grpc/xds: add test utility crate

Cargo.toml

@@ -21,6 +21,7 @@ members = [
   "xds-client",
   "xds-client-opentelemetry",
   "tonic-xds",
+  "xds-test-util",
   "interop",
   "tests/disable_comments",
   "tests/wellknown",

tonic-xds/Cargo.toml

@@ -63,6 +63,7 @@ workspace = true
 
 [dev-dependencies]
 xds-client = { version = "0.1.0-alpha.2", path = "../xds-client", features = ["test-util"] }
+xds-test-util = { path = "../xds-test-util" }
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "test-util"] }
 tonic = { version = "0.14", features = [ "server", "channel", "tls-ring" ] }
 tonic-prost = "0.14"

tonic-xds/src/client/mod.rs

@@ -7,3 +7,6 @@ pub(crate) mod loadbalance;
 #[allow(dead_code)]
 pub(crate) mod retry;
 pub(crate) mod route;
+
+#[cfg(test)]
+mod xds_e2e;

tonic-xds/src/client/xds_e2e.rs

@@ -0,0 +1,288 @@
+//! End-to-end tests: a real xDS channel resolving through a fake ADS control
+//! plane to live gRPC backends.
+//!
+//! Unlike the unit tests in [`super::channel`], which inject endpoints through
+//! `build_grpc_channel_from_parts`, these tests exercise the full pipeline:
+//! bootstrap loading, the ADS transport, and LDS -> RDS -> CDS -> EDS resolution
+//! served by [`XdsTestControlPlaneService`]. Traffic is routed to greeter (echo)
+//! backends spawned by the tests.
+mod test {
+    use std::collections::HashMap;
+    use std::net::SocketAddr;
+    use xds_test_util::{RunningControlPlane, XdsTestControlPlaneService, config};
+
+    use crate::testutil::grpc::{GreeterClient, HelloRequest, spawn_greeter_server};
+    use crate::{BootstrapConfig, XdsChannelBuilder, XdsChannelConfig, XdsChannelGrpc, XdsUri};
+    use tokio::time::Duration;
+
+    /// Starts the fake ADS control plane on an ephemeral port. Returns the running
+    /// control plane (which injects config and shuts down on drop) and its address.
+    async fn start_control_plane() -> (RunningControlPlane, SocketAddr) {
+        let running = XdsTestControlPlaneService::new()
+            .start()
+            .await
+            .expect("start control plane");
+        let addr = running.addr();
+        (running, addr)
+    }
+
+    /// Builds a real xDS channel whose bootstrap points at `cp_addr` and whose
+    /// target resolves the listener `listener_name`.
+    fn build_channel(cp_addr: SocketAddr, listener_name: &str) -> XdsChannelGrpc {
+        let bootstrap_json = format!(
+            r#"{{"xds_servers":[{{"server_uri":"http://{cp_addr}"}}],"node":{{"id":"test"}}}}"#
+        );
+        let bootstrap = BootstrapConfig::from_json(&bootstrap_json).expect("parse bootstrap");
+        let target = XdsUri::parse(&format!("xds:///{listener_name}")).expect("parse target");
+        XdsChannelBuilder::new(XdsChannelConfig::new(target).with_bootstrap(bootstrap))
+            .build_grpc_channel()
+            .expect("build xds channel")
+    }
+
+    /// Sends `say_hello` in a loop until a reply starting with `want_prefix` is
+    /// observed (xDS resolution and config updates are asynchronous), returning that
+    /// reply. Panics if it never arrives.
+    async fn say_hello_until_prefix(
+        client: &mut GreeterClient<XdsChannelGrpc>,
+        want_prefix: &str,
+    ) -> String {
+        const RETRIES: usize = 100;
+        let mut last = None;
+        for _ in 0..RETRIES {
+            match client
+                .say_hello(HelloRequest {
+                    name: "world".to_string(),
+                })
+                .await
+            {
+                Ok(response) => {
+                    let message = response.into_inner().message;
+                    if message.starts_with(want_prefix) {
+                        return message;
+                    }
+                    last = Some(Ok(message));
+                }
+                Err(status) => last = Some(Err(status)),
+            }
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
+        panic!("never observed reply starting with {want_prefix:?}; last seen: {last:?}");
+    }
+
+    /// End-to-end test: verifies that an xDS channel routes traffic to the correct backend.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn xds_channel_e2e_routes_to_backend() {
+        let backend = spawn_greeter_server("backend", None, None)
+            .await
+            .expect("spawn greeter backend");
+        let backend_addr = backend.addr;
+
+        let (control_plane, cp_addr) = start_control_plane().await;
+
+        // Configure LDS (inline route) -> CDS -> EDS pointing at the backend.
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Lds,
+            HashMap::from([(
+                "my-service".to_string(),
+                config::build_inline_listener("my-service", "my-cluster"),
+            )]),
+        );
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Cds,
+            HashMap::from([(
+                "my-cluster".to_string(),
+                config::build_cluster("my-cluster"),
+            )]),
+        );
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Eds,
+            HashMap::from([(
+                "my-cluster".to_string(),
+                config::build_cla(
+                    "my-cluster",
+                    &[(backend_addr.ip().to_string(), backend_addr.port())],
+                ),
+            )]),
+        );
+
+        let mut client = GreeterClient::new(build_channel(cp_addr, "my-service"));
+        let reply = say_hello_until_prefix(&mut client, "backend:").await;
+        assert_eq!(reply, "backend: world");
+
+        // The control plane observed exactly one ADS stream from the client.
+        let counts = control_plane.get_service().get_subscriber_counts();
+        assert_eq!(counts.get(&config::AdsTypeUrl::Lds), Some(&1));
+        assert_eq!(counts.get(&config::AdsTypeUrl::Cds), Some(&1));
+        assert_eq!(counts.get(&config::AdsTypeUrl::Eds), Some(&1));
+
+        let _ = backend.shutdown.send(());
+    }
+
+    /// The client is initially routing to
+    /// one cluster, update the RDS route to point at a different cluster and assert
+    /// traffic shifts to the new backend — exercising the control plane pushing a
+    /// live update to a connected client.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn xds_channel_e2e_route_update_shifts_traffic() {
+        let backend_a = spawn_greeter_server("backend-a", None, None)
+            .await
+            .expect("spawn backend-a");
+        let backend_b = spawn_greeter_server("backend-b", None, None)
+            .await
+            .expect("spawn backend-b");
+
+        let (control_plane, cp_addr) = start_control_plane().await;
+
+        // LDS -> RDS "route-config"; both clusters and their endpoints are configured
+        // up front, and the route initially targets cluster-a.
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Lds,
+            HashMap::from([(
+                "my-service".to_string(),
+                config::build_rds_listener("my-service", "route-config"),
+            )]),
+        );
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Rds,
+            HashMap::from([(
+                "route-config".to_string(),
+                config::build_route_config("route-config", "cluster-a"),
+            )]),
+        );
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Cds,
+            HashMap::from([
+                ("cluster-a".to_string(), config::build_cluster("cluster-a")),
+                ("cluster-b".to_string(), config::build_cluster("cluster-b")),
+            ]),
+        );
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Eds,
+            HashMap::from([
+                (
+                    "cluster-a".to_string(),
+                    config::build_cla(
+                        "cluster-a",
+                        &[(backend_a.addr.ip().to_string(), backend_a.addr.port())],
+                    ),
+                ),
+                (
+                    "cluster-b".to_string(),
+                    config::build_cla(
+                        "cluster-b",
+                        &[(backend_b.addr.ip().to_string(), backend_b.addr.port())],
+                    ),
+                ),
+            ]),
+        );
+
+        let mut client = GreeterClient::new(build_channel(cp_addr, "my-service"));
+
+        // Initially routed to cluster-a -> backend-a.
+        let reply = say_hello_until_prefix(&mut client, "backend-a:").await;
+        assert_eq!(reply, "backend-a: world");
+
+        // Update the RDS route to target cluster-b; the control plane pushes the new
+        // RouteConfiguration to the connected client.
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Rds,
+            HashMap::from([(
+                "route-config".to_string(),
+                config::build_route_config("route-config", "cluster-b"),
+            )]),
+        );
+
+        // Traffic shifts to cluster-b -> backend-b.
+        let reply = say_hello_until_prefix(&mut client, "backend-b:").await;
+        assert_eq!(reply, "backend-b: world");
+
+        let _ = backend_a.shutdown.send(());
+        let _ = backend_b.shutdown.send(());
+    }
+
+    /// P2C load balancing: with several EDS endpoints behind one cluster, traffic
+    /// should spread across all of them. tonic-xds uses power-of-two-choices as its
+    /// balancer (see the unit-level `test_xds_channel_grpc_with_p2c_lb`); this is the
+    /// full-pipeline version through the control plane.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn xds_channel_e2e_p2c_spreads_across_backends() {
+        const NUM_BACKENDS: usize = 5;
+        const NUM_REQUESTS: usize = 1000;
+
+        // Spawn several greeter backends, each reporting a distinct name.
+        let mut backends = Vec::new();
+        for i in 0..NUM_BACKENDS {
+            backends.push(
+                spawn_greeter_server(&format!("backend-{i}"), None, None)
+                    .await
+                    .expect("spawn backend"),
+            );
+        }
+        let endpoints: Vec<(String, u16)> = backends
+            .iter()
+            .map(|backend| (backend.addr.ip().to_string(), backend.addr.port()))
+            .collect();
+
+        let (control_plane, cp_addr) = start_control_plane().await;
+
+        // Configure LDS (inline route) -> CDS -> EDS with all backends in one cluster.
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Lds,
+            HashMap::from([(
+                "my-service".to_string(),
+                config::build_inline_listener("my-service", "my-cluster"),
+            )]),
+        );
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Cds,
+            HashMap::from([(
+                "my-cluster".to_string(),
+                config::build_cluster("my-cluster"),
+            )]),
+        );
+        control_plane.get_service().set_xds_config(
+            &config::AdsTypeUrl::Eds,
+            HashMap::from([(
+                "my-cluster".to_string(),
+                config::build_cla("my-cluster", &endpoints),
+            )]),
+        );
+
+        let mut client = GreeterClient::new(build_channel(cp_addr, "my-service"));
+
+        // Warm up until xDS resolution completes and the first RPC succeeds.
+        say_hello_until_prefix(&mut client, "backend-").await;
+
+        // Tally which backend served each request.
+        let mut counts: HashMap<String, usize> = HashMap::new();
+        for _ in 0..NUM_REQUESTS {
+            let reply = client
+                .say_hello(HelloRequest {
+                    name: "world".to_string(),
+                })
+                .await
+                .expect("rpc succeeds")
+                .into_inner()
+                .message;
+            let server = reply.split(':').next().unwrap_or_default().to_string();
+            *counts.entry(server).or_default() += 1;
+        }
+
+        // Every backend should have received a fair share of the traffic.
+        assert_eq!(counts.values().sum::<usize>(), NUM_REQUESTS);
+        // Each backend should receive at least 50% of the fair share of requests.
+        let min_per_backend = (NUM_REQUESTS / NUM_BACKENDS) / 2;
+        for i in 0..NUM_BACKENDS {
+            let name = format!("backend-{i}");
+            let count = counts.get(&name).copied().unwrap_or(0);
+            assert!(
+                count >= min_per_backend,
+                "backend {name} received {count} requests (< {min_per_backend}); distribution: {counts:?}",
+            );
+        }
+
+        for backend in backends {
+            let _ = backend.shutdown.send(());
+        }
+    }
+}

xds-test-util/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "xds-test-util"
+version = "0.1.0"
+edition = "2024"
+rust-version.workspace = true
+homepage = "https://github.com/grpc/grpc-rust"
+repository = "https://github.com/grpc/grpc-rust"
+description = """
+Test utilities for gRPC xDS implementations.
+"""
+keywords = ["grpc", "xds", "test"]
+license = "MIT"
+publish = false
+
+[lints]
+workspace = true
+
+[dependencies]
+envoy-types = "0.7"
+prost = "0.14"
+tonic = { version = "0.14", features = ["server"] }
+tokio = { version = "1", features = ["sync", "rt", "net"] }
+tokio-stream = { version = "0.1", features = ["net"] }
+strum = { version = "0.28.0", features = ["derive"] }
+tracing = "0.1.44"