Update Kopia(filesystem) uploader to support incremental and deleteExtraFile during restore (#10066)

Update Kopia(filesystem) uploader to support incremental and deleteExtraFile during restore

Signed-off-by: Wenkai Yin(尹文开) <yinw@vmware.com>

pkg/datamover/restore_micro_service.go

@@ -180,7 +180,9 @@ func (r *RestoreMicroService) RunCancelableDataPath(ctx context.Context) (string
 	}
 	log.Info("fs init")
 
-	if err := dp.StartRestore(dd.Spec.SnapshotID, r.sourceTargetPath, dd.Spec.DataMoverConfig, &datapath.RestoreStartParam{}); err != nil {
+	if err := dp.StartRestore(dd.Spec.SnapshotID, r.sourceTargetPath, dd.Spec.DataMoverConfig, &datapath.RestoreStartParam{
+		Incremental: dd.Spec.RestoreType == string(velerov1api.VolumeDataPolicyTypeIncremental),
+	}); err != nil {
 		return "", errors.Wrap(err, "error starting data path restore")
 	}
 

pkg/datapath/data_path.go

@@ -61,6 +61,7 @@ type BackupStartParam struct {
 
 // RestoreStartParam define the input param for restore start
 type RestoreStartParam struct {
+	Incremental bool
 }
 
 type generalDataPath struct {
@@ -232,6 +233,8 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
 
 	dp.wgDataPath.Add(1)
 
+	restoreParam := param.(*RestoreStartParam)
+
 	go func() {
 		dp.log.Info("Start data path restore")
 
@@ -240,7 +243,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
 			dp.wgDataPath.Done()
 		}()
 
-		totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, target.VolMode, uploaderConfigs, dp)
+		totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental, target.VolMode, uploaderConfigs, dp)
 
 		if err == provider.ErrorCanceled {
 			dp.callbacks.OnCancelled(context.Background(), dp.namespace, dp.jobName)

pkg/datapath/data_path_test.go

@@ -184,7 +184,7 @@ func TestAsyncRestore(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			dp := newGeneralDataPath("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*generalDataPath)
 			mockProvider := providerMock.NewProvider(t)
-			mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.TotalBytes, test.err)
+			mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.TotalBytes, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			dp.uploaderProv = mockProvider
 			dp.initialized = true

pkg/podvolume/restore_micro_service.go

@@ -184,7 +184,9 @@ func (r *RestoreMicroService) RunCancelableDataPath(ctx context.Context) (string
 
 	log.Info("Async fs br init")
 
-	if err := fsRestore.StartRestore(pvr.Spec.SnapshotID, r.sourceTargetPath, pvr.Spec.UploaderSettings, &datapath.RestoreStartParam{}); err != nil {
+	if err := fsRestore.StartRestore(pvr.Spec.SnapshotID, r.sourceTargetPath, pvr.Spec.UploaderSettings, &datapath.RestoreStartParam{
+		Incremental: pvr.Spec.RestoreType == string(velerov1api.VolumeDataPolicyTypeIncremental),
+	}); err != nil {
 		return "", errors.Wrap(err, "error starting data path restore")
 	}
 

pkg/uploader/kopia/snapshot.go

@@ -389,7 +389,7 @@ func (o *fileSystemRestoreOutput) Terminate() error {
 }
 
 // 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, uploaderCfg map[string]string,
+func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string,
 	log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 	log.Info("Start to restore...")
 
@@ -421,7 +421,7 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 	}
 
 	restoreConcurrency := runtime.NumCPU()
-
+	deleteExtra := false
 	if len(uploaderCfg) > 0 {
 		writeSparseFiles, err := uploaderutil.GetWriteSparseFiles(uploaderCfg)
 		if err != nil {
@@ -438,9 +438,14 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		if concurrency > 0 {
 			restoreConcurrency = concurrency
 		}
+
+		deleteExtra, err = uploaderutil.GetDeleteExtraFiles(uploaderCfg)
+		if err != nil {
+			return 0, 0, errors.Wrap(err, "failed to get delete extra files config")
+		}
 	}
 
-	log.Debugf("Restore filesystem output %v, concurrency %d", fsOutput, restoreConcurrency)
+	log.Debugf("Restore filesystem output %v, concurrency %d, incremental %v, delete extra %v", fsOutput, restoreConcurrency, incremental, deleteExtra)
 
 	err = fsOutput.Init(ctx)
 	if err != nil {
@@ -448,14 +453,22 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 	}
 
 	var output RestoreOutput
+	// kopiaOutput is the output passed to Kopia's restore.Entry function.
+	// We must pass the unwrapped fsOutput (*restore.FilesystemOutput) directly for file system restores.
+	// This is because Kopia internally uses a strict type assertion (c.output.(*FilesystemOutput))
+	// to determine if it should execute the deleteExtra logic. If we pass the wrapped
+	// fileSystemRestoreOutput, the type assertion fails and extra files are not deleted.
+	var kopiaOutput restore.Output
 	if volMode == uploader.PersistentVolumeBlock {
 		output = &BlockOutput{
 			FilesystemOutput: fsOutput,
 		}
+		kopiaOutput = output
 	} else {
 		output = &fileSystemRestoreOutput{
 			FilesystemOutput: fsOutput,
 		}
+		kopiaOutput = fsOutput
 	}
 
 	defer func() {
@@ -464,8 +477,10 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
 		}
 	}()
 
-	stat, err := restoreEntryFunc(kopiaCtx, rep, output, rootEntry, restore.Options{
+	stat, err := restoreEntryFunc(kopiaCtx, rep, kopiaOutput, rootEntry, restore.Options{
 		Parallel:               restoreConcurrency,
+		Incremental:            incremental,
+		DeleteExtra:            deleteExtra,
 		RestoreDirEntryAtDepth: math.MaxInt32,
 		Cancel:                 cancleCh,
 		ProgressCallback: func(ctx context.Context, stats restore.Stats) {

pkg/uploader/kopia/snapshot_test.go

@@ -681,6 +681,7 @@ func TestRestore(t *testing.T) {
 		expectedCount       int32
 		expectedError       error
 		volMode             uploader.PersistentVolumeMode
+		incremental         bool
 	}
 
 	// Define test cases
@@ -818,7 +819,7 @@ func TestRestore(t *testing.T) {
 			repoWriterMock.On("OpenObject", mock.Anything, mock.Anything).Return(em, nil)
 
 			progress := new(Progress)
-			bytesRestored, fileCount, err := Restore(t.Context(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.volMode, map[string]string{}, logrus.New(), nil)
+			bytesRestored, fileCount, err := Restore(t.Context(), repoWriterMock, progress, tc.snapshotID, tc.dest, tc.incremental, tc.volMode, map[string]string{}, logrus.New(), nil)
 
 			// Check if the returned error matches the expected error
 			if tc.expectedError != nil {

pkg/uploader/provider/block.go

@@ -159,6 +159,7 @@ func (bp *blockProvider) RunRestore(
 	ctx context.Context,
 	snapshotID string,
 	volumePath string,
+	incremental bool,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) (int64, error) {

pkg/uploader/provider/block_test.go

@@ -454,6 +454,7 @@ func TestBlockProviderRunRestore(t *testing.T) {
 				t.Context(),
 				tc.snapshotID,
 				tc.volumePath,
+				false,
 				uploader.PersistentVolumeBlock,
 				map[string]string{},
 				tc.updater,

pkg/uploader/provider/kopia.go

@@ -211,6 +211,7 @@ func (kp *kopiaProvider) RunRestore(
 	ctx context.Context,
 	snapshotID string,
 	volumePath string,
+	incremental bool,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) (int64, error) {
@@ -234,7 +235,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 := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, volMode, uploaderCfg, log, restoreCancel)
+	size, fileCount, err := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, incremental, volMode, uploaderCfg, log, restoreCancel)
 
 	if err != nil {
 		return 0, errors.Wrapf(err, "Failed to run kopia restore")

pkg/uploader/provider/kopia_test.go

@@ -119,20 +119,21 @@ func TestRunBackup(t *testing.T) {
 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, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error)
+		hookRestoreFunc func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error)
 		notError        bool
 		volMode         uploader.PersistentVolumeMode
+		incremental     bool
 	}{
 		{
 			name: "normal restore",
-			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) {
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 				return 0, 0, nil
 			},
 			notError: true,
 		},
 		{
 			name: "normal block mode restore",
-			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) {
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
 				return 0, 0, nil
 			},
 			volMode:  uploader.PersistentVolumeBlock,
@@ -140,7 +141,7 @@ func TestRunRestore(t *testing.T) {
 		},
 		{
 			name: "failed to restore",
-			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) {
+			hookRestoreFunc: func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, 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,
@@ -157,7 +158,7 @@ func TestRunRestore(t *testing.T) {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
 			kopiaRestoreFunc = tc.hookRestoreFunc
-			_, err := kp.RunRestore(t.Context(), "", "/var", tc.volMode, map[string]string{}, &updater)
+			_, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {

pkg/uploader/provider/mocks/Provider.go

@@ -223,8 +223,8 @@ func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, pa
 }
 
 // RunRestore provides a mock function for the type Provider
-func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) {
-	ret := _mock.Called(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
+func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) {
+	ret := _mock.Called(ctx, snapshotID, volumePath, incremental, volMode, uploaderConfig, updater)
 
 	if len(ret) == 0 {
 		panic("no return value specified for RunRestore")
@@ -232,16 +232,16 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume
 
 	var r0 int64
 	var r1 error
-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok {
-		return returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok {
+		return returnFunc(ctx, snapshotID, volumePath, incremental, volMode, uploaderConfig, updater)
 	}
-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
-		r0 = returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
+		r0 = returnFunc(ctx, snapshotID, volumePath, incremental, volMode, uploaderConfig, updater)
 	} else {
 		r0 = ret.Get(0).(int64)
 	}
-	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
-		r1 = returnFunc(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, bool, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r1 = returnFunc(ctx, snapshotID, volumePath, incremental, volMode, uploaderConfig, updater)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -257,14 +257,15 @@ type Provider_RunRestore_Call struct {
 //   - ctx context.Context
 //   - snapshotID string
 //   - volumePath string
+//   - incremental bool
 //   - volMode uploader.PersistentVolumeMode
 //   - uploaderConfig map[string]string
 //   - updater uploader.ProgressUpdater
-func (_e *Provider_Expecter) RunRestore(ctx interface{}, snapshotID interface{}, volumePath interface{}, volMode interface{}, uploaderConfig interface{}, updater interface{}) *Provider_RunRestore_Call {
-	return &Provider_RunRestore_Call{Call: _e.mock.On("RunRestore", ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)}
+func (_e *Provider_Expecter) RunRestore(ctx interface{}, snapshotID interface{}, volumePath interface{}, incremental interface{}, volMode interface{}, uploaderConfig interface{}, updater interface{}) *Provider_RunRestore_Call {
+	return &Provider_RunRestore_Call{Call: _e.mock.On("RunRestore", ctx, snapshotID, volumePath, incremental, volMode, uploaderConfig, updater)}
 }
 
-func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater)) *Provider_RunRestore_Call {
+func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater)) *Provider_RunRestore_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		var arg0 context.Context
 		if args[0] != nil {
@@ -278,17 +279,21 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID
 		if args[2] != nil {
 			arg2 = args[2].(string)
 		}
-		var arg3 uploader.PersistentVolumeMode
+		var arg3 bool
 		if args[3] != nil {
-			arg3 = args[3].(uploader.PersistentVolumeMode)
+			arg3 = args[3].(bool)
 		}
-		var arg4 map[string]string
+		var arg4 uploader.PersistentVolumeMode
 		if args[4] != nil {
-			arg4 = args[4].(map[string]string)
+			arg4 = args[4].(uploader.PersistentVolumeMode)
 		}
-		var arg5 uploader.ProgressUpdater
+		var arg5 map[string]string
 		if args[5] != nil {
-			arg5 = args[5].(uploader.ProgressUpdater)
+			arg5 = args[5].(map[string]string)
+		}
+		var arg6 uploader.ProgressUpdater
+		if args[6] != nil {
+			arg6 = args[6].(uploader.ProgressUpdater)
 		}
 		run(
 			arg0,
@@ -297,6 +302,7 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID
 			arg3,
 			arg4,
 			arg5,
+			arg6,
 		)
 	})
 	return _c
@@ -307,7 +313,7 @@ func (_c *Provider_RunRestore_Call) Return(n int64, err error) *Provider_RunRest
 	return _c
 }
 
-func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error)) *Provider_RunRestore_Call {
+func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error)) *Provider_RunRestore_Call {
 	_c.Call.Return(run)
 	return _c
 }

pkg/uploader/provider/provider.go

@@ -64,6 +64,7 @@ type Provider interface {
 		ctx context.Context,
 		snapshotID string,
 		volumePath string,
+		incremental bool,
 		volMode uploader.PersistentVolumeMode,
 		uploaderConfig map[string]string,
 		updater uploader.ProgressUpdater) (int64, error)

pkg/uploader/util/uploader_config.go

@@ -28,6 +28,7 @@ const (
 	ParallelFilesUpload = "ParallelFilesUpload"
 	WriteSparseFiles    = "WriteSparseFiles"
 	RestoreConcurrency  = "ParallelFilesDownload"
+	DeleteExtraFiles    = "DeleteExtraFiles"
 )
 
 func StoreBackupConfig(config *velerov1api.UploaderConfigForBackup) map[string]string {
@@ -47,6 +48,13 @@ func StoreRestoreConfig(config *velerov1api.UploaderConfigForRestore) map[string
 	if config.ParallelFilesDownload > 0 {
 		data[RestoreConcurrency] = strconv.Itoa(config.ParallelFilesDownload)
 	}
+
+	if config.DeleteExtraFiles != nil {
+		data[DeleteExtraFiles] = strconv.FormatBool(*config.DeleteExtraFiles)
+	} else {
+		data[DeleteExtraFiles] = strconv.FormatBool(false)
+	}
+
 	return data
 }
 
@@ -85,3 +93,15 @@ func GetRestoreConcurrency(uploaderCfg map[string]string) (int, error) {
 	}
 	return 0, nil
 }
+
+func GetDeleteExtraFiles(uploaderCfg map[string]string) (bool, error) {
+	deleteExtraFiles, ok := uploaderCfg[DeleteExtraFiles]
+	if ok {
+		deleteExtraFilesBool, err := strconv.ParseBool(deleteExtraFiles)
+		if err != nil {
+			return false, errors.Wrap(err, "failed to parse DeleteExtraFiles config")
+		}
+		return deleteExtraFilesBool, nil
+	}
+	return false, nil
+}

pkg/uploader/util/uploader_config_test.go

@@ -58,6 +58,7 @@ func TestStoreRestoreConfig(t *testing.T) {
 			},
 			expectedData: map[string]string{
 				WriteSparseFiles: "true",
+				DeleteExtraFiles: "false",
 			},
 		},
 		{
@@ -67,6 +68,7 @@ func TestStoreRestoreConfig(t *testing.T) {
 			},
 			expectedData: map[string]string{
 				WriteSparseFiles: "false",
+				DeleteExtraFiles: "false",
 			},
 		},
 		{
@@ -76,6 +78,7 @@ func TestStoreRestoreConfig(t *testing.T) {
 			},
 			expectedData: map[string]string{
 				WriteSparseFiles: "false", // Assuming default value is false for nil case
+				DeleteExtraFiles: "false",
 			},
 		},
 		{
@@ -86,6 +89,37 @@ func TestStoreRestoreConfig(t *testing.T) {
 			expectedData: map[string]string{
 				RestoreConcurrency: "5",
 				WriteSparseFiles:   "false",
+				DeleteExtraFiles:   "false",
+			},
+		},
+		{
+			name: "DeleteExtraFiles is true",
+			config: &velerov1api.UploaderConfigForRestore{
+				DeleteExtraFiles: &boolTrue,
+			},
+			expectedData: map[string]string{
+				WriteSparseFiles: "false",
+				DeleteExtraFiles: "true",
+			},
+		},
+		{
+			name: "DeleteExtraFiles is false",
+			config: &velerov1api.UploaderConfigForRestore{
+				DeleteExtraFiles: &boolFalse,
+			},
+			expectedData: map[string]string{
+				WriteSparseFiles: "false",
+				DeleteExtraFiles: "false",
+			},
+		},
+		{
+			name: "DeleteExtraFiles is nil",
+			config: &velerov1api.UploaderConfigForRestore{
+				DeleteExtraFiles: nil,
+			},
+			expectedData: map[string]string{
+				WriteSparseFiles: "false",
+				DeleteExtraFiles: "false", // Assuming default value is false for nil case
 			},
 		},
 	}
@@ -240,3 +274,51 @@ func TestGetRestoreConcurrency(t *testing.T) {
 		})
 	}
 }
+
+func TestGetDeleteExtraFiles(t *testing.T) {
+	tests := []struct {
+		name           string
+		uploaderCfg    map[string]string
+		expectedResult bool
+		expectedError  error
+	}{
+		{
+			name:           "Valid DeleteExtraFiles (true)",
+			uploaderCfg:    map[string]string{DeleteExtraFiles: "true"},
+			expectedResult: true,
+			expectedError:  nil,
+		},
+		{
+			name:           "Valid DeleteExtraFiles (false)",
+			uploaderCfg:    map[string]string{DeleteExtraFiles: "false"},
+			expectedResult: false,
+			expectedError:  nil,
+		},
+		{
+			name:           "Invalid DeleteExtraFiles (not a boolean)",
+			uploaderCfg:    map[string]string{DeleteExtraFiles: "invalid"},
+			expectedResult: false,
+			expectedError:  errors.Wrap(errors.New("strconv.ParseBool: parsing \"invalid\": invalid syntax"), "failed to parse DeleteExtraFiles config"),
+		},
+		{
+			name:           "Missing DeleteExtraFiles",
+			uploaderCfg:    map[string]string{},
+			expectedResult: false,
+			expectedError:  nil,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result, err := GetDeleteExtraFiles(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)
+			}
+		})
+	}
+}