Add API to initialize xDS server

src/python/grpcio/grpc/__init__.py

@@ -2014,7 +2014,8 @@ def server(thread_pool,
            interceptors=None,
            options=None,
            maximum_concurrent_rpcs=None,
-           compression=None):
+           compression=None,
+           xds=False):
     """Creates a Server with which RPCs can be serviced.
 
     Args:
@@ -2035,6 +2036,8 @@ def server(thread_pool,
       compression: An element of grpc.compression, e.g.
         grpc.compression.Gzip. This compression algorithm will be used for the
         lifetime of the server unless overridden. This is an EXPERIMENTAL option.
+      xds: If set to true, retrieves server configuration via xDS. This is an
+        EXPERIMENTAL option.
 
     Returns:
       A Server object.
@@ -2044,7 +2047,8 @@ def server(thread_pool,
                                  () if handlers is None else handlers,
                                  () if interceptors is None else interceptors,
                                  () if options is None else options,
-                                 maximum_concurrent_rpcs, compression)
+                                 maximum_concurrent_rpcs, compression,
+                                 xds)
 
 
 @contextlib.contextmanager

src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi

@@ -397,6 +397,16 @@ cdef extern from "grpc/grpc.h":
   void grpc_server_register_completion_queue(grpc_server *server,
                                              grpc_completion_queue *cq,
                                              void *reserved) nogil
+
+  ctypedef struct grpc_server_config_fetcher:
+    pass
+
+  void grpc_server_set_config_fetcher(
+       grpc_server* server, grpc_server_config_fetcher* config_fetcher) nogil
+
+  grpc_server_config_fetcher* grpc_server_config_fetcher_xds_create() nogil
+
+
   int grpc_server_add_insecure_http2_port(
       grpc_server *server, const char *addr) nogil
   void grpc_server_start(grpc_server *server) nogil

src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi

@@ -15,7 +15,7 @@
 
 cdef class Server:
 
-  def __cinit__(self, object arguments):
+  def __cinit__(self, object arguments, bint xds):
     fork_handlers_and_grpc_init()
     self.references = []
     self.registered_completion_queues = []
@@ -25,6 +25,8 @@ cdef class Server:
     self.c_server = NULL
     cdef _ChannelArgs channel_args = _ChannelArgs(arguments)
     self.c_server = grpc_server_create(channel_args.c_args(), NULL)
+    if xds:
+      grpc_server_set_config_fetcher(self.c_server, grpc_server_config_fetcher_xds_create())
     self.references.append(arguments)
 
   def request_call(

src/python/grpcio/grpc/_server.py

@@ -945,9 +945,9 @@ class _Server(grpc.Server):
 
     # pylint: disable=too-many-arguments
     def __init__(self, thread_pool, generic_handlers, interceptors, options,
-                 maximum_concurrent_rpcs, compression):
+                 maximum_concurrent_rpcs, compression, xds):
         completion_queue = cygrpc.CompletionQueue()
-        server = cygrpc.Server(_augment_options(options, compression))
+        server = cygrpc.Server(_augment_options(options, compression), xds)
         server.register_completion_queue(completion_queue)
         self._state = _ServerState(completion_queue, server, generic_handlers,
                                    _interceptor.service_pipeline(interceptors),
@@ -989,7 +989,7 @@ class _Server(grpc.Server):
 
 
 def create_server(thread_pool, generic_rpc_handlers, interceptors, options,
-                  maximum_concurrent_rpcs, compression):
+                  maximum_concurrent_rpcs, compression, xds):
     _validate_generic_rpc_handlers(generic_rpc_handlers)
     return _Server(thread_pool, generic_rpc_handlers, interceptors, options,
-                   maximum_concurrent_rpcs, compression)
+                   maximum_concurrent_rpcs, compression, xds)

src/python/grpcio_tests/tests/unit/_xds_credentials_test.py

@@ -74,6 +74,17 @@ class XdsCredentialsTest(unittest.TestCase):
             self.assertEqual(response, request)
         server.stop(None)
 
+    def test_start_xds_server(self):
+        server = grpc.server(futures.ThreadPoolExecutor(), xds=True)
+        server.add_generic_rpc_handlers((_GenericHandler(),))
+        server_fallback_creds = grpc.insecure_server_credentials()
+        server_creds = grpc.xds_server_credentials(server_fallback_creds)
+        port = server.add_secure_port("localhost:0", server_creds)
+        server.start()
+        server.stop(None)
+        # No exceptions thrown. A more comprehensive suite of tests will be
+        # provided by the interop tests.
+
 if __name__ == "__main__":
     logging.basicConfig()
     unittest.main()