fix(controller): dag continue on failed. Fixes #2596 (#2597)

USERS.md

@@ -57,6 +57,7 @@ Currently, the following organizations are **officially** using Argo Workflows:
 1. [Riskified](https://www.riskified.com)
 1. [SAP Fieldglass](https://www.fieldglass.com/)
 1. [SAP Hybris](https://cx.sap.com/)
+1. [SegmentStream](https://segmentstream.com)
 1. [Sidecar Technologies](https://hello.getsidecar.com/)
 1. [Styra](https://www.styra.com/)
 1. [Threekit](https://www.threekit.com/)

test/e2e/functional_test.go

@@ -78,6 +78,78 @@ spec:
 		})
 }
 
+func (s *FunctionalSuite) TestContinueOnFailDag() {
+	s.Given().
+		Workflow(`
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: continue-on-failed-dag
+  labels:
+    argo-e2e: true
+spec:
+  entrypoint: workflow-ignore
+  parallelism: 2
+  templates:
+    - name: workflow-ignore
+      dag:
+        failFast: false
+        tasks:
+          - name: A
+            template: whalesay
+          - name: B
+            template: boom
+            continueOn:
+              failed: true
+            dependencies:
+              - A
+          - name: C
+            template: whalesay
+            dependencies:
+              - A
+          - name: D
+            template: whalesay
+            dependencies:
+              - B
+              - C
+
+    - name: boom
+      dag:
+        tasks:
+          - name: B-1
+            template: whalesplosion
+
+    - name: whalesay
+      container:
+        imagePullPolicy: IfNotPresent
+        image: docker/whalesay:latest
+
+    - name: whalesplosion
+      container:
+        imagePullPolicy: IfNotPresent
+        image: docker/whalesay:latest
+        command: ["sh", "-c", "sleep 10; exit 1"]
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(30 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.NodeFailed, status.Phase)
+			assert.Len(t, status.Nodes, 6)
+
+			bStatus := status.Nodes.FindByDisplayName("B")
+			if assert.NotNil(t, bStatus) {
+				assert.Equal(t, wfv1.NodeFailed, bStatus.Phase)
+			}
+
+			dStatus := status.Nodes.FindByDisplayName("D")
+			if assert.NotNil(t, dStatus) {
+				assert.Equal(t, wfv1.NodeSucceeded, dStatus.Phase)
+			}
+		})
+}
+
 func (s *FunctionalSuite) TestFastFailOnPodTermination() {
 	// TODO: Test fails due to using a service account with insufficient permissions, skipping for now
 	// pods is forbidden: User "system:serviceaccount:argo:default" cannot list resource "pods" in API group "" in the namespace "argo"

workflow/controller/dag.go

@@ -83,7 +83,10 @@ func (d *dagContext) assertBranchFinished(targetTaskNames []string) bool {
 				return d.assertBranchFinished(taskObject.Dependencies)
 			}
 		} else if !taskNode.Successful() {
-			flag = true
+			taskObject := d.getTask(targetTaskName)
+			if !taskObject.ContinuesOn(taskNode.Phase) {
+				flag = true
+			}
 		}
 
 		// In failFast situation, if node is successful, it will run to leaf node, above