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()