✨ add skipper observer to flagger

pkg/apis/flagger/v1beta1/provider.go

@@ -9,4 +9,5 @@ const (
 	GlooProvider       string = "gloo"
 	NGINXProvider      string = "nginx"
 	KubernetesProvider string = "kubernetes"
+	SkipperProvider    string = "skipper"
 )

pkg/metrics/observers/factory.go

@@ -56,6 +56,10 @@ func (factory Factory) Observer(provider string) Interface {
 		return &HttpObserver{
 			client: factory.Client,
 		}
+	case provider == flaggerv1.SkipperProvider:
+		return &SkipperObserver{
+			client: factory.Client,
+		}
 	default:
 		return &IstioObserver{
 			client: factory.Client,

pkg/metrics/observers/skipper.go

@@ -0,0 +1,110 @@
+package observers
+
+import (
+	"fmt"
+	"regexp"
+	"time"
+
+	flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+	"github.com/weaveworks/flagger/pkg/metrics/providers"
+)
+
+var skipperQueries = map[string]string{
+	"request-success-rate": `
+	{{- $route := printf "kube_%s__%s.*__%s" namespace ingress service }}
+	sum(
+		rate(
+			skipper_response_duration_seconds_bucket{
+				namespace="{{ namespace }}",
+				route=~"{{ $route }}",
+				code!~"5..",
+				le="+Inf"
+			}[{{ interval }}]
+		)
+	)
+	/ 
+	sum(
+		rate(
+			skipper_response_duration_seconds_bucket{
+				namespace="{{ namespace }}",
+				route=~"{{ $route }}",
+				le="+Inf"
+			}[{{ interval }}]
+		)
+	)
+	* 100`,
+	"request-duration": `
+	{{- $route := printf "kube_%s__%s.*__%s" namespace ingress service }}
+	sum(
+		rate(
+			skipper_response_duration_seconds_sum{
+				namespace="{{ namespace }}",
+				route=~"{{ $route }}"
+			}[{{ interval }}]
+		)
+	) 
+	/ 
+	sum(
+		rate(
+			skipper_response_duration_seconds_count{
+				namespace="{{ namespace }}",
+				route=~"{{ $route }}"
+			}[{{ interval }}]
+		)
+	) 
+	* 1000`,
+}
+
+// SkipperObserver Implentation for Skipper (https://github.com/zalando/skipper)
+type SkipperObserver struct {
+	client providers.Interface
+}
+
+// GetRequestSuccessRate return value for Skipper Request Success Rate
+func (ob *SkipperObserver) GetRequestSuccessRate(model flaggerv1.MetricTemplateModel) (float64, error) {
+
+	model = encodeModelForSkipper(model)
+
+	query, err := RenderQuery(skipperQueries["request-success-rate"], model)
+	if err != nil {
+		return 0, fmt.Errorf("rendering query failed: %w", err)
+	}
+
+	value, err := ob.client.RunQuery(query)
+	if err != nil {
+		return 0, fmt.Errorf("running query failed: %w", err)
+	}
+
+	return value, nil
+}
+
+// GetRequestDuration return value for Skipper Request Duration
+func (ob *SkipperObserver) GetRequestDuration(model flaggerv1.MetricTemplateModel) (time.Duration, error) {
+
+	model = encodeModelForSkipper(model)
+
+	query, err := RenderQuery(skipperQueries["request-duration"], model)
+	if err != nil {
+		return 0, fmt.Errorf("rendering query failed: %w", err)
+	}
+
+	value, err := ob.client.RunQuery(query)
+	if err != nil {
+		return 0, fmt.Errorf("running query failed: %w", err)
+	}
+
+	ms := time.Duration(int64(value)) * time.Millisecond
+	return ms, nil
+}
+
+// encodeModelForSkipper replaces non word character in model with underscore to match route names
+// https://github.com/zalando/skipper/blob/dd70bd65e7f99cfb5dd6b6f71885d9fe3b2707f6/dataclients/kubernetes/ingress.go#L101
+func encodeModelForSkipper(model flaggerv1.MetricTemplateModel) flaggerv1.MetricTemplateModel {
+	nonWord := regexp.MustCompile(`\W`)
+	model.Ingress = nonWord.ReplaceAllString(model.Ingress, "_")
+	model.Name = nonWord.ReplaceAllString(model.Name, "_")
+	model.Namespace = nonWord.ReplaceAllString(model.Namespace, "_")
+	model.Service = nonWord.ReplaceAllString(model.Service, "_")
+	model.Target = nonWord.ReplaceAllString(model.Target, "_")
+	return model
+}

pkg/metrics/observers/skipper_test.go

@@ -0,0 +1,106 @@
+package observers
+
+import (
+	"errors"
+	"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"
+	"github.com/weaveworks/flagger/pkg/metrics/providers"
+)
+
+func TestSkipperObserver_GetRequestSuccessRate(t *testing.T) {
+	t.Run("ok", func(t *testing.T) {
+		expected := ` sum( rate( skipper_response_duration_seconds_bucket{ namespace="skipper", route=~"kube_skipper__skipper_ingress.*__backend", code=~"[4|5]..", le="+Inf" }[1m] ) ) / sum( rate( skipper_response_duration_seconds_bucket{ namespace="skipper", route=~"kube_skipper__skipper_ingress.*__backend", le="+Inf" }[1m] ) ) * 100`
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			promql := r.URL.Query()["query"][0]
+			assert.Equal(t, expected, promql)
+
+			json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1,"100"]}]}}`
+			w.Write([]byte(json))
+		}))
+		defer ts.Close()
+
+		client, err := providers.NewPrometheusProvider(flaggerv1.MetricTemplateProvider{
+			Type:      "prometheus",
+			Address:   ts.URL,
+			SecretRef: nil,
+		}, nil)
+		require.NoError(t, err)
+
+		observer := &SkipperObserver{
+			client: client,
+		}
+
+		val, err := observer.GetRequestSuccessRate(flaggerv1.MetricTemplateModel{
+			Namespace: "skipper",
+			Interval:  "1m",
+			Service:   "backend",
+			Ingress:   "skipper-ingress",
+		})
+		require.NoError(t, err)
+
+		assert.Equal(t, float64(100), val)
+	})
+
+	t.Run("no values", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			json := `{"status":"success","data":{"resultType":"vector","result":[]}}`
+			w.Write([]byte(json))
+		}))
+		defer ts.Close()
+
+		client, err := providers.NewPrometheusProvider(flaggerv1.MetricTemplateProvider{
+			Type:      "prometheus",
+			Address:   ts.URL,
+			SecretRef: nil,
+		}, nil)
+		require.NoError(t, err)
+
+		observer := &SkipperObserver{
+			client: client,
+		}
+
+		_, err = observer.GetRequestSuccessRate(flaggerv1.MetricTemplateModel{})
+		require.True(t, errors.Is(err, providers.ErrNoValuesFound))
+	})
+}
+
+func TestSkipperObserver_GetRequestDuration(t *testing.T) {
+	expected := ` sum( rate( skipper_response_duration_seconds_sum{ namespace="skipper", route=~"kube_skipper__skipper_ingress.*__backend" }[1m] ) ) / sum( rate( skipper_response_duration_seconds_count{ namespace="skipper", route=~"kube_skipper__skipper_ingress.*__backend" }[1m] ) ) * 1000`
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		promql := r.URL.Query()["query"][0]
+		assert.Equal(t, expected, promql)
+
+		json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1,"100"]}]}}`
+		w.Write([]byte(json))
+	}))
+	defer ts.Close()
+
+	client, err := providers.NewPrometheusProvider(flaggerv1.MetricTemplateProvider{
+		Type:      "prometheus",
+		Address:   ts.URL,
+		SecretRef: nil,
+	}, nil)
+	require.NoError(t, err)
+
+	observer := &SkipperObserver{
+		client: client,
+	}
+
+	val, err := observer.GetRequestDuration(flaggerv1.MetricTemplateModel{
+		Namespace: "skipper",
+		Interval:  "1m",
+		Service:   "backend",
+		Ingress:   "skipper-ingress",
+	})
+	require.NoError(t, err)
+
+	assert.Equal(t, 100*time.Millisecond, val)
+}