rws_proxy/ingress/
mod.rs

1//! Kubernetes Ingress watcher and router.
2//!
3//! [`KubernetesIngressWatcher`] watches the Kubernetes API for Ingress
4//! resources and maintains a live route table. [`IngressRouter`] implements
5//! [`Application`] and routes incoming HTTP requests to the appropriate
6//! upstream service using the live rule table.
7//!
8//! # Prerequisites
9//!
10//! The watcher communicates with the Kubernetes API over plain HTTP/1.1 by
11//! default. For in-cluster use, expose the API via `kubectl proxy` and point
12//! the watcher at `http://localhost:8001`:
13//!
14//! ```text
15//! kubectl proxy &
16//! export RWS_K8S_API_SERVER=http://localhost:8001
17//! export RWS_K8S_TOKEN=
18//! export RWS_K8S_NAMESPACE=default
19//! ```
20//!
21//! Or, with the `http-client` or `http2` feature enabled, connect directly
22//! to the in-cluster API server over TLS — see
23//! [`KubernetesIngressWatcher::from_service_account`].
24//!
25//! # Example
26//!
27//! ```rust,no_run
28//! # extern crate rws_proxy as rust_web_server;
29//! use rust_web_server::ingress::{IngressRouter, KubernetesIngressWatcher};
30//! use rust_web_server::server::Server;
31//!
32//! let watcher = KubernetesIngressWatcher::from_env().expect("K8s env not set");
33//! watcher.start();
34//!
35//! let app = IngressRouter::new(watcher);
36//! // Server::run(app);  // pass to your server
37//! ```
38
39#[cfg(test)]
40mod tests;
41
42#[cfg(any(feature = "http-client", feature = "http2"))]
43mod tls;
44// `pub(crate)` (not just `mod`) so `service_discovery::etcd` can reuse
45// `read_chunked_lines` for its own gRPC-gateway JSON watch stream — the same
46// "chunked NDJSON-shaped event stream" problem this module already solved.
47#[doc(hidden)]
48pub mod watch;
49
50use std::io::{Read, Write};
51use std::net::TcpStream;
52use std::sync::{Arc, RwLock};
53use std::time::Duration;
54
55use crate::application::Application;
56use crate::request::Request;
57use crate::response::{Response, STATUS_CODE_REASON_PHRASE};
58use crate::server::ConnectionInfo;
59
60/// How long a watch connection is allowed to sit idle before we treat it
61/// as stalled and reconnect. Generous — legitimate watch connections can
62/// be quiet for long stretches on an unchanging cluster.
63const WATCH_READ_TIMEOUT: Duration = Duration::from_secs(300);
64/// Backoff between watch reconnect attempts (including the very first
65/// connection failing, e.g. because the API server doesn't support watch).
66const WATCH_RECONNECT_BACKOFF: Duration = Duration::from_secs(2);
67
68// ── PathType ──────────────────────────────────────────────────────────────────
69
70/// Mirrors Kubernetes' `Ingress.spec.rules[].http.paths[].pathType`.
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72pub enum PathType {
73    /// Element-wise path-segment prefix match (the Kubernetes default and
74    /// this crate's fallback when `pathType` is absent from older/lenient
75    /// API responses).
76    Prefix,
77    /// The request path must equal the rule's path exactly (ignoring any
78    /// query string).
79    Exact,
80    /// Matching is left to the Ingress controller. This router treats it
81    /// the same as `Prefix`, same as most controllers do in practice.
82    ImplementationSpecific,
83}
84
85impl PathType {
86    fn parse(s: &str) -> Self {
87        match s {
88            "Exact" => PathType::Exact,
89            "ImplementationSpecific" => PathType::ImplementationSpecific,
90            _ => PathType::Prefix,
91        }
92    }
93}
94
95// ── IngressRule ───────────────────────────────────────────────────────────────
96
97/// A single routing rule parsed from a Kubernetes Ingress resource.
98#[derive(Debug, Clone, PartialEq)]
99pub struct IngressRule {
100    /// Value of `spec.rules[].host`.  Empty string means match-all.
101    pub host: String,
102    /// Value of `spec.rules[].http.paths[].path`.
103    pub path: String,
104    /// Value of `spec.rules[].http.paths[].pathType`. Defaults to
105    /// [`PathType::Prefix`] if the field is absent.
106    pub path_type: PathType,
107    /// Kubernetes service name (`spec.rules[].http.paths[].backend.service.name`).
108    pub service_name: String,
109    /// Kubernetes service port number.
110    pub service_port: u16,
111    /// Namespace the Ingress lives in.
112    pub namespace: String,
113}
114
115impl IngressRule {
116    /// Build the upstream Kubernetes DNS address.
117    ///
118    /// Returns `"{service_name}.{namespace}.svc.cluster.local:{service_port}"`.
119    pub fn upstream_addr(&self) -> String {
120        format!(
121            "{}.{}.svc.cluster.local:{}",
122            self.service_name, self.namespace, self.service_port
123        )
124    }
125
126    /// Returns `true` if this rule matches the given `host` header value and
127    /// request `uri`.
128    ///
129    /// * If `self.host` is non-empty, the incoming `host` must match
130    ///   (case-insensitive).
131    /// * Any query string on `uri` is ignored for path matching.
132    /// * [`PathType::Exact`] requires the request path to equal `self.path`
133    ///   exactly.
134    /// * [`PathType::Prefix`] (and [`PathType::ImplementationSpecific`])
135    ///   match on whole path segments — `/foo` matches `/foo`, `/foo/`, and
136    ///   `/foo/bar`, but **not** `/foobar` — not a raw byte prefix.
137    pub fn matches(&self, host: &str, uri: &str) -> bool {
138        if !self.host.is_empty() && !self.host.eq_ignore_ascii_case(host) {
139            return false;
140        }
141        let request_path = uri.split('?').next().unwrap_or(uri);
142        match self.path_type {
143            PathType::Exact => request_path == self.path,
144            PathType::Prefix | PathType::ImplementationSpecific => {
145                path_prefix_matches(&self.path, request_path)
146            }
147        }
148    }
149}
150
151/// Element-wise prefix match per Kubernetes' `pathType: Prefix` semantics.
152fn path_prefix_matches(prefix: &str, request_path: &str) -> bool {
153    if prefix == "/" {
154        return true;
155    }
156    let trimmed = prefix.trim_end_matches('/');
157    request_path == trimmed || request_path.starts_with(&format!("{trimmed}/"))
158}
159
160// ── JSON helpers ──────────────────────────────────────────────────────────────
161
162/// Find the first occurrence of `"field": "VALUE"` in `json` and return `VALUE`.
163fn extract_str_field<'a>(json: &'a str, field: &str) -> Option<&'a str> {
164    let needle = format!("\"{}\":", field);
165    let start = json.find(needle.as_str())?;
166    let after_colon = &json[start + needle.len()..];
167    let after_colon = after_colon.trim_start_matches(' ');
168    if !after_colon.starts_with('"') {
169        return None;
170    }
171    let inner = &after_colon[1..];
172    let end = inner.find('"')?;
173    Some(&inner[..end])
174}
175
176/// Find the *last* occurrence of `"field": "VALUE"` in `json` and return
177/// `VALUE` — used to find the `namespace` field belonging to *this*
178/// Ingress item by searching backward from its `spec`, since `metadata`
179/// (and the `namespace` field within it) always precedes `spec` in a real
180/// Kubernetes object's JSON encoding.
181fn extract_last_str_field<'a>(json: &'a str, field: &str) -> Option<&'a str> {
182    let needle = format!("\"{}\":", field);
183    let start = json.rfind(needle.as_str())?;
184    let after_colon = &json[start + needle.len()..];
185    let after_colon = after_colon.trim_start_matches(' ');
186    if !after_colon.starts_with('"') {
187        return None;
188    }
189    let inner = &after_colon[1..];
190    let end = inner.find('"')?;
191    Some(&inner[..end])
192}
193
194/// Find the first occurrence of `"field": NUMBER` in `json` and parse as `u16`.
195fn extract_u16_field(json: &str, field: &str) -> Option<u16> {
196    let needle = format!("\"{}\":", field);
197    let start = json.find(needle.as_str())?;
198    let after_colon = &json[start + needle.len()..];
199    let after_colon = after_colon.trim_start_matches(' ');
200    let end = after_colon.find(|c: char| !c.is_ascii_digit())?;
201    after_colon[..end].parse().ok()
202}
203
204// ── parse_ingress_list ────────────────────────────────────────────────────────
205
206/// Parse a Kubernetes Ingress list JSON body into a `Vec<IngressRule>`.
207///
208/// This is a minimal, hand-rolled parser that handles the common formatting
209/// returned by the Kubernetes API server.  It does not depend on any external
210/// JSON library.
211///
212/// `ingress_class`, if `Some`, restricts the result to Ingress objects whose
213/// `spec.ingressClassName` equals it exactly — an Ingress with no
214/// `ingressClassName` at all never matches a `Some` filter. Pass `None` to
215/// accept every Ingress regardless of class (the historical, and still
216/// default, behavior — see [`KubernetesIngressWatcher::ingress_class`]).
217pub fn parse_ingress_list(json: &str, ingress_class: Option<&str>) -> Vec<IngressRule> {
218    let mut rules = Vec::new();
219
220    // Every `"spec"` occurrence starts one Ingress item's spec object. The
221    // section between this occurrence and the next one (or end of string)
222    // is exactly what the old split()-based version of this parser treated
223    // as "the current item" — kept as-is here, just computed via explicit
224    // byte offsets instead of `str::split` so we can also look *backward*
225    // from each occurrence (see `extract_last_str_field` below).
226    let spec_positions: Vec<usize> = json.match_indices("\"spec\"").map(|(i, _)| i).collect();
227    for (idx, &pos) in spec_positions.iter().enumerate() {
228        let section_start = pos + "\"spec\"".len();
229        let section_end = spec_positions.get(idx + 1).copied().unwrap_or(json.len());
230        let section = &json[section_start..section_end];
231
232        // `namespace` lives in `metadata`, which precedes `spec` for every
233        // real Kubernetes object — search backward from this `spec`
234        // occurrence, not forward into `section`. (A prior version of this
235        // parser searched `section` itself, which never actually contains
236        // `namespace` for a real API response — metadata always comes
237        // first — so it silently always fell back to `"default"` unless
238        // the real namespace happened to also be "default".)
239        let namespace = extract_last_str_field(&json[..pos], "namespace")
240            .unwrap_or("default")
241            .to_string();
242
243        if let Some(wanted) = ingress_class {
244            if extract_str_field(section, "ingressClassName") != Some(wanted) {
245                continue;
246            }
247        }
248
249        // Within this spec section, look for rules.
250        let rules_sections: Vec<&str> = section.split("\"rules\"").collect();
251        for rules_section in rules_sections.iter().skip(1) {
252            // Extract host (may be absent).
253            let host = extract_str_field(rules_section, "host").unwrap_or("").to_string();
254
255            // Within the rules section, split on "paths".
256            let paths_sections: Vec<&str> = rules_section.split("\"paths\"").collect();
257            for paths_section in paths_sections.iter().skip(1) {
258                // Within each paths entry, split on path objects.
259                // Each path entry looks like: {"path":"/foo","backend":...}
260                // We split on `"path"` and take alternating sections.
261                let path_entries: Vec<&str> = paths_section.split("\"path\"").collect();
262                for path_entry in path_entries.iter().skip(1) {
263                    let path = extract_str_field(path_entry, "path")
264                        .or_else(|| {
265                            // The split consumed "path" so the value comes right after ":"
266                            let after_colon = path_entry.trim_start_matches(':').trim_start_matches(' ');
267                            if after_colon.starts_with('"') {
268                                let inner = &after_colon[1..];
269                                inner.find('"').map(|end| &inner[..end])
270                            } else {
271                                None
272                            }
273                        })
274                        .unwrap_or("/")
275                        .to_string();
276                    let path_type = extract_str_field(path_entry, "pathType")
277                        .map(PathType::parse)
278                        .unwrap_or(PathType::Prefix);
279
280                    let service_name =
281                        extract_str_field(path_entry, "name").unwrap_or("").to_string();
282                    let service_port =
283                        extract_u16_field(path_entry, "number").unwrap_or(80);
284
285                    if !service_name.is_empty() {
286                        rules.push(IngressRule {
287                            host: host.clone(),
288                            path,
289                            path_type,
290                            service_name,
291                            service_port,
292                            namespace: namespace.clone(),
293                        });
294                    }
295                }
296            }
297        }
298    }
299
300    rules
301}
302
303// ── KubernetesIngressWatcher ──────────────────────────────────────────────────
304
305/// Watches a Kubernetes API server for Ingress resources and maintains a live
306/// routing table.
307pub struct KubernetesIngressWatcher {
308    api_server: String,
309    token: String,
310    namespace: String,
311    ingress_class: Option<String>,
312    poll_interval_secs: u64,
313    rules: Arc<RwLock<Vec<IngressRule>>>,
314    #[cfg(any(feature = "http-client", feature = "http2"))]
315    tls: Option<Arc<tls::InClusterConfig>>,
316}
317
318impl KubernetesIngressWatcher {
319    /// Create a watcher from explicit values.
320    ///
321    /// `api_server` should be a plain-HTTP URL such as `http://localhost:8001`.
322    /// The default namespace is `"default"`.
323    pub fn new(api_server: impl Into<String>, token: impl Into<String>) -> Self {
324        Self {
325            api_server: api_server.into(),
326            token: token.into(),
327            namespace: "default".to_string(),
328            ingress_class: None,
329            poll_interval_secs: 30,
330            rules: Arc::new(RwLock::new(Vec::new())),
331            #[cfg(any(feature = "http-client", feature = "http2"))]
332            tls: None,
333        }
334    }
335
336    /// Configure from the Kubernetes service account files mounted at
337    /// `/var/run/secrets/kubernetes.io/serviceaccount/` and the
338    /// `KUBERNETES_SERVICE_HOST`/`KUBERNETES_SERVICE_PORT` environment
339    /// variables Kubernetes injects into every pod, connecting to the
340    /// in-cluster API server directly over TLS — no `kubectl proxy`
341    /// sidecar needed.
342    ///
343    /// Requires the `http-client` or `http2` feature (both already pull in
344    /// `rustls`); trusts only the cluster's own CA certificate
345    /// (`.../ca.crt`), not the public root store `crate::http_client` uses,
346    /// since the API server's certificate is signed by that private CA.
347    #[cfg(any(feature = "http-client", feature = "http2"))]
348    pub fn from_service_account() -> Result<Self, String> {
349        let cfg = tls::load()?;
350        let namespace = cfg.namespace.clone();
351        let mut watcher = Self::new("", "");
352        watcher.namespace = namespace;
353        watcher.tls = Some(Arc::new(cfg));
354        Ok(watcher)
355    }
356
357    /// Attempt to configure from the Kubernetes service account files at
358    /// `/var/run/secrets/kubernetes.io/serviceaccount/`.
359    ///
360    /// Building without the `http-client` or `http2` feature has no TLS
361    /// implementation available, so this always fails — use `kubectl
362    /// proxy` and configure via environment variables instead:
363    ///
364    /// ```text
365    /// kubectl proxy &
366    /// export RWS_K8S_API_SERVER=http://localhost:8001
367    /// ```
368    #[cfg(not(any(feature = "http-client", feature = "http2")))]
369    pub fn from_service_account() -> Result<Self, String> {
370        Err(
371            "In-cluster TLS (https://kubernetes.default.svc) requires the `http-client` or \
372             `http2` feature. Enable one of those, or use `kubectl proxy` and set \
373             RWS_K8S_API_SERVER=http://localhost:8001 along with RWS_K8S_TOKEN and \
374             RWS_K8S_NAMESPACE, then call KubernetesIngressWatcher::from_env()."
375                .to_string(),
376        )
377    }
378
379    /// Configure from environment variables `RWS_K8S_API_SERVER`, `RWS_K8S_TOKEN`,
380    /// and (optionally) `RWS_K8S_NAMESPACE`.
381    ///
382    /// Returns `Err` if `RWS_K8S_API_SERVER` is not set.
383    pub fn from_env() -> Result<Self, String> {
384        let api_server = std::env::var("RWS_K8S_API_SERVER").map_err(|_| {
385            "RWS_K8S_API_SERVER environment variable is not set".to_string()
386        })?;
387        let token = std::env::var("RWS_K8S_TOKEN").unwrap_or_default();
388        let namespace = std::env::var("RWS_K8S_NAMESPACE").unwrap_or_else(|_| "default".to_string());
389        let mut watcher = Self::new(api_server, token);
390        watcher.namespace = namespace;
391        Ok(watcher)
392    }
393
394    /// Override the namespace filter.  Use `"all"` or empty string for all namespaces.
395    pub fn namespace(mut self, ns: impl Into<String>) -> Self {
396        self.namespace = ns.into();
397        self
398    }
399
400    /// Restrict watched Ingress objects to those whose
401    /// `spec.ingressClassName` equals `class` exactly. Unset (the default)
402    /// accepts every Ingress cluster- or namespace-wide regardless of
403    /// class — fine for a single-controller cluster, but on a
404    /// multi-controller cluster (e.g. running alongside `nginx-ingress`)
405    /// leaves this watcher picking up Ingress objects meant for the other
406    /// controller too, so set this explicitly in that case.
407    pub fn ingress_class(mut self, class: impl Into<String>) -> Self {
408        self.ingress_class = Some(class.into());
409        self
410    }
411
412    /// Override the polling interval in seconds (default: 30).
413    ///
414    /// This remains a periodic full resync even when the watch connection
415    /// (always attempted alongside it — see the [module docs](self)) is
416    /// healthy, the same "trust but verify" pattern real Kubernetes
417    /// controllers use: the watch stream delivers low-latency updates, and
418    /// this interval is the safety net against a missed or silently
419    /// swallowed event.
420    pub fn poll_interval_secs(mut self, secs: u64) -> Self {
421        self.poll_interval_secs = secs;
422        self
423    }
424
425    /// Spawn background threads that keep the rule table up to date: a
426    /// periodic full resync every `poll_interval_secs`, and (best-effort) a
427    /// long-lived watch connection that triggers an immediate resync on any
428    /// change instead of waiting for the next interval. Call once at
429    /// startup.
430    pub fn start(&self) {
431        self.clone_inner().poll_loop();
432        self.clone_inner().watch_loop();
433    }
434
435    fn clone_inner(&self) -> WatcherHandle {
436        WatcherHandle {
437            api_server: self.api_server.clone(),
438            token: self.token.clone(),
439            namespace: self.namespace.clone(),
440            ingress_class: self.ingress_class.clone(),
441            poll_interval_secs: self.poll_interval_secs,
442            rules: Arc::clone(&self.rules),
443            #[cfg(any(feature = "http-client", feature = "http2"))]
444            tls: self.tls.clone(),
445        }
446    }
447
448    /// Return a snapshot of the current rule list.
449    pub fn rules(&self) -> Vec<IngressRule> {
450        self.rules.read().unwrap().clone()
451    }
452
453    /// Perform one synchronous poll cycle.
454    ///
455    /// Exported for testing without starting the background thread.
456    pub fn poll(&self) -> Result<(), String> {
457        let new_rules = self.do_poll()?;
458        *self.rules.write().unwrap() = new_rules;
459        Ok(())
460    }
461
462    fn list_path(&self) -> String {
463        list_path(&self.namespace)
464    }
465
466    fn do_poll(&self) -> Result<Vec<IngressRule>, String> {
467        let path = self.list_path();
468        #[cfg(any(feature = "http-client", feature = "http2"))]
469        let body = fetch(&self.api_server, &self.token, self.tls.as_deref(), &path)?;
470        #[cfg(not(any(feature = "http-client", feature = "http2")))]
471        let body = http_get_plain(&self.api_server, &path, &self.token)?;
472        Ok(parse_ingress_list(&body, self.ingress_class.as_deref()))
473    }
474}
475
476fn list_path(namespace: &str) -> String {
477    if namespace.is_empty() || namespace == "all" {
478        "/apis/networking.k8s.io/v1/ingresses".to_string()
479    } else {
480        format!("/apis/networking.k8s.io/v1/namespaces/{namespace}/ingresses")
481    }
482}
483
484/// Fetch the ingress list body, over TLS if `tls_cfg` is configured,
485/// falling back to plain HTTP otherwise. Shared by
486/// [`KubernetesIngressWatcher::do_poll`] and [`WatcherHandle::poll_once`].
487#[cfg(any(feature = "http-client", feature = "http2"))]
488fn fetch(
489    api_server: &str,
490    token: &str,
491    tls_cfg: Option<&tls::InClusterConfig>,
492    path: &str,
493) -> Result<String, String> {
494    if let Some(cfg) = tls_cfg {
495        return tls::https_get(
496            &cfg.host,
497            cfg.port,
498            "kubernetes.default.svc",
499            cfg.client_config.clone(),
500            &cfg.token,
501            path,
502            Duration::from_secs(10),
503        );
504    }
505    http_get_plain(api_server, path, token)
506}
507
508// Internal handle used by the background threads (avoids having to make
509// KubernetesIngressWatcher Clone while sharing the rules Arc).
510struct WatcherHandle {
511    api_server: String,
512    token: String,
513    namespace: String,
514    ingress_class: Option<String>,
515    poll_interval_secs: u64,
516    rules: Arc<RwLock<Vec<IngressRule>>>,
517    #[cfg(any(feature = "http-client", feature = "http2"))]
518    tls: Option<Arc<tls::InClusterConfig>>,
519}
520
521impl WatcherHandle {
522    fn poll_loop(self) {
523        // Do an initial poll before sleeping.
524        self.poll_once();
525        let interval = Duration::from_secs(self.poll_interval_secs);
526        std::thread::spawn(move || loop {
527            std::thread::sleep(interval);
528            self.poll_once();
529        });
530    }
531
532    fn poll_once(&self) {
533        let path = list_path(&self.namespace);
534        let result = {
535            #[cfg(any(feature = "http-client", feature = "http2"))]
536            {
537                fetch(&self.api_server, &self.token, self.tls.as_deref(), &path)
538            }
539            #[cfg(not(any(feature = "http-client", feature = "http2")))]
540            {
541                http_get_plain(&self.api_server, &path, &self.token)
542            }
543        };
544        match result {
545            Ok(body) => {
546                let new_rules = parse_ingress_list(&body, self.ingress_class.as_deref());
547                *self.rules.write().unwrap() = new_rules;
548            }
549            Err(e) => {
550                eprintln!("ingress watcher: poll failed: {}", e);
551            }
552        }
553    }
554
555    /// Spawn the watch-connection thread — see the [module docs](self) (and
556    /// `watch`'s own docs) for why this triggers a full re-list on any event
557    /// rather than applying deltas incrementally.
558    fn watch_loop(self) {
559        std::thread::spawn(move || loop {
560            if let Err(e) = self.watch_once() {
561                eprintln!("ingress watcher: watch connection error: {e}");
562            }
563            std::thread::sleep(WATCH_RECONNECT_BACKOFF);
564        });
565    }
566
567    fn watch_once(&self) -> Result<(), String> {
568        let path = format!("{}?watch=true", list_path(&self.namespace));
569
570        #[cfg(any(feature = "http-client", feature = "http2"))]
571        if let Some(cfg) = &self.tls {
572            let stream = tls::tls_connect(
573                &cfg.host,
574                cfg.port,
575                "kubernetes.default.svc",
576                cfg.client_config.clone(),
577                WATCH_READ_TIMEOUT,
578            )?;
579            return self.stream_watch(stream, "kubernetes.default.svc", &cfg.token, &path);
580        }
581
582        let (host, addr) = parse_plain_host_addr(&self.api_server)?;
583        let tcp = TcpStream::connect(&addr)
584            .map_err(|e| format!("ingress watcher: connect to {addr} failed: {e}"))?;
585        tcp.set_read_timeout(Some(WATCH_READ_TIMEOUT)).map_err(|e| e.to_string())?;
586        tcp.set_write_timeout(Some(Duration::from_secs(5))).map_err(|e| e.to_string())?;
587        self.stream_watch(tcp, &host, &self.token, &path)
588    }
589
590    fn stream_watch<S: Read + Write>(
591        &self,
592        mut stream: S,
593        host_header: &str,
594        token: &str,
595        path: &str,
596    ) -> Result<(), String> {
597        let auth_header = if token.is_empty() {
598            String::new()
599        } else {
600            format!("Authorization: Bearer {token}\r\n")
601        };
602        let request = format!(
603            "GET {path} HTTP/1.1\r\nHost: {host_header}\r\n{auth_header}Accept: application/json\r\nConnection: keep-alive\r\n\r\n"
604        );
605        stream
606            .write_all(request.as_bytes())
607            .map_err(|e| format!("ingress watcher: watch request write failed: {e}"))?;
608
609        watch::read_chunked_lines(stream, |_line| {
610            // Every event (ADDED/MODIFIED/DELETED/ERROR) is treated
611            // identically: it means *something* changed, so re-list from
612            // scratch rather than trying to apply it as a delta — see the
613            // module docs for why.
614            self.poll_once();
615        })
616    }
617}
618
619/// Parse `host:port` out of a plain-HTTP `api_server` URL, returning both
620/// the bare host (for the `Host:` header) and the `host:port` dial address.
621fn parse_plain_host_addr(api_server: &str) -> Result<(String, String), String> {
622    let rest = api_server
623        .strip_prefix("http://")
624        .ok_or_else(|| format!("ingress watcher: api_server must start with http://, got: {api_server}"))?;
625    let host_port = rest.split('/').next().unwrap_or(rest);
626    let (host, port) = if let Some(colon) = host_port.rfind(':') {
627        let port_str = &host_port[colon + 1..];
628        if let Ok(p) = port_str.parse::<u16>() {
629            (&host_port[..colon], p)
630        } else {
631            (host_port, 80u16)
632        }
633    } else {
634        (host_port, 80u16)
635    };
636    Ok((host.to_string(), format!("{host}:{port}")))
637}
638
639// ── plain-HTTP/1.1 GET helper ─────────────────────────────────────────────────
640
641/// Issue a plain-HTTP/1.1 GET to `{api_server}{path}` with an optional Bearer
642/// token and return the response body as a string.
643fn http_get_plain(api_server: &str, path: &str, token: &str) -> Result<String, String> {
644    let (host, addr) = parse_plain_host_addr(api_server)?;
645    let mut stream = TcpStream::connect(&addr)
646        .map_err(|e| format!("ingress watcher: connect to {} failed: {}", addr, e))?;
647    stream.set_read_timeout(Some(Duration::from_secs(10))).map_err(|e| e.to_string())?;
648    stream.set_write_timeout(Some(Duration::from_secs(5))).map_err(|e| e.to_string())?;
649
650    let auth_header = if token.is_empty() {
651        String::new()
652    } else {
653        format!("Authorization: Bearer {}\r\n", token)
654    };
655
656    let request = format!(
657        "GET {} HTTP/1.1\r\nHost: {}\r\n{}Accept: application/json\r\nConnection: close\r\n\r\n",
658        path, host, auth_header
659    );
660
661    stream.write_all(request.as_bytes()).map_err(|e| e.to_string())?;
662
663    let mut buf = Vec::with_capacity(8192);
664    let mut tmp = [0u8; 4096];
665    loop {
666        match stream.read(&mut tmp) {
667            Ok(0) => break,
668            Ok(n) => buf.extend_from_slice(&tmp[..n]),
669            Err(e) => return Err(format!("ingress watcher: read failed: {}", e)),
670        }
671    }
672
673    parse_http1_response(&buf)
674}
675
676/// Split a complete HTTP/1.1 response into its status code and body,
677/// returning the body as a string if the status is 2xx. Shared by the
678/// plain-HTTP path above and the TLS path in `tls::https_get`.
679fn parse_http1_response(buf: &[u8]) -> Result<String, String> {
680    let header_end = buf
681        .windows(4)
682        .position(|w| w == b"\r\n\r\n")
683        .ok_or_else(|| "ingress watcher: incomplete HTTP response (no header end)".to_string())?;
684
685    let header_str = std::str::from_utf8(&buf[..header_end]).unwrap_or("");
686    let status_line = header_str.lines().next().unwrap_or("");
687    let parts: Vec<&str> = status_line.splitn(3, ' ').collect();
688    if parts.len() < 2 {
689        return Err(format!("ingress watcher: malformed status line: {}", status_line));
690    }
691    let status: u16 = parts[1].parse().unwrap_or(0);
692    if status < 200 || status >= 300 {
693        return Err(format!("ingress watcher: API returned status {}", status));
694    }
695
696    let body_bytes = &buf[header_end + 4..];
697    std::str::from_utf8(body_bytes)
698        .map(|s| s.to_string())
699        .map_err(|e| format!("ingress watcher: non-UTF-8 response body: {}", e))
700}
701
702// ── IngressRouter ─────────────────────────────────────────────────────────────
703
704/// An [`Application`] that routes incoming requests using the live Ingress rule table.
705///
706/// Finds the first matching [`IngressRule`] and forwards the request to
707/// `{service_name}.{namespace}.svc.cluster.local:{service_port}` over HTTP/1.1.
708/// Returns `404 Not Found` when no rule matches.
709pub struct IngressRouter {
710    watcher: KubernetesIngressWatcher,
711    connect_timeout: Duration,
712    read_timeout: Duration,
713}
714
715impl IngressRouter {
716    /// Wrap a watcher in an `IngressRouter` with default timeouts.
717    pub fn new(watcher: KubernetesIngressWatcher) -> Self {
718        Self {
719            watcher,
720            connect_timeout: Duration::from_secs(5),
721            read_timeout: Duration::from_secs(30),
722        }
723    }
724
725    /// Override the TCP connect timeout (default: 5 000 ms).
726    pub fn connect_timeout_ms(mut self, ms: u64) -> Self {
727        self.connect_timeout = Duration::from_millis(ms);
728        self
729    }
730
731    /// Override the response read timeout (default: 30 000 ms).
732    pub fn read_timeout_ms(mut self, ms: u64) -> Self {
733        self.read_timeout = Duration::from_millis(ms);
734        self
735    }
736}
737
738impl Application for IngressRouter {
739    fn execute(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
740        let host = request
741            .get_header("host".to_string())
742            .map(|h| h.value.as_ref())
743            .unwrap_or("");
744
745        let rules = self.watcher.rules();
746        let matched = rules.iter().find(|r| r.matches(host, &request.request_uri));
747
748        match matched {
749            Some(rule) => {
750                let upstream_host = format!(
751                    "{}.{}.svc.cluster.local",
752                    rule.service_name, rule.namespace
753                );
754                crate::proxy::forward_pooled(
755                    crate::proxy::shared_pool(),
756                    request,
757                    crate::proxy::Peer::of(connection),
758                    &upstream_host,
759                    rule.service_port,
760                    self.connect_timeout,
761                    self.read_timeout,
762                )
763                .map(|mut response| {
764                    crate::proxy::status::received(&mut response, None);
765                    response
766                })
767                .or_else(|e| {
768                    let mut response = bad_gateway();
769                    crate::proxy::status::failed_with(&mut response, &e, None);
770                    Ok(response)
771                })
772            }
773            None => Ok(not_found()),
774        }
775    }
776}
777
778fn bad_gateway() -> Response {
779    crate::error::problem(STATUS_CODE_REASON_PHRASE.n502_bad_gateway, "")
780}
781
782fn not_found() -> Response {
783    crate::error::problem::problem_with_text(STATUS_CODE_REASON_PHRASE.n404_not_found, "no matching ingress rule", "404 No matching ingress rule")
784}
rws_proxy/ingress/mod.rs

rws_proxy/ingress/
mod.rs
