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 6354d88433af3..80758506a443d 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 2a0c3b148127e..238f6b3221980 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("");