Merge pull request #1863 from jlore-decathlon/feat/externalmetrics

feat(provider): add External Metrics provider
2026-04-15 06:57:34 +00:00 · 2026-04-14 14:54:30 +05:30
parent d29c60d2e6 9b37d18e16
commit 95cebe58c1
10 changed files with 522 additions and 13 deletions
@@ -0,0 +1,156 @@
+/*
+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 providers
+
+import (
+	"fmt"
+	"net/url"
+	"strings"
+	"time"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/rest"
+	externalmetrics_client "k8s.io/metrics/pkg/client/external_metrics"
+)
+
+// ExternalMetricsProvider fetches metrics from an ExternalMetricsProvider.
+type ExternalMetricsProvider struct {
+	client externalmetrics_client.NamespacedMetricsGetter
+}
+
+// NewExternalMetricsProvider takes a provider spec, credentials, and a
+// rest config, and returns a client ready to execute queries against the
+// External Metrics API server.
+func NewExternalMetricsProvider(
+	provider flaggerv1.MetricTemplateProvider,
+	credentials map[string][]byte,
+	config *rest.Config,
+) (*ExternalMetricsProvider, error) {
+	if config == nil {
+		return nil, fmt.Errorf(
+			"could not initialize ExternalMetricsProvider: rest config is nil",
+		)
+	}
+
+	// clone to avoid mutating the shared config
+	restConfig := rest.CopyConfig(config)
+
+	// apply overrides from MetricTemplateProvider
+	if provider.Address != "" {
+		restConfig.Host = provider.Address
+	}
+	restConfig.TLSClientConfig.Insecure = provider.InsecureSkipVerify
+	if tokenBytes, ok := credentials["token"]; ok {
+		restConfig.BearerToken = string(tokenBytes)
+	}
+
+	restConfig.Timeout = 5 * time.Second
+
+	client, err := externalmetrics_client.NewForConfig(restConfig)
+	if err != nil {
+		return nil, fmt.Errorf("error creating external metric client: %w", err)
+	}
+
+	return &ExternalMetricsProvider{
+		client: client,
+	}, nil
+}
+
+// RunQuery retrieves the ExternalMetricValue from the External Metrics API
+// at the ExternalMetricsProvider's address, using the provided query string,
+// and returns the *first* result as a float64.
+func (p *ExternalMetricsProvider) RunQuery(query string) (float64, error) {
+	namespace, metricName, selector, err := parseExternalMetricsQuery(query)
+	if err != nil {
+		return 0, fmt.Errorf("error parsing metric query: %w", err)
+	}
+
+	nm := p.client.NamespacedMetrics(namespace)
+	metricsList, err := nm.List(metricName, selector)
+	if err != nil {
+		return 0, fmt.Errorf("error querying external metrics API: %w", err)
+	}
+
+	if len(metricsList.Items) < 1 {
+		return 0, fmt.Errorf("no external metrics found: %w", ErrNoValuesFound)
+	}
+
+	vs := metricsList.Items[0].Value.AsApproximateFloat64()
+
+	return vs, nil
+}
+
+// IsOnline tests that the External Metrics API is reachable by looking for dummy metrics.
+// If we don't get a network error, we assume the service is online.
+func (p *ExternalMetricsProvider) IsOnline() (bool, error) {
+	nm := p.client.NamespacedMetrics("kube-system")
+	_, err := nm.List("dummy-metric", labels.Everything())
+
+	if err != nil {
+		return false, fmt.Errorf("external metrics service unavailable: %w", err)
+	}
+	return true, nil
+}
+
+// parseExternalMetricsQuery parses a query string in the format:
+//
+//	<namespace>/<metricName>?labelSelector=<urlencoded label selectors>
+//
+// where only the metricName is required.
+// and returns the namespace, metricName, and labelSelector separately.
+func parseExternalMetricsQuery(query string) (namespace string, metricName string, labelSelector labels.Selector, err error) {
+	// Adding a dummy protocol so we can leverage url.Parse for parsing the query string, easily extracting the path and query parameters.
+	u, err := url.Parse("dummy:///" + query)
+	if err != nil {
+		return "", "", labels.Everything(), fmt.Errorf("malformed query string, expected <namespace>/<metricName>?labelSelector=<urlencoded label selectors>, got %s", query)
+	}
+	path := strings.TrimPrefix(u.Path, "/")
+	parts := strings.Split(path, "/")
+	if len(parts) > 2 {
+		return "", "", labels.Everything(), fmt.Errorf("malformed query string, too many slashes, expected <namespace>/<metricName>?labelSelector=<urlencoded label selectors>, got %s", query)
+	}
+
+	namespace = "default"
+	switch len(parts) {
+	case 1:
+		// Format: "metric"
+		metricName = parts[0]
+	case 2:
+		// Format: "namespace/metric" or "/metric"
+		if parts[0] != "" {
+			namespace = parts[0]
+		}
+		metricName = parts[1]
+	}
+	if metricName == "" {
+		return "", "", labels.Everything(), fmt.Errorf("metric name cannot be empty")
+	}
+
+	qp := u.Query()
+	rawSelector := qp.Get("labelSelector")
+	if rawSelector == "" {
+		labelSelector = labels.Everything()
+	} else {
+		labelSelector, err = labels.Parse(rawSelector)
+		if err != nil {
+			return "", "", labels.Everything(), fmt.Errorf("error parsing label selector from string %s: %w", rawSelector, err)
+		}
+	}
+
+	return namespace, metricName, labelSelector, nil
+}
@@ -0,0 +1,287 @@
+/*
+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 providers
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/inf.v0"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/rest"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8stesting "k8s.io/client-go/testing"
+	emv1beta1 "k8s.io/metrics/pkg/apis/external_metrics/v1beta1"
+	fakeemc "k8s.io/metrics/pkg/client/external_metrics/fake"
+)
+
+const (
+	testMetricName          = "myMetric"
+	testMetricNamespace     = "default"
+	testMetricServerAddress = "https://external-metrics.default.svc.cluster.local"
+	testQuery               = "default/myMetric?labelSelector=label1%3Dvalue1"
+)
+
+var (
+	testMetricLabels       = [...]string{"label1"}
+	testMetricLabelsValues = [...]string{"value1"}
+	// 11111e-4 = 1.1111
+	testMetricValue = resource.NewDecimalQuantity(*inf.NewDec(11111, 4), resource.DecimalSI)
+)
+
+func TestExternalMetrics_NewProvider(t *testing.T) {
+	tests := []struct {
+		name               string
+		Address            string
+		InsecureSkipVerify bool
+		creds              map[string][]byte
+		config             *rest.Config
+	}{
+		{
+			name:               "Custom provider address and token",
+			Address:            testMetricServerAddress,
+			InsecureSkipVerify: false,
+			creds: map[string][]byte{
+				"token": []byte("test-token"),
+			},
+			config: &rest.Config{},
+		},
+		{
+			name:               "In cluster, automatic address and token",
+			Address:            "",
+			InsecureSkipVerify: true,
+			creds:              map[string][]byte{},
+			config: &rest.Config{
+				Host:            "https://kubernetes.default.svc",
+				BearerToken:     "fake-token",
+				TLSClientConfig: rest.TLSClientConfig{Insecure: true},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mtp := flaggerv1.MetricTemplateProvider{
+				Address:            tt.Address,
+				InsecureSkipVerify: tt.InsecureSkipVerify,
+			}
+			emp, err := NewExternalMetricsProvider(mtp, tt.creds, tt.config)
+			require.NoError(t, err)
+			require.NotNil(t, emp)
+		})
+	}
+}
+
+func TestExternalMetrics_NewProvider_NilConfig(t *testing.T) {
+	mtp := flaggerv1.MetricTemplateProvider{}
+	_, err := NewExternalMetricsProvider(mtp, nil, nil)
+	require.Error(t, err)
+}
+
+func TestExternalMetrics_ParseQuery(t *testing.T) {
+	tests := []struct {
+		name              string
+		query             string
+		wantNamespace     string
+		wantMetricName    string
+		wantLabelSelector string
+		wantErr           bool
+	}{
+		{
+			name:              "General case",
+			query:             testQuery,
+			wantNamespace:     testMetricNamespace,
+			wantMetricName:    testMetricName,
+			wantLabelSelector: labels.Set{testMetricLabels[0]: testMetricLabelsValues[0]}.AsSelector().String(),
+			wantErr:           false,
+		},
+		{
+			name:              "Still OK without labelSelector",
+			query:             testQuery[:strings.Index(testQuery, "?")],
+			wantNamespace:     testMetricNamespace,
+			wantMetricName:    testMetricName,
+			wantLabelSelector: labels.Everything().String(),
+			wantErr:           false,
+		},
+		{
+			name:              "No namespace uses default",
+			query:             "/metric_only",
+			wantNamespace:     "default",
+			wantMetricName:    "metric_only",
+			wantLabelSelector: labels.Everything().String(),
+			wantErr:           false,
+		},
+		{
+			name:    "Missing metric name - namespaceonly",
+			query:   "namespaceonly/",
+			wantErr: true,
+		},
+		{
+			name:    "Missing metric name - slash only",
+			query:   "/",
+			wantErr: true,
+		},
+		{
+			name:    "Missing metric name - empty",
+			query:   "",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotNamespace, gotMetricName, gotLabelSelector, err := parseExternalMetricsQuery(tt.query)
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tt.wantNamespace, gotNamespace)
+				assert.Equal(t, tt.wantMetricName, gotMetricName)
+				assert.Equal(t, tt.wantLabelSelector, gotLabelSelector.String())
+			}
+		})
+	}
+}
+
+func TestExternalMetrics_RunQuery(t *testing.T) {
+	tests := []struct {
+		name    string
+		query   string
+		setup   func(*fakeemc.FakeExternalMetricsClient)
+		want    float64
+		wantErr bool
+	}{
+		{
+			name:  "Full query with label selector",
+			query: testQuery,
+			setup: func(client *fakeemc.FakeExternalMetricsClient) {
+				client.Fake.AddReactor("list", "*", func(action k8stesting.Action) (bool, runtime.Object, error) {
+					return true, &emv1beta1.ExternalMetricValueList{
+						Items: []emv1beta1.ExternalMetricValue{
+							{
+								MetricName: testMetricName,
+								Value:      *testMetricValue,
+								MetricLabels: map[string]string{
+									testMetricLabels[0]: testMetricLabelsValues[0],
+								},
+								Timestamp: metav1.Now(),
+							},
+						},
+					}, nil
+				})
+			},
+			want: testMetricValue.AsApproximateFloat64(),
+		},
+		{
+			name:  "Namespace and metric only",
+			query: "namespace/" + testMetricName,
+			setup: func(client *fakeemc.FakeExternalMetricsClient) {
+				client.Fake.AddReactor("list", "*", func(action k8stesting.Action) (bool, runtime.Object, error) {
+					return true, &emv1beta1.ExternalMetricValueList{
+						Items: []emv1beta1.ExternalMetricValue{{
+							MetricName: testMetricName,
+							Value:      *testMetricValue,
+							Timestamp:  metav1.Now(),
+						}},
+					}, nil
+				})
+			},
+			want: testMetricValue.AsApproximateFloat64(),
+		},
+		{
+			name:  "Metric only, default namespace",
+			query: testMetricName,
+			setup: func(client *fakeemc.FakeExternalMetricsClient) {
+				client.Fake.AddReactor("list", "*", func(action k8stesting.Action) (bool, runtime.Object, error) {
+					return true, &emv1beta1.ExternalMetricValueList{
+						Items: []emv1beta1.ExternalMetricValue{{
+							MetricName: testMetricName,
+							Value:      *testMetricValue,
+							Timestamp:  metav1.Now(),
+						}},
+					}, nil
+				})
+			},
+			want: testMetricValue.AsApproximateFloat64(),
+		},
+		{
+			name:    "Fails on invalid query",
+			query:   "namespace/metric/extra",
+			wantErr: true,
+		},
+		{
+			name:  "Fails when external metrics API returns an error",
+			query: testQuery,
+			setup: func(client *fakeemc.FakeExternalMetricsClient) {
+				client.Fake.AddReactor("list", "*", func(action k8stesting.Action) (bool, runtime.Object, error) {
+					return true, nil, errors.New("backend unavailable")
+				})
+			},
+			wantErr: true,
+		},
+		{
+			name:  "Fails when no external metrics are returned",
+			query: testQuery,
+			setup: func(client *fakeemc.FakeExternalMetricsClient) {
+				client.Fake.AddReactor("list", "*", func(action k8stesting.Action) (bool, runtime.Object, error) {
+					return true, &emv1beta1.ExternalMetricValueList{}, nil
+				})
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			emclient := &fakeemc.FakeExternalMetricsClient{}
+			if tt.setup != nil {
+				tt.setup(emclient)
+			}
+
+			emp := &ExternalMetricsProvider{
+				client: emclient,
+			}
+
+			got, err := emp.RunQuery(tt.query)
+			if tt.wantErr {
+				require.Error(t, err)
+				assert.Zero(t, got)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestExternalMetrics_IsOnline(t *testing.T) {
+	emp := &ExternalMetricsProvider{
+		client: &fakeemc.FakeExternalMetricsClient{},
+	}
+
+	online, err := emp.IsOnline()
+	require.NoError(t, err)
+	assert.True(t, online)
+}
@@ -38,6 +38,8 @@ func (factory Factory) Provider(metricInterval string, provider flaggerv1.Metric
 		return NewPrometheusProvider(provider, credentials)
 	case "datadog":
 		return NewDatadogProvider(metricInterval, provider, credentials)
+	case "externalmetrics":
+		return NewExternalMetricsProvider(provider, credentials, config)
 	case "cloudwatch":
 		return NewCloudWatchProvider(metricInterval, provider)
 	case "newrelic":