From 0020301af3724c92bc4ce02b1cce0fe0ab08ebd7 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 9 Jul 2026 13:52:48 +0000 Subject: [PATCH 1/7] fix: harden scrape edge cases Signed-off-by: Gregor Zeitlinger --- ...prometheus-metrics-exporter-httpserver.txt | 4 +- .../metrics/it/exporter/test/ExporterIT.java | 6 +- .../it/exporter/test/HttpServerIT.java | 9 +++ .../config/ExporterPushgatewayProperties.java | 13 ++-- .../io/prometheus/metrics/config/Util.java | 59 +++++++++++++-- .../config/ExporterPropertiesTest.java | 6 +- .../ExporterPushgatewayPropertiesTest.java | 4 +- .../config/OpenMetrics2PropertiesTest.java | 10 +-- .../prometheus/metrics/config/UtilTest.java | 12 ++- .../metrics/core/metrics/Buffer.java | 27 ++++++- .../metrics/core/metrics/BufferTest.java | 73 +++++++++++++++++++ .../InvalidQueryParameterException.java | 8 ++ .../common/PrometheusHttpRequest.java | 31 +++++++- .../common/PrometheusScrapeHandler.java | 33 +++++++-- .../common/PrometheusScrapeHandlerTest.java | 32 +++++++- .../BlockingRejectedExecutionHandler.java | 18 ----- .../exporter/httpserver/HTTPServer.java | 27 +++---- .../httpserver/HttpExchangeAdapter.java | 58 +++++++-------- .../exporter/httpserver/HTTPServerTest.java | 37 +++++++++- .../httpserver/HttpExchangeAdapterTest.java | 23 ++++++ .../model/registry/MetricNameFilter.java | 15 +++- 21 files changed, 395 insertions(+), 110 deletions(-) create mode 100644 prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java create mode 100644 prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java delete mode 100644 prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/BlockingRejectedExecutionHandler.java diff --git a/docs/apidiffs/current_vs_latest/prometheus-metrics-exporter-httpserver.txt b/docs/apidiffs/current_vs_latest/prometheus-metrics-exporter-httpserver.txt index 17fccaa45..8bc12b84f 100644 --- a/docs/apidiffs/current_vs_latest/prometheus-metrics-exporter-httpserver.txt +++ b/docs/apidiffs/current_vs_latest/prometheus-metrics-exporter-httpserver.txt @@ -1,2 +1,4 @@ Comparing source compatibility of prometheus-metrics-exporter-httpserver-1.8.1-SNAPSHOT.jar against prometheus-metrics-exporter-httpserver-1.8.0.jar -No changes. +*** MODIFIED CLASS: PUBLIC io.prometheus.metrics.exporter.httpserver.HTTPServer (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 + diff --git a/integration-tests/it-exporter/it-exporter-test/src/test/java/io/prometheus/metrics/it/exporter/test/ExporterIT.java b/integration-tests/it-exporter/it-exporter-test/src/test/java/io/prometheus/metrics/it/exporter/test/ExporterIT.java index 5a80d8bdf..3ba88dcc1 100644 --- a/integration-tests/it-exporter/it-exporter-test/src/test/java/io/prometheus/metrics/it/exporter/test/ExporterIT.java +++ b/integration-tests/it-exporter/it-exporter-test/src/test/java/io/prometheus/metrics/it/exporter/test/ExporterIT.java @@ -157,7 +157,11 @@ void testErrorHandling() throws IOException { start("error"); Response response = scrape("GET", ""); assertThat(response.status).isEqualTo(500); - assertThat(response.stringBody()).contains("Simulating an error."); + assertErrorResponseBody(response.stringBody()); + } + + protected void assertErrorResponseBody(String body) { + assertThat(body).contains("Simulating an error."); } @Test diff --git a/integration-tests/it-exporter/it-exporter-test/src/test/java/io/prometheus/metrics/it/exporter/test/HttpServerIT.java b/integration-tests/it-exporter/it-exporter-test/src/test/java/io/prometheus/metrics/it/exporter/test/HttpServerIT.java index 4c7e61472..d04664718 100644 --- a/integration-tests/it-exporter/it-exporter-test/src/test/java/io/prometheus/metrics/it/exporter/test/HttpServerIT.java +++ b/integration-tests/it-exporter/it-exporter-test/src/test/java/io/prometheus/metrics/it/exporter/test/HttpServerIT.java @@ -1,5 +1,7 @@ package io.prometheus.metrics.it.exporter.test; +import static org.assertj.core.api.Assertions.assertThat; + import java.io.IOException; import java.net.URISyntaxException; @@ -7,4 +9,11 @@ class HttpServerIT extends ExporterIT { public HttpServerIT() throws IOException, URISyntaxException { super("exporter-httpserver-sample"); } + + @Override + protected void assertErrorResponseBody(String body) { + assertThat(body) + .isEqualTo("An internal error occurred while scraping metrics.\n") + .doesNotContain("Simulating an error."); + } } diff --git a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/ExporterPushgatewayProperties.java b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/ExporterPushgatewayProperties.java index 10e8c0f33..8accc12bb 100644 --- a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/ExporterPushgatewayProperties.java +++ b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/ExporterPushgatewayProperties.java @@ -94,9 +94,8 @@ static ExporterPushgatewayProperties load(PropertySource propertySource) if (scheme != null) { if (!scheme.equals("http") && !scheme.equals("https")) { throw new PrometheusPropertiesException( - String.format( - "%s.%s: Illegal value. Expecting 'http' or 'https'. Found: %s", - PREFIX, SCHEME, scheme)); + Util.invalidValueMessage( + PREFIX + "." + SCHEME, "Illegal value. Expecting 'http' or 'https'.", scheme)); } } @@ -119,10 +118,10 @@ static ExporterPushgatewayProperties load(PropertySource propertySource) return EscapingScheme.DOTS_ESCAPING; default: throw new PrometheusPropertiesException( - String.format( - "%s.%s: Illegal value. Expecting 'allow-utf-8', 'values', 'underscores', " - + "or 'dots'. Found: %s", - PREFIX, ESCAPING_SCHEME, scheme)); + Util.invalidValueMessage( + PREFIX + "." + ESCAPING_SCHEME, + "Illegal value. Expecting 'allow-utf-8', 'values', 'underscores', or 'dots'.", + scheme)); } } diff --git a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/Util.java b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/Util.java index 20bd75699..a266e587b 100644 --- a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/Util.java +++ b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/Util.java @@ -24,7 +24,7 @@ static Boolean loadBoolean(String prefix, String propertyName, PropertySource pr String fullKey = prefix.isEmpty() ? propertyName : prefix + "." + propertyName; if (!"true".equalsIgnoreCase(property) && !"false".equalsIgnoreCase(property)) { throw new PrometheusPropertiesException( - String.format("%s: Expecting 'true' or 'false'. Found: %s", fullKey, property)); + invalidValueMessage(fullKey, "Expecting 'true' or 'false'.", property)); } return Boolean.parseBoolean(property); } @@ -88,7 +88,8 @@ static List loadDoubleList( } } catch (NumberFormatException e) { throw new PrometheusPropertiesException( - fullKey + "=" + property + ": Expecting comma separated list of double values"); + invalidValueMessage( + fullKey, "Expecting comma separated list of double values", property)); } } return Arrays.asList(result); @@ -130,7 +131,7 @@ static Integer loadInteger(String prefix, String propertyName, PropertySource pr return Integer.parseInt(property); } catch (NumberFormatException e) { throw new PrometheusPropertiesException( - fullKey + "=" + property + ": Expecting integer value"); + invalidValueMessage(fullKey, "Expecting integer value", property)); } } return null; @@ -146,7 +147,7 @@ static Double loadDouble(String prefix, String propertyName, PropertySource prop return Double.parseDouble(property); } catch (NumberFormatException e) { throw new PrometheusPropertiesException( - fullKey + "=" + property + ": Expecting double value"); + invalidValueMessage(fullKey, "Expecting double value", property)); } } return null; @@ -162,7 +163,7 @@ static Long loadLong(String prefix, String propertyName, PropertySource property return Long.parseLong(property); } catch (NumberFormatException e) { throw new PrometheusPropertiesException( - fullKey + "=" + property + ": Expecting long value"); + invalidValueMessage(fullKey, "Expecting long value", property)); } } return null; @@ -197,4 +198,52 @@ static void assertValue( throw new PrometheusPropertiesException(fullMessage); } } + + static String invalidValueMessage(String fullKey, String message, String found) { + String separator = message.endsWith(".") ? " " : ". "; + return fullKey + ": " + message + separator + "Found: " + escape(found); + } + + static String escape(String value) { + StringBuilder result = new StringBuilder(value.length() + 2); + result.append('"'); + int maxLength = Math.min(value.length(), 100); + for (int i = 0; i < maxLength; i++) { + char c = value.charAt(i); + switch (c) { + case '\b': + result.append("\\b"); + break; + case '\t': + result.append("\\t"); + break; + case '\n': + result.append("\\n"); + break; + case '\f': + result.append("\\f"); + break; + case '\r': + result.append("\\r"); + break; + case '"': + result.append("\\\""); + break; + case '\\': + result.append("\\\\"); + break; + default: + if (c < 0x20 || c == 0x7f) { + result.append(String.format("\\u%04x", (int) c)); + } else { + result.append(c); + } + } + } + if (value.length() > maxLength) { + result.append("..."); + } + result.append('"'); + return result.toString(); + } } diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPropertiesTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPropertiesTest.java index 514c5ff52..0a5d20ef3 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPropertiesTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPropertiesTest.java @@ -28,7 +28,7 @@ void load() { Map.of("io.prometheus.exporter.include_created_timestamps", "invalid")))) .withMessage( "io.prometheus.exporter.include_created_timestamps: Expecting 'true' or 'false'. Found:" - + " invalid"); + + " \"invalid\""); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> @@ -37,7 +37,7 @@ void load() { Map.of("io.prometheus.exporter.exemplars_on_all_metric_types", "invalid")))) .withMessage( "io.prometheus.exporter.exemplars_on_all_metric_types: Expecting 'true' or 'false'." - + " Found: invalid"); + + " Found: \"invalid\""); } private static ExporterProperties load(Map map) { @@ -85,6 +85,6 @@ void prometheusTimestampsInMs() { Map.of("io.prometheus.exporter.prometheus_timestamps_in_ms", "invalid")))) .withMessage( "io.prometheus.exporter.prometheus_timestamps_in_ms: Expecting 'true' or 'false'." - + " Found: invalid"); + + " Found: \"invalid\""); } } diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPushgatewayPropertiesTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPushgatewayPropertiesTest.java index c92e6f2f9..dbec7c8a1 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPushgatewayPropertiesTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPushgatewayPropertiesTest.java @@ -27,7 +27,7 @@ void load() { .isThrownBy(() -> load(Map.of("io.prometheus.exporter.pushgateway.scheme", "foo"))) .withMessage( "io.prometheus.exporter.pushgateway.scheme: Illegal value. Expecting 'http' or 'https'." - + " Found: foo"); + + " Found: \"foo\""); } @Test @@ -60,7 +60,7 @@ void loadWithInvalidEscapingScheme() { () -> load(Map.of("io.prometheus.exporter.pushgateway.escaping_scheme", "invalid"))) .withMessage( "io.prometheus.exporter.pushgateway.escaping_scheme: Illegal value. Expecting" - + " 'allow-utf-8', 'values', 'underscores', or 'dots'. Found: invalid"); + + " 'allow-utf-8', 'values', 'underscores', or 'dots'. Found: \"invalid\""); } @Test diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/OpenMetrics2PropertiesTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/OpenMetrics2PropertiesTest.java index e7a273464..df9e9a04b 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/OpenMetrics2PropertiesTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/OpenMetrics2PropertiesTest.java @@ -38,7 +38,7 @@ void loadInvalidValue() { .isThrownBy( () -> load(new HashMap<>(Map.of("io.prometheus.openmetrics2.enabled", "invalid")))) .withMessage( - "io.prometheus.openmetrics2.enabled: Expecting 'true' or 'false'. Found: invalid"); + "io.prometheus.openmetrics2.enabled: Expecting 'true' or 'false'. Found: \"invalid\""); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> @@ -47,7 +47,7 @@ void loadInvalidValue() { Map.of("io.prometheus.openmetrics2.content_negotiation", "invalid")))) .withMessage( "io.prometheus.openmetrics2.content_negotiation: Expecting 'true' or 'false'. Found:" - + " invalid"); + + " \"invalid\""); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> @@ -56,7 +56,7 @@ void loadInvalidValue() { Map.of("io.prometheus.openmetrics2.composite_values", "invalid")))) .withMessage( "io.prometheus.openmetrics2.composite_values: Expecting 'true' or 'false'. Found:" - + " invalid"); + + " \"invalid\""); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> @@ -65,7 +65,7 @@ void loadInvalidValue() { Map.of("io.prometheus.openmetrics2.exemplar_compliance", "invalid")))) .withMessage( "io.prometheus.openmetrics2.exemplar_compliance: Expecting 'true' or 'false'. Found:" - + " invalid"); + + " \"invalid\""); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> @@ -74,7 +74,7 @@ void loadInvalidValue() { Map.of("io.prometheus.openmetrics2.native_histograms", "invalid")))) .withMessage( "io.prometheus.openmetrics2.native_histograms: Expecting 'true' or 'false'. Found:" - + " invalid"); + + " \"invalid\""); } private static OpenMetrics2Properties load(Map map) { diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/UtilTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/UtilTest.java index e4d7fa829..3fadf15c2 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/UtilTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/UtilTest.java @@ -51,6 +51,16 @@ void loadOptionalDuration_invalidNumber_throws() { assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy(() -> Util.loadOptionalDuration("", "foo", propertySource)) - .withMessage("foo=abc: Expecting long value"); + .withMessage("foo: Expecting long value. Found: \"abc\""); + } + + @Test + void invalidValueMessageEscapesRawValue() { + Map regularProperties = new HashMap<>(Map.of("foo", "bad\n\"value")); + PropertySource propertySource = new PropertySource(regularProperties); + + assertThatExceptionOfType(PrometheusPropertiesException.class) + .isThrownBy(() -> Util.loadBoolean("", "foo", propertySource)) + .withMessage("foo: Expecting 'true' or 'false'. Found: \"bad\\n\\\"value\""); } } 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 1c47f867c..b3a52652c 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 @@ -2,6 +2,7 @@ import io.prometheus.metrics.model.snapshots.DataPointSnapshot; import java.util.Arrays; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; @@ -18,6 +19,7 @@ class Buffer { private static final long bufferActiveBit = 1L << 63; + private static final long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(1); // 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 improve, we stripe the AtomicLong into N instances, @@ -34,8 +36,14 @@ class Buffer { ReentrantLock appendLock = new ReentrantLock(); ReentrantLock runLock = new ReentrantLock(); Condition bufferFilled = appendLock.newCondition(); + private final long maxSpinWaitNanos; Buffer() { + this(DEFAULT_MAX_SPIN_WAIT_NANOS); + } + + Buffer(long maxSpinWaitNanos) { + this.maxSpinWaitNanos = maxSpinWaitNanos; stripedObservationCounts = new AtomicLong[Runtime.getRuntime().availableProcessors()]; for (int i = 0; i < stripedObservationCounts.length; i++) { stripedObservationCounts[i] = new AtomicLong(0); @@ -43,7 +51,7 @@ class Buffer { } boolean append(double value) { - int index = Math.abs((int) Thread.currentThread().getId()) % stripedObservationCounts.length; + int index = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length); AtomicLong observationCountForThread = stripedObservationCounts[index]; long count = observationCountForThread.incrementAndGet(); if ((count & bufferActiveBit) == 0) { @@ -54,6 +62,10 @@ boolean append(double value) { } } + static int stripeIndex(long threadId, int stripeCount) { + return (int) Math.floorMod(threadId, stripeCount); + } + private void doAppend(double amount) { appendLock.lock(); try { @@ -74,7 +86,7 @@ void reset() { reset = true; } - @SuppressWarnings("ThreadPriorityCheck") + @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) T run( Function complete, Supplier createResult, @@ -82,6 +94,7 @@ T run( double[] buffer; int bufferSize; T result; + boolean timedOut = false; runLock.lock(); try { @@ -91,14 +104,19 @@ T run( expectedCount += observationCount.getAndAdd(bufferActiveBit); } + long deadline = System.nanoTime() + maxSpinWaitNanos; 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. + if (System.nanoTime() - deadline >= 0) { + timedOut = true; + break; + } Thread.yield(); } - result = createResult.get(); + result = timedOut ? null : createResult.get(); // Signal that the buffer is inactive. long expectedBufferSize = 0; @@ -137,6 +155,9 @@ T run( for (int i = 0; i < bufferSize; i++) { observeFunction.accept(buffer[i]); } + if (timedOut) { + throw new IllegalStateException("Timed out while waiting for in-flight observations."); + } return result; } } 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 new file mode 100644 index 000000000..7f8e09dbf --- /dev/null +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -0,0 +1,73 @@ +package io.prometheus.metrics.core.metrics; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; + +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.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; + +class BufferTest { + + @Test + void stripeIndexDoesNotOverflowWhenThreadIdNarrowsToIntegerMinValue() { + assertThat(Buffer.stripeIndex(2_147_483_648L, 3)).isEqualTo(2); + assertThat(Buffer.stripeIndex(2_147_483_648L, 6)).isEqualTo(2); + assertThat(Buffer.stripeIndex(2_147_483_648L, 12)).isEqualTo(8); + } + + @Test + void timeoutDeactivatesBufferAndReplaysBufferedObservations() throws InterruptedException { + Buffer buffer = new Buffer(TimeUnit.SECONDS.toNanos(1)); + CountDownLatch spinWaitStarted = new CountDownLatch(1); + AtomicBoolean keepWaiting = new AtomicBoolean(true); + List replayedObservations = new ArrayList<>(); + AtomicBoolean timedOut = new AtomicBoolean(false); + + Thread runner = + new Thread( + () -> { + try { + buffer.run( + expectedCount -> { + spinWaitStarted.countDown(); + return !keepWaiting.get(); + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + replayedObservations::add); + } catch (IllegalStateException expected) { + timedOut.set(true); + } + }); + runner.start(); + + assertThat(spinWaitStarted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(buffer.append(1.0)).isTrue(); + runner.join(5_000); + + assertThat(timedOut).isTrue(); + assertThat(replayedObservations).containsExactly(1.0); + assertThat(buffer.append(2.0)).isFalse(); + } + + @Test + void timeoutDoesNotCreateSnapshot() { + Buffer buffer = new Buffer(TimeUnit.MILLISECONDS.toNanos(1)); + + assertThatExceptionOfType(IllegalStateException.class) + .isThrownBy( + () -> + buffer.run( + expectedCount -> false, + () -> { + throw new AssertionError("snapshot should not be created"); + }, + ignored -> {})) + .withMessage("Timed out while waiting for in-flight observations."); + } +} diff --git a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java new file mode 100644 index 000000000..0a22300f6 --- /dev/null +++ b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java @@ -0,0 +1,8 @@ +package io.prometheus.metrics.exporter.common; + +class InvalidQueryParameterException extends RuntimeException { + + InvalidQueryParameterException(String message) { + super(message); + } +} diff --git a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequest.java b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequest.java index a0c692c23..f8cc39e41 100644 --- a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequest.java +++ b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequest.java @@ -49,15 +49,42 @@ default String getParameter(String name) { @SuppressWarnings("JdkObsolete") default String[] getParameterValues(String name) { try { + int maxQueryStringLength = 64 * 1024; + int maxQueryParameterCount = 1024; ArrayList result = new ArrayList<>(); String queryString = getQueryString(); if (queryString != null) { - String[] pairs = queryString.split("&"); - for (String pair : pairs) { + if (queryString.length() > maxQueryStringLength) { + throw new InvalidQueryParameterException( + "Query string too long: " + + queryString.length() + + " characters (max " + + maxQueryStringLength + + ")"); + } + int start = 0; + int parameterCount = 0; + for (int end = queryString.indexOf('&'); start <= queryString.length(); ) { + parameterCount++; + if (parameterCount > maxQueryParameterCount) { + throw new InvalidQueryParameterException( + "Too many query parameters: " + + parameterCount + + " (max " + + maxQueryParameterCount + + ")"); + } + String pair = + end == -1 ? queryString.substring(start) : queryString.substring(start, end); int idx = pair.indexOf("="); if (idx != -1 && URLDecoder.decode(pair.substring(0, idx), "UTF-8").equals(name)) { result.add(URLDecoder.decode(pair.substring(idx + 1), "UTF-8")); } + if (end == -1) { + break; + } + start = end + 1; + end = queryString.indexOf('&', start); } } if (result.isEmpty()) { diff --git a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusScrapeHandler.java b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusScrapeHandler.java index 20328382a..d97a90205 100644 --- a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusScrapeHandler.java +++ b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusScrapeHandler.java @@ -60,10 +60,19 @@ public PrometheusScrapeHandler(PrometheusProperties config, PrometheusRegistry r public void handleRequest(PrometheusHttpExchange exchange) throws IOException { try { PrometheusHttpRequest request = exchange.getRequest(); - MetricSnapshots snapshots = scrape(request); + String[] includedNames = null; + String debugParam = null; + try { + includedNames = request.getParameterValues("name[]"); + debugParam = request.getParameter("debug"); + } catch (InvalidQueryParameterException e) { + writeInvalidQueryParametersResponse(exchange); + return; + } + MetricSnapshots snapshots = scrape(request, includedNames); String acceptHeader = request.getHeader("Accept"); EscapingScheme escapingScheme = EscapingScheme.fromAcceptHeader(acceptHeader); - if (writeDebugResponse(snapshots, escapingScheme, exchange)) { + if (writeDebugResponse(snapshots, escapingScheme, debugParam, exchange)) { return; } ExpositionFormatWriter writer = expositionFormats.findWriter(acceptHeader); @@ -136,9 +145,9 @@ private Predicate makeNameFilter(@Nullable String[] includedNames) { return result; } - private MetricSnapshots scrape(PrometheusHttpRequest request) { + private MetricSnapshots scrape(PrometheusHttpRequest request, @Nullable String[] includedNames) { - Predicate filter = makeNameFilter(request.getParameterValues("name[]")); + Predicate filter = makeNameFilter(includedNames); if (filter != null) { return registry.scrape(filter, request); } else { @@ -147,9 +156,11 @@ private MetricSnapshots scrape(PrometheusHttpRequest request) { } private boolean writeDebugResponse( - MetricSnapshots snapshots, EscapingScheme escapingScheme, PrometheusHttpExchange exchange) + MetricSnapshots snapshots, + EscapingScheme escapingScheme, + @Nullable String debugParam, + PrometheusHttpExchange exchange) throws IOException { - String debugParam = exchange.getRequest().getParameter("debug"); PrometheusHttpResponse response = exchange.getResponse(); if (debugParam == null) { return false; @@ -184,6 +195,16 @@ private boolean writeDebugResponse( } } + private void writeInvalidQueryParametersResponse(PrometheusHttpExchange exchange) + throws IOException { + PrometheusHttpResponse response = exchange.getResponse(); + response.setHeader("Content-Type", "text/plain; charset=utf-8"); + byte[] message = "Invalid query parameters".getBytes(StandardCharsets.UTF_8); + try (OutputStream outputStream = response.sendHeadersAndGetBody(400, message.length)) { + outputStream.write(message); + } + } + private boolean shouldUseCompression(PrometheusHttpRequest request) { if (preferUncompressedResponse) { return false; diff --git a/prometheus-metrics-exporter-common/src/test/java/io/prometheus/metrics/exporter/common/PrometheusScrapeHandlerTest.java b/prometheus-metrics-exporter-common/src/test/java/io/prometheus/metrics/exporter/common/PrometheusScrapeHandlerTest.java index 07ea8ac52..4f190b61f 100644 --- a/prometheus-metrics-exporter-common/src/test/java/io/prometheus/metrics/exporter/common/PrometheusScrapeHandlerTest.java +++ b/prometheus-metrics-exporter-common/src/test/java/io/prometheus/metrics/exporter/common/PrometheusScrapeHandlerTest.java @@ -164,6 +164,36 @@ void testMultipleMetricNameFilters() throws IOException { assertThat(body).doesNotContain("metric_three"); } + @Test + void testRejectsTooManyQueryParameters() throws IOException { + StringBuilder queryString = new StringBuilder("name[]=test_counter"); + for (int i = 0; i < 1024; i++) { + queryString.append("&name[]=metric_").append(i); + } + + TestHttpExchange exchange = new TestHttpExchange("GET", queryString.toString()); + handler.handleRequest(exchange); + + assertThat(exchange.getResponseCode()).isEqualTo(400); + assertThat(exchange.getResponseHeaders().get("Content-Type")) + .isEqualTo("text/plain; charset=utf-8"); + assertThat(exchange.getResponseBody()).isEqualTo("Invalid query parameters"); + } + + @Test + void testRejectsTooLongQueryString() throws IOException { + StringBuilder queryString = new StringBuilder("name[]="); + for (int i = 0; i < 64 * 1024; i++) { + queryString.append("a"); + } + + TestHttpExchange exchange = new TestHttpExchange("GET", queryString.toString()); + handler.handleRequest(exchange); + + assertThat(exchange.getResponseCode()).isEqualTo(400); + assertThat(exchange.getResponseBody()).isEqualTo("Invalid query parameters"); + } + /** Test implementation of PrometheusHttpExchange for testing. */ private static class TestHttpExchange implements PrometheusHttpExchange { private final TestHttpRequest request; @@ -216,7 +246,7 @@ public Map getResponseHeaders() { } public String getResponseBody() { - return rawResponseBody.toString(StandardCharsets.UTF_8); + return new String(rawResponseBody.toByteArray(), StandardCharsets.UTF_8); } public boolean isGzipCompressed() { diff --git a/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/BlockingRejectedExecutionHandler.java b/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/BlockingRejectedExecutionHandler.java deleted file mode 100644 index 023d3f2f0..000000000 --- a/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/BlockingRejectedExecutionHandler.java +++ /dev/null @@ -1,18 +0,0 @@ -package io.prometheus.metrics.exporter.httpserver; - -import java.util.concurrent.RejectedExecutionHandler; -import java.util.concurrent.ThreadPoolExecutor; - -class BlockingRejectedExecutionHandler implements RejectedExecutionHandler { - - @Override - public void rejectedExecution(Runnable runnable, ThreadPoolExecutor threadPoolExecutor) { - if (!threadPoolExecutor.isShutdown()) { - try { - threadPoolExecutor.getQueue().put(runnable); - } catch (InterruptedException ignored) { - // ignore - } - } - } -} diff --git a/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java b/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java index e93b122b0..fe55e6e9d 100644 --- a/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java +++ b/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java @@ -12,14 +12,13 @@ import io.prometheus.metrics.model.registry.PrometheusRegistry; import java.io.Closeable; import java.io.IOException; -import java.io.InputStream; import java.net.InetAddress; import java.net.InetSocketAddress; import java.security.PrivilegedActionException; import java.security.PrivilegedExceptionAction; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import javax.annotation.Nullable; @@ -39,6 +38,10 @@ @StableApi public class HTTPServer implements Closeable { + private static final int DEFAULT_MIN_THREADS = 1; + private static final int DEFAULT_MAX_THREADS = 10; + private static final int DEFAULT_QUEUE_SIZE = 100; + static { if (!System.getProperties().containsKey("sun.net.httpserver.maxReqTime")) { System.setProperty("sun.net.httpserver.maxReqTime", "60"); @@ -153,22 +156,13 @@ public void handle(HttpExchange exchange) throws IOException { } } } else { - drainInputAndClose(exchange); + exchange.getRequestBody().close(); exchange.sendResponseHeaders(403, -1); } } }; } - private void drainInputAndClose(HttpExchange httpExchange) throws IOException { - InputStream inputStream = httpExchange.getRequestBody(); - byte[] b = new byte[4096]; - while (inputStream.read(b) != -1) { - // nop - } - inputStream.close(); - } - /** Stop the HTTP server. Same as {@link #close()}. */ public void stop() { close(); @@ -337,13 +331,12 @@ private ExecutorService makeExecutorService() { return executorService; } else { return new ThreadPoolExecutor( - 1, - 10, + DEFAULT_MIN_THREADS, + DEFAULT_MAX_THREADS, 120, TimeUnit.SECONDS, - new SynchronousQueue<>(true), - NamedDaemonThreadFactory.defaultThreadFactory(true), - new BlockingRejectedExecutionHandler()); + new ArrayBlockingQueue<>(DEFAULT_QUEUE_SIZE), + NamedDaemonThreadFactory.defaultThreadFactory(true)); } } diff --git a/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HttpExchangeAdapter.java b/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HttpExchangeAdapter.java index df99837cb..b18dbb83d 100644 --- a/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HttpExchangeAdapter.java +++ b/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HttpExchangeAdapter.java @@ -6,8 +6,6 @@ import io.prometheus.metrics.exporter.common.PrometheusHttpResponse; import java.io.IOException; import java.io.OutputStream; -import java.io.PrintWriter; -import java.io.StringWriter; import java.net.URI; import java.nio.charset.StandardCharsets; import java.util.Collections; @@ -18,6 +16,10 @@ public class HttpExchangeAdapter implements PrometheusHttpExchange { + private static final Logger logger = Logger.getLogger(HttpExchangeAdapter.class.getName()); + private static final byte[] ERROR_RESPONSE = + "An internal error occurred while scraping metrics.\n".getBytes(StandardCharsets.UTF_8); + private final HttpExchange httpExchange; private final HttpRequest request = new HttpRequest(); private final HttpResponse response = new HttpResponse(); @@ -92,52 +94,42 @@ public HttpResponse getResponse() { @Override public void handleException(IOException e) throws IOException { - sendErrorResponseWithStackTrace(e); + sendErrorResponse(e); } @Override public void handleException(RuntimeException e) { - sendErrorResponseWithStackTrace(e); + sendErrorResponse(e); } - private void sendErrorResponseWithStackTrace(Exception requestHandlerException) { + private void sendErrorResponse(Exception requestHandlerException) { if (!responseSent) { responseSent = true; + logger.log( + Level.SEVERE, + "The Prometheus metrics HTTPServer caught an Exception during scrape.", + requestHandlerException); try { - StringWriter stringWriter = new StringWriter(); - PrintWriter printWriter = new PrintWriter(stringWriter); - printWriter.write("An Exception occurred while scraping metrics: "); - requestHandlerException.printStackTrace(new PrintWriter(printWriter)); - byte[] stackTrace = stringWriter.toString().getBytes(StandardCharsets.UTF_8); httpExchange.getResponseHeaders().set("Content-Type", "text/plain; charset=utf-8"); - httpExchange.sendResponseHeaders(500, stackTrace.length); - httpExchange.getResponseBody().write(stackTrace); + httpExchange.sendResponseHeaders(500, ERROR_RESPONSE.length); + httpExchange.getResponseBody().write(ERROR_RESPONSE); } catch (IOException errorWriterException) { - // We want to avoid logging so that we don't mess with application logs when the HTTPServer - // is used in a Java agent. - // However, if we can't even send an error response to the client there's nothing we can do - // but logging a message. - Logger.getLogger(this.getClass().getName()) - .log( - Level.SEVERE, - "The Prometheus metrics HTTPServer caught an Exception during scrape and " - + "failed to send an error response to the client.", - errorWriterException); - Logger.getLogger(this.getClass().getName()) - .log( - Level.SEVERE, - "Original Exception that caused the Prometheus scrape error:", - requestHandlerException); + // If we can't even send an error response to the client, logging is the only remaining + // signal. + logger.log( + Level.SEVERE, + "The Prometheus metrics HTTPServer caught an Exception during scrape and " + + "failed to send an error response to the client.", + errorWriterException); } } else { // If the exception occurs after response headers have been sent, it's too late to respond // with HTTP 500. - Logger.getLogger(this.getClass().getName()) - .log( - Level.SEVERE, - "The Prometheus metrics HTTPServer caught an Exception while trying to send " - + "the metrics response.", - requestHandlerException); + logger.log( + Level.SEVERE, + "The Prometheus metrics HTTPServer caught an Exception while trying to send " + + "the metrics response.", + requestHandlerException); } } diff --git a/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HTTPServerTest.java b/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HTTPServerTest.java index ff2d55048..f0f8ffd83 100644 --- a/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HTTPServerTest.java +++ b/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HTTPServerTest.java @@ -25,6 +25,7 @@ import java.security.Principal; import java.util.List; import java.util.concurrent.Executors; +import java.util.concurrent.ThreadPoolExecutor; import javax.net.ssl.SSLContext; import javax.security.auth.Subject; import org.junit.jupiter.api.BeforeEach; @@ -158,6 +159,21 @@ void metricsCustomRootPath() throws Exception { "my_counter_total 1.0"); } + @Test + void defaultExecutorHasBoundedQueueAndNonBlockingRejection() throws Exception { + HTTPServer server = HTTPServer.builder().port(0).buildAndStart(); + try { + assertThat(server.executorService).isInstanceOf(ThreadPoolExecutor.class); + ThreadPoolExecutor executor = (ThreadPoolExecutor) server.executorService; + assertThat(executor.getMaximumPoolSize()).isEqualTo(10); + assertThat(executor.getQueue().remainingCapacity()).isEqualTo(100); + assertThat(executor.getRejectedExecutionHandler()) + .isInstanceOf(ThreadPoolExecutor.AbortPolicy.class); + } finally { + server.stop(); + } + } + @Test void registryThrows() throws Exception { HTTPServer server = @@ -171,7 +187,13 @@ public MetricSnapshots scrape(PrometheusScrapeRequest scrapeRequest) { } }) .buildAndStart(); - run(server, "/metrics", 500, "An Exception occurred while scraping metrics"); + run( + server, + "/metrics", + 500, + "An internal error occurred while scraping metrics.", + "IllegalStateException", + "test"); } @Test @@ -237,6 +259,16 @@ void healthDisabled() throws Exception { private static void run( HTTPServer server, String path, int expectedStatusCode, String expectedBody) throws Exception { + run(server, path, expectedStatusCode, expectedBody, new String[0]); + } + + private static void run( + HTTPServer server, + String path, + int expectedStatusCode, + String expectedBody, + String... unexpectedBody) + throws Exception { // we cannot use try-with-resources or even client.close(), or the test will fail with Java 17 @SuppressWarnings("resource") final HttpClient client = HttpClient.newBuilder().build(); @@ -248,6 +280,9 @@ private static void run( client.send(request, HttpResponse.BodyHandlers.ofString()); assertThat(response.statusCode()).isEqualTo(expectedStatusCode); assertThat(response.body()).contains(expectedBody); + if (unexpectedBody.length > 0) { + assertThat(response.body()).doesNotContain(unexpectedBody); + } } finally { server.stop(); } diff --git a/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HttpExchangeAdapterTest.java b/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HttpExchangeAdapterTest.java index 19bc0d66e..8aaef9c42 100644 --- a/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HttpExchangeAdapterTest.java +++ b/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HttpExchangeAdapterTest.java @@ -2,11 +2,14 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import com.sun.net.httpserver.Headers; import com.sun.net.httpserver.HttpExchange; +import java.io.ByteArrayOutputStream; import java.net.URI; +import java.nio.charset.StandardCharsets; import java.util.List; import org.junit.jupiter.api.Test; @@ -46,4 +49,24 @@ void getHeadersWhenNotPresent() { HttpExchangeAdapter adapter = new HttpExchangeAdapter(httpExchange); assertThat(adapter.getRequest().getHeaders("Accept").hasMoreElements()).isFalse(); } + + @Test + void handleExceptionReturnsGenericMessageWithoutStackTrace() throws Exception { + HttpExchange httpExchange = mock(HttpExchange.class); + Headers headers = new Headers(); + ByteArrayOutputStream responseBody = new ByteArrayOutputStream(); + when(httpExchange.getResponseHeaders()).thenReturn(headers); + when(httpExchange.getResponseBody()).thenReturn(responseBody); + HttpExchangeAdapter adapter = new HttpExchangeAdapter(httpExchange); + + adapter.handleException(new IllegalStateException("secret failure")); + + String body = new String(responseBody.toByteArray(), StandardCharsets.UTF_8); + assertThat(body).isEqualTo("An internal error occurred while scraping metrics.\n"); + assertThat(body).doesNotContain("IllegalStateException"); + assertThat(body).doesNotContain("secret failure"); + assertThat(body).doesNotContain("at "); + assertThat(headers.getFirst("Content-Type")).isEqualTo("text/plain; charset=utf-8"); + verify(httpExchange).sendResponseHeaders(500, body.getBytes(StandardCharsets.UTF_8).length); + } } diff --git a/prometheus-metrics-model/src/main/java/io/prometheus/metrics/model/registry/MetricNameFilter.java b/prometheus-metrics-model/src/main/java/io/prometheus/metrics/model/registry/MetricNameFilter.java index 642f27150..ded171500 100644 --- a/prometheus-metrics-model/src/main/java/io/prometheus/metrics/model/registry/MetricNameFilter.java +++ b/prometheus-metrics-model/src/main/java/io/prometheus/metrics/model/registry/MetricNameFilter.java @@ -6,6 +6,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.LinkedHashSet; import java.util.function.Predicate; import javax.annotation.Nullable; @@ -26,10 +27,10 @@ private MetricNameFilter( Collection nameIsNotEqualTo, Collection nameStartsWith, Collection nameDoesNotStartWith) { - this.nameIsEqualTo = unmodifiableCollection(new ArrayList<>(nameIsEqualTo)); - this.nameIsNotEqualTo = unmodifiableCollection(new ArrayList<>(nameIsNotEqualTo)); - this.nameStartsWith = unmodifiableCollection(new ArrayList<>(nameStartsWith)); - this.nameDoesNotStartWith = unmodifiableCollection(new ArrayList<>(nameDoesNotStartWith)); + this.nameIsEqualTo = unmodifiableCollection(new LinkedHashSet<>(nameIsEqualTo)); + this.nameIsNotEqualTo = unmodifiableCollection(new LinkedHashSet<>(nameIsNotEqualTo)); + this.nameStartsWith = unmodifiableCollection(new LinkedHashSet<>(nameStartsWith)); + this.nameDoesNotStartWith = unmodifiableCollection(new LinkedHashSet<>(nameDoesNotStartWith)); } @Override @@ -44,6 +45,9 @@ private boolean matchesNameEqualTo(String metricName) { if (nameIsEqualTo.isEmpty()) { return true; } + if (nameIsEqualTo.contains(metricName)) { + return true; + } for (String name : nameIsEqualTo) { // The following ignores suffixes like _total. // "request_count" and "request_count_total" both match a metric named "request_count". @@ -58,6 +62,9 @@ private boolean matchesNameNotEqualTo(String metricName) { if (nameIsNotEqualTo.isEmpty()) { return false; } + if (nameIsNotEqualTo.contains(metricName)) { + return true; + } for (String name : nameIsNotEqualTo) { // The following ignores suffixes like _total. // "request_count" and "request_count_total" both match a metric named "request_count". From 92115dff5293a5e3fa92c77309c4374240e33043 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 10 Jul 2026 17:38:10 +0000 Subject: [PATCH 2/7] fix: update stale prometheus-metrics-exporter-common API diff --- .../current_vs_latest/prometheus-metrics-exporter-common.txt | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/apidiffs/current_vs_latest/prometheus-metrics-exporter-common.txt b/docs/apidiffs/current_vs_latest/prometheus-metrics-exporter-common.txt index e83e9a6f5..daa5ef822 100644 --- a/docs/apidiffs/current_vs_latest/prometheus-metrics-exporter-common.txt +++ b/docs/apidiffs/current_vs_latest/prometheus-metrics-exporter-common.txt @@ -1,2 +1,4 @@ Comparing source compatibility of prometheus-metrics-exporter-common-1.8.1-SNAPSHOT.jar against prometheus-metrics-exporter-common-1.8.0.jar -No changes. +*** MODIFIED CLASS: PUBLIC io.prometheus.metrics.exporter.common.PrometheusScrapeHandler (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 + From 49b71f4782043a26c231e4f594420a54821ca8b4 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Wed, 15 Jul 2026 13:33:06 +0000 Subject: [PATCH 3/7] fix: address scrape hardening review Signed-off-by: Gregor Zeitlinger --- .../common/InvalidQueryParameterException.java | 13 +++++++++++++ .../exporter/common/PrometheusHttpRequest.java | 6 +++--- .../common/PrometheusHttpRequestTest.java | 17 +++++++++++++++++ .../metrics/exporter/httpserver/HTTPServer.java | 3 ++- .../exporter/httpserver/HTTPServerTest.java | 3 ++- 5 files changed, 37 insertions(+), 5 deletions(-) diff --git a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java index 0a22300f6..0622952e7 100644 --- a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java +++ b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java @@ -1,8 +1,21 @@ package io.prometheus.metrics.exporter.common; +import java.io.UnsupportedEncodingException; +import java.net.URLDecoder; + class InvalidQueryParameterException extends RuntimeException { InvalidQueryParameterException(String message) { super(message); } + + // decode with Charset is only available in Java 10+, but we want to support Java 8 + @SuppressWarnings("JdkObsolete") + static String urlDecode(String value) throws UnsupportedEncodingException { + try { + return URLDecoder.decode(value, "UTF-8"); + } catch (IllegalArgumentException e) { + throw new InvalidQueryParameterException("Invalid percent-encoding in query string"); + } + } } diff --git a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequest.java b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequest.java index f8cc39e41..31ce2cc73 100644 --- a/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequest.java +++ b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequest.java @@ -3,7 +3,6 @@ import io.prometheus.metrics.annotations.StableApi; import io.prometheus.metrics.model.registry.PrometheusScrapeRequest; import java.io.UnsupportedEncodingException; -import java.net.URLDecoder; import java.util.ArrayList; import java.util.Enumeration; import javax.annotation.Nullable; @@ -77,8 +76,9 @@ default String[] getParameterValues(String name) { String pair = end == -1 ? queryString.substring(start) : queryString.substring(start, end); int idx = pair.indexOf("="); - if (idx != -1 && URLDecoder.decode(pair.substring(0, idx), "UTF-8").equals(name)) { - result.add(URLDecoder.decode(pair.substring(idx + 1), "UTF-8")); + if (idx != -1 + && InvalidQueryParameterException.urlDecode(pair.substring(0, idx)).equals(name)) { + result.add(InvalidQueryParameterException.urlDecode(pair.substring(idx + 1))); } if (end == -1) { break; diff --git a/prometheus-metrics-exporter-common/src/test/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequestTest.java b/prometheus-metrics-exporter-common/src/test/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequestTest.java index a2b630ecd..c2af705a4 100644 --- a/prometheus-metrics-exporter-common/src/test/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequestTest.java +++ b/prometheus-metrics-exporter-common/src/test/java/io/prometheus/metrics/exporter/common/PrometheusHttpRequestTest.java @@ -1,6 +1,7 @@ package io.prometheus.metrics.exporter.common; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import java.util.Collections; import java.util.Enumeration; @@ -80,6 +81,22 @@ void testGetParameterValuesIgnoresParametersWithoutEquals() { assertThat(values).containsExactly("value1", "value2"); } + @Test + void testGetParameterValuesRejectsMalformedEncodedName() { + PrometheusHttpRequest request = new TestPrometheusHttpRequest("name%ZZ=value"); + + assertThatExceptionOfType(InvalidQueryParameterException.class) + .isThrownBy(() -> request.getParameterValues("name")); + } + + @Test + void testGetParameterValuesRejectsMalformedEncodedValue() { + PrometheusHttpRequest request = new TestPrometheusHttpRequest("name=value%ZZ"); + + assertThatExceptionOfType(InvalidQueryParameterException.class) + .isThrownBy(() -> request.getParameterValues("name")); + } + /** Test implementation of PrometheusHttpRequest for testing default methods. */ private static class TestPrometheusHttpRequest implements PrometheusHttpRequest { private final String queryString; diff --git a/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java b/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java index fe55e6e9d..ff43144f0 100644 --- a/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java +++ b/prometheus-metrics-exporter-httpserver/src/main/java/io/prometheus/metrics/exporter/httpserver/HTTPServer.java @@ -38,7 +38,7 @@ @StableApi public class HTTPServer implements Closeable { - private static final int DEFAULT_MIN_THREADS = 1; + private static final int DEFAULT_MIN_THREADS = 10; private static final int DEFAULT_MAX_THREADS = 10; private static final int DEFAULT_QUEUE_SIZE = 100; @@ -158,6 +158,7 @@ public void handle(HttpExchange exchange) throws IOException { } else { exchange.getRequestBody().close(); exchange.sendResponseHeaders(403, -1); + exchange.close(); } } }; diff --git a/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HTTPServerTest.java b/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HTTPServerTest.java index f0f8ffd83..5585dc2ab 100644 --- a/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HTTPServerTest.java +++ b/prometheus-metrics-exporter-httpserver/src/test/java/io/prometheus/metrics/exporter/httpserver/HTTPServerTest.java @@ -46,7 +46,7 @@ void setUp() { } @Test - public void testSubjectDoAs() throws Exception { + void testSubjectDoAs() throws Exception { final String user = "joe"; final Subject subject = new Subject(); subject.getPrincipals().add(() -> user); @@ -165,6 +165,7 @@ void defaultExecutorHasBoundedQueueAndNonBlockingRejection() throws Exception { try { assertThat(server.executorService).isInstanceOf(ThreadPoolExecutor.class); ThreadPoolExecutor executor = (ThreadPoolExecutor) server.executorService; + assertThat(executor.getCorePoolSize()).isEqualTo(10); assertThat(executor.getMaximumPoolSize()).isEqualTo(10); assertThat(executor.getQueue().remainingCapacity()).isEqualTo(100); assertThat(executor.getRejectedExecutionHandler()) From 30e3a2351f265226365ec1ab1e6802619726d1d8 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 08:37:42 +0000 Subject: [PATCH 4/7] fix: harden scrape buffering and redact config errors Signed-off-by: Gregor Zeitlinger --- .../config/ExporterPushgatewayProperties.java | 5 +- .../config/PrometheusPropertiesLoader.java | 3 +- .../io/prometheus/metrics/config/Util.java | 59 +---- .../config/ExporterPropertiesTest.java | 9 +- .../ExporterPushgatewayPropertiesTest.java | 4 +- .../config/OpenMetrics2PropertiesTest.java | 17 +- .../PrometheusPropertiesLoaderTest.java | 5 +- .../prometheus/metrics/config/UtilTest.java | 10 +- .../metrics/core/metrics/Buffer.java | 207 ++++++++++-------- .../metrics/core/metrics/Histogram.java | 10 +- .../metrics/core/metrics/BufferTest.java | 43 ++++ .../exporter/pushgateway/PushGateway.java | 4 +- .../exporter/pushgateway/PushGatewayTest.java | 10 + 13 files changed, 209 insertions(+), 177 deletions(-) diff --git a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/ExporterPushgatewayProperties.java b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/ExporterPushgatewayProperties.java index 8accc12bb..e97ade191 100644 --- a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/ExporterPushgatewayProperties.java +++ b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/ExporterPushgatewayProperties.java @@ -95,7 +95,7 @@ static ExporterPushgatewayProperties load(PropertySource propertySource) if (!scheme.equals("http") && !scheme.equals("https")) { throw new PrometheusPropertiesException( Util.invalidValueMessage( - PREFIX + "." + SCHEME, "Illegal value. Expecting 'http' or 'https'.", scheme)); + PREFIX + "." + SCHEME, "Illegal value. Expecting 'http' or 'https'.")); } } @@ -120,8 +120,7 @@ static ExporterPushgatewayProperties load(PropertySource propertySource) throw new PrometheusPropertiesException( Util.invalidValueMessage( PREFIX + "." + ESCAPING_SCHEME, - "Illegal value. Expecting 'allow-utf-8', 'values', 'underscores', or 'dots'.", - scheme)); + "Illegal value. Expecting 'allow-utf-8', 'values', 'underscores', or 'dots'.")); } } diff --git a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/PrometheusPropertiesLoader.java b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/PrometheusPropertiesLoader.java index 6d52b71ef..04bbfe4a5 100644 --- a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/PrometheusPropertiesLoader.java +++ b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/PrometheusPropertiesLoader.java @@ -169,8 +169,7 @@ private static Properties loadPropertiesFromFile() throws PrometheusPropertiesEx try (InputStream stream = Files.newInputStream(Paths.get(path))) { properties.load(stream); } catch (IOException e) { - throw new PrometheusPropertiesException( - "Failed to read Prometheus properties from " + path + ": " + e.getMessage(), e); + throw new PrometheusPropertiesException("Failed to read Prometheus properties file.", e); } } return properties; diff --git a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/Util.java b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/Util.java index a266e587b..d90810faf 100644 --- a/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/Util.java +++ b/prometheus-metrics-config/src/main/java/io/prometheus/metrics/config/Util.java @@ -24,7 +24,7 @@ static Boolean loadBoolean(String prefix, String propertyName, PropertySource pr String fullKey = prefix.isEmpty() ? propertyName : prefix + "." + propertyName; if (!"true".equalsIgnoreCase(property) && !"false".equalsIgnoreCase(property)) { throw new PrometheusPropertiesException( - invalidValueMessage(fullKey, "Expecting 'true' or 'false'.", property)); + invalidValueMessage(fullKey, "Expecting 'true' or 'false'.")); } return Boolean.parseBoolean(property); } @@ -88,8 +88,7 @@ static List loadDoubleList( } } catch (NumberFormatException e) { throw new PrometheusPropertiesException( - invalidValueMessage( - fullKey, "Expecting comma separated list of double values", property)); + invalidValueMessage(fullKey, "Expecting comma separated list of double values")); } } return Arrays.asList(result); @@ -131,7 +130,7 @@ static Integer loadInteger(String prefix, String propertyName, PropertySource pr return Integer.parseInt(property); } catch (NumberFormatException e) { throw new PrometheusPropertiesException( - invalidValueMessage(fullKey, "Expecting integer value", property)); + invalidValueMessage(fullKey, "Expecting integer value")); } } return null; @@ -147,7 +146,7 @@ static Double loadDouble(String prefix, String propertyName, PropertySource prop return Double.parseDouble(property); } catch (NumberFormatException e) { throw new PrometheusPropertiesException( - invalidValueMessage(fullKey, "Expecting double value", property)); + invalidValueMessage(fullKey, "Expecting double value")); } } return null; @@ -163,7 +162,7 @@ static Long loadLong(String prefix, String propertyName, PropertySource property return Long.parseLong(property); } catch (NumberFormatException e) { throw new PrometheusPropertiesException( - invalidValueMessage(fullKey, "Expecting long value", property)); + invalidValueMessage(fullKey, "Expecting long value")); } } return null; @@ -199,51 +198,7 @@ static void assertValue( } } - static String invalidValueMessage(String fullKey, String message, String found) { - String separator = message.endsWith(".") ? " " : ". "; - return fullKey + ": " + message + separator + "Found: " + escape(found); - } - - static String escape(String value) { - StringBuilder result = new StringBuilder(value.length() + 2); - result.append('"'); - int maxLength = Math.min(value.length(), 100); - for (int i = 0; i < maxLength; i++) { - char c = value.charAt(i); - switch (c) { - case '\b': - result.append("\\b"); - break; - case '\t': - result.append("\\t"); - break; - case '\n': - result.append("\\n"); - break; - case '\f': - result.append("\\f"); - break; - case '\r': - result.append("\\r"); - break; - case '"': - result.append("\\\""); - break; - case '\\': - result.append("\\\\"); - break; - default: - if (c < 0x20 || c == 0x7f) { - result.append(String.format("\\u%04x", (int) c)); - } else { - result.append(c); - } - } - } - if (value.length() > maxLength) { - result.append("..."); - } - result.append('"'); - return result.toString(); + static String invalidValueMessage(String fullKey, String message) { + return fullKey + ": " + message; } } diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPropertiesTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPropertiesTest.java index 0a5d20ef3..2f912ff77 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPropertiesTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPropertiesTest.java @@ -27,8 +27,7 @@ void load() { new HashMap<>( Map.of("io.prometheus.exporter.include_created_timestamps", "invalid")))) .withMessage( - "io.prometheus.exporter.include_created_timestamps: Expecting 'true' or 'false'. Found:" - + " \"invalid\""); + "io.prometheus.exporter.include_created_timestamps: Expecting 'true' or 'false'."); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> @@ -36,8 +35,7 @@ void load() { new HashMap<>( Map.of("io.prometheus.exporter.exemplars_on_all_metric_types", "invalid")))) .withMessage( - "io.prometheus.exporter.exemplars_on_all_metric_types: Expecting 'true' or 'false'." - + " Found: \"invalid\""); + "io.prometheus.exporter.exemplars_on_all_metric_types: Expecting 'true' or 'false'."); } private static ExporterProperties load(Map map) { @@ -84,7 +82,6 @@ void prometheusTimestampsInMs() { new HashMap<>( Map.of("io.prometheus.exporter.prometheus_timestamps_in_ms", "invalid")))) .withMessage( - "io.prometheus.exporter.prometheus_timestamps_in_ms: Expecting 'true' or 'false'." - + " Found: \"invalid\""); + "io.prometheus.exporter.prometheus_timestamps_in_ms: Expecting 'true' or 'false'."); } } diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPushgatewayPropertiesTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPushgatewayPropertiesTest.java index dbec7c8a1..4715662f3 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPushgatewayPropertiesTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/ExporterPushgatewayPropertiesTest.java @@ -27,7 +27,7 @@ void load() { .isThrownBy(() -> load(Map.of("io.prometheus.exporter.pushgateway.scheme", "foo"))) .withMessage( "io.prometheus.exporter.pushgateway.scheme: Illegal value. Expecting 'http' or 'https'." - + " Found: \"foo\""); + + ""); } @Test @@ -60,7 +60,7 @@ void loadWithInvalidEscapingScheme() { () -> load(Map.of("io.prometheus.exporter.pushgateway.escaping_scheme", "invalid"))) .withMessage( "io.prometheus.exporter.pushgateway.escaping_scheme: Illegal value. Expecting" - + " 'allow-utf-8', 'values', 'underscores', or 'dots'. Found: \"invalid\""); + + " 'allow-utf-8', 'values', 'underscores', or 'dots'."); } @Test diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/OpenMetrics2PropertiesTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/OpenMetrics2PropertiesTest.java index df9e9a04b..0546a138f 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/OpenMetrics2PropertiesTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/OpenMetrics2PropertiesTest.java @@ -37,8 +37,7 @@ void loadInvalidValue() { assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> load(new HashMap<>(Map.of("io.prometheus.openmetrics2.enabled", "invalid")))) - .withMessage( - "io.prometheus.openmetrics2.enabled: Expecting 'true' or 'false'. Found: \"invalid\""); + .withMessage("io.prometheus.openmetrics2.enabled: Expecting 'true' or 'false'."); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> @@ -46,17 +45,14 @@ void loadInvalidValue() { new HashMap<>( Map.of("io.prometheus.openmetrics2.content_negotiation", "invalid")))) .withMessage( - "io.prometheus.openmetrics2.content_negotiation: Expecting 'true' or 'false'. Found:" - + " \"invalid\""); + "io.prometheus.openmetrics2.content_negotiation: Expecting 'true' or 'false'."); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> load( new HashMap<>( Map.of("io.prometheus.openmetrics2.composite_values", "invalid")))) - .withMessage( - "io.prometheus.openmetrics2.composite_values: Expecting 'true' or 'false'. Found:" - + " \"invalid\""); + .withMessage("io.prometheus.openmetrics2.composite_values: Expecting 'true' or 'false'."); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> @@ -64,17 +60,14 @@ void loadInvalidValue() { new HashMap<>( Map.of("io.prometheus.openmetrics2.exemplar_compliance", "invalid")))) .withMessage( - "io.prometheus.openmetrics2.exemplar_compliance: Expecting 'true' or 'false'. Found:" - + " \"invalid\""); + "io.prometheus.openmetrics2.exemplar_compliance: Expecting 'true' or 'false'."); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy( () -> load( new HashMap<>( Map.of("io.prometheus.openmetrics2.native_histograms", "invalid")))) - .withMessage( - "io.prometheus.openmetrics2.native_histograms: Expecting 'true' or 'false'. Found:" - + " \"invalid\""); + .withMessage("io.prometheus.openmetrics2.native_histograms: Expecting 'true' or 'false'."); } private static OpenMetrics2Properties load(Map map) { diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/PrometheusPropertiesLoaderTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/PrometheusPropertiesLoaderTest.java index 532b00295..8f910cceb 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/PrometheusPropertiesLoaderTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/PrometheusPropertiesLoaderTest.java @@ -31,9 +31,8 @@ void propertiesShouldBeLoadedFromPropertiesFile() { void cantLoadPropertiesFile() { assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy(() -> PrometheusPropertiesLoader.load(new Properties())) - .withMessage( - "Failed to read Prometheus properties from nonexistent.properties:" - + " nonexistent.properties"); + .withMessage("Failed to read Prometheus properties file.") + .withMessageNotContaining("nonexistent.properties"); } @Test diff --git a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/UtilTest.java b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/UtilTest.java index 3fadf15c2..c3ed65ec7 100644 --- a/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/UtilTest.java +++ b/prometheus-metrics-config/src/test/java/io/prometheus/metrics/config/UtilTest.java @@ -51,16 +51,18 @@ void loadOptionalDuration_invalidNumber_throws() { assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy(() -> Util.loadOptionalDuration("", "foo", propertySource)) - .withMessage("foo: Expecting long value. Found: \"abc\""); + .withMessage("foo: Expecting long value"); } @Test - void invalidValueMessageEscapesRawValue() { - Map regularProperties = new HashMap<>(Map.of("foo", "bad\n\"value")); + void invalidValueMessageRedactsRawValue() { + String secret = "bad\n\"secret-value"; + Map regularProperties = new HashMap<>(Map.of("foo", secret)); PropertySource propertySource = new PropertySource(regularProperties); assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy(() -> Util.loadBoolean("", "foo", propertySource)) - .withMessage("foo: Expecting 'true' or 'false'. Found: \"bad\\n\\\"value\""); + .withMessage("foo: Expecting 'true' or 'false'.") + .withMessageNotContaining(secret); } } 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 b3a52652c..faeb460cc 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 @@ -9,79 +9,96 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import javax.annotation.Nullable; -/** - * Metrics support concurrent write and scrape operations. - * - *

This is implemented by switching to a Buffer when the scrape starts, and applying the values - * from the buffer after the scrape ends. - */ +/** Metrics support concurrent write and scrape operations. */ class Buffer { - private static final long bufferActiveBit = 1L << 63; private static final long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(1); - // 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 improve, we stripe the AtomicLong into N instances, - // where N is the number of available processors. Each record operation chooses the appropriate - // instance to use based on the modulo of its thread id and N. This is a more naive / simple - // implementation compared to the striping used under the hood in java.util.concurrent classes - // like LongAdder - contention and hot spots can still occur if recording thread ids happen to - // resolve to the same index. Further improvement is possible. - private final AtomicLong[] stripedObservationCounts; - private double[] observationBuffer = new double[0]; - private int bufferPos = 0; - private boolean reset = false; + private static final int DEFAULT_MAX_BUFFER_SIZE = 1_000_000; + private static final int INITIAL_BUFFER_SIZE = 128; + + private static final class Generation { + private double[] values = new double[0]; + private int size; + private boolean active = true; + } + private final AtomicLong[] stripedObservationCounts; + private final AtomicLong appendersInFlight = new AtomicLong(); + private final AtomicLong appendPhase = new AtomicLong(); + private boolean reset; + private long observationCountOffset; + @Nullable private volatile Generation activeGeneration; ReentrantLock appendLock = new ReentrantLock(); ReentrantLock runLock = new ReentrantLock(); - Condition bufferFilled = appendLock.newCondition(); + private final Condition bufferSpaceAvailable = appendLock.newCondition(); private final long maxSpinWaitNanos; + private final int maxBufferSize; + private final Runnable beforeAppendLock; Buffer() { - this(DEFAULT_MAX_SPIN_WAIT_NANOS); + this(DEFAULT_MAX_SPIN_WAIT_NANOS, DEFAULT_MAX_BUFFER_SIZE, () -> {}); } Buffer(long maxSpinWaitNanos) { + this(maxSpinWaitNanos, DEFAULT_MAX_BUFFER_SIZE, () -> {}); + } + + Buffer(long maxSpinWaitNanos, int maxBufferSize, Runnable beforeAppendLock) { + if (maxBufferSize <= 0) throw new IllegalArgumentException("maxBufferSize must be positive"); this.maxSpinWaitNanos = maxSpinWaitNanos; + this.maxBufferSize = maxBufferSize; + this.beforeAppendLock = beforeAppendLock; stripedObservationCounts = new AtomicLong[Runtime.getRuntime().availableProcessors()]; for (int i = 0; i < stripedObservationCounts.length; i++) { - stripedObservationCounts[i] = new AtomicLong(0); + stripedObservationCounts[i] = new AtomicLong(); } } boolean append(double value) { - int index = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length); - AtomicLong observationCountForThread = stripedObservationCounts[index]; - long count = observationCountForThread.incrementAndGet(); - if ((count & bufferActiveBit) == 0) { - return false; // sign bit not set -> buffer not active. - } else { - doAppend(value); - return true; - } - } - - static int stripeIndex(long threadId, int stripeCount) { - return (int) Math.floorMod(threadId, stripeCount); - } - - private void doAppend(double amount) { + AtomicLong counter = + stripedObservationCounts[ + stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length)]; + appendersInFlight.incrementAndGet(); + long phase = appendPhase.get(); + long count = counter.incrementAndGet(); + boolean phaseChanged = appendPhase.get() != phase; + appendersInFlight.decrementAndGet(); + if (phaseChanged || (phase & 1L) != 0 || (count & bufferActiveBit) == 0) return false; + Generation generation = activeGeneration; + if (generation == null) return false; + beforeAppendLock.run(); appendLock.lock(); try { - if (bufferPos >= observationBuffer.length) { - observationBuffer = Arrays.copyOf(observationBuffer, observationBuffer.length + 128); + Generation current = activeGeneration; + if (current != null && current != generation) generation = current; + if (!generation.active) return false; + while (generation.size >= maxBufferSize && generation.active) { + bufferSpaceAvailable.awaitUninterruptibly(); } - observationBuffer[bufferPos] = amount; - bufferPos++; - - bufferFilled.signalAll(); + if (!generation.active) return false; + if (generation.size >= generation.values.length) { + int doubled = + generation.values.length > maxBufferSize / 2 + ? maxBufferSize + : generation.values.length * 2; + generation.values = + Arrays.copyOf( + generation.values, + Math.min(maxBufferSize, Math.max(INITIAL_BUFFER_SIZE, Math.max(1, doubled)))); + } + generation.values[generation.size++] = value; + return true; } finally { appendLock.unlock(); } } - /** Must be called by the runnable in the run() method. */ + static int stripeIndex(long threadId, int stripeCount) { + return (int) Math.floorMod(threadId, stripeCount); + } + void reset() { reset = true; } @@ -91,25 +108,47 @@ T run( Function complete, Supplier createResult, Consumer observeFunction) { + return run(complete, createResult, observeFunction, true, complete); + } + + @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) + T run( + Function complete, + Supplier createResult, + Consumer observeFunction, + boolean failOnTimeout) { + return run(complete, createResult, observeFunction, failOnTimeout, complete); + } + + @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) + T run( + Function complete, + Supplier createResult, + Consumer observeFunction, + boolean failOnTimeout, + Function afterTimeout) { + Generation generation = new Generation(); double[] buffer; int bufferSize; - T result; boolean timedOut = false; - + T result; runLock.lock(); try { - // Signal that the buffer is active. - long expectedCount = 0L; - for (AtomicLong observationCount : stripedObservationCounts) { - expectedCount += observationCount.getAndAdd(bufferActiveBit); + phaseTransition(); + long expectedCount; + appendLock.lock(); + try { + activeGeneration = generation; + long total = 0; + for (AtomicLong counter : stripedObservationCounts) + total += counter.getAndAdd(bufferActiveBit); + expectedCount = total - observationCountOffset; + } finally { + appendLock.unlock(); } - + appendPhase.incrementAndGet(); long deadline = System.nanoTime() + maxSpinWaitNanos; 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. if (System.nanoTime() - deadline >= 0) { timedOut = true; break; @@ -117,47 +156,39 @@ T run( Thread.yield(); } result = timedOut ? null : 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; - + phaseTransition(); appendLock.lock(); try { - while (bufferPos < expectedBufferSize) { - // Wait until all in-flight threads have added their observations to the buffer. - bufferFilled.await(); + generation.active = false; + for (AtomicLong counter : stripedObservationCounts) counter.addAndGet(bufferActiveBit); + if (reset) { + observationCountOffset += expectedCount; + reset = false; } + activeGeneration = null; + buffer = generation.values; + bufferSize = generation.size; + generation.values = new double[0]; + generation.size = 0; + bufferSpaceAvailable.signalAll(); } finally { appendLock.unlock(); } - - buffer = observationBuffer; - bufferSize = bufferPos; - observationBuffer = new double[0]; - bufferPos = 0; - } catch (InterruptedException e) { - throw new RuntimeException(e); + appendPhase.incrementAndGet(); + for (int i = 0; i < bufferSize; i++) observeFunction.accept(buffer[i]); + if (timedOut && failOnTimeout) { + if (afterTimeout.apply(expectedCount)) return createResult.get(); + throw new IllegalStateException("Timed out while waiting for in-flight observations."); + } + return result; } finally { runLock.unlock(); } + } - for (int i = 0; i < bufferSize; i++) { - observeFunction.accept(buffer[i]); - } - if (timedOut) { - throw new IllegalStateException("Timed out while waiting for in-flight observations."); - } - return result; + @SuppressWarnings("ThreadPriorityCheck") + private void phaseTransition() { + appendPhase.incrementAndGet(); + while (appendersInFlight.get() != 0) Thread.yield(); } } diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java index c4bb1f5fe..9efcac436 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 @@ -339,7 +339,9 @@ private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { createdTimeMillis); } }, - v -> doObserve(v, true)); + v -> doObserve(v, true), + true, + expectedCount -> count.sum() >= expectedCount); } private boolean addToNativeBucket(double value, ConcurrentHashMap buckets) { @@ -444,7 +446,8 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { } return null; }, - v -> doObserve(v, true)); + v -> doObserve(v, true), + false); } else if (nativeBucketCreated) { // If a new bucket was created we need to check if nativeMaxBuckets is exceeded // and scale down if so. @@ -488,7 +491,8 @@ private void maybeScaleDown(AtomicBoolean wasReset) { doubleBucketWidth(); return null; }, - v -> doObserve(v, true)); + v -> doObserve(v, true), + false); } // maybeReset is called in the synchronized block while new observations go into the buffer. 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 7f8e09dbf..c2f315bb2 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 @@ -10,6 +10,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; class BufferTest { @@ -70,4 +71,46 @@ void timeoutDoesNotCreateSnapshot() { ignored -> {})) .withMessage("Timed out while waiting for in-flight observations."); } + + @Test + void stalledAppenderAfterStripeActivationDoesNotBlockRun() throws InterruptedException { + CountDownLatch started = new CountDownLatch(1); + CountDownLatch proceed = new CountDownLatch(1); + CountDownLatch stalled = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicReference appended = new AtomicReference<>(); + Buffer buffer = + new Buffer( + TimeUnit.SECONDS.toNanos(1), + 16, + () -> { + stalled.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + Thread runner = + new Thread( + () -> + buffer.run( + ignored -> { + started.countDown(); + return proceed.getCount() == 0; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {})); + runner.start(); + assertThat(started.await(5, TimeUnit.SECONDS)).isTrue(); + Thread appender = new Thread(() -> appended.set(buffer.append(1.0))); + appender.start(); + assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue(); + proceed.countDown(); + runner.join(5_000); + assertThat(runner.isAlive()).isFalse(); + release.countDown(); + appender.join(5_000); + assertThat(appended).hasValue(false); + } } diff --git a/prometheus-metrics-exporter-pushgateway/src/main/java/io/prometheus/metrics/exporter/pushgateway/PushGateway.java b/prometheus-metrics-exporter-pushgateway/src/main/java/io/prometheus/metrics/exporter/pushgateway/PushGateway.java index 5bf26b6c1..b7dd844ec 100644 --- a/prometheus-metrics-exporter-pushgateway/src/main/java/io/prometheus/metrics/exporter/pushgateway/PushGateway.java +++ b/prometheus-metrics-exporter-pushgateway/src/main/java/io/prometheus/metrics/exporter/pushgateway/PushGateway.java @@ -554,9 +554,9 @@ public PushGateway build() { getEscapingScheme(properties), getConnectionTimeout(properties), getReadTimeout(properties)); - } catch (MalformedURLException e) { + } catch (MalformedURLException | IllegalArgumentException e) { throw new PrometheusPropertiesException( - address + ": Invalid address. Expecting :"); + "Invalid Pushgateway address. Expecting :"); } catch (UnsupportedEncodingException e) { throw new RuntimeException(e); // cannot happen, UTF-8 is always supported } diff --git a/prometheus-metrics-exporter-pushgateway/src/test/java/io/prometheus/metrics/exporter/pushgateway/PushGatewayTest.java b/prometheus-metrics-exporter-pushgateway/src/test/java/io/prometheus/metrics/exporter/pushgateway/PushGatewayTest.java index bae8fdd91..8c33543f9 100644 --- a/prometheus-metrics-exporter-pushgateway/src/test/java/io/prometheus/metrics/exporter/pushgateway/PushGatewayTest.java +++ b/prometheus-metrics-exporter-pushgateway/src/test/java/io/prometheus/metrics/exporter/pushgateway/PushGatewayTest.java @@ -6,6 +6,7 @@ import static org.mockserver.model.HttpResponse.response; import io.prometheus.metrics.config.EscapingScheme; +import io.prometheus.metrics.config.PrometheusPropertiesException; import io.prometheus.metrics.core.metrics.Gauge; import io.prometheus.metrics.model.registry.PrometheusRegistry; import java.io.IOException; @@ -48,6 +49,15 @@ void testInvalidURLThrowsRuntimeException() { }); } + @Test + void testInvalidURLDoesNotExposeCredentials() { + String secretAddress = "user:secret@[::"; + assertThatExceptionOfType(PrometheusPropertiesException.class) + .isThrownBy(() -> PushGateway.builder().address(secretAddress).build()) + .withMessage("Invalid Pushgateway address. Expecting :") + .withMessageNotContaining("secret"); + } + @Test void testMultipleSlashesAreStrippedFromURL() throws NoSuchFieldException, IllegalAccessException { final PushGateway pushGateway = From ec7f83da8b3e368b02d558b0b52d62310883538a Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 08:40:18 +0000 Subject: [PATCH 5/7] fix: satisfy scrape buffer style checks Signed-off-by: Gregor Zeitlinger --- .../metrics/core/metrics/Buffer.java | 43 ++++++++++++++----- 1 file changed, 32 insertions(+), 11 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 faeb460cc..caf885b56 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 @@ -46,7 +46,9 @@ private static final class Generation { } Buffer(long maxSpinWaitNanos, int maxBufferSize, Runnable beforeAppendLock) { - if (maxBufferSize <= 0) throw new IllegalArgumentException("maxBufferSize must be positive"); + if (maxBufferSize <= 0) { + throw new IllegalArgumentException("maxBufferSize must be positive"); + } this.maxSpinWaitNanos = maxSpinWaitNanos; this.maxBufferSize = maxBufferSize; this.beforeAppendLock = beforeAppendLock; @@ -65,19 +67,29 @@ boolean append(double value) { long count = counter.incrementAndGet(); boolean phaseChanged = appendPhase.get() != phase; appendersInFlight.decrementAndGet(); - if (phaseChanged || (phase & 1L) != 0 || (count & bufferActiveBit) == 0) return false; + if (phaseChanged || (phase & 1L) != 0 || (count & bufferActiveBit) == 0) { + return false; + } Generation generation = activeGeneration; - if (generation == null) return false; + if (generation == null) { + return false; + } beforeAppendLock.run(); appendLock.lock(); try { Generation current = activeGeneration; - if (current != null && current != generation) generation = current; - if (!generation.active) return false; + if (current != null && current != generation) { + generation = current; + } + if (!generation.active) { + return false; + } while (generation.size >= maxBufferSize && generation.active) { bufferSpaceAvailable.awaitUninterruptibly(); } - if (!generation.active) return false; + if (!generation.active) { + return false; + } if (generation.size >= generation.values.length) { int doubled = generation.values.length > maxBufferSize / 2 @@ -140,8 +152,9 @@ T run( try { activeGeneration = generation; long total = 0; - for (AtomicLong counter : stripedObservationCounts) + for (AtomicLong counter : stripedObservationCounts) { total += counter.getAndAdd(bufferActiveBit); + } expectedCount = total - observationCountOffset; } finally { appendLock.unlock(); @@ -160,7 +173,9 @@ T run( appendLock.lock(); try { generation.active = false; - for (AtomicLong counter : stripedObservationCounts) counter.addAndGet(bufferActiveBit); + for (AtomicLong counter : stripedObservationCounts) { + counter.addAndGet(bufferActiveBit); + } if (reset) { observationCountOffset += expectedCount; reset = false; @@ -175,9 +190,13 @@ T run( appendLock.unlock(); } appendPhase.incrementAndGet(); - for (int i = 0; i < bufferSize; i++) observeFunction.accept(buffer[i]); + for (int i = 0; i < bufferSize; i++) { + observeFunction.accept(buffer[i]); + } if (timedOut && failOnTimeout) { - if (afterTimeout.apply(expectedCount)) return createResult.get(); + if (afterTimeout.apply(expectedCount)) { + return createResult.get(); + } throw new IllegalStateException("Timed out while waiting for in-flight observations."); } return result; @@ -189,6 +208,8 @@ T run( @SuppressWarnings("ThreadPriorityCheck") private void phaseTransition() { appendPhase.incrementAndGet(); - while (appendersInFlight.get() != 0) Thread.yield(); + while (appendersInFlight.get() != 0) { + Thread.yield(); + } } } From 1a9a50dc4d6c56cdcf47245cb1b9a1631f7842c1 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 09:06:53 +0000 Subject: [PATCH 6/7] fix: preserve observations across buffer generations Signed-off-by: Gregor Zeitlinger --- .../metrics/core/metrics/Buffer.java | 84 ++++++++++--------- .../metrics/core/metrics/Histogram.java | 29 ++++--- .../metrics/core/metrics/Summary.java | 14 +++- .../metrics/core/metrics/BufferTest.java | 76 +++++++++++++++++ 4 files changed, 149 insertions(+), 54 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 caf885b56..db4fc9892 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 @@ -27,6 +27,7 @@ private static final class Generation { private final AtomicLong[] stripedObservationCounts; private final AtomicLong appendersInFlight = new AtomicLong(); private final AtomicLong appendPhase = new AtomicLong(); + private final ReentrantLock observationLock = new ReentrantLock(); private boolean reset; private long observationCountOffset; @Nullable private volatile Generation activeGeneration; @@ -78,10 +79,7 @@ boolean append(double value) { appendLock.lock(); try { Generation current = activeGeneration; - if (current != null && current != generation) { - generation = current; - } - if (!generation.active) { + if (current != generation || !generation.active) { return false; } while (generation.size >= maxBufferSize && generation.active) { @@ -115,21 +113,21 @@ void reset() { reset = true; } - @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) - T run( - Function complete, - Supplier createResult, - Consumer observeFunction) { - return run(complete, createResult, observeFunction, true, complete); + T observeDirect(Supplier observeFunction) { + observationLock.lock(); + try { + return observeFunction.get(); + } finally { + observationLock.unlock(); + } } @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) T run( Function complete, Supplier createResult, - Consumer observeFunction, - boolean failOnTimeout) { - return run(complete, createResult, observeFunction, failOnTimeout, complete); + Consumer observeFunction) { + return run(complete, createResult, observeFunction, true); } @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) @@ -137,13 +135,12 @@ T run( Function complete, Supplier createResult, Consumer observeFunction, - boolean failOnTimeout, - Function afterTimeout) { + boolean failOnTimeout) { Generation generation = new Generation(); double[] buffer; int bufferSize; boolean timedOut = false; - T result; + T result = null; runLock.lock(); try { phaseTransition(); @@ -168,35 +165,40 @@ T run( } Thread.yield(); } - result = timedOut ? null : createResult.get(); - phaseTransition(); - appendLock.lock(); + observationLock.lock(); try { - generation.active = false; - for (AtomicLong counter : stripedObservationCounts) { - counter.addAndGet(bufferActiveBit); - } - if (reset) { - observationCountOffset += expectedCount; - reset = false; - } - activeGeneration = null; - buffer = generation.values; - bufferSize = generation.size; - generation.values = new double[0]; - generation.size = 0; - bufferSpaceAvailable.signalAll(); + result = timedOut ? null : createResult.get(); } finally { - appendLock.unlock(); - } - appendPhase.incrementAndGet(); - for (int i = 0; i < bufferSize; i++) { - observeFunction.accept(buffer[i]); + try { + phaseTransition(); + appendLock.lock(); + try { + generation.active = false; + for (AtomicLong counter : stripedObservationCounts) { + counter.addAndGet(bufferActiveBit); + } + if (reset) { + observationCountOffset += expectedCount; + reset = false; + } + activeGeneration = null; + buffer = generation.values; + bufferSize = generation.size; + generation.values = new double[0]; + generation.size = 0; + bufferSpaceAvailable.signalAll(); + } finally { + appendLock.unlock(); + } + appendPhase.incrementAndGet(); + for (int i = 0; i < bufferSize; i++) { + observeFunction.accept(buffer[i]); + } + } finally { + observationLock.unlock(); + } } if (timedOut && failOnTimeout) { - if (afterTimeout.apply(expectedCount)) { - return createResult.get(); - } throw new IllegalStateException("Timed out while waiting for in-flight observations."); } return result; 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 9efcac436..faddd0f27 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 @@ -243,7 +243,8 @@ public void observe(double value) { return; } if (!buffer.append(value)) { - doObserve(value, false); + boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value)); + maybeResetOrScaleDown(value, nativeBucketCreated); } if (exemplarSampler != null) { exemplarSampler.observe(value); @@ -257,14 +258,15 @@ public void observeWithExemplar(double value, Labels labels) { return; } if (!buffer.append(value)) { - doObserve(value, false); + boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value)); + maybeResetOrScaleDown(value, nativeBucketCreated); } if (exemplarSampler != null) { exemplarSampler.observeWithExemplar(value, labels); } } - private void doObserve(double value, boolean fromBuffer) { + private boolean doObserve(double value) { // 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. @@ -287,6 +289,11 @@ private void doObserve(double value, boolean fromBuffer) { count .increment(); // must be the last step, because count is used to signal that the operation // is complete. + return nativeBucketCreated; + } + + private void doObserve(double value, boolean fromBuffer) { + boolean nativeBucketCreated = doObserve(value); if (!fromBuffer) { // maybeResetOrScaleDown will switch to the buffer, // which won't work if we are currently still processing observations from the buffer. @@ -302,7 +309,7 @@ private void doObserve(double value, boolean fromBuffer) { private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { Exemplars exemplars = exemplarSampler != null ? exemplarSampler.collect() : Exemplars.EMPTY; return buffer.run( - expectedCount -> count.sum() == expectedCount, + expectedCount -> count.sum() >= expectedCount, () -> { if (classicUpperBounds.length == 0) { // native only @@ -339,9 +346,7 @@ private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { createdTimeMillis); } }, - v -> doObserve(v, true), - true, - expectedCount -> count.sum() >= expectedCount); + v -> doObserve(v, true)); } private boolean addToNativeBucket(double value, ConcurrentHashMap buckets) { @@ -439,7 +444,7 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { // If nativeSchema < initialNativeSchema the histogram has been scaled down. // So if resetDurationExpired we will reset it to restore the original native schema. buffer.run( - expectedCount -> count.sum() == expectedCount, + expectedCount -> count.sum() >= expectedCount, () -> { if (maybeReset()) { wasReset.set(true); @@ -456,7 +461,11 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { if (wasReset.get()) { // We just discarded the newly observed value. Observe it again. if (!buffer.append(value)) { - doObserve(value, true); + buffer.observeDirect( + () -> { + doObserve(value, true); + return null; + }); } } } @@ -471,7 +480,7 @@ private void maybeScaleDown(AtomicBoolean wasReset) { return; } buffer.run( - expectedCount -> count.sum() == expectedCount, + expectedCount -> count.sum() >= expectedCount, () -> { // Now we are in the synchronized block while new observations go into the buffer. // Check again if we need to limit the bucket size, because another thread might 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 043f31129..832c10619 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 @@ -185,7 +185,11 @@ public void observe(double value) { return; } if (!buffer.append(value)) { - doObserve(value); + buffer.observeDirect( + () -> { + doObserve(value); + return null; + }); } if (exemplarSampler != null) { exemplarSampler.observe(value); @@ -198,7 +202,11 @@ public void observeWithExemplar(double value, Labels labels) { return; } if (!buffer.append(value)) { - doObserve(value); + buffer.observeDirect( + () -> { + doObserve(value); + return null; + }); } if (exemplarSampler != null) { exemplarSampler.observeWithExemplar(value, labels); @@ -217,7 +225,7 @@ private void doObserve(double amount) { private SummarySnapshot.SummaryDataPointSnapshot collect(Labels labels) { return buffer.run( - expectedCount -> count.sum() == expectedCount, + expectedCount -> count.sum() >= expectedCount, // Note: Exemplars are currently hard-coded as empty for Summary metrics. // While exemplars are sampled during observe() and observeWithExemplar() calls // via the exemplarSampler field, they are not included in the snapshot to maintain 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 c2f315bb2..abc7ae991 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 @@ -10,6 +10,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; @@ -113,4 +114,79 @@ void stalledAppenderAfterStripeActivationDoesNotBlockRun() throws InterruptedExc appender.join(5_000); assertThat(appended).hasValue(false); } + + @Test + void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException { + CountDownLatch firstRunStarted = new CountDownLatch(1); + CountDownLatch firstRunMayFinish = new CountDownLatch(1); + CountDownLatch stalled = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch secondRunStarted = new CountDownLatch(1); + AtomicBoolean appended = new AtomicBoolean(); + AtomicLong completedObservations = new AtomicLong(); + Buffer buffer = + new Buffer( + TimeUnit.SECONDS.toNanos(1), + 16, + () -> { + stalled.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + Thread firstRun = + new Thread( + () -> + buffer.run( + ignored -> { + firstRunStarted.countDown(); + return firstRunMayFinish.getCount() == 0; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {})); + firstRun.start(); + assertThat(firstRunStarted.await(5, TimeUnit.SECONDS)).isTrue(); + + Thread appender = + new Thread( + () -> { + appended.set(buffer.append(1.0)); + if (!appended.get()) { + buffer.observeDirect( + () -> { + completedObservations.incrementAndGet(); + return null; + }); + } + }); + appender.start(); + assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue(); + + firstRunMayFinish.countDown(); + firstRun.join(5_000); + assertThat(firstRun.isAlive()).isFalse(); + + Thread secondRun = + new Thread( + () -> + buffer.run( + expectedCount -> { + secondRunStarted.countDown(); + return completedObservations.get() >= expectedCount; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {})); + secondRun.start(); + assertThat(secondRunStarted.await(5, TimeUnit.SECONDS)).isTrue(); + release.countDown(); + appender.join(5_000); + secondRun.join(5_000); + + assertThat(appender.isAlive()).isFalse(); + assertThat(secondRun.isAlive()).isFalse(); + assertThat(appended).isFalse(); + assertThat(completedObservations).hasValue(1); + } } From b45478d3fd184956b1f938731c158dad90698376 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 09:26:58 +0000 Subject: [PATCH 7/7] docs: refresh histogram API diff Signed-off-by: Gregor Zeitlinger --- docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt b/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt index ffb4a1d52..136f7f6f1 100644 --- a/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt +++ b/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt @@ -1,4 +1,6 @@ Comparing source compatibility of prometheus-metrics-core-1.8.1-SNAPSHOT.jar against prometheus-metrics-core-1.8.0.jar *** MODIFIED CLASS: PUBLIC io.prometheus.metrics.core.exemplars.ExemplarSampler (not serializable) === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 +*** MODIFIED CLASS: PUBLIC io.prometheus.metrics.core.metrics.Histogram$DataPoint (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0