modify uploader provider interface for block data mover

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

pkg/datapath/file_system.go

@@ -183,7 +183,7 @@ func (fs *fileSystemBR) StartBackup(source AccessPoint, uploaderConfig map[strin
 		}()
 
 		snapshotID, emptySnapshot, totalBytes, incrementalBytes, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, backupParam.RealSource, backupParam.Tags, backupParam.ForceFull,
-			backupParam.ParentSnapshot, source.VolMode, uploaderConfig, fs)
+			backupParam.ParentSnapshot, provider.CBTParam{}, source.VolMode, uploaderConfig, fs)
 
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)

pkg/datapath/file_system_test.go

@@ -96,7 +96,7 @@ func TestAsyncBackup(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("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.result.Backup.IncrementalBytes, test.err)
+			mockProvider.On("RunBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Backup.SnapshotID, test.result.Backup.EmptySnapshot, test.result.Backup.TotalBytes, test.result.Backup.IncrementalBytes, test.err)
 			mockProvider.On("Close", mock.Anything).Return(nil)
 			fs.uploaderProv = mockProvider
 			fs.initialized = true

pkg/uploader/provider/kopia.go

@@ -118,6 +118,7 @@ func (kp *kopiaProvider) RunBackup(
 	tags map[string]string,
 	forceFull bool,
 	parentSnapshot string,
+	_ CBTParam,
 	volMode uploader.PersistentVolumeMode,
 	uploaderCfg map[string]string,
 	updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {

pkg/uploader/provider/kopia_test.go

@@ -106,7 +106,7 @@ func TestRunBackup(t *testing.T) {
 				tc.volMode = uploader.PersistentVolumeFilesystem
 			}
 			BackupFunc = tc.hookBackupFunc
-			_, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", tc.volMode, map[string]string{}, &updater)
+			_, _, _, _, err := kp.RunBackup(t.Context(), "var", "", nil, false, "", CBTParam{}, tc.volMode, map[string]string{}, &updater)
 			if tc.notError {
 				assert.NoError(t, err)
 			} else {

pkg/uploader/provider/mocks/Provider.go

@@ -1,115 +1,17 @@
-// Code generated by mockery v2.53.5. DO NOT EDIT.
+// Code generated by mockery; DO NOT EDIT.
+// github.com/vektra/mockery
+// template: testify
 
 package mocks
 
 import (
-	context "context"
+	"context"
 
 	mock "github.com/stretchr/testify/mock"
-
-	uploader "github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 )
 
-// Provider is an autogenerated mock type for the Provider type
-type Provider struct {
-	mock.Mock
-}
-
-// Close provides a mock function with given fields: ctx
-func (_m *Provider) Close(ctx context.Context) error {
-	ret := _m.Called(ctx)
-
-	if len(ret) == 0 {
-		panic("no return value specified for Close")
-	}
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
-		r0 = rf(ctx)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// 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 map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {
-	ret := _m.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-
-	if len(ret) == 0 {
-		panic("no return value specified for RunBackup")
-	}
-
-	var r0 string
-	var r1 bool
-	var r2 int64
-	var r3 int64
-	var r4 error
-	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, int64, int64, 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, 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, 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, map[string]string, uploader.ProgressUpdater) int64); ok {
-		r2 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	} else {
-		r2 = ret.Get(2).(int64)
-	}
-
-	if rf, ok := ret.Get(3).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
-		r3 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	} else {
-		r3 = ret.Get(3).(int64)
-	}
-
-	if rf, ok := ret.Get(4).(func(context.Context, string, string, map[string]string, bool, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
-		r4 = rf(ctx, path, realSource, tags, forceFull, parentSnapshot, volMode, uploaderCfg, updater)
-	} else {
-		r4 = ret.Error(4)
-	}
-
-	return r0, r1, r2, r3, r4
-}
-
-// 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) (int64, error) {
-	ret := _m.Called(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
-
-	if len(ret) == 0 {
-		panic("no return value specified for RunRestore")
-	}
-
-	var r0 int64
-	var r1 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok {
-		return rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
-	}
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
-		r0 = rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
-	} else {
-		r0 = ret.Get(0).(int64)
-	}
-
-	if rf, ok := ret.Get(1).(func(context.Context, string, string, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
-		r1 = rf(ctx, snapshotID, volumePath, volMode, uploaderConfig, updater)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
 // NewProvider creates a new instance of Provider. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
 // The first argument is typically a *testing.T value.
 func NewProvider(t interface {
@@ -123,3 +25,289 @@ func NewProvider(t interface {
 
 	return mock
 }
+
+// Provider is an autogenerated mock type for the Provider type
+type Provider struct {
+	mock.Mock
+}
+
+type Provider_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *Provider) EXPECT() *Provider_Expecter {
+	return &Provider_Expecter{mock: &_m.Mock}
+}
+
+// Close provides a mock function for the type Provider
+func (_mock *Provider) Close(ctx context.Context) error {
+	ret := _mock.Called(ctx)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Close")
+	}
+
+	var r0 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context) error); ok {
+		r0 = returnFunc(ctx)
+	} else {
+		r0 = ret.Error(0)
+	}
+	return r0
+}
+
+// Provider_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close'
+type Provider_Close_Call struct {
+	*mock.Call
+}
+
+// Close is a helper method to define mock.On call
+//   - ctx context.Context
+func (_e *Provider_Expecter) Close(ctx interface{}) *Provider_Close_Call {
+	return &Provider_Close_Call{Call: _e.mock.On("Close", ctx)}
+}
+
+func (_c *Provider_Close_Call) Run(run func(ctx context.Context)) *Provider_Close_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 context.Context
+		if args[0] != nil {
+			arg0 = args[0].(context.Context)
+		}
+		run(
+			arg0,
+		)
+	})
+	return _c
+}
+
+func (_c *Provider_Close_Call) Return(err error) *Provider_Close_Call {
+	_c.Call.Return(err)
+	return _c
+}
+
+func (_c *Provider_Close_Call) RunAndReturn(run func(ctx context.Context) error) *Provider_Close_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// RunBackup provides a mock function for the type Provider
+func (_mock *Provider) RunBackup(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error) {
+	ret := _mock.Called(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RunBackup")
+	}
+
+	var r0 string
+	var r1 bool
+	var r2 int64
+	var r3 int64
+	var r4 error
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (string, bool, int64, int64, error)); ok {
+		return returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	}
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) string); ok {
+		r0 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+	if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) bool); ok {
+		r1 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r1 = ret.Get(1).(bool)
+	}
+	if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
+		r2 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r2 = ret.Get(2).(int64)
+	}
+	if returnFunc, ok := ret.Get(3).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
+		r3 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r3 = ret.Get(3).(int64)
+	}
+	if returnFunc, ok := ret.Get(4).(func(context.Context, string, string, map[string]string, bool, string, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
+		r4 = returnFunc(ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)
+	} else {
+		r4 = ret.Error(4)
+	}
+	return r0, r1, r2, r3, r4
+}
+
+// Provider_RunBackup_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunBackup'
+type Provider_RunBackup_Call struct {
+	*mock.Call
+}
+
+// RunBackup is a helper method to define mock.On call
+//   - ctx context.Context
+//   - path string
+//   - realSource string
+//   - tags map[string]string
+//   - forceFull bool
+//   - parentSnapshot string
+//   - cbtParam provider.CBTParam
+//   - volMode uploader.PersistentVolumeMode
+//   - uploaderCfg map[string]string
+//   - updater uploader.ProgressUpdater
+func (_e *Provider_Expecter) RunBackup(ctx interface{}, path interface{}, realSource interface{}, tags interface{}, forceFull interface{}, parentSnapshot interface{}, cbtParam interface{}, volMode interface{}, uploaderCfg interface{}, updater interface{}) *Provider_RunBackup_Call {
+	return &Provider_RunBackup_Call{Call: _e.mock.On("RunBackup", ctx, path, realSource, tags, forceFull, parentSnapshot, cbtParam, volMode, uploaderCfg, updater)}
+}
+
+func (_c *Provider_RunBackup_Call) Run(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater)) *Provider_RunBackup_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 context.Context
+		if args[0] != nil {
+			arg0 = args[0].(context.Context)
+		}
+		var arg1 string
+		if args[1] != nil {
+			arg1 = args[1].(string)
+		}
+		var arg2 string
+		if args[2] != nil {
+			arg2 = args[2].(string)
+		}
+		var arg3 map[string]string
+		if args[3] != nil {
+			arg3 = args[3].(map[string]string)
+		}
+		var arg4 bool
+		if args[4] != nil {
+			arg4 = args[4].(bool)
+		}
+		var arg5 string
+		if args[5] != nil {
+			arg5 = args[5].(string)
+		}
+		var arg6 provider.CBTParam
+		if args[6] != nil {
+			arg6 = args[6].(provider.CBTParam)
+		}
+		var arg7 uploader.PersistentVolumeMode
+		if args[7] != nil {
+			arg7 = args[7].(uploader.PersistentVolumeMode)
+		}
+		var arg8 map[string]string
+		if args[8] != nil {
+			arg8 = args[8].(map[string]string)
+		}
+		var arg9 uploader.ProgressUpdater
+		if args[9] != nil {
+			arg9 = args[9].(uploader.ProgressUpdater)
+		}
+		run(
+			arg0,
+			arg1,
+			arg2,
+			arg3,
+			arg4,
+			arg5,
+			arg6,
+			arg7,
+			arg8,
+			arg9,
+		)
+	})
+	return _c
+}
+
+func (_c *Provider_RunBackup_Call) Return(s string, b bool, n int64, n1 int64, err error) *Provider_RunBackup_Call {
+	_c.Call.Return(s, b, n, n1, err)
+	return _c
+}
+
+func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, path string, realSource string, tags map[string]string, forceFull bool, parentSnapshot string, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, updater uploader.ProgressUpdater) (string, bool, int64, int64, error)) *Provider_RunBackup_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// 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)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RunRestore")
+	}
+
+	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, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
+		r0 = returnFunc(ctx, snapshotID, volumePath, 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)
+	} else {
+		r1 = ret.Error(1)
+	}
+	return r0, r1
+}
+
+// Provider_RunRestore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunRestore'
+type Provider_RunRestore_Call struct {
+	*mock.Call
+}
+
+// RunRestore is a helper method to define mock.On call
+//   - ctx context.Context
+//   - snapshotID string
+//   - volumePath string
+//   - 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 (_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 {
+	_c.Call.Run(func(args mock.Arguments) {
+		var arg0 context.Context
+		if args[0] != nil {
+			arg0 = args[0].(context.Context)
+		}
+		var arg1 string
+		if args[1] != nil {
+			arg1 = args[1].(string)
+		}
+		var arg2 string
+		if args[2] != nil {
+			arg2 = args[2].(string)
+		}
+		var arg3 uploader.PersistentVolumeMode
+		if args[3] != nil {
+			arg3 = args[3].(uploader.PersistentVolumeMode)
+		}
+		var arg4 map[string]string
+		if args[4] != nil {
+			arg4 = args[4].(map[string]string)
+		}
+		var arg5 uploader.ProgressUpdater
+		if args[5] != nil {
+			arg5 = args[5].(uploader.ProgressUpdater)
+		}
+		run(
+			arg0,
+			arg1,
+			arg2,
+			arg3,
+			arg4,
+			arg5,
+		)
+	})
+	return _c
+}
+
+func (_c *Provider_RunRestore_Call) Return(n int64, err error) *Provider_RunRestore_Call {
+	_c.Call.Return(n, err)
+	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 {
+	_c.Call.Return(run)
+	return _c
+}

pkg/uploader/provider/provider.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/cbtservice"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 )
 
@@ -37,6 +38,11 @@ const backupProgressCheckInterval = 10 * time.Second
 
 var ErrorCanceled error = errors.New("uploader is canceled")
 
+type CBTParam struct {
+	Source  cbtservice.SourceInfo
+	Service cbtservice.Service
+}
+
 // Provider which is designed for one pod volume to do the backup or restore
 type Provider interface {
 	// RunBackup which will do backup for one specific volume and return snapshotID, isSnapshotEmpty, error
@@ -48,6 +54,7 @@ type Provider interface {
 		tags map[string]string,
 		forceFull bool,
 		parentSnapshot string,
+		cbtParam CBTParam,
 		volMode uploader.PersistentVolumeMode,
 		uploaderCfg map[string]string,
 		updater uploader.ProgressUpdater) (string, bool, int64, int64, error)