Fix failFast bug: When a node in the middle fails, the entire workflo…

…w will hang (argoproj#1468)

test/e2e/expectedfailures/dag-disbale-failFast-2.yaml

@@ -0,0 +1,54 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+ generateName: dag-primay-branch-
+spec:
+ entrypoint: statis
+ templates:
+ - name: a
+ container:
+ image: docker/whalesay:latest
+ command: [cowsay]
+ args: ["hello world"]
+ - name: b
+ retryStrategy:
+ limit: 2
+ container:
+ image: alpine:latest
+ command: [sh, -c]
+ args: ["sleep 30; echo haha"]
+ - name: c
+ retryStrategy:
+ limit: 3
+ container:
+ image: alpine:latest
+ command: [sh, -c]
+ args: ["echo intentional failure; exit 2"]
+ - name: d
+ container:
+ image: alpine:latest
+ command: [sh, -c]
+ args: ["echo intentional failure; exit 2"]
+ - name: e
+ container:
+ image: alpine:latest
+ command: [sh, -c]
+ args: ["sleep 30; echo haha"]
+ - name: statis
+ dag:
+ failFast: false
+ tasks:
+ - name: A
+ template: a
+ - name: B
+ dependencies: [A]
+ template: b
+ - name: C
+ dependencies: [A]
+ template: c
+ - name: D
+ dependencies: [B]
+ template: d
+ - name: E
+ dependencies: [D]
+ template: e

workflow/controller/dag.go

@@ -61,6 +61,30 @@ func (d *dagContext) getTaskNode(taskName string) *wfv1.NodeStatus {
  return &node
 }
 
+// Assert all branch finished for failFast:disable function
+func (d *dagContext) assertBranchFinished(targetTaskName string) bool {
+ // We should ensure that from the bottom to the top,
+ // all the nodes of this branch have at least one failure.
+ // If successful, we should continue to run down until the leaf node
+ taskNode := d.getTaskNode(targetTaskName)
+ if taskNode == nil {
+ taskObject := d.getTask(targetTaskName)
+ if taskObject != nil {
+ // Make sure all the dependency node have one failed
+ for _, tmpTaskName := range taskObject.Dependencies {
+ // Recursive check until top root node
+ return d.assertBranchFinished(tmpTaskName)
+ }
+ }
+ } else if !taskNode.Successful() {
+ return true
+ }
+
+ // In failFast situation, if node is successful, it will run to leaf node, above
+ // the function, we have already check the leaf node status
+ return false
+}
+
 // assessDAGPhase assesses the overall DAG status
 func (d *dagContext) assessDAGPhase(targetTasks []string, nodes map[string]wfv1.NodeStatus) wfv1.NodePhase {
  // First check all our nodes to see if anything is still running. If so, then the DAG is
@@ -93,17 +117,20 @@ func (d *dagContext) assessDAGPhase(targetTasks []string, nodes map[string]wfv1.
  if d.tmpl.DAG.FailFast != nil && !*d.tmpl.DAG.FailFast {
  tmpOverAllFinished := true
  // If all the nodes have finished, we should mark the failed node to finish overall workflow
- // So we should check all the targetTasks have finished
+ // So we should check all the targetTasks branch have finished
  for _, tmpDepName := range targetTasks {
  tmpDepNode := d.getTaskNode(tmpDepName)
  if tmpDepNode == nil {
+ // If leaf node is nil, we should check it's parent node and recursive check
+ if !d.assertBranchFinished(tmpDepName) {
+ tmpOverAllFinished = false
+ }
+ } else if tmpDepNode.Type == wfv1.NodeTypeRetry && hasMoreRetries(tmpDepNode, d.wf) {
  tmpOverAllFinished = false
  break
  }
- if tmpDepNode.Type == wfv1.NodeTypeRetry && hasMoreRetries(tmpDepNode, d.wf) {
- tmpOverAllFinished = false
- break
- }
+
+ //If leaf node has finished, we should mark the error workflow
  }
  if !tmpOverAllFinished {
  return wfv1.NodeRunning