From 9de8c9ba97ebb7ee61d9c0a758a17211b97e5377 Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Fri, 25 Sep 2026 18:01:15 +0200 Subject: [PATCH] version change --- Cargo.toml | 2 +- src/utils/ingress/consul.rs | 91 ++++++++++ src/utils/ingress/discovery.rs | 67 ++++++++ src/utils/ingress/kuberconsul.rs | 63 +++++++ src/utils/ingress/kubernetes.rs | 163 ++++++++++++++++++ src/utils/ingress/kubewatcher.rs | 279 +++++++++++++++++++++++++++++++ src/utils/ingress/mod.rs | 0 7 files changed, 664 insertions(+), 1 deletion(-) create mode 100644 src/utils/ingress/consul.rs create mode 100644 src/utils/ingress/discovery.rs create mode 100644 src/utils/ingress/kuberconsul.rs create mode 100644 src/utils/ingress/kubernetes.rs create mode 100644 src/utils/ingress/kubewatcher.rs create mode 100644 src/utils/ingress/mod.rs diff --git a/Cargo.toml b/Cargo.toml index 546a80d..2df1cda 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "aralez" -version = "0.93.5" +version = "0.94.1" edition = "2021" license = "Apache-2.0" description = "Reverse proxy built on top of Cloudflare's Pingora" diff --git a/src/utils/ingress/consul.rs b/src/utils/ingress/consul.rs new file mode 100644 index 0000000..c374137 --- /dev/null +++ b/src/utils/ingress/consul.rs @@ -0,0 +1,91 @@ +use crate::utils::httpclient; +use crate::utils::kuberconsul::{list_to_upstreams, ServiceDiscovery}; +use crate::utils::parceyaml::build_headers; +use crate::utils::structs::{Configuration, GlobalServiceMapping, UpstreamsDashMap}; +use async_trait::async_trait; +use dashmap::DashMap; +use pingora::prelude::sleep; +use rand::RngExt; +use serde::Deserialize; +use std::env; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc::Sender; + +#[derive(Debug, Deserialize)] +pub struct ConsulService { + #[serde(rename = "Address")] + pub address: String, + #[serde(rename = "ServicePort")] + pub port: u16, +} +pub struct ConsulDiscovery; + +#[async_trait] +impl ServiceDiscovery for ConsulDiscovery { + async fn fetch_upstreams(&self, config: Arc, toreturn: Sender) { + loop { + let upstreams = UpstreamsDashMap::new(); + + if let Some(consul) = config.consul.clone() { + let servers = consul.servers.unwrap_or_else(|| { + vec![format!( + "{}:{}", + env::var("CONSUL_SERVICE_HOST").unwrap_or_else(|_| "0.0.0.0".to_string()), + env::var("CONSUL_SERVICE_PORT").unwrap_or_else(|_| "0".to_string()) + )] + }); + let end = servers.len().saturating_sub(1); + let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 }; + let consul_data = servers.get(num).unwrap().to_string(); + let ss = format!("{}/v1/catalog/service/", consul_data); + let dc = format!("{}/v1/catalog/services?filter=%22aralez.service=yes%22+in+ServiceTags", consul_data); + + if let Some(infolist) = httpclient::for_consul_list(dc.as_str(), consul.token.clone()).await { + for (k, v) in infolist { + let client_header_map: DashMap, Vec<(String, Arc)>> = DashMap::new(); + let mut client_header_list = Vec::new(); + build_headers(&v.client_headers, config.as_ref(), &mut client_header_list); + + let sever_header_map: DashMap, Vec<(String, Arc)>> = DashMap::new(); + let mut server_header_list = Vec::new(); + build_headers(&v.server_headers, config.as_ref(), &mut server_header_list); + + if !client_header_list.is_empty() { + let path_key = v.path.clone(); + client_header_map.insert(Arc::from(path_key), client_header_list); + config.client_headers.insert(Arc::from(v.host.clone()), client_header_map); + } + if !server_header_list.is_empty() { + let path_key = v.path.clone(); + sever_header_map.insert(Arc::from(path_key), server_header_list); + config.server_headers.insert(Arc::from(v.host.clone()), sever_header_map); + } + let pref = format!("{}{}", ss, k); + let gsm = &GlobalServiceMapping { + upstream: k.clone(), + hostname: v.host.clone(), + path: Some(v.path), + to_https: v.to_https, + redirect_to: v.redirect, + authorization: v.auth, + rate_limit: v.rate, + x4xx_limit: v.xrate, + client_headers: v.client_headers, + server_headers: v.server_headers, + }; + + let list = httpclient::for_consul(pref.as_str(), consul.token.clone(), gsm).await; + list_to_upstreams(list, &upstreams, gsm); + } + } + } + + if let Some(lt) = crate::utils::kuberconsul::clone_compare(&upstreams, &config).await { + let _ = toreturn.send(lt).await; + } + + sleep(Duration::from_secs(5)).await; + } + } +} diff --git a/src/utils/ingress/discovery.rs b/src/utils/ingress/discovery.rs new file mode 100644 index 0000000..fca615a --- /dev/null +++ b/src/utils/ingress/discovery.rs @@ -0,0 +1,67 @@ +use crate::utils::consul::ConsulDiscovery; +use crate::utils::filewatch; +use crate::utils::kuberconsul::ServiceDiscovery; +use crate::utils::kubernetes::KubernetesDiscovery; +use crate::utils::structs::{Configuration, UpstreamsDashMap}; +use crate::web::webserver; +use async_trait::async_trait; +use std::sync::Arc; +use tokio::sync::mpsc::Sender; + +pub struct APIUpstreamProvider { + pub config_api_enabled: bool, + pub address: String, + pub masterkey: Option, + pub certs_dir: String, + pub config_dir: String, + pub upstreams_file: String, + pub file_server_address: Option, + pub file_server_folder: Option, + pub current_upstreams: Arc, + pub full_upstreams: Arc, +} + +pub struct FromFileProvider { + pub path: String, +} + +pub struct ConsulProvider { + pub config: Arc, +} + +pub struct KubernetesProvider { + pub config: Arc, +} + +#[async_trait] +pub trait Discovery { + async fn start(&self, tx: Sender); +} + +#[async_trait] +impl Discovery for APIUpstreamProvider { + async fn start(&self, toreturn: Sender) { + webserver::run_server(self, toreturn, self.current_upstreams.clone(), self.full_upstreams.clone()).await; + } +} + +#[async_trait] +impl Discovery for FromFileProvider { + async fn start(&self, tx: Sender) { + tokio::spawn(filewatch::start(self.path.clone(), tx)); + } +} + +#[async_trait] +impl Discovery for ConsulProvider { + async fn start(&self, tx: Sender) { + tokio::spawn(ConsulDiscovery.fetch_upstreams(self.config.clone(), tx)); + } +} + +#[async_trait] +impl Discovery for KubernetesProvider { + async fn start(&self, tx: Sender) { + tokio::spawn(KubernetesDiscovery.fetch_upstreams(self.config.clone(), tx)); + } +} diff --git a/src/utils/ingress/kuberconsul.rs b/src/utils/ingress/kuberconsul.rs new file mode 100644 index 0000000..1fec46a --- /dev/null +++ b/src/utils/ingress/kuberconsul.rs @@ -0,0 +1,63 @@ +use crate::utils::structs::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap}; +use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams}; +use async_trait::async_trait; +use dashmap::DashMap; +use std::sync::atomic::AtomicUsize; +use std::sync::Arc; +use tokio::fs::File; +use tokio::io::AsyncReadExt; +use tokio::sync::mpsc::Sender; + +#[allow(clippy::type_complexity)] +pub fn list_to_upstreams(lt: Option, (Vec>, AtomicUsize)>>, upstreams: &UpstreamsDashMap, i: &GlobalServiceMapping) { + if let Some(list) = lt { + let key: Arc = Arc::from(i.hostname.as_str()); + match upstreams.get(&key) { + Some(upstr) => { + for (k, v) in list { + upstr.value().insert(k, v); + } + } + None => { + upstreams.insert(key, list); + } + }; + } +} + +pub fn match_path(conf: &GlobalServiceMapping, upstreams: &DashMap, (Vec>, AtomicUsize)>, values: Vec>) { + let path = conf.path.as_deref().unwrap_or("/"); + upstreams.insert(Arc::from(path), (values, AtomicUsize::new(0))); +} + +pub async fn read_token(path: &str) -> String { + let mut file = File::open(path).await.unwrap(); + let mut contents = String::new(); + file.read_to_string(&mut contents).await.unwrap(); + contents.trim().to_string() +} + +#[async_trait] +pub trait ServiceDiscovery { + async fn fetch_upstreams(&self, config: Arc, toreturn: Sender); +} + +pub async fn clone_compare(upstreams: &UpstreamsDashMap, config: &Arc) -> Option { + if !compare_dashmaps(upstreams, &config.upstreams) { + let tosend = Configuration { + upstreams: Default::default(), + client_headers: config.client_headers.clone(), + server_headers: config.server_headers.clone(), + consul: config.consul.clone(), + kubernetes: config.kubernetes.clone(), + typecfg: config.typecfg.clone(), + extraparams: config.extraparams.clone(), + }; + clone_dashmap_into(upstreams, &config.upstreams); + clone_dashmap_into(upstreams, &tosend.upstreams); + + print_upstreams(&tosend.upstreams, &tosend.extraparams); + return Some(tosend); + } + None +} diff --git a/src/utils/ingress/kubernetes.rs b/src/utils/ingress/kubernetes.rs new file mode 100644 index 0000000..2be6a3c --- /dev/null +++ b/src/utils/ingress/kubernetes.rs @@ -0,0 +1,163 @@ +use crate::utils::httpclient; +use crate::utils::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate}; +use crate::utils::parceyaml::build_headers; +use crate::utils::structs::{Configuration, GlobalServiceMapping, UpstreamsDashMap}; +use async_trait::async_trait; +use dashmap::DashMap; +use log::error; +use rand::RngExt; +use std::env; +use std::sync::Arc; + +use crate::utils::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery}; +use rustls::crypto::ring::default_provider; +use rustls::crypto::CryptoProvider; +use tokio::sync::mpsc::Sender; + +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct KubeEndpointSliceList { + #[serde(default)] + pub items: Vec, +} + +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct KubeEndpointSlice { + pub ports: Option>, + #[serde(default)] + pub endpoints: Vec, +} + +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct KubeEndpoint { + #[serde(default)] + pub addresses: Vec, + pub conditions: Option, +} + +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct KubeConditions { + pub ready: Option, +} + +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct KubePort { + pub port: Option, +} + +pub struct KubernetesDiscovery; + +#[async_trait] +impl ServiceDiscovery for KubernetesDiscovery { + async fn fetch_upstreams(&self, config: Arc, toreturn: Sender) { + if let Some(kuber) = config.kubernetes.clone() { + let servers = kuber.servers.unwrap_or_else(|| { + vec![format!( + "{}:{}", + env::var("KUBERNETES_SERVICE_HOST").unwrap_or_else(|_| "0.0.0.0".to_string()), + env::var("KUBERNETES_SERVICE_PORT_HTTPS").unwrap_or_else(|_| "0".to_string()) + )] + }); + + let end = servers.len().saturating_sub(1); + 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 token = crate::utils::kuberconsul::read_token(path.as_str()).await; + + if CryptoProvider::get_default().is_none() { + default_provider().install_default().expect("Failed to install rustls crypto provider"); + } + + let (tx, mut rx) = tokio::sync::mpsc::channel::(100); + let srv = server.clone(); + let tok = token.clone(); + tokio::spawn(async move { + if let Err(e) = start_ingress_watcher_with_config(srv.as_str(), tok.as_str(), tx, "aralez").await { + error!("Fatal watcher error: {:?}", e); + } + }); + + let upstreams = UpstreamsDashMap::new(); + + while let Some(update) = rx.recv().await { + let host_key: Arc = Arc::from(update.host.as_str()); + + if update.is_deleted { + upstreams.remove(&host_key); + config.client_headers.remove(&host_key); + } else { + let service = GlobalServiceMapping { + upstream: update.service_name.clone(), + hostname: update.host.clone(), + path: Some(update.path.clone()), + to_https: None, + redirect_to: None, + authorization: None, + rate_limit: update.rate_limit, + x4xx_limit: update.x4xx_limit, + client_headers: None, + server_headers: None, + }; + + let cheader_list: DashMap, Vec<(String, Arc)>> = DashMap::new(); + let sheader_list: DashMap, Vec<(String, Arc)>> = DashMap::new(); + + let mut client_headers_list = Vec::new(); + let mut server_headers_list = Vec::new(); + + let mut chl: Option> = None; + let mut shl: Option> = None; + + if let Some(hdr) = service.client_headers.as_ref() { + chl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned()); + } + if let Some(hdr) = service.server_headers.as_ref() { + shl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned()); + } + + if let Some(ch) = update.client_headers { + chl.get_or_insert_with(Vec::new).extend(ch); + } + if let Some(hdr) = update.server_headers.as_ref() { + shl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned()); + } + + build_headers(&chl, config.as_ref(), &mut client_headers_list); + if !client_headers_list.is_empty() { + let path_key = service.path.as_deref().unwrap_or("/"); + cheader_list.insert(Arc::from(path_key), client_headers_list); + config.client_headers.insert(host_key.clone(), cheader_list); + } + + build_headers(&shl, config.as_ref(), &mut server_headers_list); + if !server_headers_list.is_empty() { + let path_key = service.path.as_deref().unwrap_or("/"); + sheader_list.insert(Arc::from(path_key), server_headers_list); + config.server_headers.insert(host_key.clone(), sheader_list); + } + + let url = format!( + "https://{}/apis/discovery.k8s.io/v1/namespaces/{}/endpointslices?labelSelector=kubernetes.io%2Fservice-name%3D{}", + server, update.namespace, update.service_name + ); + + let list = httpclient::for_kuber(&url, &token, &service).await; + + if list.is_none() { + error!("Endpoint URL: {} returned empty/invalid response", url); + } + + list_to_upstreams(list, &upstreams, &service); + } + if let Some(lt) = clone_compare(&upstreams, &config).await { + let _ = toreturn.send(lt).await; + } + } + } + } +} diff --git a/src/utils/ingress/kubewatcher.rs b/src/utils/ingress/kubewatcher.rs new file mode 100644 index 0000000..2807568 --- /dev/null +++ b/src/utils/ingress/kubewatcher.rs @@ -0,0 +1,279 @@ +use dashmap::DashMap; +use futures::StreamExt; +use hyper::Uri; +use k8s_openapi::api::core::v1::{Node, Service}; +use k8s_openapi::api::networking::v1::Ingress; +use kube::api::{Patch, PatchParams}; +use kube::{ + api::Api, + client::Client, + config::{AuthInfo, Config}, + runtime::watcher::{self, Event}, +}; +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 client_headers: Option>, + pub server_headers: Option>, +} + +#[derive(Debug, Clone)] +struct AralezAnnotation { + rate_limit: Option, + x4xx_limit: 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, + tx: mpsc::Sender, + target_class: &'static str, +) -> Result<(), Box> { + let client = create_custom_k8s_client(api_server, token, true).await?; + let ingresses: Api = Api::all(client.clone()); + + let watcher_config = watcher::Config::default().any_semantic(); + let mut stream = watcher::watcher(ingresses, watcher_config).boxed(); + + while let Some(event) = stream.next().await { + match event { + Ok(Event::Apply(ing)) => { + process_ingress_change(&client, &ing, target_class, false, &tx).await; + } + Ok(Event::Delete(ing)) => { + process_ingress_change(&client, &ing, target_class, true, &tx).await; + } + Ok(Event::InitApply(ing)) => { + process_ingress_change(&client, &ing, target_class, false, &tx).await; + } + Ok(Event::InitDone) => { + info!("Ingress watcher initial sync complete"); + } + Err(err) => { + error!("Kubernetes Ingress watch error: {:?}", err); + } + _ => {} + } + } + Ok(()) +} + +async fn process_ingress_change(client: &Client, ing: &Ingress, target_class: &str, is_deleted: bool, tx: &mpsc::Sender) { + let spec = match &ing.spec { + Some(s) => s, + None => return, + }; + let class_name = spec.ingress_class_name.as_deref().unwrap_or_default(); + if class_name != target_class { + return; + } + + if !is_deleted { + if let (Some(name), Some(ns)) = (ing.metadata.name.as_deref(), ing.metadata.namespace.as_deref()) { + 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 = 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 annot = parse_ingress_config(ing); + let update = AralezRouteUpdate { + host: host.clone(), + path, + service_name, + is_deleted, + 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 Err(e) = tx.send(update).await { + error!("Failed to send route update to channel: {:?}", e); + } + } + } + } + } + } +} + +pub async fn create_custom_k8s_client(api_server_url: &str, bearer_token: &str, insecure_tls: bool) -> Result> { + let full_url = if !api_server_url.starts_with("http://") && !api_server_url.starts_with("https://") { + format!("https://{}", api_server_url) + } else { + api_server_url.to_string() + }; + let uri: Uri = full_url.parse()?; + let mut config = Config::new(uri); + config.auth_info = AuthInfo { + token: Some(bearer_token.trim().to_string().into()), + ..Default::default() + }; + config.connect_timeout = Some(Duration::from_secs(10)); + config.read_timeout = None; + if insecure_tls { + config.accept_invalid_certs = true; + } + let client = Client::try_from(config)?; + Ok(client) +} + +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(); + + let client_headers = annotations.and_then(|a| a.get("aralez.rs/client_headers")).and_then(|ch| serde_json::from_str(ch).ok()); + + 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, + } +} + +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!({ + "status": { + "loadBalancer": { + "ingress": [ + { + "ip": external_ip + } + ] + } + } + }); + let params = PatchParams::default(); + + 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) + } + } +} + +pub async fn discover_node_ip(client: &Client) -> Option { + let nodes: Api = Api::all(client.clone()); + if let Ok(node_list) = nodes.list(&Default::default()).await { + for node in node_list { + if let Some(status) = node.status { + if let Some(addresses) = status.addresses { + for addr in &addresses { + if addr.type_ == "ExternalIP" { + return Some(addr.address.clone()); + } + } + for addr in &addresses { + if addr.type_ == "InternalIP" { + return Some(addr.address.clone()); + } + } + } + } + } + } + None +} +pub async fn discover_controller_ip(client: &Client, namespace: &str, service_name: &str) -> Option { + let services: Api = Api::namespaced(client.clone(), namespace); + if let Ok(svc) = services.get(service_name).await { + if let Some(status) = svc.status { + if let Some(lb) = status.load_balancer { + if let Some(ingress_list) = lb.ingress { + for ing in ingress_list { + if let Some(ip) = ing.ip { + return Some(ip); + } + if let Some(hostname) = ing.hostname { + return Some(hostname); + } + } + } + } + } + } + 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() +} diff --git a/src/utils/ingress/mod.rs b/src/utils/ingress/mod.rs new file mode 100644 index 0000000..e69de29