Let querier plugin decide to delete state object on error

CHANGELOG.rst

@@ -49,6 +49,9 @@ in development
   put the workflow in a PAUSED state in mistral. (improvement)
 * Add support for evaluating jinja expressions in mistral workflow definition where yaql
   expressions are typically accepted. (improvement)
+* Let querier plugin decide whether to delete state object on error. Mistral querier will
+  delete state object on workflow completion or when the workflow or task references no
+  longer exists. (improvement)
 
 2.1.1 - December 16, 2016
 -------------------------

contrib/runners/mistral_v2/query/mistral_v2.py

@@ -1,11 +1,13 @@
 import uuid
 
+from mistralclient.api import base as mistralclient_base
 from mistralclient.api import client as mistral
 from oslo_config import cfg
 import retrying
 
 from st2common.query.base import Querier
 from st2common.constants import action as action_constants
+from st2common.exceptions import resultstracker as exceptions
 from st2common import log as logging
 from st2common.services import action as action_service
 from st2common.util import jsonify
@@ -31,6 +33,8 @@ def get_instance():
 
 
 class MistralResultsQuerier(Querier):
+    delete_state_object_on_error = False
+
     def __init__(self, id, *args, **kwargs):
         super(MistralResultsQuerier, self).__init__(*args, **kwargs)
         self._base_url = get_url_without_trailing_slash(cfg.CONF.mistral.v2_base_url)
@@ -66,6 +70,9 @@ def query(self, execution_id, query_context):
         try:
             result = self._get_workflow_result(mistral_exec_id)
             result['tasks'] = self._get_workflow_tasks(mistral_exec_id)
+        except exceptions.ReferenceNotFoundError as exc:
+            LOG.exception('[%s] Unable to find reference.', execution_id)
+            return (action_constants.LIVEACTION_STATUS_FAILED, exc.message)
         except Exception:
             LOG.exception('[%s] Unable to fetch mistral workflow result and tasks. %s',
                           execution_id, query_context)
@@ -87,7 +94,12 @@ def _get_workflow_result(self, exec_id):
         :type exec_id: ``str``
         :rtype: (``str``, ``dict``)
         """
-        execution = self._client.executions.get(exec_id)
+        try:
+            execution = self._client.executions.get(exec_id)
+        except mistralclient_base.APIException as mistral_exc:
+            if 'not found' in mistral_exc.message:
+                raise exceptions.ReferenceNotFoundError(mistral_exc.message)
+            raise mistral_exc
 
         result = jsonify.try_loads(execution.output) if execution.state in DONE_STATES else {}
 
@@ -105,10 +117,15 @@ def _get_workflow_tasks(self, exec_id):
         :type exec_id: ``str``
         :rtype: ``list``
         """
-        wf_tasks = [
-            self._client.tasks.get(task.id)
-            for task in self._client.tasks.list(workflow_execution_id=exec_id)
-        ]
+        wf_tasks = []
+
+        try:
+            for task in self._client.tasks.list(workflow_execution_id=exec_id):
+                wf_tasks.append(self._client.tasks.get(task.id))
+        except mistralclient_base.APIException as mistral_exc:
+            if 'not found' in mistral_exc.message:
+                raise exceptions.ReferenceNotFoundError(mistral_exc.message)
+            raise mistral_exc
 
         return [self._format_task_result(task=wf_task.to_dict()) for wf_task in wf_tasks]
 

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

@@ -22,6 +22,7 @@
 import mock
 from mock import call
 
+from mistralclient.api import base as mistralclient_base
 from mistralclient.api.v2 import executions
 from mistralclient.api.v2 import tasks
 from oslo_config import cfg
@@ -361,6 +362,17 @@ def test_query_get_workflow_retry_exhausted(self):
         calls = [call(MOCK_QRY_CONTEXT['mistral']['execution_id']) for i in range(0, 2)]
         executions.ExecutionManager.get.assert_has_calls(calls)
 
+    @mock.patch.object(
+        executions.ExecutionManager, 'get',
+        mock.MagicMock(
+            side_effect=mistralclient_base.APIException(
+                error_code=404, error_message='Workflow not found.')))
+    def test_query_get_workflow_not_found(self):
+        (status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)
+
+        self.assertEqual(action_constants.LIVEACTION_STATUS_FAILED, status)
+        self.assertEqual('Workflow not found.', result)
+
     @mock.patch.object(
         executions.ExecutionManager, 'get',
         mock.MagicMock(return_value=MOCK_WF_EX))
@@ -487,6 +499,23 @@ def test_query_get_workflow_tasks_retry_exhausted(self):
 
         tasks.TaskManager.get.assert_has_calls(calls)
 
+    @mock.patch.object(
+        executions.ExecutionManager, 'get',
+        mock.MagicMock(return_value=MOCK_WF_EX))
+    @mock.patch.object(
+        tasks.TaskManager, 'list',
+        mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
+    @mock.patch.object(
+        tasks.TaskManager, 'get',
+        mock.MagicMock(
+            side_effect=mistralclient_base.APIException(
+                error_code=404, error_message='Task not found.')))
+    def test_query_get_workflow_tasks_not_found(self):
+        (status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)
+
+        self.assertEqual(action_constants.LIVEACTION_STATUS_FAILED, status)
+        self.assertEqual('Task not found.', result)
+
     def test_query_missing_context(self):
         self.assertRaises(Exception, self.querier.query, uuid.uuid4().hex, {})