You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/14 14:18:04 UTC
(pulsar) 02/17: [improve][broker] Make exposing topic level metric dynamic (#21006)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6224c44fb801523c2668b8f64ac0f970790a17e6
Author: Hang Chen <ch...@apache.org>
AuthorDate: Sun Sep 13 03:06:46 2026 +0800
[improve][broker] Make exposing topic level metric dynamic (#21006)
Co-authored-by: Lari Hotari <lh...@apache.org>
(cherry picked from commit 91f690ff7b30a5d4160a3eb80ab0cc709c70195f)
---
.../apache/pulsar/broker/ServiceConfiguration.java | 21 +++++++++-----
.../org/apache/pulsar/broker/PulsarService.java | 6 +---
.../prometheus/PrometheusMetricsGenerator.java | 31 ++++++++++++++++-----
.../prometheus/PulsarPrometheusMetricsServlet.java | 13 +++++++--
.../pulsar/broker/stats/PrometheusMetricsTest.java | 32 ++++++++++++++++++++++
5 files changed, 81 insertions(+), 22 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 7b778269fd19..ea8c23696617 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -3439,9 +3439,10 @@ public class ServiceConfiguration implements PulsarConfiguration {
private boolean authenticateMetricsEndpoint = false;
@FieldContext(
category = CATEGORY_METRICS,
+ dynamic = true,
doc = "If true, export topic level metrics otherwise namespace level"
)
- private boolean exposeTopicLevelMetricsInPrometheus = true;
+ private volatile boolean exposeTopicLevelMetricsInPrometheus = true;
@FieldContext(
category = CATEGORY_METRICS,
doc = "Set to true to enable the broker to cache the metrics response; the default is false. "
@@ -3451,24 +3452,28 @@ public class ServiceConfiguration implements PulsarConfiguration {
private boolean metricsBufferResponse = false;
@FieldContext(
category = CATEGORY_METRICS,
+ dynamic = true,
doc = "If true, export consumer level metrics otherwise namespace level"
)
- private boolean exposeConsumerLevelMetricsInPrometheus = false;
+ private volatile boolean exposeConsumerLevelMetricsInPrometheus = false;
@FieldContext(
category = CATEGORY_METRICS,
+ dynamic = true,
doc = "If true, export producer level metrics otherwise namespace level"
)
- private boolean exposeProducerLevelMetricsInPrometheus = false;
+ private volatile boolean exposeProducerLevelMetricsInPrometheus = false;
@FieldContext(
category = CATEGORY_METRICS,
+ dynamic = true,
doc = "If true, export managed ledger metrics (aggregated by namespace)"
)
- private boolean exposeManagedLedgerMetricsInPrometheus = true;
+ private volatile boolean exposeManagedLedgerMetricsInPrometheus = true;
@FieldContext(
category = CATEGORY_METRICS,
+ dynamic = true,
doc = "If true, export managed cursor metrics"
)
- private boolean exposeManagedCursorMetricsInPrometheus = false;
+ private volatile boolean exposeManagedCursorMetricsInPrometheus = false;
@FieldContext(
category = CATEGORY_METRICS,
doc = "Classname of Pluggable JVM GC metrics logger that can log GC specific metrics")
@@ -3476,11 +3481,12 @@ public class ServiceConfiguration implements PulsarConfiguration {
@FieldContext(
category = CATEGORY_METRICS,
+ dynamic = true,
doc = "Enable expose the precise backlog stats.\n"
+ " Set false to use published counter and consumed counter to calculate,\n"
+ " this would be more efficient but may be inaccurate. Default is false."
)
- private boolean exposePreciseBacklogInPrometheus = false;
+ private volatile boolean exposePreciseBacklogInPrometheus = false;
@FieldContext(
category = CATEGORY_METRICS,
@@ -3492,10 +3498,11 @@ public class ServiceConfiguration implements PulsarConfiguration {
@FieldContext(
category = CATEGORY_METRICS,
+ dynamic = true,
doc = "Enable expose the backlog size for each subscription when generating stats.\n"
+ " Locking is used for fetching the status so default to false."
)
- private boolean exposeSubscriptionBacklogSizeInPrometheus = false;
+ private volatile boolean exposeSubscriptionBacklogSizeInPrometheus = false;
@FieldContext(
category = CATEGORY_METRICS,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 83161b3353d4..5333d48a2825 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -1161,11 +1161,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
}
private synchronized void createMetricsServlet() {
- this.metricsServlet = new PulsarPrometheusMetricsServlet(
- this, config.isExposeTopicLevelMetricsInPrometheus(),
- config.isExposeConsumerLevelMetricsInPrometheus(),
- config.isExposeProducerLevelMetricsInPrometheus(),
- config.isSplitTopicAndPartitionLabelInPrometheus());
+ this.metricsServlet = new PulsarPrometheusMetricsServlet(this);
if (pendingMetricsProviders != null) {
pendingMetricsProviders.forEach(provider -> metricsServlet.addRawMetricsProvider(provider));
this.pendingMetricsProviders = null;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
index 2bb6372c1cbf..285b6aaefc93 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java
@@ -47,6 +47,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.function.BooleanSupplier;
import java.util.zip.CRC32;
import java.util.zip.Deflater;
import lombok.extern.slf4j.Slf4j;
@@ -297,10 +298,10 @@ public class PrometheusMetricsGenerator implements AutoCloseable {
}
private final PulsarService pulsar;
- private final boolean includeTopicMetrics;
- private final boolean includeConsumerMetrics;
- private final boolean includeProducerMetrics;
- private final boolean splitTopicAndPartitionIndexLabel;
+ private final BooleanSupplier includeTopicMetrics;
+ private final BooleanSupplier includeConsumerMetrics;
+ private final BooleanSupplier includeProducerMetrics;
+ private final BooleanSupplier splitTopicAndPartitionIndexLabel;
private final Clock clock;
private volatile int initialBufferSize = DEFAULT_INITIAL_BUFFER_SIZE;
@@ -308,6 +309,20 @@ public class PrometheusMetricsGenerator implements AutoCloseable {
public PrometheusMetricsGenerator(PulsarService pulsar, boolean includeTopicMetrics,
boolean includeConsumerMetrics, boolean includeProducerMetrics,
boolean splitTopicAndPartitionIndexLabel, Clock clock) {
+ this(pulsar, () -> includeTopicMetrics, () -> includeConsumerMetrics, () -> includeProducerMetrics,
+ () -> splitTopicAndPartitionIndexLabel, clock);
+ }
+
+ public PrometheusMetricsGenerator(PulsarService pulsar, Clock clock) {
+ this(pulsar, pulsar.getConfiguration()::isExposeTopicLevelMetricsInPrometheus,
+ pulsar.getConfiguration()::isExposeConsumerLevelMetricsInPrometheus,
+ pulsar.getConfiguration()::isExposeProducerLevelMetricsInPrometheus,
+ pulsar.getConfiguration()::isSplitTopicAndPartitionLabelInPrometheus, clock);
+ }
+
+ private PrometheusMetricsGenerator(PulsarService pulsar, BooleanSupplier includeTopicMetrics,
+ BooleanSupplier includeConsumerMetrics, BooleanSupplier includeProducerMetrics,
+ BooleanSupplier splitTopicAndPartitionIndexLabel, Clock clock) {
this.pulsar = pulsar;
this.includeTopicMetrics = includeTopicMetrics;
this.includeConsumerMetrics = includeConsumerMetrics;
@@ -326,15 +341,17 @@ public class PrometheusMetricsGenerator implements AutoCloseable {
generateSystemMetrics(stream, pulsar.getConfiguration().getClusterName());
- NamespaceStatsAggregator.generate(pulsar, includeTopicMetrics, includeConsumerMetrics,
- includeProducerMetrics, splitTopicAndPartitionIndexLabel, metricStreams);
+ boolean exportTopicMetrics = includeTopicMetrics.getAsBoolean();
+ NamespaceStatsAggregator.generate(pulsar, exportTopicMetrics, includeConsumerMetrics.getAsBoolean(),
+ includeProducerMetrics.getAsBoolean(), splitTopicAndPartitionIndexLabel.getAsBoolean(),
+ metricStreams);
if (pulsar.getWorkerServiceOpt().isPresent()) {
pulsar.getWorkerService().generateFunctionsStats(stream);
}
if (pulsar.getConfiguration().isTransactionCoordinatorEnabled()) {
- TransactionAggregator.generate(pulsar, metricStreams, includeTopicMetrics);
+ TransactionAggregator.generate(pulsar, metricStreams, exportTopicMetrics);
}
metricStreams.flushAllToStream(stream);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PulsarPrometheusMetricsServlet.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PulsarPrometheusMetricsServlet.java
index 7593c6697259..5260249a22bb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PulsarPrometheusMetricsServlet.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PulsarPrometheusMetricsServlet.java
@@ -48,12 +48,19 @@ public class PulsarPrometheusMetricsServlet extends PrometheusMetricsServlet {
public PulsarPrometheusMetricsServlet(PulsarService pulsar, boolean includeTopicMetrics,
boolean includeConsumerMetrics, boolean includeProducerMetrics,
boolean splitTopicAndPartitionLabel) {
+ this(pulsar, new PrometheusMetricsGenerator(pulsar, includeTopicMetrics, includeConsumerMetrics,
+ includeProducerMetrics, splitTopicAndPartitionLabel, Clock.systemUTC()));
+ }
+
+ public PulsarPrometheusMetricsServlet(PulsarService pulsar) {
+ this(pulsar, new PrometheusMetricsGenerator(pulsar, Clock.systemUTC()));
+ }
+
+ private PulsarPrometheusMetricsServlet(PulsarService pulsar, PrometheusMetricsGenerator metricsGenerator) {
super(pulsar.getConfiguration().getMetricsServletTimeoutMs(), pulsar.getConfiguration().getClusterName(),
EXECUTOR_MAX_THREADS);
MetricsExports.initialize();
- prometheusMetricsGenerator =
- new PrometheusMetricsGenerator(pulsar, includeTopicMetrics, includeConsumerMetrics,
- includeProducerMetrics, splitTopicAndPartitionLabel, Clock.systemUTC());
+ prometheusMetricsGenerator = metricsGenerator;
gzipCompressionEnabledForMetrics = isGzipCompressionEnabledForEndpoint(
pulsar.getConfiguration().getHttpServerGzipCompressionExcludedPaths(), DEFAULT_METRICS_PATH);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
index b86b92c5e820..8cd1611e5c34 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java
@@ -87,6 +87,7 @@ import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentMessageExpiryMonitor;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient;
import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageRoutingMode;
@@ -134,6 +135,37 @@ public class PrometheusMetricsTest extends BrokerTestBase {
super.internalCleanup();
}
+ @Test
+ public void testDynamicMetricsConfiguration() throws Exception {
+ String topic = "persistent://my-property/my-ns/dynamic-metrics";
+ try (Producer<byte[]> producer = pulsarClient.newProducer().topic(topic)
+ .producerName("dynamic-producer").create();
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic)
+ .subscriptionName("dynamic-subscription").consumerName("dynamic-consumer").subscribe()) {
+ producer.send(new byte[] {1});
+ consumer.acknowledge(consumer.receive(10, TimeUnit.SECONDS));
+ PrometheusMetricsClient metricsClient = new PrometheusMetricsClient("localhost",
+ pulsar.getListenPortHTTP().orElseThrow());
+ for (boolean enabled : new boolean[] {false, true, false}) {
+ admin.brokers().updateDynamicConfiguration("exposeTopicLevelMetricsInPrometheus",
+ Boolean.toString(enabled));
+ admin.brokers().updateDynamicConfiguration("exposeConsumerLevelMetricsInPrometheus",
+ Boolean.toString(enabled));
+ admin.brokers().updateDynamicConfiguration("exposeProducerLevelMetricsInPrometheus",
+ Boolean.toString(enabled));
+ Awaitility.await().untilAsserted(() -> {
+ Collection<Metric> metrics = metricsClient.getMetrics().getNameToDataPoints().values();
+ assertThat(metrics.stream().anyMatch(metric -> topic.equals(metric.tags.get("topic"))))
+ .isEqualTo(enabled);
+ assertThat(metrics.stream().anyMatch(metric ->
+ "dynamic-producer".equals(metric.tags.get("producer_name")))).isEqualTo(enabled);
+ assertThat(metrics.stream().anyMatch(metric ->
+ "dynamic-consumer".equals(metric.tags.get("consumer_name")))).isEqualTo(enabled);
+ });
+ }
+ }
+ }
+
@Test
public void testPublishRateLimitedTimes() throws Exception {
cleanup();