feat: Cache semaphore limit lookup (#14205)

Signed-off-by: Dmitri Rabinowitz <[email protected]>

Author: drabinowitz

config/config.go

@@ -108,6 +108,10 @@ type Config struct {
 	// Workflow retention by number of workflows
 	RetentionPolicy *RetentionPolicy `json:"retentionPolicy,omitempty"`
 
+	// SemaphoreLimitCacheSeconds specifies the duration in seconds before the workflow controller will re-fetch the limit
+	// for a semaphore from its associated ConfigMap(s). Defaults to 0 seconds (re-fetch every time the semaphore is checked).
+	SemaphoreLimitCacheSeconds *int64 `json:"semaphoreLimitCacheSeconds,omitempty"`
+
 	// NavColor is an ui navigation bar background color
 	NavColor string `json:"navColor,omitempty"`
 

docs/scaling.md

@@ -137,6 +137,12 @@ In order to protect users against infinite recursion, the controller has a defau
 
 This protection can be disabled with the [environment variable](environment-variables.md#controller) `DISABLE_MAX_RECURSION=true`
 
+### Caching Semaphore Limit ConfigMap Requests
+
+By default the controller will reload the ConfigMap(s) referenced by a semaphore from kube every time that workflow is queued. If you notice high latency from queuing workflows leveraging semaphores you can cache semaphore limits by editing the `semaphoreLimitCacheSeconds` parameter in [`workflow-controller-configmap.yaml`](workflow-controller-configmap.yaml).
+
+Note that this will mean that Argo will not immediately pick up changes to your config map limits.
+
 ## Miscellaneous
 
 See also [Running At Massive Scale](running-at-massive-scale.md).

docs/workflow-controller-configmap.yaml

@@ -335,6 +335,10 @@ data:
   #   failed: 3
   #   errored: 3
 
+  # SemaphoreLimitCacheSeconds specifies the duration in seconds before the workflow controller will re-fetch the limit
+  # for a semaphore from its associated ConfigMap(s). Defaults to 0 seconds (re-fetch every time the semaphore is checked).
+  semaphoreLimitCacheSeconds: "0"
+
   # Default values that will apply to all Workflows from this controller, unless overridden on the Workflow-level
   # See more: docs/default-workflow-specs.md
   workflowDefaults: |

workflow/controller/controller.go

@@ -411,7 +411,12 @@ func (wfc *WorkflowController) createSynchronizationManager(ctx context.Context)
 		return exists
 	}
 
-	wfc.syncManager = sync.NewLockManager(getSyncLimit, nextWorkflow, isWFDeleted)
+	syncLimitCacheTTL := time.Duration(0)
+	if wfc.Config.SemaphoreLimitCacheSeconds != nil {
+		syncLimitCacheTTL = time.Duration(*wfc.Config.SemaphoreLimitCacheSeconds) * time.Second
+	}
+
+	wfc.syncManager = sync.NewLockManager(getSyncLimit, syncLimitCacheTTL, nextWorkflow, isWFDeleted)
 }
 
 // list all running workflows to initialize throttler and syncManager

workflow/controller/operator_concurrency_test.go

@@ -163,7 +163,7 @@ func TestSemaphoreTmplLevel(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	var cm apiv1.ConfigMap
 	wfv1.MustUnmarshal([]byte(configMap), &cm)
@@ -224,7 +224,7 @@ func TestSemaphoreScriptTmplLevel(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	var cm apiv1.ConfigMap
 	wfv1.MustUnmarshal([]byte(configMap), &cm)
@@ -284,7 +284,7 @@ func TestSemaphoreScriptConfigMapInDifferentNamespace(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	var cm apiv1.ConfigMap
 	wfv1.MustUnmarshal([]byte(configMap), &cm)
@@ -346,7 +346,7 @@ func TestSemaphoreResourceTmplLevel(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	var cm apiv1.ConfigMap
 	wfv1.MustUnmarshal([]byte(configMap), &cm)
@@ -408,7 +408,7 @@ func TestSemaphoreWithOutConfigMap(t *testing.T) {
 	defer cancel()
 
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 
 	t.Run("SemaphoreRefWithOutConfigMap", func(t *testing.T) {
@@ -464,7 +464,7 @@ func TestMutexInDAG(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	t.Run("MutexWithDAG", func(t *testing.T) {
 		wf := wfv1.MustUnmarshalWorkflow(DAGWithMutex)
@@ -536,7 +536,7 @@ func TestMutexInDAGWithInterpolation(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	t.Run("InterpolatedMutexWithDAG", func(t *testing.T) {
 		wf := wfv1.MustUnmarshalWorkflow(DAGWithInterpolatedMutex)
@@ -601,7 +601,7 @@ func TestSynchronizationWithRetry(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	var cm apiv1.ConfigMap
 	wfv1.MustUnmarshal([]byte(configMap), &cm)
@@ -809,7 +809,7 @@ func TestSynchronizationWithStep(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	var cm apiv1.ConfigMap
 	wfv1.MustUnmarshal([]byte(configMap), &cm)
@@ -886,7 +886,7 @@ func TestSynchronizationWithStepRetry(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 	var cm apiv1.ConfigMap
 	wfv1.MustUnmarshal([]byte(configMap), &cm)
@@ -950,7 +950,7 @@ func TestSynchronizationForPendingShuttingdownWfs(t *testing.T) {
 	cancel, controller := newController()
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 
 	t.Run("PendingShuttingdownTerminatingWf", func(t *testing.T) {

workflow/controller/operator_test.go

@@ -8581,7 +8581,7 @@ func TestMutexWfPendingWithNoPod(t *testing.T) {
 	cancel, controller := newController(wf)
 	defer cancel()
 	ctx := context.Background()
-	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), func(key string) {
+	controller.syncManager = sync.NewLockManager(GetSyncLimitFunc(ctx, controller.kubeclientset), 0, func(key string) {
 	}, workflowExistenceFunc)
 
 	// preempt lock

workflow/sync/common.go

@@ -13,5 +13,10 @@ type semaphore interface {
 	getCurrentPending() []string
 	getName() string
 	getLimit() int
+	getLimitTimestamp() time.Time
+	resetLimitTimestamp()
 	resize(n int) bool
 }
+
+// expose for overriding in tests
+var nowFn = time.Now

workflow/sync/multiple_test.go

@@ -42,7 +42,7 @@ func TestMultipleMutexLock(t *testing.T) {
 	kube := fake.NewSimpleClientset()
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("MultipleMutex", func(t *testing.T) {
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {},
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {},
 			WorkflowExistenceFunc)
 		wfall := templatedWorkflow("all",
 			`    mutexes:
@@ -122,7 +122,7 @@ func TestMultipleMutexLock(t *testing.T) {
 		assert.True(t, wfUpdate)
 	})
 	t.Run("MultipleMutexOrdering", func(t *testing.T) {
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {},
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {},
 			WorkflowExistenceFunc)
 		wfall := templatedWorkflow("all",
 			`    mutexes:
@@ -204,7 +204,7 @@ func TestMutexAndSemaphore(t *testing.T) {
 
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("MutexSemaphore", func(t *testing.T) {
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {},
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {},
 			WorkflowExistenceFunc)
 		wfmands1 := templatedWorkflow("mands1",
 			`    mutexes:
@@ -322,7 +322,7 @@ func TestPriority(t *testing.T) {
 	kube := fake.NewSimpleClientset()
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("Priority", func(t *testing.T) {
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {},
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {},
 			WorkflowExistenceFunc)
 		wflow := templatedWorkflow("prioritylow",
 			`    mutexes:
@@ -399,7 +399,7 @@ func TestDuplicates(t *testing.T) {
 	kube := fake.NewSimpleClientset()
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("Mutex", func(t *testing.T) {
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {},
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {},
 			WorkflowExistenceFunc)
 		wfdupmutex := templatedWorkflow("mutex",
 			`    mutexes:
@@ -410,7 +410,7 @@ func TestDuplicates(t *testing.T) {
 		assert.Error(t, err)
 	})
 	t.Run("Semaphore", func(t *testing.T) {
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {},
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {},
 			WorkflowExistenceFunc)
 		wfdupsemaphore := templatedWorkflow("semaphore",
 			`    semaphores:

workflow/sync/mutex_test.go

@@ -114,7 +114,7 @@ func TestMutexLock(t *testing.T) {
 	kube := fake.NewSimpleClientset()
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("InitializeSynchronization", func(t *testing.T) {
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {
 		}, WorkflowExistenceFunc)
 		wf := wfv1.MustUnmarshalWorkflow(mutexwfstatus)
 		wfclientset := fakewfclientset.NewSimpleClientset(wf)
@@ -127,7 +127,7 @@ func TestMutexLock(t *testing.T) {
 	})
 	t.Run("WfLevelMutexAcquireAndRelease", func(t *testing.T) {
 		var nextWorkflow string
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {
 			nextWorkflow = key
 		}, WorkflowExistenceFunc)
 		wf := wfv1.MustUnmarshalWorkflow(mutexWf)
@@ -208,7 +208,7 @@ func TestMutexLock(t *testing.T) {
 
 	t.Run("WfLevelMutexOthernamespace", func(t *testing.T) {
 		var nextWorkflow string
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {
 			nextWorkflow = key
 		}, WorkflowExistenceFunc)
 		wf := wfv1.MustUnmarshalWorkflow(mutexWfNamespaced)
@@ -399,7 +399,7 @@ func TestMutexTmplLevel(t *testing.T) {
 	syncLimitFunc := GetSyncLimitFunc(kube)
 	t.Run("TemplateLevelAcquireAndRelease", func(t *testing.T) {
 		// var nextKey string
-		syncManager := NewLockManager(syncLimitFunc, func(key string) {
+		syncManager := NewLockManager(syncLimitFunc, 0, func(key string) {
 			// nextKey = key
 		}, WorkflowExistenceFunc)
 		wf := wfv1.MustUnmarshalWorkflow(mutexWfWithTmplLevel)

workflow/sync/semaphore.go

@@ -10,25 +10,27 @@ import (
 )
 
 type prioritySemaphore struct {
-	name         string
-	limit        int
-	pending      *priorityQueue
-	semaphore    *sema.Weighted
-	lockHolder   map[string]bool
-	nextWorkflow NextWorkflow
-	log          *log.Entry
+	name           string
+	limit          int
+	limitTimestamp time.Time
+	pending        *priorityQueue
+	semaphore      *sema.Weighted
+	lockHolder     map[string]bool
+	nextWorkflow   NextWorkflow
+	log            *log.Entry
 }
 
 var _ semaphore = &prioritySemaphore{}
 
 func NewSemaphore(name string, limit int, nextWorkflow NextWorkflow, lockType string) *prioritySemaphore {
 	return &prioritySemaphore{
-		name:         name,
-		limit:        limit,
-		pending:      &priorityQueue{itemByKey: make(map[string]*item)},
-		semaphore:    sema.NewWeighted(int64(limit)),
-		lockHolder:   make(map[string]bool),
-		nextWorkflow: nextWorkflow,
+		name:           name,
+		limit:          limit,
+		limitTimestamp: nowFn(),
+		pending:        &priorityQueue{itemByKey: make(map[string]*item)},
+		semaphore:      sema.NewWeighted(int64(limit)),
+		lockHolder:     make(map[string]bool),
+		nextWorkflow:   nextWorkflow,
 		log: log.WithFields(log.Fields{
 			lockType: name,
 		}),
@@ -43,6 +45,14 @@ func (s *prioritySemaphore) getLimit() int {
 	return s.limit
 }
 
+func (s *prioritySemaphore) getLimitTimestamp() time.Time {
+	return s.limitTimestamp
+}
+
+func (s *prioritySemaphore) resetLimitTimestamp() {
+	s.limitTimestamp = nowFn()
+}
+
 func (s *prioritySemaphore) getCurrentPending() []string {
 	var keys []string
 	for _, item := range s.pending.items {
@@ -72,6 +82,7 @@ func (s *prioritySemaphore) resize(n int) bool {
 		s.log.Infof("%s semaphore resized from %d to %d", s.name, cur, n)
 		s.semaphore = semaphore
 		s.limit = n
+		s.limitTimestamp = nowFn()
 	}
 	return status
 }

workflow/sync/sync_manager.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 	runtimeutil "k8s.io/apimachinery/pkg/util/runtime"
@@ -19,20 +20,22 @@ type (
 )
 
 type Manager struct {
-	syncLockMap  map[string]semaphore
-	lock         *sync.RWMutex
-	nextWorkflow NextWorkflow
-	getSyncLimit GetSyncLimit
-	isWFDeleted  IsWorkflowDeleted
+	syncLockMap       map[string]semaphore
+	lock              *sync.RWMutex
+	nextWorkflow      NextWorkflow
+	getSyncLimit      GetSyncLimit
+	syncLimitCacheTTL time.Duration
+	isWFDeleted       IsWorkflowDeleted
 }
 
-func NewLockManager(getSyncLimit GetSyncLimit, nextWorkflow NextWorkflow, isWFDeleted IsWorkflowDeleted) *Manager {
+func NewLockManager(getSyncLimit GetSyncLimit, syncLimitCacheTTL time.Duration, nextWorkflow NextWorkflow, isWFDeleted IsWorkflowDeleted) *Manager {
 	return &Manager{
-		syncLockMap:  make(map[string]semaphore),
-		lock:         &sync.RWMutex{},
-		nextWorkflow: nextWorkflow,
-		getSyncLimit: getSyncLimit,
-		isWFDeleted:  isWFDeleted,
+		syncLockMap:       make(map[string]semaphore),
+		lock:              &sync.RWMutex{},
+		nextWorkflow:      nextWorkflow,
+		getSyncLimit:      getSyncLimit,
+		syncLimitCacheTTL: syncLimitCacheTTL,
+		isWFDeleted:       isWFDeleted,
 	}
 }
 
@@ -462,12 +465,18 @@ func (sm *Manager) isSemaphoreSizeChanged(semaphore semaphore) (bool, int, error
 }
 
 func (sm *Manager) checkAndUpdateSemaphoreSize(semaphore semaphore) error {
+	if nowFn().Sub(semaphore.getLimitTimestamp()) < sm.syncLimitCacheTTL {
+		return nil
+	}
+
 	changed, newLimit, err := sm.isSemaphoreSizeChanged(semaphore)
 	if err != nil {
 		return err
 	}
 	if changed {
 		semaphore.resize(newLimit)
+	} else {
+		semaphore.resetLimitTimestamp()
 	}
 	return nil
 }