From fb35a2e06c903d7f0b7e6e7e7c2fcdeb3996e5f7 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 14 Sep 2026 14:03:08 +0200 Subject: [PATCH 1/4] [FLINK-40641][metrics] Assert the exposition contract of the Prometheus reporter No existing test asserts what a family does not contain, so the missing `_sum` and the Meter's missing count were uncovered. `TestHistogram` has a minimum of 7 and a maximum of 6, so a coherent one is added. Generated-by: Claude Code (Claude Opus 5) --- .../prometheus/CoherentTestHistogram.java | 99 +++++++ .../PrometheusExpositionContractTest.java | 246 ++++++++++++++++++ .../flink/metrics/prometheus/TestUtils.java | 63 +++++ 3 files changed, 408 insertions(+) create mode 100644 flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/CoherentTestHistogram.java create mode 100644 flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusExpositionContractTest.java diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/CoherentTestHistogram.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/CoherentTestHistogram.java new file mode 100644 index 0000000000000..526e10e694a7b --- /dev/null +++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/CoherentTestHistogram.java @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.metrics.prometheus; + +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.HistogramStatistics; + +/** + * A histogram whose minimum is below every quantile and every quantile below its maximum. + * + *

The shared {@link org.apache.flink.metrics.util.TestHistogram} returns an ordinal value per + * accessor, so its minimum is 7 and its maximum 6. + */ +class CoherentTestHistogram implements Histogram { + + private static final long COUNT = 42; + private static final long MIN = 1; + private static final long MAX = 98; + + static double expectedQuantile(double quantile) { + return 2 + quantile * 95; + } + + static long expectedCount() { + return COUNT; + } + + static long expectedMin() { + return MIN; + } + + static long expectedMax() { + return MAX; + } + + @Override + public void update(long value) {} + + @Override + public long getCount() { + return COUNT; + } + + @Override + public HistogramStatistics getStatistics() { + return new HistogramStatistics() { + @Override + public double getQuantile(double quantile) { + return expectedQuantile(quantile); + } + + @Override + public long[] getValues() { + return new long[0]; + } + + @Override + public int size() { + return (int) COUNT; + } + + @Override + public double getMean() { + return (MIN + MAX) / 2.0; + } + + @Override + public double getStdDev() { + return 1; + } + + @Override + public long getMax() { + return MAX; + } + + @Override + public long getMin() { + return MIN; + } + }; + } +} diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusExpositionContractTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusExpositionContractTest.java new file mode 100644 index 0000000000000..81f08dcdcfd2c --- /dev/null +++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusExpositionContractTest.java @@ -0,0 +1,246 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.metrics.prometheus; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.MetricConfig; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.SimpleCounter; +import org.apache.flink.metrics.util.TestMeter; +import org.apache.flink.testutils.logging.LoggerAuditingExtension; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.slf4j.event.Level; + +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Asserts the data the reporter exposes: family names and types, labels and values. A failure here + * means users broke, not that the client changed its syntax. + */ +class PrometheusExpositionContractTest { + + private static final String LOGICAL_SCOPE = "logical_scope"; + private static final String[] LABEL_NAMES = {"label1", "label2"}; + private static final String[] LABEL_VALUES = {"value1", "value2"}; + private static final String LABELS = "{label1=\"value1\",label2=\"value2\",}"; + + /** An unexpected warning means a metric silently went missing. */ + @RegisterExtension + private final LoggerAuditingExtension loggerExtension = + new LoggerAuditingExtension(PrometheusReporter.class, Level.WARN); + + private PrometheusReporter reporter; + private MetricGroup metricGroup; + + @BeforeEach + void setUp() { + reporter = TestUtils.reporterOnFreePort(); + metricGroup = + TestUtils.createTestMetricGroup( + LOGICAL_SCOPE, TestUtils.toMap(LABEL_NAMES, LABEL_VALUES)); + } + + @AfterEach + void tearDown() { + if (reporter != null) { + reporter.close(); + } + assertThat(loggerExtension.getMessages()) + .as("the reporter logged a warning, so a metric was dropped") + .isEmpty(); + } + + /** + * A Flink Counter and Meter export as a gauge, since a Prometheus counter may not decrease. So + * simpleclient 0.10.0's {@code _total} rule, which applies only to a counter family, reaches + * nothing we emit, including a metric already named {@code _total}. + */ + @Test + void counterGaugeAndMeterAreGaugesAndKeepTheirNameExactly() throws Exception { + reporter.notifyOfAddedMetric(counterWith(7), "numRecordsIn", metricGroup); + reporter.notifyOfAddedMetric(counterWith(11), "records-consumed-total", metricGroup); + reporter.notifyOfAddedMetric((Gauge) () -> 3, "someGauge", metricGroup); + reporter.notifyOfAddedMetric(new TestMeter(), "someMeter", metricGroup); + + final String body = TestUtils.scrapeBody(reporter); + + for (String name : + Arrays.asList("numRecordsIn", "records_consumed_total", "someGauge", "someMeter")) { + assertThat(body).contains("# TYPE " + scoped(name) + " gauge\n"); + } + assertThat(body).contains(scoped("numRecordsIn") + LABELS + " 7.0\n"); + assertThat(body).contains(scoped("records_consumed_total") + LABELS + " 11.0\n"); + assertThat(body).contains(scoped("someGauge") + LABELS + " 3.0\n"); + // A Meter exports its rate and nothing else; there is no companion count. + assertThat(body).contains(scoped("someMeter") + LABELS + " 5.0\n"); + assertThat(TestUtils.samplesNamed(body, scoped("someMeter") + "_count")).isZero(); + } + + /** + * A Flink metric is a gauge or a summary, so none of the family shapes a newer client may emit + * may appear, whichever client is bundled. + */ + @Test + void noReservedFamilyShapeIsEverEmitted() throws Exception { + reporter.notifyOfAddedMetric(counterWith(7), "someCounter", metricGroup); + reporter.notifyOfAddedMetric(new TestMeter(), "someMeter", metricGroup); + reporter.notifyOfAddedMetric(new CoherentTestHistogram(), "someHistogram", metricGroup); + + final String body = TestUtils.scrapeBody(reporter); + + assertThat( + Arrays.stream(body.split("\n")) + .filter(line -> line.startsWith("# TYPE ")) + .collect(Collectors.toList())) + .containsExactlyInAnyOrder( + "# TYPE " + scoped("someCounter") + " gauge", + "# TYPE " + scoped("someMeter") + " gauge", + "# TYPE " + scoped("someHistogram") + " summary"); + for (String suffix : Arrays.asList("_created", "_info", "_bucket", "_gcount", "_gsum")) { + assertThat(body).doesNotContain(suffix + "{").doesNotContain(suffix + " "); + } + } + + /** + * A Flink Histogram becomes a summary with a count and eight quantiles, the minimum and maximum + * being the 0 and 1 quantiles. There is no {@code _sum}: the statistics expose only a sample of + * recent values (FLINK-29037). + */ + @Test + void histogramIsACountAndEightQuantilesWithNoSum() throws Exception { + reporter.notifyOfAddedMetric(new CoherentTestHistogram(), "someHistogram", metricGroup); + + final String body = TestUtils.scrapeBody(reporter); + final String family = scoped("someHistogram"); + + assertThat(body).contains("# TYPE " + family + " summary\n"); + assertThat(TestUtils.samplesNamed(body, family + "_count")).isOne(); + assertThat(body) + .contains( + family + + "_count" + + LABELS + + " " + + CoherentTestHistogram.expectedCount() + + ".0\n"); + assertThat(TestUtils.samplesNamed(body, family + "_sum")).isZero(); + + assertThat(TestUtils.samplesNamed(body, family)).isEqualTo(8); + assertThat(quantile(body, family, "0.0")).isEqualTo(CoherentTestHistogram.expectedMin()); + assertThat(quantile(body, family, "1.0")).isEqualTo(CoherentTestHistogram.expectedMax()); + for (String q : Arrays.asList("0.5", "0.75", "0.95", "0.98", "0.99", "0.999")) { + assertThat(quantile(body, family, q)) + .isEqualTo(CoherentTestHistogram.expectedQuantile(Double.parseDouble(q))); + } + } + + /** The family name is the prefixed logical scope and metric name, sanitised. */ + @Test + void unsupportedCharactersAreReplacedInTheFamilyName() throws Exception { + final MetricGroup group = + TestUtils.createTestMetricGroup( + "my.scope-here", TestUtils.toMap(LABEL_NAMES, LABEL_VALUES)); + reporter.notifyOfAddedMetric(counterWith(1), "my:metric-name", group); + + // A colon is legal in a metric name and is not replaced. + assertThat(TestUtils.scrapeBody(reporter)) + .contains("# TYPE flink_my_scope_here_my:metric_name gauge\n"); + } + + /** + * Label values are sanitised by default and passed through when {@code + * filterLabelValueCharacters} is off. Neither branch is covered today. + */ + @Test + void labelValuesAreFilteredUnlessTheOptionIsDisabled() throws Exception { + final Map variables = new LinkedHashMap<>(); + variables.put("", "a job, with \"quotes\" and\na newline"); + final MetricGroup group = TestUtils.createTestMetricGroup(LOGICAL_SCOPE, variables); + + reporter.open(new MetricConfig()); + reporter.notifyOfAddedMetric(counterWith(1), "filtered", group); + + assertThat(TestUtils.scrapeBody(reporter)) + .contains("job_name=\"a_job__with__quotes__and_a_newline\""); + + final PrometheusReporter unfiltered = TestUtils.reporterOnFreePort(); + try { + final MetricConfig config = new MetricConfig(); + config.setProperty( + PrometheusPushGatewayReporterOptions.FILTER_LABEL_VALUE_CHARACTER.key(), + "false"); + unfiltered.open(config); + unfiltered.notifyOfAddedMetric(counterWith(1), "unfiltered", group); + + // The client escapes what the filter would otherwise have removed. + assertThat(TestUtils.scrapeBody(unfiltered)) + .contains("job_name=\"a job, with \\\"quotes\\\" and\\na newline\""); + } finally { + unfiltered.close(); + } + } + + /** + * A user Gauge may return any double, and a histogram with no observations reports {@code NaN}. + */ + @Test + void nonFiniteGaugeValuesReachTheWire() throws Exception { + reporter.notifyOfAddedMetric((Gauge) () -> Double.NaN, "nan", metricGroup); + reporter.notifyOfAddedMetric( + (Gauge) () -> Double.POSITIVE_INFINITY, "posInf", metricGroup); + reporter.notifyOfAddedMetric( + (Gauge) () -> Double.NEGATIVE_INFINITY, "negInf", metricGroup); + + final String body = TestUtils.scrapeBody(reporter); + + assertThat(body).contains(scoped("nan") + LABELS + " NaN\n"); + assertThat(body).contains(scoped("posInf") + LABELS + " +Inf\n"); + assertThat(body).contains(scoped("negInf") + LABELS + " -Inf\n"); + } + + private static double quantile(String body, String family, String q) { + final String needle = "quantile=\"" + q + "\",} "; + for (String line : body.split("\n")) { + if (line.startsWith(family + "{") && line.contains(needle)) { + return Double.parseDouble(line.substring(line.indexOf(needle) + needle.length())); + } + } + throw new AssertionError("no sample for quantile " + q); + } + + private static Counter counterWith(long count) { + final Counter counter = new SimpleCounter(); + counter.inc(count); + return counter; + } + + private static String scoped(String metricName) { + return "flink_" + LOGICAL_SCOPE + "_" + metricName; + } +} diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/TestUtils.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/TestUtils.java index f99844b55912f..90099d4b2413f 100644 --- a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/TestUtils.java +++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/TestUtils.java @@ -17,15 +17,78 @@ package org.apache.flink.metrics.prometheus; +import org.apache.flink.metrics.Metric; import org.apache.flink.metrics.MetricGroup; import org.apache.flink.metrics.util.TestMetricGroup; import org.apache.flink.runtime.metrics.scope.ScopeFormat; +import org.apache.flink.util.NetUtils; +import org.apache.flink.util.PortRange; +import javax.annotation.Nullable; + +import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.util.Arrays; import java.util.LinkedHashMap; import java.util.Map; class TestUtils { + private static final HttpClient HTTP_CLIENT = HttpClient.newHttpClient(); + + public static PrometheusReporter reporterOnFreePort() { + try (NetUtils.Port port = NetUtils.getAvailablePort()) { + return new PrometheusReporter(new PortRange(String.valueOf(port.getPort()))); + } catch (Exception e) { + throw new RuntimeException("Could not start a reporter on a free port.", e); + } + } + + public static HttpResponse scrape(int port, @Nullable String accept) + throws IOException, InterruptedException { + return request(port, "/metrics", accept); + } + + public static HttpResponse request(int port, String path, @Nullable String accept) + throws IOException, InterruptedException { + final HttpRequest.Builder request = + HttpRequest.newBuilder().uri(URI.create("http://localhost:" + port + path)).GET(); + if (accept != null) { + request.header("Accept", accept); + } + return HTTP_CLIENT.send(request.build(), HttpResponse.BodyHandlers.ofString()); + } + + /** The reporter's current output. */ + public static String scrapeBody(PrometheusReporter reporter) + throws IOException, InterruptedException { + return scrape(reporter.getPort(), null).body(); + } + + /** How many samples in the body carry this exact sample name. */ + public static long samplesNamed(String body, String sampleName) { + return Arrays.stream(body.split("\n")) + .filter(l -> l.startsWith(sampleName + "{") || l.startsWith(sampleName + " ")) + .count(); + } + + /** Registers a metric the way {@code MetricRegistryImpl} does, swallowing what it throws. */ + public static void registerLikeRegistry( + AbstractPrometheusReporter reporter, + Metric metric, + String metricName, + MetricGroup group) { + try { + reporter.notifyOfAddedMetric(metric, metricName, group); + } catch (Exception e) { + // The registry catches and logs whatever a reporter throws, so a metric lost this way + // looks the same to a user as one the reporter dropped itself. + } + } + public static MetricGroup createTestMetricGroup( String logicalScope, Map variables) { return TestMetricGroup.newBuilder() From c78d436e9c5ffe32dc10a3feb3086e0cab7ea673 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 14 Sep 2026 14:03:08 +0200 Subject: [PATCH 2/4] [FLINK-40641][metrics] Pin the reserved-name collisions of the Prometheus reporter A summary occupies more names than its own, and which ones is a property of the client. When that set grows a metric stops being exported and we only log a warning. Generated-by: Claude Code (Claude Opus 5) --- .../PrometheusReservedNameTest.java | 166 ++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReservedNameTest.java diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReservedNameTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReservedNameTest.java new file mode 100644 index 0000000000000..ddd9322f8739f --- /dev/null +++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReservedNameTest.java @@ -0,0 +1,166 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.metrics.prometheus; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Metric; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.SimpleCounter; +import org.apache.flink.testutils.logging.LoggerAuditingExtension; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.slf4j.event.Level; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Records what happens when a user metric collides with a name reserved for a summary. + * + *

Which names a summary occupies is a property of the client. When that set grows, a user metric + * stops being exported and we only log a warning. The suffixes are the format's reserved set, not + * the set the client checks. + */ +class PrometheusReservedNameTest { + + private static final String LOGICAL_SCOPE = "logical_scope"; + private static final String BASE = "hazard"; + + private static final List RESERVED_SUFFIXES = + Arrays.asList( + "_count", "_sum", "_created", "_bucket", "_total", "_info", "_gcount", "_gsum"); + + /** Losing a metric is only acceptable because we say so in the log. */ + @RegisterExtension + private final LoggerAuditingExtension loggerExtension = + new LoggerAuditingExtension(PrometheusReporter.class, Level.WARN); + + private PrometheusReporter reporter; + private MetricGroup metricGroup; + + enum Disposition { + BOTH_EXPORTED, + SECOND_LOST + } + + enum Order { + GAUGE_FIRST, + SUMMARY_FIRST + } + + @BeforeEach + void setUp() { + reporter = TestUtils.reporterOnFreePort(); + metricGroup = TestUtils.createTestMetricGroup(LOGICAL_SCOPE, Collections.emptyMap()); + } + + @AfterEach + void tearDown() { + if (reporter != null) { + reporter.close(); + } + } + + static Stream collisions() { + final List rows = new ArrayList<>(); + for (String suffix : RESERVED_SUFFIXES) { + for (Order order : Order.values()) { + // Only the names a summary actually occupies can collide. On the client Flink + // bundles today that is the count and the sum; a client that also reserves the + // creation timestamp flips the _created rows. + final Disposition expected = + suffix.equals("_count") || suffix.equals("_sum") + ? Disposition.SECOND_LOST + : Disposition.BOTH_EXPORTED; + rows.add(Arguments.of(suffix, order, expected)); + } + } + return rows.stream(); + } + + @ParameterizedTest(name = "gauge named {0} and summary, {1}") + @MethodSource("collisions") + void reservedNameCollisionDisposition(String suffix, Order order, Disposition expected) + throws Exception { + final String gaugeName = BASE + suffix; + + if (order == Order.GAUGE_FIRST) { + register(counter(), gaugeName); + register(new CoherentTestHistogram(), BASE); + } else { + register(new CoherentTestHistogram(), BASE); + register(counter(), gaugeName); + } + + final String body = TestUtils.scrapeBody(reporter); + final boolean gaugeExported = hasFamilyOfType(body, scoped(gaugeName), "gauge"); + final boolean summaryExported = hasFamilyOfType(body, scoped(BASE), "summary"); + + final Disposition actual = + gaugeExported && summaryExported + ? Disposition.BOTH_EXPORTED + : Disposition.SECOND_LOST; + assertThat(actual) + .as( + "a gauge named %s registered %s alongside a summary named ", + suffix, order) + .isEqualTo(expected); + + if (expected == Disposition.SECOND_LOST) { + // Whichever was registered second is the one that went missing. + assertThat(order == Order.GAUGE_FIRST ? summaryExported : gaugeExported).isFalse(); + assertThat(order == Order.GAUGE_FIRST ? gaugeExported : summaryExported).isTrue(); + assertThat(loggerExtension.getMessages()) + .anyMatch( + message -> message.contains("There was a problem registering metric")); + } else { + assertThat(loggerExtension.getMessages()).isEmpty(); + } + } + + private void register(Metric metric, String metricName) { + // The registry swallows whatever a reporter throws, so a user sees a log line and a missing + // metric either way. The test has to treat both paths as the same outcome. + TestUtils.registerLikeRegistry(reporter, metric, metricName, metricGroup); + } + + private static boolean hasFamilyOfType(String body, String familyName, String type) { + return body.contains("# TYPE " + familyName + " " + type + "\n"); + } + + private static Counter counter() { + final Counter counter = new SimpleCounter(); + counter.inc(5); + return counter; + } + + private static String scoped(String metricName) { + return "flink_" + LOGICAL_SCOPE + "_" + metricName; + } +} From 05f947d4021ca6c8bc897cf90401846e1c6625a3 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 14 Sep 2026 14:03:08 +0200 Subject: [PATCH 3/4] [FLINK-40641][metrics] Pin the served paths and content negotiation of the Prometheus reporter Today only `/metrics` is scraped, so the root going unserved would go unnoticed, and no test sends an `Accept` header. A real Prometheus server asks for OpenMetrics. Generated-by: Claude Code (Claude Opus 5) --- .../prometheus/PrometheusHttpSurfaceTest.java | 151 ++++++++++++++++++ 1 file changed, 151 insertions(+) create mode 100644 flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusHttpSurfaceTest.java diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusHttpSurfaceTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusHttpSurfaceTest.java new file mode 100644 index 0000000000000..329aa583f98e5 --- /dev/null +++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusHttpSurfaceTest.java @@ -0,0 +1,151 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.metrics.prometheus; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.SimpleCounter; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.net.ConnectException; +import java.util.Collections; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Asserts the paths that serve metrics and what the reporter answers for a given {@code Accept} + * header. Neither is covered today, and a real Prometheus server asks for OpenMetrics. + */ +class PrometheusHttpSurfaceTest { + + private static final String LOGICAL_SCOPE = "logical_scope"; + private static final String TEXT_PLAIN_0_0_4 = "text/plain; version=0.0.4; charset=utf-8"; + + /** + * Verbatim from a Prometheus 3.11.2 server scraping a logging endpoint. This is what production + * sends, and it is the row that matters. + */ + private static final String PROMETHEUS_3_ACCEPT = + "application/openmetrics-text;version=1.0.0;escaping=allow-utf-8;q=0.6," + + "application/openmetrics-text;version=0.0.1;q=0.5," + + "text/plain;version=1.0.0;escaping=allow-utf-8;q=0.4," + + "text/plain;version=0.0.4;q=0.3,*/*;q=0.2"; + + private PrometheusReporter reporter; + + @BeforeEach + void setUp() { + reporter = TestUtils.reporterOnFreePort(); + final MetricGroup group = + TestUtils.createTestMetricGroup(LOGICAL_SCOPE, Collections.emptyMap()); + final Counter counter = new SimpleCounter(); + counter.inc(7); + reporter.notifyOfAddedMetric(counter, "someCounter", group); + reporter.notifyOfAddedMetric(new CoherentTestHistogram(), "someHistogram", group); + } + + @AfterEach + void tearDown() { + if (reporter != null) { + reporter.close(); + } + } + + /** + * Metrics are served on the root as well as on {@code /metrics}. A stock {@code prometheus.yml} + * defaults to {@code /metrics}, so a client that stopped serving the root would go unnoticed. + */ + @Test + void metricsAreServedOnTheRootAndOnTheMetricsPath() throws Exception { + for (String path : new String[] {"/", "/metrics"}) { + assertThat(TestUtils.request(reporter.getPort(), path, null).statusCode()) + .as("status of GET %s", path) + .isEqualTo(200); + assertThat(TestUtils.request(reporter.getPort(), path, null).body()) + .as("families served on GET %s", path) + .contains("# TYPE " + scoped("someCounter") + " gauge") + .contains("# TYPE " + scoped("someHistogram") + " summary"); + } + } + + static Stream acceptHeaders() { + return Stream.of( + Arguments.of("no Accept header", null, TEXT_PLAIN_0_0_4), + Arguments.of("*/*", "*/*", TEXT_PLAIN_0_0_4), + Arguments.of("legacy text", "text/plain;version=0.0.4", TEXT_PLAIN_0_0_4), + Arguments.of( + "OpenMetrics", + "application/openmetrics-text;version=1.0.0", + TEXT_PLAIN_0_0_4), + Arguments.of("Prometheus 3.11.2", PROMETHEUS_3_ACCEPT, TEXT_PLAIN_0_0_4)); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("acceptHeaders") + void contentTypeForAcceptHeader(String name, String accept, String expectedContentType) + throws Exception { + assertThat( + TestUtils.scrape(reporter.getPort(), accept) + .headers() + .firstValue("content-type")) + .hasValue(expectedContentType); + } + + /** There is no dedicated health endpoint: an unmatched path answers with the metrics body. */ + @Test + void healthPathServesTheMetricsBody() throws Exception { + assertThat(TestUtils.request(reporter.getPort(), "/-/healthy", null).body()) + .contains(scoped("someCounter")); + } + + @Test + void sampleNameFilterIsHonoured() throws Exception { + final String filtered = + TestUtils.request( + reporter.getPort(), + "/metrics?name%5B%5D=" + scoped("someCounter"), + null) + .body(); + + assertThat(filtered) + .contains(scoped("someCounter")) + .doesNotContain(scoped("someHistogram")); + } + + @Test + void closingTheReporterReleasesThePort() throws Exception { + final int port = reporter.getPort(); + reporter.close(); + reporter = null; + + assertThatThrownBy(() -> TestUtils.scrape(port, null)).isInstanceOf(ConnectException.class); + } + + private static String scoped(String metricName) { + return "flink_" + LOGICAL_SCOPE + "_" + metricName; + } +} From cdc393efc33bb2aa3087cfb9f99835a184cef959 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 14 Sep 2026 14:03:08 +0200 Subject: [PATCH 4/4] [FLINK-40641][metrics] Cover the PrometheusPushGatewayReporter wire FLINK-40592 covers the Authorization header, and with it the method and job path. This covers the encoding, body, content type, shutdown and failure. A path prefix in `hostUrl` is preserved. Generated-by: Claude Code (Claude Opus 5) --- .../PrometheusPushGatewayWireTest.java | 251 ++++++++++++++++++ 1 file changed, 251 insertions(+) create mode 100644 flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayWireTest.java diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayWireTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayWireTest.java new file mode 100644 index 0000000000000..fce213a002abc --- /dev/null +++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayWireTest.java @@ -0,0 +1,251 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.metrics.prometheus; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.MetricConfig; +import org.apache.flink.metrics.SimpleCounter; +import org.apache.flink.testutils.logging.LoggerAuditingExtension; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.slf4j.event.Level; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.DELETE_ON_SHUTDOWN; +import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.GROUPING_KEY; +import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.HOST_URL; +import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.JOB_NAME; +import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.RANDOM_JOB_NAME_SUFFIX; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Asserts the requests the push reporter sends: the path, the job name and grouping key encoded + * into it, the body, and what happens on shutdown and on failure. The Authorization header is + * covered by {@link PrometheusPushGatewayReporterAuthenticationTest}. + */ +class PrometheusPushGatewayWireTest { + + private static final String LOGICAL_SCOPE = "scope"; + + @RegisterExtension + private final LoggerAuditingExtension loggerExtension = + new LoggerAuditingExtension(PrometheusPushGatewayReporter.class, Level.WARN); + + private RecordingGateway gateway; + private boolean expectWarnings; + + @BeforeEach + void startGateway() throws IOException { + gateway = new RecordingGateway(); + } + + @AfterEach + void stopGateway() { + if (gateway != null) { + gateway.stop(); + } + if (!expectWarnings) { + assertThat(loggerExtension.getMessages()) + .as("the reporter logged a warning, so the gateway rejected a request") + .isEmpty(); + } + } + + /** + * A gateway reached through an ingress is configured with a path, and that path has to survive + * into the request. + */ + @Test + void aPathPrefixInTheHostUrlIsPreserved() throws Exception { + final MetricConfig config = config("myJob"); + config.setProperty(HOST_URL.key(), gateway.baseUrl() + "/behind/an/ingress"); + + reportAndClose(config); + + assertThat(gateway.requests().get(0).path) + .isEqualTo("/behind/an/ingress/metrics/job/myJob"); + } + + @Test + void groupingKeyEntriesBecomePathSegments() throws Exception { + final MetricConfig config = config("myJob"); + config.setProperty(GROUPING_KEY.key(), "k1=v1;k2=v2"); + + reportAndClose(config); + + // The grouping key is a map, so the segment order is not fixed; the pairs are. + assertThat(gateway.requests().get(0).path) + .startsWith("/metrics/job/myJob/") + .contains("/k1/v1") + .contains("/k2/v2"); + } + + @Test + void aSlashInTheJobNameIsBase64Encoded() throws Exception { + reportAndClose(config("my/job")); + + assertThat(gateway.requests().get(0).path).isEqualTo("/metrics/job@base64/bXkvam9i"); + } + + @Test + void aSlashInAGroupingKeyValueIsBase64Encoded() throws Exception { + final MetricConfig config = config("myJob"); + config.setProperty(GROUPING_KEY.key(), "instance=host/1"); + + reportAndClose(config); + + assertThat(gateway.requests().get(0).path) + .isEqualTo("/metrics/job/myJob/instance@base64/aG9zdC8x"); + } + + @Test + void theRandomJobNameSuffixIsAppendedToTheConfiguredName() throws Exception { + final MetricConfig config = config("myJob"); + config.setProperty(RANDOM_JOB_NAME_SUFFIX.key(), "true"); + + reportAndClose(config); + + assertThat(gateway.requests().get(0).path).matches("/metrics/job/myJob[0-9a-f]{32}"); + } + + @Test + void thePushedBodyIsTheExpositionAndTheContentTypeIsTheLegacyTextFormat() throws Exception { + reportAndClose(config("myJob")); + + final RecordedRequest push = gateway.requests().get(0); + assertThat(push.contentType).isEqualTo("text/plain; version=0.0.4; charset=utf-8"); + assertThat(push.body).contains("flink_" + LOGICAL_SCOPE + "_someCounter 7.0\n"); + // A delete carries no body. Its method is asserted by the authentication test. + assertThat(gateway.requests().get(1).body).isEmpty(); + } + + @Test + void nothingIsDeletedWhenDeleteOnShutdownIsDisabled() throws Exception { + final MetricConfig config = config("myJob"); + config.setProperty(DELETE_ON_SHUTDOWN.key(), "false"); + + reportAndClose(config); + + assertThat(gateway.requests()).hasSize(1); + assertThat(gateway.requests().get(0).method).isEqualTo("PUT"); + } + + @Test + void aFailedPushIsLoggedAndSwallowed() throws Exception { + expectWarnings = true; + gateway.respondWith(500); + + reportAndClose(config("myJob")); + + assertThat(loggerExtension.getMessages()) + .anyMatch(message -> message.contains("Failed to push metrics to PushGateway")); + } + + private MetricConfig config(String jobName) { + final MetricConfig config = new MetricConfig(); + config.setProperty(HOST_URL.key(), gateway.baseUrl()); + config.setProperty(JOB_NAME.key(), jobName); + config.setProperty(RANDOM_JOB_NAME_SUFFIX.key(), "false"); + config.setProperty(DELETE_ON_SHUTDOWN.key(), "true"); + return config; + } + + private void reportAndClose(MetricConfig config) { + final PrometheusPushGatewayReporter reporter = + new PrometheusPushGatewayReporterFactory().createMetricReporter(config); + try { + final Counter counter = new SimpleCounter(); + counter.inc(7); + reporter.notifyOfAddedMetric( + counter, + "someCounter", + TestUtils.createTestMetricGroup(LOGICAL_SCOPE, Collections.emptyMap())); + reporter.report(); + } finally { + reporter.close(); + } + } + + private static final class RecordedRequest { + private final String method; + private final String path; + private final String contentType; + private final String body; + + private RecordedRequest(HttpExchange exchange, byte[] body) { + this.method = exchange.getRequestMethod(); + this.path = exchange.getRequestURI().getRawPath(); + this.contentType = exchange.getRequestHeaders().getFirst("Content-Type"); + this.body = new String(body, StandardCharsets.UTF_8); + } + } + + private static final class RecordingGateway { + private final HttpServer server; + private final List requests = + Collections.synchronizedList(new ArrayList<>()); + private volatile int responseCode = 202; + + private RecordingGateway() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/", + exchange -> { + try { + // Record before responding, so a client that has its response has our + // record too. + requests.add( + new RecordedRequest( + exchange, exchange.getRequestBody().readAllBytes())); + exchange.sendResponseHeaders(responseCode, -1); + } finally { + exchange.close(); + } + }); + server.start(); + } + + private String baseUrl() { + return "http://127.0.0.1:" + server.getAddress().getPort(); + } + + private void respondWith(int code) { + responseCode = code; + } + + private List requests() { + return requests; + } + + private void stop() { + server.stop(0); + } + } +}