From cd408372c2b8ebd05dd52cea95b944f9d1284ab9 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 14 Sep 2026 09:22:25 +0200 Subject: [PATCH 1/2] [FLINK-40648][metrics] Do not report a metric whose variables collide into one label name Two metric group variables that differ only in characters the filter replaces become the same label name. Prometheus rejects such a body and abandons the whole scrape, so reporting the metric costs every other metric of the process. Removal has to tolerate a metric that was never registered. Generated-by: Claude Code (Claude Opus 5) --- .../AbstractPrometheusReporter.java | 30 +++++++++- .../prometheus/PrometheusReporterTest.java | 60 +++++++++++++++++++ 2 files changed, 88 insertions(+), 2 deletions(-) diff --git a/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java b/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java index 6354d88433af36..80758506a443d2 100644 --- a/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java +++ b/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java @@ -45,6 +45,8 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.regex.Pattern; import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.ALLOW_LIST; @@ -69,6 +71,9 @@ public abstract class AbstractPrometheusReporter implements MetricReporter { private final List allowLists = new ArrayList<>(); + /** Label names already reported as duplicated, so each is logged once and not per metric. */ + private final Set reportedDuplicateLabels = ConcurrentHashMap.newKeySet(); + @VisibleForTesting static String replaceInvalidChars(final String input) { // https://prometheus.io/docs/instrumenting/writing_exporters/ @@ -121,8 +126,16 @@ public void notifyOfAddedMetric( List dimensionValues = new LinkedList<>(); for (final Map.Entry dimension : group.getAllVariables().entrySet()) { final String key = dimension.getKey(); - dimensionKeys.add( - CHARACTER_FILTER.filterCharacters(key.substring(1, key.length() - 1))); + final String labelName = + CHARACTER_FILTER.filterCharacters(key.substring(1, key.length() - 1)); + if (dimensionKeys.contains(labelName)) { + // Prometheus refuses an exposition carrying the same label name twice and + // abandons the whole scrape, so reporting this metric would cost every other + // metric of this process as well. + warnAboutDuplicateLabel(labelName, metricName); + return; + } + dimensionKeys.add(labelName); dimensionValues.add(labelValueCharactersFilter.filterCharacters(dimension.getValue())); } @@ -158,6 +171,15 @@ public void notifyOfAddedMetric( } } + private void warnAboutDuplicateLabel(String labelName, String metricName) { + if (reportedDuplicateLabels.add(labelName)) { + log.warn( + "Multiple metric group variables map to the label name {}. Metrics carrying them, such as {}, will not be reported.", + labelName, + metricName); + } + } + private static String getScopedName(String metricName, MetricGroup group) { return SCOPE_PREFIX + getLogicalScope(group) @@ -259,6 +281,10 @@ public void notifyOfRemovedMetric( synchronized (this) { final AbstractMap.SimpleImmutableEntry collectorWithCount = collectorsWithCountByMetricName.get(scopedMetricName); + if (collectorWithCount == null) { + // The metric was refused, so there is nothing to remove. + return; + } final Integer count = collectorWithCount.getValue(); final Collector collector = collectorWithCount.getKey(); diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java index 2a0c3b148127ee..238f6b32219804 100644 --- a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java +++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java @@ -41,12 +41,14 @@ import java.net.http.HttpResponse; import java.util.Arrays; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; import java.util.NoSuchElementException; import static org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterOptions.ALLOW_LIST; import static org.apache.flink.metrics.prometheus.PrometheusReporterFactory.ARG_PORT; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Basic test for {@link PrometheusReporter}. */ @@ -183,6 +185,64 @@ void metricIsRemovedWhileOtherMetricsWithSameNameExist() assertThat(response).contains("some_value").doesNotContain(labelValueThatShouldBeRemoved); } + /** + * Two variables that differ only in characters the filter replaces become one label name, + * twice. Prometheus rejects such a body and abandons the whole scrape. + */ + @Test + void metricIsNotReportedWhenTwoVariablesSanitiseToTheSameLabelName() + throws IOException, InterruptedException { + final Map colliding = new LinkedHashMap<>(); + colliding.put("", "v1"); + colliding.put("", "v2"); + + reporter.notifyOfAddedMetric( + new SimpleCounter(), + "colliding", + TestUtils.createTestMetricGroup(LOGICAL_SCOPE, colliding)); + + final String response = pollMetrics(reporter.getPort()).body(); + + assertThat(response).doesNotContain("a_b=\"v1\",a_b=\"v2\""); + assertThat(response).doesNotContain(SCOPE_PREFIX + "colliding"); + } + + @Test + void removingARefusedMetricDoesNotThrow() { + final Map colliding = new LinkedHashMap<>(); + colliding.put("", "v1"); + colliding.put("", "v2"); + final MetricGroup group = TestUtils.createTestMetricGroup(LOGICAL_SCOPE, colliding); + final Counter counter = new SimpleCounter(); + + reporter.notifyOfAddedMetric(counter, "colliding", group); + + // The registry removes every metric it added, including the ones we refused. + assertThatCode(() -> reporter.notifyOfRemovedMetric(counter, "colliding", group)) + .doesNotThrowAnyException(); + } + + /** One unreportable metric must not cost the rest of the process its metrics. */ + @Test + void otherMetricsAreStillReportedAlongsideAnUnreportableOne() + throws IOException, InterruptedException { + final Map colliding = new LinkedHashMap<>(); + colliding.put("", "v1"); + colliding.put("", "v2"); + + reporter.notifyOfAddedMetric( + new SimpleCounter(), + "colliding", + TestUtils.createTestMetricGroup(LOGICAL_SCOPE, colliding)); + final Counter healthy = new SimpleCounter(); + healthy.inc(3); + reporter.notifyOfAddedMetric(healthy, "healthy", metricGroup); + + final String response = pollMetrics(reporter.getPort()).body(); + + assertThat(response).contains(SCOPE_PREFIX + "healthy"); + } + @Test void invalidCharactersAreReplacedWithUnderscore() { assertThat(PrometheusReporter.replaceInvalidChars("")).isEqualTo(""); From 0140aaf8eba8f677582b030f0232c5a185407de2 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 14 Sep 2026 09:51:26 +0200 Subject: [PATCH 2/2] [FLINK-32649][metrics] Do not report a metric whose variables differ from an earlier metric of the same name A collector is stored under the scoped metric name alone, so a metric whose variables differ from the first one registered reuses it and its values are published under the other metric's label names. Generated-by: Claude Code (Claude Opus 5) --- .../AbstractPrometheusReporter.java | 66 +++++++++++++------ .../prometheus/PrometheusReporterTest.java | 50 +++++++++++++- 2 files changed, 96 insertions(+), 20 deletions(-) diff --git a/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java b/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java index 80758506a443d2..dcfabf5ff299c3 100644 --- a/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java +++ b/flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java @@ -37,7 +37,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.AbstractMap; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -66,14 +65,15 @@ public abstract class AbstractPrometheusReporter implements MetricReporter { @VisibleForTesting static final String SCOPE_PREFIX = "flink" + SCOPE_SEPARATOR; @VisibleForTesting static final char DEFAULT_SCOPE_SEPARATOR = '.'; - private final Map> - collectorsWithCountByMetricName = new HashMap<>(); + private final Map collectorsByMetricName = new HashMap<>(); private final List allowLists = new ArrayList<>(); /** Label names already reported as duplicated, so each is logged once and not per metric. */ private final Set reportedDuplicateLabels = ConcurrentHashMap.newKeySet(); + private final Set reportedLabelNameMismatches = ConcurrentHashMap.newKeySet(); + @VisibleForTesting static String replaceInvalidChars(final String input) { // https://prometheus.io/docs/instrumenting/writing_exporters/ @@ -146,11 +146,18 @@ public void notifyOfAddedMetric( Integer count = 0; synchronized (this) { - if (collectorsWithCountByMetricName.containsKey(scopedMetricName)) { - final AbstractMap.SimpleImmutableEntry collectorWithCount = - collectorsWithCountByMetricName.get(scopedMetricName); - collector = collectorWithCount.getKey(); - count = collectorWithCount.getValue(); + final RegisteredCollector registered = collectorsByMetricName.get(scopedMetricName); + if (registered != null) { + if (!registered.labelNames.equals(dimensionKeys)) { + // The collector was built with the label names of the metric that registered + // first, and a Collector owns its name in the registry, so a second one cannot + // take its place. Reporting this metric anyway would publish its values under + // the other metric's label names. + warnAboutLabelNameMismatch(scopedMetricName, metricName); + return; + } + collector = registered.collector; + count = registered.count; } else { collector = createCollector( @@ -166,8 +173,8 @@ public void notifyOfAddedMetric( } } addMetric(metric, dimensionValues, collector); - collectorsWithCountByMetricName.put( - scopedMetricName, new AbstractMap.SimpleImmutableEntry<>(collector, count + 1)); + collectorsByMetricName.put( + scopedMetricName, new RegisteredCollector(collector, dimensionKeys, count + 1)); } } @@ -180,6 +187,28 @@ private void warnAboutDuplicateLabel(String labelName, String metricName) { } } + private void warnAboutLabelNameMismatch(String scopedMetricName, String metricName) { + if (reportedLabelNameMismatches.add(scopedMetricName)) { + log.warn( + "Metrics named {} do not all carry the same metric group variables. Only those matching the first one reported, such as {}, will be reported.", + scopedMetricName, + metricName); + } + } + + /** A registered collector, the label names it was built with, and how many metrics use it. */ + private static final class RegisteredCollector { + private final Collector collector; + private final List labelNames; + private final int count; + + private RegisteredCollector(Collector collector, List labelNames, int count) { + this.collector = collector; + this.labelNames = labelNames; + this.count = count; + } + } + private static String getScopedName(String metricName, MetricGroup group) { return SCOPE_PREFIX + getLogicalScope(group) @@ -279,14 +308,13 @@ public void notifyOfRemovedMetric( final String scopedMetricName = getScopedName(metricName, group); synchronized (this) { - final AbstractMap.SimpleImmutableEntry collectorWithCount = - collectorsWithCountByMetricName.get(scopedMetricName); - if (collectorWithCount == null) { - // The metric was refused, so there is nothing to remove. + final RegisteredCollector registered = collectorsByMetricName.get(scopedMetricName); + if (registered == null) { + // The metric was never reported, so there is nothing to remove. return; } - final Integer count = collectorWithCount.getValue(); - final Collector collector = collectorWithCount.getKey(); + final int count = registered.count; + final Collector collector = registered.collector; removeMetric(metric, dimensionValues, collector); @@ -296,11 +324,11 @@ public void notifyOfRemovedMetric( } catch (Exception e) { log.warn("There was a problem unregistering metric {}.", scopedMetricName, e); } - collectorsWithCountByMetricName.remove(scopedMetricName); + collectorsByMetricName.remove(scopedMetricName); } else { - collectorsWithCountByMetricName.put( + collectorsByMetricName.put( scopedMetricName, - new AbstractMap.SimpleImmutableEntry<>(collector, count - 1)); + new RegisteredCollector(collector, registered.labelNames, count - 1)); } } } diff --git a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java index 238f6b32219804..022794ec544037 100644 --- a/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java +++ b/flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java @@ -170,7 +170,9 @@ void metricIsRemovedWhileOtherMetricsWithSameNameExist() Counter metric1 = new SimpleCounter(); Counter metric2 = new SimpleCounter(); - final Map variables2 = new HashMap<>(metricGroup.getAllVariables()); + // The variables are copied in iteration order: a metric reusing another's collector has to + // carry the same label names in the same order, since the values are bound by position. + final Map variables2 = new LinkedHashMap<>(metricGroup.getAllVariables()); final Map.Entry entryToModify = variables2.entrySet().iterator().next(); final String labelValueThatShouldBeRemoved = entryToModify.getValue(); variables2.put(entryToModify.getKey(), "some_value"); @@ -243,6 +245,52 @@ void otherMetricsAreStillReportedAlongsideAnUnreportableOne() assertThat(response).contains(SCOPE_PREFIX + "healthy"); } + /** + * Reporting a metric whose variables differ from the first one of that name would publish its + * values under that one's label names. + */ + @Test + void metricIsNotReportedWhenItsVariablesDifferFromAnEarlierMetricOfTheSameName() + throws IOException, InterruptedException { + final Counter first = new SimpleCounter(); + first.inc(1); + final Counter second = new SimpleCounter(); + second.inc(2); + + reporter.notifyOfAddedMetric(first, "m", groupWith("x", "1", "y", "2")); + reporter.notifyOfAddedMetric(second, "m", groupWith("x", "1", "z", "9")); + + final String response = pollMetrics(reporter.getPort()).body(); + + assertThat(response).contains("y=\"2\"").doesNotContain("y=\"9\""); + assertThat(response).doesNotContain("z="); + } + + /** The same name with the same variables is the case the shared collector exists for. */ + @Test + void metricIsReportedWhenItsVariablesMatchAnEarlierMetricOfTheSameName() + throws IOException, InterruptedException { + final Counter first = new SimpleCounter(); + first.inc(1); + final Counter second = new SimpleCounter(); + second.inc(2); + + reporter.notifyOfAddedMetric(first, "m", groupWith("x", "1", "y", "2")); + reporter.notifyOfAddedMetric(second, "m", groupWith("x", "1", "y", "other")); + + final String response = pollMetrics(reporter.getPort()).body(); + + assertThat(response).contains("y=\"2\"").contains("y=\"other\""); + } + + private static MetricGroup groupWith(String... keysAndValues) { + final Map variables = new HashMap<>(); + for (int i = 0; i < keysAndValues.length; i += 2) { + variables.put("<" + keysAndValues[i] + ">", keysAndValues[i + 1]); + } + return TestUtils.createTestMetricGroup(LOGICAL_SCOPE, variables); + } + @Test void invalidCharactersAreReplacedWithUnderscore() { assertThat(PrometheusReporter.replaceInvalidChars("")).isEqualTo("");