Skip to content
This repository was archived by the owner on Apr 7, 2026. It is now read-only.

Commit afdef61

Browse files
kinsaurraldekinsaurralde
committed
Add GCPFallback support
1 parent 3d585cf commit afdef61

1 file changed

Lines changed: 180 additions & 48 deletions

File tree

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

Lines changed: 180 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -51,19 +51,24 @@
5151
import com.google.api.gax.rpc.StatusCode;
5252
import com.google.api.gax.rpc.StatusCode.Code;
5353
import com.google.api.gax.rpc.StreamController;
54+
import com.google.api.gax.rpc.TransportChannel;
5455
import com.google.api.gax.rpc.TransportChannelProvider;
5556
import com.google.api.gax.rpc.UnaryCallSettings;
5657
import com.google.api.gax.rpc.UnaryCallable;
5758
import com.google.api.gax.rpc.UnavailableException;
5859
import com.google.api.gax.rpc.WatchdogProvider;
5960
import com.google.api.pathtemplate.PathTemplate;
61+
import com.google.auth.Credentials;
6062
import com.google.cloud.RetryHelper;
6163
import com.google.cloud.RetryHelper.RetryHelperException;
6264
import com.google.cloud.grpc.GcpManagedChannel;
6365
import com.google.cloud.grpc.GcpManagedChannelBuilder;
6466
import com.google.cloud.grpc.GcpManagedChannelOptions;
6567
import com.google.cloud.grpc.GcpManagedChannelOptions.GcpMetricsOptions;
6668
import com.google.cloud.grpc.GrpcTransportOptions;
69+
import com.google.cloud.grpc.fallback.GcpFallbackChannel;
70+
import com.google.cloud.grpc.fallback.GcpFallbackChannelOptions;
71+
import com.google.cloud.grpc.fallback.GcpFallbackOpenTelemetry;
6772
import com.google.cloud.spanner.AdminRequestsPerMinuteExceededException;
6873
import com.google.cloud.spanner.BackupId;
6974
import com.google.cloud.spanner.ErrorCode;
@@ -185,9 +190,21 @@
185190
import com.google.spanner.v1.SpannerGrpc;
186191
import com.google.spanner.v1.Transaction;
187192
import io.grpc.CallCredentials;
193+
import io.grpc.CallOptions;
194+
import io.grpc.Channel;
195+
import io.grpc.ClientCall;
196+
import io.grpc.ClientInterceptor;
188197
import io.grpc.Context;
198+
import io.grpc.ForwardingChannelBuilder2;
199+
import io.grpc.ManagedChannel;
189200
import io.grpc.ManagedChannelBuilder;
190201
import io.grpc.MethodDescriptor;
202+
import io.grpc.NameResolver;
203+
import io.grpc.StatusRuntimeException;
204+
import io.grpc.auth.MoreCallCredentials;
205+
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
206+
import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
207+
import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioSocketChannel;
191208
import java.io.IOException;
192209
import java.io.UnsupportedEncodingException;
193210
import java.net.URLDecoder;
@@ -214,6 +231,7 @@
214231
import java.util.concurrent.Future;
215232
import java.util.concurrent.ScheduledExecutorService;
216233
import java.util.concurrent.TimeUnit;
234+
import java.util.concurrent.atomic.AtomicReference;
217235
import java.util.stream.Collectors;
218236
import java.util.stream.Stream;
219237
import javax.annotation.Nullable;
@@ -341,65 +359,89 @@ public GapicSpannerRpc(final SpannerOptions options) {
341359
this.isDynamicChannelPoolEnabled = options.isDynamicChannelPoolEnabled();
342360
this.baseGrpcCallContext = createBaseCallContext();
343361

362+
boolean isEnableDirectAccess = options.isEnableDirectAccess();
363+
344364
if (initializeStubs) {
345-
// First check if SpannerOptions provides a TransportChannelProvider. Create one
346-
// with information gathered from SpannerOptions if none is provided
347-
InstantiatingGrpcChannelProvider.Builder defaultChannelProviderBuilder =
348-
InstantiatingGrpcChannelProvider.newBuilder()
349-
.setChannelConfigurator(options.getChannelConfigurator())
350-
.setEndpoint(options.getEndpoint())
351-
.setMaxInboundMessageSize(MAX_MESSAGE_SIZE)
352-
.setMaxInboundMetadataSize(MAX_METADATA_SIZE)
353-
.setPoolSize(options.getNumChannels())
354-
355-
// Set a keepalive time of 120 seconds to help long running
356-
// commit GRPC calls succeed
357-
.setKeepAliveTimeDuration(Duration.ofSeconds(GRPC_KEEPALIVE_SECONDS))
358-
359-
// Then check if SpannerOptions provides an InterceptorProvider. Create a default
360-
// SpannerInterceptorProvider if none is provided
361-
.setInterceptorProvider(
362-
SpannerInterceptorProvider.create(
363-
MoreObjects.firstNonNull(
364-
options.getInterceptorProvider(),
365-
SpannerInterceptorProvider.createDefault(options.getOpenTelemetry())))
366-
// This sets the trace context headers.
367-
.withTraceContext(endToEndTracingEnabled, options.getOpenTelemetry())
368-
// This sets the response compressor (Server -> Client).
369-
.withEncoding(compressorName))
370-
.setHeaderProvider(headerProviderWithUserAgent)
371-
.setAllowNonDefaultServiceAccount(true);
372-
boolean isEnableDirectAccess = options.isEnableDirectAccess();
373-
if (isEnableDirectAccess) {
374-
defaultChannelProviderBuilder.setAttemptDirectPath(true);
375-
if (isEnableDirectPathBoundToken()) {
376-
// This will let the credentials try to fetch a hard-bound access token if the runtime
377-
// environment supports it.
378-
defaultChannelProviderBuilder.setAllowHardBoundTokenTypes(
379-
Collections.singletonList(InstantiatingGrpcChannelProvider.HardBoundTokenTypes.ALTS));
365+
CredentialsProvider credentialsProvider =
366+
GrpcTransportOptions.setUpCredentialsProvider(options);
367+
368+
InstantiatingGrpcChannelProvider.Builder defaultChannelProviderBuilder = getDefaultChannelProviderBuilder(options, headerProviderWithUserAgent, isEnableDirectAccess);
369+
370+
if (options.getChannelProvider() == null && isEnableDirectAccess && isEnableGcpFallbackEnv()) {
371+
InstantiatingGrpcChannelProvider.Builder cloudPathProviderBuilder = getDefaultChannelProviderBuilder(options, headerProviderWithUserAgent, /*isEnableDirectAccess=*/false);
372+
cloudPathProviderBuilder.setAttemptDirectPath(false);
373+
374+
final AtomicReference<ManagedChannelBuilder> cloudPathBuilderRef = new AtomicReference<>();
375+
cloudPathProviderBuilder.setChannelConfigurator(builder -> {
376+
cloudPathBuilderRef.set(builder);
377+
return builder;
378+
});
379+
380+
try (TransportChannel ignored = cloudPathProviderBuilder.build().getTransportChannel()) {
381+
} catch (Exception e) {
382+
throw asSpannerException(e);
380383
}
381-
defaultChannelProviderBuilder.setAttemptDirectPathXds();
382-
}
383384

384-
options.enablegRPCMetrics(defaultChannelProviderBuilder);
385+
ManagedChannelBuilder cloudPathBuilder = cloudPathBuilderRef.get();
386+
if (cloudPathBuilder == null) {
387+
throw new IllegalStateException("CloudPath builder was not captured.");
388+
}
385389

386-
if (options.isUseVirtualThreads()) {
387-
ExecutorService executor =
388-
tryCreateVirtualThreadPerTaskExecutor("spanner-virtual-grpc-executor");
389-
if (executor != null) {
390-
defaultChannelProviderBuilder.setExecutor(executor);
390+
try {
391+
Credentials credentials = credentialsProvider.getCredentials();
392+
if (credentials != null) {
393+
cloudPathBuilder.intercept(new ClientInterceptor() {
394+
@Override
395+
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
396+
MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
397+
return next.newCall(method, callOptions.withCallCredentials(MoreCallCredentials.from(credentials)));
398+
}
399+
});
400+
}
401+
} catch (Exception e) {
402+
throw asSpannerException(e);
391403
}
404+
405+
defaultChannelProviderBuilder.setChannelConfigurator(directPathBuilder -> {
406+
String jsonApiConfig = parseGrpcGcpApiConfig();
407+
GcpManagedChannelOptions gcpOptions = options.getGrpcGcpOptions();
408+
if (gcpOptions == null) {
409+
gcpOptions = GcpManagedChannelOptions.newBuilder().build();
410+
}
411+
412+
GcpManagedChannelBuilder primaryGcpBuilder =
413+
GcpManagedChannelBuilder.forDelegateBuilder(directPathBuilder)
414+
.withApiConfigJsonString(jsonApiConfig)
415+
.withOptions(gcpOptions);
416+
417+
GcpManagedChannelBuilder fallbackGcpBuilder =
418+
GcpManagedChannelBuilder.forDelegateBuilder(cloudPathBuilder)
419+
.withApiConfigJsonString(jsonApiConfig)
420+
.withOptions(gcpOptions);
421+
422+
GcpFallbackOpenTelemetry fallbackTelemetry = GcpFallbackOpenTelemetry.newBuilder()
423+
.withSdk(options.getOpenTelemetry())
424+
.build();
425+
426+
return new FallbackChannelBuilder(
427+
primaryGcpBuilder,
428+
fallbackGcpBuilder,
429+
GcpFallbackChannelOptions.newBuilder()
430+
.setPrimaryChannelName("directpath")
431+
.setFallbackChannelName("cloudpath")
432+
.setMinFailedCalls(1)
433+
.setGcpFallbackOpenTelemetry(fallbackTelemetry)
434+
.build()
435+
);
436+
});
392437
}
393-
// If it is enabled in options uses the channel pool provided by the gRPC-GCP extension.
394-
maybeEnableGrpcGcpExtension(defaultChannelProviderBuilder, options);
395438

439+
// First check if SpannerOptions provides a TransportChannelProvider. Create one
440+
// with information gathered from SpannerOptions if none is provided
396441
TransportChannelProvider channelProvider =
397442
MoreObjects.firstNonNull(
398443
options.getChannelProvider(), defaultChannelProviderBuilder.build());
399444

400-
CredentialsProvider credentialsProvider =
401-
GrpcTransportOptions.setUpCredentialsProvider(options);
402-
403445
spannerWatchdog =
404446
Executors.newSingleThreadScheduledExecutor(
405447
new ThreadFactoryBuilder()
@@ -572,6 +614,57 @@ private static String parseGrpcGcpApiConfig() {
572614
}
573615
}
574616

617+
private InstantiatingGrpcChannelProvider.Builder getDefaultChannelProviderBuilder(final SpannerOptions options, final HeaderProvider headerProviderWithUserAgent, Boolean isEnableDirectAccess) {
618+
InstantiatingGrpcChannelProvider.Builder defaultChannelProviderBuilder =
619+
InstantiatingGrpcChannelProvider.newBuilder()
620+
.setChannelConfigurator(options.getChannelConfigurator())
621+
.setEndpoint(options.getEndpoint())
622+
.setMaxInboundMessageSize(MAX_MESSAGE_SIZE)
623+
.setMaxInboundMetadataSize(MAX_METADATA_SIZE)
624+
.setPoolSize(options.getNumChannels())
625+
626+
// Set a keepalive time of 120 seconds to help long running
627+
// commit GRPC calls succeed
628+
.setKeepAliveTimeDuration(Duration.ofSeconds(GRPC_KEEPALIVE_SECONDS))
629+
630+
// Then check if SpannerOptions provides an InterceptorProvider. Create a default
631+
// SpannerInterceptorProvider if none is provided
632+
.setInterceptorProvider(
633+
SpannerInterceptorProvider.create(
634+
MoreObjects.firstNonNull(
635+
options.getInterceptorProvider(),
636+
SpannerInterceptorProvider.createDefault(options.getOpenTelemetry())))
637+
// This sets the trace context headers.
638+
.withTraceContext(endToEndTracingEnabled, options.getOpenTelemetry())
639+
// This sets the response compressor (Server -> Client).
640+
.withEncoding(compressorName))
641+
.setHeaderProvider(headerProviderWithUserAgent)
642+
.setAllowNonDefaultServiceAccount(true);
643+
if (isEnableDirectAccess) {
644+
defaultChannelProviderBuilder.setAttemptDirectPath(true);
645+
defaultChannelProviderBuilder.setAttemptDirectPathXds();
646+
if (isEnableDirectPathBoundToken()) {
647+
// This will let the credentials try to fetch a hard-bound access token if the runtime
648+
// environment supports it.
649+
defaultChannelProviderBuilder.setAllowHardBoundTokenTypes(
650+
Collections.singletonList(InstantiatingGrpcChannelProvider.HardBoundTokenTypes.ALTS));
651+
}
652+
}
653+
654+
options.enablegRPCMetrics(defaultChannelProviderBuilder);
655+
656+
if (options.isUseVirtualThreads()) {
657+
ExecutorService executor =
658+
tryCreateVirtualThreadPerTaskExecutor("spanner-virtual-grpc-executor");
659+
if (executor != null) {
660+
defaultChannelProviderBuilder.setExecutor(executor);
661+
}
662+
}
663+
// If it is enabled in options uses the channel pool provided by the gRPC-GCP extension.
664+
maybeEnableGrpcGcpExtension(defaultChannelProviderBuilder, options);
665+
return defaultChannelProviderBuilder;
666+
}
667+
575668
// Enhance gRPC-GCP options with metrics and dynamic channel pool configuration.
576669
private static GcpManagedChannelOptions grpcGcpOptionsWithMetricsAndDcp(SpannerOptions options) {
577670
GcpManagedChannelOptions grpcGcpOptions =
@@ -715,6 +808,10 @@ public static boolean isEnableDirectPathBoundToken() {
715808
return !Boolean.parseBoolean(System.getenv("GOOGLE_SPANNER_DISABLE_DIRECT_ACCESS_BOUND_TOKEN"));
716809
}
717810

811+
public static boolean isEnableGcpFallbackEnv() {
812+
return Boolean.parseBoolean(System.getenv("GOOGLE_SPANNER_ENABLE_GCP_FALLBACK"));
813+
}
814+
718815
private static final RetrySettings ADMIN_REQUESTS_LIMIT_EXCEEDED_RETRY_SETTINGS =
719816
RetrySettings.newBuilder()
720817
.setInitialRetryDelayDuration(Duration.ofSeconds(5L))
@@ -2313,4 +2410,39 @@ private static Duration systemProperty(String name, int defaultValue) {
23132410
String stringValue = System.getProperty(name, "");
23142411
return Duration.ofSeconds(stringValue.isEmpty() ? defaultValue : Integer.parseInt(stringValue));
23152412
}
2413+
2414+
// Wrapper class to build the GcpFallbackChannel using GAX's configuration
2415+
private static class FallbackChannelBuilder extends ForwardingChannelBuilder2<FallbackChannelBuilder> {
2416+
private final GcpFallbackChannelOptions options;
2417+
2418+
private final GcpManagedChannelBuilder primaryGcpBuilder;
2419+
private final GcpManagedChannelBuilder fallbackGcpBuilder;
2420+
2421+
private FallbackChannelBuilder(
2422+
GcpManagedChannelBuilder primary,
2423+
GcpManagedChannelBuilder fallback,
2424+
GcpFallbackChannelOptions options) {
2425+
this.primaryGcpBuilder = primary;
2426+
this.fallbackGcpBuilder = fallback;
2427+
this.options = options;
2428+
}
2429+
2430+
/**
2431+
* Delegates all configuration calls (e.g., interceptors, userAgent) to the primary builder.
2432+
* This ensures the primary channel receives all of GAX's standard configuration.
2433+
*/
2434+
@Override
2435+
protected ManagedChannelBuilder<?> delegate() {
2436+
return primaryGcpBuilder;
2437+
}
2438+
2439+
/**
2440+
* Overrides the build method to return our custom GcpFallbackChannel
2441+
* instead of a standard gRPC channel.
2442+
*/
2443+
@Override
2444+
public ManagedChannel build() {
2445+
return new GcpFallbackChannel(options, primaryGcpBuilder, fallbackGcpBuilder);
2446+
}
2447+
}
23162448
}

0 commit comments

Comments
 (0)