diff --git a/go.mod b/go.mod index 380d84c81..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.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 diff --git a/go.sum b/go.sum index df944af02..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.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= 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") +} 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) { 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)") +} 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.