diff --git a/bookkeeper-prometheus-metrics-provider/build.gradle.kts b/bookkeeper-prometheus-metrics-provider/build.gradle.kts new file mode 100644 index 0000000000000..7555da2e71a5d --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/build.gradle.kts @@ -0,0 +1,41 @@ +/* + * 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. + */ + +plugins { + id("pulsar.public-java-library-conventions") +} + +dependencies { + implementation(libs.slog) + implementation(libs.bookkeeper.stats.api) + // Prometheus Java client 1.x. The registry and naming helpers live in -model, the callback gauges in + // -core, the JVM instrumentation in -instrumentation-jvm and the text exposition writer in + // -exposition-textformats. + api(libs.prometheus.metrics.model) + implementation(libs.prometheus.metrics.core) + implementation(libs.prometheus.metrics.instrumentation.jvm) + implementation(libs.prometheus.metrics.exposition.textformats) + implementation(libs.javax.servlet.api) + implementation(libs.jetty.ee8.servlet) + implementation(libs.guava) + implementation(libs.netty.common) + implementation(libs.datasketches.java) + + testImplementation(libs.netty.buffer) +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/DataSketchesOpStatsLogger.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/DataSketchesOpStatsLogger.java new file mode 100644 index 0000000000000..f9f6a882ca677 --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/DataSketchesOpStatsLogger.java @@ -0,0 +1,219 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import io.netty.util.concurrent.FastThreadLocal; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.LongAdder; +import java.util.concurrent.locks.StampedLock; +import org.apache.bookkeeper.stats.OpStatsData; +import org.apache.bookkeeper.stats.OpStatsLogger; +import org.apache.datasketches.kll.KllDoublesSketch; + +/** + * OpStatsLogger implementation that uses DataSketches library to calculate the approximated latency quantiles. + */ +public class DataSketchesOpStatsLogger implements OpStatsLogger { + + /* + * Use 2 rotating thread local accessor so that we can safely swap them. + */ + private volatile ThreadLocalAccessor current; + private volatile ThreadLocalAccessor replacement; + + /* + * These are the sketches where all the aggregated results are published. + */ + private volatile KllDoublesSketch successResult; + private volatile KllDoublesSketch failResult; + + private final LongAdder successCountAdder = new LongAdder(); + private final LongAdder failCountAdder = new LongAdder(); + + private final LongAdder successSumAdder = new LongAdder(); + private final LongAdder failSumAdder = new LongAdder(); + + private Map labels; + + // used for lazy registration for thread scoped metrics + private boolean threadInitialized; + + public DataSketchesOpStatsLogger(Map labels) { + this.current = new ThreadLocalAccessor(); + this.replacement = new ThreadLocalAccessor(); + this.labels = labels; + } + + @Override + public void registerFailedEvent(long eventLatency, TimeUnit unit) { + double valueMillis = unit.toMicros(eventLatency) / 1000.0; + + failCountAdder.increment(); + failSumAdder.add((long) valueMillis); + + LocalData localData = current.localData.get(); + + long stamp = localData.lock.readLock(); + try { + localData.failSketch.update(valueMillis); + } finally { + localData.lock.unlockRead(stamp); + } + } + + @Override + public void registerSuccessfulEvent(long eventLatency, TimeUnit unit) { + double valueMillis = unit.toMicros(eventLatency) / 1000.0; + + successCountAdder.increment(); + successSumAdder.add((long) valueMillis); + + LocalData localData = current.localData.get(); + + long stamp = localData.lock.readLock(); + try { + localData.successSketch.update(valueMillis); + } finally { + localData.lock.unlockRead(stamp); + } + } + + @Override + public void registerSuccessfulValue(long value) { + successCountAdder.increment(); + successSumAdder.add(value); + + LocalData localData = current.localData.get(); + + long stamp = localData.lock.readLock(); + try { + localData.successSketch.update(value); + } finally { + localData.lock.unlockRead(stamp); + } + } + + @Override + public void registerFailedValue(long value) { + failCountAdder.increment(); + failSumAdder.add(value); + + LocalData localData = current.localData.get(); + + long stamp = localData.lock.readLock(); + try { + localData.failSketch.update(value); + } finally { + localData.lock.unlockRead(stamp); + } + } + + @Override + public OpStatsData toOpStatsData() { + // Not relevant as we don't use JMX here + throw new UnsupportedOperationException(); + } + + @Override + public void clear() { + // Not relevant as we don't use JMX here + throw new UnsupportedOperationException(); + } + + public void rotateLatencyCollection() { + // Swap current with replacement + ThreadLocalAccessor local = current; + current = replacement; + replacement = local; + + final KllDoublesSketch aggregateSuccess = KllDoublesSketch.newHeapInstance(); + final KllDoublesSketch aggregateFail = KllDoublesSketch.newHeapInstance(); + local.map.forEach((localData, b) -> { + long stamp = localData.lock.writeLock(); + try { + aggregateSuccess.merge(localData.successSketch); + aggregateFail.merge(localData.failSketch); + localData.successSketch = KllDoublesSketch.newHeapInstance(); + localData.failSketch = KllDoublesSketch.newHeapInstance(); + } finally { + localData.lock.unlockWrite(stamp); + } + }); + + successResult = aggregateSuccess; + failResult = aggregateFail; + } + + public long getCount(boolean success) { + return success ? successCountAdder.sum() : failCountAdder.sum(); + } + + public long getSum(boolean success) { + return success ? successSumAdder.sum() : failSumAdder.sum(); + } + + public double getQuantileValue(boolean success, double quantile) { + KllDoublesSketch s = success ? successResult : failResult; + return (s != null && !s.isEmpty()) ? s.getQuantile(quantile) : Double.NaN; + } + + public Map getLabels() { + return labels; + } + + public boolean isThreadInitialized() { + return threadInitialized; + } + + public void initializeThread(Map labels) { + this.labels = labels; + this.threadInitialized = true; + } + + private static class LocalData { + private KllDoublesSketch successSketch = KllDoublesSketch.newHeapInstance(); + private KllDoublesSketch failSketch = KllDoublesSketch.newHeapInstance(); + private final StampedLock lock = new StampedLock(); + } + + private static class ThreadLocalAccessor { + private final Map map = new ConcurrentHashMap<>(); + private final FastThreadLocal localData = new FastThreadLocal() { + + @Override + protected LocalData initialValue() throws Exception { + LocalData localData = new LocalData(); + map.put(localData, Boolean.TRUE); + return localData; + } + + @Override + protected void onRemoval(LocalData value) throws Exception { + map.remove(value); + } + }; + } + + @Override + public String toString() { + return "DataSketchesOpStatsLogger{labels=" + labels + ", id=" + System.identityHashCode(this) + "}"; + } +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/LongAdderCounter.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/LongAdderCounter.java new file mode 100644 index 0000000000000..a5a8e4d2f029f --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/LongAdderCounter.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.pulsar.metrics.prometheus.bookkeeper; + +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.LongAdder; +import org.apache.bookkeeper.stats.Counter; + +/** + * {@link Counter} implementation based on {@link LongAdder}. + * + *

LongAdder keeps a counter per-thread and then aggregates to get the result, in order to avoid contention between + * multiple threads. + */ +public class LongAdderCounter implements Counter { + private final LongAdder counter = new LongAdder(); + + private Map labels; + + // used for lazy registration for thread scoped metric + private boolean threadInitialized; + + public LongAdderCounter(Map labels) { + this.labels = labels; + } + + @Override + public void clear() { + counter.reset(); + } + + @Override + public void inc() { + counter.increment(); + } + + @Override + public void dec() { + counter.decrement(); + } + + @Override + public void addCount(long delta) { + counter.add(delta); + } + + /** + * When counter is used to count time. + * consistent with the {@link DataSketchesOpStatsLogger#registerSuccessfulEvent(long, TimeUnit)} 's logic + * */ + @Override + public void addLatency(long eventLatency, TimeUnit unit) { + long valueMillis = unit.toMillis(eventLatency); + counter.add(valueMillis); + } + + @Override + public Long get() { + return counter.sum(); + } + + public Map getLabels() { + return labels; + } + + public boolean isThreadInitialized() { + return threadInitialized; + } + + public void initializeThread(Map labels) { + this.labels = labels; + this.threadInitialized = true; + } +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusMetricsProvider.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusMetricsProvider.java new file mode 100644 index 0000000000000..084e820ee3e0e --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusMetricsProvider.java @@ -0,0 +1,259 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +// CHECKSTYLE.OFF: IllegalImport + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.util.concurrent.MoreExecutors; +import io.netty.util.concurrent.DefaultThreadFactory; +import io.netty.util.internal.PlatformDependent; +import io.prometheus.metrics.core.metrics.GaugeWithCallback; +import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import io.prometheus.metrics.model.snapshots.PrometheusNaming; +import java.io.IOException; +import java.io.Writer; +import java.lang.management.BufferPoolMXBean; +import java.lang.management.ManagementFactory; +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; +import lombok.CustomLog; +import org.apache.bookkeeper.stats.StatsLogger; +import org.apache.bookkeeper.stats.StatsProvider; +import org.apache.bookkeeper.stats.ThreadRegistry; +import org.apache.commons.configuration2.Configuration; +import org.apache.commons.lang3.StringUtils; +import org.eclipse.jetty.ee8.servlet.ServletContextHandler; +import org.eclipse.jetty.ee8.servlet.ServletHolder; +import org.eclipse.jetty.server.Server; +// CHECKSTYLE.ON: IllegalImport + +/** + * A Prometheus based {@link StatsProvider} implementation. + */ +@CustomLog +public class PrometheusMetricsProvider implements StatsProvider { + + private ScheduledExecutorService executor; + + public static final String PROMETHEUS_STATS_HTTP_ENABLE = "prometheusStatsHttpEnable"; + public static final boolean DEFAULT_PROMETHEUS_STATS_HTTP_ENABLE = true; + + public static final String PROMETHEUS_STATS_HTTP_ADDRESS = "prometheusStatsHttpAddress"; + public static final String DEFAULT_PROMETHEUS_STATS_HTTP_ADDR = "0.0.0.0"; + + public static final String PROMETHEUS_STATS_HTTP_PORT = "prometheusStatsHttpPort"; + public static final int DEFAULT_PROMETHEUS_STATS_HTTP_PORT = 8000; + + public static final String PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS = "prometheusStatsLatencyRolloverSeconds"; + public static final int DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS = 60; + + final PrometheusRegistry registry; + + Server server; + + /* + * These acts a registry of the metrics defined in this provider + */ + final ConcurrentMap counters = new ConcurrentHashMap<>(); + final ConcurrentMap> gauges = new ConcurrentHashMap<>(); + final ConcurrentMap opStats = new ConcurrentHashMap<>(); + final ConcurrentMap threadScopedOpStats = + new ConcurrentHashMap<>(); + final ConcurrentMap threadScopedCounters = + new ConcurrentHashMap<>(); + + public PrometheusMetricsProvider() { + this(PrometheusRegistry.defaultRegistry); + } + + public PrometheusMetricsProvider(PrometheusRegistry registry) { + this.registry = registry; + } + + @Override + public void start(Configuration conf) { + boolean httpEnabled = conf.getBoolean(PROMETHEUS_STATS_HTTP_ENABLE, DEFAULT_PROMETHEUS_STATS_HTTP_ENABLE); + boolean bkHttpServerEnabled = conf.getBoolean("httpServerEnabled", false); + boolean exposeDefaultJVMMetrics = conf.getBoolean("exposeDefaultJVMMetrics", true); + // only start its own http server when prometheus http is enabled and bk http server is not enabled. + if (httpEnabled && !bkHttpServerEnabled) { + String httpAddr = conf.getString(PROMETHEUS_STATS_HTTP_ADDRESS, DEFAULT_PROMETHEUS_STATS_HTTP_ADDR); + int httpPort = conf.getInt(PROMETHEUS_STATS_HTTP_PORT, DEFAULT_PROMETHEUS_STATS_HTTP_PORT); + InetSocketAddress httpEndpoint = InetSocketAddress.createUnresolved(httpAddr, httpPort); + this.server = new Server(httpEndpoint); + ServletContextHandler context = new ServletContextHandler(); + context.setContextPath("/"); + server.setHandler(context); + + context.addServlet(new ServletHolder(new PrometheusServlet(this)), "/metrics"); + + try { + server.start(); + log.info().attr("endpoint", httpEndpoint).log("Started Prometheus stats endpoint"); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + if (exposeDefaultJVMMetrics) { + // Include standard JVM stats. Note that the metric names produced by the Prometheus Java client 1.x + // JvmMetrics differ from the ones the legacy simpleclient hotspot exports produced, for example + // jvm_memory_bytes_used is now jvm_memory_used_bytes. + registerJvmMetrics(); + + // Netty tracks direct memory allocated through unsafe, which is more accurate than the JVM's own + // accounting, so these two are exported in addition to the standard JVM metrics. + registerGaugeQuietly("jvm_memory_direct_bytes_used", + "Direct memory currently allocated by Netty", + () -> getDirectMemoryUsage.get()); + + registerGaugeQuietly("jvm_memory_direct_bytes_max", + "Maximum direct memory available to the JVM", + () -> (double) PlatformDependent.estimateMaxDirectMemory()); + } + + executor = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("metrics")); + + int latencyRolloverSeconds = conf.getInt(PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS, + DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS); + + executor.scheduleAtFixedRate(() -> { + rotateLatencyCollection(); + }, 1, latencyRolloverSeconds, TimeUnit.SECONDS); + + } + + @Override + public void stop() { + if (server != null) { + try { + server.stop(); + } catch (Exception e) { + log.warn().exception(e).log("Failed to shutdown Jetty server"); + } finally { + ThreadRegistry.clear(); + } + } + if (executor != null) { + MoreExecutors.shutdownAndAwaitTermination(executor, 5, TimeUnit.SECONDS); + } + } + + @Override + public StatsLogger getStatsLogger(String scope) { + return new PrometheusStatsLogger(PrometheusMetricsProvider.this, scope, Collections.emptyMap()); + } + + @Override + public void writeAllMetrics(Writer writer) throws IOException { + PrometheusTextFormat prometheusTextFormat = new PrometheusTextFormat(); + PrometheusTextFormat.writeMetricsCollectedByPrometheusClient(writer, registry); + + gauges.forEach((sc, gauge) -> prometheusTextFormat.writeGauge(writer, sc.getScope(), gauge)); + counters.forEach((sc, counter) -> prometheusTextFormat.writeCounter(writer, sc.getScope(), counter)); + opStats.forEach((sc, opStatLogger) -> + prometheusTextFormat.writeOpStat(writer, sc.getScope(), opStatLogger)); + } + + @Override + public String getStatsName(String... statsComponents) { + String completeName; + if (statsComponents.length == 0) { + return ""; + } else if (statsComponents[0].isEmpty()) { + completeName = StringUtils.join(statsComponents, '_', 1, statsComponents.length); + } else { + completeName = StringUtils.join(statsComponents, '_'); + } + return PrometheusNaming.sanitizeMetricName(completeName); + } + + @VisibleForTesting + void rotateLatencyCollection() { + opStats.forEach((name, metric) -> { + metric.rotateLatencyCollection(); + }); + } + + private void registerJvmMetrics() { + try { + JvmMetrics.builder().register(registry); + } catch (Exception e) { + // Ignore if these were already registered, which happens when more than one provider instance shares + // the default registry. + log.debug().exception(e).log("Failed to register JVM metrics"); + } + } + + private void registerGaugeQuietly(String name, String help, Supplier valueSupplier) { + try { + GaugeWithCallback.builder() + .name(name) + .help(help) + .callback(callback -> callback.call(valueSupplier.get())) + .register(registry); + } catch (Exception e) { + // Ignore if these were already registered + log.debug().exception(e).attr("metric", name).log("Failed to register Prometheus gauge"); + } + } + + /* + * Try to get Netty counter of used direct memory. This will be correct, unlike the JVM values. + */ + private static final AtomicLong directMemoryUsage; + private static final Optional poolMxBeanOp; + private static final Supplier getDirectMemoryUsage; + + static { + if (PlatformDependent.useDirectBufferNoCleaner()) { + poolMxBeanOp = Optional.empty(); + AtomicLong tmpDirectMemoryUsage = null; + try { + Field field = PlatformDependent.class.getDeclaredField("DIRECT_MEMORY_COUNTER"); + field.setAccessible(true); + tmpDirectMemoryUsage = (AtomicLong) field.get(null); + } catch (Throwable t) { + log.warn().exceptionMessage(t) + .log("Failed to access netty DIRECT_MEMORY_COUNTER field"); + } + directMemoryUsage = tmpDirectMemoryUsage; + getDirectMemoryUsage = () -> directMemoryUsage != null ? directMemoryUsage.get() : Double.NaN; + } else { + directMemoryUsage = null; + List platformMXBeans = ManagementFactory.getPlatformMXBeans(BufferPoolMXBean.class); + poolMxBeanOp = platformMXBeans.stream() + .filter(bufferPoolMXBean -> bufferPoolMXBean.getName().equals("direct")).findAny(); + getDirectMemoryUsage = () -> poolMxBeanOp.isPresent() + ? (double) poolMxBeanOp.get().getMemoryUsed() : Double.NaN; + } + } +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusServlet.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusServlet.java new file mode 100644 index 0000000000000..ea75ae95e881d --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusServlet.java @@ -0,0 +1,60 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import io.prometheus.metrics.expositionformats.PrometheusTextFormatWriter; +import java.io.IOException; +import java.io.Writer; +import javax.servlet.ServletException; +import javax.servlet.http.HttpServlet; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; + +/** + * Servlet used to export metrics in prometheus text format. + */ +public class PrometheusServlet extends HttpServlet { + private static final long serialVersionUID = 1L; + + private final transient PrometheusMetricsProvider provider; + + public PrometheusServlet(PrometheusMetricsProvider provider) { + this.provider = provider; + } + + @Override + protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { + resp.setStatus(HttpServletResponse.SC_OK); + resp.setContentType(PrometheusTextFormatWriter.CONTENT_TYPE); + + Writer writer = resp.getWriter(); + try { + provider.writeAllMetrics(writer); + writer.flush(); + } finally { + writer.close(); + } + } + + @Override + protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { + doGet(req, resp); + } + +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusStatsLogger.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusStatsLogger.java new file mode 100644 index 0000000000000..4b3ac2555b738 --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusStatsLogger.java @@ -0,0 +1,100 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import com.google.common.base.Joiner; +import io.prometheus.metrics.model.snapshots.PrometheusNaming; +import java.util.Map; +import java.util.TreeMap; +import org.apache.bookkeeper.stats.Counter; +import org.apache.bookkeeper.stats.Gauge; +import org.apache.bookkeeper.stats.OpStatsLogger; +import org.apache.bookkeeper.stats.StatsLogger; + +/** + * A {@code Prometheus} based {@link StatsLogger} implementation. + */ +public class PrometheusStatsLogger implements StatsLogger { + + private final PrometheusMetricsProvider provider; + private final String scope; + private final Map labels; + + PrometheusStatsLogger(PrometheusMetricsProvider provider, String scope, Map labels) { + this.provider = provider; + this.scope = scope; + this.labels = labels; + } + + @Override + public OpStatsLogger getOpStatsLogger(String name) { + return provider.opStats.computeIfAbsent(scopeContext(name), x -> new DataSketchesOpStatsLogger(labels)); + } + + @Override + public OpStatsLogger getThreadScopedOpStatsLogger(String name) { + return provider.threadScopedOpStats.computeIfAbsent(scopeContext(name), + x -> new ThreadScopedDataSketchesStatsLogger(provider, x, labels)); + } + + @Override + public Counter getCounter(String name) { + return provider.counters.computeIfAbsent(scopeContext(name), x -> new LongAdderCounter(labels)); + } + + public Counter getThreadScopedCounter(String name) { + return provider.threadScopedCounters.computeIfAbsent(scopeContext(name), + x -> new ThreadScopedLongAdderCounter(provider, x, labels)); + } + + @Override + public void registerGauge(String name, Gauge gauge) { + provider.gauges.computeIfAbsent(scopeContext(name), x -> new SimpleGauge(gauge, labels)); + } + + @Override + public void unregisterGauge(String name, Gauge gauge) { + // no-op + } + + @Override + public void removeScope(String name, StatsLogger statsLogger) { + // no-op + } + + @Override + public StatsLogger scope(String name) { + return new PrometheusStatsLogger(provider, completeName(name), labels); + } + + @Override + public StatsLogger scopeLabel(String labelName, String labelValue) { + Map newLabels = new TreeMap<>(labels); + newLabels.put(labelName, labelValue); + return new PrometheusStatsLogger(provider, scope, newLabels); + } + + private ScopeContext scopeContext(String name) { + return new ScopeContext(completeName(name), labels); + } + + private String completeName(String name) { + return PrometheusNaming.sanitizeMetricName(scope.isEmpty() ? name : Joiner.on('_').join(scope, name)); + } +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusTextFormat.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusTextFormat.java new file mode 100644 index 0000000000000..13bd79e158a72 --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/PrometheusTextFormat.java @@ -0,0 +1,199 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import io.prometheus.metrics.expositionformats.PrometheusTextFormatWriter; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import io.prometheus.metrics.model.snapshots.MetricSnapshots; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.Writer; +import java.nio.charset.StandardCharsets; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** + * Logic to write metrics in Prometheus text format. + */ +public class PrometheusTextFormat { + + Set metricNameSet = new HashSet<>(); + + void writeGauge(Writer w, String name, SimpleGauge gauge) { + // Example: + // # TYPE bookie_storage_entries_count gauge + // bookie_storage_entries_count 519 + try { + writeType(w, name, "gauge"); + w.append(name); + writeLabels(w, gauge.getLabels()); + w.append(' ').append(gauge.getSample().toString()).append('\n'); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + void writeCounter(Writer w, String name, LongAdderCounter counter) { + // Example: + // # TYPE jvm_threads_started_total counter + // jvm_threads_started_total 59 + try { + writeType(w, name, "counter"); + w.append(name); + writeLabels(w, counter.getLabels()); + w.append(' ').append(counter.get().toString()).append('\n'); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + void writeOpStat(Writer w, String name, DataSketchesOpStatsLogger opStat) { + // Example: + // # TYPE bookie_journal_JOURNAL_ADD_ENTRY summary + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.5",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.75",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.95",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.99",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.999",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.9999",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="1.0",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY_count{success="false",} 0.0 + // bookie_journal_JOURNAL_ADD_ENTRY_sum{success="false",} 0.0 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.5",} 1.706 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.75",} 1.89 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.95",} 2.121 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.99",} 10.708 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.999",} 10.902 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.9999",} 10.902 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="1.0",} 10.902 + // bookie_journal_JOURNAL_ADD_ENTRY_count{success="true",} 658.0 + // bookie_journal_JOURNAL_ADD_ENTRY_sum{success="true",} 1265.0800000000002 + try { + writeType(w, name, "summary"); + writeQuantile(w, opStat, name, false, 0.5); + writeQuantile(w, opStat, name, false, 0.75); + writeQuantile(w, opStat, name, false, 0.95); + writeQuantile(w, opStat, name, false, 0.99); + writeQuantile(w, opStat, name, false, 0.999); + writeQuantile(w, opStat, name, false, 0.9999); + writeQuantile(w, opStat, name, false, 1.0); + writeCount(w, opStat, name, false); + writeSum(w, opStat, name, false); + + writeQuantile(w, opStat, name, true, 0.5); + writeQuantile(w, opStat, name, true, 0.75); + writeQuantile(w, opStat, name, true, 0.95); + writeQuantile(w, opStat, name, true, 0.99); + writeQuantile(w, opStat, name, true, 0.999); + writeQuantile(w, opStat, name, true, 0.9999); + writeQuantile(w, opStat, name, true, 1.0); + writeCount(w, opStat, name, true); + writeSum(w, opStat, name, true); + + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private void writeLabels(Writer w, Map labels) throws IOException { + if (labels.isEmpty()) { + return; + } + + w.append('{'); + writeLabelsNoBraces(w, labels); + w.append('}'); + } + + private void writeLabelsNoBraces(Writer w, Map labels) throws IOException { + if (labels.isEmpty()) { + return; + } + + boolean isFirst = true; + for (Map.Entry e : labels.entrySet()) { + if (!isFirst) { + w.append(','); + } + isFirst = false; + w.append(e.getKey()) + .append("=\"") + .append(e.getValue()) + .append('"'); + } + } + + private void writeQuantile(Writer w, DataSketchesOpStatsLogger opStat, String name, Boolean success, + double quantile) throws IOException { + w.append(name) + .append("{success=\"").append(success.toString()) + .append("\",quantile=\"").append(Double.toString(quantile)) + .append("\""); + if (!opStat.getLabels().isEmpty()) { + w.append(", "); + writeLabelsNoBraces(w, opStat.getLabels()); + } + w.append("} ") + .append(Double.toString(opStat.getQuantileValue(success, quantile))).append('\n'); + } + + private void writeCount(Writer w, DataSketchesOpStatsLogger opStat, String name, Boolean success) + throws IOException { + w.append(name).append("_count{success=\"").append(success.toString()).append("\""); + if (!opStat.getLabels().isEmpty()) { + w.append(", "); + writeLabelsNoBraces(w, opStat.getLabels()); + } + w.append("} ") + .append(Long.toString(opStat.getCount(success))).append('\n'); + } + + private void writeSum(Writer w, DataSketchesOpStatsLogger opStat, String name, Boolean success) + throws IOException { + w.append(name).append("_sum{success=\"").append(success.toString()).append("\""); + if (!opStat.getLabels().isEmpty()) { + w.append(", "); + writeLabelsNoBraces(w, opStat.getLabels()); + } + w.append("} ") + .append(Double.toString(opStat.getSum(success))).append('\n'); + } + + /** + * Writes everything registered in the Prometheus Java client registry, which is where the JVM metrics and the + * Netty direct-memory gauges live. The client library's own text format writer is used here rather than a + * hand-rolled one so that every snapshot type is rendered exactly as the Prometheus exposition format specifies. + */ + static void writeMetricsCollectedByPrometheusClient(Writer w, PrometheusRegistry registry) throws IOException { + MetricSnapshots snapshots = registry.scrape(); + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + PrometheusTextFormatWriter.create().write(buffer, snapshots); + w.write(buffer.toString(StandardCharsets.UTF_8)); + } + + void writeType(Writer w, String name, String type) throws IOException { + if (metricNameSet.contains(name)) { + return; + } + metricNameSet.add(name); + w.append("# TYPE ").append(name).append(" ").append(type).append("\n"); + } + +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ScopeContext.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ScopeContext.java new file mode 100644 index 0000000000000..4f15c006569ef --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ScopeContext.java @@ -0,0 +1,56 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import java.util.Map; +import java.util.Objects; + +/** + * Holder for a scope and a set of associated labels. + */ +public class ScopeContext { + private final String scope; + private final Map labels; + + public ScopeContext(String scope, Map labels) { + this.scope = scope; + this.labels = labels; + } + + public String getScope() { + return scope; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ScopeContext that = (ScopeContext) o; + return Objects.equals(scope, that.scope) && Objects.equals(labels, that.labels); + } + + @Override + public int hashCode() { + return Objects.hash(scope, labels); + } +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/SimpleGauge.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/SimpleGauge.java new file mode 100644 index 0000000000000..384cc682c2ff3 --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/SimpleGauge.java @@ -0,0 +1,44 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import java.util.Map; +import org.apache.bookkeeper.stats.Gauge; + +/** + * A {@link Gauge} implementation that forwards on the value supplier. + */ +public class SimpleGauge { + + private final Map labels; + private final Gauge gauge; + + public SimpleGauge(final Gauge gauge, Map labels) { + this.gauge = gauge; + this.labels = labels; + } + + Number getSample() { + return gauge.getSample(); + } + + public Map getLabels() { + return labels; + } +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ThreadScopedDataSketchesStatsLogger.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ThreadScopedDataSketchesStatsLogger.java new file mode 100644 index 0000000000000..e1850262a5115 --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ThreadScopedDataSketchesStatsLogger.java @@ -0,0 +1,137 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import lombok.CustomLog; +import org.apache.bookkeeper.stats.OpStatsData; +import org.apache.bookkeeper.stats.OpStatsLogger; +import org.apache.bookkeeper.stats.ThreadRegistry; + +/** + * OpStatsLogger implementation that lazily registers OpStatsLoggers per thread + * with added labels for the threadpool/thresd name and thread no. + */ +@CustomLog +public class ThreadScopedDataSketchesStatsLogger implements OpStatsLogger { + private ThreadLocal statsLoggers; + private DataSketchesOpStatsLogger defaultStatsLogger; + private Map originalLabels; + private ScopeContext scopeContext; + private PrometheusMetricsProvider provider; + + public ThreadScopedDataSketchesStatsLogger(PrometheusMetricsProvider provider, + ScopeContext scopeContext, + Map labels) { + this.provider = provider; + this.scopeContext = scopeContext; + this.originalLabels = labels; + this.defaultStatsLogger = new DataSketchesOpStatsLogger(labels); + + Map defaultLabels = new HashMap<>(labels); + defaultLabels.put("threadPool", "?"); + defaultLabels.put("thread", "?"); + this.defaultStatsLogger.initializeThread(defaultLabels); + + this.statsLoggers = ThreadLocal.withInitial(() -> { + return new DataSketchesOpStatsLogger(labels); + }); + } + + @Override + public void registerFailedEvent(long eventLatency, TimeUnit unit) { + getStatsLogger().registerFailedEvent(eventLatency, unit); + } + + @Override + public void registerSuccessfulEvent(long eventLatency, TimeUnit unit) { + getStatsLogger().registerSuccessfulEvent(eventLatency, unit); + } + + @Override + public void registerSuccessfulValue(long value) { + getStatsLogger().registerSuccessfulValue(value); + } + + @Override + public void registerFailedValue(long value) { + getStatsLogger().registerFailedValue(value); + } + + @Override + public OpStatsData toOpStatsData() { + // Not relevant as we don't use JMX here + throw new UnsupportedOperationException(); + } + + @Override + public void clear() { + // Not relevant as we don't use JMX here + throw new UnsupportedOperationException(); + } + + private DataSketchesOpStatsLogger getStatsLogger() { + DataSketchesOpStatsLogger statsLogger = statsLoggers.get(); + + // Lazy registration + // Update the stats logger with the thread labels then add to the provider + // If for some reason this thread did not get registered, + // then we fallback to a standard OpsStatsLogger (defaultStatsLogger) + if (!statsLogger.isThreadInitialized()) { + ThreadRegistry.ThreadPoolThread tpt = ThreadRegistry.get(); + if (tpt == null) { + log.warn().attr("thread", Thread.currentThread()) + .attr("defaultStatsLogger", defaultStatsLogger) + .log("Thread was not registered in the thread registry. Using default stats logger"); + statsLoggers.set(defaultStatsLogger); + DataSketchesOpStatsLogger previous = provider.opStats + .put(new ScopeContext(scopeContext.getScope(), originalLabels), defaultStatsLogger); + // If we overwrite a logger, metrics will not be collected correctly + if (previous != null && previous != defaultStatsLogger) { + log.error().attr("thread", Thread.currentThread()) + .attr("newLogger", defaultStatsLogger) + .attr("previousLogger", previous) + .log("Invalid state. Overwrote a stats logger"); + throw new IllegalStateException("Invalid state. Overwrote a stats logger."); + } + return defaultStatsLogger; + } else { + Map threadScopedlabels = new HashMap<>(originalLabels); + threadScopedlabels.put("threadPool", tpt.getThreadPool()); + threadScopedlabels.put("thread", String.valueOf(tpt.getOrdinal())); + + statsLogger.initializeThread(threadScopedlabels); + DataSketchesOpStatsLogger previous = provider.opStats + .put(new ScopeContext(scopeContext.getScope(), threadScopedlabels), statsLogger); + // If we overwrite a logger, metrics will not be collected correctly + if (previous != null && previous != statsLogger) { + log.error().attr("thread", Thread.currentThread()) + .attr("newLogger", defaultStatsLogger) + .attr("previousLogger", previous) + .log("Invalid state. Overwrote a stats logger"); + throw new IllegalStateException("Invalid state. Overwrote a stats logger."); + } + } + } + + return statsLogger; + } +} \ No newline at end of file diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ThreadScopedLongAdderCounter.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ThreadScopedLongAdderCounter.java new file mode 100644 index 0000000000000..1c2287f5b9c85 --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/ThreadScopedLongAdderCounter.java @@ -0,0 +1,111 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.stats.Counter; +import org.apache.bookkeeper.stats.ThreadRegistry; + +/** + * {@link Counter} implementation that lazily registers LongAdderCounters per thread + * * with added labels for the threadpool/thread name and thread no. + */ +public class ThreadScopedLongAdderCounter implements Counter { + private ThreadLocal counters; + private LongAdderCounter defaultCounter; + private Map originalLabels; + private ScopeContext scopeContext; + private PrometheusMetricsProvider provider; + + public ThreadScopedLongAdderCounter(PrometheusMetricsProvider provider, + ScopeContext scopeContext, + Map labels) { + this.provider = provider; + this.scopeContext = scopeContext; + this.originalLabels = new HashMap<>(labels); + this.defaultCounter = new LongAdderCounter(labels); + Map defaultLabels = new HashMap<>(labels); + defaultLabels.put("threadPool", "?"); + defaultLabels.put("thread", "?"); + this.defaultCounter.initializeThread(defaultLabels); + + this.counters = ThreadLocal.withInitial(() -> { + return new LongAdderCounter(labels); + }); + } + + @Override + public void clear() { + getCounter().clear(); + } + + @Override + public void inc() { + getCounter().inc(); + } + + @Override + public void dec() { + getCounter().dec(); + } + + @Override + public void addCount(long delta) { + getCounter().addCount(delta); + } + + @Override + public void addLatency(long eventLatency, TimeUnit unit) { + getCounter().addLatency(eventLatency, unit); + } + + @Override + public Long get() { + return getCounter().get(); + } + + private LongAdderCounter getCounter() { + LongAdderCounter counter = counters.get(); + + // Lazy registration + // Update the counter with the thread labels then add to the provider + // If for some reason this thread did not get registered, + // then we fallback to a standard counter (defaultCounter) + if (!counter.isThreadInitialized()) { + ThreadRegistry.ThreadPoolThread tpt = ThreadRegistry.get(); + + if (tpt == null) { + counters.set(defaultCounter); + provider.counters.put(new ScopeContext(scopeContext.getScope(), originalLabels), defaultCounter); + return defaultCounter; + } else { + Map threadScopedlabels = new HashMap<>(originalLabels); + threadScopedlabels.put("threadPool", tpt.getThreadPool()); + threadScopedlabels.put("thread", String.valueOf(tpt.getOrdinal())); + + counter.initializeThread(threadScopedlabels); + provider.counters.put(new ScopeContext(scopeContext.getScope(), threadScopedlabels), counter); + } + } + + return counter; + } +} diff --git a/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/package-info.java b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/package-info.java new file mode 100644 index 0000000000000..6bd1f9d6bf0d3 --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/main/java/org/apache/pulsar/metrics/prometheus/bookkeeper/package-info.java @@ -0,0 +1,23 @@ +/* + * 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. + */ + +/** + * A stats provider implementation for BookKeeper that uses Jetty 12. + */ +package org.apache.pulsar.metrics.prometheus.bookkeeper; diff --git a/bookkeeper-prometheus-metrics-provider/src/test/java/org/apache/pulsar/metrics/prometheus/bookkeeper/TestPrometheusFormatter.java b/bookkeeper-prometheus-metrics-provider/src/test/java/org/apache/pulsar/metrics/prometheus/bookkeeper/TestPrometheusFormatter.java new file mode 100644 index 0000000000000..5d9fa92cc47e4 --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/test/java/org/apache/pulsar/metrics/prometheus/bookkeeper/TestPrometheusFormatter.java @@ -0,0 +1,297 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import static com.google.common.base.Preconditions.checkArgument; +import static org.testng.AssertJUnit.assertEquals; +import static org.testng.AssertJUnit.assertTrue; +import static org.testng.AssertJUnit.fail; +import com.google.common.base.MoreObjects; +import com.google.common.base.Splitter; +import com.google.common.collect.ArrayListMultimap; +import com.google.common.collect.Multimap; +import io.prometheus.metrics.core.metrics.GaugeWithCallback; +import io.prometheus.metrics.instrumentation.jvm.JvmMetrics; +import io.prometheus.metrics.model.registry.PrometheusRegistry; +import java.io.IOException; +import java.io.StringWriter; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.TimeUnit; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.apache.bookkeeper.stats.Counter; +import org.apache.bookkeeper.stats.OpStatsLogger; +import org.apache.bookkeeper.stats.StatsLogger; +import org.testng.annotations.Test; + +/** + * Test for {@link PrometheusMetricsProvider}. + */ +public class TestPrometheusFormatter { + + @Test(timeOut = 30000) + public void testStatsOutput() throws Exception { + PrometheusMetricsProvider provider = new PrometheusMetricsProvider(); + StatsLogger statsLogger = provider.getStatsLogger("test"); + Counter counter = statsLogger.getCounter("my_counter"); + + counter.inc(); + counter.inc(); + + OpStatsLogger opStats = statsLogger.getOpStatsLogger("op"); + opStats.registerSuccessfulEvent(10, TimeUnit.MILLISECONDS); + opStats.registerSuccessfulEvent(5, TimeUnit.MILLISECONDS); + + OpStatsLogger opStats1 = statsLogger.scopeLabel("test_label", "test_value") + .getOpStatsLogger("op_label"); + opStats1.registerSuccessfulEvent(10, TimeUnit.MILLISECONDS); + opStats1.registerSuccessfulEvent(5, TimeUnit.MILLISECONDS); + opStats1.registerFailedEvent(1, TimeUnit.MILLISECONDS); + + provider.rotateLatencyCollection(); + + StringWriter writer = new StringWriter(); + provider.writeAllMetrics(writer); + + writer.write("jvm_memory_direct_bytes_max{} 4.77626368E8\n"); + writer.write("jvm_memory_pool_bytes_used{pool=\"Code Cache\"} 3347712.0\n"); + writer.write("jvm_memory_pool_bytes_used{pool=\"CodeHeap 'non-nmethods'\"} 1207168.0\n"); + System.out.println(writer); + Multimap metrics = parseMetrics(writer.toString()); + System.out.println(metrics); + + List cm = (List) metrics.get("test_my_counter"); + assertEquals(1, cm.size()); + assertEquals(0, cm.get(0).tags.size()); + assertEquals(2.0, cm.get(0).value, 0.0); + + // test_op_sum + cm = (List) metrics.get("test_op_sum"); + assertEquals(2, cm.size()); + Metric m = cm.get(0); + assertEquals(1, cm.get(0).tags.size()); + assertEquals(0.0, m.value, 0.0); + assertEquals(1, m.tags.size()); + assertEquals("false", m.tags.get("success")); + + m = cm.get(1); + assertEquals(1, cm.get(0).tags.size()); + assertEquals(15.0, m.value, 0.0); + assertEquals(1, m.tags.size()); + assertEquals("true", m.tags.get("success")); + + // test_op_count + cm = (List) metrics.get("test_op_count"); + assertEquals(2, cm.size()); + m = cm.get(0); + assertEquals(1, cm.get(0).tags.size()); + assertEquals(0.0, m.value, 0.0); + assertEquals(1, m.tags.size()); + assertEquals("false", m.tags.get("success")); + + m = cm.get(1); + assertEquals(1, cm.get(0).tags.size()); + assertEquals(2.0, m.value, 0.0); + assertEquals(1, m.tags.size()); + assertEquals("true", m.tags.get("success")); + + // Latency + cm = (List) metrics.get("test_op"); + assertEquals(14, cm.size()); + + boolean found = false; + for (Metric mt : cm) { + if ("true".equals(mt.tags.get("success")) && "1.0".equals(mt.tags.get("quantile"))) { + assertEquals(10.0, mt.value, 0.0); + found = true; + } + } + + assertTrue(found); + + // test_op_label_sum + cm = (List) metrics.get("test_op_label_sum"); + assertEquals(2, cm.size()); + m = cm.get(0); + assertEquals(2, m.tags.size()); + assertEquals(1.0, m.value, 0.0); + assertEquals("false", m.tags.get("success")); + assertEquals("test_value", m.tags.get("test_label")); + + m = cm.get(1); + assertEquals(15.0, m.value, 0.0); + assertEquals(2, m.tags.size()); + assertEquals("true", m.tags.get("success")); + assertEquals("test_value", m.tags.get("test_label")); + + // test_op_label_count + cm = (List) metrics.get("test_op_label_count"); + assertEquals(2, cm.size()); + m = cm.get(0); + assertEquals(1, m.value, 0.0); + assertEquals(2, m.tags.size()); + assertEquals("false", m.tags.get("success")); + assertEquals("test_value", m.tags.get("test_label")); + + m = cm.get(1); + assertEquals(2.0, m.value, 0.0); + assertEquals(2, m.tags.size()); + assertEquals("true", m.tags.get("success")); + assertEquals("test_value", m.tags.get("test_label")); + + // Latency + cm = (List) metrics.get("test_op_label"); + assertEquals(14, cm.size()); + + found = false; + for (Metric mt : cm) { + if ("true".equals(mt.tags.get("success")) + && "test_value".equals(mt.tags.get("test_label")) + && "1.0".equals(mt.tags.get("quantile"))) { + assertEquals(10.0, mt.value, 0.0); + found = true; + } + } + + assertTrue(found); + } + + @Test + public void testWriteMetricsCollectedByPrometheusClient() { + // A dedicated registry rather than the default one, so this test does not collide with any other test that + // registers the same JVM metrics. + PrometheusRegistry registry = new PrometheusRegistry(); + JvmMetrics.builder().register(registry); + GaugeWithCallback.builder() + .name("jvm_memory_direct_bytes_used") + .help("-") + .callback(callback -> callback.call(1.0)) + .register(registry); + GaugeWithCallback.builder() + .name("jvm_memory_direct_bytes_max") + .help("-") + .callback(callback -> callback.call(100.0)) + .register(registry); + PrometheusMetricsProvider provider = new PrometheusMetricsProvider(registry); + StringWriter writer = new StringWriter(); + try { + provider.rotateLatencyCollection(); + provider.writeAllMetrics(writer); + String output = writer.toString(); + parseMetrics(output); + assertTrue(output.contains("# TYPE jvm_memory_direct_bytes_max gauge")); + assertTrue(output.contains("# TYPE jvm_memory_direct_bytes_used gauge")); + assertTrue(output.contains("# TYPE jvm_gc_collection_seconds summary")); + // Prometheus Java client 1.x renamed several JVM metrics relative to the legacy simpleclient + // hotspot exports: jvm_memory_pool_bytes_committed is now jvm_memory_pool_committed_bytes, and + // the process counter is exposed under its full process_cpu_seconds_total name. + assertTrue(output.contains("# TYPE jvm_memory_pool_committed_bytes gauge")); + assertTrue(output.contains("# TYPE jvm_memory_used_bytes gauge")); + assertTrue(output.contains("# TYPE process_cpu_seconds_total counter")); + } catch (Exception e) { + fail(); + } + + } + + @Test + public void testPrometheusTypeDuplicate() throws IOException { + PrometheusTextFormat prometheusTextFormat = new PrometheusTextFormat(); + StringWriter writer = new StringWriter(); + prometheusTextFormat.writeType(writer, "counter", "gauge"); + prometheusTextFormat.writeType(writer, "counter", "gauge"); + String string = writer.toString(); + assertEquals("# TYPE counter gauge\n", string); + } + + + /** + * Hacky parsing of Prometheus text format. Sould be good enough for unit tests + */ + private static Multimap parseMetrics(String metrics) { + Multimap parsed = ArrayListMultimap.create(); + + // Example of lines are + // jvm_threads_current{cluster="standalone",} 203.0 + // or + // pulsar_subscriptions_count{cluster="standalone", namespace="sample/standalone/ns1", + // topic="persistent://sample/standalone/ns1/test-2"} 0.0 1517945780897 + // Values may be negative and may carry an exponent, e.g. 1.0E-5. + Pattern pattern = Pattern.compile("^(\\w+)(\\{([^\\}]*)\\})?\\s(-?[\\d\\w\\.+-]+)(\\s(\\d+))?$"); + // Label values are quoted strings and may contain anything except a quote. jvm_runtime_info, for + // example, reports values such as "OpenJDK Runtime Environment" and "21.0.11+10-LTS". + Pattern formatPattern = + Pattern.compile("^(\\w+)(\\{(\\w+=\"[^\"]*\"(,\\s?\\w+=\"[^\"]*\")*)?\\})?" + + "\\s(-?[\\d\\w\\.+-]+)(\\s(\\d+))?$"); + Pattern tagsPattern = Pattern.compile("(\\w+)=\"([^\"]+)\"(,\\s?)?"); + + Splitter.on("\n").split(metrics).forEach(line -> { + if (line.isEmpty() || line.startsWith("#")) { + return; + } + + System.err.println("LINE: '" + line + "'"); + Matcher matcher = pattern.matcher(line); + Matcher formatMatcher = formatPattern.matcher(line); + System.err.println("Matches: " + matcher.matches()); + System.err.println(matcher); + assertTrue(matcher.matches()); + assertTrue("failed to validate line: " + line, formatMatcher.matches()); + + assertEquals(6, matcher.groupCount()); + System.err.println("groups: " + matcher.groupCount()); + for (int i = 0; i < matcher.groupCount(); i++) { + System.err.println(" GROUP " + i + " -- " + matcher.group(i)); + } + + checkArgument(matcher.matches()); + checkArgument(formatMatcher.matches()); + String name = matcher.group(1); + + Metric m = new Metric(); + m.value = Double.parseDouble(matcher.group(4)); + + String tags = matcher.group(3); + if (tags != null) { + Matcher tagsMatcher = tagsPattern.matcher(tags); + while (tagsMatcher.find()) { + String tag = tagsMatcher.group(1); + String value = tagsMatcher.group(2); + m.tags.put(tag, value); + } + } + + parsed.put(name, m); + }); + + return parsed; + } + + static class Metric { + Map tags = new TreeMap<>(); + double value; + + @Override + public String toString() { + return MoreObjects.toStringHelper(this).add("tags", tags).add("value", value).toString(); + } + } +} diff --git a/bookkeeper-prometheus-metrics-provider/src/test/java/org/apache/pulsar/metrics/prometheus/bookkeeper/TestPrometheusMetricsProvider.java b/bookkeeper-prometheus-metrics-provider/src/test/java/org/apache/pulsar/metrics/prometheus/bookkeeper/TestPrometheusMetricsProvider.java new file mode 100644 index 0000000000000..e9c4b4570038a --- /dev/null +++ b/bookkeeper-prometheus-metrics-provider/src/test/java/org/apache/pulsar/metrics/prometheus/bookkeeper/TestPrometheusMetricsProvider.java @@ -0,0 +1,169 @@ +/* + * 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.pulsar.metrics.prometheus.bookkeeper; + +import static org.testng.Assert.assertNotEquals; +import static org.testng.AssertJUnit.assertEquals; +import static org.testng.AssertJUnit.assertNotNull; +import static org.testng.AssertJUnit.assertNull; +import static org.testng.AssertJUnit.assertSame; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import java.io.StringWriter; +import java.util.Collections; +import java.util.HashMap; +import java.util.concurrent.TimeUnit; +import lombok.Cleanup; +import org.apache.bookkeeper.stats.Counter; +import org.apache.bookkeeper.stats.StatsLogger; +import org.apache.commons.configuration2.PropertiesConfiguration; +import org.testng.annotations.Test; + +/** + * Unit test of {@link PrometheusMetricsProvider}. + */ +public class TestPrometheusMetricsProvider { + + @Test + public void testStartNoHttp() { + PropertiesConfiguration config = new PropertiesConfiguration(); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_ENABLE, false); + PrometheusMetricsProvider provider = new PrometheusMetricsProvider(); + try { + provider.start(config); + assertNull(provider.server); + } finally { + provider.stop(); + } + } + + @Test + public void testStartNoHttpWhenBkHttpEnabled() { + PropertiesConfiguration config = new PropertiesConfiguration(); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_ENABLE, true); + config.setProperty("httpServerEnabled", true); + @Cleanup("stop") PrometheusMetricsProvider provider = new PrometheusMetricsProvider(); + provider.start(config); + assertNull(provider.server); + } + + @Test + public void testStartWithHttp() { + PropertiesConfiguration config = new PropertiesConfiguration(); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_ENABLE, true); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_PORT, 0); // ephemeral + PrometheusMetricsProvider provider = new PrometheusMetricsProvider(); + try { + provider.start(config); + assertNotNull(provider.server); + } finally { + provider.stop(); + } + } + + @Test + public void testStartWithHttpSpecifyAddr() { + PropertiesConfiguration config = new PropertiesConfiguration(); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_ENABLE, true); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_PORT, 0); // ephemeral + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_ADDRESS, "127.0.0.1"); + PrometheusMetricsProvider provider = new PrometheusMetricsProvider(); + try { + provider.start(config); + assertNotNull(provider.server); + } finally { + provider.stop(); + } + } + + @Test + public void testCounter() { + LongAdderCounter counter = new LongAdderCounter(Collections.emptyMap()); + long value = counter.get(); + assertEquals(0L, value); + counter.inc(); + assertEquals(1L, counter.get().longValue()); + counter.dec(); + assertEquals(0L, counter.get().longValue()); + counter.addCount(3); + assertEquals(3L, counter.get().longValue()); + } + + @Test + public void testCounter2() { + LongAdderCounter counter = new LongAdderCounter(Collections.emptyMap()); + long value = counter.get(); + assertEquals(0L, value); + counter.addLatency(3 * 1000 * 1000L, TimeUnit.NANOSECONDS); + assertEquals(3L, counter.get().longValue()); + } + + @Test + public void testTwoCounters() throws Exception { + PrometheusMetricsProvider provider = new PrometheusMetricsProvider(); + StatsLogger statsLogger = provider.getStatsLogger("test"); + + Counter counter1 = statsLogger.getCounter("counter"); + Counter counter2 = statsLogger.getCounter("counter"); + assertEquals(counter1, counter2); + assertSame(counter1, counter2); + + assertEquals(1, provider.counters.size()); + } + + @Test + public void testJvmDirectMemoryMetrics() throws Exception { + PropertiesConfiguration config = new PropertiesConfiguration(); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_ENABLE, true); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_PORT, 0); + config.setProperty(PrometheusMetricsProvider.PROMETHEUS_STATS_HTTP_ADDRESS, "127.0.0.1"); + ByteBuf byteBuf = ByteBufAllocator.DEFAULT.directBuffer(25); + PrometheusMetricsProvider provider = new PrometheusMetricsProvider(); + try { + provider.start(config); + assertNotNull(provider.server); + StringWriter writer = new StringWriter(); + provider.writeAllMetrics(writer); + String s = writer.toString(); + String[] split = s.split(System.lineSeparator()); + HashMap map = new HashMap<>(); + for (String str : split) { + String[] aux = str.split(" "); + map.put(aux[0], aux[1]); + } + // The Prometheus Java client 1.x writer omits the empty label set that the legacy simpleclient + // rendered, so these lines are now "jvm_memory_direct_bytes_max " rather than + // "jvm_memory_direct_bytes_max{} ". + String directBytesMax = map.get("jvm_memory_direct_bytes_max"); + assertNotNull(directBytesMax); + assertNotEquals("Nan", directBytesMax); + assertNotEquals("-1", directBytesMax); + String directBytesUsed = map.get("jvm_memory_direct_bytes_used"); + assertNotNull(directBytesUsed); + assertNotEquals("Nan", directBytesUsed); + // this condition is flaky + //assertTrue(Double.parseDouble(directBytesUsed) > 25); + // ensure byteBuffer doesn't gc + byteBuf.release(); + } finally { + provider.stop(); + } + } + +} diff --git a/conf/bookkeeper.conf b/conf/bookkeeper.conf index df433612e8f03..b13cb7ed213d8 100644 --- a/conf/bookkeeper.conf +++ b/conf/bookkeeper.conf @@ -595,7 +595,7 @@ compactionRateByBytes=1000000 # enableStatistics=true # Stats Provider Class (if statistics are enabled) -statsProviderClass=org.apache.bookkeeper.stats.prometheus.PrometheusMetricsProvider +statsProviderClass=org.apache.pulsar.metrics.prometheus.bookkeeper.PrometheusMetricsProvider # Default port for Prometheus metrics exporter prometheusStatsHttpPort=8000 diff --git a/distribution/server/build.gradle.kts b/distribution/server/build.gradle.kts index 1963b41cbfbe1..b614cfe75bc93 100644 --- a/distribution/server/build.gradle.kts +++ b/distribution/server/build.gradle.kts @@ -103,7 +103,9 @@ dependencies { distLib(project(":pulsar-broker-auth-oidc")) distLib(project(":pulsar-broker-auth-sasl")) distLib(project(":pulsar-client-auth-sasl")) - distLib(libs.bookkeeper.prometheus.metrics.provider) + // Pulsar's own BookKeeper stats provider, built on the Prometheus Java client 1.x. It replaces + // org.apache.bookkeeper.stats:prometheus-metrics-provider, which is still on the legacy simpleclient. + distLib(project(":pulsar-bookkeeper-prometheus-metrics-provider")) distLib(project(":pulsar-package-management:pulsar-package-bookkeeper-storage")) { exclude(group = "org.objenesis") } diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt index 02abeeaf3749f..4735bb727fd70 100644 --- a/distribution/server/src/assemble/LICENSE.bin.txt +++ b/distribution/server/src/assemble/LICENSE.bin.txt @@ -339,11 +339,17 @@ The Apache Software License, Version 2.0 - io.prometheus-simpleclient_tracer_otel_agent-0.16.0.jar * Prometheus exporter - io.prometheus-prometheus-metrics-config-1.8.0.jar + - io.prometheus-prometheus-metrics-core-1.8.0.jar - io.prometheus-prometheus-metrics-exporter-common-1.8.0.jar - io.prometheus-prometheus-metrics-exporter-httpserver-1.8.0.jar - io.prometheus-prometheus-metrics-exposition-formats-no-protobuf-1.8.0.jar - io.prometheus-prometheus-metrics-exposition-textformats-1.8.0.jar + - io.prometheus-prometheus-metrics-instrumentation-jvm-1.8.0.jar - io.prometheus-prometheus-metrics-model-1.8.0.jar + - io.prometheus-prometheus-metrics-tracer-common-1.8.0.jar + - io.prometheus-prometheus-metrics-tracer-initializer-1.8.0.jar + - io.prometheus-prometheus-metrics-tracer-otel-1.8.0.jar + - io.prometheus-prometheus-metrics-tracer-otel-agent-1.8.0.jar * Jakarta Bean Validation API - jakarta.validation-jakarta.validation-api-3.0.2.jar - javax.validation-validation-api-1.1.0.Final.jar @@ -376,7 +382,6 @@ The Apache Software License, Version 2.0 - org.apache.bookkeeper.http-http-server-4.18.0.jar - org.apache.bookkeeper.http-vertx-http-server-4.18.0.jar - org.apache.bookkeeper.stats-bookkeeper-stats-api-4.18.0.jar - - org.apache.bookkeeper.stats-prometheus-metrics-provider-4.18.0.jar - org.apache.distributedlog-distributedlog-common-4.18.0.jar - org.apache.distributedlog-distributedlog-core-4.18.0-tests.jar - org.apache.distributedlog-distributedlog-core-4.18.0.jar diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 8e052c9b823fd..0413f195cb54b 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -91,8 +91,12 @@ aircompressor = "2.0.3" completable-futures = "0.3.6" re2j = "1.8" # Metrics / Observability +# Legacy Prometheus simpleclient (client_java 0.x). Still required by BookKeeper's and ZooKeeper's +# stats providers and by Pulsar's own broker/proxy/functions metrics, which have not been migrated yet. prometheus = "0.16.0" prometheus-jmx = "0.16.1" +# Prometheus Java client 1.x (the successor to simpleclient) +prometheus-client-java = "1.8.0" dropwizardmetrics = "4.2.39" hdrHistogram = "2.2.2" perfmark = "0.27.0" @@ -313,6 +317,11 @@ simpleclient-servlet = { module = "io.prometheus:simpleclient_servlet", version. simpleclient-common = { module = "io.prometheus:simpleclient_common", version.ref = "prometheus" } simpleclient-log4j2 = { module = "io.prometheus:simpleclient_log4j2", version.ref = "prometheus" } prometheus-jmx-collector = { module = "io.prometheus.jmx:collector", version.ref = "prometheus-jmx" } +# Prometheus Java client 1.x +prometheus-metrics-core = { module = "io.prometheus:prometheus-metrics-core", version.ref = "prometheus-client-java" } +prometheus-metrics-model = { module = "io.prometheus:prometheus-metrics-model", version.ref = "prometheus-client-java" } +prometheus-metrics-instrumentation-jvm = { module = "io.prometheus:prometheus-metrics-instrumentation-jvm", version.ref = "prometheus-client-java" } +prometheus-metrics-exposition-textformats = { module = "io.prometheus:prometheus-metrics-exposition-textformats", version.ref = "prometheus-client-java" } # OpenTelemetry opentelemetry-bom = { module = "io.opentelemetry:opentelemetry-bom", version.ref = "opentelemetry" } opentelemetry-bom-alpha = { module = "io.opentelemetry:opentelemetry-bom-alpha", version.ref = "opentelemetry-alpha" } diff --git a/settings.gradle.kts b/settings.gradle.kts index 28be05724f712..c96c26c2d7990 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -173,6 +173,10 @@ include("pulsar-broker") include("pulsar-package-management:pulsar-package-bookkeeper-storage") project(":pulsar-package-management:pulsar-package-bookkeeper-storage").projectDir = file("pulsar-package-management/bookkeeper-storage") +// Tier 6.5 — BookKeeper stats provider on the Prometheus Java client 1.x +include("pulsar-bookkeeper-prometheus-metrics-provider") +project(":pulsar-bookkeeper-prometheus-metrics-provider").projectDir = file("bookkeeper-prometheus-metrics-provider") + // Tier 6.5 — jetty upgrade modules include("jetty-upgrade:pulsar-zookeeper-prometheus-metrics") project(":jetty-upgrade:pulsar-zookeeper-prometheus-metrics").projectDir = file("jetty-upgrade/zookeeper-prometheus-metrics")