Repository: argoproj/argo
Pull request #4687 (2020-12-09 17:55:13)

feat: Add a minimal prometheus server manifest
Signed-off-by: Simon Behar <simbeh7@gmail.com>

Checklist:

* [ ] 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`. 

-->


simster7: > Could you do a hands-on with me and Bala?

Would be happy to. Today at standup?
alexec: Stand-up demo - yes please. What would be neat? `make start RUN_MODE=kubernetes PROFILE=prometheus` and it just works.
simster7: > What would be neat? make start RUN_MODE=kubernetes PROFILE=prometheus and it just works.

Will have this ready by standup time

Patch:
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 d1d1ac2f9..75d807b36 100644
--- a/cmd/workflow-controller/main.go
+++ b/cmd/workflow-controller/main.go
@@ -39,6 +39,7 @@ func NewRootCommand() *cobra.Command {
 		logLevel                 string // --loglevel
 		glogLevel                int    // --gloglevel
 		workflowWorkers          int    // --workflow-workers
+		workflowTTLWorkers       int    // --workflow-ttl-workers
 		podWorkers               int    // --pod-workers
 		burst                    int
 		qps                      float32
@@ -84,7 +85,7 @@ func NewRootCommand() *cobra.Command {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
-			go wfController.Run(ctx, workflowWorkers, podWorkers)
+			go wfController.Run(ctx, workflowWorkers, workflowTTLWorkers, podWorkers)
 
 			// Wait forever
 			select {}
@@ -100,6 +101,7 @@ func NewRootCommand() *cobra.Command {
 	command.Flags().StringVar(&logLevel, "loglevel", "info", "Set the logging level. One of: debug|info|warn|error")
 	command.Flags().IntVar(&glogLevel, "gloglevel", 0, "Set the glog logging level")
 	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(&burst, "burst", 30, "Maximum burst for throttle.")
 	command.Flags().Float32Var(&qps, "qps", 20.0, "Queries per second")
diff --git a/docs/argo-server-sso.md b/docs/argo-server-sso.md
index 6a0482935..3c2e053c4 100644
--- a/docs/argo-server-sso.md
+++ b/docs/argo-server-sso.md
@@ -107,3 +107,53 @@ If no rule matches, we deny the user access.
     ```
     
     The precedence must be the lowest of all your service accounts. 
+
+## Sharing the Argo CD Dex Instance using Oauth2
+
+It is possible to have the Argo Workflows Server use the Argo CD Dex instance for SSO, for instance if you use Okta with SAML which cannot integrate with Argo Workflows directly. In order to make this happen, you will need the following:
+
+- You must be using at least Dex [v2.23.0](https://github.com/dexidp/dex/releases/tag/v2.23.0), because that's when `staticClients[].secretEnv` was added.
+- A secret created above with a `client-id` and `client-secret` to be used by both Dex and Argo Workflows Server. It is called `argo-workflows-sso` in this example.
+- `--auth-mode=sso` server argument added
+- A Dex `staticClients` configured for `argo-workflows-sso`
+- The `sso` configuration filled out in Argo Workflows Server to match
+
+What this might look like in your chart configuration:
+
+`argo-cd/values.yaml`:
+```yaml
+     dex:
+       image:
+         tag: v2.23.0
+       env:
+         - name: ARGO_WORKFLOWS_SSO_CLIENT_SECRET
+           valueFrom:
+             secretKeyRef:
+               name: argo-workflows-sso
+               key: client-secret
+     server:
+       config:
+         dex.config: |
+           staticClients:
+           - id: argo-workflows-sso
+             name: Argo Workflow
+             redirectURIs:
+               - https://argo-workflows.mydomain.com/oauth2/callback
+             secretEnv: ARGO_WORKFLOWS_SSO_CLIENT_SECRET
+```
+
+`argo/values.yaml`:
+```yaml
+     server:
+       extraArgs:
+         - --auth-mode=sso
+       sso:
+         issuer: https://argo-cd.mydomain.com/api/dex
+         clientId:
+           name: argo-workflows-sso
+           key: client-id
+         clientSecret:
+           name: argo-workflows-sso
+           key: client-secret
+         redirectUrl: https://argo-workflows.mydomain.com/oauth2/callback
+```
diff --git a/docs/scaling.md b/docs/scaling.md
index 96b6373b0..8b9ebe39a 100644
--- a/docs/scaling.md
+++ b/docs/scaling.md
@@ -10,8 +10,7 @@ You cannot horizontally scale the controller.
 
 You can scale the controller vertically:
 
-- If you have workflows with many steps, increase `--pod-workers`.
-- If you have many workflows, increase `--workflow-workers`. 
+- If you have many workflows, increase `--workflow-workers` and `--workflow-ttl-workers`. 
 - Increase both `--qps` and `--burst`.
 
 You will need to increase the controller's memory and CPU.
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/pkg/apis/workflow/v1alpha1/workflow_types.go b/pkg/apis/workflow/v1alpha1/workflow_types.go
index 41186e6f2..fb8526132 100644
--- a/pkg/apis/workflow/v1alpha1/workflow_types.go
+++ b/pkg/apis/workflow/v1alpha1/workflow_types.go
@@ -2510,6 +2510,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 +2528,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 +2605,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 +2621,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/gatekeeper.go b/server/auth/gatekeeper.go
index 55c135dc7..00403413a 100644
--- a/server/auth/gatekeeper.go
+++ b/server/auth/gatekeeper.go
@@ -131,7 +131,7 @@ func (s gatekeeper) getClients(ctx context.Context) (versioned.Interface, kubern
 	authorization := getAuthHeader(md)
 	mode, valid := s.Modes.GetMode(authorization)
 	if !valid {
-		return nil, nil, nil, status.Error(codes.Unauthenticated, "token not valid for requested mode")
+		return nil, nil, nil, status.Error(codes.Unauthenticated, "token not valid for running mode")
 	}
 	switch mode {
 	case Client:
diff --git a/server/auth/mode.go b/server/auth/mode.go
index 7e5e836f4..bd244527d 100644
--- a/server/auth/mode.go
+++ b/server/auth/mode.go
@@ -31,11 +31,14 @@ func (m Modes) Add(value string) error {
 }
 
 func (m Modes) GetMode(authorisation string) (Mode, bool) {
-	if strings.HasPrefix(authorisation, sso.Prefix) {
-		return SSO, m[SSO]
+	if m[SSO] && strings.HasPrefix(authorisation, sso.Prefix) {
+		return SSO, true
 	}
-	if strings.HasPrefix(authorisation, "Bearer ") || strings.HasPrefix(authorisation, "Basic ") {
-		return Client, m[Client]
+	if m[Client] && strings.HasPrefix(authorisation, "Bearer ") || strings.HasPrefix(authorisation, "Basic ") {
+		return Client, true
 	}
-	return Server, m[Server]
+	if m[Server] {
+		return Server, true
+	}
+	return "", false
 }
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/shared/template-resolution.ts b/ui/src/app/shared/template-resolution.ts
index 316b463ec..8e2c1ab42 100644
--- a/ui/src/app/shared/template-resolution.ts
+++ b/ui/src/app/shared/template-resolution.ts
@@ -16,7 +16,7 @@ export function getResolvedTemplates(workflow: models.Workflow, node: models.Nod
         if (templRef.StorageNeeded) {
             tmpl = workflow.status.storedTemplates[templRef.StoredTemplateName];
         } else if (tmpTemplate.template) {
-            tmpl = workflow.spec.templates.find(item => item.name === tmpTemplate.template);
+            tmpl = (workflow.status.storedWorkflowTemplateSpec || workflow.spec).templates.find(item => item.name === tmpTemplate.template);
         }
         if (!tmpl) {
             const name = templRef.StoredTemplateName || tmpTemplate.template;
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/ui/src/models/workflows.ts b/ui/src/models/workflows.ts
index 77f5a2f0a..8218ab907 100644
--- a/ui/src/models/workflows.ts
+++ b/ui/src/models/workflows.ts
@@ -611,9 +611,9 @@ export interface WorkflowStatus {
     conditions?: Condition[];
 
     /**
-     * StoredWorkflowSpec is a Workflow Spec of top level WorkflowTemplate.
+     * StoredWorkflowTemplateSpec is a Workflow Spec of top level WorkflowTemplate.
      */
-    storedWorkflowSpec?: WorkflowSpec;
+    storedWorkflowTemplateSpec?: WorkflowSpec;
 }
 
 export interface Condition {
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 e32defdff..5d64eb36d 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -156,9 +156,9 @@ func (wfc *WorkflowController) newThrottler() sync.Throttler {
 }
 
 // RunTTLController runs the workflow TTL controller
-func (wfc *WorkflowController) runTTLController(ctx context.Context) {
+func (wfc *WorkflowController) runTTLController(ctx context.Context, workflowTTLWorkers int) {
 	ttlCtrl := ttlcontroller.NewController(wfc.wfclientset, wfc.wfInformer)
-	err := ttlCtrl.Run(ctx.Done())
+	err := ttlCtrl.Run(ctx.Done(), workflowTTLWorkers)
 	if err != nil {
 		panic(err)
 	}
@@ -178,7 +178,7 @@ var indexers = cache.Indexers{
 }
 
 // Run starts an Workflow resource controller
-func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, podWorkers int) {
+func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWorkers, podWorkers int) {
 	defer wfc.wfQueue.ShutDown()
 	defer wfc.podQueue.ShutDown()
 
@@ -237,7 +237,7 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, podWorkers in
 				go wfc.workflowGarbageCollector(ctx.Done())
 				go wfc.archivedWorkflowGarbageCollector(ctx.Done())
 
-				go wfc.runTTLController(ctx)
+				go wfc.runTTLController(ctx, workflowTTLWorkers)
 				go wfc.runCronController(ctx)
 				go wfc.metrics.RunServer(ctx)
 				go wait.Until(wfc.syncWorkflowPhaseMetrics, 15*time.Second, ctx.Done())
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index d00dba8de..27a3ef202 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -909,21 +909,25 @@ func (woc *wfOperationCtx) podReconciliation() error {
 		}
 		if _, ok := seenPods[nodeID]; !ok {
 
+			// grace-period to allow informer sync
+			recentlyStarted := recentlyStarted(node)
+			woc.log.WithFields(log.Fields{"nodeName": node.Name, "nodePhase": node.Phase, "recentlyStarted": recentlyStarted}).Info("Workflow pod is missing")
+			metrics.PodMissingMetric.WithLabelValues(strconv.FormatBool(recentlyStarted), string(node.Phase)).Inc()
+
 			// 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() {
 				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(defaultRequeueTime)
+				continue
+			}
 
-			node.Message = "pod deleted"
-			node.Phase = wfv1.NodeError
-			// FinishedAt must be set since retry strategy depends on it to determine the backoff duration.
-			// See processNodeRetries for more details.
-			node.FinishedAt = metav1.Time{Time: time.Now().UTC()}
-			woc.wf.Status.Nodes[nodeID] = node
-			woc.log.WithField("displayName", node.DisplayName).WithField("templateName", node.TemplateName).
-				WithField("node", node.Name).Error("Pod for node deleted")
-			woc.updated = true
+			woc.markNodePhase(node.Name, wfv1.NodeError, "pod deleted")
 		} else {
 			// At this point we are certain that the pod associated with our node is running or has been run;
 			// it is safe to extract the k8s-node information given this knowledge.
@@ -937,6 +941,10 @@ func (woc *wfOperationCtx) podReconciliation() error {
 	return nil
 }
 
+func recentlyStarted(node wfv1.NodeStatus) bool {
+	return time.Since(node.StartedAt.Time) <= envutil.LookupEnvDurationOr("RECENTLY_STARTED_POD_DURATION", 10*time.Second)
+}
+
 // shouldPrintPodSpec return eligible to print to the pod spec
 func (woc *wfOperationCtx) shouldPrintPodSpec(node wfv1.NodeStatus) bool {
 	return woc.controller.Config.PodSpecLogStrategy.AllPods ||
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 39deed247..6345e6e0b 100644
--- a/workflow/controller/workflowpod.go
+++ b/workflow/controller/workflowpod.go
@@ -153,11 +153,11 @@ func (woc *wfOperationCtx) createWorkflowPod(nodeName string, mainCtr apiv1.Cont
 	mainCtr.Name = common.MainContainerName
 	// Allow customization of main container resources.
 	if isResourcesSpecified(woc.controller.Config.MainContainer) {
-		mainCtr.Resources = woc.controller.Config.MainContainer.Resources
+		mainCtr.Resources = *woc.controller.Config.MainContainer.Resources.DeepCopy()
 	}
 	// Container resources in workflow spec takes precedence over the main container's configuration in controller.
 	if isResourcesSpecified(tmpl.Container) && tmpl.Container.Name == common.MainContainerName {
-		mainCtr.Resources = tmpl.Container.Resources
+		mainCtr.Resources = *tmpl.Container.Resources.DeepCopy()
 	}
 
 	var activeDeadlineSeconds *int64
@@ -571,9 +571,9 @@ func (woc *wfOperationCtx) newExecContainer(name string, tmpl *wfv1.Template) *a
 		}
 	}
 	if isResourcesSpecified(woc.controller.Config.Executor) {
-		exec.Resources = woc.controller.Config.Executor.Resources
+		exec.Resources = *woc.controller.Config.Executor.Resources.DeepCopy()
 	} else if woc.controller.Config.ExecutorResources != nil {
-		exec.Resources = *woc.controller.Config.ExecutorResources
+		exec.Resources = *woc.controller.Config.ExecutorResources.DeepCopy()
 	}
 	if woc.controller.Config.KubeConfig != nil {
 		path := woc.controller.Config.KubeConfig.MountPath
diff --git a/workflow/executor/common/wait/wait.go b/workflow/executor/common/wait/wait.go
index 0b5910c5b..63bded23a 100644
--- a/workflow/executor/common/wait/wait.go
+++ b/workflow/executor/common/wait/wait.go
@@ -26,22 +26,32 @@ func UntilTerminated(kubernetesInterface kubernetes.Interface, namespace, podNam
 }
 
 func untilTerminatedAux(podInterface v1.PodInterface, containerID string, listOptions metav1.ListOptions) (bool, error) {
+	for {
+		timedOut, done, err := doWatch(podInterface, containerID, listOptions)
+		if !timedOut {
+			return done, err
+		}
+		log.Infof("Pod watch timed out, restarting watch on %s", containerID)
+	}
+}
+
+func doWatch(podInterface v1.PodInterface, containerID string, listOptions metav1.ListOptions) (bool, bool, error) {
 	w, err := podInterface.Watch(listOptions)
 	if err != nil {
-		return true, fmt.Errorf("could not watch pod: %w", err)
+		return false, true, fmt.Errorf("could not watch pod: %w", err)
 	}
 	defer w.Stop()
 	for event := range w.ResultChan() {
 		pod, ok := event.Object.(*corev1.Pod)
 		if !ok {
-			return false, apierrors.FromObject(event.Object)
+			return false, false, apierrors.FromObject(event.Object)
 		}
 		for _, s := range pod.Status.ContainerStatuses {
 			if common.GetContainerID(&s) == containerID && s.State.Terminated != nil {
-				return true, nil
+				return false, true, nil
 			}
 		}
 		listOptions.ResourceVersion = pod.ResourceVersion
 	}
-	return true, nil
+	return true, false, nil
 }
diff --git a/workflow/executor/resource.go b/workflow/executor/resource.go
index 8b4d35228..f39b88c6d 100644
--- a/workflow/executor/resource.go
+++ b/workflow/executor/resource.go
@@ -227,6 +227,10 @@ func checkResourceState(resourceNamespace string, resourceName string, successRe
 	}()
 
 	for {
+		if checkIfResourceDeleted(resourceName, resourceNamespace) {
+			return false, errors.Errorf(errors.CodeNotFound, "Resource %s in namespace %s has been deleted somehow.", resourceName, resourceNamespace)
+		}
+
 		jsonBytes, err := readJSON(reader)
 
 		if err != nil {
@@ -254,10 +258,6 @@ func checkResourceState(resourceNamespace string, resourceName string, successRe
 			return true, resultErr
 		}
 
-		if checkIfResourceDeleted(resourceName, resourceNamespace) {
-			return false, errors.Errorf(errors.CodeNotFound, "Resource %s in namespace %s has been deleted somehow.", resourceName, resourceNamespace)
-		}
-
 		log.Info(string(jsonBytes))
 		ls := gjsonLabels{json: jsonBytes}
 		for _, req := range failReqs {
diff --git a/workflow/metrics/pod_missing_metric.go b/workflow/metrics/pod_missing_metric.go
new file mode 100644
index 000000000..0ba2bd20c
--- /dev/null
+++ b/workflow/metrics/pod_missing_metric.go
@@ -0,0 +1,14 @@
+package metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var (
+	PodMissingMetric = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: argoNamespace,
+			Name:      "pod_missing",
+			Help:      "Incidents of pod missing",
+		},
+		[]string{"recently_started", "node_phase"},
+	)
+)
diff --git a/workflow/metrics/server.go b/workflow/metrics/server.go
index ce06cd3c7..d6223fb9c 100644
--- a/workflow/metrics/server.go
+++ b/workflow/metrics/server.go
@@ -70,6 +70,7 @@ func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
 		ch <- metric.Desc()
 	}
 	m.logMetric.Describe(ch)
+	PodMissingMetric.Describe(ch)
 }
 
 func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
@@ -77,6 +78,7 @@ func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
 		ch <- metric
 	}
 	m.logMetric.Collect(ch)
+	PodMissingMetric.Collect(ch)
 }
 
 func (m *Metrics) garbageCollector(ctx context.Context) {
diff --git a/workflow/ttlcontroller/ttlcontroller.go b/workflow/ttlcontroller/ttlcontroller.go
index dc4597fa4..6624b7c5d 100644
--- a/workflow/ttlcontroller/ttlcontroller.go
+++ b/workflow/ttlcontroller/ttlcontroller.go
@@ -24,6 +24,8 @@ import (
 
 const (
 	workflowTTLResyncPeriod = 20 * time.Minute
+	// 1s is usually enough time for the informer to get synced and be up-to-date
+	enoughTimeForInformerSync = time.Second
 )
 
 type ConfigSupplier func() *config.Config
@@ -49,28 +51,29 @@ func NewController(wfClientset wfclientset.Interface, wfInformer cache.SharedInd
 	wfInformer.AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: func(obj interface{}) bool {
 			un, ok := obj.(*unstructured.Unstructured)
-			return ok && un.GetLabels()[common.LabelKeyCompleted] == "true" && un.GetLabels()[common.LabelKeyWorkflowArchivingStatus] != "Pending"
+			return ok && un.GetDeletionTimestamp() == nil && un.GetLabels()[common.LabelKeyCompleted] == "true" && un.GetLabels()[common.LabelKeyWorkflowArchivingStatus] != "Pending"
 		},
 		Handler: cache.ResourceEventHandlerFuncs{
 			AddFunc: controller.enqueueWF,
 			UpdateFunc: func(old, new interface{}) {
 				controller.enqueueWF(new)
 			},
-			DeleteFunc: controller.enqueueWF,
 		},
 	})
 	return controller
 }
 
-func (c *Controller) Run(stopCh <-chan struct{}) error {
+func (c *Controller) Run(stopCh <-chan struct{}, workflowTTLWorkers int) error {
 	defer runtimeutil.HandleCrash()
 	defer c.workqueue.ShutDown()
-	log.Infof("Starting workflow TTL controller (resync %v)", c.resyncPeriod)
+	log.Infof("Starting workflow TTL controller (resync %v, workflowTTLWorkers %d)", c.resyncPeriod, workflowTTLWorkers)
 	go c.wfInformer.Run(stopCh)
 	if ok := cache.WaitForCacheSync(stopCh, c.wfInformer.HasSynced); !ok {
 		return fmt.Errorf("failed to wait for caches to sync")
 	}
-	go wait.Until(c.runWorker, time.Second, stopCh)
+	for i := 0; i < workflowTTLWorkers; i++ {
+		go wait.Until(c.runWorker, time.Second, stopCh)
+	}
 	log.Info("Started workflow TTL worker")
 	<-stopCh
 	log.Info("Shutting workflow TTL worker")
@@ -129,9 +132,6 @@ func (c *Controller) enqueueWF(obj interface{}) {
 		log.Warnf("'%v' is not an unstructured", obj)
 		return
 	}
-	if un.GetDeletionTimestamp() != nil {
-		return
-	}
 	wf, err := util.FromUnstructured(un)
 	if err != nil {
 		log.Warnf("Failed to unmarshal workflow %v object: %v", obj, err)
@@ -152,7 +152,12 @@ func (c *Controller) enqueueWF(obj interface{}) {
 		runtimeutil.HandleError(err)
 		return
 	}
-	//c.workqueue.Add(key)
+	// if we try and delete in the next second, it is almost certain that the informer is out of sync. Because we
+	// double-check that sees if the workflow in the informer is already deleted and we'll make 2 API requests when
+	// one is enough.
+	if addAfter < enoughTimeForInformerSync {
+		addAfter = enoughTimeForInformerSync
+	}
 	log.Infof("Queueing workflow %s/%s for delete in %v", wf.Namespace, wf.Name, addAfter)
 	c.workqueue.AddAfter(key, addAfter)
 }
@@ -177,6 +182,9 @@ func (c *Controller) deleteWorkflow(key string) error {
 		log.Warnf("Key '%s' in index is not an unstructured", key)
 		return nil
 	}
+	if un.GetDeletionTimestamp() != nil {
+		return nil
+	}
 	wf, err := util.FromUnstructured(un)
 	if err != nil {
 		log.Warnf("Failed to unmarshal key '%s' to workflow object: %v", key, err)
@@ -187,9 +195,14 @@ func (c *Controller) deleteWorkflow(key string) error {
 
 		err = c.wfclientset.ArgoprojV1alpha1().Workflows(wf.Namespace).Delete(wf.Name, &metav1.DeleteOptions{PropagationPolicy: commonutil.GetDeletePropagation()})
 		if err != nil {
-			return err
+			if apierr.IsNotFound(err) {
+				log.Infof("workflow already deleted '%s'", key)
+			} else {
+				return err
+			}
+		} else {
+			log.Infof("Successfully deleted '%s'", key)
 		}
-		log.Infof("Successfully deleted '%s'", key)
 	}
 	return nil
 }


Test patch:
diff --git a/server/auth/mode_test.go b/server/auth/mode_test.go
index 2fe5bd9b6..2a55e0e9f 100644
--- a/server/auth/mode_test.go
+++ b/server/auth/mode_test.go
@@ -60,4 +60,16 @@ func TestModes_GetMode(t *testing.T) {
 			assert.Equal(t, SSO, mode)
 		}
 	})
+
+	m = Modes{
+		Client: false,
+		SSO:    false,
+		Server: true,
+	}
+	t.Run("Server and Auth", func(t *testing.T) {
+		mode, valid := m.GetMode("Bearer ")
+		if assert.True(t, valid) {
+			assert.Equal(t, Server, mode)
+		}
+	})
 }
diff --git a/test/e2e/fixtures/then.go b/test/e2e/fixtures/then.go
index 554a59b74..c1a7251af 100644
--- a/test/e2e/fixtures/then.go
+++ b/test/e2e/fixtures/then.go
@@ -6,6 +6,7 @@ import (
 	"time"
 
 	apiv1 "k8s.io/api/core/v1"
+	apierr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
@@ -55,7 +56,14 @@ func (t *Then) expectWorkflow(workflowName string, block func(t *testing.T, meta
 		t.t.FailNow()
 	}
 	return t
+}
 
+func (t *Then) ExpectWorkflowDeleted() *Then {
+	_, err := t.client.Get(t.wf.Name, metav1.GetOptions{})
+	if err == nil || !apierr.IsNotFound(err) {
+		t.t.Fatalf("expected workflow to be deleted: %v", err)
+	}
+	return t
 }
 
 func (t *Then) ExpectCron(block func(t *testing.T, cronWf *wfv1.CronWorkflow)) *Then {
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index c8e829519..21092aab0 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -88,6 +88,30 @@ func (s *FunctionalSuite) TestDeletingRunningPodWithOrErrorRetryPolicy() {
 		})
 }
 
+func (s *FunctionalSuite) TestWorkflowTTL() {
+	s.Given().
+		Workflow(`
+metadata:
+  generateName: workflow-ttl-
+  labels:
+    argo-e2e: true
+spec:
+  ttlStrategy:
+    secondsAfterCompletion: 0
+  entrypoint: main
+  templates:
+    - name: main
+      container:
+        image: argoproj/argosay:v2
+`).
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow().
+		Wait(3 * time.Second). // enough time for TTL controller to delete the workflow
+		Then().
+		ExpectWorkflowDeleted()
+}
+
 // in this test we create a poi quota, and then  we create a workflow that needs one more pod than the quota allows
 // because we run them in parallel, the first node will run to completion, and then the second one
 func (s *FunctionalSuite) TestResourceQuota() {
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/workflow/metrics/metrics_test.go b/workflow/metrics/metrics_test.go
index 95c8bef5a..2616b9dcf 100644
--- a/workflow/metrics/metrics_test.go
+++ b/workflow/metrics/metrics_test.go
@@ -104,7 +104,7 @@ func TestMetricGC(t *testing.T) {
 	go m.garbageCollector(ctx)
 
 	// Ensure we get at least one TTL run
-	time.Sleep(1*time.Second + time.Millisecond)
+	time.Sleep(1*time.Second + 100*time.Millisecond)
 
 	assert.Len(t, m.customMetrics, 0)
 }
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)
 	})
 }
diff --git a/workflow/ttlcontroller/ttlcontroller_test.go b/workflow/ttlcontroller/ttlcontroller_test.go
index bd6d507e2..4eb6a5f59 100644
--- a/workflow/ttlcontroller/ttlcontroller_test.go
+++ b/workflow/ttlcontroller/ttlcontroller_test.go
@@ -352,6 +352,11 @@ func newTTLController() *Controller {
 	}
 }
 
+func enqueueWF(controller *Controller, un *unstructured.Unstructured) {
+	controller.enqueueWF(un)
+	time.Sleep(100*time.Millisecond + enoughTimeForInformerSync)
+}
+
 func TestEnqueueWF(t *testing.T) {
 	var err error
 	var un *unstructured.Unstructured
@@ -363,7 +368,7 @@ func TestEnqueueWF(t *testing.T) {
 	wf := test.LoadWorkflowFromBytes([]byte(completedWf))
 	un, err = util.ToUnstructured(wf)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 0, controller.workqueue.Len())
 
 	// Veirfy we do not enqueue if workflow finished is not exceed the TTL
@@ -371,7 +376,7 @@ func TestEnqueueWF(t *testing.T) {
 	wf.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-5 * time.Second)}
 	un, err = util.ToUnstructured(wf)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 0, controller.workqueue.Len())
 
 	// Verify we enqueue when ttl is expired
@@ -379,7 +384,7 @@ func TestEnqueueWF(t *testing.T) {
 	wf.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-11 * time.Second)}
 	un, err = util.ToUnstructured(wf)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 1, controller.workqueue.Len())
 }
 
@@ -396,7 +401,7 @@ func TestTTLStrategySucceded(t *testing.T) {
 	wf.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-5 * time.Second)}
 	un, err = util.ToUnstructured(wf)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 0, controller.workqueue.Len())
 
 	wf1 := test.LoadWorkflowFromBytes([]byte(succeededWf))
@@ -404,7 +409,7 @@ func TestTTLStrategySucceded(t *testing.T) {
 	wf1.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-11 * time.Second)}
 	un, err = util.ToUnstructured(wf1)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 1, controller.workqueue.Len())
 
 	wf2 := test.LoadWorkflowFromBytes([]byte(wftRefWithTTLinWFT))
@@ -413,7 +418,7 @@ func TestTTLStrategySucceded(t *testing.T) {
 	assert.NoError(t, err)
 	_, err = controller.wfclientset.ArgoprojV1alpha1().Workflows("default").Create(wf2)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	controller.processNextWorkItem()
 	assert.Equal(t, 1, controller.workqueue.Len())
 
@@ -423,7 +428,7 @@ func TestTTLStrategySucceded(t *testing.T) {
 	assert.NoError(t, err)
 	_, err = controller.wfclientset.ArgoprojV1alpha1().Workflows("default").Create(wf3)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	controller.processNextWorkItem()
 	assert.Equal(t, 1, controller.workqueue.Len())
 
@@ -442,7 +447,7 @@ func TestTTLStrategyFailed(t *testing.T) {
 	wf.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-5 * time.Second)}
 	un, err = util.ToUnstructured(wf)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 0, controller.workqueue.Len())
 
 	wf1 := test.LoadWorkflowFromBytes([]byte(failedWf))
@@ -450,11 +455,10 @@ func TestTTLStrategyFailed(t *testing.T) {
 	wf1.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-11 * time.Second)}
 	un, err = util.ToUnstructured(wf1)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 1, controller.workqueue.Len())
 
 }
-
 func TestNoTTLStrategyFailed(t *testing.T) {
 	var err error
 	var un *unstructured.Unstructured
@@ -464,14 +468,14 @@ func TestNoTTLStrategyFailed(t *testing.T) {
 	wf.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-5 * time.Second)}
 	un, err = util.ToUnstructured(wf)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 0, controller.workqueue.Len())
 
 	wf1 := test.LoadWorkflowFromBytes([]byte(failedWf))
 	wf1.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-11 * time.Second)}
 	un, err = util.ToUnstructured(wf1)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 0, controller.workqueue.Len())
 
 }
@@ -489,7 +493,7 @@ func TestNoTTLStrategyFailedButTTLSecondsAfterFinished(t *testing.T) {
 	wf.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-5 * time.Second)}
 	un, err = util.ToUnstructured(wf)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 0, controller.workqueue.Len())
 
 	wf1 := test.LoadWorkflowFromBytes([]byte(failedWf))
@@ -499,7 +503,7 @@ func TestNoTTLStrategyFailedButTTLSecondsAfterFinished(t *testing.T) {
 	wf1.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-11 * time.Second)}
 	un, err = util.ToUnstructured(wf1)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 1, controller.workqueue.Len())
 }
 
@@ -516,7 +520,7 @@ func TestTTLStrategyFromUnstructured(t *testing.T) {
 	wf.Status.FinishedAt = metav1.Time{Time: controller.clock.Now().Add(-6 * time.Second)}
 	un, err = util.ToUnstructured(wf)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 1, controller.workqueue.Len())
 
 	controller1 := newTTLController()
@@ -527,7 +531,7 @@ func TestTTLStrategyFromUnstructured(t *testing.T) {
 	wf1.Status.FinishedAt = metav1.Time{Time: controller1.clock.Now().Add(-6 * time.Second)}
 	un, err = util.ToUnstructured(wf1)
 	assert.NoError(t, err)
-	controller1.enqueueWF(un)
+	enqueueWF(controller1, un)
 	assert.Equal(t, 1, controller1.workqueue.Len())
 
 	controller2 := newTTLController()
@@ -538,7 +542,7 @@ func TestTTLStrategyFromUnstructured(t *testing.T) {
 	wf2.Status.FinishedAt = metav1.Time{Time: controller2.clock.Now().Add(-6 * time.Second)}
 	un, err = util.ToUnstructured(wf2)
 	assert.NoError(t, err)
-	controller2.enqueueWF(un)
+	enqueueWF(controller2, un)
 	assert.Equal(t, 1, controller2.workqueue.Len())
 
 	controller3 := newTTLController()
@@ -549,7 +553,7 @@ func TestTTLStrategyFromUnstructured(t *testing.T) {
 	un, err = util.ToUnstructured(wf3)
 	t.Log(wf3.Spec.TTLStrategy)
 	assert.NoError(t, err)
-	controller.enqueueWF(un)
+	enqueueWF(controller, un)
 	assert.Equal(t, 0, controller3.workqueue.Len())
 }