From 10e741c523b33fc470d1eddbc41ac71403030267 Mon Sep 17 00:00:00 2001 From: Michael Ingley Date: Tue, 23 Jun 2026 16:59:27 -0500 Subject: [PATCH 1/2] feat(tonic-xds): parse A32 circuit breaking config Signed-off-by: Michael Ingley --- tonic-xds/src/client/channel.rs | 1 + tonic-xds/src/xds/cache.rs | 1 + tonic-xds/src/xds/cluster_discovery.rs | 1 + .../src/xds/resource/circuit_breaking.rs | 123 ++++++++++++++++++ tonic-xds/src/xds/resource/cluster.rs | 36 ++++- tonic-xds/src/xds/resource/mod.rs | 1 + 6 files changed, 162 insertions(+), 1 deletion(-) create mode 100644 tonic-xds/src/xds/resource/circuit_breaking.rs diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index 5132d128a..27a0f6ab8 100644 --- a/tonic-xds/src/client/channel.rs +++ b/tonic-xds/src/client/channel.rs @@ -564,6 +564,7 @@ mod tests { eds_service_name: None, lb_policy: LbPolicy::RoundRobin, security: None, + circuit_breaking: Default::default(), }) } diff --git a/tonic-xds/src/xds/cache.rs b/tonic-xds/src/xds/cache.rs index 14773324f..f73f12fe8 100644 --- a/tonic-xds/src/xds/cache.rs +++ b/tonic-xds/src/xds/cache.rs @@ -195,6 +195,7 @@ mod tests { eds_service_name: None, lb_policy: lb, security: None, + circuit_breaking: Default::default(), }) } diff --git a/tonic-xds/src/xds/cluster_discovery.rs b/tonic-xds/src/xds/cluster_discovery.rs index 4c9e232a0..708df4c9f 100644 --- a/tonic-xds/src/xds/cluster_discovery.rs +++ b/tonic-xds/src/xds/cluster_discovery.rs @@ -319,6 +319,7 @@ mod tests { eds_service_name: None, lb_policy: LbPolicy::RoundRobin, security: None, + circuit_breaking: Default::default(), } } diff --git a/tonic-xds/src/xds/resource/circuit_breaking.rs b/tonic-xds/src/xds/resource/circuit_breaking.rs new file mode 100644 index 000000000..f6cda7801 --- /dev/null +++ b/tonic-xds/src/xds/resource/circuit_breaking.rs @@ -0,0 +1,123 @@ +//! Validated configuration types for [gRFC A32] circuit breaking. +//! +//! gRPC supports only the `max_requests` threshold from Envoy's CDS +//! `CircuitBreakers` config. Other threshold fields are intentionally ignored +//! because they are connection-pool or retry specific and do not apply to gRPC's +//! A32 request limiter. +//! +//! [gRFC A32]: https://github.com/grpc/proposal/blob/master/A32-xds-circuit-breaking.md + +use envoy_types::pb::envoy::config::cluster::v3::{CircuitBreakers, circuit_breakers::Thresholds}; +use envoy_types::pb::envoy::config::core::v3::RoutingPriority; + +/// Default max concurrent requests per cluster from A32. +pub(crate) const DEFAULT_MAX_REQUESTS: u32 = 1024; + +/// Validated A32 circuit-breaking configuration for a cluster. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct CircuitBreakingConfig { + /// Maximum number of in-flight requests allowed for the upstream cluster. + /// + /// This scaffolds the parsed CDS value only; enforcement is wired in a + /// follow-up change so request-lifetime accounting can be handled correctly + /// for streaming RPCs. + pub(crate) max_requests: u32, +} + +impl CircuitBreakingConfig { + /// Build circuit-breaking config from a CDS `CircuitBreakers` message. + /// + /// A32 uses the first threshold for `RoutingPriority::Default`. If no + /// applicable threshold or `max_requests` value is present, the gRPC default + /// of 1024 is used. + pub(crate) fn from_proto(circuit_breakers: Option<&CircuitBreakers>) -> Self { + let max_requests = circuit_breakers + .and_then(first_default_threshold) + .and_then(|threshold| threshold.max_requests.as_ref()) + .map(|value| value.value) + .unwrap_or(DEFAULT_MAX_REQUESTS); + + Self { max_requests } + } +} + +impl Default for CircuitBreakingConfig { + fn default() -> Self { + Self { + max_requests: DEFAULT_MAX_REQUESTS, + } + } +} + +fn first_default_threshold(circuit_breakers: &CircuitBreakers) -> Option<&Thresholds> { + circuit_breakers.thresholds.iter().find(|threshold| { + RoutingPriority::try_from(threshold.priority).ok() == Some(RoutingPriority::Default) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use envoy_types::pb::google::protobuf::UInt32Value; + + fn threshold(priority: RoutingPriority, max_requests: Option) -> Thresholds { + Thresholds { + priority: priority as i32, + max_requests: max_requests.map(|value| UInt32Value { value }), + ..Default::default() + } + } + + #[test] + fn defaults_when_circuit_breakers_absent() { + assert_eq!( + CircuitBreakingConfig::from_proto(None), + CircuitBreakingConfig { + max_requests: DEFAULT_MAX_REQUESTS, + } + ); + } + + #[test] + fn defaults_when_default_threshold_absent() { + let circuit_breakers = CircuitBreakers { + thresholds: vec![threshold(RoutingPriority::High, Some(7))], + ..Default::default() + }; + + assert_eq!( + CircuitBreakingConfig::from_proto(Some(&circuit_breakers)).max_requests, + DEFAULT_MAX_REQUESTS + ); + } + + #[test] + fn defaults_when_default_threshold_has_no_max_requests() { + let circuit_breakers = CircuitBreakers { + thresholds: vec![threshold(RoutingPriority::Default, None)], + ..Default::default() + }; + + assert_eq!( + CircuitBreakingConfig::from_proto(Some(&circuit_breakers)).max_requests, + DEFAULT_MAX_REQUESTS + ); + } + + #[test] + fn uses_first_default_threshold_max_requests() { + let circuit_breakers = CircuitBreakers { + thresholds: vec![ + threshold(RoutingPriority::High, Some(9)), + threshold(RoutingPriority::Default, Some(11)), + threshold(RoutingPriority::Default, Some(13)), + ], + ..Default::default() + }; + + assert_eq!( + CircuitBreakingConfig::from_proto(Some(&circuit_breakers)).max_requests, + 11 + ); + } +} diff --git a/tonic-xds/src/xds/resource/cluster.rs b/tonic-xds/src/xds/resource/cluster.rs index a21619e32..435062928 100644 --- a/tonic-xds/src/xds/resource/cluster.rs +++ b/tonic-xds/src/xds/resource/cluster.rs @@ -6,6 +6,7 @@ use prost::Message; use xds_client::resource::TypeUrl; use xds_client::{Error, Resource}; +use super::circuit_breaking::CircuitBreakingConfig; use super::security::{ClusterSecurityConfig, parse_transport_socket}; /// Validated Cluster resource. @@ -20,6 +21,8 @@ pub(crate) struct ClusterResource { /// TLS security config parsed from `transport_socket`. `None` means the /// cluster uses plaintext connections. pub security: Option, + /// Parsed A32 circuit-breaking config. Enforcement is wired separately. + pub circuit_breaking: CircuitBreakingConfig, } /// Load balancing policies. @@ -67,12 +70,14 @@ impl Resource for ClusterResource { }; let security = parse_transport_socket(message.transport_socket)?; + let circuit_breaking = CircuitBreakingConfig::from_proto(message.circuit_breakers.as_ref()); Ok(ClusterResource { name, eds_service_name, lb_policy, security, + circuit_breaking, }) } } @@ -88,7 +93,12 @@ impl ClusterResource { #[cfg(test)] mod tests { use super::*; - use envoy_types::pb::envoy::config::cluster::v3::cluster::EdsClusterConfig; + use crate::xds::resource::circuit_breaking::DEFAULT_MAX_REQUESTS; + use envoy_types::pb::envoy::config::cluster::v3::{ + CircuitBreakers, circuit_breakers::Thresholds, cluster::EdsClusterConfig, + }; + use envoy_types::pb::envoy::config::core::v3::RoutingPriority; + use envoy_types::pb::google::protobuf::UInt32Value; fn make_cluster(name: &str) -> Cluster { Cluster { @@ -105,6 +115,10 @@ mod tests { assert_eq!(validated.name, "my-cluster"); assert_eq!(validated.lb_policy, LbPolicy::RoundRobin); assert!(validated.eds_service_name.is_none()); + assert_eq!( + validated.circuit_breaking.max_requests, + DEFAULT_MAX_REQUESTS + ); } #[test] @@ -141,6 +155,26 @@ mod tests { assert_eq!(validated.lb_policy, LbPolicy::LeastRequest); } + #[test] + fn test_circuit_breaking_max_requests() { + let cluster = Cluster { + name: "cb-cluster".to_string(), + lb_policy: cluster::LbPolicy::RoundRobin as i32, + circuit_breakers: Some(CircuitBreakers { + thresholds: vec![Thresholds { + priority: RoutingPriority::Default as i32, + max_requests: Some(UInt32Value { value: 17 }), + ..Default::default() + }], + ..Default::default() + }), + ..Default::default() + }; + + let validated = ClusterResource::validate(cluster).unwrap(); + assert_eq!(validated.circuit_breaking.max_requests, 17); + } + #[test] fn test_unsupported_lb_policy_is_rejected() { let cluster = Cluster { diff --git a/tonic-xds/src/xds/resource/mod.rs b/tonic-xds/src/xds/resource/mod.rs index e3e90daea..aa64fc940 100644 --- a/tonic-xds/src/xds/resource/mod.rs +++ b/tonic-xds/src/xds/resource/mod.rs @@ -10,6 +10,7 @@ //! //! These are *validated* types containing only the fields relevant to gRPC +pub(crate) mod circuit_breaking; pub(crate) mod cluster; pub(crate) mod endpoints; pub(crate) mod hash_policy; From 9afebba99a96f5071b7093bdb95a0c83284fab98 Mon Sep 17 00:00:00 2001 From: Michael Ingley Date: Thu, 25 Jun 2026 14:04:23 -0500 Subject: [PATCH 2/2] fix(tonic-xds): defer circuit breaking CDS wiring Signed-off-by: Michael Ingley --- tonic-xds/src/client/channel.rs | 1 - tonic-xds/src/xds/cache.rs | 1 - tonic-xds/src/xds/cluster_discovery.rs | 1 - .../src/xds/resource/circuit_breaking.rs | 4 +++ tonic-xds/src/xds/resource/cluster.rs | 36 +------------------ 5 files changed, 5 insertions(+), 38 deletions(-) diff --git a/tonic-xds/src/client/channel.rs b/tonic-xds/src/client/channel.rs index 27a0f6ab8..5132d128a 100644 --- a/tonic-xds/src/client/channel.rs +++ b/tonic-xds/src/client/channel.rs @@ -564,7 +564,6 @@ mod tests { eds_service_name: None, lb_policy: LbPolicy::RoundRobin, security: None, - circuit_breaking: Default::default(), }) } diff --git a/tonic-xds/src/xds/cache.rs b/tonic-xds/src/xds/cache.rs index f73f12fe8..14773324f 100644 --- a/tonic-xds/src/xds/cache.rs +++ b/tonic-xds/src/xds/cache.rs @@ -195,7 +195,6 @@ mod tests { eds_service_name: None, lb_policy: lb, security: None, - circuit_breaking: Default::default(), }) } diff --git a/tonic-xds/src/xds/cluster_discovery.rs b/tonic-xds/src/xds/cluster_discovery.rs index 708df4c9f..4c9e232a0 100644 --- a/tonic-xds/src/xds/cluster_discovery.rs +++ b/tonic-xds/src/xds/cluster_discovery.rs @@ -319,7 +319,6 @@ mod tests { eds_service_name: None, lb_policy: LbPolicy::RoundRobin, security: None, - circuit_breaking: Default::default(), } } diff --git a/tonic-xds/src/xds/resource/circuit_breaking.rs b/tonic-xds/src/xds/resource/circuit_breaking.rs index f6cda7801..28c7200bc 100644 --- a/tonic-xds/src/xds/resource/circuit_breaking.rs +++ b/tonic-xds/src/xds/resource/circuit_breaking.rs @@ -5,6 +5,10 @@ //! because they are connection-pool or retry specific and do not apply to gRPC's //! A32 request limiter. //! +//! This parser intentionally stays detached from `ClusterResource` until +//! enforcement lands; otherwise cluster validation would advertise support before +//! requests are actually limited. +//! //! [gRFC A32]: https://github.com/grpc/proposal/blob/master/A32-xds-circuit-breaking.md use envoy_types::pb::envoy::config::cluster::v3::{CircuitBreakers, circuit_breakers::Thresholds}; diff --git a/tonic-xds/src/xds/resource/cluster.rs b/tonic-xds/src/xds/resource/cluster.rs index 435062928..a21619e32 100644 --- a/tonic-xds/src/xds/resource/cluster.rs +++ b/tonic-xds/src/xds/resource/cluster.rs @@ -6,7 +6,6 @@ use prost::Message; use xds_client::resource::TypeUrl; use xds_client::{Error, Resource}; -use super::circuit_breaking::CircuitBreakingConfig; use super::security::{ClusterSecurityConfig, parse_transport_socket}; /// Validated Cluster resource. @@ -21,8 +20,6 @@ pub(crate) struct ClusterResource { /// TLS security config parsed from `transport_socket`. `None` means the /// cluster uses plaintext connections. pub security: Option, - /// Parsed A32 circuit-breaking config. Enforcement is wired separately. - pub circuit_breaking: CircuitBreakingConfig, } /// Load balancing policies. @@ -70,14 +67,12 @@ impl Resource for ClusterResource { }; let security = parse_transport_socket(message.transport_socket)?; - let circuit_breaking = CircuitBreakingConfig::from_proto(message.circuit_breakers.as_ref()); Ok(ClusterResource { name, eds_service_name, lb_policy, security, - circuit_breaking, }) } } @@ -93,12 +88,7 @@ impl ClusterResource { #[cfg(test)] mod tests { use super::*; - use crate::xds::resource::circuit_breaking::DEFAULT_MAX_REQUESTS; - use envoy_types::pb::envoy::config::cluster::v3::{ - CircuitBreakers, circuit_breakers::Thresholds, cluster::EdsClusterConfig, - }; - use envoy_types::pb::envoy::config::core::v3::RoutingPriority; - use envoy_types::pb::google::protobuf::UInt32Value; + use envoy_types::pb::envoy::config::cluster::v3::cluster::EdsClusterConfig; fn make_cluster(name: &str) -> Cluster { Cluster { @@ -115,10 +105,6 @@ mod tests { assert_eq!(validated.name, "my-cluster"); assert_eq!(validated.lb_policy, LbPolicy::RoundRobin); assert!(validated.eds_service_name.is_none()); - assert_eq!( - validated.circuit_breaking.max_requests, - DEFAULT_MAX_REQUESTS - ); } #[test] @@ -155,26 +141,6 @@ mod tests { assert_eq!(validated.lb_policy, LbPolicy::LeastRequest); } - #[test] - fn test_circuit_breaking_max_requests() { - let cluster = Cluster { - name: "cb-cluster".to_string(), - lb_policy: cluster::LbPolicy::RoundRobin as i32, - circuit_breakers: Some(CircuitBreakers { - thresholds: vec![Thresholds { - priority: RoutingPriority::Default as i32, - max_requests: Some(UInt32Value { value: 17 }), - ..Default::default() - }], - ..Default::default() - }), - ..Default::default() - }; - - let validated = ClusterResource::validate(cluster).unwrap(); - assert_eq!(validated.circuit_breaking.max_requests, 17); - } - #[test] fn test_unsupported_lb_policy_is_rejected() { let cluster = Cluster {