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 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 + 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..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 @@ -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'.")); } } @@ -119,10 +118,9 @@ 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'.")); } } 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 20bd75699..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( - String.format("%s: Expecting 'true' or 'false'. Found: %s", fullKey, property)); + invalidValueMessage(fullKey, "Expecting 'true' or 'false'.")); } return Boolean.parseBoolean(property); } @@ -88,7 +88,7 @@ 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")); } } return Arrays.asList(result); @@ -130,7 +130,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")); } } return null; @@ -146,7 +146,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")); } } return null; @@ -162,7 +162,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")); } } return null; @@ -197,4 +197,8 @@ static void assertValue( throw new PrometheusPropertiesException(fullMessage); } } + + 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 514c5ff52..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 c92e6f2f9..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 e7a273464..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 e4d7fa829..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,6 +51,18 @@ void loadOptionalDuration_invalidNumber_throws() { assertThatExceptionOfType(PrometheusPropertiesException.class) .isThrownBy(() -> Util.loadOptionalDuration("", "foo", propertySource)) - .withMessage("foo=abc: Expecting long value"); + .withMessage("foo: Expecting long value"); + } + + @Test + 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'.") + .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 1c47f867c..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 @@ -2,141 +2,216 @@ 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; 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; - // 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 long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(1); + 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 final ReentrantLock observationLock = new ReentrantLock(); + 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, 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 = Math.abs((int) 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; + 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; } - } - - private void doAppend(double amount) { + 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 != generation || !generation.active) { + return false; } - observationBuffer[bufferPos] = amount; - bufferPos++; - - bufferFilled.signalAll(); + while (generation.size >= maxBufferSize && generation.active) { + bufferSpaceAvailable.awaitUninterruptibly(); + } + 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; } - @SuppressWarnings("ThreadPriorityCheck") + T observeDirect(Supplier observeFunction) { + observationLock.lock(); + try { + return observeFunction.get(); + } finally { + observationLock.unlock(); + } + } + + @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) T run( Function complete, Supplier createResult, Consumer observeFunction) { + return run(complete, createResult, observeFunction, true); + } + + @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) + T run( + Function complete, + Supplier createResult, + Consumer observeFunction, + boolean failOnTimeout) { + Generation generation = new Generation(); double[] buffer; int bufferSize; - T result; - + boolean timedOut = false; + T result = null; 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. - Thread.yield(); - } - result = createResult.get(); - - // Signal that the buffer is inactive. - long expectedBufferSize = 0; - if (reset) { - for (AtomicLong observationCount : stripedObservationCounts) { - expectedBufferSize += observationCount.getAndSet(0) & ~bufferActiveBit; - } - reset = false; - } else { - for (AtomicLong observationCount : stripedObservationCounts) { - expectedBufferSize += observationCount.addAndGet(bufferActiveBit); + if (System.nanoTime() - deadline >= 0) { + timedOut = true; + break; } + Thread.yield(); } - expectedBufferSize -= expectedCount; - - appendLock.lock(); + observationLock.lock(); try { - while (bufferPos < expectedBufferSize) { - // Wait until all in-flight threads have added their observations to the buffer. - bufferFilled.await(); - } + result = timedOut ? null : createResult.get(); } finally { - appendLock.unlock(); + 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(); + } } - - buffer = observationBuffer; - bufferSize = bufferPos; - observationBuffer = new double[0]; - bufferPos = 0; - } catch (InterruptedException e) { - throw new RuntimeException(e); + if (timedOut && failOnTimeout) { + 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]); + @SuppressWarnings("ThreadPriorityCheck") + private void phaseTransition() { + appendPhase.incrementAndGet(); + while (appendersInFlight.get() != 0) { + Thread.yield(); } - 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 c4bb1f5fe..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 @@ -437,14 +444,15 @@ 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); } 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. @@ -453,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; + }); } } } @@ -468,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 @@ -488,7 +500,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/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 new file mode 100644 index 000000000..abc7ae991 --- /dev/null +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -0,0 +1,192 @@ +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 java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +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."); + } + + @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); + } + + @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); + } +} 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..0622952e7 --- /dev/null +++ b/prometheus-metrics-exporter-common/src/main/java/io/prometheus/metrics/exporter/common/InvalidQueryParameterException.java @@ -0,0 +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 a0c692c23..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; @@ -49,15 +48,43 @@ 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 (idx != -1 + && InvalidQueryParameterException.urlDecode(pair.substring(0, idx)).equals(name)) { + result.add(InvalidQueryParameterException.urlDecode(pair.substring(idx + 1))); + } + 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/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-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..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 @@ -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 = 10; + 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,14 @@ public void handle(HttpExchange exchange) throws IOException { } } } else { - drainInputAndClose(exchange); + exchange.getRequestBody().close(); exchange.sendResponseHeaders(403, -1); + exchange.close(); } } }; } - 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 +332,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..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 @@ -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; @@ -45,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); @@ -158,6 +159,22 @@ 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.getCorePoolSize()).isEqualTo(10); + 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 +188,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 +260,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 +281,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-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 = 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".