adding support for azure managed prometheus

CHANGELOG.md

@@ -55,6 +55,7 @@ Here is an overview of all **stable** additions:
 - **General**: Introduce new ArangoDB Scaler ([#4000](https://github.com/kedacore/keda/issues/4000))
 - **Prometheus Metrics**: Introduce scaler activity in Prometheus metrics ([#4114](https://github.com/kedacore/keda/issues/4114))
 - **Prometheus Metrics**: Introduce scaler latency in Prometheus metrics ([#4037](https://github.com/kedacore/keda/issues/4037))
+- **Prometheus Scaler**: Extend Prometheus Scaler to support Azure managed service for Prometheus ([#4153](https://github.com/kedacore/keda/issues/4153))
 
 Here is an overview of all new **experimental** features:
 

pkg/scalers/azure/azure_managed_prometheus_http_round_tripper.go

@@ -0,0 +1,82 @@
+package azure
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
+	az "github.com/Azure/go-autorest/autorest/azure"
+
+	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
+	"github.com/kedacore/keda/v2/pkg/util"
+)
+
+var azureManagedPrometheusResourceURLInCloud = map[string]string{
+	"AZUREPUBLICCLOUD":       "https://prometheus.monitor.azure.com/.default",
+	"AZUREUSGOVERNMENTCLOUD": "https://prometheus.monitor.azure.us/.default",
+	"AZURECHINACLOUD":        "https://prometheus.monitor.azure.cn/.default",
+}
+
+type azureManagedPrometheusHTTPRoundTripper struct {
+	chainedCredential *azidentity.ChainedTokenCredential
+	next              http.RoundTripper
+	resourceURL       string
+}
+
+// Tries to get a round tripper.
+// If the pod identity represents azure auth, it creates a round tripper and returns that. Returns error if fails to create one.
+// If its not azure auth, then this becomes a no-op. Neither returns round tripper nor error.
+func TryAndGetAzureManagedPrometheusHTTPRoundTripper(podIdentity kedav1alpha1.AuthPodIdentity, triggerMetadata map[string]string) (http.RoundTripper, error) {
+	switch podIdentity.Provider {
+	case kedav1alpha1.PodIdentityProviderAzureWorkload, kedav1alpha1.PodIdentityProviderAzure:
+
+		if triggerMetadata == nil {
+			return nil, fmt.Errorf("trigger metadata cannot be nil")
+		}
+
+		chainedCred, err := NewChainedCredential(podIdentity.IdentityID, podIdentity.Provider)
+		if err != nil {
+			return nil, err
+		}
+
+		azureManagedPrometheusResourceURLProvider := func(env az.Environment) (string, error) {
+			if resource, ok := azureManagedPrometheusResourceURLInCloud[strings.ToUpper(env.Name)]; ok {
+				return resource, nil
+			}
+
+			return "", fmt.Errorf("azure managed prometheus is not available in cloud %s", env.Name)
+		}
+
+		resourceURLBasedOnCloud, err := ParseEnvironmentProperty(triggerMetadata, "azureManagedPrometheusResourceURL", azureManagedPrometheusResourceURLProvider)
+		if err != nil {
+			return nil, err
+		}
+
+		transport := util.CreateHTTPTransport(false)
+		rt := &azureManagedPrometheusHTTPRoundTripper{
+			next:              transport,
+			chainedCredential: chainedCred,
+			resourceURL:       resourceURLBasedOnCloud,
+		}
+		return rt, nil
+	}
+
+	// Not azure managed prometheus. Don't create a round tripper and don't return error.
+	return nil, nil
+}
+
+// Sets Auhtorization header for requests
+func (rt *azureManagedPrometheusHTTPRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	token, err := rt.chainedCredential.GetToken(req.Context(), policy.TokenRequestOptions{Scopes: []string{rt.resourceURL}})
+
+	if err != nil {
+		return nil, err
+	}
+
+	bearerAccessToken := "Bearer " + token.Token
+	req.Header.Set("Authorization", bearerAccessToken)
+
+	return rt.next.RoundTrip(req)
+}

pkg/scalers/azure/azure_managed_prometheus_http_round_tripper_test.go

@@ -0,0 +1,32 @@
+package azure
+
+import (
+	"testing"
+
+	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
+)
+
+type testTryAndGetAzureManagedPrometheusHTTPRoundTripperData struct {
+	testName            string
+	podIdentityProvider kedav1alpha1.PodIdentityProvider
+	isError             bool
+}
+
+var testTryAndGetAzureManagedPrometheusHTTPRoundTripperTestData = []testTryAndGetAzureManagedPrometheusHTTPRoundTripperData{
+	{"test azure workload identity trigger metadata absent", kedav1alpha1.PodIdentityProviderAzureWorkload, true},
+	{"test azure pod identity trigger metadata absent", kedav1alpha1.PodIdentityProviderAzureWorkload, true},
+	{"test not azure identity", kedav1alpha1.PodIdentityProviderNone, false},
+}
+
+func TestTryAndGetAzureManagedPrometheusHTTPRoundTripper(t *testing.T) {
+	for _, testData := range testTryAndGetAzureManagedPrometheusHTTPRoundTripperTestData {
+		_, err := TryAndGetAzureManagedPrometheusHTTPRoundTripper(kedav1alpha1.AuthPodIdentity{Provider: testData.podIdentityProvider}, nil)
+		if testData.isError {
+			if err == nil {
+				t.Errorf("Test: %v; Expected error but got success. testData: %v", testData.testName, testData)
+			}
+		} else if err != nil {
+			t.Errorf("Test: %v; Expected success but got error: %v", testData.testName, err)
+		}
+	}
+}

pkg/scalers/prometheus_scaler.go

@@ -16,6 +16,7 @@ import (
 	"k8s.io/metrics/pkg/apis/external_metrics"
 
 	"github.com/kedacore/keda/v2/pkg/scalers/authentication"
+	"github.com/kedacore/keda/v2/pkg/scalers/azure"
 	kedautil "github.com/kedacore/keda/v2/pkg/util"
 )
 
@@ -92,17 +93,33 @@ func NewPrometheusScaler(config *ScalerConfig) (Scaler, error) {
 
 	httpClient := kedautil.CreateHTTPClient(config.GlobalHTTPTimeout, meta.unsafeSsl)
 
-	if meta.prometheusAuth != nil && (meta.prometheusAuth.CA != "" || meta.prometheusAuth.EnableTLS) {
-		// create http.RoundTripper with auth settings from ScalerConfig
-		transport, err := authentication.CreateHTTPRoundTripper(
-			authentication.NetHTTP,
-			meta.prometheusAuth,
-		)
+	if meta.prometheusAuth != nil {
+		if meta.prometheusAuth.CA != "" || meta.prometheusAuth.EnableTLS {
+			// create http.RoundTripper with auth settings from ScalerConfig
+			transport, err := authentication.CreateHTTPRoundTripper(
+				authentication.NetHTTP,
+				meta.prometheusAuth,
+			)
+			if err != nil {
+				logger.V(1).Error(err, "init Prometheus client http transport")
+				return nil, err
+			}
+			httpClient.Transport = transport
+		}
+	} else {
+		// could be the case of azure managed prometheus. Try and get the roundtripper.
+		// If its not the case of azure managed prometheus, we will get both transport and err as nil and proceed assuming no auth.
+		transport, err := azure.TryAndGetAzureManagedPrometheusHTTPRoundTripper(config.PodIdentity, config.TriggerMetadata)
+
 		if err != nil {
-			logger.V(1).Error(err, "init Prometheus client http transport")
+			logger.V(1).Error(err, "error while init Azure Managed Prometheus client http transport")
 			return nil, err
 		}
-		httpClient.Transport = transport
+
+		// transport should not be nil if its a case of azure managed prometheus
+		if transport != nil {
+			httpClient.Transport = transport
+		}
 	}
 
 	return &prometheusScaler{
@@ -195,14 +212,27 @@ func parsePrometheusMetadata(config *ScalerConfig) (meta *prometheusMetadata, er
 
 	meta.scalerIndex = config.ScalerIndex
 
+	err = parseAuthConfig(config, meta)
+	if err != nil {
+		return nil, err
+	}
+
+	return meta, nil
+}
+
+func parseAuthConfig(config *ScalerConfig, meta *prometheusMetadata) error {
 	// parse auth configs from ScalerConfig
 	auth, err := authentication.GetAuthConfigs(config.TriggerMetadata, config.AuthParams)
 	if err != nil {
-		return nil, err
+		return err
+	}
+
+	if auth != nil && config.PodIdentity.Provider != "" {
+		return fmt.Errorf("pod identity cannot be enabled with other auth types")
 	}
 	meta.prometheusAuth = auth
 
-	return meta, nil
+	return nil
 }
 
 func (s *prometheusScaler) Close(context.Context) error {

pkg/scalers/prometheus_scaler_test.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/assert"
+
+	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
 )
 
 type parsePrometheusMetadataTestData struct {
@@ -60,38 +62,47 @@ var prometheusMetricIdentifiers = []prometheusMetricIdentifier{
 }
 
 type prometheusAuthMetadataTestData struct {
-	metadata   map[string]string
-	authParams map[string]string
-	isError    bool
+	metadata            map[string]string
+	authParams          map[string]string
+	podIdentityProvider kedav1alpha1.PodIdentityProvider
+	isError             bool
 }
 
 var testPrometheusAuthMetadata = []prometheusAuthMetadataTestData{
 	// success TLS
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls"}, map[string]string{"ca": "caaa", "cert": "ceert", "key": "keey"}, false},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls"}, map[string]string{"ca": "caaa", "cert": "ceert", "key": "keey"}, "", false},
 	// TLS, ca is optional
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls"}, map[string]string{"cert": "ceert", "key": "keey"}, false},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls"}, map[string]string{"cert": "ceert", "key": "keey"}, "", false},
 	// fail TLS, key not given
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls"}, map[string]string{"ca": "caaa", "cert": "ceert"}, true},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls"}, map[string]string{"ca": "caaa", "cert": "ceert"}, "", true},
 	// fail TLS, cert not given
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls"}, map[string]string{"ca": "caaa", "key": "keey"}, true},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls"}, map[string]string{"ca": "caaa", "key": "keey"}, "", true},
 	// success bearer default
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "bearer"}, map[string]string{"bearerToken": "tooooken"}, false},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "bearer"}, map[string]string{"bearerToken": "tooooken"}, "", false},
 	// fail bearerAuth with no token
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "bearer"}, map[string]string{}, true},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "bearer"}, map[string]string{}, "", true},
 	// success basicAuth
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "basic"}, map[string]string{"username": "user", "password": "pass"}, false},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "basic"}, map[string]string{"username": "user", "password": "pass"}, "", false},
 	// fail basicAuth with no username
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "basic"}, map[string]string{}, true},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "basic"}, map[string]string{}, "", true},
 
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls, basic"}, map[string]string{"ca": "caaa", "cert": "ceert", "key": "keey", "username": "user", "password": "pass"}, false},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls, basic"}, map[string]string{"ca": "caaa", "cert": "ceert", "key": "keey", "username": "user", "password": "pass"}, "", false},
 
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls,basic"}, map[string]string{"username": "user", "password": "pass"}, true},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls,basic"}, map[string]string{"username": "user", "password": "pass"}, "", true},
 	// success custom auth
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "custom"}, map[string]string{"customAuthHeader": "header", "customAuthValue": "value"}, false},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "custom"}, map[string]string{"customAuthHeader": "header", "customAuthValue": "value"}, "", false},
 	// fail custom auth with no customAuthHeader
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "custom"}, map[string]string{"customAuthHeader": ""}, true},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "custom"}, map[string]string{"customAuthHeader": ""}, "", true},
 	// fail custom auth with no customAuthValue
-	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "custom"}, map[string]string{"customAuthValue": ""}, true},
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "custom"}, map[string]string{"customAuthValue": ""}, "", true},
+
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "tls,basic"}, map[string]string{"username": "user", "password": "pass"}, "", true},
+	// pod identity and other auth modes enabled together
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up", "authModes": "basic"}, map[string]string{"username": "user", "password": "pass"}, "azure-workload", true},
+	// azure workload identity
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up"}, nil, "azure-workload", false},
+	// azure pod identity
+	{map[string]string{"serverAddress": "http://localhost:9090", "metricName": "http_requests_total", "threshold": "100", "query": "up"}, nil, "azure", false},
 }
 
 func TestPrometheusParseMetadata(t *testing.T) {
@@ -127,7 +138,7 @@ func TestPrometheusGetMetricSpecForScaling(t *testing.T) {
 
 func TestPrometheusScalerAuthParams(t *testing.T) {
 	for _, testData := range testPrometheusAuthMetadata {
-		meta, err := parsePrometheusMetadata(&ScalerConfig{TriggerMetadata: testData.metadata, AuthParams: testData.authParams})
+		meta, err := parsePrometheusMetadata(&ScalerConfig{TriggerMetadata: testData.metadata, AuthParams: testData.authParams, PodIdentity: kedav1alpha1.AuthPodIdentity{Provider: testData.podIdentityProvider}})
 
 		if err != nil && !testData.isError {
 			t.Error("Expected success but got error", err)
@@ -137,11 +148,13 @@ func TestPrometheusScalerAuthParams(t *testing.T) {
 		}
 
 		if err == nil {
-			if (meta.prometheusAuth.EnableBearerAuth && !strings.Contains(testData.metadata["authModes"], "bearer")) ||
-				(meta.prometheusAuth.EnableBasicAuth && !strings.Contains(testData.metadata["authModes"], "basic")) ||
-				(meta.prometheusAuth.EnableTLS && !strings.Contains(testData.metadata["authModes"], "tls")) ||
-				(meta.prometheusAuth.EnableCustomAuth && !strings.Contains(testData.metadata["authModes"], "custom")) {
-				t.Error("wrong auth mode detected")
+			if meta.prometheusAuth != nil {
+				if (meta.prometheusAuth.EnableBearerAuth && !strings.Contains(testData.metadata["authModes"], "bearer")) ||
+					(meta.prometheusAuth.EnableBasicAuth && !strings.Contains(testData.metadata["authModes"], "basic")) ||
+					(meta.prometheusAuth.EnableTLS && !strings.Contains(testData.metadata["authModes"], "tls")) ||
+					(meta.prometheusAuth.EnableCustomAuth && !strings.Contains(testData.metadata["authModes"], "custom")) {
+					t.Error("wrong auth mode detected")
+				}
 			}
 		}
 	}

tests/.env

@@ -16,6 +16,7 @@ AZURE_DEVOPS_PROJECT=
 TF_AZURE_EVENTHBUS_MANAGEMENT_CONNECTION_STRING=
 TF_AZURE_KEYVAULT_URI=
 TF_AZURE_LOG_ANALYTICS_WORKSPACE_ID=
+TF_AZURE_MANAGED_PROMETHEUS_QUERY_ENDPOINT=
 TF_AZURE_RESOURCE_GROUP=
 TF_AZURE_SERVICE_BUS_CONNECTION_STRING=
 TF_AZURE_SP_APP_ID=