wait PV detachment before deleting PV

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

changelogs/unreleased/9932-Lyndon-Li

@@ -0,0 +1 @@
+Wait restorePV detached before binding the cloned PV to avoid confusing the CSI driver

pkg/exposer/generic_restore_test.go

@@ -443,6 +443,28 @@ func TestRebindVolume(t *testing.T) {
 			},
 			err: "error to delete restore PVC fake-restore: error to delete pvc fake-restore: fake-delete-error",
 		},
+		{
+			name:            "wait volume detached fail",
+			targetPVCName:   "fake-target-pvc",
+			targetNamespace: "fake-ns",
+			ownerRestore:    restore,
+			kubeClientObj: []runtime.Object{
+				targetPVCObj,
+				restorePVCObj,
+				restorePVObj,
+				restorePod,
+			},
+			kubeReactors: []reactor{
+				{
+					verb:     "list",
+					resource: "volumeattachments",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("fake-list-error")
+					},
+				},
+			},
+			err: "error waiting for retained PV fake-restore-pv to detach: error listing volumeattachment: error listing volumeattachment: fake-list-error",
+		},
 		{
 			name:            "rebind pv fail",
 			targetPVCName:   "fake-target-pvc",

pkg/util/kube/pvc_pv_test.go

@@ -26,6 +26,7 @@ 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/runtime/schema"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
 
@@ -2239,9 +2240,7 @@ func TestRebindPV(t *testing.T) {
 						"key1": "val3",
 						"key2": "val2",
 					},
-					Annotations: map[string]string{
-						"anno1": "val1",
-					},
+					Annotations: nil,
 				},
 				Spec: corev1api.PersistentVolumeSpec{
 					Capacity: sourcePV.Spec.Capacity,
@@ -2364,3 +2363,93 @@ func TestClonePVSource(t *testing.T) {
 		})
 	}
 }
+
+func TestWaitVolumeDetached(t *testing.T) {
+	pvName := "test-pv"
+	otherPVName := "other-pv"
+
+	volAttach1 := &storagev1api.VolumeAttachment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "va-1",
+		},
+		Spec: storagev1api.VolumeAttachmentSpec{
+			Source: storagev1api.VolumeAttachmentSource{
+				PersistentVolumeName: &pvName,
+			},
+		},
+	}
+
+	volAttach2 := &storagev1api.VolumeAttachment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "va-2",
+		},
+		Spec: storagev1api.VolumeAttachmentSpec{
+			Source: storagev1api.VolumeAttachmentSource{
+				PersistentVolumeName: &otherPVName,
+			},
+		},
+	}
+
+	tests := []struct {
+		name         string
+		kubeReactors []reactor
+		timeout      time.Duration
+		expectedErr  string
+		storageObjs  []runtime.Object
+	}{
+		{
+			name:    "no volume attachments",
+			timeout: time.Second,
+		},
+		{
+			name:        "volume attachments exist and deleted",
+			timeout:     time.Second,
+			storageObjs: []runtime.Object{volAttach1, volAttach2},
+			kubeReactors: []reactor{
+				{
+					verb:     "get",
+					resource: "volumeattachments",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, apierrors.NewNotFound(schema.GroupResource{Group: "storage.k8s.io", Resource: "volumeattachments"}, "va-1")
+					},
+				},
+			},
+		},
+		{
+			name:        "volume attachments exist and not deleted",
+			timeout:     2 * time.Millisecond,
+			storageObjs: []runtime.Object{volAttach1, volAttach2},
+			expectedErr: "timeout waiting for volume test-pv to be detached",
+		},
+		{
+			name:        "get returns error",
+			timeout:     time.Second,
+			storageObjs: []runtime.Object{volAttach1, volAttach2},
+			kubeReactors: []reactor{
+				{
+					verb:     "get",
+					resource: "volumeattachments",
+					reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
+						return true, nil, errors.New("fake-error")
+					},
+				},
+			},
+			expectedErr: "error waiting for volume test-pv to be detached: fake-error",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeKubeClient := fake.NewSimpleClientset(test.storageObjs...)
+			for _, reactor := range test.kubeReactors {
+				fakeKubeClient.Fake.PrependReactor(reactor.verb, reactor.resource, reactor.reactorFunc)
+			}
+			err := WaitVolumeDetached(t.Context(), fakeKubeClient.StorageV1(), pvName, test.timeout)
+			if test.expectedErr != "" {
+				assert.EqualError(t, err, test.expectedErr)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}