Repository: argoproj/argo (go)

Issue:
Workflows with mutexes get stuck
## Summary 

I'm trying to run a few workflows that are using a mutex, however if I run ~25 steps on the same mutex, the workflows get stuck and can't grab the mutex.

What happened/what you expected to happen?

The workflows get stucks and hangs indefinitely. Deleting the stuck workflows with the web-ui or kubectl does not release the mutex so new workflows can't run until the workflow-controller have been restarted.

## Diagnostics

What Kubernetes provider are you using? 

We are using Azure AKS with Kubernetes v1.18.10.

What version of Argo Workflows are you running? 

I have first encountered the issue with our argo v2.11.7 (bf3fec176cf6bdf3e23b2cb73ec7d4e3d051ca40) with a cluster install.
Then i tested this in a different k8s cluster (still on azure) with argo v2.11.6, v2.11.7 and v2.11.8 in a namespaced environment.
The logs in this issue will be from my test argo workow v2.11.8 (310e099f82520030246a7c9d66f3efaadac9ade2).

The issue is only appearing with this specific workflowtemplate architecture. The result is mostly random.

WorkflowTemplate used to reproduce the issue:
```yaml
---
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
  name: test
spec:
  entrypoint: entrypoint
  onExit: on-exit
  templates:
  - name: entrypoint
    steps:
    - - name: generate-jobs
        template: generate-jobs
    - - arguments:
          parameters:
          - name: mutex_name
            value: '{{item}}'
        name: test
        template: main-function
        withParam: '{{steps.generate-jobs.outputs.result}}'

  - name: generate-jobs
    script:
      command: [ cat ]
      image: alpine:3.6
      source: |
        ["config", "config", "config", "config", "config"]

  - name: main-function
    inputs:
      parameters:
      - name: mutex_name
    synchronization:
      mutex:
        name: '{{inputs.parameters.mutex_name}}'
    steps:
    - - name: check
        templateRef:
          name: test
          template: main-command

  - name: main-command
    container:
      command: [ sh ]
      args: [ "-c", "sleep 5" ]
      image: alpine:3.6

  - name: on-exit
    script:
      command:
      - cat
      image: alpine:3.6
      source: |
        echo fixme
```

Shell command to trigger the issue:
```
for i in $(seq 5); do argo submit --from workflowtemplate/test ; done
```

I take as an example the case where I launch 5 workflows from the workflow template.
One workflow has succeeded, but all others are stuck.
(I'm using github gists to keep the issue readable)

The workflow that succeeded: [GIST](https://gist.githubusercontent.com/Fumesover/0360ea227560f4212e1a22ac40fda4b4/raw/11a9a1ffc9f39345a6170ee14eb4a47d6ca7eebc/test-d44k6.yaml)
A stuck workflow: [GIST](https://gist.githubusercontent.com/Fumesover/0360ea227560f4212e1a22ac40fda4b4/raw/11a9a1ffc9f39345a6170ee14eb4a47d6ca7eebc/test-hkkhv.yaml)
workflow-controller logs: [GIST](https://gist.githubusercontent.com/Fumesover/0360ea227560f4212e1a22ac40fda4b4/raw/11a9a1ffc9f39345a6170ee14eb4a47d6ca7eebc/workflow-controller-79d879b95d-q6fqk.logs)

Apparently, the mutex is still held by one of the "argo nodes" (and not by a pod). But I can't figure out what kind of object it is in Kubernetes, so I don't know where to look for the cause of this behavior.

I'm on the argoproj slack (`@Albin Parou`) if you have any questions regarding this issue. :slightly_smiling_face: 

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.


Pull request description:
fix(controller): Fixed workflow stuck with mutex lock
Signed-off-by: Saravanan Balasubramanian <sarabala1979@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`. 

-->
Fixes #4703 

Patch:
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/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/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/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/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/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/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/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())
 }