Merge pull request #84948 from zouyee/PriorityFunction

clean up PriorityFunction

pkg/scheduler/algorithm/priorities/balanced_resource_allocation_test.go

@@ -409,9 +409,8 @@ func TestBalancedResourceAllocation(t *testing.T) {
 					info.TransientInfo.TransNodeInfo.RequestedVolumes = len(test.pod.Spec.Volumes)
 				}
 			}
-			function := priorityFunction(BalancedResourceAllocationMap, nil, nil)
 
-			list, err := function(test.pod, snapshot, test.nodes)
+			list, err := runMapReducePriority(BalancedResourceAllocationMap, nil, nil, test.pod, snapshot, test.nodes)
 
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)

pkg/scheduler/algorithm/priorities/image_locality_test.go

@@ -185,7 +185,7 @@ func TestImageLocalityPriority(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			snapshot := nodeinfosnapshot.NewSnapshot(test.pods, test.nodes)
-			list, err := priorityFunction(ImageLocalityPriorityMap, nil, &priorityMetadata{totalNumNodes: len(test.nodes)})(test.pod, snapshot, test.nodes)
+			list, err := runMapReducePriority(ImageLocalityPriorityMap, nil, &priorityMetadata{totalNumNodes: len(test.nodes)}, test.pod, snapshot, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}

pkg/scheduler/algorithm/priorities/least_requested_test.go

@@ -254,7 +254,7 @@ func TestLeastRequested(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			snapshot := nodeinfosnapshot.NewSnapshot(test.pods, test.nodes)
-			list, err := priorityFunction(LeastRequestedPriorityMap, nil, nil)(test.pod, snapshot, test.nodes)
+			list, err := runMapReducePriority(LeastRequestedPriorityMap, nil, nil, test.pod, snapshot, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}

pkg/scheduler/algorithm/priorities/most_requested_test.go

@@ -211,7 +211,7 @@ func TestMostRequested(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			snapshot := nodeinfosnapshot.NewSnapshot(test.pods, test.nodes)
-			list, err := priorityFunction(MostRequestedPriorityMap, nil, nil)(test.pod, snapshot, test.nodes)
+			list, err := runMapReducePriority(MostRequestedPriorityMap, nil, nil, test.pod, snapshot, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}

pkg/scheduler/algorithm/priorities/node_affinity_test.go

@@ -168,8 +168,7 @@ func TestNodeAffinityPriority(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			snapshot := nodeinfosnapshot.NewSnapshot(nil, test.nodes)
-			nap := priorityFunction(CalculateNodeAffinityPriorityMap, CalculateNodeAffinityPriorityReduce, nil)
+			list, err := runMapReducePriority(CalculateNodeAffinityPriorityMap, CalculateNodeAffinityPriorityReduce, nil, test.pod, snapshot, test.nodes)
-			list, err := nap(test.pod, snapshot, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}

pkg/scheduler/algorithm/priorities/node_prefer_avoid_pods_test.go

@@ -142,7 +142,7 @@ func TestNodePreferAvoidPriority(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			snapshot := nodeinfosnapshot.NewSnapshot(nil, test.nodes)
-			list, err := priorityFunction(CalculateNodePreferAvoidPodsPriorityMap, nil, nil)(test.pod, snapshot, test.nodes)
+			list, err := runMapReducePriority(CalculateNodePreferAvoidPodsPriorityMap, nil, nil, test.pod, snapshot, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}

pkg/scheduler/algorithm/priorities/requested_to_capacity_ratio_test.go

@@ -241,7 +241,7 @@ func TestRequestedToCapacityRatio(t *testing.T) {
 		newPod := buildResourcesPod("", test.requested)
 
 		snapshot := nodeinfosnapshot.NewSnapshot(scheduledPods, nodes)
-		list, err := priorityFunction(RequestedToCapacityRatioResourceAllocationPriorityDefault().PriorityMap, nil, nil)(newPod, snapshot, nodes)
+		list, err := runMapReducePriority(RequestedToCapacityRatioResourceAllocationPriorityDefault().PriorityMap, nil, nil, newPod, snapshot, nodes)
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
 		}
@@ -390,7 +390,7 @@ func TestResourceBinPackingSingleExtended(t *testing.T) {
 			functionShape, _ := NewFunctionShape([]FunctionShapePoint{{0, 0}, {100, 10}})
 			resourceToWeightMap := ResourceToWeightMap{v1.ResourceName("intel.com/foo"): 1}
 			prior := RequestedToCapacityRatioResourceAllocationPriority(functionShape, resourceToWeightMap)
-			list, err := priorityFunction(prior.PriorityMap, nil, nil)(test.pod, snapshot, test.nodes)
+			list, err := runMapReducePriority(prior.PriorityMap, nil, nil, test.pod, snapshot, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}
@@ -615,7 +615,7 @@ func TestResourceBinPackingMultipleExtended(t *testing.T) {
 			functionShape, _ := NewFunctionShape([]FunctionShapePoint{{0, 0}, {100, 10}})
 			resourceToWeightMap := ResourceToWeightMap{v1.ResourceName("intel.com/foo"): 3, v1.ResourceName("intel.com/bar"): 5}
 			prior := RequestedToCapacityRatioResourceAllocationPriority(functionShape, resourceToWeightMap)
-			list, err := priorityFunction(prior.PriorityMap, nil, nil)(test.pod, snapshot, test.nodes)
+			list, err := runMapReducePriority(prior.PriorityMap, nil, nil, test.pod, snapshot, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}

pkg/scheduler/algorithm/priorities/resource_limits.go

@@ -19,7 +19,7 @@ package priorities
 import (
 	"fmt"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 	schedulernodeinfo "k8s.io/kubernetes/pkg/scheduler/nodeinfo"
 
@@ -43,7 +43,7 @@ func ResourceLimitsPriorityMap(pod *v1.Pod, meta interface{}, nodeInfo *schedule
 
 	// compute pod limits
 	var podLimits *schedulernodeinfo.Resource
-	if priorityMeta, ok := meta.(*priorityMetadata); ok {
+	if priorityMeta, ok := meta.(*priorityMetadata); ok && priorityMeta != nil {
 		// We were able to parse metadata, use podLimits from there.
 		podLimits = priorityMeta.podLimits
 	} else {

pkg/scheduler/algorithm/priorities/resource_limits_test.go

@@ -20,7 +20,7 @@ import (
 	"reflect"
 	"testing"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 	nodeinfosnapshot "k8s.io/kubernetes/pkg/scheduler/nodeinfo/snapshot"
@@ -144,14 +144,12 @@ func TestResourceLimitsPriority(t *testing.T) {
 			}
 
 			for _, hasMeta := range []bool{true, false} {
-				var function PriorityFunction
+				meta := metadata
-				if hasMeta {
+				if !hasMeta {
-					function = priorityFunction(ResourceLimitsPriorityMap, nil, metadata)
+					meta = nil
-				} else {
-					function = priorityFunction(ResourceLimitsPriorityMap, nil, nil)
 				}
 
-				list, err := function(test.pod, snapshot, test.nodes)
+				list, err := runMapReducePriority(ResourceLimitsPriorityMap, nil, meta, test.pod, snapshot, test.nodes)
 
 				if err != nil {
 					t.Errorf("unexpected error: %v", err)

pkg/scheduler/algorithm/priorities/selector_spreading_test.go

@@ -355,8 +355,7 @@ func TestSelectorSpreadPriority(t *testing.T) {
 			)
 			metaData := metaDataProducer(test.pod, nodes, snapshot)
 
-			ttp := priorityFunction(selectorSpread.CalculateSpreadPriorityMap, selectorSpread.CalculateSpreadPriorityReduce, metaData)
+			list, err := runMapReducePriority(selectorSpread.CalculateSpreadPriorityMap, selectorSpread.CalculateSpreadPriorityReduce, metaData, test.pod, snapshot, makeNodeList(test.nodes))
-			list, err := ttp(test.pod, snapshot, makeNodeList(test.nodes))
 			if err != nil {
 				t.Errorf("unexpected error: %v \n", err)
 			}
@@ -593,8 +592,7 @@ func TestZoneSelectorSpreadPriority(t *testing.T) {
 				1,
 			)
 			metaData := metaDataProducer(test.pod, nodes, snapshot)
-			ttp := priorityFunction(selectorSpread.CalculateSpreadPriorityMap, selectorSpread.CalculateSpreadPriorityReduce, metaData)
+			list, err := runMapReducePriority(selectorSpread.CalculateSpreadPriorityMap, selectorSpread.CalculateSpreadPriorityReduce, metaData, test.pod, snapshot, makeLabeledNodeList(labeledNodes))
-			list, err := ttp(test.pod, snapshot, makeLabeledNodeList(labeledNodes))
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}
@@ -783,8 +781,7 @@ func TestZoneSpreadPriority(t *testing.T) {
 				1,
 			)
 			metaData := metaDataProducer(test.pod, nodes, snapshot)
-			ttp := priorityFunction(zoneSpread.CalculateAntiAffinityPriorityMap, zoneSpread.CalculateAntiAffinityPriorityReduce, metaData)
+			list, err := runMapReducePriority(zoneSpread.CalculateAntiAffinityPriorityMap, zoneSpread.CalculateAntiAffinityPriorityReduce, metaData, test.pod, snapshot, makeLabeledNodeList(test.nodes))
-			list, err := ttp(test.pod, snapshot, makeLabeledNodeList(test.nodes))
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}

pkg/scheduler/algorithm/priorities/spreading_perf_test.go

@@ -98,8 +98,7 @@ func BenchmarkTestSelectorSpreadPriority(b *testing.B) {
 				meta := &priorityMetadata{
 					podSelectors: getSelectors(pod, ss.serviceLister, ss.controllerLister, ss.replicaSetLister, ss.statefulSetLister),
 				}
-				ttp := priorityFunction(ss.CalculateSpreadPriorityMap, ss.CalculateSpreadPriorityReduce, meta)
+				_, err := runMapReducePriority(ss.CalculateSpreadPriorityMap, ss.CalculateSpreadPriorityReduce, meta, pod, snapshot, filteredNodes)
-				_, err := ttp(pod, snapshot, filteredNodes)
 				if err != nil {
 					b.Fatal(err)
 				}

pkg/scheduler/algorithm/priorities/taint_toleration_test.go

@@ -228,8 +228,7 @@ func TestTaintAndToleration(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			snapshot := nodeinfosnapshot.NewSnapshot(nil, test.nodes)
-			ttp := priorityFunction(ComputeTaintTolerationPriorityMap, ComputeTaintTolerationPriorityReduce, nil)
+			list, err := runMapReducePriority(ComputeTaintTolerationPriorityMap, ComputeTaintTolerationPriorityReduce, nil, test.pod, snapshot, test.nodes)
-			list, err := ttp(test.pod, snapshot, test.nodes)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}

pkg/scheduler/algorithm/priorities/test_util.go

@@ -19,7 +19,7 @@ package priorities
 import (
 	"sort"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
@@ -58,27 +58,25 @@ func makeNodeWithExtendedResource(node string, milliCPU, memory int64, extendedR
 	}
 }
 
-func priorityFunction(mapFn PriorityMapFunction, reduceFn PriorityReduceFunction, metaData interface{}) PriorityFunction {
+func runMapReducePriority(mapFn PriorityMapFunction, reduceFn PriorityReduceFunction, metaData interface{}, pod *v1.Pod, sharedLister schedulerlisters.SharedLister, nodes []*v1.Node) (framework.NodeScoreList, error) {
-	return func(pod *v1.Pod, sharedLister schedulerlisters.SharedLister, nodes []*v1.Node) (framework.NodeScoreList, error) {
+	result := make(framework.NodeScoreList, 0, len(nodes))
-		result := make(framework.NodeScoreList, 0, len(nodes))
+	for i := range nodes {
-		for i := range nodes {
+		nodeInfo, err := sharedLister.NodeInfos().Get(nodes[i].Name)
-			nodeInfo, err := sharedLister.NodeInfos().Get(nodes[i].Name)
+		if err != nil {
-			if err != nil {
+			return nil, err
-				return nil, err
-			}
-			hostResult, err := mapFn(pod, metaData, nodeInfo)
-			if err != nil {
-				return nil, err
-			}
-			result = append(result, hostResult)
 		}
-		if reduceFn != nil {
+		hostResult, err := mapFn(pod, metaData, nodeInfo)
-			if err := reduceFn(pod, metaData, sharedLister, result); err != nil {
+		if err != nil {
-				return nil, err
+			return nil, err
-			}
 		}
-		return result, nil
+		result = append(result, hostResult)
 	}
+	if reduceFn != nil {
+		if err := reduceFn(pod, metaData, sharedLister, result); err != nil {
+			return nil, err
+		}
+	}
+	return result, nil
 }
 
 func sortNodeScoreList(out framework.NodeScoreList) {

pkg/scheduler/algorithm/priorities/types.go

@@ -38,11 +38,6 @@ type PriorityReduceFunction func(pod *v1.Pod, meta interface{}, sharedLister sch
 // is now used for only for priority functions. For predicates please use PredicateMetadataProducer.
 type PriorityMetadataProducer func(pod *v1.Pod, filteredNodes []*v1.Node, sharedLister schedulerlisters.SharedLister) interface{}
 
-// PriorityFunction is a function that computes scores for all nodes.
-// DEPRECATED
-// Use Map-Reduce pattern for priority functions.
-type PriorityFunction func(pod *v1.Pod, sharedLister schedulerlisters.SharedLister, nodes []*v1.Node) (framework.NodeScoreList, error)
-
 // PriorityConfig is a config used for a priority function.
 type PriorityConfig struct {
 	Name   string

pkg/scheduler/algorithm_factory.go

@@ -70,19 +70,13 @@ type PredicateMetadataProducerFactory func(PluginFactoryArgs) predicates.Predica
 // FitPredicateFactory produces a FitPredicate from the given args.
 type FitPredicateFactory func(PluginFactoryArgs) predicates.FitPredicate
 
-// PriorityFunctionFactory produces a PriorityConfig from the given args.
+// PriorityFunctionFactory produces map & reduce priority functions
-// DEPRECATED
-// Use Map-Reduce pattern for priority functions.
-type PriorityFunctionFactory func(PluginFactoryArgs) priorities.PriorityFunction
-
-// PriorityFunctionFactory2 produces map & reduce priority functions
 // from a given args.
-// FIXME: Rename to PriorityFunctionFactory.
+type PriorityFunctionFactory func(PluginFactoryArgs) (priorities.PriorityMapFunction, priorities.PriorityReduceFunction)
-type PriorityFunctionFactory2 func(PluginFactoryArgs) (priorities.PriorityMapFunction, priorities.PriorityReduceFunction)
 
 // PriorityConfigFactory produces a PriorityConfig from the given function and weight
 type PriorityConfigFactory struct {
-	MapReduceFunction PriorityFunctionFactory2
+	MapReduceFunction PriorityFunctionFactory
 	Weight            int64
 }