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);