فهرست منبع

[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: