# -*- coding: utf-8 -*- # Copyright (C) 2014 Yahoo! Inc. All Rights Reserved. # # 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. import functools from taskflow.engines.action_engine import compiler from taskflow import exceptions as exc from taskflow import states from taskflow.types import tree from taskflow.utils import misc def _cached_get(cache, cache_key, atom_name, fetch_func, *args, **kwargs): """Tries to get a previously saved value or fetches it and caches it.""" value, value_found = None, False try: value, value_found = cache[cache_key][atom_name] except KeyError: try: value = fetch_func(*args, **kwargs) value_found = True except (exc.StorageFailure, exc.NotFound): pass cache[cache_key][atom_name] = value, value_found return value, value_found def _fetch_predecessor_tree(graph, atom): """Creates a tree of predecessors, rooted at given atom.""" root = tree.Node(atom) stack = [(root, atom)] while stack: parent, node = stack.pop() for pred_node in graph.predecessors(node): pred_node_data = graph.nodes[pred_node] if pred_node_data['kind'] == compiler.FLOW_END: # Jump over and/or don't show flow end nodes... for pred_pred_node in graph.predecessors(pred_node): stack.append((parent, pred_pred_node)) else: child = tree.Node(pred_node, **pred_node_data) parent.add(child) # And go further backwards... stack.append((child, pred_node)) return root class FailureFormatter(object): """Formats a failure and connects it to associated atoms & engine.""" _BUILDERS = { states.EXECUTE: (_fetch_predecessor_tree, 'predecessors'), } def __init__(self, engine, hide_inputs_outputs_of=()): self._hide_inputs_outputs_of = hide_inputs_outputs_of self._engine = engine def _format_node(self, storage, cache, node): """Formats a single tree node into a string version.""" if node.metadata['kind'] == compiler.FLOW: flow = node.item flow_name = flow.name return "Flow '%s'" % (flow_name) elif node.metadata['kind'] in compiler.ATOMS: atom = node.item atom_name = atom.name atom_attrs = {} intention, intention_found = _cached_get( cache, 'intentions', atom_name, storage.get_atom_intention, atom_name) if intention_found: atom_attrs['intention'] = intention state, state_found = _cached_get(cache, 'states', atom_name, storage.get_atom_state, atom_name) if state_found: atom_attrs['state'] = state if atom_name not in self._hide_inputs_outputs_of: # When the cache does not exist for this atom this # will be called with the rest of these arguments # used to populate the cache. fetch_mapped_args = functools.partial( storage.fetch_mapped_args, atom.rebind, atom_name=atom_name, optional_args=atom.optional) requires, requires_found = _cached_get(cache, 'requires', atom_name, fetch_mapped_args) if requires_found: atom_attrs['requires'] = requires provides, provides_found = _cached_get( cache, 'provides', atom_name, storage.get_execute_result, atom_name) if provides_found: atom_attrs['provides'] = provides if atom_attrs: return "Atom '%s' %s" % (atom_name, atom_attrs) else: return "Atom '%s'" % (atom_name) else: raise TypeError("Unable to format node, unknown node" " kind '%s' encountered" % node.metadata['kind']) def format(self, fail, atom_matcher): """Returns a (exc_info, details) tuple about the failure. The ``exc_info`` tuple should be a standard three element (exctype, value, traceback) tuple that will be used for further logging. A non-empty string is typically returned for ``details``; it should contain any string info about the failure (with any specific details the ``exc_info`` may not have/contain). """ buff = misc.StringIO() storage = self._engine.storage compilation = self._engine.compilation if fail.exc_info is None: # Remote failures will not have a 'exc_info' tuple, so just use # the captured traceback that was captured by the creator when it # failed... buff.write_nl(fail.pformat(traceback=True)) if storage is None or compilation is None: # Somehow we got called before prepared and/or compiled; ok # that's weird, skip doing the rest... return (fail.exc_info, buff.getvalue()) hierarchy = compilation.hierarchy graph = compilation.execution_graph atom_node = hierarchy.find_first_match(atom_matcher) atom = None atom_intention = None if atom_node is not None: atom = atom_node.item atom_intention = storage.get_atom_intention(atom.name) if atom is not None and atom_intention in self._BUILDERS: # Cache as much as we can, since the path of various atoms # may cause the same atom to be seen repeatedly depending on # the graph structure... cache = { 'intentions': {}, 'provides': {}, 'requires': {}, 'states': {}, } builder, kind = self._BUILDERS[atom_intention] rooted_tree = builder(graph, atom) child_count = rooted_tree.child_count(only_direct=False) buff.write_nl('%s %s (most recent first):' % (child_count, kind)) formatter = functools.partial(self._format_node, storage, cache) direct_child_count = rooted_tree.child_count(only_direct=True) for i, child in enumerate(rooted_tree, 1): if i == direct_child_count: buff.write(child.pformat(stringify_node=formatter, starting_prefix=" ")) else: buff.write_nl(child.pformat(stringify_node=formatter, starting_prefix=" ")) return (fail.exc_info, buff.getvalue())