Roughly 11 months ago, I started working on solving the biggest issue with Pulsar metrics: the lack of ability to monitor a pulsar broker with a large topic count: 10k, 100k, and future support of 1M. This started by mapping the existing functionality and then enumerating all the problems I saw (all documented in this doc).
This PIP is a parent PIP. It aims to gradually solve (using sub-PIPs) all the current metric system's problems and provide the ability to monitor a broker with a large topic count, which is currently lacking. As a parent PIP, it will describe each problem and its solution at a high level, leaving fine-grained details to the sub-PIPs. The parent PIP ensures all solutions align and does not contradict each other.
The basic building block to solve the monitoring ability of large topic count is aggregating internally (to topic groups) and adding fine-grained filtering. We could have shoe-horned it into the existing metric system, but we thought adding that to a system already ingrained with many problems would be wrong and hard to do gradually, as so many things will break. This is why the second-biggest design decision presented here is consolidating all existing metric libraries into a single one - OpenTelemetry. The parent PIP will explain why OpenTelemetry was chosen out of existing solutions and why it far exceeds all other options. I’ve been working closely with the OpenTelemetry community in the past eight months: brainstorming this integration, and raising issues, in an afford to remove serious blockers to making this migration successful.
I made every effort to summarize this document so that it can be concise yet clear. I understand it is an effort to read it and, more so, provide meaningful feedback on such a large document; hence I’m very grateful for each individual who does so.
I think this design will help improve the user experience immensely, so it is worth the time spent reading it.
Working with Metrics today as a user or a developer is hard and has many severe issues.
From the user perspective:
From the developer perspective:
This proposal offers several key changes to solve that:
Aggregation and Filtering combined solves the cardinality without sacrificing the level of detail when needed and most importantly, you determine which topic/group/namespace it happens on.
Since this change is so invasive, it requires a single metrics library to implement all of it on top of; Hence the third big change point is consolidating all four ways to define and record metrics to a single one, a new one: OpenTelemetry Metrics (Java SDK, and also Python and Go for the Pulsar Function runners). Introducing OpenTelemetry (OTel) solves also the biggest pain point from the developer perspective, since it's a superb metrics library offering everything you need, and there is going to be a single way - only it. Also, it solves the robustness for Plugin authors which will use OpenTelemetry. It so happens that it also solves all the numerous problems described in the doc itself.
The solution will be introduced as another layer with feature toggles, so you can work with existing system, and/or OTel, until gradually deprecating existing system. Pulsar OTel Metrics will support exporting as Prometheus HTTP endpoint (/metrics but different port) for backward compatability and also OTLP, so you can push the metrics to OTel Collector and from there ship it to any destination.
It's a big breaking change for Pulsar users on many fronts: names, semantics, configuration. Read at the end of this doc to learn exactly what will change for the user (in high level).
In my opinion, it will make Pulsar user experience so much better, they will want to migrate to it, despite the breaking change.
This was a very short summary. You are most welcomed to read the full design document below and express feedback, so we can make it better.
Any software we know in the world today, exposes metrics to show what transpires in it in an aggregated way. A metric is defined as:
pulsar_rate_incluster=europe-cluster, topic=orders1679403600, which means Tuesday, March 21, 2023 1:00:00 PMpulsar_rate_in it means 15,345 bytes received in the interval (e.g., 1min)Composing it all together looks like this:
pulsar_rate_in {cluster="europe-cluster", topic="orders"} 1679403600 15345
Metrics usually come in all kinds of types:
As opposed to logs which tell you a step-by-step story, metrics give you an aggregated view of Pulsar behavior over time.
Pulsar main feature is messaging: receiving messages from producers, and dispatching messages to consumers via subscriptions.
The metrics related to messaging are divided into a couple of aggregation levels: broker, namespace, topic, subscription, consumer and producer.
Some levels can have high cardinality, hence Pulsar offers several configuration to minimize the amount of unique time series exported, by the following toggles:
exposeTopicLevelMetricsInPrometheus - Settings this value to false will cause metrics to be reported in namespace level granularity only, while if true, the metrics will be reported in topic level granularity.exposeConsumerMetricsInPrometheus - Setting this value to false will filter out any consumer level metric, i.e. filtering out pulsar_consumer_* metrics.exposeProducerLevelMetricsInPrometheus - Setting this value to false will filter out any producer level metric, i.e. filtering out pulsar_producer_* metrics.cluster=eu-cluster, topic=ordersThe current metric system has several problems which act as the motivation for this PIP. Each subsection below explains the background to the problem and the actual problem. No prior knowledge is required.
In Pulsar there are multiple ways to define and record metrics - i.e. several metric libraries:
StatsBucketsRate.Summary - An extension for Prometheus Client library providing a more performant version of Summary.OrderedExecutors.PulsarZooKeeperClient and several Pulsar plugins are the most prominent examplesLongAdder to act as Counters, AtomicLong or primitive long with atomic updater to act as Gauge.Having multiple metric libraries is a problem for several reasons:
Summary by Pulsar Metrics library uses a fixed time window (1 min) to reset and start accumulating metrics, while Pulsar Client summary uses a moving time window of 10 minutes.StatsBucket by Pulsar Metrics library resets its bucket counters every interval (1 min) while Prometheus Client Histogram does not.I would summarize it with one word: confusion.
Pulsar, as users know, is unique by allowing you to use a very high number of topics in a cluster - up to 1M. It’s not uncommon to find a broker with 10k up to 100k topics hosted on it.
For each topic, Pulsar exposes roughly 80 - 100 unique time series (metrics). A single broker with 100k topics will emit 10M unique time series (UTS). This usually results in the following for the user:
Hence, the common user behavior is:
stats Admin API, to filter only the metrics they need, and aggregate to a level with reasonable cardinality.The filtering supported today is toggle-based (all or nothing) for certain levels, hence very coarse. You can toggle between namespace to topic level. If you chose topic level, you can toggle consumers and producers level metrics separately.
The aggregations provided today are:
Summary, as explain in the Background section, is used to provide quantile values like p95, p75, for certain measurements - mostly latencies, but sometimes sizes. It is widely used in Pulsar: Ledger Offloader, Resource Groups, Replicated Subscriptions, Schema Registry, Broker Operability, Transaction Managements, Pulsar Functions and Topics, just to name a few.
The biggest problem with Summaries is that quantiles are not aggregatable mathematically. You can’t know what is the p95 of schema registry “get” operation latency across the cluster by knowing each p95 per broker. Any math you’ll do on it will result in large numerical error. This means that beyond the scope of a broker or label set (i.e. topic/partition) it’s unusable.
For example, the user is mostly interested in the topic publish latency, and not the topic-partition publish latency. Without aggregating, it’s impossible to know the publish latency for a topic with partitions, because we need to aggregate the publish latency of each topic-partition, but we can’t aggregate summaries as explained above.
This is not the case for Explicit Bucket Histograms or the new type called Exponential Bucket histogram. They are aggregatable and produce quantiles (extrapolating) with a small margin of error.
Most metrics framework offers you objects you create and then register to a registry. The exporting is taken care of for you. In Prometheus for example, you create a Counter, and register it to static collector registry. The exporter simply exports all collectors (the counter being one) registered to the static registry.
In Pulsar, since it has 4 different libraries you can use to define a metric, the exporter had to be written by Pulsar, to patch all of them together into a single response to the GET/metrics request.
If you have used Prometheus Client, you’re all set, as that integration was written for you. The problem is most usages are not that library, since it has serious performance issues, especially on high cardinality metrics (like topics).
Using all other libraries, you’re basically required to write a function that has the following signature: void writeMetrics(SimpleTextOutputStream stream) for each class containing metrics. Then you add a call to that function in PrometheusMetricsGenerator. The argument stream is basically a byte-array you’re writing bytes into, which represents the response body for /metrics that is about to be delivered. You need to be aware of that, and write the current state of each metric you have in your class, in Prometheus Exposition format.
This presents multiple problems:
PrometheusMetricStreams class), which holds a Stream per metric name.FunctionMetricsResource writing their own Writer using Heap-based ByteBuf instead of direct memory)cluster label must be added manually by any function.Plugins have their own metrics. Most plugins were written to run inside Pulsar (you supply JARs or NARs loaded on Broker initialization). Pulsar doesn't provide a single interface through which you create metric objects and register them and integrate with Pulsar metrics reporting. Due to that, the following happens:
PrometheusRawMetricsProvider which contains a single method:void generate(SimpleTextOutputStream stream). This basically means they need to implement this function, so it will read the metrics from the framework they chose, and write it in Prometheus exposition format, in bytes.Due to that, most plugin developers are forced to write their metric exporting logic on their own, causing more work to be done for them.
Since the interface is very low level, it creates several difficulties going forward:
Due to multiple libraries and lack of high level metrics interface for plugins, it’s basically impossible to add another export format in a performant manner suitable for latency sensitive system such as Pulsar. The metrics system today is coupled to Prometheus format, thus prevents any addition of a new, better format.
Take for example, OTLP, a new protocol for Traces/Logs/Metrics. It’s more optimized than Prometheus, since for example, for histograms, it mentions the attributes once for all buckets, rather than repeating it for each bucket like Prometheus format.
OTLP can’t be added to Pulsar as another exporting mechanism, in current metric system.
The way Rate is built forces the developer to both:
SimpleTextOutputStream (each class has such a method)reset() of the Rateinstance, to a function which runs periodically.Both are not something a developer can understand on its own, therefor easy to forget to do, or even call twice by mistake. Also wastes time to learn how to do it.
Each metric, in Prometheus format should contain a line starting with #HELP which allows Prometheus to parse that line and add a description to this metric, which is later used by UIs like Grafana to be better explain the available metrics.
Since there are 4 metric libraries, only Prometheus Client offers the typed option of including a description. Most metrics are not using it, hence lack a descent help line.
The main histogram used in Pulsar is StatsBucket. It has two major problems:
Bucket counters in it are reset to 0 every 1 min (configurable). This goes against Prometheus assumption that bucket counters only increase. As such, it prevents using Prometheus functions on it like calculating quantiles (histogram_qunatile), which is the main reason to use histograms.
When exported, the bucket label is encoded in the metric name, and not as le label as Prometheus expects, hence makes it impossible to use histogram_quantile and calculate quantiles on it.
For example: pulsar_storage_write_latency_le_10 Should have been pulsar_storage_write_latency{le=”10”}
Rate, StatsBucket, Summary, some exported JVM metrics and Pulsar Function metrics are reset to 0 every 1 min (configurable). This means that if from some reason, Prometheus or any other agent, fails to scrape the metrics for several minutes, you lost the visibility to Pulsar during those minutes. When using counters / histograms which are only incremented, the rate is calculated as delta on the counter values hence if two measurements 5 minutes apart, will still give you a descent average on that period. Same goes for histogram quantile calculation.
Most usage of Prometheus Client library is done using the static Collector registry. This exposes it to flaky behavior across tests, as static variables are shared across tests, and not cleaned between them. When using non-static registry, it inherently resets itself every new test, but this is not the case here.
A Pulsar Function author has a single way to add metrics to their function, using method ``void recordMetric(String metricName, double value)onBaseContext` interface.
What it does, is record this value in a Prometheus Summary named pulsar_function_user_metric_, under the label metric={metricName}.
This Summary metric is also being reset every 1min by the wrapper code running the user function.
It has the following problems:
There is a configuration specifying if a Partitioned Topic, composed of several partitions each is a Topic, will be printed using topic label only (i.e. topic=incoming-logs-partition-2) or split into topic and partition (i.e. topic=incoming-logs, partition=2).
The problem is that this configuration is only applied to messaging related metrics (namespace, topic, producer, consumer, subscription). It is not applied to any other metric which contains the topic label, such as Transactions, Ledger Offloader, etc. This creates inconsistency in reported metrics.
Pulsar Functions are launched by instances (processes) of Pulsar Function Worker. It supports 3 types of runtimes (function launchers):
The metrics of the wrapper code, which also includes the function custom metrics (metrics the function authors adds), are exposed on /metrics endpoint by the wrapper code. In the case of Kubernetes, the pod is annotated such that Prometheus operators will scrape those metrics directly. In the case of Thread runtime, the metrics are integrated into the Function Worker metrics. In the case or Process runtime, the Function Worker is the one scraping the /metrics from each function process, concatenating them and add it to Function Worker /metrics.
Process runtime also includes many JVM and system level metrics registered using Prometheus Client built-in exporters. Due to this (not Pulsar code) it doesn't contain any special label identifying this function.
When the Function Worker scrapes each /metric endpoint, it simply concat the response, and since no unique label exists, the metrics override each other.
For example, if a Function Worker launched 3 processes, one for each function, then each will contain jvm_memory_bytes_used{area="heap"} 2000000, with different numeric value. In it there is no unique label. When concatenating the response from the three functions processes, without any process/function we will not know from each process this arrived from, and they will override each other.
pulsar_txn for transactions related metrics or pulsar_schema for Pulsar Schema metrics. Some don’t, like metrics related to messaging (topic metrics) - for example pulsar_bytes_in or pulsar_entry_size_le_*.brk_ while others start with pulsar_. Some are even replaced from brk_ to pulsar_ during metric export.This makes it very hard:
pulsar_* will catch all other pulsar’s metrics.pulsar_ledgeroffloader_but it’s impossible for metrics such as messaging metrics which doesn't really have a prefix.We will introduce a new metrics library that will be the only metrics library used once this PIP implementation reaches its final phase. The chosen library is OpenTelemetry Java SDK. Full details on what is OpenTelemetry and why it was chosen over other metric libraries is located at the Detailed Design section below. This section focuses on describing it in high level.
OpenTelemetry is a project that provides several components: API, SDK, Collector and protocol. It’s main purpose is defining a standard way and reference implementation to define, export and manipulate observability signals: metrics, logs and traces. In this explanation we’ll focus on the metrics signal. The API is a set of interfaces used to define and report metrics. The SDK is the implementation of that API and also added functionality such as export of metrics and ability to change how certain metrics are collected and measured. The protocol is called OTLP, and it's the most efficient protocol to transmit metrics, logs and traces to its final destination (be it a database or vendor). The Collector is a light-weight process that allows receiving/pulling metrics/logs/traces in various formats, transform them (i.e. filter, aggregate, maintain state, etc.), and export them to many destinations (many vendors and databases).
The project’s API also has a part for reporting logs and traces. One its core features is the ability to share common context that can be used across metrics, logs and traces reporting (think Thread Local contains an attribute which is also used when reporting metric’s baggage, logs and traces, hence creates a link between them).
In this PIP scope we will only use the Metrics API and SDK. Specifically we will use the Java SDK in Pulsar Broker, Proxy and Function Worker, but also the Go and Python SDK for the wrapper code which executes functions written in Go and Python.
We will keep the current metric system as is, and add a new layer of metrics using OpenTelemetry Java SDK: All of Pulsar’s metrics will be created also using OpenTelemetry. A feature flag will allow enabling OpenTelemetry metrics (init, recording and exporting). All the features and changes described here will be done only in the OpenTelemetry layer, allowing to keep the old version working until you’re ready to switch using the OTel (OpenTelemetry) implementation. In the far future, once OTel usage has stabilized and became widely adopted we’ll deprecate current metric system and eventually remove it. We will also make sure there is feature flag to turn off current Prometheus based metric system. There‘s no need to have an abstraction layer on both the current metric system and the new one (OTel) as OpenTelemetry API is the abstraction layer, and it’s the industry standard.
One very big breaking change (there are several described in the High Level Design, and also summarized in the Backward Compatibility section) is the naming. We are changing all metric names due to several reasons:
le attribute (when exported to Prometheus format) and not inside the metric name.The move to OpenTelemetry is not without cost, as we both need to improve the library and adjust the way we use metrics in Pulsar with it. The Detailed Design contains a section detailing exactly what we need to improve in OpenTelemetry to make sure it fits Apache Pulsar. It mostly consists of making it performant (almost allocation free), and small additions like removing attribute sets limit per instrument.
The changes we need to make in Pulsar to use it are detailed in this high level design section below. It’s composed of two big changes: changing all histograms to be only at namespace level instead of topic level, and switching from Summary to Histogram (Explicit Bucket).
There will be a sub-PIP detailed exactly how OpenTelemetry will be integrated into Pulsar, detailing how users will be able to configure it: their own views, exporters, etc. We need to see how to support that, given we want to introduce our own filtering predicate.
Here's a very short idea of how the code will look like using OpenTelemetry. In reality, we will use batch callback and different design. The sub-PIP will specify the exact details.
class TopicInstruments { // ... meter.counterBuilder("pulsar.messaging.topic.messages.received.size") .setUnit("bytes") .buildWithCallback(observableLongMeasurement -> { for (topic : getTopics()) { val size = topic.getMessagesReceivedSize(); observableLongMeasurement.record(size, topic.getAttributes()); } }); // ... } class PersistentTopic { // ... Attributes topicAttributes LongAdder messagesReceivedSize // ... init(String topicName) { Attributes topicAttributes = Attributes.builder() .put("topic", topicName) .put("namespace", namespaceName) .build(); } // ... messageReceived() { // ... messagesReceivedSize.add(msgSize); } getMessagesReceivedSize() { return messagesReceivedSize.value(); } }
When you have cardinality issues, specifically topics being the root cause since Pulsar support up to 1M topics cluster wide, roughly translated to 100k topics per single broker. The best way to solve it is to reduce the cardinality.
We will introduce a new term called Topic Metric Group. Each topic will be mapped to a single group. The groups are the tool we’ll use to reduce cardinality to a descent level, which roughly means 10 - 10k. A cardinality scale your time series database can handle properly.
For each metric Pulsar currently have in topic-level (topic is one of its attributes), we will create another metric, bearing a different name which will have its attributes in the group level (topicMetricGroup will be the attribute, without topic attribute). Each time we’ll record a value in a topic metric, we’ll also record it in the equivalent Topic Metric Group metric. For consistency, we will do the same also for namespace level.
Example:
If today you chose topic level metric, which means you don’t have namespace level metrics, you’ll have the following metric:
pulsar_in_messages_total{topic="orders_company_foobar", namespace="finance", ...}
After the change (including naming changes discussed later), you’ll have:
pulsar_messaging_**namespace**_in_messages_total{namespace="finance"} pulsar_messaging_**group**_in_messages_total{topicMetricGroup="orders", namespace="finance"} pulsar_messaging_**topic**_in_messages_total{topic="orders_company_foobar" topicMetricGroup="orders", namespace="finance"}
There will be a dynamic configuration allowing to specify the mapping from topic to topic group, in a convenient way. Since there can be many ways to configure that, we’ll define a plugin interface which will provide that mapping given a topic, and we’ll provide a default implementation for it. This will allow advanced users to customize to their needs.
The detailed design section contains a more detailed description of this mapping configuration. There will be a sub-PIP detailing how the mapping will be configured, which tools we need to add to allow knowing which topics are in each group, and more.
The aggregation above using both namespace and Topic Metric Group, reduces the cardinality. The only thing left for us to do is add a great user experience tool to control which metrics you’d like to see exported, in fine granularity. A bit like logging levels, but with more fine-grained controls.
We would like to have a dynamic configuration allowing us to toggle which metric we wish to keep or drop. For example:
We will introduce a new configuration, containing Filtering Rules, evaluated in order. Each rule will define a selector allowing you to select (metric, attributes) pairs based on all sorts of criteria, and then defining actions such as drop all, keep all, keep only, drop only, determining for each (metric, attribute) pair if it should be filtered out or not.
This configuration will be dynamic, hence allowing you to build namespace / group level dashboards to monitor Pulsar. Once you see a group misbehaving, you can dynamically “open” the filter to allow this group’s topic metric, and you can decide to only allow the certain metric you suspect is problematic. Once you find out the topic, you allow (stop dropping) all of its metrics, to enable you to debug it. Once you’re done, you can roll back to group level metrics.
We want to allow users to customize and build a filter matching their needs, hence we’ll create a plugin interface, which can decide for a given metric (name, attributes, unit, etc.) if they want to filter it or not.
The detailed design section will include a more detailed description of the plugin interface and default implementation configuration file. It will also include explanation how we plan to implement that in stages: 1st stage on our own and 2nd stage by introducing push down predicate into OpenTelemetry MetricReader.
The fine-grained filtering mechanism described makes using the configuration toggles we have today deprecated hence we will remove them. This includes exposeConsumerLevelMetricsInPrometheus, exposeTopicLevelMetricsInPrometheus and exposeProducerLevelMetricsInPrometheus.
The aggregation to groups and namespaces which are of reasonable cardinality, coupled with the ability to decide on which metrics and specific attribute sets in them, you wish to export using the filters, solves the cardinality issue. Visibility to monitoring data is not sacrificed since dynamic configuration allows you to zoom in to get the finer details and later shut it off.
Histograms are primarily used to record latencies: pulsar_storage_write_latency_le_*, pulsar_storage_ledger_write_latency_le_*, and pulsar_compaction_latency_*
We have several issues with them:
The other problem we have is the topic move. Topics can move between brokers, due to automatic load balancing, or an operator decision to unload a topic. Today, when a topic is unloaded, the broker stops reporting the metrics for it.
In OTel, the API doesn't support remove() for an attribute set on an instrument (counter, histogram, etc.). Normally, if it was supported, we could have called remove for each instrument which contains an attribute-set for that topic.
OTel has two categories of instruments: synchronous and asynchronous. Synchronous instruments, keeps the value of the object in memory and updates it upon recording a new value (behind the scenes: it adds for example +2 to an AtomicLong it maintains for a Counter and the attribute set). Asynchronous instruments on the other hand are defined using a callback. When a call is made to collect the values for each instrument, a callback is invoked to retrieve the (attributes, value) pairs. when you record a value to an attribute set in a synchronous instrument, it will remain in memory forever until restart. Async instruments on the other hand, retrieve the (attributes, value) pairs upon collect, and forget them afterward.
Since remove() doesn't exist for instruments, async instruments are the closest thing we have in OTel. Thus Counter and UpDownCounters for topic instruments can be asynchronous instruments, hence when a topic is unloaded, the next callback will simply not record their values.
Histograms are problematic in that sense, since they yet to have an asynchronous version - alas they are only synchronous. Hence, if we use them for topic instruments, used attributes can never be cleared from memory for them thus we’ll have an “attributes” (topic) leak, as over time it will only grow. This is why currently we can’t use them for topics or topic groups.
We have opened an issue and making progress, but this is a long process.
Coupling the two together - cost, confusion, lack of clear use, and lack of support from OTel - we will change those topic level histograms to be namespace / broker level.
As explained before, instruments don’t support remove(). Topic, its subscriptions, producers and consumers - each has its own set of metrics. Also, each is ephemeral. Topics can move or be deleted, producers and consumers can stop and start many times, and subscriptions move with topics.
Hence, for topic, producers, consumer and subscription (and topic group) we will use asynchronous instruments. This means we’ll keep the state using our own LongAdder, AtomicLong or primitive long using atomic updater. When creating the asynchronous instrument, the callback will retrieve the value from the variable. For example, when a producer is removed, in the next collection cycle, the callback will not report metrics for it since it doesn't exist, and thus it will disappear from memory of OTel as well.
OTel has a special batch callback mechanism, allowing you to supply a single callback for multiple instruments, making it more efficient, and we plan to use it.
As explained before, we’ll have instruments per aggregation level: broker, namespace, topic group, topic, subscription, producer and consumer.
Broker and namespace level will use synchronous instruments, since the amount of namespaces is not expected to have high cardinality, hence not removing them is not a big attribute leak issue. Asynchronous instruments are also needed es explain in “Supporting Admin Rest API Statistics endpoints” section below.
Summary by design can’t be aggregated across topics or hosts (See Background and Motivation). That was the reason OTel doesn‘t support them. I opened an issue for that, but it doesn’t seem like something that can be added to OTel.
Another consideration is CPU. In benchmarks done, it seems that updating the summaries that are based on Apache Data Sketches cost 5% of CPU time, which is a lot compared with a simple +1 to a counter in a bucket in an explicit bucket histogram.
Due to those reasons, it makes sense to switch all summaries to histograms (explicit bucket type).
From the same reasons as histograms, they will be broker/namespace level only (not topic).
Most summaries are used for latency reporting, from same reasons of multi tenancy and careful inspection of existing summaries, we’ve concluded we can convert them to be namespace level. The domains affected by it are:
The complete list of which summaries are affected is at the Detailed Design section
Pulsar Functions metrics uses summary for user defined metrics, but we assume the quantiles are actually meaningless since some use it to record the value “1” just to obtain count, and some record value to obtain a sum. We will convert them to Histogram without buckets, just providing sum and count.
metric={user-defined-name}We’ve inspected our summaries and histograms, and it seems the bucket ranges are common per unit: ms, seconds and bytes.
OTel has the notion of views. They are created upon init of OTel and can be used to specify the buckets for a set of instruments based on instrument selector (name wildcard, …). We have opened an issue for OTel SDK Specifications, which was merged to allow specifying a unit in an instrument selector. We only need to implement it in OTel Java SDK (See issue I’ve opened).
As described in the motivation and background section, Pulsar reset certain type of metrics every configurable interval (i.e. 1min) : rates, explicit bucket histograms and summaries.
As also explained, it makes hard to use for histograms, and redundant and less accurate for rates. Most time-series databases can easily calculate rates based on ever-increasing counters.
Hence, in OpenTelemetry we won’t report rates, we’ll switch to counters. That means of course name changing, but as described in this document, all names will be changed anyhow.
For histograms, we will simply never reset them.
Summaries are converted to histograms so also never reset anymore.
We will keep Rates around primarily for the statistics returned through Admin API (i.e. Topic Stats, …), but they will never be exposed through OpenTelemetry.
It is worth noting that OpenTelemetry supports the notion of Aggregation Temporality. In short, it allows you to define for a given instruments if you want it to be exported as delta or cumulative. Some exporters like Prometheus only support Cumulative, and will override it all to be such. OTLP supports delta. Currently, we’re not explicitly supporting configuring views / readers to allow that, but it’s something very easily added in the future, by the community. It will be perfect for people using OpenSearch or Elasticsearch.
Today a topic is reported using the attribute topic={topicName}. If the topic is actually a partition for a partitioned topic, it will look like topic={partitionedTopicName}-partition-{partitionNum}. There is a configuration name splitTopicAndPartitionLabelInPrometheus which makes that be reported instead as topic={partitionedTopicName}, partition={partitionNum}.
This is not consistent, in such that not all metrics using topic attribute did the split accordingly.
In OTel metrics we will ignore that flag and always report it as topic={partitionedTopicName}, partition={partitionNum}. We will make it consistent across any topic attribute usage. Eventually this flag will be removed (probably in the next major version of Pulsar).
OTel has a built-in Prometheus MetricReader (and exporter) which exposes /metrics endpoint, which we will use.
Pulsar current metric system has a caching mechanism for /metrics responses. This was developed since creating the response for high topic count was a CPU hog and in some cases memory hog. In our case we plan to use Filtering and Aggregation (Metric Topic Group) to drive the response to a reasonable size, hence won’t need to implement that in OTel metrics.
OTel also has built-in OTLP exporter. OTLP is the efficient protocol OTel has, which OTel Collector supports, and some vendors as well. We wish to use it, yet it seems that it is very heavy on memory allocation. Hence, we will need to improve it to make it allocation free as much as possible.
OTel supports a static (Global) OTel instance, but we will refrain from it to make sure test data doesn't leak between tests.
In OTel we will create an instance of MeterProvider during Pulsar init. This object is the factory for Meter which by itself is a factory for instruments (Counter, Histogram, etc.). We will pass along a Pulsar Meter instance to each class that needs to create instruments. The exact details will be detailed in a sub-PIP of adding OpenTelemetry to Pulsar.
OTel doesn't force you to supply documentation.
We will create a static code analysis rule failing the build if it finds instrument creation without description.
We will optionally try to somehow create an automated way to export all metrics metadata - instrument name, documentation, in an easy-to-read format to be used for documentation purposes on Pulsar website.
BookKeeper (client and server) has their own custom metrics library. It’s built upon a set of interfaces. Pulsar metric system has an implementation for it.
We will create another implementation to patch it to OTel.
We will modify all popular plugin interfaces Pulsar has, such that they will accept an OpenTelemetry instance to be used to grab the MeterProvider instance and use it to create their own Meter with the plugin name and version which they will use to create their own instruments.
A user which has decided to turn on OTel metrics will have to verify all Pulsar plugins it uses have been upgraded to use the modified interface, otherwise their metrics will not be exported.
Plugin authors will need to release a new version of their plugin which implements the new interface and registers the metrics using OpenTelemetry.
One big advantage is that using OTel supports plugins running in stand-alone mode. Some plugins have the option to run some of their code outside Pulsar. By using OTel API, they can integrate either via their own SDK or Pulsar SDK (via OpenTelemetry instance).
A detailed list is at the Detailed Design.
A sub-PIP will be dedicated to integrating with plugins.
Pulsar has several REST API endpoints for retrieving detailed metrics. They are exposing rates, up-down counters and counters.
OTel instruments doesn't have methods to retrieve the current value. The only facility exposing that is the MetricReader that reads the entire metric set. Thus, any metric that is exposed also through Admin REST API will have to have its state maintained by Pulsar, either using LongAdder, AtomicLong or primitive long with atomic updater. The matching OTel instrument will be defined as Asynchronous instrument, meaning it is defined by supplying a callback function will be executed to retrieve the instrument current value. The callback will simply retrieve the state from the matching LongAdder, AtomicLong or primitive long (or double to that end) and return it as the current value.
We will change Rate, in such a way that won’t require the user creating it, calling reset() periodically and manually. All rates will be created via a manager class of some sort, and it will be the one responsible for scheduling resets. Rates will always expose a sum counter and a counter to save up on multiple variables. A sub-PIP will explain in detail how this will be achieved.
Pulsar supports the notion of Pulsar Functions. It is user-supplied functions, either in Go, Java or Python, which can read messages, and write messages. You can use it to read all messages from a topic and write them to external system like S3 (Sink), or the other way around: read from an external system and write it to a topic (e.g. DB Change log to Pulsar topic - source). The other option is simply transforming the message received and writing it to another topic.
A user can submit a function, and can also configure the amount of instances it will have.
The code responsible for coordinating the execution of those functions is located in a component called Function Worker, which can be run as stand-alone process or as part of Pulsar process. You can run many Function Workers, yet only one function as the leader.
The leader takes care of splitting the work of executing the function instances between the different Function Workers (for load balancing purposes).
The Function worker has three runtimes, as in, three options to execute the functions it is in charge of:
Each Function Worker has its own metrics it exposes.
Each Function has its own metrics it exposes.
In the Thread runtime, all metrics are funneled into Pulsar metrics (exposing a method which writes them into the SimpleTextOutputFormat).
In the Process runtime, the function is executed in its own process, hence there is a wrapper main() function executing the user supplied function in the process. This main() function has general function execution metrics (e.g. how many messages received, etc.), and also the function metrics (user custom metrics). All the metrics are defined using a single library: Prometheus client. The metrics are exposed using the client’s HTTP server exposing /metrics endpoint exposing the two categories of metrics.
In the Kubernetes runtime, the pod is annotated with Prometheus Operator annotation, and it is expected it will be installed, hence Prometheus will scrape those metrics directly from the process running in the pod.
In the Process runtime, the Function Worker is iterating over all processes it launched, and for each it issues a GET request to /metrics. The responses are concatenated together and printed to SimpleTextOutputStream.
The general function execution metrics comes in two forms: cumulative and 1 min ones. The latter names ends with _1min, and they get reset every 1min. (e.g. *_received_1min).
Each process launched also launches a gRPC server supporting commands. Two of those are related to metrics: resetMetrics and getAndResetMetrics. They reset all metrics - both the general framework ones and the customer user ones.
Prometheus client was also configured to emit several metrics using built in exporters: memory pool, JMX metrics, etc.
In phase 1, we’ll keep the reporting of user defined metrics for function authors as is, and mainly focus on the other issues which are: metrics scraping for each runtime, and the 1min metrics. In phase 2, we’ll also add the option to define metrics for pulsar function authors via OTel.
Thread Runtime
The framework will use Pulsar OTel SDK, or it’s standalone function worker SDK, thus what ever export method it uses, it will use as well (prometheus, OTLP, …)
Kubernetes Runtime
OTel supports exporting metrics via /metrics endpoint using Prometheus format. We’ll support the same as it is today done with Prometheus client.
We’ll also support configuring the pod, so it can send metrics via OTLP to defined destination.
Process Runtime
The existing scraping of /metrics solution was not good:
OTel supports both exporting metrics as Prometheus and pushing them using OTLP. Making the Function worker pull the metrics, in effect - to be the hub - is super complicated. It’s much easier to simply let the processes be scraped or push OTLP metrics - configured the same as Pulsar is.
At phase 1 we won’t support Process Runtime.
At phase 2, we can use Prometheus HTTP Service Discovery, and expose such an endpoint in the Function Worker leader. Via health pings it gets from each worker, they can also report each process metrics port, thus allowing prometheus to scrape the metrics directly from each process. We’ll garner feedback from the community to see how important is Process runtime, as we have K8s runtime which is much more robust.
We’ll not define any 1min metric. Any TSDB can calculate that from the cumulative counter.
getMetrics RPCWe can define our own MetricReader which we can then filter to return the same metrics as we return today: general function metrics and user-defined metrics.
resetMetrics RPCOTel doesn't support metric reset, and it also violates Prometheus since it expects metrics to be cumulative. Thus, we will remove that method.
OTel has an SDK for Python and Go, thus we’ll use it to export metrics.
A sub-pip will be created for Function Metrics which will include detailed design for it.
As mentioned in the high level design section, we’ll have a plugin interface, allowing to have multiple implementations and to customize the way a topic is mapped to a group.
The default implementation this PIP will provide will be rule based, and described below.
We’ll have a configuration, that can look something like this:
bi-data // group name namespace = bi-ns // condition of the form: attribute name = expression topic = bi-* // condition of the form: attribute name = expression incoming-logs namespace = * topic = incoming-logs-*
The configuration will contain a list of rules. Each rule begins with a group name, and list of matchers: one for namespace and one for topic. The rules will be evaluated in order, and once a topic was matched, we’ll stop iterating the rules.
There will be a sub-PIP detailing the plugin and default implementation in fine-grained detail: where it will be stored, how it will support changing it dynamically, performance, etc.
These are the list of plugins Pulsar currently uses
AdditionalServletAdditionalServletWithPulsarService - we can use PulsarService to integrateEntryFilter - need to add init method to supply OTelDelayedDeliveryTrackerFactory - accepts PulsarServiceTopicFactory - accepts BrokerServiceResourceUsageTransportManager - need to add init methodAuthenticationProvider - need to add parameter to initialize methodModularLoadManager - accepts PulsarServiceSchemaStorageFactory - accepts PulsarServiceJvmGCMetricsLogger - need to add init().TransactionMetadataStoreProvider - need to add initTransactionBufferProvider - need to add initTransactionPendingAckStoreProvider - need to add initLedgerOffloader - need to add init()AuthorizationProvider - need to add parameter to initialize methodWorkerService - need to modify init methodsPackageStorageProviderProtocolHandler - need to modify init methodBrokerInterceptor - accepts PulsarServiceBrokerEntryMetadataInterceptorIn a sub-PIP we will consider how to update those interfaces effectively allowing to pass OpenTelemetry instance, so they can create their own Meter or have access to an auxiliary class and supply only a Pulsar meter. Note that each a Meter has its own name and version and those are emitted as two additional attributes - e.g. {..., otel_scope_name="s3_offloader_plugin" otel_scope_version="1.2", ...}, to avoid any metric name collision.
It’s the new emerging industry-wide standard for observability and specific metrics, as opposed to just a library or a standard adopted and promoted by a single entity/company.
It’s much more sophisticated than the other libraries
OpenTelemetry interface for reporting traces/metrics/logs, the integration of it will not require any special development efforts.Below I will list the libraries I found and why I don’t think they are suitable.
sl4fj-api which is used by 60k artifacts—as such, picking it as the standard for today, seems like “betting” on the wrong project.instrument = map(attributes→values). In Micrometer, it’s designed in a way that each (instrument, attributes) is a metric on its own. Less elegant and more confusing.MetricReader interface and the MetricsExporter interfaces were designed to receive the metrics collected from memory by the SDK using a list, and for each collection cycle, an allocation of at least 70M objects (Metric data points).MetricsProducerNice to have
Issues completed while writing the design
We have two ways to do that:
As mentioned in the high level design, we will define an interface allowing to have multiple built-in and also custom implementations of filtering. The interface will determine per each metric data point if it will be filtered or not. The data point is composed of: instrument name, attributes, unit, type.
Our default implementation will be ruled based, and will have the following configuration. I used here HOCON to make it less verbose. Exact syntax will be determined in sub-PIP. The configuration will be dynamic, allowing operators to change it in runtime. The exact mechanisms will be detailed in the sub-PIP.
rules { // All instruments starting with pulsar_, with a topic attribute // will be dropped by default. This will keep only topicMetricGroup and namespace level. default { instrumentSelect = "pulsar_*" attrSelect { topic = "*" } filterInstruments { dropAll = true } } // single topic, highest granularity bi-data { instrumentSelect = "pulsar_*" attrSelect { topicGroup = "bi-data" } filterInstruments { keepAll = true } } // multiple topics, highest granularity, only metrics I need receipts { instrumentSelect = "pulsar_*" attrSelect { topic = "receipts-us-*" } filterInstruments { keepOnly = ["pulsar_rate_*"] } } // single topic, don't want subscriptions and consumer level logs { instrumentSelect = "pulsar_*" attrSelect { topic = "logs" } filterInstruments { dropOnly = ["pulsar_subscription_*", "pulsar_consumer_*"] } } }
The configuration is made up of a list of “filtering rules”. Each rule stipulates the following:
instrumentSelect - ability to select one a more instruments apply this filtering rule for.pulsar_*attrSelect - ability to select a group of attributes to apply this rule on, within the selected instruments.topic=receipt-us-*filterInstruments - For each instrument matched, we allow to set whether we wish to drop the attributes selected for certain instruments or keep them.dropOnly - list the instruments we wish to drop selected attributes for. Example: if instrumentSelect is pulsar_* , attrSelect is topic="incoming-logs" and drop is pulsar_subscription_* and pulsar_consumer_* this means all messaging instruments in the topic level will remain and subscription and consumer level metrics will be dropped, only for “incoming-logs” topic.keepOnly - The opposite of drop.dropAllkeepAllOrder matter for the list of filtering rules. This allows us to set a default which applies to a wide range of instruments and then override it for certain instruments.
We will supply a default filtering rules configuration which should make sense.
In certain cases the number of rules can reach to 10k or 20k. Since each rule is essentially a regular expression, we will need to cache the resolve of (instrument, attribute) to true/false. For 10k rules, meaning 10k regular expression this might be too much, so we can use aho-corasik algorithm to match only on select few regular expressions.
We have proposed an issue in OpenTelemetry Java SDK to have a push-down predicate, allowing us to do the filtering while iterating over instruments and their attributes. It means OTel SDK will not allocate a datapoints objects if a certain (instrument, attributes) pair is filtered out. See section above on what we wish to fix in OTel to see issue link.
There will be a sub-PIP detailing the exact configuration syntax, storage, how it will be made dynamic, plugin details and more. We will to take into account user experience defining and updating that configuration, especially when it will reach a large size. Also, we want the experience of adding a rule to be smooth, so the CLI should offer auto-complete of metric names if possible and optionally to have a UI dedicated for it. We will also try to protect and validate the filtering rules such that we can reject updating it if the data points amount will exceed a certain threshold - this is to protect against accidental mistake in the configuration (we will accept it if a force parameter will be supplied).
As mentioned in the design, summaries will be converted into histograms, and those and existing histograms will be modified to be at the granularity level of namespace.
Prometheus Client
LedgerOffloaderStatsImplbrk_ledgeroffloader_read_offload_index_latencybrk_ledgeroffloader_read_offload_data_latencybrk_ledgeroffloader_read_ledger_latencyResourceGroupServicepulsar_resource_group_aggregate_usage_secsTime required to aggregate usage of all resource groups, in seconds.pulsar_resource_group_calculate_quota_secsTime required to calculate quota of all resource groups, in secondsReplicatedSubscriptionsSnapshotBuilderpulsar_replicated_subscriptions_snapshot_msTime taken to create a consistent snapshot across clustersSchemaRegistryStatspulsar_schema_del_ops_latencypulsar_schema_get_ops_latencypulsar_schema_put_ops_latencyBrokerOperabilityMetricstopic_load_timesTransactionBufferClientStatsImplpulsar_txn_tb_client_abort_latencypulsar_txn_tb_client_commit_latencyPendingAckHandleStatsImplpulsar_txn_tp_commit_latencyContextImplpulsar_function_user_metric_*FunctionStatsManagerpulsar_function_``process_latency_mspulsar_function_``process_latency_ms_1minWorkerStatsManagerpulsar_function_worker_start_up_time_msschedule_execution_time_total_msOur Summary
ModularLoadManagerImplpulsar_broker_load_manager_bundle_assigmentpulsar_broker_lookupAbstractTopicpulsar_broker_publish_latencyOpsStatLogger (uses DataSketches)
PulsarZooKeeperClientStatsBucket (Our version of Explicit Bucket Histogram)
ManagedLedgerMBeanImplpulsar_storage_write_latency_le_*pulsar_storage_ledger_write_latency_le_*pulsar_entry_size_le_*CompactionRecordpulsar_compaction_latency_*TransactionMetadataStoreStatspulsar_txn_execution_latency_le_Pulsar repository contains multiple dashboards. We will create the same dashboards using new names alongside existing ones. We’ll add dashboards for different granularity levels: namespace, group and zoom in on specific group/topic. The dashboards will look the same as existing, since the main changes are done to the query of each panel since the metric name has changed, not the semantics.
This will be specified in a sub-PIP.
All changes reported here are applicable to the newly added OTel metrics layer. At first, as mentioned in the document, we’ll be able to use both existing metrics system and OTel metric system - you can toggle each. Once everything will be stabilized, we’ll deprecate the current metric system.
pulsar_ but should be pulsar_messaging_.exposeConsumerLevelMetricsInPrometheus, exposeTopicLevelMetricsInPrometheus and exposeProducerLevelMetricsInPrometheus.topic will not contain partition number, but it will be specified via partitionNum attribute.splitTopicAndPartitionLabelInPrometheus will be deprecated and eventually removed._1min metrics are removed from Pulsar Function metricsresetMetrics RPC operation in processes running Pulsar Function will be removedThe sub-PIPs will specify the exact API changes relevant to their constrained scope, since it will be much easier to review as such.
The sub-PIPs will specify the exact security concerns relevant to their constrained scope, since it will be much easier to review as such.