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

rust_web_server/ingress/
mod.rs
