Samrat002 commented on code in PR #28427: URL: https://github.com/apache/flink/pull/28427#discussion_r3619886274
########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3FileSystemFactoryMetricsTest.java: ########## @@ -0,0 +1,62 @@ +/* + * 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.groups.UnregisteredMetricsGroup; + +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; + +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 s3FactoryUsesSchemeAsFilesystemTypeLabel() { Review Comment: Done. The factory metric-scope test is parameterized across both `s3` and `s3a`. ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3MetricsEmissionITCase.java: ########## @@ -0,0 +1,199 @@ +/* + * 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.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.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +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 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 String MINIO_IMAGE = "minio/minio:RELEASE.2022-02-07T08-17-33Z"; + private static final int MINIO_PORT = 9000; Review Comment: Agreed. MinIO has been removed from the test and from the image constants. The integration test now uses the existing centralized `DockerImageVersions.SEAWEEDFS` image and SeaweedFS S3 configuration. ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/NativeS3MetricsEmissionITCase.java: ########## @@ -0,0 +1,199 @@ +/* + * 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.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.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +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 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 String MINIO_IMAGE = "minio/minio:RELEASE.2022-02-07T08-17-33Z"; Review Comment: The local image tag is gone. The test now reuses the existing centralized `DockerImageVersions.SEAWEEDFS` constant. ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/metrics/S3MetricHistogramTest.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.fs.s3native.metrics; + +import org.apache.flink.metrics.HistogramStatistics; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link S3MetricHistogram}. */ +class S3MetricHistogramTest { Review Comment: Done. `S3MetricHistogram` and its plugin-local test were removed. The implementation is now `SlidingWindowHistogram` in `flink-metrics-core`, with its tests colocated there. ########## 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 Review Comment: This is not compensating for a missing filesystem close hook. The metric group must be attached before HA/blob and other startup services create and cache filesystem clients; attaching later cannot retrofit the SDK publisher into clients that already exist. Filesystem lifetime can still match the JM/TM process lifetime. A close hook would be a separate resource-cleanup improvement. -- 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]
