Update Restore Exposer and PVC CSI to support in-place restore (#10104)

1. Update Restore Exposer to support exposing with existing PV for in-place restore
2. Update PVC CSI RIA to continue the restore process for in-place restore

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

pkg/apis/velero/v1/restore_types.go

@@ -154,10 +154,6 @@ type RestoreSpec struct {
 	UploaderConfig *UploaderConfigForRestore `json:"uploaderConfig,omitempty"`
 }
 
-func (r *RestoreSpec) IsVolumeDataInplaceRestore() bool {
-	return r.ExistingVolumeDataPolicy == VolumeDataPolicyTypeFull || r.ExistingVolumeDataPolicy == VolumeDataPolicyTypeIncremental
-}
-
 // UploaderConfigForRestore defines the configuration for the restore.
 type UploaderConfigForRestore struct {
 	// WriteSparseFiles is a flag to indicate whether write files sparsely or not.
@@ -464,6 +460,10 @@ type Restore struct {
 	Status RestoreStatus `json:"status,omitempty"`
 }
 
+func (r *Restore) IsVolumeDataInplaceRestore() bool {
+	return r.Spec.ExistingVolumeDataPolicy == VolumeDataPolicyTypeFull || r.Spec.ExistingVolumeDataPolicy == VolumeDataPolicyTypeIncremental
+}
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // RestoreList is a list of Restores.

pkg/apis/velero/v1/restore_types_test.go

@@ -0,0 +1,69 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	"testing"
+)
+
+func TestIsVolumeDataInplaceRestore(t *testing.T) {
+	tests := []struct {
+		name                     string
+		existingVolumeDataPolicy VolumeDataPolicyType
+		expected                 bool
+	}{
+		{
+			name:                     "empty policy",
+			existingVolumeDataPolicy: "",
+			expected:                 false,
+		},
+		{
+			name:                     "none policy",
+			existingVolumeDataPolicy: VolumeDataPolicyTypeNone,
+			expected:                 false,
+		},
+		{
+			name:                     "full policy",
+			existingVolumeDataPolicy: VolumeDataPolicyTypeFull,
+			expected:                 true,
+		},
+		{
+			name:                     "incremental policy",
+			existingVolumeDataPolicy: VolumeDataPolicyTypeIncremental,
+			expected:                 true,
+		},
+		{
+			name:                     "unknown policy",
+			existingVolumeDataPolicy: VolumeDataPolicyType("unknown"),
+			expected:                 false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			restore := &Restore{
+				Spec: RestoreSpec{
+					ExistingVolumeDataPolicy: tc.existingVolumeDataPolicy,
+				},
+			}
+			actual := restore.IsVolumeDataInplaceRestore()
+			if actual != tc.expected {
+				t.Errorf("expected %v, got %v", tc.expected, actual)
+			}
+		})
+	}
+}

pkg/controller/data_download_controller.go

@@ -936,6 +936,7 @@ func (r *DataDownloadReconciler) setupExposeParam(dd *velerov2alpha1api.DataDown
 
 	return exposer.GenericRestoreExposeParam{
 		TargetPVCName:         dd.Spec.TargetVolume.PVC,
+		TargetPVName:          dd.Spec.TargetVolume.PV,
 		TargetNamespace:       dd.Spec.TargetVolume.Namespace,
 		HostingPodLabels:      hostingPodLabels,
 		HostingPodAnnotations: hostingPodAnnotation,

pkg/controller/data_download_controller_test.go

@@ -61,7 +61,6 @@ func dataDownloadBuilder() *builder.DataDownloadBuilder {
 		BackupStorageLocation("bsl-loc").
 		DataMover("velero").
 		SnapshotID("test-snapshot-id").TargetVolume(velerov2alpha1api.TargetVolumeSpec{
-		PV:        "test-pv",
 		PVC:       "test-pvc",
 		Namespace: "test-ns",
 	})
@@ -183,6 +182,7 @@ func TestDataDownloadReconcile(t *testing.T) {
 		dd                       *velerov2alpha1api.DataDownload
 		notCreateDD              bool
 		targetPVC                *corev1api.PersistentVolumeClaim
+		targetPV                 *corev1api.PersistentVolume
 		dataMgr                  *datapath.Manager
 		needErrs                 []bool
 		needCreateFSBR           bool
@@ -194,6 +194,7 @@ func TestDataDownloadReconcile(t *testing.T) {
 		isPeekExposeErr          bool
 		isNilExposer             bool
 		notNilExpose             bool
+		mockExpose               bool
 		notMockCleanUp           bool
 		mockInit                 bool
 		mockInitErr              error
@@ -351,6 +352,16 @@ func TestDataDownloadReconcile(t *testing.T) {
 			targetPVC: builder.ForPersistentVolumeClaim("test-ns", "test-pvc").StorageClass("sc").Result(),
 			expected:  dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).Phase(velerov2alpha1api.DataDownloadPhaseAccepted).Result(),
 		},
+		{
+			name:           "dd succeeds for accepted with target PV set",
+			dd:             dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).TargetVolume(velerov2alpha1api.TargetVolumeSpec{PVC: "test-pvc", Namespace: "test-ns", PV: "test-pv"}).Result(),
+			targetPVC:      builder.ForPersistentVolumeClaim("test-ns", "test-pvc").StorageClass("sc").Result(),
+			targetPV:       builder.ForPersistentVolume("test-pv").Result(),
+			expected:       dataDownloadBuilder().Finalizers([]string{DataUploadDownloadFinalizer}).TargetVolume(velerov2alpha1api.TargetVolumeSpec{PVC: "test-pvc", Namespace: "test-ns", PV: "test-pv"}).Phase(velerov2alpha1api.DataDownloadPhaseAccepted).Result(),
+			mockExpose:     true,
+			notMockCleanUp: true,
+			notNilExpose:   true,
+		},
 		{
 			name:     "prepare timeout on accepted",
 			dd:       dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseAccepted).Finalizers([]string{DataUploadDownloadFinalizer}).AcceptedTimestamp(&metav1.Time{Time: time.Now().Add(-time.Minute * 30)}).Result(),
@@ -487,6 +498,10 @@ func TestDataDownloadReconcile(t *testing.T) {
 				objects = append(objects, test.targetPVC)
 			}
 
+			if test.targetPV != nil {
+				objects = append(objects, test.targetPV)
+			}
+
 			r, err := initDataDownloadReconciler(t, objects, test.needErrs...)
 			require.NoError(t, err)
 
@@ -543,7 +558,7 @@ func TestDataDownloadReconcile(t *testing.T) {
 				return asyncBR
 			}
 
-			if test.isExposeErr || test.isGetExposeErr || test.isGetExposeNil || test.isPeekExposeErr || test.isNilExposer || test.notNilExpose {
+			if test.isExposeErr || test.isGetExposeErr || test.isGetExposeNil || test.isPeekExposeErr || test.isNilExposer || test.notNilExpose || test.mockExpose {
 				if test.isNilExposer {
 					r.restoreExposer = nil
 				} else {
@@ -551,6 +566,8 @@ func TestDataDownloadReconcile(t *testing.T) {
 						ep := exposermockes.NewGenericRestoreExposer(t)
 						if test.isExposeErr {
 							ep.On("Expose", mock.Anything, mock.Anything, mock.Anything).Return(errors.New("Error to expose restore exposer"))
+						} else if test.mockExpose {
+							ep.On("Expose", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 						} else if test.notNilExpose {
 							hostingPod := builder.ForPod("test-ns", "test-name").Volumes(&corev1api.Volume{Name: "test-pvc"}).Result()
 							hostingPod.ObjectMeta.SetUID("test-uid")
@@ -1319,6 +1336,7 @@ func TestDataDownloadSetupExposeParam(t *testing.T) {
 
 	baseDataDownload := dataDownloadBuilder().Result()
 	baseDataDownload.Namespace = velerov1api.DefaultNamespace
+	baseDataDownload.Spec.TargetVolume.PV = "pv-1"
 	baseDataDownload.Spec.OperationTimeout = metav1.Duration{Duration: time.Minute * 10}
 	baseDataDownload.Spec.SnapshotSize = 5368709120 // 5Gi
 
@@ -1428,6 +1446,7 @@ func TestDataDownloadSetupExposeParam(t *testing.T) {
 
 			// Core fields
 			assert.Equal(t, baseDataDownload.Spec.TargetVolume.PVC, got.TargetPVCName)
+			assert.Equal(t, baseDataDownload.Spec.TargetVolume.PV, got.TargetPVName)
 			assert.Equal(t, baseDataDownload.Spec.TargetVolume.Namespace, got.TargetNamespace)
 			assert.Equal(t, baseDataDownload.Spec.DataMover, got.DataMover)
 

pkg/exposer/csi_snapshot_test.go

@@ -1230,6 +1230,9 @@ func TestGetExpose(t *testing.T) {
 		Spec: corev1api.PersistentVolumeClaimSpec{
 			VolumeName: "fake-pv-name",
 		},
+		Status: corev1api.PersistentVolumeClaimStatus{
+			Phase: corev1api.ClaimBound,
+		},
 	}
 
 	backupPV := &corev1api.PersistentVolume{

pkg/exposer/generic_restore.go

@@ -43,6 +43,9 @@ type GenericRestoreExposeParam struct {
 	// TargetPVCName is the target volume name to be restored
 	TargetPVCName string
 
+	// TargetPVName is the target persistent volume name to be restored
+	TargetPVName string
+
 	// TargetNamespace is the namespace of the volume to be restored
 	TargetNamespace string
 
@@ -144,9 +147,11 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
 	curLog := e.log.WithFields(logrus.Fields{
 		"owner":            ownerObject.Name,
 		"target PVC":       param.TargetPVCName,
+		"target PV":        param.TargetPVName,
 		"target namespace": param.TargetNamespace,
 	})
 
+	curLog.Info("Waiting for target PVC to be consumed")
 	selectedNode, targetPVC, err := kube.WaitPVCConsumed(
 		ctx,
 		e.kubeClient.CoreV1(),
@@ -196,7 +201,16 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
 		}
 	}
 
-	restorePVC, err := e.createRestorePVC(ctx, ownerObject, targetPVC, selectedNode, param.DataMover)
+	curLog.Info("Creating restore PVC")
+
+	var targetPV *corev1api.PersistentVolume
+	if len(param.TargetPVName) > 0 {
+		targetPV, err = e.kubeClient.CoreV1().PersistentVolumes().Get(ctx, param.TargetPVName, metav1.GetOptions{})
+		if err != nil {
+			return errors.Wrapf(err, "fail to get the target PV %s", param.TargetPVName)
+		}
+	}
+	restorePVC, err := e.createRestorePVC(ctx, ownerObject, targetPVC, targetPV, selectedNode, param.DataMover, param.ExposeTimeout)
 	if err != nil {
 		return errors.Wrap(err, "error to create restore pvc")
 	}
@@ -205,10 +219,17 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
 
 	defer func() {
 		if err != nil {
-			kube.DeletePVAndPVCIfAny(ctx, e.kubeClient.CoreV1(), restorePVC.Name, restorePVC.Namespace, 0, curLog)
+			if len(param.TargetPVName) == 0 {
+				kube.DeletePVAndPVCIfAny(ctx, e.kubeClient.CoreV1(), restorePVC.Name, restorePVC.Namespace, 0, curLog)
+			} else {
+				// cannot delete PV if param.TargetPVName is set because the PV is not created by the Expose process.
+				// It's the existing PV used for in-place restore.
+				kube.DeletePVCIfAny(ctx, e.kubeClient.CoreV1(), restorePVC.Name, restorePVC.Namespace, 0, curLog)
+			}
 		}
 	}()
 
+	curLog.Info("Creating restore pod")
 	restorePod, err := e.createRestorePod(
 		ctx,
 		ownerObject,
@@ -808,7 +829,7 @@ func (e *genericRestoreExposer) createRestorePod(
 	return e.kubeClient.CoreV1().Pods(ownerObject.Namespace).Create(ctx, pod, metav1.CreateOptions{})
 }
 
-func (e *genericRestoreExposer) createRestorePVC(ctx context.Context, ownerObject corev1api.ObjectReference, targetPVC *corev1api.PersistentVolumeClaim, selectedNode string, dataMover string) (*corev1api.PersistentVolumeClaim, error) {
+func (e *genericRestoreExposer) createRestorePVC(ctx context.Context, ownerObject corev1api.ObjectReference, targetPVC *corev1api.PersistentVolumeClaim, targetPV *corev1api.PersistentVolume, selectedNode string, dataMover string, operationTimeout time.Duration) (*corev1api.PersistentVolumeClaim, error) {
 	restorePVCName := ownerObject.Name
 
 	pvcObj := &corev1api.PersistentVolumeClaim{
@@ -834,11 +855,15 @@ func (e *genericRestoreExposer) createRestorePVC(ctx context.Context, ownerObjec
 			Resources:        targetPVC.Spec.Resources,
 		},
 	}
+	if targetPV != nil {
+		pvcObj.Spec.VolumeName = targetPV.Name
+	}
 
 	if selectedNode != "" {
-		pvcObj.Annotations = map[string]string{
-			kube.KubeAnnSelectedNode: selectedNode,
+		if pvcObj.Annotations == nil {
+			pvcObj.Annotations = make(map[string]string)
 		}
+		pvcObj.Annotations[kube.KubeAnnSelectedNode] = selectedNode
 	}
 
 	if dataMover == datamover.DataMoverTypeVeleroBlock {
@@ -849,5 +874,20 @@ func (e *genericRestoreExposer) createRestorePVC(ctx context.Context, ownerObjec
 		*pvcObj.Spec.VolumeMode = corev1api.PersistentVolumeBlock
 	}
 
-	return e.kubeClient.CoreV1().PersistentVolumeClaims(pvcObj.Namespace).Create(ctx, pvcObj, metav1.CreateOptions{})
+	restorePVC, err := e.kubeClient.CoreV1().PersistentVolumeClaims(pvcObj.Namespace).Create(ctx, pvcObj, metav1.CreateOptions{})
+	if err != nil {
+		return nil, errors.Wrapf(err, "fail to create the restore PVC %s in namespace %s", pvcObj.Name, pvcObj.Namespace)
+	}
+
+	if targetPV != nil {
+		if _, err = kube.ResetPVBinding(ctx, e.kubeClient.CoreV1(), targetPV, nil, restorePVC); err != nil {
+			return nil, errors.Wrapf(err, "fail to reset PV %s binding to restore PVC %s/%s", targetPV.Name, restorePVC.Namespace, restorePVC.Name)
+		}
+
+		if _, err = kube.WaitPVCBound(ctx, e.kubeClient.CoreV1(), e.kubeClient.CoreV1(), restorePVC.Name, restorePVC.Namespace, operationTimeout); err != nil {
+			return nil, errors.Wrapf(err, "fail to wait restore PVC %s/%s bound", restorePVC.Namespace, restorePVC.Name)
+		}
+	}
+
+	return restorePVC, nil
 }

pkg/exposer/generic_restore_test.go

@@ -61,6 +61,11 @@ func TestRestoreExpose(t *testing.T) {
 			StorageClassName: &scName,
 		},
 	}
+	targetPVObj := &corev1api.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-target-pv",
+		},
+	}
 
 	modeFilesystem := corev1api.PersistentVolumeFilesystem
 	targetPVCObjWithVolumeMode := &corev1api.PersistentVolumeClaim{
@@ -118,6 +123,7 @@ func TestRestoreExpose(t *testing.T) {
 		ownerRestore    *velerov1.Restore
 		targetPVCName   string
 		targetNamespace string
+		targetPVName    string
 		kubeReactors    []reactor
 		cacheVolume     *CacheConfigs
 		dataMover       string
@@ -184,7 +190,7 @@ func TestRestoreExpose(t *testing.T) {
 					},
 				},
 			},
-			err: "error to create restore pvc: fake-create-error",
+			err: "error to create restore pvc: fail to create the restore PVC fake-restore in namespace velero: fake-create-error",
 		},
 		{
 			name:            "succeed",
@@ -199,6 +205,45 @@ func TestRestoreExpose(t *testing.T) {
 			expectBackupPod: true,
 			expectBackupPVC: true,
 		},
+		{
+			name:            "succeed with target PV set",
+			targetPVCName:   "fake-target-pvc",
+			targetNamespace: "fake-ns",
+			targetPVName:    "fake-target-pv",
+			ownerRestore:    restore,
+			kubeClientObj: []runtime.Object{
+				targetPVCObj,
+				targetPVObj,
+				daemonSet,
+				storageClass,
+			},
+			kubeReactors: []reactor{
+				{
+					verb:     "get",
+					resource: "persistentvolumeclaims",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						getAction := action.(clientTesting.GetAction)
+						if getAction.GetName() == "fake-restore" {
+							return true, &corev1api.PersistentVolumeClaim{
+								ObjectMeta: metav1.ObjectMeta{
+									Name:      "fake-restore",
+									Namespace: velerov1.DefaultNamespace,
+								},
+								Spec: corev1api.PersistentVolumeClaimSpec{
+									VolumeName: "fake-target-pv",
+								},
+								Status: corev1api.PersistentVolumeClaimStatus{
+									Phase: corev1api.ClaimBound,
+								},
+							}, nil
+						}
+						return false, nil, nil
+					},
+				},
+			},
+			expectBackupPod: true,
+			expectBackupPVC: true,
+		},
 		{
 			name:            "succeed, cache config, no cache volume",
 			targetPVCName:   "fake-target-pvc",
@@ -310,6 +355,7 @@ func TestRestoreExpose(t *testing.T) {
 				GenericRestoreExposeParam{
 					TargetPVCName:    test.targetPVCName,
 					TargetNamespace:  test.targetNamespace,
+					TargetPVName:     test.targetPVName,
 					HostingPodLabels: map[string]string{},
 					Resources:        corev1api.ResourceRequirements{},
 					ExposeTimeout:    time.Millisecond,
@@ -397,6 +443,9 @@ func TestRebindVolume(t *testing.T) {
 		Spec: corev1api.PersistentVolumeClaimSpec{
 			VolumeName: "fake-restore-pv",
 		},
+		Status: corev1api.PersistentVolumeClaimStatus{
+			Phase: corev1api.ClaimBound,
+		},
 	}
 
 	restorePVObj := &corev1api.PersistentVolume{

pkg/podvolume/restorer.go

@@ -297,6 +297,10 @@ func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, backu
 		pvr.Spec.UploaderSettings = uploaderutil.StoreRestoreConfig(restore.Spec.UploaderConfig)
 	}
 
+	if restore.IsVolumeDataInplaceRestore() {
+		pvr.Spec.RestoreType = string(restore.Spec.ExistingVolumeDataPolicy)
+	}
+
 	return pvr
 }
 

pkg/restore/actions/csi/pvc_action.go

@@ -20,17 +20,20 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"time"
 
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 
 	"github.com/cockroachdb/errors"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilrand "k8s.io/apimachinery/pkg/util/rand"
+	"k8s.io/client-go/kubernetes"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
@@ -44,6 +47,7 @@ import (
 	uploaderUtil "github.com/vmware-tanzu/velero/pkg/uploader/util"
 	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
 const (
@@ -53,12 +57,13 @@ const (
 
 // pvcRestoreItemAction is a restore item action plugin for Velero
 type pvcRestoreItemAction struct {
-	log      logrus.FieldLogger
-	crClient crclient.Client
+	log        logrus.FieldLogger
+	crClient   crclient.Client
+	kubeClient kubernetes.Interface
 }
 
 // AppliesTo returns information indicating that the
-// PVCRestoreItemAction should be run while restoring PVCs.
+// PVCCSIRestoreItemAction should be run while restoring PVCs.
 func (p *pvcRestoreItemAction) AppliesTo() (velero.ResourceSelector, error) {
 	return velero.ResourceSelector{
 		IncludedResources: []string{"persistentvolumeclaims"},
@@ -83,28 +88,150 @@ func (p *pvcRestoreItemAction) Execute(
 	}
 
 	logger := p.log.WithFields(logrus.Fields{
-		"Action":  "PVCRestoreItemAction",
+		"Action":  "PVCCSIRestoreItemAction",
 		"PVC":     pvc.Namespace + "/" + pvc.Name,
 		"Restore": input.Restore.Namespace + "/" + input.Restore.Name,
 	})
-	logger.Info("Starting PVCRestoreItemAction for PVC")
+	logger.Info("Starting PVCCSIRestoreItemAction for PVC")
 
+	// make sure this RIA only runs for CSI snapshot
 	vsName, nameOK := pvcFromBackup.Annotations[velerov1api.VolumeSnapshotLabel]
 	if !nameOK {
-		logger.Info("Skipping PVCRestoreItemAction for PVC, PVC does not have a CSI VolumeSnapshot.")
+		logger.Info("Skipping PVCCSIRestoreItemAction for PVC, PVC does not have a CSI VolumeSnapshot.")
 		return &velero.RestoreItemActionExecuteOutput{
 			UpdatedItem: input.Item,
 		}, nil
 	}
 
-	// If PVC already exists, returns early.
-	if p.isResourceExist(pvc, *input.Restore) {
+	pvcExists, existingPVC, err := p.isResourceExist(&pvc, *input.Restore)
+	if err != nil {
+		logger.Error(err)
+		return nil, errors.WithStack(err)
+	}
+
+	var output *velero.RestoreItemActionExecuteOutput
+	if boolptr.IsSetToFalse(input.Restore.Spec.RestorePVs) {
+		output, err = p.executeWithoutPVRestore(logger, input, pvcExists, &pvc)
+	} else {
+		backup := new(velerov1api.Backup)
+		if err := p.crClient.Get(context.TODO(), crclient.ObjectKey{Namespace: input.Restore.Namespace, Name: input.Restore.Spec.BackupName}, backup); err != nil {
+			return nil, fmt.Errorf("fail to get backup for restore: %s", err.Error())
+		}
+		if boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData) {
+			output, err = p.executeWithDataMove(logger, input, backup, pvcExists, existingPVC, &pvc, &pvcFromBackup)
+		} else {
+			output, err = p.executeWithoutDataMove(logger, input, pvcExists, &pvc, vsName)
+		}
+	}
+	if err != nil {
+		logger.Error(err)
+		return nil, errors.WithStack(err)
+	}
+
+	logger.Info("Returning from PVCCSIRestoreItemAction for PVC")
+
+	return output, nil
+}
+
+func (p *pvcRestoreItemAction) executeWithoutPVRestore(logger *logrus.Entry, input *velero.RestoreItemActionExecuteInput, pvcExists bool, pvc *corev1api.PersistentVolumeClaim) (*velero.RestoreItemActionExecuteOutput, error) {
+	if pvcExists {
 		logger.Warnf("PVC already exists. Skip restore this PVC.")
 		return &velero.RestoreItemActionExecuteOutput{
 			UpdatedItem: input.Item,
 		}, nil
 	}
 
+	logger.Info("Restore did not request for PVs to be restored from snapshot")
+	pvc.Spec.VolumeName = ""
+	pvc.Spec.DataSource = nil
+	pvc.Spec.DataSourceRef = nil
+
+	unstructuredPVC, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvc)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return &velero.RestoreItemActionExecuteOutput{
+		UpdatedItem: &unstructured.Unstructured{Object: unstructuredPVC},
+	}, nil
+}
+
+func (p *pvcRestoreItemAction) executeWithoutDataMove(logger *logrus.Entry, input *velero.RestoreItemActionExecuteInput, pvcExists bool, pvc *corev1api.PersistentVolumeClaim, vsName string) (*velero.RestoreItemActionExecuteOutput, error) {
+	if pvcExists {
+		logger.Warnf("PVC already exists. Skip restore this PVC.")
+		return &velero.RestoreItemActionExecuteOutput{
+			UpdatedItem: input.Item,
+		}, nil
+	}
+
+	//To avoid confilcs, vs and vsc get a new uniq name based in restore UID
+	// and vs name old name
+	newVSName := util.GenerateSha256FromRestoreUIDAndVsName(string(input.Restore.UID), vsName)
+
+	logger.Debugf("Setting PVC source to VolumeSnapshot new name: %s", newVSName)
+	resetPVCSourceToVolumeSnapshot(pvc, newVSName)
+
+	// Force-restore the VolumeSnapshot even when restore resource filters
+	// would otherwise exclude it (mirrors backup-side must-include).
+	annotations := pvc.GetAnnotations()
+	if annotations == nil {
+		annotations = map[string]string{}
+	}
+	annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+	pvc.SetAnnotations(annotations)
+
+	unstructuredPVC, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvc)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return &velero.RestoreItemActionExecuteOutput{
+		UpdatedItem: &unstructured.Unstructured{Object: unstructuredPVC},
+		AdditionalItems: []velero.ResourceIdentifier{
+			{
+				GroupResource: kuberesource.VolumeSnapshots,
+				Name:          vsName,
+				Namespace:     pvc.Namespace,
+			},
+		},
+	}, nil
+}
+
+func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *velero.RestoreItemActionExecuteInput, backup *velerov1api.Backup, pvcExists bool, existingPVC, pvc, pvcFromBackup *corev1api.PersistentVolumeClaim) (*velero.RestoreItemActionExecuteOutput, error) {
+	var existingPV *corev1api.PersistentVolume
+	var err error
+	if pvcExists {
+		// If PVC already exists and is not in-place restore, returns early.
+		if !input.Restore.IsVolumeDataInplaceRestore() {
+			logger.Warnf("PVC already exists and ExistingVolumeDataPolicy is not in-place restore. Skip restore this PVC.")
+			return &velero.RestoreItemActionExecuteOutput{
+				UpdatedItem: input.Item,
+			}, nil
+		}
+
+		// the existing PVC should be deleted here rather than in the Exposer, otherwise the target PVC cannot be restored
+		existingPV, err = p.prepareForInplaceRestore(context.Background(), logger, pvc, existingPVC, backup.Spec.CSISnapshotTimeout.Duration)
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+	}
+
+	logger.Info("Start DataMover restore.")
+
+	// If PVC doesn't have a DataUploadNameLabel, which should be created
+	// during backup, then CSI cannot handle the volume during to restore,
+	// so return early to let Velero tries to fall back to Velero native snapshot.
+	if _, ok := pvcFromBackup.Annotations[velerov1api.DataUploadNameAnnotation]; !ok {
+		logger.Warnf("PVC doesn't have a DataUpload for data mover. Return.")
+		return &velero.RestoreItemActionExecuteOutput{
+			UpdatedItem: input.Item,
+		}, nil
+	}
+
+	operationID := label.GetValidName(
+		string(velerov1api.AsyncOperationIDPrefixDataDownload) +
+			string(input.Restore.UID) + "." + string(pvcFromBackup.UID))
+
 	// If cross-namespace restore is configured, change the namespace
 	// for PVC object to be restored
 	newNamespace, ok := input.Restore.Spec.NamespaceMapping[pvc.GetNamespace()]
@@ -113,90 +240,24 @@ func (p *pvcRestoreItemAction) Execute(
 		newNamespace = pvc.Namespace
 	}
 
-	operationID := ""
-
-	additionalItems := []velero.ResourceIdentifier{}
-	if boolptr.IsSetToFalse(input.Restore.Spec.RestorePVs) {
-		logger.Info("Restore did not request for PVs to be restored from snapshot")
-		pvc.Spec.VolumeName = ""
-		pvc.Spec.DataSource = nil
-		pvc.Spec.DataSourceRef = nil
-	} else {
-		backup := new(velerov1api.Backup)
-		err := p.crClient.Get(
-			context.TODO(),
-			crclient.ObjectKey{
-				Namespace: input.Restore.Namespace,
-				Name:      input.Restore.Spec.BackupName,
-			},
-			backup,
-		)
-
-		if err != nil {
-			logger.Error("Fail to get backup for restore.")
-			return nil, fmt.Errorf("fail to get backup for restore: %s", err.Error())
-		}
-
-		if boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData) {
-			logger.Info("Start DataMover restore.")
-
-			// If PVC doesn't have a DataUploadNameLabel, which should be created
-			// during backup, then CSI cannot handle the volume during to restore,
-			// so return early to let Velero tries to fall back to Velero native snapshot.
-			if _, ok := pvcFromBackup.Annotations[velerov1api.DataUploadNameAnnotation]; !ok {
-				logger.Warnf("PVC doesn't have a DataUpload for data mover. Return.")
-				return &velero.RestoreItemActionExecuteOutput{
-					UpdatedItem: input.Item,
-				}, nil
-			}
-
-			operationID = label.GetValidName(
-				string(velerov1api.AsyncOperationIDPrefixDataDownload) +
-					string(input.Restore.UID) + "." + string(pvcFromBackup.UID))
-			dataDownload, err := restoreFromDataUploadResult(
-				context.Background(), input.Restore, backup, &pvc, newNamespace,
-				operationID, p.crClient)
-			if err != nil {
-				logger.Errorf("Fail to restore from DataUploadResult: %s", err.Error())
-				return nil, errors.WithStack(err)
-			}
-			logger.Infof("DataDownload %s/%s is created successfully.",
-				dataDownload.Namespace, dataDownload.Name)
-		} else {
-			//To avoid confilcs, vs and vsc get a new uniq name based in restore UID
-			// and vs name old name
-			newVSName := util.GenerateSha256FromRestoreUIDAndVsName(string(input.Restore.UID), vsName)
-
-			p.log.Debugf("Setting PVC source to VolumeSnapshot new name: %s", newVSName)
-			resetPVCSourceToVolumeSnapshot(&pvc, newVSName)
-
-			additionalItems = append(additionalItems, velero.ResourceIdentifier{
-				GroupResource: kuberesource.VolumeSnapshots,
-				Name:          vsName,
-				Namespace:     pvc.Namespace,
-			})
-
-			// Force-restore the VolumeSnapshot even when restore resource filters
-			// would otherwise exclude it (mirrors backup-side must-include).
-			annotations := pvc.GetAnnotations()
-			if annotations == nil {
-				annotations = map[string]string{}
-			}
-			annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
-			pvc.SetAnnotations(annotations)
-		}
+	dataDownload, err := restoreFromDataUploadResult(
+		context.Background(), input.Restore, backup, pvc, existingPV, newNamespace,
+		operationID, p.crClient)
+	if err != nil {
+		logger.Errorf("Fail to restore from DataUploadResult: %s", err.Error())
+		return nil, errors.WithStack(err)
 	}
+	logger.Infof("DataDownload %s/%s is created successfully.",
+		dataDownload.Namespace, dataDownload.Name)
 
-	pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&pvc)
+	unstructuredPVC, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvc)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
-	logger.Info("Returning from PVCRestoreItemAction for PVC")
 
 	return &velero.RestoreItemActionExecuteOutput{
-		UpdatedItem:     &unstructured.Unstructured{Object: pvcMap},
-		OperationID:     operationID,
-		AdditionalItems: additionalItems,
+		UpdatedItem: &unstructured.Unstructured{Object: unstructuredPVC},
+		OperationID: operationID,
 	}, nil
 }
 
@@ -406,8 +467,13 @@ func newDataDownload(
 	backup *velerov1api.Backup,
 	dataUploadResult *velerov2alpha1.DataUploadResult,
 	pvc *corev1api.PersistentVolumeClaim,
+	pv *corev1api.PersistentVolume,
 	newNamespace, operationID string,
 ) *velerov2alpha1.DataDownload {
+	pvName := ""
+	if pv != nil {
+		pvName = pv.Name
+	}
 	dataDownload := &velerov2alpha1.DataDownload{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: velerov2alpha1.SchemeGroupVersion.String(),
@@ -434,6 +500,7 @@ func newDataDownload(
 		Spec: velerov2alpha1.DataDownloadSpec{
 			TargetVolume: velerov2alpha1.TargetVolumeSpec{
 				PVC:       pvc.Name,
+				PV:        pvName,
 				Namespace: newNamespace,
 				FSType:    dataUploadResult.FSType,
 			},
@@ -449,6 +516,9 @@ func newDataDownload(
 	if restore.Spec.UploaderConfig != nil {
 		dataDownload.Spec.DataMoverConfig = uploaderUtil.StoreRestoreConfig(restore.Spec.UploaderConfig)
 	}
+	if restore.IsVolumeDataInplaceRestore() {
+		dataDownload.Spec.RestoreType = string(restore.Spec.ExistingVolumeDataPolicy)
+	}
 	return dataDownload
 }
 
@@ -457,6 +527,7 @@ func restoreFromDataUploadResult(
 	restore *velerov1api.Restore,
 	backup *velerov1api.Backup,
 	pvc *corev1api.PersistentVolumeClaim,
+	pv *corev1api.PersistentVolume,
 	newNamespace, operationID string,
 	crClient crclient.Client,
 ) (*velerov2alpha1.DataDownload, error) {
@@ -481,6 +552,7 @@ func restoreFromDataUploadResult(
 		backup,
 		dataUploadResult,
 		pvc,
+		pv,
 		newNamespace,
 		operationID,
 	)
@@ -493,9 +565,9 @@ func restoreFromDataUploadResult(
 }
 
 func (p *pvcRestoreItemAction) isResourceExist(
-	pvc corev1api.PersistentVolumeClaim,
+	pvc *corev1api.PersistentVolumeClaim,
 	restore velerov1api.Restore,
-) bool {
+) (bool, *corev1api.PersistentVolumeClaim, error) {
 	// get target namespace to restore into, if different from source namespace
 	targetNamespace := pvc.Namespace
 	if target, ok := restore.Spec.NamespaceMapping[pvc.Namespace]; ok {
@@ -503,17 +575,61 @@ func (p *pvcRestoreItemAction) isResourceExist(
 	}
 
 	tmpPVC := new(corev1api.PersistentVolumeClaim)
-	if err := p.crClient.Get(
+	err := p.crClient.Get(
 		context.Background(),
 		crclient.ObjectKey{
 			Name:      pvc.Name,
 			Namespace: targetNamespace,
 		},
 		tmpPVC,
-	); err == nil {
-		return true
+	)
+	if err == nil {
+		return true, tmpPVC, nil
 	}
-	return false
+	if apierrors.IsNotFound(err) {
+		return false, nil, nil
+	}
+	return false, nil, errors.Wrapf(err, "fail to get PVC %s in namespace %s", pvc.Name, targetNamespace)
+}
+
+func (p *pvcRestoreItemAction) prepareForInplaceRestore(ctx context.Context, logger *logrus.Entry, targetPVC *corev1api.PersistentVolumeClaim, existingPVC *corev1api.PersistentVolumeClaim, operationTimeout time.Duration) (*corev1api.PersistentVolume, error) {
+	if existingPVC.Status.Phase != corev1api.ClaimBound {
+		return nil, errors.New("ExistingVolumeDataPolicy is in-place restore, but the existing PVC is not bound.")
+	}
+
+	// set the "selected-node" annotation to target PVC to make sure the target pod is scheduled to the same node
+	selectedNode, exists := existingPVC.Annotations[kube.KubeAnnSelectedNode]
+	if exists {
+		logger.Infof("Setting %q annotation to %q for target PVC to keep the same selected node as the existing PVC", kube.KubeAnnSelectedNode, existingPVC.Annotations[kube.KubeAnnSelectedNode])
+		if targetPVC.Annotations == nil {
+			targetPVC.Annotations = map[string]string{}
+		}
+		targetPVC.Annotations[kube.KubeAnnSelectedNode] = selectedNode
+	}
+
+	var err error
+	logger.Info("ExistingVolumeDataPolicy is in-place restore. Deleting the existing PVC but keep the PV...")
+	pv := &corev1api.PersistentVolume{}
+	if err = p.crClient.Get(context.Background(), crclient.ObjectKey{Name: existingPVC.Spec.VolumeName}, pv); err != nil {
+		return nil, errors.Errorf("Fail to get PV %s: %s", existingPVC.Spec.VolumeName, err.Error())
+	}
+
+	// set reclaim policy to retain
+	updatedPV, err := kube.SetPVReclaimPolicy(ctx, p.kubeClient.CoreV1(), pv, corev1api.PersistentVolumeReclaimRetain)
+	if err != nil {
+		return nil, errors.Wrapf(err, "fail to set PV reclaim policy to retain for PV %s", pv.Name)
+	}
+	if updatedPV != nil {
+		pv = updatedPV
+	}
+
+	if err = kube.EnsureDeletePVC(ctx, p.kubeClient.CoreV1(), existingPVC.Name, existingPVC.Namespace, operationTimeout); err != nil {
+		return nil, errors.Wrapf(err, "fail to delete the existing PVC %s in namespace %s", existingPVC.Name, existingPVC.Namespace)
+	}
+
+	logger.Info("Existing PVC deleted")
+
+	return pv, nil
 }
 
 func NewPvcRestoreItemAction(f client.Factory) plugincommon.HandlerInitializer {
@@ -523,9 +639,15 @@ func NewPvcRestoreItemAction(f client.Factory) plugincommon.HandlerInitializer {
 			return nil, err
 		}
 
+		kubeClient, err := f.KubeClient()
+		if err != nil {
+			return nil, err
+		}
+
 		return &pvcRestoreItemAction{
-			log:      logger,
-			crClient: crClient,
+			log:        logger,
+			crClient:   crClient,
+			kubeClient: kubeClient,
 		}, nil
 	}
 }

pkg/restore/actions/csi/pvc_action_test.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/validation"
+	"k8s.io/client-go/kubernetes/fake"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
@@ -371,6 +372,7 @@ func TestExecute(t *testing.T) {
 		backup               *velerov1api.Backup
 		restore              *velerov1api.Restore
 		pvc                  *corev1api.PersistentVolumeClaim
+		pv                   *corev1api.PersistentVolume
 		pvcFromBackup        *corev1api.PersistentVolumeClaim
 		vs                   *snapshotv1api.VolumeSnapshot
 		dataUploadResult     *corev1api.ConfigMap
@@ -378,9 +380,11 @@ func TestExecute(t *testing.T) {
 		expectedDataDownload *velerov2alpha1.DataDownload
 		expectedPVC          *corev1api.PersistentVolumeClaim
 		preCreatePVC         bool
+		kubeClientObj        []runtime.Object
 	}{
 		{
 			name:        "Don't restore PV",
+			backup:      builder.ForBackup("velero", "testBackup").Result(),
 			restore:     builder.ForRestore("velero", "testRestore").Backup("testBackup").RestorePVs(false).Result(),
 			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName")).Result(),
 			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName")).VolumeName("").Result(),
@@ -486,6 +490,26 @@ func TestExecute(t *testing.T) {
 			pvc:          builder.ForPersistentVolumeClaim("restore", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
 			preCreatePVC: true,
 		},
+		{
+			name:             "PVC exists and in-place restore set",
+			backup:           builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result(),
+			restore:          builder.ForRestore("velero", "testRestore").Backup("testBackup").ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeFull)).ItemOperationTimeout(time.Minute * 10).ObjectMeta(builder.WithUID("uid")).Result(),
+			pvc:              builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			pv:               builder.ForPersistentVolume("testPV").ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).Result(),
+			dataUploadResult: builder.ForConfigMap("velero", "testCM").Data("uid", "{}").ObjectMeta(builder.WithLabels(velerov1api.RestoreUIDLabel, "uid", velerov1api.PVCNamespaceNameLabel, "velero.testPVC", velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)))).Result(),
+			preCreatePVC:     true,
+			kubeClientObj: []runtime.Object{
+				builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").Phase(corev1api.ClaimBound).ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", velerov1api.VolumeSnapshotRestoreSize, "10Gi", velerov1api.DataUploadNameAnnotation, "velero/")).Result(),
+			},
+			expectedDataDownload: func() *velerov2alpha1.DataDownload {
+				d := builder.ForDataDownload("velero", "name").TargetVolume(velerov2alpha1.TargetVolumeSpec{PVC: "testPVC", Namespace: "velero", PV: "testPV"}).
+					ObjectMeta(builder.WithOwnerReference([]metav1.OwnerReference{{APIVersion: velerov1api.SchemeGroupVersion.String(), Kind: "Restore", Name: "testRestore", UID: "uid", Controller: boolptr.True()}}),
+						builder.WithLabelsMap(map[string]string{velerov1api.AsyncOperationIDLabel: "dd-uid.", velerov1api.RestoreNameLabel: "testRestore", velerov1api.RestoreUIDLabel: "uid"}),
+						builder.WithGenerateName("testRestore-")).Result()
+				d.Spec.RestoreType = "full"
+				return d
+			}(),
+		},
 	}
 
 	for _, tc := range tests {
@@ -499,6 +523,10 @@ func TestExecute(t *testing.T) {
 				object = append(object, tc.vs)
 			}
 
+			if tc.pv != nil {
+				object = append(object, tc.pv)
+			}
+
 			input := new(velero.RestoreItemActionExecuteInput)
 
 			if tc.pvc != nil {
@@ -524,8 +552,9 @@ func TestExecute(t *testing.T) {
 			}
 
 			pvcRIA := pvcRestoreItemAction{
-				log:      logrus.New(),
-				crClient: velerotest.NewFakeControllerRuntimeClient(t, object...),
+				log:        logrus.New(),
+				crClient:   velerotest.NewFakeControllerRuntimeClient(t, object...),
+				kubeClient: fake.NewSimpleClientset(tc.kubeClientObj...),
 			}
 
 			output, err := pvcRIA.Execute(input)
@@ -596,6 +625,7 @@ func TestNewPvcRestoreItemAction(t *testing.T) {
 
 	f1 := &factorymocks.Factory{}
 	f1.On("KubebuilderClient").Return(crClient, nil)
+	f1.On("KubeClient").Return(nil, nil)
 	plugin1 := NewPvcRestoreItemAction(f1)
 	_, err1 := plugin1(logger)
 	require.NoError(t, err1)

pkg/util/kube/pvc_pv.go

@@ -95,6 +95,10 @@ func WaitPVCBound(ctx context.Context, pvcGetter corev1client.CoreV1Interface,
 			return false, nil
 		}
 
+		if tmpPVC.Status.Phase != corev1api.ClaimBound {
+			return false, nil
+		}
+
 		updated = tmpPVC
 
 		return true, nil
@@ -112,6 +116,16 @@ func WaitPVCBound(ctx context.Context, pvcGetter corev1client.CoreV1Interface,
 	return pv, err
 }
 
+// DeletePVCIfAny deletes a PVC by namespace and name if it exists, and log an error when the deletion fails
+func DeletePVCIfAny(ctx context.Context, client corev1client.CoreV1Interface, pvcName, pvcNamespace string, ensureTimeout time.Duration, log logrus.FieldLogger) {
+	if err := EnsureDeletePVC(ctx, client, pvcName, pvcNamespace, ensureTimeout); err != nil {
+		if apierrors.IsNotFound(err) {
+			return
+		}
+		log.Warnf("failed to delete pvc %s/%s with err %v", pvcNamespace, pvcName, err)
+	}
+}
+
 // DeletePVIfAny deletes a PV by name if it exists, and log an error when the deletion fails
 func DeletePVIfAny(ctx context.Context, pvGetter corev1client.CoreV1Interface, pvName string, log logrus.FieldLogger) {
 	err := pvGetter.PersistentVolumes().Delete(ctx, pvName, metav1.DeleteOptions{})

pkg/util/kube/pvc_pv_test.go

@@ -61,6 +61,9 @@ func TestWaitPVCBound(t *testing.T) {
 		Spec: corev1api.PersistentVolumeClaimSpec{
 			VolumeName: "fake-pv",
 		},
+		Status: corev1api.PersistentVolumeClaimStatus{
+			Phase: corev1api.ClaimBound,
+		},
 	}
 
 	pvObj := &corev1api.PersistentVolume{
@@ -303,6 +306,105 @@ func TestWaitPVCConsumed(t *testing.T) {
 }
 
 func TestDeletePVCIfAny(t *testing.T) {
+	pvcObject := &corev1api.PersistentVolumeClaim{
+		TypeMeta: metav1.TypeMeta{
+			Kind: "fake-kind-1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "fake-namespace",
+			Name:      "fake-pvc",
+		},
+	}
+
+	tests := []struct {
+		name          string
+		pvcName       string
+		pvcNamespace  string
+		kubeClientObj []runtime.Object
+		kubeReactors  []reactor
+		logMessage    string
+		logLevel      string
+		ensureTimeout time.Duration
+	}{
+		{
+			name:         "pvc not found",
+			pvcName:      "fake-pvc",
+			pvcNamespace: "fake-namespace",
+		},
+		{
+			name:         "failed to delete pvc",
+			pvcName:      "fake-pvc",
+			pvcNamespace: "fake-namespace",
+			kubeReactors: []reactor{
+				{
+					verb:     "delete",
+					resource: "persistentvolumeclaims",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("fake-delete-error")
+					},
+				},
+			},
+			kubeClientObj: []runtime.Object{
+				pvcObject,
+			},
+			logMessage: "failed to delete pvc fake-namespace/fake-pvc with err error to delete pvc fake-pvc: fake-delete-error",
+			logLevel:   "level=warning",
+		},
+		{
+			name:         "delete pvc success",
+			pvcName:      "fake-pvc",
+			pvcNamespace: "fake-namespace",
+			kubeClientObj: []runtime.Object{
+				pvcObject,
+			},
+		},
+		{
+			name:         "delete pvc success but wait fail",
+			pvcName:      "fake-pvc",
+			pvcNamespace: "fake-namespace",
+			kubeClientObj: []runtime.Object{
+				pvcObject,
+			},
+			kubeReactors: []reactor{
+				{
+					verb:     "delete",
+					resource: "persistentvolumeclaims",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, pvcObject, nil
+					},
+				},
+			},
+			ensureTimeout: time.Second,
+			logMessage:    "failed to delete pvc fake-namespace/fake-pvc with err timeout to assure pvc fake-pvc is deleted, finalizers in pvc []",
+			logLevel:      "level=warning",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
+
+			for _, reactor := range test.kubeReactors {
+				fakeKubeClient.Fake.PrependReactor(reactor.verb, reactor.resource, reactor.reactorFunc)
+			}
+
+			var kubeClient kubernetes.Interface = fakeKubeClient
+
+			logMessage := ""
+			DeletePVCIfAny(t.Context(), kubeClient.CoreV1(), test.pvcName, test.pvcNamespace, test.ensureTimeout, velerotest.NewSingleLogger(&logMessage))
+
+			if len(test.logMessage) > 0 {
+				assert.Contains(t, logMessage, test.logMessage)
+			}
+
+			if len(test.logLevel) > 0 {
+				assert.Contains(t, logMessage, test.logLevel)
+			}
+		})
+	}
+}
+
+func TestDeletePVAndPVCIfAny(t *testing.T) {
 	pvObject := &corev1api.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "fake-pv",