Merge branch 'main' into data-mover-support-fs-type

pkg/backup/backup_test.go

@@ -6107,15 +6107,43 @@ func TestGetNamespaceFilter(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			// First call (populates cache)
 			result := req.GetNamespaceFilter(tt.namespace)
 
 			if tt.expectNil {
 				assert.Nil(t, result)
+
+				// Verify negative cache
+				val, ok := req.NamespaceFilterCache.Load(tt.namespace)
+				assert.True(t, ok)
+				assert.Nil(t, val)
 			} else {
 				assert.NotNil(t, result)
 				// Ensure the returned filter points to the correct reference in our map
 				assert.Same(t, filterMap[tt.expectMatched], result)
+
+				// Verify positive cache
+				val, ok := req.NamespaceFilterCache.Load(tt.namespace)
+				assert.True(t, ok)
+				assert.Same(t, filterMap[tt.expectMatched], val)
 			}
+
+			// Second call (hits cache)
+			result2 := req.GetNamespaceFilter(tt.namespace)
+			assert.Same(t, result, result2)
 		})
 	}
 }
+
+func TestGetNamespaceFilter_CacheBypass(t *testing.T) {
+	req := &Request{
+		NamespacedFilterMap: make(map[string]*ResolvedNamespaceFilter),
+	}
+
+	cachedFilter := &ResolvedNamespaceFilter{}
+	req.NamespaceFilterCache.Store("cached-ns", cachedFilter)
+
+	// Since NamespacedFilterMap is empty, this would normally return nil,
+	// but the cache should return our cachedFilter.
+	assert.Same(t, cachedFilter, req.GetNamespaceFilter("cached-ns"))
+}

pkg/backup/request.go

@@ -100,6 +100,10 @@ type Request struct {
 	// NamespacedFilterPatterns preserves the order of patterns for first-match semantics
 	// and caches pre-compiled globs to avoid repeated compilation in the hot path.
 	NamespacedFilterPatterns []NamespacedFilterPattern
+
+	// NamespaceFilterCache memoizes the resolved filter for a given namespace.
+	// sync.Map is used because item backuppers access this concurrently.
+	NamespaceFilterCache sync.Map
 }
 
 // NamespacedFilterPattern pairs a namespace pattern string with its pre-compiled
@@ -149,22 +153,37 @@ func (r *Request) StopWorkerPool() {
 
 // GetNamespaceFilter returns the resolved filter for a namespace, or nil
 // if the namespace should use global filters. Uses first-match semantics
-// when multiple patterns could match the same namespace.
+// when multiple patterns could match the same namespace, but exact matches
+// always take precedence over glob patterns regardless of definition order.
 func (r *Request) GetNamespaceFilter(namespace string) *ResolvedNamespaceFilter {
 	if r.NamespacedFilterMap == nil {
 		return nil
 	}
 
-	// First check for exact match
+	// 1. Check the concurrent cache first
+	if val, ok := r.NamespaceFilterCache.Load(namespace); ok {
+		if val == nil {
+			return nil
+		}
+		return val.(*ResolvedNamespaceFilter)
+	}
+
+	// 2. Check for exact match first
 	if f, ok := r.NamespacedFilterMap[namespace]; ok {
+		r.NamespaceFilterCache.Store(namespace, f)
 		return f
 	}
 
-	// Walk patterns in definition order using pre-compiled globs (no allocation per call)
+	// 3. Walk patterns in definition order using pre-compiled globs
 	for _, p := range r.NamespacedFilterPatterns {
 		if p.Compiled != nil && p.Compiled.Match(namespace) {
-			return r.NamespacedFilterMap[p.Pattern]
+			filter := r.NamespacedFilterMap[p.Pattern]
+			r.NamespaceFilterCache.Store(namespace, filter)
+			return filter
 		}
 	}
+
+	// 4. Cache the miss
+	r.NamespaceFilterCache.Store(namespace, nil)
 	return nil
 }

pkg/controller/restore_finalizer_controller.go

@@ -421,7 +421,16 @@ func (ctx *finalizerContext) patchDynamicPVWithVolumeInfo() (errs results.Result
 					// patch PV's reclaim policy and label using the corresponding data stored in volume info
 					if needPatch(pv, volInfo.PVInfo) {
 						updatedPV := pv.DeepCopy()
-						updatedPV.Labels = volInfo.PVInfo.Labels
+
+						if updatedPV.Labels == nil {
+							updatedPV.Labels = make(map[string]string)
+						}
+						for k, v := range volInfo.PVInfo.Labels {
+							if _, exists := updatedPV.Labels[k]; !exists {
+								updatedPV.Labels[k] = v
+							}
+						}
+
 						updatedPV.Spec.PersistentVolumeReclaimPolicy = corev1api.PersistentVolumeReclaimPolicy(volInfo.PVInfo.ReclaimPolicy)
 						if err := kubeutil.PatchResource(pv, updatedPV, ctx.crClient); err != nil {
 							return false, err
@@ -553,13 +562,10 @@ func needPatch(newPV *corev1api.PersistentVolume, pvInfo *volume.PVInfo) bool {
 	}
 
 	newPVLabels, pvLabels := newPV.Labels, pvInfo.Labels
-	for k, v := range pvLabels {
+	for k := range pvLabels {
 		if _, ok := newPVLabels[k]; !ok {
 			return true
 		}
-		if newPVLabels[k] != v {
-			return true
-		}
 	}
 
 	return false

pkg/controller/restore_finalizer_controller_test.go

@@ -634,6 +634,87 @@ func Test_restoreFinalizerReconciler_finishProcessing(t *testing.T) {
 	}
 }
 
+func TestNeedPatch(t *testing.T) {
+	tests := []struct {
+		name     string
+		newPV    *corev1api.PersistentVolume
+		pvInfo   *volume.PVInfo
+		expected bool
+	}{
+		{
+			name: "reclaim policy differs",
+			newPV: builder.ForPersistentVolume("pv1").
+				ReclaimPolicy(corev1api.PersistentVolumeReclaimDelete).Result(),
+			pvInfo: &volume.PVInfo{
+				ReclaimPolicy: string(corev1api.PersistentVolumeReclaimRetain),
+				Labels:        map[string]string{},
+			},
+			expected: true,
+		},
+		{
+			name: "backup has label new PV does not",
+			newPV: builder.ForPersistentVolume("pv1").
+				ObjectMeta(builder.WithLabels("existing", "val")).
+				ReclaimPolicy(corev1api.PersistentVolumeReclaimDelete).Result(),
+			pvInfo: &volume.PVInfo{
+				ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+				Labels:        map[string]string{"existing": "val", "missing": "val"},
+			},
+			expected: true,
+		},
+		{
+			name: "same labels same values",
+			newPV: builder.ForPersistentVolume("pv1").
+				ObjectMeta(builder.WithLabels("key", "val")).
+				ReclaimPolicy(corev1api.PersistentVolumeReclaimDelete).Result(),
+			pvInfo: &volume.PVInfo{
+				ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+				Labels:        map[string]string{"key": "val"},
+			},
+			expected: false,
+		},
+		{
+			name: "same label key different values",
+			newPV: builder.ForPersistentVolume("pv1").
+				ObjectMeta(builder.WithLabels("topology.kubernetes.io/zone", "us-west-2a")).
+				ReclaimPolicy(corev1api.PersistentVolumeReclaimDelete).Result(),
+			pvInfo: &volume.PVInfo{
+				ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+				Labels:        map[string]string{"topology.kubernetes.io/zone": "us-east-1a"},
+			},
+			expected: false,
+		},
+		{
+			name: "new PV has labels backup does not",
+			newPV: builder.ForPersistentVolume("pv1").
+				ObjectMeta(builder.WithLabels("provisioner-label", "val")).
+				ReclaimPolicy(corev1api.PersistentVolumeReclaimDelete).Result(),
+			pvInfo: &volume.PVInfo{
+				ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+				Labels:        map[string]string{},
+			},
+			expected: false,
+		},
+		{
+			name: "both labels nil",
+			newPV: builder.ForPersistentVolume("pv1").
+				ReclaimPolicy(corev1api.PersistentVolumeReclaimDelete).Result(),
+			pvInfo: &volume.PVInfo{
+				ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete),
+				Labels:        nil,
+			},
+			expected: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			result := needPatch(tc.newPV, tc.pvInfo)
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}
+
 func TestRestoreOperationList(t *testing.T) {
 	var empty []*itemoperation.RestoreOperation
 	tests := []struct {

pkg/exposer/generic_restore.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/errors"
+	"github.com/google/uuid"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -425,16 +426,19 @@ func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject co
 
 	curLog.WithField("restore PV", restorePV.Name).WithField("retained", (retained != nil)).Info("Restore PV is retained")
 
+	var rebindPV *corev1api.PersistentVolume
+
 	defer func() {
 		if retained != nil {
 			curLog.WithField("retained PV", retained.Name).Info("Deleting retained PV on error")
 			kube.DeletePVIfAny(ctx, e.kubeClient.CoreV1(), retained.Name, curLog)
 		}
-	}()
 
-	if retained != nil {
-		restorePV = retained
-	}
+		if rebindPV != nil {
+			curLog.WithField("rebind PV", rebindPV.Name).Info("Deleting rebind PV on error")
+			kube.DeletePVIfAny(ctx, e.kubeClient.CoreV1(), rebindPV.Name, curLog)
+		}
+	}()
 
 	err = kube.EnsureDeletePod(ctx, e.kubeClient.CoreV1(), restorePodName, ownerObject.Namespace, param.OperationTimeout)
 	if err != nil {
@@ -448,42 +452,38 @@ func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject co
 
 	curLog.WithField("restore PVC", restorePVCName).Info("Restore PVC is deleted")
 
-	_, err = kube.RebindPVC(ctx, e.kubeClient.CoreV1(), targetPVC, restorePV.Name)
+	rebindPV, err = kube.RebindPV(ctx, e.kubeClient.CoreV1(), uuid.NewString(), retained, targetPVC, orgReclaim, param.TargetFSType)
 	if err != nil {
-		return errors.Wrapf(err, "error to rebind target PVC %s/%s to %s", targetPVC.Namespace, targetPVC.Name, restorePV.Name)
+		return errors.Wrapf(err, "error rebinding PV for target PVC %s", param.TargetPVCName)
 	}
 
-	curLog.WithField("tartet PVC", fmt.Sprintf("%s/%s", targetPVC.Namespace, targetPVC.Name)).WithField("restore PV", restorePV.Name).Info("Target PVC is rebound to restore PV")
+	curLog.WithField("rebind PV", rebindPV.Name).Info("Rebind PV is created")
 
-	var matchLabel map[string]string
-	if targetPVC.Spec.Selector != nil {
-		matchLabel = targetPVC.Spec.Selector.MatchLabels
-	}
-
-	restorePVName := restorePV.Name
-	restorePV, err = kube.ResetPVBinding(ctx, e.kubeClient.CoreV1(), restorePV, matchLabel, targetPVC)
+	err = kube.EnsureDeletePV(ctx, e.kubeClient.CoreV1(), retained.Name, param.OperationTimeout)
 	if err != nil {
-		return errors.Wrapf(err, "error to reset binding info for restore PV %s", restorePVName)
+		return errors.Wrapf(err, "error deleting PV %s", retained.Name)
 	}
 
-	curLog.WithField("restore PV", restorePV.Name).Info("Restore PV is rebound")
-
-	restorePV, err = kube.WaitPVBound(ctx, e.kubeClient.CoreV1(), restorePV.Name, targetPVC.Name, targetPVC.Namespace, param.OperationTimeout)
-	if err != nil {
-		return errors.Wrapf(err, "error to wait restore PV bound, restore PV %s", restorePVName)
-	}
-
-	curLog.WithField("restore PV", restorePV.Name).Info("Restore PV is ready")
+	curLog.WithField("retained PV", retained.Name).Info("Retained PV is deleted")
 
 	retained = nil
 
-	_, err = kube.SetPVReclaimPolicy(ctx, e.kubeClient.CoreV1(), restorePV, orgReclaim)
+	_, err = kube.RebindPVC(ctx, e.kubeClient.CoreV1(), targetPVC, rebindPV.Name)
 	if err != nil {
-		curLog.WithField("restore PV", restorePV.Name).WithError(err).Warn("Restore PV's reclaim policy is not restored")
-	} else {
-		curLog.WithField("restore PV", restorePV.Name).Info("Restore PV's reclaim policy is restored")
+		return errors.Wrapf(err, "error to rebind target PVC %s/%s to %s", targetPVC.Namespace, targetPVC.Name, rebindPV.Name)
 	}
 
+	curLog.WithField("rebind PV", rebindPV.Name).Info("Target PVC is rebound to rebind PV")
+
+	_, err = kube.WaitPVBound(ctx, e.kubeClient.CoreV1(), rebindPV.Name, targetPVC.Name, targetPVC.Namespace, param.OperationTimeout)
+	if err != nil {
+		return errors.Wrapf(err, "error to wait rebind PV ready, rebind PV %s", rebindPV.Name)
+	}
+
+	curLog.WithField("rebind PV", rebindPV.Name).Info("Rebind PV is ready")
+
+	rebindPV = nil
+
 	return nil
 }
 

pkg/exposer/generic_restore_test.go

@@ -352,8 +352,6 @@ func TestRebindVolume(t *testing.T) {
 		},
 	}
 
-	hookCount := 0
-
 	tests := []struct {
 		name            string
 		kubeClientObj   []runtime.Object
@@ -445,6 +443,50 @@ func TestRebindVolume(t *testing.T) {
 			},
 			err: "error to delete restore PVC fake-restore: error to delete pvc fake-restore: fake-delete-error",
 		},
+		{
+			name:            "rebind pv fail",
+			targetPVCName:   "fake-target-pvc",
+			targetNamespace: "fake-ns",
+			ownerRestore:    restore,
+			kubeClientObj: []runtime.Object{
+				targetPVCObj,
+				restorePVCObj,
+				restorePVObj,
+				restorePod,
+			},
+			kubeReactors: []reactor{
+				{
+					verb:     "create",
+					resource: "persistentvolumes",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("fake-create-error")
+					},
+				},
+			},
+			err: "error rebinding PV for target PVC fake-target-pvc: fake-create-error",
+		},
+		{
+			name:            "delete retained pv fail",
+			targetPVCName:   "fake-target-pvc",
+			targetNamespace: "fake-ns",
+			ownerRestore:    restore,
+			kubeClientObj: []runtime.Object{
+				targetPVCObj,
+				restorePVCObj,
+				restorePVObj,
+				restorePod,
+			},
+			kubeReactors: []reactor{
+				{
+					verb:     "delete",
+					resource: "persistentvolumes",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("fake-delete-error")
+					},
+				},
+			},
+			err: "error deleting PV fake-restore-pv: error to delete pv fake-restore-pv: fake-delete-error",
+		},
 		{
 			name:            "rebind target pvc fail",
 			targetPVCName:   "fake-target-pvc",
@@ -465,10 +507,10 @@ func TestRebindVolume(t *testing.T) {
 					},
 				},
 			},
-			err: "error to rebind target PVC fake-ns/fake-target-pvc to fake-restore-pv: error patching PVC: fake-patch-error",
+			err: "error to rebind target PVC fake-ns/fake-target-pvc to",
 		},
 		{
-			name:            "reset pv binding fail",
+			name:            "wait rebind PV ready fail",
 			targetPVCName:   "fake-target-pvc",
 			targetNamespace: "fake-ns",
 			ownerRestore:    restore,
@@ -478,34 +520,7 @@ func TestRebindVolume(t *testing.T) {
 				restorePVObj,
 				restorePod,
 			},
-			kubeReactors: []reactor{
-				{
-					verb:     "patch",
-					resource: "persistentvolumes",
-					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
-						if hookCount == 0 {
-							hookCount++
-							return false, nil, nil
-						} else {
-							return true, nil, errors.New("fake-patch-error")
-						}
-					},
-				},
-			},
-			err: "error to reset binding info for restore PV fake-restore-pv: error patching PV: fake-patch-error",
-		},
-		{
-			name:            "wait restore PV bound fail",
-			targetPVCName:   "fake-target-pvc",
-			targetNamespace: "fake-ns",
-			ownerRestore:    restore,
-			kubeClientObj: []runtime.Object{
-				targetPVCObj,
-				restorePVCObj,
-				restorePVObj,
-				restorePod,
-			},
-			err: "error to wait restore PV bound, restore PV fake-restore-pv: error to wait for bound of PV: context deadline exceeded",
+			err: "error to wait rebind PV ready, rebind PV",
 		},
 	}
 
@@ -533,14 +548,16 @@ func TestRebindVolume(t *testing.T) {
 				}
 			}
 
-			hookCount = 0
-
 			err := exposer.RebindVolume(t.Context(), ownerObject, GenericRestoreRebindVolumeParam{
 				TargetPVCName:    test.targetPVCName,
 				TargetNamespace:  test.targetNamespace,
 				OperationTimeout: time.Millisecond,
 			})
-			assert.EqualError(t, err, test.err)
+			if test.err != "" {
+				assert.ErrorContains(t, err, test.err)
+			} else {
+				assert.NoError(t, err)
+			}
 		})
 	}
 }

pkg/util/kube/pvc_pv.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"maps"
 	"strings"
 	"time"
 
@@ -161,6 +162,42 @@ func EnsureDeletePVC(ctx context.Context, pvcGetter corev1client.CoreV1Interface
 	return nil
 }
 
+func EnsureDeletePV(ctx context.Context, pvGetter corev1client.CoreV1Interface, pvName string, timeout time.Duration) error {
+	err := pvGetter.PersistentVolumes().Delete(ctx, pvName, metav1.DeleteOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "error to delete pv %s", pvName)
+	}
+
+	if timeout == 0 {
+		return nil
+	}
+
+	var updated *corev1api.PersistentVolume
+	err = wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
+		pv, err := pvGetter.PersistentVolumes().Get(ctx, pvName, metav1.GetOptions{})
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				return true, nil
+			}
+
+			return false, errors.Wrapf(err, "error to get pv %s", pvName)
+		}
+
+		updated = pv
+		return false, nil
+	})
+
+	if err != nil {
+		if errors.Is(err, context.DeadlineExceeded) {
+			return errors.Errorf("timeout to assure pv %s is deleted, finalizers in pv %v", pvName, updated.Finalizers)
+		} else {
+			return errors.Wrapf(err, "error to ensure pv deleted for %s", pvName)
+		}
+	}
+
+	return nil
+}
+
 // EnsurePVDeleted ensures a PV has been deleted. This function is supposed to be called after EnsureDeletePVC
 // If timeout is 0, it doesn't wait and return nil
 func EnsurePVDeleted(ctx context.Context, pvGetter corev1client.CoreV1Interface, pvName string, timeout time.Duration) error {
@@ -269,6 +306,87 @@ func ResetPVBinding(ctx context.Context, pvGetter corev1client.CoreV1Interface,
 	return updated, nil
 }
 
+func RebindPV(ctx context.Context, pvGetter corev1client.CoreV1Interface, pvName string, source *corev1api.PersistentVolume,
+	pvc *corev1api.PersistentVolumeClaim, policy corev1api.PersistentVolumeReclaimPolicy, fsType string) (*corev1api.PersistentVolume, error) {
+	if source == nil {
+		return nil, errors.New("source PV is required to rebind PV")
+	}
+
+	if pvc == nil {
+		return nil, errors.New("target PVC is required to rebind PV")
+	}
+
+	pvLabel := make(map[string]string)
+
+	maps.Copy(pvLabel, source.Labels)
+
+	if pvc.Spec.Selector != nil {
+		maps.Copy(pvLabel, pvc.Spec.Selector.MatchLabels)
+	}
+
+	pvAnnotations := make(map[string]string)
+	maps.Copy(pvAnnotations, source.Annotations)
+	delete(pvAnnotations, KubeAnnBoundByController)
+
+	pv := &corev1api.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        pvName,
+			Labels:      pvLabel,
+			Annotations: pvAnnotations,
+		},
+		Spec: corev1api.PersistentVolumeSpec{
+			Capacity:                      source.Spec.Capacity,
+			PersistentVolumeSource:        clonePVSource(&source.Spec.PersistentVolumeSource, fsType),
+			AccessModes:                   source.Spec.AccessModes,
+			PersistentVolumeReclaimPolicy: policy,
+			StorageClassName:              source.Spec.StorageClassName,
+			VolumeMode:                    pvc.Spec.VolumeMode,
+			NodeAffinity:                  source.Spec.NodeAffinity,
+			VolumeAttributesClassName:     source.Spec.VolumeAttributesClassName,
+			MountOptions:                  source.Spec.MountOptions,
+			ClaimRef: &corev1api.ObjectReference{
+				Kind:      pvc.Kind,
+				Namespace: pvc.Namespace,
+				Name:      pvc.Name,
+			},
+		},
+	}
+
+	return pvGetter.PersistentVolumes().Create(ctx, pv, metav1.CreateOptions{})
+}
+
+func clonePVSource(source *corev1api.PersistentVolumeSource, newFSType string) corev1api.PersistentVolumeSource {
+	newSource := source.DeepCopy()
+
+	if newFSType != "" {
+		if newSource.CSI != nil {
+			newSource.CSI.FSType = newFSType
+		} else if newSource.AWSElasticBlockStore != nil {
+			newSource.AWSElasticBlockStore.FSType = newFSType
+		} else if newSource.AzureDisk != nil {
+			newSource.AzureDisk.FSType = &newFSType
+		} else if newSource.VsphereVolume != nil {
+			newSource.VsphereVolume.FSType = newFSType
+		} else if newSource.GCEPersistentDisk != nil {
+			newSource.GCEPersistentDisk.FSType = newFSType
+		} else if newSource.Cinder != nil {
+			newSource.Cinder.FSType = newFSType
+		} else if newSource.ISCSI != nil {
+			newSource.ISCSI.FSType = newFSType
+		} else if newSource.RBD != nil {
+			newSource.RBD.FSType = newFSType
+		} else if newSource.FC != nil {
+			newSource.FC.FSType = newFSType
+		} else if newSource.Local != nil {
+			newSource.Local.FSType = &newFSType
+		} else if newSource.FlexVolume != nil {
+			newSource.FlexVolume.FSType = newFSType
+		}
+	}
+
+	return *newSource
+}
+
 // SetPVReclaimPolicy sets the specified reclaim policy to a PV
 func SetPVReclaimPolicy(ctx context.Context, pvGetter corev1client.CoreV1Interface, pv *corev1api.PersistentVolume,
 	policy corev1api.PersistentVolumeReclaimPolicy) (*corev1api.PersistentVolume, error) {

pkg/util/kube/pvc_pv_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
@@ -707,7 +708,7 @@ func TestEnsureDeletePVC(t *testing.T) {
 	}
 }
 
-func TestEnsureDeletePV(t *testing.T) {
+func TestEnsurePVDeleted(t *testing.T) {
 	pvObject := &corev1api.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "fake-pv",
@@ -2049,3 +2050,317 @@ func TestGetVolumeTopology(t *testing.T) {
 		})
 	}
 }
+
+func TestEnsureDeletePV(t *testing.T) {
+	pvObj := &corev1api.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "fake-pv",
+		},
+	}
+
+	tests := []struct {
+		name          string
+		pvName        string
+		timeout       time.Duration
+		kubeClientObj []runtime.Object
+		kubeReactors  []reactor
+		expectedErr   string
+	}{
+		{
+			name:   "delete error",
+			pvName: "fake-pv",
+			kubeReactors: []reactor{
+				{
+					verb:     "delete",
+					resource: "persistentvolumes",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("delete error")
+					},
+				},
+			},
+			expectedErr: "error to delete pv fake-pv: delete error",
+		},
+		{
+			name:          "success without wait",
+			pvName:        "fake-pv",
+			timeout:       0,
+			kubeClientObj: []runtime.Object{pvObj},
+		},
+		{
+			name:    "success with wait",
+			pvName:  "fake-pv",
+			timeout: time.Second,
+			kubeReactors: []reactor{
+				{
+					verb:     "get",
+					resource: "persistentvolumes",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, apierrors.NewNotFound(corev1api.Resource("persistentvolumes"), "fake-pv")
+					},
+				},
+			},
+			kubeClientObj: []runtime.Object{pvObj},
+		},
+		{
+			name:          "get error during wait",
+			pvName:        "fake-pv",
+			timeout:       time.Millisecond,
+			kubeClientObj: []runtime.Object{pvObj},
+			kubeReactors: []reactor{
+				{
+					verb:     "get",
+					resource: "persistentvolumes",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("get error")
+					},
+				},
+			},
+			expectedErr: "error to ensure pv deleted for fake-pv: error to get pv fake-pv: get error",
+		},
+		{
+			name:          "wait timeout",
+			pvName:        "fake-pv",
+			timeout:       time.Millisecond,
+			kubeClientObj: []runtime.Object{pvObj},
+			kubeReactors: []reactor{
+				{
+					verb:     "delete",
+					resource: "persistentvolumes",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, nil // fake delete, pv will still be in tracker
+					},
+				},
+			},
+			expectedErr: "timeout to assure pv fake-pv is deleted, finalizers in pv []",
+		},
+	}
+
+	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)
+			}
+
+			err := EnsureDeletePV(t.Context(), fakeKubeClient.CoreV1(), test.pvName, test.timeout)
+			if test.expectedErr != "" {
+				assert.EqualError(t, err, test.expectedErr)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestRebindPV(t *testing.T) {
+	sourcePV := &corev1api.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "source-pv",
+			Labels: map[string]string{
+				"key1": "val1",
+			},
+			Annotations: map[string]string{
+				"anno1":                  "val1",
+				KubeAnnBoundByController: "true",
+			},
+		},
+		Spec: corev1api.PersistentVolumeSpec{
+			PersistentVolumeSource: corev1api.PersistentVolumeSource{
+				CSI: &corev1api.CSIPersistentVolumeSource{
+					Driver:       "fake-driver",
+					VolumeHandle: "fake-handle",
+				},
+			},
+			AccessModes:                   []corev1api.PersistentVolumeAccessMode{corev1api.ReadWriteOnce},
+			PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimRetain,
+			StorageClassName:              "fake-sc",
+		},
+	}
+
+	targetPVC := &corev1api.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "fake-ns",
+			Name:      "target-pvc",
+		},
+		Spec: corev1api.PersistentVolumeClaimSpec{
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					"key1": "val3",
+					"key2": "val2",
+				},
+			},
+		},
+	}
+
+	tests := []struct {
+		name          string
+		pvName        string
+		sourcePV      *corev1api.PersistentVolume
+		targetPVC     *corev1api.PersistentVolumeClaim
+		kubeClientObj []runtime.Object
+		kubeReactors  []reactor
+		expectedErr   string
+		expected      *corev1api.PersistentVolume
+	}{
+		{
+			name:        "source is nil",
+			expectedErr: "source PV is required to rebind PV",
+		},
+		{
+			name:        "target pvc is nil",
+			sourcePV:    sourcePV,
+			expectedErr: "target PVC is required to rebind PV",
+		},
+		{
+			name:      "create error",
+			pvName:    "rebind-pv",
+			sourcePV:  sourcePV,
+			targetPVC: targetPVC,
+			kubeReactors: []reactor{
+				{
+					verb:     "create",
+					resource: "persistentvolumes",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("create error")
+					},
+				},
+			},
+			expectedErr: "create error",
+		},
+		{
+			name:      "success",
+			pvName:    "rebind-pv",
+			sourcePV:  sourcePV,
+			targetPVC: targetPVC,
+			expected: &corev1api.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "rebind-pv",
+					Labels: map[string]string{
+						"key1": "val3",
+						"key2": "val2",
+					},
+					Annotations: map[string]string{
+						"anno1": "val1",
+					},
+				},
+				Spec: corev1api.PersistentVolumeSpec{
+					Capacity: sourcePV.Spec.Capacity,
+					PersistentVolumeSource: corev1api.PersistentVolumeSource{
+						CSI: &corev1api.CSIPersistentVolumeSource{
+							Driver:       "fake-driver",
+							VolumeHandle: "fake-handle",
+							FSType:       "ext4",
+						},
+					},
+					AccessModes:                   sourcePV.Spec.AccessModes,
+					PersistentVolumeReclaimPolicy: corev1api.PersistentVolumeReclaimDelete,
+					StorageClassName:              sourcePV.Spec.StorageClassName,
+					VolumeMode:                    targetPVC.Spec.VolumeMode,
+					NodeAffinity:                  sourcePV.Spec.NodeAffinity,
+					ClaimRef: &corev1api.ObjectReference{
+						Kind:      targetPVC.Kind,
+						Namespace: "fake-ns",
+						Name:      "target-pvc",
+					},
+				},
+			},
+		},
+	}
+
+	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)
+			}
+
+			pv, err := RebindPV(t.Context(), fakeKubeClient.CoreV1(), test.pvName, test.sourcePV, test.targetPVC, corev1api.PersistentVolumeReclaimDelete, "ext4")
+
+			if test.expectedErr != "" {
+				assert.EqualError(t, err, test.expectedErr)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, test.expected, pv)
+			}
+		})
+	}
+}
+
+func TestClonePVSource(t *testing.T) {
+	fsTypeExt4 := "ext4"
+
+	tests := []struct {
+		name      string
+		source    *corev1api.PersistentVolumeSource
+		newFSType string
+		expected  corev1api.PersistentVolumeSource
+	}{
+		{
+			name: "no new fsType",
+			source: &corev1api.PersistentVolumeSource{
+				CSI: &corev1api.CSIPersistentVolumeSource{
+					Driver: "fake-driver",
+				},
+			},
+			newFSType: "",
+			expected: corev1api.PersistentVolumeSource{
+				CSI: &corev1api.CSIPersistentVolumeSource{
+					Driver: "fake-driver",
+				},
+			},
+		},
+		{
+			name: "csi source with new fsType",
+			source: &corev1api.PersistentVolumeSource{
+				CSI: &corev1api.CSIPersistentVolumeSource{
+					Driver: "fake-driver",
+					FSType: "ext3",
+				},
+			},
+			newFSType: "ext4",
+			expected: corev1api.PersistentVolumeSource{
+				CSI: &corev1api.CSIPersistentVolumeSource{
+					Driver: "fake-driver",
+					FSType: "ext4",
+				},
+			},
+		},
+		{
+			name: "awsEBS source with new fsType",
+			source: &corev1api.PersistentVolumeSource{
+				AWSElasticBlockStore: &corev1api.AWSElasticBlockStoreVolumeSource{
+					VolumeID: "fake-id",
+				},
+			},
+			newFSType: "ext4",
+			expected: corev1api.PersistentVolumeSource{
+				AWSElasticBlockStore: &corev1api.AWSElasticBlockStoreVolumeSource{
+					VolumeID: "fake-id",
+					FSType:   "ext4",
+				},
+			},
+		},
+		{
+			name: "azureDisk source with new fsType",
+			source: &corev1api.PersistentVolumeSource{
+				AzureDisk: &corev1api.AzureDiskVolumeSource{
+					DiskName: "fake-disk",
+				},
+			},
+			newFSType: "ext4",
+			expected: corev1api.PersistentVolumeSource{
+				AzureDisk: &corev1api.AzureDiskVolumeSource{
+					DiskName: "fake-disk",
+					FSType:   &fsTypeExt4,
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			actual := clonePVSource(test.source, test.newFSType)
+			assert.Equal(t, test.expected, actual)
+		})
+	}
+}