Add start and end time to logs for workflows, dags and tasks

Author: Andreas Moll

examples/timing.py

@@ -18,10 +18,14 @@ def random_sleep(data, store, signal, context):
 
 # the callback function for the task that prints the run times
 def print_times(data, store, signal, context):
-    dag_log = store.get(key='log.{}'.format(context.dag_name),
+    dag_log = store.get(key='log.{}.tasks'.format(context.dag_name),
                         section=DataStoreDocumentSection.Meta)
     for task, fields in dag_log.items():
-        print(task, 'on', fields['worker'], 'took', fields['duration'], 'seconds')
+        # The print task has not finished yet, so there is no duration available
+        if task != context.task_name:
+            print(task, 'on', fields['worker'], 'took', fields['duration'], 'seconds')
+        else:
+            print(task, 'on', fields['worker'], 'is still running')
 
 
 # create the main DAG

lightflow/models/datastore.py

@@ -3,6 +3,7 @@
 from bson.binary import Binary
 from bson.objectid import ObjectId
 import pickle
+from datetime import datetime
 from gridfs import GridFS
 from urllib.parse import quote_plus
 
@@ -358,7 +359,7 @@ def _encode_value(self, value):
         Returns:
             object: The encoded value ready to be stored in MongoDB.
         """
-        if isinstance(value, (int, float, str, bool)):
+        if isinstance(value, (int, float, str, bool, datetime)):
             return value
         elif isinstance(value, list):
             return [self._encode_value(item) for item in value]
@@ -388,7 +389,7 @@ def _decode_value(self, value):
         Returns:
             object: The decoded value as a valid Python object.
         """
-        if isinstance(value, (int, float, str, bool)):
+        if isinstance(value, (int, float, str, bool, datetime)):
             return value
         elif isinstance(value, list):
             return [self._decode_value(item) for item in value]

lightflow/queue/jobs.py

@@ -1,5 +1,5 @@
 import celery
-from datetime import datetime
+from datetime import datetime, timezone
 from functools import partial
 
 from lightflow.logger import get_logger
@@ -26,7 +26,7 @@ def execute_workflow(self, workflow, workflow_id=None):
         workflow_id (string): If a workflow ID is provided the workflow run will use
                               this ID, if not a new ID will be auto generated.
     """
-    start_time = datetime.now()
+    start_time = datetime.utcnow()
 
     logger.info('Running workflow <{}>'.format(workflow.name))
     data_store = DataStore(**self.app.user_options['config'].data_store,
@@ -39,7 +39,7 @@ def execute_workflow(self, workflow, workflow_id=None):
         workflow_id = data_store.add(payload={
                                          'name': workflow.name,
                                          'queue': workflow.queue,
-                                         'start_time': datetime.utcnow()
+                                         'start_time': start_time
                                      })
         logger.info('Created workflow ID: {}'.format(workflow_id))
 
@@ -48,7 +48,7 @@ def execute_workflow(self, workflow, workflow_id=None):
                     job_type=JobType.Workflow,
                     name=workflow.name,
                     queue=workflow.queue,
-                    time=datetime.utcnow(),
+                    time=start_time,
                     workflow_id=workflow_id,
                     duration=None)
 
@@ -62,7 +62,7 @@ def execute_workflow(self, workflow, workflow_id=None):
                             'type': JobType.Workflow,
                             'workflow_id': workflow_id,
                             'queue': workflow.queue,
-                            'start_time': datetime.utcnow(),
+                            'start_time': start_time,
                             'arguments': workflow.provided_arguments})
 
     # run the DAGs in the workflow
@@ -71,6 +71,16 @@ def execute_workflow(self, workflow, workflow_id=None):
                  signal_server=signal_server,
                  workflow_id=workflow_id)
 
+    end_time = datetime.utcnow()
+    duration = (end_time - start_time).total_seconds()
+
+    # update data store with provenance information
+    store_doc = data_store.get(workflow_id)
+    store_doc.set(key='end_time', value=end_time,
+                  section=DataStoreDocumentSection.Meta)
+    store_doc.set(key='duration', value=duration,
+                  section=DataStoreDocumentSection.Meta)
+
     # send custom celery event that the workflow has succeeded
     event_name = JobEventName.Succeeded if not workflow.is_stopped \
         else JobEventName.Aborted
@@ -79,9 +89,9 @@ def execute_workflow(self, workflow, workflow_id=None):
                     job_type=JobType.Workflow,
                     name=workflow.name,
                     queue=workflow.queue,
-                    time=datetime.utcnow(),
+                    time=end_time,
                     workflow_id=workflow_id,
-                    duration=(datetime.now() - start_time).total_seconds())
+                    duration=duration)
 
     logger.info('Finished workflow <{}>'.format(workflow.name))
 
@@ -101,15 +111,23 @@ def execute_dag(self, dag, workflow_id, data=None):
                               the first tasks in the dag. This allows the transfer of
                               data from dag to dag.
     """
-    start_time = datetime.now()
+    start_time = datetime.utcnow()
     logger.info('Running DAG <{}>'.format(dag.name))
 
+    store_doc = DataStore(**self.app.user_options['config'].data_store,
+                          auto_connect=True).get(workflow_id)
+    store_loc = 'log.{}'.format(dag.name)
+
+    # update data store with provenance information
+    store_doc.set(key='{}.start_time'.format(store_loc), value=start_time,
+                  section=DataStoreDocumentSection.Meta)
+
     # send custom celery event that the dag has been started
     self.send_event(JobEventName.Started,
                     job_type=JobType.Dag,
                     name=dag.name,
                     queue=dag.queue,
-                    time=datetime.utcnow(),
+                    time=start_time,
                     workflow_id=workflow_id,
                     duration=None)
 
@@ -128,15 +146,24 @@ def execute_dag(self, dag, workflow_id, data=None):
             signal=signal,
             data=data)
 
+    end_time = datetime.utcnow()
+    duration = (end_time - start_time).total_seconds()
+
+    # update data store with provenance information
+    store_doc.set(key='{}.end_time'.format(store_loc), value=end_time,
+                  section=DataStoreDocumentSection.Meta)
+    store_doc.set(key='{}.duration'.format(store_loc), value=duration,
+                  section=DataStoreDocumentSection.Meta)
+
     # send custom celery event that the dag has succeeded
     event_name = JobEventName.Succeeded if not signal.is_stopped else JobEventName.Aborted
     self.send_event(event_name,
                     job_type=JobType.Dag,
                     name=dag.name,
                     queue=dag.queue,
-                    time=datetime.utcnow(),
+                    time=end_time,
                     workflow_id=workflow_id,
-                    duration=(datetime.now() - start_time).total_seconds())
+                    duration=duration)
 
     logger.info('Finished DAG <{}>'.format(dag.name))
 
@@ -153,10 +180,11 @@ def execute_task(self, task, workflow_id, data=None):
         data (MultiTaskData): An optional MultiTaskData object that contains the data
                               that has been passed down from upstream tasks.
     """
-    start_time = datetime.now()
+    start_time = datetime.utcnow()
 
     store_doc = DataStore(**self.app.user_options['config'].data_store,
                           auto_connect=True).get(workflow_id)
+    store_loc = 'log.{}.tasks.{}'.format(task.dag_name, task.name)
 
     def handle_callback(message, event_type, exc=None):
         msg = '{}: {}'.format(message, str(exc)) if exc is not None else message
@@ -169,26 +197,40 @@ def handle_callback(message, event_type, exc=None):
         else:
             logger.info(msg)
 
-        # store a log into the persistent data store
+        current_time = datetime.utcnow()
+
+        # store provenance information about a task
         if event_type != JobEventName.Started:
-            duration = (datetime.now() - start_time).total_seconds()
+            duration = (current_time - start_time).total_seconds()
+
+            store_doc.set(key='{}.end_time'.format(store_loc),
+                          value=current_time,
+                          section=DataStoreDocumentSection.Meta)
 
-            store_doc.set(key='log.{}.{}.duration'.format(task.dag_name, task.name),
+            store_doc.set(key='{}.duration'.format(store_loc),
                           value=duration,
                           section=DataStoreDocumentSection.Meta)
+        else:
+            # store provenance information about a task
+            store_doc.set(key='{}.start_time'.format(store_loc),
+                          value=start_time,
+                          section=DataStoreDocumentSection.Meta)
 
-            store_doc.set(key='log.{}.{}.worker'.format(task.dag_name, task.name),
+            store_doc.set(key='{}.worker'.format(store_loc),
                           value=self.request.hostname,
                           section=DataStoreDocumentSection.Meta)
-        else:
+
+            store_doc.set(key='{}.queue'.format(store_loc),
+                          value=task.queue,
+                          section=DataStoreDocumentSection.Meta)
             duration = None
 
         # send custom celery event
         self.send_event(event_type,
                         job_type=JobType.Task,
                         name=task.name,
                         queue=task.queue,
-                        time=datetime.utcnow(),
+                        time=current_time,
                         workflow_id=workflow_id,
                         duration=duration)
 

lightflow/version.py

@@ -1 +1 @@
-__version__ = '1.10.1'
+__version__ = '1.10.2'