From e0dedfbc230b420ed42106196ef5b44a60bd2a54 Mon Sep 17 00:00:00 2001 From: Jefftree Date: Wed, 11 Mar 2026 00:40:59 -0400 Subject: [PATCH 1/2] Fix aggregated discovery retry on fetch failure --- .../pkg/apiserver/handler_discovery.go | 42 +++++++++++-------- 1 file changed, 25 insertions(+), 17 deletions(-) diff --git a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery.go b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery.go index b6ad297a9c6..e9c516ca6f4 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery.go @@ -33,7 +33,6 @@ import ( "k8s.io/apimachinery/pkg/runtime/serializer" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/sets" - "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apiserver/pkg/authentication/user" "k8s.io/apiserver/pkg/endpoints" discoveryendpoint "k8s.io/apiserver/pkg/endpoints/discovery/aggregated" @@ -53,6 +52,12 @@ var APIRegistrationGroupVersion metav1.GroupVersion = metav1.GroupVersion{Group: // first (mirrors v1 discovery behavior) var APIRegistrationGroupPriority int = 20001 +// discoveryRefreshInterval is the interval at which the discovery manager +// re-fetches discovery documents from aggregated API servers to detect changes. +// This also caps the exponential backoff for failed fetches, so retries never +// wait longer than the normal refresh period. +const discoveryRefreshInterval = 1 * time.Minute + // Aggregated discovery content-type GVK. var v2Beta1GVK = schema.GroupVersionKind{ Group: "apidiscovery.k8s.io", @@ -199,7 +204,7 @@ func NewDiscoveryManager( apiServices: make(map[string]groupVersionInfo), cachedResults: make(map[serviceKey]cachedResult), dirtyAPIServiceQueue: workqueue.NewTypedRateLimitingQueueWithConfig( - workqueue.DefaultTypedControllerRateLimiter[string](), + workqueue.NewTypedItemExponentialFailureRateLimiter[string](500*time.Millisecond, discoveryRefreshInterval), workqueue.TypedRateLimitingQueueConfig[string]{Name: "discovery-manager"}, ), codecs: codecs, @@ -432,7 +437,10 @@ func (dm *discoveryManager) syncAPIService(apiServiceName string) error { dm.mergedDiscoveryHandler.AddGroupVersion(gv.Group, entry) dm.mergedDiscoveryHandler.SetGroupVersionPriority(metav1.GroupVersion(gv), info.groupPriority, info.versionPriority) - return nil + // Return the error so the worker retries with backoff. The stale entry + // is already added above, so discovery reflects the current state while + // we wait for the service to become reachable. + return err } func (dm *discoveryManager) getAPIServiceKeys() []string { @@ -496,26 +504,17 @@ func (dm *discoveryManager) Run(stopCh <-chan struct{}, discoverySyncedCh chan<- dm.dirtyAPIServiceQueue.AddRateLimited(next) } else { dm.dirtyAPIServiceQueue.Forget(next) + // Re-enqueue after the refresh interval to pick up changes + // in the aggregated server's discovery document. + dm.markDirty(next) + dm.dirtyAPIServiceQueue.AddAfter(next, discoveryRefreshInterval) } }() } }() } - wait.PollUntil(1*time.Minute, func() (done bool, err error) { - dm.servicesLock.Lock() - defer dm.servicesLock.Unlock() - - now := time.Now() - - // Mark all non-local APIServices as dirty - for key, info := range dm.apiServices { - info.lastMarkedDirty = now - dm.apiServices[key] = info - dm.dirtyAPIServiceQueue.Add(key) - } - return false, nil - }, stopCh) + <-stopCh } // Takes a snapshot of all currently used services by known APIServices and @@ -604,6 +603,15 @@ func (dm *discoveryManager) getInfoForAPIService(name string) (groupVersionInfo, return result, ok } +func (dm *discoveryManager) markDirty(name string) { + dm.servicesLock.Lock() + defer dm.servicesLock.Unlock() + if info, exists := dm.apiServices[name]; exists { + info.lastMarkedDirty = time.Now() + dm.apiServices[name] = info + } +} + func (dm *discoveryManager) setInfoForAPIService(name string, result *groupVersionInfo) (oldValueIfExisted *groupVersionInfo) { dm.servicesLock.Lock() defer dm.servicesLock.Unlock() From 9fcdecad9feff782ca80f925974231ca81ef3935 Mon Sep 17 00:00:00 2001 From: Jefftree Date: Wed, 11 Mar 2026 09:58:41 -0400 Subject: [PATCH 2/2] Fix TestBasic timeout caused by continuous re-enqueue --- .../kube-aggregator/pkg/apiserver/handler_discovery_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery_test.go b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery_test.go index bf2f23e3228..23cc27f6c04 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery_test.go +++ b/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery_test.go @@ -1102,7 +1102,9 @@ func (q *completerWorkqueue) Add(item string) { } func (q *completerWorkqueue) AddAfter(item string, duration time.Duration) { - q.Add(item) + // Normally apiservices are continuously re-synced and never leave the + // queue. Make this a no-op in the unit test so that isComplete() + // reflects whether all explicitly dirtied items have been processed. } func (q *completerWorkqueue) AddRateLimited(item string) {