Make logger/event handlers configureable (#27)

* Make event_handler a config function

* Convert Slack to use event handler instead of using slack directly

* Notify about regular pipeline start/end

* Also add user in non-interactive runs

This also makes the function patchable in case we run in a built docker container and pass in the real "starter" of a run in an env variable...

* Merge PipelineStart/EndEvents into RunStarted/Finished  

Also move slack notification event handler to its own package and rename event_base.py to events.py

Author: jankatins

data_integration/config.py

@@ -4,8 +4,9 @@
 import multiprocessing
 import pathlib
 import typing
+import functools
 
-from . import pipelines
+from . import pipelines, events
 
 
 def root_pipeline() -> 'pipelines.Pipeline':
@@ -63,14 +64,24 @@ def base_url() -> str:
     return 'http://127.0.0.1:5000/data-integration'
 
 
-def slack_token() -> str:
+def slack_token() -> typing.Optional[str]:
     """
     When not None, then this slack webhook is notified of failed nodes.
     Slack channel's token (i.e. THISIS/ASLACK/TOCKEN) can be retrieved from the
     channel's app "Incoming WebHooks" configuration as part part of the Webhook URL
     """
     return None
 
+@functools.lru_cache(maxsize=None)
+def event_handlers() -> [events.EventHandler]:
+    """User specific envent handlers (mainly to notify chat systems)"""
+    configured_handlers = []
+    if slack_token():
+        from data_integration.notification import slack
+        configured_handlers.append(slack.Slack())
+    return configured_handlers
+
+
 def password_masks() -> typing.List[str]:
     """Any passwords which should be masked in the UI or logs"""
 

data_integration/events.py

@@ -0,0 +1,28 @@
+import json
+import abc
+import datetime
+
+
+class Event():
+    def __init__(self) -> None:
+        """
+        Base class for events that are emitted from mara.
+        """
+        pass
+
+    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 EventHandler(abc.ABC):
+    @abc.abstractmethod
+    def handle_event(self, event: Event):
+        pass
+
+
+def notify_configured_event_handlers(event: Event):
+    from . import config
+    all_handlers = config.event_handlers()
+    for handler in all_handlers:
+        handler.handle_event(event)

data_integration/execution.py

@@ -13,11 +13,14 @@
 from multiprocessing import queues
 
 from . import pipelines, config
-from .logging import logger, events, system_statistics, run_log, node_cost, slack
+from .logging import logger, pipeline_events, system_statistics, run_log, node_cost
+from . import events
 
 
 def run_pipeline(pipeline: pipelines.Pipeline, nodes: {pipelines.Node} = None,
-                 with_upstreams: bool = False) -> [events.Event]:
+                 with_upstreams: bool = False,
+                 interactively_started: bool = False
+                 ) -> [events.Event]:
     """
     Runs a pipeline in a forked sub process. Acts as a generator that yields events from the sub process.
 
@@ -129,17 +132,23 @@ def track_finished_pipelines():
                     in dict(running_pipelines).items():  # type: pipelines.Pipeline
                     if len(set(running_pipeline.nodes.values()) & processed_nodes) == len(running_pipeline.nodes):
                         succeeded = running_pipeline not in failed_pipelines
-                        event_queue.put(events.Output(
+                        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())}'))
-                        event_queue.put(events.NodeFinished(
+                        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))
                         del running_pipelines[running_pipeline]
                         processed_nodes.add(running_pipeline)
 
             # announce run start
-            event_queue.put(events.RunStarted(node_path=pipeline.path(), start_time=run_start_time, pid=os.getpid()))
+            event_queue.put(pipeline_events.RunStarted(node_path=pipeline.path(),
+                                                       start_time=run_start_time,
+                                                       pid=os.getpid(),
+                                                       interactively_started=interactively_started,
+                                                       node_ids=[node.id for node in (nodes or [])],
+                                                       is_root_pipeline=(pipeline.parent is None))
+                            )
 
             # run as long
             # - as task processes are still running
@@ -173,8 +182,8 @@ def track_finished_pipelines():
                             # book keeping and event emission
                             pipeline_start_time = datetime.datetime.now()
                             running_pipelines[next_node] = [pipeline_start_time, 0]
-                            event_queue.put(events.NodeStarted(next_node.path(), pipeline_start_time, True))
-                            event_queue.put(events.Output(
+                            event_queue.put(pipeline_events.NodeStarted(next_node.path(), pipeline_start_time, True))
+                            event_queue.put(pipeline_events.Output(
                                 node_path=next_node.path(), format=logger.Format.ITALICS,
                                 message='★ ' + node_cost.format_duration(
                                     node_durations_and_run_times.get(tuple(next_node.path()), [0, 0])[0])))
@@ -190,13 +199,13 @@ def track_finished_pipelines():
                                 queue([sub_pipeline])
 
                             except Exception as e:
-                                event_queue.put(events.NodeStarted(
+                                event_queue.put(pipeline_events.NodeStarted(
                                     node_path=next_node.path(), start_time=task_start_time, is_pipeline=True))
                                 logger.log(message=f'Could not launch parallel tasks', format=logger.Format.ITALICS,
                                            is_error=True)
                                 logger.log(message=traceback.format_exc(),
-                                           format=events.Output.Format.VERBATIM, is_error=True)
-                                event_queue.put(events.NodeFinished(
+                                           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))
 
@@ -209,8 +218,9 @@ def track_finished_pipelines():
                             # run a task in a subprocess
                             if next_node.parent in running_pipelines:
                                 running_pipelines[next_node.parent][1] += 1
-                            event_queue.put(events.NodeStarted(next_node.path(), datetime.datetime.now(), False))
-                            event_queue.put(events.Output(
+                            event_queue.put(
+                                pipeline_events.NodeStarted(next_node.path(), datetime.datetime.now(), False))
+                            event_queue.put(pipeline_events.Output(
                                 node_path=next_node.path(), format=logger.Format.ITALICS,
                                 message='★ ' + node_cost.format_duration(
                                     node_durations_and_run_times.get(tuple(next_node.path()), [0, 0])[0])))
@@ -238,12 +248,12 @@ def track_finished_pipelines():
 
                         end_time = datetime.datetime.now()
                         event_queue.put(
-                            events.Output(task_process.task.path(),
-                                          ('succeeded' if succeeded else 'failed') + ',  '
-                                          + logger.format_time_difference(task_process.start_time, end_time),
-                                          format=logger.Format.ITALICS, is_error=not succeeded))
-                        event_queue.put(events.NodeFinished(task_process.task.path(), task_process.start_time,
-                                                            end_time, False, succeeded))
+                            pipeline_events.Output(task_process.task.path(),
+                                                   ('succeeded' if succeeded else 'failed') + ',  '
+                                                   + logger.format_time_difference(task_process.start_time, end_time),
+                                                   format=logger.Format.ITALICS, is_error=not succeeded))
+                        event_queue.put(pipeline_events.NodeFinished(task_process.task.path(), task_process.start_time,
+                                                                     end_time, False, succeeded))
 
                 # check if some pipelines finished
                 track_finished_pipelines()
@@ -252,8 +262,8 @@ def track_finished_pipelines():
                 time.sleep(0.001)
 
         except:
-            event_queue.put(events.Output(node_path=pipeline.path(), message=traceback.format_exc(),
-                                          format=logger.Format.ITALICS, is_error=True))
+            event_queue.put(pipeline_events.Output(node_path=pipeline.path(), message=traceback.format_exc(),
+                                                   format=logger.Format.ITALICS, is_error=True))
 
         # run again because `dequeue` might have moved more nodes to `finished_nodes`
         track_finished_pipelines()
@@ -263,27 +273,23 @@ def track_finished_pipelines():
         statistics_process.join()
 
         # run finished
-        event_queue.put(events.RunFinished(node_path=pipeline.path(), end_time=datetime.datetime.now(),
-                                           succeeded=not failed_pipelines))
+        event_queue.put(pipeline_events.RunFinished(node_path=pipeline.path(), end_time=datetime.datetime.now(),
+                                                    succeeded=not failed_pipelines,
+                                                    interactively_started=interactively_started))
 
     # fork the process and run `run`
     run_process = multiprocessing_context.Process(target=run, name='pipeline-' + '-'.join(pipeline.path()))
     run_process.start()
 
     runlogger = run_log.RunLogger()
-    event_handlers = [runlogger]
-
-    # todo: make event handlers configurable (e.g. for slack)
-    if config.slack_token():
-        event_handlers.append(slack.Slack())
 
     # process messages from forked child processes
     while True:
         try:
             while not event_queue.empty():
                 event = event_queue.get(False)
-                for event_handler in event_handlers:
-                    event_handler.handle_event(event)
+                runlogger.handle_event(event)
+                events.notify_configured_event_handlers(event)
                 yield event
         except queues.Empty:
             pass
@@ -294,10 +300,10 @@ def track_finished_pipelines():
             # Catching GeneratorExit needs to end in a return!
             return
         except:
-            output_event = events.Output(node_path=pipeline.path(), message=traceback.format_exc(),
-                                format=logger.Format.ITALICS, is_error=True)
-            for event_handler in event_handlers:
-                event_handler.handle_event(output_event)
+            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)
             yield output_event
             run_log.close_open_run_after_error(runlogger.run_id)
             return

data_integration/logging/logger.py

@@ -4,13 +4,13 @@
 import sys
 from datetime import datetime
 
-from ..logging import events
+from ..logging import pipeline_events
 import data_integration.config
 
-Format = events.Output.Format
+Format = pipeline_events.Output.Format
 
 
-def log(message: str, format: events.Output.Format = Format.STANDARD,
+def log(message: str, format: pipeline_events.Output.Format = Format.STANDARD,
         is_error: bool = False) -> None:
     """
     Logs text messages.
@@ -32,7 +32,7 @@ def log(message: str, format: events.Output.Format = Format.STANDARD,
             message = message.replace(mask, '***')
     if message:
         if _event_queue:
-            _event_queue.put(events.Output(_node_path, message, format, is_error))
+            _event_queue.put(pipeline_events.Output(_node_path, message, format, is_error))
         elif is_error:
             sys.stderr.write(message + '\n')
         else:

data_integration/logging/events.py …a_integration/logging/pipeline_events.pydata_integration/logging/events.py renamed to data_integration/logging/pipeline_events.py data_integration/logging/events.py renamed to data_integration/logging/pipeline_events.py

@@ -1,27 +1,12 @@
 """Events that are emitted during pipeline execution"""
 
-import abc
 import datetime
 import json
 
 import enum
+import typing as t
 
-
-class Event():
-    def __init__(self) -> None:
-        """
-        Base class for events that are emitted from mara.
-        """
-
-    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 EventHandler(abc.ABC):
-    @abc.abstractmethod
-    def handle_event(self, event: Event):
-        pass
+from ..events import Event
 
 
 class PipelineEvent(Event):
@@ -42,33 +27,51 @@ def to_json(self):
 
 
 class RunStarted(PipelineEvent):
-    def __init__(self, node_path: [str], start_time: datetime.datetime, pid: int) -> None:
+    def __init__(self, node_path: [str],
+                 start_time: datetime.datetime,
+                 pid: int,
+                 is_root_pipeline: bool = False,
+                 node_ids: t.Optional[t.List[str]] = None,
+                 interactively_started: bool = False) -> None:
         """
         A pipeline run started
         Args:
             node_path: The path of the pipeline that was run
             start_time: The time when the run started
             pid: The process id of the process that runs the pipeline
+            node_ids: list of node.ids which should be run
+            is_root_pipeline: whether this pipeline run runs the root pipeline
+            interactively_started: whether or not the run was started interactively
         """
         super().__init__([])
         self.node_path = node_path
         self.start_time = start_time
         self.pid = pid
+        self.interactively_started = interactively_started
+        self.is_root_pipeline = is_root_pipeline
+        self.node_ids = node_ids or []
+
+        self.user: str = get_user_display_name(interactively_started)
 
 
 class RunFinished(PipelineEvent):
-    def __init__(self, node_path: [str], end_time: datetime.datetime, succeeded: bool) -> None:
+    def __init__(self, node_path: [str],
+                 end_time: datetime.datetime,
+                 succeeded: bool,
+                 interactively_started: bool = False) -> None:
         """
         A pipeline run finished
         Args:
             node_path: The path of the pipeline that was run
             end_time: The time when the run finished
             succeeded: Whether the run succeeded
+            interactively_started: whether or not the run was started interactively
         """
         super().__init__([])
         self.node_path = node_path
         self.end_time = end_time
         self.succeeded = succeeded
+        self.interactively_started = interactively_started
 
 
 class NodeStarted(PipelineEvent):
@@ -126,3 +129,21 @@ def __init__(self, node_path: [str], message: str,
         self.format = format
         self.is_error = is_error
         self.timestamp = datetime.datetime.now()
+
+
+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
+    if the run was started interactively, else None.
+
+    :param interactively_started: True if the run was triggered interactively
+
+    Patch if you have more sophisticated needs.
+    """
+    import os
+    if 'MARA_RUN_USER_DISPLAY_NAME' in os.environ:
+        return os.environ.get('MARA_RUN_USER_DISPLAY_NAME')
+    if not interactively_started:
+        return None
+    return os.environ.get('SUDO_USER') or os.environ.get('USER') or os.getlogin()