Activation threshold (activationValue) for Rabbitmq

CHANGELOG.md

@@ -137,6 +137,7 @@ To learn more about our roadmap, we recommend reading [this document](ROADMAP.md
 - **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))
 
 ### Deprecations

pkg/scalers/rabbitmq_scaler.go

@@ -27,14 +27,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 (
@@ -63,6 +64,7 @@ type rabbitMQMetadata struct {
 	queueName             string
 	mode                  string        // QueueLength or MessageRate
 	value                 float64       // trigger value (queue length or publish/sec. rate)
+	activationValue       float64       // 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
@@ -286,6 +288,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
@@ -301,6 +304,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.ParseFloat(activationValue, 64)
+		if err != nil {
+			return nil, fmt.Errorf("can't parse %s: %s", rabbitActivationValueTriggerConfigName, err)
+		}
+		meta.activationValue = activation
+	}
+
 	// Parse deprecated `queueLength` value
 	if deprecatedQueueLengthPresent {
 		queueLength, err := strconv.ParseFloat(deprecatedQueueLengthValue, 64)
@@ -310,6 +322,10 @@ func parseTrigger(meta *rabbitMQMetadata, config *ScalerConfig) (*rabbitMQMetada
 		meta.mode = rabbitModeQueueLength
 		meta.value = queueLength
 
+		if meta.activationValue > queueLength {
+			return nil, fmt.Errorf("activationValue is greater than value")
+		}
+
 		return meta, nil
 	}
 
@@ -339,6 +355,10 @@ func parseTrigger(meta *rabbitMQMetadata, config *ScalerConfig) (*rabbitMQMetada
 		return nil, fmt.Errorf("protocol %s not supported; must be http to use mode %s", meta.protocol, rabbitModeMessageRate)
 	}
 
+	if meta.activationValue > meta.value {
+		return nil, fmt.Errorf("activationValue is greater than value")
+	}
+
 	return meta, nil
 }
 
@@ -376,9 +396,9 @@ func (s *rabbitMQScaler) IsActive(ctx context.Context) (bool, error) {
 	}
 
 	if s.metadata.mode == rabbitModeQueueLength {
-		return messages > 0, nil
+		return float64(messages) > s.metadata.activationValue, nil
 	}
-	return publishRate > 0 || messages > 0, nil
+	return publishRate > s.metadata.activationValue || float64(messages) > s.metadata.activationValue, 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,12 @@ 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{}},
+	// invalid activationValue > queueLength
+	{map[string]string{"activationValue": "20", "queueLength": "10", "queueName": "sample", "hostFromEnv": host}, true, map[string]string{}},
+	// malformed activationValue
+	{map[string]string{"activationValue": "AA", "queueLength": "10", "queueName": "sample", "hostFromEnv": host}, true, map[string]string{}},
 	// http and excludeUnacknowledged
 	{map[string]string{"mode": "QueueLength", "value": "1000", "queueName": "sample", "host": "http://", "useRegex": "true", "excludeUnacknowledged": "true"}, false, map[string]string{}},
 	// amqp and excludeUnacknowledged

tests/scalers_go/rabbitmq/rabbitmq_helper.go

@@ -4,6 +4,7 @@
 package rabbitmq
 
 import (
+	"fmt"
 	"testing"
 
 	"k8s.io/client-go/kubernetes"
@@ -198,6 +199,9 @@ func RMQPublishMessages(t *testing.T, namespace, connectionString, queueName str
 		MessageCount: messageCount,
 	}
 
+	// Before push messages remove previous jobs if any
+	_, _ = helper.ExecuteCommand(fmt.Sprintf("kubectl delete jobs/rabbitmq-publish-%s --namespace %s", queueName, namespace))
+
 	helper.KubectlApplyWithTemplate(t, data, "rmqPublishTemplate", publishTemplate)
 }
 

tests/scalers_go/rabbitmq/rabbitmq_queue_amqp/rabbitmq_queue_amqp_test.go

@@ -7,6 +7,7 @@ import (
 	"encoding/base64"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/joho/godotenv"
 	"github.com/stretchr/testify/assert"
@@ -58,6 +59,7 @@ spec:
         hostFromEnv: RabbitApiHost
         mode: QueueLength
         value: '10'
+        activationValue: '5'
 `
 )
 
@@ -87,6 +89,8 @@ func TestScaler(t *testing.T) {
 
 	testScaling(t, kc)
 
+	testActivationValue(t, kc)
+
 	// cleanup
 	t.Log("--- cleaning up ---")
 	DeleteKubernetesResources(t, kc, testNamespace, data, templates)
@@ -117,3 +121,15 @@ func testScaling(t *testing.T, kc *kubernetes.Clientset) {
 	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 0, 60, 1),
 		"replica count should be 0 after 1 minute")
 }
+
+func testActivationValue(t *testing.T, kc *kubernetes.Clientset) {
+	t.Log("--- testing activation value ---")
+	messagesToQueue := 3
+	RMQPublishMessages(t, rmqNamespace, connectionString, queueName, messagesToQueue)
+
+	// Wait for replicas
+	time.Sleep(60 * time.Second)
+
+	assert.True(t, WaitForDeploymentReplicaReadyCount(t, kc, deploymentName, testNamespace, 0, 60, 1),
+		"replica count should be 0 after 1 minute")
+}