Merge pull request #8181 from MenD32/fix/topology-spread-binpacking

fix: binpacking simulator scale up optimization on pods with topology…

Author: k8s-ci-robot

cluster-autoscaler/estimator/binpacking_estimator.go

@@ -20,13 +20,16 @@ import (
 	"fmt"
 	"strconv"
 
+	"slices"
+
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
 	"k8s.io/autoscaler/cluster-autoscaler/metrics"
 	core_utils "k8s.io/autoscaler/cluster-autoscaler/simulator"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/clustersnapshot"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/framework"
 	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/podtopologyspread"
 )
 
 // BinpackingNodeEstimator estimates the number of needed nodes to handle the given amount of pods.
@@ -171,7 +174,8 @@ func (e *BinpackingNodeEstimator) tryToScheduleOnNewNodes(
 
 		if estimationState.lastNodeName != "" {
 			// Try to schedule the pod on only newly created node.
-			if err := e.clusterSnapshot.SchedulePod(pod, estimationState.lastNodeName); err == nil {
+			err := e.clusterSnapshot.SchedulePod(pod, estimationState.lastNodeName)
+			if err == nil {
 				// The pod was scheduled on the newly created node.
 				found = true
 				estimationState.trackScheduledPod(pod, estimationState.lastNodeName)
@@ -180,6 +184,24 @@ func (e *BinpackingNodeEstimator) tryToScheduleOnNewNodes(
 				return false, err
 			}
 			// The pod can't be scheduled on the newly created node because of scheduling predicates.
+
+			// Check if node failed because of topology constraints.
+			if isPodUsingHostNameTopologyKey(pod) && hasTopologyConstraintError(err) {
+				// If the pod can't be scheduled on the last node because of topology constraints, we can stop binpacking.
+				// The pod can't be scheduled on any new node either, because it has the same topology constraints.
+				nodeName, err := e.clusterSnapshot.SchedulePodOnAnyNodeMatching(pod, func(nodeInfo *framework.NodeInfo) bool {
+					return nodeInfo.Node().Name != estimationState.lastNodeName // only skip the last node that failed scheduling
+				})
+				if err != nil && err.Type() == clustersnapshot.SchedulingInternalError {
+					// Unexpected error.
+					return false, err
+				}
+				if nodeName != "" {
+					// The pod was scheduled on a different node, so we can continue binpacking.
+					found = true
+					estimationState.trackScheduledPod(pod, nodeName)
+				}
+			}
 		}
 
 		if !found {
@@ -240,6 +262,33 @@ func (e *BinpackingNodeEstimator) addNewNodeToSnapshot(
 	return nil
 }
 
+// isTopologyConstraintError determines if an error is related to pod topology spread constraints
+// by checking the predicate name and reasons
+func hasTopologyConstraintError(err clustersnapshot.SchedulingError) bool {
+	if err == nil {
+		return false
+	}
+
+	// Check reasons for mentions of topology or constraints
+	return slices.Contains(err.FailingPredicateReasons(), podtopologyspread.ErrReasonConstraintsNotMatch)
+}
+
+// isPodUsingHostNameTopoKey returns true if the pod has any topology spread
+// constraint that uses the kubernetes.io/hostname topology key
+func isPodUsingHostNameTopologyKey(pod *apiv1.Pod) bool {
+	if pod == nil || pod.Spec.TopologySpreadConstraints == nil {
+		return false
+	}
+
+	for _, constraint := range pod.Spec.TopologySpreadConstraints {
+		if constraint.TopologyKey == apiv1.LabelHostname {
+			return true
+		}
+	}
+
+	return false
+}
+
 func observeBinpackingHeterogeneity(podsEquivalenceGroups []PodEquivalenceGroup, nodeTemplate *framework.NodeInfo) {
 	node := nodeTemplate.Node()
 	var instanceType, cpuCount string

cluster-autoscaler/estimator/binpacking_estimator_test.go

@@ -178,13 +178,13 @@ func TestBinpackingEstimate(t *testing.T) {
 			podsEquivalenceGroup: []PodEquivalenceGroup{makePodEquivalenceGroup(
 				BuildTestPod(
 					"estimatee",
-					20,
-					100,
+					200,
+					200,
 					WithNamespace("universe"),
 					WithLabels(map[string]string{
 						"app": "estimatee",
 					}),
-					WithMaxSkew(2, "kubernetes.io/hostname")), 8)},
+					WithMaxSkew(2, "kubernetes.io/hostname", 1)), 8)},
 			expectNodeCount: 4,
 			expectPodCount:  8,
 		},
@@ -201,10 +201,27 @@ func TestBinpackingEstimate(t *testing.T) {
 					WithLabels(map[string]string{
 						"app": "estimatee",
 					}),
-					WithMaxSkew(2, "topology.kubernetes.io/zone")), 8)},
+					WithMaxSkew(2, "topology.kubernetes.io/zone", 1)), 8)},
 			expectNodeCount: 1,
 			expectPodCount:  2,
 		},
+		{
+			name:       "hostname topology spreading with maxSkew=1 with a large scaleup handles scheduling pods retroactively",
+			millicores: 1000,
+			memory:     5000,
+			podsEquivalenceGroup: []PodEquivalenceGroup{makePodEquivalenceGroup(
+				BuildTestPod(
+					"estimatee",
+					20,
+					100,
+					WithNamespace("universe"),
+					WithLabels(map[string]string{
+						"app": "estimatee",
+					}),
+					WithMaxSkew(1, "kubernetes.io/hostname", 3)), 12)},
+			expectNodeCount: 3,
+			expectPodCount:  12,
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {

cluster-autoscaler/utils/test/test_utils.go

@@ -160,8 +160,8 @@ func WithHostPort(hostport int32) func(*apiv1.Pod) {
 	}
 }
 
-// WithMaxSkew sets a namespace to the pod.
-func WithMaxSkew(maxSkew int32, topologySpreadingKey string) func(*apiv1.Pod) {
+// WithMaxSkew sets a topology spread constraint to the pod.
+func WithMaxSkew(maxSkew int32, topologySpreadingKey string, minDomains int32) func(*apiv1.Pod) {
 	return func(pod *apiv1.Pod) {
 		if maxSkew > 0 {
 			pod.Spec.TopologySpreadConstraints = []apiv1.TopologySpreadConstraint{
@@ -174,6 +174,7 @@ func WithMaxSkew(maxSkew int32, topologySpreadingKey string) func(*apiv1.Pod) {
 							"app": "estimatee",
 						},
 					},
+					MinDomains: &minDomains,
 				},
 			}
 		}