Repository: argoproj/argo
Pull request #4740 (2020-12-14 21:00:13)

chore(example): Add watch timeout and print out workflow status message
The workflow in the current example would only exit the loop when it reaches completion. It's better to exit loop immediately when the workflow has failed. 

Signed-off-by: terrytangyuan <terrytangyuan@gmail.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/Dockerfile b/Dockerfile
index 708beab64..3be5b5c0e 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -142,7 +142,7 @@ ARG IMAGE_OS=linux
 COPY --from=argoexec-base /etc/ssh/ssh_known_hosts /etc/ssh/ssh_known_hosts
 COPY --from=argoexec-base /etc/nsswitch.conf /etc/nsswitch.conf
 COPY --from=argoexec-base /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
-COPY --from=argo-build /go/src/github.com/argoproj/argo/argo-server.crt argo-server.crt
-COPY --from=argo-build /go/src/github.com/argoproj/argo/argo-server.key argo-server.key
+COPY --from=argo-build --chown=8737 /go/src/github.com/argoproj/argo/argo-server.crt argo-server.crt
+COPY --from=argo-build --chown=8737 /go/src/github.com/argoproj/argo/argo-server.key argo-server.key
 COPY --from=argo-build /go/src/github.com/argoproj/argo/dist/argo-${IMAGE_OS}-* /bin/argo
 ENTRYPOINT [ "argo" ]
diff --git a/Dockerfile.dev b/Dockerfile.dev
index e86d1c4de..7d3cd300b 100644
--- a/Dockerfile.dev
+++ b/Dockerfile.dev
@@ -93,7 +93,7 @@ FROM scratch as argocli
 USER 8737
 COPY --from=argoexec-base /etc/ssh/ssh_known_hosts /etc/ssh/ssh_known_hosts
 COPY --from=argoexec-base /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
-COPY argo-server.crt argo-server.crt
-COPY argo-server.key argo-server.key
+COPY --chown=8737 argo-server.crt argo-server.crt
+COPY --chown=8737 argo-server.key argo-server.key
 COPY argo /bin/
 ENTRYPOINT [ "argo" ]
diff --git a/Makefile b/Makefile
index da9a93081..4141680de 100644
--- a/Makefile
+++ b/Makefile
@@ -68,6 +68,10 @@ LOG_LEVEL             := debug
 UPPERIO_DB_DEBUG      := 0
 NAMESPACED            := true
 
+ifeq ($(PROFILE),prometheus)
+RUN_MODE              := kubernetes
+endif
+
 ALWAYS_OFFLOAD_NODE_STATUS := false
 ifeq ($(PROFILE),mysql)
 ALWAYS_OFFLOAD_NODE_STATUS := true
@@ -430,7 +434,13 @@ ifeq ($(RUN_MODE),kubernetes)
 	kubectl -n $(KUBE_NAMESPACE) scale deploy/workflow-controller --replicas 1
 	kubectl -n $(KUBE_NAMESPACE) scale deploy/argo-server --replicas 1
 endif
-	kubectl -n $(KUBE_NAMESPACE) wait --for=condition=Ready pod --all -l app --timeout 2m
+ifeq ($(RUN_MODE),kubernetes)
+	kubectl -n $(KUBE_NAMESPACE) wait --for=condition=Ready pod -l app=argo-server --timeout 1m
+	kubectl -n $(KUBE_NAMESPACE) wait --for=condition=Ready pod -l app=workflow-controller --timeout 1m
+endif
+ifeq ($(PROFILE),prometheus)
+	kubectl -n $(KUBE_NAMESPACE) wait --for=condition=Ready pod -l app=prometheus --timeout 1m
+endif
 	./hack/port-forward.sh
 	# Check dex, minio, postgres and mysql are in hosts file
 ifeq ($(AUTH_MODE),sso)
diff --git a/cmd/argo/commands/wait.go b/cmd/argo/commands/wait.go
index 04992f127..8b3efee35 100644
--- a/cmd/argo/commands/wait.go
+++ b/cmd/argo/commands/wait.go
@@ -74,7 +74,8 @@ func waitOnOne(serviceClient workflowpkg.WorkflowServiceClient, ctx context.Cont
 	req := &workflowpkg.WatchWorkflowsRequest{
 		Namespace: namespace,
 		ListOptions: &metav1.ListOptions{
-			FieldSelector: util.GenerateFieldSelectorFromWorkflowName(wfName),
+			FieldSelector:   util.GenerateFieldSelectorFromWorkflowName(wfName),
+			ResourceVersion: "0",
 		},
 	}
 	stream, err := serviceClient.WatchWorkflows(ctx, req)
diff --git a/cmd/argo/commands/watch.go b/cmd/argo/commands/watch.go
index 934676dbb..138aec0a4 100644
--- a/cmd/argo/commands/watch.go
+++ b/cmd/argo/commands/watch.go
@@ -58,7 +58,8 @@ func watchWorkflow(ctx context.Context, serviceClient workflowpkg.WorkflowServic
 	req := &workflowpkg.WatchWorkflowsRequest{
 		Namespace: namespace,
 		ListOptions: &metav1.ListOptions{
-			FieldSelector: util.GenerateFieldSelectorFromWorkflowName(workflow),
+			FieldSelector:   util.GenerateFieldSelectorFromWorkflowName(workflow),
+			ResourceVersion: "0",
 		},
 	}
 	stream, err := serviceClient.WatchWorkflows(ctx, req)
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/examples/example-golang/main.go b/examples/example-golang/main.go
index a0e82c7be..2914e8dcb 100644
--- a/examples/example-golang/main.go
+++ b/examples/example-golang/main.go
@@ -11,6 +11,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/utils/pointer"
 
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	wfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned"
@@ -60,7 +61,7 @@ func main() {
 
 	// wait for the workflow to complete
 	fieldSelector := fields.ParseSelectorOrDie(fmt.Sprintf("metadata.name=%s", createdWf.Name))
-	watchIf, err := wfClient.Watch(metav1.ListOptions{FieldSelector: fieldSelector.String()})
+	watchIf, err := wfClient.Watch(metav1.ListOptions{FieldSelector: fieldSelector.String(), TimeoutSeconds: pointer.Int64Ptr(180)})
 	errors.CheckError(err)
 	defer watchIf.Stop()
 	for next := range watchIf.ResultChan() {
@@ -69,7 +70,7 @@ func main() {
 			continue
 		}
 		if !wf.Status.FinishedAt.IsZero() {
-			fmt.Printf("Workflow %s %s at %v\n", wf.Name, wf.Status.Phase, wf.Status.FinishedAt)
+			fmt.Printf("Workflow %s %s at %v. Message: %s.\n", wf.Name, wf.Status.Phase, wf.Status.FinishedAt, wf.Status.Message)
 			break
 		}
 	}
diff --git a/hack/port-forward.sh b/hack/port-forward.sh
index f17b5eeb6..e6d37500e 100755
--- a/hack/port-forward.sh
+++ b/hack/port-forward.sh
@@ -6,11 +6,12 @@ pf() {
   name=$1
   resource=$2
   port=$3
+  dest_port=${4:-"$port"}
   pid=$(lsof -i ":$port" | grep -v PID | awk '{print $2}' || true)
   if [ "$pid" != "" ]; then
     kill $pid
   fi
-  kubectl -n argo port-forward "$resource" "$port:$port" > /dev/null &
+  kubectl -n argo port-forward "$resource" "$port:$dest_port" > /dev/null &
   # wait until port forward is established
 	until lsof -i ":$port" > /dev/null ; do sleep 1s ; done
   info "$name on http://localhost:$port"
@@ -44,3 +45,7 @@ fi
 if [[ "$(kubectl -n argo get pod -l app=workflow-controller -o name)" != "" ]]; then
   pf "Workflow Controller" deploy/workflow-controller 9090
 fi
+
+if [[ "$(kubectl -n argo get pod -l app=prometheus -o name)" != "" ]]; then
+  pf "Prometheus Server" deploy/prometheus 9091 9090
+fi
diff --git a/manifests/quick-start/base/kustomization.yaml b/manifests/quick-start/base/kustomization.yaml
index add9be781..d404572cc 100644
--- a/manifests/quick-start/base/kustomization.yaml
+++ b/manifests/quick-start/base/kustomization.yaml
@@ -13,4 +13,4 @@ resources:
 
 patchesStrategicMerge:
   - overlays/workflow-controller-configmap.yaml
-  - overlays/argo-server-deployment.yaml
\ No newline at end of file
+  - overlays/argo-server-deployment.yaml
diff --git a/manifests/quick-start/base/prometheus/kustomization.yaml b/manifests/quick-start/base/prometheus/kustomization.yaml
new file mode 100644
index 000000000..5d4cf2f4c
--- /dev/null
+++ b/manifests/quick-start/base/prometheus/kustomization.yaml
@@ -0,0 +1,6 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+resources:
+  - prometheus-deployment.yaml
+  - prometheus-config-cluster.yaml
diff --git a/manifests/quick-start/base/prometheus/prometheus-config-cluster.yaml b/manifests/quick-start/base/prometheus/prometheus-config-cluster.yaml
new file mode 100644
index 000000000..267347ab4
--- /dev/null
+++ b/manifests/quick-start/base/prometheus/prometheus-config-cluster.yaml
@@ -0,0 +1,12 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: prometheus-config
+data:
+  prometheus.yaml: |
+    global:
+      scrape_interval: 15s
+    scrape_configs:
+      - job_name: 'argo'
+        static_configs:
+          - targets: ['workflow-controller-metrics:9090']
diff --git a/manifests/quick-start/base/prometheus/prometheus-deployment.yaml b/manifests/quick-start/base/prometheus/prometheus-deployment.yaml
new file mode 100644
index 000000000..886b4e3a1
--- /dev/null
+++ b/manifests/quick-start/base/prometheus/prometheus-deployment.yaml
@@ -0,0 +1,37 @@
+# This manifests creates a minimal Prometheus server to scrape and display the metrics emitted by the workflow
+# controller. To open this server, create an external IP for the prometheus service or use kubectl port-forward,
+# then open:
+#   
+#   localhost:9091/graph
+#
+# Note: this assumes the workflow-controller is emitting metrics in the default port (9090). This will need to
+# be modified if the default is overriden.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: prometheus
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: prometheus
+  template:
+    metadata:
+      labels:
+        app: prometheus
+      name: prometheus
+    spec:
+      containers:
+        - name: prometheus
+          image: prom/prometheus
+          args:
+            - --config.file=/config/prometheus.yaml
+          volumeMounts:
+          - name: config
+            mountPath: /config
+      volumes:
+        - name: config
+          configMap:
+            name: prometheus-config
+
+
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 41186e6f2..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.
@@ -2510,6 +2521,9 @@ func (ss *SemaphoreStatus) LockWaiting(holderKey, lockKey string, currentHolders
 func (ss *SemaphoreStatus) LockAcquired(holderKey, lockKey string, currentHolders []string) bool {
 	i, semaphoreHolding := ss.GetHolding(lockKey)
 	items := strings.Split(holderKey, "/")
+	if len(items) == 0 {
+		return false
+	}
 	holdingName := items[len(items)-1]
 	if i < 0 {
 		ss.Holding = append(ss.Holding, SemaphoreHolding{Semaphore: lockKey, Holders: []string{holdingName}})
@@ -2525,6 +2539,9 @@ func (ss *SemaphoreStatus) LockAcquired(holderKey, lockKey string, currentHolder
 func (ss *SemaphoreStatus) LockReleased(holderKey, lockKey string) bool {
 	i, semaphoreHolding := ss.GetHolding(lockKey)
 	items := strings.Split(holderKey, "/")
+	if len(items) == 0 {
+		return false
+	}
 	holdingName := items[len(items)-1]
 	if i >= 0 {
 		semaphoreHolding.Holders = slice.RemoveString(semaphoreHolding.Holders, holdingName)
@@ -2599,6 +2616,9 @@ func (ms *MutexStatus) LockWaiting(holderKey, lockKey string, currentHolders []s
 func (ms *MutexStatus) LockAcquired(holderKey, lockKey string, currentHolders []string) bool {
 	i, mutexHolding := ms.GetHolding(lockKey)
 	items := strings.Split(holderKey, "/")
+	if len(items) == 0 {
+		return false
+	}
 	holdingName := items[len(items)-1]
 	if i < 0 {
 		ms.Holding = append(ms.Holding, MutexHolding{Mutex: lockKey, Holder: holdingName})
@@ -2612,8 +2632,13 @@ func (ms *MutexStatus) LockAcquired(holderKey, lockKey string, currentHolders []
 }
 
 func (ms *MutexStatus) LockReleased(holderKey, lockKey string) bool {
-	i, _ := ms.GetHolding(lockKey)
-	if i >= 0 {
+	i, holder := ms.GetHolding(lockKey)
+	items := strings.Split(holderKey, "/")
+	if len(items) == 0 {
+		return false
+	}
+	holdingName := items[len(items)-1]
+	if i >= 0 && holder.Holder == holdingName {
 		ms.Holding = append(ms.Holding[:i], ms.Holding[i+1:]...)
 		return true
 	}
diff --git a/server/auth/sso/sso.go b/server/auth/sso/sso.go
index 47c0b7ae8..39db5146d 100644
--- a/server/auth/sso/sso.go
+++ b/server/auth/sso/sso.go
@@ -18,6 +18,7 @@ import (
 	"gopkg.in/square/go-jose.v2"
 	"gopkg.in/square/go-jose.v2/jwt"
 	apiv1 "k8s.io/api/core/v1"
+	apierr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 
@@ -139,10 +140,13 @@ func newSso(
 	// if it fails, then the get will fail, and the pod restart
 	// it may fail due to race condition with another pod - which is fine,
 	// when it restart it'll get the new key
-	_, _ = secretsIf.Create(&apiv1.Secret{
+	_, err = secretsIf.Create(&apiv1.Secret{
 		ObjectMeta: metav1.ObjectMeta{Name: secretName},
 		Data:       map[string][]byte{cookieEncryptionPrivateKeySecretKey: x509.MarshalPKCS1PrivateKey(generatedKey)},
 	})
+	if err != nil && !apierr.IsAlreadyExists(err) {
+		return nil, fmt.Errorf("failed to create secret: %w", err)
+	}
 	secret, err := secretsIf.Get(secretName, metav1.GetOptions{})
 	if err != nil {
 		return nil, fmt.Errorf("failed to read secret: %w", err)
diff --git a/server/workflow/workflow_server.go b/server/workflow/workflow_server.go
index 99cd84389..f6084d5e8 100644
--- a/server/workflow/workflow_server.go
+++ b/server/workflow/workflow_server.go
@@ -3,6 +3,7 @@ package workflow
 import (
 	"encoding/json"
 	"fmt"
+	"io"
 	"sort"
 
 	log "github.com/sirupsen/logrus"
@@ -210,13 +211,7 @@ func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest,
 			return nil
 		case event, open := <-watch.ResultChan():
 			if !open {
-				log.Debug("Re-establishing workflow watch")
-				watch.Stop()
-				watch, err = wfIf.Watch(*opts)
-				if err != nil {
-					return err
-				}
-				continue
+				return io.EOF
 			}
 			log.Debug("Received workflow event")
 			wf, ok := event.Object.(*wfv1.Workflow)
@@ -234,8 +229,6 @@ func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest,
 			if err != nil {
 				return err
 			}
-			// when we re-establish, we want to start at the same place
-			opts.ResourceVersion = wf.ResourceVersion
 		}
 	}
 }
@@ -264,13 +257,7 @@ func (s *workflowServer) WatchEvents(req *workflowpkg.WatchEventsRequest, ws wor
 			return nil
 		case event, open := <-watch.ResultChan():
 			if !open {
-				log.Debug("Re-establishing event watch")
-				watch.Stop()
-				watch, err = eventInterface.Watch(*opts)
-				if err != nil {
-					return err
-				}
-				continue
+				return io.EOF
 			}
 			log.Debug("Received event")
 			e, ok := event.Object.(*corev1.Event)
@@ -283,8 +270,6 @@ func (s *workflowServer) WatchEvents(req *workflowpkg.WatchEventsRequest, ws wor
 			if err != nil {
 				return err
 			}
-			// when we re-establish, we want to start at the same place
-			opts.ResourceVersion = e.ResourceVersion
 		}
 	}
 }
diff --git a/ui/src/app/assets/styles/heebo.css b/ui/src/app/assets/styles/heebo.css
deleted file mode 100644
index 4a5e16ae3..000000000
--- a/ui/src/app/assets/styles/heebo.css
+++ /dev/null
@@ -1,18 +0,0 @@
-/* hebrew */
-@font-face {
-    font-family: 'Heebo';
-    font-style: normal;
-    font-weight: 400;
-    font-display: swap;
-    src: local('Heebo'), local('Heebo-Regular'), url(../fonts/Heebo-Regular.ttf) format('ttf');
-    unicode-range: U+0590-05FF, U+20AA, U+25CC, U+FB1D-FB4F;
-}
-/* latin */
-@font-face {
-    font-family: 'Heebo';
-    font-style: normal;
-    font-weight: 400;
-    font-display: swap;
-    src: local('Heebo'), local('Heebo-Regular'), url(../fonts/Heebo-Regular.ttf) format('ttf');
-    unicode-range: U+0000-00FF, U+0131, U+0152-0153, U+02BB-02BC, U+02C6, U+02DA, U+02DC, U+2000-206F, U+2074, U+20AC, U+2122, U+2191, U+2193, U+2212, U+2215, U+FEFF, U+FFFD;
-}
\ No newline at end of file
diff --git a/ui/src/app/index.html b/ui/src/app/index.html
index ba2c7aa92..71eb50fb8 100644
--- a/ui/src/app/index.html
+++ b/ui/src/app/index.html
@@ -5,7 +5,6 @@
     <meta charset="UTF-8">
     <title>Argo</title>
     <base href="/">
-    <link href="assets/styles/heebo.css" rel="stylesheet">
     <meta name="viewport" content="width=device-width, initial-scale=1">
     <meta name="robots" content="noindex">
     <link rel="icon" type="image/png" href="assets/favicon/favicon-32x32.png" sizes="32x32">
diff --git a/ui/src/app/shared/services/requests.ts b/ui/src/app/shared/services/requests.ts
index fbfb44615..6898e8ec9 100644
--- a/ui/src/app/shared/services/requests.ts
+++ b/ui/src/app/shared/services/requests.ts
@@ -43,7 +43,22 @@ export default {
         return Observable.create((observer: Observer<any>) => {
             const eventSource = new EventSource(url);
             eventSource.onmessage = x => observer.next(x.data);
-            eventSource.onerror = x => observer.error(x);
+            eventSource.onerror = x => {
+                switch (eventSource.readyState) {
+                    case EventSource.CONNECTING:
+                        observer.error(new Error('Failed to connect to ' + url));
+                        break;
+                    case EventSource.OPEN:
+                        observer.error(new Error('Error in open connection to ' + url));
+                        break;
+                    case EventSource.CLOSED:
+                        observer.error(new Error('Connection closed to ' + url));
+                        break;
+                    default:
+                        observer.error(new Error('Unknown error with ' + url));
+                }
+            };
+
             return () => {
                 eventSource.close();
             };
diff --git a/ui/src/app/workflows/components/workflow-dag/workflow-dag.scss b/ui/src/app/workflows/components/workflow-dag/workflow-dag.scss
index fc50339c5..10e2e6de5 100644
--- a/ui/src/app/workflows/components/workflow-dag/workflow-dag.scss
+++ b/ui/src/app/workflows/components/workflow-dag/workflow-dag.scss
@@ -39,7 +39,7 @@
 
   .label {
     text-anchor: middle;
-    font-family: "Heebo", sans-serif;
+    font-family: sans-serif;
     text-rendering: optimizeLegibility;
   }
 
diff --git a/ui/src/app/workflows/components/workflow-details/workflow-details.tsx b/ui/src/app/workflows/components/workflow-details/workflow-details.tsx
index 7670a8b64..1d70c6ce6 100644
--- a/ui/src/app/workflows/components/workflow-details/workflow-details.tsx
+++ b/ui/src/app/workflows/components/workflow-details/workflow-details.tsx
@@ -47,10 +47,6 @@ export class WorkflowDetails extends React.Component<RouteComponentProps<any>, W
     private changesSubscription: Subscription;
     private timelineComponent: WorkflowTimeline;
 
-    private get resourceVersion() {
-        return this.state.workflow && this.state.workflow.metadata.resourceVersion;
-    }
-
     private get selectedTabKey() {
         return new URLSearchParams(this.props.location.search).get('tab') || 'workflow';
     }
@@ -336,7 +332,7 @@ export class WorkflowDetails extends React.Component<RouteComponentProps<any>, W
         try {
             this.ensureUnsubscribed();
             this.changesSubscription = services.workflows
-                .watch({name, namespace, resourceVersion: this.resourceVersion})
+                .watch({name, namespace, resourceVersion: '0'})
                 .map(changeEvent => changeEvent.object)
                 .subscribe(
                     workflow => this.setState({workflow, error: null}),
diff --git a/util/logs/workflow-logger.go b/util/logs/workflow-logger.go
index 5156a707d..10495277b 100644
--- a/util/logs/workflow-logger.go
+++ b/util/logs/workflow-logger.go
@@ -148,7 +148,7 @@ func WorkflowLogs(ctx context.Context, wfClient versioned.Interface, kubeClient
 	}
 
 	if req.GetLogOptions().Follow {
-		wfListOptions := metav1.ListOptions{FieldSelector: "metadata.name=" + req.GetName()}
+		wfListOptions := metav1.ListOptions{FieldSelector: "metadata.name=" + req.GetName(), ResourceVersion: "0"}
 		wfWatch, err := wfInterface.Watch(wfListOptions)
 		if err != nil {
 			return err
@@ -189,8 +189,6 @@ func WorkflowLogs(ctx context.Context, wfClient versioned.Interface, kubeClient
 					if event.Type == watch.Deleted || wf.Status.Fulfilled() {
 						return
 					}
-					// in case we re-establish the watch, make sure we start at the same place
-					wfListOptions.ResourceVersion = wf.ResourceVersion
 				}
 			}
 		}()
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 5d64eb36d..4168e8d1c 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
@@ -111,7 +109,6 @@ const (
 	workflowTemplateResyncPeriod        = 20 * time.Minute
 	podResyncPeriod                     = 30 * time.Minute
 	clusterWorkflowTemplateResyncPeriod = 20 * time.Minute
-	enoughTimeForInformerSync           = 1 * time.Second
 )
 
 // NewWorkflowController instantiates a new WorkflowController
@@ -132,8 +129,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,16 +138,17 @@ func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Int
 
 	wfc.metrics = metrics.New(wfc.getMetricsServerConfig())
 
-	workqueue.SetProvider(wfc.metrics)
-	wfc.wfQueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "workflow_queue")
+	workqueue.SetProvider(wfc.metrics) // must execute SetProvider before we created the queues
+	wfc.wfQueue = workqueue.NewNamedRateLimitingQueue(&fixedItemIntervalRateLimiter{}, "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
 }
 
 func (wfc *WorkflowController) newThrottler() sync.Throttler {
-	return sync.NewThrottler(wfc.Config.Parallelism, func(key string) { wfc.wfQueue.Add(key) })
+	return sync.NewThrottler(wfc.Config.Parallelism, func(key string) { wfc.wfQueue.AddRateLimited(key) })
 }
 
 // RunTTLController runs the workflow TTL controller
@@ -178,12 +174,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 +229,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())
 
@@ -293,7 +291,7 @@ func (wfc *WorkflowController) createSynchronizationManager() error {
 	}
 
 	nextWorkflow := func(key string) {
-		wfc.wfQueue.AddAfter(key, enoughTimeForInformerSync)
+		wfc.wfQueue.Add(key)
 	}
 
 	wfc.syncManager = sync.NewLockManager(getSyncLimit, nextWorkflow)
@@ -392,54 +390,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 +627,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)
 			}
 		}
 	}
@@ -696,9 +688,7 @@ func (wfc *WorkflowController) enqueueWfFromPodLabel(obj interface{}) error {
 		// Ignore pods unrelated to workflow (this shouldn't happen unless the watch is setup incorrectly)
 		return fmt.Errorf("Watch returned pod unrelated to any workflow")
 	}
-	// add this change after 1s - this reduces the number of workflow reconciliations -
-	//with each reconciliation doing more work
-	wfc.wfQueue.AddAfter(pod.ObjectMeta.Namespace+"/"+workflowName, enoughTimeForInformerSync)
+	wfc.wfQueue.AddRateLimited(pod.ObjectMeta.Namespace + "/" + workflowName)
 	return nil
 }
 
@@ -734,6 +724,7 @@ func (wfc *WorkflowController) addWorkflowInformerHandlers() {
 				AddFunc: func(obj interface{}) {
 					key, err := cache.MetaNamespaceKeyFunc(obj)
 					if err == nil {
+						// for a new workflow, we do not want to rate limit its execution using AddRateLimited
 						wfc.wfQueue.AddAfter(key, wfc.Config.InitialDelay.Duration)
 						priority, creation := getWfPriority(obj)
 						wfc.throttler.Add(key, priority, creation)
@@ -747,7 +738,7 @@ func (wfc *WorkflowController) addWorkflowInformerHandlers() {
 					}
 					key, err := cache.MetaNamespaceKeyFunc(new)
 					if err == nil {
-						wfc.wfQueue.Add(key)
+						wfc.wfQueue.AddRateLimited(key)
 						priority, creation := getWfPriority(new)
 						wfc.throttler.Add(key, priority, creation)
 					}
@@ -758,7 +749,7 @@ func (wfc *WorkflowController) addWorkflowInformerHandlers() {
 					key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
 					if err == nil {
 						wfc.releaseAllWorkflowLocks(obj)
-						wfc.wfQueue.Add(key)
+						wfc.wfQueue.AddRateLimited(key)
 						wfc.throttler.Remove(key)
 					}
 				},
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index acf441606..e72c949a2 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -118,8 +118,7 @@ var (
 // maxOperationTime is the maximum time a workflow operation is allowed to run
 // for before requeuing the workflow onto the workqueue.
 var (
-	maxOperationTime   = envutil.LookupEnvDurationOr("MAX_OPERATION_TIME", 30*time.Second)
-	defaultRequeueTime = envutil.LookupEnvDurationOr("DEFAULT_REQUEUE_TIME", maxOperationTime/2)
+	maxOperationTime = envutil.LookupEnvDurationOr("MAX_OPERATION_TIME", 30*time.Second)
 )
 
 // failedNodeStatus is a subset of NodeStatus that is only used to Marshal certain fields into a JSON of failed nodes
@@ -277,7 +276,7 @@ func (woc *wfOperationCtx) operate() {
 		// Workflow will not be requeued if workflow steps are in pending state.
 		// Workflow needs to requeue on its deadline,
 		if woc.workflowDeadline != nil {
-			woc.requeue(time.Until(*woc.workflowDeadline))
+			woc.requeueAfter(time.Until(*woc.workflowDeadline))
 		}
 
 		if woc.execWf.Spec.Metrics != nil {
@@ -330,7 +329,7 @@ func (woc *wfOperationCtx) operate() {
 			// Error was most likely caused by a lack of resources.
 			// In this case, Workflow will be in pending state and requeue.
 			woc.markWorkflowPhase(wfv1.NodePending, fmt.Sprintf("Waiting for a PVC to be created. %v", err))
-			woc.requeue(defaultRequeueTime)
+			woc.requeue()
 			return
 		}
 		err = fmt.Errorf("pvc create error: %w", err)
@@ -475,6 +474,7 @@ func (woc *wfOperationCtx) setGlobalParameters(executionParameters wfv1.Argument
 		cTimeVar := fmt.Sprintf("%s.%s", common.GlobalVarWorkflowCreationTimestamp, string(char))
 		woc.globalParams[cTimeVar] = strftime.Format("%"+string(char), woc.wf.ObjectMeta.CreationTimestamp.Time)
 	}
+	woc.globalParams[common.GlobalVarWorkflowCreationTimestamp+".s"] = strconv.FormatInt(woc.wf.ObjectMeta.CreationTimestamp.Time.Unix(), 10)
 
 	if workflowParameters, err := json.Marshal(woc.execWf.Spec.Arguments.Parameters); err == nil {
 		woc.globalParams[common.GlobalVarWorkflowParameters] = string(workflowParameters)
@@ -555,13 +555,14 @@ func (woc *wfOperationCtx) persistUpdates() {
 
 	woc.log.WithFields(log.Fields{"resourceVersion": woc.wf.ResourceVersion, "phase": woc.wf.Status.Phase}).Info("Workflow update successful")
 
-	if os.Getenv("INFORMER_WRITE_BACK") != "false" {
+	switch os.Getenv("INFORMER_WRITE_BACK") {
+	case "true":
 		if err := woc.writeBackToInformer(); err != nil {
 			woc.markWorkflowError(err)
 			return
 		}
-	} else {
-		time.Sleep(enoughTimeForInformerSync)
+	case "false":
+		time.Sleep(1 * time.Second)
 	}
 
 	// It is important that we *never* label pods as completed until we successfully updated the workflow
@@ -574,17 +575,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)
 		}
 	}
 }
@@ -681,15 +682,16 @@ func (woc *wfOperationCtx) reapplyUpdate(wfClient v1alpha1.WorkflowInterface, no
 }
 
 // requeue this workflow onto the workqueue for later processing
-func (woc *wfOperationCtx) requeue(afterDuration time.Duration) {
-	key, err := cache.MetaNamespaceKeyFunc(woc.wf)
-	if err != nil {
-		woc.log.Errorf("Failed to requeue workflow %s: %v", woc.wf.ObjectMeta.Name, err)
-		return
-	}
+func (woc *wfOperationCtx) requeueAfter(afterDuration time.Duration) {
+	key, _ := cache.MetaNamespaceKeyFunc(woc.wf)
 	woc.controller.wfQueue.AddAfter(key, afterDuration)
 }
 
+func (woc *wfOperationCtx) requeue() {
+	key, _ := cache.MetaNamespaceKeyFunc(woc.wf)
+	woc.controller.wfQueue.AddRateLimited(key)
+}
+
 // processNodeRetries updates the retry node state based on the child node state and the retry strategy and returns the node.
 func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrategy wfv1.RetryStrategy, opts *executeTemplateOpts) (*wfv1.NodeStatus, bool, error) {
 	if node.Fulfilled() {
@@ -769,7 +771,7 @@ func (woc *wfOperationCtx) processNodeRetries(node *wfv1.NodeStatus, retryStrate
 
 		// See if we have waited past the deadline
 		if time.Now().Before(waitingDeadline) {
-			woc.requeue(timeToWait)
+			woc.requeueAfter(timeToWait)
 			retryMessage := fmt.Sprintf("Backoff for %s", humanize.Duration(timeToWait))
 			return woc.markNodePhase(node.Name, node.Phase, retryMessage), false, nil
 		}
@@ -916,7 +918,14 @@ func (woc *wfOperationCtx) podReconciliation() error {
 
 			// If the node is pending and the pod does not exist, it could be the case that we want to try to submit it
 			// again instead of marking it as an error. Check if that's the case.
-			if node.Pending() || recentlyStarted {
+			if node.Pending() {
+				continue
+			}
+			if recentlyStarted {
+				// If the pod was deleted, then we it is possible that the controller never get another informer message about it.
+				// In this case, the workflow will only be requeued after the resync period (20m). This means
+				// workflow will not update for 20m. Requeuing here prevents that happening.
+				woc.requeue()
 				continue
 			}
 
@@ -1591,7 +1600,7 @@ func (woc *wfOperationCtx) executeTemplate(nodeName string, orgTmpl wfv1.Templat
 	// Check if we took too long operating on this workflow and immediately return if we did
 	if time.Now().UTC().After(woc.deadline) {
 		woc.log.Warnf("Deadline exceeded")
-		woc.requeue(defaultRequeueTime)
+		woc.requeue()
 		return node, ErrDeadlineExceeded
 	}
 
@@ -2336,7 +2345,7 @@ func (woc *wfOperationCtx) executeScript(nodeName string, templateScope string,
 func (woc *wfOperationCtx) requeueIfTransientErr(err error, nodeName string) (*wfv1.NodeStatus, error) {
 	if errorsutil.IsTransientErr(err) {
 		// Our error was most likely caused by a lack of resources.
-		woc.requeue(defaultRequeueTime)
+		woc.requeue()
 		return woc.markNodePending(nodeName, err), nil
 	}
 	return nil, err
@@ -2665,7 +2674,7 @@ func (woc *wfOperationCtx) executeSuspend(nodeName string, templateScope string,
 	}
 
 	if requeueTime != nil {
-		woc.requeue(time.Until(*requeueTime))
+		woc.requeueAfter(time.Until(*requeueTime))
 	}
 
 	_ = woc.markNodePhase(nodeName, wfv1.NodeRunning)
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]
+
+}
diff --git a/workflow/controller/rate_limiters.go b/workflow/controller/rate_limiters.go
new file mode 100644
index 000000000..41f949191
--- /dev/null
+++ b/workflow/controller/rate_limiters.go
@@ -0,0 +1,23 @@
+package controller
+
+import (
+	"time"
+
+	"k8s.io/client-go/util/workqueue"
+
+	"github.com/argoproj/argo/util/env"
+)
+
+type fixedItemIntervalRateLimiter struct{}
+
+func (r *fixedItemIntervalRateLimiter) When(interface{}) time.Duration {
+	return env.LookupEnvDurationOr("DEFAULT_REQUEUE_TIME", 2*time.Second)
+}
+
+func (r *fixedItemIntervalRateLimiter) Forget(interface{}) {}
+
+func (r *fixedItemIntervalRateLimiter) NumRequeues(interface{}) int {
+	return 1
+}
+
+var _ workqueue.RateLimiter = &fixedItemIntervalRateLimiter{}
diff --git a/workflow/controller/steps.go b/workflow/controller/steps.go
index 4c74f2ca3..cb9111340 100644
--- a/workflow/controller/steps.go
+++ b/workflow/controller/steps.go
@@ -93,7 +93,11 @@ func (woc *wfOperationCtx) executeSteps(nodeName string, tmplCtx *templateresolu
 
 		sgNode := woc.executeStepGroup(stepGroup.Steps, sgNodeName, &stepsCtx)
 
-		if !sgNode.Fulfilled() {
+		if sgNode.Fulfilled() {
+			if tmpl.Synchronization != nil {
+				woc.controller.syncManager.Release(woc.wf, node.ID, tmpl.Synchronization)
+			}
+		} else {
 			woc.log.Infof("Workflow step group node %s not yet completed", sgNode.ID)
 			return node, nil
 		}
diff --git a/workflow/controller/workflowpod.go b/workflow/controller/workflowpod.go
index 6345e6e0b..0cf562d35 100644
--- a/workflow/controller/workflowpod.go
+++ b/workflow/controller/workflowpod.go
@@ -142,7 +142,7 @@ func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Cont
 	if exists {
 		existing, ok := obj.(*apiv1.Pod)
 		if ok {
-			woc.log.WithField("podPhase", existing.Status.Phase).Infof("Skipped pod %s (%s) creation: already exists", nodeName, nodeID)
+			woc.log.WithField("podPhase", existing.Status.Phase).Debugf("Skipped pod %s (%s) creation: already exists", nodeName, nodeID)
 			return existing, nil
 		}
 	}


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/manifests/prometheus/kustomization.yaml b/test/e2e/manifests/prometheus/kustomization.yaml
new file mode 100644
index 000000000..3738b023c
--- /dev/null
+++ b/test/e2e/manifests/prometheus/kustomization.yaml
@@ -0,0 +1,15 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+resources:
+  - ../../../../manifests/quick-start/minimal
+  - ../../../../manifests/quick-start/base/prometheus
+
+patchesStrategicMerge:
+  - ../mixins/argo-server-deployment.yaml
+  - ../mixins/workflow-controller-configmap.yaml
+  - ../mixins/workflow-controller-deployment.yaml
+  - ../mixins/cluster-workflow-template-rbac.yaml
+
+commonLabels:
+  "app.kubernetes.io/part-of": "argo"
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..b0a413740 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
@@ -256,15 +256,7 @@ func unmarshalArtifact(yamlStr string) *wfv1.Artifact {
 }
 
 func expectWorkflow(controller *WorkflowController, name string, test func(wf *wfv1.Workflow)) {
-	obj, exists, err := controller.wfInformer.GetStore().GetByKey(name)
-	if err != nil {
-		panic(err)
-	}
-	if !exists {
-		test(nil)
-		return
-	}
-	wf, err := util.FromUnstructured(obj.(*unstructured.Unstructured))
+	wf, err := controller.wfclientset.ArgoprojV1alpha1().Workflows("").Get(name, metav1.GetOptions{})
 	if err != nil {
 		panic(err)
 	}
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index 52c83f32a..4cf64b966 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -10,6 +10,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/argoproj/pkg/strftime"
+
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/assert"
@@ -240,6 +242,11 @@ func TestGlobalParams(t *testing.T) {
 	if assert.Contains(t, woc.globalParams, "workflow.creationTimestamp") {
 		assert.NotContains(t, woc.globalParams["workflow.creationTimestamp"], "UTC")
 	}
+	for char := range strftime.FormatChars {
+		assert.Contains(t, woc.globalParams, fmt.Sprintf("%s.%s", "workflow.creationTimestamp", string(char)))
+	}
+	assert.Contains(t, woc.globalParams, "workflow.creationTimestamp.s")
+
 	assert.Contains(t, woc.globalParams, "workflow.duration")
 	assert.Contains(t, woc.globalParams, "workflow.name")
 	assert.Contains(t, woc.globalParams, "workflow.namespace")
diff --git a/workflow/sync/mutex_test.go b/workflow/sync/mutex_test.go
index 57b4613bb..39782c006 100644
--- a/workflow/sync/mutex_test.go
+++ b/workflow/sync/mutex_test.go
@@ -311,6 +311,7 @@ func TestMutexTmplLevel(t *testing.T) {
 		assert.False(t, wfUpdate)
 		assert.False(t, status)
 
+		assert.Equal(t, "synchronization-tmpl-level-mutex-vjcdk-3941195474", wf.Status.Synchronization.Mutex.Holding[0].Holder)
 		concurrenyMgr.Release(wf, "synchronization-tmpl-level-mutex-vjcdk-3941195474", tmpl.Synchronization)
 		assert.NotNil(t, wf.Status.Synchronization)
 		assert.NotNil(t, wf.Status.Synchronization.Mutex)
@@ -324,5 +325,11 @@ func TestMutexTmplLevel(t *testing.T) {
 		assert.NotNil(t, wf.Status.Synchronization)
 		assert.NotNil(t, wf.Status.Synchronization.Mutex)
 		assert.Equal(t, "synchronization-tmpl-level-mutex-vjcdk-2216915482", wf.Status.Synchronization.Mutex.Holding[0].Holder)
+
+		assert.NotEqual(t, "synchronization-tmpl-level-mutex-vjcdk-3941195474", wf.Status.Synchronization.Mutex.Holding[0].Holder)
+		concurrenyMgr.Release(wf, "synchronization-tmpl-level-mutex-vjcdk-3941195474", tmpl.Synchronization)
+		assert.NotNil(t, wf.Status.Synchronization)
+		assert.NotNil(t, wf.Status.Synchronization.Mutex)
+		assert.NotEmpty(t, wf.Status.Synchronization.Mutex.Holding)
 	})
 }