[feature] separate function and critical modules

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

Author: ipaqsa

pkg/module_manager/module_manager.go

@@ -604,6 +604,15 @@ func (mm *ModuleManager) GetModule(name string) *modules.BasicModule {
 	return mm.modules.Get(name)
 }
 
+func (mm *ModuleManager) GetCritical(moduleName string) bool {
+	module := mm.GetModule(moduleName)
+	if module == nil {
+		return false
+	}
+
+	return module.GetCritical()
+}
+
 func (mm *ModuleManager) GetFunctionalDependencies() map[string][]string {
 	return mm.moduleScheduler.GetFunctionalDependencies()
 }

pkg/task/functional/scheduler.go

@@ -24,14 +24,8 @@ 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
+	requests  map[string]*Request
 	done      map[string]struct{}
 	scheduled map[string]struct{}
 
@@ -48,48 +42,27 @@ type Request struct {
 	Labels       map[string]string
 }
 
-func NewScheduler(qService *queue.Service, logger *log.Logger) *Scheduler {
-	return &Scheduler{
+// NewScheduler creates a scheduler instance and starts it
+func NewScheduler(ctx context.Context, qService *queue.Service, logger *log.Logger) *Scheduler {
+	s := &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()
+		scheduled:    make(map[string]struct{}),
+		requests:     make(map[string]*Request),
+		done:         make(map[string]struct{}),
+		doneCh:       make(chan string, channelsBuffer),
+		processCh:    make(chan *Request, channelsBuffer),
 	}
 
-	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)
+		s.runScheduleLoop(ctx)
 	}()
 
 	go func() {
-		defer s.wg.Done()
-		s.runProcessLoop(batchCtx)
+		s.runProcessLoop(ctx)
 	}()
+
+	return s
 }
 
 // runScheduleLoop launches the scheduling loop for a batch.
@@ -121,6 +94,28 @@ func (s *Scheduler) runProcessLoop(ctx context.Context) {
 	}
 }
 
+// Add adds request to process
+func (s *Scheduler) Add(reqs ...*Request) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	for _, req := range reqs {
+		s.logger.Debug("add request", slog.Any("request", req))
+		s.requests[req.Name] = req
+		delete(s.done, req.Name)
+		delete(s.scheduled, req.Name)
+	}
+}
+
+// Remove removes module from done
+// TODO(ipaqsa): add stop module run task
+func (s *Scheduler) Remove(name string) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	delete(s.done, name)
+}
+
 // reschedule marks module done and schedule new modules to be processed
 func (s *Scheduler) reschedule(name string) {
 	if name != Root {
@@ -155,7 +150,7 @@ func (s *Scheduler) reschedule(name string) {
 
 		// schedule module if ready
 		if ready {
-			s.logger.Debug("trigger module scheduling", slog.String("module", req.Name), slog.Any("trigger", name))
+			s.logger.Debug("trigger scheduling", slog.String("scheduled", req.Name), slog.Any("done", name))
 			s.scheduled[req.Name] = struct{}{}
 			s.processCh <- req
 		}
@@ -193,17 +188,3 @@ func (s *Scheduler) Finished() bool {
 
 	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/service/service.go

@@ -109,7 +109,7 @@ 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.functionalScheduler = functional.NewScheduler(ctx, svc.queueService, logger.Named("functional-scheduler"))
 
 	svc.initFactory()
 

pkg/task/tasks/converge-modules/task.go

@@ -381,14 +381,9 @@ func (s *Task) CreateConvergeModulesTasks(ctx context.Context,
 			continue
 		}
 
-		basicModule := s.moduleManager.GetModule(modules[0])
-		if basicModule == nil {
-			continue
-		}
-
-		if !basicModule.GetCritical() {
+		// skip functional modules
+		if !s.moduleManager.GetCritical(modules[0]) {
 			functionalModules = append(functionalModules, modules...)
-			continue
 		}
 
 		newLogLabels := utils.MergeLabels(logLabels)
@@ -521,7 +516,7 @@ func (s *Task) CreateConvergeModulesTasks(ctx context.Context,
 
 	// schedule functional modules in parallel queues
 	if len(schedulerRequests) > 0 {
-		s.functionalScheduler.Start(ctx, schedulerRequests)
+		s.functionalScheduler.Add(schedulerRequests...)
 	}
 
 	// as resultingTasks contains new ensureCRDsTasks we invalidate