RabbitMQ: provide separate paremeters for user and password (#6180)

Signed-off-by: wangrushen <[email protected]>
Signed-off-by: Zbynek Roubalik <[email protected]>
Co-authored-by: Zbynek Roubalik <[email protected]>

CHANGELOG.md

@@ -62,6 +62,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 - **CloudEventSource**: Provide ClusterCloudEventSource around the management of ScaledJobs resources ([#3523](https://github.com/kedacore/keda/issues/3523))
 - **CloudEventSource**: Provide ClusterCloudEventSource around the management of TriggerAuthentication/ClusterTriggerAuthentication resources ([#3524](https://github.com/kedacore/keda/issues/3524))
 - **Github Action**: Fix panic when env for runnerScopeFromEnv or ownerFromEnv is empty ([#6156](https://github.com/kedacore/keda/issues/6156))
+- **RabbitMQ Scaler**: provide separate paremeters for user and password ([#2513](https://github.com/kedacore/keda/issues/2513))
 
 #### Experimental
 

pkg/scalers/rabbitmq_scaler.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/url"
 	"path"
+	"reflect"
 	"regexp"
 	"strconv"
 	"strings"
@@ -83,6 +84,9 @@ type rabbitMQMetadata struct {
 	timeout               time.Duration // custom http timeout for a specific trigger
 	triggerIndex          int           // scaler index
 
+	username string
+	password string
+
 	// TLS
 	ca          string
 	cert        string
@@ -149,12 +153,20 @@ func NewRabbitMQScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
 	if meta.protocol == amqpProtocol {
 		// Override vhost if requested.
 		host := meta.host
-		if meta.vhostName != "" {
+		if meta.vhostName != "" || (meta.username != "" && meta.password != "") {
 			hostURI, err := amqp.ParseURI(host)
 			if err != nil {
 				return nil, fmt.Errorf("error parsing rabbitmq connection string: %w", err)
 			}
-			hostURI.Vhost = meta.vhostName
+			if meta.vhostName != "" {
+				hostURI.Vhost = meta.vhostName
+			}
+
+			if meta.username != "" && meta.password != "" {
+				hostURI.Username = meta.username
+				hostURI.Password = meta.password
+			}
+
 			host = hostURI.String()
 		}
 
@@ -232,6 +244,28 @@ func resolveTLSAuthParams(config *scalersconfig.ScalerConfig, meta *rabbitMQMeta
 	return nil
 }
 
+func resolveAuth(config *scalersconfig.ScalerConfig, meta *rabbitMQMetadata) error {
+	usernameVal, err := getParameterFromConfigV2(config, "username", reflect.TypeOf(meta.username),
+		UseAuthentication(true), UseResolvedEnv(true), IsOptional(true))
+	if err != nil {
+		return err
+	}
+	meta.username = usernameVal.(string)
+
+	passwordVal, err := getParameterFromConfigV2(config, "password", reflect.TypeOf(meta.username),
+		UseAuthentication(true), UseResolvedEnv(true), IsOptional(true))
+	if err != nil {
+		return err
+	}
+	meta.password = passwordVal.(string)
+
+	if (meta.username != "" || meta.password != "") && (meta.username == "" || meta.password == "") {
+		return fmt.Errorf("username and password must be given together")
+	}
+
+	return nil
+}
+
 func parseRabbitMQMetadata(config *scalersconfig.ScalerConfig) (*rabbitMQMetadata, error) {
 	meta := rabbitMQMetadata{
 		connectionName: connectionName(config),
@@ -252,6 +286,11 @@ func parseRabbitMQMetadata(config *scalersconfig.ScalerConfig) (*rabbitMQMetadat
 		return nil, err
 	}
 
+	// Resolve username and password
+	if err := resolveAuth(config, &meta); err != nil {
+		return nil, err
+	}
+
 	meta.keyPassword = config.AuthParams["keyPassword"]
 
 	if config.PodIdentity.Provider == v1alpha1.PodIdentityProviderAzureWorkload {
@@ -596,6 +635,10 @@ func (s *rabbitMQScaler) getQueueInfoViaHTTP(ctx context.Context) (*queueInfo, e
 	vhost, subpaths := getVhostAndPathFromURL(parsedURL.Path, s.metadata.vhostName)
 	parsedURL.Path = subpaths
 
+	if s.metadata.username != "" && s.metadata.password != "" {
+		parsedURL.User = url.UserPassword(s.metadata.username, s.metadata.password)
+	}
+
 	var getQueueInfoManagementURI string
 	if s.metadata.useRegex {
 		getQueueInfoManagementURI = fmt.Sprintf("%s/api/queues%s?page=1&use_regex=true&pagination=false&name=%s&page_size=%d", parsedURL.String(), vhost, url.QueryEscape(s.metadata.queueName), s.metadata.pageSize)

pkg/scalers/rabbitmq_scaler_test.go

@@ -18,7 +18,9 @@ import (
 )
 
 const (
-	host = "myHostSecret"
+	host             = "myHostSecret"
+	rabbitMQUsername = "myUsernameSecret"
+	rabbitMQPassword = "myPasswordSecret"
 )
 
 type parseRabbitMQMetadataTestData struct {
@@ -43,7 +45,9 @@ type rabbitMQMetricIdentifier struct {
 }
 
 var sampleRabbitMqResolvedEnv = map[string]string{
-	host: "amqp://user:[email protected]:5236/vhost",
+	host:             "amqp://user:[email protected]:5236/vhost",
+	rabbitMQUsername: "user",
+	rabbitMQPassword: "Password",
 }
 
 var testRabbitMQMetadata = []parseRabbitMQMetadataTestData{
@@ -151,6 +155,18 @@ var testRabbitMQAuthParamData = []parseRabbitMQAuthParamTestData{
 	{map[string]string{"queueName": "sample", "hostFromEnv": host}, v1alpha1.AuthPodIdentity{}, map[string]string{"tls": "enable", "ca": "caaa", "cert": "ceert"}, true, true, false},
 	// failure, TLS invalid
 	{map[string]string{"queueName": "sample", "hostFromEnv": host}, v1alpha1.AuthPodIdentity{}, map[string]string{"tls": "yes", "ca": "caaa", "cert": "ceert", "key": "kee"}, true, true, false},
+	// success, username and password
+	{map[string]string{"queueName": "sample", "hostFromEnv": host}, v1alpha1.AuthPodIdentity{}, map[string]string{"username": "user", "password": "PASSWORD"}, false, false, false},
+	// failure, username but no password
+	{map[string]string{"queueName": "sample", "hostFromEnv": host}, v1alpha1.AuthPodIdentity{}, map[string]string{"username": "user"}, true, false, false},
+	// failure, password but no username
+	{map[string]string{"queueName": "sample", "hostFromEnv": host}, v1alpha1.AuthPodIdentity{}, map[string]string{"password": "PASSWORD"}, true, false, false},
+	// success, username and password from env
+	{map[string]string{"queueName": "sample", "hostFromEnv": host, "usernameFromEnv": rabbitMQUsername, "passwordFromEnv": rabbitMQPassword}, v1alpha1.AuthPodIdentity{}, map[string]string{}, false, false, false},
+	// failure, username from env but not password
+	{map[string]string{"queueName": "sample", "hostFromEnv": host, "usernameFromEnv": rabbitMQUsername}, v1alpha1.AuthPodIdentity{}, map[string]string{}, true, false, false},
+	// failure, password from env but not username
+	{map[string]string{"queueName": "sample", "hostFromEnv": host, "passwordFromEnv": rabbitMQPassword}, v1alpha1.AuthPodIdentity{}, map[string]string{}, true, false, false},
 	// success, WorkloadIdentity
 	{map[string]string{"queueName": "sample", "hostFromEnv": host, "protocol": "http"}, v1alpha1.AuthPodIdentity{Provider: v1alpha1.PodIdentityProviderAzureWorkload, IdentityID: kedautil.StringPointer("client-id")}, map[string]string{"workloadIdentityResource": "rabbitmq-resource-id"}, false, false, true},
 	// failure, WoekloadIdentity not supported for amqp

tests/scalers/rabbitmq/rabbitmq_helper.go

@@ -139,6 +139,47 @@ data:
 ---
 apiVersion: apps/v1
 kind: Deployment
+metadata:
+  name: {{.DeploymentName}}
+  namespace: {{.TestNamespace}}
+  labels:
+    app: {{.DeploymentName}}
+spec:
+  replicas: 0
+  selector:
+    matchLabels:
+      app: {{.DeploymentName}}
+  template:
+    metadata:
+      labels:
+        app: {{.DeploymentName}}
+    spec:
+      containers:
+      - name: rabbitmq-consumer
+        image: ghcr.io/kedacore/tests-rabbitmq
+        imagePullPolicy: Always
+        command:
+          - receive
+        args:
+          - '{{.Connection}}'
+        envFrom:
+        - secretRef:
+            name: {{.SecretName}}
+`
+
+	RMQTargetDeploymentWithAuthEnvTemplate = `
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{.SecretName}}
+  namespace: {{.TestNamespace}}
+data:
+  RabbitApiHost: {{.Base64Connection}}
+  RabbitUsername: {{.Base64Username}}
+  RabbitPassword: {{.Base64Password}}
+---
+apiVersion: apps/v1
+kind: Deployment
 metadata:
   name: {{.DeploymentName}}
   namespace: {{.TestNamespace}}