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..dcfabf5ff299c 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; @@ -45,6 +44,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; @@ -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> - 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/ @@ -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())); } @@ -133,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( @@ -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 labelNames; + private final int count; + + private RegisteredCollector(Collector collector, List 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 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)); } } } 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..022794ec54403 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}. */ @@ -168,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"); @@ -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 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"); + } + + /** + * 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("");