Make workerSize configurable

cmd/volumeprovider/app/app.go

@@ -58,13 +58,16 @@ type CephOptions struct {
 	KeyEncryptionKeyPath string
 
 	VolumeEventStoreOptions eventrecorder.EventStoreOptions
+
+	WorkerSize int
 }
 
 func (o *Options) Defaults() {
 	o.Ceph.ConnectTimeout = 10 * time.Second
 	o.Ceph.BurstFactor = 10
 	o.Ceph.BurstDurationInSeconds = 15
 	o.Ceph.PopulatorBufferSize = 5 * 1024 * 1024
+	o.Ceph.WorkerSize = 15
 }
 
 func (o *Options) AddFlags(fs *pflag.FlagSet) {
@@ -88,6 +91,8 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.IntVar(&o.Ceph.VolumeEventStoreOptions.MaxEvents, "volume-event-max-events", 100, "Maximum number of volume events that can be stored.")
 	fs.DurationVar(&o.Ceph.VolumeEventStoreOptions.TTL, "volume-event-ttl", 5*time.Minute, "Time to live for volume events.")
 	fs.DurationVar(&o.Ceph.VolumeEventStoreOptions.ResyncInterval, "volume-event-resync-interval", 1*time.Minute, "Interval for resynchronizing the volume events.")
+
+	fs.IntVar(&o.Ceph.WorkerSize, "worker-size", o.Ceph.WorkerSize, "Defines the factor to calculate the burst limits.")
 }
 
 func (o *Options) MarkFlagsRequired(cmd *cobra.Command) {
@@ -162,6 +167,13 @@ func configureCephAuth(opts *CephOptions) (func() error, error) {
 func Run(ctx context.Context, opts Options) error {
 	log := ctrl.LoggerFrom(ctx)
 	setupLog := log.WithName("setup")
+
+	if opts.Ceph.WorkerSize <= 1 {
+		err := fmt.Errorf("invalid configuration: worker-size must be greater than 1, but got %d", opts.Ceph.WorkerSize)
+		setupLog.Error(err, "Worker size validation failed")
+		return err
+	}
+
 	var wg sync.WaitGroup
 
 	cleanup, err := configureCephAuth(&opts.Ceph)
@@ -252,9 +264,10 @@ func Run(ctx context.Context, opts Options) error {
 		snapshotEvents,
 		encryptor,
 		controllers.ImageReconcilerOptions{
-			Monitors: opts.Ceph.Monitors,
-			Client:   opts.Ceph.Client,
-			Pool:     opts.Ceph.Pool,
+			Monitors:   opts.Ceph.Monitors,
+			Client:     opts.Ceph.Client,
+			Pool:       opts.Ceph.Pool,
+			WorkerSize: opts.Ceph.WorkerSize,
 		},
 	)
 	if err != nil {
@@ -280,6 +293,7 @@ func Run(ctx context.Context, opts Options) error {
 		controllers.SnapshotReconcilerOptions{
 			Pool:                opts.Ceph.Pool,
 			PopulatorBufferSize: opts.Ceph.PopulatorBufferSize,
+			WorkerSize:          opts.Ceph.WorkerSize,
 		},
 	)
 	if err != nil {

internal/controllers/image_controller.go

@@ -38,9 +38,10 @@ const (
 )
 
 type ImageReconcilerOptions struct {
-	Monitors string
-	Client   string
-	Pool     string
+	Monitors   string
+	Client     string
+	Pool       string
+	WorkerSize int
 }
 
 func NewImageReconciler(
@@ -95,6 +96,10 @@ func NewImageReconciler(
 		return nil, fmt.Errorf("must specify ceph client")
 	}
 
+	if opts.WorkerSize == 0 {
+		opts.WorkerSize = 15
+	}
+
 	return &ImageReconciler{
 		log:            log,
 		conn:           conn,
@@ -109,6 +114,7 @@ func NewImageReconciler(
 		client:         opts.Client,
 		pool:           opts.Pool,
 		keyEncryption:  keyEncryption,
+		workerSize:     opts.WorkerSize,
 	}, nil
 }
 
@@ -131,14 +137,13 @@ type ImageReconciler struct {
 	pool     string
 
 	keyEncryption encryption.Encryptor
+
+	workerSize int
 }
 
 func (r *ImageReconciler) Start(ctx context.Context) error {
 	log := r.log
 
-	//todo make configurable
-	workerSize := 15
-
 	imgEventReg, err := r.imageEvents.AddHandler(event.HandlerFunc[*providerapi.Image](func(evt event.Event[*providerapi.Image]) {
 		r.queue.Add(evt.Object.ID)
 	}))
@@ -180,7 +185,7 @@ func (r *ImageReconciler) Start(ctx context.Context) error {
 	}()
 
 	var wg sync.WaitGroup
-	for i := 0; i < workerSize; i++ {
+	for i := 0; i < r.workerSize; i++ {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()

internal/controllers/snapshot_controller.go

@@ -29,6 +29,7 @@ import (
 type SnapshotReconcilerOptions struct {
 	Pool                string
 	PopulatorBufferSize int64
+	WorkerSize          int
 }
 
 func NewSnapshotReconciler(
@@ -68,6 +69,10 @@ func NewSnapshotReconciler(
 		opts.PopulatorBufferSize = 5 * 1024 * 1024
 	}
 
+	if opts.WorkerSize == 0 {
+		opts.WorkerSize = 15
+	}
+
 	return &SnapshotReconciler{
 		log:                 log,
 		conn:                conn,
@@ -78,6 +83,7 @@ func NewSnapshotReconciler(
 		events:              events,
 		pool:                opts.Pool,
 		populatorBufferSize: opts.PopulatorBufferSize,
+		workerSize:          opts.WorkerSize,
 	}, nil
 }
 
@@ -94,14 +100,13 @@ type SnapshotReconciler struct {
 
 	pool                string
 	populatorBufferSize int64
+
+	workerSize int
 }
 
 func (r *SnapshotReconciler) Start(ctx context.Context) error {
 	log := r.log
 
-	//todo make configurable
-	workerSize := 15
-
 	reg, err := r.events.AddHandler(event.HandlerFunc[*providerapi.Snapshot](func(event event.Event[*providerapi.Snapshot]) {
 		r.queue.Add(event.Object.ID)
 	}))
@@ -118,7 +123,7 @@ func (r *SnapshotReconciler) Start(ctx context.Context) error {
 	}()
 
 	var wg sync.WaitGroup
-	for i := 0; i < workerSize; i++ {
+	for i := 0; i < r.workerSize; i++ {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()

tests/integration/integration_suite_test.go

@@ -100,6 +100,7 @@ var _ = BeforeSuite(func() {
 				TTL:            eventTTL,
 				ResyncInterval: resyncInterval,
 			},
+			WorkerSize: 15,
 		},
 	}