Repository: argoproj/argo
Pull request #3403 (2020-07-07 16:28:49)

fix: Ensure non-leaf DAG tasks have their onExit handler's run
Fixes: https://github.com/argoproj/argo/issues/3395


Patch:
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index ac286b2c7..efb3a19e2 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -366,11 +366,12 @@ func (woc *wfOperationCtx) executeDAGTask(dagCtx *dagContext, taskName string) {
 			connectDependencies(nodeName)
 			return
 		}
+		// Recurse into all of this node's dependencies
+		for _, dep := range taskDependencies {
+			woc.executeDAGTask(dagCtx, dep)
+		}
 		if !proceed {
-			// This node's dependencies are not completed yet, recurse into them, then return
-			for _, dep := range taskDependencies {
-				woc.executeDAGTask(dagCtx, dep)
-			}
+			// This node's dependencies are not completed yet, return
 			return
 		}
 		if !execute {
diff --git a/workflow/metrics/metrics.go b/workflow/metrics/metrics.go
index a61ec250c..819c3ce8b 100644
--- a/workflow/metrics/metrics.go
+++ b/workflow/metrics/metrics.go
@@ -2,6 +2,7 @@ package metrics
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -34,6 +35,8 @@ type metric struct {
 }
 
 type Metrics struct {
+	// Ensures mutual exclusion in workflows map
+	workflowsMutex  sync.Mutex
 	metricsConfig   ServerConfig
 	telemetryConfig ServerConfig
 
@@ -89,6 +92,9 @@ func (m *Metrics) allMetrics() []prometheus.Metric {
 }
 
 func (m *Metrics) WorkflowAdded(key string, phase v1alpha1.NodePhase) {
+	m.workflowsMutex.Lock()
+	defer m.workflowsMutex.Unlock()
+
 	if m.workflows[key] {
 		return
 	}
@@ -99,7 +105,10 @@ func (m *Metrics) WorkflowAdded(key string, phase v1alpha1.NodePhase) {
 }
 
 func (m *Metrics) WorkflowUpdated(key string, fromPhase, toPhase v1alpha1.NodePhase) {
-	if fromPhase == toPhase || !m.workflows[key] {
+	m.workflowsMutex.Lock()
+	hasKey := m.workflows[key]
+	m.workflowsMutex.Unlock()
+	if fromPhase == toPhase || !hasKey {
 		return
 	}
 	m.WorkflowDeleted(key, fromPhase)
@@ -107,6 +116,9 @@ func (m *Metrics) WorkflowUpdated(key string, fromPhase, toPhase v1alpha1.NodePh
 }
 
 func (m *Metrics) WorkflowDeleted(key string, phase v1alpha1.NodePhase) {
+	m.workflowsMutex.Lock()
+	defer m.workflowsMutex.Unlock()
+
 	if !m.workflows[key] {
 		return
 	}


Test patch:
diff --git a/workflow/controller/dag_test.go b/workflow/controller/dag_test.go
index 7ec11902e..9cb21e162 100644
--- a/workflow/controller/dag_test.go
+++ b/workflow/controller/dag_test.go
@@ -1643,3 +1643,130 @@ func TestOnExitDAGPhase(t *testing.T) {
 
 	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
 }
+
+var testOnExitNonLeaf = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: exit-handler-bug-example
+spec:
+  arguments: {}
+  entrypoint: dag
+  templates:
+  - arguments: {}
+    dag:
+      tasks:
+      - arguments: {}
+        name: step-2
+        onExit: on-exit
+        template: step-template
+      - arguments: {}
+        dependencies:
+        - step-2
+        name: step-3
+        onExit: on-exit
+        template: step-template
+    inputs: {}
+    metadata: {}
+    name: dag
+    outputs: {}
+  - arguments: {}
+    container:
+      args:
+      - echo exit-handler-step-{{pod.name}}
+      command:
+      - sh
+      - -c
+      image: alpine:latest
+      name: ""
+      resources: {}
+    inputs: {}
+    metadata: {}
+    name: on-exit
+    outputs: {}
+  - arguments: {}
+    container:
+      args:
+      - echo step {{pod.name}}
+      command:
+      - sh
+      - -c
+      image: alpine:latest
+      name: ""
+      resources: {}
+    inputs: {}
+    metadata: {}
+    name: step-template
+    outputs: {}
+status:
+  nodes:
+    exit-handler-bug-example:
+      children:
+      - exit-handler-bug-example-3054913383
+      displayName: exit-handler-bug-example
+      finishedAt: "2020-07-07T16:15:54Z"
+      id: exit-handler-bug-example
+      name: exit-handler-bug-example
+      outboundNodes:
+      - exit-handler-bug-example-3038135764
+      phase: Running
+      startedAt: "2020-07-07T16:15:33Z"
+      templateName: dag
+      templateScope: local/exit-handler-bug-example
+      type: DAG
+    exit-handler-bug-example-3054913383:
+      boundaryID: exit-handler-bug-example
+      displayName: step-2
+      finishedAt: "2020-07-07T16:15:37Z"
+      hostNodeName: minikube
+      id: exit-handler-bug-example-3054913383
+      name: exit-handler-bug-example.step-2
+      outputs:
+        artifacts:
+        - archiveLogs: true
+          name: main-logs
+          s3:
+            accessKeySecret:
+              key: accesskey
+              name: my-minio-cred
+            bucket: my-bucket
+            endpoint: minio:9000
+            insecure: true
+            key: exit-handler-bug-example/exit-handler-bug-example-3054913383/main.log
+            secretKeySecret:
+              key: secretkey
+              name: my-minio-cred
+        exitCode: "0"
+      phase: Succeeded
+      resourcesDuration:
+        cpu: 4
+        memory: 2
+      startedAt: "2020-07-07T16:15:33Z"
+      templateName: step-template
+      templateScope: local/exit-handler-bug-example
+      type: Pod
+  phase: Running
+  startedAt: "2020-07-07T16:15:33Z"
+`
+
+func TestOnExitNonLeaf(t *testing.T) {
+	cancel, controller := newController()
+	defer cancel()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	wf := unmarshalWF(testOnExitNonLeaf)
+	wf, err := wfcset.Create(wf)
+	assert.NoError(t, err)
+	woc := newWorkflowOperationCtx(wf, controller)
+
+	woc.operate()
+	retryNode := woc.getNodeByName("step-2.onExit")
+	if assert.NotNil(t, retryNode) {
+		assert.Equal(t, wfv1.NodePending, retryNode.Phase)
+	}
+	retryNode = woc.getNodeByName("exit-handler-bug-example.step-3")
+	if assert.NotNil(t, retryNode) {
+		assert.Equal(t, wfv1.NodePending, retryNode.Phase)
+	}
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+}