Add storage capacity scoring to VolumeBinding plugin

Author: bells17

pkg/features/kube_features.go

@@ -677,6 +677,13 @@ const (
 	// Enables trafficDistribution field on Services.
 	ServiceTrafficDistribution featuregate.Feature = "ServiceTrafficDistribution"
 
+	// owner: @cupnes
+	// kep: https://kep.k8s.io/4049
+	//
+	// Enables scoring nodes by available storage capacity with
+	// StorageCapacityScoring feature gate (dynamic provisioning only).
+	StorageCapacityScoring featuregate.Feature = "StorageCapacityScoring"
+
 	// owner: @gjkim42 @SergeyKanzhelev @matthyx @tzneal
 	// kep: http://kep.k8s.io/753
 	//

pkg/features/versioned_kube_features.go

@@ -753,6 +753,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate
 		{Version: version.MustParse("1.31"), Default: true, PreRelease: featuregate.GA, LockToDefault: true}, // GA in 1.31, remove in 1.33
 	},
 
+	StorageCapacityScoring: {
+		{Version: version.MustParse("1.33"), Default: false, PreRelease: featuregate.Alpha},
+	},
+
 	StorageVersionMigrator: {
 		{Version: version.MustParse("1.30"), Default: false, PreRelease: featuregate.Alpha},
 	},

pkg/scheduler/apis/config/v1/defaults.go

@@ -193,15 +193,28 @@ func SetDefaults_VolumeBindingArgs(obj *configv1.VolumeBindingArgs) {
 		obj.BindTimeoutSeconds = ptr.To[int64](600)
 	}
 	if len(obj.Shape) == 0 && feature.DefaultFeatureGate.Enabled(features.VolumeCapacityPriority) {
-		obj.Shape = []configv1.UtilizationShapePoint{
-			{
-				Utilization: 0,
-				Score:       0,
-			},
-			{
-				Utilization: 100,
-				Score:       int32(config.MaxCustomPriorityScore),
-			},
+		if feature.DefaultFeatureGate.Enabled(features.StorageCapacityScoring) {
+			obj.Shape = []configv1.UtilizationShapePoint{
+				{
+					Utilization: 0,
+					Score:       int32(config.MaxCustomPriorityScore),
+				},
+				{
+					Utilization: 100,
+					Score:       0,
+				},
+			}
+		} else {
+			obj.Shape = []configv1.UtilizationShapePoint{
+				{
+					Utilization: 0,
+					Score:       0,
+				},
+				{
+					Utilization: 100,
+					Score:       int32(config.MaxCustomPriorityScore),
+				},
+			}
 		}
 	}
 }

pkg/scheduler/framework/plugins/feature/feature.go

@@ -33,4 +33,5 @@ type Features struct {
 	EnableSchedulingQueueHint                    bool
 	EnableAsyncPreemption                        bool
 	EnablePodLevelResources                      bool
+	EnableStorageCapacityScoring                 bool
 }

pkg/scheduler/framework/plugins/registry.go

@@ -59,6 +59,7 @@ func NewInTreeRegistry() runtime.Registry {
 		EnableSchedulingQueueHint:                    feature.DefaultFeatureGate.Enabled(features.SchedulerQueueingHints),
 		EnableAsyncPreemption:                        feature.DefaultFeatureGate.Enabled(features.SchedulerAsyncPreemption),
 		EnablePodLevelResources:                      feature.DefaultFeatureGate.Enabled(features.PodLevelResources),
+		EnableStorageCapacityScoring:                 feature.DefaultFeatureGate.Enabled(features.StorageCapacityScoring),
 	}
 
 	registry := runtime.Registry{

pkg/scheduler/framework/plugins/volumebinding/binder.go

@@ -102,13 +102,19 @@ func (b *BindingInfo) StorageResource() *StorageResource {
 	}
 }
 
+// DynamicProvision represents a dynamically provisioned volume.
+type DynamicProvision struct {
+	PVC          *v1.PersistentVolumeClaim
+	NodeCapacity *storagev1.CSIStorageCapacity
+}
+
 // PodVolumes holds pod's volumes information used in volume scheduling.
 type PodVolumes struct {
 	// StaticBindings are binding decisions for PVCs which can be bound to
 	// pre-provisioned static PVs.
 	StaticBindings []*BindingInfo
 	// DynamicProvisions are PVCs that require dynamic provisioning
-	DynamicProvisions []*v1.PersistentVolumeClaim
+	DynamicProvisions []*DynamicProvision
 }
 
 // InTreeToCSITranslator contains methods required to check migratable status
@@ -310,7 +316,7 @@ func (b *volumeBinder) FindPodVolumes(logger klog.Logger, pod *v1.Pod, podVolume
 
 	var (
 		staticBindings    []*BindingInfo
-		dynamicProvisions []*v1.PersistentVolumeClaim
+		dynamicProvisions []*DynamicProvision
 	)
 	defer func() {
 		// Although we do not distinguish nil from empty in this function, for
@@ -377,6 +383,16 @@ func (b *volumeBinder) FindPodVolumes(logger klog.Logger, pod *v1.Pod, podVolume
 	return
 }
 
+// ConvertDynamicProvisionsToPVCs converts a slice of *DynamicProvision to a
+// slice of PersistentVolumeClaim
+func convertDynamicProvisionsToPVCs(dynamicProvisions []*DynamicProvision) []*v1.PersistentVolumeClaim {
+	pvcs := make([]*v1.PersistentVolumeClaim, 0, len(dynamicProvisions))
+	for _, dynamicProvision := range dynamicProvisions {
+		pvcs = append(pvcs, dynamicProvision.PVC)
+	}
+	return pvcs
+}
+
 // AssumePodVolumes will take the matching PVs and PVCs to provision in pod's
 // volume information for the chosen node, and:
 // 1. Update the pvCache with the new prebound PV.
@@ -423,20 +439,21 @@ func (b *volumeBinder) AssumePodVolumes(logger klog.Logger, assumedPod *v1.Pod,
 	}
 
 	// Assume PVCs
-	newProvisionedPVCs := []*v1.PersistentVolumeClaim{}
-	for _, claim := range podVolumes.DynamicProvisions {
+	newProvisionedPVCs := []*DynamicProvision{}
+	for _, dynamicProvision := range podVolumes.DynamicProvisions {
 		// The claims from method args can be pointing to watcher cache. We must not
 		// modify these, therefore create a copy.
-		claimClone := claim.DeepCopy()
+		claimClone := dynamicProvision.PVC.DeepCopy()
 		metav1.SetMetaDataAnnotation(&claimClone.ObjectMeta, volume.AnnSelectedNode, nodeName)
 		err = b.pvcCache.Assume(claimClone)
 		if err != nil {
+			pvcs := convertDynamicProvisionsToPVCs(newProvisionedPVCs)
 			b.revertAssumedPVs(newBindings)
-			b.revertAssumedPVCs(newProvisionedPVCs)
+			b.revertAssumedPVCs(pvcs)
 			return
 		}
 
-		newProvisionedPVCs = append(newProvisionedPVCs, claimClone)
+		newProvisionedPVCs = append(newProvisionedPVCs, &DynamicProvision{PVC: claimClone})
 	}
 
 	podVolumes.StaticBindings = newBindings
@@ -446,8 +463,9 @@ func (b *volumeBinder) AssumePodVolumes(logger klog.Logger, assumedPod *v1.Pod,
 
 // RevertAssumedPodVolumes will revert assumed PV and PVC cache.
 func (b *volumeBinder) RevertAssumedPodVolumes(podVolumes *PodVolumes) {
+	pvcs := convertDynamicProvisionsToPVCs(podVolumes.DynamicProvisions)
 	b.revertAssumedPVs(podVolumes.StaticBindings)
-	b.revertAssumedPVCs(podVolumes.DynamicProvisions)
+	b.revertAssumedPVCs(pvcs)
 }
 
 // BindPodVolumes gets the cached bindings and PVCs to provision in pod's volumes information,
@@ -464,7 +482,7 @@ func (b *volumeBinder) BindPodVolumes(ctx context.Context, assumedPod *v1.Pod, p
 	}()
 
 	bindings := podVolumes.StaticBindings
-	claimsToProvision := podVolumes.DynamicProvisions
+	claimsToProvision := convertDynamicProvisionsToPVCs(podVolumes.DynamicProvisions)
 
 	// Start API operations
 	err = b.bindAPIUpdate(ctx, assumedPod, bindings, claimsToProvision)
@@ -886,8 +904,8 @@ func (b *volumeBinder) findMatchingVolumes(logger klog.Logger, pod *v1.Pod, clai
 // checkVolumeProvisions checks given unbound claims (the claims have gone through func
 // findMatchingVolumes, and do not have matching volumes for binding), and return true
 // if all of the claims are eligible for dynamic provision.
-func (b *volumeBinder) checkVolumeProvisions(logger klog.Logger, pod *v1.Pod, claimsToProvision []*v1.PersistentVolumeClaim, node *v1.Node) (provisionSatisfied, sufficientStorage bool, dynamicProvisions []*v1.PersistentVolumeClaim, err error) {
-	dynamicProvisions = []*v1.PersistentVolumeClaim{}
+func (b *volumeBinder) checkVolumeProvisions(logger klog.Logger, pod *v1.Pod, claimsToProvision []*v1.PersistentVolumeClaim, node *v1.Node) (provisionSatisfied, sufficientStorage bool, dynamicProvisions []*DynamicProvision, err error) {
+	dynamicProvisions = []*DynamicProvision{}
 
 	// We return early with provisionedClaims == nil if a check
 	// fails or we encounter an error.
@@ -915,7 +933,7 @@ func (b *volumeBinder) checkVolumeProvisions(logger klog.Logger, pod *v1.Pod, cl
 		}
 
 		// Check storage capacity.
-		sufficient, err := b.hasEnoughCapacity(logger, provisioner, claim, class, node)
+		sufficient, capacity, err := b.hasEnoughCapacity(logger, provisioner, claim, class, node)
 		if err != nil {
 			return false, false, nil, err
 		}
@@ -924,8 +942,10 @@ func (b *volumeBinder) checkVolumeProvisions(logger klog.Logger, pod *v1.Pod, cl
 			return true, false, nil, nil
 		}
 
-		dynamicProvisions = append(dynamicProvisions, claim)
-
+		dynamicProvisions = append(dynamicProvisions, &DynamicProvision{
+			PVC:          claim,
+			NodeCapacity: capacity,
+		})
 	}
 	logger.V(4).Info("Provisioning for claims of pod that has no matching volumes...", "claimCount", len(claimsToProvision), "pod", klog.KObj(pod), "node", klog.KObj(node))
 
@@ -945,12 +965,12 @@ func (b *volumeBinder) revertAssumedPVCs(claims []*v1.PersistentVolumeClaim) {
 }
 
 // hasEnoughCapacity checks whether the provisioner has enough capacity left for a new volume of the given size
-// that is available from the node.
-func (b *volumeBinder) hasEnoughCapacity(logger klog.Logger, provisioner string, claim *v1.PersistentVolumeClaim, storageClass *storagev1.StorageClass, node *v1.Node) (bool, error) {
+// that is available from the node. This function returns the node capacity based on the PVC's storage class.
+func (b *volumeBinder) hasEnoughCapacity(logger klog.Logger, provisioner string, claim *v1.PersistentVolumeClaim, storageClass *storagev1.StorageClass, node *v1.Node) (bool, *storagev1.CSIStorageCapacity, error) {
 	quantity, ok := claim.Spec.Resources.Requests[v1.ResourceStorage]
 	if !ok {
 		// No capacity to check for.
-		return true, nil
+		return true, nil, nil
 	}
 
 	// Only enabled for CSI drivers which opt into it.
@@ -960,19 +980,19 @@ func (b *volumeBinder) hasEnoughCapacity(logger klog.Logger, provisioner string,
 			// Either the provisioner is not a CSI driver or the driver does not
 			// opt into storage capacity scheduling. Either way, skip
 			// capacity checking.
-			return true, nil
+			return true, nil, nil
 		}
-		return false, err
+		return false, nil, err
 	}
 	if driver.Spec.StorageCapacity == nil || !*driver.Spec.StorageCapacity {
-		return true, nil
+		return true, nil, nil
 	}
 
 	// Look for a matching CSIStorageCapacity object(s).
 	// TODO (for beta): benchmark this and potentially introduce some kind of lookup structure (https://github.com/kubernetes/enhancements/issues/1698#issuecomment-654356718).
 	capacities, err := b.csiStorageCapacityLister.List(labels.Everything())
 	if err != nil {
-		return false, err
+		return false, nil, err
 	}
 
 	sizeInBytes := quantity.Value()
@@ -981,15 +1001,15 @@ func (b *volumeBinder) hasEnoughCapacity(logger klog.Logger, provisioner string,
 			capacitySufficient(capacity, sizeInBytes) &&
 			b.nodeHasAccess(logger, node, capacity) {
 			// Enough capacity found.
-			return true, nil
+			return true, capacity, nil
 		}
 	}
 
 	// TODO (?): this doesn't give any information about which pools where considered and why
 	// they had to be rejected. Log that above? But that might be a lot of log output...
 	logger.V(4).Info("Node has no accessible CSIStorageCapacity with enough capacity for PVC",
 		"node", klog.KObj(node), "PVC", klog.KObj(claim), "size", sizeInBytes, "storageClass", klog.KObj(storageClass))
-	return false, nil
+	return false, nil, nil
 }
 
 func capacitySufficient(capacity *storagev1.CSIStorageCapacity, sizeInBytes int64) bool {