Merge pull request #63929 from losipiuk/lo/scheduler-priorities-ca-friendly

pkg/scheduler/algorithm/priorities/BUILD

@@ -19,6 +19,7 @@ go_library(
         "node_label.go",
         "node_prefer_avoid_pods.go",
         "reduce.go",
+        "requested_to_capacity_ratio.go",
         "resource_allocation.go",
         "resource_limits.go",
         "selector_spreading.go",
@@ -58,6 +59,7 @@ go_test(
         "node_affinity_test.go",
         "node_label_test.go",
         "node_prefer_avoid_pods_test.go",
+        "requested_to_capacity_ratio_test.go",
         "resource_limits_test.go",
         "selector_spreading_test.go",
         "taint_toleration_test.go",
@@ -70,6 +72,7 @@ go_test(
         "//pkg/scheduler/api:go_default_library",
         "//pkg/scheduler/cache:go_default_library",
         "//pkg/scheduler/testing:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/k8s.io/api/apps/v1beta1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/extensions/v1beta1:go_default_library",

pkg/scheduler/algorithm/priorities/requested_to_capacity_ratio.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package priorities
+
+import (
+	"fmt"
+
+	schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
+	schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
+)
+
+// FunctionShape represents shape of scoring function.
+// For safety use NewFunctionShape which performs precondition checks for struct creation.
+type FunctionShape []FunctionShapePoint
+
+// FunctionShapePoint represents single point in scoring function shape.
+type FunctionShapePoint struct {
+	// Utilization is function argument.
+	Utilization int64
+	// Score is function value.
+	Score int64
+}
+
+var (
+	// give priority to least utilized nodes by default
+	defaultFunctionShape, _ = NewFunctionShape([]FunctionShapePoint{{0, 10}, {100, 0}})
+)
+
+const (
+	minUtilization = 0
+	maxUtilization = 100
+	minScore       = 0
+	maxScore       = schedulerapi.MaxPriority
+)
+
+// NewFunctionShape creates instance of FunctionShape in a safe way performing all
+// necessary sanity checks.
+func NewFunctionShape(points []FunctionShapePoint) (FunctionShape, error) {
+
+	n := len(points)
+
+	if n == 0 {
+		return nil, fmt.Errorf("at least one point must be specified")
+	}
+
+	for i := 1; i < n; i++ {
+		if points[i-1].Utilization >= points[i].Utilization {
+			return nil, fmt.Errorf("utilization values must be sorted. Utilization[%d]==%d >= Utilization[%d]==%d", i-1, points[i-1].Utilization, i, points[i].Utilization)
+		}
+	}
+
+	for i, point := range points {
+		if point.Utilization < minUtilization {
+			return nil, fmt.Errorf("utilization values must not be less than %d. Utilization[%d]==%d", minUtilization, i, point.Utilization)
+		}
+		if point.Utilization > maxUtilization {
+			return nil, fmt.Errorf("utilization values must not be greater than %d. Utilization[%d]==%d", maxUtilization, i, point.Utilization)
+		}
+		if point.Score < minScore {
+			return nil, fmt.Errorf("score values must not be less than %d. Score[%d]==%d", minScore, i, point.Score)
+		}
+		if point.Score > maxScore {
+			return nil, fmt.Errorf("score valuses not be greater than %d. Score[%d]==%d", maxScore, i, point.Score)
+		}
+	}
+
+	// We make defensive copy so we make no assumption if array passed as argument is not changed afterwards
+	pointsCopy := make(FunctionShape, n)
+	copy(pointsCopy, points)
+	return pointsCopy, nil
+}
+
+// RequestedToCapacityRatioResourceAllocationPriorityDefault creates a requestedToCapacity based
+// ResourceAllocationPriority using default resource scoring function shape.
+// The default function assigns 1.0 to resource when all capacity is available
+// and 0.0 when requested amount is equal to capacity.
+func RequestedToCapacityRatioResourceAllocationPriorityDefault() *ResourceAllocationPriority {
+	return RequestedToCapacityRatioResourceAllocationPriority(defaultFunctionShape)
+}
+
+// RequestedToCapacityRatioResourceAllocationPriority creates a requestedToCapacity based
+// ResourceAllocationPriority using provided resource scoring function shape.
+func RequestedToCapacityRatioResourceAllocationPriority(scoringFunctionShape FunctionShape) *ResourceAllocationPriority {
+	return &ResourceAllocationPriority{"RequestedToCapacityRatioResourceAllocationPriority", buildRequestedToCapacityRatioScorerFunction(scoringFunctionShape)}
+}
+
+func buildRequestedToCapacityRatioScorerFunction(scoringFunctionShape FunctionShape) func(*schedulercache.Resource, *schedulercache.Resource, bool, int, int) int64 {
+	rawScoringFunction := buildBrokenLinearFunction(scoringFunctionShape)
+
+	resourceScoringFunction := func(requested, capacity int64) int64 {
+		if capacity == 0 || requested > capacity {
+			return rawScoringFunction(maxUtilization)
+		}
+
+		return rawScoringFunction(maxUtilization - (capacity-requested)*maxUtilization/capacity)
+	}
+
+	return func(requested, allocable *schedulercache.Resource, includeVolumes bool, requestedVolumes int, allocatableVolumes int) int64 {
+		cpuScore := resourceScoringFunction(requested.MilliCPU, allocable.MilliCPU)
+		memoryScore := resourceScoringFunction(requested.Memory, allocable.Memory)
+		return (cpuScore + memoryScore) / 2
+	}
+}
+
+// Creates a function which is built using linear segments. Segments are defined via shape array.
+// Shape[i].Utilization slice represents points on "utilization" axis where different segments meet.
+// Shape[i].Score represents function values at meeting points.
+//
+// function f(p) is defined as:
+//   shape[0].Score for p < f[0].Utilization
+//   shape[i].Score for p == shape[i].Utilization
+//   shape[n-1].Score for p > shape[n-1].Utilization
+// and linear between points (p < shape[i].Utilization)
+func buildBrokenLinearFunction(shape FunctionShape) func(int64) int64 {
+	n := len(shape)
+	return func(p int64) int64 {
+		for i := 0; i < n; i++ {
+			if p <= shape[i].Utilization {
+				if i == 0 {
+					return shape[0].Score
+				}
+				return shape[i-1].Score + (shape[i].Score-shape[i-1].Score)*(p-shape[i-1].Utilization)/(shape[i].Utilization-shape[i-1].Utilization)
+			}
+		}
+		return shape[n-1].Score
+	}
+}

pkg/scheduler/algorithm/priorities/requested_to_capacity_ratio_test.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package priorities
+
+import (
+	"reflect"
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
+	schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
+)
+
+func TestCreatingFunctionShapeErrorsIfEmptyPoints(t *testing.T) {
+	var err error
+	_, err = NewFunctionShape([]FunctionShapePoint{})
+	assert.Equal(t, "at least one point must be specified", err.Error())
+}
+
+func TestCreatingFunctionShapeErrorsIfXIsNotSorted(t *testing.T) {
+	var err error
+	_, err = NewFunctionShape([]FunctionShapePoint{{10, 1}, {15, 2}, {20, 3}, {19, 4}, {25, 5}})
+	assert.Equal(t, "utilization values must be sorted. Utilization[2]==20 >= Utilization[3]==19", err.Error())
+
+	_, err = NewFunctionShape([]FunctionShapePoint{{10, 1}, {20, 2}, {20, 3}, {22, 4}, {25, 5}})
+	assert.Equal(t, "utilization values must be sorted. Utilization[1]==20 >= Utilization[2]==20", err.Error())
+}
+
+func TestCreatingFunctionPointNotInAllowedRange(t *testing.T) {
+	var err error
+	_, err = NewFunctionShape([]FunctionShapePoint{{-1, 0}, {100, 10}})
+	assert.Equal(t, "utilization values must not be less than 0. Utilization[0]==-1", err.Error())
+
+	_, err = NewFunctionShape([]FunctionShapePoint{{0, 0}, {101, 10}})
+	assert.Equal(t, "utilization values must not be greater than 100. Utilization[1]==101", err.Error())
+
+	_, err = NewFunctionShape([]FunctionShapePoint{{0, -1}, {100, 10}})
+	assert.Equal(t, "score values must not be less than 0. Score[0]==-1", err.Error())
+
+	_, err = NewFunctionShape([]FunctionShapePoint{{0, 0}, {100, 11}})
+	assert.Equal(t, "score valuses not be greater than 10. Score[1]==11", err.Error())
+}
+
+func TestBrokenLinearFunction(t *testing.T) {
+	type Assertion struct {
+		p        int64
+		expected int64
+	}
+	type Test struct {
+		points     []FunctionShapePoint
+		assertions []Assertion
+	}
+
+	tests := []Test{
+		{
+			points: []FunctionShapePoint{{10, 1}, {90, 9}},
+			assertions: []Assertion{
+				{p: -10, expected: 1},
+				{p: 0, expected: 1},
+				{p: 9, expected: 1},
+				{p: 10, expected: 1},
+				{p: 15, expected: 1},
+				{p: 19, expected: 1},
+				{p: 20, expected: 2},
+				{p: 89, expected: 8},
+				{p: 90, expected: 9},
+				{p: 99, expected: 9},
+				{p: 100, expected: 9},
+				{p: 110, expected: 9},
+			},
+		},
+		{
+			points: []FunctionShapePoint{{0, 2}, {40, 10}, {100, 0}},
+			assertions: []Assertion{
+				{p: -10, expected: 2},
+				{p: 0, expected: 2},
+				{p: 20, expected: 6},
+				{p: 30, expected: 8},
+				{p: 40, expected: 10},
+				{p: 70, expected: 5},
+				{p: 100, expected: 0},
+				{p: 110, expected: 0},
+			},
+		},
+		{
+			points: []FunctionShapePoint{{0, 2}, {40, 2}, {100, 2}},
+			assertions: []Assertion{
+				{p: -10, expected: 2},
+				{p: 0, expected: 2},
+				{p: 20, expected: 2},
+				{p: 30, expected: 2},
+				{p: 40, expected: 2},
+				{p: 70, expected: 2},
+				{p: 100, expected: 2},
+				{p: 110, expected: 2},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		functionShape, err := NewFunctionShape(test.points)
+		assert.Nil(t, err)
+		function := buildBrokenLinearFunction(functionShape)
+		for _, assertion := range test.assertions {
+			assert.InDelta(t, assertion.expected, function(assertion.p), 0.1, "points=%v, p=%f", test.points, assertion.p)
+		}
+	}
+}
+
+func TestRequestedToCapacityRatio(t *testing.T) {
+	type resources struct {
+		cpu int64
+		mem int64
+	}
+
+	type nodeResources struct {
+		capacity resources
+		used     resources
+	}
+
+	type test struct {
+		test               string
+		requested          resources
+		nodes              map[string]nodeResources
+		expectedPriorities schedulerapi.HostPriorityList
+	}
+
+	tests := []test{
+		{
+			test:      "nothing scheduled, nothing requested (default - least requested nodes have priority)",
+			requested: resources{0, 0},
+			nodes: map[string]nodeResources{
+				"node1": {
+					capacity: resources{4000, 10000},
+					used:     resources{0, 0},
+				},
+				"node2": {
+					capacity: resources{4000, 10000},
+					used:     resources{0, 0},
+				},
+			},
+			expectedPriorities: []schedulerapi.HostPriority{{Host: "node1", Score: 10}, {Host: "node2", Score: 10}},
+		},
+		{
+			test:      "nothing scheduled, resources requested, differently sized machines (default - least requested nodes have priority)",
+			requested: resources{3000, 5000},
+			nodes: map[string]nodeResources{
+				"node1": {
+					capacity: resources{4000, 10000},
+					used:     resources{0, 0},
+				},
+				"node2": {
+					capacity: resources{6000, 10000},
+					used:     resources{0, 0},
+				},
+			},
+			expectedPriorities: []schedulerapi.HostPriority{{Host: "node1", Score: 4}, {Host: "node2", Score: 5}},
+		},
+		{
+			test:      "no resources requested, pods scheduled with resources (default - least requested nodes have priority)",
+			requested: resources{0, 0},
+			nodes: map[string]nodeResources{
+				"node1": {
+					capacity: resources{4000, 10000},
+					used:     resources{3000, 5000},
+				},
+				"node2": {
+					capacity: resources{6000, 10000},
+					used:     resources{3000, 5000},
+				},
+			},
+			expectedPriorities: []schedulerapi.HostPriority{{Host: "node1", Score: 4}, {Host: "node2", Score: 5}},
+		},
+	}
+
+	buildResourcesPod := func(node string, requestedResources resources) *v1.Pod {
+		return &v1.Pod{Spec: v1.PodSpec{
+			NodeName: node,
+			Containers: []v1.Container{
+				{
+					Resources: v1.ResourceRequirements{
+						Requests: v1.ResourceList{
+							v1.ResourceCPU:    *resource.NewMilliQuantity(requestedResources.cpu, resource.DecimalSI),
+							v1.ResourceMemory: *resource.NewQuantity(requestedResources.mem, resource.DecimalSI),
+						},
+					},
+				},
+			},
+		},
+		}
+	}
+
+	for _, test := range tests {
+
+		nodeNames := make([]string, 0)
+		for nodeName := range test.nodes {
+			nodeNames = append(nodeNames, nodeName)
+		}
+		sort.Strings(nodeNames)
+
+		nodes := make([]*v1.Node, 0)
+		for _, nodeName := range nodeNames {
+			node := test.nodes[nodeName]
+			nodes = append(nodes, makeNode(nodeName, node.capacity.cpu, node.capacity.mem))
+		}
+
+		scheduledPods := make([]*v1.Pod, 0)
+		for name, node := range test.nodes {
+			scheduledPods = append(scheduledPods,
+				buildResourcesPod(name, node.used))
+		}
+
+		newPod := buildResourcesPod("", test.requested)
+
+		nodeNameToInfo := schedulercache.CreateNodeNameToInfoMap(scheduledPods, nodes)
+		list, err := priorityFunction(RequestedToCapacityRatioResourceAllocationPriorityDefault().PriorityMap, nil, nil)(newPod, nodeNameToInfo, nodes)
+		if err != nil {
+			t.Errorf("unexpected error: %v", err)
+		}
+		if !reflect.DeepEqual(test.expectedPriorities, list) {
+			t.Errorf("%s: expected %#v, got %#v", test.test, test.expectedPriorities, list)
+		}
+	}
+}

pkg/scheduler/algorithmprovider/defaults/compatibility_test.go

@@ -603,6 +603,103 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
 				},
 			},
 		},
+		// Do not change this JSON after the corresponding release has been tagged.
+		// A failure indicates backwards compatibility with the specified release was broken.
+		"1.11": {
+			JSON: `{
+		  "kind": "Policy",
+		  "apiVersion": "v1",
+		  "predicates": [
+			{"name": "MatchNodeSelector"},
+			{"name": "PodFitsResources"},
+			{"name": "PodFitsHostPorts"},
+			{"name": "HostName"},
+			{"name": "NoDiskConflict"},
+			{"name": "NoVolumeZoneConflict"},
+			{"name": "PodToleratesNodeTaints"},
+			{"name": "CheckNodeMemoryPressure"},
+			{"name": "CheckNodeDiskPressure"},
+			{"name": "CheckNodePIDPressure"},
+			{"name": "CheckNodeCondition"},
+			{"name": "MaxEBSVolumeCount"},
+			{"name": "MaxGCEPDVolumeCount"},
+			{"name": "MaxAzureDiskVolumeCount"},
+			{"name": "MatchInterPodAffinity"},
+			{"name": "GeneralPredicates"},
+			{"name": "CheckVolumeBinding"},
+			{"name": "TestServiceAffinity", "argument": {"serviceAffinity" : {"labels" : ["region"]}}},
+			{"name": "TestLabelsPresence",  "argument": {"labelsPresence"  : {"labels" : ["foo"], "presence":true}}}
+		  ],"priorities": [
+			{"name": "EqualPriority",   "weight": 2},
+			{"name": "ImageLocalityPriority",   "weight": 2},
+			{"name": "LeastRequestedPriority",   "weight": 2},
+			{"name": "BalancedResourceAllocation",   "weight": 2},
+			{"name": "SelectorSpreadPriority",   "weight": 2},
+			{"name": "NodePreferAvoidPodsPriority",   "weight": 2},
+			{"name": "NodeAffinityPriority",   "weight": 2},
+			{"name": "TaintTolerationPriority",   "weight": 2},
+			{"name": "InterPodAffinityPriority",   "weight": 2},
+			{"name": "MostRequestedPriority",   "weight": 2},
+			{
+				"name": "RequestedToCapacityRatioPriority",
+				"weight": 2,
+				"argument": {
+				"requestedToCapacityRatioArguments": {
+					"shape": [
+						{"utilization": 0,  "score": 0},
+						{"utilization": 50, "score": 7}
+					]
+				}
+			}}
+		  ]
+		}`,
+			ExpectedPolicy: schedulerapi.Policy{
+				Predicates: []schedulerapi.PredicatePolicy{
+					{Name: "MatchNodeSelector"},
+					{Name: "PodFitsResources"},
+					{Name: "PodFitsHostPorts"},
+					{Name: "HostName"},
+					{Name: "NoDiskConflict"},
+					{Name: "NoVolumeZoneConflict"},
+					{Name: "PodToleratesNodeTaints"},
+					{Name: "CheckNodeMemoryPressure"},
+					{Name: "CheckNodeDiskPressure"},
+					{Name: "CheckNodePIDPressure"},
+					{Name: "CheckNodeCondition"},
+					{Name: "MaxEBSVolumeCount"},
+					{Name: "MaxGCEPDVolumeCount"},
+					{Name: "MaxAzureDiskVolumeCount"},
+					{Name: "MatchInterPodAffinity"},
+					{Name: "GeneralPredicates"},
+					{Name: "CheckVolumeBinding"},
+					{Name: "TestServiceAffinity", Argument: &schedulerapi.PredicateArgument{ServiceAffinity: &schedulerapi.ServiceAffinity{Labels: []string{"region"}}}},
+					{Name: "TestLabelsPresence", Argument: &schedulerapi.PredicateArgument{LabelsPresence: &schedulerapi.LabelsPresence{Labels: []string{"foo"}, Presence: true}}},
+				},
+				Priorities: []schedulerapi.PriorityPolicy{
+					{Name: "EqualPriority", Weight: 2},
+					{Name: "ImageLocalityPriority", Weight: 2},
+					{Name: "LeastRequestedPriority", Weight: 2},
+					{Name: "BalancedResourceAllocation", Weight: 2},
+					{Name: "SelectorSpreadPriority", Weight: 2},
+					{Name: "NodePreferAvoidPodsPriority", Weight: 2},
+					{Name: "NodeAffinityPriority", Weight: 2},
+					{Name: "TaintTolerationPriority", Weight: 2},
+					{Name: "InterPodAffinityPriority", Weight: 2},
+					{Name: "MostRequestedPriority", Weight: 2},
+					{
+						Name:   "RequestedToCapacityRatioPriority",
+						Weight: 2,
+						Argument: &schedulerapi.PriorityArgument{
+							RequestedToCapacityRatioArguments: &schedulerapi.RequestedToCapacityRatioArguments{
+								UtilizationShape: []schedulerapi.UtilizationShapePoint{
+									{Utilization: 0, Score: 0},
+									{Utilization: 50, Score: 7},
+								}},
+						},
+					},
+				},
+			},
+		},
 	}
 
 	registeredPredicates := sets.NewString(factory.ListRegisteredFitPredicates()...)

pkg/scheduler/algorithmprovider/defaults/defaults.go

@@ -100,6 +100,11 @@ func init() {
 	factory.RegisterPriorityFunction2("ImageLocalityPriority", priorities.ImageLocalityPriorityMap, nil, 1)
 	// Optional, cluster-autoscaler friendly priority function - give used nodes higher priority.
 	factory.RegisterPriorityFunction2("MostRequestedPriority", priorities.MostRequestedPriorityMap, nil, 1)
+	factory.RegisterPriorityFunction2(
+		"RequestedToCapacityRatioPriority",
+		priorities.RequestedToCapacityRatioResourceAllocationPriorityDefault().PriorityMap,
+		nil,
+		1)
 }
 
 func defaultPredicates() sets.String {

pkg/scheduler/api/types.go

@@ -109,6 +109,8 @@ type PriorityArgument struct {
 	// The priority function that checks whether a particular node has a certain label
 	// defined or not, regardless of value
 	LabelPreference *LabelPreference
+	// The RequestedToCapacityRatio priority function is parametrized with function shape.
+	RequestedToCapacityRatioArguments *RequestedToCapacityRatioArguments
 }
 
 // ServiceAffinity holds the parameters that are used to configure the corresponding predicate in scheduler policy configuration.
@@ -143,6 +145,20 @@ type LabelPreference struct {
 	Presence bool
 }
 
+// RequestedToCapacityRatioArguments holds arguments specific to RequestedToCapacityRatio priority function
+type RequestedToCapacityRatioArguments struct {
+	// Array of point defining priority function shape
+	UtilizationShape []UtilizationShapePoint
+}
+
+// UtilizationShapePoint represents single point of priority function shape
+type UtilizationShapePoint struct {
+	// Utilization (x axis). Valid values are 0 to 100. Fully utilized node maps to 100.
+	Utilization int
+	// Score assigned to given utilization (y axis). Valid values are 0 to 10.
+	Score int
+}
+
 // ExtenderManagedResource describes the arguments of extended resources
 // managed by an extender.
 type ExtenderManagedResource struct {

pkg/scheduler/api/v1/types.go

@@ -91,6 +91,8 @@ type PriorityArgument struct {
 	// The priority function that checks whether a particular node has a certain label
 	// defined or not, regardless of value
 	LabelPreference *LabelPreference `json:"labelPreference"`
+	// The RequestedToCapacityRatio priority function is parametrized with function shape.
+	RequestedToCapacityRatioArguments *RequestedToCapacityRatioArguments `json:"requestedToCapacityRatioArguments"`
 }
 
 // ServiceAffinity holds the parameters that are used to configure the corresponding predicate in scheduler policy configuration.
@@ -125,6 +127,20 @@ type LabelPreference struct {
 	Presence bool `json:"presence"`
 }
 
+// RequestedToCapacityRatioArguments holds arguments specific to RequestedToCapacityRatio priority function
+type RequestedToCapacityRatioArguments struct {
+	// Array of point defining priority function shape
+	UtilizationShape []UtilizationShapePoint `json:"shape"`
+}
+
+// UtilizationShapePoint represents single point of priority function shape
+type UtilizationShapePoint struct {
+	// Utilization (x axis). Valid values are 0 to 100. Fully utilized node maps to 100.
+	Utilization int `json:"utilization"`
+	// Score assigned to given utilization (y axis). Valid values are 0 to 10.
+	Score int `json:"score"`
+}
+
 // ExtenderManagedResource describes the arguments of extended resources
 // managed by an extender.
 type ExtenderManagedResource struct {

pkg/scheduler/api/v1/zz_generated.deepcopy.go

@@ -538,6 +538,15 @@ func (in *PriorityArgument) DeepCopyInto(out *PriorityArgument) {
 			**out = **in
 		}
 	}
+	if in.RequestedToCapacityRatioArguments != nil {
+		in, out := &in.RequestedToCapacityRatioArguments, &out.RequestedToCapacityRatioArguments
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(RequestedToCapacityRatioArguments)
+			(*in).DeepCopyInto(*out)
+		}
+	}
 	return
 }
 
@@ -577,6 +586,27 @@ func (in *PriorityPolicy) DeepCopy() *PriorityPolicy {
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *RequestedToCapacityRatioArguments) DeepCopyInto(out *RequestedToCapacityRatioArguments) {
+	*out = *in
+	if in.UtilizationShape != nil {
+		in, out := &in.UtilizationShape, &out.UtilizationShape
+		*out = make([]UtilizationShapePoint, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RequestedToCapacityRatioArguments.
+func (in *RequestedToCapacityRatioArguments) DeepCopy() *RequestedToCapacityRatioArguments {
+	if in == nil {
+		return nil
+	}
+	out := new(RequestedToCapacityRatioArguments)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ServiceAffinity) DeepCopyInto(out *ServiceAffinity) {
 	*out = *in
 	if in.Labels != nil {
@@ -614,6 +644,22 @@ func (in *ServiceAntiAffinity) DeepCopy() *ServiceAntiAffinity {
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *UtilizationShapePoint) DeepCopyInto(out *UtilizationShapePoint) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new UtilizationShapePoint.
+func (in *UtilizationShapePoint) DeepCopy() *UtilizationShapePoint {
+	if in == nil {
+		return nil
+	}
+	out := new(UtilizationShapePoint)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Victims) DeepCopyInto(out *Victims) {
 	*out = *in
 	if in.Pods != nil {

pkg/scheduler/api/zz_generated.deepcopy.go

@@ -538,6 +538,15 @@ func (in *PriorityArgument) DeepCopyInto(out *PriorityArgument) {
 			**out = **in
 		}
 	}
+	if in.RequestedToCapacityRatioArguments != nil {
+		in, out := &in.RequestedToCapacityRatioArguments, &out.RequestedToCapacityRatioArguments
+		if *in == nil {
+			*out = nil
+		} else {
+			*out = new(RequestedToCapacityRatioArguments)
+			(*in).DeepCopyInto(*out)
+		}
+	}
 	return
 }
 
@@ -577,6 +586,27 @@ func (in *PriorityPolicy) DeepCopy() *PriorityPolicy {
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *RequestedToCapacityRatioArguments) DeepCopyInto(out *RequestedToCapacityRatioArguments) {
+	*out = *in
+	if in.UtilizationShape != nil {
+		in, out := &in.UtilizationShape, &out.UtilizationShape
+		*out = make([]UtilizationShapePoint, len(*in))
+		copy(*out, *in)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RequestedToCapacityRatioArguments.
+func (in *RequestedToCapacityRatioArguments) DeepCopy() *RequestedToCapacityRatioArguments {
+	if in == nil {
+		return nil
+	}
+	out := new(RequestedToCapacityRatioArguments)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ServiceAffinity) DeepCopyInto(out *ServiceAffinity) {
 	*out = *in
 	if in.Labels != nil {
@@ -614,6 +644,22 @@ func (in *ServiceAntiAffinity) DeepCopy() *ServiceAntiAffinity {
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *UtilizationShapePoint) DeepCopyInto(out *UtilizationShapePoint) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new UtilizationShapePoint.
+func (in *UtilizationShapePoint) DeepCopy() *UtilizationShapePoint {
+	if in == nil {
+		return nil
+	}
+	out := new(UtilizationShapePoint)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Victims) DeepCopyInto(out *Victims) {
 	*out = *in
 	if in.Pods != nil {

pkg/scheduler/factory/BUILD

@@ -103,12 +103,14 @@ go_test(
         "//pkg/api/testing:go_default_library",
         "//pkg/scheduler:go_default_library",
         "//pkg/scheduler/algorithm:go_default_library",
+        "//pkg/scheduler/algorithm/priorities:go_default_library",
         "//pkg/scheduler/api:go_default_library",
         "//pkg/scheduler/api/latest:go_default_library",
         "//pkg/scheduler/cache:go_default_library",
         "//pkg/scheduler/core:go_default_library",
         "//pkg/scheduler/testing:go_default_library",
         "//pkg/scheduler/util:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/policy/v1beta1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",

pkg/scheduler/factory/plugins.go

@@ -322,6 +322,15 @@ func RegisterCustomPriorityFunction(policy schedulerapi.PriorityPolicy) string {
 				},
 				Weight: policy.Weight,
 			}
+		} else if policy.Argument.RequestedToCapacityRatioArguments != nil {
+			pcf = &PriorityConfigFactory{
+				MapReduceFunction: func(args PluginFactoryArgs) (algorithm.PriorityMapFunction, algorithm.PriorityReduceFunction) {
+					scoringFunctionShape := buildScoringFunctionShapeFromRequestedToCapacityRatioArguments(policy.Argument.RequestedToCapacityRatioArguments)
+					p := priorities.RequestedToCapacityRatioResourceAllocationPriority(scoringFunctionShape)
+					return p.PriorityMap, nil
+				},
+				Weight: policy.Weight,
+			}
 		}
 	} else if existingPcf, ok := priorityFunctionMap[policy.Name]; ok {
 		glog.V(2).Infof("Priority type %s already registered, reusing.", policy.Name)
@@ -340,6 +349,19 @@ func RegisterCustomPriorityFunction(policy schedulerapi.PriorityPolicy) string {
 	return RegisterPriorityConfigFactory(policy.Name, *pcf)
 }
 
+func buildScoringFunctionShapeFromRequestedToCapacityRatioArguments(arguments *schedulerapi.RequestedToCapacityRatioArguments) priorities.FunctionShape {
+	n := len(arguments.UtilizationShape)
+	points := make([]priorities.FunctionShapePoint, 0, n)
+	for _, point := range arguments.UtilizationShape {
+		points = append(points, priorities.FunctionShapePoint{Utilization: int64(point.Utilization), Score: int64(point.Score)})
+	}
+	shape, err := priorities.NewFunctionShape(points)
+	if err != nil {
+		glog.Fatalf("invalid RequestedToCapacityRatioPriority arguments: %s", err.Error())
+	}
+	return shape
+}
+
 // IsPriorityFunctionRegistered is useful for testing providers.
 func IsPriorityFunctionRegistered(name string) bool {
 	schedulerFactoryMutex.Lock()
@@ -494,6 +516,9 @@ func validatePriorityOrDie(priority schedulerapi.PriorityPolicy) {
 		if priority.Argument.LabelPreference != nil {
 			numArgs++
 		}
+		if priority.Argument.RequestedToCapacityRatioArguments != nil {
+			numArgs++
+		}
 		if numArgs != 1 {
 			glog.Fatalf("Exactly 1 priority argument is required, numArgs: %v, Priority: %s", numArgs, priority.Name)
 		}

pkg/scheduler/factory/plugins_test.go

@@ -19,7 +19,9 @@ package factory
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"k8s.io/kubernetes/pkg/scheduler/algorithm"
+	"k8s.io/kubernetes/pkg/scheduler/algorithm/priorities"
 	"k8s.io/kubernetes/pkg/scheduler/api"
 )
 
@@ -80,3 +82,19 @@ func TestValidatePriorityConfigOverFlow(t *testing.T) {
 		}
 	}
 }
+
+func TestBuildScoringFunctionShapeFromRequestedToCapacityRatioArguments(t *testing.T) {
+	arguments := api.RequestedToCapacityRatioArguments{
+		UtilizationShape: []api.UtilizationShapePoint{
+			{Utilization: 10, Score: 1},
+			{Utilization: 30, Score: 5},
+			{Utilization: 70, Score: 2},
+		}}
+	builtShape := buildScoringFunctionShapeFromRequestedToCapacityRatioArguments(&arguments)
+	expectedShape, _ := priorities.NewFunctionShape([]priorities.FunctionShapePoint{
+		{Utilization: 10, Score: 1},
+		{Utilization: 30, Score: 5},
+		{Utilization: 70, Score: 2},
+	})
+	assert.Equal(t, expectedShape, builtShape)
+}