[FLINK-32649][metrics] Do not report a metric whose variables differ from an earlier metric of the same name

flink-metrics/flink-metrics-prometheus/src/main/java/org/apache/flink/metrics/prometheus/AbstractPrometheusReporter.java

@@ -37,14 +37,15 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 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;
@@ -64,11 +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<String, AbstractMap.SimpleImmutableEntry<Collector, Integer>>
-            collectorsWithCountByMetricName = new HashMap<>();
+    private final Map<String, RegisteredCollector> collectorsByMetricName = new HashMap<>();
 
     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();
+
+    private final Set<String> reportedLabelNameMismatches = 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()));
         }
 
@@ -133,11 +146,18 @@ public void notifyOfAddedMetric(
         Integer count = 0;
 
         synchronized (this) {
-            if (collectorsWithCountByMetricName.containsKey(scopedMetricName)) {
-                final AbstractMap.SimpleImmutableEntry<Collector, Integer> 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(
@@ -153,8 +173,39 @@ 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));
+        }
+    }
+
+    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 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<String> labelNames;
+        private final int count;
+
+        private RegisteredCollector(Collector collector, List<String> labelNames, int count) {
+            this.collector = collector;
+            this.labelNames = labelNames;
+            this.count = count;
         }
     }
 
@@ -257,10 +308,13 @@ public void notifyOfRemovedMetric(
 
         final String scopedMetricName = getScopedName(metricName, group);
         synchronized (this) {
-            final AbstractMap.SimpleImmutableEntry<Collector, Integer> collectorWithCount =
-                    collectorsWithCountByMetricName.get(scopedMetricName);
-            final Integer count = collectorWithCount.getValue();
-            final Collector collector = collectorWithCount.getKey();
+            final RegisteredCollector registered = collectorsByMetricName.get(scopedMetricName);
+            if (registered == null) {
+                // The metric was never reported, so there is nothing to remove.
+                return;
+            }
+            final int count = registered.count;
+            final Collector collector = registered.collector;
 
             removeMetric(metric, dimensionValues, collector);
 
@@ -270,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));
             }
         }
     }

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}. */
@@ -168,7 +170,9 @@ void metricIsRemovedWhileOtherMetricsWithSameNameExist()
         Counter metric1 = new SimpleCounter();
         Counter metric2 = new SimpleCounter();
 
-        final Map<String, String> 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<String, String> variables2 = new LinkedHashMap<>(metricGroup.getAllVariables());
         final Map.Entry<String, String> entryToModify = variables2.entrySet().iterator().next();
         final String labelValueThatShouldBeRemoved = entryToModify.getValue();
         variables2.put(entryToModify.getKey(), "some_value");
@@ -183,6 +187,110 @@ 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");
+    }
+
+    /**
+     * 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<String, String> 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("");