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,38 +96,38 @@ 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
 
 	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
 	}