xds: add tests & misc fixes based on outstanding items (#6935)

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

@@ -43,7 +43,6 @@ import java.net.UnknownHostException;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
-import java.util.NoSuchElementException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
@@ -72,6 +71,18 @@ public final class XdsClientWrapperForServerSds {
   private final ScheduledExecutorService timeService;
   private final XdsClient.ListenerWatcher listenerWatcher;
 
+  /**
+   * Thrown when no suitable management server was found in the bootstrap file.
+   */
+  public static final class ManagementServerNotFoundException extends Exception {
+
+    private static final long serialVersionUID = 1;
+
+    public ManagementServerNotFoundException(String msg) {
+      super(msg);
+    }
+  }
+
   /**
    * Factory method for creating a {@link XdsClientWrapperForServerSds}.
    *
@@ -80,11 +91,12 @@ public final class XdsClientWrapperForServerSds {
    * @param syncContext {@link SynchronizationContext} needed by {@link XdsClient}.
    */
   public static XdsClientWrapperForServerSds newInstance(
-      int port, Bootstrapper bootstrapper, SynchronizationContext syncContext) throws IOException {
+      int port, Bootstrapper bootstrapper, SynchronizationContext syncContext)
+      throws IOException, ManagementServerNotFoundException {
     Bootstrapper.BootstrapInfo bootstrapInfo = bootstrapper.readBootstrap();
     final List<Bootstrapper.ServerInfo> serverList = bootstrapInfo.getServers();
     if (serverList.isEmpty()) {
-      throw new NoSuchElementException("No management server provided by bootstrap");
+      throw new ManagementServerNotFoundException("No management server provided by bootstrap");
     }
     final Node node = bootstrapInfo.getNode();
     ScheduledExecutorService timeService = SharedResourceHolder.get(timeServiceResource);

xds/src/main/java/io/grpc/xds/internal/sds/SdsProtocolNegotiators.java

@@ -34,6 +34,7 @@ import io.grpc.netty.ProtocolNegotiationEvent;
 import io.grpc.xds.Bootstrapper;
 import io.grpc.xds.XdsAttributes;
 import io.grpc.xds.XdsClientWrapperForServerSds;
+import io.grpc.xds.XdsClientWrapperForServerSds.ManagementServerNotFoundException;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelHandlerContext;
@@ -60,7 +61,7 @@ public final class SdsProtocolNegotiators {
 
   private static final Logger logger = Logger.getLogger(SdsProtocolNegotiators.class.getName());
 
-  private static final AsciiString SCHEME = AsciiString.of("https");
+  private static final AsciiString SCHEME = AsciiString.of("http");
 
   /** Returns a {@link ProtocolNegotiatorFactory} to be used on {@link NettyChannelBuilder}. */
   public static ProtocolNegotiatorFactory clientProtocolNegotiatorFactory() {
@@ -75,13 +76,14 @@ public final class SdsProtocolNegotiators {
    */
   public static ProtocolNegotiator serverProtocolNegotiator(
       int port, SynchronizationContext syncContext) {
-    XdsClientWrapperForServerSds xdsClientWrapperForServerSds =
+    XdsClientWrapperForServerSds xdsClientWrapperForServerSds;
-        ServerSdsProtocolNegotiator.getXdsClientWrapperForServerSds(port, syncContext);
+    try {
-    if (xdsClientWrapperForServerSds == null) {
+      xdsClientWrapperForServerSds = XdsClientWrapperForServerSds.newInstance(
+          port, Bootstrapper.getInstance(), syncContext);
+      return new ServerSdsProtocolNegotiator(xdsClientWrapperForServerSds);
+    } catch (IOException | ManagementServerNotFoundException e) {
       logger.log(Level.INFO, "Fallback to plaintext for server at port {0}", port);
       return InternalProtocolNegotiators.serverPlaintext();
-    } else {
-      return new ServerSdsProtocolNegotiator(xdsClientWrapperForServerSds);
     }
   }
 
@@ -251,17 +253,6 @@ public final class SdsProtocolNegotiators {
           checkNotNull(xdsClientWrapperForServerSds, "xdsClientWrapperForServerSds");
     }
 
-    private static XdsClientWrapperForServerSds getXdsClientWrapperForServerSds(
-        int port, SynchronizationContext syncContext) {
-      try {
-        return XdsClientWrapperForServerSds.newInstance(
-            port, Bootstrapper.getInstance(), syncContext);
-      } catch (IOException e) {
-        logger.log(Level.FINE, "Fallback to plaintext due to exception", e);
-        return null;
-      }
-    }
-
     @Override
     public AsciiString scheme() {
       return SCHEME;

xds/src/main/java/io/grpc/xds/internal/sds/XdsServerBuilder.java

@@ -159,6 +159,7 @@ public final class XdsServerBuilder extends ServerBuilder<XdsServerBuilder> {
               panicMode = true;
             }
           });
+    // TODO(sanjaypujare): move this to start() after creating an XdsServer wrapper
     InternalProtocolNegotiator.ProtocolNegotiator serverProtocolNegotiator =
         SdsProtocolNegotiators.serverProtocolNegotiator(port, syncContext);
     return buildServer(serverProtocolNegotiator);

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

@@ -33,6 +33,7 @@ import io.envoyproxy.envoy.api.v2.auth.UpstreamTlsContext;
 import io.grpc.Attributes;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.NameResolver;
+import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
 import io.grpc.testing.GrpcCleanupRule;
@@ -43,6 +44,7 @@ import io.grpc.xds.internal.sds.CommonTlsContextTestsUtil;
 import io.grpc.xds.internal.sds.SdsProtocolNegotiators;
 import io.grpc.xds.internal.sds.XdsChannelBuilder;
 import io.grpc.xds.internal.sds.XdsServerBuilder;
+import io.netty.handler.ssl.NotSslRecordException;
 import java.io.IOException;
 import java.net.Inet4Address;
 import java.net.InetSocketAddress;
@@ -110,6 +112,44 @@ public class XdsSdsClientServerTest {
     XdsClient.ListenerWatcher unused = performMtlsTestAndGetListenerWatcher(upstreamTlsContext);
   }
 
+  @Test
+  public void tlsServer_plaintextClient_expectException() throws IOException, URISyntaxException {
+    DownstreamTlsContext downstreamTlsContext =
+        CommonTlsContextTestsUtil.buildDownstreamTlsContextFromFilenames(
+            SERVER_1_KEY_FILE, SERVER_1_PEM_FILE, null);
+    buildServerWithTlsContext(downstreamTlsContext);
+
+    SimpleServiceGrpc.SimpleServiceBlockingStub blockingStub =
+        getBlockingStub(/* upstreamTlsContext= */ null, /* overrideAuthority= */ null);
+    try {
+      unaryRpc("buddy", blockingStub);
+      fail("exception expected");
+    } catch (StatusRuntimeException sre) {
+      assertThat(sre.getStatus().getCode()).isEqualTo(Status.UNAVAILABLE.getCode());
+      assertThat(sre.getStatus().getDescription()).contains("Network closed");
+    }
+  }
+
+  @Test
+  public void plaintextServer_tlsClient_expectException() throws IOException, URISyntaxException {
+    buildServerWithTlsContext(/* downstreamTlsContext= */ null);
+
+    // for TLS, client only needs trustCa
+    UpstreamTlsContext upstreamTlsContext =
+        CommonTlsContextTestsUtil.buildUpstreamTlsContextFromFilenames(
+            /* privateKey= */ null, /* certChain= */ null, CA_PEM_FILE);
+
+    SimpleServiceGrpc.SimpleServiceBlockingStub blockingStub =
+        getBlockingStub(upstreamTlsContext, /* overrideAuthority= */ "foo.test.google.fr");
+    try {
+      unaryRpc("buddy", blockingStub);
+      fail("exception expected");
+    } catch (StatusRuntimeException sre) {
+      assertThat(sre).hasCauseThat().isInstanceOf(NotSslRecordException.class);
+      assertThat(sre).hasCauseThat().hasMessageThat().contains("not an SSL/TLS record");
+    }
+  }
+
   /** mTLS - client auth enabled then update server certs to untrusted. */
   @Test
   public void mtlsClientServer_changeServerContext_expectException()

xds/src/test/java/io/grpc/xds/internal/sds/SdsProtocolNegotiatorsTest.java

@@ -288,7 +288,8 @@ public class SdsProtocolNegotiatorsTest {
   public void serverSdsProtocolNegotiator_nullSyncContext_expectPlaintext() {
     InternalProtocolNegotiator.ProtocolNegotiator protocolNegotiator =
         SdsProtocolNegotiators.serverProtocolNegotiator(/* port= */ 7000, /* syncContext= */ null);
-    assertThat(protocolNegotiator.scheme().toString()).isEqualTo("http");
+    assertThat(protocolNegotiator.getClass().getSimpleName())
+        .isEqualTo("ServerPlaintextNegotiator");
   }
 
   private static final class FakeGrpcHttp2ConnectionHandler extends GrpcHttp2ConnectionHandler {