fix(volume): fix panic bug when enable ModifyVolume feature (#5058) (#…

…5082)

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
Co-authored-by: Bo Liu <liubo02@pingcap.com>
Co-authored-by: csuzhangxc <csuzhangxc@gmail.com>

pkg/manager/volumes/phase.go

@@ -77,44 +77,55 @@ func (p *podVolModifier) getVolumePhase(vol *ActualVolume) VolumePhase {
 		return VolumePhaseModified
 	}
 
-	if p.waitForNextTime(vol.PVC, vol.Desired.StorageClass) {
+	if p.waitForNextTime(vol.PVC, vol.StorageClass, vol.Desired.StorageClass) {
 		return VolumePhasePending
 	}
 
 	return VolumePhasePreparing
 }
 
-func isVolumeExpansionSupported(sc *storagev1.StorageClass) bool {
+func isVolumeExpansionSupported(sc *storagev1.StorageClass) (bool, error) {
+	if sc == nil {
+		// always assume expansion is supported
+		return true, fmt.Errorf("expansion cap of volume is unknown")
+	}
 	if sc.AllowVolumeExpansion == nil {
-		return false
+		return false, nil
 	}
-	return *sc.AllowVolumeExpansion
+	return *sc.AllowVolumeExpansion, nil
 }
 
 func (p *podVolModifier) validate(vol *ActualVolume) error {
 	if vol.Desired == nil {
 		return fmt.Errorf("can't match desired volume")
 	}
-	if vol.Desired.StorageClass == nil {
-		// TODO: support default storage class
-		return fmt.Errorf("can't change storage class to the default one")
-	}
-	desired := vol.Desired.Size
-	actual := getStorageSize(vol.PVC.Spec.Resources.Requests)
+	desired := vol.Desired.GetStorageSize()
+	actual := vol.GetStorageSize()
 	result := desired.Cmp(actual)
 	switch {
 	case result == 0:
 	case result < 0:
 		return fmt.Errorf("can't shrunk size from %s to %s", &actual, &desired)
 	case result > 0:
-		if !isVolumeExpansionSupported(vol.StorageClass) {
+		supported, err := isVolumeExpansionSupported(vol.StorageClass)
+		if err != nil {
+			klog.Warningf("volume expansion of storage class %s may be not supported, but it will be tried", vol.GetStorageClassName())
+		}
+		if !supported {
 			return fmt.Errorf("volume expansion is not supported by storageclass %s", vol.StorageClass.Name)
 		}
 	}
-	m := p.getVolumeModifier(vol.Desired.StorageClass)
+
+	m := p.getVolumeModifier(vol.StorageClass, vol.Desired.StorageClass)
 	if m == nil {
 		return nil
 	}
+
+	// if no pv permission but have sc permission: cannot change sc
+	if isStorageClassChanged(vol.GetStorageClassName(), vol.Desired.GetStorageClassName()) && vol.PV == nil {
+		return fmt.Errorf("cannot change storage class (%s to %s), because there is no permission to get persistent volume", vol.GetStorageClassName(), vol.Desired.GetStorageClassName())
+	}
+
 	desiredPVC := vol.PVC.DeepCopy()
 	desiredPVC.Spec.Resources.Requests[corev1.ResourceStorage] = desired
 
@@ -128,7 +139,7 @@ func isPVCRevisionChanged(pvc *corev1.PersistentVolumeClaim) bool {
 	return specRevision != statusRevision
 }
 
-func (p *podVolModifier) waitForNextTime(pvc *corev1.PersistentVolumeClaim, sc *storagev1.StorageClass) bool {
+func (p *podVolModifier) waitForNextTime(pvc *corev1.PersistentVolumeClaim, actualSc, desciredSc *storagev1.StorageClass) bool {
 	str, ok := pvc.Annotations[annoKeyPVCLastTransitionTimestamp]
 	if !ok {
 		return false
@@ -139,7 +150,7 @@ func (p *podVolModifier) waitForNextTime(pvc *corev1.PersistentVolumeClaim, sc *
 	}
 	d := time.Since(timestamp)
 
-	m := p.getVolumeModifier(sc)
+	m := p.getVolumeModifier(actualSc, desciredSc)
 
 	waitDur := defaultModifyWaitingDuration
 	if m != nil {
@@ -156,23 +167,14 @@ func (p *podVolModifier) waitForNextTime(pvc *corev1.PersistentVolumeClaim, sc *
 
 func needModify(pvc *corev1.PersistentVolumeClaim, desired *DesiredVolume) bool {
 	size := desired.Size
-	scName := ""
-	if desired.StorageClass != nil {
-		scName = desired.StorageClass.Name
-	}
+	scName := desired.GetStorageClassName()
 
 	return isPVCStatusMatched(pvc, scName, size)
 }
 
 func isPVCStatusMatched(pvc *corev1.PersistentVolumeClaim, scName string, size resource.Quantity) bool {
-	isChanged := false
-	oldSc, ok := pvc.Annotations[annoKeyPVCStatusStorageClass]
-	if !ok {
-		oldSc = ignoreNil(pvc.Spec.StorageClassName)
-	}
-	if oldSc != scName {
-		isChanged = true
-	}
+	oldSc := getStorageClassNameFromPVC(pvc)
+	isChanged := isStorageClassChanged(oldSc, scName)
 
 	oldSize, ok := pvc.Annotations[annoKeyPVCStatusStorageSize]
 	if !ok {
@@ -188,3 +190,10 @@ func isPVCStatusMatched(pvc *corev1.PersistentVolumeClaim, scName string, size r
 
 	return isChanged
 }
+
+func isStorageClassChanged(pre, cur string) bool {
+	if cur != "" && pre != cur {
+		return true
+	}
+	return false
+}

pkg/manager/volumes/phase_test.go

@@ -43,6 +43,11 @@ func newTestPVCForGetVolumePhase(size string, sc *string, annotations map[string
 			},
 			StorageClassName: sc,
 		},
+		Status: corev1.PersistentVolumeClaimStatus{
+			Capacity: corev1.ResourceList{
+				corev1.ResourceStorage: q,
+			},
+		},
 	}
 }
 
@@ -169,13 +174,22 @@ func TestGetVolumePhase(t *testing.T) {
 			expected: VolumePhasePreparing,
 		},
 		{
-			desc:  "invalid sc",
+			desc:  "sc is not set",
 			pvc:   newTestPVCForGetVolumePhase(oldSize, &oldScName, nil),
 			oldSc: newStorageClassForGetVolumePhase(oldScName, "ebs.csi.aws.com", true),
 			sc:    nil,
 			size:  oldSize,
 
-			expected: VolumePhaseCannotModify,
+			expected: VolumePhaseModified,
+		},
+		{
+			desc:  "sc is not set, but size is changed",
+			pvc:   newTestPVCForGetVolumePhase(oldSize, &oldScName, nil),
+			oldSc: newStorageClassForGetVolumePhase(oldScName, "ebs.csi.aws.com", true),
+			sc:    nil,
+			size:  newSize,
+
+			expected: VolumePhasePreparing,
 		},
 		{
 			desc:  "invalid size",
@@ -217,14 +231,22 @@ func TestGetVolumePhase(t *testing.T) {
 		actual := ActualVolume{
 			PVC:          c.pvc,
 			StorageClass: c.oldSc,
+			PV: &corev1.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test",
+				},
+			},
 		}
 		if !c.noDesired {
 			actual.Desired = &DesiredVolume{
 				StorageClass: c.sc,
 				Size:         resource.MustParse(c.size),
 			}
+			if c.sc != nil {
+				actual.Desired.StorageClassName = &c.sc.Name
+			}
 		}
 		phase := pvm.getVolumePhase(&actual)
-		g.Expect(phase).Should(Equal(c.expected), c.desc)
+		g.Expect(phase.String()).Should(Equal(c.expected.String()), c.desc)
 	}
 }