Make ChannelCapacity configurable in poollet

poollet/bucketpoollet/cmd/bucketpoollet/app/app.go

@@ -61,6 +61,8 @@ type Options struct {
 	BucketRuntimeSocketDiscoveryTimeout time.Duration
 	BucketClassMapperSyncTimeout        time.Duration
 
+	ChannelCapacity int
+
 	WatchFilterValue string
 }
 
@@ -81,6 +83,8 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.DurationVar(&o.BucketRuntimeSocketDiscoveryTimeout, "bucket-runtime-discovery-timeout", 20*time.Second, "Timeout for discovering the bucket runtime socket.")
 	fs.DurationVar(&o.BucketClassMapperSyncTimeout, "bcm-sync-timeout", 10*time.Second, "Timeout waiting for the bucket class mapper to sync.")
 
+	fs.IntVar(&o.ChannelCapacity, "channel-capacity", 1024, "channel capacity for the bucket event generator")
+
 	fs.StringVar(&o.WatchFilterValue, "watch-filter", "", "Value to filter for while watching.")
 }
 
@@ -187,7 +191,9 @@ func Run(ctx context.Context, opts Options) error {
 			return nil, err
 		}
 		return res.Buckets, nil
-	}, irievent.GeneratorOptions{})
+	}, irievent.GeneratorOptions{
+		ChannelCapacity: opts.ChannelCapacity,
+	})
 	if err := mgr.Add(bucketEvents); err != nil {
 		return fmt.Errorf("error adding bucket event generator: %w", err)
 	}

poollet/machinepoollet/cmd/machinepoollet/app/app.go

@@ -71,6 +71,8 @@ type Options struct {
 	DialTimeout                          time.Duration
 	MachineClassMapperSyncTimeout        time.Duration
 
+	ChannelCapacity int
+
 	ServerFlags server.Flags
 
 	AddressesOptions addresses.GetOptions
@@ -97,6 +99,8 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.DurationVar(&o.DialTimeout, "dial-timeout", 1*time.Second, "Timeout for dialing to the machine runtime endpoint.")
 	fs.DurationVar(&o.MachineClassMapperSyncTimeout, "mcm-sync-timeout", 10*time.Second, "Timeout waiting for the machine class mapper to sync.")
 
+	fs.IntVar(&o.ChannelCapacity, "channel-capacity", 1024, "channel capacity for the machine event generator")
+
 	o.ServerFlags.BindFlags(fs)
 
 	o.AddressesOptions.BindFlags(fs)
@@ -261,7 +265,9 @@ func Run(ctx context.Context, opts Options) error {
 			return nil, err
 		}
 		return res.Machines, nil
-	}, irievent.GeneratorOptions{})
+	}, irievent.GeneratorOptions{
+		ChannelCapacity: opts.ChannelCapacity,
+	})
 	if err := mgr.Add(machineEvents); err != nil {
 		return fmt.Errorf("error adding machine event generator: %w", err)
 	}

poollet/volumepoollet/cmd/volumepoollet/app/app.go

@@ -62,6 +62,8 @@ type Options struct {
 	VolumeRuntimeSocketDiscoveryTimeout time.Duration
 	VolumeClassMapperSyncTimeout        time.Duration
 
+	ChannelCapacity int
+
 	WatchFilterValue string
 }
 
@@ -81,6 +83,9 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.DurationVar(&o.DialTimeout, "dial-timeout", 1*time.Second, "Timeout for dialing to the volume runtime endpoint.")
 	fs.DurationVar(&o.VolumeRuntimeSocketDiscoveryTimeout, "volume-runtime-discovery-timeout", 20*time.Second, "Timeout for discovering the volume runtime socket.")
 	fs.DurationVar(&o.VolumeClassMapperSyncTimeout, "vcm-sync-timeout", 10*time.Second, "Timeout waiting for the volume class mapper to sync.")
+
+	fs.IntVar(&o.ChannelCapacity, "channel-capacity", 1024, "channel capacity for the volume event generator")
+
 	fs.StringVar(&o.WatchFilterValue, "watch-filter", "", "Value to filter for while watching.")
 }
 
@@ -187,7 +192,9 @@ func Run(ctx context.Context, opts Options) error {
 			return nil, err
 		}
 		return res.Volumes, nil
-	}, irievent.GeneratorOptions{})
+	}, irievent.GeneratorOptions{
+		ChannelCapacity: opts.ChannelCapacity,
+	})
 	if err := mgr.Add(volumeEvents); err != nil {
 		return fmt.Errorf("error adding volume event generator: %w", err)
 	}