xds: fix URI creation used to instantiate DNS name resolver (#8129)

When creating the URI using Channel authority for instantiating a DNS resolver in the cluster_resolver LB policy, a "dns" scheme needs to be manually attached and the Channel authority would be used as the URI path (same as creating Channel with target). Otherwise, the Channel authority will just be used as the scheme and causing name resolver not found.

The change also handles name resolver lookup more defensively. Although it should not happen, if there does have bug causing DNS resolver not being able to be loaded, the cluster_resolver LB policy propagates the INTERNAL error to upstream.

xds/src/main/java/io/grpc/xds/ClusterResolverLoadBalancer.java

@@ -337,7 +337,7 @@ void shutdown() {
       }
     }
 
-    private class EdsClusterState extends ClusterState implements EdsResourceWatcher {
+    private final class EdsClusterState extends ClusterState implements EdsResourceWatcher {
 
       private EdsClusterState(String name, @Nullable String edsServiceName,
           @Nullable String lrsServerName, @Nullable Long maxConcurrentRequests,
@@ -464,8 +464,10 @@ public void run() {
       }
     }
 
-    private class LogicalDnsClusterState extends ClusterState {
-      private final NameResolver resolver;
+    private final class LogicalDnsClusterState extends ClusterState {
+      private final NameResolver.Factory nameResolverFactory;
+      private final NameResolver.Args nameResolverArgs;
+      private NameResolver resolver;
       @Nullable
       private BackoffPolicy backoffPolicy;
       @Nullable
@@ -474,31 +476,45 @@ private class LogicalDnsClusterState extends ClusterState {
       private LogicalDnsClusterState(String name, @Nullable String lrsServerName,
           @Nullable Long maxConcurrentRequests, @Nullable UpstreamTlsContext tlsContext) {
         super(name, null, lrsServerName, maxConcurrentRequests, tlsContext);
-        NameResolver.Args args = helper.getNameResolverArgs();
-        URI uri;
-        try {
-          uri = new URI(authority);
-        } catch (URISyntaxException e) {
-          // TODO(chengyuanzhang): unlikely to happen, but maybe handle it more gracefully.
-          throw new AssertionError("Bug, invalid authority: " + authority, e);
-        }
-        resolver = helper.getNameResolverRegistry().asFactory().newNameResolver(uri, args);
+        nameResolverFactory =
+            checkNotNull(helper.getNameResolverRegistry().asFactory(), "nameResolverFactory");
+        nameResolverArgs = checkNotNull(helper.getNameResolverArgs(), "nameResolverArgs");
       }
 
       @Override
       void start() {
+        URI uri;
+        try {
+          uri = new URI("dns", "", "/" + authority, null);
+        } catch (URISyntaxException e) {
+          status =
+              Status.INTERNAL.withDescription("Bug, invalid authority: " + authority).withCause(e);
+          handleEndpointResolutionError();
+          return;
+        }
+        resolver = nameResolverFactory.newNameResolver(uri, nameResolverArgs);
+        if (resolver == null) {
+          status = Status.INTERNAL.withDescription("Cannot find DNS resolver");
+          handleEndpointResolutionError();
+          return;
+        }
         resolver.start(new NameResolverListener());
       }
 
       void refresh() {
+        if (resolver == null) {
+          return;
+        }
         cancelBackoff();
         resolver.refresh();
       }
 
       @Override
       void shutdown() {
         super.shutdown();
-        resolver.shutdown();
+        if (resolver != null) {
+          resolver.shutdown();
+        }
         cancelBackoff();
       }
 

xds/src/test/java/io/grpc/xds/ClusterResolverLoadBalancerTest.java

@@ -175,15 +175,14 @@ public void setUp() throws URISyntaxException {
     lbRegistry.register(new FakeLoadBalancerProvider(WEIGHTED_TARGET_POLICY_NAME));
     lbRegistry.register(
         new FakeLoadBalancerProvider("pick_first")); // needed by logical_dns
-    URI targetUri = new URI(AUTHORITY);
     NameResolver.Args args = NameResolver.Args.newBuilder()
         .setDefaultPort(8080)
         .setProxyDetector(GrpcUtil.NOOP_PROXY_DETECTOR)
         .setSynchronizationContext(syncContext)
         .setServiceConfigParser(mock(ServiceConfigParser.class))
         .setChannelLogger(mock(ChannelLogger.class))
         .build();
-    nsRegistry.register(new FakeNameResolverProvider(targetUri));
+    nsRegistry.register(new FakeNameResolverProvider());
     when(helper.getNameResolverRegistry()).thenReturn(nsRegistry);
     when(helper.getNameResolverArgs()).thenReturn(args);
     when(helper.getSynchronizationContext()).thenReturn(syncContext);
@@ -962,25 +961,18 @@ void deliverError(Status error) {
   }
 
   private class FakeNameResolverProvider extends NameResolverProvider {
-    private final URI expectedUri;
-
-    FakeNameResolverProvider(URI expectedUri) {
-      this.expectedUri = expectedUri;
-    }
-
     @Override
     public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
-      if (expectedUri.equals(targetUri)) {
-        FakeNameResolver resolver = new FakeNameResolver(targetUri);
-        resolvers.add(resolver);
-        return resolver;
-      }
-      return null;
+      assertThat(targetUri.getScheme()).isEqualTo("dns");
+      assertThat(targetUri.getPath()).isEqualTo("/" + AUTHORITY);
+      FakeNameResolver resolver = new FakeNameResolver();
+      resolvers.add(resolver);
+      return resolver;
     }
 
     @Override
     public String getDefaultScheme() {
-      return "fake";
+      return "dns";
     }
 
     @Override
@@ -995,17 +987,12 @@ protected int priority() {
   }
 
   private class FakeNameResolver extends NameResolver {
-    private final URI uri;
     private Listener2 listener;
     private int refreshCount;
 
-    FakeNameResolver(URI uri) {
-      this.uri = uri;
-    }
-
     @Override
     public String getServiceAuthority() {
-      return uri.getAuthority();
+      throw new UnsupportedOperationException("should not be called");
     }
 
     @Override
@@ -1025,7 +1012,6 @@ public void shutdown() {
 
     private void deliverEndpointAddresses(List<EquivalentAddressGroup> addresses) {
       listener.onResult(ResolutionResult.newBuilder().setAddresses(addresses).build());
-
     }
 
     private void deliverError(Status error) {