KEDA Prometheus Autoscale (#2690)

* feat: keda prometheus autoscaler

* test: keda prometheus autoscaler

docs/reference/api.yaml

@@ -4720,12 +4720,19 @@ definitions:
         items:
           type: object
           $ref: "#/definitions/AutoScaleSchedule"
+      prometheus:
+        type: array
+        items:
+          type: object
+          $ref: "#/definitions/AutoScalePrometheus"
       version:
         type: integer
   AutoScaleSchedule:
     description: Auto Scale schedules struct
     type: object
     properties:
+      name:
+        type: string
       minReplicas:
         type: integer
       start:
@@ -4734,6 +4741,22 @@ definitions:
         type: string
       timezone:
         type: string
+  AutoScalePrometheus:
+    description: Auto Scale prometheus struct
+    type: object
+    properties:
+      name:
+        type: string
+      query:
+        type: string
+      threshold:
+        type: number
+        x-go-custom-type: "float64"
+      activationThreshold:
+        type: number
+        x-go-custom-type: "float64"
+      prometheusAddress:
+        type: string
   AppCName:
     description: Application CNames
     type: object

provision/kubernetes/autoscale.go

@@ -5,13 +5,16 @@
 package kubernetes
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"html/template"
 	"strconv"
 	"strings"
 
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
 	"github.com/pkg/errors"
+	"github.com/tsuru/config"
 	tsuruErrors "github.com/tsuru/tsuru/errors"
 	"github.com/tsuru/tsuru/provision"
 	appsv1 "k8s.io/api/apps/v1"
@@ -189,17 +192,31 @@ func scaledObjectToSpec(scaledObject kedav1alpha1.ScaledObject) provision.AutoSc
 	}
 
 	for _, metric := range scaledObject.Spec.Triggers {
-		if metric.Type == "cron" {
+		switch metric.Type {
+		case "cron":
 			minReplicas, _ := strconv.Atoi(metric.Metadata["desiredReplicas"])
 
 			spec.Schedules = append(spec.Schedules, provision.AutoScaleSchedule{
 				MinReplicas: minReplicas,
 				Start:       metric.Metadata["start"],
 				End:         metric.Metadata["end"],
 				Timezone:    metric.Metadata["timezone"],
+				Name:        metric.Metadata["scheduleName"],
 			})
-		}
-		if metric.Type == "cpu" {
+
+		case "prometheus":
+			thresholdValue, _ := strconv.ParseFloat(metric.Metadata["threshold"], 64)
+			activationThresholdValue, _ := strconv.ParseFloat(metric.Metadata["activationThreshold"], 64)
+
+			spec.Prometheus = append(spec.Prometheus, provision.AutoScalePrometheus{
+				Name:                metric.Metadata["prometheusMetricName"],
+				Query:               metric.Metadata["query"],
+				Threshold:           thresholdValue,
+				ActivationThreshold: activationThresholdValue,
+				PrometheusAddress:   metric.Metadata["serverAddress"],
+			})
+
+		case "cpu":
 			cpuValue := metric.Metadata["value"]
 			if metric.MetricType == autoscalingv2.UtilizationMetricType {
 				//percentage based, so multiple by 10
@@ -345,7 +362,7 @@ func setAutoScale(ctx context.Context, client *ClusterClient, a provision.App, s
 	labels = labels.WithoutIsolated().WithoutRoutable()
 	hpaName := hpaNameForApp(a, depInfo.process)
 
-	if len(spec.Schedules) > 0 {
+	if len(spec.Schedules) > 0 || len(spec.Prometheus) > 0 {
 		err = setKEDAAutoscale(ctx, client, spec, a, depInfo, hpaName, labels)
 		if err != nil {
 			return errors.WithStack(err)
@@ -502,6 +519,7 @@ func newKEDAScaledObject(ctx context.Context, spec provision.AutoScaleSpec, a pr
 		kedaTriggers = append(kedaTriggers, kedav1alpha1.ScaleTriggers{
 			Type: "cron",
 			Metadata: map[string]string{
+				"scheduleName":    schedule.Name,
 				"desiredReplicas": strconv.Itoa(schedule.MinReplicas),
 				"start":           schedule.Start,
 				"end":             schedule.End,
@@ -510,6 +528,15 @@ func newKEDAScaledObject(ctx context.Context, spec provision.AutoScaleSpec, a pr
 		})
 	}
 
+	for _, prometheus := range spec.Prometheus {
+		prometheusTrigger, err := buildPrometheusTrigger(ns, prometheus)
+		if err != nil {
+			return nil, err
+		}
+
+		kedaTriggers = append(kedaTriggers, *prometheusTrigger)
+	}
+
 	return &kedav1alpha1.ScaledObject{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      hpaName,
@@ -534,6 +561,51 @@ func newKEDAScaledObject(ctx context.Context, spec provision.AutoScaleSpec, a pr
 	}, nil
 }
 
+func buildPrometheusTrigger(ns string, prometheus provision.AutoScalePrometheus) (*kedav1alpha1.ScaleTriggers, error) {
+	if prometheus.PrometheusAddress == "" {
+		defaultPrometheusAddress, err := buildDefaultPrometheusAddress(ns)
+		if err != nil {
+			return nil, err
+		}
+
+		prometheus.PrometheusAddress = defaultPrometheusAddress
+	}
+
+	return &kedav1alpha1.ScaleTriggers{
+		Type: "prometheus",
+		Metadata: map[string]string{
+			"serverAddress":        prometheus.PrometheusAddress,
+			"query":                prometheus.Query,
+			"threshold":            strconv.FormatFloat(prometheus.Threshold, 'f', -1, 64),
+			"activationThreshold":  strconv.FormatFloat(prometheus.ActivationThreshold, 'f', -1, 64),
+			"prometheusMetricName": prometheus.Name,
+		},
+	}, nil
+}
+
+func buildDefaultPrometheusAddress(ns string) (string, error) {
+	prometheusAddressTemplate, err := config.GetString("kubernetes:keda:prometheus-address-template")
+	if err != nil {
+		return "", err
+	}
+
+	tmpl, err := template.New("prometheusAddress").Parse(prometheusAddressTemplate)
+	if err != nil {
+		return "", err
+	}
+
+	var buf bytes.Buffer
+
+	err = tmpl.Execute(&buf, map[string]string{
+		"namespace": ns,
+	})
+	if err != nil {
+		return "", err
+	}
+
+	return buf.String(), nil
+}
+
 func buildHPABehavior() *autoscalingv2.HorizontalPodAutoscalerBehavior {
 	// setting ground to allow user customization regarding Up/Down scale behavior
 	// should be a feature in the future