From 6f7e9d0acd823bf56a347f60d1847f39c9e12cf6 Mon Sep 17 00:00:00 2001 From: Bin Fan Date: Thu, 3 Sep 2026 10:29:58 -0700 Subject: [PATCH] fix(core): clear the buffer active bit when the snapshot supplier throws Buffer.run() sets the buffer-active bit on every striped observation counter, waits for in-flight observations to land, calls createResult.get(), and then clears the bit again. The surrounding finally only releases runLock, so when createResult.get() throws, the bit is left set permanently. Two things follow from a bit that stays set: * append() keeps returning true, so observations go into the buffer instead of being recorded, and are replayed at an arbitrary later point. * The next run() derives expectedCount from counters that still carry the sign bit, so complete.apply(expectedCount) may never be satisfied. The wait loop then spins on Thread.yield() indefinitely while holding runLock, which blocks every subsequent scrape of that registry. Whether it spins or instead produces a corrupt snapshot depends on the number of stripes, because the sign-bit contributions cancel out for an even Runtime.availableProcessors(): 2 * Long.MIN_VALUE == 0. Nothing clears the bit in-process, so the only recovery is a restart. Move the wait-and-create block into its own try, with the "signal that the buffer is inactive" step and the buffer drain in the matching finally. The drain has to move with it: leaving bufferPos set makes the next run() skip its own wait and replay stale values on top of the new ones. Found in production in Alluxio on prometheus-metrics-core 1.0.0, where an OutOfMemoryError raised inside createResult.get() wedged the /metrics endpoint of several workers for 4.5 days until they were restarted. Co-Authored-By: Claude Opus 5 Signed-off-by: Bin Fan --- .../metrics/core/metrics/Buffer.java | 70 ++++++++++--------- .../metrics/core/metrics/BufferTest.java | 22 ++++++ 2 files changed, 59 insertions(+), 33 deletions(-) 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 c2017995e..04734b2df 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 @@ -95,43 +95,47 @@ T run( expectedCount += observationCount.getAndAdd(bufferActiveBit); } - while (!complete.apply(expectedCount)) { - // Wait until all in-flight threads have added their observations to the histogram / - // summary. - // we can't use a condition here, because the other thread doesn't have a lock as it's on - // the fast path. - Thread.yield(); - } - result = createResult.get(); - - // Signal that the buffer is inactive. - long expectedBufferSize = 0; - if (reset) { - for (AtomicLong observationCount : stripedObservationCounts) { - expectedBufferSize += observationCount.getAndSet(0) & ~bufferActiveBit; - } - reset = false; - } else { - for (AtomicLong observationCount : stripedObservationCounts) { - expectedBufferSize += observationCount.addAndGet(bufferActiveBit); - } - } - expectedBufferSize -= expectedCount; - - appendLock.lock(); try { - while (bufferPos < expectedBufferSize) { - // Wait until all in-flight threads have added their observations to the buffer. - bufferFilled.await(); + while (!complete.apply(expectedCount)) { + // Wait until all in-flight threads have added their observations to the histogram / + // summary. + // we can't use a condition here, because the other thread doesn't have a lock as it's on + // the fast path. + Thread.yield(); } + result = createResult.get(); } finally { - appendLock.unlock(); - } + // Signal that the buffer is inactive. This has to happen even when the block above throws: + // the bit is what makes append() buffer instead of record, so a bit left set sends every + // later run() into the wait loop above forever, holding runLock and with it the scrape. + long expectedBufferSize = 0; + if (reset) { + for (AtomicLong observationCount : stripedObservationCounts) { + expectedBufferSize += observationCount.getAndSet(0) & ~bufferActiveBit; + } + reset = false; + } else { + for (AtomicLong observationCount : stripedObservationCounts) { + expectedBufferSize += observationCount.addAndGet(bufferActiveBit); + } + } + expectedBufferSize -= expectedCount; + + appendLock.lock(); + try { + while (bufferPos < expectedBufferSize) { + // Wait until all in-flight threads have added their observations to the buffer. + bufferFilled.await(); + } + } finally { + appendLock.unlock(); + } - buffer = observationBuffer; - bufferSize = bufferPos; - observationBuffer = new double[0]; - bufferPos = 0; + buffer = observationBuffer; + bufferSize = bufferPos; + observationBuffer = new double[0]; + bufferPos = 0; + } } catch (InterruptedException e) { throw new RuntimeException(e); } finally { 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 dccd3b4eb..b5f6cca68 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 @@ -1,11 +1,33 @@ package io.prometheus.metrics.core.metrics; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import org.junit.jupiter.api.Test; class BufferTest { + @Test + void bufferIsDeactivatedWhenCreateResultThrows() { + Buffer buffer = new Buffer(); + assertThat(buffer.append(1.0)).isFalse(); + + assertThatThrownBy( + () -> + buffer.run( + count -> true, + () -> { + throw new IllegalStateException("failed to create the snapshot"); + }, + value -> {})) + .isInstanceOf(IllegalStateException.class) + .hasMessage("failed to create the snapshot"); + + // The buffer has to be inactive again. While it stayed active, append() kept buffering + // observations instead of recording them, and the next run() never left its wait loop. + assertThat(buffer.append(2.0)).isFalse(); + } + @Test void stripeIndexDoesNotOverflowWhenThreadIdNarrowsToIntegerMinValue() { assertThat(Buffer.stripeIndex(2_147_483_648L, 3)).isEqualTo(2);