feat(operator): added adapter for custom metrics (#682)

Signed-off-by: Florian Bacher <florian.bacher@dynatrace.com>

operator/Makefile

@@ -83,15 +83,17 @@ vet: ## Run go vet against code.
 
 .PHONY: test
 test: manifests fmt vet generate envtest ## Run tests.
+	go test ./cmd/... -v -coverprofile cover-cmd.out
 	go test ./apis/... -v -coverprofile cover-apis.out
 	go test ./controllers/... -v -coverprofile cover-pkg.out
 	go test ./webhooks/... -v -coverprofile cover-main.out
+	sed -i '/mode: set/d' "cover-cmd.out"
 	sed -i '/mode: set/d' "cover-apis.out"
 	sed -i '/mode: set/d' "cover-pkg.out"
 	sed -i '/mode: set/d' "cover-main.out"
 	echo "mode: set" > cover.out
-	cat cover-main.out cover-pkg.out cover-apis.out >> cover.out
-	rm cover-pkg.out cover-main.out cover-apis.out
+	cat cover-cmd.out cover-main.out cover-pkg.out cover-apis.out >> cover.out
+	rm cover-cmd.out cover-pkg.out cover-main.out cover-apis.out
 
 .PHONY: component-test
 component-test: manifests generate envtest ## Run tests.

operator/cmd/metrics/adapter/adapter.go

@@ -0,0 +1,79 @@
+package adapter
+
+import (
+	"context"
+	"flag"
+	"fmt"
+
+	kmprovider "github.com/keptn/lifecycle-toolkit/operator/cmd/metrics/adapter/provider"
+	"k8s.io/apiserver/pkg/server/options"
+	"k8s.io/component-base/logs"
+	"k8s.io/klog/v2"
+	basecmd "sigs.k8s.io/custom-metrics-apiserver/pkg/cmd"
+	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
+)
+
+const (
+	flagPort                   = "adapter-port"
+	flagCertificateDirectory   = "adapter-certs-dir"
+	defaultCertificatePairName = "apiserver"
+)
+
+var (
+	port    int
+	certDir string
+)
+
+type MetricsAdapter struct {
+	basecmd.AdapterBase
+}
+
+// RunAdapter starts the Keptn Metrics adapter to provide KeptnMetrics via the Kubernetes Custom Metrics API.
+// Runs until the given context is done.
+func (a *MetricsAdapter) RunAdapter(ctx context.Context) {
+
+	logs.InitLogs()
+	defer logs.FlushLogs()
+
+	addFlags()
+
+	fmt.Println("Starting Keptn Metrics Adapter")
+	// initialize the flags, with one custom flag for the message
+	cmd := &MetricsAdapter{}
+	// make sure you get the klog flags
+	logs.AddGoFlags(flag.CommandLine)
+	cmd.Flags().AddGoFlagSet(flag.CommandLine)
+	if err := cmd.Flags().Parse([]string{}); err != nil {
+		klog.Fatalf("Could not parse flags: %v", err)
+	}
+
+	cmd.CustomMetricsAdapterServerOptions.SecureServing.BindPort = port
+	cmd.CustomMetricsAdapterServerOptions.SecureServing.ServerCert = options.GeneratableKeyCert{
+		PairName:      defaultCertificatePairName,
+		CertDirectory: certDir,
+	}
+
+	prov := cmd.makeProviderOrDie(ctx)
+
+	cmd.WithCustomMetrics(prov)
+
+	if err := cmd.Run(ctx.Done()); err != nil {
+		klog.Fatalf("Could not run custom metrics adapter: %v", err)
+	}
+	klog.Info("Finishing Keptn Metrics Adapter")
+}
+
+func (a *MetricsAdapter) makeProviderOrDie(ctx context.Context) provider.CustomMetricsProvider {
+	client, err := a.DynamicClient()
+	if err != nil {
+		klog.Fatalf("unable to construct dynamic client: %v", err)
+	}
+
+	return kmprovider.NewProvider(ctx, client)
+}
+
+func addFlags() {
+	flag.IntVar(&port, flagPort, 6443, "Port of the metrics adapter endpoint")
+	flag.StringVar(&certDir, flagCertificateDirectory, "/tmp/metrics-adapter/serving-certs", "Directory in which to look for certificates for the Metrics Adapter.")
+	flag.Parse()
+}

operator/cmd/metrics/adapter/provider/common.go

@@ -0,0 +1,11 @@
+package provider
+
+import "github.com/pkg/errors"
+
+const (
+	metricsGroup        = "metrics.keptn.sh"
+	metricsResource     = "keptnmetrics"
+	defaultMetricsValue = "0.0"
+)
+
+var ErrMetricNotFound = errors.New("no metric value found")

operator/cmd/metrics/adapter/provider/custom_metrics.go

@@ -0,0 +1,106 @@
+package provider
+
+import (
+	"sync"
+
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/metrics/pkg/apis/custom_metrics"
+	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
+)
+
+type CustomMetricValue struct {
+	Value  custom_metrics.MetricValue
+	Labels map[string]string
+}
+
+type CustomMetricsCache struct {
+	mtx     sync.RWMutex
+	metrics map[string]CustomMetricValue
+}
+
+// Update adds a new metricValue for the given metricName to the cache. If an item has already been present for the provided
+// metricName, the previous value will be replaced.
+func (cm *CustomMetricsCache) Update(metricName string, metricValue CustomMetricValue) {
+	cm.mtx.Lock()
+	defer cm.mtx.Unlock()
+	if cm.metrics == nil {
+		cm.metrics = map[string]CustomMetricValue{}
+	}
+
+	cm.metrics[metricName] = metricValue
+}
+
+// Delete will delete the value for the given metricName
+func (cm *CustomMetricsCache) Delete(metricName string) {
+	cm.mtx.Lock()
+	defer cm.mtx.Unlock()
+
+	delete(cm.metrics, metricName)
+}
+
+// List returns a slice of provider.CustomMetricInfo objects containing all the available metrics
+// that are currently present in the cache
+func (cm *CustomMetricsCache) List() []provider.CustomMetricInfo {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+	res := make([]provider.CustomMetricInfo, len(cm.metrics))
+
+	i := 0
+	for metricInfo := range cm.metrics {
+		res[i] = generateCustomMetricInfo(metricInfo)
+		i++
+	}
+	return res
+}
+
+// ListByLabelSelector returns a slice of provider.CustomMetricInfo objects containing all the available metrics
+// that are currently present in the cache and match with the provided labels
+func (cm *CustomMetricsCache) ListByLabelSelector(selector labels.Selector) []provider.CustomMetricInfo {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+	res := []provider.CustomMetricInfo{}
+	for metricInfo, metricValue := range cm.metrics {
+		if selector.Matches(labels.Set(metricValue.Labels)) {
+			res = append(res, generateCustomMetricInfo(metricInfo))
+		}
+	}
+	return res
+}
+
+// Get returns the metric value for the given metric name
+func (cm *CustomMetricsCache) Get(metricName string) (*CustomMetricValue, error) {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+	metric, ok := cm.metrics[metricName]
+	if !ok {
+		return nil, ErrMetricNotFound
+	}
+	return &metric, nil
+}
+
+// GetValuesByLabel returns a slice of CustomMetricValue objects containing the values of all
+// available metrics that match with the given label
+func (cm *CustomMetricsCache) GetValuesByLabel(selector labels.Selector) []CustomMetricValue {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	res := []CustomMetricValue{}
+	for _, value := range cm.metrics {
+		if selector.Matches(labels.Set(value.Labels)) {
+			res = append(res, value)
+		}
+	}
+	return res
+}
+
+func generateCustomMetricInfo(name string) provider.CustomMetricInfo {
+	return provider.CustomMetricInfo{
+		GroupResource: schema.GroupResource{
+			Group:    metricsGroup,
+			Resource: metricsResource,
+		},
+		Metric:     name,
+		Namespaced: true,
+	}
+}