Register RequestedToCapacityRatioPriority priority function

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,83 @@ 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}
+		  ]
+		}`,
+			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},
+				},
+			},
+		},
 	}
 
 	registeredPredicates := sets.NewString(factory.ListRegisteredFitPredicates()...)

pkg/scheduler/algorithmprovider/defaults/defaults.go

@@ -100,6 +100,15 @@ 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)
+
+	// TODO use RequestedToCapacityRatioResourceAllocationPriority(scoringFunctionShape functionShape)
+	// and build scoringFunctionShape based on configuration passed to scheduler on startup.
+	// What is proper way of passing configuration here? config file? startup flag?
+	factory.RegisterPriorityFunction2(
+		"RequestedToCapacityRatioPriority",
+		priorities.RequestedToCapacityRatioResourceAllocationPriorityDefault().PriorityMap,
+		nil,
+		1)
 }
 
 func defaultPredicates() sets.String {