feat: make manifest hydration queue concurrency configurable (#27926)

cmd/argocd-application-controller/commands/argocd_application_controller.go

@@ -70,6 +70,7 @@ func NewCommand() *cobra.Command {
 		syncTimeout                      int
 		statusProcessors                 int
 		operationProcessors              int
+		hydrationProcessors              int
 		glogLevel                        int
 		metricsPort                      int
 		metricsCacheExpiration           time.Duration
@@ -243,7 +244,7 @@ func NewCommand() *cobra.Command {
 				cancel()
 			}()
 
-			go appController.Run(ctx, statusProcessors, operationProcessors)
+			go appController.Run(ctx, statusProcessors, operationProcessors, hydrationProcessors)
 
 			<-ctx.Done()
 
@@ -263,6 +264,7 @@ func NewCommand() *cobra.Command {
 	command.Flags().StringVar(&commitServerAddress, "commit-server", env.StringFromEnv("ARGOCD_APPLICATION_CONTROLLER_COMMIT_SERVER", common.DefaultCommitServerAddr), "Commit server address.")
 	command.Flags().IntVar(&statusProcessors, "status-processors", env.ParseNumFromEnv("ARGOCD_APPLICATION_CONTROLLER_STATUS_PROCESSORS", 20, 0, math.MaxInt32), "Number of application status processors")
 	command.Flags().IntVar(&operationProcessors, "operation-processors", env.ParseNumFromEnv("ARGOCD_APPLICATION_CONTROLLER_OPERATION_PROCESSORS", 10, 0, math.MaxInt32), "Number of application operation processors")
+	command.Flags().IntVar(&hydrationProcessors, "hydration-processors", env.ParseNumFromEnv("ARGOCD_APPLICATION_CONTROLLER_HYDRATION_PROCESSORS", 5, 1, math.MaxInt32), "Number of manifest hydration processors (only relevant when the Source Hydrator is enabled)")
 	command.Flags().StringVar(&cmdutil.LogFormat, "logformat", env.StringFromEnv("ARGOCD_APPLICATION_CONTROLLER_LOGFORMAT", "json"), "Set the logging format. One of: json|text")
 	command.Flags().StringVar(&cmdutil.LogLevel, "loglevel", env.StringFromEnv("ARGOCD_APPLICATION_CONTROLLER_LOGLEVEL", "info"), "Set the logging level. One of: debug|info|warn|error")
 	command.Flags().IntVar(&glogLevel, "gloglevel", 0, "Set the glog logging level")

cmd/argocd-application-controller/commands/argocd_application_controller_test.go

@@ -0,0 +1,20 @@
+package commands
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestNewCommand_HydrationProcessorsFlag pins down the contract for the manifest hydration concurrency
+// knob added for https://github.com/argoproj/argo-cd/issues/27926: the flag exists and defaults to a value
+// greater than 1 (so the default deployment exercises hydration concurrency and tests are more likely to
+// catch races, per the maintainer's guidance on the issue).
+func TestNewCommand_HydrationProcessorsFlag(t *testing.T) {
+	cmd := NewCommand()
+
+	f := cmd.Flags().Lookup("hydration-processors")
+	require.NotNil(t, f, "expected --hydration-processors flag to be registered")
+	assert.Equal(t, "5", f.DefValue, "default hydration processors should be greater than 1")
+}

controller/appcontroller.go

@@ -888,7 +888,20 @@ func (ctrl *ApplicationController) hideSecretData(destCluster *appv1.Cluster, ap
 }
 
 // Run starts the Application CRD controller.
-func (ctrl *ApplicationController) Run(ctx context.Context, statusProcessors int, operationProcessors int) {
+// normalizeHydrationProcessors clamps the configured number of manifest hydration workers to a safe
+// minimum. The --hydration-processors flag / ARGOCD_APPLICATION_CONTROLLER_HYDRATION_PROCESSORS env var
+// can be set to 0 or a negative value on the command line (the env default is clamped, but an explicit
+// flag value is not). Starting zero workers would silently stall hydration, so fall back to a single
+// worker and warn. See https://github.com/argoproj/argo-cd/issues/27926.
+func normalizeHydrationProcessors(hydrationProcessors int) int {
+	if hydrationProcessors < 1 {
+		log.Warnf("hydration-processors was set to %d; hydration requires at least one worker, using 1 instead", hydrationProcessors)
+		return 1
+	}
+	return hydrationProcessors
+}
+
+func (ctrl *ApplicationController) Run(ctx context.Context, statusProcessors int, operationProcessors int, hydrationProcessors int) {
 	defer runtime.HandleCrash()
 	defer ctrl.appRefreshQueue.ShutDown()
 	defer ctrl.appComparisonTypeRefreshQueue.ShutDown()
@@ -956,15 +969,29 @@ func (ctrl *ApplicationController) Run(ctx context.Context, statusProcessors int
 	}, time.Second, ctx.Done())
 
 	if ctrl.hydrator != nil {
+		// The app hydrate queue is keyed per application. Its only job is to decide whether the
+		// app needs hydration and, if so, enqueue the (deduped) hydration key. The Hydrating
+		// status mark and all subsequent per-app status writes live on the hydration queue side,
+		// so this worker is just an enqueuer and a single goroutine is sufficient
+		// (https://github.com/argoproj/argo-cd/issues/27926).
 		go wait.Until(func() {
 			for ctrl.processAppHydrateQueueItem() {
 			}
 		}, time.Second, ctx.Done())
 
-		go wait.Until(func() {
-			for ctrl.processHydrationQueueItem() {
-			}
-		}, time.Second, ctx.Done())
+		// The hydration queue does the heavy lifting (marking apps Hydrating, generating
+		// manifests, committing to the hydrated branch, and writing the per-app statuses) and is
+		// keyed by {SourceRepoURL, SourceTargetRevision, DestinationBranch}. Because it is a
+		// rate-limiting workqueue, the same key is never processed by two workers at once, so
+		// additional workers only parallelize hydration across *distinct* keys. The per-key dedup
+		// is also what makes the status writes safe to do inside this worker rather than ahead of
+		// time on the app hydrate queue. This is the concurrency knob requested in #27926.
+		for range normalizeHydrationProcessors(hydrationProcessors) {
+			go wait.Until(func() {
+				for ctrl.processHydrationQueueItem() {
+				}
+			}, time.Second, ctx.Done())
+		}
 	}
 
 	<-ctx.Done()

controller/appcontroller_test.go

@@ -1161,8 +1161,14 @@ func TestFinalizeAppDeletion(t *testing.T) {
 			}}, nil)
 
 			fakeAppCs := ctrl.applicationClientset.(*appclientset.Clientset)
+			// The embedded testing.Fake's RWMutex protects ReactionChain. Wrap the swap so it
+			// does not race with informer-driven Patches reading the chain via Invokes: this
+			// subtest's invalid Destination (name + server) makes the namespace indexer in
+			// newApplicationInformerAndLister invoke setAppCondition → Patch concurrently.
+			fakeAppCs.Lock()
 			defaultReactor := fakeAppCs.ReactionChain[0]
 			fakeAppCs.ReactionChain = nil
+			fakeAppCs.Unlock()
 			fakeAppCs.AddReactor("get", "*", func(action kubetesting.Action) (handled bool, ret runtime.Object, err error) {
 				return defaultReactor.React(action)
 			})

controller/hydration_concurrency_test.go

@@ -0,0 +1,178 @@
+package controller
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/client-go/util/workqueue"
+
+	hydratortypes "github.com/argoproj/argo-cd/v3/controller/hydrator/types"
+	"github.com/argoproj/argo-cd/v3/pkg/ratelimiter"
+)
+
+// TestNormalizeHydrationProcessors verifies that the hydration worker count is clamped to a safe minimum.
+// The --hydration-processors CLI flag can be set to 0 or a negative value (the env default is clamped, but
+// an explicit flag value is not), which would otherwise start zero workers and silently stall hydration.
+// See https://github.com/argoproj/argo-cd/issues/27926.
+func TestNormalizeHydrationProcessors(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name       string
+		configured int
+		want       int
+	}{
+		{"zero clamps to one", 0, 1},
+		{"negative clamps to one", -3, 1},
+		{"one stays one", 1, 1},
+		{"default preserved", 5, 5},
+		{"large value preserved", 50, 50},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.want, normalizeHydrationProcessors(tc.configured))
+		})
+	}
+}
+
+func newTestHydrationQueue() workqueue.TypedRateLimitingInterface[hydratortypes.HydrationQueueKey] {
+	return workqueue.NewTypedRateLimitingQueue(
+		ratelimiter.NewCustomAppControllerRateLimiter[hydratortypes.HydrationQueueKey](ratelimiter.GetDefaultAppRateLimiterConfig()),
+	)
+}
+
+// TestHydrationQueue_DistinctKeysProcessConcurrently asserts the core feature claim of #27926: when the
+// hydration queue is drained by multiple workers, distinct hydration keys are processed concurrently, while
+// no single key is ever processed by two workers at the same time. The worker loop mirrors the one started
+// in ApplicationController.Run.
+func TestHydrationQueue_DistinctKeysProcessConcurrently(t *testing.T) {
+	t.Parallel()
+
+	queue := newTestHydrationQueue()
+	t.Cleanup(queue.ShutDown)
+
+	const numKeys = 12
+	const numWorkers = 4
+	for i := range numKeys {
+		queue.Add(hydratortypes.HydrationQueueKey{SourceRepoURL: fmt.Sprintf("https://example.com/repo-%d", i)})
+	}
+
+	var (
+		mu             sync.Mutex
+		inFlight       int
+		maxConcurrent  int
+		processed      int
+		perKeyActive   = map[hydratortypes.HydrationQueueKey]bool{}
+		sameKeyOverlap bool
+	)
+
+	processNext := func() bool {
+		key, shutdown := queue.Get()
+		if shutdown {
+			return false
+		}
+		defer queue.Done(key)
+
+		mu.Lock()
+		if perKeyActive[key] {
+			sameKeyOverlap = true
+		}
+		perKeyActive[key] = true
+		inFlight++
+		if inFlight > maxConcurrent {
+			maxConcurrent = inFlight
+		}
+		mu.Unlock()
+
+		// Hold the key briefly so that concurrent processing of distinct keys is observable.
+		time.Sleep(20 * time.Millisecond)
+
+		mu.Lock()
+		inFlight--
+		perKeyActive[key] = false
+		processed++
+		allDone := processed == numKeys
+		mu.Unlock()
+
+		if allDone {
+			queue.ShutDown()
+		}
+		return true
+	}
+
+	var wg sync.WaitGroup
+	for range numWorkers {
+		wg.Go(func() {
+			for processNext() {
+			}
+		})
+	}
+	wg.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, numKeys, processed, "every distinct key should be processed exactly once")
+	assert.False(t, sameKeyOverlap, "the same hydration key must never be processed by two workers at once")
+	assert.GreaterOrEqual(t, maxConcurrent, 2, "distinct hydration keys should be processed concurrently with multiple workers")
+}
+
+// TestHydrationQueue_SameKeyNotProcessedConcurrently asserts the dedup guarantee the feature relies on: a
+// hydration key that is re-enqueued while it is still in-flight is withheld until the in-flight processing
+// calls Done, so the same key is never handed to two workers simultaneously. See #27926.
+func TestHydrationQueue_SameKeyNotProcessedConcurrently(t *testing.T) {
+	t.Parallel()
+
+	queue := newTestHydrationQueue()
+	t.Cleanup(queue.ShutDown)
+
+	key := hydratortypes.HydrationQueueKey{SourceRepoURL: "https://example.com/repo", DestinationBranch: "env/dev"}
+	queue.Add(key)
+
+	started := make(chan struct{})
+	release := make(chan struct{})
+	secondPickup := make(chan hydratortypes.HydrationQueueKey, 1)
+
+	// Worker 1 takes the key and holds it (does not call Done) until released.
+	go func() {
+		k, shutdown := queue.Get()
+		if shutdown {
+			return
+		}
+		close(started)
+		<-release
+		queue.Done(k)
+	}()
+
+	<-started
+	// Re-enqueue the same key while it is still in-flight. The workqueue must not hand it to another
+	// worker until the first one calls Done.
+	queue.Add(key)
+
+	// Worker 2 attempts to take work; it must block while the key is in-flight.
+	go func() {
+		k, shutdown := queue.Get()
+		if !shutdown {
+			secondPickup <- k
+		}
+	}()
+
+	select {
+	case <-secondPickup:
+		t.Fatal("the same hydration key was delivered to a second worker while still in-flight")
+	case <-time.After(150 * time.Millisecond):
+		// Expected: the re-added key is withheld until the first worker calls Done.
+	}
+
+	// Release the first worker; the re-added key should now become available to worker 2.
+	close(release)
+	select {
+	case got := <-secondPickup:
+		assert.Equal(t, key, got)
+		queue.Done(got)
+	case <-time.After(2 * time.Second):
+		t.Fatal("re-added key was not delivered after the first worker finished")
+	}
+}