Merge pull request #106154 from gnufied/recover-expansion-failure-123

Recover expansion failure

api/openapi-spec/swagger.json

@@ -7767,7 +7767,7 @@
         },
         "resources": {
           "$ref": "#/definitions/io.k8s.api.core.v1.ResourceRequirements",
-          "description": "Resources represents the minimum resources the volume should have. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#resources"
+          "description": "Resources represents the minimum resources the volume should have. If RecoverVolumeExpansionFailure feature is enabled users are allowed to specify resource requirements that are lower than previous value but must still be higher than capacity recorded in the status field of the claim. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#resources"
         },
         "selector": {
           "$ref": "#/definitions/io.k8s.apimachinery.pkg.apis.meta.v1.LabelSelector",
@@ -7798,6 +7798,13 @@
           },
           "type": "array"
         },
+        "allocatedResources": {
+          "additionalProperties": {
+            "$ref": "#/definitions/io.k8s.apimachinery.pkg.api.resource.Quantity"
+          },
+          "description": "The storage resource within AllocatedResources tracks the capacity allocated to a PVC. It may be larger than the actual capacity when a volume expansion operation is requested. For storage quota, the larger value from allocatedResources and PVC.spec.resources is used. If allocatedResources is not set, PVC.spec.resources alone is used for quota calculation. If a volume expansion capacity request is lowered, allocatedResources is only lowered if there are no expansion operations in progress and if the actual volume capacity is equal or lower than the requested capacity. This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.",
+          "type": "object"
+        },
         "capacity": {
           "additionalProperties": {
             "$ref": "#/definitions/io.k8s.apimachinery.pkg.api.resource.Quantity"
@@ -7817,6 +7824,10 @@
         "phase": {
           "description": "Phase represents the current phase of PersistentVolumeClaim.",
           "type": "string"
+        },
+        "resizeStatus": {
+          "description": "ResizeStatus stores status of resize operation. ResizeStatus is not set by default but when expansion is complete resizeStatus is set to empty string by resize controller or kubelet. This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.",
+          "type": "string"
         }
       },
       "type": "object"

pkg/api/persistentvolumeclaim/util.go

@@ -74,6 +74,21 @@ func EnforceDataSourceBackwardsCompatibility(pvcSpec, oldPVCSpec *core.Persisten
 	}
 }
 
+func DropDisabledFieldsFromStatus(pvc, oldPVC *core.PersistentVolumeClaim) {
+	if !utilfeature.DefaultFeatureGate.Enabled(features.ExpandPersistentVolumes) && oldPVC.Status.Conditions == nil {
+		pvc.Status.Conditions = nil
+	}
+
+	if !utilfeature.DefaultFeatureGate.Enabled(features.RecoverVolumeExpansionFailure) {
+		if !allocatedResourcesInUse(oldPVC) {
+			pvc.Status.AllocatedResources = nil
+		}
+		if !resizeStatusInUse(oldPVC) {
+			pvc.Status.ResizeStatus = nil
+		}
+	}
+}
+
 func dataSourceInUse(oldPVCSpec *core.PersistentVolumeClaimSpec) bool {
 	if oldPVCSpec == nil {
 		return false
@@ -118,3 +133,25 @@ func NormalizeDataSources(pvcSpec *core.PersistentVolumeClaimSpec) {
 		pvcSpec.DataSource = pvcSpec.DataSourceRef.DeepCopy()
 	}
 }
+
+func resizeStatusInUse(oldPVC *core.PersistentVolumeClaim) bool {
+	if oldPVC == nil {
+		return false
+	}
+	if oldPVC.Status.ResizeStatus != nil {
+		return true
+	}
+	return false
+}
+
+func allocatedResourcesInUse(oldPVC *core.PersistentVolumeClaim) bool {
+	if oldPVC == nil {
+		return false
+	}
+
+	if oldPVC.Status.AllocatedResources != nil {
+		return true
+	}
+
+	return false
+}

pkg/api/persistentvolumeclaim/util_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"k8s.io/apimachinery/pkg/api/resource"
 
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
@@ -288,3 +289,111 @@ func TestDataSourceRef(t *testing.T) {
 		})
 	}
 }
+
+func TestDropDisabledFieldsFromStatus(t *testing.T) {
+	tests := []struct {
+		name     string
+		feature  bool
+		pvc      *core.PersistentVolumeClaim
+		oldPVC   *core.PersistentVolumeClaim
+		expected *core.PersistentVolumeClaim
+	}{
+		{
+			name:     "for:newPVC=hasAllocatedResource,oldPVC=doesnot,featuregate=false; should drop field",
+			feature:  false,
+			pvc:      withAllocatedResource("5G"),
+			oldPVC:   getPVC(),
+			expected: getPVC(),
+		},
+		{
+			name:     "for:newPVC=hasAllocatedResource,oldPVC=doesnot,featuregate=true; should keep field",
+			feature:  true,
+			pvc:      withAllocatedResource("5G"),
+			oldPVC:   getPVC(),
+			expected: withAllocatedResource("5G"),
+		},
+		{
+			name:     "for:newPVC=hasAllocatedResource,oldPVC=hasAllocatedResource,featuregate=true; should keep field",
+			feature:  true,
+			pvc:      withAllocatedResource("5G"),
+			oldPVC:   withAllocatedResource("5G"),
+			expected: withAllocatedResource("5G"),
+		},
+		{
+			name:     "for:newPVC=hasAllocatedResource,oldPVC=hasAllocatedResource,featuregate=false; should keep field",
+			feature:  false,
+			pvc:      withAllocatedResource("10G"),
+			oldPVC:   withAllocatedResource("5G"),
+			expected: withAllocatedResource("10G"),
+		},
+		{
+			name:     "for:newPVC=hasAllocatedResource,oldPVC=nil,featuregate=false; should drop field",
+			feature:  false,
+			pvc:      withAllocatedResource("5G"),
+			oldPVC:   nil,
+			expected: getPVC(),
+		},
+		{
+			name:     "for:newPVC=hasResizeStatus,oldPVC=nil, featuregate=false should drop field",
+			feature:  false,
+			pvc:      withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+			oldPVC:   nil,
+			expected: getPVC(),
+		},
+		{
+			name:     "for:newPVC=hasResizeStatus,oldPVC=doesnot,featuregate=true; should keep field",
+			feature:  true,
+			pvc:      withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+			oldPVC:   getPVC(),
+			expected: withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+		},
+		{
+			name:     "for:newPVC=hasResizeStatus,oldPVC=hasResizeStatus,featuregate=true; should keep field",
+			feature:  true,
+			pvc:      withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+			oldPVC:   withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+			expected: withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+		},
+		{
+			name:     "for:newPVC=hasResizeStatus,oldPVC=hasResizeStatus,featuregate=false; should keep field",
+			feature:  false,
+			pvc:      withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+			oldPVC:   withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+			expected: withResizeStatus(core.PersistentVolumeClaimNodeExpansionFailed),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, test.feature)()
+
+			DropDisabledFieldsFromStatus(test.pvc, test.oldPVC)
+
+			if !reflect.DeepEqual(*test.expected, *test.pvc) {
+				t.Errorf("Unexpected change: %+v", cmp.Diff(test.expected, test.pvc))
+			}
+		})
+	}
+}
+
+func getPVC() *core.PersistentVolumeClaim {
+	return &core.PersistentVolumeClaim{}
+}
+
+func withAllocatedResource(q string) *core.PersistentVolumeClaim {
+	return &core.PersistentVolumeClaim{
+		Status: core.PersistentVolumeClaimStatus{
+			AllocatedResources: core.ResourceList{
+				core.ResourceStorage: resource.MustParse(q),
+			},
+		},
+	}
+}
+
+func withResizeStatus(status core.PersistentVolumeClaimResizeStatus) *core.PersistentVolumeClaim {
+	return &core.PersistentVolumeClaim{
+		Status: core.PersistentVolumeClaimStatus{
+			ResizeStatus: &status,
+		},
+	}
+}

pkg/apis/apps/v1/zz_generated.defaults.go

@@ -924,6 +924,7 @@ func SetObjectDefaults_StatefulSet(in *v1.StatefulSet) {
 		corev1.SetDefaults_ResourceList(&a.Spec.Resources.Limits)
 		corev1.SetDefaults_ResourceList(&a.Spec.Resources.Requests)
 		corev1.SetDefaults_ResourceList(&a.Status.Capacity)
+		corev1.SetDefaults_ResourceList(&a.Status.AllocatedResources)
 	}
 }
 

pkg/apis/apps/v1beta1/zz_generated.defaults.go

@@ -478,6 +478,7 @@ func SetObjectDefaults_StatefulSet(in *v1beta1.StatefulSet) {
 		v1.SetDefaults_ResourceList(&a.Spec.Resources.Limits)
 		v1.SetDefaults_ResourceList(&a.Spec.Resources.Requests)
 		v1.SetDefaults_ResourceList(&a.Status.Capacity)
+		v1.SetDefaults_ResourceList(&a.Status.AllocatedResources)
 	}
 }
 

pkg/apis/apps/v1beta2/zz_generated.defaults.go

@@ -924,6 +924,7 @@ func SetObjectDefaults_StatefulSet(in *v1beta2.StatefulSet) {
 		v1.SetDefaults_ResourceList(&a.Spec.Resources.Limits)
 		v1.SetDefaults_ResourceList(&a.Spec.Resources.Requests)
 		v1.SetDefaults_ResourceList(&a.Status.Capacity)
+		v1.SetDefaults_ResourceList(&a.Status.AllocatedResources)
 	}
 }
 

pkg/apis/core/types.go

@@ -430,6 +430,9 @@ type PersistentVolumeClaimSpec struct {
 	// +optional
 	Selector *metav1.LabelSelector
 	// Resources represents the minimum resources required
+	// If RecoverVolumeExpansionFailure feature is enabled users are allowed to specify resource requirements
+	// that are lower than previous value but must still be higher than capacity recorded in the
+	// status field of the claim.
 	// +optional
 	Resources ResourceRequirements
 	// VolumeName is the binding reference to the PersistentVolume backing this
@@ -486,6 +489,26 @@ const (
 	PersistentVolumeClaimFileSystemResizePending PersistentVolumeClaimConditionType = "FileSystemResizePending"
 )
 
+// +enum
+type PersistentVolumeClaimResizeStatus string
+
+const (
+	// When expansion is complete, the empty string is set by resize controller or kubelet.
+	PersistentVolumeClaimNoExpansionInProgress PersistentVolumeClaimResizeStatus = ""
+	// State set when resize controller starts expanding the volume in control-plane
+	PersistentVolumeClaimControllerExpansionInProgress PersistentVolumeClaimResizeStatus = "ControllerExpansionInProgress"
+	// State set when expansion has failed in resize controller with a terminal error.
+	// Transient errors such as timeout should not set this status and should leave ResizeStatus
+	// unmodified, so as resize controller can resume the volume expansion.
+	PersistentVolumeClaimControllerExpansionFailed PersistentVolumeClaimResizeStatus = "ControllerExpansionFailed"
+	// State set when resize controller has finished expanding the volume but further expansion is needed on the node.
+	PersistentVolumeClaimNodeExpansionPending PersistentVolumeClaimResizeStatus = "NodeExpansionPending"
+	// State set when kubelet starts expanding the volume.
+	PersistentVolumeClaimNodeExpansionInProgress PersistentVolumeClaimResizeStatus = "NodeExpansionInProgress"
+	// State set when expansion has failed in kubelet with a terminal error. Transient errors don't set NodeExpansionFailed.
+	PersistentVolumeClaimNodeExpansionFailed PersistentVolumeClaimResizeStatus = "NodeExpansionFailed"
+)
+
 // PersistentVolumeClaimCondition represents the current condition of PV claim
 type PersistentVolumeClaimCondition struct {
 	Type   PersistentVolumeClaimConditionType
@@ -513,6 +536,24 @@ type PersistentVolumeClaimStatus struct {
 	Capacity ResourceList
 	// +optional
 	Conditions []PersistentVolumeClaimCondition
+	// The storage resource within AllocatedResources tracks the capacity allocated to a PVC. It may
+	// be larger than the actual capacity when a volume expansion operation is requested.
+	// For storage quota, the larger value from allocatedResources and PVC.spec.resources is used.
+	// If allocatedResources is not set, PVC.spec.resources alone is used for quota calculation.
+	// If a volume expansion capacity request is lowered, allocatedResources is only
+	// lowered if there are no expansion operations in progress and if the actual volume capacity
+	// is equal or lower than the requested capacity.
+	// This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.
+	// +featureGate=RecoverVolumeExpansionFailure
+	// +optional
+	AllocatedResources ResourceList
+	// ResizeStatus stores status of resize operation.
+	// ResizeStatus is not set by default but when expansion is complete resizeStatus is set to empty
+	// string by resize controller or kubelet.
+	// This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.
+	// +featureGate=RecoverVolumeExpansionFailure
+	// +optional
+	ResizeStatus *PersistentVolumeClaimResizeStatus
 }
 
 // PersistentVolumeAccessMode defines various access modes for PV.

pkg/apis/core/v1/zz_generated.conversion.go

@@ -5171,6 +5171,8 @@ func autoConvert_v1_PersistentVolumeClaimStatus_To_core_PersistentVolumeClaimSta
 	out.AccessModes = *(*[]core.PersistentVolumeAccessMode)(unsafe.Pointer(&in.AccessModes))
 	out.Capacity = *(*core.ResourceList)(unsafe.Pointer(&in.Capacity))
 	out.Conditions = *(*[]core.PersistentVolumeClaimCondition)(unsafe.Pointer(&in.Conditions))
+	out.AllocatedResources = *(*core.ResourceList)(unsafe.Pointer(&in.AllocatedResources))
+	out.ResizeStatus = (*core.PersistentVolumeClaimResizeStatus)(unsafe.Pointer(in.ResizeStatus))
 	return nil
 }
 
@@ -5184,6 +5186,8 @@ func autoConvert_core_PersistentVolumeClaimStatus_To_v1_PersistentVolumeClaimSta
 	out.AccessModes = *(*[]v1.PersistentVolumeAccessMode)(unsafe.Pointer(&in.AccessModes))
 	out.Capacity = *(*v1.ResourceList)(unsafe.Pointer(&in.Capacity))
 	out.Conditions = *(*[]v1.PersistentVolumeClaimCondition)(unsafe.Pointer(&in.Conditions))
+	out.AllocatedResources = *(*v1.ResourceList)(unsafe.Pointer(&in.AllocatedResources))
+	out.ResizeStatus = (*v1.PersistentVolumeClaimResizeStatus)(unsafe.Pointer(in.ResizeStatus))
 	return nil
 }
 

pkg/apis/core/v1/zz_generated.defaults.go

@@ -155,6 +155,7 @@ func SetObjectDefaults_PersistentVolumeClaim(in *v1.PersistentVolumeClaim) {
 	SetDefaults_ResourceList(&in.Spec.Resources.Limits)
 	SetDefaults_ResourceList(&in.Spec.Resources.Requests)
 	SetDefaults_ResourceList(&in.Status.Capacity)
+	SetDefaults_ResourceList(&in.Status.AllocatedResources)
 }
 
 func SetObjectDefaults_PersistentVolumeClaimList(in *v1.PersistentVolumeClaimList) {

pkg/apis/core/validation/validation.go

@@ -2020,20 +2020,26 @@ func ValidatePersistentVolumeStatusUpdate(newPv, oldPv *core.PersistentVolume) f
 	return allErrs
 }
 
-// PersistentVolumeClaimSpecValidationOptions contains the different settings for PersistentVolumeClaim validation
 type PersistentVolumeClaimSpecValidationOptions struct {
 	// Allow spec to contain the "ReadWiteOncePod" access mode
 	AllowReadWriteOncePod bool
+	// Allow pvc expansion after PVC is created and bound to a PV
+	EnableExpansion bool
+	// Allow users to recover from previously failing expansion operation
+	EnableRecoverFromExpansionFailure bool
 }
 
 func ValidationOptionsForPersistentVolumeClaim(pvc, oldPvc *core.PersistentVolumeClaim) PersistentVolumeClaimSpecValidationOptions {
 	opts := PersistentVolumeClaimSpecValidationOptions{
-		AllowReadWriteOncePod: utilfeature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
+		AllowReadWriteOncePod:             utilfeature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
+		EnableExpansion:                   utilfeature.DefaultFeatureGate.Enabled(features.ExpandPersistentVolumes),
+		EnableRecoverFromExpansionFailure: utilfeature.DefaultFeatureGate.Enabled(features.RecoverVolumeExpansionFailure),
 	}
 	if oldPvc == nil {
 		// If there's no old PVC, use the options based solely on feature enablement
 		return opts
 	}
+
 	if helper.ContainsAccessMode(oldPvc.Spec.AccessModes, core.ReadWriteOncePod) {
 		// If the old object allowed "ReadWriteOncePod", continue to allow it in the new object
 		opts.AllowReadWriteOncePod = true
@@ -2173,7 +2179,7 @@ func ValidatePersistentVolumeClaimUpdate(newPvc, oldPvc *core.PersistentVolumeCl
 		allErrs = append(allErrs, ValidateImmutableAnnotation(newPvc.ObjectMeta.Annotations[v1.BetaStorageClassAnnotation], oldPvc.ObjectMeta.Annotations[v1.BetaStorageClassAnnotation], v1.BetaStorageClassAnnotation, field.NewPath("metadata"))...)
 	}
 
-	if utilfeature.DefaultFeatureGate.Enabled(features.ExpandPersistentVolumes) {
+	if opts.EnableExpansion {
 		// lets make sure storage values are same.
 		if newPvc.Status.Phase == core.ClaimBound && newPvcClone.Spec.Resources.Requests != nil {
 			newPvcClone.Spec.Resources.Requests["storage"] = oldPvc.Spec.Resources.Requests["storage"] // +k8s:verify-mutation:reason=clone
@@ -2181,13 +2187,23 @@ func ValidatePersistentVolumeClaimUpdate(newPvc, oldPvc *core.PersistentVolumeCl
 
 		oldSize := oldPvc.Spec.Resources.Requests["storage"]
 		newSize := newPvc.Spec.Resources.Requests["storage"]
+		statusSize := oldPvc.Status.Capacity["storage"]
 
 		if !apiequality.Semantic.DeepEqual(newPvcClone.Spec, oldPvcClone.Spec) {
 			specDiff := diff.ObjectDiff(newPvcClone.Spec, oldPvcClone.Spec)
 			allErrs = append(allErrs, field.Forbidden(field.NewPath("spec"), fmt.Sprintf("spec is immutable after creation except resources.requests for bound claims\n%v", specDiff)))
 		}
 		if newSize.Cmp(oldSize) < 0 {
-			allErrs = append(allErrs, field.Forbidden(field.NewPath("spec", "resources", "requests", "storage"), "field can not be less than previous value"))
+			if !opts.EnableRecoverFromExpansionFailure {
+				allErrs = append(allErrs, field.Forbidden(field.NewPath("spec", "resources", "requests", "storage"), "field can not be less than previous value"))
+			} else {
+				// This validation permits reducing pvc requested size up to capacity recorded in pvc.status
+				// so that users can recover from volume expansion failure, but Kubernetes does not actually
+				// support volume shrinking
+				if newSize.Cmp(statusSize) <= 0 {
+					allErrs = append(allErrs, field.Forbidden(field.NewPath("spec", "resources", "requests", "storage"), "field can not be less than status.capacity"))
+				}
+			}
 		}
 
 	} else {
@@ -2220,8 +2236,15 @@ func validateStorageClassUpgrade(oldAnnotations, newAnnotations map[string]strin
 		(!newAnnotationExist || newScInAnnotation == oldSc) /* condition 4 */
 }
 
+var resizeStatusSet = sets.NewString(string(core.PersistentVolumeClaimNoExpansionInProgress),
+	string(core.PersistentVolumeClaimControllerExpansionInProgress),
+	string(core.PersistentVolumeClaimControllerExpansionFailed),
+	string(core.PersistentVolumeClaimNodeExpansionPending),
+	string(core.PersistentVolumeClaimNodeExpansionInProgress),
+	string(core.PersistentVolumeClaimNodeExpansionFailed))
+
 // ValidatePersistentVolumeClaimStatusUpdate validates an update to status of a PersistentVolumeClaim
-func ValidatePersistentVolumeClaimStatusUpdate(newPvc, oldPvc *core.PersistentVolumeClaim) field.ErrorList {
+func ValidatePersistentVolumeClaimStatusUpdate(newPvc, oldPvc *core.PersistentVolumeClaim, validationOpts PersistentVolumeClaimSpecValidationOptions) field.ErrorList {
 	allErrs := ValidateObjectMetaUpdate(&newPvc.ObjectMeta, &oldPvc.ObjectMeta, field.NewPath("metadata"))
 	if len(newPvc.ResourceVersion) == 0 {
 		allErrs = append(allErrs, field.Required(field.NewPath("resourceVersion"), ""))
@@ -2229,10 +2252,32 @@ func ValidatePersistentVolumeClaimStatusUpdate(newPvc, oldPvc *core.PersistentVo
 	if len(newPvc.Spec.AccessModes) == 0 {
 		allErrs = append(allErrs, field.Required(field.NewPath("Spec", "accessModes"), ""))
 	}
+
 	capPath := field.NewPath("status", "capacity")
 	for r, qty := range newPvc.Status.Capacity {
 		allErrs = append(allErrs, validateBasicResource(qty, capPath.Key(string(r)))...)
 	}
+	if validationOpts.EnableRecoverFromExpansionFailure {
+		resizeStatusPath := field.NewPath("status", "resizeStatus")
+		if newPvc.Status.ResizeStatus != nil {
+			resizeStatus := *newPvc.Status.ResizeStatus
+			if !resizeStatusSet.Has(string(resizeStatus)) {
+				allErrs = append(allErrs, field.NotSupported(resizeStatusPath, resizeStatus, resizeStatusSet.List()))
+			}
+		}
+		allocPath := field.NewPath("status", "allocatedResources")
+		for r, qty := range newPvc.Status.AllocatedResources {
+			if r != core.ResourceStorage {
+				allErrs = append(allErrs, field.NotSupported(allocPath, r, []string{string(core.ResourceStorage)}))
+				continue
+			}
+			if errs := validateBasicResource(qty, allocPath.Key(string(r))); len(errs) > 0 {
+				allErrs = append(allErrs, errs...)
+			} else {
+				allErrs = append(allErrs, ValidateResourceQuantityValue(string(core.ResourceStorage), qty, allocPath.Key(string(r)))...)
+			}
+		}
+	}
 	return allErrs
 }
 

pkg/apis/core/validation/validation_test.go

@@ -1862,12 +1862,97 @@ func TestValidatePersistentVolumeClaimUpdate(t *testing.T) {
 		},
 		VolumeName: "volume",
 	})
+	validClaimShrinkInitial := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("15G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimBound,
+		Capacity: core.ResourceList{
+			core.ResourceStorage: resource.MustParse("10G"),
+		},
+	})
+
+	unboundShrink := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("12G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimPending,
+		Capacity: core.ResourceList{
+			core.ResourceStorage: resource.MustParse("10G"),
+		},
+	})
+
+	validClaimShrink := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceStorage: resource.MustParse("13G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimBound,
+		Capacity: core.ResourceList{
+			core.ResourceStorage: resource.MustParse("10G"),
+		},
+	})
+
+	invalidShrinkToStatus := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceStorage: resource.MustParse("10G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimBound,
+		Capacity: core.ResourceList{
+			core.ResourceStorage: resource.MustParse("10G"),
+		},
+	})
+
+	invalidClaimShrink := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceStorage: resource.MustParse("3G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimBound,
+		Capacity: core.ResourceList{
+			core.ResourceStorage: resource.MustParse("10G"),
+		},
+	})
 
 	scenarios := map[string]struct {
-		isExpectedFailure bool
+		isExpectedFailure          bool
-		oldClaim          *core.PersistentVolumeClaim
+		oldClaim                   *core.PersistentVolumeClaim
-		newClaim          *core.PersistentVolumeClaim
+		newClaim                   *core.PersistentVolumeClaim
-		enableResize      bool
+		enableResize               bool
+		enableRecoverFromExpansion bool
 	}{
 		"valid-update-volumeName-only": {
 			isExpectedFailure: false,
@@ -2037,12 +2122,53 @@ func TestValidatePersistentVolumeClaimUpdate(t *testing.T) {
 			newClaim:          validClaimRWOPAccessModeAddAnnotation,
 			enableResize:      false,
 		},
+		"valid-expand-shrink-resize-enabled": {
+			oldClaim:                   validClaimShrinkInitial,
+			newClaim:                   validClaimShrink,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+		},
+		"invalid-expand-shrink-resize-enabled": {
+			oldClaim:                   validClaimShrinkInitial,
+			newClaim:                   invalidClaimShrink,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+			isExpectedFailure:          true,
+		},
+		"invalid-expand-shrink-to-status-resize-enabled": {
+			oldClaim:                   validClaimShrinkInitial,
+			newClaim:                   invalidShrinkToStatus,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+			isExpectedFailure:          true,
+		},
+		"invalid-expand-shrink-recover-disabled": {
+			oldClaim:                   validClaimShrinkInitial,
+			newClaim:                   validClaimShrink,
+			enableResize:               true,
+			enableRecoverFromExpansion: false,
+			isExpectedFailure:          true,
+		},
+		"invalid-expand-shrink-resize-disabled": {
+			oldClaim:                   validClaimShrinkInitial,
+			newClaim:                   validClaimShrink,
+			enableResize:               false,
+			enableRecoverFromExpansion: true,
+			isExpectedFailure:          true,
+		},
+		"unbound-size-shrink-resize-enabled": {
+			oldClaim:                   validClaimShrinkInitial,
+			newClaim:                   unboundShrink,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+			isExpectedFailure:          true,
+		},
 	}
 
 	for name, scenario := range scenarios {
 		t.Run(name, func(t *testing.T) {
-			// ensure we have a resource version specified for updates
 			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ExpandPersistentVolumes, scenario.enableResize)()
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, scenario.enableRecoverFromExpansion)()
 			scenario.oldClaim.ResourceVersion = "1"
 			scenario.newClaim.ResourceVersion = "1"
 			opts := ValidationOptionsForPersistentVolumeClaim(scenario.newClaim, scenario.oldClaim)
@@ -2067,35 +2193,45 @@ func TestValidationOptionsForPersistentVolumeClaim(t *testing.T) {
 			oldPvc:                 nil,
 			enableReadWriteOncePod: true,
 			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
-				AllowReadWriteOncePod: true,
+				AllowReadWriteOncePod:             true,
+				EnableExpansion:                   true,
+				EnableRecoverFromExpansionFailure: false,
 			},
 		},
 		"rwop allowed because feature enabled": {
 			oldPvc:                 pvcWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOnce}),
 			enableReadWriteOncePod: true,
 			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
-				AllowReadWriteOncePod: true,
+				AllowReadWriteOncePod:             true,
+				EnableExpansion:                   true,
+				EnableRecoverFromExpansionFailure: false,
 			},
 		},
 		"rwop not allowed because not used and feature disabled": {
 			oldPvc:                 pvcWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOnce}),
 			enableReadWriteOncePod: false,
 			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
-				AllowReadWriteOncePod: false,
+				AllowReadWriteOncePod:             false,
+				EnableExpansion:                   true,
+				EnableRecoverFromExpansionFailure: false,
 			},
 		},
 		"rwop allowed because used and feature enabled": {
 			oldPvc:                 pvcWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOncePod}),
 			enableReadWriteOncePod: true,
 			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
-				AllowReadWriteOncePod: true,
+				AllowReadWriteOncePod:             true,
+				EnableExpansion:                   true,
+				EnableRecoverFromExpansionFailure: false,
 			},
 		},
 		"rwop allowed because used and feature disabled": {
 			oldPvc:                 pvcWithAccessModes([]core.PersistentVolumeAccessMode{core.ReadWriteOncePod}),
 			enableReadWriteOncePod: false,
 			expectValidationOpts: PersistentVolumeClaimSpecValidationOptions{
-				AllowReadWriteOncePod: true,
+				AllowReadWriteOncePod:             true,
+				EnableExpansion:                   true,
+				EnableRecoverFromExpansionFailure: false,
 			},
 		},
 	}
@@ -15771,11 +15907,86 @@ func TestValidatePersistentVolumeClaimStatusUpdate(t *testing.T) {
 			{Type: core.PersistentVolumeClaimResizing, Status: core.ConditionTrue},
 		},
 	})
+	validAllocatedResources := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimPending,
+		Conditions: []core.PersistentVolumeClaimCondition{
+			{Type: core.PersistentVolumeClaimResizing, Status: core.ConditionTrue},
+		},
+		AllocatedResources: core.ResourceList{
+			core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+		},
+	})
+
+	invalidAllocatedResources := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+			core.ReadOnlyMany,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimPending,
+		Conditions: []core.PersistentVolumeClaimCondition{
+			{Type: core.PersistentVolumeClaimResizing, Status: core.ConditionTrue},
+		},
+		AllocatedResources: core.ResourceList{
+			core.ResourceName(core.ResourceStorage): resource.MustParse("-10G"),
+		},
+	})
+
+	noStoraegeClaimStatus := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+		},
+		Resources: core.ResourceRequirements{
+			Requests: core.ResourceList{
+				core.ResourceName(core.ResourceStorage): resource.MustParse("10G"),
+			},
+		},
+	}, core.PersistentVolumeClaimStatus{
+		Phase: core.ClaimPending,
+		AllocatedResources: core.ResourceList{
+			core.ResourceName(core.ResourceCPU): resource.MustParse("10G"),
+		},
+	})
+	progressResizeStatus := core.PersistentVolumeClaimControllerExpansionInProgress
+	invalidResizeStatus := core.PersistentVolumeClaimResizeStatus("foo")
+
+	validResizeStatusPVC := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+		},
+	}, core.PersistentVolumeClaimStatus{
+		ResizeStatus: &progressResizeStatus,
+	})
+
+	invalidResizeStatusPVC := testVolumeClaimWithStatus("foo", "ns", core.PersistentVolumeClaimSpec{
+		AccessModes: []core.PersistentVolumeAccessMode{
+			core.ReadWriteOnce,
+		},
+	}, core.PersistentVolumeClaimStatus{
+		ResizeStatus: &invalidResizeStatus,
+	})
+
 	scenarios := map[string]struct {
-		isExpectedFailure bool
+		isExpectedFailure          bool
-		oldClaim          *core.PersistentVolumeClaim
+		oldClaim                   *core.PersistentVolumeClaim
-		newClaim          *core.PersistentVolumeClaim
+		newClaim                   *core.PersistentVolumeClaim
-		enableResize      bool
+		enableResize               bool
+		enableRecoverFromExpansion bool
 	}{
 		"condition-update-with-enabled-feature-gate": {
 			isExpectedFailure: false,
@@ -15783,13 +15994,51 @@ func TestValidatePersistentVolumeClaimStatusUpdate(t *testing.T) {
 			newClaim:          validConditionUpdate,
 			enableResize:      true,
 		},
+		"status-update-with-valid-allocatedResources-feature-enabled": {
+			isExpectedFailure:          false,
+			oldClaim:                   validClaim,
+			newClaim:                   validAllocatedResources,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+		},
+		"status-update-with-invalid-allocatedResources-feature-enabled": {
+			isExpectedFailure:          true,
+			oldClaim:                   validClaim,
+			newClaim:                   invalidAllocatedResources,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+		},
+		"status-update-with-no-storage-update": {
+			isExpectedFailure:          true,
+			oldClaim:                   validClaim,
+			newClaim:                   noStoraegeClaimStatus,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+		},
+		"status-update-with-valid-pvc-resize-status": {
+			isExpectedFailure:          false,
+			oldClaim:                   validClaim,
+			newClaim:                   validResizeStatusPVC,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+		},
+		"status-update-with-invalid-pvc-resize-status": {
+			isExpectedFailure:          true,
+			oldClaim:                   validClaim,
+			newClaim:                   invalidResizeStatusPVC,
+			enableResize:               true,
+			enableRecoverFromExpansion: true,
+		},
 	}
 	for name, scenario := range scenarios {
 		t.Run(name, func(t *testing.T) {
+			validateOpts := PersistentVolumeClaimSpecValidationOptions{
+				EnableRecoverFromExpansionFailure: scenario.enableRecoverFromExpansion,
+			}
 			// ensure we have a resource version specified for updates
 			scenario.oldClaim.ResourceVersion = "1"
 			scenario.newClaim.ResourceVersion = "1"
-			errs := ValidatePersistentVolumeClaimStatusUpdate(scenario.newClaim, scenario.oldClaim)
+			errs := ValidatePersistentVolumeClaimStatusUpdate(scenario.newClaim, scenario.oldClaim, validateOpts)
 			if len(errs) == 0 && scenario.isExpectedFailure {
 				t.Errorf("Unexpected success for scenario: %s", name)
 			}

pkg/apis/core/zz_generated.deepcopy.go

@@ -2977,6 +2977,18 @@ func (in *PersistentVolumeClaimStatus) DeepCopyInto(out *PersistentVolumeClaimSt
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.AllocatedResources != nil {
+		in, out := &in.AllocatedResources, &out.AllocatedResources
+		*out = make(ResourceList, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val.DeepCopy()
+		}
+	}
+	if in.ResizeStatus != nil {
+		in, out := &in.ResizeStatus, &out.ResizeStatus
+		*out = new(PersistentVolumeClaimResizeStatus)
+		**out = **in
+	}
 	return
 }
 

pkg/controller/resourcequota/resource_quota_monitor.go

@@ -23,7 +23,7 @@ import (
 
 	"k8s.io/klog/v2"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
@@ -148,6 +148,10 @@ func (qm *QuotaMonitor) controllerFor(resource schema.GroupVersionResource) (cac
 				oldService := oldObj.(*v1.Service)
 				newService := newObj.(*v1.Service)
 				notifyUpdate = core.GetQuotaServiceType(oldService) != core.GetQuotaServiceType(newService)
+			case schema.GroupResource{Resource: "persistentvolumeclaims"}:
+				oldPVC := oldObj.(*v1.PersistentVolumeClaim)
+				newPVC := newObj.(*v1.PersistentVolumeClaim)
+				notifyUpdate = core.RequiresQuotaReplenish(newPVC, oldPVC)
 			}
 			if notifyUpdate {
 				event := &event{

pkg/controller/volume/expand/expand_controller.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -44,12 +45,14 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	cloudprovider "k8s.io/cloud-provider"
 	"k8s.io/kubernetes/pkg/controller/volume/events"
+	"k8s.io/kubernetes/pkg/features"
 	proxyutil "k8s.io/kubernetes/pkg/proxy/util"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/csimigration"
 	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
 	"k8s.io/kubernetes/pkg/volume/util/subpath"
+	volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
 	"k8s.io/kubernetes/pkg/volume/util/volumepathhandler"
 )
 
@@ -302,19 +305,31 @@ func (expc *expandController) expand(pvc *v1.PersistentVolumeClaim, pv *v1.Persi
 		return util.DeleteAnnPreResizeCapacity(pv, expc.GetKubeClient())
 	}
 
-	pvc, err := util.MarkResizeInProgressWithResizer(pvc, resizerName, expc.kubeClient)
+	var generatedOptions volumetypes.GeneratedOperations
-	if err != nil {
+	var err error
-		klog.V(5).Infof("Error setting PVC %s in progress with error : %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+
-		return err
+	if utilfeature.DefaultFeatureGate.Enabled(features.RecoverVolumeExpansionFailure) {
+		generatedOptions, err = expc.operationGenerator.GenerateExpandAndRecoverVolumeFunc(pvc, pv, resizerName)
+		if err != nil {
+			klog.Errorf("Error starting ExpandVolume for pvc %s with %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+			return err
+		}
+	} else {
+		pvc, err := util.MarkResizeInProgressWithResizer(pvc, resizerName, expc.kubeClient)
+		if err != nil {
+			klog.Errorf("Error setting PVC %s in progress with error : %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+			return err
+		}
+
+		generatedOptions, err = expc.operationGenerator.GenerateExpandVolumeFunc(pvc, pv)
+		if err != nil {
+			klog.Errorf("Error starting ExpandVolume for pvc %s with %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+			return err
+		}
 	}
 
-	generatedOperations, err := expc.operationGenerator.GenerateExpandVolumeFunc(pvc, pv)
-	if err != nil {
-		klog.Errorf("Error starting ExpandVolume for pvc %s with %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
-		return err
-	}
 	klog.V(5).Infof("Starting ExpandVolume for volume %s", util.GetPersistentVolumeClaimQualifiedName(pvc))
-	_, detailedErr := generatedOperations.Run()
+	_, detailedErr := generatedOptions.Run()
 
 	return detailedErr
 }
@@ -357,8 +372,15 @@ func (expc *expandController) getPersistentVolume(ctx context.Context, pvc *v1.P
 // isNodeExpandComplete returns true if  pvc.Status.Capacity >= pv.Spec.Capacity
 func (expc *expandController) isNodeExpandComplete(pvc *v1.PersistentVolumeClaim, pv *v1.PersistentVolume) bool {
 	klog.V(4).Infof("pv %q capacity = %v, pvc %s capacity = %v", pv.Name, pv.Spec.Capacity[v1.ResourceStorage], pvc.ObjectMeta.Name, pvc.Status.Capacity[v1.ResourceStorage])
-	pvcCap, pvCap := pvc.Status.Capacity[v1.ResourceStorage], pv.Spec.Capacity[v1.ResourceStorage]
+	pvcSpecCap := pvc.Spec.Resources.Requests.Storage()
-	return pvcCap.Cmp(pvCap) >= 0
+	pvcStatusCap, pvCap := pvc.Status.Capacity[v1.ResourceStorage], pv.Spec.Capacity[v1.ResourceStorage]
+
+	// since we allow shrinking volumes, we must compare both pvc status and capacity
+	// with pv spec capacity.
+	if pvcStatusCap.Cmp(*pvcSpecCap) >= 0 && pvcStatusCap.Cmp(pvCap) >= 0 {
+		return true
+	}
+	return false
 }
 
 // Implementing VolumeHost interface

pkg/features/kube_features.go

@@ -816,6 +816,13 @@ const (
 	// Honor Persistent Volume Reclaim Policy when it is "Delete" irrespective of PV-PVC
 	// deletion ordering.
 	HonorPVReclaimPolicy featuregate.Feature = "HonorPVReclaimPolicy"
+
+	// owner: @gnufied
+	// kep: http://kep.k8s.io/1790
+	// alpha: v1.23
+	//
+	// Allow users to recover from volume expansion failure
+	RecoverVolumeExpansionFailure featuregate.Feature = "RecoverVolumeExpansionFailure"
 )
 
 func init() {
@@ -935,6 +942,7 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 	IdentifyPodOS:                                  {Default: false, PreRelease: featuregate.Alpha},
 	PodAndContainerStatsFromCRI:                    {Default: false, PreRelease: featuregate.Alpha},
 	HonorPVReclaimPolicy:                           {Default: false, PreRelease: featuregate.Alpha},
+	RecoverVolumeExpansionFailure:                  {Default: false, PreRelease: featuregate.Alpha},
 
 	// inherited features from generic apiserver, relisted here to get a conflict if it is changed
 	// unintentionally on either side:

pkg/quota/v1/evaluator/core/persistent_volume_claims.go

@@ -159,25 +159,50 @@ func (p *pvcEvaluator) Usage(item runtime.Object) (corev1.ResourceList, error) {
 		result[storageClassClaim] = *(resource.NewQuantity(1, resource.DecimalSI))
 	}
 
-	// charge for storage
+	requestedStorage := p.getStorageUsage(pvc)
-	if request, found := pvc.Spec.Resources.Requests[corev1.ResourceStorage]; found {
+	if requestedStorage != nil {
-		roundedRequest := request.DeepCopy()
+		result[corev1.ResourceRequestsStorage] = *requestedStorage
-		if !roundedRequest.RoundUp(0) {
-			// Ensure storage requests are counted as whole byte values, to pass resourcequota validation.
-			// See http://issue.k8s.io/94313
-			request = roundedRequest
-		}
-
-		result[corev1.ResourceRequestsStorage] = request
 		// charge usage to the storage class (if present)
 		if len(storageClassRef) > 0 {
 			storageClassStorage := corev1.ResourceName(storageClassRef + storageClassSuffix + string(corev1.ResourceRequestsStorage))
-			result[storageClassStorage] = request
+			result[storageClassStorage] = *requestedStorage
 		}
 	}
+
 	return result, nil
 }
 
+func (p *pvcEvaluator) getStorageUsage(pvc *corev1.PersistentVolumeClaim) *resource.Quantity {
+	var result *resource.Quantity
+	roundUpFunc := func(i *resource.Quantity) *resource.Quantity {
+		roundedRequest := i.DeepCopy()
+		if !roundedRequest.RoundUp(0) {
+			// Ensure storage requests are counted as whole byte values, to pass resourcequota validation.
+			// See http://issue.k8s.io/94313
+			return &roundedRequest
+		}
+		return i
+	}
+
+	if userRequest, ok := pvc.Spec.Resources.Requests[corev1.ResourceStorage]; ok {
+		result = roundUpFunc(&userRequest)
+	}
+
+	if utilfeature.DefaultFeatureGate.Enabled(k8sfeatures.RecoverVolumeExpansionFailure) && result != nil {
+		if len(pvc.Status.AllocatedResources) == 0 {
+			return result
+		}
+
+		// if AllocatedResources is set and is greater than user request, we should use it.
+		if allocatedRequest, ok := pvc.Status.AllocatedResources[corev1.ResourceStorage]; ok {
+			if allocatedRequest.Cmp(*result) > 0 {
+				result = roundUpFunc(&allocatedRequest)
+			}
+		}
+	}
+	return result
+}
+
 // UsageStats calculates aggregate usage for the object.
 func (p *pvcEvaluator) UsageStats(options quota.UsageStatsOptions) (quota.UsageStats, error) {
 	return generic.CalculateUsageStats(options, p.listFuncByNamespace, generic.MatchesNoScopeFunc, p.Usage)
@@ -200,3 +225,13 @@ func toExternalPersistentVolumeClaimOrError(obj runtime.Object) (*corev1.Persist
 	}
 	return pvc, nil
 }
+
+// RequiresQuotaReplenish enables quota monitoring for PVCs.
+func RequiresQuotaReplenish(pvc, oldPVC *corev1.PersistentVolumeClaim) bool {
+	if utilfeature.DefaultFeatureGate.Enabled(k8sfeatures.RecoverVolumeExpansionFailure) {
+		if oldPVC.Status.AllocatedResources.Storage() != pvc.Status.AllocatedResources.Storage() {
+			return true
+		}
+	}
+	return false
+}

pkg/quota/v1/evaluator/core/persistent_volume_claims_test.go

@@ -25,7 +25,11 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	quota "k8s.io/apiserver/pkg/quota/v1"
 	"k8s.io/apiserver/pkg/quota/v1/generic"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/kubernetes/pkg/apis/core"
 	api "k8s.io/kubernetes/pkg/apis/core"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 func testVolumeClaim(name string, namespace string, spec api.PersistentVolumeClaimSpec) *api.PersistentVolumeClaim {
@@ -85,8 +89,9 @@ func TestPersistentVolumeClaimEvaluatorUsage(t *testing.T) {
 
 	evaluator := NewPersistentVolumeClaimEvaluator(nil)
 	testCases := map[string]struct {
-		pvc   *api.PersistentVolumeClaim
+		pvc                        *api.PersistentVolumeClaim
-		usage corev1.ResourceList
+		usage                      corev1.ResourceList
+		enableRecoverFromExpansion bool
 	}{
 		"pvc-usage": {
 			pvc: validClaim,
@@ -95,6 +100,7 @@ func TestPersistentVolumeClaimEvaluatorUsage(t *testing.T) {
 				corev1.ResourcePersistentVolumeClaims: resource.MustParse("1"),
 				generic.ObjectCountQuotaResourceNameFor(schema.GroupResource{Resource: "persistentvolumeclaims"}): resource.MustParse("1"),
 			},
+			enableRecoverFromExpansion: true,
 		},
 		"pvc-usage-by-class": {
 			pvc: validClaimByStorageClass,
@@ -105,6 +111,7 @@ func TestPersistentVolumeClaimEvaluatorUsage(t *testing.T) {
 				V1ResourceByStorageClass(classGold, corev1.ResourcePersistentVolumeClaims):                        resource.MustParse("1"),
 				generic.ObjectCountQuotaResourceNameFor(schema.GroupResource{Resource: "persistentvolumeclaims"}): resource.MustParse("1"),
 			},
+			enableRecoverFromExpansion: true,
 		},
 
 		"pvc-usage-rounded": {
@@ -114,6 +121,7 @@ func TestPersistentVolumeClaimEvaluatorUsage(t *testing.T) {
 				corev1.ResourcePersistentVolumeClaims: resource.MustParse("1"),
 				generic.ObjectCountQuotaResourceNameFor(schema.GroupResource{Resource: "persistentvolumeclaims"}): resource.MustParse("1"),
 			},
+			enableRecoverFromExpansion: true,
 		},
 		"pvc-usage-by-class-rounded": {
 			pvc: validClaimByStorageClassWithNonIntegerStorage,
@@ -124,15 +132,52 @@ func TestPersistentVolumeClaimEvaluatorUsage(t *testing.T) {
 				V1ResourceByStorageClass(classGold, corev1.ResourcePersistentVolumeClaims):                        resource.MustParse("1"),
 				generic.ObjectCountQuotaResourceNameFor(schema.GroupResource{Resource: "persistentvolumeclaims"}): resource.MustParse("1"),
 			},
+			enableRecoverFromExpansion: true,
+		},
+		"pvc-usage-higher-allocated-resource": {
+			pvc: getPVCWithAllocatedResource("5G", "10G"),
+			usage: corev1.ResourceList{
+				corev1.ResourceRequestsStorage:        resource.MustParse("10G"),
+				corev1.ResourcePersistentVolumeClaims: resource.MustParse("1"),
+				generic.ObjectCountQuotaResourceNameFor(schema.GroupResource{Resource: "persistentvolumeclaims"}): resource.MustParse("1"),
+			},
+			enableRecoverFromExpansion: true,
+		},
+		"pvc-usage-lower-allocated-resource": {
+			pvc: getPVCWithAllocatedResource("10G", "5G"),
+			usage: corev1.ResourceList{
+				corev1.ResourceRequestsStorage:        resource.MustParse("10G"),
+				corev1.ResourcePersistentVolumeClaims: resource.MustParse("1"),
+				generic.ObjectCountQuotaResourceNameFor(schema.GroupResource{Resource: "persistentvolumeclaims"}): resource.MustParse("1"),
+			},
+			enableRecoverFromExpansion: true,
 		},
 	}
 	for testName, testCase := range testCases {
-		actual, err := evaluator.Usage(testCase.pvc)
+		t.Run(testName, func(t *testing.T) {
-		if err != nil {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, testCase.enableRecoverFromExpansion)()
-			t.Errorf("%s unexpected error: %v", testName, err)
+			actual, err := evaluator.Usage(testCase.pvc)
-		}
+			if err != nil {
-		if !quota.Equals(testCase.usage, actual) {
+				t.Errorf("%s unexpected error: %v", testName, err)
-			t.Errorf("%s expected:\n%v\n, actual:\n%v", testName, testCase.usage, actual)
+			}
-		}
+			if !quota.Equals(testCase.usage, actual) {
+				t.Errorf("%s expected:\n%v\n, actual:\n%v", testName, testCase.usage, actual)
+			}
+		})
+
 	}
 }
+
+func getPVCWithAllocatedResource(pvcSize, allocatedSize string) *api.PersistentVolumeClaim {
+	validPVCWithAllocatedResources := testVolumeClaim("foo", "ns", api.PersistentVolumeClaimSpec{
+		Resources: api.ResourceRequirements{
+			Requests: api.ResourceList{
+				core.ResourceStorage: resource.MustParse(pvcSize),
+			},
+		},
+	})
+	validPVCWithAllocatedResources.Status.AllocatedResources = api.ResourceList{
+		api.ResourceName(api.ResourceStorage): resource.MustParse(allocatedSize),
+	}
+	return validPVCWithAllocatedResources
+}

pkg/registry/core/persistentvolumeclaim/strategy.go

@@ -27,12 +27,10 @@ import (
 	"k8s.io/apiserver/pkg/registry/generic"
 	"k8s.io/apiserver/pkg/storage"
 	"k8s.io/apiserver/pkg/storage/names"
-	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	pvcutil "k8s.io/kubernetes/pkg/api/persistentvolumeclaim"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/validation"
-	"k8s.io/kubernetes/pkg/features"
 	"sigs.k8s.io/structured-merge-diff/v4/fieldpath"
 )
 
@@ -66,7 +64,6 @@ func (persistentvolumeclaimStrategy) GetResetFields() map[fieldpath.APIVersion]*
 func (persistentvolumeclaimStrategy) PrepareForCreate(ctx context.Context, obj runtime.Object) {
 	pvc := obj.(*api.PersistentVolumeClaim)
 	pvc.Status = api.PersistentVolumeClaimStatus{}
-
 	pvcutil.DropDisabledFields(&pvc.Spec)
 
 	// For data sources, we need to do 2 things to implement KEP 1495
@@ -153,16 +150,17 @@ func (persistentvolumeclaimStatusStrategy) GetResetFields() map[fieldpath.APIVer
 
 // PrepareForUpdate sets the Spec field which is not allowed to be changed when updating a PV's Status
 func (persistentvolumeclaimStatusStrategy) PrepareForUpdate(ctx context.Context, obj, old runtime.Object) {
-	newPv := obj.(*api.PersistentVolumeClaim)
+	newPVC := obj.(*api.PersistentVolumeClaim)
-	oldPv := old.(*api.PersistentVolumeClaim)
+	oldPVC := old.(*api.PersistentVolumeClaim)
-	newPv.Spec = oldPv.Spec
+	newPVC.Spec = oldPVC.Spec
-	if !utilfeature.DefaultFeatureGate.Enabled(features.ExpandPersistentVolumes) && oldPv.Status.Conditions == nil {
+	pvcutil.DropDisabledFieldsFromStatus(newPVC, oldPVC)
-		newPv.Status.Conditions = nil
-	}
 }
 
 func (persistentvolumeclaimStatusStrategy) ValidateUpdate(ctx context.Context, obj, old runtime.Object) field.ErrorList {
-	return validation.ValidatePersistentVolumeClaimStatusUpdate(obj.(*api.PersistentVolumeClaim), old.(*api.PersistentVolumeClaim))
+	newPvc := obj.(*api.PersistentVolumeClaim)
+	oldPvc := old.(*api.PersistentVolumeClaim)
+	opts := validation.ValidationOptionsForPersistentVolumeClaim(newPvc, oldPvc)
+	return validation.ValidatePersistentVolumeClaimStatusUpdate(newPvc, oldPvc, opts)
 }
 
 // WarningsOnUpdate returns warnings for the given update.

pkg/volume/csi/csi_client.go

@@ -347,8 +347,12 @@ func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, opts csiResizeOp
 
 	resp, err := nodeClient.NodeExpandVolume(ctx, req)
 	if err != nil {
+		if !isFinalError(err) {
+			return opts.newSize, volumetypes.NewUncertainProgressError(err.Error())
+		}
 		return opts.newSize, err
 	}
+
 	updatedQuantity := resource.NewQuantity(resp.CapacityBytes, resource.BinarySI)
 	return *updatedQuantity, nil
 }

pkg/volume/testing/testing.go

@@ -85,6 +85,8 @@ const (
 
 	FailVolumeExpansion = "fail-expansion-test"
 
+	AlwaysFailNodeExpansion = "always-fail-node-expansion"
+
 	deviceNotMounted     = "deviceNotMounted"
 	deviceMountUncertain = "deviceMountUncertain"
 	deviceMounted        = "deviceMounted"
@@ -178,6 +180,7 @@ type FakeVolumePlugin struct {
 	LimitKey               string
 	ProvisionDelaySeconds  int
 	SupportsRemount        bool
+	DisableNodeExpansion   bool
 
 	// default to false which means it is attachable by default
 	NonAttachable bool
@@ -464,13 +467,17 @@ func (plugin *FakeVolumePlugin) ExpandVolumeDevice(spec *Spec, newSize resource.
 }
 
 func (plugin *FakeVolumePlugin) RequiresFSResize() bool {
-	return true
+	return !plugin.DisableNodeExpansion
 }
 
 func (plugin *FakeVolumePlugin) NodeExpand(resizeOptions NodeResizeOptions) (bool, error) {
 	if resizeOptions.VolumeSpec.Name() == FailWithInUseVolumeName {
 		return false, volumetypes.NewFailedPreconditionError("volume-in-use")
 	}
+	if resizeOptions.VolumeSpec.Name() == AlwaysFailNodeExpansion {
+		return false, fmt.Errorf("Test failure: NodeExpand")
+	}
+
 	// Set up fakeVolumePlugin not support STAGE_UNSTAGE for testing the behavior
 	// so as volume can be node published before we can resize
 	if resizeOptions.CSIVolumePhase == volume.CSIVolumeStaged {

pkg/volume/util/operationexecutor/fakegenerator.go

@@ -104,6 +104,10 @@ func (f *fakeOGCounter) GenerateExpandVolumeFunc(*v1.PersistentVolumeClaim, *v1.
 	return f.recordFuncCall("GenerateExpandVolumeFunc"), nil
 }
 
+func (f *fakeOGCounter) GenerateExpandAndRecoverVolumeFunc(*v1.PersistentVolumeClaim, *v1.PersistentVolume, string) (volumetypes.GeneratedOperations, error) {
+	return f.recordFuncCall("GenerateExpandVolumeFunc"), nil
+}
+
 func (f *fakeOGCounter) GenerateExpandInUseVolumeFunc(volumeToMount VolumeToMount, actualStateOfWorld ActualStateOfWorldMounterUpdater) (volumetypes.GeneratedOperations, error) {
 	return f.recordFuncCall("GenerateExpandInUseVolumeFunc"), nil
 }

pkg/volume/util/operationexecutor/operation_executor_test.go

@@ -658,6 +658,16 @@ func (fopg *fakeOperationGenerator) GenerateExpandVolumeFunc(pvc *v1.PersistentV
 	}, nil
 }
 
+func (fopg *fakeOperationGenerator) GenerateExpandAndRecoverVolumeFunc(pvc *v1.PersistentVolumeClaim, pv *v1.PersistentVolume, resizerName string) (volumetypes.GeneratedOperations, error) {
+	opFunc := func() volumetypes.OperationContext {
+		startOperationAndBlock(fopg.ch, fopg.quit)
+		return volumetypes.NewOperationContext(nil, nil, false)
+	}
+	return volumetypes.GeneratedOperations{
+		OperationFunc: opFunc,
+	}, nil
+}
+
 func (fopg *fakeOperationGenerator) GenerateExpandInUseVolumeFunc(volumeToMount VolumeToMount, actualStateOfWorld ActualStateOfWorldMounterUpdater) (volumetypes.GeneratedOperations, error) {
 	opFunc := func() volumetypes.OperationContext {
 		startOperationAndBlock(fopg.ch, fopg.quit)

pkg/volume/util/operationexecutor/operation_generator.go

@@ -25,6 +25,8 @@ import (
 	"strings"
 	"time"
 
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -150,10 +152,50 @@ type OperationGenerator interface {
 
 	GenerateExpandVolumeFunc(*v1.PersistentVolumeClaim, *v1.PersistentVolume) (volumetypes.GeneratedOperations, error)
 
+	GenerateExpandAndRecoverVolumeFunc(*v1.PersistentVolumeClaim, *v1.PersistentVolume, string) (volumetypes.GeneratedOperations, error)
+
 	// Generates the volume file system resize function, which can resize volume's file system to expected size without unmounting the volume.
 	GenerateExpandInUseVolumeFunc(volumeToMount VolumeToMount, actualStateOfWorld ActualStateOfWorldMounterUpdater) (volumetypes.GeneratedOperations, error)
 }
 
+type inTreeResizeOpts struct {
+	resizerName  string
+	pvc          *v1.PersistentVolumeClaim
+	pv           *v1.PersistentVolume
+	volumeSpec   *volume.Spec
+	volumePlugin volume.ExpandableVolumePlugin
+}
+
+type inTreeResizeResponse struct {
+	pvc *v1.PersistentVolumeClaim
+	pv  *v1.PersistentVolume
+	err error
+
+	// Indicates whether kubelet should assume resize operation as finished.
+	// For kubelet - resize operation could be assumed as finished even if
+	// actual resizing is *not* finished. This can happen, because certain prechecks
+	// are failing and kubelet should not retry expansion, or it could happen
+	// because resize operation is genuinely finished.
+	assumeResizeOpAsFinished bool
+
+	// indicates that resize operation was called on underlying volume driver
+	// mainly useful for testing.
+	resizeCalled bool
+
+	// indicates whether entire volume expansion is finished or not
+	// only used from nodeExpansion calls. Mainly used for testing.
+	resizeFinished bool
+}
+
+type nodeResizeOperationOpts struct {
+	vmt                VolumeToMount
+	pvc                *v1.PersistentVolumeClaim
+	pv                 *v1.PersistentVolume
+	pluginResizeOpts   volume.NodeResizeOptions
+	volumePlugin       volume.NodeExpandableVolumePlugin
+	actualStateOfWorld ActualStateOfWorldMounterUpdater
+}
+
 func (og *operationGenerator) GenerateVolumesAreAttachedFunc(
 	attachedVolumes []AttachedVolume,
 	nodeName types.NodeName,
@@ -1595,7 +1637,6 @@ func (og *operationGenerator) GenerateExpandVolumeFunc(
 	}
 
 	expandVolumeFunc := func() volumetypes.OperationContext {
-
 		migrated := false
 
 		newSize := pvc.Spec.Resources.Requests[v1.ResourceStorage]
@@ -1617,7 +1658,7 @@ func (og *operationGenerator) GenerateExpandVolumeFunc(
 			// k8s doesn't have transactions, we can't guarantee that after updating PV - updating PVC will be
 			// successful, that is why all PVCs for which pvc.Spec.Size > pvc.Status.Size must be reprocessed
 			// until they reflect user requested size in pvc.Status.Size
-			updateErr := util.UpdatePVSize(pv, newSize, og.kubeClient)
+			_, updateErr := util.UpdatePVSize(pv, newSize, og.kubeClient)
 			if updateErr != nil {
 				detailedErr := fmt.Errorf("error updating PV spec capacity for volume %q with : %v", util.GetPersistentVolumeClaimQualifiedName(pvc), updateErr)
 				return volumetypes.NewOperationContext(detailedErr, detailedErr, migrated)
@@ -1632,7 +1673,7 @@ func (og *operationGenerator) GenerateExpandVolumeFunc(
 		// reflects user requested size.
 		if !volumePlugin.RequiresFSResize() || !fsVolume {
 			klog.V(4).Infof("Controller resizing done for PVC %s", util.GetPersistentVolumeClaimQualifiedName(pvc))
-			err := util.MarkResizeFinished(pvc, newSize, og.kubeClient)
+			_, err := util.MarkResizeFinished(pvc, newSize, og.kubeClient)
 			if err != nil {
 				detailedErr := fmt.Errorf("error marking pvc %s as resized : %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
 				return volumetypes.NewOperationContext(detailedErr, detailedErr, migrated)
@@ -1640,7 +1681,7 @@ func (og *operationGenerator) GenerateExpandVolumeFunc(
 			successMsg := fmt.Sprintf("ExpandVolume succeeded for volume %s", util.GetPersistentVolumeClaimQualifiedName(pvc))
 			og.recorder.Eventf(pvc, v1.EventTypeNormal, kevents.VolumeResizeSuccess, successMsg)
 		} else {
-			err := util.MarkForFSResize(pvc, og.kubeClient)
+			_, err := util.MarkForFSResize(pvc, og.kubeClient)
 			if err != nil {
 				detailedErr := fmt.Errorf("error updating pvc %s condition for fs resize : %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
 				klog.Warning(detailedErr)
@@ -1672,6 +1713,210 @@ func (og *operationGenerator) GenerateExpandVolumeFunc(
 	}, nil
 }
 
+func (og *operationGenerator) GenerateExpandAndRecoverVolumeFunc(
+	pvc *v1.PersistentVolumeClaim,
+	pv *v1.PersistentVolume, resizerName string) (volumetypes.GeneratedOperations, error) {
+
+	volumeSpec := volume.NewSpecFromPersistentVolume(pv, false)
+
+	volumePlugin, err := og.volumePluginMgr.FindExpandablePluginBySpec(volumeSpec)
+	if err != nil {
+		return volumetypes.GeneratedOperations{}, fmt.Errorf("error finding plugin for expanding volume: %q with error %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+	}
+
+	if volumePlugin == nil {
+		return volumetypes.GeneratedOperations{}, fmt.Errorf("can not find plugin for expanding volume: %q", util.GetPersistentVolumeClaimQualifiedName(pvc))
+	}
+
+	expandVolumeFunc := func() volumetypes.OperationContext {
+		resizeOpts := inTreeResizeOpts{
+			pvc:          pvc,
+			pv:           pv,
+			resizerName:  resizerName,
+			volumePlugin: volumePlugin,
+			volumeSpec:   volumeSpec,
+		}
+		migrated := false
+		resp := og.expandAndRecoverFunction(resizeOpts)
+		if resp.err != nil {
+			return volumetypes.NewOperationContext(resp.err, resp.err, migrated)
+		}
+		return volumetypes.NewOperationContext(nil, nil, migrated)
+	}
+
+	eventRecorderFunc := func(err *error) {
+		if *err != nil {
+			og.recorder.Eventf(pvc, v1.EventTypeWarning, kevents.VolumeResizeFailed, (*err).Error())
+		}
+	}
+
+	return volumetypes.GeneratedOperations{
+		OperationName:     "expand_volume",
+		OperationFunc:     expandVolumeFunc,
+		EventRecorderFunc: eventRecorderFunc,
+		CompleteFunc:      util.OperationCompleteHook(util.GetFullQualifiedPluginNameForVolume(volumePlugin.GetPluginName(), volumeSpec), "expand_volume"),
+	}, nil
+}
+
+func (og *operationGenerator) expandAndRecoverFunction(resizeOpts inTreeResizeOpts) inTreeResizeResponse {
+	pvc := resizeOpts.pvc
+	pv := resizeOpts.pv
+	resizerName := resizeOpts.resizerName
+	volumePlugin := resizeOpts.volumePlugin
+	volumeSpec := resizeOpts.volumeSpec
+
+	pvcSpecSize := pvc.Spec.Resources.Requests[v1.ResourceStorage]
+	pvcStatusSize := pvc.Status.Capacity[v1.ResourceStorage]
+	pvSize := pv.Spec.Capacity[v1.ResourceStorage]
+
+	resizeResponse := inTreeResizeResponse{
+		pvc:          pvc,
+		pv:           pv,
+		resizeCalled: false,
+	}
+
+	// by default we are expanding to full-fill size requested in pvc.Spec.Resources
+	newSize := pvcSpecSize
+	resizeStatus := v1.PersistentVolumeClaimNoExpansionInProgress
+	if pvc.Status.ResizeStatus != nil {
+		resizeStatus = *pvc.Status.ResizeStatus
+	}
+	var allocatedSize *resource.Quantity
+	t, ok := pvc.Status.AllocatedResources[v1.ResourceStorage]
+	if ok {
+		allocatedSize = &t
+	}
+	var err error
+
+	if pvSize.Cmp(pvcSpecSize) < 0 {
+		// pv is not of requested size yet and hence will require expanding
+
+		switch resizeStatus {
+		case v1.PersistentVolumeClaimControllerExpansionInProgress:
+		case v1.PersistentVolumeClaimNodeExpansionPending:
+		case v1.PersistentVolumeClaimNodeExpansionInProgress:
+		case v1.PersistentVolumeClaimNodeExpansionFailed:
+			if allocatedSize != nil {
+				newSize = *allocatedSize
+			}
+		default:
+			newSize = pvcSpecSize
+		}
+	} else {
+		// PV has already been expanded and hence we can be here for following reasons:
+		//   1. If expansion is pending on the node and this was just a spurious update event
+		//      we don't need to do anything and let kubelet handle it.
+		//   2. It could be that - although we successfully expanded the volume, we failed to
+		//      record our work in API objects, in which case - we should resume resizing operation
+		//      and let API objects be updated.
+		//   3. Controller successfully expanded the volume, but expansion is failing on the node
+		//      and before kubelet can retry failed node expansion - controller must verify if it is
+		//      safe to do so.
+		//   4. While expansion was still pending on the node, user reduced the pvc size.
+		switch resizeStatus {
+		case v1.PersistentVolumeClaimNodeExpansionInProgress:
+		case v1.PersistentVolumeClaimNodeExpansionPending:
+			// we don't need to do any work. We could be here because of a spurious update event.
+			// This is case #1
+			return resizeResponse
+		case v1.PersistentVolumeClaimNodeExpansionFailed:
+			// This is case#3
+			pvc, err = og.markForPendingNodeExpansion(pvc, pv)
+			resizeResponse.pvc = pvc
+			resizeResponse.err = err
+			return resizeResponse
+		case v1.PersistentVolumeClaimControllerExpansionInProgress:
+		case v1.PersistentVolumeClaimControllerExpansionFailed:
+		case v1.PersistentVolumeClaimNoExpansionInProgress:
+			// This is case#2 or it could also be case#4 when user manually shrunk the PVC
+			// after expanding it.
+			if allocatedSize != nil {
+				newSize = *allocatedSize
+			}
+		default:
+			// It is impossible for ResizeStatus to be nil and allocatedSize to be not nil but somehow
+			// if we do end up in this state, it is safest to resume expansion to last recorded size in
+			// allocatedSize variable.
+			if pvc.Status.ResizeStatus == nil && allocatedSize != nil {
+				newSize = *allocatedSize
+			} else {
+				newSize = pvcSpecSize
+			}
+		}
+	}
+
+	pvc, err = util.MarkControllerReisizeInProgress(pvc, resizerName, newSize, og.kubeClient)
+	if err != nil {
+		msg := fmt.Errorf("error updating pvc %s with resize in progress: %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+		resizeResponse.err = msg
+		resizeResponse.pvc = pvc
+		return resizeResponse
+	}
+
+	updatedSize, err := volumePlugin.ExpandVolumeDevice(volumeSpec, newSize, pvcStatusSize)
+	resizeResponse.resizeCalled = true
+
+	if err != nil {
+		msg := fmt.Errorf("error expanding pvc %s: %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+		resizeResponse.err = msg
+		resizeResponse.pvc = pvc
+		return resizeResponse
+	}
+
+	// update PV size
+	var updateErr error
+	pv, updateErr = util.UpdatePVSize(pv, updatedSize, og.kubeClient)
+	// if updating PV failed, we are going to leave the PVC in ControllerExpansionInProgress state, so as expansion can be retried to previously set allocatedSize value.
+	if updateErr != nil {
+		msg := fmt.Errorf("error updating pv for pvc %s: %v", util.GetPersistentVolumeClaimQualifiedName(pvc), updateErr)
+		resizeResponse.err = msg
+		return resizeResponse
+	}
+	resizeResponse.pv = pv
+
+	fsVolume, _ := util.CheckVolumeModeFilesystem(volumeSpec)
+
+	if !volumePlugin.RequiresFSResize() || !fsVolume {
+		pvc, err = util.MarkResizeFinished(pvc, updatedSize, og.kubeClient)
+		if err != nil {
+			msg := fmt.Errorf("error marking pvc %s as resized: %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+			resizeResponse.err = msg
+			return resizeResponse
+		}
+		resizeResponse.pvc = pvc
+		successMsg := fmt.Sprintf("ExpandVolume succeeded for volume %s", util.GetPersistentVolumeClaimQualifiedName(pvc))
+		og.recorder.Eventf(pvc, v1.EventTypeNormal, kevents.VolumeResizeSuccess, successMsg)
+	} else {
+		pvc, err = og.markForPendingNodeExpansion(pvc, pv)
+		resizeResponse.pvc = pvc
+		if err != nil {
+			msg := fmt.Errorf("error marking pvc %s for node expansion: %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+			resizeResponse.err = msg
+			return resizeResponse
+		}
+	}
+	return resizeResponse
+}
+
+func (og *operationGenerator) markForPendingNodeExpansion(pvc *v1.PersistentVolumeClaim, pv *v1.PersistentVolume) (*v1.PersistentVolumeClaim, error) {
+	var err error
+	pvc, err = util.MarkForFSResize(pvc, og.kubeClient)
+	if err != nil {
+		msg := fmt.Errorf("error marking pvc %s for node expansion: %v", util.GetPersistentVolumeClaimQualifiedName(pvc), err)
+		return pvc, msg
+	}
+	// store old PVC capacity in pv, so as if PVC gets deleted while node expansion was pending
+	// we can restore size of pvc from PV annotation and still perform expansion on the node
+	oldCapacity := pvc.Status.Capacity[v1.ResourceStorage]
+	err = util.AddAnnPreResizeCapacity(pv, oldCapacity, og.kubeClient)
+	if err != nil {
+		detailedErr := fmt.Errorf("error updating pv %s annotation (%s) with pre-resize capacity %s: %v", pv.ObjectMeta.Name, util.AnnPreResizeCapacity, oldCapacity.String(), err)
+		klog.Warning(detailedErr)
+		return pvc, detailedErr
+	}
+	return pvc, nil
+}
+
 func (og *operationGenerator) GenerateExpandInUseVolumeFunc(
 	volumeToMount VolumeToMount,
 	actualStateOfWorld ActualStateOfWorldMounterUpdater) (volumetypes.GeneratedOperations, error) {
@@ -1825,6 +2070,7 @@ func (og *operationGenerator) nodeExpandVolume(
 	if expandableVolumePlugin != nil &&
 		expandableVolumePlugin.RequiresFSResize() &&
 		volumeToMount.VolumeSpec.PersistentVolume != nil {
+
 		pv := volumeToMount.VolumeSpec.PersistentVolume
 		pvc, err := og.kubeClient.CoreV1().PersistentVolumeClaims(pv.Spec.ClaimRef.Namespace).Get(context.TODO(), pv.Spec.ClaimRef.Name, metav1.GetOptions{})
 		if err != nil {
@@ -1832,56 +2078,200 @@ func (og *operationGenerator) nodeExpandVolume(
 			return false, fmt.Errorf("mountVolume.NodeExpandVolume get PVC failed : %v", err)
 		}
 
-		pvcStatusCap := pvc.Status.Capacity[v1.ResourceStorage]
+		if volumeToMount.VolumeSpec.ReadOnly {
-		pvSpecCap := pv.Spec.Capacity[v1.ResourceStorage]
+			simpleMsg, detailedMsg := volumeToMount.GenerateMsg("MountVolume.NodeExpandVolume failed", "requested read-only file system")
-		if pvcStatusCap.Cmp(pvSpecCap) < 0 {
+			klog.Warningf(detailedMsg)
-			// File system resize was requested, proceed
+			og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeWarning, kevents.FileSystemResizeFailed, simpleMsg)
-			klog.V(4).InfoS(volumeToMount.GenerateMsgDetailed("MountVolume.NodeExpandVolume entering", fmt.Sprintf("DevicePath %q", volumeToMount.DevicePath)), "pod", klog.KObj(volumeToMount.Pod))
+			og.recorder.Eventf(pvc, v1.EventTypeWarning, kevents.FileSystemResizeFailed, simpleMsg)
-
-			if volumeToMount.VolumeSpec.ReadOnly {
-				simpleMsg, detailedMsg := volumeToMount.GenerateMsg("MountVolume.NodeExpandVolume failed", "requested read-only file system")
-				klog.Warningf(detailedMsg)
-				og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeWarning, kevents.FileSystemResizeFailed, simpleMsg)
-				og.recorder.Eventf(pvc, v1.EventTypeWarning, kevents.FileSystemResizeFailed, simpleMsg)
-				return true, nil
-			}
-			rsOpts.VolumeSpec = volumeToMount.VolumeSpec
-			rsOpts.NewSize = pvSpecCap
-			rsOpts.OldSize = pvcStatusCap
-			resizeDone, resizeErr := expandableVolumePlugin.NodeExpand(rsOpts)
-			if resizeErr != nil {
-				// if driver returned FailedPrecondition error that means
-				// volume expansion should not be retried on this node but
-				// expansion operation should not block mounting
-				if volumetypes.IsFailedPreconditionError(resizeErr) {
-					actualStateOfWorld.MarkForInUseExpansionError(volumeToMount.VolumeName)
-					klog.Errorf(volumeToMount.GenerateErrorDetailed("MountVolume.NodeExapndVolume failed with %v", resizeErr).Error())
-					return true, nil
-				}
-				return false, resizeErr
-			}
-			// Volume resizing is not done but it did not error out. This could happen if a CSI volume
-			// does not have node stage_unstage capability but was asked to resize the volume before
-			// node publish. In which case - we must retry resizing after node publish.
-			if !resizeDone {
-				return false, nil
-			}
-			simpleMsg, detailedMsg := volumeToMount.GenerateMsg("MountVolume.NodeExpandVolume succeeded", "")
-			og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeNormal, kevents.FileSystemResizeSuccess, simpleMsg)
-			og.recorder.Eventf(pvc, v1.EventTypeNormal, kevents.FileSystemResizeSuccess, simpleMsg)
-			klog.InfoS(detailedMsg, "pod", klog.KObj(volumeToMount.Pod))
-			// File system resize succeeded, now update the PVC's Capacity to match the PV's
-			err = util.MarkFSResizeFinished(pvc, pvSpecCap, og.kubeClient)
-			if err != nil {
-				// On retry, NodeExpandVolume will be called again but do nothing
-				return false, fmt.Errorf("mountVolume.NodeExpandVolume update PVC status failed : %v", err)
-			}
 			return true, nil
 		}
+		resizeOp := nodeResizeOperationOpts{
+			vmt:                volumeToMount,
+			pvc:                pvc,
+			pv:                 pv,
+			pluginResizeOpts:   rsOpts,
+			volumePlugin:       expandableVolumePlugin,
+			actualStateOfWorld: actualStateOfWorld,
+		}
+		if utilfeature.DefaultFeatureGate.Enabled(features.RecoverVolumeExpansionFailure) {
+			resizeResponse := og.callNodeExpandOnPlugin(resizeOp)
+			return resizeResponse.assumeResizeOpAsFinished, resizeResponse.err
+		} else {
+			return og.legacyCallNodeExpandOnPlugin(resizeOp)
+		}
 	}
 	return true, nil
 }
 
+// callNodeExpandOnPlugin is newer version of calling node expansion on plugins, which does support
+// recovery from volume expansion failure.
+func (og *operationGenerator) callNodeExpandOnPlugin(resizeOp nodeResizeOperationOpts) inTreeResizeResponse {
+	pvc := resizeOp.pvc
+	pv := resizeOp.pv
+	volumeToMount := resizeOp.vmt
+	rsOpts := resizeOp.pluginResizeOpts
+	actualStateOfWorld := resizeOp.actualStateOfWorld
+	expandableVolumePlugin := resizeOp.volumePlugin
+
+	var err error
+	pvcStatusCap := pvc.Status.Capacity[v1.ResourceStorage]
+	pvSpecCap := pv.Spec.Capacity[v1.ResourceStorage]
+
+	resizeResponse := inTreeResizeResponse{
+		pvc: pvc,
+		pv:  pv,
+	}
+
+	if permitNodeExpansion(pvc, pv) {
+		// File system resize was requested, proceed
+		klog.V(4).InfoS(volumeToMount.GenerateMsgDetailed("MountVolume.NodeExpandVolume entering", fmt.Sprintf("DevicePath %q", volumeToMount.DevicePath)), "pod", klog.KObj(volumeToMount.Pod))
+
+		rsOpts.VolumeSpec = volumeToMount.VolumeSpec
+		rsOpts.NewSize = pvSpecCap
+		rsOpts.OldSize = pvcStatusCap
+		pvc, err = util.MarkNodeExpansionInProgress(pvc, og.kubeClient)
+
+		if err != nil {
+			msg := volumeToMount.GenerateErrorDetailed("MountVolume.NodeExpandVolume failed to mark node expansion in progress: %v", err)
+			klog.Errorf(msg.Error())
+			resizeResponse.err = msg
+			return resizeResponse
+		}
+
+		resizeDone, resizeErr := expandableVolumePlugin.NodeExpand(rsOpts)
+		resizeResponse.resizeCalled = true
+
+		if resizeErr != nil {
+			if volumetypes.IsOperationFinishedError(resizeErr) {
+				var markFailedError error
+				pvc, markFailedError = util.MarkNodeExpansionFailed(pvc, og.kubeClient)
+				// update the pvc with node expansion object
+				resizeResponse.pvc = pvc
+				resizeResponse.assumeResizeOpAsFinished = true
+				if markFailedError != nil {
+					klog.Errorf(volumeToMount.GenerateErrorDetailed("MountMount.NodeExpandVolume failed to mark node expansion as failed: %v", err).Error())
+				}
+			}
+
+			// if driver returned FailedPrecondition error that means
+			// volume expansion should not be retried on this node but
+			// expansion operation should not block mounting
+			if volumetypes.IsFailedPreconditionError(resizeErr) {
+				actualStateOfWorld.MarkForInUseExpansionError(volumeToMount.VolumeName)
+				klog.Errorf(volumeToMount.GenerateErrorDetailed("MountVolume.NodeExapndVolume failed with %v", resizeErr).Error())
+				resizeResponse.assumeResizeOpAsFinished = true
+				return resizeResponse
+			}
+
+			resizeResponse.err = resizeErr
+			return resizeResponse
+		}
+		resizeResponse.resizeFinished = resizeDone
+
+		// Volume resizing is not done but it did not error out. This could happen if a CSI volume
+		// does not have node stage_unstage capability but was asked to resize the volume before
+		// node publish. In which case - we must retry resizing after node publish.
+		if !resizeDone {
+			return resizeResponse
+		}
+
+		simpleMsg, detailedMsg := volumeToMount.GenerateMsg("MountVolume.NodeExpandVolume succeeded", "")
+		og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeNormal, kevents.FileSystemResizeSuccess, simpleMsg)
+		og.recorder.Eventf(pvc, v1.EventTypeNormal, kevents.FileSystemResizeSuccess, simpleMsg)
+		klog.InfoS(detailedMsg, "pod", klog.KObj(volumeToMount.Pod))
+
+		// File system resize succeeded, now update the PVC's Capacity to match the PV's
+		pvc, err = util.MarkFSResizeFinished(pvc, pvSpecCap, og.kubeClient)
+		resizeResponse.pvc = pvc
+
+		if err != nil {
+			resizeResponse.err = fmt.Errorf("mountVolume.NodeExpandVolume update PVC status failed : %v", err)
+			// On retry, NodeExpandVolume will be called again but do nothing
+			return resizeResponse
+		}
+		resizeResponse.assumeResizeOpAsFinished = true
+		return resizeResponse
+	}
+	// somehow a resize operation was queued, but we can not perform any resizing because
+	// prechecks required for node expansion failed. Kubelet should not retry expanding the volume.
+	resizeResponse.assumeResizeOpAsFinished = true
+	return resizeResponse
+}
+
+// legacyCallNodeExpandOnPlugin is old version of calling node expansion on plugin, which does not support
+// recovery from volume expansion failure
+func (og *operationGenerator) legacyCallNodeExpandOnPlugin(resizeOp nodeResizeOperationOpts) (bool, error) {
+	pvc := resizeOp.pvc
+	pv := resizeOp.pv
+	volumeToMount := resizeOp.vmt
+	rsOpts := resizeOp.pluginResizeOpts
+	actualStateOfWorld := resizeOp.actualStateOfWorld
+	expandableVolumePlugin := resizeOp.volumePlugin
+
+	var err error
+
+	pvcStatusCap := pvc.Status.Capacity[v1.ResourceStorage]
+	pvSpecCap := pv.Spec.Capacity[v1.ResourceStorage]
+	if pvcStatusCap.Cmp(pvSpecCap) < 0 {
+		// File system resize was requested, proceed
+		klog.V(4).InfoS(volumeToMount.GenerateMsgDetailed("MountVolume.NodeExpandVolume entering", fmt.Sprintf("DevicePath %q", volumeToMount.DevicePath)), "pod", klog.KObj(volumeToMount.Pod))
+
+		rsOpts.VolumeSpec = volumeToMount.VolumeSpec
+		rsOpts.NewSize = pvSpecCap
+		rsOpts.OldSize = pvcStatusCap
+		resizeDone, resizeErr := expandableVolumePlugin.NodeExpand(rsOpts)
+		if resizeErr != nil {
+			// if driver returned FailedPrecondition error that means
+			// volume expansion should not be retried on this node but
+			// expansion operation should not block mounting
+			if volumetypes.IsFailedPreconditionError(resizeErr) {
+				actualStateOfWorld.MarkForInUseExpansionError(volumeToMount.VolumeName)
+				klog.Errorf(volumeToMount.GenerateErrorDetailed("MountVolume.NodeExapndVolume failed with %v", resizeErr).Error())
+				return true, nil
+			}
+			return false, resizeErr
+		}
+		// Volume resizing is not done but it did not error out. This could happen if a CSI volume
+		// does not have node stage_unstage capability but was asked to resize the volume before
+		// node publish. In which case - we must retry resizing after node publish.
+		if !resizeDone {
+			return false, nil
+		}
+		simpleMsg, detailedMsg := volumeToMount.GenerateMsg("MountVolume.NodeExpandVolume succeeded", "")
+		og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeNormal, kevents.FileSystemResizeSuccess, simpleMsg)
+		og.recorder.Eventf(pvc, v1.EventTypeNormal, kevents.FileSystemResizeSuccess, simpleMsg)
+		klog.InfoS(detailedMsg, "pod", klog.KObj(volumeToMount.Pod))
+		// File system resize succeeded, now update the PVC's Capacity to match the PV's
+		_, err = util.MarkFSResizeFinished(pvc, pvSpecCap, og.kubeClient)
+		if err != nil {
+			// On retry, NodeExpandVolume will be called again but do nothing
+			return false, fmt.Errorf("mountVolume.NodeExpandVolume update PVC status failed : %v", err)
+		}
+		return true, nil
+	}
+	return true, nil
+}
+
+func permitNodeExpansion(pvc *v1.PersistentVolumeClaim, pv *v1.PersistentVolume) bool {
+	pvcStatusCap := pvc.Status.Capacity[v1.ResourceStorage]
+	pvSpecCap := pv.Spec.Capacity[v1.ResourceStorage]
+	// if pvc.Status.Cap is >= pv.Spec.Cap then volume is already expanded
+	if pvcStatusCap.Cmp(pvSpecCap) >= 0 {
+		return false
+	}
+
+	resizeStatus := pvc.Status.ResizeStatus
+	// if resizestatus is nil or NodeExpansionInProgress or NodeExpansionPending then we should allow volume expansion on
+	// the node to proceed. We are making an exception for resizeStatus being nil because it will support use cases where
+	// resizeStatus may not be set (old control-plane expansion controller etc).
+	if resizeStatus == nil || *resizeStatus == v1.PersistentVolumeClaimNodeExpansionPending || *resizeStatus == v1.PersistentVolumeClaimNodeExpansionInProgress {
+		return true
+	} else {
+		klog.Infof("volume %s/%s can not be expanded because resizeStaus is: %s", pvc.Namespace, pvc.Name, *resizeStatus)
+		return false
+	}
+}
+
 func checkMountOptionSupport(og *operationGenerator, volumeToMount VolumeToMount, plugin volume.VolumePlugin) error {
 	mountOptions := util.MountOptionFromSpec(volumeToMount.VolumeSpec)
 

pkg/volume/util/operationexecutor/operation_generator_test.go

@@ -17,6 +17,12 @@ limitations under the License.
 package operationexecutor
 
 import (
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/client-go/tools/record"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/kubernetes/pkg/features"
 	"os"
 	"testing"
 
@@ -27,7 +33,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	fakeclient "k8s.io/client-go/kubernetes/fake"
-	"k8s.io/client-go/tools/record"
 	"k8s.io/component-base/metrics/legacyregistry"
 	"k8s.io/csi-translation-lib/plugins"
 	"k8s.io/kubernetes/pkg/volume"
@@ -111,6 +116,267 @@ func TestOperationGenerator_GenerateUnmapVolumeFunc_PluginName(t *testing.T) {
 	}
 }
 
+func TestOperationGenerator_GenerateExpandAndRecoverVolumeFunc(t *testing.T) {
+	var tests = []struct {
+		name                 string
+		pvc                  *v1.PersistentVolumeClaim
+		pv                   *v1.PersistentVolume
+		recoverFeatureGate   bool
+		disableNodeExpansion bool
+		// expectations of test
+		expectedResizeStatus  v1.PersistentVolumeClaimResizeStatus
+		expectedAllocatedSize resource.Quantity
+		expectResizeCall      bool
+	}{
+		{
+			name:                  "pvc.spec.size > pv.spec.size, recover_expansion=on",
+			pvc:                   getTestPVC("test-vol0", "2G", "1G", "", v1.PersistentVolumeClaimNoExpansionInProgress),
+			pv:                    getTestPV("test-vol0", "1G"),
+			recoverFeatureGate:    true,
+			expectedResizeStatus:  v1.PersistentVolumeClaimNodeExpansionPending,
+			expectedAllocatedSize: resource.MustParse("2G"),
+			expectResizeCall:      true,
+		},
+		{
+			name:                  "pvc.spec.size = pv.spec.size, recover_expansion=on",
+			pvc:                   getTestPVC("test-vol0", "1G", "1G", "", v1.PersistentVolumeClaimNoExpansionInProgress),
+			pv:                    getTestPV("test-vol0", "1G"),
+			recoverFeatureGate:    true,
+			expectedResizeStatus:  v1.PersistentVolumeClaimNodeExpansionPending,
+			expectedAllocatedSize: resource.MustParse("1G"),
+			expectResizeCall:      true,
+		},
+		{
+			name:                  "pvc.spec.size = pv.spec.size, recover_expansion=on",
+			pvc:                   getTestPVC("test-vol0", "1G", "1G", "1G", v1.PersistentVolumeClaimNodeExpansionPending),
+			pv:                    getTestPV("test-vol0", "1G"),
+			recoverFeatureGate:    true,
+			expectedResizeStatus:  v1.PersistentVolumeClaimNodeExpansionPending,
+			expectedAllocatedSize: resource.MustParse("1G"),
+			expectResizeCall:      false,
+		},
+		{
+			name:                  "pvc.spec.size > pv.spec.size, recover_expansion=on, disable_node_expansion=true",
+			pvc:                   getTestPVC("test-vol0", "2G", "1G", "", v1.PersistentVolumeClaimNoExpansionInProgress),
+			pv:                    getTestPV("test-vol0", "1G"),
+			disableNodeExpansion:  true,
+			recoverFeatureGate:    true,
+			expectedResizeStatus:  v1.PersistentVolumeClaimNoExpansionInProgress,
+			expectedAllocatedSize: resource.MustParse("2G"),
+			expectResizeCall:      true,
+		},
+		{
+			name:                  "pv.spec.size >= pvc.spec.size, recover_expansion=on, resize_status=node_expansion_failed",
+			pvc:                   getTestPVC("test-vol0", "2G", "1G", "2G", v1.PersistentVolumeClaimNodeExpansionFailed),
+			pv:                    getTestPV("test-vol0", "2G"),
+			recoverFeatureGate:    true,
+			expectedResizeStatus:  v1.PersistentVolumeClaimNodeExpansionPending,
+			expectedAllocatedSize: resource.MustParse("2G"),
+			expectResizeCall:      false,
+		},
+	}
+	for i := range tests {
+		test := tests[i]
+		t.Run(test.name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, test.recoverFeatureGate)()
+			volumePluginMgr, fakePlugin := volumetesting.GetTestKubeletVolumePluginMgr(t)
+			fakePlugin.DisableNodeExpansion = test.disableNodeExpansion
+			pvc := test.pvc
+			pv := test.pv
+			og := getTestOperationGenerator(volumePluginMgr, pvc, pv)
+			rsOpts := inTreeResizeOpts{
+				pvc:          pvc,
+				pv:           pv,
+				resizerName:  fakePlugin.GetPluginName(),
+				volumePlugin: fakePlugin,
+			}
+			ogInstance, _ := og.(*operationGenerator)
+
+			expansionResponse := ogInstance.expandAndRecoverFunction(rsOpts)
+			if expansionResponse.err != nil {
+				t.Fatalf("GenerateExpandAndRecoverVolumeFunc failed: %v", expansionResponse.err)
+			}
+			updatedPVC := expansionResponse.pvc
+			assert.Equal(t, *updatedPVC.Status.ResizeStatus, test.expectedResizeStatus)
+			actualAllocatedSize := updatedPVC.Status.AllocatedResources.Storage()
+			if test.expectedAllocatedSize.Cmp(*actualAllocatedSize) != 0 {
+				t.Fatalf("GenerateExpandAndRecoverVolumeFunc failed: expected allocated size %s, got %s", test.expectedAllocatedSize.String(), actualAllocatedSize.String())
+			}
+			if test.expectResizeCall != expansionResponse.resizeCalled {
+				t.Fatalf("GenerateExpandAndRecoverVolumeFunc failed: expected resize called %t, got %t", test.expectResizeCall, expansionResponse.resizeCalled)
+			}
+		})
+	}
+}
+
+func TestOperationGenerator_callNodeExpansionOnPlugin(t *testing.T) {
+	var tests = []struct {
+		name               string
+		pvc                *v1.PersistentVolumeClaim
+		pv                 *v1.PersistentVolume
+		recoverFeatureGate bool
+
+		// expectations of test
+		expectedResizeStatus     v1.PersistentVolumeClaimResizeStatus
+		expectedStatusSize       resource.Quantity
+		expectResizeCall         bool
+		assumeResizeOpAsFinished bool
+		expectError              bool
+	}{
+		{
+			name:               "pv.spec.cap > pvc.status.cap, resizeStatus=node_expansion_failed",
+			pvc:                getTestPVC("test-vol0", "2G", "1G", "", v1.PersistentVolumeClaimNodeExpansionFailed),
+			pv:                 getTestPV("test-vol0", "2G"),
+			recoverFeatureGate: true,
+
+			expectedResizeStatus:     v1.PersistentVolumeClaimNodeExpansionFailed,
+			expectResizeCall:         false,
+			assumeResizeOpAsFinished: true,
+			expectedStatusSize:       resource.MustParse("1G"),
+		},
+		{
+			name:                     "pv.spec.cap > pvc.status.cap, resizeStatus=node_expansion_pending",
+			pvc:                      getTestPVC("test-vol0", "2G", "1G", "2G", v1.PersistentVolumeClaimNodeExpansionPending),
+			pv:                       getTestPV("test-vol0", "2G"),
+			recoverFeatureGate:       true,
+			expectedResizeStatus:     v1.PersistentVolumeClaimNoExpansionInProgress,
+			expectResizeCall:         true,
+			assumeResizeOpAsFinished: true,
+			expectedStatusSize:       resource.MustParse("2G"),
+		},
+		{
+			name:                     "pv.spec.cap > pvc.status.cap, resizeStatus=node_expansion_pending, reize_op=failing",
+			pvc:                      getTestPVC(volumetesting.AlwaysFailNodeExpansion, "2G", "1G", "2G", v1.PersistentVolumeClaimNodeExpansionPending),
+			pv:                       getTestPV(volumetesting.AlwaysFailNodeExpansion, "2G"),
+			recoverFeatureGate:       true,
+			expectError:              true,
+			expectedResizeStatus:     v1.PersistentVolumeClaimNodeExpansionFailed,
+			expectResizeCall:         true,
+			assumeResizeOpAsFinished: true,
+			expectedStatusSize:       resource.MustParse("1G"),
+		},
+	}
+
+	for i := range tests {
+		test := tests[i]
+		t.Run(test.name, func(t *testing.T) {
+			defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.RecoverVolumeExpansionFailure, test.recoverFeatureGate)()
+			volumePluginMgr, fakePlugin := volumetesting.GetTestKubeletVolumePluginMgr(t)
+
+			pvc := test.pvc
+			pv := test.pv
+			pod := getTestPod("test-pod", pvc.Name)
+			og := getTestOperationGenerator(volumePluginMgr, pvc, pv)
+
+			vmt := VolumeToMount{
+				Pod:        pod,
+				VolumeName: v1.UniqueVolumeName(pv.Name),
+				VolumeSpec: volume.NewSpecFromPersistentVolume(pv, false),
+			}
+			resizeOp := nodeResizeOperationOpts{
+				pvc:                pvc,
+				pv:                 pv,
+				volumePlugin:       fakePlugin,
+				vmt:                vmt,
+				actualStateOfWorld: nil,
+			}
+			ogInstance, _ := og.(*operationGenerator)
+			expansionResponse := ogInstance.callNodeExpandOnPlugin(resizeOp)
+
+			pvc = expansionResponse.pvc
+			pvcStatusCap := pvc.Status.Capacity[v1.ResourceStorage]
+
+			if !test.expectError && expansionResponse.err != nil {
+				t.Errorf("For test %s, expected no error got: %v", test.name, expansionResponse.err)
+			}
+			if test.expectError && expansionResponse.err == nil {
+				t.Errorf("For test %s, expected error but got none", test.name)
+			}
+
+			if test.expectResizeCall != expansionResponse.resizeCalled {
+				t.Errorf("For test %s, expected resize called %t, got %t", test.name, test.expectResizeCall, expansionResponse.resizeCalled)
+			}
+			if test.assumeResizeOpAsFinished != expansionResponse.assumeResizeOpAsFinished {
+				t.Errorf("For test %s, expected assumeResizeOpAsFinished %t, got %t", test.name, test.assumeResizeOpAsFinished, expansionResponse.assumeResizeOpAsFinished)
+			}
+			if test.expectedResizeStatus != *pvc.Status.ResizeStatus {
+				t.Errorf("For test %s, expected resizeStatus %v, got %v", test.name, test.expectedResizeStatus, *pvc.Status.ResizeStatus)
+			}
+			if pvcStatusCap.Cmp(test.expectedStatusSize) != 0 {
+				t.Errorf("For test %s, expected status size %s, got %s", test.name, test.expectedStatusSize.String(), pvcStatusCap.String())
+			}
+		})
+	}
+}
+
+func getTestPod(podName, pvcName string) *v1.Pod {
+	return &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      podName,
+			UID:       "test-pod-uid",
+			Namespace: "ns",
+		},
+		Spec: v1.PodSpec{
+			Volumes: []v1.Volume{
+				{
+					Name: pvcName,
+					VolumeSource: v1.VolumeSource{
+						PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+							ClaimName: pvcName,
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func getTestPVC(volumeName string, specSize, statusSize, allocatedSize string, resizeStatus v1.PersistentVolumeClaimResizeStatus) *v1.PersistentVolumeClaim {
+	pvc := &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "claim01",
+			Namespace: "ns",
+			UID:       "test-uid",
+		},
+		Spec: v1.PersistentVolumeClaimSpec{
+			AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+			Resources:   v1.ResourceRequirements{Requests: v1.ResourceList{v1.ResourceStorage: resource.MustParse(specSize)}},
+			VolumeName:  volumeName,
+		},
+		Status: v1.PersistentVolumeClaimStatus{
+			Phase: v1.ClaimBound,
+		},
+	}
+	if len(statusSize) > 0 {
+		pvc.Status.Capacity = v1.ResourceList{v1.ResourceStorage: resource.MustParse(statusSize)}
+	}
+	if len(allocatedSize) > 0 {
+		pvc.Status.AllocatedResources = v1.ResourceList{v1.ResourceStorage: resource.MustParse(allocatedSize)}
+	}
+	if len(resizeStatus) > 0 {
+		pvc.Status.ResizeStatus = &resizeStatus
+	}
+	return pvc
+}
+
+func getTestPV(volumeName string, specSize string) *v1.PersistentVolume {
+	return &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: volumeName,
+			UID:  "test-uid",
+		},
+		Spec: v1.PersistentVolumeSpec{
+			AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+			Capacity: v1.ResourceList{
+				v1.ResourceStorage: resource.MustParse(specSize),
+			},
+		},
+		Status: v1.PersistentVolumeStatus{
+			Phase: v1.VolumeBound,
+		},
+	}
+}
+
 func findMetricWithNameAndLabels(metricFamilyName string, labelFilter map[string]string) *io_prometheus_client.Metric {
 	metricFamily := getMetricFamily(metricFamilyName)
 	if metricFamily == nil {
@@ -145,8 +411,8 @@ func isLabelsMatchWithMetric(labelFilter map[string]string, metric *io_prometheu
 	return true
 }
 
-func getTestOperationGenerator(volumePluginMgr *volume.VolumePluginMgr) OperationGenerator {
+func getTestOperationGenerator(volumePluginMgr *volume.VolumePluginMgr, objects ...runtime.Object) OperationGenerator {
-	fakeKubeClient := fakeclient.NewSimpleClientset()
+	fakeKubeClient := fakeclient.NewSimpleClientset(objects...)
 	fakeRecorder := &record.FakeRecorder{}
 	fakeHandler := volumetesting.NewBlockVolumePathHandler()
 	operationGenerator := NewOperationGenerator(

pkg/volume/util/resize_util.go

@@ -28,7 +28,9 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
 	volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
 	"k8s.io/mount-utils"
@@ -61,7 +63,7 @@ func ClaimToClaimKey(claim *v1.PersistentVolumeClaim) string {
 func UpdatePVSize(
 	pv *v1.PersistentVolume,
 	newSize resource.Quantity,
-	kubeClient clientset.Interface) error {
+	kubeClient clientset.Interface) (*v1.PersistentVolume, error) {
 	pvClone := pv.DeepCopy()
 	pvClone.Spec.Capacity[v1.ResourceStorage] = newSize
 
@@ -84,7 +86,8 @@ func AddAnnPreResizeCapacity(
 	}
 	pvClone.ObjectMeta.Annotations[AnnPreResizeCapacity] = oldCapacity.String()
 
-	return PatchPV(pv, pvClone, kubeClient)
+	_, err := PatchPV(pv, pvClone, kubeClient)
+	return err
 }
 
 // DeleteAnnPreResizeCapacity deletes volume.alpha.kubernetes.io/pre-resize-capacity from the pv
@@ -97,35 +100,35 @@ func DeleteAnnPreResizeCapacity(
 	}
 	pvClone := pv.DeepCopy()
 	delete(pvClone.ObjectMeta.Annotations, AnnPreResizeCapacity)
-
+	_, err := PatchPV(pv, pvClone, kubeClient)
-	return PatchPV(pv, pvClone, kubeClient)
+	return err
 }
 
 // PatchPV creates and executes a patch for pv
 func PatchPV(
 	oldPV *v1.PersistentVolume,
 	newPV *v1.PersistentVolume,
-	kubeClient clientset.Interface) error {
+	kubeClient clientset.Interface) (*v1.PersistentVolume, error) {
 	oldData, err := json.Marshal(oldPV)
 	if err != nil {
-		return fmt.Errorf("unexpected error marshaling old PV %q with error : %v", oldPV.Name, err)
+		return oldPV, fmt.Errorf("unexpected error marshaling old PV %q with error : %v", oldPV.Name, err)
 	}
 
 	newData, err := json.Marshal(newPV)
 	if err != nil {
-		return fmt.Errorf("unexpected error marshaling new PV %q with error : %v", newPV.Name, err)
+		return oldPV, fmt.Errorf("unexpected error marshaling new PV %q with error : %v", newPV.Name, err)
 	}
 
 	patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, oldPV)
 	if err != nil {
-		return fmt.Errorf("error Creating two way merge patch for PV %q with error : %v", oldPV.Name, err)
+		return oldPV, fmt.Errorf("error Creating two way merge patch for PV %q with error : %v", oldPV.Name, err)
 	}
 
-	_, err = kubeClient.CoreV1().PersistentVolumes().Patch(context.TODO(), oldPV.Name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{})
+	updatedPV, err := kubeClient.CoreV1().PersistentVolumes().Patch(context.TODO(), oldPV.Name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{})
 	if err != nil {
-		return fmt.Errorf("error Patching PV %q with error : %v", oldPV.Name, err)
+		return oldPV, fmt.Errorf("error Patching PV %q with error : %v", oldPV.Name, err)
 	}
-	return nil
+	return updatedPV, nil
 }
 
 // MarkResizeInProgressWithResizer marks cloudprovider resizing as in progress
@@ -147,6 +150,23 @@ func MarkResizeInProgressWithResizer(
 	return PatchPVCStatus(pvc /*oldPVC*/, newPVC, kubeClient)
 }
 
+func MarkControllerReisizeInProgress(pvc *v1.PersistentVolumeClaim, resizerName string, newSize resource.Quantity, kubeClient clientset.Interface) (*v1.PersistentVolumeClaim, error) {
+	// Mark PVC as Resize Started
+	progressCondition := v1.PersistentVolumeClaimCondition{
+		Type:               v1.PersistentVolumeClaimResizing,
+		Status:             v1.ConditionTrue,
+		LastTransitionTime: metav1.Now(),
+	}
+	controllerExpansionInProgress := v1.PersistentVolumeClaimControllerExpansionInProgress
+	conditions := []v1.PersistentVolumeClaimCondition{progressCondition}
+	newPVC := pvc.DeepCopy()
+	newPVC = MergeResizeConditionOnPVC(newPVC, conditions)
+	newPVC.Status.ResizeStatus = &controllerExpansionInProgress
+	newPVC.Status.AllocatedResources = v1.ResourceList{v1.ResourceStorage: newSize}
+	newPVC = setResizer(newPVC, resizerName)
+	return PatchPVCStatus(pvc /*oldPVC*/, newPVC, kubeClient)
+}
+
 // SetClaimResizer sets resizer annotation on PVC
 func SetClaimResizer(
 	pvc *v1.PersistentVolumeClaim,
@@ -168,7 +188,7 @@ func setResizer(pvc *v1.PersistentVolumeClaim, resizerName string) *v1.Persisten
 // MarkForFSResize marks file system resizing as pending
 func MarkForFSResize(
 	pvc *v1.PersistentVolumeClaim,
-	kubeClient clientset.Interface) error {
+	kubeClient clientset.Interface) (*v1.PersistentVolumeClaim, error) {
 	pvcCondition := v1.PersistentVolumeClaimCondition{
 		Type:               v1.PersistentVolumeClaimFileSystemResizePending,
 		Status:             v1.ConditionTrue,
@@ -177,16 +197,20 @@ func MarkForFSResize(
 	}
 	conditions := []v1.PersistentVolumeClaimCondition{pvcCondition}
 	newPVC := pvc.DeepCopy()
+	if utilfeature.DefaultFeatureGate.Enabled(features.RecoverVolumeExpansionFailure) {
+		expansionPendingOnNode := v1.PersistentVolumeClaimNodeExpansionPending
+		newPVC.Status.ResizeStatus = &expansionPendingOnNode
+	}
 	newPVC = MergeResizeConditionOnPVC(newPVC, conditions)
-	_, err := PatchPVCStatus(pvc /*oldPVC*/, newPVC, kubeClient)
+	updatedPVC, err := PatchPVCStatus(pvc /*oldPVC*/, newPVC, kubeClient)
-	return err
+	return updatedPVC, err
 }
 
 // MarkResizeFinished marks all resizing as done
 func MarkResizeFinished(
 	pvc *v1.PersistentVolumeClaim,
 	newSize resource.Quantity,
-	kubeClient clientset.Interface) error {
+	kubeClient clientset.Interface) (*v1.PersistentVolumeClaim, error) {
 	return MarkFSResizeFinished(pvc, newSize, kubeClient)
 }
 
@@ -194,12 +218,65 @@ func MarkResizeFinished(
 func MarkFSResizeFinished(
 	pvc *v1.PersistentVolumeClaim,
 	newSize resource.Quantity,
-	kubeClient clientset.Interface) error {
+	kubeClient clientset.Interface) (*v1.PersistentVolumeClaim, error) {
 	newPVC := pvc.DeepCopy()
+
 	newPVC.Status.Capacity[v1.ResourceStorage] = newSize
+
+	// if RecoverVolumeExpansionFailure is enabled, we need to reset ResizeStatus back to nil
+	if utilfeature.DefaultFeatureGate.Enabled(features.RecoverVolumeExpansionFailure) {
+		expansionFinished := v1.PersistentVolumeClaimNoExpansionInProgress
+		newPVC.Status.ResizeStatus = &expansionFinished
+	}
+
 	newPVC = MergeResizeConditionOnPVC(newPVC, []v1.PersistentVolumeClaimCondition{})
-	_, err := PatchPVCStatus(pvc /*oldPVC*/, newPVC, kubeClient)
+	updatedPVC, err := PatchPVCStatus(pvc /*oldPVC*/, newPVC, kubeClient)
-	return err
+	return updatedPVC, err
+}
+
+func MarkControllerExpansionFailed(pvc *v1.PersistentVolumeClaim, kubeClient clientset.Interface) (*v1.PersistentVolumeClaim, error) {
+	expansionFailedOnController := v1.PersistentVolumeClaimControllerExpansionFailed
+	newPVC := pvc.DeepCopy()
+	newPVC.Status.ResizeStatus = &expansionFailedOnController
+	patchBytes, err := createPVCPatch(pvc, newPVC, false /* addResourceVersionCheck */)
+	if err != nil {
+		return pvc, fmt.Errorf("patchPVCStatus failed to patch PVC %q: %v", pvc.Name, err)
+	}
+
+	updatedClaim, updateErr := kubeClient.CoreV1().PersistentVolumeClaims(pvc.Namespace).
+		Patch(context.TODO(), pvc.Name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}, "status")
+	if updateErr != nil {
+		return pvc, fmt.Errorf("patchPVCStatus failed to patch PVC %q: %v", pvc.Name, updateErr)
+	}
+	return updatedClaim, nil
+}
+
+// MarkNodeExpansionFailed marks a PVC for node expansion as failed. Kubelet should not retry expansion
+// of volumes which are in failed state.
+func MarkNodeExpansionFailed(pvc *v1.PersistentVolumeClaim, kubeClient clientset.Interface) (*v1.PersistentVolumeClaim, error) {
+	expansionFailedOnNode := v1.PersistentVolumeClaimNodeExpansionFailed
+	newPVC := pvc.DeepCopy()
+	newPVC.Status.ResizeStatus = &expansionFailedOnNode
+	patchBytes, err := createPVCPatch(pvc, newPVC, false /* addResourceVersionCheck */)
+	if err != nil {
+		return pvc, fmt.Errorf("patchPVCStatus failed to patch PVC %q: %v", pvc.Name, err)
+	}
+
+	updatedClaim, updateErr := kubeClient.CoreV1().PersistentVolumeClaims(pvc.Namespace).
+		Patch(context.TODO(), pvc.Name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}, "status")
+	if updateErr != nil {
+		return pvc, fmt.Errorf("patchPVCStatus failed to patch PVC %q: %v", pvc.Name, updateErr)
+	}
+	return updatedClaim, nil
+}
+
+// MarkNodeExpansionInProgress marks pvc expansion in progress on node
+func MarkNodeExpansionInProgress(pvc *v1.PersistentVolumeClaim, kubeClient clientset.Interface) (*v1.PersistentVolumeClaim, error) {
+	nodeExpansionInProgress := v1.PersistentVolumeClaimNodeExpansionInProgress
+	newPVC := pvc.DeepCopy()
+	newPVC.Status.ResizeStatus = &nodeExpansionInProgress
+	updatedPVC, err := PatchPVCStatus(pvc /* oldPVC */, newPVC, kubeClient)
+	return updatedPVC, err
 }
 
 // PatchPVCStatus updates PVC status using PATCH verb
@@ -210,22 +287,22 @@ func PatchPVCStatus(
 	oldPVC *v1.PersistentVolumeClaim,
 	newPVC *v1.PersistentVolumeClaim,
 	kubeClient clientset.Interface) (*v1.PersistentVolumeClaim, error) {
-	patchBytes, err := createPVCPatch(oldPVC, newPVC)
+	patchBytes, err := createPVCPatch(oldPVC, newPVC, true /* addResourceVersionCheck */)
 	if err != nil {
-		return nil, fmt.Errorf("patchPVCStatus failed to patch PVC %q: %v", oldPVC.Name, err)
+		return oldPVC, fmt.Errorf("patchPVCStatus failed to patch PVC %q: %v", oldPVC.Name, err)
 	}
 
 	updatedClaim, updateErr := kubeClient.CoreV1().PersistentVolumeClaims(oldPVC.Namespace).
 		Patch(context.TODO(), oldPVC.Name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}, "status")
 	if updateErr != nil {
-		return nil, fmt.Errorf("patchPVCStatus failed to patch PVC %q: %v", oldPVC.Name, updateErr)
+		return oldPVC, fmt.Errorf("patchPVCStatus failed to patch PVC %q: %v", oldPVC.Name, updateErr)
 	}
 	return updatedClaim, nil
 }
 
 func createPVCPatch(
 	oldPVC *v1.PersistentVolumeClaim,
-	newPVC *v1.PersistentVolumeClaim) ([]byte, error) {
+	newPVC *v1.PersistentVolumeClaim, addResourceVersionCheck bool) ([]byte, error) {
 	oldData, err := json.Marshal(oldPVC)
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal old data: %v", err)
@@ -241,9 +318,11 @@ func createPVCPatch(
 		return nil, fmt.Errorf("failed to create 2 way merge patch: %v", err)
 	}
 
-	patchBytes, err = addResourceVersion(patchBytes, oldPVC.ResourceVersion)
+	if addResourceVersionCheck {
-	if err != nil {
+		patchBytes, err = addResourceVersion(patchBytes, oldPVC.ResourceVersion)
-		return nil, fmt.Errorf("failed to add resource version: %v", err)
+		if err != nil {
+			return nil, fmt.Errorf("failed to add resource version: %v", err)
+		}
 	}
 
 	return patchBytes, nil

pkg/volume/util/resize_util_test.go

@@ -155,7 +155,7 @@ func TestCreatePVCPatch(t *testing.T) {
 	pvc2.Status.Capacity = v1.ResourceList{
 		v1.ResourceName("size"): resource.MustParse("10G"),
 	}
-	patchBytes, err := createPVCPatch(pvc1, pvc2)
+	patchBytes, err := createPVCPatch(pvc1, pvc2, true /* addResourceVersionCheck */)
 	if err != nil {
 		t.Errorf("error creating patch bytes %v", err)
 	}

staging/src/k8s.io/api/core/v1/generated.proto

@@ -2665,6 +2665,9 @@ message PersistentVolumeClaimSpec {
   optional k8s.io.apimachinery.pkg.apis.meta.v1.LabelSelector selector = 4;
 
   // Resources represents the minimum resources the volume should have.
+  // If RecoverVolumeExpansionFailure feature is enabled users are allowed to specify resource requirements
+  // that are lower than previous value but must still be higher than capacity recorded in the
+  // status field of the claim.
   // More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#resources
   // +optional
   optional ResourceRequirements resources = 2;
@@ -2735,6 +2738,26 @@ message PersistentVolumeClaimStatus {
   // +patchMergeKey=type
   // +patchStrategy=merge
   repeated PersistentVolumeClaimCondition conditions = 4;
+
+  // The storage resource within AllocatedResources tracks the capacity allocated to a PVC. It may
+  // be larger than the actual capacity when a volume expansion operation is requested.
+  // For storage quota, the larger value from allocatedResources and PVC.spec.resources is used.
+  // If allocatedResources is not set, PVC.spec.resources alone is used for quota calculation.
+  // If a volume expansion capacity request is lowered, allocatedResources is only
+  // lowered if there are no expansion operations in progress and if the actual volume capacity
+  // is equal or lower than the requested capacity.
+  // This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.
+  // +featureGate=RecoverVolumeExpansionFailure
+  // +optional
+  map<string, k8s.io.apimachinery.pkg.api.resource.Quantity> allocatedResources = 5;
+
+  // ResizeStatus stores status of resize operation.
+  // ResizeStatus is not set by default but when expansion is complete resizeStatus is set to empty
+  // string by resize controller or kubelet.
+  // This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.
+  // +featureGate=RecoverVolumeExpansionFailure
+  // +optional
+  optional string resizeStatus = 6;
 }
 
 // PersistentVolumeClaimTemplate is used to produce

staging/src/k8s.io/api/core/v1/types.go

@@ -472,6 +472,9 @@ type PersistentVolumeClaimSpec struct {
 	// +optional
 	Selector *metav1.LabelSelector `json:"selector,omitempty" protobuf:"bytes,4,opt,name=selector"`
 	// Resources represents the minimum resources the volume should have.
+	// If RecoverVolumeExpansionFailure feature is enabled users are allowed to specify resource requirements
+	// that are lower than previous value but must still be higher than capacity recorded in the
+	// status field of the claim.
 	// More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#resources
 	// +optional
 	Resources ResourceRequirements `json:"resources,omitempty" protobuf:"bytes,2,opt,name=resources"`
@@ -526,6 +529,26 @@ const (
 	PersistentVolumeClaimFileSystemResizePending PersistentVolumeClaimConditionType = "FileSystemResizePending"
 )
 
+// +enum
+type PersistentVolumeClaimResizeStatus string
+
+const (
+	// When expansion is complete, the empty string is set by resize controller or kubelet.
+	PersistentVolumeClaimNoExpansionInProgress PersistentVolumeClaimResizeStatus = ""
+	// State set when resize controller starts expanding the volume in control-plane
+	PersistentVolumeClaimControllerExpansionInProgress PersistentVolumeClaimResizeStatus = "ControllerExpansionInProgress"
+	// State set when expansion has failed in resize controller with a terminal error.
+	// Transient errors such as timeout should not set this status and should leave ResizeStatus
+	// unmodified, so as resize controller can resume the volume expansion.
+	PersistentVolumeClaimControllerExpansionFailed PersistentVolumeClaimResizeStatus = "ControllerExpansionFailed"
+	// State set when resize controller has finished expanding the volume but further expansion is needed on the node.
+	PersistentVolumeClaimNodeExpansionPending PersistentVolumeClaimResizeStatus = "NodeExpansionPending"
+	// State set when kubelet starts expanding the volume.
+	PersistentVolumeClaimNodeExpansionInProgress PersistentVolumeClaimResizeStatus = "NodeExpansionInProgress"
+	// State set when expansion has failed in kubelet with a terminal error. Transient errors don't set NodeExpansionFailed.
+	PersistentVolumeClaimNodeExpansionFailed PersistentVolumeClaimResizeStatus = "NodeExpansionFailed"
+)
+
 // PersistentVolumeClaimCondition contails details about state of pvc
 type PersistentVolumeClaimCondition struct {
 	Type   PersistentVolumeClaimConditionType `json:"type" protobuf:"bytes,1,opt,name=type,casttype=PersistentVolumeClaimConditionType"`
@@ -564,6 +587,24 @@ type PersistentVolumeClaimStatus struct {
 	// +patchMergeKey=type
 	// +patchStrategy=merge
 	Conditions []PersistentVolumeClaimCondition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type" protobuf:"bytes,4,rep,name=conditions"`
+	// The storage resource within AllocatedResources tracks the capacity allocated to a PVC. It may
+	// be larger than the actual capacity when a volume expansion operation is requested.
+	// For storage quota, the larger value from allocatedResources and PVC.spec.resources is used.
+	// If allocatedResources is not set, PVC.spec.resources alone is used for quota calculation.
+	// If a volume expansion capacity request is lowered, allocatedResources is only
+	// lowered if there are no expansion operations in progress and if the actual volume capacity
+	// is equal or lower than the requested capacity.
+	// This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.
+	// +featureGate=RecoverVolumeExpansionFailure
+	// +optional
+	AllocatedResources ResourceList `json:"allocatedResources,omitempty" protobuf:"bytes,5,rep,name=allocatedResources,casttype=ResourceList,castkey=ResourceName"`
+	// ResizeStatus stores status of resize operation.
+	// ResizeStatus is not set by default but when expansion is complete resizeStatus is set to empty
+	// string by resize controller or kubelet.
+	// This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.
+	// +featureGate=RecoverVolumeExpansionFailure
+	// +optional
+	ResizeStatus *PersistentVolumeClaimResizeStatus `json:"resizeStatus,omitempty" protobuf:"bytes,6,opt,name=resizeStatus,casttype=PersistentVolumeClaimResizeStatus"`
 }
 
 type PersistentVolumeAccessMode string

staging/src/k8s.io/api/core/v1/types_swagger_doc_generated.go

@@ -1297,7 +1297,7 @@ var map_PersistentVolumeClaimSpec = map[string]string{
 	"":                 "PersistentVolumeClaimSpec describes the common attributes of storage devices and allows a Source for provider-specific attributes",
 	"accessModes":      "AccessModes contains the desired access modes the volume should have. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#access-modes-1",
 	"selector":         "A label query over volumes to consider for binding.",
-	"resources":        "Resources represents the minimum resources the volume should have. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#resources",
+	"resources":        "Resources represents the minimum resources the volume should have. If RecoverVolumeExpansionFailure feature is enabled users are allowed to specify resource requirements that are lower than previous value but must still be higher than capacity recorded in the status field of the claim. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#resources",
 	"volumeName":       "VolumeName is the binding reference to the PersistentVolume backing this claim.",
 	"storageClassName": "Name of the StorageClass required by the claim. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#class-1",
 	"volumeMode":       "volumeMode defines what type of volume is required by the claim. Value of Filesystem is implied when not included in claim spec.",
@@ -1310,11 +1310,13 @@ func (PersistentVolumeClaimSpec) SwaggerDoc() map[string]string {
 }
 
 var map_PersistentVolumeClaimStatus = map[string]string{
-	"":            "PersistentVolumeClaimStatus is the current status of a persistent volume claim.",
+	"":                   "PersistentVolumeClaimStatus is the current status of a persistent volume claim.",
-	"phase":       "Phase represents the current phase of PersistentVolumeClaim.",
+	"phase":              "Phase represents the current phase of PersistentVolumeClaim.",
-	"accessModes": "AccessModes contains the actual access modes the volume backing the PVC has. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#access-modes-1",
+	"accessModes":        "AccessModes contains the actual access modes the volume backing the PVC has. More info: https://kubernetes.io/docs/concepts/storage/persistent-volumes#access-modes-1",
-	"capacity":    "Represents the actual resources of the underlying volume.",
+	"capacity":           "Represents the actual resources of the underlying volume.",
-	"conditions":  "Current Condition of persistent volume claim. If underlying persistent volume is being resized then the Condition will be set to 'ResizeStarted'.",
+	"conditions":         "Current Condition of persistent volume claim. If underlying persistent volume is being resized then the Condition will be set to 'ResizeStarted'.",
+	"allocatedResources": "The storage resource within AllocatedResources tracks the capacity allocated to a PVC. It may be larger than the actual capacity when a volume expansion operation is requested. For storage quota, the larger value from allocatedResources and PVC.spec.resources is used. If allocatedResources is not set, PVC.spec.resources alone is used for quota calculation. If a volume expansion capacity request is lowered, allocatedResources is only lowered if there are no expansion operations in progress and if the actual volume capacity is equal or lower than the requested capacity. This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.",
+	"resizeStatus":       "ResizeStatus stores status of resize operation. ResizeStatus is not set by default but when expansion is complete resizeStatus is set to empty string by resize controller or kubelet. This is an alpha field and requires enabling RecoverVolumeExpansionFailure feature.",
 }
 
 func (PersistentVolumeClaimStatus) SwaggerDoc() map[string]string {

staging/src/k8s.io/api/core/v1/zz_generated.deepcopy.go

@@ -2975,6 +2975,18 @@ func (in *PersistentVolumeClaimStatus) DeepCopyInto(out *PersistentVolumeClaimSt
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
 	}
+	if in.AllocatedResources != nil {
+		in, out := &in.AllocatedResources, &out.AllocatedResources
+		*out = make(ResourceList, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val.DeepCopy()
+		}
+	}
+	if in.ResizeStatus != nil {
+		in, out := &in.ResizeStatus, &out.ResizeStatus
+		*out = new(PersistentVolumeClaimResizeStatus)
+		**out = **in
+	}
 	return
 }
 

staging/src/k8s.io/api/testdata/HEAD/apps.v1.StatefulSet.json

@@ -1642,39 +1642,43 @@
               "reason": "523",
               "message": "524"
             }
-          ]
+          ],
+          "allocatedResources": {
+            "鐳VDɝ": "844"
+          },
+          "resizeStatus": "但Ǭľa执mÎDƃ"
         }
       }
     ],
     "serviceName": "525",
-    "podManagementPolicy": "婵=ǻ",
+    "podManagementPolicy": "ƌ妜;)t罎j´A",
     "updateStrategy": {
-      "type": "ɝÔȗ$`Ž#",
+      "type": "徙蔿Yċʤw俣Ǫ",
       "rollingUpdate": {
-        "partition": -1644040574
+        "partition": 2000146968
       }
     },
-    "revisionHistoryLimit": -1205784111,
+    "revisionHistoryLimit": 560461224,
-    "minReadySeconds": 1505300966
+    "minReadySeconds": 2059121580
   },
   "status": {
-    "observedGeneration": 7422250233075984176,
+    "observedGeneration": -632886252136267545,
-    "replicas": -326265137,
+    "replicas": 750655684,
-    "readyReplicas": 1683394621,
+    "readyReplicas": -1012893423,
-    "currentReplicas": 1862659237,
+    "currentReplicas": -1295777734,
-    "updatedReplicas": 798811297,
+    "updatedReplicas": -1394190312,
     "currentRevision": "526",
     "updateRevision": "527",
-    "collisionCount": -1290326833,
+    "collisionCount": 1077247354,
     "conditions": [
       {
-        "type": "´Aƺå嫹^Ȇɀ*ǹ",
+        "type": "堏ȑ湸睑L暱ʖ妾崗",
-        "status": "蟷尨BABȳ",
+        "status": "ij敪賻yʗHiv\u003c",
-        "lastTransitionTime": "2464-04-30T23:47:03Z",
+        "lastTransitionTime": "2814-04-22T10:44:02Z",
         "reason": "528",
         "message": "529"
       }
     ],
-    "availableReplicas": -1012893423
+    "availableReplicas": 747018016
   }
 }

staging/src/k8s.io/api/testdata/HEAD/apps.v1.StatefulSet.yaml

@@ -31,10 +31,10 @@ metadata:
   selfLink: "5"
   uid: "7"
 spec:
-  minReadySeconds: 1505300966
+  minReadySeconds: 2059121580
-  podManagementPolicy: 婵=ǻ
+  podManagementPolicy: ƌ妜;)t罎j´A
   replicas: 896585016
-  revisionHistoryLimit: -1205784111
+  revisionHistoryLimit: 560461224
   selector:
     matchExpressions:
     - key: 50-u--25cu87--r7p-w1e67-8pj5t-kl-v0q6b68--nu5oii38fn-8.629b-jd-8c45-0-8--6n--w0--w---196g8d--iv1-5--5ht-a-29--0qso796/3___47._49pIB_o61ISU4--A_.XK_._M99
@@ -1060,8 +1060,8 @@ spec:
           volumePath: "103"
   updateStrategy:
     rollingUpdate:
-      partition: -1644040574
+      partition: 2000146968
-    type: ɝÔȗ$`Ž#
+    type: 徙蔿Yċʤw俣Ǫ
   volumeClaimTemplates:
   - metadata:
       annotations:
@@ -1123,6 +1123,8 @@ spec:
     status:
       accessModes:
       - v}鮩澊聝楧
+      allocatedResources:
+        鐳VDɝ: "844"
       capacity:
         问Ð7ɞŶJŖ)j{驟ʦcȃ: "657"
       conditions:
@@ -1133,19 +1135,20 @@ spec:
         status: Q¢鬣_棈Ý泷
         type: ņȎZȐ樾'Ż£劾ů
       phase: 忣àÂƺ琰Ȃ芋醳鮩!廊臚cɶċ
+      resizeStatus: 但Ǭľa执mÎDƃ
 status:
-  availableReplicas: -1012893423
+  availableReplicas: 747018016
-  collisionCount: -1290326833
+  collisionCount: 1077247354
   conditions:
-  - lastTransitionTime: "2464-04-30T23:47:03Z"
+  - lastTransitionTime: "2814-04-22T10:44:02Z"
     message: "529"
     reason: "528"
-    status: 蟷尨BABȳ
+    status: ij敪賻yʗHiv<
-    type: ´Aƺå嫹^Ȇɀ*ǹ
+    type: 堏ȑ湸睑L暱ʖ妾崗
-  currentReplicas: 1862659237
+  currentReplicas: -1295777734
   currentRevision: "526"
-  observedGeneration: 7422250233075984176
+  observedGeneration: -632886252136267545
-  readyReplicas: 1683394621
+  readyReplicas: -1012893423
-  replicas: -326265137
+  replicas: 750655684
   updateRevision: "527"
-  updatedReplicas: 798811297
+  updatedReplicas: -1394190312

staging/src/k8s.io/api/testdata/HEAD/apps.v1beta1.StatefulSet.json

@@ -1642,39 +1642,43 @@
               "reason": "523",
               "message": "524"
             }
-          ]
+          ],
+          "allocatedResources": {
+            "鐳VDɝ": "844"
+          },
+          "resizeStatus": "但Ǭľa执mÎDƃ"
         }
       }
     ],
     "serviceName": "525",
-    "podManagementPolicy": "婵=ǻ",
+    "podManagementPolicy": "ƌ妜;)t罎j´A",
     "updateStrategy": {
-      "type": "ɝÔȗ$`Ž#",
+      "type": "徙蔿Yċʤw俣Ǫ",
       "rollingUpdate": {
-        "partition": -1644040574
+        "partition": 2000146968
       }
     },
-    "revisionHistoryLimit": -1205784111,
+    "revisionHistoryLimit": 560461224,
-    "minReadySeconds": 1505300966
+    "minReadySeconds": 2059121580
   },
   "status": {
-    "observedGeneration": 2345785178116014414,
+    "observedGeneration": -8200913189823252840,
-    "replicas": 923301621,
+    "replicas": 1892314617,
-    "readyReplicas": 2036280873,
+    "readyReplicas": -1893854851,
-    "currentReplicas": 2102009515,
+    "currentReplicas": 658548230,
-    "updatedReplicas": -1974512490,
+    "updatedReplicas": -301228056,
     "currentRevision": "526",
     "updateRevision": "527",
-    "collisionCount": -1001798049,
+    "collisionCount": 446542989,
     "conditions": [
       {
-        "type": "Ǒl徙蔿Yċʤw",
+        "type": "Ǚ3洠º襊Ł靫挕欰ij敪賻yʗHiv",
-        "status": "ǹ脡È6",
+        "status": "V汦\u003e蒃U",
-        "lastTransitionTime": "2744-07-10T16:37:22Z",
+        "lastTransitionTime": "2800-08-07T22:03:04Z",
         "reason": "528",
         "message": "529"
       }
     ],
-    "availableReplicas": -180607525
+    "availableReplicas": -2059927818
   }
 }

staging/src/k8s.io/api/testdata/HEAD/apps.v1beta1.StatefulSet.yaml

@@ -31,10 +31,10 @@ metadata:
   selfLink: "5"
   uid: "7"
 spec:
-  minReadySeconds: 1505300966
+  minReadySeconds: 2059121580
-  podManagementPolicy: 婵=ǻ
+  podManagementPolicy: ƌ妜;)t罎j´A
   replicas: 896585016
-  revisionHistoryLimit: -1205784111
+  revisionHistoryLimit: 560461224
   selector:
     matchExpressions:
     - key: 50-u--25cu87--r7p-w1e67-8pj5t-kl-v0q6b68--nu5oii38fn-8.629b-jd-8c45-0-8--6n--w0--w---196g8d--iv1-5--5ht-a-29--0qso796/3___47._49pIB_o61ISU4--A_.XK_._M99
@@ -1060,8 +1060,8 @@ spec:
           volumePath: "103"
   updateStrategy:
     rollingUpdate:
-      partition: -1644040574
+      partition: 2000146968
-    type: ɝÔȗ$`Ž#
+    type: 徙蔿Yċʤw俣Ǫ
   volumeClaimTemplates:
   - metadata:
       annotations:
@@ -1123,6 +1123,8 @@ spec:
     status:
       accessModes:
       - v}鮩澊聝楧
+      allocatedResources:
+        鐳VDɝ: "844"
       capacity:
         问Ð7ɞŶJŖ)j{驟ʦcȃ: "657"
       conditions:
@@ -1133,19 +1135,20 @@ spec:
         status: Q¢鬣_棈Ý泷
         type: ņȎZȐ樾'Ż£劾ů
       phase: 忣àÂƺ琰Ȃ芋醳鮩!廊臚cɶċ
+      resizeStatus: 但Ǭľa执mÎDƃ
 status:
-  availableReplicas: -180607525
+  availableReplicas: -2059927818
-  collisionCount: -1001798049
+  collisionCount: 446542989
   conditions:
-  - lastTransitionTime: "2744-07-10T16:37:22Z"
+  - lastTransitionTime: "2800-08-07T22:03:04Z"
     message: "529"
     reason: "528"
-    status: ǹ脡È6
+    status: V汦>蒃U
-    type: Ǒl徙蔿Yċʤw
+    type: Ǚ3洠º襊Ł靫挕欰ij敪賻yʗHiv
-  currentReplicas: 2102009515
+  currentReplicas: 658548230
   currentRevision: "526"
-  observedGeneration: 2345785178116014414
+  observedGeneration: -8200913189823252840
-  readyReplicas: 2036280873
+  readyReplicas: -1893854851
-  replicas: 923301621
+  replicas: 1892314617
   updateRevision: "527"
-  updatedReplicas: -1974512490
+  updatedReplicas: -301228056

staging/src/k8s.io/api/testdata/HEAD/apps.v1beta2.StatefulSet.json

@@ -1642,39 +1642,43 @@
               "reason": "523",
               "message": "524"
             }
-          ]
+          ],
+          "allocatedResources": {
+            "鐳VDɝ": "844"
+          },
+          "resizeStatus": "但Ǭľa执mÎDƃ"
         }
       }
     ],
     "serviceName": "525",
-    "podManagementPolicy": "婵=ǻ",
+    "podManagementPolicy": "ƌ妜;)t罎j´A",
     "updateStrategy": {
-      "type": "ɝÔȗ$`Ž#",
+      "type": "徙蔿Yċʤw俣Ǫ",
       "rollingUpdate": {
-        "partition": -1644040574
+        "partition": 2000146968
       }
     },
-    "revisionHistoryLimit": -1205784111,
+    "revisionHistoryLimit": 560461224,
-    "minReadySeconds": 1505300966
+    "minReadySeconds": 2059121580
   },
   "status": {
-    "observedGeneration": 7422250233075984176,
+    "observedGeneration": -632886252136267545,
-    "replicas": -326265137,
+    "replicas": 750655684,
-    "readyReplicas": 1683394621,
+    "readyReplicas": -1012893423,
-    "currentReplicas": 1862659237,
+    "currentReplicas": -1295777734,
-    "updatedReplicas": 798811297,
+    "updatedReplicas": -1394190312,
     "currentRevision": "526",
     "updateRevision": "527",
-    "collisionCount": -1290326833,
+    "collisionCount": 1077247354,
     "conditions": [
       {
-        "type": "´Aƺå嫹^Ȇɀ*ǹ",
+        "type": "堏ȑ湸睑L暱ʖ妾崗",
-        "status": "蟷尨BABȳ",
+        "status": "ij敪賻yʗHiv\u003c",
-        "lastTransitionTime": "2464-04-30T23:47:03Z",
+        "lastTransitionTime": "2814-04-22T10:44:02Z",
         "reason": "528",
         "message": "529"
       }
     ],
-    "availableReplicas": -1012893423
+    "availableReplicas": 747018016
   }
 }

staging/src/k8s.io/api/testdata/HEAD/apps.v1beta2.StatefulSet.yaml

@@ -31,10 +31,10 @@ metadata:
   selfLink: "5"
   uid: "7"
 spec:
-  minReadySeconds: 1505300966
+  minReadySeconds: 2059121580
-  podManagementPolicy: 婵=ǻ
+  podManagementPolicy: ƌ妜;)t罎j´A
   replicas: 896585016
-  revisionHistoryLimit: -1205784111
+  revisionHistoryLimit: 560461224
   selector:
     matchExpressions:
     - key: 50-u--25cu87--r7p-w1e67-8pj5t-kl-v0q6b68--nu5oii38fn-8.629b-jd-8c45-0-8--6n--w0--w---196g8d--iv1-5--5ht-a-29--0qso796/3___47._49pIB_o61ISU4--A_.XK_._M99
@@ -1060,8 +1060,8 @@ spec:
           volumePath: "103"
   updateStrategy:
     rollingUpdate:
-      partition: -1644040574
+      partition: 2000146968
-    type: ɝÔȗ$`Ž#
+    type: 徙蔿Yċʤw俣Ǫ
   volumeClaimTemplates:
   - metadata:
       annotations:
@@ -1123,6 +1123,8 @@ spec:
     status:
       accessModes:
       - v}鮩澊聝楧
+      allocatedResources:
+        鐳VDɝ: "844"
       capacity:
         问Ð7ɞŶJŖ)j{驟ʦcȃ: "657"
       conditions:
@@ -1133,19 +1135,20 @@ spec:
         status: Q¢鬣_棈Ý泷
         type: ņȎZȐ樾'Ż£劾ů
       phase: 忣àÂƺ琰Ȃ芋醳鮩!廊臚cɶċ
+      resizeStatus: 但Ǭľa执mÎDƃ
 status:
-  availableReplicas: -1012893423
+  availableReplicas: 747018016
-  collisionCount: -1290326833
+  collisionCount: 1077247354
   conditions:
-  - lastTransitionTime: "2464-04-30T23:47:03Z"
+  - lastTransitionTime: "2814-04-22T10:44:02Z"
     message: "529"
     reason: "528"
-    status: 蟷尨BABȳ
+    status: ij敪賻yʗHiv<
-    type: ´Aƺå嫹^Ȇɀ*ǹ
+    type: 堏ȑ湸睑L暱ʖ妾崗
-  currentReplicas: 1862659237
+  currentReplicas: -1295777734
   currentRevision: "526"
-  observedGeneration: 7422250233075984176
+  observedGeneration: -632886252136267545
-  readyReplicas: 1683394621
+  readyReplicas: -1012893423
-  replicas: -326265137
+  replicas: 750655684
   updateRevision: "527"
-  updatedReplicas: 798811297
+  updatedReplicas: -1394190312

staging/src/k8s.io/api/testdata/HEAD/core.v1.PersistentVolumeClaim.json

@@ -95,6 +95,10 @@
         "reason": "34",
         "message": "35"
       }
-    ]
+    ],
+    "allocatedResources": {
+      " u衲\u003c¿燥": "98"
+    },
+    "resizeStatus": "{舁吉蓨O澘"
   }
 }

staging/src/k8s.io/api/testdata/HEAD/core.v1.PersistentVolumeClaim.yaml

@@ -58,6 +58,8 @@ spec:
 status:
   accessModes:
   - l殛瓷雼浢Ü礽
+  allocatedResources:
+    ' u衲<¿燥': "98"
   capacity:
     '{囥': "721"
   conditions:
@@ -68,3 +70,4 @@ status:
     status: Ka縳讋ɮ衺勽Ƙq
     type: n(鲼ƳÐƣKʘńw:5塋訩塶"=
   phase: gɸ=ǤÆ碛,1ZƜ/C龷Ȫ
+  resizeStatus: '{舁吉蓨O澘'

staging/src/k8s.io/client-go/applyconfigurations/core/v1/persistentvolumeclaimstatus.go

@@ -25,10 +25,12 @@ import (
 // PersistentVolumeClaimStatusApplyConfiguration represents an declarative configuration of the PersistentVolumeClaimStatus type for use
 // with apply.
 type PersistentVolumeClaimStatusApplyConfiguration struct {
-	Phase       *v1.PersistentVolumeClaimPhase                     `json:"phase,omitempty"`
+	Phase              *v1.PersistentVolumeClaimPhase                     `json:"phase,omitempty"`
-	AccessModes []v1.PersistentVolumeAccessMode                    `json:"accessModes,omitempty"`
+	AccessModes        []v1.PersistentVolumeAccessMode                    `json:"accessModes,omitempty"`
-	Capacity    *v1.ResourceList                                   `json:"capacity,omitempty"`
+	Capacity           *v1.ResourceList                                   `json:"capacity,omitempty"`
-	Conditions  []PersistentVolumeClaimConditionApplyConfiguration `json:"conditions,omitempty"`
+	Conditions         []PersistentVolumeClaimConditionApplyConfiguration `json:"conditions,omitempty"`
+	AllocatedResources *v1.ResourceList                                   `json:"allocatedResources,omitempty"`
+	ResizeStatus       *v1.PersistentVolumeClaimResizeStatus              `json:"resizeStatus,omitempty"`
 }
 
 // PersistentVolumeClaimStatusApplyConfiguration constructs an declarative configuration of the PersistentVolumeClaimStatus type for use with
@@ -75,3 +77,19 @@ func (b *PersistentVolumeClaimStatusApplyConfiguration) WithConditions(values ..
 	}
 	return b
 }
+
+// WithAllocatedResources sets the AllocatedResources field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the AllocatedResources field is set to the value of the last call.
+func (b *PersistentVolumeClaimStatusApplyConfiguration) WithAllocatedResources(value v1.ResourceList) *PersistentVolumeClaimStatusApplyConfiguration {
+	b.AllocatedResources = &value
+	return b
+}
+
+// WithResizeStatus sets the ResizeStatus field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the ResizeStatus field is set to the value of the last call.
+func (b *PersistentVolumeClaimStatusApplyConfiguration) WithResizeStatus(value v1.PersistentVolumeClaimResizeStatus) *PersistentVolumeClaimStatusApplyConfiguration {
+	b.ResizeStatus = &value
+	return b
+}

staging/src/k8s.io/client-go/applyconfigurations/internal/internal.go

@@ -5291,6 +5291,11 @@ var schemaYAML = typed.YAMLObject(`types:
           elementType:
             scalar: string
           elementRelationship: atomic
+    - name: allocatedResources
+      type:
+        map:
+          elementType:
+            namedType: io.k8s.apimachinery.pkg.api.resource.Quantity
     - name: capacity
       type:
         map:
@@ -5307,6 +5312,9 @@ var schemaYAML = typed.YAMLObject(`types:
     - name: phase
       type:
         scalar: string
+    - name: resizeStatus
+      type:
+        scalar: string
 - name: io.k8s.api.core.v1.PersistentVolumeClaimTemplate
   map:
     fields: