Event logging in case of error (#33)

* Notify when the runlogger.handle_event() throws an exception

The can happen e.g. when the mara DB is not available for some reason

* Make event handling robust against exceptions during handling

* Make the commandline only succeed if a  RunFinished event arrived

* Clean the message cache when a pipeline run starts and finishes

Now that we keep the event handlers arround and not recreate them on each pipeline run
we have to clean up the state.

This will screw the case where we have two runs next to each other *in the browser*.
IMO thats anyways not supported... fixing thid requires giving all events a run id...

* Use OutputEvent and remove GenericExceptionEvent

Author: jankatins

data_integration/events.py

@@ -1,6 +1,7 @@
 import json
 import abc
 import datetime
+import sys
 
 
 class Event():
@@ -23,6 +24,14 @@ def handle_event(self, event: Event):
 
 def notify_configured_event_handlers(event: Event):
     from . import config
-    all_handlers = config.event_handlers()
+    try:
+        all_handlers = config.event_handlers()
+    except BaseException as e:
+        print(f"Exception while getting configured event handlers: {repr(e)}", file=sys.stderr)
+        return
+
     for handler in all_handlers:
-        handler.handle_event(event)
+        try:
+            handler.handle_event(event)
+        except BaseException as e:
+            print(f"Handler {repr(handler)} could report about {repr(event)}: {repr(e)}", file=sys.stderr)

data_integration/execution.py

@@ -8,6 +8,7 @@
 import functools
 import multiprocessing
 import os
+import sys
 import signal
 import time
 import traceback
@@ -284,13 +285,24 @@ def track_finished_pipelines():
 
     runlogger = run_log.RunLogger()
 
+    def _notify_all(event):
+        try:
+            runlogger.handle_event(event)
+        except BaseException as e:
+            # This includes the case when the mara DB is not reachable when writing the event.
+            # Not sure if we should just ignore that, but at least get other notifications
+            # out in case of an error
+            events.notify_configured_event_handlers(event)
+            # this will notify the UI in case of a problem later on
+            raise e
+        events.notify_configured_event_handlers(event)
+
     # process messages from forked child processes
     while True:
         try:
             while not event_queue.empty():
                 event = event_queue.get(False)
-                runlogger.handle_event(event)
-                events.notify_configured_event_handlers(event)
+                _notify_all(event)
                 yield event
         except queues.Empty:
             pass
@@ -301,12 +313,35 @@ def track_finished_pipelines():
             # Catching GeneratorExit needs to end in a return!
             return
         except:
-            output_event = pipeline_events.Output(node_path=pipeline.path(), message=traceback.format_exc(),
-                                                  format=logger.Format.ITALICS, is_error=True)
-            runlogger.handle_event(output_event)
-            events.notify_configured_event_handlers(output_event)
+            def _create_exception_output_event(msg: str = ''):
+                if msg:
+                    msg = msg + '\n'
+                return pipeline_events.Output(node_path=pipeline.path(), message=msg + traceback.format_exc(),
+                                              format=logger.Format.ITALICS, is_error=True)
+
+            output_event = _create_exception_output_event()
+            exception_events = []
+            try:
+                _notify_all(output_event)
+            except BaseException as e:
+                # we are already in the generic exception handler, so we cannot do anything
+                # if we still fail, as we have to get to the final close_open_run_after_error()
+                # and 'return'...
+                msg = "Could not notify about final output event"
+                exception_events.append(_create_exception_output_event(msg))
             yield output_event
-            run_log.close_open_run_after_error(runlogger.run_id)
+            try:
+                run_log.close_open_run_after_error(runlogger.run_id)
+            except BaseException as e:
+                msg = "Exception during 'close_open_run_after_error()'"
+                exception_events.append(_create_exception_output_event(msg))
+
+            # At least try to notify the UI
+            for e in exception_events:
+                print(f"{repr(e)}", file=sys.stderr)
+                yield e
+                events.notify_configured_event_handlers(e)
+
             return
         if not run_process.is_alive():
             break

data_integration/logging/pipeline_events.py

@@ -21,10 +21,6 @@ def __init__(self, node_path: [str]) -> None:
         super().__init__()
         self.node_path = node_path
 
-    def to_json(self):
-        return json.dumps({field: value.isoformat() if isinstance(value, datetime.datetime) else value
-                           for field, value in self.__dict__.items()})
-
 
 class RunStarted(PipelineEvent):
     def __init__(self, node_path: [str],
@@ -131,7 +127,7 @@ def __init__(self, node_path: [str], message: str,
         self.timestamp = datetime.datetime.now()
 
 
-def get_user_display_name(interactively_started:bool) -> t.Optional[str]:
+def get_user_display_name(interactively_started: bool) -> t.Optional[str]:
     """Gets the display name for the user which started a run
 
     Defaults to MARA_RUN_USER_DISPLAY_NAME and falls back to the current OS-level name

data_integration/notification/notifier.py

@@ -1,62 +1,66 @@
-import abc
-
-from data_integration import events
-from data_integration.logging import pipeline_events
-
-
-class ChatNotifier(events.EventHandler, abc.ABC):
-
-    def __init__(self):
-        """ Abstract class for sending notifications to chat bots when pipeline errors occur"""
-
-        # keep a list of log messages and error log messsages for each node
-        self.node_output: {tuple: {bool: [events.Event]}} = None
-
-
-    def handle_event(self, event: events.Event):
-        """
-        Send notifications for failed tasks and interactively started pipelines
-
-        Args:
-            event: The current event of interest
-        """
-
-        if isinstance(event, pipeline_events.Output):
-            # collect the output and error output of each node so that it can be shown if something fails
-            key = tuple(event.node_path)
-
-            if not self.node_output:
-                self.node_output = {}
-
-            if not key in self.node_output:
-                self.node_output[key] = {True: [], False: []}
-
-            self.node_output[key][event.is_error].append(event)
-
-        elif isinstance(event, pipeline_events.NodeFinished):
-            if not event.succeeded and event.is_pipeline is False:
-                self.send_task_failed_message(event)
-
-
-        elif isinstance(event, pipeline_events.RunStarted):
-            if event.interactively_started:
-                self.send_run_started_interactively_message(event)
-
-        elif isinstance(event, pipeline_events.RunFinished):
-            if event.interactively_started:
-                self.send_run_finished_interactively_message(event)
-
-    @abc.abstractmethod
-    def send_run_started_interactively_message(self, event: pipeline_events.RunStarted):
-        """Send a notification that somebody manually triggered the run of a pipeline"""
-        pass
-
-    @abc.abstractmethod
-    def send_run_finished_interactively_message(self, event: pipeline_events.RunFinished):
-        """Send a notification that a manually triggered pipeline run finished"""
-        pass
-
-    @abc.abstractmethod
-    def send_task_failed_message(self, event: pipeline_events.NodeFinished):
-        """Send a notification that a task failed"""
-        pass
+import abc
+
+from data_integration import events
+from data_integration.logging import pipeline_events
+
+
+class ChatNotifier(events.EventHandler, abc.ABC):
+
+    def __init__(self):
+        """ Abstract class for sending notifications to chat bots when pipeline errors occur"""
+
+        # keep a list of log messages and error log messages for each node
+        self.node_output: {tuple: {bool: [events.Event]}} = None
+
+
+    def handle_event(self, event: events.Event):
+        """
+        Send notifications for failed tasks and interactively started pipelines
+
+        Args:
+            event: The current event of interest
+        """
+
+        if isinstance(event, pipeline_events.Output):
+            # collect the output and error output of each node so that it can be shown if something fails
+            key = tuple(event.node_path)
+
+            if not self.node_output:
+                self.node_output = {}
+
+            if not key in self.node_output:
+                self.node_output[key] = {True: [], False: []}
+
+            self.node_output[key][event.is_error].append(event)
+
+        elif isinstance(event, pipeline_events.NodeFinished):
+            if not event.succeeded and event.is_pipeline is False:
+                self.send_task_failed_message(event)
+
+
+        elif isinstance(event, pipeline_events.RunStarted):
+            if event.interactively_started:
+                self.send_run_started_interactively_message(event)
+            # reset the saved outputs, just to be sure...
+            self.node_output = None
+
+        elif isinstance(event, pipeline_events.RunFinished):
+            if event.interactively_started:
+                self.send_run_finished_interactively_message(event)
+            # reset the saved outputs
+            self.node_output = None
+
+    @abc.abstractmethod
+    def send_run_started_interactively_message(self, event: pipeline_events.RunStarted):
+        """Send a notification that somebody manually triggered the run of a pipeline"""
+        pass
+
+    @abc.abstractmethod
+    def send_run_finished_interactively_message(self, event: pipeline_events.RunFinished):
+        """Send a notification that a manually triggered pipeline run finished"""
+        pass
+
+    @abc.abstractmethod
+    def send_task_failed_message(self, event: pipeline_events.NodeFinished):
+        """Send a notification that a task failed"""
+        pass

data_integration/ui/cli.py

@@ -42,15 +42,15 @@ def run_pipeline(pipeline: pipelines.Pipeline, nodes: {pipelines.Node} = None,
 
     theme = plain if disable_colors else colorful
 
-    succeeded = True
+    succeeded = False
     for event in execution.run_pipeline(pipeline, nodes, with_upstreams, interactively_started=interactively_started):
         if isinstance(event, pipeline_events.Output):
             print(f'{theme[PATH_COLOR]}{" / ".join(event.node_path)}{":" if event.node_path else ""}{theme[RESET_ALL]} '
                   + theme[event.format] + (theme[ERROR_COLOR] if event.is_error else '')
                   + event.message + theme[RESET_ALL])
         elif isinstance(event, pipeline_events.RunFinished):
-            if not event.succeeded:
-                succeeded = False
+            if event.succeeded:
+                succeeded = True
 
     return succeeded