This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new df122f59096 [FLINK-39546][s3] Improve observability in
flink-s3-fs-native by exposing operation-level S3 metrics
df122f59096 is described below
commit df122f590962d55917b0f9033f587467f8596490
Author: Samrat <[email protected]>
AuthorDate: Wed Aug 19 18:14:56 2026 +0530
[FLINK-39546][s3] Improve observability in flink-s3-fs-native by exposing
operation-level S3 metrics
---
docs/content.zh/docs/deployment/filesystems/s3.md | 6 +
docs/content/docs/deployment/filesystems/s3.md | 6 +
.../flink/core/fs/ConnectionLimitingFactory.java | 14 +-
.../java/org/apache/flink/core/fs/FileSystem.java | 41 ++++
.../flink/core/fs/PluginFileSystemFactory.java | 18 +-
.../org/apache/flink/core/plugin/MetricsAware.java | 47 +++++
.../flink/core/fs/FileSystemAttachMetricsTest.java | 231 +++++++++++++++++++++
flink-filesystems/flink-s3-fs-native/README.md | 20 ++
.../fs/s3native/NativeS3FileSystemFactory.java | 83 +++++++-
.../apache/flink/fs/s3native/S3ClientProvider.java | 20 +-
.../fs/s3native/metrics/AwsSdkMetricBridge.java | 165 +++++++++++++++
.../fs/s3native/metrics/S3MetricRecorder.java | 177 ++++++++++++++++
.../fs/s3native/NativeS3FileSystemFactoryTest.java | 44 ++++
.../s3native/metrics/AwsSdkMetricBridgeTest.java | 187 +++++++++++++++++
.../NativeS3FileSystemFactoryMetricsTest.java | 95 +++++++++
.../metrics/NativeS3MetricsEmissionITCase.java | 218 +++++++++++++++++++
.../fs/s3native/metrics/S3MetricRecorderTest.java | 120 +++++++++++
.../flink/metrics/SlidingWindowHistogram.java | 151 ++++++++++++++
.../flink/metrics/SlidingWindowHistogramTest.java | 73 +++++++
.../runtime/entrypoint/ClusterEntrypoint.java | 2 +
.../runtime/taskexecutor/TaskManagerRunner.java | 3 +
21 files changed, 1715 insertions(+), 6 deletions(-)
diff --git a/docs/content.zh/docs/deployment/filesystems/s3.md
b/docs/content.zh/docs/deployment/filesystems/s3.md
index 44e6dde5c58..f8f7bad7a8e 100644
--- a/docs/content.zh/docs/deployment/filesystems/s3.md
+++ b/docs/content.zh/docs/deployment/filesystems/s3.md
@@ -172,6 +172,12 @@ In addition to the [common
configuration](#common-configuration) options (`s3.ac
View the detailed configuration at
[native-s3-fs](https://github.com/apache/flink/tree/master/flink-filesystems/flink-s3-fs-native#configuration-options)
+#### Metrics
+
+The Native S3 FileSystem can publish AWS SDK operation metrics into Flink's
process-level metric group on the JobManager and TaskManager. Metrics are
enabled by default with `s3.metrics.enabled: true` and are scoped under
`filesystem.filesystem_type.s3` or `filesystem.filesystem_type.s3a`.
+
+The default metric set includes `api_call_count`, `api_call_duration_ms`,
`throttle_count`, `retry_count`, and reporter-derived `iops`. Use
`s3.metrics.allowlist` to restrict the registered metrics, or set
`s3.metrics.enabled: false` to disable S3 operation metrics. Empty allowlists
are rejected.
+
---
### Presto S3 FileSystem
diff --git a/docs/content/docs/deployment/filesystems/s3.md
b/docs/content/docs/deployment/filesystems/s3.md
index a49ac14e678..36c2b55bad3 100644
--- a/docs/content/docs/deployment/filesystems/s3.md
+++ b/docs/content/docs/deployment/filesystems/s3.md
@@ -180,6 +180,12 @@ s3.region: us-east-1
s3.checksum-validation.enabled: false
```
+#### Metrics
+
+The Native S3 FileSystem can publish AWS SDK operation metrics into Flink's
process-level metric group on the JobManager and TaskManager. Metrics are
enabled by default with `s3.metrics.enabled: true` and are scoped under
`filesystem.filesystem_type.s3` or `filesystem.filesystem_type.s3a`.
+
+The default metric set includes `api_call_count`, `api_call_duration_ms`,
`throttle_count`, `retry_count`, and reporter-derived `iops`. Use
`s3.metrics.allowlist` to restrict the registered metrics, or set
`s3.metrics.enabled: false` to disable S3 operation metrics. Empty allowlists
are rejected.
+
---
### Presto S3 FileSystem
diff --git
a/flink-core/src/main/java/org/apache/flink/core/fs/ConnectionLimitingFactory.java
b/flink-core/src/main/java/org/apache/flink/core/fs/ConnectionLimitingFactory.java
index 7fdb2dac4f2..1d57aeea387 100644
---
a/flink-core/src/main/java/org/apache/flink/core/fs/ConnectionLimitingFactory.java
+++
b/flink-core/src/main/java/org/apache/flink/core/fs/ConnectionLimitingFactory.java
@@ -21,6 +21,8 @@ package org.apache.flink.core.fs;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
import
org.apache.flink.core.fs.LimitedConnectionsFileSystem.ConnectionLimitingSettings;
+import org.apache.flink.core.plugin.MetricsAware;
+import org.apache.flink.metrics.MetricGroup;
import java.io.IOException;
import java.net.URI;
@@ -29,7 +31,7 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
/** A wrapping factory that adds a {@link LimitedConnectionsFileSystem} to a
file system. */
@Internal
-public class ConnectionLimitingFactory implements FileSystemFactory {
+public class ConnectionLimitingFactory implements FileSystemFactory,
MetricsAware {
private final FileSystemFactory factory;
@@ -59,6 +61,16 @@ public class ConnectionLimitingFactory implements
FileSystemFactory {
factory.configure(config);
}
+ /**
+ * Forwards metric registration to the wrapped factory when it supports
{@link MetricsAware}.
+ */
+ @Override
+ public void setMetricGroup(MetricGroup metricGroup) {
+ if (factory instanceof MetricsAware) {
+ ((MetricsAware) factory).setMetricGroup(metricGroup);
+ }
+ }
+
@Override
public FileSystem create(URI fsUri) throws IOException {
FileSystem original = factory.create(fsUri);
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
index 3699b2a2962..ba6a0a9d953 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
@@ -32,7 +32,9 @@ import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.core.fs.local.LocalFileSystemFactory;
+import org.apache.flink.core.plugin.MetricsAware;
import org.apache.flink.core.plugin.PluginManager;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.TemporaryClassLoaderContext;
import org.apache.flink.util.WrappingProxy;
@@ -316,6 +318,45 @@ public abstract class FileSystem implements IFileSystem {
}
}
+ /**
+ * 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) {
+ ((MetricsAware) factory).setMetricGroup(fsGroup);
+ }
+ }
+ } finally {
+ LOCK.unlock();
+ }
+ }
+
/**
* Initializes the shared file system settings.
*
diff --git
a/flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java
b/flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java
index e7eb5892363..c2ff0d5e893 100644
---
a/flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java
+++
b/flink-core/src/main/java/org/apache/flink/core/fs/PluginFileSystemFactory.java
@@ -18,6 +18,8 @@
package org.apache.flink.core.fs;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.plugin.MetricsAware;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.util.TemporaryClassLoaderContext;
import org.apache.flink.util.WrappingProxy;
@@ -30,7 +32,7 @@ import java.util.List;
* {@link FileSystem} operations.
*/
public class PluginFileSystemFactory
- implements FileSystemFactory, WrappingProxy<FileSystemFactory> {
+ implements FileSystemFactory, WrappingProxy<FileSystemFactory>,
MetricsAware {
private final FileSystemFactory inner;
private final ClassLoader loader;
@@ -58,6 +60,20 @@ public class PluginFileSystemFactory
inner.configure(config);
}
+ /**
+ * Forwards the metric group to the wrapped factory if it opts into
metrics, using the plugin
+ * classloader so the factory observes the same isolation as every other
call routed through
+ * this wrapper.
+ */
+ @Override
+ public void setMetricGroup(final MetricGroup metricGroup) {
+ if (inner instanceof MetricsAware) {
+ try (TemporaryClassLoaderContext ignored =
TemporaryClassLoaderContext.of(loader)) {
+ ((MetricsAware) inner).setMetricGroup(metricGroup);
+ }
+ }
+ }
+
@Override
public FileSystem create(final URI fsUri) throws IOException {
try (TemporaryClassLoaderContext ignored =
TemporaryClassLoaderContext.of(loader)) {
diff --git
a/flink-core/src/main/java/org/apache/flink/core/plugin/MetricsAware.java
b/flink-core/src/main/java/org/apache/flink/core/plugin/MetricsAware.java
new file mode 100644
index 00000000000..5a2c21bcfe0
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/plugin/MetricsAware.java
@@ -0,0 +1,47 @@
+/*
+ * 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.plugin;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.metrics.MetricGroup;
+
+/**
+ * Opt-in contract for {@link Plugin}s that need a runtime-owned {@link
MetricGroup}.
+ *
+ * <p>The runtime calls {@link #setMetricGroup(MetricGroup)} after {@link
+ * Plugin#configure(org.apache.flink.configuration.Configuration)} and before
the plugin emits
+ * metrics.
+ *
+ * <p>{@code setMetricGroup} may be called more than once, for example when an
embedded runtime is
+ * restarted in the same JVM while retaining plugin instances. Re-applying the
same {@link
+ * MetricGroup} must be a no-op, and a different group must scope metrics
created afterwards. The
+ * runtime owns the group; implementations may call {@link
MetricGroup#addGroup} to derive nested
+ * scopes from it.
+ */
+@PublicEvolving
+public interface MetricsAware {
+
+ /**
+ * Hands the plugin a runtime-owned {@link MetricGroup} to register its
metrics against. See the
+ * class-level two-phase init contract.
+ *
+ * @param metricGroup the group to register metrics under; never {@code
null}.
+ */
+ void setMetricGroup(MetricGroup metricGroup);
+}
diff --git
a/flink-core/src/test/java/org/apache/flink/core/fs/FileSystemAttachMetricsTest.java
b/flink-core/src/test/java/org/apache/flink/core/fs/FileSystemAttachMetricsTest.java
new file mode 100644
index 00000000000..fbe91609b8f
--- /dev/null
+++
b/flink-core/src/test/java/org/apache/flink/core/fs/FileSystemAttachMetricsTest.java
@@ -0,0 +1,231 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.core.plugin.MetricsAware;
+import org.apache.flink.core.plugin.TestingPluginManager;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link FileSystem#attachMetrics(MetricGroup)}. */
+class FileSystemAttachMetricsTest {
+
+ @AfterEach
+ void resetFileSystems() {
+ // Restore the default, plugin-less factory registry so other tests
are unaffected.
+ FileSystem.initialize(new Configuration(), null);
+ }
+
+ @Test
+ void attachMetricsReachesPluginLoadedMetricsAwareFactory() {
+ // Headline case: plugin file systems are registered behind wrappers
such as
+ // PluginFileSystemFactory. attachMetrics must still reach the real
factory, otherwise the
+ // metric group is silently never delivered and no metrics are ever
emitted.
+ TestingFileSystemFactory factory = new
TestingFileSystemFactory("metrics-test-fs");
+ initializeWithPlugins(factory);
+
+ RecordingMetricGroup processGroup = new RecordingMetricGroup();
+ FileSystem.attachMetrics(processGroup);
+
+ // Unwrapped through PluginFileSystemFactory and invoked exactly once.
+ assertThat(factory.setMetricGroupCalls).hasValue(1);
+ // The group handed to the factory is the "filesystem" child of the
process group, not the
+ // process group itself.
+ assertThat(processGroup.childGroupNames).containsExactly("filesystem");
+
assertThat(factory.receivedGroup.get()).isNotNull().isNotSameAs(processGroup);
+ }
+
+ @Test
+ void attachMetricsReachesMetricsAwareFactoryBehindConnectionLimiter() {
+ TestingFileSystemFactory factory = new
TestingFileSystemFactory("limited-test-fs");
+ Configuration config = new Configuration();
+ config.set(CoreOptions.fileSystemConnectionLimit(factory.getScheme()),
1);
+ initializeWithPlugins(config, factory);
+
+ FileSystem.attachMetrics(new UnregisteredMetricsGroup());
+
+ assertThat(factory.setMetricGroupCalls).hasValue(1);
+ }
+
+ @Test
+ void attachMetricsSkipsFactoriesThatAreNotMetricsAware() {
+ // A factory that does not implement MetricsAware is filtered out by
the instanceof check in
+ // attachMetrics, so it simply receives no group. One representative
factory is enough.
+ initializeWithPlugins(new PlainFileSystemFactory("plain-test-fs"));
+
+ assertThatCode(() -> FileSystem.attachMetrics(new
UnregisteredMetricsGroup()))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void attachMetricsPropagatesFactoryFailure() {
+ initializeWithPlugins(
+ new TestingFileSystemFactory(
+ "throwing-test-fs",
+ group -> {
+ throw new RuntimeException(
+ "intentional failure from a misbehaving
plugin");
+ }));
+
+ assertThatThrownBy(() -> FileSystem.attachMetrics(new
UnregisteredMetricsGroup()))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessage("intentional failure from a misbehaving plugin");
+ }
+
+ @Test
+ void attachMetricsForwardsToFactoryOnEveryInvocation() {
+ TestingFileSystemFactory factory = new
TestingFileSystemFactory("idem-test-fs");
+ initializeWithPlugins(factory);
+
+ MetricGroup group = new UnregisteredMetricsGroup();
+ FileSystem.attachMetrics(group);
+ FileSystem.attachMetrics(group);
+
+ // The hook forwards on every call; collapsing duplicate registrations
is the factory's
+ // responsibility, verified end-to-end in
NativeS3FileSystemFactoryMetricsTest
+ // #repeatedAttachmentWithSameGroupDoesNotCreateNewFilesystemTypeGroup.
+ assertThat(factory.setMetricGroupCalls).hasValue(2);
+ }
+
+ @Test
+ void setMetricGroupIsNotInvokedWhenAttachMetricsIsNeverCalled() {
+ TestingFileSystemFactory factory = new
TestingFileSystemFactory("never-test-fs");
+ initializeWithPlugins(factory);
+
+ assertThat(factory.setMetricGroupCalls).hasValue(0);
+ }
+
+ @Test
+ void pluginFileSystemFactoryForwardsMetricGroupToInner() {
+ TestingFileSystemFactory inner = new
TestingFileSystemFactory("wrapped-fs");
+ FileSystemFactory wrapper = PluginFileSystemFactory.of(inner);
+
+ // The wrapper must itself be MetricsAware so attachMetrics reaches it
without unwrapping.
+ assertThat(wrapper).isInstanceOf(MetricsAware.class);
+
+ MetricGroup group = new UnregisteredMetricsGroup();
+ ((MetricsAware) wrapper).setMetricGroup(group);
+
+ assertThat(inner.setMetricGroupCalls).hasValue(1);
+ assertThat(inner.receivedGroup.get()).isSameAs(group);
+ }
+
+ private static void initializeWithPlugins(FileSystemFactory... factories) {
+ initializeWithPlugins(new Configuration(), factories);
+ }
+
+ private static void initializeWithPlugins(
+ Configuration config, FileSystemFactory... factories) {
+ Map<Class<?>, Iterator<?>> plugins = new HashMap<>();
+ plugins.put(FileSystemFactory.class,
Arrays.asList(factories).iterator());
+ FileSystem.initialize(config, new TestingPluginManager(plugins));
+ }
+
+ // ------------------------------------------------------------------------
+ // test factories
+ // ------------------------------------------------------------------------
+
+ private static class PlainFileSystemFactory implements FileSystemFactory {
+ private final String scheme;
+
+ PlainFileSystemFactory(String scheme) {
+ this.scheme = scheme;
+ }
+
+ @Override
+ public void configure(Configuration config) {}
+
+ @Override
+ public String getScheme() {
+ return scheme;
+ }
+
+ @Override
+ public FileSystem create(URI fsUri) throws IOException {
+ throw new UnsupportedOperationException(
+ "This test factory does not create file systems.");
+ }
+ }
+
+ private static class TestingFileSystemFactory extends
PlainFileSystemFactory
+ implements MetricsAware {
+ final AtomicInteger setMetricGroupCalls = new AtomicInteger();
+ final AtomicReference<MetricGroup> receivedGroup = new
AtomicReference<>();
+ private final Consumer<MetricGroup> onSetMetricGroup;
+
+ TestingFileSystemFactory(String scheme) {
+ this(scheme, metricGroup -> {});
+ }
+
+ TestingFileSystemFactory(String scheme, Consumer<MetricGroup>
onSetMetricGroup) {
+ super(scheme);
+ this.onSetMetricGroup = onSetMetricGroup;
+ }
+
+ @Override
+ public void setMetricGroup(MetricGroup metricGroup) {
+ setMetricGroupCalls.incrementAndGet();
+ receivedGroup.set(metricGroup);
+ onSetMetricGroup.accept(metricGroup);
+ }
+ }
+
+ /**
+ * Records the names of child groups created directly under it, sharing
the list with children.
+ */
+ private static class RecordingMetricGroup extends UnregisteredMetricsGroup
{
+ final List<String> childGroupNames;
+
+ RecordingMetricGroup() {
+ this(Collections.synchronizedList(new ArrayList<>()));
+ }
+
+ private RecordingMetricGroup(List<String> childGroupNames) {
+ this.childGroupNames = childGroupNames;
+ }
+
+ @Override
+ public MetricGroup addGroup(String name) {
+ childGroupNames.add(name);
+ return new RecordingMetricGroup(childGroupNames);
+ }
+ }
+}
diff --git a/flink-filesystems/flink-s3-fs-native/README.md
b/flink-filesystems/flink-s3-fs-native/README.md
index d8be409626b..b65583f989d 100644
--- a/flink-filesystems/flink-s3-fs-native/README.md
+++ b/flink-filesystems/flink-s3-fs-native/README.md
@@ -78,6 +78,26 @@ input.sinkTo(FileSink.forRowFormat(new
Path("s3://my-bucket/output"),
| s3.async.enabled | true | Enable async read/write with TransferManager |
| s3.read.buffer.size | 262144 (256KB) | Read buffer size per stream (64KB -
4MB) |
+### Metrics
+
+When the native S3 plugin is loaded in a JobManager or TaskManager, it can
publish AWS SDK operation metrics into Flink's process-level metric group.
Metrics are scoped under `filesystem.filesystem_type.<scheme>`, where
`<scheme>` is `s3` or `s3a`.
+
+| Key | Default | Description |
+|-----|---------|-------------|
+| s3.metrics.enabled | true | Enable S3 operation metrics. Set to `false` to
avoid attaching the AWS SDK metric publisher |
+| s3.metrics.allowlist | `api_call_count`, `api_call_duration_ms`,
`throttle_count`, `retry_count`, `iops` | Metrics to register. Use `*` to
register every metric emitted by the plugin. Empty lists are rejected; set
`s3.metrics.enabled: false` to disable metrics. `iops` is derived by reporters
from `api_call_count`, so allowing `iops` also registers `api_call_count` |
+| s3.metrics.histogram.window-size | 1024 | Number of recent samples retained
per `api_call_duration_ms` histogram |
+
+The plugin emits the following metric names:
+
+| Metric | Type | Labels | Description |
+|--------|------|--------|-------------|
+| api_call_count | Counter | `op`, `status_class` | Number of completed S3 API
calls, grouped by operation and result class |
+| api_call_duration_ms | Histogram | `op` | Completed S3 API call latency in
milliseconds |
+| throttle_count | Counter | `op` | Number of throttled S3 responses (`429` or
`503`) |
+| retry_count | Counter | `op`, `reason` | Number of AWS SDK retries, grouped
by retry reason |
+| iops | Derived rate | `op`, `status_class` | Reporter-side rate derived from
`api_call_count` |
+
### Credentials Provider
| Key | Default | Description |
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
index 23d6c22314c..ff52b2dfb0a 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
@@ -27,6 +27,11 @@ import
org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystemFactory;
+import org.apache.flink.core.plugin.MetricsAware;
+import org.apache.flink.fs.s3native.metrics.AwsSdkMetricBridge;
+import org.apache.flink.fs.s3native.metrics.S3MetricRecorder;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SlidingWindowHistogram;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StringUtils;
@@ -34,11 +39,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
import java.io.IOException;
import java.net.URI;
import java.time.Duration;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
/**
@@ -55,10 +62,11 @@ import java.util.Map;
* @see org.apache.flink.core.fs.FileSystemFactory
*/
@Experimental
-public class NativeS3FileSystemFactory implements FileSystemFactory {
+public class NativeS3FileSystemFactory implements FileSystemFactory,
MetricsAware {
private static final Logger LOG =
LoggerFactory.getLogger(NativeS3FileSystemFactory.class);
+ private static final String CONFIGURATION_PREFIX = "s3";
private static final String INVALID_ENTROPY_KEY_CHARS =
"^.*[~#@*+%{}<>\\[\\]|\"\\\\].*$";
public static final long S3_MULTIPART_MIN_PART_SIZE = 5L << 20;
@@ -410,9 +418,47 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
+ "size here causes 'failed to acquire a
connection' timeouts "
+ "under parallel checkpoint
upload/restore. Defaults to 256.");
+ public static final ConfigOption<Boolean> METRICS_ENABLED =
+ ConfigOptions.key(CONFIGURATION_PREFIX + ".metrics.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Master switch for publishing S3 operation metrics
to Flink's metric "
+ + "system.");
+
+ public static final ConfigOption<List<String>> METRICS_ALLOWLIST =
+ ConfigOptions.key(CONFIGURATION_PREFIX + ".metrics.allowlist")
+ .stringType()
+ .asList()
+
.defaultValues(S3MetricRecorder.DEFAULT_ALLOWLIST.toArray(new String[0]))
+ .withDescription(
+ "Names of S3 metrics to register. Replaces the
default list; use \"*\" "
+ + "to register every metric emitted by the
plugin. An empty list "
+ + "is invalid. The iops metric is derived
from api_call_count.");
+
+ public static final ConfigOption<Integer> METRICS_HISTOGRAM_WINDOW_SIZE =
+ ConfigOptions.key(CONFIGURATION_PREFIX +
".metrics.histogram.window-size")
+ .intType()
+ .defaultValue(SlidingWindowHistogram.DEFAULT_WINDOW_SIZE)
+ .withDescription(
+ "Number of recent values retained by S3 latency
histograms. Must be "
+ + "positive.");
+
@Nullable private Configuration flinkConfig;
@Nullable private BucketConfigProvider bucketConfigProvider;
+ @GuardedBy("this")
+ @Nullable
+ private MetricGroup pluginMetrics;
+
+ @GuardedBy("this")
+ @Nullable
+ private MetricGroup attachedMetricGroup;
+
+ @GuardedBy("this")
+ @Nullable
+ private AwsSdkMetricBridge metricBridge;
+
@Override
public String getScheme() {
return "s3";
@@ -430,6 +476,40 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
this.bucketConfigProvider = new BucketConfigProvider(config);
}
+ @Override
+ public synchronized void setMetricGroup(MetricGroup metricGroup) {
+ if (metricGroup == attachedMetricGroup) {
+ return;
+ }
+ this.attachedMetricGroup = metricGroup;
+ this.pluginMetrics = metricGroup.addGroup("filesystem_type",
getScheme());
+ if (metricBridge != null) {
+ metricBridge.setMetricGroup(pluginMetrics);
+ }
+ }
+
+ /**
+ * Returns the stable SDK metric publisher shared by all clients, or
{@code null} when metrics
+ * are disabled. The publisher must be installed even before a metric
group is available because
+ * runtime startup may cache a file system before attaching metrics.
+ */
+ @Nullable
+ private synchronized AwsSdkMetricBridge resolveMetricBridge(Configuration
config) {
+ if (!config.get(METRICS_ENABLED)) {
+ return null;
+ }
+ if (metricBridge == null) {
+ metricBridge =
+ new AwsSdkMetricBridge(
+ config.get(METRICS_ALLOWLIST),
+ config.get(METRICS_HISTOGRAM_WINDOW_SIZE));
+ if (pluginMetrics != null) {
+ metricBridge.setMetricGroup(pluginMetrics);
+ }
+ }
+ return metricBridge;
+ }
+
@Override
public FileSystem create(URI fsUri) throws IOException {
Configuration config = this.flinkConfig;
@@ -632,6 +712,7 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
.retryCircuitBreakerEnabled(config.get(RETRY_CIRCUIT_BREAKER_ENABLED))
.credentialsProviderClasses(credentialsProviderClasses)
.encryptionConfig(encryptionConfig)
+ .metricPublisher(resolveMetricBridge(config))
.useCrt(crtEnabled);
if (crtEnabled) {
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
index aa31db64696..634409da4d3 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
@@ -36,6 +36,7 @@ import
software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
import software.amazon.awssdk.http.apache.ApacheHttpClient;
import software.amazon.awssdk.http.crt.AwsCrtHttpClient;
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import software.amazon.awssdk.metrics.MetricPublisher;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain;
import software.amazon.awssdk.retries.StandardRetryStrategy;
@@ -464,6 +465,10 @@ class S3ClientProvider implements AutoCloseableAsync {
private long crtMinPartSizeInBytes =
NativeS3FileSystemFactory.PART_UPLOAD_MIN_SIZE.defaultValue();
+ // Optional AWS SDK metric publisher (e.g. the Flink metric bridge).
Attached to both the
+ // sync and async clients via the shared ClientOverrideConfiguration.
Null = no metrics.
+ @Nullable private MetricPublisher metricPublisher;
+
public Builder accessKey(@Nullable String accessKey) {
this.accessKey = accessKey;
return this;
@@ -633,6 +638,11 @@ class S3ClientProvider implements AutoCloseableAsync {
return this;
}
+ public Builder metricPublisher(@Nullable MetricPublisher
metricPublisher) {
+ this.metricPublisher = metricPublisher;
+ return this;
+ }
+
S3ClientProvider build() {
if (endpoint == null) {
endpoint = System.getProperty("s3.endpoint");
@@ -674,7 +684,7 @@ class S3ClientProvider implements AutoCloseableAsync {
retryMaxBackoff,
retryThrottleBaseDelay);
- ClientOverrideConfiguration overrideConfig =
+ ClientOverrideConfiguration.Builder overrideConfigBuilder =
ClientOverrideConfiguration.builder()
.retryStrategy(
StandardRetryStrategy.builder()
@@ -687,8 +697,7 @@ class S3ClientProvider implements AutoCloseableAsync {
retryThrottleBaseDelay,
retryMaxBackoff))
.circuitBreakerEnabled(retryCircuitBreakerEnabled)
- .build())
- .build();
+ .build());
if (useCrt) {
LOG.info(
@@ -698,6 +707,11 @@ class S3ClientProvider implements AutoCloseableAsync {
: "(CRT runtime default)");
}
+ if (metricPublisher != null) {
+ overrideConfigBuilder.addMetricPublisher(metricPublisher);
+ }
+ ClientOverrideConfiguration overrideConfig =
overrideConfigBuilder.build();
+
S3Client s3Client =
buildSyncClient(
credentialsProvider, awsRegion, s3Config,
overrideConfig, endpointUri);
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java
new file mode 100644
index 00000000000..0f885944b17
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java
@@ -0,0 +1,165 @@
+/*
+ * 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.fs.s3native.metrics;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SlidingWindowHistogram;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.metrics.CoreMetric;
+import software.amazon.awssdk.http.HttpMetric;
+import software.amazon.awssdk.metrics.MetricCollection;
+import software.amazon.awssdk.metrics.MetricPublisher;
+import software.amazon.awssdk.metrics.SdkMetric;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+/** Maps AWS SDK metric records to native S3 filesystem metrics. */
+@Internal
+public final class AwsSdkMetricBridge implements MetricPublisher {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AwsSdkMetricBridge.class);
+ private static final String UNKNOWN_OPERATION = "Unknown";
+
+ @Nullable private final Collection<String> configuredAllowlist;
+ private final int histogramWindowSize;
+ private final AtomicReference<S3MetricRecorder> metrics = new
AtomicReference<>();
+
+ public AwsSdkMetricBridge(MetricGroup metricGroup) {
+ this(
+ metricGroup,
+ S3MetricRecorder.DEFAULT_ALLOWLIST,
+ SlidingWindowHistogram.DEFAULT_WINDOW_SIZE);
+ }
+
+ public AwsSdkMetricBridge(
+ MetricGroup metricGroup,
+ @Nullable Collection<String> allowlist,
+ int histogramWindowSize) {
+ this(allowlist, histogramWindowSize);
+ setMetricGroup(metricGroup);
+ }
+
+ public AwsSdkMetricBridge(@Nullable Collection<String> allowlist, int
histogramWindowSize) {
+ this.configuredAllowlist = allowlist == null ? null : new
ArrayList<>(allowlist);
+ this.histogramWindowSize = histogramWindowSize;
+ }
+
+ /** Starts publishing subsequent SDK metrics to the given Flink metric
group. */
+ public void setMetricGroup(MetricGroup metricGroup) {
+ metrics.set(new S3MetricRecorder(metricGroup, configuredAllowlist,
histogramWindowSize));
+ }
+
+ @Override
+ public void publish(MetricCollection apiCall) {
+ final S3MetricRecorder currentMetrics = metrics.get();
+ if (currentMetrics == null) {
+ return;
+ }
+ try {
+ translate(apiCall, currentMetrics);
+ } catch (Exception e) {
+ LOG.warn("Failed to publish S3 SDK metrics", e);
+ }
+ }
+
+ private void translate(MetricCollection apiCall, S3MetricRecorder
currentMetrics) {
+ final String operation = first(apiCall, CoreMetric.OPERATION_NAME,
UNKNOWN_OPERATION);
+ final Duration duration = first(apiCall, CoreMetric.API_CALL_DURATION,
null);
+ if (duration != null) {
+ currentMetrics.recordDuration(operation, duration.toMillis());
+ }
+
+ int throttleResponses = 0;
+ boolean sawServerError = false;
+ Integer lastStatus = null;
+ for (MetricCollection attempt : apiCall.children()) {
+ for (Integer status :
attempt.metricValues(HttpMetric.HTTP_STATUS_CODE)) {
+ if (status != null) {
+ lastStatus = status;
+ if (isThrottle(status)) {
+ throttleResponses++;
+ } else if (status >= 500) {
+ sawServerError = true;
+ }
+ }
+ }
+ }
+
+ currentMetrics.recordApiCall(
+ operation,
+ statusClass(lastStatus, first(apiCall,
CoreMetric.API_CALL_SUCCESSFUL, null)));
+ currentMetrics.recordThrottles(operation, throttleResponses);
+
+ final Integer retries = first(apiCall, CoreMetric.RETRY_COUNT, 0);
+ if (retries != null) {
+ currentMetrics.recordRetries(
+ operation, retries, retryReason(throttleResponses > 0,
sawServerError));
+ }
+ }
+
+ private static boolean isThrottle(int status) {
+ return status == 429 || status == 503;
+ }
+
+ private static String statusClass(Integer status, Boolean successful) {
+ if (status == null) {
+ if (Boolean.TRUE.equals(successful)) {
+ return "2xx";
+ }
+ return Boolean.FALSE.equals(successful) ? "error" : "unknown";
+ }
+ if (isThrottle(status)) {
+ return "throttled";
+ }
+ if (status >= 200 && status < 300) {
+ return "2xx";
+ }
+ if (status >= 400 && status < 500) {
+ return "4xx";
+ }
+ if (status >= 500) {
+ return "5xx";
+ }
+ return "other";
+ }
+
+ private static String retryReason(boolean sawThrottle, boolean
sawServerError) {
+ if (sawThrottle) {
+ return "throttled";
+ }
+ return sawServerError ? "5xx" : "other";
+ }
+
+ private static <T> T first(MetricCollection collection, SdkMetric<T>
metric, T defaultValue) {
+ final List<T> values = collection.metricValues(metric);
+ return values.isEmpty() ? defaultValue : values.get(0);
+ }
+
+ @Override
+ public void close() {}
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/S3MetricRecorder.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/S3MetricRecorder.java
new file mode 100644
index 00000000000..852e66ab115
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/S3MetricRecorder.java
@@ -0,0 +1,177 @@
+/*
+ * 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.fs.s3native.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 native S3 operation metrics supplied by the AWS SDK adapter. */
+@Internal
+public final class S3MetricRecorder {
+
+ static final String API_CALL_COUNT = "api_call_count";
+ static final String API_CALL_DURATION_MS = "api_call_duration_ms";
+ static final String THROTTLE_COUNT = "throttle_count";
+ static final String RETRY_COUNT = "retry_count";
+ static final String IOPS = "iops";
+
+ public static final List<String> DEFAULT_ALLOWLIST =
+ Collections.unmodifiableList(
+ Arrays.asList(
+ API_CALL_COUNT,
+ API_CALL_DURATION_MS,
+ THROTTLE_COUNT,
+ RETRY_COUNT,
+ IOPS));
+
+ private static final String WILDCARD = "*";
+ private static final String LABEL_OPERATION = "op";
+ private static final String LABEL_STATUS_CLASS = "status_class";
+ private static final String LABEL_REASON = "reason";
+
+ private final MetricGroup metricGroup;
+ private final int histogramWindowSize;
+ private final boolean allowAll;
+ private final Set<String> allowlist;
+ private final Map<String, Counter> counters = new ConcurrentHashMap<>();
+ private final Map<String, Histogram> histograms = new
ConcurrentHashMap<>();
+
+ S3MetricRecorder(
+ MetricGroup metricGroup,
+ @Nullable Collection<String> configuredAllowlist,
+ int histogramWindowSize) {
+ this.metricGroup = Preconditions.checkNotNull(metricGroup);
+ Preconditions.checkArgument(
+ histogramWindowSize > 0, "histogramWindowSize must be
positive");
+ this.histogramWindowSize = histogramWindowSize;
+
+ final Set<String> normalizedAllowlist =
normalizeAllowlist(configuredAllowlist);
+ Preconditions.checkArgument(
+ !normalizedAllowlist.isEmpty(),
+ "S3 metrics allowlist must not be empty. Disable metrics with "
+ + "s3.metrics.enabled=false instead.");
+ if (normalizedAllowlist.contains(WILDCARD)) {
+ this.allowAll = true;
+ this.allowlist = Collections.emptySet();
+ } else {
+ if (normalizedAllowlist.contains(IOPS)) {
+ normalizedAllowlist.add(API_CALL_COUNT);
+ }
+ this.allowAll = false;
+ this.allowlist = normalizedAllowlist;
+ }
+ }
+
+ void recordApiCall(String operation, String statusClass) {
+ if (!isMetricEnabled(API_CALL_COUNT)) {
+ return;
+ }
+ final String normalizedOperation =
Preconditions.checkNotNull(operation);
+ final String normalizedStatusClass =
Preconditions.checkNotNull(statusClass);
+ counters.computeIfAbsent(
+ "api-call|" + normalizedOperation + '|' +
normalizedStatusClass,
+ ignored ->
+ metricGroup
+ .addGroup(LABEL_OPERATION,
normalizedOperation)
+ .addGroup(LABEL_STATUS_CLASS,
normalizedStatusClass)
+ .counter(API_CALL_COUNT, new
ThreadSafeSimpleCounter()))
+ .inc();
+ }
+
+ void recordDuration(String operation, long durationMillis) {
+ if (!isMetricEnabled(API_CALL_DURATION_MS)) {
+ return;
+ }
+ final String normalizedOperation =
Preconditions.checkNotNull(operation);
+ histograms
+ .computeIfAbsent(
+ normalizedOperation,
+ ignored ->
+ metricGroup
+ .addGroup(LABEL_OPERATION,
normalizedOperation)
+ .histogram(
+ API_CALL_DURATION_MS,
+ new
SlidingWindowHistogram(histogramWindowSize)))
+ .update(durationMillis);
+ }
+
+ void recordThrottles(String operation, long count) {
+ if (count <= 0 || !isMetricEnabled(THROTTLE_COUNT)) {
+ return;
+ }
+ final String normalizedOperation =
Preconditions.checkNotNull(operation);
+ counters.computeIfAbsent(
+ "throttle|" + normalizedOperation,
+ ignored ->
+ metricGroup
+ .addGroup(LABEL_OPERATION,
normalizedOperation)
+ .counter(THROTTLE_COUNT, new
ThreadSafeSimpleCounter()))
+ .inc(count);
+ }
+
+ void recordRetries(String operation, long count, String reason) {
+ if (count <= 0 || !isMetricEnabled(RETRY_COUNT)) {
+ return;
+ }
+ final String normalizedOperation =
Preconditions.checkNotNull(operation);
+ final String normalizedReason = Preconditions.checkNotNull(reason);
+ counters.computeIfAbsent(
+ "retry|" + normalizedOperation + '|' +
normalizedReason,
+ ignored ->
+ metricGroup
+ .addGroup(LABEL_OPERATION,
normalizedOperation)
+ .addGroup(LABEL_REASON,
normalizedReason)
+ .counter(RETRY_COUNT, new
ThreadSafeSimpleCounter()))
+ .inc(count);
+ }
+
+ boolean isMetricEnabled(String metricName) {
+ return allowAll || allowlist.contains(metricName);
+ }
+
+ private static Set<String> normalizeAllowlist(
+ @Nullable Collection<String> configuredAllowlist) {
+ final Set<String> normalized = new HashSet<>();
+ if (configuredAllowlist == null) {
+ return normalized;
+ }
+ for (String metric : configuredAllowlist) {
+ if (metric != null && !metric.trim().isEmpty()) {
+ normalized.add(metric.trim());
+ }
+ }
+ return normalized;
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
index 902bc9ae9b6..5ed2edd1614 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
@@ -21,10 +21,12 @@ package org.apache.flink.fs.s3native;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.fs.s3native.metrics.AwsSdkMetricBridge;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.metrics.MetricPublisher;
import java.io.IOException;
import java.net.URI;
@@ -96,6 +98,48 @@ class NativeS3FileSystemFactoryTest {
assertThat(fs.getUri()).isEqualTo(URI.create("s3://test-bucket/"));
}
+ @Test
+ void testMetricsPublisherIsInstalledBeforeMetricGroupAttachment() throws
Exception {
+ NativeS3FileSystem fs = createFs(baseConfig());
+ List<MetricPublisher> syncPublishers =
+ fs.getClientProvider()
+ .getS3Client()
+ .serviceClientConfiguration()
+ .overrideConfiguration()
+ .metricPublishers();
+ List<MetricPublisher> asyncPublishers =
+ fs.getClientProvider()
+ .getAsyncClient()
+ .serviceClientConfiguration()
+ .overrideConfiguration()
+ .metricPublishers();
+
+
assertThat(syncPublishers).singleElement().isInstanceOf(AwsSdkMetricBridge.class);
+ assertThat(asyncPublishers).containsExactly(syncPublishers.get(0));
+ }
+
+ @Test
+ void testMetricsPublisherIsNotInstalledWhenMetricsAreDisabled() throws
Exception {
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.METRICS_ENABLED, false);
+ NativeS3FileSystem fs = createFs(config);
+
+ assertThat(
+ fs.getClientProvider()
+ .getS3Client()
+ .serviceClientConfiguration()
+ .overrideConfiguration()
+ .metricPublishers())
+ .isEmpty();
+ assertThat(
+ fs.getClientProvider()
+ .getAsyncClient()
+ .serviceClientConfiguration()
+ .overrideConfiguration()
+ .metricPublishers())
+ .isEmpty();
+ }
+
@Test
void testCreateFileSystemWithCustomEndpoint() throws Exception {
// Global: endpoint A; bucket: endpoint B → bucket endpoint is used
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridgeTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridgeTest.java
new file mode 100644
index 00000000000..b001eb4e95d
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridgeTest.java
@@ -0,0 +1,187 @@
+/*
+ * 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.fs.s3native.metrics;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.SlidingWindowHistogram;
+import org.apache.flink.metrics.testutils.MetricListener;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import software.amazon.awssdk.core.metrics.CoreMetric;
+import software.amazon.awssdk.http.HttpMetric;
+import software.amazon.awssdk.metrics.MetricCollection;
+import software.amazon.awssdk.metrics.MetricCollector;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class AwsSdkMetricBridgeTest {
+
+ @Test
+ void mapsSuccessfulCallAndDuration() {
+ final MetricListener listener = new MetricListener();
+ final AwsSdkMetricBridge bridge = new
AwsSdkMetricBridge(listener.getMetricGroup());
+
+ bridge.publish(apiCall("PutObject", Duration.ofMillis(120), true, 0,
200));
+
+ assertThat(count(listener, "op", "PutObject", "status_class", "2xx",
"api_call_count"))
+ .isEqualTo(1L);
+ final Histogram histogram = histogram(listener, "op", "PutObject",
"api_call_duration_ms");
+ assertThat(histogram.getCount()).isEqualTo(1L);
+ assertThat(histogram.getStatistics().getMax()).isEqualTo(120L);
+ }
+
+ @Test
+ void mapsAwsThrottlingAndRetries() {
+ final MetricListener listener = new MetricListener();
+ final AwsSdkMetricBridge bridge = new
AwsSdkMetricBridge(listener.getMetricGroup());
+
+ bridge.publish(apiCall("UploadPart", Duration.ofMillis(900), true, 2,
503, 503, 200));
+
+ assertThat(count(listener, "op", "UploadPart",
"throttle_count")).isEqualTo(2L);
+ assertThat(count(listener, "op", "UploadPart", "reason", "throttled",
"retry_count"))
+ .isEqualTo(2L);
+ assertThat(count(listener, "op", "UploadPart", "status_class", "2xx",
"api_call_count"))
+ .isEqualTo(1L);
+ }
+
+ @ParameterizedTest(name = "HTTP {0} (successful={1}) -> status_class {2}")
+ @CsvSource({
+ "200, true, 2xx",
+ "404, false, 4xx",
+ "500, false, 5xx",
+ "503, false, throttled",
+ "429, false, throttled",
+ "302, false, other"
+ })
+ void mapsAwsHttpStatus(int status, boolean successful, String
expectedClass) {
+ final MetricListener listener = new MetricListener();
+ final AwsSdkMetricBridge bridge = new
AwsSdkMetricBridge(listener.getMetricGroup());
+
+ bridge.publish(apiCall("GetObject", Duration.ofMillis(15), successful,
0, status));
+
+ assertThat(
+ count(
+ listener,
+ "op",
+ "GetObject",
+ "status_class",
+ expectedClass,
+ "api_call_count"))
+ .isEqualTo(1L);
+ }
+
+ @Test
+ void mapsAwsServerErrorRetry() {
+ final MetricListener listener = new MetricListener();
+ final AwsSdkMetricBridge bridge = new
AwsSdkMetricBridge(listener.getMetricGroup());
+
+ bridge.publish(apiCall("GetObject", Duration.ofMillis(50), true, 1,
500, 200));
+
+ assertThat(count(listener, "op", "GetObject", "reason", "5xx",
"retry_count"))
+ .isEqualTo(1L);
+ assertThat(listener.getCounter("op", "GetObject",
"throttle_count")).isEmpty();
+ }
+
+ @Test
+ void mapsMissingSdkFieldsToUnknown() {
+ final MetricListener listener = new MetricListener();
+ final AwsSdkMetricBridge bridge = new
AwsSdkMetricBridge(listener.getMetricGroup());
+
+ bridge.publish(MetricCollector.create("ApiCall").collect());
+
+ assertThat(count(listener, "op", "Unknown", "status_class", "unknown",
"api_call_count"))
+ .isEqualTo(1L);
+ }
+
+ @Test
+ void startsPublishingAfterMetricGroupIsAttached() {
+ final AwsSdkMetricBridge bridge =
+ new AwsSdkMetricBridge(
+ S3MetricRecorder.DEFAULT_ALLOWLIST,
+ SlidingWindowHistogram.DEFAULT_WINDOW_SIZE);
+ final MetricCollection call = apiCall("PutObject",
Duration.ofMillis(120), true, 0, 200);
+
+ bridge.publish(call);
+
+ final MetricListener listener = new MetricListener();
+ bridge.setMetricGroup(listener.getMetricGroup());
+ bridge.publish(call);
+
+ assertThat(count(listener, "op", "PutObject", "status_class", "2xx",
"api_call_count"))
+ .isEqualTo(1L);
+ }
+
+ @Test
+ void reattachingMetricGroupRedirectsSubsequentMetrics() {
+ final MetricListener firstListener = new MetricListener();
+ final MetricListener secondListener = new MetricListener();
+ final AwsSdkMetricBridge bridge = new
AwsSdkMetricBridge(firstListener.getMetricGroup());
+ final MetricCollection call = apiCall("PutObject",
Duration.ofMillis(120), true, 0, 200);
+
+ bridge.publish(call);
+ bridge.setMetricGroup(secondListener.getMetricGroup());
+ bridge.publish(call);
+
+ assertThat(count(firstListener, "op", "PutObject", "status_class",
"2xx", "api_call_count"))
+ .isEqualTo(1L);
+ assertThat(
+ count(
+ secondListener,
+ "op",
+ "PutObject",
+ "status_class",
+ "2xx",
+ "api_call_count"))
+ .isEqualTo(1L);
+ }
+
+ private static MetricCollection apiCall(
+ String operation,
+ Duration duration,
+ boolean successful,
+ int retries,
+ int... attemptStatuses) {
+ final MetricCollector apiCall = MetricCollector.create("ApiCall");
+ apiCall.reportMetric(CoreMetric.OPERATION_NAME, operation);
+ apiCall.reportMetric(CoreMetric.API_CALL_DURATION, duration);
+ apiCall.reportMetric(CoreMetric.API_CALL_SUCCESSFUL, successful);
+ apiCall.reportMetric(CoreMetric.RETRY_COUNT, retries);
+ for (int status : attemptStatuses) {
+ final MetricCollector attempt =
apiCall.createChild("ApiCallAttempt");
+ attempt.reportMetric(HttpMetric.HTTP_STATUS_CODE, status);
+ }
+ return apiCall.collect();
+ }
+
+ private static long count(MetricListener listener, String... identifier) {
+ return listener.getCounter(identifier)
+ .map(Counter::getCount)
+ .orElseThrow(() -> new AssertionError("Missing counter"));
+ }
+
+ private static Histogram histogram(MetricListener listener, String...
identifier) {
+ return listener.getHistogram(identifier)
+ .orElseThrow(() -> new AssertionError("Missing histogram"));
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3FileSystemFactoryMetricsTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3FileSystemFactoryMetricsTest.java
new file mode 100644
index 00000000000..483307bc8ee
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3FileSystemFactoryMetricsTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.fs.s3native.metrics;
+
+import org.apache.flink.fs.s3native.NativeS3AFileSystemFactory;
+import org.apache.flink.fs.s3native.NativeS3FileSystemFactory;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SlidingWindowHistogram;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that the native S3 factories register metrics under a {@code
filesystem_type} label whose
+ * value is the factory scheme, so {@code s3://} and {@code s3a://} traffic
remain distinguishable.
+ */
+class NativeS3FileSystemFactoryMetricsTest {
+
+ @Test
+ void definesS3SpecificMetricOptions() {
+
assertThat(NativeS3FileSystemFactory.METRICS_ENABLED.key()).isEqualTo("s3.metrics.enabled");
+
assertThat(NativeS3FileSystemFactory.METRICS_ENABLED.defaultValue()).isTrue();
+ assertThat(NativeS3FileSystemFactory.METRICS_ALLOWLIST.key())
+ .isEqualTo("s3.metrics.allowlist");
+ assertThat(NativeS3FileSystemFactory.METRICS_ALLOWLIST.defaultValue())
+ .containsExactlyElementsOf(S3MetricRecorder.DEFAULT_ALLOWLIST);
+
assertThat(NativeS3FileSystemFactory.METRICS_HISTOGRAM_WINDOW_SIZE.key())
+ .isEqualTo("s3.metrics.histogram.window-size");
+
assertThat(NativeS3FileSystemFactory.METRICS_HISTOGRAM_WINDOW_SIZE.defaultValue())
+ .isEqualTo(SlidingWindowHistogram.DEFAULT_WINDOW_SIZE);
+ }
+
+ @ParameterizedTest(name = "filesystem_type={1}")
+ @MethodSource("factories")
+ void factoryUsesSchemeAsFilesystemTypeLabel(
+ NativeS3FileSystemFactory factory, String expectedScheme) {
+ RecordingGroup group = new RecordingGroup();
+ factory.setMetricGroup(group);
+ assertThat(group.keyedGroups).containsEntry("filesystem_type",
expectedScheme);
+ }
+
+ private static Stream<Arguments> factories() {
+ return Stream.of(
+ Arguments.of(new NativeS3FileSystemFactory(), "s3"),
+ Arguments.of(new NativeS3AFileSystemFactory(), "s3a"));
+ }
+
+ @Test
+ void repeatedAttachmentWithSameGroupDoesNotCreateNewFilesystemTypeGroup() {
+ RecordingGroup group = new RecordingGroup();
+ NativeS3FileSystemFactory factory = new NativeS3FileSystemFactory();
+
+ factory.setMetricGroup(group);
+ factory.setMetricGroup(group);
+
+ assertThat(group.addGroupCalls).isEqualTo(1);
+ }
+
+ private static final class RecordingGroup extends UnregisteredMetricsGroup
{
+ final Map<String, String> keyedGroups = new HashMap<>();
+ int addGroupCalls;
+
+ @Override
+ public MetricGroup addGroup(String key, String value) {
+ addGroupCalls++;
+ keyedGroups.put(key, value);
+ return this;
+ }
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3MetricsEmissionITCase.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3MetricsEmissionITCase.java
new file mode 100644
index 00000000000..9a97ad293b2
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3MetricsEmissionITCase.java
@@ -0,0 +1,218 @@
+/*
+ * 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.fs.s3native.metrics;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.fs.s3native.NativeS3AFileSystemFactory;
+import org.apache.flink.fs.s3native.NativeS3FileSystemFactory;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.util.AutoCloseableAsync;
+import org.apache.flink.util.DockerImageVersions;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+
+import java.io.FileNotFoundException;
+import java.net.URI;
+import java.time.Duration;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end test proving that real S3 operations performed through {@code
NativeS3FileSystem} are
+ * translated into Flink metrics by {@link AwsSdkMetricBridge} and become
visible in a real Flink
+ * metric registry.
+ *
+ * <p>Unlike {@link AwsSdkMetricBridgeTest} (which drives the bridge with
synthesized SDK records
+ * and a fake {@code MetricGroup}), this test exercises the full chain: the
AWS SDK actually invokes
+ * the registered {@link software.amazon.awssdk.metrics.MetricPublisher} after
each completed API
+ * call, the bridge registers and updates {@link Counter}/{@link Histogram}
handles, and the
+ * assertions read those handles back through {@link MetricListener}'s real
{@code MetricRegistry}.
+ *
+ * <p>Assertions use only GET/HEAD/LIST round trips, which carry no request
body and are therefore
+ * unaffected by the request-checksum behaviour newer AWS SDK versions apply
to {@code PutObject}.
+ *
+ * <p>Requires Docker; auto-skipped when Docker is unavailable.
+ */
+@Testcontainers(disabledWithoutDocker = true)
+class NativeS3MetricsEmissionITCase {
+
+ private static final int SEAWEEDFS_PORT = 8333;
+ private static final String ACCESS_KEY = "metricsAccessKey";
+ private static final String SECRET_KEY = "metricsSecretKey";
+ private static final String BUCKET = "flip576-metrics";
+
+ @Container
+ private static final GenericContainer<?> SEAWEEDFS =
+ new GenericContainer<>(DockerImageVersions.SEAWEEDFS)
+ .withEnv("AWS_ACCESS_KEY_ID", ACCESS_KEY)
+ .withEnv("AWS_SECRET_ACCESS_KEY", SECRET_KEY)
+ .withCommand("server", "-s3", "-s3.port=" +
SEAWEEDFS_PORT, "-dir=/data")
+ .withExposedPorts(SEAWEEDFS_PORT)
+ .waitingFor(
+ Wait.forHttp("/healthz")
+ .forPort(SEAWEEDFS_PORT)
+ .withStartupTimeout(Duration.ofMinutes(2)))
+ .withStartupAttempts(3);
+
+ private static String endpoint() {
+ return String.format(
+ "http://%s:%d";, SEAWEEDFS.getHost(),
SEAWEEDFS.getMappedPort(SEAWEEDFS_PORT));
+ }
+
+ @BeforeAll
+ static void createBucket() {
+ try (S3Client client =
+ S3Client.builder()
+ .endpointOverride(URI.create(endpoint()))
+ .region(Region.US_EAST_1)
+ .credentialsProvider(
+ StaticCredentialsProvider.create(
+ AwsBasicCredentials.create(ACCESS_KEY,
SECRET_KEY)))
+ .serviceConfiguration(
+
S3Configuration.builder().pathStyleAccessEnabled(true).build())
+ .build()) {
+
client.createBucket(CreateBucketRequest.builder().bucket(BUCKET).build());
+ }
+ }
+
+ @ParameterizedTest(name = "filesystem_type={1}")
+ @MethodSource("factories")
+ void realS3OperationsEmitFlinkMetrics(NativeS3FileSystemFactory factory,
String expectedScheme)
+ throws Exception {
+ Configuration config = new Configuration();
+ config.set(NativeS3FileSystemFactory.ENDPOINT, endpoint());
+ config.set(NativeS3FileSystemFactory.ACCESS_KEY, ACCESS_KEY);
+ config.set(NativeS3FileSystemFactory.SECRET_KEY, SECRET_KEY);
+ config.set(NativeS3FileSystemFactory.REGION, "us-east-1");
+ config.set(NativeS3FileSystemFactory.PATH_STYLE_ACCESS, true);
+ config.set(NativeS3FileSystemFactory.CHUNKED_ENCODING_ENABLED, false);
+ config.set(NativeS3FileSystemFactory.CHECKSUM_VALIDATION_ENABLED,
false);
+ config.set(NativeS3FileSystemFactory.METRICS_ENABLED, true);
+
+ factory.configure(config);
+
+ final Path bucketPath = new Path(expectedScheme + "://" + BUCKET +
"/");
+ FileSystem fs = factory.create(bucketPath.toUri());
+ try {
+ MetricListener metricListener = new MetricListener();
+ // Mirror what FileSystem#attachMetrics hands to the factory: the
"filesystem" child of
+ // the process-level group.
+ MetricGroup fsGroup =
metricListener.getMetricGroup().addGroup("filesystem");
+ factory.setMetricGroup(fsGroup);
+
+ // (1) A successful listing -> ListObjectsV2 (2xx). No request
body, no checksum.
+ FileStatus[] listing = fs.listStatus(bucketPath);
+ assertThat(listing).isNotNull();
+
+ // (2) A lookup of a key that does not exist -> HeadObject
classified as an error
+ // (4xx).
+ try {
+ fs.getFileStatus(new Path(bucketPath, "does-not-exist-" +
System.nanoTime()));
+ } catch (FileNotFoundException expected) {
+ // expected: the object is absent
+ }
+
+ // The SDK publishes metrics after each completed call; the sync
client typically
+ // publishes inline, but poll to remain robust against any
asynchronous delivery.
+ CommonTestUtils.waitUtil(
+ () -> listObjectsSuccessCount(metricListener,
expectedScheme) > 0L,
+ Duration.ofSeconds(30),
+ "Expected a ListObjectsV2 api_call_count metric to be
emitted by real S3 traffic");
+
+ // --- api_call_count (Counter) for the successful listing ---
+ long listCalls = listObjectsSuccessCount(metricListener,
expectedScheme);
+ assertThat(listCalls).as("ListObjectsV2 (2xx)
api_call_count").isGreaterThan(0L);
+
+ // --- api_call_duration_ms (Histogram) for the listing ---
+ Optional<Histogram> listDuration =
+ metricListener.getHistogram(
+ "filesystem",
+ "filesystem_type",
+ expectedScheme,
+ "op",
+ "ListObjectsV2",
+ "api_call_duration_ms");
+ assertThat(listDuration).as("ListObjectsV2 duration
histogram").isPresent();
+ assertThat(listDuration.get().getCount()).isGreaterThan(0L);
+
+ // --- the failed lookup is recorded and classified as a client
error (4xx) ---
+ long headErrorCalls =
+ counter(
+ metricListener,
+ "filesystem",
+ "filesystem_type",
+ expectedScheme,
+ "op",
+ "HeadObject",
+ "status_class",
+ "4xx",
+ "api_call_count");
+ assertThat(headErrorCalls)
+ .as("HeadObject (4xx) api_call_count for the missing key")
+ .isGreaterThan(0L);
+ } finally {
+ ((AutoCloseableAsync) fs).closeAsync().get();
+ }
+ }
+
+ private static Stream<Arguments> factories() {
+ return Stream.of(
+ Arguments.of(new NativeS3FileSystemFactory(), "s3"),
+ Arguments.of(new NativeS3AFileSystemFactory(), "s3a"));
+ }
+
+ private static long listObjectsSuccessCount(MetricListener listener,
String expectedScheme) {
+ return counter(
+ listener,
+ "filesystem",
+ "filesystem_type",
+ expectedScheme,
+ "op",
+ "ListObjectsV2",
+ "status_class",
+ "2xx",
+ "api_call_count");
+ }
+
+ private static long counter(MetricListener listener, String... identifier)
{
+ return
listener.getCounter(identifier).map(Counter::getCount).orElse(0L);
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/S3MetricRecorderTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/S3MetricRecorderTest.java
new file mode 100644
index 00000000000..062a9b2b41e
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/S3MetricRecorderTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.fs.s3native.metrics;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.SlidingWindowHistogram;
+import org.apache.flink.metrics.testutils.MetricListener;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class S3MetricRecorderTest {
+
+ @Test
+ void restrictedAllowlistRegistersOnlySelectedMetrics() {
+ final MetricListener listener = new MetricListener();
+ final S3MetricRecorder recorder =
+ recorder(listener,
Collections.singletonList(S3MetricRecorder.API_CALL_COUNT));
+
+ recordAllMetrics(recorder);
+
+ assertThat(count(listener, "op", "Write", "status_class", "2xx",
"api_call_count"))
+ .isEqualTo(1L);
+ assertThat(listener.getHistogram("op", "Write",
"api_call_duration_ms")).isEmpty();
+ assertThat(listener.getCounter("op", "Write",
"throttle_count")).isEmpty();
+ assertThat(listener.getCounter("op", "Write", "reason", "throttled",
"retry_count"))
+ .isEmpty();
+ }
+
+ @Test
+ void iopsEnablesSourceCounter() {
+ final MetricListener listener = new MetricListener();
+ final S3MetricRecorder recorder =
+ recorder(listener,
Collections.singletonList(S3MetricRecorder.IOPS));
+
+ recorder.recordApiCall("Read", "2xx");
+
+ assertThat(count(listener, "op", "Read", "status_class", "2xx",
"api_call_count"))
+ .isEqualTo(1L);
+ }
+
+ @Test
+ void accumulatesMetricsAcrossCalls() {
+ final MetricListener listener = new MetricListener();
+ final S3MetricRecorder recorder = recorder(listener,
S3MetricRecorder.DEFAULT_ALLOWLIST);
+
+ recorder.recordApiCall("Read", "2xx");
+ recorder.recordApiCall("Read", "2xx");
+ recorder.recordDuration("Read", 10L);
+ recorder.recordDuration("Read", 30L);
+
+ assertThat(count(listener, "op", "Read", "status_class", "2xx",
"api_call_count"))
+ .isEqualTo(2L);
+ assertThat(
+ listener.getHistogram("op", "Read",
"api_call_duration_ms")
+ .orElseThrow(AssertionError::new)
+ .getStatistics()
+ .getValues())
+ .containsExactly(10L, 30L);
+ }
+
+ @Test
+ void wildcardIncludesMetricsOutsideDefaultAllowlist() {
+ final S3MetricRecorder wildcardRecorder =
+ recorder(new MetricListener(), Collections.singletonList("*"));
+ final S3MetricRecorder defaultRecorder =
+ recorder(new MetricListener(),
S3MetricRecorder.DEFAULT_ALLOWLIST);
+
+ assertThat(wildcardRecorder.isMetricEnabled("future_metric")).isTrue();
+ assertThat(defaultRecorder.isMetricEnabled("future_metric")).isFalse();
+ }
+
+ @Test
+ void rejectsEmptyAllowlist() {
+ final MetricListener listener = new MetricListener();
+
+ assertThatThrownBy(() -> recorder(listener, Collections.emptyList()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("allowlist must not be empty");
+ }
+
+ private static S3MetricRecorder recorder(
+ MetricListener listener, java.util.Collection<String> allowlist) {
+ return new S3MetricRecorder(
+ listener.getMetricGroup(), allowlist,
SlidingWindowHistogram.DEFAULT_WINDOW_SIZE);
+ }
+
+ private static void recordAllMetrics(S3MetricRecorder recorder) {
+ recorder.recordApiCall("Write", "2xx");
+ recorder.recordDuration("Write", 10L);
+ recorder.recordThrottles("Write", 1L);
+ recorder.recordRetries("Write", 1L, "throttled");
+ }
+
+ private static long count(MetricListener listener, String... identifier) {
+ return listener.getCounter(identifier)
+ .map(Counter::getCount)
+ .orElseThrow(() -> new AssertionError("Missing counter"));
+ }
+}
diff --git
a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/SlidingWindowHistogram.java
b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/SlidingWindowHistogram.java
new file mode 100644
index 00000000000..acfbbf2df87
--- /dev/null
+++
b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/SlidingWindowHistogram.java
@@ -0,0 +1,151 @@
+/*
+ * 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.metrics;
+
+import org.apache.flink.annotation.Internal;
+
+import java.io.Serializable;
+import java.util.Arrays;
+
+/** A thread-safe histogram retaining a fixed-size window of the most recent
values. */
+@Internal
+public class SlidingWindowHistogram implements Histogram, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final int DEFAULT_WINDOW_SIZE = 1024;
+
+ private final long[] window;
+ private int size;
+ private int next;
+ private long count;
+
+ public SlidingWindowHistogram(int windowSize) {
+ if (windowSize <= 0) {
+ throw new IllegalArgumentException("windowSize must be positive");
+ }
+ this.window = new long[windowSize];
+ }
+
+ @Override
+ public synchronized void update(long value) {
+ window[next] = value;
+ next = (next + 1) % window.length;
+ size = Math.min(size + 1, window.length);
+ count++;
+ }
+
+ @Override
+ public synchronized long getCount() {
+ return count;
+ }
+
+ @Override
+ public synchronized HistogramStatistics getStatistics() {
+ return new SlidingWindowStatistics(getValuesSnapshot());
+ }
+
+ protected synchronized long[] getValuesSnapshot() {
+ return Arrays.copyOf(window, size);
+ }
+
+ private static final class SlidingWindowStatistics extends
HistogramStatistics {
+
+ private final long[] sortedValues;
+ private final double mean;
+ private final double standardDeviation;
+
+ private SlidingWindowStatistics(long[] values) {
+ Arrays.sort(values);
+ this.sortedValues = values;
+ this.mean = calculateMean(values);
+ this.standardDeviation = calculateStandardDeviation(values, mean);
+ }
+
+ @Override
+ public double getQuantile(double quantile) {
+ if (sortedValues.length == 0) {
+ return 0.0;
+ }
+ final double position = quantile * (sortedValues.length + 1);
+ if (position < 1) {
+ return sortedValues[0];
+ }
+ if (position >= sortedValues.length) {
+ return sortedValues[sortedValues.length - 1];
+ }
+ final int lower = (int) position;
+ final double fraction = position - lower;
+ return sortedValues[lower - 1]
+ + fraction * (sortedValues[lower] - sortedValues[lower -
1]);
+ }
+
+ @Override
+ public long[] getValues() {
+ return Arrays.copyOf(sortedValues, sortedValues.length);
+ }
+
+ @Override
+ public int size() {
+ return sortedValues.length;
+ }
+
+ @Override
+ public double getMean() {
+ return mean;
+ }
+
+ @Override
+ public double getStdDev() {
+ return standardDeviation;
+ }
+
+ @Override
+ public long getMax() {
+ return sortedValues.length == 0 ? 0L :
sortedValues[sortedValues.length - 1];
+ }
+
+ @Override
+ public long getMin() {
+ return sortedValues.length == 0 ? 0L : sortedValues[0];
+ }
+
+ private static double calculateMean(long[] values) {
+ if (values.length == 0) {
+ return 0.0;
+ }
+ double sum = 0.0;
+ for (long value : values) {
+ sum += value;
+ }
+ return sum / values.length;
+ }
+
+ private static double calculateStandardDeviation(long[] values, double
mean) {
+ if (values.length <= 1) {
+ return 0.0;
+ }
+ double sum = 0.0;
+ for (long value : values) {
+ final double difference = value - mean;
+ sum += difference * difference;
+ }
+ return Math.sqrt(sum / (values.length - 1));
+ }
+ }
+}
diff --git
a/flink-metrics/flink-metrics-core/src/test/java/org/apache/flink/metrics/SlidingWindowHistogramTest.java
b/flink-metrics/flink-metrics-core/src/test/java/org/apache/flink/metrics/SlidingWindowHistogramTest.java
new file mode 100644
index 00000000000..5a3a345a538
--- /dev/null
+++
b/flink-metrics/flink-metrics-core/src/test/java/org/apache/flink/metrics/SlidingWindowHistogramTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.metrics;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class SlidingWindowHistogramTest extends AbstractHistogramTest {
+
+ @Test
+ void retainsMostRecentValues() {
+ final SlidingWindowHistogram histogram = new SlidingWindowHistogram(4);
+ for (long value = 1; value <= 10; value++) {
+ histogram.update(value);
+ }
+
+ assertThat(histogram.getCount()).isEqualTo(10L);
+ assertThat(histogram.getStatistics().getValues()).containsExactly(7L,
8L, 9L, 10L);
+ }
+
+ @Test
+ void computesStatistics() {
+ final SlidingWindowHistogram histogram = new
SlidingWindowHistogram(10);
+ for (long value = 1; value <= 5; value++) {
+ histogram.update(value);
+ }
+
+ final HistogramStatistics statistics = histogram.getStatistics();
+ assertThat(statistics.getMean()).isEqualTo(3.0);
+ assertThat(statistics.getStdDev()).isEqualTo(Math.sqrt(2.5));
+ assertThat(statistics.getQuantile(0.5)).isEqualTo(3.0);
+ }
+
+ @Test
+ void statisticsAreSnapshots() {
+ final SlidingWindowHistogram histogram = new SlidingWindowHistogram(2);
+ histogram.update(1L);
+ final HistogramStatistics statistics = histogram.getStatistics();
+
+ histogram.update(2L);
+
+ assertThat(statistics.getValues()).containsExactly(1L);
+ }
+
+ @Test
+ void satisfiesHistogramContract() {
+ testHistogram(10, new SlidingWindowHistogram(10));
+ }
+
+ @Test
+ void rejectsNonPositiveWindowSize() {
+ assertThatThrownBy(() -> new SlidingWindowHistogram(0))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("windowSize must be positive");
+ }
+}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index de80e508eea..dca34f97720 100755
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -419,6 +419,8 @@ public abstract class ClusterEntrypoint implements
AutoCloseableAsync, FatalErro
ConfigurationUtils.getSystemResourceMetricsProbingInterval(
configuration));
+ FileSystem.attachMetrics(processMetricGroup);
+
archivedApplicationStore =
createArchivedApplicationStore(
configuration,
commonRpcService.getScheduledExecutor());
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index c8f0e24f732..7240b4fc1f5 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -640,6 +640,9 @@ public class TaskManagerRunner implements FatalErrorHandler
{
resourceID,
taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+ // Attach file system metrics once the process-level MetricGroup
exists.
+ FileSystem.attachMetrics(taskManagerMetricGroup.f0);
+
final ExecutorService ioExecutor =
Executors.newFixedThreadPool(
taskManagerServicesConfiguration.getNumIoThreads(),
