Repository: argoproj/argo
Pull request #4728 (2020-12-12 00:56:37)

refactor(controller): Enhanced pod clean-up scalability
Signed-off-by: Alex Collins <alex_collins@intuit.com>

Checklist:

* [x] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).

<!--

Please leave your PR in draft if you don't need a review yet. 

To fix failing `CI / Codegen`, run `make codegen`. 

-->


alexec: ```
    --- FAIL: TestPodCleanupSuite/TestOnPodSuccess (9.44s)
        pod_cleanup_test.go:125: 
            	Error Trace:	pod_cleanup_test.go:125
            	            				then.go:86
            	            				then.go:54
            	            				then.go:73
            	            				pod_cleanup_test.go:123
            	Error:      	Expected value not to be nil.
            	Test:       	TestPodCleanupSuite/TestOnPodSuccess
            	Messages:   	failed pod is NOT deleted
```
alexec: ```
=== FAIL: test/e2e TestCLIWithServerOverHTTP1Suite/TestSynchronizationWfLevelMutex (1.24s)
mode: HTTP1
../../dist/argo -n argo submit functional/synchronization-mutex-wf-level-1.yaml
Name:                synchronization-wf-level-mutex-1
Namespace:           argo
ServiceAccount:      default
Status:              Pending
Created:             Wed Dec 16 17:03:13 +0000 (now)
Progress:            

Submitting workflow synchronization-wf-level-mutex 
Waiting for 1s
Done waiting
../../dist/argo -n argo get synchronization-wf-level-mutex
Name:                synchronization-wf-level-mutex
Namespace:           argo
ServiceAccount:      default
Status:              Running
Created:             Wed Dec 16 17:03:13 +0000 (1 second ago)
Started:             Wed Dec 16 17:03:13 +0000 (1 second ago)
Duration:            1 second
Progress:            0/1

STEP                               TEMPLATE  PODNAME                         DURATION  MESSAGE
 ◷ synchronization-wf-level-mutex  whalesay  synchronization-wf-level-mutex  1s          

    --- FAIL: TestCLIWithServerOverHTTP1Suite/TestSynchronizationWfLevelMutex (1.24s)
        cli_test.go:1255: 
            	Error Trace:	cli_test.go:1255
            	            				when.go:288
            	            				when.go:297
            	            				cli_test.go:1254
            	Error:      	"Name:                synchronization-wf-level-mutex\nNamespace:           argo\nServiceAccount:      default\nStatus:              Running\nCreated:             Wed Dec 16 17:03:13 +0000 (1 second ago)\nStarted:             Wed Dec 16 17:03:13 +0000 (1 second ago)\nDuration:            1 second\nProgress:            0/1\n\n\x1b[39mSTEP\x1b[0m                               TEMPLATE  PODNAME                         DURATION  MESSAGE\n \x1b[33m◷\x1b[0m synchronization-wf-level-mutex  whalesay  synchronization-wf-level-mutex  1s          \n" does not contain "Pending"
            	Test:       	TestCLIWithServerOverHTTP1Suite/TestSynchronizationWfLevelMutex
```

Patch:
diff --git a/cmd/workflow-controller/main.go b/cmd/workflow-controller/main.go
index 75d807b36..e184d45ca 100644
--- a/cmd/workflow-controller/main.go
+++ b/cmd/workflow-controller/main.go
@@ -41,6 +41,7 @@ func NewRootCommand() *cobra.Command {
 		workflowWorkers          int    // --workflow-workers
 		workflowTTLWorkers       int    // --workflow-ttl-workers
 		podWorkers               int    // --pod-workers
+		podCleanupWorkers        int    // --pod-cleanup-workers
 		burst                    int
 		qps                      float32
 		namespaced               bool   // --namespaced
@@ -85,7 +86,7 @@ func NewRootCommand() *cobra.Command {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
-			go wfController.Run(ctx, workflowWorkers, workflowTTLWorkers, podWorkers)
+			go wfController.Run(ctx, workflowWorkers, workflowTTLWorkers, podWorkers, podCleanupWorkers)
 
 			// Wait forever
 			select {}
@@ -103,6 +104,7 @@ func NewRootCommand() *cobra.Command {
 	command.Flags().IntVar(&workflowWorkers, "workflow-workers", 32, "Number of workflow workers")
 	command.Flags().IntVar(&workflowTTLWorkers, "workflow-ttl-workers", 4, "Number of workflow TTL workers")
 	command.Flags().IntVar(&podWorkers, "pod-workers", 32, "Number of pod workers")
+	command.Flags().IntVar(&podCleanupWorkers, "pod-cleanup-workers", 4, "Number of pod cleanup workers")
 	command.Flags().IntVar(&burst, "burst", 30, "Maximum burst for throttle.")
 	command.Flags().Float32Var(&qps, "qps", 20.0, "Queries per second")
 	command.Flags().BoolVar(&namespaced, "namespaced", false, "run workflow-controller as namespaced mode")
diff --git a/pkg/apis/workflow/v1alpha1/workflow_types.go b/pkg/apis/workflow/v1alpha1/workflow_types.go
index fb8526132..71a02d961 100644
--- a/pkg/apis/workflow/v1alpha1/workflow_types.go
+++ b/pkg/apis/workflow/v1alpha1/workflow_types.go
@@ -1176,21 +1176,32 @@ var _ ArgumentsProvider = &Arguments{}
 type Nodes map[string]NodeStatus
 
 func (n Nodes) FindByDisplayName(name string) *NodeStatus {
+	return n.Find(NodeWithDisplayName(name))
+}
+
+func (in Nodes) Any(f func(NodeStatus) bool) bool {
+	return in.Find(f) != nil
+}
+
+func (n Nodes) Find(f func(NodeStatus) bool) *NodeStatus {
 	for _, i := range n {
-		if i.DisplayName == name {
+		if f(i) {
 			return &i
 		}
 	}
 	return nil
 }
 
-func (in Nodes) Any(f func(node NodeStatus) bool) bool {
-	for _, i := range in {
-		if f(i) {
-			return true
-		}
-	}
-	return false
+func NodeWithDisplayName(name string) func(n NodeStatus) bool {
+	return func(n NodeStatus) bool { return n.DisplayName == name }
+}
+
+func FailedPodNode(n NodeStatus) bool {
+	return n.Type == NodeTypePod && n.Phase == NodeFailed
+}
+
+func SucceededPodNode(n NodeStatus) bool {
+	return n.Type == NodeTypePod && n.Phase == NodeSucceeded
 }
 
 // UserContainer is a container specified by a user.
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 5d64eb36d..4130839bb 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"os"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/argoproj/pkg/errors"
@@ -90,8 +89,7 @@ type WorkflowController struct {
 	podInformer           cache.SharedIndexInformer
 	wfQueue               workqueue.RateLimitingInterface
 	podQueue              workqueue.RateLimitingInterface
-	completedPods         chan string
-	gcPods                chan string // pods to be deleted depend on GC strategy
+	podCleanupQueue       workqueue.RateLimitingInterface // pods to be deleted or labelled depend on GC strategy
 	throttler             sync.Throttler
 	workflowKeyLock       syncpkg.KeyLock // used to lock workflows for exclusive modification or access
 	session               sqlbuilder.Database
@@ -132,8 +130,6 @@ func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Int
 		cliExecutorImagePullPolicy: executorImagePullPolicy,
 		containerRuntimeExecutor:   containerRuntimeExecutor,
 		configController:           config.NewController(namespace, configMap, kubeclientset, config.EmptyConfigFunc),
-		completedPods:              make(chan string, 512),
-		gcPods:                     make(chan string, 512),
 		workflowKeyLock:            syncpkg.NewKeyLock(),
 		cacheFactory:               controllercache.NewCacheFactory(kubeclientset, namespace),
 		eventRecorderManager:       events.NewEventRecorderManager(kubeclientset),
@@ -143,10 +139,11 @@ func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Int
 
 	wfc.metrics = metrics.New(wfc.getMetricsServerConfig())
 
-	workqueue.SetProvider(wfc.metrics)
+	workqueue.SetProvider(wfc.metrics) // must execute SetProvider before we created the queues
 	wfc.wfQueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "workflow_queue")
 	wfc.throttler = wfc.newThrottler()
 	wfc.podQueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pod_queue")
+	wfc.podCleanupQueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pod_cleanup_queue")
 
 	return &wfc, nil
 }
@@ -178,12 +175,13 @@ var indexers = cache.Indexers{
 }
 
 // Run starts an Workflow resource controller
-func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWorkers, podWorkers int) {
+func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWorkers, podWorkers, podCleanupWorkers int) {
 	defer wfc.wfQueue.ShutDown()
 	defer wfc.podQueue.ShutDown()
+	defer wfc.podCleanupQueue.ShutDown()
 
 	log.WithField("version", argo.GetVersion().Version).Info("Starting Workflow Controller")
-	log.Infof("Workers: workflow: %d, pod: %d", wfWorkers, podWorkers)
+	log.Infof("Workers: workflow: %d, pod: %d, pod cleanup: %d", wfWorkers, podWorkers, podCleanupWorkers)
 
 	wfc.wfInformer = util.NewWorkflowInformer(wfc.dynamicInterface, wfc.GetManagedNamespace(), workflowResyncPeriod, wfc.tweakListOptions, indexers)
 	wfc.wftmplInformer = informer.NewTolerantWorkflowTemplateInformer(wfc.dynamicInterface, workflowTemplateResyncPeriod, wfc.managedNamespace)
@@ -232,8 +230,9 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
 				logCtx.Info("started leading")
 				ctx, cancel = context.WithCancel(ctx)
 
-				go wfc.podLabeler(ctx.Done())
-				go wfc.podGarbageCollector(ctx.Done())
+				for i := 0; i < podCleanupWorkers; i++ {
+					go wait.Until(wfc.runPodCleanup, time.Second, ctx.Done())
+				}
 				go wfc.workflowGarbageCollector(ctx.Done())
 				go wfc.archivedWorkflowGarbageCollector(ctx.Done())
 
@@ -392,54 +391,49 @@ func (wfc *WorkflowController) UpdateConfig() {
 	}
 }
 
-// podLabeler will label all pods on the controllers completedPod channel as completed
-func (wfc *WorkflowController) podLabeler(stopCh <-chan struct{}) {
-	for {
-		select {
-		case <-stopCh:
-			return
-		case pod := <-wfc.completedPods:
-			parts := strings.Split(pod, "/")
-			if len(parts) != 2 {
-				log.WithFields(log.Fields{"pod": pod}).Warn("Unexpected item on completed pod channel")
-				continue
-			}
-			namespace := parts[0]
-			podName := parts[1]
-			err := common.AddPodLabel(wfc.kubeclientset, podName, namespace, common.LabelKeyCompleted, "true")
-			if err != nil {
-				if !apierr.IsNotFound(err) {
-					log.WithFields(log.Fields{"namespace": namespace, "pod": podName, "err": err}).Error("Failed to labeled pod completed")
-				}
-			} else {
-				log.WithFields(log.Fields{"namespace": namespace, "pod": podName}).Info("Labeled pod completed")
-			}
-		}
+func (wfc *WorkflowController) queuePodForCleanup(namespace string, podName string, action podCleanupAction) {
+	wfc.podCleanupQueue.AddRateLimited(newPodCleanupKey(namespace, podName, action))
+}
+
+func (wfc *WorkflowController) runPodCleanup() {
+	for wfc.processNextPodCleanupItem() {
 	}
 }
 
-// podGarbageCollector will delete all pods on the controllers gcPods channel as completed
-func (wfc *WorkflowController) podGarbageCollector(stopCh <-chan struct{}) {
-	for {
-		select {
-		case <-stopCh:
-			return
-		case pod := <-wfc.gcPods:
-			parts := strings.Split(pod, "/")
-			if len(parts) != 2 {
-				log.WithFields(log.Fields{"pod": pod}).Warn("Unexpected item on gcPods channel")
-				continue
-			}
-			namespace := parts[0]
-			podName := parts[1]
-			err := common.DeletePod(wfc.kubeclientset, podName, namespace)
+// all pods will ultimately be cleaned up by either deleting them, or labelling them
+func (wfc *WorkflowController) processNextPodCleanupItem() bool {
+	key, quit := wfc.podCleanupQueue.Get()
+	if quit {
+		return false
+	}
+	defer wfc.podCleanupQueue.Done(key)
+	namespace, podName, action := parsePodCleanupKey(key.(podCleanupKey))
+	logCtx := log.WithFields(log.Fields{"key": key, "action": action})
+	logCtx.Info("cleaning up pod")
+	err := func() error {
+		pods := wfc.kubeclientset.CoreV1().Pods(namespace)
+		switch action {
+		case labelPodCompleted:
+			_, err := pods.Patch(podName, types.MergePatchType, []byte(`{"metadata": {"labels": {"workflows.argoproj.io/completed": "true"}}}`))
 			if err != nil {
-				log.WithFields(log.Fields{"namespace": namespace, "pod": podName, "err": err}).Error("Failed to delete pod for gc")
-			} else {
-				log.WithFields(log.Fields{"namespace": namespace, "pod": podName}).Info("Delete pod for gc successfully")
+				return err
+			}
+		case deletePod:
+			propagation := metav1.DeletePropagationBackground
+			err := pods.Delete(podName, &metav1.DeleteOptions{PropagationPolicy: &propagation})
+			if err != nil && !apierr.IsNotFound(err) {
+				return err
 			}
 		}
+		return nil
+	}()
+	if err != nil {
+		logCtx.WithError(err).Warn("failed to clean-up pod")
+		if errorsutil.IsTransientErr(err) {
+			wfc.podCleanupQueue.AddRateLimited(key)
+		}
 	}
+	return true
 }
 
 func (wfc *WorkflowController) workflowGarbageCollector(stopCh <-chan struct{}) {
@@ -634,8 +628,7 @@ func (wfc *WorkflowController) processNextItem() bool {
 		}
 		if doPodGC {
 			for podName := range woc.completedPods {
-				pod := fmt.Sprintf("%s/%s", woc.wf.ObjectMeta.Namespace, podName)
-				woc.controller.gcPods <- pod
+				woc.controller.queuePodForCleanup(woc.wf.Namespace, podName, deletePod)
 			}
 		}
 	}
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 27a3ef202..a8732fa30 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -574,17 +574,17 @@ func (woc *wfOperationCtx) persistUpdates() {
 		switch woc.execWf.Spec.PodGC.Strategy {
 		case wfv1.PodGCOnPodSuccess:
 			for podName := range woc.succeededPods {
-				woc.controller.gcPods <- fmt.Sprintf("%s/%s", woc.wf.ObjectMeta.Namespace, podName)
+				woc.controller.queuePodForCleanup(woc.wf.Namespace, podName, deletePod)
 			}
 		case wfv1.PodGCOnPodCompletion:
 			for podName := range woc.completedPods {
-				woc.controller.gcPods <- fmt.Sprintf("%s/%s", woc.wf.ObjectMeta.Namespace, podName)
+				woc.controller.queuePodForCleanup(woc.wf.Namespace, podName, deletePod)
 			}
 		}
 	} else {
 		// label pods which will not be deleted
 		for podName := range woc.completedPods {
-			woc.controller.completedPods <- fmt.Sprintf("%s/%s", woc.wf.ObjectMeta.Namespace, podName)
+			woc.controller.queuePodForCleanup(woc.wf.Namespace, podName, labelPodCompleted)
 		}
 	}
 }
diff --git a/workflow/controller/pod_cleanup_key.go b/workflow/controller/pod_cleanup_key.go
new file mode 100644
index 000000000..6728aa636
--- /dev/null
+++ b/workflow/controller/pod_cleanup_key.go
@@ -0,0 +1,31 @@
+package controller
+
+import (
+	"fmt"
+	"strings"
+)
+
+// Should I use "clean-up" or "cleanup"?
+// * cleanup is a noun - e.g "The cleanup"
+// * clean-up is a verb - e.g. "I clean-up"
+
+type podCleanupKey = string // describes the pod to cleanup + the cleanup action to take
+type podCleanupAction = string
+
+const (
+	deletePod         podCleanupAction = "deletePod"
+	labelPodCompleted podCleanupAction = "labelPodCompleted"
+)
+
+func newPodCleanupKey(namespace string, podName string, action podCleanupAction) podCleanupKey {
+	return fmt.Sprintf("%s/%s/%v", namespace, podName, action)
+}
+
+func parsePodCleanupKey(k podCleanupKey) (namespace string, podName string, action podCleanupAction) {
+	parts := strings.Split(k, "/")
+	if len(parts) != 3 {
+		return "", "", ""
+	}
+	return parts[0], parts[1], parts[2]
+
+}


Test patch:
diff --git a/test/e2e/fixtures/then.go b/test/e2e/fixtures/then.go
index c1a7251af..971c4faa8 100644
--- a/test/e2e/fixtures/then.go
+++ b/test/e2e/fixtures/then.go
@@ -66,6 +66,32 @@ func (t *Then) ExpectWorkflowDeleted() *Then {
 	return t
 }
 
+// Check on a specific node in the workflow.
+// If no node matches the selector, then the NodeStatus and Pod will be nil.
+// If the pod does not exist (e.g. because it was deleted) then the Pod will be nil too.
+func (t *Then) ExpectWorkflowNode(selector func(status wfv1.NodeStatus) bool, f func(t *testing.T, status *wfv1.NodeStatus, pod *apiv1.Pod)) *Then {
+	return t.expectWorkflow(t.wf.Name, func(tt *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+		n := status.Nodes.Find(selector)
+		var p *apiv1.Pod
+		if n != nil {
+			println("Found node", "id="+n.ID, "type="+n.Type)
+			if n.Type == wfv1.NodeTypePod {
+				var err error
+				p, err = t.kubeClient.CoreV1().Pods(t.wf.Namespace).Get(n.ID, metav1.GetOptions{})
+				if err != nil {
+					if !apierr.IsNotFound(err) {
+						t.t.Fatal(err)
+					}
+					p = nil // i did not expect to need to nil the pod, but here we are
+				}
+			}
+		} else {
+			println("Did not find node")
+		}
+		f(tt, n, p)
+	})
+}
+
 func (t *Then) ExpectCron(block func(t *testing.T, cronWf *wfv1.CronWorkflow)) *Then {
 	t.t.Helper()
 	if t.cronWf == nil {
diff --git a/test/e2e/pod_cleanup_test.go b/test/e2e/pod_cleanup_test.go
new file mode 100644
index 000000000..1a7d6a8ef
--- /dev/null
+++ b/test/e2e/pod_cleanup_test.go
@@ -0,0 +1,195 @@
+// +build e2e
+
+package e2e
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
+	corev1 "k8s.io/api/core/v1"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	"github.com/argoproj/argo/test/e2e/fixtures"
+	"github.com/argoproj/argo/workflow/common"
+)
+
+type PodCleanupSuite struct {
+	fixtures.E2ESuite
+}
+
+const enoughTimeForPodCleanup = 5 * time.Second
+
+func (s *PodCleanupSuite) TestNone() {
+	s.Given().
+		Workflow(`
+metadata:
+  generateName: test-pod-cleanup-
+  labels:
+    argo-e2e: true
+spec:
+  entrypoint: main
+  templates:
+    - name: main
+      container:
+        image: argoproj/argosay:v2
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow().
+		Wait(enoughTimeForPodCleanup).
+		Then().
+		ExpectWorkflowNode(wfv1.SucceededPodNode, func(t *testing.T, n *wfv1.NodeStatus, p *corev1.Pod) {
+			if assert.NotNil(t, n) && assert.NotNil(t, p) {
+				assert.Equal(t, "true", p.Labels[common.LabelKeyCompleted])
+			}
+		})
+}
+
+func (s *PodCleanupSuite) TestOnPodCompletion() {
+	s.Given().
+		Workflow(`
+metadata:
+  generateName: test-pod-cleanup-on-pod-success-
+  labels:
+    argo-e2e: true
+spec:
+  podGC: 
+    strategy: OnPodCompletion
+  entrypoint: main
+  templates:
+    - name: main
+      steps:
+        - - name: success
+            template: success
+          - name: failure
+            template: failure
+    - name: success
+      container:
+        image: argoproj/argosay:v2
+    - name: failure
+      container:
+        image: argoproj/argosay:v2
+        args: [exit, 1]
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow().
+		Wait(enoughTimeForPodCleanup).
+		Then().
+		ExpectWorkflowNode(wfv1.FailedPodNode, func(t *testing.T, n *wfv1.NodeStatus, p *corev1.Pod) {
+			if assert.NotNil(t, n) {
+				assert.Nil(t, p, "failed pod is deleted")
+			}
+		}).
+		ExpectWorkflowNode(wfv1.SucceededPodNode, func(t *testing.T, n *wfv1.NodeStatus, p *corev1.Pod) {
+			if assert.NotNil(t, n) {
+				assert.Nil(t, p, "successful pod is deleted")
+			}
+		})
+}
+
+func (s *PodCleanupSuite) TestOnPodSuccess() {
+	s.Given().
+		Workflow(`
+metadata:
+  generateName: test-pod-cleanup-on-pod-success-
+  labels:
+    argo-e2e: true
+spec:
+  podGC: 
+    strategy: OnPodSuccess
+  entrypoint: main
+  templates:
+    - name: main
+      steps:
+        - - name: success
+            template: success
+          - name: failure
+            template: failure
+    - name: success
+      container:
+        image: argoproj/argosay:v2
+    - name: failure
+      container:
+        image: argoproj/argosay:v2
+        args: [exit, 1]
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow().
+		Wait(enoughTimeForPodCleanup).
+		Then().
+		ExpectWorkflowNode(wfv1.FailedPodNode, func(t *testing.T, n *wfv1.NodeStatus, p *corev1.Pod) {
+			if assert.NotNil(t, n) {
+				assert.NotNil(t, p, "failed pod is NOT deleted")
+			}
+		}).
+		ExpectWorkflowNode(wfv1.SucceededPodNode, func(t *testing.T, n *wfv1.NodeStatus, p *corev1.Pod) {
+			if assert.NotNil(t, n) {
+				assert.Nil(t, p, "successful pod is deleted")
+			}
+		})
+}
+
+func (s *PodCleanupSuite) TestOnWorkflowCompletion() {
+	s.Given().
+		Workflow(`
+metadata:
+  generateName: test-pod-cleanup-on-workflow-completion-
+  labels:
+    argo-e2e: true
+spec:
+  podGC: 
+    strategy: OnWorkflowCompletion
+  entrypoint: main
+  templates:
+    - name: main
+      container:
+        image: argoproj/argosay:v2
+        args: [exit, 1]
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow().
+		Wait(enoughTimeForPodCleanup).
+		Then().
+		ExpectWorkflowNode(wfv1.FailedPodNode, func(t *testing.T, n *wfv1.NodeStatus, p *corev1.Pod) {
+			if assert.NotNil(t, n) {
+				assert.Nil(t, p, "failed pod is deleted")
+			}
+		})
+}
+
+func (s *PodCleanupSuite) TestOnWorkflowSuccess() {
+	s.Given().
+		Workflow(`
+metadata:
+  generateName: test-pod-cleanup-on-workflow-success-
+  labels:
+    argo-e2e: true
+spec:
+  podGC: 
+    strategy: OnWorkflowSuccess
+  entrypoint: main
+  templates:
+    - name: main
+      container:
+        image: argoproj/argosay:v2
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow().
+		Wait(enoughTimeForPodCleanup).
+		Then().
+		ExpectWorkflowNode(wfv1.SucceededPodNode, func(t *testing.T, n *wfv1.NodeStatus, p *corev1.Pod) {
+			if assert.NotNil(t, n) {
+				assert.Nil(t, p, "successful pod is deleted")
+			}
+		})
+}
+
+func TestPodCleanupSuite(t *testing.T) {
+	suite.Run(t, new(PodCleanupSuite))
+}
diff --git a/test/e2e/testdata/sleepy-retry-on-error-workflow.yaml b/test/e2e/testdata/sleepy-retry-on-error-workflow.yaml
index 326bcf6d8..892ddd4a5 100644
--- a/test/e2e/testdata/sleepy-retry-on-error-workflow.yaml
+++ b/test/e2e/testdata/sleepy-retry-on-error-workflow.yaml
@@ -1,7 +1,7 @@
 kind: Workflow
 apiVersion: argoproj.io/v1alpha1
 metadata:
-  name: sleepy-retry-on-error
+  generateName: sleepy-retry-on-error-
   labels:
     argo-e2e: "true"
 spec:
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index 2176c6079..5a0219055 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -156,7 +156,6 @@ func newController(options ...interface{}) (context.CancelFunc, *WorkflowControl
 		kubeclientset:        kube,
 		dynamicInterface:     dynamicClient,
 		wfclientset:          wfclientset,
-		completedPods:        make(chan string, 16),
 		workflowKeyLock:      sync.NewKeyLock(),
 		wfArchive:            sqldb.NullWorkflowArchive,
 		hydrator:             hydratorfake.Noop,
@@ -180,6 +179,7 @@ func newController(options ...interface{}) (context.CancelFunc, *WorkflowControl
 		wfc.wfQueue = workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
 		wfc.throttler = wfc.newThrottler()
 		wfc.podQueue = workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+		wfc.podCleanupQueue = workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
 	}
 
 	// always compare to WorkflowController.Run to see what this block of code should be doing