Implement runner, API, and client code for Inquiries

CHANGELOG.rst

@@ -37,7 +37,12 @@ Added
   #3763
 * Add new Jinja filters ``from_json_string``, ``from_yaml_string``, and ``jsonpath_query``.
   #3763
-
+* Add new "Inquiry" capability, which adds ability to "ask a question", usually in a workflow.
+  Create a new runner type: "inquirer" to support this, as well as new API endpoints and
+  client commands for interacting with Inquiries
+
+  Contributed by mierdin. #3653
+
 Fixed
 ~~~~~
 

contrib/core/actions/ask.yaml

@@ -0,0 +1,5 @@
+---
+name: "ask"
+runner_type: "inquirer"
+description: "Action for initiating an Inquiry (usually in a workflow)"
+enabled: true

contrib/examples/actions/chain-test-inquiry-parent.yaml

@@ -0,0 +1,6 @@
+---
+name: chain-test-inquiry-parent
+description: A workflow to test Inquiry pausing nested actionchain workflows
+runner_type: action-chain
+entry_point: chains/chain_test_inquiry_parent.yaml
+enabled: true

contrib/examples/actions/chain-test-inquiry.yaml

@@ -0,0 +1,6 @@
+---
+name: chain-test-inquiry
+description: A basic ActionChain workflow illustrating the use of Inquiries
+runner_type: action-chain
+entry_point: chains/chain_test_inquiry.yaml
+enabled: true

contrib/examples/actions/chains/chain_test_inquiry.yaml

@@ -0,0 +1,19 @@
+chain:
+
+  - name: task1
+    ref: core.ask
+    params:
+      route: developers
+      schema:
+        type: object
+        properties:
+          secondfactor:
+            type: string
+            description: Please enter second factor for authenticating to "foo" service
+            required: True
+    on-success: "task2"
+
+  - name: task2
+    ref: core.local
+    params:
+      cmd: echo "We can now authenticate to "foo" service with {{ task1.result.response.secondfactor }}"

contrib/examples/actions/chains/chain_test_inquiry_parent.yaml

@@ -0,0 +1,10 @@
+chain:
+
+    - name: task1
+      ref: examples.chain-test-inquiry
+      on-success: "task2"
+
+    - name: task2
+      ref: core.local
+      params:
+        cmd: echo "parent task2"

contrib/examples/actions/mistral-ask-basic.meta.yaml

@@ -0,0 +1,7 @@
+---
+description: A basic Mistral workflow illustrating the use of Inquiries
+enabled: true
+entry_point: workflows/mistral-ask-basic.yaml
+name: mistral-ask-basic
+pack: examples
+runner_type: mistral-v2

contrib/examples/actions/mistral-ask-parent.meta.yaml

@@ -0,0 +1,8 @@
+---
+description: >
+    Workflow for demonstrating "ask" functionality
+enabled: true
+entry_point: workflows/mistral-ask-parent.yaml
+name: mistral-ask-parent
+pack: examples
+runner_type: mistral-v2

contrib/examples/actions/workflows/mistral-ask-basic.yaml

@@ -0,0 +1,26 @@
+version: '2.0'
+
+examples.mistral-ask-basic:
+    description: A basic Mistral workflow illustrating the use of Inquiries
+    type: direct
+    output:
+        result: <% task(task1).result.response %>
+    tasks:
+        task1:
+            action: core.ask
+            input:
+              route: developers
+              schema:
+                type: object
+                properties:
+                  secondfactor:
+                    type: string
+                    description: Please enter second factor for authenticating to "foo" service
+                    required: True
+            on-success:
+              - task2
+
+        task2:
+            action: core.local
+            input:
+              cmd: echo "We can now authenticate to 'foo' service with <% task(task1).result.response.secondfactor %>"

contrib/examples/actions/workflows/mistral-ask-parent.yaml

@@ -0,0 +1,10 @@
+version: '2.0'
+
+examples.mistral-ask-parent:
+    description: A workflow for testing core.ask's ability to pause nested workflows
+    type: direct
+    output:
+        result: <% task(task1).result.result %>
+    tasks:
+        task1:
+            action: examples.mistral-ask-basic

contrib/examples/rules/notify_inquiry_developers.yaml

@@ -0,0 +1,14 @@
+---
+name: "notify_inquiry_developers"
+pack: "examples"
+description: Notify developers of an Inquiry action
+enabled: false
+
+trigger:
+  type: core.st2.generic.inquiry
+
+action:
+  ref: slack.post_message
+  parameters:
+    channel: "#{{ trigger.route }}"
+    message: 'Inquiry {{trigger.id}} is awaiting an approval action'

contrib/runners/action_chain_runner/action_chain_runner.py

@@ -423,6 +423,11 @@ def _run_chain(self, action_parameters, resuming=False):
                 del result['tasks'][-1]
                 result['tasks'].append(updated_task_result)
 
+                # Also need to update context_result so the updated result
+                # is available to Jinja expressions
+                updated_task_name = updated_task_result['name']
+                context_result[updated_task_name]['result'] = updated_task_result['result']
+
             # If the last task was canceled, then canceled the chain altogether.
             if liveaction.status == action_constants.LIVEACTION_STATUS_CANCELED:
                 chain_status = action_constants.LIVEACTION_STATUS_CANCELED

contrib/runners/action_chain_runner/tests/unit/test_actionchain_pause_resume.py

@@ -28,6 +28,7 @@
 from st2common.services import action as action_service
 from st2common.transport.liveaction import LiveActionPublisher
 from st2common.transport.publishers import CUDPublisher
+from st2common.util import action_db as action_utils
 
 from st2tests import DbTestCase
 from st2tests import fixturesloader
@@ -37,6 +38,7 @@
 TEST_FIXTURES = {
     'chains': [
         'test_pause_resume.yaml',
+        'test_pause_resume_context_result',
         'test_pause_resume_with_published_vars.yaml',
         'test_pause_resume_with_error.yaml',
         'test_pause_resume_with_subworkflow.yaml',
@@ -47,6 +49,7 @@
     ],
     'actions': [
         'test_pause_resume.yaml',
+        'test_pause_resume_context_result',
         'test_pause_resume_with_published_vars.yaml',
         'test_pause_resume_with_error.yaml',
         'test_pause_resume_with_subworkflow.yaml',
@@ -728,3 +731,56 @@ def test_chain_pause_resume_last_task_failed_with_no_next_task(self):
             liveaction.result['tasks'][0]['state'],
             action_constants.LIVEACTION_STATUS_FAILED
         )
+
+    def test_chain_pause_resume_status_change(self):
+        """Tests context_result is updated when last task's status changes between pause and resume
+        """
+
+        # A temp file is created during test setup. Ensure the temp file exists.
+        # The test action chain will stall until this file is deleted. This gives
+        # the unit test a moment to run any test related logic.
+        path = self.temp_file_path
+        self.assertTrue(os.path.exists(path))
+
+        action = TEST_PACK + '.' + 'test_pause_resume_context_result'
+        params = {'tempfile': path}
+        liveaction = LiveActionDB(action=action, parameters=params)
+        liveaction, execution = action_service.request(liveaction)
+        liveaction = LiveAction.get_by_id(str(liveaction.id))
+
+        # This workflow runs 'core.ask' so will pause on its own. We just need to
+        # wait until the liveaction is pausing.
+        liveaction = self._wait_for_status(liveaction, action_constants.LIVEACTION_STATUS_PAUSING)
+        self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_PAUSING)
+
+        # Delete the temporary file that the action chain is waiting on.
+        os.remove(path)
+        self.assertFalse(os.path.exists(path))
+
+        # Wait until the liveaction is paused.
+        liveaction = self._wait_for_status(liveaction, action_constants.LIVEACTION_STATUS_PAUSED)
+        self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_PAUSED)
+
+        # Wait for non-blocking threads to complete. Ensure runner is not running.
+        MockLiveActionPublisherNonBlocking.wait_all()
+
+        last_task_liveaction_id = liveaction.result['tasks'][-1]['liveaction_id']
+        action_utils.update_liveaction_status(
+            status=action_constants.LIVEACTION_STATUS_SUCCEEDED,
+            result={'foo': 'bar'},
+            liveaction_id=last_task_liveaction_id)
+
+        # Request action chain to resume.
+        liveaction, execution = action_service.request_resume(liveaction, USERNAME)
+
+        # Wait until the liveaction is completed.
+        liveaction = self._wait_for_status(liveaction, action_constants.LIVEACTION_STATUS_SUCCEEDED)
+        self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
+
+        # Wait for non-blocking threads to complete.
+        MockLiveActionPublisherNonBlocking.wait_all()
+
+        # Check liveaction result.
+        self.assertIn('tasks', liveaction.result)
+        self.assertEqual(len(liveaction.result['tasks']), 2)
+        self.assertEqual(liveaction.result['tasks'][0]['result']['foo'], 'bar')

contrib/runners/inquirer/inquirer.py

@@ -0,0 +1,112 @@
+# Licensed to the StackStorm, Inc ('StackStorm') under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import uuid
+
+from st2common import log as logging
+from st2common.constants.action import LIVEACTION_STATUS_PENDING
+from st2common.constants.triggers import INQUIRY_TRIGGER
+from st2common.models.system.common import ResourceReference
+from st2common.persistence.execution import ActionExecution
+from st2common.runners.base import ActionRunner
+from st2common.services import action as action_service
+from st2common.transport.reactor import TriggerDispatcher
+from st2common.util import action_db as action_utils
+
+LOG = logging.getLogger(__name__)
+
+__all__ = [
+    'get_runner',
+    'Inquirer',
+]
+
+# constants to lookup in runner_parameters.
+RUNNER_SCHEMA = 'schema'
+RUNNER_ROLES = 'roles'
+RUNNER_USERS = 'users'
+RUNNER_ROUTE = 'route'
+RUNNER_TTL = 'ttl'
+
+DEFAULT_SCHEMA = {
+    "title": "response_data",
+    "type": "object",
+    "properties": {
+        "continue": {
+            "type": "boolean",
+            "description": "Would you like to continue the workflow?",
+            "required": True
+        }
+    }
+}
+
+
+def get_runner():
+    return Inquirer(str(uuid.uuid4()))
+
+
+class Inquirer(ActionRunner):
+    """This runner implements the ability to ask for more input during a workflow
+    """
+
+    def __init__(self, runner_id):
+        super(Inquirer, self).__init__(runner_id=runner_id)
+        self.trigger_dispatcher = TriggerDispatcher(LOG)
+
+    def pre_run(self):
+        super(Inquirer, self).pre_run()
+
+        # TODO :This is awful, but the way "runner_parameters" and other variables get
+        # assigned on the runner instance is even worse. Those arguments should
+        # be passed to the constructor.
+        self.schema = self.runner_parameters.get(RUNNER_SCHEMA, DEFAULT_SCHEMA)
+        self.roles_param = self.runner_parameters.get(RUNNER_ROLES, [])
+        self.users_param = self.runner_parameters.get(RUNNER_USERS, [])
+        self.route = self.runner_parameters.get(RUNNER_ROUTE, "")
+        self.ttl = self.runner_parameters.get(RUNNER_TTL, 1440)
+
+    def run(self, action_parameters):
+
+        liveaction_db = action_utils.get_liveaction_by_id(self.liveaction_id)
+        exc = ActionExecution.get(liveaction__id=str(liveaction_db.id))
+
+        # Assemble and dispatch trigger
+        trigger_ref = ResourceReference.to_string_reference(
+            pack=INQUIRY_TRIGGER['pack'],
+            name=INQUIRY_TRIGGER['name']
+        )
+        trigger_payload = {
+            "id": str(exc.id),
+            "route": self.route
+        }
+        self.trigger_dispatcher.dispatch(trigger_ref, trigger_payload)
+
+        # We only want to request a pause if this has a parent
+        if liveaction_db.context.get("parent"):
+
+            # Get the root liveaction and request that it pauses
+            root_liveaction = action_service.get_root_liveaction(liveaction_db)
+            action_service.request_pause(
+                root_liveaction,
+                self.context.get('user', None)
+            )
+
+        result = {
+            "schema": self.schema,
+            "roles": self.roles_param,
+            "users": self.users_param,
+            "route": self.route,
+            "ttl": self.ttl
+        }
+        return (LIVEACTION_STATUS_PENDING, result, None)

contrib/runners/inquirer/runner.yaml

@@ -0,0 +1,37 @@
+- description: A runner for Inquiries
+  enabled: true
+  name: inquirer
+  runner_module: inquirer
+  runner_parameters:
+    schema:
+      default:
+        title: response_data
+        type: object
+        properties:
+          continue:
+           type: boolean
+           description: Would you like to continue the workflow?
+           required: True
+      required: true
+      description: A JSON schema that will be used to validate the response data
+      type: object
+    route:
+      default: ""
+      required: false
+      description: An arbitrary value for allowing rules to route to proper notification channel
+      type: string
+    roles:
+      default: []
+      required: false
+      description: A list of roles that are permitted to respond to the action (if nothing provided, all are permitted) - REQUIRES ENTERPRISE FEATURES 
+      type: array
+    users:
+      default: []
+      required: false
+      description: A list of usernames that are permitted to respond to the action (if nothing provided, all are permitted)
+      type: array
+    ttl:
+      default: 1440
+      required: true
+      description: Time (in minutes) to wait before timing out the inquiry if no response is received
+      type: integer