diff --git a/pkg/hook/controller/kubernetes_bindings_controller.go b/pkg/hook/controller/kubernetes_bindings_controller.go index 80c966d5..1b7666fc 100644 --- a/pkg/hook/controller/kubernetes_bindings_controller.go +++ b/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 } } diff --git a/pkg/hook/controller/kubernetes_bindings_controller_test.go b/pkg/hook/controller/kubernetes_bindings_controller_test.go new file mode 100644 index 00000000..eae26159 --- /dev/null +++ b/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") +} diff --git a/pkg/kube_events_manager/factory.go b/pkg/kube_events_manager/factory.go index 95c27c20..14dac288 100644 --- a/pkg/kube_events_manager/factory.go +++ b/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 diff --git a/pkg/kube_events_manager/factory_test.go b/pkg/kube_events_manager/factory_test.go new file mode 100644 index 00000000..ba7f0f1c --- /dev/null +++ b/pkg/kube_events_manager/factory_test.go @@ -0,0 +1,136 @@ +package kubeeventsmanager + +import ( + "context" + "sync" + "testing" + + "github.com/deckhouse/deckhouse/pkg/log" + . "github.com/onsi/gomega" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/flant/kube-client/fake" +) + +var configMapsGVR = schema.GroupVersionResource{Version: "v1", Resource: "configmaps"} + +// recordingHandler is a minimal cache.ResourceEventHandler that records names of +// added objects. +type recordingHandler struct { + mu sync.Mutex + added []string +} + +func (h *recordingHandler) OnAdd(obj interface{}, _ bool) { + u, ok := obj.(*unstructured.Unstructured) + if !ok { + return + } + h.mu.Lock() + h.added = append(h.added, u.GetName()) + h.mu.Unlock() +} + +func (h *recordingHandler) OnUpdate(_, _ interface{}) {} + +func (h *recordingHandler) OnDelete(_ interface{}) {} + +func (h *recordingHandler) addedNames() []string { + h.mu.Lock() + defer h.mu.Unlock() + return append([]string(nil), h.added...) +} + +func testWatchErrorHandler() *WatchErrorHandler { + return newWatchErrorHandler("factory-test", "ConfigMap", nil, nil, log.NewNop()) +} + +// TestFactoryStore_SurvivesConsumerContextCancel is the regression test for the +// shared-informer lifetime bug: the factory context must derive from the store's +// base context, not from the context of whichever consumer created the factory +// first. Before the fix, cancelling the first consumer's context killed the +// shared informer for every other consumer, silently freezing their snapshots. +func TestFactoryStore_SurvivesConsumerContextCancel(t *testing.T) { + g := NewWithT(t) + + fc := fake.NewFakeCluster(fake.ClusterVersionV121) + createNsWithLabels(fc, "default", map[string]string{}) + + store := NewFactoryStore(context.Background()) + index := FactoryIndex{GVR: configMapsGVR, Namespace: "default"} + + ctx1, cancel1 := context.WithCancel(context.Background()) + defer cancel1() + + handler1 := &recordingHandler{} + g.Expect(store.Start(ctx1, "informer-1", fc.Client.Dynamic(), index, handler1, testWatchErrorHandler())).To(Succeed()) + + handler2 := &recordingHandler{} + g.Expect(store.Start(context.Background(), "informer-2", fc.Client.Dynamic(), index, handler2, testWatchErrorHandler())).To(Succeed()) + + // The first consumer goes away: its context is cancelled and its handler is + // removed — exactly what StopMonitor does for one of two hooks sharing an index. + cancel1() + store.Stop("informer-1", index) + + // The shared informer must still be running for the second consumer. + store.mu.Lock() + factory, ok := store.data[index] + store.mu.Unlock() + g.Expect(ok).To(BeTrue(), "factory must stay in the store while a consumer remains") + select { + case <-factory.done: + t.Fatal("shared informer died together with the first consumer's context") + default: + } + + // And events must still be delivered to the second consumer. + createCM(fc, "default", testCM("after-first-consumer-gone")) + g.Eventually(handler2.addedNames, "5s", "10ms"). + Should(ContainElement("after-first-consumer-gone"), + "second consumer must keep receiving events after the first consumer is gone") +} + +// TestFactoryStore_RecreatesDeadFactory is the regression test for the +// success-on-corpse bug: Start on an index whose informer goroutine has already +// exited must build a fresh factory instead of silently reusing the stopped one +// (whose HasSynced() still returns true and which never delivers events). +func TestFactoryStore_RecreatesDeadFactory(t *testing.T) { + g := NewWithT(t) + + fc := fake.NewFakeCluster(fake.ClusterVersionV121) + createNsWithLabels(fc, "default", map[string]string{}) + + store := NewFactoryStore(context.Background()) + index := FactoryIndex{GVR: configMapsGVR, Namespace: "default"} + + handler1 := &recordingHandler{} + g.Expect(store.Start(context.Background(), "informer-1", fc.Client.Dynamic(), index, handler1, testWatchErrorHandler())).To(Succeed()) + + // Kill the informer out of band, keeping the factory in the store — the state + // the pre-fix lifetime bug used to leave behind. + store.mu.Lock() + dead := store.data[index] + store.mu.Unlock() + dead.cancel() + <-dead.done + + handler2 := &recordingHandler{} + g.Expect(store.Start(context.Background(), "informer-2", fc.Client.Dynamic(), index, handler2, testWatchErrorHandler())).To(Succeed()) + + store.mu.Lock() + fresh, ok := store.data[index] + store.mu.Unlock() + g.Expect(ok).To(BeTrue()) + g.Expect(fresh).ToNot(BeIdenticalTo(dead), "a dead factory must be replaced, not reused") + select { + case <-fresh.done: + t.Fatal("recreated factory is not running") + default: + } + + createCM(fc, "default", testCM("after-recreate")) + g.Eventually(handler2.addedNames, "5s", "10ms"). + Should(ContainElement("after-recreate"), "consumer on the recreated factory must receive events") +} 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() diff --git a/pkg/shell-operator/manager_events_handler.go b/pkg/shell-operator/manager_events_handler.go index 38f55c92..01df8525 100644 --- a/pkg/shell-operator/manager_events_handler.go +++ b/pkg/shell-operator/manager_events_handler.go @@ -2,6 +2,8 @@ package shell_operator import ( "context" + "log/slog" + "runtime/debug" "github.com/deckhouse/deckhouse/pkg/log" @@ -73,12 +75,16 @@ func (m *ManagerEventsHandler) Start() { select { case crontab := <-m.scheduleManager.Ch(): if m.scheduleCb != nil { - tailTasks = m.scheduleCb(ctx, crontab) + tailTasks = runEventCb(logEntry, "schedule", func() []task.Task { + return m.scheduleCb(ctx, crontab) + }) } case kubeEvent := <-m.kubeEventsManager.Ch(): if m.kubeEventCb != nil { - tailTasks = m.kubeEventCb(ctx, kubeEvent) + tailTasks = runEventCb(logEntry, "kubernetes", func() []task.Task { + return m.kubeEventCb(ctx, kubeEvent) + }) } case <-m.ctx.Done(): @@ -90,3 +96,22 @@ func (m *ManagerEventsHandler) Start() { } }() } + +// runEventCb invokes an event callback with panic isolation, mirroring the +// recover in TaskQueue.processOne: a panicking hook dispatch must not kill +// the events goroutine and with it the whole process. On panic the event's +// tail tasks are dropped and nil is returned. +func runEventCb(logEntry *log.Logger, binding string, cb func() []task.Task) []task.Task { + defer func() { + if r := recover(); r != nil { + logEntry.Warn( + "panic recovered in ManagerEventsHandler", + slog.String(pkg.LogKeyBinding, binding), + slog.Any(pkg.LogKeyError, r), + slog.String(pkg.LogKeyStack, string(debug.Stack())), + ) + } + }() + + return cb() +} diff --git a/pkg/shell-operator/manager_events_handler_test.go b/pkg/shell-operator/manager_events_handler_test.go new file mode 100644 index 00000000..8e171488 --- /dev/null +++ b/pkg/shell-operator/manager_events_handler_test.go @@ -0,0 +1,99 @@ +package shell_operator + +import ( + "context" + "testing" + "time" + + "github.com/deckhouse/deckhouse/pkg/log" + "github.com/stretchr/testify/require" + + kemtypes "github.com/flant/shell-operator/pkg/kube_events_manager/types" + "github.com/flant/shell-operator/pkg/task" + "github.com/flant/shell-operator/pkg/task/queue" +) + +type fakeKubeEventEmitter struct { + ch chan kemtypes.KubeEvent +} + +func (f *fakeKubeEventEmitter) Ch() chan kemtypes.KubeEvent { return f.ch } + +type fakeScheduleEmitter struct { + ch chan string +} + +func (f *fakeScheduleEmitter) Ch() chan string { return f.ch } + +func newTestEventsHandler(kubeCh chan kemtypes.KubeEvent, schedCh chan string) *ManagerEventsHandler { + return newManagerEventsHandler(context.Background(), &managerEventsHandlerConfig{ + tqs: queue.NewTaskQueueSet(), + mgr: &fakeKubeEventEmitter{ch: kubeCh}, + smgr: &fakeScheduleEmitter{ch: schedCh}, + logger: log.NewNop(), + }) +} + +// TestManagerEventsHandlerSurvivesPanicInKubeEventCb proves the panic +// escalation bug: a panic inside the kube-event callback (in production - the +// nil HookController dereference during hook dispatch) must not kill the +// events goroutine and the whole process. The handler has to log the panic +// and keep processing subsequent events. +// +// On unfixed code this test crashes the whole test binary - that crash is the +// process-level SIGSEGV from the incident, reproduced in miniature. +func TestManagerEventsHandlerSurvivesPanicInKubeEventCb(t *testing.T) { + kubeCh := make(chan kemtypes.KubeEvent) + m := newTestEventsHandler(kubeCh, make(chan string)) + + processed := make(chan string, 1) + m.WithKubeEventHandler(func(_ context.Context, ev kemtypes.KubeEvent) []task.Task { + if ev.MonitorId == "panic" { + panic("runtime error: invalid memory address or nil pointer dereference") + } + processed <- ev.MonitorId + return nil + }) + + m.Start() + defer m.cancel() + + kubeCh <- kemtypes.KubeEvent{MonitorId: "panic"} + kubeCh <- kemtypes.KubeEvent{MonitorId: "after-panic"} + + select { + case got := <-processed: + require.Equal(t, "after-panic", got) + case <-time.After(5 * time.Second): + t.Fatal("events goroutine died after panic: second event was never processed") + } +} + +// TestManagerEventsHandlerSurvivesPanicInScheduleCb covers the same panic +// isolation for the schedule callback branch. +func TestManagerEventsHandlerSurvivesPanicInScheduleCb(t *testing.T) { + schedCh := make(chan string) + m := newTestEventsHandler(make(chan kemtypes.KubeEvent), schedCh) + + processed := make(chan string, 1) + m.WithScheduleEventHandler(func(_ context.Context, crontab string) []task.Task { + if crontab == "panic" { + panic("schedule hook dispatch failed") + } + processed <- crontab + return nil + }) + + m.Start() + defer m.cancel() + + schedCh <- "panic" + schedCh <- "* * * * *" + + select { + case got := <-processed: + require.Equal(t, "* * * * *", got) + case <-time.After(5 * time.Second): + t.Fatal("events goroutine died after panic: second event was never processed") + } +}