diff --git a/changelogs/unreleased/9932-Lyndon-Li b/changelogs/unreleased/9932-Lyndon-Li new file mode 100644 index 000000000..a09169cf2 --- /dev/null +++ b/changelogs/unreleased/9932-Lyndon-Li @@ -0,0 +1 @@ +Wait restorePV detached before binding the cloned PV to avoid confusing the CSI driver \ No newline at end of file diff --git a/pkg/exposer/generic_restore.go b/pkg/exposer/generic_restore.go index f0ad76123..5d0f34d99 100644 --- a/pkg/exposer/generic_restore.go +++ b/pkg/exposer/generic_restore.go @@ -452,6 +452,13 @@ func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject co curLog.WithField("restore PVC", restorePVCName).Info("Restore PVC is deleted") + err = kube.WaitVolumeDetached(ctx, e.kubeClient.StorageV1(), retained.Name, param.OperationTimeout) + if err != nil { + return errors.Wrapf(err, "error waiting for retained PV %s to detach", retained.Name) + } + + curLog.WithField("retained PV", retained.Name).Info("Retained PV is detached") + rebindPV, err = kube.RebindPV(ctx, e.kubeClient.CoreV1(), uuid.NewString(), retained, targetPVC, orgReclaim, param.TargetFSType) if err != nil { return errors.Wrapf(err, "error rebinding PV for target PVC %s", param.TargetPVCName) diff --git a/pkg/exposer/generic_restore_test.go b/pkg/exposer/generic_restore_test.go index c10bec06b..75da686e6 100644 --- a/pkg/exposer/generic_restore_test.go +++ b/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", diff --git a/pkg/util/kube/pvc_pv.go b/pkg/util/kube/pvc_pv.go index 7dea36f08..182b18995 100644 --- a/pkg/util/kube/pvc_pv.go +++ b/pkg/util/kube/pvc_pv.go @@ -324,15 +324,10 @@ func RebindPV(ctx context.Context, pvGetter corev1client.CoreV1Interface, pvName 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, + Name: pvName, + Labels: pvLabel, }, Spec: corev1api.PersistentVolumeSpec{ Capacity: source.Spec.Capacity, @@ -358,6 +353,10 @@ func RebindPV(ctx context.Context, pvGetter corev1client.CoreV1Interface, pvName func clonePVSource(source *corev1api.PersistentVolumeSource, newFSType string) corev1api.PersistentVolumeSource { newSource := source.DeepCopy() + if newSource.CSI != nil && newSource.CSI.VolumeAttributes != nil { + delete(newSource.CSI.VolumeAttributes, "storage.kubernetes.io/csiProvisionerIdentity") + } + if newFSType != "" { if newSource.CSI != nil { newSource.CSI.FSType = newFSType @@ -684,22 +683,71 @@ func GetPVAttachedNode(ctx context.Context, pv string, storageClient storagev1.S return "", nil } -func GetPVAttachedNodes(ctx context.Context, pv string, storageClient storagev1.StorageV1Interface) ([]string, error) { +func getPVAttachment(ctx context.Context, pv string, storageClient storagev1.StorageV1Interface) ([]*storagev1api.VolumeAttachment, error) { vaList, err := storageClient.VolumeAttachments().List(ctx, metav1.ListOptions{}) if err != nil { return nil, errors.Wrapf(err, "error listing volumeattachment") } - nodes := []string{} + attachments := []*storagev1api.VolumeAttachment{} for _, va := range vaList.Items { if va.Spec.Source.PersistentVolumeName != nil && *va.Spec.Source.PersistentVolumeName == pv { - nodes = append(nodes, va.Spec.NodeName) + attachments = append(attachments, &va) } } + return attachments, nil +} + +func GetPVAttachedNodes(ctx context.Context, pv string, storageClient storagev1.StorageV1Interface) ([]string, error) { + attachments, err := getPVAttachment(ctx, pv, storageClient) + if err != nil { + return nil, errors.Wrap(err, "error listing volumeattachment") + } + + nodes := []string{} + for _, attach := range attachments { + nodes = append(nodes, attach.Spec.NodeName) + } + return nodes, nil } +func WaitVolumeDetached(ctx context.Context, storageClient storagev1.StorageV1Interface, pv string, timeout time.Duration) error { + attachments, err := getPVAttachment(ctx, pv, storageClient) + if err != nil { + return errors.Wrap(err, "error listing volumeattachment") + } + + err = wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) { + left := []*storagev1api.VolumeAttachment{} + for _, attach := range attachments { + if _, err := storageClient.VolumeAttachments().Get(ctx, attach.Name, metav1.GetOptions{}); err == nil { + left = append(left, attach) + } else if !apierrors.IsNotFound(err) { + return false, err // Return the error if it's not a NotFound error + } + } + + if len(left) == 0 { + return true, nil + } + + attachments = left + + return false, nil + }) + + if err != nil { + if errors.Is(err, context.DeadlineExceeded) { + return errors.Errorf("timeout waiting for volume %s to be detached", pv) + } + return errors.Wrapf(err, "error waiting for volume %s to be detached", pv) + } + + return nil +} + func GetVolumeTopology(ctx context.Context, volumeClient corev1client.CoreV1Interface, storageClient storagev1.StorageV1Interface, pvName string, scName string) (*corev1api.NodeSelector, error) { if pvName == "" || scName == "" { return nil, errors.Errorf("invalid parameter, pv %s, sc %s", pvName, scName) diff --git a/pkg/util/kube/pvc_pv_test.go b/pkg/util/kube/pvc_pv_test.go index 0f1876ecd..9b93f2971 100644 --- a/pkg/util/kube/pvc_pv_test.go +++ b/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) + } + }) + } +}