Activation threshold (activationValue) for Rabbitmq

CHANGELOG.md

@@ -46,7 +46,7 @@
 - **Metric API Scaler:** Improve error handling on not-ok response ([#2317](https://github.com/kedacore/keda/issues/2317))
 - **Prometheus Scaler:** Support for `X-Scope-OrgID` header in Prometheus scaler ([#2667](https://github.com/kedacore/keda/issues/2667))
 - **RabbitMQ Scaler:** Include `vhost` for RabbitMQ when retrieving queue info with `useRegex` ([#2498](https://github.com/kedacore/keda/issues/2498))
-- **RabbitMQ Scaler:** Add activation threshold `minMetricValue` ([#2800](https://github.com/kedacore/keda/issues/2800))
+- **RabbitMQ Scaler:** Add activation threshold `activationValue` ([#2800](https://github.com/kedacore/keda/issues/2800))
 
 ### Breaking Changes
 

pkg/scalers/rabbitmq_scaler.go

@@ -29,15 +29,15 @@ func init() {
 }
 
 const (
-	rabbitQueueLengthMetricName           = "queueLength"
-	rabbitModeTriggerConfigName           = "mode"
-	rabbitValueTriggerConfigName          = "value"
-	rabbitMinMetricValueTriggerConfigName = "minMetricValue"
-	rabbitModeQueueLength                 = "QueueLength"
-	rabbitModeMessageRate                 = "MessageRate"
-	defaultRabbitMQQueueLength            = 20
-	rabbitMetricType                      = "External"
-	rabbitRootVhostPath                   = "/%2F"
+	rabbitQueueLengthMetricName            = "queueLength"
+	rabbitModeTriggerConfigName            = "mode"
+	rabbitValueTriggerConfigName           = "value"
+	rabbitActivationValueTriggerConfigName = "activationValue"
+	rabbitModeQueueLength                  = "QueueLength"
+	rabbitModeMessageRate                  = "MessageRate"
+	defaultRabbitMQQueueLength             = 20
+	rabbitMetricType                       = "External"
+	rabbitRootVhostPath                    = "/%2F"
 )
 
 const (
@@ -62,19 +62,19 @@ type rabbitMQScaler struct {
 }
 
 type rabbitMQMetadata struct {
-	queueName      string
-	mode           string        // QueueLength or MessageRate
-	value          int           // trigger value (queue length or publish/sec. rate)
-	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
-	useRegex       bool          // specify if the queueName contains a rexeg
-	pageSize       int64         // specify the page size if useRegex is enabled
-	operation      string        // specify the operation to apply in case of multiples queues
-	metricName     string        // custom metric name for trigger
-	timeout        time.Duration // custom http timeout for a specific trigger
-	scalerIndex    int           // scaler index
-	minMetricValue float64       // min metric value
+	queueName         string
+	mode              string        // QueueLength or MessageRate
+	value             int           // trigger value (queue length or publish/sec. rate)
+	actionvationValue int           // activation value
+	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
+	useRegex          bool          // specify if the queueName contains a rexeg
+	pageSize          int64         // specify the page size if useRegex is enabled
+	operation         string        // specify the operation to apply in case of multiples queues
+	metricName        string        // custom metric name for trigger
+	timeout           time.Duration // custom http timeout for a specific trigger
+	scalerIndex       int           // scaler index
 }
 
 type queueInfo struct {
@@ -264,7 +264,7 @@ func parseTrigger(meta *rabbitMQMetadata, config *ScalerConfig) (*rabbitMQMetada
 	deprecatedQueueLengthValue, deprecatedQueueLengthPresent := config.TriggerMetadata[rabbitQueueLengthMetricName]
 	mode, modePresent := config.TriggerMetadata[rabbitModeTriggerConfigName]
 	value, valuePresent := config.TriggerMetadata[rabbitValueTriggerConfigName]
-	minMetricValueStr, minMetricValuePresent := config.TriggerMetadata[rabbitMinMetricValueTriggerConfigName]
+	activationValue, activationValuePresent := config.TriggerMetadata[rabbitActivationValueTriggerConfigName]
 
 	// Initialize to default trigger settings
 	meta.mode = rabbitModeQueueLength
@@ -280,13 +280,13 @@ func parseTrigger(meta *rabbitMQMetadata, config *ScalerConfig) (*rabbitMQMetada
 		return nil, fmt.Errorf("queueLength is deprecated; configure only %s and %s", rabbitModeTriggerConfigName, rabbitValueTriggerConfigName)
 	}
 
-	// Parse minMetricValue
-	if minMetricValuePresent {
-		minMetricValue, err := strconv.ParseFloat(minMetricValueStr, 64)
+	// Parse activation value
+	if activationValuePresent {
+		activation, err := strconv.Atoi(activationValue)
 		if err != nil {
-			return nil, fmt.Errorf("can't parse %s: %s", rabbitMinMetricValueTriggerConfigName, err)
+			return nil, fmt.Errorf("can't parse %s: %s", rabbitActivationValueTriggerConfigName, err)
 		}
-		meta.minMetricValue = minMetricValue
+		meta.actionvationValue = activation
 	}
 
 	// Parse deprecated `queueLength` value
@@ -364,9 +364,9 @@ func (s *rabbitMQScaler) IsActive(ctx context.Context) (bool, error) {
 	}
 
 	if s.metadata.mode == rabbitModeQueueLength {
-		return float64(messages) > s.metadata.minMetricValue, nil
+		return messages > s.metadata.actionvationValue, nil
 	}
-	return publishRate > s.metadata.minMetricValue || float64(messages) > s.metadata.minMetricValue, nil
+	return publishRate > float64(s.metadata.actionvationValue) || float64(messages) > float64(s.metadata.actionvationValue), nil
 }
 
 func (s *rabbitMQScaler) getQueueStatus() (int, float64, error) {

pkg/scalers/rabbitmq_scaler_test.go

@@ -111,10 +111,10 @@ var testRabbitMQMetadata = []parseRabbitMQMetadataTestData{
 	{map[string]string{"mode": "MessageRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true", "pageSize": "-1"}, true, map[string]string{}},
 	// invalid pageSize
 	{map[string]string{"mode": "MessageRate", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true", "pageSize": "a"}, true, map[string]string{}},
-	// minMetricValue passed
-	{map[string]string{"minMetricValue": "10.0", "queueLength": "10", "queueName": "sample", "hostFromEnv": host}, false, map[string]string{}},
-	// malformed minMetricValue
-	{map[string]string{"minMetricValue": "AA", "queueLength": "10", "queueName": "sample", "hostFromEnv": host}, true, map[string]string{}},
+	// activationValue passed
+	{map[string]string{"activationValue": "10", "queueLength": "20", "queueName": "sample", "hostFromEnv": host}, false, map[string]string{}},
+	// malformed activationValue
+	{map[string]string{"activationValue": "AA", "queueLength": "10", "queueName": "sample", "hostFromEnv": host}, true, map[string]string{}},
 }
 
 var rabbitMQMetricIdentifiers = []rabbitMQMetricIdentifier{

tests/scalers/rabbitmq-queue-amqp.test.ts

@@ -1,7 +1,7 @@
 import * as sh from 'shelljs'
 import test from 'ava'
 import { RabbitMQHelper } from './rabbitmq-helpers'
-import {waitForDeploymentReplicaCount} from "./helpers";
+import {waitForDeploymentReplicaCount, sleep} from "./helpers";
 
 const testNamespace = 'rabbitmq-queue-amqp-test'
 const rabbitmqNamespace = 'rabbitmq-amqp-test'
@@ -11,7 +11,7 @@ const password = "test-password"
 const vhost = "test-vh"
 const connectionString = `amqp://${username}:${password}@rabbitmq.${rabbitmqNamespace}.svc.cluster.local/${vhost}`
 const messageCount = 500
-const minMetricValue = 10
+const activationValue = 10
 test.before(t => {
   RabbitMQHelper.installRabbit(t, username, password, vhost, rabbitmqNamespace)
 
@@ -35,10 +35,10 @@ test.serial(`Deployment should scale to 4 with ${messageCount} messages on the q
   t.true(await waitForDeploymentReplicaCount(0, 'test-deployment', testNamespace, 50, 5000), 'Replica count should be 0 after 3 minutes')
 })
 
-test.serial(`Deployment shouldn't scale because the 5 messages on the queue are less than ${minMetricValue}(minMetricValue)`, async t => {
+test.serial(`Deployment shouldn't scale because the 5 messages on the queue are less than ${activationValue}(activationValue)`, async t => {
   RabbitMQHelper.publishMessages(t, testNamespace, connectionString, 5, queueName)
-
-  // with messages published, the consumer deployment should start receiving the messages
+  await sleep(20000);
+  // since the messages are less than activationValue, the consumer deployment should not scale
   t.true(await waitForDeploymentReplicaCount(0, 'test-deployment', testNamespace, 20, 5000), 'Replica count remain 0 after 10 seconds')
 })
 
@@ -111,4 +111,4 @@ spec:
       hostFromEnv: RabbitMqHost
       mode: QueueLength
       value: '50'
-      minMetricValue: '10'`
+      activationValue: '10'`