Activation threshold (activationValue) for Rabbitmq

CHANGELOG.md

@@ -55,6 +55,7 @@
 - **Prometheus Scaler:** Check and properly inform user that `threshold` is not set ([#2793](https://github.com/kedacore/keda/issues/2793))
 - **Prometheus Scaler:** Support for `X-Scope-OrgID` header ([#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 `activationValue` ([#2800](https://github.com/kedacore/keda/issues/2800))
 - **Selenium Grid Scaler:** Consider `maxSession` grid info when scaling. ([#2618](https://github.com/kedacore/keda/issues/2618))
 
 ### Breaking Changes

pkg/scalers/rabbitmq_scaler.go

@@ -29,14 +29,15 @@ func init() {
 }
 
 const (
-	rabbitQueueLengthMetricName  = "queueLength"
-	rabbitModeTriggerConfigName  = "mode"
-	rabbitValueTriggerConfigName = "value"
-	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 (
@@ -61,18 +62,19 @@ type rabbitMQScaler struct {
 }
 
 type rabbitMQMetadata struct {
-	queueName   string
-	mode        string        // QueueLength or MessageRate
-	value       int64         // 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
+	queueName         string
+	mode              string        // QueueLength or MessageRate
+	value             int64         // trigger value (queue length or publish/sec. rate)
+	actionvationValue int64         // 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 {
@@ -262,6 +264,7 @@ func parseTrigger(meta *rabbitMQMetadata, config *ScalerConfig) (*rabbitMQMetada
 	deprecatedQueueLengthValue, deprecatedQueueLengthPresent := config.TriggerMetadata[rabbitQueueLengthMetricName]
 	mode, modePresent := config.TriggerMetadata[rabbitModeTriggerConfigName]
 	value, valuePresent := config.TriggerMetadata[rabbitValueTriggerConfigName]
+	activationValue, activationValuePresent := config.TriggerMetadata[rabbitActivationValueTriggerConfigName]
 
 	// Initialize to default trigger settings
 	meta.mode = rabbitModeQueueLength
@@ -277,6 +280,15 @@ func parseTrigger(meta *rabbitMQMetadata, config *ScalerConfig) (*rabbitMQMetada
 		return nil, fmt.Errorf("queueLength is deprecated; configure only %s and %s", rabbitModeTriggerConfigName, rabbitValueTriggerConfigName)
 	}
 
+	// Parse activation value
+	if activationValuePresent {
+		activation, err := strconv.ParseInt(activationValue, 10, 64)
+		if err != nil {
+			return nil, fmt.Errorf("can't parse %s: %s", rabbitActivationValueTriggerConfigName, err)
+		}
+		meta.actionvationValue = activation
+	}
+
 	// Parse deprecated `queueLength` value
 	if deprecatedQueueLengthPresent {
 		queueLength, err := strconv.ParseInt(deprecatedQueueLengthValue, 10, 64)
@@ -352,9 +364,9 @@ func (s *rabbitMQScaler) IsActive(ctx context.Context) (bool, error) {
 	}
 
 	if s.metadata.mode == rabbitModeQueueLength {
-		return messages > 0, nil
+		return messages > s.metadata.actionvationValue, nil
 	}
-	return publishRate > 0 || messages > 0, nil
+	return publishRate > float64(s.metadata.actionvationValue) || messages > s.metadata.actionvationValue, nil
 }
 
 func (s *rabbitMQScaler) getQueueStatus() (int64, float64, error) {

pkg/scalers/rabbitmq_scaler_test.go

@@ -44,7 +44,7 @@ var testRabbitMQMetadata = []parseRabbitMQMetadataTestData{
 	// missing queueName
 	{map[string]string{"queueLength": "10", "hostFromEnv": host}, true, map[string]string{}},
 	// host defined in authParams
-	{map[string]string{"queueLength": "10"}, true, map[string]string{"host": host}},
+	{map[string]string{"queueLength": "10", "hostFromEnv": host}, true, map[string]string{"host": host}},
 	// properly formed metadata with http protocol
 	{map[string]string{"queueLength": "10", "queueName": "sample", "host": host, "protocol": "http"}, false, map[string]string{}},
 	// queue name with slashes
@@ -111,6 +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{}},
+	// 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 activationValue = 10
 test.before(t => {
   RabbitMQHelper.installRabbit(t, username, password, vhost, rabbitmqNamespace)
 
@@ -35,6 +35,13 @@ 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 ${activationValue}(activationValue)`, async t => {
+  RabbitMQHelper.publishMessages(t, testNamespace, connectionString, 5, queueName)
+  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')
+})
+
 test.after.always.cb('clean up rabbitmq-queue deployment', t => {
   const resources = [
     'scaledobject.keda.sh/test-scaledobject',
@@ -103,4 +110,5 @@ spec:
       queueName: {{QUEUE_NAME}}
       hostFromEnv: RabbitMqHost
       mode: QueueLength
-      value: '50'`
+      value: '50'
+      activationValue: '10'`