diff --git a/tonic-xds/Cargo.toml b/tonic-xds/Cargo.toml index c174d485c..49d73713e 100644 --- a/tonic-xds/Cargo.toml +++ b/tonic-xds/Cargo.toml @@ -114,6 +114,7 @@ allowed_external_types = [ # not major released "prost::*", "opentelemetry::*", + "shared_http_body::*", "tower_service::Service", "tower::BoxError", diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index 742463a66..33c9a3e8c 100644 --- a/tonic-xds/src/client/channel.rs +++ b/tonic-xds/src/client/channel.rs @@ -22,8 +22,8 @@ * */ -use crate::client::cluster::ClusterClientRegistryGrpc; -use crate::client::endpoint::{EndpointAddress, EndpointChannel}; +use crate::client::cluster::{ClusterClientRegistry, ClusterClientRegistryGrpc}; +use crate::client::endpoint::{EndpointAddress, EndpointChannel, MakeConnector}; use crate::client::lb::{ClusterDiscovery, XdsLbService}; use crate::client::route::{PreRouteInterceptor, Router, XdsRoutingLayer}; use crate::xds::bootstrap::{BootstrapConfig, BootstrapError}; @@ -34,19 +34,23 @@ use crate::xds::cluster_discovery::{GrpcMakeConnector, XdsClusterDiscovery}; use crate::xds::resource_manager::XdsResourceManager; use crate::xds::routing::XdsRouter; use crate::{TonicCallCredentials, XdsUri}; -use http::Request; +use http::{Request, Response}; +use http_body::Body; +use shared_http_body::SharedBody; #[cfg(feature = "_tls-any")] use std::collections::HashMap; use std::fmt::Debug; use std::sync::Arc; use std::task::{Context, Poll}; use tonic::{body::Body as TonicBody, client::GrpcService, transport::channel::Channel}; -use tower::{BoxError, Service, ServiceBuilder, util::BoxCloneSyncService}; +use tower::{BoxError, Service, ServiceBuilder, load::Load, util::BoxCloneSyncService}; use xds_client::{ ClientConfig, MetricsRecorder, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient, }; -use crate::client::retry::{GrpcRetryPolicy, RetryLayer}; +use crate::client::retry::{ + GrpcRetryPolicy, RetryClassifier, RetryConfig, RetryLayer, RetryPolicy, +}; /// Configuration for building [`XdsChannel`] / [`XdsChannelGrpc`]. #[derive(Clone, Debug)] @@ -297,11 +301,48 @@ impl XdsChannelBuilder { } fn build_tonic_grpc_channel(&self) -> Result { - let bootstrap = match self.config.bootstrap.clone() { + let bootstrap = self.load_bootstrap()?; + + // The cert-provider registry is gRPC-specific (gRFC A29): it is consumed + // only by the built-in gRPC connector factory. Build it here, before the + // bootstrap is moved into `build_xds_runtime`, so the generic transport + // path (`build_transport_channel`) never pays for it. + #[cfg(feature = "_tls-any")] + let cert_provider_registry = Arc::new(CertProviderRegistry::from_bootstrap( + &bootstrap.certificate_providers, + self.cert_providers.clone(), + )?); + + let (xds_client, cache, resource_manager) = self.build_xds_runtime(bootstrap)?; + + Ok(self.build_from_cache( + cache, + #[cfg(feature = "_tls-any")] + cert_provider_registry, + xds_client, + resource_manager, + )) + } + + /// Loads the bootstrap config from the builder, falling back to the + /// environment (`GRPC_XDS_BOOTSTRAP` / `GRPC_XDS_BOOTSTRAP_CONFIG`). + fn load_bootstrap(&self) -> Result { + Ok(match self.config.bootstrap.clone() { Some(b) => b, None => BootstrapConfig::from_env()?, - }; + }) + } + /// Builds the transport-agnostic xDS runtime: the ADS [`XdsClient`], the + /// shared [`XdsCache`], and the [`XdsResourceManager`] that drives it. + /// + /// This is everything both the gRPC and the generic transport build paths + /// need, and nothing gRPC-specific (the cert-provider registry is built by + /// the caller). Consumes `bootstrap` (its node is moved into the client). + fn build_xds_runtime( + &self, + bootstrap: BootstrapConfig, + ) -> Result<(XdsClient, Arc, XdsResourceManager), BuildError> { let listener_name = self.config.target_uri.target.clone(); let server_uri = bootstrap.server_uri().to_owned(); @@ -326,12 +367,6 @@ impl XdsChannelBuilder { transport_builder = transport_builder.with_call_credentials(creds); } - #[cfg(feature = "_tls-any")] - let cert_provider_registry = Arc::new(CertProviderRegistry::from_bootstrap( - &bootstrap.certificate_providers, - self.cert_providers.clone(), - )?); - let node = Node::try_from(bootstrap.node)?; let client_config = ClientConfig::new(node, &server_uri).with_target(self.config.target_uri.to_string()); @@ -346,13 +381,7 @@ impl XdsChannelBuilder { let resource_manager = XdsResourceManager::new(xds_client.clone(), cache.clone(), listener_name); - Ok(self.build_from_cache( - cache, - #[cfg(feature = "_tls-any")] - cert_provider_registry, - xds_client, - resource_manager, - )) + Ok((xds_client, cache, resource_manager)) } /// Internal builder that wires the service stack from a pre-built cache. @@ -378,32 +407,137 @@ impl XdsChannelBuilder { let discovery: Arc< dyn ClusterDiscovery>, > = Arc::new(XdsClusterDiscovery::new(cache, GrpcMakeConnector::new())); - let retry_policy = GrpcRetryPolicy::default(); - let resources = Arc::new(XdsChannelResources { _resource_manager: resource_manager, _xds_client: xds_client, }); - let routing_layer = XdsRoutingLayer::new(router, self.pre_route.clone(), self.authority()); + // gRPC bridges the retry engine's cloneable `SharedBody` back into a + // `tonic` body, which is what the built-in `Channel` connector expects. + self.build_stack( + router, + self.pre_route.clone(), + discovery, + Arc::new(ClusterClientRegistryGrpc::new()), + GrpcRetryPolicy::default(), + Some(resources), + |req: Request>| req.map(TonicBody::new), + ) + } + + /// Wires the routing -> retry -> load-balancing service stack over a + /// pre-built cluster discovery and type-erases it into a cloneable, + /// `Send + Sync` boxed [`tower::Service`]. This is the transport-generic + /// core shared by [`build_grpc_channel`](Self::build_grpc_channel) and + /// [`build_transport_channel`](Self::build_transport_channel). + /// + /// - `S` is the per-endpoint service produced by the cluster discovery. + /// - `B` is the inbound request body; `EB` the body the per-endpoint service + /// accepts; `Res` the response body. + /// - `post_retry_map` bridges the retry engine's cloneable [`SharedBody`] + /// (needed so a request can be replayed) back to `EB`. For gRPC this + /// rewraps it into a `tonic` body; for a transport whose service already + /// accepts `SharedBody` it is the identity. + #[allow(clippy::too_many_arguments)] + fn build_stack( + &self, + router: Arc, + interceptor: Option>, + discovery: Arc>, + cluster_registry: Arc, Response>>, + retry_policy: RetryPolicy, + resources: Option>, + post_retry_map: MapFn, + ) -> BoxCloneSyncService, Response, BoxError> + where + S: Service, Response = Response> + Load + Send + 'static, + >>::Error: Into, + >>::Future: Send, + ::Metric: Debug, + C: RetryClassifier + Send + Sync + 'static, + B: Body + Unpin + Send + 'static, + B::Data: Clone + Send + Sync, + B::Error: Clone + Send + Sync, + EB: Send + 'static, + Res: Send + 'static, + MapFn: FnMut(Request>) -> Request + Clone + Send + Sync + 'static, + { + let routing_layer = XdsRoutingLayer::new(router, interceptor, self.authority()); let retry_layer = RetryLayer::new(retry_policy); - let cluster_registry = Arc::new(ClusterClientRegistryGrpc::new()); let lb_service = XdsLbService::new(cluster_registry, discovery); let inner = ServiceBuilder::new() .layer(routing_layer) .layer(retry_layer) - .map_request(|req: Request>| { - req.map(TonicBody::new) - }) + .map_request(post_retry_map) .service(lb_service); BoxCloneSyncService::new(XdsChannel { config: self.config.clone(), inner, - _resources: Some(resources), + _resources: resources, }) } + /// Builds a transport-generic xDS channel: the routing + retry + + /// load-balancing stack over a caller-provided [`MakeConnector`], returned + /// as a type-erased, cloneable, `Send + Sync` [`tower::Service`]. + /// + /// Unlike [`build_grpc_channel`](Self::build_grpc_channel), the per-endpoint + /// transport is supplied by the caller via `make_connector`, and the retry + /// behavior by `retry_classifier` (which decides retryability for that + /// transport) plus `retry_config`. The gRPC channel is the special case + /// where the connector speaks HTTP/2 + gRPC and the classifier follows the + /// gRPC retry spec. + /// + /// The returned service accepts `http::Request` and yields + /// `http::Response`, where `B` / `Res` are the request / response body + /// types of the caller's transport. The caller's connector service accepts a + /// [`SharedBody`]-wrapped body so that requests can be cloned for retries. + pub fn build_transport_channel( + &self, + make_connector: MC, + retry_classifier: C, + retry_config: RetryConfig, + ) -> Result, Response, BoxError>, BuildError> + where + MC: MakeConnector, + MC::Service: Service>, Response = Response> + Load, + >>>::Error: Into, + >>>::Future: Send, + ::Metric: Debug, + C: RetryClassifier + Send + Sync + 'static, + B: Body + Unpin + Send + 'static, + B::Data: Clone + Send + Sync, + B::Error: Clone + Send + Sync, + Res: Send + 'static, + { + let bootstrap = self.load_bootstrap()?; + let (xds_client, cache, resource_manager) = self.build_xds_runtime(bootstrap)?; + + let router: Arc = Arc::new(XdsRouter::new(&cache)); + let discovery: Arc> = + Arc::new(XdsClusterDiscovery::new(cache, make_connector)); + + let resources = Arc::new(XdsChannelResources { + _resource_manager: resource_manager, + _xds_client: xds_client, + }); + + let retry_policy = RetryPolicy::new(retry_config, retry_classifier); + + // The caller's connector already accepts `SharedBody`, so the post-retry + // map is the identity — no body rewrapping is needed. + Ok(self.build_stack( + router, + self.pre_route.clone(), + discovery, + Arc::new(ClusterClientRegistry::new()), + retry_policy, + Some(resources), + |req: Request>| req, + )) + } + /// Builds an `XdsChannelGrpc`, which is a type-erased gRPC channel. // TODO: Support HTTP and other channel types (not just gRPC). This will // require a generic `build()` or separate `build_http_channel()` method. @@ -421,22 +555,15 @@ impl XdsChannelBuilder { retry_policy: GrpcRetryPolicy, interceptor: Option>, ) -> XdsChannelGrpc { - let routing_layer = XdsRoutingLayer::new(router, interceptor, self.authority()); - let retry_layer = RetryLayer::new(retry_policy); - let cluster_registry = Arc::new(ClusterClientRegistryGrpc::new()); - let lb_service = XdsLbService::new(cluster_registry, discovery); - let inner = ServiceBuilder::new() - .layer(routing_layer) - .layer(retry_layer) - .map_request(|req: Request>| { - req.map(TonicBody::new) - }) - .service(lb_service); - BoxCloneSyncService::new(XdsChannel { - config: self.config.clone(), - inner, - _resources: None, - }) + self.build_stack( + router, + interceptor, + discovery, + Arc::new(ClusterClientRegistryGrpc::new()), + retry_policy, + None, + |req: Request>| req.map(TonicBody::new), + ) } /// Channel-level authority used as the routing key for matching against diff --git a/tonic-xds/src/client/retry.rs b/tonic-xds/src/client/retry.rs index 2d0060deb..52f6d455a 100644 --- a/tonic-xds/src/client/retry.rs +++ b/tonic-xds/src/client/retry.rs @@ -52,7 +52,7 @@ use crate::client::circuit_breaking::is_local_circuit_breaker_drop; /// /// These are errors where the request was definitely **not** sent, making it safe to retry. /// Walks the full error source chain via [`std::error::Error::source`]. -pub(crate) fn is_retryable_connection_error(err: &(dyn std::error::Error + 'static)) -> bool { +pub fn is_retryable_connection_error(err: &(dyn std::error::Error + 'static)) -> bool { let mut current: Option<&(dyn std::error::Error + 'static)> = Some(err); while let Some(e) = current { if let Some(io_err) = e.downcast_ref::() { @@ -77,14 +77,14 @@ pub(crate) fn is_retryable_grpc_status_code( code != tonic::Code::Ok && retryable_codes.contains(&code) } -/// Transport-specific retry decisions. [`RetryPolicy`] owns everything else +/// Transport-specific retry decisions. The retry engine owns everything else /// (attempt cap, backoff, body cloning), so a classifier only decides *whether* /// a response is retryable and optionally mutates the request before each retry. /// /// This is the seam that lets non-gRPC transports (e.g. plain HTTP) reuse the /// shared retry engine by supplying their own retryable-status logic without /// duplicating any retry state machine. -pub(crate) trait RetryClassifier: Clone { +pub trait RetryClassifier: Clone { /// Whether the request should be retried, given either the transport response /// or a connection-level error. Implementations typically retry on a retryable /// connection error (see [`is_retryable_connection_error`]) or a retryable @@ -115,7 +115,7 @@ const MIN_BACKOFF: Duration = Duration::from_millis(1); /// - `max_interval` defaults to `10 * base_interval`. /// - `max_interval` must be >= `base_interval`; if not, it is clamped to `base_interval`. #[derive(Debug, Clone, PartialEq)] -pub(crate) struct RetryBackoffConfig { +pub struct RetryBackoffConfig { pub(crate) base_interval: Duration, pub(crate) max_interval: Duration, pub(crate) backoff_multiplier: f64, @@ -125,7 +125,7 @@ impl RetryBackoffConfig { /// Create a new backoff config with the given `base_interval`. /// `max_interval` defaults to `10 * base_interval`. /// `backoff_multiplier` defaults to `2.0`. - pub(crate) fn new(base_interval: Duration) -> Self { + pub fn new(base_interval: Duration) -> Self { let base_interval = base_interval.max(MIN_BACKOFF); Self { max_interval: base_interval * 10, @@ -136,14 +136,14 @@ impl RetryBackoffConfig { /// Set the maximum backoff interval. /// Values < 1ms are treated as 1ms. Values < `base_interval` are clamped to `base_interval`. - pub(crate) fn max_interval(mut self, max_interval: Duration) -> Self { + pub fn max_interval(mut self, max_interval: Duration) -> Self { let max_interval = max_interval.max(MIN_BACKOFF); self.max_interval = max_interval.max(self.base_interval); self } /// Set the backoff multiplier (default: 2.0). - pub(crate) fn backoff_multiplier(mut self, multiplier: f64) -> Self { + pub fn backoff_multiplier(mut self, multiplier: f64) -> Self { self.backoff_multiplier = multiplier; self } @@ -167,27 +167,27 @@ impl Default for RetryBackoffConfig { /// - `num_retries` must be >= 1. Values of 0 are clamped to 1. /// - `num_retries` is capped so total attempts (num_retries + 1) never exceed 5. #[derive(Debug, Clone)] -pub(crate) struct RetryConfig { +pub struct RetryConfig { pub(crate) num_retries: u32, pub(crate) retry_backoff: RetryBackoffConfig, } impl RetryConfig { /// Create a new retry config with defaults. - pub(crate) fn new() -> Self { + pub fn new() -> Self { Self::default() } /// Set the number of retries (total attempts = num_retries + 1). /// Values of 0 are clamped to 1. Values that would exceed 5 total attempts are capped. - pub(crate) fn num_retries(mut self, num_retries: u32) -> Self { + pub fn num_retries(mut self, num_retries: u32) -> Self { // Safety: clamp panics if min > max. Here min=1, max=MAX_ATTEMPTS-1=4 (const). self.num_retries = num_retries.clamp(1, MAX_ATTEMPTS - 1); self } /// Set the backoff configuration. - pub(crate) fn retry_backoff(mut self, backoff: RetryBackoffConfig) -> Self { + pub fn retry_backoff(mut self, backoff: RetryBackoffConfig) -> Self { self.retry_backoff = backoff; self } diff --git a/tonic-xds/src/client/xds_e2e.rs b/tonic-xds/src/client/xds_e2e.rs index 739aa412f..8e4e56032 100644 --- a/tonic-xds/src/client/xds_e2e.rs +++ b/tonic-xds/src/client/xds_e2e.rs @@ -72,6 +72,91 @@ mod test { .expect("build xds channel") } + // --- Transport-generic build path (`build_transport_channel`) --- + // + // The types below are written entirely against tonic-xds' *public* API, + // mirroring what an out-of-crate transport (e.g. plain HTTP) would provide: + // a `MakeConnector` whose per-endpoint service accepts a `SharedBody`-wrapped + // body, plus a caller-supplied `RetryClassifier`. This exercises the generic + // stack instead of the built-in gRPC one. + use crate::{ + BoxFuture, ClusterConfig, Connector, EndpointAddress, EndpointChannel, MakeConnector, + RetryClassifier, RetryConfig, is_retryable_connection_error, + }; + use http::{Request, Response}; + use shared_http_body::SharedBody; + use std::sync::Arc; + use tonic::body::Body as TonicBody; + use tonic::transport::Endpoint; + use tower::util::BoxCloneSyncService; + use tower::{BoxError, ServiceExt as _}; + + /// Per-endpoint service: a lazily-connected tonic channel adapted to accept + /// a `SharedBody`-wrapped request body, wrapped in `EndpointChannel` for + /// in-flight load reporting. + type SharedBodyEndpoint = EndpointChannel< + BoxCloneSyncService>, Response, BoxError>, + >; + + /// A non-gRPC retry classifier: retries only connection-level errors, reusing + /// the public [`is_retryable_connection_error`] helper. + #[derive(Clone)] + struct ConnErrorRetryClassifier; + + impl RetryClassifier for ConnErrorRetryClassifier { + fn is_retryable(&self, res: &Result, BoxError>) -> bool { + matches!(res, Err(e) if is_retryable_connection_error(e.as_ref())) + } + } + + struct SharedBodyConnector; + + impl Connector for SharedBodyConnector { + type Service = SharedBodyEndpoint; + + fn connect(&self, addr: &EndpointAddress) -> BoxFuture { + let channel = Endpoint::from_shared(format!("http://{addr}")) + .expect("valid endpoint uri") + .connect_lazy(); + let svc = channel + .map_request(|req: Request>| req.map(TonicBody::new)) + .map_err(|e| -> BoxError { e.into() }); + let ep = EndpointChannel::new(BoxCloneSyncService::new(svc)); + Box::pin(async move { ep }) + } + } + + struct SharedBodyMakeConnector; + + impl MakeConnector for SharedBodyMakeConnector { + type Service = SharedBodyEndpoint; + + fn make_connector( + &self, + _cluster: ClusterConfig<'_>, + ) -> Result + Send + Sync>, BoxError> { + Ok(Arc::new(SharedBodyConnector)) + } + } + + /// Like [`build_channel`], but drives the transport-generic build path with a + /// custom connector and retry classifier. Returns an `XdsChannelGrpc` because + /// the outbound body is still a `tonic` body (`B = TonicBody`). + fn build_generic_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_transport_channel( + SharedBodyMakeConnector, + ConnErrorRetryClassifier, + RetryConfig::new(), + ) + .expect("build transport 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. @@ -152,10 +237,52 @@ mod test { 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. + /// End-to-end test for the transport-generic build path: verifies that a + /// channel built via [`XdsChannelBuilder::build_transport_channel`] with a + /// caller-supplied connector and [`RetryClassifier`] routes traffic to the + /// correct backend, just like the built-in gRPC path. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn xds_channel_e2e_transport_generic_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_generic_channel(cp_addr, "my-service")); + let reply = say_hello_until_prefix(&mut client, "backend:").await; + assert_eq!(reply, "backend: world"); + + let _ = backend.shutdown.send(()); + } + #[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) diff --git a/tonic-xds/src/lib.rs b/tonic-xds/src/lib.rs index 2c69499ed..5ca1f8c20 100644 --- a/tonic-xds/src/lib.rs +++ b/tonic-xds/src/lib.rs @@ -174,6 +174,9 @@ pub use client::channel::{ pub use client::endpoint::{ ClusterConfig, Connector, EndpointAddress, EndpointChannel, MakeConnector, }; +pub use client::retry::{ + RetryBackoffConfig, RetryClassifier, RetryConfig, is_retryable_connection_error, +}; pub use client::route::PreRouteInterceptor; pub use common::async_util::BoxFuture; pub use xds::bootstrap::{BootstrapConfig, BootstrapError};