diff --git a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java index d6d4072dd..99a4c1b6e 100644 --- a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java +++ b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsBlobStore.java @@ -36,6 +36,7 @@ import com.salesforce.multicloudj.common.exceptions.InvalidArgumentException; import com.salesforce.multicloudj.common.exceptions.ResourceNotFoundException; import com.salesforce.multicloudj.common.exceptions.SubstrateSdkException; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; import java.io.File; import java.io.InputStream; import java.io.OutputStream; @@ -98,6 +99,7 @@ public class AwsBlobStore extends AbstractBlobStore implements AwsSdkService { private final S3Client s3Client; private final AwsTransformer transformer; + private final MetricsPublisher metricsPublisher; public AwsBlobStore() { this(new Builder(), null); @@ -107,6 +109,7 @@ public AwsBlobStore(Builder builder, S3Client s3Client) { super(builder); this.s3Client = s3Client; this.transformer = builder.getTransformerSupplier().get(bucket); + this.metricsPublisher = builder.getMetricsPublisher(); } /** Helper function to determine if any of the HttpClient configuration options have been set */ @@ -722,6 +725,12 @@ public void close() { if (s3Client != null) { s3Client.close(); } + // The AWS SDK v2 does not close MetricPublishers registered via addMetricPublisher + // (ownership stays with the caller), so release the publisher here to honor the + // MetricsPublisher lifecycle contract. + if (metricsPublisher != null) { + metricsPublisher.close(); + } } @Getter @@ -756,20 +765,28 @@ private static S3Client buildS3Client(Builder builder) { if (shouldConfigureHttpClient(builder)) { b.httpClient(generateHttpClient(builder)); } - if (builder.getRetryConfig() != null) { + // A single overrideConfiguration call is used for both retry and metrics because the + // AWS SDK replaces (rather than merges) the override configuration on each call. + if (builder.getRetryConfig() != null || builder.getMetricsPublisher() != null) { // Create a temporary transformer instance for retry strategy conversion AwsTransformer transformer = builder.getTransformerSupplier().get(builder.getBucket()); b.overrideConfiguration( config -> { - config.retryStrategy(transformer.toAwsRetryStrategy(builder.getRetryConfig())); - // Set API call timeouts if provided - if (builder.getRetryConfig().getAttemptTimeout() != null) { - config.apiCallAttemptTimeout( - Duration.ofMillis(builder.getRetryConfig().getAttemptTimeout())); + if (builder.getRetryConfig() != null) { + config.retryStrategy(transformer.toAwsRetryStrategy(builder.getRetryConfig())); + // Set API call timeouts if provided + if (builder.getRetryConfig().getAttemptTimeout() != null) { + config.apiCallAttemptTimeout( + Duration.ofMillis(builder.getRetryConfig().getAttemptTimeout())); + } + if (builder.getRetryConfig().getTotalTimeout() != null) { + config.apiCallTimeout( + Duration.ofMillis(builder.getRetryConfig().getTotalTimeout())); + } } - if (builder.getRetryConfig().getTotalTimeout() != null) { - config.apiCallTimeout( - Duration.ofMillis(builder.getRetryConfig().getTotalTimeout())); + if (builder.getMetricsPublisher() != null) { + config.addMetricPublisher( + new AwsMetricsPublisherAdapter(builder.getMetricsPublisher())); } }); } diff --git a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsMetricsPublisherAdapter.java b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsMetricsPublisherAdapter.java new file mode 100644 index 000000000..56f3e23cb --- /dev/null +++ b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/AwsMetricsPublisherAdapter.java @@ -0,0 +1,63 @@ +package com.salesforce.multicloudj.blob.aws; + +import com.salesforce.multicloudj.common.observability.Metric; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; +import java.util.ArrayList; +import java.util.List; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricPublisher; +import software.amazon.awssdk.metrics.MetricRecord; + +/** + * Bridges the AWS SDK's {@link software.amazon.awssdk.metrics.MetricPublisher} SPI to the + * cloud-agnostic {@link MetricsPublisher}. + * + *

The AWS SDK reports metrics as a tree of {@link MetricCollection}s: a root collection for the + * API call with nested child collections for lower layers such as the HTTP client (where + * connection-pool metrics like {@code MaxConcurrency}, {@code LeasedConcurrency}, {@code + * PendingConcurrencyAcquires}, and {@code ConcurrencyAcquireDuration} live). This adapter flattens + * the entire tree into neutral {@link Metric} instances, tagging each with the name of the + * collection that produced it, and forwards them to the configured {@link MetricsPublisher}. + * + *

Forwarding the whole tree is intentional: the connection-pool counters that every provider + * emits are a guaranteed subset (found under the {@code HttpClient} collection), and AWS callers + * additionally receive the SDK's native request- and attempt-level metrics as a provider-specific + * superset. Consumers that only care about pool saturation can filter by the {@code HttpClient} + * category; the metric names in that category match the cloud-agnostic {@link + * com.salesforce.multicloudj.common.observability.ConnectionPoolMetrics} vocabulary. + */ +public class AwsMetricsPublisherAdapter implements MetricPublisher { + + private final MetricsPublisher delegate; + + public AwsMetricsPublisherAdapter(MetricsPublisher delegate) { + this.delegate = delegate; + } + + @Override + public void publish(MetricCollection metricCollection) { + List metrics = new ArrayList<>(); + flatten(metricCollection, metrics); + delegate.publish(metrics); + } + + private void flatten(MetricCollection collection, List out) { + String category = collection.name(); + for (MetricRecord record : collection) { + out.add( + Metric.builder() + .name(record.metric().name()) + .value(record.value()) + .category(category) + .build()); + } + for (MetricCollection child : collection.children()) { + flatten(child, out); + } + } + + @Override + public void close() { + delegate.close(); + } +} diff --git a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStore.java b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStore.java index 5074af9d5..6858a92dd 100644 --- a/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStore.java +++ b/blob/blob-aws/src/main/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStore.java @@ -3,6 +3,7 @@ import com.salesforce.multicloudj.blob.async.driver.AbstractAsyncBlobStore; import com.salesforce.multicloudj.blob.async.driver.AsyncBlobStore; import com.salesforce.multicloudj.blob.async.driver.AsyncBlobStoreProvider; +import com.salesforce.multicloudj.blob.aws.AwsMetricsPublisherAdapter; import com.salesforce.multicloudj.blob.aws.AwsSdkService; import com.salesforce.multicloudj.blob.aws.AwsTransformer; import com.salesforce.multicloudj.blob.aws.AwsTransformerSupplier; @@ -38,6 +39,7 @@ import com.salesforce.multicloudj.common.exceptions.InvalidArgumentException; import com.salesforce.multicloudj.common.exceptions.ResourceNotFoundException; import com.salesforce.multicloudj.common.exceptions.SubstrateSdkException; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; import com.salesforce.multicloudj.sts.model.CredentialsOverrider; import java.io.File; import java.io.IOException; @@ -97,6 +99,7 @@ public class AwsAsyncBlobStore extends AbstractAsyncBlobStore implements AwsSdkS private final S3AsyncClient client; private final S3TransferManager transferManager; private final AwsTransformer transformer; + private final MetricsPublisher metricsPublisher; public AwsAsyncBlobStore( String bucket, @@ -106,10 +109,31 @@ public AwsAsyncBlobStore( S3AsyncClient client, S3TransferManager transferManager, AwsTransformerSupplier transformerSupplier) { + this( + bucket, + region, + credentialsOverrider, + validator, + client, + transferManager, + transformerSupplier, + null); + } + + public AwsAsyncBlobStore( + String bucket, + String region, + CredentialsOverrider credentialsOverrider, + BlobStoreValidator validator, + S3AsyncClient client, + S3TransferManager transferManager, + AwsTransformerSupplier transformerSupplier, + MetricsPublisher metricsPublisher) { super(AwsConstants.PROVIDER_ID, bucket, region, credentialsOverrider, validator); this.client = client; this.transferManager = transferManager; this.transformer = transformerSupplier.get(bucket); + this.metricsPublisher = metricsPublisher; } @Override @@ -572,6 +596,12 @@ public void close() { if (client != null) { client.close(); } + // The AWS SDK v2 does not close MetricPublishers registered via addMetricPublisher + // (ownership stays with the caller), so release the publisher here to honor the + // MetricsPublisher lifecycle contract. + if (metricsPublisher != null) { + metricsPublisher.close(); + } } public static Builder builder() { @@ -713,21 +743,30 @@ private static void applyCommonConfig( } builder.multipartConfiguration(configBuilder.build()); - // Configure retry strategy if specified - if (config.getRetryConfig() != null) { + // Configure retry strategy and/or metrics if specified. A single overrideConfiguration + // call is used for both because the AWS SDK replaces (rather than merges) the override + // configuration on each call. + if (config.getRetryConfig() != null || config.getMetricsPublisher() != null) { // Create a temporary transformer instance for retry strategy conversion AwsTransformer transformer = config.getTransformerSupplier().get(config.getBucket()); builder.overrideConfiguration( overrideConfig -> { - overrideConfig.retryStrategy(transformer.toAwsRetryStrategy(config.getRetryConfig())); - // Set API call timeouts if provided - if (config.getRetryConfig().getAttemptTimeout() != null) { - overrideConfig.apiCallAttemptTimeout( - Duration.ofMillis(config.getRetryConfig().getAttemptTimeout())); + if (config.getRetryConfig() != null) { + overrideConfig.retryStrategy( + transformer.toAwsRetryStrategy(config.getRetryConfig())); + // Set API call timeouts if provided + if (config.getRetryConfig().getAttemptTimeout() != null) { + overrideConfig.apiCallAttemptTimeout( + Duration.ofMillis(config.getRetryConfig().getAttemptTimeout())); + } + if (config.getRetryConfig().getTotalTimeout() != null) { + overrideConfig.apiCallTimeout( + Duration.ofMillis(config.getRetryConfig().getTotalTimeout())); + } } - if (config.getRetryConfig().getTotalTimeout() != null) { - overrideConfig.apiCallTimeout( - Duration.ofMillis(config.getRetryConfig().getTotalTimeout())); + if (config.getMetricsPublisher() != null) { + overrideConfig.addMetricPublisher( + new AwsMetricsPublisherAdapter(config.getMetricsPublisher())); } }); } @@ -743,6 +782,10 @@ private static void applyCommonConfig( } } + // Note: unlike the standard S3AsyncClient path, the CRT-backed client (used when parallel + // downloads are enabled) uses a native HTTP stack rather than the SDK's metric-publisher SPI, + // so a MetricsPublisher supplied via withMetricsPublisher(...) does not emit connection-pool + // metrics on this path. This is a documented limitation; see withMetricsPublisher's javadoc. private static void applyCommonConfig( S3CrtAsyncClientBuilder builder, Builder config, Region regionObj) { // Configure region @@ -845,7 +888,8 @@ public AsyncBlobStore build() { getValidator(), client, tm, - getTransformerSupplier()); + getTransformerSupplier(), + getMetricsPublisher()); } } } diff --git a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsBlobStoreTest.java b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsBlobStoreTest.java index cdc5725b3..48b0962f3 100644 --- a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsBlobStoreTest.java +++ b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsBlobStoreTest.java @@ -52,6 +52,7 @@ import com.salesforce.multicloudj.common.exceptions.ResourceNotFoundException; import com.salesforce.multicloudj.common.exceptions.UnAuthorizedException; import com.salesforce.multicloudj.common.exceptions.UnknownException; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; import com.salesforce.multicloudj.common.retries.RetryConfig; import com.salesforce.multicloudj.sts.model.CredentialsOverrider; import com.salesforce.multicloudj.sts.model.CredentialsType; @@ -235,6 +236,105 @@ void testProviderId() { assertEquals("aws", aws.getProviderId()); } + /** + * Captures the override-configuration builder produced when a client is built, so wiring tests + * can assert what was applied to it. Re-stubs the already-open static S3Client mock. + */ + private ClientOverrideConfiguration.Builder captureOverrideConfigFor( + AwsBlobStore.Builder builder) { + ClientOverrideConfiguration.Builder configBuilder = + mock(ClientOverrideConfiguration.Builder.class); + when(configBuilder.retryStrategy(any(RetryStrategy.class))).thenReturn(configBuilder); + when(configBuilder.apiCallAttemptTimeout(any(Duration.class))).thenReturn(configBuilder); + when(configBuilder.apiCallTimeout(any(Duration.class))).thenReturn(configBuilder); + when(configBuilder.addMetricPublisher(any())).thenReturn(configBuilder); + + S3ClientBuilder mockBuilder = mock(S3ClientBuilder.class); + when(mockBuilder.region(any())).thenReturn(mockBuilder); + when(mockBuilder.credentialsProvider(any())).thenReturn(mockBuilder); + doAnswer( + invocation -> { + Consumer consumer = invocation.getArgument(0); + consumer.accept(configBuilder); + return mockBuilder; + }) + .when(mockBuilder) + .overrideConfiguration(any(Consumer.class)); + when(mockBuilder.build()).thenReturn(mock(S3Client.class)); + + s3Client.when(S3Client::builder).thenReturn(mockBuilder); + builder.build(); + return configBuilder; + } + + @Test + void testMetricsPublisherIsWiredWhenConfigured() { + RecordingMetricsPublisher publisher = new RecordingMetricsPublisher(); + AwsBlobStore.Builder builder = + (AwsBlobStore.Builder) + new AwsBlobStore.Builder() + .withTransformerSupplier(transformerSupplier) + .withBucket("bucket-1") + .withRegion("us-east-2") + .withMetricsPublisher(publisher); + + ClientOverrideConfiguration.Builder configBuilder = captureOverrideConfigFor(builder); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(software.amazon.awssdk.metrics.MetricPublisher.class); + verify(configBuilder).addMetricPublisher(captor.capture()); + Assertions.assertInstanceOf(AwsMetricsPublisherAdapter.class, captor.getValue()); + verify(configBuilder, times(0)).retryStrategy(any(RetryStrategy.class)); + } + + @Test + void testMetricsPublisherAndRetryCoexistInSingleOverride() { + RecordingMetricsPublisher publisher = new RecordingMetricsPublisher(); + RetryConfig retryConfig = + RetryConfig.builder() + .mode(RetryConfig.Mode.EXPONENTIAL) + .maxAttempts(3) + .initialDelayMillis(100L) + .multiplier(2.0) + .maxDelayMillis(5000L) + .build(); + AwsBlobStore.Builder builder = + (AwsBlobStore.Builder) + new AwsBlobStore.Builder() + .withTransformerSupplier(transformerSupplier) + .withBucket("bucket-1") + .withRegion("us-east-2") + .withRetryConfig(retryConfig) + .withMetricsPublisher(publisher); + + ClientOverrideConfiguration.Builder configBuilder = captureOverrideConfigFor(builder); + + verify(configBuilder).addMetricPublisher(any()); + verify(configBuilder).retryStrategy(any(RetryStrategy.class)); + } + + @Test + void testNoMetricPublisherWhenUnset() { + AwsBlobStore.Builder builder = + (AwsBlobStore.Builder) + new AwsBlobStore.Builder() + .withTransformerSupplier(transformerSupplier) + .withBucket("bucket-1") + .withRegion("us-east-2"); + + ClientOverrideConfiguration.Builder configBuilder = captureOverrideConfigFor(builder); + + verify(configBuilder, times(0)).addMetricPublisher(any()); + } + + /** Minimal publisher used to prove wiring without external dependencies. */ + private static final class RecordingMetricsPublisher + implements com.salesforce.multicloudj.common.observability.MetricsPublisher { + @Override + public void publish( + List metrics) {} + } + @Test void testShouldConfigureHttpClient() { var builderWithProxy = @@ -2103,6 +2203,23 @@ void testClose() { verify(mockS3Client, times(1)).close(); } + @Test + void testCloseReleasesMetricsPublisher() { + // The AWS SDK v2 does not close caller-registered MetricPublishers, so the blob store must + // release the supplied publisher itself when it closes. + MetricsPublisher metricsPublisher = mock(MetricsPublisher.class); + AwsBlobStore.Builder builder = new AwsBlobStore.Builder(); + builder.withTransformerSupplier(transformerSupplier); + builder.withBucket("bucket-1"); + builder.withRegion("us-east-2"); + builder.withMetricsPublisher(metricsPublisher); + AwsBlobStore store = builder.build(); + + store.close(); + + verify(metricsPublisher, times(1)).close(); + } + // ---- New overload: updateObjectRetention(String, String, ObjectRetentionConfig) ---- private GetObjectRetentionResponse currentRetention( diff --git a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsMetricsPublisherAdapterTest.java b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsMetricsPublisherAdapterTest.java new file mode 100644 index 000000000..d60ac852c --- /dev/null +++ b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/AwsMetricsPublisherAdapterTest.java @@ -0,0 +1,136 @@ +package com.salesforce.multicloudj.blob.aws; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.salesforce.multicloudj.common.observability.Metric; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.http.HttpMetric; +import software.amazon.awssdk.metrics.MetricCollection; +import software.amazon.awssdk.metrics.MetricCollector; + +class AwsMetricsPublisherAdapterTest { + + /** Simple capturing publisher used to assert what the adapter forwards. */ + private static final class CapturingPublisher implements MetricsPublisher { + private final List> batches = new ArrayList<>(); + private final AtomicInteger closeCount = new AtomicInteger(); + + @Override + public void publish(List metrics) { + batches.add(metrics); + } + + @Override + public void close() { + closeCount.incrementAndGet(); + } + } + + /** + * Builds a realistic AWS metric tree: an ApiCall root collection with a nested HttpClient child, + * mirroring how the SDK reports connection-pool metrics under the HTTP layer. + */ + private static MetricCollection buildApiCallTreeWithHttpChild() { + MetricCollector root = MetricCollector.create("ApiCall"); + root.reportMetric(CoreMetric.API_CALL_DURATION, Duration.ofMillis(42)); + root.reportMetric(CoreMetric.OPERATION_NAME, "GetObject"); + + MetricCollector httpChild = root.createChild("HttpClient"); + httpChild.reportMetric(HttpMetric.MAX_CONCURRENCY, 100); + httpChild.reportMetric(HttpMetric.LEASED_CONCURRENCY, 7); + httpChild.reportMetric(HttpMetric.PENDING_CONCURRENCY_ACQUIRES, 3); + + return root.collect(); + } + + @Test + void publishFlattensTreeAndTagsCategoryThenForwards() { + CapturingPublisher publisher = new CapturingPublisher(); + AwsMetricsPublisherAdapter adapter = new AwsMetricsPublisherAdapter(publisher); + + adapter.publish(buildApiCallTreeWithHttpChild()); + + assertEquals(1, publisher.batches.size()); + List forwarded = publisher.batches.get(0); + + assertEquals(5, forwarded.size()); + + Map byName = + forwarded.stream().collect(Collectors.toMap(Metric::getName, m -> m)); + + assertEquals("ApiCall", byName.get(CoreMetric.API_CALL_DURATION.name()).getCategory()); + assertEquals(Duration.ofMillis(42), byName.get(CoreMetric.API_CALL_DURATION.name()).getValue()); + assertEquals("ApiCall", byName.get(CoreMetric.OPERATION_NAME.name()).getCategory()); + assertEquals("GetObject", byName.get(CoreMetric.OPERATION_NAME.name()).getValue()); + + assertEquals("HttpClient", byName.get(HttpMetric.MAX_CONCURRENCY.name()).getCategory()); + assertEquals(100, byName.get(HttpMetric.MAX_CONCURRENCY.name()).getValue()); + assertEquals("HttpClient", byName.get(HttpMetric.LEASED_CONCURRENCY.name()).getCategory()); + assertEquals(7, byName.get(HttpMetric.LEASED_CONCURRENCY.name()).getValue()); + assertEquals( + "HttpClient", byName.get(HttpMetric.PENDING_CONCURRENCY_ACQUIRES.name()).getCategory()); + assertEquals(3, byName.get(HttpMetric.PENDING_CONCURRENCY_ACQUIRES.name()).getValue()); + } + + @Test + void publishWithNoChildrenForwardsOnlyRootMetrics() { + CapturingPublisher publisher = new CapturingPublisher(); + AwsMetricsPublisherAdapter adapter = new AwsMetricsPublisherAdapter(publisher); + + MetricCollector root = MetricCollector.create("ApiCall"); + root.reportMetric(CoreMetric.RETRY_COUNT, 2); + + adapter.publish(root.collect()); + + assertEquals(1, publisher.batches.size()); + List forwarded = publisher.batches.get(0); + assertEquals(1, forwarded.size()); + assertEquals(CoreMetric.RETRY_COUNT.name(), forwarded.get(0).getName()); + assertEquals(2, forwarded.get(0).getValue()); + assertEquals("ApiCall", forwarded.get(0).getCategory()); + } + + @Test + void publishFlattensDeeplyNestedChildren() { + CapturingPublisher publisher = new CapturingPublisher(); + AwsMetricsPublisherAdapter adapter = new AwsMetricsPublisherAdapter(publisher); + + MetricCollector root = MetricCollector.create("ApiCall"); + root.reportMetric(CoreMetric.OPERATION_NAME, "PutObject"); + MetricCollector attempt = root.createChild("ApiCallAttempt"); + attempt.reportMetric(CoreMetric.BACKOFF_DELAY_DURATION, Duration.ofMillis(10)); + MetricCollector http = attempt.createChild("HttpClient"); + http.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5); + + adapter.publish(root.collect()); + + List forwarded = publisher.batches.get(0); + assertEquals(3, forwarded.size()); + + Map categoryByName = + forwarded.stream().collect(Collectors.toMap(Metric::getName, Metric::getCategory)); + assertEquals("ApiCall", categoryByName.get(CoreMetric.OPERATION_NAME.name())); + assertEquals("ApiCallAttempt", categoryByName.get(CoreMetric.BACKOFF_DELAY_DURATION.name())); + assertEquals("HttpClient", categoryByName.get(HttpMetric.AVAILABLE_CONCURRENCY.name())); + } + + @Test + void closeDelegatesToUnderlyingPublisher() { + CapturingPublisher publisher = new CapturingPublisher(); + AwsMetricsPublisherAdapter adapter = new AwsMetricsPublisherAdapter(publisher); + + adapter.close(); + + assertEquals(1, publisher.closeCount.get()); + assertTrue(publisher.batches.isEmpty()); + } +} diff --git a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStoreTest.java b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStoreTest.java index 8b1f251d5..8581a9f73 100644 --- a/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStoreTest.java +++ b/blob/blob-aws/src/test/java/com/salesforce/multicloudj/blob/aws/async/AwsAsyncBlobStoreTest.java @@ -2021,6 +2021,91 @@ void testBuildS3AsyncClientWithRetryConfigWithTotalTimeout() { assertEquals(BUCKET, store.getBucket()); } + /** + * Captures the override-configuration builder produced when an async client is built, so wiring + * tests can assert what was applied to it. Re-stubs the already-open static S3AsyncClient mock. + */ + private ClientOverrideConfiguration.Builder captureAsyncOverrideConfigFor( + AwsAsyncBlobStore.Builder builder) { + ClientOverrideConfiguration.Builder configBuilder = + mock(ClientOverrideConfiguration.Builder.class); + when(configBuilder.retryStrategy(any(RetryStrategy.class))).thenReturn(configBuilder); + when(configBuilder.apiCallAttemptTimeout(any(Duration.class))).thenReturn(configBuilder); + when(configBuilder.apiCallTimeout(any(Duration.class))).thenReturn(configBuilder); + when(configBuilder.addMetricPublisher(any())).thenReturn(configBuilder); + + S3AsyncClientBuilder mockBuilder = mock(S3AsyncClientBuilder.class); + when(mockBuilder.region(any())).thenReturn(mockBuilder); + when(mockBuilder.credentialsProvider(any())).thenReturn(mockBuilder); + doAnswer( + invocation -> { + Consumer consumer = invocation.getArgument(0); + consumer.accept(configBuilder); + return mockBuilder; + }) + .when(mockBuilder) + .overrideConfiguration(any(Consumer.class)); + when(mockBuilder.build()).thenReturn(mock(S3AsyncClient.class)); + + s3Client.when(S3AsyncClient::builder).thenReturn(mockBuilder); + builder.build(); + return configBuilder; + } + + @Test + void testAsyncMetricsPublisherIsWiredWhenConfigured() { + AwsAsyncBlobStore.Builder builder = + (AwsAsyncBlobStore.Builder) + new AwsAsyncBlobStore.Builder() + .withBucket(BUCKET) + .withRegion(REGION) + .withMetricsPublisher(metrics -> {}); + + ClientOverrideConfiguration.Builder configBuilder = captureAsyncOverrideConfigFor(builder); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(software.amazon.awssdk.metrics.MetricPublisher.class); + verify(configBuilder).addMetricPublisher(captor.capture()); + Assertions.assertInstanceOf( + com.salesforce.multicloudj.blob.aws.AwsMetricsPublisherAdapter.class, captor.getValue()); + verify(configBuilder, times(0)).retryStrategy(any(RetryStrategy.class)); + } + + @Test + void testAsyncMetricsPublisherAndRetryCoexistInSingleOverride() { + RetryConfig retryConfig = + RetryConfig.builder() + .mode(RetryConfig.Mode.EXPONENTIAL) + .maxAttempts(3) + .initialDelayMillis(100L) + .multiplier(2.0) + .maxDelayMillis(5000L) + .build(); + AwsAsyncBlobStore.Builder builder = + (AwsAsyncBlobStore.Builder) + new AwsAsyncBlobStore.Builder() + .withBucket(BUCKET) + .withRegion(REGION) + .withRetryConfig(retryConfig) + .withMetricsPublisher(metrics -> {}); + + ClientOverrideConfiguration.Builder configBuilder = captureAsyncOverrideConfigFor(builder); + + verify(configBuilder).addMetricPublisher(any()); + verify(configBuilder).retryStrategy(any(RetryStrategy.class)); + } + + @Test + void testAsyncNoMetricPublisherWhenUnset() { + AwsAsyncBlobStore.Builder builder = + (AwsAsyncBlobStore.Builder) + new AwsAsyncBlobStore.Builder().withBucket(BUCKET).withRegion(REGION); + + ClientOverrideConfiguration.Builder configBuilder = captureAsyncOverrideConfigFor(builder); + + verify(configBuilder, times(0)).addMetricPublisher(any()); + } + @Test void testBuildS3AsyncClientWithoutRetryConfig() { // Test without retry config (default behavior) diff --git a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/async/client/AsyncBucketClient.java b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/async/client/AsyncBucketClient.java index 6383738fa..06b4c398d 100644 --- a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/async/client/AsyncBucketClient.java +++ b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/async/client/AsyncBucketClient.java @@ -30,6 +30,7 @@ import com.salesforce.multicloudj.blob.driver.UploadResponse; import com.salesforce.multicloudj.common.exceptions.ExceptionHandler; import com.salesforce.multicloudj.common.exceptions.SubstrateSdkException; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; import com.salesforce.multicloudj.common.observability.MultiCloudJLogger; import com.salesforce.multicloudj.common.observability.OperationContext; import com.salesforce.multicloudj.common.observability.TracingPolicy; @@ -765,6 +766,16 @@ public Builder withTracingPolicy(TracingPolicy tracingPolicy) { return this; } + /** + * Method to supply an opt-in publisher for cloud-agnostic client metrics (for example HTTP + * connection-pool saturation). Off by default. + */ + @Override + public Builder withMetricsPublisher(MetricsPublisher metricsPublisher) { + super.withMetricsPublisher(metricsPublisher); + return this; + } + public Builder withUseTransferListener(Boolean useTransferListener) { ((AsyncBlobStoreProvider.Builder) storeBuilder).withUseTransferListener(useTransferListener); return this; diff --git a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/client/BucketClient.java b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/client/BucketClient.java index eb9d74648..c4cd4a0af 100644 --- a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/client/BucketClient.java +++ b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/client/BucketClient.java @@ -5,6 +5,7 @@ import com.salesforce.multicloudj.blob.driver.BlobInfo; import com.salesforce.multicloudj.blob.driver.BlobMetadata; import com.salesforce.multicloudj.blob.driver.BlobSpanNames; +import com.salesforce.multicloudj.blob.driver.BlobStoreBuilder; import com.salesforce.multicloudj.blob.driver.BucketVersioningConfiguration; import com.salesforce.multicloudj.blob.driver.ByteArray; import com.salesforce.multicloudj.blob.driver.CopyFromRequest; @@ -29,6 +30,7 @@ import com.salesforce.multicloudj.blob.driver.UploadResponse; import com.salesforce.multicloudj.common.exceptions.ExceptionHandler; import com.salesforce.multicloudj.common.exceptions.SubstrateSdkException; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; import com.salesforce.multicloudj.common.observability.MultiCloudJLogger; import com.salesforce.multicloudj.common.observability.OperationContext; import com.salesforce.multicloudj.common.observability.TracingPolicy; @@ -1248,6 +1250,19 @@ public BlobBuilder withTracingPolicy(TracingPolicy tracingPolicy) { return this; } + /** + * Method to supply an opt-in publisher for cloud-agnostic client metrics (for example + * HTTP connection-pool saturation). Off by default; see {@link + * BlobStoreBuilder#withMetricsPublisher(MetricsPublisher)} for provider support details. + * + * @param metricsPublisher The publisher to receive client metrics, or {@code null} to disable. + * @return An instance of self + */ + public BlobBuilder withMetricsPublisher(MetricsPublisher metricsPublisher) { + this.blobStoreBuilder.withMetricsPublisher(metricsPublisher); + return this; + } + /** * Builds and returns an instance of BucketClient. * diff --git a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobClientBuilder.java b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobClientBuilder.java index 90c32500c..a94ada748 100644 --- a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobClientBuilder.java +++ b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobClientBuilder.java @@ -1,5 +1,6 @@ package com.salesforce.multicloudj.blob.driver; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; import com.salesforce.multicloudj.common.observability.TracingPolicy; import com.salesforce.multicloudj.common.retries.RetryConfig; import com.salesforce.multicloudj.common.service.SdkService; @@ -302,6 +303,19 @@ public BlobClientBuilder withTracingPolicy(TracingPolicy tracingPolicy) { return this; } + /** + * Method to supply an opt-in publisher for cloud-agnostic client metrics (for example HTTP + * connection-pool saturation). Off by default; see {@link + * BlobStoreBuilder#withMetricsPublisher(MetricsPublisher)} for provider support details. + * + * @param metricsPublisher The publisher to receive client metrics, or {@code null} to disable. + * @return An instance of self + */ + public BlobClientBuilder withMetricsPublisher(MetricsPublisher metricsPublisher) { + this.storeBuilder.withMetricsPublisher(metricsPublisher); + return this; + } + /** * Builds and returns an instance of the target client implementation. * diff --git a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobStoreBuilder.java b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobStoreBuilder.java index d5bc0bd17..d92fecec3 100644 --- a/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobStoreBuilder.java +++ b/blob/blob-client/src/main/java/com/salesforce/multicloudj/blob/driver/BlobStoreBuilder.java @@ -1,5 +1,6 @@ package com.salesforce.multicloudj.blob.driver; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; import com.salesforce.multicloudj.common.observability.TracingPolicy; import com.salesforce.multicloudj.common.provider.SdkProvider; import com.salesforce.multicloudj.common.retries.RetryConfig; @@ -41,6 +42,7 @@ public abstract class BlobStoreBuilder implements SdkProvi private Boolean useEnvironmentVariableProxyValues; private String quotaProjectId; private TracingPolicy tracingPolicy; + private MetricsPublisher metricsPublisher; public BlobStoreBuilder providerId(String providerId) { this.providerId = providerId; @@ -132,6 +134,46 @@ public BlobStoreBuilder withIdleConnectionTimeout(Duration idleConnectionTime return this; } + /** + * Method to supply a cloud-agnostic metrics publisher. When set, the client forwards collected + * client-level metrics (for example connection-pool saturation and request/acquisition latency) + * to this publisher. When left unset, no metrics are collected or published and client behavior + * is unchanged. + * + *

Across every provider that supports this hook, the same four connection-pool counters are + * emitted under the {@code HttpClient} category — {@code MaxConcurrency}, {@code + * LeasedConcurrency}, {@code AvailableConcurrency}, and {@code PendingConcurrencyAcquires} — so + * pool saturation can be observed uniformly regardless of the backing cloud. The underlying cloud + * SDKs expose different telemetry models, so some providers additionally emit a richer set: + * + *

+ * + *

Alibaba is not yet wired to this hook; supplying a publisher on an Alibaba-backed client is + * a no-op today and is planned for a follow-up. + * + *

Supplying a publisher is always safe: if a provider cannot produce a given metric it is + * simply not emitted, and if a provider's connection pool is unavailable the hook degrades to a + * no-op. It never alters client behavior and never throws. + * + * @param metricsPublisher The publisher to receive client metrics, or {@code null} to disable. + * @return An instance of self + */ + public BlobStoreBuilder withMetricsPublisher(MetricsPublisher metricsPublisher) { + this.metricsPublisher = metricsPublisher; + return this; + } + /** * Method to supply credentialsOverrider * diff --git a/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStore.java b/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStore.java index e192cc98e..8679d09fd 100644 --- a/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStore.java +++ b/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStore.java @@ -92,6 +92,7 @@ import com.salesforce.multicloudj.common.gcp.GcpConstants; import com.salesforce.multicloudj.common.gcp.GcpCredentialsProvider; import com.salesforce.multicloudj.common.gcp.GcpRetryClassifier; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; import com.salesforce.multicloudj.common.provider.Provider; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -129,6 +130,7 @@ import org.apache.http.client.config.RequestConfig; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -143,6 +145,7 @@ public class GcpBlobStore extends AbstractBlobStore { private final MultipartUploadClient multipartUploadClient; private final TransferManager transferManager; private final GcpTransformer transformer; + private final MetricsPublisher metricsPublisher; private static final String TAG_PREFIX = "gcp-tag-"; private static final String RESPONSE_CONTENT_DISPOSITION = "response-content-disposition"; @@ -160,6 +163,7 @@ public GcpBlobStore( this.multipartUploadClient = mpuClient; this.transferManager = transferManager; this.transformer = builder.transformerSupplier.get(bucket); + this.metricsPublisher = builder.getMetricsPublisher(); } @Override @@ -1522,6 +1526,11 @@ public void close() { if (storage != null) { storage.close(); } + // The GCS SDK does not own the publisher, so release it here to honor the + // MetricsPublisher lifecycle contract (a buffered/threaded publisher must be shut down). + if (metricsPublisher != null) { + metricsPublisher.close(); + } } catch (Exception e) { throw new SubstrateSdkException("Failed to close GCP storage clients", e); } @@ -1627,7 +1636,10 @@ private static boolean shouldConfigureHttpClient(Builder builder) { return builder.getProxyEndpoint() != null || builder.getMaxConnections() != null || builder.getSocketTimeout() != null - || builder.getIdleConnectionTimeout() != null; + || builder.getIdleConnectionTimeout() != null + // A metrics-only builder still needs the explicitly-owned transport so the pool-sampling + // interceptor in buildHttpClient() can be installed; otherwise metrics silently no-op. + || builder.getMetricsPublisher() != null; } /** Creates HttpTransportOptions with ApacheHttpTransport */ @@ -1750,15 +1762,24 @@ private static TransferManager buildTransferManager(Builder builder, Storage sto private static CloseableHttpClient buildHttpClient(Builder builder) { HttpClientBuilder httpClientBuilder = ApacheHttpTransport.newDefaultHttpClientBuilder(); httpClientBuilder.setDefaultRequestConfig(buildRequestConfig(builder)); - // Performance note (directory / many-small-object workloads): GCS traffic all targets a - // single host, so it maps to one Apache HTTP route whose default per-route connection cap - // is 20. The TransferManager used for directory operations spawns 2 x availableProcessors - // workers, which on multi-core hosts exceeds that cap and leaves workers blocked waiting for - // a connection. For such workloads, raise this via withMaxConnections (which sets both - // maxConnTotal and maxConnPerRoute below) together with withTransferManagerThreadPoolSize; - // in a controlled benchmark this roughly tripled small-file directory throughput. The knob is - // left unset by default so single-object callers keep the lean default connection footprint. - if (builder.getMaxConnections() != null) { + if (builder.getMetricsPublisher() != null) { + // Sampling pool saturation needs a handle to the pool's live stats, so install an + // explicitly-owned connection manager (seeded with the same pool sizes + // ApacheHttpTransport.newDefaultHttpClientBuilder() applies) and read from it. + PoolingHttpClientConnectionManager connectionManager = buildConnectionManager(builder); + httpClientBuilder.setConnectionManager(connectionManager); + httpClientBuilder.addInterceptorLast( + new GcpConnectionPoolMetricsInterceptor( + connectionManager::getTotalStats, builder.getMetricsPublisher())); + } else if (builder.getMaxConnections() != null) { + // Performance note (directory / many-small-object workloads): GCS traffic all targets a + // single host, so it maps to one Apache HTTP route whose default per-route connection cap + // is 20. The TransferManager used for directory operations spawns 2 x availableProcessors + // workers, which on multi-core hosts exceeds that cap and leaves workers blocked waiting + // for a connection. For such workloads, raise this via withMaxConnections (which sets both + // maxConnTotal and maxConnPerRoute below) together with withTransferManagerThreadPoolSize; + // in a controlled benchmark this roughly tripled small-file directory throughput. The + // knob is left unset by default so single-object callers keep a lean pool footprint. int maxConns = builder.getMaxConnections(); httpClientBuilder.setMaxConnTotal(maxConns); httpClientBuilder.setMaxConnPerRoute(maxConns); @@ -1770,6 +1791,21 @@ private static CloseableHttpClient buildHttpClient(Builder builder) { return httpClientBuilder.build(); } + private static PoolingHttpClientConnectionManager buildConnectionManager(Builder builder) { + PoolingHttpClientConnectionManager connectionManager = + new PoolingHttpClientConnectionManager(); + if (builder.getMaxConnections() != null) { + connectionManager.setMaxTotal(builder.getMaxConnections()); + connectionManager.setDefaultMaxPerRoute(builder.getMaxConnections()); + } else { + // Mirror ApacheHttpTransport.newDefaultHttpClientBuilder()'s pool sizes so that + // enabling metrics never silently shrinks the pool from its uninstrumented default. + connectionManager.setMaxTotal(200); + connectionManager.setDefaultMaxPerRoute(20); + } + return connectionManager; + } + private static RequestConfig buildRequestConfig(Builder builder) { RequestConfig.Builder requestConfigBuilder = RequestConfig.custom(); if (builder.getSocketTimeout() != null) { diff --git a/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpConnectionPoolMetricsInterceptor.java b/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpConnectionPoolMetricsInterceptor.java new file mode 100644 index 000000000..c9a1397cb --- /dev/null +++ b/blob/blob-gcp/src/main/java/com/salesforce/multicloudj/blob/gcp/GcpConnectionPoolMetricsInterceptor.java @@ -0,0 +1,63 @@ +package com.salesforce.multicloudj.blob.gcp; + +import com.salesforce.multicloudj.common.observability.ConnectionPoolMetrics; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; +import org.apache.http.HttpResponse; +import org.apache.http.HttpResponseInterceptor; +import org.apache.http.pool.PoolStats; +import org.apache.http.protocol.HttpContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Samples GCP HTTP connection-pool utilization and forwards it to a {@link MetricsPublisher}. + * + *

The GCP Cloud Storage client is configured with an Apache {@code + * PoolingHttpClientConnectionManager}, which exposes point-in-time pool statistics via {@code + * getTotalStats()} but provides no push callback. This interceptor is installed on the Apache HTTP + * client so that after every response the current pool statistics are read and published as + * cloud-agnostic {@link com.salesforce.multicloudj.common.observability.Metric}s, giving the + * per-request sampling cadence needed for connection-pool observability. + * + *

The hook degrades safely: if no pool manager was supplied (for example a custom transport that + * does not use pooling), or if reading statistics fails, the interceptor does nothing and the HTTP + * response is unaffected. + */ +class GcpConnectionPoolMetricsInterceptor implements HttpResponseInterceptor { + + private static final Logger logger = + LoggerFactory.getLogger(GcpConnectionPoolMetricsInterceptor.class); + + private final PoolStatsSupplier poolStatsSupplier; + + private final MetricsPublisher metricsPublisher; + + @FunctionalInterface + interface PoolStatsSupplier { + PoolStats get(); + } + + GcpConnectionPoolMetricsInterceptor( + PoolStatsSupplier poolStatsSupplier, MetricsPublisher metricsPublisher) { + this.poolStatsSupplier = poolStatsSupplier; + this.metricsPublisher = metricsPublisher; + } + + @Override + public void process(HttpResponse response, HttpContext context) { + if (metricsPublisher == null || poolStatsSupplier == null) { + return; + } + try { + PoolStats stats = poolStatsSupplier.get(); + if (stats == null) { + return; + } + metricsPublisher.publish( + ConnectionPoolMetrics.from( + stats.getMax(), stats.getLeased(), stats.getAvailable(), stats.getPending())); + } catch (RuntimeException e) { + logger.debug("Failed to sample GCP connection pool metrics", e); + } + } +} diff --git a/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStoreTest.java b/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStoreTest.java index f52d5e983..88a13fc23 100644 --- a/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStoreTest.java +++ b/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpBlobStoreTest.java @@ -2331,6 +2331,34 @@ void testBuildHttpClient_usesExplicitMaxConnections() throws Exception { assertEquals(123, connectionManager.getDefaultMaxPerRoute()); } + @Test + void testBuildHttpClient_metricsEnabledInstallsPoolWithDefaultSizing() throws Exception { + // Enabling metrics swaps in an explicitly-owned connection manager so pool stats can be + // sampled. That swap must not silently shrink the pool from the transport's 200/20 default. + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) new GcpBlobStore.Builder().withMetricsPublisher(metrics -> {}); + + PoolingHttpClientConnectionManager connectionManager = + extractConnectionManager(invokeBuildHttpClient(builder)); + + assertEquals(200, connectionManager.getMaxTotal()); + assertEquals(20, connectionManager.getDefaultMaxPerRoute()); + } + + @Test + void testBuildHttpClient_metricsEnabledHonorsExplicitMaxConnections() throws Exception { + // On the metrics path, an explicit maxConnections must still size the installed pool. + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) + new GcpBlobStore.Builder().withMaxConnections(50).withMetricsPublisher(metrics -> {}); + + PoolingHttpClientConnectionManager connectionManager = + extractConnectionManager(invokeBuildHttpClient(builder)); + + assertEquals(50, connectionManager.getMaxTotal()); + assertEquals(50, connectionManager.getDefaultMaxPerRoute()); + } + private static CloseableHttpClient invokeBuildHttpClient(GcpBlobStore.Builder builder) throws Exception { Method method = @@ -2348,6 +2376,49 @@ private static PoolingHttpClientConnectionManager extractConnectionManager( return (PoolingHttpClientConnectionManager) field.get(httpClient); } + private static org.apache.http.HttpResponseInterceptor[] extractResponseInterceptors( + CloseableHttpClient httpClient) throws Exception { + // Apache wires response interceptors into the InternalHttpClient's private "execChain": a + // stack of ClientExecChain wrappers that each delegate through a "requestExecutor" field until + // reaching the ProtocolExec, which owns the ImmutableHttpProcessor holding the interceptors. + Field execChainField = httpClient.getClass().getDeclaredField("execChain"); + execChainField.setAccessible(true); + Object node = execChainField.get(httpClient); + + Object httpProcessor = null; + while (node != null) { + Field processorField = findFieldOrNull(node.getClass(), "httpProcessor"); + if (processorField != null) { + processorField.setAccessible(true); + httpProcessor = processorField.get(node); + break; + } + Field delegateField = findFieldOrNull(node.getClass(), "requestExecutor"); + if (delegateField == null) { + break; + } + delegateField.setAccessible(true); + node = delegateField.get(node); + } + assertNotNull(httpProcessor, "could not locate httpProcessor in the client exec chain"); + + Field interceptorsField = + httpProcessor.getClass().getDeclaredField("responseInterceptors"); + interceptorsField.setAccessible(true); + return (org.apache.http.HttpResponseInterceptor[]) interceptorsField.get(httpProcessor); + } + + private static Field findFieldOrNull(Class type, String name) { + for (Class c = type; c != null; c = c.getSuperclass()) { + try { + return c.getDeclaredField(name); + } catch (NoSuchFieldException ignored) { + // keep walking up the hierarchy + } + } + return null; + } + @Test void testShouldConfigureHttpClient_falseWhenNothingSet() throws Exception { GcpBlobStore.Builder builder = new GcpBlobStore.Builder(); @@ -2385,6 +2456,33 @@ void testShouldConfigureHttpClient_trueWhenProxyEndpointSet() throws Exception { assertTrue(invokeShouldConfigureHttpClient(builder)); } + @Test + void testShouldConfigureHttpClient_trueWhenOnlyMetricsPublisherSet() throws Exception { + // Regression: a metrics-only builder must still own its transport so the pool-sampling + // interceptor gets installed; otherwise metrics silently no-op for the common enable path. + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) new GcpBlobStore.Builder().withMetricsPublisher(metrics -> {}); + assertTrue(invokeShouldConfigureHttpClient(builder)); + } + + @Test + void testBuildHttpClient_metricsOnlyInstallsPoolMetricsInterceptor() throws Exception { + // End-to-end wiring: a metrics-only builder must actually register the pool-sampling + // interceptor on the built HttpClient — not merely pass the shouldConfigureHttpClient gate. + GcpBlobStore.Builder builder = + (GcpBlobStore.Builder) new GcpBlobStore.Builder().withMetricsPublisher(metrics -> {}); + + org.apache.http.HttpResponseInterceptor[] interceptors = + extractResponseInterceptors(invokeBuildHttpClient(builder)); + + boolean installed = + java.util.Arrays.stream(interceptors) + .anyMatch(i -> i instanceof GcpConnectionPoolMetricsInterceptor); + assertTrue( + installed, + "metrics-only client must register GcpConnectionPoolMetricsInterceptor on its transport"); + } + private static boolean invokeShouldConfigureHttpClient(GcpBlobStore.Builder builder) throws Exception { Method method = diff --git a/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpConnectionPoolMetricsInterceptorTest.java b/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpConnectionPoolMetricsInterceptorTest.java new file mode 100644 index 000000000..267582e57 --- /dev/null +++ b/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/GcpConnectionPoolMetricsInterceptorTest.java @@ -0,0 +1,89 @@ +package com.salesforce.multicloudj.blob.gcp; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.salesforce.multicloudj.common.observability.ConnectionPoolMetrics; +import com.salesforce.multicloudj.common.observability.Metric; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import org.apache.http.pool.PoolStats; +import org.junit.jupiter.api.Test; + +class GcpConnectionPoolMetricsInterceptorTest { + + /** Capturing publisher used to assert what the interceptor forwards. */ + private static final class CapturingPublisher implements MetricsPublisher { + private final List> batches = new ArrayList<>(); + + @Override + public void publish(List metrics) { + batches.add(metrics); + } + } + + @Test + void processPublishesCurrentPoolStats() { + CapturingPublisher publisher = new CapturingPublisher(); + PoolStats stats = new PoolStats(7, 3, 20, 100); + GcpConnectionPoolMetricsInterceptor interceptor = + new GcpConnectionPoolMetricsInterceptor(() -> stats, publisher); + + interceptor.process(null, null); + + assertEquals(1, publisher.batches.size()); + Map byName = + publisher.batches.get(0).stream() + .collect(Collectors.toMap(Metric::getName, Metric::getValue)); + assertEquals(100, byName.get(ConnectionPoolMetrics.MAX_CONCURRENCY)); + assertEquals(7, byName.get(ConnectionPoolMetrics.LEASED_CONCURRENCY)); + assertEquals(20, byName.get(ConnectionPoolMetrics.AVAILABLE_CONCURRENCY)); + assertEquals(3, byName.get(ConnectionPoolMetrics.PENDING_CONCURRENCY_ACQUIRES)); + } + + @Test + void processIsNoOpWhenPublisherIsNull() { + GcpConnectionPoolMetricsInterceptor interceptor = + new GcpConnectionPoolMetricsInterceptor(() -> new PoolStats(1, 1, 1, 1), null); + interceptor.process(null, null); + } + + @Test + void processIsNoOpWhenPoolStatsSupplierIsNull() { + CapturingPublisher publisher = new CapturingPublisher(); + GcpConnectionPoolMetricsInterceptor interceptor = + new GcpConnectionPoolMetricsInterceptor(null, publisher); + + interceptor.process(null, null); + + assertTrue(publisher.batches.isEmpty()); + } + + @Test + void processIsNoOpWhenPoolStatsAreNull() { + CapturingPublisher publisher = new CapturingPublisher(); + GcpConnectionPoolMetricsInterceptor interceptor = + new GcpConnectionPoolMetricsInterceptor(() -> null, publisher); + + interceptor.process(null, null); + + assertTrue(publisher.batches.isEmpty()); + } + + @Test + void processSwallowsSupplierFailure() { + CapturingPublisher publisher = new CapturingPublisher(); + GcpConnectionPoolMetricsInterceptor interceptor = + new GcpConnectionPoolMetricsInterceptor( + () -> { + throw new IllegalStateException("pool unavailable"); + }, + publisher); + + interceptor.process(null, null); + assertTrue(publisher.batches.isEmpty()); + } +} diff --git a/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/MetricsSamplingBenchmark.java b/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/MetricsSamplingBenchmark.java new file mode 100644 index 000000000..da942f0c0 --- /dev/null +++ b/blob/blob-gcp/src/test/java/com/salesforce/multicloudj/blob/gcp/MetricsSamplingBenchmark.java @@ -0,0 +1,105 @@ +package com.salesforce.multicloudj.blob.gcp; + +import com.salesforce.multicloudj.common.observability.ConnectionPoolMetrics; +import com.salesforce.multicloudj.common.observability.Metric; +import com.salesforce.multicloudj.common.observability.MetricsPublisher; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.http.pool.PoolStats; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.RunnerException; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; + +/** + * Measures the per-response CPU cost that PR #532's connection-pool sampling adds to the GCP + * request path, in isolation from the network. + * + *

The overhead question for #532 cannot be answered by a live-upload wall-clock comparison: each + * upload is ~100-200 ms of network round-trip with tens of ms of jitter, while the sampling work is + * a handful of getter reads plus a small list allocation — so the network dominates the signal by + * four-to-five orders of magnitude and swamps any measurement. This microbenchmark instead + * exercises the exact production path with no I/O, so JMH can report a stable ns/op figure with a + * confidence interval. + * + *

What is measured is the real path, not a stand-in: {@link + * GcpConnectionPoolMetricsInterceptor#process} reading the pool statistics via its {@code + * PoolStatsSupplier}, translating them through {@link ConnectionPoolMetrics#from} (four {@link + * Metric} allocations plus the backing list), and handing the result to the publisher. The + * publisher is a no-op sink that forwards to a {@link Blackhole} so that JMH cannot + * dead-code-eliminate the work while the benchmark still measures sampling cost rather than any + * real sink's cost. The {@link PoolStats} value models a saturated pool (max=4, leased=4, + * available=0, pending=20) — the same shape the accuracy probe observed — so the measured path is + * representative of load. + * + *

Run standalone (recommended, clean JMH lifecycle): + * 

+ *   mvn -q test-compile -pl blob/blob-gcp
+ *   mvn -q exec:java -pl blob/blob-gcp \
+ *     -Dexec.classpathScope=test \
+ *     -Dexec.mainClass=com.salesforce.multicloudj.blob.gcp.MetricsSamplingBenchmark
+ *
+ * or invoke {@link #main} from an IDE. Reports average time per {@code process(...)} call. + */ +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 5, time = 1) +@Measurement(iterations = 5, time = 1) +@Fork(2) +@State(Scope.Thread) +public class MetricsSamplingBenchmark { + + /** No-op publisher that pushes into a Blackhole so the sampled metrics are not optimised away. */ + private static final class SinkPublisher implements MetricsPublisher { + private Blackhole blackhole; + + @Override + public void publish(List metrics) { + blackhole.consume(metrics); + } + } + + private GcpConnectionPoolMetricsInterceptor interceptor; + private SinkPublisher publisher; + + @Setup(Level.Trial) + public void setup() { + publisher = new SinkPublisher(); + // Production wires `connectionManager::getTotalStats`, which allocates a fresh PoolStats on + // every call (httpcore 4.x computes stats on demand). Model that allocation here so the + // measured cost is not understated: a saturated-pool snapshot PoolStats(leased, pending, + // available, max) constructed per invocation, the same shape the accuracy probe observed. + interceptor = + new GcpConnectionPoolMetricsInterceptor(() -> new PoolStats(4, 20, 0, 4), publisher); + } + + /** + * The full per-response sampling path: read pool stats -> {@link ConnectionPoolMetrics#from} -> + * publish. {@code response}/{@code context} are unused by the interceptor, so null is passed. + */ + @Benchmark + public void sampleAndPublish(Blackhole bh) { + publisher.blackhole = bh; + interceptor.process(null, null); + } + + public static void main(String[] args) throws RunnerException { + Options opt = + new OptionsBuilder() + .include(MetricsSamplingBenchmark.class.getSimpleName()) + .build(); + new Runner(opt).run(); + } +} diff --git a/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/ConnectionPoolMetrics.java b/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/ConnectionPoolMetrics.java new file mode 100644 index 000000000..fa306b89c --- /dev/null +++ b/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/ConnectionPoolMetrics.java @@ -0,0 +1,53 @@ +package com.salesforce.multicloudj.common.observability; + +import java.util.Arrays; +import java.util.List; + +/** + * Helper for producing cloud-agnostic connection-pool utilization {@link Metric}s. + * + *

HTTP connection pools expose a small, universal set of counters — the maximum number of + * connections the pool may hold, how many are currently leased (in use), how many are idle and + * available for reuse, and how many callers are blocked waiting to acquire one. This helper + * translates those raw counts into neutral {@link Metric} instances tagged with the {@link + * #CATEGORY_HTTP_CLIENT} category so that connection-pool saturation can be observed uniformly + * regardless of which cloud SDK produced the numbers. + * + *

The metric names mirror the vocabulary a substrate client already emits for the HTTP layer, + * so operators see a single, consistent set of names across providers. + */ +public final class ConnectionPoolMetrics { + + /** Category tagged on every connection-pool metric, identifying the HTTP client layer. */ + public static final String CATEGORY_HTTP_CLIENT = "HttpClient"; + + /** Maximum number of connections the pool is configured to hold. */ + public static final String MAX_CONCURRENCY = "MaxConcurrency"; + + /** Number of connections currently leased (checked out and in use). */ + public static final String LEASED_CONCURRENCY = "LeasedConcurrency"; + + /** Number of idle connections currently available for reuse. */ + public static final String AVAILABLE_CONCURRENCY = "AvailableConcurrency"; + + /** Number of callers currently blocked waiting to acquire a connection. */ + public static final String PENDING_CONCURRENCY_ACQUIRES = "PendingConcurrencyAcquires"; + + private ConnectionPoolMetrics() {} + + public static List from( + int maxConcurrency, + int leasedConcurrency, + int availableConcurrency, + int pendingConcurrencyAcquires) { + return Arrays.asList( + metric(MAX_CONCURRENCY, maxConcurrency), + metric(LEASED_CONCURRENCY, leasedConcurrency), + metric(AVAILABLE_CONCURRENCY, availableConcurrency), + metric(PENDING_CONCURRENCY_ACQUIRES, pendingConcurrencyAcquires)); + } + + private static Metric metric(String name, int value) { + return Metric.builder().name(name).value(value).category(CATEGORY_HTTP_CLIENT).build(); + } +} diff --git a/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/Metric.java b/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/Metric.java new file mode 100644 index 000000000..e05bf82b2 --- /dev/null +++ b/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/Metric.java @@ -0,0 +1,42 @@ +package com.salesforce.multicloudj.common.observability; + +import lombok.Builder; +import lombok.Getter; + +/** + * A single cloud-agnostic metric data point emitted by a substrate client. + * + *

This is a neutral value object with no dependency on any cloud provider's telemetry API. A + * provider implementation is responsible for translating its native metric events into instances + * of this type before handing them to a {@link MetricsPublisher}. + * + *

The {@link #category} preserves the logical layer that produced the metric (for example {@code + * "ApiCall"} or {@code "HttpClient"}) so operators can distinguish, e.g., connection-pool + * saturation metrics from request-level metrics without needing provider-specific types. + * + *

Provider-neutral contract. The only metrics guaranteed on every supported provider are + * the four connection-pool counters defined in {@link ConnectionPoolMetrics} ({@code + * MaxConcurrency}, {@code LeasedConcurrency}, {@code AvailableConcurrency}, {@code + * PendingConcurrencyAcquires}), all emitted under {@link ConnectionPoolMetrics#CATEGORY_HTTP_CLIENT + * "HttpClient"} with an {@link Integer} {@link #value}. Any other name/category (for example an + * AWS-shaped {@code ApiCall}/{@code ApiCallAttempt} metric, whose value may be a {@code Duration}, + * {@code String}, or numeric type) is a provider-specific superset that is not portable across + * clouds. A consumer written to the neutral contract should filter by category {@code "HttpClient"} + * and the four names above, and must not assume the {@link #value} type of any other metric. + */ +@Builder +@Getter +public class Metric { + + /** The name of the metric, for example {@code "MaxConcurrency"} or {@code "ApiCallDuration"}. */ + private final String name; + + /** The recorded value of the metric. */ + private final Object value; + + /** + * The logical layer that produced this metric (for example {@code "ApiCall"} or {@code + * "HttpClient"}). May be {@code null} when the producing layer is unknown. + */ + private final String category; +} diff --git a/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/MetricsPublisher.java b/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/MetricsPublisher.java new file mode 100644 index 000000000..f6ace6168 --- /dev/null +++ b/multicloudj-common/src/main/java/com/salesforce/multicloudj/common/observability/MetricsPublisher.java @@ -0,0 +1,32 @@ +package com.salesforce.multicloudj.common.observability; + +import java.util.List; + +/** + * Cloud-agnostic sink for client-level metrics. + * + *

Substrate clients (blob, docstore, etc.) collect metrics from the underlying cloud SDK and, + * when a publisher is configured, forward them here as neutral {@link Metric} instances. This + * surface intentionally has no dependency on any provider's telemetry API so a single operator + * implementation can receive metrics regardless of which cloud is backing the client. + * + *

Implementations may be invoked concurrently from multiple threads and must be thread-safe. + * The {@link #publish(List)} method should return quickly, deferring any expensive aggregation or + * transmission to a background thread, and must never propagate an exception to the caller. + */ +public interface MetricsPublisher { + + /** + * Notifies the publisher of a batch of newly collected metrics. + * + * @param metrics the metrics collected for a single operation; never {@code null} + */ + void publish(List metrics); + + /** + * Releases any resources held by the publisher. The default implementation does nothing. + * + *

Called when the owning client is closed. + */ + default void close() {} +} diff --git a/multicloudj-common/src/test/java/com/salesforce/multicloudj/common/observability/ConnectionPoolMetricsTest.java b/multicloudj-common/src/test/java/com/salesforce/multicloudj/common/observability/ConnectionPoolMetricsTest.java new file mode 100644 index 000000000..1185197e2 --- /dev/null +++ b/multicloudj-common/src/test/java/com/salesforce/multicloudj/common/observability/ConnectionPoolMetricsTest.java @@ -0,0 +1,43 @@ +package com.salesforce.multicloudj.common.observability; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import org.junit.jupiter.api.Test; + +class ConnectionPoolMetricsTest { + + @Test + void fromProducesFourMetricsWithHttpClientCategory() { + List metrics = ConnectionPoolMetrics.from(100, 7, 20, 3); + + assertEquals(4, metrics.size()); + metrics.forEach(m -> assertEquals(ConnectionPoolMetrics.CATEGORY_HTTP_CLIENT, m.getCategory())); + } + + @Test + void fromMapsRawCountsToNamedMetrics() { + List metrics = ConnectionPoolMetrics.from(100, 7, 20, 3); + Map byName = + metrics.stream().collect(Collectors.toMap(Metric::getName, Metric::getValue)); + + assertEquals(100, byName.get(ConnectionPoolMetrics.MAX_CONCURRENCY)); + assertEquals(7, byName.get(ConnectionPoolMetrics.LEASED_CONCURRENCY)); + assertEquals(20, byName.get(ConnectionPoolMetrics.AVAILABLE_CONCURRENCY)); + assertEquals(3, byName.get(ConnectionPoolMetrics.PENDING_CONCURRENCY_ACQUIRES)); + } + + @Test + void fromHandlesZeroedPoolCounts() { + List metrics = ConnectionPoolMetrics.from(0, 0, 0, 0); + Map byName = + metrics.stream().collect(Collectors.toMap(Metric::getName, Metric::getValue)); + + assertEquals(0, byName.get(ConnectionPoolMetrics.MAX_CONCURRENCY)); + assertEquals(0, byName.get(ConnectionPoolMetrics.LEASED_CONCURRENCY)); + assertEquals(0, byName.get(ConnectionPoolMetrics.AVAILABLE_CONCURRENCY)); + assertEquals(0, byName.get(ConnectionPoolMetrics.PENDING_CONCURRENCY_ACQUIRES)); + } +}