mistral/mistral/engine/task_handler.py

# Copyright 2015 - Mirantis, Inc.
# Copyright 2015 - StackStorm, Inc.
# Copyright 2016 - Nokia Networks.
# Copyright 2016 - Brocade Communications Systems, Inc.
#
#    Licensed 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.

from oslo_log import log as logging
from osprofiler import profiler
import traceback as tb

from mistral.db.v2 import api as db_api
from mistral.engine import tasks
from mistral.engine import workflow_handler as wf_handler
from mistral import exceptions as exc
from mistral.services import scheduler
from mistral.workbook import parser as spec_parser
from mistral.workflow import base as wf_base
from mistral.workflow import commands as wf_cmds
from mistral.workflow import states


"""Responsible for running tasks and handling results."""

LOG = logging.getLogger(__name__)

_CHECK_TASK_START_ALLOWED_PATH = (
    'mistral.engine.task_handler._check_task_start_allowed'
)


@profiler.trace('task-handler-run-task')
def run_task(wf_cmd):
    """Runs workflow task.

    :param wf_cmd: Workflow command.
    """

    task = _build_task_from_command(wf_cmd)

    try:
        task.run()
    except exc.MistralException as e:
        wf_ex = wf_cmd.wf_ex
        task_spec = wf_cmd.task_spec

        msg = (
            "Failed to run task [wf=%s, task=%s]: %s\n%s" %
            (wf_ex, task_spec.get_name(), e, tb.format_exc())
        )

        LOG.error(msg)

        task.set_state(states.ERROR, msg)

        wf_handler.fail_workflow(wf_ex, msg)

        return

    if task.is_waiting():
        _schedule_check_task_start_allowed(task.task_ex)

    if task.is_completed():
        wf_handler.schedule_on_task_complete(task.task_ex)


@profiler.trace('task-handler-on-task-complete')
def on_action_complete(action_ex):
    """Handles action completion event.

    :param action_ex: Action execution.
    """

    task_ex = action_ex.task_execution

    if not task_ex:
        return

    task_spec = spec_parser.get_task_spec(task_ex.spec)

    wf_ex = task_ex.workflow_execution

    task = _create_task(
        wf_ex,
        spec_parser.get_workflow_spec_by_id(wf_ex.workflow_id),
        task_spec,
        task_ex.in_context,
        task_ex
    )

    try:
        task.on_action_complete(action_ex)
    except exc.MistralException as e:
        wf_ex = task_ex.workflow_execution

        msg = ("Failed to handle action completion [wf=%s, task=%s,"
               " action=%s]: %s\n%s" %
               (wf_ex.name, task_ex.name, action_ex.name, e, tb.format_exc()))

        LOG.error(msg)

        task.set_state(states.ERROR, msg)

        wf_handler.fail_workflow(wf_ex, msg)

        return

    if task.is_completed():
        wf_handler.schedule_on_task_complete(task_ex)


def fail_task(task_ex, msg):
    task = _build_task_from_execution(
        spec_parser.get_workflow_spec_by_id(task_ex.workflow_id),
        task_ex
    )

    task.set_state(states.ERROR, msg)

    wf_handler.fail_workflow(task_ex.workflow_execution, msg)


def continue_task(task_ex):
    task = _build_task_from_execution(
        spec_parser.get_workflow_spec_by_id(task_ex.workflow_id),
        task_ex
    )

    try:
        task.set_state(states.RUNNING, None)

        task.run()
    except exc.MistralException as e:
        wf_ex = task_ex.workflow_execution

        msg = (
            "Failed to run task [wf=%s, task=%s]: %s\n%s" %
            (wf_ex, task_ex.name, e, tb.format_exc())
        )

        LOG.error(msg)

        task.set_state(states.ERROR, msg)

        wf_handler.fail_workflow(wf_ex, msg)

        return

    if task.is_completed():
        wf_handler.schedule_on_task_complete(task_ex)


def complete_task(task_ex, state, state_info):
    task = _build_task_from_execution(
        spec_parser.get_workflow_spec_by_id(task_ex.workflow_id),
        task_ex
    )

    try:
        task.complete(state, state_info)
    except exc.MistralException as e:
        wf_ex = task_ex.workflow_execution

        msg = (
            "Failed to complete task [wf=%s, task=%s]: %s\n%s" %
            (wf_ex, task_ex.name, e, tb.format_exc())
        )

        LOG.error(msg)

        task.set_state(states.ERROR, msg)

        wf_handler.fail_workflow(wf_ex, msg)

        return

    if task.is_completed():
        wf_handler.schedule_on_task_complete(task_ex)


def _build_task_from_execution(wf_spec, task_ex, task_spec=None):
    return _create_task(
        task_ex.workflow_execution,
        wf_spec,
        task_spec or spec_parser.get_task_spec(task_ex.spec),
        task_ex.in_context,
        task_ex
    )


@profiler.trace('task-handler-build-task-from-command')
def _build_task_from_command(cmd):
    if isinstance(cmd, wf_cmds.RunExistingTask):
        task = _create_task(
            cmd.wf_ex,
            cmd.wf_spec,
            spec_parser.get_task_spec(cmd.task_ex.spec),
            cmd.ctx,
            task_ex=cmd.task_ex,
            unique_key=cmd.task_ex.unique_key
        )

        if cmd.reset:
            task.reset()

        return task

    if isinstance(cmd, wf_cmds.RunTask):
        task = _create_task(
            cmd.wf_ex,
            cmd.wf_spec,
            cmd.task_spec,
            cmd.ctx,
            unique_key=cmd.unique_key
        )

        if cmd.is_waiting():
            task.defer()

        return task

    raise exc.MistralError('Unsupported workflow command: %s' % cmd)


def _create_task(wf_ex, wf_spec, task_spec, ctx, task_ex=None,
                 unique_key=None):
    if task_spec.get_with_items():
        return tasks.WithItemsTask(
            wf_ex,
            wf_spec,
            task_spec,
            ctx,
            task_ex,
            unique_key
        )

    return tasks.RegularTask(
        wf_ex,
        wf_spec,
        task_spec,
        ctx,
        task_ex,
        unique_key
    )


def _check_task_start_allowed(task_ex_id):
    with db_api.transaction():
        task_ex = db_api.get_task_execution(task_ex_id)

        wf_ctrl = wf_base.get_controller(
            task_ex.workflow_execution,
            spec_parser.get_workflow_spec_by_id(task_ex.workflow_id)
        )

        if wf_ctrl.is_task_start_allowed(task_ex):
            continue_task(task_ex)

            return

        # TODO(rakhmerov): Algorithm for increasing rescheduling delay.
        _schedule_check_task_start_allowed(task_ex, 1)


def _schedule_check_task_start_allowed(task_ex, delay=0):
    """Schedules task preconditions check.

    This method provides transactional decoupling of task preconditions
    check from events that can potentially satisfy those preconditions.

    It's needed in non-locking model in order to avoid 'phantom read'
    phenomena when reading state of multiple tasks to see if a task that
    depends on them can start. Just starting a separate transaction
    without using scheduler is not safe due to concurrency window that
    we'll have in this case (time between transactions) whereas scheduler
    is a special component that is designed to be resistant to failures.

    :param task_ex: Task execution.
    :param delay: Delay.
    :return:
    """
    key = 'th_c_t_s_a-%s' % task_ex.id

    scheduler.schedule_call(
        None,
        _CHECK_TASK_START_ALLOWED_PATH,
        delay,
        unique_key=key,
        task_ex_id=task_ex.id
    )