Преглед изворни кода

[oozie] Import workflow improvements

Link resolution throws more verbose errors.
Allow end nodes to have any name.
Allow kill nodes to have any name.
abec пре 13 година
родитељ
комит
e748e7a
2 измењених фајлова са 117 додато и 42 уклоњено
  1. 115 40
      apps/oozie/src/oozie/import_workflow.py
  2. 2 2
      apps/oozie/src/oozie/models.py

+ 115 - 40
apps/oozie/src/oozie/import_workflow.py

@@ -52,6 +52,7 @@ OOZIE_NAMESPACES = ['uri:oozie:workflow:0.1', 'uri:oozie:workflow:0.2', 'uri:ooz
 
 LINKS = ('ok', 'error', 'path')
 
+
 def _save_links(workflow, root):
   """
   Iterates over all links in the passed XML doc and creates links.
@@ -78,56 +79,26 @@ def _save_links(workflow, root):
   Note: Nodes are looked up by workflow and name.
   """
   # Iterate over nodes
-  for node in root:
+  for child_el in root:
+    # Skip special nodes (like comments).
+    if not isinstance(child_el.tag, basestring):
+      continue
+
     # Iterate over node members
     # Join nodes have attributes which point to the next node
     # Start node has attribute which points to first node
-    parent = Node.objects.get(workflow=workflow, name=node.attrib.get('name', xml_tag(node))).get_full_node()
+    parent = Node.objects.get(workflow=workflow, name=child_el.attrib.get('name', xml_tag(child_el))).get_full_node()
 
     if isinstance(parent, Start):
-      workflow.start = parent
-      to = node.attrib['to']
-      child = Node.objects.get(workflow=workflow, name=to)
-      obj = Link.objects.create(name='to', parent=parent, child=child)
-      obj.save()
+      _start_relationships(workflow, parent, child_el)
 
     elif isinstance(parent, Join):
-      to = node.attrib['to']
-      child = Node.objects.get(workflow=workflow, name=to)
-      obj = Link.objects.create(name='to', parent=parent, child=child)
-      obj.save()
+      _join_relationships(workflow, parent, child_el)
 
     elif isinstance(parent, Decision):
-      for switch in node:
-        for case in switch:
-          to = case.attrib['to']
-          child = Node.objects.get(workflow=workflow, name=to)
-
-          if xml_tag(case) == 'default':
-            name = 'default'
-            obj = Link.objects.create(name=name, parent=parent, child=child)
-
-          else:
-            name = 'start'
-            comment = case.text.strip()
-            obj = Link.objects.create(name=name, parent=parent, child=child, comment=comment)
-
-          obj.save()
-
+      _decision_relationships(workflow, parent, child_el)
     else:
-      for el in node:
-        # Links
-        name = xml_tag(el)
-        if name in LINKS:
-          if name == 'path':
-            to = el.attrib['start']
-            name = 'start'
-          else:
-            to = el.attrib['to']
-
-          child = Node.objects.get(workflow=workflow, name=to)
-          obj = Link.objects.create(name=name, parent=parent, child=child)
-          obj.save()
+      _node_relationships(workflow, parent, child_el)
 
   workflow.end = End.objects.get(workflow=workflow).get_full_node()
   workflow.save()
@@ -135,6 +106,109 @@ def _save_links(workflow, root):
   _resolve_fork_relationships(workflow)
   _resolve_decision_relationships(workflow)
 
+def _start_relationships(workflow, parent, child_el):
+  """
+  Resolve start node links.
+  Will always use 'to' link type.
+  """
+  if 'to' not in child_el.attrib:
+    raise RuntimeError(_("Node %s has a link that is missing 'to' attribute.") % parent.name)
+
+  workflow.start = parent
+  to = child_el.attrib['to']
+
+  try:
+    child = Node.objects.get(workflow=workflow, name=to)
+  except Node.DoesNotExist, e:
+    raise RuntimeError(_("Node %s has not been defined.") % to)
+
+  obj = Link.objects.create(name='to', parent=parent, child=child)
+  obj.save()
+
+def _join_relationships(workflow, parent, child_el):
+  """
+  Resolves join node links.
+  Will always use 'to' link type.
+  """
+  if 'to' not in child_el.attrib:
+    raise RuntimeError(_("Node %s has a link that is missing 'to' attribute.") % parent.name)
+
+  to = child_el.attrib['to']
+
+  try:
+    child = Node.objects.get(workflow=workflow, name=to)
+  except Node.DoesNotExist, e:
+    raise RuntimeError(_("Node %s has not been defined.") % to)
+
+  obj = Link.objects.create(name='to', parent=parent, child=child)
+  obj.save()
+
+def _decision_relationships(workflow, parent, child_el):
+  """
+  Resolves the switch statement like nature of decision nodes.
+  Will use 'to' link type, except for default case.
+  """
+  for switch in child_el:
+    # Skip special nodes (like comments).
+    if not isinstance(switch.tag, basestring):
+      continue
+
+    for case in switch:
+      # Skip special nodes (like comments).
+      if not isinstance(case.tag, basestring):
+        continue
+
+      if 'to' not in case.attrib:
+        raise RuntimeError(_("Node %s has a link that is missing 'to' attribute.") % parent.name)
+
+      to = case.attrib['to']
+      try:
+        child = Node.objects.get(workflow=workflow, name=to)
+      except Node.DoesNotExist, e:
+        raise RuntimeError(_("Node %s has not been defined.") % to)
+
+      if xml_tag(case) == 'default':
+        name = 'default'
+        obj = Link.objects.create(name=name, parent=parent, child=child)
+
+      else:
+        name = 'start'
+        comment = case.text.strip()
+        obj = Link.objects.create(name=name, parent=parent, child=child, comment=comment)
+
+      obj.save()
+
+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.
+  """
+  for el in child_el:
+    # Skip special nodes (like comments).
+    if not isinstance(el.tag, basestring):
+      continue
+
+    # Links
+    name = xml_tag(el)
+    if name in LINKS:
+      if name == 'path':
+        if 'start' not in el.attrib:
+          raise RuntimeError(_("Node %s has a link that is missing 'start' attribute.") % parent.name)
+        to = el.attrib['start']
+        name = 'start'
+      else:
+        if 'to' not in el.attrib:
+          raise RuntimeError(_("Node %s has a link that is missing 'to' attribute.") % parent.name)
+        to = el.attrib['to']
+
+      try:
+        child = Node.objects.get(workflow=workflow, name=to)
+      except Node.DoesNotExist, e:
+        raise RuntimeError("Node %s has not been defined" % to)
+
+      obj = Link.objects.create(name=name, parent=parent, child=child)
+      obj.save()
+
 
 def _resolve_fork_relationships(workflow):
   """
@@ -444,6 +518,7 @@ def import_workflow(workflow, workflow_definition, fs=None):
 
   # Transform XML using XSLT
   transformed_root = transform(workflow_definition_root)
+  print etree.tostring(transformed_root, pretty_print=True)
 
   # Resolve workflow dependencies and node types and link dependencies
   nodes = _prepare_nodes(workflow, transformed_root)

+ 2 - 2
apps/oozie/src/oozie/models.py

@@ -378,8 +378,8 @@ class Workflow(Job):
 
   def get_hierarchy(self):
     node = self.start
-    return self.get_hierarchy_rec(node=node) + [[Kill.objects.get(name='kill', workflow=node.workflow)],
-                                           [End.objects.get(name='end', workflow=node.workflow)]]
+    return self.get_hierarchy_rec(node=node) + [[Kill.objects.get(workflow=node.workflow)],
+                                           [End.objects.get(workflow=node.workflow)]]
 
   def get_hierarchy_rec(self, node=None):
     if node is None: