xds,googleapis: Allow wrapping NameResolver to inject XdsClient

Since there is no longer a single global XdsClient, it makes more sense
to allow things like the c2p name resolver to inject its own bootstrap
even if there is one defined in an environment variable.
GoogleCloudToProdNameResolver can now pass an XdsClient instance to
XdsNameResolver, and SharedXdsClientPoolProvider allows
GoogleCloudToProdNameResolver to choose the bootstrap for that one
specific target.

Since XdsNameResolver is no longer in control of the XdsClient pool the
XdsClient instance is now passed to ClusterImplLb. A channel will now
only access the global XdsClient pool exactly once: in the name
resolver.

BootstrapInfo is purposefully being shared across channels, as we really
want to share things like credentials which can have significant memory
use and may have caches which reduce I/O when shared. That is why
SharedXdsClientPoolProvider receives BootstrapInfo instead of
Map<String,?>.

Verifying BootstrapInfo.server() is not empty was moved from
SharedXdsClientPoolProvider to GrpcBootstrapperImpl so avoid
getOrCreate() throwing an exception in only that one case. It might make
sense to move that to BootstrapperImpl, but that will need more
investigation.

A lot of tests needed updating because XdsClientPoolProvider is no
longer responsible for parsing the bootstrap, so we now need bootstraps
even if XdsClientPoolProvider will ignore it.

This also fixes a bug in GoogleCloudToProdNameResolver where it would
initialize the delegate even when it failed to create the bootstrap.
That would certainly cause all RPCs on the channel to fail because of
the missing bootstrap and it defeated the point of `succeeded == false`
and `refresh()` which was supposed to retry contacting the metadata
server.

The server tests were enhanced to give a useful error when
server.start() throws an exception, as otherwise the real error is lost.

Author: ejona86

googleapis/src/main/java/io/grpc/googleapis/GoogleCloudToProdNameResolver.java

@@ -20,10 +20,11 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.CharStreams;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import io.grpc.MetricRecorder;
 import io.grpc.NameResolver;
 import io.grpc.NameResolverRegistry;
 import io.grpc.Status;
@@ -32,6 +33,13 @@
 import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.SharedResourceHolder;
 import io.grpc.internal.SharedResourceHolder.Resource;
+import io.grpc.xds.InternalGrpcBootstrapperImpl;
+import io.grpc.xds.InternalSharedXdsClientPoolProvider;
+import io.grpc.xds.InternalSharedXdsClientPoolProvider.XdsClientResult;
+import io.grpc.xds.XdsNameResolverProvider;
+import io.grpc.xds.client.Bootstrapper.BootstrapInfo;
+import io.grpc.xds.client.XdsClient;
+import io.grpc.xds.client.XdsInitializationException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
@@ -41,7 +49,6 @@
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
-import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.Executor;
 import java.util.logging.Level;
@@ -63,52 +70,53 @@ final class GoogleCloudToProdNameResolver extends NameResolver {
   static final String C2P_AUTHORITY = "traffic-director-c2p.xds.googleapis.com";
   @VisibleForTesting
   static boolean isOnGcp = InternalCheckGcpEnvironment.isOnGcp();
-  @VisibleForTesting
-  static boolean xdsBootstrapProvided =
-      System.getenv("GRPC_XDS_BOOTSTRAP") != null
-          || System.getProperty("io.grpc.xds.bootstrap") != null
-          || System.getenv("GRPC_XDS_BOOTSTRAP_CONFIG") != null
-          || System.getProperty("io.grpc.xds.bootstrapConfig") != null;
-  @VisibleForTesting
-  static boolean enableFederation =
-      Strings.isNullOrEmpty(System.getenv("GRPC_EXPERIMENTAL_XDS_FEDERATION"))
-          || Boolean.parseBoolean(System.getenv("GRPC_EXPERIMENTAL_XDS_FEDERATION"));
 
   private static final String serverUriOverride =
       System.getenv("GRPC_TEST_ONLY_GOOGLE_C2P_RESOLVER_TRAFFIC_DIRECTOR_URI");
 
-  private HttpConnectionProvider httpConnectionProvider = HttpConnectionFactory.INSTANCE;
+  @GuardedBy("GoogleCloudToProdNameResolver.class")
+  private static BootstrapInfo bootstrapInfo;
+  private static HttpConnectionProvider httpConnectionProvider = HttpConnectionFactory.INSTANCE;
+  private static int c2pId = new Random().nextInt();
+
+  private static synchronized BootstrapInfo getBootstrapInfo()
+      throws XdsInitializationException, IOException {
+    if (bootstrapInfo != null) {
+      return bootstrapInfo;
+    }
+    BootstrapInfo bootstrapInfoTmp =
+        InternalGrpcBootstrapperImpl.parseBootstrap(generateBootstrap());
+    // Avoid setting global when testing
+    if (httpConnectionProvider == HttpConnectionFactory.INSTANCE) {
+      bootstrapInfo = bootstrapInfoTmp;
+    }
+    return bootstrapInfoTmp;
+  }
+
   private final String authority;
   private final SynchronizationContext syncContext;
   private final Resource<Executor> executorResource;
-  private final BootstrapSetter bootstrapSetter;
+  private final String target;
+  private final MetricRecorder metricRecorder;
   private final NameResolver delegate;
-  private final Random rand;
   private final boolean usingExecutorResource;
-  // It's not possible to use both PSM and DirectPath C2P in the same application.
-  // Delegate to DNS if user-provided bootstrap is found.
-  private final String schemeOverride =
-      !isOnGcp
-      || (xdsBootstrapProvided && !enableFederation)
-      ? "dns" : "xds";
+  private final String schemeOverride = !isOnGcp ? "dns" : "xds";
+  private XdsClientResult xdsClientPool;
+  private XdsClient xdsClient;
   private Executor executor;
-  private Listener2 listener;
   private boolean succeeded;
   private boolean resolving;
   private boolean shutdown;
 
-  GoogleCloudToProdNameResolver(URI targetUri, Args args, Resource<Executor> executorResource,
-      BootstrapSetter bootstrapSetter) {
-    this(targetUri, args, executorResource, new Random(), bootstrapSetter,
+  GoogleCloudToProdNameResolver(URI targetUri, Args args, Resource<Executor> executorResource) {
+    this(targetUri, args, executorResource,
         NameResolverRegistry.getDefaultRegistry().asFactory());
   }
 
   @VisibleForTesting
   GoogleCloudToProdNameResolver(URI targetUri, Args args, Resource<Executor> executorResource,
-      Random rand, BootstrapSetter bootstrapSetter, NameResolver.Factory nameResolverFactory) {
+      NameResolver.Factory nameResolverFactory) {
     this.executorResource = checkNotNull(executorResource, "executorResource");
-    this.bootstrapSetter = checkNotNull(bootstrapSetter, "bootstrapSetter");
-    this.rand = checkNotNull(rand, "rand");
     String targetPath = checkNotNull(checkNotNull(targetUri, "targetUri").getPath(), "targetPath");
     Preconditions.checkArgument(
         targetPath.startsWith("/"),
@@ -118,9 +126,14 @@ final class GoogleCloudToProdNameResolver extends NameResolver {
     authority = GrpcUtil.checkAuthority(targetPath.substring(1));
     syncContext = checkNotNull(args, "args").getSynchronizationContext();
     targetUri = overrideUriScheme(targetUri, schemeOverride);
-    if (schemeOverride.equals("xds") && enableFederation) {
+    if (schemeOverride.equals("xds")) {
       targetUri = overrideUriAuthority(targetUri, C2P_AUTHORITY);
+      args = args.toBuilder()
+          .setArg(XdsNameResolverProvider.XDS_CLIENT_SUPPLIER, () -> xdsClient)
+          .build();
     }
+    target = targetUri.toString();
+    metricRecorder = args.getMetricRecorder();
     delegate = checkNotNull(nameResolverFactory, "nameResolverFactory").newNameResolver(
         targetUri, args);
     executor = args.getOffloadExecutor();
@@ -150,7 +163,7 @@ private void resolve() {
 
     resolving = true;
     if (logger.isLoggable(Level.FINE)) {
-      logger.fine("resolve with schemaOverride = " + schemeOverride);
+      logger.log(Level.FINE, "start with schemaOverride = {0}", schemeOverride);
     }
 
     if (schemeOverride.equals("dns")) {
@@ -168,28 +181,28 @@ private void resolve() {
     class Resolve implements Runnable {
       @Override
       public void run() {
-        ImmutableMap<String, ?> rawBootstrap = null;
+        BootstrapInfo bootstrapInfo = null;
         try {
-          // User provided bootstrap configs are only supported with federation. If federation is
-          // not enabled or there is no user provided config, we set a custom bootstrap override.
-          // Otherwise, we don't set the override, which will allow a user provided bootstrap config
-          // to take effect.
-          if (!enableFederation || !xdsBootstrapProvided) {
-            rawBootstrap = generateBootstrap(queryZoneMetadata(METADATA_URL_ZONE),
-                queryIpv6SupportMetadata(METADATA_URL_SUPPORT_IPV6));
-          }
+          bootstrapInfo = getBootstrapInfo();
         } catch (IOException e) {
           listener.onError(
               Status.INTERNAL.withDescription("Unable to get metadata").withCause(e));
+        } catch (XdsInitializationException e) {
+          listener.onError(
+              Status.INTERNAL.withDescription("Unable to create c2p bootstrap").withCause(e));
+        } catch (Throwable t) {
+          listener.onError(
+              Status.INTERNAL.withDescription("Unexpected error creating c2p bootstrap")
+              .withCause(t));
         } finally {
-          final ImmutableMap<String, ?> finalRawBootstrap = rawBootstrap;
+          final BootstrapInfo finalBootstrapInfo = bootstrapInfo;
           syncContext.execute(new Runnable() {
             @Override
             public void run() {
-              if (!shutdown) {
-                if (finalRawBootstrap != null) {
-                  bootstrapSetter.setBootstrap(finalRawBootstrap);
-                }
+              if (!shutdown && finalBootstrapInfo != null) {
+                xdsClientPool = InternalSharedXdsClientPoolProvider.getOrCreate(
+                    target, finalBootstrapInfo, metricRecorder, null);
+                xdsClient = xdsClientPool.getObject();
                 delegate.start(listener);
                 succeeded = true;
               }
@@ -203,9 +216,16 @@ public void run() {
     executor.execute(new Resolve());
   }
 
-  private ImmutableMap<String, ?> generateBootstrap(String zone, boolean supportIpv6) {
+  @VisibleForTesting
+  static ImmutableMap<String, ?> generateBootstrap() throws IOException {
+    return generateBootstrap(
+        queryZoneMetadata(METADATA_URL_ZONE),
+        queryIpv6SupportMetadata(METADATA_URL_SUPPORT_IPV6));
+  }
+
+  private static ImmutableMap<String, ?> generateBootstrap(String zone, boolean supportIpv6) {
     ImmutableMap.Builder<String, Object> nodeBuilder = ImmutableMap.builder();
-    nodeBuilder.put("id", "C2P-" + (rand.nextInt() & Integer.MAX_VALUE));
+    nodeBuilder.put("id", "C2P-" + (c2pId & Integer.MAX_VALUE));
     if (!zone.isEmpty()) {
       nodeBuilder.put("locality", ImmutableMap.of("zone", zone));
     }
@@ -250,12 +270,15 @@ public void shutdown() {
     if (delegate != null) {
       delegate.shutdown();
     }
+    if (xdsClient != null) {
+      xdsClient = xdsClientPool.returnObject(xdsClient);
+    }
     if (executor != null && usingExecutorResource) {
       executor = SharedResourceHolder.release(executorResource, executor);
     }
   }
 
-  private String queryZoneMetadata(String url) throws IOException {
+  private static String queryZoneMetadata(String url) throws IOException {
     HttpURLConnection con = null;
     String respBody;
     try {
@@ -275,7 +298,7 @@ private String queryZoneMetadata(String url) throws IOException {
     return index == -1 ? "" : respBody.substring(index + 1);
   }
 
-  private boolean queryIpv6SupportMetadata(String url) throws IOException {
+  private static boolean queryIpv6SupportMetadata(String url) throws IOException {
     HttpURLConnection con = null;
     try {
       con = httpConnectionProvider.createConnection(url);
@@ -294,8 +317,17 @@ private boolean queryIpv6SupportMetadata(String url) throws IOException {
   }
 
   @VisibleForTesting
-  void setHttpConnectionProvider(HttpConnectionProvider httpConnectionProvider) {
-    this.httpConnectionProvider = httpConnectionProvider;
+  static void setHttpConnectionProvider(HttpConnectionProvider httpConnectionProvider) {
+    if (httpConnectionProvider == null) {
+      GoogleCloudToProdNameResolver.httpConnectionProvider = HttpConnectionFactory.INSTANCE;
+    } else {
+      GoogleCloudToProdNameResolver.httpConnectionProvider = httpConnectionProvider;
+    }
+  }
+
+  @VisibleForTesting
+  static void setC2pId(int c2pId) {
+    GoogleCloudToProdNameResolver.c2pId = c2pId;
   }
 
   private static URI overrideUriScheme(URI uri, String scheme) {
@@ -335,8 +367,4 @@ public HttpURLConnection createConnection(String url) throws IOException {
   interface HttpConnectionProvider {
     HttpURLConnection createConnection(String url) throws IOException;
   }
-
-  public interface BootstrapSetter {
-    void setBootstrap(Map<String, ?> bootstrap);
-  }
 }

googleapis/src/main/java/io/grpc/googleapis/GoogleCloudToProdNameResolverProvider.java

@@ -22,13 +22,11 @@
 import io.grpc.NameResolver.Args;
 import io.grpc.NameResolverProvider;
 import io.grpc.internal.GrpcUtil;
-import io.grpc.xds.InternalSharedXdsClientPoolProvider;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.net.URI;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Map;
 
 /**
  * A provider for {@link GoogleCloudToProdNameResolver}.
@@ -52,8 +50,7 @@ public GoogleCloudToProdNameResolverProvider() {
   public NameResolver newNameResolver(URI targetUri, Args args) {
     if (scheme.equals(targetUri.getScheme())) {
       return new GoogleCloudToProdNameResolver(
-          targetUri, args, GrpcUtil.SHARED_CHANNEL_EXECUTOR,
-          new SharedXdsClientPoolProviderBootstrapSetter());
+          targetUri, args, GrpcUtil.SHARED_CHANNEL_EXECUTOR);
     }
     return null;
   }
@@ -77,12 +74,4 @@ protected int priority() {
   public Collection<Class<? extends SocketAddress>> getProducedSocketAddressTypes() {
     return Collections.singleton(InetSocketAddress.class);
   }
-
-  private static final class SharedXdsClientPoolProviderBootstrapSetter
-      implements GoogleCloudToProdNameResolver.BootstrapSetter {
-    @Override
-    public void setBootstrap(Map<String, ?> bootstrap) {
-      InternalSharedXdsClientPoolProvider.setDefaultProviderBootstrapOverride(bootstrap);
-    }
-  }
 }