diff --git a/docs/content/docs/ops/metrics.md b/docs/content/docs/ops/metrics.md index c6ad679ce2d54..3d057bb218f1d 100644 --- a/docs/content/docs/ops/metrics.md +++ b/docs/content/docs/ops/metrics.md @@ -1963,6 +1963,9 @@ Note that the metrics are only available via reporters. watermark.currentWatermark The last watermark this split has received (in milliseconds). +

Note: Split metrics are reported to metric reporters only. By default they are not + exposed via the REST API or the web UI, because the number of splits a source reads over its lifetime is + unbounded; see metrics.internal.query-service.filter.excludes to change this.

Gauge diff --git a/docs/layouts/shortcodes/generated/metric_configuration.html b/docs/layouts/shortcodes/generated/metric_configuration.html index c5eecb7eab54d..f855980619ca8 100644 --- a/docs/layouts/shortcodes/generated/metric_configuration.html +++ b/docs/layouts/shortcodes/generated/metric_configuration.html @@ -14,6 +14,12 @@ Duration Update interval for the metric fetcher used by the web UI. Decrease this value for faster updating metrics. Increase this value if the metric fetcher causes too much load. Setting this value to 0 disables the metric fetching completely. + +
metrics.internal.query-service.filter.excludes
+ "*.operator.split.watermark:*:*" + List<String> + The metrics that should be excluded from Flink's internal metric query service, which backs the metrics exposed via the REST API and the web UI. Excluded metrics are still reported to configured metric reporters. Filters use the same format as metrics.reporter.<name>.filter.excludes.
By default the per-split watermark metrics of sources are excluded, because the number of splits a source reads over its lifetime is unbounded and the REST API retains the metrics of every split it has ever seen. +
metrics.internal.query-service.port
"0" diff --git a/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java index 9a22534cb993c..b91cb74fee26e 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/MetricOptions.java @@ -391,6 +391,31 @@ public static Configuration forReporter(Configuration configuration, String repo + "The range of the priority is from 1 (MIN_PRIORITY) to 10 (MAX_PRIORITY). " + "Warning, increasing this value may bring the main Flink components down."); + /** + * Metrics matching one of these filters are not exposed through Flink's internal metric query + * service and therefore neither through the REST API nor the web UI. They are still reported to + * configured metric reporters. + */ + public static final ConfigOption> QUERY_SERVICE_EXCLUDES = + key("metrics.internal.query-service.filter.excludes") + .stringType() + .asList() + .defaultValues("*.operator.split.watermark:*:*") + .withDescription( + Description.builder() + .text( + "The metrics that should be excluded from Flink's internal metric query service," + + " which backs the metrics exposed via the REST API and the web UI." + + " Excluded metrics are still reported to configured metric reporters." + + " Filters use the same format as %s.", + code("metrics.reporter..filter.excludes")) + .linebreak() + .text( + "By default the per-split watermark metrics of sources are excluded, because the" + + " number of splits a source reads over its lifetime is unbounded and the" + + " REST API retains the metrics of every split it has ever seen.") + .build()); + /** * The config parameter defining the update interval for the metric fetcher used by the web UI * in milliseconds. diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryConfiguration.java index e81441f650cdc..d1970b374bdd3 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryConfiguration.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryConfiguration.java @@ -20,6 +20,9 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.MetricOptions; +import org.apache.flink.metrics.Metric; +import org.apache.flink.runtime.metrics.filter.DefaultReporterFilters; +import org.apache.flink.runtime.metrics.filter.ReporterFilter; import org.apache.flink.runtime.metrics.scope.ScopeFormats; import org.apache.flink.util.Preconditions; @@ -39,12 +42,19 @@ public class MetricRegistryConfiguration { private final long queryServiceMessageSizeLimit; + // decides which metrics are exposed through the metric query service + private final ReporterFilter queryServiceFilter; + public MetricRegistryConfiguration( - ScopeFormats scopeFormats, char delimiter, long queryServiceMessageSizeLimit) { + ScopeFormats scopeFormats, + char delimiter, + long queryServiceMessageSizeLimit, + ReporterFilter queryServiceFilter) { this.scopeFormats = Preconditions.checkNotNull(scopeFormats); this.delimiter = delimiter; this.queryServiceMessageSizeLimit = queryServiceMessageSizeLimit; + this.queryServiceFilter = Preconditions.checkNotNull(queryServiceFilter); } // ------------------------------------------------------------------------ @@ -63,6 +73,10 @@ public long getQueryServiceMessageSizeLimit() { return queryServiceMessageSizeLimit; } + public ReporterFilter getQueryServiceFilter() { + return queryServiceFilter; + } + // ------------------------------------------------------------------------ // Static factory methods // ------------------------------------------------------------------------ @@ -92,10 +106,30 @@ public static MetricRegistryConfiguration fromConfiguration( delim = '.'; } + ReporterFilter queryServiceFilter; + try { + queryServiceFilter = + DefaultReporterFilters.queryServiceMetricsFromConfiguration(configuration); + LOG.info( + "Metrics matching {} will not be exposed through the metric query service (REST API / web UI).", + configuration.get(MetricOptions.QUERY_SERVICE_EXCLUDES)); + } catch (Exception e) { + LOG.warn( + "Failed to parse {}, using the default value.", + MetricOptions.QUERY_SERVICE_EXCLUDES.key(), + e); + queryServiceFilter = + DefaultReporterFilters.queryServiceMetricsFromConfiguration( + new Configuration()); + } + // padding to account for serialization overhead final long messageSizeLimitPadding = 256; return new MetricRegistryConfiguration( - scopeFormats, delim, maximumFrameSize - messageSizeLimitPadding); + scopeFormats, + delim, + maximumFrameSize - messageSizeLimitPadding, + queryServiceFilter); } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java index 5a14c8c85dacc..8543ecc4e53a2 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java @@ -30,6 +30,7 @@ import org.apache.flink.metrics.reporter.Scheduled; import org.apache.flink.runtime.clusterframework.types.ResourceID; import org.apache.flink.runtime.metrics.dump.MetricQueryService; +import org.apache.flink.runtime.metrics.filter.ReporterFilter; import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup; import org.apache.flink.runtime.metrics.groups.FrontMetricGroup; import org.apache.flink.runtime.metrics.groups.ReporterScopedSettings; @@ -89,6 +90,8 @@ public class MetricRegistryImpl implements MetricRegistry, AutoCloseableAsync { private final long maximumFramesize; + private final ReporterFilter queryServiceFilter; + @Nullable private MetricQueryService queryService; @Nullable private RpcService metricQueryServiceRpcService; @@ -160,6 +163,7 @@ public MetricRegistryImpl( ScheduledExecutorService reporterScheduledExecutor, ScheduledExecutorService viewUpdaterScheduledExecutor) { this.maximumFramesize = config.getQueryServiceMessageSizeLimit(); + this.queryServiceFilter = config.getQueryServiceFilter(); this.scopeFormats = config.getScopeFormats(); this.globalDelimiter = config.getDelimiter(); this.terminationFuture = new CompletableFuture<>(); @@ -566,7 +570,16 @@ public void register(Metric metric, String metricName, AbstractMetricGroup gr } try { if (queryService != null) { - queryService.addMetric(metricName, metric, group); + final String logicalScope = + group.getLogicalScope(CharacterFilter.NO_OP_FILTER); + if (queryServiceFilter.filter(metric, metricName, logicalScope)) { + queryService.addMetric(metricName, metric, group); + } else { + LOG.trace( + "Skip exposing {}.{} through the metric query service due to filter rules.", + logicalScope, + metricName); + } } } catch (Exception e) { LOG.warn("Error while registering metric: {}.", metricName, e); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/filter/DefaultReporterFilters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/filter/DefaultReporterFilters.java index 5f9288c097abf..3a233d53bee25 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/filter/DefaultReporterFilters.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/filter/DefaultReporterFilters.java @@ -30,6 +30,7 @@ import org.apache.flink.traces.SpanBuilder; import java.util.Arrays; +import java.util.Collections; import java.util.EnumSet; import java.util.List; import java.util.function.BiFunction; @@ -38,7 +39,8 @@ /** * Default {@link ReporterFilter} implementation that filters metrics based on {@link - * MetricOptions#REPORTER_INCLUDES}/{@link MetricOptions#REPORTER_EXCLUDES}. + * MetricOptions#REPORTER_INCLUDES}/{@link MetricOptions#REPORTER_EXCLUDES}, or on {@link + * MetricOptions#QUERY_SERVICE_EXCLUDES} for the metric query service. */ public class DefaultReporterFilters { @@ -138,6 +140,21 @@ public static ReporterFilter metricsFromConfiguration(Configuration conf MetricReporterFilter::new); } + /** + * Creates the filter applied to metrics before they are exposed through the metric query + * service. The query service has no include list; everything not excluded via {@link + * MetricOptions#QUERY_SERVICE_EXCLUDES} is exposed. + */ + public static ReporterFilter queryServiceMetricsFromConfiguration( + Configuration configuration) { + final List excludeFilters = + configuration.get(MetricOptions.QUERY_SERVICE_EXCLUDES).stream() + .map(DefaultReporterFilters::parse) + .collect(Collectors.toList()); + + return new MetricReporterFilter(Collections.singletonList(parse("*:*:*")), excludeFilters); + } + public static ReporterFilter tracesFromConfiguration(Configuration configuration) { return fromConfiguration( configuration, diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java index d83af16b6a8c9..b02b0bd08b550 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/MetricRegistryImplTest.java @@ -18,6 +18,7 @@ package org.apache.flink.runtime.metrics; +import org.apache.flink.api.common.JobID; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.EventOptions; import org.apache.flink.configuration.MetricOptions; @@ -68,7 +69,9 @@ import java.util.Optional; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId; import static org.apache.flink.util.Preconditions.checkNotNull; import static org.assertj.core.api.Assertions.assertThat; @@ -622,6 +625,97 @@ void testMetricFiltering() { assertThat(reporter.getLastAddedMetric()).isNotEmpty(); } + @Test + void testQueryServiceExcludesSplitWatermarkMetricsByDefault() throws Exception { + final NotificationCapturingMetricReporter reporter = + new NotificationCapturingMetricReporter(); + final MetricRegistryImpl registry = + new MetricRegistryImpl( + MetricRegistryTestUtils.defaultMetricRegistryConfiguration(), + Collections.singletonList( + ReporterSetupBuilder.METRIC_SETUP_BUILDER.forReporter( + "test", reporter))); + try { + registry.startQueryService(new TestingRpcService(), new ResourceID("mqs")); + + final MetricGroup operatorGroup = + TaskManagerMetricGroup.createTaskManagerMetricGroup( + registry, "host", new ResourceID("tm")) + .addJob(new JobID(), "job") + .addTask(createExecutionAttemptId(), "task") + .getOrAddOperator("source"); + + operatorGroup + .addGroup("split", "split-0") + .addGroup("watermark") + .gauge("currentWatermark", () -> 42L); + // reporters are unaffected by the query service filter + assertThat(reporter.getLastAddedMetricName()).hasValue("currentWatermark"); + + operatorGroup.counter("marker"); + + assertThat(queryMetricNames(registry, "marker")) + .contains("marker") + .doesNotContain("currentWatermark"); + } finally { + registry.closeAsync().get(); + } + } + + @Test + void testQueryServiceExcludesAreConfigurable() throws Exception { + final Configuration configuration = new Configuration(); + configuration.set( + MetricOptions.QUERY_SERVICE_EXCLUDES, Collections.singletonList("*:excluded")); + final MetricRegistryImpl registry = + new MetricRegistryImpl(MetricRegistryTestUtils.fromConfiguration(configuration)); + try { + registry.startQueryService(new TestingRpcService(), new ResourceID("mqs")); + + final MetricGroup group = + TaskManagerMetricGroup.createTaskManagerMetricGroup( + registry, "host", new ResourceID("tm")); + group.counter("excluded"); + group.counter("included"); + + assertThat(queryMetricNames(registry, "included")) + .contains("included") + .doesNotContain("excluded"); + } finally { + registry.closeAsync().get(); + } + } + + /** + * Queries the metric query service until the dump contains {@code awaitedMetricName} and + * returns the names of all dumped metrics. Metrics are added to the query service + * asynchronously but in registration order, so once the awaited metric is present every metric + * registered before it has been processed as well. + */ + private static List queryMetricNames( + MetricRegistryImpl registry, String awaitedMetricName) throws Exception { + final MetricQueryServiceGateway gateway = + checkNotNull(registry.getMetricQueryServiceGateway()); + final MetricDumpSerialization.MetricDumpDeserializer deserializer = + new MetricDumpSerialization.MetricDumpDeserializer(); + + List names = Collections.emptyList(); + for (int attempt = 0; attempt < 100 && !names.contains(awaitedMetricName); attempt++) { + if (attempt > 0) { + Thread.sleep(50); + } + names = + deserializer + .deserialize( + gateway.queryMetrics(Duration.ofSeconds(5)) + .get(5, TimeUnit.SECONDS)) + .stream() + .map(dump -> dump.name) + .collect(Collectors.toList()); + } + return names; + } + @Test void testSpanFiltering() { final String includedGroupName = "foo"; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/filter/DefaultReporterFiltersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/filter/DefaultReporterFiltersTest.java index c93cb4f020e1f..d40626508bdcc 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/filter/DefaultReporterFiltersTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/filter/DefaultReporterFiltersTest.java @@ -89,6 +89,75 @@ void testParseMetricTypesCaseIgnored() { assertThat(types).containsExactlyInAnyOrder(MetricType.METER, MetricType.COUNTER); } + @Test + void testQueryServiceFilterExcludesSplitWatermarkMetricsByDefault() { + final ReporterFilter metricFilter = + DefaultReporterFilters.queryServiceMetricsFromConfiguration(new Configuration()); + + assertThat( + metricFilter.filter( + GAUGE, + "currentWatermark", + "taskmanager.job.task.operator.split.watermark")) + .isFalse(); + assertThat( + metricFilter.filter( + GAUGE, + "accumulatedIdleTimeMs", + "taskmanager.job.task.operator.split.watermark")) + .isFalse(); + assertThat(metricFilter.filter(COUNTER, "numRecordsIn", "taskmanager.job.task.operator")) + .isTrue(); + assertThat(metricFilter.filter(GAUGE, "currentInputWatermark", "taskmanager.job.task")) + .isTrue(); + } + + @Test + void testQueryServiceFilterFromConfiguration() { + Configuration configuration = new Configuration(); + configuration.set( + MetricOptions.QUERY_SERVICE_EXCLUDES, + Arrays.asList("*:excluded*", "*.custom:*:gauge")); + + final ReporterFilter metricFilter = + DefaultReporterFilters.queryServiceMetricsFromConfiguration(configuration); + + assertThat(metricFilter.filter(COUNTER, "excludedCounter", "taskmanager")).isFalse(); + assertThat(metricFilter.filter(COUNTER, "included", "taskmanager")).isTrue(); + assertThat(metricFilter.filter(GAUGE, "included", "taskmanager.custom")).isFalse(); + assertThat(metricFilter.filter(COUNTER, "included", "taskmanager.custom")).isTrue(); + // the default exclusion is replaced, not extended + assertThat( + metricFilter.filter( + GAUGE, + "currentWatermark", + "taskmanager.job.task.operator.split.watermark")) + .isTrue(); + } + + @Test + void testQueryServiceFilterWithoutExcludesExposesSplitWatermarkMetrics() { + final Configuration listForm = new Configuration(); + listForm.set(MetricOptions.QUERY_SERVICE_EXCLUDES, Collections.emptyList()); + + final Configuration flatStringForm = new Configuration(); + flatStringForm.setString(MetricOptions.QUERY_SERVICE_EXCLUDES.key(), ""); + + for (Configuration configuration : Arrays.asList(listForm, flatStringForm)) { + final ReporterFilter metricFilter = + DefaultReporterFilters.queryServiceMetricsFromConfiguration(configuration); + + assertThat( + metricFilter.filter( + GAUGE, + "currentWatermark", + "taskmanager.job.task.operator.split.watermark")) + .isTrue(); + assertThat(metricFilter.filter(COUNTER, "numRecordsIn", "taskmanager.job.task")) + .isTrue(); + } + } + @Test void testFromConfigurationIncludeByScope() { Configuration configuration = new Configuration();