feat(controller): add watch-list for initial cluster cache sync (argoproj#77) (argoproj#78)

* feat(controller): add watch-list for initial cluster cache sync

The cluster cache's initial sync currently issues a LIST with
ResourceVersion=0 per resource kind. That serves from the apiserver
watch cache (no etcd) but returns every object in a single response
body, because the apiserver ignores the limit parameter when serving
from cache. On clusters with many objects of one kind (e.g. 17k
StatefulSets on oxygen) the response exceeds what the GKE
control-plane load balancer will carry, and the HTTP/2 stream is
reset with INTERNAL_ERROR before the controller reads it. The cluster
cache never finishes initializing, so every Application on the shard
goes into ComparisonError and stays there.

This adds an opt-in watch-list path (KEP-3157, GA in k8s 1.32) that
opens a WATCH with sendInitialEvents=true instead of a LIST. The
apiserver streams each object as an individual ADDED event, then
sends a bookmark annotated with k8s.io/initial-events-end to mark the
snapshot boundary. Same watch-cache source, same etcd-free guarantee,
but the response is a stream of small events rather than one giant
body — long-lived watch streams are what the LB already handles for
every informer in the cluster.

Enable with ARGOCD_CLUSTER_CACHE_WATCH_LIST=true. On clusters where
the WatchList feature gate is disabled server-side (k8s <1.32 by
default), the first rejected request latches a per-cluster flag and
subsequent kinds fall back to the LIST path, so the cost is one 400
round-trip at most.

* add retry/backoff to watchListResources

Mirror the retry.OnError pattern from listResources so both initial-sync
paths honor the same listRetryLimit/listRetryUseBackoff/listRetryFunc
knobs. The retry predicate skips BadRequest/Invalid so the
fallback-to-LIST latch stays a single fast round-trip on clusters
without the WatchList feature gate.

Addresses review feedback on argoproj#77.

Author: fsouza

controller/cache/cache.go

@@ -75,6 +75,12 @@ const (
 	// EnvClusterCacheEventsProcessingInterval is the env variable to control the interval between processing events when BatchEventsProcessing is enabled
 	EnvClusterCacheEventsProcessingInterval = "ARGOCD_CLUSTER_CACHE_EVENTS_PROCESSING_INTERVAL"
 
+	// EnvClusterCacheWatchList enables watch-list (sendInitialEvents) for
+	// initial cluster cache sync. Requires the WatchList feature gate on the
+	// target apiserver (GA in k8s 1.32). Falls back to LIST per cluster if
+	// the apiserver rejects the request.
+	EnvClusterCacheWatchList = "ARGOCD_CLUSTER_CACHE_WATCH_LIST"
+
 	// AnnotationIgnoreResourceUpdates when set to true on an untracked resource,
 	// argo will apply `ignoreResourceUpdates` configuration on it.
 	AnnotationIgnoreResourceUpdates = "argocd.argoproj.io/ignore-resource-updates"
@@ -115,6 +121,9 @@ var (
 
 	// clusterCacheEventsProcessingInterval specifies the interval between processing events when BatchEventsProcessing is enabled
 	clusterCacheEventsProcessingInterval = 100 * time.Millisecond
+
+	// clusterCacheWatchList specifies whether to use watch-list for initial sync
+	clusterCacheWatchList = false
 )
 
 func init() {
@@ -128,6 +137,7 @@ func init() {
 	clusterCacheRetryUseBackoff = env.ParseBoolFromEnv(EnvClusterCacheRetryUseBackoff, false)
 	clusterCacheBatchEventsProcessing = env.ParseBoolFromEnv(EnvClusterCacheBatchEventsProcessing, true)
 	clusterCacheEventsProcessingInterval = env.ParseDurationFromEnv(EnvClusterCacheEventsProcessingInterval, clusterCacheEventsProcessingInterval, 0, math.MaxInt64)
+	clusterCacheWatchList = env.ParseBoolFromEnv(EnvClusterCacheWatchList, clusterCacheWatchList)
 }
 
 type LiveStateCache interface {
@@ -587,6 +597,7 @@ func (c *liveStateCache) getCluster(cluster *appv1.Cluster) (clustercache.Cluste
 		clustercache.SetRespectRBAC(respectRBAC),
 		clustercache.SetBatchEventsProcessing(clusterCacheBatchEventsProcessing),
 		clustercache.SetEventProcessingInterval(clusterCacheEventsProcessingInterval),
+		clustercache.SetWatchList(clusterCacheWatchList),
 	}
 
 	clusterCache = clustercache.NewClusterCache(clusterCacheConfig, clusterCacheOpts...)

gitops-engine/pkg/cache/cluster.go

@@ -26,10 +26,12 @@ package cache
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"runtime/debug"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -55,11 +57,17 @@ import (
 	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2/textlogger"
 	"k8s.io/kubectl/pkg/util/openapi"
+	"k8s.io/utils/ptr"
 
 	"github.com/argoproj/gitops-engine/pkg/utils/kube"
 	"github.com/argoproj/gitops-engine/pkg/utils/tracing"
 )
 
+// errWatchListUnsupported signals that the apiserver rejected a
+// sendInitialEvents watch request and the caller should fall back
+// to a LIST-based sync.
+var errWatchListUnsupported = errors.New("apiserver does not support watch-list (sendInitialEvents)")
+
 const (
 	watchResourcesRetryTimeout = 1 * time.Second
 	ClusterRetryTimeout        = 10 * time.Second
@@ -284,6 +292,17 @@ type clusterCache struct {
 	// Maps any resource's UID to its direct children's ResourceKeys
 	// Eliminates need for O(n) graph building during hierarchy traversal
 	parentUIDToChildren map[types.UID][]kube.ResourceKey
+
+	// watchListEnabled opts this cache into using watch-list
+	// (sendInitialEvents) for initial state sync instead of LIST. Requires
+	// the WatchList feature gate on the apiserver (GA in k8s 1.32).
+	watchListEnabled bool
+	// watchListUnsupported latches to true after the first sendInitialEvents
+	// watch is rejected by the apiserver, so subsequent syncs skip straight
+	// to the LIST-based path. This avoids a 400 round-trip per group/kind
+	// on clusters where watchListEnabled is set but the feature gate is
+	// disabled server-side.
+	watchListUnsupported atomic.Bool
 }
 
 type clusterCacheSync struct {
@@ -705,6 +724,138 @@ func runSynced(lock sync.Locker, action func() error) error {
 	return action()
 }
 
+// watchListResources loads the initial state of a resource kind using a
+// watch request with sendInitialEvents=true (KEP-3157) instead of a LIST.
+// This is served from the apiserver's watch cache (no etcd round-trip) and
+// delivers objects as a stream of ADDED events terminated by a bookmark,
+// avoiding the single oversized response body that a ResourceVersion=0 LIST
+// produces on clusters with many objects of one kind.
+//
+// Returns errWatchListUnsupported if the apiserver rejects the request
+// because the WatchList feature gate is disabled (k8s <1.32 by default).
+// Callers should fall back to listResources on that error.
+func (c *clusterCache) watchListResources(ctx context.Context, resClient dynamic.ResourceInterface, callback func(*unstructured.Unstructured) error) (string, error) {
+	if err := c.listSemaphore.Acquire(ctx, 1); err != nil {
+		return "", fmt.Errorf("failed to acquire list semaphore: %w", err)
+	}
+	defer c.listSemaphore.Release(1)
+
+	// Bound the initial-state stream so a slow or stalled apiserver does not
+	// tie up a watch slot indefinitely. Reuse the watch resync timeout as the
+	// upper bound, matching the behaviour of the subsequent long-running watch.
+	timeoutSeconds := int64(c.watchResyncTimeout.Seconds())
+	opts := metav1.ListOptions{
+		SendInitialEvents:    ptr.To(true),
+		ResourceVersionMatch: metav1.ResourceVersionMatchNotOlderThan,
+		AllowWatchBookmarks:  true,
+		TimeoutSeconds:       &timeoutSeconds,
+	}
+
+	var listRetry wait.Backoff
+	if c.listRetryUseBackoff {
+		listRetry = retry.DefaultBackoff
+	} else {
+		listRetry = retry.DefaultRetry
+	}
+	listRetry.Steps = int(c.listRetryLimit)
+
+	retryable := func(err error) bool {
+		// BadRequest/Invalid means the feature gate is off server-side; no
+		// point retrying, let the caller fall back to LIST immediately.
+		if apierrors.IsBadRequest(err) || apierrors.IsInvalid(err) {
+			return false
+		}
+		return c.listRetryFunc(err)
+	}
+
+	var w watch.Interface
+	var retryCount int64
+	err := retry.OnError(listRetry, retryable, func() error {
+		var ierr error
+		w, ierr = resClient.Watch(ctx, opts)
+		if ierr != nil && c.listRetryLimit > 1 && retryable(ierr) {
+			retryCount++
+			c.log.Info(fmt.Sprintf("Error starting watch-list: %v (try %d/%d)", ierr, retryCount, c.listRetryLimit))
+		}
+		//nolint:wrapcheck // wrap outside the retry
+		return ierr
+	})
+	if err != nil {
+		if apierrors.IsBadRequest(err) || apierrors.IsInvalid(err) {
+			return "", fmt.Errorf("%w: %v", errWatchListUnsupported, err)
+		}
+		return "", fmt.Errorf("failed to start watch-list: %w", err)
+	}
+	defer w.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return "", ctx.Err()
+		case event, ok := <-w.ResultChan():
+			if !ok {
+				return "", errors.New("watch-list stream closed before initial-events-end bookmark")
+			}
+			switch event.Type {
+			case watch.Error:
+				status := apierrors.FromObject(event.Object)
+				if apierrors.IsBadRequest(status) || apierrors.IsInvalid(status) {
+					return "", fmt.Errorf("%w: %v", errWatchListUnsupported, status)
+				}
+				return "", fmt.Errorf("watch-list received error event: %w", status)
+			case watch.Bookmark:
+				obj, ok := event.Object.(*unstructured.Unstructured)
+				if !ok {
+					return "", fmt.Errorf("watch-list bookmark has unexpected type %T", event.Object)
+				}
+				if obj.GetAnnotations()[metav1.InitialEventsAnnotationKey] == "true" {
+					return obj.GetResourceVersion(), nil
+				}
+			case watch.Added, watch.Modified:
+				obj, ok := event.Object.(*unstructured.Unstructured)
+				if !ok {
+					return "", fmt.Errorf("watch-list event object has unexpected type %T", event.Object)
+				}
+				if err := callback(obj); err != nil {
+					return "", err
+				}
+			case watch.Deleted:
+				// Deletion events before the end-of-initial-state bookmark are
+				// not expected; ignore rather than fail so a race between the
+				// snapshot and a concurrent delete does not abort the sync.
+			}
+		}
+	}
+}
+
+// eachResourceWithFallback streams the current state of a resource kind,
+// preferring watch-list (sendInitialEvents) and falling back to a
+// ResourceVersion=0 LIST if the apiserver rejects watch-list. The per-cluster
+// watchListUnsupported flag makes the fallback sticky so clusters without the
+// feature gate take at most one rejected watch round-trip.
+func (c *clusterCache) eachResourceWithFallback(ctx context.Context, resClient dynamic.ResourceInterface, callback func(*unstructured.Unstructured) error) (string, error) {
+	if c.watchListEnabled && !c.watchListUnsupported.Load() {
+		rv, err := c.watchListResources(ctx, resClient, callback)
+		if err == nil {
+			return rv, nil
+		}
+		if !errors.Is(err, errWatchListUnsupported) {
+			return "", err
+		}
+		c.watchListUnsupported.Store(true)
+		c.log.Info("watch-list unsupported by apiserver, falling back to LIST for remaining kinds", "host", c.config.Host, "error", err)
+	}
+	return c.listResources(ctx, resClient, func(listPager *pager.ListPager) error {
+		return listPager.EachListItem(ctx, metav1.ListOptions{}, func(obj runtime.Object) error {
+			un, ok := obj.(*unstructured.Unstructured)
+			if !ok {
+				return fmt.Errorf("watch-list fallback received unexpected object type %T", obj)
+			}
+			return callback(un)
+		})
+	})
+}
+
 // listResources creates list pager and enforces number of concurrent list requests
 // The callback should not wait on any locks that may be held by other callers.
 func (c *clusterCache) listResources(ctx context.Context, resClient dynamic.ResourceInterface, callback func(*pager.ListPager) error) (string, error) {
@@ -763,15 +914,9 @@ func (c *clusterCache) listResources(ctx context.Context, resClient dynamic.Reso
 // loadInitialState loads the state of all the resources retrieved by the given resource client.
 func (c *clusterCache) loadInitialState(ctx context.Context, api kube.APIResourceInfo, resClient dynamic.ResourceInterface, ns string, lock bool) (string, error) {
 	var items []*Resource
-	resourceVersion, err := c.listResources(ctx, resClient, func(listPager *pager.ListPager) error {
-		return listPager.EachListItem(ctx, metav1.ListOptions{}, func(obj runtime.Object) error {
-			if un, ok := obj.(*unstructured.Unstructured); !ok {
-				return fmt.Errorf("object %s/%s has an unexpected type", un.GroupVersionKind().String(), un.GetName())
-			} else {
-				items = append(items, c.newResource(un))
-			}
-			return nil
-		})
+	resourceVersion, err := c.eachResourceWithFallback(ctx, resClient, func(un *unstructured.Unstructured) error {
+		items = append(items, c.newResource(un))
+		return nil
 	})
 	if err != nil {
 		return "", fmt.Errorf("failed to load initial state of resource %s: %w", api.GroupKind.String(), err)
@@ -1073,18 +1218,12 @@ func (c *clusterCache) sync() error {
 		lock.Unlock()
 
 		return c.processApi(client, api, func(resClient dynamic.ResourceInterface, ns string) error {
-			resourceVersion, err := c.listResources(ctx, resClient, func(listPager *pager.ListPager) error {
-				return listPager.EachListItem(context.Background(), metav1.ListOptions{}, func(obj runtime.Object) error {
-					if un, ok := obj.(*unstructured.Unstructured); !ok {
-						return fmt.Errorf("object %s/%s has an unexpected type", un.GroupVersionKind().String(), un.GetName())
-					} else {
-						newRes := c.newResource(un)
-						lock.Lock()
-						c.setNode(newRes)
-						lock.Unlock()
-					}
-					return nil
-				})
+			resourceVersion, err := c.eachResourceWithFallback(ctx, resClient, func(un *unstructured.Unstructured) error {
+				newRes := c.newResource(un)
+				lock.Lock()
+				c.setNode(newRes)
+				lock.Unlock()
+				return nil
 			})
 			if err != nil {
 				if c.isRestrictedResource(err) {

gitops-engine/pkg/cache/settings.go

@@ -87,6 +87,18 @@ func SetListPageSize(listPageSize int64) UpdateSettingsFunc {
 	}
 }
 
+// SetWatchList enables watch-list (sendInitialEvents) for initial resource
+// state sync. When enabled, the cache opens a watch with sendInitialEvents=true
+// instead of issuing a LIST, which streams objects individually and avoids the
+// single oversized response that a watch-cache LIST produces on large clusters.
+// Requires the WatchList feature gate on the apiserver (GA in k8s 1.32).
+// Falls back to LIST automatically if the apiserver rejects the request.
+func SetWatchList(enabled bool) UpdateSettingsFunc {
+	return func(cache *clusterCache) {
+		cache.watchListEnabled = enabled
+	}
+}
+
 // SetListPageBufferSize sets the number of pages to prefetch for list pager.
 func SetListPageBufferSize(listPageBufferSize int32) UpdateSettingsFunc {
 	return func(cache *clusterCache) {