xds: eliminate downstreamTlsContext from XdsServerBuilder

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

@@ -70,6 +70,7 @@ public final class XdsClientWrapperForServerSds {
   @Nullable private final XdsClient xdsClient;
   private final int port;
   private final ScheduledExecutorService timeService;
+  private final XdsClient.ListenerWatcher listenerWatcher;
 
   /**
    * Factory method for creating a {@link XdsClientWrapperForServerSds}.
@@ -106,15 +107,14 @@ public static XdsClientWrapperForServerSds newInstance(
     this.port = port;
     this.xdsClient = xdsClient;
     this.timeService = timeService;
-    xdsClient.watchListenerData(
-        port,
+    this.listenerWatcher =
         new XdsClient.ListenerWatcher() {
           @Override
           public void onListenerChanged(XdsClient.ListenerUpdate update) {
             logger.log(
                 Level.INFO,
-                "Setting myListener from ConfigUpdate listener :{0}",
-                update.getListener().toString());
+                "Setting myListener from ConfigUpdate listener: {0}",
+                update.getListener());
             curListener = update.getListener();
           }
 
@@ -126,9 +126,10 @@ public void onError(Status error) {
               curListener = null;
             }
             // TODO(sanjaypujare): Implement logic for other cases based on final design.
-            logger.log(Level.SEVERE, "ListenerWatcher in XdsClientWrapperForServerSds:{0}", error);
+            logger.log(Level.SEVERE, "ListenerWatcher in XdsClientWrapperForServerSds: {0}", error);
           }
-        });
+        };
+    xdsClient.watchListenerData(port, listenerWatcher);
   }
 
   /**
@@ -157,6 +158,11 @@ public DownstreamTlsContext getDownstreamTlsContext(Channel channel) {
     return null;
   }
 
+  @VisibleForTesting
+  XdsClient.ListenerWatcher getListenerWatcher() {
+    return listenerWatcher;
+  }
+
   private static final class FilterChainComparator implements Comparator<FilterChain> {
     private final InetSocketAddress localAddress;
 

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

@@ -16,7 +16,6 @@
 
 package io.grpc.xds.internal.sds;
 
-import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -74,22 +73,19 @@ public static ProtocolNegotiatorFactory clientProtocolNegotiatorFactory(
 
   /**
    * Creates an SDS based {@link ProtocolNegotiator} for a {@link io.grpc.netty.NettyServerBuilder}.
-   * Passing {@code null} for downstreamTlsContext will fall back to plaintext.
    * If xDS returns no DownstreamTlsContext, it will fall back to plaintext.
    *
-   * @param downstreamTlsContext passed in {@link XdsServerBuilder#tlsContext}.
    * @param port the listening port passed to {@link XdsServerBuilder#forPort(int)}.
    */
   public static ProtocolNegotiator serverProtocolNegotiator(
-      @Nullable DownstreamTlsContext downstreamTlsContext, int port,
-      SynchronizationContext syncContext) {
+      int port, SynchronizationContext syncContext) {
     XdsClientWrapperForServerSds xdsClientWrapperForServerSds =
         ServerSdsProtocolNegotiator.getXdsClientWrapperForServerSds(port, syncContext);
-    if (xdsClientWrapperForServerSds == null && downstreamTlsContext == null) {
+    if (xdsClientWrapperForServerSds == null) {
       logger.log(Level.INFO, "Fallback to plaintext for server at port {0}", port);
       return InternalProtocolNegotiators.serverPlaintext();
     } else {
-      return new ServerSdsProtocolNegotiator(downstreamTlsContext, xdsClientWrapperForServerSds);
+      return new ServerSdsProtocolNegotiator(xdsClientWrapperForServerSds);
     }
   }
 
@@ -267,18 +263,13 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
   @VisibleForTesting
   public static final class ServerSdsProtocolNegotiator implements ProtocolNegotiator {
 
-    @Nullable private final DownstreamTlsContext downstreamTlsContext;
-    @Nullable private final XdsClientWrapperForServerSds xdsClientWrapperForServerSds;
+    private final XdsClientWrapperForServerSds xdsClientWrapperForServerSds;
 
     /** Constructor. */
     @VisibleForTesting
-    public ServerSdsProtocolNegotiator(
-        @Nullable DownstreamTlsContext downstreamTlsContext,
-        @Nullable XdsClientWrapperForServerSds xdsClientWrapperForServerSds) {
-      checkArgument(downstreamTlsContext != null || xdsClientWrapperForServerSds != null,
-          "both downstreamTlsContext and xdsClientWrapperForServerSds cannot be null");
-      this.downstreamTlsContext = downstreamTlsContext;
-      this.xdsClientWrapperForServerSds = xdsClientWrapperForServerSds;
+    public ServerSdsProtocolNegotiator(XdsClientWrapperForServerSds xdsClientWrapperForServerSds) {
+      this.xdsClientWrapperForServerSds =
+          checkNotNull(xdsClientWrapperForServerSds, "xdsClientWrapperForServerSds");
     }
 
     private static XdsClientWrapperForServerSds getXdsClientWrapperForServerSds(
@@ -299,8 +290,7 @@ public AsciiString scheme() {
 
     @Override
     public ChannelHandler newHandler(GrpcHttp2ConnectionHandler grpcHandler) {
-      return new HandlerPickerHandler(grpcHandler, downstreamTlsContext,
-          xdsClientWrapperForServerSds);
+      return new HandlerPickerHandler(grpcHandler, xdsClientWrapperForServerSds);
     }
 
     @Override
@@ -315,16 +305,13 @@ public void close() {
   static final class HandlerPickerHandler
       extends ChannelInboundHandlerAdapter {
     private final GrpcHttp2ConnectionHandler grpcHandler;
-    private final DownstreamTlsContext downstreamTlsContextFromBuilder;
     private final XdsClientWrapperForServerSds xdsClientWrapperForServerSds;
 
     HandlerPickerHandler(
         GrpcHttp2ConnectionHandler grpcHandler,
-        @Nullable DownstreamTlsContext downstreamTlsContext,
         @Nullable XdsClientWrapperForServerSds xdsClientWrapperForServerSds) {
       checkNotNull(grpcHandler, "grpcHandler");
       this.grpcHandler = grpcHandler;
-      this.downstreamTlsContextFromBuilder = downstreamTlsContext;
       this.xdsClientWrapperForServerSds = xdsClientWrapperForServerSds;
     }
 
@@ -339,9 +326,6 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
             xdsClientWrapperForServerSds == null
                 ? null
                 : xdsClientWrapperForServerSds.getDownstreamTlsContext(ctx.channel());
-        if (isTlsContextEmpty(downstreamTlsContext)) {
-          downstreamTlsContext = downstreamTlsContextFromBuilder;
-        }
         if (isTlsContextEmpty(downstreamTlsContext)) {
           logger.log(Level.INFO, "Fallback to plaintext for {0}", ctx.channel().localAddress());
           ctx.pipeline()

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

@@ -17,7 +17,6 @@
 package io.grpc.xds.internal.sds;
 
 import com.google.common.annotations.VisibleForTesting;
-import io.envoyproxy.envoy.api.v2.auth.DownstreamTlsContext;
 import io.grpc.BindableService;
 import io.grpc.CompressorRegistry;
 import io.grpc.DecompressorRegistry;
@@ -51,9 +50,6 @@ public final class XdsServerBuilder extends ServerBuilder<XdsServerBuilder> {
   private final NettyServerBuilder delegate;
   private final int port;
 
-  // TODO (sanjaypujare) integrate with xDS client to get downstreamTlsContext from LDS
-  @Nullable private DownstreamTlsContext downstreamTlsContext;
-
   private XdsServerBuilder(NettyServerBuilder nettyDelegate, int port) {
     this.delegate = nettyDelegate;
     this.port = port;
@@ -130,15 +126,6 @@ public XdsServerBuilder intercept(ServerInterceptor interceptor) {
     return this;
   }
 
-  /**
-   * Set the DownstreamTlsContext for the server. This is a temporary workaround until integration
-   * with xDS client is implemented to get LDS. Passing {@code null} will fall back to plaintext.
-   */
-  public XdsServerBuilder tlsContext(@Nullable DownstreamTlsContext downstreamTlsContext) {
-    this.downstreamTlsContext = downstreamTlsContext;
-    return this;
-  }
-
   /** Creates a gRPC server builder for the given port. */
   public static XdsServerBuilder forPort(int port) {
     NettyServerBuilder nettyDelegate = NettyServerBuilder.forAddress(new InetSocketAddress(port));
@@ -173,8 +160,7 @@ void panic(final Throwable t) {
             }
           });
     InternalProtocolNegotiator.ProtocolNegotiator serverProtocolNegotiator =
-        SdsProtocolNegotiators.serverProtocolNegotiator(
-            this.downstreamTlsContext, port, syncContext);
+        SdsProtocolNegotiators.serverProtocolNegotiator(port, syncContext);
     return buildServer(serverProtocolNegotiator);
   }
 

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

@@ -20,6 +20,11 @@
 import static io.grpc.ConnectivityState.CONNECTING;
 import static io.grpc.ConnectivityState.TRANSIENT_FAILURE;
 import static io.grpc.xds.XdsLbPolicies.EDS_POLICY_NAME;
+import static io.grpc.xds.internal.sds.CommonTlsContextTestsUtil.BAD_CLIENT_KEY_FILE;
+import static io.grpc.xds.internal.sds.CommonTlsContextTestsUtil.BAD_CLIENT_PEM_FILE;
+import static io.grpc.xds.internal.sds.CommonTlsContextTestsUtil.CA_PEM_FILE;
+import static io.grpc.xds.internal.sds.CommonTlsContextTestsUtil.CLIENT_KEY_FILE;
+import static io.grpc.xds.internal.sds.CommonTlsContextTestsUtil.CLIENT_PEM_FILE;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.same;
@@ -56,7 +61,7 @@
 import io.grpc.xds.XdsClient.RefCountedXdsClientObjectPool;
 import io.grpc.xds.XdsClient.XdsClientFactory;
 import io.grpc.xds.XdsLoadBalancerProvider.XdsConfig;
-import io.grpc.xds.internal.sds.SecretVolumeSslContextProviderTest;
+import io.grpc.xds.internal.sds.CommonTlsContextTestsUtil;
 import io.grpc.xds.internal.sds.SslContextProvider;
 import io.grpc.xds.internal.sds.TlsContextManager;
 import java.net.InetSocketAddress;
@@ -79,11 +84,6 @@
  */
 @RunWith(JUnit4.class)
 public class CdsLoadBalancerTest {
-  private static final String CLIENT_PEM_FILE = "client.pem";
-  private static final String CLIENT_KEY_FILE = "client.key";
-  private static final String BADCLIENT_PEM_FILE = "badclient.pem";
-  private static final String BADCLIENT_KEY_FILE = "badclient.key";
-  private static final String CA_PEM_FILE = "ca.pem";
 
   private final RefCountedXdsClientObjectPool xdsClientPool = new RefCountedXdsClientObjectPool(
       new XdsClientFactory() {
@@ -356,7 +356,7 @@ public void handleCdsConfigs_withUpstreamTlsContext()  {
     verify(xdsClient).watchClusterData(eq("foo.googleapis.com"), clusterWatcherCaptor1.capture());
 
     UpstreamTlsContext upstreamTlsContext =
-        SecretVolumeSslContextProviderTest.buildUpstreamTlsContextFromFilenames(
+        CommonTlsContextTestsUtil.buildUpstreamTlsContextFromFilenames(
             CLIENT_KEY_FILE, CLIENT_PEM_FILE, CA_PEM_FILE);
 
     SslContextProvider<UpstreamTlsContext> mockSslContextProvider =
@@ -415,8 +415,8 @@ public void handleCdsConfigs_withUpstreamTlsContext()  {
     reset(mockTlsContextManager);
     reset(helper);
     UpstreamTlsContext upstreamTlsContext1 =
-        SecretVolumeSslContextProviderTest.buildUpstreamTlsContextFromFilenames(
-            BADCLIENT_KEY_FILE, BADCLIENT_PEM_FILE, CA_PEM_FILE);
+        CommonTlsContextTestsUtil.buildUpstreamTlsContextFromFilenames(
+            BAD_CLIENT_KEY_FILE, BAD_CLIENT_PEM_FILE, CA_PEM_FILE);
     SslContextProvider<UpstreamTlsContext> mockSslContextProvider1 =
         (SslContextProvider<UpstreamTlsContext>) mock(SslContextProvider.class);
     doReturn(upstreamTlsContext1).when(mockSslContextProvider1).getSource();

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

@@ -18,6 +18,7 @@
 
 import static com.google.common.truth.Truth.assertThat;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -139,6 +140,26 @@ public static Iterable<Object[]> data() {
   private XdsClientWrapperForServerSds xdsClientWrapperForServerSds;
   private final DownstreamTlsContext[] tlsContexts = new DownstreamTlsContext[3];
 
+  /** Creates XdsClientWrapperForServerSds: also used by other classes. */
+  public static XdsClientWrapperForServerSds createXdsClientWrapperForServerSds(
+      int port, DownstreamTlsContext downstreamTlsContext) {
+    XdsClient mockXdsClient = mock(XdsClient.class);
+    XdsClientWrapperForServerSds xdsClientWrapperForServerSds =
+        new XdsClientWrapperForServerSds(port, mockXdsClient, null);
+    generateListenerUpdateToWatcher(
+        port, downstreamTlsContext, xdsClientWrapperForServerSds.getListenerWatcher());
+    return xdsClientWrapperForServerSds;
+  }
+
+  static void generateListenerUpdateToWatcher(
+      int port, DownstreamTlsContext tlsContext, XdsClient.ListenerWatcher registeredWatcher) {
+    EnvoyServerProtoData.Listener listener =
+        XdsSdsClientServerTest.buildListener("listener1", "0.0.0.0", port, tlsContext);
+    XdsClient.ListenerUpdate listenerUpdate =
+        XdsClient.ListenerUpdate.newBuilder().setListener(listener).build();
+    registeredWatcher.onListenerChanged(listenerUpdate);
+  }
+
   @Before
   public void setUp() throws IOException {
     MockitoAnnotations.initMocks(this);