Prometheus metrics scraper

apply deislabs#54

Author: vbehar

README.md

@@ -184,6 +184,7 @@ The following table lists the supported annotations for Kubernetes `Deployments`
 | `osiris.dm.gg/enabled` | Enable the zeroscaler component to scrape and analyze metrics from the deployment's or statefulSet's pods and scale the deployment/statefulSet to zero when idle. Allowed values: `y`, `yes`, `true`, `on`, `1`. | _no value_ (= disabled) |
 | `osiris.dm.gg/minReplicas` | The minimum number of replicas to set on the deployment/statefulSet when Osiris will scale up. If you set `2`, Osiris will scale the deployment/statefulSet from `0` to `2` replicas directly. Osiris won't collect metrics from deployments/statefulSets which have more than `minReplicas` replicas - to avoid useless collections of metrics. | `1` |
 | `osiris.dm.gg/metricsCheckInterval` | The interval in which Osiris would repeatedly track the pod http request metrics. The value is the number of seconds of the interval. Note that this value override the global value defined by the `zeroscaler.metricsCheckInterval` Helm value. | _value of the `zeroscaler.metricsCheckInterval` Helm value_ |
+| `osiris.dm.gg/metricsCollector` | Configure the collection of metrics for a pod. The value is a JSON object with at least a `type` string, and an optional `implementation` object. See the *Metrics Scraping* section for more. | `{ "type": "osiris" }` |
 
 #### Pod Annotations
 
@@ -210,6 +211,40 @@ The following table lists the supported annotations for Kubernetes `Services` an
 
 Note that you might see an `osiris.dm.gg/selector` annotation - this is for internal use only, and you shouldn't try to set/update or delete it.
 
+#### Metrics Scraping Configuration
+
+Scraping the metrics from the pods is done automatically using Osiris provided sidecar container by default. But if you don't want to use the auto-injected sidecar container, you can also configure a custom metrics scraper, using the `osiris.dm.gg/metricsCollector` annotation on your deployment/statefulset.
+
+The following scrapers are supported:
+
+**osiris**
+
+This is the default scraper, which doesn't need any configuration.
+
+**prometheus**
+
+The prometheus scraper retrieves metrics about the request count from your own prometheus endpoint. To use it, your application need to expose an endpoint with metrics in the prometheus format.
+You can then set the following annotation:
+
+```
+annotations:
+  osiris.dm.gg/metricsCollector: |
+    {
+      "type": "prometheus",
+      "implementation": {
+        "port": 8080,
+        "path": "/metrics",
+        "requestCountMetricName": "requests"
+      }
+    }
+```
+
+The schema of the prometheus implementation configuration is:
+- a mandatory `port` integer
+- an optional `path` string - default to `/metrics` if not set
+- a mandatory `requestCountMetricName` string, for the name of the metric that expose the number of requests
+- an optional `requestCountMetricLabels` object, for all labels that should match the metric for request count
+
 ### Demo
 
 Deploy the example application `hello-osiris` :

go.mod

@@ -11,6 +11,8 @@ require (
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/imdario/mergo v0.3.11 // indirect
 	github.com/kelseyhightower/envconfig v1.4.0
+	github.com/prometheus/client_model v0.2.0
+	github.com/prometheus/common v0.10.0
 	github.com/satori/go.uuid v1.2.0
 	github.com/stretchr/testify v1.5.1
 	golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a // indirect

go.sum

@@ -395,6 +395,7 @@ github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaO
 github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
 github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
+github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
 github.com/mholt/certmagic v0.6.2-0.20190624175158-6a42ef9fe8c2/go.mod h1:g4cOPxcjV0oFq3qwpjSA30LReKD8AoIfwAY9VvG35NY=
 github.com/miekg/dns v1.1.3/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
@@ -461,10 +462,12 @@ github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/prometheus/client_model v0.2.0 h1:uq5h0d+GuxiXLJLNABMgp2qUWDPiLvgCzz2dUR+/W/M=
 github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
 github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
+github.com/prometheus/common v0.10.0 h1:RyRA7RzGXQZiW+tGMr7sxa85G1z0yOpM1qq5c8lNawc=
 github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
 github.com/prometheus/procfs v0.0.0-20180125133057-cb4147076ac7/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=

pkg/deployments/zeroscaler/metrics_collector.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
 	"net/http"
 	"sync"
 	"time"
@@ -18,56 +17,47 @@ import (
 	"k8s.io/client-go/tools/cache"
 
 	k8s "github.com/dailymotion/osiris/pkg/kubernetes"
-	"github.com/dailymotion/osiris/pkg/metrics"
 )
 
-const (
-	proxyContainerName = "osiris-proxy"
-	proxyPortName      = "osiris-metrics"
-)
-
-type metricsCollector struct {
-	kubeClient           kubernetes.Interface
+type metricsCollectorConfig struct {
 	appKind              string
 	appName              string
 	appNamespace         string
 	selector             labels.Selector
 	metricsCheckInterval time.Duration
-	podsInformer         cache.SharedIndexInformer
-	appPods              map[string]*corev1.Pod
-	appPodsLock          sync.Mutex
-	httpClient           *http.Client
-	cancelFunc           func()
+	scraperConfig        metricsScraperConfig
+}
+
+type metricsCollector struct {
+	config       metricsCollectorConfig
+	scraper      metricsScraper
+	kubeClient   kubernetes.Interface
+	podsInformer cache.SharedIndexInformer
+	appPods      map[string]*corev1.Pod
+	appPodsLock  sync.Mutex
+	httpClient   *http.Client
+	cancelFunc   func()
 }
 
 func newMetricsCollector(
 	kubeClient kubernetes.Interface,
-	appKind string,
-	appName string,
-	appNamespace string,
-	selector labels.Selector,
-	metricsCheckInterval time.Duration,
-) *metricsCollector {
+	config metricsCollectorConfig,
+) (*metricsCollector, error) {
+	s, err := newMetricsScraper(config.scraperConfig)
+	if err != nil {
+		return nil, err
+	}
 	m := &metricsCollector{
-		kubeClient:           kubeClient,
-		appKind:              appKind,
-		appName:              appName,
-		appNamespace:         appNamespace,
-		selector:             selector,
-		metricsCheckInterval: metricsCheckInterval,
+		config:     config,
+		scraper:    s,
+		kubeClient: kubeClient,
 		podsInformer: k8s.PodsIndexInformer(
 			kubeClient,
-			appNamespace,
+			config.appNamespace,
 			nil,
-			selector,
+			config.selector,
 		),
 		appPods: map[string]*corev1.Pod{},
-		// A very aggressive timeout. When collecting metrics, we want to do it very
-		// quickly to minimize the possibility that some pods we've checked on have
-		// served requests while we've been checking on OTHER pods.
-		httpClient: &http.Client{
-			Timeout: 2 * time.Second,
-		},
 	}
 	m.podsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: m.syncAppPod,
@@ -76,7 +66,7 @@ func newMetricsCollector(
 		},
 		DeleteFunc: m.syncDeletedAppPod,
 	})
-	return m
+	return m, nil
 }
 
 func (m *metricsCollector) run(ctx context.Context) {
@@ -86,16 +76,16 @@ func (m *metricsCollector) run(ctx context.Context) {
 		<-ctx.Done()
 		glog.Infof(
 			"Stopping metrics collection for %s %s in namespace %s",
-			m.appKind,
-			m.appName,
-			m.appNamespace,
+			m.config.appKind,
+			m.config.appName,
+			m.config.appNamespace,
 		)
 	}()
 	glog.Infof(
 		"Starting metrics collection for %s %s in namespace %s",
-		m.appKind,
-		m.appName,
-		m.appNamespace,
+		m.config.appKind,
+		m.config.appName,
+		m.config.appNamespace,
 	)
 	go m.podsInformer.Run(ctx.Done())
 	// When this exits, the cancel func will stop the informer
@@ -123,7 +113,7 @@ func (m *metricsCollector) syncDeletedAppPod(obj interface{}) {
 func (m *metricsCollector) collectMetrics(ctx context.Context) {
 	requestCountsByProxy := map[string]uint64{}
 	var lastTotalRequestCount uint64
-	ticker := time.NewTicker(m.metricsCheckInterval)
+	ticker := time.NewTicker(m.config.metricsCheckInterval)
 	defer ticker.Stop()
 	for {
 		select {
@@ -135,26 +125,17 @@ func (m *metricsCollector) collectMetrics(ctx context.Context) {
 			// all.
 			timer := time.NewTimer(3 * time.Second)
 			for _, pod := range m.appPods {
-				podMetricsPort, ok := getMetricsPort(pod)
-				if !ok {
-					mustNotDecide = true
-					continue
-				}
-				url := fmt.Sprintf(
-					"http://%s:%d/metrics",
-					pod.Status.PodIP,
-					podMetricsPort,
-				)
 				scrapeWG.Add(1)
-				go func() {
+				go func(pod *corev1.Pod) {
 					defer scrapeWG.Done()
 					// Get the results
-					prc, ok := m.scrape(url)
-					if !ok {
+					prc := m.scraper.Scrap(pod)
+					if prc == nil {
 						mustNotDecide = true
+					} else {
+						requestCountsByProxy[prc.ProxyID] = prc.RequestCount
 					}
-					requestCountsByProxy[prc.ProxyID] = prc.RequestCount
-				}()
+				}(pod)
 			}
 			m.appPodsLock.Unlock()
 			scrapeWG.Wait()
@@ -180,65 +161,12 @@ func (m *metricsCollector) collectMetrics(ctx context.Context) {
 	}
 }
 
-func getMetricsPort(pod *corev1.Pod) (int32, bool) {
-	for _, c := range pod.Spec.Containers {
-		if c.Name == proxyContainerName && len(c.Ports) > 0 {
-			for _, port := range c.Ports {
-				if port.Name == proxyPortName {
-					return port.ContainerPort, true
-				}
-			}
-		}
-	}
-	return 0, false
-}
-
-func (m *metricsCollector) scrape(
-	target string,
-) (metrics.ProxyRequestCount, bool) {
-	prc := metrics.ProxyRequestCount{}
-	// Requests made with this client time out after 2 seconds
-	resp, err := m.httpClient.Get(target)
-	if err != nil {
-		glog.Errorf("Error requesting metrics from %s: %s", target, err)
-		return prc, false
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != 200 {
-		glog.Errorf(
-			"Received unexpected HTTP response code %d when requesting metrics "+
-				"from %s",
-			resp.StatusCode,
-			target,
-		)
-		return prc, false
-	}
-	bodyBytes, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		glog.Errorf(
-			"Error reading metrics request response from %s: %s",
-			target,
-			err,
-		)
-		return prc, false
-	}
-	if err := json.Unmarshal(bodyBytes, &prc); err != nil {
-		glog.Errorf(
-			"Error umarshaling metrics request response from %s: %s",
-			target,
-			err,
-		)
-		return prc, false
-	}
-	return prc, true
-}
-
 func (m *metricsCollector) scaleToZero(ctx context.Context) {
 	glog.Infof(
 		"Scale to zero starting for %s %s in namespace %s",
-		m.appKind,
-		m.appName,
-		m.appNamespace,
+		m.config.appKind,
+		m.config.appName,
+		m.config.appNamespace,
 	)
 
 	patches := []k8s.PatchOperation{{
@@ -248,41 +176,41 @@ func (m *metricsCollector) scaleToZero(ctx context.Context) {
 	}}
 	patchesBytes, _ := json.Marshal(patches)
 	var err error
-	switch m.appKind {
+	switch m.config.appKind {
 	case "Deployment":
-		_, err = m.kubeClient.AppsV1().Deployments(m.appNamespace).Patch(
+		_, err = m.kubeClient.AppsV1().Deployments(m.config.appNamespace).Patch(
 			ctx,
-			m.appName,
+			m.config.appName,
 			k8s_types.JSONPatchType,
 			patchesBytes,
 			metav1.PatchOptions{},
 		)
 	case "StatefulSet":
-		_, err = m.kubeClient.AppsV1().StatefulSets(m.appNamespace).Patch(
+		_, err = m.kubeClient.AppsV1().StatefulSets(m.config.appNamespace).Patch(
 			ctx,
-			m.appName,
+			m.config.appName,
 			k8s_types.JSONPatchType,
 			patchesBytes,
 			metav1.PatchOptions{},
 		)
 	default:
-		err = fmt.Errorf("unknown kind '%s'", m.appKind)
+		err = fmt.Errorf("unknown kind '%s'", m.config.appKind)
 	}
 	if err != nil {
 		glog.Errorf(
 			"Error scaling %s %s in namespace %s to zero: %s",
-			m.appKind,
-			m.appName,
-			m.appNamespace,
+			m.config.appKind,
+			m.config.appName,
+			m.config.appNamespace,
 			err,
 		)
 		return
 	}
 
 	glog.Infof(
 		"Scaled %s %s in namespace %s to zero",
-		m.appKind,
-		m.appName,
-		m.appNamespace,
+		m.config.appKind,
+		m.config.appName,
+		m.config.appNamespace,
 	)
 }