diff --git a/pip/pip-406.md b/pip/pip-406.md new file mode 100644 index 0000000000000..b541d48c380ce --- /dev/null +++ b/pip/pip-406.md @@ -0,0 +1,176 @@ +# PIP-406: Introduce metrics related to dispatch throttled events + +# Background knowledge + +## Motivation + +Currently, users can monitor subscription backlogs using the `pulsar_subscription_back_log_no_delayed` metric. +However, if [dispatch throttling](https://pulsar.apache.org/docs/next/concepts-throttling/) is configured at the broker/topic/subscription level, +this metric may not accurately reflect whether the backlog is due to insufficient consumer capacity, as it could be caused by dispatch throttling. + +## Goals + +Introduce metrics to indicate the count of `messages/bytes throttled` for **broker/topic/subscription** level rate limit. +This allows users to write PromQL queries to identify subscriptions with high backlogs but low or no throttling, pinpointing backlogs caused by insufficient consumer capacity. + +## In Scope + +- Introduce the metric `pulsar_subscription_dispatch_throttled_msg_events` to represent the total number of times message dispatching was throttled on a subscription +- Introduce the metric `pulsar_subscription_dispatch_throttled_bytes_events` to represent the total number of times byte dispatching was throttled on a subscription + +Since throttling on a subscription can be caused by multiple rate limit(broker/topic/subscription), +these metrics will include a label `reason: broker/topic/subscription` to indicate which throttler is responsible. + +## Out of Scope +- These states are not persistent and will reset upon broker restart/ topic re-load / subscription reconnected. + +# High Level Design +1. Maintain metrics counters for throttling caused by `broker/topic/subscription` rate limit. Increment the appropriate counter whenever throttling occurs. +2. Output these fields when retrieving metrics. + + +# Detailed Design + +## Design & Implementation Details +1. Maintain these fields in AbstractBaseDispatcher. +```java + public final LongAdder dispatchThrottledMsgEventsBySubscriptionLimit = new LongAdder(); + public final LongAdder dispatchThrottledMsgEventsByTopicLimit = new LongAdder(); + public final LongAdder dispatchThrottledMsgEventsByBrokerLimit = new LongAdder(); + public final LongAdder dispatchThrottledBytesEventsBySubscriptionLimit = new LongAdder(); + public final LongAdder dispatchThrottledBytesEventsByTopicLimit = new LongAdder(); + public final LongAdder dispatchThrottledBytesEventsByBrokerLimit = new LongAdder(); +``` + +2. Each time a read occurs, if the expected number of messages or bytes is reduced, increment the metric by one. +```diff + private boolean applyDispatchRateLimitsToReadLimits(DispatchRateLimiter rateLimiter, + MutablePair readLimits, + DispatchRateLimiter.Type limiterType) { ++ int originalMessagesToRead = readLimits.getLeft(); ++ long originalBytesToRead = readLimits.getRight(); + int availablePermitsOnMsg = (int) rateLimiter.getAvailableDispatchRateLimitOnMsg(); + if (availablePermitsOnMsg >= 0) { + readLimits.setLeft(Math.min(readLimits.getLeft(), availablePermitsOnMsg)); + } + long availablePermitsOnByte = rateLimiter.getAvailableDispatchRateLimitOnByte(); + if (availablePermitsOnByte >= 0) { + readLimits.setRight(Math.min(readLimits.getRight(), availablePermitsOnByte)); + } ++ if (readLimits.getLeft() < originalMessagesToRead) { ++ switch (limiterType) { ++ case BROKER -> dispatchThrottledMsgEventsByBrokerLimit.increment(); ++ case TOPIC -> dispatchThrottledMsgEventsByTopicLimit.increment(); ++ case SUBSCRIPTION -> dispatchThrottledMsgEventsBySubscriptionLimit.increment(); ++ default -> {} ++ } ++ } ++ if (readLimits.getRight() < originalBytesToRead) { ++ switch (limiterType) { ++ case BROKER -> dispatchThrottledBytesEventsByBrokerLimit.increment(); ++ case TOPIC -> dispatchThrottledBytesEventsByTopicLimit.increment(); ++ case SUBSCRIPTION -> dispatchThrottledBytesEventsBySubscriptionLimit.increment(); ++ default -> {} ++ } ++ } +``` +3. Print these metrics when retrieving metrics(TopicStats#printTopicStats). +```java +// write dispatch throttling metrics with `reason` labels to identify specific throttling +// causes: by subscription limit, by topic limit, or by broker limit. +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_msg_events", + subsStats.dispatchThrottledMsgEventsBySubscriptionLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "subscription"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_bytes_events", + subsStats.dispatchThrottledBytesEventsBySubscriptionLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "subscription"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_msg_events", + subsStats.dispatchThrottledMsgEventsByTopicLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "topic"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_bytes_events", + subsStats.dispatchThrottledBytesEventsByTopicLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "topic"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_msg_events", + subsStats.dispatchThrottledMsgEventsByBrokerLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "broker"); +writeTopicMetric(stream, "pulsar_subscription_dispatch_throttled_bytes_events", + subsStats.dispatchThrottledBytesEventsByBrokerLimit, cluster, namespace, topic, + splitTopicAndPartitionIndexLabel, "subscription", sub, + "reason", "broker");} +``` + +## Public-facing Changes + +### Metrics + +1. `pulsar_subscription_dispatch_throttled_msg_events`: + - Description: The total number of times message dispatching was throttled on a subscription + - Attributes: + - tenant + - namespace + - topic + - subscription + - `reason`: broker/topic/subscription + - Unit: messages count + +2. `pulsar_subscription_dispatch_throttled_bytes_events`: + - Description: The total number of times byte dispatching was throttled on a subscription + - Attributes: + - tenant + - namespace + - topic + - subscription + - `reason`: broker/topic/subscription + - Unit: messages count + +### API +Add get throttle count interface on admin api `SubscriptionStats`. +```java + /** + * Gets the total number of times message dispatching was throttled on a subscription due to broker rate limits. + * @return the count of throttled message events by subscription limit, default is 0. + */ +long getDispatchThrottledMsgEventsBySubscriptionLimit(); + +/** + * Gets the total number of times bytes dispatching was throttled on a subscription due to broker rate limits. + * @return the count of throttled bytes by subscription limit, default is 0. + */ +long getDispatchThrottledBytesEventsBySubscriptionLimit(); + +/** + * Gets the total number of times message dispatching was throttled on a subscription due to topic rate limits. + * @return the count of throttled message events by topic limit, default is 0. + */ +long getDispatchThrottledMsgEventsByTopicLimit(); + +/** + * Gets the total number of times bytes dispatching was throttled on a subscription due to topic rate limits. + * @return the count of throttled bytes events by topic limit, default is 0. + */ +long getDispatchThrottledBytesEventsByTopicLimit(); + +/** + * Gets the total number of times message dispatching was throttled on a subscription due to broker rate limits. + * @return the count of throttled message events by broker limit, default is 0. + */ +long getDispatchThrottledMsgEventsByBrokerLimit(); + +/** + * Gets the total number of times bytes dispatching was throttled on a subscription due to broker rate limits. + * @return the count of throttled bytes events by broker limit, default is 0. + */ +long getDispatchThrottledBytesEventsByBrokerLimit(); +``` + +### Configuration +- None + +# Backward & Forward Compatibility +- Full Compatibility +