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.spec.addresses.iter().filter(|a| address_supported(a) && !a.value.is_empty()).map(|a| a.value.as_str()).collect()
101}
102
103/// The Service every one of `gateways` should have, allocating container
104/// ports for new listener ports (keeping every existing allocation).
105pub fn desired(snapshot: &Snapshot, gateways: &[&Gateway], settings: &ServiceSettings) -> BTreeMap<(String, String), Service> {
106    // Container ports in use by any of our Services: the pods are shared.
107    let mut used: BTreeSet<u16> = snapshot.services.iter().filter(|s| is_ours(s)).flat_map(|s| container_ports(Some(s)).into_values()).collect();
108    let mut out = BTreeMap::new();
109    for gateway in gateways {
110        let namespace = gateway.namespace().unwrap_or_default();
111        let existing = container_ports(current(snapshot, gateway));
112        let mut wanted: BTreeSet<(i32, String)> = BTreeSet::new();
113        for listener in &gateway.spec.listeners {
114            wanted.insert((listener.port, transport(&listener.protocol).to_string()));
115        }
116        let mut ports = Vec::new();
117        for (port, transport) in wanted {
118            let container = match existing.get(&(port, transport.clone())) {
119                Some(container) => *container,
120                None => match (settings.port_range.0..=settings.port_range.1).find(|p| !used.contains(p)) {
121                    Some(free) => {
122                        used.insert(free);
123                        free
124                    }
125                    None => continue, // range exhausted: the listener stays unprogrammed
126                },
127            };
128            ports.push(ServicePort {
129                name: Some(format!("{}-{port}", transport.to_ascii_lowercase())),
130                port,
131                protocol: Some(transport),
132                target_port: Some(IntOrString::Int(container as i32)),
133                ..Default::default()
134            });
135        }
136        let infrastructure = gateway.spec.infrastructure.clone().unwrap_or_default();
137        let mut labels = infrastructure.labels.clone();
138        labels.insert(GATEWAY_NAME_LABEL.to_string(), gateway.name_any());
139        labels.insert("app.kubernetes.io/managed-by".to_string(), MANAGED_BY.to_string());
140        let service = Service {
141            metadata: ObjectMeta {
142                name: Some(service_name(&gateway.name_any())),
143                namespace: Some(namespace.clone()),
144                labels: Some(labels),
145                annotations: (!infrastructure.annotations.is_empty()).then(|| infrastructure.annotations.clone()),
146                owner_references: gateway.metadata.uid.clone().map(|uid| {
147                    vec![OwnerReference {
148                        api_version: format!("{GATEWAY_GROUP}/v1"),
149                        kind: "Gateway".to_string(),
150                        name: gateway.name_any(),
151                        uid,
152                        controller: Some(true),
153                        block_owner_deletion: Some(true),
154                    }]
155                }),
156                ..Default::default()
157            },
158            spec: Some(ServiceSpec {
159                type_: Some(settings.service_type.clone()),
160                // The first static address (a load balancer takes one per Service).
161                load_balancer_ip: requested_addresses(gateway).first().map(|a| a.to_string()),
162                selector: None,
163                ports: Some(ports),
164                ..Default::default()
165            }),
166            status: None,
167        };
168        out.insert((namespace, service_name(&gateway.name_any())), service);
169    }
170    out
171}
172
173/// `endpointslice.kubernetes.io/managed-by` on the EndpointSlices rws writes
174/// (not the EndpointSlice controller's value, so it leaves them alone; a
175/// label value, so no `/`).
176pub const ENDPOINTS_MANAGED_BY: &str = "gateway-controller.rws8.tech";
177
178impl ServiceSettings {
179    /// The labels rws's serving pods are watched by.
180    pub fn pod_watch_labels(&self) -> BTreeMap<String, String> {
181        if self.provisioned {
182            BTreeMap::from([("app.kubernetes.io/name".to_string(), super::provision::APP_NAME.to_string())])
183        } else {
184            self.pod_selector.clone()
185        }
186    }
187}
188
189/// The serving pods: in the pod namespace, matching `selector`, with an IP,
190/// not being deleted; (IP, ready, node, name).
191pub fn serving_pods(snapshot: &Snapshot, settings: &ServiceSettings, selector: &BTreeMap<String, String>) -> Vec<(String, bool, Option<String>, String)> {
192    let mut pods: Vec<_> = snapshot
193        .pods
194        .iter()
195        .filter(|p| p.namespace().as_deref() == Some(settings.pod_namespace.as_str()))
196        .filter(|p| selector.iter().all(|(k, v)| p.labels().get(k) == Some(v)))
197        .filter(|p| p.metadata.deletion_timestamp.is_none())
198        .filter_map(|p| {
199            let status = p.status.as_ref()?;
200            let ip = status.pod_ip.clone().filter(|ip| !ip.is_empty())?;
201            let ready = status.conditions.iter().flatten().any(|c| c.type_ == "Ready" && c.status == "True");
202            Some((ip, ready, p.spec.as_ref().and_then(|s| s.node_name.clone()), p.name_any()))
203        })
204        .collect();
205    pods.sort();
206    pods
207}
208
209/// The EndpointSlice of each Gateway Service: the serving pods on the
210/// Service's container ports (one slice per address family in use).
211pub fn endpoint_slices(snapshot: &Snapshot, services: &BTreeMap<(String, String), Service>, settings: &ServiceSettings) -> BTreeMap<(String, String), EndpointSlice> {
212    let shared = serving_pods(snapshot, settings, &settings.pod_selector);
213    let mut out = BTreeMap::new();
214    for ((namespace, name), service) in services {
215        // Provisioned: the Gateway's own pods.
216        let own;
217        let pods = if settings.provisioned {
218            let gateway = service.labels().get(GATEWAY_NAME_LABEL).cloned().unwrap_or_default();
219            own = serving_pods(snapshot, settings, &super::provision::pod_labels(namespace, &gateway));
220            &own
221        } else {
222            &shared
223        };
224        let ports: Vec<EndpointPort> = service
225            .spec
226            .iter()
227            .flat_map(|s| s.ports.iter().flatten())
228            .filter_map(|p| match p.target_port {
229                Some(IntOrString::Int(target)) => Some(EndpointPort { name: p.name.clone(), port: Some(target), protocol: p.protocol.clone(), app_protocol: None }),
230                _ => None,
231            })
232            .collect();
233        for (family, v6) in [("IPv4", false), ("IPv6", true)] {
234            let endpoints: Vec<Endpoint> = pods
235                .iter()
236                .filter(|(ip, ..)| ip.contains(':') == v6)
237                .map(|(ip, ready, node, pod)| Endpoint {
238                    addresses: vec![ip.clone()],
239                    conditions: Some(EndpointConditions { ready: Some(*ready), serving: Some(*ready), terminating: Some(false) }),
240                    node_name: node.clone(),
241                    target_ref: Some(k8s_openapi::api::core::v1::ObjectReference {
242                        kind: Some("Pod".to_string()),
243                        name: Some(pod.clone()),
244                        namespace: Some(settings.pod_namespace.clone()),
245                        ..Default::default()
246                    }),
247                    ..Default::default()
248                })
249                .collect();
250            if endpoints.is_empty() && v6 {
251                continue;
252            }
253            let slice_name = format!("{name}-{}", family.to_ascii_lowercase());
254            let slice = EndpointSlice {
255                metadata: ObjectMeta {
256                    name: Some(slice_name.clone()),
257                    namespace: Some(namespace.clone()),
258                    labels: Some(BTreeMap::from([
259                        ("kubernetes.io/service-name".to_string(), name.clone()),
260                        ("endpointslice.kubernetes.io/managed-by".to_string(), ENDPOINTS_MANAGED_BY.to_string()),
261                        ("app.kubernetes.io/managed-by".to_string(), MANAGED_BY.to_string()),
262                    ])),
263                    owner_references: service.metadata.owner_references.clone(),
264                    ..Default::default()
265                },
266                address_type: family.to_string(),
267                endpoints,
268                ports: Some(ports.clone()),
269            };
270            out.insert((namespace.clone(), slice_name), slice);
271        }
272    }
273    out
274}
275
276/// Whether an EndpointSlice already has what `desired` asks for.
277pub fn same_slice(current: &EndpointSlice, desired: &EndpointSlice) -> bool {
278    let endpoints = |s: &EndpointSlice| {
279        let mut e: Vec<_> = s.endpoints.iter().map(|e| (e.addresses.clone(), e.conditions.as_ref().and_then(|c| c.ready))).collect();
280        e.sort();
281        e
282    };
283    let ports = |s: &EndpointSlice| {
284        let mut p: Vec<_> = s.ports.iter().flatten().map(|p| (p.name.clone(), p.port, p.protocol.clone())).collect();
285        p.sort();
286        p
287    };
288    current.address_type == desired.address_type && endpoints(current) == endpoints(desired) && ports(current) == ports(desired)
289}
290
291/// Whether `current` already is what `desired` asks for (the fields rws
292/// sets; the API server fills in others, like `clusterIP` and node ports).
293pub fn same(current: &Service, desired: &Service) -> bool {
294    let labels_ok = desired.metadata.labels.iter().flatten().all(|(k, v)| current.labels().get(k) == Some(v));
295    let annotations_ok = desired.metadata.annotations.iter().flatten().all(|(k, v)| current.annotations().get(k) == Some(v));
296    let (Some(c), Some(d)) = (current.spec.as_ref(), desired.spec.as_ref()) else { return false };
297    let ports = |s: &ServiceSpec| -> Vec<(i32, Option<String>, Option<IntOrString>)> {
298        let mut p: Vec<_> = s.ports.iter().flatten().map(|p| (p.port, p.protocol.clone(), p.target_port.clone())).collect();
299        p.sort_by(|a, b| (a.0, &a.1).cmp(&(b.0, &b.1)));
300        p
301    };
302    labels_ok && annotations_ok && c.type_ == d.type_ && c.selector == d.selector && c.load_balancer_ip == d.load_balancer_ip && ports(c) == ports(d)
303}
304
305#[cfg(test)]
306mod tests;
rws_gateway_api/gateway_api/services.rs

rws_gateway_api/gateway_api/
services.rs
