st2actions.container.RunnerContainer._create_auth_token() - Code Metrics - Inspection of "Rearrange PythonRunner.Action logger creation." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — kale/action-datastore (#6)

by Manas

created 2016-02-02 00:01 UTC

_create_auth_token() A

↳ Parent: st2actions.container.RunnerContainer

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	7
rs	9.4285
cc	3

# 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 json
import sys
import traceback

from st2common import log as logging
from st2common.util import date as date_utils
from st2common.constants import action as action_constants
from st2common.exceptions import actionrunner
from st2common.exceptions.param import ParamException
from st2common.models.db.executionstate import ActionExecutionStateDB
from st2common.models.system.action import ResolvedActionParameters
from st2common.persistence.execution import ActionExecution
from st2common.persistence.executionstate import ActionExecutionState
from st2common.services import access, executions
from st2common.util.action_db import (get_action_by_ref, get_runnertype_by_name)
from st2common.util.action_db import (update_liveaction_status, get_liveaction_by_id)
from st2common.util import param as param_utils

from st2actions.container.service import RunnerContainerService
from st2actions.runners import get_runner, AsyncActionRunner

LOG = logging.getLogger(__name__)

__all__ = [
    'RunnerContainer',
    'get_runner_container'
]


class RunnerContainer(object):

    def dispatch(self, liveaction_db):
        action_db = get_action_by_ref(liveaction_db.action)
        if not action_db:
            raise Exception('Action %s not found in DB.' % (liveaction_db.action))

        runnertype_db = get_runnertype_by_name(action_db.runner_type['name'])

        extra = {'liveaction_db': liveaction_db, 'runnertype_db': runnertype_db}
        LOG.info('Dispatching Action to a runner', extra=extra)

        # Get runner instance.
        runner = self._get_runner(runnertype_db, action_db, liveaction_db)
        LOG.debug('Runner instance for RunnerType "%s" is: %s', runnertype_db.name, runner)

        # Process the request.
        if liveaction_db.status == action_constants.LIVEACTION_STATUS_CANCELING:
            liveaction_db = self._do_cancel(runner=runner, runnertype_db=runnertype_db,
                                            action_db=action_db, liveaction_db=liveaction_db)
        else:
            liveaction_db = self._do_run(runner=runner, runnertype_db=runnertype_db,
                                         action_db=action_db, liveaction_db=liveaction_db)

        return liveaction_db.result

    def _do_run(self, runner, runnertype_db, action_db, liveaction_db):
        # Create a temporary auth token which will be available
        # for the duration of the action execution.
        runner.auth_token = self._create_auth_token(runner.context)

        updated_liveaction_db = None
        try:
            # Finalized parameters are resolved and then rendered. This process could
            # fail. Handle the exception and report the error correctly.
            try:
                runner_params, action_params = param_utils.render_final_params(
                    runnertype_db.runner_parameters, action_db.parameters, liveaction_db.parameters,
                    liveaction_db.context)
                runner.runner_parameters = runner_params
            except ParamException as e:
                raise actionrunner.ActionRunnerException(str(e))

            LOG.debug('Performing pre-run for runner: %s', runner.runner_id)
            runner.pre_run()

            # Mask secret parameters in the log context
            resolved_action_params = ResolvedActionParameters(action_db=action_db,
                                                              runner_type_db=runnertype_db,
                                                              runner_parameters=runner_params,
                                                              action_parameters=action_params)
            extra = {'runner': runner, 'parameters': resolved_action_params}
            LOG.debug('Performing run for runner: %s' % (runner.runner_id), extra=extra)

            (status, result, context) = runner.run(action_params)

            try:
                result = json.loads(result)
            except:
                pass

            action_completed = status in action_constants.LIVEACTION_COMPLETED_STATES
            if isinstance(runner, AsyncActionRunner) and not action_completed:
                self._setup_async_query(liveaction_db.id, runnertype_db, context)
        except:
            LOG.exception('Failed to run action.')
            _, ex, tb = sys.exc_info()
            # mark execution as failed.
            status = action_constants.LIVEACTION_STATUS_FAILED
            # include the error message and traceback to try and provide some hints.
            result = {'error': str(ex), 'traceback': ''.join(traceback.format_tb(tb, 20))}
            context = None
        finally:
            # Log action completion
            extra = {'result': result, 'status': status}
            LOG.debug('Action "%s" completed.' % (action_db.name), extra=extra)

            # Always clean-up the auth_token
            try:
                LOG.debug('Setting status: %s for liveaction: %s', status, liveaction_db.id)
                updated_liveaction_db = self._update_live_action_db(liveaction_db.id, status,
                                                                    result, context)
            except:
                error = 'Cannot update LiveAction object for id: %s, status: %s, result: %s.' % (
                    liveaction_db.id, status, result)
                LOG.exception(error)
                raise

            executions.update_execution(updated_liveaction_db)
            extra = {'liveaction_db': updated_liveaction_db}
            LOG.debug('Updated liveaction after run', extra=extra)

            # Always clean-up the auth_token
            self._clean_up_auth_token(runner=runner, status=status)

        LOG.debug('Performing post_run for runner: %s', runner.runner_id)
        runner.post_run(status=status, result=result)
        runner.container_service = None

        LOG.debug('Runner do_run result', extra={'result': updated_liveaction_db.result})
        LOG.audit('Liveaction completed', extra={'liveaction_db': updated_liveaction_db})

        return updated_liveaction_db

    def _do_cancel(self, runner, runnertype_db, action_db, liveaction_db):
        try:
            extra = {'runner': runner}
            LOG.debug('Performing cancel for runner: %s', (runner.runner_id), extra=extra)

            runner.cancel()

            liveaction_db = update_liveaction_status(
                status=action_constants.LIVEACTION_STATUS_CANCELED,
                end_timestamp=date_utils.get_datetime_utc_now(),
                liveaction_db=liveaction_db)

            executions.update_execution(liveaction_db)

            LOG.debug('Performing post_run for runner: %s', runner.runner_id)
            result = {'error': 'Execution canceled by user.'}
            runner.post_run(status=liveaction_db.status, result=result)
            runner.container_service = None
        except:
            _, ex, tb = sys.exc_info()
            # include the error message and traceback to try and provide some hints.
            result = {'error': str(ex), 'traceback': ''.join(traceback.format_tb(tb, 20))}
            LOG.exception('Failed to cancel action %s.' % (liveaction_db.id), extra=result)
        finally:
            # Always clean-up the auth_token
            status = liveaction_db.status
            self._clean_up_auth_token(runner=runner, status=status)

        return liveaction_db

    def _clean_up_auth_token(self, runner, status):
        """
        Clean up the temporary auth token for the current action.
        """
        # Deletion of the runner generated auth token is delayed until the token expires.
        # Async actions such as Mistral workflows uses the auth token to launch other
        # actions in the workflow. If the auth token is deleted here, then the actions
        # in the workflow will fail with unauthorized exception.
        is_async_runner = isinstance(runner, AsyncActionRunner)
        action_completed = status in action_constants.LIVEACTION_COMPLETED_STATES

        if not is_async_runner or (is_async_runner and action_completed):
            try:
                self._delete_auth_token(runner.auth_token)
            except:
                LOG.exception('Unable to clean-up auth_token.')

            return True

        return False

    def _update_live_action_db(self, liveaction_id, status, result, context):
        """
        Update LiveActionDB object for the provided liveaction id.
        """
        liveaction_db = get_liveaction_by_id(liveaction_id)
        if status in action_constants.LIVEACTION_COMPLETED_STATES:
            end_timestamp = date_utils.get_datetime_utc_now()
        else:
            end_timestamp = None

        liveaction_db = update_liveaction_status(status=status,
                                                 result=result,
                                                 context=context,
                                                 end_timestamp=end_timestamp,
                                                 liveaction_db=liveaction_db)
        return liveaction_db

    def _get_entry_point_abs_path(self, pack, entry_point):
        return RunnerContainerService.get_entry_point_abs_path(pack=pack,
                                                               entry_point=entry_point)

    def _get_action_libs_abs_path(self, pack, entry_point):
        return RunnerContainerService.get_action_libs_abs_path(pack=pack,
                                                               entry_point=entry_point)

    def _get_rerun_reference(self, context):
        execution_id = context.get('re-run', {}).get('ref')
        return ActionExecution.get_by_id(execution_id) if execution_id else None

    def _get_runner(self, runnertype_db, action_db, liveaction_db):
        runner = get_runner(runnertype_db.runner_module)

        resolved_entry_point = self._get_entry_point_abs_path(action_db.pack,
                                                              action_db.entry_point)

        runner.container_service = RunnerContainerService()
        runner.action = action_db
        runner.action_name = action_db.name
        runner.liveaction = liveaction_db
        runner.liveaction_id = str(liveaction_db.id)
        runner.execution = ActionExecution.get(liveaction__id=runner.liveaction_id)
        runner.execution_id = str(runner.execution.id)
        runner.entry_point = resolved_entry_point
        runner.context = getattr(liveaction_db, 'context', dict())
        runner.callback = getattr(liveaction_db, 'callback', dict())
        runner.libs_dir_path = self._get_action_libs_abs_path(action_db.pack,
                                                              action_db.entry_point)

        # For re-run, get the ActionExecutionDB in which the re-run is based on.
        rerun_ref_id = runner.context.get('re-run', {}).get('ref')
        runner.rerun_ex_ref = ActionExecution.get(id=rerun_ref_id) if rerun_ref_id else None

        return runner

    def _create_auth_token(self, context):
        if not context:
            return None
        user = context.get('user', None)
        if not user:
            return None
        return access.create_token(user)

    def _delete_auth_token(self, auth_token):
        if auth_token:
            access.delete_token(auth_token.token)

    def _setup_async_query(self, liveaction_id, runnertype_db, query_context):
        query_module = getattr(runnertype_db, 'query_module', None)
        if not query_module:
            LOG.error('No query module specified for runner %s.', runnertype_db)
            return
        try:
            self._create_execution_state(liveaction_id, runnertype_db, query_context)
        except:
            LOG.exception('Unable to create action execution state db model ' +
                          'for liveaction_id %s', liveaction_id)

    def _create_execution_state(self, liveaction_id, runnertype_db, query_context):
        state_db = ActionExecutionStateDB(
            execution_id=liveaction_id,
            query_module=runnertype_db.query_module,
            query_context=query_context)
        try:
            return ActionExecutionState.add_or_update(state_db)
        except:
            LOG.exception('Unable to create execution state db for liveaction_id %s.'
                          % liveaction_id)
            return None


def get_runner_container():
    return RunnerContainer()


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import json
17			import sys
18			import traceback
19
20			from st2common import log as logging
21			from st2common.util import date as date_utils
22			from st2common.constants import action as action_constants
23			from st2common.exceptions import actionrunner
24			from st2common.exceptions.param import ParamException
25			from st2common.models.db.executionstate import ActionExecutionStateDB
26			from st2common.models.system.action import ResolvedActionParameters
27			from st2common.persistence.execution import ActionExecution
28			from st2common.persistence.executionstate import ActionExecutionState
29			from st2common.services import access, executions
30			from st2common.util.action_db import (get_action_by_ref, get_runnertype_by_name)
31			from st2common.util.action_db import (update_liveaction_status, get_liveaction_by_id)
32			from st2common.util import param as param_utils
33
34			from st2actions.container.service import RunnerContainerService
35			from st2actions.runners import get_runner, AsyncActionRunner
36
37			LOG = logging.getLogger(__name__)
38
39			__all__ = [
40			'RunnerContainer',
41			'get_runner_container'
42			]
43
44
45			class RunnerContainer(object):
46
47			def dispatch(self, liveaction_db):
48			action_db = get_action_by_ref(liveaction_db.action)
49			if not action_db:
50			raise Exception('Action %s not found in DB.' % (liveaction_db.action))
51
52			runnertype_db = get_runnertype_by_name(action_db.runner_type['name'])
53
54			extra = {'liveaction_db': liveaction_db, 'runnertype_db': runnertype_db}
55			LOG.info('Dispatching Action to a runner', extra=extra)
56
57			# Get runner instance.
58			runner = self._get_runner(runnertype_db, action_db, liveaction_db)
59			LOG.debug('Runner instance for RunnerType "%s" is: %s', runnertype_db.name, runner)
60
61			# Process the request.
62			if liveaction_db.status == action_constants.LIVEACTION_STATUS_CANCELING:
63			liveaction_db = self._do_cancel(runner=runner, runnertype_db=runnertype_db,
64			action_db=action_db, liveaction_db=liveaction_db)
65			else:
66			liveaction_db = self._do_run(runner=runner, runnertype_db=runnertype_db,
67			action_db=action_db, liveaction_db=liveaction_db)
68
69			return liveaction_db.result
70
71			def _do_run(self, runner, runnertype_db, action_db, liveaction_db):
72			# Create a temporary auth token which will be available
73			# for the duration of the action execution.
74			runner.auth_token = self._create_auth_token(runner.context)
75
76			updated_liveaction_db = None
77			try:
78			# Finalized parameters are resolved and then rendered. This process could
79			# fail. Handle the exception and report the error correctly.
80			try:
81			runner_params, action_params = param_utils.render_final_params(
82			runnertype_db.runner_parameters, action_db.parameters, liveaction_db.parameters,
83			liveaction_db.context)
84			runner.runner_parameters = runner_params
85			except ParamException as e:
86			raise actionrunner.ActionRunnerException(str(e))
87
88			LOG.debug('Performing pre-run for runner: %s', runner.runner_id)
89			runner.pre_run()
90
91			# Mask secret parameters in the log context
92			resolved_action_params = ResolvedActionParameters(action_db=action_db,
93			runner_type_db=runnertype_db,
94			runner_parameters=runner_params,
95			action_parameters=action_params)
96			extra = {'runner': runner, 'parameters': resolved_action_params}
97			LOG.debug('Performing run for runner: %s' % (runner.runner_id), extra=extra)
			0 ignored issues – show Bug introduced 2016-01-14 18:36 UTC by Report Bug Copy Issue Report The variable `extra` was used before it was assigned. Loading history...
98			(status, result, context) = runner.run(action_params)
99
100			try:
101			result = json.loads(result)
102			except:
103			pass
104
105			action_completed = status in action_constants.LIVEACTION_COMPLETED_STATES
106			if isinstance(runner, AsyncActionRunner) and not action_completed:
107			self._setup_async_query(liveaction_db.id, runnertype_db, context)
108			except:
109			LOG.exception('Failed to run action.')
110			_, ex, tb = sys.exc_info()
111			# mark execution as failed.
112			status = action_constants.LIVEACTION_STATUS_FAILED
113			# include the error message and traceback to try and provide some hints.
114			result = {'error': str(ex), 'traceback': ''.join(traceback.format_tb(tb, 20))}
115			context = None
116			finally:
117			# Log action completion
118			extra = {'result': result, 'status': status}
119			LOG.debug('Action "%s" completed.' % (action_db.name), extra=extra)
120
121			# Always clean-up the auth_token
122			try:
123			LOG.debug('Setting status: %s for liveaction: %s', status, liveaction_db.id)
124			updated_liveaction_db = self._update_live_action_db(liveaction_db.id, status,
125			result, context)
126			except:
127			error = 'Cannot update LiveAction object for id: %s, status: %s, result: %s.' % (
128			liveaction_db.id, status, result)
129			LOG.exception(error)
130			raise
131
132			executions.update_execution(updated_liveaction_db)
133			extra = {'liveaction_db': updated_liveaction_db}
134			LOG.debug('Updated liveaction after run', extra=extra)
135
136			# Always clean-up the auth_token
137			self._clean_up_auth_token(runner=runner, status=status)
138
139			LOG.debug('Performing post_run for runner: %s', runner.runner_id)
140			runner.post_run(status=status, result=result)
141			runner.container_service = None
142
143			LOG.debug('Runner do_run result', extra={'result': updated_liveaction_db.result})
144			LOG.audit('Liveaction completed', extra={'liveaction_db': updated_liveaction_db})
145
146			return updated_liveaction_db
147
148			def _do_cancel(self, runner, runnertype_db, action_db, liveaction_db):
149			try:
150			extra = {'runner': runner}
151			LOG.debug('Performing cancel for runner: %s', (runner.runner_id), extra=extra)
152
153			runner.cancel()
154
155			liveaction_db = update_liveaction_status(
156			status=action_constants.LIVEACTION_STATUS_CANCELED,
157			end_timestamp=date_utils.get_datetime_utc_now(),
158			liveaction_db=liveaction_db)
159
160			executions.update_execution(liveaction_db)
161
162			LOG.debug('Performing post_run for runner: %s', runner.runner_id)
163			result = {'error': 'Execution canceled by user.'}
164			runner.post_run(status=liveaction_db.status, result=result)
165			runner.container_service = None
166			except:
167			_, ex, tb = sys.exc_info()
168			# include the error message and traceback to try and provide some hints.
169			result = {'error': str(ex), 'traceback': ''.join(traceback.format_tb(tb, 20))}
170			LOG.exception('Failed to cancel action %s.' % (liveaction_db.id), extra=result)
171			finally:
172			# Always clean-up the auth_token
173			status = liveaction_db.status
174			self._clean_up_auth_token(runner=runner, status=status)
175
176			return liveaction_db
177
178			def _clean_up_auth_token(self, runner, status):
179			"""
180			Clean up the temporary auth token for the current action.
181			"""
182			# Deletion of the runner generated auth token is delayed until the token expires.
183			# Async actions such as Mistral workflows uses the auth token to launch other
184			# actions in the workflow. If the auth token is deleted here, then the actions
185			# in the workflow will fail with unauthorized exception.
186			is_async_runner = isinstance(runner, AsyncActionRunner)
187			action_completed = status in action_constants.LIVEACTION_COMPLETED_STATES
188
189			if not is_async_runner or (is_async_runner and action_completed):
190			try:
191			self._delete_auth_token(runner.auth_token)
192			except:
193			LOG.exception('Unable to clean-up auth_token.')
194
195			return True
196
197			return False
198
199			def _update_live_action_db(self, liveaction_id, status, result, context):
200			"""
201			Update LiveActionDB object for the provided liveaction id.
202			"""
203			liveaction_db = get_liveaction_by_id(liveaction_id)
204			if status in action_constants.LIVEACTION_COMPLETED_STATES:
205			end_timestamp = date_utils.get_datetime_utc_now()
206			else:
207			end_timestamp = None
208
209			liveaction_db = update_liveaction_status(status=status,
210			result=result,
211			context=context,
212			end_timestamp=end_timestamp,
213			liveaction_db=liveaction_db)
214			return liveaction_db
215
216			def _get_entry_point_abs_path(self, pack, entry_point):
217			return RunnerContainerService.get_entry_point_abs_path(pack=pack,
218			entry_point=entry_point)
219
220			def _get_action_libs_abs_path(self, pack, entry_point):
221			return RunnerContainerService.get_action_libs_abs_path(pack=pack,
222			entry_point=entry_point)
223
224			def _get_rerun_reference(self, context):
225			execution_id = context.get('re-run', {}).get('ref')
226			return ActionExecution.get_by_id(execution_id) if execution_id else None
227
228			def _get_runner(self, runnertype_db, action_db, liveaction_db):
229			runner = get_runner(runnertype_db.runner_module)
230
231			resolved_entry_point = self._get_entry_point_abs_path(action_db.pack,
232			action_db.entry_point)
233
234			runner.container_service = RunnerContainerService()
235			runner.action = action_db
236			runner.action_name = action_db.name
237			runner.liveaction = liveaction_db
238			runner.liveaction_id = str(liveaction_db.id)
239			runner.execution = ActionExecution.get(liveaction__id=runner.liveaction_id)
240			runner.execution_id = str(runner.execution.id)
241			runner.entry_point = resolved_entry_point
242			runner.context = getattr(liveaction_db, 'context', dict())
243			runner.callback = getattr(liveaction_db, 'callback', dict())
244			runner.libs_dir_path = self._get_action_libs_abs_path(action_db.pack,
245			action_db.entry_point)
246
247			# For re-run, get the ActionExecutionDB in which the re-run is based on.
248			rerun_ref_id = runner.context.get('re-run', {}).get('ref')
249			runner.rerun_ex_ref = ActionExecution.get(id=rerun_ref_id) if rerun_ref_id else None
250
251			return runner
252
253			def _create_auth_token(self, context):
254			if not context:
255			return None
256			user = context.get('user', None)
257			if not user:
258			return None
259			return access.create_token(user)
260
261			def _delete_auth_token(self, auth_token):
262			if auth_token:
263			access.delete_token(auth_token.token)
264
265			def _setup_async_query(self, liveaction_id, runnertype_db, query_context):
266			query_module = getattr(runnertype_db, 'query_module', None)
267			if not query_module:
268			LOG.error('No query module specified for runner %s.', runnertype_db)
269			return
270			try:
271			self._create_execution_state(liveaction_id, runnertype_db, query_context)
272			except:
273			LOG.exception('Unable to create action execution state db model ' +
274			'for liveaction_id %s', liveaction_id)
275
276			def _create_execution_state(self, liveaction_id, runnertype_db, query_context):
277			state_db = ActionExecutionStateDB(
278			execution_id=liveaction_id,
279			query_module=runnertype_db.query_module,
280			query_context=query_context)
281			try:
282			return ActionExecutionState.add_or_update(state_db)
283			except:
284			LOG.exception('Unable to create execution state db for liveaction_id %s.'
285			% liveaction_id)
286			return None
287
288
289			def get_runner_container():
290			return RunnerContainer()
291

Plexxi / st2

GitHub Access Token became invalid

Pull Request — kale/action-datastore (#6)

_create_auth_token() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like