rws_gateway_api/gateway_api/
mesh.rs1pub mod config;
27pub mod inject;
28pub mod sidecar;
29
30use std::collections::{BTreeMap, HashMap};
31use std::sync::Arc;
32
33use k8s_openapi::api::core::v1::Service;
34use k8s_openapi::api::discovery::v1::EndpointSlice;
35use kube::ResourceExt;
36use serde::{Deserialize, Serialize};
37
38use super::resources::{GrpcRoute, HttpRoute, ReferenceGrant};
39use super::routes::CompiledHttpRoute;
40use super::translate::{ListenerConfig, Snapshot};
41use super::types::*;
42
43#[derive(Clone, Debug, Default, Serialize, Deserialize)]
46#[serde(rename_all = "camelCase")]
47pub struct MeshSnapshot {
48 #[serde(default)]
49 pub http_routes: Vec<HttpRoute>,
50 #[serde(default)]
51 pub grpc_routes: Vec<GrpcRoute>,
52 #[serde(default)]
53 pub services: Vec<Service>,
54 #[serde(default)]
55 pub endpoint_slices: Vec<EndpointSlice>,
56 #[serde(default)]
57 pub reference_grants: Vec<ReferenceGrant>,
58}
59
60pub fn is_service_parent(parent: &ParentReference) -> bool {
62 parent.group() == "" && parent.kind() == "Service"
63}
64
65fn parent_service(parent: &ParentReference, route_namespace: &str) -> (String, String) {
67 (parent.namespace.clone().unwrap_or_else(|| route_namespace.to_string()), parent.name.clone())
68}
69
70pub fn snapshot(from: &Snapshot) -> MeshSnapshot {
72 let meshy = |parents: &[ParentReference]| parents.iter().any(is_service_parent);
73 let http_routes: Vec<HttpRoute> = from.http_routes.iter().filter(|r| meshy(&r.spec.parent_refs)).cloned().collect();
74 let grpc_routes: Vec<GrpcRoute> = from.grpc_routes.iter().filter(|r| meshy(&r.spec.parent_refs)).cloned().collect();
75 let mut wanted: std::collections::BTreeSet<(String, String)> = Default::default();
77 let mut note = |namespace: &str, parents: &[ParentReference], backends: Vec<&BackendObjectReference>| {
78 for p in parents.iter().filter(|p| is_service_parent(p)) {
79 wanted.insert(parent_service(p, namespace));
80 }
81 for b in backends {
82 wanted.insert((b.namespace.clone().unwrap_or_else(|| namespace.to_string()), b.name.clone()));
83 }
84 };
85 for r in &http_routes {
86 let namespace = r.namespace().unwrap_or_default();
87 note(&namespace, &r.spec.parent_refs, r.spec.rules.iter().flat_map(|rule| rule.backend_refs.iter().map(|b| &b.backend_ref.object)).collect());
88 }
89 for r in &grpc_routes {
90 let namespace = r.namespace().unwrap_or_default();
91 note(&namespace, &r.spec.parent_refs, r.spec.rules.iter().flat_map(|rule| rule.backend_refs.iter().map(|b| &b.backend_ref.object)).collect());
92 }
93 let named = |namespace: Option<String>, name: String| wanted.contains(&(namespace.unwrap_or_default(), name));
94 let services = from.services.iter().filter(|s| named(s.namespace(), s.name_any())).cloned().collect();
95 let endpoint_slices = from
96 .endpoint_slices
97 .iter()
98 .filter(|e| e.labels().get("kubernetes.io/service-name").is_some_and(|service| named(e.namespace(), service.clone())))
99 .cloned()
100 .collect();
101 let mut mesh = MeshSnapshot { http_routes, grpc_routes, services, endpoint_slices, reference_grants: from.reference_grants.clone() };
102 trim(&mut mesh);
103 mesh
104}
105
106fn trim(mesh: &mut MeshSnapshot) {
110 fn meta(m: &mut kube::core::ObjectMeta) {
111 m.managed_fields = None;
112 m.resource_version = None;
113 m.generation = None;
114 m.uid = None;
115 m.annotations = None;
116 }
117 for r in &mut mesh.http_routes {
118 meta(&mut r.metadata);
119 r.status = None;
120 }
121 for r in &mut mesh.grpc_routes {
122 meta(&mut r.metadata);
123 r.status = None;
124 }
125 for s in &mut mesh.services {
126 meta(&mut s.metadata);
127 s.status = None;
128 }
129 for e in &mut mesh.endpoint_slices {
130 meta(&mut e.metadata);
131 }
132 for g in &mut mesh.reference_grants {
133 meta(&mut g.metadata);
134 }
135}
136
137pub type MeshTable = HashMap<(String, u16), Arc<Vec<ListenerConfig>>>;
140
141fn service_parents(parents: &[ParentReference], namespace: &str) -> Vec<((String, String), Option<i32>)> {
143 parents.iter().filter(|p| is_service_parent(p)).map(|p| (parent_service(p, namespace), p.port)).collect()
144}
145
146pub fn table(mesh: &MeshSnapshot, client_namespace: &str) -> MeshTable {
148 let snapshot = compile_snapshot(mesh);
149 let mut attached: Vec<((String, String), Option<i32>, String, CompiledHttpRoute)> = Vec::new();
151 for r in &mesh.http_routes {
152 let namespace = r.namespace().unwrap_or_default();
153 let compiled = without_hostnames(super::routes::compile(r, &snapshot));
154 for (service, port) in service_parents(&r.spec.parent_refs, &namespace) {
155 attached.push((service, port, namespace.clone(), compiled.clone()));
156 }
157 }
158 for r in &mesh.grpc_routes {
159 let namespace = r.namespace().unwrap_or_default();
160 let compiled = without_hostnames(super::routes::compile_grpc(r, &snapshot));
161 for (service, port) in service_parents(&r.spec.parent_refs, &namespace) {
162 attached.push((service, port, namespace.clone(), compiled.clone()));
163 }
164 }
165 let mut table = MeshTable::new();
166 for service in &mesh.services {
167 let key = (service.namespace().unwrap_or_default(), service.name_any());
168 let routes_here = |namespace: &str| attached.iter().filter(|(s, _, n, _)| *s == key && n == namespace).collect::<Vec<_>>();
169 let consumer = if client_namespace != key.0 { routes_here(client_namespace) } else { Vec::new() };
171 let chosen = if consumer.is_empty() { routes_here(&key.0) } else { consumer };
172 if chosen.is_empty() {
173 continue;
174 }
175 let Some(spec) = &service.spec else { continue };
176 let ips = cluster_ips(spec);
177 for port in spec.ports.iter().flatten() {
178 let routes: Vec<CompiledHttpRoute> = chosen.iter().filter(|(_, limit, _, _)| limit.is_none_or(|p| p == port.port)).map(|(_, _, _, r)| r.clone()).collect();
179 if routes.is_empty() {
180 continue;
181 }
182 let listener = ListenerConfig {
183 gateway: key.clone(),
184 name: port.name.clone().unwrap_or_else(|| port.port.to_string()),
185 port: port.port,
186 protocol: "HTTP".to_string(),
187 hostname: None,
188 container_port: port.port as u16,
189 tls: None,
190 client_ca: None,
191 backend_client_cert: None,
192 request_filters: Vec::new(),
193 http_routes: routes,
194 };
195 let listeners = Arc::new(vec![listener]);
196 for ip in &ips {
197 table.insert((ip.clone(), port.port as u16), Arc::clone(&listeners));
198 }
199 }
200 }
201 table
202}
203
204fn cluster_ips(spec: &k8s_openapi::api::core::v1::ServiceSpec) -> Vec<String> {
206 let mut ips: Vec<String> = spec.cluster_ips.clone().unwrap_or_default();
207 if let Some(ip) = &spec.cluster_ip {
208 if !ips.contains(ip) {
209 ips.push(ip.clone());
210 }
211 }
212 ips.retain(|ip| !ip.is_empty() && ip != "None");
213 ips
214}
215
216fn without_hostnames(mut route: CompiledHttpRoute) -> CompiledHttpRoute {
218 route.hostnames.clear();
219 route
220}
221
222fn compile_snapshot(mesh: &MeshSnapshot) -> Snapshot {
226 let mut grants = mesh.reference_grants.clone();
227 let mut implied = |kind: &str, namespace: &str, parents: &[ParentReference]| {
228 for ((service_namespace, name), _) in service_parents(parents, namespace) {
229 if service_namespace != namespace {
230 grants.push(implied_grant(kind, namespace, &service_namespace, &name));
231 }
232 }
233 };
234 for r in &mesh.http_routes {
235 implied("HTTPRoute", &r.namespace().unwrap_or_default(), &r.spec.parent_refs);
236 }
237 for r in &mesh.grpc_routes {
238 implied("GRPCRoute", &r.namespace().unwrap_or_default(), &r.spec.parent_refs);
239 }
240 Snapshot { services: mesh.services.clone(), endpoint_slices: mesh.endpoint_slices.clone(), reference_grants: grants, ..Default::default() }
241}
242
243fn implied_grant(kind: &str, from_namespace: &str, to_namespace: &str, to_name: &str) -> ReferenceGrant {
244 let mut grant = ReferenceGrant::new(
245 "rws-mesh-consumer",
246 &super::resources::reference_grants(),
247 ReferenceGrantSpec {
248 from: vec![ReferenceGrantFrom { group: GATEWAY_GROUP.to_string(), kind: kind.to_string(), namespace: from_namespace.to_string() }],
249 to: vec![ReferenceGrantTo { group: String::new(), kind: "Service".to_string(), name: Some(to_name.to_string()) }],
250 },
251 );
252 grant.metadata.namespace = Some(to_namespace.to_string());
253 grant
254}
255
256pub fn parent_acceptance(parent: &ParentReference, route_namespace: &str, services: &[Service]) -> (bool, &'static str, String) {
259 let (namespace, name) = parent_service(parent, route_namespace);
260 let Some(service) = services.iter().find(|s| s.namespace().as_deref() == Some(namespace.as_str()) && s.name_any() == name) else {
261 return (false, "NoMatchingParent", format!("Service {namespace}/{name} not found"));
262 };
263 if let Some(port) = parent.port {
264 let has_port = service.spec.as_ref().and_then(|s| s.ports.as_ref()).is_some_and(|ports| ports.iter().any(|p| p.port == port));
265 if !has_port {
266 return (false, "NoMatchingParent", format!("Service {namespace}/{name} has no port {port}"));
267 }
268 }
269 (true, "Accepted", "Route is accepted".to_string())
270}
271
272pub fn only_parent_references(references: &[&BackendObjectReference], parents: &[ParentReference], route_namespace: &str) -> bool {
275 let parents = service_parents(parents, route_namespace);
276 references.iter().all(|r| {
277 let namespace = r.namespace.clone().unwrap_or_else(|| route_namespace.to_string());
278 namespace == route_namespace || parents.iter().any(|((n, name), _)| *n == namespace && *name == r.name)
279 })
280}
281
282pub fn describe(table: &MeshTable) -> BTreeMap<String, usize> {
284 table.iter().map(|((ip, port), listeners)| (format!("{ip}:{port}"), listeners.iter().map(|l| l.http_routes.len()).sum())).collect()
285}
286
287#[cfg(test)]
288mod tests;