diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcChannelTelemetry.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcChannelTelemetry.java new file mode 100644 index 000000000000..891e868c769d --- /dev/null +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcChannelTelemetry.java @@ -0,0 +1,651 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.gax.grpc.ChannelPrimer; +import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider; +import com.google.common.annotations.VisibleForTesting; +import io.grpc.ConnectivityState; +import io.grpc.ManagedChannel; +import io.grpc.internal.GrpcUtil; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.DoubleHistogram; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.metrics.MeterProvider; +import java.lang.ref.WeakReference; +import java.lang.reflect.Method; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; +import java.util.function.LongSupplier; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Package-private telemetry and diagnostics for gRPC channels, supporting CloudPath, DirectPath, + * and DirectPath over Cloud Interconnect. + */ +final class GrpcChannelTelemetry { + + static final String C2P_PREFIX = "google-c2p:///"; + static final String C2P_EXPERIMENTAL_PREFIX = "google-c2p-experimental:///"; + static final String FORCE_XDS_PARAM = "force-xds"; + static final String DIRECT_PATH_INTERCONNECT_HOST = "storage-direct.googleapis.com"; + static final String DIRECT_PATH_INTERCONNECT_PREPROD_HOST = + "storage-preprod-direct.sandbox.googleapis.com"; + static final String XDS_LB_POLICY_NAME = "xds_cluster_manager_experimental"; + static final String DNS_LB_POLICY_NAME = "pick_first"; + static final String STORAGE_METER_NAME = StorageClientMetrics.METER_NAME; + static final String CHANNEL_CREATION_LATENCY_INSTRUMENT = + "internal/client/channel_creation_latency"; + static final String CHANNEL_CREATION_LATENCY_VIEW_NAME = + "storage.googleapis.com/internal/client/channel_creation_latency"; + static final String CHANNEL_CREATION_LATENCY_DESCRIPTION = + "Time from client construction until the first gRPC channel became ready."; + static final String CHANNEL_CREATION_LATENCY_UNIT = "s"; + static final AttributeKey TRANSPORT_TYPE_KEY = AttributeKey.stringKey("transport_type"); + static final Duration DEFAULT_CHANNEL_READY_TIMEOUT = Duration.ofSeconds(30); + + @VisibleForTesting + static final Logger LOG = Logger.getLogger(GrpcChannelTelemetry.class.getName()); + + private static final String DIRECT_PATH_INTERCONNECT_INFIX = "-direct."; + private static final String ENV_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT = + "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT"; + private static final boolean GRPC_FORCE_XDS_SUPPORTED = detectGrpcForceXdsSupport(); + private static final ScheduledExecutorService DEFAULT_SCHEDULER = createDefaultScheduler(); + + /** The transport used to reach the GCS service. */ + enum TransportType { + CLOUD_PATH("CloudPath"), + DIRECT_PATH("DirectPath"), + DIRECT_PATH_INTERCONNECT("DirectPathInterconnect"); + + private final String label; + + TransportType(String label) { + this.label = label; + } + + @Override + public String toString() { + return label; + } + } + + /** + * Callback invoked once the first gRPC channel transitions to {@link ConnectivityState#READY}. + */ + @FunctionalInterface + interface ChannelReadyCallback { + void onChannelReady(TransportType selectedTransport, Duration latency); + } + + @VisibleForTesting + static boolean hasQueryParameter(@Nullable String uri, String key) { + if (uri == null || key == null) { + return false; + } + int fragmentPos = uri.indexOf('#'); + String addressable = fragmentPos >= 0 ? uri.substring(0, fragmentPos) : uri; + int queryPos = addressable.indexOf('?'); + if (queryPos < 0) { + return false; + } + String query = addressable.substring(queryPos + 1); + int start = 0; + while (start <= query.length()) { + int ampPos = query.indexOf('&', start); + String param = ampPos >= 0 ? query.substring(start, ampPos) : query.substring(start); + int eqPos = param.indexOf('='); + String paramName = eqPos >= 0 ? param.substring(0, eqPos) : param; + if (paramName.equals(key)) { + return true; + } + if (ampPos < 0) { + break; + } + start = ampPos + 1; + } + return false; + } + + @VisibleForTesting + static TransportType detectTransportType(@Nullable String endpoint) { + if (endpoint == null + || (!endpoint.startsWith(C2P_PREFIX) && !endpoint.startsWith(C2P_EXPERIMENTAL_PREFIX))) { + return TransportType.CLOUD_PATH; + } + if (hasQueryParameter(endpoint, FORCE_XDS_PARAM)) { + return TransportType.DIRECT_PATH_INTERCONNECT; + } + return TransportType.DIRECT_PATH; + } + + @VisibleForTesting + static String c2pTargetHost(@Nullable String endpoint) { + if (endpoint == null) { + return ""; + } + String rest; + if (endpoint.startsWith(C2P_PREFIX)) { + rest = endpoint.substring(C2P_PREFIX.length()); + } else if (endpoint.startsWith(C2P_EXPERIMENTAL_PREFIX)) { + rest = endpoint.substring(C2P_EXPERIMENTAL_PREFIX.length()); + } else { + return ""; + } + int end = rest.length(); + for (int i = 0; i < rest.length(); i++) { + char c = rest.charAt(i); + if (c == '/' || c == '?' || c == '#' || c == ':') { + end = i; + break; + } + } + return rest.substring(0, end); + } + + @VisibleForTesting + static String authorityHost(@Nullable String authority) { + if (authority == null || authority.isEmpty()) { + return ""; + } + int colon = authority.indexOf(':'); + return colon >= 0 ? authority.substring(0, colon) : authority; + } + + @VisibleForTesting + static boolean isDirectPathInterconnectHost(@Nullable String host) { + String h = authorityHost(host); + return DIRECT_PATH_INTERCONNECT_HOST.equals(h) + || DIRECT_PATH_INTERCONNECT_PREPROD_HOST.equals(h); + } + + @VisibleForTesting + static TransportType selectedTransportType( + TransportType requestedTransport, + @Nullable String lbPolicyName, + boolean grpcSupportsForceXds) { + if (requestedTransport == TransportType.CLOUD_PATH || DNS_LB_POLICY_NAME.equals(lbPolicyName)) { + return TransportType.CLOUD_PATH; + } + if (requestedTransport == TransportType.DIRECT_PATH_INTERCONNECT && !grpcSupportsForceXds) { + return TransportType.DIRECT_PATH; + } + return requestedTransport; + } + + @VisibleForTesting + static boolean grpcForceXdsIsSupported() { + return GRPC_FORCE_XDS_SUPPORTED; + } + + static boolean isDirectPathInterconnectEnabled(boolean attemptDirectPathXdsOverInterconnect) { + return isDirectPathInterconnectEnabled( + attemptDirectPathXdsOverInterconnect, + System.getenv(ENV_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT)); + } + + @VisibleForTesting + static boolean isDirectPathInterconnectEnabled( + boolean attemptDirectPathXdsOverInterconnect, @Nullable String envValue) { + if (envValue != null) { + String trimmed = envValue.trim(); + if ("true".equalsIgnoreCase(trimmed)) { + return true; + } + if ("false".equalsIgnoreCase(trimmed)) { + return false; + } + } + return attemptDirectPathXdsOverInterconnect; + } + + @VisibleForTesting + static String resolveEffectiveTarget( + InstantiatingGrpcChannelProvider provider, boolean attemptDirectPathXdsOverInterconnect) { + String endpoint = provider.getEndpoint(); + if (endpoint == null) { + return ""; + } + if (endpoint.startsWith(C2P_PREFIX) || endpoint.startsWith(C2P_EXPERIMENTAL_PREFIX)) { + return endpoint; + } + boolean interconnectEnabled = + isDirectPathInterconnectEnabled(attemptDirectPathXdsOverInterconnect); + if (!provider.needsCredentials() + && provider.canUseDirectPath() + && provider.isDirectPathXdsEnabled()) { + int colon = endpoint.lastIndexOf(':'); + String serviceAddress = colon >= 0 ? endpoint.substring(0, colon) : endpoint; + if (interconnectEnabled) { + return C2P_PREFIX + + rewriteDirectPathInterconnectHost(serviceAddress) + + "?" + + FORCE_XDS_PARAM; + } + return C2P_PREFIX + serviceAddress; + } + return endpoint; + } + + @VisibleForTesting + static void logChannelConfiguration( + String effectiveEndpoint, + @Nullable String authority, + boolean interconnectEnabled, + boolean grpcSupportsForceXds) { + TransportType requestedTransport = detectTransportType(effectiveEndpoint); + String effectiveAuthority = + (authority == null || authority.isEmpty()) ? c2pTargetHost(effectiveEndpoint) : authority; + if (effectiveAuthority.isEmpty()) { + effectiveAuthority = authorityHost(effectiveEndpoint); + } + String grpcVersion = GrpcUtil.getGrpcBuildVersion().toString(); + + if (!grpcSupportsForceXds + && (interconnectEnabled || requestedTransport == TransportType.DIRECT_PATH_INTERCONNECT)) { + LOG.log( + Level.WARNING, + "DirectPath over Interconnect requires a gRPC version supporting force-xds (found " + + grpcVersion + + "). The force-xds query parameter will be ignored by the google-c2p resolver;" + + " outside GCE this channel will fall back to CloudPath, and on GCE it will use" + + " standard DirectPath. endpoint=" + + effectiveEndpoint); + return; + } + if (interconnectEnabled && requestedTransport != TransportType.DIRECT_PATH_INTERCONNECT) { + LOG.log( + Level.WARNING, + "DirectPath over Interconnect is enabled, but the effective endpoint does not request" + + " it. Check whether a custom endpoint, non-default universe domain, disabled" + + " DirectPath, or unauthenticated credentials overrode the DirectPath over" + + " Interconnect target. requested_transport_type=" + + requestedTransport + + ", endpoint=" + + effectiveEndpoint); + return; + } + if (requestedTransport == TransportType.DIRECT_PATH_INTERCONNECT + && !isDirectPathInterconnectHost(effectiveAuthority)) { + LOG.log( + Level.WARNING, + "DirectPath over Interconnect is active, but the channel authority is not a DirectPath" + + " over Interconnect host; the service will route RPCs as standard DirectPath. Use '" + + DIRECT_PATH_INTERCONNECT_HOST + + "' (or '" + + DIRECT_PATH_INTERCONNECT_PREPROD_HOST + + "' in preprod). requested_transport_type=" + + requestedTransport + + ", endpoint=" + + effectiveEndpoint + + ", authority=" + + effectiveAuthority + + ", grpc_version=" + + grpcVersion); + return; + } + LOG.log( + Level.INFO, + "Configured gRPC transport. requested_transport_type=" + + requestedTransport + + ", endpoint=" + + effectiveEndpoint + + ", authority=" + + effectiveAuthority + + ", grpc_version=" + + grpcVersion); + } + + @VisibleForTesting + static String channelReadyNote( + TransportType requestedTransport, + TransportType selectedTransport, + @Nullable String lbPolicyName, + boolean ready) { + if (!ready) { + if (requestedTransport == TransportType.DIRECT_PATH_INTERCONNECT) { + return " Verify that TD bootstrap and xDS reachability are configured, or check gRPC logs" + + " with GRPC_VERBOSITY=debug GRPC_TRACE=xds_client,xds_resolver,c2p_resolver."; + } + if (requestedTransport == TransportType.DIRECT_PATH) { + return " Check gRPC logs with GRPC_VERBOSITY=debug" + + " GRPC_TRACE=xds_client,xds_resolver,c2p_resolver."; + } + return ""; + } + if (requestedTransport != TransportType.CLOUD_PATH + && selectedTransport == TransportType.CLOUD_PATH) { + return " Note: google-c2p resolver fell back to DNS (" + + DNS_LB_POLICY_NAME + + "); traffic is using CloudPath instead of " + + requestedTransport + + "."; + } + if (requestedTransport != TransportType.CLOUD_PATH + && lbPolicyName != null + && !lbPolicyName.isEmpty() + && !XDS_LB_POLICY_NAME.equals(lbPolicyName) + && !DNS_LB_POLICY_NAME.equals(lbPolicyName)) { + return " Note: unexpected LB policy '" + + lbPolicyName + + "' for " + + requestedTransport + + " (expected '" + + XDS_LB_POLICY_NAME + + "')."; + } + return ""; + } + + @VisibleForTesting + static void logChannelReady( + TransportType requestedTransport, + TransportType selectedTransport, + @Nullable String lbPolicyName, + Duration elapsed, + boolean ready, + @Nullable String statusDescription) { + String lb = (lbPolicyName == null || lbPolicyName.isEmpty()) ? "unknown" : lbPolicyName; + String note = channelReadyNote(requestedTransport, selectedTransport, lbPolicyName, ready); + long elapsedMs = elapsed.toMillis(); + if (ready) { + LOG.log( + Level.INFO, + "gRPC channel [0] is ready. transport_type=" + + selectedTransport + + ", requested_transport_type=" + + requestedTransport + + ", lb_policy=" + + lb + + ", elapsed_ms=" + + elapsedMs + + note); + } else { + LOG.log( + Level.INFO, + "gRPC channel [0] did not become ready before deadline. transport_type=" + + selectedTransport + + ", requested_transport_type=" + + requestedTransport + + ", lb_policy=" + + lb + + ", elapsed_ms=" + + elapsedMs + + (statusDescription != null && !statusDescription.isEmpty() + ? ", status=" + statusDescription + : "") + + note); + } + } + + @VisibleForTesting + static ChannelReadyCallback makeChannelMetricsCallback(@Nullable MeterProvider meterProvider) { + if (meterProvider == null) { + return (t, d) -> {}; + } + Meter meter = + meterProvider + .meterBuilder(STORAGE_METER_NAME) + .setInstrumentationVersion(StorageOptions.version()) + .build(); + DoubleHistogram histogram = + meter + .histogramBuilder(CHANNEL_CREATION_LATENCY_INSTRUMENT) + .setDescription(CHANNEL_CREATION_LATENCY_DESCRIPTION) + .setUnit(CHANNEL_CREATION_LATENCY_UNIT) + .build(); + return (selectedTransport, latency) -> { + double seconds = latency.toNanos() / 1_000_000_000.0; + histogram.record(seconds, Attributes.of(TRANSPORT_TYPE_KEY, selectedTransport.toString())); + }; + } + + static CompletableFuture startChannelTelemetry( + ManagedChannel channel, + String effectiveEndpoint, + boolean grpcSupportsForceXds, + Duration timeout, + @Nullable ChannelReadyCallback onReady) { + return startChannelTelemetry( + new WeakReference<>(channel), + effectiveEndpoint, + grpcSupportsForceXds, + timeout, + onReady, + DEFAULT_SCHEDULER, + System::nanoTime, + GrpcChannelTelemetry::defaultExtractLbPolicyName); + } + + @VisibleForTesting + static CompletableFuture startChannelTelemetry( + WeakReference channelRef, + String effectiveEndpoint, + boolean grpcSupportsForceXds, + Duration timeout, + @Nullable ChannelReadyCallback onReady, + ScheduledExecutorService scheduler, + LongSupplier nanoTimeSupplier, + Function lbPolicyExtractor) { + TransportType requestedTransport = detectTransportType(effectiveEndpoint); + ChannelReadyWatcher watcher = + new ChannelReadyWatcher( + channelRef, + requestedTransport, + grpcSupportsForceXds, + timeout, + onReady, + scheduler, + nanoTimeSupplier, + lbPolicyExtractor); + return watcher.start(); + } + + static ChannelPrimer createChannelPrimer( + String effectiveEndpoint, + boolean grpcSupportsForceXds, + @Nullable ChannelReadyCallback onReady) { + AtomicBoolean firstChannelObserved = new AtomicBoolean(false); + return managedChannel -> { + if (firstChannelObserved.compareAndSet(false, true)) { + startChannelTelemetry( + managedChannel, + effectiveEndpoint, + grpcSupportsForceXds, + DEFAULT_CHANNEL_READY_TIMEOUT, + onReady); + } + }; + } + + private GrpcChannelTelemetry() {} + + private static boolean detectGrpcForceXdsSupport() { + try { + Class clazz = Class.forName("io.grpc.googleapis.GoogleCloudToProdNameResolver"); + clazz.getDeclaredField("forceXds"); + return true; + } catch (ReflectiveOperationException | SecurityException e) { + return false; + } + } + + private static ScheduledExecutorService createDefaultScheduler() { + ScheduledThreadPoolExecutor executor = + new ScheduledThreadPoolExecutor( + 1, + runnable -> { + Thread t = new Thread(runnable, "gcs-grpc-channel-telemetry"); + t.setDaemon(true); + return t; + }); + executor.setRemoveOnCancelPolicy(true); + return executor; + } + + private static String rewriteDirectPathInterconnectHost(String serviceAddress) { + if (!serviceAddress.contains(DIRECT_PATH_INTERCONNECT_INFIX)) { + int dotIndex = serviceAddress.indexOf('.'); + if (dotIndex > 0) { + return serviceAddress.substring(0, dotIndex) + + DIRECT_PATH_INTERCONNECT_INFIX + + serviceAddress.substring(dotIndex + 1); + } + } + return serviceAddress; + } + + @Nullable + private static String defaultExtractLbPolicyName(@Nullable ManagedChannel channel) { + if (channel == null) { + return null; + } + try { + Method m = channel.getClass().getMethod("getLoadBalancingPolicyName"); + Object result = m.invoke(channel); + if (result instanceof String) { + return (String) result; + } + } catch (ReflectiveOperationException | SecurityException ignored) { + // Standard io.grpc.ManagedChannel does not expose getLoadBalancingPolicyName publicly. + } + return null; + } + + private static final class ChannelReadyWatcher { + private final WeakReference channelRef; + private final TransportType requestedTransport; + private final boolean grpcSupportsForceXds; + private final Duration timeout; + @Nullable private final ChannelReadyCallback onReady; + private final ScheduledExecutorService scheduler; + private final LongSupplier nanoTimeSupplier; + private final Function lbPolicyExtractor; + private final AtomicBoolean completed = new AtomicBoolean(false); + private final CompletableFuture completionFuture = new CompletableFuture<>(); + private volatile long startNanos; + @Nullable private volatile ScheduledFuture timeoutFuture; + + ChannelReadyWatcher( + WeakReference channelRef, + TransportType requestedTransport, + boolean grpcSupportsForceXds, + Duration timeout, + @Nullable ChannelReadyCallback onReady, + ScheduledExecutorService scheduler, + LongSupplier nanoTimeSupplier, + Function lbPolicyExtractor) { + this.channelRef = channelRef; + this.requestedTransport = requestedTransport; + this.grpcSupportsForceXds = grpcSupportsForceXds; + this.timeout = timeout; + this.onReady = onReady; + this.scheduler = scheduler; + this.nanoTimeSupplier = nanoTimeSupplier; + this.lbPolicyExtractor = lbPolicyExtractor; + } + + CompletableFuture start() { + this.startNanos = nanoTimeSupplier.getAsLong(); + this.timeoutFuture = + scheduler.schedule( + this::onTimeout, Math.max(0L, timeout.toMillis()), TimeUnit.MILLISECONDS); + step(); + return completionFuture; + } + + private void step() { + if (completed.get()) { + return; + } + ManagedChannel ch = channelRef.get(); + if (ch == null) { + completeFailure(null, "channel destroyed before becoming ready"); + return; + } + ConnectivityState state; + try { + state = ch.getState(true); + } catch (UnsupportedOperationException e) { + completeFailure(ch, "channel does not support connectivity state inspection"); + return; + } + if (state == ConnectivityState.READY) { + if (completed.compareAndSet(false, true)) { + cancelTimeout(); + Duration elapsed = elapsedDuration(); + String lbPolicyName = lbPolicyExtractor.apply(ch); + TransportType selected = + selectedTransportType(requestedTransport, lbPolicyName, grpcSupportsForceXds); + logChannelReady(requestedTransport, selected, lbPolicyName, elapsed, true, null); + if (onReady != null) { + onReady.onChannelReady(selected, elapsed); + } + completionFuture.complete(null); + } + return; + } + if (state == ConnectivityState.SHUTDOWN) { + completeFailure(ch, "channel shut down before becoming ready"); + return; + } + try { + ch.notifyWhenStateChanged(state, this::step); + } catch (UnsupportedOperationException e) { + completeFailure(ch, "channel does not support state change notifications"); + } + } + + private void onTimeout() { + completeFailure(channelRef.get(), "DEADLINE_EXCEEDED"); + } + + private void completeFailure(@Nullable ManagedChannel ch, String statusDescription) { + if (completed.compareAndSet(false, true)) { + cancelTimeout(); + Duration elapsed = elapsedDuration(); + String lbPolicyName = ch != null ? lbPolicyExtractor.apply(ch) : null; + TransportType selected = + selectedTransportType(requestedTransport, lbPolicyName, grpcSupportsForceXds); + logChannelReady( + requestedTransport, selected, lbPolicyName, elapsed, false, statusDescription); + completionFuture.complete(null); + } + } + + private Duration elapsedDuration() { + return Duration.ofNanos(Math.max(0L, nanoTimeSupplier.getAsLong() - startNanos)); + } + + private void cancelTimeout() { + ScheduledFuture f = this.timeoutFuture; + if (f != null) { + f.cancel(false); + } + } + } +} 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 eaeaa04db14a..bce29df5c866 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 @@ -326,7 +326,7 @@ private Tuple> resolveSettingsAndOpts() throw InstantiatingGrpcChannelProvider.newBuilder() .setEndpoint(endpoint) .setAllowNonDefaultServiceAccount(true) - .setAttemptDirectPath(attemptDirectPath || attemptDirectPathXdsOverInterconnect) + .setAttemptDirectPath(attemptDirectPath) .setAttemptDirectPathXdsOverInterconnect(attemptDirectPathXdsOverInterconnect); if (!DIRECT_PATH_BOUND_TOKEN_DISABLED) { @@ -346,16 +346,36 @@ private Tuple> resolveSettingsAndOpts() throw channelProviderBuilder.setChannelConfigurator(ManagedChannelBuilder::usePlaintext); } + @Nullable MeterProvider grpcMeterProvider = null; if (enableGrpcClientMetrics) { - ChannelConfigurator channelConfigurator = - OpenTelemetryBootstrappingUtils.enableGrpcMetrics( + OpenTelemetryBootstrappingUtils.GrpcMetricsBootstrapResult metricsBootstrap = + OpenTelemetryBootstrappingUtils.bootstrapGrpcMetrics( ChannelConfigurator.lift(channelProviderBuilder.getChannelConfigurator()), endpoint, this.getProjectId(), this.getUniverseDomain(), !grpcClientMetricsManuallyEnabled); - channelProviderBuilder.setChannelConfigurator(channelConfigurator); - } + channelProviderBuilder.setChannelConfigurator(metricsBootstrap.getChannelConfigurator()); + grpcMeterProvider = metricsBootstrap.getMeterProvider(); + } + + InstantiatingGrpcChannelProvider baseProvider = channelProviderBuilder.build(); + InstantiatingGrpcChannelProvider telemetryProvider = + credentials instanceof NoCredentials + ? baseProvider + : (InstantiatingGrpcChannelProvider) baseProvider.withCredentials(credentials); + boolean interconnectEnabled = + GrpcChannelTelemetry.isDirectPathInterconnectEnabled(attemptDirectPathXdsOverInterconnect); + boolean grpcSupportsForceXds = GrpcChannelTelemetry.grpcForceXdsIsSupported(); + String effectiveTarget = + GrpcChannelTelemetry.resolveEffectiveTarget(telemetryProvider, interconnectEnabled); + GrpcChannelTelemetry.logChannelConfiguration( + effectiveTarget, null, interconnectEnabled, grpcSupportsForceXds); + GrpcChannelTelemetry.ChannelReadyCallback channelReadyCallback = + GrpcChannelTelemetry.makeChannelMetricsCallback(grpcMeterProvider); + channelProviderBuilder.setChannelPrimer( + GrpcChannelTelemetry.createChannelPrimer( + effectiveTarget, grpcSupportsForceXds, channelReadyCallback)); builder.setTransportChannelProvider(channelProviderBuilder.build()); RetrySettings baseRetrySettings = getRetrySettings(); 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 0188e4667d4a..b5dbaea247e2 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 @@ -32,6 +32,7 @@ import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.common.AttributesBuilder; import io.opentelemetry.api.internal.StringUtils; +import io.opentelemetry.api.metrics.MeterProvider; import io.opentelemetry.contrib.gcp.resource.GCPResourceProvider; import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.common.CompletableResultCode; @@ -100,6 +101,17 @@ static ChannelConfigurator enableGrpcMetrics( @Nullable String projectId, String universeDomain, boolean shouldSuppressExceptions) { + return bootstrapGrpcMetrics( + channelConfigurator, endpoint, projectId, universeDomain, shouldSuppressExceptions) + .getChannelConfigurator(); + } + + static GrpcMetricsBootstrapResult bootstrapGrpcMetrics( + @Nullable ChannelConfigurator channelConfigurator, + String endpoint, + @Nullable String projectId, + String universeDomain, + boolean shouldSuppressExceptions) { GCPResourceProvider resourceProvider = new GCPResourceProvider(); Attributes detectedAttributes = resourceProvider.getAttributes(); @@ -109,7 +121,9 @@ static ChannelConfigurator enableGrpcMetrics( log.warning( "Unable to determine the Project ID in order to report metrics. No gRPC client metrics" + " will be reported."); - return channelConfigurator != null ? channelConfigurator : ChannelConfigurator.identity(); + ChannelConfigurator fallback = + channelConfigurator != null ? channelConfigurator : ChannelConfigurator.identity(); + return new GrpcMetricsBootstrapResult(fallback, null); } String projectIdToUse = detectedProjectId == null ? projectId : detectedProjectId; @@ -143,7 +157,26 @@ static ChannelConfigurator enableGrpcMetrics( grpcOpenTelemetry.configureChannelBuilder(b); return b; }; - return otelConfigurator.andThen(channelConfigurator); + return new GrpcMetricsBootstrapResult(otelConfigurator.andThen(channelConfigurator), provider); + } + + static final class GrpcMetricsBootstrapResult { + private final ChannelConfigurator channelConfigurator; + @Nullable private final MeterProvider meterProvider; + + GrpcMetricsBootstrapResult( + ChannelConfigurator channelConfigurator, @Nullable MeterProvider meterProvider) { + this.channelConfigurator = channelConfigurator; + this.meterProvider = meterProvider; + } + + ChannelConfigurator getChannelConfigurator() { + return channelConfigurator; + } + + @Nullable MeterProvider getMeterProvider() { + return meterProvider; + } } @SuppressWarnings("rawtypes") // ManagedChannelBuilder @@ -183,13 +216,30 @@ public ManagedChannelBuilder apply(ManagedChannelBuilder input) { } @VisibleForTesting - static String getCloudMonitoringEndpoint(String endpoint, String universeDomain) { + static String getCloudMonitoringEndpoint(String endpoint, @Nullable String universeDomain) { String metricServiceEndpoint = "monitoring.googleapis.com"; + String host = GrpcChannelTelemetry.c2pTargetHost(endpoint); + int port = 443; + if (host.isEmpty()) { + int colon = endpoint.lastIndexOf(':'); + if (colon >= 0) { + host = endpoint.substring(0, colon); + try { + port = Integer.parseInt(endpoint.substring(colon + 1)); + } catch (NumberFormatException ignored) { + port = 443; + } + } else { + host = endpoint; + } + } - // use contains instead of equals because endpoint has a port in it - if (universeDomain != null && endpoint.contains("storage." + universeDomain)) { + if (universeDomain != null + && (host.equals("storage." + universeDomain) + || host.equals("storage-direct." + universeDomain))) { metricServiceEndpoint = "monitoring." + universeDomain; - } else if (!endpoint.contains("storage.googleapis.com")) { + } else if (!host.equals("storage.googleapis.com") + && !GrpcChannelTelemetry.isDirectPathInterconnectHost(host)) { String canonicalEndpoint = "storage.googleapis.com"; String privateEndpoint = "private.googleapis.com"; String restrictedEndpoint = "restricted.googleapis.com"; @@ -198,15 +248,16 @@ static String getCloudMonitoringEndpoint(String endpoint, String universeDomain) privateEndpoint = "private." + universeDomain; restrictedEndpoint = "restricted." + universeDomain; } + final String resolvedHost = host; String match = ImmutableList.of(canonicalEndpoint, privateEndpoint, restrictedEndpoint).stream() - .filter(s -> endpoint.contains(s) || endpoint.contains("google-c2p:///" + s)) + .filter(s -> resolvedHost.contains(s) || endpoint.contains("google-c2p:///" + s)) .collect(Collectors.joining()); if (!StringUtils.isNullOrEmpty(match)) { metricServiceEndpoint = match; } } - return metricServiceEndpoint + ":" + endpoint.split(":")[1]; + return metricServiceEndpoint + ":" + port; } static final ImmutableList CLIENT_LATENCY_HISTOGRAMS = @@ -294,6 +345,7 @@ static SdkMeterProvider createMeterProvider( "By", "A view of grpc/client/attempt/sent_total_compressed_message_size with histogram boundaries" + " more appropriate for Google Cloud Storage RPCs"); + registerGrpcChannelViews(providerBuilder); return buildMeterProvider( cloudMonitoringExporter, @@ -448,6 +500,19 @@ private static SdkMeterProvider buildMeterProvider( return providerBuilder.build(); } + @VisibleForTesting + static SdkMeterProviderBuilder registerGrpcChannelViews(SdkMeterProviderBuilder providerBuilder) { + addHistogramView( + providerBuilder, + latencyHistogramBoundaries(), + GrpcChannelTelemetry.CHANNEL_CREATION_LATENCY_INSTRUMENT, + GrpcChannelTelemetry.CHANNEL_CREATION_LATENCY_VIEW_NAME, + GrpcChannelTelemetry.STORAGE_METER_NAME, + GrpcChannelTelemetry.CHANNEL_CREATION_LATENCY_UNIT, + GrpcChannelTelemetry.CHANNEL_CREATION_LATENCY_DESCRIPTION); + return providerBuilder; + } + @VisibleForTesting static SdkMeterProviderBuilder registerClientViews(SdkMeterProviderBuilder providerBuilder) { for (String metric : CLIENT_COUNTERS) { diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcChannelTelemetryTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcChannelTelemetryTest.java new file mode 100644 index 000000000000..003f157fab13 --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcChannelTelemetryTest.java @@ -0,0 +1,747 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.api.gax.grpc.ChannelPrimer; +import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider; +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.cloud.NoCredentials; +import com.google.cloud.storage.GrpcChannelTelemetry.ChannelReadyCallback; +import com.google.cloud.storage.GrpcChannelTelemetry.TransportType; +import com.google.storage.v2.StorageSettings; +import io.grpc.CallOptions; +import io.grpc.ClientCall; +import io.grpc.ConnectivityState; +import io.grpc.ManagedChannel; +import io.grpc.MethodDescriptor; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder; +import io.opentelemetry.sdk.metrics.data.HistogramPointData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.lang.ref.WeakReference; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.logging.Handler; +import java.util.logging.Level; +import java.util.logging.LogRecord; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public final class GrpcChannelTelemetryTest { + + private CapturingLogHandler logHandler; + private Level previousLevel; + private boolean previousUseParentHandlers; + private ScheduledExecutorService scheduler; + + @Before + public void setUp() { + logHandler = new CapturingLogHandler(); + previousLevel = GrpcChannelTelemetry.LOG.getLevel(); + previousUseParentHandlers = GrpcChannelTelemetry.LOG.getUseParentHandlers(); + GrpcChannelTelemetry.LOG.setLevel(Level.ALL); + GrpcChannelTelemetry.LOG.setUseParentHandlers(false); + GrpcChannelTelemetry.LOG.addHandler(logHandler); + scheduler = Executors.newSingleThreadScheduledExecutor(); + } + + @After + public void tearDown() { + GrpcChannelTelemetry.LOG.removeHandler(logHandler); + GrpcChannelTelemetry.LOG.setLevel(previousLevel); + GrpcChannelTelemetry.LOG.setUseParentHandlers(previousUseParentHandlers); + scheduler.shutdownNow(); + } + + @Test + public void detectTransportType_allCases() { + assertThat(GrpcChannelTelemetry.detectTransportType(null)).isEqualTo(TransportType.CLOUD_PATH); + assertThat(GrpcChannelTelemetry.detectTransportType("")).isEqualTo(TransportType.CLOUD_PATH); + assertThat(GrpcChannelTelemetry.detectTransportType("storage.googleapis.com:443")) + .isEqualTo(TransportType.CLOUD_PATH); + assertThat(GrpcChannelTelemetry.detectTransportType("https://storage.googleapis.com")) + .isEqualTo(TransportType.CLOUD_PATH); + assertThat(GrpcChannelTelemetry.detectTransportType("google-c2p:///storage.googleapis.com")) + .isEqualTo(TransportType.DIRECT_PATH); + assertThat( + GrpcChannelTelemetry.detectTransportType( + "google-c2p-experimental:///storage.googleapis.com")) + .isEqualTo(TransportType.DIRECT_PATH); + assertThat( + GrpcChannelTelemetry.detectTransportType( + "google-c2p:///storage-direct.googleapis.com?force-xds")) + .isEqualTo(TransportType.DIRECT_PATH_INTERCONNECT); + assertThat( + GrpcChannelTelemetry.detectTransportType( + "google-c2p:///storage-direct.googleapis.com?foo=1&force-xds=true")) + .isEqualTo(TransportType.DIRECT_PATH_INTERCONNECT); + assertThat( + GrpcChannelTelemetry.detectTransportType( + "google-c2p-experimental:///storage-direct.googleapis.com?force-xds")) + .isEqualTo(TransportType.DIRECT_PATH_INTERCONNECT); + assertThat( + GrpcChannelTelemetry.detectTransportType( + "google-c2p:///storage.googleapis.com?other=force-xds")) + .isEqualTo(TransportType.DIRECT_PATH); + assertThat( + GrpcChannelTelemetry.detectTransportType( + "google-c2p:///storage.googleapis.com#force-xds")) + .isEqualTo(TransportType.DIRECT_PATH); + assertThat( + GrpcChannelTelemetry.detectTransportType( + "google-c2p:///storage-direct.googleapis.com?force-xds#section")) + .isEqualTo(TransportType.DIRECT_PATH_INTERCONNECT); + assertThat( + GrpcChannelTelemetry.detectTransportType( + "google-c2p:///storage.googleapis.com#frag?force-xds=1")) + .isEqualTo(TransportType.DIRECT_PATH); + } + + @Test + public void c2pTargetHost_authorityHost_isDirectPathInterconnectHost() { + assertThat( + GrpcChannelTelemetry.c2pTargetHost( + "google-c2p:///storage-direct.googleapis.com?force-xds")) + .isEqualTo("storage-direct.googleapis.com"); + assertThat( + GrpcChannelTelemetry.c2pTargetHost( + "google-c2p-experimental:///storage-preprod-direct.sandbox.googleapis.com:443?force-xds")) + .isEqualTo("storage-preprod-direct.sandbox.googleapis.com"); + assertThat(GrpcChannelTelemetry.c2pTargetHost("google-c2p:///storage.googleapis.com/path")) + .isEqualTo("storage.googleapis.com"); + assertThat(GrpcChannelTelemetry.c2pTargetHost("google-c2p:///storage.googleapis.com#frag")) + .isEqualTo("storage.googleapis.com"); + assertThat(GrpcChannelTelemetry.c2pTargetHost("storage.googleapis.com:443")).isEmpty(); + assertThat(GrpcChannelTelemetry.c2pTargetHost(null)).isEmpty(); + + assertThat(GrpcChannelTelemetry.authorityHost("storage-direct.googleapis.com:443")) + .isEqualTo("storage-direct.googleapis.com"); + assertThat(GrpcChannelTelemetry.authorityHost("storage-direct.googleapis.com")) + .isEqualTo("storage-direct.googleapis.com"); + assertThat(GrpcChannelTelemetry.authorityHost(null)).isEmpty(); + assertThat(GrpcChannelTelemetry.authorityHost("")).isEmpty(); + + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectHost("storage-direct.googleapis.com")) + .isTrue(); + assertThat( + GrpcChannelTelemetry.isDirectPathInterconnectHost("storage-direct.googleapis.com:443")) + .isTrue(); + assertThat( + GrpcChannelTelemetry.isDirectPathInterconnectHost( + "storage-preprod-direct.sandbox.googleapis.com")) + .isTrue(); + assertThat( + GrpcChannelTelemetry.isDirectPathInterconnectHost( + "storage-preprod-direct.sandbox.googleapis.com:443")) + .isTrue(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectHost("storage.googleapis.com")) + .isFalse(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectHost("custom.example.com")).isFalse(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectHost(null)).isFalse(); + } + + @Test + public void selectedTransportType_allCases() { + assertThat(GrpcChannelTelemetry.selectedTransportType(TransportType.CLOUD_PATH, null, true)) + .isEqualTo(TransportType.CLOUD_PATH); + assertThat( + GrpcChannelTelemetry.selectedTransportType( + TransportType.CLOUD_PATH, GrpcChannelTelemetry.XDS_LB_POLICY_NAME, true)) + .isEqualTo(TransportType.CLOUD_PATH); + assertThat( + GrpcChannelTelemetry.selectedTransportType( + TransportType.DIRECT_PATH, GrpcChannelTelemetry.XDS_LB_POLICY_NAME, true)) + .isEqualTo(TransportType.DIRECT_PATH); + assertThat( + GrpcChannelTelemetry.selectedTransportType( + TransportType.DIRECT_PATH, GrpcChannelTelemetry.DNS_LB_POLICY_NAME, true)) + .isEqualTo(TransportType.CLOUD_PATH); + assertThat( + GrpcChannelTelemetry.selectedTransportType( + TransportType.DIRECT_PATH_INTERCONNECT, + GrpcChannelTelemetry.XDS_LB_POLICY_NAME, + true)) + .isEqualTo(TransportType.DIRECT_PATH_INTERCONNECT); + assertThat( + GrpcChannelTelemetry.selectedTransportType( + TransportType.DIRECT_PATH_INTERCONNECT, + GrpcChannelTelemetry.DNS_LB_POLICY_NAME, + true)) + .isEqualTo(TransportType.CLOUD_PATH); + assertThat( + GrpcChannelTelemetry.selectedTransportType( + TransportType.DIRECT_PATH_INTERCONNECT, + GrpcChannelTelemetry.XDS_LB_POLICY_NAME, + false)) + .isEqualTo(TransportType.DIRECT_PATH); + } + + @Test + public void logChannelConfiguration_validConfigurations_logInfo() { + GrpcChannelTelemetry.logChannelConfiguration( + "storage.googleapis.com:443", "storage.googleapis.com", false, true); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.INFO); + assertThat(logHandler.records.get(0).getMessage()) + .contains("Configured gRPC transport. requested_transport_type=CloudPath"); + + logHandler.clear(); + GrpcChannelTelemetry.logChannelConfiguration( + "google-c2p:///storage.googleapis.com", "storage.googleapis.com", false, true); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.INFO); + assertThat(logHandler.records.get(0).getMessage()) + .contains("Configured gRPC transport. requested_transport_type=DirectPath"); + + logHandler.clear(); + GrpcChannelTelemetry.logChannelConfiguration( + "google-c2p:///storage-direct.googleapis.com?force-xds", null, true, true); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.INFO); + assertThat(logHandler.records.get(0).getMessage()) + .contains("Configured gRPC transport. requested_transport_type=DirectPathInterconnect"); + assertThat(logHandler.records.get(0).getMessage()) + .contains("authority=storage-direct.googleapis.com"); + + logHandler.clear(); + GrpcChannelTelemetry.logChannelConfiguration( + "google-c2p:///storage-preprod-direct.sandbox.googleapis.com?force-xds", null, true, true); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.INFO); + assertThat(logHandler.records.get(0).getMessage()) + .contains("authority=storage-preprod-direct.sandbox.googleapis.com"); + } + + @Test + public void logChannelConfiguration_warnings() { + // Priority 1: gRPC version does not support force-xds + GrpcChannelTelemetry.logChannelConfiguration( + "google-c2p:///storage-direct.googleapis.com?force-xds", null, true, false); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.WARNING); + assertThat(logHandler.records.get(0).getMessage()) + .contains("DirectPath over Interconnect requires a gRPC version supporting force-xds"); + + // Priority 2: interconnectEnabled = true, but effectiveEndpoint is CloudPath or DirectPath + logHandler.clear(); + GrpcChannelTelemetry.logChannelConfiguration( + "storage.googleapis.com:443", "storage.googleapis.com", true, true); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.WARNING); + assertThat(logHandler.records.get(0).getMessage()) + .contains( + "DirectPath over Interconnect is enabled, but the effective endpoint does not request" + + " it."); + assertThat(logHandler.records.get(0).getMessage()) + .contains("requested_transport_type=CloudPath"); + + logHandler.clear(); + GrpcChannelTelemetry.logChannelConfiguration( + "google-c2p:///storage.googleapis.com", "storage.googleapis.com", true, true); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.WARNING); + assertThat(logHandler.records.get(0).getMessage()) + .contains("requested_transport_type=DirectPath"); + + // Priority 3: force-xds active, but authority is not a -direct host + logHandler.clear(); + GrpcChannelTelemetry.logChannelConfiguration( + "google-c2p:///storage.googleapis.com?force-xds", "storage.googleapis.com", true, true); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.WARNING); + assertThat(logHandler.records.get(0).getMessage()) + .contains( + "DirectPath over Interconnect is active, but the channel authority is not a DirectPath" + + " over Interconnect host"); + } + + @Test + public void logChannelReady_and_channelReadyNote() { + // Ready on DirectPathInterconnect + GrpcChannelTelemetry.logChannelReady( + TransportType.DIRECT_PATH_INTERCONNECT, + TransportType.DIRECT_PATH_INTERCONNECT, + GrpcChannelTelemetry.XDS_LB_POLICY_NAME, + Duration.ofMillis(42), + true, + null); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getLevel()).isEqualTo(Level.INFO); + assertThat(logHandler.records.get(0).getMessage()) + .isEqualTo( + "gRPC channel [0] is ready. transport_type=DirectPathInterconnect," + + " requested_transport_type=DirectPathInterconnect," + + " lb_policy=xds_cluster_manager_experimental, elapsed_ms=42"); + + // Ready with DNS fallback (pick_first) + logHandler.clear(); + GrpcChannelTelemetry.logChannelReady( + TransportType.DIRECT_PATH_INTERCONNECT, + TransportType.CLOUD_PATH, + GrpcChannelTelemetry.DNS_LB_POLICY_NAME, + Duration.ofMillis(15), + true, + null); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getMessage()) + .contains( + "Note: google-c2p resolver fell back to DNS (pick_first); traffic is using CloudPath" + + " instead of DirectPathInterconnect."); + + // Ready with unexpected LB policy + logHandler.clear(); + GrpcChannelTelemetry.logChannelReady( + TransportType.DIRECT_PATH, + TransportType.DIRECT_PATH, + "round_robin", + Duration.ofMillis(20), + true, + null); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getMessage()) + .contains( + "Note: unexpected LB policy 'round_robin' for DirectPath (expected" + + " 'xds_cluster_manager_experimental')."); + + // Not ready on DirectPathInterconnect + logHandler.clear(); + GrpcChannelTelemetry.logChannelReady( + TransportType.DIRECT_PATH_INTERCONNECT, + TransportType.DIRECT_PATH_INTERCONNECT, + null, + Duration.ofMillis(30000), + false, + "DEADLINE_EXCEEDED"); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getMessage()) + .contains("gRPC channel [0] did not become ready before deadline."); + assertThat(logHandler.records.get(0).getMessage()).contains("status=DEADLINE_EXCEEDED"); + assertThat(logHandler.records.get(0).getMessage()) + .contains("GRPC_VERBOSITY=debug GRPC_TRACE=xds_client,xds_resolver,c2p_resolver"); + } + + @Test + public void startChannelTelemetry_immediateReady_invokesCallbackOnce() throws Exception { + FakeManagedChannel channel = + new FakeManagedChannel(ConnectivityState.READY, GrpcChannelTelemetry.XDS_LB_POLICY_NAME); + AtomicLong clockNanos = new AtomicLong(1_000_000_000L); + AtomicInteger callbackCount = new AtomicInteger(0); + AtomicReference reportedTransport = new AtomicReference<>(); + AtomicReference reportedLatency = new AtomicReference<>(); + + GrpcChannelTelemetry.startChannelTelemetry( + new WeakReference<>(channel), + "google-c2p:///storage-direct.googleapis.com?force-xds", + true, + Duration.ofSeconds(30), + (transport, latency) -> { + callbackCount.incrementAndGet(); + reportedTransport.set(transport); + reportedLatency.set(latency); + }, + scheduler, + () -> clockNanos.addAndGet(25_000_000L), + FakeManagedChannel::getLbPolicy) + .get(5, TimeUnit.SECONDS); + + assertThat(callbackCount.get()).isEqualTo(1); + assertThat(reportedTransport.get()).isEqualTo(TransportType.DIRECT_PATH_INTERCONNECT); + assertThat(reportedLatency.get()).isEqualTo(Duration.ofMillis(25)); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getMessage()).contains("gRPC channel [0] is ready."); + } + + @Test + public void startChannelTelemetry_asyncTransitionToReady_invokesCallbackOnce() throws Exception { + FakeManagedChannel channel = + new FakeManagedChannel(ConnectivityState.IDLE, GrpcChannelTelemetry.XDS_LB_POLICY_NAME); + AtomicLong clockNanos = new AtomicLong(0L); + AtomicInteger callbackCount = new AtomicInteger(0); + AtomicReference reportedTransport = new AtomicReference<>(); + AtomicReference reportedLatency = new AtomicReference<>(); + + CompletableFuture future = + GrpcChannelTelemetry.startChannelTelemetry( + new WeakReference<>(channel), + "google-c2p:///storage-direct.googleapis.com?force-xds", + true, + Duration.ofSeconds(30), + (transport, latency) -> { + callbackCount.incrementAndGet(); + reportedTransport.set(transport); + reportedLatency.set(latency); + }, + scheduler, + clockNanos::get, + FakeManagedChannel::getLbPolicy); + + assertThat(callbackCount.get()).isEqualTo(0); + + clockNanos.set(10_000_000L); + channel.transitionTo(ConnectivityState.CONNECTING); + assertThat(callbackCount.get()).isEqualTo(0); + + clockNanos.set(55_000_000L); + channel.transitionTo(ConnectivityState.READY); + future.get(5, TimeUnit.SECONDS); + + assertThat(callbackCount.get()).isEqualTo(1); + assertThat(reportedTransport.get()).isEqualTo(TransportType.DIRECT_PATH_INTERCONNECT); + assertThat(reportedLatency.get()).isEqualTo(Duration.ofMillis(55)); + } + + @Test + public void startChannelTelemetry_shutdownBeforeReady_doesNotInvokeCallback() throws Exception { + FakeManagedChannel channel = new FakeManagedChannel(ConnectivityState.CONNECTING, null); + AtomicInteger callbackCount = new AtomicInteger(0); + + CompletableFuture future = + GrpcChannelTelemetry.startChannelTelemetry( + new WeakReference<>(channel), + "google-c2p:///storage-direct.googleapis.com?force-xds", + true, + Duration.ofSeconds(30), + (transport, latency) -> callbackCount.incrementAndGet(), + scheduler, + System::nanoTime, + FakeManagedChannel::getLbPolicy); + + channel.transitionTo(ConnectivityState.SHUTDOWN); + future.get(5, TimeUnit.SECONDS); + + assertThat(callbackCount.get()).isEqualTo(0); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getMessage()) + .contains("status=channel shut down before becoming ready"); + } + + @Test + public void startChannelTelemetry_timeoutBeforeReady_doesNotInvokeCallback() throws Exception { + FakeManagedChannel channel = new FakeManagedChannel(ConnectivityState.CONNECTING, null); + AtomicInteger callbackCount = new AtomicInteger(0); + + CompletableFuture future = + GrpcChannelTelemetry.startChannelTelemetry( + new WeakReference<>(channel), + "google-c2p:///storage-direct.googleapis.com?force-xds", + true, + Duration.ofMillis(20), + (transport, latency) -> callbackCount.incrementAndGet(), + scheduler, + System::nanoTime, + FakeManagedChannel::getLbPolicy); + + future.get(5, TimeUnit.SECONDS); + + assertThat(callbackCount.get()).isEqualTo(0); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getMessage()).contains("status=DEADLINE_EXCEEDED"); + } + + @Test + public void startChannelTelemetry_clearedWeakReference_doesNotInvokeCallback() throws Exception { + WeakReference clearedRef = new WeakReference<>(null); + AtomicInteger callbackCount = new AtomicInteger(0); + + GrpcChannelTelemetry.startChannelTelemetry( + clearedRef, + "google-c2p:///storage-direct.googleapis.com?force-xds", + true, + Duration.ofSeconds(30), + (transport, latency) -> callbackCount.incrementAndGet(), + scheduler, + System::nanoTime, + ch -> null) + .get(5, TimeUnit.SECONDS); + + assertThat(callbackCount.get()).isEqualTo(0); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getMessage()) + .contains("status=channel destroyed before becoming ready"); + } + + @Test + public void createChannelPrimer_onlyObservesFirstChannel() { + AtomicInteger callbackCount = new AtomicInteger(0); + ChannelPrimer primer = + GrpcChannelTelemetry.createChannelPrimer( + "google-c2p:///storage-direct.googleapis.com?force-xds", + true, + (transport, latency) -> callbackCount.incrementAndGet()); + + FakeManagedChannel ch0 = + new FakeManagedChannel(ConnectivityState.READY, GrpcChannelTelemetry.XDS_LB_POLICY_NAME); + FakeManagedChannel ch1 = + new FakeManagedChannel(ConnectivityState.READY, GrpcChannelTelemetry.XDS_LB_POLICY_NAME); + + primer.primeChannel(ch0); + primer.primeChannel(ch1); + + assertThat(callbackCount.get()).isEqualTo(1); + assertThat(logHandler.records).hasSize(1); + } + + @Test + public void makeChannelMetricsCallback_recordsExpectedHistogramMetric() { + ChannelReadyCallback noopCallback = GrpcChannelTelemetry.makeChannelMetricsCallback(null); + noopCallback.onChannelReady(TransportType.CLOUD_PATH, Duration.ofMillis(10)); + + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProviderBuilder providerBuilder = + SdkMeterProvider.builder().registerMetricReader(metricReader); + OpenTelemetryBootstrappingUtils.registerGrpcChannelViews(providerBuilder); + + try (SdkMeterProvider meterProvider = providerBuilder.build()) { + ChannelReadyCallback callback = + GrpcChannelTelemetry.makeChannelMetricsCallback(meterProvider); + callback.onChannelReady(TransportType.DIRECT_PATH_INTERCONNECT, Duration.ofMillis(125)); + callback.onChannelReady(TransportType.DIRECT_PATH, Duration.ofMillis(50)); + callback.onChannelReady(TransportType.CLOUD_PATH, Duration.ofMillis(250)); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat(metrics).hasSize(1); + MetricData metricData = metrics.iterator().next(); + assertThat(metricData.getName()) + .isEqualTo(GrpcChannelTelemetry.CHANNEL_CREATION_LATENCY_VIEW_NAME); + assertThat(metricData.getUnit()) + .isEqualTo(GrpcChannelTelemetry.CHANNEL_CREATION_LATENCY_UNIT); + assertThat(metricData.getDescription()) + .isEqualTo(GrpcChannelTelemetry.CHANNEL_CREATION_LATENCY_DESCRIPTION); + + Collection points = metricData.getHistogramData().getPoints(); + assertThat(points).hasSize(3); + for (HistogramPointData point : points) { + String transportType = point.getAttributes().get(GrpcChannelTelemetry.TRANSPORT_TYPE_KEY); + assertThat(point.getCount()).isEqualTo(1); + if ("DirectPathInterconnect".equals(transportType)) { + assertThat(point.getSum()).isWithin(1e-6).of(0.125); + } else if ("DirectPath".equals(transportType)) { + assertThat(point.getSum()).isWithin(1e-6).of(0.050); + } else if ("CloudPath".equals(transportType)) { + assertThat(point.getSum()).isWithin(1e-6).of(0.250); + } else { + throw new AssertionError("Unexpected transport_type: " + transportType); + } + } + } + } + + @Test + public void startChannelTelemetry_dnsFallback_reportsCloudPath() throws Exception { + FakeManagedChannel channel = + new FakeManagedChannel(ConnectivityState.READY, GrpcChannelTelemetry.DNS_LB_POLICY_NAME); + AtomicReference reportedTransport = new AtomicReference<>(); + + GrpcChannelTelemetry.startChannelTelemetry( + new WeakReference<>(channel), + "google-c2p:///storage-direct.googleapis.com?force-xds", + true, + Duration.ofSeconds(30), + (transport, latency) -> reportedTransport.set(transport), + scheduler, + System::nanoTime, + FakeManagedChannel::getLbPolicy) + .get(5, TimeUnit.SECONDS); + + assertThat(reportedTransport.get()).isEqualTo(TransportType.CLOUD_PATH); + assertThat(logHandler.records).hasSize(1); + assertThat(logHandler.records.get(0).getMessage()) + .contains("transport_type=CloudPath, requested_transport_type=DirectPathInterconnect"); + } + + @Test + public void isDirectPathInterconnectEnabled_envPrecedence() { + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectEnabled(false, "true")).isTrue(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectEnabled(false, " TRUE ")).isTrue(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectEnabled(true, "false")).isFalse(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectEnabled(true, " FALSE ")).isFalse(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectEnabled(true, "invalid")).isTrue(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectEnabled(false, "invalid")).isFalse(); + assertThat(GrpcChannelTelemetry.isDirectPathInterconnectEnabled(true, null)).isTrue(); + } + + @Test + public void grpcStorageOptions_attemptDirectPathFalse_overridesDirectPathOverInterconnect() + throws Exception { + GrpcStorageOptions options = + StorageOptions.grpc() + .setAttemptDirectPath(false) + .setAttemptDirectPathXdsOverInterconnect(true) + .setCredentials(GoogleCredentials.create(new AccessToken("fake-token", null))) + .build(); + + StorageSettings settings = options.getStorageSettings(); + InstantiatingGrpcChannelProvider provider = + (InstantiatingGrpcChannelProvider) + settings.getTransportChannelProvider().withCredentials(options.getCredentials()); + + assertThat(provider.canUseDirectPath()).isFalse(); + assertThat(GrpcChannelTelemetry.resolveEffectiveTarget(provider, true)) + .isEqualTo("storage.googleapis.com:443"); + + if (!Boolean.parseBoolean(System.getenv("GOOGLE_CLOUD_DISABLE_DIRECT_PATH"))) { + GrpcStorageOptions enabledOptions = + StorageOptions.grpc() + .setAttemptDirectPathXdsOverInterconnect(true) + .setCredentials(GoogleCredentials.create(new AccessToken("fake-token", null))) + .build(); + InstantiatingGrpcChannelProvider enabledProvider = + (InstantiatingGrpcChannelProvider) + enabledOptions + .getStorageSettings() + .getTransportChannelProvider() + .withCredentials(enabledOptions.getCredentials()); + assertThat(GrpcChannelTelemetry.resolveEffectiveTarget(enabledProvider, true)) + .isEqualTo("google-c2p:///storage-direct.googleapis.com?force-xds"); + + GrpcStorageOptions preprodOptions = + StorageOptions.grpc() + .setHost("https://storage-preprod.sandbox.googleapis.com") + .setAttemptDirectPathXdsOverInterconnect(true) + .setCredentials(GoogleCredentials.create(new AccessToken("fake-token", null))) + .build(); + InstantiatingGrpcChannelProvider preprodProvider = + (InstantiatingGrpcChannelProvider) + preprodOptions + .getStorageSettings() + .getTransportChannelProvider() + .withCredentials(preprodOptions.getCredentials()); + assertThat(GrpcChannelTelemetry.resolveEffectiveTarget(preprodProvider, true)) + .isEqualTo("google-c2p:///storage-preprod-direct.sandbox.googleapis.com?force-xds"); + + GrpcStorageOptions unauthenticatedOptions = + StorageOptions.grpc() + .setAttemptDirectPathXdsOverInterconnect(true) + .setCredentials(NoCredentials.getInstance()) + .build(); + InstantiatingGrpcChannelProvider unauthenticatedProvider = + (InstantiatingGrpcChannelProvider) + unauthenticatedOptions.getStorageSettings().getTransportChannelProvider(); + assertThat(GrpcChannelTelemetry.resolveEffectiveTarget(unauthenticatedProvider, true)) + .isEqualTo("storage.googleapis.com:443"); + } + } + + private static final class FakeManagedChannel extends ManagedChannel { + private ConnectivityState state; + private final String lbPolicy; + private Runnable stateChangeCallback; + + FakeManagedChannel(ConnectivityState initialState, String lbPolicy) { + this.state = initialState; + this.lbPolicy = lbPolicy; + } + + static String getLbPolicy(ManagedChannel channel) { + return channel instanceof FakeManagedChannel ? ((FakeManagedChannel) channel).lbPolicy : null; + } + + @Override + public ConnectivityState getState(boolean requestConnection) { + return state; + } + + @Override + public void notifyWhenStateChanged(ConnectivityState source, Runnable callback) { + if (state != source) { + callback.run(); + } else { + this.stateChangeCallback = callback; + } + } + + void transitionTo(ConnectivityState newState) { + this.state = newState; + Runnable cb = this.stateChangeCallback; + this.stateChangeCallback = null; + if (cb != null) { + cb.run(); + } + } + + @Override + public ManagedChannel shutdown() { + transitionTo(ConnectivityState.SHUTDOWN); + return this; + } + + @Override + public boolean isShutdown() { + return state == ConnectivityState.SHUTDOWN; + } + + @Override + public boolean isTerminated() { + return state == ConnectivityState.SHUTDOWN; + } + + @Override + public ManagedChannel shutdownNow() { + transitionTo(ConnectivityState.SHUTDOWN); + return this; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) { + return true; + } + + @Override + public ClientCall newCall( + MethodDescriptor methodDescriptor, CallOptions callOptions) { + throw new UnsupportedOperationException(); + } + + @Override + public String authority() { + return "storage-direct.googleapis.com"; + } + } + + private static final class CapturingLogHandler extends Handler { + private final List records = new ArrayList<>(); + + @Override + public void publish(LogRecord record) { + records.add(record); + } + + @Override + public void flush() {} + + @Override + public void close() {} + + void clear() { + records.clear(); + } + } +} diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/OpenTelemetryBootstrappingUtilsTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/OpenTelemetryBootstrappingUtilsTest.java index c5844cfaee41..429f24dc1e9a 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/OpenTelemetryBootstrappingUtilsTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/OpenTelemetryBootstrappingUtilsTest.java @@ -215,4 +215,28 @@ public void createClientMeterProvider_preservesDetectedAttributes() { assertThat(provider.toString()).contains("host.id=\"test-host-123\""); } } + + @Test + public void testGetCloudMonitoringEndpoint() { + assertThat( + OpenTelemetryBootstrappingUtils.getCloudMonitoringEndpoint( + "storage.googleapis.com:443", null)) + .isEqualTo("monitoring.googleapis.com:443"); + assertThat( + OpenTelemetryBootstrappingUtils.getCloudMonitoringEndpoint( + "storage-direct.googleapis.com:443", null)) + .isEqualTo("monitoring.googleapis.com:443"); + assertThat( + OpenTelemetryBootstrappingUtils.getCloudMonitoringEndpoint( + "storage-preprod-direct.sandbox.googleapis.com:443", null)) + .isEqualTo("monitoring.googleapis.com:443"); + assertThat( + OpenTelemetryBootstrappingUtils.getCloudMonitoringEndpoint( + "storage-direct.apis-tpclp.goog:443", "apis-tpclp.goog")) + .isEqualTo("monitoring.apis-tpclp.goog:443"); + assertThat( + OpenTelemetryBootstrappingUtils.getCloudMonitoringEndpoint( + "google-c2p:///storage-direct.googleapis.com?force-xds", null)) + .isEqualTo("monitoring.googleapis.com:443"); + } }