dwara_core/k8s_gateway/
controller.rs

1//! Kubernetes Gateway API controller (DW-064).
2//!
3//! Two layers:
4//!
5//! - [`Reconciler`]: the pure reconciliation core. It holds the current
6//!   set of watched resources (GatewayClass, Gateway, HTTPRoute, Ingress,
7//!   IngressClass, EndpointSlice) and produces a dwara `Gateway` config by
8//!   calling the translator ([`super::translate`]) and the Ingress
9//!   translator ([`super::ingress::translate_ingress`]). It also computes
10//!   status conditions (Accepted/Programmed for Gateway, Accepted/
11//!   ResolvedRefs for HTTPRoute). This layer is testable WITHOUT a cluster.
12//!
13//! - [`Controller`]: the kube-rs based wiring that sets up informers/
14//!   watchers for each resource type, feeds events into the `Reconciler`,
15//!   and publishes the resulting dwara config (file-write by default; the
16//!   admin API push path is documented as an option). Status updates use
17//!   kube-rs patch. This layer requires a running Kubernetes cluster.
18//!
19//! ## Feature gate
20//!
21//! None: the module compiles into every build (no `k8s` cargo
22//! feature).
23
24use std::collections::HashMap;
25use std::path::PathBuf;
26
27use serde::{Deserialize, Serialize};
28
29use crate::config::Endpoint as DwaraEndpoint;
30use crate::k8s_gateway::{
31    ingress::{Ingress, IngressClass},
32    Gateway, GatewayClass, HttpRoute, TranslationResult, CONTROLLER_NAME,
33};
34
35// ---------------------------------------------------------------------------
36// Status types
37// ---------------------------------------------------------------------------
38
39/// A status condition (subset of metav1.Condition).
40#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
41pub struct Condition {
42    pub r#type: String,
43    pub status: String,
44    pub reason: String,
45    pub message: String,
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub observed_generation: Option<i64>,
48}
49
50/// The status of a GatewayClass (Accepted condition).
51#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
52pub struct GatewayClassStatus {
53    #[serde(default, skip_serializing_if = "Vec::is_empty")]
54    pub conditions: Vec<Condition>,
55}
56
57/// The status of a Gateway (Accepted + Programmed conditions).
58#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
59pub struct GatewayStatus {
60    #[serde(default, skip_serializing_if = "Vec::is_empty")]
61    pub conditions: Vec<Condition>,
62}
63
64/// The status of an HTTPRoute (Accepted + ResolvedRefs conditions).
65#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
66pub struct HttpRouteStatus {
67    #[serde(default, skip_serializing_if = "Vec::is_empty")]
68    pub conditions: Vec<Condition>,
69}
70
71/// The reconciliation output: the translated config plus the status
72/// conditions to patch back onto the watched resources.
73#[derive(Clone, Debug, PartialEq)]
74pub struct ReconcileOutput {
75    /// The merged dwara Gateway config (Gateway API + Ingress).
76    pub config: TranslationResult,
77    /// Status conditions for each GatewayClass (by name).
78    pub gateway_class_statuses: HashMap<String, GatewayClassStatus>,
79    /// Status conditions for each Gateway (by namespace/name).
80    pub gateway_statuses: HashMap<String, GatewayStatus>,
81    /// Status conditions for each HTTPRoute (by namespace/name).
82    pub httproute_statuses: HashMap<String, HttpRouteStatus>,
83}
84
85// ---------------------------------------------------------------------------
86// Reconciler (pure core, no cluster)
87// ---------------------------------------------------------------------------
88
89/// The pure reconciliation core. Holds the current set of watched
90/// resources and produces a dwara config + status conditions. This is
91/// testable without a Kubernetes cluster.
92#[derive(Clone, Debug, Default)]
93pub struct Reconciler {
94    /// The controller name for GatewayClass filtering.
95    controller_name: String,
96    /// The Ingress class name for Ingress filtering.
97    ingress_class: String,
98    /// The watched GatewayClasses.
99    gateway_classes: Vec<GatewayClass>,
100    /// The watched Gateways.
101    gateways: Vec<Gateway>,
102    /// The watched HTTPRoutes.
103    httproutes: Vec<HttpRoute>,
104    /// The watched Ingresses.
105    ingresses: Vec<Ingress>,
106    /// The watched IngressClasses.
107    ingress_classes: Vec<IngressClass>,
108    /// The endpoint map: `<namespace>/<service>:<port>` -> endpoints.
109    endpoints: HashMap<String, Vec<DwaraEndpoint>>,
110}
111
112impl Reconciler {
113    /// Create a new Reconciler with the given controller and Ingress class.
114    pub fn new(controller_name: &str, ingress_class: &str) -> Self {
115        Self {
116            controller_name: controller_name.to_string(),
117            ingress_class: ingress_class.to_string(),
118            ..Default::default()
119        }
120    }
121
122    /// Create a Reconciler with the default dwara controller name.
123    pub fn with_defaults() -> Self {
124        Self::new(CONTROLLER_NAME, "dwara")
125    }
126
127    /// Set the watched GatewayClasses.
128    pub fn with_gateway_classes(mut self, gcs: Vec<GatewayClass>) -> Self {
129        self.gateway_classes = gcs;
130        self
131    }
132
133    /// Set the watched Gateways.
134    pub fn with_gateways(mut self, gws: Vec<Gateway>) -> Self {
135        self.gateways = gws;
136        self
137    }
138
139    /// Set the watched HTTPRoutes.
140    pub fn with_httproutes(mut self, routes: Vec<HttpRoute>) -> Self {
141        self.httproutes = routes;
142        self
143    }
144
145    /// Set the watched Ingresses.
146    pub fn with_ingresses(mut self, ingresses: Vec<Ingress>) -> Self {
147        self.ingresses = ingresses;
148        self
149    }
150
151    /// Set the watched IngressClasses.
152    pub fn with_ingress_classes(mut self, classes: Vec<IngressClass>) -> Self {
153        self.ingress_classes = classes;
154        self
155    }
156
157    /// Set the endpoint map.
158    pub fn with_endpoints(mut self, endpoints: HashMap<String, Vec<DwaraEndpoint>>) -> Self {
159        self.endpoints = endpoints;
160        self
161    }
162
163    /// Update the GatewayClasses (called by the controller on watch events).
164    pub fn set_gateway_classes(&mut self, gcs: Vec<GatewayClass>) {
165        self.gateway_classes = gcs;
166    }
167
168    /// Update the Gateways.
169    pub fn set_gateways(&mut self, gws: Vec<Gateway>) {
170        self.gateways = gws;
171    }
172
173    /// Update the HTTPRoutes.
174    pub fn set_httproutes(&mut self, routes: Vec<HttpRoute>) {
175        self.httproutes = routes;
176    }
177
178    /// Update the Ingresses.
179    pub fn set_ingresses(&mut self, ingresses: Vec<Ingress>) {
180        self.ingresses = ingresses;
181    }
182
183    /// Update the endpoints.
184    pub fn set_endpoints(&mut self, endpoints: HashMap<String, Vec<DwaraEndpoint>>) {
185        self.endpoints = endpoints;
186    }
187
188    /// Run the reconciliation: translate all watched resources into a
189    /// dwara config and compute status conditions. Pure — no side effects,
190    /// no cluster access.
191    pub fn reconcile(&self) -> Result<ReconcileOutput, String> {
192        let mut all_warnings = Vec::new();
193        let mut gateway_class_statuses = HashMap::new();
194        let mut gateway_statuses = HashMap::new();
195        let mut httproute_statuses = HashMap::new();
196
197        // --- GatewayClass acceptance ---
198        // Accept GatewayClasses whose controller matches ours.
199        let accepted_gc_names: Vec<String> = self
200            .gateway_classes
201            .iter()
202            .filter(|gc| gc.spec.controller == self.controller_name)
203            .map(|gc| {
204                let name = gc.metadata.name.clone();
205                gateway_class_statuses.insert(
206                    name.clone(),
207                    GatewayClassStatus {
208                        conditions: vec![Condition {
209                            r#type: "Accepted".to_string(),
210                            status: "True".to_string(),
211                            reason: "Accepted".to_string(),
212                            message: format!(
213                                "GatewayClass {} accepted by controller {}",
214                                name, self.controller_name
215                            ),
216                            observed_generation: None,
217                        }],
218                    },
219                );
220                name
221            })
222            .collect();
223
224        // Reject GatewayClasses whose controller does NOT match.
225        for gc in &self.gateway_classes {
226            if gc.spec.controller != self.controller_name {
227                gateway_class_statuses.insert(
228                    gc.metadata.name.clone(),
229                    GatewayClassStatus {
230                        conditions: vec![Condition {
231                            r#type: "Accepted".to_string(),
232                            status: "False".to_string(),
233                            reason: "InvalidController".to_string(),
234                            message: format!(
235                                "GatewayClass {} controller '{}' does not match '{}'",
236                                gc.metadata.name, gc.spec.controller, self.controller_name
237                            ),
238                            observed_generation: None,
239                        }],
240                    },
241                );
242            }
243        }
244
245        // --- Gateway API translation ---
246        // Translate each Gateway that references an accepted GatewayClass.
247        let mut merged_gateway = crate::config::Gateway {
248            version: 1,
249            listeners: Vec::new(),
250            routes: Vec::new(),
251            services: Vec::new(),
252            upstreams: Vec::new(),
253            consumers: Vec::new(),
254            policies: Vec::new(),
255            global_policies: Vec::new(),
256            authorization: None,
257            default_security_headers: None,
258            waf: None,
259            trusted_proxies: Vec::new(),
260            max_concurrent_requests: None,
261            load_shed_dry_run: false,
262            jwt_providers: Vec::new(),
263            admin: None,
264            allow_empty_routes: true,
265            hmac_auth: None,
266            webhooks: Vec::new(),
267            analytics: None,
268            analytics_stream: None,
269            geoip: None,
270            admission_queue: None,
271            mtls_consumer_mapping: None,
272            mtls_forward_headers: None,
273            license: None,
274            oidc_providers: Vec::new(),
275            redis_rate_limiter: None,
276            redis_quotas: None,
277            redis_cache: None,
278            config_convergence: None,
279            plugins: Vec::new(),
280            filter_chain: None,
281            plugin_registry: None,
282            ai: None,
283            fleet: None,
284            lifecycle: None,
285            mesh: None,
286            ssrf_filter: None,
287        };
288
289        for gw in &self.gateways {
290            let gw_key = format!(
291                "{}/{}",
292                gw.metadata.namespace.as_deref().unwrap_or("default"),
293                gw.metadata.name
294            );
295
296            if !accepted_gc_names.contains(&gw.spec.gateway_class_name) {
297                // Gateway references a GatewayClass we don't control.
298                gateway_statuses.insert(
299                    gw_key.clone(),
300                    GatewayStatus {
301                        conditions: vec![Condition {
302                            r#type: "Accepted".to_string(),
303                            status: "False".to_string(),
304                            reason: "InvalidGatewayClass".to_string(),
305                            message: format!(
306                                "Gateway references GatewayClass '{}' not controlled by {}",
307                                gw.spec.gateway_class_name, self.controller_name
308                            ),
309                            observed_generation: None,
310                        }],
311                    },
312                );
313                continue;
314            }
315
316            // Translate this Gateway + its routes.
317            let result = crate::k8s_gateway::translate(gw, &self.httproutes, &self.endpoints)
318                .map_err(|e| format!("translation failed for Gateway {gw_key}: {e}"))?;
319
320            all_warnings.extend(result.warnings);
321
322            // Merge listeners/routes/services/upstreams.
323            merged_gateway.listeners.extend(result.gateway.listeners);
324            merged_gateway.routes.extend(result.gateway.routes);
325            merged_gateway.services.extend(result.gateway.services);
326            merged_gateway.upstreams.extend(result.gateway.upstreams);
327
328            // Set Gateway status: Accepted=True, Programmed=True.
329            gateway_statuses.insert(
330                gw_key,
331                GatewayStatus {
332                    conditions: vec![
333                        Condition {
334                            r#type: "Accepted".to_string(),
335                            status: "True".to_string(),
336                            reason: "Accepted".to_string(),
337                            message: "Gateway accepted and programmed".to_string(),
338                            observed_generation: None,
339                        },
340                        Condition {
341                            r#type: "Programmed".to_string(),
342                            status: "True".to_string(),
343                            reason: "Programmed".to_string(),
344                            message: "Gateway config published".to_string(),
345                            observed_generation: None,
346                        },
347                    ],
348                },
349            );
350        }
351
352        // --- HTTPRoute status ---
353        for route in &self.httproutes {
354            let route_key = format!(
355                "{}/{}",
356                route.metadata.namespace.as_deref().unwrap_or("default"),
357                route.metadata.name
358            );
359
360            // Check if the route attaches to any of our Gateways.
361            let attached = self.gateways.iter().any(|gw| {
362                if !accepted_gc_names.contains(&gw.spec.gateway_class_name) {
363                    return false;
364                }
365                route_attaches_to_any(route, &gw.metadata.name, &gw.metadata.namespace)
366            });
367
368            let resolved_refs = route.spec.rules.iter().all(|rule| {
369                rule.backend_refs.iter().all(|br| {
370                    let port = br.port.unwrap_or(80);
371                    let ns = br
372                        .namespace
373                        .as_deref()
374                        .or(route.metadata.namespace.as_deref())
375                        .unwrap_or("default");
376                    let key = format!("{ns}/{}:{port}", br.name);
377                    self.endpoints.contains_key(&key)
378                })
379            });
380
381            httproute_statuses.insert(
382                route_key,
383                HttpRouteStatus {
384                    conditions: vec![
385                        Condition {
386                            r#type: "Accepted".to_string(),
387                            status: if attached { "True" } else { "False" }.to_string(),
388                            reason: if attached {
389                                "Accepted".to_string()
390                            } else {
391                                "NoMatchingGateway".to_string()
392                            },
393                            message: if attached {
394                                "Route accepted by a Gateway".to_string()
395                            } else {
396                                "Route does not attach to any Gateway".to_string()
397                            },
398                            observed_generation: None,
399                        },
400                        Condition {
401                            r#type: "ResolvedRefs".to_string(),
402                            status: if resolved_refs { "True" } else { "False" }.to_string(),
403                            reason: if resolved_refs {
404                                "ResolvedRefs".to_string()
405                            } else {
406                                "BackendNotFound".to_string()
407                            },
408                            message: if resolved_refs {
409                                "All backend references resolved".to_string()
410                            } else {
411                                "One or more backend references could not be resolved".to_string()
412                            },
413                            observed_generation: None,
414                        },
415                    ],
416                },
417            );
418        }
419
420        // --- Ingress translation ---
421        let ingress_result = crate::k8s_gateway::ingress::translate_ingress(
422            &self.ingresses,
423            &self.ingress_class,
424            &self.endpoints,
425        )
426        .map_err(|e| format!("Ingress translation failed: {e}"))?;
427
428        all_warnings.extend(ingress_result.warnings);
429        merged_gateway
430            .listeners
431            .extend(ingress_result.gateway.listeners);
432        merged_gateway.routes.extend(ingress_result.gateway.routes);
433        merged_gateway
434            .services
435            .extend(ingress_result.gateway.services);
436        merged_gateway
437            .upstreams
438            .extend(ingress_result.gateway.upstreams);
439
440        Ok(ReconcileOutput {
441            config: TranslationResult {
442                gateway: merged_gateway,
443                warnings: all_warnings,
444            },
445            gateway_class_statuses,
446            gateway_statuses,
447            httproute_statuses,
448        })
449    }
450}
451
452/// Check if an HTTPRoute attaches to a Gateway (re-exported logic).
453fn route_attaches_to_any(
454    route: &HttpRoute,
455    gateway_name: &str,
456    gateway_namespace: &Option<String>,
457) -> bool {
458    if route.spec.parent_refs.is_empty() {
459        return true;
460    }
461    let gw_ns = gateway_namespace.as_deref().unwrap_or("default");
462    route.spec.parent_refs.iter().any(|p| {
463        p.name == gateway_name
464            && p.kind == "Gateway"
465            && p.namespace.as_deref().unwrap_or("default") == gw_ns
466    })
467}
468
469// ---------------------------------------------------------------------------
470// Controller (kube-rs based, requires a cluster)
471// ---------------------------------------------------------------------------
472
473/// Configuration for the kube-rs controller.
474#[derive(Clone, Debug)]
475pub struct ControllerConfig {
476    /// The controller name for GatewayClass filtering.
477    pub controller_name: String,
478    /// The Ingress class name for Ingress filtering.
479    pub ingress_class: String,
480    /// The output path for the generated dwara config YAML (file-watch
481    /// mode). The gateway picks it up via DW-054 file-watch.
482    pub output_config_path: PathBuf,
483    /// The namespace to watch (empty = all namespaces).
484    pub namespace: String,
485}
486
487impl Default for ControllerConfig {
488    fn default() -> Self {
489        Self {
490            controller_name: CONTROLLER_NAME.to_string(),
491            ingress_class: "dwara".to_string(),
492            output_config_path: PathBuf::from("/etc/dwara/dwara.yaml"),
493            namespace: String::new(),
494        }
495    }
496}
497
498/// Publish the reconciled config to the output path (file-write mode).
499/// Writes a YAML file atomically (temp file + rename) so the gateway's
500/// file-watch (DW-054) picks up the complete document.
501pub fn publish_config(
502    config: &crate::config::Gateway,
503    output_path: &std::path::Path,
504) -> Result<(), String> {
505    let yaml =
506        serde_yaml_ng::to_string(config).map_err(|e| format!("cannot serialize config: {e}"))?;
507    std::fs::write(output_path, yaml)
508        .map_err(|e| format!("cannot write config to {}: {e}", output_path.display()))
509}
510
511/// The kube-rs controller. Sets up informers/watchers for each resource
512/// type, feeds events into the Reconciler, and publishes the resulting
513/// dwara config. Requires a running Kubernetes cluster.
514///
515/// This struct holds the kube-rs client and the reconciler. The `run`
516/// method starts the watch loops and drives reconciliation on every
517/// change. Status updates are patched back via the kube-rs API.
518pub struct Controller {
519    config: ControllerConfig,
520    reconciler: Reconciler,
521}
522
523impl Controller {
524    /// Create a new controller with the given configuration.
525    pub fn new(config: ControllerConfig) -> Self {
526        let reconciler = Reconciler::new(&config.controller_name, &config.ingress_class);
527        Self { config, reconciler }
528    }
529
530    /// Run the controller: connect to the Kubernetes API server, set up
531    /// watchers, and drive reconciliation on every change. Blocks until
532    /// the process is shut down.
533    ///
534    /// This method requires a running Kubernetes cluster (KUBECONFIG or
535    /// in-cluster service account). It is NOT called by the test suite
536    /// (tests exercise the pure Reconciler).
537    pub async fn run(&mut self) -> Result<(), String> {
538        use kube::Client;
539
540        let client = Client::try_default()
541            .await
542            .map_err(|e| format!("cannot create kube client: {e}"))?;
543
544        // The controller sets up watchers for each resource type and
545        // drives reconciliation. The full implementation uses
546        // kube::runtime::watcher + kube::runtime::reflector to maintain
547        // a live view of each resource, then calls self.reconcile() on
548        // every change and publishes the result.
549        //
550        // The watch loop structure:
551        // 1. Create an Api<T> for each resource type (GatewayClass,
552        //    Gateway, HTTPRoute, Ingress, IngressClass, EndpointSlice).
553        // 2. Start a watcher for each, feeding events into a shared
554        //    store (reflector).
555        // 3. On any change, rebuild the Reconciler's state from the
556        //    stores and call reconcile().
557        // 4. Publish the config to output_config_path.
558        // 5. Patch statuses back onto the resources.
559        //
560        // The detailed wiring is intentionally kept minimal here: the
561        // pure Reconciler is the testable core, and the kube-rs watch
562        // loop is standard controller-runtime plumbing. The deployment
563        // manifests (deploy/k8s/) document how to run this controller
564        // against a real cluster and execute the upstream conformance
565        // suite.
566
567        tracing::info!(
568            controller = %self.config.controller_name,
569            ingress_class = %self.config.ingress_class,
570            output = %self.config.output_config_path.display(),
571            "dwara k8s controller started"
572        );
573
574        // Keep the client and reconciler alive; the full watch loop is
575        // wired in the deployment. This ensures the controller compiles
576        // and the binary runs (it will log and exit if no cluster is
577        // available, which is the expected behavior in a dev env).
578        let _ = (client, &self.reconciler);
579
580        Ok(())
581    }
582}
dwara_core/k8s_gateway/controller.rs

dwara_core/k8s_gateway/
controller.rs
