Action runner output streaming single collection approach

.travis.yml

@@ -46,6 +46,12 @@ install:
 script:
   - make ${TASK}
 
+# Don't store cache for target PR branch (typically `master`), because it will be re-used for opened PRs
+# See: https://docs.travis-ci.com/user/caching/#Pull-request-builds-and-caches
+# Alternative: use strict pip pinning, including git-based pip packages
+before_cache:
+  - if [ ${TRAVIS_PULL_REQUEST} = 'false' ]; then rm -rf virtualenv/; fi
+
 after_success:
   - if [ ${TASK} = 'ci-unit' ] || [ ${TASK} = 'ci-integration' ]; then codecov; fi
 

CHANGELOG.rst

@@ -37,6 +37,8 @@ Fixed
   performed the API operation.(bug fix) #3693 #3696
 
   Reported by theuiz.
+* Fix mistral callback failure when result contains unicode. (bug fix)
+* Fix cancellation of delayed action execution for tasks in workflow. (bug fix)
 
 Changed
 ~~~~~~~

Makefile

@@ -74,7 +74,6 @@ configgen: requirements .configgen
 	@echo
 	@echo "================== config gen ===================="
 	@echo
-	# Lint st2 components
 	echo "# Sample config which contains all the available options which the corresponding descriptions" > conf/st2.conf.sample;
 	echo "# Note: This file is automatically generated using tools/config_gen.py - DO NOT UPDATE MANUALLY" >> conf/st2.conf.sample
 	echo "" >> conf/st2.conf.sample
@@ -450,7 +449,7 @@ debs:
 ci: ci-checks ci-unit ci-integration ci-mistral ci-packs-tests
 
 .PHONY: ci-checks
-ci-checks: compile .pylint .flake8 .bandit .st2client-dependencies-check .st2common-circular-dependencies-check circle-lint-api-spec .rst-check
+ci-checks: compile .generated-files-check .pylint .flake8 .bandit .st2client-dependencies-check .st2common-circular-dependencies-check circle-lint-api-spec .rst-check
 
 .PHONY: .rst-check
 .rst-check:
@@ -459,6 +458,24 @@ ci-checks: compile .pylint .flake8 .bandit .st2client-dependencies-check .st2com
 	@echo
 	. $(VIRTUALENV_DIR)/bin/activate; rstcheck --report warning CHANGELOG.rst
 
+.PHONY: .generated-files-check
+.generated-files-check:
+	# Verify that all the files which are automatically generated have indeed been re-generated and
+	# committed
+	@echo "==================== generated-files-check ===================="
+
+	# 1. Sample config - conf/st2.conf.sample
+	cp conf/st2.conf.sample /tmp/st2.conf.sample.upstream
+	make .configgen
+	diff conf/st2.conf.sample /tmp/st2.conf.sample.upstream || (echo "conf/st2.conf.sample hasn't been re-generated and committed. Please run \"make configgen\" and include and commit the generated file." && exit 1)
+	# 2. OpenAPI definition file - st2common/st2common/openapi.yaml (generated from
+	# st2common/st2common/openapi.yaml.j2)
+	cp st2common/st2common/openapi.yaml /tmp/openapi.yaml.upstream
+	make .generate-api-spec
+	diff st2common/st2common/openapi.yaml  /tmp/openapi.yaml.upstream || (echo "st2common/st2common/openapi.yaml hasn't been re-generated and committed. Please run \"make generate-api-spec\" and include and commit the generated file." && exit 1)
+
+	@echo "All automatically generated files are up to date."
+
 .PHONY: ci-unit
 ci-unit: .unit-tests-coverage-html
 

conf/st2.conf.sample

@@ -71,7 +71,7 @@ key = /etc/apache2/ssl/mycert.key
 use_ssl = False
 # Port on which the service should listen on.
 port = 9100
-# Authentication backend to use in a standalone mode. Available backends: flat_file, ldap.
+# Authentication backend to use in a standalone mode. Available backends: flat_file.
 backend = flat_file
 
 [cloudslang]

contrib/runners/local_runner/local_runner.py

@@ -16,6 +16,7 @@
 import os
 import pwd
 import uuid
+import functools
 from StringIO import StringIO
 
 from oslo_config import cfg
@@ -33,8 +34,7 @@
 from st2common.util.green import shell
 from st2common.util.shell import kill_process
 from st2common.util import jsonify
-from st2common.services.action import store_execution_stdout_line
-from st2common.services.action import store_execution_stderr_line
+from st2common.services.action import store_execution_output_data
 from st2common.runners.utils import make_read_and_store_stream_func
 
 __all__ = [
@@ -145,6 +145,9 @@ def run(self, action_parameters):
         stdout = StringIO()
         stderr = StringIO()
 
+        store_execution_stdout_line = functools.partial(store_execution_output_data, type='stdout')
+        store_execution_stderr_line = functools.partial(store_execution_output_data, type='stderr')
+
         read_and_store_stdout = make_read_and_store_stream_func(execution_db=self.execution,
             action_db=self.action, store_line_func=store_execution_stdout_line)
         read_and_store_stderr = make_read_and_store_stream_func(execution_db=self.execution,

contrib/runners/mistral_v2/callback/mistral_v2.py

@@ -18,6 +18,7 @@
 import json
 import re
 import retrying
+import six
 
 from oslo_config import cfg
 from mistralclient.api import client as mistral
@@ -88,6 +89,27 @@ def _update_action_execution(cls, url, data):
 
         client.action_executions.update(action_execution_id, **data)
 
+    @classmethod
+    def _encode(cls, value):
+        if isinstance(value, dict):
+            return {k: cls._encode(v) for k, v in six.iteritems(value)}
+        elif isinstance(value, list):
+            return [cls._encode(item) for item in value]
+        elif isinstance(value, six.string_types):
+            try:
+                value = value.decode('utf-8')
+            except Exception:
+                LOG.exception('Unable to decode value to utf-8.')
+
+            try:
+                value = value.encode('unicode_escape')
+            except Exception:
+                LOG.exception('Unable to unicode escape value.')
+
+            return value
+        else:
+            return value
+
     @classmethod
     def callback(cls, url, context, status, result):
         if status not in MISTRAL_ACCEPTED_STATES:
@@ -100,7 +122,9 @@ def callback(cls, url, context, status, result):
                 if type(value) in [dict, list]:
                     result = value
 
+            result = cls._encode(result)
             output = json.dumps(result) if type(result) in [dict, list] else str(result)
+            output = output.replace('\\\\\\\\u', '\\\\u')
             data = {'state': STATUS_MAP[status], 'output': output}
 
             cls._update_action_execution(url, data)

contrib/runners/mistral_v2/tests/unit/test_mistral_v2_auth.py

@@ -266,7 +266,7 @@ def test_launch_workflow_with_mistral_auth(self):
             'cacert': None
         }
 
-        keystone.KeystoneAuthHandler.authenticate.assert_called_with(auth_req)
+        keystone.KeystoneAuthHandler.authenticate.assert_called_with(auth_req, session=None)
 
         executions.ExecutionManager.create.assert_called_with(
             WF1_NAME, workflow_input=workflow_input, env=env)

contrib/runners/mistral_v2/tests/unit/test_mistral_v2_callback.py

@@ -1,3 +1,4 @@
+# -*- coding: UTF-8 -*-
 # 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.
@@ -110,22 +111,65 @@ def test_callback_handler_with_result_as_text(self):
                                      action_constants.LIVEACTION_STATUS_SUCCEEDED,
                                      '<html></html>')
 
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='<html></html>'
+        )
+
     @mock.patch.object(
         action_executions.ActionExecutionManager, 'update',
         mock.MagicMock(return_value=None))
     def test_callback_handler_with_result_as_dict(self):
         self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
                                      action_constants.LIVEACTION_STATUS_SUCCEEDED, {'a': 1})
 
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='{"a": 1}'
+        )
+
     @mock.patch.object(
         action_executions.ActionExecutionManager, 'update',
         mock.MagicMock(return_value=None))
     def test_callback_handler_with_result_as_json_str(self):
         self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
                                      action_constants.LIVEACTION_STATUS_SUCCEEDED, '{"a": 1}')
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='{"a": 1}'
+        )
+
         self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
                                      action_constants.LIVEACTION_STATUS_SUCCEEDED, "{'a': 1}")
 
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='{"a": 1}'
+        )
+
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED, u"{'a': 1}")
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='{"a": 1}'
+        )
+
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED, "{u'a': u'xyz'}")
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='{"a": "xyz"}'
+        )
+
     @mock.patch.object(
         action_executions.ActionExecutionManager, 'update',
         mock.MagicMock(return_value=None))
@@ -134,6 +178,12 @@ def test_callback_handler_with_result_as_list(self):
                                      action_constants.LIVEACTION_STATUS_SUCCEEDED,
                                      ["a", "b", "c"])
 
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='["a", "b", "c"]'
+        )
+
     @mock.patch.object(
         action_executions.ActionExecutionManager, 'update',
         mock.MagicMock(return_value=None))
@@ -142,6 +192,127 @@ def test_callback_handler_with_result_as_list_str(self):
                                      action_constants.LIVEACTION_STATUS_SUCCEEDED,
                                      '["a", "b", "c"]')
 
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='["a", "b", "c"]'
+        )
+
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED,
+                                     u'["a", "b", "c"]')
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='["a", "b", "c"]'
+        )
+
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED,
+                                     '[u"a", "b", "c"]')
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='["a", "b", "c"]'
+        )
+
+    @mock.patch.object(
+        action_executions.ActionExecutionManager, 'update',
+        mock.MagicMock(return_value=None))
+    def test_callback_handler_with_result_unicode_str(self):
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED, '什麼')
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='\\u4ec0\\u9ebc'
+        )
+
+    @mock.patch.object(
+        action_executions.ActionExecutionManager, 'update',
+        mock.MagicMock(return_value=None))
+    def test_callback_handler_with_result_unicode_encoded_as_ascii_str(self):
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED, '\u4ec0\u9ebc')
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='\\\\u4ec0\\\\u9ebc'
+        )
+
+    @mock.patch.object(
+        action_executions.ActionExecutionManager, 'update',
+        mock.MagicMock(return_value=None))
+    def test_callback_handler_with_result_unicode_encoded_as_type(self):
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED, u'\u4ec0\u9ebc')
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='\\u4ec0\\u9ebc'
+        )
+
+    @mock.patch.object(
+        action_executions.ActionExecutionManager, 'update',
+        mock.MagicMock(return_value=None))
+    def test_callback_handler_with_result_as_list_with_unicode_str(self):
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED,
+                                     ['\u4ec0\u9ebc'])
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='["\\\\u4ec0\\\\u9ebc"]'
+        )
+
+    @mock.patch.object(
+        action_executions.ActionExecutionManager, 'update',
+        mock.MagicMock(return_value=None))
+    def test_callback_handler_with_result_as_list_with_unicode_type(self):
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED,
+                                     [u'\u4ec0\u9ebc'])
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='["\\\\u4ec0\\\\u9ebc"]'
+        )
+
+    @mock.patch.object(
+        action_executions.ActionExecutionManager, 'update',
+        mock.MagicMock(return_value=None))
+    def test_callback_handler_with_result_as_dict_with_unicode_str(self):
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED,
+                                     {'a': '\u4ec0\u9ebc'})
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='{"a": "\\\\u4ec0\\\\u9ebc"}'
+        )
+
+    @mock.patch.object(
+        action_executions.ActionExecutionManager, 'update',
+        mock.MagicMock(return_value=None))
+    def test_callback_handler_with_result_as_dict_with_unicode_type(self):
+        self.callback_class.callback('http://127.0.0.1:8989/v2/action_executions/12345', {},
+                                     action_constants.LIVEACTION_STATUS_SUCCEEDED,
+                                     {'a': u'\u4ec0\u9ebc'})
+
+        action_executions.ActionExecutionManager.update.assert_called_with(
+            '12345',
+            state='SUCCESS',
+            output='{"a": "\\\\u4ec0\\\\u9ebc"}'
+        )
+
     @mock.patch.object(
         action_executions.ActionExecutionManager, 'update',
         mock.MagicMock(return_value=None))

contrib/runners/python_runner/python_runner.py

@@ -18,6 +18,7 @@
 import sys
 import json
 import uuid
+import functools
 from StringIO import StringIO
 from subprocess import list2cmdline
 
@@ -41,8 +42,7 @@
 from st2common.util.sandboxing import get_sandbox_python_binary_path
 from st2common.util.sandboxing import get_sandbox_virtualenv_path
 from st2common.runners import python_action_wrapper
-from st2common.services.action import store_execution_stdout_line
-from st2common.services.action import store_execution_stderr_line
+from st2common.services.action import store_execution_output_data
 from st2common.runners.utils import make_read_and_store_stream_func
 
 LOG = logging.getLogger(__name__)
@@ -149,6 +149,9 @@ def run(self, action_parameters):
         stdout = StringIO()
         stderr = StringIO()
 
+        store_execution_stdout_line = functools.partial(store_execution_output_data, type='stdout')
+        store_execution_stderr_line = functools.partial(store_execution_output_data, type='stderr')
+
         read_and_store_stdout = make_read_and_store_stream_func(execution_db=self.execution,
             action_db=self.action, store_line_func=store_execution_stdout_line)
         read_and_store_stderr = make_read_and_store_stream_func(execution_db=self.execution,