fix: Watch APIServices like we do CRDs (#28289)

Signed-off-by: Zach Aller <[email protected]>

gitops-engine/pkg/cache/cluster.go

@@ -902,26 +902,113 @@ func (c *clusterCache) watchEvents(ctx context.Context, api kube.APIResourceInfo
 							c.log.Error(err, "Failed to start missing watch")
 						}
 					}
-					err = runSynced(&c.lock, func() error {
-						openAPISchema, gvkParser, err := c.kubectl.LoadOpenAPISchema(c.config)
-						if err != nil {
-							return fmt.Errorf("failed to load open api schema while handling CRD change: %w", err)
-						}
-						if gvkParser != nil {
-							c.gvkParser = gvkParser
-						}
-						c.openAPISchema = openAPISchema
-						return nil
-					})
-					if err != nil {
+					if err := c.reloadOpenAPISchema(); err != nil {
 						c.log.Error(err, "Failed to reload open api schema")
 					}
+				} else if kube.IsAPIService(obj) {
+					// When an aggregated API's extension apiserver becomes available, the
+					// kube-apiserver starts serving new group/kinds that were not present
+					// during the initial discovery (e.g. the extension apiserver was down
+					// when Argo CD started). Re-run discovery so those resources get
+					// watched, mirroring how CRD events are handled above. Otherwise the
+					// new kinds remain invisible until the next manual cache invalidation
+					// or full resync.
+					deleted := event.Type == watch.Deleted
+					if deleted || isAPIServiceAvailable(obj) {
+						// The kube-apiserver's aggregated discovery can lag behind an
+						// APIService reporting Available, so the group may not be served yet
+						// when we first re-run discovery. Reconcile watches with a bounded
+						// retry (in a goroutine to avoid blocking this watch) until the
+						// APIService's own group is served or we exhaust our attempts.
+						group, _, _ := unstructured.NestedString(obj.Object, "spec", "group")
+						c.log.Info("Reconciling Kubernetes APIs, watches, and Open API schemas due to APIService event", "eventType", event.Type, "name", obj.GetName(), "group", group)
+						go c.reconcileAPIServiceWatches(group, !deleted)
+					}
 				}
 			}
 		}
 	})
 }
 
+// reloadOpenAPISchema reloads the cluster's OpenAPI schema and GVK parser. It is
+// called after the set of served APIs changes (e.g. due to a CRD or APIService
+// event) so that field management and schema-aware operations stay in sync with
+// the cluster.
+func (c *clusterCache) reloadOpenAPISchema() error {
+	return runSynced(&c.lock, func() error {
+		openAPISchema, gvkParser, err := c.kubectl.LoadOpenAPISchema(c.config)
+		if err != nil {
+			return fmt.Errorf("failed to load open api schema: %w", err)
+		}
+		if gvkParser != nil {
+			c.gvkParser = gvkParser
+		}
+		c.openAPISchema = openAPISchema
+		return nil
+	})
+}
+
+// reconcileAPIServiceWatches re-runs discovery and starts any missing watches in
+// response to an APIService event. Aggregated discovery can lag behind an
+// APIService reporting Available, so when waitForGroup is true (Added/Modified
+// while Available) it retries with exponential backoff (9 attempts, intervals
+// growing 500ms -> ~8.5s for a ~25s total budget) until the APIService's group
+// is served by the cluster (i.e. a watch for it has been started) or the attempts
+// are exhausted. For deletions it reconciles once.
+func (c *clusterCache) reconcileAPIServiceWatches(group string, waitForGroup bool) {
+	err := wait.ExponentialBackoff(wait.Backoff{
+		Duration: 500 * time.Millisecond,
+		Factor:   1.5,
+		Steps:    9,
+	}, func() (bool, error) {
+		if err := runSynced(&c.lock, func() error {
+			return c.startMissingWatches()
+		}); err != nil {
+			c.log.Error(err, "Failed to start missing watches after APIService event")
+		}
+		if err := c.reloadOpenAPISchema(); err != nil {
+			c.log.Error(err, "Failed to reload open api schema after APIService event")
+		}
+		return !waitForGroup || group == "" || c.isGroupWatched(group), nil
+	})
+	if err != nil {
+		c.log.Info("Aggregated API group is not served by discovery yet after retries; it will be picked up on the next resync", "group", group)
+	}
+}
+
+// isGroupWatched reports whether the cache has started watching any resource in
+// the given API group.
+func (c *clusterCache) isGroupWatched(group string) bool {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	for gk := range c.apisMeta {
+		if gk.Group == group {
+			return true
+		}
+	}
+	return false
+}
+
+// isAPIServiceAvailable reports whether the given APIService object has an
+// Available condition set to True, indicating its backing (aggregated) apiserver
+// is ready to serve its API group.
+func isAPIServiceAvailable(obj *unstructured.Unstructured) bool {
+	conditions, found, err := unstructured.NestedSlice(obj.Object, "status", "conditions")
+	if err != nil || !found {
+		return false
+	}
+	for _, condition := range conditions {
+		cond, ok := condition.(map[string]any)
+		if !ok {
+			continue
+		}
+		if cond["type"] == "Available" {
+			return cond["status"] == "True"
+		}
+	}
+	return false
+}
+
 // processApi processes all the resources for a given API. First we construct an API client for the given API. Then we
 // call the callback. If we're managing the whole cluster, we call the callback with the client and an empty namespace.
 // If we're managing specific namespaces, we call the callback for each namespace.

gitops-engine/pkg/cache/cluster_test.go

@@ -1092,6 +1092,68 @@ func TestAppendAPIResource(t *testing.T) {
 	assert.ElementsMatch(t, []kube.APIResourceInfo{resourceInfo}, cluster.apiResources)
 }
 
+func TestIsAPIServiceAvailable(t *testing.T) {
+	t.Parallel()
+
+	apiService := func(conditions []any) *unstructured.Unstructured {
+		obj := &unstructured.Unstructured{Object: map[string]any{
+			"apiVersion": "apiregistration.k8s.io/v1",
+			"kind":       "APIService",
+			"metadata":   map[string]any{"name": "v1beta1.metrics.k8s.io"},
+			"spec":       map[string]any{"group": "metrics.k8s.io", "version": "v1beta1"},
+		}}
+		if conditions != nil {
+			obj.Object["status"] = map[string]any{"conditions": conditions}
+		}
+		return obj
+	}
+
+	testCases := []struct {
+		name     string
+		obj      *unstructured.Unstructured
+		expected bool
+	}{
+		{
+			name:     "no status",
+			obj:      apiService(nil),
+			expected: false,
+		},
+		{
+			name:     "no conditions",
+			obj:      apiService([]any{}),
+			expected: false,
+		},
+		{
+			name: "available true",
+			obj: apiService([]any{
+				map[string]any{"type": "Available", "status": "True"},
+			}),
+			expected: true,
+		},
+		{
+			name: "available false",
+			obj: apiService([]any{
+				map[string]any{"type": "Available", "status": "False"},
+			}),
+			expected: false,
+		},
+		{
+			name: "unrelated condition only",
+			obj: apiService([]any{
+				map[string]any{"type": "SomethingElse", "status": "True"},
+			}),
+			expected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.expected, isAPIServiceAvailable(tc.obj))
+		})
+	}
+}
+
 func ExampleNewClusterCache_resourceUpdatedEvents() {
 	// kubernetes cluster config here
 	config := &rest.Config{}

gitops-engine/pkg/utils/kube/kube.go

@@ -185,6 +185,14 @@ func IsCRD(obj *unstructured.Unstructured) bool {
 	return IsCRDGroupVersionKind(obj.GroupVersionKind())
 }
 
+func IsAPIServiceGroupVersionKind(gvk schema.GroupVersionKind) bool {
+	return gvk.Kind == APIServiceKind && gvk.Group == "apiregistration.k8s.io"
+}
+
+func IsAPIService(obj *unstructured.Unstructured) bool {
+	return IsAPIServiceGroupVersionKind(obj.GroupVersionKind())
+}
+
 // ServerResourceForGroupVersionKind looks up and returns the API resource from
 // the server for a given GVK scheme. If verb is set to the non-empty string,
 // it will return the API resource which supports the verb. There are some edge

gitops-engine/pkg/utils/kube/kube_test.go

@@ -447,3 +447,44 @@ func TestServerResourceGroupForGroupVersionKind(t *testing.T) {
 		}
 	})
 }
+
+func TestIsAPIService(t *testing.T) {
+	t.Parallel()
+	testCases := []struct {
+		name     string
+		obj      *unstructured.Unstructured
+		expected bool
+	}{
+		{
+			name: "is an APIService",
+			obj: &unstructured.Unstructured{Object: map[string]any{
+				"apiVersion": "apiregistration.k8s.io/v1",
+				"kind":       "APIService",
+			}},
+			expected: true,
+		},
+		{
+			name: "wrong group",
+			obj: &unstructured.Unstructured{Object: map[string]any{
+				"apiVersion": "example.com/v1",
+				"kind":       "APIService",
+			}},
+			expected: false,
+		},
+		{
+			name: "wrong kind",
+			obj: &unstructured.Unstructured{Object: map[string]any{
+				"apiVersion": "apiregistration.k8s.io/v1",
+				"kind":       "Service",
+			}},
+			expected: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			assert.Equal(t, tc.expected, IsAPIService(tc.obj))
+		})
+	}
+}

test/e2e/aggregated_apiservice_test.go

@@ -0,0 +1,122 @@
+package e2e
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	. "github.com/argoproj/argo-cd/gitops-engine/pkg/sync/common"
+
+	applicationpkg "github.com/argoproj/argo-cd/v3/pkg/apiclient/application"
+	. "github.com/argoproj/argo-cd/v3/pkg/apis/application/v1alpha1"
+	. "github.com/argoproj/argo-cd/v3/test/e2e/fixture"
+	. "github.com/argoproj/argo-cd/v3/test/e2e/fixture/app"
+	"github.com/argoproj/argo-cd/v3/util/errors"
+	utilio "github.com/argoproj/argo-cd/v3/util/io"
+)
+
+const wardleAPIServiceName = "v1alpha1.wardle.example.com"
+
+// TestAPIServiceLateRegistrationIsDiscovered verifies that when an aggregated
+// (extension) API server is registered AFTER Argo CD is already running, the
+// destination cluster cache picks up the newly served group/kind WITHOUT a manual
+// cache invalidation or hard refresh.
+//
+// The scenario reproduces the original bug: starting Argo CD without the
+// aggregated apiserver running means its group/kind is absent from discovery, so
+// the cluster cache never watches it. Argo CD watches APIService objects and, on
+// the fix, reacts to an APIService becoming Available by re-running discovery and
+// starting the missing watches - mirroring the existing CRD handling.
+//
+// The assertion deliberately checks the application RESOURCE TREE (which is built
+// purely from resources the cluster cache is watching) rather than the resource
+// sync status. The sync status is not a reliable signal here because
+// GetManagedLiveObjs falls back to a direct live API GET for kinds that are not
+// watched, so a resource can show as Synced even though the cache never observed
+// it - which is exactly why it would be missing from the UI/tree.
+func TestAPIServiceLateRegistrationIsDiscovered(t *testing.T) {
+	// The aggregated apiserver infrastructure is cluster-scoped (Namespace,
+	// APIService, ClusterRole(Binding)s) and is not managed by an Argo CD app. It
+	// is labeled with e2e.argoproj.io=true so fixture.EnsureCleanState tears it
+	// down between tests; a dangling APIService backed by a deleted service would
+	// otherwise degrade discovery for subsequent tests.
+	const serverManifests = "testdata/aggregated-apiserver-server/manifests.yaml"
+
+	flunderManifest := `apiVersion: wardle.example.com/v1alpha1
+kind: Flunder
+metadata:
+  name: e2e-flunder
+`
+
+	ctx := Given(t)
+	ctx.
+		Path("aggregated-apiserver").
+		When().
+		CreateApp().
+		Sync().
+		Then().
+		// Baseline: the app - and therefore the destination cluster cache - is
+		// synced and running before the aggregated API exists.
+		Expect(OperationPhaseIs(OperationSucceeded)).
+		Expect(SyncStatusIs(SyncStatusCodeSynced)).
+		Expect(ResourceSyncStatusIs("ConfigMap", "aggregated-apiserver-canary", SyncStatusCodeSynced)).
+		When().
+		And(func() {
+			// Register the extension apiserver AFTER Argo CD is up and wait until it
+			// is actually serving its API group.
+			errors.NewHandler(t).FailOnErr(Run("", "kubectl", "apply", "-f", serverManifests))
+			errors.NewHandler(t).FailOnErr(Run("", "kubectl", "-n", "wardle", "rollout", "status", "deployment/wardle-server", "--timeout=180s"))
+			errors.NewHandler(t).FailOnErr(Run("", "kubectl", "wait", "--for=condition=Available", "apiservice/"+wardleAPIServiceName, "--timeout=180s"))
+		}).
+		// Deliberately NO InvalidateCache and NO RefreshTypeHard here - the cache must
+		// discover the new kind on its own in response to the APIService event.
+		AddFile("flunder.yaml", flunderManifest).
+		Refresh(RefreshTypeNormal).
+		Sync().
+		Then().
+		Expect(OperationPhaseIs(OperationSucceeded)).
+		Expect(SyncStatusIs(SyncStatusCodeSynced)).
+		And(func(app *Application) {
+			// The Flunder appears in the resource tree only if the cluster cache
+			// discovered the aggregated kind and is watching it. This is the behavior
+			// under test: without reacting to the APIService event, the cache never
+			// watches wardle.example.com and the Flunder is absent from the tree (and
+			// the UI) until a manual invalidation.
+			closer, cdClient := ArgoCDClientset.NewApplicationClientOrDie()
+			defer utilio.Close(closer)
+
+			require.Eventually(t, func() bool {
+				// Trigger a normal (soft) refresh so the controller recomputes the
+				// resource tree from the live cluster cache. This does NOT invalidate
+				// the cache or re-run discovery: without the APIService fix the kind
+				// remains unwatched, so the Flunder never enters the tree regardless of
+				// how many times we refresh.
+				refresh := string(RefreshTypeNormal)
+				if _, err := cdClient.Get(context.Background(), &applicationpkg.ApplicationQuery{
+					Name:         &app.Name,
+					AppNamespace: &app.Namespace,
+					Project:      []string{app.Spec.Project},
+					Refresh:      &refresh,
+				}); err != nil {
+					t.Logf("app refresh failed: %v", err)
+				}
+				tree, err := cdClient.ResourceTree(context.Background(), &applicationpkg.ResourcesQuery{
+					ApplicationName: &app.Name,
+					AppNamespace:    &app.Namespace,
+					Project:         &app.Spec.Project,
+				})
+				if err != nil {
+					t.Logf("failed to get resource tree: %v", err)
+					return false
+				}
+				for _, node := range tree.Nodes {
+					if node.Group == "wardle.example.com" && node.Kind == "Flunder" && node.Name == "e2e-flunder" {
+						return true
+					}
+				}
+				return false
+			}, 90*time.Second, 3*time.Second, "Flunder should appear in the application resource tree without a cache invalidation")
+		})
+}

test/e2e/fixture/fixture.go

@@ -792,6 +792,12 @@ func EnsureCleanState(t *testing.T, opts ...TestOption) *TestState {
 			_, err := Run("", "kubectl", "delete", "crd", "-l", TestingLabel+"=true", "--wait=false")
 			return err
 		},
+		func() error {
+			// delete old aggregated APIServices created by tests. A dangling APIService
+			// backed by a deleted extension apiserver degrades cluster discovery.
+			_, err := Run("", "kubectl", "delete", "apiservice", "-l", TestingLabel+"=true", "--wait=false")
+			return err
+		},
 		func() error {
 			err := updateSettingConfigMap(func(cm *corev1.ConfigMap) error {
 				cm.Data = map[string]string{}

test/e2e/testdata/aggregated-apiserver-server/manifests.yaml

@@ -0,0 +1,150 @@
+# A minimal aggregated (extension) API server based on the Kubernetes
+# "sample-apiserver" (wardle). It registers the wardle.example.com/v1alpha1 group
+# (Flunder/Fischer kinds) via an APIService. These manifests are intentionally
+# NOT part of any Argo CD application path; the e2e test applies them directly
+# with kubectl AFTER Argo CD is already running, to verify that the cluster cache
+# picks up the newly served kinds without a manual cache invalidation.
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: wardle
+  # e2e.argoproj.io lets fixture EnsureCleanState catch leaks if t.Cleanup fails.
+  labels:
+    e2e.argoproj.io: "true"
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: apiserver
+  namespace: wardle
+---
+# Allows the aggregated apiserver to delegate authn/authz to the kube-apiserver.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: wardle:system:auth-delegator
+  labels:
+    e2e.argoproj.io: "true"
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: system:auth-delegator
+subjects:
+- kind: ServiceAccount
+  name: apiserver
+  namespace: wardle
+---
+# Allows the aggregated apiserver to read the extension-apiserver-authentication
+# configmap (front-proxy CA / requestheader config) from kube-system.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: wardle-auth-reader
+  namespace: kube-system
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: extension-apiserver-authentication-reader
+subjects:
+- kind: ServiceAccount
+  name: apiserver
+  namespace: wardle
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: wardle-aggregated-apiserver-clusterrole
+  labels:
+    e2e.argoproj.io: "true"
+rules:
+- apiGroups: [""]
+  resources: ["namespaces"]
+  verbs: ["get", "watch", "list"]
+- apiGroups: ["admissionregistration.k8s.io"]
+  resources: ["mutatingwebhookconfigurations", "validatingwebhookconfigurations"]
+  verbs: ["get", "watch", "list"]
+- apiGroups: ["flowcontrol.apiserver.k8s.io"]
+  resources: ["prioritylevelconfigurations", "flowschemas"]
+  verbs: ["get", "watch", "list"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: wardle-aggregated-apiserver-clusterrolebinding
+  labels:
+    e2e.argoproj.io: "true"
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: wardle-aggregated-apiserver-clusterrole
+subjects:
+- kind: ServiceAccount
+  name: apiserver
+  namespace: wardle
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: wardle-server
+  namespace: wardle
+  labels:
+    apiserver: "true"
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      apiserver: "true"
+  template:
+    metadata:
+      labels:
+        apiserver: "true"
+    spec:
+      serviceAccountName: apiserver
+      containers:
+      - name: wardle-server
+        image: registry.k8s.io/e2e-test-images/sample-apiserver:1.29.2
+        imagePullPolicy: IfNotPresent
+        args:
+        - --etcd-servers=http://127.0.0.1:2379
+        - --secure-port=8443
+        ports:
+        - containerPort: 8443
+      - name: etcd
+        image: registry.k8s.io/etcd:3.5.16-0
+        imagePullPolicy: IfNotPresent
+        command: ["/usr/local/bin/etcd"]
+        args:
+        - --data-dir=/tmp/etcd
+        - --listen-client-urls=http://127.0.0.1:2379
+        - --advertise-client-urls=http://127.0.0.1:2379
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: api
+  namespace: wardle
+spec:
+  ports:
+  - port: 443
+    protocol: TCP
+    targetPort: 8443
+  selector:
+    apiserver: "true"
+---
+apiVersion: apiregistration.k8s.io/v1
+kind: APIService
+metadata:
+  name: v1alpha1.wardle.example.com
+  # e2e.argoproj.io lets fixture EnsureCleanState clean up this cluster-scoped
+  # APIService between tests; a dangling APIService degrades discovery.
+  labels:
+    e2e.argoproj.io: "true"
+spec:
+  insecureSkipTLSVerify: true
+  group: wardle.example.com
+  groupPriorityMinimum: 1000
+  versionPriority: 15
+  service:
+    name: api
+    namespace: wardle
+  version: v1alpha1

test/e2e/testdata/aggregated-apiserver/configmap.yaml

@@ -0,0 +1,6 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: aggregated-apiserver-canary
+data:
+  purpose: "ensures the destination cluster cache is synced before the APIService is registered"