Add New Relic as a metrics provider

pkg/metrics/providers/factory.go

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

pkg/metrics/providers/newrelic.go

@@ -0,0 +1,159 @@
+package providers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"time"
+
+	flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+)
+
+const (
+	newrelicInsightsDefaultHost = "https://insights-api.newrelic.com"
+
+	newrelicQueryKeySecretKey  = "newrelic_query_key"
+	newrelicAccountIdSecretKey = "newrelic_account_id"
+
+	newrelicQueryKeyHeaderKey = "X-Query-Key"
+)
+
+// NewRelicProvider executes newrelic queries
+type NewRelicProvider struct {
+	insightsQueryEndpoint string
+
+	timeout   time.Duration
+	queryKey  string
+	fromDelta int64
+}
+
+type newRelicResponse struct {
+	Results []struct {
+		Result *float64 `json:"result"`
+	} `json:"results"`
+}
+
+// NewNewRelicProvider takes a canary spec, a provider spec and the credentials map, and
+// returns a NewRelic client ready to execute queries against the Insights API
+func NewNewRelicProvider(
+	metricInterval string,
+	provider flaggerv1.MetricTemplateProvider,
+	credentials map[string][]byte,
+) (*NewRelicProvider, error) {
+	address := provider.Address
+	if address == "" {
+		address = newrelicInsightsDefaultHost
+	}
+
+	accountId, ok := credentials[newrelicAccountIdSecretKey]
+	if !ok {
+		return nil, fmt.Errorf("newrelic credentials does not contain the key '%s'", newrelicAccountIdSecretKey)
+	}
+
+	queryEndpoint := fmt.Sprintf("%s/v1/accounts/%s/query", address, accountId)
+	nr := NewRelicProvider{
+		timeout:               5 * time.Second,
+		insightsQueryEndpoint: queryEndpoint,
+	}
+
+	if b, ok := credentials[newrelicQueryKeySecretKey]; ok {
+		nr.queryKey = string(b)
+	} else {
+		return nil, fmt.Errorf("newrelic credentials does not contain the key ''%s", newrelicQueryKeySecretKey)
+	}
+
+	md, err := time.ParseDuration(metricInterval)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing metric interval: %w", err)
+	}
+
+	nr.fromDelta = int64(md.Seconds())
+	return &nr, nil
+}
+
+// RunQuery executes the new relic query against the New Relic Insights API
+// and returns the the first result
+func (p *NewRelicProvider) RunQuery(query string) (float64, error) {
+	req, err := p.newInsightsRequest(query)
+	if err != nil {
+		return 0, err
+	}
+
+	ctx, cancel := context.WithTimeout(req.Context(), p.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 r.StatusCode != http.StatusOK {
+		return 0, fmt.Errorf("error response: %s: %w", string(b), err)
+	}
+
+	var res newRelicResponse
+	if err := json.Unmarshal(b, &res); err != nil {
+		return 0, fmt.Errorf("error unmarshaling result: %w, '%s'", err, string(b))
+	}
+
+	if len(res.Results) != 1 {
+		return 0, fmt.Errorf("invalid response: %s: %w", string(b), ErrNoValuesFound)
+	}
+
+	if res.Results[0].Result == nil {
+		return 0, fmt.Errorf("invalid response: %s: %w", string(b), ErrNoValuesFound)
+	}
+
+	return *res.Results[0].Result, nil
+}
+
+// IsOnline calls the NewRelic's insights API with
+// and returns an error if the request is rejected
+func (p *NewRelicProvider) IsOnline() (bool, error) {
+	req, err := p.newInsightsRequest("SELECT * FROM Metric")
+	if err != nil {
+		return false, fmt.Errorf("error http.NewRequest: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(req.Context(), p.timeout)
+	defer cancel()
+	r, err := http.DefaultClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return false, fmt.Errorf("request failed: %w", err)
+	}
+
+	defer r.Body.Close()
+
+	b, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		return false, fmt.Errorf("error reading body: %w", err)
+	}
+
+	if r.StatusCode != http.StatusOK {
+		return false, fmt.Errorf("error response: %s", string(b))
+	}
+
+	return true, nil
+}
+
+func (p *NewRelicProvider) newInsightsRequest(query string) (*http.Request, error) {
+	req, err := http.NewRequest("GET", p.insightsQueryEndpoint, nil)
+	if err != nil {
+		return nil, fmt.Errorf("error http.NewRequest: %w", err)
+	}
+
+	req.Header.Set(newrelicQueryKeyHeaderKey, p.queryKey)
+
+	q := req.URL.Query()
+	q.Add("nrql", fmt.Sprintf("%s SINCE %d seconds ago", query, p.fromDelta))
+	req.URL.RawQuery = q.Encode()
+
+	return req, nil
+}

pkg/metrics/providers/newrelic_test.go

@@ -0,0 +1,126 @@
+package providers
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+)
+
+func TestNewNewRelicProvider(t *testing.T) {
+	queryKey := "query-key"
+	accountId := "51312"
+	cs := map[string][]byte{
+		"newrelic_query_key":  []byte(queryKey),
+		"newrelic_account_id": []byte(accountId),
+	}
+
+	duration := "100s"
+	secondsDuration, err := time.ParseDuration(duration)
+	require.NoError(t, err)
+
+	nr, err := NewNewRelicProvider("100s", flaggerv1.MetricTemplateProvider{}, cs)
+	require.NoError(t, err)
+	assert.Equal(t, "https://insights-api.newrelic.com/v1/accounts/51312/query", nr.insightsQueryEndpoint)
+	assert.Equal(t, int64(secondsDuration.Seconds()), nr.fromDelta)
+	assert.Equal(t, queryKey, nr.queryKey)
+}
+
+func TestNewRelicProvider_RunQuery(t *testing.T) {
+	queryKey := "query-key"
+	accountId := "51312"
+	t.Run("ok", func(t *testing.T) {
+		q := `SELECT sum(nginx_ingress_controller_requests) / 1 FROM Metric WHERE status = '200'`
+		eq := `SELECT sum(nginx_ingress_controller_requests) / 1 FROM Metric WHERE status = '200' SINCE 60 seconds ago`
+		er := 1.11111
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			aq := r.URL.Query().Get("nrql")
+			assert.Equal(t, eq, aq)
+			assert.Equal(t, queryKey, r.Header.Get(newrelicQueryKeyHeaderKey))
+
+			json := fmt.Sprintf(`{"results":[{"result": %f}]}`, er)
+			w.Write([]byte(json))
+		}))
+		defer ts.Close()
+
+		nr, err := NewNewRelicProvider("1m",
+			flaggerv1.MetricTemplateProvider{
+				Address: ts.URL,
+			},
+			map[string][]byte{
+				"newrelic_query_key":  []byte(queryKey),
+				"newrelic_account_id": []byte(accountId),
+			},
+		)
+		require.NoError(t, err)
+
+		f, err := nr.RunQuery(q)
+		assert.NoError(t, err)
+		assert.Equal(t, er, f)
+	})
+
+	t.Run("no values", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			json := fmt.Sprintf(`{"results": []}`)
+			w.Write([]byte(json))
+		}))
+		defer ts.Close()
+
+		dp, err := NewNewRelicProvider(
+			"1m",
+			flaggerv1.MetricTemplateProvider{Address: ts.URL},
+			map[string][]byte{
+				"newrelic_query_key":  []byte(queryKey),
+				"newrelic_account_id": []byte(accountId)},
+		)
+		require.NoError(t, err)
+		_, err = dp.RunQuery("")
+		require.True(t, errors.Is(err, ErrNoValuesFound))
+	})
+}
+
+func TestNewReelicProvider_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) {
+			queryKey := "query-key"
+			accountId := "51312"
+			query := `SELECT * FROM Metric SINCE 60 seconds ago`
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				assert.Equal(t, queryKey, r.Header.Get(newrelicQueryKeyHeaderKey))
+				assert.Equal(t, query, r.URL.Query().Get("nrql"))
+				w.WriteHeader(c.code)
+			}))
+			defer ts.Close()
+
+			dp, err := NewNewRelicProvider(
+				"1m",
+				flaggerv1.MetricTemplateProvider{Address: ts.URL},
+				map[string][]byte{
+					"newrelic_query_key":  []byte(queryKey),
+					"newrelic_account_id": []byte(accountId),
+				},
+			)
+			require.NoError(t, err)
+
+			_, err = dp.IsOnline()
+			if c.errExpected {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}