add shared generic data path

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

pkg/controller/pod_volume_backup_controller.go

@@ -27,49 +27,60 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	clocks "k8s.io/utils/clock"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"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/podvolume"
 	"github.com/vmware-tanzu/velero/pkg/repository"
-	repokey "github.com/vmware-tanzu/velero/pkg/repository/keys"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
-	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
-	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
-// NewUploaderProviderFunc is used for unit test to mock function
-var NewUploaderProviderFunc = provider.NewUploaderProvider
+const pVBRRequestor string = "pod-volume-backup-restore"
+
+// NewPodVolumeBackupReconciler creates the PodVolumeBackupReconciler instance
+func NewPodVolumeBackupReconciler(client client.Client, ensurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter,
+	nodeName string, scheme *runtime.Scheme, metrics *metrics.ServerMetrics, logger logrus.FieldLogger) *PodVolumeBackupReconciler {
+	return &PodVolumeBackupReconciler{
+		Client:            client,
+		logger:            logger.WithField("controller", "PodVolumeBackup"),
+		repositoryEnsurer: ensurer,
+		credentialGetter:  credentialGetter,
+		nodeName:          nodeName,
+		fileSystem:        filesystem.NewFileSystem(),
+		clock:             &clocks.RealClock{},
+		scheme:            scheme,
+		metrics:           metrics,
+		dataPathMgr:       datapath.NewManager(1),
+	}
+}
 
 // PodVolumeBackupReconciler reconciles a PodVolumeBackup object
 type PodVolumeBackupReconciler struct {
-	Scheme           *runtime.Scheme
-	Client           client.Client
-	Clock            clocks.WithTickerAndDelayedExecution
-	Metrics          *metrics.ServerMetrics
-	CredentialGetter *credentials.CredentialGetter
-	NodeName         string
-	FileSystem       filesystem.Interface
-	Log              logrus.FieldLogger
-}
-
-type BackupProgressUpdater struct {
-	PodVolumeBackup *velerov1api.PodVolumeBackup
-	Log             logrus.FieldLogger
-	Ctx             context.Context
-	Cli             client.Client
+	client.Client
+	scheme            *runtime.Scheme
+	clock             clocks.WithTickerAndDelayedExecution
+	metrics           *metrics.ServerMetrics
+	credentialGetter  *credentials.CredentialGetter
+	repositoryEnsurer *repository.Ensurer
+	nodeName          string
+	fileSystem        filesystem.Interface
+	logger            logrus.FieldLogger
+	dataPathMgr       *datapath.Manager
 }
 
 // +kubebuilder:rbac:groups=velero.io,resources=podvolumebackups,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=velero.io,resources=podvolumebackups/status,verbs=get;update;patch
 
 func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	log := r.Log.WithFields(logrus.Fields{
+	log := r.logger.WithFields(logrus.Fields{
 		"controller":      "podvolumebackup",
 		"podvolumebackup": req.NamespacedName,
 	})
@@ -92,7 +103,7 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	log.Info("PodVolumeBackup starting")
 
 	// Only process items for this node.
-	if pvb.Spec.Node != r.NodeName {
+	if pvb.Spec.Node != r.nodeName {
 		return ctrl.Result{}, nil
 	}
 
@@ -104,15 +115,30 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		return ctrl.Result{}, nil
 	}
 
-	r.Metrics.RegisterPodVolumeBackupEnqueue(r.NodeName)
+	callbacks := datapath.Callbacks{
+		OnCompleted: r.OnDataPathCompleted,
+		OnFailed:    r.OnDataPathFailed,
+		OnCancelled: r.OnDataPathCancelled,
+		OnProgress:  r.OnDataPathProgress,
+	}
+
+	fsBackup, err := r.dataPathMgr.CreateFileSystemBR(pvb.Name, pVBRRequestor, ctx, r.Client, pvb.Namespace, callbacks, log)
+	if err != nil {
+		if err == datapath.ConcurrentLimitExceed {
+			return ctrl.Result{Requeue: true, RequeueAfter: time.Minute}, nil
+		} else {
+			return r.errorOut(ctx, &pvb, err, "error to create data path", log)
+		}
+	}
+
+	r.metrics.RegisterPodVolumeBackupEnqueue(r.nodeName)
 
 	// Update status to InProgress.
 	original := pvb.DeepCopy()
 	pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseInProgress
-	pvb.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
+	pvb.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now()}
 	if err := r.Client.Patch(ctx, &pvb, client.MergeFrom(original)); err != nil {
-		log.WithError(err).Error("error updating PodVolumeBackup status")
-		return ctrl.Result{}, err
+		return r.errorOut(ctx, &pvb, err, "error updating PodVolumeBackup status", log)
 	}
 
 	var pod corev1.Pod
@@ -121,45 +147,19 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		Name:      pvb.Spec.Pod.Name,
 	}
 	if err := r.Client.Get(ctx, podNamespacedName, &pod); err != nil {
-		return r.updateStatusToFailed(ctx, &pvb, err, fmt.Sprintf("getting pod %s/%s", pvb.Spec.Pod.Namespace, pvb.Spec.Pod.Name), log)
+		return r.errorOut(ctx, &pvb, err, fmt.Sprintf("getting pod %s/%s", pvb.Spec.Pod.Namespace, pvb.Spec.Pod.Name), log)
 	}
 
-	volDir, err := kube.GetVolumeDirectory(ctx, log, &pod, pvb.Spec.Volume, r.Client)
+	path, err := exposer.GetPodVolumeHostPath(ctx, &pod, pvb.Spec.Volume, r.Client, r.fileSystem, log)
 	if err != nil {
-		return r.updateStatusToFailed(ctx, &pvb, err, "getting volume directory name", log)
+		return r.errorOut(ctx, &pvb, err, "error exposing host path for pod volume", log)
 	}
 
-	pathGlob := fmt.Sprintf("/host_pods/%s/volumes/*/%s", string(pvb.Spec.Pod.UID), volDir)
-	log.WithField("pathGlob", pathGlob).Debug("Looking for path matching glob")
+	log.WithField("path", path.ByPath).Debugf("Found host path")
 
-	path, err := kube.SinglePathMatch(pathGlob, r.FileSystem, log)
-	if err != nil {
-		return r.updateStatusToFailed(ctx, &pvb, err, "identifying unique volume path on host", log)
-	}
-	log.WithField("path", path).Debugf("Found path matching glob")
-
-	backupLocation := &velerov1api.BackupStorageLocation{}
-	if err := r.Client.Get(context.Background(), client.ObjectKey{
-		Namespace: pvb.Namespace,
-		Name:      pvb.Spec.BackupStorageLocation,
-	}, backupLocation); err != nil {
-		return ctrl.Result{}, errors.Wrap(err, "error getting backup storage location")
-	}
-
-	backupRepo, err := repository.GetBackupRepository(ctx, r.Client, pvb.Namespace, repository.BackupRepositoryKey{
-		VolumeNamespace: pvb.Spec.Pod.Namespace,
-		BackupLocation:  pvb.Spec.BackupStorageLocation,
-		RepositoryType:  podvolume.GetPvbRepositoryType(&pvb),
-	})
-	if err != nil {
-		return ctrl.Result{}, errors.Wrap(err, "error getting backup repository")
-	}
-
-	var uploaderProv provider.Provider
-	uploaderProv, err = NewUploaderProviderFunc(ctx, r.Client, pvb.Spec.UploaderType, pvb.Spec.RepoIdentifier,
-		backupLocation, backupRepo, r.CredentialGetter, repokey.RepoKeySelector(), log)
-	if err != nil {
-		return r.updateStatusToFailed(ctx, &pvb, err, "error creating uploader", log)
+	if err := fsBackup.Init(ctx, pvb.Spec.BackupStorageLocation, pvb.Spec.Pod.Namespace, pvb.Spec.UploaderType,
+		podvolume.GetPvbRepositoryType(&pvb), r.repositoryEnsurer, r.credentialGetter); err != nil {
+		return r.errorOut(ctx, &pvb, err, "error to initialize data path", log)
 	}
 
 	// If this is a PVC, look for the most recent completed pod volume backup for it and get
@@ -177,41 +177,98 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		}
 	}
 
-	defer func() {
-		if err := uploaderProv.Close(ctx); err != nil {
-			log.Errorf("failed to close uploader provider with error %v", err)
-		}
-	}()
+	if err := fsBackup.StartBackup(path, parentSnapshotID, false, pvb.Spec.Tags); err != nil {
+		return r.errorOut(ctx, &pvb, err, "error starting data path backup", log)
+	}
 
-	snapshotID, emptySnapshot, err := uploaderProv.RunBackup(ctx, path, pvb.Spec.Tags, parentSnapshotID, r.NewBackupProgressUpdater(ctx, &pvb, log))
-	if err != nil {
-		return r.updateStatusToFailed(ctx, &pvb, err, fmt.Sprintf("running backup, stderr=%v", err), log)
+	log.WithField("path", path.ByPath).Info("Async fs backup data path started")
+
+	return ctrl.Result{}, nil
+}
+
+func (r *PodVolumeBackupReconciler) OnDataPathCompleted(ctx context.Context, namespace string, pvbName string, result datapath.Result) {
+	defer r.closeDataPath(ctx, pvbName)
+
+	log := r.logger.WithField("pvb", pvbName)
+
+	log.WithField("PVB", pvbName).Info("Async fs backup data path completed")
+
+	var pvb velerov1api.PodVolumeBackup
+	if err := r.Client.Get(ctx, types.NamespacedName{Name: pvbName, Namespace: namespace}, &pvb); err != nil {
+		log.WithError(err).Warn("Failed to get PVB on completion")
+		return
 	}
 
 	// Update status to Completed with path & snapshot ID.
-	original = pvb.DeepCopy()
-	pvb.Status.Path = path
+	original := pvb.DeepCopy()
+	pvb.Status.Path = result.Backup.Source.ByPath
 	pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseCompleted
-	pvb.Status.SnapshotID = snapshotID
-	pvb.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
-	if emptySnapshot {
+	pvb.Status.SnapshotID = result.Backup.SnapshotID
+	pvb.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+	if result.Backup.EmptySnapshot {
 		pvb.Status.Message = "volume was empty so no snapshot was taken"
 	}
-	if err = r.Client.Patch(ctx, &pvb, client.MergeFrom(original)); err != nil {
+
+	if err := r.Client.Patch(ctx, &pvb, client.MergeFrom(original)); err != nil {
 		log.WithError(err).Error("error updating PodVolumeBackup status")
-		return ctrl.Result{}, err
 	}
 
 	latencyDuration := pvb.Status.CompletionTimestamp.Time.Sub(pvb.Status.StartTimestamp.Time)
 	latencySeconds := float64(latencyDuration / time.Second)
-	backupName := fmt.Sprintf("%s/%s", req.Namespace, pvb.OwnerReferences[0].Name)
-	generateOpName := fmt.Sprintf("%s-%s-%s-%s-%s-backup", pvb.Name, backupRepo.Name, pvb.Spec.BackupStorageLocation, pvb.Namespace, pvb.Spec.UploaderType)
-	r.Metrics.ObservePodVolumeOpLatency(r.NodeName, req.Name, generateOpName, backupName, latencySeconds)
-	r.Metrics.RegisterPodVolumeOpLatencyGauge(r.NodeName, req.Name, generateOpName, backupName, latencySeconds)
-	r.Metrics.RegisterPodVolumeBackupDequeue(r.NodeName)
+	backupName := fmt.Sprintf("%s/%s", pvb.Namespace, pvb.OwnerReferences[0].Name)
+	generateOpName := fmt.Sprintf("%s-%s-%s-%s-backup", pvb.Name, pvb.Spec.BackupStorageLocation, pvb.Spec.Pod.Namespace, pvb.Spec.UploaderType)
+	r.metrics.ObservePodVolumeOpLatency(r.nodeName, pvb.Name, generateOpName, backupName, latencySeconds)
+	r.metrics.RegisterPodVolumeOpLatencyGauge(r.nodeName, pvb.Name, generateOpName, backupName, latencySeconds)
+	r.metrics.RegisterPodVolumeBackupDequeue(r.nodeName)
 
 	log.Info("PodVolumeBackup completed")
-	return ctrl.Result{}, nil
+}
+
+func (r *PodVolumeBackupReconciler) OnDataPathFailed(ctx context.Context, namespace string, pvbName string, err error) {
+	defer r.closeDataPath(ctx, pvbName)
+
+	log := r.logger.WithField("pvb", pvbName)
+
+	log.WithError(err).Error("Async fs backup data path failed")
+
+	var pvb velerov1api.PodVolumeBackup
+	if err := r.Client.Get(ctx, types.NamespacedName{Name: pvbName, Namespace: namespace}, &pvb); err != nil {
+		log.WithError(err).Warn("Failed to get PVB on failure")
+	} else {
+		_, _ = r.errorOut(ctx, &pvb, err, "data path backup failed", log)
+	}
+}
+
+func (r *PodVolumeBackupReconciler) OnDataPathCancelled(ctx context.Context, namespace string, pvbName string) {
+	defer r.closeDataPath(ctx, pvbName)
+
+	log := r.logger.WithField("pvb", pvbName)
+
+	log.Warn("Async fs backup data path canceled")
+
+	var pvb velerov1api.PodVolumeBackup
+	if err := r.Client.Get(ctx, types.NamespacedName{Name: pvbName, Namespace: namespace}, &pvb); err != nil {
+		log.WithError(err).Warn("Failed to get PVB on cancel")
+	} else {
+		_, _ = r.errorOut(ctx, &pvb, err, "data path backup canceled", log)
+	}
+}
+
+func (r *PodVolumeBackupReconciler) OnDataPathProgress(ctx context.Context, namespace string, pvbName string, progress *uploader.Progress) {
+	log := r.logger.WithField("pvb", pvbName)
+
+	var pvb velerov1api.PodVolumeBackup
+	if err := r.Client.Get(ctx, types.NamespacedName{Name: pvbName, Namespace: namespace}, &pvb); err != nil {
+		log.WithError(err).Warn("Failed to get PVB on progress")
+		return
+	}
+
+	original := pvb.DeepCopy()
+	pvb.Status.Progress = velerov1api.PodVolumeOperationProgress{TotalBytes: progress.TotalBytes, BytesDone: progress.BytesDone}
+
+	if err := r.Client.Patch(ctx, &pvb, client.MergeFrom(original)); err != nil {
+		log.WithError(err).Error("Failed to update progress")
+	}
 }
 
 // SetupWithManager registers the PVB controller.
@@ -279,36 +336,32 @@ func (r *PodVolumeBackupReconciler) getParentSnapshot(ctx context.Context, log l
 	return mostRecentPVB.Status.SnapshotID
 }
 
-func (r *PodVolumeBackupReconciler) updateStatusToFailed(ctx context.Context, pvb *velerov1api.PodVolumeBackup, err error, msg string, log logrus.FieldLogger) (ctrl.Result, error) {
-	if err = UpdatePVBStatusToFailed(ctx, r.Client, pvb, errors.WithMessage(err, msg).Error(), r.Clock.Now()); err != nil {
-		log.WithError(err).Error("error updating PodVolumeBackup status")
-		return ctrl.Result{}, err
+func (r *PodVolumeBackupReconciler) closeDataPath(ctx context.Context, pvbName string) {
+	fsBackup := r.dataPathMgr.GetAsyncBR(pvbName)
+	if fsBackup != nil {
+		fsBackup.Close(ctx)
 	}
-	return ctrl.Result{}, nil
+
+	r.dataPathMgr.RemoveAsyncBR(pvbName)
 }
 
-func UpdatePVBStatusToFailed(ctx context.Context, c client.Client, pvb *velerov1api.PodVolumeBackup, errString string, time time.Time) error {
+func (r *PodVolumeBackupReconciler) errorOut(ctx context.Context, pvb *velerov1api.PodVolumeBackup, err error, msg string, log logrus.FieldLogger) (ctrl.Result, error) {
+	r.closeDataPath(ctx, pvb.Name)
+	_ = UpdatePVBStatusToFailed(ctx, r.Client, pvb, errors.WithMessage(err, msg).Error(), r.clock.Now(), log)
+
+	return ctrl.Result{}, err
+}
+
+func UpdatePVBStatusToFailed(ctx context.Context, c client.Client, pvb *velerov1api.PodVolumeBackup, errString string, time time.Time, log logrus.FieldLogger) error {
 	original := pvb.DeepCopy()
 	pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseFailed
 	pvb.Status.Message = errString
 	pvb.Status.CompletionTimestamp = &metav1.Time{Time: time}
 
-	return c.Patch(ctx, pvb, client.MergeFrom(original))
-}
-
-func (r *PodVolumeBackupReconciler) NewBackupProgressUpdater(ctx context.Context, pvb *velerov1api.PodVolumeBackup, log logrus.FieldLogger) *BackupProgressUpdater {
-	return &BackupProgressUpdater{pvb, log, ctx, r.Client}
-}
-
-// UpdateProgress which implement ProgressUpdater interface to update pvb progress status
-func (b *BackupProgressUpdater) UpdateProgress(p *uploader.Progress) {
-	original := b.PodVolumeBackup.DeepCopy()
-	b.PodVolumeBackup.Status.Progress = velerov1api.PodVolumeOperationProgress{TotalBytes: p.TotalBytes, BytesDone: p.BytesDone}
-	if b.Cli == nil {
-		b.Log.Errorf("failed to update backup pod %s volume %s progress with uninitailize client", b.PodVolumeBackup.Spec.Pod.Name, b.PodVolumeBackup.Spec.Volume)
-		return
-	}
-	if err := b.Cli.Patch(b.Ctx, b.PodVolumeBackup, client.MergeFrom(original)); err != nil {
-		b.Log.Errorf("update backup pod %s volume %s progress with %v", b.PodVolumeBackup.Spec.Pod.Name, b.PodVolumeBackup.Spec.Volume, err)
+	if err := c.Patch(ctx, pvb, client.MergeFrom(original)); err != nil {
+		log.WithError(err).Error("error updating PodVolumeBackup status")
+		return err
+	} else {
+		return nil
 	}
 }

pkg/controller/pod_volume_backup_controller_test.go

@@ -30,6 +30,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/utils/clock"
 	testclocks "k8s.io/utils/clock/testing"
 	ctrl "sigs.k8s.io/controller-runtime"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -38,10 +39,10 @@ import (
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"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"
-	"github.com/vmware-tanzu/velero/pkg/uploader"
-	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 )
 
 const name = "pvb-1"
@@ -92,6 +93,38 @@ func buildBackupRepo() *velerov1api.BackupRepository {
 	}
 }
 
+type fakeFSBR struct {
+	pvb    *velerov1api.PodVolumeBackup
+	client kbclient.Client
+	clock  clock.WithTickerAndDelayedExecution
+}
+
+func (b *fakeFSBR) Init(ctx context.Context, bslName string, sourceNamespace string, uploaderType string, repositoryType string, repositoryEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter) error {
+	return nil
+}
+
+func (b *fakeFSBR) StartBackup(source datapath.AccessPoint, parentSnapshot string, forceFull bool, tags map[string]string) error {
+	pvb := b.pvb
+
+	original := b.pvb.DeepCopy()
+	pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseCompleted
+	pvb.Status.CompletionTimestamp = &metav1.Time{Time: b.clock.Now()}
+
+	b.client.Patch(ctx, pvb, kbclient.MergeFrom(original))
+
+	return nil
+}
+
+func (b *fakeFSBR) StartRestore(snapshotID string, target datapath.AccessPoint) error {
+	return nil
+}
+
+func (b *fakeFSBR) Cancel() {
+}
+
+func (b *fakeFSBR) Close(ctx context.Context) {
+}
+
 var _ = Describe("PodVolumeBackup Reconciler", func() {
 	type request struct {
 		pvb               *velerov1api.PodVolumeBackup
@@ -102,6 +135,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 		expected          *velerov1api.PodVolumeBackup
 		expectedRequeue   ctrl.Result
 		expectedErrMsg    string
+		dataMgr           *datapath.Manager
 	}
 
 	// `now` will be used to set the fake clock's time; capture
@@ -141,20 +175,31 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 
 			Expect(err).To(BeNil())
 
+			if test.dataMgr == nil {
+				test.dataMgr = datapath.NewManager(1)
+			}
+
+			datapath.FSBRCreator = func(string, string, kbclient.Client, string, datapath.Callbacks, logrus.FieldLogger) datapath.AsyncBR {
+				return &fakeFSBR{
+					pvb:    test.pvb,
+					client: fakeClient,
+					clock:  testclocks.NewFakeClock(now),
+				}
+			}
+
 			// Setup reconciler
 			Expect(velerov1api.AddToScheme(scheme.Scheme)).To(Succeed())
 			r := PodVolumeBackupReconciler{
 				Client:           fakeClient,
-				Clock:            testclocks.NewFakeClock(now),
-				Metrics:          metrics.NewPodVolumeMetrics(),
-				CredentialGetter: &credentials.CredentialGetter{FromFile: credentialFileStore},
-				NodeName:         "test_node",
-				FileSystem:       fakeFS,
-				Log:              velerotest.NewLogger(),
-			}
-			NewUploaderProviderFunc = func(ctx context.Context, client kbclient.Client, uploaderType, repoIdentifier string, bsl *velerov1api.BackupStorageLocation, backupRepo *velerov1api.BackupRepository, credGetter *credentials.CredentialGetter, repoKeySelector *corev1.SecretKeySelector, log logrus.FieldLogger) (provider.Provider, error) {
-				return &fakeProvider{}, nil
+				clock:            testclocks.NewFakeClock(now),
+				metrics:          metrics.NewPodVolumeMetrics(),
+				credentialGetter: &credentials.CredentialGetter{FromFile: credentialFileStore},
+				nodeName:         "test_node",
+				fileSystem:       fakeFS,
+				logger:           velerotest.NewLogger(),
+				dataPathMgr:      test.dataMgr,
 			}
+
 			actualResult, err := r.Reconcile(ctx, ctrl.Request{
 				NamespacedName: types.NamespacedName{
 					Namespace: velerov1api.DefaultNamespace,
@@ -328,29 +373,17 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			expectedRequeue: ctrl.Result{},
 		}),
+		Entry("pvb should be requeued when exceeding max concurrent number", request{
+			pvb:               pvbBuilder().Phase("").Node("test_node").Result(),
+			pod:               podBuilder().Result(),
+			bsl:               bslBuilder().Result(),
+			backupRepo:        buildBackupRepo(),
+			dataMgr:           datapath.NewManager(0),
+			expectedProcessed: false,
+			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
+				Phase("").
+				Result(),
+			expectedRequeue: ctrl.Result{Requeue: true, RequeueAfter: time.Minute},
+		}),
 	)
 })
-
-type fakeProvider struct {
-}
-
-func (f *fakeProvider) RunBackup(
-	ctx context.Context,
-	path string,
-	tags map[string]string,
-	parentSnapshot string,
-	updater uploader.ProgressUpdater) (string, bool, error) {
-	return "", false, nil
-}
-
-func (f *fakeProvider) RunRestore(
-	ctx context.Context,
-	snapshotID string,
-	volumePath string,
-	updater uploader.ProgressUpdater) error {
-	return nil
-}
-
-func (f *fakeProvider) Close(ctx context.Context) error {
-	return nil
-}

pkg/controller/pod_volume_restore_controller.go

@@ -39,40 +39,37 @@ import (
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/datapath"
+	"github.com/vmware-tanzu/velero/pkg/exposer"
 	"github.com/vmware-tanzu/velero/pkg/podvolume"
 	"github.com/vmware-tanzu/velero/pkg/repository"
-	repokey "github.com/vmware-tanzu/velero/pkg/repository/keys"
 	"github.com/vmware-tanzu/velero/pkg/restorehelper"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
-	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
-	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
-func NewPodVolumeRestoreReconciler(logger logrus.FieldLogger, client client.Client, credentialGetter *credentials.CredentialGetter) *PodVolumeRestoreReconciler {
+func NewPodVolumeRestoreReconciler(client client.Client, ensurer *repository.Ensurer,
+	credentialGetter *credentials.CredentialGetter, logger logrus.FieldLogger) *PodVolumeRestoreReconciler {
 	return &PodVolumeRestoreReconciler{
-		Client:           client,
-		logger:           logger.WithField("controller", "PodVolumeRestore"),
-		credentialGetter: credentialGetter,
-		fileSystem:       filesystem.NewFileSystem(),
-		clock:            &clocks.RealClock{},
+		Client:            client,
+		logger:            logger.WithField("controller", "PodVolumeRestore"),
+		repositoryEnsurer: ensurer,
+		credentialGetter:  credentialGetter,
+		fileSystem:        filesystem.NewFileSystem(),
+		clock:             &clocks.RealClock{},
+		dataPathMgr:       datapath.NewManager(1),
 	}
 }
 
 type PodVolumeRestoreReconciler struct {
 	client.Client
-	logger           logrus.FieldLogger
-	credentialGetter *credentials.CredentialGetter
-	fileSystem       filesystem.Interface
-	clock            clocks.WithTickerAndDelayedExecution
-}
-
-type RestoreProgressUpdater struct {
-	PodVolumeRestore *velerov1api.PodVolumeRestore
-	Log              logrus.FieldLogger
-	Ctx              context.Context
-	Cli              client.Client
+	logger            logrus.FieldLogger
+	repositoryEnsurer *repository.Ensurer
+	credentialGetter  *credentials.CredentialGetter
+	fileSystem        filesystem.Interface
+	clock             clocks.WithTickerAndDelayedExecution
+	dataPathMgr       *datapath.Manager
 }
 
 // +kubebuilder:rbac:groups=velero.io,resources=podvolumerestores,verbs=get;list;watch;create;update;patch;delete
@@ -113,41 +110,69 @@ func (c *PodVolumeRestoreReconciler) Reconcile(ctx context.Context, req ctrl.Req
 	}
 
 	log.Info("Restore starting")
+
+	callbacks := datapath.Callbacks{
+		OnCompleted: c.OnDataPathCompleted,
+		OnFailed:    c.OnDataPathFailed,
+		OnCancelled: c.OnDataPathCancelled,
+		OnProgress:  c.OnDataPathProgress,
+	}
+
+	fsRestore, err := c.dataPathMgr.CreateFileSystemBR(pvr.Name, pVBRRequestor, ctx, c.Client, pvr.Namespace, callbacks, log)
+	if err != nil {
+		if err == datapath.ConcurrentLimitExceed {
+			return ctrl.Result{Requeue: true, RequeueAfter: time.Minute}, nil
+		} else {
+			return c.errorOut(ctx, pvr, err, "error to create data path", log)
+		}
+	}
+
 	original := pvr.DeepCopy()
 	pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseInProgress
 	pvr.Status.StartTimestamp = &metav1.Time{Time: c.clock.Now()}
 	if err = c.Patch(ctx, pvr, client.MergeFrom(original)); err != nil {
-		log.WithError(err).Error("Unable to update status to in progress")
-		return ctrl.Result{}, err
+		return c.errorOut(ctx, pvr, err, "error to update status to in progress", log)
 	}
 
-	if err = c.processRestore(ctx, pvr, pod, log); err != nil {
-		if e := UpdatePVRStatusToFailed(ctx, c, pvr, err.Error(), c.clock.Now()); e != nil {
-			log.WithError(err).Error("Unable to update status to failed")
-		}
-
-		log.WithError(err).Error("Unable to process the PodVolumeRestore")
-		return ctrl.Result{}, err
+	volumePath, err := exposer.GetPodVolumeHostPath(ctx, pod, pvr.Spec.Volume, c.Client, c.fileSystem, log)
+	if err != nil {
+		return c.errorOut(ctx, pvr, err, "error exposing host path for pod volume", log)
 	}
 
-	original = pvr.DeepCopy()
-	pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseCompleted
-	pvr.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
-	if err = c.Patch(ctx, pvr, client.MergeFrom(original)); err != nil {
-		log.WithError(err).Error("Unable to update status to completed")
-		return ctrl.Result{}, err
+	log.WithField("path", volumePath.ByPath).Debugf("Found host path")
+
+	if err := fsRestore.Init(ctx, pvr.Spec.BackupStorageLocation, pvr.Spec.Pod.Namespace, pvr.Spec.UploaderType,
+		podvolume.GetPvrRepositoryType(pvr), c.repositoryEnsurer, c.credentialGetter); err != nil {
+		return c.errorOut(ctx, pvr, err, "error to initialize data path", log)
 	}
-	log.Info("Restore completed")
+
+	if err := fsRestore.StartRestore(pvr.Spec.SnapshotID, volumePath); err != nil {
+		return c.errorOut(ctx, pvr, err, "error starting data path restore", log)
+	}
+
+	log.WithField("path", volumePath.ByPath).Info("Async fs restore data path started")
+
 	return ctrl.Result{}, nil
 }
 
-func UpdatePVRStatusToFailed(ctx context.Context, c client.Client, pvr *velerov1api.PodVolumeRestore, errString string, time time.Time) error {
-	original := pvr.DeepCopy()
-	pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseFailed
-	pvr.Status.Message = errString
-	pvr.Status.CompletionTimestamp = &metav1.Time{Time: time}
+func (c *PodVolumeRestoreReconciler) errorOut(ctx context.Context, pvr *velerov1api.PodVolumeRestore, err error, msg string, log logrus.FieldLogger) (ctrl.Result, error) {
+	c.closeDataPath(ctx, pvr.Name)
+	_ = UpdatePVRStatusToFailed(ctx, c.Client, pvr, errors.WithMessage(err, msg).Error(), c.clock.Now(), log)
+	return ctrl.Result{}, err
+}
 
-	return c.Patch(ctx, pvr, client.MergeFrom(original))
+func UpdatePVRStatusToFailed(ctx context.Context, c client.Client, pvb *velerov1api.PodVolumeRestore, errString string, time time.Time, log logrus.FieldLogger) error {
+	original := pvb.DeepCopy()
+	pvb.Status.Phase = velerov1api.PodVolumeRestorePhaseFailed
+	pvb.Status.Message = errString
+	pvb.Status.CompletionTimestamp = &metav1.Time{Time: time}
+
+	if err := c.Patch(ctx, pvb, client.MergeFrom(original)); err != nil {
+		log.WithError(err).Error("error updating PodVolumeRestore status")
+		return err
+	} else {
+		return nil
+	}
 }
 
 func (c *PodVolumeRestoreReconciler) shouldProcess(ctx context.Context, log logrus.FieldLogger, pvr *velerov1api.PodVolumeRestore) (bool, *corev1api.Pod, error) {
@@ -232,54 +257,23 @@ func getInitContainerIndex(pod *corev1api.Pod) int {
 	return -1
 }
 
-func (c *PodVolumeRestoreReconciler) processRestore(ctx context.Context, req *velerov1api.PodVolumeRestore, pod *corev1api.Pod, log logrus.FieldLogger) error {
-	volumeDir, err := kube.GetVolumeDirectory(ctx, log, pod, req.Spec.Volume, c.Client)
-	if err != nil {
-		return errors.Wrap(err, "error getting volume directory name")
+func (c *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, namespace string, pvrName string, result datapath.Result) {
+	defer c.closeDataPath(ctx, pvrName)
+
+	log := c.logger.WithField("pvr", pvrName)
+
+	log.WithField("PVR", pvrName).Info("Async fs restore data path completed")
+
+	var pvr velerov1api.PodVolumeRestore
+	if err := c.Client.Get(ctx, types.NamespacedName{Name: pvrName, Namespace: namespace}, &pvr); err != nil {
+		log.WithError(err).Warn("Failed to get PVR on completion")
+		return
 	}
 
-	// Get the full path of the new volume's directory as mounted in the daemonset pod, which
-	// will look like: /host_pods/<new-pod-uid>/volumes/<volume-plugin-name>/<volume-dir>
-	volumePath, err := kube.SinglePathMatch(
-		fmt.Sprintf("/host_pods/%s/volumes/*/%s", string(req.Spec.Pod.UID), volumeDir),
-		c.fileSystem, log)
-	if err != nil {
-		return errors.Wrap(err, "error identifying path of volume")
-	}
-
-	backupLocation := &velerov1api.BackupStorageLocation{}
-	if err := c.Get(ctx, client.ObjectKey{
-		Namespace: req.Namespace,
-		Name:      req.Spec.BackupStorageLocation,
-	}, backupLocation); err != nil {
-		return errors.Wrap(err, "error getting backup storage location")
-	}
-
-	// need to check backup repository in source namespace rather than in pod namespace
-	// such as in case of namespace mapping issue
-	backupRepo, err := repository.GetBackupRepository(ctx, c.Client, req.Namespace, repository.BackupRepositoryKey{
-		VolumeNamespace: req.Spec.SourceNamespace,
-		BackupLocation:  req.Spec.BackupStorageLocation,
-		RepositoryType:  podvolume.GetPvrRepositoryType(req),
-	})
-	if err != nil {
-		return errors.Wrap(err, "error getting backup repository")
-	}
-
-	uploaderProv, err := provider.NewUploaderProvider(ctx, c.Client, req.Spec.UploaderType,
-		req.Spec.RepoIdentifier, backupLocation, backupRepo, c.credentialGetter, repokey.RepoKeySelector(), log)
-	if err != nil {
-		return errors.Wrap(err, "error creating uploader")
-	}
-
-	defer func() {
-		if err := uploaderProv.Close(ctx); err != nil {
-			log.Errorf("failed to close uploader provider with error %v", err)
-		}
-	}()
-
-	if err = uploaderProv.RunRestore(ctx, req.Spec.SnapshotID, volumePath, c.NewRestoreProgressUpdater(ctx, req, log)); err != nil {
-		return errors.Wrapf(err, "error running restore err=%v", err)
+	volumePath := result.Restore.Target.ByPath
+	if volumePath == "" {
+		_, _ = c.errorOut(ctx, &pvr, errors.New("path is empty"), "invalid restore target", log)
+		return
 	}
 
 	// Remove the .velero directory from the restored volume (it may contain done files from previous restores
@@ -291,7 +285,7 @@ func (c *PodVolumeRestoreReconciler) processRestore(ctx context.Context, req *ve
 	}
 
 	var restoreUID types.UID
-	for _, owner := range req.OwnerReferences {
+	for _, owner := range pvr.OwnerReferences {
 		if boolptr.IsSetToTrue(owner.Controller) {
 			restoreUID = owner.UID
 			break
@@ -301,32 +295,80 @@ func (c *PodVolumeRestoreReconciler) processRestore(ctx context.Context, req *ve
 	// Create the .velero directory within the volume dir so we can write a done file
 	// for this restore.
 	if err := os.MkdirAll(filepath.Join(volumePath, ".velero"), 0755); err != nil {
-		return errors.Wrap(err, "error creating .velero directory for done file")
+		_, _ = c.errorOut(ctx, &pvr, err, "error creating .velero directory for done file", log)
+		return
 	}
 
 	// Write a done file with name=<restore-uid> into the just-created .velero dir
 	// within the volume. The velero init container on the pod is waiting
 	// for this file to exist in each restored volume before completing.
 	if err := os.WriteFile(filepath.Join(volumePath, ".velero", string(restoreUID)), nil, 0644); err != nil { //nolint:gosec
-		return errors.Wrap(err, "error writing done file")
-	}
-
-	return nil
-}
-
-func (c *PodVolumeRestoreReconciler) NewRestoreProgressUpdater(ctx context.Context, pvr *velerov1api.PodVolumeRestore, log logrus.FieldLogger) *RestoreProgressUpdater {
-	return &RestoreProgressUpdater{pvr, log, ctx, c.Client}
-}
-
-// UpdateProgress which implement ProgressUpdater interface to update pvr progress status
-func (c *RestoreProgressUpdater) UpdateProgress(p *uploader.Progress) {
-	original := c.PodVolumeRestore.DeepCopy()
-	c.PodVolumeRestore.Status.Progress = velerov1api.PodVolumeOperationProgress{TotalBytes: p.TotalBytes, BytesDone: p.BytesDone}
-	if c.Cli == nil {
-		c.Log.Errorf("failed to update restore pod %s volume %s progress with uninitailize client", c.PodVolumeRestore.Spec.Pod.Name, c.PodVolumeRestore.Spec.Volume)
+		_, _ = c.errorOut(ctx, &pvr, err, "error writing done file", log)
 		return
 	}
-	if err := c.Cli.Patch(c.Ctx, c.PodVolumeRestore, client.MergeFrom(original)); err != nil {
-		c.Log.Errorf("update restore pod %s volume %s progress with %v", c.PodVolumeRestore.Spec.Pod.Name, c.PodVolumeRestore.Spec.Volume, err)
+
+	original := pvr.DeepCopy()
+	pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseCompleted
+	pvr.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
+	if err := c.Patch(ctx, &pvr, client.MergeFrom(original)); err != nil {
+		log.WithError(err).Error("error updating PodVolumeRestore status")
+	}
+
+	log.Info("Restore completed")
+}
+
+func (c *PodVolumeRestoreReconciler) OnDataPathFailed(ctx context.Context, namespace string, pvrName string, err error) {
+	defer c.closeDataPath(ctx, pvrName)
+
+	log := c.logger.WithField("pvr", pvrName)
+
+	log.WithError(err).Info("Async fs restore data path failed")
+
+	var pvr velerov1api.PodVolumeRestore
+	if err := c.Client.Get(ctx, types.NamespacedName{Name: pvrName, Namespace: namespace}, &pvr); err != nil {
+		log.WithError(err).Warn("Failed to get PVR on failure")
+	} else {
+		_, _ = c.errorOut(ctx, &pvr, err, "data path restore failed", log)
 	}
 }
+
+func (c *PodVolumeRestoreReconciler) OnDataPathCancelled(ctx context.Context, namespace string, pvrName string) {
+	defer c.closeDataPath(ctx, pvrName)
+
+	log := c.logger.WithField("pvr", pvrName)
+
+	log.Info("Async fs restore data path canceled")
+
+	var pvr velerov1api.PodVolumeRestore
+	if err := c.Client.Get(ctx, types.NamespacedName{Name: pvrName, Namespace: namespace}, &pvr); err != nil {
+		log.WithError(err).Warn("Failed to get PVR on cancel")
+	} else {
+		_, _ = c.errorOut(ctx, &pvr, err, "data path restore canceled", log)
+	}
+}
+
+func (c *PodVolumeRestoreReconciler) OnDataPathProgress(ctx context.Context, namespace string, pvrName string, progress *uploader.Progress) {
+	log := c.logger.WithField("pvr", pvrName)
+
+	var pvr velerov1api.PodVolumeRestore
+	if err := c.Client.Get(ctx, types.NamespacedName{Name: pvrName, Namespace: namespace}, &pvr); err != nil {
+		log.WithError(err).Warn("Failed to get PVB on progress")
+		return
+	}
+
+	original := pvr.DeepCopy()
+	pvr.Status.Progress = velerov1api.PodVolumeOperationProgress{TotalBytes: progress.TotalBytes, BytesDone: progress.BytesDone}
+
+	if err := c.Client.Patch(ctx, &pvr, client.MergeFrom(original)); err != nil {
+		log.WithError(err).Error("Failed to update progress")
+	}
+}
+
+func (c *PodVolumeRestoreReconciler) closeDataPath(ctx context.Context, pvbName string) {
+	fsRestore := c.dataPathMgr.GetAsyncBR(pvbName)
+	if fsRestore != nil {
+		fsRestore.Close(ctx)
+	}
+
+	c.dataPathMgr.RemoveAsyncBR(pvbName)
+}