github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 11:28:58 +00:00
Code Issues Projects Releases Wiki Activity

Wire GVExclusionManager

Browse Source
This commit is contained in:
Richa Banker
2025-11-19 20:28:05 -08:00
parent 79375cf7de
commit 1713b6d8f2
5 changed files with 142 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
pkg/controlplane/apiserver/server.go
View File

@@ -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
})
}

10
staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager.go
View File

@@ -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
}

26
staging/src/k8s.io/apiserver/pkg/util/peerproxy/gv_exclusion_manager_test.go
View File

@@ -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"}: {},

101
staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy.go
View File

@@ -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)
}
}

31
staging/src/k8s.io/apiserver/pkg/util/peerproxy/peerproxy_handler.go
View File

@@ -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.