[shell-operator] fix: isolate panics in ManagerEventsHandler callbacks, self-heal kubernetes bindings, own shared informer lifetime

pkg/hook/controller/hook_controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
+	metricsstorage "github.com/deckhouse/deckhouse/pkg/metrics-storage"
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 
 	bctx "github.com/flant/shell-operator/pkg/hook/binding_context"
@@ -64,6 +65,11 @@ func (hc *HookController) InitKubernetesBindings(bindings []htypes.OnKubernetesE
 	bindingCtrl := NewKubernetesBindingsController(logger)
 	bindingCtrl.WithKubeEventsManager(kubeEventMgr)
 	bindingCtrl.WithKubernetesBindings(bindings)
+	// KubeEventsSource is a narrow interface; recover the metric storage from the
+	// real manager when available so the controller can count missing monitors.
+	if mgr, ok := kubeEventMgr.(interface{ MetricStorage() metricsstorage.Storage }); ok {
+		bindingCtrl.WithMetricStorage(mgr.MetricStorage())
+	}
 	hc.KubernetesController = bindingCtrl
 	hc.kubernetesBindings = bindings
 	hc.logger = logger

pkg/hook/controller/kubernetes_bindings_controller.go

@@ -7,12 +7,14 @@ import (
 	"sync"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
+	metricsstorage "github.com/deckhouse/deckhouse/pkg/metrics-storage"
 
 	"github.com/flant/shell-operator/pkg"
 	bctx "github.com/flant/shell-operator/pkg/hook/binding_context"
 	htypes "github.com/flant/shell-operator/pkg/hook/types"
 	kubeeventsmanager "github.com/flant/shell-operator/pkg/kube_events_manager"
 	kemtypes "github.com/flant/shell-operator/pkg/kube_events_manager/types"
+	"github.com/flant/shell-operator/pkg/metrics"
 	utils "github.com/flant/shell-operator/pkg/utils/labels"
 )
 
@@ -49,6 +51,8 @@ type kubernetesBindingsController struct {
 
 	// dependencies
 	kubeEventsManager kubeeventsmanager.KubeEventsSource
+	// optional, only for the binding_monitor_missing_total counter
+	metricStorage metricsstorage.Storage
 
 	logger *log.Logger
 
@@ -76,34 +80,47 @@ func (c *kubernetesBindingsController) WithKubeEventsManager(kubeEventsManager k
 	c.kubeEventsManager = kubeEventsManager
 }
 
+// WithMetricStorage sets an optional storage for the binding_monitor_missing_total
+// counter. Without it the missing-monitor state is still logged, just not counted.
+func (c *kubernetesBindingsController) WithMetricStorage(mstor metricsstorage.Storage) {
+	c.metricStorage = mstor
+}
+
 // 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 +292,19 @@ 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.Error("no monitor for configured kubernetes binding, snapshot is empty",
+				slog.String(pkg.LogKeyBinding, bindingName),
+				slog.String(pkg.LogKeyMonitorID, monitorID))
+			if c.metricStorage != nil {
+				c.metricStorage.CounterAdd(metrics.BindingMonitorMissingTotal, 1.0,
+					map[string]string{pkg.MetricKeyBinding: bindingName})
+			}
+			return nil
 		}
 	}
 

pkg/hook/controller/kubernetes_bindings_controller_test.go

@@ -0,0 +1,132 @@
+package controller
+
+import (
+	"context"
+	"testing"
+
+	"github.com/deckhouse/deckhouse/pkg/log"
+	. "github.com/onsi/gomega"
+
+	"github.com/flant/kube-client/fake"
+	pkg "github.com/flant/shell-operator/pkg"
+	"github.com/flant/shell-operator/pkg/hook/config"
+	kubeeventsmanager "github.com/flant/shell-operator/pkg/kube_events_manager"
+	"github.com/flant/shell-operator/pkg/metric"
+	"github.com/flant/shell-operator/pkg/metrics"
+)
+
+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")
+}
+
+// TestSnapshotsFor_MissingMonitorIsCounted ensures the empty-snapshot state is
+// observable: reading a snapshot for a configured binding whose monitor is not
+// running must increment binding_monitor_missing_total with the binding label.
+func TestSnapshotsFor_MissingMonitorIsCounted(t *testing.T) {
+	g := NewWithT(t)
+
+	fc := fake.NewFakeCluster(fake.ClusterVersionV121)
+	mgr := kubeeventsmanager.NewKubeEventsManager(context.Background(), fc.Client, log.NewNop())
+
+	var (
+		calls     int
+		gotMetric string
+		gotValue  float64
+		gotLabels map[string]string
+	)
+	storage := metric.NewStorageMock(t)
+	storage.CounterAddMock.Set(func(name string, value float64, labels map[string]string) {
+		calls++
+		gotMetric, gotValue, gotLabels = name, value, labels
+	})
+	// The storage must be set before InitKubernetesBindings: the controller
+	// captures it from the manager at init time.
+	mgr.WithMetricStorage(storage)
+
+	cfg := &config.HookConfig{}
+	g.Expect(cfg.LoadAndValidate([]byte(twoBindingsHookConfig))).ShouldNot(HaveOccurred())
+
+	hc := NewHookController()
+	hc.InitKubernetesBindings(cfg.OnKubernetesEvents, mgr, log.NewNop())
+
+	// Bindings are configured but never enabled: no monitor is running.
+	binding := cfg.OnKubernetesEvents[0].BindingName
+	g.Expect(hc.KubernetesController.SnapshotsFor(binding)).To(BeNil())
+
+	g.Expect(calls).To(Equal(1))
+	g.Expect(gotMetric).To(Equal(metrics.BindingMonitorMissingTotal))
+	g.Expect(gotValue).To(Equal(1.0))
+	g.Expect(gotLabels).To(Equal(map[string]string{pkg.MetricKeyBinding: binding}))
+}