gaborgsomogyi commented on code in PR #28427:
URL: https://github.com/apache/flink/pull/28427#discussion_r3734252292
##########
flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java:
##########
@@ -316,6 +318,53 @@ public static List<FileSystemFactory>
getRegisteredFileSystemFactories() {
}
}
+ /**
+ * Hands a runtime-owned, process-level {@link MetricGroup} to every
registered {@link
+ * FileSystemFactory} that opts into metrics via {@link MetricsAware}.
+ *
+ * <p>This is the second phase of file system initialization. {@link
#initialize(Configuration,
+ * PluginManager)} runs at process startup, before the {@code
MetricRegistry} exists; this
+ * method is therefore invoked separately, once the registry and a
process-level {@link
+ * MetricGroup} are available. It is called from the TaskManager and
JobManager entrypoints
+ * only. Contexts without a process-level {@link MetricGroup} (CLI,
HistoryServer, YARN client)
+ * simply never call it, and their file system plugins continue to operate
without emitting
+ * metrics.
+ *
+ * <p>The call is idempotent: factories receive a child group {@code
<process>.filesystem}, and
+ * {@link MetricGroup#addGroup} returns the same child on repeated calls
with the same parent,
+ * so re-invocation does not register duplicate metrics. Factories that do
not implement {@link
+ * MetricsAware} are skipped.
+ *
+ * @param processMetricGroup the process-level metric group to register
file system metrics
+ * under.
+ */
+ @Internal
+ public static void attachMetrics(MetricGroup processMetricGroup) {
+ checkNotNull(processMetricGroup, "processMetricGroup");
+ LOCK.lock();
+ try {
+ final MetricGroup fsGroup =
processMetricGroup.addGroup("filesystem");
+ for (FileSystemFactory factory : FS_FACTORIES.values()) {
+ // Plugin-loaded factories are wrapped in a
PluginFileSystemFactory, which is itself
+ // MetricsAware and forwards setMetricGroup to the inner
factory under the plugin
+ // classloader, so this plain instanceof reaches both wrapped
and direct factories.
+ if (factory instanceof MetricsAware) {
+ try {
+ ((MetricsAware) factory).setMetricGroup(fsGroup);
+ } catch (Exception e) {
+ // A misbehaving plugin must never break process
startup.
+ LOG.warn(
+ "Failed to attach metrics to file system
factory {}",
+ factory.getClass().getName(),
+ e);
+ }
Review Comment:
If it's `MetricsAware` instance then why do we treat exception as acceptable
warning? I mean if the user configured metrics then it's a fair expectation to
work, right?
##########
flink-core/src/main/java/org/apache/flink/core/fs/metrics/FileSystemMetricOptions.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs.metrics;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.metrics.SlidingWindowHistogram;
+import org.apache.flink.util.Preconditions;
+
+import java.util.List;
+
+/** Creates filesystem metric options under a cloud-specific configuration
prefix. */
+@Internal
+public final class FileSystemMetricOptions {
+
+ private FileSystemMetricOptions() {}
+
+ public static ConfigOption<Boolean> metricsEnabled(String
configurationPrefix) {
+ final String prefix = validatePrefix(configurationPrefix);
+ return ConfigOptions.key(prefix + ".metrics.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Master switch for publishing "
+ + prefix
+ + " filesystem operation metrics to Flink's
metric system.");
+ }
+
+ public static ConfigOption<List<String>> metricsAllowlist(String
configurationPrefix) {
+ final String prefix = validatePrefix(configurationPrefix);
+ return ConfigOptions.key(prefix + ".metrics.allowlist")
+ .stringType()
+ .asList()
+
.defaultValues(FileSystemMetricRecorder.DEFAULT_ALLOWLIST.toArray(new
String[0]))
+ .withDescription(
+ "Names of "
+ + prefix
+ + " filesystem metrics to register. Replaces
the default list; use "
+ + "\"*\" to register every emitted metric. An
empty list is invalid. "
+ + "The iops metric is derived from
api_call_count.");
Review Comment:
I've the feeling that the last sentence is S3 native specific, right?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java:
##########
@@ -682,6 +692,13 @@ S3ClientProvider build() {
: "(CRT runtime default)");
}
+ if (metricPublisher != null) {
+ // Re-wrap the immutable override config with the publisher
attached. The same
+ // publisher feeds both the sync and async clients below.
+ overrideConfig =
+
overrideConfig.toBuilder().addMetricPublisher(metricPublisher).build();
+ }
Review Comment:
More like a nit but I think nothing blocks us to build `overrideConfig` only
once, right?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java:
##########
@@ -321,9 +325,53 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
+ "When not set, the default chain is
used: delegation tokens -> "
+ "static credentials (if configured) ->
DefaultCredentialsProvider.");
+ public static final ConfigOption<Boolean> METRICS_ENABLED =
+ ConfigOptions.key("s3.metrics.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Master switch for publishing S3 operation metrics
to Flink's metric "
+ + "system. When false, no metric publisher
is attached to the SDK "
+ + "and no metric is registered. Metrics
are only emitted under the "
+ + "TaskManager and JobManager entrypoints,
which provide a "
+ + "process-level metric group; other
contexts (CLI, etc.) emit "
+ + "none regardless of this setting.");
+
+ public static final ConfigOption<List<String>> METRICS_ALLOWLIST =
+ ConfigOptions.key("s3.metrics.allowlist")
+ .stringType()
+ .asList()
+ .defaultValues(
+ "api_call_count",
+ "api_call_duration_ms",
+ "throttle_count",
+ "retry_count",
+ "iops")
Review Comment:
Having a `*` is fine, the main idea is not to type all when everything is
needed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java:
##########
@@ -640,6 +640,11 @@ public static TaskExecutor startTaskManager(
resourceID,
taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+ // Second-phase init for file system plugins that opt into metrics
(e.g.
+ // flink-s3-fs-native): hand them the process-level metric group now
that the
+ // MetricRegistry exists. See FileSystem#attachMetrics and
MetricsAware.
+ FileSystem.attachMetrics(taskManagerMetricGroup.f0);
Review Comment:
Makes sense, thanks for clarifying. Confirming what we landed on:
`MetricsAware` is a reusable contract, but the setter/two-phase pattern is
specifically needed for filesystems, not a general requirement. Two reasons:
`FileSystemFactory.create(URI)` is `@PublicEvolving` with many external
implementors, so adding a `MetricGroup` parameter would break the public API;
and `FileSystem.CACHE` can already hold instances created before any
`MetricGroup` exists, so parameter injection could not retrofit them anyway.
Other plugin families without that constraint (e.g.
`StateChangelogStorageFactory`) should keep using plain parameter passing. Good
with this approach for filesystems.
##########
flink-core/src/main/java/org/apache/flink/core/fs/metrics/FileSystemMetricRecorder.java:
##########
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.fs.metrics;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SlidingWindowHistogram;
+import org.apache.flink.metrics.ThreadSafeSimpleCounter;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** Registers normalized filesystem operation metrics supplied by
cloud-specific adapters. */
+@Internal
+public final class FileSystemMetricRecorder {
Review Comment:
I think there is a global design issue here. This code part is in core and
assumes that all FS plugins are having the exact same metrics set. This will
never be true. Just to give an example this maps HTTP status codes which
assumes all filesystems are HTTP based. Instead a more dynamic approach need to
be chosen which is not doing any prediction how a filesystem works.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
