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 29f730e6f5..a39c220e88 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( 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 f87bede40c..3a4988c494 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 + * + *

+ * + *

{@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/Buffer.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java index 9283e60fc1..11671ec17f 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 d03ac9b97b..2bc450cebe 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 = @@ -244,13 +249,33 @@ 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); } } + @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)) { @@ -259,7 +284,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 +292,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 +370,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 +390,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 +481,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 d75ac6c74d..6cf28ffee6 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)) { @@ -214,13 +240,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/BatchObserveTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BatchObserveTest.java new file mode 100644 index 0000000000..d05500c8b1 --- /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); + } +} 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 064bb81863..35cc0359c5 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 0000000000..cf16eda571 --- /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); + } + } +}