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

Kubernetes-commit: c72f9f73d92ff5e7a28f06c110acbc9626e4252d

Author: richabanker

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,16 +145,28 @@ 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 {
 	GroupPriorityMinimum int
 	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,15 +207,15 @@ 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) {
 	rdm.lock.Lock()
 	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,53 +540,62 @@ 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
 	if mediaType.Convert == nil ||
 		(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,
 		)
 	}