[traefik] Implement observer interface

pkg/metrics/observers/factory.go

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

pkg/metrics/observers/traefik.go

@@ -0,0 +1,76 @@
+package observers
+
+import (
+	"fmt"
+	"time"
+
+	flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1beta1"
+	"github.com/weaveworks/flagger/pkg/metrics/providers"
+)
+
+var traefikQueries = map[string]string{
+	"request-success-rate": `
+	sum(
+		rate(
+			traefik_service_request_duration_seconds_bucket{
+				service=~"{{ namespace }}-{{ target }}-canary-[0-9a-zA-Z-]+@kubernetescrd",
+				code!~"5..",
+				le="+Inf"
+			}[{{ interval }}]
+		)
+	)
+	/
+	sum(
+		rate(
+			traefik_service_request_duration_seconds_bucket{
+				service=~"{{ namespace }}-{{ target }}-canary-[0-9a-zA-Z-]+@kubernetescrd",
+				le="+Inf"
+			}[{{ interval }}]
+		)
+	) * 100`,
+	"request-duration": `
+	histogram_quantile(
+		0.99,
+		sum(
+			rate(
+				traefik_service_request_duration_seconds_bucket{
+					service=~"{{ namespace }}-{{ target }}-canary-[0-9a-zA-Z-]+@kubernetescrd"
+				}[{{ interval }}]
+			)
+		) by (le)
+	)`,
+}
+
+type TraefikObserver struct {
+	client providers.Interface
+}
+
+func (ob *TraefikObserver) GetRequestSuccessRate(model flaggerv1.MetricTemplateModel) (float64, error) {
+
+	query, err := RenderQuery(traefikQueries["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
+}
+
+func (ob *TraefikObserver) GetRequestDuration(model flaggerv1.MetricTemplateModel) (time.Duration, error) {
+	query, err := RenderQuery(traefikQueries["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
+}

pkg/metrics/observers/traefik_test.go

@@ -0,0 +1,102 @@
+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 TestTraefikObserver_GetRequestSuccessRate(t *testing.T) {
+	t.Run("ok", func(t *testing.T) {
+		expected := ` sum( rate( traefik_service_request_duration_seconds_bucket{ service=~"default-podinfo-canary-[0-9a-zA-Z-]+@kubernetescrd", code!~"5..", le="+Inf" }[1m] ) ) / sum( rate( traefik_service_request_duration_seconds_bucket{ service=~"default-podinfo-canary-[0-9a-zA-Z-]+@kubernetescrd", 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 := &TraefikObserver{client: client}
+
+		val, err := observer.GetRequestSuccessRate(flaggerv1.MetricTemplateModel{
+			Name:      "podinfo",
+			Namespace: "default",
+			Target:    "podinfo",
+			Service:   "podinfo",
+			Interval:  "1m",
+		})
+		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 := &TraefikObserver{client: client}
+		_, err = observer.GetRequestSuccessRate(flaggerv1.MetricTemplateModel{})
+		require.True(t, errors.Is(err, providers.ErrNoValuesFound))
+	})
+}
+
+func TestTraefikObserver_GetRequestDuration(t *testing.T) {
+	expected := ` histogram_quantile( 0.99, sum( rate( traefik_service_request_duration_seconds_bucket{ service=~"default-podinfo-canary-[0-9a-zA-Z-]+@kubernetescrd" }[1m] ) ) by (le) )`
+
+	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 := &TraefikObserver{client: client}
+
+	val, err := observer.GetRequestDuration(flaggerv1.MetricTemplateModel{
+		Name:      "podinfo",
+		Namespace: "default",
+		Target:    "podinfo",
+		Service:   "podinfo",
+		Interval:  "1m",
+	})
+	require.NoError(t, err)
+
+	assert.Equal(t, 100*time.Millisecond, val)
+}