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 index d5109324dec2..d996f2c42f62 100644 --- 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 @@ -37,13 +37,15 @@ 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.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 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. + * 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 { @@ -134,6 +136,87 @@ void failedGetObject_publishesUnsuccessfulApiCallMetrics() throws InterruptedExc 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"); 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 067b30f4e10d..0316182230b7 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 @@ -31,10 +31,12 @@ 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; @@ -82,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; @@ -93,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"); private static final String CRT_CLIENT_CLASSPATH = "software.amazon.awssdk.crt.s3.S3Client"; private final CopyObjectHelper copyObjectHelper; @@ -137,7 +142,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) { @@ -460,7 +493,9 @@ public void afterMarshalling(Context.AfterMarshalling context, .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_PATH, executionAttributes.getAttribute(RESPONSE_FILE_PATH)) .put(S3InternalSdkHttpExecutionAttribute.RESPONSE_FILE_OPTION, - executionAttributes.getAttribute(RESPONSE_FILE_OPTION)); + executionAttributes.getAttribute(RESPONSE_FILE_OPTION)) + .put(S3InternalSdkHttpExecutionAttribute.METRIC_PUBLISHERS, + executionAttributes.getAttribute(REQUEST_METRIC_PUBLISHERS)); SdkRequest request = context.request(); if (request instanceof AwsRequest) { @@ -517,10 +552,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 96a642932080..4b4a897e1431 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 @@ -166,14 +166,15 @@ 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. + List requestMetricPublishers = + httpExecutionAttributes.getAttribute(S3InternalSdkHttpExecutionAttribute.METRIC_PUBLISHERS); + List effectivePublishers = resolveEffectiveMetricPublishers(requestMetricPublishers, + metricPublishers); SdkHttpExecutionAttributes adapterAttributes = httpExecutionAttributes; - if (!metricPublishers.isEmpty()) { + if (!effectivePublishers.isEmpty()) { adapterAttributes = httpExecutionAttributes.toBuilder() .put(S3InternalSdkHttpExecutionAttribute.METRIC_PUBLISHERS, - metricPublishers) + effectivePublishers) .build(); } @@ -232,6 +233,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 AwsSigningConfig awsSigningConfig(Region signingRegion, SdkHttpExecutionAttributes httpExecutionAttributes) { CrtCredentialsProviderAdapter requestAdapter = httpExecutionAttributes.getAttribute(S3InternalSdkHttpExecutionAttribute.CRT_CREDENTIALS_PROVIDER_ADAPTER); 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 0f7baad4a7c7..f207927e8871 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 @@ -70,11 +70,12 @@ public final class S3InternalSdkHttpExecutionAttribute extends SdkHttpExecuti new S3InternalSdkHttpExecutionAttribute<>(CrtCredentialsProviderAdapter.class); /** - * Metric publishers that this request's CRT telemetry is published to. + * The metric publishers this request's CRT telemetry is published to: the request-level publishers if the request + * set any, otherwise the client-level publishers. The CRT transport resolves the two and folds the effective set in. */ @SuppressWarnings("unchecked") public static final S3InternalSdkHttpExecutionAttribute> METRIC_PUBLISHERS = - new S3InternalSdkHttpExecutionAttribute<>((Class>) (Class) List.class); + 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 3baf717a1e03..f9b1bbb3b8fe 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 @@ -32,6 +32,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; @@ -181,9 +182,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()); } @@ -197,10 +200,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) -> { @@ -237,10 +242,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); } @@ -250,10 +257,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); } @@ -269,6 +278,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())); @@ -282,7 +292,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 05acbb596371..6bd38e982625 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 @@ -54,6 +54,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; @@ -68,10 +69,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; @@ -977,6 +981,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); + } + private static CopyObjectRequest.Builder copyRequestBuilder() { return CopyObjectRequest.builder() .sourceBucket(SOURCE_BUCKET) 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 147ca786703e..3e11fd3123f0 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 @@ -38,6 +38,10 @@ import com.github.tomakehurst.wiremock.stubbing.Scenario; import java.net.URI; import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executor; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; @@ -50,8 +54,11 @@ import software.amazon.awssdk.core.ResponseInputStream; import software.amazon.awssdk.core.async.AsyncResponseTransformer; import software.amazon.awssdk.core.exception.SdkClientException; +import software.amazon.awssdk.core.metrics.CoreMetric; import software.amazon.awssdk.crt.CrtResource; 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; @@ -244,10 +251,58 @@ 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); + }); + } + } + private static class SpyableExecutor implements Executor { @Override 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/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); + } +}