[feature] separate function and critical modules

Signed-off-by: Stepan Paksashvili <stepan.paksashvili@flant.com>

Author: ipaqsa

pkg/app/app.go

@@ -53,7 +53,7 @@ var (
 	CRDsFilters = "doc-,_"
 
 	// NumberOfParallelQueues defines the number of precreated parallel queues for parallel execution
-	NumberOfParallelQueues   = 15
+	NumberOfParallelQueues   = 20
 	ParallelQueuePrefix      = "parallel_queue"
 	ParallelQueueNamePattern = ParallelQueuePrefix + "_%d"
 )

pkg/module_manager/module_manager.go

@@ -604,6 +604,10 @@ func (mm *ModuleManager) GetModule(name string) *modules.BasicModule {
 	return mm.modules.Get(name)
 }
 
+func (mm *ModuleManager) GetFunctionalDependencies() map[string][]string {
+	return mm.moduleScheduler.GetFunctionalDependencies()
+}
+
 func (mm *ModuleManager) GetModuleNames() []string {
 	return mm.modules.NamesInOrder()
 }

pkg/module_manager/scheduler/scheduler.go

@@ -550,6 +550,42 @@ func (s *Scheduler) GetUpdatedByExtender(moduleName string) (string, error) {
 	return vertex.GetUpdatedBy(), err
 }
 
+func (s *Scheduler) GetFunctionalDependencies() map[string][]string {
+	result := make(map[string][]string, len(s.modules))
+
+	for _, module := range s.modules {
+		for _, ext := range s.extenders {
+			if tope, ok := ext.ext.(extenders.TopologicalExtender); ok {
+				if parents := tope.GetTopologicalHints(module.GetName()); len(parents) > 0 {
+					var deps []string
+					for _, parent := range parents {
+						if !s.IsModuleCritical(parent) {
+							deps = append(deps, parent)
+						}
+					}
+
+					result[module.GetName()] = deps
+				}
+				break
+			}
+		}
+	}
+
+	return result
+}
+
+func (s *Scheduler) IsModuleCritical(moduleName string) bool {
+	s.l.Lock()
+	defer s.l.Unlock()
+
+	vertex, err := s.dag.Vertex(moduleName)
+	if err != nil {
+		return false
+	}
+
+	return vertex.GetModule().GetCritical()
+}
+
 func (s *Scheduler) IsModuleEnabled(moduleName string) bool {
 	s.l.Lock()
 	defer s.l.Unlock()

pkg/task/functional/scheduler.go

@@ -0,0 +1,209 @@
+package functional
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+
+	"github.com/deckhouse/deckhouse/pkg/log"
+
+	"github.com/flant/addon-operator/pkg/app"
+	"github.com/flant/addon-operator/pkg/task"
+	"github.com/flant/addon-operator/pkg/task/queue"
+	sh_task "github.com/flant/shell-operator/pkg/task"
+)
+
+const (
+	channelsBuffer = 24
+
+	Root = ""
+)
+
+type Scheduler struct {
+	queueService *queue.Service
+	logger       *log.Logger
+
+	// batch control
+	cancel context.CancelFunc
+
+	// for safe shutdown on replacement
+	wg *sync.WaitGroup
+
+	mtx       sync.Mutex
+	requests  []*Request
+	done      map[string]struct{}
+	scheduled map[string]struct{}
+
+	doneCh    chan string
+	processCh chan *Request
+}
+
+type Request struct {
+	Name         string
+	Description  string
+	Dependencies []string
+	IsReloadAll  bool
+	DoStartup    bool
+	Labels       map[string]string
+}
+
+func NewScheduler(qService *queue.Service, logger *log.Logger) *Scheduler {
+	return &Scheduler{
+		queueService: qService,
+		logger:       logger,
+		wg:           new(sync.WaitGroup),
+	}
+}
+
+// Start schedules a new batch, canceling the previous one if active.
+func (s *Scheduler) Start(ctx context.Context, modules []*Request) {
+	// cancel the previous batch.
+	if s.cancel != nil {
+		s.cancel()
+		// wait for batch goroutines to finish
+		s.wg.Wait()
+	}
+
+	s.logger.Debug("following functional modules will be scheduled", slog.Any("modules", modules))
+
+	// initialize new batch state.
+	batchCtx, cancel := context.WithCancel(ctx)
+	s.cancel = cancel
+
+	s.mtx.Lock()
+	s.done = make(map[string]struct{}, len(modules))
+	s.scheduled = make(map[string]struct{}, len(modules))
+	s.requests = modules
+	s.mtx.Unlock()
+
+	s.doneCh = make(chan string, channelsBuffer)
+	s.processCh = make(chan *Request, channelsBuffer)
+
+	s.wg.Add(2)
+	go func() {
+		defer s.wg.Done()
+		s.runScheduleLoop(batchCtx)
+	}()
+
+	go func() {
+		defer s.wg.Done()
+		s.runProcessLoop(batchCtx)
+	}()
+}
+
+// runScheduleLoop launches the scheduling loop for a batch.
+func (s *Scheduler) runScheduleLoop(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case name, ok := <-s.doneCh:
+			if !ok {
+				return
+			}
+			s.reschedule(name)
+		}
+	}
+}
+
+// runProcessLoop waits for requests to be processed
+func (s *Scheduler) runProcessLoop(ctx context.Context) {
+	var idx int
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case req := <-s.processCh:
+			s.handleRequest(idx, req)
+			idx++
+		}
+	}
+}
+
+// reschedule marks module done and schedule new modules to be processed
+func (s *Scheduler) reschedule(name string) {
+	if name != Root {
+		// skip modules that not present in the batch
+		if _, ok := s.scheduled[name]; !ok {
+			return
+		}
+
+		// mark module done
+		s.done[name] = struct{}{}
+	}
+
+	for _, req := range s.requests {
+		// skip already processed
+		if _, ok := s.done[req.Name]; ok {
+			continue
+		}
+
+		// skip already scheduled
+		if _, ok := s.scheduled[req.Name]; ok {
+			continue
+		}
+
+		// check if all dependencies done
+		ready := true
+		for _, dep := range req.Dependencies {
+			if _, ok := s.done[dep]; !ok {
+				ready = false
+				break
+			}
+		}
+
+		// schedule module if ready
+		if ready {
+			s.logger.Debug("trigger module scheduling", slog.String("module", req.Name), slog.Any("trigger", name))
+			s.scheduled[req.Name] = struct{}{}
+			s.processCh <- req
+		}
+	}
+}
+
+// handleRequest creates a ModuleRun task for request in a parallel queue
+func (s *Scheduler) handleRequest(idx int, req *Request) {
+	queueName := fmt.Sprintf(app.ParallelQueueNamePattern, idx%(app.NumberOfParallelQueues-1))
+
+	moduleTask := sh_task.NewTask(task.ModuleRun).
+		WithLogLabels(req.Labels).
+		WithQueueName(queueName).
+		WithMetadata(task.HookMetadata{
+			EventDescription: req.Description,
+			ModuleName:       req.Name,
+			DoModuleStartup:  req.DoStartup,
+			IsReloadAll:      req.IsReloadAll,
+		})
+
+	_ = s.queueService.AddLastTaskToQueue(queueName, moduleTask)
+}
+
+// Done sends signal that module processing done
+func (s *Scheduler) Done(name string) {
+	if s.doneCh != nil {
+		s.doneCh <- name
+	}
+}
+
+// Finished defines if processing done
+func (s *Scheduler) Finished() bool {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	return len(s.done) == len(s.requests)
+}
+
+// Stop is the graceful shutdown
+func (s *Scheduler) Stop() {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	if s.cancel != nil {
+		s.cancel()
+		s.wg.Wait()
+
+		close(s.doneCh)
+		close(s.processCh)
+	}
+}

pkg/task/hook_metadata.go

@@ -27,6 +27,7 @@ type HookMetadata struct {
 	BindingType         types.BindingType
 	BindingContext      []bindingcontext.BindingContext
 	AllowFailure        bool // Task considered as 'ok' if hook failed. False by default. Can be true for some schedule hooks.
+	Critical            bool
 
 	DoModuleStartup bool // Execute onStartup and kubernetes@Synchronization hooks for module
 	IsReloadAll     bool // ModuleRun task is a part of 'Reload all modules' process.

pkg/task/queue/queue.go

@@ -68,6 +68,15 @@ func (s *Service) AddLastTaskToMain(t sh_task.Task) error {
 	return nil
 }
 
+func (s *Service) GetQueueLength(queueName string) int {
+	q := s.engine.TaskQueues.GetByName(queueName)
+	if q == nil {
+		return 0
+	}
+
+	return q.Length()
+}
+
 func (s *Service) AddLastTaskToQueue(queueName string, t sh_task.Task) error {
 	q := s.engine.TaskQueues.GetByName(queueName)
 	if q == nil {

pkg/task/service/converge.go

@@ -106,7 +106,7 @@ func (s *TaskHandlerService) UpdateFirstConvergeStatus(convergeTasks int) {
 	case converge.FirstStarted:
 		// Mark as done when all convergence tasks are completed
 		if convergeTasks == 0 {
-			log.Info("First converge is finished. Operator is ready now.")
+			s.logger.Debug("first converge is finished. Operator is ready now.")
 			s.convergeState.SetFirstRunPhase(converge.FirstDone)
 		}
 	}

pkg/task/service/service.go

@@ -14,6 +14,7 @@ import (
 	"github.com/flant/addon-operator/pkg/module_manager"
 	"github.com/flant/addon-operator/pkg/task"
 	discovercrds "github.com/flant/addon-operator/pkg/task/discover-crds"
+	"github.com/flant/addon-operator/pkg/task/functional"
 	paralleltask "github.com/flant/addon-operator/pkg/task/parallel"
 	taskqueue "github.com/flant/addon-operator/pkg/task/queue"
 	applykubeconfigvalues "github.com/flant/addon-operator/pkg/task/tasks/apply-kube-config-values"
@@ -57,6 +58,8 @@ type TaskHandlerService struct {
 	// a map of channels to communicate with parallel queues and its lock
 	parallelTaskChannels *paralleltask.TaskChannels
 
+	functionalScheduler *functional.Scheduler
+
 	helm *helm.ClientFactory
 
 	// helmResourcesManager monitors absent resources created for modules.
@@ -106,6 +109,8 @@ func NewTaskHandlerService(ctx context.Context, config *TaskHandlerServiceConfig
 		Handle: svc.Handle,
 	}, logger.Named("task-queue-service"))
 
+	svc.functionalScheduler = functional.NewScheduler(svc.queueService, logger.Named("functional-scheduler"))
+
 	svc.initFactory()
 
 	return svc
@@ -175,8 +180,33 @@ func (s *TaskHandlerService) ParallelHandle(ctx context.Context, t sh_task.Task)
 	s.logTaskEnd(t, res, logger)
 
 	hm := task.HookMetadataAccessor(t)
+
+	s.logger.Debug("parallel task done",
+		slog.String("task_type", string(t.GetType())),
+		slog.String("module", hm.ModuleName),
+		slog.Bool("critical", hm.Critical),
+		slog.String("result", string(res.Status)))
+
+	if !hm.Critical {
+		if t.GetType() == task.ModuleRun {
+			if res.Status == queue.Success && len(res.AfterTasks) == 0 {
+				s.functionalScheduler.Done(hm.ModuleName)
+			}
+
+			if res.Status == queue.Fail {
+				if s.queueService.GetQueueLength(t.GetQueueName()) > 1 {
+					res.Status = queue.Success
+					_ = s.queueService.AddLastTaskToQueue(t.GetQueueName(), t)
+				}
+			}
+		}
+
+		return res
+	}
+
 	if hm.ParallelRunMetadata == nil || len(hm.ParallelRunMetadata.ChannelId) == 0 {
 		s.logger.Warn("Parallel task had no communication channel set")
+		return res
 	}
 
 	if parallelChannel, ok := s.parallelTaskChannels.Get(hm.ParallelRunMetadata.ChannelId); ok {
@@ -242,6 +272,10 @@ func (s *TaskHandlerService) GetQueueService() *taskqueue.Service {
 	return s.queueService
 }
 
+func (s *TaskHandlerService) GetFunctionalScheduler() *functional.Scheduler {
+	return s.functionalScheduler
+}
+
 func (s *TaskHandlerService) GetConvergeState() *converge.ConvergeState {
 	return s.convergeState
 }