Repository: argoproj/argo (go)

Issue:
Fix `Unable to delete PDB resource for workflow` error
## Summary 

Unexpected error.

## Diagnostics

```
evel=error msg="Unable to delete PDB resource for workflow." err="poddisruptionbudgets.policy \"canary-8rpjz\" not found" namespace=argo workflow=canary-8rpjz
```

---
<!-- 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:
fix(controller): Tolerate PDB delete race. Fixes #3706
Checklist:

* [x] Either (a) I've created an [enhancement proposal](https://github.com/argoproj/argo/issues/new/choose) and discussed it with the community, (b) this is a bug fix, or (c) this is a chore.
* [x] The title of the PR is (a) [conventional](https://www.conventionalcommits.org/en/v1.0.0/), (b) states what changed, and (c) suffixes the related issues number. E.g. `"fix(controller): Updates such and such. Fixes #1234"`.  
* [x] I've signed the CLA.
* [x] I have written unit and/or e2e tests for my change. PRs without these are unlikely to be merged.
* [x] My builds are green. Try syncing with master if they are not. 
* [x] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).

Fixes #3706

Patch:
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
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index bcb5ef110..83a2d7d94 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -2819,10 +2819,9 @@ func (woc *wfOperationCtx) deletePDBResource() error {
 	if woc.wfSpec.PodDisruptionBudget == nil {
 		return nil
 	}
-	var err error
-	_ = wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
-		err = woc.controller.kubeclientset.PolicyV1beta1().PodDisruptionBudgets(woc.wf.Namespace).Delete(woc.wf.Name, &metav1.DeleteOptions{})
-		if err != nil {
+	err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
+		err := woc.controller.kubeclientset.PolicyV1beta1().PodDisruptionBudgets(woc.wf.Namespace).Delete(woc.wf.Name, &metav1.DeleteOptions{})
+		if err != nil && !apierr.IsNotFound(err) {
 			woc.log.WithField("err", err).Warn("Failed to delete PDB.")
 			if !retry.IsRetryableKubeAPIError(err) {
 				return false, err


Test patch:
diff --git a/hack/test_report.go b/hack/test_report.go
index 0b9c85c8f..9ed240076 100755
--- a/hack/test_report.go
+++ b/hack/test_report.go
@@ -4,6 +4,7 @@ import (
 	"encoding/xml"
 	"fmt"
 	"io/ioutil"
+	"strconv"
 	"strings"
 )
 
@@ -38,14 +39,40 @@ func testReport() {
 	for _, s := range v.TestSuites {
 		for _, c := range s.TestCases {
 			if c.Failure.Text != "" {
+				x := newFailureText(s.Name, c.Failure.Text)
 				// https://docs.github.com/en/actions/reference/workflow-commands-for-github-actions#setting-an-error-message
 				// Replace ‘/n’ with ‘%0A’ for multiple strings output.
-				parts := strings.SplitN(c.Failure.Text, ":", 3)
-				file := strings.ReplaceAll(s.Name, "github.com/argoproj/argo/", "") + "/" + parts[0]
-				line := parts[1]
-				message := strings.ReplaceAll(strings.TrimSpace(parts[2]), "\n", "%0A")
-				_, _ = fmt.Printf("::error file=%s,line=%v,col=0::%s\n", file, line, message)
+				_, _ = fmt.Printf("::error file=%s,line=%v,col=0::%s\n", x.file, x.line, x.message)
 			}
 		}
 	}
 }
+
+type failureText struct {
+	file    string
+	line    int
+	message string
+}
+
+func trimStdoutLines(text string) string {
+	split := strings.Split(text, "\n")
+	for i, s := range split {
+		if strings.Contains(s, "_test.go") {
+			return strings.Join(split[i:], "\n")
+		}
+	}
+	panic(text)
+}
+
+func newFailureText(suite, text string) failureText {
+	text = trimStdoutLines(text)
+	parts := strings.SplitN(text, ":", 3)
+	file := strings.TrimPrefix(suite, "github.com/argoproj/argo/") + "/" + parts[0]
+	line, _ := strconv.Atoi(parts[1])
+	message := strings.ReplaceAll(strings.TrimSpace(parts[2]), "\n", "%0A")
+	return failureText{
+		file:    file,
+		line:    line,
+		message: message,
+	}
+}
diff --git a/hack/test_report_test.go b/hack/test_report_test.go
new file mode 100644
index 000000000..15577aea6
--- /dev/null
+++ b/hack/test_report_test.go
@@ -0,0 +1,40 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_newFailureText(t *testing.T) {
+	x := newFailureText("github.com/argoproj/argo/controller", `time="2020-08-10T11:04:52-07:00" level=info msg="Processing workflow" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:52-07:00" level=info msg="Updated phase  -> Running" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:52-07:00" level=info msg="Created PDB resource for workflow." namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:52-07:00" level=info msg="Pod node artifact-repo-config-ref initialized Pending" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:52-07:00" level=info msg="Created pod: artifact-repo-config-ref (artifact-repo-config-ref)" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:52-07:00" level=info msg="Workflow update successful" namespace= phase=Running resourceVersion= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Updated phase Running -> Succeeded" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Marking workflow completed" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Deleted PDB resource for workflow." namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Processing workflow" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=error msg="Unexpected pod phase for artifact-repo-config-ref: "
+time="2020-08-10T11:04:53-07:00" level=info msg="Updating node artifact-repo-config-ref status Pending -> Error"
+time="2020-08-10T11:04:53-07:00" level=info msg="Updating node artifact-repo-config-ref message: Unexpected pod phase for artifact-repo-config-ref: "
+time="2020-08-10T11:04:53-07:00" level=info msg="Updated phase Succeeded -> Error" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Updated message  -> Unexpected pod phase for artifact-repo-config-ref: " namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Marking workflow completed" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=warning msg="Failed to delete PDB." err="poddisruptionbudgets.policy \"artifact-repo-config-ref\" not found" namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Deleted PDB resource for workflow." namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Checking daemoned children of " namespace= workflow=artifact-repo-config-ref
+time="2020-08-10T11:04:53-07:00" level=info msg="Workflow update successful" namespace= phase=Error resourceVersion= workflow=artifact-repo-config-ref
+operator_test.go:2912: 
+Error Trace:	operator_test.go:2912
+Error:      	Received unexpected error:
+            	poddisruptionbudgets.policy "artifact-repo-config-ref" not found
+Test:       	TestPDBCreation`)
+	assert.Equal(t, failureText{
+		file: "controller/operator_test.go",
+		line: 2912,
+		message: `Error Trace:	operator_test.go:2912%0AError:      	Received unexpected error:%0A            	poddisruptionbudgets.policy "artifact-repo-config-ref" not found%0ATest:       	TestPDBCreation`,
+	}, x)
+}
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)
+	}
+}
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index 5d915b524..9d47028b6 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -2885,47 +2885,54 @@ var pdbwf = `
 apiVersion: argoproj.io/v1alpha1
 kind: Workflow
 metadata:
-  name: artifact-repo-config-ref
+  name: my-pdb-wf
 spec:
-  entrypoint: whalesay
+  entrypoint: main
   poddisruptionbudget:
     minavailable: 100%
   templates:
-  - name: whalesay
+  - name: main
     container:
       image: docker/whalesay:latest
-      command: [sh, -c]
-      args: ["cowsay hello world | tee /tmp/hello_world.txt"]
 `
 
 func TestPDBCreation(t *testing.T) {
-	cancel, controller := newController()
-	defer cancel()
-	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
 	wf := unmarshalWF(pdbwf)
-	wf, err := wfcset.Create(wf)
-	assert.NoError(t, err)
+	cancel, controller := newController(wf)
+	defer cancel()
 	woc := newWorkflowOperationCtx(wf, controller)
 	woc.operate()
 	pdb, _ := controller.kubeclientset.PolicyV1beta1().PodDisruptionBudgets("").Get(woc.wf.Name, metav1.GetOptions{})
-	assert.NotNil(t, pdb)
 	assert.Equal(t, pdb.Name, wf.Name)
 	woc.markWorkflowSuccess()
+	_, err := controller.kubeclientset.PolicyV1beta1().PodDisruptionBudgets("").Get(woc.wf.Name, metav1.GetOptions{})
+	assert.EqualError(t, err, "poddisruptionbudgets.policy \"my-pdb-wf\" not found")
+}
+
+func TestPDBCreationRaceDelete(t *testing.T) {
+	wf := unmarshalWF(pdbwf)
+	cancel, controller := newController(wf)
+	defer cancel()
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	pod, err := controller.kubeclientset.CoreV1().Pods("").Get("my-pdb-wf", metav1.GetOptions{})
+	assert.NoError(t, err)
+	pod.Status.Phase = apiv1.PodSucceeded
+	_, err = controller.kubeclientset.CoreV1().Pods("").Update(pod)
+	assert.NoError(t, err)
+	err = controller.kubeclientset.PolicyV1beta1().PodDisruptionBudgets("").Delete(woc.wf.Name, nil)
+	assert.NoError(t, err)
 	woc.operate()
-	pdb, _ = controller.kubeclientset.PolicyV1beta1().PodDisruptionBudgets("").Get(woc.wf.Name, metav1.GetOptions{})
-	assert.Nil(t, pdb)
+	assert.Equal(t, wfv1.NodeSucceeded, woc.wf.Status.Phase)
 }
 
 func TestStatusConditions(t *testing.T) {
-	cancel, controller := newController()
-	defer cancel()
-	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
 	wf := unmarshalWF(pdbwf)
-	wf, err := wfcset.Create(wf)
-	assert.NoError(t, err)
+	cancel, controller := newController(wf)
+	defer cancel()
 	woc := newWorkflowOperationCtx(wf, controller)
 	woc.operate()
-	assert.Equal(t, len(woc.wf.Status.Conditions), 0)
+	assert.Empty(t, woc.wf.Status.Conditions)
 	woc.markWorkflowSuccess()
 	assert.Equal(t, woc.wf.Status.Conditions[0].Status, metav1.ConditionStatus("True"))
 }