From 154204a541e418f8b55964a6f7c4d295eccf5318 Mon Sep 17 00:00:00 2001 From: Roman Berezkin Date: Tue, 7 Jul 2026 13:38:11 +0300 Subject: [PATCH 1/6] fix: make module hook registration idempotent and race-free - AddHook replaces the previously registered hook object in all bindings instead of appending a duplicate: a stale entry keeps a nil HookController and crashes the kube-events dispatcher after a registration retry - registered/controllersReady flags are read and written under the storage lock, giving a happens-before edge between setting hook controllers and observing the readiness flag - RegisterHooks is serialized with a dedicated mutex so two concurrent ModuleRun tasks cannot register the same module twice - getHooks sorts and returns a copy instead of the shared slice Signed-off-by: Roman Berezkin --- go.mod | 2 +- go.sum | 6 +- pkg/module_manager/models/modules/basic.go | 19 +- .../models/modules/hook_storage.go | 78 ++++++- .../models/modules/hook_storage_test.go | 210 ++++++++++++++++++ 5 files changed, 300 insertions(+), 15 deletions(-) create mode 100644 pkg/module_manager/models/modules/hook_storage_test.go diff --git a/go.mod b/go.mod index 380d84c81..bfc428306 100644 --- a/go.mod +++ b/go.mod @@ -10,7 +10,7 @@ require ( github.com/dominikbraun/graph v0.23.0 github.com/ettle/strcase v0.2.0 github.com/flant/kube-client v1.9.1 - github.com/flant/shell-operator v1.20.1 + github.com/flant/shell-operator v1.20.2-0.20260707115217-ac90cf8d1033 github.com/go-chi/chi/v5 v5.3.0 github.com/go-openapi/loads v0.23.2 github.com/go-openapi/spec v0.22.1 diff --git a/go.sum b/go.sum index df944af02..f3ca12757 100644 --- a/go.sum +++ b/go.sum @@ -156,8 +156,10 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/flant/kube-client v1.9.1 h1:B5Sa++5arl3N/KBdL1KNMlUBtzQXKz1o2FIiFb4X+VA= github.com/flant/kube-client v1.9.1/go.mod h1:jAQ01+f5Q8UJJ4RYZlk5xqoA+OOalpr/QJ3e3Q/saqw= -github.com/flant/shell-operator v1.20.1 h1:v8+5r9H2bQGUDhh2hPC/GXzw1vfC8hFhO0fxSGnVCD0= -github.com/flant/shell-operator v1.20.1/go.mod h1:Qj8agNbfPm6x7C3TmCpY1fCtayhqS2Sxncui7grrOVw= +github.com/flant/shell-operator v1.20.2-0.20260707103551-b1bf9c86d4e6 h1:yDFBK1LSNs/olu0JEjNM+4bbSlM07p6YT5ZXWyeLdTk= +github.com/flant/shell-operator v1.20.2-0.20260707103551-b1bf9c86d4e6/go.mod h1:Qj8agNbfPm6x7C3TmCpY1fCtayhqS2Sxncui7grrOVw= +github.com/flant/shell-operator v1.20.2-0.20260707115217-ac90cf8d1033 h1:WsxzIwmXv3zR751CacmEXRMKeZ42gQjFED+sTbBFu0g= +github.com/flant/shell-operator v1.20.2-0.20260707115217-ac90cf8d1033/go.mod h1:Qj8agNbfPm6x7C3TmCpY1fCtayhqS2Sxncui7grrOVw= github.com/flopp/go-findfont v0.1.0 h1:lPn0BymDUtJo+ZkV01VS3661HL6F4qFlkhcJN55u6mU= github.com/flopp/go-findfont v0.1.0/go.mod h1:wKKxRDjD024Rh7VMwoU90i6ikQRCr+JTHB5n4Ejkqvw= github.com/fluxcd/flagger v1.36.1 h1:X2PumtNwZz9YSGaOtZLFm2zAKLgHhFkbNv8beg7ifyc= diff --git a/pkg/module_manager/models/modules/basic.go b/pkg/module_manager/models/modules/basic.go index 8cf2862ce..fda75e591 100644 --- a/pkg/module_manager/models/modules/basic.go +++ b/pkg/module_manager/models/modules/basic.go @@ -257,14 +257,17 @@ func (bm *BasicModule) DeregisterHooks() { bm.hasReadiness = false } -// HooksControllersReady returns controllersReady status of the hook storage +// HooksControllersReady returns controllersReady status of the hook storage. +// Reading under the storage lock also gives a happens-before edge: a reader +// that observed true sees all WithHookController writes made before the flag +// was set. func (bm *BasicModule) HooksControllersReady() bool { - return bm.hooks.controllersReady + return bm.hooks.isControllersReady() } // SetHooksControllersReady sets controllersReady status of the hook storage to true func (bm *BasicModule) SetHooksControllersReady() { - bm.hooks.controllersReady = true + bm.hooks.setControllersReady() } // ResetState drops the module state @@ -294,7 +297,13 @@ func (bm *BasicModule) ResetState() { // RegisterHooks searches and registers all module hooks from a filesystem or GoHook Registry func (bm *BasicModule) RegisterHooks(logger *log.Logger) ([]*hooks.ModuleHook, error) { - if bm.hooks.registered { + // Serialize whole-module registration: two concurrent ModuleRun tasks can + // both reach this point. The loser must wait and take the fast path below + // instead of re-publishing hooks with not-yet-set controllers. + bm.hooks.registrationMu.Lock() + defer bm.hooks.registrationMu.Unlock() + + if bm.hooks.isRegistered() { logger.Debug("Module hooks already registered") return nil, nil } @@ -320,7 +329,7 @@ func (bm *BasicModule) RegisterHooks(logger *log.Logger) ([]*hooks.ModuleHook, e return nil, fmt.Errorf("register hooks: %w", err) } - bm.hooks.registered = true + bm.hooks.setRegistered() bm.hasReadiness = searchModuleHooksResult.HasReadiness return searchModuleHooksResult.Hooks, nil diff --git a/pkg/module_manager/models/modules/hook_storage.go b/pkg/module_manager/models/modules/hook_storage.go index 8728d4aaf..3e43c29e0 100644 --- a/pkg/module_manager/models/modules/hook_storage.go +++ b/pkg/module_manager/models/modules/hook_storage.go @@ -12,9 +12,13 @@ import ( type HooksStorage struct { registered bool controllersReady bool - lock sync.RWMutex - byBinding map[sh_op_types.BindingType][]*hooks.ModuleHook - byName map[string]*hooks.ModuleHook + // registrationMu serializes whole-module hook registration. Separate from + // lock: registration execs hook binaries and must not hold the index lock + // across that. + registrationMu sync.Mutex + lock sync.RWMutex + byBinding map[sh_op_types.BindingType][]*hooks.ModuleHook + byName map[string]*hooks.ModuleHook } func newHooksStorage() *HooksStorage { @@ -31,12 +35,66 @@ func (hs *HooksStorage) AddHook(hk *hooks.ModuleHook) { hName := hk.GetName() + // Re-registration (e.g. ModuleRun retry after a transient failure) must + // replace the previous object, not accumulate duplicates: a stale entry + // keeps a nil HookController and crashes the kube-events dispatcher. + if old, ok := hs.byName[hName]; ok { + hs.removeHookFromBindings(old) + } + hs.byName[hName] = hk for _, binding := range hk.GetHookConfig().Bindings() { hs.byBinding[binding] = append(hs.byBinding[binding], hk) } } +// removeHookFromBindings deletes all byBinding entries pointing to the given +// hook. Call under hs.lock. +func (hs *HooksStorage) removeHookFromBindings(hk *hooks.ModuleHook) { + for binding, hks := range hs.byBinding { + filtered := make([]*hooks.ModuleHook, 0, len(hks)) + for _, h := range hks { + if h != hk { + filtered = append(filtered, h) + } + } + + if len(filtered) == 0 { + delete(hs.byBinding, binding) + } else { + hs.byBinding[binding] = filtered + } + } +} + +func (hs *HooksStorage) isRegistered() bool { + hs.lock.RLock() + defer hs.lock.RUnlock() + + return hs.registered +} + +func (hs *HooksStorage) setRegistered() { + hs.lock.Lock() + defer hs.lock.Unlock() + + hs.registered = true +} + +func (hs *HooksStorage) isControllersReady() bool { + hs.lock.RLock() + defer hs.lock.RUnlock() + + return hs.controllersReady +} + +func (hs *HooksStorage) setControllersReady() { + hs.lock.Lock() + defer hs.lock.Unlock() + + hs.controllersReady = true +} + func (hs *HooksStorage) getHooks(bt ...sh_op_types.BindingType) []*hooks.ModuleHook { hs.lock.RLock() defer hs.lock.RUnlock() @@ -49,16 +107,22 @@ func (hs *HooksStorage) getHooks(bt ...sh_op_types.BindingType) []*hooks.ModuleH return []*hooks.ModuleHook{} } - sort.Slice(res, func(i, j int) bool { - oi, oj := res[i].Order(t), res[j].Order(t) + // Sort a copy: the shared slice must not be mutated under RLock, and + // callers must not observe later index updates through the returned + // header. + out := make([]*hooks.ModuleHook, len(res)) + copy(out, res) + + sort.Slice(out, func(i, j int) bool { + oi, oj := out[i].Order(t), out[j].Order(t) if oi != oj { return oi < oj } - return res[i].GetName() < res[j].GetName() + return out[i].GetName() < out[j].GetName() }) - return res + return out } // return all hooks diff --git a/pkg/module_manager/models/modules/hook_storage_test.go b/pkg/module_manager/models/modules/hook_storage_test.go new file mode 100644 index 000000000..d77dbd389 --- /dev/null +++ b/pkg/module_manager/models/modules/hook_storage_test.go @@ -0,0 +1,210 @@ +package modules + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/deckhouse/deckhouse/pkg/log" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + + gohook "github.com/flant/addon-operator/pkg/module_manager/go_hook" + "github.com/flant/addon-operator/pkg/module_manager/models/hooks" + "github.com/flant/addon-operator/pkg/module_manager/models/hooks/kind" + "github.com/flant/addon-operator/pkg/utils" + "github.com/flant/shell-operator/pkg/hook/config" + "github.com/flant/shell-operator/pkg/hook/controller" + sh_op_types "github.com/flant/shell-operator/pkg/hook/types" +) + +// newKubeGoHook returns a go hook with a single OnKubernetesEvent binding. +// Config is precompiled, so no binary is executed on InitializeHookConfig. +func newKubeGoHook(name string) *kind.GoHook { + gh := kind.NewGoHook(&gohook.HookConfig{ + Kubernetes: []gohook.KubernetesConfig{ + { + Name: "pods", + ApiVersion: "v1", + Kind: "Pod", + FilterFunc: func(_ *unstructured.Unstructured) (gohook.FilterResult, error) { + return nil, nil + }, + }, + }, + Logger: log.NewNop(), + }, func(_ context.Context, _ *gohook.HookInput) error { return nil }) + + gh.AddMetadata(&gohook.HookMetadata{Name: name, Path: "/hooks/" + name}) + + return gh +} + +// newInitializedModuleHook wraps a go hook into ModuleHook and loads its config, +// so GetHookConfig().Bindings() returns OnKubernetesEvent. +func newInitializedModuleHook(t *testing.T, name string) *hooks.ModuleHook { + t.Helper() + + mh := hooks.NewModuleHook(newKubeGoHook(name)) + require.NoError(t, mh.InitializeHookConfig()) + + return mh +} + +// TestAddHookIsIdempotentPerBinding proves the byBinding duplication bug: +// re-registering a hook with the same name must replace the old entry, +// not append a second one (Trigger A root cause). +func TestAddHookIsIdempotentPerBinding(t *testing.T) { + hs := newHooksStorage() + + first := newInitializedModuleHook(t, "hooks/license") + // A retry creates a fresh object for the same hook (searchModuleHooks + // calls NewModuleHook on every attempt). + second := newInitializedModuleHook(t, "hooks/license") + + hs.AddHook(first) + hs.AddHook(second) + + byName := hs.getHookByName("hooks/license") + require.Same(t, second, byName, "byName keeps the most recent object") + + byBinding := hs.getHooks(sh_op_types.OnKubernetesEvent) + require.Len(t, byBinding, 1, + "byBinding must hold exactly one entry per hook name after re-registration") + require.Same(t, second, byBinding[0], + "byBinding entry must be the most recently registered object") +} + +// flakyGoHook fails config loading a given number of times. +// Simulates the transient exec failure (ECHILD) that aborted hook +// registration mid-loop in production. +type flakyGoHook struct { + *kind.GoHook + + mu sync.Mutex + failures int +} + +func (f *flakyGoHook) GetConfigForModule(moduleKind string) (*config.HookConfig, error) { + f.mu.Lock() + defer f.mu.Unlock() + + if f.failures > 0 { + f.failures-- + return nil, errors.New("exec file '/hooks/go-hooks-bin': waitid: no child processes") + } + + return f.GoHook.GetConfigForModule(moduleKind) +} + +// TestRegistrationRetryLeavesNoOrphanHooks reproduces the production incident: +// attempt 1 publishes some hooks and fails mid-loop, attempt 2 re-registers +// fresh objects and attaches controllers only to them. Every hook visible via +// GetHooks must have a controller afterwards - a nil one is exactly the value +// that crashed the operator with SIGSEGV. +func TestRegistrationRetryLeavesNoOrphanHooks(t *testing.T) { + logger := log.NewNop() + + bm, err := NewBasicModule("flant-integration", t.TempDir(), 1, utils.Values{}, nil, nil) + require.NoError(t, err) + bm.WithDependencies(stubDeps(logger)) + + // Attempt 1: the second hook fails on config load, registerHooks aborts. + // The first hook is already published to byBinding without a controller. + okHook1 := hooks.NewModuleHook(newKubeGoHook("hooks/ok")) + failingHook1 := hooks.NewModuleHook(&flakyGoHook{GoHook: newKubeGoHook("hooks/license"), failures: 1}) + + err = bm.registerHooks([]*hooks.ModuleHook{okHook1, failingHook1}, logger) + require.Error(t, err, "attempt 1 must fail on the failing hook") + + // Attempt 2: ModuleRun retry - fresh hook objects, the failure is gone. + okHook2 := hooks.NewModuleHook(newKubeGoHook("hooks/ok")) + failingHook2 := hooks.NewModuleHook(&flakyGoHook{GoHook: newKubeGoHook("hooks/license")}) + + err = bm.registerHooks([]*hooks.ModuleHook{okHook2, failingHook2}, logger) + require.NoError(t, err, "attempt 2 must succeed") + + // RegisterModuleHooks attaches controllers only to hooks returned by the + // successful attempt, then raises the readiness flag. + for _, hk := range []*hooks.ModuleHook{okHook2, failingHook2} { + hk.WithHookController(controller.NewHookController()) + } + bm.SetHooksControllersReady() + + // Invariant: controllersReady == true implies every hook in byBinding has + // a controller. + kubeHooks := bm.GetHooks(sh_op_types.OnKubernetesEvent) + for _, hk := range kubeHooks { + assert.NotNilf(t, hk.GetHookController(), + "hook %q has no controller: stale duplicate left by failed attempt 1", hk.GetName()) + } + require.Len(t, kubeHooks, 2, + "byBinding must contain one entry per hook, without stale duplicates") +} + +// TestControllersReadyFlagIsRaceFree proves the data race on controllersReady: +// the flag is written by the registration goroutine and read by the kube-events +// dispatcher without synchronization. Run with -race. +func TestControllersReadyFlagIsRaceFree(t *testing.T) { + bm, err := NewBasicModule("race-flag", t.TempDir(), 1, utils.Values{}, nil, nil) + require.NoError(t, err) + + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + for range 1000 { + bm.SetHooksControllersReady() + } + }() + + go func() { + defer wg.Done() + for range 1000 { + _ = bm.HooksControllersReady() + } + }() + + wg.Wait() + require.True(t, bm.HooksControllersReady()) +} + +// TestConcurrentRegisterHooksIsSafe proves Trigger B entry: two ModuleRun +// workers can both pass the `registered` check before either sets it, so the +// module is registered twice (duplicates + data race on the flag). Run with -race. +func TestConcurrentRegisterHooksIsSafe(t *testing.T) { + moduleDir := t.TempDir() + hooksDir := filepath.Join(moduleDir, "hooks") + require.NoError(t, os.MkdirAll(hooksDir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(hooksDir, "startup.sh"), []byte(`#!/bin/sh +if [ "$1" = "--config" ]; then + echo '{"configVersion":"v1","onStartup":10}' + exit 0 +fi +exit 0 +`), 0o755)) + + logger := log.NewNop() + + bm, err := NewBasicModule("concurrent", moduleDir, 1, utils.Values{}, nil, nil) + require.NoError(t, err) + bm.WithDependencies(stubDeps(logger)) + + var wg sync.WaitGroup + for range 2 { + wg.Add(1) + go func() { + defer wg.Done() + _, _ = bm.RegisterHooks(logger) + }() + } + wg.Wait() + + require.Len(t, bm.GetHooks(sh_op_types.OnStartup), 1, + "concurrent registration must not produce duplicate hooks") +} From 125ee1271a052f344ab835d0d4b159184cb5232e Mon Sep 17 00:00:00 2001 From: Roman Berezkin Date: Thu, 9 Jul 2026 20:04:19 +0300 Subject: [PATCH 2/6] fix: aggregate per-hook errors when enabling kubernetes bindings One hook whose monitor could not start aborted HandleModuleEnableKubernetesBindings for the whole module, so the remaining hooks never emitted their Synchronization contexts. Collect errors with errors.Join instead of returning on the first one. Ported from flant/addon-operator#798. Signed-off-by: Roman Berezkin --- pkg/module_manager/module_manager.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/pkg/module_manager/module_manager.go b/pkg/module_manager/module_manager.go index 47c6961ca..550ab9b1b 100644 --- a/pkg/module_manager/module_manager.go +++ b/pkg/module_manager/module_manager.go @@ -1018,6 +1018,13 @@ func (mm *ModuleManager) HandleModuleEnableKubernetesBindings(ctx context.Contex kubeHooks := ml.GetHooks(OnKubernetesEvent) + // Best-effort: try to enable bindings for every hook instead of aborting on the + // first failure. A single hook whose monitor cannot start (e.g. a transient + // apiserver/CRD error during startup) must not prevent the remaining hooks from + // starting their monitors and emitting their Synchronization contexts. Errors are + // aggregated so the caller still retries, and the failing hooks are named. + var errs []error + for _, mh := range kubeHooks { err := mh.GetHookController().HandleEnableKubernetesBindings(ctx, func(info controller.BindingExecutionInfo) { if createTaskFn != nil { @@ -1025,11 +1032,11 @@ func (mm *ModuleManager) HandleModuleEnableKubernetesBindings(ctx context.Contex } }) if err != nil { - return fmt.Errorf("handle enable kubernetes bindings for '%s': %w", mh.GetName(), err) + errs = append(errs, fmt.Errorf("handle enable kubernetes bindings for '%s': %w", mh.GetName(), err)) } } - return nil + return errors.Join(errs...) } func (mm *ModuleManager) EnableModuleScheduleBindings(moduleName string) { From 4135eae755da72f4027e83183743e5712bc52248 Mon Sep 17 00:00:00 2001 From: Roman Berezkin Date: Fri, 10 Jul 2026 15:56:15 +0300 Subject: [PATCH 3/6] chore: bump shell-operator to idempotent EnableKubernetesBindings The best-effort enable loop (errors.Join) is only safe together with the idempotent EnableKubernetesBindings from shell-operator c52b11d: on an older shell-operator, pass 1 starts monitors for every successful hook and the retry then hits the alreadyEnabled early-return, which yields zero Synchronization contexts for all of them - a wider loss window than before the aggregation. Re-point to the tagged release once flant/shell-operator#922 merges. Signed-off-by: Roman Berezkin --- go.mod | 2 +- go.sum | 6 ++---- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/go.mod b/go.mod index bfc428306..5b40f483f 100644 --- a/go.mod +++ b/go.mod @@ -10,7 +10,7 @@ require ( github.com/dominikbraun/graph v0.23.0 github.com/ettle/strcase v0.2.0 github.com/flant/kube-client v1.9.1 - github.com/flant/shell-operator v1.20.2-0.20260707115217-ac90cf8d1033 + github.com/flant/shell-operator v1.20.2-0.20260709170407-c52b11d7fbe8 github.com/go-chi/chi/v5 v5.3.0 github.com/go-openapi/loads v0.23.2 github.com/go-openapi/spec v0.22.1 diff --git a/go.sum b/go.sum index f3ca12757..ae6e4dfef 100644 --- a/go.sum +++ b/go.sum @@ -156,10 +156,8 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/flant/kube-client v1.9.1 h1:B5Sa++5arl3N/KBdL1KNMlUBtzQXKz1o2FIiFb4X+VA= github.com/flant/kube-client v1.9.1/go.mod h1:jAQ01+f5Q8UJJ4RYZlk5xqoA+OOalpr/QJ3e3Q/saqw= -github.com/flant/shell-operator v1.20.2-0.20260707103551-b1bf9c86d4e6 h1:yDFBK1LSNs/olu0JEjNM+4bbSlM07p6YT5ZXWyeLdTk= -github.com/flant/shell-operator v1.20.2-0.20260707103551-b1bf9c86d4e6/go.mod h1:Qj8agNbfPm6x7C3TmCpY1fCtayhqS2Sxncui7grrOVw= -github.com/flant/shell-operator v1.20.2-0.20260707115217-ac90cf8d1033 h1:WsxzIwmXv3zR751CacmEXRMKeZ42gQjFED+sTbBFu0g= -github.com/flant/shell-operator v1.20.2-0.20260707115217-ac90cf8d1033/go.mod h1:Qj8agNbfPm6x7C3TmCpY1fCtayhqS2Sxncui7grrOVw= +github.com/flant/shell-operator v1.20.2-0.20260709170407-c52b11d7fbe8 h1:YoRN3pd10muB4z4f+fxsSnKe/oOqKbvdOpt05iJcx+M= +github.com/flant/shell-operator v1.20.2-0.20260709170407-c52b11d7fbe8/go.mod h1:Qj8agNbfPm6x7C3TmCpY1fCtayhqS2Sxncui7grrOVw= github.com/flopp/go-findfont v0.1.0 h1:lPn0BymDUtJo+ZkV01VS3661HL6F4qFlkhcJN55u6mU= github.com/flopp/go-findfont v0.1.0/go.mod h1:wKKxRDjD024Rh7VMwoU90i6ikQRCr+JTHB5n4Ejkqvw= github.com/fluxcd/flagger v1.36.1 h1:X2PumtNwZz9YSGaOtZLFm2zAKLgHhFkbNv8beg7ifyc= From 8f37af1712a520b1308e9d3b84a26f759faa9230 Mon Sep 17 00:00:00 2001 From: Roman Berezkin Date: Fri, 10 Jul 2026 15:56:30 +0300 Subject: [PATCH 4/6] test: cover Synchronization retry after a partial enable failure Encodes the acceptance scenario for the lost-Synchronization incident: pass 1 fails on one hook while the remaining hooks still emit their Synchronization contexts, pass 2 re-emits contexts for every hook. The failing hook sorts first, so the test catches both regressions: a revert of the best-effort aggregation in HandleModuleEnableKubernetesBindings (pass 1 would emit zero contexts) and a non-idempotent shell-operator Enable (pass 2 would emit only the failed hook's context). Signed-off-by: Roman Berezkin --- .../module_manager_sync_retry_test.go | 121 ++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 pkg/module_manager/module_manager_sync_retry_test.go diff --git a/pkg/module_manager/module_manager_sync_retry_test.go b/pkg/module_manager/module_manager_sync_retry_test.go new file mode 100644 index 000000000..40a8d375c --- /dev/null +++ b/pkg/module_manager/module_manager_sync_retry_test.go @@ -0,0 +1,121 @@ +package module_manager + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "testing" + + "github.com/deckhouse/deckhouse/pkg/log" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/flant/addon-operator/pkg/module_manager/models/hooks" + "github.com/flant/addon-operator/pkg/module_manager/models/modules" + "github.com/flant/addon-operator/pkg/utils" + "github.com/flant/kube-client/fake" + "github.com/flant/shell-operator/pkg/hook/controller" + kubeeventsmanager "github.com/flant/shell-operator/pkg/kube_events_manager" +) + +// flakyMonitorSource wraps the real KubeEventsManager and fails AddMonitor for +// the given kind a limited number of times - the transient apiserver error that +// aborts EnableKubernetesBindings for a single hook. +type flakyMonitorSource struct { + kubeeventsmanager.KubeEventsManager + + mu sync.Mutex + failKind string + failures int +} + +func (f *flakyMonitorSource) AddMonitor(config *kubeeventsmanager.MonitorConfig) error { + f.mu.Lock() + defer f.mu.Unlock() + + if config.Kind == f.failKind && f.failures > 0 { + f.failures-- + return errors.New("transient: apiserver unavailable") + } + + return f.KubeEventsManager.AddMonitor(config) +} + +func writeShellHook(t *testing.T, hooksDir, name, kind string) { + t.Helper() + + script := `#!/bin/sh +if [ "$1" = "--config" ]; then + echo '{"configVersion":"v1","kubernetes":[{"name":"objects","apiVersion":"v1","kind":"` + kind + `","executeHookOnEvent":["Added"]}]}' + exit 0 +fi +exit 0 +` + require.NoError(t, os.WriteFile(filepath.Join(hooksDir, name), []byte(script), 0o755)) +} + +// TestHandleModuleEnableKubernetesBindings_RetryAfterPartialFailure encodes the +// acceptance scenario for the lost-Synchronization incident: on the first pass +// one hook cannot start its monitor, the other hooks still emit their +// Synchronization contexts and the error names the failing hook; on the retry +// pass every hook emits its Synchronization context again, so nothing is lost +// and the module can proceed. +// The failing hook sorts FIRST: with the old fail-fast loop in +// HandleModuleEnableKubernetesBindings the pass-1 assert would then see zero +// contexts, so a revert of the best-effort aggregation fails the test too. +// Requires shell-operator with idempotent EnableKubernetesBindings (>= v1.20.2). +func TestHandleModuleEnableKubernetesBindings_RetryAfterPartialFailure(t *testing.T) { + logger := log.NewNop() + + moduleDir := t.TempDir() + hooksDir := filepath.Join(moduleDir, "hooks") + require.NoError(t, os.MkdirAll(hooksDir, 0o755)) + writeShellHook(t, hooksDir, "0-fail.sh", "ConfigMap") // the failing one, sorts first + writeShellHook(t, hooksDir, "a.sh", "Pod") + writeShellHook(t, hooksDir, "b.sh", "Pod") + + fc := fake.NewFakeCluster(fake.ClusterVersionV121) + mgr := kubeeventsmanager.NewKubeEventsManager(context.Background(), fc.Client, logger) + source := &flakyMonitorSource{KubeEventsManager: mgr, failKind: "ConfigMap", failures: 1} + + bm, err := modules.NewBasicModule("retry-sync", moduleDir, 1, utils.Values{}, nil, nil) + require.NoError(t, err) + // Registration dereferences the container for shell hooks; an empty one is enough. + bm.WithDependencies(&hooks.HookExecutionDependencyContainer{}) + + hks, err := bm.RegisterHooks(logger) + require.NoError(t, err) + require.Len(t, hks, 3) + + for _, hk := range hks { + hookCtrl := controller.NewHookController() + hookCtrl.InitKubernetesBindings(hk.GetHookConfig().OnKubernetesEvents, source, logger) + hk.WithHookController(hookCtrl) + } + bm.SetHooksControllersReady() + + mm := NewModuleManager(context.Background(), &ModuleManagerConfig{}, logger) + mm.modules.Add(bm) + + synced := make(map[string]int) + collect := func(mh *hooks.ModuleHook, _ controller.BindingExecutionInfo) { + synced[filepath.Base(mh.GetName())]++ + } + + // Pass 1: the first hook fails, the remaining ones must still emit their + // Synchronization contexts (with fail-fast the map would be empty). + err = mm.HandleModuleEnableKubernetesBindings(context.Background(), "retry-sync", collect) + require.Error(t, err) + assert.Contains(t, err.Error(), "0-fail.sh", "the error must name the failing hook") + assert.Equal(t, map[string]int{"a.sh": 1, "b.sh": 1}, synced, + "hooks after the failing one must emit Synchronization contexts on the failing pass") + + // Pass 2 (ModuleRun retry): every hook emits its Synchronization context again. + clear(synced) + err = mm.HandleModuleEnableKubernetesBindings(context.Background(), "retry-sync", collect) + require.NoError(t, err) + assert.Equal(t, map[string]int{"0-fail.sh": 1, "a.sh": 1, "b.sh": 1}, synced, + "retry must emit Synchronization for previously succeeded hooks too (idempotent Enable)") +} From 6c4521c4969ddaa81a79f1f4364d11852bbdfd94 Mon Sep 17 00:00:00 2001 From: Roman Berezkin Date: Fri, 10 Jul 2026 15:56:30 +0300 Subject: [PATCH 5/6] fix: do not silently drop Synchronization tasks when a queue is missing A Synchronization task that is built but never queued left its binding invisible to SynchronizationState.IsCompleted (vacuously completed) and its kubernetes events locked forever, with only an Error log. - ensure hook queues exist at the start of the QueueSynchronizationTasks phase: queues are normally created in the Startup phase only when DoModuleStartup is set, and a converge restart can displace that task, making every later AddLastTaskToQueue fail deterministically - collect queueing errors and fail the phase instead of skipping the task: the retry rebuilds every Synchronization context because EnableKubernetesBindings is idempotent - log 'append' only for tasks that were actually queued Signed-off-by: Roman Berezkin --- pkg/task/tasks/module-run/task.go | 44 ++++++++++++++++++++++++++----- 1 file changed, 37 insertions(+), 7 deletions(-) diff --git a/pkg/task/tasks/module-run/task.go b/pkg/task/tasks/module-run/task.go index 01ade299a..78a3aafd3 100644 --- a/pkg/task/tasks/module-run/task.go +++ b/pkg/task/tasks/module-run/task.go @@ -2,6 +2,8 @@ package modulerun import ( "context" + "errors" + "fmt" "log/slog" "runtime/trace" "strings" @@ -235,6 +237,12 @@ func (s *Task) Handle(ctx context.Context) (res queue.TaskResult) { //nolint:non s.logger.Debug("ModuleRun phase", slog.String(pkg.LogKeyPhase, string(baseModule.GetPhase()))) + // Hook queues are normally created in the Startup phase, but only when + // DoModuleStartup is set, and a converge restart can displace that task. + // A missing queue would make every AddLastTaskToQueue below fail + // deterministically, so ensure the queues exist (the call is idempotent). + s.CreateAndStartQueuesForModuleHooks(hm.ModuleName) + // ModuleHookRun.Synchronization tasks for bindings with the "main" queue. mainSyncTasks := make([]sh_task.Task, 0) // ModuleHookRun.Synchronization tasks to add in parallel queues. @@ -310,30 +318,52 @@ func (s *Task) Handle(ctx context.Context) (res queue.TaskResult) { //nolint:non // Fail to enable bindings: cannot start Kubernetes monitors. moduleRunErr = err } else { + // A Synchronization task that is built but never queued would leave its + // binding invisible to SynchronizationState.IsCompleted (vacuously + // completed) and its kubernetes events locked forever. Collect queueing + // errors and fail the phase instead of dropping tasks silently: the + // retry rebuilds every Synchronization context, EnableKubernetesBindings + // is idempotent. + var queueErrs []error + + queued := make([]sh_task.Task, 0, len(parallelSyncTasksToWait)+len(parallelSyncTasks)) + // Queue parallel tasks that should be waited. for _, tsk := range parallelSyncTasksToWait { if err := s.queueService.AddLastTaskToQueue(tsk.GetQueueName(), tsk); err != nil { - s.logger.Error("queue is not found while EnableKubernetesBindings task", - slog.String(pkg.LogKeyQueue, tsk.GetQueueName())) + queueErrs = append(queueErrs, + fmt.Errorf("queue Synchronization task to '%s': %w", tsk.GetQueueName(), err)) continue } + queued = append(queued, tsk) + thm := task.HookMetadataAccessor(tsk) baseModule.Synchronization().QueuedForBinding(thm) } - s.logTaskAdd("append", parallelSyncTasksToWait...) - // Queue regular parallel tasks. for _, tsk := range parallelSyncTasks { if err := s.queueService.AddLastTaskToQueue(tsk.GetQueueName(), tsk); err != nil { - s.logger.Error("queue is not found while EnableKubernetesBindings task", - slog.String(pkg.LogKeyQueue, tsk.GetQueueName())) + queueErrs = append(queueErrs, + fmt.Errorf("queue Synchronization task to '%s': %w", tsk.GetQueueName(), err)) + + continue } + + queued = append(queued, tsk) } - s.logTaskAdd("append", parallelSyncTasks...) + s.logTaskAdd("append", queued...) + + if len(queueErrs) > 0 { + moduleRunErr = errors.Join(queueErrs...) + + res.Status = queue.Repeat + + return res + } if len(parallelSyncTasksToWait) == 0 { // Skip waiting tasks in parallel queues, proceed to schedule bindings. From 39f74737d964a2ee11e4a2e83c3d46faebd7b872 Mon Sep 17 00:00:00 2001 From: Roman Berezkin Date: Fri, 10 Jul 2026 16:25:53 +0300 Subject: [PATCH 6/6] chore: bump shell-operator to v1.20.2 Release tag for flant/shell-operator#922 (tree-identical to the previously pinned branch pseudo-version): panic isolation in ManagerEventsHandler, idempotent self-healing EnableKubernetesBindings, binding_monitor_missing_total, FactoryStore lifetime owned by the manager + factory_informer_dead_total. The best-effort enable loop (errors.Join) is only safe together with the idempotent EnableKubernetesBindings from this release: on an older shell-operator, pass 1 starts monitors for every successful hook and the retry then hits the alreadyEnabled early-return, which yields zero Synchronization contexts for all of them. Signed-off-by: Roman Berezkin --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 5b40f483f..118374e07 100644 --- a/go.mod +++ b/go.mod @@ -10,7 +10,7 @@ require ( github.com/dominikbraun/graph v0.23.0 github.com/ettle/strcase v0.2.0 github.com/flant/kube-client v1.9.1 - github.com/flant/shell-operator v1.20.2-0.20260709170407-c52b11d7fbe8 + github.com/flant/shell-operator v1.20.2 github.com/go-chi/chi/v5 v5.3.0 github.com/go-openapi/loads v0.23.2 github.com/go-openapi/spec v0.22.1 diff --git a/go.sum b/go.sum index ae6e4dfef..c84e18995 100644 --- a/go.sum +++ b/go.sum @@ -156,8 +156,8 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/flant/kube-client v1.9.1 h1:B5Sa++5arl3N/KBdL1KNMlUBtzQXKz1o2FIiFb4X+VA= github.com/flant/kube-client v1.9.1/go.mod h1:jAQ01+f5Q8UJJ4RYZlk5xqoA+OOalpr/QJ3e3Q/saqw= -github.com/flant/shell-operator v1.20.2-0.20260709170407-c52b11d7fbe8 h1:YoRN3pd10muB4z4f+fxsSnKe/oOqKbvdOpt05iJcx+M= -github.com/flant/shell-operator v1.20.2-0.20260709170407-c52b11d7fbe8/go.mod h1:Qj8agNbfPm6x7C3TmCpY1fCtayhqS2Sxncui7grrOVw= +github.com/flant/shell-operator v1.20.2 h1:rI2lIkhZxIfMVR0NFMSZ5xyXMvtBj+Wai2MPopb/Yp0= +github.com/flant/shell-operator v1.20.2/go.mod h1:Qj8agNbfPm6x7C3TmCpY1fCtayhqS2Sxncui7grrOVw= github.com/flopp/go-findfont v0.1.0 h1:lPn0BymDUtJo+ZkV01VS3661HL6F4qFlkhcJN55u6mU= github.com/flopp/go-findfont v0.1.0/go.mod h1:wKKxRDjD024Rh7VMwoU90i6ikQRCr+JTHB5n4Ejkqvw= github.com/fluxcd/flagger v1.36.1 h1:X2PumtNwZz9YSGaOtZLFm2zAKLgHhFkbNv8beg7ifyc=