diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index 742463a66..a7f0f6232 100644 --- a/tonic-xds/src/client/channel.rs +++ b/tonic-xds/src/client/channel.rs @@ -26,11 +26,14 @@ use crate::client::cluster::ClusterClientRegistryGrpc; use crate::client::endpoint::{EndpointAddress, EndpointChannel}; use crate::client::lb::{ClusterDiscovery, XdsLbService}; use crate::client::route::{PreRouteInterceptor, Router, XdsRoutingLayer}; +use crate::common::async_util::AbortOnDrop; use crate::xds::bootstrap::{BootstrapConfig, BootstrapError}; -use crate::xds::cache::XdsCache; +use crate::xds::cache::{CacheWatch, XdsCache}; #[cfg(feature = "_tls-any")] use crate::xds::cert_provider::{CertProviderError, CertProviderRegistry, CertificateProvider}; use crate::xds::cluster_discovery::{GrpcMakeConnector, XdsClusterDiscovery}; +use crate::xds::resource::ListenerResource; +use crate::xds::resource::route_config::RouteConfigMetadata; use crate::xds::resource_manager::XdsResourceManager; use crate::xds::routing::XdsRouter; use crate::{TonicCallCredentials, XdsUri}; @@ -46,7 +49,13 @@ use xds_client::{ ClientConfig, MetricsRecorder, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient, }; -use crate::client::retry::{GrpcRetryPolicy, RetryLayer}; +use crate::client::retry::{GrpcRetryPolicy, RetryConfig, RetryLayer, RetryPolicy}; + +/// Extracts a [`RetryConfig`] from a Listener's [`RouteConfigMetadata`], returning +/// `None` when no retry config is present. Registered via +/// [`XdsChannelBuilder::with_retry_config_from_metadata`]. +type RetryConfigFromMetadata = + Arc Option + Send + Sync>; /// Configuration for building [`XdsChannel`] / [`XdsChannelGrpc`]. #[derive(Clone, Debug)] @@ -121,6 +130,10 @@ pub enum BuildError { struct XdsChannelResources { _resource_manager: XdsResourceManager, _xds_client: XdsClient, + /// Background task that hot-swaps the retry config from LDS updates. `None` + /// when no retry-config extractor is registered. Aborted when the last + /// channel clone drops. + _retry_watch_task: Option, } /// `XdsChannel` is an xDS-capable [`tower::Service`] implementation. @@ -195,6 +208,7 @@ pub struct XdsChannelBuilder { config: Arc, recorder: Option>, pre_route: Option>, + retry_config_from_metadata: Option, #[cfg(feature = "_tls-any")] cert_providers: HashMap>, } @@ -216,6 +230,14 @@ impl Debug for XdsChannelBuilder { .pre_route .as_deref() .map_or("None", |i| std::any::type_name_of_val(i)), + ) + .field( + "retry_config_from_metadata", + &if self.retry_config_from_metadata.is_some() { + "Some(..)" + } else { + "None" + }, ); #[cfg(feature = "_tls-any")] s.field( @@ -234,6 +256,7 @@ impl XdsChannelBuilder { config: Arc::new(config), recorder: None, pre_route: None, + retry_config_from_metadata: None, #[cfg(feature = "_tls-any")] cert_providers: HashMap::new(), } @@ -266,6 +289,24 @@ impl XdsChannelBuilder { self } + /// Drives the retry configuration from the control plane. + /// + /// The provided closure extracts a [`RetryConfig`] from each Listener's + /// [`RouteConfigMetadata`](crate::RouteConfigMetadata) (LDS). A background task + /// applies it on every Listener update and hot-swaps the result into the retry + /// policy, so retry behavior tracks the control plane without rebuilding the + /// channel; returning `None` keeps the previous config. + /// + /// When unset, the channel uses a fixed default retry configuration. + #[must_use] + pub fn with_retry_config_from_metadata(mut self, extract: F) -> Self + where + F: Fn(&RouteConfigMetadata) -> Option + Send + Sync + 'static, + { + self.retry_config_from_metadata = Some(Arc::new(extract)); + self + } + /// Registers a custom [`CertificateProvider`] under an xDS certificate /// provider instance name, resolved by CDS `UpstreamTlsContext` references. /// Shadows a bootstrap `file_watcher` instance of the same name. @@ -367,6 +408,16 @@ impl XdsChannelBuilder { resource_manager: XdsResourceManager, ) -> XdsChannelGrpc { let router: Arc = Arc::new(XdsRouter::new(&cache)); + + // Retry config is control-plane-driven: if the caller registered an + // extractor, seed the default policy and spawn a task that hot-swaps the + // config from LDS (Listener metadata) updates. `watch_listener()` must be + // taken before `cache` is moved into the cluster discovery. + let retry_policy = GrpcRetryPolicy::default(); + let retry_watch_task = self.retry_config_from_metadata.clone().map(|extract| { + spawn_retry_config_watch(retry_policy.clone(), cache.watch_listener(), extract) + }); + #[cfg(feature = "_tls-any")] let discovery: Arc< dyn ClusterDiscovery>, @@ -378,11 +429,11 @@ 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, + _retry_watch_task: retry_watch_task, }); let routing_layer = XdsRoutingLayer::new(router, self.pre_route.clone(), self.authority()); @@ -446,6 +497,32 @@ impl XdsChannelBuilder { } } +/// Spawns a background task that keeps a [`RetryPolicy`]'s configuration in sync +/// with the control plane. On each Listener (LDS) update, `extract` is applied to +/// the listener's [`RouteConfigMetadata`]: a returned `Some(config)` is hot-swapped +/// into the policy's shared config, while `None` leaves the previous config in +/// place. +/// +/// The returned [`AbortOnDrop`] handle stops the task when dropped, tying the +/// task's lifetime to the channel resources that hold it. +fn spawn_retry_config_watch( + policy: RetryPolicy, + mut watcher: CacheWatch, + extract: RetryConfigFromMetadata, +) -> AbortOnDrop +where + C: Send + 'static, +{ + let handle = tokio::spawn(async move { + while let Some(listener) = watcher.next().await { + if let Some(config) = extract(&listener.metadata) { + policy.update_config(config); + } + } + }); + AbortOnDrop(handle) +} + #[cfg(test)] mod tests { use super::{XdsChannelBuilder, XdsChannelConfig}; @@ -921,4 +998,39 @@ mod tests { let _channel = builder.build_from_cache(cache, xds_client, resource_manager); // Construction should succeed without panicking. } + + #[tokio::test] + async fn retry_config_hot_swaps_from_listener_metadata() { + use crate::client::retry::RetryConfig; + use crate::xds::resource::ListenerResource; + use crate::xds::resource::listener::RouteSource; + use crate::xds::resource::route_config::RouteConfigMetadata; + + let cache = XdsCache::new(); + let policy = GrpcRetryPolicy::default(); + // Default config: a single retry. + assert_eq!(policy.load_config().num_retries, 1); + + // The watch task recomputes the config from every Listener (LDS) update + // and hot-swaps it into the config shared across policy clones. + let extract: super::RetryConfigFromMetadata = + Arc::new(|_meta: &RouteConfigMetadata| Some(RetryConfig::new().num_retries(3))); + let _task = + super::spawn_retry_config_watch(policy.clone(), cache.watch_listener(), extract); + + cache.update_listener(Arc::new(ListenerResource { + name: "listener".into(), + route_source: RouteSource::Rds("rc".into()), + metadata: RouteConfigMetadata::default(), + })); + + // Yield until the spawned task observes the update and hot-swaps. + for _ in 0..100 { + if policy.load_config().num_retries == 3 { + break; + } + tokio::task::yield_now().await; + } + assert_eq!(policy.load_config().num_retries, 3); + } } diff --git a/tonic-xds/src/client/retry.rs b/tonic-xds/src/client/retry.rs index 2d0060deb..ab81c398f 100644 --- a/tonic-xds/src/client/retry.rs +++ b/tonic-xds/src/client/retry.rs @@ -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 } @@ -155,7 +155,7 @@ impl Default for RetryBackoffConfig { } } -/// Transport-agnostic retry knobs shared by every [`RetryClassifier`]. +/// Transport-agnostic retry knobs shared by every retry classifier. /// /// Built via [`RetryConfig::new`] with defaults, then customized via builder methods. /// @@ -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/lib.rs b/tonic-xds/src/lib.rs index 2c69499ed..c939d198a 100644 --- a/tonic-xds/src/lib.rs +++ b/tonic-xds/src/lib.rs @@ -174,6 +174,7 @@ pub use client::channel::{ pub use client::endpoint::{ ClusterConfig, Connector, EndpointAddress, EndpointChannel, MakeConnector, }; +pub use client::retry::{RetryBackoffConfig, RetryConfig}; pub use client::route::PreRouteInterceptor; pub use common::async_util::BoxFuture; pub use xds::bootstrap::{BootstrapConfig, BootstrapError}; diff --git a/tonic-xds/src/xds/cache.rs b/tonic-xds/src/xds/cache.rs index 12ae472d5..570fccf7c 100644 --- a/tonic-xds/src/xds/cache.rs +++ b/tonic-xds/src/xds/cache.rs @@ -33,7 +33,9 @@ use std::sync::Arc; use dashmap::DashMap; use tokio::sync::watch; -use crate::xds::resource::{ClusterResource, EndpointsResource, RouteConfigResource}; +use crate::xds::resource::{ + ClusterResource, EndpointsResource, ListenerResource, RouteConfigResource, +}; /// A wrapper around [`watch::Receiver`] that exposes only a single `next()` /// method, preventing misuse of the raw watch API. @@ -123,6 +125,10 @@ impl WatchMap { /// - **Cluster layer**: calls [`watch_cluster`](Self::watch_cluster) to await /// CDS readiness, then [`watch_endpoints`](Self::watch_endpoints) to track EDS updates. pub(crate) struct XdsCache { + /// Most recent Listener (LDS), surfaced so consumers can read listener-level + /// `metadata` (e.g. the control-plane-driven retry-config watcher). + listener_tx: watch::Sender>>, + /// Active route configuration (from LDS inline or RDS). route_config_tx: watch::Sender>>, @@ -136,14 +142,26 @@ pub(crate) struct XdsCache { impl XdsCache { /// Creates a new empty cache with no resources. pub(crate) fn new() -> Self { + let (listener_tx, _) = watch::channel(None); let (route_config_tx, _) = watch::channel(None); Self { + listener_tx, route_config_tx, clusters: WatchMap::new(), endpoints: WatchMap::new(), } } + /// Updates the active listener and notifies all watchers. + pub(crate) fn update_listener(&self, listener: Arc) { + self.listener_tx.send_replace(Some(listener)); + } + + /// Watches listener changes. + pub(crate) fn watch_listener(&self) -> CacheWatch { + CacheWatch::new(self.listener_tx.subscribe()) + } + /// Updates the active route configuration and notifies all watchers. pub(crate) fn update_route_config(&self, config: Arc) { self.route_config_tx.send_replace(Some(config)); diff --git a/tonic-xds/src/xds/resource/listener.rs b/tonic-xds/src/xds/resource/listener.rs index e56ead10e..64dc57bdd 100644 --- a/tonic-xds/src/xds/resource/listener.rs +++ b/tonic-xds/src/xds/resource/listener.rs @@ -33,7 +33,7 @@ use prost::Message; use xds_client::resource::TypeUrl; use xds_client::{Error, Resource}; -use super::route_config::RouteConfigResource; +use super::route_config::{RouteConfigMetadata, RouteConfigResource}; /// How the listener obtains its route configuration. #[derive(Debug)] @@ -47,11 +47,15 @@ pub(crate) enum RouteSource { /// Validated Listener resource. /// /// Extracts the route source from the -/// `ApiListener` -> `HttpConnectionManager` -> `route_specifier` chain. +/// `ApiListener` -> `HttpConnectionManager` -> `route_specifier` chain, and +/// preserves the listener-level `metadata` (`filter_metadata`) so consumers can +/// read config attached to the Listener (e.g. a retry-config extractor). #[derive(Debug)] pub(crate) struct ListenerResource { pub name: String, pub route_source: RouteSource, + /// Listener-level `metadata`. Empty when the Listener carried no metadata. + pub metadata: RouteConfigMetadata, } impl Resource for ListenerResource { @@ -72,6 +76,14 @@ impl Resource for ListenerResource { fn validate(message: Self::Message) -> xds_client::Result { let name = message.name; + // Preserve listener-level metadata (filter_metadata / typed_filter_metadata) + // so downstream consumers (e.g. a retry-config extractor) can read config + // attached to the Listener. + let metadata = message + .metadata + .map(RouteConfigMetadata::from_proto) + .unwrap_or_default(); + // gRPC listeners must have an ApiListener. let api_listener = message .api_listener @@ -98,6 +110,7 @@ impl Resource for ListenerResource { Ok(ListenerResource { name, route_source: RouteSource::Rds(rds.route_config_name), + metadata, }) } RouteSpecifier::RouteConfig(route_config) => { @@ -105,6 +118,7 @@ impl Resource for ListenerResource { Ok(ListenerResource { name, route_source: RouteSource::Inline(validated), + metadata, }) } RouteSpecifier::ScopedRoutes(_) => Err(Error::Validation( diff --git a/tonic-xds/src/xds/resource_manager.rs b/tonic-xds/src/xds/resource_manager.rs index 7f646d0a0..6ca814fef 100644 --- a/tonic-xds/src/xds/resource_manager.rs +++ b/tonic-xds/src/xds/resource_manager.rs @@ -180,6 +180,9 @@ impl CascadeState { result: Ok(listener), done, } => { + // Surface the listener (and its metadata) to consumers such as the + // control-plane-driven retry-config watcher. + cache.update_listener(Arc::clone(&listener)); match &listener.route_source { RouteSource::Inline(rc) => { // Drop any existing RDS watcher — routes are inline. @@ -371,6 +374,7 @@ mod tests { }], metadata: Default::default(), }), + metadata: Default::default(), }) } @@ -378,6 +382,7 @@ mod tests { Arc::new(ListenerResource { name: "listener".into(), route_source: RouteSource::Rds(rds_name.into()), + metadata: Default::default(), }) }