fix: Mutex not being released on step completion (#4847)

.gitignore

@@ -37,3 +37,5 @@ git-ask-pass.sh
 /go-diagrams/
 /.run/
 pprof
+pkg/apiclient/sensor/sensor.swagger.json
+pkg/apiclient/eventsource/eventsource.swagger.json

workflow/controller/operator.go

@@ -1702,6 +1702,9 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
  woc.computeMetrics(processedTmpl.Metrics.Prometheus, localScope, realTimeScope, false)
  }
  }
+ if processedTmpl.Synchronization != nil {
+ woc.controller.syncManager.Release(woc.wf, node.ID, processedTmpl.Synchronization)
+ }
  return retryParentNode, nil
  }
  lastChildNode := getChildNodeIndex(retryParentNode, woc.wf.Status.Nodes, -1)

workflow/controller/operator_concurrency_test.go

@@ -398,3 +398,87 @@ func TestMutexInDAG(t *testing.T) {
  }
  })
 }
+
+const RetryWfWithSemaphore = `
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata: 
+ name: script-wf
+ namespace: default
+spec: 
+ entrypoint: step1
+ retryStrategy:
+ limit: 10 
+ templates:
+ - name: step1
+ steps:
+ - - name: hello1
+ template: whalesay
+ - - name: hello2
+ template: whalesay 
+ - name: whalesay
+ daemon: true
+ synchronization: 
+ semaphore: 
+ configMapKeyRef: 
+ key: template
+ name: my-config
+ container: 
+ args: 
+ - "hello world"
+ command: 
+ - cowsay
+ image: "docker/whalesay:latest"
+`
+
+func TestSynchronizationWithRetry(t *testing.T) {
+ assert := assert.New(t)
+ cancel, controller := newController()
+ defer cancel()
+ ctx := context.Background()
+ controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+ })
+ var cm v1.ConfigMap
+ err := yaml.Unmarshal([]byte(configMap), &cm)
+ assert.NoError(err)
+ _, err = controller.kubeclientset.CoreV1().ConfigMaps("default").Create(ctx, &cm, metav1.CreateOptions{})
+ assert.NoError(err)
+ t.Run("WorkflowWithRetry", func(t *testing.T) {
+ wf := unmarshalWF(RetryWfWithSemaphore)
+ wf, err := controller.wfclientset.ArgoprojV1alpha1().Workflows(wf.Namespace).Create(ctx, wf, metav1.CreateOptions{})
+ assert.NoError(err)
+ woc := newWorkflowOperationCtx(wf, controller)
+ woc.operate(ctx)
+ for _, node := range woc.wf.Status.Nodes {
+ if node.Name == "hello1" {
+ assert.Equal(wfv1.NodePending, node.Phase)
+ }
+ }
+
+ // Updating Pod state
+ makePodsPhase(ctx, woc, v1.PodSucceeded)
+
+ // Release the lock from hello1
+ woc = newWorkflowOperationCtx(woc.wf, controller)
+ woc.operate(ctx)
+ for _, node := range woc.wf.Status.Nodes {
+ if node.Name == "hello1" {
+ assert.Equal(wfv1.NodeSucceeded, node.Phase)
+ }
+ if node.Name == "hello2" {
+ assert.Equal(wfv1.NodePending, node.Phase)
+ }
+ }
+ // Updating Pod state
+ makePodsPhase(ctx, woc, v1.PodSucceeded)
+
+ // Release the lock from hello2
+ woc = newWorkflowOperationCtx(woc.wf, controller)
+ woc.operate(ctx)
+ // Nobody is waiting for the lock
+ assert.Empty(woc.wf.Status.Synchronization.Semaphore.Waiting)
+ // Nobody is holding the lock
+ assert.Empty(woc.wf.Status.Synchronization.Semaphore.Holding[0].Holders)
+
+ })
+}

workflow/sync/sync_manager.go

@@ -155,6 +155,7 @@ func (cm *Manager) Release(wf *wfv1.Workflow, nodeName string, syncRef *wfv1.Syn
 
  if syncLockHolder, ok := cm.syncLockMap[lockName.EncodeName()]; ok {
  syncLockHolder.release(holderKey)
+ syncLockHolder.removeFromQueue(holderKey)
  log.Debugf("%s sync lock is released by %s", lockName.EncodeName(), holderKey)
  lockKey := lockName.EncodeName()
  wf.Status.Synchronization.GetStatus(syncRef.GetType()).LockReleased(holderKey, lockKey)