diff --git a/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt b/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt index ffb4a1d52..136f7f6f1 100644 --- a/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt +++ b/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt @@ -1,4 +1,6 @@ Comparing source compatibility of prometheus-metrics-core-1.8.1-SNAPSHOT.jar against prometheus-metrics-core-1.8.0.jar *** MODIFIED CLASS: PUBLIC io.prometheus.metrics.core.exemplars.ExemplarSampler (not serializable) === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 +*** MODIFIED CLASS: PUBLIC io.prometheus.metrics.core.metrics.Histogram$DataPoint (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java new file mode 100644 index 000000000..e4f5b30ff --- /dev/null +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java @@ -0,0 +1,120 @@ +package io.prometheus.metrics.core.metrics; + +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Experimental accumulator for classic-only histogram data points. + * + *

Each recording thread owns a cell with two buffers. A snapshot advances the global epoch, + * waits only for observations that had already entered the previous epoch, and then drains the + * inactive buffers. Recording threads therefore never contend on a shared monitor. + * + *

Cells are retained for the lifetime of the data point so observations made by short-lived + * threads remain available to later snapshots. A cell is static and does not reference its owning + * accumulator, so a thread-local value cannot retain a removed or cleared data point. + */ +@SuppressWarnings("ThreadLocalUsage") +final class ClassicOnlyAccumulator { + + private static final long NOT_WRITING = -1; + + private final int bucketCount; + private final AtomicLong epoch = new AtomicLong(); + private final ConcurrentLinkedQueue cells = new ConcurrentLinkedQueue<>(); + private final ThreadLocal threadCell = + new ThreadLocal() { + @Override + protected Cell initialValue() { + Cell cell = new Cell(bucketCount); + cells.add(cell); + return cell; + } + }; + + // Accessed only while holding this accumulator's monitor. + private final long[] collectedBuckets; + private long collectedCount; + private double collectedSum; + + ClassicOnlyAccumulator(int bucketCount) { + this.bucketCount = bucketCount; + this.collectedBuckets = new long[bucketCount]; + } + + void observe(int bucket, double value) { + Cell cell = threadCell.get(); + while (true) { + long observedEpoch = epoch.get(); + cell.writingEpoch = observedEpoch; + if (epoch.get() != observedEpoch) { + cell.writingEpoch = NOT_WRITING; + continue; + } + try { + CellBuffer buffer = cell.buffers[(int) (observedEpoch & 1)]; + buffer.buckets[bucket]++; + buffer.sum += value; + buffer.count++; + return; + } finally { + // Publishes all plain writes above to a snapshot waiting on writingEpoch. + cell.writingEpoch = NOT_WRITING; + } + } + } + + @SuppressWarnings("ThreadPriorityCheck") + synchronized Snapshot snapshot() { + long inactiveEpoch = epoch.getAndIncrement(); + int inactiveBuffer = (int) (inactiveEpoch & 1); + + for (Cell cell : cells) { + while (cell.writingEpoch == inactiveEpoch) { + Thread.yield(); + } + CellBuffer buffer = cell.buffers[inactiveBuffer]; + for (int i = 0; i < bucketCount; i++) { + collectedBuckets[i] += buffer.buckets[i]; + buffer.buckets[i] = 0; + } + collectedCount += buffer.count; + collectedSum += buffer.sum; + buffer.count = 0; + buffer.sum = 0; + } + + return new Snapshot(collectedBuckets.clone(), collectedCount, collectedSum); + } + + private static final class Cell { + private final CellBuffer[] buffers; + private volatile long writingEpoch = NOT_WRITING; + + private Cell(int bucketCount) { + buffers = new CellBuffer[] {new CellBuffer(bucketCount), new CellBuffer(bucketCount)}; + } + } + + private static final class CellBuffer { + private final long[] buckets; + private long count; + private double sum; + + private CellBuffer(int bucketCount) { + buckets = new long[bucketCount]; + } + } + + static final class Snapshot { + final long[] buckets; + final long count; + final double sum; + + private Snapshot(long[] buckets, long count, double sum) { + this.buckets = buckets; + this.count = count; + this.sum = sum; + } + } +} diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java index c4bb1f5fe..5ead70345 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java @@ -205,6 +205,7 @@ public class DataPoint implements DistributionDataPoint { private final LongAdder nativeZeroCount = new LongAdder(); private final LongAdder count = new LongAdder(); private final DoubleAdder sum = new DoubleAdder(); + @Nullable private final ClassicOnlyAccumulator classicOnlyAccumulator; private volatile int nativeSchema = nativeInitialSchema; // integer in [-4, 8] or CLASSIC_HISTOGRAM private volatile double nativeZeroThreshold = Histogram.this.nativeMinZeroThreshold; @@ -223,16 +224,24 @@ private DataPoint() { for (int i = 0; i < classicUpperBounds.length; i++) { classicBuckets[i] = new LongAdder(); } + classicOnlyAccumulator = + isClassicOnly() ? new ClassicOnlyAccumulator(classicUpperBounds.length) : null; maybeScheduleNextReset(); } @Override public double getSum() { + if (classicOnlyAccumulator != null) { + return classicOnlyAccumulator.snapshot().sum; + } return sum.sum(); } @Override public long getCount() { + if (classicOnlyAccumulator != null) { + return classicOnlyAccumulator.snapshot().count; + } return count.sum(); } @@ -242,7 +251,9 @@ public void observe(double value) { // See https://github.com/prometheus/client_golang/issues/1275 on ignoring NaN observations. return; } - if (!buffer.append(value)) { + if (classicOnlyAccumulator != null) { + classicOnlyAccumulator.observe(findClassicBucket(value), value); + } else if (!buffer.append(value)) { doObserve(value, false); } if (exemplarSampler != null) { @@ -256,7 +267,9 @@ public void observeWithExemplar(double value, Labels labels) { // See https://github.com/prometheus/client_golang/issues/1275 on ignoring NaN observations. return; } - if (!buffer.append(value)) { + if (classicOnlyAccumulator != null) { + classicOnlyAccumulator.observe(findClassicBucket(value), value); + } else if (!buffer.append(value)) { doObserve(value, false); } if (exemplarSampler != null) { @@ -266,12 +279,8 @@ public void observeWithExemplar(double value, Labels labels) { private void doObserve(double value, boolean fromBuffer) { // classicUpperBounds is an empty array if this is a native histogram only. - for (int i = 0; i < classicUpperBounds.length; ++i) { - // The last bucket is +Inf, so we always increment. - if (value <= classicUpperBounds[i]) { - classicBuckets[i].add(1); - break; - } + if (classicUpperBounds.length > 0) { + classicBuckets[findClassicBucket(value)].add(1); } boolean nativeBucketCreated = false; if (Histogram.this.nativeInitialSchema != CLASSIC_HISTOGRAM) { @@ -301,6 +310,15 @@ private void doObserve(double value, boolean fromBuffer) { private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { Exemplars exemplars = exemplarSampler != null ? exemplarSampler.collect() : Exemplars.EMPTY; + if (classicOnlyAccumulator != null) { + ClassicOnlyAccumulator.Snapshot snapshot = classicOnlyAccumulator.snapshot(); + return new HistogramSnapshot.HistogramDataPointSnapshot( + ClassicHistogramBuckets.of(classicUpperBounds, snapshot.buckets), + snapshot.sum, + labels, + exemplars, + createdTimeMillis); + } return buffer.run( expectedCount -> count.sum() == expectedCount, () -> { @@ -342,6 +360,20 @@ private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { v -> doObserve(v, true)); } + private boolean isClassicOnly() { + return Histogram.this.nativeInitialSchema == CLASSIC_HISTOGRAM; + } + + private int findClassicBucket(double value) { + for (int i = 0; i < classicUpperBounds.length; ++i) { + // The last bucket is +Inf, so we always return from this loop. + if (value <= classicUpperBounds[i]) { + return i; + } + } + throw new IllegalStateException("Classic histogram is missing the +Inf bucket."); + } + private boolean addToNativeBucket(double value, ConcurrentHashMap buckets) { boolean newBucketCreated = false; int bucketIndex; diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/HistogramTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/HistogramTest.java index cbfd5fade..127fd6a3e 100644 --- a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/HistogramTest.java +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/HistogramTest.java @@ -41,6 +41,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.LongAdder; import java.util.stream.Collectors; import org.junit.jupiter.api.AfterEach; @@ -1598,6 +1599,117 @@ void testObserveMultithreaded() assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); } + @Test + void testClassicOnlyCollectWhileObserversAreRunning() throws Exception { + Histogram histogram = Histogram.builder().name("test").classicOnly().build(); + DistributionDataPoint dataPoint = histogram.labelValues(); + int observerCount = 8; + int observationsPerThread = 25_000; + ExecutorService executor = Executors.newFixedThreadPool(observerCount + 1); + CountDownLatch ready = new CountDownLatch(observerCount); + CountDownLatch start = new CountDownLatch(1); + AtomicBoolean observersFinished = new AtomicBoolean(); + List> observers = new ArrayList<>(); + + for (int i = 0; i < observerCount; i++) { + observers.add( + executor.submit( + () -> { + ready.countDown(); + start.await(); + for (int observation = 0; observation < observationsPerThread; observation++) { + dataPoint.observe(1.25); + } + return null; + })); + } + + Future collector = + executor.submit( + () -> { + long previousCount = 0; + start.await(); + while (!observersFinished.get()) { + HistogramSnapshot.HistogramDataPointSnapshot snapshot = + histogram.collect().getDataPoints().get(0); + assertClassicOnlySnapshotIsCoherent(snapshot, previousCount); + previousCount = snapshot.getCount(); + } + return null; + }); + + assertThat(ready.await(5, TimeUnit.SECONDS)).isTrue(); + start.countDown(); + for (Future observer : observers) { + observer.get(10, TimeUnit.SECONDS); + } + observersFinished.set(true); + collector.get(10, TimeUnit.SECONDS); + + HistogramSnapshot.HistogramDataPointSnapshot snapshot = + histogram.collect().getDataPoints().get(0); + assertClassicOnlySnapshotIsCoherent(snapshot, observerCount * observationsPerThread); + assertThat(dataPoint.getCount()).isEqualTo(observerCount * observationsPerThread); + assertThat(dataPoint.getSum()) + .isCloseTo(observerCount * observationsPerThread * 1.25, offset(0.000_001)); + + executor.shutdown(); + assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void testClassicOnlyRetainsShortLivedThreadCellsAndClearCreatesFreshDataPoint() throws Exception { + Histogram histogram = + Histogram.builder().name("test").classicOnly().labelNames("status").build(); + DistributionDataPoint oldDataPoint = histogram.labelValues("200"); + int threadCount = 100; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + CountDownLatch start = new CountDownLatch(1); + List> observations = new ArrayList<>(); + for (int i = 0; i < threadCount; i++) { + observations.add( + executor.submit( + () -> { + start.await(); + oldDataPoint.observe(2.0); + return null; + })); + } + start.countDown(); + for (Future observation : observations) { + observation.get(5, TimeUnit.SECONDS); + } + executor.shutdown(); + assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + + assertThat(oldDataPoint.getCount()).isEqualTo(threadCount); + assertThat(oldDataPoint.getSum()).isEqualTo(threadCount * 2.0); + assertThat(getBucket(histogram, 2.5, "status", "200").getCount()).isEqualTo(threadCount); + + histogram.clear(); + DistributionDataPoint newDataPoint = histogram.labelValues("200"); + assertThat(newDataPoint).isNotSameAs(oldDataPoint); + assertThat(newDataPoint.getCount()).isZero(); + assertThat(newDataPoint.getSum()).isZero(); + + newDataPoint.observe(3.0); + assertThat(newDataPoint.getCount()).isOne(); + assertThat(newDataPoint.getSum()).isEqualTo(3.0); + assertThat(oldDataPoint.getCount()).isEqualTo(threadCount); + assertThat(oldDataPoint.getSum()).isEqualTo(threadCount * 2.0); + } + + private static void assertClassicOnlySnapshotIsCoherent( + HistogramSnapshot.HistogramDataPointSnapshot snapshot, long minimumCount) { + assertThat(snapshot.getCount()).isGreaterThanOrEqualTo(minimumCount); + assertThat(snapshot.getSum()).isCloseTo(snapshot.getCount() * 1.25, offset(0.000_001)); + long bucketTotal = 0; + for (ClassicHistogramBucket bucket : snapshot.getClassicBuckets()) { + bucketTotal += bucket.getCount(); + } + assertThat(bucketTotal).isEqualTo(snapshot.getCount()); + } + @Test void testNativeResetDuration() { // Test that nativeResetDuration can be configured without error and the histogram