From 1a2485b87a9e876c5b65fc6ed04efe216188fe6d Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Wed, 9 Sep 2026 18:42:41 +0200 Subject: [PATCH] Implemented EndpointSlices #55 --- src/utils/httpclient.rs | 41 +++++++++++++++++++------------- src/utils/kubernetes.rs | 52 +++++++++++++++++++++++++---------------- 2 files changed, 57 insertions(+), 36 deletions(-) diff --git a/src/utils/httpclient.rs b/src/utils/httpclient.rs index e9c017e..cb8b1c1 100644 --- a/src/utils/httpclient.rs +++ b/src/utils/httpclient.rs @@ -1,6 +1,6 @@ use crate::utils::consul::ConsulService; use crate::utils::kuberconsul::match_path; -use crate::utils::kubernetes::KubeEndpoints; +use crate::utils::kubernetes::KubeEndpointSliceList; use crate::utils::structs::{GlobalServiceMapping, InnerMap}; use ahash::HashMap; use dashmap::DashMap; @@ -36,7 +36,9 @@ pub struct ConsulServicesInternal { #[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); @@ -75,6 +77,7 @@ fn parse_services(json: Vec) -> Result { }) .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()?; @@ -103,18 +106,29 @@ pub async fn for_consul(url: &str, token: Option, conf: &GlobalServiceMa pub async fn for_kuber(url: &str, token: &str, conf: &GlobalServiceMapping) -> Option, (Vec>, AtomicUsize)>> { if let Some(data) = getfromapi(url, Some(token.to_string()), "kubernetes").await { - let endpoints: KubeEndpoints = serde_json::from_slice(&data).ok()?; + let slice_list: KubeEndpointSliceList = serde_json::from_slice(&data).ok()?; let upstreams: DashMap, (Vec>, AtomicUsize)> = DashMap::new(); let mut inner_vec = Vec::new(); - if let Some(subsets) = endpoints.subsets { - for subset in subsets { - if let (Some(addresses), Some(ports)) = (subset.addresses, subset.ports) { - for addr in addresses { - for port in &ports { + for slice in slice_list.items { + let ports = match &slice.ports { + Some(p) if !p.is_empty() => p, + _ => continue, + }; + + for ep in &slice.endpoints { + let is_ready = ep.conditions.as_ref().and_then(|c| c.ready).unwrap_or(true); + + if !is_ready { + continue; + } + + for addr in &ep.addresses { + for port in ports { + if let Some(port_num) = port.port { let to_add = Arc::from(InnerMap { - address: Arc::from(addr.ip.as_str()), - port: port.port, + address: Arc::from(addr.as_str()), + port: port_num, is_ssl: false, is_http2: false, to_https: conf.to_https.unwrap_or(false), @@ -130,6 +144,7 @@ pub async fn for_kuber(url: &str, token: &str, conf: &GlobalServiceMapping) -> O } } } + if !inner_vec.is_empty() { match_path(conf, &upstreams, inner_vec); return Some(upstreams); @@ -137,6 +152,7 @@ pub async fn for_kuber(url: &str, token: &str, conf: &GlobalServiceMapping) -> O } None } + pub async fn getfromapi(url: &str, token: Option, provider: &str) -> Option> { let (host, port, path, is_tls) = parse_url(&url).ok()?; @@ -226,10 +242,3 @@ 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/kubernetes.rs b/src/utils/kubernetes.rs index 5ccf813..83365af 100644 --- a/src/utils/kubernetes.rs +++ b/src/utils/kubernetes.rs @@ -15,25 +15,44 @@ use rustls::crypto::CryptoProvider; use tokio::sync::mpsc::Sender; #[derive(Debug, serde::Deserialize)] -pub struct KubeEndpoints { - pub subsets: Option>, +#[serde(rename_all = "camelCase")] +pub struct KubeEndpointSliceList { + #[serde(default)] + pub items: Vec, } #[derive(Debug, serde::Deserialize)] -pub struct KubeSubset { - pub addresses: Option>, +#[serde(rename_all = "camelCase")] +pub struct KubeEndpointSlice { pub ports: Option>, + #[serde(default)] + pub endpoints: Vec, } #[derive(Debug, serde::Deserialize)] -pub struct KubeAddress { - pub ip: String, +#[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, + pub serving: Option, + pub terminating: Option, +} + +#[derive(Debug, serde::Deserialize)] +#[serde(rename_all = "camelCase")] pub struct KubePort { - pub port: u16, + pub name: Option, + pub port: Option, + pub protocol: Option, } + pub struct KubernetesDiscovery; #[async_trait] @@ -52,7 +71,6 @@ 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 token = crate::utils::kuberconsul::read_token(path.as_str()).await; if CryptoProvider::get_default().is_none() { @@ -126,7 +144,12 @@ 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, update.namespace, update.service_name); + + 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() { @@ -142,14 +165,3 @@ 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() { - if let Ok(contents) = fs::read_to_string(ns_path) { - return Some(contents.trim().to_string()); - } - } - env::var("POD_NAMESPACE").ok() -} -*/