Merge pull request #7141 from qiuming-best/support-restore-sparse

Allow sparse option for Kopia & Restic restore

changelogs/unreleased/7141-qiuming-best

@@ -0,0 +1 @@
+Allow sparse option for Kopia & Restic restore 

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

@@ -121,13 +121,12 @@ spec:
                 description: Tags are a map of key-value pairs that should be applied
                   to the volume backup as tags.
                 type: object
-              uploaderConfig:
-                description: UploaderConfig specifies the configuration for the uploader.
-                properties:
-                  parallelFilesUpload:
-                    description: ParallelFilesUpload is the number of files parallel
-                      uploads to perform when using the uploader.
-                    type: integer
+              uploaderSettings:
+                additionalProperties:
+                  type: string
+                description: UploaderSettings are a map of key-value pairs that should
+                  be applied to the uploader configuration.
+                nullable: true
                 type: object
               uploaderType:
                 description: UploaderType is the type of the uploader to handle the

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

@@ -119,6 +119,13 @@ spec:
                 description: SourceNamespace is the original namespace for namaspace
                   mapping.
                 type: string
+              uploaderSettings:
+                additionalProperties:
+                  type: string
+                description: UploaderSettings are a map of key-value pairs that should
+                  be applied to the uploader configuration.
+                nullable: true
+                type: object
               uploaderType:
                 description: UploaderType is the type of the uploader to handle the
                   data transfer.

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

@@ -418,6 +418,16 @@ spec:
                   restore from the most recent successful backup created from this
                   schedule.
                 type: string
+              uploaderConfig:
+                description: UploaderConfig specifies the configuration for the restore.
+                nullable: true
+                properties:
+                  writeSparseFiles:
+                    description: WriteSparseFiles is a flag to indicate whether write
+                      files sparsely or not.
+                    nullable: true
+                    type: boolean
+                type: object
             required:
             - backupName
             type: object

config/crd/v2alpha1/bases/velero.io_datauploads.yaml

@@ -125,14 +125,6 @@ spec:
                 description: SourcePVC is the name of the PVC which the snapshot is
                   taken for.
                 type: string
-              uploaderConfig:
-                description: UploaderConfig specifies the configuration for the uploader.
-                properties:
-                  parallelFilesUpload:
-                    description: ParallelFilesUpload is the number of files parallel
-                      uploads to perform when using the uploader.
-                    type: integer
-                type: object
             required:
             - backupStorageLocation
             - operationTimeout

pkg/apis/velero/shared/uploader_config.go

@@ -1,23 +0,0 @@
-/*
-Copyright The Velero Contributors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package shared
-
-// UploaderConfig defines the configuration for the uploader.
-type UploaderConfig struct {
-	// ParallelFilesUpload is the number of files parallel uploads to perform when using the uploader.
-	ParallelFilesUpload int `json:"parallelFilesUpload,omitempty"`
-}

pkg/apis/velero/v1/backup_types.go

@@ -19,8 +19,6 @@ package v1
 import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 )
 
 type Metadata struct {
@@ -181,7 +179,14 @@ type BackupSpec struct {
 	// UploaderConfig specifies the configuration for the uploader.
 	// +optional
 	// +nullable
-	UploaderConfig shared.UploaderConfig `json:"uploaderConfig,omitempty"`
+	UploaderConfig *UploaderConfigForBackup `json:"uploaderConfig,omitempty"`
+}
+
+// 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.
+	// +optional
+	ParallelFilesUpload int `json:"parallelFilesUpload,omitempty"`
 }
 
 // BackupHooks contains custom behaviors that should be executed at different phases of the backup.

pkg/apis/velero/v1/pod_volume_backup_types.go

@@ -52,8 +52,11 @@ type PodVolumeBackupSpec struct {
 	// +optional
 	Tags map[string]string `json:"tags,omitempty"`
 
-	// UploaderConfig specifies the configuration for the uploader.
-	UploaderConfig shared.UploaderConfig `json:"uploaderConfig,omitempty"`
+	// UploaderSettings are a map of key-value pairs that should be applied to the
+	// uploader configuration.
+	// +optional
+	// +nullable
+	UploaderSettings map[string]string `json:"uploaderSettings,omitempty"`
 }
 
 // PodVolumeBackupPhase represents the lifecycle phase of a PodVolumeBackup.

pkg/apis/velero/v1/pod_volume_restore_type.go

@@ -48,6 +48,12 @@ type PodVolumeRestoreSpec struct {
 
 	// SourceNamespace is the original namespace for namaspace mapping.
 	SourceNamespace string `json:"sourceNamespace"`
+
+	// UploaderSettings are a map of key-value pairs that should be applied to the
+	// uploader configuration.
+	// +optional
+	// +nullable
+	UploaderSettings map[string]string `json:"uploaderSettings,omitempty"`
 }
 
 // PodVolumeRestorePhase represents the lifecycle phase of a PodVolumeRestore.

pkg/apis/velero/v1/restore_types.go

@@ -123,6 +123,19 @@ type RestoreSpec struct {
 	// +optional
 	// +nullable
 	ResourceModifier *v1.TypedLocalObjectReference `json:"resourceModifier,omitempty"`
+
+	// UploaderConfig specifies the configuration for the restore.
+	// +optional
+	// +nullable
+	UploaderConfig *UploaderConfigForRestore `json:"uploaderConfig,omitempty"`
+}
+
+// UploaderConfigForRestore defines the configuration for the restore.
+type UploaderConfigForRestore struct {
+	// WriteSparseFiles is a flag to indicate whether write files sparsely or not.
+	// +optional
+	// +nullable
+	WriteSparseFiles *bool `json:"writeSparseFiles,omitempty"`
 }
 
 // RestoreHooks contains custom behaviors that should be executed during or post restore.

pkg/apis/velero/v1/zz_generated.deepcopy.go

@@ -381,7 +381,11 @@ func (in *BackupSpec) DeepCopyInto(out *BackupSpec) {
 		*out = new(bool)
 		**out = **in
 	}
-	out.UploaderConfig = in.UploaderConfig
+	if in.UploaderConfig != nil {
+		in, out := &in.UploaderConfig, &out.UploaderConfig
+		*out = new(UploaderConfigForBackup)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupSpec.
@@ -972,7 +976,13 @@ func (in *PodVolumeBackupSpec) DeepCopyInto(out *PodVolumeBackupSpec) {
 			(*out)[key] = val
 		}
 	}
-	out.UploaderConfig = in.UploaderConfig
+	if in.UploaderSettings != nil {
+		in, out := &in.UploaderSettings, &out.UploaderSettings
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PodVolumeBackupSpec.
@@ -1014,7 +1024,7 @@ func (in *PodVolumeRestore) DeepCopyInto(out *PodVolumeRestore) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
-	out.Spec = in.Spec
+	in.Spec.DeepCopyInto(&out.Spec)
 	in.Status.DeepCopyInto(&out.Status)
 }
 
@@ -1072,6 +1082,13 @@ func (in *PodVolumeRestoreList) DeepCopyObject() runtime.Object {
 func (in *PodVolumeRestoreSpec) DeepCopyInto(out *PodVolumeRestoreSpec) {
 	*out = *in
 	out.Pod = in.Pod
+	if in.UploaderSettings != nil {
+		in, out := &in.UploaderSettings, &out.UploaderSettings
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PodVolumeRestoreSpec.
@@ -1349,6 +1366,11 @@ func (in *RestoreSpec) DeepCopyInto(out *RestoreSpec) {
 		*out = new(corev1.TypedLocalObjectReference)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.UploaderConfig != nil {
+		in, out := &in.UploaderConfig, &out.UploaderConfig
+		*out = new(UploaderConfigForRestore)
+		(*in).DeepCopyInto(*out)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RestoreSpec.
@@ -1646,6 +1668,41 @@ func (in *StorageType) DeepCopy() *StorageType {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *UploaderConfigForBackup) DeepCopyInto(out *UploaderConfigForBackup) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new UploaderConfigForBackup.
+func (in *UploaderConfigForBackup) DeepCopy() *UploaderConfigForBackup {
+	if in == nil {
+		return nil
+	}
+	out := new(UploaderConfigForBackup)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *UploaderConfigForRestore) DeepCopyInto(out *UploaderConfigForRestore) {
+	*out = *in
+	if in.WriteSparseFiles != nil {
+		in, out := &in.WriteSparseFiles, &out.WriteSparseFiles
+		*out = new(bool)
+		**out = **in
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new UploaderConfigForRestore.
+func (in *UploaderConfigForRestore) DeepCopy() *UploaderConfigForRestore {
+	if in == nil {
+		return nil
+	}
+	out := new(UploaderConfigForRestore)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *VolumeSnapshotLocation) DeepCopyInto(out *VolumeSnapshotLocation) {
 	*out = *in

pkg/apis/velero/v2alpha1/data_upload_types.go

@@ -51,7 +51,7 @@ type DataUploadSpec struct {
 	// DataMoverConfig is for data-mover-specific configuration fields.
 	// +optional
 	// +nullable
-	DataMoverConfig *map[string]string `json:"dataMoverConfig,omitempty"`
+	DataMoverConfig map[string]string `json:"dataMoverConfig,omitempty"`
 
 	// Cancel indicates request to cancel the ongoing DataUpload. It can be set
 	// when the DataUpload is in InProgress phase
@@ -60,9 +60,6 @@ type DataUploadSpec struct {
 	// OperationTimeout specifies the time used to wait internal operations,
 	// before returning error as timeout.
 	OperationTimeout metav1.Duration `json:"operationTimeout"`
-
-	// UploaderConfig specifies the configuration for the uploader.
-	UploaderConfig shared.UploaderConfig `json:"uploaderConfig,omitempty"`
 }
 
 type SnapshotType string

pkg/apis/velero/v2alpha1/zz_generated.deepcopy.go

@@ -226,17 +226,12 @@ func (in *DataUploadSpec) DeepCopyInto(out *DataUploadSpec) {
 	}
 	if in.DataMoverConfig != nil {
 		in, out := &in.DataMoverConfig, &out.DataMoverConfig
-		*out = new(map[string]string)
-		if **in != nil {
-			in, out := *in, *out
-			*out = make(map[string]string, len(*in))
-			for key, val := range *in {
-				(*out)[key] = val
-			}
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
 		}
 	}
 	out.OperationTimeout = in.OperationTimeout
-	out.UploaderConfig = in.UploaderConfig
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DataUploadSpec.

pkg/builder/backup_builder.go

@@ -302,6 +302,9 @@ func (b *BackupBuilder) DataMover(name string) *BackupBuilder {
 
 // ParallelFilesUpload sets the Backup's uploader parallel uploads
 func (b *BackupBuilder) ParallelFilesUpload(parallel int) *BackupBuilder {
+	if b.object.Spec.UploaderConfig == nil {
+		b.object.Spec.UploaderConfig = &velerov1api.UploaderConfigForBackup{}
+	}
 	b.object.Spec.UploaderConfig.ParallelFilesUpload = parallel
 	return b
 }

pkg/builder/data_upload_builder.go

@@ -103,7 +103,7 @@ func (d *DataUploadBuilder) OperationTimeout(timeout metav1.Duration) *DataUploa
 }
 
 // DataMoverConfig sets the DataUpload's DataMoverConfig.
-func (d *DataUploadBuilder) DataMoverConfig(config *map[string]string) *DataUploadBuilder {
+func (d *DataUploadBuilder) DataMoverConfig(config map[string]string) *DataUploadBuilder {
 	d.object.Spec.DataMoverConfig = config
 	return d
 }

pkg/builder/restore_builder.go

@@ -171,3 +171,9 @@ func (b *RestoreBuilder) ItemOperationTimeout(timeout time.Duration) *RestoreBui
 	b.object.Spec.ItemOperationTimeout.Duration = timeout
 	return b
 }
+
+// WriteSparseFiles sets the Restore's uploader write sparse files
+func (b *RestoreBuilder) WriteSparseFiles(val bool) *RestoreBuilder {
+	b.object.Spec.UploaderConfig.WriteSparseFiles = &val
+	return b
+}

pkg/cmd/cli/restore/create.go

@@ -98,6 +98,7 @@ type CreateOptions struct {
 	AllowPartiallyFailed      flag.OptionalBool
 	ItemOperationTimeout      time.Duration
 	ResourceModifierConfigMap string
+	WriteSparseFiles          flag.OptionalBool
 	client                    kbclient.WithWatch
 }
 
@@ -109,6 +110,7 @@ func NewCreateOptions() *CreateOptions {
 		RestoreVolumes:          flag.NewOptionalBool(nil),
 		PreserveNodePorts:       flag.NewOptionalBool(nil),
 		IncludeClusterResources: flag.NewOptionalBool(nil),
+		WriteSparseFiles:        flag.NewOptionalBool(nil),
 	}
 }
 
@@ -146,6 +148,9 @@ func (o *CreateOptions) BindFlags(flags *pflag.FlagSet) {
 	flags.BoolVarP(&o.Wait, "wait", "w", o.Wait, "Wait for the operation to complete.")
 
 	flags.StringVar(&o.ResourceModifierConfigMap, "resource-modifier-configmap", "", "Reference to the resource modifier configmap that restore will use")
+
+	f = flags.VarPF(&o.WriteSparseFiles, "write-sparse-files", "", "Whether to write sparse files during restoring volumes")
+	f.NoOptDefVal = cmd.TRUE
 }
 
 func (o *CreateOptions) Complete(args []string, f client.Factory) error {
@@ -318,6 +323,9 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
 			ItemOperationTimeout: metav1.Duration{
 				Duration: o.ItemOperationTimeout,
 			},
+			UploaderConfig: &api.UploaderConfigForRestore{
+				WriteSparseFiles: o.WriteSparseFiles.Value,
+			},
 		},
 	}
 

pkg/cmd/cli/restore/create_test.go

@@ -84,6 +84,7 @@ func TestCreateCommand(t *testing.T) {
 		includeClusterResources := "true"
 		allowPartiallyFailed := "true"
 		itemOperationTimeout := "10m0s"
+		writeSparseFiles := "true"
 
 		flags := new(pflag.FlagSet)
 		o := NewCreateOptions()
@@ -106,7 +107,7 @@ func TestCreateCommand(t *testing.T) {
 		flags.Parse([]string{"--include-cluster-resources", includeClusterResources})
 		flags.Parse([]string{"--allow-partially-failed", allowPartiallyFailed})
 		flags.Parse([]string{"--item-operation-timeout", itemOperationTimeout})
-
+		flags.Parse([]string{"--write-sparse-files", writeSparseFiles})
 		client := velerotest.NewFakeControllerRuntimeClient(t).(kbclient.WithWatch)
 
 		f.On("Namespace").Return(mock.Anything)
@@ -142,7 +143,7 @@ func TestCreateCommand(t *testing.T) {
 		require.Equal(t, includeClusterResources, o.IncludeClusterResources.String())
 		require.Equal(t, allowPartiallyFailed, o.AllowPartiallyFailed.String())
 		require.Equal(t, itemOperationTimeout, o.ItemOperationTimeout.String())
-
+		require.Equal(t, writeSparseFiles, o.WriteSparseFiles.String())
 	})
 
 	t.Run("create a restore from schedule", func(t *testing.T) {

pkg/cmd/util/output/backup_describer.go

@@ -88,9 +88,9 @@ func DescribeBackup(
 			DescribeResourcePolicies(d, backup.Spec.ResourcePolicy)
 		}
 
-		if backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
+		if backup.Spec.UploaderConfig != nil && backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
 			d.Println()
-			DescribeUploaderConfig(d, backup.Spec)
+			DescribeUploaderConfigForBackup(d, backup.Spec)
 		}
 
 		status := backup.Status
@@ -135,8 +135,8 @@ func DescribeResourcePolicies(d *Describer, resPolicies *v1.TypedLocalObjectRefe
 	d.Printf("\tName:\t%s\n", resPolicies.Name)
 }
 
-// DescribeUploaderConfig describes uploader config in human-readable format
-func DescribeUploaderConfig(d *Describer, spec velerov1api.BackupSpec) {
+// DescribeUploaderConfigForBackup describes uploader config in human-readable format
+func DescribeUploaderConfigForBackup(d *Describer, spec velerov1api.BackupSpec) {
 	d.Printf("Uploader config:\n")
 	d.Printf("\tParallel files upload:\t%d\n", spec.UploaderConfig.ParallelFilesUpload)
 }

pkg/cmd/util/output/backup_describer_test.go

@@ -28,7 +28,7 @@ func TestDescribeUploaderConfig(t *testing.T) {
 		buf:    &bytes.Buffer{},
 	}
 	d.out.Init(d.buf, 0, 8, 2, ' ', 0)
-	DescribeUploaderConfig(d, input)
+	DescribeUploaderConfigForBackup(d, input)
 	d.out.Flush()
 	expect := `Uploader config:
   Parallel files upload:  10

pkg/cmd/util/output/restore_describer.go

@@ -32,6 +32,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/downloadrequest"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/results"
 )
 
@@ -177,6 +178,11 @@ func DescribeRestore(ctx context.Context, kbClient kbclient.Client, restore *vel
 		d.Println()
 		d.Printf("Preserve Service NodePorts:\t%s\n", BoolPointerString(restore.Spec.PreserveNodePorts, "false", "true", "auto"))
 
+		if restore.Spec.UploaderConfig != nil && boolptr.IsSetToTrue(restore.Spec.UploaderConfig.WriteSparseFiles) {
+			d.Println()
+			DescribeUploaderConfigForRestore(d, restore.Spec)
+		}
+
 		d.Println()
 		describeRestoreItemOperations(ctx, kbClient, d, restore, details, insecureSkipTLSVerify, caCertFile)
 
@@ -193,6 +199,12 @@ func DescribeRestore(ctx context.Context, kbClient kbclient.Client, restore *vel
 	})
 }
 
+// DescribeUploaderConfigForRestore describes uploader config in human-readable format
+func DescribeUploaderConfigForRestore(d *Describer, spec velerov1api.RestoreSpec) {
+	d.Printf("Uploader config:\n")
+	d.Printf("\tWrite Sparse Files:\t%T\n", boolptr.IsSetToTrue(spec.UploaderConfig.WriteSparseFiles))
+}
+
 func describeRestoreItemOperations(ctx context.Context, kbClient kbclient.Client, d *Describer, restore *velerov1api.Restore, details bool, insecureSkipTLSVerify bool, caCertPath string) {
 	status := restore.Status
 	if status.RestoreItemOperationsAttempted > 0 {

pkg/cmd/util/output/schedule_describer.go

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

pkg/controller/data_download_controller.go

@@ -333,7 +333,7 @@ func (r *DataDownloadReconciler) runCancelableDataPath(ctx context.Context, fsRe
 	}
 	log.WithField("path", path.ByPath).Info("fs init")
 
-	if err := fsRestore.StartRestore(dd.Spec.SnapshotID, path); err != nil {
+	if err := fsRestore.StartRestore(dd.Spec.SnapshotID, path, dd.Spec.DataMoverConfig); err != nil {
 		return r.errorOut(ctx, dd, err, fmt.Sprintf("error starting data path %s restore", path.ByPath), log)
 	}
 

pkg/controller/data_upload_controller.go

@@ -343,7 +343,8 @@ func (r *DataUploadReconciler) runCancelableDataUpload(ctx context.Context, fsBa
 	tags := map[string]string{
 		velerov1api.AsyncOperationIDLabel: du.Labels[velerov1api.AsyncOperationIDLabel],
 	}
-	if err := fsBackup.StartBackup(path, fmt.Sprintf("%s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC), "", false, tags, du.Spec.UploaderConfig); err != nil {
+
+	if err := fsBackup.StartBackup(path, fmt.Sprintf("%s/%s", du.Spec.SourceNamespace, du.Spec.SourcePVC), "", false, tags, du.Spec.DataMoverConfig); err != nil {
 		return r.errorOut(ctx, du, err, "error starting data path backup", log)
 	}
 

pkg/controller/data_upload_controller_test.go

@@ -45,7 +45,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
@@ -297,7 +296,7 @@ func (f *fakeDataUploadFSBR) Init(ctx context.Context, bslName string, sourceNam
 	return nil
 }
 
-func (f *fakeDataUploadFSBR) StartBackup(source datapath.AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfigs shared.UploaderConfig) error {
+func (f *fakeDataUploadFSBR) StartBackup(source datapath.AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfigs map[string]string) error {
 	du := f.du
 	original := f.du.DeepCopy()
 	du.Status.Phase = velerov2alpha1api.DataUploadPhaseCompleted
@@ -307,7 +306,7 @@ func (f *fakeDataUploadFSBR) StartBackup(source datapath.AccessPoint, realSource
 	return nil
 }
 
-func (f *fakeDataUploadFSBR) StartRestore(snapshotID string, target datapath.AccessPoint) error {
+func (f *fakeDataUploadFSBR) StartRestore(snapshotID string, target datapath.AccessPoint, uploaderConfigs map[string]string) error {
 	return nil
 }
 

pkg/controller/pod_volume_backup_controller.go

@@ -178,7 +178,7 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		}
 	}
 
-	if err := fsBackup.StartBackup(path, "", parentSnapshotID, false, pvb.Spec.Tags, pvb.Spec.UploaderConfig); err != nil {
+	if err := fsBackup.StartBackup(path, "", parentSnapshotID, false, pvb.Spec.Tags, pvb.Spec.UploaderSettings); err != nil {
 		return r.errorOut(ctx, &pvb, err, "error starting data path backup", log)
 	}
 

pkg/controller/pod_volume_backup_controller_test.go

@@ -37,7 +37,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
@@ -104,7 +103,7 @@ func (b *fakeFSBR) Init(ctx context.Context, bslName string, sourceNamespace str
 	return nil
 }
 
-func (b *fakeFSBR) StartBackup(source datapath.AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfigs shared.UploaderConfig) error {
+func (b *fakeFSBR) StartBackup(source datapath.AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfigs map[string]string) error {
 	pvb := b.pvb
 
 	original := b.pvb.DeepCopy()
@@ -116,7 +115,7 @@ func (b *fakeFSBR) StartBackup(source datapath.AccessPoint, realSource string, p
 	return nil
 }
 
-func (b *fakeFSBR) StartRestore(snapshotID string, target datapath.AccessPoint) error {
+func (b *fakeFSBR) StartRestore(snapshotID string, target datapath.AccessPoint, uploaderConfigs map[string]string) error {
 	return nil
 }
 

pkg/controller/pod_volume_restore_controller.go

@@ -147,7 +147,7 @@ func (c *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		return c.errorOut(ctx, pvr, err, "error to initialize data path", log)
 	}
 
-	if err := fsRestore.StartRestore(pvr.Spec.SnapshotID, volumePath); err != nil {
+	if err := fsRestore.StartRestore(pvr.Spec.SnapshotID, volumePath, pvr.Spec.UploaderSettings); err != nil {
 		return c.errorOut(ctx, pvr, err, "error starting data path restore", log)
 	}
 

pkg/datapath/file_system.go

@@ -24,7 +24,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	repokey "github.com/vmware-tanzu/velero/pkg/repository/keys"
@@ -130,14 +129,14 @@ func (fs *fileSystemBR) Close(ctx context.Context) {
 	fs.log.WithField("user", fs.jobName).Info("FileSystemBR is closed")
 }
 
-func (fs *fileSystemBR) StartBackup(source AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfigs shared.UploaderConfig) error {
+func (fs *fileSystemBR) StartBackup(source AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfig map[string]string) error {
 	if !fs.initialized {
 		return errors.New("file system data path is not initialized")
 	}
 
 	go func() {
 		snapshotID, emptySnapshot, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, realSource, tags, forceFull,
-			parentSnapshot, source.VolMode, uploaderConfigs, fs)
+			parentSnapshot, source.VolMode, uploaderConfig, fs)
 
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)
@@ -151,13 +150,13 @@ func (fs *fileSystemBR) StartBackup(source AccessPoint, realSource string, paren
 	return nil
 }
 
-func (fs *fileSystemBR) StartRestore(snapshotID string, target AccessPoint) error {
+func (fs *fileSystemBR) StartRestore(snapshotID string, target AccessPoint, uploaderConfigs map[string]string) error {
 	if !fs.initialized {
 		return errors.New("file system data path is not initialized")
 	}
 
 	go func() {
-		err := fs.uploaderProv.RunRestore(fs.ctx, snapshotID, target.ByPath, target.VolMode, fs)
+		err := fs.uploaderProv.RunRestore(fs.ctx, snapshotID, target.ByPath, target.VolMode, uploaderConfigs, fs)
 
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)

pkg/datapath/file_system_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 	providerMock "github.com/vmware-tanzu/velero/pkg/uploader/provider/mocks"
@@ -101,7 +100,7 @@ func TestAsyncBackup(t *testing.T) {
 			fs.initialized = true
 			fs.callbacks = test.callbacks
 
-			err := fs.StartBackup(AccessPoint{ByPath: test.path}, "", "", false, nil, shared.UploaderConfig{})
+			err := fs.StartBackup(AccessPoint{ByPath: test.path}, "", "", false, nil, map[string]string{})
 			require.Equal(t, nil, err)
 
 			<-finish
@@ -179,12 +178,12 @@ func TestAsyncRestore(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			fs := newFileSystemBR("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*fileSystemBR)
 			mockProvider := providerMock.NewProvider(t)
-			mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.err)
+			mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.err)
 			fs.uploaderProv = mockProvider
 			fs.initialized = true
 			fs.callbacks = test.callbacks
 
-			err := fs.StartRestore(test.snapshot, AccessPoint{ByPath: test.path})
+			err := fs.StartRestore(test.snapshot, AccessPoint{ByPath: test.path}, map[string]string{})
 			require.Equal(t, nil, err)
 
 			<-finish

pkg/datapath/mocks/types.go

@@ -11,8 +11,6 @@ import (
 	mock "github.com/stretchr/testify/mock"
 
 	repository "github.com/vmware-tanzu/velero/pkg/repository"
-
-	shared "github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 )
 
 // AsyncBR is an autogenerated mock type for the AsyncBR type
@@ -44,13 +42,13 @@ func (_m *AsyncBR) Init(ctx context.Context, bslName string, sourceNamespace str
 	return r0
 }
 
-// StartBackup provides a mock function with given fields: source, realSource, parentSnapshot, forceFull, tags, uploaderConfig
-func (_m *AsyncBR) StartBackup(source datapath.AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfig shared.UploaderConfig) error {
-	ret := _m.Called(source, realSource, parentSnapshot, forceFull, tags, uploaderConfig)
+// StartBackup provides a mock function with given fields: source, realSource, parentSnapshot, forceFull, tags, dataMoverConfig
+func (_m *AsyncBR) StartBackup(source datapath.AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, dataMoverConfig map[string]string) error {
+	ret := _m.Called(source, realSource, parentSnapshot, forceFull, tags, dataMoverConfig)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(datapath.AccessPoint, string, string, bool, map[string]string, shared.UploaderConfig) error); ok {
-		r0 = rf(source, realSource, parentSnapshot, forceFull, tags, uploaderConfig)
+	if rf, ok := ret.Get(0).(func(datapath.AccessPoint, string, string, bool, map[string]string, map[string]string) error); ok {
+		r0 = rf(source, realSource, parentSnapshot, forceFull, tags, dataMoverConfig)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -58,13 +56,13 @@ func (_m *AsyncBR) StartBackup(source datapath.AccessPoint, realSource string, p
 	return r0
 }
 
-// StartRestore provides a mock function with given fields: snapshotID, target
-func (_m *AsyncBR) StartRestore(snapshotID string, target datapath.AccessPoint) error {
-	ret := _m.Called(snapshotID, target)
+// StartRestore provides a mock function with given fields: snapshotID, target, dataMoverConfig
+func (_m *AsyncBR) StartRestore(snapshotID string, target datapath.AccessPoint, dataMoverConfig map[string]string) error {
+	ret := _m.Called(snapshotID, target, dataMoverConfig)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, datapath.AccessPoint) error); ok {
-		r0 = rf(snapshotID, target)
+	if rf, ok := ret.Get(0).(func(string, datapath.AccessPoint, map[string]string) error); ok {
+		r0 = rf(snapshotID, target, dataMoverConfig)
 	} else {
 		r0 = ret.Error(0)
 	}

pkg/datapath/types.go

@@ -20,7 +20,6 @@ import (
 	"context"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 )
@@ -63,10 +62,10 @@ type AsyncBR interface {
 	Init(ctx context.Context, bslName string, sourceNamespace string, uploaderType string, repositoryType string, repoIdentifier string, repositoryEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter) error
 
 	// StartBackup starts an asynchronous data path instance for backup
-	StartBackup(source AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfig shared.UploaderConfig) error
+	StartBackup(source AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, dataMoverConfig map[string]string) error
 
 	// StartRestore starts an asynchronous data path instance for restore
-	StartRestore(snapshotID string, target AccessPoint) error
+	StartRestore(snapshotID string, target AccessPoint, dataMoverConfig map[string]string) error
 
 	// Cancel cancels an asynchronous data path instance
 	Cancel()

pkg/podvolume/backupper.go

@@ -36,6 +36,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	"github.com/vmware-tanzu/velero/pkg/repository"
+	uploaderutil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
@@ -398,7 +399,6 @@ func newPodVolumeBackup(backup *velerov1api.Backup, pod *corev1api.Pod, volume c
 			BackupStorageLocation: backup.Spec.StorageLocation,
 			RepoIdentifier:        repoIdentifier,
 			UploaderType:          uploaderType,
-			UploaderConfig:        backup.Spec.UploaderConfig,
 		},
 	}
 
@@ -417,5 +417,9 @@ func newPodVolumeBackup(backup *velerov1api.Backup, pod *corev1api.Pod, volume c
 		pvb.Spec.Tags["pvc-uid"] = string(pvc.UID)
 	}
 
+	if backup.Spec.UploaderConfig != nil {
+		pvb.Spec.UploaderSettings = uploaderutil.StoreBackupConfig(backup.Spec.UploaderConfig)
+	}
+
 	return pvb
 }

pkg/podvolume/restorer.go

@@ -36,6 +36,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/nodeagent"
 	"github.com/vmware-tanzu/velero/pkg/repository"
+	uploaderutil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
@@ -177,7 +178,6 @@ func (r *restorer) RestorePodVolumes(data RestoreData) []error {
 		}
 
 		volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, repoIdentifier, backupInfo.uploaderType, data.SourceNamespace, pvc)
-
 		if err := veleroclient.CreateRetryGenerateName(r.crClient, r.ctx, volumeRestore); err != nil {
 			errs = append(errs, errors.WithStack(err))
 			continue
@@ -286,6 +286,11 @@ func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, backu
 		// this label is not used by velero, but useful for debugging.
 		pvr.Labels[velerov1api.PVCUIDLabel] = string(pvc.UID)
 	}
+
+	if restore.Spec.UploaderConfig != nil {
+		pvr.Spec.UploaderSettings = uploaderutil.StoreRestoreConfig(restore.Spec.UploaderConfig)
+	}
+
 	return pvr
 }
 

pkg/uploader/kopia/snapshot.go

@@ -38,10 +38,10 @@ import (
 	"github.com/kopia/kopia/snapshot/snapshotfs"
 	"github.com/pkg/errors"
 
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	"github.com/vmware-tanzu/velero/pkg/kopia"
 	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
+	uploaderutil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 )
 
 // All function mainly used to make testing more convenient
@@ -105,11 +105,18 @@ func getDefaultPolicy() *policy.Policy {
 	}
 }
 
-func setupPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceInfo snapshot.SourceInfo, uploaderCfg shared.UploaderConfig) (*policy.Tree, error) {
+func setupPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceInfo snapshot.SourceInfo, uploaderCfg map[string]string) (*policy.Tree, error) {
 	// some internal operations from Kopia code retrieves policies from repo directly, so we need to persist the policy to repo
 	curPolicy := getDefaultPolicy()
-	if uploaderCfg.ParallelFilesUpload > 0 {
-		curPolicy.UploadPolicy.MaxParallelFileReads = newOptionalInt(uploaderCfg.ParallelFilesUpload)
+
+	if len(uploaderCfg) > 0 {
+		parallelUpload, err := uploaderutil.GetParallelFilesUpload(uploaderCfg)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to get uploader config")
+		}
+		if parallelUpload > 0 {
+			curPolicy.UploadPolicy.MaxParallelFileReads = newOptionalInt(parallelUpload)
+		}
 	}
 
 	err := setPolicyFunc(ctx, rep, sourceInfo, curPolicy)
@@ -133,7 +140,7 @@ func setupPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceInfo snap
 
 // Backup backup specific sourcePath and update progress
 func Backup(ctx context.Context, fsUploader SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string,
-	forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg shared.UploaderConfig, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
+	forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
 	if fsUploader == nil {
 		return nil, false, errors.New("get empty kopia uploader")
 	}
@@ -229,7 +236,7 @@ func SnapshotSource(
 	forceFull bool,
 	parentSnapshot string,
 	snapshotTags map[string]string,
-	uploaderCfg shared.UploaderConfig,
+	uploaderCfg map[string]string,
 	log logrus.FieldLogger,
 	description string,
 ) (string, int64, error) {
@@ -361,7 +368,7 @@ func findPreviousSnapshotManifest(ctx context.Context, rep repo.Repository, sour
 }
 
 // Restore restore specific sourcePath with given snapshotID and update progress
-func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode,
+func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string,
 	log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 	log.Info("Start to restore...")
 
@@ -392,6 +399,18 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		IgnorePermissionErrors: true,
 	}
 
+	if len(uploaderCfg) > 0 {
+		writeSparseFiles, err := uploaderutil.GetWriteSparseFiles(uploaderCfg)
+		if err != nil {
+			return 0, 0, errors.Wrap(err, "failed to get uploader config")
+		}
+		if writeSparseFiles {
+			fsOutput.WriteSparseFiles = true
+		}
+	}
+
+	log.Debugf("Restore filesystem output %v", fsOutput)
+
 	err = fsOutput.Init(ctx)
 	if err != nil {
 		return 0, 0, errors.Wrap(err, "error to init output")

pkg/uploader/kopia/snapshot_test.go

@@ -35,7 +35,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	repomocks "github.com/vmware-tanzu/velero/pkg/repository/mocks"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 	uploadermocks "github.com/vmware-tanzu/velero/pkg/uploader/mocks"
@@ -97,7 +96,7 @@ func TestSnapshotSource(t *testing.T) {
 	testCases := []struct {
 		name        string
 		args        []mockArgs
-		uploaderCfg shared.UploaderConfig
+		uploaderCfg map[string]string
 		notError    bool
 	}{
 		{
@@ -153,7 +152,7 @@ func TestSnapshotSource(t *testing.T) {
 			notError: false,
 		},
 		{
-			name: "set policy with ParallelFilesUpload",
+			name: "set policy with parallel files upload",
 			args: []mockArgs{
 				{methodName: "LoadSnapshot", returns: []interface{}{manifest, nil}},
 				{methodName: "SaveSnapshot", returns: []interface{}{manifest.ID, nil}},
@@ -163,8 +162,10 @@ func TestSnapshotSource(t *testing.T) {
 				{methodName: "Upload", returns: []interface{}{manifest, nil}},
 				{methodName: "Flush", returns: []interface{}{nil}},
 			},
-			uploaderCfg: shared.UploaderConfig{ParallelFilesUpload: 10},
-			notError:    true,
+			uploaderCfg: map[string]string{
+				"ParallelFilesUpload": "10",
+			},
+			notError: true,
 		},
 		{
 			name: "failed to upload snapshot",
@@ -646,9 +647,9 @@ func TestBackup(t *testing.T) {
 			var snapshotInfo *uploader.SnapshotInfo
 			var err error
 			if tc.isEmptyUploader {
-				snapshotInfo, isSnapshotEmpty, err = Backup(context.Background(), nil, s.repoWriterMock, tc.sourcePath, "", tc.forceFull, tc.parentSnapshot, tc.volMode, shared.UploaderConfig{}, tc.tags, &logrus.Logger{})
+				snapshotInfo, isSnapshotEmpty, err = Backup(context.Background(), nil, s.repoWriterMock, tc.sourcePath, "", tc.forceFull, tc.parentSnapshot, tc.volMode, map[string]string{}, tc.tags, &logrus.Logger{})
 			} else {
-				snapshotInfo, isSnapshotEmpty, err = Backup(context.Background(), s.uploderMock, s.repoWriterMock, tc.sourcePath, "", tc.forceFull, tc.parentSnapshot, tc.volMode, shared.UploaderConfig{}, tc.tags, &logrus.Logger{})
+				snapshotInfo, isSnapshotEmpty, err = Backup(context.Background(), s.uploderMock, s.repoWriterMock, tc.sourcePath, "", tc.forceFull, tc.parentSnapshot, tc.volMode, map[string]string{}, tc.tags, &logrus.Logger{})
 			}
 			// Check if the returned error matches the expected error
 			if tc.expectedError != nil {
@@ -787,7 +788,7 @@ func TestRestore(t *testing.T) {
 			repoWriterMock.On("OpenObject", mock.Anything, mock.Anything).Return(em, nil)
 
 			progress := new(Progress)
-			bytesRestored, fileCount, err := Restore(context.Background(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.volMode, logrus.New(), nil)
+			bytesRestored, fileCount, err := Restore(context.Background(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.volMode, map[string]string{}, logrus.New(), nil)
 
 			// Check if the returned error matches the expected error
 			if tc.expectedError != nil {

pkg/uploader/provider/kopia.go

@@ -30,7 +30,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/uploader/kopia"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	repokeys "github.com/vmware-tanzu/velero/pkg/repository/keys"
 	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
@@ -120,7 +119,7 @@ func (kp *kopiaProvider) RunBackup(
 	forceFull bool,
 	parentSnapshot string,
 	volMode uploader.PersistentVolumeMode,
-	uploaderCfg shared.UploaderConfig,
+	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) (string, bool, error) {
 	if updater == nil {
 		return "", false, errors.New("Need to initial backup progress updater first")
@@ -205,6 +204,7 @@ func (kp *kopiaProvider) RunRestore(
 	snapshotID string,
 	volumePath string,
 	volMode uploader.PersistentVolumeMode,
+	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) error {
 	log := kp.log.WithFields(logrus.Fields{
 		"snapshotID": snapshotID,
@@ -228,7 +228,7 @@ func (kp *kopiaProvider) RunRestore(
 	// We use the cancel channel to control the restore cancel, so don't pass a context with cancel to Kopia restore.
 	// Otherwise, Kopia restore will not response to the cancel control but return an arbitrary error.
 	// Kopia restore cancel is not designed as well as Kopia backup which uses the context to control backup cancel all the way.
-	size, fileCount, err := RestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, volMode, log, restoreCancel)
+	size, fileCount, err := RestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, volMode, uploaderCfg, log, restoreCancel)
 
 	if err != nil {
 		return errors.Wrapf(err, "Failed to run kopia restore")

pkg/uploader/provider/kopia_test.go

@@ -34,7 +34,6 @@ import (
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	"github.com/vmware-tanzu/velero/internal/credentials/mocks"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	udmrepo "github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
@@ -69,34 +68,34 @@ func TestRunBackup(t *testing.T) {
 
 	testCases := []struct {
 		name           string
-		hookBackupFunc func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg shared.UploaderConfig, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error)
+		hookBackupFunc func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error)
 		volMode        uploader.PersistentVolumeMode
 		notError       bool
 	}{
 		{
 			name: "success to backup",
-			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg shared.UploaderConfig, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
+			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
 				return &uploader.SnapshotInfo{}, false, nil
 			},
 			notError: true,
 		},
 		{
 			name: "get error to backup",
-			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg shared.UploaderConfig, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
+			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
 				return &uploader.SnapshotInfo{}, false, errors.New("failed to backup")
 			},
 			notError: false,
 		},
 		{
 			name: "got empty snapshot",
-			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg shared.UploaderConfig, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
+			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
 				return nil, true, errors.New("snapshot is empty")
 			},
 			notError: false,
 		},
 		{
 			name: "success to backup block mode volume",
-			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg shared.UploaderConfig, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
+			hookBackupFunc: func(ctx context.Context, fsUploader kopia.SnapshotUploader, repoWriter repo.RepositoryWriter, sourcePath string, realSource string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, tags map[string]string, log logrus.FieldLogger) (*uploader.SnapshotInfo, bool, error) {
 				return &uploader.SnapshotInfo{}, false, nil
 			},
 			volMode:  uploader.PersistentVolumeBlock,
@@ -109,7 +108,7 @@ func TestRunBackup(t *testing.T) {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
 			BackupFunc = tc.hookBackupFunc
-			_, _, err := kp.RunBackup(context.Background(), "var", "", nil, false, "", tc.volMode, shared.UploaderConfig{}, &updater)
+			_, _, err := kp.RunBackup(context.Background(), "var", "", nil, false, "", tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {
@@ -126,27 +125,27 @@ func TestRunRestore(t *testing.T) {
 
 	testCases := []struct {
 		name            string
-		hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error)
+		hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error)
 		notError        bool
 		volMode         uploader.PersistentVolumeMode
 	}{
 		{
 			name: "normal restore",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 				return 0, 0, nil
 			},
 			notError: true,
 		},
 		{
 			name: "failed to restore",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 				return 0, 0, errors.New("failed to restore")
 			},
 			notError: false,
 		},
 		{
 			name: "normal block mode restore",
-			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 				return 0, 0, nil
 			},
 			volMode:  uploader.PersistentVolumeBlock,
@@ -160,7 +159,7 @@ func TestRunRestore(t *testing.T) {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
 			RestoreFunc = tc.hookRestoreFunc
-			err := kp.RunRestore(context.Background(), "", "/var", tc.volMode, &updater)
+			err := kp.RunRestore(context.Background(), "", "/var", tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {

pkg/uploader/provider/mocks/Provider.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v2.22.1. DO NOT EDIT.
+// Code generated by mockery v2.20.0. DO NOT EDIT.
 
 package mocks
 
@@ -7,8 +7,6 @@ import (
 
 	mock "github.com/stretchr/testify/mock"
 
-	shared "github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
-
 	uploader "github.com/vmware-tanzu/velero/pkg/uploader"
 )
 
@@ -32,28 +30,28 @@ func (_m *Provider) Close(ctx context.Context) error {
 }
 
 // RunBackup provides a mock function with given fields: ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater
-func (_m *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg shared.UploaderConfig, updater uploader.ProgressUpdater) (string, bool, error) {
+func (_m *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, error) {
 	ret := _m.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
 
 	var r0 string
 	var r1 bool
 	var r2 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, shared.UploaderConfig, uploader.ProgressUpdater) (string, bool, error)); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, error)); ok {
 		return rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
 	}
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, shared.UploaderConfig, uploader.ProgressUpdater) string); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok {
 		r0 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
 	} else {
 		r0 = ret.Get(0).(string)
 	}
 
-	if rf, ok := ret.Get(1).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, shared.UploaderConfig, uploader.ProgressUpdater) bool); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok {
 		r1 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
 	} else {
 		r1 = ret.Get(1).(bool)
 	}
 
-	if rf, ok := ret.Get(2).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, shared.UploaderConfig, uploader.ProgressUpdater) error); ok {
+	if rf, ok := ret.Get(2).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
 		r2 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
 	} else {
 		r2 = ret.Error(2)
@@ -62,13 +60,13 @@ func (_m *Provider) RunBackup(ctx context.Context, path string, realSource strin
 	return r0, r1, r2
 }
 
-// RunRestore provides a mock function with given fields: ctx, snapshotID, volumePath, volMode, updater
-func (_m *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, updater uploader.ProgressUpdater) error {
-	ret := _m.Called(ctx, snapshotID, volumePath, volMode, updater)
+// RunRestore provides a mock function with given fields: ctx, snapshotID, volumePath, volMode, uploaderConfig, updater
+func (_m *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) error {
+	ret := _m.Called(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, uploader.ProgressUpdater) error); ok {
-		r0 = rf(ctx, snapshotID, volumePath, volMode, updater)
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r0 = rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
 	} else {
 		r0 = ret.Error(0)
 	}

pkg/uploader/provider/provider.go

@@ -28,7 +28,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 )
@@ -50,7 +49,7 @@ type Provider interface {
 		forceFull bool,
 		parentSnapshot string,
 		volMode uploader.PersistentVolumeMode,
-		uploaderCfg shared.UploaderConfig,
+		uploaderCfg map[string]string,
 		updater uploader.ProgressUpdater) (string, bool, error)
 	// RunRestore which will do restore for one specific volume with given snapshot id and return error
 	// updater is used for updating backup progress which implement by third-party
@@ -59,6 +58,7 @@ type Provider interface {
 		snapshotID string,
 		volumePath string,
 		volMode uploader.PersistentVolumeMode,
+		uploaderConfig map[string]string,
 		updater uploader.ProgressUpdater) error
 	// Close which will close related repository
 	Close(ctx context.Context) error

pkg/uploader/provider/restic.go

@@ -27,10 +27,10 @@ import (
 	v1 "k8s.io/api/core/v1"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/restic"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
+	uploaderutil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 )
 
@@ -123,7 +123,7 @@ func (rp *resticProvider) RunBackup(
 	forceFull bool,
 	parentSnapshot string,
 	volMode uploader.PersistentVolumeMode,
-	uploaderCfg shared.UploaderConfig,
+	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) (string, bool, error) {
 	if updater == nil {
 		return "", false, errors.New("Need to initial backup progress updater first")
@@ -146,8 +146,14 @@ func (rp *resticProvider) RunBackup(
 		"parentSnapshot": parentSnapshot,
 	})
 
-	if uploaderCfg.ParallelFilesUpload > 0 {
-		log.Warnf("ParallelFilesUpload is set to %d, but restic does not support parallel file uploads. Ignoring.", uploaderCfg.ParallelFilesUpload)
+	if len(uploaderCfg) > 0 {
+		parallelFilesUpload, err := uploaderutil.GetParallelFilesUpload(uploaderCfg)
+		if err != nil {
+			return "", false, errors.Wrap(err, "failed to get uploader config")
+		}
+		if parallelFilesUpload > 0 {
+			log.Warnf("ParallelFilesUpload is set to %d, but restic does not support parallel file uploads. Ignoring.", parallelFilesUpload)
+		}
 	}
 
 	backupCmd := resticBackupCMDFunc(rp.repoIdentifier, rp.credentialsFile, path, tags)
@@ -191,6 +197,7 @@ func (rp *resticProvider) RunRestore(
 	snapshotID string,
 	volumePath string,
 	volMode uploader.PersistentVolumeMode,
+	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) error {
 	if updater == nil {
 		return errors.New("Need to initial backup progress updater first")
@@ -210,8 +217,34 @@ func (rp *resticProvider) RunRestore(
 	if len(rp.extraFlags) != 0 {
 		restoreCmd.ExtraFlags = append(restoreCmd.ExtraFlags, rp.extraFlags...)
 	}
+
+	extraFlags, err := rp.parseRestoreExtraFlags(uploaderCfg)
+	if err != nil {
+		return errors.Wrap(err, "failed to parse uploader config")
+	} else if len(extraFlags) != 0 {
+		restoreCmd.ExtraFlags = append(restoreCmd.ExtraFlags, extraFlags...)
+	}
+
 	stdout, stderr, err := restic.RunRestore(restoreCmd, log, updater)
 
-	log.Infof("Run command=%s, stdout=%s, stderr=%s", restoreCmd.Command, stdout, stderr)
+	log.Infof("Run command=%v, stdout=%s, stderr=%s", restoreCmd, stdout, stderr)
 	return err
 }
+
+func (rp *resticProvider) parseRestoreExtraFlags(uploaderCfg map[string]string) ([]string, error) {
+	extraFlags := []string{}
+	if len(uploaderCfg) == 0 {
+		return extraFlags, nil
+	}
+
+	writeSparseFiles, err := uploaderutil.GetWriteSparseFiles(uploaderCfg)
+	if err != nil {
+		return extraFlags, errors.Wrap(err, "failed to get uploader config")
+	}
+
+	if writeSparseFiles {
+		extraFlags = append(extraFlags, "--sparse")
+	}
+
+	return extraFlags, nil
+}

pkg/uploader/provider/restic_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"os"
+	"reflect"
 	"strings"
 	"testing"
 
@@ -31,7 +32,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/restic"
@@ -150,9 +150,9 @@ func TestResticRunBackup(t *testing.T) {
 			}
 			if !tc.nilUpdater {
 				updater := FakeBackupProgressUpdater{PodVolumeBackup: &velerov1api.PodVolumeBackup{}, Log: tc.rp.log, Ctx: context.Background(), Cli: fake.NewClientBuilder().WithScheme(util.VeleroScheme).Build()}
-				_, _, err = tc.rp.RunBackup(context.Background(), "var", "", map[string]string{}, false, parentSnapshot, tc.volMode, shared.UploaderConfig{}, &updater)
+				_, _, err = tc.rp.RunBackup(context.Background(), "var", "", map[string]string{}, false, parentSnapshot, tc.volMode, map[string]string{}, &updater)
 			} else {
-				_, _, err = tc.rp.RunBackup(context.Background(), "var", "", map[string]string{}, false, parentSnapshot, tc.volMode, shared.UploaderConfig{}, nil)
+				_, _, err = tc.rp.RunBackup(context.Background(), "var", "", map[string]string{}, false, parentSnapshot, tc.volMode, map[string]string{}, nil)
 			}
 
 			tc.rp.log.Infof("test name %v error %v", tc.name, err)
@@ -223,9 +223,9 @@ func TestResticRunRestore(t *testing.T) {
 			var err error
 			if !tc.nilUpdater {
 				updater := FakeBackupProgressUpdater{PodVolumeBackup: &velerov1api.PodVolumeBackup{}, Log: tc.rp.log, Ctx: context.Background(), Cli: fake.NewClientBuilder().WithScheme(util.VeleroScheme).Build()}
-				err = tc.rp.RunRestore(context.Background(), "", "var", tc.volMode, &updater)
+				err = tc.rp.RunRestore(context.Background(), "", "var", tc.volMode, map[string]string{}, &updater)
 			} else {
-				err = tc.rp.RunRestore(context.Background(), "", "var", tc.volMode, nil)
+				err = tc.rp.RunRestore(context.Background(), "", "var", tc.volMode, map[string]string{}, nil)
 			}
 
 			tc.rp.log.Infof("test name %v error %v", tc.name, err)
@@ -411,3 +411,42 @@ func TestNewResticUploaderProvider(t *testing.T) {
 		})
 	}
 }
+
+func TestParseUploaderConfig(t *testing.T) {
+	rp := &resticProvider{}
+
+	testCases := []struct {
+		name           string
+		uploaderConfig map[string]string
+		expectedFlags  []string
+	}{
+		{
+			name: "SparseFilesEnabled",
+			uploaderConfig: map[string]string{
+				"WriteSparseFiles": "true",
+			},
+			expectedFlags: []string{"--sparse"},
+		},
+		{
+			name: "SparseFilesDisabled",
+			uploaderConfig: map[string]string{
+				"writeSparseFiles": "false",
+			},
+			expectedFlags: []string{},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			result, err := rp.parseRestoreExtraFlags(testCase.uploaderConfig)
+			if err != nil {
+				t.Errorf("Test case %s failed with error: %v", testCase.name, err)
+				return
+			}
+
+			if !reflect.DeepEqual(result, testCase.expectedFlags) {
+				t.Errorf("Test case %s failed. Expected: %v, Got: %v", testCase.name, testCase.expectedFlags, result)
+			}
+		})
+	}
+}

pkg/uploader/util/uploader_config.go

@@ -0,0 +1,70 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import (
+	"strconv"
+
+	"github.com/pkg/errors"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+)
+
+const (
+	parallelFilesUpload = "ParallelFilesUpload"
+	writeSparseFiles    = "WriteSparseFiles"
+)
+
+func StoreBackupConfig(config *velerov1api.UploaderConfigForBackup) map[string]string {
+	data := make(map[string]string)
+	data[parallelFilesUpload] = strconv.Itoa(config.ParallelFilesUpload)
+	return data
+}
+
+func StoreRestoreConfig(config *velerov1api.UploaderConfigForRestore) map[string]string {
+	data := make(map[string]string)
+	if config.WriteSparseFiles != nil {
+		data[writeSparseFiles] = strconv.FormatBool(*config.WriteSparseFiles)
+	} else {
+		data[writeSparseFiles] = strconv.FormatBool(false)
+	}
+	return data
+}
+
+func GetParallelFilesUpload(uploaderCfg map[string]string) (int, error) {
+	parallelFilesUpload, ok := uploaderCfg[parallelFilesUpload]
+	if ok {
+		parallelFilesUploadInt, err := strconv.Atoi(parallelFilesUpload)
+		if err != nil {
+			return 0, errors.Wrap(err, "failed to parse ParallelFilesUpload config")
+		}
+		return parallelFilesUploadInt, nil
+	}
+	return 0, nil
+}
+
+func GetWriteSparseFiles(uploaderCfg map[string]string) (bool, error) {
+	writeSparseFiles, ok := uploaderCfg[writeSparseFiles]
+	if ok {
+		writeSparseFilesBool, err := strconv.ParseBool(writeSparseFiles)
+		if err != nil {
+			return false, errors.Wrap(err, "failed to parse WriteSparseFiles config")
+		}
+		return writeSparseFilesBool, nil
+	}
+	return false, nil
+}

pkg/uploader/util/uploader_config_test.go

@@ -0,0 +1,149 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/pkg/errors"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+)
+
+func TestStoreBackupConfig(t *testing.T) {
+	config := &velerov1api.UploaderConfigForBackup{
+		ParallelFilesUpload: 3,
+	}
+
+	expectedData := map[string]string{
+		parallelFilesUpload: "3",
+	}
+
+	result := StoreBackupConfig(config)
+
+	if !reflect.DeepEqual(result, expectedData) {
+		t.Errorf("Expected: %v, but got: %v", expectedData, result)
+	}
+}
+
+func TestStoreRestoreConfig(t *testing.T) {
+	boolTrue := true
+	config := &velerov1api.UploaderConfigForRestore{
+		WriteSparseFiles: &boolTrue,
+	}
+
+	expectedData := map[string]string{
+		writeSparseFiles: "true",
+	}
+
+	result := StoreRestoreConfig(config)
+
+	if !reflect.DeepEqual(result, expectedData) {
+		t.Errorf("Expected: %v, but got: %v", expectedData, result)
+	}
+}
+
+func TestGetParallelFilesUpload(t *testing.T) {
+	tests := []struct {
+		name           string
+		uploaderCfg    map[string]string
+		expectedResult int
+		expectedError  error
+	}{
+		{
+			name:           "Valid ParallelFilesUpload",
+			uploaderCfg:    map[string]string{parallelFilesUpload: "5"},
+			expectedResult: 5,
+			expectedError:  nil,
+		},
+		{
+			name:           "Missing ParallelFilesUpload",
+			uploaderCfg:    map[string]string{},
+			expectedResult: 0,
+			expectedError:  nil,
+		},
+		{
+			name:           "Invalid ParallelFilesUpload (not a number)",
+			uploaderCfg:    map[string]string{parallelFilesUpload: "invalid"},
+			expectedResult: 0,
+			expectedError:  errors.Wrap(errors.New("strconv.Atoi: parsing \"invalid\": invalid syntax"), "failed to parse ParallelFilesUpload config"),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result, err := GetParallelFilesUpload(test.uploaderCfg)
+
+			if result != test.expectedResult {
+				t.Errorf("Expected result %d, but got %d", test.expectedResult, result)
+			}
+
+			if (err == nil && test.expectedError != nil) || (err != nil && test.expectedError == nil) || (err != nil && test.expectedError != nil && err.Error() != test.expectedError.Error()) {
+				t.Errorf("Expected error '%v', but got '%v'", test.expectedError, err)
+			}
+		})
+	}
+}
+
+func TestGetWriteSparseFiles(t *testing.T) {
+	tests := []struct {
+		name           string
+		uploaderCfg    map[string]string
+		expectedResult bool
+		expectedError  error
+	}{
+		{
+			name:           "Valid WriteSparseFiles (true)",
+			uploaderCfg:    map[string]string{writeSparseFiles: "true"},
+			expectedResult: true,
+			expectedError:  nil,
+		},
+		{
+			name:           "Valid WriteSparseFiles (false)",
+			uploaderCfg:    map[string]string{writeSparseFiles: "false"},
+			expectedResult: false,
+			expectedError:  nil,
+		},
+		{
+			name:           "Invalid WriteSparseFiles (not a boolean)",
+			uploaderCfg:    map[string]string{writeSparseFiles: "invalid"},
+			expectedResult: false,
+			expectedError:  errors.Wrap(errors.New("strconv.ParseBool: parsing \"invalid\": invalid syntax"), "failed to parse WriteSparseFiles config"),
+		},
+		{
+			name:           "Missing WriteSparseFiles",
+			uploaderCfg:    map[string]string{},
+			expectedResult: false,
+			expectedError:  nil,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result, err := GetWriteSparseFiles(test.uploaderCfg)
+
+			if result != test.expectedResult {
+				t.Errorf("Expected result %t, but got %t", test.expectedResult, result)
+			}
+
+			if (err == nil && test.expectedError != nil) || (err != nil && test.expectedError == nil) || (err != nil && test.expectedError != nil && err.Error() != test.expectedError.Error()) {
+				t.Errorf("Expected error '%v', but got '%v'", test.expectedError, err)
+			}
+		})
+	}
+}