KEDA Autoscaler for BullWorkers

[avneeshkumar]

I have different queues and I am running a consumer group for each queue. Now I need to upscale or downscale the number of consumers instances in the consumer groups based on number of items in queue.

Now in KEDA supports https://keda.sh/docs/2.10/scalers/redis-lists/ and Redis Stream https://keda.sh/docs/2.10/scalers/redis-streams/

Is there someway to get the queue length that can be used directly with KEDA Autoscalers ?

[manast]

You can get the number of jobs waiting in the queue: https://api.docs.bullmq.io/classes/Queue.html#getWaitingCount

[avneeshkumar]

Hey @manast Yes I know that I can get that from inside the nodejs. But i don't want my worker to update that information somewhere on Azure application Insights or AWS cloudwatch. I want to use KEDA redis scalers.

For every queue I see following redis items

1) "bull:queueName:meta" 
2) "bull:lqueueName:wait"
4) "bull:lqueueName:id"
6) "bull:queueName:events"

bull:queueName:meta - This is of type hash and contains following :-

127.0.0.1:6379> hgetall bull:local-triggered-campaign-unprocessed-5:meta
1) "opts.maxLenEvents"
2) "10000"

bull:lqueueName:wait - This is of type list and it contains following :-

1) "fa5c3adc-3c82-4724-b7ac-253164b00c6e"
2) "2"
3) "b3725de9-a17c-4e62-9cd3-849a15ca1241"

bull:lqueueName:id - This is of type string and contains a numeric value

bull:queueName:event - This is of type stream

if I do a llen bull:queueName:wait this will give me approximate number of events in queue ? Is my understanding correct ?

And what number bull:lqueueName:id this stores ?

[lohart13]

I have been using the "bull:_queue_name_:wait" key with KEDA which allows scaling objects/jobs based on the qty of jobs in the wait list.

In the KEDA triggers section for redis, set the listName equal to "bull:_queue_name_:wait" (where _queue_name_ is the name of the queue) and then use listLength to trigger the scaling of the jobs/objects in relation to the number of waiting jobs. See example:

triggers:
- type: redis
  metadata:
    address: localhost:6379
    listName: bull:_queue_name_:wait
    listLength: "5"

[lohart13]

@manast has something changed in v4.0.0 and above when using job priority?

The I'm using v4.1.0 and the "bull:_queue_name_:wait" key doesn't seem to exist anymore for a queue which has jobs that have a priority set.

Without the wait key, KEDA is unable to scale the jobs/objects.

Is this something that can be added or is it best to revert to v3.15.x?

[manast]

There are breaking changes in v4.0.0, https://docs.bullmq.io/changelog#4.0.0-2023-06-21
If you use priority jobs you will need to use getPrioritizedCount https://api.docs.bullmq.io/classes/QueueGetters.html#getPrioritizedCount

[lohart13]

Thanks, this is working after changing to the "bull:_queue_name_:prioritized" key as the listName in the keda trigger config.

[manast]

note that prioritized is a ZSET not a LIST so you need to use ZCARD to calculate the number of jobs in there https://redis.io/commands/zcard/

[lohart13]

Keda checks the type of the key passed to listName and automatically selects the command to use to calculate the length of type as per the below.
https://github.com/kedacore/keda/blob/a57d00aad63fb7799a7686a9e4de7bd58ef92cf2/pkg/scalers/redis_scaler.go#L71-L83