diff --git a/src/utils/ingress/consul.rs b/src/utils/ingress/consul.rs deleted file mode 100644 index c374137..0000000 --- a/src/utils/ingress/consul.rs +++ /dev/null @@ -1,91 +0,0 @@ -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 deleted file mode 100644 index fca615a..0000000 --- a/src/utils/ingress/discovery.rs +++ /dev/null @@ -1,67 +0,0 @@ -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 deleted file mode 100644 index 1fec46a..0000000 --- a/src/utils/ingress/kuberconsul.rs +++ /dev/null @@ -1,63 +0,0 @@ -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 deleted file mode 100644 index 2be6a3c..0000000 --- a/src/utils/ingress/kubernetes.rs +++ /dev/null @@ -1,163 +0,0 @@ -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 deleted file mode 100644 index 2807568..0000000 --- a/src/utils/ingress/kubewatcher.rs +++ /dev/null @@ -1,279 +0,0 @@ -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 deleted file mode 100644 index e69de29..0000000