Samrat002 commented on code in PR #28427:
URL: https://github.com/apache/flink/pull/28427#discussion_r3619884447
##########
flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java:
##########
@@ -377,6 +377,33 @@ protected void initializeServices(Configuration
configuration, PluginManager plu
configuration.set(JobManagerOptions.ADDRESS,
commonRpcService.getAddress());
configuration.set(JobManagerOptions.PORT,
commonRpcService.getPort());
+ metricRegistry = createMetricRegistry(configuration,
pluginManager, rpcSystem);
+
+ final RpcService metricQueryServiceRpcService =
+ MetricUtils.startRemoteMetricsRpcService(
+ configuration,
+ commonRpcService.getAddress(),
+ configuration.get(JobManagerOptions.BIND_HOST),
+ rpcSystem);
+ metricRegistry.startQueryService(metricQueryServiceRpcService,
null);
+
+ final String hostname = RpcUtils.getHostname(commonRpcService);
+
+ processMetricGroup =
+ MetricUtils.instantiateProcessMetricGroup(
+ metricRegistry,
+ hostname,
+
ConfigurationUtils.getSystemResourceMetricsProbingInterval(
+ configuration));
+
+ // Second-phase init for file system plugins that opt into metrics
(e.g.
+ // flink-s3-fs-native): hand them the process-level metric group
before any file system
+ // is used. This must run ahead of the HA services and BlobServer
below, because those
+ // may open external file systems (e.g. S3 HA/blob storage),
creating them first would
+ // cache metric-less file system clients for the rest of the
process lifetime. See
+ // FileSystem#attachMetrics and MetricsAware.
Review Comment:
The metric-registry, query-service, and process-group block is moved as-is.
The functional addition is `FileSystem.attachMetrics(processMetricGroup)`
immediately after the process group exists, so filesystem clients created by
later startup services receive the metric scope.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java:
##########
@@ -640,6 +640,11 @@ public static TaskExecutor startTaskManager(
resourceID,
taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
+ // Second-phase init for file system plugins that opt into metrics
(e.g.
+ // flink-s3-fs-native): hand them the process-level metric group now
that the
+ // MetricRegistry exists. See FileSystem#attachMetrics and
MetricsAware.
+ FileSystem.attachMetrics(taskManagerMetricGroup.f0);
Review Comment:
`MetricsAware` is intentionally only the opt-in contract; it does not add
global plugin discovery. A non-filesystem plugin family would use the same
pattern by forwarding `setMetricGroup(...)` through its wrappers and invoking
it at the runtime point where that family already owns an appropriate metric
group. The filesystem attachment is the first concrete integration of that
pattern.
##########
flink-core/src/test/java/org/apache/flink/core/fs/FileSystemAttachMetricsTest.java:
##########
@@ -0,0 +1,213 @@
+/*
+ * 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.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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * Tests for {@link FileSystem#attachMetrics(MetricGroup)} and the {@link
MetricsAware} two-phase
+ * init contract.
+ *
+ * <p>The headline case is {@link
#attachMetricsReachesPluginLoadedMetricsAwareFactory()}: plugin
Review Comment:
Done. The class JavaDoc is now a single-line test description;
behavior-specific explanations stay with the relevant test cases.
##########
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
Review Comment:
Agreed. The numeric estimate was removed. After the extraction in
47babe49e34, the AWS bridge class documentation is now only a one-line
statement of its mapping responsibility.
--
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]
