feat: Check the workflow is not being deleted for Synchronization workflow

workflow/controller/controller.go

@@ -110,6 +110,7 @@ const (
 	workflowTemplateResyncPeriod        = 20 * time.Minute
 	podResyncPeriod                     = 30 * time.Minute
 	clusterWorkflowTemplateResyncPeriod = 20 * time.Minute
+	workflowExistenceCheckPeriod        = 1 * time.Minute
 )
 
 // NewWorkflowController instantiates a new WorkflowController
@@ -243,6 +244,8 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
 				go wait.Until(wfc.syncWorkflowPhaseMetrics, 15*time.Second, ctx.Done())
 				go wait.Until(wfc.syncPodPhaseMetrics, 15*time.Second, ctx.Done())
 
+				go wait.Until(wfc.syncManager.CheckWorkflowExistence, workflowExistenceCheckPeriod, ctx.Done())
+
 				for i := 0; i < wfWorkers; i++ {
 					go wait.Until(wfc.runWorker, time.Second, ctx.Done())
 				}
@@ -297,7 +300,16 @@ func (wfc *WorkflowController) createSynchronizationManager(ctx context.Context)
 		wfc.wfQueue.AddRateLimited(key)
 	}
 
-	wfc.syncManager = sync.NewLockManager(getSyncLimit, nextWorkflow)
+	isWFDeleted := func(key string) bool {
+		_, exists, err := wfc.wfInformer.GetIndexer().GetByKey(key)
+		if err != nil {
+			log.WithFields(log.Fields{"key": key, "error": err}).Error("Failed to get workflow from informer")
+			return false
+		}
+		return exists
+	}
+
+	wfc.syncManager = sync.NewLockManager(getSyncLimit, nextWorkflow, isWFDeleted)
 
 	labelSelector := v1Label.NewSelector()
 	req, _ := v1Label.NewRequirement(common.LabelKeyPhase, selection.Equals, []string{string(wfv1.NodeRunning)})

workflow/controller/operator_concurrency_test.go

@@ -100,6 +100,10 @@ spec:
           name: workflow-controller-configmap1
 `
 
+var workflowExistenceFunc = func(key string) bool {
+	return true
+}
+
 func GetSyncLimitFunc(ctx context.Context, kube kubernetes.Interface) func(string) (int, error) {
 	syncLimitConfig := func(lockName string) (int, error) {
 		items := strings.Split(lockName, "/")
@@ -128,7 +132,7 @@ func TestSemaphoreTmplLevel(t *testing.T) {
 	defer cancel()
 	ctx := context.Background()
 	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
-	})
+	}, workflowExistenceFunc)
 	var cm v1.ConfigMap
 	err := yaml.Unmarshal([]byte(configMap), &cm)
 	assert.NoError(t, err)
@@ -191,7 +195,7 @@ func TestSemaphoreScriptTmplLevel(t *testing.T) {
 	defer cancel()
 	ctx := context.Background()
 	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
-	})
+	}, workflowExistenceFunc)
 	var cm v1.ConfigMap
 	err := yaml.Unmarshal([]byte(configMap), &cm)
 	assert.NoError(t, err)
@@ -253,7 +257,7 @@ func TestSemaphoreResourceTmplLevel(t *testing.T) {
 	defer cancel()
 	ctx := context.Background()
 	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
-	})
+	}, workflowExistenceFunc)
 	var cm v1.ConfigMap
 	err := yaml.Unmarshal([]byte(configMap), &cm)
 	assert.NoError(t, err)
@@ -316,7 +320,7 @@ func TestSemaphoreWithOutConfigMap(t *testing.T) {
 
 	ctx := context.Background()
 	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
-	})
+	}, workflowExistenceFunc)
 
 	t.Run("SemaphoreRefWithOutConfigMap", func(t *testing.T) {
 		wf := unmarshalWF(wfWithSemaphore)
@@ -373,7 +377,7 @@ func TestMutexInDAG(t *testing.T) {
 	defer cancel()
 	ctx := context.Background()
 	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
-	})
+	}, workflowExistenceFunc)
 	t.Run("MutexWithDAG", func(t *testing.T) {
 		wf := unmarshalWF(DAGWithMutex)
 		wf, err := controller.wfclientset.ArgoprojV1alpha1().Workflows(wf.Namespace).Create(ctx, wf, metav1.CreateOptions{})
@@ -437,7 +441,7 @@ func TestSynchronizationWithRetry(t *testing.T) {
 	defer cancel()
 	ctx := context.Background()
 	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
-	})
+	}, workflowExistenceFunc)
 	var cm v1.ConfigMap
 	err := yaml.Unmarshal([]byte(configMap), &cm)
 	assert.NoError(err)

workflow/sync/common.go

@@ -9,6 +9,7 @@ type Semaphore interface {
 	addToQueue(holderKey string, priority int32, creationTime time.Time)
 	removeFromQueue(holderKey string)
 	getCurrentHolders() []string
+	getCurrentPending() []string
 	getName() string
 	getLimit() int
 	resize(n int) bool

workflow/sync/mutex.go

@@ -11,6 +11,10 @@ type PriorityMutex struct {
 	lock  *sync.Mutex
 }
 
+func (m *PriorityMutex) getCurrentPending() []string {
+	return m.mutex.getCurrentPending()
+}
+
 var _ Semaphore = &PriorityMutex{}
 
 // NewMutex creates new mutex lock object

workflow/sync/mutex_test.go

@@ -94,7 +94,7 @@ func TestMutexLock(t *testing.T) {
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("InitializeSynchronization", func(t *testing.T) {
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(mutexwfstatus)
 		wfclientset := fakewfclientset.NewSimpleClientset(wf)
 
@@ -108,7 +108,7 @@ func TestMutexLock(t *testing.T) {
 		var nextWorkflow string
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
 			nextWorkflow = key
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(mutexWf)
 		wf1 := wf.DeepCopy()
 		wf2 := wf.DeepCopy()
@@ -287,7 +287,7 @@ func TestMutexTmplLevel(t *testing.T) {
 		//var nextKey string
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
 			//nextKey = key
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(mutexWfWithTmplLevel)
 		tmpl := wf.Spec.Templates[1]
 

workflow/sync/semaphore.go

@@ -46,6 +46,14 @@ func (s *PrioritySemaphore) getLimit() int {
 	return s.limit
 }
 
+func (s *PrioritySemaphore) getCurrentPending() []string {
+	var keys []string
+	for _, item := range s.pending.items {
+		keys = append(keys, item.key)
+	}
+	return keys
+}
+
 func (s *PrioritySemaphore) getCurrentHolders() []string {
 	var keys []string
 	for k := range s.lockHolder {

workflow/sync/sync_manager.go

@@ -2,6 +2,7 @@ package sync
 
 import (
 	"fmt"
+	"strings"
 	"sync"
 
 	log "github.com/sirupsen/logrus"
@@ -11,20 +12,52 @@ import (
 
 type NextWorkflow func(string)
 type GetSyncLimit func(string) (int, error)
+type IsWorkflowDeleted func(string) bool
 
 type Manager struct {
 	syncLockMap  map[string]Semaphore
 	lock         *sync.Mutex
 	nextWorkflow NextWorkflow
 	getSyncLimit GetSyncLimit
+	isWFDeleted  IsWorkflowDeleted
 }
 
-func NewLockManager(getSyncLimit GetSyncLimit, nextWorkflow NextWorkflow) *Manager {
+func NewLockManager(getSyncLimit GetSyncLimit, nextWorkflow NextWorkflow, isWFDeleted IsWorkflowDeleted) *Manager {
 	return &Manager{
 		syncLockMap:  make(map[string]Semaphore),
 		lock:         &sync.Mutex{},
 		nextWorkflow: nextWorkflow,
 		getSyncLimit: getSyncLimit,
+		isWFDeleted:  isWFDeleted,
+	}
+}
+
+func (cm *Manager) getWorkflowKey(key string) (string, error) {
+	if key == "" {
+		return "", fmt.Errorf("holderkey is empty")
+	}
+	items := strings.Split(key, "/")
+	if len(items) < 2 {
+		return "", fmt.Errorf("invalid holderkey format")
+	}
+	return fmt.Sprintf("%s/%s", items[0], items[1]), nil
+}
+
+func (cm *Manager) CheckWorkflowExistence() {
+	log.Infof("Check the workflow existence")
+	for _, lock := range cm.syncLockMap {
+		keys := lock.getCurrentHolders()
+		keys = append(keys, lock.getCurrentPending()...)
+		for _, holderKeys := range keys {
+			wfKey, err := cm.getWorkflowKey(holderKeys)
+			if err != nil {
+				continue
+			}
+			if !cm.isWFDeleted(wfKey) {
+				lock.release(holderKeys)
+				lock.removeFromQueue(holderKeys)
+			}
+		}
 	}
 }
 

workflow/sync/sync_manager_test.go

@@ -304,6 +304,10 @@ spec:
      args: ["hello world"]
 `
 
+var WorkflowExistenceFunc = func(s string) bool {
+	return false
+}
+
 func unmarshalWF(yamlStr string) *wfv1.Workflow {
 	var wf wfv1.Workflow
 	err := yaml.Unmarshal([]byte(yamlStr), &wf)
@@ -350,7 +354,7 @@ func TestSemaphoreWfLevel(t *testing.T) {
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("InitializeSynchronization", func(t *testing.T) {
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(wfWithStatus)
 		wfclientset := fakewfclientset.NewSimpleClientset(wf)
 
@@ -362,7 +366,7 @@ func TestSemaphoreWfLevel(t *testing.T) {
 	t.Run("InitializeSynchronizationWithInvalid", func(t *testing.T) {
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
 
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(wfWithStatus)
 		invalidSync := []wfv1.SemaphoreHolding{{Semaphore: "default/configmap/my-config1/workflow", Holders: []string{"hello-world-vcrg5"}}}
 		wf.Status.Synchronization.Semaphore.Holding = invalidSync
@@ -377,7 +381,7 @@ func TestSemaphoreWfLevel(t *testing.T) {
 		var nextKey string
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
 			nextKey = key
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(wfWithSemaphore)
 		wf1 := wf.DeepCopy()
 		wf2 := wf.DeepCopy()
@@ -470,7 +474,7 @@ func TestResizeSemaphoreSize(t *testing.T) {
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("WfLevelAcquireAndRelease", func(t *testing.T) {
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(wfWithSemaphore)
 		wf.CreationTimestamp = metav1.Time{Time: time.Now()}
 		wf1 := wf.DeepCopy()
@@ -541,7 +545,7 @@ func TestSemaphoreTmplLevel(t *testing.T) {
 		//var nextKey string
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
 			//nextKey = key
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(wfWithTmplSemaphore)
 		tmpl := wf.Spec.Templates[2]
 
@@ -601,7 +605,7 @@ func TestTriggerWFWithAvailableLock(t *testing.T) {
 		triggerCount := 0
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
 			triggerCount++
-		})
+		}, WorkflowExistenceFunc)
 		var wfs []wfv1.Workflow
 		for i := 0; i < 3; i++ {
 			wf := unmarshalWF(wfWithSemaphore)
@@ -637,7 +641,7 @@ func TestMutexWfLevel(t *testing.T) {
 		//var nextKey string
 		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
 			//nextKey = key
-		})
+		}, WorkflowExistenceFunc)
 		wf := unmarshalWF(wfWithMutex)
 		wf1 := wf.DeepCopy()
 		wf2 := wf.DeepCopy()
@@ -674,3 +678,48 @@ func TestMutexWfLevel(t *testing.T) {
 		assert.Len(t, mutex.mutex.pending.items, 0)
 	})
 }
+
+func TestCheckWorkflowExistence(t *testing.T) {
+	assert := assert.New(t)
+	kube := fake.NewSimpleClientset()
+	var cm v1.ConfigMap
+	err := yaml.Unmarshal([]byte(configMap), &cm)
+	cm.Data["workflow"] = "1"
+	assert.NoError(err)
+
+	ctx := context.Background()
+	_, err = kube.CoreV1().ConfigMaps("default").Create(ctx, &cm, metav1.CreateOptions{})
+	assert.NoError(err)
+
+	syncLimitFunc := GetSyncLimitFunc(kube)
+	t.Run("WorkflowDeleted", func(t *testing.T) {
+		concurrenyMgr := NewLockManager(syncLimitFunc, func(key string) {
+			//nextKey = key
+		}, func(s string) bool {
+			return strings.Contains(s, "test1")
+		})
+		wfMutex := unmarshalWF(wfWithMutex)
+		wfMutex1 := wfMutex.DeepCopy()
+		wfMutex1.Name = "test1"
+		wfSema := unmarshalWF(wfWithSemaphore)
+		wfSema1 := wfSema.DeepCopy()
+		wfSema1.Name = "test2"
+		_, _, _, _ = concurrenyMgr.TryAcquire(wfMutex, "", wfMutex.Spec.Synchronization)
+		_, _, _, _ = concurrenyMgr.TryAcquire(wfMutex1, "", wfMutex.Spec.Synchronization)
+		_, _, _, _ = concurrenyMgr.TryAcquire(wfSema, "", wfSema.Spec.Synchronization)
+		_, _, _, _ = concurrenyMgr.TryAcquire(wfSema1, "", wfSema.Spec.Synchronization)
+		mutex := concurrenyMgr.syncLockMap["default/Mutex/my-mutex"].(*PriorityMutex)
+		semaphore := concurrenyMgr.syncLockMap["default/ConfigMap/my-config/workflow"]
+
+		assert.Len(mutex.getCurrentHolders(), 1)
+		assert.Len(mutex.getCurrentPending(), 1)
+		assert.Len(semaphore.getCurrentHolders(), 1)
+		assert.Len(semaphore.getCurrentPending(), 1)
+		concurrenyMgr.CheckWorkflowExistence()
+		assert.Len(mutex.getCurrentHolders(), 0)
+		assert.Len(mutex.getCurrentPending(), 1)
+		assert.Len(semaphore.getCurrentHolders(), 0)
+		assert.Len(semaphore.getCurrentPending(), 0)
+	})
+
+}