Browse Source

[oozie] Workflow breaks when two branches join to the same node

Disable 'convert to decision node' view if there are some nodes below the join
Test for creation of cycle when editing a workflow
Added some tests
Romain Rigaux 13 years ago
parent
commit
d42549305a
3 changed files with 94 additions and 6 deletions
  1. 22 0
      apps/oozie/src/oozie/models.py
  2. 62 4
      apps/oozie/src/oozie/tests.py
  3. 10 2
      apps/oozie/src/oozie/views/editor.py

+ 22 - 0
apps/oozie/src/oozie/models.py

@@ -382,6 +382,28 @@ class Workflow(Job):
 
     return copy
 
+
+  def has_cycle(self):
+    """
+    Topological sort for detecting cycles in the directed graph.
+    """
+    queue = set([self.start])
+    removed_edges = set()
+
+    while queue:
+      node = queue.pop()
+      edges = set(node.get_children_links())
+      for edge in edges:
+        removed_edges.add(edge)
+        # Edge has no other incoming edges
+        if not set(edge.child.get_parent_links()) - removed_edges:
+          queue.add(edge.child)
+
+    graph_edges = set([edge for node in self.node_set.all() for edge in node.get_children_links()])
+
+    return len(graph_edges - removed_edges) > 0 # Graph does not have unseen edges
+
+
   def find_parameters(self):
     params = set()
 

+ 62 - 4
apps/oozie/src/oozie/tests.py

@@ -255,6 +255,43 @@ class TestEditor:
     for field in node.PARAM_FIELDS:
       assert_equal(translation_regex.sub(r'${\1}', getattr(jobsub_design.get_root_action(), field)), getattr(node, field))
 
+
+  def test_workflow_has_cycle(self):
+    action1 = Node.objects.get(name='action-name-1')
+    action2 = Node.objects.get(name='action-name-2')
+    action3 = Node.objects.get(name='action-name-3')
+
+    assert_false(self.wf.has_cycle())
+
+    ok = action3.get_link('ok')
+    ok.child = action1
+    ok.save()
+
+    assert_true(self.wf.has_cycle())
+
+
+  def test_workflow_has_cycle_in_fork(self):
+    action1 = Node.objects.get(name='action-name-1')
+    action2 = Node.objects.get(name='action-name-2')
+    action3 = Node.objects.get(name='action-name-3')
+    action4 = add_action(self.wf.id, action3.id, 'action-name-4')
+
+    move_up(self.c, self.wf, action2)
+    move_up(self.c, self.wf, action4)
+
+    # start
+    # 1 2
+    # 3 4
+
+    assert_false(self.wf.has_cycle())
+
+    ok = action4.get_link('ok')
+    ok.child = action2
+    ok.save()
+
+    assert_true(self.wf.has_cycle())
+
+
   def test_decision_node(self):
     action1 = Node.objects.get(name='action-name-1')
     action2 = Node.objects.get(name='action-name-2')
@@ -262,13 +299,17 @@ class TestEditor:
 
     move_down(self.c, self.wf, action1)
     fork = action1.get_parent()
+    assert_false(fork.has_decisions())
 
     # 1 2
     #  3
     response = self.c.get(reverse('oozie:edit_workflow_fork', args=[fork.id]), {}, follow=True)
-    assert_equal(200, response.status_code)
+    assert_true('this Fork has some other actions below' in response.content, response.content)
 
-    assert_false(fork.has_decisions())
+    self.c.post(reverse('oozie:delete_action', args=[action3.id]), {})
+
+    response = self.c.get(reverse('oozie:edit_workflow_fork', args=[fork.id]), {}, follow=True)
+    assert_false('this Fork has some other actions below' in response.content, response.content)
 
     # Missing information for converting to decision
     response = self.c.post(reverse('oozie:edit_workflow_fork', args=[fork.id]), {
@@ -277,7 +318,6 @@ class TestEditor:
         u'form-1-comment': [u''], u'form-1-id': [u'%s' % action2.id],
         u'child': [u'%s' % self.wf.end.id]}, follow=True)
     assert_true('This field is required' in response.content, response.content)
-    assert_equal(200, response.status_code)
     assert_false(fork.has_decisions())
 
     # Convert to decision
@@ -286,10 +326,10 @@ class TestEditor:
         u'form-0-comment': [u'output'], u'form-0-id': [u'%s' % action1.id],
         u'form-1-comment': [u'output'], u'form-1-id': [u'%s' % action2.id],
         u'child': [u'%s' % self.wf.end.id]}, follow=True)
-    assert_equal(200, response.status_code)
 
     raise SkipTest
     # Mystery below, link_formset.save() does not appear to save the links during a test
+    assert_true('Decision' in response.content, response.content)
     assert_equal(Fork.ACTION_DECISION_TYPE, fork.node_type)
     assert_true(fork.has_decisions(), response.content)
 
@@ -302,6 +342,12 @@ class TestEditor:
         '        <map-reduce>\n'
         '           <job-tracker>${jobTracker}</job-tracker>\n'
         '            <name-node>${nameNode}</name-node>\n'
+        '            <configuration>\n'
+        '                <property>\n'
+        '                    <name>sleep</name>\n'
+        '                    <value>${SLEEP}</value>\n'
+        '                </property>\n'
+        '            </configuration>\n'
         '        </map-reduce>\n'
         '        <ok to="action-name-2"/>\n'
         '        <error to="kill"/>\n'
@@ -310,6 +356,12 @@ class TestEditor:
         '        <map-reduce>\n'
         '            <job-tracker>${jobTracker}</job-tracker>\n'
         '            <name-node>${nameNode}</name-node>\n'
+        '            <configuration>\n'
+        '                <property>\n'
+        '                    <name>sleep</name>\n'
+        '                    <value>${SLEEP}</value>\n'
+        '                </property>\n'
+        '            </configuration>\n'
         '        </map-reduce>\n'
         '        <ok to="action-name-3"/>\n'
         '        <error to="kill"/>\n'
@@ -318,6 +370,12 @@ class TestEditor:
         '        <map-reduce>\n'
         '            <job-tracker>${jobTracker}</job-tracker>\n'
         '            <name-node>${nameNode}</name-node>\n'
+        '            <configuration>\n'
+        '                <property>\n'
+        '                    <name>sleep</name>\n'
+        '                    <value>${SLEEP}</value>\n'
+        '                </property>\n'
+        '            </configuration>\n'
         '        </map-reduce>\n'
         '        <ok to="end"/>\n'
         '        <error to="kill"/>\n'

+ 10 - 2
apps/oozie/src/oozie/views/editor.py

@@ -239,6 +239,10 @@ def edit_workflow(request, workflow):
       if workflow_form.is_valid() and actions_formset.is_valid():
         workflow_form.save()
         actions_formset.save()
+
+        if workflow.has_cycle():
+          raise PopupException(_('Sorry, this operation is not creating a cycle which would break the workflow.'))
+
         return redirect(reverse('oozie:list_workflows'))
     except Exception, e:
       request.error(_('Sorry, this operation is not supported: %(error)s') % {'error': e})
@@ -442,7 +446,6 @@ def import_action(request, workflow, parent_action_id):
 @check_action_edition_permission
 def edit_workflow_fork(request, action):
   fork = action
-
   LinkFormSet = modelformset_factory(Link, form=LinkForm, max_num=0)
 
   if request.method == 'POST':
@@ -463,7 +466,12 @@ def edit_workflow_fork(request, action):
 
       return redirect(reverse('oozie:edit_workflow', kwargs={'workflow': fork.workflow.id}))
   else:
-    link_formset = LinkFormSet(queryset=fork.get_children_links().exclude(name__in=['related', 'default']))
+    if filter(lambda link: link.child.id != action.workflow.end.id,
+              [link for link in fork.get_child_join().get_children_links()]):
+      raise PopupException(_('Sorry, this Fork has some other actions below its Join and cannot be converted. '
+                             'Please delete the nodes below the Join.'))
+
+    link_formset = LinkFormSet(queryset=fork.get_children_links())
     default_link = Link(parent=fork, name='default', comment='default')
     default_link_form = DefaultLinkForm(action=fork, instance=default_link)