refactor: initialize default factory store and reset in KubeEventsManager for consistent informer management

Signed-off-by: Evsyukov Denis <denis.evsyukov@flant.com>

Author: Evsyukov Denis

pkg/kube_events_manager/factory.go

@@ -6,16 +6,24 @@ import (
 	"sync"
 	"time"
 
-	"github.com/deckhouse/deckhouse/pkg/log"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/tools/cache"
+
+	"github.com/deckhouse/deckhouse/pkg/log"
+)
+
+var (
+	DefaultFactoryStore *FactoryStore
+	DefaultSyncTime     = 100 * time.Millisecond
 )
 
-var DefaultSyncTime = 100 * time.Millisecond
+func init() {
+	DefaultFactoryStore = NewFactoryStore()
+}
 
 type FactoryIndex struct {
 	GVR           schema.GroupVersionResource
@@ -42,6 +50,12 @@ func NewFactoryStore() *FactoryStore {
 	}
 }
 
+func (c *FactoryStore) Reset() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.data = make(map[FactoryIndex]Factory)
+}
+
 func (c *FactoryStore) add(index FactoryIndex, f dynamicinformer.DynamicSharedInformerFactory) {
 	ctx, cancel := context.WithCancel(context.Background())
 	c.data[index] = Factory{

pkg/kube_events_manager/resource_informer.go

@@ -37,7 +37,6 @@ type resourceInformer struct {
 	GroupVersionResource schema.GroupVersionResource
 	ListOptions          metav1.ListOptions
 
-	defaultFactoryStore *FactoryStore
 	// A cache of objects and filterResults. It is a part of the Monitor's snapshot.
 	cachedObjects map[string]*kemtypes.ObjectAndFilterResult
 	cacheLock     sync.RWMutex
@@ -92,7 +91,6 @@ func newResourceInformer(ns, name string, cfg *resourceInformerConfig) *resource
 		cachedObjectsInfo:      &CachedObjectsInfo{},
 		cachedObjectsIncrement: &CachedObjectsInfo{},
 		logger:                 cfg.logger,
-		defaultFactoryStore:    NewFactoryStore(),
 	}
 	return informer
 }
@@ -457,13 +455,13 @@ func (ei *resourceInformer) start() {
 	go func() {
 		if ei.ctx != nil {
 			<-ei.ctx.Done()
-			ei.defaultFactoryStore.Stop(ei.id, ei.FactoryIndex)
+			DefaultFactoryStore.Stop(ei.id, ei.FactoryIndex)
 		}
 	}()
 
 	// TODO: separate handler and informer
 	errorHandler := newWatchErrorHandler(ei.Monitor.Metadata.DebugName, ei.Monitor.Kind, ei.Monitor.Metadata.LogLabels, ei.metricStorage, ei.logger.Named("watch-error-handler"))
-	err := ei.defaultFactoryStore.Start(ei.ctx, ei.id, ei.KubeClient.Dynamic(), ei.FactoryIndex, ei, errorHandler)
+	err := DefaultFactoryStore.Start(ei.ctx, ei.id, ei.KubeClient.Dynamic(), ei.FactoryIndex, ei, errorHandler)
 	if err != nil {
 		ei.Monitor.Logger.Error("cache is not synced for informer", slog.String("debugName", ei.Monitor.Metadata.DebugName))
 		return

test/hook/context/generator.go

@@ -67,6 +67,8 @@ func NewBindingContextController(config string, logger *log.Logger, version ...f
 
 	b.KubeEventsManager = kubeeventsmanager.NewKubeEventsManager(ctx, b.fakeCluster.Client, b.logger.Named("kube-events-manager"))
 	b.KubeEventsManager.WithMetricStorage(metricstorage.NewMetricStorage(ctx, "metrics-prefix", false, log.NewNop()))
+	// Re-create factory to drop informers created using different b.fakeCluster.Client.
+	kubeeventsmanager.DefaultFactoryStore.Reset()
 
 	b.ScheduleManager = schedulemanager.NewScheduleManager(ctx, b.logger.Named("schedule-manager"))