diff --git a/Cargo.lock b/Cargo.lock index 9fb1db5..aefe94b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -138,7 +138,7 @@ checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "aralez" -version = "0.92.19" +version = "0.93.1" dependencies = [ "ahash", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index 162cabe..e6f46fd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "aralez" -version = "0.92.19" +version = "0.93.1" edition = "2021" license = "Apache-2.0" description = "Reverse proxy built on top of Cloudflare's Pingora" diff --git a/etc/consul/apache.hcl b/etc/consul/apache.hcl new file mode 100644 index 0000000..6351ca6 --- /dev/null +++ b/etc/consul/apache.hcl @@ -0,0 +1,67 @@ +job "apache" { + meta { + uuid = uuidv4() + } + datacenters = ["eu-ams-1"] + type = "service" + constraint { + attribute = node.unique.name + operator = "regexp" + value = "consul[0-9]" + } + update { + max_parallel = 1 + min_healthy_time = "10s" + healthy_deadline = "3m" + progress_deadline = "10m" + auto_revert = false + canary = 0 + } + migrate { + max_parallel = 1 + health_check = "checks" + min_healthy_time = "10s" + healthy_deadline = "5m" + } + group "Apache" { + network { + port "apache" { to = 80 } + } + constraint { + operator = "distinct_hosts" + value = "true" + } + count = 3 + restart { + attempts = 2 + interval = "30m" + delay = "15s" + mode = "fail" + } + task "health" { + service { + port = "apache" + tags = [ + "aralez.service=yes", + "aralez.host=apache.blablabla.com", + "aralez.path=/" + ] + check { + type = "http" + port = "apache" + path = "/" + interval = "5s" + timeout = "2s" + } + } + + driver = "docker" + + config { + image = "httpd:latest" + force_pull = true + ports = ["apache"] + } + } + } +} diff --git a/etc/consul/echo.hcl b/etc/consul/echo.hcl new file mode 100644 index 0000000..d2c486c --- /dev/null +++ b/etc/consul/echo.hcl @@ -0,0 +1,72 @@ +job "echo" { + meta { + uuid = uuidv4() + } + datacenters = ["eu-ams-1"] + type = "service" + constraint { + attribute = node.unique.name + operator = "regexp" + value = "consul[0-9]" + } + update { + max_parallel = 1 + min_healthy_time = "10s" + healthy_deadline = "3m" + progress_deadline = "10m" + auto_revert = false + canary = 0 + } + migrate { + max_parallel = 1 + health_check = "checks" + min_healthy_time = "10s" + healthy_deadline = "5m" + } + group "Echo" { + network { + port "echo" { to = 80 } + } + constraint { + operator = "distinct_hosts" + value = "true" + } + count = 3 + restart { + attempts = 2 + interval = "30m" + delay = "15s" + mode = "fail" + } + task "server" { + service { + port = "echo" + tags = [ + "aralez.service=yes", + "aralez.host=echo.blablabla.com", + "aralez.path=/", + "aralez.rate=100", + "aralez.4xx_rate=50", + "aralez.to_https=false", + "aralez.client_headers=WyJYLUVjaG8tU2VuZDpIb29vb29vb29vb29vIiwiWC1FY2hvLVJlY3Y6WW9vb29vb29vb29vbyJdCg==", + "aralez.server_headers=WyJYLVByb3h5LUZyb206IEFyYWxleiIsIlgtQmFyaS1MdXlzOiBBcGVyIEphbiIsIlgtQW5rYXAtSGVhZGVyOiBCZXNhbXAgQW5rYXAgSGVhZGVyIl0K" + ] + check { + type = "http" + port = "echo" + path = "/" + interval = "5s" + timeout = "2s" + } + } + + driver = "docker" + + config { + image = "ealen/echo-server:latest" + force_pull = true + ports = ["echo"] + } + } + } +} \ No newline at end of file diff --git a/etc/consul/nginx.hcl b/etc/consul/nginx.hcl new file mode 100644 index 0000000..1f57769 --- /dev/null +++ b/etc/consul/nginx.hcl @@ -0,0 +1,73 @@ +job "nginx" { + meta { + uuid = uuidv4() + } + datacenters = ["eu-ams-1"] + type = "service" + constraint { + attribute = node.unique.name + operator = "regexp" + value = "consul[0-9]" + } + update { + max_parallel = 1 + min_healthy_time = "10s" + healthy_deadline = "3m" + progress_deadline = "10m" + auto_revert = false + canary = 0 + } + migrate { + max_parallel = 1 + health_check = "checks" + min_healthy_time = "10s" + healthy_deadline = "5m" + } + group "NginX" { + network { + port "nginx" { to = 80 } + } + constraint { + operator = "distinct_hosts" + value = "true" + } + count = 3 + restart { + attempts = 2 + interval = "30m" + delay = "15s" + mode = "fail" + } + task "server" { + service { + port = "nginx" + tags = [ + "aralez.service=yes", + "aralez.host=nginx.blablabla.com", + "aralez.path=/", + "aralez.rate=20", + "aralez.4xx_rate=10", + "aralez.to_https=true", + "aralez.auth=basic root:toor", + "aralez.client_headers=WwogICAgIlgtU29tZS1IZWFkZXI6QUFBQUFBQUFBQUFBQUFBIiwKICAgICJYLUVtb3MtSGVhZGVyOlpaWlpaWlpaWlpaWlpaWiIsCiAgICAiWC1PdGhlci1oZWFkZXI6VmF5VmF5VmF5IiwKICAgICJYLVJlaHRvLWhlYWRlcjpaYXlaYXlaYXkiCl0KCg==", + "aralez.server_headers=WyJYLVByb3h5LUZyb206IEFyYWxleiIsIlgtQmFyaS1MdXlzOiBBcGVyIEphbiIsIlgtQW5rYXAtSGVhZGVyOiBCZXNhbXAgQW5rYXAgSGVhZGVyIl0K" + ] + check { + type = "http" + port = "nginx" + path = "/" + interval = "5s" + timeout = "2s" + } + } + + driver = "docker" + + config { + image = "nginx:latest" + force_pull = true + ports = ["nginx"] + } + } + } +} \ No newline at end of file diff --git a/src/utils.rs b/src/utils.rs index 35b3bb4..5ae7492 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,4 +1,5 @@ pub mod auth; +pub mod consul; pub mod discovery; mod filewatch; pub mod fordebug; @@ -8,6 +9,7 @@ pub mod healthcheck; pub mod httpclient; pub mod jwt; pub mod kuberconsul; +pub mod kubernetes; pub mod kubewatcher; pub mod lazylock; pub mod metrics; diff --git a/src/utils/consul.rs b/src/utils/consul.rs new file mode 100644 index 0000000..c374137 --- /dev/null +++ b/src/utils/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/discovery.rs b/src/utils/discovery.rs index 4e873de..fca615a 100644 --- a/src/utils/discovery.rs +++ b/src/utils/discovery.rs @@ -1,5 +1,7 @@ +use crate::utils::consul::ConsulDiscovery; use crate::utils::filewatch; -use crate::utils::kuberconsul::{ConsulDiscovery, KubernetesDiscovery, ServiceDiscovery}; +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; diff --git a/src/utils/httpclient.rs b/src/utils/httpclient.rs index 26f2623..f09214c 100644 --- a/src/utils/httpclient.rs +++ b/src/utils/httpclient.rs @@ -1,33 +1,117 @@ -use crate::utils::kuberconsul::{match_path, ConsulService, KubeEndpoints}; +use crate::utils::consul::ConsulService; +use crate::utils::kuberconsul::match_path; +use crate::utils::kubernetes::KubeEndpoints; use crate::utils::structs::{GlobalServiceMapping, InnerMap}; +use ahash::HashMap; +use base64::{engine::general_purpose, Engine as _}; use dashmap::DashMap; +use log::warn; use pingora_core::connectors::http::Connector; use pingora_core::listeners::ALPN; use pingora_core::prelude::HttpPeer; use pingora_http::RequestHeader; +use serde::{Deserialize, Serialize}; use std::sync::atomic::AtomicUsize; use std::sync::{Arc, LazyLock}; use std::time::Duration; pub static CONNECTOR: LazyLock = LazyLock::new(|| Connector::new(None)); +#[derive(Debug, Default, Serialize, Deserialize, Clone)] +pub struct ConsulServicesInternal { + #[serde(rename = "aralez.host")] + pub host: String, + #[serde(rename = "aralez.path")] + pub path: String, + #[serde(rename = "aralez.auth")] + pub auth: Option, + #[serde(rename = "aralez.redirect")] + pub redirect: Option, + #[serde(rename = "aralez.rate")] + pub rate: Option, + #[serde(rename = "aralez.4xx_rate")] + pub xrate: Option, + #[serde(rename = "aralez.client_headers")] + pub client_headers: Option>, + #[serde(rename = "aralez.server_headers")] + pub server_headers: Option>, + #[serde(rename = "aralez.to_https")] + pub to_https: Option, +} +pub type ConsulServices = HashMap; +pub async fn for_consul_list(url: &str, token: Option) -> Option { + if let Some(data) = getfromapi(url, token, "consul").await { + let yo = parse_services(data); + if let Ok(y) = yo { + return Some(y); + } + return None; + } + None +} + +fn parse_services(json: Vec) -> Result { + let raw: HashMap> = serde_json::from_slice(&json)?; + Ok(raw + .into_iter() + .map(|(service, tags)| { + let mut internal = ConsulServicesInternal::default(); + for tag in tags { + if let Some((key, value)) = tag.split_once('=') { + match key { + "aralez.host" => internal.host = value.to_string(), + "aralez.path" => internal.path = value.to_string(), + "aralez.rate" => internal.rate = value.parse::().ok(), + "aralez.4xx_rate" => internal.xrate = value.parse::().ok(), + "aralez.to_https" => internal.to_https = value.parse::().ok(), + "aralez.auth" => internal.auth = Option::from(value.to_string()), + "aralez.redirect" => internal.redirect = Option::from(value.to_string()), + "aralez.client_headers" => { + let decoded_bytes = general_purpose::STANDARD.decode(value).unwrap_or_default(); + let decoded: Result, serde_json::Error> = serde_json::from_slice(&decoded_bytes); + match decoded { + Ok(dd) => internal.client_headers = Some(dd), + Err(er) => { + warn!("Could not decode client headers: {}", er); + internal.client_headers = None + } + } + } + "aralez.server_headers" => { + let decoded_bytes = general_purpose::STANDARD.decode(value).unwrap_or_default(); + let decoded: Result, serde_json::Error> = serde_json::from_slice(&decoded_bytes); + match decoded { + Ok(dd) => internal.server_headers = Some(dd), + Err(er) => { + warn!("Could not decode server headers: {}", er); + internal.server_headers = None + } + } + } + _ => {} + } + } + } + + (service, internal) + }) + .collect()) +} pub async fn for_consul(url: &str, token: Option, conf: &GlobalServiceMapping) -> Option, (Vec>, AtomicUsize)>> { if let Some(data) = getfromapi(url, token, "consul").await { let endpoints: Vec = serde_json::from_slice(&data).ok()?; let mut inner_vec = Vec::new(); let upstreams: DashMap, (Vec>, AtomicUsize)> = DashMap::new(); for subsets in endpoints { - let addr = subsets.tagged_addresses.get("lan_ipv4").unwrap().address.clone(); - let prt = subsets.tagged_addresses.get("lan_ipv4").unwrap().port; let to_add = Arc::from(InnerMap { - address: Arc::from(&*addr), - port: prt, + address: Arc::from(&*subsets.address), + port: subsets.port, is_ssl: false, is_http2: false, to_https: conf.to_https.unwrap_or(false), rate_limit: conf.rate_limit, x4xx_limit: conf.x4xx_limit, - redirect_to: None, + redirect_to: conf.redirect_to.clone().map(Arc::::from), healthcheck: None, authorization: None, }); @@ -164,3 +248,10 @@ fn parse_url(url: &str) -> Result<(&str, u16, &str, bool), &'static str> { Ok((host, port, uri, is_https)) } + +// fn base64_decode(encoded: &str) -> Option { +// let decoded_bytes = general_purpose::STANDARD.decode(encoded).unwrap_or_default(); +// let decoded: Result, serde_json::Error> = serde_json::from_slice(&decoded_bytes); +// println!("Decoded: {:?}", decoded); +// None +// } diff --git a/src/utils/kuberconsul.rs b/src/utils/kuberconsul.rs index 028abc0..1fec46a 100644 --- a/src/utils/kuberconsul.rs +++ b/src/utils/kuberconsul.rs @@ -1,63 +1,13 @@ -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, InnerMap, UpstreamsDashMap}; use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams}; use async_trait::async_trait; use dashmap::DashMap; -use log::error; -use pingora::prelude::sleep; -use rand::RngExt; -use serde::Deserialize; -use std::collections::HashMap; -use std::env; -use std::fs; -use std::path::Path; use std::sync::atomic::AtomicUsize; use std::sync::Arc; -use std::time::Duration; use tokio::fs::File; use tokio::io::AsyncReadExt; - -use rustls::crypto::ring::default_provider; -use rustls::crypto::CryptoProvider; use tokio::sync::mpsc::Sender; -#[derive(Debug, serde::Deserialize)] -pub struct KubeEndpoints { - pub subsets: Option>, -} - -#[derive(Debug, serde::Deserialize)] -pub struct KubeSubset { - pub addresses: Option>, - pub ports: Option>, -} - -#[derive(Debug, serde::Deserialize)] -pub struct KubeAddress { - pub ip: String, -} - -#[derive(Debug, serde::Deserialize)] -pub struct KubePort { - pub port: u16, -} - -#[derive(Debug, Deserialize)] -pub struct ConsulService { - #[serde(rename = "ServiceTaggedAddresses")] - pub tagged_addresses: HashMap, -} - -#[derive(Debug, Deserialize)] -pub struct ConsulTaggedAddress { - #[serde(rename = "Address")] - pub address: String, - #[serde(rename = "Port")] - pub port: u16, -} - #[allow(clippy::type_complexity)] pub fn list_to_upstreams(lt: Option, (Vec>, AtomicUsize)>>, upstreams: &UpstreamsDashMap, i: &GlobalServiceMapping) { if let Some(list) = lt { @@ -80,7 +30,7 @@ pub fn match_path(conf: &GlobalServiceMapping, upstreams: &DashMap, (Ve upstreams.insert(Arc::from(path), (values, AtomicUsize::new(0))); } -async fn read_token(path: &str) -> String { +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(); @@ -92,172 +42,7 @@ pub trait ServiceDiscovery { async fn fetch_upstreams(&self, config: Arc, toreturn: Sender); } -pub struct KubernetesDiscovery; -pub struct ConsulDiscovery; - -#[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 namespace = get_current_namespace().unwrap_or_else(|| "default".to_string()); - let token = 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, - sticky_sessions: update.sticky_sessions, - 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://{}/api/v1/namespaces/{}/endpoints/{}", server, 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; - } - } - } - } -} - -fn get_current_namespace() -> Option { - let ns_path = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"; - if Path::new(ns_path).exists() { - if let Ok(contents) = fs::read_to_string(ns_path) { - return Some(contents.trim().to_string()); - } - } - env::var("POD_NAMESPACE").ok() -} - -#[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); - - if let Some(svc) = consul.services { - for i in &svc { - let header_list = DashMap::new(); - let mut hl = Vec::new(); - build_headers(&i.client_headers, config.as_ref(), &mut hl); - if !hl.is_empty() { - let path_key = i.path.as_deref().unwrap_or("/"); - header_list.insert(Arc::from(path_key), hl); - config.client_headers.insert(Arc::from(i.hostname.clone()), header_list); - } - - let pref = format!("{}{}", ss, i.upstream); - let list = httpclient::for_consul(pref.as_str(), consul.token.clone(), i).await; - list_to_upstreams(list, &upstreams, i); - } - } - } - - if let Some(lt) = clone_compare(&upstreams, &config).await { - let _ = toreturn.send(lt).await; - } - - sleep(Duration::from_secs(5)).await; - } - } -} - -async fn clone_compare(upstreams: &UpstreamsDashMap, config: &Arc) -> Option { +pub async fn clone_compare(upstreams: &UpstreamsDashMap, config: &Arc) -> Option { if !compare_dashmaps(upstreams, &config.upstreams) { let tosend = Configuration { upstreams: Default::default(), diff --git a/src/utils/kubernetes.rs b/src/utils/kubernetes.rs new file mode 100644 index 0000000..eeb3640 --- /dev/null +++ b/src/utils/kubernetes.rs @@ -0,0 +1,157 @@ +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::fs; +use std::path::Path; +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)] +pub struct KubeEndpoints { + pub subsets: Option>, +} + +#[derive(Debug, serde::Deserialize)] +pub struct KubeSubset { + pub addresses: Option>, + pub ports: Option>, +} + +#[derive(Debug, serde::Deserialize)] +pub struct KubeAddress { + pub ip: String, +} + +#[derive(Debug, serde::Deserialize)] +pub struct KubePort { + pub port: u16, +} +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 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() { + 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, + // sticky_sessions: update.sticky_sessions, + 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://{}/api/v1/namespaces/{}/endpoints/{}", server, 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; + } + } + } + } +} + +fn get_current_namespace() -> Option { + let ns_path = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"; + if Path::new(ns_path).exists() { + if let Ok(contents) = fs::read_to_string(ns_path) { + return Some(contents.trim().to_string()); + } + } + env::var("POD_NAMESPACE").ok() +} diff --git a/src/utils/state.rs b/src/utils/state.rs index 4c9b16e..da0bf73 100644 --- a/src/utils/state.rs +++ b/src/utils/state.rs @@ -16,14 +16,3 @@ pub fn is_first_run() -> bool { let state = GLOBAL_STATE.read().unwrap(); state.first_run } - -/* -impl SharedState { - pub fn mark_first_run(&mut self) { - self.first_run = false; - } - pub fn is_first_run(&self) -> bool { - self.first_run - } -} -*/ diff --git a/src/utils/structs.rs b/src/utils/structs.rs index c369712..1ec6b1b 100644 --- a/src/utils/structs.rs +++ b/src/utils/structs.rs @@ -20,11 +20,12 @@ pub struct Extraparams { #[derive(Debug, Default, Clone, Serialize, Deserialize)] pub struct GlobalServiceMapping { + pub redirect_to: Option, + pub authorization: Option, pub upstream: String, pub hostname: String, pub path: Option, pub to_https: Option, - pub sticky_sessions: Option, pub rate_limit: Option, pub x4xx_limit: Option, pub client_headers: Option>, @@ -34,14 +35,12 @@ pub struct GlobalServiceMapping { #[derive(Clone, Default, Debug, Serialize, Deserialize)] pub struct Kubernetes { pub servers: Option>, - pub services: Option>, pub tokenpath: Option, } #[derive(Clone, Default, Debug, Serialize, Deserialize)] pub struct Consul { pub servers: Option>, - pub services: Option>, pub token: Option, } #[derive(Debug, Default, Serialize, Deserialize)]