rws_gateway_api/gateway_api/
services.rs

1//! The Service rws keeps for each Gateway: the Gateway's own address
2//! (`LoadBalancer` by default), one port per listener port, each mapped to a
3//! container port allocated from a range — the pods tell Gateways apart by
4//! the port a connection arrives on, and never bind ports below 1024.
5//!
6//! Allocations live in the Services themselves: a listener port keeps its
7//! container port for as long as its Service has it, so every replica, and
8//! every restart, agrees on the mapping without any other storage.
9
10use std::collections::{BTreeMap, BTreeSet};
11
12use k8s_openapi::api::core::v1::{Service, ServicePort, ServiceSpec};
13use k8s_openapi::api::discovery::v1::{Endpoint, EndpointConditions, EndpointPort, EndpointSlice};
14use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ObjectMeta, OwnerReference};
15use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;
16use kube::ResourceExt;
17
18use super::resources::Gateway;
19use super::translate::{Snapshot, GATEWAY_NAME_LABEL, MANAGED_BY};
20use super::types::{GatewaySpecAddress, GATEWAY_GROUP};
21
22/// How Gateways' Services are made.
23#[derive(Clone, Debug)]
24pub struct ServiceSettings {
25    /// Labels of the pods that serve the traffic (the shared Deployment's).
26    pub pod_selector: BTreeMap<String, String>,
27    /// Their namespace. A Service only selects pods in its own namespace, so
28    /// each Gateway's Service has no selector: rws writes its EndpointSlice.
29    pub pod_namespace: String,
30    /// Provisioned mode: each Gateway's own pods ([`super::provision`])
31    /// instead of `pod_selector`.
32    pub provisioned: bool,
33    /// Container ports to allocate from (inclusive).
34    pub port_range: (u16, u16),
35    /// `LoadBalancer` (default), `NodePort` or `ClusterIP`.
36    pub service_type: String,
37}
38
39impl Default for ServiceSettings {
40    fn default() -> Self {
41        ServiceSettings {
42            pod_selector: BTreeMap::from([("app.kubernetes.io/name".to_string(), "rws".to_string())]),
43            pod_namespace: "default".to_string(),
44            provisioned: false,
45            port_range: (10000, 10999),
46            service_type: "LoadBalancer".to_string(),
47        }
48    }
49}
50
51/// The transport a listener protocol needs.
52pub fn transport(protocol: &str) -> &'static str {
53    if protocol == "UDP" { "UDP" } else { "TCP" }
54}
55
56/// The Service's name for a Gateway: `rws-<gateway>`, shortened with a hash
57/// past the 63 characters a Service name may have.
58pub fn service_name(gateway: &str) -> String {
59    let name = format!("rws-{gateway}");
60    if name.len() <= 63 {
61        return name;
62    }
63    let hash: String = crate::core::sha256::sha256(gateway.as_bytes())[..4].iter().map(|b| format!("{b:02x}")).collect();
64    let head: String = name.chars().take(54).collect();
65    format!("{}-{hash}", head.trim_end_matches(['-', '.']))
66}
67
68fn is_ours(service: &Service) -> bool {
69    service.labels().get("app.kubernetes.io/managed-by").map(String::as_str) == Some(MANAGED_BY)
70}
71
72/// The Service rws manages for this Gateway, as it is now.
73pub fn current<'a>(snapshot: &'a Snapshot, gateway: &Gateway) -> Option<&'a Service> {
74    let namespace = gateway.namespace().unwrap_or_default();
75    let name = service_name(&gateway.name_any());
76    snapshot.services.iter().find(|s| s.namespace().as_deref() == Some(namespace.as_str()) && s.name_any() == name && is_ours(s))
77}
78
79/// (listener port, transport) → container port, as the Gateway's current
80/// Service maps it.
81pub fn container_ports(service: Option<&Service>) -> BTreeMap<(i32, String), u16> {
82    let mut ports = BTreeMap::new();
83    for port in service.and_then(|s| s.spec.as_ref()).and_then(|s| s.ports.as_ref()).into_iter().flatten() {
84        if let Some(IntOrString::Int(target)) = port.target_port {
85            let transport = port.protocol.clone().unwrap_or_else(|| "TCP".to_string());
86            ports.insert((port.port, transport), target as u16);
87        }
88    }
89    ports
90}
91
92/// Address types rws can ask a load balancer for: `IPAddress` (the default).
93pub fn address_supported(address: &GatewaySpecAddress) -> bool {
94    address.address_type.as_deref().unwrap_or("IPAddress") == "IPAddress"
95}
96
97/// The static addresses a Gateway asks for (`spec.addresses` with a value;
98/// one with only a type asks for any address).
99pub fn requested_addresses(gateway: &Gateway) -> Vec<&str> {
100    gateway
101        .spec
102        .addresses
103        .iter()
104        .filter(|a| address_supported(a) && !a.value.is_empty() && matches!(a.routability.as_deref().unwrap_or(""), "" | super::addresses::SENTINEL))
105        .map(|a| a.value.as_str())
106        .collect()
107}
108
109/// The Service every one of `gateways` should have, allocating container
110/// ports for new listener ports (keeping every existing allocation).
111pub fn desired(snapshot: &Snapshot, gateways: &[&Gateway], settings: &ServiceSettings) -> BTreeMap<(String, String), Service> {
112    // Container ports in use by any of our Services: the pods are shared.
113    let mut used: BTreeSet<u16> = snapshot.services.iter().filter(|s| is_ours(s)).flat_map(|s| container_ports(Some(s)).into_values()).collect();
114    let mut out = BTreeMap::new();
115    for gateway in gateways {
116        let namespace = gateway.namespace().unwrap_or_default();
117        let existing = container_ports(current(snapshot, gateway));
118        let mut wanted: BTreeSet<(i32, String)> = BTreeSet::new();
119        for listener in &gateway.spec.listeners {
120            wanted.insert((listener.port, transport(&listener.protocol).to_string()));
121        }
122        let mut ports = Vec::new();
123        for (port, transport) in wanted {
124            let container = match existing.get(&(port, transport.clone())) {
125                Some(container) => *container,
126                None => match (settings.port_range.0..=settings.port_range.1).find(|p| !used.contains(p)) {
127                    Some(free) => {
128                        used.insert(free);
129                        free
130                    }
131                    None => continue, // range exhausted: the listener stays unprogrammed
132                },
133            };
134            ports.push(ServicePort {
135                name: Some(format!("{}-{port}", transport.to_ascii_lowercase())),
136                port,
137                protocol: Some(transport),
138                target_port: Some(IntOrString::Int(container as i32)),
139                ..Default::default()
140            });
141        }
142        let infrastructure = gateway.spec.infrastructure.clone().unwrap_or_default();
143        let mut labels = infrastructure.labels.clone();
144        labels.insert(GATEWAY_NAME_LABEL.to_string(), gateway.name_any());
145        labels.insert("app.kubernetes.io/managed-by".to_string(), MANAGED_BY.to_string());
146        let service = Service {
147            metadata: ObjectMeta {
148                name: Some(service_name(&gateway.name_any())),
149                namespace: Some(namespace.clone()),
150                labels: Some(labels),
151                annotations: (!infrastructure.annotations.is_empty()).then(|| infrastructure.annotations.clone()),
152                owner_references: gateway.metadata.uid.clone().map(|uid| {
153                    vec![OwnerReference {
154                        api_version: format!("{GATEWAY_GROUP}/v1"),
155                        kind: "Gateway".to_string(),
156                        name: gateway.name_any(),
157                        uid,
158                        controller: Some(true),
159                        block_owner_deletion: Some(true),
160                    }]
161                }),
162                ..Default::default()
163            },
164            spec: Some(ServiceSpec {
165                type_: Some(settings.service_type.clone()),
166                // The first static address (a load balancer takes one per Service).
167                load_balancer_ip: requested_addresses(gateway).first().map(|a| a.to_string()),
168                // A `Cluster` address asked for by value (GEP-5093). A
169                // Service's ClusterIP can't change: an existing one is kept,
170                // and the entry is reported unassigned if it differs.
171                cluster_ip: match current(snapshot, gateway).and_then(|s| s.spec.as_ref()).and_then(|s| s.cluster_ip.clone()) {
172                    Some(existing) => Some(existing),
173                    None => super::addresses::requested_cluster_ip(&gateway.spec.addresses).map(str::to_string),
174                },
175                selector: None,
176                ports: Some(ports),
177                ..Default::default()
178            }),
179            status: None,
180        };
181        out.insert((namespace, service_name(&gateway.name_any())), service);
182    }
183    out
184}
185
186/// `endpointslice.kubernetes.io/managed-by` on the EndpointSlices rws writes
187/// (not the EndpointSlice controller's value, so it leaves them alone; a
188/// label value, so no `/`).
189pub const ENDPOINTS_MANAGED_BY: &str = "gateway-controller.rws8.tech";
190
191impl ServiceSettings {
192    /// The labels rws's serving pods are watched by.
193    pub fn pod_watch_labels(&self) -> BTreeMap<String, String> {
194        if self.provisioned {
195            BTreeMap::from([("app.kubernetes.io/name".to_string(), super::provision::APP_NAME.to_string())])
196        } else {
197            self.pod_selector.clone()
198        }
199    }
200}
201
202/// The serving pods: in the pod namespace, matching `selector`, with an IP,
203/// not being deleted; (IP, ready, node, name).
204pub fn serving_pods(snapshot: &Snapshot, settings: &ServiceSettings, selector: &BTreeMap<String, String>) -> Vec<(String, bool, Option<String>, String)> {
205    let mut pods: Vec<_> = snapshot
206        .pods
207        .iter()
208        .filter(|p| p.namespace().as_deref() == Some(settings.pod_namespace.as_str()))
209        .filter(|p| selector.iter().all(|(k, v)| p.labels().get(k) == Some(v)))
210        .filter(|p| p.metadata.deletion_timestamp.is_none())
211        .filter_map(|p| {
212            let status = p.status.as_ref()?;
213            let ip = status.pod_ip.clone().filter(|ip| !ip.is_empty())?;
214            let ready = status.conditions.iter().flatten().any(|c| c.type_ == "Ready" && c.status == "True");
215            Some((ip, ready, p.spec.as_ref().and_then(|s| s.node_name.clone()), p.name_any()))
216        })
217        .collect();
218    pods.sort();
219    pods
220}
221
222/// The EndpointSlice of each Gateway Service: the serving pods on the
223/// Service's container ports (one slice per address family in use).
224pub fn endpoint_slices(snapshot: &Snapshot, services: &BTreeMap<(String, String), Service>, settings: &ServiceSettings) -> BTreeMap<(String, String), EndpointSlice> {
225    let shared = serving_pods(snapshot, settings, &settings.pod_selector);
226    let mut out = BTreeMap::new();
227    for ((namespace, name), service) in services {
228        // Provisioned: the Gateway's own pods.
229        let own;
230        let pods = if settings.provisioned {
231            let gateway = service.labels().get(GATEWAY_NAME_LABEL).cloned().unwrap_or_default();
232            own = serving_pods(snapshot, settings, &super::provision::pod_labels(namespace, &gateway));
233            &own
234        } else {
235            &shared
236        };
237        let ports: Vec<EndpointPort> = service
238            .spec
239            .iter()
240            .flat_map(|s| s.ports.iter().flatten())
241            .filter_map(|p| match p.target_port {
242                Some(IntOrString::Int(target)) => Some(EndpointPort { name: p.name.clone(), port: Some(target), protocol: p.protocol.clone(), app_protocol: None }),
243                _ => None,
244            })
245            .collect();
246        for (family, v6) in [("IPv4", false), ("IPv6", true)] {
247            let endpoints: Vec<Endpoint> = pods
248                .iter()
249                .filter(|(ip, ..)| ip.contains(':') == v6)
250                .map(|(ip, ready, node, pod)| Endpoint {
251                    addresses: vec![ip.clone()],
252                    conditions: Some(EndpointConditions { ready: Some(*ready), serving: Some(*ready), terminating: Some(false) }),
253                    node_name: node.clone(),
254                    target_ref: Some(k8s_openapi::api::core::v1::ObjectReference {
255                        kind: Some("Pod".to_string()),
256                        name: Some(pod.clone()),
257                        namespace: Some(settings.pod_namespace.clone()),
258                        ..Default::default()
259                    }),
260                    ..Default::default()
261                })
262                .collect();
263            if endpoints.is_empty() && v6 {
264                continue;
265            }
266            let slice_name = format!("{name}-{}", family.to_ascii_lowercase());
267            let slice = EndpointSlice {
268                metadata: ObjectMeta {
269                    name: Some(slice_name.clone()),
270                    namespace: Some(namespace.clone()),
271                    labels: Some(BTreeMap::from([
272                        ("kubernetes.io/service-name".to_string(), name.clone()),
273                        ("endpointslice.kubernetes.io/managed-by".to_string(), ENDPOINTS_MANAGED_BY.to_string()),
274                        ("app.kubernetes.io/managed-by".to_string(), MANAGED_BY.to_string()),
275                    ])),
276                    owner_references: service.metadata.owner_references.clone(),
277                    ..Default::default()
278                },
279                address_type: family.to_string(),
280                endpoints,
281                ports: Some(ports.clone()),
282            };
283            out.insert((namespace.clone(), slice_name), slice);
284        }
285    }
286    out
287}
288
289/// Whether an EndpointSlice already has what `desired` asks for.
290pub fn same_slice(current: &EndpointSlice, desired: &EndpointSlice) -> bool {
291    let endpoints = |s: &EndpointSlice| {
292        let mut e: Vec<_> = s.endpoints.iter().map(|e| (e.addresses.clone(), e.conditions.as_ref().and_then(|c| c.ready))).collect();
293        e.sort();
294        e
295    };
296    let ports = |s: &EndpointSlice| {
297        let mut p: Vec<_> = s.ports.iter().flatten().map(|p| (p.name.clone(), p.port, p.protocol.clone())).collect();
298        p.sort();
299        p
300    };
301    current.address_type == desired.address_type && endpoints(current) == endpoints(desired) && ports(current) == ports(desired)
302}
303
304/// Whether `current` already is what `desired` asks for (the fields rws
305/// sets; the API server fills in others, like `clusterIP` and node ports).
306pub fn same(current: &Service, desired: &Service) -> bool {
307    let labels_ok = desired.metadata.labels.iter().flatten().all(|(k, v)| current.labels().get(k) == Some(v));
308    let annotations_ok = desired.metadata.annotations.iter().flatten().all(|(k, v)| current.annotations().get(k) == Some(v));
309    let (Some(c), Some(d)) = (current.spec.as_ref(), desired.spec.as_ref()) else { return false };
310    let ports = |s: &ServiceSpec| -> Vec<(i32, Option<String>, Option<IntOrString>)> {
311        let mut p: Vec<_> = s.ports.iter().flatten().map(|p| (p.port, p.protocol.clone(), p.target_port.clone())).collect();
312        p.sort_by(|a, b| (a.0, &a.1).cmp(&(b.0, &b.1)));
313        p
314    };
315    labels_ok && annotations_ok && c.type_ == d.type_ && c.selector == d.selector && c.load_balancer_ip == d.load_balancer_ip && ports(c) == ports(d)
316}
317
318#[cfg(test)]
319mod tests;
rws_gateway_api/gateway_api/services.rs

rws_gateway_api/gateway_api/
services.rs
