MistralResultsQuerier._format_task_result() - Code Metrics - Inspection of "Optimize workflow task queries in st2resultstracke..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#3420)

by W

created 2017-05-18 22:40 UTC

MistralResultsQuerier._format_task_result() A

↳ Parent: MistralResultsQuerier

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
dl	0
loc	19
rs	9.4285
c	0
b	0
f	0

import time
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 action_db as action_utils
from st2common.util import jsonify
from st2common.util.url import get_url_without_trailing_slash
from st2common.util.workflow import mistral as utils


LOG = logging.getLogger(__name__)

DONE_STATES = {
    'ERROR': action_constants.LIVEACTION_STATUS_FAILED,
    'SUCCESS': action_constants.LIVEACTION_STATUS_SUCCEEDED,
    'CANCELLED': action_constants.LIVEACTION_STATUS_CANCELED
}

ACTIVE_STATES = {
    'RUNNING': action_constants.LIVEACTION_STATUS_RUNNING
}

CANCELED_STATES = [
    action_constants.LIVEACTION_STATUS_CANCELED,
    action_constants.LIVEACTION_STATUS_CANCELING
]


def get_instance():
    return MistralResultsQuerier(str(uuid.uuid4()))


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)
        self._client = mistral.client(
            mistral_url=self._base_url,
            username=cfg.CONF.mistral.keystone_username,
            api_key=cfg.CONF.mistral.keystone_password,
            project_name=cfg.CONF.mistral.keystone_project_name,
            auth_url=cfg.CONF.mistral.keystone_auth_url,
            cacert=cfg.CONF.mistral.cacert,
            insecure=cfg.CONF.mistral.insecure)

    @retrying.retry(
        retry_on_exception=utils.retry_on_exceptions,
        wait_exponential_multiplier=cfg.CONF.mistral.retry_exp_msec,
        wait_exponential_max=cfg.CONF.mistral.retry_exp_max_msec,
        stop_max_delay=cfg.CONF.mistral.retry_stop_max_msec)
    def query(self, execution_id, query_context, last_query_time=None):
        """
        Queries mistral for workflow results using v2 APIs.
        :param execution_id: st2 execution_id (context to be used for logging/audit)
        :type execution_id: ``str``
        :param query_context: context for the query to be made to mistral. This contains mistral
                              execution id.
        :type query_context: ``object``
        :param last_query_time: Timestamp of last query.
        :type last_query_time: ``float``
        :rtype: (``str``, ``object``)
        """
        dt_last_query_time = (
            time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(last_query_time))
            if last_query_time else None
        )

        liveaction_db = action_utils.get_liveaction_by_id(execution_id)

        mistral_exec_id = query_context.get('mistral', {}).get('execution_id', None)
        if not mistral_exec_id:
            raise Exception('[%s] Missing mistral workflow execution ID in query context. %s',
                            execution_id, query_context)

        try:
            wf_result = self._get_workflow_result(mistral_exec_id)

            wf_tasks_result = self._get_workflow_tasks(
                mistral_exec_id,
                last_query_time=dt_last_query_time
            )

            result = self._format_query_result(
                liveaction_db.result,
                wf_result,
                wf_tasks_result
            )
        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)
            raise

        status = self._determine_execution_status(
            liveaction_db,
            result['extra']['state'],
            result['tasks']
        )

        LOG.debug('[%s] mistral workflow execution status: %s' % (execution_id, status))
        LOG.debug('[%s] mistral workflow execution result: %s' % (execution_id, result))

        return (status, result)

    def _get_workflow_result(self, exec_id):
        """
        Returns the workflow status and output. Mistral workflow status will be converted
        to st2 action status.
        :param exec_id: Mistral execution ID
        :type exec_id: ``str``
        :rtype: (``str``, ``dict``)
        """
        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 {}

        result['extra'] = {
            'state': execution.state,
            'state_info': execution.state_info
        }

        return result

    def _get_workflow_tasks(self, exec_id, last_query_time=None):
        """
        Returns the list of tasks for a workflow execution.
        :param exec_id: Mistral execution ID
        :type exec_id: ``str``
        :param last_query_time: Timestamp to filter tasks
        :type last_query_time: ``str``
        :rtype: ``list``
        """
        result = []

        try:
            query_filters = {}

            if last_query_time:
                query_filters['updated_at'] = 'gte:%s' % last_query_time

            wf_tasks = self._client.tasks.list(workflow_execution_id=exec_id, **query_filters)

            for wf_task in wf_tasks:
                result.append(self._client.tasks.get(wf_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=entry.to_dict()) for entry in result]

    def _format_task_result(self, task):
        """
        Format task result to follow the unified workflow result format.
        """
        result = {
            'id': task['id'],
            'name': task['name'],
            'workflow_execution_id': task.get('workflow_execution_id', None),
            'workflow_name': task['workflow_name'],
            'created_at': task.get('created_at', None),
            'updated_at': task.get('updated_at', None),
            'state': task.get('state', None),
            'state_info': task.get('state_info', None)
        }

        for attr in ['result', 'input', 'published']:
            result[attr] = jsonify.try_loads(task.get(attr, None))

        return result

    def _format_query_result(self, current_result, new_wf_result, new_wf_tasks_result):
        result = new_wf_result

        new_wf_task_ids = [entry['id'] for entry in new_wf_tasks_result]

        old_wf_tasks_result_to_keep = [
            entry for entry in current_result.get('tasks', [])
            if entry['id'] not in new_wf_task_ids
        ] 


        result['tasks'] = old_wf_tasks_result_to_keep + new_wf_tasks_result

        return result

    def _determine_execution_status(self, liveaction_db, wf_state, tasks):
        # Determine if liveaction is canceled or being canceled.
        is_action_canceled = liveaction_db.status in CANCELED_STATES

        # Identify the list of tasks that are not still running.
        active_tasks = [t for t in tasks if t['state'] in ACTIVE_STATES]

        # Keep the execution in running state if there are active tasks.
        # In certain use cases, Mistral sets the workflow state to
        # completion prior to task completion.
        if is_action_canceled and active_tasks:
            status = action_constants.LIVEACTION_STATUS_CANCELING
        elif is_action_canceled and not active_tasks and wf_state not in DONE_STATES:
            status = action_constants.LIVEACTION_STATUS_CANCELING
        elif not is_action_canceled and active_tasks and wf_state == 'CANCELLED':
            status = action_constants.LIVEACTION_STATUS_CANCELING
        elif wf_state in DONE_STATES and active_tasks:
            status = action_constants.LIVEACTION_STATUS_RUNNING
        elif wf_state in DONE_STATES and not active_tasks:
            status = DONE_STATES[wf_state]
        else:
            status = action_constants.LIVEACTION_STATUS_RUNNING

        return status


1			import time
2			import uuid
3
4			from mistralclient.api import base as mistralclient_base
5			from mistralclient.api import client as mistral
6			from oslo_config import cfg
7			import retrying
8
9			from st2common.query.base import Querier
10			from st2common.constants import action as action_constants
11			from st2common.exceptions import resultstracker as exceptions
12			from st2common import log as logging
13			from st2common.services import action as action_service
			0 ignored issues – show Unused Code introduced 2017-05-18 22:45 UTC by Report Bug Copy Issue Report Unused action imported from st2common.services as action_service Loading history...
14			from st2common.util import action_db as action_utils
15			from st2common.util import jsonify
16			from st2common.util.url import get_url_without_trailing_slash
17			from st2common.util.workflow import mistral as utils
18
19
20			LOG = logging.getLogger(__name__)
21
22			DONE_STATES = {
23			'ERROR': action_constants.LIVEACTION_STATUS_FAILED,
24			'SUCCESS': action_constants.LIVEACTION_STATUS_SUCCEEDED,
25			'CANCELLED': action_constants.LIVEACTION_STATUS_CANCELED
26			}
27
28			ACTIVE_STATES = {
29			'RUNNING': action_constants.LIVEACTION_STATUS_RUNNING
30			}
31
32			CANCELED_STATES = [
33			action_constants.LIVEACTION_STATUS_CANCELED,
34			action_constants.LIVEACTION_STATUS_CANCELING
35			]
36
37
38			def get_instance():
39			return MistralResultsQuerier(str(uuid.uuid4()))
40
41
42			class MistralResultsQuerier(Querier):
43			delete_state_object_on_error = False
44
45			def __init__(self, id, args, *kwargs):
			0 ignored issues – show Bug Best Practice introduced 2017-04-18 09:25 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `id`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
46			super(MistralResultsQuerier, self).__init__(args, *kwargs)
47			self._base_url = get_url_without_trailing_slash(cfg.CONF.mistral.v2_base_url)
48			self._client = mistral.client(
49			mistral_url=self._base_url,
50			username=cfg.CONF.mistral.keystone_username,
51			api_key=cfg.CONF.mistral.keystone_password,
52			project_name=cfg.CONF.mistral.keystone_project_name,
53			auth_url=cfg.CONF.mistral.keystone_auth_url,
54			cacert=cfg.CONF.mistral.cacert,
55			insecure=cfg.CONF.mistral.insecure)
56
57			@retrying.retry(
58			retry_on_exception=utils.retry_on_exceptions,
59			wait_exponential_multiplier=cfg.CONF.mistral.retry_exp_msec,
60			wait_exponential_max=cfg.CONF.mistral.retry_exp_max_msec,
61			stop_max_delay=cfg.CONF.mistral.retry_stop_max_msec)
62			def query(self, execution_id, query_context, last_query_time=None):
63			"""
64			Queries mistral for workflow results using v2 APIs.
65			:param execution_id: st2 execution_id (context to be used for logging/audit)
66			:type execution_id: ``str``
67			:param query_context: context for the query to be made to mistral. This contains mistral
68			execution id.
69			:type query_context: ``object``
70			:param last_query_time: Timestamp of last query.
71			:type last_query_time: ``float``
72			:rtype: (``str``, ``object``)
73			"""
74			dt_last_query_time = (
75			time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(last_query_time))
76			if last_query_time else None
77			)
78
79			liveaction_db = action_utils.get_liveaction_by_id(execution_id)
80
81			mistral_exec_id = query_context.get('mistral', {}).get('execution_id', None)
82			if not mistral_exec_id:
83			raise Exception('[%s] Missing mistral workflow execution ID in query context. %s',
84			execution_id, query_context)
85
86			try:
87			wf_result = self._get_workflow_result(mistral_exec_id)
88
89			wf_tasks_result = self._get_workflow_tasks(
90			mistral_exec_id,
91			last_query_time=dt_last_query_time
92			)
93
94			result = self._format_query_result(
95			liveaction_db.result,
96			wf_result,
97			wf_tasks_result
98			)
99			except exceptions.ReferenceNotFoundError as exc:
100			LOG.exception('[%s] Unable to find reference.', execution_id)
101			return (action_constants.LIVEACTION_STATUS_FAILED, exc.message)
102			except Exception:
103			LOG.exception('[%s] Unable to fetch mistral workflow result and tasks. %s',
104			execution_id, query_context)
105			raise
106
107			status = self._determine_execution_status(
108			liveaction_db,
109			result['extra']['state'],
110			result['tasks']
111			)
112
113			LOG.debug('[%s] mistral workflow execution status: %s' % (execution_id, status))
114			LOG.debug('[%s] mistral workflow execution result: %s' % (execution_id, result))
115
116			return (status, result)
117
118			def _get_workflow_result(self, exec_id):
119			"""
120			Returns the workflow status and output. Mistral workflow status will be converted
121			to st2 action status.
122			:param exec_id: Mistral execution ID
123			:type exec_id: ``str``
124			:rtype: (``str``, ``dict``)
125			"""
126			try:
127			execution = self._client.executions.get(exec_id)
128			except mistralclient_base.APIException as mistral_exc:
129			if 'not found' in mistral_exc.message:
130			raise exceptions.ReferenceNotFoundError(mistral_exc.message)
131			raise mistral_exc
132
133			result = jsonify.try_loads(execution.output) if execution.state in DONE_STATES else {}
134
135			result['extra'] = {
136			'state': execution.state,
137			'state_info': execution.state_info
138			}
139
140			return result
141
142			def _get_workflow_tasks(self, exec_id, last_query_time=None):
143			"""
144			Returns the list of tasks for a workflow execution.
145			:param exec_id: Mistral execution ID
146			:type exec_id: ``str``
147			:param last_query_time: Timestamp to filter tasks
148			:type last_query_time: ``str``
149			:rtype: ``list``
150			"""
151			result = []
152
153			try:
154			query_filters = {}
155
156			if last_query_time:
157			query_filters['updated_at'] = 'gte:%s' % last_query_time
158
159			wf_tasks = self._client.tasks.list(workflow_execution_id=exec_id, **query_filters)
160
161			for wf_task in wf_tasks:
162			result.append(self._client.tasks.get(wf_task.id))
163			except mistralclient_base.APIException as mistral_exc:
164			if 'not found' in mistral_exc.message:
165			raise exceptions.ReferenceNotFoundError(mistral_exc.message)
166			raise mistral_exc
167
168			return [self._format_task_result(task=entry.to_dict()) for entry in result]
169
170			def _format_task_result(self, task):
171			"""
172			Format task result to follow the unified workflow result format.
173			"""
174			result = {
175			'id': task['id'],
176			'name': task['name'],
177			'workflow_execution_id': task.get('workflow_execution_id', None),
178			'workflow_name': task['workflow_name'],
179			'created_at': task.get('created_at', None),
180			'updated_at': task.get('updated_at', None),
181			'state': task.get('state', None),
182			'state_info': task.get('state_info', None)
183			}
184
185			for attr in ['result', 'input', 'published']:
186			result[attr] = jsonify.try_loads(task.get(attr, None))
187
188			return result
189
190			def _format_query_result(self, current_result, new_wf_result, new_wf_tasks_result):
191			result = new_wf_result
192
193			new_wf_task_ids = [entry['id'] for entry in new_wf_tasks_result]
194
195			old_wf_tasks_result_to_keep = [
196			entry for entry in current_result.get('tasks', [])
197			if entry['id'] not in new_wf_task_ids
198			]
			0 ignored issues – show Coding Style introduced 2017-05-18 22:45 UTC by Report Bug Copy Issue Report Trailing whitespace Loading history...
199
200			result['tasks'] = old_wf_tasks_result_to_keep + new_wf_tasks_result
201
202			return result
203
204			def _determine_execution_status(self, liveaction_db, wf_state, tasks):
205			# Determine if liveaction is canceled or being canceled.
206			is_action_canceled = liveaction_db.status in CANCELED_STATES
207
208			# Identify the list of tasks that are not still running.
209			active_tasks = [t for t in tasks if t['state'] in ACTIVE_STATES]
210
211			# Keep the execution in running state if there are active tasks.
212			# In certain use cases, Mistral sets the workflow state to
213			# completion prior to task completion.
214			if is_action_canceled and active_tasks:
215			status = action_constants.LIVEACTION_STATUS_CANCELING
216			elif is_action_canceled and not active_tasks and wf_state not in DONE_STATES:
217			status = action_constants.LIVEACTION_STATUS_CANCELING
218			elif not is_action_canceled and active_tasks and wf_state == 'CANCELLED':
219			status = action_constants.LIVEACTION_STATUS_CANCELING
220			elif wf_state in DONE_STATES and active_tasks:
221			status = action_constants.LIVEACTION_STATUS_RUNNING
222			elif wf_state in DONE_STATES and not active_tasks:
223			status = DONE_STATES[wf_state]
224			else:
225			status = action_constants.LIVEACTION_STATUS_RUNNING
226
227			return status
228

StackStorm / st2

Pull Request — master (#3420)

MistralResultsQuerier._format_task_result() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like