mirror of
https://github.com/fluxcd/flagger.git
synced 2026-04-15 06:57:34 +00:00
add a Graphite metrics provider
This adds a Graphite metric provider to address issue #911. Co-authored-by: Mike Ball <mike.ball@warnermedia.com> Co-authored-by: Nathan Mische <Nathan.Mische@warnermedia.com> Signed-off-by: Mike Ball <mike.ball@warnermedia.com> Signed-off-by: Nathan Mische <Nathan.Mische@warnermedia.com>
This commit is contained in:
Mike Ball
committed by
Mike Ball
Mike Ball
co-authored by
Nathan Mische
Nathan Mische
parent
39a3898234
commit
4654f2cba9
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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, " ")
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user