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..d5109324dec2 --- /dev/null +++ b/services/s3/src/it/java/software/amazon/awssdk/services/s3/crt/S3CrtClientMetricPublisherIntegrationTest.java @@ -0,0 +1,180 @@ +/* + * 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 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.BeforeEach; +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.testutils.RandomTempFile; +import software.amazon.awssdk.testutils.service.AwsTestBase; + +/** + * Verifies that the CRT-based S3 client publishes CRT native request telemetry to a client-level {@link MetricPublisher} + * configured via {@code crtBuilder().addMetricPublisher(...)}. 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; + + private static final CapturingMetricPublisher CLIENT_PUBLISHER = new CapturingMetricPublisher(); + private static S3AsyncClient crtClient; + + @BeforeAll + public static void setup() throws Exception { + S3IntegrationTestBase.setUp(); + S3IntegrationTestBase.createBucket(BUCKET); + + crtClient = S3AsyncClient.crtBuilder() + .region(S3IntegrationTestBase.DEFAULT_REGION) + .credentialsProvider(AwsTestBase.CREDENTIALS_PROVIDER_CHAIN) + .minimumPartSizeInBytes(PART_SIZE) + .thresholdInBytes(PART_SIZE) + .addMetricPublisher(CLIENT_PUBLISHER) + .build(); + + 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() { + crtClient.close(); + S3IntegrationTestBase.deleteBucketAndAllContents(BUCKET); + } + + @BeforeEach + public void resetPublisher() { + CLIENT_PUBLISHER.clear(); + } + + @Test + void singlePartGetObject_publishesApiCallMetrics() throws InterruptedException { + crtClient.getObject(b -> b.bucket(BUCKET).key(SMALL_KEY), AsyncResponseTransformer.toBytes()).join(); + + List collections = CLIENT_PUBLISHER.awaitAtLeast(1, Duration.ofSeconds(30)); + assertThat(collections).isNotEmpty(); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + } + + @Test + void multipartGetObject_publishesOneCollectionPerUnderlyingRequest() throws InterruptedException { + crtClient.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 = CLIENT_PUBLISHER.awaitAtLeast(2, Duration.ofSeconds(60)); + assertThat(collections).hasSizeGreaterThanOrEqualTo(2); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + } + + @Test + void putObject_publishesApiCallMetrics() throws InterruptedException { + crtClient.putObject(b -> b.bucket(BUCKET).key("put-metrics-key"), + AsyncRequestBody.fromBytes(new byte[SMALL_SIZE])).join(); + + List collections = CLIENT_PUBLISHER.awaitAtLeast(1, Duration.ofSeconds(30)); + assertThat(collections).isNotEmpty(); + collections.forEach(S3CrtClientMetricPublisherIntegrationTest::assertIsCrtApiCallCollection); + } + + @Test + void failedGetObject_publishesUnsuccessfulApiCallMetrics() throws InterruptedException { + try { + crtClient.getObject(b -> b.bucket(BUCKET).key("does-not-exist-" + System.nanoTime()), + AsyncResponseTransformer.toBytes()).join(); + } catch (RuntimeException expected) { + // 404 - the metrics for the failed attempt should still be published below. + } + + List collections = CLIENT_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)); + } + + 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() { + } + + void clear() { + collections.clear(); + } + + 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 66569eb39f97..9a0c9e9bbdd1 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; @@ -29,6 +30,7 @@ import software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption; 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; @@ -379,6 +381,33 @@ 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. + * + *

The SDK does not close these publishers when the client is closed; the caller retains ownership and is + * responsible for closing them. + * + * @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 8ef59ac7cca8..067b30f4e10d 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,6 +30,7 @@ import java.net.URI; import java.nio.file.Path; import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -64,6 +65,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; @@ -236,7 +238,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 { @@ -261,6 +264,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(); @@ -406,6 +410,23 @@ public DefaultS3CrtClientBuilder advancedOptions(Map metricPublishers) { + Validate.paramNotNull(metricPublishers, "metricPublishers"); + 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); 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 b48d9528800c..96a642932080 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; @@ -35,6 +34,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; @@ -59,6 +59,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; @@ -71,14 +72,18 @@ @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 S3CrtAsyncHttpClient(Builder builder) { s3NativeClientConfiguration = builder.clientConfiguration; this.s3ClientOptions = createS3ClientOption(); + this.metricPublishers = builder.metricPublishers == null ? Collections.emptyList() : builder.metricPublishers; this.crtS3Client = new S3Client(s3ClientOptions); } @@ -88,6 +93,7 @@ private S3CrtAsyncHttpClient(Builder builder) { Builder builder) { s3NativeClientConfiguration = builder.clientConfiguration; s3ClientOptions = createS3ClientOption(); + this.metricPublishers = builder.metricPublishers == null ? Collections.emptyList() : builder.metricPublishers; this.crtS3Client = crtS3Client; } @@ -160,12 +166,22 @@ public CompletableFuture execute(AsyncExecuteRequest asyncRequest) { Path responseFilePath = httpExecutionAttributes.getAttribute(RESPONSE_FILE_PATH); S3MetaRequestOptions.ResponseFileOption responseFileOption = httpExecutionAttributes.getAttribute(RESPONSE_FILE_OPTION); + // The adapter reads its inputs from the execution attributes, so attach the client-level publishers here when + // there are any. toBuilder() preserves everything already in the bag (including CRT_PROGRESS_LISTENER); skip the + // copy entirely when no publishers are configured, to avoid rebuilding the bag on every request. + SdkHttpExecutionAttributes adapterAttributes = httpExecutionAttributes; + if (!metricPublishers.isEmpty()) { + adapterAttributes = httpExecutionAttributes.toBuilder() + .put(S3InternalSdkHttpExecutionAttribute.METRIC_PUBLISHERS, + metricPublishers) + .build(); + } + S3CrtResponseHandlerAdapter responseHandler = - new S3CrtResponseHandlerAdapter( - executeFuture, - asyncRequest.responseHandler(), - httpExecutionAttributes.getAttribute(CRT_PROGRESS_LISTENER), - s3MetaRequestFuture); + new S3CrtResponseHandlerAdapter(executeFuture, + asyncRequest.responseHandler(), + adapterAttributes, + s3MetaRequestFuture); URI endpoint = getEndpoint(uri); @@ -244,7 +260,7 @@ private static URI getEndpoint(URI uri) { @Override public String clientName() { - return "s3crt"; + return CLIENT_NAME; } private static S3MetaRequestOptions.MetaRequestType requestType(String operationName) { @@ -297,12 +313,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/S3CrtResponseHandlerAdapter.java b/services/s3/src/main/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapter.java index 7c778b173687..1e48b923e786 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,12 @@ 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 java.nio.ByteBuffer; import java.time.Duration; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -28,6 +30,8 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +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; @@ -35,15 +39,24 @@ 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.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.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; @@ -65,20 +78,21 @@ public final class S3CrtResponseHandlerAdapter implements S3MetaRequestResponseH private final PublisherListener progressListener; private final Duration s3MetaRequestTimeout; + private final List metricPublishers; private volatile boolean responseHandlingInitiated; public S3CrtResponseHandlerAdapter(CompletableFuture executeFuture, SdkAsyncHttpResponseHandler responseHandler, - PublisherListener progressListener, + SdkHttpExecutionAttributes httpExecutionAttributes, CompletableFuture metaRequestFuture) { - this(executeFuture, responseHandler, progressListener, metaRequestFuture, META_REQUEST_TIMEOUT); + this(executeFuture, responseHandler, httpExecutionAttributes, metaRequestFuture, META_REQUEST_TIMEOUT); } @SdkTestInternalApi public S3CrtResponseHandlerAdapter(CompletableFuture executeFuture, SdkAsyncHttpResponseHandler responseHandler, - PublisherListener progressListener, + SdkHttpExecutionAttributes httpExecutionAttributes, CompletableFuture metaRequestFuture, Duration s3MetaRequestTimeout) { this.resultFuture = executeFuture; @@ -97,7 +111,11 @@ public S3CrtResponseHandlerAdapter(CompletableFuture executeFuture, }); this.responseHandler = responseHandler; + PublisherListener progressListener = httpExecutionAttributes.getAttribute(CRT_PROGRESS_LISTENER); this.progressListener = progressListener == null ? new NoOpPublisherListener() : progressListener; + List publishers = + httpExecutionAttributes.getAttribute(S3InternalSdkHttpExecutionAttribute.METRIC_PUBLISHERS); + this.metricPublishers = publishers == null ? Collections.emptyList() : publishers; this.s3MetaRequestTimeout = s3MetaRequestTimeout; } @@ -304,6 +322,68 @@ 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 only reads the metrics object and publishes, and never throws back into the native callback. + */ + @Override + public void onTelemetry(S3RequestMetrics requestMetrics) { + if (metricPublishers.isEmpty()) { + return; + } + try { + MetricCollector apiCall = MetricCollector.create("ApiCall"); + apiCall.reportMetric(CoreMetric.SERVICE_ID, "S3"); + 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); + + 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); + + MetricCollector httpClient = attempt.createChild("HttpClient"); + httpClient.reportMetric(HttpMetric.HTTP_CLIENT_NAME, S3CrtAsyncHttpClient.CLIENT_NAME); + + // TODO: map when CRT exposes them - HTTP status, endpoint URL, TTFB/TTLB, connection-pool metrics. + + MetricCollection collection = apiCall.collect(); + metricPublishers.forEach(p -> p.publish(collection)); + } catch (RuntimeException e) { + log.warn(() -> "Failed to publish CRT S3 request metrics", e); + } + } + + /** + * 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 c9138539c81c..0f7baad4a7c7 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 @@ -67,6 +69,13 @@ public final class S3InternalSdkHttpExecutionAttribute extends SdkHttpExecuti public static final S3InternalSdkHttpExecutionAttribute CRT_CREDENTIALS_PROVIDER_ADAPTER = new S3InternalSdkHttpExecutionAttribute<>(CrtCredentialsProviderAdapter.class); + /** + * Metric publishers that this request's CRT telemetry is published to. + */ + @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/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..4445a3ff6969 --- /dev/null +++ b/services/s3/src/test/java/software/amazon/awssdk/services/s3/internal/crt/S3CrtResponseHandlerAdapterMetricsTest.java @@ -0,0 +1,190 @@ +/* + * 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.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.crt.CrtRuntimeException; +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; + + @Test + public void onTelemetry_publishesApiCallTreeWithMappedValues() throws Exception { + CapturingPublisher publisher = new CapturingPublisher(); + adapterPublishingTo(publisher).onTelemetry(successfulGetObjectMetrics()); + + 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)); + + 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"); + + MetricCollection httpClient = childNamed(attempt, "HttpClient"); + assertThat(httpClient.metricValues(HttpMetric.HTTP_CLIENT_NAME)).containsExactly("s3crt"); + } + + @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(); + adapterPublishingTo(publisher).onTelemetry(metrics); + + 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_failedAttempt_reportsApiCallUnsuccessful() throws Exception { + S3RequestMetrics metrics = successfulGetObjectMetrics(); + when(metrics.isApiCallSuccessful()).thenReturn(false); + + CapturingPublisher publisher = new CapturingPublisher(); + adapterPublishingTo(publisher).onTelemetry(metrics); + + 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(); + S3CrtResponseHandlerAdapter adapter = adapterPublishingTo(publisher); + S3RequestMetrics metrics = successfulGetObjectMetrics(); + + adapter.onTelemetry(metrics); + adapter.onTelemetry(metrics); + adapter.onTelemetry(metrics); + + assertThat(publisher.collections).hasSize(3); + } + + @Test + public void onTelemetry_retriedRequest_publishesSeparateCollectionsWithIncreasingRetryCount() throws Exception { + CapturingPublisher publisher = new CapturingPublisher(); + 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); + + 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.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"); + return metrics; + } + + private static S3CrtResponseHandlerAdapter adapterPublishingTo(MetricPublisher... publishers) { + SdkHttpExecutionAttributes attributes = + SdkHttpExecutionAttributes.builder() + .put(S3InternalSdkHttpExecutionAttribute.METRIC_PUBLISHERS, Arrays.asList(publishers)) + .build(); + return new S3CrtResponseHandlerAdapter(new CompletableFuture<>(), + mock(SdkAsyncHttpResponseHandler.class), + attributes, + new CompletableFuture<>()); + } + + 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 ArrayList<>(); + + @Override + public void publish(MetricCollection metricCollection) { + collections.add(metricCollection); + } + + @Override + public void close() { + } + } +} 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 d8c55f5820cc..796f2f119675 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 @@ -46,6 +46,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; @@ -69,7 +70,7 @@ public void setup() { sdkResponseHandler = spy(new TestResponseHandler()); responseHandlerAdapter = new S3CrtResponseHandlerAdapter(future, sdkResponseHandler, - null, + SdkHttpExecutionAttributes.builder().build(), CompletableFuture.completedFuture(s3MetaRequest)); } @@ -102,7 +103,7 @@ public void successfulResponse_shouldCompleteFutureSuccessfully() throws Excepti public void s3MetaRequestNotFinish_shouldFailFuture() throws Exception { S3CrtResponseHandlerAdapter responseHandlerAdapter = new S3CrtResponseHandlerAdapter(future, sdkResponseHandler, - null, + SdkHttpExecutionAttributes.builder().build(), new CompletableFuture<>(), Duration.ofMillis(10)); int statusCode = 200;