Minor improvements, type changes

src/core/proxyhttp.rs

@@ -43,7 +43,7 @@ pub struct Context {
     upstream_peer: Option<Arc<InnerMap>>,
     extraparams: arc_swap::Guard<Arc<Extraparams>>,
     client_headers: Option<Vec<(String, Arc<str>)>>,
-    x4xx_limit: Option<u32>,
+    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,15 @@ 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 xrate > 0 {
+                                _ctx.x4xx_limit = xrate;
                                 if let Some(rk) = rate_key {
                                     let count = REQUESTS_4XX.get(&rk).unwrap_or(0);
-                                if count > rate {
+                                    if count > xrate {
                                         let header = ResponseHeader::build(429, None)?;
                                         session.set_keepalive(None);
                                         session.write_response_header(Box::new(header), true).await?;
@@ -146,8 +149,7 @@ 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());
+                            if rate > 0 {
                                 let curr_window_requests = RATE_LIMITER.observe(&rate_key, 1);
                                 if curr_window_requests > rate {
                                     let header = ResponseHeader::build(429, None)?;
@@ -156,6 +158,7 @@ impl ProxyHttp for LB {
                                     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);

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<TcpSocketOptions> = Some(TcpSocketOptions::default());
-    // let mut tcp_options = TcpSocketOptions::default();
-
     let mut tcp_options: Option<TcpSocketOptions> = None;
     if let Some(idle) = cfg.tcp_keepalive_idle {
         let mut to = TcpSocketOptions::default();

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 {

src/utils/httpclient.rs

@@ -90,10 +90,10 @@ pub async fn for_consul(url: &str, token: Option<String>, 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::<str>::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<String>, 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> {

src/utils/parceyaml.rs

@@ -172,10 +172,10 @@ async fn populate_headers_and_auth(config: &mut Configuration, parsed: &Config)
     let server_global_headers: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = 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,
                             }));

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(());

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<Arc<str>, DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>>;
 
@@ -11,11 +11,12 @@ pub type Headers = DashMap<Arc<str>, DashMap<Arc<str>, Vec<(String, Arc<str>)>>>
 
 #[derive(Clone, Debug, Default)]
 pub struct Extraparams {
-    pub to_https: Option<bool>,
-    pub sticky_sessions: Option<u64>,
+    pub sticky_sessions: u64,
+    pub rate_limit: isize,
+    pub x4xx_limit: u32,
+    pub to_https: bool,
+    pub ssl_verify: bool,
     pub authentication: Option<Arc<InnerAuth>>,
-    pub rate_limit: Option<isize>,
-    pub x4xx_limit: Option<u32>,
 }
 
 #[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<isize>,
-    pub x4xx_limit: Option<u32>,
-    pub healthcheck: Option<bool>,
+    pub rate_limit: isize,
+    pub x4xx_limit: u32,
+    pub healthcheck: bool,
     pub redirect_to: Option<Arc<str>>,
     pub authorization: Option<Arc<InnerAuth>>,
 }
@@ -190,9 +191,9 @@ pub struct InnerMapForJson {
     pub is_ssl: bool,
     pub is_http2: bool,
     pub to_https: bool,
-    pub rate_limit: Option<isize>,
-    pub x4xx_limit: Option<u32>,
-    pub healthcheck: Option<bool>,
+    pub rate_limit: isize,
+    pub x4xx_limit: u32,
+    pub healthcheck: bool,
 }
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct UpstreamSnapshotForJson {