Add Kuma routing and metrics

Signed-off-by: John Harris <john.harris@konghq.com>

pkg/metrics/observers/factory.go

@@ -84,6 +84,10 @@ func (factory Factory) Observer(provider string) Interface {
 		return &OsmObserver{
 			client: factory.Client,
 		}
+	case provider == flaggerv1.KumaProvider:
+		return &KumaObserver{
+			client: factory.Client,
+		}
 	default:
 		return &IstioObserver{
 			client: factory.Client,

pkg/metrics/observers/kuma.go

@@ -0,0 +1,94 @@
+/*
+Copyright 2021 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 observers
+
+import (
+	"fmt"
+	"time"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	"github.com/fluxcd/flagger/pkg/metrics/providers"
+)
+
+// TODO [@johnharris85]: Do we also need to select by mesh here? These could be duplicated (but in different meshes).
+// 	We're currently getting the mesh name from an annotation on the Canary object, but that isn't propagated to the
+// 	MetricTemplateModel.
+var kumaQueries = map[string]string{
+	"request-success-rate": `
+	sum(
+		rate(
+			envoy_cluster_upstream_rq{
+				envoy_cluster_name=~"{{ target }}-canary_{{ namespace }}_svc_[0-9a-zA-Z-]+",
+				envoy_response_code!~"5.*"
+			}[{{ interval }}]
+		)
+	) 
+	/ 
+	sum(
+		rate(
+			envoy_cluster_upstream_rq{
+				envoy_cluster_name=~"{{ target }}-canary_{{ namespace }}_svc_[0-9a-zA-Z-]+",
+			}[{{ interval }}]
+		)
+	) 
+	* 100`,
+	"request-duration": `
+	histogram_quantile(
+		0.99,
+		sum(
+			rate(
+				envoy_cluster_upstream_rq_time_bucket{
+					envoy_cluster_name=~"{{ target }}-canary_{{ namespace }}_svc_[0-9a-zA-Z-]+",
+				}[{{ interval }}]
+			)
+		) by (le)
+	)`,
+}
+
+type KumaObserver struct {
+	client providers.Interface
+}
+
+func (ob *KumaObserver) GetRequestSuccessRate(model flaggerv1.MetricTemplateModel) (float64, error) {
+	query, err := RenderQuery(kumaQueries["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 *KumaObserver) GetRequestDuration(model flaggerv1.MetricTemplateModel) (time.Duration, error) {
+	query, err := RenderQuery(kumaQueries["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/kuma_test.go

@@ -0,0 +1,100 @@
+/*
+Copyright 2021 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 observers
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	"github.com/fluxcd/flagger/pkg/metrics/providers"
+)
+
+func TestKumaObserver_GetRequestSuccessRate(t *testing.T) {
+	expected := ` sum( rate( envoy_cluster_upstream_rq{ envoy_cluster_name=~"podinfo-canary_default_svc_[0-9a-zA-Z-]+", envoy_response_code!~"5.*" }[1m] ) ) / sum( rate( envoy_cluster_upstream_rq{ envoy_cluster_name=~"podinfo-canary_default_svc_[0-9a-zA-Z-]+", }[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 := &KumaObserver{
+		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)
+}
+
+func TestKumaObserver_GetRequestDuration(t *testing.T) {
+	expected := ` histogram_quantile( 0.99, sum( rate( envoy_cluster_upstream_rq_time_bucket{ envoy_cluster_name=~"podinfo-canary_default_svc_[0-9a-zA-Z-]+", }[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 := &KumaObserver{
+		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)
+}

pkg/router/factory.go

@@ -170,6 +170,13 @@ func (factory *Factory) MeshRouter(provider string, labelSelector string) Interf
 			smiClient:     factory.meshClient,
 			targetMesh:    flaggerv1.OsmProvider,
 		}
+	case provider == flaggerv1.KumaProvider:
+		return &KumaRouter{
+			logger:        factory.logger,
+			flaggerClient: factory.flaggerClient,
+			kubeClient:    factory.kubeClient,
+			kumaClient:    factory.meshClient,
+		}
 	case provider == flaggerv1.KubernetesProvider:
 		return &NopRouter{}
 	default:

pkg/router/kuma.go

@@ -0,0 +1,212 @@
+package router
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	kumav1alpha1 "github.com/fluxcd/flagger/pkg/apis/kuma/v1alpha1"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"go.uber.org/zap"
+	"k8s.io/client-go/kubernetes"
+
+	clientset "github.com/fluxcd/flagger/pkg/client/clientset/versioned"
+)
+
+// KumaRouter is managing TrafficRoute objects
+type KumaRouter struct {
+	kubeClient    kubernetes.Interface
+	kumaClient    clientset.Interface
+	flaggerClient clientset.Interface
+	logger        *zap.SugaredLogger
+}
+
+// Reconcile creates or updates the Kuma TrafficRoute
+func (kr *KumaRouter) Reconcile(canary *flaggerv1.Canary) error {
+	apexName, primaryName, canaryName := canary.GetServiceNames()
+
+	trSpec := kumav1alpha1.TrafficRouteSpec{
+		Sources: []*kumav1alpha1.Selector{
+			{
+				Match: map[string]string{
+					"kuma.io/service": "*",
+				},
+			},
+		},
+		Destinations: []*kumav1alpha1.Selector{
+			{
+				Match: map[string]string{
+					"kuma.io/service": fmt.Sprintf("%s_%s_svc_%d", apexName, canary.Namespace, canary.Spec.Service.Port),
+				},
+			},
+		},
+		Conf: &kumav1alpha1.TrafficRouteConf{
+			Split: []*kumav1alpha1.TrafficRouteSplit{
+				{
+					Weight: uint32(100),
+					Destination: map[string]string{
+						"kuma.io/service": fmt.Sprintf("%s_%s_svc_%d", primaryName, canary.Namespace, canary.Spec.Service.Port),
+					},
+				},
+				{
+					Weight: uint32(0),
+					Destination: map[string]string{
+						"kuma.io/service": fmt.Sprintf("%s_%s_svc_%d", canaryName, canary.Namespace, canary.Spec.Service.Port),
+					},
+				},
+			},
+		},
+	}
+
+	tr, err := kr.kumaClient.KumaV1alpha1().TrafficRoutes().Get(context.TODO(), apexName, metav1.GetOptions{})
+
+	// create TrafficRoute
+	if errors.IsNotFound(err) {
+		metadata := canary.Spec.Service.Apex
+		if metadata == nil {
+			metadata = &flaggerv1.CustomMetadata{}
+		}
+		if metadata.Labels == nil {
+			metadata.Labels = make(map[string]string)
+		}
+		if metadata.Annotations == nil {
+			metadata.Annotations = make(map[string]string)
+			metadata.Annotations[fmt.Sprintf("%d.service.kuma.io", canary.Spec.Service.Port)] = "http"
+		}
+
+		meshName, ok := canary.Annotations["kuma.io/mesh"]
+		if !ok {
+			meshName = "default"
+		}
+
+		t := &kumav1alpha1.TrafficRoute{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: apexName,
+				OwnerReferences: []metav1.OwnerReference{
+					*metav1.NewControllerRef(canary, schema.GroupVersionKind{
+						Group:   flaggerv1.SchemeGroupVersion.Group,
+						Version: flaggerv1.SchemeGroupVersion.Version,
+						Kind:    flaggerv1.CanaryKind,
+					}),
+				},
+				Annotations: filterMetadata(metadata.Annotations),
+			},
+			Spec: trSpec,
+			Mesh: meshName,
+		}
+
+		_, err := kr.kumaClient.KumaV1alpha1().TrafficRoutes().Create(context.TODO(), t, metav1.CreateOptions{})
+
+		if err != nil {
+			return fmt.Errorf("TrafficRoute %s create error: %w", apexName, err)
+		}
+
+		kr.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
+			Infof("TrafficRoute %s created", t.GetName())
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("TrafficRoute %s get query error: %w", apexName, err)
+	}
+
+	// update TrafficRoute
+	if diff := cmp.Diff(trSpec, tr.Spec, cmpopts.IgnoreFields(kumav1alpha1.TrafficRouteSplit{}, "Weight")); diff != "" {
+		trClone := tr.DeepCopy()
+		trClone.Spec = trSpec
+
+		_, err := kr.kumaClient.KumaV1alpha1().TrafficRoutes().Update(context.TODO(), trClone, metav1.UpdateOptions{})
+
+		if err != nil {
+			return fmt.Errorf("TrafficRoute %s update error: %w", apexName, err)
+		}
+
+		kr.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
+			Infof("TrafficRoute %s.%s updated", apexName, canary.Namespace)
+		return nil
+	}
+
+	return nil
+}
+
+// GetRoutes returns the destinations weight for primary and canary
+func (kr *KumaRouter) GetRoutes(canary *flaggerv1.Canary) (
+	primaryWeight int,
+	canaryWeight int,
+	mirrored bool,
+	err error,
+) {
+	apexName, primaryName, canaryName := canary.GetServiceNames()
+	tr, err := kr.kumaClient.KumaV1alpha1().TrafficRoutes().Get(context.TODO(), apexName, metav1.GetOptions{})
+
+	if err != nil {
+		err = fmt.Errorf("TrafficRoute %s get query error %v", apexName, err)
+		return
+	}
+
+	for _, split := range tr.Spec.Conf.Split {
+		if strings.Split(split.Destination["kuma.io/service"], "_")[0] == primaryName {
+			primaryWeight = int(split.Weight)
+			canaryWeight = 100 - primaryWeight
+		}
+	}
+
+	if primaryWeight == 0 && canaryWeight == 0 {
+		err = fmt.Errorf("TrafficRoute %s does not contain routes for %s and %s",
+			apexName, primaryName, canaryName)
+	}
+
+	mirrored = false
+
+	return
+}
+
+// SetRoutes updates the destinations weight for primary and canary
+func (kr *KumaRouter) SetRoutes(
+	canary *flaggerv1.Canary,
+	primaryWeight int,
+	canaryWeight int,
+	_ bool,
+) error {
+	apexName, primaryName, canaryName := canary.GetServiceNames()
+	tr, err := kr.kumaClient.KumaV1alpha1().TrafficRoutes().Get(context.TODO(), apexName, metav1.GetOptions{})
+
+	if err != nil {
+		return fmt.Errorf("TrafficRoute %s get query error %v", apexName, err)
+	}
+
+	conf := &kumav1alpha1.TrafficRouteConf{
+		Split: []*kumav1alpha1.TrafficRouteSplit{
+			{
+				Weight: uint32(primaryWeight),
+				Destination: map[string]string{
+					"kuma.io/service": fmt.Sprintf("%s_%s_svc_%d", primaryName, canary.Namespace, canary.Spec.Service.Port),
+				},
+			},
+			{
+				Weight: uint32(canaryWeight),
+				Destination: map[string]string{
+					"kuma.io/service": fmt.Sprintf("%s_%s_svc_%d", canaryName, canary.Namespace, canary.Spec.Service.Port),
+				},
+			},
+		},
+	}
+
+	trClone := tr.DeepCopy()
+	trClone.Spec.Conf = conf
+
+	_, err = kr.kumaClient.KumaV1alpha1().TrafficRoutes().Update(context.TODO(), trClone, metav1.UpdateOptions{})
+	if err != nil {
+		return fmt.Errorf("TrafficRoute %s update error %v", apexName, err)
+	}
+
+	return nil
+}
+
+func (kr *KumaRouter) Finalize(_ *flaggerv1.Canary) error {
+	return nil
+}

pkg/router/kuma_test.go

@@ -0,0 +1,78 @@
+/*
+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 router
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestKumaRouter_Reconcile(t *testing.T) {
+	canary := newTestSMICanary()
+	mocks := newFixture(canary)
+	router := &KumaRouter{
+		logger:        mocks.logger,
+		flaggerClient: mocks.flaggerClient,
+		kumaClient:    mocks.meshClient,
+		kubeClient:    mocks.kubeClient,
+	}
+
+	// init
+	err := router.Reconcile(canary)
+	require.NoError(t, err)
+
+	// test insert
+	trafficRoute, err := router.kumaClient.KumaV1alpha1().TrafficRoutes().Get(context.TODO(), "podinfo", metav1.GetOptions{})
+	require.NoError(t, err)
+
+	splits := trafficRoute.Spec.Conf.Split
+	require.Len(t, splits, 2)
+	assert.Equal(t, uint32(100), splits[0].Weight)
+	assert.Equal(t, uint32(0), splits[1].Weight)
+
+}
+
+func TestKumaRouter_Routes(t *testing.T) {
+	canary := newTestSMICanary()
+	mocks := newFixture(canary)
+	router := &KumaRouter{
+		logger:        mocks.logger,
+		flaggerClient: mocks.flaggerClient,
+		kumaClient:    mocks.meshClient,
+		kubeClient:    mocks.kubeClient,
+	}
+
+	// init
+	err := router.Reconcile(canary)
+	require.NoError(t, err)
+
+	// test set routers
+	err = router.SetRoutes(canary, 50, 50, false)
+	require.NoError(t, err)
+
+	trafficRoute, err := router.kumaClient.KumaV1alpha1().TrafficRoutes().Get(context.TODO(), "podinfo", metav1.GetOptions{})
+	require.NoError(t, err)
+
+	primary := trafficRoute.Spec.Conf.Split[0]
+	assert.Equal(t, uint32(50), primary.Weight)
+
+}