openstack/taskflow
Code Issues Proposed changes

We were not notifying when errors occured so fix that.

Browse Source 
Previously we seemed to have forgotten to notify the listener
when a task errored and allowed said listeners to get the reason
why the task failed so fix that and allow for the exception to
be returned back on result fetching and rethrown.
This commit is contained in:
Joshua Harlow
2013-05-31 22:58:33 -07:00
parent a194c8ea52
commit 520f7ace49
3 changed files with 42 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
taskflow/job.py
+14 -12
View File
@@ -124,7 +124,7 @@ class Job(object):
book so that it can be retrieved later."""
metadata = None
flow_details = self.logbook[flow.name]
if state == states.SUCCESS:
if state in (states.SUCCESS, states.FAILURE):
metadata = {
'result': result,
}
@@ -134,17 +134,19 @@ class Job(object):
def _task_result_fetcher(self, _context, flow, task):
flow_details = self.logbook[flow.name]
# See if it completed before so that we can use its results instead
# of having to recompute them.
task_state = task_and_state(task, states.SUCCESS)
if task_state in flow_details:
# TODO(harlowja): should we be a little more cautious about
# duplicate task results? Maybe we shouldn't allow them to
# have the same name in the first place?
task_details = flow_details[task_state][0]
if task_details.metadata and 'result' in task_details.metadata:
return (True, task_details.metadata['result'])
return (False, None)
# See if it completed before (or failed before) so that we can use its
# results instead of having to recompute it.
for s in (states.SUCCESS, states.FAILURE):
name = task_and_state(task, s)
if name in flow_details:
# TODO(harlowja): should we be a little more cautious about
# duplicate task results? Maybe we shouldn't allow them to
# have the same name in the first place?
details = flow_details[name][0]
if details.metadata and 'result' in details.metadata:
return (True, s == states.FAILURE,
details.metadata['result'])
return (False, False, None)
def associate(self, flow, parents=True):
"""Attachs the needed resumption and state change tracking listeners
taskflow/patterns/ordered_flow.py
+26 -6
View File
@@ -72,7 +72,7 @@ class Flow(object):
# to call the parents...
self.parents = parents
# This should be a functor that returns whether a given task has
# already ran by returning a pair of (has_result, result).
# already ran by returning a pair of (has_result, was_error, result).
#
# NOTE(harlowja): This allows for resumption by skipping tasks which
# have already occurred. The previous return value is needed due to
@@ -135,7 +135,7 @@ class Flow(object):
LOG.exception("Dropping exception catched when"
" notifying about ordering failure.")
def run_task(task, result=None, simulate_run=False):
def run_task(task, failed=False, result=None, simulate_run=False):
try:
self._on_task_start(context, task)
if not simulate_run:
@@ -144,6 +144,20 @@ class Flow(object):
inputs = {}
inputs.update(kwargs)
result = task(context, *args, **inputs)
else:
if failed:
if not result:
# If no exception or exception message was provided
# or captured from the previous run then we need to
# form one for this task.
result = "%s failed running." % (task)
if isinstance(result, basestring):
result = exc.InvalidStateException(result)
if not isinstance(result, Exception):
LOG.warn("Can not raise a non-exception"
" object: %s", result)
result = exc.InvalidStateException()
raise result
# Keep a pristine copy of the result
# so that if said result is altered by other further
# states the one here will not be. This ensures that
@@ -160,6 +174,11 @@ class Flow(object):
except Exception as e:
cause = FlowFailure(task, self, e)
with excutils.save_and_reraise_exception():
try:
self._on_task_error(context, task, e)
except Exception:
LOG.exception("Dropping exception catched when"
" notifying about task failure.")
self.rollback(context, cause)
last_task = 0
@@ -170,14 +189,15 @@ class Flow(object):
if self.state == states.INTERRUPTED:
was_interrupted = True
break
(has_result, result) = result_fetcher(self, task)
(has_result, was_error, result) = result_fetcher(self, task)
if not has_result:
break
# Fake running the task so that we trigger the same
# notifications and state changes (and rollback that
# would have happened in a normal flow).
last_task = i + 1
run_task(task, result=result, simulate_run=True)
run_task(task, failed=was_error, result=result,
simulate_run=True)
if was_interrupted:
return
@@ -216,10 +236,10 @@ class Flow(object):
for f in self.listeners:
f(context, self, old_state)
def _on_task_error(self, context, task):
def _on_task_error(self, context, task, exc):
# Notify any listeners that the task has errored.
for f in self.task_listeners:
f(context, states.FAILURE, self, task)
f(context, states.FAILURE, self, task, result=exc)
def _on_task_start(self, context, task):
# Notify any listeners that we are about to start the given task.
taskflow/tests/unit/test_linear_flow.py
+2 -2
View File
@@ -208,8 +208,8 @@ class LinearFlowTest(unittest.TestCase):
def result_fetcher(_ctx, _wf, task):
if task.name in result_storage:
return (True, result_storage.get(task.name))
return (False, None)
return (True, False, result_storage.get(task.name))
return (False, False, None)
def task_listener(_ctx, state, _wf, task, result=None):
if state not in (states.SUCCESS, states.FAILURE,):