diff --git a/.changes/next-release/feature-AWSCRTbasedS3Client-ec658f9.json b/.changes/next-release/feature-AWSCRTbasedS3Client-ec658f9.json new file mode 100644 index 000000000000..6ac6547fd0c0 --- /dev/null +++ b/.changes/next-release/feature-AWSCRTbasedS3Client-ec658f9.json @@ -0,0 +1,6 @@ +{ + "type": "feature", + "category": "AWS CRT-based S3 Client", + "contributor": "", + "description": "Added support for publishing SDK metrics from the AWS CRT-based S3 async client through client-level and request-level `MetricPublisher` configurations. This feature requires AWS CRT Java 0.48.5 or later." +} diff --git a/services/s3/src/it/java/software/amazon/awssdk/services/s3/crt/S3CrtClientMetricPublisherIntegrationTest.java b/services/s3/src/it/java/software/amazon/awssdk/services/s3/crt/S3CrtClientMetricPublisherIntegrationTest.java new file mode 100644 index 000000000000..a55fc3607a3d --- /dev/null +++ b/services/s3/src/it/java/software/amazon/awssdk/services/s3/crt/S3CrtClientMetricPublisherIntegrationTest.java @@ -0,0 +1,254 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.services.s3.crt; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static software.amazon.awssdk.testutils.service.S3BucketUtils.temporaryBucketName; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import software.amazon.awssdk.core.async.AsyncRequestBody; +import software.amazon.awssdk.core.async.AsyncResponseTransformer; +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.services.s3.S3AsyncClient; +import software.amazon.awssdk.services.s3.S3IntegrationTestBase; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; +import software.amazon.awssdk.services.s3.model.S3Exception; +import software.amazon.awssdk.services.s3.model.Tag; +import software.amazon.awssdk.services.s3.model.TaggingDirective; +import software.amazon.awssdk.testutils.RandomTempFile; +import software.amazon.awssdk.testutils.service.AwsTestBase; + +/** + * Verifies that the CRT-based S3 client publishes CRT native request telemetry to client-level and request-level + * {@link MetricPublisher}s. Each underlying CRT request attempt is published as its own + * {@code ApiCall -> ApiCallAttempt -> HttpClient} {@link MetricCollection}, so a multipart transfer yields several. + */ +@Timeout(value = 5, unit = TimeUnit.MINUTES) +public class S3CrtClientMetricPublisherIntegrationTest extends S3IntegrationTestBase { + private static final String BUCKET = temporaryBucketName(S3CrtClientMetricPublisherIntegrationTest.class); + private static final String SMALL_KEY = "small-single-part"; + private static final String LARGE_KEY = "large-multipart"; + private static final long PART_SIZE = 8L * 1024 * 1024; + // Below the 8MB threshold -> single underlying GET. + private static final int SMALL_SIZE = 100 * 1024; + // Comfortably above the threshold -> the meta-request fans out into multiple underlying part requests. + private static final int LARGE_SIZE = 17 * 1024 * 1024; + + @BeforeAll + public static void setup() throws Exception { + S3IntegrationTestBase.setUp(); + S3IntegrationTestBase.createBucket(BUCKET); + + S3IntegrationTestBase.s3.putObject(PutObjectRequest.builder().bucket(BUCKET).key(SMALL_KEY).build(), + new RandomTempFile(SMALL_SIZE).toPath()); + S3IntegrationTestBase.s3.putObject(PutObjectRequest.builder().bucket(BUCKET).key(LARGE_KEY).build(), + new RandomTempFile(LARGE_SIZE).toPath()); + } + + @AfterAll + public static void cleanup() { + S3IntegrationTestBase.deleteBucketAndAllContents(BUCKET); + } + + @Test + void singlePartGetObject_publishesApiCallMetrics() throws InterruptedException { + CapturingMetricPublisher publisher = new CapturingMetricPublisher(); + try (S3AsyncClient client = crtClientWith(publisher)) { + client.getObject(b -> b.bucket(BUCKET).key(SMALL_KEY), AsyncResponseTransformer.toBytes()).join(); + } + + List collections = publisher.awaitAtLeast(1, Duration.ofSeconds(30)); + assertThat(collections).isNotEmpty(); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + } + + @Test + void multipartGetObject_publishesOneCollectionPerUnderlyingRequest() throws InterruptedException { + CapturingMetricPublisher publisher = new CapturingMetricPublisher(); + try (S3AsyncClient client = crtClientWith(publisher)) { + client.getObject(b -> b.bucket(BUCKET).key(LARGE_KEY), AsyncResponseTransformer.toBytes()).join(); + } + + // A multipart download fans out into several underlying requests, so we expect more than one collection. + List collections = publisher.awaitAtLeast(2, Duration.ofSeconds(60)); + assertThat(collections).hasSizeGreaterThanOrEqualTo(2); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + } + + @Test + void putObject_publishesApiCallMetrics() throws InterruptedException { + CapturingMetricPublisher publisher = new CapturingMetricPublisher(); + try (S3AsyncClient client = crtClientWith(publisher)) { + client.putObject(b -> b.bucket(BUCKET).key("put-metrics-key"), + AsyncRequestBody.fromBytes(new byte[SMALL_SIZE])).join(); + } + + List collections = publisher.awaitAtLeast(1, Duration.ofSeconds(30)); + assertThat(collections).isNotEmpty(); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + } + + @Test + void failedGetObject_publishesUnsuccessfulApiCallMetrics() throws InterruptedException { + CapturingMetricPublisher publisher = new CapturingMetricPublisher(); + try (S3AsyncClient client = crtClientWith(publisher)) { + assertThatThrownBy(() -> + client.getObject(b -> b.bucket(BUCKET).key("does-not-exist-" + System.nanoTime()), + AsyncResponseTransformer.toBytes()).join()) + .hasCauseInstanceOf(S3Exception.class) + .satisfies(e -> assertThat(((S3Exception) e.getCause()).statusCode()).isEqualTo(404)); + } + + List collections = publisher.awaitAtLeast(1, Duration.ofSeconds(30)); + assertThat(collections).isNotEmpty(); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + assertThat(collections).anySatisfy( + c -> assertThat(c.metricValues(CoreMetric.API_CALL_SUCCESSFUL)).contains(false)); + } + + @Test + void requestLevelPublisher_overridesClientLevelPublisher() throws InterruptedException { + CapturingMetricPublisher clientPublisher = new CapturingMetricPublisher(); + CapturingMetricPublisher requestPublisher = new CapturingMetricPublisher(); + + try (S3AsyncClient client = crtClientWith(clientPublisher)) { + client.getObject(b -> b.bucket(BUCKET).key(SMALL_KEY) + .overrideConfiguration(o -> o.addMetricPublisher(requestPublisher)), + AsyncResponseTransformer.toBytes()).join(); + + // The request-level publisher receives the CRT telemetry ... + List requestCollections = requestPublisher.awaitAtLeast(1, Duration.ofSeconds(30)); + assertThat(requestCollections).isNotEmpty(); + requestCollections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + + // ... and this client's own client-level publisher receives nothing, since request-level takes precedence. + assertThat(clientPublisher.awaitAtLeast(1, Duration.ofSeconds(1))).isEmpty(); + } + } + + @Test + void copyObjectWithRequestLevelPublisher_publishesSubRequestTelemetryToIt() throws InterruptedException { + CapturingMetricPublisher clientPublisher = new CapturingMetricPublisher(); + CapturingMetricPublisher requestPublisher = new CapturingMetricPublisher(); + String destinationKey = "copy-dest-" + System.nanoTime(); + + try (S3AsyncClient client = crtClientWith(clientPublisher)) { + client.copyObject(b -> b.sourceBucket(BUCKET).sourceKey(LARGE_KEY) + .destinationBucket(BUCKET).destinationKey(destinationKey) + .overrideConfiguration(o -> o.addMetricPublisher(requestPublisher))) + .join(); + + List collections = requestPublisher.awaitAtLeast(2, Duration.ofSeconds(60)); + assertThat(collections).hasSizeGreaterThanOrEqualTo(2); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + + // request-level takes precedence, so this client's own client-level publisher sees none of the sub-requests. + assertThat(clientPublisher.awaitAtLeast(1, Duration.ofSeconds(1))).isEmpty(); + } + } + + @Test + void copyObjectWithTaggingDirective_requestLevelPublisher_receivesTaggingSubRequestMetrics() throws InterruptedException { + // Tag the source so a taggingDirective(COPY) copy actually issues GetObjectTagging + PutObjectTagging sub-requests. + S3IntegrationTestBase.s3.putObjectTagging( + r -> r.bucket(BUCKET).key(LARGE_KEY) + .tagging(t -> t.tagSet(Tag.builder().key("env").value("t").build()))); + + CapturingMetricPublisher clientPublisher = new CapturingMetricPublisher(); + CapturingMetricPublisher requestPublisher = new CapturingMetricPublisher(); + String destinationKey = "copy-tagged-dest-" + System.nanoTime(); + + try (S3AsyncClient client = crtClientWith(clientPublisher)) { + client.copyObject(b -> b.sourceBucket(BUCKET).sourceKey(LARGE_KEY) + .destinationBucket(BUCKET).destinationKey(destinationKey) + .taggingDirective(TaggingDirective.COPY) + .overrideConfiguration(o -> o.addMetricPublisher(requestPublisher))) + .join(); + + List collections = requestPublisher.awaitAtLeast(2, Duration.ofSeconds(60)); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + assertThat(collections).anySatisfy( + c -> assertThat(c.metricValues(CoreMetric.OPERATION_NAME)).contains("GetObjectTagging")); + assertThat(collections).anySatisfy( + c -> assertThat(c.metricValues(CoreMetric.OPERATION_NAME)).contains("PutObjectTagging")); + + // request-level takes precedence, so this client's own client-level publisher sees none of the sub-requests. + assertThat(clientPublisher.awaitAtLeast(1, Duration.ofSeconds(1))).isEmpty(); + } + } + + private static S3AsyncClient crtClientWith(MetricPublisher clientLevelPublisher) { + return S3AsyncClient.crtBuilder() + .region(S3IntegrationTestBase.DEFAULT_REGION) + .credentialsProvider(AwsTestBase.CREDENTIALS_PROVIDER_CHAIN) + .minimumPartSizeInBytes(PART_SIZE) + .thresholdInBytes(PART_SIZE) + .addMetricPublisher(clientLevelPublisher) + .build(); + } + + private static void assertIsCrtApiCallCollection(MetricCollection apiCall) { + assertThat(apiCall.name()).isEqualTo("ApiCall"); + assertThat(apiCall.metricValues(CoreMetric.SERVICE_ID)).containsExactly("S3"); + assertThat(apiCall.metricValues(CoreMetric.OPERATION_NAME)).isNotEmpty(); + assertThat(apiCall.metricValues(CoreMetric.API_CALL_DURATION)).isNotEmpty(); + + MetricCollection attempt = apiCall.childrenWithName("ApiCallAttempt").findFirst().orElse(null); + assertThat(attempt).as("ApiCallAttempt child").isNotNull(); + + MetricCollection httpClient = attempt.childrenWithName("HttpClient").findFirst().orElse(null); + assertThat(httpClient).as("HttpClient child").isNotNull(); + assertThat(httpClient.metricValues(HttpMetric.HTTP_CLIENT_NAME)).containsExactly("s3crt"); + } + + /** + * Thread-safe capture of published collections. {@code onTelemetry} fires on CRT native threads and can lag the + * completion of the operation future, so tests poll {@link #awaitAtLeast(int, Duration)} rather than reading + * immediately. + */ + private static final class CapturingMetricPublisher implements MetricPublisher { + private final List collections = new CopyOnWriteArrayList<>(); + + @Override + public void publish(MetricCollection metricCollection) { + collections.add(metricCollection); + } + + @Override + public void close() { + } + + List awaitAtLeast(int min, Duration timeout) throws InterruptedException { + long deadlineNanos = System.nanoTime() + timeout.toNanos(); + while (collections.size() < min && System.nanoTime() < deadlineNanos) { + Thread.sleep(100); + } + return new ArrayList<>(collections); + } + } +} diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java index 71c0d9815fd8..0f8e8d5cfaf4 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/S3CrtAsyncClientBuilder.java @@ -17,6 +17,7 @@ import java.net.URI; import java.nio.file.Path; +import java.util.Collection; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executor; @@ -27,8 +28,10 @@ import software.amazon.awssdk.core.checksums.RequestChecksumCalculation; import software.amazon.awssdk.core.checksums.ResponseChecksumValidation; import software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption; +import software.amazon.awssdk.core.metrics.CoreMetric; import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity; import software.amazon.awssdk.identity.spi.IdentityProvider; +import software.amazon.awssdk.metrics.MetricPublisher; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.s3.crt.S3CrtHttpConfiguration; import software.amazon.awssdk.services.s3.crt.S3CrtRetryConfiguration; @@ -391,6 +394,40 @@ default S3CrtAsyncClientBuilder retryConfiguration(Consumer, ?> advancedOptions); + /** + * Sets the {@link MetricPublisher}s that this client publishes metrics to. This overrides any previously configured + * publishers. + * + *

Each underlying CRT request attempt is published as a separate {@code "ApiCall"} metric collection. A single + * high-level transfer (for example a multipart {@code getObject}) fans out into several underlying requests, so + * several collections are published per transfer. + * + *

When migrating from another S3 client, note that the CRT-based client does not publish + * {@link CoreMetric#CREDENTIALS_FETCH_DURATION}, {@link CoreMetric#READ_THROUGHPUT}, or + * {@link CoreMetric#WRITE_THROUGHPUT}. + * + *

The SDK does not close these publishers when the client is closed; the caller retains ownership and is + * responsible for closing them. Metric publication is asynchronous and best effort. Closing the client waits for a + * bounded time while pending metric collections are published, but may return before collections generated during + * client shutdown are delivered. If a publisher is closed concurrently with this delivery, those metric collections + * may not be published and a debug message may be logged. This does not affect the outcome of the S3 operation. + * + * @param metricPublishers the metric publishers to use + * @return this builder for method chaining. + */ + S3CrtAsyncClientBuilder metricPublishers(Collection metricPublishers); + + /** + * Adds a {@link MetricPublisher} that this client publishes metrics to. Can be called multiple times to add several + * publishers. + * + *

See {@link #metricPublishers(Collection)} for details on how metrics are published and on publisher ownership. + * + * @param metricPublisher the metric publisher to add + * @return this builder for method chaining. + */ + S3CrtAsyncClientBuilder addMetricPublisher(MetricPublisher metricPublisher); + @Override S3AsyncClient build(); -} \ No newline at end of file +} diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClient.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClient.java index 33fe80c7a454..0013bbb615dc 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClient.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClient.java @@ -30,10 +30,13 @@ import java.net.URI; import java.nio.file.Path; import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executor; +import java.util.function.Function; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.annotations.SdkTestInternalApi; import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; @@ -64,6 +67,7 @@ import software.amazon.awssdk.http.SdkHttpExecutionAttributes; import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity; import software.amazon.awssdk.identity.spi.IdentityProvider; +import software.amazon.awssdk.metrics.MetricPublisher; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.s3.DelegatingS3AsyncClient; import software.amazon.awssdk.services.s3.S3AsyncClient; @@ -80,6 +84,7 @@ import software.amazon.awssdk.services.s3.model.GetObjectResponse; import software.amazon.awssdk.services.s3.model.PutObjectRequest; import software.amazon.awssdk.services.s3.model.PutObjectResponse; +import software.amazon.awssdk.services.s3.model.S3Request; import software.amazon.awssdk.services.s3.presignedurl.AsyncPresignedUrlExtension; import software.amazon.awssdk.utils.AttributeMap; import software.amazon.awssdk.utils.CollectionUtils; @@ -91,6 +96,8 @@ public final class DefaultS3CrtAsyncClient extends DelegatingS3AsyncClient imple public static final ExecutionAttribute RESPONSE_FILE_PATH = new ExecutionAttribute<>("responseFilePath"); public static final ExecutionAttribute RESPONSE_FILE_OPTION = new ExecutionAttribute<>("responseFileOption"); + public static final ExecutionAttribute> REQUEST_METRIC_PUBLISHERS = + new ExecutionAttribute<>("requestMetricPublishers"); public static final ExecutionAttribute RESPONSE_FILE_DELETE_ON_FAILURE = new ExecutionAttribute<>("responseFileDeleteOnFailure"); private static final String CRT_CLIENT_CLASSPATH = "software.amazon.awssdk.crt.s3.S3Client"; @@ -139,7 +146,35 @@ public CompletableFuture getObject(GetObjectRequest getObject @Override public CompletableFuture copyObject(CopyObjectRequest copyObjectRequest) { - return copyObjectHelper.copyObject(copyObjectRequest); + // copyObject's sub-requests bypass invokeOperation, so stash once here; CopyObjectHelper propagates the copy's + // override (with the stashed attribute) to each sub-request. + return copyObjectHelper.copyObject(stashRequestMetricPublishers(copyObjectRequest)); + } + + /** + * All operations funnel through here. If the request carries request-level metric publishers, move them off the + * request override (so the inner standard client's resolveMetricPublishers stays empty -> NoOp, no hollow ApiCall) + * and stash them in an execution attribute that the CRT transport reads to publish telemetry to them instead of + * (overriding) the client-level publishers. + */ + @Override + protected CompletableFuture invokeOperation( + T request, Function> operation) { + return operation.apply(stashRequestMetricPublishers(request)); + } + + @SuppressWarnings("unchecked") + static T stashRequestMetricPublishers(T request) { + AwsRequestOverrideConfiguration override = request.overrideConfiguration().orElse(null); + if (override == null || CollectionUtils.isNullOrEmpty(override.metricPublishers())) { + return request; + } + AwsRequestOverrideConfiguration newOverride = + override.toBuilder() + .metricPublishers(Collections.emptyList()) + .putExecutionAttribute(REQUEST_METRIC_PUBLISHERS, override.metricPublishers()) + .build(); + return (T) request.toBuilder().overrideConfiguration(newOverride).build(); } private static S3AsyncClient initializeS3AsyncClient(DefaultS3CrtClientBuilder builder) { @@ -240,7 +275,8 @@ private static S3CrtAsyncHttpClient.Builder initializeS3CrtAsyncHttpClient(Defau .withMaxRetries(builder.retryConfiguration.numRetries()))); } return S3CrtAsyncHttpClient.builder() - .s3ClientConfiguration(nativeClientBuilder.build()); + .s3ClientConfiguration(nativeClientBuilder.build()) + .metricPublishers(builder.metricPublishers); } public static final class DefaultS3CrtClientBuilder implements S3CrtAsyncClientBuilder { @@ -265,6 +301,7 @@ public static final class DefaultS3CrtClientBuilder implements S3CrtAsyncClientB private Long thresholdInBytes; private Executor futureCompletionExecutor; private Boolean disableS3ExpressSessionAuth; + private List metricPublishers; private AttributeMap.Builder advancedOptions = AttributeMap.builder(); @@ -410,6 +447,24 @@ public DefaultS3CrtClientBuilder advancedOptions(Map metricPublishers) { + Validate.paramNotNull(metricPublishers, "metricPublishers"); + Validate.noNullElements(metricPublishers, "metricPublishers must not contain null elements"); + this.metricPublishers = new ArrayList<>(metricPublishers); + return this; + } + + @Override + public DefaultS3CrtClientBuilder addMetricPublisher(MetricPublisher metricPublisher) { + Validate.paramNotNull(metricPublisher, "metricPublisher"); + if (this.metricPublishers == null) { + this.metricPublishers = new ArrayList<>(); + } + this.metricPublishers.add(metricPublisher); + return this; + } + @Override public S3CrtAsyncClient build() { return new DefaultS3CrtAsyncClient(this); @@ -444,6 +499,8 @@ public void afterMarshalling(Context.AfterMarshalling context, executionAttributes.getAttribute(RESPONSE_FILE_PATH)) .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION, executionAttributes.getAttribute(RESPONSE_FILE_OPTION)) + .put(S3InternalSdkHttpExecutionAttribute.METRIC_PUBLISHERS, + executionAttributes.getAttribute(REQUEST_METRIC_PUBLISHERS)) .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_DELETE_ON_FAILURE, executionAttributes.getAttribute(RESPONSE_FILE_DELETE_ON_FAILURE)); @@ -502,10 +559,6 @@ private static void validateOverrideConfiguration(SdkRequest request) { throw new UnsupportedOperationException("Request-level signer override is not supported"); } - if (!CollectionUtils.isNullOrEmpty(overrideConfiguration.metricPublishers())) { - throw new UnsupportedOperationException("Request-level Metric Publishers override is not supported"); - } - if (overrideConfiguration.apiCallAttemptTimeout().isPresent()) { throw new UnsupportedOperationException("Request-level apiCallAttemptTimeout override is not supported"); } diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java index 0ccc0e0f60cc..2c53168ba3ca 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClient.java @@ -15,7 +15,6 @@ package software.amazon.awssdk.services.s3.internal.crt; -import static software.amazon.awssdk.services.s3.crt.S3CrtSdkHttpExecutionAttribute.CRT_PROGRESS_LISTENER; import static software.amazon.awssdk.services.s3.crt.S3CrtSdkHttpExecutionAttribute.METAREQUEST_PAUSE_OBSERVABLE; import static software.amazon.awssdk.services.s3.internal.crt.CrtChecksumUtils.checksumConfig; import static software.amazon.awssdk.services.s3.internal.crt.S3InternalSdkHttpExecutionAttribute.CRT_PAUSE_RESUME_TOKEN; @@ -37,6 +36,7 @@ import java.nio.file.Path; import java.time.Duration; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; @@ -62,6 +62,7 @@ import software.amazon.awssdk.http.SdkHttpRequest; import software.amazon.awssdk.http.async.AsyncExecuteRequest; import software.amazon.awssdk.http.async.SdkAsyncHttpClient; +import software.amazon.awssdk.metrics.MetricPublisher; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.utils.AttributeMap; import software.amazon.awssdk.utils.NumericUtils; @@ -74,14 +75,20 @@ @SdkInternalApi public final class S3CrtAsyncHttpClient implements SdkAsyncHttpClient { + static final String CLIENT_NAME = "s3crt"; + private final S3Client crtS3Client; private final S3NativeClientConfiguration s3NativeClientConfiguration; private final S3ClientOptions s3ClientOptions; + private final List metricPublishers; + private final S3CrtMetricPublisherDispatcher metricPublisherDispatcher; private S3CrtAsyncHttpClient(Builder builder) { s3NativeClientConfiguration = builder.clientConfiguration; this.s3ClientOptions = createS3ClientOption(); + this.metricPublishers = copyMetricPublishers(builder.metricPublishers); + this.metricPublisherDispatcher = new S3CrtMetricPublisherDispatcher(); this.crtS3Client = new S3Client(s3ClientOptions); } @@ -91,9 +98,17 @@ private S3CrtAsyncHttpClient(Builder builder) { Builder builder) { s3NativeClientConfiguration = builder.clientConfiguration; s3ClientOptions = createS3ClientOption(); + this.metricPublishers = copyMetricPublishers(builder.metricPublishers); + this.metricPublisherDispatcher = new S3CrtMetricPublisherDispatcher(); this.crtS3Client = crtS3Client; } + private static List copyMetricPublishers(List metricPublishers) { + return metricPublishers == null + ? Collections.emptyList() + : Collections.unmodifiableList(new ArrayList<>(metricPublishers)); + } + @SdkTestInternalApi public S3ClientOptions s3ClientOptions() { return s3ClientOptions; @@ -164,12 +179,22 @@ public CompletableFuture execute(AsyncExecuteRequest asyncRequest) { S3MetaRequestOptions.ResponseFileOption responseFileOption = httpExecutionAttributes.getAttribute(RESPONSE_FILE_OPTION); Boolean responseFileDeleteOnFailure = httpExecutionAttributes.getAttribute(RESPONSE_FILE_DELETE_ON_FAILURE); + List requestMetricPublishers = + httpExecutionAttributes.getAttribute(S3InternalSdkHttpExecutionAttribute.METRIC_PUBLISHERS); + List effectivePublishers = resolveEffectiveMetricPublishers(requestMetricPublishers, + metricPublishers); + S3CrtResponseHandlerAdapter responseHandler = - new S3CrtResponseHandlerAdapter( - executeFuture, - asyncRequest.responseHandler(), - httpExecutionAttributes.getAttribute(CRT_PROGRESS_LISTENER), - s3MetaRequestFuture); + S3CrtResponseHandlerAdapter.builder() + .executeFuture(executeFuture) + .responseHandler(asyncRequest.responseHandler()) + .httpExecutionAttributes(httpExecutionAttributes) + .metaRequestFuture(s3MetaRequestFuture) + .metricPublishers(effectivePublishers) + .metricPublisherDispatcher(metricPublisherDispatcher) + .maxConcurrency(effectivePublishers.isEmpty() ? + null : crtS3Client.getMaxActiveConnections()) + .build(); URI endpoint = getEndpoint(uri); @@ -233,6 +258,18 @@ public CompletableFuture execute(AsyncExecuteRequest asyncRequest) { return executeFuture; } + /** + * Request-level publishers (if any were set on the request override) take precedence over the client-level + * publishers; otherwise the client-level publishers are used. + */ + static List resolveEffectiveMetricPublishers(List requestLevel, + List clientLevel) { + if (requestLevel != null && !requestLevel.isEmpty()) { + return requestLevel; + } + return clientLevel; + } + private static boolean isResponseFileAlreadyExistsError(Throwable throwable, Path responseFilePath) { return responseFilePath != null && throwable instanceof CrtRuntimeException @@ -267,7 +304,7 @@ private static URI getEndpoint(URI uri) { @Override public String clientName() { - return "s3crt"; + return CLIENT_NAME; } private static S3MetaRequestOptions.MetaRequestType requestType(String operationName) { @@ -312,8 +349,17 @@ private static HttpRequest toCrtRequest(AsyncExecuteRequest asyncRequest, Comple @Override public void close() { - s3NativeClientConfiguration.close(); - crtS3Client.close(); + CompletableFuture shutdownCompleteFuture = crtS3Client.getShutdownCompleteFuture(); + shutdownCompleteFuture.whenComplete((ignored, throwable) -> metricPublisherDispatcher.startAsyncClose()); + try { + metricPublisherDispatcher.flushPendingMetrics(); + } finally { + try { + s3NativeClientConfiguration.close(); + } finally { + crtS3Client.close(); + } + } } public static Builder builder() { @@ -322,12 +368,18 @@ public static Builder builder() { public static final class Builder implements SdkAsyncHttpClient.Builder { private S3NativeClientConfiguration clientConfiguration; + private List metricPublishers; public Builder s3ClientConfiguration(S3NativeClientConfiguration clientConfiguration) { this.clientConfiguration = clientConfiguration; return this; } + public Builder metricPublishers(List metricPublishers) { + this.metricPublishers = metricPublishers; + return this; + } + @Override public SdkAsyncHttpClient build() { return new S3CrtAsyncHttpClient(this); diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java new file mode 100644 index 000000000000..8350ec81b9f2 --- /dev/null +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcher.java @@ -0,0 +1,280 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.services.s3.internal.crt; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import software.amazon.awssdk.annotations.SdkInternalApi; +import software.amazon.awssdk.annotations.SdkTestInternalApi; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; +import software.amazon.awssdk.utils.Logger; +import software.amazon.awssdk.utils.SdkAutoCloseable; +import software.amazon.awssdk.utils.ThreadFactoryBuilder; + +/** + * Publishes CRT metric collections away from CRT native callback threads. + * + *

Collections are submitted without blocking to a single bounded queue and published in acceptance order by one + * daemon worker. A slow or blocked publisher can delay the other publishers configured for the client, but it cannot + * block a CRT callback thread or cause the queue to grow without bound. + */ +@SdkInternalApi +final class S3CrtMetricPublisherDispatcher implements SdkAutoCloseable { + private static final Logger log = Logger.loggerFor(S3CrtMetricPublisherDispatcher.class); + // Uses CloudWatchMetricPublisher's 128-entry queue as an existing SDK precedent. This limits each client + // to retaining at most 128 queued metric collections awaiting publication. + private static final int DEFAULT_QUEUE_CAPACITY = 128; + // Allows time for queued metrics to be published while preventing a slow publisher from delaying shutdown + // indefinitely. + private static final Duration DEFAULT_CLOSE_TIMEOUT = Duration.ofSeconds(5); + // Delay between attempts to enqueue a flush marker when the queue is full; retries stop when + // DEFAULT_CLOSE_TIMEOUT expires. + private static final long FLUSH_RETRY_DELAY_NANOS = TimeUnit.MILLISECONDS.toNanos(10); + private static final ThreadFactory CLOSE_THREAD_FACTORY = + new ThreadFactoryBuilder() + .daemonThreads(true) + .threadNamePrefix("s3-crt-metric-publisher-cleanup") + .build(); + + private final ThreadPoolExecutor executor; + private final int queueCapacity; + private final Duration closeTimeout; + private final AtomicLong queueFullDropCount = new AtomicLong(); + private final AtomicLong closedDispatcherDropCount = new AtomicLong(); + private final AtomicBoolean dispatchAttempted = new AtomicBoolean(); + private final AtomicBoolean asyncCloseStarted = new AtomicBoolean(); + private final AtomicBoolean closed = new AtomicBoolean(); + + S3CrtMetricPublisherDispatcher() { + this(DEFAULT_QUEUE_CAPACITY, DEFAULT_CLOSE_TIMEOUT); + } + + @SdkTestInternalApi + S3CrtMetricPublisherDispatcher(int queueCapacity, Duration closeTimeout) { + if (queueCapacity <= 0) { + throw new IllegalArgumentException("queueCapacity must be greater than zero"); + } + if (closeTimeout.isNegative()) { + throw new IllegalArgumentException("closeTimeout must not be negative"); + } + this.queueCapacity = queueCapacity; + this.closeTimeout = closeTimeout; + this.executor = new ThreadPoolExecutor( + 1, 1, + 0, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(queueCapacity), + new ThreadFactoryBuilder() + .daemonThreads(true) + .threadNamePrefix("s3-crt-metric-publisher") + .build(), + this::handleRejectedTask); + } + + /** + * Dispatches one collection to all configured publishers as a single queue task. Queue admission is non-blocking, + * and publisher code is only invoked by the executor worker. + */ + void dispatch(MetricCollection collection, List publishers) { + dispatchAttempted.set(true); + executor.execute(new PublishTask(collection, publishers)); + } + + /** + * Waits for collections accepted before the marker task to be published, up to the configured close timeout. + */ + void flushPendingMetrics() { + if (!dispatchAttempted.get() || executor.isShutdown()) { + return; + } + + long deadlineNanos = System.nanoTime() + closeTimeout.toNanos(); + FlushMarkerTask flushMarker = submitFlushMarker(deadlineNanos); + if (flushMarker == null) { + return; + } + + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { + logFlushTimeout(); + return; + } + + try { + if (!flushMarker.await(remainingNanos)) { + logFlushTimeout(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.debug(() -> "Interrupted while waiting for queued CRT S3 metrics to be published."); + } + } + + private void publish(MetricCollection collection, List publishers) { + for (MetricPublisher publisher : publishers) { + try { + publisher.publish(collection); + } catch (RuntimeException e) { + log.debug(() -> "Failed to publish CRT S3 metrics through " + publisher.getClass().getName(), e); + } + } + } + + private FlushMarkerTask submitFlushMarker(long deadlineNanos) { + while (!executor.isShutdown()) { + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { + logFlushTimeout(); + return null; + } + + FlushMarkerTask flushMarker = new FlushMarkerTask(); + executor.execute(flushMarker); + if (!flushMarker.rejected()) { + return flushMarker; + } + + try { + TimeUnit.NANOSECONDS.sleep(Math.min(remainingNanos, FLUSH_RETRY_DELAY_NANOS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.debug(() -> "Interrupted while waiting to flush queued CRT S3 metrics."); + return null; + } + } + return null; + } + + private void handleRejectedTask(Runnable task, ThreadPoolExecutor pool) { + if (task instanceof FlushMarkerTask) { + ((FlushMarkerTask) task).reject(); + } else if (pool.isShutdown()) { + recordClosedDispatcherDrop(); + } else { + recordQueueFullDrop(); + } + } + + private void logFlushTimeout() { + log.debug(() -> "Timed out waiting for queued CRT S3 metrics to be published."); + } + + private void recordQueueFullDrop() { + long dropped = queueFullDropCount.incrementAndGet(); + if (isPowerOfTwo(dropped)) { + log.debug(() -> "Dropped CRT S3 metric collections because the metric publisher queue is full. " + + "Queue capacity: " + queueCapacity + ", total dropped: " + dropped + "."); + } + } + + private void recordClosedDispatcherDrop() { + long dropped = closedDispatcherDropCount.incrementAndGet(); + if (isPowerOfTwo(dropped)) { + log.debug(() -> "Dropped CRT S3 metric collections because the metric publisher dispatcher is closed. " + + "Total dropped after close: " + dropped + "."); + } + } + + void startAsyncClose() { + if (asyncCloseStarted.compareAndSet(false, true)) { + if (!dispatchAttempted.get()) { + close(); + return; + } + CLOSE_THREAD_FACTORY.newThread(this::close).start(); + } + } + + @Override + public void close() { + if (!closed.compareAndSet(false, true)) { + return; + } + executor.shutdown(); + + boolean terminated = false; + try { + terminated = executor.awaitTermination(closeTimeout.toNanos(), TimeUnit.NANOSECONDS); + } catch (InterruptedException e) { + int queuedCollectionCount = forceShutdown(); + Thread.currentThread().interrupt(); + log.debug(() -> "Interrupted while draining CRT S3 metric publisher queue. Undelivered queued collections: " + + queuedCollectionCount + "."); + return; + } + + if (!terminated) { + int queuedCollectionCount = forceShutdown(); + log.debug(() -> "Timed out draining CRT S3 metric publisher queue. Undelivered queued collections: " + + queuedCollectionCount + "."); + } + } + + private int forceShutdown() { + return (int) executor.shutdownNow() + .stream() + .filter(PublishTask.class::isInstance) + .count(); + } + + private static boolean isPowerOfTwo(long value) { + return (value & (value - 1)) == 0; + } + + private final class PublishTask implements Runnable { + private final MetricCollection collection; + private final List publishers; + + private PublishTask(MetricCollection collection, List publishers) { + this.collection = collection; + this.publishers = publishers; + } + + @Override + public void run() { + publish(collection, publishers); + } + } + + private static final class FlushMarkerTask implements Runnable { + private final CountDownLatch completion = new CountDownLatch(1); + private volatile boolean rejected; + + @Override + public void run() { + completion.countDown(); + } + + private void reject() { + rejected = true; + } + + private boolean rejected() { + return rejected; + } + + private boolean await(long timeoutNanos) throws InterruptedException { + return completion.await(timeoutNanos, TimeUnit.NANOSECONDS); + } + } +} diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java index ec8ad6a1cc21..c420d72fbe9b 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java @@ -17,10 +17,14 @@ import static software.amazon.awssdk.core.http.HttpResponseHandler.X_AMZN_REQUEST_ID_HEADER_ALTERNATE; import static software.amazon.awssdk.core.http.HttpResponseHandler.X_AMZ_ID_2_HEADER; +import static software.amazon.awssdk.services.s3.crt.S3CrtSdkHttpExecutionAttribute.CRT_PROGRESS_LISTENER; import static software.amazon.awssdk.utils.FunctionalUtils.runAndLogError; +import static software.amazon.awssdk.utils.NumericUtils.saturatedCast; +import java.net.URI; import java.nio.ByteBuffer; import java.time.Duration; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -29,23 +33,34 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.LongSupplier; +import java.util.function.Supplier; import software.amazon.awssdk.annotations.SdkInternalApi; -import software.amazon.awssdk.annotations.SdkTestInternalApi; import software.amazon.awssdk.awscore.exception.AwsErrorDetails; import software.amazon.awssdk.awscore.exception.AwsServiceException; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.core.async.listener.PublisherListener; import software.amazon.awssdk.core.exception.SdkClientException; +import software.amazon.awssdk.core.metrics.CoreMetric; import software.amazon.awssdk.crt.CRT; +import software.amazon.awssdk.crt.CrtRuntimeException; import software.amazon.awssdk.crt.http.HttpHeader; +import software.amazon.awssdk.crt.http.HttpManagerMetrics; import software.amazon.awssdk.crt.s3.S3FinishedResponseContext; import software.amazon.awssdk.crt.s3.S3MetaRequestProgress; import software.amazon.awssdk.crt.s3.S3MetaRequestResponseHandler; +import software.amazon.awssdk.crt.s3.S3RequestMetrics; import software.amazon.awssdk.http.AbortableInputStream; +import software.amazon.awssdk.http.HttpMetric; import software.amazon.awssdk.http.HttpStatusFamily; +import software.amazon.awssdk.http.SdkHttpExecutionAttributes; import software.amazon.awssdk.http.SdkHttpFullResponse; import software.amazon.awssdk.http.SdkHttpResponse; import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricCollector; +import software.amazon.awssdk.metrics.MetricPublisher; +import software.amazon.awssdk.metrics.SdkMetric; import software.amazon.awssdk.services.s3.model.S3Exception; import software.amazon.awssdk.utils.Logger; import software.amazon.awssdk.utils.async.SimplePublisher; @@ -67,6 +82,9 @@ public final class S3CrtResponseHandlerAdapter implements S3MetaRequestResponseH private final PublisherListener progressListener; private final Duration s3MetaRequestTimeout; + private final List metricPublishers; + private final Integer maxConcurrency; + private final S3CrtMetricPublisherDispatcher metricPublisherDispatcher; private volatile boolean responseHandlingInitiated; @@ -74,24 +92,17 @@ public final class S3CrtResponseHandlerAdapter implements S3MetaRequestResponseH // the later cancel-derived AWS_ERROR_S3_CANCELED must not re-notify. See #6715. private final AtomicBoolean responseHandlerNotified = new AtomicBoolean(false); - public S3CrtResponseHandlerAdapter(CompletableFuture executeFuture, - SdkAsyncHttpResponseHandler responseHandler, - PublisherListener progressListener, - CompletableFuture metaRequestFuture) { - this(executeFuture, responseHandler, progressListener, metaRequestFuture, META_REQUEST_TIMEOUT); - } - - @SdkTestInternalApi - public S3CrtResponseHandlerAdapter(CompletableFuture executeFuture, - SdkAsyncHttpResponseHandler responseHandler, - PublisherListener progressListener, - CompletableFuture metaRequestFuture, - Duration s3MetaRequestTimeout) { - this.resultFuture = executeFuture; - this.metaRequestFuture = metaRequestFuture; - this.responseHandler = responseHandler; + private S3CrtResponseHandlerAdapter(Builder builder) { + this.resultFuture = builder.executeFuture; + this.metaRequestFuture = builder.metaRequestFuture; + this.responseHandler = builder.responseHandler; + PublisherListener progressListener = + builder.httpExecutionAttributes.getAttribute(CRT_PROGRESS_LISTENER); this.progressListener = progressListener == null ? new NoOpPublisherListener() : progressListener; - this.s3MetaRequestTimeout = s3MetaRequestTimeout; + this.metricPublishers = builder.metricPublishers; + this.maxConcurrency = builder.maxConcurrency; + this.s3MetaRequestTimeout = builder.s3MetaRequestTimeout; + this.metricPublisherDispatcher = builder.metricPublisherDispatcher; // Registered last: the callback reads responseHandler and s3MetaRequestTimeout, so all fields must be assigned // first in case resultFuture is already complete and the callback runs synchronously here. @@ -112,6 +123,68 @@ public S3CrtResponseHandlerAdapter(CompletableFuture executeFuture, }); } + public static Builder builder() { + return new Builder(); + } + + public static final class Builder { + private CompletableFuture executeFuture; + private SdkAsyncHttpResponseHandler responseHandler; + private SdkHttpExecutionAttributes httpExecutionAttributes; + private CompletableFuture metaRequestFuture; + private Duration s3MetaRequestTimeout = META_REQUEST_TIMEOUT; + private List metricPublishers = Collections.emptyList(); + private Integer maxConcurrency; + private S3CrtMetricPublisherDispatcher metricPublisherDispatcher; + + private Builder() { + } + + public Builder executeFuture(CompletableFuture executeFuture) { + this.executeFuture = executeFuture; + return this; + } + + public Builder responseHandler(SdkAsyncHttpResponseHandler responseHandler) { + this.responseHandler = responseHandler; + return this; + } + + public Builder httpExecutionAttributes(SdkHttpExecutionAttributes httpExecutionAttributes) { + this.httpExecutionAttributes = httpExecutionAttributes; + return this; + } + + public Builder metaRequestFuture(CompletableFuture metaRequestFuture) { + this.metaRequestFuture = metaRequestFuture; + return this; + } + + public Builder s3MetaRequestTimeout(Duration s3MetaRequestTimeout) { + this.s3MetaRequestTimeout = s3MetaRequestTimeout; + return this; + } + + public Builder metricPublishers(List metricPublishers) { + this.metricPublishers = metricPublishers; + return this; + } + + public Builder maxConcurrency(Integer maxConcurrency) { + this.maxConcurrency = maxConcurrency; + return this; + } + + Builder metricPublisherDispatcher(S3CrtMetricPublisherDispatcher metricPublisherDispatcher) { + this.metricPublisherDispatcher = metricPublisherDispatcher; + return this; + } + + public S3CrtResponseHandlerAdapter build() { + return new S3CrtResponseHandlerAdapter(this); + } + } + private S3MetaRequestWrapper s3MetaRequest() { try { return metaRequestFuture.get(s3MetaRequestTimeout.toMillis(), TimeUnit.MILLISECONDS); @@ -332,6 +405,107 @@ public void onProgress(S3MetaRequestProgress progress) { this.progressListener.subscriberOnNext(progress); } + /** + * Invoked by CRT once per underlying HTTP request (e.g. each ranged part GET / upload part) with that request's + * telemetry. We publish each as its own nested ApiCall -> ApiCallAttempt -> HttpClient MetricCollection, mirroring + * the node placement of the standard client so all metric publishers render it consistently. Runs on a CRT native + * thread; it reads and copies the metrics object before dispatching the completed collection to the client-scoped + * bounded queue and single worker, and never throws back into the native callback. + */ + @Override + public void onTelemetry(S3RequestMetrics requestMetrics) { + if (metricPublishers.isEmpty()) { + return; + } + try { + MetricCollector apiCall = MetricCollector.create("ApiCall"); + report(apiCall, CoreMetric.SERVICE_ID, requestMetrics::getServiceId); + report(apiCall, CoreMetric.OPERATION_NAME, requestMetrics::getOperationName); + report(apiCall, CoreMetric.API_CALL_SUCCESSFUL, requestMetrics::isApiCallSuccessful); + report(apiCall, CoreMetric.RETRY_COUNT, requestMetrics::getRetryCount); + reportDuration(apiCall, CoreMetric.API_CALL_DURATION, requestMetrics::getApiCallDurationNs); + reportServiceEndpoint(apiCall, requestMetrics::getServiceEndpoint); + + // TODO: Map CredentialsFetchDuration, ReadThroughput, and WriteThroughput when CRT telemetry exposes them. + MetricCollector attempt = apiCall.createChild("ApiCallAttempt"); + reportDuration(attempt, CoreMetric.SIGNING_DURATION, requestMetrics::getSigningDurationNs); + reportDuration(attempt, CoreMetric.SERVICE_CALL_DURATION, requestMetrics::getServiceCallDurationNs); + reportDuration(attempt, CoreMetric.BACKOFF_DELAY_DURATION, requestMetrics::getBackoffDelayDurationNs); + report(attempt, CoreMetric.AWS_REQUEST_ID, requestMetrics::getAwsRequestId); + report(attempt, CoreMetric.AWS_EXTENDED_REQUEST_ID, requestMetrics::getAwsExtendedRequestId); + reportDuration(attempt, CoreMetric.TIME_TO_FIRST_BYTE, requestMetrics::getTimeToFirstByte); + reportDuration(attempt, CoreMetric.TIME_TO_LAST_BYTE, requestMetrics::getTimeToLastByte); + reportResponseStatus(attempt, requestMetrics.getResponseStatus()); + + MetricCollector httpClient = attempt.createChild("HttpClient"); + httpClient.reportMetric(HttpMetric.HTTP_CLIENT_NAME, S3CrtAsyncHttpClient.CLIENT_NAME); + if (maxConcurrency != null) { + httpClient.reportMetric(HttpMetric.MAX_CONCURRENCY, maxConcurrency); + } + reportDuration(httpClient, HttpMetric.CONCURRENCY_ACQUIRE_DURATION, + requestMetrics::getConnectionAcquisitionDurationNs); + reportHttpManagerMetrics(httpClient, requestMetrics.getHttpManagerMetrics()); + + MetricCollection collection = apiCall.collect(); + metricPublisherDispatcher.dispatch(collection, metricPublishers); + } catch (RuntimeException e) { + log.debug(() -> "Failed to collect CRT S3 request metrics", e); + } + } + + private void reportServiceEndpoint(MetricCollector collector, Supplier getter) { + try { + String endpoint = getter.get(); + if (endpoint != null) { + collector.reportMetric(CoreMetric.SERVICE_ENDPOINT, URI.create(endpoint)); + } + } catch (CrtRuntimeException | IllegalArgumentException e) { + log.trace(() -> "CRT metric " + CoreMetric.SERVICE_ENDPOINT.name() + " unavailable: " + e.getMessage()); + } + } + + private void reportResponseStatus(MetricCollector collector, int responseStatus) { + if (responseStatus > 0) { + collector.reportMetric(HttpMetric.HTTP_STATUS_CODE, responseStatus); + } + } + + private void reportHttpManagerMetrics(MetricCollector collector, HttpManagerMetrics managerMetrics) { + if (managerMetrics == null) { + return; + } + collector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, + saturatedCast(managerMetrics.getAvailableConcurrency())); + collector.reportMetric(HttpMetric.LEASED_CONCURRENCY, + saturatedCast(managerMetrics.getLeasedConcurrency())); + collector.reportMetric(HttpMetric.PENDING_CONCURRENCY_ACQUIRES, + saturatedCast(managerMetrics.getPendingConcurrencyAcquires())); + } + + /** + * Reports a metric from a CRT getter, guarding against {@link CrtRuntimeException} (error code 14358, + * AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE) which CRT throws when the value is not available for this request. + * A missing/unavailable value skips only that one metric. + */ + private void report(MetricCollector collector, SdkMetric metric, Supplier getter) { + try { + T value = getter.get(); + if (value != null) { + collector.reportMetric(metric, value); + } + } catch (CrtRuntimeException e) { + log.trace(() -> "CRT metric " + metric.name() + " unavailable: " + e.getMessage()); + } + } + + private void reportDuration(MetricCollector collector, SdkMetric metric, LongSupplier nanos) { + try { + collector.reportMetric(metric, Duration.ofNanos(nanos.getAsLong())); + } catch (CrtRuntimeException e) { + log.trace(() -> "CRT metric " + metric.name() + " unavailable: " + e.getMessage()); + } + } + private static SdkHttpResponse.Builder populateSdkHttpResponse(SdkHttpResponse.Builder respBuilder, int statusCode, HttpHeader[] headers) { if (headers != null) { diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3InternalSdkHttpExecutionAttribute.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3InternalSdkHttpExecutionAttribute.java index 7ddf40998a48..fd024cbea4fd 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3InternalSdkHttpExecutionAttribute.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3InternalSdkHttpExecutionAttribute.java @@ -16,6 +16,7 @@ package software.amazon.awssdk.services.s3.internal.crt; import java.nio.file.Path; +import java.util.List; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.core.checksums.RequestChecksumCalculation; import software.amazon.awssdk.core.checksums.ResponseChecksumValidation; @@ -23,6 +24,7 @@ import software.amazon.awssdk.crt.s3.ResumeToken; import software.amazon.awssdk.crt.s3.S3MetaRequestOptions; import software.amazon.awssdk.http.SdkHttpExecutionAttribute; +import software.amazon.awssdk.metrics.MetricPublisher; import software.amazon.awssdk.regions.Region; @SdkInternalApi @@ -70,6 +72,14 @@ public final class S3InternalSdkHttpExecutionAttribute extends SdkHttpExecuti public static final S3InternalSdkHttpExecutionAttribute CRT_CREDENTIALS_PROVIDER_ADAPTER = new S3InternalSdkHttpExecutionAttribute<>(CrtCredentialsProviderAdapter.class); + /** + * The request-level metric publishers passed to the CRT transport. The transport resolves these against its configured + * client-level publishers before constructing the response handler. + */ + @SuppressWarnings("unchecked") + public static final S3InternalSdkHttpExecutionAttribute> METRIC_PUBLISHERS = + new S3InternalSdkHttpExecutionAttribute<>((Class>) (Class) List.class); + private S3InternalSdkHttpExecutionAttribute(Class valueClass) { super(valueClass); } diff --git a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/CopyObjectHelper.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/CopyObjectHelper.java index 4075ef8961fa..4d00a0dbdde9 100644 --- a/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/CopyObjectHelper.java +++ b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/multipart/CopyObjectHelper.java @@ -33,6 +33,7 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; import software.amazon.awssdk.annotations.SdkInternalApi; +import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration; import software.amazon.awssdk.core.ResponseBytes; import software.amazon.awssdk.core.async.AsyncRequestBody; import software.amazon.awssdk.core.async.AsyncResponseTransformer; @@ -191,9 +192,11 @@ private void copyInParts(CopyObjectRequest copyObjectRequest, } private CompletableFuture> fetchSourceTagging(CopyObjectRequest copyObjectRequest, String sourceVersionId) { + AwsRequestOverrideConfiguration override = copyObjectRequest.overrideConfiguration().orElse(null); return s3AsyncClient.getObjectTagging(r -> r.bucket(copyObjectRequest.sourceBucket()) .key(copyObjectRequest.sourceKey()) - .versionId(sourceVersionId)) + .versionId(sourceVersionId) + .overrideConfiguration(override)) .thenApply(response -> CollectionUtils.isNullOrEmpty(response.tagSet()) ? null : response.tagSet()); } @@ -207,10 +210,12 @@ private CompletableFuture> fetchSourceAnnotations(CopyObject Map annotationBodies = new ConcurrentHashMap<>(); Queue> fetchFutures = new ConcurrentLinkedQueue<>(); CompletableFuture> result = new CompletableFuture<>(); + AwsRequestOverrideConfiguration override = copyObjectRequest.overrideConfiguration().orElse(null); s3AsyncClient.listObjectAnnotationsPaginator(r -> r.bucket(copyObjectRequest.sourceBucket()) .key(copyObjectRequest.sourceKey()) - .versionId(sourceVersionId)) + .versionId(sourceVersionId) + .overrideConfiguration(override)) .subscribe(response -> fetchAnnotationBodies(copyObjectRequest, sourceVersionId, response, annotationBodies, fetchFutures)) .whenComplete((v, t) -> { @@ -247,10 +252,12 @@ private void fetchAnnotationBodies(CopyObjectRequest copyObjectRequest, private CompletableFuture fetchSingleAnnotation(CopyObjectRequest copyObjectRequest, String sourceVersionId, String annotationName) { + AwsRequestOverrideConfiguration override = copyObjectRequest.overrideConfiguration().orElse(null); return s3AsyncClient.getObjectAnnotation(r -> r.bucket(copyObjectRequest.sourceBucket()) .key(copyObjectRequest.sourceKey()) .versionId(sourceVersionId) - .annotationName(annotationName), + .annotationName(annotationName) + .overrideConfiguration(override), AsyncResponseTransformer.toBytes()) .thenApply(ResponseBytes::asByteArray); } @@ -260,10 +267,12 @@ private CompletableFuture putTaggingPostComplet CompleteMultipartUploadResponse completeResponse, List tags) { + AwsRequestOverrideConfiguration override = copyObjectRequest.overrideConfiguration().orElse(null); return s3AsyncClient.putObjectTagging(r -> r.bucket(copyObjectRequest.destinationBucket()) .key(copyObjectRequest.destinationKey()) .versionId(completeResponse.versionId()) - .tagging(Tagging.builder().tagSet(tags).build())) + .tagging(Tagging.builder().tagSet(tags).build()) + .overrideConfiguration(override)) .thenApply(r -> completeResponse); } @@ -279,6 +288,7 @@ private CompletableFuture writeAnnotationsToDes String destVersionId = completeResponse.versionId(); String destETag = completeResponse.eTag(); + AwsRequestOverrideConfiguration override = copyObjectRequest.overrideConfiguration().orElse(null); log.debug(() -> String.format("Writing %d annotations to destination object", annotations.size())); @@ -292,7 +302,8 @@ private CompletableFuture writeAnnotationsToDes .key(copyObjectRequest.destinationKey()) .versionId(destVersionId) .objectIfMatch(destETag) - .annotationName(entry.getKey()), + .annotationName(entry.getKey()) + .overrideConfiguration(override), AsyncRequestBody.fromBytes(entry.getValue())) .thenRun(() -> succeededAnnotations.add(entry.getKey()))); } diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/CopyObjectHelperTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/CopyObjectHelperTest.java index ce8e46b686a9..9960242bd2b6 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/CopyObjectHelperTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/CopyObjectHelperTest.java @@ -62,6 +62,7 @@ import software.amazon.awssdk.services.s3.model.CopyPartResult; import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest; import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse; +import software.amazon.awssdk.services.s3.model.GetObjectTaggingRequest; import software.amazon.awssdk.services.s3.model.GetObjectTaggingResponse; import software.amazon.awssdk.services.s3.model.HeadObjectRequest; import software.amazon.awssdk.services.s3.model.HeadObjectResponse; @@ -77,10 +78,13 @@ import software.amazon.awssdk.core.async.AsyncResponseTransformer; import software.amazon.awssdk.services.s3.model.AnnotationDirective; import software.amazon.awssdk.services.s3.model.AnnotationEntry; +import software.amazon.awssdk.services.s3.model.GetObjectAnnotationRequest; import software.amazon.awssdk.services.s3.model.GetObjectAnnotationResponse; import software.amazon.awssdk.services.s3.model.ListObjectAnnotationsRequest; import software.amazon.awssdk.services.s3.model.ListObjectAnnotationsResponse; +import software.amazon.awssdk.services.s3.model.PutObjectAnnotationRequest; import software.amazon.awssdk.services.s3.model.PutObjectAnnotationResponse; +import software.amazon.awssdk.services.s3.model.PutObjectTaggingRequest; import software.amazon.awssdk.services.s3.model.PutObjectTaggingResponse; import software.amazon.awssdk.services.s3.model.S3Exception; import software.amazon.awssdk.services.s3.paginators.ListObjectAnnotationsPublisher; @@ -986,6 +990,85 @@ void multiPartCopy_cancelReturnFuture_shouldCancelCreateMpuFuture() { assertThat(createMpuFuture).isCompletedExceptionally(); } + @SuppressWarnings("unchecked") + @Test + void multiPartCopy_taggingDirectiveCopy_propagatesOverrideToTaggingSubRequests() { + AwsRequestOverrideConfiguration override = + AwsRequestOverrideConfiguration.builder().putHeader("x-custom", "value").build(); + + stubSuccessfulHeadObjectCall(4000L); + when(s3AsyncClient.getObjectTagging(any(Consumer.class))) + .thenReturn(CompletableFuture.completedFuture(GetObjectTaggingResponse.builder() + .tagSet(Arrays.asList(Tag.builder().key("k").value("v").build())).build())); + stubSuccessfulCreateMulipartCall(); + stubSuccessfulUploadPartCopyCalls(); + stubSuccessfulCompleteMultipartCall(); + when(s3AsyncClient.putObjectTagging(any(Consumer.class))) + .thenReturn(CompletableFuture.completedFuture(PutObjectTaggingResponse.builder().build())); + + copyHelper.copyObject(copyRequestBuilder().taggingDirective(TaggingDirective.COPY) + .overrideConfiguration(override) + .build()).join(); + + ArgumentCaptor> getCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(s3AsyncClient).getObjectTagging(getCaptor.capture()); + GetObjectTaggingRequest.Builder getBuilder = GetObjectTaggingRequest.builder(); + getCaptor.getValue().accept(getBuilder); + assertThat(getBuilder.build().overrideConfiguration()).contains(override); + + ArgumentCaptor> putCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(s3AsyncClient).putObjectTagging(putCaptor.capture()); + PutObjectTaggingRequest.Builder putBuilder = PutObjectTaggingRequest.builder(); + putCaptor.getValue().accept(putBuilder); + assertThat(putBuilder.build().overrideConfiguration()).contains(override); + } + + @SuppressWarnings("unchecked") + @Test + void multiPartCopy_annotationDirectiveCopy_propagatesOverrideToAnnotationSubRequests() { + AwsRequestOverrideConfiguration override = + AwsRequestOverrideConfiguration.builder().putHeader("x-custom", "value").build(); + + stubSuccessfulHeadObjectCall(4000L); + // one source annotation so getObjectAnnotation (source) and putObjectAnnotation (dest) are actually invoked + when(s3AsyncClient.listObjectAnnotations(any(ListObjectAnnotationsRequest.class))) + .thenReturn(CompletableFuture.completedFuture(ListObjectAnnotationsResponse.builder() + .annotations(AnnotationEntry.builder().annotationName("anno1").build()).build())); + when(s3AsyncClient.getObjectAnnotation(any(Consumer.class), any(AsyncResponseTransformer.class))) + .thenReturn(CompletableFuture.completedFuture( + ResponseBytes.fromByteArray(GetObjectAnnotationResponse.builder().build(), "body".getBytes()))); + stubSuccessfulCreateMulipartCall(); + stubSuccessfulUploadPartCopyCalls(); + stubSuccessfulCompleteMultipartCall(); + when(s3AsyncClient.putObjectAnnotation(any(Consumer.class), any(AsyncRequestBody.class))) + .thenReturn(CompletableFuture.completedFuture(PutObjectAnnotationResponse.builder().build())); + + copyHelper.copyObject(copyRequestBuilder().annotationDirective(AnnotationDirective.COPY) + .overrideConfiguration(override) + .build()).join(); + + // ListObjectAnnotations (source) carries the copy's override. + ArgumentCaptor> listCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(s3AsyncClient).listObjectAnnotationsPaginator(listCaptor.capture()); + ListObjectAnnotationsRequest.Builder listBuilder = ListObjectAnnotationsRequest.builder(); + listCaptor.getValue().accept(listBuilder); + assertThat(listBuilder.build().overrideConfiguration()).contains(override); + + // GetObjectAnnotation (source) carries the copy's override. + ArgumentCaptor> getCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(s3AsyncClient).getObjectAnnotation(getCaptor.capture(), any(AsyncResponseTransformer.class)); + GetObjectAnnotationRequest.Builder getBuilder = GetObjectAnnotationRequest.builder(); + getCaptor.getValue().accept(getBuilder); + assertThat(getBuilder.build().overrideConfiguration()).contains(override); + + // PutObjectAnnotation (dest) carries the copy's override. + ArgumentCaptor> putCaptor = ArgumentCaptor.forClass(Consumer.class); + verify(s3AsyncClient).putObjectAnnotation(putCaptor.capture(), any(AsyncRequestBody.class)); + PutObjectAnnotationRequest.Builder putBuilder = PutObjectAnnotationRequest.builder(); + putCaptor.getValue().accept(putBuilder); + assertThat(putBuilder.build().overrideConfiguration()).contains(override); + } + @Test void singlePartCopy_withProgressListener_reportsLengthBytesAndCompletion() { PublisherListener progressListener = mock(PublisherListener.class); diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClientTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClientTest.java index 3a5edc98513e..101abdd7ab43 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClientTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/DefaultS3CrtAsyncClientTest.java @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.nio.file.Path; +import java.util.Collections; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -155,6 +156,13 @@ void invalidConfig_shouldThrowException(long value) { "positive"); } + @Test + void metricPublishers_withNullElement_throws() { + assertThatThrownBy(() -> S3AsyncClient.crtBuilder().metricPublishers(Collections.singletonList(null))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("metricPublishers must not contain null elements"); + } + @Test void crtClient_with_crossRegionAccessEnabled_asTrue() { try (S3AsyncClient crossRegionCrtClient = S3AsyncClient.crtBuilder().crossRegionAccessEnabled(true).build()) { diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java index c07e566a6ed3..84e9697b8a91 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtAsyncHttpClientTest.java @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption.CRT_MEMORY_BUFFER_DISABLED; @@ -39,9 +40,16 @@ import java.nio.file.FileAlreadyExistsException; import java.nio.file.Path; import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Stream; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -67,11 +75,15 @@ import software.amazon.awssdk.crt.s3.S3ClientOptions; import software.amazon.awssdk.crt.s3.S3MetaRequest; import software.amazon.awssdk.crt.s3.S3MetaRequestOptions; +import software.amazon.awssdk.crt.s3.S3RequestMetrics; +import software.amazon.awssdk.http.HttpMetric; import software.amazon.awssdk.http.SdkHttpMethod; import software.amazon.awssdk.http.SdkHttpRequest; import software.amazon.awssdk.http.async.AsyncExecuteRequest; import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler; import software.amazon.awssdk.http.async.SdkHttpContentPublisher; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.s3.crt.S3CrtHttpConfiguration; import software.amazon.awssdk.testutils.RandomTempFile; @@ -92,6 +104,7 @@ public class S3CrtAsyncHttpClientTest { @BeforeEach public void methodSetup() { s3Client = Mockito.mock(S3Client.class); + when(s3Client.getShutdownCompleteFuture()).thenReturn(CompletableFuture.completedFuture(null)); responseHandler = Mockito.mock(SdkAsyncHttpResponseHandler.class); contentPublisher = Mockito.mock(SdkHttpContentPublisher.class); s3NativeClientConfiguration = S3NativeClientConfiguration.builder() @@ -150,6 +163,59 @@ public void getObject_shouldSetMetaRequestTypeCorrectly() { assertThat(actual.getOperationName()).isEqualTo("GetObject"); } + @Test + public void execute_withMetricPublisher_publishesCrtMaxConcurrency() { + int expectedMaxConcurrency = 64; + MetricPublisher publisher = Mockito.mock(MetricPublisher.class); + when(s3Client.getMaxActiveConnections()).thenReturn(expectedMaxConcurrency); + asyncHttpClient = new S3CrtAsyncHttpClient( + s3Client, + S3CrtAsyncHttpClient.builder() + .s3ClientConfiguration(s3NativeClientConfiguration) + .metricPublishers(Collections.singletonList(publisher))); + + try { + S3MetaRequestOptions requestOptions = makeRequest(getExecuteRequestBuilder().build()); + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + + ArgumentCaptor metricCollectionCaptor = ArgumentCaptor.forClass(MetricCollection.class); + verify(publisher, timeout(1_000)).publish(metricCollectionCaptor.capture()); + MetricCollection attempt = + metricCollectionCaptor.getValue().childrenWithName("ApiCallAttempt").findFirst() + .orElseThrow(AssertionError::new); + MetricCollection httpClient = + attempt.childrenWithName("HttpClient").findFirst().orElseThrow(AssertionError::new); + assertThat(httpClient.metricValues(HttpMetric.MAX_CONCURRENCY)).containsExactly(expectedMaxConcurrency); + } finally { + asyncHttpClient.close(); + } + } + + @Test + public void metricPublishers_areCopiedWhenClientIsConstructed() { + MetricPublisher originalPublisher = Mockito.mock(MetricPublisher.class); + MetricPublisher addedAfterConstruction = Mockito.mock(MetricPublisher.class); + List configuredPublishers = new ArrayList<>(); + configuredPublishers.add(originalPublisher); + asyncHttpClient = new S3CrtAsyncHttpClient( + s3Client, + S3CrtAsyncHttpClient.builder() + .s3ClientConfiguration(s3NativeClientConfiguration) + .metricPublishers(configuredPublishers)); + + configuredPublishers.add(addedAfterConstruction); + + try { + S3MetaRequestOptions requestOptions = makeRequest(getExecuteRequestBuilder().build()); + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + + verify(originalPublisher, timeout(1_000)).publish(any()); + verify(addedAfterConstruction, Mockito.after(100).never()).publish(any()); + } finally { + asyncHttpClient.close(); + } + } + @Test public void putObject_shouldSetMetaRequestTypeCorrectly() { AsyncExecuteRequest asyncExecuteRequest = getExecuteRequestBuilder().putHttpExecutionAttribute(OPERATION_NAME, @@ -430,6 +496,82 @@ public void closeHttpClient_shouldCloseUnderlyingResources() { s3NativeClientConfiguration.close(); } + @Test + public void close_withQueuedMetrics_waitsBeforeClosingCrtClient() throws Exception { + CompletableFuture shutdownCompleteFuture = new CompletableFuture<>(); + when(s3Client.getShutdownCompleteFuture()).thenReturn(shutdownCompleteFuture); + BlockingMetricPublisher publisher = new BlockingMetricPublisher(); + asyncHttpClient = new S3CrtAsyncHttpClient( + s3Client, + S3CrtAsyncHttpClient.builder() + .s3ClientConfiguration(s3NativeClientConfiguration) + .metricPublishers(Collections.singletonList(publisher))); + S3MetaRequestOptions requestOptions = makeRequest(getExecuteRequestBuilder().build()); + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + assertThat(publisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + + CompletableFuture closeFuture = CompletableFuture.runAsync(asyncHttpClient::close); + try { + verify(s3Client, timeout(1_000)).getShutdownCompleteFuture(); + assertThat(closeFuture.isDone()).isFalse(); + verify(s3Client, Mockito.never()).close(); + + publisher.release.countDown(); + closeFuture.get(1, TimeUnit.SECONDS); + assertThat(publisher.publishCount).hasValue(2); + verify(s3Client).close(); + } finally { + publisher.release.countDown(); + shutdownCompleteFuture.complete(null); + } + } + + @Test + public void close_metricDispatcherClosesOnlyAfterCrtShutdownCompletes() throws Exception { + CompletableFuture shutdownCompleteFuture = new CompletableFuture<>(); + when(s3Client.getShutdownCompleteFuture()).thenReturn(shutdownCompleteFuture); + AtomicInteger publishCount = new AtomicInteger(); + AtomicReference publishThread = new AtomicReference<>(); + CountDownLatch published = new CountDownLatch(1); + MetricPublisher publisher = new MetricPublisher() { + @Override + public void publish(MetricCollection metricCollection) { + publishThread.set(Thread.currentThread()); + publishCount.incrementAndGet(); + published.countDown(); + } + + @Override + public void close() { + } + }; + asyncHttpClient = new S3CrtAsyncHttpClient( + s3Client, + S3CrtAsyncHttpClient.builder() + .s3ClientConfiguration(s3NativeClientConfiguration) + .metricPublishers(Collections.singletonList(publisher))); + S3MetaRequestOptions requestOptions = makeRequest(getExecuteRequestBuilder().build()); + + try { + asyncHttpClient.close(); + + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + assertThat(published.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(publishCount).hasValue(1); + assertThat(publishThread.get().isAlive()).isTrue(); + + shutdownCompleteFuture.complete(null); + publishThread.get().join(1_000); + assertThat(publishThread.get().isAlive()).isFalse(); + + requestOptions.getResponseHandler().onTelemetry(Mockito.mock(S3RequestMetrics.class)); + assertThat(publishCount).hasValue(1); + } finally { + shutdownCompleteFuture.complete(null); + } + } + @Test void build_shouldPassThroughParameters() { String signingRegion = "us-west-2"; @@ -706,6 +848,27 @@ void execute_whenMakeMetaRequestThrows_shouldCloseAdapter() { Mockito.verify(adapter).close(); } + private static final class BlockingMetricPublisher implements MetricPublisher { + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private final AtomicInteger publishCount = new AtomicInteger(); + + @Override + public void publish(MetricCollection metricCollection) { + started.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + publishCount.incrementAndGet(); + } + + @Override + public void close() { + } + } + private AsyncExecuteRequest.Builder getExecuteRequestBuilder() { return getExecuteRequestBuilder(443); } diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtClientWiremockTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtClientWiremockTest.java index 3aa14683337c..910dea8232d2 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtClientWiremockTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtClientWiremockTest.java @@ -46,9 +46,12 @@ import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.AfterEach; @@ -65,7 +68,10 @@ import software.amazon.awssdk.core.async.AsyncResponseTransformer; import software.amazon.awssdk.core.async.BlockingOutputStreamAsyncRequestBody; import software.amazon.awssdk.core.exception.SdkClientException; +import software.amazon.awssdk.core.metrics.CoreMetric; import software.amazon.awssdk.crt.Log; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.s3.S3AsyncClient; import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse; @@ -348,6 +354,33 @@ void overrideResponseCompletionExecutor_shouldCompleteWithCustomExecutor(WireMoc } } + @Test + void getObject_notFound_publishesUnsuccessfulApiCallMetrics(WireMockRuntimeInfo wiremock) throws InterruptedException { + stubFor(any(anyUrl()).willReturn(aResponse().withStatus(404).withBody( + "NoSuchKey"))); + + CapturingMetricPublisher publisher = new CapturingMetricPublisher(); + try (S3AsyncClient client = S3AsyncClient.crtBuilder() + .region(Region.US_EAST_1) + .endpointOverride(URI.create("http://localhost:" + wiremock.getHttpPort())) + .credentialsProvider( + StaticCredentialsProvider.create(AwsBasicCredentials.create("key", "secret"))) + .addMetricPublisher(publisher) + .build()) { + + assertThatThrownBy(() -> client.getObject(r -> r.bucket(BUCKET).key(KEY), + AsyncResponseTransformer.toBytes()).join()) + .hasRootCauseInstanceOf(S3Exception.class); + + List collections = publisher.awaitAtLeast(1, Duration.ofSeconds(10)); + assertThat(collections).anySatisfy(c -> { + assertThat(c.name()).isEqualTo("ApiCall"); + assertThat(c.metricValues(CoreMetric.SERVICE_ID)).containsExactly("S3"); + assertThat(c.metricValues(CoreMetric.API_CALL_SUCCESSFUL)).contains(false); + }); + } + } + @Test void getObject_fullObjectChecksumMismatch_shouldThrowException(WireMockRuntimeInfo wiremock) { int partSize = 256 * 1024; @@ -465,4 +498,25 @@ public void execute(Runnable command) { command.run(); } } + + private static final class CapturingMetricPublisher implements MetricPublisher { + private final List collections = new CopyOnWriteArrayList<>(); + + @Override + public void publish(MetricCollection metricCollection) { + collections.add(metricCollection); + } + + @Override + public void close() { + } + + List awaitAtLeast(int min, Duration timeout) throws InterruptedException { + long deadlineNanos = System.nanoTime() + timeout.toNanos(); + while (collections.size() < min && System.nanoTime() < deadlineNanos) { + Thread.sleep(50); + } + return new ArrayList<>(collections); + } + } } diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java new file mode 100644 index 000000000000..8439b5cc56f5 --- /dev/null +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherDispatcherTest.java @@ -0,0 +1,357 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.services.s3.internal.crt; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricCollector; +import software.amazon.awssdk.metrics.MetricPublisher; + +public class S3CrtMetricPublisherDispatcherTest { + + @Test + public void dispatch_blockingPublisher_doesNotBlockCallingThread() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(1)); + BlockingPublisher publisher = new BlockingPublisher(); + AtomicReference dispatchThread = new AtomicReference<>(); + + try { + CompletableFuture dispatch = + CompletableFuture.runAsync(() -> { + dispatchThread.set(Thread.currentThread()); + dispatcher.dispatch(collection("one"), Collections.singletonList(publisher)); + }); + + dispatch.get(1, TimeUnit.SECONDS); + assertThat(publisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(publisher.publishThread.get()).isNotSameAs(dispatchThread.get()); + assertThat(publisher.publishThread.get().isDaemon()).isTrue(); + } finally { + publisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void dispatch_blockingPublisher_delaysOtherPublisherWithoutBlockingCaller() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(1)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + CapturingPublisher healthyPublisher = new CapturingPublisher(1); + + try { + dispatcher.dispatch(collection("one"), Arrays.asList(blockingPublisher, healthyPublisher)); + + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.completed.await(50, TimeUnit.MILLISECONDS)).isFalse(); + + blockingPublisher.release.countDown(); + assertThat(healthyPublisher.completed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.collectionNames).containsExactly("one"); + } finally { + blockingPublisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void dispatch_multiplePublishers_preservesGlobalAcceptanceOrder() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(1)); + List publications = new CopyOnWriteArrayList<>(); + CountDownLatch completed = new CountDownLatch(6); + MetricPublisher firstPublisher = recordingPublisher("first", publications, completed); + MetricPublisher secondPublisher = recordingPublisher("second", publications, completed); + List publishers = Arrays.asList(firstPublisher, secondPublisher); + + try { + dispatcher.dispatch(collection("one"), publishers); + dispatcher.dispatch(collection("two"), publishers); + dispatcher.dispatch(collection("three"), publishers); + + assertThat(completed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(publications).containsExactly("first-one", "second-one", + "first-two", "second-two", + "first-three", "second-three"); + } finally { + dispatcher.close(); + } + } + + @Test + public void dispatch_fullQueue_dropsWholeCollection() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofSeconds(1)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + CapturingPublisher healthyPublisher = new CapturingPublisher(2); + List publishers = Arrays.asList(blockingPublisher, healthyPublisher); + + try { + dispatcher.dispatch(collection("one"), publishers); + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + + dispatcher.dispatch(collection("two"), publishers); + dispatcher.dispatch(collection("three"), publishers); + + blockingPublisher.release.countDown(); + + assertThat(healthyPublisher.completed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.collectionNames).containsExactly("one", "two"); + } finally { + blockingPublisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void flushPendingMetrics_fullQueue_retriesUntilMarkerIsAccepted() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofSeconds(2)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + MetricPublisher queuedPublisher = mock(MetricPublisher.class); + AtomicReference flushThread = new AtomicReference<>(); + + CompletableFuture flushFuture = null; + try { + dispatcher.dispatch(collection("one"), Collections.singletonList(blockingPublisher)); + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + dispatcher.dispatch(collection("two"), Collections.singletonList(queuedPublisher)); + + flushFuture = CompletableFuture.runAsync(() -> { + flushThread.set(Thread.currentThread()); + dispatcher.flushPendingMetrics(); + }); + + assertThat(waitForThreadState(flushThread, Thread.State.TIMED_WAITING, Duration.ofSeconds(1))).isTrue(); + assertThat(flushFuture.isDone()).isFalse(); + + blockingPublisher.release.countDown(); + + flushFuture.get(1, TimeUnit.SECONDS); + verify(queuedPublisher).publish(collectionNamed("two")); + } finally { + blockingPublisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void dispatch_publisherThrows_doesNotAffectRemainingPublishersAndWorkerContinues() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(1)); + ThrowingFirstPublisher throwingPublisher = new ThrowingFirstPublisher(); + CapturingPublisher healthyPublisher = new CapturingPublisher(2); + List publishers = Arrays.asList(throwingPublisher, healthyPublisher); + + try { + dispatcher.dispatch(collection("one"), publishers); + dispatcher.dispatch(collection("two"), publishers); + + assertThat(throwingPublisher.secondCallCompleted.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.completed.await(1, TimeUnit.SECONDS)).isTrue(); + assertThat(healthyPublisher.collectionNames).containsExactly("one", "two"); + } finally { + dispatcher.close(); + } + } + + @Test + public void close_drainsQueuedCollectionsWithoutClosingPublishers() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(4, Duration.ofSeconds(2)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + MetricPublisher publisherSpy = mock(MetricPublisher.class); + + dispatcher.dispatch(collection("one"), Collections.singletonList(blockingPublisher)); + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + dispatcher.dispatch(collection("two"), Collections.singletonList(publisherSpy)); + + CompletableFuture closeFuture = CompletableFuture.runAsync(dispatcher::close); + blockingPublisher.release.countDown(); + + closeFuture.get(1, TimeUnit.SECONDS); + verify(publisherSpy).publish(collectionNamed("two")); + verify(publisherSpy, never()).close(); + } + + @Test + public void close_blockedPublisher_returnsAfterTimeoutAndDropsQueuedCollections() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofMillis(50)); + BlockingPublisher blockingPublisher = new BlockingPublisher(); + MetricPublisher queuedPublisher = mock(MetricPublisher.class); + + try { + dispatcher.dispatch(collection("one"), Collections.singletonList(blockingPublisher)); + assertThat(blockingPublisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + dispatcher.dispatch(collection("two"), Collections.singletonList(queuedPublisher)); + + long startNanos = System.nanoTime(); + dispatcher.close(); + long closeDurationMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + + assertThat(closeDurationMillis).isLessThan(1_000); + verify(queuedPublisher, never()).publish(org.mockito.ArgumentMatchers.any()); + } finally { + blockingPublisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void startAsyncClose_blockedPublisher_doesNotBlockCallingThread() throws Exception { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofSeconds(1)); + BlockingPublisher publisher = new BlockingPublisher(); + + try { + dispatcher.dispatch(collection("one"), Collections.singletonList(publisher)); + assertThat(publisher.started.await(1, TimeUnit.SECONDS)).isTrue(); + + CompletableFuture.runAsync(dispatcher::startAsyncClose).get(1, TimeUnit.SECONDS); + } finally { + publisher.release.countDown(); + dispatcher.close(); + } + } + + @Test + public void dispatch_closedDispatcher_dropsWithoutInvokingPublisher() { + S3CrtMetricPublisherDispatcher dispatcher = + new S3CrtMetricPublisherDispatcher(1, Duration.ofSeconds(1)); + MetricPublisher publisher = mock(MetricPublisher.class); + + dispatcher.close(); + dispatcher.dispatch(collection("one"), Collections.singletonList(publisher)); + + verify(publisher, never()).publish(org.mockito.ArgumentMatchers.any()); + } + + private static MetricPublisher recordingPublisher(String publisherName, + List publications, + CountDownLatch completed) { + return new MetricPublisher() { + @Override + public void publish(MetricCollection metricCollection) { + publications.add(publisherName + "-" + metricCollection.name()); + completed.countDown(); + } + + @Override + public void close() { + } + }; + } + + private static MetricCollection collection(String name) { + return MetricCollector.create(name).collect(); + } + + private static MetricCollection collectionNamed(String name) { + return org.mockito.ArgumentMatchers.argThat(collection -> collection != null && name.equals(collection.name())); + } + + private static boolean waitForThreadState(AtomicReference threadReference, + Thread.State state, + Duration timeout) throws InterruptedException { + long deadlineNanos = System.nanoTime() + timeout.toNanos(); + while (System.nanoTime() < deadlineNanos) { + Thread thread = threadReference.get(); + if (thread != null && thread.getState() == state) { + return true; + } + TimeUnit.MILLISECONDS.sleep(1); + } + return false; + } + + private static final class BlockingPublisher implements MetricPublisher { + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private final AtomicReference publishThread = new AtomicReference<>(); + + @Override + public void publish(MetricCollection metricCollection) { + publishThread.set(Thread.currentThread()); + started.countDown(); + boolean released = false; + while (!released) { + try { + release.await(); + released = true; + } catch (InterruptedException ignored) { + // Deliberately model a publisher that does not respond to interruption. + } + } + } + + @Override + public void close() { + } + } + + private static final class CapturingPublisher implements MetricPublisher { + private final List collectionNames = new CopyOnWriteArrayList<>(); + private final CountDownLatch completed; + + private CapturingPublisher(int expectedCollections) { + this.completed = new CountDownLatch(expectedCollections); + } + + @Override + public void publish(MetricCollection metricCollection) { + collectionNames.add(metricCollection.name()); + completed.countDown(); + } + + @Override + public void close() { + } + } + + private static final class ThrowingFirstPublisher implements MetricPublisher { + private final AtomicInteger calls = new AtomicInteger(); + private final CountDownLatch secondCallCompleted = new CountDownLatch(1); + + @Override + public void publish(MetricCollection metricCollection) { + if (calls.incrementAndGet() == 1) { + throw new IllegalStateException("expected"); + } + secondCallCompleted.countDown(); + } + + @Override + public void close() { + } + } +} diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherResolutionTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherResolutionTest.java new file mode 100644 index 000000000000..5000aeeaa583 --- /dev/null +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtMetricPublisherResolutionTest.java @@ -0,0 +1,133 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.services.s3.internal.crt; + +import static java.util.Collections.emptyList; +import static java.util.Collections.singletonList; +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; +import software.amazon.awssdk.services.s3.model.CopyObjectRequest; +import software.amazon.awssdk.services.s3.model.GetObjectRequest; + +/** + * Unit tests for request-level metric-publisher support on the CRT client. + */ +public class S3CrtMetricPublisherResolutionTest { + + private static MetricPublisher publisher(String name) { + return new MetricPublisher() { + @Override + public void publish(MetricCollection metricCollection) { + } + + @Override + public void close() { + } + + @Override + public String toString() { + return name; + } + }; + } + + // ---- effective-publisher resolution matrix ---- + + @Test + public void resolveEffectiveMetricPublishers_noPublishers_returnsEmpty() { + assertThat(S3CrtAsyncHttpClient.resolveEffectiveMetricPublishers(emptyList(), emptyList())).isEmpty(); + assertThat(S3CrtAsyncHttpClient.resolveEffectiveMetricPublishers(null, emptyList())).isEmpty(); + } + + @Test + public void resolveEffectiveMetricPublishers_requestLevelOnly_usesRequestLevel() { + MetricPublisher request = publisher("request"); + assertThat(S3CrtAsyncHttpClient.resolveEffectiveMetricPublishers(singletonList(request), emptyList())) + .containsExactly(request); + } + + @Test + public void resolveEffectiveMetricPublishers_clientLevelOnly_usesClientLevel() { + MetricPublisher client = publisher("client"); + assertThat(S3CrtAsyncHttpClient.resolveEffectiveMetricPublishers(emptyList(), singletonList(client))) + .containsExactly(client); + assertThat(S3CrtAsyncHttpClient.resolveEffectiveMetricPublishers(null, singletonList(client))) + .containsExactly(client); + } + + @Test + public void resolveEffectiveMetricPublishers_requestAndClientLevel_requestTakesPrecedence() { + MetricPublisher client = publisher("client"); + MetricPublisher request = publisher("request"); + assertThat(S3CrtAsyncHttpClient.resolveEffectiveMetricPublishers(singletonList(request), singletonList(client))) + .containsExactly(request); + } + + // ---- stashRequestMetricPublishers: move publishers off the request into the execution attribute ---- + + @Test + public void stash_movesRequestPublishersToExecutionAttributeAndStripsOverride() { + MetricPublisher request = publisher("request"); + GetObjectRequest in = GetObjectRequest.builder() + .bucket("b").key("k") + .overrideConfiguration(o -> o.addMetricPublisher(request)) + .build(); + + GetObjectRequest out = DefaultS3CrtAsyncClient.stashRequestMetricPublishers(in); + + // stripped off the override so the inner standard client's resolveMetricPublishers() sees nothing + assertThat(out.overrideConfiguration().get().metricPublishers()).isEmpty(); + // stashed in the execution attribute the CRT transport reads + assertThat(out.overrideConfiguration().get().executionAttributes() + .getAttribute(DefaultS3CrtAsyncClient.REQUEST_METRIC_PUBLISHERS)) + .containsExactly(request); + } + + @Test + public void stash_copyObjectRequest_movesPublishersOffOverrideIntoAttribute() { + MetricPublisher request = publisher("request"); + CopyObjectRequest in = CopyObjectRequest.builder() + .sourceBucket("src").sourceKey("k") + .destinationBucket("dst").destinationKey("k2") + .overrideConfiguration(o -> o.addMetricPublisher(request)) + .build(); + + CopyObjectRequest out = DefaultS3CrtAsyncClient.stashRequestMetricPublishers(in); + + assertThat(out.overrideConfiguration().get().metricPublishers()).isEmpty(); + assertThat(out.overrideConfiguration().get().executionAttributes() + .getAttribute(DefaultS3CrtAsyncClient.REQUEST_METRIC_PUBLISHERS)) + .containsExactly(request); + } + + @Test + public void stash_noOverrideConfiguration_returnsSameRequest() { + GetObjectRequest in = GetObjectRequest.builder().bucket("b").key("k").build(); + assertThat(DefaultS3CrtAsyncClient.stashRequestMetricPublishers(in)).isSameAs(in); + } + + @Test + public void stash_overrideWithoutPublishers_returnsSameRequest() { + GetObjectRequest in = GetObjectRequest.builder() + .bucket("b").key("k") + .overrideConfiguration(o -> o.putRawQueryParameter("x", "y")) + .build(); + assertThat(DefaultS3CrtAsyncClient.stashRequestMetricPublishers(in)).isSameAs(in); + } +} diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterMetricsTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterMetricsTest.java new file mode 100644 index 000000000000..efe88eb657e6 --- /dev/null +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterMetricsTest.java @@ -0,0 +1,279 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.services.s3.internal.crt; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import java.net.URI; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.crt.CrtRuntimeException; +import software.amazon.awssdk.crt.http.HttpManagerMetrics; +import software.amazon.awssdk.crt.s3.S3RequestMetrics; +import software.amazon.awssdk.http.HttpMetric; +import software.amazon.awssdk.http.SdkHttpExecutionAttributes; +import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; + +/** + * Unit tests for {@link S3CrtResponseHandlerAdapter#onTelemetry(S3RequestMetrics)} - the mapping of CRT native request + * telemetry onto the {@code ApiCall -> ApiCallAttempt -> HttpClient} {@link MetricCollection} that is published to the + * configured publishers. CRT invokes {@code onTelemetry} once per underlying request attempt, so each call is published + * as its own top-level {@code ApiCall} collection. + */ +public class S3CrtResponseHandlerAdapterMetricsTest { + + private static final int AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE = 14358; + private static final long API_CALL_NANOS = 98_000_000L; + private static final long SIGNING_NANOS = 1_000_000L; + private static final long SERVICE_CALL_NANOS = 35_000_000L; + private static final long TIME_TO_FIRST_BYTE_NANOS = 40_000_000L; + private static final long TIME_TO_LAST_BYTE_NANOS = 90_000_000L; + private static final long CONNECTION_ACQUISITION_NANOS = 2_000_000L; + private static final String SERVICE_ENDPOINT = "https://s3.us-west-2.amazonaws.com"; + private static final int MAX_CONCURRENCY = 10; + private S3CrtMetricPublisherDispatcher metricPublisherDispatcher; + + @BeforeEach + public void setup() { + metricPublisherDispatcher = new S3CrtMetricPublisherDispatcher(); + } + + @AfterEach + public void teardown() { + metricPublisherDispatcher.close(); + } + + @Test + public void onTelemetry_publishesApiCallTreeWithMappedValues() throws Exception { + CapturingPublisher publisher = new CapturingPublisher(1); + adapterPublishingTo(publisher).onTelemetry(successfulGetObjectMetrics()); + publisher.awaitExpectedCollections(); + + assertThat(publisher.collections).hasSize(1); + MetricCollection apiCall = publisher.collections.get(0); + assertThat(apiCall.name()).isEqualTo("ApiCall"); + assertThat(apiCall.metricValues(CoreMetric.SERVICE_ID)).containsExactly("S3"); + assertThat(apiCall.metricValues(CoreMetric.OPERATION_NAME)).containsExactly("GetObject"); + assertThat(apiCall.metricValues(CoreMetric.API_CALL_SUCCESSFUL)).containsExactly(true); + assertThat(apiCall.metricValues(CoreMetric.RETRY_COUNT)).containsExactly(0); + assertThat(apiCall.metricValues(CoreMetric.API_CALL_DURATION)).containsExactly(Duration.ofNanos(API_CALL_NANOS)); + assertThat(apiCall.metricValues(CoreMetric.SERVICE_ENDPOINT)).containsExactly(URI.create(SERVICE_ENDPOINT)); + + MetricCollection attempt = childNamed(apiCall, "ApiCallAttempt"); + assertThat(attempt.metricValues(CoreMetric.SIGNING_DURATION)).containsExactly(Duration.ofNanos(SIGNING_NANOS)); + assertThat(attempt.metricValues(CoreMetric.SERVICE_CALL_DURATION)).containsExactly(Duration.ofNanos(SERVICE_CALL_NANOS)); + assertThat(attempt.metricValues(CoreMetric.BACKOFF_DELAY_DURATION)).containsExactly(Duration.ofNanos(0)); + assertThat(attempt.metricValues(CoreMetric.AWS_REQUEST_ID)).containsExactly("REQ-123"); + assertThat(attempt.metricValues(CoreMetric.AWS_EXTENDED_REQUEST_ID)).containsExactly("EXT-456"); + assertThat(attempt.metricValues(CoreMetric.TIME_TO_FIRST_BYTE)) + .containsExactly(Duration.ofNanos(TIME_TO_FIRST_BYTE_NANOS)); + assertThat(attempt.metricValues(CoreMetric.TIME_TO_LAST_BYTE)) + .containsExactly(Duration.ofNanos(TIME_TO_LAST_BYTE_NANOS)); + assertThat(attempt.metricValues(HttpMetric.HTTP_STATUS_CODE)).containsExactly(200); + + MetricCollection httpClient = childNamed(attempt, "HttpClient"); + assertThat(httpClient.metricValues(HttpMetric.HTTP_CLIENT_NAME)).containsExactly("s3crt"); + assertThat(httpClient.metricValues(HttpMetric.MAX_CONCURRENCY)).containsExactly(MAX_CONCURRENCY); + assertThat(httpClient.metricValues(HttpMetric.CONCURRENCY_ACQUIRE_DURATION)) + .containsExactly(Duration.ofNanos(CONNECTION_ACQUISITION_NANOS)); + assertThat(httpClient.metricValues(HttpMetric.AVAILABLE_CONCURRENCY)).containsExactly(4); + assertThat(httpClient.metricValues(HttpMetric.LEASED_CONCURRENCY)).containsExactly(2); + assertThat(httpClient.metricValues(HttpMetric.PENDING_CONCURRENCY_ACQUIRES)).containsExactly(1); + } + + @Test + public void onTelemetry_unavailableMetric_isSkipped_othersStillPublished() throws Exception { + S3RequestMetrics metrics = successfulGetObjectMetrics(); + // CRT throws this when a datum is not available for the request; only that one metric should be skipped. + when(metrics.getBackoffDelayDurationNs()).thenThrow(new CrtRuntimeException(AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE)); + + CapturingPublisher publisher = new CapturingPublisher(1); + adapterPublishingTo(publisher).onTelemetry(metrics); + publisher.awaitExpectedCollections(); + + assertThat(publisher.collections).hasSize(1); + MetricCollection attempt = childNamed(publisher.collections.get(0), "ApiCallAttempt"); + assertThat(attempt.metricValues(CoreMetric.BACKOFF_DELAY_DURATION)).isEmpty(); + assertThat(attempt.metricValues(CoreMetric.SIGNING_DURATION)).isNotEmpty(); + assertThat(attempt.metricValues(CoreMetric.AWS_REQUEST_ID)).isNotEmpty(); + } + + @Test + public void onTelemetry_optionalHttpMetricsUnavailable_areSkipped() throws Exception { + S3RequestMetrics metrics = successfulGetObjectMetrics(); + when(metrics.getTimeToFirstByte()).thenThrow(new CrtRuntimeException(AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE)); + when(metrics.getConnectionAcquisitionDurationNs()) + .thenThrow(new CrtRuntimeException(AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE)); + when(metrics.getResponseStatus()).thenReturn(-1); + when(metrics.getHttpManagerMetrics()).thenReturn(null); + + CapturingPublisher publisher = new CapturingPublisher(1); + adapterPublishingTo(publisher).onTelemetry(metrics); + publisher.awaitExpectedCollections(); + + assertThat(publisher.collections).hasSize(1); + MetricCollection attempt = childNamed(publisher.collections.get(0), "ApiCallAttempt"); + assertThat(attempt.metricValues(CoreMetric.TIME_TO_FIRST_BYTE)).isEmpty(); + assertThat(attempt.metricValues(CoreMetric.TIME_TO_LAST_BYTE)).isNotEmpty(); + assertThat(attempt.metricValues(HttpMetric.HTTP_STATUS_CODE)).isEmpty(); + + MetricCollection httpClient = childNamed(attempt, "HttpClient"); + assertThat(httpClient.metricValues(HttpMetric.HTTP_CLIENT_NAME)).containsExactly("s3crt"); + assertThat(httpClient.metricValues(HttpMetric.CONCURRENCY_ACQUIRE_DURATION)).isEmpty(); + assertThat(httpClient.metricValues(HttpMetric.AVAILABLE_CONCURRENCY)).isEmpty(); + assertThat(httpClient.metricValues(HttpMetric.LEASED_CONCURRENCY)).isEmpty(); + assertThat(httpClient.metricValues(HttpMetric.PENDING_CONCURRENCY_ACQUIRES)).isEmpty(); + } + + @Test + public void onTelemetry_failedAttempt_reportsApiCallUnsuccessful() throws Exception { + S3RequestMetrics metrics = successfulGetObjectMetrics(); + when(metrics.isApiCallSuccessful()).thenReturn(false); + + CapturingPublisher publisher = new CapturingPublisher(1); + adapterPublishingTo(publisher).onTelemetry(metrics); + publisher.awaitExpectedCollections(); + + MetricCollection apiCall = publisher.collections.get(0); + assertThat(apiCall.metricValues(CoreMetric.API_CALL_SUCCESSFUL)).containsExactly(false); + assertThat(apiCall.metricValues(CoreMetric.API_CALL_DURATION)).isNotEmpty(); + } + + @Test + public void onTelemetry_noPublishers_isNoOp_andReadsNoMetrics() { + S3RequestMetrics metrics = mock(S3RequestMetrics.class); + adapterPublishingTo().onTelemetry(metrics); + // The empty-publisher early-return happens before any collection is built or any native getter is read. + verifyNoInteractions(metrics); + } + + @Test + public void onTelemetry_calledPerAttempt_publishesOneCollectionEach() throws Exception { + CapturingPublisher publisher = new CapturingPublisher(3); + S3CrtResponseHandlerAdapter adapter = adapterPublishingTo(publisher); + S3RequestMetrics metrics = successfulGetObjectMetrics(); + + adapter.onTelemetry(metrics); + adapter.onTelemetry(metrics); + adapter.onTelemetry(metrics); + publisher.awaitExpectedCollections(); + + assertThat(publisher.collections).hasSize(3); + } + + @Test + public void onTelemetry_retriedRequest_publishesSeparateCollectionsWithIncreasingRetryCount() throws Exception { + CapturingPublisher publisher = new CapturingPublisher(2); + S3CrtResponseHandlerAdapter adapter = adapterPublishingTo(publisher); + + // First attempt fails (retry count 0), the retry succeeds (retry count 1). CRT delivers these as two separate + // onTelemetry callbacks, so we publish two top-level ApiCall collections - not one ApiCall with two attempts. + S3RequestMetrics firstAttempt = successfulGetObjectMetrics(); + when(firstAttempt.isApiCallSuccessful()).thenReturn(false); + when(firstAttempt.getRetryCount()).thenReturn(0); + adapter.onTelemetry(firstAttempt); + + S3RequestMetrics retryAttempt = successfulGetObjectMetrics(); + when(retryAttempt.getRetryCount()).thenReturn(1); + adapter.onTelemetry(retryAttempt); + publisher.awaitExpectedCollections(); + + assertThat(publisher.collections).hasSize(2); + assertThat(publisher.collections.get(0).metricValues(CoreMetric.API_CALL_SUCCESSFUL)).containsExactly(false); + assertThat(publisher.collections.get(0).metricValues(CoreMetric.RETRY_COUNT)).containsExactly(0); + assertThat(publisher.collections.get(1).metricValues(CoreMetric.API_CALL_SUCCESSFUL)).containsExactly(true); + assertThat(publisher.collections.get(1).metricValues(CoreMetric.RETRY_COUNT)).containsExactly(1); + } + + private static S3RequestMetrics successfulGetObjectMetrics() throws Exception { + S3RequestMetrics metrics = mock(S3RequestMetrics.class); + when(metrics.getServiceId()).thenReturn("S3"); + when(metrics.getServiceEndpoint()).thenReturn(SERVICE_ENDPOINT); + when(metrics.getOperationName()).thenReturn("GetObject"); + when(metrics.isApiCallSuccessful()).thenReturn(true); + when(metrics.getRetryCount()).thenReturn(0); + when(metrics.getApiCallDurationNs()).thenReturn(API_CALL_NANOS); + when(metrics.getSigningDurationNs()).thenReturn(SIGNING_NANOS); + when(metrics.getServiceCallDurationNs()).thenReturn(SERVICE_CALL_NANOS); + when(metrics.getBackoffDelayDurationNs()).thenReturn(0L); + when(metrics.getAwsRequestId()).thenReturn("REQ-123"); + when(metrics.getAwsExtendedRequestId()).thenReturn("EXT-456"); + when(metrics.getTimeToFirstByte()).thenReturn(TIME_TO_FIRST_BYTE_NANOS); + when(metrics.getTimeToLastByte()).thenReturn(TIME_TO_LAST_BYTE_NANOS); + when(metrics.getResponseStatus()).thenReturn(200); + when(metrics.getConnectionAcquisitionDurationNs()).thenReturn(CONNECTION_ACQUISITION_NANOS); + + HttpManagerMetrics managerMetrics = mock(HttpManagerMetrics.class); + when(managerMetrics.getAvailableConcurrency()).thenReturn(4L); + when(managerMetrics.getLeasedConcurrency()).thenReturn(2L); + when(managerMetrics.getPendingConcurrencyAcquires()).thenReturn(1L); + when(metrics.getHttpManagerMetrics()).thenReturn(managerMetrics); + return metrics; + } + + private S3CrtResponseHandlerAdapter adapterPublishingTo(MetricPublisher... publishers) { + return S3CrtResponseHandlerAdapter.builder() + .executeFuture(new CompletableFuture<>()) + .responseHandler(mock(SdkAsyncHttpResponseHandler.class)) + .httpExecutionAttributes(SdkHttpExecutionAttributes.builder().build()) + .metaRequestFuture(new CompletableFuture<>()) + .metricPublishers(Arrays.asList(publishers)) + .metricPublisherDispatcher(metricPublisherDispatcher) + .maxConcurrency(MAX_CONCURRENCY) + .build(); + } + + private static MetricCollection childNamed(MetricCollection parent, String name) { + return parent.childrenWithName(name).findFirst().orElseThrow(() -> new AssertionError("missing child: " + name)); + } + + private static final class CapturingPublisher implements MetricPublisher { + private final List collections = new CopyOnWriteArrayList<>(); + private final CountDownLatch expectedCollections; + + private CapturingPublisher(int expectedCollections) { + this.expectedCollections = new CountDownLatch(expectedCollections); + } + + @Override + public void publish(MetricCollection metricCollection) { + collections.add(metricCollection); + expectedCollections.countDown(); + } + + @Override + public void close() { + } + + private void awaitExpectedCollections() throws InterruptedException { + assertThat(expectedCollections.await(5, TimeUnit.SECONDS)).isTrue(); + } + } +} diff --git a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java index 70e519ebc356..67c33059e7f8 100644 --- a/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterTest.java @@ -33,6 +33,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import org.junit.After; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -47,6 +48,7 @@ import software.amazon.awssdk.crt.http.HttpHeader; import software.amazon.awssdk.crt.s3.S3FinishedResponseContext; import software.amazon.awssdk.crt.s3.S3MetaRequest; +import software.amazon.awssdk.http.SdkHttpExecutionAttributes; import software.amazon.awssdk.http.SdkHttpResponse; import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler; import software.amazon.awssdk.services.s3.model.S3Exception; @@ -63,15 +65,27 @@ public class S3CrtResponseHandlerAdapterTest { @Mock private S3MetaRequestWrapper s3MetaRequest; private CompletableFuture future; + private S3CrtMetricPublisherDispatcher metricPublisherDispatcher; @Before public void setup() { future = new CompletableFuture<>(); sdkResponseHandler = spy(new TestResponseHandler()); - responseHandlerAdapter = new S3CrtResponseHandlerAdapter(future, - sdkResponseHandler, - null, - CompletableFuture.completedFuture(s3MetaRequest)); + metricPublisherDispatcher = new S3CrtMetricPublisherDispatcher(); + responseHandlerAdapter = S3CrtResponseHandlerAdapter.builder() + .executeFuture(future) + .responseHandler(sdkResponseHandler) + .httpExecutionAttributes( + SdkHttpExecutionAttributes.builder().build()) + .metaRequestFuture( + CompletableFuture.completedFuture(s3MetaRequest)) + .metricPublisherDispatcher(metricPublisherDispatcher) + .build(); + } + + @After + public void teardown() { + metricPublisherDispatcher.close(); } @Test @@ -101,11 +115,15 @@ public void successfulResponse_shouldCompleteFutureSuccessfully() throws Excepti @Test public void s3MetaRequestNotFinish_shouldFailFuture() throws Exception { - S3CrtResponseHandlerAdapter responseHandlerAdapter = new S3CrtResponseHandlerAdapter(future, - sdkResponseHandler, - null, - new CompletableFuture<>(), - Duration.ofMillis(10)); + S3CrtResponseHandlerAdapter responseHandlerAdapter = + S3CrtResponseHandlerAdapter.builder() + .executeFuture(future) + .responseHandler(sdkResponseHandler) + .httpExecutionAttributes(SdkHttpExecutionAttributes.builder().build()) + .metaRequestFuture(new CompletableFuture<>()) + .s3MetaRequestTimeout(Duration.ofMillis(10)) + .metricPublisherDispatcher(metricPublisherDispatcher) + .build(); int statusCode = 200; responseHandlerAdapter.onResponseHeaders(statusCode, new HttpHeader[0]); responseHandlerAdapter.onResponseBody(ByteBuffer.wrap("helloworld1".getBytes()), 1, 2);