Repository: argoproj/argo (go)

Issue:
Argo OnExit does not work properly when using step template in a dag
## Summary 

Argo OnExit does not work properly when using step template in a dag

## Diagnostics

Argo version: v2.9.4

```
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: exit
    container:
      image: docker/whalesay:latest
      command: [cowsay]
      args: ["exit!"]
  - name: steps
    steps:
      - - name: step-a
          template: a
  - name: statis
    dag:
      tasks:
      - name: A
        onExit: exit
        template: steps # <--- if change "a", it works!!
```

In above case, exit will never be called.

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.


Pull request description:
fix: Ensure target task's onExit handlers are run
Fixes #3710

Patch:
diff --git a/docs/windows.md b/docs/windows.md
index 4d0de559b..e90f8b5a1 100644
--- a/docs/windows.md
+++ b/docs/windows.md
@@ -80,6 +80,27 @@ hello-hybrid-plqpp-1977432187: "Hello from Windows Container!"
 hello-hybrid-plqpp-764774907: Hello from Linux Container!
 ```
 
+## Artifact mount path
+
+Artifacts work mostly the same way as on Linux. All paths get automatically mapped to the `C:` drive. For example:
+
+```yaml
+ # ...
+    - name: print-message
+      inputs:
+        artifacts:
+          # unpack the message input artifact
+          # and put it at C:\message
+          - name: message
+            path: "/message" # gets mapped to C:\message
+      nodeSelector:
+        kubernetes.io/os: windows
+      container:
+        image: mcr.microsoft.com/windows/nanoserver:1809
+        command: ["cmd", "/c"]
+        args: ["dir C:\\message"]   # List the C:\message directory
+```
+
 ## Building the workflow executor image for Windows
 
 To build the workflow executor image for Windows you need a Windows machine running Windows Server 2019 with Docker installed like described [in the docs](https://docs.docker.com/ee/docker-ee/windows/docker-ee/#install-docker-engine---enterprise).
@@ -90,4 +111,4 @@ You then clone the project and run the Docker build with the Dockerfile for Wind
 git clone https://github.com/argoproj/argo.git
 cd argo
 docker build -t myargoexec -f .\Dockerfile.windows --target argoexec .
-```
\ No newline at end of file
+```
diff --git a/workflow/controller/dag.go b/workflow/controller/dag.go
index 447620d69..7e7a31e94 100644
--- a/workflow/controller/dag.go
+++ b/workflow/controller/dag.go
@@ -232,6 +232,21 @@ func (woc *wfOperationCtx) executeDAG(nodeName string, tmplCtx *templateresoluti
 	// kick off execution of each target task asynchronously
 	for _, taskName := range targetTasks {
 		woc.executeDAGTask(dagCtx, taskName)
+
+		// It is possible that target tasks are not reconsidered (i.e. executeDAGTask is not called on them) once they are
+		// complete (since the DAG itself will have succeeded). To ensure that their exit handlers are run we also run them here. Note that
+		// calls to runOnExitNode are idempotent: it is fine if they are called more than once for the same task.
+		taskNode := dagCtx.getTaskNode(taskName)
+		if taskNode != nil && taskNode.Fulfilled() {
+			if taskNode.Completed() {
+				// Run the node's onExit node, if any. Since this is a target task, we don't need to consider the status
+				// of the onExit node before continuing. That will be done in assesDAGPhase
+				_, _, err := woc.runOnExitNode(dagCtx.GetTask(taskName).OnExit, taskName, taskNode.Name, dagCtx.boundaryID, dagCtx.tmplCtx)
+				if err != nil {
+					return node, err
+				}
+			}
+		}
 	}
 
 	// check if we are still running any tasks in this dag and return early if we do


Test patch:
diff --git a/workflow/controller/dag_test.go b/workflow/controller/dag_test.go
index 2c71470b8..c29c90d40 100644
--- a/workflow/controller/dag_test.go
+++ b/workflow/controller/dag_test.go
@@ -1875,3 +1875,158 @@ func TestDagOptionalInputArtifacts(t *testing.T) {
 		assert.Equal(t, wfv1.NodePending, optionalInputArtifactsNode.Phase)
 	}
 }
+
+var testDagTargetTaskOnExit = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: dag-primay-branch-6bnnl
+spec:
+  arguments: {}
+  entrypoint: statis
+  templates:
+  - arguments: {}
+    container:
+      args:
+      - hello world
+      command:
+      - cowsay
+      image: docker/whalesay:latest
+      name: ""
+      resources: {}
+    inputs: {}
+    metadata: {}
+    name: a
+    outputs: {}
+  - arguments: {}
+    container:
+      args:
+      - exit!
+      command:
+      - cowsay
+      image: docker/whalesay:latest
+      name: ""
+      resources: {}
+    inputs: {}
+    metadata: {}
+    name: exit
+    outputs: {}
+  - arguments: {}
+    inputs: {}
+    metadata: {}
+    name: steps
+    outputs: {}
+    steps:
+    - - arguments: {}
+        name: step-a
+        template: a
+  - arguments: {}
+    dag:
+      tasks:
+      - arguments: {}
+        name: A
+        onExit: exit
+        template: steps
+    inputs: {}
+    metadata: {}
+    name: statis
+    outputs: {}
+status:
+  nodes:
+    dag-primay-branch-6bnnl:
+      children:
+      - dag-primay-branch-6bnnl-1650817843
+      displayName: dag-primay-branch-6bnnl
+      finishedAt: "2020-08-10T14:30:19Z"
+      id: dag-primay-branch-6bnnl
+      name: dag-primay-branch-6bnnl
+      outboundNodes:
+      - dag-primay-branch-6bnnl-1181733215
+      phase: Running
+      startedAt: "2020-08-10T14:30:08Z"
+      templateName: statis
+      templateScope: local/dag-primay-branch-6bnnl
+      type: DAG
+    dag-primay-branch-6bnnl-1181733215:
+      boundaryID: dag-primay-branch-6bnnl-1650817843
+      displayName: step-a
+      finishedAt: "2020-08-10T14:30:11Z"
+      hostNodeName: minikube
+      id: dag-primay-branch-6bnnl-1181733215
+      name: dag-primay-branch-6bnnl.A[0].step-a
+      outputs:
+        artifacts:
+        - archiveLogs: true
+          name: main-logs
+          s3:
+            accessKeySecret:
+              key: accesskey
+              name: my-minio-cred
+            bucket: my-bucket
+            endpoint: minio:9000
+            insecure: true
+            key: dag-primay-branch-6bnnl/dag-primay-branch-6bnnl-1181733215/main.log
+            secretKeySecret:
+              key: secretkey
+              name: my-minio-cred
+        exitCode: "0"
+      phase: Succeeded
+      resourcesDuration:
+        cpu: 2
+        memory: 1
+      startedAt: "2020-08-10T14:30:08Z"
+      templateName: a
+      templateScope: local/dag-primay-branch-6bnnl
+      type: Pod
+    dag-primay-branch-6bnnl-1650817843:
+      boundaryID: dag-primay-branch-6bnnl
+      children:
+      - dag-primay-branch-6bnnl-3841864351
+      - dag-primay-branch-6bnnl-1342580575
+      displayName: A
+      finishedAt: "2020-08-10T14:30:13Z"
+      id: dag-primay-branch-6bnnl-1650817843
+      name: dag-primay-branch-6bnnl.A
+      outboundNodes:
+      - dag-primay-branch-6bnnl-1181733215
+      phase: Running
+      startedAt: "2020-08-10T14:30:08Z"
+      templateName: steps
+      templateScope: local/dag-primay-branch-6bnnl
+      type: Steps
+    dag-primay-branch-6bnnl-3841864351:
+      boundaryID: dag-primay-branch-6bnnl-1650817843
+      children:
+      - dag-primay-branch-6bnnl-1181733215
+      displayName: '[0]'
+      finishedAt: "2020-08-10T14:30:13Z"
+      id: dag-primay-branch-6bnnl-3841864351
+      name: dag-primay-branch-6bnnl.A[0]
+      phase: Running
+      startedAt: "2020-08-10T14:30:08Z"
+      templateName: steps
+      templateScope: local/dag-primay-branch-6bnnl
+      type: StepGroup
+  phase: Running
+  resourcesDuration:
+    cpu: 5
+    memory: 2
+  startedAt: "2020-08-10T14:30:08Z"
+`
+
+func TestDagTargetTaskOnExit(t *testing.T) {
+	cancel, controller := newController()
+	defer cancel()
+	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
+
+	wf := unmarshalWF(testDagTargetTaskOnExit)
+	wf, err := wfcset.Create(wf)
+	assert.NoError(t, err)
+	woc := newWorkflowOperationCtx(wf, controller)
+
+	woc.operate()
+	onExitNode := woc.wf.GetNodeByName("A.onExit")
+	if assert.NotNil(t, onExitNode) {
+		assert.Equal(t, wfv1.NodePending, onExitNode.Phase)
+	}
+}