[addon-operator] fix: make module hook registration idempotent and race-free

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
 	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

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.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 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=

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

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

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")
+}