Update Block uploader to support increase restore (#10244)

Update Block uploader to support increase restore

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

pkg/datapath/data_path.go

@@ -245,7 +245,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
 			dp.wgDataPath.Done()
 		}()
 
-		totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental, target.VolMode, uploaderConfigs, dp)
+		totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental, provider.CBTParam{}, 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, 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, mock.Anything).Return(test.result.Restore.TotalBytes, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			dp.uploaderProv = mockProvider
 			dp.initialized = true

pkg/uploader/block/snapshot.go

@@ -198,18 +198,44 @@ func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull
 }
 
 // Restore restore specific sourcePath with given snapshotID and update progress
-func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, error) {
+func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, error) {
 	log.Info("Start to restore...")
 
 	snapshot, err := rep.GetSnapshot(ctx, udmrepo.ID(snapshotID))
 	if err != nil {
 		return 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
 	}
+	log.Infof("Restore from snapshot %s, incremental %v, cbt source %v, description %s, created time %v, tags %v", snapshotID, incremental, cbtSource, snapshot.Description, snapshot.EndTime, snapshot.Tags)
 
-	log.Infof("Restore from snapshot %s, description %s, created time %v, tags %v", snapshotID, snapshot.Description, snapshot.EndTime, snapshot.Tags)
+	var volumeSnapshot, changeID, volumeID string
+	if incremental {
+		if snapshot.Tags == nil {
+			log.Warnf("No tag from snapshot %s, fallback to full restore", snapshotID)
+			incremental = false
+		} else if snapshot.Tags[uploader.CBTChangeIDTag] == "" {
+			log.Warnf("No ChangeID tag from snapshot %s, fallback to full restore", snapshotID)
+			incremental = false
+		} else if snapshot.Tags[uploader.CBTVolumeIDTag] == "" {
+			log.Warnf("No VolumeID tag from snapshot %s, fallback to full restore", snapshotID)
+			incremental = false
+		} else if snapshot.Tags[uploader.CBTVolumeIDTag] != cbtSource.VolumeID {
+			log.Warnf("VolumeID %s from snapshot %s is not expected as %s, fallback to full restore", snapshot.Tags[uploader.CBTVolumeIDTag], snapshotID, cbtSource.VolumeID)
+			incremental = false
+		} else {
+			volumeSnapshot = cbtSource.Snapshot
+			changeID = snapshot.Tags[uploader.CBTChangeIDTag]
+			volumeID = snapshot.Tags[uploader.CBTVolumeIDTag]
+		}
+	}
 
-	bitmap := cbt.NewBitmap(blockSize, uint64(snapshot.TotalSize), "", "", "")
-	bitmap.SetFull()
+	bitmap := cbt.NewBitmap(blockSize, uint64(snapshot.TotalSize), volumeSnapshot, changeID, volumeID)
+	if incremental {
+		if err = cbt.SetBitmapOrFull(ctx, cbtService, bitmap); err != nil {
+			log.WithError(err).Warnf("Failed to create CBT with source %v, fallback to full restore", cbtSource)
+		}
+	} else {
+		bitmap.SetFull()
+	}
 
 	destPath, err := filepath.Abs(dest)
 	if err != nil {

pkg/uploader/block/snapshot_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/vmware-tanzu/velero/pkg/cbtservice"
+	cbtservicemocks "github.com/vmware-tanzu/velero/pkg/cbtservice/mocks"
 	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
 	udmrepomocks "github.com/vmware-tanzu/velero/pkg/repository/udmrepo/mocks"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
@@ -121,6 +122,23 @@ func TestBackup(t *testing.T) {
 				assert.Positive(t, info.Size)
 			},
 		},
+		{
+			name: "success with CBT",
+			setupOpenDev: func(t *testing.T) *os.File {
+				t.Helper()
+				return tempFile(t, "test-block-data")
+			},
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				blkup.On("Backup", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+					Return(udmrepo.Snapshot{RootObject: udmrepo.ObjectMetadata{ID: "root"}}, int64(8), nil)
+				repo.On("SaveSnapshot", mock.Anything, mock.Anything).Return(udmrepo.ID("snap-001"), nil)
+				repo.On("Flush", mock.Anything).Return(nil)
+			},
+			checkInfo: func(t *testing.T, info uploader.SnapshotInfo) {
+				t.Helper()
+				assert.Equal(t, "snap-001", info.ID)
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -184,6 +202,7 @@ func TestSnapshotSource(t *testing.T) {
 		expectedErrStr string
 		expectedSnapID string
 		expectedSize   int64
+		cbtService     func(t *testing.T) cbtservice.Service
 	}{
 		{
 			name: "uploader Backup error",
@@ -216,7 +235,10 @@ func TestSnapshotSource(t *testing.T) {
 		{
 			name: "success with nil cbtService falls back to full bitmap",
 			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
-				blkup.On("Backup", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+				blkup.On("Backup", mock.Anything, mock.Anything, mock.MatchedBy(func(iter cbttypes.Iterator) bool {
+					// In full mode, the iterator should cover the whole range if it's a full backup
+					return iter != nil
+				}), mock.Anything).
 					Return(udmrepo.Snapshot{RootObject: udmrepo.ObjectMetadata{ID: "root"}}, int64(512), nil)
 				repo.On("SaveSnapshot", mock.Anything, mock.Anything).Return(udmrepo.ID("snap-success"), nil)
 				repo.On("Flush", mock.Anything).Return(nil)
@@ -239,6 +261,46 @@ func TestSnapshotSource(t *testing.T) {
 			},
 			expectedSnapID: "snap-tags",
 		},
+		{
+			name: "success with cbtService getting allocated blocks",
+			cbtService: func(t *testing.T) cbtservice.Service {
+				t.Helper()
+				m := cbtservicemocks.NewService(t)
+				m.On("GetAllocatedBlocks", mock.Anything, "snap-1", mock.Anything).
+					Run(func(args mock.Arguments) {
+						record := args.Get(2).(func([]cbtservice.Range) error)
+						record([]cbtservice.Range{{Offset: 0, Length: 1024}})
+					}).Return(nil)
+				return m
+			},
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				blkup.On("Backup", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+					Return(udmrepo.Snapshot{RootObject: udmrepo.ObjectMetadata{ID: "root"}}, int64(1024), nil)
+				repo.On("SaveSnapshot", mock.Anything, mock.Anything).Return(udmrepo.ID("snap-cbt-alloc"), nil)
+				repo.On("Flush", mock.Anything).Return(nil)
+			},
+			expectedSnapID: "snap-cbt-alloc",
+			expectedSize:   1024,
+		},
+		{
+			name: "cbtService error falls back to full",
+			cbtService: func(t *testing.T) cbtservice.Service {
+				t.Helper()
+				m := cbtservicemocks.NewService(t)
+				m.On("GetAllocatedBlocks", mock.Anything, "snap-1", mock.Anything).
+					Return(errors.New("CBT error"))
+				return m
+			},
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				// Should be called with parentObject as empty because of fallback
+				blkup.On("Backup", mock.Anything, udmrepo.ID(""), mock.Anything, mock.Anything).
+					Return(udmrepo.Snapshot{}, int64(2048), nil)
+				repo.On("SaveSnapshot", mock.Anything, mock.Anything).Return(udmrepo.ID("snap-cbt-fallback"), nil)
+				repo.On("Flush", mock.Anything).Return(nil)
+			},
+			expectedSnapID: "snap-cbt-fallback",
+			expectedSize:   2048,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -249,14 +311,19 @@ func TestSnapshotSource(t *testing.T) {
 
 			tc.setupMocks(mockBlkup, mockRepo)
 
-			cbtSrc := cbtservice.SourceInfo{ChangeID: "cid-1", VolumeID: "vid-1"}
+			cbtSrc := cbtservice.SourceInfo{Snapshot: "snap-1", ChangeID: "cid-1", VolumeID: "vid-1"}
 			snapshotTags := map[string]string{"custom": "val"}
 
+			var cbtSvc cbtservice.Service
+			if tc.cbtService != nil {
+				cbtSvc = tc.cbtService(t)
+			}
+
 			snapID, size, err := snapshotSource(
 				ctx, mockRepo, mockBlkup,
 				baseSource,
 				true, "",
-				cbtSrc, nil,
+				cbtSrc, cbtSvc,
 				snapshotTags, map[string]string{},
 				testLog(), "Block Uploader",
 			)
@@ -547,6 +614,9 @@ func TestRestore(t *testing.T) {
 
 	testCases := []struct {
 		name           string
+		incremental    bool
+		cbtSource      cbtservice.SourceInfo
+		cbtService     func(t *testing.T) cbtservice.Service
 		setupMocks     func(blkup *mockUploader, repo *udmrepomocks.BackupRepo)
 		setupOpenDev   func(t *testing.T) *os.File
 		expectedErrStr string
@@ -583,7 +653,7 @@ func TestRestore(t *testing.T) {
 			expectedErrStr: "error restoring to block dev",
 		},
 		{
-			name: "success returns size",
+			name: "success returns size (full restore)",
 			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
 				repo.On("GetSnapshot", mock.Anything, udmrepo.ID("snap-001")).
 					Return(storedSnap, nil)
@@ -596,6 +666,102 @@ func TestRestore(t *testing.T) {
 			},
 			expectedSize: 4096,
 		},
+		{
+			name:        "incremental restore success",
+			incremental: true,
+			cbtSource:   cbtservice.SourceInfo{Snapshot: "snap-cbt", VolumeID: "vol-1"},
+			cbtService: func(t *testing.T) cbtservice.Service {
+				t.Helper()
+				m := cbtservicemocks.NewService(t)
+				m.On("GetChangedBlocks", mock.Anything, "snap-cbt", "cid-1", mock.Anything).
+					Run(func(args mock.Arguments) {
+						record := args.Get(3).(func([]cbtservice.Range) error)
+						record([]cbtservice.Range{{Offset: 0, Length: 512}})
+					}).Return(nil)
+				return m
+			},
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				snapWithTags := udmrepo.Snapshot{
+					Tags: map[string]string{
+						uploader.CBTChangeIDTag: "cid-1",
+						uploader.CBTVolumeIDTag: "vol-1",
+					},
+					TotalSize: 1024,
+				}
+				repo.On("GetSnapshot", mock.Anything, udmrepo.ID("snap-001")).Return(snapWithTags, nil)
+				blkup.On("Restore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+					Return(int64(512), int64(512), nil)
+			},
+			setupOpenDev: func(t *testing.T) *os.File {
+				t.Helper()
+				return tempFile(t, "")
+			},
+			expectedSize: 512,
+		},
+		{
+			name:        "incremental restore fallback - missing tags",
+			incremental: true,
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				repo.On("GetSnapshot", mock.Anything, udmrepo.ID("snap-001")).Return(storedSnap, nil)
+				blkup.On("Restore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+					Return(int64(4096), int64(4096), nil)
+			},
+			setupOpenDev: func(t *testing.T) *os.File {
+				t.Helper()
+				return tempFile(t, "")
+			},
+			expectedSize: 4096,
+		},
+		{
+			name:        "incremental restore fallback - VolumeID mismatch",
+			incremental: true,
+			cbtSource:   cbtservice.SourceInfo{VolumeID: "vol-actual"},
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				snapWithTags := udmrepo.Snapshot{
+					Tags: map[string]string{
+						uploader.CBTChangeIDTag: "cid-1",
+						uploader.CBTVolumeIDTag: "vol-expected",
+					},
+				}
+				repo.On("GetSnapshot", mock.Anything, udmrepo.ID("snap-001")).Return(snapWithTags, nil)
+				blkup.On("Restore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+					Return(int64(4096), int64(4096), nil)
+			},
+			setupOpenDev: func(t *testing.T) *os.File {
+				t.Helper()
+				return tempFile(t, "")
+			},
+			expectedSize: 4096,
+		},
+		{
+			name:        "incremental restore fallback - CBT service error",
+			incremental: true,
+			cbtSource:   cbtservice.SourceInfo{Snapshot: "snap-cbt", VolumeID: "vol-1"},
+			cbtService: func(t *testing.T) cbtservice.Service {
+				t.Helper()
+				m := cbtservicemocks.NewService(t)
+				m.On("GetChangedBlocks", mock.Anything, "snap-cbt", "cid-1", mock.Anything).
+					Return(errors.New("CBT error"))
+				return m
+			},
+			setupMocks: func(blkup *mockUploader, repo *udmrepomocks.BackupRepo) {
+				snapWithTags := udmrepo.Snapshot{
+					Tags: map[string]string{
+						uploader.CBTChangeIDTag: "cid-1",
+						uploader.CBTVolumeIDTag: "vol-1",
+					},
+					TotalSize: 1024,
+				}
+				repo.On("GetSnapshot", mock.Anything, udmrepo.ID("snap-001")).Return(snapWithTags, nil)
+				blkup.On("Restore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+					Return(int64(1024), int64(1024), nil)
+			},
+			setupOpenDev: func(t *testing.T) *os.File {
+				t.Helper()
+				return tempFile(t, "")
+			},
+			expectedSize: 1024,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -617,7 +783,12 @@ func TestRestore(t *testing.T) {
 				}
 			}
 
-			size, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", map[string]string{}, testLog())
+			var cbtSvc cbtservice.Service
+			if tc.cbtService != nil {
+				cbtSvc = tc.cbtService(t)
+			}
+
+			size, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog())
 
 			if tc.expectedErrStr != "" {
 				require.Error(t, err)

pkg/uploader/provider/block.go

@@ -160,6 +160,7 @@ func (bp *blockProvider) RunRestore(
 	snapshotID string,
 	volumePath string,
 	incremental bool,
+	cbtParam CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) (int64, error) {
@@ -175,7 +176,7 @@ func (bp *blockProvider) RunRestore(
 
 	blkUploader := block.NewUploader(ctx, bp.bkRepo, updater, log)
 
-	size, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, uploaderCfg, log)
+	size, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log)
 
 	if err == block.ErrCanceled {
 		log.Warn("Block restore is canceled")

pkg/uploader/provider/block_test.go

@@ -439,9 +439,9 @@ func TestBlockProviderRunRestore(t *testing.T) {
 			var capturedSnapshotID string
 			var capturedVolumePath string
 
-			blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, snapshotID string, volumePath string, _ map[string]string, _ logrus.FieldLogger) (int64, error) {
+			blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, error) {
 				capturedSnapshotID = snapshotID
-				capturedVolumePath = volumePath
+				capturedVolumePath = dest
 				return tc.mockRestoreSize, tc.mockRestoreErr
 			}
 
@@ -455,6 +455,7 @@ func TestBlockProviderRunRestore(t *testing.T) {
 				tc.snapshotID,
 				tc.volumePath,
 				false,
+				CBTParam{},
 				uploader.PersistentVolumeBlock,
 				map[string]string{},
 				tc.updater,

pkg/uploader/provider/kopia.go

@@ -212,6 +212,7 @@ func (kp *kopiaProvider) RunRestore(
 	snapshotID string,
 	volumePath string,
 	incremental bool,
+	_ CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) (int64, error) {

pkg/uploader/provider/kopia_test.go

@@ -158,7 +158,7 @@ func TestRunRestore(t *testing.T) {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
 			kopiaRestoreFunc = tc.hookRestoreFunc
-			_, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, tc.volMode, map[string]string{}, &updater)
+			_, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, 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, incremental bool, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) {
-	ret := _mock.Called(ctx, snapshotID, volumePath, incremental, volMode, uploaderConfig, updater)
+func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) {
+	ret := _mock.Called(ctx, snapshotID, volumePath, incremental, cbtParam, 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, 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, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok {
+		return returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, 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)
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
+		r0 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
 	} else {
 		r0 = ret.Get(0).(int64)
 	}
-	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)
+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r1 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -258,14 +258,15 @@ type Provider_RunRestore_Call struct {
 //   - snapshotID string
 //   - volumePath string
 //   - incremental bool
+//   - cbtParam provider.CBTParam
 //   - volMode uploader.PersistentVolumeMode
 //   - uploaderConfig map[string]string
 //   - updater uploader.ProgressUpdater
-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 (_e *Provider_Expecter) RunRestore(ctx interface{}, snapshotID interface{}, volumePath interface{}, incremental interface{}, cbtParam interface{}, volMode interface{}, uploaderConfig interface{}, updater interface{}) *Provider_RunRestore_Call {
+	return &Provider_RunRestore_Call{Call: _e.mock.On("RunRestore", ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)}
 }
 
-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 {
+func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, 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 {
@@ -283,17 +284,21 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID
 		if args[3] != nil {
 			arg3 = args[3].(bool)
 		}
-		var arg4 uploader.PersistentVolumeMode
+		var arg4 provider.CBTParam
 		if args[4] != nil {
-			arg4 = args[4].(uploader.PersistentVolumeMode)
+			arg4 = args[4].(provider.CBTParam)
 		}
-		var arg5 map[string]string
+		var arg5 uploader.PersistentVolumeMode
 		if args[5] != nil {
-			arg5 = args[5].(map[string]string)
+			arg5 = args[5].(uploader.PersistentVolumeMode)
 		}
-		var arg6 uploader.ProgressUpdater
+		var arg6 map[string]string
 		if args[6] != nil {
-			arg6 = args[6].(uploader.ProgressUpdater)
+			arg6 = args[6].(map[string]string)
+		}
+		var arg7 uploader.ProgressUpdater
+		if args[7] != nil {
+			arg7 = args[7].(uploader.ProgressUpdater)
 		}
 		run(
 			arg0,
@@ -303,6 +308,7 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID
 			arg4,
 			arg5,
 			arg6,
+			arg7,
 		)
 	})
 	return _c
@@ -313,7 +319,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, incremental bool, 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, cbtParam provider.CBTParam, 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

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