Samrat002 commented on code in PR #28427: URL: https://github.com/apache/flink/pull/28427#discussion_r3619822345
########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/S3MetricHistogram.java: ########## @@ -0,0 +1,149 @@ +/* + * 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.Histogram; +import org.apache.flink.metrics.HistogramStatistics; + +import java.util.Arrays; + +/** + * Minimal sliding-window {@link Histogram} used by {@link AwsSdkMetricBridge} for {@code + * api_call_duration_ms}. + * + * <p>Backed by a fixed-size circular buffer holding the most recent {@code windowSize} samples + * (default {@value #DEFAULT_WINDOW_SIZE}). This bounds memory regardless of request volume and + * keeps {@link #update(long)} O(1); statistics are computed from a sorted snapshot of the window. + * + * <p>flink-s3-fs-native deliberately keeps a minimal dependency footprint and does not depend on + * flink-runtime, so {@code DescriptiveStatisticsHistogram} is not available; this is a small + * self-contained equivalent. + */ +@Internal +public class S3MetricHistogram implements Histogram { Review Comment: Correct, it had nothing S3-specific. The class is removed in 47babe49e34 and the reusable implementation now lives as `SlidingWindowHistogram` in `flink-metrics-core`. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -321,9 +325,53 @@ public class NativeS3FileSystemFactory implements FileSystemFactory { + "When not set, the default chain is used: delegation tokens -> " + "static credentials (if configured) -> DefaultCredentialsProvider."); + public static final ConfigOption<Boolean> METRICS_ENABLED = + ConfigOptions.key("s3.metrics.enabled") Review Comment: Done in 47babe49e34. `FileSystemMetricOptions` now creates the enabled, allowlist, and histogram options from a cloud-specific configuration prefix. Native S3 passes `s3`, so the current keys remain unchanged while other clouds can reuse the option shape. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -321,9 +325,53 @@ public class NativeS3FileSystemFactory implements FileSystemFactory { + "When not set, the default chain is used: delegation tokens -> " + "static credentials (if configured) -> DefaultCredentialsProvider."); + public static final ConfigOption<Boolean> METRICS_ENABLED = + ConfigOptions.key("s3.metrics.enabled") + .booleanType() + .defaultValue(true) + .withDescription( + "Master switch for publishing S3 operation metrics to Flink's metric " + + "system. When false, no metric publisher is attached to the SDK " + + "and no metric is registered. Metrics are only emitted under the " + + "TaskManager and JobManager entrypoints, which provide a " + + "process-level metric group; other contexts (CLI, etc.) emit " + + "none regardless of this setting."); + + public static final ConfigOption<List<String>> METRICS_ALLOWLIST = + ConfigOptions.key("s3.metrics.allowlist") + .stringType() + .asList() + .defaultValues( + "api_call_count", + "api_call_duration_ms", + "throttle_count", + "retry_count", + "iops") Review Comment: Yes. The allowlist replaces the default list, so retaining the defaults while adding `foo` requires `ORIGINAL_DEFAULT,foo`; `*` enables every metric emitted by the plugin. The shared option description now states the replacement behavior explicitly. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -341,6 +389,46 @@ public void configure(Configuration config) { this.bucketConfigProvider = new BucketConfigProvider(config); } + @Override + public synchronized void setMetricGroup(MetricGroup metricGroup) { + // filesystem_type label value is the scheme ("s3" / "s3a"). This is deliberate: s3:// and + // s3a:// are served by separate factory instances, so keeping the scheme as the label value + // lets their traffic be told apart, and sibling FS plugins register the same label key with + // their own scheme. May be called more than once (see MetricsAware); reset the cached + // bridge + // so a re-attach with a different group re-scopes metrics created afterwards. + this.pluginMetrics = metricGroup.addGroup("filesystem_type", getScheme()); Review Comment: `addGroup("filesystem_type", getScheme())` creates a key/value group: the key is `filesystem_type` and the value is `s3` or `s3a`. In the metric scope these are separate components (for example `filesystem.filesystem_type.s3...`), and reporters can expose the pair as the `filesystem_type=s3` variable; it is not the literal string `filesystem_type_s3`. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java: ########## @@ -341,6 +389,46 @@ public void configure(Configuration config) { this.bucketConfigProvider = new BucketConfigProvider(config); } + @Override + public synchronized void setMetricGroup(MetricGroup metricGroup) { + // filesystem_type label value is the scheme ("s3" / "s3a"). This is deliberate: s3:// and + // s3a:// are served by separate factory instances, so keeping the scheme as the label value + // lets their traffic be told apart, and sibling FS plugins register the same label key with + // their own scheme. May be called more than once (see MetricsAware); reset the cached + // bridge + // so a re-attach with a different group re-scopes metrics created afterwards. + this.pluginMetrics = metricGroup.addGroup("filesystem_type", getScheme()); + this.metricBridge = null; + } + + /** + * Returns the SDK metric publisher to attach to clients, or {@code null} when metrics are + * disabled or no metric group has been attached yet (e.g. CLI / embedded usage). The bridge is + * built lazily and cached so all clients of this factory share one set of metric handles. + */ + @Nullable + private AwsSdkMetricBridge resolveMetricBridge(Configuration config) { + final MetricGroup metrics = this.pluginMetrics; + if (metrics == null || !config.get(METRICS_ENABLED)) { + return null; + } + AwsSdkMetricBridge bridge = this.metricBridge; + if (bridge == null) { + synchronized (this) { + bridge = this.metricBridge; + if (bridge == null) { + bridge = + new AwsSdkMetricBridge( + metrics, + config.get(METRICS_ALLOWLIST), + config.get(METRICS_HISTOGRAM_WINDOW_SIZE)); + this.metricBridge = bridge; + } + } + } + return bridge; Review Comment: Good catch. I removed the lock-free double-checked path in 47babe49e34. `setMetricGroup(...)` and bridge resolution now synchronize on the same monitor, and `pluginMetrics`, `attachedMetricGroup`, and `metricBridge` are all `@GuardedBy("this")`. `transient` is unnecessary because the factory is not serializable. ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridgeTest.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.fs.s3native.metrics; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; + +import org.junit.jupiter.api.Test; +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 java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link AwsSdkMetricBridge}'s translation of SDK metric records into Flink metrics. */ +class AwsSdkMetricBridgeTest { + + @Test + void successfulCallIncrementsApiCallCountAndRecordsDuration() { Review Comment: Done. The status mapping is a `@ParameterizedTest` covering 200, 404, 500, 503, 429, and a non-error 302 case. The refactored version is in 47babe49e34. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
