Add support for per-scaler override on the RabbitMQ vhost.

CHANGELOG.md

@@ -27,6 +27,7 @@
 - Mask password in postgres scaler auto generated metricName. ([PR #1381](https://github.com/kedacore/keda/pull/1381))
 - Bug fix for pending jobs in ScaledJob's accurateScalingStrategy . ([#1323](https://github.com/kedacore/keda/issues/1323))
 - Fix memory leak because of unclosed scalers. ([#1413](https://github.com/kedacore/keda/issues/1413))
+- Override the vhost on a RabbitMQ scaler via `vhostName` in the metadata. ([#1451](https://github.com/kedacore/keda/pull/1451))
 
 ### Breaking Changes
 

pkg/scalers/rabbitmq_scaler.go

@@ -42,8 +42,9 @@ type rabbitMQScaler struct {
 type rabbitMQMetadata struct {
 	queueName   string
 	queueLength int
-	host        string // connection string for either HTTP or AMQP protocol
-	protocol    string // either http or amqp protocol
+	host        string  // connection string for either HTTP or AMQP protocol
+	protocol    string  // either http or amqp protocol
+	vhostName   *string // override the vhost from the connection info
 }
 
 type queueInfo struct {
@@ -69,7 +70,18 @@ func NewRabbitMQScaler(config *ScalerConfig) (Scaler, error) {
 		}, nil
 	}
 
-	conn, ch, err := getConnectionAndChannel(meta.host)
+	// Override vhost if requested.
+	host := meta.host
+	if meta.vhostName != nil {
+		hostURI, err := amqp.ParseURI(host)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing rabbitmq connection string: %s", err)
+		}
+		hostURI.Vhost = *meta.vhostName
+		host = hostURI.String()
+	}
+
+	conn, ch, err := getConnectionAndChannel(host)
 	if err != nil {
 		return nil, fmt.Errorf("error establishing rabbitmq connection: %s", err)
 	}
@@ -126,6 +138,11 @@ func parseRabbitMQMetadata(config *ScalerConfig) (*rabbitMQMetadata, error) {
 		meta.queueLength = defaultRabbitMQQueueLength
 	}
 
+	// Resolve vhostName
+	if val, ok := config.TriggerMetadata["vhostName"]; ok {
+		meta.vhostName = &val
+	}
+
 	return &meta, nil
 }
 
@@ -208,6 +225,11 @@ func (s *rabbitMQScaler) getQueueInfoViaHTTP() (*queueInfo, error) {
 
 	vhost := parsedURL.Path
 
+	// Override vhost if requested.
+	if s.metadata.vhostName != nil {
+		vhost = "/" + *s.metadata.vhostName
+	}
+
 	if vhost == "" || vhost == "/" || vhost == "//" {
 		vhost = "/%2F"
 	}

pkg/scalers/rabbitmq_scaler_test.go

@@ -46,6 +46,10 @@ var testRabbitMQMetadata = []parseRabbitMQMetadataTestData{
 	{map[string]string{"queueLength": "10", "queueName": "sample", "host": host, "protocol": "http"}, false, map[string]string{}},
 	// queue name with slashes
 	{map[string]string{"queueLength": "10", "queueName": "namespace/name", "hostFromEnv": host}, false, map[string]string{}},
+	// vhost passed
+	{map[string]string{"vhostName": "myVhost", "queueName": "namespace/name", "hostFromEnv": host}, false, map[string]string{}},
+	// vhost passed but empty
+	{map[string]string{"vhostName": "", "queueName": "namespace/name", "hostFromEnv": host}, false, map[string]string{}},
 }
 
 var rabbitMQMetricIdentifiers = []rabbitMQMetricIdentifier{
@@ -90,81 +94,100 @@ type getQueueInfoTestData struct {
 	response       string
 	responseStatus int
 	isActive       bool
+	extraMetadata  map[string]string
+	vhostPath      string
 }
 
 var testQueueInfoTestData = []getQueueInfoTestData{
-	{`{"messages": 4, "messages_unacknowledged": 1, "name": "evaluate_trials"}`, http.StatusOK, true},
-	{`{"messages": 1, "messages_unacknowledged": 1, "name": "evaluate_trials"}`, http.StatusOK, true},
-	{`{"messages": 1, "messages_unacknowledged": 0, "name": "evaluate_trials"}`, http.StatusOK, true},
-	{`{"messages": 0, "messages_unacknowledged": 0, "name": "evaluate_trials"}`, http.StatusOK, false},
-	{`Password is incorrect`, http.StatusUnauthorized, false},
+	{`{"messages": 4, "messages_unacknowledged": 1, "name": "evaluate_trials"}`, http.StatusOK, true, nil, ""},
+	{`{"messages": 1, "messages_unacknowledged": 1, "name": "evaluate_trials"}`, http.StatusOK, true, nil, ""},
+	{`{"messages": 1, "messages_unacknowledged": 0, "name": "evaluate_trials"}`, http.StatusOK, true, nil, ""},
+	{`{"messages": 0, "messages_unacknowledged": 0, "name": "evaluate_trials"}`, http.StatusOK, false, nil, ""},
+	{`Password is incorrect`, http.StatusUnauthorized, false, nil, ""},
 }
 
 var vhostPathes = []string{"/myhost", "", "/", "//", "/%2F"}
 
+var testQueueInfoTestDataSingleVhost = []getQueueInfoTestData{
+	{`{"messages": 4, "messages_unacknowledged": 1, "name": "evaluate_trials"}`, http.StatusOK, true, map[string]string{"hostFromEnv": "plainHost", "vhostName": "myhost"}, "/myhost"},
+	{`{"messages": 4, "messages_unacknowledged": 1, "name": "evaluate_trials"}`, http.StatusOK, true, map[string]string{"hostFromEnv": "plainHost", "vhostName": "/"}, "/"},
+	{`{"messages": 4, "messages_unacknowledged": 1, "name": "evaluate_trials"}`, http.StatusOK, true, map[string]string{"hostFromEnv": "plainHost", "vhostName": ""}, "/"},
+}
+
 func TestGetQueueInfo(t *testing.T) {
+	allTestData := []getQueueInfoTestData{}
 	for _, testData := range testQueueInfoTestData {
-		testData := testData
 		for _, vhostPath := range vhostPathes {
-			expectedVhost := "myhost"
+			testData := testData
+			testData.vhostPath = vhostPath
+			allTestData = append(allTestData, testData)
+		}
+	}
+	allTestData = append(allTestData, testQueueInfoTestDataSingleVhost...)
 
-			if vhostPath != "/myhost" {
-				expectedVhost = "%2F"
+	for _, testData := range allTestData {
+		testData := testData
+		expectedVhost := "myhost"
+
+		if testData.vhostPath != "/myhost" {
+			expectedVhost = "%2F"
+		}
+
+		var apiStub = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			expectedPath := "/api/queues/" + expectedVhost + "/evaluate_trials"
+			if r.RequestURI != expectedPath {
+				t.Error("Expect request path to =", expectedPath, "but it is", r.RequestURI)
 			}
 
-			var apiStub = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				expectedPath := "/api/queues/" + expectedVhost + "/evaluate_trials"
-				if r.RequestURI != expectedPath {
-					t.Error("Expect request path to =", expectedPath, "but it is", r.RequestURI)
-				}
+			w.WriteHeader(testData.responseStatus)
+			_, err := w.Write([]byte(testData.response))
+			if err != nil {
+				t.Error("Expect request path to =", testData.response, "but it is", err)
+			}
+		}))
 
-				w.WriteHeader(testData.responseStatus)
-				_, err := w.Write([]byte(testData.response))
-				if err != nil {
-					t.Error("Expect request path to =", testData.response, "but it is", err)
-				}
-			}))
+		resolvedEnv := map[string]string{host: fmt.Sprintf("%s%s", apiStub.URL, testData.vhostPath), "plainHost": apiStub.URL}
 
-			resolvedEnv := map[string]string{host: fmt.Sprintf("%s%s", apiStub.URL, vhostPath)}
+		metadata := map[string]string{
+			"queueLength": "10",
+			"queueName":   "evaluate_trials",
+			"hostFromEnv": host,
+			"protocol":    "http",
+		}
+		for k, v := range testData.extraMetadata {
+			metadata[k] = v
+		}
 
-			metadata := map[string]string{
-				"queueLength": "10",
-				"queueName":   "evaluate_trials",
-				"hostFromEnv": host,
-				"protocol":    "http",
-			}
+		s, err := NewRabbitMQScaler(
+			&ScalerConfig{
+				ResolvedEnv:       resolvedEnv,
+				TriggerMetadata:   metadata,
+				AuthParams:        map[string]string{},
+				GlobalHTTPTimeout: 1000 * time.Millisecond,
+			},
+		)
 
-			s, err := NewRabbitMQScaler(
-				&ScalerConfig{
-					ResolvedEnv:       resolvedEnv,
-					TriggerMetadata:   metadata,
-					AuthParams:        map[string]string{},
-					GlobalHTTPTimeout: 1000 * time.Millisecond,
-				},
-			)
+		if err != nil {
+			t.Error("Expect success", err)
+		}
+
+		ctx := context.TODO()
+		active, err := s.IsActive(ctx)
 
+		if testData.responseStatus == http.StatusOK {
 			if err != nil {
 				t.Error("Expect success", err)
 			}
 
-			ctx := context.TODO()
-			active, err := s.IsActive(ctx)
-
-			if testData.responseStatus == http.StatusOK {
-				if err != nil {
-					t.Error("Expect success", err)
-				}
-
-				if active != testData.isActive {
-					if testData.isActive {
-						t.Error("Expect to be active")
-					} else {
-						t.Error("Expect to not be active")
-					}
+			if active != testData.isActive {
+				if testData.isActive {
+					t.Error("Expect to be active")
+				} else {
+					t.Error("Expect to not be active")
 				}
-			} else if !strings.Contains(err.Error(), testData.response) {
-				t.Error("Expect error to be like '", testData.response, "' but it's '", err, "'")
 			}
+		} else if !strings.Contains(err.Error(), testData.response) {
+			t.Error("Expect error to be like '", testData.response, "' but it's '", err, "'")
 		}
 	}
 }