[improve][pip] PIP-323: Complete Backlog Quota Telemetry

pip/pip-323.md

@@ -0,0 +1,171 @@
+# PIP-323: Complete Backlog Quota Telemetry
+
+# Background knowledge
+
+## Backlog
+
+A topic in Pulsar is the place where messages are written to. They are consumed by subscriptions. A topic can have many
+subscriptions, and it is those that maintains the state of message acknowledgment, per subscription - which messages
+were acknowledged and which were not. 
+
+A subscription backlog is the set of unacknowledged messages in that subscription.
+A subscription backlog size is the sum of the size of the unacknowledged messages (in bytes)..
+
+Since a topic can have many subscriptions, and each has its own backlog, how does one define a backlog for a topic?
+A topic backlog is defined as the backlog of the subscription which has the **oldest** unacknowledged message. 
+Since acknowledged messages can be interleaved with unacknowledged messages, calculating the exact size of that 
+subscription backlog can be expensive as it requires I/O operations to read the messages from the ledgers.
+For that reason, the topic backlog size is actually defined to be the *estimated* backlog size of that subscription. 
+It does so by summarizing the size of all the ledgers, starting from the current active one (the one being written to),
+up to the ledger which contains the oldest unacknowledged message for that subscription (There is actually a faster 
+way to calculate it, but this was the definition chosen for this estimation in Pulsar).
+
+A topic backlog age is the age of the oldest unacknowledged message (same subscription as defined for topic backlog size).
+If that message was written 30 minutes ago, its age is 30 minutes, and so is the topic backlog age.
+
+## Backlog Quota
+
+Pulsar has a feature called [backlog quota](https://pulsar.apache.org/docs/3.1.x/cookbooks-retention-expiry/#backlog-quotas). 
+It allows a user to define a quota - in effect, a limit - which limits the topic backlog.
+There are two types of quotas:
+
+1. Size based: The limit is for the topic backlog size (as we defined above).
+2. Time based: The limit is for the topic backlog age (as we defined above).
+
+Once a topic backlog exceeds either one of those limits, an action is taken to hold the backlog to that limit:
+
+* The producer write is placed on hold for a certain amount of time before failing.
+* The producer write is failed
+* The subscriptions oldest unacknowledged messages will be acknowledged in-order until both the topic backlog size or 
+  age will fall inside the limit (quota). The process is called backlog eviction (happens every interval).
+
+The quotas can be defined as a default value for any topic, by using the following broker configuration keys:
+`backlogQuotaDefaultLimitBytes` and `backlogQuotaDefaultLimitSecond`.
+
+The quota can also be specified directly for all topics in a given namespace using the namespace policy, 
+or a specific topic using a topic policy. 
+
+## Monitoring Backlog Quota
+
+The user today can calculate quota used for size based limit, since there are two metrics exposed today on 
+a topic level: `pulsar_storage_backlog_quota_limit` and `pulsar_storage_backlog_size`. 
+You can just divide the two to get a percentage and know how close the topic backlog to its size limit.
+
+For the time-based limit, the only metric exposed today is the quota itself - `pulsar_storage_backlog_quota_limit_time`
+
+## Backlog Quota Eviction in the Broker
+
+The broker has a method called `BrokerService.monitorBacklogQuota()`. It is scheduled to run every x seconds,
+as defined by the configuration `backlogQuotaCheckIntervalInSeconds`. 
+This method loops over all persistent topics, and for each topic is checks whether the topic backlog exceeded
+either one of those topics. 
+
+As mentioned before, checking backlog size is a memory-only calculation, since
+each topic has the list of ledgers stored in-memory, including the size of each ledger. Same goes for the subscriptions,
+they are all stored in memory, and the `ManagedCursor` keeps track of the subscription with the oldest unacknowledged 
+message, thus retrieveing it is O(1). Checking backlog based on time is costly if configuration key
+`preciseTimeBasedBacklogQuotaCheck` was set to true. In that case, it needs to read the oldest message to obtain
+its public timestamp, which is expensive in terms of I/O. If it was set to false, it's in-memory access only, since
+it uses the age of the ledger instead of the message, and the ledgers metadata is kept in memory.
+
+For each topic which has exceeded its quota, if the policy chosen is eviction, then the process it performed
+synchronously. This process consumes I/O, as it needs read messages (using skip) to know where to stop acknowledging
+messages.
+
+
+# Motivation
+
+Users which have defined backlog quota based on time, have no means today to monitor the backlog quota usage, 
+time-wise, to know whether the topic backlog is close to its time limit or even passed it.
+
+If it has passed it, the user has no means to know if it happened, when and how many times.
+
+
+# Goals
+
+## In Scope
+- Allow the user to know the backlog quota usage for time-based quota, per topic
+- Allow the user to know how many times backlog eviction happened, and for which backlog quota type
+
+## Out of Scope
+
+None
+
+
+# High Level Design
+
+We'll use the existing backlog monitoring process running in intervals. For each topic, the subscription with 
+the oldest unacknowledged message is retrieved, to calculate the topic backlog age. At that point, we will
+cache the following for the oldest unacknowledged message:
+* Subscription name 
+* Message position
+* Message publish timestamp
+
+That cache will allow us to add a metric exposing the topic backlog age - `pulsar_storage_backlog_age_seconds`, 
+which will be both consistent (same ones used for deciding on backlog eviction) and cheap to retrieve 
+(no additional I/O involved). 
+Coupled with the existing `pulsar_storage_backlog_quota_limit_time` metric, the user can use both to divide and
+get the usage of the quota (both are in seconds units).
+
+We will add the subscription name containing the oldest unacknowledged message to the Admin API
+topic stats endpoints (`{tenant}/{namespace}/{topic}/stats` and `{tenant}/{namespace}/{topic}/partitioned-stats`),
+allowing the user a complete workflow: alert using metrics when topic backlog is about to be exceeded, then
+query topic stats for that topic to retrieve the subscription name which contains the oldest message.
+For completeness, we will also add the backlog quota limits, both age and size, and the age of oldest 
+unacknowledged message.
+
+We will add a metric allowing the user to know how many times the usage exceeded the quota, both for time or size -
+`pulsar_storage_backlog_quota_exceeded_evictions_total`, where the `quota_type` label will be either `time` or 
+`size`. Monitoring that counter over time will allow the user to know when a topic backlog exceeded its quota,
+and if backlog eviction was chosen as action, then it happened, and how many times. 
+
+Some users may want the backlog quota check to happen more frequently, and as a consequence, the backlog age 
+metric more frequently updated. They can modify `backlogQuotaCheckIntervalInSeconds` configuration key, but without
+knowing how long this check takes, it will be hard for them. Hence, we will add the metric
+`pulsar_storage_backlog_quota_check_duration_seconds` which will be of histogram type.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Public API
+Adding the following to the response of topic stats, of both `{tenant}/{namespace}/{topic}/stats` 
+and `{tenant}/{namespace}/{topic}/partitioned-stats`:
+
+* `backlogQuotaLimitSize` - the size in bytes of the topic backlog quota 
+* `backlogQuotaLimitTime` - the topic backlog age quota, in seconds. 
+* `oldestBacklogMessageAgeSeconds` - the age of the oldest unacknowledged (i.e. backlog) message, measured by 
+   the time elapsed from its published time, in seconds. This value is recorded every backlog quota check 
+   interval, hence it represents the value seen in the last check.
+* `oldestBacklogMessageSubscriptionName` - the name of the subscription containing the oldest unacknowledged message.
+  This value is recorded every backlog quota check interval, hence it represents the value seen in the last check.
+
+
+### Metrics
+
+| Name                                                           | Description                                                                                         | Attributes                                             | Units   |
+|----------------------------------------------------------------|-----------------------------------------------------------------------------------------------------|--------------------------------------------------------|---------|
+| `pulsar_storage_backlog_age_seconds`                           | Gauge. The age of the oldest unacknowledged message (backlog)                                       | cluster, namespace, topic                              | seconds |
+| `pulsar_storage_backlog_quota_exceeded_evictions_total`        | Counter. The number of times a backlog was evicted since it has exceeded its quota                  | cluster, namespace, topic, quota_type = (time \| size) |         | 
+| `pulsar_storage_backlog_quota_check_duration_seconds`          | Histogram. The duration of the backlog quota check process.                                         | cluster                                                | seconds | 
+| `pulsar_broker_storage_backlog_quota_exceeded_evictions_total` | Counter. The number of times a backlog was evicted since it has exceeded its quota, in broker level | cluster, quota_type = (time \| size)                   |         | 
+
+* Since `pulsar_storage_backlog_age_seconds` can not be aggregated, with proper meaning, to a namespace-level, it will
+  not be included as a metric when configuration key `exposeTopicLevelMetricsInPrometheus` is set to false.
+* `pulsar_storage_backlog_quota_exceeded_evictions_total` will be included as a metric also in namespace aggregation.
+
+# Alternatives
+
+One alternative is to separate the backlog quota check into 2 separate processes, running in their own frequency:
+1. Check backlog quota exceeded for all persistent topics. The result will be marked in memory.
+   If precise time backlog quota was configured then this will the I/O cost as described before.
+2. Evict messages for those topics marked.
+
+This *may* enable more frequent updates to the backlog age metric making it more fresh, but the cost associated with it
+might be high, since it might result in more frequent I/O calls, especially with many topics. 
+Another disadvantage is that it makes the backlog check and eviction more complex.
+
+# Links
+
+* Mailing List discussion thread: https://lists.apache.org/thread/xv33xjjzc3t2n06ynz2gmcd4s06ckrqh
+* Mailing List voting thread: https://lists.apache.org/thread/x2ypnft3x5jdyyxbwgvzxgcw20o44vps