From 9d5c654fab4ee6a4a6edef6961ad4b452b4ac778 Mon Sep 17 00:00:00 2001 From: Ruslan Gorbunov Date: Thu, 2 Jul 2026 16:09:04 +0300 Subject: [PATCH 1/4] [shell-operator] fix: protect shared informer lifecycle Signed-off-by: Ruslan Gorbunov --- pkg/kube_events_manager/factory.go | 87 +++++++++++++++---- .../kube_events_manager.go | 2 +- pkg/kube_events_manager/monitor_test.go | 2 +- 3 files changed, 73 insertions(+), 18 deletions(-) diff --git a/pkg/kube_events_manager/factory.go b/pkg/kube_events_manager/factory.go index 95c27c20..ec544781 100644 --- a/pkg/kube_events_manager/factory.go +++ b/pkg/kube_events_manager/factory.go @@ -40,14 +40,22 @@ type Factory struct { } type FactoryStore struct { - mu sync.Mutex - data map[FactoryIndex]*Factory + 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 stoppedCh map[FactoryIndex]chan struct{} } -func NewFactoryStore() *FactoryStore { +func NewFactoryStore(ctx context.Context) *FactoryStore { fs := &FactoryStore{ data: make(map[FactoryIndex]*Factory), + baseCtx: ctx, stoppedCh: make(map[FactoryIndex]chan struct{}), } return fs @@ -60,8 +68,11 @@ func (c *FactoryStore) Reset() { 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 +85,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 +134,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 +181,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,10 +237,18 @@ 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())) + // 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())) + } if ch, ok := c.stoppedCh[index]; ok { close(ch) diff --git a/pkg/kube_events_manager/kube_events_manager.go b/pkg/kube_events_manager/kube_events_manager.go index 3a86ebcc..14ba419c 100644 --- a/pkg/kube_events_manager/kube_events_manager.go +++ b/pkg/kube_events_manager/kube_events_manager.go @@ -75,7 +75,7 @@ func NewKubeEventsManager(ctx context.Context, client *klient.Client, logger *lo ctx: cctx, cancel: cancel, KubeClient: client, - factoryStore: NewFactoryStore(), + factoryStore: NewFactoryStore(cctx), m: sync.RWMutex{}, Monitors: make(map[string]Monitor), KubeEventCh: make(chan kemtypes.KubeEvent, 1), diff --git a/pkg/kube_events_manager/monitor_test.go b/pkg/kube_events_manager/monitor_test.go index 11c9cea5..814ab9a9 100644 --- a/pkg/kube_events_manager/monitor_test.go +++ b/pkg/kube_events_manager/monitor_test.go @@ -58,7 +58,7 @@ func Test_Monitor_should_handle_dynamic_ns_events(t *testing.T) { metricStorage.GaugeSetMock.When(metrics.KubeSnapshotObjects, 2, map[string]string(nil)).Then() metricStorage.GaugeSetMock.When(metrics.KubeSnapshotObjects, 3, map[string]string(nil)).Then() - mon := NewMonitor(context.Background(), fc.Client, metricStorage, NewFactoryStore(), monitorCfg, func(ev kemtypes.KubeEvent) { + mon := NewMonitor(context.Background(), fc.Client, metricStorage, NewFactoryStore(context.Background()), monitorCfg, func(ev kemtypes.KubeEvent) { objsMutex.Lock() objsFromEvents = append(objsFromEvents, snapshotResourceIDs(ev.Objects)...) objsMutex.Unlock() From 6f24334e1270940b4a047620a43ce6ccb984c701 Mon Sep 17 00:00:00 2001 From: Ruslan Gorbunov Date: Mon, 6 Jul 2026 16:36:53 +0300 Subject: [PATCH 2/4] [shell-operator] chore: add tests for shared informer lifetime handling Signed-off-by: Ruslan Gorbunov --- pkg/kube_events_manager/factory_test.go | 135 ++++++++++++++++++++++++ 1 file changed, 135 insertions(+) create mode 100644 pkg/kube_events_manager/factory_test.go diff --git a/pkg/kube_events_manager/factory_test.go b/pkg/kube_events_manager/factory_test.go new file mode 100644 index 00000000..6a25356e --- /dev/null +++ b/pkg/kube_events_manager/factory_test.go @@ -0,0 +1,135 @@ +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_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") +} From 563037d745658a50a2cf3c50a03a435155db8c12 Mon Sep 17 00:00:00 2001 From: Ruslan Gorbunov Date: Mon, 6 Jul 2026 17:32:23 +0300 Subject: [PATCH 3/4] [shell-operator] fix: wait for factory informer shutdown Signed-off-by: Ruslan Gorbunov --- pkg/kube_events_manager/factory.go | 55 ++++++++++++------------- pkg/kube_events_manager/factory_test.go | 47 +++++++++++++++++++++ 2 files changed, 74 insertions(+), 28 deletions(-) diff --git a/pkg/kube_events_manager/factory.go b/pkg/kube_events_manager/factory.go index ec544781..0d588e5f 100644 --- a/pkg/kube_events_manager/factory.go +++ b/pkg/kube_events_manager/factory.go @@ -48,15 +48,13 @@ type FactoryStore struct { // 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 - stoppedCh map[FactoryIndex]chan struct{} + baseCtx context.Context } func NewFactoryStore(ctx context.Context) *FactoryStore { fs := &FactoryStore{ - data: make(map[FactoryIndex]*Factory), - baseCtx: ctx, - stoppedCh: make(map[FactoryIndex]chan struct{}), + data: make(map[FactoryIndex]*Factory), + baseCtx: ctx, } return fs } @@ -65,7 +63,6 @@ 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(index FactoryIndex, f dynamicinformer.DynamicSharedInformerFactory) { @@ -249,41 +246,43 @@ func (c *FactoryStore) Stop(informerId string, index FactoryIndex) { 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) - } } } 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())) } } diff --git a/pkg/kube_events_manager/factory_test.go b/pkg/kube_events_manager/factory_test.go index 6a25356e..c1a38085 100644 --- a/pkg/kube_events_manager/factory_test.go +++ b/pkg/kube_events_manager/factory_test.go @@ -86,6 +86,53 @@ func Test_FactoryStore_SharedInformer_SurvivesFirstConsumerCancel(t *testing.T) "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) From 8c7d3c76844780afc18259ec4621f48a7f6f4241 Mon Sep 17 00:00:00 2001 From: Ruslan Gorbunov Date: Wed, 8 Jul 2026 11:05:25 +0300 Subject: [PATCH 4/4] [shell-operator] fix: handle nil context in factory store Signed-off-by: Ruslan Gorbunov --- pkg/kube_events_manager/factory.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pkg/kube_events_manager/factory.go b/pkg/kube_events_manager/factory.go index 0d588e5f..e3766643 100644 --- a/pkg/kube_events_manager/factory.go +++ b/pkg/kube_events_manager/factory.go @@ -52,6 +52,10 @@ type FactoryStore struct { } func NewFactoryStore(ctx context.Context) *FactoryStore { + if ctx == nil { + ctx = context.Background() + } + fs := &FactoryStore{ data: make(map[FactoryIndex]*Factory), baseCtx: ctx,