From b2d300a2e12bfab8207eddab75ed0bf400438d63 Mon Sep 17 00:00:00 2001 From: Callum Donald Date: Mon, 28 Sep 2026 23:52:29 +0100 Subject: [PATCH 1/3] refactor(core): track weighted observation tickets in Buffer Buffer counts one ticket per observation, and collection spins until the metric's count adder matches the ticket total. That invariant makes a batched observation ("this value occurred n times") impossible to add without desynchronising collection permanently: the count adder would overshoot the expected count and never match it again, so every subsequent scrape of that data point would spin to its deadline and throw. Teach Buffer to count weighted tickets. append(value, weight) claims the ticket range (count - weight, count] with a single atomic add, so a batch cannot straddle a collector's activation -- it is either entirely inside that collection's expected count or entirely outside it, never split. The late-reader guard generalises from a point test to a range test and is identical for weight 1. Buffered generations carry a lazily allocated weights array, null while every entry has weight 1 so that scrapes racing only single observations allocate exactly as before, and replay passes the weight through a primitive WeightedObserver rather than boxing each value into a Consumer. Histogram.doObserve and Summary.doObserve gain a private multiplicity parameter to match. No public API change: observe(double) and append(double) take the same path they did before. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Callum Donald --- .../metrics/core/metrics/Buffer.java | 74 ++++-- .../metrics/core/metrics/Histogram.java | 41 +-- .../metrics/core/metrics/Summary.java | 14 +- .../metrics/core/metrics/BufferTest.java | 18 +- .../metrics/BufferWeightedAppendTest.java | 233 ++++++++++++++++++ 5 files changed, 338 insertions(+), 42 deletions(-) create mode 100644 prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferWeightedAppendTest.java diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java index 9283e60fc..11671ec17 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java @@ -8,7 +8,6 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; -import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; import javax.annotation.Nullable; @@ -22,8 +21,13 @@ * observations into the live metric state. * *

The default collection wait is five seconds. A generation is capped at one million buffered - * observations (about eight MiB of double storage) to keep a stalled collection from growing - * without bound; the cap applies backpressure rather than dropping observations. + * entries (about eight MiB of double storage, plus eight MiB of weights once a batched observation + * has been buffered) to keep a stalled collection from growing without bound; the cap applies + * backpressure rather than dropping observations. + * + *

Batched observations ({@code weight} identical values recorded as one operation) are tracked + * with the same ticket protocol: one atomic add claims the whole ticket range of the batch, so a + * batch is either entirely inside a collection's expected count or entirely outside it. */ class Buffer { private static final long BUFFER_ACTIVE_BIT = 1L << 63; @@ -39,10 +43,18 @@ class Buffer { /** Observations buffered during one collection cycle. */ private static final class Generation { private double[] values = EMPTY_BUFFER; + // Multiplicity of each buffered value. Allocated on the first weighted append only; null means + // every buffered value has weight 1, which keeps single observations free of the extra array. + @Nullable private long[] weights; private int size; private boolean active = true; } + /** Replays one buffered entry, {@code value} observed {@code weight} times, into the metric. */ + interface WeightedObserver { + void observe(double value, long weight); + } + // Tracking observation counts requires an AtomicLong for coordination between recording and // collecting. AtomicLong does much worse under contention than the LongAdder instances used // elsewhere to hold aggregated state. To reduce contention, the count is striped across the @@ -109,10 +121,29 @@ boolean append(double value) { if ((count & BUFFER_ACTIVE_BIT) == 0) { return false; } - return appendToActiveGeneration(value, stripe, count); + return appendToActiveGeneration(value, 1L, stripe, count); } - private boolean appendToActiveGeneration(double value, int stripe, long count) { + /** + * Like {@link #append(double)}, for {@code weight} identical observations of {@code value} + * recorded as one operation. + * + *

The batch claims its ticket range {@code (count - weight, count]} with a single atomic add, + * so it cannot straddle a collector's activation: either all of its tickets predate the + * activation and the batch is included in that collection's expected count (direct path), or none + * do and the batch is buffered for replay after the snapshot. + */ + boolean append(double value, long weight) { + int stripe = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length); + AtomicLong counter = stripedObservationCounts[stripe]; + long count = counter.addAndGet(weight); + if ((count & BUFFER_ACTIVE_BIT) == 0) { + return false; + } + return appendToActiveGeneration(value, weight, stripe, count); + } + + private boolean appendToActiveGeneration(double value, long weight, int stripe, long count) { // Allow tests to pause between allocating an observation ticket and reading the generation. beforeGenerationRead.run(); Generation generation = activeGeneration; @@ -126,10 +157,11 @@ private boolean appendToActiveGeneration(double value, int stripe, long count) { if (current != generation || !generation.active) { return false; } - if ((count & ~BUFFER_ACTIVE_BIT) <= generationStartCounts[stripe]) { - // This observation incremented its stripe in an earlier generation. The current collector - // already includes it in expectedCount, so buffering it here would make the collector wait - // for an observation that is only replayed after that same wait finishes. + if ((count & ~BUFFER_ACTIVE_BIT) - weight < generationStartCounts[stripe]) { + // This observation claimed its tickets in an earlier generation (for weight 1 this is the + // familiar count <= generationStartCounts[stripe]). The current collector already includes + // it in expectedCount, so buffering it here would make the collector wait for an + // observation that is only replayed after that same wait finishes. return false; } while (generation.size >= maxBufferSize && generation.active) { @@ -148,9 +180,18 @@ private boolean appendToActiveGeneration(double value, int stripe, long count) { generation.values.length > maxBufferSize / 2 ? maxBufferSize : generation.values.length * 2; - generation.values = - Arrays.copyOf( - generation.values, Math.min(maxBufferSize, Math.max(INITIAL_BUFFER_SIZE, doubled))); + int newLength = Math.min(maxBufferSize, Math.max(INITIAL_BUFFER_SIZE, doubled)); + generation.values = Arrays.copyOf(generation.values, newLength); + if (generation.weights != null) { + generation.weights = Arrays.copyOf(generation.weights, newLength); + } + } + if (weight != 1L && generation.weights == null) { + generation.weights = new long[generation.values.length]; + Arrays.fill(generation.weights, 0, generation.size, 1L); + } + if (generation.weights != null) { + generation.weights[generation.size] = weight; } generation.values[generation.size++] = value; return true; @@ -185,7 +226,7 @@ T observeDirect(Supplier observeFunction) { T run( Function complete, Supplier createResult, - Consumer observeFunction) { + WeightedObserver observeFunction) { return requireNonNull(run(complete, createResult, observeFunction, true)); } @@ -194,10 +235,11 @@ T run( T run( Function complete, Supplier createResult, - Consumer observeFunction, + WeightedObserver observeFunction, boolean failOnTimeout) { Generation generation = new Generation(); double[] buffer; + long[] weights; int bufferSize; boolean timedOut = false; T result = null; @@ -241,15 +283,17 @@ T run( reset = false; } buffer = generation.values; + weights = generation.weights; bufferSize = generation.size; generation.values = EMPTY_BUFFER; + generation.weights = null; generation.size = 0; bufferSpaceAvailable.signalAll(); } finally { appendLock.unlock(); } for (int i = 0; i < bufferSize; i++) { - observeFunction.accept(buffer[i]); + observeFunction.observe(buffer[i], weights == null ? 1L : weights[i]); } // Keep the inactive generation visible until replay completes. An appender that loses the // generation race must take observationLock before observing directly. 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 d03ac9b97..e983448c1 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 @@ -244,7 +244,7 @@ public void observe(double value) { } if (!buffer.append(value)) { boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value)); - maybeResetOrScaleDown(value, nativeBucketCreated); + maybeResetOrScaleDown(value, 1L, nativeBucketCreated); } if (exemplarSampler != null) { exemplarSampler.observe(value); @@ -259,7 +259,7 @@ public void observeWithExemplar(double value, Labels labels) { } if (!buffer.append(value)) { boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value)); - maybeResetOrScaleDown(value, nativeBucketCreated); + maybeResetOrScaleDown(value, 1L, nativeBucketCreated); } if (exemplarSampler != null) { exemplarSampler.observeWithExemplar(value, labels); @@ -267,27 +267,37 @@ public void observeWithExemplar(double value, Labels labels) { } private boolean doObserve(double value) { + return doObserve(value, 1L); + } + + /** + * Records {@code n} identical observations of {@code value}. All of them land in the same + * classic bucket and the same native bucket, so the bucket lookups happen once and the cost + * does not depend on {@code n}. + */ + private boolean doObserve(double value, long n) { // 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); + classicBuckets[i].add(n); break; } } boolean nativeBucketCreated = false; if (Histogram.this.nativeInitialSchema != CLASSIC_HISTOGRAM) { if (value > nativeZeroThreshold) { - nativeBucketCreated = addToNativeBucket(value, nativeBucketsForPositiveValues); + nativeBucketCreated = addToNativeBucket(value, nativeBucketsForPositiveValues, n); } else if (value < -nativeZeroThreshold) { - nativeBucketCreated = addToNativeBucket(-value, nativeBucketsForNegativeValues); + nativeBucketCreated = addToNativeBucket(-value, nativeBucketsForNegativeValues, n); } else { - nativeZeroCount.add(1); + nativeZeroCount.add(n); } } - sum.add(value); - count - .increment(); // must be the last step, because count is used to signal that the operation + // The product is the correctly rounded exact sum of the batch. Multiplying by 1 is exact, so + // single observations add precisely value, as before. + sum.add(value * n); + count.add(n); // must be the last step, because count is used to signal that the operation // is complete. return nativeBucketCreated; } @@ -335,7 +345,8 @@ private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { this::doObserve); } - private boolean addToNativeBucket(double value, ConcurrentHashMap buckets) { + private boolean addToNativeBucket( + double value, ConcurrentHashMap buckets, long n) { boolean newBucketCreated = false; int bucketIndex; if (Double.isInfinite(value)) { @@ -354,7 +365,7 @@ private boolean addToNativeBucket(double value, ConcurrentHashMap */ - private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { + private void maybeResetOrScaleDown(double value, long n, boolean nativeBucketCreated) { AtomicBoolean wasReset = new AtomicBoolean(false); if (resetDurationExpired && nativeSchema < nativeInitialSchema) { // If nativeSchema < initialNativeSchema the histogram has been scaled down. @@ -445,11 +456,11 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { maybeScaleDown(wasReset); } if (wasReset.get()) { - // We just discarded the newly observed value. Observe it again. - if (!buffer.append(value)) { + // We just discarded the newly observed value(s). Observe them again. + if (!buffer.append(value, n)) { buffer.observeDirect( () -> { - doObserve(value); + doObserve(value, n); return null; }); } diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java index d75ac6c74..f09a31314 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java @@ -214,13 +214,21 @@ public void observeWithExemplar(double value, Labels labels) { } private void doObserve(double amount) { - sum.add(amount); + doObserve(amount, 1L); + } + + private void doObserve(double amount, long n) { + sum.add(amount * n); if (quantileValues != null) { - quantileValues.observe(amount); + // The quantile sketch has no weighted insert, so this part of a batch costs one insert per + // observation. count and sum are still updated in constant time. + for (long i = 0; i < n; i++) { + quantileValues.observe(amount); + } } // count must be incremented last, because in collect() the count // indicates the number of completed observations. - count.increment(); + count.add(n); } private SummarySnapshot.SummaryDataPointSnapshot collect(Labels labels) { diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java index 064bb8186..35cc0359c 100644 --- a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -43,7 +43,7 @@ void timeoutDeactivatesBufferAndReplaysBufferedObservations() throws Interrupted return false; }, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - replayedObservations::add); + (value, weight) -> replayedObservations.add(value)); } catch (IllegalStateException expected) { timedOut.set(true); } @@ -73,7 +73,7 @@ void timeoutDoesNotCreateSnapshot() { () -> { throw new AssertionError("snapshot should not be created"); }, - ignored -> {})) + (value, weight) -> {})) .withMessage("Timed out while waiting for in-flight observations."); } @@ -103,7 +103,7 @@ private void assertFailedSnapshotIsRecovered(Throwable failure) { } throw (RuntimeException) failure; }, - replayedObservations::add)) + (value, weight) -> replayedObservations.add(value))) .isSameAs(failure); assertThat(replayedObservations).containsExactly(1.0); assertThat(buffer.append(2.0)).isFalse(); @@ -111,7 +111,7 @@ private void assertFailedSnapshotIsRecovered(Throwable failure) { buffer.run( ignored -> true, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - replayedObservations::add)) + (value, weight) -> replayedObservations.add(value))) .isNotNull(); assertThat(replayedObservations).containsExactly(1.0); } @@ -142,7 +142,7 @@ void fullBufferUnblocksAppenderWhenGenerationIsDeactivated() throws InterruptedE return false; }, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - ignored -> {}); + (value, weight) -> {}); } catch (IllegalStateException expected) { timedOut.set(true); } @@ -191,7 +191,7 @@ void interruptedAppenderLeavesBoundedBufferWait() throws InterruptedException { return false; }, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - ignored -> {}); + (value, weight) -> {}); } catch (IllegalStateException expected) { // The runner is only used to hold the generation open for this test. } @@ -282,7 +282,7 @@ private static void assertLateAppenderHandoff(boolean reset, boolean pauseBefore } return snapshot; }, - ignored -> completedObservations.incrementAndGet())); + (value, weight) -> completedObservations.addAndGet(weight))); awaitLatch(firstSnapshotStarted); // Increment while generation A is active, but do not read activeGeneration yet. @@ -311,7 +311,7 @@ private static void assertLateAppenderHandoff(boolean reset, boolean pauseBefore () -> new CounterSnapshot.CounterDataPointSnapshot( completedObservations.get(), Labels.EMPTY, null, 0), - ignored -> completedObservations.incrementAndGet())); + (value, weight) -> completedObservations.addAndGet(weight))); awaitLatch(secondRunStarted); assertThat(secondExpectedCount).hasValue(1); // An observation arriving after B's activation still belongs in B's buffer. It must not @@ -331,7 +331,7 @@ private static void assertLateAppenderHandoff(boolean reset, boolean pauseBefore () -> new CounterSnapshot.CounterDataPointSnapshot( completedObservations.get(), Labels.EMPTY, null, 0), - ignored -> completedObservations.incrementAndGet()) + (value, weight) -> completedObservations.addAndGet(weight)) .getValue()) .isEqualTo(2); } finally { diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferWeightedAppendTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferWeightedAppendTest.java new file mode 100644 index 000000000..cf16eda57 --- /dev/null +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferWeightedAppendTest.java @@ -0,0 +1,233 @@ +package io.prometheus.metrics.core.metrics; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.prometheus.metrics.model.snapshots.CounterSnapshot; +import io.prometheus.metrics.model.snapshots.Labels; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import org.junit.jupiter.api.Test; + +/** Deterministic coverage of the weighted ticket protocol used by batched observations. */ +class BufferWeightedAppendTest { + + private static CounterSnapshot.CounterDataPointSnapshot snapshot(long value) { + return new CounterSnapshot.CounterDataPointSnapshot(value, Labels.EMPTY, null, 0); + } + + @Test + void weightedAppendWithoutActiveGenerationIsDirectAndCountedInFull() { + Buffer buffer = new Buffer(); + AtomicLong completed = new AtomicLong(); + AtomicLong expected = new AtomicLong(-1); + List replayed = new ArrayList<>(); + + assertThat(buffer.append(1.5, 5)).isFalse(); + buffer.observeDirect(() -> completed.addAndGet(5)); + + CounterSnapshot.CounterDataPointSnapshot result = + buffer.run( + expectedCount -> { + expected.set(expectedCount); + return completed.get() == expectedCount; + }, + () -> snapshot(completed.get()), + (value, weight) -> replayed.add(value + "x" + weight)); + + // The collector expects the full weight, not one ticket per append() call. + assertThat(expected).hasValue(5); + assertThat(result.getValue()).isEqualTo(5); + assertThat(replayed).isEmpty(); + } + + @Test + void weightedAppendDuringActiveGenerationIsBufferedAndReplayedWithWeight() throws Exception { + Buffer buffer = new Buffer(); + AtomicLong completed = new AtomicLong(); + List replayed = new ArrayList<>(); + CountDownLatch snapshotStarted = new CountDownLatch(1); + CountDownLatch finishSnapshot = new CountDownLatch(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future run = + executor.submit( + () -> + buffer.run( + expectedCount -> completed.get() == expectedCount, + () -> { + snapshotStarted.countDown(); + await(finishSnapshot); + return snapshot(completed.get()); + }, + (value, weight) -> { + completed.addAndGet(weight); + replayed.add(value + "x" + weight); + })); + await(snapshotStarted); + + // Generation is active: singles and batches interleave in one generation. The weights array + // is allocated lazily by the first batch and must back-fill the singles before it. + assertThat(buffer.append(1.0)).isTrue(); + assertThat(buffer.append(2.0, 3)).isTrue(); + assertThat(buffer.append(3.0)).isTrue(); + assertThat(buffer.append(4.0, 1_000_000_000L)).isTrue(); + + finishSnapshot.countDown(); + assertThat(run.get(10, TimeUnit.SECONDS).getValue()).isEqualTo(0); + } finally { + finishSnapshot.countDown(); + executor.shutdownNow(); + } + assertThat(replayed).containsExactly("1.0x1", "2.0x3", "3.0x1", "4.0x1000000000"); + assertThat(completed).hasValue(1_000_000_005L); + + // The replayed batches are now part of the live state; the next collection must not wait. + CounterSnapshot.CounterDataPointSnapshot next = + buffer.run( + expectedCount -> completed.get() == expectedCount, + () -> snapshot(completed.get()), + (value, weight) -> completed.addAndGet(weight)); + assertThat(next.getValue()).isEqualTo(1_000_000_005L); + } + + @Test + void weightsSurviveGenerationGrowth() throws Exception { + Buffer buffer = new Buffer(); + AtomicLong completed = new AtomicLong(); + List replayedWeights = new ArrayList<>(); + CountDownLatch snapshotStarted = new CountDownLatch(1); + CountDownLatch finishSnapshot = new CountDownLatch(1); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future run = + executor.submit( + () -> + buffer.run( + expectedCount -> completed.get() == expectedCount, + () -> { + snapshotStarted.countDown(); + await(finishSnapshot); + return snapshot(completed.get()); + }, + (value, weight) -> { + completed.addAndGet(weight); + replayedWeights.add(weight); + })); + await(snapshotStarted); + // First entry is a batch, then enough singles to force the arrays to grow past the initial + // 128 slots (twice), then another batch. + assertThat(buffer.append(0.5, 42)).isTrue(); + for (int i = 0; i < 300; i++) { + assertThat(buffer.append(i)).isTrue(); + } + assertThat(buffer.append(0.25, 7)).isTrue(); + finishSnapshot.countDown(); + run.get(10, TimeUnit.SECONDS); + } finally { + finishSnapshot.countDown(); + executor.shutdownNow(); + } + assertThat(replayedWeights).hasSize(302); + assertThat(replayedWeights.get(0)).isEqualTo(42); + assertThat(replayedWeights.subList(1, 301)).containsOnly(1L); + assertThat(replayedWeights.get(301)).isEqualTo(7); + assertThat(completed).hasValue(42 + 300 + 7); + } + + /** + * A batch takes its tickets while generation A is active, but only reads activeGeneration after A + * finished and B started. B's expected count includes the whole batch, so the batch must be + * observed directly, not buffered into B. This is the weighted version of the late-appender + * handoff in {@link BufferTest}. + */ + @Test + void batchTicketsClaimedBeforeActivationAreObservedDirectlyNotBuffered() throws Exception { + long weight = 7; + CountDownLatch firstSnapshotStarted = new CountDownLatch(1); + CountDownLatch finishFirstSnapshot = new CountDownLatch(1); + CountDownLatch ticketsClaimed = new CountDownLatch(1); + CountDownLatch readGeneration = new CountDownLatch(1); + CountDownLatch secondRunStarted = new CountDownLatch(1); + AtomicLong completed = new AtomicLong(); + AtomicLong secondExpectedCount = new AtomicLong(); + AtomicBoolean pauseAppender = new AtomicBoolean(true); + Buffer buffer = + new Buffer( + TimeUnit.SECONDS.toNanos(5), + 16, + () -> { + if (pauseAppender.compareAndSet(true, false)) { + ticketsClaimed.countDown(); + await(readGeneration); + } + }); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future firstRun = + executor.submit( + () -> + buffer.run( + expectedCount -> completed.get() == expectedCount, + () -> { + firstSnapshotStarted.countDown(); + await(finishFirstSnapshot); + return snapshot(completed.get()); + }, + (value, w) -> completed.addAndGet(w))); + await(firstSnapshotStarted); + + Future appender = + executor.submit( + () -> { + boolean appended = buffer.append(1.0, weight); + if (!appended) { + buffer.observeDirect(() -> completed.addAndGet(weight)); + } + return appended; + }); + await(ticketsClaimed); + finishFirstSnapshot.countDown(); + assertThat(firstRun.get(10, TimeUnit.SECONDS).getValue()).isEqualTo(0); + + Future secondRun = + executor.submit( + () -> + buffer.run( + expectedCount -> { + secondExpectedCount.set(expectedCount); + secondRunStarted.countDown(); + return completed.get() == expectedCount; + }, + () -> snapshot(completed.get()), + (value, w) -> completed.addAndGet(w))); + await(secondRunStarted); + // B counted the whole batch, not a single ticket. + assertThat(secondExpectedCount).hasValue(weight); + + readGeneration.countDown(); + assertThat(secondRun.get(10, TimeUnit.SECONDS).getValue()).isEqualTo(weight); + assertThat(appender.get(10, TimeUnit.SECONDS)).isFalse(); + assertThat(completed).hasValue(weight); + } finally { + finishFirstSnapshot.countDown(); + readGeneration.countDown(); + executor.shutdownNow(); + } + } + + private static void await(CountDownLatch latch) { + try { + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } +} From 586892adcd0fdd7553ec8b004642e1a9bfc9cba9 Mon Sep 17 00:00:00 2001 From: Callum Donald Date: Mon, 28 Sep 2026 23:54:59 +0100 Subject: [PATCH 2/3] feat(core): add observe(value, count) for batched observations Recording the same value n times required a loop over observe(double), which repeated work that does not depend on the multiplicity: the buffer ticket, the classic bucket scan, findBucketIndex, the bucket map lookup, the schema-maintenance check and the exemplar-sampler call. Since all n values are identical they land in exactly one classic bucket and one native bucket, so the whole operation can be done once with a count passed to the four adders. Add observe(double value, long count) to DistributionDataPoint with a default implementation that loops, so existing implementations of this @StableApi interface keep compiling and behave correctly. Histogram and Summary override it: one bucket lookup, then add(count) on the bucket, zero-count, sum and count accumulators. The cost does not depend on count -- a batch of a million costs the same as a batch of sixteen. Semantics: buckets and counts are exactly what count single calls produce, and the batch is atomic with respect to scrapes, which a loop never was. The sum is increased by the correctly rounded product value * count, which is at least as accurate as count successive additions and identical for count == 1; making it bit-for-bit equal to a sequential loop would require O(count) work inside the collector's lock, for a reproducibility that a striped DoubleAdder does not offer anyway. count == 0 is a no-op, a negative count throws IllegalArgumentException, and NaN is ignored as in observe(double). At most one exemplar is sampled per batch. Summary batches count and sum in constant time; with quantiles configured the CKMS sketch has no weighted insert, so that part still costs one insert per observation. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Callum Donald --- .../datapoints/DistributionDataPoint.java | 32 ++ .../metrics/core/metrics/Histogram.java | 25 ++ .../metrics/core/metrics/Summary.java | 26 ++ .../core/metrics/BatchObserveTest.java | 317 ++++++++++++++++++ 4 files changed, 400 insertions(+) create mode 100644 prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BatchObserveTest.java diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/datapoints/DistributionDataPoint.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/datapoints/DistributionDataPoint.java index f87bede40..3a4988c49 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/datapoints/DistributionDataPoint.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/datapoints/DistributionDataPoint.java @@ -33,6 +33,38 @@ public interface DistributionDataPoint extends DataPoint, TimerApi { /** Observe {@code value}, and create a custom exemplar with the given labels. */ void observeWithExemplar(double value, Labels labels); + /** + * Observe {@code value} {@code count} times, as a single operation. + * + *

Use this to record pre-aggregated data ("this value occurred {@code count} times") without + * paying the per-observation cost of calling {@link #observe(double)} in a loop. Buckets and the + * observation count end up exactly as if {@link #observe(double)} had been called {@code count} + * times. The implementations in this library additionally guarantee that + * + *

    + *
  • the batch is applied atomically with respect to scrapes, so a snapshot contains either + * all of it or none of it, + *
  • the sum is increased by the correctly rounded product {@code value * count} rather than + * by {@code count} successive floating point additions (the product is at least as + * accurate, and for {@code count == 1} the two are identical), + *
  • at most one exemplar is sampled for the batch. + *
+ * + *

{@code count == 0} is a no-op. A negative {@code count} throws {@link + * IllegalArgumentException}. {@code NaN} values are ignored, as in {@link #observe(double)}. + * + *

The default implementation loops over {@link #observe(double)}. Histograms and summaries + * override it with an implementation whose cost does not depend on {@code count}. + */ + default void observe(double value, long count) { + if (count < 0) { + throw new IllegalArgumentException("Negative count " + count + " is illegal."); + } + for (long i = 0; i < count; i++) { + observe(value); + } + } + @Override default Timer startTimer() { return new Timer(this::observe); 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 e983448c1..2bc450ceb 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 @@ -196,6 +196,11 @@ public void observeWithExemplar(double amount, Labels labels) { getNoLabels().observeWithExemplar(amount, labels); } + @Override + public void observe(double amount, long count) { + getNoLabels().observe(amount, count); + } + public class DataPoint implements DistributionDataPoint { private final LongAdder[] classicBuckets; private final ConcurrentHashMap nativeBucketsForPositiveValues = @@ -251,6 +256,26 @@ public void observe(double value) { } } + @Override + public void observe(double value, long count) { + if (count < 0) { + throw new IllegalArgumentException( + "Negative count " + count + " is illegal for Histogram metrics."); + } + if (count == 0 || Double.isNaN(value)) { + // See https://github.com/prometheus/client_golang/issues/1275 on ignoring NaN observations. + return; + } + if (!buffer.append(value, count)) { + boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value, count)); + maybeResetOrScaleDown(value, count, nativeBucketCreated); + } + if (exemplarSampler != null) { + // One exemplar candidate per batch: a batch has one value and one current span context. + exemplarSampler.observe(value); + } + } + @Override public void observeWithExemplar(double value, Labels labels) { if (Double.isNaN(value)) { diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java index f09a31314..6cf28ffee 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java @@ -110,6 +110,11 @@ public void observeWithExemplar(double amount, Labels labels) { getNoLabels().observeWithExemplar(amount, labels); } + @Override + public void observe(double amount, long count) { + getNoLabels().observe(amount, count); + } + @Override public SummarySnapshot collect() { return (SummarySnapshot) super.collect(); @@ -196,6 +201,27 @@ public void observe(double value) { } } + @Override + public void observe(double value, long count) { + if (count < 0) { + throw new IllegalArgumentException( + "Negative count " + count + " is illegal for Summary metrics."); + } + if (count == 0 || Double.isNaN(value)) { + return; + } + if (!buffer.append(value, count)) { + buffer.observeDirect( + () -> { + doObserve(value, count); + return null; + }); + } + if (exemplarSampler != null) { + exemplarSampler.observe(value); + } + } + @Override public void observeWithExemplar(double value, Labels labels) { if (Double.isNaN(value)) { diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BatchObserveTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BatchObserveTest.java new file mode 100644 index 000000000..d05500c8b --- /dev/null +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BatchObserveTest.java @@ -0,0 +1,317 @@ +package io.prometheus.metrics.core.metrics; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import io.prometheus.metrics.model.snapshots.ClassicHistogramBucket; +import io.prometheus.metrics.model.snapshots.HistogramSnapshot; +import io.prometheus.metrics.model.snapshots.NativeHistogramBucket; +import io.prometheus.metrics.model.snapshots.SummarySnapshot; +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; + +/** {@code observe(value, count)} must leave a histogram exactly as {@code count} single calls. */ +class BatchObserveTest { + + private static Histogram hybrid(String name) { + return Histogram.builder().name(name).nativeInitialSchema(5).build(); + } + + /** + * Every integer-valued field of the data point. The sum is checked separately: a batch adds the + * correctly rounded product, sequential observation accumulates one rounding per addition, so the + * two agree to within a few ulps rather than bit for bit (see {@link #assertSameSum}). + */ + private static String describe(HistogramSnapshot s) { + HistogramSnapshot.HistogramDataPointSnapshot dp = s.getDataPoints().get(0); + StringBuilder sb = new StringBuilder(); + sb.append("count=").append(dp.getCount()); + sb.append(" schema=").append(dp.getNativeSchema()); + sb.append(" zeroCount=").append(dp.getNativeZeroCount()); + sb.append(" zeroThreshold=").append(dp.getNativeZeroThreshold()); + sb.append(" classic="); + for (ClassicHistogramBucket b : dp.getClassicBuckets()) { + sb.append('[').append(b.getUpperBound()).append('=').append(b.getCount()).append(']'); + } + sb.append(" pos="); + for (NativeHistogramBucket b : dp.getNativeBucketsForPositiveValues()) { + sb.append('[').append(b.getBucketIndex()).append('=').append(b.getCount()).append(']'); + } + sb.append(" neg="); + for (NativeHistogramBucket b : dp.getNativeBucketsForNegativeValues()) { + sb.append('[').append(b.getBucketIndex()).append('=').append(b.getCount()).append(']'); + } + return sb.toString(); + } + + private static void assertSameSum(HistogramSnapshot batch, HistogramSnapshot seq, long n) { + double b = batch.getDataPoints().get(0).getSum(); + double s = seq.getDataPoints().get(0).getSum(); + if (Double.isInfinite(s) || s == 0.0) { + assertThat(b).isEqualTo(s); + } else { + // Sequential accumulation carries at most one rounding error per addition. + assertThat(b).isCloseTo(s, org.assertj.core.data.Offset.offset(Math.ulp(s) * n)); + } + } + + @Test + void batchMatchesSequentialForAssortedValues() { + double[] values = { + 0.0, + 1e-9, + -1e-9, + 0.5, + 1.0, + 1.5, + 3.7, + -2.25, + 1e300, + Double.POSITIVE_INFINITY, + Double.NEGATIVE_INFINITY, + 0.25, + 1024.0 + }; + long[] counts = {1, 2, 7, 1000, 12345}; + for (double value : values) { + for (long n : counts) { + Histogram seq = hybrid("seq"); + Histogram batch = hybrid("batch"); + for (long i = 0; i < n; i++) { + seq.observe(value); + } + batch.observe(value, n); + HistogramSnapshot bs = batch.collect(); + HistogramSnapshot ss = seq.collect(); + assertThat(describe(bs)).as("value=%s n=%s", value, n).isEqualTo(describe(ss)); + assertSameSum(bs, ss, n); + } + } + } + + @Test + void sumIsTheCorrectlyRoundedProduct() { + // 0.1 added ten times accumulates rounding error; the batch sum is the single correctly + // rounded product. Both are legitimate; this pins down which one the batch produces. + Histogram batch = hybrid("batch"); + batch.observe(0.1, 10); + assertThat(batch.collect().getDataPoints().get(0).getSum()).isEqualTo(1.0); + Histogram seq = hybrid("seq"); + for (int i = 0; i < 10; i++) { + seq.observe(0.1); + } + assertThat(seq.collect().getDataPoints().get(0).getSum()).isEqualTo(0.9999999999999999); + } + + @Test + void batchMatchesSequentialAcrossScaleDown() { + Histogram seq = + Histogram.builder() + .name("seq") + .nativeOnly() + .nativeInitialSchema(5) + .nativeMaxNumberOfBuckets(8) + .build(); + Histogram batch = + Histogram.builder() + .name("batch") + .nativeOnly() + .nativeInitialSchema(5) + .nativeMaxNumberOfBuckets(8) + .build(); + for (int i = 1; i <= 40; i++) { + double v = i * 0.37; + for (int k = 0; k < 3; k++) { + seq.observe(v); + } + batch.observe(v, 3); + } + HistogramSnapshot bs = batch.collect(); + HistogramSnapshot ss = seq.collect(); + assertThat(describe(bs)).isEqualTo(describe(ss)); + assertSameSum(bs, ss, 120); + } + + @Test + void resetReappliesTheWholeBatch() throws Exception { + // Scale the histogram down, then flag the reset duration as expired (as HistogramTest does for + // the client_golang cases). The next observation resets the histogram and is re-applied; for a + // batch, the whole batch must be re-applied. + Histogram seq = scaledDown("seq"); + Histogram batch = scaledDown("batch"); + expireResetDuration(seq); + expireResetDuration(batch); + for (int k = 0; k < 9; k++) { + seq.observe(2.5); + } + batch.observe(2.5, 9); + HistogramSnapshot.HistogramDataPointSnapshot dp = batch.collect().getDataPoints().get(0); + assertThat(dp.getNativeSchema()).isEqualTo(5); // reset restored the initial schema + assertThat(dp.getCount()).isEqualTo(9); + assertThat(dp.getSum()).isEqualTo(22.5); + assertThat(describe(batch.collect())).isEqualTo(describe(seq.collect())); + } + + private static Histogram scaledDown(String name) { + Histogram h = + Histogram.builder() + .name(name) + .nativeOnly() + .nativeInitialSchema(5) + .nativeMaxNumberOfBuckets(4) + .build(); + for (int i = 1; i <= 64; i++) { + h.observe(i * 0.61); + } + assertThat(h.collect().getDataPoints().get(0).getNativeSchema()).isLessThan(5); + return h; + } + + private static void expireResetDuration(Histogram h) throws Exception { + Field flag = Histogram.DataPoint.class.getDeclaredField("resetDurationExpired"); + flag.setAccessible(true); + flag.set(h.getNoLabels(), true); + } + + @Test + void zeroIsANoOpNegativeIsRejectedNaNIsIgnored() { + Histogram h = hybrid("edge"); + h.observe(1.0, 0); + assertThat(h.collect().getDataPoints().get(0).getCount()).isZero(); + assertThatThrownBy(() -> h.observe(1.0, -1)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Negative count -1"); + h.observe(Double.NaN, 100); + assertThat(h.collect().getDataPoints().get(0).getCount()).isZero(); + assertThat(h.collect().getDataPoints().get(0).getSum()).isZero(); + } + + @Test + void interfaceDefaultLoops() { + List seen = new ArrayList<>(); + io.prometheus.metrics.core.datapoints.DistributionDataPoint dp = + new io.prometheus.metrics.core.datapoints.DistributionDataPoint() { + @Override + public long getCount() { + return seen.size(); + } + + @Override + public double getSum() { + return 0; + } + + @Override + public void observe(double value) { + seen.add(value); + } + + @Override + public void observeWithExemplar( + double value, io.prometheus.metrics.model.snapshots.Labels labels) { + seen.add(value); + } + }; + dp.observe(4.0, 3); + assertThat(seen).containsExactly(4.0, 4.0, 4.0); + assertThatThrownBy(() -> dp.observe(4.0, -2)).isInstanceOf(IllegalArgumentException.class); + } + + @Test + void batchesAreNeverSplitByConcurrentScrapes() throws Exception { + Histogram h = hybrid("concurrent"); + int threads = 4; + int iterations = 2000; + long batchSize = 5; + ExecutorService pool = Executors.newFixedThreadPool(threads + 1); + CountDownLatch start = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + List> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + futures.add( + pool.submit( + () -> { + try { + start.await(); + for (int i = 0; i < iterations; i++) { + h.observe(1.0 + (i % 7) * 0.25, batchSize); + } + } catch (Throwable e) { + failure.compareAndSet(null, e); + } + })); + } + futures.add( + pool.submit( + () -> { + try { + start.await(); + for (int i = 0; i < 200; i++) { + HistogramSnapshot.HistogramDataPointSnapshot dp = + h.collect().getDataPoints().get(0); + // Every snapshot must contain whole batches only. + if (dp.getCount() % batchSize != 0) { + throw new AssertionError("torn batch: count=" + dp.getCount()); + } + } + } catch (Throwable e) { + failure.compareAndSet(null, e); + } + })); + start.countDown(); + for (Future f : futures) { + f.get(120, TimeUnit.SECONDS); + } + pool.shutdown(); + assertThat(failure.get()).isNull(); + + HistogramSnapshot.HistogramDataPointSnapshot dp = h.collect().getDataPoints().get(0); + long expected = (long) threads * iterations * batchSize; + assertThat(dp.getCount()).isEqualTo(expected); + long classicTotal = 0; + for (ClassicHistogramBucket b : dp.getClassicBuckets()) { + classicTotal += b.getCount(); + } + assertThat(classicTotal).isEqualTo(expected); + long nativeTotal = dp.getNativeZeroCount(); + for (NativeHistogramBucket b : dp.getNativeBucketsForPositiveValues()) { + nativeTotal += b.getCount(); + } + for (NativeHistogramBucket b : dp.getNativeBucketsForNegativeValues()) { + nativeTotal += b.getCount(); + } + assertThat(nativeTotal).isEqualTo(expected); + } + + @Test + void summaryBatchMatchesSequential() { + Summary seq = Summary.builder().name("seq").quantile(0.5).quantile(0.99).build(); + Summary batch = Summary.builder().name("batch").quantile(0.5).quantile(0.99).build(); + for (int i = 0; i < 100; i++) { + for (int k = 0; k < 4; k++) { + seq.observe(i); + } + batch.observe(i, 4); + } + SummarySnapshot.SummaryDataPointSnapshot s = seq.collect().getDataPoints().get(0); + SummarySnapshot.SummaryDataPointSnapshot b = batch.collect().getDataPoints().get(0); + assertThat(b.getCount()).isEqualTo(s.getCount()); + assertThat(b.getSum()).isEqualTo(s.getSum()); + assertThat(b.getQuantiles().get(0).getValue()).isEqualTo(s.getQuantiles().get(0).getValue()); + assertThat(b.getQuantiles().get(1).getValue()).isEqualTo(s.getQuantiles().get(1).getValue()); + + Summary plain = Summary.builder().name("plain").build(); + plain.observe(0.5, 1_000_000_000L); + SummarySnapshot.SummaryDataPointSnapshot p = plain.collect().getDataPoints().get(0); + assertThat(p.getCount()).isEqualTo(1_000_000_000L); + assertThat(p.getSum()).isEqualTo(5.0e8); + } +} From 6d9e957b2fbb58d3d59746afafec81269b2d724c Mon Sep 17 00:00:00 2001 From: Callum Donald Date: Mon, 28 Sep 2026 23:56:19 +0100 Subject: [PATCH 3/3] perf: add batched observation benchmarks to HistogramBenchmark LoopN records runs of identical values with N calls to observe(value), BatchN with one observe(value, N); the ratio is the speedup. Batch1024 and Batch1M both make ten calls per op, so comparing them shows whether the cost of a call depends on the count. Also add prometheusNativeSingleThread, mirroring the existing prometheusClassicSingleThread. Single-threaded numbers are much less noisy than the four-thread ones and are what exposes a change in per-call overhead on the native path. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Callum Donald --- .../benchmarks/HistogramBenchmark.java | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/HistogramBenchmark.java b/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/HistogramBenchmark.java index 29f730e6f..a39c220e8 100644 --- a/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/HistogramBenchmark.java +++ b/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/HistogramBenchmark.java @@ -279,6 +279,79 @@ public Histogram prometheusNative( return histogram.noLabels; } + @Benchmark + @Threads(1) + public Histogram prometheusNativeSingleThread( + RandomNumbers randomNumbers, PrometheusNativeHistogram histogram) { + for (int i = 0; i < randomNumbers.randomNumbers.length; i++) { + histogram.noLabels.observe(randomNumbers.randomNumbers[i]); + } + return histogram.noLabels; + } + + /** + * Batched observation: the same 10240 observations recorded as runs of identical values. LoopN + * records each run with N calls to observe(value), BatchN with one observe(value, N). The ratio + * between a LoopN and the matching BatchN is the speedup; comparing Batch1024 with Batch1M shows + * whether the cost of a call depends on the count. + */ + @Benchmark + @Threads(4) + public Histogram prometheusNativeLoop16( + RandomNumbers randomNumbers, PrometheusNativeHistogram histogram) { + for (int i = 0; i < randomNumbers.randomNumbers.length / 16; i++) { + double value = randomNumbers.randomNumbers[i]; + for (int k = 0; k < 16; k++) { + histogram.noLabels.observe(value); + } + } + return histogram.noLabels; + } + + @Benchmark + @Threads(4) + public Histogram prometheusNativeBatch16( + RandomNumbers randomNumbers, PrometheusNativeHistogram histogram) { + for (int i = 0; i < randomNumbers.randomNumbers.length / 16; i++) { + histogram.noLabels.observe(randomNumbers.randomNumbers[i], 16); + } + return histogram.noLabels; + } + + @Benchmark + @Threads(4) + public Histogram prometheusNativeLoop1024( + RandomNumbers randomNumbers, PrometheusNativeHistogram histogram) { + for (int i = 0; i < randomNumbers.randomNumbers.length / 1024; i++) { + double value = randomNumbers.randomNumbers[i]; + for (int k = 0; k < 1024; k++) { + histogram.noLabels.observe(value); + } + } + return histogram.noLabels; + } + + @Benchmark + @Threads(4) + public Histogram prometheusNativeBatch1024( + RandomNumbers randomNumbers, PrometheusNativeHistogram histogram) { + for (int i = 0; i < randomNumbers.randomNumbers.length / 1024; i++) { + histogram.noLabels.observe(randomNumbers.randomNumbers[i], 1024); + } + return histogram.noLabels; + } + + @Benchmark + @Threads(4) + public Histogram prometheusNativeBatch1M( + RandomNumbers randomNumbers, PrometheusNativeHistogram histogram) { + // Ten batches of a million: the whole op is ten calls, so the per-observation cost is ~0. + for (int i = 0; i < 10; i++) { + histogram.noLabels.observe(randomNumbers.randomNumbers[i], 1_000_000); + } + return histogram.noLabels; + } + @Benchmark @Threads(4) public io.prometheus.client.Histogram simpleclient(