feat: add google-c2p dependence to DirectPath (#1521)

* feat: add google-c2p dependence to DirectPath

Co-authored-by: Chanseok Oh <chanseok@google.com>

Author: mohanli-ml

build.gradle

@@ -40,6 +40,7 @@ ext {
     'maven.io_grpc_grpc_protobuf': "io.grpc:grpc-protobuf:${libraries['version.io_grpc']}",
     'maven.io_grpc_grpc_netty_shaded': "io.grpc:grpc-netty-shaded:${libraries['version.io_grpc']}",
     'maven.io_grpc_grpc_alts': "io.grpc:grpc-alts:${libraries['version.io_grpc']}",
+    'maven.io_grpc_grpc_xds': "io.grpc:grpc-xds:${libraries['version.io_grpc']}",
     'maven.com_google_protobuf': "com.google.protobuf:protobuf-java:${libraries['version.com_google_protobuf']}",
     'maven.com_google_protobuf_java_util': "com.google.protobuf:protobuf-java-util:${libraries['version.com_google_protobuf']}")
 }

gax-grpc/build.gradle

@@ -20,6 +20,8 @@ dependencies {
     libraries['maven.io_grpc_grpc_protobuf'],
     libraries['maven.io_grpc_grpc_stub'])
 
+  runtimeOnly libraries['maven.io_grpc_grpc_xds']
+
   compileOnly libraries['maven.com_google_auto_value_auto_value']
 
   testImplementation( project(':gax').sourceSets.test.output,

gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java

@@ -83,6 +83,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
   static final String DIRECT_PATH_ENV_VAR = "GOOGLE_CLOUD_ENABLE_DIRECT_PATH";
   private static final String DIRECT_PATH_ENV_DISABLE_DIRECT_PATH =
       "GOOGLE_CLOUD_DISABLE_DIRECT_PATH";
+  private static final String DIRECT_PATH_ENV_ENABLE_XDS = "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS";
   static final long DIRECT_PATH_KEEP_ALIVE_TIME_SECONDS = 3600;
   static final long DIRECT_PATH_KEEP_ALIVE_TIMEOUT_SECONDS = 20;
   // reduce the thundering herd problem of too many channels trying to (re)connect at the same time
@@ -330,16 +331,23 @@ private ManagedChannel createSingleChannel() throws IOException, GeneralSecurity
 
     ManagedChannelBuilder<?> builder;
 
-    // TODO(weiranf): Add API in ComputeEngineCredentials to check default service account.
+    // Check DirectPath traffic.
+    boolean isDirectPathXdsEnabled = false;
     if (isDirectPathEnabled(serviceAddress)
         && isNonDefaultServiceAccountAllowed()
         && isOnComputeEngine()) {
-      builder = ComputeEngineChannelBuilder.forAddress(serviceAddress, port);
+      isDirectPathXdsEnabled = Boolean.parseBoolean(envProvider.getenv(DIRECT_PATH_ENV_ENABLE_XDS));
+      if (isDirectPathXdsEnabled) {
+        // google-c2p resolver target must not have a port number
+        builder = ComputeEngineChannelBuilder.forTarget("google-c2p:///" + serviceAddress);
+      } else {
+        builder = ComputeEngineChannelBuilder.forAddress(serviceAddress, port);
+        builder.defaultServiceConfig(directPathServiceConfig);
+      }
       // Set default keepAliveTime and keepAliveTimeout when directpath environment is enabled.
       // Will be overridden by user defined values if any.
       builder.keepAliveTime(DIRECT_PATH_KEEP_ALIVE_TIME_SECONDS, TimeUnit.SECONDS);
       builder.keepAliveTimeout(DIRECT_PATH_KEEP_ALIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
-      builder.defaultServiceConfig(directPathServiceConfig);
     } else {
       ChannelCredentials channelCredentials = createMtlsChannelCredentials();
       if (channelCredentials != null) {
@@ -348,10 +356,13 @@ && isOnComputeEngine()) {
         builder = ManagedChannelBuilder.forAddress(serviceAddress, port);
       }
     }
+    // google-c2p resolver requires service config lookup
+    if (!isDirectPathXdsEnabled) {
+      // See https://github.com/googleapis/gapic-generator/issues/2816
+      builder.disableServiceConfigLookUp();
+    }
     builder =
         builder
-            // See https://github.com/googleapis/gapic-generator/issues/2816
-            .disableServiceConfigLookUp()
             .intercept(new GrpcChannelUUIDInterceptor())
             .intercept(headerInterceptor)
             .intercept(metadataHandlerInterceptor)

gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java

@@ -67,6 +67,7 @@
 
 @RunWith(JUnit4.class)
 public class InstantiatingGrpcChannelProviderTest extends AbstractMtlsTransportChannelTest {
+
   @Test
   public void testEndpoint() {
     String endpoint = "localhost:8080";
@@ -164,11 +165,8 @@ public void testToBuilder() {
     Duration keepaliveTime = Duration.ofSeconds(1);
     Duration keepaliveTimeout = Duration.ofSeconds(2);
     ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> channelConfigurator =
-        new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
-          @Override
-          public ManagedChannelBuilder apply(ManagedChannelBuilder input) {
-            throw new UnsupportedOperationException();
-          }
+        builder -> {
+          throw new UnsupportedOperationException();
         };
     Map<String, ?> directPathServiceConfig = ImmutableMap.of("loadbalancingConfig", "grpclb");
 
@@ -266,16 +264,13 @@ public void testWithGCECredentials() throws IOException {
     executor.shutdown();
 
     ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> channelConfigurator =
-        new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
-          @Override
-          public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
-            if (InstantiatingGrpcChannelProvider.isOnComputeEngine()) {
-              assertThat(channelBuilder instanceof ComputeEngineChannelBuilder).isTrue();
-            } else {
-              assertThat(channelBuilder instanceof ComputeEngineChannelBuilder).isFalse();
-            }
-            return channelBuilder;
+        channelBuilder -> {
+          if (InstantiatingGrpcChannelProvider.isOnComputeEngine()) {
+            assertThat(channelBuilder).isInstanceOf(ComputeEngineChannelBuilder.class);
+          } else {
+            assertThat(channelBuilder).isNotInstanceOf(ComputeEngineChannelBuilder.class);
           }
+          return channelBuilder;
         };
 
     TransportChannelProvider provider =
@@ -304,13 +299,10 @@ public void testWithNonGCECredentials() throws IOException {
     executor.shutdown();
 
     ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> channelConfigurator =
-        new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
-          @Override
-          public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
-            // Clients with non-GCE credentials will not attempt DirectPath.
-            assertThat(channelBuilder instanceof ComputeEngineChannelBuilder).isFalse();
-            return channelBuilder;
-          }
+        channelBuilder -> {
+          // Clients with non-GCE credentials will not attempt DirectPath.
+          assertThat(channelBuilder instanceof ComputeEngineChannelBuilder).isFalse();
+          return channelBuilder;
         };
 
     TransportChannelProvider provider =
@@ -366,13 +358,10 @@ public void testWithNoDirectPathFlagSet() throws IOException {
     executor.shutdown();
 
     ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> channelConfigurator =
-        new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
-          @Override
-          public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
-            // Clients without setting attemptDirectPath flag will not attempt DirectPath
-            assertThat(channelBuilder instanceof ComputeEngineChannelBuilder).isFalse();
-            return channelBuilder;
-          }
+        channelBuilder -> {
+          // Clients without setting attemptDirectPath flag will not attempt DirectPath
+          assertThat(channelBuilder instanceof ComputeEngineChannelBuilder).isFalse();
+          return channelBuilder;
         };
 
     TransportChannelProvider provider =

gax/src/main/java/com/google/api/gax/rpc/StubSettings.java

@@ -176,6 +176,7 @@ public ApiTracerFactory getTracerFactory() {
     return tracerFactory;
   }
 
+  @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
         .add("backgroundExecutorProvider", backgroundExecutorProvider)
@@ -535,6 +536,7 @@ protected static void applyToAllUnaryMethods(
 
     public abstract <B extends StubSettings<B>> StubSettings<B> build() throws IOException;
 
+    @Override
     public String toString() {
       return MoreObjects.toStringHelper(this)
           .add("backgroundExecutorProvider", backgroundExecutorProvider)