[improve][monitor] Restore the BookKeeper Prometheus stats provider on the Prometheus Java client 1.x

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)
+}

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<String, String> labels;
+
+    // used for lazy registration for thread scoped metrics
+    private boolean threadInitialized;
+
+    public DataSketchesOpStatsLogger(Map<String, String> 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<String, String> getLabels() {
+        return labels;
+    }
+
+    public boolean isThreadInitialized() {
+        return threadInitialized;
+    }
+
+    public void initializeThread(Map<String, String> 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<LocalData, Boolean> map = new ConcurrentHashMap<>();
+        private final FastThreadLocal<LocalData> localData = new FastThreadLocal<LocalData>() {
+
+            @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) + "}";
+    }
+}

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}.
+ *
+ * <p>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<String, String> labels;
+
+    // used for lazy registration for thread scoped metric
+    private boolean threadInitialized;
+
+    public LongAdderCounter(Map<String, String> 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<String, String> getLabels() {
+        return labels;
+    }
+
+    public boolean isThreadInitialized() {
+        return threadInitialized;
+    }
+
+    public void initializeThread(Map<String, String> labels) {
+        this.labels = labels;
+        this.threadInitialized = true;
+    }
+}