[FLINK-39546][s3] Improve observability in flink-s3-fs-native by exposing operation-level S3 metrics

[Samrat002]

What is the purpose of the change

flink-s3-fs-native currently emits no metrics. When a job's checkpoints, savepoints, or sinks go through it, operators have no visibility into how Flink is actually talking to S3: request volume, latency, throttling, or retries, which makes diagnosing slow or failing checkpoints largely guesswork.

This change makes the native S3 filesystem report operation-level S3 metrics into Flink's metric system. It does so by bridging the AWS SDK's built-in metrics SPI into Flink Counter/Histogram instruments, so every completed S3 API call is counted, timed, and classified.

More details on : FLIP-576

Brief change log

flink-core

• Add MetricsAware (@PublicEvolving, org.apache.flink.core.plugin): a FileSystemFactory implements it to be handed a MetricGroup.
• Add FileSystem.attachMetrics(MetricGroup) (@Internal): creates a filesystem child group and forwards it to every registered MetricsAware factory; resilient to a misbehaving factory and idempotent.
• PluginFileSystemFactory now implements MetricsAware and forwards setMetricGroup to the wrapped inner factory under the plugin classloader. Without this, plugin-loaded filesystems (the normal deployment mode) would silently never receive the group.

flink-runtime

• ClusterEntrypoint (JobManager) and TaskManagerRunner (TaskManager) call FileSystem.attachMetrics(processMetricGroup) during startup. The ClusterEntrypoint service-init order was adjusted so this runs before HA/blob services cache filesystem clients, otherwise those early clients would be created without a metric group.

flink-s3-fs-native

• NativeS3FileSystemFactory / NativeS3AFileSystemFactory implement MetricsAware and tag metrics with a filesystem_type label set to the scheme (s3 vs s3a), so the two stay distinguishable.
• AwsSdkMetricBridge implements software.amazon.awssdk.metrics.MetricPublisher and translates each MetricCollection into Flink metrics: api_call_count (labels: op, status_class), api_call_duration_ms (histogram, label op), throttle_count (label op), retry_count (labels: op, reason).
• S3MetricHistogram: a bounded sliding-window histogram backing the duration metric.
• S3ClientProvider registers the publisher on the sync/async clients.
• New config options: s3.metrics.enabled (off by default), s3.metrics.allowlist, s3.metrics.histogram.window-size.

Verifying this change

This change added tests and can be verified as follows.

Automated tests

• AwsSdkMetricBridgeTest — translation of SDK records to Flink metrics; status_class classification (2xx/4xx/5xx/throttled); retry attribution; allowlist behavior (explicit list, * wildcard, empty → defaults).
• S3MetricHistogramTest — sliding-window statistics.
• NativeS3FileSystemFactoryMetricsTest — the filesystem_type label resolves to s3 / s3a per factory.
• FileSystemAttachMetricsTest (flink-core) — attachMetrics unwraps PluginFileSystemFactory to reach the real factory, skips non-MetricsAware factories, survives a throwing factory, and is idempotent.
• NativeS3MetricsEmissionITCase — MinIO via Testcontainers; real GET/HEAD/LIST round trips, asserting the counters/histograms are readable back through a real MetricRegistry (MetricListener). Auto-skips without Docker.

Manual end-to-end against real AWS S3

I ran a standalone cluster built from this branch with s3.metrics.enabled: true and the SLF4J reporter, and submitted a large-state streaming job checkpointing to s3://<bucket>/checkpoints (HashMap backend, filesystem checkpoint storage, 10 s interval). The native plugin loaded (Plugin loader ... s3-fs-native), built its client via the SDK default credential chain, and wrote real checkpoint objects to S3. The reporter then showed the metrics on both the TaskManager (data-plane writes) and the JobManager (checkpoint coordination / multipart):

# TaskManager
...filesystem.filesystem_type.s3.op.PutObject.status_class.2xx.api_call_count: 31
...filesystem.filesystem_type.s3.op.PutObject.api_call_duration_ms: count=31, min=343, max=8329, mean=3383.2, p99=8329.0
...filesystem.filesystem_type.s3.op.ListObjectsV2.status_class.2xx.api_call_count: 31
...filesystem.filesystem_type.s3.op.ListObjectsV2.reason.other.retry_count: 1
...filesystem.filesystem_type.s3.op.HeadObject.status_class.4xx.api_call_count: 25

# JobManager
...filesystem.filesystem_type.s3.op.CreateMultipartUpload.status_class.2xx.api_call_count: 25
...filesystem.filesystem_type.s3.op.UploadPart.status_class.2xx.api_call_count: 25
...filesystem.filesystem_type.s3.op.CompleteMultipartUpload.status_class.2xx.api_call_count: 25
...filesystem.filesystem_type.s3.op.DeleteObject.status_class.2xx.api_call_count: 48
...filesystem.filesystem_type.s3.op.HeadObject.status_class.2xx.api_call_count: 48

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: yes, native-s3-fs

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented?
  this change introduces the feature, followup documentation is up next.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

[flinkbot]

CI report:

• 7cae5fe Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[gaborgsomogyi]

Thanks for the efforts! Left some high level questions

[Izeren]

Hi @Samrat002, thank you for the change. I have left clarification questions.

Overall, it looks good, but I would like you to extract more cloud agnostic code from native FS. This way, the shims we will need to do to repeat this in Azure, GCS will be really thin. Another concern I have is regrading duplicating non-trivial code to avoid dependency footprint. It seems to me that would be worth extracting lightweight dependency rather than duplicating the code.

[Izeren]

Hi @Samrat002, I went through some of it today, and I can see that there are many comments that I have left before that were not replied. Could you please provide replies in threads before I take a second pass.

[Izeren]

Thank you @Samrat002, high level LGTM. But I would like some brief explanation regarding backwards compatibility of the change:

-import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
+import org.apache.flink.metrics.SlidingWindowHistogram;

My original suggestion was rather "move" DescriptiveStatistics to metrics-core and make both runtime-core and native-fs import it from metrics-core (which is supposedly lighter dependency to take).

In current implementation, there is a behavioural change of DescriptiveStatisticsHistogram + we still bring in custom implementation which is otherwise unjustified.

[Samrat002]

Thanks @Izeren . I've decoupled this. DescriptiveStatisticsHistogram is now reverted to its original commons-math3 implementation, so there's no behavioural or backwards-compat change to the runtime histogram.

On reusing it from metrics-core, the blocker is that DescriptiveStatisticsHistogramStatistics pulls in commons-math3 (Percentile, StandardDeviation, …). Moving it to flink-metrics-core would force commons-math3 onto the leanest, most widely-depended module (every reporter/connector), which felt like a worse trade than a small dependency-free histogram. And since the S3 plugin only depends on flink-runtime in test scope, its main code can't reference DescriptiveStatisticsHistogram anyway. org.apache.flink.metrics is parent-first for plugins and already on the classpath, so SlidingWindowHistogram lives there for the plugin to use. If we later want to unify the runtime histogram onto a shared sliding-window base,

I'd propose doing that as its own JIRA with proper percentile-compat testing rather than as a side effect here. WDUT?

[Izeren]

Thanks @Izeren . I've decoupled this

Thank you @Samrat002, I agree, lets do improvement of histogram implementation separately. This PR is big enough

[Samrat002]

@gaborgsomogyi PTAL whenever time

[gaborgsomogyi]

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.

[Samrat002]

I have moved the fixed metric schema and recorder out of flink-core into flink-s3-fs-native. flink-core now only provides the MetricGroup attachment and makes no assumptions about metric names, labels, protocols, status classes, or retry reasons.

The S3 plugin owns its allowlist defaults, iops dependency, and AWS HTTP/retry classification.

[gaborgsomogyi]

TaskManager never records S3 filesystem metrics — attachMetrics() runs after the S3 filesystem is already cached

ClusterEntrypoint.java intentionally moves attachMetrics() early, with a comment explaining why:

// Attach file system metrics before HA/blob services can create cached clients.
FileSystem.attachMetrics(processMetricGroup);

TaskManagerRunner.java doesn't get the same treatment — FileSystem.attachMetrics(taskManagerMetricGroup.f0) is inserted inside startTaskManager(), but something earlier in TM startup already resolves the checkpoint S3 path and caches the filesystem instance first. Since FileSystem's static cache holds instances for the process lifetime, and the metric publisher is baked in immutably at creation time, that one early instance never gets metrics — and neither does anything reusing it afterward.

Reproduced on a live deployment: TM log shows the S3 filesystem created at 13:30:43, while startTaskManager() (where attachMetrics() lives) doesn't start until 13:30:46. Result: 89 completed checkpoints, hundreds of GB uploaded, zero filesystem.* metrics ever registered on that TM — while JM (correct ordering) showed metrics fine for the same job.

Suggest applying the same fix pattern to TaskManagerRunner as ClusterEntrypoint: move attachMetrics() before whatever first touches the checkpoint/state filesystem.

[gaborgsomogyi]

Now that the TM fix makes publisher attachment order-independent, is ClusterEntrypoint's early attachMetrics() reordering still needed? If not, could we drop it (or update its comment, since it's no longer required for correctness)?

[gaborgsomogyi]

I think my last comment was not well formed. Now we have 3 places to solve have metrics in s3 issue:

• JM: here we do reordering
• TM: Here we don't do reordering
• resolveMetricBridge which does this dynamically and works for JM and TM too

This is kind of a brainsplit and I assume that we can drop JM reordering, right?

[Samrat002]

I think my last comment was not well formed. Now we have 3 places to solve have metrics in s3 issue:

• JM: here we do reordering
• TM: Here we don't do reordering
• resolveMetricBridge which does this dynamically and works for JM and TM too

This is kind of a brainsplit and I assume that we can drop JM reordering, right?

Yes, the split-brain behaviour existed because the JM relied on startup reordering while the TM relied on late bridge attachment.

I have removed the JM-specific reordering and restored ClusterEntrypoint’s original initialisation order. JM and TM now follow the same lifecycle: each calls FileSystem.attachMetrics() when its process metric group becomes available. The two runtime call sites only provide their respective metric groups. The shared AwsSdkMetricBridge is now the single mechanism responsible for handling initialisation order and cached S3 clients:

• If the S3 client is created first, it retains the stable bridge, which begins publishing when setMetricGroup() is called later.
• If the metric group is attached first, the bridge receives the stored group when the client is created.

Therefore, there are still separate JM and TM attachment points, but no separate correctness strategies: both use the same order-independent bridge behaviour.

[gaborgsomogyi]

LGTM. I need some testing before merge to be sure

[Samrat002]

CI failure is related to https://issues.apache.org/jira/browse/FLINK-40074

[gaborgsomogyi]

@flinkbot run azure