Samrat002 commented on code in PR #28427: URL: https://github.com/apache/flink/pull/28427#discussion_r3619821598
########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java: ########## @@ -0,0 +1,281 @@ +/* + * 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.MetricGroup; +import org.apache.flink.metrics.ThreadSafeSimpleCounter; +import org.apache.flink.util.Preconditions; + +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.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Bridges AWS SDK v2's {@link MetricPublisher} into Flink metrics for {@code flink-s3-fs-native}. + * + * <p>The SDK invokes {@link #publish(MetricCollection)} asynchronously after every completed API + * call, on its internal completion executor. The bridge reads a small, fixed set of fields and + * emits the default metric surface of FLIP-576 against the {@code filesystem_type}-labelled scope + * it is handed at construction: + * + * <ul> + * <li>{@code api_call_count} (Counter) — labels {@code op}, {@code status_class} + * <li>{@code api_call_duration_ms} (Histogram) — label {@code op} + * <li>{@code throttle_count} (Counter) — label {@code op} + * <li>{@code retry_count} (Counter) — labels {@code op}, {@code reason} + * </ul> + * + * <p>({@code iops} from the default allowlist is derived at reporter time as the rate of {@code + * api_call_count}, so it is not a separately registered metric.) + * + * <p><b>Allowlist.</b> Only metrics whose name is in the allowlist passed at construction are + * registered; the rest are skipped on the hot path. A {@code "*"} entry registers everything. The + * default set is the five FLIP-576 metrics ({@code api_call_count}, {@code api_call_duration_ms}, + * {@code throttle_count}, {@code retry_count}, {@code iops}). + * + * <p><b>Cardinality.</b> {@code op} comes from the SDK operation name (a closed set of ~15 values + * for S3), {@code status_class} is a closed classifier ({@code 2xx}, {@code 4xx}, {@code 5xx}, + * {@code throttled}, {@code other}, {@code error}, {@code unknown}), and {@code reason} is a closed + * enum ({@code throttled}, {@code 5xx}, {@code other}). Metric handles are cached in bounded maps, + * so {@link #publish} is a map lookup plus a counter increment with no per-record allocation. + * + * <p><b>Thread-safety.</b> Counters use {@link ThreadSafeSimpleCounter} and the histogram is + * synchronized, so concurrent publishes from the SDK completion executor are safe. The bridge is + * also safe to share across multiple S3 clients of the same plugin instance. + */ +@Internal +public final class AwsSdkMetricBridge implements MetricPublisher { + + private static final Logger LOG = LoggerFactory.getLogger(AwsSdkMetricBridge.class); + + static final String API_CALL_COUNT = "api_call_count"; Review Comment: Agreed. Moved the cloud-neutral metric names and default allowlist into `FileSystemMetricRecorder` in `flink-core`. The AWS bridge now consumes that shared vocabulary. Included in 47babe49e34. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java: ########## @@ -0,0 +1,281 @@ +/* + * 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.MetricGroup; +import org.apache.flink.metrics.ThreadSafeSimpleCounter; +import org.apache.flink.util.Preconditions; + +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.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Bridges AWS SDK v2's {@link MetricPublisher} into Flink metrics for {@code flink-s3-fs-native}. + * + * <p>The SDK invokes {@link #publish(MetricCollection)} asynchronously after every completed API + * call, on its internal completion executor. The bridge reads a small, fixed set of fields and + * emits the default metric surface of FLIP-576 against the {@code filesystem_type}-labelled scope + * it is handed at construction: + * + * <ul> + * <li>{@code api_call_count} (Counter) — labels {@code op}, {@code status_class} + * <li>{@code api_call_duration_ms} (Histogram) — label {@code op} + * <li>{@code throttle_count} (Counter) — label {@code op} + * <li>{@code retry_count} (Counter) — labels {@code op}, {@code reason} + * </ul> + * + * <p>({@code iops} from the default allowlist is derived at reporter time as the rate of {@code + * api_call_count}, so it is not a separately registered metric.) + * + * <p><b>Allowlist.</b> Only metrics whose name is in the allowlist passed at construction are + * registered; the rest are skipped on the hot path. A {@code "*"} entry registers everything. The + * default set is the five FLIP-576 metrics ({@code api_call_count}, {@code api_call_duration_ms}, + * {@code throttle_count}, {@code retry_count}, {@code iops}). + * + * <p><b>Cardinality.</b> {@code op} comes from the SDK operation name (a closed set of ~15 values + * for S3), {@code status_class} is a closed classifier ({@code 2xx}, {@code 4xx}, {@code 5xx}, + * {@code throttled}, {@code other}, {@code error}, {@code unknown}), and {@code reason} is a closed + * enum ({@code throttled}, {@code 5xx}, {@code other}). Metric handles are cached in bounded maps, + * so {@link #publish} is a map lookup plus a counter increment with no per-record allocation. + * + * <p><b>Thread-safety.</b> Counters use {@link ThreadSafeSimpleCounter} and the histogram is + * synchronized, so concurrent publishes from the SDK completion executor are safe. The bridge is + * also safe to share across multiple S3 clients of the same plugin instance. + */ +@Internal +public final class AwsSdkMetricBridge implements MetricPublisher { + + private static final Logger LOG = LoggerFactory.getLogger(AwsSdkMetricBridge.class); + + 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"; + + /** The default-on metric set from FLIP-576. {@code iops} is derived, not registered. */ + static final List<String> DEFAULT_ALLOWLIST = + Arrays.asList(API_CALL_COUNT, API_CALL_DURATION_MS, THROTTLE_COUNT, RETRY_COUNT, IOPS); + + private static final String WILDCARD = "*"; + + private static final String LABEL_OP = "op"; + private static final String LABEL_STATUS_CLASS = "status_class"; + private static final String LABEL_REASON = "reason"; + + private static final String UNKNOWN_OP = "Unknown"; + + private final MetricGroup fsScope; + private final int histogramWindowSize; + + private final boolean allowAll; + private final Set<String> allowlist; + + // op and label sets are closed, so these maps are bounded by construction. + private final ConcurrentHashMap<String, Counter> counters = new ConcurrentHashMap<>(); + private final ConcurrentHashMap<String, S3MetricHistogram> histograms = + new ConcurrentHashMap<>(); + + public AwsSdkMetricBridge(MetricGroup fsScope) { + this(fsScope, DEFAULT_ALLOWLIST, S3MetricHistogram.DEFAULT_WINDOW_SIZE); + } + + public AwsSdkMetricBridge(MetricGroup fsScope, int histogramWindowSize) { + this(fsScope, DEFAULT_ALLOWLIST, histogramWindowSize); + } + + public AwsSdkMetricBridge( + MetricGroup fsScope, @Nullable Collection<String> allowlist, int histogramWindowSize) { + this.fsScope = Preconditions.checkNotNull(fsScope, "fsScope must not be null"); + Preconditions.checkArgument( + histogramWindowSize > 0, "histogramWindowSize must be positive"); + this.histogramWindowSize = histogramWindowSize; + + if (allowlist == null || allowlist.isEmpty()) { + LOG.warn( + "S3 metrics allowlist is empty; falling back to the default metric set {}", + DEFAULT_ALLOWLIST); + this.allowAll = false; + this.allowlist = new HashSet<>(DEFAULT_ALLOWLIST); + } else if (allowlist.contains(WILDCARD)) { + this.allowAll = true; + this.allowlist = new HashSet<>(); + } else { + this.allowAll = false; + this.allowlist = new HashSet<>(allowlist); + } + } + + private boolean allowed(String metricName) { + return allowAll || allowlist.contains(metricName); + } + + @Override + public void publish(MetricCollection apiCall) { + try { + translate(apiCall); + } catch (Throwable t) { + // Defence in depth: a metric failure must never affect S3 IO. + LOG.debug("Failed to publish S3 SDK metrics", t); + } + } + + private void translate(MetricCollection apiCall) { + final String op = first(apiCall, CoreMetric.OPERATION_NAME, UNKNOWN_OP); + + final Duration duration = first(apiCall, CoreMetric.API_CALL_DURATION, null); + if (duration != null && allowed(API_CALL_DURATION_MS)) { + histogram(op).update(duration.toMillis()); + } + + // HTTP_STATUS_CODE lives on the per-attempt children, not on the top-level ApiCall record. + // status_class reflects the overall outcome (last attempt); the retry reason reflects the + // failures that triggered the retries (any attempt), so they are tracked separately. + 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; + } + } + } + } + + final Boolean successful = first(apiCall, CoreMetric.API_CALL_SUCCESSFUL, null); + if (allowed(API_CALL_COUNT)) { + apiCallCount(op, statusClass(lastStatus, successful)).inc(); + } + + if (throttleResponses > 0 && allowed(THROTTLE_COUNT)) { + throttleCount(op).inc(throttleResponses); + } + + final Integer retries = first(apiCall, CoreMetric.RETRY_COUNT, 0); + if (retries != null && retries > 0 && allowed(RETRY_COUNT)) { + retryCount(op, retryReason(throttleResponses > 0, sawServerError)).inc(retries); + } + } + + private static boolean isThrottle(int status) { + return status == 429 || status == 503; + } + + private static String statusClass(Integer status, Boolean successful) { Review Comment: Done in 47babe49e34. `FileSystemMetricRecorder` now owns normalized status classes, labels, allowlist handling, metric registration, and handle caches. `AwsSdkMetricBridge` is limited to extracting SDK fields and mapping concrete AWS responses into that model. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/metrics/AwsSdkMetricBridge.java: ########## @@ -0,0 +1,281 @@ +/* + * 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.MetricGroup; +import org.apache.flink.metrics.ThreadSafeSimpleCounter; +import org.apache.flink.util.Preconditions; + +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.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Bridges AWS SDK v2's {@link MetricPublisher} into Flink metrics for {@code flink-s3-fs-native}. + * + * <p>The SDK invokes {@link #publish(MetricCollection)} asynchronously after every completed API + * call, on its internal completion executor. The bridge reads a small, fixed set of fields and + * emits the default metric surface of FLIP-576 against the {@code filesystem_type}-labelled scope + * it is handed at construction: + * + * <ul> + * <li>{@code api_call_count} (Counter) — labels {@code op}, {@code status_class} + * <li>{@code api_call_duration_ms} (Histogram) — label {@code op} + * <li>{@code throttle_count} (Counter) — label {@code op} + * <li>{@code retry_count} (Counter) — labels {@code op}, {@code reason} + * </ul> + * + * <p>({@code iops} from the default allowlist is derived at reporter time as the rate of {@code + * api_call_count}, so it is not a separately registered metric.) + * + * <p><b>Allowlist.</b> Only metrics whose name is in the allowlist passed at construction are + * registered; the rest are skipped on the hot path. A {@code "*"} entry registers everything. The + * default set is the five FLIP-576 metrics ({@code api_call_count}, {@code api_call_duration_ms}, + * {@code throttle_count}, {@code retry_count}, {@code iops}). + * + * <p><b>Cardinality.</b> {@code op} comes from the SDK operation name (a closed set of ~15 values + * for S3), {@code status_class} is a closed classifier ({@code 2xx}, {@code 4xx}, {@code 5xx}, + * {@code throttled}, {@code other}, {@code error}, {@code unknown}), and {@code reason} is a closed + * enum ({@code throttled}, {@code 5xx}, {@code other}). Metric handles are cached in bounded maps, + * so {@link #publish} is a map lookup plus a counter increment with no per-record allocation. + * + * <p><b>Thread-safety.</b> Counters use {@link ThreadSafeSimpleCounter} and the histogram is + * synchronized, so concurrent publishes from the SDK completion executor are safe. The bridge is + * also safe to share across multiple S3 clients of the same plugin instance. + */ +@Internal +public final class AwsSdkMetricBridge implements MetricPublisher { + + private static final Logger LOG = LoggerFactory.getLogger(AwsSdkMetricBridge.class); + + 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"; + + /** The default-on metric set from FLIP-576. {@code iops} is derived, not registered. */ + static final List<String> DEFAULT_ALLOWLIST = + Arrays.asList(API_CALL_COUNT, API_CALL_DURATION_MS, THROTTLE_COUNT, RETRY_COUNT, IOPS); + + private static final String WILDCARD = "*"; + + private static final String LABEL_OP = "op"; + private static final String LABEL_STATUS_CLASS = "status_class"; + private static final String LABEL_REASON = "reason"; + + private static final String UNKNOWN_OP = "Unknown"; + + private final MetricGroup fsScope; + private final int histogramWindowSize; + + private final boolean allowAll; + private final Set<String> allowlist; + + // op and label sets are closed, so these maps are bounded by construction. + private final ConcurrentHashMap<String, Counter> counters = new ConcurrentHashMap<>(); + private final ConcurrentHashMap<String, S3MetricHistogram> histograms = + new ConcurrentHashMap<>(); + + public AwsSdkMetricBridge(MetricGroup fsScope) { + this(fsScope, DEFAULT_ALLOWLIST, S3MetricHistogram.DEFAULT_WINDOW_SIZE); + } + + public AwsSdkMetricBridge(MetricGroup fsScope, int histogramWindowSize) { + this(fsScope, DEFAULT_ALLOWLIST, histogramWindowSize); + } + + public AwsSdkMetricBridge( + MetricGroup fsScope, @Nullable Collection<String> allowlist, int histogramWindowSize) { + this.fsScope = Preconditions.checkNotNull(fsScope, "fsScope must not be null"); + Preconditions.checkArgument( + histogramWindowSize > 0, "histogramWindowSize must be positive"); + this.histogramWindowSize = histogramWindowSize; + + if (allowlist == null || allowlist.isEmpty()) { + LOG.warn( + "S3 metrics allowlist is empty; falling back to the default metric set {}", + DEFAULT_ALLOWLIST); + this.allowAll = false; + this.allowlist = new HashSet<>(DEFAULT_ALLOWLIST); + } else if (allowlist.contains(WILDCARD)) { + this.allowAll = true; + this.allowlist = new HashSet<>(); + } else { + this.allowAll = false; + this.allowlist = new HashSet<>(allowlist); + } + } + + private boolean allowed(String metricName) { + return allowAll || allowlist.contains(metricName); + } + + @Override + public void publish(MetricCollection apiCall) { + try { + translate(apiCall); + } catch (Throwable t) { + // Defence in depth: a metric failure must never affect S3 IO. + LOG.debug("Failed to publish S3 SDK metrics", t); + } + } + + private void translate(MetricCollection apiCall) { + final String op = first(apiCall, CoreMetric.OPERATION_NAME, UNKNOWN_OP); + + final Duration duration = first(apiCall, CoreMetric.API_CALL_DURATION, null); + if (duration != null && allowed(API_CALL_DURATION_MS)) { + histogram(op).update(duration.toMillis()); + } + + // HTTP_STATUS_CODE lives on the per-attempt children, not on the top-level ApiCall record. + // status_class reflects the overall outcome (last attempt); the retry reason reflects the + // failures that triggered the retries (any attempt), so they are tracked separately. + 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; + } + } + } + } + + final Boolean successful = first(apiCall, CoreMetric.API_CALL_SUCCESSFUL, null); + if (allowed(API_CALL_COUNT)) { + apiCallCount(op, statusClass(lastStatus, successful)).inc(); + } + + if (throttleResponses > 0 && allowed(THROTTLE_COUNT)) { + throttleCount(op).inc(throttleResponses); + } + + final Integer retries = first(apiCall, CoreMetric.RETRY_COUNT, 0); + if (retries != null && retries > 0 && allowed(RETRY_COUNT)) { + retryCount(op, retryReason(throttleResponses > 0, sawServerError)).inc(retries); + } + } + + 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) { Review Comment: Done in 47babe49e34. Retry registration and the normalized `RetryReason` values are cloud-neutral in `FileSystemMetricRecorder`; the AWS adapter only maps its attempts to `THROTTLED`, `SERVER_ERROR`, or `OTHER`. ########## 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 Review Comment: Done in 47babe49e34. The S3-local histogram was removed and replaced by dependency-free `SlidingWindowHistogram` in the existing lightweight `flink-metrics-core` module. The runtime histogram shares its window storage while retaining its existing statistics type, and the native plugin does not gain a `flink-runtime` dependency. -- 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]
