From 0ec4f5da1a17f22e5c08ef76366f0173d04d2f6d Mon Sep 17 00:00:00 2001 From: Vaishnav Kale <116202759+Vaishnav88sk@users.noreply.github.com> Date: Fri, 3 Jul 2026 08:37:21 +0530 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feature:=20expose=20flags=20to=20co?= =?UTF-8?q?nfigure=20spoke=20agent=20controller=20workers=20(#1560)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: expose flags to configure spoke agent controller workers Signed-off-by: Vaishnav88sk * fix: address PR review feedback on worker validation and formatting Signed-off-by: Vaishnav88sk * refactor: consolidate worker count flags into a single flag Signed-off-by: Vaishnav88sk --------- Signed-off-by: Vaishnav88sk --- pkg/work/spoke/options.go | 15 +++++++++++++++ pkg/work/spoke/spokeagent.go | 25 ++++++++----------------- 2 files changed, 23 insertions(+), 17 deletions(-) diff --git a/pkg/work/spoke/options.go b/pkg/work/spoke/options.go index e2e7c7a77..56d55c3d4 100644 --- a/pkg/work/spoke/options.go +++ b/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 +} diff --git a/pkg/work/spoke/spokeagent.go b/pkg/work/spoke/spokeagent.go index ae0ca0da7..4c86802a4 100644 --- a/pkg/work/spoke/spokeagent.go +++ b/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()