Cargo version update

fix linter warnings

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "aralez"
-version = "0.94.1"
-edition = "2021"
+version = "0.95.1"
+edition = "2024"
 license = "Apache-2.0"
 description = "Reverse proxy built on top of Cloudflare's Pingora"
 exclude = ["configs/*"]
@@ -27,12 +27,12 @@ async-trait = "0.1.92"
 log = "0.4.34"
 notify = "9.0.0-rc.4"
 axum = { version = "0.8.9" }
-serde = { version = "1.0.229", features = ["derive"] }
+serde = { version = "1.0.229", features = ["derive", "rc"] }
+#serde = { version = "1.0.229", features = ["derive"] }
 noyalib = { version = "0.0.44", features = ["compat-serde-yaml"] }
 rand = "0.10.2"
 base64 = "0.23.1"
 jsonwebtoken = { version = "11.1.0", default-features = false, features = ["use_pem", "rust_crypto"] }
-#sha2 = { version = "0.11.0-rc.5", default-features = false }
 sha2 = { version = "0.11.0", default-features = false }
 base16ct = { version = "1.0.0", features = ["alloc"] }
 urlencoding = "2.1.3"
@@ -68,4 +68,7 @@ anyhow = "1.0.103"
 inventory = "0.3.24"
 elasticsearch = "9.1.0-alpha.1"
 humantime = "2.4.0"
-
+reqwest = "0.12.28"
+hickory-resolver = { version = "0.26", features = ["tokio"] }
+aws-sdk-route53 = "1.127.0"
+aws-config = { version = "1.12.0", features = ["behavior-version-latest"] }

src/auth/types.rs

@@ -11,10 +11,10 @@ pub trait AuthValidator {
 }
 pub async fn authenticate(auth: &InnerAuth, session: &mut Session) -> bool {
     match &*auth.auth_type {
-        "basic" => BasicAuth(&*auth.auth_cred).validate(session).await,
-        "apikey" => ApiKeyAuth(&*auth.auth_cred).validate(session).await,
+        "basic" => BasicAuth(&auth.auth_cred).validate(session).await,
+        "apikey" => ApiKeyAuth(&auth.auth_cred).validate(session).await,
         "jwt" => JwtAuth().validate(session).await,
-        "forward" => ForwardAuth(&*auth.auth_cred).validate(session).await,
+        "forward" => ForwardAuth(&auth.auth_cred).validate(session).await,
         _ => {
             log::warn!("Unsupported authentication mechanism : {}", &*auth.auth_type);
             false

src/core/bgservice.rs

@@ -72,6 +72,7 @@ impl BackgroundService for LB {
             file_server_folder: self.config.file_server_folder.clone(),
             current_upstreams: self.ump_upst.clone(),
             full_upstreams: self.ump_full.clone(),
+            acme_dns_provider: self.config.acme_dns_provider.clone(),
         };
         tokio::spawn(async move { api_load.start(tx_api).await });
         let uu = self.ump_upst.clone();
@@ -80,7 +81,8 @@ impl BackgroundService for LB {
         let (hc_method, hc_interval) = (self.config.hc_method.clone(), self.config.hc_interval);
         tokio::spawn(async move { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await });
         tokio::spawn(async move { calc_cache_metrics().await });
-        tokio::spawn(async move { refresh_order(certdir, confdir).await });
+        let acme_provider = self.config.acme_dns_provider.clone();
+        tokio::spawn(async move { refresh_order(certdir, confdir, acme_provider).await });
 
         init_access_logging(self.config.access_log.clone()).await;
         loop {

src/core/proxyhttp.rs

@@ -8,9 +8,9 @@ use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use axum::body::Bytes;
 use log::error;
+use pingora::ErrorSource::Upstream;
 use pingora::http::{Method, RequestHeader, ResponseHeader, StatusCode};
 use pingora::prelude::*;
-use pingora::ErrorSource::Upstream;
 use pingora_cache::cache_control::CacheControl;
 use pingora_cache::{CacheKey, CacheMetaDefaults, RespCacheable};
 use pingora_core::listeners::ALPN;
@@ -34,6 +34,7 @@ pub struct LB {
     pub config: Arc<AppConfig>,
     pub extraparams: Arc<ArcSwap<Extraparams>>,
     pub cache_enabled: bool,
+    pub master_server: bool,
 }
 
 pub struct Context {
@@ -43,7 +44,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 +61,7 @@ impl ProxyHttp for LB {
             upstream_peer: None,
             extraparams: self.extraparams.load(),
             client_headers: None,
-            x4xx_limit: None,
+            x4xx_limit: 0,
         }
     }
 
@@ -105,11 +106,15 @@ impl ProxyHttp for LB {
     }
 
     async fn request_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<bool> {
+        if self.master_server {
+            let _ = session.respond_error(400).await;
+            return Ok(true);
+        };
         ACTIVE_SESSIONS.inc();
         let hostname = return_header_host_from_upstream(session, &self.ump_upst);
         _ctx.hostname = hostname;
         let mut backend_id = None;
-        if let Some(_) = _ctx.extraparams.sticky_sessions {
+        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=") {
@@ -128,17 +133,30 @@ impl ProxyHttp for LB {
                     None => return Ok(false),
                     Some(ref innermap) => {
                         if let Some(auth) = _ctx.extraparams.authentication.as_ref().or(innermap.authorization.as_ref()) {
-                            if !authenticate(&auth, session).await {
+                            if !authenticate(auth, session).await {
                                 let _ = session.respond_error(401).await;
                                 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 +164,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 +178,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 +221,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 let Some(_) = ctx.extraparams.sticky_sessions {
+                    if ctx.extraparams.sticky_sessions > 0 {
                         let mut s = String::with_capacity(64);
                         write!(
                             &mut s,
@@ -237,9 +231,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 +297,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 +315,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,12 +341,10 @@ impl ProxyHttp for LB {
         };
         calc_metrics(m);
         ACTIVE_SESSIONS.dec();
-        if let Some(_) = ctx.x4xx_limit.or(ctx.extraparams.x4xx_limit) {
-            if (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);
-                }
+        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);
             }
         }
         access_log(response_code, &self.request_summary(session, ctx), session).await;

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};
 
@@ -34,6 +34,7 @@ pub fn run() {
     let file = parameters.conf.clone().unwrap();
     let maincfg = crate::utils::parceyaml::parce_main_config(file.as_str());
 
+    let master_server = maincfg.cluster_member.as_deref() == Some("master");
     let cache_enabled = maincfg.cache_size_mb.is_some();
     CACHE_TTL.set(maincfg.cache_ttl.unwrap_or(60)).unwrap();
     if let Some(size) = maincfg.cache_size_mb {
@@ -51,11 +52,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);
@@ -68,10 +70,9 @@ pub fn run() {
         client_headers: ch_config,
         server_headers: sh_config,
         extraparams: ec_config,
-        cache_enabled: cache_enabled,
+        cache_enabled,
+        master_server,
     };
-    // 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 +85,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();
@@ -164,8 +162,8 @@ pub fn run() {
     if let Err(e) = write_pid_file(pf.as_str()) {
         panic!("Failed to write PID file: {} : {}", pf, e);
     }
-    let mut signals = Signals::new(&[SIGINT, SIGTERM, SIGQUIT]).unwrap();
-    for sig in signals.forever() {
+    let mut signals = Signals::new([SIGINT, SIGTERM, SIGQUIT]).unwrap();
+    if let Some(sig) = signals.forever().next() {
         match sig {
             SIGINT => info!("SIGINT received! Exiting..."),
             SIGTERM => info!("SIGTERM received! Exiting..."),
@@ -175,6 +173,5 @@ pub fn run() {
             }
             _ => unreachable!(),
         }
-        break;
     }
 }

src/core/webserver.rs

@@ -1,7 +1,8 @@
 use crate::auth::jwt::Claims;
 use crate::ingress::APIUpstreamProvider;
 use crate::tls::acme::http01::{acme_create, acme_order, http01_challenge};
-use crate::utils::metrics::{get_memory_usage, get_open_files, MEMORY_USAGE, OPEN_FILES};
+use crate::utils::lazylock::NODE_STATUS;
+use crate::utils::metrics::{MEMORY_USAGE, OPEN_FILES, get_memory_usage, get_open_files};
 use crate::utils::tools::{upstreams_liveness_json, upstreams_to_json};
 use crate::utils::types::{Config, Configuration, UpstreamsDashMap};
 use axum::body::Body;
@@ -10,9 +11,9 @@ use axum::http::{Response, StatusCode};
 use axum::response::IntoResponse;
 use axum::routing::{any, get, post};
 use axum::{Json, Router};
-use jsonwebtoken::{encode, EncodingKey, Header};
+use jsonwebtoken::{EncodingKey, Header, encode};
 use log::{debug, error, info, warn};
-use prometheus::{gather, Encoder, TextEncoder};
+use prometheus::{Encoder, TextEncoder, gather};
 use serde::Serialize;
 use signal_hook::{consts::SIGQUIT, iterator::Signals};
 use std::collections::HashMap;
@@ -32,6 +33,7 @@ struct OutToken {
 #[derive(Clone)]
 pub struct AppState {
     master_key: Option<String>,
+    pub acme_dns_provider: Option<String>,
     pub cert_creds: String,
     pub certs_dir: String,
     upstreams_file: String,
@@ -53,6 +55,7 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
         config_api_enabled: config.config_api_enabled,
         current_upstreams: upstreams_curr,
         full_upstreams: upstreams_full,
+        acme_dns_provider: config.acme_dns_provider.clone(),
     };
     let app = Router::new()
         // .route("/{*wildcard}", get(senderror))
@@ -61,13 +64,14 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
         .route("/acme_order/{*domain}", any(acme_order))
         .route("/.well-known/acme-challenge/{*token}", any(http01_challenge))
         .route("/conf", post(conf))
+        .route("/workerconf", post(workerconf))
         .route("/metrics", get(metrics))
         .route("/status", get(status))
         .with_state(app_state);
 
     let mut static_handle: Option<tokio::task::JoinHandle<()>> = None;
     if let (Some(address), Some(folder)) = (&config.file_server_address, &config.file_server_folder) {
-        let static_listen = port_is_available("File Server", &address).await;
+        let static_listen = port_is_available("File Server", address).await;
         let static_files = ServeDir::new(folder);
         let static_serve: Router = Router::new().fallback_service(static_files);
         info!("Starting file server on: {}", address);
@@ -82,14 +86,14 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
 
     let listener = port_is_available("Config API", &config.address).await;
     info!("Starting the API server on: {}", config.address);
-    let api_server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    // let api_server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    let api_server = tokio::spawn(async move { axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>()).await.unwrap() });
 
     let (tx, mut rx) = mpsc::channel(1);
     std::thread::spawn(move || {
-        let mut signals = Signals::new(&[SIGQUIT]).unwrap();
-        for sig in signals.forever() {
+        let mut signals = Signals::new([SIGQUIT]).unwrap();
+        if let Some(sig) = signals.forever().next() {
             tx.blocking_send(sig).unwrap();
-            break;
         }
     });
     rx.recv().await;
@@ -109,7 +113,7 @@ async fn conf(State(st): State<AppState>, Query(params): Query<HashMap<String, S
     let parsed = noyalib::from_str::<Config>(strcontent);
     match parsed {
         Ok(_) => {
-            if let Some(_) = params.get("save") {
+            if params.contains_key("save") {
                 tokio::spawn(async move { apply_config(content.as_str(), st, true).await });
             } else {
                 tokio::spawn(async move { apply_config(content.as_str(), st, false).await });
@@ -123,6 +127,54 @@ async fn conf(State(st): State<AppState>, Query(params): Query<HashMap<String, S
     }
 }
 
+use axum::extract::ConnectInfo;
+use axum::http::HeaderMap;
+async fn workerconf(State(st): State<AppState>, ConnectInfo(addr): ConnectInfo<SocketAddr>, headers: HeaderMap, content: String) -> impl IntoResponse {
+    let api_key = headers.get("API-KEY").and_then(|v| v.to_str().ok());
+    if api_key != st.master_key.as_deref() {
+        warn!("Unauthorised access to workerconf endpoint from {}", addr.ip());
+        return Response::builder()
+            .status(StatusCode::UNAUTHORIZED)
+            .body(Body::from("Unauthorized: Invalid API-KEY\n"))
+            .unwrap();
+    }
+    if let Some(member) = NODE_STATUS.get() {
+        if let Some(mbr) = member.as_ref() {
+            match mbr.as_ref() {
+                "worker" => {
+                    info!("Accepted new config from master server {}", addr.ip());
+                    let strcontent = content.as_str();
+                    let parsed = noyalib::from_str::<Config>(strcontent);
+                    match parsed {
+                        Ok(_) => {
+                            tokio::spawn(async move { apply_config(content.as_str(), st, true).await });
+                            return Response::builder().status(StatusCode::OK).body(Body::from("Accepted! Applying in background\n")).unwrap();
+                        }
+                        Err(err) => {
+                            error!("Failed to parse upstreams file: {}", err);
+                            return Response::builder().status(StatusCode::BAD_GATEWAY).body(Body::from(format!("Failed: {}\n", err))).unwrap();
+                        }
+                    }
+                }
+                _ => {
+                    return Response::builder()
+                        .status(StatusCode::BAD_GATEWAY)
+                        .body(Body::from("Node is not running on worker mode !\n"))
+                        .unwrap();
+                }
+            }
+        }
+        return Response::builder()
+            .status(StatusCode::BAD_GATEWAY)
+            .body(Body::from("Node is not running on worker mode !\n"))
+            .unwrap();
+    }
+    Response::builder()
+        .status(StatusCode::BAD_GATEWAY)
+        .body(Body::from("Node is not running on worker mode !\n"))
+        .unwrap()
+}
+
 async fn apply_config(content: &str, st: AppState, save: bool) {
     let sl = crate::utils::parceyaml::load_configuration(content, "content").await;
     if let Some(serverlist) = sl.0 {
@@ -215,7 +267,7 @@ async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String,
                     .status(StatusCode::OK)
                     .header("Content-Type", "application/json")
                     .body(Body::from(j))
-                    .unwrap()
+                    .unwrap();
             }
             Err(e) => {
                 return Response::builder()

src/ingress/discovery.rs

@@ -12,6 +12,7 @@ pub struct APIUpstreamProvider {
     pub config_api_enabled: bool,
     pub address: String,
     pub masterkey: Option<String>,
+    pub acme_dns_provider: Option<String>,
     pub certs_dir: String,
     pub config_dir: String,
     pub upstreams_file: String,

src/ingress/kubewatcher.rs

@@ -107,7 +107,7 @@ async fn process_ingress_change(client: &Client, ing: &Ingress, target_class: &s
                 let key_clone = key.clone();
                 tokio::spawn(async move {
                     if let Err(e) = patch_ingress_status_address(&client_clone, &ns_clone, &name_clone, &external_ip).await {
-                        error!("Failed to patch ingress status address {}/{}: {}", &ns_clone, &name_clone, e);
+                        error!("Failed to patch ingress status address {}/{}: {}", ns_clone, name_clone, e);
                         PATCHED_INGRESSES.remove(&key_clone);
                     }
                 });

src/logging/core.rs

@@ -160,7 +160,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
     }
     if let Some(location) = location {
         let parts: Vec<&str> = location.split(',').map(|s| s.trim()).collect();
-        let path = match parts.get(0).filter(|s| !s.is_empty()) {
+        let path = match parts.first().filter(|s| !s.is_empty()) {
             Some(p) => p,
             None => {
                 error!("Invalid log location string provided; falling back to stdout");
@@ -213,7 +213,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
 }
 
 pub async fn init_access_log(level_str: &str) {
-    let level = LogLevel::from_str(level_str);
+    let level = LogLevel::parse(level_str);
     let _ = ACCESS_LOG.set(level);
 }
 
@@ -225,7 +225,7 @@ pub enum LogLevel {
 }
 
 impl LogLevel {
-    pub fn from_str(s: &str) -> Self {
+    pub fn parse(s: &str) -> Self {
         match s {
             "all" => LogLevel::Access,
             "error" => LogLevel::Error,
@@ -288,7 +288,7 @@ pub async fn access_log(response_code: u16, summary: &str, session: &Session) {
         return;
     }
     if let Some(sender) = ACCESS_LOG_SENDER.get() {
-        if let Err(_) = sender.try_send(msg) {
+        if sender.try_send(msg).is_err() {
             LOGGING_ERRORS.inc();
         }
     }

src/tls/acme/dns/cloudflare.rs

@@ -0,0 +1,74 @@
+use crate::tls::acme::lookup::lookup_wait;
+use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
+use log::info;
+use reqwest::Client;
+use serde_json::json;
+
+pub struct CloudflareProvider {
+    pub client: Client,
+    pub api_token: String,
+    pub zone_id: String,
+}
+
+impl CloudflareProvider {
+    pub fn new() -> Self {
+        let api_token = std::env::var("CLOUDFLARE_API_TOKEN").expect("Environment variable CLOUDFLARE_API_TOKEN is required");
+        let zone_id = std::env::var("CLOUDFLARE_ZONE_ID").expect("Environment variable CLOUDFLARE_ZONE_ID is required");
+
+        Self {
+            client: Client::new(),
+            api_token,
+            zone_id,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl DnsProvider for CloudflareProvider {
+    async fn create_txt_record(&self, _domain: &str, name: &str, value: &str) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
+        let url = format!("https://api.cloudflare.com/client/v4/zones/{}/dns_records", self.zone_id);
+
+        let response = self
+            .client
+            .post(&url)
+            .bearer_auth(&self.api_token)
+            .json(&json!({
+                "type": "TXT",
+                "name": name,
+                "content": value,
+                "ttl": 120,
+            }))
+            .send()
+            .await?;
+
+        let res_json: serde_json::Value = response.json().await?;
+        if !res_json["success"].as_bool().unwrap_or(false) {
+            return Err(format!("Cloudflare API error: {:?}", res_json["errors"]).into());
+        }
+
+        let record_id = res_json["result"]["id"].as_str().ok_or("missing id")?.to_string();
+        info!("Created TXT record name: {}, id: {}", name, record_id);
+        lookup_wait(name, Some("1.1.1.1"), 60, value).await;
+        Ok(record_id)
+    }
+
+    async fn delete_txt_record(&self, record_id: &str, record_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let url = format!("https://api.cloudflare.com/client/v4/zones/{}/dns_records/{}", self.zone_id, record_id);
+
+        let response = self.client.delete(&url).bearer_auth(&self.api_token).send().await?;
+
+        let res_json: serde_json::Value = response.json().await?;
+        if !res_json["success"].as_bool().unwrap_or(false) {
+            return Err(format!("Cloudflare API error deleting record: {:?}", res_json["errors"]).into());
+        }
+        info!("Deleted TXT record name: {}, id: {}", record_name, record_id);
+        Ok(())
+    }
+}
+
+inventory::submit! {
+    DnsBackendPlugin {
+        name: "cloudflare",
+        factory: || Box::new(CloudflareProvider::new()),
+    }
+}

src/tls/acme/dns/digitalocean.rs

@@ -0,0 +1,70 @@
+use crate::tls::acme::lookup::lookup_wait;
+use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
+use async_trait::async_trait;
+use log::info;
+use serde_json::json;
+use std::env;
+
+pub struct DigitalOceanProvider {
+    token: String,
+    client: reqwest::Client,
+}
+
+impl DigitalOceanProvider {
+    pub fn new() -> Self {
+        let token = env::var("DIGITALOCEAN_TOKEN").expect("Environment variable DIGITALOCEAN_TOKEN is required");
+
+        Self {
+            token,
+            client: reqwest::Client::new(),
+        }
+    }
+}
+
+#[async_trait]
+impl DnsProvider for DigitalOceanProvider {
+    async fn create_txt_record(&self, domain: &str, name: &str, value: &str) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
+        let relative_name = name.strip_suffix(&format!(".{}", domain)).unwrap_or(name);
+        let url = format!("https://api.digitalocean.com/v2/domains/{}/records", domain);
+        let res = self
+            .client
+            .post(&url)
+            .bearer_auth(&self.token)
+            .json(&json!({
+                "type": "TXT",
+                "name": relative_name,
+                "data": value,
+                "ttl": 30
+            }))
+            .send()
+            .await?;
+
+        if !res.status().is_success() {
+            let err_text = res.text().await?;
+            return Err(format!("DigitalOcean API error: {}", err_text).into());
+        }
+
+        let json: serde_json::Value = res.json().await?;
+        let record_id = json["domain_record"]["id"].as_i64().ok_or("Missing record ID in DigitalOcean response")?.to_string();
+        info!("Created TXT record name: {}, id: {}", name, record_id);
+        lookup_wait(name, Some("1.1.1.1"), 60, value).await;
+        Ok(record_id)
+    }
+
+    async fn delete_txt_record(&self, record_id: &str, domain: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let url = format!("https://api.digitalocean.com/v2/domains/{}/records/{}", domain, record_id);
+        let res = self.client.delete(&url).bearer_auth(&self.token).send().await?;
+        if !res.status().is_success() {
+            let err_text = res.text().await?;
+            return Err(format!("Failed to delete DigitalOcean record: {}", err_text).into());
+        }
+        Ok(())
+    }
+}
+
+inventory::submit! {
+    DnsBackendPlugin {
+        name: "digitalocean",
+        factory: || Box::new(DigitalOceanProvider::new()),
+    }
+}

src/tls/acme/dns/mod.rs

@@ -0,0 +1,3 @@
+pub mod cloudflare;
+pub mod digitalocean;
+pub mod route53;

src/tls/acme/dns/route53.rs

@@ -0,0 +1,123 @@
+// ------------------------ //
+// This file is vibecoded !
+// ------------------------ //
+
+use crate::tls::acme::lookup::lookup_wait;
+use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
+use aws_config::SdkConfig;
+use aws_sdk_route53::Client;
+use aws_sdk_route53::types::{Change, ChangeAction, ChangeBatch, ResourceRecord, ResourceRecordSet, RrType};
+use log::info;
+use tokio::runtime::Handle;
+
+pub struct Route53Provider {
+    pub client: Client,
+    pub hosted_zone_id: String,
+}
+
+impl Route53Provider {
+    pub fn new() -> Self {
+        let hosted_zone_id = std::env::var("AWS_HOSTED_ZONE_ID").expect("Environment variable AWS_HOSTED_ZONE_ID is required");
+        // Load AWS SDK config synchronously using Tokio runtime handle to fit factory construction
+        let config: SdkConfig = if let Ok(handle) = Handle::try_current() {
+            tokio::task::block_in_place(|| handle.block_on(aws_config::load_from_env()))
+        } else {
+            tokio::runtime::Runtime::new().unwrap().block_on(aws_config::load_from_env())
+        };
+        let client = Client::new(&config);
+        Self { client, hosted_zone_id }
+    }
+
+    pub async fn get_authoritative_ns_ip(&self) -> Option<String> {
+        // Query Route53 for the zone's details (includes assigned Name Servers)
+        let response = self.client.get_hosted_zone().id(&self.hosted_zone_id).send().await.ok()?;
+
+        // Pick the first name server assigned to this zone (e.g., "ns-1289.awsdns-29.org")
+        let ns_domain = response.delegation_set()?.name_servers().first()?;
+
+        // Resolve that nameserver's domain name to an IP address
+        let ips = tokio::net::lookup_host(format!("{}:53", ns_domain)).await.ok()?;
+
+        let ip = ips.into_iter().next()?.ip().to_string();
+        info!("Targeting Route53 authoritative NS {} ({}) for polling", ns_domain, ip);
+        Some(ip)
+    }
+}
+
+#[async_trait::async_trait]
+impl DnsProvider for Route53Provider {
+    async fn create_txt_record(&self, _domain: &str, name: &str, value: &str) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
+        // AWS Route53 requires TXT strings to be wrapped in double quotes
+        let formatted_value = format!("\"{}\"", value);
+
+        let change = Change::builder()
+            .action(ChangeAction::Upsert)
+            .resource_record_set(
+                ResourceRecordSet::builder()
+                    .name(name)
+                    .r#type(RrType::Txt)
+                    .ttl(60)
+                    .resource_records(ResourceRecord::builder().value(formatted_value).build()?)
+                    .build()?,
+            )
+            .build()?;
+
+        let change_batch = ChangeBatch::builder().changes(change).build()?;
+
+        let response = self
+            .client
+            .change_resource_record_sets()
+            .hosted_zone_id(&self.hosted_zone_id)
+            .change_batch(change_batch)
+            .send()
+            .await?;
+
+        let record_id = response.change_info().ok_or("Missing ChangeInfo in Route53 response")?.id().to_string();
+        info!("Created Route53 TXT record name: {}, change_id: {}", name, record_id);
+        let target_dns_ip = self.get_authoritative_ns_ip().await;
+        lookup_wait(name, target_dns_ip.as_deref(), 60, value).await;
+        Ok(record_id)
+    }
+
+    async fn delete_txt_record(&self, _record_id: &str, record_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        // Since Route53 requires record value context to delete, we look up or pass existing TXT record values.
+        // For ACME cleanup, issuing a DELETE with wildcards/upsert strategy or direct deletion:
+        let list_res = self
+            .client
+            .list_resource_record_sets()
+            .hosted_zone_id(&self.hosted_zone_id)
+            .start_record_name(record_name)
+            .start_record_type(RrType::Txt)
+            .max_items(1)
+            .send()
+            .await?;
+
+        if let Some(record_set) = list_res.resource_record_sets().first() {
+            if record_set.name().trim_end_matches('.') == record_name.trim_end_matches('.') {
+                let change = Change::builder().action(ChangeAction::Delete).resource_record_set(record_set.clone()).build()?;
+
+                let change_batch = ChangeBatch::builder().changes(change).build()?;
+
+                self.client
+                    .change_resource_record_sets()
+                    .hosted_zone_id(&self.hosted_zone_id)
+                    .change_batch(change_batch)
+                    .send()
+                    .await?;
+
+                info!("Deleted Route53 TXT record name: {}", record_name);
+                return Ok(());
+            }
+        }
+
+        info!("Route53 TXT record name: {} already deleted or not found", record_name);
+        Ok(())
+    }
+}
+
+inventory::submit! {
+    DnsBackendPlugin {
+        name: "route53",
+        factory: || Box::new(Route53Provider::new()),
+    }
+}

src/tls/acme/http01.rs

@@ -14,32 +14,32 @@ pub async fn acme_create(State(state): State<webserver::AppState>) -> impl IntoR
                 .status(StatusCode::OK)
                 .header("Content-Type", "text/plain")
                 .body(Body::from(txt))
-                .unwrap()
+                .unwrap();
         }
         Err(e) => {
             return Response::builder()
                 .status(StatusCode::INTERNAL_SERVER_ERROR)
                 .body(Body::from(format!("Failed to create account: {}", e)))
-                .unwrap()
+                .unwrap();
         }
     };
 }
 #[allow(clippy::needless_return)]
 pub async fn acme_order(State(state): State<webserver::AppState>, axum::extract::Path(domain): axum::extract::Path<String>) -> impl IntoResponse {
     let domain_clean = domain.trim_matches('/');
-    match order::order(domain_clean, state.cert_creds.as_str(), state.certs_dir).await {
+    match order::order(domain_clean, state.cert_creds.as_str(), state.certs_dir, state.acme_dns_provider).await {
         Ok(txt) => {
             return Response::builder()
                 .status(StatusCode::OK)
                 .header("Content-Type", "text/plain")
                 .body(Body::from(txt))
-                .unwrap()
+                .unwrap();
         }
         Err(e) => {
             return Response::builder()
                 .status(StatusCode::INTERNAL_SERVER_ERROR)
                 .body(Body::from(format!("Failed to order a certificate: {}", e)))
-                .unwrap()
+                .unwrap();
         }
     };
 }

src/tls/acme/lookup.rs

@@ -0,0 +1,45 @@
+use hickory_resolver::TokioResolver;
+use hickory_resolver::config::{NameServerConfig, ResolverConfig, ResolverOpts};
+use hickory_resolver::net::runtime::TokioRuntimeProvider;
+use hickory_resolver::proto::rr::RData;
+use log::info;
+use std::net::IpAddr;
+use std::str::FromStr;
+use std::time::Duration;
+
+pub async fn lookup_wait(record_name: &str, dns_server_ip: Option<&str>, max_wait_secs: u64, expected_value: &str) -> bool {
+    let target_ip = dns_server_ip.and_then(|ip| IpAddr::from_str(ip).ok()).unwrap_or_else(|| IpAddr::from([1, 1, 1, 1]));
+    let name_server = NameServerConfig::udp(target_ip);
+    let config = ResolverConfig::from_name_servers(vec![name_server]);
+    let mut opts = ResolverOpts::default();
+    opts.cache_size = 0;
+
+    let resolver = TokioResolver::builder_with_config(config, TokioRuntimeProvider::default())
+        .with_options(opts)
+        .build()
+        .expect("failed to build DNS resolver");
+
+    let start = std::time::Instant::now();
+    let poll_interval = Duration::from_secs(3);
+    while start.elapsed().as_secs() < max_wait_secs {
+        if let Ok(txt_lookup) = resolver.txt_lookup(record_name).await {
+            for record in txt_lookup.answers() {
+                if let RData::TXT(txt) = &record.data {
+                    for data in &txt.txt_data {
+                        if let Ok(txt_str) = std::str::from_utf8(data) {
+                            if txt_str == expected_value {
+                                tokio::time::sleep(Duration::from_secs(5)).await;
+                                info!("DNS record propagated successfully for {} : {}", record_name, txt_str);
+                                return true;
+                            }
+                        }
+                    }
+                }
+            }
+        } else {
+            info!("Waiting for TXT record {}", record_name);
+        }
+        tokio::time::sleep(poll_interval).await;
+    }
+    false
+}

src/tls/acme/mod.rs

@@ -1,6 +1,9 @@
 pub mod account;
+pub mod dns;
 pub mod http01;
+pub mod lookup;
 pub mod order;
+pub mod types;
 
 pub use account::*;
 pub use order::*;

src/tls/acme/order.rs

@@ -1,36 +1,37 @@
 use crate::tls::acme::account::get_account;
+use crate::tls::acme::types::get_provider;
 use crate::utils::lazylock::CHALLENGES;
-use crate::utils::parceyaml::DOMAINS;
+use crate::utils::lazylock::DOMAINS;
+use base64::Engine;
+use base64::engine::general_purpose::URL_SAFE_NO_PAD;
 use instant_acme::{ChallengeType, Identifier, NewOrder, RetryPolicy};
 use log::{error, info};
 use pingora::prelude::sleep;
 use rcgen::{CertificateParams, DistinguishedName, KeyPair};
+use sha2::{Digest, Sha256};
 use std::fs;
 use std::time::Duration;
 use x509_parser::prelude::*;
 
-pub async fn refresh_order(certs_dir: String, autoconf_dir: String) {
+pub async fn refresh_order(certs_dir: String, autoconf_dir: String, provider: Option<String>) {
     let credsfile = autoconf_dir + "/acme_credentials.json";
     loop {
         for item in DOMAINS.iter() {
-            let _what = order(item.key(), credsfile.as_str(), certs_dir.clone()).await;
+            let _what = order(item.key(), credsfile.as_str(), certs_dir.clone(), provider.clone()).await;
         }
         sleep(Duration::from_secs(12 * 3600)).await;
     }
 }
-pub async fn order(domain: &str, credsfile: &str, certs_dir: String) -> Result<String, Box<dyn std::error::Error>> {
-    let crt = certs_dir.clone() + "/" + domain + ".crt";
-    let key = certs_dir.clone() + "/" + domain + ".key";
+pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: Option<String>) -> Result<String, Box<dyn std::error::Error>> {
+    let crt = format!("{}/{}.crt", certs_dir, domain);
+    let key = format!("{}/{}.key", certs_dir, domain);
 
     if DOMAINS.get(domain).is_none() {
         DOMAINS.insert(domain.to_string(), true);
-        let mut newlist: Vec<String> = Vec::new();
-        for item in DOMAINS.iter() {
-            newlist.push(item.key().to_string());
-        }
+        let newlist: Vec<String> = DOMAINS.iter().map(|item| item.key().to_string()).collect();
         if let Ok(json_content) = serde_json::to_string_pretty(&newlist) {
             let autocfg_file = credsfile.replace("/acme_credentials.json", "/domains.json");
-            if let Err(err) = std::fs::write(&autocfg_file, json_content) {
+            if let Err(err) = fs::write(&autocfg_file, json_content) {
                 error!("Error Updating domains for certificates: {} : {}", domain, err);
                 return Err(Box::from(err));
             }
@@ -40,7 +41,6 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String) -> Result<S
     if let Ok(expiry) = cert_expiry(crt.as_str()) {
         let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs();
         if expiry > now + 30 * 24 * 3600 {
-            // println!("Fresh certificate exists. Not renewing !");
             return Ok("Fresh certificate exists. Not renewing ! \n".to_string());
         }
     };
@@ -49,14 +49,43 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String) -> Result<S
     let mut order = account.new_order(&NewOrder::new(&[Identifier::Dns(domain.to_string())])).await?;
 
     let mut authorizations = order.authorizations();
+
     while let Some(auth) = authorizations.next().await {
         let mut auth = auth?;
-        let mut challenge_handle = auth.challenge(ChallengeType::Http01).ok_or("no http01 challenge found")?;
+        let domain_name = auth.identifier().to_string();
+        let (challenge_type, is_dns) = match &provider {
+            Some(_) => (ChallengeType::Dns01, true),
+            None => (ChallengeType::Http01, false),
+        };
+
+        let mut challenge_handle = auth.challenge(challenge_type).ok_or("Requested challenge type not offered by ACME server")?;
         let key_auth = challenge_handle.key_authorization();
         let key_auth_str = key_auth.as_str().to_string();
-        let token = key_auth_str.split('.').next().ok_or("invalid key authorization")?.to_string();
-        CHALLENGES.write().unwrap().insert(token, key_auth_str);
-        challenge_handle.set_ready().await?;
+
+        if is_dns {
+            let prov = provider.as_ref().unwrap();
+            let dns_provider = get_provider(prov.clone());
+            let dns_value = calculate_dns_value(&key_auth_str);
+            let clean_domain = domain_name.trim_start_matches("*.").to_string();
+            let record_name = format!("_acme-challenge.{}", clean_domain);
+
+            match dns_provider.create_txt_record(&clean_domain, &record_name, &dns_value).await {
+                Ok(record_id) => {
+                    let ready_res = challenge_handle.set_ready().await;
+                    let _ = dns_provider.delete_txt_record(&record_id, &record_name).await;
+                    ready_res?;
+                }
+                Err(e) => {
+                    eprintln!("Failed to create DNS record: {}", e);
+                    return Err(e);
+                }
+            }
+        } else {
+            let token = key_auth_str.split('.').next().ok_or("Invalid key authorization")?.to_string();
+            CHALLENGES.write().unwrap().insert(token, key_auth_str);
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+            challenge_handle.set_ready().await?;
+        }
     }
 
     let status = order.poll_ready(&RetryPolicy::default()).await?;
@@ -68,12 +97,9 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String) -> Result<S
     let signing_request = params.serialize_request(&private_key)?;
     let csr_der = signing_request.der();
     order.finalize_csr(csr_der).await?;
-
-    // poll for certificate
     let cert_chain_pem = order.poll_certificate(&RetryPolicy::default()).await?;
     CHALLENGES.write().unwrap().clear();
     let private_key_pem = private_key.serialize_pem();
-
     fs::write(crt, cert_chain_pem)?;
     fs::write(key, private_key_pem)?;
     Ok("Certificate is successfully generated \n".to_string())
@@ -86,3 +112,8 @@ fn cert_expiry(path: &str) -> Result<u64, Box<dyn std::error::Error>> {
     let expiry = cert.validity().not_after.timestamp() as u64;
     Ok(expiry)
 }
+
+fn calculate_dns_value(key_auth: &str) -> String {
+    let hash = Sha256::digest(key_auth.as_bytes());
+    URL_SAFE_NO_PAD.encode(hash)
+}

src/tls/acme/types.rs

@@ -0,0 +1,32 @@
+use async_trait::async_trait;
+use std::collections::HashMap;
+use std::sync::LazyLock;
+
+pub struct DnsBackendPlugin {
+    pub name: &'static str,
+    pub factory: fn() -> Box<dyn DnsProvider>,
+}
+
+inventory::collect!(DnsBackendPlugin);
+
+static FACTORIES: LazyLock<HashMap<&'static str, fn() -> Box<dyn DnsProvider>>> = LazyLock::new(|| {
+    let mut map = HashMap::new();
+    for plugin in inventory::iter::<DnsBackendPlugin> {
+        map.insert(plugin.name, plugin.factory);
+    }
+    map
+});
+
+#[async_trait]
+pub trait DnsProvider: Send + Sync {
+    async fn create_txt_record(&self, domain: &str, name: &str, value: &str) -> Result<String, Box<dyn std::error::Error + Send + Sync>>;
+
+    async fn delete_txt_record(&self, record_id: &str, record_name: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
+}
+
+pub fn get_provider(provider: String) -> Box<dyn DnsProvider> {
+    FACTORIES
+        .get(provider.as_str())
+        .map(|factory| factory())
+        .unwrap_or_else(|| panic!("Unknown DNS provider '{}'. Available providers: {:?}", provider, FACTORIES.keys().collect::<Vec<_>>()))
+}

src/tls/grades.rs

@@ -22,7 +22,7 @@ pub enum TlsGrade {
 }
 
 impl TlsGrade {
-    pub fn from_str(s: &str) -> Option<Self> {
+    pub fn parse(s: &str) -> Option<Self> {
         match s.to_ascii_lowercase().as_str() {
             "high" => Some(TlsGrade::High),
             "medium" => Some(TlsGrade::Medium),
@@ -39,7 +39,7 @@ pub fn prefer_h2<'a>(_ssl: &mut SslRef, alpn_in: &'a [u8]) -> Result<&'a [u8], A
 }
 
 pub fn set_tsl_grade(tls_settings: &mut TlsSettings, grade: &str) {
-    let config_grade = TlsGrade::from_str(grade);
+    let config_grade = TlsGrade::parse(grade);
     match config_grade {
         Some(TlsGrade::High) => {
             let _ = tls_settings.set_min_proto_version(Some(SslVersion::TLS1_2));

src/tls/load.rs

@@ -54,7 +54,7 @@ impl Certificates {
                     cert_infos.push(cert)
                 }
                 None => {
-                    error!("Unable to load certificate info | public: {}, private: {}", &config.cert_path, &config.key_path);
+                    error!("Unable to load certificate info | public: {}, private: {}", config.cert_path, config.key_path);
                     return None;
                 }
             }

src/utils/deser.rs

@@ -0,0 +1,297 @@
+use crate::utils::hcclient::{HttpClientRequest, httpclient};
+use crate::utils::lazylock::{GLOBAL_PARAMS, MASTER_KEY, WORKERS};
+use crate::utils::types::{Consul, Extraparams, Headers, Kubernetes, UpstreamsDashMap};
+use bytes::Bytes;
+use log::{error, info, warn};
+use noyalib::to_string;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::ops::Deref;
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct UpstreamConfigDto {
+    pub paths: HashMap<String, PathConfigDto>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct PathConfigDto {
+    pub is_http2: bool,
+    pub healthcheck: bool,
+    pub is_ssl: bool,
+    pub to_https: bool,
+    pub rate_limit: isize,
+    pub x4xx_limit: u32,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub redirect_to: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub authorization: Option<HashMap<String, String>>,
+    pub servers: Vec<String>,
+    // #[serde(skip_serializing_if = "Option::is_none")]
+    // pub client_headers: Option<Vec<String>>,
+    // #[serde(skip_serializing_if = "Option::is_none")]
+    // pub server_headers: Option<Vec<String>>,
+}
+
+#[derive(Clone, Debug)]
+pub struct SendConfigDto {
+    pub upstreams: HashMap<String, UpstreamConfigDto>,
+    pub client_headers: Headers,
+    pub server_headers: Headers,
+    pub consul: Option<Consul>,
+    pub kubernetes: Option<Kubernetes>,
+    pub typecfg: String,
+    pub extraparams: Extraparams,
+}
+
+pub fn serialize_config(data: &UpstreamsDashMap) -> String {
+    let config = GLOBAL_PARAMS.read().unwrap();
+    let mut snapshot: HashMap<String, UpstreamConfigDto> = HashMap::new();
+
+    for outer_entry in data.iter() {
+        let domain = outer_entry.key().to_string();
+        let mut paths: HashMap<String, PathConfigDto> = HashMap::new();
+
+        for inner_entry in outer_entry.value().iter() {
+            let path = inner_entry.key().to_string();
+            let (vec_inner, _atomic_val) = inner_entry.value();
+
+            if let Some(first) = vec_inner.first() {
+                let servers = vec_inner.iter().map(|item| format!("{}:{}", item.address, item.port)).collect();
+                let mut auth = None;
+                if let Some(ath) = first.authorization.clone() {
+                    let mut a = HashMap::new();
+                    a.insert("type".to_string(), ath.auth_type.clone().to_string());
+                    a.insert("data".to_string(), ath.auth_cred.clone().to_string());
+                    auth = Some(a);
+                }
+                paths.insert(
+                    path,
+                    PathConfigDto {
+                        is_http2: first.is_http2,
+                        healthcheck: first.healthcheck,
+                        is_ssl: first.is_ssl,
+                        to_https: first.to_https,
+                        rate_limit: first.rate_limit,
+                        x4xx_limit: first.x4xx_limit,
+                        redirect_to: first.redirect_to.clone().as_ref().map(|s| s.to_string()),
+                        authorization: auth,
+                        servers,
+                    },
+                );
+            }
+        }
+
+        snapshot.insert(domain, UpstreamConfigDto { paths });
+    }
+    let payload = SendConfigDto {
+        upstreams: snapshot,
+        client_headers: config.client_headers.clone(),
+        server_headers: config.server_headers.clone(),
+        consul: config.consul.clone(),
+        kubernetes: config.kubernetes.clone(),
+        typecfg: config.typecfg.clone(),
+        extraparams: config.extraparams.clone(),
+    };
+    to_yaml(&payload)
+}
+
+#[derive(Serialize, Debug)]
+struct YamlAuth {
+    #[serde(rename = "type")]
+    auth_type: String,
+    data: String,
+}
+
+#[derive(Serialize)]
+struct YamlPath {
+    servers: Vec<String>,
+    #[serde(skip_serializing_if = "is_false")]
+    is_http2: bool,
+    #[serde(skip_serializing_if = "is_false")]
+    to_https: bool,
+    #[serde(skip_serializing_if = "is_false")]
+    is_ssl: bool,
+    #[serde(skip_serializing_if = "is_true")] // healthcheck true is default
+    healthcheck: bool,
+    #[serde(skip_serializing_if = "is_zero_isize")]
+    rate_limit: isize,
+    #[serde(skip_serializing_if = "is_zero_u32")]
+    x4xx_limit: u32,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    redirect_to: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    authorization: Option<YamlAuth>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    client_headers: Option<Vec<String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    server_headers: Option<Vec<String>>,
+}
+
+#[derive(Serialize)]
+struct YamlUpstream {
+    paths: HashMap<String, YamlPath>,
+}
+
+#[derive(Serialize)]
+struct YamlConfig {
+    provider: String,
+    sticky_sessions: u64,
+    to_https: bool,
+    rate_limit: isize,
+    x4xx_limit: u32,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    server_headers: Vec<String>,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    client_headers: Vec<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    authorization: Option<YamlAuth>,
+    upstreams: HashMap<String, YamlUpstream>,
+}
+
+fn is_false(v: &bool) -> bool {
+    !v
+}
+fn is_true(v: &bool) -> bool {
+    *v
+}
+fn is_zero_isize(v: &isize) -> bool {
+    *v == 0
+}
+fn is_zero_u32(v: &u32) -> bool {
+    *v == 0
+}
+
+pub fn to_yaml(config: &SendConfigDto) -> String {
+    let server_headers: Vec<String> = config
+        .server_headers
+        .get("GLOBAL_SERVER_HEADERS")
+        .map(|paths| {
+            paths
+                .value()
+                .iter()
+                .flat_map(|inner| inner.value().iter().map(|(k, v)| format!("{}:{}", k, v)).collect::<Vec<_>>())
+                .collect()
+        })
+        .unwrap_or_default();
+
+    let client_headers: Vec<String> = config
+        .client_headers
+        .get("GLOBAL_CLIENT_HEADERS")
+        .map(|paths| {
+            paths
+                .value()
+                .iter()
+                .flat_map(|inner| inner.value().iter().map(|(k, v)| format!("{}:{}", k, v)).collect::<Vec<_>>())
+                .collect()
+        })
+        .unwrap_or_default();
+
+    let authorization = config.extraparams.authentication.as_ref().map(|auth| YamlAuth {
+        auth_type: auth.auth_type.to_string(),
+        data: auth.auth_cred.to_string(),
+    });
+
+    let upstreams: HashMap<String, YamlUpstream> = config
+        .upstreams
+        .iter()
+        .map(|(domain, upstream)| {
+            let paths = upstream
+                .paths
+                .iter()
+                .map(|(path, pcfg)| {
+                    let ch: Option<Vec<String>> = {
+                        if let Some(domain_ref) = config.client_headers.get(domain.as_str()) {
+                            if let Some(path_ref) = domain_ref.get(path.as_str()) {
+                                let collected: Vec<String> = path_ref.value().iter().map(|(k, v)| format!("{}:{}", k, v)).collect();
+                                if !collected.is_empty() { Some(collected) } else { None }
+                            } else {
+                                None
+                            }
+                        } else {
+                            None
+                        }
+                    };
+
+                    let sh: Option<Vec<String>> = {
+                        if let Some(domain_ref) = config.server_headers.get(domain.as_str()) {
+                            if let Some(path_ref) = domain_ref.get(path.as_str()) {
+                                let collected: Vec<String> = path_ref.value().iter().map(|(k, v)| format!("{}:{}", k, v)).collect();
+                                if !collected.is_empty() { Some(collected) } else { None }
+                            } else {
+                                None
+                            }
+                        } else {
+                            None
+                        }
+                    };
+
+                    let auth = pcfg.authorization.as_ref().map(|a| YamlAuth {
+                        auth_type: a.get("type").cloned().unwrap_or_default(),
+                        data: a.get("data").cloned().unwrap_or_default(),
+                    });
+
+                    (
+                        path.clone(),
+                        YamlPath {
+                            servers: pcfg.servers.clone(),
+                            is_http2: pcfg.is_http2,
+                            is_ssl: pcfg.is_ssl,
+                            healthcheck: pcfg.healthcheck,
+                            rate_limit: pcfg.rate_limit,
+                            x4xx_limit: pcfg.x4xx_limit,
+                            to_https: pcfg.to_https,
+                            redirect_to: pcfg.redirect_to.clone(),
+                            authorization: auth,
+                            client_headers: ch,
+                            server_headers: sh,
+                        },
+                    )
+                })
+                .collect();
+            (domain.clone(), YamlUpstream { paths })
+        })
+        .collect();
+
+    let yaml_config = YamlConfig {
+        provider: config.typecfg.clone(),
+        sticky_sessions: config.extraparams.sticky_sessions,
+        to_https: config.extraparams.to_https,
+        rate_limit: config.extraparams.rate_limit,
+        x4xx_limit: config.extraparams.x4xx_limit,
+        server_headers,
+        client_headers,
+        authorization,
+        upstreams,
+    };
+
+    to_string(&yaml_config).unwrap_or_default()
+}
+
+pub async fn send_to_worker(data: String) {
+    let masterkey = MASTER_KEY.get().expect("Master key not set");
+    if let Some(workers) = WORKERS.get() {
+        for worker in workers {
+            let prefix = if worker.tls { "https://" } else { "http://" };
+            let worker_url = format!("{}{}:{}", prefix, worker.address, worker.port);
+            let status = httpclient(HttpClientRequest {
+                method: "POST",
+                tls: worker.tls,
+                host: worker.address.as_ref(),
+                path: "/workerconf",
+                address: worker.address.as_ref(),
+                port: worker.port,
+                url: worker_url.as_str(),
+                payload: Bytes::from(data.clone()),
+                headers: Some(vec![("API-KEY", masterkey.clone().expect("Master key must be set in the cluster mode").deref())]),
+            })
+            .await;
+            if (200..500).contains(&status.2) {
+                info!("Successfully updated Worker {}, status {}", worker_url, status.2);
+            } else {
+                error!("Cannot connect to worker node, status {}", status.2);
+            }
+        }
+    } else {
+        warn!("Workers list has not been initialized yet.");
+    }
+}

src/utils/fordebug.rs

@@ -8,12 +8,13 @@ pub static DEALLOC_COUNT: AtomicUsize = AtomicUsize::new(0);
 pub static ALLOC_BYTES: AtomicUsize = AtomicUsize::new(0);
 #[allow(dead_code)]
 unsafe impl GlobalAlloc for CountingAllocator {
+    #[allow(unsafe_op_in_unsafe_fn)]
     unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
         ALLOC_COUNT.fetch_add(1, Ordering::Relaxed);
         ALLOC_BYTES.fetch_add(layout.size(), Ordering::Relaxed);
         System.alloc(layout)
     }
-
+    #[allow(unsafe_op_in_unsafe_fn)]
     unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
         DEALLOC_COUNT.fetch_add(1, Ordering::Relaxed);
         System.dealloc(ptr, layout)

src/utils/hcclient.rs

@@ -9,7 +9,30 @@ use std::time::Duration;
 
 pub static HC_CONNECTOR: LazyLock<Connector> = LazyLock::new(|| Connector::new(None));
 
-pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address: &str, port: u16, url: &str, payload: Bytes) -> (bool, bool) {
+pub struct HttpClientRequest<'a> {
+    pub method: &'a str,
+    pub tls: bool,
+    pub host: &'a str,
+    pub path: &'a str,
+    pub address: &'a str,
+    pub port: u16,
+    pub url: &'a str,
+    pub payload: Bytes,
+    pub headers: Option<Vec<(&'a str, &'a str)>>, // ← add this
+}
+
+pub async fn httpclient(req: HttpClientRequest<'_>) -> (bool, bool, u16) {
+    let HttpClientRequest {
+        method,
+        tls,
+        host,
+        path,
+        address,
+        port,
+        url,
+        payload,
+        headers,
+    } = req;
     let method = match method {
         "HEAD" => "HEAD",
         "GET" => "GET",
@@ -36,7 +59,7 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address
             } else {
                 log::warn!("Fail to connect to addr: {}, tls {}, host {}", address, tls, host);
             }
-            return (false, false);
+            return (false, false, 500);
         }
     };
 
@@ -47,7 +70,7 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address
         }
         None => {
             if ping_grpc(url).await {
-                return (true, true);
+                return (true, true, 204);
             }
         }
     }
@@ -56,7 +79,7 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address
         Ok(r) => r,
         Err(e) => {
             log::warn!("Failed to build request: {}", e);
-            return (false, false);
+            return (false, false, 500);
         }
     };
 
@@ -64,113 +87,33 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address
     let body_bytes = if method == "POST" {
         req.insert_header("Content-Type", "application/json").ok();
         req.insert_header("Content-Length", payload.len().to_string()).ok();
+        if let Some(headers) = headers {
+            for (k, v) in headers {
+                req.insert_header(k.to_owned(), v).ok();
+            }
+        }
         payload
     } else {
         Bytes::new()
     };
     if let Err(e) = http_session.write_request_header(Box::new(req)).await {
         log::warn!("Write header failed: {}", e);
-        return (false, false);
+        return (false, false, 500);
     }
     if let Err(e) = http_session.write_request_body(body_bytes, true).await {
         log::warn!("Write body failed: {}", e);
-        return (false, false);
+        return (false, false, 500);
     }
     let status = match http_session.read_response_header().await {
         Ok(_) => http_session.response_header().map(|r| r.status.as_u16()).unwrap_or(500),
         Err(e) => {
             if !is_h2 && ping_grpc(url).await {
-                return (true, true);
+                return (true, true, 204);
             }
             log::warn!("Health Check read failed ({}) : {} - {}", if is_h2 { "H2" } else { "H1" }, host, e);
-            return (false, false);
+            return (false, false, 500);
         }
     };
     HC_CONNECTOR.release_http_session(http_session, &peer, None).await;
-    ((200..500).contains(&status), is_h2)
+    ((200..500).contains(&status), is_h2, status)
 }
-/*
-pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address: &str, port: u16, url: String) -> (bool, bool) {
-    let method = match method {
-        "HEAD" => "HEAD",
-        "GET" => "GET",
-        "POST" => "POST",
-        _ => "GET",
-    };
-    let mut peer = HttpPeer::new((address, port), tls, host.to_string());
-
-    if tls {
-        peer.options.verify_cert = false;
-        peer.options.verify_hostname = false;
-    }
-
-    let mut is_h2 = false;
-    peer.options.total_connection_timeout = Option::from(Duration::from_secs(5));
-    peer.options.alpn = ALPN::H2H1;
-    let (mut http_session, _) = match HC_CONNECTOR.get_http_session(&peer).await {
-        Ok(s) => s,
-        Err(e) => {
-            if let Some(msg) = e.context {
-                log::warn!("{}, type: {}", msg.as_str(), e.etype.as_str());
-            } else {
-                log::warn!("Fail to connect to addr: {}, tls {}, host {}", address, tls, host);
-            }
-            return (false, false);
-        }
-    };
-
-    match http_session.as_http2() {
-        Some(_) => {
-            peer.options.alpn = ALPN::H2;
-            is_h2 = true;
-        }
-        None => {
-            if ping_grpc(url.as_str()).await {
-                return (true, true);
-            }
-        }
-    }
-
-    let mut req = match RequestHeader::build(method, path.as_bytes(), None) {
-        Ok(r) => r,
-        Err(e) => {
-            log::warn!("Failed to build request: {}", e);
-            return (false, false);
-        }
-    };
-
-    req.insert_header("Host", host).ok();
-
-    if let Err(e) = http_session.write_request_header(Box::new(req)).await {
-        log::warn!("Write header failed: {}", e);
-        return (false, false);
-    }
-
-    let status = if is_h2 {
-        if let Err(e) = http_session.write_request_body(Bytes::new(), true).await {
-            log::warn!("Write body failed: {}", e);
-            return (false, false);
-        }
-        match http_session.read_response_header().await {
-            Ok(_) => http_session.response_header().map(|r| r.status.as_u16()).unwrap_or(500),
-            Err(e) => {
-                if ping_grpc(url.as_str()).await {
-                    return (true, true);
-                }
-                log::warn!("Health Check H2 read failed: {} - {}", host, e);
-                return (false, false);
-            }
-        }
-    } else {
-        match http_session.read_response_header().await {
-            Ok(_) => http_session.response_header().map(|r| r.status.as_u16()).unwrap_or(500),
-            Err(e) => {
-                log::warn!("Health Check H1 read failed: {} - {}", e, host);
-                return (false, false);
-            }
-        }
-    };
-    HC_CONNECTOR.release_http_session(http_session, &peer, None).await;
-    ((200..500).contains(&status), is_h2)
-}
-*/

src/utils/healthcheck.rs

@@ -1,38 +1,54 @@
-use crate::utils::hcclient::httpclient;
-use crate::utils::lazylock::REVERSE_STORE;
+use crate::utils::deser::{send_to_worker, serialize_config};
+use crate::utils::hcclient::{HttpClientRequest, httpclient};
+use crate::utils::lazylock::{IM_A_MASTER, NODE_STATUS, 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 log::info;
 use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
 use std::time::Duration;
 use tokio::net::TcpStream;
 use tokio::time::{interval, timeout};
 use tokio_native_tls::native_tls::TlsConnector;
+
 const EMPTY_PAYLOAD: Bytes = Bytes::from_static(b"");
 
 pub async fn hc2(upslist: Arc<UpstreamsDashMap>, fullist: Arc<UpstreamsDashMap>, idlist: Arc<UpstreamsIdMap>, params: (&str, u64)) {
     let mut period = interval(Duration::from_secs(params.1));
+    if let Some(member) = NODE_STATUS.get() {
+        if let Some(mbr) = member.as_ref() {
+            match mbr.as_ref() {
+                "worker" => {
+                    info!("Disabling health-checks. Running in cluster mode as {}", mbr);
+                    return;
+                }
+                "master" => {
+                    info!("Disabling client/server routing. Running in cluster mode as {}", mbr);
+                }
+                _ => {}
+            }
+        }
+    }
     loop {
         tokio::select! {
             _ = period.tick() => {
-                let totest = build_upstreams(&fullist, params.0).await;
+                let totest = initiate_upstreams(&fullist, params.0).await;
                 if !compare_dashmaps(&totest, &upslist) {
                     clone_dashmap_into(&totest, &upslist);
                     clone_idmap_into(&totest, &idlist);
+                    if let Some(true) = IM_A_MASTER.get(){
+                        let binary_output = serialize_config(&totest);
+                        let _ = send_to_worker(binary_output).await;
+                    }
                     REVERSE_STORE.clear();
                 }
             }
         }
     }
 }
-
-pub async fn initiate_upstreams(fullist: UpstreamsDashMap) -> UpstreamsDashMap {
-    build_upstreams(&fullist, "HEAD").await
-}
-
-async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsDashMap {
+pub(crate) async fn initiate_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsDashMap {
     let totest: UpstreamsDashMap = DashMap::new();
     let fclone = clone_dashmap(fullist);
     for val in fclone.iter() {
@@ -44,16 +60,16 @@ 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
+                    upstream.is_ssl
                 };
                 let mut scheme = InnerMap {
                     address: upstream.address.clone(),
                     port: upstream.port,
                     is_ssl: tls,
-                    is_http2: false,
+                    is_http2: upstream.is_http2,
                     to_https: upstream.to_https,
                     rate_limit: upstream.rate_limit,
                     x4xx_limit: upstream.x4xx_limit,
@@ -62,13 +78,24 @@ 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 {
                         format!("http://{}:{}{}", upstream.address, upstream.port, path)
                     };
-                    let resp = httpclient(method, tls, host, path, upstream.address.as_ref(), upstream.port, link.as_str(), EMPTY_PAYLOAD).await;
+                    let resp = httpclient(HttpClientRequest {
+                        method,
+                        tls,
+                        host,
+                        path,
+                        address: upstream.address.as_ref(),
+                        port: upstream.port,
+                        url: link.as_str(),
+                        payload: EMPTY_PAYLOAD,
+                        headers: None,
+                    })
+                    .await;
                     if resp.0 {
                         if resp.1 {
                             scheme.is_http2 = resp.1;

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,
                             });
@@ -154,7 +154,7 @@ pub async fn for_kuber(url: &str, token: &str, conf: &GlobalServiceMapping) -> O
 }
 
 pub async fn getfromapi(url: &str, token: Option<String>, provider: &str) -> Option<Vec<u8>> {
-    let (host, port, path, is_tls) = parse_url(&url).ok()?;
+    let (host, port, path, is_tls) = parse_url(url).ok()?;
 
     let mut peer = HttpPeer::new((host, port), is_tls, host.to_string());
     peer.options.total_connection_timeout = Some(Duration::from_secs(5));
@@ -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/lazylock.rs

@@ -1,8 +1,9 @@
+use crate::utils::types::{Configuration, WorkerNodes};
 use dashmap::DashMap;
 use moka::sync::Cache;
+use pingora_cache::MemCache;
 use pingora_cache::eviction::simple_lru::Manager;
 use pingora_cache::lock::CacheLock;
-use pingora_cache::MemCache;
 use pingora_limits::rate::Rate;
 use std::collections::HashMap;
 use std::net::IpAddr;
@@ -19,3 +20,9 @@ pub static CACHE_LOCK: LazyLock<CacheLock> = LazyLock::new(|| CacheLock::new(Dur
 pub static EVICTION: OnceLock<Manager> = OnceLock::new();
 pub static CACHE_TTL: OnceLock<u64> = OnceLock::new();
 pub static CHALLENGES: LazyLock<RwLock<HashMap<String, String>>> = LazyLock::new(|| RwLock::new(HashMap::new()));
+pub static DOMAINS: LazyLock<DashMap<String, bool>> = LazyLock::new(DashMap::new);
+pub static NODE_STATUS: OnceLock<Option<Arc<str>>> = OnceLock::new();
+pub static MASTER_KEY: OnceLock<Option<Arc<str>>> = OnceLock::new();
+pub static GLOBAL_PARAMS: LazyLock<RwLock<Configuration>> = LazyLock::new(|| RwLock::new(Configuration::default()));
+pub static IM_A_MASTER: OnceLock<bool> = OnceLock::new();
+pub static WORKERS: OnceLock<Vec<WorkerNodes>> = OnceLock::new();

src/utils/mod.rs

@@ -4,6 +4,7 @@ pub mod hcclient;
 pub mod healthcheck;
 pub mod httpclient;
 // pub mod ingress;
+pub mod deser;
 pub mod lazylock;
 pub mod metrics;
 pub mod parceyaml;
@@ -11,4 +12,5 @@ pub mod state;
 pub mod tools;
 pub mod types;
 pub(crate) mod watch;
+
 pub use watch::*;

src/utils/parceyaml.rs

@@ -1,18 +1,17 @@
+use crate::utils::deser::{send_to_worker, serialize_config};
 use crate::utils::healthcheck;
-use crate::utils::lazylock::REVERSE_STORE;
+use crate::utils::lazylock::{DOMAINS, GLOBAL_PARAMS, IM_A_MASTER, MASTER_KEY, NODE_STATUS, REVERSE_STORE, WORKERS};
 use crate::utils::state::{is_first_run, mark_not_first_run};
-use crate::utils::tools::{clone_dashmap, clone_dashmap_into, print_upstreams};
+use crate::utils::tools::clone_dashmap_into;
 use crate::utils::types::*;
 use dashmap::DashMap;
 use log::{error, info, warn};
 use std::collections::HashMap;
 use std::path::Path;
+use std::sync::Arc;
 use std::sync::atomic::AtomicUsize;
-use std::sync::{Arc, LazyLock};
 use std::{env, fs};
 
-pub static DOMAINS: LazyLock<DashMap<String, bool>> = LazyLock::new(DashMap::new);
-
 pub async fn load_configuration(d: &str, kind: &str) -> (Option<Configuration>, String) {
     let mut conf_files = Vec::new();
     let yaml_data = match kind {
@@ -172,10 +171,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);
@@ -225,12 +224,12 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) {
                             server_list.push(Arc::from(InnerMap {
                                 address: Arc::from(ip),
                                 port,
-                                is_ssl: false,
-                                is_http2: false,
+                                is_ssl: path_config.is_ssl.unwrap_or(false),
+                                is_http2: path_config.is_http2.unwrap_or(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,
                             }));
@@ -241,28 +240,48 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) {
             }
             config.client_headers.insert(Arc::from(hostname.clone()), client_header_list);
             config.server_headers.insert(Arc::from(hostname.clone()), server_header_list);
+
             imtdashmap.insert(Arc::from(hostname.clone()), path_map);
         }
 
+        {
+            let mut config_guard = GLOBAL_PARAMS.write().expect("Global config lock poisoned");
+            config_guard.client_headers = config.client_headers.clone();
+            config_guard.server_headers = config.server_headers.clone();
+            config_guard.extraparams = config.extraparams.clone();
+            config_guard.kubernetes = config.kubernetes.clone();
+            config_guard.consul = config.consul.clone();
+            config_guard.typecfg = config.typecfg.clone();
+        }
+
         if is_first_run() {
             clone_dashmap_into(&imtdashmap, &config.upstreams);
             mark_not_first_run();
         } else {
-            let y = clone_dashmap(&imtdashmap);
-            let r = healthcheck::initiate_upstreams(y).await;
+            let r = healthcheck::initiate_upstreams(&imtdashmap, "HEAD").await;
             clone_dashmap_into(&r, &config.upstreams);
         }
         info!("Upstream Config:");
         REVERSE_STORE.clear();
-        print_upstreams(&config.upstreams, &config.extraparams);
+
+        if let Some(true) = IM_A_MASTER.get() {
+            if !is_first_run() {
+                let binary_output = serialize_config(&config.upstreams);
+                let _ = send_to_worker(binary_output).await;
+            }
+        }
+        // print_upstreams(&config.upstreams, &config.extraparams);
     }
 }
 pub fn parce_main_config(path: &str) -> AppConfig {
     let data = fs::read_to_string(path).expect("Failed to read main config file");
     let mut cfo: AppConfig = noyalib::from_str(&data).expect("Failed to parse main config file");
-    if let Ok(jwt_key) = env::var("JWT_KEY") {
+    if let Ok(jwt_key) = env::var("MASTER_KEY") {
+        cfo.master_key = Some(jwt_key);
+    } else if let Ok(jwt_key) = env::var("JWT_KEY") {
         cfo.master_key = Some(jwt_key);
     };
+
     // log_builder(&cfo, &cfo.log_file);
     cfo.hc_method = cfo.hc_method.to_uppercase();
     if let Some((ip, port_str)) = cfo.config_address.rsplit_once(':') {
@@ -281,6 +300,39 @@ pub fn parce_main_config(path: &str) -> AppConfig {
     }
 
     cfo.proxy_tls_grade = parce_tls_grades(cfo.proxy_tls_grade.clone());
+    if let Some(member) = cfo.cluster_member.clone() {
+        NODE_STATUS.set(Some(Arc::from(member.as_str()))).expect("Config can only be initialized once");
+        let mk = cfo.master_key.clone().expect("In CLuster mode master key must be set");
+        MASTER_KEY.set(Some(Arc::from(mk.as_str()))).expect("Config can only be initialized once");
+        if member == "master" {
+            IM_A_MASTER.set(true).expect("Master status can only be initialized once");
+            let worker_str = cfo.worker_nodes.as_deref().expect("At least one worker must be set in master mode");
+            let mut wks = Vec::new();
+            for worker in worker_str.split(',') {
+                let clean = worker.replace("//", "");
+                let wk_vec = clean.trim().split(':').collect::<Vec<&str>>();
+                let wk = WorkerNodes {
+                    address: Arc::from(wk_vec[1]),
+                    port: wk_vec[2].parse::<u16>().expect("Invalid worker port number"),
+                    tls: match wk_vec[0] {
+                        "https" => true,
+                        "http" => false,
+                        _ => panic!("invalid protocol string"),
+                    },
+                    key: cfo.master_key.clone().map(|s| Arc::from(s)),
+                };
+                wks.push(wk);
+                // println!("{:?}", wk);
+                // if let Some((host, port_str)) = worker.trim().split_once(':') {
+                //     let port = port_str.parse::<u16>().expect("Invalid worker port number");
+                //     wks.push((host.to_string(), port));
+                // } else {
+                //     panic!("Invalid worker format, expected host:port");
+                // }
+            }
+            WORKERS.set(wks).expect("Workers list can only be initialized once");
+        }
+    }
     cfo
 }
 

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,21 +1,22 @@
 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)>>;
 
 pub type UpstreamsIdMap = DashMap<String, Arc<InnerMap>>;
 pub type Headers = DashMap<Arc<str>, DashMap<Arc<str>, Vec<(String, Arc<str>)>>>;
 
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug, Default, Serialize)]
 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)]
@@ -91,6 +92,8 @@ pub struct PathConfig {
     pub healthcheck: Option<bool>,
     pub redirect_to: Option<String>,
     pub authorization: Option<Auth>,
+    pub is_ssl: Option<bool>,
+    pub is_http2: Option<bool>,
 }
 #[derive(Debug, Default)]
 pub struct Configuration {
@@ -133,12 +136,15 @@ pub struct AppConfig {
     pub log_file: Option<String>,
     pub log_pattern: Option<String>,
     pub log_structured: Option<String>,
+    pub acme_dns_provider: Option<String>,
     pub tcp_keepalive_idle: Option<u64>,
     pub tcp_keepalive_interval: Option<u64>,
     pub tcp_keepalive_count: Option<usize>,
+    pub cluster_member: Option<String>,
+    pub worker_nodes: Option<String>,
 }
 
-#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
+#[derive(Debug, Default, Clone, PartialEq, Eq, Hash, Serialize)]
 pub struct InnerAuth {
     pub auth_type: Arc<str>,
     pub auth_cred: Arc<str>,
@@ -151,13 +157,20 @@ 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>>,
 }
 
+#[allow(dead_code)]
+impl Default for InnerMap {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[allow(dead_code)]
 impl InnerMap {
     pub fn new() -> Self {
@@ -183,12 +196,20 @@ 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 {
     pub backends: Vec<InnerMapForJson>,
     pub requests: usize,
 }
+
+#[derive(Debug)]
+pub struct WorkerNodes {
+    pub address: Arc<str>,
+    pub key: Option<Arc<str>>,
+    pub port: u16,
+    pub tls: bool,
+}

src/utils/watch.rs

@@ -38,13 +38,11 @@ pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
     while let Some(event) = local_rx.recv().await {
         match event {
             Ok(e) => match e.kind {
-                EventKind::Modify(ModifyKind::Data(_)) | EventKind::Create(..) | EventKind::Remove(..) => {
-                    if start.elapsed() > Duration::from_secs(2) {
-                        start = Instant::now();
-                        let snd = load_configuration(file_path, "filepath").await.0;
-                        if let Some(snd) = snd {
-                            toreturn.send(snd).await.unwrap();
-                        }
+                EventKind::Modify(ModifyKind::Data(_)) | EventKind::Create(..) | EventKind::Remove(..) if start.elapsed() > Duration::from_secs(2) => {
+                    start = Instant::now();
+                    let snd = load_configuration(file_path, "filepath").await.0;
+                    if let Some(snd) = snd {
+                        toreturn.send(snd).await.unwrap();
                     }
                 }
                 _ => (),
@@ -64,13 +62,11 @@ pub fn folder_watch(path: String, sender: std::sync::mpsc::Sender<Vec<Certificat
     loop {
         match rx.recv_timeout(Duration::from_secs(1)) {
             Ok(Ok(event)) => match &event.kind {
-                EventKind::Modify(ModifyKind::Data(_)) | EventKind::Create(_) | EventKind::Remove(_) => {
-                    if start.elapsed() > Duration::from_secs(1) {
-                        start = Instant::now();
-                        let certificate_configs = listdir(path.clone());
-                        sender.send(certificate_configs)?;
-                        info!("Certificate changed: {:?}, {:?}", event.kind, event.paths);
-                    }
+                EventKind::Modify(ModifyKind::Data(_)) | EventKind::Create(_) | EventKind::Remove(_) if start.elapsed() > Duration::from_secs(1) => {
+                    start = Instant::now();
+                    let certificate_configs = listdir(path.clone());
+                    sender.send(certificate_configs)?;
+                    info!("Certificate changed: {:?}, {:?}", event.kind, event.paths);
                 }
                 _ => {}
             },