Aggregate mulitple NodePreference custom priorities to a single score plugin.

Author: liu-cong

pkg/scheduler/algorithm/priorities/BUILD

@@ -64,7 +64,6 @@ go_test(
         "metadata_test.go",
         "most_requested_test.go",
         "node_affinity_test.go",
-        "node_label_test.go",
         "node_prefer_avoid_pods_test.go",
         "requested_to_capacity_ratio_test.go",
         "resource_limits_test.go",

pkg/scheduler/algorithm/priorities/node_label.go

@@ -27,15 +27,15 @@ import (
 
 // NodeLabelPrioritizer contains information to calculate node label priority.
 type NodeLabelPrioritizer struct {
-	label    string
-	presence bool
+	presentLabelsPreference []string
+	absentLabelsPreference  []string
 }
 
 // NewNodeLabelPriority creates a NodeLabelPrioritizer.
-func NewNodeLabelPriority(label string, presence bool) (PriorityMapFunction, PriorityReduceFunction) {
+func NewNodeLabelPriority(presentLabelsPreference []string, absentLabelsPreference []string) (PriorityMapFunction, PriorityReduceFunction) {
 	labelPrioritizer := &NodeLabelPrioritizer{
-		label:    label,
-		presence: presence,
+		presentLabelsPreference: presentLabelsPreference,
+		absentLabelsPreference:  absentLabelsPreference,
 	}
 	return labelPrioritizer.CalculateNodeLabelPriorityMap, nil
 }
@@ -49,11 +49,20 @@ func (n *NodeLabelPrioritizer) CalculateNodeLabelPriorityMap(pod *v1.Pod, meta i
 		return framework.NodeScore{}, fmt.Errorf("node not found")
 	}
 
-	exists := labels.Set(node.Labels).Has(n.label)
 	score := int64(0)
-	if (exists && n.presence) || (!exists && !n.presence) {
-		score = framework.MaxNodeScore
+	for _, label := range n.presentLabelsPreference {
+		if labels.Set(node.Labels).Has(label) {
+			score += framework.MaxNodeScore
+		}
 	}
+	for _, label := range n.absentLabelsPreference {
+		if !labels.Set(node.Labels).Has(label) {
+			score += framework.MaxNodeScore
+		}
+	}
+	// Take average score for each label to ensure the score doesn't exceed MaxNodeScore.
+	score /= int64(len(n.presentLabelsPreference) + len(n.absentLabelsPreference))
+
 	return framework.NodeScore{
 		Name:  node.Name,
 		Score: score,

pkg/scheduler/algorithm/priorities/node_label_test.go

@@ -400,15 +400,35 @@ func RegisterCustomPriorityFunction(policy schedulerapi.PriorityPolicy, args *pl
 				Weight: policy.Weight,
 			}
 		} else if policy.Argument.LabelPreference != nil {
+			// We use the NodeLabel plugin name for all NodeLabel custom priorities.
+			// It may get called multiple times but we essentially only register one instance of NodeLabel priority.
+			// This name is then used to find the registered plugin and run the plugin instead of the priority.
+			name = nodelabel.Name
+			if args.NodeLabelArgs == nil {
+				args.NodeLabelArgs = &nodelabel.Args{}
+			}
+			if policy.Argument.LabelPreference.Presence {
+				args.NodeLabelArgs.PresentLabelsPreference = append(args.NodeLabelArgs.PresentLabelsPreference, policy.Argument.LabelPreference.Label)
+			} else {
+				args.NodeLabelArgs.AbsentLabelsPreference = append(args.NodeLabelArgs.AbsentLabelsPreference, policy.Argument.LabelPreference.Label)
+			}
+			schedulerFactoryMutex.RLock()
+			weight := policy.Weight
+			if existing, ok := priorityFunctionMap[name]; ok {
+				// If there are n NodeLabel priority configured in the policy, the weight for the corresponding
+				// priority is n*(weight of each priority in policy).
+				weight += existing.Weight
+			}
 			pcf = &PriorityConfigFactory{
-				MapReduceFunction: func(args PluginFactoryArgs) (priorities.PriorityMapFunction, priorities.PriorityReduceFunction) {
+				MapReduceFunction: func(_ PluginFactoryArgs) (priorities.PriorityMapFunction, priorities.PriorityReduceFunction) {
 					return priorities.NewNodeLabelPriority(
-						policy.Argument.LabelPreference.Label,
-						policy.Argument.LabelPreference.Presence,
+						args.NodeLabelArgs.PresentLabelsPreference,
+						args.NodeLabelArgs.AbsentLabelsPreference,
 					)
 				},
-				Weight: policy.Weight,
+				Weight: weight,
 			}
+			schedulerFactoryMutex.RUnlock()
 		} else if policy.Argument.RequestedToCapacityRatioArguments != nil {
 			scoringFunctionShape, resources := buildScoringFunctionShapeFromRequestedToCapacityRatioArguments(policy.Argument.RequestedToCapacityRatioArguments)
 			args.RequestedToCapacityRatioArgs = &requestedtocapacityratio.Args{

pkg/scheduler/algorithm_factory.go

@@ -103,7 +103,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 			wantPrioritizers: sets.NewString(
 				"ServiceSpreadingPriority",
 				"TestServiceAntiAffinity",
-				"TestLabelPreference",
 			),
 			wantPlugins: map[string][]config.Plugin{
 				"FilterPlugin": {
@@ -116,6 +115,7 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 				},
 				"ScorePlugin": {
 					{Name: "NodeResourcesLeastAllocated", Weight: 1},
+					{Name: "NodeLabel", Weight: 4},
 				},
 			},
 		},
@@ -151,7 +151,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 				"EqualPriority",
 				"SelectorSpreadPriority",
 				"TestServiceAntiAffinity",
-				"TestLabelPreference",
 			),
 			wantPlugins: map[string][]config.Plugin{
 				"FilterPlugin": {
@@ -167,6 +166,7 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 				"ScorePlugin": {
 					{Name: "NodeResourcesBalancedAllocation", Weight: 2},
 					{Name: "NodeResourcesLeastAllocated", Weight: 2},
+					{Name: "NodeLabel", Weight: 4},
 				},
 			},
 		},
@@ -207,7 +207,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 				"EqualPriority",
 				"SelectorSpreadPriority",
 				"TestServiceAntiAffinity",
-				"TestLabelPreference",
 			),
 			wantPlugins: map[string][]config.Plugin{
 				"FilterPlugin": {
@@ -229,6 +228,7 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 					{Name: "ImageLocality", Weight: 2},
 					{Name: "NodeResourcesLeastAllocated", Weight: 2},
 					{Name: "NodeAffinity", Weight: 2},
+					{Name: "NodeLabel", Weight: 4},
 				},
 			},
 		},
@@ -1263,6 +1263,7 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 		"NodeResourcesBalancedAllocation": "BalancedResourceAllocation",
 		"NodeResourcesMostAllocated":      "MostRequestedPriority",
 		"RequestedToCapacityRatio":        "RequestedToCapacityRatioPriority",
+		"NodeLabel":                       "TestLabelPreference",
 	}
 
 	for _, tc := range testcases {

pkg/scheduler/apis/config/testing/compatibility_test.go

@@ -98,6 +98,8 @@ func TestCreateFromConfig(t *testing.T) {
 		],
 		"priorities" : [
 			{"name" : "RackSpread", "weight" : 3, "argument" : {"serviceAntiAffinity" : {"label" : "rack"}}},
+			{"name" : "LabelPreference1", "weight" : 3, "argument" : {"labelPreference" : {"label" : "l1", "presence": true}}},
+			{"name" : "LabelPreference2", "weight" : 3, "argument" : {"labelPreference" : {"label" : "l2", "presence": false}}},
 			{"name" : "PriorityOne", "weight" : 2},
 			{"name" : "PriorityTwo", "weight" : 1}		]
 	}`)
@@ -114,29 +116,36 @@ func TestCreateFromConfig(t *testing.T) {
 		t.Errorf("Wrong hardPodAffinitySymmetricWeight, ecpected: %d, got: %d", v1.DefaultHardPodAffinitySymmetricWeight, hpa)
 	}
 
-	// Verify that custom predicates are converted to framework plugins.
-	if !pluginExists(nodelabel.Name, "FilterPlugin", conf) {
-		t.Error("NodeLabel plugin not exist in framework.")
+	// Verify that node label predicate/priority are converted to framework plugins.
+	if _, ok := findPlugin(nodelabel.Name, "FilterPlugin", conf); !ok {
+		t.Fatalf("NodeLabel plugin not exist in framework.")
 	}
-	// Verify that the policy config is converted to plugin config for custom predicates.
+	nodeLabelScorePlugin, ok := findPlugin(nodelabel.Name, "ScorePlugin", conf)
+	if !ok {
+		t.Fatalf("NodeLabel plugin not exist in framework.")
+	}
+	if nodeLabelScorePlugin.Weight != 6 {
+		t.Errorf("Wrong weight. Got: %v, want: 6", nodeLabelScorePlugin.Weight)
+	}
+	// Verify that the policy config is converted to plugin config for node label predicate/priority.
 	nodeLabelConfig := findPluginConfig(nodelabel.Name, conf)
 	encoding, err := json.Marshal(nodeLabelConfig)
 	if err != nil {
 		t.Errorf("Failed to marshal %+v: %v", nodeLabelConfig, err)
 	}
-	want := `{"Name":"NodeLabel","Args":{"presentLabels":["zone"],"absentLabels":["foo"]}}`
+	want := `{"Name":"NodeLabel","Args":{"presentLabels":["zone"],"absentLabels":["foo"],"presentLabelsPreference":["l1"],"absentLabelsPreference":["l2"]}}`
 	if string(encoding) != want {
 		t.Errorf("Config for NodeLabel plugin mismatch. got: %v, want: %v", string(encoding), want)
 	}
 }
 
-func pluginExists(name, extensionPoint string, schedConf *Config) bool {
+func findPlugin(name, extensionPoint string, schedConf *Config) (schedulerapi.Plugin, bool) {
 	for _, pl := range schedConf.Framework.ListPlugins()[extensionPoint] {
 		if pl.Name == name {
-			return true
+			return pl, true
 		}
 	}
-	return false
+	return schedulerapi.Plugin{}, false
 }
 
 func findPluginConfig(name string, schedConf *Config) schedulerapi.PluginConfig {

pkg/scheduler/factory_test.go

@@ -256,6 +256,12 @@ func NewDefaultConfigProducerRegistry() *ConfigProducerRegistry {
 			return
 		})
 
+	registry.RegisterPriority(nodelabel.Name,
+		func(args ConfigProducerArgs) (plugins config.Plugins, pluginConfig []config.PluginConfig) {
+			plugins.Score = appendToPluginSet(plugins.Score, nodelabel.Name, &args.Weight)
+			pluginConfig = append(pluginConfig, makePluginConfig(nodelabel.Name, args.NodeLabelArgs))
+			return
+		})
 	return registry
 }
 

pkg/scheduler/framework/plugins/default_registry.go

@@ -7,6 +7,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/scheduler/algorithm/predicates:go_default_library",
+        "//pkg/scheduler/algorithm/priorities:go_default_library",
         "//pkg/scheduler/framework/plugins/migration:go_default_library",
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//pkg/scheduler/nodeinfo:go_default_library",
@@ -22,6 +23,7 @@ go_test(
     deps = [
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//pkg/scheduler/nodeinfo:go_default_library",
+        "//pkg/scheduler/nodeinfo/snapshot:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",