234 lines
8.6 KiB
Rust
234 lines
8.6 KiB
Rust
//! Reconcile loop for the GIngress controller.
|
|
//!
|
|
//! After any watcher detects a change (Ingress, Secret, Endpoints),
|
|
//! the reconciler reads all fragments from the ConfigStore, cross-references them,
|
|
//! assembles a complete `ProxyConfig`, validates it, and signals a reload.
|
|
|
|
use gingress_proxy::config::{ConfigStore, Endpoint, ProxyConfig, RouteRule, TlsCert};
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::sync::Arc;
|
|
|
|
/// Reconcile the full proxy configuration from current k8s state.
|
|
pub struct Reconciler {
|
|
store: Arc<ConfigStore>,
|
|
}
|
|
|
|
impl Reconciler {
|
|
pub fn new(store: Arc<ConfigStore>) -> Self {
|
|
Self { store }
|
|
}
|
|
|
|
/// Trigger a full reconciliation.
|
|
///
|
|
/// 1. Reads all route fragments (from Ingress watcher)
|
|
/// 2. Reads all TLS certs (from Secret watcher)
|
|
/// 3. Reads all upstream endpoints (from Endpoint watcher)
|
|
/// 4. Cross-references: matches TLS secrets to Ingress hosts,
|
|
/// matches upstreams to route backends
|
|
/// 5. Validates the configuration
|
|
/// 6. Writes the assembled ProxyConfig to the store
|
|
/// 7. Signals reload
|
|
pub fn reconcile(&self) {
|
|
tracing::debug!("Reconciliation started");
|
|
|
|
// Step 1: Gather all routes from ingress-scoped keys
|
|
// Keys look like: "ingress:<ns>/<name>:route:<host>"
|
|
let mut routes: HashMap<String, Vec<RouteRule>> = HashMap::new();
|
|
for key in self.store.keys_with_prefix("ingress:") {
|
|
if !key.contains(":route:") {
|
|
continue;
|
|
}
|
|
// Extract host from "ingress:<ns>/<name>:route:<host>"
|
|
if let Some(host) = key.split(":route:").nth(1) {
|
|
if let Some(rules) = self.store.get::<Vec<RouteRule>>(&key) {
|
|
if !rules.is_empty() {
|
|
routes.entry(host.to_string()).or_default().extend(rules);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Step 2: Gather all TLS certs
|
|
let mut tls_certs: HashMap<String, TlsCert> = HashMap::new();
|
|
for key in self.store.keys_with_prefix("tls:") {
|
|
if let Some(cert) = self.store.get::<TlsCert>(&key) {
|
|
tls_certs.insert(cert.host.clone(), cert);
|
|
}
|
|
}
|
|
|
|
// Step 3: Gather all upstreams keyed by backend ("<ns>/<name>:<port>")
|
|
let mut upstreams: HashMap<String, Vec<Endpoint>> = HashMap::new();
|
|
for key in self.store.keys_with_prefix("upstream:") {
|
|
if let Some(eps) = self.store.get::<Vec<Endpoint>>(&key) {
|
|
if !eps.is_empty() {
|
|
upstreams.insert(key.clone(), eps);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Step 4: Gather rate limits, headers, session affinity, and websocket hosts
|
|
let rate_limits = self.collect_rate_limits(&routes);
|
|
let headers = self.collect_headers();
|
|
let session_affinity = self.collect_session_affinity(&routes);
|
|
let websocket_hosts = self.collect_websocket_hosts();
|
|
|
|
// Step 5: Build the complete ProxyConfig
|
|
let cfg = ProxyConfig {
|
|
routes,
|
|
tls: tls_certs,
|
|
upstreams,
|
|
rate_limits,
|
|
headers,
|
|
session_affinity,
|
|
websocket_hosts,
|
|
};
|
|
|
|
// Step 6: Validate
|
|
let warnings = self.validate_config(&cfg);
|
|
for w in &warnings {
|
|
tracing::warn!("{}", w);
|
|
}
|
|
|
|
// Step 7: Store the assembled config as the canonical snapshot
|
|
self.store.set(
|
|
"_assembled",
|
|
&serde_json::to_value(&cfg).unwrap_or_default(),
|
|
);
|
|
|
|
self.store.signal_reload();
|
|
tracing::info!(
|
|
routes = cfg.routes.len(),
|
|
tls_hosts = cfg.tls.len(),
|
|
upstreams = cfg.upstreams.len(),
|
|
warnings = warnings.len(),
|
|
"Reconciliation complete"
|
|
);
|
|
}
|
|
|
|
/// Cross-reference: for each Ingress TLS entry, find the Secret cert.
|
|
///
|
|
/// The Ingress TLS section maps: `hosts: [example.com]` → `secretName: my-cert`.
|
|
/// The Secret watcher stores the cert at key `tls:<secretName>`.
|
|
/// We already map secretName → host in ingress_watcher, so this is a no-op
|
|
/// when the ingress_watcher uses correct key mapping.
|
|
pub fn cross_reference_tls(&self) -> HashMap<String, TlsCert> {
|
|
let mut host_certs: HashMap<String, TlsCert> = HashMap::new();
|
|
|
|
// TLS secret name → host mapping is stored by the ingress watcher
|
|
// at key: "tls-host:<secretName>" → Vec<String> (hosts)
|
|
for key in self.store.keys_with_prefix("tls-host:") {
|
|
let secret_name = &key["tls-host:".len()..];
|
|
let hosts: Vec<String> = self.store.get::<Vec<String>>(&key).unwrap_or_default();
|
|
|
|
// Look up the actual cert: key "tls:<host>" (stored by secret watcher
|
|
// using the certificate's SAN/CN host, but also "tls-secret:<secretName>")
|
|
let cert_key = format!("tls-secret:{}", secret_name);
|
|
if let Some(cert) = self.store.get::<TlsCert>(&cert_key) {
|
|
for host in hosts {
|
|
host_certs.insert(host, cert.clone());
|
|
}
|
|
}
|
|
}
|
|
|
|
host_certs
|
|
}
|
|
|
|
/// Validate the assembled configuration. Returns warnings.
|
|
fn validate_config(&self, cfg: &ProxyConfig) -> Vec<String> {
|
|
let mut warnings = Vec::new();
|
|
|
|
// Check: every TLS host has a route
|
|
for host in cfg.tls.keys() {
|
|
if !cfg.routes.contains_key(host) {
|
|
warnings.push(format!(
|
|
"TLS configured for host '{}' but no routes exist for this host",
|
|
host
|
|
));
|
|
}
|
|
}
|
|
|
|
// Check: every route backend has upstream endpoints
|
|
for (host, rules) in &cfg.routes {
|
|
for rule in rules {
|
|
let backend_key =
|
|
format!("upstream:{}/{}", rule.backend.namespace, rule.backend.name);
|
|
|
|
if !cfg.upstreams.contains_key(&backend_key) {
|
|
warnings.push(format!(
|
|
"Host '{}' routes to backend {}/{}:{} but no endpoints found",
|
|
host, rule.backend.namespace, rule.backend.name, rule.backend.port
|
|
));
|
|
}
|
|
}
|
|
}
|
|
|
|
// Check: orphaned upstreams (no route references them)
|
|
let mut referenced_backends: HashSet<String> = HashSet::new();
|
|
for rules in cfg.routes.values() {
|
|
for rule in rules {
|
|
let bk = format!("upstream:{}/{}", rule.backend.namespace, rule.backend.name);
|
|
referenced_backends.insert(bk);
|
|
}
|
|
}
|
|
for upstream_key in cfg.upstreams.keys() {
|
|
if !referenced_backends.contains(upstream_key) {
|
|
warnings.push(format!(
|
|
"Upstream '{}' has no routes referencing it (orphaned)",
|
|
upstream_key
|
|
));
|
|
}
|
|
}
|
|
|
|
warnings
|
|
}
|
|
|
|
/// Collect rate limit policies for all hosts that have routes.
|
|
fn collect_rate_limits(
|
|
&self,
|
|
routes: &HashMap<String, Vec<RouteRule>>,
|
|
) -> HashMap<String, gingress_proxy::config::RateLimitPolicy> {
|
|
let mut limits = HashMap::new();
|
|
for host in routes.keys() {
|
|
let key = format!("rate_limit:{}", host);
|
|
if let Some(policy) = self.store.get(&key) {
|
|
limits.insert(host.clone(), policy);
|
|
}
|
|
}
|
|
limits
|
|
}
|
|
|
|
/// Collect header operations for all hosts.
|
|
fn collect_headers(&self) -> HashMap<String, Vec<gingress_proxy::config::HeaderOp>> {
|
|
let mut headers = HashMap::new();
|
|
for key in self.store.keys_with_prefix("headers:") {
|
|
let host = &key["headers:".len()..];
|
|
if let Some(ops) = self.store.get(&key) {
|
|
headers.insert(host.to_string(), ops);
|
|
}
|
|
}
|
|
headers
|
|
}
|
|
|
|
/// Collect session affinity configs for all hosts that have routes.
|
|
fn collect_session_affinity(
|
|
&self,
|
|
_routes: &HashMap<String, Vec<RouteRule>>,
|
|
) -> HashMap<String, gingress_proxy::config::SessionAffinityConfig> {
|
|
let mut affinity = HashMap::new();
|
|
for key in self.store.keys_with_prefix("session_affinity:") {
|
|
let host = &key["session_affinity:".len()..];
|
|
if let Some(cfg) = self.store.get(&key) {
|
|
affinity.insert(host.to_string(), cfg);
|
|
}
|
|
}
|
|
affinity
|
|
}
|
|
|
|
/// Collect WebSocket-enabled hosts.
|
|
fn collect_websocket_hosts(&self) -> Vec<String> {
|
|
self.store
|
|
.get::<Vec<String>>("websocket:hosts")
|
|
.unwrap_or_default()
|
|
}
|
|
}
|