Merge pull request #273 from botify-labs/enhancement/272/Implement-workflow-cancelation

Enhancement/272/implement workflow cancelation

Author: ybastide

simpleflow/history.py

@@ -40,6 +40,10 @@ def __init__(self, history):
         self._markers = collections.OrderedDict()
         self._timers = {}
         self._tasks = []
+        self._cancel_requested = None
+        self._cancel_failed = None
+        self.started_decision_id = None
+        self.completed_decision_id = None
 
     @property
     def swf_history(self):
@@ -82,6 +86,38 @@ def signals(self):
         """
         return self._signals
 
+    @property
+    def cancel_requested(self):
+        """
+        :return: Last cancel requested event, if any.
+        :rtype: Optional[dict]
+        """
+        return self._cancel_requested
+
+    @property
+    def cancel_failed(self):
+        """
+        :return: Last cancel failed event, if any.
+        :rtype: Optional[dict]
+        """
+        return self._cancel_failed
+
+    @property
+    def cancel_requested_id(self):
+        """
+        :return: ID of last cancel requested event, if any.
+        :rtype: Optional[int]
+        """
+        return self._cancel_requested['event_id'] if self._cancel_requested else None
+
+    @property
+    def cancel_failed_decision_task_completed_event_id(self):
+        """
+        :return: ID of last cancel failed event, if any.
+        :rtype: Optional[int]
+        """
+        return self._cancel_failed['decision_task_completed_event_id'] if self._cancel_failed else None
+
     @property
     def signaled_workflows(self):
         """
@@ -222,6 +258,18 @@ def get_activity():
             activity['state'] = event.state
             activity['details'] = getattr(event, 'details', '')
             activity['cancelled_timestamp'] = event.timestamp
+        elif event.state == 'cancel_requested':
+            activity = {
+                'type': 'activity',
+                'id': event.activity_id,
+                'state': event.state,
+                'cancel_requested_timestamp': event.timestamp,
+            }
+            if event.activity_id not in self._activities:
+                self._activities[event.activity_id] = activity
+                self._tasks.append(activity)
+            else:
+                self._activities[event.activity_id].update(activity)
 
     def parse_child_workflow_event(self, events, event):
         """Aggregate all the attributes of a workflow in a single entry.
@@ -371,13 +419,34 @@ def parse_workflow_event(self, events, event):
             }
             self._signals[event.signal_name] = signal
             self._tasks.append(signal)
+        elif event.state == 'cancel_requested':
+            cancel_requested = {
+                'type': event.state,
+                'cause': getattr(event, 'cause', None),
+                'external_initiated_event_id': getattr(event, 'external_initiated_event_id', None),
+                'external_run_id': getattr(event, 'external_workflow_execution', {}).get('runId'),
+                'external_workflow_id': getattr(event, 'external_workflow_execution', {}).get('workflowId'),
+                'event_id': event.id,
+                'timestamp': event.timestamp,
+            }
+            self._cancel_requested = cancel_requested
+        elif event.state == 'cancel_failed':
+            cancel_failed = {
+                'type': event.state,
+                'cause': getattr(event, 'cause', None),
+                'event_id': event.id,
+                'decision_task_completed_event_id': event.decision_task_completed_event_id,
+                'timestamp': event.timestamp,
+            }
+            self._cancel_failed = cancel_failed
 
     def parse_external_workflow_event(self, events, event):
         """
         Parse an external workflow event.
         :param events:
         :param event:
         """
+
         def get_workflow(workflows):
             initiated_event = events[event.initiated_event_id - 1]
             return workflows[initiated_event.workflow_id]
@@ -452,6 +521,7 @@ def get_workflow(workflows):
             workflow = get_workflow(self._external_workflows_canceling)
             workflow['run_id'] = event.workflow_execution['runId']
             workflow['workflow_id'] = event.workflow_execution['workflowId']
+            workflow['cancel_requested_event_id'] = event.id
             workflow['cancel_requested_timestamp'] = event.timestamp
 
     def parse_marker_event(self, events, event):
@@ -522,8 +592,15 @@ def parse_timer_event(self, events, event):
             timer['cancel_failed_event_id'] = event.id
             timer['cancel_failed_event_timestamp'] = event.timestamp
 
+    def parse_decision_event(self, events, event):
+        if event.state == 'started':
+            self.started_decision_id = event.id
+        if event.state == 'completed':
+            self.completed_decision_id = event.id
+
     TYPE_TO_PARSER = {
         'ActivityTask': parse_activity_event,
+        'DecisionTask': parse_decision_event,
         'ChildWorkflowExecution': parse_child_workflow_event,
         'WorkflowExecution': parse_workflow_event,
         'ExternalWorkflowExecution': parse_external_workflow_event,

simpleflow/local/executor.py

@@ -11,6 +11,7 @@
 from simpleflow.signal import WaitForSignal
 from simpleflow.task import ActivityTask, WorkflowTask, SignalTask, MarkerTask
 from simpleflow.activity import Activity
+from simpleflow.utils import format_exc
 from simpleflow.workflow import Workflow
 from swf.models.history import builder
 from simpleflow.history import History
@@ -89,9 +90,9 @@ def submit(self, func, *args, **kwargs):
             state = 'completed'
         except Exception as err:
             future._exception = err
-            logger.info('rescuing exception: {}'.format(err))
+            logger.exception('rescuing exception: {}'.format(err))
             if isinstance(func, Activity) and func.raises_on_failure:
-                message = err.args[0] if err.args else ''
+                message = format_exc(err)
                 raise exceptions.TaskFailed(func.name, message)
             state = 'failed'
         finally:

simpleflow/swf/executor.py

@@ -185,8 +185,14 @@ def __init__(self, domain, workflow_class, task_list=None, repair_with=None,
             self.force_activities = re.compile(force_activities)
         else:
             self.force_activities = None
+        self._open_activity_count = 0
+        self._decisions = []
+        self._append_timer = False  # Append an immediate timer decision
+        self._tasks = TaskRegistry()
+        self._idempotent_tasks_to_submit = set()
+        self._execution = None
+        self.current_priority = None
 
-    # noinspection PyAttributeOutsideInit
     def reset(self):
         """
         Clears the state of the execution.
@@ -892,10 +898,12 @@ def replay(self, decision_response, decref_workflow=True):
         """
         self.reset()
 
+        # noinspection PyUnresolvedReferences
         history = decision_response.history
         self._history = History(history)
         self._history.parse()
         self.build_execution_context(decision_response)
+        # noinspection PyUnresolvedReferences
         self._execution = decision_response.execution
 
         workflow_started_event = history[0]
@@ -906,8 +914,17 @@ def replay(self, decision_response, decref_workflow=True):
         kwargs = input.get('kwargs', {})
 
         self.before_replay()
+
         try:
             self.propagate_signals()
+            if self._history.cancel_requested:
+                decisions = self.handle_cancel_requested()
+                if decisions is not None:
+                    self.after_replay()
+                    self.after_closed()
+                    if decref_workflow:
+                        self.decref_workflow()
+                    return decisions
             result = self.run_workflow(*args, **kwargs)
         except exceptions.ExecutionBlocked:
             logger.info('{} open activities ({} decisions)'.format(
@@ -997,6 +1014,9 @@ def on_completed(self):
         except NotImplementedError:
             pass
 
+    def on_canceled(self):
+        self._workflow.on_canceled(self._history)
+
     def fail(self, reason, details=None):
         self.on_failure(reason, details)
 
@@ -1022,11 +1042,13 @@ def build_execution_context(self, decision_response):
         :param decision_response:
         :type  decision_response: swf.responses.Response
         """
+        # noinspection PyUnresolvedReferences
         execution = decision_response.execution
         if not execution:
             # For tests that don't provide an execution object.
             return
 
+        # noinspection PyUnresolvedReferences
         history = decision_response.history
         workflow_started_event = history[0]
         self._execution_context = dict(
@@ -1179,3 +1201,30 @@ def get_event_details(self, event_type, event_name):
             raise ValueError('Unimplemented type {!r} for get_event_details'.format(
                 event_type
             ))
+
+    def handle_cancel_requested(self):
+        decision = swf.models.decision.WorkflowExecutionDecision()
+        is_current_decision = self._history.completed_decision_id < self._history.cancel_requested_id
+        should_cancel = self._workflow.should_cancel(self._history)
+        if not should_cancel:
+            return None  # ignore cancel
+        if is_current_decision:
+            self.on_canceled()
+            decision.cancel()
+            return [decision]
+        if self._history.cancel_failed:
+            logger.warning('failed: %s', self._history.cancel_failed)
+        if self._history.cancel_failed and self._history.cancel_failed_decision_task_completed_event_id == self._history.completed_decision_id:
+            # Per http://docs.aws.amazon.com/amazonswf/latest/apireference/API_Decision.html,
+            # we should call RespondDecisionTaskCompleted without any decisions; however this hangs the workflow...
+
+            # <1 WorkflowExecution : started>, <2 DecisionTask : scheduled>, <3 DecisionTask : started>,
+            # <4 DecisionTask : completed>, <5 ActivityTask : scheduled>, <6 ActivityTask : started>,
+            # <7 WorkflowExecution : cancel_requested>, <8 DecisionTask : scheduled>, <9 DecisionTask : started>,
+            # <10 ActivityTask : completed>, <11 DecisionTask : scheduled>, <12 DecisionTask : completed>,
+            # <13 WorkflowExecution : cancel_failed>, <14 DecisionTask : started>
+
+            # return []
+            pass
+        decision.cancel()
+        return [decision]

simpleflow/workflow.py

@@ -167,6 +167,15 @@ def on_completed(self, history):
         """
         pass
 
+    def on_canceled(self, history):
+        """
+        Method called on canceling the execution.
+
+        :param history:
+        :type history: simpleflow.history.History
+        """
+        pass
+
     def get_execution_context(self):
         """
         Get an execution context from the executor.
@@ -208,3 +217,11 @@ def start_timer(self, timer_id, timeout, control=None):
 
     def cancel_timer(self, timer_id):
         return CancelTimerTask(timer_id)
+
+    def should_cancel(self, history):
+        """
+        Called by the executor if cancel requested.
+        :param history:
+        :return:
+        """
+        return True

tests/integration/cassettes/test_cancel_requested.yaml

@@ -0,0 +1,87 @@
+interactions:
+- request:
+    body: !!python/unicode '{"domain": "TestDomain", "workflowType": {"version": "example",
+      "name": "example"}}'
+    headers:
+      Authorization: ['AWS4-HMAC-SHA256 Credential=1234AB/20160823/us-east-1/swf/aws4_request,SignedHeaders=host;x-amz-date;x-amz-target,Signature=foobar']
+      Content-Encoding: [amz-1.0]
+      Content-Length: ['83']
+      Content-Type: [application/json; charset=UTF-8]
+      Host: [swf.us-east-1.amazonaws.com]
+      User-Agent: [Boto/2.46.1 Python/2.7.12+ Linux/4.8.0-53-generic]
+      X-Amz-Date: [20170613T163306Z]
+      X-Amz-Target: [com.amazonaws.swf.service.model.SimpleWorkflowService.DescribeWorkflowType]
+    method: POST
+    uri: https://swf.us-east-1.amazonaws.com/
+  response:
+    body: {string: !!python/unicode '{"configuration":{"defaultChildPolicy":"TERMINATE","defaultExecutionStartToCloseTimeout":"300","defaultTaskList":{"name":"None"},"defaultTaskStartToCloseTimeout":"300"},"typeInfo":{"creationDate":1.487851963534E9,"status":"REGISTERED","workflowType":{"name":"example","version":"example"}}}'}
+    headers:
+      content-length: ['290']
+      content-type: [application/json]
+      x-amzn-requestid: [fab29807-5055-11e7-b293-fb22db0b9b82]
+    status: {code: 200, message: OK}
+- request:
+    body: !!python/unicode '{"domain": "TestDomain", "taskList": {"name": "test-simpleflow-workflow-5f7b987ba10e4b4db22192c044052319"},
+      "childPolicy": "TERMINATE", "input": "{\"args\":[true],\"kwargs\":{}}", "workflowType":
+      {"version": "example", "name": "example"}, "workflowId": "test-simpleflow-workflow"}'
+    headers:
+      Authorization: ['AWS4-HMAC-SHA256 Credential=1234AB/20160823/us-east-1/swf/aws4_request,SignedHeaders=host;x-amz-date;x-amz-target,Signature=foobar']
+      Content-Encoding: [amz-1.0]
+      Content-Length: ['280']
+      Content-Type: [application/json; charset=UTF-8]
+      Host: [swf.us-east-1.amazonaws.com]
+      User-Agent: [Boto/2.46.1 Python/2.7.12+ Linux/4.8.0-53-generic]
+      X-Amz-Date: [20170613T163306Z]
+      X-Amz-Target: [com.amazonaws.swf.service.model.SimpleWorkflowService.StartWorkflowExecution]
+    method: POST
+    uri: https://swf.us-east-1.amazonaws.com/
+  response:
+    body: {string: !!python/unicode '{"runId":"22X56OQhwfy+AtGx1iSo3HX51AK2DdtKzfUnU+RQElbms="}'}
+    headers:
+      content-length: ['58']
+      content-type: [application/json]
+      x-amzn-requestid: [faeeb73a-5055-11e7-970d-a3e28b0572da]
+    status: {code: 200, message: OK}
+- request:
+    body: !!python/unicode '{"domain": "TestDomain", "execution": {"workflowId": "test-simpleflow-workflow",
+      "runId": "22X56OQhwfy+AtGx1iSo3HX51AK2DdtKzfUnU+RQElbms="}}'
+    headers:
+      Authorization: ['AWS4-HMAC-SHA256 Credential=1234AB/20160823/us-east-1/swf/aws4_request,SignedHeaders=host;x-amz-date;x-amz-target,Signature=foobar']
+      Content-Encoding: [amz-1.0]
+      Content-Length: ['140']
+      Content-Type: [application/json; charset=UTF-8]
+      Host: [swf.us-east-1.amazonaws.com]
+      User-Agent: [Boto/2.46.1 Python/2.7.12+ Linux/4.8.0-53-generic]
+      X-Amz-Date: [20170613T163309Z]
+      X-Amz-Target: [com.amazonaws.swf.service.model.SimpleWorkflowService.DescribeWorkflowExecution]
+    method: POST
+    uri: https://swf.us-east-1.amazonaws.com/
+  response:
+    body: {string: !!python/unicode '{"executionConfiguration":{"childPolicy":"TERMINATE","executionStartToCloseTimeout":"300","taskList":{"name":"test-simpleflow-workflow-5f7b987ba10e4b4db22192c044052319"},"taskStartToCloseTimeout":"300"},"executionInfo":{"cancelRequested":false,"closeStatus":"CANCELED","closeTimestamp":1.49737158914E9,"execution":{"runId":"22X56OQhwfy+AtGx1iSo3HX51AK2DdtKzfUnU+RQElbms=","workflowId":"test-simpleflow-workflow"},"executionStatus":"CLOSED","startTimestamp":1.497371587073E9,"workflowType":{"name":"example","version":"example"}},"latestActivityTaskTimestamp":1.49737158784E9,"openCounts":{"openActivityTasks":1,"openChildWorkflowExecutions":0,"openDecisionTasks":0,"openLambdaFunctions":0,"openTimers":0}}'}
+    headers:
+      content-length: ['705']
+      content-type: [application/json]
+      x-amzn-requestid: [fc630789-5055-11e7-8d87-ed8dd5432c37]
+    status: {code: 200, message: OK}
+- request:
+    body: !!python/unicode '{"domain": "TestDomain", "execution": {"workflowId": "test-simpleflow-workflow",
+      "runId": "22X56OQhwfy+AtGx1iSo3HX51AK2DdtKzfUnU+RQElbms="}}'
+    headers:
+      Authorization: ['AWS4-HMAC-SHA256 Credential=1234AB/20160823/us-east-1/swf/aws4_request,SignedHeaders=host;x-amz-date;x-amz-target,Signature=foobar']
+      Content-Encoding: [amz-1.0]
+      Content-Length: ['140']
+      Content-Type: [application/json; charset=UTF-8]
+      Host: [swf.us-east-1.amazonaws.com]
+      User-Agent: [Boto/2.46.1 Python/2.7.12+ Linux/4.8.0-53-generic]
+      X-Amz-Date: [20170613T163510Z]
+      X-Amz-Target: [com.amazonaws.swf.service.model.SimpleWorkflowService.GetWorkflowExecutionHistory]
+    method: POST
+    uri: https://swf.us-east-1.amazonaws.com/
+  response:
+    body: {string: !!python/unicode '{"events":[{"eventId":1,"eventTimestamp":1.497371587073E9,"eventType":"WorkflowExecutionStarted","workflowExecutionStartedEventAttributes":{"childPolicy":"TERMINATE","executionStartToCloseTimeout":"300","input":"{\"args\":[true],\"kwargs\":{}}","parentInitiatedEventId":0,"taskList":{"name":"test-simpleflow-workflow-5f7b987ba10e4b4db22192c044052319"},"taskStartToCloseTimeout":"300","workflowType":{"name":"example","version":"example"}}},{"decisionTaskScheduledEventAttributes":{"startToCloseTimeout":"300","taskList":{"name":"test-simpleflow-workflow-5f7b987ba10e4b4db22192c044052319"}},"eventId":2,"eventTimestamp":1.497371587073E9,"eventType":"DecisionTaskScheduled"},{"decisionTaskStartedEventAttributes":{"identity":"{\"hostname\":\"zeb-Precision-T3610\",\"pid\":13202,\"user\":\"zeb\"}","scheduledEventId":2},"eventId":3,"eventTimestamp":1.497371587123E9,"eventType":"DecisionTaskStarted"},{"decisionTaskCompletedEventAttributes":{"scheduledEventId":2,"startedEventId":3},"eventId":4,"eventTimestamp":1.497371587661E9,"eventType":"DecisionTaskCompleted"},{"activityTaskScheduledEventAttributes":{"activityId":"activity-tests.integration.workflow.cancel_workflow-1","activityType":{"name":"tests.integration.workflow.cancel_workflow","version":"example"},"decisionTaskCompletedEventId":4,"heartbeatTimeout":"300","input":"{\"args\":[],\"kwargs\":{}}","scheduleToCloseTimeout":"300","scheduleToStartTimeout":"300","startToCloseTimeout":"300","taskList":{"name":"test-simpleflow-workflow-5f7b987ba10e4b4db22192c044052319"}},"eventId":5,"eventTimestamp":1.497371587661E9,"eventType":"ActivityTaskScheduled"},{"activityTaskStartedEventAttributes":{"identity":"{\"hostname\":\"zeb-Precision-T3610\",\"pid\":13203,\"user\":\"zeb\"}","scheduledEventId":5},"eventId":6,"eventTimestamp":1.49737158784E9,"eventType":"ActivityTaskStarted"},{"eventId":7,"eventTimestamp":1.497371588667E9,"eventType":"WorkflowExecutionCancelRequested","workflowExecutionCancelRequestedEventAttributes":{"externalInitiatedEventId":0}},{"decisionTaskScheduledEventAttributes":{"startToCloseTimeout":"300","taskList":{"name":"test-simpleflow-workflow-5f7b987ba10e4b4db22192c044052319"}},"eventId":8,"eventTimestamp":1.497371588667E9,"eventType":"DecisionTaskScheduled"},{"decisionTaskStartedEventAttributes":{"identity":"{\"hostname\":\"zeb-Precision-T3610\",\"pid\":13204,\"user\":\"zeb\"}","scheduledEventId":8},"eventId":9,"eventTimestamp":1.497371588696E9,"eventType":"DecisionTaskStarted"},{"decisionTaskCompletedEventAttributes":{"scheduledEventId":8,"startedEventId":9},"eventId":10,"eventTimestamp":1.49737158914E9,"eventType":"DecisionTaskCompleted"},{"eventId":11,"eventTimestamp":1.49737158914E9,"eventType":"WorkflowExecutionCanceled","workflowExecutionCanceledEventAttributes":{"decisionTaskCompletedEventId":10}}]}'}
+    headers:
+      content-length: ['2803']
+      content-type: [application/json]
+      x-amzn-requestid: [44f5d3ac-5056-11e7-8887-8d61a3f8196e]
+    status: {code: 200, message: OK}
+version: 1