Repository: argoproj/argo (go)

Issue:
When not specifying port for Postgres DB port is defaulting to 0
## Summary 

It looks like since the changes in https://github.com/argoproj/argo/pull/4307 if you do not specify a port for your PG configuration it will now default to `0` since the port type has changed to an integer and the concatenation logic will append `:0` to the connection string.

---

Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.


Pull request description:
fix(controller): Support default database port. Fixes #4756
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`. 

-->


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/config/config.go b/config/config.go
index 0fee9b66c..e4ab9b33a 100644
--- a/config/config.go
+++ b/config/config.go
@@ -1,6 +1,8 @@
 package config
 
 import (
+	"fmt"
+
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -170,25 +172,31 @@ type ConnectionPool struct {
 	ConnMaxLifetime TTL `json:"connMaxLifetime,omitempty"`
 }
 
-type PostgreSQLConfig struct {
+type DatabaseConfig struct {
 	Host           string                  `json:"host"`
-	Port           int                     `json:"port"`
+	Port           int                     `json:"port,omitempty"`
 	Database       string                  `json:"database"`
 	TableName      string                  `json:"tableName,omitempty"`
 	UsernameSecret apiv1.SecretKeySelector `json:"userNameSecret,omitempty"`
 	PasswordSecret apiv1.SecretKeySelector `json:"passwordSecret,omitempty"`
-	SSL            bool                    `json:"ssl,omitempty"`
-	SSLMode        string                  `json:"sslMode,omitempty"`
+}
+
+func (c DatabaseConfig) GetHostname() string {
+	if c.Port == 0 {
+		return c.Host
+	}
+	return fmt.Sprintf("%s:%v", c.Host, c.Port)
+}
+
+type PostgreSQLConfig struct {
+	DatabaseConfig
+	SSL     bool   `json:"ssl,omitempty"`
+	SSLMode string `json:"sslMode,omitempty"`
 }
 
 type MySQLConfig struct {
-	Host           string                  `json:"host"`
-	Port           int                     `json:"port"`
-	Database       string                  `json:"database"`
-	TableName      string                  `json:"tableName,omitempty"`
-	Options        map[string]string       `json:"options,omitempty"`
-	UsernameSecret apiv1.SecretKeySelector `json:"userNameSecret,omitempty"`
-	PasswordSecret apiv1.SecretKeySelector `json:"passwordSecret,omitempty"`
+	DatabaseConfig
+	Options map[string]string `json:"options,omitempty"`
 }
 
 // S3ArtifactRepository defines the controller configuration for an S3 artifact repository
diff --git a/persist/sqldb/sqldb.go b/persist/sqldb/sqldb.go
index 465192645..d60a8eac9 100644
--- a/persist/sqldb/sqldb.go
+++ b/persist/sqldb/sqldb.go
@@ -50,7 +50,7 @@ func CreatePostGresDBSession(kubectlConfig kubernetes.Interface, namespace strin
 	var settings = postgresql.ConnectionURL{
 		User:     string(userNameByte),
 		Password: string(passwordByte),
-		Host:     fmt.Sprintf("%s:%v", cfg.Host, cfg.Port),
+		Host:     cfg.GetHostname(),
 		Database: cfg.Database,
 	}
 
@@ -95,7 +95,7 @@ func CreateMySQLDBSession(kubectlConfig kubernetes.Interface, namespace string,
 	session, err := mysql.Open(mysql.ConnectionURL{
 		User:     string(userNameByte),
 		Password: string(passwordByte),
-		Host:     fmt.Sprintf("%s:%v", cfg.Host, cfg.Port),
+		Host:     cfg.GetHostname(),
 		Database: cfg.Database,
 	})
 	if err != nil {
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/config/config_test.go b/config/config_test.go
index 88abcf311..3a1cf5702 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -12,3 +12,8 @@ func TestArtifactRepository_IsArchiveLogs(t *testing.T) {
 	assert.False(t, (&ArtifactRepository{ArchiveLogs: pointer.BoolPtr(false)}).IsArchiveLogs())
 	assert.True(t, (&ArtifactRepository{ArchiveLogs: pointer.BoolPtr(true)}).IsArchiveLogs())
 }
+
+func TestDatabaseConfig(t *testing.T) {
+	assert.Equal(t, "my-host", DatabaseConfig{Host: "my-host"}.GetHostname())
+	assert.Equal(t, "my-host:1234", DatabaseConfig{Host: "my-host", Port: 1234}.GetHostname())
+}
diff --git a/test/e2e/argo_server_test.go b/test/e2e/argo_server_test.go
index a0f5b73a4..8c98a6c51 100644
--- a/test/e2e/argo_server_test.go
+++ b/test/e2e/argo_server_test.go
@@ -727,7 +727,8 @@ func (s *ArgoServerSuite) TestWorkflowService() {
         {
           "name": "run-workflow",
           "container": {
-            "image": "argoproj/argosay:v2"
+            "image": "argoproj/argosay:v2",
+            "args": ["sleep", "10s"]   
           }
         }
       ],
diff --git a/test/e2e/cli_test.go b/test/e2e/cli_test.go
index 930e7aec6..750254287 100644
--- a/test/e2e/cli_test.go
+++ b/test/e2e/cli_test.go
@@ -1239,45 +1239,6 @@ func (s *CLISuite) TestRetryOmit() {
 		WaitForWorkflow()
 }
 
-func (s *CLISuite) TestSynchronizationWfLevelMutex() {
-	s.NeedsOffloading()
-	s.Given().
-		Workflow("@functional/synchronization-mutex-wf-level.yaml").
-		When().
-		RunCli([]string{"submit", "functional/synchronization-mutex-wf-level-1.yaml"}, func(t *testing.T, output string, err error) {
-			if assert.NoError(t, err) {
-				assert.Contains(t, output, "synchronization-wf-level-mutex")
-			}
-		}).
-		SubmitWorkflow().
-		Wait(1*time.Second).
-		RunCli([]string{"get", "synchronization-wf-level-mutex"}, func(t *testing.T, output string, err error) {
-			assert.Contains(t, output, "Pending")
-		}).
-		WaitForWorkflow(fixtures.ToBeCompleted, 120*time.Second).
-		Then().
-		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
-			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
-		})
-}
-
-func (s *CLISuite) TestTemplateLevelMutex() {
-	s.NeedsOffloading()
-	s.Given().
-		Workflow("@functional/synchronization-mutex-tmpl-level.yaml").
-		When().
-		SubmitWorkflow().
-		Wait(3*time.Second).
-		RunCli([]string{"get", "synchronization-tmpl-level-mutex"}, func(t *testing.T, output string, err error) {
-			assert.Contains(t, output, "Waiting for")
-		}).
-		WaitForWorkflow().
-		Then().
-		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
-			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
-		})
-}
-
 func (s *CLISuite) TestResourceTemplateStopAndTerminate() {
 	s.NeedsOffloading()
 	s.Run("ResourceTemplateStop", func() {
diff --git a/test/e2e/fixtures/given.go b/test/e2e/fixtures/given.go
index 278b8b782..55aae39f6 100644
--- a/test/e2e/fixtures/given.go
+++ b/test/e2e/fixtures/given.go
@@ -140,11 +140,11 @@ var NoError = func(t *testing.T, output string, err error) {
 	assert.NoError(t, err, output)
 }
 
-var OutputContains = func(contains string) func(t *testing.T, output string, err error) {
+var OutputRegexp = func(rx string) func(t *testing.T, output string, err error) {
 	return func(t *testing.T, output string, err error) {
 		t.Helper()
 		if assert.NoError(t, err, output) {
-			assert.Contains(t, output, contains)
+			assert.Regexp(t, rx, output)
 		}
 	}
 }
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/fixtures/when.go b/test/e2e/fixtures/when.go
index 894f3347f..3d0e726d0 100644
--- a/test/e2e/fixtures/when.go
+++ b/test/e2e/fixtures/when.go
@@ -170,6 +170,10 @@ var ToBeDone Condition = func(wf *wfv1.Workflow) bool {
 
 var ToBeArchived Condition = func(wf *wfv1.Workflow) bool { return wf.Labels[common.LabelKeyWorkflowArchivingStatus] == "Archived" }
 
+var ToBeWaitingOnAMutex Condition = func(wf *wfv1.Workflow) bool {
+	return wf.Status.Synchronization != nil && wf.Status.Synchronization.Mutex != nil
+}
+
 // Wait for a workflow to meet a condition:
 // Options:
 // * `time.Duration` - change the timeout - 30s by default
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index 21092aab0..f46a4be6b 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -42,7 +42,7 @@ func (s *FunctionalSuite) TestDeletingPendingPod() {
 		When().
 		SubmitWorkflow().
 		WaitForWorkflow(fixtures.ToStart, "to start").
-		Exec("kubectl", []string{"-n", "argo", "delete", "pod", "-l", "workflows.argoproj.io/workflow"}, fixtures.OutputContains(`pod "sleepy" deleted`)).
+		Exec("kubectl", []string{"-n", "argo", "delete", "pod", "-l", "workflows.argoproj.io/workflow"}, fixtures.OutputRegexp(`pod "sleepy-.*" deleted`)).
 		WaitForWorkflow().
 		Then().
 		ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
@@ -59,14 +59,14 @@ func (s *FunctionalSuite) TestDeletingRunningPod() {
 		When().
 		SubmitWorkflow().
 		WaitForWorkflow(fixtures.ToBeRunning, "to be running").
-		Exec("kubectl", []string{"-n", "argo", "delete", "pod", "-l", "workflows.argoproj.io/workflow"}, fixtures.NoError).
+		Exec("kubectl", []string{"-n", "argo", "delete", "pod", "-l", "workflows.argoproj.io/workflow"}, fixtures.OutputRegexp(`pod "sleepy-.*" deleted`)).
 		WaitForWorkflow().
 		Then().
 		ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
 			assert.Equal(t, wfv1.NodeError, status.Phase)
 			assert.Len(t, status.Nodes, 1)
-			if assert.Contains(t, status.Nodes, "sleepy") {
-				assert.Equal(t, "pod deleted during operation", status.Nodes["sleepy"].Message)
+			if assert.Contains(t, status.Nodes, metadata.Name) {
+				assert.Equal(t, "pod deleted during operation", status.Nodes[metadata.Name].Message)
 			}
 		})
 }
@@ -88,6 +88,36 @@ func (s *FunctionalSuite) TestDeletingRunningPodWithOrErrorRetryPolicy() {
 		})
 }
 
+func (s *FunctionalSuite) TestSynchronizationWfLevelMutex() {
+	s.Given().
+		Workflow("@functional/synchronization-mutex-wf-level-1.yaml").
+		When().
+		SubmitWorkflow().
+		Given().
+		Workflow("@functional/synchronization-mutex-wf-level.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(fixtures.ToBeWaitingOnAMutex, "to be waiting on a mutex").
+		WaitForWorkflow().
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+		})
+}
+
+func (s *FunctionalSuite) TestTemplateLevelMutex() {
+	s.Given().
+		Workflow("@functional/synchronization-mutex-tmpl-level.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(fixtures.ToBeWaitingOnAMutex, "to be waiting on a mutex").
+		WaitForWorkflow().
+		Then().
+		ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+		})
+}
+
 func (s *FunctionalSuite) TestWorkflowTTL() {
 	s.Given().
 		Workflow(`
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/test/e2e/testdata/sleepy-workflow.yaml b/test/e2e/testdata/sleepy-workflow.yaml
index e67113534..87ce94b5e 100644
--- a/test/e2e/testdata/sleepy-workflow.yaml
+++ b/test/e2e/testdata/sleepy-workflow.yaml
@@ -1,7 +1,7 @@
 kind: Workflow
 apiVersion: argoproj.io/v1alpha1
 metadata:
-  name: sleepy
+  generateName: sleepy-
   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