Repository: argoproj/argo (go)

Issue:
synchronization semaphore lost on delete
## Summary 

I had a large number of pending workflows (~50-100) stuck behind my synchronization semaphore (~20) and decided to shed load by deleting all workflows. I expected this to free up slots in the semaphore but it seems like deleting the workflows did not release those slots.

## Diagnostics

What Kubernetes provider are you using? 
running on EKS
```
$ kubectl --version
Client Version: version.Info{Major:"1", Minor:"19", GitVersion:"v1.19.3", GitCommit:"1e11e4a2108024935ecfcb2912226cedeafd99df", GitTreeState:"clean", BuildDate:"2020-10-14T12:50:19Z", GoVersion:"go1.15.2", Compiler:"gc", Platform:"linux/amd64"}
Server Version: version.Info{Major:"1", Minor:"18+", GitVersion:"v1.18.9-eks-d1db3c", GitCommit:"d1db3c46e55f95d6a7d3e5578689371318f95ff9", GitTreeState:"clean", BuildDate:"2020-10-20T22:18:07Z", GoVersion:"go1.13.15", Compiler:"gc", Platform:"linux/amd64"}
```

What version of Argo Workflows are you running? 
v2.11.7

## Reproduction
```yaml
# example_lock.yml
apiVersion: v1
kind: ConfigMap
metadata:
  name: my-config
data:
  workflow: "3"
```

```yaml
# example_workflow.yml
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: synchronization-wf-level-
spec:
  entrypoint: whalesay
  synchronization:
    semaphore:
      configMapKeyRef:
        name: my-config
        key: workflow
  templates:
    - name: whalesay
      container:
        image: alpine:latest
        command: [sh, -c, "sleep 5; echo hi"]
        imagePullPolicy: "IfNotPresent"

```


```bash
kubectl apply -f example_lock.yml -n argo
# do the below a few times:
# run workflow with some concurrency
argo submit example_workflow.yml -n argo & argo submit example_workflow.yml -n argo & argo submit example_workflow.yml -n argo & argo submit example_workflow.yml -n argo & argo submit exam
ple_workflow.yml -n argo
# delete running containers
argo delete --all -n argo
```

Paste the logs from the workflow controller:
```
time="2020-12-02T19:53:50Z" level=warning msg="Workflow processing has been postponed due to concurrency limit" key=argo/synchronization-wf-level-hwm2l message="Waiting for argo/ConfigMap/my-config/workflow lock. Lock status: 0/3 "
time="2020-12-02T19:53:50Z" level=warning msg="Workflow processing has been postponed due to concurrency limit" key=argo/synchronization-wf-level-94zjm message="Waiting for argo/ConfigMap/my-config/workflow lock. Lock status: 0/3 "
time="2020-12-02T19:53:51Z" level=warning msg="Workflow processing has been postponed due to concurrency limit" key=argo/synchronization-wf-level-hwm2l message="Waiting for argo/ConfigMap/my-config/workflow lock. Lock status: 0/3 "
time="2020-12-02T19:53:51Z" level=warning msg="Workflow processing has been postponed due to concurrency limit" key=argo/synchronization-wf-level-94zjm message="Waiting for argo/ConfigMap/my-config/workflow lock. Lock status: 0/3 "
time="2020-12-02T19:53:52Z" level=warning msg="Workflow processing has been postponed due to concurrency limit" key=argo/synchronization-wf-level-hwm2l message="Waiting for argo/ConfigMap/my-config/workflow lock. Lock status: 0/3 "
time="2020-12-02T19:53:57Z" level=warning msg="Workflow processing has been postponed due to concurrency limit" key=argo/synchronization-wf-level-94zjm message="Waiting for argo/ConfigMap/my-config/workflow lock. Lock status: 0/3 "
time="2020-12-02T19:53:59Z" level=warning msg="Workflow processing has been postponed due to concurrency limit" key=argo/synchronization-wf-level-5wsrz message="Waiting for argo/ConfigMap/my-config/workflow lock. Lock status: 3/3 "
time="2020-12-02T19:53:59Z" level=warning msg="Workflow processing has been postponed due to concurrency limit" key=argo/synchronization-wf-level-lkx7p message="Waiting for argo/ConfigMap/my-config/workflow lock. Lock status: 3/3 "
```

---
<!-- 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:
fix(controller): Cleanup the synchronize  pending queue once Workflow get deleted
… 

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 #4632 

Patch:
diff --git a/workflow/sync/sync_manager.go b/workflow/sync/sync_manager.go
index fe0ae39ef..6cd14d9ae 100644
--- a/workflow/sync/sync_manager.go
+++ b/workflow/sync/sync_manager.go
@@ -183,6 +183,16 @@ func (cm *Manager) ReleaseAll(wf *wfv1.Workflow) bool {
 				log.Infof("%s released a lock from %s", resourceKey, holding.Semaphore)
 			}
 		}
+
+		// Remove the pending Workflow level semaphore keys
+		for _, waiting := range wf.Status.Synchronization.Semaphore.Waiting {
+			syncLockHolder := cm.syncLockMap[waiting.Semaphore]
+			if syncLockHolder == nil {
+				continue
+			}
+			resourceKey := getResourceKey(wf.Namespace, wf.Name, wf.Name)
+			syncLockHolder.removeFromQueue(resourceKey)
+		}
 		wf.Status.Synchronization.Semaphore = nil
 	}
 
@@ -198,6 +208,16 @@ func (cm *Manager) ReleaseAll(wf *wfv1.Workflow) bool {
 			wf.Status.Synchronization.Mutex.LockReleased(holding.Holder, holding.Mutex)
 			log.Infof("%s released a lock from %s", resourceKey, holding.Mutex)
 		}
+
+		// Remove the pending Workflow level mutex keys
+		for _, waiting := range wf.Status.Synchronization.Mutex.Waiting {
+			syncLockHolder := cm.syncLockMap[waiting.Mutex]
+			if syncLockHolder == nil {
+				continue
+			}
+			resourceKey := getResourceKey(wf.Namespace, wf.Name, wf.Name)
+			syncLockHolder.removeFromQueue(resourceKey)
+		}
 		wf.Status.Synchronization.Mutex = nil
 	}
 


Test patch:
diff --git a/workflow/sync/sync_manager_test.go b/workflow/sync/sync_manager_test.go
index 5a12c594d..d11ee32e4 100644
--- a/workflow/sync/sync_manager_test.go
+++ b/workflow/sync/sync_manager_test.go
@@ -284,6 +284,24 @@ status:
   phase: Running
   startedAt: "2020-06-04T19:55:11Z"
 `
+const wfWithMutex = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+ name: hello-world
+ namespace: default
+spec:
+ entrypoint: whalesay
+ synchronization:
+   mutex:
+     name: my-mutex
+ templates:
+ - name: whalesay
+   container:
+     image: docker/whalesay:latest
+     command: [cowsay]
+     args: ["hello world"]
+`
 
 func unmarshalWF(yamlStr string) *wfv1.Workflow {
 	var wf wfv1.Workflow
@@ -423,6 +441,14 @@ func TestSemaphoreWfLevel(t *testing.T) {
 
 		concurrenyMgr.ReleaseAll(wf2)
 		assert.Nil(t, wf2.Status.Synchronization)
+
+		sema := concurrenyMgr.syncLockMap["default/ConfigMap/my-config/workflow"].(*PrioritySemaphore)
+		assert.NotNil(t, sema)
+		assert.Len(t, sema.pending.items, 2)
+		concurrenyMgr.ReleaseAll(wf1)
+		assert.Len(t, sema.pending.items, 1)
+		concurrenyMgr.ReleaseAll(wf3)
+		assert.Len(t, sema.pending.items, 0)
 	})
 }
 
@@ -589,3 +615,50 @@ func TestTriggerWFWithAvailableLock(t *testing.T) {
 		assert.Equal(2, triggerCount)
 	})
 }
+
+func TestMutexWfLevel(t *testing.T) {
+	kube := fake.NewSimpleClientset()
+	syncLimitFunc := GetSyncLimitFunc(kube)
+	t.Run("WorkflowLevelMutexAcquireAndRelease", func(t *testing.T) {
+		//var nextKey string
+		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
+			//nextKey = key
+		})
+		wf := unmarshalWF(wfWithMutex)
+		wf1 := wf.DeepCopy()
+		wf2 := wf.DeepCopy()
+
+		status, wfUpdate, msg, err := concurrenyMgr.TryAcquire(wf, "", wf.Spec.Synchronization)
+		assert.NoError(t, err)
+		assert.Empty(t, msg)
+		assert.True(t, status)
+		assert.True(t, wfUpdate)
+		assert.NotNil(t, wf.Status.Synchronization)
+		assert.NotNil(t, wf.Status.Synchronization.Mutex)
+		assert.NotNil(t, wf.Status.Synchronization.Mutex.Holding)
+
+		wf1.Name = "two"
+		status, wfUpdate, msg, err = concurrenyMgr.TryAcquire(wf1, "", wf1.Spec.Synchronization)
+		assert.NoError(t, err)
+		assert.NotEmpty(t, msg)
+		assert.False(t, status)
+		assert.True(t, wfUpdate)
+
+		wf2.Name = "three"
+		status, wfUpdate, msg, err = concurrenyMgr.TryAcquire(wf2, "", wf2.Spec.Synchronization)
+		assert.NoError(t, err)
+		assert.NotEmpty(t, msg)
+		assert.False(t, status)
+		assert.True(t, wfUpdate)
+
+		mutex := concurrenyMgr.syncLockMap["default/Mutex/my-mutex"].(*PriorityMutex)
+		assert.NotNil(t, mutex)
+		assert.Len(t, mutex.mutex.pending.items, 2)
+		concurrenyMgr.ReleaseAll(wf1)
+		assert.Len(t, mutex.mutex.pending.items, 1)
+		concurrenyMgr.ReleaseAll(wf2)
+		assert.Len(t, mutex.mutex.pending.items, 0)
+
+	})
+
+}
\ No newline at end of file