Allow sparse option for Kopia & Restic restore

Signed-off-by: Ming Qiu <mqiu@vmware.com>

changelogs/unreleased/7141-qiuming-best

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

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

@@ -35,6 +35,14 @@ spec:
           spec:
             description: BackupSpec defines the specification for a Velero backup.
             properties:
+              backupConfig:
+                description: BackupConfig defines the configuration for the backup.
+                properties:
+                  parallelFilesUpload:
+                    description: ParallelFilesUpload is the number of files parallel
+                      uploads to perform when using the uploader.
+                    type: integer
+                type: object
               csiSnapshotTimeout:
                 description: CSISnapshotTimeout specifies the time used to wait for
                   CSI VolumeSnapshot status turns to ReadyToUse during creation, before
@@ -477,15 +485,6 @@ spec:
                 description: TTL is a time.Duration-parseable string describing how
                   long the Backup should be retained for.
                 type: string
-              uploaderConfig:
-                description: UploaderConfig specifies the configuration for the uploader.
-                nullable: true
-                properties:
-                  parallelFilesUpload:
-                    description: ParallelFilesUpload is the number of files parallel
-                      uploads to perform when using the uploader.
-                    type: integer
-                type: object
               volumeSnapshotLocations:
                 description: VolumeSnapshotLocations is a list containing names of
                   VolumeSnapshotLocations associated with this backup.

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

@@ -121,13 +121,11 @@ 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.
                 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,12 @@ 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.
+                type: object
               uploaderType:
                 description: UploaderType is the type of the uploader to handle the
                   data transfer.

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

@@ -387,6 +387,14 @@ spec:
                 - name
                 type: object
                 x-kubernetes-map-type: atomic
+              restoreConfig:
+                description: RestoreConfig specifies the configuration for the restore.
+                properties:
+                  writeSparseFiles:
+                    description: WriteSparseFiles is a flag to indicate whether write
+                      files sparsely or not.
+                    type: boolean
+                type: object
               restorePVs:
                 description: RestorePVs specifies whether to restore all included
                   PVs from snapshot

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

@@ -65,6 +65,14 @@ spec:
                 description: Template is the definition of the Backup to be run on
                   the provided schedule
                 properties:
+                  backupConfig:
+                    description: BackupConfig defines the configuration for the backup.
+                    properties:
+                      parallelFilesUpload:
+                        description: ParallelFilesUpload is the number of files parallel
+                          uploads to perform when using the uploader.
+                        type: integer
+                    type: object
                   csiSnapshotTimeout:
                     description: CSISnapshotTimeout specifies the time used to wait
                       for CSI VolumeSnapshot status turns to ReadyToUse during creation,
@@ -514,16 +522,6 @@ spec:
                     description: TTL is a time.Duration-parseable string describing
                       how long the Backup should be retained for.
                     type: string
-                  uploaderConfig:
-                    description: UploaderConfig specifies the configuration for the
-                      uploader.
-                    nullable: true
-                    properties:
-                      parallelFilesUpload:
-                        description: ParallelFilesUpload is the number of files parallel
-                          uploads to perform when using the uploader.
-                        type: integer
-                    type: object
                   volumeSnapshotLocations:
                     description: VolumeSnapshotLocations is a list containing names
                       of VolumeSnapshotLocations associated with this backup.

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

design/velero-uploader-configuration.md

@@ -25,7 +25,7 @@ type UploaderConfig struct {
 }
 ```
 
-### Integration with Backup CRD
+### Integration with Backup & Restore CRD
 The Velero CLI will support an uploader configuration-related flag, allowing users to set the value when creating backups or restores. This value will be stored in the `UploaderConfig` field within the `Backup` CRD and `Restore` CRD:
 
 ```go
@@ -122,5 +122,44 @@ Roughly, the process is as follows:
 3. Each respective controller within the CRs calls the uploader, and the ParallelFilesUpload from UploaderConfig in CRs is passed to the uploader. 
 4. When the uploader subsequently calls the Kopia API, it can use the ParallelFilesUpload to set the MaxParallelFileReads parameter, and if the uploader calls the Restic command it would output one warning log for Restic does not support this feature.
 
+### Sparse Option For Kopia & Restic Restore
+In many system files, there are numerous zero bytes or empty blocks that still occupy physical storage space. Sparse backup employs a more intelligent approach by only backing up the actual data-containing portions. For those empty blocks or zero bytes, it merely records their presence without actually storing them. This can significantly reduce the storage space required for backups, especially in situations where there is a substantial amount of empty data in large file systems.
+
+Below are the key steps that should be added to support this new feature.
+#### Velero CLI 
+The Velero CLI will support a `--write-sparse-files` flag, allowing users to set the `WriteSparseFiles` value when creating restores with Restic or Kopia uploader.
+
+#### UploaderConfig 
+below the sub-option `WriteSparseFiles` is added into UploaderConfig:
+
+```go
+type UploaderConfig struct {
+    // +optional
+	WriteSparseFiles bool `json:"writeSparseFiles,omitempty"`
+}
+```
+
+### Enable Sparse in Restic
+For Restic, it could be enabled by pass the flag `--sparse` in creating restore:
+
+```bash
+restic restore create --sparse $snapshotID
+```
+
+### Enable Sparse in Kopia
+For Kopia, it could be enabled this feature by the `WriteSparseFiles` field in the [FilesystemOutput](https://pkg.go.dev/github.com/kopia/kopia@v0.13.0/snapshot/restore#FilesystemOutput).
+
+```golang
+fsOutput := &restore.FilesystemOutput{
+		WriteSparseFiles:       veleroCfg.WriteSparseFiles,
+	}
+```
+
+Roughly, the process is as follows: 
+1. Users pass the WriteSparseFiles parameter and its value through the Velero CLI. This parameter and its value are stored as a sub-option within UploaderConfig and then placed into the Restore CR. 
+2. When users perform file system restores, UploaderConfig is passed to the PodVolumeRestore CR. When users use the Data-mover for restores, it is passed to the DataDownload CR. 
+3. Each respective controller within the CRs calls the uploader, and the WriteSparseFiles from UploaderConfig in CRs is passed to the uploader. 
+4. When the uploader subsequently calls the Kopia API, it can use the WriteSparseFiles to set the WriteSparseFiles parameter, and if the uploader calls the Restic command it would append `--sparse` flag within the restore command.
+
 ## Alternatives Considered
 To enhance extensibility further, the option of storing `UploaderConfig` in a Kubernetes ConfigMap can be explored, this approach would allow the addition and modification of configuration options without the need to modify the CRD.

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 {
@@ -178,10 +176,16 @@ type BackupSpec struct {
 	// +optional
 	DataMover string `json:"datamover,omitempty"`
 
-	// UploaderConfig specifies the configuration for the uploader.
+	// BackupConfig defines the configuration for the backup.
 	// +optional
-	// +nullable
-	UploaderConfig shared.UploaderConfig `json:"uploaderConfig,omitempty"`
+	BackupConfig *BackupConfig `json:"backupConfig,omitempty"`
+}
+
+// BackupConfig defines the configuration for the backup.
+type BackupConfig 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,10 @@ 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
+	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,11 @@ 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
+	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,17 @@ type RestoreSpec struct {
 	// +optional
 	// +nullable
 	ResourceModifier *v1.TypedLocalObjectReference `json:"resourceModifier,omitempty"`
+
+	// RestoreConfig specifies the configuration for the restore.
+	// +optional
+	RestoreConfig *RestoreConfig `json:"restoreConfig,omitempty"`
+}
+
+// RestoreConfig defines the configuration for the restore.
+type RestoreConfig struct {
+	// WriteSparseFiles is a flag to indicate whether write files sparsely or not.
+	// +optional
+	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

@@ -38,6 +38,21 @@ func (in *Backup) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BackupConfig) DeepCopyInto(out *BackupConfig) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupConfig.
+func (in *BackupConfig) DeepCopy() *BackupConfig {
+	if in == nil {
+		return nil
+	}
+	out := new(BackupConfig)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *BackupHooks) DeepCopyInto(out *BackupHooks) {
 	*out = *in
@@ -381,7 +396,11 @@ func (in *BackupSpec) DeepCopyInto(out *BackupSpec) {
 		*out = new(bool)
 		**out = **in
 	}
-	out.UploaderConfig = in.UploaderConfig
+	if in.BackupConfig != nil {
+		in, out := &in.BackupConfig, &out.BackupConfig
+		*out = new(BackupConfig)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BackupSpec.
@@ -952,7 +971,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.
@@ -994,7 +1019,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)
 }
 
@@ -1052,6 +1077,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.
@@ -1115,6 +1147,21 @@ func (in *Restore) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *RestoreConfig) DeepCopyInto(out *RestoreConfig) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RestoreConfig.
+func (in *RestoreConfig) DeepCopy() *RestoreConfig {
+	if in == nil {
+		return nil
+	}
+	out := new(RestoreConfig)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *RestoreHooks) DeepCopyInto(out *RestoreHooks) {
 	*out = *in
@@ -1329,6 +1376,11 @@ func (in *RestoreSpec) DeepCopyInto(out *RestoreSpec) {
 		*out = new(corev1.TypedLocalObjectReference)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.RestoreConfig != nil {
+		in, out := &in.RestoreConfig, &out.RestoreConfig
+		*out = new(RestoreConfig)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RestoreSpec.

pkg/apis/velero/v2alpha1/data_upload_types.go

@@ -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

@@ -236,7 +236,6 @@ func (in *DataUploadSpec) DeepCopyInto(out *DataUploadSpec) {
 		}
 	}
 	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,7 +302,10 @@ func (b *BackupBuilder) DataMover(name string) *BackupBuilder {
 
 // ParallelFilesUpload sets the Backup's uploader parallel uploads
 func (b *BackupBuilder) ParallelFilesUpload(parallel int) *BackupBuilder {
-	b.object.Spec.UploaderConfig.ParallelFilesUpload = parallel
+	if b.object.Spec.BackupConfig == nil {
+		b.object.Spec.BackupConfig = &velerov1api.BackupConfig{}
+	}
+	b.object.Spec.BackupConfig.ParallelFilesUpload = parallel
 	return b
 }
 

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.RestoreConfig.WriteSparseFiles = val
+	return b
+}

pkg/cmd/cli/nodeagent/server.go

@@ -194,6 +194,7 @@ func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, confi
 		Scheme:   scheme,
 		NewCache: cache.BuilderWithOptions(cacheOption),
 	})
+
 	if err != nil {
 		cancelFunc()
 		return nil, err

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 when restoring volumes using restic or kopia")
+	f.NoOptDefVal = cmd.FALSE
 }
 
 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,
 			},
+			RestoreConfig: &api.RestoreConfig{
+				WriteSparseFiles: boolptr.IsSetToTrue(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 := "false"
 
 		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/const.go

@@ -1,3 +1,4 @@
 package cmd
 
 var TRUE string = "true"
+var FALSE string = "false"

pkg/cmd/util/output/backup_describer.go

@@ -88,7 +88,7 @@ func DescribeBackup(
 			DescribeResourcePolicies(d, backup.Spec.ResourcePolicy)
 		}
 
-		if backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
+		if backup.Spec.BackupConfig != nil && backup.Spec.BackupConfig.ParallelFilesUpload > 0 {
 			d.Println()
 			DescribeUploaderConfig(d, backup.Spec)
 		}
@@ -138,7 +138,7 @@ func DescribeResourcePolicies(d *Describer, resPolicies *v1.TypedLocalObjectRefe
 // DescribeUploaderConfig describes uploader config in human-readable format
 func DescribeUploaderConfig(d *Describer, spec velerov1api.BackupSpec) {
 	d.Printf("Uploader config:\n")
-	d.Printf("\tParallel files upload:\t%d\n", spec.UploaderConfig.ParallelFilesUpload)
+	d.Printf("\tParallel files upload:\t%d\n", spec.BackupConfig.ParallelFilesUpload)
 }
 
 // DescribeBackupSpec describes a backup spec in human-readable format.

pkg/cmd/util/output/restore_describer.go

@@ -177,6 +177,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.RestoreConfig != nil && restore.Spec.RestoreConfig.WriteSparseFiles {
+			d.Println()
+			d.Printf("Write Sparse Files:\t%T\n", restore.Spec.RestoreConfig.WriteSparseFiles)
+		}
+
 		d.Println()
 		describeRestoreItemOperations(ctx, kbClient, d, restore, details, insecureSkipTLSVerify, caCertFile)
 

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.ParallelFilesUpload > 0 {
+		if schedule.Spec.Template.BackupConfig != nil && schedule.Spec.Template.BackupConfig.ParallelFilesUpload > 0 {
 			d.Println()
 			DescribeUploaderConfig(d, schedule.Spec.Template)
 		}

pkg/controller/data_download_controller.go

@@ -335,7 +335,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

@@ -344,7 +344,7 @@ 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"
@@ -284,7 +283,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
@@ -294,7 +293,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, make(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}, make(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, uploaderConfigs
+func (_m *AsyncBR) StartRestore(snapshotID string, target datapath.AccessPoint, uploaderConfigs map[string]string) error {
+	ret := _m.Called(snapshotID, target, uploaderConfigs)
 
 	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, uploaderConfigs)
 	} 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, uploaderConfigs map[string]string) error
 
 	// Cancel cancels an asynchronous data path instance
 	Cancel()

pkg/podvolume/backupper.go

@@ -38,6 +38,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
+	"github.com/vmware-tanzu/velero/pkg/util/uploaderconfig"
 )
 
 // Backupper can execute pod volume backups of volumes in a pod.
@@ -300,7 +301,11 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
 			}
 		}
 
-		volumeBackup := newPodVolumeBackup(backup, pod, volume, repoIdentifier, b.uploaderType, pvc)
+		volumeBackup, err := newPodVolumeBackup(backup, pod, volume, repoIdentifier, b.uploaderType, pvc)
+		if err != nil {
+			errs = append(errs, errors.Wrapf(err, "error creating PodVolumeBackup for volume %s", volumeName))
+			continue
+		}
 		if err := veleroclient.CreateRetryGenerateName(b.crClient, b.ctx, volumeBackup); err != nil {
 			errs = append(errs, err)
 			continue
@@ -353,7 +358,7 @@ func isHostPathVolume(volume *corev1api.Volume, pvc *corev1api.PersistentVolumeC
 	return pv.Spec.HostPath != nil, nil
 }
 
-func newPodVolumeBackup(backup *velerov1api.Backup, pod *corev1api.Pod, volume corev1api.Volume, repoIdentifier, uploaderType string, pvc *corev1api.PersistentVolumeClaim) *velerov1api.PodVolumeBackup {
+func newPodVolumeBackup(backup *velerov1api.Backup, pod *corev1api.Pod, volume corev1api.Volume, repoIdentifier, uploaderType string, pvc *corev1api.PersistentVolumeClaim) (*velerov1api.PodVolumeBackup, error) {
 	pvb := &velerov1api.PodVolumeBackup{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:    backup.Namespace,
@@ -392,7 +397,6 @@ func newPodVolumeBackup(backup *velerov1api.Backup, pod *corev1api.Pod, volume c
 			BackupStorageLocation: backup.Spec.StorageLocation,
 			RepoIdentifier:        repoIdentifier,
 			UploaderType:          uploaderType,
-			UploaderConfig:        backup.Spec.UploaderConfig,
 		},
 	}
 
@@ -411,5 +415,15 @@ func newPodVolumeBackup(backup *velerov1api.Backup, pod *corev1api.Pod, volume c
 		pvb.Spec.Tags["pvc-uid"] = string(pvc.UID)
 	}
 
-	return pvb
+	if backup.Spec.BackupConfig != nil {
+		configJSON, err := uploaderconfig.MarshalToPVBConfig(backup.Spec.BackupConfig)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to marshal backup config")
+		}
+		pvb.Spec.UploaderSettings = map[string]string{
+			uploaderconfig.PodVolumeBackups: configJSON,
+		}
+	}
+
+	return pvb, nil
 }

pkg/podvolume/restorer.go

@@ -38,6 +38,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
+	"github.com/vmware-tanzu/velero/pkg/util/uploaderconfig"
 )
 
 type RestoreData struct {
@@ -176,7 +177,11 @@ func (r *restorer) RestorePodVolumes(data RestoreData) []error {
 			}
 		}
 
-		volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, repoIdentifier, backupInfo.uploaderType, data.SourceNamespace, pvc)
+		volumeRestore, err := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, repoIdentifier, backupInfo.uploaderType, data.SourceNamespace, pvc)
+		if err != nil {
+			errs = append(errs, errors.Wrapf(err, "error creating PodVolumeRestore for volume %s", volume))
+			continue
+		}
 
 		if err := veleroclient.CreateRetryGenerateName(r.crClient, r.ctx, volumeRestore); err != nil {
 			errs = append(errs, errors.WithStack(err))
@@ -246,7 +251,7 @@ ForEachVolume:
 	return errs
 }
 
-func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, backupLocation, volume, snapshot, repoIdentifier, uploaderType, sourceNamespace string, pvc *corev1api.PersistentVolumeClaim) *velerov1api.PodVolumeRestore {
+func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, backupLocation, volume, snapshot, repoIdentifier, uploaderType, sourceNamespace string, pvc *corev1api.PersistentVolumeClaim) (*velerov1api.PodVolumeRestore, error) {
 	pvr := &velerov1api.PodVolumeRestore{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace:    restore.Namespace,
@@ -285,7 +290,19 @@ 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)
 	}
-	return pvr
+
+	if restore.Spec.RestoreConfig != nil {
+		configJSON, err := uploaderconfig.MarshalToPVRConfig(restore.Spec.RestoreConfig)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to marshal restore config")
+		}
+
+		pvr.Spec.UploaderSettings = map[string]string{
+			uploaderconfig.PodVolumeRestores: configJSON,
+		}
+	}
+
+	return pvr, nil
 }
 
 func getVolumesRepositoryType(volumes map[string]volumeBackupInfo) (string, error) {

pkg/uploader/kopia/snapshot.go

@@ -18,6 +18,7 @@ package kopia
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"math"
 	"os"
@@ -38,7 +39,8 @@ import (
 	"github.com/kopia/kopia/snapshot/snapshotfs"
 	"github.com/pkg/errors"
 
-	"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/kopia"
 	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
@@ -105,11 +107,21 @@ 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
+	backupCfg := velerov1api.BackupConfig{}
+	// currently, we only have one uploader config in one uploader config so we can just loop through it
+	for configItem, jsonConfig := range uploaderCfg {
+		err := json.Unmarshal([]byte(jsonConfig), &backupCfg)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to parse %s uploader config", configItem)
+		}
+		break
+	}
+
 	curPolicy := getDefaultPolicy()
-	if uploaderCfg.ParallelFilesUpload > 0 {
-		curPolicy.UploadPolicy.MaxParallelFileReads = newOptionalInt(uploaderCfg.ParallelFilesUpload)
+	if backupCfg.ParallelFilesUpload > 0 {
+		curPolicy.UploadPolicy.MaxParallelFileReads = newOptionalInt(backupCfg.ParallelFilesUpload)
 	}
 
 	err := setPolicyFunc(ctx, rep, sourceInfo, curPolicy)
@@ -133,7 +145,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 +241,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 +373,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...")
 
@@ -384,13 +396,25 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		return 0, 0, errors.Wrapf(err, "Unable to resolve path %v", dest)
 	}
 
+	restoreCfg := velerov1api.RestoreConfig{}
+	// currently, we only have one uploader config in map so we can just loop through it
+	for configItem, jsonConfig := range uploaderCfg {
+		err := json.Unmarshal([]byte(jsonConfig), &restoreCfg)
+		if err != nil {
+			return 0, 0, errors.Wrapf(err, "failed to parse %s uploader config", configItem)
+		}
+		break
+	}
+
 	fsOutput := &restore.FilesystemOutput{
 		TargetPath:             path,
 		OverwriteDirectories:   true,
 		OverwriteFiles:         true,
 		OverwriteSymlinks:      true,
 		IgnorePermissionErrors: true,
+		WriteSparseFiles:       restoreCfg.WriteSparseFiles,
 	}
+	log.Debugf("Restore filesystem output %v", fsOutput)
 
 	err = fsOutput.Init(ctx)
 	if err != nil {

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
 	}{
 		{
@@ -163,7 +162,7 @@ func TestSnapshotSource(t *testing.T) {
 				{methodName: "Upload", returns: []interface{}{manifest, nil}},
 				{methodName: "Flush", returns: []interface{}{nil}},
 			},
-			uploaderCfg: shared.UploaderConfig{ParallelFilesUpload: 10},
+			uploaderCfg: map[string]string{"ParallelFilesUpload": "10"},
 			notError:    true,
 		},
 		{
@@ -646,9 +645,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, make(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, make(map[string]string), tc.tags, &logrus.Logger{})
 			}
 			// Check if the returned error matches the expected error
 			if tc.expectedError != nil {
@@ -787,7 +786,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, make(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

@@ -18,6 +18,7 @@ package provider
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"os"
 	"strings"
@@ -27,7 +28,6 @@ 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"
@@ -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,18 @@ 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)
+	backupCfg := velerov1api.BackupConfig{}
+	// currently, we only have one uploader config in one uploader config so we can just loop through it
+	for configItem, jsonConfig := range uploaderCfg {
+		err := json.Unmarshal([]byte(jsonConfig), &backupCfg)
+		if err != nil {
+			return "", false, errors.Wrapf(err, "failed to parse %s config", configItem)
+		}
+		break
+	}
+
+	if backupCfg.ParallelFilesUpload > 0 {
+		log.Warnf("ParallelFilesUpload is set to %d, but restic does not support parallel file uploads. Ignoring.", backupCfg.ParallelFilesUpload)
 	}
 
 	backupCmd := resticBackupCMDFunc(rp.repoIdentifier, rp.credentialsFile, path, tags)
@@ -191,6 +201,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 +221,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{}
+	restoreCfg := velerov1api.RestoreConfig{}
+	// currently, we only have one uploader config in map so we can just loop through it
+	for configItem, jsonConfig := range uploaderCfg {
+		err := json.Unmarshal([]byte(jsonConfig), &restoreCfg)
+		if err != nil {
+			return extraFlags, errors.Wrapf(err, "failed to parse %s uploader config", configItem)
+		}
+		break
+	}
+
+	if restoreCfg.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,13 +32,13 @@ 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"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
+	"github.com/vmware-tanzu/velero/pkg/util/uploaderconfig"
 )
 
 func TestResticRunBackup(t *testing.T) {
@@ -150,9 +151,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 +224,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 +412,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{
+				uploaderconfig.PodVolumeRestores: `{"WriteSparseFiles": true}`,
+			},
+			expectedFlags: []string{"--sparse"},
+		},
+		{
+			name: "SparseFilesDisabled",
+			uploaderConfig: map[string]string{
+				uploaderconfig.PodVolumeRestores: `{"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/util/uploaderconfig/uploaderconfig.go

@@ -0,0 +1,82 @@
+package uploaderconfig
+
+import (
+	"encoding/json"
+
+	"github.com/pkg/errors"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+)
+
+const (
+	PodVolumeBackups  = "PodVolumeBackups"
+	PodVolumeRestores = "PodVolumeRestores"
+	DataUploads       = "DataUploads"
+	DataDownloads     = "DataDownloads"
+)
+
+type PVBConfig struct {
+	ParallelFilesUpload int `json:"parallelFilesUpload,omitempty"`
+}
+
+type PVRConfig struct {
+	WriteSparseFiles bool `json:"writeSparseFiles,omitempty"`
+}
+
+func MarshalToPVBConfig(backupConfig *velerov1api.BackupConfig) (string, error) {
+	jsonData, err := json.Marshal(backupConfig)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to marshal backup config")
+	}
+
+	var pvb PVBConfig
+	err = json.Unmarshal(jsonData, &pvb)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to unmarshal backup config")
+	}
+
+	finalJSONData, err := json.Marshal(pvb)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to marshal backup config")
+	}
+
+	return string(finalJSONData), nil
+}
+
+func MarshalToPVRConfig(restoreConfig *velerov1api.RestoreConfig) (string, error) {
+	jsonData, err := json.Marshal(restoreConfig)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to marshal restore config")
+	}
+
+	var pvr PVRConfig
+	err = json.Unmarshal(jsonData, &pvr)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to unmarshal restore config")
+	}
+
+	finalJSONData, err := json.Marshal(pvr)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to marshal restore config")
+	}
+
+	return string(finalJSONData), nil
+}
+
+func ParseBackupConfig(str string) (velerov1api.BackupConfig, error) {
+	var config velerov1api.BackupConfig
+	err := json.Unmarshal([]byte(str), &config)
+	if err != nil {
+		return velerov1api.BackupConfig{}, err
+	}
+	return config, nil
+}
+
+func ParseRestoreConfig(str string) (velerov1api.RestoreConfig, error) {
+	var config velerov1api.RestoreConfig
+	err := json.Unmarshal([]byte(str), &config)
+	if err != nil {
+		return velerov1api.RestoreConfig{}, err
+	}
+	return config, nil
+}