Add support for k8s Resource metrics as source of scaling (#7)

azun · web-flow · commit edbc6c0bf411 · 2021-04-23T19:28:52.000+03:00
diff --git a/api/v1alpha1/kratos_types.go b/api/v1alpha1/kratos_types.go
@@ -134,6 +134,9 @@ type ResourceMetricSource struct {
 	Name v1.ResourceName `json:"name" protobuf:"bytes,1,name=name"`
 	// target specifies the target value for the given metric
 	Target MetricTarget `json:"target" protobuf:"bytes,2,name=target"`
+	// container is the name of the container in the pods of the scaling target.
+	// +optional
+	Container string `json:"container" protobuf:"bytes,3,opt,name=container"`
 }
 
 // PodsMetricSource indicates how to scale on a metric describing each pod in
diff --git a/config/crd/scaling.core.adobe.com_kratos.yaml b/config/crd/scaling.core.adobe.com_kratos.yaml
@@ -397,6 +397,9 @@ spec:
                     resource:
                       description: resource refers to a resource metric (such as those specified in requests and limits) known to Kubernetes describing each pod in the current scale target (e.g. CPU or memory). Such metrics are built in to Kubernetes, and have special scaling options on top of those available to normal per-pod metrics using the "pods" source.
                       properties:
+                        container:
+                          description: container is the name of the container in the pods of the scaling target.
+                          type: string
                         name:
                           description: name is the name of the resource in question.
                           type: string
diff --git a/examples/prometheus-scaler.yaml b/examples/prometheus-scaler.yaml
diff --git a/examples/resource-scaler.yaml b/examples/resource-scaler.yaml
@@ -0,0 +1,32 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: kratos-resource-example
+data:
+  kratosSpec: |-
+    algorithm:
+      type: hpa
+    minReplicas: 1
+    maxReplicas: 4
+    stabilizationWindowSeconds: 30
+    target:
+      apiVersion: apps/v1
+      kind: Deployment
+      name: nginx
+    metrics:
+      - type: Resource
+        resource:
+          name: cpu
+          target:
+            type: AverageValue # AverageValue/Utilization/Value
+            averageValue: 10
+      - type: Resource
+        resource:
+          name: memory
+          container: nginx
+          target:
+            type: AverageValue # AverageValue/Utilization/Value
+            averageValue: 100
+            # averageUtilization
+            # value
diff --git a/go.mod b/go.mod
@@ -17,6 +17,8 @@ require (
 	k8s.io/api v0.18.6
 	k8s.io/apimachinery v0.18.6
 	k8s.io/client-go v0.18.6
+	k8s.io/kubernetes v1.13.0
+	k8s.io/metrics v0.18.4
 	sigs.k8s.io/controller-runtime v0.6.2
 	sigs.k8s.io/controller-tools v0.4.0 // indirect
 	sigs.k8s.io/yaml v1.2.0
diff --git a/go.sum b/go.sum
@@ -929,6 +929,8 @@ k8s.io/client-go v0.18.2/go.mod h1:Xcm5wVGXX9HAA2JJ2sSBUn3tCJ+4SVlCbl2MNNv+CIU=
 k8s.io/client-go v0.18.4/go.mod h1:f5sXwL4yAZRkAtzOxRWUhA/N8XzGCb+nPZI8PfobZ9g=
 k8s.io/client-go v0.18.6 h1:I+oWqJbibLSGsZj8Xs8F0aWVXJVIoUHWaaJV3kUN/Zw=
 k8s.io/client-go v0.18.6/go.mod h1:/fwtGLjYMS1MaM5oi+eXhKwG+1UHidUEXRh6cNsdO0Q=
+k8s.io/client-go v1.5.1 h1:XaX/lo2/u3/pmFau8HN+sB5C/b4dc4Dmm2eXjBH4p1E=
+k8s.io/client-go v11.0.0+incompatible h1:LBbX2+lOwY9flffWlJM7f1Ct8V2SRNiMRDFeiwnJo9o=
 k8s.io/code-generator v0.18.2/go.mod h1:+UHX5rSbxmR8kzS+FAv7um6dtYrZokQvjHpDSYRVkTc=
 k8s.io/code-generator v0.18.4/go.mod h1:TgNEVx9hCyPGpdtCWA34olQYLkh3ok9ar7XfSsr8b6c=
 k8s.io/code-generator v0.18.6/go.mod h1:TgNEVx9hCyPGpdtCWA34olQYLkh3ok9ar7XfSsr8b6c=
@@ -949,7 +951,9 @@ k8s.io/kube-openapi v0.0.0-20200410145947-61e04a5be9a6 h1:Oh3Mzx5pJ+yIumsAD0MOEC
 k8s.io/kube-openapi v0.0.0-20200410145947-61e04a5be9a6/go.mod h1:GRQhZsXIAJ1xR0C9bd8UpWHZ5plfAS9fzPjJuQ6JL3E=
 k8s.io/kubectl v0.18.4 h1:l9DUYPTEMs1+qNtoqPpTyaJOosvj7l7tQqphCO1K52s=
 k8s.io/kubectl v0.18.4/go.mod h1:EzB+nfeUWk6fm6giXQ8P4Fayw3dsN+M7Wjy23mTRtB0=
+k8s.io/kubernetes v1.13.0 h1:qTfB+u5M92k2fCCCVP2iuhgwwSOv1EkAkvQY1tQODD8=
 k8s.io/kubernetes v1.13.0/go.mod h1:ocZa8+6APFNC2tX1DZASIbocyYT5jHzqFVsY5aoB7Jk=
+k8s.io/metrics v0.18.4 h1:iP0U2VhD1BHXIv98OrkFKb19ItRQZLhy834jySbltzI=
 k8s.io/metrics v0.18.4/go.mod h1:luze4fyI9JG4eLDZy0kFdYEebqNfi0QrG4xNEbPkHOs=
 k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7MpmSnvtrOieolJKoE=
diff --git a/metrics/metrics_fetcher.go b/metrics/metrics_fetcher.go
@@ -16,32 +16,50 @@ package metrics
 import (
 	"errors"
 	"fmt"
+	"time"
 
+	"k8s.io/apimachinery/pkg/labels"
+
+	"github.com/adobe/kratos/api/common"
 	"github.com/adobe/kratos/api/v1alpha1"
+	metrics "k8s.io/metrics/pkg/client/clientset/versioned"
+)
+
+const (
+	defaultCacheTtl    = time.Minute * 5
+	defaultCallTimeout = time.Second * 10
 )
 
 type MetricValue struct {
 	Value int64
 }
 
 type MetricsFetcher interface {
-	Fetch(scaleMetric *v1alpha1.ScaleMetric) ([]MetricValue, error)
+	Fetch(scaleMetric *v1alpha1.ScaleMetric, namespace string, selector labels.Selector) ([]MetricValue, error)
 }
 
 type MetricsFactory struct {
 	prometheusFetcher MetricsFetcher
+	resourceFetcher   MetricsFetcher
 }
 
-func NewMetricsFactory(defaultPrometheusUrl string) *MetricsFactory {
+func NewMetricsFactory(params *common.KratosParameters) *MetricsFactory {
+	mc, err := metrics.NewForConfig(params.ClientConfig)
+	if err != nil {
+		panic(err.Error())
+	}
 	return &MetricsFactory{
-		prometheusFetcher: newPrometheusMetricsFetcher(defaultPrometheusUrl),
+		prometheusFetcher: newPrometheusMetricsFetcher(params.DefaultPrometheusUrl),
+		resourceFetcher:   newResourceMetricsFetcher(mc),
 	}
 }
 
 func (facade *MetricsFactory) GetMetricsFetcher(scaleMetric *v1alpha1.ScaleMetric) (MetricsFetcher, error) {
 	switch scaleMetric.Type {
 	case v1alpha1.PrometheusScaleMetricType:
 		return facade.prometheusFetcher, nil
+	case v1alpha1.ResourceScaleMetricType:
+		return facade.resourceFetcher, nil
 	default:
 		return nil, errors.New(fmt.Sprintf("Unknown metric type %s \n", scaleMetric.Type))
 	}
diff --git a/metrics/metrics_fetcher_test.go b/metrics/metrics_fetcher_test.go
@@ -20,8 +20,10 @@ import (
 )
 
 var _ = Describe("MetricsFactory", func() {
+
 	It("Unknown fetcher type", func() {
-		metricsFactory := NewMetricsFactory("testUrl")
+
+		metricsFactory := NewMetricsFactory(fakeKratosSpec)
 		scaleMetric := &v1alpha1.ScaleMetric{
 			Type: "testType",
 		}
@@ -31,7 +33,7 @@ var _ = Describe("MetricsFactory", func() {
 	})
 
 	It("Prometheus fetcher type", func() {
-		metricsFactory := NewMetricsFactory("testUrl")
+		metricsFactory := NewMetricsFactory(fakeKratosSpec)
 		scaleMetric := &v1alpha1.ScaleMetric{
 			Type: v1alpha1.PrometheusScaleMetricType,
 		}
@@ -40,4 +42,15 @@ var _ = Describe("MetricsFactory", func() {
 		Expect(err).To(BeNil(), "no error for supported metrics fetcher type")
 		Expect(fetcher).NotTo(BeNil())
 	})
+
+	It("Resource fetcher type", func() {
+		metricsFactory := NewMetricsFactory(fakeKratosSpec)
+		scaleMetric := &v1alpha1.ScaleMetric{
+			Type: v1alpha1.ResourceScaleMetricType,
+		}
+		fetcher, err := metricsFactory.GetMetricsFetcher(scaleMetric)
+
+		Expect(err).To(BeNil(), "no error for supported metrics fetcher type")
+		Expect(fetcher).NotTo(BeNil())
+	})
 })
diff --git a/metrics/metrics_suite_test.go b/metrics/metrics_suite_test.go
@@ -14,12 +14,32 @@ written permission of Adobe.
 package metrics
 
 import (
+	"path/filepath"
+	"testing"
+
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
-	"testing"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+
+	"github.com/adobe/kratos/api/common"
+	scalingv1alpha1 "github.com/adobe/kratos/api/v1alpha1"
+	// +kubebuilder:scaffold:imports
 )
 
+// These tests use Ginkgo (BDD-style Go testing framework). Refer to
+// http://onsi.github.io/ginkgo/ to learn more about Ginkgo.
+
+var cfg *rest.Config
+var k8sClient client.Client
+var testEnv *envtest.Environment
+var fakeKratosSpec *common.KratosParameters
+
 func TestMetrics(t *testing.T) {
 	RegisterFailHandler(Fail)
 
@@ -29,9 +49,35 @@ func TestMetrics(t *testing.T) {
 }
 
 var _ = BeforeSuite(func(done Done) {
+	logf.SetLogger(zap.LoggerTo(GinkgoWriter, true))
+
+	By("bootstrapping test environment")
+	testEnv = &envtest.Environment{
+		CRDDirectoryPaths: []string{filepath.Join("..", "config", "crd", "bases")},
+	}
+
+	var err error
+	cfg, err = testEnv.Start()
+	Expect(err).ToNot(HaveOccurred())
+	Expect(cfg).ToNot(BeNil())
+
+	err = scalingv1alpha1.AddToScheme(scheme.Scheme)
+	Expect(err).NotTo(HaveOccurred())
+
+	// +kubebuilder:scaffold:scheme
+
+	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
+	Expect(err).ToNot(HaveOccurred())
+	Expect(k8sClient).ToNot(BeNil())
+	fakeKratosSpec = &common.KratosParameters{
+		ClientConfig: cfg,
+		Client:       k8sClient,
+	}
 	close(done)
 }, 60)
 
 var _ = AfterSuite(func() {
-
+	By("tearing down the test environment")
+	err := testEnv.Stop()
+	Expect(err).ToNot(HaveOccurred())
 })
diff --git a/metrics/prometheus_fetcher.go b/metrics/prometheus_fetcher.go
@@ -24,16 +24,12 @@ import (
 	"github.com/adobe/kratos/cache"
 	"github.com/go-logr/logr"
 	"github.com/prometheus/client_golang/api"
-	"github.com/prometheus/client_golang/api/prometheus/v1"
+	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
 	"github.com/prometheus/common/model"
+	"k8s.io/apimachinery/pkg/labels"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
-const (
-	defaultCacheTtl    = time.Minute * 5
-	defaultCallTimeout = time.Second * 10
-)
-
 type prometheusMetricsFetcher struct {
 	defaultUrl             string
 	prometheusClientsCache *cache.TTLCache
@@ -58,7 +54,7 @@ func newPrometheusMetricsFetcher(defaultUrl string) *prometheusMetricsFetcher {
 	return fetcher
 }
 
-func (p *prometheusMetricsFetcher) Fetch(scaleMetric *v1alpha1.ScaleMetric) ([]MetricValue, error) {
+func (p *prometheusMetricsFetcher) Fetch(scaleMetric *v1alpha1.ScaleMetric, namespace string, selector labels.Selector) ([]MetricValue, error) {
 	client, err := p.getOrCreateClient(scaleMetric.Prometheus.PrometheusEndpoint)
 
 	if err != nil {
diff --git a/metrics/prometheus_fetcher_test.go b/metrics/prometheus_fetcher_test.go
@@ -72,7 +72,7 @@ var _ = Describe("PrometheusFetcher", func() {
 			Prometheus: &v1alpha1.PrometheusMetricSource{
 				MetricQuery: "count(up)",
 			}}
-		_, err := fetcher.Fetch(scaleMetric)
+		_, err := fetcher.Fetch(scaleMetric, "", nil)
 
 		Expect(err).ToNot(BeNil(), "non prometheus response should result in error")
 	})
@@ -89,7 +89,7 @@ var _ = Describe("PrometheusFetcher", func() {
 			Prometheus: &v1alpha1.PrometheusMetricSource{
 				MetricQuery: query,
 			}}
-		fetchResults, err := fetcher.Fetch(scaleMetric)
+		fetchResults, err := fetcher.Fetch(scaleMetric, "", nil)
 
 		Expect(err).To(BeNil(), "no errors on scalar value")
 		Expect(len(fetchResults)).To(Equal(1), "scalar value should result in single item")
@@ -106,7 +106,7 @@ var _ = Describe("PrometheusFetcher", func() {
 			Prometheus: &v1alpha1.PrometheusMetricSource{
 				MetricQuery: query,
 			}}
-		fetchResults, err := fetcher.Fetch(scaleMetric)
+		fetchResults, err := fetcher.Fetch(scaleMetric, "", nil)
 
 		Expect(err).To(BeNil(), "no errors on vector value")
 		Expect(len(fetchResults)).To(Equal(0), "empty vector value should result in empty metrics")
@@ -135,7 +135,7 @@ var _ = Describe("PrometheusFetcher", func() {
 			Prometheus: &v1alpha1.PrometheusMetricSource{
 				MetricQuery: query,
 			}}
-		fetchResults, err := fetcher.Fetch(scaleMetric)
+		fetchResults, err := fetcher.Fetch(scaleMetric, "", nil)
 
 		Expect(err).To(BeNil(), "no errors on vector value")
 		Expect(len(fetchResults)).To(Equal(len(samples)), "vector size should be equal to returned metrics size")
diff --git a/metrics/resource_fetcher.go b/metrics/resource_fetcher.go
@@ -0,0 +1,86 @@
+/*
+
+Copyright 2020 Adobe
+All Rights Reserved.
+
+NOTICE: Adobe permits you to use, modify, and distribute this file in
+accordance with the terms of the Adobe license agreement accompanying
+it. If you have received this file from a source other than Adobe,
+then your use, modification, or distribution of it requires the prior
+written permission of Adobe.
+
+*/
+
+package metrics
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/adobe/kratos/api/v1alpha1"
+	"github.com/go-logr/logr"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	// "k8s.io/client-go/util/retry"
+	"k8s.io/apimachinery/pkg/labels"
+	clientset "k8s.io/metrics/pkg/client/clientset/versioned"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+type resourceMetricsFetcher struct {
+	metricsClient clientset.Interface
+	log           logr.Logger
+}
+
+func newResourceMetricsFetcher(metricsClient clientset.Interface) *resourceMetricsFetcher {
+	fetcher := &resourceMetricsFetcher{
+		metricsClient: metricsClient,
+		log:           log.Log.WithName("resourceMetrics-fetcher"),
+	}
+
+	return fetcher
+}
+
+func (r *resourceMetricsFetcher) Fetch(scaleMetric *v1alpha1.ScaleMetric, namespace string, selector labels.Selector) ([]MetricValue, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultCallTimeout)
+	defer cancel()
+
+	opts := metav1.ListOptions{
+		LabelSelector: selector.String(),
+	}
+
+	podMetricsList, err := r.metricsClient.MetricsV1beta1().PodMetricses(namespace).List(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+
+	r.log.Info("Fetched resource metrics", "selector", selector.String(), "container", scaleMetric.Resource.Container, "metrics", podMetricsList.Items)
+
+	ret := []MetricValue{}
+
+	for _, podMetrics := range podMetricsList.Items {
+		var accum int64 = 0
+		found := false
+		for _, container := range podMetrics.Containers {
+			if scaleMetric.Resource.Container == "" || scaleMetric.Resource.Container == container.Name {
+				found = true
+				switch scaleMetric.Resource.Name {
+				case corev1.ResourceCPU:
+					accum += container.Usage.Cpu().Value()
+				case corev1.ResourceMemory:
+					accum += container.Usage.Memory().Value()
+				default:
+					return nil, fmt.Errorf("Unsuported resource type %s", scaleMetric.Resource.Name)
+				}
+			}
+		}
+		if !found {
+			return nil, fmt.Errorf("container %s not present in metrics for pod %s/%s", scaleMetric.Resource.Container, namespace, podMetrics.Name)
+		}
+		ret = append(ret, MetricValue{accum})
+	}
+
+	return ret, nil
+}
diff --git a/metrics/resource_fetcher_test.go b/metrics/resource_fetcher_test.go
diff --git a/scale/scale_facade.go b/scale/scale_facade.go
diff --git a/scale/scale_target.go b/scale/scale_target.go

Original file line number	Diff line number	Diff line change
`@@ -134,6 +134,9 @@ type ResourceMetricSource struct {`
`134`	`134`	Name v1.ResourceName `json:"name" protobuf:"bytes,1,name=name"`
`135`	`135`	`// target specifies the target value for the given metric`
`136`	`136`	Target MetricTarget `json:"target" protobuf:"bytes,2,name=target"`
	`137`	`+ // container is the name of the container in the pods of the scaling target.`
	`138`	`+ // +optional`
	`139`	+ Container string `json:"container" protobuf:"bytes,3,opt,name=container"`
`137`	`140`	`}`
`138`	`141`
`139`	`142`	`// PodsMetricSource indicates how to scale on a metric describing each pod in`