mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-09-19 22:44:21 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #9932 from Lyndon-Li/pv-rebind-test-2

Browse Source 
Wait restorePV detached before binding the cloned PV
This commit is contained in:
lyndon-li
2026-06-24 11:42:52 +08:00
committed by GitHub
parent dd5db4e863 3b66458edd
commit 960a20f12c
5 changed files with 180 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelogs/unreleased/9932-Lyndon-Li
+1
View File
@@ -0,0 +1 @@
Wait restorePV detached before binding the cloned PV to avoid confusing the CSI driver
pkg/exposer/generic_restore.go
+7
View File
@@ -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)
pkg/exposer/generic_restore_test.go
+22
View File
@@ -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.go
+58 -10
View File
@@ -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)
pkg/util/kube/pvc_pv_test.go
+92 -3
View File
@@ -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)
}
})
}
}