From 3329c34a091877cc587657274cd78b44f9303a7a Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Wed, 9 Sep 2026 15:39:28 +0200 Subject: [PATCH] working on #55 --- Cargo.lock | 1 + Cargo.toml | 1 + src/utils/kubernetes.rs | 10 +-- src/utils/kubewatcher.rs | 170 ++++++++++++++++++++------------------- 4 files changed, 95 insertions(+), 87 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 38edab2..21b65b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -156,6 +156,7 @@ dependencies = [ "jsonwebtoken", "k8s-openapi", "kube", + "lazy_static", "libc", "log", "log4rs", diff --git a/Cargo.toml b/Cargo.toml index b3686d6..141043e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,3 +62,4 @@ rustls = { version = "0.23", default-features = false, features = ["ring"] } kube = { version = "4.2.0", features = ["runtime", "client", "derive"] } k8s-openapi = { version = "0.28.0", features = ["v1_32"] } futures = "0.3.34" +lazy_static = "1.5.0" diff --git a/src/utils/kubernetes.rs b/src/utils/kubernetes.rs index eeb3640..5ccf813 100644 --- a/src/utils/kubernetes.rs +++ b/src/utils/kubernetes.rs @@ -7,8 +7,6 @@ use dashmap::DashMap; use log::error; use rand::RngExt; use std::env; -use std::fs; -use std::path::Path; use std::sync::Arc; use crate::utils::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery}; @@ -54,7 +52,7 @@ impl ServiceDiscovery for KubernetesDiscovery { let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 }; let server = servers.get(num).unwrap().to_string(); let path = kuber.tokenpath.unwrap_or_else(|| "/var/run/secrets/kubernetes.io/serviceaccount/token".to_string()); - let namespace = get_current_namespace().unwrap_or_else(|| "default".to_string()); + // let namespace = get_current_namespace().unwrap_or_else(|| "default".to_string()); let token = crate::utils::kuberconsul::read_token(path.as_str()).await; if CryptoProvider::get_default().is_none() { @@ -86,7 +84,6 @@ impl ServiceDiscovery for KubernetesDiscovery { to_https: None, redirect_to: None, authorization: None, - // sticky_sessions: update.sticky_sessions, rate_limit: update.rate_limit, x4xx_limit: update.x4xx_limit, client_headers: None, @@ -129,7 +126,7 @@ impl ServiceDiscovery for KubernetesDiscovery { sheader_list.insert(Arc::from(path_key), server_headers_list); config.server_headers.insert(host_key.clone(), sheader_list); } - let url = format!("https://{}/api/v1/namespaces/{}/endpoints/{}", server, namespace, update.service_name); + let url = format!("https://{}/api/v1/namespaces/{}/endpoints/{}", server, update.namespace, update.service_name); let list = httpclient::for_kuber(&url, &token, &service).await; if list.is_none() { @@ -145,7 +142,7 @@ impl ServiceDiscovery for KubernetesDiscovery { } } } - +/* fn get_current_namespace() -> Option { let ns_path = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"; if Path::new(ns_path).exists() { @@ -155,3 +152,4 @@ fn get_current_namespace() -> Option { } env::var("POD_NAMESPACE").ok() } +*/ diff --git a/src/utils/kubewatcher.rs b/src/utils/kubewatcher.rs index b45db3e..2807568 100644 --- a/src/utils/kubewatcher.rs +++ b/src/utils/kubewatcher.rs @@ -1,3 +1,4 @@ +use dashmap::DashMap; use futures::StreamExt; use hyper::Uri; use k8s_openapi::api::core::v1::{Node, Service}; @@ -11,17 +12,18 @@ use kube::{ }; use log::{error, info}; use serde_json::json; +use std::sync::Arc; use std::time::Duration; use tokio::sync::mpsc; pub struct AralezRouteUpdate { pub host: String, pub path: String, + pub namespace: String, pub service_name: String, pub is_deleted: bool, pub rate_limit: Option, pub x4xx_limit: Option, - pub sticky_sessions: Option, pub client_headers: Option>, pub server_headers: Option>, } @@ -30,11 +32,14 @@ pub struct AralezRouteUpdate { struct AralezAnnotation { rate_limit: Option, x4xx_limit: Option, - sticky_sessions: Option, - pub client_headers: Option>, - pub server_headers: Option>, + namespace: String, + client_headers: Option>, + server_headers: Option>, } +lazy_static::lazy_static! { + static ref PATCHED_INGRESSES: Arc> = Arc::new(DashMap::new()); +} pub async fn start_ingress_watcher_with_config( api_server: &str, token: &str, @@ -82,53 +87,59 @@ async fn process_ingress_change(client: &Client, ing: &Ingress, target_class: &s if !is_deleted { if let (Some(name), Some(ns)) = (ing.metadata.name.as_deref(), ing.metadata.namespace.as_deref()) { - let svc_name = std::env::var("ARALEZ_SERVICE_NAME").unwrap_or_else(|_| "aralez-service".to_string()); - let external_ip = match discover_controller_ip(&client, ns, &svc_name).await { - Some(ip) => ip, - None => discover_node_ip(&client) - .await - .unwrap_or_else(|| std::env::var("ARALEZ_EXTERNAL_IP").unwrap_or_else(|_| "127.0.0.1".to_string())), - }; - let client_clone = client.clone(); - let name_clone = name.to_string(); - let ns_clone = ns.to_string(); - let status = ing.status.clone(); - tokio::spawn(async move { - patch_ingress_status_address(&client_clone, &ns_clone, &name_clone, status.as_ref(), &external_ip).await; - }); + let key = format!("{}/{}", ns, name); + let external_ip = resolve_external_ip(client).await; + let status_already_correct = ing + .status + .as_ref() + .and_then(|s| s.load_balancer.as_ref()) + .and_then(|lb| lb.ingress.as_ref()) + .map(|list| list.iter().any(|i| i.ip.as_deref() == Some(&external_ip))) + .unwrap_or(false); + + if status_already_correct { + PATCHED_INGRESSES.insert(key, external_ip); + } else { + PATCHED_INGRESSES.insert(key.clone(), external_ip.clone()); + let client_clone = client.clone(); + let name_clone = name.to_string(); + let ns_clone = ns.to_string(); + let key_clone = key.clone(); + tokio::spawn(async move { + if let Err(e) = patch_ingress_status_address(&client_clone, &ns_clone, &name_clone, &external_ip).await { + error!("Failed to patch ingress status address {}/{}: {}", &ns_clone, &name_clone, e); + PATCHED_INGRESSES.remove(&key_clone); + } + }); + } + } + } else { + if let (Some(name), Some(ns)) = (ing.metadata.name.as_deref(), ing.metadata.namespace.as_deref()) { + PATCHED_INGRESSES.remove(&format!("{}/{}", ns, name)); } } if let Some(rules) = &spec.rules { for rule in rules { - let host = match &rule.host { - Some(h) => h.clone(), - None => "*".to_string(), - }; + let host = rule.host.clone().unwrap_or_else(|| "*".to_string()); if let Some(http) = &rule.http { for path_entry in &http.paths { let path = path_entry.path.clone().unwrap_or_else(|| "/".to_string()); if let Some(service) = &path_entry.backend.service { let service_name = service.name.clone(); - let mut update = AralezRouteUpdate { + let annot = parse_ingress_config(ing); + let update = AralezRouteUpdate { host: host.clone(), path, service_name, is_deleted, - rate_limit: None, - x4xx_limit: None, - sticky_sessions: None, - client_headers: None, - server_headers: None, + rate_limit: annot.rate_limit, + x4xx_limit: annot.x4xx_limit, + client_headers: annot.client_headers, + server_headers: annot.server_headers, + namespace: annot.namespace, }; - if let Some(annot) = parse_ingress_config(ing) { - update.rate_limit = annot.rate_limit; - update.x4xx_limit = annot.x4xx_limit; - update.sticky_sessions = annot.sticky_sessions; - update.client_headers = annot.client_headers; - update.server_headers = annot.server_headers; - } if let Err(e) = tx.send(update).await { error!("Failed to send route update to channel: {:?}", e); } @@ -160,54 +171,26 @@ pub async fn create_custom_k8s_client(api_server_url: &str, bearer_token: &str, Ok(client) } -fn parse_ingress_config(ing: &Ingress) -> Option { - if let Some(annotations) = &ing.metadata.annotations { - let mut clh: Option> = None; - let mut srh: Option> = None; +fn parse_ingress_config(ing: &Ingress) -> AralezAnnotation { + let namespace = ing.metadata.namespace.clone().unwrap_or_else(|| "default".to_string()); + let annotations = ing.metadata.annotations.as_ref(); - if let Some(ch) = annotations.get("aralez.rs/client_headers") { - let data: Vec = serde_json::from_str(ch).unwrap_or_default(); - clh = Some(data); - } - if let Some(sh) = annotations.get("aralez.rs/server_headers") { - let data: Vec = serde_json::from_str(sh).unwrap_or_default(); - srh = Some(data); - }; + let client_headers = annotations.and_then(|a| a.get("aralez.rs/client_headers")).and_then(|ch| serde_json::from_str(ch).ok()); - let annots = AralezAnnotation { - rate_limit: annotations.get("aralez.rs/rate_limit").and_then(|v| v.parse().ok()), - x4xx_limit: annotations.get("aralez.rs/x4xx_limit").and_then(|v| v.parse().ok()), - sticky_sessions: annotations.get("aralez.rs/sticky_sessions").and_then(|v| v.parse().ok()), - client_headers: clh, - server_headers: srh, - }; - return Some(annots); + let server_headers = annotations.and_then(|a| a.get("aralez.rs/server_headers")).and_then(|sh| serde_json::from_str(sh).ok()); + + AralezAnnotation { + rate_limit: annotations.and_then(|a| a.get("aralez.rs/rate_limit")).and_then(|v| v.parse().ok()), + x4xx_limit: annotations.and_then(|a| a.get("aralez.rs/x4xx_limit")).and_then(|v| v.parse().ok()), + client_headers, + server_headers, + namespace, } - None } -pub async fn patch_ingress_status_address( - client: &Client, - namespace: &str, - name: &str, - current_status: Option<&k8s_openapi::api::networking::v1::IngressStatus>, - external_ip: &str, -) { - if let Some(status) = current_status { - if let Some(lb) = &status.load_balancer { - if let Some(ing_list) = &lb.ingress { - if ing_list.iter().any(|i| i.ip.as_deref() == Some(external_ip)) { - return; - } - } - } - } - +pub async fn patch_ingress_status_address(client: &Client, namespace: &str, name: &str, external_ip: &str) -> Result<(), kube::Error> { let ingresses: Api = Api::namespaced(client.clone(), namespace); - let patch = json!({ - "apiVersion": "networking.k8s.io/v1", - "kind": "Ingress", "status": { "loadBalancer": { "ingress": [ @@ -218,12 +201,17 @@ pub async fn patch_ingress_status_address( } } }); + let params = PatchParams::default(); - let params = PatchParams::apply("aralez-ingress-controller").force(); - - match ingresses.patch_status(name, ¶ms, &Patch::Apply(patch)).await { - Ok(_) => info!("Patched Ingress {}/{} ADDRESS to {}", namespace, name, external_ip), - Err(e) => error!("Failed to patch status for Ingress {}/{}: {:?}", namespace, name, e), + match ingresses.patch_status(name, ¶ms, &Patch::Strategic(patch)).await { + Ok(_) => { + info!("Successfully patched Ingress {}/{} ADDRESS to {}", namespace, name, external_ip); + Ok(()) + } + Err(e) => { + error!("Failed to patch status for Ingress {}/{}: {:?}", namespace, name, e); + Err(e) + } } } @@ -269,3 +257,23 @@ pub async fn discover_controller_ip(client: &Client, namespace: &str, service_na } None } + +pub async fn resolve_external_ip(client: &Client) -> String { + if let Ok(ip) = std::env::var("ARALEZ_EXTERNAL_IP") { + let trimmed = ip.trim(); + if !trimmed.is_empty() { + return trimmed.to_string(); + } + } + let svc_name = std::env::var("ARALEZ_SERVICE_NAME").unwrap_or_else(|_| "aralez-service".to_string()); + let svc_ns = std::env::var("ARALEZ_SERVICE_NAMESPACE").unwrap_or_else(|_| "aralez".to_string()); + + if let Some(ip) = discover_controller_ip(client, &svc_ns, &svc_name).await { + return ip; + } + if let Some(ip) = discover_node_ip(client).await { + return ip; + } + + "127.0.0.1".to_string() +}