Merge pull request #8380 from sseago/worker-count

Add --item-block-worker-count flag to velero install and server

changelogs/unreleased/8380-sseago

@@ -0,0 +1 @@
+Add --item-block-worker-count flag to velero install and server

pkg/cmd/cli/install/install.go

@@ -89,6 +89,7 @@ type Options struct {
 	BackupRepoConfigMap             string
 	RepoMaintenanceJobConfigMap     string
 	NodeAgentConfigMap              string
+	ItemBlockWorkerCount            int
 }
 
 // BindFlags adds command line values to the options struct.
@@ -182,6 +183,12 @@ func (o *Options) BindFlags(flags *pflag.FlagSet) {
 		o.NodeAgentConfigMap,
 		"The name of ConfigMap containing node-agent configurations.",
 	)
+	flags.IntVar(
+		&o.ItemBlockWorkerCount,
+		"item-block-worker-count",
+		o.ItemBlockWorkerCount,
+		"Number of worker threads to process ItemBlocks. Default is one. Optional.",
+	)
 }
 
 // NewInstallOptions instantiates a new, default InstallOptions struct.
@@ -283,6 +290,7 @@ func (o *Options) AsVeleroOptions() (*install.VeleroOptions, error) {
 		BackupRepoConfigMap:             o.BackupRepoConfigMap,
 		RepoMaintenanceJobConfigMap:     o.RepoMaintenanceJobConfigMap,
 		NodeAgentConfigMap:              o.NodeAgentConfigMap,
+		ItemBlockWorkerCount:            o.ItemBlockWorkerCount,
 	}, nil
 }
 

pkg/cmd/server/config/config.go

@@ -53,6 +53,8 @@ const (
 	DefaultMaintenanceJobCPULimit    = "0"
 	DefaultMaintenanceJobMemRequest  = "0"
 	DefaultMaintenanceJobMemLimit    = "0"
+
+	DefaultItemBlockWorkerCount = 1
 )
 
 var (
@@ -179,6 +181,7 @@ type Config struct {
 	RepoMaintenanceJobConfig       string
 	PodResources                   kube.PodResources
 	KeepLatestMaintenanceJobs      int
+	ItemBlockWorkerCount           int
 }
 
 func GetDefaultConfig() *Config {
@@ -216,6 +219,7 @@ func GetDefaultConfig() *Config {
 			MemoryLimit:   DefaultMaintenanceJobMemLimit,
 		},
 		KeepLatestMaintenanceJobs: DefaultKeepLatestMaintenanceJobs,
+		ItemBlockWorkerCount:      DefaultItemBlockWorkerCount,
 	}
 
 	return config
@@ -294,4 +298,10 @@ func (c *Config) BindFlags(flags *pflag.FlagSet) {
 		c.RepoMaintenanceJobConfig,
 		"The name of ConfigMap containing repository maintenance Job configurations.",
 	)
+	flags.IntVar(
+		&c.ItemBlockWorkerCount,
+		"item-block-worker-count",
+		c.ItemBlockWorkerCount,
+		"Number of worker threads to process ItemBlocks. Default is one. Optional.",
+	)
 }

pkg/cmd/server/server.go

@@ -620,6 +620,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			s.credentialFileStore,
 			s.config.MaxConcurrentK8SConnections,
 			s.config.DefaultSnapshotMoveData,
+			s.config.ItemBlockWorkerCount,
 			s.crClient,
 		).SetupWithManager(s.mgr); err != nil {
 			s.logger.Fatal(err, "unable to create controller", "controller", constant.ControllerBackup)

pkg/controller/backup_controller.go

@@ -86,6 +86,7 @@ type backupReconciler struct {
 	maxConcurrentK8SConnections int
 	defaultSnapshotMoveData     bool
 	globalCRClient              kbclient.Client
+	itemBlockWorkerCount        int
 }
 
 func NewBackupReconciler(
@@ -110,6 +111,7 @@ func NewBackupReconciler(
 	credentialStore credentials.FileStore,
 	maxConcurrentK8SConnections int,
 	defaultSnapshotMoveData bool,
+	itemBlockWorkerCount int,
 	globalCRClient kbclient.Client,
 ) *backupReconciler {
 	b := &backupReconciler{
@@ -135,6 +137,7 @@ func NewBackupReconciler(
 		credentialFileStore:         credentialStore,
 		maxConcurrentK8SConnections: maxConcurrentK8SConnections,
 		defaultSnapshotMoveData:     defaultSnapshotMoveData,
+		itemBlockWorkerCount:        itemBlockWorkerCount,
 		globalCRClient:              globalCRClient,
 	}
 	b.updateTotalBackupMetric()

pkg/install/deployment.go

@@ -57,6 +57,7 @@ type podTemplateConfig struct {
 	backupRepoConfigMap             string
 	repoMaintenanceJobConfigMap     string
 	nodeAgentConfigMap              string
+	itemBlockWorkerCount            int
 }
 
 func WithImage(image string) podTemplateOption {
@@ -212,6 +213,12 @@ func WithRepoMaintenanceJobConfigMap(repoMaintenanceJobConfigMap string) podTemp
 	}
 }
 
+func WithItemBlockWorkerCount(itemBlockWorkerCount int) podTemplateOption {
+	return func(c *podTemplateConfig) {
+		c.itemBlockWorkerCount = itemBlockWorkerCount
+	}
+}
+
 func Deployment(namespace string, opts ...podTemplateOption) *appsv1.Deployment {
 	// TODO: Add support for server args
 	c := &podTemplateConfig{
@@ -297,6 +304,10 @@ func Deployment(namespace string, opts ...podTemplateOption) *appsv1.Deployment
 		args = append(args, fmt.Sprintf("--repo-maintenance-job-configmap=%s", c.repoMaintenanceJobConfigMap))
 	}
 
+	if c.itemBlockWorkerCount > 0 {
+		args = append(args, fmt.Sprintf("--item-block-worker-count=%d", c.itemBlockWorkerCount))
+	}
+
 	deployment := &appsv1.Deployment{
 		ObjectMeta: objectMeta(namespace, "velero"),
 		TypeMeta: metav1.TypeMeta{

pkg/install/resources.go

@@ -267,6 +267,7 @@ type VeleroOptions struct {
 	BackupRepoConfigMap             string
 	RepoMaintenanceJobConfigMap     string
 	NodeAgentConfigMap              string
+	ItemBlockWorkerCount            int
 }
 
 func AllCRDs() *unstructured.UnstructuredList {
@@ -353,6 +354,7 @@ func AllResources(o *VeleroOptions) *unstructured.UnstructuredList {
 		WithScheduleSkipImmediately(o.ScheduleSkipImmediately),
 		WithPodResources(o.PodResources),
 		WithKeepLatestMaintenanceJobs(o.KeepLatestMaintenanceJobs),
+		WithItemBlockWorkerCount(o.ItemBlockWorkerCount),
 	}
 
 	if len(o.Features) > 0 {