From 0e9ade27f3179237374f90129ce5c653b3109fdd Mon Sep 17 00:00:00 2001
From: Lucas Thiesen <lucas.thiesen@zalando.de>
Date: Tue, 25 Apr 2023 20:05:47 +0200
Subject: [PATCH] [WIP] Add hostname RPS metric collector

Signed-off-by: Lucas Thiesen <lucas.thiesen@zalando.de>
---
 pkg/collector/hostname_collector.go      | 93 ++++++++++++++++++++++++
 pkg/collector/hostname_collector_test.go |  3 +
 pkg/server/start.go                      | 17 +++++
 3 files changed, 113 insertions(+)
 create mode 100644 pkg/collector/hostname_collector.go
 create mode 100644 pkg/collector/hostname_collector_test.go

diff --git a/pkg/collector/hostname_collector.go b/pkg/collector/hostname_collector.go
new file mode 100644
index 00000000..1948a5ee
--- /dev/null
+++ b/pkg/collector/hostname_collector.go
@@ -0,0 +1,93 @@
+package collector
+
+import (
+	"fmt"
+	"time"
+
+	autoscalingv2 "k8s.io/api/autoscaling/v2beta2"
+)
+
+const (
+	HostnameMetricType = "hostname"
+	HostnameRPSQuery   = `scalar(sum(rate(%s{host=~"%s"}[1m])))`
+)
+
+type HostnameCollectorPlugin struct {
+	metricName string
+	promPlugin CollectorPlugin
+}
+
+type HostnameCollector struct {
+	interval      time.Duration
+	promCollector Collector
+}
+
+func NewHostnameCollectorPlugin(
+	promPlugin CollectorPlugin,
+	metricName string,
+) (*HostnameCollectorPlugin, error) {
+	if metricName == "" {
+		return nil, fmt.Errorf("failed to initialize hostname collector plugin, metric name was not defined")
+	}
+
+	return &HostnameCollectorPlugin{
+		metricName: metricName,
+		promPlugin: promPlugin,
+	}, nil
+}
+
+func (p *HostnameCollectorPlugin) NewCollector(
+	hpa *autoscalingv2.HorizontalPodAutoscaler,
+	config *MetricConfig,
+	interval time.Duration,
+) (Collector, error) {
+	// Need to copy config and add a promQL query in order to get
+	// RPS data from a specific hostname from prometheus. The idea
+	// of the copy is to not modify the original config struct.
+	confCopy := *config
+	hostname := config.Config["hostname"]
+
+	if hostname == "" {
+		return nil, fmt.Errorf("hostname not specified, unable to create collector")
+	}
+
+	confCopy.Config = map[string]string{
+		"query": fmt.Sprintf(HostnameRPSQuery, p.metricName, hostname),
+	}
+
+	c, err := p.promPlugin.NewCollector(hpa, &confCopy, interval)
+	if err != nil {
+		return nil, err
+	}
+
+	return &HostnameCollector{
+		interval:      interval,
+		promCollector: c,
+	}, nil
+}
+
+// GetMetrics gets hostname metrics from Prometheus
+func (c *HostnameCollector) GetMetrics() ([]CollectedMetric, error) {
+	v, err := c.promCollector.GetMetrics()
+	if err != nil {
+		return nil, err
+	}
+
+	if len(v) != 1 {
+		return nil, fmt.Errorf("expected to only get one metric value, got %d", len(v))
+	}
+
+	// TBD(Lucas):
+	// The explanation bellow is only true if we want to implement object metrics.
+	// I believe external metrics would suffice.
+	// Apparently in case of k8s <v1.14 the value is not average for replica
+	// In this case we need to calculate RPS per replica manually. Check skipper
+	// collector. Anyway I need to check wether in the hostname metric I want
+	// average RPS or total. I probably want average but still need to check...
+	return v, nil
+}
+
+// Interval returns the interval at which the collector should run.
+func (c *HostnameCollector) Interval() time.Duration {
+	return c.interval
+}
diff --git a/pkg/collector/hostname_collector_test.go b/pkg/collector/hostname_collector_test.go
new file mode 100644
index 00000000..cc290c7b
--- /dev/null
+++ b/pkg/collector/hostname_collector_test.go
@@ -0,0 +1,3 @@
+package collector
+
+// TODO(Lucas)
diff --git a/pkg/server/start.go b/pkg/server/start.go
index 86388c51..2fbf4017 100644
--- a/pkg/server/start.go
+++ b/pkg/server/start.go
@@ -65,6 +65,7 @@ func NewCommandStartAdapterServer(stopCh <-chan struct{}) *cobra.Command {
 		MetricsAddress:                    ":7979",
 		ZMONTokenName:                     "zmon",
 		CredentialsDir:                    "/meta/credentials",
+		HostnameRPSMetricName:             "skipper_serve_host_duration_seconds_count",
 	}
 
 	cmd := &cobra.Command{
@@ -132,6 +133,8 @@ func NewCommandStartAdapterServer(stopCh <-chan struct{}) *cobra.Command {
 	flags.DurationVar(&o.DefaultScheduledScalingWindow, "scaling-schedule-default-scaling-window", 10*time.Minute, "Default rampup and rampdown window duration for ScalingSchedules")
 	flags.IntVar(&o.RampSteps, "scaling-schedule-ramp-steps", 10, "Number of steps used to rampup and rampdown ScalingSchedules. It's used to guarantee won't avoid reaching the max scaling due to the 10% minimum change rule.")
 	flags.StringVar(&o.DefaultTimeZone, "scaling-schedule-default-time-zone", "Europe/Berlin", "Default time zone to use for ScalingSchedules.")
+	flags.StringVar(&o.HostnameRPSMetricName, "hostname-rps-metric-name", o.HostnameRPSMetricName, ""+
+		"The name of the metric that should be used to query prometheus for RPS per hostname.")
 	return cmd
 }
 
@@ -218,6 +221,18 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
 				}
 			}
 		}
+
+		// Hostname collector, like skipper's, depends on prometheus being enabled.
+		// Also, to enable hostname metric its necessary to pass the metric name that
+		// will be used. This was built this way so we can support hostname metrics to
+		// any ingress provider, e.g. Skipper, Nginx, envoy etc, in a simple way.
+		if o.HostnameRPSMetricName != "" {
+			hostnamePlugin, err := collector.NewHostnameCollectorPlugin(promPlugin, o.HostnameRPSMetricName)
+			collectorFactory.RegisterExternalCollector([]string{collector.HostnameMetricType}, hostnamePlugin)
+			if err != nil {
+				return fmt.Errorf("failed to register hostname collector plugin: %v", err)
+			}
+		}
 	}
 
 	if o.InfluxDBAddress != "" {
@@ -445,4 +460,6 @@ type AdapterServerOptions struct {
 	RampSteps int
 	// Default time zone to use for ScalingSchedules.
 	DefaultTimeZone string
+	// Name of the Prometheus metric that stores RPS by hostname for Hostname RPS metrics.
+	HostnameRPSMetricName string
 }