diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java index 0793d4614223..36a57febed45 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java @@ -984,7 +984,9 @@ public Storage create(StorageOptions options) { OpenTelemetry otel = options.getOpenTelemetry(); DefaultRetrier retrier = new DefaultRetrier( - OtelStorageDecorator.retryContextDecorator(otel), grpcStorageOptions); + OtelStorageDecorator.retryContextDecorator(otel), + grpcStorageOptions, + grpcStorageOptions.getStorageClientMetrics()); if (ZeroCopyReadinessChecker.isReady()) { LOGGER.config("zero-copy protobuf deserialization available, using it"); StorageStubSettings baseSettings = diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OpenTelemetryBootstrappingUtils.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OpenTelemetryBootstrappingUtils.java index d83ce6bb91d9..8a0bef3db50f 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OpenTelemetryBootstrappingUtils.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OpenTelemetryBootstrappingUtils.java @@ -26,6 +26,7 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; +import com.google.common.net.HostAndPort; import io.grpc.ManagedChannelBuilder; import io.grpc.opentelemetry.GrpcOpenTelemetry; import io.opentelemetry.api.common.AttributeKey; @@ -103,28 +104,12 @@ static ChannelConfigurator enableGrpcMetrics( GCPResourceProvider resourceProvider = new GCPResourceProvider(); Attributes detectedAttributes = resourceProvider.getAttributes(); - @Nullable String detectedProjectId = - detectedAttributes.get(AttributeKey.stringKey("cloud.account.id")); - if (projectId == null && detectedProjectId == null) { - log.warning( - "Unable to determine the Project ID in order to report metrics. No gRPC client metrics" - + " will be reported."); + String projectIdToUse = + resolveProjectIdForMetrics(detectedAttributes, projectId, "gRPC client metrics"); + if (projectIdToUse == null) { return channelConfigurator != null ? channelConfigurator : ChannelConfigurator.identity(); } - String projectIdToUse = detectedProjectId == null ? projectId : detectedProjectId; - if (!projectIdToUse.equals(projectId)) { - log.warning( - "The Project ID configured for gRPC client metrics is " - + projectIdToUse - + ", but the Project ID of the storage client is " - + projectId - + ". Make sure that the service account in use has the required metric writing role " - + "(roles/monitoring.metricWriter) in the project " - + projectIdToUse - + ", or metrics will not be written."); - } - String metricServiceEndpoint = getCloudMonitoringEndpoint(endpoint, universeDomain); SdkMeterProvider provider = createMeterProvider( @@ -206,7 +191,143 @@ static String getCloudMonitoringEndpoint(String endpoint, String universeDomain) metricServiceEndpoint = match; } } - return metricServiceEndpoint + ":" + endpoint.split(":")[1]; + HostAndPort parsed = parseHostAndPort(endpoint); + int port = parsed != null ? parsed.getPortOrDefault(443) : 443; + return metricServiceEndpoint + ":" + port; + } + + @Nullable + private static HostAndPort parseHostAndPort(@Nullable String host) { + if (host == null || host.trim().isEmpty()) { + return null; + } + String s = host.trim(); + if (s.startsWith("google-c2p:///")) { + s = s.substring("google-c2p:///".length()); + } else if (s.startsWith("https://")) { + s = s.substring("https://".length()); + } else if (s.startsWith("http://")) { + s = s.substring("http://".length()); + } + int slashIdx = s.indexOf('/'); + if (slashIdx >= 0) { + s = s.substring(0, slashIdx); + } + if (s.isEmpty()) { + return null; + } + try { + return HostAndPort.fromString(s); + } catch (IllegalArgumentException e) { + return null; + } + } + + static String formatStorageEndpoint(@Nullable String host) { + HostAndPort parsed = parseHostAndPort(host); + if (parsed == null) { + return "storage.googleapis.com:443"; + } + return parsed.withDefaultPort(443).toString(); + } + + @Nullable + static String extractServerAddress(@Nullable String host) { + HostAndPort parsed = parseHostAndPort(host); + return parsed != null ? parsed.getHost() : null; + } + + @Nullable + private static String resolveProjectIdForMetrics( + Attributes detectedAttributes, + @Nullable String configuredProjectId, + String metricsTypeLabel) { + @Nullable String detectedProjectId = + detectedAttributes.get(AttributeKey.stringKey("cloud.account.id")); + if (detectedProjectId != null && detectedProjectId.trim().isEmpty()) { + detectedProjectId = null; + } + @Nullable String cleanConfiguredProjectId = + (configuredProjectId != null && !configuredProjectId.trim().isEmpty()) + ? configuredProjectId.trim() + : null; + if (cleanConfiguredProjectId == null && detectedProjectId == null) { + log.warning( + "Unable to determine the Project ID in order to report metrics. No " + + metricsTypeLabel + + " will be reported."); + return null; + } + String projectIdToUse = + detectedProjectId != null ? detectedProjectId : cleanConfiguredProjectId; + if (cleanConfiguredProjectId != null && !projectIdToUse.equals(cleanConfiguredProjectId)) { + log.warning( + "The Project ID configured for " + + metricsTypeLabel + + " is " + + projectIdToUse + + ", but the Project ID of the storage client is " + + cleanConfiguredProjectId + + ". Make sure that the service account in use has the required metric writing" + + " role (roles/monitoring.metricWriter) in the project " + + projectIdToUse + + ", or metrics will not be written."); + } + return projectIdToUse; + } + + @Nullable + static SdkMeterProvider createDefaultClientMeterProvider( + @Nullable String host, + @Nullable String configuredProjectId, + @Nullable String universeDomain, + @Nullable Duration metricInterval, + String api) { + try { + GCPResourceProvider resourceProvider = new GCPResourceProvider(); + Attributes detectedAttributes = resourceProvider.getAttributes(); + return createDefaultClientMeterProvider( + host, configuredProjectId, universeDomain, metricInterval, api, detectedAttributes); + } catch (Exception e) { + log.warning("Failed to create default client MeterProvider: " + e.getMessage()); + return null; + } + } + + @VisibleForTesting + @Nullable + static SdkMeterProvider createDefaultClientMeterProvider( + @Nullable String host, + @Nullable String configuredProjectId, + @Nullable String universeDomain, + @Nullable Duration metricInterval, + String api, + Attributes detectedAttributes) { + try { + String projectIdToUse = + resolveProjectIdForMetrics( + detectedAttributes, configuredProjectId, "storage client metrics"); + if (projectIdToUse == null) { + return null; + } + Attributes attributesToUse = + Attributes.builder() + .putAll(detectedAttributes) + .put(AttributeKey.stringKey("cloud.account.id"), projectIdToUse) + .build(); + String formattedEndpoint = formatStorageEndpoint(host); + String metricServiceEndpoint = getCloudMonitoringEndpoint(formattedEndpoint, universeDomain); + return createClientMeterProvider( + metricServiceEndpoint, + projectIdToUse, + attributesToUse, + metricInterval, + /* shouldSuppressExceptions= */ true, + api); + } catch (Exception e) { + log.warning("Failed to create default client MeterProvider: " + e.getMessage()); + return null; + } } static final ImmutableList CLIENT_LATENCY_HISTOGRAMS = diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java index d2f46d959e4c..a3b48b90ac19 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java @@ -22,18 +22,21 @@ import com.google.api.gax.retrying.BasicResultRetryAlgorithm; import com.google.api.gax.retrying.ResultRetryAlgorithm; import com.google.api.gax.retrying.RetrySettings; +import com.google.cloud.BaseServiceException; import com.google.cloud.storage.Backoff.Jitterer; import com.google.cloud.storage.Conversions.Decoder; import com.google.cloud.storage.spi.v1.HttpRpcContext; import com.google.common.base.MoreObjects; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import io.opentelemetry.context.Scope; import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.function.UnaryOperator; import org.checkerframework.checker.nullness.qual.NonNull; +import org.checkerframework.checker.nullness.qual.Nullable; final class Retrying { @@ -43,7 +46,7 @@ final class Retrying { */ interface Retrier { /** - * A convenience method to invoke a callable, and possibly retry it again if and exception is + * A convenience method to invoke a callable, and possibly retry it again if an exception is * thrown. If the result of {@code c} is non-null, Decoder {@code d} will be applied to the * result before returning. * @@ -54,18 +57,32 @@ interface Retrier { * T}, can optionally return null * @param d A post process mapping {@link Function} which can be used to transform the result * from {@code c} if it is successful and non-null + * @param metricsContext optional metrics context for tracking logical operation metrics * @return A {@code U} (possibly null) after applying {@code f} to the result of {@code c} * @throws StorageException if {@code c} fails due to any retry exhaustion */ Model run( - ResultRetryAlgorithm alg, Callable c, Decoder d); + ResultRetryAlgorithm alg, + Callable c, + Decoder d, + @Nullable StorageMetricsContext metricsContext); + + default Model run( + ResultRetryAlgorithm alg, Callable c, Decoder d) { + return run(alg, c, d, null); + } - /** - * Produce a new {@link RetrierWithAlg} where the provided ResultRetryAlgorithm is bound for the - * life of the produced instance. - */ default RetrierWithAlg withAlg(ResultRetryAlgorithm alg) { - return new RetrierWithAlgImpl(this, alg); + return withAlg(alg, null); + } + + default RetrierWithAlg withAlg( + ResultRetryAlgorithm alg, @Nullable StorageMetricsContext metricsContext) { + return new DefaultRetrierWithAlg(this, alg, metricsContext); + } + + default Retrier withMetricsContext(@Nullable StorageMetricsContext metricsContext) { + return metricsContext == null ? this : new ContextBoundRetrier(this, metricsContext); } static Retrier attemptOnce() { @@ -80,22 +97,13 @@ static Retrier attemptOnce() { */ interface RetrierWithAlg extends Retrier { - /** - * A convenience method to invoke a callable, and possibly retry it again if and exception is - * thrown. If the result of {@code c} is non-null, Decoder {@code d} will be applied to the - * result before returning. - * - * @param The result type of {@code c} - * @param The result type of any mapping that takes place via {@code d} - * @param c The {@link Callable} which will be passed to runWithRetries producing some {@code - * T}, can optionally return null - * @param d A post process mapping {@link Function} which can be used to transform the result - * from {@code c} if it is successful and non-null - * @return A {@code U} (possibly null) after applying {@code f} to the result of {@code c} - * @throws StorageException if {@code c} fails due to any retry exhaustion - */ Model run(Callable c, Decoder d); + Model run( + Callable c, + Decoder d, + @Nullable StorageMetricsContext metricsContext); + static RetrierWithAlg attemptOnce() { return AttemptOnceRetrier.INSTANCE_WITH_ALG; } @@ -107,8 +115,15 @@ static final class AttemptOnceRetrier implements Retrier { @Override public Model run( - ResultRetryAlgorithm alg, Callable c, Decoder d) { - try { + ResultRetryAlgorithm alg, + Callable c, + Decoder d, + @Nullable StorageMetricsContext metricsContext) { + try (Scope ignore = + metricsContext != null ? metricsContext.getOtelContext().makeCurrent() : null) { + if (metricsContext != null) { + metricsContext.incrementAttempts(); + } Response call = c.call(); return call == null ? null : d.decode(call); } catch (Exception iae) { @@ -117,75 +132,176 @@ public Model run( } } - static final class RetrierWithAlgImpl implements RetrierWithAlg { + static final class ContextBoundRetrier implements Retrier { + private final Retrier delegate; + private final StorageMetricsContext metricsContext; + + ContextBoundRetrier(Retrier delegate, StorageMetricsContext metricsContext) { + this.delegate = delegate; + this.metricsContext = metricsContext; + } + + @Override + public Model run( + ResultRetryAlgorithm alg, + Callable c, + Decoder d, + @Nullable StorageMetricsContext explicitCtx) { + return delegate.run(alg, c, d, explicitCtx != null ? explicitCtx : metricsContext); + } + + @Override + public Model run( + ResultRetryAlgorithm alg, Callable c, Decoder d) { + return delegate.run(alg, c, d, metricsContext); + } + + @Override + public RetrierWithAlg withAlg(ResultRetryAlgorithm alg) { + return delegate.withAlg(alg, metricsContext); + } + + @Override + public RetrierWithAlg withAlg( + ResultRetryAlgorithm alg, @Nullable StorageMetricsContext explicitCtx) { + return delegate.withAlg(alg, explicitCtx != null ? explicitCtx : metricsContext); + } + + @Override + public Retrier withMetricsContext(@Nullable StorageMetricsContext explicitCtx) { + return explicitCtx != null ? new ContextBoundRetrier(delegate, explicitCtx) : this; + } + } + + static final class DefaultRetrierWithAlg implements RetrierWithAlg { private final Retrier retrier; private final ResultRetryAlgorithm alg; + @Nullable private final StorageMetricsContext boundMetricsContext; - private RetrierWithAlgImpl(Retrier retrier, ResultRetryAlgorithm alg) { + private DefaultRetrierWithAlg( + Retrier retrier, + ResultRetryAlgorithm alg, + @Nullable StorageMetricsContext boundMetricsContext) { this.retrier = retrier; this.alg = alg; + this.boundMetricsContext = boundMetricsContext; + } + + @Override + public Model run(Callable c, Decoder d) { + return retrier.run(alg, c, d, boundMetricsContext); } @Override public Model run( ResultRetryAlgorithm alg, Callable c, Decoder d) { - return retrier.run(alg, c, d); + return retrier.run(alg, c, d, boundMetricsContext); } @Override - public Model run(Callable c, Decoder d) { - return retrier.run(alg, c, d); + public Model run( + Callable c, + Decoder d, + @Nullable StorageMetricsContext metricsContext) { + return retrier.run(alg, c, d, metricsContext != null ? metricsContext : boundMetricsContext); + } + + @Override + public Model run( + ResultRetryAlgorithm alg, + Callable c, + Decoder d, + @Nullable StorageMetricsContext metricsContext) { + return retrier.run(alg, c, d, metricsContext != null ? metricsContext : boundMetricsContext); } } static final class DefaultRetrier implements Retrier { private final UnaryOperator decorator; private final RetryingDependencies deps; + @Nullable private final StorageClientMetrics metrics; DefaultRetrier(UnaryOperator decorator, RetryingDependencies deps) { + this(decorator, deps, null); + } + + DefaultRetrier( + UnaryOperator decorator, + RetryingDependencies deps, + @Nullable StorageClientMetrics metrics) { this.decorator = decorator; this.deps = deps; + this.metrics = metrics; } @Override public Model run( - ResultRetryAlgorithm alg, Callable c, Decoder d) { - RetryContext ctx = - decorator.apply( - RetryContext.of( - RetryContext.directScheduledExecutorService(), - deps, - alg, - Jitterer.threadLocalRandom())); - AtomicReference failure = new AtomicReference<>(); - AtomicBoolean attemptAgain = new AtomicBoolean(false); - do { - attemptAgain.set(false); + ResultRetryAlgorithm alg, + Callable c, + Decoder d, + @Nullable StorageMetricsContext metricsContext) { + try (Scope ignore = + metricsContext != null ? metricsContext.getOtelContext().makeCurrent() : null) { + RetryContext ctx = + decorator.apply( + RetryContext.of( + RetryContext.directScheduledExecutorService(), + deps, + alg, + Jitterer.threadLocalRandom())); + AtomicReference failure = new AtomicReference<>(); + AtomicBoolean attemptAgain = new AtomicBoolean(false); + Throwable terminalThrowable = null; + try { - Response result = c.call(); - return result == null ? null : d.decode(result); - } catch (StorageException se) { - // we hope for this case - ctx.recordError(se, () -> attemptAgain.set(true), failure::set); - } catch (IllegalArgumentException iae) { - // IllegalArgumentException can happen if there is no json in the body and we try - // to parse it Our retry algorithms have special case for this, so in an effort to - // keep compatibility with those existing behaviors, explicitly rethrow an - // IllegalArgumentException that may have happened - ctx.recordError(iae, () -> attemptAgain.set(true), failure::set); - } catch (Exception e) { - // Wire in this fall through just in case. - // all of our retry algorithms are centered around StorageException so this helps - // those be more effective - ctx.recordError(StorageException.coalesce(e), () -> attemptAgain.set(true), failure::set); + do { + attemptAgain.set(false); + if (metricsContext != null) { + metricsContext.incrementAttempts(); + } + try { + Response result = c.call(); + Model decoded = result == null ? null : d.decode(result); + if (metricsContext != null) { + metricsContext.recordAttemptSuccess(metrics); + } + return decoded; + } catch (StorageException se) { + if (metricsContext != null) { + metricsContext.recordAttemptFailure(metrics, se); + } + ctx.recordError(se, () -> attemptAgain.set(true), failure::set); + } catch (IllegalArgumentException iae) { + if (metricsContext != null) { + metricsContext.recordAttemptFailure(metrics, iae); + } + ctx.recordError(iae, () -> attemptAgain.set(true), failure::set); + } catch (Exception e) { + if (metricsContext != null) { + metricsContext.recordAttemptFailure(metrics, e); + } + BaseServiceException coalesced = StorageException.coalesce(e); + ctx.recordError(coalesced, () -> attemptAgain.set(true), failure::set); + } + } while (attemptAgain.get()); + + Exception throwable = failure.get(); + terminalThrowable = throwable; + if (throwable instanceof StorageException) { + throw (StorageException) throwable; + } else if (throwable != null) { + throw StorageException.coalesce(throwable); + } else { + throw new StorageException(0, "Retry failed without exception"); + } + } catch (Throwable t) { + terminalThrowable = t; + throw t; + } finally { + if (metricsContext != null && metricsContext.shouldRecordOperationOnRetrierExit()) { + metricsContext.recordOperation(metrics, terminalThrowable); + } } - } while (attemptAgain.get()); - - Exception throwable = failure.get(); - if (throwable instanceof StorageException) { - throw (StorageException) throwable; - } else { - throw StorageException.coalesce(throwable); } } } @@ -199,11 +315,14 @@ static final class HttpRetrier implements Retrier { @Override public Model run( - ResultRetryAlgorithm alg, Callable c, Decoder d) { + ResultRetryAlgorithm alg, + Callable c, + Decoder d, + @Nullable StorageMetricsContext metricsContext) { HttpRpcContext httpRpcContext = HttpRpcContext.getInstance(); try { httpRpcContext.newInvocationId(); - return delegate.run(alg, c, d); + return delegate.run(alg, c, d, metricsContext); } finally { httpRpcContext.clearInvocationId(); } @@ -212,10 +331,21 @@ public Model run( @NonNull static GrpcCallContext newCallContext() { - return GrpcCallContext.createDefault() - .withExtraHeaders( - ImmutableMap.of( - "x-goog-gcs-idempotency-token", ImmutableList.of(UUID.randomUUID().toString()))); + return newCallContext(null); + } + + @NonNull + static GrpcCallContext newCallContext(@Nullable StorageMetricsContext metricsContext) { + GrpcCallContext context = + GrpcCallContext.createDefault() + .withExtraHeaders( + ImmutableMap.of( + "x-goog-gcs-idempotency-token", + ImmutableList.of(UUID.randomUUID().toString()))); + if (metricsContext != null) { + context = metricsContext.attach(context); + } + return context; } static ResultRetryAlgorithm neverRetry() { diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageClientMetrics.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageClientMetrics.java index d7e1d2308e67..b7e80c4aee02 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageClientMetrics.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageClientMetrics.java @@ -16,6 +16,7 @@ package com.google.cloud.storage; +import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.common.AttributesBuilder; @@ -25,11 +26,13 @@ import io.opentelemetry.api.metrics.LongUpDownCounter; import io.opentelemetry.api.metrics.Meter; import io.opentelemetry.api.metrics.MeterProvider; -import java.util.UUID; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import java.util.logging.Logger; import org.checkerframework.checker.nullness.qual.Nullable; /** Package-private instrument registry for OpenTelemetry client metrics. */ -final class StorageClientMetrics { +final class StorageClientMetrics implements AutoCloseable { + private static final Logger LOGGER = Logger.getLogger(StorageClientMetrics.class.getName()); static final String METER_NAME = "com.google.cloud.storage"; // Standard client attribute keys and values @@ -39,8 +42,6 @@ final class StorageClientMetrics { AttributeKey.stringKey("gcp.client.service"); static final AttributeKey KEY_GCP_CLIENT_ARTIFACT = AttributeKey.stringKey("gcp.client.artifact"); - static final AttributeKey KEY_GCP_CLIENT_INSTANCE_ID = - AttributeKey.stringKey("gcp.client.instance_id"); static final AttributeKey KEY_GCP_STORAGE_OPERATION = AttributeKey.stringKey("gcp.storage.operation"); @@ -69,11 +70,6 @@ final class StorageClientMetrics { static final String NETWORK_IO_DIRECTION_RECEIVE = "receive"; static final String NETWORK_IO_DIRECTION_TRANSMIT = "transmit"; - static final String ATTRIBUTE_GCP_CLIENT_VERSION = "gcp.client.version"; - static final String ATTRIBUTE_GCP_CLIENT_SERVICE = "gcp.client.service"; - static final String ATTRIBUTE_GCP_CLIENT_ARTIFACT = "gcp.client.artifact"; - static final String ATTRIBUTE_GCP_CLIENT_INSTANCE_ID = "gcp.client.instance_id"; - static final String SERVICE_STORAGE = "storage"; static final String ARTIFACT_STORAGE = "com.google.cloud:google-cloud-storage"; @@ -103,34 +99,65 @@ final class StorageClientMetrics { static final String METRIC_GCP_STORAGE_CLIENT_AUTH_CREDENTIAL_REFRESH_DURATION = "gcp.storage.client.auth.credential_refresh.duration"; - @Nullable private final DoubleHistogram rpcClientCallDuration; - @Nullable private final DoubleHistogram httpClientRequestDuration; - @Nullable private final DoubleHistogram gcpClientRequestDuration; - @Nullable private final LongCounter operations; - @Nullable private final LongCounter attempts; - @Nullable private final LongCounter errors; - @Nullable private final DoubleHistogram ttfb; - @Nullable private final LongHistogram requestBodySize; - @Nullable private final LongHistogram responseBodySize; - - @Nullable private final LongUpDownCounter requestActive; - @Nullable private final DoubleHistogram serverDuration; - @Nullable private final LongCounter serverUnreached; - @Nullable private final DoubleHistogram stallDuration; - @Nullable private final LongCounter networkIo; - @Nullable private final DoubleHistogram credentialRefreshDuration; - - private final Attributes clientAttributes; + @Nullable + static StorageClientMetrics fromOptions(StorageOptions options) { + if (!options.isEnableOtelMetrics()) { + return null; + } + MeterProvider mp = options.getMeterProvider(); + if (mp == null && options.isCustomOpenTelemetryConfigured()) { + OpenTelemetry otel = options.getOpenTelemetry(); + if (otel != null) { + mp = otel.getMeterProvider(); + } + } + SdkMeterProvider ownedMeterProvider = null; + if (mp == null) { + ownedMeterProvider = + OpenTelemetryBootstrappingUtils.createDefaultClientMeterProvider( + options.getHost(), + options.getProjectId(), + options.getUniverseDomain(), + options.getMetricInterval(), + options instanceof GrpcStorageOptions ? "grpc" : "json"); + mp = ownedMeterProvider; + } + if (mp == null) { + return null; + } + Meter meter = + mp.meterBuilder(METER_NAME).setInstrumentationVersion(StorageOptions.version()).build(); + Attributes clientAttributes = + Attributes.builder() + .put(KEY_GCP_CLIENT_SERVICE, SERVICE_STORAGE) + .put(KEY_GCP_CLIENT_VERSION, StorageOptions.version()) + .put(KEY_GCP_CLIENT_ARTIFACT, ARTIFACT_STORAGE) + .build(); + AttributesBuilder baseBuilder = + clientAttributes.toBuilder() + .put(KEY_RPC_SYSTEM_NAME, options instanceof GrpcStorageOptions ? "grpc" : "http"); + String serverAddress = OpenTelemetryBootstrappingUtils.extractServerAddress(options.getHost()); + if (serverAddress != null && !serverAddress.isEmpty()) { + baseBuilder.put(KEY_SERVER_ADDRESS, serverAddress); + } + return new StorageClientMetrics( + meter, + clientAttributes, + baseBuilder.build(), + ownedMeterProvider, + /* enableOtelMetrics= */ true, + options.isEnableOtelDebugMetrics()); + } static StorageClientMetrics create( MeterProvider meterProvider, boolean enableOtelMetrics, boolean enableOtelDebugMetrics) { return create( - meterProvider, UUID.randomUUID().toString(), enableOtelMetrics, enableOtelDebugMetrics); + meterProvider, (SdkMeterProvider) null, enableOtelMetrics, enableOtelDebugMetrics); } static StorageClientMetrics create( MeterProvider meterProvider, - String instanceId, + @Nullable SdkMeterProvider ownedMeterProvider, boolean enableOtelMetrics, boolean enableOtelDebugMetrics) { Meter meter = @@ -143,31 +170,47 @@ static StorageClientMetrics create( .put(KEY_GCP_CLIENT_SERVICE, SERVICE_STORAGE) .put(KEY_GCP_CLIENT_VERSION, StorageOptions.version()) .put(KEY_GCP_CLIENT_ARTIFACT, ARTIFACT_STORAGE) - .put(KEY_GCP_CLIENT_INSTANCE_ID, instanceId) .build(); return new StorageClientMetrics( - meter, clientAttributes, enableOtelMetrics, enableOtelDebugMetrics); - } - - StorageClientMetrics(Meter meter, boolean enableOtelMetrics, boolean enableOtelDebugMetrics) { - this( meter, - Attributes.builder() - .put(KEY_GCP_CLIENT_SERVICE, SERVICE_STORAGE) - .put(KEY_GCP_CLIENT_VERSION, StorageOptions.version()) - .put(KEY_GCP_CLIENT_ARTIFACT, ARTIFACT_STORAGE) - .put(KEY_GCP_CLIENT_INSTANCE_ID, UUID.randomUUID().toString()) - .build(), + clientAttributes, + clientAttributes, + ownedMeterProvider, enableOtelMetrics, enableOtelDebugMetrics); } - StorageClientMetrics( + @Nullable private final DoubleHistogram rpcClientCallDuration; + @Nullable private final DoubleHistogram httpClientRequestDuration; + @Nullable private final DoubleHistogram gcpClientRequestDuration; + @Nullable private final LongCounter operations; + @Nullable private final LongCounter attempts; + @Nullable private final LongCounter errors; + @Nullable private final DoubleHistogram ttfb; + @Nullable private final LongHistogram requestBodySize; + @Nullable private final LongHistogram responseBodySize; + + @Nullable private final LongUpDownCounter requestActive; + @Nullable private final DoubleHistogram serverDuration; + @Nullable private final LongCounter serverUnreached; + @Nullable private final DoubleHistogram stallDuration; + @Nullable private final LongCounter networkIo; + @Nullable private final DoubleHistogram credentialRefreshDuration; + + private final Attributes clientAttributes; + private final Attributes baseAttributes; + @Nullable private final SdkMeterProvider ownedMeterProvider; + + private StorageClientMetrics( Meter meter, Attributes clientAttributes, + Attributes baseAttributes, + @Nullable SdkMeterProvider ownedMeterProvider, boolean enableOtelMetrics, boolean enableOtelDebugMetrics) { this.clientAttributes = clientAttributes; + this.baseAttributes = baseAttributes; + this.ownedMeterProvider = ownedMeterProvider; if (enableOtelMetrics) { this.rpcClientCallDuration = @@ -396,4 +439,35 @@ Attributes attachClientAttributes(@Nullable Attributes other) { @Nullable DoubleHistogram getCredentialRefreshDuration() { return credentialRefreshDuration; } + + StorageMetricsContext createContext(String operationName) { + return StorageMetricsContext.create(operationName, baseAttributes); + } + + StorageMetricsContext createMultiStepContext(String operationName) { + return StorageMetricsContext.createForMultiStep(operationName, baseAttributes); + } + + @Nullable + static StorageMetricsContext createContext( + @Nullable StorageClientMetrics metrics, String operationName) { + return metrics != null ? metrics.createContext(operationName) : null; + } + + @Nullable + static StorageMetricsContext createMultiStepContext( + @Nullable StorageClientMetrics metrics, String operationName) { + return metrics != null ? metrics.createMultiStepContext(operationName) : null; + } + + @Override + public void close() { + if (ownedMeterProvider != null) { + try { + ownedMeterProvider.close(); + } catch (Exception e) { + LOGGER.warning("Failed to close storageClientMetrics: " + e.getMessage()); + } + } + } } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java index 672c9ce34041..846c8b1f3787 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java @@ -81,7 +81,8 @@ public abstract class StorageOptions extends ServiceOptions collected; + + @Before + public void setUp() { + reader = InMemoryMetricReader.create(); + provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + metrics = StorageClientMetrics.create(provider, true, false); + deps = RetryingDependencies.simple(NanoClock.getDefaultClock(), RETRY_SETTINGS); + baseRetrier = new DefaultRetrier(UnaryOperator.identity(), deps, metrics); + alwaysRetrier = baseRetrier.withAlg(Retrying.alwaysRetry()); + neverRetrier = baseRetrier.withAlg(Retrying.neverRetry()); + collected = null; + } + + @Test + public void successfulCallOnFirstAttempt_recordsExpectedMetrics() { + String result = + alwaysRetrier.run( + () -> "success", Decoder.identity(), StorageMetricsContext.create("GetObject")); + assertThat(result).isEqualTo("success"); + assertOutcomeMetrics(STATUS_OK, 1L, 0L); + assertThat(assertDuration(STATUS_OK).getAttributes().get(KEY_GCP_CLIENT_SERVICE)) + .isEqualTo("storage"); + } + + @Test + public void retriedCallSuccessOnThirdAttempt_recordsAttemptsAndErrors() { + AtomicInteger attemptCounter = new AtomicInteger(0); + String result = + alwaysRetrier.run( + () -> { + int count = attemptCounter.incrementAndGet(); + if (count < 3) { + throw new StorageException(503, "Service Unavailable attempt " + count); + } + return "recovered"; + }, + Decoder.identity(), + StorageMetricsContext.create("GetObject")); + + assertThat(result).isEqualTo("recovered"); + assertThat(attemptCounter.get()).isEqualTo(3); + assertOutcomeMetrics(STATUS_OK, 3L, 2L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_ERROR_TYPE, STATUS_UNAVAILABLE, 2L); + assertDuration(STATUS_OK); + } + + @Test + public void decoderThrowsException_recordsOnlyAttemptFailureNotSuccess() { + StorageException se = + assertThrows( + StorageException.class, + () -> + neverRetrier.run( + () -> "raw-response", + raw -> { + throw new IllegalArgumentException("decode failed: " + raw); + }, + StorageMetricsContext.create("GetObject"))); + + assertThat(se.getCause()).isInstanceOf(IllegalArgumentException.class); + assertOutcomeMetrics(STATUS_UNKNOWN, 1L, 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_ERROR_TYPE, STATUS_OK, 0L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_ERROR_TYPE, STATUS_UNKNOWN, 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_ERROR_TYPE, STATUS_UNKNOWN, 1L); + assertDuration(STATUS_UNKNOWN); + } + + @Test + public void terminalFailureAllAttemptsFail_recordsExhaustedRetries() { + AtomicInteger attemptCounter = new AtomicInteger(0); + StorageException se = + assertThrows( + StorageException.class, + () -> + alwaysRetrier.run( + () -> { + int count = attemptCounter.incrementAndGet(); + throw new StorageException(503, "Permanent failure attempt " + count); + }, + Decoder.identity(), + StorageMetricsContext.create("GetObject"))); + + assertThat(attemptCounter.get()).isEqualTo(3); + assertThat(se.getCode()).isEqualTo(503); + assertOutcomeMetrics(STATUS_UNAVAILABLE, 3L, 3L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_ERROR_TYPE, STATUS_UNAVAILABLE, 3L); + assertDuration(STATUS_UNAVAILABLE); + } + + @Test + public void metricsDisabled_recordsNoMetrics() { + StorageClientMetrics disabledMetrics = StorageClientMetrics.create(provider, false, false); + RetrierWithAlg retrier = + new DefaultRetrier(UnaryOperator.identity(), deps, disabledMetrics) + .withAlg(Retrying.alwaysRetry()); + + assertThat(retrier.run(() -> "no-op metrics", Decoder.identity())).isEqualTo("no-op metrics"); + assertThat(collect()).isEmpty(); + } + + @Test + public void nullMetrics_executesWithoutRecording() { + RetrierWithAlg retrier = + new DefaultRetrier(UnaryOperator.identity(), deps, null).withAlg(Retrying.alwaysRetry()); + + assertThat(retrier.run(() -> "null-metrics", Decoder.identity())).isEqualTo("null-metrics"); + } + + @Test + public void callingRunWithoutMetricsContext_whenMetricsEnabled_recordsNoMetrics() { + assertThat(alwaysRetrier.run(() -> "no-context", Decoder.identity())).isEqualTo("no-context"); + assertThat(collect()).isEmpty(); + } + + @Test + public void httpStorageOptions_createRetrier_wiresMetricsWhenEnabled() { + assertCreateRetrierWiresMetricsWhenEnabled( + HttpStorageOptions.newBuilder() + .setMeterProvider(provider) + .setEnableOtelMetrics(true) + .build()); + } + + @Test + public void httpStorageOptions_createRetrier_noMetricsWhenDisabled() { + assertCreateRetrierNoMetricsWhenDisabled( + HttpStorageOptions.newBuilder() + .setMeterProvider(provider) + .setEnableOtelMetrics(false) + .build()); + } + + @Test + public void grpcStorageOptions_createRetrier_wiresMetricsWhenEnabled() { + assertCreateRetrierWiresMetricsWhenEnabled( + GrpcStorageOptions.newBuilder() + .setMeterProvider(provider) + .setEnableOtelMetrics(true) + .build()); + } + + @Test + public void grpcStorageOptions_createRetrier_noMetricsWhenDisabled() { + assertCreateRetrierNoMetricsWhenDisabled( + GrpcStorageOptions.newBuilder() + .setMeterProvider(provider) + .setEnableOtelMetrics(false) + .build()); + } + + @Test + public void nonRetryableError_singleAttemptTerminalFailure_recordsErrorAndTerminalOperation() { + StorageException se = + assertThrows( + StorageException.class, + () -> + neverRetrier.run( + () -> { + throw new StorageException(404, "Not Found"); + }, + Decoder.identity(), + StorageMetricsContext.create("GetObject"))); + + assertThat(se.getCode()).isEqualTo(404); + assertOutcomeMetrics(STATUS_NOT_FOUND, 1L, 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_ERROR_TYPE, STATUS_NOT_FOUND, 1L); + assertDuration(STATUS_NOT_FOUND); + } + + @Test + public void recordAttemptError_specificExceptions_mappedToCanonicalTypes() { + AtomicInteger attemptCounter = new AtomicInteger(0); + StorageException se = + assertThrows( + StorageException.class, + () -> + alwaysRetrier.run( + () -> { + int count = attemptCounter.incrementAndGet(); + if (count == 1) { + throw new StatusRuntimeException(Status.DEADLINE_EXCEEDED); + } else if (count == 2) { + throw new SocketTimeoutException("Read timed out"); + } else { + throw new UnknownHostException("storage.googleapis.com"); + } + }, + Decoder.identity(), + StorageMetricsContext.create("GetObject"))); + + assertThat(attemptCounter.get()).isEqualTo(3); + assertThat(se).isNotNull(); + assertOutcomeMetrics(STATUS_DNS_FAILURE, 3L, 3L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_ERROR_TYPE, STATUS_TIMEOUT, 2L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_ERROR_TYPE, STATUS_DNS_FAILURE, 1L); + } + + @Test + public void illegalArgumentException_parsingFailure_recordedAndRethrown() { + StorageException se = + assertThrows( + StorageException.class, + () -> + neverRetrier.run( + () -> { + throw new IllegalArgumentException("malformed json response"); + }, + Decoder.identity(), + StorageMetricsContext.create("GetObject"))); + + assertThat(se.getCause()).isInstanceOf(IllegalArgumentException.class); + assertThat(se.getCause()).hasMessageThat().contains("malformed json response"); + assertOutcomeMetrics(STATUS_UNKNOWN, 1L, 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_ERROR_TYPE, STATUS_UNKNOWN, 1L); + } + + @Test + public void run_withCustomMetricsContext_recordsSpecificOperationAttribute() { + StorageMetricsContext customContext = StorageMetricsContext.create("GetObject"); + String result = + baseRetrier.run( + Retrying.neverRetry(), () -> "custom-op-result", Decoder.identity(), customContext); + assertThat(result).isEqualTo("custom-op-result"); + + assertAttr(METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_GCP_STORAGE_OPERATION, "GetObject", 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_GCP_CLIENT_METHOD, "GetObject", 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_RPC_METHOD, "GetObject", 0L); + + HistogramPointData durationPoint = assertDuration(STATUS_OK); + assertThat(durationPoint.getAttributes().get(KEY_GCP_STORAGE_OPERATION)).isEqualTo("GetObject"); + assertThat(durationPoint.getAttributes().get(KEY_GCP_CLIENT_METHOD)).isEqualTo("GetObject"); + assertThat(durationPoint.getAttributes().get(KEY_RPC_METHOD)) + .isEqualTo("google.storage.v2.Storage/GetObject"); + + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_GCP_STORAGE_OPERATION, "GetObject", 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_RPC_METHOD, "GetObject", 0L); + } + + @Test + public void withAlg_withCustomMetricsContext_propagatesContext() { + StorageMetricsContext customContext = StorageMetricsContext.create("GetObject"); + String result = neverRetrier.run(() -> "custom-op-result", Decoder.identity(), customContext); + assertThat(result).isEqualTo("custom-op-result"); + + assertAttr(METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_GCP_STORAGE_OPERATION, "GetObject", 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_GCP_STORAGE_OPERATION, "GetObject", 1L); + } + + @Test + public void newCallContext_attachesMetricsContext() { + StorageMetricsContext customContext = StorageMetricsContext.create("GetObject"); + GrpcCallContext callContext = Retrying.newCallContext(customContext); + assertThat(StorageMetricsContext.from(callContext)).isSameInstanceAs(customContext); + assertThat(StorageMetricsContext.from(Retrying.newCallContext(null))).isNull(); + } + + @Test + public void run_withCustomMetricsContext_recordsOperationOnRetriesAndErrors() { + StorageMetricsContext customContext = StorageMetricsContext.create("GetObject"); + AtomicInteger attempts = new AtomicInteger(0); + String result = + baseRetrier.run( + Retrying.alwaysRetry(), + () -> { + if (attempts.incrementAndGet() < 2) { + throw new StorageException(503, "Unavailable"); + } + return "recovered"; + }, + Decoder.identity(), + customContext); + assertThat(result).isEqualTo("recovered"); + + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_GCP_STORAGE_OPERATION, "GetObject", 2L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_RPC_METHOD, "GetObject", 0L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_GCP_STORAGE_OPERATION, "GetObject", 1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_RPC_METHOD, "GetObject", 0L); + } + + @Test + public void run_withCustomMetricsContext_propagatesOtelContextScope() { + StorageMetricsContext customContext = StorageMetricsContext.create("GetObject"); + Context[] observedContext = new Context[1]; + + baseRetrier.run( + Retrying.neverRetry(), + () -> { + observedContext[0] = Context.current(); + return "ok"; + }, + Decoder.identity(), + customContext); + + assertThat(observedContext[0]).isEqualTo(customContext.getOtelContext()); + } + + @Test + public void withAlg_withMultiStepMetricsContext_runTwoArgs_recordsAttemptsAndErrors() { + StorageMetricsContext multiContext = StorageMetricsContext.createForMultiStep("MultiStepOp"); + RetrierWithAlg retrier = baseRetrier.withAlg(Retrying.alwaysRetry(), multiContext); + AtomicInteger attempts = new AtomicInteger(0); + + String result = + retrier.run( + () -> { + if (attempts.incrementAndGet() < 2) { + throw new StorageException(503, "Unavailable"); + } + return "recovered"; + }, + Decoder.identity()); + + assertThat(result).isEqualTo("recovered"); + assertThat(attempts.get()).isEqualTo(2); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_OPERATIONS)).isEqualTo(0L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ATTEMPTS)).isEqualTo(2L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ERRORS)).isEqualTo(1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_GCP_STORAGE_OPERATION, "MultiStepOp", 2L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_GCP_STORAGE_OPERATION, "MultiStepOp", 1L); + + multiContext.recordOperation(metrics, null); + collected = null; + assertOutcomeMetrics(STATUS_OK, 2L, 1L); + assertDuration(STATUS_OK); + } + + @Test + public void withAlg_withMultiStepMetricsContext_runThreeArgs_recordsAttemptsAndErrors() { + StorageMetricsContext multiContext = StorageMetricsContext.createForMultiStep("MultiStepOp"); + RetrierWithAlg retrier = baseRetrier.withAlg(Retrying.neverRetry(), multiContext); + AtomicInteger attempts = new AtomicInteger(0); + + String result = + retrier.run( + Retrying.alwaysRetry(), + () -> { + if (attempts.incrementAndGet() < 2) { + throw new StorageException(503, "Unavailable"); + } + return "recovered"; + }, + Decoder.identity()); + + assertThat(result).isEqualTo("recovered"); + assertThat(attempts.get()).isEqualTo(2); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_OPERATIONS)).isEqualTo(0L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ATTEMPTS)).isEqualTo(2L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ERRORS)).isEqualTo(1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_GCP_STORAGE_OPERATION, "MultiStepOp", 2L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_GCP_STORAGE_OPERATION, "MultiStepOp", 1L); + + multiContext.recordOperation(metrics, null); + collected = null; + assertOutcomeMetrics(STATUS_OK, 2L, 1L); + assertDuration(STATUS_OK); + } + + @Test + public void withMetricsContext_runThreeArgs_recordsAttemptsAndErrors() { + StorageMetricsContext multiContext = StorageMetricsContext.createForMultiStep("MultiStepOp"); + Retrier contextBoundRetrier = baseRetrier.withMetricsContext(multiContext); + AtomicInteger attempts = new AtomicInteger(0); + + String result = + contextBoundRetrier.run( + Retrying.alwaysRetry(), + () -> { + if (attempts.incrementAndGet() < 2) { + throw new StorageException(503, "Unavailable"); + } + return "recovered"; + }, + Decoder.identity()); + + assertThat(result).isEqualTo("recovered"); + assertThat(attempts.get()).isEqualTo(2); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_OPERATIONS)).isEqualTo(0L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ATTEMPTS)).isEqualTo(2L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ERRORS)).isEqualTo(1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_GCP_STORAGE_OPERATION, "MultiStepOp", 2L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_GCP_STORAGE_OPERATION, "MultiStepOp", 1L); + + multiContext.recordOperation(metrics, null); + collected = null; + assertOutcomeMetrics(STATUS_OK, 2L, 1L); + assertDuration(STATUS_OK); + } + + @Test + public void withMetricsContext_withAlg_runTwoArgs_recordsAttemptsAndErrors() { + StorageMetricsContext multiContext = StorageMetricsContext.createForMultiStep("MultiStepOp"); + Retrier contextBoundRetrier = baseRetrier.withMetricsContext(multiContext); + RetrierWithAlg retrierWithAlg = contextBoundRetrier.withAlg(Retrying.alwaysRetry()); + AtomicInteger attempts = new AtomicInteger(0); + + String result = + retrierWithAlg.run( + () -> { + if (attempts.incrementAndGet() < 2) { + throw new StorageException(503, "Unavailable"); + } + return "recovered"; + }, + Decoder.identity()); + + assertThat(result).isEqualTo("recovered"); + assertThat(attempts.get()).isEqualTo(2); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_OPERATIONS)).isEqualTo(0L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ATTEMPTS)).isEqualTo(2L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ERRORS)).isEqualTo(1L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, KEY_GCP_STORAGE_OPERATION, "MultiStepOp", 2L); + assertAttr(METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_GCP_STORAGE_OPERATION, "MultiStepOp", 1L); + + multiContext.recordOperation(metrics, null); + collected = null; + assertOutcomeMetrics(STATUS_OK, 2L, 1L); + assertDuration(STATUS_OK); + } + + private void assertCreateRetrierWiresMetricsWhenEnabled(StorageOptions options) { + assertThat(options.getStorageClientMetrics()).isNotNull(); + Retrier retrier = options.createRetrier(); + assertThat(retrier).isNotNull(); + + StorageMetricsContext ctx = options.getStorageClientMetrics().createContext("GetObject"); + assertThat(retrier.run(Retrying.neverRetry(), () -> "options-success", Decoder.identity(), ctx)) + .isEqualTo("options-success"); + assertAttr(METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, STATUS_OK, 1L); + } + + private void assertCreateRetrierNoMetricsWhenDisabled(StorageOptions options) { + assertThat(options.getStorageClientMetrics()).isNull(); + Retrier retrier = options.createRetrier(); + assertThat(retrier.run(Retrying.neverRetry(), () -> "options-disabled", Decoder.identity())) + .isEqualTo("options-disabled"); + assertThat(collect()).isEmpty(); + } + + private Collection collect() { + if (collected == null) { + collected = reader.collectAllMetrics(); + } + return collected; + } + + private void assertOutcomeMetrics( + String terminalStatus, long expectedAttempts, long expectedErrors) { + assertAttr(METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, terminalStatus, 1L); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ATTEMPTS)) + .isEqualTo(expectedAttempts); + assertThat(getCounterValue(collect(), METRIC_GCP_STORAGE_CLIENT_ERRORS)) + .isEqualTo(expectedErrors); + } + + private void assertAttr( + String metricName, AttributeKey key, String expectedValue, long expectedCount) { + assertThat(getCounterValueWithAttribute(collect(), metricName, key, expectedValue)) + .isEqualTo(expectedCount); + } + + private HistogramPointData assertDuration(String expectedTerminalStatus) { + HistogramPointData point = getHistogramPoint(collect(), METRIC_GCP_CLIENT_REQUEST_DURATION); + assertThat(point).isNotNull(); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getAttributes().get(KEY_ERROR_TYPE)).isEqualTo(expectedTerminalStatus); + return point; + } + + private static @Nullable MetricData findMetric( + Collection metrics, String metricName) { + for (MetricData md : metrics) { + if (md.getName().equals(metricName) || md.getName().equals(metricName.replace(".", "/"))) { + return md; + } + } + return null; + } + + private static long getCounterValue(Collection metrics, String metricName) { + MetricData md = findMetric(metrics, metricName); + return md == null + ? 0L + : md.getLongSumData().getPoints().stream().mapToLong(LongPointData::getValue).sum(); + } + + private static long getCounterValueWithAttribute( + Collection metrics, + String metricName, + AttributeKey key, + String expectedValue) { + MetricData md = findMetric(metrics, metricName); + if (md == null) { + return 0L; + } + return md.getLongSumData().getPoints().stream() + .filter(p -> expectedValue.equals(p.getAttributes().get(key))) + .mapToLong(LongPointData::getValue) + .sum(); + } + + private static @Nullable HistogramPointData getHistogramPoint( + Collection metrics, String metricName) { + MetricData md = findMetric(metrics, metricName); + if (md == null || !md.getHistogramData().getPoints().iterator().hasNext()) { + return null; + } + return md.getHistogramData().getPoints().iterator().next(); + } +} diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageClientMetricsTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageClientMetricsTest.java index 9d3b8fa95e4d..505aab2280d2 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageClientMetricsTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageClientMetricsTest.java @@ -21,14 +21,21 @@ import com.google.common.collect.ImmutableList; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.metrics.Aggregation; +import io.opentelemetry.sdk.metrics.InstrumentType; import io.opentelemetry.sdk.metrics.SdkMeterProvider; import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; import io.opentelemetry.sdk.metrics.data.HistogramPointData; import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.export.CollectionRegistration; +import io.opentelemetry.sdk.metrics.export.MetricReader; import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import java.util.stream.Collectors; import org.junit.Before; @@ -79,8 +86,6 @@ public void standardMetrics_initializedWithoutDebug() { .isEqualTo(StorageOptions.version()); assertThat(metrics.getClientAttributes().get(StorageClientMetrics.KEY_GCP_CLIENT_ARTIFACT)) .isEqualTo("com.google.cloud:google-cloud-storage"); - assertThat(metrics.getClientAttributes().get(StorageClientMetrics.KEY_GCP_CLIENT_INSTANCE_ID)) - .isNotEmpty(); assertInstruments(metrics, DEBUG_INSTRUMENTS, /* isNull= */ true); } @@ -275,8 +280,7 @@ public void latencyHistogram_recordsSecondsScaleValuesIntoExpectedBuckets() { @Test public void attachClientAttributes_mergesAdditionalAttributes() { SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); - StorageClientMetrics metrics = - StorageClientMetrics.create(provider, "test-instance", true, false); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); Attributes extra = Attributes.of( @@ -287,8 +291,6 @@ public void attachClientAttributes_mergesAdditionalAttributes() { Attributes merged = metrics.attachClientAttributes(extra); - assertThat(merged.get(StorageClientMetrics.KEY_GCP_CLIENT_INSTANCE_ID)) - .isEqualTo("test-instance"); assertThat(merged.get(StorageClientMetrics.KEY_GCP_CLIENT_VERSION)) .isEqualTo(StorageOptions.version()); assertThat(merged.get(StorageClientMetrics.KEY_GCP_CLIENT_ARTIFACT)) @@ -307,8 +309,7 @@ public void attachClientAttributes_mergesAdditionalAttributes() { @Test public void clientAttributesBuilder_containsBaseClientAttributes() { SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); - StorageClientMetrics metrics = - StorageClientMetrics.create(provider, "test-instance", true, false); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); Attributes built = metrics.clientAttributesBuilder().build(); assertThat(built).isEqualTo(metrics.getClientAttributes()); @@ -323,6 +324,126 @@ public void attributeKeys_haveExpectedNames() { .isEqualTo("http.response.status_code"); } + @Test + public void close_closesOwnedMeterProvider() { + TestMetricReader reader = new TestMetricReader(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, provider, true, false); + metrics.close(); + assertThat(reader.shutdownCalled.get()).isTrue(); + } + + @Test + public void close_doesNotThrowWhenOwnedMeterProviderIsNull() { + TestMetricReader reader = new TestMetricReader(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = + StorageClientMetrics.create(provider, (SdkMeterProvider) null, true, false); + metrics.close(); + assertThat(reader.shutdownCalled.get()).isFalse(); + } + + @Test + public void createContext_httpOptions_populatesRpcSystemNameAndServerAddress() { + SdkMeterProvider meterProvider = SdkMeterProvider.builder().build(); + HttpStorageOptions httpOptions = + StorageOptions.http() + .setEnableOtelMetrics(true) + .setMeterProvider(meterProvider) + .setHost("https://storage.googleapis.com") + .build(); + StorageClientMetrics httpMetrics = httpOptions.getStorageClientMetrics(); + assertThat(httpMetrics).isNotNull(); + + StorageMetricsContext httpContext = httpMetrics.createContext("GetObject"); + + assertThat(httpContext).isNotNull(); + assertThat(httpContext.getBaseAttributes().get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME)) + .isEqualTo("http"); + assertThat(httpContext.getBaseAttributes().get(StorageClientMetrics.KEY_SERVER_ADDRESS)) + .isEqualTo("storage.googleapis.com"); + assertThat(httpContext.shouldRecordOperationOnRetrierExit()).isTrue(); + } + + @Test + public void createMultiStepContext_disablesRecordOperationOnRetrierExit() { + SdkMeterProvider meterProvider = SdkMeterProvider.builder().build(); + HttpStorageOptions httpOptions = + StorageOptions.http() + .setEnableOtelMetrics(true) + .setMeterProvider(meterProvider) + .setHost("https://storage.googleapis.com") + .build(); + StorageClientMetrics httpMetrics = httpOptions.getStorageClientMetrics(); + + StorageMetricsContext httpMultiStep = + StorageClientMetrics.createMultiStepContext(httpMetrics, "WriteObject"); + + assertThat(httpMultiStep).isNotNull(); + assertThat(httpMultiStep.getBaseAttributes().get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME)) + .isEqualTo("http"); + assertThat(httpMultiStep.shouldRecordOperationOnRetrierExit()).isFalse(); + } + + @Test + public void createContext_grpcOptions_extractsServerAddressFromC2pUri() { + SdkMeterProvider meterProvider = SdkMeterProvider.builder().build(); + GrpcStorageOptions grpcOptions = + StorageOptions.grpc() + .setEnableOtelMetrics(true) + .setMeterProvider(meterProvider) + .setHost("google-c2p:///storage.googleapis.com") + .build(); + StorageClientMetrics grpcMetrics = grpcOptions.getStorageClientMetrics(); + assertThat(grpcMetrics).isNotNull(); + + StorageMetricsContext grpcContext = + StorageClientMetrics.createContext(grpcMetrics, "GetObject"); + + assertThat(grpcContext).isNotNull(); + assertThat(grpcContext.getBaseAttributes().get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME)) + .isEqualTo("grpc"); + assertThat(grpcContext.getBaseAttributes().get(StorageClientMetrics.KEY_SERVER_ADDRESS)) + .isEqualTo("storage.googleapis.com"); + } + + @Test + public void createContext_nullMetrics_returnsNull() { + assertThat(StorageClientMetrics.createContext(null, "GetObject")).isNull(); + assertThat(StorageClientMetrics.createMultiStepContext(null, "WriteObject")).isNull(); + } + + static final class TestMetricReader implements MetricReader { + private final MetricReader delegate = InMemoryMetricReader.create(); + final AtomicBoolean shutdownCalled = new AtomicBoolean(false); + + @Override + public void register(CollectionRegistration registration) { + delegate.register(registration); + } + + @Override + public AggregationTemporality getAggregationTemporality(InstrumentType instrumentType) { + return delegate.getAggregationTemporality(instrumentType); + } + + @Override + public Aggregation getDefaultAggregation(InstrumentType instrumentType) { + return delegate.getDefaultAggregation(instrumentType); + } + + @Override + public CompletableResultCode forceFlush() { + return delegate.forceFlush(); + } + + @Override + public CompletableResultCode shutdown() { + shutdownCalled.set(true); + return delegate.shutdown(); + } + } + private static int findBucketIndex(List boundaries, double value) { for (int i = 0; i < boundaries.size(); i++) { if (value <= boundaries.get(i)) { diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageOptionsTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageOptionsTest.java index f2dc6f3fda8d..59d63d957435 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageOptionsTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageOptionsTest.java @@ -20,7 +20,9 @@ import static com.google.common.truth.Truth.assertWithMessage; import static org.junit.Assert.assertThrows; +import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.metrics.MeterProvider; +import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.metrics.SdkMeterProvider; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -356,7 +358,7 @@ public void setMetricInterval_acceptsSubSixtySecondsWithCustomMeterProvider() th @Test public void setMetricInterval_acceptsSubSixtySecondsWithCustomOpenTelemetry() throws Exception { - io.opentelemetry.api.OpenTelemetry customOtel = io.opentelemetry.api.OpenTelemetry.noop(); + OpenTelemetry customOtel = OpenTelemetry.noop(); forEachTransport( builder -> { @@ -458,4 +460,75 @@ public void deserializationWithNullMetricInterval_defaultsToSixtySeconds() throw assertThat(deserialized.hashCode()).isEqualTo(defaultOptions.hashCode()); }); } + + @Test + public void meterProvider_usedWhenExplicitlySet() throws Exception { + MeterProvider customMeterProvider = SdkMeterProvider.builder().build(); + forEachTransport( + builder -> { + StorageOptions options = + builder.setEnableOtelMetrics(true).setMeterProvider(customMeterProvider).build(); + StorageClientMetrics metrics = options.getStorageClientMetrics(); + assertThat(metrics).isNotNull(); + }); + } + + @Test + public void customOpenTelemetry_meterProviderUsedWhenConfigured() throws Exception { + SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder().build(); + OpenTelemetry customOtel = + OpenTelemetrySdk.builder().setMeterProvider(sdkMeterProvider).build(); + forEachTransport( + builder -> { + StorageOptions options = + builder.setEnableOtelMetrics(true).setOpenTelemetry(customOtel).build(); + assertThat(options.isCustomOpenTelemetryConfigured()).isTrue(); + StorageClientMetrics metrics = options.getStorageClientMetrics(); + assertThat(metrics).isNotNull(); + }); + } + + @Test + public void closeStorageClientMetrics_cleansUpMetricsAndClosesSdkOwnedProvider() + throws Exception { + Field metricsField = StorageOptions.class.getDeclaredField("storageClientMetrics"); + metricsField.setAccessible(true); + + forEachTransport( + builder -> { + StorageClientMetricsTest.TestMetricReader reader = + new StorageClientMetricsTest.TestMetricReader(); + SdkMeterProvider mockOwnedProvider = + SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageOptions options = + builder.setEnableOtelMetrics(true).setMeterProvider(mockOwnedProvider).build(); + StorageClientMetrics metrics = + StorageClientMetrics.create(mockOwnedProvider, mockOwnedProvider, true, false); + metricsField.set(options, metrics); + + options.closeStorageClientMetrics(); + + assertThat(options.getStorageClientMetrics()).isNull(); + assertThat(reader.shutdownCalled.get()).isTrue(); + }); + } + + @Test + public void closeStorageClientMetrics_doesNotFailOnUserSuppliedMeterProvider() throws Exception { + forEachTransport( + builder -> { + StorageClientMetricsTest.TestMetricReader reader = + new StorageClientMetricsTest.TestMetricReader(); + MeterProvider userProvider = + SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageOptions options = + builder.setEnableOtelMetrics(true).setMeterProvider(userProvider).build(); + assertThat(options.getStorageClientMetrics()).isNotNull(); + + options.closeStorageClientMetrics(); + + assertThat(options.getStorageClientMetrics()).isNull(); + assertThat(reader.shutdownCalled.get()).isFalse(); + }); + } }