diff --git a/pkg/controlplane/apiserver/server.go b/pkg/controlplane/apiserver/server.go index 8853cf4b79b..b12891fdfe7 100644 --- a/pkg/controlplane/apiserver/server.go +++ b/pkg/controlplane/apiserver/server.go @@ -230,17 +230,25 @@ func (c completedConfig) New(name string, delegationTarget genericapiserver.Dele return nil }) - // RunGVDeletionWorkers processes GVs from deleted CRDs/APIServices. If a GV is no longer in use, - // it is marked for removal from peer-discovery (with a deletion timestamp), triggering a grace period before cleanup. - s.GenericAPIServer.AddPostStartHookOrDie("gv-deletion-workers", func(context genericapiserver.PostStartHookContext) error { - go c.Extra.PeerProxy.RunGVDeletionWorkers(context, 1) + // RunActiveGVTracker monitors CRDs/APIServices and maintains the active GV list for exclusion from peer-discovery. + s.GenericAPIServer.AddPostStartHookOrDie("peer-discovery-gv-active-tracker", func(context genericapiserver.PostStartHookContext) error { + go c.Extra.PeerProxy.RunPeerDiscoveryActiveGVTracker(context, 1) return nil }) - // RunExcludedGVsReaper removes GVs from the peer-discovery exclusion list after their grace period expires. - // This ensures we don't include stale CRDs/aggregated APIs from peer discovery in the aggregated discovery. - s.GenericAPIServer.AddPostStartHookOrDie("excluded-groups-reaper", func(context genericapiserver.PostStartHookContext) error { - go c.Extra.PeerProxy.RunExcludedGVsReaper(context.Done()) + // RunReaper removes expired GVs from the exclusion set for peer-discovery after their grace period. + // This ensures we don't indefinitely exclude GVs that are no longer present. + s.GenericAPIServer.AddPostStartHookOrDie("peer-discovery-gv-exclusion-reaper", func(context genericapiserver.PostStartHookContext) error { + go c.Extra.PeerProxy.RunPeerDiscoveryReaper(context) + return nil + }) + + // RunPeerDiscoveryRefilter re-applies the exclusion filter to the existing peer discovery cache + // whenever the exclusion set changes (e.g., CRD or aggregated API added/deleted). This is different from the + // initial filtering that happens when peer discovery is first fetched - this worker ensures + // the already-cached data stays consistent with the current exclusion set. + s.GenericAPIServer.AddPostStartHookOrDie("peer-discovery-refilter", func(context genericapiserver.PostStartHookContext) error { + go c.Extra.PeerProxy.RunPeerDiscoveryRefilter(context, 1) return nil }) } diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager.go index 40d32d9420c..b1f9e4cac63 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager.go @@ -228,9 +228,9 @@ func (m *GVExclusionManager) onDeleteEvent(gvs []schema.GroupVersion) { m.refilterQueue.Add("refilter") } -// RunActiveGVTracker runs Worker 1: Active GV Tracker +// RunPeerDiscoveryActiveGVTracker runs Worker 1: Active GV Tracker // This worker is triggered by CRD/APIService events and rebuilds the active GV set. -func (m *GVExclusionManager) RunActiveGVTracker(ctx context.Context, workers int) { +func (m *GVExclusionManager) RunPeerDiscoveryActiveGVTracker(ctx context.Context, workers int) { defer m.activeGVQueue.ShutDown() klog.Infof("Starting %d Active GV Tracker worker(s)", workers) @@ -327,9 +327,9 @@ func (m *GVExclusionManager) diffGVs(old, new map[schema.GroupVersion]struct{}) return false } -// RunReaper runs Worker 2: Reaper +// RunPeerDiscoveryReaper runs Worker 2: Reaper // This worker periodically removes expired GVs from recentlyDeletedGVs. -func (m *GVExclusionManager) RunReaper(stopCh <-chan struct{}) { +func (m *GVExclusionManager) RunPeerDiscoveryReaper(ctx context.Context) { klog.Infof("Starting GV Reaper with %s interval", m.reaperCheckInterval) ticker := time.NewTicker(m.reaperCheckInterval) defer ticker.Stop() @@ -338,7 +338,7 @@ func (m *GVExclusionManager) RunReaper(stopCh <-chan struct{}) { select { case <-ticker.C: m.reapExpiredGVs() - case <-stopCh: + case <-ctx.Done(): klog.Info("GV Reaper stopped") return } diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager_test.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager_test.go index cba80306456..a7324b86fe1 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager_test.go @@ -59,13 +59,14 @@ func TestGetExclusionSet(t *testing.T) { {Group: "deprecated", Version: "v1beta1"}: time.Now(), {Group: "legacy", Version: "v1alpha1"}: time.Now(), }, + // Reaper hasnt removed deleted GVs yet. wantGVs: []schema.GroupVersion{ {Group: "deprecated", Version: "v1beta1"}, {Group: "legacy", Version: "v1alpha1"}, }, }, { - name: "active and deleted GVs", + name: "different active and deleted GVs", activeGVs: map[schema.GroupVersion]struct{}{ {Group: "apps", Version: "v1"}: {}, {Group: "batch", Version: "v1"}: {}, @@ -73,12 +74,30 @@ func TestGetExclusionSet(t *testing.T) { deletedGVs: map[schema.GroupVersion]time.Time{ {Group: "old", Version: "v1"}: time.Now(), }, + // Include both active GVs and recently deleted GVs. + // Deleted GVs remain in the exclusion set until the reaper removes them after the grace period. wantGVs: []schema.GroupVersion{ {Group: "apps", Version: "v1"}, {Group: "batch", Version: "v1"}, {Group: "old", Version: "v1"}, }, }, + { + // A GV can appear in both active and deleted sets if: + // 1. CRD was deleted (moved from active to deleted) + // 2. CRD was recreated (added back to active) + // 3. Reaper hasn't cleaned up the deleted entry yet + name: "same GV in both active and deleted", + activeGVs: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + deletedGVs: map[schema.GroupVersion]time.Time{ + {Group: "apps", Version: "v1"}: time.Now(), + }, + wantGVs: []schema.GroupVersion{ + {Group: "apps", Version: "v1"}, + }, + }, } for _, tt := range tests { @@ -284,6 +303,9 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { }, }, { + // Recently deleted GVs are still filtered from peer discovery during the + // grace period (before the reaper cleans them up) to avoid routing requests + // to peers for GVs that were just deleted locally. name: "filter deleted GVs", deletedGVs: map[schema.GroupVersion]time.Time{ {Group: "custom", Version: "v1alpha1"}: time.Now(), @@ -307,6 +329,8 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { }, }, { + // Both active GVs and recently deleted GVs (within grace period, not yet + // cleaned up by the reaper) are filtered from peer discovery. name: "filter both active and deleted GVs", activeGVs: map[schema.GroupVersion]struct{}{ {Group: "apps", Version: "v1"}: {}, diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go index e89eea6ad35..b6dce3b1ac0 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go @@ -49,7 +49,7 @@ const ( // available on this server. localDiscoveryRefreshInterval = 30 * time.Minute // defaultExclusionGracePeriod is the default duration to wait before - // removing a group from the exclusion set after it is deleted from + // removing a groupversion from the exclusion set after it is deleted from // CRDs and aggregated APIs. // This is to allow time for all peer API servers to also observe // the deleted CRD or aggregated API before this server stops excluding it @@ -67,12 +67,55 @@ type Interface interface { HasFinishedSync() bool RunLocalDiscoveryCacheSync(stopCh <-chan struct{}) error RunPeerDiscoveryCacheSync(ctx context.Context, workers int) - RunExcludedGVsReaper(stopCh <-chan struct{}) - RunGVDeletionWorkers(ctx context.Context, workers int) GetPeerResources() map[string][]apidiscoveryv2.APIGroupDiscovery RegisterCacheInvalidationCallback(cb func()) + + // RegisterCRDInformerHandlers registers event handlers on the CRD informer to track + // which GroupVersions are served locally by CRDs. When a CRD is created or updated, + // its GV is added to the exclusion set. When deleted, the GV is marked for exclusion + // during a grace period to allow peers to observe the deletion. The extractor function + // extracts the GroupVersion from a CRD object. + // + // This exclusion is necessary because peer discovery is not refreshed when a local + // CRD is deleted. Without exclusion, the deleted GV might still appear in cached peer + // discovery data, causing requests to be incorrectly routed to a peer for a GV that + // no longer exists locally. Therefore, we intentionally exclude CRD GVs from peer + // discovery from the start and only rely on the local apiserver's view of the CRD + // to serve it in peer-aggregated discovery. RegisterCRDInformerHandlers(crdInformer cache.SharedIndexInformer, extractor GVExtractor) error + + // RegisterAPIServiceInformerHandlers registers event handlers on the APIService informer + // to track which GroupVersions are served locally by aggregated APIServices. When an + // APIService is created or updated, its GV is added to the exclusion set. When deleted, + // the GV is marked for exclusion during a grace period. + // + // This exclusion is necessary because peer discovery is not refreshed when a local + // aggregated APIService is deleted. Without exclusion, the deleted GV might still appear + // in cached peer discovery data, causing requests to be incorrectly routed to a peer. + // Therefore, we intentionally exclude aggregated APIService GVs from peer discovery + // from the start and only rely on the local apiserver's view to serve them in + // peer-aggregated discovery. RegisterAPIServiceInformerHandlers(apiServiceInformer cache.SharedIndexInformer, extractor GVExtractor) error + + // RunPeerDiscoveryActiveGVTracker starts a worker that processes CRD/APIService informer + // events to rebuild the set of actively served GroupVersions. This worker is triggered + // whenever a CRD or APIService is added or updated and updates the exclusion + // set accordingly. + RunPeerDiscoveryActiveGVTracker(ctx context.Context, workers int) + + // RunPeerDiscoveryReaper starts a background worker that periodically removes expired + // GroupVersions from the exclusion set. When a CRD/APIService is deleted, its GV remains + // in the exclusion set for a grace period (default 5 minutes) to allow all peer API servers + // to observe the deletion. The reaper runs at a configured interval (default 1 minute) + // and removes GVs whose grace period has elapsed. + RunPeerDiscoveryReaper(ctx context.Context) + + // RunPeerDiscoveryRefilter starts a worker that re-applies exclusion filtering to the + // cached peer discovery data whenever the exclusion set changes. This ensures that + // already-cached peer discovery responses are immediately updated to exclude newly added + // or updated local GVs, rather than waiting for the next peer lease event to trigger a + // cache refresh of peer discovery data. + RunPeerDiscoveryRefilter(ctx context.Context, workers int) } // New creates a new instance to implement unknown version proxy @@ -102,16 +145,15 @@ func NewPeerProxyHandler( Name: peerDiscoveryControllerName, }), apiserverIdentityInformer: leaseInformer, - excludedGVs: make(map[schema.GroupVersion]*time.Time), - exclusionGracePeriod: defaultExclusionGracePeriod, - reaperCheckInterval: defaultExclusionReaperInterval, - gvDeletionQueue: workqueue.NewTypedRateLimitingQueueWithConfig( - workqueue.DefaultTypedControllerRateLimiter[string](), - workqueue.TypedRateLimitingQueueConfig[string]{ - Name: "gv-deletion", - }), } + h.gvExclusionManager = NewGVExclusionManager( + defaultExclusionGracePeriod, + defaultExclusionReaperInterval, + &h.peerDiscoveryInfoCache, + &h.cacheInvalidationCallback, + ) + if parts := strings.Split(identityLeaseLabelSelector, "="); len(parts) != 2 { return nil, fmt.Errorf("invalid identityLeaseLabelSelector provided, must be of the form key=value, received: %v", identityLeaseLabelSelector) } @@ -169,3 +211,40 @@ func NewPeerProxyHandler( h.leaseRegistration = peerDiscoveryRegistration return h, nil } + +// RegisterCRDInformerHandlers registers event handlers for CRD informer. +func (h *peerProxyHandler) RegisterCRDInformerHandlers(crdInformer cache.SharedIndexInformer, extractor GVExtractor) error { + if h.gvExclusionManager != nil { + return h.gvExclusionManager.RegisterCRDInformerHandlers(crdInformer, extractor) + } + return nil +} + +// RegisterAPIServiceInformerHandlers registers event handlers for APIService informer. +func (h *peerProxyHandler) RegisterAPIServiceInformerHandlers(apiServiceInformer cache.SharedIndexInformer, extractor GVExtractor) error { + if h.gvExclusionManager != nil { + return h.gvExclusionManager.RegisterAPIServiceInformerHandlers(apiServiceInformer, extractor) + } + return nil +} + +// RunPeerDiscoveryActiveGVTracker starts the worker that tracks active GVs from CRDs/APIServices. +func (h *peerProxyHandler) RunPeerDiscoveryActiveGVTracker(ctx context.Context, workers int) { + if h.gvExclusionManager != nil { + h.gvExclusionManager.RunPeerDiscoveryActiveGVTracker(ctx, workers) + } +} + +// RunPeerDiscoveryReaper starts the worker that removes expired GVs from the exclusion set. +func (h *peerProxyHandler) RunPeerDiscoveryReaper(ctx context.Context) { + if h.gvExclusionManager != nil { + h.gvExclusionManager.RunPeerDiscoveryReaper(ctx) + } +} + +// RunPeerDiscoveryRefilter starts the worker that refilters peer discovery cache. +func (h *peerProxyHandler) RunPeerDiscoveryRefilter(ctx context.Context, workers int) { + if h.gvExclusionManager != nil { + h.gvExclusionManager.RunPeerDiscoveryRefilter(ctx, workers) + } +} diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler.go index aed2909ca9a..87b86de39e3 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler.go @@ -90,25 +90,8 @@ type peerProxyHandler struct { peerLeaseQueue workqueue.TypedRateLimitingInterface[string] serializer runtime.NegotiatedSerializer cacheInvalidationCallback atomic.Pointer[func()] - // Exclusion set for groups that should not be included in peer proxying - // or peer-aggregated discovery (e.g., CRDs/APIServices) - // - // This map has three states for a group: - // - Not in map: Group is not excluded. - // - In map, value is nil: Group is actively excluded. - // - In map, value is non-nil: Group is pending deletion (grace period). - excludedGVs map[schema.GroupVersion]*time.Time - excludedGVsMu sync.RWMutex - crdInformer cache.SharedIndexInformer - crdExtractor GVExtractor - apiServiceInformer cache.SharedIndexInformer - apiServiceExtractor GVExtractor - - exclusionGracePeriod time.Duration - reaperCheckInterval time.Duration - - // Worker queue for processing GV deletions asynchronously - gvDeletionQueue workqueue.TypedRateLimitingInterface[string] + // Manager for GV exclusions (CRDs/APIServices) + gvExclusionManager *GVExclusionManager } // PeerDiscoveryCacheEntry holds the GVRs and group-level discovery info for a peer. @@ -141,12 +124,10 @@ func (h *peerProxyHandler) WaitForCacheSync(stopCh <-chan struct{}) error { return fmt.Errorf("error while waiting for peer-identity-lease event handler registration sync") } - if h.crdInformer != nil && !cache.WaitForNamedCacheSync("peer-discovery-crd-informer", stopCh, h.crdInformer.HasSynced) { - return fmt.Errorf("error while waiting for crd informer sync") - } - - if h.apiServiceInformer != nil && !cache.WaitForNamedCacheSync("peer-discovery-api-service-informer", stopCh, h.apiServiceInformer.HasSynced) { - return fmt.Errorf("error while waiting for apiservice informer sync") + if h.gvExclusionManager != nil { + if !h.gvExclusionManager.WaitForCacheSync(stopCh) { + return fmt.Errorf("error while waiting for gv exclusion manager cache sync") + } } // Wait for localDiscoveryInfoCache to be populated.