From 438649a93410cfc5342c9862abdd62a7e1574e98 Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Wed, 17 Dec 2025 17:19:49 -0800 Subject: [PATCH 1/8] Rename exclusion_filter.go to gv_exclusion_manager.go --- .../peerproxy/{exclusion_filter.go => gv_exclusion_manager.go} | 0 .../{exclusion_filter_test.go => gv_exclusion_manager_test.go} | 0 2 files changed, 0 insertions(+), 0 deletions(-) rename staging/src/k8s.io/apiserver/pkg/util/peerproxy/{exclusion_filter.go => gv_exclusion_manager.go} (100%) rename staging/src/k8s.io/apiserver/pkg/util/peerproxy/{exclusion_filter_test.go => gv_exclusion_manager_test.go} (100%) diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/exclusion_filter.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager.go similarity index 100% rename from staging/src/k8s.io/apiserver/pkg/util/peerproxy/exclusion_filter.go rename to staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager.go diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/exclusion_filter_test.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager_test.go similarity index 100% rename from staging/src/k8s.io/apiserver/pkg/util/peerproxy/exclusion_filter_test.go rename to staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager_test.go From 41761eb7242377fd838858ec494964027dc1f927 Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Wed, 17 Dec 2025 17:20:14 -0800 Subject: [PATCH 2/8] Add GVExclusionManager to process CRD and aggregated APIs GVs --- .../util/peerproxy/gv_exclusion_manager.go | 610 ++++++++++++------ .../pkg/util/peerproxy/peer_discovery.go | 8 +- 2 files changed, 399 insertions(+), 219 deletions(-) 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 a808d06157a..40d32d9420c 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 @@ -18,10 +18,13 @@ package peerproxy import ( "context" + "sync/atomic" "time" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "k8s.io/klog/v2" apidiscoveryv2 "k8s.io/api/apidiscovery/v2" @@ -33,25 +36,80 @@ import ( // in peer-aggregated discovery that were deleted but still appear in a peer's discovery. type GVExtractor func(obj interface{}) []schema.GroupVersion +// GVExclusionManager manages the exclusion of group-versions from peer discovery. +// It maintains two atomic maps: +// - currentlyActiveGVs: All GVs currently served by CRDs and aggregated APIServices +// - recentlyDeletedGVs: GVs belonging to CRDs and aggregated APIServices that were recently deleted, +// tracked with deletion timestamp for grace period +// +// It runs three workers: +// 1. Active GV Tracker: Triggered on CRD/APIService events, rebuilds active GVs +// 2. Reaper: Periodically removes expired GVs from recentlyDeletedGVs +// 3. Peer Discovery Re-filter: Rate-limited worker that filters peer cache +type GVExclusionManager struct { + // Atomic maps for lock-free access + currentlyActiveGVs atomic.Value // map[schema.GroupVersion]struct{} + recentlyDeletedGVs atomic.Value // map[schema.GroupVersion]time.Time + + // Informers for fetching active GVs + crdInformer cache.SharedIndexInformer + crdExtractor GVExtractor + apiServiceInformer cache.SharedIndexInformer + apiServiceExtractor GVExtractor + + // Worker 1: triggered by CRD/APIService events + activeGVQueue workqueue.TypedRateLimitingInterface[string] + // Worker 2: periodic reaper's configuration + exclusionGracePeriod time.Duration + reaperCheckInterval time.Duration + // Worker 3: triggered by Active/Deleted GV changes + refilterQueue workqueue.TypedRateLimitingInterface[string] + + peerDiscoveryCache *atomic.Value // peerProxyHandler.peerDiscoveryInfoCache + invalidationCallback *atomic.Pointer[func()] +} + +// NewGVExclusionManager creates a new GV exclusion manager. +func NewGVExclusionManager( + exclusionGracePeriod time.Duration, + reaperCheckInterval time.Duration, + peerDiscoveryCache *atomic.Value, + invalidationCallback *atomic.Pointer[func()], +) *GVExclusionManager { + mgr := &GVExclusionManager{ + exclusionGracePeriod: exclusionGracePeriod, + reaperCheckInterval: reaperCheckInterval, + peerDiscoveryCache: peerDiscoveryCache, + invalidationCallback: invalidationCallback, + activeGVQueue: workqueue.NewTypedRateLimitingQueueWithConfig( + workqueue.DefaultTypedControllerRateLimiter[string](), + workqueue.TypedRateLimitingQueueConfig[string]{ + Name: "active-gv-tracker", + }), + refilterQueue: workqueue.NewTypedRateLimitingQueueWithConfig( + workqueue.DefaultTypedControllerRateLimiter[string](), + workqueue.TypedRateLimitingQueueConfig[string]{ + Name: "peer-discovery-refilter", + }), + } + + mgr.currentlyActiveGVs.Store(map[schema.GroupVersion]struct{}{}) + mgr.recentlyDeletedGVs.Store(map[schema.GroupVersion]time.Time{}) + + return mgr +} + // RegisterCRDInformerHandlers registers event handlers for CRD informer using a custom extractor. // The extractor function is responsible for extracting GroupVersions from CRD objects. -func (h *peerProxyHandler) RegisterCRDInformerHandlers(crdInformer cache.SharedIndexInformer, extractor GVExtractor) error { - h.crdInformer = crdInformer - h.crdExtractor = extractor - _, err := h.crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ +func (m *GVExclusionManager) RegisterCRDInformerHandlers(crdInformer cache.SharedIndexInformer, extractor GVExtractor) error { + m.crdInformer = crdInformer + m.crdExtractor = extractor + _, err := m.crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { - gvs := extractor(obj) - if gvs == nil { - return - } - h.addExcludedGVs(gvs) + m.handleGVUpdate() }, UpdateFunc: func(oldObj, newObj interface{}) { - gvs := extractor(newObj) - if gvs == nil { - return - } - h.addExcludedGVs(gvs) + m.handleGVUpdate() }, DeleteFunc: func(obj interface{}) { // Handle tombstone objects @@ -62,9 +120,7 @@ func (h *peerProxyHandler) RegisterCRDInformerHandlers(crdInformer cache.SharedI if gvs == nil { return } - for _, gv := range gvs { - h.gvDeletionQueue.Add(gv.String()) - } + m.onDeleteEvent(gvs) }, }) return err @@ -73,23 +129,15 @@ func (h *peerProxyHandler) RegisterCRDInformerHandlers(crdInformer cache.SharedI // RegisterAPIServiceInformerHandlers registers event handlers for APIService informer using a custom extractor. // The extractor function is responsible for extracting GroupVersions from APIService objects // and determining if they represent aggregated APIs. -func (h *peerProxyHandler) RegisterAPIServiceInformerHandlers(apiServiceInformer cache.SharedIndexInformer, extractor GVExtractor) error { - h.apiServiceInformer = apiServiceInformer - h.apiServiceExtractor = extractor - _, err := h.apiServiceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ +func (m *GVExclusionManager) RegisterAPIServiceInformerHandlers(apiServiceInformer cache.SharedIndexInformer, extractor GVExtractor) error { + m.apiServiceInformer = apiServiceInformer + m.apiServiceExtractor = extractor + _, err := m.apiServiceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { - gvs := extractor(obj) - if gvs == nil { - return - } - h.addExcludedGVs(gvs) + m.handleGVUpdate() }, UpdateFunc: func(oldObj, newObj interface{}) { - gvs := extractor(newObj) - if gvs == nil { - return - } - h.addExcludedGVs(gvs) + m.handleGVUpdate() }, DeleteFunc: func(obj interface{}) { // Handle tombstone objects @@ -101,30 +149,272 @@ func (h *peerProxyHandler) RegisterAPIServiceInformerHandlers(apiServiceInformer if gvs == nil { return } - for _, gv := range gvs { - h.gvDeletionQueue.Add(gv.String()) - } + m.onDeleteEvent(gvs) }, }) return err } -// addExcludedGVs adds group-versions to the exclusion set, filters them from the peer discovery cache, -// and triggers invalidation callbacks if the cache changed. -func (h *peerProxyHandler) addExcludedGVs(gvs []schema.GroupVersion) { +// WaitForCacheSync waits for the informer caches to sync. +func (m *GVExclusionManager) WaitForCacheSync(stopCh <-chan struct{}) bool { + synced := []cache.InformerSynced{} + if m.crdInformer != nil { + synced = append(synced, m.crdInformer.HasSynced) + } + if m.apiServiceInformer != nil { + synced = append(synced, m.apiServiceInformer.HasSynced) + } + if len(synced) == 0 { + return true + } + return cache.WaitForNamedCacheSync("gv-exclusion-manager", stopCh, synced...) +} + +// getExclusionSet returns the combined exclusion set i.e., union(currentlyActiveGVs, recentlyDeletedGVs) +func (m *GVExclusionManager) getExclusionSet() map[schema.GroupVersion]struct{} { + activeMap := m.loadCurrentlyActiveGVs() + deletedMap := m.loadRecentlyDeletedGVs() + + exclusionSet := make(map[schema.GroupVersion]struct{}, len(activeMap)+len(deletedMap)) + for gv := range activeMap { + exclusionSet[gv] = struct{}{} + } + for gv := range deletedMap { + exclusionSet[gv] = struct{}{} + } + + return exclusionSet +} + +// handleGVUpdate is called when a CRD or APIService event occurs. +// This triggers Worker 1 to rebuild the active GV set. +func (m *GVExclusionManager) handleGVUpdate() { + m.activeGVQueue.Add("sync") +} + +// onDeleteEvent is called when a CRD or APIService is deleted. +// This removes the GVs from currentlyActiveGVs, adds them to recentlyDeletedGVs +// with current timestamp, and queues Worker 3 to refilter the peer discovery cache. +func (m *GVExclusionManager) onDeleteEvent(gvs []schema.GroupVersion) { if len(gvs) == 0 { return } - h.excludedGVsMu.Lock() - for _, gv := range gvs { - h.excludedGVs[gv] = nil + // Remove from currentlyActiveGVs + activeMap := m.loadCurrentlyActiveGVs() + newActiveMap := make(map[schema.GroupVersion]struct{}, len(activeMap)) + for k, v := range activeMap { + newActiveMap[k] = v } - h.excludedGVsMu.Unlock() + for _, gv := range gvs { + delete(newActiveMap, gv) + } + m.currentlyActiveGVs.Store(newActiveMap) - // Load current peer cache and filter it - cache := h.peerDiscoveryInfoCache.Load() + // Add to recentlyDeletedGVs + deletedMap := m.loadRecentlyDeletedGVs() + newDeletedMap := make(map[schema.GroupVersion]time.Time, len(deletedMap)+len(gvs)) + for k, v := range deletedMap { + newDeletedMap[k] = v + } + now := time.Now() + for _, gv := range gvs { + newDeletedMap[gv] = now + klog.V(4).Infof("GV %s deleted: moved to recentlyDeletedGVs", gv.String()) + } + m.recentlyDeletedGVs.Store(newDeletedMap) + + // Queue Worker 3 + m.refilterQueue.Add("refilter") +} + +// RunActiveGVTracker 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) { + defer m.activeGVQueue.ShutDown() + + klog.Infof("Starting %d Active GV Tracker worker(s)", workers) + for i := 0; i < workers; i++ { + go wait.UntilWithContext(ctx, m.runActiveGVTrackerWorker, time.Second) + } + + <-ctx.Done() + klog.Info("Active GV Tracker workers stopped") +} + +func (m *GVExclusionManager) runActiveGVTrackerWorker(ctx context.Context) { + for m.processNextActiveGV(ctx) { + } +} + +func (m *GVExclusionManager) processNextActiveGV(ctx context.Context) bool { + key, shutdown := m.activeGVQueue.Get() + if shutdown { + return false + } + defer m.activeGVQueue.Done(key) + + select { + case <-ctx.Done(): + return false + default: + } + + m.reconcileActiveGVs() + return true +} + +// reconcileActiveGVs fetches all GVs from CRD and APIService informers, +// detects diffs with the previous state, and queues Worker 3 if changes detected. +func (m *GVExclusionManager) reconcileActiveGVs() { + freshGVs := make(map[schema.GroupVersion]struct{}) + + // Fetch GVs from CRD informer + if m.crdInformer != nil && m.crdExtractor != nil { + crdList := m.crdInformer.GetStore().List() + for _, item := range crdList { + gvs := m.crdExtractor(item) + for _, gv := range gvs { + freshGVs[gv] = struct{}{} + } + } + } + + // Fetch GVs from APIService informer + if m.apiServiceInformer != nil && m.apiServiceExtractor != nil { + apiSvcList := m.apiServiceInformer.GetStore().List() + for _, item := range apiSvcList { + gvs := m.apiServiceExtractor(item) + for _, gv := range gvs { + freshGVs[gv] = struct{}{} + } + } + } + + // Load previous active GVs and detect diff + previousGVs := m.loadCurrentlyActiveGVs() + diffDetected := m.diffGVs(previousGVs, freshGVs) + + if diffDetected { + m.currentlyActiveGVs.Store(freshGVs) + klog.V(4).Infof("Active GVs updated: %d GVs now active", len(freshGVs)) + + // Queue Worker 3 for re-filtering + m.refilterQueue.Add("refilter") + } else { + klog.V(4).Infof("No diff detected in active GVs") + } +} + +// diffGVs checks if there's a presence/absence difference between two GV maps +func (m *GVExclusionManager) diffGVs(old, new map[schema.GroupVersion]struct{}) bool { + if len(old) != len(new) { + return true + } + + for gv := range old { + if _, exists := new[gv]; !exists { + return true + } + } + + for gv := range new { + if _, exists := old[gv]; !exists { + return true + } + } + + return false +} + +// RunReaper runs Worker 2: Reaper +// This worker periodically removes expired GVs from recentlyDeletedGVs. +func (m *GVExclusionManager) RunReaper(stopCh <-chan struct{}) { + klog.Infof("Starting GV Reaper with %s interval", m.reaperCheckInterval) + ticker := time.NewTicker(m.reaperCheckInterval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + m.reapExpiredGVs() + case <-stopCh: + klog.Info("GV Reaper stopped") + return + } + } +} + +// reapExpiredGVs removes GVs from recentlyDeletedGVs that have exceeded the grace period. +func (m *GVExclusionManager) reapExpiredGVs() { + deletedMap := m.loadRecentlyDeletedGVs() + now := time.Now() + newDeletedMap := make(map[schema.GroupVersion]time.Time) + anyReaped := false + + for gv, deletionTime := range deletedMap { + if now.Sub(deletionTime) > m.exclusionGracePeriod { + klog.V(4).Infof("Reaping GV %s (grace period expired)", gv.String()) + anyReaped = true + // Don't add to new map (effectively removing it) + } else { + newDeletedMap[gv] = deletionTime + } + } + + if anyReaped { + // Atomic swap + m.recentlyDeletedGVs.Store(newDeletedMap) + } +} + +// RunPeerDiscoveryRefilter runs Worker 3: Peer Discovery Re-filter +// This worker filters the peer discovery cache using the exclusion set. +func (m *GVExclusionManager) RunPeerDiscoveryRefilter(ctx context.Context, workers int) { + defer m.refilterQueue.ShutDown() + + klog.Infof("Starting %d Peer Discovery Re-filter worker(s)", workers) + for i := 0; i < workers; i++ { + go wait.UntilWithContext(ctx, m.runRefilterWorker, time.Second) + } + + <-ctx.Done() + klog.Info("Peer Discovery Re-filter workers stopped") +} + +func (m *GVExclusionManager) runRefilterWorker(ctx context.Context) { + for m.processNextRefilter(ctx) { + } +} + +func (m *GVExclusionManager) processNextRefilter(ctx context.Context) bool { + key, shutdown := m.refilterQueue.Get() + if shutdown { + return false + } + defer m.refilterQueue.Done(key) + + select { + case <-ctx.Done(): + return false + default: + } + + m.refilterPeerDiscoveryCache() + return true +} + +// refilterPeerDiscoveryCache filters the peer discovery cache using the exclusion set. +// Only updates the cache and calls invalidation callback if filtering actually changed content. +func (m *GVExclusionManager) refilterPeerDiscoveryCache() { + if m.peerDiscoveryCache == nil { + klog.Warning("peerDiscoveryCache reference not set") + return + } + + cache := m.peerDiscoveryCache.Load() if cache == nil { + klog.V(4).Infof("Peer discovery cache is empty, skipping re-filter") return } @@ -134,69 +424,90 @@ func (h *peerProxyHandler) addExcludedGVs(gvs []schema.GroupVersion) { return } - // Always trigger filter and callbacks, to ensure late-appearing GVs are filtered. - if peerDiscovery, peerDiscoveryChanged := h.filterPeerDiscoveryCache(cacheMap); peerDiscoveryChanged { - h.storePeerDiscoveryCacheAndInvalidate(peerDiscovery) + if len(cacheMap) == 0 { + klog.V(4).Infof("Peer discovery cache is empty, skipping re-filter") + return + } + + // Filter the cache + filteredCache, changed := m.FilterPeerDiscoveryCache(cacheMap) + + if changed { + // Atomic swap peer cache + m.peerDiscoveryCache.Store(filteredCache) + + // Call invalidation callback + if m.invalidationCallback != nil { + if callback := m.invalidationCallback.Load(); callback != nil { + (*callback)() + } + } + + klog.V(4).Infof("Peer discovery cache re-filtered, %d GVs excluded", len(m.getExclusionSet())) + } else { + klog.V(4).Infof("No changes after filtering peer discovery cache") } } -// filterPeerDiscoveryCache filters the provided peer discovery cache, -// excluding GVs in h.excludedGVs. -func (h *peerProxyHandler) filterPeerDiscoveryCache(cacheMap map[string]PeerDiscoveryCacheEntry) (map[string]PeerDiscoveryCacheEntry, bool) { - h.excludedGVsMu.RLock() - if len(h.excludedGVs) == 0 { - // No exclusions, no filtering needed. - h.excludedGVsMu.RUnlock() +// FilterPeerDiscoveryCache applies the current exclusion set to the provided peer cache. +// This should be called when new peers are added to ensure their discovery is filtered. +// Returns the filtered cache and a boolean indicating if any changes were made. +func (m *GVExclusionManager) FilterPeerDiscoveryCache(cacheMap map[string]PeerDiscoveryCacheEntry) (map[string]PeerDiscoveryCacheEntry, bool) { + exclusionSet := m.getExclusionSet() + if len(exclusionSet) == 0 { return cacheMap, false } - excludedCloned := make(map[schema.GroupVersion]struct{}, len(h.excludedGVs)) - for gv := range h.excludedGVs { - excludedCloned[gv] = struct{}{} - } - h.excludedGVsMu.RUnlock() - filtered := make(map[string]PeerDiscoveryCacheEntry, len(cacheMap)) - peerDiscoveryChanged := false + anyChanged := false + for peerID, entry := range cacheMap { - newEntry, peerChanged := h.filterPeerCacheEntry(entry, excludedCloned) - filtered[peerID] = newEntry - if peerChanged { - peerDiscoveryChanged = true + filteredEntry, changed := m.filterPeerCacheEntry(entry, exclusionSet) + filtered[peerID] = filteredEntry + if changed { + anyChanged = true } } - return filtered, peerDiscoveryChanged + return filtered, anyChanged } -// filterPeerCacheEntry filters a single peer's cache entry for excluded groups. -// Returns the filtered entry and whether any excluded group was present. -func (h *peerProxyHandler) filterPeerCacheEntry(entry PeerDiscoveryCacheEntry, excluded map[schema.GroupVersion]struct{}) (PeerDiscoveryCacheEntry, bool) { +// filterPeerCacheEntry filters a single peer's cache entry for excluded GVs. +// Returns the filtered entry and whether any GVs were excluded. +func (m *GVExclusionManager) filterPeerCacheEntry( + entry PeerDiscoveryCacheEntry, + exclusionSet map[schema.GroupVersion]struct{}, +) (PeerDiscoveryCacheEntry, bool) { filteredGVRs := make(map[schema.GroupVersionResource]bool, len(entry.GVRs)) - peerDiscoveryChanged := false + anyExcluded := false for existingGVR, v := range entry.GVRs { gv := schema.GroupVersion{Group: existingGVR.Group, Version: existingGVR.Version} - if _, skip := excluded[gv]; skip { - peerDiscoveryChanged = true + if _, excluded := exclusionSet[gv]; excluded { + anyExcluded = true continue } filteredGVRs[existingGVR] = v } - if !peerDiscoveryChanged { + if !anyExcluded { return entry, false } // Filter the group discovery list - filteredGroups := h.filterGroupDiscovery(entry.GroupDiscovery, excluded) + filteredGroups := m.filterGroupDiscovery(entry.GroupDiscovery, exclusionSet) + return PeerDiscoveryCacheEntry{ GVRs: filteredGVRs, GroupDiscovery: filteredGroups, }, true } -func (h *peerProxyHandler) filterGroupDiscovery(groupDiscoveries []apidiscoveryv2.APIGroupDiscovery, excluded map[schema.GroupVersion]struct{}) []apidiscoveryv2.APIGroupDiscovery { +// filterGroupDiscovery filters group discovery entries, removing excluded GVs. +func (m *GVExclusionManager) filterGroupDiscovery( + groupDiscoveries []apidiscoveryv2.APIGroupDiscovery, + exclusionSet map[schema.GroupVersion]struct{}, +) []apidiscoveryv2.APIGroupDiscovery { var filteredDiscovery []apidiscoveryv2.APIGroupDiscovery for _, groupDiscovery := range groupDiscoveries { filteredGroup := apidiscoveryv2.APIGroupDiscovery{ @@ -205,7 +516,7 @@ func (h *peerProxyHandler) filterGroupDiscovery(groupDiscoveries []apidiscoveryv for _, version := range groupDiscovery.Versions { gv := schema.GroupVersion{Group: groupDiscovery.Name, Version: version.Version} - if _, found := excluded[gv]; found { + if _, found := exclusionSet[gv]; found { // This version is excluded, skip it continue } @@ -220,149 +531,20 @@ func (h *peerProxyHandler) filterGroupDiscovery(groupDiscoveries []apidiscoveryv return filteredDiscovery } -// RunGVDeletionWorkers starts workers to process GroupVersions from deleted CRDs and APIServices. -// When a GV is processed, it is checked for usage by other resources. If no longer in use, -// it is marked with a deletion timestamp, initiating a grace period. After this period, -// the RunExcludedGVsReaper is responsible for removing the GV from the exclusion set. -func (h *peerProxyHandler) RunGVDeletionWorkers(ctx context.Context, workers int) { - defer func() { - klog.Info("Shutting down GV deletion workers") - h.gvDeletionQueue.ShutDown() - }() - klog.Infof("Starting %d GV deletion worker(s)", workers) - for i := 0; i < workers; i++ { - go func(workerID int) { - for h.processNextGVDeletion() { - select { - case <-ctx.Done(): - klog.Infof("GV deletion worker %d shutting down", workerID) - return - default: - } - } - }(i) - } - - <-ctx.Done() // Wait till context is done to call deferred shutdown function -} - -// processNextGVDeletion processes a single item from the GV deletion queue. -func (h *peerProxyHandler) processNextGVDeletion() bool { - gvString, shutdown := h.gvDeletionQueue.Get() - if shutdown { - return false - } - defer h.gvDeletionQueue.Done(gvString) - - gv, err := schema.ParseGroupVersion(gvString) - if err != nil { - klog.Errorf("Failed to parse GroupVersion %q: %v", gvString, err) - return true - } - - h.markGVForDeletionIfUnused(gv) - return true -} - -// markGVForDeletionIfUnused checks if a group-version is still in use. -// If not, it marks it for deletion by setting a timestamp. -func (h *peerProxyHandler) markGVForDeletionIfUnused(gv schema.GroupVersion) { - // Best-effort check if GV is still in use. This is racy - a new CRD/APIService could be - // created after this check completes and returns false. - if h.isGVUsed(gv) { - return - } - - h.excludedGVsMu.Lock() - defer h.excludedGVsMu.Unlock() - - // Only mark if it's currently active (nil timestamp) - if ts, exists := h.excludedGVs[gv]; exists && ts == nil { - now := time.Now() - h.excludedGVs[gv] = &now - klog.V(4).Infof("Marking group-version %q for deletion, grace period started", gv) - } -} - -// isGVUsed checks the informer stores to see if any CRD or APIService -// is still using the given group-version. -// -// This is necessary because multiple CRDs or aggregated APIs can share the same group-version, -// but define different resources (kinds) or versions. Deleting one CRD or APIService -// for a group-version does not mean the group-version is entirely gone—other CRDs or APIs in that group-version -// may still exist. We must only remove a group-version from the exclusion set when there are -// no remaining CRDs or APIService objects for that group-version, to ensure we do not prematurely -// allow peer-aggregated discovery or proxying for APIs that are still served locally. -func (h *peerProxyHandler) isGVUsed(gv schema.GroupVersion) bool { - // Check CRD informer store for the specific group-version - if h.crdInformer != nil && h.crdExtractor != nil { - crdList := h.crdInformer.GetStore().List() - for _, item := range crdList { - gvs := h.crdExtractor(item) - if gvs == nil { - continue - } - for _, extractedGV := range gvs { - if extractedGV == gv { - return true - } - } +func (m *GVExclusionManager) loadCurrentlyActiveGVs() map[schema.GroupVersion]struct{} { + if val := m.currentlyActiveGVs.Load(); val != nil { + if gvMap, ok := val.(map[schema.GroupVersion]struct{}); ok { + return gvMap } } - - // Check APIService informer store for the specific group-version - if h.apiServiceInformer != nil && h.apiServiceExtractor != nil { - apiSvcList := h.apiServiceInformer.GetStore().List() - for _, item := range apiSvcList { - gvs := h.apiServiceExtractor(item) - if gvs == nil { - continue - } - for _, extractedGV := range gvs { - if extractedGV == gv { - return true - } - } - } - } - - return false + return map[schema.GroupVersion]struct{}{} } -// RunExcludedGVsReaper starts a goroutine that periodically cleans up -// excluded group-versions that have passed their grace period. -func (h *peerProxyHandler) RunExcludedGVsReaper(stopCh <-chan struct{}) { - klog.Infof("Starting excluded group-versions reaper with %s interval", h.reaperCheckInterval) - ticker := time.NewTicker(h.reaperCheckInterval) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - h.reapExcludedGVs() - case <-stopCh: - klog.Info("Shutting down excluded group-versions reaper") - return - } - } -} - -// reapExcludedGVs is the garbage collector function. -// It removes group-versions from excludedGVs if their grace period has expired. -func (h *peerProxyHandler) reapExcludedGVs() { - h.excludedGVsMu.Lock() - defer h.excludedGVsMu.Unlock() - - now := time.Now() - for gv, deletionTime := range h.excludedGVs { - if deletionTime == nil { - // Still actively excluded - continue - } - - if now.Sub(*deletionTime) > h.exclusionGracePeriod { - klog.V(4).Infof("Reaping excluded group-version %q (grace period expired)", gv) - delete(h.excludedGVs, gv) +func (m *GVExclusionManager) loadRecentlyDeletedGVs() map[schema.GroupVersion]time.Time { + if val := m.recentlyDeletedGVs.Load(); val != nil { + if gvMap, ok := val.(map[schema.GroupVersion]time.Time); ok { + return gvMap } } + return map[schema.GroupVersion]time.Time{} } diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go index 12d06dd4a7e..31b2bbc037a 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go @@ -115,11 +115,9 @@ func (h *peerProxyHandler) syncPeerDiscoveryCache(ctx context.Context) error { } } - // Apply exclusion filter to the cache. - if len(newCache) != 0 { - if filteredCache, peerDiscoveryChanged := h.filterPeerDiscoveryCache(newCache); peerDiscoveryChanged { - newCache = filteredCache - } + // Apply current exclusion filter to new peer cache entries. + if filtered, changed := h.gvExclusionManager.FilterPeerDiscoveryCache(newCache); changed { + newCache = filtered } h.storePeerDiscoveryCacheAndInvalidate(newCache) From 79375cf7dee64515d65d702a1e1d9c3a6ff4f0ec Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Wed, 19 Nov 2025 20:10:38 -0800 Subject: [PATCH 3/8] Add tests for GVExclusionManager --- .../peerproxy/gv_exclusion_manager_test.go | 573 +++++++++++++++--- 1 file changed, 490 insertions(+), 83 deletions(-) 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 bafa19341ba..cba80306456 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 @@ -17,8 +17,9 @@ limitations under the License. package peerproxy import ( - "reflect" + "sync/atomic" "testing" + "time" "k8s.io/apimachinery/pkg/runtime/schema" @@ -26,106 +27,512 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -func TestFilteringLogic(t *testing.T) { - gvr := func(g, v, r string) schema.GroupVersionResource { - return schema.GroupVersionResource{Group: g, Version: v, Resource: r} - } - gv := func(g, v string) schema.GroupVersion { - return schema.GroupVersion{Group: g, Version: v} - } - - testCases := []struct { - name string - initialPeerCache PeerDiscoveryCacheEntry - excludedGVs map[schema.GroupVersion]struct{} - wantFilteredGVRs map[schema.GroupVersionResource]bool - wantFilteredGroups []apidiscoveryv2.APIGroupDiscovery - wantChange bool +func TestGetExclusionSet(t *testing.T) { + tests := []struct { + name string + activeGVs map[schema.GroupVersion]struct{} + deletedGVs map[schema.GroupVersion]time.Time + wantGVs []schema.GroupVersion }{ { - name: "no exclusions", - initialPeerCache: PeerDiscoveryCacheEntry{ - GVRs: map[schema.GroupVersionResource]bool{gvr("apps", "v1", "deployments"): true}, - GroupDiscovery: []apidiscoveryv2.APIGroupDiscovery{{ - ObjectMeta: metav1.ObjectMeta{Name: "apps"}, - Versions: []apidiscoveryv2.APIVersionDiscovery{{Version: "v1"}}, - }}, - }, - excludedGVs: map[schema.GroupVersion]struct{}{}, - wantFilteredGVRs: map[schema.GroupVersionResource]bool{gvr("apps", "v1", "deployments"): true}, - wantFilteredGroups: []apidiscoveryv2.APIGroupDiscovery{{ - ObjectMeta: metav1.ObjectMeta{Name: "apps"}, - Versions: []apidiscoveryv2.APIVersionDiscovery{{Version: "v1"}}, - }}, - wantChange: false, + name: "empty state", + activeGVs: nil, + deletedGVs: nil, + wantGVs: []schema.GroupVersion{}, }, { - name: "exclude one GV that exists", - initialPeerCache: PeerDiscoveryCacheEntry{ - GVRs: map[schema.GroupVersionResource]bool{ - gvr("apps", "v1", "deployments"): true, - gvr("apps", "v1", "statefulsets"): true, - gvr("batch", "v1", "jobs"): true, - }, - GroupDiscovery: []apidiscoveryv2.APIGroupDiscovery{ - { - ObjectMeta: metav1.ObjectMeta{Name: "apps"}, - Versions: []apidiscoveryv2.APIVersionDiscovery{{Version: "v1"}}, - }, - { - ObjectMeta: metav1.ObjectMeta{Name: "batch"}, - Versions: []apidiscoveryv2.APIVersionDiscovery{{Version: "v1"}}, - }, - }, + name: "only active GVs", + activeGVs: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + {Group: "batch", Version: "v1"}: {}, + {Group: "custom", Version: "v1alpha1"}: {}, }, - excludedGVs: map[schema.GroupVersion]struct{}{ - gv("apps", "v1"): {}, + wantGVs: []schema.GroupVersion{ + {Group: "apps", Version: "v1"}, + {Group: "batch", Version: "v1"}, + {Group: "custom", Version: "v1alpha1"}, }, - wantFilteredGVRs: map[schema.GroupVersionResource]bool{gvr("batch", "v1", "jobs"): true}, - wantFilteredGroups: []apidiscoveryv2.APIGroupDiscovery{ - { - ObjectMeta: metav1.ObjectMeta{Name: "batch"}, - Versions: []apidiscoveryv2.APIVersionDiscovery{{Version: "v1"}}, - }, - }, - wantChange: true, }, { - name: "exclude a GV that does not exist", - initialPeerCache: PeerDiscoveryCacheEntry{ - GVRs: map[schema.GroupVersionResource]bool{gvr("apps", "v1", "deployments"): true}, - GroupDiscovery: []apidiscoveryv2.APIGroupDiscovery{{ - ObjectMeta: metav1.ObjectMeta{Name: "apps"}, - Versions: []apidiscoveryv2.APIVersionDiscovery{{Version: "v1"}}, - }}, + name: "only deleted GVs", + deletedGVs: map[schema.GroupVersion]time.Time{ + {Group: "deprecated", Version: "v1beta1"}: time.Now(), + {Group: "legacy", Version: "v1alpha1"}: time.Now(), }, - excludedGVs: map[schema.GroupVersion]struct{}{ - gv("foo", "v1"): {}, + wantGVs: []schema.GroupVersion{ + {Group: "deprecated", Version: "v1beta1"}, + {Group: "legacy", Version: "v1alpha1"}, + }, + }, + { + name: "active and deleted GVs", + activeGVs: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + {Group: "batch", Version: "v1"}: {}, + }, + deletedGVs: map[schema.GroupVersion]time.Time{ + {Group: "old", Version: "v1"}: time.Now(), + }, + wantGVs: []schema.GroupVersion{ + {Group: "apps", Version: "v1"}, + {Group: "batch", Version: "v1"}, + {Group: "old", Version: "v1"}, }, - wantFilteredGVRs: map[schema.GroupVersionResource]bool{gvr("apps", "v1", "deployments"): true}, - wantFilteredGroups: []apidiscoveryv2.APIGroupDiscovery{{ - ObjectMeta: metav1.ObjectMeta{Name: "apps"}, - Versions: []apidiscoveryv2.APIVersionDiscovery{{Version: "v1"}}, - }}, - wantChange: false, }, } - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - h := &peerProxyHandler{} - filteredEntry, changed := h.filterPeerCacheEntry(tc.initialPeerCache, tc.excludedGVs) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) - if changed != tc.wantChange { - t.Errorf("want change to be %v, got %v", tc.wantChange, changed) + if tt.activeGVs != nil { + mgr.currentlyActiveGVs.Store(tt.activeGVs) + } + if tt.deletedGVs != nil { + mgr.recentlyDeletedGVs.Store(tt.deletedGVs) } - if !reflect.DeepEqual(filteredEntry.GVRs, tc.wantFilteredGVRs) { - t.Errorf("filtered GVRs mismatch: got %v, want %v", filteredEntry.GVRs, tc.wantFilteredGVRs) + exclusionSet := mgr.getExclusionSet() + if len(exclusionSet) != len(tt.wantGVs) { + t.Errorf("Want exclusion set size %d, got %d", len(tt.wantGVs), len(exclusionSet)) } - if !reflect.DeepEqual(filteredEntry.GroupDiscovery, tc.wantFilteredGroups) { - t.Errorf("filtered GroupDiscovery mismatch: got %v, want %v", filteredEntry.GroupDiscovery, tc.wantFilteredGroups) + for _, gv := range tt.wantGVs { + if _, found := exclusionSet[gv]; !found { + t.Errorf("Want GV %v in exclusion set, but not found", gv) + } + } + }) + } +} + +func TestReapExpiredGVs(t *testing.T) { + tests := []struct { + name string + gracePeriod time.Duration + deletedGVs map[schema.GroupVersion]time.Time + wantReaped []schema.GroupVersion + }{ + { + name: "empty state", + gracePeriod: 100 * time.Millisecond, + deletedGVs: map[schema.GroupVersion]time.Time{}, + wantReaped: []schema.GroupVersion{}, + }, + { + name: "reap old GV keep recent", + gracePeriod: 100 * time.Millisecond, + deletedGVs: map[schema.GroupVersion]time.Time{ + {Group: "old", Version: "v1"}: time.Now().Add(-200 * time.Millisecond), + {Group: "recent", Version: "v1"}: time.Now().Add(-50 * time.Millisecond), + {Group: "new", Version: "v1"}: time.Now(), + }, + wantReaped: []schema.GroupVersion{ + {Group: "old", Version: "v1"}, + }, + }, + { + name: "all expired", + gracePeriod: 50 * time.Millisecond, + deletedGVs: map[schema.GroupVersion]time.Time{ + {Group: "old1", Version: "v1"}: time.Now().Add(-100 * time.Millisecond), + {Group: "old2", Version: "v1"}: time.Now().Add(-200 * time.Millisecond), + }, + wantReaped: []schema.GroupVersion{ + {Group: "old1", Version: "v1"}, + {Group: "old2", Version: "v1"}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mgr := NewGVExclusionManager(tt.gracePeriod, 50*time.Millisecond, &atomic.Value{}, &atomic.Pointer[func()]{}) + mgr.recentlyDeletedGVs.Store(tt.deletedGVs) + mgr.reapExpiredGVs() + result := mgr.loadRecentlyDeletedGVs() + + for _, gv := range tt.wantReaped { + if _, found := result[gv]; found { + t.Errorf("GV %v should have been reaped but still exists", gv) + } + } + }) + } +} + +func TestDetectDiff(t *testing.T) { + mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + + tests := []struct { + name string + old map[schema.GroupVersion]struct{} + new map[schema.GroupVersion]struct{} + want bool + }{ + { + name: "both empty", + old: map[schema.GroupVersion]struct{}{}, + new: map[schema.GroupVersion]struct{}{}, + want: false, + }, + { + name: "identical", + old: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + new: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + want: false, + }, + { + name: "added GV", + old: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + new: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + {Group: "batch", Version: "v1"}: {}, + }, + want: true, + }, + { + name: "removed GV", + old: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + {Group: "batch", Version: "v1"}: {}, + }, + new: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + want: true, + }, + { + name: "different GVs same size", + old: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + new: map[schema.GroupVersion]struct{}{ + {Group: "batch", Version: "v1"}: {}, + }, + want: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := mgr.diffGVs(tt.old, tt.new) + if result != tt.want { + t.Errorf("detectDiff() = %v, want %v", result, tt.want) + } + }) + } +} + +func TestFilterPeerDiscoveryCache(t *testing.T) { + tests := []struct { + name string + activeGVs map[schema.GroupVersion]struct{} + deletedGVs map[schema.GroupVersion]time.Time + cacheMap map[string]PeerDiscoveryCacheEntry + wantChanged bool + wantPeerGVRs map[string]int // peer name -> GVR count + }{ + { + name: "empty exclusion no changes", + activeGVs: nil, + cacheMap: map[string]PeerDiscoveryCacheEntry{ + "peer1": { + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "apps", Version: "v1", Resource: "deployments"}: true, + }, + }, + "peer2": { + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "batch", Version: "v1", Resource: "jobs"}: true, + }, + }, + }, + wantChanged: false, + wantPeerGVRs: map[string]int{ + "peer1": 1, + "peer2": 1, + }, + }, + { + name: "filter active GVs", + activeGVs: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + cacheMap: map[string]PeerDiscoveryCacheEntry{ + "peer1": { + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "apps", Version: "v1", Resource: "deployments"}: true, + }, + }, + "peer2": { + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "batch", Version: "v1", Resource: "jobs"}: true, + }, + }, + }, + wantChanged: true, + wantPeerGVRs: map[string]int{ + "peer1": 0, // apps/v1 filtered out + "peer2": 1, // unchanged + }, + }, + { + name: "filter deleted GVs", + deletedGVs: map[schema.GroupVersion]time.Time{ + {Group: "custom", Version: "v1alpha1"}: time.Now(), + }, + cacheMap: map[string]PeerDiscoveryCacheEntry{ + "peer1": { + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "custom", Version: "v1alpha1", Resource: "myresources"}: true, + }, + }, + "peer2": { + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "apps", Version: "v1", Resource: "deployments"}: true, + }, + }, + }, + wantChanged: true, + wantPeerGVRs: map[string]int{ + "peer1": 0, // custom/v1alpha1 filtered out + "peer2": 1, // unchanged + }, + }, + { + name: "filter both active and deleted GVs", + activeGVs: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + deletedGVs: map[schema.GroupVersion]time.Time{ + {Group: "custom", Version: "v1alpha1"}: time.Now(), + }, + cacheMap: map[string]PeerDiscoveryCacheEntry{ + "peer1": { + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "apps", Version: "v1", Resource: "deployments"}: true, + {Group: "custom", Version: "v1alpha1", Resource: "myresources"}: true, + {Group: "batch", Version: "v1", Resource: "jobs"}: true, + }, + }, + }, + wantChanged: true, + wantPeerGVRs: map[string]int{ + "peer1": 1, // apps/v1 and custom/v1alpha1 filtered out, batch/v1 remains + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + + if tt.activeGVs != nil { + mgr.currentlyActiveGVs.Store(tt.activeGVs) + } + if tt.deletedGVs != nil { + mgr.recentlyDeletedGVs.Store(tt.deletedGVs) + } + + filtered, changed := mgr.FilterPeerDiscoveryCache(tt.cacheMap) + if changed != tt.wantChanged { + t.Errorf("Want changed=%v, got %v", tt.wantChanged, changed) + } + + if len(filtered) != len(tt.cacheMap) { + t.Errorf("Want %d peers in filtered cache, got %d", len(tt.cacheMap), len(filtered)) + } + + for peerName, wantCount := range tt.wantPeerGVRs { + entry, found := filtered[peerName] + if !found { + t.Errorf("Peer %s not found in filtered cache", peerName) + continue + } + if len(entry.GVRs) != wantCount { + t.Errorf("Peer %s: want %d GVRs, got %d", peerName, wantCount, len(entry.GVRs)) + } + } + }) + } +} + +func TestFilterPeerCacheEntry(t *testing.T) { + tests := []struct { + name string + entry PeerDiscoveryCacheEntry + exclusionSet map[schema.GroupVersion]struct{} + wantChanged bool + wantGVRs []schema.GroupVersionResource + }{ + { + name: "filter GVRs and groups", + entry: PeerDiscoveryCacheEntry{ + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "apps", Version: "v1", Resource: "deployments"}: true, + {Group: "batch", Version: "v1", Resource: "jobs"}: true, + {Group: "custom", Version: "v1", Resource: "myresources"}: true, + }, + GroupDiscovery: []apidiscoveryv2.APIGroupDiscovery{ + { + ObjectMeta: metav1.ObjectMeta{Name: "apps"}, + Versions: []apidiscoveryv2.APIVersionDiscovery{ + {Version: "v1"}, + }, + }, + { + ObjectMeta: metav1.ObjectMeta{Name: "batch"}, + Versions: []apidiscoveryv2.APIVersionDiscovery{ + {Version: "v1"}, + }, + }, + }, + }, + exclusionSet: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + wantChanged: true, + wantGVRs: []schema.GroupVersionResource{ + {Group: "batch", Version: "v1", Resource: "jobs"}, + {Group: "custom", Version: "v1", Resource: "myresources"}, + }, + }, + { + name: "no changes when exclusion doesn't match", + entry: PeerDiscoveryCacheEntry{ + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "batch", Version: "v1", Resource: "jobs"}: true, + }, + }, + exclusionSet: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + wantChanged: false, + wantGVRs: []schema.GroupVersionResource{ + {Group: "batch", Version: "v1", Resource: "jobs"}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + + filtered, changed := mgr.filterPeerCacheEntry(tt.entry, tt.exclusionSet) + + if changed != tt.wantChanged { + t.Errorf("Want changed=%v, got %v", tt.wantChanged, changed) + } + + for _, gvr := range tt.wantGVRs { + if _, found := filtered.GVRs[gvr]; !found { + t.Errorf("Want GVR %v to be present", gvr) + } + } + }) + } +} + +func TestFilterGroupDiscovery(t *testing.T) { + tests := []struct { + name string + groupDiscoveries []apidiscoveryv2.APIGroupDiscovery + exclusionSet map[schema.GroupVersion]struct{} + wantGroups map[string][]string // group name -> list of versions + }{ + { + name: "partial version exclusion", + groupDiscoveries: []apidiscoveryv2.APIGroupDiscovery{ + { + ObjectMeta: metav1.ObjectMeta{Name: "apps"}, + Versions: []apidiscoveryv2.APIVersionDiscovery{ + {Version: "v1"}, + {Version: "v1beta1"}, + }, + }, + { + ObjectMeta: metav1.ObjectMeta{Name: "batch"}, + Versions: []apidiscoveryv2.APIVersionDiscovery{ + {Version: "v1"}, + }, + }, + }, + exclusionSet: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + wantGroups: map[string][]string{ + "apps": {"v1beta1"}, + "batch": {"v1"}, + }, + }, + { + name: "all versions excluded", + groupDiscoveries: []apidiscoveryv2.APIGroupDiscovery{ + { + ObjectMeta: metav1.ObjectMeta{Name: "apps"}, + Versions: []apidiscoveryv2.APIVersionDiscovery{ + {Version: "v1"}, + }, + }, + { + ObjectMeta: metav1.ObjectMeta{Name: "batch"}, + Versions: []apidiscoveryv2.APIVersionDiscovery{ + {Version: "v1"}, + }, + }, + }, + exclusionSet: map[schema.GroupVersion]struct{}{ + {Group: "apps", Version: "v1"}: {}, + }, + wantGroups: map[string][]string{ + "batch": {"v1"}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + + filtered := mgr.filterGroupDiscovery(tt.groupDiscoveries, tt.exclusionSet) + for groupName, wantVersionStrs := range tt.wantGroups { + var group *apidiscoveryv2.APIGroupDiscovery + for i := range filtered { + if filtered[i].Name == groupName { + group = &filtered[i] + break + } + } + + if group == nil { + t.Errorf("Want group %s not found in filtered results", groupName) + continue + } + + if len(group.Versions) != len(wantVersionStrs) { + t.Errorf("Group %s: want %d versions, got %d", groupName, len(wantVersionStrs), len(group.Versions)) + continue + } + + for _, wantVer := range wantVersionStrs { + found := false + for _, ver := range group.Versions { + if ver.Version == wantVer { + found = true + break + } + } + if !found { + t.Errorf("Group %s: want version %s not found", groupName, wantVer) + } + } } }) } From 1713b6d8f242aa85188956401aeabb3c0fa20e66 Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Wed, 19 Nov 2025 20:28:05 -0800 Subject: [PATCH 4/8] Wire GVExclusionManager --- pkg/controlplane/apiserver/server.go | 24 +++-- .../util/peerproxy/gv_exclusion_manager.go | 10 +- .../peerproxy/gv_exclusion_manager_test.go | 26 ++++- .../apiserver/pkg/util/peerproxy/peerproxy.go | 101 ++++++++++++++++-- .../pkg/util/peerproxy/peerproxy_handler.go | 31 ++---- 5 files changed, 142 insertions(+), 50 deletions(-) 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. From aaed7525ddcf04b00c6d1a96ed13e7663b3a22ad Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Wed, 21 Jan 2026 16:41:01 -0800 Subject: [PATCH 5/8] Consolidate delete handler, reaper logic into same sync loop using the same worker - activeGVQueue --- pkg/controlplane/apiserver/server.go | 23 +-- .../util/peerproxy/gv_exclusion_manager.go | 189 ++++++++---------- .../peerproxy/gv_exclusion_manager_test.go | 8 +- 3 files changed, 85 insertions(+), 135 deletions(-) diff --git a/pkg/controlplane/apiserver/server.go b/pkg/controlplane/apiserver/server.go index b12891fdfe7..d536a1dd59c 100644 --- a/pkg/controlplane/apiserver/server.go +++ b/pkg/controlplane/apiserver/server.go @@ -224,30 +224,11 @@ func (c completedConfig) New(name string, delegationTarget genericapiserver.Dele return err }) - // Run peer-discovery sync loop - s.GenericAPIServer.AddPostStartHookOrDie("peer-discovery-cache-sync", func(context genericapiserver.PostStartHookContext) error { + // Run peer-discovery workers + s.GenericAPIServer.AddPostStartHookOrDie("peer-discovery-workers", func(context genericapiserver.PostStartHookContext) error { go c.Extra.PeerProxy.RunPeerDiscoveryCacheSync(context, 1) - return nil - }) - - // 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 - }) - - // 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 b1f9e4cac63..950a05f2635 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 @@ -42,10 +42,11 @@ type GVExtractor func(obj interface{}) []schema.GroupVersion // - recentlyDeletedGVs: GVs belonging to CRDs and aggregated APIServices that were recently deleted, // tracked with deletion timestamp for grace period // -// It runs three workers: -// 1. Active GV Tracker: Triggered on CRD/APIService events, rebuilds active GVs -// 2. Reaper: Periodically removes expired GVs from recentlyDeletedGVs -// 3. Peer Discovery Re-filter: Rate-limited worker that filters peer cache +// It runs two workers and a periodic ticker: +// 1. Active GV Tracker: Triggered on CRD/APIService events or reaper ticks, +// rebuilds active GVs and reaps expired deleted GVs +// 2. Peer Discovery Re-filter: Rate-limited worker that filters peer cache +// 3. Reaper Ticker: Periodically triggers the Active GV Tracker to reap expired GVs type GVExclusionManager struct { // Atomic maps for lock-free access currentlyActiveGVs atomic.Value // map[schema.GroupVersion]struct{} @@ -57,12 +58,12 @@ type GVExclusionManager struct { apiServiceInformer cache.SharedIndexInformer apiServiceExtractor GVExtractor - // Worker 1: triggered by CRD/APIService events + // Worker 1: triggered by CRD/APIService events or reaper ticks activeGVQueue workqueue.TypedRateLimitingInterface[string] - // Worker 2: periodic reaper's configuration + // Reaper ticker configuration exclusionGracePeriod time.Duration reaperCheckInterval time.Duration - // Worker 3: triggered by Active/Deleted GV changes + // Worker 2: triggered by Active/Deleted GV changes refilterQueue workqueue.TypedRateLimitingInterface[string] peerDiscoveryCache *atomic.Value // peerProxyHandler.peerDiscoveryInfoCache @@ -112,15 +113,7 @@ func (m *GVExclusionManager) RegisterCRDInformerHandlers(crdInformer cache.Share m.handleGVUpdate() }, DeleteFunc: func(obj interface{}) { - // Handle tombstone objects - if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { - obj = tombstone.Obj - } - gvs := extractor(obj) - if gvs == nil { - return - } - m.onDeleteEvent(gvs) + m.handleGVUpdate() }, }) return err @@ -140,16 +133,7 @@ func (m *GVExclusionManager) RegisterAPIServiceInformerHandlers(apiServiceInform m.handleGVUpdate() }, DeleteFunc: func(obj interface{}) { - // Handle tombstone objects - if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { - obj = tombstone.Obj - } - - gvs := extractor(obj) - if gvs == nil { - return - } - m.onDeleteEvent(gvs) + m.handleGVUpdate() }, }) return err @@ -187,49 +171,15 @@ func (m *GVExclusionManager) getExclusionSet() map[schema.GroupVersion]struct{} } // handleGVUpdate is called when a CRD or APIService event occurs. -// This triggers Worker 1 to rebuild the active GV set. +// This triggers reconciliation which rebuilds the active GV set +// and also reaps expired GVs if indicated so. func (m *GVExclusionManager) handleGVUpdate() { m.activeGVQueue.Add("sync") } -// onDeleteEvent is called when a CRD or APIService is deleted. -// This removes the GVs from currentlyActiveGVs, adds them to recentlyDeletedGVs -// with current timestamp, and queues Worker 3 to refilter the peer discovery cache. -func (m *GVExclusionManager) onDeleteEvent(gvs []schema.GroupVersion) { - if len(gvs) == 0 { - return - } - - // Remove from currentlyActiveGVs - activeMap := m.loadCurrentlyActiveGVs() - newActiveMap := make(map[schema.GroupVersion]struct{}, len(activeMap)) - for k, v := range activeMap { - newActiveMap[k] = v - } - for _, gv := range gvs { - delete(newActiveMap, gv) - } - m.currentlyActiveGVs.Store(newActiveMap) - - // Add to recentlyDeletedGVs - deletedMap := m.loadRecentlyDeletedGVs() - newDeletedMap := make(map[schema.GroupVersion]time.Time, len(deletedMap)+len(gvs)) - for k, v := range deletedMap { - newDeletedMap[k] = v - } - now := time.Now() - for _, gv := range gvs { - newDeletedMap[gv] = now - klog.V(4).Infof("GV %s deleted: moved to recentlyDeletedGVs", gv.String()) - } - m.recentlyDeletedGVs.Store(newDeletedMap) - - // Queue Worker 3 - m.refilterQueue.Add("refilter") -} - // RunPeerDiscoveryActiveGVTracker runs Worker 1: Active GV Tracker -// This worker is triggered by CRD/APIService events and rebuilds the active GV set. +// This worker is triggered by CRD/APIService events and +// rebuilds the active GV set and reaps expired GVs if indicated so. func (m *GVExclusionManager) RunPeerDiscoveryActiveGVTracker(ctx context.Context, workers int) { defer m.activeGVQueue.ShutDown() @@ -264,8 +214,11 @@ func (m *GVExclusionManager) processNextActiveGV(ctx context.Context) bool { return true } -// reconcileActiveGVs fetches all GVs from CRD and APIService informers, -// detects diffs with the previous state, and queues Worker 3 if changes detected. +// reconcileActiveGVs does the following +// 1. fetches all GVs from CRD and APIService informers +// 2. detects diffs with the previous state +// 3. adds deleted GVs to recentlyDeletedGVs +// 4. reaps expired GVs, and queues Worker 2 if changes detected. func (m *GVExclusionManager) reconcileActiveGVs() { freshGVs := make(map[schema.GroupVersion]struct{}) @@ -293,42 +246,80 @@ func (m *GVExclusionManager) reconcileActiveGVs() { // Load previous active GVs and detect diff previousGVs := m.loadCurrentlyActiveGVs() - diffDetected := m.diffGVs(previousGVs, freshGVs) + deletedGVs, activeGVsChanged := diffGVs(previousGVs, freshGVs) - if diffDetected { - m.currentlyActiveGVs.Store(freshGVs) - klog.V(4).Infof("Active GVs updated: %d GVs now active", len(freshGVs)) + // Update recentlyDeletedGVs: add newly deleted GVs and reap expired ones + recentlyDeletedChanged := m.updateRecentlyDeletedGVs(deletedGVs) - // Queue Worker 3 for re-filtering + if activeGVsChanged || recentlyDeletedChanged { + if activeGVsChanged { + m.currentlyActiveGVs.Store(freshGVs) + klog.V(4).Infof("Active GVs updated: %d GVs now active", len(freshGVs)) + } + // Queue re-filtering of peer discovery cache m.refilterQueue.Add("refilter") } else { - klog.V(4).Infof("No diff detected in active GVs") + klog.V(4).Infof("No changes detected in active or recently deleted GVs") } } -// diffGVs checks if there's a presence/absence difference between two GV maps -func (m *GVExclusionManager) diffGVs(old, new map[schema.GroupVersion]struct{}) bool { - if len(old) != len(new) { - return true +// updateRecentlyDeletedGVs adds newly deleted GVs to recentlyDeletedGVs +// and reaps expired ones. Returns true if any changes were made. +func (m *GVExclusionManager) updateRecentlyDeletedGVs(deletedGVs []schema.GroupVersion) bool { + deletedMap := m.loadRecentlyDeletedGVs() + // Early return if nothing to do + if len(deletedGVs) == 0 && len(deletedMap) == 0 { + return false } + now := time.Now() + newDeletedMap := make(map[schema.GroupVersion]time.Time, len(deletedMap)+len(deletedGVs)) + changed := false + + // Copy existing entries, reaping expired ones + for gv, deletionTime := range deletedMap { + if now.Sub(deletionTime) > m.exclusionGracePeriod { + klog.V(4).Infof("Reaping GV %s (grace period expired)", gv.String()) + changed = true + // Don't add to new map (effectively removing it) + } else { + newDeletedMap[gv] = deletionTime + } + } + + // Add newly deleted GVs + for _, gv := range deletedGVs { + newDeletedMap[gv] = now + klog.V(4).Infof("GV %s deleted: moved to recentlyDeletedGVs", gv.String()) + changed = true + } + + if changed { + m.recentlyDeletedGVs.Store(newDeletedMap) + } + + return changed +} + +// diffGVs compares old and new GV maps, returns GVs that were deleted (in old but not new) +// and a boolean indicating if there were any changes (additions or deletions). +func diffGVs(old, new map[schema.GroupVersion]struct{}) ([]schema.GroupVersion, bool) { + var deletedGVs []schema.GroupVersion + hasChanges := len(old) != len(new) + + // Find deleted GVs (in old but not in new) for gv := range old { if _, exists := new[gv]; !exists { - return true + deletedGVs = append(deletedGVs, gv) + hasChanges = true } } - for gv := range new { - if _, exists := old[gv]; !exists { - return true - } - } - - return false + return deletedGVs, hasChanges } // RunPeerDiscoveryReaper runs Worker 2: Reaper -// This worker periodically removes expired GVs from recentlyDeletedGVs. +// This worker periodically triggers reconciliation which also reaps expired GVs. func (m *GVExclusionManager) RunPeerDiscoveryReaper(ctx context.Context) { klog.Infof("Starting GV Reaper with %s interval", m.reaperCheckInterval) ticker := time.NewTicker(m.reaperCheckInterval) @@ -337,7 +328,8 @@ func (m *GVExclusionManager) RunPeerDiscoveryReaper(ctx context.Context) { for { select { case <-ticker.C: - m.reapExpiredGVs() + // Trigger reconciliation which will also reap expired GVs + m.activeGVQueue.Add("reap") case <-ctx.Done(): klog.Info("GV Reaper stopped") return @@ -345,30 +337,7 @@ func (m *GVExclusionManager) RunPeerDiscoveryReaper(ctx context.Context) { } } -// reapExpiredGVs removes GVs from recentlyDeletedGVs that have exceeded the grace period. -func (m *GVExclusionManager) reapExpiredGVs() { - deletedMap := m.loadRecentlyDeletedGVs() - now := time.Now() - newDeletedMap := make(map[schema.GroupVersion]time.Time) - anyReaped := false - - for gv, deletionTime := range deletedMap { - if now.Sub(deletionTime) > m.exclusionGracePeriod { - klog.V(4).Infof("Reaping GV %s (grace period expired)", gv.String()) - anyReaped = true - // Don't add to new map (effectively removing it) - } else { - newDeletedMap[gv] = deletionTime - } - } - - if anyReaped { - // Atomic swap - m.recentlyDeletedGVs.Store(newDeletedMap) - } -} - -// RunPeerDiscoveryRefilter runs Worker 3: Peer Discovery Re-filter +// RunPeerDiscoveryRefilter runs Worker 2: Peer Discovery Re-filter // This worker filters the peer discovery cache using the exclusion set. func (m *GVExclusionManager) RunPeerDiscoveryRefilter(ctx context.Context, workers int) { defer m.refilterQueue.ShutDown() 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 a7324b86fe1..1e3ea220e32 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 @@ -168,7 +168,7 @@ func TestReapExpiredGVs(t *testing.T) { t.Run(tt.name, func(t *testing.T) { mgr := NewGVExclusionManager(tt.gracePeriod, 50*time.Millisecond, &atomic.Value{}, &atomic.Pointer[func()]{}) mgr.recentlyDeletedGVs.Store(tt.deletedGVs) - mgr.reapExpiredGVs() + mgr.updateRecentlyDeletedGVs(nil) result := mgr.loadRecentlyDeletedGVs() for _, gv := range tt.wantReaped { @@ -181,7 +181,7 @@ func TestReapExpiredGVs(t *testing.T) { } func TestDetectDiff(t *testing.T) { - mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + _ = NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) tests := []struct { name string @@ -241,9 +241,9 @@ func TestDetectDiff(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - result := mgr.diffGVs(tt.old, tt.new) + _, result := diffGVs(tt.old, tt.new) if result != tt.want { - t.Errorf("detectDiff() = %v, want %v", result, tt.want) + t.Errorf("diffGVs() = %v, want %v", result, tt.want) } }) } From 4df467d42d103bb512441d250336ffd38fe12105 Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Sat, 24 Jan 2026 20:33:37 -0800 Subject: [PATCH 6/8] separate raw and filtered peer discovery caches --- .../util/peerproxy/gv_exclusion_manager.go | 119 +++++++++--------- .../peerproxy/gv_exclusion_manager_test.go | 96 +++++++++----- .../pkg/util/peerproxy/peer_discovery.go | 30 ++--- .../pkg/util/peerproxy/peer_discovery_test.go | 5 +- .../apiserver/pkg/util/peerproxy/peerproxy.go | 6 +- .../pkg/util/peerproxy/peerproxy_handler.go | 27 ++-- .../util/peerproxy/peerproxy_handler_test.go | 2 +- 7 files changed, 147 insertions(+), 138 deletions(-) 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 950a05f2635..9a1ede91767 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 @@ -66,22 +66,27 @@ type GVExclusionManager struct { // Worker 2: triggered by Active/Deleted GV changes refilterQueue workqueue.TypedRateLimitingInterface[string] - peerDiscoveryCache *atomic.Value // peerProxyHandler.peerDiscoveryInfoCache - invalidationCallback *atomic.Pointer[func()] + // rawPeerDiscoveryCache is written only by peerLeaseQueue worker + // when peer leases change + rawPeerDiscoveryCache *atomic.Value // map[string]PeerDiscoveryCacheEntry + // filteredPeerDiscoveryCache is written only by the refilter worker + // when raw cache or exclusion set changes. + filteredPeerDiscoveryCache atomic.Value // map[string]PeerDiscoveryCacheEntry + invalidationCallback *atomic.Pointer[func()] } // NewGVExclusionManager creates a new GV exclusion manager. func NewGVExclusionManager( exclusionGracePeriod time.Duration, reaperCheckInterval time.Duration, - peerDiscoveryCache *atomic.Value, + rawPeerDiscoveryCache *atomic.Value, invalidationCallback *atomic.Pointer[func()], ) *GVExclusionManager { mgr := &GVExclusionManager{ - exclusionGracePeriod: exclusionGracePeriod, - reaperCheckInterval: reaperCheckInterval, - peerDiscoveryCache: peerDiscoveryCache, - invalidationCallback: invalidationCallback, + exclusionGracePeriod: exclusionGracePeriod, + reaperCheckInterval: reaperCheckInterval, + rawPeerDiscoveryCache: rawPeerDiscoveryCache, + invalidationCallback: invalidationCallback, activeGVQueue: workqueue.NewTypedRateLimitingQueueWithConfig( workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{ @@ -96,10 +101,21 @@ func NewGVExclusionManager( mgr.currentlyActiveGVs.Store(map[schema.GroupVersion]struct{}{}) mgr.recentlyDeletedGVs.Store(map[schema.GroupVersion]time.Time{}) + mgr.filteredPeerDiscoveryCache.Store(map[string]PeerDiscoveryCacheEntry{}) return mgr } +// GetFilteredPeerDiscoveryCache returns the filtered peer discovery cache. +func (m *GVExclusionManager) GetFilteredPeerDiscoveryCache() map[string]PeerDiscoveryCacheEntry { + if cache := m.filteredPeerDiscoveryCache.Load(); cache != nil { + if cacheMap, ok := cache.(map[string]PeerDiscoveryCacheEntry); ok { + return cacheMap + } + } + return map[string]PeerDiscoveryCacheEntry{} +} + // RegisterCRDInformerHandlers registers event handlers for CRD informer using a custom extractor. // The extractor function is responsible for extracting GroupVersions from CRD objects. func (m *GVExclusionManager) RegisterCRDInformerHandlers(crdInformer cache.SharedIndexInformer, extractor GVExtractor) error { @@ -256,8 +272,7 @@ func (m *GVExclusionManager) reconcileActiveGVs() { m.currentlyActiveGVs.Store(freshGVs) klog.V(4).Infof("Active GVs updated: %d GVs now active", len(freshGVs)) } - // Queue re-filtering of peer discovery cache - m.refilterQueue.Add("refilter") + m.TriggerRefilter() } else { klog.V(4).Infof("No changes detected in active or recently deleted GVs") } @@ -373,83 +388,57 @@ func (m *GVExclusionManager) processNextRefilter(ctx context.Context) bool { return true } -// refilterPeerDiscoveryCache filters the peer discovery cache using the exclusion set. -// Only updates the cache and calls invalidation callback if filtering actually changed content. +// refilterPeerDiscoveryCache reads the raw peer discovery cache, +// applies exclusion filtering, and stores the result to the filtered cache. func (m *GVExclusionManager) refilterPeerDiscoveryCache() { - if m.peerDiscoveryCache == nil { - klog.Warning("peerDiscoveryCache reference not set") - return - } - - cache := m.peerDiscoveryCache.Load() - if cache == nil { - klog.V(4).Infof("Peer discovery cache is empty, skipping re-filter") - return - } - - cacheMap, ok := cache.(map[string]PeerDiscoveryCacheEntry) - if !ok { - klog.Warning("Invalid cache type in peerDiscoveryInfoCache") - return + var cacheMap map[string]PeerDiscoveryCacheEntry + if m.rawPeerDiscoveryCache != nil { + if rawCache := m.rawPeerDiscoveryCache.Load(); rawCache != nil { + cacheMap, _ = rawCache.(map[string]PeerDiscoveryCacheEntry) + } } if len(cacheMap) == 0 { - klog.V(4).Infof("Peer discovery cache is empty, skipping re-filter") + m.filteredPeerDiscoveryCache.Store(map[string]PeerDiscoveryCacheEntry{}) + klog.V(4).Infof("Raw peer discovery cache is empty or unavailable") return } - // Filter the cache - filteredCache, changed := m.FilterPeerDiscoveryCache(cacheMap) + filteredCache := m.filterPeerDiscoveryCache(cacheMap) + m.filteredPeerDiscoveryCache.Store(filteredCache) - if changed { - // Atomic swap peer cache - m.peerDiscoveryCache.Store(filteredCache) - - // Call invalidation callback - if m.invalidationCallback != nil { - if callback := m.invalidationCallback.Load(); callback != nil { - (*callback)() - } + if m.invalidationCallback != nil { + if callback := m.invalidationCallback.Load(); callback != nil { + (*callback)() } - - klog.V(4).Infof("Peer discovery cache re-filtered, %d GVs excluded", len(m.getExclusionSet())) - } else { - klog.V(4).Infof("No changes after filtering peer discovery cache") } + + klog.V(4).Infof("Peer discovery cache re-filtered, %d GVs in exclusion set", len(m.getExclusionSet())) + } -// FilterPeerDiscoveryCache applies the current exclusion set to the provided peer cache. -// This should be called when new peers are added to ensure their discovery is filtered. -// Returns the filtered cache and a boolean indicating if any changes were made. -func (m *GVExclusionManager) FilterPeerDiscoveryCache(cacheMap map[string]PeerDiscoveryCacheEntry) (map[string]PeerDiscoveryCacheEntry, bool) { +// filterPeerDiscoveryCache applies the current exclusion set to the provided peer cache. +func (m *GVExclusionManager) filterPeerDiscoveryCache(cacheMap map[string]PeerDiscoveryCacheEntry) map[string]PeerDiscoveryCacheEntry { exclusionSet := m.getExclusionSet() if len(exclusionSet) == 0 { - return cacheMap, false + return cacheMap } filtered := make(map[string]PeerDiscoveryCacheEntry, len(cacheMap)) - anyChanged := false - for peerID, entry := range cacheMap { - filteredEntry, changed := m.filterPeerCacheEntry(entry, exclusionSet) - filtered[peerID] = filteredEntry - if changed { - anyChanged = true - } + filtered[peerID] = m.filterPeerCacheEntry(entry, exclusionSet) } - return filtered, anyChanged + return filtered } // filterPeerCacheEntry filters a single peer's cache entry for excluded GVs. -// Returns the filtered entry and whether any GVs were excluded. func (m *GVExclusionManager) filterPeerCacheEntry( entry PeerDiscoveryCacheEntry, exclusionSet map[schema.GroupVersion]struct{}, -) (PeerDiscoveryCacheEntry, bool) { +) PeerDiscoveryCacheEntry { filteredGVRs := make(map[schema.GroupVersionResource]bool, len(entry.GVRs)) anyExcluded := false - for existingGVR, v := range entry.GVRs { gv := schema.GroupVersion{Group: existingGVR.Group, Version: existingGVR.Version} if _, excluded := exclusionSet[gv]; excluded { @@ -459,17 +448,17 @@ func (m *GVExclusionManager) filterPeerCacheEntry( filteredGVRs[existingGVR] = v } + // If no GVRs were excluded, the exclusion set doesn't intersect with this peer's GVs, + // so we can return the entry unchanged without filtering GroupDiscovery. if !anyExcluded { - return entry, false + return entry } - - // Filter the group discovery list filteredGroups := m.filterGroupDiscovery(entry.GroupDiscovery, exclusionSet) return PeerDiscoveryCacheEntry{ GVRs: filteredGVRs, GroupDiscovery: filteredGroups, - }, true + } } // filterGroupDiscovery filters group discovery entries, removing excluded GVs. @@ -517,3 +506,9 @@ func (m *GVExclusionManager) loadRecentlyDeletedGVs() map[schema.GroupVersion]ti } return map[schema.GroupVersion]time.Time{} } + +// TriggerRefilter triggers the refilter worker to apply exclusions to the filtered cache. +// This should be called by peerLeaseQueue after updating the raw peer discovery cache. +func (m *GVExclusionManager) TriggerRefilter() { + m.refilterQueue.Add("refilter") +} 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 1e3ea220e32..85e921ccc45 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 @@ -184,16 +184,18 @@ func TestDetectDiff(t *testing.T) { _ = NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) tests := []struct { - name string - old map[schema.GroupVersion]struct{} - new map[schema.GroupVersion]struct{} - want bool + name string + old map[schema.GroupVersion]struct{} + new map[schema.GroupVersion]struct{} + wantChanged bool + wantDeleted []schema.GroupVersion }{ { - name: "both empty", - old: map[schema.GroupVersion]struct{}{}, - new: map[schema.GroupVersion]struct{}{}, - want: false, + name: "both empty", + old: map[schema.GroupVersion]struct{}{}, + new: map[schema.GroupVersion]struct{}{}, + wantChanged: false, + wantDeleted: nil, }, { name: "identical", @@ -203,7 +205,8 @@ func TestDetectDiff(t *testing.T) { new: map[schema.GroupVersion]struct{}{ {Group: "apps", Version: "v1"}: {}, }, - want: false, + wantChanged: false, + wantDeleted: nil, }, { name: "added GV", @@ -214,7 +217,8 @@ func TestDetectDiff(t *testing.T) { {Group: "apps", Version: "v1"}: {}, {Group: "batch", Version: "v1"}: {}, }, - want: true, + wantChanged: true, + wantDeleted: nil, // No deletions, only addition }, { name: "removed GV", @@ -225,7 +229,10 @@ func TestDetectDiff(t *testing.T) { new: map[schema.GroupVersion]struct{}{ {Group: "apps", Version: "v1"}: {}, }, - want: true, + wantChanged: true, + wantDeleted: []schema.GroupVersion{ + {Group: "batch", Version: "v1"}, + }, }, { name: "different GVs same size", @@ -235,15 +242,33 @@ func TestDetectDiff(t *testing.T) { new: map[schema.GroupVersion]struct{}{ {Group: "batch", Version: "v1"}: {}, }, - want: true, + wantChanged: true, + wantDeleted: []schema.GroupVersion{ + {Group: "apps", Version: "v1"}, + }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - _, result := diffGVs(tt.old, tt.new) - if result != tt.want { - t.Errorf("diffGVs() = %v, want %v", result, tt.want) + deletedGVs, changed := diffGVs(tt.old, tt.new) + if changed != tt.wantChanged { + t.Errorf("diffGVs() changed = %v, want %v", changed, tt.wantChanged) + } + if len(deletedGVs) != len(tt.wantDeleted) { + t.Errorf("diffGVs() deleted count = %d, want %d", len(deletedGVs), len(tt.wantDeleted)) + } + for _, wantGV := range tt.wantDeleted { + found := false + for _, gotGV := range deletedGVs { + if gotGV == wantGV { + found = true + break + } + } + if !found { + t.Errorf("diffGVs() missing deleted GV %v", wantGV) + } } }) } @@ -255,7 +280,6 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { activeGVs map[schema.GroupVersion]struct{} deletedGVs map[schema.GroupVersion]time.Time cacheMap map[string]PeerDiscoveryCacheEntry - wantChanged bool wantPeerGVRs map[string]int // peer name -> GVR count }{ { @@ -273,7 +297,6 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { }, }, }, - wantChanged: false, wantPeerGVRs: map[string]int{ "peer1": 1, "peer2": 1, @@ -296,7 +319,6 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { }, }, }, - wantChanged: true, wantPeerGVRs: map[string]int{ "peer1": 0, // apps/v1 filtered out "peer2": 1, // unchanged @@ -322,7 +344,6 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { }, }, }, - wantChanged: true, wantPeerGVRs: map[string]int{ "peer1": 0, // custom/v1alpha1 filtered out "peer2": 1, // unchanged @@ -347,7 +368,6 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { }, }, }, - wantChanged: true, wantPeerGVRs: map[string]int{ "peer1": 1, // apps/v1 and custom/v1alpha1 filtered out, batch/v1 remains }, @@ -365,10 +385,7 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { mgr.recentlyDeletedGVs.Store(tt.deletedGVs) } - filtered, changed := mgr.FilterPeerDiscoveryCache(tt.cacheMap) - if changed != tt.wantChanged { - t.Errorf("Want changed=%v, got %v", tt.wantChanged, changed) - } + filtered := mgr.filterPeerDiscoveryCache(tt.cacheMap) if len(filtered) != len(tt.cacheMap) { t.Errorf("Want %d peers in filtered cache, got %d", len(tt.cacheMap), len(filtered)) @@ -393,9 +410,30 @@ func TestFilterPeerCacheEntry(t *testing.T) { name string entry PeerDiscoveryCacheEntry exclusionSet map[schema.GroupVersion]struct{} - wantChanged bool wantGVRs []schema.GroupVersionResource }{ + { + name: "empty exclusion set returns entry unchanged", + entry: PeerDiscoveryCacheEntry{ + GVRs: map[schema.GroupVersionResource]bool{ + {Group: "apps", Version: "v1", Resource: "deployments"}: true, + {Group: "batch", Version: "v1", Resource: "jobs"}: true, + }, + GroupDiscovery: []apidiscoveryv2.APIGroupDiscovery{ + { + ObjectMeta: metav1.ObjectMeta{Name: "apps"}, + Versions: []apidiscoveryv2.APIVersionDiscovery{ + {Version: "v1"}, + }, + }, + }, + }, + exclusionSet: map[schema.GroupVersion]struct{}{}, + wantGVRs: []schema.GroupVersionResource{ + {Group: "apps", Version: "v1", Resource: "deployments"}, + {Group: "batch", Version: "v1", Resource: "jobs"}, + }, + }, { name: "filter GVRs and groups", entry: PeerDiscoveryCacheEntry{ @@ -422,7 +460,6 @@ func TestFilterPeerCacheEntry(t *testing.T) { exclusionSet: map[schema.GroupVersion]struct{}{ {Group: "apps", Version: "v1"}: {}, }, - wantChanged: true, wantGVRs: []schema.GroupVersionResource{ {Group: "batch", Version: "v1", Resource: "jobs"}, {Group: "custom", Version: "v1", Resource: "myresources"}, @@ -438,7 +475,6 @@ func TestFilterPeerCacheEntry(t *testing.T) { exclusionSet: map[schema.GroupVersion]struct{}{ {Group: "apps", Version: "v1"}: {}, }, - wantChanged: false, wantGVRs: []schema.GroupVersionResource{ {Group: "batch", Version: "v1", Resource: "jobs"}, }, @@ -449,11 +485,7 @@ func TestFilterPeerCacheEntry(t *testing.T) { t.Run(tt.name, func(t *testing.T) { mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) - filtered, changed := mgr.filterPeerCacheEntry(tt.entry, tt.exclusionSet) - - if changed != tt.wantChanged { - t.Errorf("Want changed=%v, got %v", tt.wantChanged, changed) - } + filtered := mgr.filterPeerCacheEntry(tt.entry, tt.exclusionSet) for _, gvr := range tt.wantGVRs { if _, found := filtered.GVRs[gvr]; !found { diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go index 31b2bbc037a..b7e9d362d69 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go @@ -115,21 +115,17 @@ func (h *peerProxyHandler) syncPeerDiscoveryCache(ctx context.Context) error { } } - // Apply current exclusion filter to new peer cache entries. - if filtered, changed := h.gvExclusionManager.FilterPeerDiscoveryCache(newCache); changed { - newCache = filtered - } - - h.storePeerDiscoveryCacheAndInvalidate(newCache) + // Store unfiltered data to raw cache and trigger refilter. + // The refilter worker (single writer to filtered cache) will apply exclusions. + h.storeRawPeerDiscoveryCacheAndTriggerRefilter(newCache) return fetchDiscoveryErr } -// storePeerDiscoveryCacheAndInvalidate stores the new peer discovery cache and always calls the invalidation callback if set. -func (h *peerProxyHandler) storePeerDiscoveryCacheAndInvalidate(newCache map[string]PeerDiscoveryCacheEntry) { - h.peerDiscoveryInfoCache.Store(newCache) - if callback := h.cacheInvalidationCallback.Load(); callback != nil { - (*callback)() - } +// storeRawPeerDiscoveryCacheAndTriggerRefilter stores the raw (unfiltered) peer discovery cache +// and immediately refilters to update the filtered cache. +func (h *peerProxyHandler) storeRawPeerDiscoveryCacheAndTriggerRefilter(newCache map[string]PeerDiscoveryCacheEntry) { + h.rawPeerDiscoveryCache.Store(newCache) + h.gvExclusionManager.TriggerRefilter() } func (h *peerProxyHandler) fetchNewDiscoveryFor(ctx context.Context, serverID string) (PeerDiscoveryCacheEntry, error) { @@ -285,14 +281,8 @@ func (h *peerProxyHandler) isValidPeerIdentityLease(obj interface{}) (*v1.Lease, func (h *peerProxyHandler) findServiceableByPeerFromPeerDiscoveryCache(gvr schema.GroupVersionResource) []string { var serviceableByIDs []string - cache := h.peerDiscoveryInfoCache.Load() - if cache == nil { - return serviceableByIDs - } - - cacheMap, ok := cache.(map[string]PeerDiscoveryCacheEntry) - if !ok { - klog.Warning("Invalid cache type in peerDiscoveryInfoCache") + cacheMap := h.gvExclusionManager.GetFilteredPeerDiscoveryCache() + if len(cacheMap) == 0 { return serviceableByIDs } diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery_test.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery_test.go index e15e4c314b9..c479fbcece3 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery_test.go @@ -192,6 +192,7 @@ func TestRunPeerDiscoveryCacheSync(t *testing.T) { } go h.RunPeerDiscoveryCacheSync(ctx, 1) + go h.RunPeerDiscoveryRefilter(ctx, 1) // Wait for initial cache update. initialCache := map[string]PeerDiscoveryCacheEntry{} @@ -204,7 +205,7 @@ func TestRunPeerDiscoveryCacheSync(t *testing.T) { return false, ctx.Err() default: } - gotCache := h.peerDiscoveryInfoCache.Load() + gotCache := h.gvExclusionManager.GetFilteredPeerDiscoveryCache() return assert.ObjectsAreEqual(initialCache, gotCache), nil }) if err != nil { @@ -251,7 +252,7 @@ func TestRunPeerDiscoveryCacheSync(t *testing.T) { return false, ctx.Err() default: } - gotCache := h.peerDiscoveryInfoCache.Load() + gotCache := h.gvExclusionManager.GetFilteredPeerDiscoveryCache() r := assert.ObjectsAreEqual(tt.wantCache, gotCache) return r, nil }) 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 b6dce3b1ac0..ca0a383c22d 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go @@ -138,7 +138,7 @@ func NewPeerProxyHandler( localDiscoveryInfoCache: atomic.Value{}, localDiscoveryCacheTicker: time.NewTicker(localDiscoveryRefreshInterval), localDiscoveryInfoCachePopulated: make(chan struct{}), - peerDiscoveryInfoCache: atomic.Value{}, + rawPeerDiscoveryCache: atomic.Value{}, peerLeaseQueue: workqueue.NewTypedRateLimitingQueueWithConfig( workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{ @@ -150,7 +150,7 @@ func NewPeerProxyHandler( h.gvExclusionManager = NewGVExclusionManager( defaultExclusionGracePeriod, defaultExclusionReaperInterval, - &h.peerDiscoveryInfoCache, + &h.rawPeerDiscoveryCache, &h.cacheInvalidationCallback, ) @@ -176,7 +176,7 @@ func NewPeerProxyHandler( discoveryClient.NoPeerDiscovery = true h.discoveryClient = discoveryClient h.localDiscoveryInfoCache.Store(map[schema.GroupVersionResource]bool{}) - h.peerDiscoveryInfoCache.Store(map[string]PeerDiscoveryCacheEntry{}) + h.rawPeerDiscoveryCache.Store(map[string]PeerDiscoveryCacheEntry{}) proxyTransport, err := transport.New(proxyClientConfig) if err != nil { 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 87b86de39e3..cd7227f220e 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 @@ -79,14 +79,12 @@ type peerProxyHandler struct { localDiscoveryCacheTicker *time.Ticker localDiscoveryInfoCachePopulated chan struct{} localDiscoveryInfoCachePopulatedOnce sync.Once - // Cache that stores resources and groups served by peer apiservers. - // The map is from string to PeerDiscoveryCacheEntry where the string - // is the serverID of the peer apiserver. - // Refreshed if a new apiserver identity lease is added, deleted or - // holderIndentity change is observed in the lease. - peerDiscoveryInfoCache atomic.Value // map[string]struct{GVRs map[schema.GroupVersionResource]bool; Groups []apidiscoveryv2.APIGroupDiscovery} - proxyTransport http.RoundTripper - // Worker queue that keeps the peerDiscoveryInfoCache up-to-date. + // rawPeerDiscoveryCache stores unfiltered resources and groups served by peer apiservers. + // The map is from string (serverID) to PeerDiscoveryCacheEntry. + // Written ONLY by peerLeaseQueue worker when peer leases change. + rawPeerDiscoveryCache atomic.Value // map[string]PeerDiscoveryCacheEntry + proxyTransport http.RoundTripper + // Worker queue that keeps the rawPeerDiscoveryCache up-to-date. peerLeaseQueue workqueue.TypedRateLimitingInterface[string] serializer runtime.NegotiatedSerializer cacheInvalidationCallback atomic.Pointer[func()] @@ -284,16 +282,9 @@ func (r *responder) Error(w http.ResponseWriter, req *http.Request, err error) { // Returns a map of serverID -> []apidiscoveryv2.APIGroupDiscovery served by peer servers func (h *peerProxyHandler) GetPeerResources() map[string][]apidiscoveryv2.APIGroupDiscovery { result := make(map[string][]apidiscoveryv2.APIGroupDiscovery) - - peerCache := h.peerDiscoveryInfoCache.Load() - if peerCache == nil { - klog.V(4).Infof("GetPeerResources: peer cache is nil") - return result - } - - cacheMap, ok := peerCache.(map[string]PeerDiscoveryCacheEntry) - if !ok { - klog.Warning("Invalid cache type in peerDiscoveryGVRCache") + cacheMap := h.gvExclusionManager.GetFilteredPeerDiscoveryCache() + if len(cacheMap) == 0 { + klog.V(4).Infof("GetPeerResources: peer cache is empty") return result } diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler_test.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler_test.go index 95983701142..afa4ca86ef4 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler_test.go @@ -387,7 +387,7 @@ func newFakePeerProxyHandler(informerFinishedSync bool, return nil, err } ppH.localDiscoveryInfoCache.Store(localCache) - ppH.peerDiscoveryInfoCache.Store(peerCache) + ppH.gvExclusionManager.filteredPeerDiscoveryCache.Store(peerCache) ppH.finishedSync.Store(informerFinishedSync) return ppH, nil From d03e9c87087debd9fe4940bb2a96100de88189cf Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Tue, 27 Jan 2026 13:23:00 -0800 Subject: [PATCH 7/8] Use single worker for ActiveGVTracker and Refliter workqueues --- pkg/controlplane/apiserver/server.go | 4 +-- .../util/peerproxy/gv_exclusion_manager.go | 25 ++++++++----------- .../pkg/util/peerproxy/peer_discovery.go | 8 +----- .../pkg/util/peerproxy/peer_discovery_test.go | 2 +- .../apiserver/pkg/util/peerproxy/peerproxy.go | 12 ++++----- 5 files changed, 21 insertions(+), 30 deletions(-) diff --git a/pkg/controlplane/apiserver/server.go b/pkg/controlplane/apiserver/server.go index d536a1dd59c..3382ff69eb9 100644 --- a/pkg/controlplane/apiserver/server.go +++ b/pkg/controlplane/apiserver/server.go @@ -227,9 +227,9 @@ func (c completedConfig) New(name string, delegationTarget genericapiserver.Dele // Run peer-discovery workers s.GenericAPIServer.AddPostStartHookOrDie("peer-discovery-workers", func(context genericapiserver.PostStartHookContext) error { go c.Extra.PeerProxy.RunPeerDiscoveryCacheSync(context, 1) - go c.Extra.PeerProxy.RunPeerDiscoveryActiveGVTracker(context, 1) + go c.Extra.PeerProxy.RunPeerDiscoveryActiveGVTracker(context) go c.Extra.PeerProxy.RunPeerDiscoveryReaper(context) - go c.Extra.PeerProxy.RunPeerDiscoveryRefilter(context, 1) + go c.Extra.PeerProxy.RunPeerDiscoveryRefilter(context) 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 9a1ede91767..0015345e0bb 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 @@ -193,16 +193,15 @@ func (m *GVExclusionManager) handleGVUpdate() { m.activeGVQueue.Add("sync") } -// RunPeerDiscoveryActiveGVTracker runs Worker 1: Active GV Tracker +// RunPeerDiscoveryActiveGVTracker runs the Active GV Tracker worker. // This worker is triggered by CRD/APIService events and -// rebuilds the active GV set and reaps expired GVs if indicated so. -func (m *GVExclusionManager) RunPeerDiscoveryActiveGVTracker(ctx context.Context, workers int) { +// rebuilds the active GV set and reaps expired GVs. +// Only a single worker is used to avoid race conditions on atomic store operations. +func (m *GVExclusionManager) RunPeerDiscoveryActiveGVTracker(ctx context.Context) { defer m.activeGVQueue.ShutDown() - klog.Infof("Starting %d Active GV Tracker worker(s)", workers) - for i := 0; i < workers; i++ { - go wait.UntilWithContext(ctx, m.runActiveGVTrackerWorker, time.Second) - } + klog.Info("Starting Active GV Tracker worker") + go wait.UntilWithContext(ctx, m.runActiveGVTrackerWorker, time.Second) <-ctx.Done() klog.Info("Active GV Tracker workers stopped") @@ -344,7 +343,7 @@ func (m *GVExclusionManager) RunPeerDiscoveryReaper(ctx context.Context) { select { case <-ticker.C: // Trigger reconciliation which will also reap expired GVs - m.activeGVQueue.Add("reap") + m.activeGVQueue.Add("sync") case <-ctx.Done(): klog.Info("GV Reaper stopped") return @@ -352,15 +351,13 @@ func (m *GVExclusionManager) RunPeerDiscoveryReaper(ctx context.Context) { } } -// RunPeerDiscoveryRefilter runs Worker 2: Peer Discovery Re-filter +// RunPeerDiscoveryRefilter runs the Peer Discovery Re-filter worker. // This worker filters the peer discovery cache using the exclusion set. -func (m *GVExclusionManager) RunPeerDiscoveryRefilter(ctx context.Context, workers int) { +func (m *GVExclusionManager) RunPeerDiscoveryRefilter(ctx context.Context) { defer m.refilterQueue.ShutDown() - klog.Infof("Starting %d Peer Discovery Re-filter worker(s)", workers) - for i := 0; i < workers; i++ { - go wait.UntilWithContext(ctx, m.runRefilterWorker, time.Second) - } + klog.Info("Starting Peer Discovery Re-filter worker") + go wait.UntilWithContext(ctx, m.runRefilterWorker, time.Second) <-ctx.Done() klog.Info("Peer Discovery Re-filter workers stopped") diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go index b7e9d362d69..041377d0817 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery.go @@ -117,15 +117,9 @@ func (h *peerProxyHandler) syncPeerDiscoveryCache(ctx context.Context) error { // Store unfiltered data to raw cache and trigger refilter. // The refilter worker (single writer to filtered cache) will apply exclusions. - h.storeRawPeerDiscoveryCacheAndTriggerRefilter(newCache) - return fetchDiscoveryErr -} - -// storeRawPeerDiscoveryCacheAndTriggerRefilter stores the raw (unfiltered) peer discovery cache -// and immediately refilters to update the filtered cache. -func (h *peerProxyHandler) storeRawPeerDiscoveryCacheAndTriggerRefilter(newCache map[string]PeerDiscoveryCacheEntry) { h.rawPeerDiscoveryCache.Store(newCache) h.gvExclusionManager.TriggerRefilter() + return fetchDiscoveryErr } func (h *peerProxyHandler) fetchNewDiscoveryFor(ctx context.Context, serverID string) (PeerDiscoveryCacheEntry, error) { diff --git a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery_test.go b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery_test.go index c479fbcece3..a275e112ff6 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peer_discovery_test.go @@ -192,7 +192,7 @@ func TestRunPeerDiscoveryCacheSync(t *testing.T) { } go h.RunPeerDiscoveryCacheSync(ctx, 1) - go h.RunPeerDiscoveryRefilter(ctx, 1) + go h.RunPeerDiscoveryRefilter(ctx) // Wait for initial cache update. initialCache := map[string]PeerDiscoveryCacheEntry{} 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 ca0a383c22d..bc3a44c954b 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go @@ -101,7 +101,7 @@ type Interface interface { // 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) + RunPeerDiscoveryActiveGVTracker(ctx context.Context) // RunPeerDiscoveryReaper starts a background worker that periodically removes expired // GroupVersions from the exclusion set. When a CRD/APIService is deleted, its GV remains @@ -115,7 +115,7 @@ type Interface interface { // 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) + RunPeerDiscoveryRefilter(ctx context.Context) } // New creates a new instance to implement unknown version proxy @@ -229,9 +229,9 @@ func (h *peerProxyHandler) RegisterAPIServiceInformerHandlers(apiServiceInformer } // RunPeerDiscoveryActiveGVTracker starts the worker that tracks active GVs from CRDs/APIServices. -func (h *peerProxyHandler) RunPeerDiscoveryActiveGVTracker(ctx context.Context, workers int) { +func (h *peerProxyHandler) RunPeerDiscoveryActiveGVTracker(ctx context.Context) { if h.gvExclusionManager != nil { - h.gvExclusionManager.RunPeerDiscoveryActiveGVTracker(ctx, workers) + h.gvExclusionManager.RunPeerDiscoveryActiveGVTracker(ctx) } } @@ -243,8 +243,8 @@ func (h *peerProxyHandler) RunPeerDiscoveryReaper(ctx context.Context) { } // RunPeerDiscoveryRefilter starts the worker that refilters peer discovery cache. -func (h *peerProxyHandler) RunPeerDiscoveryRefilter(ctx context.Context, workers int) { +func (h *peerProxyHandler) RunPeerDiscoveryRefilter(ctx context.Context) { if h.gvExclusionManager != nil { - h.gvExclusionManager.RunPeerDiscoveryRefilter(ctx, workers) + h.gvExclusionManager.RunPeerDiscoveryRefilter(ctx) } } From f4882eeaa63b8df9e47710ca02ac228a593e13f2 Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Sun, 8 Feb 2026 22:03:00 -0800 Subject: [PATCH 8/8] Use AddAfter for GV reaping instead of periodic ticker --- pkg/controlplane/apiserver/server.go | 1 - .../util/peerproxy/gv_exclusion_manager.go | 41 +++++-------------- .../peerproxy/gv_exclusion_manager_test.go | 24 ++++------- .../apiserver/pkg/util/peerproxy/peerproxy.go | 21 +--------- 4 files changed, 22 insertions(+), 65 deletions(-) diff --git a/pkg/controlplane/apiserver/server.go b/pkg/controlplane/apiserver/server.go index 3382ff69eb9..5e1bc1fc122 100644 --- a/pkg/controlplane/apiserver/server.go +++ b/pkg/controlplane/apiserver/server.go @@ -228,7 +228,6 @@ func (c completedConfig) New(name string, delegationTarget genericapiserver.Dele s.GenericAPIServer.AddPostStartHookOrDie("peer-discovery-workers", func(context genericapiserver.PostStartHookContext) error { go c.Extra.PeerProxy.RunPeerDiscoveryCacheSync(context, 1) go c.Extra.PeerProxy.RunPeerDiscoveryActiveGVTracker(context) - go c.Extra.PeerProxy.RunPeerDiscoveryReaper(context) go c.Extra.PeerProxy.RunPeerDiscoveryRefilter(context) 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 0015345e0bb..e7fb0071031 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 @@ -42,11 +42,11 @@ type GVExtractor func(obj interface{}) []schema.GroupVersion // - recentlyDeletedGVs: GVs belonging to CRDs and aggregated APIServices that were recently deleted, // tracked with deletion timestamp for grace period // -// It runs two workers and a periodic ticker: -// 1. Active GV Tracker: Triggered on CRD/APIService events or reaper ticks, -// rebuilds active GVs and reaps expired deleted GVs +// It runs two workers: +// 1. Active GV Tracker: Triggered on CRD/APIService events, rebuilds active GVs +// and reaps expired deleted GVs. When a GV is deleted, a delayed sync is scheduled +// after the grace period to reap it. // 2. Peer Discovery Re-filter: Rate-limited worker that filters peer cache -// 3. Reaper Ticker: Periodically triggers the Active GV Tracker to reap expired GVs type GVExclusionManager struct { // Atomic maps for lock-free access currentlyActiveGVs atomic.Value // map[schema.GroupVersion]struct{} @@ -58,11 +58,10 @@ type GVExclusionManager struct { apiServiceInformer cache.SharedIndexInformer apiServiceExtractor GVExtractor - // Worker 1: triggered by CRD/APIService events or reaper ticks + // Worker 1: triggered by CRD/APIService events or delayed reap scheduling activeGVQueue workqueue.TypedRateLimitingInterface[string] - // Reaper ticker configuration + // Grace period before reaping deleted GVs from the exclusion set exclusionGracePeriod time.Duration - reaperCheckInterval time.Duration // Worker 2: triggered by Active/Deleted GV changes refilterQueue workqueue.TypedRateLimitingInterface[string] @@ -78,13 +77,11 @@ type GVExclusionManager struct { // NewGVExclusionManager creates a new GV exclusion manager. func NewGVExclusionManager( exclusionGracePeriod time.Duration, - reaperCheckInterval time.Duration, rawPeerDiscoveryCache *atomic.Value, invalidationCallback *atomic.Pointer[func()], ) *GVExclusionManager { mgr := &GVExclusionManager{ exclusionGracePeriod: exclusionGracePeriod, - reaperCheckInterval: reaperCheckInterval, rawPeerDiscoveryCache: rawPeerDiscoveryCache, invalidationCallback: invalidationCallback, activeGVQueue: workqueue.NewTypedRateLimitingQueueWithConfig( @@ -196,7 +193,6 @@ func (m *GVExclusionManager) handleGVUpdate() { // RunPeerDiscoveryActiveGVTracker runs the Active GV Tracker worker. // This worker is triggered by CRD/APIService events and // rebuilds the active GV set and reaps expired GVs. -// Only a single worker is used to avoid race conditions on atomic store operations. func (m *GVExclusionManager) RunPeerDiscoveryActiveGVTracker(ctx context.Context) { defer m.activeGVQueue.ShutDown() @@ -301,7 +297,11 @@ func (m *GVExclusionManager) updateRecentlyDeletedGVs(deletedGVs []schema.GroupV } } - // Add newly deleted GVs + // Schedule a delayed sync to reap expired GVs after the grace period. + if len(deletedGVs) > 0 { + m.activeGVQueue.AddAfter("sync", m.exclusionGracePeriod) + } + for _, gv := range deletedGVs { newDeletedMap[gv] = now klog.V(4).Infof("GV %s deleted: moved to recentlyDeletedGVs", gv.String()) @@ -332,25 +332,6 @@ func diffGVs(old, new map[schema.GroupVersion]struct{}) ([]schema.GroupVersion, return deletedGVs, hasChanges } -// RunPeerDiscoveryReaper runs Worker 2: Reaper -// This worker periodically triggers reconciliation which also reaps expired GVs. -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() - - for { - select { - case <-ticker.C: - // Trigger reconciliation which will also reap expired GVs - m.activeGVQueue.Add("sync") - case <-ctx.Done(): - klog.Info("GV Reaper stopped") - return - } - } -} - // RunPeerDiscoveryRefilter runs the Peer Discovery Re-filter worker. // This worker filters the peer discovery cache using the exclusion set. func (m *GVExclusionManager) RunPeerDiscoveryRefilter(ctx context.Context) { 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 85e921ccc45..bc8125bf20e 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 @@ -17,6 +17,7 @@ limitations under the License. package peerproxy import ( + "slices" "sync/atomic" "testing" "time" @@ -102,7 +103,7 @@ func TestGetExclusionSet(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + mgr := NewGVExclusionManager(5*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) if tt.activeGVs != nil { mgr.currentlyActiveGVs.Store(tt.activeGVs) @@ -166,7 +167,7 @@ func TestReapExpiredGVs(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - mgr := NewGVExclusionManager(tt.gracePeriod, 50*time.Millisecond, &atomic.Value{}, &atomic.Pointer[func()]{}) + mgr := NewGVExclusionManager(tt.gracePeriod, &atomic.Value{}, &atomic.Pointer[func()]{}) mgr.recentlyDeletedGVs.Store(tt.deletedGVs) mgr.updateRecentlyDeletedGVs(nil) result := mgr.loadRecentlyDeletedGVs() @@ -181,7 +182,7 @@ func TestReapExpiredGVs(t *testing.T) { } func TestDetectDiff(t *testing.T) { - _ = NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + _ = NewGVExclusionManager(5*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) tests := []struct { name string @@ -259,15 +260,8 @@ func TestDetectDiff(t *testing.T) { t.Errorf("diffGVs() deleted count = %d, want %d", len(deletedGVs), len(tt.wantDeleted)) } for _, wantGV := range tt.wantDeleted { - found := false - for _, gotGV := range deletedGVs { - if gotGV == wantGV { - found = true - break - } - } - if !found { - t.Errorf("diffGVs() missing deleted GV %v", wantGV) + if !slices.Contains(deletedGVs, wantGV) { + t.Errorf("diffGVs() deleted missing GV %v", wantGV) } } }) @@ -376,7 +370,7 @@ func TestFilterPeerDiscoveryCache(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + mgr := NewGVExclusionManager(5*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) if tt.activeGVs != nil { mgr.currentlyActiveGVs.Store(tt.activeGVs) @@ -483,7 +477,7 @@ func TestFilterPeerCacheEntry(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + mgr := NewGVExclusionManager(5*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) filtered := mgr.filterPeerCacheEntry(tt.entry, tt.exclusionSet) @@ -555,7 +549,7 @@ func TestFilterGroupDiscovery(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - mgr := NewGVExclusionManager(5*time.Minute, 1*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) + mgr := NewGVExclusionManager(5*time.Minute, &atomic.Value{}, &atomic.Pointer[func()]{}) filtered := mgr.filterGroupDiscovery(tt.groupDiscoveries, tt.exclusionSet) for groupName, wantVersionStrs := range tt.wantGroups { 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 bc3a44c954b..9644be4cfd0 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go +++ b/staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go @@ -55,9 +55,6 @@ const ( // the deleted CRD or aggregated API before this server stops excluding it // in peer-aggregated discovery and while proxying requests to peers. defaultExclusionGracePeriod = 5 * time.Minute - // defaultExclusionReaperInterval is the interval at which the we - // clean up deleted groups from the exclusion list. - defaultExclusionReaperInterval = 1 * time.Minute ) // Interface defines how the Mixed Version Proxy filter interacts with the underlying system. @@ -100,16 +97,10 @@ type Interface interface { // 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. + // set accordingly. When a GV is deleted, a delayed sync is automatically scheduled + // after the grace period to reap expired GVs from the exclusion set. RunPeerDiscoveryActiveGVTracker(ctx context.Context) - // 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 @@ -149,7 +140,6 @@ func NewPeerProxyHandler( h.gvExclusionManager = NewGVExclusionManager( defaultExclusionGracePeriod, - defaultExclusionReaperInterval, &h.rawPeerDiscoveryCache, &h.cacheInvalidationCallback, ) @@ -235,13 +225,6 @@ func (h *peerProxyHandler) RunPeerDiscoveryActiveGVTracker(ctx context.Context) } } -// 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) { if h.gvExclusionManager != nil {