Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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 */
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()));
}
});
}
Expand Down
Original file line number Diff line number Diff line change
@@ -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}.
*
* <p>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}.
*
* <p>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<Metric> metrics = new ArrayList<>();
flatten(metricCollection, metrics);
delegate.publish(metrics);
}

private void flatten(MetricCollection collection, List<Metric> 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();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The CRT-backed async client doesn't wire the metrics publisher.

This wires the adapter for the standard S3AsyncClientBuilder path, but the S3CrtAsyncClientBuilder overload (applyCommonConfig around line 757) has no equivalent — so withParallelDownloadsEnabled(true) + withMetricsPublisher(...) silently emits nothing (the CRT path is taken when parallel downloads are enabled).

This is defensible since CRT uses a native HTTP stack rather than the SDK metric SPI, but the builder javadoc states AWS support for "both the synchronous and asynchronous clients" with no caveat. Could we either add a one-line javadoc note that the CRT-backed async path (parallel downloads) does not emit metrics, or log a debug message when a publisher is set on the CRT path — so it's a documented limitation rather than a silent gap?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Went with the javadoc route. Added a caveat on BlobStoreBuilder.withMetricsPublisher noting the CRT-backed async client (parallel downloads) uses a native HTTP stack and doesn't emit connection-pool metrics, plus a matching comment on the CRT applyCommonConfig so it's documented at the code site too. Made the AWS bullet in the javadoc explicit about the exception rather than the unqualified "both sync and async" it promised before.

new AwsMetricsPublisherAdapter(config.getMetricsPublisher()));
}
});
}
Expand All @@ -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
Expand Down Expand Up @@ -845,7 +888,8 @@ public AsyncBlobStore build() {
getValidator(),
client,
tm,
getTransformerSupplier());
getTransformerSupplier(),
getMetricsPublisher());
}
}
}
Loading
Loading