ad07ba0d68
* Transaction in on_action_complete() must not be splitted into 2 parts, it caused the bug with after task completion logic * Fix executor behavior so that it doesn't send an error back to engine if a error came from engine itself. It should report back only errors occurred with an action itself. * YAQL and other expected Mistral exceptions in transitions should not lead to transaction rollback and rollback of action result. For example if action result came and it's valid but while evaluating transition conditions we got a YAQL exception then action result should be stored normally w/o transaction rollback and corresponding task and workflow should fail with corresponding state_info. * Fixed all tests * Minor cosmetic changes Closes-Bug: #1524477 Change-Id: I09086e40a5902bbb6c977bf195cb035e31f21246
152 lines
4.1 KiB
Python
152 lines
4.1 KiB
Python
# Copyright 2015 - Mirantis, Inc.
|
|
# Copyright 2015 - StackStorm, 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 mistral.workbook import parser as spec_parser
|
|
from mistral.workbook.v2 import tasks
|
|
from mistral.workflow import states
|
|
|
|
|
|
class WorkflowCommand(object):
|
|
"""Workflow command.
|
|
|
|
A set of workflow commands form a communication protocol between workflow
|
|
handler and its clients. When workflow handler makes a decision about
|
|
how to continue a workflow it returns a set of commands so that a caller
|
|
knows what to do next.
|
|
"""
|
|
|
|
def __init__(self, wf_ex, task_spec, ctx):
|
|
self.wf_ex = wf_ex
|
|
self.task_spec = task_spec
|
|
self.ctx = ctx or {}
|
|
|
|
|
|
class Noop(WorkflowCommand):
|
|
"""No-operation command."""
|
|
|
|
def __repr__(self):
|
|
return "NOOP [workflow=%s]" % self.wf_ex.name
|
|
|
|
|
|
class RunTask(WorkflowCommand):
|
|
"""Instruction to run a workflow task."""
|
|
|
|
def __init__(self, wf_ex, task_spec, ctx):
|
|
super(RunTask, self).__init__(wf_ex, task_spec, ctx)
|
|
self.wait_flag = False
|
|
|
|
def is_waiting(self):
|
|
return (self.wait_flag and
|
|
isinstance(self.task_spec, tasks.DirectWorkflowTaskSpec) and
|
|
self.task_spec.get_join())
|
|
|
|
def __repr__(self):
|
|
return (
|
|
"Run task [workflow=%s, task=%s, waif_flag=%s]"
|
|
% (self.wf_ex.name, self.task_spec.get_name(), self.wait_flag)
|
|
)
|
|
|
|
|
|
class RunExistingTask(WorkflowCommand):
|
|
"""Command for running already existent task."""
|
|
|
|
def __init__(self, task_ex, reset=True):
|
|
super(RunExistingTask, self).__init__(
|
|
task_ex.workflow_execution,
|
|
spec_parser.get_task_spec(task_ex.spec),
|
|
task_ex.in_context
|
|
)
|
|
|
|
self.task_ex = task_ex
|
|
self.reset = reset
|
|
|
|
|
|
class SetWorkflowState(WorkflowCommand):
|
|
"""Instruction to change a workflow state."""
|
|
|
|
def __init__(self, wf_ex, task_spec, ctx, new_state, msg):
|
|
super(SetWorkflowState, self).__init__(wf_ex, task_spec, ctx)
|
|
|
|
self.new_state = new_state
|
|
self.msg = msg
|
|
|
|
|
|
class FailWorkflow(SetWorkflowState):
|
|
"""Instruction to fail a workflow."""
|
|
|
|
def __init__(self, wf_ex, task_spec, ctx, msg=None):
|
|
super(FailWorkflow, self).__init__(
|
|
wf_ex,
|
|
task_spec,
|
|
ctx,
|
|
states.ERROR,
|
|
msg
|
|
)
|
|
|
|
def __repr__(self):
|
|
return "Fail [workflow=%s]" % self.wf_ex.name
|
|
|
|
|
|
class SucceedWorkflow(SetWorkflowState):
|
|
"""Instruction to succeed a workflow."""
|
|
|
|
def __init__(self, wf_ex, task_spec, ctx, msg=None):
|
|
super(SucceedWorkflow, self).__init__(
|
|
wf_ex,
|
|
task_spec,
|
|
ctx,
|
|
states.SUCCESS,
|
|
msg
|
|
)
|
|
|
|
def __repr__(self):
|
|
return "Succeed [workflow=%s]" % self.wf_ex.name
|
|
|
|
|
|
class PauseWorkflow(SetWorkflowState):
|
|
"""Instruction to pause a workflow."""
|
|
|
|
def __init__(self, wf_ex, task_spec, ctx, msg=None):
|
|
super(PauseWorkflow, self).__init__(
|
|
wf_ex,
|
|
task_spec,
|
|
ctx,
|
|
states.PAUSED,
|
|
msg
|
|
)
|
|
|
|
def __repr__(self):
|
|
return "Pause [workflow=%s]" % self.wf_ex.name
|
|
|
|
|
|
RESERVED_CMDS = dict(zip(
|
|
tasks.RESERVED_TASK_NAMES, [
|
|
Noop,
|
|
FailWorkflow,
|
|
SucceedWorkflow,
|
|
PauseWorkflow
|
|
]
|
|
))
|
|
|
|
|
|
def get_command_class(cmd_name):
|
|
return RESERVED_CMDS[cmd_name] if cmd_name in RESERVED_CMDS else None
|
|
|
|
|
|
def create_command(cmd_name, wf_ex, task_spec, ctx):
|
|
cmd_cls = get_command_class(cmd_name) or RunTask
|
|
|
|
return cmd_cls(wf_ex, task_spec, ctx)
|