From 77aa3f59a3bcb77ae9f6ad7d1f8f55f2dfe4dc8d Mon Sep 17 00:00:00 2001 From: Artem Kuleshov Date: Fri, 21 Aug 2026 17:38:25 +0300 Subject: [PATCH 1/2] [addon-operator] create module hook queues regardless of DoModuleStartup Signed-off-by: Artem Kuleshov --- pkg/task/tasks/module-run/task.go | 50 ++++++++++++++++++++++++------- 1 file changed, 40 insertions(+), 10 deletions(-) diff --git a/pkg/task/tasks/module-run/task.go b/pkg/task/tasks/module-run/task.go index 01ade299a..730b80906 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" @@ -186,15 +188,17 @@ func (s *Task) Handle(ctx context.Context) (res queue.TaskResult) { //nolint:non // Register module hooks on every enable. moduleRunErr = s.moduleManager.RegisterModuleHooks(baseModule, taskLogLabels) if moduleRunErr == nil { + // Start queues for module hooks. Every phase below queues tasks into them, + // and the phases are reached with or without module startup, so the queues + // must not depend on DoModuleStartup (the call is idempotent). + s.CreateAndStartQueuesForModuleHooks(baseModule.GetName()) + if hm.DoModuleStartup { s.logger.Debug("ModuleRun phase", slog.String(pkg.LogKeyPhase, string(baseModule.GetPhase()))) treg := trace.StartRegion(context.Background(), "ModuleRun-OnStartup") - // Start queues for module hooks. - s.CreateAndStartQueuesForModuleHooks(baseModule.GetName()) - // Run onStartup hooks. moduleRunErr = s.moduleManager.RunModuleHooks(ctx, baseModule, htypes.OnStartup, s.shellTask.GetLogLabels()) if moduleRunErr == nil { @@ -235,6 +239,10 @@ func (s *Task) Handle(ctx context.Context) (res queue.TaskResult) { //nolint:non s.logger.Debug("ModuleRun phase", slog.String(pkg.LogKeyPhase, string(baseModule.GetPhase()))) + // Queues are created in the Startup phase; ensure them again so that a hook + // registered later cannot lose its Synchronization task. 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 81823d06bf8438719ffae7888e8c365f6d77c7c9 Mon Sep 17 00:00:00 2001 From: Artem Kuleshov Date: Fri, 21 Aug 2026 18:00:01 +0300 Subject: [PATCH 2/2] [addon-operator] run module startup phase regardless of DoModuleStartup Signed-off-by: Artem Kuleshov --- pkg/task/tasks/module-run/task.go | 27 +++++++++++---------------- 1 file changed, 11 insertions(+), 16 deletions(-) diff --git a/pkg/task/tasks/module-run/task.go b/pkg/task/tasks/module-run/task.go index 730b80906..56cd9af57 100644 --- a/pkg/task/tasks/module-run/task.go +++ b/pkg/task/tasks/module-run/task.go @@ -188,28 +188,23 @@ func (s *Task) Handle(ctx context.Context) (res queue.TaskResult) { //nolint:non // Register module hooks on every enable. moduleRunErr = s.moduleManager.RegisterModuleHooks(baseModule, taskLogLabels) if moduleRunErr == nil { - // Start queues for module hooks. Every phase below queues tasks into them, - // and the phases are reached with or without module startup, so the queues - // must not depend on DoModuleStartup (the call is idempotent). - s.CreateAndStartQueuesForModuleHooks(baseModule.GetName()) - - if hm.DoModuleStartup { - s.logger.Debug("ModuleRun phase", - slog.String(pkg.LogKeyPhase, string(baseModule.GetPhase()))) + s.logger.Debug("ModuleRun phase", + slog.String(pkg.LogKeyPhase, string(baseModule.GetPhase()))) - treg := trace.StartRegion(context.Background(), "ModuleRun-OnStartup") + treg := trace.StartRegion(context.Background(), "ModuleRun-OnStartup") - // Run onStartup hooks. - moduleRunErr = s.moduleManager.RunModuleHooks(ctx, baseModule, htypes.OnStartup, s.shellTask.GetLogLabels()) - if moduleRunErr == nil { - s.moduleManager.SetModulePhaseAndNotify(baseModule, modules.OnStartupDone) - } + // Start queues for module hooks. + s.CreateAndStartQueuesForModuleHooks(baseModule.GetName()) - treg.End() - } else { + // Run onStartup hooks. The Startup phase already means they have not run, so this + // must not depend on DoModuleStartup, which each ModuleRun producer computes itself. + moduleRunErr = s.moduleManager.RunModuleHooks(ctx, baseModule, htypes.OnStartup, s.shellTask.GetLogLabels()) + if moduleRunErr == nil { s.moduleManager.SetModulePhaseAndNotify(baseModule, modules.OnStartupDone) } + treg.End() + res.Status = queue.Repeat return res