feat: Fallback to cloudpath

Author: nimf

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableClientContext.java

@@ -153,7 +153,8 @@ public static BigtableClientContext create(EnhancedBigtableStubSettings settings
           BigtableTransportChannelProvider.create(
               (InstantiatingGrpcChannelProvider) transportProvider.build(),
               channelPrimer,
-              channelPoolMetricsTracer);
+              channelPoolMetricsTracer,
+              settings);
 
       builder.setTransportChannelProvider(btTransportProvider);
     }

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java

@@ -115,6 +115,9 @@ public class EnhancedBigtableStubSettings extends StubSettings<EnhancedBigtableS
   private static final boolean DIRECT_PATH_ENABLED =
       Boolean.parseBoolean(System.getenv("CBT_ENABLE_DIRECTPATH"));
 
+  private static final boolean FALLBACK_ENABLED =
+      Boolean.parseBoolean(System.getenv("CBT_ENABLE_FALLBACK"));
+
   // If true, disable the bound-token-by-default feature for DirectPath.
   private static final boolean DIRECT_PATH_BOUND_TOKEN_DISABLED =
       Boolean.parseBoolean(System.getenv("CBT_DISABLE_DIRECTPATH_BOUND_TOKEN"));
@@ -362,6 +365,16 @@ public String getAppProfileId() {
     return appProfileId;
   }
 
+  /** Returns if Direct Access is enabled. */
+  public boolean isDirectpathEnabled() {
+    return DIRECT_PATH_ENABLED;
+  }
+
+  /** Returns if fallback to cloudpath is enabled. */
+  public boolean isFallbackEnabled() {
+    return FALLBACK_ENABLED;
+  }
+
   /**
    * Returns if channels will gracefully refresh connections to Cloud Bigtable service
    *

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider.java

@@ -23,13 +23,30 @@
 import com.google.api.gax.rpc.TransportChannel;
 import com.google.api.gax.rpc.TransportChannelProvider;
 import com.google.auth.Credentials;
+import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStubSettings;
 import com.google.cloud.bigtable.data.v2.stub.metrics.ChannelPoolMetricsTracer;
+import com.google.cloud.bigtable.gaxx.grpc.fallback.GcpFallbackChannel;
+import com.google.cloud.bigtable.gaxx.grpc.fallback.GcpFallbackChannelOptions;
 import com.google.common.base.Preconditions;
+
+import io.grpc.Channel;
 import io.grpc.ManagedChannel;
+import io.grpc.StatusRuntimeException;
+
+import static io.grpc.Status.Code.DEADLINE_EXCEEDED;
+import static io.grpc.Status.Code.UNAUTHENTICATED;
+import static io.grpc.Status.Code.UNAVAILABLE;
+import static io.grpc.Status.Code.UNIMPLEMENTED;
+import static io.grpc.Status.Code.UNKNOWN;
+
 import java.io.IOException;
+import java.time.Duration;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Function;
+
 import javax.annotation.Nullable;
 
 /**
@@ -42,14 +59,17 @@ public final class BigtableTransportChannelProvider implements TransportChannelP
   private final InstantiatingGrpcChannelProvider delegate;
   private final ChannelPrimer channelPrimer;
   @Nullable private final ChannelPoolMetricsTracer channelPoolMetricsTracer;
+  @Nullable private final EnhancedBigtableStubSettings settings;
 
   private BigtableTransportChannelProvider(
       InstantiatingGrpcChannelProvider instantiatingGrpcChannelProvider,
       ChannelPrimer channelPrimer,
-      ChannelPoolMetricsTracer channelPoolMetricsTracer) {
+      ChannelPoolMetricsTracer channelPoolMetricsTracer,
+      EnhancedBigtableStubSettings settings) {
     delegate = Preconditions.checkNotNull(instantiatingGrpcChannelProvider);
     this.channelPrimer = channelPrimer;
     this.channelPoolMetricsTracer = channelPoolMetricsTracer;
+    this.settings = settings;
   }
 
   @Override
@@ -72,7 +92,7 @@ public BigtableTransportChannelProvider withExecutor(Executor executor) {
     InstantiatingGrpcChannelProvider newChannelProvider =
         (InstantiatingGrpcChannelProvider) delegate.withExecutor(executor);
     return new BigtableTransportChannelProvider(
-        newChannelProvider, channelPrimer, channelPoolMetricsTracer);
+        newChannelProvider, channelPrimer, channelPoolMetricsTracer, settings);
   }
 
   @Override
@@ -85,7 +105,7 @@ public BigtableTransportChannelProvider withHeaders(Map<String, String> headers)
     InstantiatingGrpcChannelProvider newChannelProvider =
         (InstantiatingGrpcChannelProvider) delegate.withHeaders(headers);
     return new BigtableTransportChannelProvider(
-        newChannelProvider, channelPrimer, channelPoolMetricsTracer);
+        newChannelProvider, channelPrimer, channelPoolMetricsTracer, settings);
   }
 
   @Override
@@ -98,7 +118,7 @@ public TransportChannelProvider withEndpoint(String endpoint) {
     InstantiatingGrpcChannelProvider newChannelProvider =
         (InstantiatingGrpcChannelProvider) delegate.withEndpoint(endpoint);
     return new BigtableTransportChannelProvider(
-        newChannelProvider, channelPrimer, channelPoolMetricsTracer);
+        newChannelProvider, channelPrimer, channelPoolMetricsTracer, settings);
   }
 
   @Deprecated
@@ -113,7 +133,7 @@ public TransportChannelProvider withPoolSize(int size) {
     InstantiatingGrpcChannelProvider newChannelProvider =
         (InstantiatingGrpcChannelProvider) delegate.withPoolSize(size);
     return new BigtableTransportChannelProvider(
-        newChannelProvider, channelPrimer, channelPoolMetricsTracer);
+        newChannelProvider, channelPrimer, channelPoolMetricsTracer, settings);
   }
 
   /** Expected to only be called once when BigtableClientContext is created */
@@ -144,14 +164,70 @@ public TransportChannel getTransportChannel() throws IOException {
 
     BigtableChannelPool btChannelPool =
         BigtableChannelPool.create(btPoolSettings, channelFactory, channelPrimer);
+    
+    ManagedChannel resultingChannel = btChannelPool;
+
+    // TODO: Also check if directpath is possible.
+    if (settings != null && settings.isFallbackEnabled() && settings.isDirectpathEnabled()) {
+      InstantiatingGrpcChannelProvider cloudpathChannelProvider =
+          delegate.toBuilder()
+            .setAttemptDirectPath(false)
+            .setChannelPoolSettings(ChannelPoolSettings.staticallySized(1))
+            .build();
+
+      ChannelFactory cloudpathFactory =
+        () -> {
+          try {
+            GrpcTransportChannel channel =
+                (GrpcTransportChannel) cloudpathChannelProvider.getTransportChannel();
+            return (ManagedChannel) channel.getChannel();
+          } catch (IOException e) {
+            throw new java.io.UncheckedIOException(e);
+          }
+        };
+
+      BigtableChannelPool btCloupathPool =
+        BigtableChannelPool.create(btPoolSettings, cloudpathFactory, channelPrimer);
+
+      Function<Channel, String> probingFn =
+          (channel) -> {
+            try {
+              channelPrimer.sendPrimeRequestsAsync((ManagedChannel) channel).get();
+            } catch (StatusRuntimeException e) {
+              return e.getStatus().getCode().toString();
+            } catch (Exception e) {
+              return "EXCEPTION";
+            }
+            return "";
+          };
+
+      // Default options for now, but with probing.
+      // TODO: enable oTel metrics if needed.
+      GcpFallbackChannelOptions fallbackOptions = GcpFallbackChannelOptions.newBuilder()
+        .setPrimaryChannelName("DIRECTPATH")
+        .setFallbackChannelName("CLOUDPATH")
+        .setEnableFallback(true)
+        .setPeriod(Duration.ofMinutes(1))
+        .setErroneousStates(Set.of(UNAVAILABLE, UNAUTHENTICATED, DEADLINE_EXCEEDED, UNKNOWN, UNIMPLEMENTED))
+        .setFallbackProbingInterval(Duration.ofMinutes(15))
+        .setPrimaryProbingInterval(Duration.ofMinutes(1))
+        .setMinFailedCalls(3)
+        .setErrorRateThreshold(1f)
+        .setFallbackProbingFunction(probingFn)
+        .setPrimaryProbingFunction(probingFn)
+        .build();
+
+      resultingChannel = new GcpFallbackChannel(fallbackOptions, btChannelPool, btCloupathPool);
+    }
 
     if (channelPoolMetricsTracer != null) {
-      channelPoolMetricsTracer.registerChannelInsightsProvider(btChannelPool::getChannelInfos);
+      // resultingChannel is either BigtableChannelPool or GcpFallbackChannel here and both implement BigtableChannelPoolObserver.
+      channelPoolMetricsTracer.registerChannelInsightsProvider(((BigtableChannelPoolObserver) resultingChannel)::getChannelInfos);
       channelPoolMetricsTracer.registerLoadBalancingStrategy(
           btPoolSettings.getLoadBalancingStrategy().name());
     }
 
-    return GrpcTransportChannel.create(btChannelPool);
+    return GrpcTransportChannel.create(resultingChannel);
   }
 
   @Override
@@ -169,7 +245,7 @@ public TransportChannelProvider withCredentials(Credentials credentials) {
     InstantiatingGrpcChannelProvider newChannelProvider =
         (InstantiatingGrpcChannelProvider) delegate.withCredentials(credentials);
     return new BigtableTransportChannelProvider(
-        newChannelProvider, channelPrimer, channelPoolMetricsTracer);
+        newChannelProvider, channelPrimer, channelPoolMetricsTracer, settings);
   }
 
   /** Creates a BigtableTransportChannelProvider. */
@@ -178,6 +254,15 @@ public static BigtableTransportChannelProvider create(
       ChannelPrimer channelPrimer,
       ChannelPoolMetricsTracer outstandingRpcsMetricTracke) {
     return new BigtableTransportChannelProvider(
-        instantiatingGrpcChannelProvider, channelPrimer, outstandingRpcsMetricTracke);
+        instantiatingGrpcChannelProvider, channelPrimer, outstandingRpcsMetricTracke, null);
+  }
+
+  public static BigtableTransportChannelProvider create(
+      InstantiatingGrpcChannelProvider instantiatingGrpcChannelProvider,
+      ChannelPrimer channelPrimer,
+      ChannelPoolMetricsTracer outstandingRpcsMetricTracer,
+      EnhancedBigtableStubSettings settings) {
+    return new BigtableTransportChannelProvider(
+        instantiatingGrpcChannelProvider, channelPrimer, outstandingRpcsMetricTracer, settings);
   }
 }