kubernetes
/
autoscaler
mirror of https://github.com/kubernetes/autoscaler.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement scale from zero for clusterapi 

This allows a Machine{Set,Deployment} to scale up/down from 0,
providing the following annotations are set:

```yaml
apiVersion: v1
items:
- apiVersion: machine.openshift.io/v1beta1
  kind: MachineSet
  metadata:
    annotations:
      machine.openshift.io/cluster-api-autoscaler-node-group-min-size: "0"
      machine.openshift.io/cluster-api-autoscaler-node-group-max-size: "6"
      machine.openshift.io/vCPU: "2"
      machine.openshift.io/memoryMb: 8G
      machine.openshift.io/GPU: "1"
      machine.openshift.io/maxPods: "100"
```

Note that `machine.openshift.io/GPU` and `machine.openshift.io/maxPods`
are optional.

For autoscaling from zero, the autoscaler should convert the mem value
received in the appropriate annotation to bytes using powers of two
consistently with other providers and fail if the format received is not
expected. This gives robust behaviour consistent with cloud providers APIs
and providers implementations.

https://cloud.google.com/compute/all-pricing
https://www.iec.ch/si/binary.htm
https://github.com/openshift/kubernetes-autoscaler/blob/master/cluster-autoscaler/cloudprovider/aws/aws_manager.go#L366

Co-authored-by:  Enxebre <alberto.garcial@hotmail.com>
Co-authored-by:  Joel Speed <joel.speed@hotmail.co.uk>
Co-authored-by:  Michael McCune <elmiko@redhat.com>
This commit is contained in:
Andrew McDermott 2019-07-09 16:05:13 +01:00 committed by Michael McCune
parent 028584a476
commit de90a462c7
6 changed files with 790 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

155
cluster-autoscaler/cloudprovider/clusterapi/clusterapi_nodegroup.go
View File

@ -18,10 +18,16 @@ package clusterapi
import (
"fmt"
"math/rand"
"github.com/pkg/errors"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
gpuapis "k8s.io/autoscaler/cluster-autoscaler/utils/gpu"
kubeletapis "k8s.io/kubelet/pkg/apis"
schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework"
"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
@ -29,7 +35,17 @@ import (
)
const (
debugFormat = "%s (min: %d, max: %d, replicas: %d)"
// deprecatedMachineDeleteAnnotationKey should not be removed until minimum cluster-api support is v1alpha3
deprecatedMachineDeleteAnnotationKey = "cluster.k8s.io/delete-machine"
// TODO: determine what currently relies on deprecatedMachineAnnotationKey to determine when it can be removed
deprecatedMachineAnnotationKey = "cluster.k8s.io/machine"
machineDeleteAnnotationKey = "machine.openshift.io/cluster-api-delete-machine"
machineAnnotationKey = "machine.openshift.io/machine"
debugFormat = "%s (min: %d, max: %d, replicas: %d)"
// This default for the maximum number of pods comes from the machine-config-operator
// see https://github.com/openshift/machine-config-operator/blob/2f1bd6d99131fa4471ed95543a51dec3d5922b2b/templates/worker/01-worker-kubelet/_base/files/kubelet.yaml#L19
defaultMaxPods = 250
)
type nodegroup struct {
@ -234,7 +250,92 @@ func (ng *nodegroup) Nodes() ([]cloudprovider.Instance, error) {
// node by default, using manifest (most likely only kube-proxy).
// Implementation optional.
func (ng *nodegroup) TemplateNodeInfo() (*schedulerframework.NodeInfo, error) {
return nil, cloudprovider.ErrNotImplemented
if !ng.scalableResource.CanScaleFromZero() {
return nil, cloudprovider.ErrNotImplemented
}
cpu, err := ng.scalableResource.InstanceCPUCapacity()
if err != nil {
return nil, err
}
mem, err := ng.scalableResource.InstanceMemoryCapacity()
if err != nil {
return nil, err
}
gpu, err := ng.scalableResource.InstanceGPUCapacity()
if err != nil {
return nil, err
}
pod, err := ng.scalableResource.InstanceMaxPodsCapacity()
if err != nil {
return nil, err
}
if cpu.IsZero() || mem.IsZero() {
return nil, cloudprovider.ErrNotImplemented
}
if gpu.IsZero() {
gpu = zeroQuantity.DeepCopy()
}
if pod.IsZero() {
pod = *resource.NewQuantity(defaultMaxPods, resource.DecimalSI)
}
capacity := map[corev1.ResourceName]resource.Quantity{
corev1.ResourceCPU: cpu,
corev1.ResourceMemory: mem,
corev1.ResourcePods: pod,
gpuapis.ResourceNvidiaGPU: gpu,
}
nodeName := fmt.Sprintf("%s-asg-%d", ng.scalableResource.Name(), rand.Int63())
node := corev1.Node{
ObjectMeta: metav1.ObjectMeta{
Name: nodeName,
Labels: map[string]string{},
},
}
node.Status.Capacity = capacity
node.Status.Allocatable = capacity
node.Status.Conditions = cloudprovider.BuildReadyConditions()
node.Spec.Taints = ng.scalableResource.Taints()
node.Labels, err = ng.buildTemplateLabels(nodeName)
if err != nil {
return nil, err
}
nodeInfo := schedulerframework.NewNodeInfo(cloudprovider.BuildKubeProxy(ng.scalableResource.Name()))
nodeInfo.SetNode(&node)
return nodeInfo, nil
}
func (ng *nodegroup) buildTemplateLabels(nodeName string) (map[string]string, error) {
labels := cloudprovider.JoinStringMaps(ng.scalableResource.Labels(), buildGenericLabels(nodeName))
nodes, err := ng.Nodes()
if err != nil {
return nil, err
}
if len(nodes) > 0 {
node, err := ng.machineController.findNodeByProviderID(normalizedProviderString(nodes[0].Id))
if err != nil {
return nil, err
}
if node != nil {
labels = cloudprovider.JoinStringMaps(labels, extractNodeLabels(node))
}
}
return labels, nil
}
// Exist checks if the node group really exists on the cloud nodegroup
@ -289,9 +390,9 @@ func newNodeGroupFromScalableResource(controller *machineController, unstructure
return nil, err
}
// We don't scale from 0 so nodes must belong to a nodegroup
// that has a scale size of at least 1.
if found && replicas == 0 {
// Ensure that if the nodegroup has 0 replicas it is capable
// of scaling before adding it.
if found && replicas == 0 && !scalableResource.CanScaleFromZero() {
return nil, nil
}
@ -305,3 +406,47 @@ func newNodeGroupFromScalableResource(controller *machineController, unstructure
scalableResource: scalableResource,
}, nil
}
func buildGenericLabels(nodeName string) map[string]string {
// TODO revisit this function and add an explanation about what these
// labels are used for, or remove them if not necessary
m := make(map[string]string)
m[kubeletapis.LabelArch] = cloudprovider.DefaultArch
m[corev1.LabelArchStable] = cloudprovider.DefaultArch
m[kubeletapis.LabelOS] = cloudprovider.DefaultOS
m[corev1.LabelOSStable] = cloudprovider.DefaultOS
m[corev1.LabelHostname] = nodeName
return m
}
// extract a predefined list of labels from the existing node
func extractNodeLabels(node *corev1.Node) map[string]string {
m := make(map[string]string)
if node.Labels == nil {
return m
}
setLabelIfNotEmpty(m, node.Labels, kubeletapis.LabelArch)
setLabelIfNotEmpty(m, node.Labels, corev1.LabelArchStable)
setLabelIfNotEmpty(m, node.Labels, kubeletapis.LabelOS)
setLabelIfNotEmpty(m, node.Labels, corev1.LabelOSStable)
setLabelIfNotEmpty(m, node.Labels, corev1.LabelInstanceType)
setLabelIfNotEmpty(m, node.Labels, corev1.LabelInstanceTypeStable)
setLabelIfNotEmpty(m, node.Labels, corev1.LabelZoneRegion)
setLabelIfNotEmpty(m, node.Labels, corev1.LabelZoneRegionStable)
setLabelIfNotEmpty(m, node.Labels, corev1.LabelZoneFailureDomain)
return m
}
func setLabelIfNotEmpty(to, from map[string]string, key string) {
if value := from[key]; value != "" {
to[key] = value
}
}

205
cluster-autoscaler/cloudprovider/clusterapi/clusterapi_nodegroup_test.go
View File

@ -32,6 +32,7 @@ import (
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/autoscaler/cluster-autoscaler/cloudprovider"
gpuapis "k8s.io/autoscaler/cluster-autoscaler/utils/gpu"
)
const (
@ -178,10 +179,6 @@ func TestNodeGroupNewNodeGroupConstructor(t *testing.T) {
t.Errorf("expected %q, got %q", expectedDebug, ng.Debug())
}
if _, err := ng.TemplateNodeInfo(); err != cloudprovider.ErrNotImplemented {
t.Error("expected error")
}
if exists := ng.Exist(); !exists {
t.Errorf("expected %t, got %t", true, exists)
}
@ -1194,3 +1191,203 @@ func TestNodeGroupWithFailedMachine(t *testing.T) {
}))
})
}
func TestNodeGroupTemplateNodeInfo(t *testing.T) {
enableScaleAnnotations := map[string]string{
nodeGroupMinSizeAnnotationKey: "1",
nodeGroupMaxSizeAnnotationKey: "10",
}
type testCaseConfig struct {
nodeLabels map[string]string
nodegroupLabels map[string]string
includeNodes bool
expectedErr error
expectedCapacity map[corev1.ResourceName]int64
expectedNodeLabels map[string]string
}
testCases := []struct {
name string
nodeGroupAnnotations map[string]string
config testCaseConfig
}{
{
name: "When the NodeGroup cannot scale from zero",
config: testCaseConfig{
expectedErr: cloudprovider.ErrNotImplemented,
},
},
{
name: "When the NodeGroup can scale from zero",
nodeGroupAnnotations: map[string]string{
memoryKey: "2048",
cpuKey: "2",
},
config: testCaseConfig{
expectedErr: nil,
expectedCapacity: map[corev1.ResourceName]int64{
corev1.ResourceCPU: 2,
corev1.ResourceMemory: 2048 * 1024 * 1024,
corev1.ResourcePods: 250,
gpuapis.ResourceNvidiaGPU: 0,
},
expectedNodeLabels: map[string]string{
"kubernetes.io/os": "linux",
"beta.kubernetes.io/os": "linux",
"kubernetes.io/arch": "amd64",
"beta.kubernetes.io/arch": "amd64",
},
},
},
{
name: "When the NodeGroup can scale from zero and the nodegroup adds labels to the Node",
nodeGroupAnnotations: map[string]string{
memoryKey: "2048",
cpuKey: "2",
},
config: testCaseConfig{
expectedErr: nil,
nodegroupLabels: map[string]string{
"nodeGroupLabel": "value",
"anotherLabel": "anotherValue",
},
expectedCapacity: map[corev1.ResourceName]int64{
corev1.ResourceCPU: 2,
corev1.ResourceMemory: 2048 * 1024 * 1024,
corev1.ResourcePods: 250,
gpuapis.ResourceNvidiaGPU: 0,
},
expectedNodeLabels: map[string]string{
"kubernetes.io/os": "linux",
"beta.kubernetes.io/os": "linux",
"kubernetes.io/arch": "amd64",
"beta.kubernetes.io/arch": "amd64",
"nodeGroupLabel": "value",
"anotherLabel": "anotherValue",
},
},
},
{
name: "When the NodeGroup can scale from zero and the Node still exists, it includes the known node labels",
nodeGroupAnnotations: map[string]string{
memoryKey: "2048",
cpuKey: "2",
},
config: testCaseConfig{
includeNodes: true,
expectedErr: nil,
nodeLabels: map[string]string{
"kubernetes.io/os": "windows",
"kubernetes.io/arch": "arm64",
"node.kubernetes.io/instance-type": "instance1",
"anotherLabel": "nodeValue", // This should not be copied as it is not a well known label
},
nodegroupLabels: map[string]string{
"nodeGroupLabel": "value",
"anotherLabel": "anotherValue",
},
expectedCapacity: map[corev1.ResourceName]int64{
corev1.ResourceCPU: 2,
corev1.ResourceMemory: 2048 * 1024 * 1024,
corev1.ResourcePods: 250,
gpuapis.ResourceNvidiaGPU: 0,
},
expectedNodeLabels: map[string]string{
"kubernetes.io/os": "windows",
"beta.kubernetes.io/os": "linux",
"kubernetes.io/arch": "arm64",
"beta.kubernetes.io/arch": "amd64",
"nodeGroupLabel": "value",
"anotherLabel": "anotherValue",
"node.kubernetes.io/instance-type": "instance1",
},
},
},
}
test := func(t *testing.T, testConfig *testConfig, config testCaseConfig) {
if testConfig.machineDeployment != nil {
unstructured.SetNestedStringMap(testConfig.machineDeployment.Object, config.nodegroupLabels, "spec", "template", "spec", "metadata", "labels")
} else {
unstructured.SetNestedStringMap(testConfig.machineSet.Object, config.nodegroupLabels, "spec", "template", "spec", "metadata", "labels")
}
if config.includeNodes {
for i := range testConfig.nodes {
testConfig.nodes[i].SetLabels(config.nodeLabels)
}
} else {
testConfig.nodes = []*corev1.Node{}
}
controller, stop := mustCreateTestController(t, testConfig)
defer stop()
nodegroups, err := controller.nodeGroups()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if l := len(nodegroups); l != 1 {
t.Fatalf("expected 1 nodegroup, got %d", l)
}
ng := nodegroups[0]
nodeInfo, err := ng.TemplateNodeInfo()
if config.expectedErr != nil {
if err != config.expectedErr {
t.Fatalf("expected error: %v, but got: %v", config.expectedErr, err)
}
return
}
nodeAllocatable := nodeInfo.Node().Status.Allocatable
nodeCapacity := nodeInfo.Node().Status.Capacity
for resource, expectedCapacity := range config.expectedCapacity {
if gotAllocatable, ok := nodeAllocatable[resource]; !ok {
t.Errorf("Expected allocatable to have resource %q, resource not found", resource)
} else if gotAllocatable.Value() != expectedCapacity {
t.Errorf("Expected allocatable %q: %+v, Got: %+v", resource, expectedCapacity, gotAllocatable.Value())
}
if gotCapactiy, ok := nodeCapacity[resource]; !ok {
t.Errorf("Expected capacity to have resource %q, resource not found", resource)
} else if gotCapactiy.Value() != expectedCapacity {
t.Errorf("Expected capacity %q: %+v, Got: %+v", resource, expectedCapacity, gotCapactiy.Value())
}
}
// expectedNodeLabels won't have the hostname label as it is randomized, so +1 to its length
if len(nodeInfo.Node().GetLabels()) != len(config.expectedNodeLabels)+1 {
t.Errorf("Expected node labels to have len: %d, but got: %d", len(config.expectedNodeLabels)+1, len(nodeInfo.Node().GetLabels()))
}
for key, value := range nodeInfo.Node().GetLabels() {
// Exclude the hostname label as it is randomized
if key != corev1.LabelHostname {
if value != config.expectedNodeLabels[key] {
t.Errorf("Expected node label %q: %q, Got: %q", key, config.expectedNodeLabels[key], value)
}
}
}
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
t.Run("MachineSet", func(t *testing.T) {
test(t, createMachineSetTestConfig(testNamespace, RandomString(6), RandomString(6), 10, cloudprovider.JoinStringMaps(enableScaleAnnotations, tc.nodeGroupAnnotations)),
tc.config,
)
})
t.Run("MachineDeployment", func(t *testing.T) {
test(
t,
createMachineDeploymentTestConfig(testNamespace, RandomString(6), RandomString(6), 10, cloudprovider.JoinStringMaps(enableScaleAnnotations, tc.nodeGroupAnnotations)),
tc.config,
)
})
})
}
}

47
cluster-autoscaler/cloudprovider/clusterapi/clusterapi_unstructured.go
View File

@ -23,6 +23,8 @@ import (
"time"
"github.com/pkg/errors"
apiv1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
@ -163,6 +165,51 @@ func (r unstructuredScalableResource) MarkMachineForDeletion(machine *unstructur
return updateErr
}
func (r unstructuredScalableResource) Labels() map[string]string {
labels, found, err := unstructured.NestedStringMap(r.unstructured.Object, "spec", "template", "spec", "metadata", "labels")
if !found || err != nil {
return nil
}
return labels
}
func (r unstructuredScalableResource) Taints() []apiv1.Taint {
taints, found, err := unstructured.NestedSlice(r.unstructured.Object, "spec", "template", "spec", "taints")
if !found || err != nil {
return nil
}
ret := make([]apiv1.Taint, len(taints))
for i, t := range taints {
if v, ok := t.(apiv1.Taint); ok {
ret[i] = v
} else {
// if we cannot convert the interface to a Taint, return early with zero value
return nil
}
}
return ret
}
func (r unstructuredScalableResource) CanScaleFromZero() bool {
return scaleFromZeroEnabled(r.unstructured.GetAnnotations())
}
func (r unstructuredScalableResource) InstanceCPUCapacity() (resource.Quantity, error) {
return parseCPUCapacity(r.unstructured.GetAnnotations())
}
func (r unstructuredScalableResource) InstanceMemoryCapacity() (resource.Quantity, error) {
return parseMemoryCapacity(r.unstructured.GetAnnotations())
}
func (r unstructuredScalableResource) InstanceGPUCapacity() (resource.Quantity, error) {
return parseGPUCapacity(r.unstructured.GetAnnotations())
}
func (r unstructuredScalableResource) InstanceMaxPodsCapacity() (resource.Quantity, error) {
return parseMaxPodsCapacity(r.unstructured.GetAnnotations())
}
func newUnstructuredScalableResource(controller *machineController, u *unstructured.Unstructured) (*unstructuredScalableResource, error) {
minSize, maxSize, err := parseScalingBounds(u.GetAnnotations())
if err != nil {

123
cluster-autoscaler/cloudprovider/clusterapi/clusterapi_unstructured_test.go
View File

@ -18,12 +18,11 @@ package clusterapi
import (
"context"
"fmt"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/client-go/tools/cache"
"testing"
"time"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/autoscaler/cluster-autoscaler/utils/units"
)
func TestSetSize(t *testing.T) {
@ -268,3 +267,117 @@ func TestSetSizeAndReplicas(t *testing.T) {
))
})
}
func TestAnnotations(t *testing.T) {
cpuQuantity := resource.MustParse("2")
memQuantity := resource.MustParse("1024")
gpuQuantity := resource.MustParse("1")
maxPodsQuantity := resource.MustParse("42")
annotations := map[string]string{
cpuKey: cpuQuantity.String(),
memoryKey: memQuantity.String(),
gpuKey: gpuQuantity.String(),
maxPodsKey: maxPodsQuantity.String(),
}
// convert the initial memory value from Mebibytes to bytes as this conversion happens internally
// when we use InstanceMemoryCapacity()
memVal, _ := memQuantity.AsInt64()
memQuantityAsBytes := resource.NewQuantity(memVal*units.MiB, resource.DecimalSI)
test := func(t *testing.T, testConfig *testConfig) {
controller, stop := mustCreateTestController(t, testConfig)
defer stop()
testResource := testConfig.machineSet
sr, err := newUnstructuredScalableResource(controller, testResource)
if err != nil {
t.Fatal(err)
}
if cpu, err := sr.InstanceCPUCapacity(); err != nil {
t.Fatal(err)
} else if cpuQuantity.Cmp(cpu) != 0 {
t.Errorf("expected %v, got %v", cpuQuantity, cpu)
}
if mem, err := sr.InstanceMemoryCapacity(); err != nil {
t.Fatal(err)
} else if memQuantityAsBytes.Cmp(mem) != 0 {
t.Errorf("expected %v, got %v", memQuantity, mem)
}
if gpu, err := sr.InstanceGPUCapacity(); err != nil {
t.Fatal(err)
} else if gpuQuantity.Cmp(gpu) != 0 {
t.Errorf("expected %v, got %v", gpuQuantity, gpu)
}
if maxPods, err := sr.InstanceMaxPodsCapacity(); err != nil {
t.Fatal(err)
} else if maxPodsQuantity.Cmp(maxPods) != 0 {
t.Errorf("expected %v, got %v", maxPodsQuantity, maxPods)
}
}
t.Run("MachineSet", func(t *testing.T) {
test(t, createMachineSetTestConfig(RandomString(6), RandomString(6), RandomString(6), 1, annotations))
})
}
func TestCanScaleFromZero(t *testing.T) {
testConfigs := []struct {
name string
annotations map[string]string
canScale bool
}{
{
"MachineSet can scale from zero",
map[string]string{
cpuKey: "1",
memoryKey: "1024",
},
true,
},
{
"MachineSet with missing CPU info cannot scale from zero",
map[string]string{
memoryKey: "1024",
},
false,
},
{
"MachineSet with missing Memory info cannot scale from zero",
map[string]string{
cpuKey: "1",
},
false,
},
{
"MachineSet with no information cannot scale from zero",
map[string]string{},
false,
},
}
for _, tc := range testConfigs {
t.Run(tc.name, func(t *testing.T) {
msTestConfig := createMachineSetTestConfig(RandomString(6), RandomString(6), RandomString(6), 1, tc.annotations)
controller, stop := mustCreateTestController(t, msTestConfig)
defer stop()
testResource := msTestConfig.machineSet
sr, err := newUnstructuredScalableResource(controller, testResource)
if err != nil {
t.Fatal(err)
}
canScale := sr.CanScaleFromZero()
if canScale != tc.canScale {
t.Errorf("expected %v, got %v", tc.canScale, canScale)
}
})
}
}

74
cluster-autoscaler/cloudprovider/clusterapi/clusterapi_utils.go
View File

@ -22,8 +22,21 @@ import (
"strings"
"github.com/pkg/errors"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/autoscaler/cluster-autoscaler/utils/units"
)
const (
deprecatedNodeGroupMinSizeAnnotationKey = "cluster.k8s.io/cluster-api-autoscaler-node-group-min-size"
deprecatedNodeGroupMaxSizeAnnotationKey = "cluster.k8s.io/cluster-api-autoscaler-node-group-max-size"
deprecatedClusterNameLabel = "cluster.k8s.io/cluster-name"
cpuKey = "machine.openshift.io/vCPU"
memoryKey = "machine.openshift.io/memoryMb"
gpuKey = "machine.openshift.io/GPU"
maxPodsKey = "machine.openshift.io/maxPods"
)
var (
@ -50,22 +63,7 @@ var (
// machine set has a non-integral max annotation value.
errInvalidMaxAnnotation = errors.New("invalid max annotation")
// machineDeleteAnnotationKey is the annotation used by cluster-api to indicate
// that a machine should be deleted. Because this key can be affected by the
// CAPI_GROUP env variable, it is initialized here.
machineDeleteAnnotationKey = getMachineDeleteAnnotationKey()
// machineAnnotationKey is the annotation used by the cluster-api on Node objects
// to specify the name of the related Machine object. Because this can be affected
// by the CAPI_GROUP env variable, it is initialized here.
machineAnnotationKey = getMachineAnnotationKey()
// nodeGroupMinSizeAnnotationKey and nodeGroupMaxSizeAnnotationKey are the keys
// used in MachineSet and MachineDeployment annotations to specify the limits
// for the node group. Because the keys can be affected by the CAPI_GROUP env
// variable, they are initialized here.
nodeGroupMinSizeAnnotationKey = getNodeGroupMinSizeAnnotationKey()
nodeGroupMaxSizeAnnotationKey = getNodeGroupMaxSizeAnnotationKey()
zeroQuantity = resource.MustParse("0")
)
type normalizedProviderID string
@ -157,6 +155,50 @@ func normalizedProviderString(s string) normalizedProviderID {
return normalizedProviderID(split[len(split)-1])
}
func scaleFromZeroEnabled(annotations map[string]string) bool {
cpu := annotations[cpuKey]
mem := annotations[memoryKey]
if cpu != "" && mem != "" {
return true
}
return false
}
func parseKey(annotations map[string]string, key string) (resource.Quantity, error) {
if val, exists := annotations[key]; exists && val != "" {
return resource.ParseQuantity(val)
}
return zeroQuantity.DeepCopy(), nil
}
func parseCPUCapacity(annotations map[string]string) (resource.Quantity, error) {
return parseKey(annotations, cpuKey)
}
func parseMemoryCapacity(annotations map[string]string) (resource.Quantity, error) {
// The value for the memoryKey is expected to be an integer representing Mebibytes. e.g. "1024".
// https://www.iec.ch/si/binary.htm
val, exists := annotations[memoryKey]
if exists && val != "" {
valInt, err := strconv.ParseInt(val, 10, 0)
if err != nil {
return zeroQuantity.DeepCopy(), fmt.Errorf("value %q from annotation %q expected to be an integer: %v", val, memoryKey, err)
}
// Convert from Mebibytes to bytes
return *resource.NewQuantity(valInt*units.MiB, resource.DecimalSI), nil
}
return zeroQuantity.DeepCopy(), nil
}
func parseGPUCapacity(annotations map[string]string) (resource.Quantity, error) {
return parseKey(annotations, gpuKey)
}
func parseMaxPodsCapacity(annotations map[string]string) (resource.Quantity, error) {
return parseKey(annotations, maxPodsKey)
}
func clusterNameFromResource(r *unstructured.Unstructured) string {
// Use Spec.ClusterName if defined (only available on v1alpha3+ types)
clusterName, found, err := unstructured.NestedString(r.Object, "spec", "clusterName")

216
cluster-autoscaler/cloudprovider/clusterapi/clusterapi_utils_test.go
View File

@ -23,8 +23,10 @@ import (
"strings"
"testing"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/autoscaler/cluster-autoscaler/utils/units"
)
const (
@ -429,6 +431,220 @@ func TestUtilNormalizedProviderID(t *testing.T) {
}
}
func TestScaleFromZeroEnabled(t *testing.T) {
for _, tc := range []struct {
description string
enabled bool
annotations map[string]string
}{{
description: "nil annotations",
enabled: false,
}, {
description: "empty annotations",
annotations: map[string]string{},
enabled: false,
}, {
description: "non-matching annotation",
annotations: map[string]string{"foo": "bar"},
enabled: false,
}, {
description: "matching key, incomplete annotations",
annotations: map[string]string{
"foo": "bar",
cpuKey: "1",
gpuKey: "2",
},
enabled: false,
}, {
description: "matching key, complete annotations",
annotations: map[string]string{
"foo": "bar",
cpuKey: "1",
memoryKey: "2",
},
enabled: true,
}} {
t.Run(tc.description, func(t *testing.T) {
got := scaleFromZeroEnabled(tc.annotations)
if tc.enabled != got {
t.Errorf("expected %t, got %t", tc.enabled, got)
}
})
}
}
func TestParseCPUCapacity(t *testing.T) {
for _, tc := range []struct {
description string
annotations map[string]string
expectedQuantity resource.Quantity
expectedError bool
}{{
description: "nil annotations",
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: false,
}, {
description: "empty annotations",
annotations: map[string]string{},
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: false,
}, {
description: "bad quantity",
annotations: map[string]string{cpuKey: "not-a-quantity"},
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: true,
}, {
description: "valid quantity with units",
annotations: map[string]string{cpuKey: "123m"},
expectedError: false,
expectedQuantity: resource.MustParse("123m"),
}, {
description: "valid quantity without units",
annotations: map[string]string{cpuKey: "1"},
expectedError: false,
expectedQuantity: resource.MustParse("1"),
}, {
description: "valid fractional quantity without units",
annotations: map[string]string{cpuKey: "0.1"},
expectedError: false,
expectedQuantity: resource.MustParse("0.1"),
}} {
t.Run(tc.description, func(t *testing.T) {
got, err := parseCPUCapacity(tc.annotations)
if tc.expectedError && err == nil {
t.Fatal("expected an error")
}
if tc.expectedQuantity.Cmp(got) != 0 {
t.Errorf("expected %v, got %v", tc.expectedQuantity.String(), got.String())
}
})
}
}
func TestParseMemoryCapacity(t *testing.T) {
for _, tc := range []struct {
description string
annotations map[string]string
expectedQuantity resource.Quantity
expectedError bool
}{{
description: "nil annotations",
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: false,
}, {
description: "empty annotations",
annotations: map[string]string{},
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: false,
}, {
description: "bad quantity",
annotations: map[string]string{memoryKey: "not-a-quantity"},
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: true,
}, {
description: "valid quantity",
annotations: map[string]string{memoryKey: "456"},
expectedError: false,
expectedQuantity: *resource.NewQuantity(456*units.MiB, resource.DecimalSI),
}, {
description: "quantity with unit type (Mi)",
annotations: map[string]string{memoryKey: "456Mi"},
expectedError: true,
expectedQuantity: zeroQuantity.DeepCopy(),
}, {
description: "quantity with unit type (Gi)",
annotations: map[string]string{memoryKey: "8Gi"},
expectedError: true,
expectedQuantity: zeroQuantity.DeepCopy(),
}} {
t.Run(tc.description, func(t *testing.T) {
got, err := parseMemoryCapacity(tc.annotations)
if tc.expectedError && err == nil {
t.Fatal("expected an error")
}
if tc.expectedQuantity.Cmp(got) != 0 {
t.Errorf("expected %v, got %v", tc.expectedQuantity.String(), got.String())
}
})
}
}
func TestParseGPUCapacity(t *testing.T) {
for _, tc := range []struct {
description string
annotations map[string]string
expectedQuantity resource.Quantity
expectedError bool
}{{
description: "nil annotations",
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: false,
}, {
description: "empty annotations",
annotations: map[string]string{},
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: false,
}, {
description: "bad quantity",
annotations: map[string]string{gpuKey: "not-a-quantity"},
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: true,
}, {
description: "valid quantity",
annotations: map[string]string{gpuKey: "13"},
expectedError: false,
expectedQuantity: resource.MustParse("13"),
}} {
t.Run(tc.description, func(t *testing.T) {
got, err := parseGPUCapacity(tc.annotations)
if tc.expectedError && err == nil {
t.Fatal("expected an error")
}
if tc.expectedQuantity.Cmp(got) != 0 {
t.Errorf("expected %v, got %v", tc.expectedQuantity.String(), got.String())
}
})
}
}
func TestParseMaxPodsCapacity(t *testing.T) {
for _, tc := range []struct {
description string
annotations map[string]string
expectedQuantity resource.Quantity
expectedError bool
}{{
description: "nil annotations",
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: false,
}, {
description: "empty annotations",
annotations: map[string]string{},
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: false,
}, {
description: "bad quantity",
annotations: map[string]string{maxPodsKey: "not-a-quantity"},
expectedQuantity: zeroQuantity.DeepCopy(),
expectedError: true,
}, {
description: "valid quantity",
annotations: map[string]string{maxPodsKey: "13"},
expectedError: false,
expectedQuantity: resource.MustParse("13"),
}} {
t.Run(tc.description, func(t *testing.T) {
got, err := parseMaxPodsCapacity(tc.annotations)
if tc.expectedError && err == nil {
t.Fatal("expected an error")
}
if tc.expectedQuantity.Cmp(got) != 0 {
t.Errorf("expected %v, got %v", tc.expectedQuantity.String(), got.String())
}
})
}
}
func Test_clusterNameFromResource(t *testing.T) {
for _, tc := range []struct {
name string