From 69076e7ae4b1a285ebf40311dd12275c08c4131a Mon Sep 17 00:00:00 2001 From: Jeff Jiang Date: Sat, 11 Jul 2026 01:28:13 -0700 Subject: [PATCH 1/6] xds test util --- Cargo.toml | 1 + tonic-xds/Cargo.toml | 1 + tonic-xds/src/client/mod.rs | 3 + tonic-xds/src/client/xds_e2e_test.rs | 207 ++++++++++++++ xds-test-util/Cargo.toml | 23 ++ xds-test-util/src/control_plane.rs | 412 +++++++++++++++++++++++++++ xds-test-util/src/lib.rs | 28 ++ 7 files changed, 675 insertions(+) create mode 100644 tonic-xds/src/client/xds_e2e_test.rs create mode 100644 xds-test-util/Cargo.toml create mode 100644 xds-test-util/src/control_plane.rs create mode 100644 xds-test-util/src/lib.rs diff --git a/Cargo.toml b/Cargo.toml index c492e8b15..b2c821450 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ members = [ "protoc-gen-rust-grpc", "xds-client", "tonic-xds", + "xds-test-util", "interop", "tests/disable_comments", "tests/wellknown", diff --git a/tonic-xds/Cargo.toml b/tonic-xds/Cargo.toml index 3559db617..a84ffde60 100644 --- a/tonic-xds/Cargo.toml +++ b/tonic-xds/Cargo.toml @@ -61,6 +61,7 @@ workspace = true [dev-dependencies] xds-client = { version = "0.1.0-alpha.1", 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" diff --git a/tonic-xds/src/client/mod.rs b/tonic-xds/src/client/mod.rs index 3e02c9b29..1687f8bcb 100644 --- a/tonic-xds/src/client/mod.rs +++ b/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_test; diff --git a/tonic-xds/src/client/xds_e2e_test.rs b/tonic-xds/src/client/xds_e2e_test.rs new file mode 100644 index 000000000..41a340abc --- /dev/null +++ b/tonic-xds/src/client/xds_e2e_test.rs @@ -0,0 +1,207 @@ +//! End-to-end test: a real xDS channel resolving through a fake ADS control +//! plane to a live gRPC backend. +//! +//! Unlike the unit tests in [`super::channel`], which inject endpoints through +//! `build_grpc_channel_from_parts`, this test exercises the full pipeline: +//! bootstrap loading, the ADS transport, and LDS -> CDS -> EDS resolution served +//! by [`XdsTestControlPlaneService`]. Traffic is routed to a greeter (echo) +//! backend spawned by the test. + +use std::collections::HashMap; +use std::time::Duration; + +use envoy_types::pb::envoy::config::cluster::v3::Cluster; +use envoy_types::pb::envoy::config::cluster::v3::cluster::{ClusterDiscoveryType, DiscoveryType}; +use envoy_types::pb::envoy::config::core::v3 as core_v3; +use envoy_types::pb::envoy::config::endpoint::v3::{ + ClusterLoadAssignment, Endpoint, LbEndpoint, LocalityLbEndpoints, lb_endpoint::HostIdentifier, +}; +use envoy_types::pb::envoy::config::listener::v3::{ApiListener, Listener}; +use envoy_types::pb::envoy::config::route::v3::route::Action; +use envoy_types::pb::envoy::config::route::v3::route_action::ClusterSpecifier; +use envoy_types::pb::envoy::config::route::v3::route_match::PathSpecifier; +use envoy_types::pb::envoy::config::route::v3::{ + Route, RouteAction, RouteConfiguration, RouteMatch, VirtualHost, +}; +use envoy_types::pb::envoy::extensions::filters::network::http_connection_manager::v3::{ + HttpConnectionManager, http_connection_manager::RouteSpecifier, +}; +use envoy_types::pb::google::protobuf::Any; +use prost::Message; +use tokio::net::TcpListener; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::transport::Server; +use xds_test_util::{ + ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, AggregatedDiscoveryServiceServer, + XdsTestControlPlaneService, +}; + +use crate::testutil::grpc::{GreeterClient, HelloRequest, spawn_greeter_server}; +use crate::{BootstrapConfig, XdsChannelBuilder, XdsChannelConfig, XdsUri}; + +const LISTENER_NAME: &str = "my-service"; +const CLUSTER_NAME: &str = "my-cluster"; +const TYPE_HCM: &str = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager"; + +/// Builds a gRPC listener whose inline route configuration sends all traffic to +/// `CLUSTER_NAME`. +fn build_listener() -> Listener { + let hcm = HttpConnectionManager { + route_specifier: Some(RouteSpecifier::RouteConfig(RouteConfiguration { + name: format!("{LISTENER_NAME}-route"), + virtual_hosts: vec![VirtualHost { + name: "default".to_string(), + domains: vec!["*".to_string()], + routes: vec![Route { + r#match: Some(RouteMatch { + path_specifier: Some(PathSpecifier::Prefix("/".to_string())), + ..Default::default() + }), + action: Some(Action::Route(RouteAction { + cluster_specifier: Some(ClusterSpecifier::Cluster( + CLUSTER_NAME.to_string(), + )), + ..Default::default() + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + })), + ..Default::default() + }; + Listener { + name: LISTENER_NAME.to_string(), + api_listener: Some(ApiListener { + api_listener: Some(Any { + type_url: TYPE_HCM.to_string(), + value: hcm.encode_to_vec(), + }), + }), + ..Default::default() + } +} + +/// Builds an EDS-discovered cluster named `CLUSTER_NAME`. +fn build_cluster() -> Cluster { + Cluster { + name: CLUSTER_NAME.to_string(), + cluster_discovery_type: Some(ClusterDiscoveryType::Type(DiscoveryType::Eds as i32)), + ..Default::default() + } +} + +/// Builds a `ClusterLoadAssignment` with a single endpoint at `host:port`. +fn build_endpoints(host: &str, port: u16) -> ClusterLoadAssignment { + ClusterLoadAssignment { + cluster_name: CLUSTER_NAME.to_string(), + endpoints: vec![LocalityLbEndpoints { + lb_endpoints: vec![LbEndpoint { + host_identifier: Some(HostIdentifier::Endpoint(Endpoint { + address: Some(core_v3::Address { + address: Some(core_v3::address::Address::SocketAddress( + core_v3::SocketAddress { + address: host.to_string(), + port_specifier: Some( + core_v3::socket_address::PortSpecifier::PortValue(u32::from( + port, + )), + ), + ..Default::default() + }, + )), + }), + ..Default::default() + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn xds_channel_e2e_routes_to_backend() { + // 1. Spawn the greeter (echo) backend. + let backend = spawn_greeter_server("backend", None, None) + .await + .expect("spawn greeter backend"); + let backend_addr = backend.addr; + + // 2. Start the fake ADS control plane on an ephemeral port. + let control_plane = XdsTestControlPlaneService::new(); + let cp_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let cp_addr = cp_listener.local_addr().unwrap(); + let cp_service = control_plane.clone(); + let cp_handle = tokio::spawn(async move { + Server::builder() + .add_service(AggregatedDiscoveryServiceServer::new(cp_service)) + .serve_with_incoming(TcpListenerStream::new(cp_listener)) + .await + }); + + // 3. Configure LDS -> CDS -> EDS pointing at the backend. + control_plane.set_xds_config( + ADS_TYPE_URL_LDS, + HashMap::from([(LISTENER_NAME.to_string(), build_listener())]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_CDS, + HashMap::from([(CLUSTER_NAME.to_string(), build_cluster())]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_EDS, + HashMap::from([( + CLUSTER_NAME.to_string(), + build_endpoints(&backend_addr.ip().to_string(), backend_addr.port()), + )]), + ); + + // 4. Build a real xDS channel via bootstrap pointing at the control plane. + 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"); + let channel = XdsChannelBuilder::new(XdsChannelConfig::new(target).with_bootstrap(bootstrap)) + .build_grpc_channel() + .expect("build xds channel"); + + let mut client = GreeterClient::new(channel); + + // 5. Send requests, retrying until xDS resolution completes. + let mut reply = None; + let mut last_err = None; + for _ in 0..50 { + match client + .say_hello(HelloRequest { + name: "world".to_string(), + }) + .await + { + Ok(response) => { + reply = Some(response.into_inner().message); + break; + } + Err(status) => { + last_err = Some(status); + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + } + + let reply = reply.unwrap_or_else(|| panic!("no successful RPC; last error: {last_err:?}")); + assert_eq!(reply, "backend: world"); + + // The control plane observed exactly one ADS stream from the client. + let counts = control_plane.get_subscriber_counts(); + assert_eq!(counts.get(ADS_TYPE_URL_LDS), Some(&1)); + assert_eq!(counts.get(ADS_TYPE_URL_CDS), Some(&1)); + assert_eq!(counts.get(ADS_TYPE_URL_EDS), Some(&1)); + + // 6. Cleanup. + let _ = backend.shutdown.send(()); + cp_handle.abort(); +} diff --git a/xds-test-util/Cargo.toml b/xds-test-util/Cargo.toml new file mode 100644 index 000000000..56b783eb9 --- /dev/null +++ b/xds-test-util/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "xds-test-util" +version = "0.1.0-alpha.1" +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 = "0.14" +tokio = { version = "1", features = ["sync", "rt"] } +tokio-stream = "0.1" diff --git a/xds-test-util/src/control_plane.rs b/xds-test-util/src/control_plane.rs new file mode 100644 index 000000000..501c8bafa --- /dev/null +++ b/xds-test-util/src/control_plane.rs @@ -0,0 +1,412 @@ +//! A fake xDS ADS control plane for tests. +//! +//! Rust port of grpc-java's `XdsTestControlPlaneService`. It is a bidi-stream +//! service that acts as a local xDS control plane. Config is injected through +//! [`set_xds_config`](XdsTestControlPlaneService::set_xds_config). +//! +//! The service maintains, per ADS resource type: +//! - a resources table (resource name to packed protobuf `Any`), +//! - a subscriber table (each active stream to its subscribed resource names), +//! - a version counter (bumped on every config set), and +//! - a per-stream nonce counter. +//! +//! All resource types are treated as state-of-the-world: whenever any resource +//! of a type changes, every subscriber to that type receives all of its +//! subscribed resources of that type. +//! +//! Incoming ADS requests share the same proto message but represent different +//! phases, which the service distinguishes: +//! 1. an initial request (new or changed subscription) — answered with a response, +//! 2. a NACK (carries `error_detail`) — logged and ignored, +//! 3. an ACK (same resource names already subscribed) — ignored. + +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::AtomicU64; +use std::sync::{Arc, Mutex}; + +use envoy_types::pb::envoy::service::discovery::v3::{DiscoveryRequest, DiscoveryResponse}; +use envoy_types::pb::google::protobuf::Any; +use prost::Message; +use tokio::sync::mpsc; + +/// ADS type URL for LDS (`Listener`) resources. +pub const ADS_TYPE_URL_LDS: &str = "type.googleapis.com/envoy.config.listener.v3.Listener"; +/// ADS type URL for RDS (`RouteConfiguration`) resources. +pub const ADS_TYPE_URL_RDS: &str = "type.googleapis.com/envoy.config.route.v3.RouteConfiguration"; +/// ADS type URL for CDS (`Cluster`) resources. +pub const ADS_TYPE_URL_CDS: &str = "type.googleapis.com/envoy.config.cluster.v3.Cluster"; +/// ADS type URL for EDS (`ClusterLoadAssignment`) resources. +pub const ADS_TYPE_URL_EDS: &str = + "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"; + +/// The ADS resource types tracked by the control plane. +const ADS_TYPE_URLS: [&str; 4] = [ + ADS_TYPE_URL_LDS, + ADS_TYPE_URL_RDS, + ADS_TYPE_URL_CDS, + ADS_TYPE_URL_EDS, +]; + +/// Sender half of a stream's outbound `DiscoveryResponse` channel. +/// +/// The channel carries bare `DiscoveryResponse`s; the (framework-specific) +/// serving adapter wraps them in the transport's success type. +type ResponseSender = mpsc::UnboundedSender; + +/// Per-(type, stream) subscription bookkeeping. +#[derive(Debug)] +struct Subscription { + /// Outbound channel used to push responses to this stream. + sender: ResponseSender, + /// Resource names currently subscribed for this type on this stream. + resource_names: HashSet, + /// Last nonce sent to this stream for this type (starts at 0). + nonce: u64, +} + +/// Mutable control-plane state, guarded by a single mutex. +#[derive(Debug)] +struct State { + /// `type_url` to (resource name to packed `Any`). + resources: HashMap>, + /// `type_url` to latest version (starts at 1, bumped on each config set). + versions: HashMap, + /// `type_url` to (stream id to subscription). + subscribers: HashMap>, +} + +impl State { + fn new() -> Self { + let mut versions = HashMap::new(); + let mut subscribers = HashMap::new(); + for type_url in ADS_TYPE_URLS { + versions.insert(type_url.to_string(), 1); + subscribers.insert(type_url.to_string(), HashMap::new()); + } + Self { + resources: HashMap::new(), + versions, + subscribers, + } + } +} + +/// Shared inner state of the control plane. +#[derive(Debug)] +struct Inner { + state: Mutex, + next_stream_id: AtomicU64, +} + +impl Inner { + /// Handles a single inbound `DiscoveryRequest` for a stream. + fn handle_request(&self, stream_id: u64, tx: &ResponseSender, req: DiscoveryRequest) { + let mut state = self.state.lock().expect("control plane state poisoned"); + + // NACK: a request carrying an error detail rejects the last response. + if req.error_detail.is_some() { + return; + } + + let type_url = req.type_url; + if type_url.is_empty() { + return; + } + + // Nonce check: if the request carries a response nonce, it must match + // the last nonce we sent on this (type, stream); otherwise ignore it. + if !req.response_nonce.is_empty() { + let matches = state + .subscribers + .get(&type_url) + .and_then(|streams| streams.get(&stream_id)) + .is_some_and(|sub| sub.nonce.to_string() == req.response_nonce); + if !matches { + return; + } + } + + let requested: HashSet = req.resource_names.into_iter().collect(); + + // ACK: identical subscription already recorded, nothing to send. + if let Some(sub) = state + .subscribers + .get(&type_url) + .and_then(|streams| streams.get(&stream_id)) + && sub.resource_names == requested + { + return; + } + + let version = state.versions.get(&type_url).copied().unwrap_or(1); + + // Borrow the resource and subscriber tables as disjoint fields so the + // response can be built from `resources` while mutating `subscribers`. + let State { + resources, + subscribers, + .. + } = &mut *state; + let type_resources = resources.get(&type_url); + let sub = subscribers + .entry(type_url.clone()) + .or_default() + .entry(stream_id) + .or_insert_with(|| Subscription { + sender: tx.clone(), + resource_names: HashSet::new(), + nonce: 0, + }); + sub.nonce += 1; + let response = build_response(&type_url, version, sub.nonce, &requested, type_resources); + let _ = sub.sender.send(response); + sub.resource_names = requested; + } + + /// Removes all subscriptions for a stream when it ends. + fn remove_stream(&self, stream_id: u64) { + let mut state = self.state.lock().expect("control plane state poisoned"); + for streams in state.subscribers.values_mut() { + streams.remove(&stream_id); + } + } +} + +/// A bidi-stream service that acts as a local xDS control plane for tests. +/// +/// Clone freely: all clones share the same underlying state, so config injected +/// through one clone is served by another. Inject config with +/// [`set_xds_config`](Self::set_xds_config) and retain a clone to push updates +/// while the server runs. +/// +/// The service implements the ADS server trait and can be registered with a +/// tonic server via [`AggregatedDiscoveryServiceServer`](crate::AggregatedDiscoveryServiceServer). +#[derive(Clone, Debug)] +pub struct XdsTestControlPlaneService { + inner: Arc, +} + +impl Default for XdsTestControlPlaneService { + fn default() -> Self { + Self::new() + } +} + +impl XdsTestControlPlaneService { + /// Creates a new control plane with no resources configured. + #[must_use] + pub fn new() -> Self { + Self { + inner: Arc::new(Inner { + state: Mutex::new(State::new()), + next_stream_id: AtomicU64::new(0), + }), + } + } + + /// Sets the full set of resources for `type_url`, replacing any previous + /// resources of that type, and pushes an update to every current subscriber. + /// + /// Each resource is packed into a protobuf `Any` tagged with `type_url`. + /// Resource types are treated as state-of-the-world. + pub fn set_xds_config(&self, type_url: &str, resources: HashMap) { + let packed: HashMap = resources + .into_iter() + .map(|(name, msg)| { + ( + name, + Any { + type_url: type_url.to_string(), + value: msg.encode_to_vec(), + }, + ) + }) + .collect(); + + let mut state = self.inner.state.lock().expect("control plane state poisoned"); + state.resources.insert(type_url.to_string(), packed); + + // getAndIncrement: the pushed response carries the pre-increment version. + let version = { + let counter = state.versions.entry(type_url.to_string()).or_insert(1); + let current = *counter; + *counter += 1; + current + }; + + let State { + resources, + subscribers, + .. + } = &mut *state; + let type_resources = resources.get(type_url); + if let Some(streams) = subscribers.get_mut(type_url) { + for sub in streams.values_mut() { + sub.nonce += 1; + let response = + build_response(type_url, version, sub.nonce, &sub.resource_names, type_resources); + let _ = sub.sender.send(response); + } + } + } + + /// Returns the currently configured resources for `type_url`, as packed + /// protobuf `Any`s keyed by resource name. + /// + /// For a protobuf-free view suitable for assertions, use + /// [`resource_names`](Self::resource_names). + #[must_use] + pub fn get_current_config(&self, type_url: &str) -> HashMap { + let state = self.inner.state.lock().expect("control plane state poisoned"); + state.resources.get(type_url).cloned().unwrap_or_default() + } + + /// Returns the names of the currently configured resources for `type_url`, + /// sorted. + /// + /// This is a protocol-agnostic view of what the control plane is serving, + /// suitable for assertions without depending on protobuf types. + #[must_use] + pub fn resource_names(&self, type_url: &str) -> Vec { + let state = self.inner.state.lock().expect("control plane state poisoned"); + state + .resources + .get(type_url) + .map(|resources| { + let mut names: Vec = resources.keys().cloned().collect(); + names.sort(); + names + }) + .unwrap_or_default() + } + + /// Returns the number of active subscribers per ADS resource type. + #[must_use] + pub fn get_subscriber_counts(&self) -> HashMap { + let state = self.inner.state.lock().expect("control plane state poisoned"); + ADS_TYPE_URLS + .iter() + .map(|type_url| { + ( + (*type_url).to_string(), + state.subscribers.get(*type_url).map_or(0, HashMap::len), + ) + }) + .collect() + } +} + +/// Tonic serving adapter for the ADS bidi stream. +mod tonic_service { + use std::pin::Pin; + use std::sync::atomic::Ordering; + + use envoy_types::pb::envoy::service::discovery::v3::{ + DeltaDiscoveryRequest, DeltaDiscoveryResponse, DiscoveryRequest, DiscoveryResponse, + aggregated_discovery_service_server::AggregatedDiscoveryService, + }; + use tokio_stream::wrappers::UnboundedReceiverStream; + use tokio_stream::{Stream, StreamExt}; + use tonic::{Request, Response, Status}; + + use super::{Arc, XdsTestControlPlaneService, mpsc}; + + #[tonic::async_trait] + impl AggregatedDiscoveryService for XdsTestControlPlaneService { + type StreamAggregatedResourcesStream = + Pin> + Send>>; + + async fn stream_aggregated_resources( + &self, + request: Request>, + ) -> Result, Status> { + let mut inbound = request.into_inner(); + let (tx, rx) = mpsc::unbounded_channel::(); + let inner = Arc::clone(&self.inner); + let stream_id = inner.next_stream_id.fetch_add(1, Ordering::Relaxed); + + tokio::spawn(async move { + while let Some(item) = inbound.next().await { + match item { + Ok(req) => inner.handle_request(stream_id, &tx, req), + Err(_) => break, + } + } + inner.remove_stream(stream_id); + }); + + let outbound = UnboundedReceiverStream::new(rx).map(Ok); + Ok(Response::new(Box::pin(outbound))) + } + + type DeltaAggregatedResourcesStream = + Pin> + Send>>; + + async fn delta_aggregated_resources( + &self, + _request: Request>, + ) -> Result, Status> { + Err(Status::unimplemented( + "delta ADS is not supported by the test control plane", + )) + } + } +} + +/// Builds a `DiscoveryResponse` containing the requested resources that exist. +fn build_response( + type_url: &str, + version: u64, + nonce: u64, + resource_names: &HashSet, + type_resources: Option<&HashMap>, +) -> DiscoveryResponse { + let resources = match type_resources { + Some(map) => resource_names + .iter() + .filter_map(|name| map.get(name).cloned()) + .collect(), + None => Vec::new(), + }; + DiscoveryResponse { + version_info: version.to_string(), + resources, + type_url: type_url.to_string(), + nonce: nonce.to_string(), + ..Default::default() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use envoy_types::pb::envoy::config::listener::v3::Listener; + + #[test] + fn set_and_get_config() { + let control_plane = XdsTestControlPlaneService::new(); + + let mut listeners = HashMap::new(); + listeners.insert( + "my-listener".to_string(), + Listener { + name: "my-listener".to_string(), + ..Default::default() + }, + ); + control_plane.set_xds_config(ADS_TYPE_URL_LDS, listeners); + + let config = control_plane.get_current_config(ADS_TYPE_URL_LDS); + assert_eq!(config.len(), 1); + let packed = config.get("my-listener").expect("listener present"); + assert_eq!(packed.type_url, ADS_TYPE_URL_LDS); + + // No streams have connected yet. + let counts = control_plane.get_subscriber_counts(); + assert_eq!(counts.len(), 4); + assert_eq!(counts.get(ADS_TYPE_URL_LDS), Some(&0)); + + // A type that was never configured has no resources. + assert!(control_plane.get_current_config(ADS_TYPE_URL_CDS).is_empty()); + } +} diff --git a/xds-test-util/src/lib.rs b/xds-test-util/src/lib.rs new file mode 100644 index 000000000..259a8922a --- /dev/null +++ b/xds-test-util/src/lib.rs @@ -0,0 +1,28 @@ +//! Test utilities for xDS implementations. +//! +//! This crate provides helpers for exercising xDS clients against an in-process +//! management server. It is intended for tests only and not for production use. +//! +//! The main entry point is [`XdsTestControlPlaneService`], a fake Aggregated +//! Discovery Service (ADS) control plane. It is a Rust port of grpc-java's +//! `XdsTestControlPlaneService`. + +mod control_plane; + +pub use control_plane::{ + ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, ADS_TYPE_URL_RDS, + XdsTestControlPlaneService, +}; + +/// Re-export of the generated ADS server adapter, so tests can register the +/// control plane with a [`tonic`] server without depending on `envoy-types` +/// module paths directly: +/// +/// ```ignore +/// let service = XdsTestControlPlaneService::new(); +/// Server::builder() +/// .add_service(AggregatedDiscoveryServiceServer::new(service.clone())) +/// .serve(addr) +/// .await?; +/// ``` +pub use envoy_types::pb::envoy::service::discovery::v3::aggregated_discovery_service_server::AggregatedDiscoveryServiceServer; From 71b0fa606e77a46a2655554f7543830d0c2f1868 Mon Sep 17 00:00:00 2001 From: Jeff Jiang Date: Sat, 11 Jul 2026 02:03:58 -0700 Subject: [PATCH 2/6] more tests --- tonic-xds/src/client/xds_e2e_test.rs | 454 ++++++++++++++++++++------- 1 file changed, 340 insertions(+), 114 deletions(-) diff --git a/tonic-xds/src/client/xds_e2e_test.rs b/tonic-xds/src/client/xds_e2e_test.rs index 41a340abc..5cbbfad7b 100644 --- a/tonic-xds/src/client/xds_e2e_test.rs +++ b/tonic-xds/src/client/xds_e2e_test.rs @@ -1,13 +1,14 @@ -//! End-to-end test: a real xDS channel resolving through a fake ADS control -//! plane to a live gRPC backend. +//! 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`, this test exercises the full pipeline: -//! bootstrap loading, the ADS transport, and LDS -> CDS -> EDS resolution served -//! by [`XdsTestControlPlaneService`]. Traffic is routed to a greeter (echo) -//! backend spawned by the test. +//! `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. use std::collections::HashMap; +use std::net::SocketAddr; use std::time::Duration; use envoy_types::pb::envoy::config::cluster::v3::Cluster; @@ -24,55 +25,32 @@ use envoy_types::pb::envoy::config::route::v3::{ Route, RouteAction, RouteConfiguration, RouteMatch, VirtualHost, }; use envoy_types::pb::envoy::extensions::filters::network::http_connection_manager::v3::{ - HttpConnectionManager, http_connection_manager::RouteSpecifier, + HttpConnectionManager, Rds, http_connection_manager::RouteSpecifier, }; use envoy_types::pb::google::protobuf::Any; use prost::Message; use tokio::net::TcpListener; +use tokio::task::JoinHandle; use tokio_stream::wrappers::TcpListenerStream; use tonic::transport::Server; use xds_test_util::{ - ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, AggregatedDiscoveryServiceServer, - XdsTestControlPlaneService, + ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, ADS_TYPE_URL_RDS, + AggregatedDiscoveryServiceServer, XdsTestControlPlaneService, }; use crate::testutil::grpc::{GreeterClient, HelloRequest, spawn_greeter_server}; -use crate::{BootstrapConfig, XdsChannelBuilder, XdsChannelConfig, XdsUri}; +use crate::{BootstrapConfig, XdsChannelBuilder, XdsChannelConfig, XdsChannelGrpc, XdsUri}; -const LISTENER_NAME: &str = "my-service"; -const CLUSTER_NAME: &str = "my-cluster"; const TYPE_HCM: &str = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager"; -/// Builds a gRPC listener whose inline route configuration sends all traffic to -/// `CLUSTER_NAME`. -fn build_listener() -> Listener { +/// Wraps a `HttpConnectionManager` route specifier in an `ApiListener` listener. +fn build_listener(name: &str, route: RouteSpecifier) -> Listener { let hcm = HttpConnectionManager { - route_specifier: Some(RouteSpecifier::RouteConfig(RouteConfiguration { - name: format!("{LISTENER_NAME}-route"), - virtual_hosts: vec![VirtualHost { - name: "default".to_string(), - domains: vec!["*".to_string()], - routes: vec![Route { - r#match: Some(RouteMatch { - path_specifier: Some(PathSpecifier::Prefix("/".to_string())), - ..Default::default() - }), - action: Some(Action::Route(RouteAction { - cluster_specifier: Some(ClusterSpecifier::Cluster( - CLUSTER_NAME.to_string(), - )), - ..Default::default() - })), - ..Default::default() - }], - ..Default::default() - }], - ..Default::default() - })), + route_specifier: Some(route), ..Default::default() }; Listener { - name: LISTENER_NAME.to_string(), + name: name.to_string(), api_listener: Some(ApiListener { api_listener: Some(Any { type_url: TYPE_HCM.to_string(), @@ -83,97 +61,141 @@ fn build_listener() -> Listener { } } -/// Builds an EDS-discovered cluster named `CLUSTER_NAME`. -fn build_cluster() -> Cluster { +/// Builds a listener whose route configuration is embedded inline and sends all +/// traffic to `cluster`. +fn build_inline_listener(name: &str, cluster: &str) -> Listener { + build_listener( + name, + RouteSpecifier::RouteConfig(build_route_config(&format!("{name}-route"), cluster)), + ) +} + +/// Builds a listener that fetches its route configuration via RDS by name. +fn build_rds_listener(name: &str, rds_name: &str) -> Listener { + build_listener( + name, + RouteSpecifier::Rds(Rds { + route_config_name: rds_name.to_string(), + ..Default::default() + }), + ) +} + +/// Builds a route configuration that sends all traffic to `cluster`. +fn build_route_config(name: &str, cluster: &str) -> RouteConfiguration { + RouteConfiguration { + name: name.to_string(), + virtual_hosts: vec![VirtualHost { + name: "default".to_string(), + domains: vec!["*".to_string()], + routes: vec![Route { + r#match: Some(RouteMatch { + path_specifier: Some(PathSpecifier::Prefix("/".to_string())), + ..Default::default() + }), + action: Some(Action::Route(RouteAction { + cluster_specifier: Some(ClusterSpecifier::Cluster(cluster.to_string())), + ..Default::default() + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + } +} + +/// Builds an EDS-discovered cluster named `name`. +fn build_cluster(name: &str) -> Cluster { Cluster { - name: CLUSTER_NAME.to_string(), + name: name.to_string(), cluster_discovery_type: Some(ClusterDiscoveryType::Type(DiscoveryType::Eds as i32)), ..Default::default() } } -/// Builds a `ClusterLoadAssignment` with a single endpoint at `host:port`. -fn build_endpoints(host: &str, port: u16) -> ClusterLoadAssignment { +/// Builds a `ClusterLoadAssignment` for `cluster` with a single endpoint. +fn build_endpoints(cluster: &str, host: &str, port: u16) -> ClusterLoadAssignment { + build_endpoints_multi(cluster, &[(host.to_string(), port)]) +} + +/// Builds a `ClusterLoadAssignment` for `cluster` with the given endpoints, all +/// placed in a single locality. +fn build_endpoints_multi(cluster: &str, endpoints: &[(String, u16)]) -> ClusterLoadAssignment { ClusterLoadAssignment { - cluster_name: CLUSTER_NAME.to_string(), + cluster_name: cluster.to_string(), endpoints: vec![LocalityLbEndpoints { - lb_endpoints: vec![LbEndpoint { - host_identifier: Some(HostIdentifier::Endpoint(Endpoint { - address: Some(core_v3::Address { - address: Some(core_v3::address::Address::SocketAddress( - core_v3::SocketAddress { - address: host.to_string(), - port_specifier: Some( - core_v3::socket_address::PortSpecifier::PortValue(u32::from( - port, - )), - ), - ..Default::default() - }, - )), - }), - ..Default::default() - })), - ..Default::default() - }], + lb_endpoints: endpoints + .iter() + .map(|(host, port)| lb_endpoint(host, *port)) + .collect(), ..Default::default() }], ..Default::default() } } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn xds_channel_e2e_routes_to_backend() { - // 1. Spawn the greeter (echo) backend. - let backend = spawn_greeter_server("backend", None, None) - .await - .expect("spawn greeter backend"); - let backend_addr = backend.addr; +/// Builds a single `LbEndpoint` at `host:port`. +fn lb_endpoint(host: &str, port: u16) -> LbEndpoint { + LbEndpoint { + host_identifier: Some(HostIdentifier::Endpoint(Endpoint { + address: Some(core_v3::Address { + address: Some(core_v3::address::Address::SocketAddress( + core_v3::SocketAddress { + address: host.to_string(), + port_specifier: Some(core_v3::socket_address::PortSpecifier::PortValue( + u32::from(port), + )), + ..Default::default() + }, + )), + }), + ..Default::default() + })), + ..Default::default() + } +} - // 2. Start the fake ADS control plane on an ephemeral port. +/// Starts the fake ADS control plane on an ephemeral port. Returns the service +/// handle (for injecting config), its address, and the server task handle. +async fn start_control_plane() -> ( + XdsTestControlPlaneService, + SocketAddr, + JoinHandle>, +) { let control_plane = XdsTestControlPlaneService::new(); - let cp_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let cp_addr = cp_listener.local_addr().unwrap(); - let cp_service = control_plane.clone(); - let cp_handle = tokio::spawn(async move { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let service = control_plane.clone(); + let handle = tokio::spawn(async move { Server::builder() - .add_service(AggregatedDiscoveryServiceServer::new(cp_service)) - .serve_with_incoming(TcpListenerStream::new(cp_listener)) + .add_service(AggregatedDiscoveryServiceServer::new(service)) + .serve_with_incoming(TcpListenerStream::new(listener)) .await }); + (control_plane, addr, handle) +} - // 3. Configure LDS -> CDS -> EDS pointing at the backend. - control_plane.set_xds_config( - ADS_TYPE_URL_LDS, - HashMap::from([(LISTENER_NAME.to_string(), build_listener())]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_CDS, - HashMap::from([(CLUSTER_NAME.to_string(), build_cluster())]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_EDS, - HashMap::from([( - CLUSTER_NAME.to_string(), - build_endpoints(&backend_addr.ip().to_string(), backend_addr.port()), - )]), - ); - - // 4. Build a real xDS channel via bootstrap pointing at the control plane. - let bootstrap_json = format!( - r#"{{"xds_servers":[{{"server_uri":"http://{cp_addr}"}}],"node":{{"id":"test"}}}}"# - ); +/// 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"); - let channel = XdsChannelBuilder::new(XdsChannelConfig::new(target).with_bootstrap(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"); - - let mut client = GreeterClient::new(channel); + .expect("build xds channel") +} - // 5. Send requests, retrying until xDS resolution completes. - let mut reply = None; - let mut last_err = None; +/// 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, + want_prefix: &str, +) -> String { + let mut last = None; for _ in 0..50 { match client .say_hello(HelloRequest { @@ -182,17 +204,54 @@ async fn xds_channel_e2e_routes_to_backend() { .await { Ok(response) => { - reply = Some(response.into_inner().message); - break; - } - Err(status) => { - last_err = Some(status); - tokio::time::sleep(Duration::from_millis(100)).await; + 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:?}"); +} - let reply = reply.unwrap_or_else(|| panic!("no successful RPC; last error: {last_err:?}")); +#[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, cp_handle) = start_control_plane().await; + + // Configure LDS (inline route) -> CDS -> EDS pointing at the backend. + control_plane.set_xds_config( + ADS_TYPE_URL_LDS, + HashMap::from([( + "my-service".to_string(), + build_inline_listener("my-service", "my-cluster"), + )]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_CDS, + HashMap::from([("my-cluster".to_string(), build_cluster("my-cluster"))]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_EDS, + HashMap::from([( + "my-cluster".to_string(), + build_endpoints( + "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. @@ -201,7 +260,174 @@ async fn xds_channel_e2e_routes_to_backend() { assert_eq!(counts.get(ADS_TYPE_URL_CDS), Some(&1)); assert_eq!(counts.get(ADS_TYPE_URL_EDS), Some(&1)); - // 6. Cleanup. let _ = backend.shutdown.send(()); cp_handle.abort(); } + +/// Analog of grpc-java's `changeClusterForRoute`: once the client is 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, cp_handle) = 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.set_xds_config( + ADS_TYPE_URL_LDS, + HashMap::from([( + "my-service".to_string(), + build_rds_listener("my-service", "route-config"), + )]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_RDS, + HashMap::from([( + "route-config".to_string(), + build_route_config("route-config", "cluster-a"), + )]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_CDS, + HashMap::from([ + ("cluster-a".to_string(), build_cluster("cluster-a")), + ("cluster-b".to_string(), build_cluster("cluster-b")), + ]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_EDS, + HashMap::from([ + ( + "cluster-a".to_string(), + build_endpoints( + "cluster-a", + &backend_a.addr.ip().to_string(), + backend_a.addr.port(), + ), + ), + ( + "cluster-b".to_string(), + build_endpoints( + "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.set_xds_config( + ADS_TYPE_URL_RDS, + HashMap::from([( + "route-config".to_string(), + 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(()); + cp_handle.abort(); +} + +/// 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 = 300; + + // 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, cp_handle) = start_control_plane().await; + + // Configure LDS (inline route) -> CDS -> EDS with all backends in one cluster. + control_plane.set_xds_config( + ADS_TYPE_URL_LDS, + HashMap::from([( + "my-service".to_string(), + build_inline_listener("my-service", "my-cluster"), + )]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_CDS, + HashMap::from([("my-cluster".to_string(), build_cluster("my-cluster"))]), + ); + control_plane.set_xds_config( + ADS_TYPE_URL_EDS, + HashMap::from([( + "my-cluster".to_string(), + build_endpoints_multi("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 = 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::(), NUM_REQUESTS); + let min_per_backend = NUM_REQUESTS / NUM_BACKENDS / 4; + 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(()); + } + cp_handle.abort(); +} From 805d6a123a949dd560d344e38eac5c2b4272a0f5 Mon Sep 17 00:00:00 2001 From: Jeff Jiang Date: Sat, 11 Jul 2026 03:02:12 -0700 Subject: [PATCH 3/6] simpler api --- tonic-xds/src/client/xds_e2e_test.rs | 44 ++++++----------- xds-test-util/Cargo.toml | 8 +-- xds-test-util/src/control_plane.rs | 74 ++++++++++++++++++++++++++++ xds-test-util/src/lib.rs | 19 +++---- 4 files changed, 102 insertions(+), 43 deletions(-) diff --git a/tonic-xds/src/client/xds_e2e_test.rs b/tonic-xds/src/client/xds_e2e_test.rs index 5cbbfad7b..06642358d 100644 --- a/tonic-xds/src/client/xds_e2e_test.rs +++ b/tonic-xds/src/client/xds_e2e_test.rs @@ -29,13 +29,9 @@ use envoy_types::pb::envoy::extensions::filters::network::http_connection_manage }; use envoy_types::pb::google::protobuf::Any; use prost::Message; -use tokio::net::TcpListener; -use tokio::task::JoinHandle; -use tokio_stream::wrappers::TcpListenerStream; -use tonic::transport::Server; use xds_test_util::{ - ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, ADS_TYPE_URL_RDS, - AggregatedDiscoveryServiceServer, XdsTestControlPlaneService, + ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, ADS_TYPE_URL_RDS, RunningControlPlane, + XdsTestControlPlaneService, }; use crate::testutil::grpc::{GreeterClient, HelloRequest, spawn_greeter_server}; @@ -156,24 +152,15 @@ fn lb_endpoint(host: &str, port: u16) -> LbEndpoint { } } -/// Starts the fake ADS control plane on an ephemeral port. Returns the service -/// handle (for injecting config), its address, and the server task handle. -async fn start_control_plane() -> ( - XdsTestControlPlaneService, - SocketAddr, - JoinHandle>, -) { - let control_plane = XdsTestControlPlaneService::new(); - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - let service = control_plane.clone(); - let handle = tokio::spawn(async move { - Server::builder() - .add_service(AggregatedDiscoveryServiceServer::new(service)) - .serve_with_incoming(TcpListenerStream::new(listener)) - .await - }); - (control_plane, addr, handle) +/// 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 @@ -224,7 +211,7 @@ async fn xds_channel_e2e_routes_to_backend() { .expect("spawn greeter backend"); let backend_addr = backend.addr; - let (control_plane, cp_addr, cp_handle) = start_control_plane().await; + let (control_plane, cp_addr) = start_control_plane().await; // Configure LDS (inline route) -> CDS -> EDS pointing at the backend. control_plane.set_xds_config( @@ -261,7 +248,6 @@ async fn xds_channel_e2e_routes_to_backend() { assert_eq!(counts.get(ADS_TYPE_URL_EDS), Some(&1)); let _ = backend.shutdown.send(()); - cp_handle.abort(); } /// Analog of grpc-java's `changeClusterForRoute`: once the client is routing to @@ -277,7 +263,7 @@ async fn xds_channel_e2e_route_update_shifts_traffic() { .await .expect("spawn backend-b"); - let (control_plane, cp_addr, cp_handle) = start_control_plane().await; + 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. @@ -346,7 +332,6 @@ async fn xds_channel_e2e_route_update_shifts_traffic() { let _ = backend_a.shutdown.send(()); let _ = backend_b.shutdown.send(()); - cp_handle.abort(); } /// P2C load balancing: with several EDS endpoints behind one cluster, traffic @@ -372,7 +357,7 @@ async fn xds_channel_e2e_p2c_spreads_across_backends() { .map(|backend| (backend.addr.ip().to_string(), backend.addr.port())) .collect(); - let (control_plane, cp_addr, cp_handle) = start_control_plane().await; + let (control_plane, cp_addr) = start_control_plane().await; // Configure LDS (inline route) -> CDS -> EDS with all backends in one cluster. control_plane.set_xds_config( @@ -429,5 +414,4 @@ async fn xds_channel_e2e_p2c_spreads_across_backends() { for backend in backends { let _ = backend.shutdown.send(()); } - cp_handle.abort(); } diff --git a/xds-test-util/Cargo.toml b/xds-test-util/Cargo.toml index 56b783eb9..d17cecdb4 100644 --- a/xds-test-util/Cargo.toml +++ b/xds-test-util/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "xds-test-util" -version = "0.1.0-alpha.1" +version = "0.1.0" edition = "2024" rust-version.workspace = true homepage = "https://github.com/grpc/grpc-rust" @@ -18,6 +18,6 @@ workspace = true [dependencies] envoy-types = "0.7" prost = "0.14" -tonic = "0.14" -tokio = { version = "1", features = ["sync", "rt"] } -tokio-stream = "0.1" +tonic = { version = "0.14", features = ["server"] } +tokio = { version = "1", features = ["sync", "rt", "net"] } +tokio-stream = { version = "0.1", features = ["net"] } diff --git a/xds-test-util/src/control_plane.rs b/xds-test-util/src/control_plane.rs index 501c8bafa..52e72a3fa 100644 --- a/xds-test-util/src/control_plane.rs +++ b/xds-test-util/src/control_plane.rs @@ -21,13 +21,19 @@ //! 3. an ACK (same resource names already subscribed) — ignored. use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; use std::sync::atomic::AtomicU64; use std::sync::{Arc, Mutex}; +use envoy_types::pb::envoy::service::discovery::v3::aggregated_discovery_service_server::AggregatedDiscoveryServiceServer; use envoy_types::pb::envoy::service::discovery::v3::{DiscoveryRequest, DiscoveryResponse}; use envoy_types::pb::google::protobuf::Any; use prost::Message; +use tokio::net::TcpListener; use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::transport::Server; /// ADS type URL for LDS (`Listener`) resources. pub const ADS_TYPE_URL_LDS: &str = "type.googleapis.com/envoy.config.listener.v3.Listener"; @@ -294,6 +300,74 @@ impl XdsTestControlPlaneService { }) .collect() } + + /// Serves the ADS control plane on an ephemeral `127.0.0.1` port in a + /// background task. + /// + /// Returns a [`RunningControlPlane`] that exposes the bound address via + /// [`addr`](RunningControlPlane::addr), derefs to this service so config + /// setters can be called on it directly, and shuts the server down when + /// dropped. + /// + /// # Errors + /// + /// Returns an error if binding the ephemeral port fails. + pub async fn start(&self) -> std::io::Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + let service = self.clone(); + let handle = tokio::spawn( + Server::builder() + .add_service(AggregatedDiscoveryServiceServer::new(service.clone())) + .serve_with_incoming(TcpListenerStream::new(listener)), + ); + Ok(RunningControlPlane { + service, + addr, + handle, + }) + } +} + +/// A running [`XdsTestControlPlaneService`] returned by +/// [`XdsTestControlPlaneService::start`]. +/// +/// Derefs to the underlying control plane, so config setters (e.g. +/// [`set_xds_config`](XdsTestControlPlaneService::set_xds_config)) can be called +/// on it directly. The server task is aborted when this handle is dropped. +#[derive(Debug)] +pub struct RunningControlPlane { + service: XdsTestControlPlaneService, + addr: SocketAddr, + handle: JoinHandle>, +} + +impl RunningControlPlane { + /// The address the ADS server is listening on. + #[must_use] + pub fn addr(&self) -> SocketAddr { + self.addr + } + + /// The underlying control plane service (shared state). + #[must_use] + pub fn service(&self) -> &XdsTestControlPlaneService { + &self.service + } +} + +impl std::ops::Deref for RunningControlPlane { + type Target = XdsTestControlPlaneService; + + fn deref(&self) -> &Self::Target { + &self.service + } +} + +impl Drop for RunningControlPlane { + fn drop(&mut self) { + self.handle.abort(); + } } /// Tonic serving adapter for the ADS bidi stream. diff --git a/xds-test-util/src/lib.rs b/xds-test-util/src/lib.rs index 259a8922a..167d8f8b5 100644 --- a/xds-test-util/src/lib.rs +++ b/xds-test-util/src/lib.rs @@ -10,19 +10,20 @@ mod control_plane; pub use control_plane::{ - ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, ADS_TYPE_URL_RDS, + ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, ADS_TYPE_URL_RDS, RunningControlPlane, XdsTestControlPlaneService, }; -/// Re-export of the generated ADS server adapter, so tests can register the -/// control plane with a [`tonic`] server without depending on `envoy-types` -/// module paths directly: +/// Re-export of the generated ADS server adapter, for advanced wiring — e.g. +/// registering the control plane on an existing [`tonic`] server alongside +/// other services. For the common case, prefer +/// [`XdsTestControlPlaneService::start`], which serves on an ephemeral port and +/// returns a [`RunningControlPlane`]: /// /// ```ignore -/// let service = XdsTestControlPlaneService::new(); -/// Server::builder() -/// .add_service(AggregatedDiscoveryServiceServer::new(service.clone())) -/// .serve(addr) -/// .await?; +/// let running = XdsTestControlPlaneService::new().start().await?; +/// running.set_xds_config(ADS_TYPE_URL_LDS, listeners); +/// let addr = running.addr(); // point your xDS bootstrap here +/// // `running` shuts the server down when dropped. /// ``` pub use envoy_types::pb::envoy::service::discovery::v3::aggregated_discovery_service_server::AggregatedDiscoveryServiceServer; From d2bbeed143d83e7f90365d21377608d298e50852 Mon Sep 17 00:00:00 2001 From: Jeff Jiang Date: Tue, 14 Jul 2026 07:02:44 -0700 Subject: [PATCH 4/6] refactor --- tonic-xds/src/client/mod.rs | 2 +- tonic-xds/src/client/xds_e2e.rs | 288 ++++++++++++++++++ tonic-xds/src/client/xds_e2e_test.rs | 417 --------------------------- xds-test-util/Cargo.toml | 2 + xds-test-util/src/config.rs | 152 ++++++++++ xds-test-util/src/control_plane.rs | 200 +++++++------ xds-test-util/src/lib.rs | 27 +- 7 files changed, 562 insertions(+), 526 deletions(-) create mode 100644 tonic-xds/src/client/xds_e2e.rs delete mode 100644 tonic-xds/src/client/xds_e2e_test.rs create mode 100644 xds-test-util/src/config.rs diff --git a/tonic-xds/src/client/mod.rs b/tonic-xds/src/client/mod.rs index 1687f8bcb..c69f71e6d 100644 --- a/tonic-xds/src/client/mod.rs +++ b/tonic-xds/src/client/mod.rs @@ -9,4 +9,4 @@ pub(crate) mod retry; pub(crate) mod route; #[cfg(test)] -mod xds_e2e_test; +mod xds_e2e; diff --git a/tonic-xds/src/client/xds_e2e.rs b/tonic-xds/src/client/xds_e2e.rs new file mode 100644 index 000000000..f85db9cb2 --- /dev/null +++ b/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, + 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 = 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::(), 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(()); + } + } +} diff --git a/tonic-xds/src/client/xds_e2e_test.rs b/tonic-xds/src/client/xds_e2e_test.rs deleted file mode 100644 index 06642358d..000000000 --- a/tonic-xds/src/client/xds_e2e_test.rs +++ /dev/null @@ -1,417 +0,0 @@ -//! 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. - -use std::collections::HashMap; -use std::net::SocketAddr; -use std::time::Duration; - -use envoy_types::pb::envoy::config::cluster::v3::Cluster; -use envoy_types::pb::envoy::config::cluster::v3::cluster::{ClusterDiscoveryType, DiscoveryType}; -use envoy_types::pb::envoy::config::core::v3 as core_v3; -use envoy_types::pb::envoy::config::endpoint::v3::{ - ClusterLoadAssignment, Endpoint, LbEndpoint, LocalityLbEndpoints, lb_endpoint::HostIdentifier, -}; -use envoy_types::pb::envoy::config::listener::v3::{ApiListener, Listener}; -use envoy_types::pb::envoy::config::route::v3::route::Action; -use envoy_types::pb::envoy::config::route::v3::route_action::ClusterSpecifier; -use envoy_types::pb::envoy::config::route::v3::route_match::PathSpecifier; -use envoy_types::pb::envoy::config::route::v3::{ - Route, RouteAction, RouteConfiguration, RouteMatch, VirtualHost, -}; -use envoy_types::pb::envoy::extensions::filters::network::http_connection_manager::v3::{ - HttpConnectionManager, Rds, http_connection_manager::RouteSpecifier, -}; -use envoy_types::pb::google::protobuf::Any; -use prost::Message; -use xds_test_util::{ - ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, ADS_TYPE_URL_RDS, RunningControlPlane, - XdsTestControlPlaneService, -}; - -use crate::testutil::grpc::{GreeterClient, HelloRequest, spawn_greeter_server}; -use crate::{BootstrapConfig, XdsChannelBuilder, XdsChannelConfig, XdsChannelGrpc, XdsUri}; - -const TYPE_HCM: &str = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager"; - -/// Wraps a `HttpConnectionManager` route specifier in an `ApiListener` listener. -fn build_listener(name: &str, route: RouteSpecifier) -> Listener { - let hcm = HttpConnectionManager { - route_specifier: Some(route), - ..Default::default() - }; - Listener { - name: name.to_string(), - api_listener: Some(ApiListener { - api_listener: Some(Any { - type_url: TYPE_HCM.to_string(), - value: hcm.encode_to_vec(), - }), - }), - ..Default::default() - } -} - -/// Builds a listener whose route configuration is embedded inline and sends all -/// traffic to `cluster`. -fn build_inline_listener(name: &str, cluster: &str) -> Listener { - build_listener( - name, - RouteSpecifier::RouteConfig(build_route_config(&format!("{name}-route"), cluster)), - ) -} - -/// Builds a listener that fetches its route configuration via RDS by name. -fn build_rds_listener(name: &str, rds_name: &str) -> Listener { - build_listener( - name, - RouteSpecifier::Rds(Rds { - route_config_name: rds_name.to_string(), - ..Default::default() - }), - ) -} - -/// Builds a route configuration that sends all traffic to `cluster`. -fn build_route_config(name: &str, cluster: &str) -> RouteConfiguration { - RouteConfiguration { - name: name.to_string(), - virtual_hosts: vec![VirtualHost { - name: "default".to_string(), - domains: vec!["*".to_string()], - routes: vec![Route { - r#match: Some(RouteMatch { - path_specifier: Some(PathSpecifier::Prefix("/".to_string())), - ..Default::default() - }), - action: Some(Action::Route(RouteAction { - cluster_specifier: Some(ClusterSpecifier::Cluster(cluster.to_string())), - ..Default::default() - })), - ..Default::default() - }], - ..Default::default() - }], - ..Default::default() - } -} - -/// Builds an EDS-discovered cluster named `name`. -fn build_cluster(name: &str) -> Cluster { - Cluster { - name: name.to_string(), - cluster_discovery_type: Some(ClusterDiscoveryType::Type(DiscoveryType::Eds as i32)), - ..Default::default() - } -} - -/// Builds a `ClusterLoadAssignment` for `cluster` with a single endpoint. -fn build_endpoints(cluster: &str, host: &str, port: u16) -> ClusterLoadAssignment { - build_endpoints_multi(cluster, &[(host.to_string(), port)]) -} - -/// Builds a `ClusterLoadAssignment` for `cluster` with the given endpoints, all -/// placed in a single locality. -fn build_endpoints_multi(cluster: &str, endpoints: &[(String, u16)]) -> ClusterLoadAssignment { - ClusterLoadAssignment { - cluster_name: cluster.to_string(), - endpoints: vec![LocalityLbEndpoints { - lb_endpoints: endpoints - .iter() - .map(|(host, port)| lb_endpoint(host, *port)) - .collect(), - ..Default::default() - }], - ..Default::default() - } -} - -/// Builds a single `LbEndpoint` at `host:port`. -fn lb_endpoint(host: &str, port: u16) -> LbEndpoint { - LbEndpoint { - host_identifier: Some(HostIdentifier::Endpoint(Endpoint { - address: Some(core_v3::Address { - address: Some(core_v3::address::Address::SocketAddress( - core_v3::SocketAddress { - address: host.to_string(), - port_specifier: Some(core_v3::socket_address::PortSpecifier::PortValue( - u32::from(port), - )), - ..Default::default() - }, - )), - }), - ..Default::default() - })), - ..Default::default() - } -} - -/// 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, - want_prefix: &str, -) -> String { - let mut last = None; - for _ in 0..50 { - 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:?}"); -} - -#[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.set_xds_config( - ADS_TYPE_URL_LDS, - HashMap::from([( - "my-service".to_string(), - build_inline_listener("my-service", "my-cluster"), - )]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_CDS, - HashMap::from([("my-cluster".to_string(), build_cluster("my-cluster"))]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_EDS, - HashMap::from([( - "my-cluster".to_string(), - build_endpoints( - "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_subscriber_counts(); - assert_eq!(counts.get(ADS_TYPE_URL_LDS), Some(&1)); - assert_eq!(counts.get(ADS_TYPE_URL_CDS), Some(&1)); - assert_eq!(counts.get(ADS_TYPE_URL_EDS), Some(&1)); - - let _ = backend.shutdown.send(()); -} - -/// Analog of grpc-java's `changeClusterForRoute`: once the client is 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.set_xds_config( - ADS_TYPE_URL_LDS, - HashMap::from([( - "my-service".to_string(), - build_rds_listener("my-service", "route-config"), - )]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_RDS, - HashMap::from([( - "route-config".to_string(), - build_route_config("route-config", "cluster-a"), - )]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_CDS, - HashMap::from([ - ("cluster-a".to_string(), build_cluster("cluster-a")), - ("cluster-b".to_string(), build_cluster("cluster-b")), - ]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_EDS, - HashMap::from([ - ( - "cluster-a".to_string(), - build_endpoints( - "cluster-a", - &backend_a.addr.ip().to_string(), - backend_a.addr.port(), - ), - ), - ( - "cluster-b".to_string(), - build_endpoints( - "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.set_xds_config( - ADS_TYPE_URL_RDS, - HashMap::from([( - "route-config".to_string(), - 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 = 300; - - // 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.set_xds_config( - ADS_TYPE_URL_LDS, - HashMap::from([( - "my-service".to_string(), - build_inline_listener("my-service", "my-cluster"), - )]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_CDS, - HashMap::from([("my-cluster".to_string(), build_cluster("my-cluster"))]), - ); - control_plane.set_xds_config( - ADS_TYPE_URL_EDS, - HashMap::from([( - "my-cluster".to_string(), - build_endpoints_multi("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 = 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::(), NUM_REQUESTS); - let min_per_backend = NUM_REQUESTS / NUM_BACKENDS / 4; - 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(()); - } -} diff --git a/xds-test-util/Cargo.toml b/xds-test-util/Cargo.toml index d17cecdb4..8308b82b9 100644 --- a/xds-test-util/Cargo.toml +++ b/xds-test-util/Cargo.toml @@ -21,3 +21,5 @@ 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" diff --git a/xds-test-util/src/config.rs b/xds-test-util/src/config.rs new file mode 100644 index 000000000..2c4a9290e --- /dev/null +++ b/xds-test-util/src/config.rs @@ -0,0 +1,152 @@ + +//! Helper functions for building xDS configuration resources for testing. + +use envoy_types::pb::envoy::config::cluster::v3::Cluster; +use envoy_types::pb::envoy::config::cluster::v3::cluster::{ClusterDiscoveryType, DiscoveryType}; +use envoy_types::pb::envoy::config::core::v3 as core_v3; +use envoy_types::pb::envoy::config::endpoint::v3::{ + ClusterLoadAssignment, Endpoint, LbEndpoint, LocalityLbEndpoints, lb_endpoint::HostIdentifier, +}; +use envoy_types::pb::envoy::config::listener::v3::{ApiListener, Listener}; +use envoy_types::pb::envoy::config::route::v3::route::Action; +use envoy_types::pb::envoy::config::route::v3::route_action::ClusterSpecifier; +use envoy_types::pb::envoy::config::route::v3::route_match::PathSpecifier; +use envoy_types::pb::envoy::config::route::v3::{ + Route, RouteAction, RouteConfiguration, RouteMatch, VirtualHost, +}; +use envoy_types::pb::envoy::extensions::filters::network::http_connection_manager::v3::{ + HttpConnectionManager, Rds, http_connection_manager::RouteSpecifier, +}; +use envoy_types::pb::google::protobuf::Any; +use prost::Message; +use strum::{Display, EnumIter, EnumString, IntoStaticStr}; + +/// ADS type URLs for the different xDS resource types. +#[derive(Debug, Display, EnumIter, IntoStaticStr, EnumString, PartialEq, Eq, Hash, Clone)] +pub enum AdsTypeUrl { + /// Listener Discovery Service (LDS) type URL. + #[strum(serialize = "type.googleapis.com/envoy.config.listener.v3.Listener")] + Lds, + /// Route Discovery Service (RDS) type URL. + #[strum(serialize = "type.googleapis.com/envoy.config.route.v3.RouteConfiguration")] + Rds, + /// Cluster Discovery Service (CDS) type URL. + #[strum(serialize = "type.googleapis.com/envoy.config.cluster.v3.Cluster")] + Cds, + /// Endpoint Discovery Service (EDS) type URL. + #[strum(serialize = "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment")] + Eds, + /// HTTP Connection Manager (HCM) type URL. + #[strum( + serialize = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager" + )] + Hcm, +} + +/// Wraps a `HttpConnectionManager` route specifier in an `ApiListener` listener. +fn build_listener(name: &str, route: RouteSpecifier) -> Listener { + let hcm = HttpConnectionManager { + route_specifier: Some(route), + ..Default::default() + }; + Listener { + name: name.to_string(), + api_listener: Some(ApiListener { + api_listener: Some(Any { + type_url: AdsTypeUrl::Hcm.to_string(), + value: hcm.encode_to_vec(), + }), + }), + ..Default::default() + } +} + +/// Builds a listener whose route configuration is embedded inline and sends all +/// traffic to `cluster`. +pub fn build_inline_listener(name: &str, cluster: &str) -> Listener { + build_listener( + name, + RouteSpecifier::RouteConfig(build_route_config(&format!("{name}-route"), cluster)), + ) +} + +/// Builds a listener that fetches its route configuration via RDS by name. +pub fn build_rds_listener(name: &str, rds_name: &str) -> Listener { + build_listener( + name, + RouteSpecifier::Rds(Rds { + route_config_name: rds_name.to_string(), + ..Default::default() + }), + ) +} + +/// Builds a route configuration that sends all traffic to `cluster`. +pub fn build_route_config(name: &str, cluster: &str) -> RouteConfiguration { + RouteConfiguration { + name: name.to_string(), + virtual_hosts: vec![VirtualHost { + name: "default".to_string(), + domains: vec!["*".to_string()], + routes: vec![Route { + r#match: Some(RouteMatch { + path_specifier: Some(PathSpecifier::Prefix("/".to_string())), + ..Default::default() + }), + action: Some(Action::Route(RouteAction { + cluster_specifier: Some(ClusterSpecifier::Cluster(cluster.to_string())), + ..Default::default() + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + } +} + +/// Builds an EDS-discovered cluster named `name`. +pub fn build_cluster(name: &str) -> Cluster { + Cluster { + name: name.to_string(), + cluster_discovery_type: Some(ClusterDiscoveryType::Type(DiscoveryType::Eds as i32)), + ..Default::default() + } +} + +/// Builds a `ClusterLoadAssignment` for `cluster` with the given endpoints, all +/// placed in a single locality. +pub fn build_cla(cluster: &str, endpoints: &[(String, u16)]) -> ClusterLoadAssignment { + ClusterLoadAssignment { + cluster_name: cluster.to_string(), + endpoints: vec![LocalityLbEndpoints { + lb_endpoints: endpoints + .iter() + .map(|(host, port)| lb_endpoint(host, *port)) + .collect(), + ..Default::default() + }], + ..Default::default() + } +} + +/// Builds a single `LbEndpoint` at `host:port`. +fn lb_endpoint(host: &str, port: u16) -> LbEndpoint { + LbEndpoint { + host_identifier: Some(HostIdentifier::Endpoint(Endpoint { + address: Some(core_v3::Address { + address: Some(core_v3::address::Address::SocketAddress( + core_v3::SocketAddress { + address: host.to_string(), + port_specifier: Some(core_v3::socket_address::PortSpecifier::PortValue( + u32::from(port), + )), + ..Default::default() + }, + )), + }), + ..Default::default() + })), + ..Default::default() + } +} diff --git a/xds-test-util/src/control_plane.rs b/xds-test-util/src/control_plane.rs index 52e72a3fa..82a931031 100644 --- a/xds-test-util/src/control_plane.rs +++ b/xds-test-util/src/control_plane.rs @@ -1,62 +1,32 @@ -//! A fake xDS ADS control plane for tests. +//! `XdsTestControlPlaneService` is a fake xDS ADS control plane for gRPC tests. //! -//! Rust port of grpc-java's `XdsTestControlPlaneService`. It is a bidi-stream -//! service that acts as a local xDS control plane. Config is injected through +//! The xDS config resources are injected through //! [`set_xds_config`](XdsTestControlPlaneService::set_xds_config). //! -//! The service maintains, per ADS resource type: -//! - a resources table (resource name to packed protobuf `Any`), -//! - a subscriber table (each active stream to its subscribed resource names), -//! - a version counter (bumped on every config set), and -//! - a per-stream nonce counter. -//! -//! All resource types are treated as state-of-the-world: whenever any resource -//! of a type changes, every subscriber to that type receives all of its -//! subscribed resources of that type. -//! -//! Incoming ADS requests share the same proto message but represent different -//! phases, which the service distinguishes: -//! 1. an initial request (new or changed subscription) — answered with a response, -//! 2. a NACK (carries `error_detail`) — logged and ignored, -//! 3. an ACK (same resource names already subscribed) — ignored. +//! The xDS config resources are served by ADS (Aggregated Discovery Service) protocol and +//! State-of-the-world (SOTW): whenever any resource of a type changes, every subscriber to +//! that type receives all of its subscribed resources of that type. use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; use std::sync::atomic::AtomicU64; use std::sync::{Arc, Mutex}; -use envoy_types::pb::envoy::service::discovery::v3::aggregated_discovery_service_server::AggregatedDiscoveryServiceServer; -use envoy_types::pb::envoy::service::discovery::v3::{DiscoveryRequest, DiscoveryResponse}; +use crate::config::*; +use envoy_types::pb::envoy::service::discovery::v3::{ + DiscoveryRequest, DiscoveryResponse, + aggregated_discovery_service_server::AggregatedDiscoveryServiceServer, +}; use envoy_types::pb::google::protobuf::Any; use prost::Message; +use strum::IntoEnumIterator; use tokio::net::TcpListener; use tokio::sync::mpsc; use tokio::task::JoinHandle; use tokio_stream::wrappers::TcpListenerStream; use tonic::transport::Server; -/// ADS type URL for LDS (`Listener`) resources. -pub const ADS_TYPE_URL_LDS: &str = "type.googleapis.com/envoy.config.listener.v3.Listener"; -/// ADS type URL for RDS (`RouteConfiguration`) resources. -pub const ADS_TYPE_URL_RDS: &str = "type.googleapis.com/envoy.config.route.v3.RouteConfiguration"; -/// ADS type URL for CDS (`Cluster`) resources. -pub const ADS_TYPE_URL_CDS: &str = "type.googleapis.com/envoy.config.cluster.v3.Cluster"; -/// ADS type URL for EDS (`ClusterLoadAssignment`) resources. -pub const ADS_TYPE_URL_EDS: &str = - "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"; - -/// The ADS resource types tracked by the control plane. -const ADS_TYPE_URLS: [&str; 4] = [ - ADS_TYPE_URL_LDS, - ADS_TYPE_URL_RDS, - ADS_TYPE_URL_CDS, - ADS_TYPE_URL_EDS, -]; - /// Sender half of a stream's outbound `DiscoveryResponse` channel. -/// -/// The channel carries bare `DiscoveryResponse`s; the (framework-specific) -/// serving adapter wraps them in the transport's success type. type ResponseSender = mpsc::UnboundedSender; /// Per-(type, stream) subscription bookkeeping. @@ -73,21 +43,21 @@ struct Subscription { /// Mutable control-plane state, guarded by a single mutex. #[derive(Debug)] struct State { - /// `type_url` to (resource name to packed `Any`). - resources: HashMap>, - /// `type_url` to latest version (starts at 1, bumped on each config set). - versions: HashMap, - /// `type_url` to (stream id to subscription). - subscribers: HashMap>, + /// xDS resources keyed by their names, organized by type URL. + resources: HashMap>, + /// Latest version number for each xDS resource type (starts at 1, bumped on each config set). + versions: HashMap, + /// Active subscribers for each xDS resource type, keyed by stream ID. + subscribers: HashMap>, } impl State { fn new() -> Self { let mut versions = HashMap::new(); let mut subscribers = HashMap::new(); - for type_url in ADS_TYPE_URLS { - versions.insert(type_url.to_string(), 1); - subscribers.insert(type_url.to_string(), HashMap::new()); + for type_url in AdsTypeUrl::iter() { + versions.insert(type_url.clone(), 1); + subscribers.insert(type_url.clone(), HashMap::new()); } Self { resources: HashMap::new(), @@ -111,13 +81,24 @@ impl Inner { // NACK: a request carrying an error detail rejects the last response. if req.error_detail.is_some() { + tracing::warn!( + "Received NACK for stream {}: {:?}", + stream_id, + req.error_detail + ); return; } - let type_url = req.type_url; - if type_url.is_empty() { - return; - } + let type_url: AdsTypeUrl = match req.type_url.parse() { + Ok(url) => url, + Err(_) => { + tracing::error!( + "Received request with empty or invalid type_url for stream {}", + stream_id + ); + return; + } + }; // Nonce check: if the request carries a response nonce, it must match // the last nonce we sent on this (type, stream); otherwise ignore it. @@ -128,9 +109,20 @@ impl Inner { .and_then(|streams| streams.get(&stream_id)) .is_some_and(|sub| sub.nonce.to_string() == req.response_nonce); if !matches { + tracing::warn!( + "Received request with mismatched nonce for stream {}: {}", + stream_id, + req.response_nonce + ); return; } } + tracing::debug!( + "Received ACK {} for stream {} with type_url {}", + req.response_nonce, + stream_id, + type_url + ); let requested: HashSet = req.resource_names.into_iter().collect(); @@ -141,6 +133,11 @@ impl Inner { .and_then(|streams| streams.get(&stream_id)) && sub.resource_names == requested { + tracing::debug!( + "Received identical subscription for stream {} with type_url {}", + stream_id, + type_url + ); return; } @@ -178,7 +175,7 @@ impl Inner { } } -/// A bidi-stream service that acts as a local xDS control plane for tests. +/// XdsTestControlPlaneService is a bidi-stream service that acts as a local xDS control plane for tests. /// /// Clone freely: all clones share the same underlying state, so config injected /// through one clone is served by another. Inject config with @@ -215,7 +212,7 @@ impl XdsTestControlPlaneService { /// /// Each resource is packed into a protobuf `Any` tagged with `type_url`. /// Resource types are treated as state-of-the-world. - pub fn set_xds_config(&self, type_url: &str, resources: HashMap) { + pub fn set_xds_config(&self, type_url: &AdsTypeUrl, resources: HashMap) { let packed: HashMap = resources .into_iter() .map(|(name, msg)| { @@ -229,12 +226,16 @@ impl XdsTestControlPlaneService { }) .collect(); - let mut state = self.inner.state.lock().expect("control plane state poisoned"); - state.resources.insert(type_url.to_string(), packed); + let mut state = self + .inner + .state + .lock() + .expect("control plane state poisoned"); + state.resources.insert(type_url.clone(), packed); // getAndIncrement: the pushed response carries the pre-increment version. let version = { - let counter = state.versions.entry(type_url.to_string()).or_insert(1); + let counter = state.versions.entry(type_url.clone()).or_insert(1); let current = *counter; *counter += 1; current @@ -245,12 +246,17 @@ impl XdsTestControlPlaneService { subscribers, .. } = &mut *state; - let type_resources = resources.get(type_url); - if let Some(streams) = subscribers.get_mut(type_url) { + let type_resources = resources.get(&type_url); + if let Some(streams) = subscribers.get_mut(&type_url) { for sub in streams.values_mut() { sub.nonce += 1; - let response = - build_response(type_url, version, sub.nonce, &sub.resource_names, type_resources); + let response = build_response( + type_url, + version, + sub.nonce, + &sub.resource_names, + type_resources, + ); let _ = sub.sender.send(response); } } @@ -262,8 +268,12 @@ impl XdsTestControlPlaneService { /// For a protobuf-free view suitable for assertions, use /// [`resource_names`](Self::resource_names). #[must_use] - pub fn get_current_config(&self, type_url: &str) -> HashMap { - let state = self.inner.state.lock().expect("control plane state poisoned"); + pub fn get_current_config(&self, type_url: &AdsTypeUrl) -> HashMap { + let state = self + .inner + .state + .lock() + .expect("control plane state poisoned"); state.resources.get(type_url).cloned().unwrap_or_default() } @@ -273,8 +283,12 @@ impl XdsTestControlPlaneService { /// This is a protocol-agnostic view of what the control plane is serving, /// suitable for assertions without depending on protobuf types. #[must_use] - pub fn resource_names(&self, type_url: &str) -> Vec { - let state = self.inner.state.lock().expect("control plane state poisoned"); + pub fn resource_names(&self, type_url: &AdsTypeUrl) -> Vec { + let state = self + .inner + .state + .lock() + .expect("control plane state poisoned"); state .resources .get(type_url) @@ -288,14 +302,20 @@ impl XdsTestControlPlaneService { /// Returns the number of active subscribers per ADS resource type. #[must_use] - pub fn get_subscriber_counts(&self) -> HashMap { - let state = self.inner.state.lock().expect("control plane state poisoned"); - ADS_TYPE_URLS - .iter() + pub fn get_subscriber_counts(&self) -> HashMap { + let state = self + .inner + .state + .lock() + .expect("control plane state poisoned"); + + state + .subscribers + .keys() .map(|type_url| { ( - (*type_url).to_string(), - state.subscribers.get(*type_url).map_or(0, HashMap::len), + type_url.clone(), + state.subscribers.get(type_url).map_or(0, HashMap::len), ) }) .collect() @@ -351,15 +371,7 @@ impl RunningControlPlane { /// The underlying control plane service (shared state). #[must_use] - pub fn service(&self) -> &XdsTestControlPlaneService { - &self.service - } -} - -impl std::ops::Deref for RunningControlPlane { - type Target = XdsTestControlPlaneService; - - fn deref(&self) -> &Self::Target { + pub fn get_service(&self) -> &XdsTestControlPlaneService { &self.service } } @@ -403,9 +415,13 @@ mod tonic_service { while let Some(item) = inbound.next().await { match item { Ok(req) => inner.handle_request(stream_id, &tx, req), - Err(_) => break, + Err(err) => { + tracing::error!("Error receiving request: {:?}", err); + break; + } } } + tracing::info!("Stream {} closed", stream_id); inner.remove_stream(stream_id); }); @@ -429,7 +445,7 @@ mod tonic_service { /// Builds a `DiscoveryResponse` containing the requested resources that exist. fn build_response( - type_url: &str, + type_url: &AdsTypeUrl, version: u64, nonce: u64, resource_names: &HashSet, @@ -468,19 +484,23 @@ mod tests { ..Default::default() }, ); - control_plane.set_xds_config(ADS_TYPE_URL_LDS, listeners); + control_plane.set_xds_config(&AdsTypeUrl::Lds, listeners); - let config = control_plane.get_current_config(ADS_TYPE_URL_LDS); + let config = control_plane.get_current_config(&AdsTypeUrl::Lds); assert_eq!(config.len(), 1); let packed = config.get("my-listener").expect("listener present"); - assert_eq!(packed.type_url, ADS_TYPE_URL_LDS); + assert_eq!(packed.type_url, AdsTypeUrl::Lds.to_string()); // No streams have connected yet. let counts = control_plane.get_subscriber_counts(); - assert_eq!(counts.len(), 4); - assert_eq!(counts.get(ADS_TYPE_URL_LDS), Some(&0)); + assert_eq!(counts.len(), AdsTypeUrl::iter().count()); + assert_eq!(counts.get(&AdsTypeUrl::Lds), Some(&0)); // A type that was never configured has no resources. - assert!(control_plane.get_current_config(ADS_TYPE_URL_CDS).is_empty()); + assert!( + control_plane + .get_current_config(&AdsTypeUrl::Cds) + .is_empty() + ); } } diff --git a/xds-test-util/src/lib.rs b/xds-test-util/src/lib.rs index 167d8f8b5..e841e4a3d 100644 --- a/xds-test-util/src/lib.rs +++ b/xds-test-util/src/lib.rs @@ -6,24 +6,15 @@ //! The main entry point is [`XdsTestControlPlaneService`], a fake Aggregated //! Discovery Service (ADS) control plane. It is a Rust port of grpc-java's //! `XdsTestControlPlaneService`. +//! +//! ```ignore +//! let running = XdsTestControlPlaneService::new().start().await?; +//! running.set_xds_config(&AdsTypeUrl::Lds, listeners); +//! let addr = running.addr(); // point your xDS bootstrap here +//! // `running` shuts the server down when dropped. +//! ``` +pub mod config; mod control_plane; -pub use control_plane::{ - ADS_TYPE_URL_CDS, ADS_TYPE_URL_EDS, ADS_TYPE_URL_LDS, ADS_TYPE_URL_RDS, RunningControlPlane, - XdsTestControlPlaneService, -}; - -/// Re-export of the generated ADS server adapter, for advanced wiring — e.g. -/// registering the control plane on an existing [`tonic`] server alongside -/// other services. For the common case, prefer -/// [`XdsTestControlPlaneService::start`], which serves on an ephemeral port and -/// returns a [`RunningControlPlane`]: -/// -/// ```ignore -/// let running = XdsTestControlPlaneService::new().start().await?; -/// running.set_xds_config(ADS_TYPE_URL_LDS, listeners); -/// let addr = running.addr(); // point your xDS bootstrap here -/// // `running` shuts the server down when dropped. -/// ``` -pub use envoy_types::pb::envoy::service::discovery::v3::aggregated_discovery_service_server::AggregatedDiscoveryServiceServer; +pub use control_plane::{RunningControlPlane, XdsTestControlPlaneService}; From 74a4e9c5acbf40962db172e0aa6e0296bec691e9 Mon Sep 17 00:00:00 2001 From: Jeff Jiang Date: Tue, 14 Jul 2026 07:37:53 -0700 Subject: [PATCH 5/6] clippy --- xds-test-util/src/control_plane.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/xds-test-util/src/control_plane.rs b/xds-test-util/src/control_plane.rs index 82a931031..8690944b8 100644 --- a/xds-test-util/src/control_plane.rs +++ b/xds-test-util/src/control_plane.rs @@ -246,8 +246,8 @@ impl XdsTestControlPlaneService { subscribers, .. } = &mut *state; - let type_resources = resources.get(&type_url); - if let Some(streams) = subscribers.get_mut(&type_url) { + let type_resources = resources.get(type_url); + if let Some(streams) = subscribers.get_mut(type_url) { for sub in streams.values_mut() { sub.nonce += 1; let response = build_response( From cd90078ca0e6ed1cf67cc2671f16c26405ef98bb Mon Sep 17 00:00:00 2001 From: Jeff Jiang Date: Tue, 14 Jul 2026 07:54:18 -0700 Subject: [PATCH 6/6] fmt --- xds-test-util/src/config.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/xds-test-util/src/config.rs b/xds-test-util/src/config.rs index 2c4a9290e..fb4ef6217 100644 --- a/xds-test-util/src/config.rs +++ b/xds-test-util/src/config.rs @@ -1,4 +1,3 @@ - //! Helper functions for building xDS configuration resources for testing. use envoy_types::pb::envoy::config::cluster::v3::Cluster;