Merge pull request #2672 from rexagod/sync-caches

fix: Clean up CR reflectors after parent CRD deletion

Author: k8s-ci-robot

internal/discovery/discovery.go

@@ -64,8 +64,8 @@ func (r *CRDiscoverer) StartDiscovery(ctx context.Context, config *rest.Config)
 					},
 					Plural: p,
 				}
-				r.AppendToMap(gotGVKP)
 				r.SafeWrite(func() {
+					r.AppendToMap(gotGVKP)
 					r.WasUpdated = true
 				})
 			}
@@ -89,8 +89,8 @@ func (r *CRDiscoverer) StartDiscovery(ctx context.Context, config *rest.Config)
 					},
 					Plural: p,
 				}
-				r.RemoveFromMap(gotGVKP)
 				r.SafeWrite(func() {
+					r.RemoveFromMap(gotGVKP)
 					r.WasUpdated = true
 				})
 			}

internal/discovery/types.go

@@ -45,6 +45,8 @@ type CRDiscoverer struct {
 	CRDsCacheCountGauge prometheus.Gauge
 	// Map is a cache of the collected GVKs.
 	Map map[string]map[string][]kindPlural
+	// GVKToReflectorStopChanMap is a map of GVKs to channels that can be used to stop their corresponding reflector.
+	GVKToReflectorStopChanMap map[string]chan struct{}
 	// m is a mutex to protect the cache.
 	m sync.RWMutex
 	// ShouldUpdate is a flag that indicates whether the cache was updated.
@@ -70,6 +72,9 @@ func (r *CRDiscoverer) AppendToMap(gvkps ...groupVersionKindPlural) {
 	if r.Map == nil {
 		r.Map = map[string]map[string][]kindPlural{}
 	}
+	if r.GVKToReflectorStopChanMap == nil {
+		r.GVKToReflectorStopChanMap = map[string]chan struct{}{}
+	}
 	for _, gvkp := range gvkps {
 		if _, ok := r.Map[gvkp.Group]; !ok {
 			r.Map[gvkp.Group] = map[string][]kindPlural{}
@@ -78,6 +83,7 @@ func (r *CRDiscoverer) AppendToMap(gvkps ...groupVersionKindPlural) {
 			r.Map[gvkp.Group][gvkp.Version] = []kindPlural{}
 		}
 		r.Map[gvkp.Group][gvkp.Version] = append(r.Map[gvkp.Group][gvkp.Version], kindPlural{Kind: gvkp.Kind, Plural: gvkp.Plural})
+		r.GVKToReflectorStopChanMap[gvkp.GroupVersionKind.String()] = make(chan struct{})
 	}
 }
 
@@ -92,6 +98,8 @@ func (r *CRDiscoverer) RemoveFromMap(gvkps ...groupVersionKindPlural) {
 		}
 		for i, el := range r.Map[gvkp.Group][gvkp.Version] {
 			if el.Kind == gvkp.Kind {
+				close(r.GVKToReflectorStopChanMap[gvkp.GroupVersionKind.String()])
+				delete(r.GVKToReflectorStopChanMap, gvkp.GroupVersionKind.String())
 				if len(r.Map[gvkp.Group][gvkp.Version]) == 1 {
 					delete(r.Map[gvkp.Group], gvkp.Version)
 					if len(r.Map[gvkp.Group]) == 0 {

internal/store/builder.go

@@ -38,6 +38,7 @@ import (
 	policyv1 "k8s.io/api/policy/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	storagev1 "k8s.io/api/storage/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
@@ -84,6 +85,8 @@ type Builder struct {
 	shard               int32
 	useAPIServerCache   bool
 	objectLimit         int64
+
+	GVKToReflectorStopChanMap *map[string]chan struct{}
 }
 
 // NewBuilder returns a new builder.
@@ -617,7 +620,11 @@ func (b *Builder) startReflector(
 ) {
 	instrumentedListWatch := watch.NewInstrumentedListerWatcher(listWatcher, b.listWatchMetrics, reflect.TypeOf(expectedType).String(), useAPIServerCache, objectLimit)
 	reflector := cache.NewReflectorWithOptions(sharding.NewShardedListWatch(b.shard, b.totalShards, instrumentedListWatch), expectedType, store, cache.ReflectorOptions{ResyncPeriod: 0})
-	go reflector.Run(b.ctx.Done())
+	if cr, ok := expectedType.(*unstructured.Unstructured); ok {
+		go reflector.Run((*b.GVKToReflectorStopChanMap)[cr.GroupVersionKind().String()])
+	} else {
+		go reflector.Run(b.ctx.Done())
+	}
 }
 
 // cacheStoresToMetricStores converts []cache.Store into []*metricsstore.MetricsStore

pkg/app/server.go

@@ -296,6 +296,8 @@ func RunKubeStateMetrics(ctx context.Context, opts *options.Options) error {
 			CRDsDeleteEventsCounter: crdsDeleteEventsCounter,
 			CRDsCacheCountGauge:     crdsCacheCountGauge,
 		}
+		// storeBuilder starts reflectors for the discovered GVKs, and as such, should close them too.
+		storeBuilder.GVKToReflectorStopChanMap = &discovererInstance.GVKToReflectorStopChanMap
 		// This starts a goroutine that will watch for any new GVKs to extract from CRDs.
 		err = discovererInstance.StartDiscovery(ctx, kubeConfig)
 		if err != nil {