Merge pull request #915 from HBOCodeLabs/graphite-metrics-provider

Add Graphite metrics provider

artifacts/flagger/crd.yaml

@@ -1106,6 +1106,7 @@ spec:
                         - datadog
                         - cloudwatch
                         - newrelic
+                        - graphite
                     address:
                       description: API address of this provider
                       type: string

charts/flagger/crds/crd.yaml

@@ -1106,6 +1106,7 @@ spec:
                         - datadog
                         - cloudwatch
                         - newrelic
+                        - graphite
                     address:
                       description: API address of this provider
                       type: string

docs/gitbook/usage/metrics.md

@@ -401,3 +401,78 @@ Reference the template in the canary analysis:
           max: 5
         interval: 1m
 ```
+
+## Graphite
+
+You can create custom metric checks using the Graphite provider.
+
+Graphite template example:
+
+```yaml
+apiVersion: flagger.app/v1beta1
+kind: MetricTemplate
+metadata:
+  name: graphite-request-success-rate
+spec:
+  provider:
+    type: graphite
+    address: http://graphite.monitoring
+  query: |
+    target=summarize(
+      asPercent(
+        sumSeries(
+          stats.timers.httpServerRequests.app.{{target}}.exception.*.method.*.outcome.{CLIENT_ERROR,INFORMATIONAL,REDIRECTION,SUCCESS}.status.*.uri.*.count
+        ),
+        sumSeries(
+          stats.timers.httpServerRequests.app.{{target}}.exception.*.method.*.outcome.*.status.*.uri.*.count
+        )
+      ),
+      {{interval}},
+      'avg'
+    )
+```
+
+Reference the template in the canary analysis:
+
+```yaml
+  analysis:
+    metrics:
+      - name: "success rate"
+        templateRef:
+          name: graphite-request-success-rate
+        thresholdRange:
+          min: 90
+        interval: 1min
+```
+
+## Graphite authentication
+
+If your Graphite API requires basic authentication, you can create a secret in the same namespace
+as the `MetricTemplate` with the basic-auth credentials:
+
+```yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: graphite-basic-auth
+  namespace: flagger
+data:
+  username: your-user
+  password: your-password
+```
+
+Then, reference the secret in the `MetricTemplate`:
+
+```yaml
+apiVersion: flagger.app/v1beta1
+kind: MetricTemplate
+metadata:
+  name: my-metric
+  namespace: flagger
+spec:
+  provider:
+    type: graphite
+    address: http://graphite.monitoring
+    secretRef:
+      name: graphite-basic-auth
+```

kustomize/base/flagger/crd.yaml

@@ -1106,6 +1106,7 @@ spec:
                         - datadog
                         - cloudwatch
                         - newrelic
+                        - graphite
                     address:
                       description: API address of this provider
                       type: string

pkg/metrics/providers/factory.go

@@ -36,6 +36,8 @@ func (factory Factory) Provider(
 		return NewCloudWatchProvider(metricInterval, provider)
 	case "newrelic":
 		return NewNewRelicProvider(metricInterval, provider, credentials)
+	case "graphite":
+		return NewGraphiteProvider(provider, credentials)
 	default:
 		return NewPrometheusProvider(provider, credentials)
 	}

pkg/metrics/providers/graphite.go

@@ -0,0 +1,222 @@
+/*
+Copyright 2020 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package providers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"math"
+	"net/http"
+	"net/url"
+	"path"
+	"regexp"
+	"strconv"
+	"time"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+)
+
+type graphiteDataPoint struct {
+	Value     *float64
+	TimeStamp time.Time
+}
+
+func (gdp *graphiteDataPoint) UnmarshalJSON(data []byte) error {
+
+	var v []interface{}
+	if err := json.Unmarshal(data, &v); err != nil {
+		return err
+	}
+
+	if len(v) != 2 {
+		return fmt.Errorf("error unmarshaling data point: %v", v)
+	}
+
+	switch v[0].(type) {
+	case nil:
+		// no value
+	case float64:
+		f, _ := v[0].(float64)
+		gdp.Value = &f
+	case string:
+		f, err := strconv.ParseFloat(v[0].(string), 64)
+		if err != nil {
+			return err
+		}
+		gdp.Value = &f
+	default:
+		f, ok := v[0].(float64)
+		if !ok {
+			return fmt.Errorf("error unmarshaling value: %v", v[0])
+		}
+		gdp.Value = &f
+	}
+
+	switch v[1].(type) {
+	case nil:
+		// no value
+	case float64:
+		ts := int64(math.Round(v[1].(float64)))
+		gdp.TimeStamp = time.Unix(ts, 0)
+	case string:
+		ts, err := strconv.ParseInt(v[1].(string), 10, 64)
+		if err != nil {
+			return err
+		}
+		gdp.TimeStamp = time.Unix(ts, 0)
+	default:
+		ts, ok := v[1].(int64)
+		if !ok {
+			return fmt.Errorf("error unmarshaling timestamp: %v", v[0])
+		}
+		gdp.TimeStamp = time.Unix(ts, 0)
+	}
+
+	return nil
+}
+
+type graphiteTargetResp struct {
+	Target     string              `json:"target"`
+	DataPoints []graphiteDataPoint `json:"datapoints"`
+}
+
+type graphiteResponse []graphiteTargetResp
+
+// GraphiteProvider executes Graphite render URL API queries.
+type GraphiteProvider struct {
+	url      url.URL
+	username string
+	password string
+	timeout  time.Duration
+}
+
+// NewGraphiteProvider takes a provider spec and credentials map,
+// validates the address, extracts the  credentials map's username
+// and password values if provided, and returns a Graphite client
+// ready to execute queries against the Graphite render URL API.
+func NewGraphiteProvider(provider flaggerv1.MetricTemplateProvider, credentials map[string][]byte) (*GraphiteProvider, error) {
+	graphiteURL, err := url.Parse(provider.Address)
+	if provider.Address == "" || err != nil {
+		return nil, fmt.Errorf("%s address %s is not a valid URL", provider.Type, provider.Address)
+	}
+
+	graph := GraphiteProvider{
+		url:     *graphiteURL,
+		timeout: 5 * time.Second,
+	}
+
+	if provider.SecretRef == nil {
+		return &graph, nil
+	}
+
+	if username, ok := credentials["username"]; ok {
+		graph.username = string(username)
+	} else {
+		return nil, fmt.Errorf("%s credentials does not contain a username", provider.Type)
+	}
+
+	if password, ok := credentials["password"]; ok {
+		graph.password = string(password)
+	} else {
+		return nil, fmt.Errorf("%s credentials does not contain a password", provider.Type)
+	}
+
+	return &graph, nil
+}
+
+// RunQuery executes the Graphite render URL API query and returns the
+// the first result as float64.
+func (g *GraphiteProvider) RunQuery(query string) (float64, error) {
+	query = g.trimQuery(query)
+	u, err := url.Parse(fmt.Sprintf("./render?%s", query))
+	if err != nil {
+		return 0, fmt.Errorf("url.Parase failed: %w", err)
+	}
+
+	q := u.Query()
+	q.Set("format", "json")
+	u.RawQuery = q.Encode()
+
+	u.Path = path.Join(g.url.Path, u.Path)
+	u = g.url.ResolveReference(u)
+
+	req, err := http.NewRequest("GET", u.String(), nil)
+	if err != nil {
+		return 0, fmt.Errorf("http.NewRequest failed: %w", err)
+	}
+
+	if g.username != "" && g.password != "" {
+		req.SetBasicAuth(g.username, g.password)
+	}
+
+	ctx, cancel := context.WithTimeout(req.Context(), g.timeout)
+	defer cancel()
+
+	r, err := http.DefaultClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return 0, fmt.Errorf("request failed: %w", err)
+	}
+	defer r.Body.Close()
+
+	b, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return 0, fmt.Errorf("error reading body: %w", err)
+	}
+
+	if 400 <= r.StatusCode {
+		return 0, fmt.Errorf("error response: %s", string(b))
+	}
+
+	var result graphiteResponse
+	err = json.Unmarshal(b, &result)
+	if err != nil {
+		return 0, fmt.Errorf("error unmarshaling result: %w, '%s'", err, string(b))
+	}
+
+	var value *float64
+	for _, tr := range result {
+		for _, dp := range tr.DataPoints {
+			if dp.Value != nil {
+				value = dp.Value
+			}
+		}
+	}
+	if value == nil {
+		return 0, ErrNoValuesFound
+	}
+
+	return *value, nil
+}
+
+// IsOnline runs a simple Graphite render URL API query and returns
+// an error if the API is unreachable.
+func (g *GraphiteProvider) IsOnline() (bool, error) {
+	_, err := g.RunQuery("target=test")
+	if err != nil && err != ErrNoValuesFound {
+		return false, fmt.Errorf("running query failed: %w", err)
+	}
+
+	return true, nil
+}
+
+// trimQuery removes whitespace from the query it's passed.
+func (g *GraphiteProvider) trimQuery(query string) string {
+	space := regexp.MustCompile(`\s+`)
+	return space.ReplaceAllString(query, " ")
+}

pkg/metrics/providers/graphite_test.go

@@ -0,0 +1,322 @@
+/*
+Copyright 2020 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package providers
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	fakeFlagger "github.com/fluxcd/flagger/pkg/client/clientset/versioned/fake"
+)
+
+func graphiteFake(query string) fakeClients {
+
+	provider := flaggerv1.MetricTemplateProvider{
+		Type:      "graphite",
+		Address:   "http://graphite:8080",
+		SecretRef: &corev1.LocalObjectReference{Name: "graphite"},
+	}
+
+	template := &flaggerv1.MetricTemplate{
+		TypeMeta: metav1.TypeMeta{APIVersion: flaggerv1.SchemeGroupVersion.String()},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default",
+			Name:      "graphite",
+		},
+		Spec: flaggerv1.MetricTemplateSpec{
+			Provider: provider,
+			Query:    query,
+		},
+	}
+
+	flaggerClient := fakeFlagger.NewSimpleClientset(template)
+
+	secret := &corev1.Secret{
+		TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "default",
+			Name:      "graphite",
+		},
+		Type: corev1.SecretTypeOpaque,
+		Data: map[string][]byte{
+			"username": []byte("username"),
+			"password": []byte("password"),
+		},
+	}
+
+	kubeClient := fake.NewSimpleClientset(secret)
+
+	return fakeClients{
+		kubeClient:    kubeClient,
+		flaggerClient: flaggerClient,
+	}
+}
+
+func TestNewGraphiteProvider(t *testing.T) {
+	secretRef := &corev1.LocalObjectReference{Name: "graphite"}
+	tests := []struct {
+		name           string
+		addr           string
+		secretRef      *corev1.LocalObjectReference
+		errExpected    bool
+		expectedErrStr string
+		credentials    map[string][]byte
+	}{{
+		name:           "a valid URL, a nil SecretRef, and an empty credentials map are specified",
+		addr:           "http://graphite:8080",
+		secretRef:      nil,
+		errExpected:    false,
+		expectedErrStr: "",
+		credentials:    map[string][]byte{},
+	}, {
+		name:           "an invalid URL is specified",
+		addr:           ":::",
+		secretRef:      nil,
+		errExpected:    true,
+		expectedErrStr: "graphite address ::: is not a valid URL",
+		credentials:    map[string][]byte{},
+	}, {
+		name:           "a valid URL, a SecretRef, and valid credentials are specified",
+		addr:           "http://graphite:8080",
+		secretRef:      secretRef,
+		errExpected:    false,
+		expectedErrStr: "",
+		credentials: map[string][]byte{
+			"username": []byte("a-username"),
+			"password": []byte("a-password"),
+		},
+	}, {
+		name:           "a valid URL, a SecretRef, and credentials without a username are specified",
+		addr:           "http://graphite:8080",
+		secretRef:      secretRef,
+		errExpected:    true,
+		expectedErrStr: "graphite credentials does not contain a username",
+		credentials: map[string][]byte{
+			"password": []byte("a-password"),
+		},
+	}, {
+		name:           "a valid URL, a SecretRef, and credentials without a password are specified",
+		addr:           "http://graphite:8080",
+		secretRef:      secretRef,
+		errExpected:    true,
+		expectedErrStr: "graphite credentials does not contain a password",
+		credentials: map[string][]byte{
+			"username": []byte("a-username"),
+		},
+	}, {
+		name:           "a valid URL, a nil SecretRef, and valid credentials are specified",
+		addr:           "http://graphite:8080",
+		secretRef:      nil,
+		errExpected:    false,
+		expectedErrStr: "",
+		credentials: map[string][]byte{
+			"username": []byte("a-username"),
+			"password": []byte("a-password"),
+		},
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			addr := test.addr
+			graph, err := NewGraphiteProvider(flaggerv1.MetricTemplateProvider{
+				Address:   addr,
+				Type:      "graphite",
+				SecretRef: test.secretRef,
+			}, test.credentials)
+
+			if test.errExpected {
+				require.Error(t, err)
+				assert.Equal(t, err.Error(), test.expectedErrStr)
+			} else {
+				username := ""
+				if uname, ok := test.credentials["username"]; ok && test.secretRef != nil {
+					username = string(uname)
+				}
+
+				password := ""
+				if pword, ok := test.credentials["password"]; ok && test.secretRef != nil {
+					password = string(pword)
+				}
+
+				require.NoError(t, err)
+				assert.Equal(t, addr, graph.url.String())
+				assert.Equal(t, username, graph.username)
+				assert.Equal(t, password, graph.password)
+			}
+		})
+	}
+}
+
+func TestGraphiteProvider_RunQuery(t *testing.T) {
+	tests := []struct {
+		name           string
+		query          string
+		expectedTarget string
+		expectedResult float64
+		errExpected    bool
+		body           string
+	}{{
+		"ok",
+		"target=sumSeries(app.http.*.*.count)&from=-2min",
+		"sumSeries(app.http.*.*.count)",
+		float64(100),
+		false,
+		`[
+			{
+				"datapoints": [
+					[
+						10,
+						1621348400
+					],
+					[
+						75,
+						1621348410
+					],
+					[
+						25,
+						1621348420
+					],
+					[
+						100,
+						1621348430
+					]
+				],
+				"target": "sumSeries(app.http.*.*.count)",
+				"tags": {
+					"aggregatedBy": "sum",
+					"name": "sumSeries(app.http.*.*.count)"
+				}
+			}
+		]`,
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				target := r.URL.Query().Get("target")
+				assert.Equal(t, test.expectedTarget, target)
+
+				header, ok := r.Header["Authorization"]
+				if assert.True(t, ok, "Authorization header not found") {
+					assert.True(t, strings.Contains(header[0], "Basic"), "Basic authorization header not found")
+				}
+
+				json := test.body
+				w.Write([]byte(json))
+			}))
+			defer ts.Close()
+
+			clients := graphiteFake(test.query)
+
+			template, err := clients.flaggerClient.FlaggerV1beta1().MetricTemplates("default").Get(context.TODO(), "graphite", metav1.GetOptions{})
+			require.NoError(t, err)
+			template.Spec.Provider.Address = ts.URL
+
+			secret, err := clients.kubeClient.CoreV1().Secrets("default").Get(context.TODO(), "graphite", metav1.GetOptions{})
+			require.NoError(t, err)
+
+			graphite, err := NewGraphiteProvider(template.Spec.Provider, secret.Data)
+			require.NoError(t, err)
+
+			val, err := graphite.RunQuery(template.Spec.Query)
+			require.NoError(t, err)
+
+			if test.errExpected {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, test.expectedResult, val)
+			}
+
+		})
+	}
+}
+
+func TestGraphiteProvider_IsOnline(t *testing.T) {
+	tests := []struct {
+		name           string
+		expectedResult bool
+		errExpected    bool
+		code           int
+		body           string
+	}{{
+		"Graphite responds 200 with valid JSON",
+		true,
+		false,
+		200,
+		"[]",
+	}, {
+		"Graphite responds 200 with invalid JSON",
+		false,
+		true,
+		200,
+		"[",
+	}, {
+		"Graphite responds 400",
+		false,
+		true,
+		400,
+		"error",
+	}, {
+		"Graphite responds 500",
+		false,
+		true,
+		500,
+		"error",
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.URL.Path != "/render" || r.URL.Query().Encode() != "format=json&target=test" {
+					w.WriteHeader(http.StatusNotFound)
+					return
+				}
+
+				w.WriteHeader(test.code)
+				fmt.Fprintf(w, test.body)
+			}))
+			defer ts.Close()
+
+			graph, err := NewGraphiteProvider(flaggerv1.MetricTemplateProvider{
+				Address: ts.URL,
+			}, map[string][]byte{})
+			require.NoError(t, err)
+
+			res, err := graph.IsOnline()
+			assert.Equal(t, res, test.expectedResult)
+
+			if test.errExpected {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}