Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Activation threshold (activationValue) for Rabbitmq #2831

Merged
merged 18 commits into from
Aug 3, 2022
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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))
JorTurFer marked this conversation as resolved.
Show resolved Hide resolved

### Breaking Changes

Expand Down
60 changes: 30 additions & 30 deletions pkg/scalers/rabbitmq_scaler.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -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 {
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
adborroto marked this conversation as resolved.
Show resolved Hide resolved
}

func (s *rabbitMQScaler) getQueueStatus() (int, float64, error) {
Expand Down
8 changes: 4 additions & 4 deletions pkg/scalers/rabbitmq_scaler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand Down
12 changes: 6 additions & 6 deletions tests/scalers/rabbitmq-queue-amqp.test.ts
Original file line number Diff line number Diff line change
@@ -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'
Expand All @@ -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)

Expand All @@ -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')
})

Expand Down Expand Up @@ -111,4 +111,4 @@ spec:
hostFromEnv: RabbitMqHost
mode: QueueLength
value: '50'
minMetricValue: '10'`
activationValue: '10'`