From 91d2d414a3b253e516fd3d7b1a147d33eb08029b Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Thu, 1 Oct 2026 15:19:18 +0200 Subject: [PATCH] Minor improvements, type changes --- src/core/proxyhttp.rs | 69 +++++++++++++++++----------------------- src/core/start.rs | 20 +++++------- src/utils/healthcheck.rs | 8 ++--- src/utils/httpclient.rs | 18 ++++------- src/utils/parceyaml.rs | 14 ++++---- src/utils/tools.rs | 14 ++++---- src/utils/types.rs | 23 +++++++------- 7 files changed, 74 insertions(+), 92 deletions(-) diff --git a/src/core/proxyhttp.rs b/src/core/proxyhttp.rs index b9d1b05..256b5b5 100644 --- a/src/core/proxyhttp.rs +++ b/src/core/proxyhttp.rs @@ -43,7 +43,7 @@ pub struct Context { upstream_peer: Option>, extraparams: arc_swap::Guard>, client_headers: Option)>>, - x4xx_limit: Option, + x4xx_limit: u32, } thread_local! { @@ -60,7 +60,7 @@ impl ProxyHttp for LB { upstream_peer: None, extraparams: self.extraparams.load(), client_headers: None, - x4xx_limit: None, + x4xx_limit: 0, } } @@ -109,7 +109,7 @@ impl ProxyHttp for LB { let hostname = return_header_host_from_upstream(session, &self.ump_upst); _ctx.hostname = hostname; let mut backend_id = None; - if _ctx.extraparams.sticky_sessions.is_some() { + if _ctx.extraparams.sticky_sessions > 0 { if let Some(cookies) = session.req_header().headers.get("cookie") { if let Ok(cookie_str) = cookies.to_str() { if let Some(pos) = cookie_str.find("backend_id=") { @@ -133,12 +133,25 @@ impl ProxyHttp for LB { return Ok(true); } } - if let Some(rate) = innermap.x4xx_limit.or(_ctx.extraparams.x4xx_limit) { - _ctx.x4xx_limit = innermap.x4xx_limit; + let rate = if innermap.rate_limit > 0 { innermap.rate_limit } else { _ctx.extraparams.rate_limit }; + let xrate = if innermap.x4xx_limit > 0 { innermap.x4xx_limit } else { _ctx.extraparams.x4xx_limit }; + if xrate > 0 || rate > 0 { let rate_key = session.client_addr().and_then(|addr| addr.as_inet()).map(|inet| inet.ip()); - if let Some(rk) = rate_key { - let count = REQUESTS_4XX.get(&rk).unwrap_or(0); - if count > rate { + if xrate > 0 { + _ctx.x4xx_limit = xrate; + if let Some(rk) = rate_key { + let count = REQUESTS_4XX.get(&rk).unwrap_or(0); + if count > xrate { + let header = ResponseHeader::build(429, None)?; + session.set_keepalive(None); + session.write_response_header(Box::new(header), true).await?; + return Ok(true); + } + } + } + if rate > 0 { + let curr_window_requests = RATE_LIMITER.observe(&rate_key, 1); + if curr_window_requests > rate { let header = ResponseHeader::build(429, None)?; session.set_keepalive(None); session.write_response_header(Box::new(header), true).await?; @@ -146,16 +159,6 @@ impl ProxyHttp for LB { } } } - if let Some(rate) = innermap.rate_limit.or(_ctx.extraparams.rate_limit) { - let rate_key = session.client_addr().and_then(|addr| addr.as_inet()).map(|inet| inet.ip()); - let curr_window_requests = RATE_LIMITER.observe(&rate_key, 1); - if curr_window_requests > rate { - let header = ResponseHeader::build(429, None)?; - session.set_keepalive(None); - session.write_response_header(Box::new(header), true).await?; - return Ok(true); - } - } if let Some(redirect_to) = &innermap.redirect_to { let uri = session.req_header().uri.path(); @@ -170,7 +173,7 @@ impl ProxyHttp for LB { return Ok(true); } - if _ctx.extraparams.to_https.unwrap_or(false) || innermap.to_https { + if _ctx.extraparams.to_https || innermap.to_https { if let Some(stream) = session.stream() { if stream.get_ssl().is_none() { if let Some(host) = _ctx.hostname.as_ref() { @@ -213,21 +216,7 @@ impl ProxyHttp for LB { peer.options.verify_cert = false; peer.options.verify_hostname = false; } - /* - Experimental optionsv - The following TCP optimizations were tested but caused performance degrade under heavy load: - peer.options.tcp_keepalive = Some(TcpKeepalive { - idle: Duration::from_secs(60), - interval: Duration::from_secs(10), - count: 5, - user_timeout: Duration::from_secs(30), - }); - - peer.options.idle_timeout = Some(Duration::from_secs(300)); - peer.options.tcp_recv_buf = Some(128 * 1024); - End of experimental options - */ - if ctx.extraparams.sticky_sessions.is_some() { + if ctx.extraparams.sticky_sessions > 0 { let mut s = String::with_capacity(64); write!( &mut s, @@ -237,9 +226,9 @@ impl ProxyHttp for LB { innermap.port, innermap.is_http2, innermap.to_https, - innermap.x4xx_limit.unwrap_or_default(), - innermap.rate_limit.unwrap_or_default(), - innermap.healthcheck.unwrap_or_default(), + innermap.x4xx_limit, + innermap.rate_limit, + innermap.healthcheck, innermap.authorization ) .unwrap_or(()); @@ -303,7 +292,7 @@ impl ProxyHttp for LB { Ok(()) } async fn response_filter(&self, _session: &mut Session, _upstream_response: &mut ResponseHeader, ctx: &mut Self::CTX) -> Result<()> { - if let Some(val) = ctx.extraparams.sticky_sessions { + if ctx.extraparams.sticky_sessions > 0 { if let Some(bid) = &ctx.backend_id { let tt = if let Some(existing) = REVERSE_STORE.get(bid) { existing.value().clone() @@ -321,7 +310,7 @@ impl ProxyHttp for LB { buf.push_str("backend_id="); buf.push_str(&tt); buf.push_str("; Path=/; Max-Age="); - buf.push_str(&val.to_string()); + buf.push_str(&ctx.extraparams.sticky_sessions.to_string()); buf.push_str("; HttpOnly; SameSite=Lax"); let _ = _upstream_response.append_header("set-cookie", buf.as_str()); } @@ -347,7 +336,7 @@ impl ProxyHttp for LB { }; calc_metrics(m); ACTIVE_SESSIONS.dec(); - if ctx.x4xx_limit.or(ctx.extraparams.x4xx_limit).is_some() && (400..=499).contains(&response_code) { + if ctx.x4xx_limit > 0 && (400..=499).contains(&response_code) { if let Some(ip) = session.client_addr().and_then(|a| a.as_inet()).map(|i| i.ip()) { let current = REQUESTS_4XX.get(&ip).unwrap_or(0); REQUESTS_4XX.insert(ip, current + 1); diff --git a/src/core/start.rs b/src/core/start.rs index e422601..6b7ddca 100644 --- a/src/core/start.rs +++ b/src/core/start.rs @@ -11,9 +11,9 @@ use dashmap::DashMap; use log::info; use pingora::tls::ssl::{SslAlert, SslRef}; use pingora_cache::eviction::simple_lru::Manager; -use pingora_core::listeners::tls::TlsSettings; use pingora_core::listeners::TcpSocketOptions; -use pingora_core::prelude::{background_service, Opt}; +use pingora_core::listeners::tls::TlsSettings; +use pingora_core::prelude::{Opt, background_service}; use pingora_core::protocols::TcpKeepalive; use pingora_core::server::Server; use privdrop::reexports::libc::SIGQUIT; @@ -23,8 +23,8 @@ use signal_hook::{ consts::{SIGINT, SIGTERM}, iterator::Signals, }; -use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::Arc; +use std::sync::mpsc::{Receiver, Sender, channel}; use std::time::Duration; use std::{fs, thread}; @@ -51,11 +51,12 @@ pub fn run() { let sh_config = Arc::new(DashMap::new()); let ec_config = Arc::new(ArcSwap::from_pointee(Extraparams { - to_https: None, - sticky_sessions: None, + to_https: false, + ssl_verify: false, + sticky_sessions: 0, authentication: None, - rate_limit: None, - x4xx_limit: None, + rate_limit: 0, + x4xx_limit: 0, })); let cfg = Arc::new(maincfg); @@ -70,8 +71,6 @@ pub fn run() { extraparams: ec_config, cache_enabled, }; - // let al = cfg.access_log.clone().unwrap_or("none".to_string()); - // init_access_log(al.as_str()); let grade = cfg.proxy_tls_grade.clone().unwrap_or("medium".to_string()); info!("TLS grade set to: [ {} ]", grade); @@ -84,9 +83,6 @@ pub fn run() { check_priv(bind_address_http.as_str()); - // let mut tcp_options: Option = Some(TcpSocketOptions::default()); - // let mut tcp_options = TcpSocketOptions::default(); - let mut tcp_options: Option = None; if let Some(idle) = cfg.tcp_keepalive_idle { let mut to = TcpSocketOptions::default(); diff --git a/src/utils/healthcheck.rs b/src/utils/healthcheck.rs index 2c3c077..a5d0bff 100644 --- a/src/utils/healthcheck.rs +++ b/src/utils/healthcheck.rs @@ -1,11 +1,11 @@ -use crate::utils::hcclient::{httpclient, HttpClientRequest}; +use crate::utils::hcclient::{HttpClientRequest, httpclient}; use crate::utils::lazylock::REVERSE_STORE; use crate::utils::tools::*; use crate::utils::types::{InnerMap, UpstreamsDashMap, UpstreamsIdMap}; use bytes::Bytes; use dashmap::DashMap; -use std::sync::atomic::AtomicUsize; use std::sync::Arc; +use std::sync::atomic::AtomicUsize; use std::time::Duration; use tokio::net::TcpStream; use tokio::time::{interval, timeout}; @@ -44,7 +44,7 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsD let mut innervec = Vec::new(); for upstream in path_entry.value().0.iter() { - let tls = if upstream.healthcheck.unwrap_or(true) { + let tls = if upstream.healthcheck { detect_tls(upstream.address.as_ref(), &upstream.port, host).await } else { false @@ -62,7 +62,7 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsD authorization: upstream.authorization.clone(), }; - if scheme.healthcheck.unwrap_or(true) { + if scheme.healthcheck { let link = if tls { format!("https://{}:{}{}", upstream.address, upstream.port, path) } else { diff --git a/src/utils/httpclient.rs b/src/utils/httpclient.rs index ca82140..041c71b 100644 --- a/src/utils/httpclient.rs +++ b/src/utils/httpclient.rs @@ -90,10 +90,10 @@ pub async fn for_consul(url: &str, token: Option, conf: &GlobalServiceMa is_ssl: false, is_http2: false, to_https: conf.to_https.unwrap_or(false), - rate_limit: conf.rate_limit, - x4xx_limit: conf.x4xx_limit, + rate_limit: conf.rate_limit.unwrap_or(0), + x4xx_limit: conf.x4xx_limit.unwrap_or(0), redirect_to: conf.redirect_to.clone().map(Arc::::from), - healthcheck: None, + healthcheck: true, authorization: None, }); inner_vec.push(to_add); @@ -132,9 +132,9 @@ pub async fn for_kuber(url: &str, token: &str, conf: &GlobalServiceMapping) -> O is_ssl: false, is_http2: false, to_https: conf.to_https.unwrap_or(false), - rate_limit: conf.rate_limit, - x4xx_limit: conf.x4xx_limit, - healthcheck: None, + rate_limit: conf.rate_limit.unwrap_or(0), + x4xx_limit: conf.x4xx_limit.unwrap_or(0), + healthcheck: true, redirect_to: None, authorization: None, }); @@ -213,11 +213,7 @@ pub async fn getfromapi(url: &str, token: Option, provider: &str) -> Opt CONNECTOR.release_http_session(http_session.0, &peer, None).await; - if status == 200 && !body_bytes.is_empty() { - Some(body_bytes) - } else { - None - } + if status == 200 && !body_bytes.is_empty() { Some(body_bytes) } else { None } } fn parse_url(url: &str) -> Result<(&str, u16, &str, bool), &'static str> { diff --git a/src/utils/parceyaml.rs b/src/utils/parceyaml.rs index a9b9bb5..ef215e8 100644 --- a/src/utils/parceyaml.rs +++ b/src/utils/parceyaml.rs @@ -172,10 +172,10 @@ async fn populate_headers_and_auth(config: &mut Configuration, parsed: &Config) let server_global_headers: DashMap, Vec<(String, Arc)>> = DashMap::new(); server_global_headers.insert(Arc::from("/"), sh); config.server_headers.insert(Arc::from("GLOBAL_SERVER_HEADERS"), server_global_headers); - config.extraparams.to_https = parsed.to_https; - config.extraparams.sticky_sessions = parsed.sticky_sessions; - config.extraparams.rate_limit = parsed.rate_limit; - config.extraparams.x4xx_limit = parsed.x4xx_limit; + config.extraparams.to_https = parsed.to_https.unwrap_or(false); + config.extraparams.sticky_sessions = parsed.sticky_sessions.unwrap_or(0); + config.extraparams.rate_limit = parsed.rate_limit.unwrap_or(0); + config.extraparams.x4xx_limit = parsed.x4xx_limit.unwrap_or(0); if let Some(rate) = &parsed.rate_limit { info!("Applied Global Rate Limit : {} request per second", rate); @@ -228,9 +228,9 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) { is_ssl: false, is_http2: false, to_https: path_config.to_https.unwrap_or(false), - rate_limit: path_config.rate_limit, - x4xx_limit: path_config.x4xx_limit, - healthcheck: path_config.healthcheck, + rate_limit: path_config.rate_limit.unwrap_or(0), + x4xx_limit: path_config.x4xx_limit.unwrap_or(0), + healthcheck: path_config.healthcheck.unwrap_or(true), redirect_to: redirect_link, authorization: path_auth, })); diff --git a/src/utils/tools.rs b/src/utils/tools.rs index 0dfc746..d9ef1ef 100644 --- a/src/utils/tools.rs +++ b/src/utils/tools.rs @@ -4,7 +4,7 @@ use crate::utils::types::{Extraparams, InnerMapForJson, UpstreamSnapshotForJson, use dashmap::DashMap; use log::{error, info}; use privdrop::PrivDrop; -use serde_json::{json, Value}; +use serde_json::{Value, json}; use sha2::{Digest, Sha256}; use std::collections::{HashMap, HashSet}; use std::fmt::Write; @@ -16,8 +16,8 @@ use std::os::unix::fs::MetadataExt; use std::os::unix::fs::OpenOptionsExt; use std::process::Command; use std::str::FromStr; -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::{env, fs, process, thread, time}; pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams) { @@ -35,8 +35,8 @@ pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams) f.is_ssl, f.is_http2, f.to_https, - f.rate_limit.unwrap_or(extraparams.rate_limit.unwrap_or(0)), - f.x4xx_limit.unwrap_or(extraparams.x4xx_limit.unwrap_or(0)) + if f.rate_limit > 0 { f.rate_limit } else { extraparams.rate_limit }, + if f.x4xx_limit > 0 { f.x4xx_limit } else { extraparams.x4xx_limit } ) .unwrap(); } @@ -138,9 +138,9 @@ pub fn clone_idmap_into(original: &UpstreamsDashMap, cloned: &UpstreamsIdMap) { x.port, x.is_http2, x.to_https, - x.x4xx_limit.unwrap_or_default(), - x.rate_limit.unwrap_or_default(), - x.healthcheck.unwrap_or_default(), + x.x4xx_limit, + x.rate_limit, + x.healthcheck, x.authorization ) .unwrap_or(()); diff --git a/src/utils/types.rs b/src/utils/types.rs index bf38131..46733d3 100644 --- a/src/utils/types.rs +++ b/src/utils/types.rs @@ -1,8 +1,8 @@ use dashmap::DashMap; use serde::{Deserialize, Serialize}; use std::collections::HashMap; -use std::sync::atomic::AtomicUsize; use std::sync::Arc; +use std::sync::atomic::AtomicUsize; pub type UpstreamsDashMap = DashMap, DashMap, (Vec>, AtomicUsize)>>; @@ -11,11 +11,12 @@ pub type Headers = DashMap, DashMap, Vec<(String, Arc)>>> #[derive(Clone, Debug, Default)] pub struct Extraparams { - pub to_https: Option, - pub sticky_sessions: Option, + pub sticky_sessions: u64, + pub rate_limit: isize, + pub x4xx_limit: u32, + pub to_https: bool, + pub ssl_verify: bool, pub authentication: Option>, - pub rate_limit: Option, - pub x4xx_limit: Option, } #[derive(Debug, Default, Clone, Serialize, Deserialize)] @@ -151,9 +152,9 @@ pub struct InnerMap { pub is_ssl: bool, pub is_http2: bool, pub to_https: bool, - pub rate_limit: Option, - pub x4xx_limit: Option, - pub healthcheck: Option, + pub rate_limit: isize, + pub x4xx_limit: u32, + pub healthcheck: bool, pub redirect_to: Option>, pub authorization: Option>, } @@ -190,9 +191,9 @@ pub struct InnerMapForJson { pub is_ssl: bool, pub is_http2: bool, pub to_https: bool, - pub rate_limit: Option, - pub x4xx_limit: Option, - pub healthcheck: Option, + pub rate_limit: isize, + pub x4xx_limit: u32, + pub healthcheck: bool, } #[derive(Debug, Default, Serialize, Deserialize)] pub struct UpstreamSnapshotForJson {