Selaa lähdekoodia

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

Abraham Elmahrek 12 vuotta sitten
vanhempi
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 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
 
 LOG = logging.getLogger(__name__)
@@ -84,6 +86,10 @@ def _save_links(workflow, root):
     if not isinstance(child_el.tag, basestring):
       continue
 
+    # Skip kill nodes.
+    if child_el.tag.endswith('kill'):
+      continue
+
     # Iterate over node members
     # Join nodes have attributes which point to the next node
     # Start node has attribute which points to first node
@@ -97,6 +103,7 @@ def _save_links(workflow, root):
 
     elif isinstance(parent, Decision):
       _decision_relationships(workflow, parent, child_el)
+
     else:
       _node_relationships(workflow, parent, child_el)
 
@@ -187,6 +194,7 @@ def _node_relationships(workflow, parent, child_el):
   """
   Resolves node links.
   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:
     # 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)
         to = el.attrib['start']
         name = 'start'
+      elif name == 'error':
+        to = 'kill'
       else:
         if 'to' not in el.attrib:
           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):
-  # Deserialize
+  """
+  Deserialize
+  """
   objs = serializers.deserialize('xml', etree.tostring(root))
 
   # 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):
+  """
+  Save nodes, but skip kill nodes because we create a single kill node to use.
+  """
   for node in nodes:
+    if node.node_type is 'kill':
+      continue
+
     try:
       # Do not overwrite start or end node
       Node.objects.get(workflow=workflow, node_type=node.node_type, name=node.name)
     except Node.DoesNotExist:
       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):
   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()
 
 
+  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):
 
   def setUp(self):