add parallel-blocks-upload for block data mover

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

config/crd/v1/bases/velero.io_backups.yaml

@@ -536,9 +536,13 @@ spec:
                 description: UploaderConfig specifies the configuration for the uploader.
                 nullable: true
                 properties:
+                  parallelBlocksUpload:
+                    description: ParallelBlocksUpload is the number of blocks parallel
+                      uploads to perform when using Velero block uploader.
+                    type: integer
                   parallelFilesUpload:
                     description: ParallelFilesUpload is the number of files parallel
-                      uploads to perform when using the uploader.
+                      uploads to perform when using the Velero fs uploader.
                     type: integer
                 type: object
               volumeGroupSnapshotLabelKey:

config/crd/v1/bases/velero.io_schedules.yaml

@@ -558,9 +558,13 @@ spec:
                       uploader.
                     nullable: true
                     properties:
+                      parallelBlocksUpload:
+                        description: ParallelBlocksUpload is the number of blocks parallel
+                          uploads to perform when using Velero block uploader.
+                        type: integer
                       parallelFilesUpload:
                         description: ParallelFilesUpload is the number of files parallel
-                          uploads to perform when using the uploader.
+                          uploads to perform when using the Velero fs uploader.
                         type: integer
                     type: object
                   volumeGroupSnapshotLabelKey:

pkg/apis/velero/v1/backup_types.go

@@ -195,9 +195,13 @@ type BackupSpec struct {
 
 // UploaderConfigForBackup defines the configuration for the uploader when doing backup.
 type UploaderConfigForBackup struct {
-	// ParallelFilesUpload is the number of files parallel uploads to perform when using the uploader.
+	// ParallelFilesUpload is the number of files parallel uploads to perform when using the Velero fs uploader.
 	// +optional
 	ParallelFilesUpload int `json:"parallelFilesUpload,omitempty"`
+
+	// ParallelBlocksUpload is the number of blocks parallel uploads to perform when using Velero block uploader.
+	// +optional
+	ParallelBlocksUpload int `json:"parallelBlocksUpload,omitempty"`
 }
 
 // BackupHooks contains custom behaviors that should be executed at different phases of the backup.

pkg/backup/actions/csi/pvc_action.go

@@ -615,10 +615,16 @@ func newDataUpload(
 		dataUpload.Spec.CSISnapshot.SnapshotClass = *vs.Spec.VolumeSnapshotClassName
 	}
 
-	if backup.Spec.UploaderConfig != nil &&
-		backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
+	if backup.Spec.UploaderConfig != nil {
 		dataUpload.Spec.DataMoverConfig = make(map[string]string)
-		dataUpload.Spec.DataMoverConfig[uploaderUtil.ParallelFilesUpload] = strconv.Itoa(backup.Spec.UploaderConfig.ParallelFilesUpload)
+
+		if backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
+			dataUpload.Spec.DataMoverConfig[uploaderUtil.ParallelFilesUpload] = strconv.Itoa(backup.Spec.UploaderConfig.ParallelFilesUpload)
+		}
+
+		if backup.Spec.UploaderConfig.ParallelBlocksUpload > 0 {
+			dataUpload.Spec.DataMoverConfig[uploaderUtil.ParallelBlocksUpload] = strconv.Itoa(backup.Spec.UploaderConfig.ParallelBlocksUpload)
+		}
 	}
 
 	return dataUpload

pkg/builder/backup_builder.go

@@ -327,7 +327,7 @@ func (b *BackupBuilder) DataMover(name string) *BackupBuilder {
 	return b
 }
 
-// ParallelFilesUpload sets the Backup's uploader parallel uploads
+// ParallelFilesUpload sets the Backup's uploader parallel files upload
 func (b *BackupBuilder) ParallelFilesUpload(parallel int) *BackupBuilder {
 	if b.object.Spec.UploaderConfig == nil {
 		b.object.Spec.UploaderConfig = &velerov1api.UploaderConfigForBackup{}
@@ -336,6 +336,15 @@ func (b *BackupBuilder) ParallelFilesUpload(parallel int) *BackupBuilder {
 	return b
 }
 
+// ParallelBlocksUpload sets the Backup's uploader parallel blocks upload
+func (b *BackupBuilder) ParallelBlocksUpload(parallel int) *BackupBuilder {
+	if b.object.Spec.UploaderConfig == nil {
+		b.object.Spec.UploaderConfig = &velerov1api.UploaderConfigForBackup{}
+	}
+	b.object.Spec.UploaderConfig.ParallelBlocksUpload = parallel
+	return b
+}
+
 func (b *BackupBuilder) BackupType(backupType velerov1api.BackupType) *BackupBuilder {
 	b.object.Spec.BackupType = backupType
 	return b

pkg/cmd/cli/backup/create.go

@@ -128,6 +128,7 @@ type CreateOptions struct {
 	ResPoliciesConfigmap            string
 	client                          kbclient.WithWatch
 	ParallelFilesUpload             int
+	ParallelBlocksUpload            int
 	BackupType                      string
 }
 
@@ -176,7 +177,8 @@ func (o *CreateOptions) BindFlags(flags *pflag.FlagSet) {
 
 	flags.StringVar(&o.ResPoliciesConfigmap, "resource-policies-configmap", "", "Reference to the resource policies configmap that backup should use")
 	flags.StringVar(&o.DataMover, "data-mover", "", "Specify the data mover to be used by the backup. If the parameter is not set or set as 'velero', the built-in data mover will be used")
-	flags.IntVar(&o.ParallelFilesUpload, "parallel-files-upload", 0, "Number of files uploads simultaneously when running a backup. This is only applicable for the kopia uploader")
+	flags.IntVar(&o.ParallelFilesUpload, "parallel-files-upload", 0, "Number of files uploads simultaneously when running a backup. This is only applicable for Velero fs uploader")
+	flags.IntVar(&o.ParallelBlocksUpload, "parallel-blocks-upload", 0, "Number of blocks simultaneously when running a backup. This is only applicable for Velero block uploader")
 	flags.StringVar(&o.BackupType, "backup-type", "", "Specify how volume data is backed up, with possible values including Full and Incremental.")
 }
 
@@ -492,6 +494,9 @@ func (o *CreateOptions) BuildBackup(namespace string) (*velerov1api.Backup, erro
 		if o.ParallelFilesUpload > 0 {
 			backupBuilder.ParallelFilesUpload(o.ParallelFilesUpload)
 		}
+		if o.ParallelBlocksUpload > 0 {
+			backupBuilder.ParallelBlocksUpload(o.ParallelBlocksUpload)
+		}
 	}
 
 	backup := backupBuilder.ObjectMeta(builder.WithLabelsMap(o.Labels.Data()), builder.WithAnnotationsMap(o.Annotations.Data())).Result()

pkg/cmd/cli/schedule/create.go

@@ -177,9 +177,10 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
 		schedule.Spec.Template.ResourcePolicy = &corev1api.TypedLocalObjectReference{Kind: resourcepolicies.ConfigmapRefType, Name: o.BackupOptions.ResPoliciesConfigmap}
 	}
 
-	if o.BackupOptions.ParallelFilesUpload > 0 {
+	if o.BackupOptions.ParallelFilesUpload > 0 || o.BackupOptions.ParallelBlocksUpload > 0 {
 		schedule.Spec.Template.UploaderConfig = &api.UploaderConfigForBackup{
-			ParallelFilesUpload: o.BackupOptions.ParallelFilesUpload,
+			ParallelFilesUpload:  o.BackupOptions.ParallelFilesUpload,
+			ParallelBlocksUpload: o.BackupOptions.ParallelBlocksUpload,
 		}
 	}
 

pkg/cmd/util/output/backup_describer.go

@@ -96,7 +96,7 @@ func DescribeBackup(
 
 		DescribeGlobalVolumePolicy(d, backup)
 
-		if backup.Spec.UploaderConfig != nil && backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
+		if backup.Spec.UploaderConfig != nil {
 			d.Println()
 			DescribeUploaderConfigForBackup(d, backup.Spec)
 		}
@@ -150,6 +150,7 @@ func DescribeGlobalVolumePolicy(d *Describer, backup *velerov1api.Backup) {
 func DescribeUploaderConfigForBackup(d *Describer, spec velerov1api.BackupSpec) {
 	d.Printf("Uploader config:\n")
 	d.Printf("\tParallel files upload:\t%d\n", spec.UploaderConfig.ParallelFilesUpload)
+	d.Printf("\tParallel blocks upload:\t%d\n", spec.UploaderConfig.ParallelBlocksUpload)
 }
 
 // DescribeBackupSpec describes a backup spec in human-readable format.

pkg/cmd/util/output/schedule_describer.go

@@ -48,7 +48,7 @@ func DescribeSchedule(schedule *v1.Schedule) string {
 			DescribeResourcePolicies(d, schedule.Spec.Template.ResourcePolicy)
 		}
 
-		if schedule.Spec.Template.UploaderConfig != nil && schedule.Spec.Template.UploaderConfig.ParallelFilesUpload > 0 {
+		if schedule.Spec.Template.UploaderConfig != nil {
 			d.Println()
 			DescribeUploaderConfigForBackup(d, schedule.Spec.Template)
 		}

pkg/uploader/block/uploader.go

@@ -34,6 +34,7 @@ import (
 	"github.com/velero-io/velero/pkg/repository/udmrepo"
 	"github.com/velero-io/velero/pkg/uploader"
 	cbt "github.com/velero-io/velero/pkg/uploader/cbt/types"
+	uploaderutil "github.com/velero-io/velero/pkg/uploader/util"
 	"github.com/velero-io/velero/pkg/util/freelist"
 )
 
@@ -97,13 +98,22 @@ func (blkup *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, b
 		backupMode = udmrepo.ObjectDataBackupModeFull
 	}
 
+	asyncWrite := runtime.GOMAXPROCS(0)
+	if len(configs) > 0 {
+		if parallelUpload, err := uploaderutil.GetParallelBlocksUpload(configs); err != nil {
+			return udmrepo.Snapshot{}, 0, errors.Wrap(err, "failed to get uploader config")
+		} else if parallelUpload > 0 {
+			asyncWrite = parallelUpload
+		}
+	}
+
 	destObj, err := blkup.repoWriter.NewObjectWriter(blkup.ctx, udmrepo.ObjectWriteOptions{
 		Description:  fmt.Sprintf("BDEV:%s-%s", getObjectName(source.realSource), snapStart.Format("2006-01-02-15-04-05")),
 		DataType:     udmrepo.ObjectDataTypeData,
 		AccessMode:   udmrepo.ObjectDataAccessModeBlock,
 		ParentObject: parentObject,
 		BackupMode:   backupMode,
-		AsyncWrites:  runtime.NumCPU(),
+		AsyncWrites:  asyncWrite,
 	})
 	if err != nil {
 		return udmrepo.Snapshot{}, 0, errors.Wrap(err, "error creating object writer")

pkg/uploader/util/uploader_config.go

@@ -25,15 +25,17 @@ import (
 )
 
 const (
-	ParallelFilesUpload = "ParallelFilesUpload"
-	WriteSparseFiles    = "WriteSparseFiles"
-	RestoreConcurrency  = "ParallelFilesDownload"
-	DeleteExtraFiles    = "DeleteExtraFiles"
+	ParallelFilesUpload  = "ParallelFilesUpload"
+	ParallelBlocksUpload = "ParallelBlocksUpload"
+	WriteSparseFiles     = "WriteSparseFiles"
+	RestoreConcurrency   = "ParallelFilesDownload"
+	DeleteExtraFiles     = "DeleteExtraFiles"
 )
 
 func StoreBackupConfig(config *velerov1api.UploaderConfigForBackup) map[string]string {
 	data := make(map[string]string)
 	data[ParallelFilesUpload] = strconv.Itoa(config.ParallelFilesUpload)
+	data[ParallelBlocksUpload] = strconv.Itoa(config.ParallelBlocksUpload)
 	return data
 }
 
@@ -70,6 +72,18 @@ func GetParallelFilesUpload(uploaderCfg map[string]string) (int, error) {
 	return 0, nil
 }
 
+func GetParallelBlocksUpload(uploaderCfg map[string]string) (int, error) {
+	parallelBlocksUpload, ok := uploaderCfg[ParallelBlocksUpload]
+	if ok {
+		parallelBlocksUploadInt, err := strconv.Atoi(parallelBlocksUpload)
+		if err != nil {
+			return 0, errors.Wrap(err, "failed to parse parallelBlocksUpload config")
+		}
+		return parallelBlocksUploadInt, nil
+	}
+	return 0, nil
+}
+
 func GetWriteSparseFiles(uploaderCfg map[string]string) (bool, error) {
 	writeSparseFiles, ok := uploaderCfg[WriteSparseFiles]
 	if ok {