rws_gateway_api/gateway_api/
controller.rs

1//! The controller loop: one watcher per kind feeding a shared store, a
2//! debounced reconcile that runs [`translate`] over a snapshot of those
3//! stores, hands the result to the data plane, and writes the status that
4//! changed.
5//!
6//! Status is written with a JSON merge patch on `/status`. For routes,
7//! `status.parents` is one list shared with other controllers: rws replaces
8//! only its own entries (by `controllerName`) and sends the object's
9//! `resourceVersion`, so a concurrent write by someone else makes the API
10//! server refuse ours (`409`) rather than lose theirs; the next reconcile
11//! retries from fresh state.
12
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::sync::Arc;
15use std::time::Duration;
16
17use futures_util::StreamExt;
18use k8s_openapi::api::core::v1::{ConfigMap, Namespace, Pod, Secret, Service};
19use k8s_openapi::api::discovery::v1::EndpointSlice;
20use k8s_openapi::apimachinery::pkg::apis::meta::v1::Time;
21use kube::api::{Api, Patch, PatchParams};
22use kube::runtime::reflector::{self, store::Writer, Store};
23use kube::runtime::{watcher, WatchStreamExt};
24use kube::{Client, Resource, ResourceExt};
25use serde::de::DeserializeOwned;
26use tokio::sync::Notify;
27
28use super::resources::{self, BackendTlsPolicy, Gateway, GatewayClass, GrpcRoute, HttpRoute, ListenerSet, ReferenceGrant, TcpRoute, TlsRoute, UdpRoute};
29use super::translate::{translate, GatewayConfig, Settings, Snapshot};
30use super::types::RouteParentStatus;
31
32/// How long changes are gathered before a reconcile.
33pub const DEBOUNCE: Duration = Duration::from_millis(100);
34/// A reconcile at least this often, changes or not (status drift, a lost event).
35pub const RESYNC: Duration = Duration::from_secs(30);
36
37/// What the controller needs besides the cluster.
38pub struct Options {
39    pub settings: Settings,
40    /// Whether this replica holds the Lease: the leader writes status and
41    /// Services, every replica serves ([`super::lease`]).
42    pub leader: Arc<AtomicBool>,
43    /// Called with every new configuration.
44    pub on_config: Arc<dyn Fn(GatewayConfig) + Send + Sync>,
45    /// The mesh configuration endpoint, when the mesh is on: every replica
46    /// publishes its snapshot's mesh part ([`super::mesh`]).
47    pub mesh: Option<super::mesh::config::Publisher>,
48}
49
50/// The watched stores.
51#[derive(Clone)]
52pub struct Stores {
53    pub classes: Store<GatewayClass>,
54    pub gateways: Store<Gateway>,
55    pub http_routes: Store<HttpRoute>,
56    pub grpc_routes: Store<GrpcRoute>,
57    pub tls_routes: Store<TlsRoute>,
58    pub tcp_routes: Store<TcpRoute>,
59    pub udp_routes: Store<UdpRoute>,
60    pub listener_sets: Store<ListenerSet>,
61    pub backend_tls_policies: Store<BackendTlsPolicy>,
62    pub reference_grants: Store<ReferenceGrant>,
63    pub services: Store<Service>,
64    pub namespaces: Store<Namespace>,
65    pub secrets: Store<Secret>,
66    pub config_maps: Store<ConfigMap>,
67    pub endpoint_slices: Store<EndpointSlice>,
68    pub pods: Store<Pod>,
69    pub deployments: Store<k8s_openapi::api::apps::v1::Deployment>,
70    /// ClusterTrustBundles, when the cluster serves them (GEP-5075).
71    pub cluster_trust_bundles: Option<Store<kube::core::DynamicObject>>,
72}
73
74impl Stores {
75    pub fn snapshot(&self) -> Snapshot {
76        fn all<K: Clone + kube::Resource>(store: &Store<K>) -> Vec<K>
77        where
78            K::DynamicType: std::hash::Hash + Eq + Clone,
79        {
80            store.state().iter().map(|o| (**o).clone()).collect()
81        }
82        Snapshot {
83            classes: all(&self.classes),
84            gateways: all(&self.gateways),
85            http_routes: all(&self.http_routes),
86            grpc_routes: all(&self.grpc_routes),
87            tls_routes: all(&self.tls_routes),
88            tcp_routes: all(&self.tcp_routes),
89            udp_routes: all(&self.udp_routes),
90            listener_sets: all(&self.listener_sets),
91            backend_tls_policies: all(&self.backend_tls_policies),
92            reference_grants: all(&self.reference_grants),
93            services: all(&self.services),
94            namespaces: all(&self.namespaces),
95            secrets: all(&self.secrets),
96            config_maps: all(&self.config_maps),
97            endpoint_slices: all(&self.endpoint_slices),
98            pods: all(&self.pods),
99            deployments: all(&self.deployments),
100            cluster_trust_bundles_served: self.cluster_trust_bundles.is_some(),
101            cluster_trust_bundles: self
102                .cluster_trust_bundles
103                .as_ref()
104                .map(|store| {
105                    store
106                        .state()
107                        .iter()
108                        .map(|o| (o.metadata.name.clone().unwrap_or_default(), o.data["spec"]["trustBundle"].as_str().unwrap_or_default().to_string()))
109                        .collect()
110                })
111                .unwrap_or_default(),
112        }
113    }
114}
115
116/// Start a watcher for `api` feeding a new store; every change wakes `changed`.
117fn watch<K>(api: Api<K>, dyntype: K::DynamicType, changed: Arc<Notify>) -> Store<K>
118where
119    K: Resource + Clone + DeserializeOwned + std::fmt::Debug + Send + Sync + 'static,
120    K::DynamicType: std::hash::Hash + Eq + Clone + Send + Sync,
121{
122    watch_with(api, dyntype, watcher::Config::default(), changed)
123}
124
125fn watch_with<K>(api: Api<K>, dyntype: K::DynamicType, config: watcher::Config, changed: Arc<Notify>) -> Store<K>
126where
127    K: Resource + Clone + DeserializeOwned + std::fmt::Debug + Send + Sync + 'static,
128    K::DynamicType: std::hash::Hash + Eq + Clone + Send + Sync,
129{
130    let writer = Writer::new(dyntype);
131    let reader = writer.as_reader();
132    let stream = reflector::reflector(writer, watcher(api, config)).default_backoff().touched_objects();
133    tokio::spawn(async move {
134        let mut stream = std::pin::pin!(stream);
135        while let Some(event) = stream.next().await {
136            if let Err(e) = event {
137                eprintln!("[gateway-api] watch error: {e}");
138            }
139            changed.notify_one();
140        }
141    });
142    reader
143}
144
145/// Start watching everything (rws's own pods by `services`' namespace and
146/// selector); returns the stores and the change signal.
147pub fn start_watches(client: &Client, services: &super::services::ServiceSettings, trust_bundles: Option<kube::core::ApiResource>) -> (Stores, Arc<Notify>) {
148    let pod_labels = services.pod_watch_labels().iter().map(|(k, v)| format!("{k}={v}")).collect::<Vec<_>>().join(",");
149    let managed = format!("app.kubernetes.io/managed-by={}", super::translate::MANAGED_BY);
150    let changed = Arc::new(Notify::new());
151    let stores = Stores {
152        classes: watch(Api::all_with(client.clone(), &resources::gateway_classes()), resources::gateway_classes(), Arc::clone(&changed)),
153        gateways: watch(Api::all_with(client.clone(), &resources::gateways()), resources::gateways(), Arc::clone(&changed)),
154        http_routes: watch(Api::all_with(client.clone(), &resources::http_routes()), resources::http_routes(), Arc::clone(&changed)),
155        grpc_routes: watch(Api::all_with(client.clone(), &resources::grpc_routes()), resources::grpc_routes(), Arc::clone(&changed)),
156        tls_routes: watch(Api::all_with(client.clone(), &resources::tls_routes()), resources::tls_routes(), Arc::clone(&changed)),
157        tcp_routes: watch(Api::all_with(client.clone(), &resources::tcp_routes()), resources::tcp_routes(), Arc::clone(&changed)),
158        udp_routes: watch(Api::all_with(client.clone(), &resources::udp_routes()), resources::udp_routes(), Arc::clone(&changed)),
159        listener_sets: watch(Api::all_with(client.clone(), &resources::listener_sets()), resources::listener_sets(), Arc::clone(&changed)),
160        backend_tls_policies: watch(Api::all_with(client.clone(), &resources::backend_tls_policies()), resources::backend_tls_policies(), Arc::clone(&changed)),
161        reference_grants: watch(Api::all_with(client.clone(), &resources::reference_grants()), resources::reference_grants(), Arc::clone(&changed)),
162        services: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
163        namespaces: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
164        secrets: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
165        config_maps: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
166        endpoint_slices: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
167        pods: watch_with(Api::namespaced(client.clone(), &services.pod_namespace), (), watcher::Config::default().labels(&pod_labels), Arc::clone(&changed)),
168        deployments: watch_with(Api::namespaced(client.clone(), &services.pod_namespace), (), watcher::Config::default().labels(&managed), Arc::clone(&changed)),
169        cluster_trust_bundles: trust_bundles.map(|resource| watch(Api::all_with(client.clone(), &resource), resource, Arc::clone(&changed))),
170    };
171    (stores, changed)
172}
173
174/// Where the cluster serves ClusterTrustBundles (`certificates.k8s.io` v1,
175/// else v1beta1), if it does: they're watched, and the feature advertised,
176/// only then (GEP-5075).
177/// Only a `404` means a version isn't served; any other error (RBAC not in
178/// effect yet right after an install, the API server busy) is retried for
179/// a minute, so a transient refusal doesn't hide the API for good.
180pub async fn served_cluster_trust_bundles(client: &Client) -> Option<kube::core::ApiResource> {
181    'versions: for version in ["v1", "v1beta1"] {
182        let gvk = kube::core::GroupVersionKind::gvk(super::translate::CLUSTER_TRUST_BUNDLE_GROUP, version, "ClusterTrustBundle");
183        let resource = kube::core::ApiResource::from_gvk_with_plural(&gvk, "clustertrustbundles");
184        let api: Api<kube::core::DynamicObject> = Api::all_with(client.clone(), &resource);
185        for _ in 0..30 {
186            match api.list(&kube::api::ListParams::default().limit(1)).await {
187                Ok(_) => return Some(resource),
188                Err(e) if not_served(&e) => continue 'versions,
189                Err(e) => {
190                    eprintln!("[gateway-api] ClusterTrustBundles {version}: {e}; retrying");
191                    tokio::time::sleep(Duration::from_secs(2)).await;
192                }
193            }
194        }
195    }
196    None
197}
198
199/// Whether an API error says the resource isn't served at all.
200fn not_served(error: &kube::Error) -> bool {
201    match error {
202        kube::Error::Api(status) => status.code == 404,
203        _ => false,
204    }
205}
206
207/// Reconcile on every change, when this replica becomes the leader, and
208/// every [`RESYNC`], forever.
209pub async fn run(client: Client, options: Options, became_leader: Arc<Notify>) {
210    let trust_bundles = served_cluster_trust_bundles(&client).await;
211    let (stores, changed) = start_watches(&client, &options.settings.services, trust_bundles);
212    loop {
213        tokio::select! {
214            _ = changed.notified() => {}
215            _ = became_leader.notified() => {}
216            _ = tokio::time::sleep(RESYNC) => {}
217        }
218        tokio::time::sleep(DEBOUNCE).await;
219        if let Err(e) = reconcile(&client, &stores, &options).await {
220            eprintln!("[gateway-api] reconcile: {e}");
221            // Try again soon even without a new event.
222            let changed = Arc::clone(&changed);
223            tokio::spawn(async move {
224                tokio::time::sleep(Duration::from_secs(1)).await;
225                changed.notify_one();
226            });
227        }
228    }
229}
230
231/// One pass: translate, hand the configuration over, write changed status.
232pub async fn reconcile(client: &Client, stores: &Stores, options: &Options) -> Result<(), String> {
233    let snapshot = stores.snapshot();
234    let now = Time(k8s_openapi::jiff::Timestamp::now());
235    let (config, desired) = translate(&snapshot, &options.settings, &now);
236    (options.on_config)(config);
237    if let Some(publisher) = &options.mesh {
238        publisher.publish(&super::mesh::snapshot(&snapshot));
239    }
240    if !options.leader.load(Ordering::SeqCst) {
241        return Ok(());
242    }
243    let mut failures = Vec::new();
244
245    // Each Gateway's Service, applied when it differs from what's there
246    // (rws owns these objects outright: server-side apply, forced).
247    for ((namespace, name), service) in &desired.services {
248        let current = snapshot.services.iter().find(|s| s.namespace().as_deref() == Some(namespace.as_str()) && s.name_any() == *name);
249        if current.is_some_and(|c| super::services::same(c, service)) {
250            continue;
251        }
252        let api: Api<Service> = Api::namespaced(client.clone(), namespace);
253        let mut applied = service.clone();
254        applied.metadata.managed_fields = None;
255        let body = serde_json::json!({"apiVersion": "v1", "kind": "Service", "metadata": applied.metadata, "spec": applied.spec});
256        if let Err(e) = api.patch(name, &PatchParams::apply(super::FIELD_MANAGER).force(), &Patch::Apply(&body)).await {
257            failures.push(format!("Service {namespace}/{name}: {e}"));
258        }
259    }
260
261    // Provisioned mode: each Gateway's Deployment; those of Gateways gone are deleted.
262    if options.settings.provision.is_some() {
263        for deployment in desired.deployments.values() {
264            let (namespace, name) = (deployment.namespace().unwrap_or_default(), deployment.name_any());
265            let current = snapshot.deployments.iter().find(|d| d.namespace().as_deref() == Some(namespace.as_str()) && d.name_any() == name);
266            if current.is_some_and(|c| super::provision::same(c, deployment)) {
267                continue;
268            }
269            let api: Api<k8s_openapi::api::apps::v1::Deployment> = Api::namespaced(client.clone(), &namespace);
270            let body = serde_json::json!({"apiVersion": "apps/v1", "kind": "Deployment", "metadata": deployment.metadata, "spec": deployment.spec});
271            if let Err(e) = api.patch(&name, &PatchParams::apply(super::FIELD_MANAGER).force(), &Patch::Apply(&body)).await {
272                failures.push(format!("Deployment {namespace}/{name}: {e}"));
273            }
274        }
275        for stale in super::provision::stale(&snapshot.deployments, &desired.deployments) {
276            let api: Api<k8s_openapi::api::apps::v1::Deployment> = Api::namespaced(client.clone(), &stale.namespace().unwrap_or_default());
277            if let Err(e) = api.delete(&stale.name_any(), &Default::default()).await {
278                failures.push(format!("deleting Deployment {}: {e}", stale.name_any()));
279            }
280        }
281    }
282
283    for ((namespace, name), slice) in &desired.endpoint_slices {
284        let current = snapshot.endpoint_slices.iter().find(|s| s.namespace().as_deref() == Some(namespace.as_str()) && s.name_any() == *name);
285        if current.is_some_and(|c| super::services::same_slice(c, slice)) {
286            continue;
287        }
288        let api: Api<EndpointSlice> = Api::namespaced(client.clone(), namespace);
289        let body = serde_json::json!({
290            "apiVersion": "discovery.k8s.io/v1", "kind": "EndpointSlice",
291            "metadata": slice.metadata, "addressType": slice.address_type, "endpoints": slice.endpoints, "ports": slice.ports,
292        });
293        if let Err(e) = api.patch(name, &PatchParams::apply(super::FIELD_MANAGER).force(), &Patch::Apply(&body)).await {
294            failures.push(format!("EndpointSlice {namespace}/{name}: {e}"));
295        }
296    }
297
298    for class in &snapshot.classes {
299        let Some(status) = desired.classes.get(&class.name_any()) else { continue };
300        if class.status.as_ref() == Some(status) {
301            continue;
302        }
303        let api: Api<GatewayClass> = Api::all_with(client.clone(), &resources::gateway_classes());
304        if let Err(e) = patch_status(&api, &class.name_any(), status, None).await {
305            failures.push(e);
306        }
307    }
308    for gateway in &snapshot.gateways {
309        let namespace = gateway.namespace().unwrap_or_default();
310        let Some(status) = desired.gateways.get(&(namespace.clone(), gateway.name_any())) else { continue };
311        if gateway.status.as_ref() == Some(status) {
312            continue;
313        }
314        let api: Api<Gateway> = Api::namespaced_with(client.clone(), &namespace, &resources::gateways());
315        // With the resourceVersion: a status written from a stale cache would
316        // drop conditions another controller added since (409, retried).
317        if let Err(e) = patch_status(&api, &gateway.name_any(), status, gateway.resource_version()).await {
318            failures.push(e);
319        }
320    }
321    for set in &snapshot.listener_sets {
322        let namespace = set.namespace().unwrap_or_default();
323        let Some(status) = desired.listener_sets.get(&(namespace.clone(), set.name_any())) else { continue };
324        if set.status.as_ref() == Some(status) {
325            continue;
326        }
327        let api: Api<ListenerSet> = Api::namespaced_with(client.clone(), &namespace, &resources::listener_sets());
328        if let Err(e) = patch_status(&api, &set.name_any(), status, set.resource_version()).await {
329            failures.push(e);
330        }
331    }
332    let controller_name = options.settings.controller_name.as_str();
333    for policy in &snapshot.backend_tls_policies {
334        let namespace = policy.namespace().unwrap_or_default();
335        let Some(ours) = desired.backend_tls_policies.get(&(namespace.clone(), policy.name_any())) else { continue };
336        let current = policy.status.as_ref().map(|s| s.ancestors.clone()).unwrap_or_default();
337        let mut ancestors: Vec<super::types::PolicyAncestorStatus> = current.iter().filter(|a| a.controller_name != controller_name).cloned().collect();
338        ancestors.extend(ours.iter().cloned());
339        if ancestors == current {
340            continue;
341        }
342        let api: Api<BackendTlsPolicy> = Api::namespaced_with(client.clone(), &namespace, &resources::backend_tls_policies());
343        let status = super::types::PolicyStatus { ancestors };
344        if let Err(e) = patch_status(&api, &policy.name_any(), &status, policy.resource_version()).await {
345            failures.push(e);
346        }
347    }
348    failures.extend(write_route_status(client, &snapshot.http_routes, &desired.http_routes, &resources::http_routes(), controller_name).await);
349    failures.extend(write_route_status(client, &snapshot.grpc_routes, &desired.grpc_routes, &resources::grpc_routes(), controller_name).await);
350    failures.extend(write_route_status(client, &snapshot.tls_routes, &desired.tls_routes, &resources::tls_routes(), controller_name).await);
351    failures.extend(write_route_status(client, &snapshot.tcp_routes, &desired.tcp_routes, &resources::tcp_routes(), controller_name).await);
352    failures.extend(write_route_status(client, &snapshot.udp_routes, &desired.udp_routes, &resources::udp_routes(), controller_name).await);
353    if failures.is_empty() {
354        Ok(())
355    } else {
356        Err(failures.join("; "))
357    }
358}
359
360/// Write our part of `status.parents` on every route of one kind where it
361/// changed; returns the failures.
362async fn write_route_status<S>(
363    client: &Client,
364    routes: &[kube::core::Object<S, super::types::RouteStatus>],
365    desired: &std::collections::BTreeMap<(String, String), Vec<RouteParentStatus>>,
366    resource: &kube::core::ApiResource,
367    controller_name: &str,
368) -> Vec<String>
369where
370    S: Clone + DeserializeOwned + std::fmt::Debug,
371{
372    let mut failures = Vec::new();
373    for route in routes {
374        let namespace = route.namespace().unwrap_or_default();
375        let Some(ours) = desired.get(&(namespace.clone(), route.name_any())) else { continue };
376        let current = route.status.as_ref().map(|s| s.parents.clone()).unwrap_or_default();
377        let parents = merge_parents(&current, ours, controller_name);
378        if parents == current {
379            continue;
380        }
381        let api: Api<kube::core::Object<S, super::types::RouteStatus>> = Api::namespaced_with(client.clone(), &namespace, resource);
382        let status = super::types::RouteStatus { parents };
383        if let Err(e) = patch_status(&api, &route.name_any(), &status, route.resource_version()).await {
384            failures.push(e);
385        }
386    }
387    failures
388}
389
390/// Other controllers' entries as they are, then ours.
391pub fn merge_parents(current: &[RouteParentStatus], ours: &[RouteParentStatus], controller_name: &str) -> Vec<RouteParentStatus> {
392    let mut parents: Vec<RouteParentStatus> = current.iter().filter(|p| p.controller_name != controller_name).cloned().collect();
393    parents.extend(ours.iter().cloned());
394    parents
395}
396
397async fn patch_status<K, S>(api: &Api<K>, name: &str, status: &S, resource_version: Option<String>) -> Result<(), String>
398where
399    K: Resource + Clone + DeserializeOwned + std::fmt::Debug,
400    S: serde::Serialize,
401{
402    let mut patch = serde_json::json!({ "status": status });
403    if let Some(version) = resource_version {
404        patch["metadata"] = serde_json::json!({ "resourceVersion": version });
405    }
406    api.patch_status(name, &PatchParams::apply(super::FIELD_MANAGER), &Patch::Merge(&patch))
407        .await
408        .map(|_| ())
409        .map_err(|e| format!("status of {name}: {e}"))
410}
411
412#[cfg(test)]
413mod tests;
rws_gateway_api/gateway_api/controller.rs

rws_gateway_api/gateway_api/
controller.rs
