[shell-operator] fix: prevent silent empty snapshots from lost kubernetes monitors 1.17

pkg/hook/controller/kubernetes_bindings_controller.go

@@ -79,31 +79,38 @@ func (c *kubernetesBindingsController) WithKubeEventsManager(kubeEventsManager k
 // EnableKubernetesBindings adds a monitor for each 'kubernetes' binding. This method
 // returns an array of BindingExecutionInfo to help construct initial tasks to run hooks.
 // Informers in each monitor are started immediately to keep up the "fresh" state of object caches.
+//
+// The method is idempotent and self-healing: per-binding monitor creation is guarded so a
+// second call does not create duplicate monitors, and a binding whose link exists but whose
+// monitor is missing (e.g. a monitor stopped out of band by DisableKubernetesBindings racing
+// with this call) is repaired by re-adding and re-starting the monitor. A Synchronization
+// BindingExecutionInfo is always returned for every binding, so a repeat call after a partial
+// failure never silently drops the Synchronization run that populates hook snapshots.
 func (c *kubernetesBindingsController) EnableKubernetesBindings() ([]BindingExecutionInfo, error) {
 	res := make([]BindingExecutionInfo, 0)
 
-	c.l.RLock()
-	alreadyEnabled := len(c.BindingMonitorLinks) == len(c.KubernetesBindings)
-	c.l.RUnlock()
-	if alreadyEnabled {
-		return res, nil
-	}
-
 	for _, config := range c.KubernetesBindings {
-		if _, found := c.getBindingMonitorLinksById(config.Monitor.Metadata.MonitorId); !found {
+		monitorID := config.Monitor.Metadata.MonitorId
+
+		_, linked := c.getBindingMonitorLinksById(monitorID)
+		// Add the monitor when there is no link yet, or when the link exists but the
+		// monitor is gone — the latter is the state a Disable/Enable race leaves behind
+		// and, without repair here, every later snapshot read for this binding returns
+		// empty while the hook still "succeeds".
+		if !linked || !c.kubeEventsManager.HasMonitor(monitorID) {
 			if err := c.kubeEventsManager.AddMonitor(config.Monitor); err != nil {
 				return nil, fmt.Errorf("run monitor: %s", err)
 			}
-			c.setBindingMonitorLinks(config.Monitor.Metadata.MonitorId, &KubernetesBindingToMonitorLink{
-				MonitorId:     config.Monitor.Metadata.MonitorId,
+			c.setBindingMonitorLinks(monitorID, &KubernetesBindingToMonitorLink{
+				MonitorId:     monitorID,
 				BindingConfig: config,
 			})
 			// Start monitor's informers to fill the cache.
-			c.kubeEventsManager.StartMonitor(config.Monitor.Metadata.MonitorId)
+			c.kubeEventsManager.StartMonitor(monitorID)
 		}
 
 		synchronizationInfo := c.HandleEvent(context.TODO(), kemtypes.KubeEvent{
-			MonitorId: config.Monitor.Metadata.MonitorId,
+			MonitorId: monitorID,
 			Type:      kemtypes.TypeSynchronization,
 		})
 		res = append(res, synchronizationInfo)
@@ -275,6 +282,15 @@ func (c *kubernetesBindingsController) SnapshotsFor(bindingName string) []kemtyp
 			if c.kubeEventsManager.HasMonitor(monitorID) {
 				return c.kubeEventsManager.GetMonitor(monitorID).Snapshot()
 			}
+
+			// A configured binding without a live monitor yields an empty snapshot while the
+			// hook still runs successfully — this is exactly how a lost/never-started monitor
+			// silently feeds empty values into a hook. Make it observable instead of silent;
+			// EnableKubernetesBindings repairs the state on the next module run.
+			c.logger.Warn("no monitor for configured kubernetes binding, snapshot is empty",
+				slog.String(pkg.LogKeyBinding, bindingName),
+				slog.String("monitorID", monitorID))
+			return nil
 		}
 	}
 

pkg/hook/controller/kubernetes_bindings_controller_test.go

@@ -0,0 +1,89 @@
+package controller
+
+import (
+	"context"
+	"testing"
+
+	"github.com/deckhouse/deckhouse/pkg/log"
+	. "github.com/onsi/gomega"
+
+	"github.com/flant/kube-client/fake"
+	"github.com/flant/shell-operator/pkg/hook/config"
+	kubeeventsmanager "github.com/flant/shell-operator/pkg/kube_events_manager"
+)
+
+const twoBindingsHookConfig = `
+configVersion: v1
+kubernetes:
+- name: binding_1
+  apiVersion: v1
+  kind: Pod
+  executeHookOnEvent: ["Added"]
+- name: binding_2
+  apiVersion: v1
+  kind: Pod
+  executeHookOnEvent: ["Added"]
+`
+
+func newTestKubeController(t *testing.T) (*HookController, *config.HookConfig, kubeeventsmanager.KubeEventsManager) {
+	t.Helper()
+	g := NewWithT(t)
+
+	fc := fake.NewFakeCluster(fake.ClusterVersionV121)
+	mgr := kubeeventsmanager.NewKubeEventsManager(context.Background(), fc.Client, log.NewNop())
+
+	cfg := &config.HookConfig{}
+	g.Expect(cfg.LoadAndValidate([]byte(twoBindingsHookConfig))).ShouldNot(HaveOccurred())
+
+	hc := NewHookController()
+	hc.InitKubernetesBindings(cfg.OnKubernetesEvents, mgr, log.NewNop())
+
+	return hc, cfg, mgr
+}
+
+// TestEnableKubernetesBindings_RepeatCallStillEmitsSynchronization is the direct
+// regression test for the incident: on the second call the old code hit the
+// alreadyEnabled early-return and yielded ZERO Synchronization contexts, so a
+// ModuleRun retry lost the values-populating Synchronization run for the module's
+// hooks. The fix must return a Synchronization context for every binding on every
+// call.
+func TestEnableKubernetesBindings_RepeatCallStillEmitsSynchronization(t *testing.T) {
+	g := NewWithT(t)
+	hc, cfg, _ := newTestKubeController(t)
+	want := len(cfg.OnKubernetesEvents)
+
+	first, err := hc.KubernetesController.EnableKubernetesBindings()
+	g.Expect(err).ShouldNot(HaveOccurred())
+	g.Expect(first).To(HaveLen(want), "first call must emit a Synchronization context per binding")
+
+	second, err := hc.KubernetesController.EnableKubernetesBindings()
+	g.Expect(err).ShouldNot(HaveOccurred())
+	g.Expect(second).To(HaveLen(want), "repeat call must still emit Synchronization contexts (regression: was 0)")
+}
+
+// TestEnableKubernetesBindings_SelfHealsMissingMonitor reproduces the permanent
+// "link present, monitor gone" state left by a Disable/Enable race and asserts
+// that a subsequent EnableKubernetesBindings repairs it — restoring the
+// self-healing behavior that existed before the idempotent early-return was added.
+func TestEnableKubernetesBindings_SelfHealsMissingMonitor(t *testing.T) {
+	g := NewWithT(t)
+	hc, cfg, mgr := newTestKubeController(t)
+
+	_, err := hc.KubernetesController.EnableKubernetesBindings()
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	// Drop a monitor out of band while its binding link stays — this is the state
+	// DisableKubernetesBindings racing with the queue worker leaves behind.
+	victim := cfg.OnKubernetesEvents[0].Monitor.Metadata.MonitorId
+	g.Expect(mgr.HasMonitor(victim)).To(BeTrue())
+	g.Expect(mgr.StopMonitor(victim)).ShouldNot(HaveOccurred())
+	g.Expect(mgr.HasMonitor(victim)).To(BeFalse())
+
+	// Snapshot for the orphaned binding is empty here — the silent-failure state.
+	g.Expect(hc.KubernetesController.SnapshotsFor(cfg.OnKubernetesEvents[0].BindingName)).To(BeNil())
+
+	// Next enable must re-create the monitor rather than early-return.
+	_, err = hc.KubernetesController.EnableKubernetesBindings()
+	g.Expect(err).ShouldNot(HaveOccurred())
+	g.Expect(mgr.HasMonitor(victim)).To(BeTrue(), "missing monitor must be repaired on next EnableKubernetesBindings")
+}

pkg/kube_events_manager/factory.go

@@ -2,6 +2,7 @@ package kubeeventsmanager
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"sync"
 	"time"
@@ -40,13 +41,24 @@ type Factory struct {
 }
 
 type FactoryStore struct {
-	mu        sync.Mutex
+	mu sync.Mutex
+	// baseCtx bounds the lifetime of every shared informer in the store. The store
+	// (via its owning kubeEventsManager) is the owner of the shared factories, so
+	// their contexts must derive from the owner's context — never from the context
+	// of whichever consumer happened to create a factory first, or stopping that
+	// consumer would kill the shared informer for every other consumer.
+	baseCtx   context.Context
 	data      map[FactoryIndex]*Factory
 	stoppedCh map[FactoryIndex]chan struct{}
 }
 
-func NewFactoryStore() *FactoryStore {
+func NewFactoryStore(ctx context.Context) *FactoryStore {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
 	fs := &FactoryStore{
+		baseCtx:   ctx,
 		data:      make(map[FactoryIndex]*Factory),
 		stoppedCh: make(map[FactoryIndex]chan struct{}),
 	}
@@ -56,12 +68,17 @@ func NewFactoryStore() *FactoryStore {
 func (c *FactoryStore) Reset() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	// Stop informer goroutines before dropping the references, otherwise they
+	// keep running detached forever.
+	for _, f := range c.data {
+		f.cancel()
+	}
 	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) {
+	ctx, cancel := context.WithCancel(c.baseCtx)
 	c.data[index] = &Factory{
 		shared:               f,
 		handlerRegistrations: make(map[string]cache.ResourceEventHandlerRegistration),
@@ -74,7 +91,7 @@ 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 {
+func (c *FactoryStore) get(client dynamic.Interface, index FactoryIndex) *Factory {
 	f, ok := c.data[index]
 	if ok {
 		log.Debug("Factory store: the factory with index found",
@@ -98,26 +115,49 @@ 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)
+
+	// A factory whose informer goroutine has already exited is unusable: event
+	// handlers can no longer be added and no events will ever be delivered, while
+	// HasSynced() on the stopped informer still returns true. Reusing it would
+	// hand the consumer a snapshot that silently never updates. Drop the corpse
+	// and build a fresh factory instead.
+	if factory.done != nil {
+		select {
+		case <-factory.done:
+			log.Warn("Factory store: informer for index is stopped, recreating factory",
+				slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()))
+
+			factory.cancel()
+			delete(c.data, index)
+			factory = c.get(client, index)
+		default:
+		}
+	}
 
 	informer := factory.shared.ForResource(index.GVR).Informer()
-	// Add error handler, ignore "already started" error.
-	_ = informer.SetWatchErrorHandler(errorHandler.handler)
+	// SetWatchErrorHandler fails only on an already started informer — harmless
+	// here, the handler set by the first consumer stays in effect.
+	if err := informer.SetWatchErrorHandler(errorHandler.handler); err != nil {
+		log.Debug("Factory store: watch error handler wasn't set",
+			slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()),
+			log.Err(err))
+	}
 
 	registration, err := informer.AddEventHandler(handler)
 	if err != nil {
-		log.Warn("Factory store: couldn't add event handler to the factory's informer",
-			slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()),
-			log.Err(err))
+		// A consumer without a registered handler would run with a snapshot that
+		// never updates — fail the start instead of degrading silently.
+		c.mu.Unlock()
+		return fmt.Errorf("add event handler for %s: %w", index.GVR.String(), err)
 	}
 
 	factory.handlerRegistrations[informerId] = registration
@@ -140,6 +180,11 @@ func (c *FactoryStore) Start(ctx context.Context, informerId string, client dyna
 		}()
 	}
 
+	c.mu.Unlock()
+
+	// Wait for the cache sync outside the store lock: a slow initial LIST of one
+	// resource must not serialize Start/Stop of every other informer (same
+	// unlock-before-wait pattern as in Stop).
 	if !informer.HasSynced() {
 		if err := wait.PollUntilContextCancel(ctx, DefaultSyncTime, true, func(_ context.Context) (bool, error) {
 			return informer.HasSynced(), nil