You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ariatosca.apache.org by ra...@apache.org on 2016/11/15 02:26:38 UTC
[03/14] incubator-ariatosca git commit: ARIA-20 Add support for
ignoring task failures
ARIA-20 Add support for ignoring task failures
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/b9c135f8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/b9c135f8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/b9c135f8
Branch: refs/heads/ARIA-18-migrate-tosca-parser
Commit: b9c135f86feda60c318ae585a1ec4aa0c22a31d8
Parents: eb3dd5d
Author: Dan Kilman <da...@gigaspaces.com>
Authored: Mon Nov 14 13:24:13 2016 +0200
Committer: Dan Kilman <da...@gigaspaces.com>
Committed: Mon Nov 14 13:26:41 2016 +0200
----------------------------------------------------------------------
aria/context/workflow.py | 2 ++
aria/events/builtin_event_handler.py | 8 +++++++-
aria/storage/models.py | 1 +
aria/workflows/api/task.py | 3 +++
aria/workflows/core/engine.py | 2 +-
aria/workflows/core/task.py | 1 +
tests/mock/context.py | 6 ++++--
tests/workflows/api/test_task.py | 8 ++++++--
tests/workflows/core/test_engine.py | 27 +++++++++++++++++++++++++--
9 files changed, 50 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/aria/context/workflow.py
----------------------------------------------------------------------
diff --git a/aria/context/workflow.py b/aria/context/workflow.py
index fb9c8ee..b84a5fb 100644
--- a/aria/context/workflow.py
+++ b/aria/context/workflow.py
@@ -49,6 +49,7 @@ class WorkflowContext(logger.LoggerMixin):
parameters=None,
task_max_attempts=1,
task_retry_interval=0,
+ task_ignore_failure=False,
**kwargs):
super(WorkflowContext, self).__init__(**kwargs)
self.name = name
@@ -61,6 +62,7 @@ class WorkflowContext(logger.LoggerMixin):
self.parameters = parameters or {}
self.task_max_attempts = task_max_attempts
self.task_retry_interval = task_retry_interval
+ self.task_ignore_failure = task_ignore_failure
# TODO: execution creation should happen somewhere else
# should be moved there, when such logical place exists
try:
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/aria/events/builtin_event_handler.py
----------------------------------------------------------------------
diff --git a/aria/events/builtin_event_handler.py b/aria/events/builtin_event_handler.py
index cba3180..2f9a3be 100644
--- a/aria/events/builtin_event_handler.py
+++ b/aria/events/builtin_event_handler.py
@@ -55,7 +55,13 @@ def _task_started(task, *args, **kwargs):
@on_failure_task_signal.connect
def _task_failed(task, *args, **kwargs):
with task.update():
- if task.retry_count < task.max_attempts - 1 or task.max_attempts == task.INFINITE_RETRIES:
+ should_retry = (
+ (task.retry_count < task.max_attempts - 1 or
+ task.max_attempts == task.INFINITE_RETRIES) and
+ # ignore_failure check here means the task will not be retries and it will be marked as
+ # failed. The engine will also look at ignore_failure so it won't fail the workflow.
+ not task.ignore_failure)
+ if should_retry:
task.status = task.RETRYING
task.retry_count += 1
task.due_at = datetime.utcnow() + timedelta(seconds=task.retry_interval)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/aria/storage/models.py
----------------------------------------------------------------------
diff --git a/aria/storage/models.py b/aria/storage/models.py
index 23f2408..94a9aa0 100644
--- a/aria/storage/models.py
+++ b/aria/storage/models.py
@@ -432,6 +432,7 @@ class Task(Model):
max_attempts = Field(type=int, default=1, validation_func=_Validation.validate_max_attempts)
retry_count = Field(type=int, default=0)
retry_interval = Field(type=(int, float), default=0)
+ ignore_failure = Field(type=bool, default=False)
# Operation specific fields
name = Field(type=basestring)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/aria/workflows/api/task.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/task.py b/aria/workflows/api/task.py
index bb8e045..7e9b8e0 100644
--- a/aria/workflows/api/task.py
+++ b/aria/workflows/api/task.py
@@ -60,6 +60,7 @@ class OperationTask(BaseTask):
node_instance,
max_attempts=None,
retry_interval=None,
+ ignore_failure=None,
inputs=None):
"""
Creates an operation task using the name, details, node instance and any additional kwargs.
@@ -77,6 +78,8 @@ class OperationTask(BaseTask):
if max_attempts is None else max_attempts)
self.retry_interval = (self.workflow_context.task_retry_interval
if retry_interval is None else retry_interval)
+ self.ignore_failure = (self.workflow_context.task_ignore_failure
+ if ignore_failure is None else ignore_failure)
class WorkflowTask(BaseTask):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/aria/workflows/core/engine.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/engine.py b/aria/workflows/core/engine.py
index b32ef11..83ad097 100644
--- a/aria/workflows/core/engine.py
+++ b/aria/workflows/core/engine.py
@@ -108,7 +108,7 @@ class Engine(logger.LoggerMixin):
self._executor.execute(task)
def _handle_ended_tasks(self, task):
- if task.status == models.Task.FAILED:
+ if task.status == models.Task.FAILED and not task.ignore_failure:
raise exceptions.ExecutorException('Workflow failed')
else:
self._execution_graph.remove_node(task.id)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/aria/workflows/core/task.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/task.py b/aria/workflows/core/task.py
index b90306a..65a8ddb 100644
--- a/aria/workflows/core/task.py
+++ b/aria/workflows/core/task.py
@@ -98,6 +98,7 @@ class OperationTask(BaseTask, logger.LoggerMixin):
execution_id=self.workflow_context.execution_id,
max_attempts=api_task.max_attempts,
retry_interval=api_task.retry_interval,
+ ignore_failure=api_task.ignore_failure
)
self.workflow_context.model.task.store(task)
self._task_id = task.id
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/tests/mock/context.py
----------------------------------------------------------------------
diff --git a/tests/mock/context.py b/tests/mock/context.py
index 4d218b5..15dbc33 100644
--- a/tests/mock/context.py
+++ b/tests/mock/context.py
@@ -19,11 +19,11 @@ from . import models
from ..storage import InMemoryModelDriver
-def simple():
+def simple(**kwargs):
storage = application_model_storage(InMemoryModelDriver())
storage.setup()
storage.deployment.store(models.get_deployment())
- return context.workflow.WorkflowContext(
+ final_kwargs = dict(
name='simple_context',
model_storage=storage,
resource_storage=None,
@@ -33,3 +33,5 @@ def simple():
task_max_attempts=models.TASK_MAX_ATTEMPTS,
task_retry_interval=models.TASK_RETRY_INTERVAL
)
+ final_kwargs.update(kwargs)
+ return context.workflow.WorkflowContext(**final_kwargs)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/tests/workflows/api/test_task.py
----------------------------------------------------------------------
diff --git a/tests/workflows/api/test_task.py b/tests/workflows/api/test_task.py
index 857f3bf..50bca45 100644
--- a/tests/workflows/api/test_task.py
+++ b/tests/workflows/api/test_task.py
@@ -69,6 +69,7 @@ class TestOperationTask(object):
inputs = {'inputs': True}
max_attempts = 10
retry_interval = 10
+ ignore_failure = True
with context.workflow.current.push(workflow_context):
model_task = api.task.OperationTask(name=name,
@@ -76,7 +77,8 @@ class TestOperationTask(object):
node_instance=node_instance,
inputs=inputs,
max_attempts=max_attempts,
- retry_interval=retry_interval)
+ retry_interval=retry_interval,
+ ignore_failure=ignore_failure)
assert model_task.name == name
assert model_task.operation_details == op_details
@@ -84,9 +86,10 @@ class TestOperationTask(object):
assert model_task.inputs == inputs
assert model_task.retry_interval == retry_interval
assert model_task.max_attempts == max_attempts
+ assert model_task.ignore_failure == ignore_failure
def test_operation_task_default_values(self):
- workflow_context = mock.context.simple()
+ workflow_context = mock.context.simple(task_ignore_failure=True)
with context.workflow.current.push(workflow_context):
model_task = api.task.OperationTask(
name='stub',
@@ -96,6 +99,7 @@ class TestOperationTask(object):
assert model_task.inputs == {}
assert model_task.retry_interval == workflow_context.task_retry_interval
assert model_task.max_attempts == workflow_context.task_max_attempts
+ assert model_task.ignore_failure == workflow_context.task_ignore_failure
class TestWorkflowTask(object):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/b9c135f8/tests/workflows/core/test_engine.py
----------------------------------------------------------------------
diff --git a/tests/workflows/core/test_engine.py b/tests/workflows/core/test_engine.py
index 744e155..6377a69 100644
--- a/tests/workflows/core/test_engine.py
+++ b/tests/workflows/core/test_engine.py
@@ -54,7 +54,11 @@ class BaseTest(object):
tasks_graph=graph)
@staticmethod
- def _op(func, ctx, inputs=None, max_attempts=None, retry_interval=None):
+ def _op(func, ctx,
+ inputs=None,
+ max_attempts=None,
+ retry_interval=None,
+ ignore_failure=None):
return api.task.OperationTask(
name='task',
operation_details={'operation': 'tests.workflows.core.test_engine.{name}'.format(
@@ -62,7 +66,8 @@ class BaseTest(object):
node_instance=ctx.model.node_instance.get('dependency_node_instance'),
inputs=inputs,
max_attempts=max_attempts,
- retry_interval=retry_interval
+ retry_interval=retry_interval,
+ ignore_failure=ignore_failure
)
@pytest.fixture(scope='function', autouse=True)
@@ -372,6 +377,24 @@ class TestRetries(BaseTest):
assert invocation2 - invocation1 >= retry_interval
assert global_test_holder.get('sent_task_signal_calls') == 2
+ def test_ignore_failure(self, workflow_context, executor):
+ @workflow
+ def mock_workflow(ctx, graph):
+ op = self._op(mock_conditional_failure_task, ctx,
+ ignore_failure=True,
+ inputs={'failure_count': 100},
+ max_attempts=100)
+ graph.add_tasks(op)
+ self._execute(
+ workflow_func=mock_workflow,
+ workflow_context=workflow_context,
+ executor=executor)
+ assert workflow_context.states == ['start', 'success']
+ assert workflow_context.exception is None
+ invocations = global_test_holder.get('invocations', [])
+ assert len(invocations) == 1
+ assert global_test_holder.get('sent_task_signal_calls') == 1
+
def mock_success_task():
pass