Fix Problems when frontend and database are in a different timezone (#34)

The problem was that a) the frontend generated timestamps without timezones which postgresql then interpreted as the UTC but which were actually in the timezone which you ran the python code and b) the closing of open runs after an exception which kills the ETL runner itself (not individual tasks) after an error happened to insert a timestamp with timezone of the time on the postgresql server.

-> the error only happens when an exception happens which triggers this code path, i.e. because something shuts down the the runner (e.g. docker instance is shut down or ctrl+c on the command line).

Should fix mara/mara-example-project-2#12

Author: jankatins

CHANGELOG.md

@@ -1,11 +1,14 @@
 # Changelog
 
+## 2.8.1 (2020-04-27)
+
+- Fix Problems when frontend and database are in a different timezone (#34)
+
 ## 2.8.0 (2020-03-25)
 
 - Implement pipeline notifications via Microsoft Teams #28
 - Make it possible to disable output coloring in command line etl runs (#31)
 
-
 ## 2.7.0 (2020-03-05)
 
 - Make event handlers configurable: this allows for e.g. adding your own notifier for specific events

data_integration/execution.py

@@ -4,6 +4,7 @@
 """
 
 import datetime
+from datetime import timezone as tz
 import functools
 import multiprocessing
 import os
@@ -99,7 +100,7 @@ def with_all_upstreams(nodes: {pipelines.Node}):
                 queue([pipeline])
 
             # book keeping
-            run_start_time = datetime.datetime.now()
+            run_start_time = datetime.datetime.now(tz.utc)
             # all nodes that already ran or that won't be run anymore
             processed_nodes: {pipelines.Node} = set()
             # running pipelines with start times and number of running children
@@ -134,10 +135,10 @@ def track_finished_pipelines():
                         succeeded = running_pipeline not in failed_pipelines
                         event_queue.put(pipeline_events.Output(
                             node_path=running_pipeline.path(), format=logger.Format.ITALICS, is_error=not succeeded,
-                            message=f'{"succeeded" if succeeded else "failed"}, {logger.format_time_difference(run_start_time, datetime.datetime.now())}'))
+                            message=f'{"succeeded" if succeeded else "failed"}, {logger.format_time_difference(run_start_time, datetime.datetime.now(tz.utc))}'))
                         event_queue.put(pipeline_events.NodeFinished(
                             node_path=running_pipeline.path(), start_time=start_time,
-                            end_time=datetime.datetime.now(), is_pipeline=True, succeeded=succeeded))
+                            end_time=datetime.datetime.now(tz.utc), is_pipeline=True, succeeded=succeeded))
                         del running_pipelines[running_pipeline]
                         processed_nodes.add(running_pipeline)
 
@@ -180,7 +181,7 @@ def track_finished_pipelines():
                             queue(list(next_node.nodes.values()))
 
                             # book keeping and event emission
-                            pipeline_start_time = datetime.datetime.now()
+                            pipeline_start_time = datetime.datetime.now(tz.utc)
                             running_pipelines[next_node] = [pipeline_start_time, 0]
                             event_queue.put(pipeline_events.NodeStarted(next_node.path(), pipeline_start_time, True))
                             event_queue.put(pipeline_events.Output(
@@ -190,7 +191,7 @@ def track_finished_pipelines():
 
                         elif isinstance(next_node, pipelines.ParallelTask):
                             # create sub tasks and queue them
-                            task_start_time = datetime.datetime.now()
+                            task_start_time = datetime.datetime.now(tz.utc)
                             try:
                                 logger.redirect_output(event_queue, next_node.path())
                                 logger.log('☆ Launching tasks', format=logger.Format.ITALICS)
@@ -207,7 +208,7 @@ def track_finished_pipelines():
                                            format=pipeline_events.Output.Format.VERBATIM, is_error=True)
                                 event_queue.put(pipeline_events.NodeFinished(
                                     node_path=next_node.path(), start_time=task_start_time,
-                                    end_time=datetime.datetime.now(), is_pipeline=True, succeeded=False))
+                                    end_time=datetime.datetime.now(tz.utc), is_pipeline=True, succeeded=False))
 
                                 failed_pipelines.add(next_node.parent)
                                 processed_nodes.add(next_node)
@@ -219,7 +220,7 @@ def track_finished_pipelines():
                             if next_node.parent in running_pipelines:
                                 running_pipelines[next_node.parent][1] += 1
                             event_queue.put(
-                                pipeline_events.NodeStarted(next_node.path(), datetime.datetime.now(), False))
+                                pipeline_events.NodeStarted(next_node.path(), datetime.datetime.now(tz.utc), False))
                             event_queue.put(pipeline_events.Output(
                                 node_path=next_node.path(), format=logger.Format.ITALICS,
                                 message='★ ' + node_cost.format_duration(
@@ -246,7 +247,7 @@ def track_finished_pipelines():
                             for parent in task_process.task.parents()[:-1]:
                                 failed_pipelines.add(parent)
 
-                        end_time = datetime.datetime.now()
+                        end_time = datetime.datetime.now(tz.utc)
                         event_queue.put(
                             pipeline_events.Output(task_process.task.path(),
                                                    ('succeeded' if succeeded else 'failed') + ',  '
@@ -273,7 +274,7 @@ def track_finished_pipelines():
         statistics_process.join()
 
         # run finished
-        event_queue.put(pipeline_events.RunFinished(node_path=pipeline.path(), end_time=datetime.datetime.now(),
+        event_queue.put(pipeline_events.RunFinished(node_path=pipeline.path(), end_time=datetime.datetime.now(tz.utc),
                                                     succeeded=not failed_pipelines,
                                                     interactively_started=interactively_started))
 
@@ -326,7 +327,7 @@ def __init__(self, task: pipelines.Task, event_queue: multiprocessing.Queue, sta
         self.task = task
         self.event_queue = event_queue
         self.status_queue = status_queue
-        self.start_time = datetime.datetime.now()
+        self.start_time = datetime.datetime.now(tz.utc)
 
     def run(self):
         # redirect stdout and stderr to queue

setup.py

@@ -7,7 +7,7 @@ def get_long_description():
 
 setup(
     name='data-integration',
-    version='2.8.0',
+    version='2.8.1',
 
     description='Opinionated lightweight ETL pipeline framework',