feat: Provide scaler for Amazon managed service for Prometheus

CHANGELOG.md

@@ -120,6 +120,7 @@ Here is an overview of all new **experimental** features:
 - **Kafka Scaler**: Add more logging to check Sarama DescribeTopics method ([#5102](https://github.com/kedacore/keda/issues/5102))
 - **Kafka Scaler**: Add support for Kerberos authentication (SASL / GSSAPI) ([#4836](https://github.com/kedacore/keda/issues/4836))
 - **Prometheus Metrics**: Introduce paused ScaledObjects in Prometheus metrics ([#4430](https://github.com/kedacore/keda/issues/4430))
+- **Prometheus Scaler**: Provide scaler for Amazon managed service for Prometheus ([#2214](https://github.com/kedacore/keda/issues/2214))
 - **Pulsar Scaler**: support endpointParams in pulsar oauth ([#5069](https://github.com/kedacore/keda/issues/5069))
 
 ### Fixes

go.mod

@@ -27,6 +27,7 @@ require (
 	github.com/aws/aws-sdk-go-v2 v1.24.1
 	github.com/aws/aws-sdk-go-v2/config v1.26.2
 	github.com/aws/aws-sdk-go-v2/credentials v1.16.13
+	github.com/aws/aws-sdk-go-v2/service/amp v1.22.1
 	github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.32.1
 	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.26.7
 	github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.18.6

go.sum

@@ -936,6 +936,8 @@ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.10/go.mod h1:6UV4SZkVvmO
 github.com/aws/aws-sdk-go-v2/internal/ini v1.3.20/go.mod h1:bfTcsThj5a9P5pIGRy0QudJ8k4+issxXX+O6Djnd5Cs=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.7.2 h1:GrSw8s0Gs/5zZ0SX+gX4zQjRnRsMJDJ2sLur1gRBhEM=
 github.com/aws/aws-sdk-go-v2/internal/ini v1.7.2/go.mod h1:6fQQgfuGmw8Al/3M2IgIllycxV7ZW7WCdVSqfBeUiCY=
+github.com/aws/aws-sdk-go-v2/service/amp v1.22.1 h1:09O7NJKub+PsLAi1S+j/melSkjQROVV2RsDGqt3i34k=
+github.com/aws/aws-sdk-go-v2/service/amp v1.22.1/go.mod h1:zXysWREb7sWv3Mr80IBeQmbbWtBD4OvA5r/W+E+aSyA=
 github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.32.1 h1:IQ+uLXwS5Eelikc5ZdR0P55XPo+tqWh+k872KdpAjFA=
 github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.32.1/go.mod h1:G63GKqSBLpBmO3tN1/PwM2NC65XvSd00zJWTZk202bc=
 github.com/aws/aws-sdk-go-v2/service/dynamodb v1.26.7 h1:X60rMbnylU1xmmhv4+/N78t+lKOCC4ELst5eR25dyqg=

pkg/scalers/aws/aws_sigv4.go

@@ -0,0 +1,116 @@
+/*
+Copyright 2024 The KEDA 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.
+*/
+
+/*
+This file contains all the logic for caching aws.Config across all the (AWS)
+triggers. The first time when an aws.Config is requested, it's cached based on
+the authentication info (roleArn, Key&Secret, keda itself) and it's returned
+every time when an aws.Config is requested for the same authentication info.
+This is required because if we don't cache and share them, each scaler
+generates and refresh it's own token although all the tokens grants the same
+permissions
+*/
+package aws
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/service/amp"
+
+	"github.com/kedacore/keda/v2/pkg/scalers/scalersconfig"
+	httputils "github.com/kedacore/keda/v2/pkg/util"
+)
+
+// roundTripper adds custom round tripper to sign requests
+type roundTripper struct {
+	client *amp.Client
+	region string
+}
+
+var (
+	// ErrAwsAMPNoAwsRegion is returned when "awsRegion" is missing from the config.
+	ErrAwsAMPNoAwsRegion = errors.New("no awsRegion given")
+)
+
+// RoundTrip adds the roundTrip logic so that the request is SigV4 signed
+func (rt *roundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	cred, err := rt.client.Options().Credentials.Retrieve(req.Context())
+	if err != nil {
+		return nil, err
+	}
+
+	// We need to sign the request because giving an empty string (not a hashed empty string)
+	// fails in the backend as not signed request hence the following value is used
+	// "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" is the sha256 of ""
+	const reqHash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+
+	err = rt.client.Options().HTTPSignerV4.SignHTTP(req.Context(), cred, req, reqHash, "aps", rt.region, time.Now())
+	if err != nil {
+		return nil, err
+	}
+	// Create default transport
+	transport := httputils.CreateHTTPTransport(false)
+
+	// Send signed request
+	return transport.RoundTrip(req)
+}
+
+// parseAwsAMPMetadata parses the data to get the AWS sepcific auth info and metadata
+func parseAwsAMPMetadata(config *scalersconfig.ScalerConfig) (*awsConfigMetadata, error) {
+	meta := awsConfigMetadata{}
+
+	auth, err := GetAwsAuthorization(config.TriggerUniqueKey, config.PodIdentity, config.TriggerMetadata, config.AuthParams, config.ResolvedEnv)
+	if err != nil {
+		return nil, err
+	}
+
+	meta.awsAuthorization = auth
+	return &meta, nil
+}
+
+// NewSigV4RoundTripper returns a new http.RoundTripper that will sign requests
+// using Amazon's Signature Verification V4 signing procedure. The request will
+// then be handed off to the next RoundTripper provided by next. If next is nil,
+// http.DefaultTransport will be used.
+//
+// Credentials for signing are retrieving used the default AWS credential chain.
+// If credentials could not be found, an error will be returned.
+func NewSigV4RoundTripper(config *scalersconfig.ScalerConfig) (http.RoundTripper, error) {
+	// parseAwsAMPMetadata can return an error if AWS info is missing
+	// but this can happen if we check for them on not AWS scalers
+	// which is probably the reason to create a SigV4RoundTripper.
+	// To prevent failures we check if the metadata is nil
+	// (missing AWS info) and we hide the error
+	metadata, _ := parseAwsAMPMetadata(config)
+	if metadata == nil {
+		return nil, nil
+	}
+	awsCfg, err := GetAwsConfig(context.Background(), metadata.awsRegion, metadata.awsAuthorization)
+	if err != nil {
+		return nil, err
+	}
+
+	client := amp.NewFromConfig(*awsCfg, func(o *amp.Options) {})
+	rt := &roundTripper{
+		client: client,
+		region: metadata.awsRegion,
+	}
+
+	return rt, nil
+}

pkg/scalers/aws/aws_sigv4_test.go

@@ -0,0 +1,51 @@
+/*
+Copyright 2024 The KEDA 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.
+*/
+
+/*
+This file contains all the logic for caching aws.Config across all the (AWS)
+triggers. The first time when an aws.Config is requested, it's cached based on
+the authentication info (roleArn, Key&Secret, keda itself) and it's returned
+every time when an aws.Config is requested for the same authentication info.
+This is required because if we don't cache and share them, each scaler
+generates and refresh it's own token although all the tokens grants the same
+permissions
+*/
+package aws
+
+import (
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/kedacore/keda/v2/pkg/util"
+)
+
+func TestSigV4RoundTripper(t *testing.T) {
+	transport := util.CreateHTTPTransport(false)
+
+	cli := &http.Client{Transport: transport}
+
+	req, err := http.NewRequest(http.MethodGet, "https://aps-workspaces.us-west-2.amazonaws.com/workspaces/ws-38377ca8-8db3-4b58-812d-b65a81837bb8/api/v1/query?query=vector(10)", strings.NewReader("Hello, world!"))
+	require.NoError(t, err)
+	r, err := cli.Do(req)
+	require.NotEmpty(t, r)
+	require.NoError(t, err)
+	defer r.Body.Close()
+
+	require.NotNil(t, req)
+}

pkg/scalers/prometheus_scaler.go

@@ -18,6 +18,7 @@ import (
 
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
 	"github.com/kedacore/keda/v2/pkg/scalers/authentication"
+	"github.com/kedacore/keda/v2/pkg/scalers/aws"
 	"github.com/kedacore/keda/v2/pkg/scalers/azure"
 	"github.com/kedacore/keda/v2/pkg/scalers/gcp"
 	"github.com/kedacore/keda/v2/pkg/scalers/scalersconfig"
@@ -130,6 +131,16 @@ func NewPrometheusScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
 		if err == nil && gcpTransport != nil {
 			httpClient.Transport = gcpTransport
 		}
+
+		awsTransport, err := aws.NewSigV4RoundTripper(config)
+		if err != nil {
+			logger.V(1).Error(err, "failed to get AWS client HTTP transport ")
+			return nil, err
+		}
+
+		if err == nil && awsTransport != nil {
+			httpClient.Transport = awsTransport
+		}
 	}
 
 	return &prometheusScaler{