Implement fallback strategy when there is no metrics

Add new field to the spec - `fallbackStrategy` that controlls what
amount of resources should be assigned to the instance when it's
impossible to run estimations due to missing metrics.

Only two modes are implemented at the moment - min and max, with default
being min (to preserve the existing behaviour).

Fallback strategy assigns min/max value based on either of 2 inputs:

* if there are more than half of the instances are running (and more
  than 10 in total), find min/max among those instances and use it as
  fallback value for new estimations.

* otherwise, use min/max from the container policy set in the spec:
  ```
	containerPolicies:
	- containerName: name
		minAllowed:
			cpu: 3
			memory: 12Gi
		maxAllowed:
			cpu: 8
			memory: 12Gi
  ```

Author: lwolf

api/v1/consumer_types.go

@@ -22,6 +22,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+// FallbackStrategy specifies how resources should be assigned
+// in case of missing Prometheus metrics.
+// Currently, two strategies are implemented: min and max
+type FallbackStrategy string
+
 type AutoscalerType string
 
 // ContainerScalingMode controls whether autoscaler is enabled for a specific
@@ -37,6 +42,9 @@ const (
 	ContainerScalingModeAuto ContainerScalingMode = "Auto"
 	// ContainerScalingModeOff means autoscaling is disabled for a container.
 	ContainerScalingModeOff ContainerScalingMode = "Off"
+
+	FallbackStrategyMin = "min"
+	FallbackStrategyMax = "max"
 )
 
 // ConsumerSpec defines the desired state of Consumer
@@ -110,6 +118,8 @@ type PrometheusAutoscalerSpec struct {
 	Offset      OffsetQuerySpec      `json:"offset"`
 	Production  ProductionQuerySpec  `json:"production"`
 	Consumption ConsumptionQuerySpec `json:"consumption"`
+	// +optional
+	FallbackStrategy *FallbackStrategy `json:"fallbackStrategy,omitempty"`
 
 	RatePerCore *int64            `json:"ratePerCore"`
 	RamPerCore  resource.Quantity `json:"ramPerCore"`

api/v1/zz_generated.deepcopy.go

@@ -98,6 +98,11 @@ spec:
                         x-kubernetes-int-or-string: true
                       criticalLag:
                         type: string
+                      fallbackStrategy:
+                        description: 'FallbackStrategy specifies how resources should
+                          be assigned in case of missing Prometheus metrics. Currently,
+                          two strategies are implemented: min and max'
+                        type: string
                       minSyncPeriod:
                         type: string
                       offset:

config/crd/bases/konsumerator.lwolf.org_consumers.yaml

@@ -13,10 +13,6 @@ import (
 	konsumeratorv1 "github.com/lwolf/konsumerator/api/v1"
 )
 
-func TestEstimateResources(t *testing.T) {
-
-}
-
 func TestNewConsumerOperator(t *testing.T) {
 	testCases := []struct {
 		name           string

controllers/consumer_test.go

@@ -84,6 +84,33 @@ func (o *operator) init(consumer *konsumeratorv1.Consumer, managedInstances apps
 	o.toEstimateInstances = make([]*appsv1.Deployment, 0)
 	o.toCreateInstances = make([]*appsv1.Deployment, 0)
 
+	fallbackValue := make(map[string]*corev1.ResourceRequirements)
+	var strategy konsumeratorv1.FallbackStrategy
+	if o.consumer.Spec.Autoscaler.Prometheus.FallbackStrategy != nil {
+		strategy = *o.consumer.Spec.Autoscaler.Prometheus.FallbackStrategy
+	} else {
+		strategy = konsumeratorv1.FallbackStrategyMin
+	}
+	// calculate min/max request.cpu value to use it as "fallback" value when metrics are missing,
+	// use it only when number of instances >10 and > 50% of instances are up and running, otherwise hard fail to min/max from the policy
+	if len(o.assignments) > 10 && len(managedInstances.Items) > (len(o.assignments)/2) {
+		fallbackValue = calculateFallbackFromRunningInstances(managedInstances.Items, strategy)
+	} else {
+		/*
+			In case metrics are missing, use fallback strategy to assign
+			min/max resources from the container policy
+			  containerPolicies:
+			    - containerName: name
+			      minAllowed:
+			        cpu: 3
+			        memory: 12Gi
+			      maxAllowed:
+			        cpu: 8
+			        memory: 12Gi
+		*/
+		fallbackValue = calculateFallbackFromPolicy(consumer.Spec.ResourcePolicy, strategy)
+	}
+
 	o.limiter = limiters.NewInstanceLimiter(consumer.Spec.ResourcePolicy, o.log)
 	o.globalLimiter = limiters.NewGlobalLimiter(consumer.Spec.ResourcePolicy, o.usedResources, o.log)
 
@@ -97,7 +124,8 @@ func (o *operator) init(consumer *konsumeratorv1.Consumer, managedInstances apps
 	if o.consumer.Spec.Autoscaler == nil || o.consumer.Spec.Autoscaler.Prometheus == nil {
 		return fmt.Errorf("Spec.Autoscaler.Prometheus can't be empty")
 	}
-	o.predictor = predictors.NewNaivePredictor(o.mp, o.consumer.Spec.Autoscaler.Prometheus)
+
+	o.predictor = predictors.NewNaivePredictor(o.mp, o.consumer.Spec.Autoscaler.Prometheus, fallbackValue, o.log)
 
 	o.syncInstanceStates(managedInstances)
 

controllers/operator.go

@@ -198,3 +198,56 @@ func sumAllRequestedResourcesInPod(containerSpecs []corev1.Container) *corev1.Re
 	}
 	return &result
 }
+
+func calculateFallbackFromPolicy(policy *konsumeratorv1.ResourcePolicy, strategy konsumeratorv1.FallbackStrategy) map[string]*corev1.ResourceRequirements {
+	if policy == nil {
+		return nil
+	}
+	res := make(map[string]*corev1.ResourceRequirements)
+	switch strategy {
+	case konsumeratorv1.FallbackStrategyMax:
+		for i := range policy.ContainerPolicies {
+			cp := policy.ContainerPolicies[i]
+			res[cp.ContainerName] = &corev1.ResourceRequirements{
+				Requests: cp.MaxAllowed,
+				Limits:   cp.MaxAllowed,
+			}
+		}
+	case konsumeratorv1.FallbackStrategyMin:
+		fallthrough
+	default:
+		for i := range policy.ContainerPolicies {
+			cp := policy.ContainerPolicies[i]
+			res[cp.ContainerName] = &corev1.ResourceRequirements{
+				Requests: cp.MinAllowed,
+				Limits:   cp.MinAllowed,
+			}
+		}
+	}
+	return res
+}
+
+func calculateFallbackFromRunningInstances(instances []appsv1.Deployment, strategy konsumeratorv1.FallbackStrategy) map[string]*corev1.ResourceRequirements {
+	res := make(map[string]*corev1.ResourceRequirements)
+	for i := range instances {
+		deploy := instances[i]
+		for c := range deploy.Spec.Template.Spec.Containers {
+			container := deploy.Spec.Template.Spec.Containers[c]
+			if _, ok := res[container.Name]; !ok {
+				res[container.Name] = &container.Resources
+				continue
+			}
+			switch strategy {
+			case konsumeratorv1.FallbackStrategyMin:
+				if container.Resources.Requests.Cpu().MilliValue() < res[container.Name].Requests.Cpu().MilliValue() {
+					res[container.Name] = &container.Resources
+				}
+			case konsumeratorv1.FallbackStrategyMax:
+				if container.Resources.Requests.Cpu().MilliValue() > res[container.Name].Requests.Cpu().MilliValue() {
+					res[container.Name] = &container.Resources
+				}
+			}
+		}
+	}
+	return res
+}

controllers/utils.go

@@ -1,6 +1,8 @@
 package controllers
 
 import (
+	"github.com/lwolf/konsumerator/pkg/helpers/tests"
+	appsv1 "k8s.io/api/apps/v1"
 	"testing"
 	"time"
 
@@ -138,8 +140,6 @@ func TestUpdateStatusAnnotations(t *testing.T) {
 
 }
 
-func TestResourceRequirementsDiff(t *testing.T) {}
-func TestResourceRequirementsSum(t *testing.T)  {}
 func TestResourceListSum(t *testing.T) {
 	testCases := map[string]struct {
 		a   corev1.ResourceList
@@ -189,4 +189,147 @@ func isResourceListEqual(a, b corev1.ResourceList) bool {
 	return true
 }
 
-func TestResourceListDiff(t *testing.T) {}
+func TestCalculateFallbackFromPolicy(t *testing.T) {
+	testCases := map[string]struct {
+		strategy      konsumeratorv1.FallbackStrategy
+		containerName []string
+		policy        *konsumeratorv1.ResourcePolicy
+		expFallback   map[string]*corev1.ResourceRequirements
+	}{
+		"should return min allowed by containerName in the map": {
+			strategy:      konsumeratorv1.FallbackStrategyMin,
+			containerName: []string{"test"},
+			policy: &konsumeratorv1.ResourcePolicy{ContainerPolicies: []konsumeratorv1.ContainerResourcePolicy{
+				tests.NewContainerResourcePolicy("test", "100m", "100M", "2", "150M"),
+			}},
+			expFallback: map[string]*corev1.ResourceRequirements{"test": tests.NewResourceRequirements("100m", "100M", "100m", "100M")},
+		},
+		"should return max allowed by containerName in the map": {
+			strategy:      konsumeratorv1.FallbackStrategyMax,
+			containerName: []string{"test"},
+			policy: &konsumeratorv1.ResourcePolicy{ContainerPolicies: []konsumeratorv1.ContainerResourcePolicy{
+				tests.NewContainerResourcePolicy("test", "100m", "100M", "2", "150M"),
+			}},
+			expFallback: map[string]*corev1.ResourceRequirements{"test": tests.NewResourceRequirements("2", "150M", "2", "150M")},
+		},
+		"multi container setup should return correct values": {
+			strategy:      konsumeratorv1.FallbackStrategyMin,
+			containerName: []string{"test", "test2"},
+			policy: &konsumeratorv1.ResourcePolicy{ContainerPolicies: []konsumeratorv1.ContainerResourcePolicy{
+				tests.NewContainerResourcePolicy("test", "100m", "100M", "2", "150M"),
+				tests.NewContainerResourcePolicy("test2", "1", "1G", "5", "2G"),
+			}},
+			expFallback: map[string]*corev1.ResourceRequirements{
+				"test":  tests.NewResourceRequirements("100m", "100M", "100m", "100M"),
+				"test2": tests.NewResourceRequirements("1", "1G", "1", "1G"),
+			},
+		},
+		"should return nil if no such policy exists": {
+			strategy:      konsumeratorv1.FallbackStrategyMax,
+			containerName: []string{"test"},
+			policy:        nil,
+			expFallback:   nil,
+		},
+	}
+	for testName, tc := range testCases {
+		t.Run(testName, func(t *testing.T) {
+			fallback := calculateFallbackFromPolicy(tc.policy, tc.strategy)
+			if tc.expFallback == nil && fallback != nil {
+				t.Fatalf("Fallback is expected to be nil, got %v", fallback)
+			}
+			if tc.expFallback != nil {
+				for _, containerName := range tc.containerName {
+					if helpers.CmpResourceList(fallback[containerName].Requests, tc.expFallback[containerName].Requests) != 0 {
+						t.Errorf("Fallback results mismatch. want %v, got %v", tc.expFallback[containerName].Requests, fallback[containerName].Requests)
+					}
+					if helpers.CmpResourceList(fallback[containerName].Limits, tc.expFallback[containerName].Limits) != 0 {
+						t.Errorf("Fallback results mismatch. want %v, got %v", tc.expFallback[containerName].Requests, fallback[containerName].Requests)
+					}
+				}
+			}
+		})
+	}
+}
+
+func TestCalculateFallbackFromRunningInstances(t *testing.T) {
+	instances := []appsv1.Deployment{
+		{
+			Spec: appsv1.DeploymentSpec{
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{Name: "busybox", Resources: *tests.NewResourceRequirements("100m", "100M", "100m", "100M")},
+							{Name: "test", Resources: *tests.NewResourceRequirements("600m", "600M", "600m", "600M")},
+						}},
+				},
+			},
+		},
+		{
+			Spec: appsv1.DeploymentSpec{
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{Name: "busybox", Resources: *tests.NewResourceRequirements("150m", "150M", "1", "100G")},
+							{Name: "test", Resources: *tests.NewResourceRequirements("50m", "50M", "2", "200G")},
+						}},
+				},
+			},
+		},
+		{
+			Spec: appsv1.DeploymentSpec{
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{Name: "busybox", Resources: *tests.NewResourceRequirements("200m", "200M", "300m", "300M")},
+							{Name: "test", Resources: *tests.NewResourceRequirements("100m", "100M", "300m", "300M")},
+						}},
+				},
+			},
+		},
+	}
+
+	testCases := map[string]struct {
+		strategy      konsumeratorv1.FallbackStrategy
+		containerName []string
+		instances     []appsv1.Deployment
+		expFallback   map[string]*corev1.ResourceRequirements
+	}{
+		"min": {
+			strategy:      konsumeratorv1.FallbackStrategyMin,
+			containerName: []string{"busybox", "test"},
+			instances:     instances,
+			expFallback: map[string]*corev1.ResourceRequirements{
+				"busybox": tests.NewResourceRequirements("100m", "100M", "100m", "100M"),
+				"test":    tests.NewResourceRequirements("50m", "50M", "2", "200G"),
+			},
+		},
+		"max": {
+			strategy:      konsumeratorv1.FallbackStrategyMax,
+			containerName: []string{"busybox", "test"},
+			instances:     instances,
+			expFallback: map[string]*corev1.ResourceRequirements{
+				"busybox": tests.NewResourceRequirements("200m", "200M", "300m", "300M"),
+				"test":    tests.NewResourceRequirements("600m", "600M", "600m", "600M"),
+			},
+		},
+	}
+	for testName, tc := range testCases {
+		t.Run(testName, func(t *testing.T) {
+			fallback := calculateFallbackFromRunningInstances(tc.instances, tc.strategy)
+			if tc.expFallback == nil && fallback != nil {
+				t.Fatalf("Fallback is expected to be nil, got %v", fallback)
+			}
+			if tc.expFallback != nil {
+				for _, containerName := range tc.containerName {
+					if helpers.CmpResourceList(fallback[containerName].Requests, tc.expFallback[containerName].Requests) != 0 {
+						t.Errorf("Fallback results mismatch. want %v, got %v", tc.expFallback[containerName].Requests, fallback[containerName].Requests)
+					}
+					if helpers.CmpResourceList(fallback[containerName].Limits, tc.expFallback[containerName].Limits) != 0 {
+						t.Errorf("Fallback results mismatch. want %v, got %v", tc.expFallback[containerName].Requests, fallback[containerName].Requests)
+					}
+				}
+			}
+
+		})
+	}
+}

controllers/utils_test.go

@@ -34,6 +34,7 @@ data:
         # available, consumer will be scaled up to recover during
         # during this time.
         recoveryTime: "30m"
+        fallbackStrategy: "max"
         # prometheus addresses to query
         address:
           - "http://prometheus-server.kube-system:9090"