Merge pull request #460 from mathetake/datadog-metrics-provider

feature: add datadog metrics provider

artifacts/flagger/crd.yaml

@@ -738,6 +738,7 @@ spec:
                   enum:
                     - prometheus
                     - influxdb
+                    - datadog
                 address:
                   description: API address of this provider
                   type: string

charts/flagger/crds/crd.yaml

@@ -738,6 +738,7 @@ spec:
                   enum:
                     - prometheus
                     - influxdb
+                    - datadog
                 address:
                   description: API address of this provider
                   type: string

kustomize/base/flagger/crd.yaml

@@ -738,6 +738,7 @@ spec:
                   enum:
                     - prometheus
                     - influxdb
+                    - datadog
                 address:
                   description: API address of this provider
                   type: string

pkg/controller/scheduler.go

@@ -939,7 +939,7 @@ func (c *Controller) runMetricChecks(canary *flaggerv1.Canary) bool {
 			}
 
 			factory := providers.Factory{}
-			provider, err := factory.Provider(template.Spec.Provider, credentials)
+			provider, err := factory.Provider(metric.Interval, template.Spec.Provider, credentials)
 			if err != nil {
 				c.recordEventErrorf(canary, "Metric template %s.%s provider %s error: %v",
 					metric.TemplateRef.Name, namespace, template.Spec.Provider.Type, err)

pkg/metrics/providers/datadog.go

@@ -0,0 +1,168 @@
+package providers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strconv"
+	"time"
+
+	flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+)
+
+// https://docs.datadoghq.com/api/
+const (
+	datadogDefaultHost = "https://api.datadoghq.com"
+
+	datadogMetricsQueryPath     = "/api/v1/query"
+	datadogAPIKeyValidationPath = "/api/v1/validate"
+
+	datadogAPIKeySecretKey = "datadog_api_key"
+	datadogAPIKeyHeaderKey = "DD-API-KEY"
+
+	datadogApplicationKeySecretKey = "datadog_application_key"
+	datadogApplicationKeyHeaderKey = "DD-APPLICATION-KEY"
+
+	datadogFromDeltaMultiplierOnMetricInterval = 10
+)
+
+// DatadogProvider executes datadog queries
+type DatadogProvider struct {
+	metricsQueryEndpoint     string
+	apiKeyValidationEndpoint string
+
+	timeout        time.Duration
+	apiKey         string
+	applicationKey string
+	fromDelta      int64
+}
+
+type datadogResponse struct {
+	Series []struct {
+		Pointlist [][]float64 `json:"pointlist"`
+	}
+}
+
+// NewDatadogProvider takes a canary spec, a provider spec and the credentials map, and
+// returns a Datadog client ready to execute queries against the API
+func NewDatadogProvider(metricInterval string,
+	provider flaggerv1.MetricTemplateProvider,
+	credentials map[string][]byte) (*DatadogProvider, error) {
+
+	address := provider.Address
+	if address == "" {
+		address = datadogDefaultHost
+	}
+
+	dd := DatadogProvider{
+		timeout:                  5 * time.Second,
+		metricsQueryEndpoint:     address + datadogMetricsQueryPath,
+		apiKeyValidationEndpoint: address + datadogAPIKeyValidationPath,
+	}
+
+	if b, ok := credentials[datadogAPIKeySecretKey]; ok {
+		dd.apiKey = string(b)
+	} else {
+		return nil, fmt.Errorf("datadog credentials does not contain datadog_api_key")
+	}
+
+	if b, ok := credentials[datadogApplicationKeySecretKey]; ok {
+		dd.applicationKey = string(b)
+	} else {
+		return nil, fmt.Errorf("datadog credentials does not contain datadog_application_key")
+	}
+
+	md, err := time.ParseDuration(metricInterval)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing metric interval: %s", err.Error())
+	}
+
+	dd.fromDelta = int64(datadogFromDeltaMultiplierOnMetricInterval * md.Seconds())
+	return &dd, nil
+}
+
+// RunQuery executes the datadog query against DatadogProvider.metricsQueryEndpoint
+// and returns the the first result as float64
+func (p *DatadogProvider) RunQuery(query string) (float64, error) {
+
+	req, err := http.NewRequest("GET", p.metricsQueryEndpoint, nil)
+	if err != nil {
+		return 0, fmt.Errorf("error http.NewRequest: %s", err.Error())
+	}
+
+	req.Header.Set(datadogAPIKeyHeaderKey, p.apiKey)
+	req.Header.Set(datadogApplicationKeyHeaderKey, p.applicationKey)
+	now := time.Now().Unix()
+	q := req.URL.Query()
+	q.Add("query", query)
+	q.Add("from", strconv.FormatInt(now-p.fromDelta, 10))
+	q.Add("to", strconv.FormatInt(now, 10))
+	req.URL.RawQuery = q.Encode()
+
+	ctx, cancel := context.WithTimeout(req.Context(), p.timeout)
+	defer cancel()
+	r, err := http.DefaultClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return 0, err
+	}
+
+	defer r.Body.Close()
+	b, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return 0, fmt.Errorf("error reading body: %s", err.Error())
+	}
+
+	if r.StatusCode != http.StatusOK {
+		return 0, fmt.Errorf("error response: %s", string(b))
+	}
+
+	var res datadogResponse
+	if err := json.Unmarshal(b, &res); err != nil {
+		return 0, fmt.Errorf("error unmarshaling result: %s, '%s'", err.Error(), string(b))
+	}
+
+	if len(res.Series) < 1 {
+		return 0, fmt.Errorf("no values found in response: %s", string(b))
+	}
+
+	s := res.Series[0]
+	vs := s.Pointlist[len(s.Pointlist)-1]
+	if len(vs) < 1 {
+		return 0, fmt.Errorf("no values found in response: %s", string(b))
+	}
+
+	return vs[1], nil
+}
+
+// IsOnline calls the Datadog's validation endpoint with api keys
+// and returns an error if the validation fails
+func (p *DatadogProvider) IsOnline() (bool, error) {
+	req, err := http.NewRequest("GET", p.apiKeyValidationEndpoint, nil)
+	if err != nil {
+		return false, fmt.Errorf("error http.NewRequest: %s", err.Error())
+	}
+
+	req.Header.Add(datadogAPIKeyHeaderKey, p.apiKey)
+	req.Header.Add(datadogApplicationKeyHeaderKey, p.applicationKey)
+
+	ctx, cancel := context.WithTimeout(req.Context(), p.timeout)
+	defer cancel()
+	r, err := http.DefaultClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return false, err
+	}
+	defer r.Body.Close()
+
+	b, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return false, fmt.Errorf("error reading body: %s", err.Error())
+	}
+
+	if r.StatusCode != http.StatusOK {
+		return false, fmt.Errorf("error response: %s", string(b))
+	}
+
+	return true, nil
+}

pkg/metrics/providers/datadog_test.go

@@ -0,0 +1,156 @@
+package providers
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+	"time"
+
+	flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+)
+
+func TestNewDatadogProvider(t *testing.T) {
+	appKey := "app-key"
+	apiKey := "api-key"
+	cs := map[string][]byte{
+		datadogApplicationKeySecretKey: []byte(appKey),
+		datadogAPIKeySecretKey:         []byte(apiKey),
+	}
+
+	mi := "100s"
+	md, err := time.ParseDuration(mi)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	dp, err := NewDatadogProvider("100s", flaggerv1.MetricTemplateProvider{}, cs)
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if exp := "https://api.datadoghq.com/api/v1/validate"; dp.apiKeyValidationEndpoint != exp {
+		t.Fatalf("apiKeyValidationEndpoint expected %s but got %s", exp, dp.apiKeyValidationEndpoint)
+	}
+
+	if exp := "https://api.datadoghq.com/api/v1/query"; dp.metricsQueryEndpoint != exp {
+		t.Fatalf("metricsQueryEndpoint expected %s but got %s", exp, dp.metricsQueryEndpoint)
+	}
+
+	if exp := int64(md.Seconds() * datadogFromDeltaMultiplierOnMetricInterval); dp.fromDelta != exp {
+		t.Fatalf("fromDelta expected %d but got %d", exp, dp.fromDelta)
+	}
+
+	if dp.applicationKey != appKey {
+		t.Fatalf("application key expected %s but got %s", appKey, dp.applicationKey)
+	}
+
+	if dp.apiKey != apiKey {
+		t.Fatalf("api key expected %s but got %s", apiKey, dp.apiKey)
+	}
+}
+
+func TestDatadogProvider_RunQuery(t *testing.T) {
+	eq := `avg:system.cpu.user\{*}by{host}`
+	appKey := "app-key"
+	apiKey := "api-key"
+	expected := 1.11111
+
+	now := time.Now().Unix()
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		aq := r.URL.Query().Get("query")
+		if aq != eq {
+			t.Errorf("\nquery expected %s bug got %s", eq, aq)
+		}
+
+		if vs := r.Header.Get(datadogApplicationKeyHeaderKey); vs != appKey {
+			t.Errorf("\n%s header expected %s but got %s", datadogApplicationKeyHeaderKey, appKey, vs)
+		}
+		if vs := r.Header.Get(datadogAPIKeyHeaderKey); vs != apiKey {
+			t.Errorf("\n%s header expected %s but got %s", datadogAPIKeyHeaderKey, apiKey, vs)
+		}
+
+		rf := r.URL.Query().Get("from")
+		if from, err := strconv.ParseInt(rf, 10, 64); err == nil && from >= now {
+			t.Errorf("\nfrom %d should be less than %d", from, now)
+		} else if err != nil {
+			t.Errorf("\nfailed to parse from: %v", err)
+		}
+
+		rt := r.URL.Query().Get("to")
+		if to, err := strconv.ParseInt(rt, 10, 64); err == nil && to < now {
+			t.Errorf("\nto %d should be greater than or equals %d", to, now)
+		} else if err != nil {
+			t.Errorf("\nfailed to parse to: %v", err)
+		}
+
+		json := fmt.Sprintf(`{"series": [{"pointlist": [[1577232000000,29325.102158814265],[1577318400000,56294.46758591842],[1577404800000,%f]]}]}`, expected)
+		w.Write([]byte(json))
+	}))
+	defer ts.Close()
+
+	dp, err := NewDatadogProvider("1m",
+		flaggerv1.MetricTemplateProvider{Address: ts.URL},
+		map[string][]byte{
+			datadogApplicationKeySecretKey: []byte(appKey),
+			datadogAPIKeySecretKey:         []byte(apiKey),
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	f, err := dp.RunQuery(eq)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if f != expected {
+		t.Fatalf("metric value expected %f but got %f", expected, f)
+	}
+}
+
+func TestDatadogProvider_IsOnline(t *testing.T) {
+	for _, c := range []struct {
+		code        int
+		errExpected bool
+	}{
+		{code: http.StatusOK, errExpected: false},
+		{code: http.StatusUnauthorized, errExpected: true},
+	} {
+		t.Run(fmt.Sprintf("%d", c.code), func(t *testing.T) {
+			appKey := "app-key"
+			apiKey := "api-key"
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if vs := r.Header.Get(datadogApplicationKeyHeaderKey); vs != appKey {
+					t.Errorf("\n%s header expected %s but got %s", datadogApplicationKeyHeaderKey, appKey, vs)
+				}
+				if vs := r.Header.Get(datadogAPIKeyHeaderKey); vs != apiKey {
+					t.Errorf("\n%s header expected %s but got %s", datadogAPIKeyHeaderKey, apiKey, vs)
+				}
+				w.WriteHeader(c.code)
+			}))
+			defer ts.Close()
+
+			dp, err := NewDatadogProvider("1m",
+				flaggerv1.MetricTemplateProvider{Address: ts.URL},
+				map[string][]byte{
+					datadogApplicationKeySecretKey: []byte(appKey),
+					datadogAPIKeySecretKey:         []byte(apiKey),
+				},
+			)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			_, err = dp.IsOnline()
+			if c.errExpected && err == nil {
+				t.Fatal("error expected but got no error")
+			} else if !c.errExpected && err != nil {
+				t.Fatalf("no error expected but got %v", err)
+			}
+		})
+	}
+}

pkg/metrics/providers/factory.go

@@ -1,14 +1,22 @@
 package providers
 
-import flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+import (
+	flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+)
 
-type Factory struct {
-}
+type Factory struct{}
+
+func (factory Factory) Provider(
+	metricInterval string,
+	provider flaggerv1.MetricTemplateProvider,
+	credentials map[string][]byte,
+) (Interface, error) {
 
-func (factory Factory) Provider(provider flaggerv1.MetricTemplateProvider, credentials map[string][]byte) (Interface, error) {
 	switch {
 	case provider.Type == "prometheus":
 		return NewPrometheusProvider(provider, credentials)
+	case provider.Type == "datadog":
+		return NewDatadogProvider(metricInterval, provider, credentials)
 	default:
 		return NewPrometheusProvider(provider, credentials)
 	}