[FLINK-41054][metrics] Add exclude filter for the metric query service

docs/content/docs/ops/metrics.md

@@ -1963,6 +1963,9 @@ Note that the metrics are only available via reporters.
       <td>watermark.currentWatermark</td>
       <td>
         The last watermark this split has received (in milliseconds).
+        <p><strong>Note:</strong> 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 <code>metrics.internal.query-service.filter.excludes</code> to change this.</p>
       </td>
       <td>Gauge</td>
     </tr>

docs/layouts/shortcodes/generated/metric_configuration.html

@@ -14,6 +14,12 @@
             <td>Duration</td>
             <td>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.</td>
         </tr>
+        <tr>
+            <td><h5>metrics.internal.query-service.filter.excludes</h5></td>
+            <td style="word-wrap: break-word;">"*.operator.split.watermark:*:*"</td>
+            <td>List&lt;String&gt;</td>
+            <td>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 <code class="highlighter-rouge">metrics.reporter.&lt;name&gt;.filter.excludes</code>.<br />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.</td>
+        </tr>
         <tr>
             <td><h5>metrics.internal.query-service.port</h5></td>
             <td style="word-wrap: break-word;">"0"</td>

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<List<String>> 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.<name>.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.

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<Metric> queryServiceFilter;
+
     public MetricRegistryConfiguration(
-            ScopeFormats scopeFormats, char delimiter, long queryServiceMessageSizeLimit) {
+            ScopeFormats scopeFormats,
+            char delimiter,
+            long queryServiceMessageSizeLimit,
+            ReporterFilter<Metric> 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<Metric> getQueryServiceFilter() {
+        return queryServiceFilter;
+    }
+
     // ------------------------------------------------------------------------
     //  Static factory methods
     // ------------------------------------------------------------------------
@@ -92,10 +106,30 @@ public static MetricRegistryConfiguration fromConfiguration(
             delim = '.';
         }
 
+        ReporterFilter<Metric> 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);
     }
 }

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<Metric> 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);

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<Metric> 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<Metric> queryServiceMetricsFromConfiguration(
+            Configuration configuration) {
+        final List<FilterSpec> excludeFilters =
+                configuration.get(MetricOptions.QUERY_SERVICE_EXCLUDES).stream()
+                        .map(DefaultReporterFilters::parse)
+                        .collect(Collectors.toList());
+
+        return new MetricReporterFilter(Collections.singletonList(parse("*:*:*")), excludeFilters);
+    }
+
     public static ReporterFilter<SpanBuilder> tracesFromConfiguration(Configuration configuration) {
         return fromConfiguration(
                 configuration,

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<String> queryMetricNames(
+            MetricRegistryImpl registry, String awaitedMetricName) throws Exception {
+        final MetricQueryServiceGateway gateway =
+                checkNotNull(registry.getMetricQueryServiceGateway());
+        final MetricDumpSerialization.MetricDumpDeserializer deserializer =
+                new MetricDumpSerialization.MetricDumpDeserializer();
+
+        List<String> 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";