diff --git a/Cargo.toml b/Cargo.toml index 3a4e1d3ab..23f4b37af 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,6 +21,7 @@ members = [ "xds-client", "xds-client-opentelemetry", "tonic-xds", + "xds-test-util", "interop", "tests/disable_comments", "tests/wellknown", diff --git a/tonic-xds/Cargo.toml b/tonic-xds/Cargo.toml index 3502905cc..14d009365 100644 --- a/tonic-xds/Cargo.toml +++ b/tonic-xds/Cargo.toml @@ -63,6 +63,7 @@ workspace = true [dev-dependencies] xds-client = { version = "0.1.0-alpha.2", path = "../xds-client", features = ["test-util"] } +xds-test-util = { path = "../xds-test-util" } tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "test-util"] } tonic = { version = "0.14", features = [ "server", "channel", "tls-ring" ] } tonic-prost = "0.14" diff --git a/tonic-xds/src/client/mod.rs b/tonic-xds/src/client/mod.rs index 3e02c9b29..c69f71e6d 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; 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/xds-test-util/Cargo.toml b/xds-test-util/Cargo.toml new file mode 100644 index 000000000..8308b82b9 --- /dev/null +++ b/xds-test-util/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "xds-test-util" +version = "0.1.0" +edition = "2024" +rust-version.workspace = true +homepage = "https://github.com/grpc/grpc-rust" +repository = "https://github.com/grpc/grpc-rust" +description = """ +Test utilities for gRPC xDS implementations. +""" +keywords = ["grpc", "xds", "test"] +license = "MIT" +publish = false + +[lints] +workspace = true + +[dependencies] +envoy-types = "0.7" +prost = "0.14" +tonic = { version = "0.14", features = ["server"] } +tokio = { version = "1", features = ["sync", "rt", "net"] } +tokio-stream = { version = "0.1", features = ["net"] } +strum = { version = "0.28.0", features = ["derive"] } +tracing = "0.1.44" diff --git a/xds-test-util/src/config.rs b/xds-test-util/src/config.rs new file mode 100644 index 000000000..fb4ef6217 --- /dev/null +++ b/xds-test-util/src/config.rs @@ -0,0 +1,151 @@ +//! 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 new file mode 100644 index 000000000..8690944b8 --- /dev/null +++ b/xds-test-util/src/control_plane.rs @@ -0,0 +1,506 @@ +//! `XdsTestControlPlaneService` is a fake xDS ADS control plane for gRPC tests. +//! +//! The xDS config resources are injected through +//! [`set_xds_config`](XdsTestControlPlaneService::set_xds_config). +//! +//! 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 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; + +/// Sender half of a stream's outbound `DiscoveryResponse` channel. +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 { + /// 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 AdsTypeUrl::iter() { + versions.insert(type_url.clone(), 1); + subscribers.insert(type_url.clone(), 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() { + tracing::warn!( + "Received NACK for stream {}: {:?}", + stream_id, + req.error_detail + ); + 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. + 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 { + 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(); + + // 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 + { + tracing::debug!( + "Received identical subscription for stream {} with type_url {}", + stream_id, + type_url + ); + 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); + } + } +} + +/// 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 +/// [`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: &AdsTypeUrl, 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.clone(), packed); + + // getAndIncrement: the pushed response carries the pre-increment version. + let version = { + let counter = state.versions.entry(type_url.clone()).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: &AdsTypeUrl) -> 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: &AdsTypeUrl) -> 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"); + + state + .subscribers + .keys() + .map(|type_url| { + ( + type_url.clone(), + state.subscribers.get(type_url).map_or(0, HashMap::len), + ) + }) + .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 get_service(&self) -> &XdsTestControlPlaneService { + &self.service + } +} + +impl Drop for RunningControlPlane { + fn drop(&mut self) { + self.handle.abort(); + } +} + +/// 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(err) => { + tracing::error!("Error receiving request: {:?}", err); + break; + } + } + } + tracing::info!("Stream {} closed", stream_id); + 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: &AdsTypeUrl, + 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(&AdsTypeUrl::Lds, listeners); + + 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, AdsTypeUrl::Lds.to_string()); + + // No streams have connected yet. + let counts = control_plane.get_subscriber_counts(); + 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(&AdsTypeUrl::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..e841e4a3d --- /dev/null +++ b/xds-test-util/src/lib.rs @@ -0,0 +1,20 @@ +//! 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`. +//! +//! ```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::{RunningControlPlane, XdsTestControlPlaneService};