Merge pull request #7955 from Lyndon-Li/data-mover-ms-new-data-path

New data path for data mover ms

changelogs/unreleased/7955-Lyndon-Li

@@ -0,0 +1 @@
+New data path for data mover ms according to design #7574

pkg/cmd/cli/repomantenance/maintenance.go

@@ -139,7 +139,7 @@ func (o *Options) runRepoPrune(f velerocli.Factory, namespace string, logger log
 			credentials.CredentialGetter{
 				FromFile:   credentialFileStore,
 				FromSecret: credentialSecretStore,
-			}, o.RepoType, cli, logger)
+			}, o.RepoType, logger)
 	}
 
 	// backupRepository

pkg/controller/data_download_controller.go

@@ -334,10 +334,19 @@ func (r *DataDownloadReconciler) runCancelableDataPath(ctx context.Context, fsRe
 	}
 
 	log.WithField("path", path.ByPath).Debug("Found host path")
-	if err := fsRestore.Init(ctx, dd.Spec.BackupStorageLocation, dd.Spec.SourceNamespace, datamover.GetUploaderType(dd.Spec.DataMover),
-		velerov1api.BackupRepositoryTypeKopia, "", r.repositoryEnsurer, r.credentialGetter); err != nil {
+
+	if err := fsRestore.Init(ctx, &datapath.FSBRInitParam{
+		BSLName:           dd.Spec.BackupStorageLocation,
+		SourceNamespace:   dd.Spec.SourceNamespace,
+		UploaderType:      datamover.GetUploaderType(dd.Spec.DataMover),
+		RepositoryType:    velerov1api.BackupRepositoryTypeKopia,
+		RepoIdentifier:    "",
+		RepositoryEnsurer: r.repositoryEnsurer,
+		CredentialGetter:  r.credentialGetter,
+	}); err != nil {
 		return r.errorOut(ctx, dd, err, "error to initialize data path", log)
 	}
+
 	log.WithField("path", path.ByPath).Info("fs init")
 
 	if err := fsRestore.StartRestore(dd.Spec.SnapshotID, path, dd.Spec.DataMoverConfig); err != nil {

pkg/controller/data_upload_controller.go

@@ -338,23 +338,38 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 
 func (r *DataUploadReconciler) runCancelableDataUpload(ctx context.Context, fsBackup datapath.AsyncBR, du *velerov2alpha1api.DataUpload, res *exposer.ExposeResult, log logrus.FieldLogger) (reconcile.Result, error) {
 	log.Info("Run cancelable dataUpload")
+
 	path, err := exposer.GetPodVolumeHostPath(ctx, res.ByPod.HostingPod, res.ByPod.VolumeName, r.client, r.fileSystem, log)
 	if err != nil {
 		return r.errorOut(ctx, du, err, "error exposing host path for pod volume", log)
 	}
 
 	log.WithField("path", path.ByPath).Debug("Found host path")
-	if err := fsBackup.Init(ctx, du.Spec.BackupStorageLocation, du.Spec.SourceNamespace, datamover.GetUploaderType(du.Spec.DataMover),
-		velerov1api.BackupRepositoryTypeKopia, "", r.repoEnsurer, r.credentialGetter); err != nil {
+
+	if err := fsBackup.Init(ctx, &datapath.FSBRInitParam{
+		BSLName:           du.Spec.BackupStorageLocation,
+		SourceNamespace:   du.Spec.SourceNamespace,
+		UploaderType:      datamover.GetUploaderType(du.Spec.DataMover),
+		RepositoryType:    velerov1api.BackupRepositoryTypeKopia,
+		RepoIdentifier:    "",
+		RepositoryEnsurer: r.repoEnsurer,
+		CredentialGetter:  r.credentialGetter,
+	}); err != nil {
 		return r.errorOut(ctx, du, err, "error to initialize data path", log)
 	}
+
 	log.WithField("path", path.ByPath).Info("fs init")
 
 	tags := map[string]string{
 		velerov1api.AsyncOperationIDLabel: du.Labels[velerov1api.AsyncOperationIDLabel],
 	}
 
-	if err := fsBackup.StartBackup(path, datamover.GetRealSource(du.Spec.SourceNamespace, du.Spec.SourcePVC), "", false, tags, du.Spec.DataMoverConfig); err != nil {
+	if err := fsBackup.StartBackup(path, du.Spec.DataMoverConfig, &datapath.FSBRStartParam{
+		RealSource:     datamover.GetRealSource(du.Spec.SourceNamespace, du.Spec.SourcePVC),
+		ParentSnapshot: "",
+		ForceFull:      false,
+		Tags:           tags,
+	}); err != nil {
 		return r.errorOut(ctx, du, err, "error starting data path backup", log)
 	}
 

pkg/controller/data_upload_controller_test.go

@@ -51,7 +51,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	"github.com/vmware-tanzu/velero/pkg/exposer"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
-	"github.com/vmware-tanzu/velero/pkg/repository"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
@@ -297,11 +296,11 @@ type fakeDataUploadFSBR struct {
 	clock      clock.WithTickerAndDelayedExecution
 }
 
-func (f *fakeDataUploadFSBR) Init(ctx context.Context, bslName string, sourceNamespace string, uploaderType string, repositoryType string, repoIdentifier string, repositoryEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter) error {
+func (f *fakeDataUploadFSBR) Init(ctx context.Context, param interface{}) error {
 	return nil
 }
 
-func (f *fakeDataUploadFSBR) StartBackup(source datapath.AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfigs map[string]string) error {
+func (f *fakeDataUploadFSBR) StartBackup(source datapath.AccessPoint, uploaderConfigs map[string]string, param interface{}) error {
 	du := f.du
 	original := f.du.DeepCopy()
 	du.Status.Phase = velerov2alpha1api.DataUploadPhaseCompleted

pkg/controller/pod_volume_backup_controller.go

@@ -159,8 +159,15 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 
 	log.WithField("path", path.ByPath).Debugf("Found host path")
 
-	if err := fsBackup.Init(ctx, pvb.Spec.BackupStorageLocation, pvb.Spec.Pod.Namespace, pvb.Spec.UploaderType,
-		podvolume.GetPvbRepositoryType(&pvb), pvb.Spec.RepoIdentifier, r.repositoryEnsurer, r.credentialGetter); err != nil {
+	if err := fsBackup.Init(ctx, &datapath.FSBRInitParam{
+		BSLName:           pvb.Spec.BackupStorageLocation,
+		SourceNamespace:   pvb.Spec.Pod.Namespace,
+		UploaderType:      pvb.Spec.UploaderType,
+		RepositoryType:    podvolume.GetPvbRepositoryType(&pvb),
+		RepoIdentifier:    pvb.Spec.RepoIdentifier,
+		RepositoryEnsurer: r.repositoryEnsurer,
+		CredentialGetter:  r.credentialGetter,
+	}); err != nil {
 		return r.errorOut(ctx, &pvb, err, "error to initialize data path", log)
 	}
 
@@ -179,7 +186,12 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		}
 	}
 
-	if err := fsBackup.StartBackup(path, "", parentSnapshotID, false, pvb.Spec.Tags, pvb.Spec.UploaderSettings); err != nil {
+	if err := fsBackup.StartBackup(path, pvb.Spec.UploaderSettings, &datapath.FSBRStartParam{
+		RealSource:     "",
+		ParentSnapshot: parentSnapshotID,
+		ForceFull:      false,
+		Tags:           pvb.Spec.Tags,
+	}); err != nil {
 		return r.errorOut(ctx, &pvb, err, "error starting data path backup", log)
 	}
 

pkg/controller/pod_volume_backup_controller_test.go

@@ -41,7 +41,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
-	"github.com/vmware-tanzu/velero/pkg/repository"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 )
 
@@ -99,11 +98,11 @@ type fakeFSBR struct {
 	clock  clock.WithTickerAndDelayedExecution
 }
 
-func (b *fakeFSBR) Init(ctx context.Context, bslName string, sourceNamespace string, uploaderType string, repositoryType string, repoIdentifier string, repositoryEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter) error {
+func (b *fakeFSBR) Init(ctx context.Context, param interface{}) error {
 	return nil
 }
 
-func (b *fakeFSBR) StartBackup(source datapath.AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, uploaderConfigs map[string]string) error {
+func (b *fakeFSBR) StartBackup(source datapath.AccessPoint, uploaderConfigs map[string]string, param interface{}) error {
 	pvb := b.pvb
 
 	original := b.pvb.DeepCopy()

pkg/controller/pod_volume_restore_controller.go

@@ -141,8 +141,15 @@ func (c *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req
 
 	log.WithField("path", volumePath.ByPath).Debugf("Found host path")
 
-	if err := fsRestore.Init(ctx, pvr.Spec.BackupStorageLocation, pvr.Spec.SourceNamespace, pvr.Spec.UploaderType,
-		podvolume.GetPvrRepositoryType(pvr), pvr.Spec.RepoIdentifier, c.repositoryEnsurer, c.credentialGetter); err != nil {
+	if err := fsRestore.Init(ctx, &datapath.FSBRInitParam{
+		BSLName:           pvr.Spec.BackupStorageLocation,
+		SourceNamespace:   pvr.Spec.SourceNamespace,
+		UploaderType:      pvr.Spec.UploaderType,
+		RepositoryType:    podvolume.GetPvrRepositoryType(pvr),
+		RepoIdentifier:    pvr.Spec.RepoIdentifier,
+		RepositoryEnsurer: c.repositoryEnsurer,
+		CredentialGetter:  c.credentialGetter,
+	}); err != nil {
 		return c.errorOut(ctx, pvr, err, "error to initialize data path", log)
 	}
 

pkg/datapath/file_system.go

@@ -33,6 +33,26 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 )
 
+// FSBRInitParam define the input param for FSBR init
+type FSBRInitParam struct {
+	BSLName           string
+	SourceNamespace   string
+	UploaderType      string
+	RepositoryType    string
+	RepoIdentifier    string
+	RepositoryEnsurer *repository.Ensurer
+	CredentialGetter  *credentials.CredentialGetter
+	Filesystem        filesystem.Interface
+}
+
+// FSBRStartParam define the input param for FSBR start
+type FSBRStartParam struct {
+	RealSource     string
+	ParentSnapshot string
+	ForceFull      bool
+	Tags           map[string]string
+}
+
 type fileSystemBR struct {
 	ctx            context.Context
 	cancel         context.CancelFunc
@@ -61,8 +81,9 @@ func newFileSystemBR(jobName string, requestorType string, client client.Client,
 	return fs
 }
 
-func (fs *fileSystemBR) Init(ctx context.Context, bslName string, sourceNamespace string, uploaderType string, repositoryType string,
-	repoIdentifier string, repositoryEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter) error {
+func (fs *fileSystemBR) Init(ctx context.Context, param interface{}) error {
+	initParam := param.(*FSBRInitParam)
+
 	var err error
 	defer func() {
 		if err != nil {
@@ -75,27 +96,27 @@ func (fs *fileSystemBR) Init(ctx context.Context, bslName string, sourceNamespac
 	backupLocation := &velerov1api.BackupStorageLocation{}
 	if err = fs.client.Get(ctx, client.ObjectKey{
 		Namespace: fs.namespace,
-		Name:      bslName,
+		Name:      initParam.BSLName,
 	}, backupLocation); err != nil {
-		return errors.Wrapf(err, "error getting backup storage location %s", bslName)
+		return errors.Wrapf(err, "error getting backup storage location %s", initParam.BSLName)
 	}
 
 	fs.backupLocation = backupLocation
 
-	fs.backupRepo, err = repositoryEnsurer.EnsureRepo(ctx, fs.namespace, sourceNamespace, bslName, repositoryType)
+	fs.backupRepo, err = initParam.RepositoryEnsurer.EnsureRepo(ctx, fs.namespace, initParam.SourceNamespace, initParam.BSLName, initParam.RepositoryType)
 	if err != nil {
-		return errors.Wrapf(err, "error to ensure backup repository %s-%s-%s", bslName, sourceNamespace, repositoryType)
+		return errors.Wrapf(err, "error to ensure backup repository %s-%s-%s", initParam.BSLName, initParam.SourceNamespace, initParam.RepositoryType)
 	}
 
-	err = fs.boostRepoConnect(ctx, repositoryType, credentialGetter)
+	err = fs.boostRepoConnect(ctx, initParam.RepositoryType, initParam.CredentialGetter)
 	if err != nil {
-		return errors.Wrapf(err, "error to boost backup repository connection %s-%s-%s", bslName, sourceNamespace, repositoryType)
+		return errors.Wrapf(err, "error to boost backup repository connection %s-%s-%s", initParam.BSLName, initParam.SourceNamespace, initParam.RepositoryType)
 	}
 
-	fs.uploaderProv, err = provider.NewUploaderProvider(ctx, fs.client, uploaderType, fs.requestorType, repoIdentifier,
-		fs.backupLocation, fs.backupRepo, credentialGetter, repokey.RepoKeySelector(), fs.log)
+	fs.uploaderProv, err = provider.NewUploaderProvider(ctx, fs.client, initParam.UploaderType, fs.requestorType, initParam.RepoIdentifier,
+		fs.backupLocation, fs.backupRepo, initParam.CredentialGetter, repokey.RepoKeySelector(), fs.log)
 	if err != nil {
-		return errors.Wrapf(err, "error creating uploader %s", uploaderType)
+		return errors.Wrapf(err, "error creating uploader %s", initParam.UploaderType)
 	}
 
 	fs.initialized = true
@@ -103,10 +124,10 @@ func (fs *fileSystemBR) Init(ctx context.Context, bslName string, sourceNamespac
 	fs.log.WithFields(
 		logrus.Fields{
 			"jobName":          fs.jobName,
-			"bsl":              bslName,
-			"source namespace": sourceNamespace,
-			"uploader":         uploaderType,
-			"repository":       repositoryType,
+			"bsl":              initParam.BSLName,
+			"source namespace": initParam.SourceNamespace,
+			"uploader":         initParam.UploaderType,
+			"repository":       initParam.RepositoryType,
 		}).Info("FileSystemBR is initialized")
 
 	return nil
@@ -129,14 +150,16 @@ 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, uploaderConfig map[string]string) error {
+func (fs *fileSystemBR) StartBackup(source AccessPoint, uploaderConfig map[string]string, param interface{}) error {
 	if !fs.initialized {
 		return errors.New("file system data path is not initialized")
 	}
 
+	backupParam := param.(*FSBRStartParam)
+
 	go func() {
-		snapshotID, emptySnapshot, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, realSource, tags, forceFull,
-			parentSnapshot, source.VolMode, uploaderConfig, fs)
+		snapshotID, emptySnapshot, err := fs.uploaderProv.RunBackup(fs.ctx, source.ByPath, backupParam.RealSource, backupParam.Tags, backupParam.ForceFull,
+			backupParam.ParentSnapshot, source.VolMode, uploaderConfig, fs)
 
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)
@@ -192,7 +215,7 @@ func (fs *fileSystemBR) Cancel() {
 
 func (fs *fileSystemBR) boostRepoConnect(ctx context.Context, repositoryType string, credentialGetter *credentials.CredentialGetter) error {
 	if repositoryType == velerov1api.BackupRepositoryTypeKopia {
-		if err := repoProvider.NewUnifiedRepoProvider(*credentialGetter, repositoryType, fs.client, fs.log).BoostRepoConnect(ctx, repoProvider.RepoParam{BackupLocation: fs.backupLocation, BackupRepo: fs.backupRepo}); err != nil {
+		if err := repoProvider.NewUnifiedRepoProvider(*credentialGetter, repositoryType, fs.log).BoostRepoConnect(ctx, repoProvider.RepoParam{BackupLocation: fs.backupLocation, BackupRepo: fs.backupRepo}); err != nil {
 			return err
 		}
 	} else {

pkg/datapath/file_system_test.go

@@ -100,13 +100,13 @@ func TestAsyncBackup(t *testing.T) {
 			fs.initialized = true
 			fs.callbacks = test.callbacks
 
-			err := fs.StartBackup(AccessPoint{ByPath: test.path}, "", "", false, nil, map[string]string{})
+			err := fs.StartBackup(AccessPoint{ByPath: test.path}, map[string]string{}, &FSBRStartParam{})
 			require.NoError(t, err)
 
 			<-finish
 
-			assert.Equal(t, asyncErr, test.err)
-			assert.Equal(t, asyncResult, test.result)
+			assert.Equal(t, test.err, asyncErr)
+			assert.Equal(t, test.result, asyncResult)
 		})
 	}
 

pkg/datapath/manager.go

@@ -47,7 +47,7 @@ func (m *Manager) CreateFileSystemBR(jobName string, requestorType string, ctx c
 	m.trackerLock.Lock()
 	defer m.trackerLock.Unlock()
 
-	if len(m.tracker) == m.cocurrentNum {
+	if len(m.tracker) >= m.cocurrentNum {
 		return nil, ConcurrentLimitExceed
 	}
 

pkg/datapath/mocks/asyncBR.go

@@ -0,0 +1,93 @@
+// Code generated by mockery v2.39.1. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+	datapath "github.com/vmware-tanzu/velero/pkg/datapath"
+)
+
+// AsyncBR is an autogenerated mock type for the AsyncBR type
+type AsyncBR struct {
+	mock.Mock
+}
+
+// Cancel provides a mock function with given fields:
+func (_m *AsyncBR) Cancel() {
+	_m.Called()
+}
+
+// Close provides a mock function with given fields: ctx
+func (_m *AsyncBR) Close(ctx context.Context) {
+	_m.Called(ctx)
+}
+
+// Init provides a mock function with given fields: ctx, param
+func (_m *AsyncBR) Init(ctx context.Context, param interface{}) error {
+	ret := _m.Called(ctx, param)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Init")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, interface{}) error); ok {
+		r0 = rf(ctx, param)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// StartBackup provides a mock function with given fields: source, dataMoverConfig, param
+func (_m *AsyncBR) StartBackup(source datapath.AccessPoint, dataMoverConfig map[string]string, param interface{}) error {
+	ret := _m.Called(source, dataMoverConfig, param)
+
+	if len(ret) == 0 {
+		panic("no return value specified for StartBackup")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(datapath.AccessPoint, map[string]string, interface{}) error); ok {
+		r0 = rf(source, dataMoverConfig, param)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// StartRestore provides a mock function with given fields: snapshotID, target, dataMoverConfig
+func (_m *AsyncBR) StartRestore(snapshotID string, target datapath.AccessPoint, dataMoverConfig map[string]string) error {
+	ret := _m.Called(snapshotID, target, dataMoverConfig)
+
+	if len(ret) == 0 {
+		panic("no return value specified for StartRestore")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, datapath.AccessPoint, map[string]string) error); ok {
+		r0 = rf(snapshotID, target, dataMoverConfig)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// NewAsyncBR creates a new instance of AsyncBR. 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 NewAsyncBR(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *AsyncBR {
+	mock := &AsyncBR{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

pkg/datapath/mocks/types.go

@@ -1,86 +0,0 @@
-// Code generated by mockery v2.20.0. DO NOT EDIT.
-
-package mocks
-
-import (
-	context "context"
-
-	credentials "github.com/vmware-tanzu/velero/internal/credentials"
-	datapath "github.com/vmware-tanzu/velero/pkg/datapath"
-
-	mock "github.com/stretchr/testify/mock"
-
-	repository "github.com/vmware-tanzu/velero/pkg/repository"
-)
-
-// AsyncBR is an autogenerated mock type for the AsyncBR type
-type AsyncBR struct {
-	mock.Mock
-}
-
-// Cancel provides a mock function with given fields:
-func (_m *AsyncBR) Cancel() {
-	_m.Called()
-}
-
-// Close provides a mock function with given fields: ctx
-func (_m *AsyncBR) Close(ctx context.Context) {
-	_m.Called(ctx)
-}
-
-// Init provides a mock function with given fields: ctx, bslName, sourceNamespace, uploaderType, repositoryType, repoIdentifier, repositoryEnsurer, credentialGetter
-func (_m *AsyncBR) Init(ctx context.Context, bslName string, sourceNamespace string, uploaderType string, repositoryType string, repoIdentifier string, repositoryEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter) error {
-	ret := _m.Called(ctx, bslName, sourceNamespace, uploaderType, repositoryType, repoIdentifier, repositoryEnsurer, credentialGetter)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, string, string, string, *repository.Ensurer, *credentials.CredentialGetter) error); ok {
-		r0 = rf(ctx, bslName, sourceNamespace, uploaderType, repositoryType, repoIdentifier, repositoryEnsurer, credentialGetter)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// 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, map[string]string) error); ok {
-		r0 = rf(source, realSource, parentSnapshot, forceFull, tags, dataMoverConfig)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// StartRestore provides a mock function with given fields: snapshotID, target, dataMoverConfig
-func (_m *AsyncBR) StartRestore(snapshotID string, target datapath.AccessPoint, dataMoverConfig map[string]string) error {
-	ret := _m.Called(snapshotID, target, dataMoverConfig)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(string, datapath.AccessPoint, map[string]string) error); ok {
-		r0 = rf(snapshotID, target, dataMoverConfig)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-type mockConstructorTestingTNewAsyncBR interface {
-	mock.TestingT
-	Cleanup(func())
-}
-
-// NewAsyncBR creates a new instance of AsyncBR. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
-func NewAsyncBR(t mockConstructorTestingTNewAsyncBR) *AsyncBR {
-	mock := &AsyncBR{}
-	mock.Mock.Test(t)
-
-	t.Cleanup(func() { mock.AssertExpectations(t) })
-
-	return mock
-}

pkg/datapath/types.go

@@ -19,8 +19,6 @@ package datapath
 import (
 	"context"
 
-	"github.com/vmware-tanzu/velero/internal/credentials"
-	"github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 )
 
@@ -32,14 +30,14 @@ type Result struct {
 
 // BackupResult represents the result of a backup
 type BackupResult struct {
-	SnapshotID    string
-	EmptySnapshot bool
-	Source        AccessPoint
+	SnapshotID    string      `json:"snapshotID"`
+	EmptySnapshot bool        `json:"emptySnapshot"`
+	Source        AccessPoint `json:"source,omitempty"`
 }
 
 // RestoreResult represents the result of a restore
 type RestoreResult struct {
-	Target AccessPoint
+	Target AccessPoint `json:"target,omitempty"`
 }
 
 // Callbacks defines the collection of callbacks during backup/restore
@@ -59,10 +57,10 @@ type AccessPoint struct {
 // AsyncBR is the interface for asynchronous data path methods
 type AsyncBR interface {
 	// Init initializes an asynchronous data path instance
-	Init(ctx context.Context, bslName string, sourceNamespace string, uploaderType string, repositoryType string, repoIdentifier string, repositoryEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter) error
+	Init(ctx context.Context, param interface{}) error
 
 	// StartBackup starts an asynchronous data path instance for backup
-	StartBackup(source AccessPoint, realSource string, parentSnapshot string, forceFull bool, tags map[string]string, dataMoverConfig map[string]string) error
+	StartBackup(source AccessPoint, dataMoverConfig map[string]string, param interface{}) error
 
 	// StartRestore starts an asynchronous data path instance for restore
 	StartRestore(snapshotID string, target AccessPoint, dataMoverConfig map[string]string) error

pkg/repository/manager.go

@@ -128,7 +128,7 @@ func NewManager(
 	mgr.providers[velerov1api.BackupRepositoryTypeKopia] = provider.NewUnifiedRepoProvider(credentials.CredentialGetter{
 		FromFile:   credentialFileStore,
 		FromSecret: credentialSecretStore,
-	}, velerov1api.BackupRepositoryTypeKopia, client, mgr.log)
+	}, velerov1api.BackupRepositoryTypeKopia, mgr.log)
 
 	return mgr
 }

pkg/repository/provider/unified_repo.go

@@ -29,7 +29,6 @@ import (
 	"github.com/kopia/kopia/repo"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
@@ -44,7 +43,6 @@ type unifiedRepoProvider struct {
 	workPath         string
 	repoService      udmrepo.BackupRepoService
 	repoBackend      string
-	cli              client.Client
 	log              logrus.FieldLogger
 }
 
@@ -75,13 +73,11 @@ const (
 func NewUnifiedRepoProvider(
 	credentialGetter credentials.CredentialGetter,
 	repoBackend string,
-	cli client.Client,
 	log logrus.FieldLogger,
 ) Provider {
 	repo := unifiedRepoProvider{
 		credentialGetter: credentialGetter,
 		repoBackend:      repoBackend,
-		cli:              cli,
 		log:              log,
 	}