Browse Source

HUE-1184 [oozie] Merge kill nodes into single kill node

Abraham Elmahrek 12 years ago
parent
commit
e8928f3

+ 25 - 2
apps/oozie/src/oozie/import_workflow.py

@@ -43,7 +43,9 @@ from django.core import serializers
 from django.utils.translation import ugettext as _
 from django.utils.translation import ugettext as _
 
 
 from conf import DEFINITION_XSLT_DIR
 from conf import DEFINITION_XSLT_DIR
-from models import Workflow, Node, Link, Start, End, Decision, DecisionEnd, Fork, Join
+from models import Workflow, Node, Link, Start, End,\
+                   Decision, DecisionEnd, Fork, Join,\
+                   Kill
 from utils import xml_tag
 from utils import xml_tag
 
 
 LOG = logging.getLogger(__name__)
 LOG = logging.getLogger(__name__)
@@ -84,6 +86,10 @@ def _save_links(workflow, root):
     if not isinstance(child_el.tag, basestring):
     if not isinstance(child_el.tag, basestring):
       continue
       continue
 
 
+    # Skip kill nodes.
+    if child_el.tag.endswith('kill'):
+      continue
+
     # Iterate over node members
     # Iterate over node members
     # Join nodes have attributes which point to the next node
     # Join nodes have attributes which point to the next node
     # Start node has attribute which points to first node
     # Start node has attribute which points to first node
@@ -97,6 +103,7 @@ def _save_links(workflow, root):
 
 
     elif isinstance(parent, Decision):
     elif isinstance(parent, Decision):
       _decision_relationships(workflow, parent, child_el)
       _decision_relationships(workflow, parent, child_el)
+
     else:
     else:
       _node_relationships(workflow, parent, child_el)
       _node_relationships(workflow, parent, child_el)
 
 
@@ -187,6 +194,7 @@ def _node_relationships(workflow, parent, child_el):
   """
   """
   Resolves node links.
   Resolves node links.
   Will use 'start' link type for fork nodes and 'to' link type for all other nodes.
   Will use 'start' link type for fork nodes and 'to' link type for all other nodes.
+  Error links will automatically resolve to a single kill node.
   """
   """
   for el in child_el:
   for el in child_el:
     # Skip special nodes (like comments).
     # Skip special nodes (like comments).
@@ -201,6 +209,8 @@ def _node_relationships(workflow, parent, child_el):
           raise RuntimeError(_("Node %s has a link that is missing 'start' attribute.") % parent.name)
           raise RuntimeError(_("Node %s has a link that is missing 'start' attribute.") % parent.name)
         to = el.attrib['start']
         to = el.attrib['start']
         name = 'start'
         name = 'start'
+      elif name == 'error':
+        to = 'kill'
       else:
       else:
         if 'to' not in el.attrib:
         if 'to' not in el.attrib:
           raise RuntimeError(_("Node %s has a link that is missing 'to' attribute.") % parent.name)
           raise RuntimeError(_("Node %s has a link that is missing 'to' attribute.") % parent.name)
@@ -406,7 +416,9 @@ def _resolve_decision_relationships(workflow):
 
 
 
 
 def _prepare_nodes(workflow, root):
 def _prepare_nodes(workflow, root):
-  # Deserialize
+  """
+  Deserialize
+  """
   objs = serializers.deserialize('xml', etree.tostring(root))
   objs = serializers.deserialize('xml', etree.tostring(root))
 
 
   # First pass is a list of nodes and their types respectively.
   # First pass is a list of nodes and their types respectively.
@@ -501,13 +513,24 @@ def _resolve_subworkflow_from_deployment_dir(fs, workflow, app_path):
 
 
 
 
 def _save_nodes(workflow, nodes):
 def _save_nodes(workflow, nodes):
+  """
+  Save nodes, but skip kill nodes because we create a single kill node to use.
+  """
   for node in nodes:
   for node in nodes:
+    if node.node_type is 'kill':
+      continue
+
     try:
     try:
       # Do not overwrite start or end node
       # Do not overwrite start or end node
       Node.objects.get(workflow=workflow, node_type=node.node_type, name=node.name)
       Node.objects.get(workflow=workflow, node_type=node.node_type, name=node.name)
     except Node.DoesNotExist:
     except Node.DoesNotExist:
       node.save()
       node.save()
 
 
+  # Create kill node
+  # Only need it if we have a node to reference it with.
+  if len(nodes) > 2:
+    Kill.objects.create(name='kill', workflow=workflow, node_type=Kill.node_type)
+
 
 
 def import_workflow(workflow, workflow_definition, fs=None):
 def import_workflow(workflow, workflow_definition, fs=None):
   xslt_definition_fh = open("%(xslt_dir)s/workflow.xslt" % {
   xslt_definition_fh = open("%(xslt_dir)s/workflow.xslt" % {

+ 33 - 0
apps/oozie/src/oozie/test_data/0.4/test-java-multiple-kill.xml

@@ -0,0 +1,33 @@
+<workflow-app name="Sequential Java" xmlns="uri:oozie:workflow:0.4">
+    <start to="TeraGenWorkflow"/>
+    <action name="TeraGenWorkflow">
+        <java>
+            <job-tracker>${jobTracker}</job-tracker>
+            <name-node>${nameNode}</name-node>
+            <main-class>org.apache.hadoop.examples.terasort.TeraGen</main-class>
+            <arg>${records}</arg>
+            <arg>${output_dir}/teragen</arg>
+            <capture-output/>
+        </java>
+        <ok to="TeraSort"/>
+        <error to="kill1"/>
+    </action>
+    <action name="TeraSort">
+        <java>
+            <job-tracker>${jobTracker}</job-tracker>
+            <name-node>${nameNode}</name-node>
+            <main-class>org.apache.hadoop.examples.terasort.TeraSort</main-class>
+            <arg>${output_dir}/teragen</arg>
+            <arg>${output_dir}/terasort</arg>
+        </java>
+        <ok to="end"/>
+        <error to="kill2"/>
+    </action>
+    <kill name="kill1">
+        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+    </kill>
+    <kill name="kill2">
+        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+    </kill>
+    <end name="end"/>
+</workflow-app>

+ 26 - 0
apps/oozie/src/oozie/tests.py

@@ -1674,6 +1674,32 @@ class TestImportWorkflow04(OozieMockBase):
     workflow.delete()
     workflow.delete()
 
 
 
 
+  def test_import_multi_kill_node(self):
+    """
+    Validates import for multiple kill nodes: xml.
+
+    Kill nodes should be skipped and a single kill node should be created.
+    """
+    workflow = Workflow.objects.new_workflow(self.user)
+    workflow.save()
+    f = open('apps/oozie/src/oozie/test_data/0.4/test-java-multiple-kill.xml')
+    import_workflow(workflow, f.read())
+    f.close()
+    workflow.save()
+    assert_equal('kill', Kill.objects.get(workflow=workflow).name)
+    assert_equal(5, len(Node.objects.filter(workflow=workflow)))
+    assert_equal(6, len(Link.objects.filter(parent__workflow=workflow)))
+    nodes = [Node.objects.filter(workflow=workflow, node_type='java')[0].get_full_node(),
+             Node.objects.filter(workflow=workflow, node_type='java')[1].get_full_node()]
+    assert_equal('org.apache.hadoop.examples.terasort.TeraGen', nodes[0].main_class)
+    assert_equal('${records} ${output_dir}/teragen', nodes[0].args)
+    assert_equal('org.apache.hadoop.examples.terasort.TeraSort', nodes[1].main_class)
+    assert_equal('${output_dir}/teragen ${output_dir}/terasort', nodes[1].args)
+    assert_true(nodes[0].capture_output)
+    assert_false(nodes[1].capture_output)
+    workflow.delete()
+
+
 class TestPermissions(OozieBase):
 class TestPermissions(OozieBase):
 
 
   def setUp(self):
   def setUp(self):