[FLINK-40648][metrics] Do not report a metric whose variables collide into one label name

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<String> allowLists = new ArrayList<>();
 
+    /** Label names already reported as duplicated, so each is logged once and not per metric. */
+    private final Set<String> 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<String> dimensionValues = new LinkedList<>();
         for (final Map.Entry<String, String> 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<Collector, Integer> 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();
 

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<String, String> colliding = new LinkedHashMap<>();
+        colliding.put("<a.b>", "v1");
+        colliding.put("<a-b>", "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<String, String> colliding = new LinkedHashMap<>();
+        colliding.put("<a.b>", "v1");
+        colliding.put("<a-b>", "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<String, String> colliding = new LinkedHashMap<>();
+        colliding.put("<a.b>", "v1");
+        colliding.put("<a-b>", "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("");