Implement scheduler changes for volume limits

pkg/scheduler/algorithm/predicates/BUILD

@@ -46,6 +46,7 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "max_attachable_volume_predicate_test.go",
         "metadata_test.go",
         "predicates_test.go",
         "utils_test.go",
@@ -53,10 +54,12 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//pkg/apis/core/v1/helper:go_default_library",
+        "//pkg/features:go_default_library",
         "//pkg/kubelet/apis:go_default_library",
         "//pkg/scheduler/algorithm:go_default_library",
         "//pkg/scheduler/cache:go_default_library",
         "//pkg/scheduler/testing:go_default_library",
+        "//pkg/volume/util:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/storage/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
@@ -64,6 +67,7 @@ go_test(
         "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/util/feature/testing:go_default_library",
     ],
 )
 

pkg/scheduler/algorithm/predicates/predicates.go

@@ -289,10 +289,11 @@ func NoDiskConflict(pod *v1.Pod, meta algorithm.PredicateMetadata, nodeInfo *sch
 
 // MaxPDVolumeCountChecker contains information to check the max number of volumes for a predicate.
 type MaxPDVolumeCountChecker struct {
-	filter     VolumeFilter
-	maxVolumes int
-	pvInfo     PersistentVolumeInfo
-	pvcInfo    PersistentVolumeClaimInfo
+	filter         VolumeFilter
+	volumeLimitKey v1.ResourceName
+	maxVolumes     int
+	pvInfo         PersistentVolumeInfo
+	pvcInfo        PersistentVolumeClaimInfo
 
 	// The string below is generated randomly during the struct's initialization.
 	// It is used to prefix volumeID generated inside the predicate() method to
@@ -313,21 +314,25 @@ type VolumeFilter struct {
 // The predicate looks for both volumes used directly, as well as PVC volumes that are backed by relevant volume
 // types, counts the number of unique volumes, and rejects the new pod if it would place the total count over
 // the maximum.
-func NewMaxPDVolumeCountPredicate(filterName string, pvInfo PersistentVolumeInfo, pvcInfo PersistentVolumeClaimInfo) algorithm.FitPredicate {
-
+func NewMaxPDVolumeCountPredicate(
+	filterName string, pvInfo PersistentVolumeInfo, pvcInfo PersistentVolumeClaimInfo) algorithm.FitPredicate {
 	var filter VolumeFilter
 	var maxVolumes int
+	var volumeLimitKey v1.ResourceName
 
 	switch filterName {
 
 	case EBSVolumeFilterType:
 		filter = EBSVolumeFilter
+		volumeLimitKey = v1.ResourceName(volumeutil.EBSVolumeLimitKey)
 		maxVolumes = getMaxVols(DefaultMaxEBSVolumes)
 	case GCEPDVolumeFilterType:
 		filter = GCEPDVolumeFilter
+		volumeLimitKey = v1.ResourceName(volumeutil.GCEVolumeLimitKey)
 		maxVolumes = getMaxVols(DefaultMaxGCEPDVolumes)
 	case AzureDiskVolumeFilterType:
 		filter = AzureDiskVolumeFilter
+		volumeLimitKey = v1.ResourceName(volumeutil.AzureVolumeLimitKey)
 		maxVolumes = getMaxVols(DefaultMaxAzureDiskVolumes)
 	default:
 		glog.Fatalf("Wrong filterName, Only Support %v %v %v ", EBSVolumeFilterType,
@@ -337,6 +342,7 @@ func NewMaxPDVolumeCountPredicate(filterName string, pvInfo PersistentVolumeInfo
 	}
 	c := &MaxPDVolumeCountChecker{
 		filter:               filter,
+		volumeLimitKey:       volumeLimitKey,
 		maxVolumes:           maxVolumes,
 		pvInfo:               pvInfo,
 		pvcInfo:              pvcInfo,
@@ -362,7 +368,6 @@ func getMaxVols(defaultVal int) int {
 }
 
 func (c *MaxPDVolumeCountChecker) filterVolumes(volumes []v1.Volume, namespace string, filteredVolumes map[string]bool) error {
-
 	for i := range volumes {
 		vol := &volumes[i]
 		if id, ok := c.filter.FilterVolume(vol); ok {
@@ -449,15 +454,23 @@ func (c *MaxPDVolumeCountChecker) predicate(pod *v1.Pod, meta algorithm.Predicat
 	}
 
 	numNewVolumes := len(newVolumes)
+	maxAttachLimit := c.maxVolumes
+
+	if utilfeature.DefaultFeatureGate.Enabled(features.AttachVolumeLimit) {
+		volumeLimits := nodeInfo.VolumeLimits()
+		if maxAttachLimitFromAllocatable, ok := volumeLimits[c.volumeLimitKey]; ok {
+			maxAttachLimit = int(maxAttachLimitFromAllocatable)
+		}
+	}
 
-	if numExistingVolumes+numNewVolumes > c.maxVolumes {
+	if numExistingVolumes+numNewVolumes > maxAttachLimit {
 		// violates MaxEBSVolumeCount or MaxGCEPDVolumeCount
 		return false, []algorithm.PredicateFailureReason{ErrMaxVolumeCountExceeded}, nil
 	}
 	if nodeInfo != nil && nodeInfo.TransientInfo != nil && utilfeature.DefaultFeatureGate.Enabled(features.BalanceAttachedNodeVolumes) {
 		nodeInfo.TransientInfo.TransientLock.Lock()
 		defer nodeInfo.TransientInfo.TransientLock.Unlock()
-		nodeInfo.TransientInfo.TransNodeInfo.AllocatableVolumesCount = c.maxVolumes - numExistingVolumes
+		nodeInfo.TransientInfo.TransNodeInfo.AllocatableVolumesCount = maxAttachLimit - numExistingVolumes
 		nodeInfo.TransientInfo.TransNodeInfo.RequestedVolumes = numNewVolumes
 	}
 	return true, nil, nil

pkg/scheduler/cache/node_info.go

@@ -412,6 +412,17 @@ func (n *NodeInfo) Clone() *NodeInfo {
 	return clone
 }
 
+// VolumeLimits returns volume limits associated with the node
+func (n *NodeInfo) VolumeLimits() map[v1.ResourceName]int64 {
+	volumeLimits := map[v1.ResourceName]int64{}
+	for k, v := range n.AllocatableResource().ScalarResources {
+		if v1helper.IsAttachableVolumeResourceName(k) {
+			volumeLimits[k] = v
+		}
+	}
+	return volumeLimits
+}
+
 // String returns representation of human readable format of this NodeInfo.
 func (n *NodeInfo) String() string {
 	podKeys := make([]string, len(n.pods))

pkg/scheduler/cache/node_info_test.go

@@ -84,7 +84,11 @@ func TestResourceList(t *testing.T) {
 				Memory:           2000,
 				EphemeralStorage: 5000,
 				AllowedPodNumber: 80,
-				ScalarResources:  map[v1.ResourceName]int64{"scalar.test/scalar1": 1, "hugepages-test": 2},
+				ScalarResources: map[v1.ResourceName]int64{
+					"scalar.test/scalar1":        1,
+					"hugepages-test":             2,
+					"attachable-volumes-aws-ebs": 39,
+				},
 			},
 			expected: map[v1.ResourceName]resource.Quantity{
 				v1.ResourceCPU:                      *resource.NewScaledQuantity(4, -3),
@@ -92,6 +96,7 @@ func TestResourceList(t *testing.T) {
 				v1.ResourcePods:                     *resource.NewQuantity(80, resource.BinarySI),
 				v1.ResourceEphemeralStorage:         *resource.NewQuantity(5000, resource.BinarySI),
 				"scalar.test/" + "scalar1":          *resource.NewQuantity(1, resource.DecimalSI),
+				"attachable-volumes-aws-ebs":        *resource.NewQuantity(39, resource.DecimalSI),
 				v1.ResourceHugePagesPrefix + "test": *resource.NewQuantity(2, resource.BinarySI),
 			},
 		},