feat: Add E2E fallback to the spanner client.

google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java

@@ -58,13 +58,17 @@
 import com.google.api.gax.rpc.UnavailableException;
 import com.google.api.gax.rpc.WatchdogProvider;
 import com.google.api.pathtemplate.PathTemplate;
+import com.google.auth.Credentials;
 import com.google.cloud.RetryHelper;
 import com.google.cloud.RetryHelper.RetryHelperException;
 import com.google.cloud.grpc.GcpManagedChannel;
 import com.google.cloud.grpc.GcpManagedChannelBuilder;
 import com.google.cloud.grpc.GcpManagedChannelOptions;
 import com.google.cloud.grpc.GcpManagedChannelOptions.GcpMetricsOptions;
 import com.google.cloud.grpc.GrpcTransportOptions;
+import com.google.cloud.grpc.fallback.GcpFallbackChannel;
+import com.google.cloud.grpc.fallback.GcpFallbackChannelOptions;
+import com.google.cloud.grpc.fallback.GcpFallbackOpenTelemetry;
 import com.google.cloud.spanner.AdminRequestsPerMinuteExceededException;
 import com.google.cloud.spanner.BackupId;
 import com.google.cloud.spanner.ErrorCode;
@@ -187,16 +191,23 @@
 import com.google.spanner.v1.SpannerGrpc;
 import com.google.spanner.v1.Transaction;
 import io.grpc.CallCredentials;
+import io.grpc.CallOptions;
 import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
 import io.grpc.Context;
+import io.grpc.ForwardingChannelBuilder2;
+import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.MethodDescriptor;
+import io.grpc.auth.MoreCallCredentials;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.net.URLDecoder;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -217,6 +228,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
@@ -347,59 +359,101 @@ public GapicSpannerRpc(final SpannerOptions options) {
     this.isDynamicChannelPoolEnabled = options.isDynamicChannelPoolEnabled();
     this.baseGrpcCallContext = createBaseCallContext();
 
+    boolean isEnableDirectAccess = options.isEnableDirectAccess();
+
     if (initializeStubs) {
-      // First check if SpannerOptions provides a TransportChannelProvider. Create one
-      // with information gathered from SpannerOptions if none is provided
+      CredentialsProvider credentialsProvider =
+          GrpcTransportOptions.setUpCredentialsProvider(options);
+
       InstantiatingGrpcChannelProvider.Builder defaultChannelProviderBuilder =
-          InstantiatingGrpcChannelProvider.newBuilder()
-              .setChannelConfigurator(options.getChannelConfigurator())
-              .setEndpoint(options.getEndpoint())
-              .setMaxInboundMessageSize(MAX_MESSAGE_SIZE)
-              .setMaxInboundMetadataSize(MAX_METADATA_SIZE)
-              .setPoolSize(options.getNumChannels())
-
-              // Set a keepalive time of 120 seconds to help long running
-              // commit GRPC calls succeed
-              .setKeepAliveTimeDuration(Duration.ofSeconds(GRPC_KEEPALIVE_SECONDS))
-
-              // Then check if SpannerOptions provides an InterceptorProvider. Create a default
-              // SpannerInterceptorProvider if none is provided
-              .setInterceptorProvider(
-                  SpannerInterceptorProvider.create(
-                          MoreObjects.firstNonNull(
-                              options.getInterceptorProvider(),
-                              SpannerInterceptorProvider.createDefault(options.getOpenTelemetry())))
-                      // This sets the trace context headers.
-                      .withTraceContext(endToEndTracingEnabled, options.getOpenTelemetry())
-                      // This sets the response compressor (Server -> Client).
-                      .withEncoding(compressorName))
-              .setHeaderProvider(headerProviderWithUserAgent)
-              .setAllowNonDefaultServiceAccount(true);
-      boolean isEnableDirectAccess = options.isEnableDirectAccess();
-      if (isEnableDirectAccess) {
-        defaultChannelProviderBuilder.setAttemptDirectPath(true);
-        if (isEnableDirectPathBoundToken()) {
-          // This will let the credentials try to fetch a hard-bound access token if the runtime
-          // environment supports it.
-          defaultChannelProviderBuilder.setAllowHardBoundTokenTypes(
-              Collections.singletonList(InstantiatingGrpcChannelProvider.HardBoundTokenTypes.ALTS));
+          createChannelProviderBuilder(options, headerProviderWithUserAgent, isEnableDirectAccess);
+
+      if (options.getChannelProvider() == null
+          && isEnableDirectAccess
+          && isEnableGcpFallbackEnv()) {
+        InstantiatingGrpcChannelProvider.Builder cloudPathProviderBuilder =
+            createChannelProviderBuilder(
+                options, headerProviderWithUserAgent, /* isEnableDirectAccess= */ false);
+
+        final AtomicReference<ManagedChannelBuilder> cloudPathBuilderRef = new AtomicReference<>();
+        cloudPathProviderBuilder.setChannelConfigurator(
+            builder -> {
+              if (options.getChannelConfigurator() != null) {
+                builder = options.getChannelConfigurator().apply(builder);
+              }
+              cloudPathBuilderRef.set(builder);
+              return builder;
+            });
+
+        // Build the cloudPathProvider to extract the builder which will be provided to
+        // FallbackChannelBuilder.
+        try (TransportChannel ignored = cloudPathProviderBuilder.build().getTransportChannel()) {
+        } catch (Exception e) {
+          throw asSpannerException(e);
         }
-        defaultChannelProviderBuilder.setAttemptDirectPathXds();
-      }
 
-      options.enablegRPCMetrics(defaultChannelProviderBuilder);
+        ManagedChannelBuilder cloudPathBuilder = cloudPathBuilderRef.get();
+        if (cloudPathBuilder == null) {
+          throw new IllegalStateException("CloudPath builder was not captured.");
+        }
 
-      if (options.isUseVirtualThreads()) {
-        ExecutorService executor =
-            tryCreateVirtualThreadPerTaskExecutor("spanner-virtual-grpc-executor");
-        if (executor != null) {
-          defaultChannelProviderBuilder.setExecutor(executor);
+        try {
+          Credentials credentials = credentialsProvider.getCredentials();
+          if (credentials != null) {
+            cloudPathBuilder.intercept(
+                new ClientInterceptor() {
+                  @Override
+                  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+                      MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+                    return next.newCall(
+                        method,
+                        callOptions.withCallCredentials(MoreCallCredentials.from(credentials)));
+                  }
+                });
+          }
+        } catch (Exception e) {
+          throw asSpannerException(e);
         }
+
+        defaultChannelProviderBuilder.setChannelConfigurator(
+            directPathBuilder -> {
+              if (options.getChannelConfigurator() != null) {
+                directPathBuilder = options.getChannelConfigurator().apply(directPathBuilder);
+              }
+
+              String jsonApiConfig = parseGrpcGcpApiConfig();
+              GcpManagedChannelOptions gcpOptions = grpcGcpOptionsWithMetricsAndDcp(options);
+              if (gcpOptions == null) {
+                gcpOptions = GcpManagedChannelOptions.newBuilder().build();
+              }
+
+              GcpManagedChannelBuilder primaryGcpBuilder =
+                  GcpManagedChannelBuilder.forDelegateBuilder(directPathBuilder)
+                      .withApiConfigJsonString(jsonApiConfig)
+                      .withOptions(gcpOptions);
+
+              GcpManagedChannelBuilder fallbackGcpBuilder =
+                  GcpManagedChannelBuilder.forDelegateBuilder(cloudPathBuilder)
+                      .withApiConfigJsonString(jsonApiConfig)
+                      .withOptions(gcpOptions);
+
+              GcpFallbackOpenTelemetry fallbackTelemetry =
+                  GcpFallbackOpenTelemetry.newBuilder()
+                      .withSdk(options.getOpenTelemetry())
+                      .disableAllMetrics()
+                      .enableMetrics(Arrays.asList("fallback_count", "call_status"))
+                      .build();
+
+              return new FallbackChannelBuilder(
+                  primaryGcpBuilder,
+                  fallbackGcpBuilder,
+                  createFallbackChannelOptions(fallbackTelemetry, 1));
+            });
       }
-      // If it is enabled in options uses the channel pool provided by the gRPC-GCP extension.
-      maybeEnableGrpcGcpExtension(defaultChannelProviderBuilder, options);
 
       boolean enableLocationApi = options.isEnableLocationApi();
+      // First check if SpannerOptions provides a TransportChannelProvider. Create one
+      // with information gathered from SpannerOptions if none is provided
       TransportChannelProvider baseChannelProvider =
           MoreObjects.firstNonNull(
               options.getChannelProvider(), defaultChannelProviderBuilder.build());
@@ -410,9 +464,6 @@ public GapicSpannerRpc(final SpannerOptions options) {
                   options.getChannelEndpointCacheFactory())
               : baseChannelProvider;
 
-      CredentialsProvider credentialsProvider =
-          GrpcTransportOptions.setUpCredentialsProvider(options);
-
       spannerWatchdog =
           Executors.newSingleThreadScheduledExecutor(
               new ThreadFactoryBuilder()
@@ -578,6 +629,17 @@ public <RequestT, ResponseT> UnaryCallable<RequestT, ResponseT> createUnaryCalla
     }
   }
 
+  @VisibleForTesting
+  GcpFallbackChannelOptions createFallbackChannelOptions(
+      GcpFallbackOpenTelemetry fallbackTelemetry, int minFailedCalls) {
+    return GcpFallbackChannelOptions.newBuilder()
+        .setPrimaryChannelName("directpath")
+        .setFallbackChannelName("cloudpath")
+        .setMinFailedCalls(minFailedCalls)
+        .setGcpFallbackOpenTelemetry(fallbackTelemetry)
+        .build();
+  }
+
   private static KeyAwareChannel extractKeyAwareChannel(TransportChannel transportChannel) {
     if (transportChannel instanceof GrpcTransportChannel) {
       Channel channel = ((GrpcTransportChannel) transportChannel).getChannel();
@@ -604,6 +666,60 @@ private static String parseGrpcGcpApiConfig() {
     }
   }
 
+  private InstantiatingGrpcChannelProvider.Builder createChannelProviderBuilder(
+      final SpannerOptions options,
+      final HeaderProvider headerProviderWithUserAgent,
+      boolean isEnableDirectAccess) {
+    InstantiatingGrpcChannelProvider.Builder defaultChannelProviderBuilder =
+        InstantiatingGrpcChannelProvider.newBuilder()
+            .setChannelConfigurator(options.getChannelConfigurator())
+            .setEndpoint(options.getEndpoint())
+            .setMaxInboundMessageSize(MAX_MESSAGE_SIZE)
+            .setMaxInboundMetadataSize(MAX_METADATA_SIZE)
+            .setPoolSize(options.getNumChannels())
+
+            // Set a keepalive time of 120 seconds to help long running
+            // commit GRPC calls succeed
+            .setKeepAliveTimeDuration(Duration.ofSeconds(GRPC_KEEPALIVE_SECONDS))
+
+            // Then check if SpannerOptions provides an InterceptorProvider. Create a default
+            // SpannerInterceptorProvider if none is provided
+            .setInterceptorProvider(
+                SpannerInterceptorProvider.create(
+                        MoreObjects.firstNonNull(
+                            options.getInterceptorProvider(),
+                            SpannerInterceptorProvider.createDefault(options.getOpenTelemetry())))
+                    // This sets the trace context headers.
+                    .withTraceContext(endToEndTracingEnabled, options.getOpenTelemetry())
+                    // This sets the response compressor (Server -> Client).
+                    .withEncoding(compressorName))
+            .setHeaderProvider(headerProviderWithUserAgent)
+            .setAllowNonDefaultServiceAccount(true);
+    if (isEnableDirectAccess) {
+      defaultChannelProviderBuilder.setAttemptDirectPath(true);
+      if (isEnableDirectPathBoundToken()) {
+        // This will let the credentials try to fetch a hard-bound access token if the runtime
+        // environment supports it.
+        defaultChannelProviderBuilder.setAllowHardBoundTokenTypes(
+            Collections.singletonList(InstantiatingGrpcChannelProvider.HardBoundTokenTypes.ALTS));
+      }
+      defaultChannelProviderBuilder.setAttemptDirectPathXds();
+    }
+
+    options.enablegRPCMetrics(defaultChannelProviderBuilder);
+
+    if (options.isUseVirtualThreads()) {
+      ExecutorService executor =
+          tryCreateVirtualThreadPerTaskExecutor("spanner-virtual-grpc-executor");
+      if (executor != null) {
+        defaultChannelProviderBuilder.setExecutor(executor);
+      }
+    }
+    // If it is enabled in options uses the channel pool provided by the gRPC-GCP extension.
+    maybeEnableGrpcGcpExtension(defaultChannelProviderBuilder, options);
+    return defaultChannelProviderBuilder;
+  }
+
   // Enhance gRPC-GCP options with metrics and dynamic channel pool configuration.
   private static GcpManagedChannelOptions grpcGcpOptionsWithMetricsAndDcp(SpannerOptions options) {
     GcpManagedChannelOptions grpcGcpOptions =
@@ -749,6 +865,15 @@ public static boolean isEnableDirectPathBoundToken() {
     return !Boolean.parseBoolean(System.getenv("GOOGLE_SPANNER_DISABLE_DIRECT_ACCESS_BOUND_TOKEN"));
   }
 
+  @VisibleForTesting static Boolean enableGcpFallbackEnv = null;
+
+  public static boolean isEnableGcpFallbackEnv() {
+    if (enableGcpFallbackEnv != null) {
+      return enableGcpFallbackEnv;
+    }
+    return Boolean.parseBoolean(System.getenv("GOOGLE_SPANNER_ENABLE_GCP_FALLBACK"));
+  }
+
   private static final RetrySettings ADMIN_REQUESTS_LIMIT_EXCEEDED_RETRY_SETTINGS =
       RetrySettings.newBuilder()
           .setInitialRetryDelayDuration(Duration.ofSeconds(5L))
@@ -2347,4 +2472,40 @@ private static Duration systemProperty(String name, int defaultValue) {
     String stringValue = System.getProperty(name, "");
     return Duration.ofSeconds(stringValue.isEmpty() ? defaultValue : Integer.parseInt(stringValue));
   }
+
+  // Wrapper class to build the GcpFallbackChannel using GAX's configuration
+  private static class FallbackChannelBuilder
+      extends ForwardingChannelBuilder2<FallbackChannelBuilder> {
+    private final GcpFallbackChannelOptions options;
+
+    private final GcpManagedChannelBuilder primaryGcpBuilder;
+    private final GcpManagedChannelBuilder fallbackGcpBuilder;
+
+    private FallbackChannelBuilder(
+        GcpManagedChannelBuilder primary,
+        GcpManagedChannelBuilder fallback,
+        GcpFallbackChannelOptions options) {
+      this.primaryGcpBuilder = primary;
+      this.fallbackGcpBuilder = fallback;
+      this.options = options;
+    }
+
+    /**
+     * Delegates all configuration calls (e.g., interceptors, userAgent) to the primary builder.
+     * This ensures the primary channel receives all of GAX's standard configuration.
+     */
+    @Override
+    protected ManagedChannelBuilder<?> delegate() {
+      return primaryGcpBuilder;
+    }
+
+    /**
+     * Overrides the build method to return our custom GcpFallbackChannel instead of a standard gRPC
+     * channel.
+     */
+    @Override
+    public ManagedChannel build() {
+      return new GcpFallbackChannel(options, primaryGcpBuilder, fallbackGcpBuilder);
+    }
+  }
 }