This is an automated email from the ASF dual-hosted git repository.
anmolnar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zookeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 6925dea5f ZOOKEEPER-4767: New implementation of prometheus quantile
metrics based on DataSketches
6925dea5f is described below
commit 6925dea5fd50c6b3fe1e0e6064216939c041c13e
Author: Yike Xiao <[email protected]>
AuthorDate: Wed Jul 8 22:00:54 2026 +0800
ZOOKEEPER-4767: New implementation of prometheus quantile metrics based on
DataSketches
Reviewers: tisonkun, anmolnar
Author: Shawyeok
Closes #2086 from Shawyeok/sketches-summary
---
.../zookeeper-prometheus-metrics/pom.xml | 6 +
.../prometheus/PrometheusMetricsProvider.java | 141 ++++++++--
.../metrics/prometheus/SketchesSummary.java | 295 +++++++++++++++++++++
.../prometheus/PrometheusMetricsProviderTest.java | 31 +--
.../PrometheusMetricsSummaryRotationTest.java | 91 +++++++
.../metrics/prometheus/SketchesSummaryTest.java | 90 +++++++
.../administrators-guide/administration.mdx | 19 +-
7 files changed, 616 insertions(+), 57 deletions(-)
diff --git a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/pom.xml
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/pom.xml
index 9cdb235c8..8350115e6 100644
--- a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/pom.xml
+++ b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/pom.xml
@@ -34,6 +34,7 @@
<properties>
<prometheus.version>1.3.10</prometheus.version>
<jetty.version>9.4.58.v20250814</jetty.version>
+ <datasketches.version>7.0.1</datasketches.version>
</properties>
<dependencies>
<dependency>
@@ -75,6 +76,11 @@
<artifactId>prometheus-metrics-config</artifactId>
<version>${prometheus.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.datasketches</groupId>
+ <artifactId>datasketches-java</artifactId>
+ <version>${datasketches.version}</version>
+ </dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
diff --git
a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/main/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsProvider.java
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/main/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsProvider.java
index 001b7f25d..2ce0273d7 100644
---
a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/main/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsProvider.java
+++
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/main/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsProvider.java
@@ -31,6 +31,11 @@
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
@@ -81,6 +86,9 @@ public class PrometheusMetricsProvider implements
MetricsProvider {
private Server server;
private int numWorkerThreads;
+ private long workerShutdownTimeoutMs = 1000;
+ private int summaryRotateSeconds = 60;
+ private ScheduledExecutorService summaryRotateExecutor;
private String host;
// SSL Configuration fields
@@ -101,7 +109,27 @@ public class PrometheusMetricsProvider implements
MetricsProvider {
public static final String HTTP_PORT = "httpPort";
public static final String EXPORT_JVM_INFO = "exportJvmInfo";
public static final String HTTPS_PORT = "httpsPort";
+ /**
+ * @deprecated DataSketches-based summaries are lock-free per-thread and
no longer require
+ * worker threads. This property is ignored. See ZOOKEEPER-4741.
+ */
+ @Deprecated
public static final String NUM_WORKER_THREADS = "numWorkerThreads";
+ /**
+ * @deprecated DataSketches-based summaries no longer use a bounded worker
queue. This property
+ * is ignored. See ZOOKEEPER-4741.
+ */
+ @Deprecated
+ public static final String MAX_QUEUE_SIZE = "maxQueueSize";
+ /** Timeout in ms for shutting down the summary rotation executor. */
+ public static final String WORKER_SHUTDOWN_TIMEOUT_MS =
"workerShutdownTimeoutMs";
+ /**
+ * Interval in seconds for rotating per-thread DataSketches into the
aggregated result that is
+ * exposed via {@code /metrics}. Quantiles from observations in the
current interval become
+ * visible after the next rotation. Default is 60 seconds.
+ */
+ public static final String PROMETHEUS_SUMMARY_ROTATE_INTERVAL_SECONDS =
+ "prometheusMetricsSummaryRotateIntervalSeconds";
public static final String SSL_KEYSTORE_LOCATION = "ssl.keyStore.location";
public static final String SSL_KEYSTORE_PASSWORD = "ssl.keyStore.password";
public static final String SSL_KEYSTORE_TYPE = "ssl.keyStore.type";
@@ -144,6 +172,14 @@ public void configure(Properties configuration) throws
MetricsProviderLifeCycleE
this.httpsPort =
Integer.parseInt(configuration.getProperty(HTTPS_PORT, "-1"));
this.exportJvmInfo =
Boolean.parseBoolean(configuration.getProperty(EXPORT_JVM_INFO, "true"));
this.numWorkerThreads =
Integer.parseInt(configuration.getProperty(NUM_WORKER_THREADS, "10"));
+ if (configuration.containsKey(NUM_WORKER_THREADS) ||
configuration.containsKey(MAX_QUEUE_SIZE)) {
+ LOG.warn("The configuration {} and {} are deprecated and ignored.
See ZOOKEEPER-4741.",
+ NUM_WORKER_THREADS, MAX_QUEUE_SIZE);
+ }
+ this.workerShutdownTimeoutMs =
+
Long.parseLong(configuration.getProperty(WORKER_SHUTDOWN_TIMEOUT_MS, "1000"));
+ this.summaryRotateSeconds =
+
Integer.parseInt(configuration.getProperty(PROMETHEUS_SUMMARY_ROTATE_INTERVAL_SECONDS,
"60"));
// If httpsPort is specified, parse all SSL properties
if (this.httpsPort != -1) {
@@ -244,6 +280,7 @@ public void start() throws
MetricsProviderLifeCycleException {
LOG.info("Prometheus metrics provider with Jetty started. HTTP
port: {}, HTTPS port: {}",
httpPort != -1 ? httpPort : "disabled", httpsPort != -1 ?
httpsPort : "disabled");
+ startSummaryRotateTask();
} catch (Exception e) {
LOG.error("Failed to start Prometheus Jetty server", e);
// Ensure server is stopped on startup failure
@@ -252,6 +289,48 @@ public void start() throws
MetricsProviderLifeCycleException {
}
}
+ private void startSummaryRotateTask() {
+ summaryRotateExecutor = new ScheduledThreadPoolExecutor(1, new
SummaryRotateThreadFactory());
+ summaryRotateExecutor.scheduleAtFixedRate(() -> {
+ try {
+ rootContext.rotateAllSummaries();
+ } catch (Exception err) {
+ LOG.error("Cannot rotate Prometheus summaries", err);
+ }
+ }, summaryRotateSeconds, summaryRotateSeconds, TimeUnit.SECONDS);
+ }
+
+ private void shutdownSummaryRotateExecutor() {
+ if (summaryRotateExecutor == null) {
+ return;
+ }
+ LOG.info("Shutting down Prometheus summary rotate executor with
timeout {}ms", workerShutdownTimeoutMs);
+ summaryRotateExecutor.shutdown();
+ try {
+ if
(!summaryRotateExecutor.awaitTermination(workerShutdownTimeoutMs,
TimeUnit.MILLISECONDS)) {
+ LOG.warn("Summary rotate executor did not terminate in {}ms;
forcing shutdown",
+ workerShutdownTimeoutMs);
+ summaryRotateExecutor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ summaryRotateExecutor.shutdownNow();
+ } finally {
+ summaryRotateExecutor = null;
+ }
+ }
+
+ private static class SummaryRotateThreadFactory implements ThreadFactory {
+ private static final AtomicInteger counter = new AtomicInteger(1);
+
+ @Override
+ public Thread newThread(Runnable runnable) {
+ Thread thread = new Thread(runnable, "PrometheusSummaryRotate-" +
counter.getAndIncrement());
+ thread.setDaemon(true);
+ return thread;
+ }
+ }
+
private void setKeyStoreScanner(SslContextFactory.Server
sslContextFactory) {
KeyStoreScanner keystoreScanner = new
KeyStoreScanner(sslContextFactory);
keystoreScanner.setScanInterval(SCAN_INTERVAL);
@@ -335,6 +414,7 @@ private ServerConnector createSslConnector(Server server,
int acceptors, int sel
@Override
public void stop() {
+ shutdownSummaryRotateExecutor();
if (server != null) {
try {
LOG.info("Stopping Prometheus Jetty server.");
@@ -482,16 +562,12 @@ public void unregisterGaugeSet(final String name) {
unregisterGauge(name);
}
- private io.prometheus.metrics.core.metrics.Summary
createPrometheusSummary(String name, DetailLevel detailLevel,
- String... labelNames) {
- io.prometheus.metrics.core.metrics.Summary.Builder builder =
io.prometheus.metrics.core.metrics.Summary
- .builder().name(name).help(name + "
summary").quantile(0.5, 0.05); // Median
-
+ private SketchesSummary createSketchesSummary(String name, DetailLevel
detailLevel, String... labelNames) {
+ SketchesSummary.Builder builder = SketchesSummary.build(name, name
+ " summary")
+ .quantile(0.5); // Median
if (detailLevel == DetailLevel.ADVANCED) {
- builder.quantile(0.95, 0.05) // 95th percentile
- .quantile(0.99, 0.05); // 99th percentile
+ builder.quantile(0.95).quantile(0.99); // 95th and 99th
percentile
}
-
if (labelNames.length > 0) {
builder.labelNames(labelNames);
}
@@ -508,9 +584,7 @@ public Summary getSummary(String name, DetailLevel
detailLevel) {
throw new IllegalArgumentException(
"Already registered a summary as " + key + " with
a different detail level");
}
- io.prometheus.metrics.core.metrics.Summary prometheusSummary =
createPrometheusSummary(key,
- detailLevel);
- return new PrometheusSummaryWrapper(prometheusSummary);
+ return new PrometheusSummaryWrapper(createSketchesSummary(key,
detailLevel), key);
});
}
@@ -524,11 +598,16 @@ public SummarySet getSummarySet(String name, DetailLevel
detailLevel) {
throw new IllegalArgumentException(
"Already registered a summary set as " + key + "
with a different detail level");
}
- io.prometheus.metrics.core.metrics.Summary prometheusSummary =
createPrometheusSummary(key, detailLevel,
- LABEL);
- return new PrometheusLabelledSummaryWrapper(prometheusSummary);
+ return new
PrometheusLabelledSummaryWrapper(createSketchesSummary(key, detailLevel,
LABEL), key);
});
}
+
+ void rotateAllSummaries() {
+ basicSummaries.values().forEach(s -> s.inner.rotate());
+ advancedSummaries.values().forEach(s -> s.inner.rotate());
+ basicSummarySets.values().forEach(s -> s.inner.rotate());
+ advancedSummarySets.values().forEach(s -> s.inner.rotate());
+ }
}
// --- Wrapper classes to adapt Prometheus metrics to ZooKeeper's metric
interfaces ---
@@ -578,29 +657,43 @@ public void inc(String key) {
}
}
- private static class PrometheusSummaryWrapper implements Summary {
- private final io.prometheus.metrics.core.metrics.Summary
prometheusSummary;
+ static class PrometheusSummaryWrapper implements Summary {
+ // VisibleForTesting
+ final SketchesSummary inner;
+ private final String name;
- public
PrometheusSummaryWrapper(io.prometheus.metrics.core.metrics.Summary
prometheusSummary) {
- this.prometheusSummary = prometheusSummary;
+ PrometheusSummaryWrapper(SketchesSummary inner, String name) {
+ this.inner = inner;
+ this.name = name;
}
@Override
public void add(long value) {
- this.prometheusSummary.observe(value);
+ try {
+ inner.observe(value);
+ } catch (IllegalArgumentException err) {
+ LOG.error("invalid delta {} for metric {}", value, name, err);
+ }
}
}
- private static class PrometheusLabelledSummaryWrapper implements
SummarySet {
- private final io.prometheus.metrics.core.metrics.Summary
prometheusSummary;
+ static class PrometheusLabelledSummaryWrapper implements SummarySet {
+ // VisibleForTesting
+ final SketchesSummary inner;
+ private final String name;
- public
PrometheusLabelledSummaryWrapper(io.prometheus.metrics.core.metrics.Summary
prometheusSummary) {
- this.prometheusSummary = prometheusSummary;
+ PrometheusLabelledSummaryWrapper(SketchesSummary inner, String name) {
+ this.inner = inner;
+ this.name = name;
}
@Override
public void add(String key, long value) {
- this.prometheusSummary.labelValues(key).observe(value);
+ try {
+ inner.labels(key).observe(value);
+ } catch (IllegalArgumentException err) {
+ LOG.error("invalid value {} for metric {} with key {}", value,
name, key, err);
+ }
}
}
}
diff --git
a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/main/java/org/apache/zookeeper/metrics/prometheus/SketchesSummary.java
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/main/java/org/apache/zookeeper/metrics/prometheus/SketchesSummary.java
new file mode 100644
index 000000000..c874f95fa
--- /dev/null
+++
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/main/java/org/apache/zookeeper/metrics/prometheus/SketchesSummary.java
@@ -0,0 +1,295 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.metrics.prometheus;
+
+import io.prometheus.metrics.model.registry.Collector;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
+import io.prometheus.metrics.model.snapshots.Labels;
+import io.prometheus.metrics.model.snapshots.MetricMetadata;
+import io.prometheus.metrics.model.snapshots.MetricSnapshot;
+import io.prometheus.metrics.model.snapshots.Quantile;
+import io.prometheus.metrics.model.snapshots.Quantiles;
+import io.prometheus.metrics.model.snapshots.SummarySnapshot;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.DoubleAdder;
+import java.util.concurrent.locks.StampedLock;
+import org.apache.datasketches.quantiles.DoublesSketch;
+import org.apache.datasketches.quantiles.DoublesUnion;
+import org.apache.datasketches.quantiles.DoublesUnionBuilder;
+
+/**
+ * A DataSketches based Summary metric, aim for lock-free and low gc overhead.
+ *
+ * <p>Per-thread {@link DoublesSketch} instances accumulate observations
without contention. A
+ * periodic {@link #rotate()} aggregates them into a single result sketch that
is exposed via
+ * {@link #collect()}. Quantile values returned to Prometheus are based on the
most recently
+ * rotated result; before the first rotation they are {@link Double#NaN}.
+ */
+public class SketchesSummary implements Collector {
+
+ private final String name;
+ private final String help;
+ private final List<String> labelNames;
+ private final List<Double> quantiles;
+ private final ConcurrentMap<List<String>, Child> children = new
ConcurrentHashMap<>();
+ private final Child noLabelsChild;
+
+ private SketchesSummary(Builder b) {
+ this.name = b.name;
+ this.help = b.help == null ? b.name : b.help;
+ this.labelNames = Collections.unmodifiableList(new
ArrayList<>(b.labelNames));
+ this.quantiles = Collections.unmodifiableList(new
ArrayList<>(b.quantiles));
+ if (this.labelNames.isEmpty()) {
+ this.noLabelsChild = new Child(this.quantiles);
+ this.children.put(Collections.<String>emptyList(),
this.noLabelsChild);
+ } else {
+ this.noLabelsChild = null;
+ }
+ }
+
+ /** Observe a value on the no-labels child. */
+ public void observe(double amt) {
+ if (noLabelsChild == null) {
+ throw new IllegalStateException("Summary has labels; use
labels(...).observe()");
+ }
+ noLabelsChild.observe(amt);
+ }
+
+ /** Get or create the child for the given label values. */
+ public Child labels(String... values) {
+ if (values.length != labelNames.size()) {
+ throw new IllegalArgumentException(
+ "Incorrect number of label values: expected " +
labelNames.size() + " got " + values.length);
+ }
+ List<String> key =
Collections.unmodifiableList(Arrays.asList(values.clone()));
+ Child existing = children.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ return children.computeIfAbsent(key, k -> new Child(quantiles));
+ }
+
+ /**
+ * Aggregate per-thread sketches across all children into their result
sketches. Must be called
+ * periodically (the {@link PrometheusMetricsProvider} schedules this) so
that quantile values
+ * become visible to scrapers.
+ */
+ public void rotate() {
+ for (Child child : children.values()) {
+ child.rotate();
+ }
+ }
+
+ /** Register this collector and return it (fluent style). */
+ public SketchesSummary register(PrometheusRegistry registry) {
+ registry.register(this);
+ return this;
+ }
+
+ @Override
+ public MetricSnapshot collect() {
+ List<SummarySnapshot.SummaryDataPointSnapshot> dataPoints = new
ArrayList<>(children.size());
+ String[] labelNamesArray = labelNames.toArray(new String[0]);
+ for (Map.Entry<List<String>, Child> entry : children.entrySet()) {
+ Child child = entry.getValue();
+ Child.Value value = child.get();
+ List<Quantile> quantileList = new
ArrayList<>(value.quantiles.size());
+ for (Map.Entry<Double, Double> q : value.quantiles.entrySet()) {
+ quantileList.add(new Quantile(q.getKey(), q.getValue()));
+ }
+ Labels labels = labelNames.isEmpty()
+ ? Labels.EMPTY
+ : Labels.of(labelNamesArray, entry.getKey().toArray(new
String[0]));
+ dataPoints.add(SummarySnapshot.SummaryDataPointSnapshot.builder()
+ .count((long) value.count)
+ .sum(value.sum)
+ .quantiles(Quantiles.of(quantileList))
+ .labels(labels)
+ .build());
+ }
+ return new SummarySnapshot(new MetricMetadata(name, help), dataPoints);
+ }
+
+ @Override
+ public String getPrometheusName() {
+ return name;
+ }
+
+ public static Builder build(String name, String help) {
+ return new Builder().name(name).help(help);
+ }
+
+ public static Builder build() {
+ return new Builder();
+ }
+
+ public static class Builder {
+ private String name;
+ private String help;
+ private final List<String> labelNames = new ArrayList<>();
+ private final List<Double> quantiles = new ArrayList<>();
+
+ public Builder name(String name) {
+ this.name = name;
+ return this;
+ }
+
+ public Builder help(String help) {
+ this.help = help;
+ return this;
+ }
+
+ public Builder labelNames(String... labelNames) {
+ for (String label : labelNames) {
+ if ("quantile".equals(label)) {
+ throw new IllegalArgumentException("Summary cannot have a
label named 'quantile'.");
+ }
+ this.labelNames.add(label);
+ }
+ return this;
+ }
+
+ public Builder quantile(double quantile) {
+ if (quantile < 0.0 || quantile > 1.0) {
+ throw new IllegalArgumentException(
+ "Quantile " + quantile + " invalid: Expected number
between 0.0 and 1.0.");
+ }
+ quantiles.add(quantile);
+ return this;
+ }
+
+ public SketchesSummary create() {
+ if (name == null) {
+ throw new IllegalStateException("SketchesSummary name must be
set");
+ }
+ return new SketchesSummary(this);
+ }
+
+ public SketchesSummary register(PrometheusRegistry registry) {
+ return create().register(registry);
+ }
+ }
+
+ /** A single child (per label-values combination). */
+ public static class Child {
+
+ public static class Value {
+ public final double count;
+ public final double sum;
+ public final SortedMap<Double, Double> quantiles;
+
+ private Value(double count, double sum, List<Double> quantiles,
DoublesSketch doublesSketch) {
+ this.count = count;
+ this.sum = sum;
+ this.quantiles =
Collections.unmodifiableSortedMap(snapshot(quantiles, doublesSketch));
+ }
+
+ private SortedMap<Double, Double> snapshot(List<Double> quantiles,
DoublesSketch doublesSketch) {
+ SortedMap<Double, Double> result = new TreeMap<>();
+ for (Double q : quantiles) {
+ result.put(q, doublesSketch != null &&
!doublesSketch.isEmpty()
+ ? doublesSketch.getQuantile(q) : Double.NaN);
+ }
+ return result;
+ }
+ }
+
+ // Having these separate leaves us open to races, however Prometheus
as a whole has other
+ // races that mean adding atomicity here wouldn't be useful. This
should be reevaluated in
+ // the future.
+ private final DoubleAdder count = new DoubleAdder();
+ private final DoubleAdder sum = new DoubleAdder();
+ private final List<Double> quantiles;
+ /*
+ * Use 2 rotating thread local accessors so that we can safely swap
them.
+ */
+ private volatile ThreadLocalAccessor current;
+ private volatile ThreadLocalAccessor replacement;
+ private volatile DoublesSketch result;
+
+ private Child(List<Double> quantiles) {
+ this.quantiles = quantiles;
+ this.current = new ThreadLocalAccessor();
+ this.replacement = new ThreadLocalAccessor();
+ }
+
+ public void observe(double amt) {
+ count.add(1);
+ sum.add(amt);
+ LocalSketch localSketch = current.localData.get();
+ long stamp = localSketch.lock.readLock();
+ try {
+ localSketch.sketch.update(amt);
+ } finally {
+ localSketch.lock.unlockRead(stamp);
+ }
+ }
+
+ public void rotate() {
+ ThreadLocalAccessor swap = current;
+ current = replacement;
+ replacement = swap;
+ List<Thread> deadThreads = new ArrayList<>();
+ final DoublesUnion aggregated = new DoublesUnionBuilder().build();
+ swap.map.forEach((thread, localSketch) -> {
+ long stamp = localSketch.lock.writeLock();
+ try {
+ aggregated.union(localSketch.sketch);
+ localSketch.sketch.reset();
+ if (!thread.isAlive()) {
+ deadThreads.add(thread);
+ }
+ } finally {
+ localSketch.lock.unlockWrite(stamp);
+ }
+ });
+ for (Thread deadThread : deadThreads) {
+ swap.map.remove(deadThread);
+ current.map.remove(deadThread);
+ }
+ result = aggregated.getResultAndReset();
+ }
+
+ public Value get() {
+ return new Value(count.sum(), sum.sum(), quantiles, result);
+ }
+ }
+
+ private static class ThreadLocalAccessor {
+ private final Map<Thread, LocalSketch> map = new ConcurrentHashMap<>();
+ private final ThreadLocal<LocalSketch> localData =
ThreadLocal.withInitial(() -> {
+ LocalSketch sketch = new LocalSketch();
+ map.put(Thread.currentThread(), sketch);
+ return sketch;
+ });
+ }
+
+ private static class LocalSketch {
+ private final DoublesSketch sketch = DoublesSketch.builder().build();
+ private final StampedLock lock = new StampedLock();
+ }
+}
diff --git
a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsProviderTest.java
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsProviderTest.java
index 2d36b0017..533468450 100644
---
a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsProviderTest.java
+++
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsProviderTest.java
@@ -287,6 +287,8 @@ public void testBasicSummary() throws Exception {
Summary summary = provider.getRootContext().getSummary("cc",
MetricsContext.DetailLevel.BASIC);
summary.add(10);
summary.add(10);
+ // Force per-thread sketches to be aggregated into the result sketch.
+ ((PrometheusMetricsProvider.PrometheusSummaryWrapper)
summary).inner.rotate();
int[] count = { 0 };
provider.dump((k, v) -> {
count[0]++;
@@ -332,6 +334,7 @@ public void testAdvancedSummary() throws Exception {
Summary summary = provider.getRootContext().getSummary("cc",
MetricsContext.DetailLevel.ADVANCED);
summary.add(10);
summary.add(10);
+ ((PrometheusMetricsProvider.PrometheusSummaryWrapper)
summary).inner.rotate();
int[] count = { 0 };
provider.dump((k, v) -> {
count[0]++;
@@ -411,33 +414,6 @@ public void testTraceCall() throws ServletException,
IOException {
verify(response).sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED);
}
- @Test
- public void testSummary_asyncAndExceedMaxQueueSize() throws Exception {
- final Properties config = new Properties();
- config.setProperty("numWorkerThreads", "1");
- config.setProperty("maxQueueSize", "1");
- config.setProperty("httpPort", "0"); // ephemeral port
- config.setProperty("exportJvmInfo", "false");
-
- PrometheusMetricsProvider metricsProvider = null;
- try {
- metricsProvider = new PrometheusMetricsProvider();
- metricsProvider.configure(config);
- metricsProvider.start();
- final Summary summary =
metricsProvider.getRootContext().getSummary("cc",
- MetricsContext.DetailLevel.ADVANCED);
-
- // make sure no error is thrown
- for (int i = 0; i < 10; i++) {
- summary.add(10);
- }
- } finally {
- if (metricsProvider != null) {
- metricsProvider.stop();
- }
- }
- }
-
@Test
public void testSummarySet() throws Exception {
final String name = "ss";
@@ -451,6 +427,7 @@ public void testSummarySet() throws Exception {
for (int i = 0; i < count; i++) {
Arrays.asList(keys).forEach(key -> summarySet.add(key, 1));
}
+ ((PrometheusMetricsProvider.PrometheusLabelledSummaryWrapper)
summarySet).inner.rotate();
// validate with dump call
final Map<String, Number> expectedMetricsMap = new HashMap<>();
diff --git
a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsSummaryRotationTest.java
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsSummaryRotationTest.java
new file mode 100644
index 000000000..6e6728e6d
--- /dev/null
+++
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/PrometheusMetricsSummaryRotationTest.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.metrics.prometheus;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
+import java.util.Properties;
+import java.util.SortedMap;
+import java.util.concurrent.TimeUnit;
+import org.apache.zookeeper.metrics.MetricsContext;
+import org.apache.zookeeper.metrics.Summary;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.opentest4j.AssertionFailedError;
+
+/**
+ * Tests about PrometheusSummary, make sure sketches rotation task works as
expected.
+ */
+public class PrometheusMetricsSummaryRotationTest {
+
+ private PrometheusMetricsProvider provider;
+
+ private final int summaryRotateSeconds = 2;
+
+ @BeforeEach
+ public void setup() throws Exception {
+ PrometheusRegistry.defaultRegistry.clear();
+ provider = new PrometheusMetricsProvider();
+ Properties configuration = new Properties();
+ configuration.setProperty("httpHost", "127.0.0.1"); // local host for
test
+ configuration.setProperty("httpPort", "0"); // ephemeral port
+ configuration.setProperty("exportJvmInfo", "false");
+
configuration.setProperty(PrometheusMetricsProvider.PROMETHEUS_SUMMARY_ROTATE_INTERVAL_SECONDS,
+ String.valueOf(summaryRotateSeconds));
+ provider.configure(configuration);
+ provider.start();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (provider != null) {
+ provider.stop();
+ }
+ PrometheusRegistry.defaultRegistry.clear();
+ }
+
+ @Test
+ void testSummaryRotation() throws InterruptedException {
+ Summary summary = provider.getRootContext().getSummary("rotation_" +
System.currentTimeMillis(),
+ MetricsContext.DetailLevel.ADVANCED);
+ for (int i = 0; i < 10; i++) {
+ summary.add(10);
+ }
+ SketchesSummary innerSummary =
+ ((PrometheusMetricsProvider.PrometheusSummaryWrapper)
summary).inner;
+ boolean assertionFailed = true;
+ int timeout = summaryRotateSeconds + 1;
+ for (int i = 0; i < timeout && assertionFailed; i++) {
+ SortedMap<Double, Double> quantiles =
innerSummary.labels().get().quantiles;
+ try {
+ assertEquals(10, quantiles.get(0.5));
+ assertEquals(10, quantiles.get(0.95));
+ assertEquals(10, quantiles.get(0.99));
+ assertionFailed = false;
+ } catch (AssertionFailedError e) {
+ TimeUnit.SECONDS.sleep(1);
+ }
+ }
+ if (assertionFailed) {
+ fail("Quantiles not updated after " + timeout + "seconds, it is
likely that the summary is not rotated.");
+ }
+ }
+}
diff --git
a/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/SketchesSummaryTest.java
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/SketchesSummaryTest.java
new file mode 100644
index 000000000..fbbd8a1c4
--- /dev/null
+++
b/zookeeper-metrics-providers/zookeeper-prometheus-metrics/src/test/java/org/apache/zookeeper/metrics/prometheus/SketchesSummaryTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.metrics.prometheus;
+
+import static java.util.Arrays.binarySearch;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Random;
+import java.util.SortedMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests about SketchesSummary, make sure the quantile error is within the
expected range.
+ */
+public class SketchesSummaryTest {
+
+ @Test
+ void testQuantileError() {
+ SketchesSummary summary = SketchesSummary.build("test", "test help")
+ .quantile(0.5)
+ .quantile(0.9)
+ .quantile(0.99)
+ .create();
+ Random random = new Random();
+ int[] samples = new int[1_000];
+ double sum = 0;
+ for (int i = 0; i < samples.length; i++) {
+ int sample = random.nextInt(1_000_000) + 1;
+ summary.observe(sample);
+ sum += sample;
+ samples[i] = sample;
+ }
+ Arrays.sort(samples);
+ summary.rotate();
+ SketchesSummary.Child.Value value = summary.labels().get();
+ assertEquals(samples.length, value.count);
+ assertEquals(sum, value.sum);
+ // The default k of DoublesSketches is 128, rank error of about 1.7%.
+ // See more: org.apache.datasketches.quantiles.DoublesSketch
+ assertQuantileError(samples, value.quantiles, 0.017);
+ }
+
+ @Test
+ public void testEmptySketch() {
+ SketchesSummary summary = SketchesSummary.build("testEmptySketch",
"test help")
+ .quantile(0.5)
+ .quantile(0.9)
+ .quantile(0.99)
+ .create();
+ summary.observe(10);
+ summary.rotate();
+ summary.observe(10);
+ summary.rotate();
+ summary.rotate();
+ for (Double quantile : summary.labels().get().quantiles.values()) {
+ assertTrue(Double.isNaN(quantile));
+ }
+ }
+
+ private static void assertQuantileError(int[] samples, SortedMap<Double,
Double> quantiles, double delta) {
+ for (Map.Entry<Double, Double> entry : quantiles.entrySet()) {
+ double quantile = entry.getKey();
+ int expected = (int) (samples.length * quantile - 1);
+ int actual = binarySearch(samples, entry.getValue().intValue());
+ if (Math.abs(expected - actual) > samples.length * delta) {
+ Assertions.fail(String.format("expected error delta: %s,
actual: %s abs(%s-%s)",
+ delta, Math.abs(expected - actual), expected, actual));
+ }
+ }
+ }
+}
diff --git
a/zookeeper-website/app/pages/_docs/docs/_mdx/admin-ops/administrators-guide/administration.mdx
b/zookeeper-website/app/pages/_docs/docs/_mdx/admin-ops/administrators-guide/administration.mdx
index 612ec6079..bdcc3bd6c 100644
---
a/zookeeper-website/app/pages/_docs/docs/_mdx/admin-ops/administrators-guide/administration.mdx
+++
b/zookeeper-website/app/pages/_docs/docs/_mdx/admin-ops/administrators-guide/administration.mdx
@@ -163,17 +163,18 @@ Since 3.6.0 ZooKeeper binary package bundles an
integration with [Prometheus.io]
If this property is set to **true** Prometheus.io will export useful metrics
about the JVM.
The default is true.
- _metricsProvider.numWorkerThreads_ :
- **New in 3.7.1:**
+ **New in 3.7.1**, **deprecated in 3.10.0:**
Number of worker threads for reporting Prometheus summary metrics.
- Default value is 1.
- If the number is less than 1, the main thread will be used.
+ This property is ignored since ZOOKEEPER-4741; the DataSketches-based summary
+ implementation is lock-free per-thread and no longer requires a worker pool.
- _metricsProvider.maxQueueSize_ :
- **New in 3.7.1:**
+ **New in 3.7.1**, **deprecated in 3.10.0:**
The max queue size for Prometheus summary metrics reporting task.
- Default value is 10000.
+ This property is ignored since ZOOKEEPER-4741; the DataSketches-based summary
+ implementation does not use a bounded worker queue.
- _metricsProvider.workerShutdownTimeoutMs_ :
**New in 3.7.1:**
- The timeout in ms for Prometheus worker threads shutdown.
+ The timeout in ms for shutting down the Prometheus summary rotation executor.
Default value is 1000ms.
- _metricsProvider.ssl.keyStore.location_ and
_metricsProvider.ssl.keyStore.password_ :
**New in 3.10.0:**
@@ -207,3 +208,9 @@ Since 3.6.0 ZooKeeper binary package bundles an integration
with [Prometheus.io]
**New in 3.10.0:**
The enabled protocols to be used in TLS negotiation for
PrometheusMetricsProvider.
Default value is the Jetty default.
+- _metricsProvider.prometheusMetricsSummaryRotateIntervalSeconds_ :
+ **New in 3.10.0:**
+ Interval in seconds for rotating per-thread DataSketches into the aggregated
+ result exposed via `/metrics`. Quantiles from observations in the current
+ interval become visible after the next rotation.
+ Default value is 60 seconds.
