Topology scoring with dynamically weighted topologies

Signed-off-by: Aldo Culquicondor <[email protected]>

Author: alculquicondor

pkg/scheduler/framework/plugins/podtopologyspread/scoring.go

@@ -39,6 +39,10 @@ type preScoreState struct {
 	IgnoredNodes sets.String
 	// TopologyPairToPodCounts is keyed with topologyPair, and valued with the number of matching pods.
 	TopologyPairToPodCounts map[topologyPair]*int64
+	// TopologyNormalizingWeight is the weight we give to the counts per topology.
+	// This allows the pod counts of smaller topologies to not be watered down by
+	// bigger ones.
+	TopologyNormalizingWeight []float64
 }
 
 // Clone implements the mandatory Clone interface. We don't really copy the data since
@@ -51,6 +55,7 @@ func (s *preScoreState) Clone() framework.StateData {
 // don't have required topologyKey(s), and initialize:
 // 1) s.TopologyPairToPodCounts: keyed with both eligible topology pair and node names.
 // 2) s.IgnoredNodes: the set of nodes that shouldn't be scored.
+// 3) s.TopologyNormalizingWeight: The weight to be given to each constraint based on the number of values in a topology.
 func (pl *PodTopologySpread) initPreScoreState(s *preScoreState, pod *v1.Pod, filteredNodes []*v1.Node) error {
 	var err error
 	if len(pod.Spec.TopologySpreadConstraints) > 0 {
@@ -67,24 +72,35 @@ func (pl *PodTopologySpread) initPreScoreState(s *preScoreState, pod *v1.Pod, fi
 	if len(s.Constraints) == 0 {
 		return nil
 	}
+	topoSize := make([]int, len(s.Constraints))
 	for _, node := range filteredNodes {
 		if !nodeLabelsMatchSpreadConstraints(node.Labels, s.Constraints) {
 			// Nodes which don't have all required topologyKeys present are ignored
 			// when scoring later.
 			s.IgnoredNodes.Insert(node.Name)
 			continue
 		}
-		for _, constraint := range s.Constraints {
+		for i, constraint := range s.Constraints {
 			// per-node counts are calculated during Score.
 			if constraint.TopologyKey == v1.LabelHostname {
 				continue
 			}
 			pair := topologyPair{key: constraint.TopologyKey, value: node.Labels[constraint.TopologyKey]}
 			if s.TopologyPairToPodCounts[pair] == nil {
 				s.TopologyPairToPodCounts[pair] = new(int64)
+				topoSize[i]++
 			}
 		}
 	}
+
+	s.TopologyNormalizingWeight = make([]float64, len(s.Constraints))
+	for i, c := range s.Constraints {
+		sz := topoSize[i]
+		if c.TopologyKey == v1.LabelHostname {
+			sz = len(filteredNodes) - len(s.IgnoredNodes)
+		}
+		s.TopologyNormalizingWeight[i] = topologyNormalizingWeight(sz)
+	}
 	return nil
 }
 
@@ -174,20 +190,20 @@ func (pl *PodTopologySpread) Score(ctx context.Context, cycleState *framework.Cy
 
 	// For each present <pair>, current node gets a credit of <matchSum>.
 	// And we sum up <matchSum> and return it as this node's score.
-	var score int64
-	for _, c := range s.Constraints {
+	var score float64
+	for i, c := range s.Constraints {
 		if tpVal, ok := node.Labels[c.TopologyKey]; ok {
+			var cnt int64
 			if c.TopologyKey == v1.LabelHostname {
-				count := countPodsMatchSelector(nodeInfo.Pods, c.Selector, pod.Namespace)
-				score += int64(count)
+				cnt = int64(countPodsMatchSelector(nodeInfo.Pods, c.Selector, pod.Namespace))
 			} else {
 				pair := topologyPair{key: c.TopologyKey, value: tpVal}
-				matchSum := *s.TopologyPairToPodCounts[pair]
-				score += matchSum
+				cnt = *s.TopologyPairToPodCounts[pair]
 			}
+			score += float64(cnt) * s.TopologyNormalizingWeight[i]
 		}
 	}
-	return score, nil
+	return int64(score), nil
 }
 
 // NormalizeScore invoked after scoring all nodes.
@@ -200,41 +216,41 @@ func (pl *PodTopologySpread) NormalizeScore(ctx context.Context, cycleState *fra
 		return nil
 	}
 
-	// Calculate the summed <total> score and <minScore>.
+	// Calculate <minScore> and <maxScore>
 	var minScore int64 = math.MaxInt64
-	var total int64
+	var maxScore int64
 	for _, score := range scores {
 		// it's mandatory to check if <score.Name> is present in m.IgnoredNodes
 		if s.IgnoredNodes.Has(score.Name) {
 			continue
 		}
-		total += score.Score
 		if score.Score < minScore {
 			minScore = score.Score
 		}
+		if score.Score > maxScore {
+			maxScore = score.Score
+		}
 	}
 
-	maxMinDiff := total - minScore
 	for i := range scores {
 		nodeInfo, err := pl.sharedLister.NodeInfos().Get(scores[i].Name)
 		if err != nil {
 			return framework.NewStatus(framework.Error, err.Error())
 		}
 		node := nodeInfo.Node()
 
-		if maxMinDiff == 0 {
-			scores[i].Score = framework.MaxNodeScore
+		if s.IgnoredNodes.Has(node.Name) {
+			scores[i].Score = 0
 			continue
 		}
 
-		if s.IgnoredNodes.Has(node.Name) {
-			scores[i].Score = 0
+		if maxScore == 0 {
+			scores[i].Score = framework.MaxNodeScore
 			continue
 		}
 
-		flippedScore := total - scores[i].Score
-		fScore := float64(framework.MaxNodeScore) * (float64(flippedScore) / float64(maxMinDiff))
-		scores[i].Score = int64(fScore)
+		s := scores[i].Score
+		scores[i].Score = framework.MaxNodeScore * (maxScore + minScore - s) / maxScore
 	}
 	return nil
 }
@@ -256,3 +272,16 @@ func getPreScoreState(cycleState *framework.CycleState) (*preScoreState, error)
 	}
 	return s, nil
 }
+
+// topologyNormalizingWeight calculates the weight for the topology, based on
+// the number of values that exist for a topology.
+// Since <size> is at least 1 (all nodes that passed the Filters are in the
+// same topology), and k8s supports 5k nodes, the result is in the interval
+// <1.09, 8.52>.
+//
+// Note: <size> could also be zero when no nodes have the required topologies,
+// however we don't care about topology weight in this case as we return a 0
+// score for all nodes.
+func topologyNormalizingWeight(size int) float64 {
+	return math.Log(float64(size + 2))
+}