From d154c63ac347862c0f97211d1b582da7cba98ebf Mon Sep 17 00:00:00 2001 From: stefanprodan Date: Wed, 29 Jan 2020 17:09:36 +0200 Subject: [PATCH] Implement Prometheus provider --- pkg/metrics/providers/factory.go | 15 ++ .../{client.go => providers/prometheus.go} | 109 ++++++----- pkg/metrics/providers/prometheus_test.go | 170 ++++++++++++++++++ pkg/metrics/providers/provider.go | 9 + 4 files changed, 246 insertions(+), 57 deletions(-) create mode 100644 pkg/metrics/providers/factory.go rename pkg/metrics/{client.go => providers/prometheus.go} (55%) create mode 100644 pkg/metrics/providers/prometheus_test.go create mode 100644 pkg/metrics/providers/provider.go diff --git a/pkg/metrics/providers/factory.go b/pkg/metrics/providers/factory.go new file mode 100644 index 00000000..e03b2aee --- /dev/null +++ b/pkg/metrics/providers/factory.go @@ -0,0 +1,15 @@ +package providers + +import flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha1" + +type Factory struct { +} + +func (factory Factory) Provider(provider flaggerv1.MetricTemplateProvider, credentials map[string][]byte) (Interface, error) { + switch { + case provider.Type == "prometheus": + return NewPrometheusProvider(provider, credentials) + default: + return NewPrometheusProvider(provider, credentials) + } +} diff --git a/pkg/metrics/client.go b/pkg/metrics/providers/prometheus.go similarity index 55% rename from pkg/metrics/client.go rename to pkg/metrics/providers/prometheus.go index a114125a..e04c03d2 100644 --- a/pkg/metrics/client.go +++ b/pkg/metrics/providers/prometheus.go @@ -1,8 +1,6 @@ -package metrics +package providers import ( - "bufio" - "bytes" "context" "encoding/json" "fmt" @@ -12,14 +10,17 @@ import ( "path" "regexp" "strconv" - "text/template" "time" + + flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha1" ) -// PrometheusClient is executing promql queries -type PrometheusClient struct { - timeout time.Duration - url url.URL +// PrometheusProvider executes promQL queries +type PrometheusProvider struct { + timeout time.Duration + url url.URL + username string + password string } type prometheusResponse struct { @@ -33,54 +34,44 @@ type prometheusResponse struct { } } -// NewPrometheusClient creates a Prometheus client for the provided URL address -func NewPrometheusClient(address string, timeout time.Duration) (*PrometheusClient, error) { - promURL, err := url.Parse(address) +// NewPrometheusProvider takes a provider spec and the credentials map, +// validates the address, extracts the username and password values if provided and +// returns a Prometheus client ready to execute queries against the API +func NewPrometheusProvider(provider flaggerv1.MetricTemplateProvider, credentials map[string][]byte) (*PrometheusProvider, error) { + promURL, err := url.Parse(provider.Address) if err != nil { - return nil, err + return nil, fmt.Errorf("%s address %s is not a valid URL", provider.Type, provider.Address) } - return &PrometheusClient{timeout: timeout, url: *promURL}, nil + prom := PrometheusProvider{ + timeout: 5 * time.Second, + url: *promURL, + } + + if provider.SecretRef != nil { + if username, ok := credentials["username"]; ok { + prom.username = string(username) + } else { + return nil, fmt.Errorf("%s credentials does not contain a username", provider.Type) + } + + if password, ok := credentials["password"]; ok { + prom.password = string(password) + } else { + return nil, fmt.Errorf("%s credentials does not contain a password", provider.Type) + } + } + + return &prom, nil } -// RenderQuery renders the promql query using the provided text template -func (p *PrometheusClient) RenderQuery(name string, namespace string, interval string, tmpl string) (string, error) { - meta := struct { - Name string - Namespace string - Interval string - }{ - name, - namespace, - interval, - } - - t, err := template.New("tmpl").Parse(tmpl) - if err != nil { - return "", err - } - var data bytes.Buffer - b := bufio.NewWriter(&data) - - if err := t.Execute(b, meta); err != nil { - return "", err - } - - err = b.Flush() - if err != nil { - return "", err - } - - return data.String(), nil -} - -// RunQuery executes the promql and converts the result to float64 -func (p *PrometheusClient) RunQuery(query string) (float64, error) { +// RunQuery executes the promQL query and returns the the first result as float64 +func (p *PrometheusProvider) RunQuery(query string) (float64, error) { if p.url.Host == "fake" { return 100, nil } - query = url.QueryEscape(p.TrimQuery(query)) + query = url.QueryEscape(p.trimQuery(query)) u, err := url.Parse(fmt.Sprintf("./api/v1/query?query=%s", query)) if err != nil { return 0, err @@ -94,6 +85,10 @@ func (p *PrometheusClient) RunQuery(query string) (float64, error) { return 0, err } + if p.username != "" && p.password != "" { + req.SetBasicAuth(p.username, p.password) + } + ctx, cancel := context.WithTimeout(req.Context(), p.timeout) defer cancel() @@ -137,14 +132,8 @@ func (p *PrometheusClient) RunQuery(query string) (float64, error) { return *value, nil } -// TrimQuery takes a promql query and removes whitespace -func (p *PrometheusClient) TrimQuery(query string) string { - space := regexp.MustCompile(`\s+`) - return space.ReplaceAllString(query, " ") -} - -// IsOnline call Prometheus status endpoint and returns an error if the API is unreachable -func (p *PrometheusClient) IsOnline() (bool, error) { +// IsOnline calls the Prometheus status endpoint and returns an error if the API is unreachable +func (p *PrometheusProvider) IsOnline() (bool, error) { u, err := url.Parse("./api/v1/status/flags") if err != nil { return false, err @@ -158,6 +147,10 @@ func (p *PrometheusClient) IsOnline() (bool, error) { return false, err } + if p.username != "" && p.password != "" { + req.SetBasicAuth(p.username, p.password) + } + ctx, cancel := context.WithTimeout(req.Context(), p.timeout) defer cancel() @@ -179,6 +172,8 @@ func (p *PrometheusClient) IsOnline() (bool, error) { return true, nil } -func (p *PrometheusClient) GetMetricsServer() string { - return p.url.String() +// trimQuery takes a promql query and removes whitespace +func (p *PrometheusProvider) trimQuery(query string) string { + space := regexp.MustCompile(`\s+`) + return space.ReplaceAllString(query, " ") } diff --git a/pkg/metrics/providers/prometheus_test.go b/pkg/metrics/providers/prometheus_test.go new file mode 100644 index 00000000..09d1f139 --- /dev/null +++ b/pkg/metrics/providers/prometheus_test.go @@ -0,0 +1,170 @@ +package providers + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" + + flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha1" + clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned" + fakeFlagger "github.com/weaveworks/flagger/pkg/client/clientset/versioned/fake" +) + +type fakeClients struct { + kubeClient kubernetes.Interface + flaggerClient clientset.Interface +} + +func prometheusFake() fakeClients { + provider := flaggerv1.MetricTemplateProvider{ + Type: "prometheus", + Address: "http://prometheus:9090", + SecretRef: &corev1.LocalObjectReference{Name: "prometheus"}, + } + + template := &flaggerv1.MetricTemplate{ + TypeMeta: metav1.TypeMeta{APIVersion: flaggerv1.SchemeGroupVersion.String()}, + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "prometheus", + }, + Spec: flaggerv1.MetricTemplateSpec{ + Provider: provider, + Query: "sum(envoy_cluster_upstream_rq)", + }, + } + + flaggerClient := fakeFlagger.NewSimpleClientset(template) + + secret := &corev1.Secret{ + TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()}, + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "prometheus", + }, + Type: corev1.SecretTypeOpaque, + Data: map[string][]byte{ + "username": []byte("username"), + "password": []byte("password"), + }, + } + + kubeClient := fake.NewSimpleClientset(secret) + + return fakeClients{ + kubeClient: kubeClient, + flaggerClient: flaggerClient, + } +} + +func TestNewPrometheusProvider(t *testing.T) { + clients := prometheusFake() + + template, err := clients.flaggerClient.FlaggerV1alpha1().MetricTemplates("default").Get("prometheus", metav1.GetOptions{}) + if err != nil { + t.Fatal(err.Error()) + } + + secret, err := clients.kubeClient.CoreV1().Secrets("default").Get("prometheus", metav1.GetOptions{}) + if err != nil { + t.Fatal(err.Error()) + } + + prom, err := NewPrometheusProvider(template.Spec.Provider, secret.Data) + if err != nil { + t.Fatal(err.Error()) + } + + if prom.url.String() != "http://prometheus:9090" { + t.Errorf("Got URL %s wanted %s", prom.url.String(), "http://prometheus:9090") + } + + if prom.password != "password" { + t.Errorf("Got password %s wanted %s", prom.password, "password") + } +} + +func TestPrometheusProvider_RunQueryWithBasicAuth(t *testing.T) { + expected := `sum(envoy_cluster_upstream_rq)` + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + promql := r.URL.Query()["query"][0] + if promql != expected { + t.Errorf("\nGot %s \nWanted %s", promql, expected) + } + + if header, ok := r.Header["Authorization"]; ok { + if !strings.Contains(header[0], "Basic") { + t.Error("Basic authorization header not found") + } + } else { + t.Error("Authorization header not found") + } + + json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1545905245.458,"100"]}]}}` + w.Write([]byte(json)) + })) + defer ts.Close() + + clients := prometheusFake() + + template, err := clients.flaggerClient.FlaggerV1alpha1().MetricTemplates("default").Get("prometheus", metav1.GetOptions{}) + if err != nil { + t.Fatal(err.Error()) + } + template.Spec.Provider.Address = ts.URL + + secret, err := clients.kubeClient.CoreV1().Secrets("default").Get("prometheus", metav1.GetOptions{}) + if err != nil { + t.Fatal(err.Error()) + } + + prom, err := NewPrometheusProvider(template.Spec.Provider, secret.Data) + if err != nil { + t.Fatal(err.Error()) + } + + val, err := prom.RunQuery(template.Spec.Query) + if err != nil { + t.Fatal(err.Error()) + } + + if val != 100 { + t.Errorf("Got %v wanted %v", val, 100) + } +} + +func TestPrometheusProvider_IsOnline(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadGateway) + })) + defer ts.Close() + + clients := prometheusFake() + + template, err := clients.flaggerClient.FlaggerV1alpha1().MetricTemplates("default").Get("prometheus", metav1.GetOptions{}) + if err != nil { + t.Fatal(err.Error()) + } + template.Spec.Provider.Address = ts.URL + template.Spec.Provider.SecretRef = nil + + prom, err := NewPrometheusProvider(template.Spec.Provider, nil) + if err != nil { + t.Fatal(err.Error()) + } + + ok, err := prom.IsOnline() + if err == nil { + t.Errorf("Got no error wanted %v", http.StatusBadGateway) + } + + if ok { + t.Errorf("Got %v wanted %v", ok, false) + } +} diff --git a/pkg/metrics/providers/provider.go b/pkg/metrics/providers/provider.go new file mode 100644 index 00000000..e64bd947 --- /dev/null +++ b/pkg/metrics/providers/provider.go @@ -0,0 +1,9 @@ +package providers + +type Interface interface { + // RunQuery executes the query and converts the first result to float64 + RunQuery(query string) (float64, error) + + // IsOnline calls the provider endpoint and returns an error if the API is unreachable + IsOnline() (bool, error) +}