Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
Add a ConfigMap `config-kafka-autoscaler` to configure autoscaler default values
**Problem**

Autoscaler default values are in https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/control-plane/pkg/autoscaler/autoscaler_api.go, however, as administrator there is no way for me to override those defaults.

Proposed ConfigMap structure:

```
defaults.min-scale: <value>
defaults.max-scale: <value>
defaults.polling-interval: <value>
defaults.cooldown-period: <value>
defaults.lag-threshold: <value>
defaults.activation-lag-threshold: <value>
```

We should also add an example entry as usual like: https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml#L24 to explain what each config does

**[Persona:](https://github.com/knative/eventing/blob/master/docs/personas.md)** 
Which persona is this feature for?
Operator

**Exit Criteria**
A measurable (binary) test that would indicate that the problem has been resolved.

**Time Estimate (optional):** 
How many developer-days do you think this may take to resolve?
1

**Additional context (optional)**

/area api

Pull request description:
Configurable default values for autoscaler
Fixes #2804

Signed-off-by: aavarghese <avarghese@us.ibm.com>

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- New configmap for specifying [default values](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/control-plane/pkg/autoscaler/autoscaler_api.go#L52-L63) for Autoscaling Kafka resources

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-autoscaler.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-autoscaler.yaml
new file mode 100644
index 000000000..d5aebc4a6
--- /dev/null
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-autoscaler.yaml
@@ -0,0 +1,29 @@
+# Copyright 2022 The Knative 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
+#
+#     https://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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-autoscaler
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+data:
+    class: "keda.autoscaling.knative.dev"
+    min-scale: "0"
+    max-scale: "50"
+    polling-interval: "10"
+    cooldown-period: "300"
+    lag-threshold: "100"
+    activation-lag-threshold: "1"
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
index 748e38fcf..368face6a 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
@@ -147,6 +147,9 @@ spec:
             - name: DESCHEDULER_CONFIG
               value: 'config-kafka-descheduler'
 
+            - name: AUTOSCALER_CONFIG
+              value: 'config-kafka-autoscaler'
+
             - name: CONFIG_LEADERELECTION_NAME
               value: config-kafka-leader-election
             - name: CONFIG_LOGGING_NAME
diff --git a/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-autoscaler.yaml b/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-autoscaler.yaml
new file mode 120000
index 000000000..b23ec2379
--- /dev/null
+++ b/control-plane/config/eventing-kafka-source/200-controller/100-config-kafka-autoscaler.yaml
@@ -0,0 +1,1 @@
+../../eventing-kafka-broker/200-controller/100-config-kafka-autoscaler.yaml
\ No newline at end of file
diff --git a/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
index 31c78f744..b343fbd65 100644
--- a/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
@@ -82,6 +82,9 @@ spec:
             - name: DESCHEDULER_CONFIG
               value: 'config-kafka-descheduler'
 
+            - name: AUTOSCALER_CONFIG
+              value: 'config-kafka-autoscaler'
+
             - name: CONFIG_LEADERELECTION_NAME
               value: config-kafka-leader-election
             - name: CONFIG_LOGGING_NAME
diff --git a/control-plane/pkg/autoscaler/config.go b/control-plane/pkg/autoscaler/config.go
new file mode 100644
index 000000000..9b84e0878
--- /dev/null
+++ b/control-plane/pkg/autoscaler/config.go
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2022 The Knative 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 autoscaler
+
+import (
+	"fmt"
+	"strconv"
+)
+
+// AutoscalerConfig contains the defaults defined in the autoscaler ConfigMap.
+type AutoscalerConfig struct {
+	AutoscalerClass    map[string]string
+	AutoscalerDefaults map[string]int32
+}
+
+func defaultConfig() *AutoscalerConfig {
+	return &AutoscalerConfig{
+		AutoscalerClass: map[string]string{
+			AutoscalingClassAnnotation: AutoscalingClassDisabledAnnotationValue,
+		},
+		AutoscalerDefaults: map[string]int32{
+			AutoscalingMinScaleAnnotation:        DefaultMinReplicaCount,
+			AutoscalingMaxScaleAnnotation:        DefaultMaxReplicaCount,
+			AutoscalingPollingIntervalAnnotation: DefaultPollingInterval,
+			AutoscalingCooldownPeriodAnnotation:  DefaultCooldownPeriod,
+			AutoscalingLagThreshold:              DefaultLagThreshold,
+			AutoscalingActivationLagThreshold:    DefaultActivationLagThreshold,
+		},
+	}
+}
+
+// NewConfigFromMap creates a AutoscalerConfig from the supplied map
+func NewConfigFromMap(data map[string]string) (*AutoscalerConfig, error) {
+	ac := defaultConfig()
+
+	if class, ok := data["class"]; ok && class == AutoscalingClassDisabledAnnotationValue { //autoscaler disabled - ignore new defaults
+		return ac, nil
+	}
+
+	for key, val := range data {
+		key = convertConfigKeyToAutoscalingAnnotation(key)
+
+		if key == AutoscalingClassAnnotation {
+			ac.AutoscalerClass[AutoscalingClassAnnotation] = val
+		} else {
+			i, err := strconv.ParseInt(val, 10, 32)
+			if err != nil {
+				return nil, fmt.Errorf("Expected value for autoscaling annotation: "+key+" should be integer but got "+val, err)
+			}
+			ac.AutoscalerDefaults[key] = int32(i)
+		}
+	}
+
+	return ac, nil
+}
+
+func convertConfigKeyToAutoscalingAnnotation(configkey string) (autoscalingannotation string) {
+	return Autoscaling + "/" + configkey
+}
diff --git a/control-plane/pkg/autoscaler/keda/keda.go b/control-plane/pkg/autoscaler/keda/keda.go
index 379d79b3d..73813a7e1 100644
--- a/control-plane/pkg/autoscaler/keda/keda.go
+++ b/control-plane/pkg/autoscaler/keda/keda.go
@@ -47,17 +47,17 @@ func GenerateScaleTarget(cg *kafkainternals.ConsumerGroup) *kedav1alpha1.ScaleTa
 	}
 }
 
-func GenerateScaleTriggers(cg *kafkainternals.ConsumerGroup, triggerAuthentication *kedav1alpha1.TriggerAuthentication) ([]kedav1alpha1.ScaleTriggers, error) {
+func GenerateScaleTriggers(cg *kafkainternals.ConsumerGroup, triggerAuthentication *kedav1alpha1.TriggerAuthentication, aconfig autoscaler.AutoscalerConfig) ([]kedav1alpha1.ScaleTriggers, error) {
 	triggers := make([]kedav1alpha1.ScaleTriggers, 0, len(cg.Spec.Template.Spec.Topics))
 	bootstrapServers := cg.Spec.Template.Spec.Configs.Configs[kafka.BootstrapServersConfigMapKey]
 	consumerGroup := cg.Spec.Template.Spec.Configs.Configs[kafka.GroupIDConfigMapKey]
 
-	lagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingLagThreshold, autoscaler.DefaultLagThreshold)
+	lagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingLagThreshold, aconfig.AutoscalerDefaults[autoscaler.AutoscalingLagThreshold])
 	if err != nil {
 		return nil, err
 	}
 
-	activationLagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingActivationLagThreshold, autoscaler.DefaultActivationLagThreshold)
+	activationLagThreshold, err := GetInt32ValueFromMap(cg.Annotations, autoscaler.AutoscalingActivationLagThreshold, aconfig.AutoscalerDefaults[autoscaler.AutoscalingActivationLagThreshold])
 	if err != nil {
 		return nil, err
 	}
diff --git a/control-plane/pkg/autoscaler/keda/resources.go b/control-plane/pkg/autoscaler/keda/resources.go
index de2c24433..b7e9730d4 100644
--- a/control-plane/pkg/autoscaler/keda/resources.go
+++ b/control-plane/pkg/autoscaler/keda/resources.go
@@ -31,21 +31,21 @@ var (
 	KedaSchemeGroupVersion = schema.GroupVersion{Group: "keda.sh", Version: "v1alpha1"}
 )
 
-func GenerateScaledObject(obj metav1.Object, gvk schema.GroupVersionKind, scaleTarget *kedav1alpha1.ScaleTarget, triggers []kedav1alpha1.ScaleTriggers) (*kedav1alpha1.ScaledObject, error) {
+func GenerateScaledObject(obj metav1.Object, gvk schema.GroupVersionKind, scaleTarget *kedav1alpha1.ScaleTarget, triggers []kedav1alpha1.ScaleTriggers, aconfig autoscaler.AutoscalerConfig) (*kedav1alpha1.ScaledObject, error) {
 
-	cooldownPeriod, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingCooldownPeriodAnnotation, autoscaler.DefaultCooldownPeriod)
+	cooldownPeriod, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingCooldownPeriodAnnotation, aconfig.AutoscalerDefaults[autoscaler.AutoscalingCooldownPeriodAnnotation])
 	if err != nil {
 		return nil, err
 	}
-	pollingInterval, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingPollingIntervalAnnotation, autoscaler.DefaultPollingInterval)
+	pollingInterval, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingPollingIntervalAnnotation, aconfig.AutoscalerDefaults[autoscaler.AutoscalingPollingIntervalAnnotation])
 	if err != nil {
 		return nil, err
 	}
-	minReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMinScaleAnnotation, autoscaler.DefaultMinReplicaCount)
+	minReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMinScaleAnnotation, aconfig.AutoscalerDefaults[autoscaler.AutoscalingMinScaleAnnotation])
 	if err != nil {
 		return nil, err
 	}
-	maxReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMaxScaleAnnotation, autoscaler.DefaultMaxReplicaCount)
+	maxReplicaCount, err := GetInt32ValueFromMap(obj.GetAnnotations(), autoscaler.AutoscalingMaxScaleAnnotation, aconfig.AutoscalerDefaults[autoscaler.AutoscalingMaxScaleAnnotation])
 	if err != nil {
 		return nil, err
 	}
diff --git a/control-plane/pkg/reconciler/consumergroup/autoscaler_config.go b/control-plane/pkg/reconciler/consumergroup/autoscaler_config.go
new file mode 100644
index 000000000..1338ef4ab
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/autoscaler_config.go
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2021 The Knative 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 consumergroup
+
+import (
+	"context"
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/autoscaler"
+)
+
+func (r Reconciler) autoscalerDefaultsFromConfigMap(ctx context.Context, configMapName string) (*autoscaler.AutoscalerConfig, error) {
+	cm, err := r.KubeClient.CoreV1().ConfigMaps(r.SystemNamespace).Get(ctx, configMapName, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("error while retrieving the %s config map in namespace %s: %+v", configMapName, r.SystemNamespace, err)
+	}
+
+	config, err := autoscaler.NewConfigFromMap(cm.Data)
+	if err != nil {
+		return nil, fmt.Errorf("error while parsing the %s config map in namespace %s: %+v", configMapName, r.SystemNamespace, err)
+	}
+
+	return config, nil
+}
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup.go b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
index 54f83c8d9..7f90bce64 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup.go
@@ -94,6 +94,7 @@ type Reconciler struct {
 
 	KafkaFeatureFlags *config.KafkaFeatureFlags
 	KedaClient        kedaclientset.Interface
+	AutoscalerConfig  string
 }
 
 func (r Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.ConsumerGroup) reconciler.Event {
@@ -465,6 +466,11 @@ func (r Reconciler) reconcileKedaObjects(ctx context.Context, cg *kafkainternals
 	var triggerAuthentication *kedav1alpha1.TriggerAuthentication
 	var secret *corev1.Secret
 
+	autoscalerDefaults, err := r.autoscalerDefaultsFromConfigMap(ctx, r.AutoscalerConfig)
+	if err != nil {
+		return err
+	}
+
 	if hasSecretSpecConfig(cg.Spec.Template.Spec.Auth) || hasNetSpecAuthConfig(cg.Spec.Template.Spec.Auth) {
 		secretData, err := r.retrieveSecretData(ctx, cg)
 		if err != nil {
@@ -477,12 +483,12 @@ func (r Reconciler) reconcileKedaObjects(ctx context.Context, cg *kafkainternals
 		}
 	}
 
-	triggers, err := keda.GenerateScaleTriggers(cg, triggerAuthentication)
+	triggers, err := keda.GenerateScaleTriggers(cg, triggerAuthentication, *autoscalerDefaults)
 	if err != nil {
 		return err
 	}
 
-	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), keda.GenerateScaleTarget(cg), triggers)
+	scaledObject, err := keda.GenerateScaledObject(cg, cg.GetGroupVersionKind(), keda.GenerateScaleTarget(cg), triggers, *autoscalerDefaults)
 	if err != nil {
 		return err
 	}
@@ -541,7 +547,7 @@ func (r *Reconciler) reconcileScaledObject(ctx context.Context, expectedScaledOb
 	if !metav1.IsControlledBy(scaledObject, obj) {
 		return fmt.Errorf("scaledobject %s/%s is not owned by %s/%s", expectedScaledObject.Namespace, expectedScaledObject.Name, obj.GetNamespace(), obj.GetName())
 	}
-	if !equality.Semantic.DeepDerivative(expectedScaledObject.Spec, scaledObject.Spec) {
+	if !equality.Semantic.DeepDerivative(scaledObject.Spec, expectedScaledObject.Spec) {
 		scaledObject.Spec = expectedScaledObject.Spec
 		if _, err = r.KedaClient.KedaV1alpha1().ScaledObjects(expectedScaledObject.Namespace).Update(ctx, scaledObject, metav1.UpdateOptions{}); err != nil {
 			return fmt.Errorf("failed to update scaled object %s/%s: %w", expectedScaledObject.Namespace, expectedScaledObject.Name, err)
@@ -565,7 +571,7 @@ func (r *Reconciler) reconcileTriggerAuthentication(ctx context.Context, expecte
 	if !metav1.IsControlledBy(triggerAuth, obj) {
 		return fmt.Errorf("triggerauthentication object %s/%s is not owned by %s/%s", expectedTriggerAuth.Namespace, expectedTriggerAuth.Name, obj.GetNamespace(), obj.GetName())
 	}
-	if !equality.Semantic.DeepDerivative(expectedTriggerAuth.Spec, triggerAuth.Spec) {
+	if !equality.Semantic.DeepDerivative(triggerAuth.Spec, expectedTriggerAuth.Spec) {
 		triggerAuth.Spec = expectedTriggerAuth.Spec
 		if _, err = r.KedaClient.KedaV1alpha1().TriggerAuthentications(expectedTriggerAuth.Namespace).Update(ctx, triggerAuth, metav1.UpdateOptions{}); err != nil {
 			return fmt.Errorf("failed to update triggerauthentication object %s/%s: %w", expectedTriggerAuth.Namespace, expectedTriggerAuth.Name, err)
diff --git a/control-plane/pkg/reconciler/consumergroup/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
index 1c5ddd364..bf9f05f90 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -75,6 +75,7 @@ type envConfig struct {
 	PodCapacity                int32  `envconfig:"POD_CAPACITY" required:"true"`
 	SchedulerPolicyConfigMap   string `envconfig:"SCHEDULER_CONFIG" required:"true"`
 	DeSchedulerPolicyConfigMap string `envconfig:"DESCHEDULER_CONFIG" required:"true"`
+	AutoscalerConfigMap        string `envconfig:"AUTOSCALER_CONFIG" required:"true"`
 }
 
 type SchedulerConfig struct {
@@ -121,6 +122,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		KafkaFeatureFlags:          config.DefaultFeaturesConfig(),
 		KedaClient:                 kedaclient.Get(ctx),
+		AutoscalerConfig:           env.AutoscalerConfigMap,
 	}
 
 	consumerInformer := consumer.Get(ctx)


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
index d7f5c3901..3b0dbe541 100644
--- a/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/consumergroup_test.go
@@ -1671,7 +1671,8 @@ func TestReconcileKind(t *testing.T) {
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
-			SystemNamespace: systemNamespace,
+			SystemNamespace:  systemNamespace,
+			AutoscalerConfig: "",
 		}
 
 		r.KafkaFeatureFlags = configapis.FromContext(store.ToContext(ctx))
diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
index dc148abaf..33e1d51ac 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -48,6 +48,8 @@ const (
 	ConfigKafkaSchedulerName = "config-kafka-scheduler"
 	// ConfigKafkaDeSchedulerName is the name of the ConfigMap to configure the descheduler.
 	ConfigKafkaDeSchedulerName = "config-kafka-descheduler"
+	// ConfigKafkaAutoscalerName is the name of the ConfigMap to configure the autoscaler.
+	ConfigKafkaAutoscalerName = "config-kafka-autoscaler"
 )
 
 func TestNewController(t *testing.T) {
@@ -97,6 +99,7 @@ func TestNewController(t *testing.T) {
 	t.Setenv("POD_CAPACITY", PodCapacity)
 	t.Setenv("SCHEDULER_CONFIG", ConfigKafkaSchedulerName)
 	t.Setenv("DESCHEDULER_CONFIG", ConfigKafkaDeSchedulerName)
+	t.Setenv("AUTOSCALER_CONFIG", ConfigKafkaAutoscalerName)
 	controller := NewController(ctx, configmap.NewStaticWatcher(&corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "config-kafka-features",