Merge pull request #4531 from StackStorm/fix_orquesta_config_context_bug

Fix a bug with "{{ config_text.foo }}" not working for executions scheduled via Orquesta workflow

Author: Kami

contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py

@@ -68,7 +68,8 @@ def _construct_st2_context(self):
             'st2': {
                 'action_execution_id': str(self.execution.id),
                 'api_url': api_util.get_full_public_api_url(),
-                'user': self.execution.context.get('user', cfg.CONF.system_user.user)
+                'user': self.execution.context.get('user', cfg.CONF.system_user.user),
+                'pack': self.execution.context.get('pack', None)
             }
         }
 

contrib/runners/orquesta_runner/tests/unit/test_basic.py

@@ -138,7 +138,11 @@ def test_run_workflow(self):
                 'workflow_execution_id': str(wf_ex_db.id),
                 'action_execution_id': str(ac_ex_db.id),
                 'api_url': 'http://127.0.0.1/v1',
-                'user': username
+                'user': username,
+                'pack': 'orquesta_tests'
+            },
+            'parent': {
+                'pack': 'orquesta_tests'
             }
         }
 
@@ -294,6 +298,40 @@ def test_run_workflow_with_unicode_input(self):
         self.assertDictEqual(lv_ac_db.result, expected_result)
         self.assertDictEqual(ac_ex_db.result, expected_result)
 
+    def test_run_workflow_action_config_context(self):
+        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'config-context.yaml')
+        wf_input = {}
+        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
+        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
+
+        # Assert action execution is running.
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)
+        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
+        self.assertEqual(wf_ex_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
+
+        # Assert task1 is already completed.
+        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
+        tk1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
+        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction['id'])
+        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+        self.assertTrue(wf_svc.is_action_execution_under_workflow_context(tk1_ac_ex_db))
+
+        # Manually handle action execution completion.
+        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)
+
+        # Assert workflow is completed.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        self.assertEqual(wf_ex_db.status, wf_states.SUCCEEDED)
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
+        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+
+        # Verify config_context works
+        self.assertEqual(wf_ex_db.output, {'msg': 'value of config key a'})
+
     def test_run_workflow_with_action_less_tasks(self):
         wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'action-less-tasks.yaml')
         wf_input = {'name': 'Thanos'}

contrib/runners/orquesta_runner/tests/unit/test_context.py

@@ -115,7 +115,8 @@ def test_runtime_context(self):
         expected_st2_ctx = {
             'action_execution_id': str(ac_ex_db.id),
             'api_url': 'http://127.0.0.1/v1',
-            'user': 'stanley'
+            'user': 'stanley',
+            'pack': 'orquesta_tests'
         }
 
         expected_st2_ctx_with_wf_ex_id = copy.deepcopy(expected_st2_ctx)

st2common/st2common/services/action.py

@@ -76,11 +76,11 @@ def create_request(liveaction, action_db=None, runnertype_db=None):
     # Use the user context from the parent action execution. Subtasks in a workflow
     # action can be invoked by a system user and so we want to use the user context
     # from the original workflow action.
-    parent_context = executions.get_parent_context(liveaction)
-    if parent_context:
-        parent_user = parent_context.get('user', None)
-        if parent_user:
-            liveaction.context['user'] = parent_user
+    parent_context = executions.get_parent_context(liveaction) or {}
+    parent_user = parent_context.get('user', None)
+
+    if parent_user:
+        liveaction.context['user'] = parent_user
 
     # Validate action
     if not action_db:
@@ -97,6 +97,9 @@ def create_request(liveaction, action_db=None, runnertype_db=None):
     if not hasattr(liveaction, 'parameters'):
         liveaction.parameters = dict()
 
+    # For consistency add pack to the context here in addition to RunnerContainer.dispatch() method
+    liveaction.context['pack'] = action_db.pack
+
     # Validate action parameters.
     schema = util_schema.get_schema_for_action_parameters(action_db, runnertype_db)
     validator = util_schema.get_validator()

st2common/st2common/services/workflows.py

@@ -537,6 +537,7 @@ def request_action_execution(wf_ex_db, task_ex_db, st2_ctx, ac_ex_req, delay=Non
 
     # Set context for the action execution.
     ac_ex_ctx = {
+        'pack': st2_ctx.get('pack'),
         'user': st2_ctx.get('user'),
         'parent': st2_ctx,
         'orquesta': {
@@ -887,7 +888,11 @@ def request_next_tasks(wf_ex_db, task_ex_id=None):
 
                 # Pass down appropriate st2 context to the task and action execution(s).
                 root_st2_ctx = wf_ex_db.context.get('st2', {})
-                st2_ctx = {'execution_id': wf_ac_ex_id, 'user': root_st2_ctx.get('user')}
+                st2_ctx = {
+                    'execution_id': wf_ac_ex_id,
+                    'user': root_st2_ctx.get('user'),
+                    'pack': root_st2_ctx.get('pack')
+                }
                 if root_st2_ctx.get('api_user'):
                     st2_ctx['api_user'] = root_st2_ctx.get('api_user')
 

st2common/st2common/util/config_loader.py

@@ -224,7 +224,7 @@ def _get_datastore_value_for_expression(self, key, value, config_schema_item=Non
 def get_config(pack, user):
     """Returns config for given pack and user.
     """
-    LOG.debug('Attempting to get config')
+    LOG.debug('Attempting to get config for pack "%s" and user "%s"' % (pack, user))
     if pack and user:
         LOG.debug('Pack and user found. Loading config.')
         config_loader = ContentPackConfigLoader(

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/config-context-action.yaml

@@ -0,0 +1,9 @@
+---
+  name: "config-context-action"
+  runner_type: "local-shell-cmd"
+  enabled: true
+  entry_point: ""
+  parameters:
+    cmd:
+      immutable: true
+      default: "echo \"{{ config_context.config_key_a }}\""

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/config-context.yaml

@@ -0,0 +1,7 @@
+---
+name: config-context
+description: Workflow which tests {{ config_context.foo }} notation works default parameter values for workflow actions.
+pack: orquesta_tests
+runner_type: orquesta
+entry_point: workflows/config-context.yaml
+enabled: true

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/workflows/config-context.yaml

@@ -0,0 +1,13 @@
+version: 1.0
+
+description: Workflow which tests {{ config_context }} functionality.
+
+output:
+  - msg: <% ctx().message %>
+
+tasks:
+  task1:
+    action: orquesta_tests.config-context-action
+    next:
+      - when: <% succeeded() %>
+        publish: message=<% result().stdout %>

st2tests/st2tests/fixtures/packs/orquesta_tests/config.schema.yaml

@@ -0,0 +1,6 @@
+---
+config_key_a:
+  description: "Sample config key."
+  type: "string"
+  default: "value of config key a"
+  required: true