Merge pull request #217 from botify-labs/enhancement/216/Implement-markers

Implement markers (#216)

Author: ybastide

.gitignore

@@ -43,6 +43,7 @@ output/*/index.html
 # Sphinx
 docs/_build
 docs/api/*
+!docs/api/.gitkeep
 
 # Coverage
 htmlcov/

examples/marker.py

@@ -0,0 +1,15 @@
+from simpleflow import Workflow, futures
+
+
+class MarkerWorkflow(Workflow):
+    name = 'example'
+    version = 'example'
+    task_list = 'example'
+
+    def run(self):
+        m = self.submit(self.record_marker('marker 1'))
+        m = self.submit(self.record_marker('marker 1', 'some details'))
+        self.submit(self.record_marker('marker 2', "2nd marker's details"))
+        futures.wait(m)
+        print('Markers: {}'.format(self.list_markers()))
+        print('Markers, all: {}'.format(self.list_markers(all=True)))

simpleflow/__init__.py

@@ -4,6 +4,7 @@
 
 from .activity import Activity  # NOQA
 from .workflow import Workflow  # NOQA
+from .signal import WaitForSignal  # NOQA
 
 from . import settings
 

simpleflow/command.py

@@ -426,11 +426,11 @@ def create_unique_task_list(workflow_id=''):
               required=False,
               default=60,
               help='Heartbeat interval in seconds (0 to disable heartbeating).')
-@click.option('--nb-workers',
+@click.option('--nb-workers', '-W',
               type=int,
               required=False,
               help='Number of parallel processes handling activity tasks.')
-@click.option('--nb-deciders',
+@click.option('--nb-deciders', '-D',
               type=int,
               required=False,
               help='Number of parallel processes handling decision tasks.')

simpleflow/executor.py

@@ -161,11 +161,18 @@ def get_execution_context(self):
         """
         return {}
 
+    @abc.abstractmethod
     def signal(self, name, *args, **kwargs):
         raise NotImplementedError
 
+    @abc.abstractmethod
     def wait_signal(self, name):
         raise NotImplementedError
 
-    def get_future_from_signal(self, name):
+    @abc.abstractmethod
+    def record_marker(self, name, details=None):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def list_markers(self, all=False):
         raise NotImplementedError

simpleflow/history.py

@@ -21,6 +21,8 @@ class History(object):
     :type _external_workflows_canceling: collections.OrderedDict[str, dict[str, Any]]
     :ivar _signals: activity events
     :type _signals: collections.OrderedDict[str, dict[str, Any]]
+    :ivar _markers: marker events
+    :type _markers: collections.OrderedDict[str, list[dict[str, Any]]]
     :ivar _tasks: ordered list of tasks/etc
     :type _tasks: list[dict[str, Any]]
     """
@@ -33,8 +35,18 @@ def __init__(self, history):
         self._external_workflows_canceling = collections.OrderedDict()
         self._signals = collections.OrderedDict()
         self._signaled_workflows = collections.defaultdict(list)
+        self._markers = collections.OrderedDict()
         self._tasks = []
 
+    @property
+    def swf_history(self):
+        """
+
+        :return: SWF history
+        :rtype: swf.models.history.History
+        """
+        return self._history
+
     @property
     def activities(self):
         """
@@ -75,8 +87,21 @@ def signaled_workflows(self):
         """
         return self._signaled_workflows
 
+    @property
+    def markers(self):
+        """
+
+        :return: Markers
+        :rtype: collections.OrderedDict[str, list[dict[str, Any]]]
+        """
+        return self._markers
+
     @property
     def tasks(self):
+        """
+        :return:
+         :rtype: list[dict[str, Any]]
+        """
         return self._tasks
 
     @property
@@ -422,11 +447,34 @@ def get_workflow(workflows):
             workflow['workflow_id'] = event.workflow_execution['workflowId']
             workflow['cancel_requested_timestamp'] = event.timestamp
 
+    def parse_marker_event(self, events, event):
+        if event.state == 'recorded':
+            marker = {
+                'type': 'marker',
+                'name': event.marker_name,
+                'state': event.state,
+                'details': getattr(event, 'details', None),
+                'recorded_event_id': event.id,
+                'recorded_event_timestamp': event.timestamp,
+            }
+            self._markers.setdefault(event.marker_name, []).append(marker)
+        elif event.state == 'failed':
+            marker = {
+                'type': 'marker',
+                'name': event.marker_name,
+                'state': event.state,
+                'cause': event.cause,
+                'record_failed_event_id': event.id,
+                'record_failed_event_timestamp': event.timestamp,
+            }
+            self._markers.setdefault(event.marker_name, []).append(marker)
+
     TYPE_TO_PARSER = {
         'ActivityTask': parse_activity_event,
         'ChildWorkflowExecution': parse_child_workflow_event,
         'WorkflowExecution': parse_workflow_event,
         'ExternalWorkflowExecution': parse_external_workflow_event,
+        'Marker': parse_marker_event,
     }
 
     def parse(self):

simpleflow/local/executor.py

@@ -1,3 +1,4 @@
+import collections
 import logging
 
 from simpleflow import (
@@ -6,8 +7,9 @@
     futures,
 )
 from simpleflow.base import Submittable
+from simpleflow.marker import Marker
 from simpleflow.signal import WaitForSignal
-from simpleflow.task import ActivityTask, WorkflowTask, SignalTask
+from simpleflow.task import ActivityTask, WorkflowTask, SignalTask, MarkerTask
 from simpleflow.activity import Activity
 from simpleflow.workflow import Workflow
 from swf.models.history import builder
@@ -22,11 +24,13 @@ class Executor(executor.Executor):
     Executes all tasks synchronously in a single local process.
 
     """
+
     def __init__(self, workflow_class):
         super(Executor, self).__init__(workflow_class)
         self.update_workflow_class()
         self.nb_activities = 0
         self.signals_sent = set()
+        self._markers = collections.OrderedDict()
 
     def update_workflow_class(self):
         """
@@ -65,6 +69,8 @@ def submit(self, func, *args, **kwargs):
                 raise NotImplementedError(
                     'wait_signal({}) before signal was sent: unsupported by the local executor'.format(signal_name)
                 )
+        elif isinstance(func, MarkerTask):
+            self._markers.setdefault(func.name, []).append(Marker(func.name, func.details))
 
         if isinstance(func, Submittable):
             task = func  # *args, **kwargs already resolved.
@@ -139,3 +145,11 @@ def signal(self, name, *args, **kwargs):
 
     def wait_signal(self, name):
         return WaitForSignal(name)
+
+    def record_marker(self, name, details=None):
+        return MarkerTask(name, details)
+
+    def list_markers(self, all=False):
+        if all:
+            return [m for ml in self._markers.values() for m in ml]
+        return [m[-1] for m in self._markers.values()]

simpleflow/marker.py

@@ -0,0 +1,11 @@
+class Marker(object):
+    def __init__(self, name, details):
+        self.name = name
+        self.details = details
+
+    def __repr__(self):
+        return '<{klass} {name!r} details={details!r}>'.format(
+            klass=self.__class__.__name__,
+            name=self.name,
+            details=self.details,
+        )