github/flagger
1
0
Fork 0
mirror of https://github.com/fluxcd/flagger.git synced 2026-04-15 06:57:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement Prometheus provider

Browse Source
This commit is contained in:
stefanprodan
2020-02-06 15:07:53 +02:00
parent d9252748d2
commit d154c63ac3
4 changed files with 246 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/metrics/providers/factory.go
+15
View File
@@ -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)
}
}
pkg/metrics/client.go → pkg/metrics/providers/prometheus.go
+52 -57
View File
@@ -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, " ")
}
pkg/metrics/providers/prometheus_test.go
+170
View File
@@ -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)
}
}
pkg/metrics/providers/provider.go
+9
View File
@@ -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)
}