feat(externalmetrics): implement ExternalMetricsProvider for querying external metrics

Co-authored-by: Johan Lore <[email protected]>
Co-authored-by: Maxime Véroone <[email protected]>

Author: jlore-decathlon

go.mod

@@ -24,11 +24,12 @@ require (
 	google.golang.org/grpc v1.76.0
 	google.golang.org/protobuf v1.36.10
 	gopkg.in/h2non/gock.v1 v1.1.2
-	k8s.io/api v0.34.1
-	k8s.io/apimachinery v0.34.1
-	k8s.io/client-go v0.34.1
-	k8s.io/code-generator v0.34.1
+	k8s.io/api v0.34.2
+	k8s.io/apimachinery v0.34.2
+	k8s.io/client-go v0.34.2
+	k8s.io/code-generator v0.34.2
 	k8s.io/klog/v2 v2.130.1
+	k8s.io/metrics v0.34.2
 	k8s.io/utils v0.0.0-20250604170112-4c0f3b243397
 	knative.dev/serving v0.46.6
 )

go.sum

@@ -275,20 +275,22 @@ gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/api v0.34.1 h1:jC+153630BMdlFukegoEL8E/yT7aLyQkIVuwhmwDgJM=
-k8s.io/api v0.34.1/go.mod h1:SB80FxFtXn5/gwzCoN6QCtPD7Vbu5w2n1S0J5gFfTYk=
-k8s.io/apimachinery v0.34.1 h1:dTlxFls/eikpJxmAC7MVE8oOeP1zryV7iRyIjB0gky4=
-k8s.io/apimachinery v0.34.1/go.mod h1:/GwIlEcWuTX9zKIg2mbw0LRFIsXwrfoVxn+ef0X13lw=
-k8s.io/client-go v0.34.1 h1:ZUPJKgXsnKwVwmKKdPfw4tB58+7/Ik3CrjOEhsiZ7mY=
-k8s.io/client-go v0.34.1/go.mod h1:kA8v0FP+tk6sZA0yKLRG67LWjqufAoSHA2xVGKw9Of8=
-k8s.io/code-generator v0.34.1 h1:WpphT26E+j7tEgIUfFr5WfbJrktCGzB3JoJH9149xYc=
-k8s.io/code-generator v0.34.1/go.mod h1:DeWjekbDnJWRwpw3s0Jat87c+e0TgkxoR4ar608yqvg=
+k8s.io/api v0.34.2 h1:fsSUNZhV+bnL6Aqrp6O7lMTy6o5x2C4XLjnh//8SLYY=
+k8s.io/api v0.34.2/go.mod h1:MMBPaWlED2a8w4RSeanD76f7opUoypY8TFYkSM+3XHw=
+k8s.io/apimachinery v0.34.2 h1:zQ12Uk3eMHPxrsbUJgNF8bTauTVR2WgqJsTmwTE/NW4=
+k8s.io/apimachinery v0.34.2/go.mod h1:/GwIlEcWuTX9zKIg2mbw0LRFIsXwrfoVxn+ef0X13lw=
+k8s.io/client-go v0.34.2 h1:Co6XiknN+uUZqiddlfAjT68184/37PS4QAzYvQvDR8M=
+k8s.io/client-go v0.34.2/go.mod h1:2VYDl1XXJsdcAxw7BenFslRQX28Dxz91U9MWKjX97fE=
+k8s.io/code-generator v0.34.2 h1:9bG6jTxmsU3HXE5BNYJTC8AZ1D6hVVfkm8yYSkdkGY0=
+k8s.io/code-generator v0.34.2/go.mod h1:dnDDEd6S/z4uZ+PG1aE58ySCi/lR4+qT3a4DddE4/2I=
 k8s.io/gengo/v2 v2.0.0-20250604051438-85fd79dbfd9f h1:SLb+kxmzfA87x4E4brQzB33VBbT2+x7Zq9ROIHmGn9Q=
 k8s.io/gengo/v2 v2.0.0-20250604051438-85fd79dbfd9f/go.mod h1:EJykeLsmFC60UQbYJezXkEsG2FLrt0GPNkU5iK5GWxU=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
 k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b h1:MloQ9/bdJyIu9lb1PzujOPolHyvO06MXG5TUIj2mNAA=
 k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b/go.mod h1:UZ2yyWbFTpuhSbFhv24aGNOdoRdJZgsIObGBUaYVsts=
+k8s.io/metrics v0.34.2 h1:zao91FNDVPRGIiHLO2vqqe21zZVPien1goyzn0hsz90=
+k8s.io/metrics v0.34.2/go.mod h1:Ydulln+8uZZctUM8yrUQX4rfq/Ay6UzsuXf24QJ37Vc=
 k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 h1:hwvWFiBzdWw1FhfY1FooPn3kzWuJ8tmbZBHi4zVsl1Y=
 k8s.io/utils v0.0.0-20250604170112-4c0f3b243397/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 knative.dev/networking v0.0.0-20250902160145-7dad473f6351 h1:Gv/UqbN0AK+ORoT5e2Kg+3+uMW/y9CCdhpXKxYaVV6k=

pkg/metrics/providers/externalmetrics.go

@@ -0,0 +1,139 @@
+/*
+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 (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"time"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	"k8s.io/metrics/pkg/apis/external_metrics"
+)
+
+const (
+	metricServiceEndpointPath = "/apis/external.metrics.k8s.io/v1beta1"
+	namespacesPath            = "/namespaces/"
+
+	autorisationHeaderKey  = "Authorization"
+	applicationBearerToken = "bearerToken"
+)
+
+// ExternalMetricsProvider executes datadog queries
+type ExternalMetricsProvider struct {
+	metricServiceEndpoint string
+	bearerToken           string // Find out if we can get authoritative answer that this is the standard
+
+	timeout   time.Duration
+	unsafeSsl bool
+}
+
+// NewExternalMetricsProvider takes a canary spec, a provider spec, and
+// returns a client ready to execute queries against the Service
+func NewExternalMetricsProvider(metricInterval string,
+	provider flaggerv1.MetricTemplateProvider,
+	credentials map[string][]byte) (*ExternalMetricsProvider, error) {
+
+	if provider.Address == "" {
+		return nil, fmt.Errorf("the Url of the external metric service must be provided")
+	}
+
+	externalMetrics := ExternalMetricsProvider{
+		metricServiceEndpoint: fmt.Sprintf("%s%s", provider.Address, metricServiceEndpointPath),
+		timeout:               5 * time.Second,
+		unsafeSsl:             provider.InsecureSkipVerify,
+	}
+
+	if b, ok := credentials[applicationBearerToken]; ok {
+		externalMetrics.bearerToken = string(b)
+	}
+
+	return &externalMetrics, nil
+}
+
+// RunQuery retrieves the ExternalMetricValue from the ExternalMetricsProvider.metricServiceUrl
+// and returns the first result as a float64
+func (p *ExternalMetricsProvider) RunQuery(query string) (float64, error) {
+
+	metricsQueryUrl := fmt.Sprintf("%s%s%s", p.metricServiceEndpoint, namespacesPath, query)
+	//TODO add labelSelector as queryString (in the docs of this provider as it's embedded in the query string)
+
+	req, err := http.NewRequest("GET", metricsQueryUrl, nil)
+	if err != nil {
+		return 0, fmt.Errorf("error http.NewRequest: %w", err)
+	}
+	if p.bearerToken != "" {
+		req.Header.Add(autorisationHeaderKey, fmt.Sprintf("Bearer %s", p.bearerToken))
+	}
+
+	ctx, cancel := context.WithTimeout(req.Context(), p.timeout)
+	defer cancel()
+	r, err := http.DefaultClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return 0, fmt.Errorf("request failed: %w", err)
+	}
+
+	defer r.Body.Close()
+	b, err := io.ReadAll(r.Body)
+	if err != nil {
+		return 0, fmt.Errorf("error reading body: %w", err)
+	}
+
+	if r.StatusCode != http.StatusOK {
+		return 0, fmt.Errorf("error response: %s: %w", string(b), err)
+	}
+
+	var res external_metrics.ExternalMetricValueList
+	if err := json.Unmarshal(b, &res); err != nil {
+		return 0, fmt.Errorf("error unmarshaling result: %w, '%s'", err, string(b))
+	}
+
+	if len(res.Items) < 1 {
+		return 0, fmt.Errorf("invalid response: %s: %w", string(b), ErrNoValuesFound)
+	}
+
+	// TODO
+	vs := res.Items[0].Value.AsApproximateFloat64()
+
+	return vs, nil
+}
+
+// IsOnline will only check the TCP endpoint reachability,
+// given that external metric servers don't have a common health check endpoint defined
+func (p *ExternalMetricsProvider) IsOnline() (bool, error) {
+	var d net.Dialer
+
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+
+	metricServiceUrl, err := url.Parse(p.metricServiceEndpoint)
+	if err != nil {
+		return false, fmt.Errorf("error parsing metric service url: %w", err)
+	}
+
+	conn, err := d.DialContext(ctx, "tcp", metricServiceUrl.Host)
+	defer conn.Close()
+	if err != nil {
+		return false, fmt.Errorf("connection failed: %w", err)
+	}
+	return true, err
+}

pkg/metrics/providers/externalmetrics_test.go

@@ -0,0 +1,131 @@
+/*
+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 (
+	json2 "encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/api/resource"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	flaggerv1 "github.com/fluxcd/flagger/pkg/apis/flagger/v1beta1"
+	"k8s.io/metrics/pkg/apis/external_metrics"
+)
+
+func TestNewExternalMetricsProvider(t *testing.T) {
+	bt := "token"
+	cs := map[string][]byte{
+		"bearerToken": []byte(bt),
+	}
+
+	provider := flaggerv1.MetricTemplateProvider{
+		Address:            "https://external-metrics.default.svc.cluster.local",
+		InsecureSkipVerify: false,
+	}
+
+	emp, err := NewExternalMetricsProvider("100s", provider, cs)
+	require.NoError(t, err)
+	assert.Equal(t, "https://external-metrics.default.svc.cluster.local/apis/external.metrics.k8s.io/v1beta1", emp.metricServiceEndpoint)
+	assert.Equal(t, provider.InsecureSkipVerify, emp.unsafeSsl)
+	assert.Equal(t, 5*time.Second, emp.timeout)
+	assert.Equal(t, bt, emp.bearerToken)
+}
+
+func TestExternalMetrics_RunQuery(t *testing.T) {
+	token := "mytoken"
+	t.Run("ok", func(t *testing.T) {
+		expected := 1.11111
+		eq := `default/myMetric?label1=value1`
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			aq := fmt.Sprintf("%s?%s", r.URL.EscapedPath(), r.URL.RawQuery)
+			assert.Equal(t, fmt.Sprintf("%s%s%s", metricServiceEndpointPath, namespacesPath, eq), aq)
+			assert.Equal(t, fmt.Sprintf("Bearer %s", token), r.Header.Get(autorisationHeaderKey))
+
+			q, err := resource.ParseQuantity(strconv.FormatFloat(expected, 'f', -1, 64))
+			assert.NoError(t, err)
+
+			ret := &external_metrics.ExternalMetricValueList{
+				TypeMeta: v1.TypeMeta{
+					APIVersion: "external.metrics.k8s.io/v1beta1",
+					Kind:       "ExternalMetricValueList",
+				},
+				ListMeta: v1.ListMeta{},
+				Items: []external_metrics.ExternalMetricValue{
+					{
+						MetricName: "myMetric",
+						MetricLabels: map[string]string{
+							"label1": "value1",
+						},
+						Value: q,
+					},
+				},
+			}
+			json, err := json2.Marshal(ret)
+			assert.NoError(t, err)
+			w.Write(json)
+		}))
+		defer ts.Close()
+
+		dp, err := NewExternalMetricsProvider("1m",
+			flaggerv1.MetricTemplateProvider{Address: ts.URL},
+			map[string][]byte{
+				applicationBearerToken: []byte(token),
+			},
+		)
+		require.NoError(t, err)
+
+		f, err := dp.RunQuery(eq)
+		require.NoError(t, err)
+		assert.Equal(t, expected, f)
+	})
+
+	t.Run("no values", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			ret := &external_metrics.ExternalMetricValueList{
+				TypeMeta: v1.TypeMeta{
+					APIVersion: "external.metrics.k8s.io/v1beta1",
+					Kind:       "ExternalMetricValueList",
+				},
+				ListMeta: v1.ListMeta{},
+				Items:    []external_metrics.ExternalMetricValue{},
+			}
+			json, err := json2.Marshal(ret)
+			assert.NoError(t, err)
+			w.Write(json)
+		}))
+		defer ts.Close()
+
+		dp, err := NewExternalMetricsProvider("1m",
+			flaggerv1.MetricTemplateProvider{Address: ts.URL},
+			map[string][]byte{
+				applicationBearerToken: []byte(token),
+			},
+		)
+		require.NoError(t, err)
+		_, err = dp.RunQuery("")
+		require.True(t, errors.Is(err, ErrNoValuesFound))
+	})
+}

pkg/metrics/providers/factory.go

@@ -29,6 +29,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(metricInterval, provider, credentials)
 	case "cloudwatch":
 		return NewCloudWatchProvider(metricInterval, provider)
 	case "newrelic":