[Autoscaling] Add Elasticsearch autoscaling controller

cmd/manager/main.go

@@ -29,6 +29,7 @@ import (
 	"github.com/elastic/cloud-on-k8s/pkg/controller/apmserver"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/association"
 	associationctl "github.com/elastic/cloud-on-k8s/pkg/controller/association/controller"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/autoscaling"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/beat"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/certificates"
 	"github.com/elastic/cloud-on-k8s/pkg/controller/common/container"
@@ -643,6 +644,7 @@ func registerControllers(mgr manager.Manager, params operator.Parameters, access
 	}{
 		{name: "APMServer", registerFunc: apmserver.Add},
 		{name: "Elasticsearch", registerFunc: elasticsearch.Add},
+		{name: "ElasticsearchAutoscaling", registerFunc: autoscaling.Add},
 		{name: "Kibana", registerFunc: kibana.Add},
 		{name: "EnterpriseSearch", registerFunc: enterprisesearch.Add},
 		{name: "Beats", registerFunc: beat.Add},

pkg/controller/autoscaling/elasticsearch.go

@@ -0,0 +1,36 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package autoscaling
+
+import (
+	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/autoscaling/elasticsearch"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/common/operator"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+const (
+	controllerName = "elasticsearch-autoscaling"
+)
+
+// Add creates a new Elasticsearch autoscaling controller and adds it to the Manager with default RBAC.
+// The Manager will set fields on the Controller and Start it when the Manager is Started.
+func Add(mgr manager.Manager, p operator.Parameters) error {
+	r := elasticsearch.NewReconciler(mgr, p)
+	c, err := common.NewController(mgr, controllerName, r, p)
+	if err != nil {
+		return err
+	}
+	// Watch for changes on Elasticsearch clusters.
+	if err := c.Watch(
+		&source.Kind{Type: &esv1.Elasticsearch{}}, &handler.EnqueueRequestForObject{},
+	); err != nil {
+		return err
+	}
+	return nil
+}

pkg/controller/autoscaling/elasticsearch/autoscaler/autoscaler.go

@@ -0,0 +1,67 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package autoscaler
+
+import (
+	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/autoscaling/elasticsearch/resources"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/autoscaling/elasticsearch/status"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/volume"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+)
+
+// GetResources calculates the resources required by all the NodeSets managed by a same autoscaling policy.
+func (ctx *Context) GetResources() resources.NodeSetsResources {
+	// 1. Scale vertically, calculating the resources for each node managed by the autoscaling policy in the context.
+	desiredNodeResources := ctx.scaleVertically()
+	ctx.Log.Info(
+		"Vertical autoscaler",
+		"state", "online",
+		"policy", ctx.AutoscalingSpec.Name,
+		"scope", "node",
+		"nodesets", ctx.NodeSets.Names(),
+		"resources", desiredNodeResources.ToInt64(),
+		"required_capacity", ctx.RequiredCapacity,
+	)
+
+	// 2. Scale horizontally by adding nodes to meet the resource requirements.
+	return ctx.scaleHorizontally(desiredNodeResources)
+}
+
+// scaleVertically calculates the desired resources for all the nodes managed by the same autoscaling policy, given the requested
+// capacity returned by the Elasticsearch autoscaling API and the AutoscalingSpec specified by the user.
+// It attempts to scale all the resources vertically until the required resources are provided or the limits set by the user are reached.
+func (ctx *Context) scaleVertically() resources.NodeResources {
+	// All resources can be computed "from scratch", without knowing the previous values.
+	// This is however not true for storage. Storage can't be scaled down, current storage capacity must be considered
+	// as a hard min. limit. This storage limit must be taken into consideration when computing the desired resources.
+	currentStorage := getStorage(ctx.AutoscalingSpec, ctx.CurrentAutoscalingStatus)
+	return ctx.nodeResources(
+		int64(ctx.AutoscalingSpec.NodeCount.Min),
+		currentStorage,
+	)
+}
+
+// getStorage returns the min. storage capacity that should be used by the autoscaling algorithm.
+// The value is the max. value of either:
+// * the current value in the status
+// * the min. value set by the user in the autoscaling spec.
+func getStorage(autoscalingSpec esv1.AutoscalingPolicySpec, currentAutoscalingStatus status.Status) resource.Quantity {
+	// If no storage spec is defined in the autoscaling status we return the default volume size.
+	storage := volume.DefaultPersistentVolumeSize.DeepCopy()
+	// Always adjust to the min value specified by the user in the limits.
+	if autoscalingSpec.IsStorageDefined() {
+		storage = autoscalingSpec.Storage.Min
+	}
+	// If a storage value is stored in the status then reuse it.
+	if currentResourcesInStatus, exists := currentAutoscalingStatus.CurrentResourcesForPolicy(autoscalingSpec.Name); exists && currentResourcesInStatus.HasRequest(corev1.ResourceStorage) {
+		storageInStatus := currentResourcesInStatus.GetRequest(corev1.ResourceStorage)
+		if storageInStatus.Cmp(storage) > 0 {
+			storage = storageInStatus
+		}
+	}
+	return storage
+}

pkg/controller/autoscaling/elasticsearch/autoscaler/autoscaler_test.go

@@ -0,0 +1,261 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package autoscaler
+
+import (
+	"testing"
+
+	esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/autoscaling/elasticsearch/resources"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/autoscaling/elasticsearch/status"
+	"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/client"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/apimachinery/pkg/api/resource"
+)
+
+func Test_applyScaleDecision(t *testing.T) {
+	defaultNodeSets := esv1.NodeSetList{{
+		Name: "default",
+	}}
+	type args struct {
+		currentNodeSets  esv1.NodeSetList
+		nodeSetsStatus   status.Status
+		requiredCapacity client.AutoscalingCapacityInfo
+		policy           esv1.AutoscalingPolicySpec
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    resources.NodeSetsResources
+		wantErr bool
+	}{
+		{
+			name: "Scale both vertically and horizontally to fulfil storage capacity request",
+			args: args{
+				currentNodeSets: defaultNodeSets,
+				nodeSetsStatus: status.Status{AutoscalingPolicyStatuses: []status.AutoscalingPolicyStatus{{
+					Name:                   "my-autoscaling-policy",
+					NodeSetNodeCount:       []resources.NodeSetNodeCount{{Name: "default", NodeCount: 3}},
+					ResourcesSpecification: resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("3G"), corev1.ResourceStorage: q("1Gi")}}}},
+				},
+				requiredCapacity: newRequiredCapacityBuilder().
+					nodeMemory("3Gi").nodeStorage("8Gi").
+					tierMemory("9Gi").tierStorage("50Gi").
+					build(),
+				policy: NewAutoscalingSpecBuilder("my-autoscaling-policy").WithNodeCounts(3, 6).WithMemory("3Gi", "4Gi").WithStorage("5Gi", "10Gi").Build(),
+			},
+			want: resources.NodeSetsResources{
+				Name:             "my-autoscaling-policy",
+				NodeSetNodeCount: []resources.NodeSetNodeCount{{Name: "default", NodeCount: 5}},
+				NodeResources:    resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("3Gi"), corev1.ResourceStorage: q("10Gi")}},
+			},
+		},
+		{
+			name: "Scale existing nodes vertically",
+			args: args{
+				currentNodeSets: defaultNodeSets,
+				nodeSetsStatus: status.Status{AutoscalingPolicyStatuses: []status.AutoscalingPolicyStatus{{
+					Name:                   "my-autoscaling-policy",
+					NodeSetNodeCount:       []resources.NodeSetNodeCount{{Name: "default", NodeCount: 3}},
+					ResourcesSpecification: resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("3G"), corev1.ResourceStorage: q("1Gi")}}}},
+				},
+				requiredCapacity: newRequiredCapacityBuilder().
+					nodeMemory("6G").
+					tierMemory("15G").
+					build(),
+				policy: NewAutoscalingSpecBuilder("my-autoscaling-policy").WithNodeCounts(3, 6).WithMemory("5G", "8G").Build(),
+			},
+			want: resources.NodeSetsResources{
+				Name:             "my-autoscaling-policy",
+				NodeSetNodeCount: []resources.NodeSetNodeCount{{Name: "default", NodeCount: 3}},
+				NodeResources:    resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("6Gi")}},
+			},
+		},
+		{
+			name: "Do not scale down storage capacity",
+			args: args{
+				currentNodeSets: defaultNodeSets,
+				nodeSetsStatus: status.Status{AutoscalingPolicyStatuses: []status.AutoscalingPolicyStatus{{
+					Name:                   "my-autoscaling-policy",
+					NodeSetNodeCount:       []resources.NodeSetNodeCount{{Name: "default", NodeCount: 3}},
+					ResourcesSpecification: resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("4G"), corev1.ResourceStorage: q("10G")}}}},
+				},
+				requiredCapacity: newRequiredCapacityBuilder().
+					nodeMemory("6G").
+					tierMemory("15G").
+					nodeStorage("1Gi").
+					tierStorage("3Gi").
+					build(),
+				policy: NewAutoscalingSpecBuilder("my-autoscaling-policy").WithNodeCounts(3, 6).WithMemory("5G", "8G").WithStorage("1G", "20G").Build(),
+			},
+			want: resources.NodeSetsResources{
+				Name:             "my-autoscaling-policy",
+				NodeSetNodeCount: []resources.NodeSetNodeCount{{Name: "default", NodeCount: 3}},
+				NodeResources:    resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("6Gi"), corev1.ResourceStorage: q("10G")}},
+			},
+		},
+		{
+			name: "Scale existing nodes vertically up to the tier limit",
+			args: args{
+				currentNodeSets: defaultNodeSets,
+				nodeSetsStatus: status.Status{AutoscalingPolicyStatuses: []status.AutoscalingPolicyStatus{{
+					Name:                   "my-autoscaling-policy",
+					NodeSetNodeCount:       []resources.NodeSetNodeCount{{Name: "default", NodeCount: 3}},
+					ResourcesSpecification: resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("4G"), corev1.ResourceStorage: q("1Gi")}}}},
+				},
+				requiredCapacity: newRequiredCapacityBuilder().
+					nodeMemory("6G").
+					tierMemory("21G").
+					build(),
+				policy: NewAutoscalingSpecBuilder("my-autoscaling-policy").WithNodeCounts(3, 6).WithMemory("5G", "8G").Build(),
+			},
+			want: resources.NodeSetsResources{
+				Name:             "my-autoscaling-policy",
+				NodeSetNodeCount: []resources.NodeSetNodeCount{{Name: "default", NodeCount: 3}},
+				NodeResources:    resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("7Gi")}},
+			},
+		},
+		{
+			name: "Scale both vertically and horizontally",
+			args: args{
+				currentNodeSets: defaultNodeSets,
+				nodeSetsStatus: status.Status{AutoscalingPolicyStatuses: []status.AutoscalingPolicyStatus{{
+					Name:                   "my-autoscaling-policy",
+					NodeSetNodeCount:       []resources.NodeSetNodeCount{{Name: "default", NodeCount: 3}},
+					ResourcesSpecification: resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("4G"), corev1.ResourceStorage: q("1Gi")}}}},
+				},
+				requiredCapacity: newRequiredCapacityBuilder().
+					nodeMemory("6G").
+					tierMemory("48G").
+					build(),
+				policy: NewAutoscalingSpecBuilder("my-autoscaling-policy").WithNodeCounts(3, 6).WithMemory("5G", "8G").Build(),
+			},
+			want: resources.NodeSetsResources{
+				Name:             "my-autoscaling-policy",
+				NodeSetNodeCount: []resources.NodeSetNodeCount{{Name: "default", NodeCount: 6}},
+				NodeResources:    resources.NodeResources{Requests: map[corev1.ResourceName]resource.Quantity{corev1.ResourceMemory: q("8G")}},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := Context{
+				Log:                      logTest,
+				AutoscalingSpec:          tt.args.policy,
+				NodeSets:                 tt.args.currentNodeSets,
+				CurrentAutoscalingStatus: tt.args.nodeSetsStatus,
+				RequiredCapacity:         tt.args.requiredCapacity,
+				StatusBuilder:            status.NewAutoscalingStatusBuilder(),
+			}
+			if got := ctx.GetResources(); !equality.Semantic.DeepEqual(got, tt.want) {
+				t.Errorf("autoscaler.GetResources() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+// - AutoscalingSpec builder
+
+type AutoscalingSpecBuilder struct {
+	name                       string
+	nodeCountMin, nodeCountMax int32
+	cpu, memory, storage       *esv1.QuantityRange
+}
+
+func NewAutoscalingSpecBuilder(name string) *AutoscalingSpecBuilder {
+	return &AutoscalingSpecBuilder{name: name}
+}
+
+func (asb *AutoscalingSpecBuilder) WithNodeCounts(min, max int) *AutoscalingSpecBuilder {
+	asb.nodeCountMin = int32(min)
+	asb.nodeCountMax = int32(max)
+	return asb
+}
+
+func (asb *AutoscalingSpecBuilder) WithMemory(min, max string) *AutoscalingSpecBuilder {
+	asb.memory = &esv1.QuantityRange{
+		Min: resource.MustParse(min),
+		Max: resource.MustParse(max),
+	}
+	return asb
+}
+
+func (asb *AutoscalingSpecBuilder) WithStorage(min, max string) *AutoscalingSpecBuilder {
+	asb.storage = &esv1.QuantityRange{
+		Min: resource.MustParse(min),
+		Max: resource.MustParse(max),
+	}
+	return asb
+}
+
+func (asb *AutoscalingSpecBuilder) WithCPU(min, max string) *AutoscalingSpecBuilder {
+	asb.cpu = &esv1.QuantityRange{
+		Min: resource.MustParse(min),
+		Max: resource.MustParse(max),
+	}
+	return asb
+}
+
+func (asb *AutoscalingSpecBuilder) Build() esv1.AutoscalingPolicySpec {
+	return esv1.AutoscalingPolicySpec{
+		NamedAutoscalingPolicy: esv1.NamedAutoscalingPolicy{
+			Name: asb.name,
+		},
+		AutoscalingResources: esv1.AutoscalingResources{
+			CPU:     asb.cpu,
+			Memory:  asb.memory,
+			Storage: asb.storage,
+			NodeCount: esv1.CountRange{
+				Min: asb.nodeCountMin,
+				Max: asb.nodeCountMax,
+			},
+		},
+	}
+}
+
+// - PolicyCapacityInfo builder
+
+type requiredCapacityBuilder struct {
+	client.AutoscalingCapacityInfo
+}
+
+func newRequiredCapacityBuilder() *requiredCapacityBuilder {
+	return &requiredCapacityBuilder{}
+}
+
+func ptr(q int64) *client.AutoscalingCapacity {
+	v := client.AutoscalingCapacity(q)
+	return &v
+}
+
+func (rcb *requiredCapacityBuilder) build() client.AutoscalingCapacityInfo {
+	return rcb.AutoscalingCapacityInfo
+}
+
+func (rcb *requiredCapacityBuilder) nodeMemory(m string) *requiredCapacityBuilder {
+	rcb.Node.Memory = ptr(value(m))
+	return rcb
+}
+
+func (rcb *requiredCapacityBuilder) tierMemory(m string) *requiredCapacityBuilder {
+	rcb.Total.Memory = ptr(value(m))
+	return rcb
+}
+
+func (rcb *requiredCapacityBuilder) nodeStorage(m string) *requiredCapacityBuilder {
+	rcb.Node.Storage = ptr(value(m))
+	return rcb
+}
+
+func (rcb *requiredCapacityBuilder) tierStorage(m string) *requiredCapacityBuilder {
+	rcb.Total.Storage = ptr(value(m))
+	return rcb
+}
+
+func value(v string) int64 {
+	q := resource.MustParse(v)
+	return q.Value()
+}