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}
46
47/// The watched stores.
48#[derive(Clone)]
49pub struct Stores {
50    pub classes: Store<GatewayClass>,
51    pub gateways: Store<Gateway>,
52    pub http_routes: Store<HttpRoute>,
53    pub grpc_routes: Store<GrpcRoute>,
54    pub tls_routes: Store<TlsRoute>,
55    pub tcp_routes: Store<TcpRoute>,
56    pub udp_routes: Store<UdpRoute>,
57    pub listener_sets: Store<ListenerSet>,
58    pub backend_tls_policies: Store<BackendTlsPolicy>,
59    pub reference_grants: Store<ReferenceGrant>,
60    pub services: Store<Service>,
61    pub namespaces: Store<Namespace>,
62    pub secrets: Store<Secret>,
63    pub config_maps: Store<ConfigMap>,
64    pub endpoint_slices: Store<EndpointSlice>,
65    pub pods: Store<Pod>,
66    pub deployments: Store<k8s_openapi::api::apps::v1::Deployment>,
67}
68
69impl Stores {
70    pub fn snapshot(&self) -> Snapshot {
71        fn all<K: Clone + kube::Resource>(store: &Store<K>) -> Vec<K>
72        where
73            K::DynamicType: std::hash::Hash + Eq + Clone,
74        {
75            store.state().iter().map(|o| (**o).clone()).collect()
76        }
77        Snapshot {
78            classes: all(&self.classes),
79            gateways: all(&self.gateways),
80            http_routes: all(&self.http_routes),
81            grpc_routes: all(&self.grpc_routes),
82            tls_routes: all(&self.tls_routes),
83            tcp_routes: all(&self.tcp_routes),
84            udp_routes: all(&self.udp_routes),
85            listener_sets: all(&self.listener_sets),
86            backend_tls_policies: all(&self.backend_tls_policies),
87            reference_grants: all(&self.reference_grants),
88            services: all(&self.services),
89            namespaces: all(&self.namespaces),
90            secrets: all(&self.secrets),
91            config_maps: all(&self.config_maps),
92            endpoint_slices: all(&self.endpoint_slices),
93            pods: all(&self.pods),
94            deployments: all(&self.deployments),
95        }
96    }
97}
98
99/// Start a watcher for `api` feeding a new store; every change wakes `changed`.
100fn watch<K>(api: Api<K>, dyntype: K::DynamicType, changed: Arc<Notify>) -> Store<K>
101where
102    K: Resource + Clone + DeserializeOwned + std::fmt::Debug + Send + Sync + 'static,
103    K::DynamicType: std::hash::Hash + Eq + Clone + Send + Sync,
104{
105    watch_with(api, dyntype, watcher::Config::default(), changed)
106}
107
108fn watch_with<K>(api: Api<K>, dyntype: K::DynamicType, config: watcher::Config, changed: Arc<Notify>) -> Store<K>
109where
110    K: Resource + Clone + DeserializeOwned + std::fmt::Debug + Send + Sync + 'static,
111    K::DynamicType: std::hash::Hash + Eq + Clone + Send + Sync,
112{
113    let writer = Writer::new(dyntype);
114    let reader = writer.as_reader();
115    let stream = reflector::reflector(writer, watcher(api, config)).default_backoff().touched_objects();
116    tokio::spawn(async move {
117        let mut stream = std::pin::pin!(stream);
118        while let Some(event) = stream.next().await {
119            if let Err(e) = event {
120                eprintln!("[gateway-api] watch error: {e}");
121            }
122            changed.notify_one();
123        }
124    });
125    reader
126}
127
128/// Start watching everything (rws's own pods by `services`' namespace and
129/// selector); returns the stores and the change signal.
130pub fn start_watches(client: &Client, services: &super::services::ServiceSettings) -> (Stores, Arc<Notify>) {
131    let pod_labels = services.pod_watch_labels().iter().map(|(k, v)| format!("{k}={v}")).collect::<Vec<_>>().join(",");
132    let managed = format!("app.kubernetes.io/managed-by={}", super::translate::MANAGED_BY);
133    let changed = Arc::new(Notify::new());
134    let stores = Stores {
135        classes: watch(Api::all_with(client.clone(), &resources::gateway_classes()), resources::gateway_classes(), Arc::clone(&changed)),
136        gateways: watch(Api::all_with(client.clone(), &resources::gateways()), resources::gateways(), Arc::clone(&changed)),
137        http_routes: watch(Api::all_with(client.clone(), &resources::http_routes()), resources::http_routes(), Arc::clone(&changed)),
138        grpc_routes: watch(Api::all_with(client.clone(), &resources::grpc_routes()), resources::grpc_routes(), Arc::clone(&changed)),
139        tls_routes: watch(Api::all_with(client.clone(), &resources::tls_routes()), resources::tls_routes(), Arc::clone(&changed)),
140        tcp_routes: watch(Api::all_with(client.clone(), &resources::tcp_routes()), resources::tcp_routes(), Arc::clone(&changed)),
141        udp_routes: watch(Api::all_with(client.clone(), &resources::udp_routes()), resources::udp_routes(), Arc::clone(&changed)),
142        listener_sets: watch(Api::all_with(client.clone(), &resources::listener_sets()), resources::listener_sets(), Arc::clone(&changed)),
143        backend_tls_policies: watch(Api::all_with(client.clone(), &resources::backend_tls_policies()), resources::backend_tls_policies(), Arc::clone(&changed)),
144        reference_grants: watch(Api::all_with(client.clone(), &resources::reference_grants()), resources::reference_grants(), Arc::clone(&changed)),
145        services: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
146        namespaces: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
147        secrets: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
148        config_maps: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
149        endpoint_slices: watch(Api::all(client.clone()), (), Arc::clone(&changed)),
150        pods: watch_with(Api::namespaced(client.clone(), &services.pod_namespace), (), watcher::Config::default().labels(&pod_labels), Arc::clone(&changed)),
151        deployments: watch_with(Api::namespaced(client.clone(), &services.pod_namespace), (), watcher::Config::default().labels(&managed), Arc::clone(&changed)),
152    };
153    (stores, changed)
154}
155
156/// Reconcile on every change, when this replica becomes the leader, and
157/// every [`RESYNC`], forever.
158pub async fn run(client: Client, options: Options, became_leader: Arc<Notify>) {
159    let (stores, changed) = start_watches(&client, &options.settings.services);
160    loop {
161        tokio::select! {
162            _ = changed.notified() => {}
163            _ = became_leader.notified() => {}
164            _ = tokio::time::sleep(RESYNC) => {}
165        }
166        tokio::time::sleep(DEBOUNCE).await;
167        if let Err(e) = reconcile(&client, &stores, &options).await {
168            eprintln!("[gateway-api] reconcile: {e}");
169            // Try again soon even without a new event.
170            let changed = Arc::clone(&changed);
171            tokio::spawn(async move {
172                tokio::time::sleep(Duration::from_secs(1)).await;
173                changed.notify_one();
174            });
175        }
176    }
177}
178
179/// One pass: translate, hand the configuration over, write changed status.
180pub async fn reconcile(client: &Client, stores: &Stores, options: &Options) -> Result<(), String> {
181    let snapshot = stores.snapshot();
182    let now = Time(k8s_openapi::jiff::Timestamp::now());
183    let (config, desired) = translate(&snapshot, &options.settings, &now);
184    (options.on_config)(config);
185    if !options.leader.load(Ordering::SeqCst) {
186        return Ok(());
187    }
188    let mut failures = Vec::new();
189
190    // Each Gateway's Service, applied when it differs from what's there
191    // (rws owns these objects outright: server-side apply, forced).
192    for ((namespace, name), service) in &desired.services {
193        let current = snapshot.services.iter().find(|s| s.namespace().as_deref() == Some(namespace.as_str()) && s.name_any() == *name);
194        if current.is_some_and(|c| super::services::same(c, service)) {
195            continue;
196        }
197        let api: Api<Service> = Api::namespaced(client.clone(), namespace);
198        let mut applied = service.clone();
199        applied.metadata.managed_fields = None;
200        let body = serde_json::json!({"apiVersion": "v1", "kind": "Service", "metadata": applied.metadata, "spec": applied.spec});
201        if let Err(e) = api.patch(name, &PatchParams::apply(super::FIELD_MANAGER).force(), &Patch::Apply(&body)).await {
202            failures.push(format!("Service {namespace}/{name}: {e}"));
203        }
204    }
205
206    // Provisioned mode: each Gateway's Deployment; those of Gateways gone are deleted.
207    if options.settings.provision.is_some() {
208        for deployment in desired.deployments.values() {
209            let (namespace, name) = (deployment.namespace().unwrap_or_default(), deployment.name_any());
210            let current = snapshot.deployments.iter().find(|d| d.namespace().as_deref() == Some(namespace.as_str()) && d.name_any() == name);
211            if current.is_some_and(|c| super::provision::same(c, deployment)) {
212                continue;
213            }
214            let api: Api<k8s_openapi::api::apps::v1::Deployment> = Api::namespaced(client.clone(), &namespace);
215            let body = serde_json::json!({"apiVersion": "apps/v1", "kind": "Deployment", "metadata": deployment.metadata, "spec": deployment.spec});
216            if let Err(e) = api.patch(&name, &PatchParams::apply(super::FIELD_MANAGER).force(), &Patch::Apply(&body)).await {
217                failures.push(format!("Deployment {namespace}/{name}: {e}"));
218            }
219        }
220        for stale in super::provision::stale(&snapshot.deployments, &desired.deployments) {
221            let api: Api<k8s_openapi::api::apps::v1::Deployment> = Api::namespaced(client.clone(), &stale.namespace().unwrap_or_default());
222            if let Err(e) = api.delete(&stale.name_any(), &Default::default()).await {
223                failures.push(format!("deleting Deployment {}: {e}", stale.name_any()));
224            }
225        }
226    }
227
228    for ((namespace, name), slice) in &desired.endpoint_slices {
229        let current = snapshot.endpoint_slices.iter().find(|s| s.namespace().as_deref() == Some(namespace.as_str()) && s.name_any() == *name);
230        if current.is_some_and(|c| super::services::same_slice(c, slice)) {
231            continue;
232        }
233        let api: Api<EndpointSlice> = Api::namespaced(client.clone(), namespace);
234        let body = serde_json::json!({
235            "apiVersion": "discovery.k8s.io/v1", "kind": "EndpointSlice",
236            "metadata": slice.metadata, "addressType": slice.address_type, "endpoints": slice.endpoints, "ports": slice.ports,
237        });
238        if let Err(e) = api.patch(name, &PatchParams::apply(super::FIELD_MANAGER).force(), &Patch::Apply(&body)).await {
239            failures.push(format!("EndpointSlice {namespace}/{name}: {e}"));
240        }
241    }
242
243    for class in &snapshot.classes {
244        let Some(status) = desired.classes.get(&class.name_any()) else { continue };
245        if class.status.as_ref() == Some(status) {
246            continue;
247        }
248        let api: Api<GatewayClass> = Api::all_with(client.clone(), &resources::gateway_classes());
249        if let Err(e) = patch_status(&api, &class.name_any(), status, None).await {
250            failures.push(e);
251        }
252    }
253    for gateway in &snapshot.gateways {
254        let namespace = gateway.namespace().unwrap_or_default();
255        let Some(status) = desired.gateways.get(&(namespace.clone(), gateway.name_any())) else { continue };
256        if gateway.status.as_ref() == Some(status) {
257            continue;
258        }
259        let api: Api<Gateway> = Api::namespaced_with(client.clone(), &namespace, &resources::gateways());
260        if let Err(e) = patch_status(&api, &gateway.name_any(), status, None).await {
261            failures.push(e);
262        }
263    }
264    for set in &snapshot.listener_sets {
265        let namespace = set.namespace().unwrap_or_default();
266        let Some(status) = desired.listener_sets.get(&(namespace.clone(), set.name_any())) else { continue };
267        if set.status.as_ref() == Some(status) {
268            continue;
269        }
270        let api: Api<ListenerSet> = Api::namespaced_with(client.clone(), &namespace, &resources::listener_sets());
271        if let Err(e) = patch_status(&api, &set.name_any(), status, None).await {
272            failures.push(e);
273        }
274    }
275    let controller_name = options.settings.controller_name.as_str();
276    for policy in &snapshot.backend_tls_policies {
277        let namespace = policy.namespace().unwrap_or_default();
278        let Some(ours) = desired.backend_tls_policies.get(&(namespace.clone(), policy.name_any())) else { continue };
279        let current = policy.status.as_ref().map(|s| s.ancestors.clone()).unwrap_or_default();
280        let mut ancestors: Vec<super::types::PolicyAncestorStatus> = current.iter().filter(|a| a.controller_name != controller_name).cloned().collect();
281        ancestors.extend(ours.iter().cloned());
282        if ancestors == current {
283            continue;
284        }
285        let api: Api<BackendTlsPolicy> = Api::namespaced_with(client.clone(), &namespace, &resources::backend_tls_policies());
286        let status = super::types::PolicyStatus { ancestors };
287        if let Err(e) = patch_status(&api, &policy.name_any(), &status, policy.resource_version()).await {
288            failures.push(e);
289        }
290    }
291    failures.extend(write_route_status(client, &snapshot.http_routes, &desired.http_routes, &resources::http_routes(), controller_name).await);
292    failures.extend(write_route_status(client, &snapshot.grpc_routes, &desired.grpc_routes, &resources::grpc_routes(), controller_name).await);
293    failures.extend(write_route_status(client, &snapshot.tls_routes, &desired.tls_routes, &resources::tls_routes(), controller_name).await);
294    failures.extend(write_route_status(client, &snapshot.tcp_routes, &desired.tcp_routes, &resources::tcp_routes(), controller_name).await);
295    failures.extend(write_route_status(client, &snapshot.udp_routes, &desired.udp_routes, &resources::udp_routes(), controller_name).await);
296    if failures.is_empty() {
297        Ok(())
298    } else {
299        Err(failures.join("; "))
300    }
301}
302
303/// Write our part of `status.parents` on every route of one kind where it
304/// changed; returns the failures.
305async fn write_route_status<S>(
306    client: &Client,
307    routes: &[kube::core::Object<S, super::types::RouteStatus>],
308    desired: &std::collections::BTreeMap<(String, String), Vec<RouteParentStatus>>,
309    resource: &kube::core::ApiResource,
310    controller_name: &str,
311) -> Vec<String>
312where
313    S: Clone + DeserializeOwned + std::fmt::Debug,
314{
315    let mut failures = Vec::new();
316    for route in routes {
317        let namespace = route.namespace().unwrap_or_default();
318        let Some(ours) = desired.get(&(namespace.clone(), route.name_any())) else { continue };
319        let current = route.status.as_ref().map(|s| s.parents.clone()).unwrap_or_default();
320        let parents = merge_parents(&current, ours, controller_name);
321        if parents == current {
322            continue;
323        }
324        let api: Api<kube::core::Object<S, super::types::RouteStatus>> = Api::namespaced_with(client.clone(), &namespace, resource);
325        let status = super::types::RouteStatus { parents };
326        if let Err(e) = patch_status(&api, &route.name_any(), &status, route.resource_version()).await {
327            failures.push(e);
328        }
329    }
330    failures
331}
332
333/// Other controllers' entries as they are, then ours.
334pub fn merge_parents(current: &[RouteParentStatus], ours: &[RouteParentStatus], controller_name: &str) -> Vec<RouteParentStatus> {
335    let mut parents: Vec<RouteParentStatus> = current.iter().filter(|p| p.controller_name != controller_name).cloned().collect();
336    parents.extend(ours.iter().cloned());
337    parents
338}
339
340async fn patch_status<K, S>(api: &Api<K>, name: &str, status: &S, resource_version: Option<String>) -> Result<(), String>
341where
342    K: Resource + Clone + DeserializeOwned + std::fmt::Debug,
343    S: serde::Serialize,
344{
345    let mut patch = serde_json::json!({ "status": status });
346    if let Some(version) = resource_version {
347        patch["metadata"] = serde_json::json!({ "resourceVersion": version });
348    }
349    api.patch_status(name, &PatchParams::apply(super::FIELD_MANAGER), &Patch::Merge(&patch))
350        .await
351        .map(|_| ())
352        .map_err(|e| format!("status of {name}: {e}"))
353}
354
355#[cfg(test)]
356mod tests;
rws_gateway_api/gateway_api/controller.rs

rws_gateway_api/gateway_api/
controller.rs
