rws_gateway_api/gateway_api/
routes.rs

1//! HTTPRoutes compiled for the data plane: matches, filters, and backends
2//! resolved to the ready endpoints of their Services (through the Service's
3//! port → EndpointSlice port), so the router does no lookups per request.
4
5use k8s_openapi::api::discovery::v1::EndpointSlice;
6use kube::ResourceExt;
7
8use super::resources::{GrpcRoute, HttpRoute};
9use super::types::L4RouteSpec;
10use super::translate::Snapshot;
11use super::types::*;
12
13#[derive(Clone, Debug, PartialEq)]
14pub struct CompiledHttpRoute {
15    /// (namespace, name).
16    pub key: (String, String),
17    pub hostnames: Vec<String>,
18    /// `creationTimestamp` (RFC 3339): older routes win ties.
19    pub created: String,
20    pub rules: Vec<CompiledRule>,
21}
22
23#[derive(Clone, Debug, PartialEq)]
24pub struct CompiledRule {
25    pub name: Option<String>,
26    /// Never empty: no matches means one match on prefix `/`.
27    pub matches: Vec<HttpRouteMatch>,
28    pub filters: Vec<HttpRouteFilter>,
29    pub backends: Vec<Backend>,
30    pub timeouts: Option<HttpRouteTimeouts>,
31    pub retry: Option<HttpRouteRetry>,
32    /// The rule's `RequestMirror` filters, resolved.
33    pub mirrors: Vec<Mirror>,
34}
35
36/// Where a `RequestMirror` filter sends copies, and how many.
37#[derive(Clone, Debug, PartialEq)]
38pub struct Mirror {
39    pub endpoints: Result<Vec<String>, String>,
40    /// Share of requests mirrored, 0.0–1.0 (`percent` or `fraction`;
41    /// all when neither is set).
42    pub share: f64,
43}
44
45impl Mirror {
46    fn from_filter(filter: &HttpRequestMirrorFilter, namespace: &str, snapshot: &Snapshot) -> Mirror {
47        let share = match (&filter.fraction, filter.percent) {
48            (Some(f), _) => f.numerator as f64 / f.denominator.unwrap_or(100).max(1) as f64,
49            (None, Some(percent)) => percent as f64 / 100.0,
50            (None, None) => 1.0,
51        };
52        Mirror { endpoints: resolve(&filter.backend_ref, namespace, "HTTPRoute", snapshot).endpoints, share: share.clamp(0.0, 1.0) }
53    }
54}
55
56#[derive(Clone, Debug, PartialEq)]
57pub struct Backend {
58    pub weight: u32,
59    /// `ip:port` of every ready endpoint; `Err` for a reference that doesn't
60    /// resolve (its share of requests gets `500`, as the spec says).
61    pub endpoints: Result<Vec<String>, String>,
62    /// The Service port's `appProtocol` (`kubernetes.io/h2c`, `kubernetes.io/ws`).
63    pub app_protocol: Option<String>,
64    /// TLS to the backend, when a BackendTLSPolicy covers its Service port;
65    /// `Err` for a policy that isn't usable (those requests get `500`).
66    pub tls: Option<Result<super::backend_tls::BackendTls, String>>,
67    pub filters: Vec<HttpRouteFilter>,
68}
69
70pub fn compile(route: &HttpRoute, snapshot: &Snapshot) -> CompiledHttpRoute {
71    let namespace = route.namespace().unwrap_or_default();
72    let rules = route
73        .spec
74        .rules
75        .iter()
76        .map(|rule| CompiledRule {
77            name: rule.name.clone(),
78            matches: if rule.matches.is_empty() {
79                vec![HttpRouteMatch { path: Some(HttpPathMatch { match_type: Some("PathPrefix".to_string()), value: Some("/".to_string()) }), ..Default::default() }]
80            } else {
81                rule.matches.clone()
82            },
83            filters: rule.filters.clone(),
84            backends: rule
85                .backend_refs
86                .iter()
87                .map(|b| {
88                    let resolved = resolve(&b.backend_ref.object, &namespace, "HTTPRoute", snapshot);
89                    Backend { weight: b.backend_ref.weight.unwrap_or(1).max(0) as u32, endpoints: resolved.endpoints, app_protocol: resolved.app_protocol, tls: resolved.tls, filters: b.filters.clone() }
90                })
91                .collect(),
92            timeouts: rule.timeouts.clone(),
93            retry: rule.retry.clone(),
94            mirrors: rule
95                .filters
96                .iter()
97                .filter(|f| f.filter_type == "RequestMirror")
98                .filter_map(|f| f.request_mirror.as_ref())
99                .map(|f| Mirror::from_filter(f, &namespace, snapshot))
100                .collect(),
101        })
102        .collect();
103    CompiledHttpRoute {
104        key: (namespace, route.name_any()),
105        hostnames: route.spec.hostnames.clone(),
106        created: route.metadata.creation_timestamp.as_ref().map(|t| t.0.to_string()).unwrap_or_default(),
107        rules,
108    }
109}
110
111/// The `matches.path` type a gRPC `method` match with only a method
112/// compiles to: the value is the method, matched as the path's second
113/// segment of `/<service>/<method>` (not an API type; internal).
114pub const GRPC_METHOD_MATCH: &str = "rws.GRPCMethod";
115
116/// A GRPCRoute compiled to the same form as an HTTPRoute: a `method` match
117/// becomes a path match on `/<service>/<method>` (`Exact` for both,
118/// `PathPrefix` `/<service>` for a service alone, [`GRPC_METHOD_MATCH`] for
119/// a method alone); `RegularExpression` matches nothing (implementation
120/// specific, not supported). Backends are reached over h2c.
121pub fn compile_grpc(route: &GrpcRoute, snapshot: &Snapshot) -> CompiledHttpRoute {
122    let namespace = route.namespace().unwrap_or_default();
123    let path = |kind: &str, value: String| Some(HttpPathMatch { match_type: Some(kind.to_string()), value: Some(value) });
124    let rules = route
125        .spec
126        .rules
127        .iter()
128        .map(|rule| {
129            let matches = if rule.matches.is_empty() {
130                vec![HttpRouteMatch { path: path("PathPrefix", "/".to_string()), ..Default::default() }]
131            } else {
132                rule.matches
133                    .iter()
134                    .map(|m| {
135                        let path = match &m.method {
136                            None => path("PathPrefix", "/".to_string()),
137                            Some(method) if method.match_type.as_deref().unwrap_or("Exact") != "Exact" => path("rws.Unsupported", String::new()),
138                            Some(method) => match (&method.service, &method.method) {
139                                (Some(service), Some(name)) => path("Exact", format!("/{service}/{name}")),
140                                (Some(service), None) => path("PathPrefix", format!("/{service}")),
141                                (None, Some(name)) => path(GRPC_METHOD_MATCH, name.clone()),
142                                (None, None) => path("PathPrefix", "/".to_string()),
143                            },
144                        };
145                        HttpRouteMatch { path, headers: m.headers.clone(), ..Default::default() }
146                    })
147                    .collect()
148            };
149            CompiledRule {
150                name: rule.name.clone(),
151                matches,
152                filters: rule.filters.clone(),
153                backends: rule
154                    .backend_refs
155                    .iter()
156                    .map(|b| {
157                        let resolved = resolve(&b.backend_ref.object, &namespace, "GRPCRoute", snapshot);
158                        Backend {
159                            weight: b.backend_ref.weight.unwrap_or(1).max(0) as u32,
160                            endpoints: resolved.endpoints,
161                            app_protocol: Some("kubernetes.io/h2c".to_string()),
162                            tls: resolved.tls,
163                            filters: b.filters.clone(),
164                        }
165                    })
166                    .collect(),
167                timeouts: None,
168                retry: None,
169                mirrors: rule
170                    .filters
171                    .iter()
172                    .filter(|f| f.filter_type == "RequestMirror")
173                    .filter_map(|f| f.request_mirror.as_ref())
174                    .map(|f| Mirror::from_filter(f, &namespace, snapshot))
175                    .collect(),
176            }
177        })
178        .collect();
179    CompiledHttpRoute {
180        key: (namespace, route.name_any()),
181        hostnames: route.spec.hostnames.clone(),
182        created: route.metadata.creation_timestamp.as_ref().map(|t| t.0.to_string()).unwrap_or_default(),
183        rules,
184    }
185}
186
187/// A TLSRoute, TCPRoute or UDPRoute: one rule per rule, no matches (the
188/// data plane picks the route by SNI or age, then a rule's backend by
189/// weight).
190pub fn compile_l4(route: &kube::core::Object<L4RouteSpec, RouteStatus>, kind: &str, snapshot: &Snapshot) -> CompiledHttpRoute {
191    let namespace = route.namespace().unwrap_or_default();
192    let rules = route
193        .spec
194        .rules
195        .iter()
196        .map(|rule| CompiledRule {
197            name: rule.name.clone(),
198            matches: Vec::new(),
199            filters: Vec::new(),
200            backends: rule
201                .backend_refs
202                .iter()
203                .map(|b| {
204                    let resolved = resolve(&b.object, &namespace, kind, snapshot);
205                    Backend { weight: b.weight.unwrap_or(1).max(0) as u32, endpoints: resolved.endpoints, app_protocol: resolved.app_protocol, tls: None, filters: Vec::new() }
206                })
207                .collect(),
208            timeouts: None,
209            retry: None,
210            mirrors: Vec::new(),
211        })
212        .collect();
213    CompiledHttpRoute {
214        key: (namespace, route.name_any()),
215        hostnames: route.spec.hostnames.clone(),
216        created: route.metadata.creation_timestamp.as_ref().map(|t| t.0.to_string()).unwrap_or_default(),
217        rules,
218    }
219}
220
221/// What a backendRef resolves to.
222pub struct Resolved {
223    /// Every ready `ip:port`; `Err` when the reference doesn't resolve.
224    pub endpoints: Result<Vec<String>, String>,
225    pub app_protocol: Option<String>,
226    pub tls: Option<Result<super::backend_tls::BackendTls, String>>,
227}
228
229/// A backendRef → its ready endpoints, `appProtocol` and BackendTLSPolicy.
230pub fn resolve(reference: &BackendObjectReference, route_namespace: &str, route_kind: &str, snapshot: &Snapshot) -> Resolved {
231    let failed = |e: String| Resolved { endpoints: Err(e), app_protocol: None, tls: None };
232    if !reference.group().is_empty() || reference.kind() != "Service" {
233        return failed(format!("unsupported backend kind {}", reference.kind()));
234    }
235    let namespace = reference.namespace.clone().unwrap_or_else(|| route_namespace.to_string());
236    if namespace != route_namespace && !super::translate::granted(snapshot, GATEWAY_GROUP, route_kind, route_namespace, "", "Service", &namespace, &reference.name) {
237        return failed("not permitted by a ReferenceGrant".to_string());
238    }
239    let Some(service) = snapshot.services.iter().find(|s| s.namespace().as_deref() == Some(namespace.as_str()) && s.name_any() == reference.name) else {
240        return failed(format!("Service {namespace}/{} not found", reference.name));
241    };
242    let Some(port) = reference.port else {
243        return failed("a Service backend needs a port".to_string());
244    };
245    let service_port = service.spec.as_ref().and_then(|s| s.ports.as_ref()).and_then(|ports| ports.iter().find(|p| p.port == port)).cloned();
246    let Some(service_port) = service_port else {
247        return failed(format!("Service {namespace}/{} has no port {port}", reference.name));
248    };
249    let endpoints = endpoints(snapshot, &namespace, &service.name_any(), service_port.name.as_deref());
250    let tls = if snapshot.backend_tls_policies.is_empty() {
251        None
252    } else {
253        super::backend_tls::for_service(&super::backend_tls::evaluate(snapshot), snapshot, &namespace, &reference.name, service_port.name.as_deref())
254    };
255    Resolved { endpoints: Ok(endpoints), app_protocol: service_port.app_protocol.clone(), tls }
256}
257
258/// Ready endpoints of a Service port, from its EndpointSlices.
259fn endpoints(snapshot: &Snapshot, namespace: &str, service: &str, port_name: Option<&str>) -> Vec<String> {
260    let mut out = Vec::new();
261    let slices = snapshot.endpoint_slices.iter().filter(|s: &&EndpointSlice| {
262        s.namespace().as_deref() == Some(namespace) && s.labels().get("kubernetes.io/service-name").map(String::as_str) == Some(service)
263    });
264    for slice in slices {
265        let Some(port) = slice
266            .ports
267            .iter()
268            .flatten()
269            .find(|p| p.name.as_deref().unwrap_or("") == port_name.unwrap_or(""))
270            .and_then(|p| p.port)
271        else {
272            continue;
273        };
274        for endpoint in &slice.endpoints {
275            if endpoint.conditions.as_ref().and_then(|c| c.ready) == Some(false) {
276                continue;
277            }
278            for address in &endpoint.addresses {
279                out.push(if address.contains(':') { format!("[{address}]:{port}") } else { format!("{address}:{port}") });
280            }
281        }
282    }
283    out.sort();
284    out.dedup();
285    out
286}
287
288#[cfg(test)]
289mod tests;
rws_gateway_api/gateway_api/routes.rs

rws_gateway_api/gateway_api/
routes.rs
