diff --git a/tonic-xds/src/xds/resource_manager.rs b/tonic-xds/src/xds/resource_manager.rs index 7f646d0a0..98543cc10 100644 --- a/tonic-xds/src/xds/resource_manager.rs +++ b/tonic-xds/src/xds/resource_manager.rs @@ -617,4 +617,69 @@ mod tests { assert!(state.rds_watcher.is_some()); assert_eq!(state.rds_name.as_deref(), Some("rc")); } + + /// End-to-end replica of the production Istio hang: an RDS update that + /// references far more clusters than the xds-client worker's command + /// channel buffers (64), reconciled through the real cascade path — + /// `handle_rds` holds the event's `ProcessingDone` token while + /// `reconcile_clusters` awaits one `watch()` per cluster. + /// + /// Before xds-client gained ADS flow control, the worker sat inside + /// `handle_response` awaiting that token and stopped draining commands; + /// once the command channel filled, `reconcile_clusters` blocked on the + /// 65th watch and the client deadlocked. This test times out under that + /// behavior and completes under the fixed worker. + #[tokio::test] + async fn rds_referencing_many_clusters_reconciles_without_deadlock() { + use std::time::Duration; + use xds_client::{ + ClientConfig, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, + XdsClient as RealXdsClient, + }; + use xds_test_util::{XdsTestControlPlaneService, config}; + + // Well past the worker's 64-slot command channel. + const CLUSTER_COUNT: usize = 100; + + let control_plane = XdsTestControlPlaneService::new() + .start() + .await + .expect("control plane failed to start"); + control_plane.get_service().set_xds_config( + &config::AdsTypeUrl::Rds, + HashMap::from([( + "rc-many".to_string(), + config::build_route_config_with_cluster_count("rc-many", CLUSTER_COUNT), + )]), + ); + + let client_config = ClientConfig::new( + Node::new("test", "0"), + format!("http://{}", control_plane.addr()), + ); + let xds_client: RealXdsClient = RealXdsClient::builder( + client_config, + TonicTransportBuilder::new(), + ProstCodec, + TokioRuntime, + ) + .build(); + + let mut watcher = xds_client.watch::("rc-many").await; + let event = tokio::time::timeout(Duration::from_secs(10), watcher.next()) + .await + .expect("timed out waiting for the RDS update") + .expect("watcher closed"); + + let cache = test_cache(); + let mut state = CascadeState::new(); + tokio::time::timeout( + Duration::from_secs(10), + state.handle_rds(event, &xds_client, &cache), + ) + .await + .expect("deadlock: reconcile_clusters blocked while the worker awaited ProcessingDone"); + + assert_eq!(state.cds_watchers.len(), CLUSTER_COUNT); + } } diff --git a/xds-client/src/client/watch.rs b/xds-client/src/client/watch.rs index c2c1e3304..cb553dceb 100644 --- a/xds-client/src/client/watch.rs +++ b/xds-client/src/client/watch.rs @@ -36,9 +36,13 @@ use crate::resource::{DecodedResource, Resource}; /// A signal to indicate that processing of a resource event is complete. /// -/// The xDS client waits for this signal before sending ACK/NACK to the server. -/// This allows watchers to add cascading subscriptions (e.g. LDS -> RDS -> CDS -> EDS) -/// that will be included in the same ACK. +/// The xDS client does not read the next response from the ADS stream until +/// every watcher has signaled for the current one (ADS flow control, per +/// gRFC A88). This lets watchers apply an update — including adding cascading +/// subscriptions (e.g. LDS -> RDS -> CDS -> EDS) — before the next update can +/// arrive. It does *not* delay the ACK/NACK, and the worker keeps processing +/// watch/unwatch commands while waiting, so holding the token cannot deadlock +/// the client. /// /// # Automatic Signaling /// diff --git a/xds-client/src/client/worker.rs b/xds-client/src/client/worker.rs index 070a7cb20..eafa3b3a6 100644 --- a/xds-client/src/client/worker.rs +++ b/xds-client/src/client/worker.rs @@ -32,6 +32,8 @@ use std::collections::hash_map::Entry; use std::collections::{HashMap, HashSet}; +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; @@ -573,6 +575,22 @@ pub(crate) struct AdsWorker { recorder: RecorderHandle, } +/// A watcher notification staged during response handling: the channel to +/// deliver on, the event (carrying its `ProcessingDone` token), and — for +/// events that participate in ADS flow control — the receiver that resolves +/// when the watcher finishes processing. +type Delivery = ( + mpsc::Sender>, + ResourceEvent, + Option>, +); + +/// In-flight watcher deliveries for one response: sends every staged event +/// (with backpressure) and resolves once all watchers signal `ProcessingDone`. +/// While unresolved it gates reading the *next* response, but nothing else — +/// see [`AdsWorker::run_connected`]. +type PendingDispatch = Pin + Send>>; + /// Outcome of a connected ADS session (see [`AdsWorker::run_connected`]). enum ConnectedOutcome { /// All `XdsClient` handles were dropped; the worker should shut down. @@ -762,14 +780,23 @@ where // A78 a stream that fails *after* receiving a response is not counted as // a server failure. let mut saw_response = false; + // Watcher deliveries for the last response, still in flight (ADS flow + // control, per gRFC A88). While `Some`, the next response is not read + // — but commands keep draining below. Awaiting the deliveries inline + // in `handle_response` instead would freeze the whole loop: a watcher + // that issues commands (e.g. cascading watches) while holding its + // `ProcessingDone` token could fill the command channel and deadlock + // against a worker that only resumes once that token drops. + let mut pending: Option = None; loop { tokio::select! { - result = stream.recv() => { + result = stream.recv(), if pending.is_none() => { match result { Ok(Some(bytes)) => { saw_response = true; - if self.handle_response(&mut stream, bytes).await.is_err() { - return ConnectedOutcome::Failed { saw_response }; + match self.handle_response(&mut stream, bytes).await { + Ok(deliveries) => pending = Self::dispatch_pending(deliveries), + Err(_) => return ConnectedOutcome::Failed { saw_response }, } } // Stream closed by server or errored; reconnect. @@ -777,6 +804,11 @@ where } } + // `unwrap` is safe: the branch is disabled when `pending` is `None`. + _ = async { pending.as_mut().unwrap().await }, if pending.is_some() => { + pending = None; + } + cmd = self.command_rx.recv() => { match cmd { Some(cmd) => { @@ -791,6 +823,29 @@ where } } + /// Build the in-flight delivery future for one response's staged watcher + /// notifications, or `None` when there is nothing to deliver. + fn dispatch_pending(deliveries: Vec) -> Option { + if deliveries.is_empty() { + return None; + } + Some(Box::pin(async move { + let mut done_rxs = Vec::with_capacity(deliveries.len()); + for (event_tx, event, done_rx) in deliveries { + // Backpressure: await if the watcher's channel is full. Send + // errors (watcher dropped) are ignored; the rejected event's + // `ProcessingDone` token resolves its receiver on drop. + let _ = event_tx.send(event).await; + if let Some(rx) = done_rx { + done_rxs.push(rx); + } + } + for rx in done_rxs { + let _ = rx.await; + } + })) + } + /// Handle a command, optionally sending network requests if connected. /// /// When `stream` is `None`, only state updates are performed (disconnected mode). @@ -974,18 +1029,24 @@ where /// - Valid resources are cached and dispatched to watchers /// - Invalid resources are cached as NACKed and errors sent to specific watchers /// - Missing resources (for types with ALL_RESOURCES_REQUIRED_IN_SOTW) are marked deleted + /// + /// Cache/state updates and the ACK/NACK happen here; the watcher + /// notifications are only *staged* and returned as [`Delivery`]s for + /// `run_connected` to drive, so a slow (or stuck) watcher delays reading + /// the next response — ADS flow control — without freezing command + /// processing. async fn handle_response( &mut self, stream: &mut S, bytes: Bytes, - ) -> Result<()> { + ) -> Result> { let response = self.codec.decode_response(bytes)?; let type_url = response.type_url.clone(); let (type_url_arc, decoder) = match self.type_states.get(&type_url) { Some(s) => (Arc::clone(&s.type_url), &s.decoder), None => { - return Ok(()); + return Ok(Vec::new()); } }; @@ -1028,27 +1089,25 @@ where .map(|r| r.name().to_string()) .collect(); - let mut processing_done_futures = self.dispatch_resources(&type_url, valid_resources).await; + // Stage watcher notifications instead of sending them here: the sends + // (and the ProcessingDone waits) happen in the returned deliveries, + // driven by `run_connected` concurrently with command processing. + // State/cache updates still happen synchronously below, so the ACK + // reflects the accepted config regardless of watcher progress. + let mut deliveries = Vec::new(); + + self.dispatch_resources(&mut deliveries, &type_url, valid_resources); // Only notify watchers for per-resource errors (where we know the name). // Top-level errors have no associated name, so no watcher to notify. for (resource_name, error) in &per_resource_errors { - self.notify_resource_error(&type_url, resource_name, error) - .await; + self.notify_resource_error(&mut deliveries, &type_url, resource_name, error); } // Detect deleted resources (per A53): // For resource types with ALL_RESOURCES_REQUIRED_IN_SOTW = true, // any previously-received resource not in this response is deleted. - let deleted_futures = self - .detect_deleted_resources(&type_url, &received_names) - .await; - processing_done_futures.extend(deleted_futures); - - // Wait for all watchers to finish processing. - for rx in processing_done_futures { - let _ = rx.await; - } + self.detect_deleted_resources(&mut deliveries, &type_url, &received_names); let has_errors = !top_level_errors.is_empty() || !per_resource_errors.is_empty(); if !has_errors { @@ -1057,7 +1116,7 @@ where if let Some(ts) = self.type_states.get_mut(&type_url) { ts.version_info = response.version_info.clone(); } - self.send_ack(stream, &response).await + self.send_ack(stream, &response).await?; } else { // Build NACK message combining both error categories let mut error_parts = Vec::new(); @@ -1076,21 +1135,23 @@ where } self.send_nack(stream, &response, error_parts.join("; ")) - .await + .await?; } + + Ok(deliveries) } - /// Dispatch decoded resources to watchers and update cache. + /// Update the cache from decoded resources and stage watcher deliveries. /// - /// Returns futures that resolve when watchers signal ProcessingDone. - /// Uses backpressure: waits if a watcher's channel is full. - async fn dispatch_resources( + /// The staged events carry `ProcessingDone` receivers that gate reading + /// the next response (ADS flow control); the sends themselves happen in + /// the [`PendingDispatch`] future, with backpressure on full channels. + fn dispatch_resources( &mut self, + deliveries: &mut Vec, type_url: &str, resources: Vec, - ) -> Vec> { - let mut processing_done_futures = Vec::new(); - + ) { let watcher_info: Vec<_> = match self.type_states.get_mut(type_url) { Some(s) => { for resource in &resources { @@ -1107,7 +1168,7 @@ where .map(|(id, entry)| (*id, entry.event_tx.clone(), entry.subscription.clone())) .collect() } - None => return processing_done_futures, + None => return, }; // Cancel resource timers for received resources (gRFC A57). @@ -1127,22 +1188,24 @@ where result: Ok(Arc::clone(&resource)), done, }; - // Use backpressure: await if channel is full. - // Ignore send errors (watcher dropped). - let _ = event_tx.send(event).await; - processing_done_futures.push(rx); + deliveries.push((event_tx, event, Some(rx))); } } } - - processing_done_futures } - /// Notify watchers of a validation error for a specific resource. + /// Stage validation-error notifications for a specific resource. /// /// Per gRFC A46/A88, errors are routed only to watchers interested in - /// that specific resource (plus wildcard watchers). - async fn notify_resource_error(&mut self, type_url: &str, resource_name: &str, error: &str) { + /// that specific resource (plus wildcard watchers). Error events do not + /// gate flow control (no `ProcessingDone` receiver is staged). + fn notify_resource_error( + &mut self, + deliveries: &mut Vec, + type_url: &str, + resource_name: &str, + error: &str, + ) { let type_state = match self.type_states.get_mut(type_url) { Some(s) => s, None => return, @@ -1166,29 +1229,29 @@ where result: Err(Error::Validation(error.to_string())), done, }; - let _ = event_tx.send(event).await; + deliveries.push((event_tx, event, None)); } } - /// Detect resources that were deleted (present in cache but not in response). + /// Detect resources that were deleted (present in cache but not in response) + /// and stage the deletion notifications. /// /// Per gRFC A53, for resource types with ALL_RESOURCES_REQUIRED_IN_SOTW = true, /// if a previously-received resource is absent from a new SotW response, /// it is treated as deleted. - async fn detect_deleted_resources( + fn detect_deleted_resources( &mut self, + deliveries: &mut Vec, type_url: &str, received_names: &HashSet, - ) -> Vec> { - let mut processing_done_futures = Vec::new(); - + ) { let type_state = match self.type_states.get_mut(type_url) { Some(s) => s, - None => return processing_done_futures, + None => return, }; if !type_state.all_resources_required_in_sotw { - return processing_done_futures; + return; } let deleted_names: Vec = type_state @@ -1211,8 +1274,7 @@ where result: Err(Error::ResourceDoesNotExist), done, }; - let _ = event_tx.send(event).await; - processing_done_futures.push(rx); + deliveries.push((event_tx, event, Some(rx))); } } @@ -1220,8 +1282,6 @@ where let counts = type_state.resource_state_counts(); self.recorder .sync_resource_counts(&type_state.type_url, &counts); - - processing_done_futures } /// Send an ACK for a response. @@ -1551,3 +1611,239 @@ mod tests { assert_eq!(gauge_for(&events, "acked"), Some(0)); } } + +/// Regression tests for the worker's ADS flow control: watcher deliveries and +/// `ProcessingDone` waits must not freeze the event loop (deadlock with +/// watchers that issue commands while holding the token) and must gate +/// reading the next response. +#[cfg(test)] +mod flow_control_tests { + use std::time::Duration; + + use bytes::Bytes; + + use crate::client::config::ClientConfig; + use crate::client::watch::ResourceEvent; + use crate::codec::XdsCodec; + use crate::error::Result; + use crate::message::{DiscoveryRequest, DiscoveryResponse, Node, ResourceAny}; + use crate::resource::{Resource, TypeUrl}; + use crate::runtime::tokio::TokioRuntime; + use crate::transport::mock::{MockServer, mock_transport}; + use crate::{XdsClient, error::Error}; + + const TEST_TYPE_URL: &str = "type.googleapis.com/test.Resource"; + + /// Minimal resource: the message is the resource name itself. + #[derive(Debug, Clone)] + struct TestResource; + + impl Resource for TestResource { + type Message = String; + const TYPE_URL: TypeUrl = TypeUrl::new(TEST_TYPE_URL); + const ALL_RESOURCES_REQUIRED_IN_SOTW: bool = false; + + fn deserialize(bytes: Bytes) -> Result { + String::from_utf8(bytes.to_vec()).map_err(|e| Error::Validation(e.to_string())) + } + + fn name(message: &Self::Message) -> &str { + message + } + + fn validate(_message: Self::Message) -> Result { + Ok(Self) + } + } + + /// Line-based codec: `type_url \n version \n nonce \n name,name,...`. + struct FakeCodec; + + impl XdsCodec for FakeCodec { + fn encode_request(&self, request: &DiscoveryRequest<'_>) -> Result { + Ok(Bytes::from(format!( + "{}\n{}\n{}\n{}", + request.type_url, + request.version_info, + request.response_nonce, + request.resource_names.join(",") + ))) + } + + fn decode_response(&self, bytes: Bytes) -> Result { + let text = + String::from_utf8(bytes.to_vec()).map_err(|e| Error::Validation(e.to_string()))?; + let mut lines = text.split('\n'); + let type_url = lines.next().unwrap_or_default().to_string(); + let version_info = lines.next().unwrap_or_default().to_string(); + let nonce = lines.next().unwrap_or_default().to_string(); + let resources = lines + .next() + .unwrap_or_default() + .split(',') + .filter(|n| !n.is_empty()) + .map(|name| ResourceAny { + type_url: type_url.clone(), + value: Bytes::from(name.to_string()), + }) + .collect(); + Ok(DiscoveryResponse { + version_info, + resources, + type_url, + nonce, + }) + } + } + + fn response(version: &str, nonce: &str, names: &[&str]) -> Bytes { + Bytes::from(format!( + "{TEST_TYPE_URL}\n{version}\n{nonce}\n{}", + names.join(",") + )) + } + + /// (version_info, response_nonce) of an encoded request. + fn parse_request(bytes: &Bytes) -> (String, String) { + let text = String::from_utf8(bytes.to_vec()).unwrap(); + let mut lines = text.split('\n'); + let _type_url = lines.next().unwrap_or_default(); + let version = lines.next().unwrap_or_default().to_string(); + let nonce = lines.next().unwrap_or_default().to_string(); + (version, nonce) + } + + /// Client watching `res-0` with an established mock stream, its initial + /// request already drained. + async fn connected_client() -> ( + XdsClient, + crate::client::watch::ResourceWatcher, + MockServer, + ) { + let (builder, mut servers) = mock_transport(); + let config = ClientConfig::new(Node::new("test", "0"), "mock:///xds"); + let client = XdsClient::builder(config, builder, FakeCodec, TokioRuntime).build(); + + let watcher = client.watch::("res-0").await; + let mut server = tokio::time::timeout(Duration::from_secs(5), servers.recv()) + .await + .expect("timed out waiting for stream") + .expect("transport dropped"); + let _initial = tokio::time::timeout(Duration::from_secs(5), server.requests.recv()) + .await + .expect("timed out waiting for initial request") + .expect("stream closed"); + (client, watcher, server) + } + + /// A watcher that issues more commands than the command channel buffers + /// while holding its `ProcessingDone` token must not deadlock the worker. + /// + /// Before the flow-control fix the worker awaited the token inside + /// `handle_response`, so it stopped draining commands; once the channel + /// filled, watcher and worker waited on each other forever. + #[tokio::test] + async fn commands_drain_while_processing_done_is_held() { + let (client, mut watcher, server) = connected_client().await; + + server + .responses + .send(Ok(Some(response("1", "n1", &["res-0"])))) + .unwrap(); + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for event") + .expect("watcher closed"); + let ResourceEvent::ResourceChanged { + result: Ok(_), + done, + } = event + else { + panic!("expected ResourceChanged(Ok)"); + }; + + // Well past COMMAND_CHANNEL_BUFFER_SIZE (64). + let mut extra = Vec::new(); + tokio::time::timeout(Duration::from_secs(5), async { + for i in 1..=100 { + extra.push(client.watch::(format!("res-{i}")).await); + } + }) + .await + .expect("deadlock: commands not drained while ProcessingDone was held"); + + // The worker is still healthy end-to-end: it reads the next response + // and delivers it. + drop(done); + server + .responses + .send(Ok(Some(response("2", "n2", &["res-0"])))) + .unwrap(); + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for second event") + .expect("watcher closed"); + assert!(matches!( + event, + ResourceEvent::ResourceChanged { result: Ok(_), .. } + )); + } + + /// The ACK goes out as soon as the response is validated and cached, and + /// the *next* response is not delivered until the previous one's + /// `ProcessingDone` tokens drop (ADS flow control). + #[tokio::test] + async fn next_response_gated_until_processing_done() { + let (_client, mut watcher, mut server) = connected_client().await; + + server + .responses + .send(Ok(Some(response("1", "n1", &["res-0"])))) + .unwrap(); + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for event") + .expect("watcher closed"); + let ResourceEvent::ResourceChanged { + result: Ok(_), + done, + } = event + else { + panic!("expected ResourceChanged(Ok)"); + }; + + // ACK is not gated on ProcessingDone. + let ack = tokio::time::timeout(Duration::from_secs(5), server.requests.recv()) + .await + .expect("ACK not sent while ProcessingDone was held") + .expect("stream closed"); + assert_eq!(parse_request(&ack), ("1".to_string(), "n1".to_string())); + + // The next response is gated on ProcessingDone. + server + .responses + .send(Ok(Some(response("2", "n2", &["res-0"])))) + .unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(200), watcher.next()) + .await + .is_err(), + "second response delivered while the first was still being processed" + ); + + drop(done); + let event = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("timed out waiting for second event") + .expect("watcher closed"); + assert!(matches!( + event, + ResourceEvent::ResourceChanged { result: Ok(_), .. } + )); + let ack = tokio::time::timeout(Duration::from_secs(5), server.requests.recv()) + .await + .expect("second ACK not sent") + .expect("stream closed"); + assert_eq!(parse_request(&ack), ("2".to_string(), "n2".to_string())); + } +} diff --git a/xds-client/src/transport/mod.rs b/xds-client/src/transport/mod.rs index 033ff8558..2f509c695 100644 --- a/xds-client/src/transport/mod.rs +++ b/xds-client/src/transport/mod.rs @@ -122,3 +122,96 @@ pub trait TransportBuilder: Send + Sync + 'static { // - `fn close(&self, server: &ServerConfig)` for explicit connection cleanup // - Metrics/observability hooks } + +/// In-crate mock transport for worker tests. +/// +/// Lives here because [`TransportStream`] is sealed: test code outside this +/// module cannot implement it. +#[cfg(test)] +pub(crate) mod mock { + use super::{Transport, TransportBuilder, TransportStream, sealed}; + use crate::client::config::ServerConfig; + use crate::error::{Error, Result}; + use bytes::Bytes; + use tokio::sync::mpsc; + + /// Test-side handle to one mock ADS stream. + pub(crate) struct MockServer { + /// Requests the worker sent (initial requests, ACKs, subscription changes). + pub(crate) requests: mpsc::UnboundedReceiver, + /// Responses to push to the worker. + pub(crate) responses: mpsc::UnboundedSender>>, + } + + /// Returns a transport builder for the worker plus the receiver on which + /// the test obtains a [`MockServer`] for every stream the worker opens. + pub(crate) fn mock_transport() -> (MockTransportBuilder, mpsc::UnboundedReceiver) { + let (servers_tx, servers_rx) = mpsc::unbounded_channel(); + ( + MockTransportBuilder { + servers: servers_tx, + }, + servers_rx, + ) + } + + pub(crate) struct MockTransportBuilder { + servers: mpsc::UnboundedSender, + } + + impl TransportBuilder for MockTransportBuilder { + type Transport = MockTransport; + + async fn build(&self, _server: &ServerConfig) -> Result { + Ok(MockTransport { + servers: self.servers.clone(), + }) + } + } + + pub(crate) struct MockTransport { + servers: mpsc::UnboundedSender, + } + + impl Transport for MockTransport { + type Stream = MockStream; + + async fn new_stream(&self, initial_requests: Vec) -> Result { + let (req_tx, req_rx) = mpsc::unbounded_channel(); + let (resp_tx, resp_rx) = mpsc::unbounded_channel(); + for request in initial_requests { + let _ = req_tx.send(request); + } + let _ = self.servers.send(MockServer { + requests: req_rx, + responses: resp_tx, + }); + Ok(MockStream { + requests: req_tx, + responses: resp_rx, + }) + } + } + + pub(crate) struct MockStream { + requests: mpsc::UnboundedSender, + responses: mpsc::UnboundedReceiver>>, + } + + impl sealed::Sealed for MockStream {} + + impl TransportStream for MockStream { + async fn send(&mut self, request: Bytes) -> Result<()> { + self.requests + .send(request) + .map_err(|_| Error::Connection("mock stream closed".into())) + } + + async fn recv(&mut self) -> Result> { + match self.responses.recv().await { + Some(result) => result, + None => Ok(None), + } + } + } +} diff --git a/xds-test-util/src/config.rs b/xds-test-util/src/config.rs index 45207df04..f211917ee 100644 --- a/xds-test-util/src/config.rs +++ b/xds-test-util/src/config.rs @@ -127,6 +127,48 @@ pub fn build_route_config(name: &str, cluster: &str) -> RouteConfiguration { } } +/// Builds a route configuration referencing `count` generated clusters +/// (`cluster-0` .. `cluster-{count-1}`), one prefix route per cluster. +/// +/// With `count = 1` this is the generated-name equivalent of +/// [`build_route_config`]. Larger counts mirror control planes that share one +/// route config across many services (e.g. Istio's per-port route configs). +pub fn build_route_config_with_cluster_count(name: &str, count: usize) -> RouteConfiguration { + let clusters: Vec = (0..count).map(|i| format!("cluster-{i}")).collect(); + build_route_config_with_clusters(name, &clusters) +} + +/// Builds a route configuration with one prefix route per cluster +/// (`/` -> that cluster), all in one virtual host. +/// +/// Mirrors control planes that share one route config across many services +/// (e.g. Istio's per-port route configs). +pub fn build_route_config_with_clusters(name: &str, clusters: &[String]) -> RouteConfiguration { + RouteConfiguration { + name: name.to_string(), + virtual_hosts: vec![VirtualHost { + name: "default".to_string(), + domains: vec!["*".to_string()], + routes: clusters + .iter() + .map(|cluster| Route { + r#match: Some(RouteMatch { + path_specifier: Some(PathSpecifier::Prefix(format!("/{cluster}"))), + ..Default::default() + }), + action: Some(Action::Route(RouteAction { + cluster_specifier: Some(ClusterSpecifier::Cluster(cluster.to_string())), + ..Default::default() + })), + ..Default::default() + }) + .collect(), + ..Default::default() + }], + ..Default::default() + } +} + /// Builds an EDS-discovered cluster named `name`. pub fn build_cluster(name: &str) -> Cluster { Cluster { @@ -172,3 +214,58 @@ fn lb_endpoint(host: &str, port: u16) -> LbEndpoint { ..Default::default() } } + +#[cfg(test)] +mod tests { + use super::*; + + fn route_clusters(rc: &RouteConfiguration) -> Vec<(String, String)> { + rc.virtual_hosts[0] + .routes + .iter() + .map(|r| { + let prefix = match r.r#match.as_ref().and_then(|m| m.path_specifier.as_ref()) { + Some(PathSpecifier::Prefix(p)) => p.clone(), + other => panic!("expected prefix match, got {other:?}"), + }; + let cluster = match r.action.as_ref() { + Some(Action::Route(a)) => match a.cluster_specifier.as_ref() { + Some(ClusterSpecifier::Cluster(c)) => c.clone(), + other => panic!("expected cluster specifier, got {other:?}"), + }, + other => panic!("expected route action, got {other:?}"), + }; + (prefix, cluster) + }) + .collect() + } + + #[test] + fn route_config_with_clusters_builds_one_route_per_cluster() { + let clusters = vec!["a".to_string(), "b".to_string()]; + let rc = build_route_config_with_clusters("rc", &clusters); + assert_eq!(rc.name, "rc"); + assert_eq!( + route_clusters(&rc), + vec![ + ("/a".to_string(), "a".to_string()), + ("/b".to_string(), "b".to_string()), + ] + ); + } + + #[test] + fn route_config_with_cluster_count_generates_names() { + let rc = build_route_config_with_cluster_count("rc", 3); + let routes = route_clusters(&rc); + assert_eq!(routes.len(), 3); + assert_eq!( + routes[0], + ("/cluster-0".to_string(), "cluster-0".to_string()) + ); + assert_eq!( + routes[2], + ("/cluster-2".to_string(), "cluster-2".to_string()) + ); + } +}