We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Orquesta workflow execution is stuck with canceling status
##### SUMMARY

If user cancels an orquesta workflow execution immediately after it was launched, there is a possibility where the workflow execution will be stuck in a canceling status. The cancelation failed because the workflow execution record has not been created yet.

##### ISSUE TYPE
 - Bug Report

##### STACKSTORM VERSION
st2 v2.8 -> v3.1

##### OS / ENVIRONMENT / INSTALL METHOD
This affects all distros and installs.

##### STEPS TO REPRODUCE
Launch an orquesta workflow execution with one or more nested subworkflows. Cancel the workflow execution almostly immediately after it was launched.

##### EXPECTED RESULTS
Workflow execution should be in canceled state.

##### ACTUAL RESULTS
Workflow execution is stuck in canceling state.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py`
Content:
```
1 # Copyright 2019 Extreme Networks, Inc.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from __future__ import absolute_import
16 
17 import uuid
18 
19 import six
20 from oslo_config import cfg
21 
22 from orquesta import exceptions as wf_exc
23 from orquesta import statuses as wf_statuses
24 
25 from st2common.constants import action as ac_const
26 from st2common import log as logging
27 from st2common.models.api import notification as notify_api_models
28 from st2common.persistence import execution as ex_db_access
29 from st2common.persistence import liveaction as lv_db_access
30 from st2common.runners import base as runners
31 from st2common.services import action as ac_svc
32 from st2common.services import workflows as wf_svc
33 from st2common.util import api as api_util
34 from st2common.util import ujson
35 
36 __all__ = [
37     'OrquestaRunner',
38     'get_runner',
39     'get_metadata'
40 ]
41 
42 
43 LOG = logging.getLogger(__name__)
44 
45 
46 class OrquestaRunner(runners.AsyncActionRunner):
47 
48     @staticmethod
49     def get_workflow_definition(entry_point):
50         with open(entry_point, 'r') as def_file:
51             return def_file.read()
52 
53     def _get_notify_config(self):
54         return (
55             notify_api_models.NotificationsHelper.from_model(notify_model=self.liveaction.notify)
56             if self.liveaction.notify
57             else None
58         )
59 
60     def _construct_context(self, wf_ex):
61         ctx = ujson.fast_deepcopy(self.context)
62         ctx['workflow_execution'] = str(wf_ex.id)
63 
64         return ctx
65 
66     def _construct_st2_context(self):
67         st2_ctx = {
68             'st2': {
69                 'action_execution_id': str(self.execution.id),
70                 'api_url': api_util.get_full_public_api_url(),
71                 'user': self.execution.context.get('user', cfg.CONF.system_user.user),
72                 'pack': self.execution.context.get('pack', None)
73             }
74         }
75 
76         if self.execution.context.get('api_user'):
77             st2_ctx['st2']['api_user'] = self.execution.context.get('api_user')
78 
79         if self.execution.context.get('source_channel'):
80             st2_ctx['st2']['source_channel'] = self.execution.context.get('source_channel')
81 
82         if self.execution.context:
83             st2_ctx['parent'] = self.execution.context
84 
85         return st2_ctx
86 
87     def run(self, action_parameters):
88         # Read workflow definition from file.
89         wf_def = self.get_workflow_definition(self.entry_point)
90 
91         try:
92             # Request workflow execution.
93             st2_ctx = self._construct_st2_context()
94             notify_cfg = self._get_notify_config()
95             wf_ex_db = wf_svc.request(wf_def, self.execution, st2_ctx, notify_cfg)
96         except wf_exc.WorkflowInspectionError as e:
97             status = ac_const.LIVEACTION_STATUS_FAILED
98             result = {'errors': e.args[1], 'output': None}
99             return (status, result, self.context)
100         except Exception as e:
101             status = ac_const.LIVEACTION_STATUS_FAILED
102             result = {'errors': [{'message': six.text_type(e)}], 'output': None}
103             return (status, result, self.context)
104 
105         if wf_ex_db.status in wf_statuses.COMPLETED_STATUSES:
106             status = wf_ex_db.status
107             result = {'output': wf_ex_db.output or None}
108 
109             if wf_ex_db.status in wf_statuses.ABENDED_STATUSES:
110                 result['errors'] = wf_ex_db.errors
111 
112             for wf_ex_error in wf_ex_db.errors:
113                 msg = '[%s] Workflow execution completed with errors.'
114                 LOG.error(msg, str(self.execution.id), extra=wf_ex_error)
115 
116             return (status, result, self.context)
117 
118         # Set return values.
119         status = ac_const.LIVEACTION_STATUS_RUNNING
120         partial_results = {}
121         ctx = self._construct_context(wf_ex_db)
122 
123         return (status, partial_results, ctx)
124 
125     @staticmethod
126     def task_pauseable(ac_ex):
127         wf_ex_pauseable = (
128             ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and
129             ac_ex.status == ac_const.LIVEACTION_STATUS_RUNNING
130         )
131 
132         return wf_ex_pauseable
133 
134     def pause(self):
135         # Pause the target workflow.
136         wf_ex_db = wf_svc.request_pause(self.execution)
137 
138         # Request pause of tasks that are workflows and still running.
139         for child_ex_id in self.execution.children:
140             child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)
141             if self.task_pauseable(child_ex):
142                 ac_svc.request_pause(
143                     lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),
144                     self.context.get('user', None)
145                 )
146 
147         if wf_ex_db.status == wf_statuses.PAUSING or ac_svc.is_children_active(self.liveaction.id):
148             status = ac_const.LIVEACTION_STATUS_PAUSING
149         else:
150             status = ac_const.LIVEACTION_STATUS_PAUSED
151 
152         return (
153             status,
154             self.liveaction.result,
155             self.liveaction.context
156         )
157 
158     @staticmethod
159     def task_resumeable(ac_ex):
160         wf_ex_resumeable = (
161             ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and
162             ac_ex.status == ac_const.LIVEACTION_STATUS_PAUSED
163         )
164 
165         return wf_ex_resumeable
166 
167     def resume(self):
168         # Resume the target workflow.
169         wf_ex_db = wf_svc.request_resume(self.execution)
170 
171         # Request resume of tasks that are workflows and still running.
172         for child_ex_id in self.execution.children:
173             child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)
174             if self.task_resumeable(child_ex):
175                 ac_svc.request_resume(
176                     lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),
177                     self.context.get('user', None)
178                 )
179 
180         return (
181             wf_ex_db.status if wf_ex_db else ac_const.LIVEACTION_STATUS_RUNNING,
182             self.liveaction.result,
183             self.liveaction.context
184         )
185 
186     @staticmethod
187     def task_cancelable(ac_ex):
188         wf_ex_cancelable = (
189             ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and
190             ac_ex.status in ac_const.LIVEACTION_CANCELABLE_STATES
191         )
192 
193         ac_ex_cancelable = (
194             ac_ex.runner['name'] not in ac_const.WORKFLOW_RUNNER_TYPES and
195             ac_ex.status in ac_const.LIVEACTION_DELAYED_STATES
196         )
197 
198         return wf_ex_cancelable or ac_ex_cancelable
199 
200     def cancel(self):
201         # Cancel the target workflow.
202         wf_svc.request_cancellation(self.execution)
203 
204         # Request cancellation of tasks that are workflows and still running.
205         for child_ex_id in self.execution.children:
206             child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)
207             if self.task_cancelable(child_ex):
208                 ac_svc.request_cancellation(
209                     lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),
210                     self.context.get('user', None)
211                 )
212 
213         status = (
214             ac_const.LIVEACTION_STATUS_CANCELING
215             if ac_svc.is_children_active(self.liveaction.id)
216             else ac_const.LIVEACTION_STATUS_CANCELED
217         )
218 
219         return (
220             status,
221             self.liveaction.result,
222             self.liveaction.context
223         )
224 
225 
226 def get_runner():
227     return OrquestaRunner(str(uuid.uuid4()))
228 
229 
230 def get_metadata():
231     return runners.get_metadata('orquesta_runner')[0]
232
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py b/contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py
--- a/contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py
+++ b/contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py
@@ -14,6 +14,8 @@
 
 from __future__ import absolute_import
 
+import sys
+import traceback
 import uuid
 
 import six
@@ -24,6 +26,7 @@
 
 from st2common.constants import action as ac_const
 from st2common import log as logging
+from st2common.exceptions import workflow as wf_svc_exc
 from st2common.models.api import notification as notify_api_models
 from st2common.persistence import execution as ex_db_access
 from st2common.persistence import liveaction as lv_db_access
@@ -198,8 +201,29 @@
         return wf_ex_cancelable or ac_ex_cancelable
 
     def cancel(self):
-        # Cancel the target workflow.
-        wf_svc.request_cancellation(self.execution)
+        result = None
+
+        # Try to cancel the target workflow execution.
+        try:
+            wf_svc.request_cancellation(self.execution)
+        # If workflow execution is not found because the action execution is cancelled
+        # before the workflow execution is created or if the workflow execution is
+        # already completed, then ignore the exception and proceed with cancellation.
+        except (wf_svc_exc.WorkflowExecutionNotFoundException,
+                wf_svc_exc.WorkflowExecutionIsCompletedException):
+            pass
+        # If there is an unknown exception, then log the error. Continue with the
+        # cancelation sequence below to cancel children and determine final status.
+        # If we rethrow the exception here, the workflow will be stuck in a canceling
+        # state with no options for user to clean up. It is safer to continue with
+        # the cancel then to revert back to some other statuses because the workflow
+        # execution will be in an unknown state.
+        except Exception:
+            _, ex, tb = sys.exc_info()
+            msg = 'Error encountered when canceling workflow execution. %s'
+            result = {'error': msg % str(ex), 'traceback': ''.join(traceback.format_tb(tb, 20))}
+            msg = '[%s] Error encountered when canceling workflow execution.'
+            LOG.exception(msg, str(self.execution.id))
 
         # Request cancellation of tasks that are workflows and still running.
         for child_ex_id in self.execution.children:
@@ -218,7 +242,7 @@
 
         return (
             status,
-            self.liveaction.result,
+            result if result else self.liveaction.result,
             self.liveaction.context
         )

{"golden_diff": "diff --git a/contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py b/contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py\n--- a/contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py\n+++ b/contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py\n@@ -14,6 +14,8 @@\n \n from __future__ import absolute_import\n \n+import sys\n+import traceback\n import uuid\n \n import six\n@@ -24,6 +26,7 @@\n \n from st2common.constants import action as ac_const\n from st2common import log as logging\n+from st2common.exceptions import workflow as wf_svc_exc\n from st2common.models.api import notification as notify_api_models\n from st2common.persistence import execution as ex_db_access\n from st2common.persistence import liveaction as lv_db_access\n@@ -198,8 +201,29 @@\n         return wf_ex_cancelable or ac_ex_cancelable\n \n     def cancel(self):\n-        # Cancel the target workflow.\n-        wf_svc.request_cancellation(self.execution)\n+        result = None\n+\n+        # Try to cancel the target workflow execution.\n+        try:\n+            wf_svc.request_cancellation(self.execution)\n+        # If workflow execution is not found because the action execution is cancelled\n+        # before the workflow execution is created or if the workflow execution is\n+        # already completed, then ignore the exception and proceed with cancellation.\n+        except (wf_svc_exc.WorkflowExecutionNotFoundException,\n+                wf_svc_exc.WorkflowExecutionIsCompletedException):\n+            pass\n+        # If there is an unknown exception, then log the error. Continue with the\n+        # cancelation sequence below to cancel children and determine final status.\n+        # If we rethrow the exception here, the workflow will be stuck in a canceling\n+        # state with no options for user to clean up. It is safer to continue with\n+        # the cancel then to revert back to some other statuses because the workflow\n+        # execution will be in an unknown state.\n+        except Exception:\n+            _, ex, tb = sys.exc_info()\n+            msg = 'Error encountered when canceling workflow execution. %s'\n+            result = {'error': msg % str(ex), 'traceback': ''.join(traceback.format_tb(tb, 20))}\n+            msg = '[%s] Error encountered when canceling workflow execution.'\n+            LOG.exception(msg, str(self.execution.id))\n \n         # Request cancellation of tasks that are workflows and still running.\n         for child_ex_id in self.execution.children:\n@@ -218,7 +242,7 @@\n \n         return (\n             status,\n-            self.liveaction.result,\n+            result if result else self.liveaction.result,\n             self.liveaction.context\n         )\n", "issue": "Orquesta workflow execution is stuck with canceling status\n##### SUMMARY\r\n\r\nIf user cancels an orquesta workflow execution immediately after it was launched, there is a possibility where the workflow execution will be stuck in a canceling status. The cancelation failed because the workflow execution record has not been created yet.\r\n\r\n##### ISSUE TYPE\r\n - Bug Report\r\n\r\n##### STACKSTORM VERSION\r\nst2 v2.8 -> v3.1\r\n\r\n##### OS / ENVIRONMENT / INSTALL METHOD\r\nThis affects all distros and installs.\r\n\r\n##### STEPS TO REPRODUCE\r\nLaunch an orquesta workflow execution with one or more nested subworkflows. Cancel the workflow execution almostly immediately after it was launched.\r\n\r\n##### EXPECTED RESULTS\r\nWorkflow execution should be in canceled state.\r\n\r\n##### ACTUAL RESULTS\r\nWorkflow execution is stuck in canceling state.\n", "before_files": [{"content": "# Copyright 2019 Extreme Networks, Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import absolute_import\n\nimport uuid\n\nimport six\nfrom oslo_config import cfg\n\nfrom orquesta import exceptions as wf_exc\nfrom orquesta import statuses as wf_statuses\n\nfrom st2common.constants import action as ac_const\nfrom st2common import log as logging\nfrom st2common.models.api import notification as notify_api_models\nfrom st2common.persistence import execution as ex_db_access\nfrom st2common.persistence import liveaction as lv_db_access\nfrom st2common.runners import base as runners\nfrom st2common.services import action as ac_svc\nfrom st2common.services import workflows as wf_svc\nfrom st2common.util import api as api_util\nfrom st2common.util import ujson\n\n__all__ = [\n    'OrquestaRunner',\n    'get_runner',\n    'get_metadata'\n]\n\n\nLOG = logging.getLogger(__name__)\n\n\nclass OrquestaRunner(runners.AsyncActionRunner):\n\n    @staticmethod\n    def get_workflow_definition(entry_point):\n        with open(entry_point, 'r') as def_file:\n            return def_file.read()\n\n    def _get_notify_config(self):\n        return (\n            notify_api_models.NotificationsHelper.from_model(notify_model=self.liveaction.notify)\n            if self.liveaction.notify\n            else None\n        )\n\n    def _construct_context(self, wf_ex):\n        ctx = ujson.fast_deepcopy(self.context)\n        ctx['workflow_execution'] = str(wf_ex.id)\n\n        return ctx\n\n    def _construct_st2_context(self):\n        st2_ctx = {\n            'st2': {\n                'action_execution_id': str(self.execution.id),\n                'api_url': api_util.get_full_public_api_url(),\n                'user': self.execution.context.get('user', cfg.CONF.system_user.user),\n                'pack': self.execution.context.get('pack', None)\n            }\n        }\n\n        if self.execution.context.get('api_user'):\n            st2_ctx['st2']['api_user'] = self.execution.context.get('api_user')\n\n        if self.execution.context.get('source_channel'):\n            st2_ctx['st2']['source_channel'] = self.execution.context.get('source_channel')\n\n        if self.execution.context:\n            st2_ctx['parent'] = self.execution.context\n\n        return st2_ctx\n\n    def run(self, action_parameters):\n        # Read workflow definition from file.\n        wf_def = self.get_workflow_definition(self.entry_point)\n\n        try:\n            # Request workflow execution.\n            st2_ctx = self._construct_st2_context()\n            notify_cfg = self._get_notify_config()\n            wf_ex_db = wf_svc.request(wf_def, self.execution, st2_ctx, notify_cfg)\n        except wf_exc.WorkflowInspectionError as e:\n            status = ac_const.LIVEACTION_STATUS_FAILED\n            result = {'errors': e.args[1], 'output': None}\n            return (status, result, self.context)\n        except Exception as e:\n            status = ac_const.LIVEACTION_STATUS_FAILED\n            result = {'errors': [{'message': six.text_type(e)}], 'output': None}\n            return (status, result, self.context)\n\n        if wf_ex_db.status in wf_statuses.COMPLETED_STATUSES:\n            status = wf_ex_db.status\n            result = {'output': wf_ex_db.output or None}\n\n            if wf_ex_db.status in wf_statuses.ABENDED_STATUSES:\n                result['errors'] = wf_ex_db.errors\n\n            for wf_ex_error in wf_ex_db.errors:\n                msg = '[%s] Workflow execution completed with errors.'\n                LOG.error(msg, str(self.execution.id), extra=wf_ex_error)\n\n            return (status, result, self.context)\n\n        # Set return values.\n        status = ac_const.LIVEACTION_STATUS_RUNNING\n        partial_results = {}\n        ctx = self._construct_context(wf_ex_db)\n\n        return (status, partial_results, ctx)\n\n    @staticmethod\n    def task_pauseable(ac_ex):\n        wf_ex_pauseable = (\n            ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and\n            ac_ex.status == ac_const.LIVEACTION_STATUS_RUNNING\n        )\n\n        return wf_ex_pauseable\n\n    def pause(self):\n        # Pause the target workflow.\n        wf_ex_db = wf_svc.request_pause(self.execution)\n\n        # Request pause of tasks that are workflows and still running.\n        for child_ex_id in self.execution.children:\n            child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)\n            if self.task_pauseable(child_ex):\n                ac_svc.request_pause(\n                    lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),\n                    self.context.get('user', None)\n                )\n\n        if wf_ex_db.status == wf_statuses.PAUSING or ac_svc.is_children_active(self.liveaction.id):\n            status = ac_const.LIVEACTION_STATUS_PAUSING\n        else:\n            status = ac_const.LIVEACTION_STATUS_PAUSED\n\n        return (\n            status,\n            self.liveaction.result,\n            self.liveaction.context\n        )\n\n    @staticmethod\n    def task_resumeable(ac_ex):\n        wf_ex_resumeable = (\n            ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and\n            ac_ex.status == ac_const.LIVEACTION_STATUS_PAUSED\n        )\n\n        return wf_ex_resumeable\n\n    def resume(self):\n        # Resume the target workflow.\n        wf_ex_db = wf_svc.request_resume(self.execution)\n\n        # Request resume of tasks that are workflows and still running.\n        for child_ex_id in self.execution.children:\n            child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)\n            if self.task_resumeable(child_ex):\n                ac_svc.request_resume(\n                    lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),\n                    self.context.get('user', None)\n                )\n\n        return (\n            wf_ex_db.status if wf_ex_db else ac_const.LIVEACTION_STATUS_RUNNING,\n            self.liveaction.result,\n            self.liveaction.context\n        )\n\n    @staticmethod\n    def task_cancelable(ac_ex):\n        wf_ex_cancelable = (\n            ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and\n            ac_ex.status in ac_const.LIVEACTION_CANCELABLE_STATES\n        )\n\n        ac_ex_cancelable = (\n            ac_ex.runner['name'] not in ac_const.WORKFLOW_RUNNER_TYPES and\n            ac_ex.status in ac_const.LIVEACTION_DELAYED_STATES\n        )\n\n        return wf_ex_cancelable or ac_ex_cancelable\n\n    def cancel(self):\n        # Cancel the target workflow.\n        wf_svc.request_cancellation(self.execution)\n\n        # Request cancellation of tasks that are workflows and still running.\n        for child_ex_id in self.execution.children:\n            child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)\n            if self.task_cancelable(child_ex):\n                ac_svc.request_cancellation(\n                    lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),\n                    self.context.get('user', None)\n                )\n\n        status = (\n            ac_const.LIVEACTION_STATUS_CANCELING\n            if ac_svc.is_children_active(self.liveaction.id)\n            else ac_const.LIVEACTION_STATUS_CANCELED\n        )\n\n        return (\n            status,\n            self.liveaction.result,\n            self.liveaction.context\n        )\n\n\ndef get_runner():\n    return OrquestaRunner(str(uuid.uuid4()))\n\n\ndef get_metadata():\n    return runners.get_metadata('orquesta_runner')[0]\n", "path": "contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py"}], "after_files": [{"content": "# Copyright 2019 Extreme Networks, Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import absolute_import\n\nimport sys\nimport traceback\nimport uuid\n\nimport six\nfrom oslo_config import cfg\n\nfrom orquesta import exceptions as wf_exc\nfrom orquesta import statuses as wf_statuses\n\nfrom st2common.constants import action as ac_const\nfrom st2common import log as logging\nfrom st2common.exceptions import workflow as wf_svc_exc\nfrom st2common.models.api import notification as notify_api_models\nfrom st2common.persistence import execution as ex_db_access\nfrom st2common.persistence import liveaction as lv_db_access\nfrom st2common.runners import base as runners\nfrom st2common.services import action as ac_svc\nfrom st2common.services import workflows as wf_svc\nfrom st2common.util import api as api_util\nfrom st2common.util import ujson\n\n__all__ = [\n    'OrquestaRunner',\n    'get_runner',\n    'get_metadata'\n]\n\n\nLOG = logging.getLogger(__name__)\n\n\nclass OrquestaRunner(runners.AsyncActionRunner):\n\n    @staticmethod\n    def get_workflow_definition(entry_point):\n        with open(entry_point, 'r') as def_file:\n            return def_file.read()\n\n    def _get_notify_config(self):\n        return (\n            notify_api_models.NotificationsHelper.from_model(notify_model=self.liveaction.notify)\n            if self.liveaction.notify\n            else None\n        )\n\n    def _construct_context(self, wf_ex):\n        ctx = ujson.fast_deepcopy(self.context)\n        ctx['workflow_execution'] = str(wf_ex.id)\n\n        return ctx\n\n    def _construct_st2_context(self):\n        st2_ctx = {\n            'st2': {\n                'action_execution_id': str(self.execution.id),\n                'api_url': api_util.get_full_public_api_url(),\n                'user': self.execution.context.get('user', cfg.CONF.system_user.user),\n                'pack': self.execution.context.get('pack', None)\n            }\n        }\n\n        if self.execution.context.get('api_user'):\n            st2_ctx['st2']['api_user'] = self.execution.context.get('api_user')\n\n        if self.execution.context.get('source_channel'):\n            st2_ctx['st2']['source_channel'] = self.execution.context.get('source_channel')\n\n        if self.execution.context:\n            st2_ctx['parent'] = self.execution.context\n\n        return st2_ctx\n\n    def run(self, action_parameters):\n        # Read workflow definition from file.\n        wf_def = self.get_workflow_definition(self.entry_point)\n\n        try:\n            # Request workflow execution.\n            st2_ctx = self._construct_st2_context()\n            notify_cfg = self._get_notify_config()\n            wf_ex_db = wf_svc.request(wf_def, self.execution, st2_ctx, notify_cfg)\n        except wf_exc.WorkflowInspectionError as e:\n            status = ac_const.LIVEACTION_STATUS_FAILED\n            result = {'errors': e.args[1], 'output': None}\n            return (status, result, self.context)\n        except Exception as e:\n            status = ac_const.LIVEACTION_STATUS_FAILED\n            result = {'errors': [{'message': six.text_type(e)}], 'output': None}\n            return (status, result, self.context)\n\n        if wf_ex_db.status in wf_statuses.COMPLETED_STATUSES:\n            status = wf_ex_db.status\n            result = {'output': wf_ex_db.output or None}\n\n            if wf_ex_db.status in wf_statuses.ABENDED_STATUSES:\n                result['errors'] = wf_ex_db.errors\n\n            for wf_ex_error in wf_ex_db.errors:\n                msg = '[%s] Workflow execution completed with errors.'\n                LOG.error(msg, str(self.execution.id), extra=wf_ex_error)\n\n            return (status, result, self.context)\n\n        # Set return values.\n        status = ac_const.LIVEACTION_STATUS_RUNNING\n        partial_results = {}\n        ctx = self._construct_context(wf_ex_db)\n\n        return (status, partial_results, ctx)\n\n    @staticmethod\n    def task_pauseable(ac_ex):\n        wf_ex_pauseable = (\n            ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and\n            ac_ex.status == ac_const.LIVEACTION_STATUS_RUNNING\n        )\n\n        return wf_ex_pauseable\n\n    def pause(self):\n        # Pause the target workflow.\n        wf_ex_db = wf_svc.request_pause(self.execution)\n\n        # Request pause of tasks that are workflows and still running.\n        for child_ex_id in self.execution.children:\n            child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)\n            if self.task_pauseable(child_ex):\n                ac_svc.request_pause(\n                    lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),\n                    self.context.get('user', None)\n                )\n\n        if wf_ex_db.status == wf_statuses.PAUSING or ac_svc.is_children_active(self.liveaction.id):\n            status = ac_const.LIVEACTION_STATUS_PAUSING\n        else:\n            status = ac_const.LIVEACTION_STATUS_PAUSED\n\n        return (\n            status,\n            self.liveaction.result,\n            self.liveaction.context\n        )\n\n    @staticmethod\n    def task_resumeable(ac_ex):\n        wf_ex_resumeable = (\n            ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and\n            ac_ex.status == ac_const.LIVEACTION_STATUS_PAUSED\n        )\n\n        return wf_ex_resumeable\n\n    def resume(self):\n        # Resume the target workflow.\n        wf_ex_db = wf_svc.request_resume(self.execution)\n\n        # Request resume of tasks that are workflows and still running.\n        for child_ex_id in self.execution.children:\n            child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)\n            if self.task_resumeable(child_ex):\n                ac_svc.request_resume(\n                    lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),\n                    self.context.get('user', None)\n                )\n\n        return (\n            wf_ex_db.status if wf_ex_db else ac_const.LIVEACTION_STATUS_RUNNING,\n            self.liveaction.result,\n            self.liveaction.context\n        )\n\n    @staticmethod\n    def task_cancelable(ac_ex):\n        wf_ex_cancelable = (\n            ac_ex.runner['name'] in ac_const.WORKFLOW_RUNNER_TYPES and\n            ac_ex.status in ac_const.LIVEACTION_CANCELABLE_STATES\n        )\n\n        ac_ex_cancelable = (\n            ac_ex.runner['name'] not in ac_const.WORKFLOW_RUNNER_TYPES and\n            ac_ex.status in ac_const.LIVEACTION_DELAYED_STATES\n        )\n\n        return wf_ex_cancelable or ac_ex_cancelable\n\n    def cancel(self):\n        result = None\n\n        # Try to cancel the target workflow execution.\n        try:\n            wf_svc.request_cancellation(self.execution)\n        # If workflow execution is not found because the action execution is cancelled\n        # before the workflow execution is created or if the workflow execution is\n        # already completed, then ignore the exception and proceed with cancellation.\n        except (wf_svc_exc.WorkflowExecutionNotFoundException,\n                wf_svc_exc.WorkflowExecutionIsCompletedException):\n            pass\n        # If there is an unknown exception, then log the error. Continue with the\n        # cancelation sequence below to cancel children and determine final status.\n        # If we rethrow the exception here, the workflow will be stuck in a canceling\n        # state with no options for user to clean up. It is safer to continue with\n        # the cancel then to revert back to some other statuses because the workflow\n        # execution will be in an unknown state.\n        except Exception:\n            _, ex, tb = sys.exc_info()\n            msg = 'Error encountered when canceling workflow execution. %s'\n            result = {'error': msg % str(ex), 'traceback': ''.join(traceback.format_tb(tb, 20))}\n            msg = '[%s] Error encountered when canceling workflow execution.'\n            LOG.exception(msg, str(self.execution.id))\n\n        # Request cancellation of tasks that are workflows and still running.\n        for child_ex_id in self.execution.children:\n            child_ex = ex_db_access.ActionExecution.get(id=child_ex_id)\n            if self.task_cancelable(child_ex):\n                ac_svc.request_cancellation(\n                    lv_db_access.LiveAction.get(id=child_ex.liveaction['id']),\n                    self.context.get('user', None)\n                )\n\n        status = (\n            ac_const.LIVEACTION_STATUS_CANCELING\n            if ac_svc.is_children_active(self.liveaction.id)\n            else ac_const.LIVEACTION_STATUS_CANCELED\n        )\n\n        return (\n            status,\n            result if result else self.liveaction.result,\n            self.liveaction.context\n        )\n\n\ndef get_runner():\n    return OrquestaRunner(str(uuid.uuid4()))\n\n\ndef get_metadata():\n    return runners.get_metadata('orquesta_runner')[0]\n", "path": "contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py"}]}