✨ feature: expose flags to configure spoke agent controller workers (#1560)

* feat: expose flags to configure spoke agent controller workers

Signed-off-by: Vaishnav88sk <vaishnavsk8804@gmail.com>

* fix: address PR review feedback on worker validation and formatting

Signed-off-by: Vaishnav88sk <vaishnavsk8804@gmail.com>

* refactor: consolidate worker count flags into a single flag

Signed-off-by: Vaishnav88sk <vaishnavsk8804@gmail.com>

---------

Signed-off-by: Vaishnav88sk <vaishnavsk8804@gmail.com>

pkg/work/spoke/options.go

@@ -1,6 +1,7 @@
 package spoke
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/spf13/pflag"
@@ -23,6 +24,8 @@ type WorkloadAgentOptions struct {
 	CloudEventsClientCodecs                []string
 	DefaultUserAgent                       string
 
+	WorkloadAgentWorkers int
+
 	ObjectReaderOption *objectreader.Options
 }
 
@@ -36,6 +39,7 @@ func NewWorkloadAgentOptions() *WorkloadAgentOptions {
 		WorkloadSourceConfig:                   "/spoke/hub-kubeconfig/kubeconfig",
 		DefaultUserAgent:                       defaultUserAgent,
 		ObjectReaderOption:                     objectreader.NewOptions(),
+		WorkloadAgentWorkers:                   10,
 	}
 }
 
@@ -56,5 +60,16 @@ func (o *WorkloadAgentOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.StringSliceVar(&o.CloudEventsClientCodecs, "cloudevents-client-codecs", o.CloudEventsClientCodecs,
 		"The codecs for cloudevents client when workload source source is based on cloudevents, the valid codecs: manifest or manifestbundle")
 
+	fs.IntVar(&o.WorkloadAgentWorkers, "workload-agent-workers",
+		o.WorkloadAgentWorkers, "The number of workers for the workload agent controllers")
+
 	o.ObjectReaderOption.AddFlags(fs)
 }
+
+// Validate checks if the options are valid
+func (o *WorkloadAgentOptions) Validate() error {
+	if o.WorkloadAgentWorkers < 1 {
+		return fmt.Errorf("workload-agent-workers must be >= 1, got %d", o.WorkloadAgentWorkers)
+	}
+	return nil
+}

pkg/work/spoke/spokeagent.go

@@ -35,19 +35,6 @@ import (
 	"open-cluster-management.io/ocm/pkg/work/spoke/controllers/statuscontroller"
 )
 
-const (
-	// If a controller queue size is too large (>500), the processing speed of the controller will drop significantly
-	// with one worker, increasing the work numbers can improve the processing speed.
-	// We compared the two situations where the worker is set to 1 and 10, when the worker is 10, the resource
-	// utilization of the kubeapi-server and work agent do not increase significantly.
-	//
-	// TODO expose a flag to set the worker for each controller
-	appliedManifestWorkFinalizeControllerWorkers = 10
-	manifestWorkFinalizeControllerWorkers        = 10
-	availableStatusControllerWorkers             = 10
-	manifestWorkAgentWorkers                     = 10
-)
-
 type WorkAgentConfig struct {
 	agentOptions *options.AgentOptions
 	workOptions  *WorkloadAgentOptions
@@ -63,6 +50,10 @@ func NewWorkAgentConfig(commonOpts *options.AgentOptions, opts *WorkloadAgentOpt
 
 // RunWorkloadAgent starts the controllers on agent to process work from hub.
 func (o *WorkAgentConfig) RunWorkloadAgent(ctx context.Context, controllerContext *controllercmd.ControllerContext) error {
+	if err := o.workOptions.Validate(); err != nil {
+		return err
+	}
+
 	// setting up contextual logger
 	logger := klog.NewKlogr()
 	podName := os.Getenv("POD_NAME")
@@ -199,11 +190,11 @@ func (o *WorkAgentConfig) RunWorkloadAgent(ctx context.Context, controllerContex
 	go hubWorkInformer.Informer().Run(ctx.Done())
 
 	go addFinalizerController.Run(ctx, 1)
-	go appliedManifestWorkFinalizeController.Run(ctx, appliedManifestWorkFinalizeControllerWorkers)
+	go appliedManifestWorkFinalizeController.Run(ctx, o.workOptions.WorkloadAgentWorkers)
 	go unmanagedAppliedManifestWorkController.Run(ctx, 1)
-	go manifestWorkController.Run(ctx, manifestWorkAgentWorkers)
-	go manifestWorkFinalizeController.Run(ctx, manifestWorkFinalizeControllerWorkers)
-	go availableStatusController.Run(ctx, availableStatusControllerWorkers)
+	go manifestWorkController.Run(ctx, o.workOptions.WorkloadAgentWorkers)
+	go manifestWorkFinalizeController.Run(ctx, o.workOptions.WorkloadAgentWorkers)
+	go availableStatusController.Run(ctx, o.workOptions.WorkloadAgentWorkers)
 
 	<-ctx.Done()