From 012505b77e95c3fcb73786f61890dd440e0d3132 Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Fri, 24 Oct 2025 15:27:15 +0200 Subject: [PATCH] Cleaning up the code --- src/utils/auth.rs | 10 --------- src/utils/consul.rs | 39 ++++++++--------------------------- src/utils/kuber.rs | 44 +++++++++------------------------------- src/utils/kuberconsul.rs | 35 ++++++++++++++++++++++++++++++++ 4 files changed, 54 insertions(+), 74 deletions(-) diff --git a/src/utils/auth.rs b/src/utils/auth.rs index d6cc600..adcdacf 100644 --- a/src/utils/auth.rs +++ b/src/utils/auth.rs @@ -40,16 +40,6 @@ impl AuthValidator for JwtAuth<'_> { if let Some(tok) = get_query_param(session, "araleztoken") { return check_jwt(tok.as_str(), jwtsecret); } - - // if let Some(header) = session.get_header("authorization") { - // let h = header.to_str().ok().unwrap().split(" ").collect::>(); - // match h.len() { - // n => { - // return check_jwt(h[n - 1], jwtsecret); - // } - // } - // } - if let Some(auth_header) = session.get_header("authorization") { if let Ok(header_str) = auth_header.to_str() { if let Some((scheme, token)) = header_str.split_once(' ') { diff --git a/src/utils/consul.rs b/src/utils/consul.rs index 37526df..53cee6e 100644 --- a/src/utils/consul.rs +++ b/src/utils/consul.rs @@ -1,4 +1,4 @@ -use crate::utils::kuberconsul::{list_to_upstreams, match_path}; +use crate::utils::kuberconsul::*; use crate::utils::parceyaml::build_headers; use crate::utils::structs::{Configuration, InnerMap, ServiceMapping, UpstreamsDashMap}; use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams}; @@ -9,30 +9,15 @@ use pingora::prelude::sleep; use rand::Rng; use reqwest::header::{HeaderMap, HeaderValue}; use reqwest::Client; -use serde::Deserialize; -use std::collections::HashMap; use std::env; use std::sync::atomic::AtomicUsize; use std::sync::Arc; use std::time::Duration; -#[derive(Debug, Deserialize)] -struct Service { - #[serde(rename = "ServiceTaggedAddresses")] - tagged_addresses: HashMap, -} - -#[derive(Debug, Deserialize)] -struct TaggedAddress { - #[serde(rename = "Address")] - address: String, - #[serde(rename = "Port")] - port: u16, -} - pub async fn start(mut toreturn: Sender, config: Arc) { let prev_upstreams = UpstreamsDashMap::new(); loop { + let upstreams = UpstreamsDashMap::new(); if let Some(consul) = config.consul.clone() { let servers = consul.servers.unwrap_or(vec![format!( "{}:{}", @@ -40,14 +25,13 @@ pub async fn start(mut toreturn: Sender, config: Arc 0 { num = rand::rng().random_range(0..end); } let consul_data = servers.get(num).unwrap().to_string(); let ss = consul_data + "/v1/catalog/service/"; - if let Some(ref svc) = consul.services { + if let Some(svc) = consul.services { for i in svc { let header_list = DashMap::new(); let mut hl = Vec::new(); @@ -62,21 +46,16 @@ pub async fn start(mut toreturn: Sender, config: Arc, conf: &ServiceMapping) } let mut inner_vec = Vec::new(); let upstreams: DashMap, AtomicUsize)> = DashMap::new(); - let endpoints: Vec = resp.json().await.ok()?; + let endpoints: Vec = resp.json().await.ok()?; 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.clone(); diff --git a/src/utils/kuber.rs b/src/utils/kuber.rs index a852a44..5e37dcc 100644 --- a/src/utils/kuber.rs +++ b/src/utils/kuber.rs @@ -1,4 +1,4 @@ -use crate::utils::kuberconsul::{list_to_upstreams, match_path}; +use crate::utils::kuberconsul::*; use crate::utils::parceyaml::build_headers; use crate::utils::structs::{Configuration, InnerMap, ServiceMapping, UpstreamsDashMap}; use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams}; @@ -15,32 +15,17 @@ use std::time::Duration; use tokio::fs::File; use tokio::io::AsyncReadExt; -#[derive(Debug, serde::Deserialize)] -struct Endpoints { - subsets: Option>, +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() } - -#[derive(Debug, serde::Deserialize)] -struct Subset { - addresses: Option>, - ports: Option>, -} - -#[derive(Debug, serde::Deserialize)] -struct Address { - ip: String, -} - -#[derive(Debug, serde::Deserialize)] -struct Port { - port: u16, -} - pub async fn start(mut toreturn: Sender, config: Arc) { let prev_upstreams = UpstreamsDashMap::new(); loop { + let upstreams = UpstreamsDashMap::new(); if let Some(kuber) = config.kubernetes.clone() { - let upstreams = UpstreamsDashMap::new(); let path = kuber.tokenpath.unwrap_or("/var/run/secrets/kubernetes.io/serviceaccount/token".to_string()); let token = read_token(path.as_str()).await; let servers = kuber.servers.unwrap_or(vec![format!( @@ -58,7 +43,6 @@ pub async fn start(mut toreturn: Sender, config: Arc 0 { @@ -70,7 +54,6 @@ pub async fn start(mut toreturn: Sender, config: Arc, config: Arc Option, AtomicUsize)>> { - let client = Client::builder().timeout(Duration::from_secs(2)).danger_accept_invalid_certs(true).build().ok()?; - let to = Duration::from_secs(1); + let to = Duration::from_secs(10); + let client = Client::builder().timeout(Duration::from_secs(10)).danger_accept_invalid_certs(true).build().ok()?; let resp = client.get(url).timeout(to).bearer_auth(token).send().await.ok()?; if !resp.status().is_success() { eprintln!("Kubernetes API returned status: {}", resp.status()); return None; } - let endpoints: Endpoints = resp.json().await.ok()?; + let endpoints: KubeEndpoints = resp.json().await.ok()?; let upstreams: DashMap, AtomicUsize)> = DashMap::new(); if let Some(subsets) = endpoints.subsets { for subset in subsets { @@ -124,10 +107,3 @@ pub async fn get_by_http(url: &str, token: &str, conf: &ServiceMapping) -> Optio } Some(upstreams) } - -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() -} diff --git a/src/utils/kuberconsul.rs b/src/utils/kuberconsul.rs index 7a9b460..626ce07 100644 --- a/src/utils/kuberconsul.rs +++ b/src/utils/kuberconsul.rs @@ -1,7 +1,42 @@ use crate::utils::structs::{InnerMap, ServiceMapping, UpstreamsDashMap}; use dashmap::DashMap; +use serde::Deserialize; +use std::collections::HashMap; use std::sync::atomic::AtomicUsize; +#[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, +} pub fn list_to_upstreams(lt: Option, AtomicUsize)>>, upstreams: &UpstreamsDashMap, i: &ServiceMapping) { if let Some(list) = lt { match upstreams.get(&i.hostname.clone()) {