From 47be2a25f29a84229355e8cc5c75546f7720c261 Mon Sep 17 00:00:00 2001 From: John Harris Date: Sat, 18 Dec 2021 14:07:59 -0800 Subject: [PATCH] Add Kuma routing and metrics Signed-off-by: John Harris --- pkg/metrics/observers/factory.go | 4 + pkg/metrics/observers/kuma.go | 94 +++++++++++++ pkg/metrics/observers/kuma_test.go | 100 ++++++++++++++ pkg/router/factory.go | 7 + pkg/router/kuma.go | 212 +++++++++++++++++++++++++++++ pkg/router/kuma_test.go | 78 +++++++++++ 6 files changed, 495 insertions(+) create mode 100644 pkg/metrics/observers/kuma.go create mode 100644 pkg/metrics/observers/kuma_test.go create mode 100644 pkg/router/kuma.go create mode 100644 pkg/router/kuma_test.go diff --git a/pkg/metrics/observers/factory.go b/pkg/metrics/observers/factory.go index 9b54dbe5..23e4bd2f 100644 --- a/pkg/metrics/observers/factory.go +++ b/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, diff --git a/pkg/metrics/observers/kuma.go b/pkg/metrics/observers/kuma.go new file mode 100644 index 00000000..699fea04 --- /dev/null +++ b/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 +} diff --git a/pkg/metrics/observers/kuma_test.go b/pkg/metrics/observers/kuma_test.go new file mode 100644 index 00000000..dda802a1 --- /dev/null +++ b/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) +} diff --git a/pkg/router/factory.go b/pkg/router/factory.go index 5b7adc25..e5a864e7 100644 --- a/pkg/router/factory.go +++ b/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: diff --git a/pkg/router/kuma.go b/pkg/router/kuma.go new file mode 100644 index 00000000..75b1b0fa --- /dev/null +++ b/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 +} diff --git a/pkg/router/kuma_test.go b/pkg/router/kuma_test.go new file mode 100644 index 00000000..52f10b89 --- /dev/null +++ b/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) + +}