feat: Add Experimental DirectPath support (#8)

* Add Experimental DirectPath support

* finish removing DirectPathEnv

* format

* remove sys prop check, this will be enabled in the next version of grpc

* english

google-cloud-bigtable/pom.xml

@@ -257,7 +257,7 @@
                 </goals>
                 <configuration>
                   <systemPropertyVariables>
-                    <bigtable.env>direct_path</bigtable.env>
+                    <bigtable.env>prod</bigtable.env>
                     <!-- TODO(igorbernstein): This property should be auto set by gax -->
                     <io.grpc.internal.DnsNameResolverProvider.enable_grpclb>true</io.grpc.internal.DnsNameResolverProvider.enable_grpclb>
                   </systemPropertyVariables>

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

@@ -42,6 +42,7 @@
 import com.google.common.collect.ImmutableSet;
 import java.util.List;
 import java.util.Set;
+import java.util.logging.Logger;
 import javax.annotation.Nonnull;
 import org.threeten.bp.Duration;
 
@@ -74,10 +75,18 @@
  * }</pre>
  */
 public class EnhancedBigtableStubSettings extends StubSettings<EnhancedBigtableStubSettings> {
+  private static final Logger logger =
+      Logger.getLogger(EnhancedBigtableStubSettings.class.getName());
+
   // The largest message that can be received is a 256 MB ReadRowsResponse.
   private static final int MAX_MESSAGE_SIZE = 256 * 1024 * 1024;
   private static final String SERVER_DEFAULT_APP_PROFILE_ID = "";
 
+  // TODO(igorbernstein2): Remove this once DirectPath goes to public beta
+  // Temporary endpoint for the DirectPath private alpha
+  private static final String DIRECT_PATH_ENV_VAR = "GOOGLE_CLOUD_ENABLE_DIRECT_PATH";
+  private static final String DIRECT_PATH_ENDPOINT = "directpath-bigtable.googleapis.com:443";
+
   private static final Set<Code> IDEMPOTENT_RETRY_CODES =
       ImmutableSet.of(Code.DEADLINE_EXCEEDED, Code.UNAVAILABLE);
 
@@ -133,6 +142,12 @@ public class EnhancedBigtableStubSettings extends StubSettings<EnhancedBigtableS
   private EnhancedBigtableStubSettings(Builder builder) {
     super(builder);
 
+    if (DIRECT_PATH_ENDPOINT.equals(builder.getEndpoint())) {
+      logger.warning(
+          "Connecting to Bigtable using DirectPath."
+              + " This is currently an experimental feature and should not be used in production.");
+    }
+
     // Since point reads & streaming reads share the same base callable that converts grpc errors
     // into ApiExceptions, they must have the same retry codes.
     Preconditions.checkState(
@@ -404,7 +419,14 @@ private Builder() {
       // Defaults provider
       BigtableStubSettings.Builder baseDefaults = BigtableStubSettings.newBuilder();
 
-      setEndpoint(baseDefaults.getEndpoint());
+      // TODO(igorbernstein): remove this once DirectPath goes to public Beta and uses the default
+      // endpoint.
+      if (isDirectPathEnabled()) {
+        setEndpoint(DIRECT_PATH_ENDPOINT);
+      } else {
+        setEndpoint(baseDefaults.getEndpoint());
+      }
+
       setTransportChannelProvider(defaultTransportChannelProvider());
       setStreamWatchdogCheckInterval(baseDefaults.getStreamWatchdogCheckInterval());
       setStreamWatchdogProvider(baseDefaults.getStreamWatchdogProvider());
@@ -503,6 +525,22 @@ private static void copyRetrySettings(
       dest.setRetryableCodes(source.getRetryableCodes());
       dest.setRetrySettings(source.getRetrySettings());
     }
+
+    // TODO(igorbernstein): Remove this once DirectPath goes to public beta
+    // Extracted from InstantiatingGrpcChannelProvider#isDirectPathEnabled
+    private static boolean isDirectPathEnabled() {
+      String whiteList = System.getenv(DIRECT_PATH_ENV_VAR);
+      if (whiteList == null) {
+        return false;
+      }
+
+      for (String service : whiteList.split(",")) {
+        if (!service.isEmpty() && DIRECT_PATH_ENDPOINT.contains(service)) {
+          return true;
+        }
+      }
+      return false;
+    }
     // </editor-fold>
 
     // <editor-fold desc="Public API">

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/it/DirectPathFallbackIT.java

@@ -23,7 +23,6 @@
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.cloud.bigtable.data.v2.BigtableDataClient;
 import com.google.cloud.bigtable.data.v2.BigtableDataSettings;
-import com.google.cloud.bigtable.test_helpers.env.DirectPathEnv;
 import com.google.cloud.bigtable.test_helpers.env.TestEnvRule;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.alts.ComputeEngineChannelBuilder;
@@ -82,8 +81,8 @@ public DirectPathFallbackIT() {
   public void setup() throws IOException {
     assume()
         .withMessage("DirectPath integration tests can only run against DirectPathEnv")
-        .that(testEnvRule.env())
-        .isInstanceOf(DirectPathEnv.class);
+        .that(testEnvRule.env().isDirectPathEnabled())
+        .isTrue();
 
     BigtableDataSettings defaultSettings = testEnvRule.env().getDataClientSettings();
     InstantiatingGrpcChannelProvider defaultTransportProvider =

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/test_helpers/env/AbstractTestEnv.java

@@ -74,6 +74,10 @@ public boolean isInstanceAdminSupported() {
     return true;
   }
 
+  public boolean isDirectPathEnabled() {
+    return "bigtable".equals(System.getenv("GOOGLE_CLOUD_ENABLE_DIRECT_PATH"));
+  }
+
   void cleanUpStale() {
     cleanupStaleTables();
     if (isInstanceAdminSupported()) {

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/test_helpers/env/TestEnvRule.java

@@ -63,9 +63,6 @@ protected void before() throws Throwable {
       case "prod":
         testEnv = ProdEnv.fromSystemProperties();
         break;
-      case "direct_path":
-        testEnv = DirectPathEnv.create();
-        break;
       default:
         throw new IllegalArgumentException(
             String.format(