feat: add batching integration for module operations

Signed-off-by: Evsyukov Denis <denis.evsyukov@flant.com>

Author: Evsyukov Denis

pkg/addon-operator/bootstrap.go

@@ -10,6 +10,7 @@ import (
 	"github.com/flant/addon-operator/pkg/kube_config_manager"
 	"github.com/flant/addon-operator/pkg/kube_config_manager/backend"
 	"github.com/flant/addon-operator/pkg/module_manager"
+	"github.com/flant/addon-operator/pkg/task/batch"
 	taskservice "github.com/flant/addon-operator/pkg/task/service"
 	shapp "github.com/flant/shell-operator/pkg/app"
 	"github.com/flant/shell-operator/pkg/debug"
@@ -46,6 +47,17 @@ func (op *AddonOperator) bootstrap() error {
 		return fmt.Errorf("assemble Debug server: %w", err)
 	}
 
+	// Initialize batching integration for module operations if enabled
+	var batchingIntegration *batch.BatchingIntegration
+	if app.BatchingEnabled {
+		batchingIntegration = batch.NewBatchingIntegration(
+			op.ModuleManager,
+			nil, // Task queue service will be set later if needed
+			op.parallelTaskChannels,
+			op.Logger,
+		)
+	}
+
 	cfg := &taskservice.TaskHandlerServiceConfig{
 		Engine:               op.engine,
 		ParallelTaskChannels: op.parallelTaskChannels,
@@ -56,9 +68,13 @@ func (op *AddonOperator) bootstrap() error {
 		KubeConfigManager:    op.KubeConfigManager,
 		ConvergeState:        op.ConvergeState,
 		CRDExtraLabels:       op.CRDExtraLabels,
+		BatchingIntegration:  batchingIntegration,
 	}
 
 	op.TaskService = taskservice.NewTaskHandlerService(op.ctx, cfg, op.Logger)
+	
+	// Store reference to batching integration
+	op.BatchingIntegration = batchingIntegration
 
 	return nil
 }

pkg/addon-operator/operator.go

@@ -25,6 +25,7 @@ import (
 	"github.com/flant/addon-operator/pkg/module_manager/models/hooks/kind"
 	"github.com/flant/addon-operator/pkg/module_manager/models/modules/events"
 	"github.com/flant/addon-operator/pkg/task"
+	"github.com/flant/addon-operator/pkg/task/batch"
 	paralleltask "github.com/flant/addon-operator/pkg/task/parallel"
 	taskservice "github.com/flant/addon-operator/pkg/task/service"
 	"github.com/flant/addon-operator/pkg/utils"
@@ -45,6 +46,9 @@ const (
 	LabelHeritage string = "heritage"
 )
 
+// BatchingIntegration is a type alias for the batch integration
+type BatchingIntegration = batch.BatchingIntegration
+
 // AddonOperator extends ShellOperator with modules and global hooks
 // and with a value storage.
 type AddonOperator struct {
@@ -94,6 +98,9 @@ type AddonOperator struct {
 	ConvergeState *converge.ConvergeState
 
 	TaskService *taskservice.TaskHandlerService
+
+	// BatchingIntegration provides batching capabilities for module operations
+	BatchingIntegration *BatchingIntegration
 }
 
 type Option func(operator *AddonOperator)
@@ -286,6 +293,11 @@ func (op *AddonOperator) Start(ctx context.Context) error {
 	op.ModuleManager.Start()
 	op.StartModuleManagerEventHandler()
 
+	// Start batching integration for module operations
+	if op.BatchingIntegration != nil {
+		op.BatchingIntegration.Start(ctx)
+	}
+
 	return nil
 }
 

pkg/app/app.go

@@ -56,6 +56,9 @@ var (
 	NumberOfParallelQueues   = 20
 	ParallelQueuePrefix      = "parallel_queue"
 	ParallelQueueNamePattern = ParallelQueuePrefix + "_%d"
+
+	// BatchingEnabled enables batching of module operations for improved performance
+	BatchingEnabled = false
 )
 
 const (
@@ -172,6 +175,11 @@ func DefineStartCommandFlags(kpApp *kingpin.Application, cmd *kingpin.CmdClause)
 		Default("").
 		StringVar(&ShellChrootDir)
 
+	cmd.Flag("batching-enabled", "Enable batching of module operations for improved performance.").
+		Envar("ADDON_OPERATOR_BATCHING_ENABLED").
+		Default("false").
+		BoolVar(&BatchingEnabled)
+
 	shapp.DefineKubeClientFlags(cmd)
 	shapp.DefineLoggingFlags(cmd)
 

pkg/task/service/service.go

@@ -48,6 +48,7 @@ type TaskHandlerServiceConfig struct {
 	KubeConfigManager    *kube_config_manager.KubeConfigManager
 	ConvergeState        *converge.ConvergeState
 	CRDExtraLabels       map[string]string
+	BatchingIntegration  interface{} // Interface to avoid circular import
 }
 
 type TaskHandlerService struct {
@@ -84,6 +85,8 @@ type TaskHandlerService struct {
 
 	taskFactory map[sh_task.TaskType]func(t sh_task.Task, logger *log.Logger) task.Task
 
+	batchingIntegration interface{} // BatchingIntegration for module operations
+
 	logger *log.Logger
 }
 
@@ -101,6 +104,7 @@ func NewTaskHandlerService(ctx context.Context, config *TaskHandlerServiceConfig
 		convergeState:        config.ConvergeState,
 		crdExtraLabels:       config.CRDExtraLabels,
 		discoveredCRDs:       discovercrds.NewDiscoveredGVKs(),
+		batchingIntegration:  config.BatchingIntegration,
 		logger:               logger,
 	}