KEDA scaler for Redis Streams

[abhirockzz]

Redis Streams, a data type introduced with Redis 5.0 is an append-only log data structure. One of its features includes "consumer groups" that allows a group of clients to co-operate consuming a different portion of the same stream of messages.

This scaler can check the Pending Entries List (see XPENDING https://redis.io/commands/xpending) for a specific Consumer Group of a Redis Stream and scale-out the Deployment as per the count returned (by XPENDING command). If you have a Redis Streams consumer application that is unable to keep up with the rate of production, KEDA can help by auto-scaling the no. of instances (part of the same consumer group) so that it can start sharing the processing workload.

• Scaler Source: Redis Streams (Pending Entries List aka PEL)
• Authentication: Password-based authentication supported by Redis

[abhirockzz]

If this sounds interesting, I can continue working on the implementation

[abhirockzz]

Here are some additional points:

1. Consumer application must acknowledge the processes message (XACK https://redis.io/commands/xack) else the “Pending Entry List” will keep increasing indefinitely if messages continue to be inserted in the stream. The scaler can keep a cap on the no. of instances by using maxReplicaCount
2. If a consumer instance has failed or crashed - in this case, simply scaling out your application may not help (since Redis treats consumer groups as a stateful entity and stores the processing state of each consumer instance, including the messages which were received but not acknowledged - see https://redis.io/topics/streams-intro#consumer-groups). This needs to be handled - examples below (each has its own caveats)
   1. Explicitly reading the history (using 0-0 ID) to process previous messages which were received but not acknowledged (see Ruby example here https://redis.io/topics/streams-intro#creating-a-consumer-group)
   2. Using XCLAIM (https://redis.io/topics/streams-intro#recovering-from-permanent-failures) to take ownership of such messages and process them (probably as an out-of-band process)

These points have been mentioned since these characteristics are specific to Redis Streams based consumer applications irrespective of the scaler

[abhirockzz]

The ScaledObject for this scaler will look something like this:

apiVersion: keda.k8s.io/v1alpha1
kind: ScaledObject
metadata:
  name: redis-streams-scaledobject
  namespace: default
  labels:
    deploymentName: redis-streams-consumer
spec:
  scaleTargetRef:
    deploymentName: redis-streams-consumer
  pollingInterval: 15
  cooldownPeriod: 200
  maxReplicaCount: 25
  minReplicaCount: 1
  triggers:
    - type: redis-streams
      metadata:
        address: REDIS_HOST
        stream: my-stream
        consumerGroup: consumer-group-1
        pendingEntriesCount: "10"
      authenticationRef:
        name: keda-redis-stream-triggerauth

stream and consumerGroup are names of the stream and a consumer group respectively and pendingEntriesCount is the threshold for the count of Pending Entries (obtained using XPENDING) for that consumer group in the stream

[tomkerkhove]

I've added it to today's standup meeting but think this would be a good addition - Will discuss and let you know!

Feel free to join if you want as well.

[yifan]

Thanks for the work. I wonder why you have chosen the count of Pending Entries as the trigger for scaling. Based my own understanding, the count should equal to the number of consumers (each will read an entry, process and acknowledge it). The increase of Pending Entries correlate to the increase of the number of consumers, it seems to does the opposite as the trigger for scaler. @abhirockzz

[accnops]

Can I use this for a job queue? From my understanding and testing, when a message is added using XADD, the consumer's Pending Entries List (PEL) does not grow. It's only when a consumer executes XREADGROUP that the message is added to the PEL. This makes this unusable, since it's specifically that consumer I want to scale.

Is this correct?

For now, I'll try using Redis lists.

[nakumgaurav]

@accnops is right. If I have an application adding messages to a stream, they are not delivered to any consumer group until an XREADGROUP is issued over that consumer group. This means that if the Job that KEDA is trying to scale consumes messages, it has to wait for someone to consume messages (!!!) using XREADGROUP, so that the messages are moved to the PEL of that group. This leaves Redis Streams pretty much useless with KEDA because if some other application has already fetched the messages, how would I retrieve them in my KEDA application that I am trying to scale?

Why not have an option to create a Job without a consumer group? This way any new messages added to a stream would trigger a Job and the user would be able to do their processing over the newly added messages.

Keda developers, please let us know if there is a workaround to this issue.

[zroubalik]

@nakumgaurav could you please open a separate issue for this?

[nakumgaurav]

Thank you @zroubalik I filed my question and a potential workaround in #2044