From c72f9f73d92ff5e7a28f06c110acbc9626e4252d Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Wed, 5 Nov 2025 21:14:26 -0800 Subject: [PATCH] discovery/aggregated: Add invalidation callback support Add callback mechanism to notify peer-aggregated discovery handler when local discovery changes, enabling efficient cache invalidation. Part of KEP-4020: Unknown Version Interoperability Proxy --- .../endpoints/discovery/aggregated/handler.go | 64 +++++++++++++------ 1 file changed, 46 insertions(+), 18 deletions(-) diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/discovery/aggregated/handler.go b/staging/src/k8s.io/apiserver/pkg/endpoints/discovery/aggregated/handler.go index 331091e4a5b..5e325ba447d 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/discovery/aggregated/handler.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/discovery/aggregated/handler.go @@ -85,6 +85,9 @@ type ResourceManager interface { // The group from the least-numbered source is used WithSource(source Source) ResourceManager + // AddInvalidationCallback adds a callback to be called when the discovery cache is invalidated. + AddInvalidationCallback(callback func()) + http.Handler } @@ -116,6 +119,10 @@ func (rm resourceManager) WithSource(source Source) ResourceManager { } } +func (rm resourceManager) AddInvalidationCallback(callback func()) { + rm.resourceDiscoveryManager.AddInvalidationCallback(callback) +} + type groupKey struct { name string @@ -138,9 +145,10 @@ type resourceDiscoveryManager struct { // Writes protected by the lock. // List of all apigroups & resources indexed by the resource manager - lock sync.RWMutex - apiGroups map[groupKey]*apidiscoveryv2.APIGroupDiscovery - versionPriorities map[groupVersionKey]priorityInfo + lock sync.RWMutex + apiGroups map[groupKey]*apidiscoveryv2.APIGroupDiscovery + versionPriorities map[groupVersionKey]priorityInfo + invalidationCallback atomic.Pointer[func()] } type priorityInfo struct { @@ -148,6 +156,17 @@ type priorityInfo struct { VersionPriority int } +func (rdm *resourceDiscoveryManager) AddInvalidationCallback(callback func()) { + rdm.invalidationCallback.Store(&callback) +} + +func (rdm *resourceDiscoveryManager) invalidateCacheLocked() { + rdm.cache.Store(nil) + if callback := rdm.invalidationCallback.Load(); callback != nil { + (*callback)() + } +} + func NewResourceManager(path string) ResourceManager { scheme := runtime.NewScheme() utilruntime.Must(apidiscoveryv2.AddToScheme(scheme)) @@ -188,7 +207,7 @@ func (rdm *resourceDiscoveryManager) SetGroupVersionPriority(source Source, gv m GroupPriorityMinimum: groupPriorityMinimum, VersionPriority: versionPriority, } - rdm.cache.Store(nil) + rdm.invalidateCacheLocked() } func (rdm *resourceDiscoveryManager) SetGroups(source Source, groups []apidiscoveryv2.APIGroupDiscovery) { @@ -196,7 +215,7 @@ func (rdm *resourceDiscoveryManager) SetGroups(source Source, groups []apidiscov defer rdm.lock.Unlock() rdm.apiGroups = nil - rdm.cache.Store(nil) + rdm.invalidateCacheLocked() for _, group := range groups { for _, version := range group.Versions { @@ -297,7 +316,7 @@ func (rdm *resourceDiscoveryManager) addGroupVersionLocked(source Source, groupN } // Reset response document so it is recreated lazily - rdm.cache.Store(nil) + rdm.invalidateCacheLocked() } func (rdm *resourceDiscoveryManager) RemoveGroupVersion(source Source, apiGroup metav1.GroupVersion) { @@ -338,7 +357,7 @@ func (rdm *resourceDiscoveryManager) RemoveGroupVersion(source Source, apiGroup } // Reset response document so it is recreated lazily - rdm.cache.Store(nil) + rdm.invalidateCacheLocked() } func (rdm *resourceDiscoveryManager) RemoveGroup(source Source, groupName string) { @@ -521,12 +540,22 @@ func (rdm *resourceDiscoveryManager) serveHTTP(resp http.ResponseWriter, req *ht response := cache.cachedResponse etag := cache.cachedResponseETag - mediaType, _, err := negotiation.NegotiateOutputMediaType(req, rdm.serializer, DiscoveryEndpointRestrictions) + writeDiscoveryResponse(&response, etag, rdm.serializer, resp, req) +} + +func writeDiscoveryResponse( + resp *apidiscoveryv2.APIGroupDiscoveryList, + etag string, + serializer runtime.NegotiatedSerializer, + w http.ResponseWriter, + req *http.Request, +) { + mediaType, _, err := negotiation.NegotiateOutputMediaType(req, serializer, DiscoveryEndpointRestrictions) if err != nil { // Should never happen. wrapper.go will only proxy requests to this // handler if the media type passes DiscoveryEndpointRestrictions utilruntime.HandleError(err) - resp.WriteHeader(http.StatusInternalServerError) + w.WriteHeader(http.StatusInternalServerError) return } var targetGV schema.GroupVersion @@ -534,40 +563,39 @@ func (rdm *resourceDiscoveryManager) serveHTTP(resp http.ResponseWriter, req *ht (mediaType.Convert.GroupVersion() != apidiscoveryv2.SchemeGroupVersion && mediaType.Convert.GroupVersion() != apidiscoveryv2beta1.SchemeGroupVersion) { utilruntime.HandleError(fmt.Errorf("expected aggregated discovery group version, got group: %s, version %s", mediaType.Convert.Group, mediaType.Convert.Version)) - resp.WriteHeader(http.StatusInternalServerError) + w.WriteHeader(http.StatusInternalServerError) return } if mediaType.Convert.GroupVersion() == apidiscoveryv2beta1.SchemeGroupVersion && utilfeature.DefaultFeatureGate.Enabled(genericfeatures.AggregatedDiscoveryRemoveBetaType) { klog.Errorf("aggregated discovery version v2beta1 is removed. Please update to use v2") - resp.WriteHeader(http.StatusNotFound) + w.WriteHeader(http.StatusNotFound) return } - targetGV = mediaType.Convert.GroupVersion() if len(etag) > 0 { // Use proper e-tag headers if one is available ServeHTTPWithETag( - &response, + resp, etag, targetGV, - rdm.serializer, - resp, + serializer, + w, req, ) } else { // Default to normal response in rare case etag is // not cached with the object for some reason. responsewriters.WriteObjectNegotiated( - rdm.serializer, + serializer, DiscoveryEndpointRestrictions, targetGV, - resp, + w, req, http.StatusOK, - &response, + resp, true, ) }