kafka-broker-dispatcher rate limiter

[lamluongbinh]

Fixes #

Proposed Changes

• Enable rate limiter support for the Knative Kafka broker dispatcher.
• Add an annotation to control vreplicas through trigger annotations.
• Fix the rate limiter logic. Based on my knowledge, we previously didn't have a functioning rate limiter since tokens were only being added but not consumed.

[linux-foundation-easycla]

• ❌ - login: @lamluongbinh / name: Michael. Lam . The commit (4ee6a65, 13961b7, d310486) is not authorized under a signed CLA. Please click here to be authorized. For further assistance with EasyCLA, please submit a support request ticket.

[knative-prow]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by: lamluongbinh
Once this PR has been reviewed and has the lgtm label, please assign pierdipi for approval. For more information see the Code Review Process.

The full list of commands accepted by this bot can be found here.

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[knative-prow]

Welcome @lamluongbinh! It looks like this is your first PR to knative-extensions/eventing-kafka-broker 🎉

[knative-prow]

Hi @lamluongbinh. Thanks for your PR.

I'm waiting for a knative-extensions member to verify that this patch is reasonable to test. If it is, they should reply with /ok-to-test on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should join the org to skip this step.

Once the patch is verified, the new status will be reflected by the ok-to-test label.

I understand the commands that are listed here.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository.

[Cali0707]

I don't think using this annotation is the correct direction to go for manually controlling the scale of triggers. There has been discussion about implementing the scale subresource for the triggers and then using that here to set the vreplicas for the egress.

Any thoughts @pierDipi @creydr ?

[lamluongbinh]

Thanks for your reply, @Cali0707. Do we have any documentation about the scale subresource that you mentioned above? Is it k8s scale subresource?

I added the vreplicas annotation here since it’s easier to manage via annotations. Making changes directly in the CRDs would be riskier and require additional modifications.

If we don’t use annotations, I’m curious if there’s a manual way to control the vreplica/rate for each trigger. This would be especially useful since each trigger sends messages to different services, each with unique needs, SLOs, and allocated resources.

[Cali0707]

The scale subresource is documented here: https://kubernetes.io/docs/tasks/extend-kubernetes/custom-resources/custom-resource-definitions/#scale-subresource

It provides a consistent API for anything wanting to scale a resource in the k8s ecosystem. With this approach, your trigger spec would look something like:

apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata: ...
spec:
  scale: 10
  # other spec stuff here

[lamluongbinh]

In Kubernetes, the scale subresource primarily facilitates scaling out physical dispatcher pods, which can enhance processing capacity and accelerate resource handling. However, simply increasing the number of pods does not directly address rate limiting. In fact, scaling out could unintentionally disrupt the rate limiting logic, as adding more pods would result in a higher overall processing rate.

The current rate limiting formula is:

Rate Limit = MaxPollRecords × vReplicas × Number of Dispatcher Pods

Scaling pods out just make us have higher rate limit for all triggers.

Current Approach
In our current implementation, rate limits can be controlled by adjusting the number of vReplicas and dispatcher pods. To support this, I have introduced an annotation that enables dynamic rate limiting by modifying the vReplicas value.

Future Considerations
If we decide to utilize the scale subresource to improve processing speed in the future, additional logic will be required to ensure rate limits remain consistent. This could be achieved by introducing new annotations or configuration options that control rate limits independently. A potential solution would be dynamically adjusting rate limits within the trigger configuration, such as:

Example 1: Defining rate limits in the spec

apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata: ...
spec:
  maxRatePerPod: 30

Example 2: Using annotations for rate limits

apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  annotations:
    kafka.eventing.knative.dev/maxRatePerPod: "30"

Ultimately, achieving a balance between scaling and rate limiting is crucial to maintaining system performance while ensuring that processing limits are not exceeded.

Any thoughts on your end?