[shell-operator] fix: protect shared informer lifecycle

pkg/kube_events_manager/factory.go

@@ -40,15 +40,25 @@ type Factory struct {
 }
 
 type FactoryStore struct {
-	mu        sync.Mutex
-	data      map[FactoryIndex]*Factory
-	stoppedCh map[FactoryIndex]chan struct{}
+	mu   sync.Mutex
+	data map[FactoryIndex]*Factory
+	// baseCtx is the lifetime anchor for every shared informer created by the
+	// store. Shared informers are long-lived, store-owned resources, so their
+	// contexts must descend from baseCtx (tied to the events-manager), NOT from
+	// the transient context of whichever consumer happened to register first.
+	// Otherwise cancelling one consumer's context would tear down the shared
+	// informer for every other consumer still using it.
+	baseCtx context.Context
 }
 
-func NewFactoryStore() *FactoryStore {
+func NewFactoryStore(ctx context.Context) *FactoryStore {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
 	fs := &FactoryStore{
-		data:      make(map[FactoryIndex]*Factory),
-		stoppedCh: make(map[FactoryIndex]chan struct{}),
+		data:    make(map[FactoryIndex]*Factory),
+		baseCtx: ctx,
 	}
 	return fs
 }
@@ -57,11 +67,13 @@ func (c *FactoryStore) Reset() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.data = make(map[FactoryIndex]*Factory)
-	c.stoppedCh = make(map[FactoryIndex]chan struct{})
 }
 
-func (c *FactoryStore) add(ctx context.Context, index FactoryIndex, f dynamicinformer.DynamicSharedInformerFactory) {
-	ctx, cancel := context.WithCancel(ctx)
+func (c *FactoryStore) add(index FactoryIndex, f dynamicinformer.DynamicSharedInformerFactory) {
+	// Derive from the store's baseCtx, not from the caller's context: the
+	// shared informer's lifetime is owned by the store and must only end when
+	// the last handler is removed (see Stop) or the manager shuts down.
+	ctx, cancel := context.WithCancel(c.baseCtx)
 	c.data[index] = &Factory{
 		shared:               f,
 		handlerRegistrations: make(map[string]cache.ResourceEventHandlerRegistration),
@@ -74,13 +86,38 @@ func (c *FactoryStore) add(ctx context.Context, index FactoryIndex, f dynamicinf
 		slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()))
 }
 
-func (c *FactoryStore) get(ctx context.Context, client dynamic.Interface, index FactoryIndex) *Factory {
+// isDead reports whether the factory's shared informer goroutine has already
+// exited (its Run returned and done was closed). A dead factory must never be
+// reused: attaching a handler to it silently succeeds while the informer never
+// lists/watches again, freezing the snapshot until the process restarts.
+func (f *Factory) isDead() bool {
+	if f.done == nil {
+		return false
+	}
+	select {
+	case <-f.done:
+		return true
+	default:
+		return false
+	}
+}
+
+func (c *FactoryStore) get(client dynamic.Interface, index FactoryIndex) *Factory {
 	f, ok := c.data[index]
-	if ok {
+	if ok && !f.isDead() {
 		log.Debug("Factory store: the factory with index found",
 			slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()))
 		return f
 	}
+	if ok {
+		// The cached factory's informer has terminated (e.g. torn down in the
+		// window of a concurrent Stop). Discard the corpse and rebuild instead
+		// of handing back a dead informer.
+		log.Warn("Factory store: cached factory is dead, recreating",
+			slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()))
+		f.cancel()
+		delete(c.data, index)
+	}
 
 	// define resyncPeriod for informer
 	resyncPeriod := randomizedResyncPeriod()
@@ -98,20 +135,25 @@ func (c *FactoryStore) get(ctx context.Context, client dynamic.Interface, index
 		client, resyncPeriod, index.Namespace, tweakListOptions)
 	factory.ForResource(index.GVR)
 
-	c.add(ctx, index, factory)
+	c.add(index, factory)
 
 	return c.data[index]
 }
 
 func (c *FactoryStore) Start(ctx context.Context, informerId string, client dynamic.Interface, index FactoryIndex, handler cache.ResourceEventHandler, errorHandler *WatchErrorHandler) error {
 	c.mu.Lock()
-	defer c.mu.Unlock()
 
-	factory := c.get(ctx, client, index)
+	factory := c.get(client, index)
 
 	informer := factory.shared.ForResource(index.GVR).Informer()
-	// Add error handler, ignore "already started" error.
-	_ = informer.SetWatchErrorHandler(errorHandler.handler)
+	// Register the watch error handler. This returns an error when the shared
+	// informer is already running (a previous consumer registered its handler);
+	// that is expected on a reused factory, so log rather than silently drop it.
+	if err := informer.SetWatchErrorHandler(errorHandler.handler); err != nil {
+		log.Debug("Factory store: couldn't set watch error handler, informer likely already started",
+			slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()),
+			log.Err(err))
+	}
 
 	registration, err := informer.AddEventHandler(handler)
 	if err != nil {
@@ -140,6 +182,12 @@ func (c *FactoryStore) Start(ctx context.Context, informerId string, client dyna
 		}()
 	}
 
+	// Release the store lock before waiting for the initial sync. Holding c.mu
+	// across a blocking cache sync would serialize Start/Stop of every other
+	// informer behind a single slow initial LIST/discovery (throttled API,
+	// heavy CRD). The informer handle is safe to poll without the lock.
+	c.mu.Unlock()
+
 	if !informer.HasSynced() {
 		if err := wait.PollUntilContextCancel(ctx, DefaultSyncTime, true, func(_ context.Context) (bool, error) {
 			return informer.HasSynced(), nil
@@ -190,45 +238,55 @@ func (c *FactoryStore) Stop(informerId string, index FactoryIndex) {
 			}
 
 			c.mu.Lock()
-			delete(c.data, index)
-
-			log.Debug("Factory store: deleted factory",
-				slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()))
-
-			if ch, ok := c.stoppedCh[index]; ok {
-				close(ch)
-				delete(c.stoppedCh, index)
+			// Only remove the factory if it is still the same instance we just
+			// cancelled. While the lock was released (waiting on done), a
+			// concurrent Start could have observed this dead factory, rebuilt
+			// it, and installed a fresh, live one under the same index — that
+			// one must not be deleted, or its informer goroutine would be
+			// orphaned and untracked.
+			if cur, ok := c.data[index]; ok && cur == f {
+				delete(c.data, index)
+
+				log.Debug("Factory store: deleted factory",
+					slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()))
 			}
 		}
 	}
 
 	c.mu.Unlock()
 }
 
-// WaitStopped blocks until there is no factory for the index or timeout
+// WaitStopped blocks until the shared informer for the index has stopped (its
+// Run goroutine has exited) or FactoryShutdownTimeout elapses.
+//
+// The stop signal is taken from the factory's own done channel rather than a
+// separate per-index channel: the informer's lifetime belongs to the factory
+// instance, so a factory that is concurrently rebuilt under the same index
+// (see get/Stop) gets its own done and cannot be falsely signalled as stopped
+// by a previous instance's teardown.
 func (c *FactoryStore) WaitStopped(index FactoryIndex) {
 	c.mu.Lock()
-
-	if _, ok := c.data[index]; !ok {
+	f, ok := c.data[index]
+	if !ok {
+		// No factory for the index: it was never created, or it has already
+		// stopped and been removed. Nothing to wait for.
 		c.mu.Unlock()
 		return
 	}
+	done := f.done
+	c.mu.Unlock()
 
-	ch, ok := c.stoppedCh[index]
-	if !ok {
-		ch = make(chan struct{})
-		close(ch)
+	if done == nil {
+		// The informer goroutine was never launched (no handler ever
+		// registered), so there is nothing running to wait for.
+		return
 	}
 
-	c.mu.Unlock()
-
-	for {
-		select {
-		case <-ch:
-			return
-		case <-time.After(FactoryShutdownTimeout):
-			log.Warn("timeout waiting for factory to stop",
-				slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()))
-		}
+	select {
+	case <-done:
+		return
+	case <-time.After(FactoryShutdownTimeout):
+		log.Warn("timeout waiting for factory to stop",
+			slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()))
 	}
 }

pkg/kube_events_manager/factory_test.go

@@ -0,0 +1,182 @@
+package kubeeventsmanager
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/deckhouse/deckhouse/pkg/log"
+	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/flant/kube-client/fake"
+	"github.com/flant/shell-operator/pkg/metric"
+)
+
+// configMapIndex is the shared FactoryIndex used by the tests below: cluster
+// "core/v1 ConfigMaps in the default namespace", no selectors. Two consumers
+// with this identical index share a single factory in the store.
+func configMapIndex() FactoryIndex {
+	return FactoryIndex{
+		GVR:       schema.GroupVersionResource{Group: "", Version: "v1", Resource: "configmaps"},
+		Namespace: "default",
+	}
+}
+
+// Test_FactoryStore_SharedInformer_SurvivesFirstConsumerCancel covers the
+// context-ownership fix (defect #1).
+//
+// Two consumers (A then B) share one shared informer for the same index. A
+// registers first. On the buggy code the factory's context descended from A's
+// context, so cancelling A (as StopMonitor does at module reload) tore down the
+// shared informer while B was still registered — B's snapshot froze until the
+// process restarted. With the fix the informer's lifetime is anchored to the
+// store's baseCtx and must survive A's cancellation, keeping B live.
+func Test_FactoryStore_SharedInformer_SurvivesFirstConsumerCancel(t *testing.T) {
+	g := NewWithT(t)
+
+	fc := fake.NewFakeCluster(fake.ClusterVersionV121)
+	createNsWithLabels(fc, "default", nil)
+	createCM(fc, "default", testCM("cm-initial"))
+
+	baseCtx, baseCancel := context.WithCancel(context.Background())
+	defer baseCancel()
+
+	store := NewFactoryStore(baseCtx)
+	index := configMapIndex()
+	errH := newWatchErrorHandler("test", "ConfigMap", nil, metric.NewStorageMock(t), log.NewNop())
+
+	var bCount atomic.Int64
+	handlerA := cache.ResourceEventHandlerFuncs{AddFunc: func(_ interface{}) {}}
+	handlerB := cache.ResourceEventHandlerFuncs{AddFunc: func(_ interface{}) { bCount.Add(1) }}
+
+	// Consumer A registers first -> creates the shared factory.
+	ctxA, cancelA := context.WithCancel(baseCtx)
+	g.Expect(store.Start(ctxA, "consumer-A", fc.Client.Dynamic(), index, handlerA, errH)).
+		ShouldNot(HaveOccurred())
+
+	// Consumer B reuses the same shared factory.
+	ctxB, cancelB := context.WithCancel(baseCtx)
+	defer cancelB()
+	g.Expect(store.Start(ctxB, "consumer-B", fc.Client.Dynamic(), index, handlerB, errH)).
+		ShouldNot(HaveOccurred())
+
+	g.Eventually(bCount.Load, "5s", "10ms").Should(BeNumerically(">=", int64(1)),
+		"consumer B must receive the initial snapshot")
+
+	// Simulate monitor A being stopped: its context is cancelled (resource_informer
+	// cancels ei.ctx on stop) AND its handler is removed (factoryStore.Stop drops
+	// the refcount from 2 to 1).
+	cancelA()
+	store.Stop("consumer-A", index)
+
+	// Give a cancelled shared informer time to actually exit on the buggy code so
+	// the assertion below is not decided by a lucky in-flight delivery. On the
+	// fixed code nothing is torn down, so this settle is a no-op.
+	time.Sleep(500 * time.Millisecond)
+
+	before := bCount.Load()
+
+	// A new object arrives AFTER A is gone. B is still registered, so B must see it.
+	createCM(fc, "default", testCM("cm-after-a-stopped"))
+
+	g.Eventually(bCount.Load, "5s", "10ms").Should(BeNumerically(">", before),
+		"shared informer must keep delivering to consumer B after consumer A is stopped")
+}
+
+// Test_FactoryStore_WaitStopped_BlocksUntilInformerExits guards the graceful
+// shutdown barrier. Before the fix WaitStopped read a per-index stoppedCh that
+// was never populated, so it built a pre-closed channel and returned
+// immediately — monitor.Wait() never actually waited for informers to stop.
+// The fix waits on the factory's own done channel, which is closed only when
+// the informer's Run goroutine exits.
+func Test_FactoryStore_WaitStopped_BlocksUntilInformerExits(t *testing.T) {
+	g := NewWithT(t)
+
+	fc := fake.NewFakeCluster(fake.ClusterVersionV121)
+	createNsWithLabels(fc, "default", nil)
+
+	baseCtx, baseCancel := context.WithCancel(context.Background())
+	defer baseCancel()
+
+	store := NewFactoryStore(baseCtx)
+	index := configMapIndex()
+	errH := newWatchErrorHandler("test", "ConfigMap", nil, metric.NewStorageMock(t), log.NewNop())
+
+	// WaitStopped on an unknown index must return immediately.
+	g.Eventually(func() bool { store.WaitStopped(index); return true }, "1s", "10ms").
+		Should(BeTrue(), "WaitStopped must not block when no factory exists for the index")
+
+	handler := cache.ResourceEventHandlerFuncs{AddFunc: func(_ interface{}) {}}
+	ctx, cancel := context.WithCancel(baseCtx)
+	g.Expect(store.Start(ctx, "consumer", fc.Client.Dynamic(), index, handler, errH)).
+		ShouldNot(HaveOccurred())
+
+	var returned atomic.Bool
+	go func() {
+		store.WaitStopped(index)
+		returned.Store(true)
+	}()
+
+	// The informer is still running, so WaitStopped must keep blocking.
+	g.Consistently(returned.Load, "500ms", "20ms").Should(BeFalse(),
+		"WaitStopped must block while the shared informer is still running")
+
+	// Stopping the last consumer cancels the factory; its Run goroutine exits
+	// and closes done, which must release WaitStopped.
+	cancel()
+	store.Stop("consumer", index)
+
+	g.Eventually(returned.Load, "5s", "10ms").Should(BeTrue(),
+		"WaitStopped must return once the shared informer has stopped")
+}
+
+// Test_Factory_isDead covers the liveness predicate (defect #2).
+func Test_Factory_isDead(t *testing.T) {
+	g := NewWithT(t)
+
+	// Never started: Run has not been launched, done is nil.
+	g.Expect((&Factory{done: nil}).isDead()).To(BeFalse(), "a never-started factory is not dead")
+
+	// Started and still running: done is open.
+	running := make(chan struct{})
+	g.Expect((&Factory{done: running}).isDead()).To(BeFalse(), "a running factory is not dead")
+
+	// Started and exited: informer.Run returned and closed done.
+	exited := make(chan struct{})
+	close(exited)
+	g.Expect((&Factory{done: exited}).isDead()).To(BeTrue(), "a factory whose Run exited is dead")
+}
+
+// Test_FactoryStore_Get_RecreatesDeadFactory covers the corpse-recreation fix
+// (defect #2): get() must never hand back a factory whose informer goroutine has
+// already exited. On the buggy code get() returned the cached factory as-is, so
+// a new consumer attached to a dead informer and its snapshot never updated. The
+// fix cancels and discards the corpse and builds a fresh factory instead.
+func Test_FactoryStore_Get_RecreatesDeadFactory(t *testing.T) {
+	g := NewWithT(t)
+
+	fc := fake.NewFakeCluster(fake.ClusterVersionV121)
+	store := NewFactoryStore(context.Background())
+	index := configMapIndex()
+
+	// Inject a dead factory (its Run has exited -> done closed) under the index.
+	dead := make(chan struct{})
+	close(dead)
+	var cancelled atomic.Bool
+	corpse := &Factory{
+		handlerRegistrations: make(map[string]cache.ResourceEventHandlerRegistration),
+		done:                 dead,
+		cancel:               func() { cancelled.Store(true) },
+	}
+	store.data[index] = corpse
+
+	got := store.get(fc.Client.Dynamic(), index)
+
+	g.Expect(got).ShouldNot(BeIdenticalTo(corpse), "get must not reuse the dead factory")
+	g.Expect(cancelled.Load()).To(BeTrue(), "the dead factory must be cancelled before discarding")
+	g.Expect(got.isDead()).To(BeFalse(), "the recreated factory must be alive")
+	g.Expect(store.data[index]).To(BeIdenticalTo(got), "the store must cache the recreated factory under the index")
+}