diff --git a/grpc-xds/Cargo.toml b/grpc-xds/Cargo.toml index 0433cfc82..aa77ef186 100644 --- a/grpc-xds/Cargo.toml +++ b/grpc-xds/Cargo.toml @@ -20,6 +20,9 @@ allowed_external_types = [] [dependencies] protobuf = "4.35.1-release" protobuf-well-known-types = "4.35.1-release" +bytes = "1.11.0" +xds-client = { version = "0.1.0-alpha.2", path = "../xds-client", default-features = false } +regex = "1" [build-dependencies] protobuf-codegen = "4.35.1-release" diff --git a/grpc-xds/src/lib.rs b/grpc-xds/src/lib.rs index 94845582e..b9b0b3dca 100644 --- a/grpc-xds/src/lib.rs +++ b/grpc-xds/src/lib.rs @@ -42,6 +42,16 @@ pub(crate) mod generated { include!(concat!(env!("OUT_DIR"), "/generated/mod.rs")); } +/// Validated xDS resource types (LDS/RDS/CDS/EDS), each implementing +/// `xds_client::Resource` so they can be deserialized, named, and validated +/// per their respective gRFCs (A27 core resource types, A28 route matching, +/// A37 aggregate clusters). +pub(crate) mod resource; + +/// [`xds_config::XdsConfig`]: the atomic xDS configuration snapshot assembled +/// from a channel's Listener/RouteConfiguration/Cluster/Endpoints resources. +pub(crate) mod xds_config; + #[cfg(test)] mod tests { //! Sanity checks that the generated xDS modules are importable and usable diff --git a/grpc-xds/src/resource/cluster.rs b/grpc-xds/src/resource/cluster.rs new file mode 100644 index 000000000..c68dacccd --- /dev/null +++ b/grpc-xds/src/resource/cluster.rs @@ -0,0 +1,507 @@ +/* + * + * Copyright 2025 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +//! Validated Cluster resource (CDS). + +use protobuf::Parse; +use xds_client::resource::TypeUrl; +use xds_client::{Error, Resource}; + +use crate::generated::envoy::config::cluster::v3::cluster::DiscoveryType; +use crate::generated::envoy::config::cluster::v3::{Cluster, cluster::ClusterDiscoveryTypeOneof}; +use crate::generated::envoy::extensions::clusters::aggregate::v3::ClusterConfig as AggregateClusterConfig; + +/// Extension name and `typed_config` type for gRFC A37 aggregate clusters. +const AGGREGATE_CLUSTER_NAME: &str = "envoy.clusters.aggregate"; +const AGGREGATE_CLUSTER_CONFIG_TYPE_URL: &str = + "type.googleapis.com/envoy.extensions.clusters.aggregate.v3.ClusterConfig"; + +// TODO: model `transport_socket` (security) and load-balancing-policy +// config once the dependency manager and LB policies exist to consume them. +/// Validated Cluster resource. +/// +/// Only the discovery mechanism is modeled for now (gRFC A27/A37). +#[derive(Debug, Clone)] +pub(crate) struct ClusterResource { + pub(crate) name: String, + pub(crate) discovery: ClusterDiscovery, +} + +/// How a cluster's endpoints are discovered, mirroring the +/// `Cluster.cluster_discovery_type` oneof plus the `envoy.clusters.aggregate` +/// custom cluster type extension (gRFC A37). +/// +/// `STATIC`, `STRICT_DNS`, and `ORIGINAL_DST` are not supported: they have no +/// gRPC xDS use case. +#[derive(Debug, Clone)] +pub(crate) enum ClusterDiscovery { + /// Endpoints are discovered via EDS. When CDS resource left `eds_service_name` unset, + /// it is resolved to the cluster name. + Eds { eds_service_name: String }, + /// Endpoints are discovered via DNS resolution of a single target. + LogicalDns { hostname: String, port: u16 }, + /// This is an aggregate cluster (gRFC A37): traffic falls over across + /// `children` in priority order. Each child is itself a top-level cluster + /// resolved independently (see `crate::xds_config::XdsConfig::clusters`). + Aggregate { children: Vec }, +} + +impl Resource for ClusterResource { + type Message = Cluster; + + const TYPE_URL: TypeUrl = TypeUrl::new("type.googleapis.com/envoy.config.cluster.v3.Cluster"); + + const ALL_RESOURCES_REQUIRED_IN_SOTW: bool = true; + + fn deserialize(bytes: bytes::Bytes) -> xds_client::Result { + Cluster::parse(&bytes) + .map_err(|e| Error::Validation(format!("failed to decode Cluster: {e}"))) + } + + fn name(message: &Self::Message) -> &str { + message.name().to_str().unwrap_or_default() + } + + fn validate(message: Self::Message) -> xds_client::Result { + let name = message.name().to_str().unwrap_or_default().to_string(); + if name.is_empty() { + return Err(Error::Validation("cluster name is empty".into())); + } + + let discovery = match message.cluster_discovery_type() { + ClusterDiscoveryTypeOneof::Type(DiscoveryType::Eds) => { + validate_eds_discovery(&message, &name)? + } + ClusterDiscoveryTypeOneof::Type(DiscoveryType::LogicalDns) => { + validate_logical_dns_discovery(&message)? + } + ClusterDiscoveryTypeOneof::ClusterType(custom) + if custom.name().to_str().unwrap_or_default() == AGGREGATE_CLUSTER_NAME => + { + validate_aggregate_discovery(custom)? + } + other => { + return Err(Error::Validation(format!( + "unsupported cluster discovery type: {other:?}" + ))); + } + }; + + Ok(ClusterResource { name, discovery }) + } +} + +fn validate_eds_discovery( + message: &Cluster, + cluster_name: &str, +) -> xds_client::Result { + let eds_cluster_config = message.eds_cluster_config(); + + // Per gRFC A27: eds_config must be set, and must point at ADS or Self. + if !eds_cluster_config.has_eds_config() { + return Err(Error::Validation( + "CDS's EDS config source is not set".into(), + )); + } + let config_source = eds_cluster_config.eds_config(); + if !config_source.has_ads() && !config_source.has_self() { + return Err(Error::Validation( + "CDS's EDS config source is not ADS or Self".into(), + )); + } + + let eds_service_name = eds_cluster_config + .service_name() + .to_str() + .unwrap_or_default(); + let eds_service_name = if eds_service_name.is_empty() { + // Per gRFC A47, `xdstp:`-scheme (federation) cluster names must set an + // explicit EDS service name rather than relying on the cluster-name fallback. + if cluster_name.starts_with("xdstp:") { + return Err(Error::Validation( + "CDS's EDS service name is not set with a new-style cluster name".into(), + )); + } + cluster_name.to_string() + } else { + eds_service_name.to_string() + }; + + Ok(ClusterDiscovery::Eds { eds_service_name }) +} + +fn validate_logical_dns_discovery(message: &Cluster) -> xds_client::Result { + if !message.has_load_assignment() { + return Err(Error::Validation( + "load_assignment not present for LOGICAL_DNS cluster".into(), + )); + } + let load_assignment = message.load_assignment(); + + let localities = load_assignment.endpoints(); + if localities.len() != 1 { + return Err(Error::Validation(format!( + "load_assignment for LOGICAL_DNS cluster must have exactly one locality, got {}", + localities.len() + ))); + } + let lb_endpoints = localities.get(0).expect("checked len == 1").lb_endpoints(); + if lb_endpoints.len() != 1 { + return Err(Error::Validation(format!( + "locality for LOGICAL_DNS cluster must have exactly one endpoint, got {}", + lb_endpoints.len() + ))); + } + let lb_endpoint = lb_endpoints.get(0).expect("checked len == 1"); + + if !lb_endpoint.has_endpoint() { + return Err(Error::Validation( + "endpoint for LOGICAL_DNS cluster not set".into(), + )); + } + let endpoint = lb_endpoint.endpoint(); + + if !endpoint.has_address() { + return Err(Error::Validation( + "socket address for endpoint for LOGICAL_DNS cluster not set".into(), + )); + } + let address = endpoint.address(); + if !address.has_socket_address() { + return Err(Error::Validation( + "socket address for endpoint for LOGICAL_DNS cluster not set".into(), + )); + } + let socket_address = address.socket_address(); + + let resolver_name = socket_address.resolver_name(); + if !resolver_name.is_empty() { + return Err(Error::Validation(format!( + "socket address for endpoint for LOGICAL_DNS cluster has unexpected custom resolver name: {resolver_name}" + ))); + } + + let hostname = socket_address.address().to_str().unwrap_or_default(); + if hostname.is_empty() { + return Err(Error::Validation( + "host for endpoint for LOGICAL_DNS cluster not set".into(), + )); + } + let port = socket_address.port_value(); + if port == 0 { + return Err(Error::Validation( + "port for endpoint for LOGICAL_DNS cluster not set".into(), + )); + } + let port = u16::try_from(port).map_err(|_| { + Error::Validation(format!( + "port for endpoint for LOGICAL_DNS cluster is out of range: {port}" + )) + })?; + + Ok(ClusterDiscovery::LogicalDns { + hostname: hostname.to_string(), + port, + }) +} + +fn validate_aggregate_discovery( + custom: crate::generated::envoy::config::cluster::v3::cluster::CustomClusterTypeView<'_>, +) -> xds_client::Result { + if !custom.has_typed_config() { + return Err(Error::Validation( + "aggregate cluster missing typed_config".into(), + )); + } + let any = custom.typed_config(); + let type_url = any.type_url().to_str().unwrap_or_default(); + if type_url != AGGREGATE_CLUSTER_CONFIG_TYPE_URL { + return Err(Error::Validation(format!( + "unexpected aggregate cluster typed_config type_url: '{type_url}'" + ))); + } + let cluster_config = AggregateClusterConfig::parse(any.value()).map_err(|e| { + Error::Validation(format!("failed to unmarshal aggregate cluster config: {e}")) + })?; + + let children: Vec = cluster_config + .clusters() + .into_iter() + .map(|c| c.to_str().unwrap_or_default().to_string()) + .collect(); + if children.is_empty() { + return Err(Error::Validation( + "aggregate cluster has empty clusters field in response".into(), + )); + } + + Ok(ClusterDiscovery::Aggregate { children }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::generated::envoy::config::cluster::v3::cluster::{ + CustomClusterType, EdsClusterConfig, + }; + use crate::generated::envoy::config::core::v3::{ + Address, AggregatedConfigSource, ConfigSource, SocketAddress, + }; + use crate::generated::envoy::config::endpoint::v3::{ + ClusterLoadAssignment, Endpoint, LbEndpoint, LocalityLbEndpoints, + }; + use protobuf::Serialize; + use protobuf_well_known_types::Any; + + fn make_ads_eds_config() -> EdsClusterConfig { + let mut eds_cfg = EdsClusterConfig::new(); + let mut config_source = ConfigSource::new(); + config_source.set_ads(AggregatedConfigSource::new()); + eds_cfg.set_eds_config(config_source); + eds_cfg + } + + fn make_cluster(name: &str) -> Cluster { + let mut cluster = Cluster::new(); + cluster.set_name(name); + cluster.set_type(DiscoveryType::Eds); + cluster.set_eds_cluster_config(make_ads_eds_config()); + cluster + } + + fn make_logical_dns_cluster(host: &str, port: u32) -> Cluster { + let mut cluster = Cluster::new(); + cluster.set_name("dns-cluster"); + cluster.set_type(DiscoveryType::LogicalDns); + + let mut socket_address = SocketAddress::new(); + socket_address.set_address(host); + socket_address.set_port_value(port); + let mut address = Address::new(); + address.set_socket_address(socket_address); + let mut endpoint = Endpoint::new(); + endpoint.set_address(address); + let mut lb_endpoint = LbEndpoint::new(); + lb_endpoint.set_endpoint(endpoint); + let mut locality_lb_endpoints = LocalityLbEndpoints::new(); + locality_lb_endpoints.lb_endpoints_mut().push(lb_endpoint); + let mut cla = ClusterLoadAssignment::new(); + cla.endpoints_mut().push(locality_lb_endpoints); + cluster.set_load_assignment(cla); + cluster + } + + #[test] + fn validate_eds_basic() { + let cluster = make_cluster("my-cluster"); + let validated = ClusterResource::validate(cluster).expect("should validate"); + assert_eq!(validated.name, "my-cluster"); + match validated.discovery { + ClusterDiscovery::Eds { eds_service_name } => { + assert_eq!(eds_service_name, "my-cluster"); + } + other => panic!("expected Eds, got {other:?}"), + } + } + + #[test] + fn validate_eds_service_name_override() { + let mut cluster = make_cluster("my-cluster"); + let mut eds_cfg = make_ads_eds_config(); + eds_cfg.set_service_name("eds-svc"); + cluster.set_eds_cluster_config(eds_cfg); + let validated = ClusterResource::validate(cluster).unwrap(); + match validated.discovery { + ClusterDiscovery::Eds { eds_service_name } => assert_eq!(eds_service_name, "eds-svc"), + other => panic!("expected Eds, got {other:?}"), + } + } + + #[test] + fn validate_eds_rejects_missing_eds_config() { + let mut cluster = Cluster::new(); + cluster.set_name("my-cluster"); + cluster.set_type(DiscoveryType::Eds); + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!(err.to_string().contains("EDS config source is not set")); + } + + #[test] + fn validate_eds_xdstp_name_requires_service_name() { + let cluster = make_cluster("xdstp://example.com/envoy.config.cluster.v3.Cluster/foo"); + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!(err.to_string().contains("new-style cluster name")); + } + + #[test] + fn validate_eds_rejects_non_ads_non_self_config_source() { + let mut cluster = make_cluster("my-cluster"); + let mut eds_cfg = EdsClusterConfig::new(); + let mut config_source = ConfigSource::new(); + config_source.set_path("/some/path"); + eds_cfg.set_eds_config(config_source); + cluster.set_eds_cluster_config(eds_cfg); + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!(err.to_string().contains("not ADS or Self")); + } + + #[test] + fn validate_eds_accepts_ads_config_source() { + let mut cluster = make_cluster("my-cluster"); + let mut eds_cfg = EdsClusterConfig::new(); + let mut config_source = ConfigSource::new(); + config_source.set_ads(AggregatedConfigSource::new()); + eds_cfg.set_eds_config(config_source); + cluster.set_eds_cluster_config(eds_cfg); + assert!(ClusterResource::validate(cluster).is_ok()); + } + + #[test] + fn validate_logical_dns() { + let cluster = make_logical_dns_cluster("example.com", 443); + let validated = ClusterResource::validate(cluster).expect("should validate"); + match validated.discovery { + ClusterDiscovery::LogicalDns { hostname, port } => { + assert_eq!(hostname, "example.com"); + assert_eq!(port, 443); + } + other => panic!("expected LogicalDns, got {other:?}"), + } + } + + #[test] + fn validate_logical_dns_rejects_out_of_range_port() { + // 65536 must be rejected outright rather than truncated to 0. + let cluster = make_logical_dns_cluster("example.com", 65_536); + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!(err.to_string().contains("out of range")); + } + + #[test] + fn validate_logical_dns_missing_load_assignment() { + let mut cluster = Cluster::new(); + cluster.set_name("dns-cluster"); + cluster.set_type(DiscoveryType::LogicalDns); + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!(err.to_string().contains("load_assignment not present")); + } + + #[test] + fn validate_aggregate() { + let mut inner = AggregateClusterConfig::new(); + inner.clusters_mut().push("child-a"); + inner.clusters_mut().push("child-b"); + + let mut any = Any::new(); + any.set_type_url( + "type.googleapis.com/envoy.extensions.clusters.aggregate.v3.ClusterConfig", + ); + any.set_value(inner.serialize().expect("serialize")); + + let mut custom = CustomClusterType::new(); + custom.set_name("envoy.clusters.aggregate"); + custom.set_typed_config(any); + + let mut cluster = Cluster::new(); + cluster.set_name("aggregate-cluster"); + cluster.set_cluster_type(custom); + + let validated = ClusterResource::validate(cluster).expect("should validate"); + match validated.discovery { + ClusterDiscovery::Aggregate { children } => { + assert_eq!(children, vec!["child-a".to_string(), "child-b".to_string()]); + } + other => panic!("expected Aggregate, got {other:?}"), + } + } + + #[test] + fn validate_aggregate_rejects_unexpected_typed_config_type_url() { + let mut inner = AggregateClusterConfig::new(); + inner.clusters_mut().push("child-a"); + + let mut any = Any::new(); + any.set_type_url("type.googleapis.com/envoy.config.cluster.v3.Cluster"); + any.set_value(inner.serialize().expect("serialize")); + + let mut custom = CustomClusterType::new(); + custom.set_name("envoy.clusters.aggregate"); + custom.set_typed_config(any); + + let mut cluster = Cluster::new(); + cluster.set_name("aggregate-cluster"); + cluster.set_cluster_type(custom); + + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!(err.to_string().contains("type_url")); + } + + #[test] + fn validate_aggregate_rejects_empty_children() { + let inner = AggregateClusterConfig::new(); + let mut any = Any::new(); + any.set_type_url( + "type.googleapis.com/envoy.extensions.clusters.aggregate.v3.ClusterConfig", + ); + any.set_value(inner.serialize().expect("serialize")); + + let mut custom = CustomClusterType::new(); + custom.set_name("envoy.clusters.aggregate"); + custom.set_typed_config(any); + + let mut cluster = Cluster::new(); + cluster.set_name("aggregate-cluster"); + cluster.set_cluster_type(custom); + + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!(err.to_string().contains("empty clusters field")); + } + + #[test] + fn validate_empty_name() { + let cluster = make_cluster(""); + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!(err.to_string().contains("cluster name is empty")); + } + + #[test] + fn validate_unsupported_discovery_type_rejected() { + let mut cluster = Cluster::new(); + cluster.set_name("static-cluster"); + cluster.set_type(DiscoveryType::Static); + let err = ClusterResource::validate(cluster).unwrap_err(); + assert!( + err.to_string() + .contains("unsupported cluster discovery type") + ); + } + + #[test] + fn deserialize_roundtrip() { + let cluster = make_cluster("test"); + let bytes = cluster.serialize().expect("serialize"); + let deserialized = ClusterResource::deserialize(bytes::Bytes::from(bytes)).unwrap(); + assert_eq!(ClusterResource::name(&deserialized), "test"); + } +} diff --git a/grpc-xds/src/resource/endpoint.rs b/grpc-xds/src/resource/endpoint.rs new file mode 100644 index 000000000..8ce280aac --- /dev/null +++ b/grpc-xds/src/resource/endpoint.rs @@ -0,0 +1,656 @@ +/* + * + * Copyright 2025 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +//! Validated ClusterLoadAssignment resource (EDS). + +use std::collections::{HashMap, HashSet}; + +use protobuf::Parse; +use xds_client::resource::TypeUrl; +use xds_client::{Error, Resource}; + +use crate::generated::envoy::config::core::v3::HealthStatus as EnvoyHealthStatus; +use crate::generated::envoy::config::endpoint::v3::{ + ClusterLoadAssignment, lb_endpoint::HostIdentifierOneof, +}; + +/// Validated ClusterLoadAssignment (EDS resource). +#[derive(Debug, Clone)] +pub(crate) struct EndpointsResource { + pub(crate) cluster_name: String, + pub(crate) localities: Vec, +} + +/// Endpoints within a single locality. +#[derive(Debug, Clone)] +pub(crate) struct LocalityLbEndpoints { + pub(crate) locality: Locality, + pub(crate) endpoints: Vec, + pub(crate) load_balancing_weight: u32, + pub(crate) priority: u32, +} + +// TODO: consider unifying with `xds_client::message::Locality` (same +// shape, used for `Node.locality`) and tonic-xds's own copy of this type. +/// Locality information for a set of endpoints. +/// +/// Kept local for now instead of reusing `xds_client::message::Locality`. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct Locality { + pub(crate) region: String, + pub(crate) zone: String, + pub(crate) sub_zone: String, +} + +/// A single validated endpoint. +#[derive(Debug, Clone)] +pub(crate) struct LbEndpoint { + pub(crate) address: EndpointAddress, + pub(crate) health_status: HealthStatus, + pub(crate) load_balancing_weight: u32, +} + +// TODO: reuse `grpc::client::name_resolution::Address` when wiring resolver. +/// A resolved `host:port` endpoint address extracted from a `SocketAddress`. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct EndpointAddress { + pub(crate) host: String, + pub(crate) port: u16, +} + +/// Health status of an endpoint (gRFC A27). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum HealthStatus { + Unknown, + Healthy, + Unhealthy, + Draining, + Degraded, + /// A status this build does not recognize, carrying the raw wire value. + /// + /// Treated as unusable, like every status other than `Unknown` and + /// `Healthy`. + Other(i32), +} + +impl From for HealthStatus { + fn from(value: EnvoyHealthStatus) -> Self { + match value { + EnvoyHealthStatus::Unknown => Self::Unknown, + EnvoyHealthStatus::Healthy => Self::Healthy, + // Envoy's TIMEOUT is documented as "interpreted by Envoy as + // UNHEALTHY". + EnvoyHealthStatus::Unhealthy | EnvoyHealthStatus::Timeout => Self::Unhealthy, + EnvoyHealthStatus::Draining => Self::Draining, + EnvoyHealthStatus::Degraded => Self::Degraded, + // `HealthStatus` is an open enum, so a newer control plane can send + // a status this build does not know. Per gRFC A27 only HEALTHY and + // UNKNOWN are usable, so an unrecognized status fails closed. + other => Self::Other(i32::from(other)), + } + } +} + +impl Resource for EndpointsResource { + type Message = ClusterLoadAssignment; + + const TYPE_URL: TypeUrl = + TypeUrl::new("type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"); + + const ALL_RESOURCES_REQUIRED_IN_SOTW: bool = false; + + fn deserialize(bytes: bytes::Bytes) -> xds_client::Result { + ClusterLoadAssignment::parse(&bytes) + .map_err(|e| Error::Validation(format!("failed to decode ClusterLoadAssignment: {e}"))) + } + + fn name(message: &Self::Message) -> &str { + message.cluster_name().to_str().unwrap_or_default() + } + + fn validate(message: Self::Message) -> xds_client::Result { + let cluster_name = message + .cluster_name() + .to_str() + .unwrap_or_default() + .to_string(); + if cluster_name.is_empty() { + return Err(Error::Validation( + "ClusterLoadAssignment missing cluster_name".into(), + )); + } + + let mut localities = Vec::new(); + // Per gRFC A27, all of these must hold across the whole resource: + // endpoint addresses are unique, a locality appears at most once per + // priority, and locality weights at one priority fit in a u32. + let mut seen_addresses = HashSet::new(); + let mut seen_localities = HashSet::new(); + let mut weight_sums: HashMap = HashMap::new(); + + for locality_endpoints in message.endpoints().iter() { + let Some(l) = locality_endpoints.locality_opt() else { + return Err(Error::Validation( + "ClusterLoadAssignment contains a locality without an ID".into(), + )); + }; + let locality = Locality { + region: l.region().to_str().unwrap_or_default().to_string(), + zone: l.zone().to_str().unwrap_or_default().to_string(), + sub_zone: l.sub_zone().to_str().unwrap_or_default().to_string(), + }; + + // Per gRFC A27: skip localities with no usable weight. An unset + // weight reads as 0 here, which is the same "skip" case. + let load_balancing_weight = locality_endpoints.load_balancing_weight().value(); + if load_balancing_weight == 0 { + continue; + } + + let priority = locality_endpoints.priority(); + + let sum = weight_sums.entry(priority).or_default(); + *sum += u64::from(load_balancing_weight); + if *sum > u64::from(u32::MAX) { + return Err(Error::Validation(format!( + "sum of locality weights at priority {priority} exceeds {}", + u32::MAX + ))); + } + + if !seen_localities.insert((locality.clone(), priority)) { + return Err(Error::Validation(format!( + "duplicate locality {locality:?} at priority {priority}" + ))); + } + + let mut endpoints = Vec::new(); + let mut endpoint_weight_sum: u64 = 0; + for lb_ep in locality_endpoints.lb_endpoints().iter() { + let Some(ep) = validate_lb_endpoint(lb_ep)? else { + continue; + }; + endpoint_weight_sum += u64::from(ep.load_balancing_weight); + if endpoint_weight_sum > u64::from(u32::MAX) { + return Err(Error::Validation(format!( + "sum of endpoint weights in locality {locality:?} exceeds {}", + u32::MAX + ))); + } + if !seen_addresses.insert(ep.address.clone()) { + return Err(Error::Validation(format!( + "duplicate endpoint address {}:{}", + ep.address.host, ep.address.port + ))); + } + endpoints.push(ep); + } + + localities.push(LocalityLbEndpoints { + locality, + endpoints, + load_balancing_weight, + priority, + }); + } + + // Per gRFC A27: priorities must run 0..N with no gaps. + let priorities: HashSet = localities.iter().map(|l| l.priority).collect(); + for priority in 0..priorities.len() as u32 { + if !priorities.contains(&priority) { + return Err(Error::Validation(format!( + "priority {priority} missing from ClusterLoadAssignment" + ))); + } + } + + Ok(EndpointsResource { + cluster_name, + localities, + }) + } +} + +fn validate_lb_endpoint( + lb_ep: crate::generated::envoy::config::endpoint::v3::LbEndpointView<'_>, +) -> xds_client::Result> { + let health_status = HealthStatus::from(lb_ep.health_status()); + + let endpoint = match lb_ep.host_identifier() { + HostIdentifierOneof::Endpoint(ep) => ep, + // Skip unsupported host_identifier variants (e.g. `endpoint_name`, + // used for LRS-only named endpoints) rather than NACKing the whole + // resource -- the control plane may be serving both Envoy proxies + // and gRPC clients. + _ => return Ok(None), + }; + + if !endpoint.has_address() { + return Err(Error::Validation("endpoint missing address".into())); + } + let address = endpoint.address(); + if !address.has_socket_address() { + return Err(Error::Validation( + "only socket addresses are supported for gRPC endpoints".into(), + )); + } + let socket_address = address.socket_address(); + + if !socket_address.has_port_value() { + return Err(Error::Validation( + "endpoint address missing numeric port".into(), + )); + } + + let host = socket_address + .address() + .to_str() + .unwrap_or_default() + .to_string(); + // Per gRFC A27: the address field must be set. + if host.is_empty() { + return Err(Error::Validation("endpoint address is empty".into())); + } + let port = socket_address.port_value(); + let port = u16::try_from(port) + .map_err(|_| Error::Validation(format!("endpoint port is out of range: {port}")))?; + let address = EndpointAddress { host, port }; + + // Per gRFC A27: if set, the weight must be at least 1. Unset means the + // endpoint carries equal weight within its locality. + let weight = match lb_ep.load_balancing_weight_opt() { + Some(w) if w.value() == 0 => { + return Err(Error::Validation( + "endpoint has a zero load_balancing_weight".into(), + )); + } + Some(w) => w.value(), + None => 1, + }; + + Ok(Some(LbEndpoint { + address, + health_status, + load_balancing_weight: weight, + })) +} + +impl EndpointsResource { + /// Returns all healthy endpoints (`Unknown` and `Healthy` status), per + /// gRFC A27's definition of usable endpoints. + pub(crate) fn healthy_endpoints(&self) -> impl Iterator { + self.localities + .iter() + .flat_map(|l| &l.endpoints) + .filter(|e| { + matches!( + e.health_status, + HealthStatus::Unknown | HealthStatus::Healthy + ) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::generated::envoy::config::core::v3::{ + Address, Locality as EnvoyLocality, SocketAddress, + }; + use crate::generated::envoy::config::endpoint::v3::{ + Endpoint, LbEndpoint as EnvoyLbEndpoint, LocalityLbEndpoints as EnvoyLocalityLbEndpoints, + }; + use protobuf::Serialize; + use protobuf_well_known_types::UInt32Value; + + fn make_weight(weight: u32) -> UInt32Value { + let mut value = UInt32Value::new(); + value.set_value(weight); + value + } + + fn make_locality_endpoints(region: &str, priority: u32) -> EnvoyLocalityLbEndpoints { + let mut locality = EnvoyLocality::new(); + locality.set_region(region); + + let mut locality_lb_endpoints = EnvoyLocalityLbEndpoints::new(); + locality_lb_endpoints.set_locality(locality); + locality_lb_endpoints.set_load_balancing_weight(make_weight(1)); + locality_lb_endpoints.set_priority(priority); + locality_lb_endpoints + } + + fn make_lb_endpoint(host: &str, port: u32, health: EnvoyHealthStatus) -> EnvoyLbEndpoint { + let mut socket_address = SocketAddress::new(); + socket_address.set_address(host); + socket_address.set_port_value(port); + let mut address = Address::new(); + address.set_socket_address(socket_address); + let mut endpoint = Endpoint::new(); + endpoint.set_address(address); + + let mut lb_endpoint = EnvoyLbEndpoint::new(); + lb_endpoint.set_endpoint(endpoint); + lb_endpoint.set_health_status(health); + lb_endpoint + } + + fn make_cla(cluster_name: &str) -> ClusterLoadAssignment { + let mut locality_lb_endpoints = make_locality_endpoints("us-east-1", 0); + locality_lb_endpoints + .lb_endpoints_mut() + .push(make_lb_endpoint( + "10.0.0.1", + 8080, + EnvoyHealthStatus::Healthy, + )); + locality_lb_endpoints + .lb_endpoints_mut() + .push(make_lb_endpoint( + "10.0.0.2", + 8080, + EnvoyHealthStatus::Unknown, + )); + locality_lb_endpoints + .lb_endpoints_mut() + .push(make_lb_endpoint( + "10.0.0.3", + 8080, + EnvoyHealthStatus::Unhealthy, + )); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name(cluster_name); + cla.endpoints_mut().push(locality_lb_endpoints); + cla + } + + #[test] + fn validate_basic() { + let cla = make_cla("my-cluster"); + let validated = EndpointsResource::validate(cla).expect("should validate"); + assert_eq!(validated.cluster_name, "my-cluster"); + assert_eq!(validated.localities.len(), 1); + assert_eq!(validated.localities[0].endpoints.len(), 3); + assert_eq!(validated.localities[0].locality.region, "us-east-1"); + } + + #[test] + fn validate_endpoint_addresses() { + let cla = make_cla("my-cluster"); + let validated = EndpointsResource::validate(cla).unwrap(); + let addr = &validated.localities[0].endpoints[0].address; + assert_eq!(addr.host, "10.0.0.1"); + assert_eq!(addr.port, 8080); + } + + #[test] + fn healthy_endpoints_excludes_unhealthy() { + let cla = make_cla("my-cluster"); + let validated = EndpointsResource::validate(cla).unwrap(); + // Healthy + Unknown = 2 (Unhealthy excluded). + assert_eq!(validated.healthy_endpoints().count(), 2); + } + + #[test] + fn healthy_endpoints_excludes_degraded_and_unrecognized() { + let mut locality_lb_endpoints = make_locality_endpoints("us-east-1", 0); + locality_lb_endpoints + .lb_endpoints_mut() + .push(make_lb_endpoint( + "10.0.0.1", + 8080, + EnvoyHealthStatus::Degraded, + )); + locality_lb_endpoints + .lb_endpoints_mut() + .push(make_lb_endpoint( + "10.0.0.2", + 8080, + EnvoyHealthStatus::from(99), + )); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name("my-cluster"); + cla.endpoints_mut().push(locality_lb_endpoints); + + let validated = EndpointsResource::validate(cla).unwrap(); + assert_eq!(validated.localities[0].endpoints.len(), 2); + assert_eq!(validated.healthy_endpoints().count(), 0); + assert_eq!( + validated.localities[0].endpoints[0].health_status, + HealthStatus::Degraded + ); + } + + #[test] + fn unrecognized_health_status_survives_the_wire_and_is_not_usable() { + let mut locality_lb_endpoints = make_locality_endpoints("us-east-1", 0); + locality_lb_endpoints + .lb_endpoints_mut() + .push(make_lb_endpoint( + "10.0.0.1", + 8080, + EnvoyHealthStatus::from(99), + )); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name("my-cluster"); + cla.endpoints_mut().push(locality_lb_endpoints); + + let bytes = cla.serialize().expect("serialize"); + let decoded = + EndpointsResource::deserialize(bytes::Bytes::from(bytes)).expect("deserialize"); + assert_eq!( + i32::from( + decoded + .endpoints() + .get(0) + .expect("locality") + .lb_endpoints() + .get(0) + .expect("endpoint") + .health_status() + ), + 99, + "protobuf runtime must preserve unrecognized enum values" + ); + + let validated = EndpointsResource::validate(decoded).unwrap(); + assert_eq!( + validated.localities[0].endpoints[0].health_status, + HealthStatus::Other(99), + "an unrecognized status must keep its raw value for debugging" + ); + assert_eq!(validated.healthy_endpoints().count(), 0); + } + + #[test] + fn validate_empty_cluster_name() { + let cla = ClusterLoadAssignment::new(); + let err = EndpointsResource::validate(cla).unwrap_err(); + assert!(err.to_string().contains("cluster_name")); + } + + #[test] + fn validate_skips_named_endpoint() { + let mut lb_endpoint = EnvoyLbEndpoint::new(); + lb_endpoint.set_endpoint_name("named-endpoint"); + + let mut locality_lb_endpoints = make_locality_endpoints("us-east-1", 0); + locality_lb_endpoints.lb_endpoints_mut().push(lb_endpoint); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name("my-cluster"); + cla.endpoints_mut().push(locality_lb_endpoints); + + let validated = EndpointsResource::validate(cla).expect("should validate"); + assert!(validated.localities[0].endpoints.is_empty()); + } + + #[test] + fn validate_rejects_missing_port() { + let mut socket_address = SocketAddress::new(); + socket_address.set_address("10.0.0.1"); + let mut address = Address::new(); + address.set_socket_address(socket_address); + let mut endpoint = Endpoint::new(); + endpoint.set_address(address); + let mut lb_endpoint = EnvoyLbEndpoint::new(); + lb_endpoint.set_endpoint(endpoint); + + let mut locality_lb_endpoints = make_locality_endpoints("us-east-1", 0); + locality_lb_endpoints.lb_endpoints_mut().push(lb_endpoint); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name("my-cluster"); + cla.endpoints_mut().push(locality_lb_endpoints); + + let err = EndpointsResource::validate(cla).unwrap_err(); + assert!(err.to_string().contains("missing numeric port")); + } + + #[test] + fn validate_rejects_out_of_range_port() { + let mut locality_lb_endpoints = make_locality_endpoints("us-east-1", 0); + locality_lb_endpoints + .lb_endpoints_mut() + .push(make_lb_endpoint( + "10.0.0.1", + 65_536, + EnvoyHealthStatus::Healthy, + )); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name("my-cluster"); + cla.endpoints_mut().push(locality_lb_endpoints); + + let err = EndpointsResource::validate(cla).unwrap_err(); + assert!(err.to_string().contains("out of range")); + } + + #[test] + fn validate_rejects_locality_without_id() { + let mut locality_lb_endpoints = EnvoyLocalityLbEndpoints::new(); + locality_lb_endpoints.set_load_balancing_weight(make_weight(1)); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name("my-cluster"); + cla.endpoints_mut().push(locality_lb_endpoints); + + let err = EndpointsResource::validate(cla).unwrap_err(); + assert!(err.to_string().contains("without an ID")); + } + + #[test] + fn validate_skips_locality_without_weight() { + let mut locality = EnvoyLocality::new(); + locality.set_region("us-east-1"); + let mut locality_lb_endpoints = EnvoyLocalityLbEndpoints::new(); + locality_lb_endpoints.set_locality(locality); + locality_lb_endpoints + .lb_endpoints_mut() + .push(make_lb_endpoint( + "10.0.0.1", + 8080, + EnvoyHealthStatus::Healthy, + )); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name("my-cluster"); + cla.endpoints_mut().push(locality_lb_endpoints); + + let validated = EndpointsResource::validate(cla).expect("should validate"); + assert!(validated.localities.is_empty()); + } + + #[test] + fn validate_rejects_duplicate_locality_at_same_priority() { + let mut cla = make_cla("my-cluster"); + let mut duplicate = make_locality_endpoints("us-east-1", 0); + duplicate.lb_endpoints_mut().push(make_lb_endpoint( + "10.0.0.4", + 8080, + EnvoyHealthStatus::Healthy, + )); + cla.endpoints_mut().push(duplicate); + + let err = EndpointsResource::validate(cla).unwrap_err(); + assert!(err.to_string().contains("duplicate locality")); + } + + #[test] + fn validate_rejects_duplicate_endpoint_address() { + let mut cla = make_cla("my-cluster"); + let mut other = make_locality_endpoints("us-west-1", 0); + other.lb_endpoints_mut().push(make_lb_endpoint( + "10.0.0.1", + 8080, + EnvoyHealthStatus::Healthy, + )); + cla.endpoints_mut().push(other); + + let err = EndpointsResource::validate(cla).unwrap_err(); + assert!(err.to_string().contains("duplicate endpoint address")); + } + + #[test] + fn validate_rejects_priority_gap() { + let mut cla = make_cla("my-cluster"); + let mut gapped = make_locality_endpoints("us-west-1", 2); + gapped.lb_endpoints_mut().push(make_lb_endpoint( + "10.0.0.4", + 8080, + EnvoyHealthStatus::Healthy, + )); + cla.endpoints_mut().push(gapped); + + let err = EndpointsResource::validate(cla).unwrap_err(); + assert!(err.to_string().contains("priority 1 missing")); + } + + #[test] + fn validate_rejects_zero_endpoint_weight() { + let mut lb_endpoint = make_lb_endpoint("10.0.0.1", 8080, EnvoyHealthStatus::Healthy); + lb_endpoint.set_load_balancing_weight(make_weight(0)); + + let mut locality_lb_endpoints = make_locality_endpoints("us-east-1", 0); + locality_lb_endpoints.lb_endpoints_mut().push(lb_endpoint); + + let mut cla = ClusterLoadAssignment::new(); + cla.set_cluster_name("my-cluster"); + cla.endpoints_mut().push(locality_lb_endpoints); + + let err = EndpointsResource::validate(cla).unwrap_err(); + assert!(err.to_string().contains("zero load_balancing_weight")); + } + + #[test] + fn deserialize_roundtrip() { + let cla = make_cla("test"); + let bytes = cla.serialize().expect("serialize"); + let deserialized = EndpointsResource::deserialize(bytes::Bytes::from(bytes)).unwrap(); + assert_eq!(EndpointsResource::name(&deserialized), "test"); + } +} diff --git a/grpc-xds/src/resource/listener.rs b/grpc-xds/src/resource/listener.rs new file mode 100644 index 000000000..e8fd79cbe --- /dev/null +++ b/grpc-xds/src/resource/listener.rs @@ -0,0 +1,346 @@ +/* + * + * Copyright 2025 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +//! Validated Listener resource (LDS). + +use std::sync::Arc; + +use protobuf::Parse; +use protobuf_well_known_types::Any; +use xds_client::resource::TypeUrl; +use xds_client::{Error, Resource}; + +use super::route::RouteConfigResource; +use crate::generated::envoy::config::listener::v3::Listener; +use crate::generated::envoy::extensions::filters::network::http_connection_manager::v3::{ + HttpConnectionManager, http_connection_manager::RouteSpecifierOneof, +}; + +/// The only `api_listener` extension gRPC supports. +const HTTP_CONNECTION_MANAGER_TYPE_URL: &str = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager"; + +/// How the listener obtains its route configuration. +#[derive(Debug, Clone)] +pub(crate) enum RouteSource { + /// Route configuration fetched dynamically via RDS, keyed by this route + /// config name. + Rds(String), + /// Route configuration embedded inline in the listener. + Inline(Arc), +} + +/// Validated Listener resource. +/// +/// Extracts the route source from the +/// `ApiListener` -> `HttpConnectionManager` -> `route_specifier` chain per +/// gRFC A27. `scoped_routes` is not supported, matching other gRPC xDS +/// client implementation. +#[derive(Debug, Clone)] +pub(crate) struct ListenerResource { + pub(crate) name: String, + pub(crate) route_source: RouteSource, +} + +impl Resource for ListenerResource { + type Message = Listener; + + const TYPE_URL: TypeUrl = TypeUrl::new("type.googleapis.com/envoy.config.listener.v3.Listener"); + + const ALL_RESOURCES_REQUIRED_IN_SOTW: bool = true; + + fn deserialize(bytes: bytes::Bytes) -> xds_client::Result { + Listener::parse(&bytes) + .map_err(|e| Error::Validation(format!("failed to decode Listener: {e}"))) + } + + fn name(message: &Self::Message) -> &str { + message.name().to_str().unwrap_or_default() + } + + fn validate(message: Self::Message) -> xds_client::Result { + let name = message.name().to_str().unwrap_or_default().to_string(); + if name.is_empty() { + return Err(Error::Validation("listener name is empty".into())); + } + + if !message.has_api_listener() { + return Err(Error::Validation( + "listener missing api_listener field".into(), + )); + } + let api_listener = message.api_listener(); + + if !api_listener.has_api_listener() { + return Err(Error::Validation( + "api_listener missing inner api_listener Any field".into(), + )); + } + let any: Any = api_listener.api_listener().to_owned(); + + let type_url = any.type_url().to_str().unwrap_or_default(); + if type_url != HTTP_CONNECTION_MANAGER_TYPE_URL { + return Err(Error::Validation(format!( + "unexpected api_listener type_url: '{type_url}'" + ))); + } + + let hcm = HttpConnectionManager::parse(any.value()).map_err(|e| { + Error::Validation(format!("failed to decode HttpConnectionManager: {e}")) + })?; + + let route_source = match hcm.route_specifier() { + RouteSpecifierOneof::Rds(rds) => { + if !rds.has_config_source() { + return Err(Error::Validation("RDS config_source is not set".into())); + } + let config_source = rds.config_source(); + if !config_source.has_ads() && !config_source.has_self() { + return Err(Error::Validation( + "RDS config_source is not ADS or Self".into(), + )); + } + + let route_config_name = rds.route_config_name().to_str().unwrap_or_default(); + if route_config_name.is_empty() { + return Err(Error::Validation("RDS route_config_name is empty".into())); + } + RouteSource::Rds(route_config_name.to_string()) + } + RouteSpecifierOneof::RouteConfig(route_config) => { + let validated = Arc::new(RouteConfigResource::validate(route_config.to_owned())?); + RouteSource::Inline(validated) + } + RouteSpecifierOneof::ScopedRoutes(_) => { + return Err(Error::Validation( + "scoped_routes not supported for gRPC".into(), + )); + } + RouteSpecifierOneof::not_set(_) => { + return Err(Error::Validation( + "HttpConnectionManager missing route_specifier".into(), + )); + } + }; + + Ok(ListenerResource { name, route_source }) + } +} + +impl ListenerResource { + /// Returns the RDS route config name for cascading subscriptions, or + /// `None` when the route configuration was embedded inline. + pub(crate) fn route_config_name(&self) -> Option<&str> { + match &self.route_source { + RouteSource::Rds(name) => Some(name), + RouteSource::Inline(_) => None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::generated::envoy::config::core::v3::{ + AggregatedConfigSource, ConfigSource, SelfConfigSource, + }; + use crate::generated::envoy::config::listener::v3::ApiListener; + use crate::generated::envoy::config::route::v3::{ + Route, RouteAction, RouteConfiguration, RouteMatch, VirtualHost, + }; + use crate::generated::envoy::extensions::filters::network::http_connection_manager::v3::Rds; + use protobuf::Serialize; + + const HCM_TYPE_URL: &str = "type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager"; + + fn wrap_hcm(hcm: &HttpConnectionManager) -> Any { + let mut any = Any::new(); + any.set_type_url(HCM_TYPE_URL); + any.set_value(hcm.serialize().expect("serialize hcm")); + any + } + + fn make_rds_listener_with_config_source( + name: &str, + route_config_name: &str, + config_source: Option, + ) -> Listener { + let mut rds = Rds::new(); + rds.set_route_config_name(route_config_name); + if let Some(config_source) = config_source { + rds.set_config_source(config_source); + } + let mut hcm = HttpConnectionManager::new(); + hcm.set_rds(rds); + + let mut api_listener = ApiListener::new(); + api_listener.set_api_listener(wrap_hcm(&hcm)); + + let mut listener = Listener::new(); + listener.set_name(name); + listener.set_api_listener(api_listener); + listener + } + + fn make_rds_listener(name: &str, route_config_name: &str) -> Listener { + let mut config_source = ConfigSource::new(); + config_source.set_ads(AggregatedConfigSource::new()); + make_rds_listener_with_config_source(name, route_config_name, Some(config_source)) + } + + #[test] + fn validate_rds_listener() { + let listener = make_rds_listener("test-listener", "route-config-1"); + let validated = ListenerResource::validate(listener).expect("should validate"); + assert_eq!(validated.name, "test-listener"); + assert!( + matches!(&validated.route_source, RouteSource::Rds(name) if name == "route-config-1") + ); + assert_eq!(validated.route_config_name(), Some("route-config-1")); + } + + #[test] + fn validate_missing_api_listener() { + let mut listener = Listener::new(); + listener.set_name("test-listener"); + let err = ListenerResource::validate(listener).unwrap_err(); + assert!(err.to_string().contains("api_listener")); + } + + #[test] + fn validate_rejects_empty_listener_name() { + let listener = make_rds_listener("", "route-config-1"); + let err = ListenerResource::validate(listener).unwrap_err(); + assert!(err.to_string().contains("listener name is empty")); + } + + #[test] + fn validate_rejects_unexpected_api_listener_type_url() { + let mut listener = make_rds_listener("test-listener", "route-config-1"); + let mut any = listener.api_listener().api_listener().to_owned(); + any.set_type_url("type.googleapis.com/envoy.config.listener.v3.Listener"); + let mut api_listener = ApiListener::new(); + api_listener.set_api_listener(any); + listener.set_api_listener(api_listener); + + let err = ListenerResource::validate(listener).unwrap_err(); + assert!(err.to_string().contains("unexpected api_listener type_url")); + } + + #[test] + fn validate_empty_rds_name() { + let listener = make_rds_listener("test-listener", ""); + let err = ListenerResource::validate(listener).unwrap_err(); + assert!(err.to_string().contains("route_config_name is empty")); + } + + #[test] + fn validate_rds_listener_accepts_self_config_source() { + let mut config_source = ConfigSource::new(); + config_source.set_self(SelfConfigSource::new()); + let listener = make_rds_listener_with_config_source( + "test-listener", + "route-config-1", + Some(config_source), + ); + assert!(ListenerResource::validate(listener).is_ok()); + } + + #[test] + fn validate_rds_listener_rejects_missing_config_source() { + let listener = + make_rds_listener_with_config_source("test-listener", "route-config-1", None); + let err = ListenerResource::validate(listener).unwrap_err(); + assert!(err.to_string().contains("config_source is not set")); + } + + #[test] + fn validate_rds_listener_rejects_non_ads_non_self_config_source() { + let mut config_source = ConfigSource::new(); + config_source.set_path("/some/path"); + let listener = make_rds_listener_with_config_source( + "test-listener", + "route-config-1", + Some(config_source), + ); + let err = ListenerResource::validate(listener).unwrap_err(); + assert!(err.to_string().contains("not ADS or Self")); + } + + #[test] + fn deserialize_valid() { + let listener = make_rds_listener("test", "rc1"); + let bytes = listener.serialize().expect("serialize"); + let deserialized = + ListenerResource::deserialize(bytes::Bytes::from(bytes)).expect("should deserialize"); + assert_eq!(ListenerResource::name(&deserialized), "test"); + } + + #[test] + fn deserialize_invalid_bytes() { + // A lone 0x80 is not a valid protobuf varint tag: this should fail + // decoding rather than silently produce a default message. + let result = ListenerResource::deserialize(bytes::Bytes::from_static(b"\x80")); + assert!(result.is_err()); + } + + #[test] + fn validate_inline_route_config() { + let mut route_match = RouteMatch::new(); + route_match.set_prefix("/"); + let mut route_action = RouteAction::new(); + route_action.set_cluster("cluster-1"); + let mut route = Route::new(); + route.set_match(route_match); + route.set_route(route_action); + + let mut vh = VirtualHost::new(); + vh.set_name("vh1"); + vh.domains_mut().push("*"); + vh.routes_mut().push(route); + + let mut route_config = RouteConfiguration::new(); + route_config.set_name("inline-rc"); + route_config.virtual_hosts_mut().push(vh); + + let mut hcm = HttpConnectionManager::new(); + hcm.set_route_config(route_config); + + let mut api_listener = ApiListener::new(); + api_listener.set_api_listener(wrap_hcm(&hcm)); + + let mut listener = Listener::new(); + listener.set_name("inline-listener"); + listener.set_api_listener(api_listener); + + let validated = ListenerResource::validate(listener).expect("should validate"); + assert_eq!(validated.name, "inline-listener"); + assert!(matches!(&validated.route_source, RouteSource::Inline(_))); + assert!(validated.route_config_name().is_none()); + + let RouteSource::Inline(rc) = &validated.route_source else { + unreachable!() + }; + assert_eq!(rc.virtual_hosts.len(), 1); + } +} diff --git a/grpc-xds/src/resource/mod.rs b/grpc-xds/src/resource/mod.rs new file mode 100644 index 000000000..fa03e5f8d --- /dev/null +++ b/grpc-xds/src/resource/mod.rs @@ -0,0 +1,48 @@ +/* + * + * Copyright 2025 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +//! The xDS data model layer: validated, owned representations of the raw +//! discovery-protocol resources (LDS/RDS/CDS/EDS), each implementing +//! [`xds_client::Resource`] so it can be deserialized, named, and validated +//! (gRFC A27) independently of any live xDS traffic. +//! The dependency manager assembles these into an [`crate::xds_config::XdsConfig`]. + +// TODO: remove once the xDS dependency manager subscribes to these resource +// types and assembles them into an XdsConfig. +#![allow(dead_code, unused_imports)] + +mod cluster; +mod endpoint; +mod listener; +mod route; + +pub(crate) use cluster::{ClusterDiscovery, ClusterResource}; +pub(crate) use endpoint::{ + EndpointAddress, EndpointsResource, HealthStatus, LbEndpoint, Locality, LocalityLbEndpoints, +}; +pub(crate) use listener::{ListenerResource, RouteSource}; +pub(crate) use route::{ + HeaderMatchSpecifier, HeaderMatcher, PathSpecifier, Route, RouteAction, RouteConfigResource, + RouteMatch, StringMatcher, VirtualHost, WeightedCluster, +}; diff --git a/grpc-xds/src/resource/route.rs b/grpc-xds/src/resource/route.rs new file mode 100644 index 000000000..aecdbc056 --- /dev/null +++ b/grpc-xds/src/resource/route.rs @@ -0,0 +1,1037 @@ +/* + * + * Copyright 2025 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +//! Validated RouteConfiguration resource (RDS). +//! +//! Models the validated data shapes only (gRFC A28). + +// TODO: implement request routing components (virtual-host domain matching, then +// path/header/fraction matching, stripping `-bin` headers from request metadata before +// evaluating header matchers) in the future resolver/interceptor layer that consumes `XdsConfig`. + +use std::collections::HashSet; + +use protobuf::Parse; +use regex::Regex; +use xds_client::resource::TypeUrl; +use xds_client::{Error, Resource}; + +use crate::generated::envoy::config::route::v3::header_matcher::HeaderMatchSpecifierOneof; +use crate::generated::envoy::config::route::v3::route::ActionOneof; +use crate::generated::envoy::config::route::v3::route_action::ClusterSpecifierOneof; +use crate::generated::envoy::config::route::v3::route_match::PathSpecifierOneof; +use crate::generated::envoy::config::route::v3::{ + HeaderMatcherView, RouteActionView, RouteConfiguration, RouteMatchView, RouteView, + VirtualHostView, +}; +use crate::generated::envoy::r#type::matcher::v3::StringMatcherView; +use crate::generated::envoy::r#type::matcher::v3::string_matcher::MatchPatternOneof; +use crate::generated::envoy::r#type::v3::fractional_percent::DenominatorType; + +/// Validated RouteConfiguration. +#[derive(Debug, Clone)] +pub(crate) struct RouteConfigResource { + pub(crate) name: String, + pub(crate) virtual_hosts: Vec, +} + +/// Validated virtual host with domain matching and routes. +#[derive(Debug, Clone)] +pub(crate) struct VirtualHost { + pub(crate) name: String, + pub(crate) domains: Vec, + pub(crate) routes: Vec, +} + +/// A validated route with match criteria and action. +#[derive(Debug, Clone)] +pub(crate) struct Route { + pub(crate) route_match: RouteMatch, + pub(crate) action: RouteAction, +} + +/// Validated route match criteria. +#[derive(Debug, Clone)] +pub(crate) struct RouteMatch { + pub(crate) path_specifier: PathSpecifier, + pub(crate) headers: Vec, + pub(crate) case_sensitive: bool, + /// Fraction of requests this route should match, as numerator out of + /// 1,000,000. `None` means always match (100%). + pub(crate) match_fraction: Option, +} + +/// Path matching specifier. +#[derive(Debug, Clone)] +pub(crate) enum PathSpecifier { + Prefix(String), + Path(String), + SafeRegex(Regex), +} + +/// Header matching criteria. +#[derive(Debug, Clone)] +pub(crate) struct HeaderMatcher { + pub(crate) name: String, + pub(crate) match_specifier: HeaderMatchSpecifier, + pub(crate) invert_match: bool, +} + +/// Header match specifier variants. +/// +/// The `String` variant carries a generic [`StringMatcher`] (exact / prefix / +/// suffix / contains / safe_regex, with optional ASCII case-insensitive +/// matching per gRFC A63). `Present`, `Absent`, and `Range` are header-specific +/// extensions beyond the generic StringMatcher. +#[derive(Debug, Clone)] +pub(crate) enum HeaderMatchSpecifier { + String(StringMatcher), + /// Match if header is present (any value). + Present, + /// Match if header is absent. + Absent, + /// Match if the header value, parsed as an integer, falls within `[start, end)`. + Range { + start: i64, + end: i64, + }, +} + +/// Route action deciding where to send traffic. +#[derive(Debug, Clone)] +pub(crate) enum RouteAction { + Cluster(String), + WeightedClusters(Vec), +} + +/// A cluster with an associated weight for traffic splitting. +#[derive(Debug, Clone)] +pub(crate) struct WeightedCluster { + pub(crate) name: String, + pub(crate) weight: u32, +} + +/// Validated `envoy.type.matcher.v3.StringMatcher`. +#[derive(Debug, Clone)] +pub(crate) enum StringMatcher { + Exact { value: String, ignore_case: bool }, + Prefix { value: String, ignore_case: bool }, + Suffix { value: String, ignore_case: bool }, + Contains { value: String, ignore_case: bool }, + SafeRegex(Regex), +} + +impl StringMatcher { + /// Parses and validates an `envoy.type.matcher.v3.StringMatcher`. + /// + /// Returns an error if the `match_pattern` oneof is unset or carries an + /// unsupported variant, a prefix/suffix/contains value is empty, or a + /// `safe_regex` fails to compile. + fn from_proto(proto: StringMatcherView<'_>) -> xds_client::Result { + let ignore_case = proto.ignore_case(); + match proto.match_pattern() { + MatchPatternOneof::Exact(value) => Ok(Self::Exact { + value: value.to_str().unwrap_or_default().to_string(), + ignore_case, + }), + MatchPatternOneof::Prefix(value) => Ok(Self::Prefix { + value: non_empty_match_value(value.to_str().unwrap_or_default(), "prefix")?, + ignore_case, + }), + MatchPatternOneof::Suffix(value) => Ok(Self::Suffix { + value: non_empty_match_value(value.to_str().unwrap_or_default(), "suffix")?, + ignore_case, + }), + MatchPatternOneof::Contains(value) => Ok(Self::Contains { + value: non_empty_match_value(value.to_str().unwrap_or_default(), "contains")?, + ignore_case, + }), + MatchPatternOneof::SafeRegex(r) => { + let pattern = r.regex(); + let pattern = pattern.to_str().unwrap_or_default(); + Ok(Self::SafeRegex(compile_regex(pattern, "string matcher")?)) + } + MatchPatternOneof::not_set(_) => Err(Error::Validation( + "StringMatcher has no match_pattern set".into(), + )), + _ => Err(Error::Validation( + "unsupported StringMatcher pattern".into(), + )), + } + } +} + +fn non_empty_match_value(value: &str, kind: &str) -> xds_client::Result { + if value.is_empty() { + return Err(Error::Validation(format!( + "empty {kind} match is not allowed" + ))); + } + Ok(value.to_string()) +} + +/// Compiles a `RegexMatcher` pattern, rejecting the empty pattern. +fn compile_regex(pattern: &str, kind: &str) -> xds_client::Result { + if pattern.is_empty() { + return Err(Error::Validation(format!( + "empty {kind} regex is not allowed" + ))); + } + Regex::new(pattern) + .map_err(|e| Error::Validation(format!("invalid {kind} regex '{pattern}': {e}"))) +} + +impl Resource for RouteConfigResource { + type Message = RouteConfiguration; + + const TYPE_URL: TypeUrl = + TypeUrl::new("type.googleapis.com/envoy.config.route.v3.RouteConfiguration"); + + const ALL_RESOURCES_REQUIRED_IN_SOTW: bool = false; + + fn deserialize(bytes: bytes::Bytes) -> xds_client::Result { + RouteConfiguration::parse(&bytes) + .map_err(|e| Error::Validation(format!("failed to decode RouteConfiguration: {e}"))) + } + + fn name(message: &Self::Message) -> &str { + message.name().to_str().unwrap_or_default() + } + + fn validate(message: Self::Message) -> xds_client::Result { + let name = message.name().to_str().unwrap_or_default().to_string(); + + let virtual_hosts_view = message.virtual_hosts(); + let mut virtual_hosts = Vec::new(); + for vh in virtual_hosts_view.iter() { + virtual_hosts.push(validate_virtual_host(vh)?); + } + + Ok(RouteConfigResource { + name, + virtual_hosts, + }) + } +} + +fn validate_virtual_host(vh: VirtualHostView<'_>) -> xds_client::Result { + let name = vh.name().to_str().unwrap_or_default().to_string(); + + let domains_view = vh.domains(); + if domains_view.is_empty() { + return Err(Error::Validation(format!( + "virtual host '{name}' has no domains" + ))); + } + let domains: Vec = domains_view + .iter() + .map(|d| d.to_str().unwrap_or_default().to_string()) + .collect(); + + let mut routes = Vec::new(); + for route in vh.routes().iter() { + if let Some(validated) = validate_route(route)? { + routes.push(validated); + } + } + + Ok(VirtualHost { + name, + domains, + routes, + }) +} + +/// Returns `Ok(None)` for routes that should be silently skipped (query +/// param matchers, unsupported cluster specifiers like `cluster_header`), +/// per gRFC A28. +fn validate_route(route: RouteView<'_>) -> xds_client::Result> { + if !route.has_match() { + return Err(Error::Validation("route missing match field".into())); + } + let route_match = route.r#match(); + + // Per A28: ignore routes with query parameter matchers. + if !route_match.query_parameters().is_empty() { + return Ok(None); + } + + let match_criteria = validate_route_match(route_match)?; + + let action = match route.action() { + ActionOneof::Route(route_action) => match validate_route_action(route_action)? { + Some(action) => action, + None => return Ok(None), + }, + // Per A28: action field must be "route", otherwise NACK. + _ => { + return Err(Error::Validation( + "only route action is supported for client routing".into(), + )); + } + }; + + Ok(Some(Route { + route_match: match_criteria, + action, + })) +} + +fn validate_route_match(rm: RouteMatchView<'_>) -> xds_client::Result { + let path_specifier = match rm.path_specifier() { + PathSpecifierOneof::Prefix(p) => { + PathSpecifier::Prefix(p.to_str().unwrap_or_default().to_string()) + } + PathSpecifierOneof::Path(p) => { + PathSpecifier::Path(p.to_str().unwrap_or_default().to_string()) + } + PathSpecifierOneof::SafeRegex(r) => { + let pattern = r.regex(); + let pattern = pattern.to_str().unwrap_or_default(); + PathSpecifier::SafeRegex(compile_regex(pattern, "path")?) + } + // Per A28: not having path_specifier will cause a NACK. + PathSpecifierOneof::not_set(_) => { + return Err(Error::Validation( + "route match missing path_specifier".into(), + )); + } + _ => { + return Err(Error::Validation( + "unsupported path specifier variant".into(), + )); + } + }; + + let case_sensitive = if rm.has_case_sensitive() { + rm.case_sensitive().value() + } else { + true + }; + + // Per A28, a matcher naming a `-bin` header must behave as if that header + // were absent, so the matcher is kept here and neutralized at match time. + // Dropping it instead would widen the route to traffic it must not match. + let headers = rm + .headers() + .iter() + .map(validate_header_matcher) + .collect::>>()?; + + // Per A28: use runtime_fraction.default_value, normalize to numerator out + // of 1,000,000. runtime_key is ignored (gRPC has no runtime config). + let match_fraction = match rm.runtime_fraction_opt() { + None => None, + Some(rf) => { + if !rf.has_default_value() { + return Err(Error::Validation( + "runtime_fraction is missing its required default_value".into(), + )); + } + let frac = rf.default_value(); + let scale = match frac.denominator() { + DenominatorType::Hundred => 10_000, + DenominatorType::TenThousand => 100, + DenominatorType::Million => 1, + _ => 1, + }; + Some(frac.numerator().saturating_mul(scale).min(1_000_000)) + } + }; + + Ok(RouteMatch { + path_specifier, + headers, + case_sensitive, + match_fraction, + }) +} + +fn validate_header_matcher(hm: HeaderMatcherView<'_>) -> xds_client::Result { + let name = hm.name().to_str().unwrap_or_default().to_string(); + if name.is_empty() { + return Err(Error::Validation("header matcher name is empty".into())); + } + + // Legacy matchers are deprecated in favor of StringMatch but remain + // widely used and are still required by gRFC A63. + #[allow(deprecated, unreachable_patterns)] + let match_specifier = match hm.header_match_specifier() { + HeaderMatchSpecifierOneof::ExactMatch(v) => { + HeaderMatchSpecifier::String(StringMatcher::Exact { + value: v.to_str().unwrap_or_default().to_string(), + ignore_case: false, + }) + } + HeaderMatchSpecifierOneof::SafeRegexMatch(r) => { + let pattern = r.regex(); + let pattern = pattern.to_str().unwrap_or_default(); + HeaderMatchSpecifier::String(StringMatcher::SafeRegex(compile_regex( + pattern, "header", + )?)) + } + HeaderMatchSpecifierOneof::RangeMatch(r) => HeaderMatchSpecifier::Range { + start: r.start(), + end: r.end(), + }, + HeaderMatchSpecifierOneof::PresentMatch(present) => { + if present { + HeaderMatchSpecifier::Present + } else { + HeaderMatchSpecifier::Absent + } + } + HeaderMatchSpecifierOneof::PrefixMatch(v) => { + HeaderMatchSpecifier::String(StringMatcher::Prefix { + value: non_empty_match_value(v.to_str().unwrap_or_default(), "prefix")?, + ignore_case: false, + }) + } + HeaderMatchSpecifierOneof::SuffixMatch(v) => { + HeaderMatchSpecifier::String(StringMatcher::Suffix { + value: non_empty_match_value(v.to_str().unwrap_or_default(), "suffix")?, + ignore_case: false, + }) + } + HeaderMatchSpecifierOneof::ContainsMatch(v) => { + HeaderMatchSpecifier::String(StringMatcher::Contains { + value: non_empty_match_value(v.to_str().unwrap_or_default(), "contains")?, + ignore_case: false, + }) + } + HeaderMatchSpecifierOneof::StringMatch(sm) => { + HeaderMatchSpecifier::String(StringMatcher::from_proto(sm)?) + } + HeaderMatchSpecifierOneof::not_set(_) => HeaderMatchSpecifier::Present, + _ => { + return Err(Error::Validation( + "unsupported header match specifier".into(), + )); + } + }; + + Ok(HeaderMatcher { + name, + match_specifier, + invert_match: hm.invert_match(), + }) +} + +/// Returns `Ok(None)` for routes whose cluster specifier is unsupported +/// (e.g. `cluster_header`) or unset, both of which A28 requires be skipped. +fn validate_route_action(ra: RouteActionView<'_>) -> xds_client::Result> { + match ra.cluster_specifier() { + ClusterSpecifierOneof::Cluster(name) => { + let name = name.to_str().unwrap_or_default(); + if name.is_empty() { + return Err(Error::Validation("cluster name is empty".into())); + } + Ok(Some(RouteAction::Cluster(name.to_string()))) + } + ClusterSpecifierOneof::WeightedClusters(wc) => { + let clusters_view = wc.clusters(); + if clusters_view.is_empty() { + return Err(Error::Validation("weighted_clusters is empty".into())); + } + let mut clusters = Vec::new(); + let mut total_weight: u64 = 0; + for c in clusters_view.iter() { + // Per A28: zero-weight entries never receive traffic, so drop them. + let weight = c.weight_opt().map(|w| w.value()).unwrap_or(0); + if weight == 0 { + continue; + } + let name = c.name().to_str().unwrap_or_default(); + if name.is_empty() { + return Err(Error::Validation("weighted cluster name is empty".into())); + } + total_weight += u64::from(weight); + if total_weight > u64::from(u32::MAX) { + return Err(Error::Validation(format!( + "sum of weighted cluster weights exceeds {}", + u32::MAX + ))); + } + clusters.push(WeightedCluster { + name: name.to_string(), + weight, + }); + } + // Per A28: the weights must add up to a non-zero total, otherwise + // there is nothing for a picker to distribute traffic across. + if clusters.is_empty() { + return Err(Error::Validation( + "weighted_clusters has no cluster with a non-zero weight".into(), + )); + } + Ok(Some(RouteAction::WeightedClusters(clusters))) + } + // Per A28: ignore the route when the cluster specifier is unsupported + // (e.g. `cluster_header`) or unset. A specifier field added to the oneof + // by a newer control plane decodes as an unknown field, which protobuf + // reports as `not_set`, so that case must be ignored rather than NACKed. + _ => Ok(None), + } +} + +impl VirtualHost { + /// Returns cluster names referenced by this virtual host, for cascading + /// CDS subscriptions after the dependency manager selects the host that + /// matches the channel authority. + #[allow(dead_code)] // TODO: remove once dependency manager calls this. + pub(crate) fn cluster_names(&self) -> HashSet { + let mut clusters = HashSet::new(); + for route in &self.routes { + match &route.action { + RouteAction::Cluster(name) => { + clusters.insert(name.clone()); + } + RouteAction::WeightedClusters(wcs) => { + for wc in wcs { + clusters.insert(wc.name.clone()); + } + } + } + } + clusters + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::generated::envoy::config::core::v3::RuntimeFractionalPercent; + use crate::generated::envoy::config::route::v3::weighted_cluster::ClusterWeight; + use crate::generated::envoy::config::route::v3::{ + HeaderMatcher as EnvoyHeaderMatcher, QueryParameterMatcher, RedirectAction, + Route as EnvoyRoute, RouteAction as EnvoyRouteAction, RouteMatch as EnvoyRouteMatch, + VirtualHost as EnvoyVirtualHost, WeightedCluster as EnvoyWeightedCluster, + }; + use crate::generated::envoy::r#type::matcher::v3::RegexMatcher; + use crate::generated::envoy::r#type::matcher::v3::StringMatcher as EnvoyStringMatcher; + use crate::generated::envoy::r#type::v3::FractionalPercent; + use protobuf::Serialize; + use protobuf_well_known_types::UInt32Value; + + fn make_route(prefix: &str, cluster: &str) -> EnvoyRoute { + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix(prefix); + + let mut route_action = EnvoyRouteAction::new(); + route_action.set_cluster(cluster); + + let mut route = EnvoyRoute::new(); + route.set_match(route_match); + route.set_route(route_action); + route + } + + fn make_route_config(name: &str) -> RouteConfiguration { + wrap_route(name, make_route("/", "cluster-1")) + } + + fn wrap_route(name: &str, route: EnvoyRoute) -> RouteConfiguration { + let mut vh = EnvoyVirtualHost::new(); + vh.set_name("vh1"); + vh.domains_mut().push("*"); + vh.routes_mut().push(route); + + let mut rc = RouteConfiguration::new(); + rc.set_name(name); + rc.virtual_hosts_mut().push(vh); + rc + } + + fn route_config_with_header(header: EnvoyHeaderMatcher) -> RouteConfiguration { + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + route_match.headers_mut().push(header); + + let mut route_action = EnvoyRouteAction::new(); + route_action.set_cluster("cluster-1"); + + let mut route = EnvoyRoute::new(); + route.set_match(route_match); + route.set_route(route_action); + + wrap_route("rc-1", route) + } + + fn validate_header(header: EnvoyHeaderMatcher) -> HeaderMatchSpecifier { + let validated = RouteConfigResource::validate(route_config_with_header(header)) + .expect("should validate"); + validated.virtual_hosts[0].routes[0].route_match.headers[0] + .match_specifier + .clone() + } + + fn make_weighted_route(weights: &[(&str, Option)]) -> EnvoyRoute { + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + + let mut wc = EnvoyWeightedCluster::new(); + for (name, weight) in weights { + let mut cw = ClusterWeight::new(); + cw.set_name(*name); + if let Some(weight) = weight { + let mut value = UInt32Value::new(); + value.set_value(*weight); + cw.set_weight(value); + } + wc.clusters_mut().push(cw); + } + + let mut route_action = EnvoyRouteAction::new(); + route_action.set_weighted_clusters(wc); + + let mut route = EnvoyRoute::new(); + route.set_match(route_match); + route.set_route(route_action); + route + } + + #[test] + fn validate_basic() { + let rc = make_route_config("rc-1"); + let validated = RouteConfigResource::validate(rc).expect("should validate"); + assert_eq!(validated.name, "rc-1"); + assert_eq!(validated.virtual_hosts.len(), 1); + assert_eq!(validated.virtual_hosts[0].routes.len(), 1); + } + + #[test] + fn validate_empty_virtual_hosts() { + let mut rc = RouteConfiguration::new(); + rc.set_name("rc-1"); + let validated = RouteConfigResource::validate(rc).expect("should validate"); + assert!(validated.virtual_hosts.is_empty()); + } + + #[test] + fn validate_virtual_host_no_domains() { + let mut vh = EnvoyVirtualHost::new(); + vh.set_name("vh1"); + vh.routes_mut().push(make_route("/", "cluster-1")); + + let mut rc = RouteConfiguration::new(); + rc.set_name("rc-1"); + rc.virtual_hosts_mut().push(vh); + + let err = RouteConfigResource::validate(rc).unwrap_err(); + assert!(err.to_string().contains("no domains")); + } + + #[test] + fn validate_route_missing_match() { + let mut route = EnvoyRoute::new(); + route.set_route(EnvoyRouteAction::new()); + + let mut vh = EnvoyVirtualHost::new(); + vh.set_name("vh1"); + vh.domains_mut().push("*"); + vh.routes_mut().push(route); + + let mut rc = RouteConfiguration::new(); + rc.set_name("rc-1"); + rc.virtual_hosts_mut().push(vh); + + let err = RouteConfigResource::validate(rc).unwrap_err(); + assert!(err.to_string().contains("missing match field")); + } + + #[test] + fn validate_route_skips_query_parameter_matchers() { + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + let mut qp = QueryParameterMatcher::new(); + qp.set_name("q"); + route_match.query_parameters_mut().push(qp); + + let mut route_action = EnvoyRouteAction::new(); + route_action.set_cluster("cluster-1"); + + let mut route = EnvoyRoute::new(); + route.set_match(route_match); + route.set_route(route_action); + + let mut vh = EnvoyVirtualHost::new(); + vh.set_name("vh1"); + vh.domains_mut().push("*"); + vh.routes_mut().push(route); + + let mut rc = RouteConfiguration::new(); + rc.set_name("rc-1"); + rc.virtual_hosts_mut().push(vh); + + let validated = RouteConfigResource::validate(rc).expect("should validate"); + assert!(validated.virtual_hosts[0].routes.is_empty()); + } + + #[test] + fn validate_route_rejects_non_route_action() { + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + let mut route = EnvoyRoute::new(); + route.set_match(route_match); + route.set_redirect(RedirectAction::new()); + + let mut vh = EnvoyVirtualHost::new(); + vh.set_name("vh1"); + vh.domains_mut().push("*"); + vh.routes_mut().push(route); + + let mut rc = RouteConfiguration::new(); + rc.set_name("rc-1"); + rc.virtual_hosts_mut().push(vh); + + let err = RouteConfigResource::validate(rc).unwrap_err(); + assert!(err.to_string().contains("only route action is supported")); + } + + #[test] + fn validate_weighted_clusters() { + let rc = wrap_route( + "rc-1", + make_weighted_route(&[("cluster-a", Some(80)), ("cluster-b", Some(20))]), + ); + + let validated = RouteConfigResource::validate(rc).expect("should validate"); + match &validated.virtual_hosts[0].routes[0].action { + RouteAction::WeightedClusters(clusters) => { + assert_eq!(clusters.len(), 2); + assert_eq!(clusters[0].name, "cluster-a"); + assert_eq!(clusters[0].weight, 80); + assert_eq!(clusters[1].name, "cluster-b"); + assert_eq!(clusters[1].weight, 20); + } + other => panic!("expected WeightedClusters, got {other:?}"), + } + } + + #[test] + fn validate_weighted_clusters_drops_zero_weight_entries() { + let rc = wrap_route( + "rc-1", + make_weighted_route(&[ + ("cluster-a", Some(80)), + ("cluster-zero", Some(0)), + ("cluster-unset", None), + ]), + ); + + let validated = RouteConfigResource::validate(rc).expect("should validate"); + match &validated.virtual_hosts[0].routes[0].action { + RouteAction::WeightedClusters(clusters) => { + assert_eq!(clusters.len(), 1); + assert_eq!(clusters[0].name, "cluster-a"); + } + other => panic!("expected WeightedClusters, got {other:?}"), + } + // Zero-weight clusters never receive traffic, so they must not be subscribed to. + assert_eq!(validated.virtual_hosts[0].cluster_names().len(), 1); + } + + #[test] + fn validate_weighted_clusters_rejects_zero_total_weight() { + let rc = wrap_route( + "rc-1", + make_weighted_route(&[("cluster-a", Some(0)), ("cluster-b", None)]), + ); + let err = RouteConfigResource::validate(rc).unwrap_err(); + assert!(err.to_string().contains("non-zero weight")); + } + + #[test] + fn validate_weighted_clusters_rejects_overflowing_total_weight() { + let rc = wrap_route( + "rc-1", + make_weighted_route(&[("cluster-a", Some(u32::MAX)), ("cluster-b", Some(1))]), + ); + let err = RouteConfigResource::validate(rc).unwrap_err(); + assert!(err.to_string().contains("exceeds")); + } + + #[test] + fn validate_skips_route_with_unset_cluster_specifier() { + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + let mut route = EnvoyRoute::new(); + route.set_match(route_match); + route.set_route(EnvoyRouteAction::new()); + + // An unset specifier -- which is also how an unknown specifier from a + // newer control plane decodes -- must skip the route, not NACK. + let validated = + RouteConfigResource::validate(wrap_route("rc-1", route)).expect("should validate"); + assert!(validated.virtual_hosts[0].routes.is_empty()); + } + + #[test] + fn validate_keeps_bin_header_matchers() { + let mut header = EnvoyHeaderMatcher::new(); + header.set_name("x-authz-bin"); + header.set_present_match(true); + + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + route_match.headers_mut().push(header); + + let mut route_action = EnvoyRouteAction::new(); + route_action.set_cluster("cluster-1"); + + let mut route = EnvoyRoute::new(); + route.set_match(route_match); + route.set_route(route_action); + + // Dropping the matcher would widen the route to every request; per A28 it + // must be kept and evaluated as though the header were absent. + let validated = + RouteConfigResource::validate(wrap_route("rc-1", route)).expect("should validate"); + assert_eq!( + validated.virtual_hosts[0].routes[0] + .route_match + .headers + .len(), + 1 + ); + } + + #[test] + fn validate_legacy_prefix_header_matcher() { + let mut header = EnvoyHeaderMatcher::new(); + header.set_name("x-tenant"); + header.set_prefix_match("prod-"); + + assert!(matches!( + validate_header(header), + HeaderMatchSpecifier::String(StringMatcher::Prefix { + value, + ignore_case: false, + }) if value == "prod-" + )); + } + + #[test] + fn validate_legacy_suffix_header_matcher() { + let mut header = EnvoyHeaderMatcher::new(); + header.set_name("x-tenant"); + header.set_suffix_match("-canary"); + + assert!(matches!( + validate_header(header), + HeaderMatchSpecifier::String(StringMatcher::Suffix { + value, + ignore_case: false, + }) if value == "-canary" + )); + } + + #[test] + fn validate_legacy_contains_header_matcher() { + let mut header = EnvoyHeaderMatcher::new(); + header.set_name("x-tenant"); + header.set_contains_match("staging"); + + assert!(matches!( + validate_header(header), + HeaderMatchSpecifier::String(StringMatcher::Contains { + value, + ignore_case: false, + }) if value == "staging" + )); + } + + #[test] + fn validate_legacy_string_header_matchers_reject_empty_values() { + let mut prefix = EnvoyHeaderMatcher::new(); + prefix.set_name("x-tenant"); + prefix.set_prefix_match(""); + + let mut suffix = EnvoyHeaderMatcher::new(); + suffix.set_name("x-tenant"); + suffix.set_suffix_match(""); + + let mut contains = EnvoyHeaderMatcher::new(); + contains.set_name("x-tenant"); + contains.set_contains_match(""); + + for (kind, header) in [ + ("prefix", prefix), + ("suffix", suffix), + ("contains", contains), + ] { + let err = RouteConfigResource::validate(route_config_with_header(header)).unwrap_err(); + assert!(err.to_string().contains(&format!("empty {kind} match"))); + } + } + + #[test] + fn validate_string_matcher_rejects_empty_prefix_suffix_and_contains() { + let mut prefix = EnvoyStringMatcher::new(); + prefix.set_prefix(""); + let mut prefix_header = EnvoyHeaderMatcher::new(); + prefix_header.set_name("x-tenant"); + prefix_header.set_string_match(prefix); + + let mut suffix = EnvoyStringMatcher::new(); + suffix.set_suffix(""); + let mut suffix_header = EnvoyHeaderMatcher::new(); + suffix_header.set_name("x-tenant"); + suffix_header.set_string_match(suffix); + + let mut contains = EnvoyStringMatcher::new(); + contains.set_contains(""); + let mut contains_header = EnvoyHeaderMatcher::new(); + contains_header.set_name("x-tenant"); + contains_header.set_string_match(contains); + + for (kind, header) in [ + ("prefix", prefix_header), + ("suffix", suffix_header), + ("contains", contains_header), + ] { + let err = RouteConfigResource::validate(route_config_with_header(header)).unwrap_err(); + assert!(err.to_string().contains(&format!("empty {kind} match"))); + } + } + + #[test] + fn cluster_names_are_scoped_to_virtual_host() { + let mut rc = make_route_config("rc-1"); + let mut other = EnvoyVirtualHost::new(); + other.set_name("vh2"); + other.domains_mut().push("other.example.com"); + other.routes_mut().push(make_route("/", "cluster-2")); + rc.virtual_hosts_mut().push(other); + + let validated = RouteConfigResource::validate(rc).unwrap(); + assert_eq!( + validated.virtual_hosts[0].cluster_names(), + HashSet::from(["cluster-1".to_string()]) + ); + assert_eq!( + validated.virtual_hosts[1].cluster_names(), + HashSet::from(["cluster-2".to_string()]) + ); + } + + #[test] + fn deserialize_roundtrip() { + let rc = make_route_config("test"); + let bytes = rc.serialize().expect("serialize"); + let deserialized = RouteConfigResource::deserialize(bytes::Bytes::from(bytes)).unwrap(); + assert_eq!(RouteConfigResource::name(&deserialized), "test"); + } + + fn route_config_with_match(route_match: EnvoyRouteMatch) -> RouteConfiguration { + let mut route_action = EnvoyRouteAction::new(); + route_action.set_cluster("cluster-1"); + + let mut route = EnvoyRoute::new(); + route.set_match(route_match); + route.set_route(route_action); + + let mut vh = EnvoyVirtualHost::new(); + vh.set_name("vh1"); + vh.domains_mut().push("*"); + vh.routes_mut().push(route); + + let mut rc = RouteConfiguration::new(); + rc.set_name("rc-1"); + rc.virtual_hosts_mut().push(vh); + rc + } + + #[test] + fn validate_rejects_empty_path_regex() { + let mut regex = RegexMatcher::new(); + regex.set_regex(""); + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_safe_regex(regex); + + let err = RouteConfigResource::validate(route_config_with_match(route_match)).unwrap_err(); + assert!(err.to_string().contains("empty path regex")); + } + + #[test] + fn validate_rejects_empty_header_regex() { + let mut regex = RegexMatcher::new(); + regex.set_regex(""); + let mut header = EnvoyHeaderMatcher::new(); + header.set_name("x-test"); + header.set_safe_regex_match(regex); + + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + route_match.headers_mut().push(header); + + let err = RouteConfigResource::validate(route_config_with_match(route_match)).unwrap_err(); + assert!(err.to_string().contains("empty header regex")); + } + + #[test] + fn validate_rejects_empty_header_matcher_name() { + let mut header = EnvoyHeaderMatcher::new(); + header.set_present_match(true); + + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + route_match.headers_mut().push(header); + + let err = RouteConfigResource::validate(route_config_with_match(route_match)).unwrap_err(); + assert!(err.to_string().contains("header matcher name is empty")); + } + + #[test] + fn validate_runtime_fraction_normalizes_denominator() { + let mut frac = FractionalPercent::new(); + frac.set_numerator(25); + frac.set_denominator(DenominatorType::Hundred); + let mut rf = RuntimeFractionalPercent::new(); + rf.set_default_value(frac); + + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + route_match.set_runtime_fraction(rf); + + let validated = + RouteConfigResource::validate(route_config_with_match(route_match)).unwrap(); + assert_eq!( + validated.virtual_hosts[0].routes[0] + .route_match + .match_fraction, + Some(250_000) + ); + } + + #[test] + fn validate_rejects_runtime_fraction_without_default_value() { + let mut route_match = EnvoyRouteMatch::new(); + route_match.set_prefix("/"); + route_match.set_runtime_fraction(RuntimeFractionalPercent::new()); + + let err = RouteConfigResource::validate(route_config_with_match(route_match)).unwrap_err(); + assert!(err.to_string().contains("default_value")); + } +} diff --git a/grpc-xds/src/xds_config.rs b/grpc-xds/src/xds_config.rs new file mode 100644 index 000000000..a2bbb801c --- /dev/null +++ b/grpc-xds/src/xds_config.rs @@ -0,0 +1,271 @@ +/* + * + * Copyright 2025 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +//! [`XdsConfig`]: the atomic xDS configuration snapshot for a channel. +//! +//! Per gRFC A74, it bundles everything needed to route and load balance a +//! single RPC -- a Listener, its RouteConfiguration, and every reachable +//! Cluster and its endpoints -- into one immutable value, so a config +//! update is atomic and never exposes a partial mix of old and new state. + +// TODO: remove once the xDS dependency manager is implemented. +#![allow(dead_code)] + +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; + +use crate::resource::{ + ClusterResource, EndpointAddress, EndpointsResource, ListenerResource, RouteConfigResource, + RouteSource, VirtualHost, +}; + +/// The atomic xDS configuration snapshot for a channel. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub(crate) struct XdsConfig { + /// The channel's Listener resource (LDS). + pub(crate) listener: Arc, + /// The Listener's resolved RouteConfiguration: either fetched via RDS or + /// embedded inline in the Listener (see + /// [`crate::resource::RouteSource`]). + pub(crate) route_config: Arc, + /// Index of the virtual host selected for this channel's data-plane + /// authority. Kept as an index so it can safely refer into the immutable + /// `route_config` without a self-referential borrow. + virtual_host_index: usize, + /// Every cluster transitively reachable from [`Self::virtual_host`], keyed + /// by cluster name. Includes aggregate clusters' descendants. + /// + /// A cluster that failed to resolve (CDS/EDS validation failure, NACK, + /// DNS lookup failure, cyclic aggregate reference, ...) is still present + /// here with an `Err` value rather than omitted, per gRFC A74, + /// so callers can distinguish "still loading" from "failed". + pub(crate) clusters: HashMap, +} + +impl XdsConfig { + /// Constructs a snapshot for a selected virtual host. + /// + /// Returns `None` if `virtual_host_index` is invalid, or if `route_config` + /// is not the route configuration the listener actually points at: the + /// same allocation for an inline config, or the same name for RDS. + pub(crate) fn try_new( + listener: Arc, + route_config: Arc, + virtual_host_index: usize, + clusters: HashMap, + ) -> Option { + route_config.virtual_hosts.get(virtual_host_index)?; + match &listener.route_source { + RouteSource::Inline(inline) if !Arc::ptr_eq(inline, &route_config) => return None, + RouteSource::Rds(name) if *name != route_config.name => return None, + _ => {} + } + Some(Self { + listener, + route_config, + virtual_host_index, + clusters, + }) + } + + /// Returns the single virtual host selected for this channel. + pub(crate) fn virtual_host(&self) -> &VirtualHost { + self.route_config + .virtual_hosts + .get(self.virtual_host_index) + .expect("XdsConfig constructor validates virtual_host_index") + } +} + +/// Resolution result for a single cluster: either its fully resolved +/// config, or the error that prevented it from resolving. +pub(crate) type ClusterResult = Result, ClusterResolutionError>; + +/// A cluster-scoped dependency-resolution failure suitable for reporting to +/// the data plane without exposing an xDS transport-layer error type. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ClusterResolutionError { + message: Arc, +} + +impl ClusterResolutionError { + pub(crate) fn new(message: impl Into>) -> Self { + Self { + message: message.into(), + } + } + + pub(crate) fn message(&self) -> &str { + &self.message + } +} + +impl fmt::Display for ClusterResolutionError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.message) + } +} + +impl std::error::Error for ClusterResolutionError {} + +/// A single resolved cluster: its static CDS configuration plus its +/// dynamically resolved children. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub(crate) struct ClusterConfig { + /// The validated CDS resource itself (name, discovery mechanism, ...). + pub(crate) cluster: Arc, + /// This cluster's dynamically resolved children. + pub(crate) children: ClusterChildren, +} + +/// A cluster's dynamically resolved children. +#[derive(Debug, Clone)] +pub(crate) enum ClusterChildren { + /// A leaf cluster (EDS or LogicalDNS): its resolved endpoints. + Leaf(LeafEndpoints), + /// An aggregate cluster (gRFC A37): the fully resolved list of leaf + /// cluster names to fall over across, in priority order. + Aggregate(AggregateClusters), +} + +/// Resolved aggregate-cluster dependencies. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub(crate) struct AggregateClusters { + pub(crate) leaf_clusters: Vec, + /// Ambient LDS/RDS/CDS errors associated with this aggregate branch. + pub(crate) resolution_note: Option>, +} + +/// A leaf cluster's resolved endpoints. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub(crate) struct LeafEndpoints { + /// A leaf cluster's resolved endpoints, if any have been resolved yet. + /// + /// `None` while the first resolution attempt is still outstanding or has + /// only ever failed; see `resolution_note` for why. + pub(crate) source: Option, + /// Ambient LDS/RDS/CDS/EDS or DNS diagnostic information for this branch. + /// + /// Unlike the `Err` side of [`ClusterResult`], a note does not mean the + /// cluster failed: `source` may still hold stale but valid endpoints. + pub(crate) resolution_note: Option>, +} + +/// The origin of a leaf cluster's endpoints. +#[derive(Debug, Clone)] +pub(crate) enum LeafEndpointSource { + /// Resolved via EDS: the validated ClusterLoadAssignment resource. + Eds(Arc), + /// Resolved via DNS resolution of a LogicalDNS cluster's target. + LogicalDns(Arc<[EndpointAddress]>), +} + +#[cfg(test)] +mod tests { + use super::*; + + fn route_config() -> Arc { + Arc::new(RouteConfigResource { + name: "routes".into(), + virtual_hosts: vec![ + VirtualHost { + name: "first".into(), + domains: vec!["first.example.com".into()], + routes: Vec::new(), + }, + VirtualHost { + name: "second".into(), + domains: vec!["second.example.com".into()], + routes: Vec::new(), + }, + ], + }) + } + + fn inline_listener(route_config: Arc) -> Arc { + Arc::new(ListenerResource { + name: "listener".into(), + route_source: RouteSource::Inline(route_config), + }) + } + + #[test] + fn selected_virtual_host_uses_route_config_index() { + let route_config = route_config(); + let listener = inline_listener(Arc::clone(&route_config)); + let config = XdsConfig::try_new( + Arc::clone(&listener), + Arc::clone(&route_config), + 1, + HashMap::new(), + ) + .expect("valid selected virtual host"); + + assert_eq!(config.virtual_host().name, "second"); + let RouteSource::Inline(inline) = &listener.route_source else { + panic!("expected inline route config"); + }; + assert!(Arc::ptr_eq(inline, &config.route_config)); + } + + #[test] + fn selected_virtual_host_rejects_invalid_index() { + let route_config = route_config(); + let listener = inline_listener(Arc::clone(&route_config)); + assert!(XdsConfig::try_new(listener, route_config, 2, HashMap::new()).is_none()); + } + + #[test] + fn selected_virtual_host_rejects_different_inline_allocation() { + let listener_route_config = route_config(); + let snapshot_route_config = route_config(); + let listener = inline_listener(listener_route_config); + assert!(XdsConfig::try_new(listener, snapshot_route_config, 0, HashMap::new()).is_none()); + } + + #[test] + fn selected_virtual_host_accepts_matching_rds_name() { + let route_config = route_config(); + let listener = Arc::new(ListenerResource { + name: "listener".into(), + route_source: RouteSource::Rds("routes".into()), + }); + assert!(XdsConfig::try_new(listener, route_config, 0, HashMap::new()).is_some()); + } + + #[test] + fn selected_virtual_host_rejects_mismatched_rds_name() { + let route_config = route_config(); + let listener = Arc::new(ListenerResource { + name: "listener".into(), + route_source: RouteSource::Rds("other-routes".into()), + }); + assert!(XdsConfig::try_new(listener, route_config, 0, HashMap::new()).is_none()); + } +}