Cargo.toml

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "aralez"
-version = "0.94.2"
+version = "0.95.2"
 edition = "2024"
 license = "Apache-2.0"
 description = "Reverse proxy built on top of Cloudflare's Pingora"
@@ -27,7 +27,8 @@ 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"
@@ -69,5 +70,6 @@ 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"] }
+redis = { version = "1.7.1", features = ["tokio-comp", "connection-manager"] }

README.md

@@ -60,30 +60,41 @@ Built on Rust, on top of **Cloudflare’s Pingora engine**, **Aralez** delivers
 
 ### `main.yaml`
 
-| Key                              | Example Value                    | Description                                                                                                                                                           |
+| Key                              | Example Value                             | Description                                                                                                                                                                                                            |
-|----------------------------------|----------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+|----------------------------------|-------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| **threads**                      | 12                               | Number of running daemon threads. Optional, defaults to 1                                                                                                             |
+| **threads**                      | `12`                                      | Number of running daemon threads. Optional, defaults to 1                                                                                                                                                              |
-| **runuser**                      | aralez                           | Optional. Username for running aralez after dropping root privileges (requires launch as root)                                                                        |
+| **runuser**                      | `aralez`                                  | Optional. Username for running aralez after dropping root privileges (requires launch as root)                                                                                                                         |
-| **rungroup**                     | aralez                           | Optional. Group for running aralez after dropping root privileges (requires launch as root)                                                                           |
+| **rungroup**                     | `aralez`                                  | Optional. Group for running aralez after dropping root privileges (requires launch as root)                                                                                                                            |
-| **daemon**                       | false                            | Run in background (boolean)                                                                                                                                           |
+| **daemon**                       | `false`                                   | Run in background (boolean)                                                                                                                                                                                            |
-| **upstream_keepalive_pool_size** | 500                              | Pool size for upstream keepalive connections                                                                                                                          |
+| **upstream_keepalive_pool_size** | `500`                                     | Pool size for upstream keepalive connections                                                                                                                                                                           |
-| **pid_file**                     | /tmp/aralez.pid                  | Path to PID file                                                                                                                                                      |
+| **pid_file**                     | `/tmp/aralez.pid`                         | Path to PID file                                                                                                                                                                                                       |
-| **config_address**               | 0.0.0.0:3000                     | HTTP API address for pushing upstreams.yaml from remote location                                                                                                      |
+| **upgrade_sock**                 | `/tmp/aralez.sock`                        | Path to socket file                                                                                                                                                                                                    |
-| **proxy_tls_grade**              | high, medium, unsafe             | Grade of TLS ciphers. `high` matches Qualys SSL Labs A+ (defaults to `medium`)                                                                                        |
+| **config_address**               | `0.0.0.0:3000`                            | HTTP API address for pushing upstreams.yaml from remote location                                                                                                                                                       |
-| **proxy_address_http**           | 0.0.0.0:6193                     | Aralez HTTP bind address                                                                                                                                              |
+| **proxy_tls_grade**              | `high, medium, unsafe`                    | Grade of TLS ciphers. `high` matches Qualys SSL Labs A+ (defaults to `medium`)                                                                                                                                         |
-| **proxy_address_tls**            | 0.0.0.0:6194                     | Aralez HTTPS bind address (Optional)                                                                                                                                  |
+| **proxy_address_http**           | `0.0.0.0:6193`                            | Aralez HTTP bind address                                                                                                                                                                                               |
-| **proxy_configs**                | /etc/aralez/                     | Direcotry containing configuration files, must be writeable by user  `aralez`                                                                                         |
+| **proxy_address_tls**            | `0.0.0.0:6194`                            | Aralez HTTPS bind address (Optional)                                                                                                                                                                                   |
-| **upstreams_conf**               | /etc/aralez/upstreams.yaml       | Location of the upstreams file                                                                                                                                        |
+| **proxy_configs**                | `/etc/aralez/`                            | Direcotry containing configuration files, must be writeable by user  `aralez`                                                                                                                                          |
-| **access_log**                   | access                           | Configure access logging. Values: `access, error`                                                                                                                     |
+| **upstreams_conf**               | `/etc/aralez/upstreams.yaml`              | Location of the upstreams file                                                                                                                                                                                         |
-| **log_level**                    | info                             | Log level: `info`, `warn`, `error`, `debug`, `trace`, `off`                                                                                                           |
+| **access_log**                   | `access`                                  | Configure access logging. Values: `access, error`                                                                                                                                                                      |
-| **log_file**                     | /tmp/aralez.log, 20, 5, compress | Optional, the location of log file, file size in megabytes, number of log files to keep, compress files .valueas are separated by comma. Defaults to None, 100, 5, No |
+| **log_level**                    | `info`                                    | Log level: `info`, `warn`, `error`, `debug`, `trace`, `off`                                                                                                                                                            |
-| **hc_method**                    | HEAD                             | Healthcheck method: HEAD, GET, POST (UPPERCASE)                                                                                                                       |
+| **log_file**                     | `/tmp/aralez.log, 20, 5, compress`        | Optional, the location of log file, file size in megabytes, number of log files to keep, compress files .valueas are separated by comma. Defaults to None, 100, 5, No                                                  |
-| **hc_interval**                  | 2                                | Interval for health checks in seconds                                                                                                                                 |
+| **log_pattern**                  | `{d(%Y-%m-%d %H:%M:%S)} {l} - {m}{n}`     | Optional log pattern, defaults to `{d(%Y-%m-%d %H:%M:%S)} {l} - {m}{n}`. See log4rs official [documentation](https://docs.rs/log4rs/latest/log4rs/encode/pattern/struct.PatternEncoderDeserializer.html#configuration) |
-| **file_server_folder**           | /some/local/folder               | Optional. Local folder to serve                                                                                                                                       |
+| **hc_method**                    | `HEAD`                                    | Healthcheck method: HEAD, GET, POST (UPPERCASE)                                                                                                                                                                        |
-| **file_server_address**          | 127.0.0.1:3002                   | Optional. Local address for file server                                                                                                                               |
+| **hc_interval**                  | `2`                                       | Interval for health checks in seconds                                                                                                                                                                                  |
-| **config_api_enabled**           | true                             | Enable/disable remote config push capability                                                                                                                          |
+| **file_server_folder**           | `/some/local/folder`                      | Optional. Local folder to serve                                                                                                                                                                                        |
-| **cache_size_mb**                | 50                               | Optional. Enable internal cache with size in megabytes of value. Cache is entirely disabled when this line is commented out.                                          |
+| **file_server_address**          | `127.0.0.1:3002`                          | Optional. Local address for file server                                                                                                                                                                                |
-| **cache_ttl**                    | 10                               | Optional. Set Default TTL in seconds for cached items, works if cache control headers are not set via upstream. Defaults to 60 if cache_size_mb is set.               |
+| **config_api_enabled**           | `true`                                    | Enable/disable remote config push capability                                                                                                                                                                           |
+| **cache_size_mb**                | `50`                                      | Optional. Enable internal cache with size in megabytes of value. Cache is entirely disabled when this line is commented out.                                                                                           |
+| **cache_enabled**                | `bool`                                    | Optional. Disabled by default                                                                                                                                                                                          |
+| **cache_storage**                | `memory, redis://host:port`               | Optional Defaults to memory works with Redis, Valkey, Dragonfly (requires redis:// prefix)                                                                                                                             |
+| **cache_ttl**                    | `60`                                      | Optional. Set internal cache with size in megabytes of value. Defaults to 50Mb if cache_enabled: true                                                                                                                  |
+| **tcp_keepalive_idle**           | `60`                                      | Seconds of inactivity before the kernel starts sending keepalive probes to a downstream client. Defaults to 60                                                                                                         |
+| **tcp_keepalive_interval**       | `10`                                      | Seconds between individual keepalive probes if the client does not respond. Defaults to 10                                                                                                                             |
+| **tcp_keepalive_count**          | `5`                                       | Number of unanswered probes before the kernel declares the connection dead and closes it. Defaults to 5                                                                                                                |
+| **cluster_member**               | `master, worker`                          | Optional. Enables cluster mode, Sets node type to master or worker. Disabled by default                                                                                                                                |
+| **worker_nodes**                 | `http(s)://host:port,http(s)://host:port` | Required is if cluster_member: master. Comma separated list of worker nodes http(s)://host:port                                                                                                                        |
+| **log_structured**               | `elasticsearch`                           | Optional. Enables structured logging and sets backend type. Only elasticsearch et the moment.                                                                                                                          |
+| **acme_dns_provider**            | `cloudflare, route53, digitalocean, etc`  | Optional. Enable ACME DNS-01 challenge, sets DNS provider.  ONly coudflare, route53, digitalocean at the moment.                                                                                                       |
 
 ---
 
@@ -297,10 +308,27 @@ openssl req -x509 -newkey rsa:4096 \
 
 ## Remote Config API
 
-Push new `upstreams.yaml` over HTTP to `config_address` (`:3000` by default). Useful for CI/CD automation or remote config updates. URL parameter. `key=MASTERKEY` is required. `MASTERKEY` is the value of `master_key` in the `main.yaml`
+Push new `upstreams.yaml` over HTTP to `config_address` (`127.0.0.1:3000` by default).
+Useful for CI/CD automation or remote config updates.
 
 ```bash
-curl -XPOST --data-binary @./etc/upstreams.txt 127.0.0.1:3000/conf?key=${MASTERKEY}
+curl -XPOST --data-binary @./etc/upstreams.txt 127.0.0.1:3000/conf
+```
+
+Don't run API on address 0.0.0.0 on production servers, keep it on 127.0.0.1 and configure appropriate upstream with supported authentication and TLS termination
+
+Example:
+
+```yaml
+  config.example.com:
+    paths:
+      "/conf":
+        authorization:
+          type: "basic"
+          data: "user:pass"
+        healthcheck: false
+        servers:
+          - "127.0.0.1:3000"
 ```
 
 ---

etc/main.yaml

@@ -4,25 +4,31 @@ runuser: aralez # Username for running aralez after dropping root privileges, re
 rungroup: aralez # Group for running aralez after dropping root privileges, requires program to start as root
 upstream_keepalive_pool_size: 500 # Pool size for upstream keepalive connections
 pid_file: /tmp/aralez.pid # Path to PID file
+#error_log: /tmp/aralez_err.log # Path to error log
 upgrade_sock: /tmp/aralez.sock # Path to socket file
+cache_enabled: true #  Optional. Disabled by default
+cache_storage: memory # memory | redis://host:port . Optional Defaults to memory works with Redis, Valkey, Dragonfly (requires redis:// prefix)
+cache_size_mb: 50 # Optional. Set internal cache with size in megabytes of value. Defaults to 50Mb if cache_enabled: true
+cache_ttl: 60 # Optional. Set Default TTL in seconds for cached items, works if cache control headers are not set via upstream. Defaults to 60 if cache_enabled: true
 config_api_enabled: true # Boolean to enable/disable remote config push capability.
 config_address: 127.0.0.1:3000 # HTTP API address for pushing upstreams.yaml from remote location
 proxy_address_http: 0.0.0.0:80 # Proxy HTTP bind address
 proxy_address_tls: 0.0.0.0:443 # Optional, Proxy TLS bind address
-proxy_configs: /opt/aralez/asyncweb/etc # Mandatory if proxy_address_tls set, should contain a certificate and key files strictly in a format {NAME}.crt, {NAME}.key.
+proxy_configs: /opt/aralez/etc # Mandatory if proxy_address_tls set, should contain a certificate and key files strictly in a format {NAME}.crt, {NAME}.key.
 proxy_tls_grade: high # Grade of TLS suite for proxy (high, medium, unsafe), matching grades of Qualys SSL Labs
 upstreams_conf: /opt/aralez/etc/upstreams.yaml # the location of upstreams file
 file_server_folder: /opt/aralez/public # Optional, local folder to serve
-file_server_address: 0.0.0.0:3002 # Optional, Local address for file server. Can set as upstream for public access.
+file_server_address: 127.0.0.1:3002 # Optional, Local address for file server. Can set as upstream for public access.
 log_level: info # info, warn, error, debug, trace, off
 log_file: /tmp/aralez.log, 20, 5, compress # Optional, the location of log file, file size in megabytes, number of log files to keep, compress files .valueas are separated by comma.
 log_pattern: "{d(%Y-%m-%d %H:%M:%S)} {h({l})} {t} - {m} {n}" # Optional log pattern, defaults to "{d(%Y-%m-%d %H:%M:%S)} {l} - {m}{n}"
-access_log: error # all, error, (Off if commented)
+access_log: error # all, error, (Off by default)
 hc_method: HEAD # Healthcheck method (HEAD, GET, POST are supported) UPPERCASE
 hc_interval: 2 #Interval for health checks in seconds
 tcp_keepalive_idle: 60 # Seconds of inactivity before the kernel starts sending keepalive probes to a downstream client
 tcp_keepalive_interval: 10 # Seconds between individual keepalive probes if the client does not respond
 tcp_keepalive_count: 5 # Number of unanswered probes before the kernel declares the connection dead and closes it
-cache_size_mb: 50 # Optional. Enable internal cache with size in megabytes of value. Cache is entirely disabled when this line is commented out
+#cluster_member: master | worker  Optional. Enables cluster mode, Sets node type to master or worker. Disabled by default
-cache_ttl: 120 # Optional. Set Default TTL in seconds for cached items, works if cache control headers are not set via upstream. Defaults to 60 if cache_size_mb is set
+#worker_nodes: http://127.0.0.1:8000, http://127.0.0.1:4000 Required is if cluster_member: master. Comma separated list of worker nodes http(s)://host:port
-
+#log_structured: elasticsearch Optional. Enables structured logging and sets backend type. Only elasticsearch et the moment.
+#acme_dns_provider: cloudflare Optional. Enable ACME DNS-01 challenge, sets DNS provider.  ONly coudflare, route53, digitalocean at the moment.

etc/upstreams.yaml

@@ -130,7 +130,7 @@ upstreams:
       "/":
         healthcheck: false
         servers:
-          - "127.0.0.1:3000"
+          - "127.0.0.1:8000"
       "/.well-known/acme-challenge":
         healthcheck: false
         servers:

src/cache/mod.rs

@@ -0,0 +1,2 @@
+pub mod redis;
+pub mod types;

src/cache/redis.rs

@@ -0,0 +1,222 @@
+use async_trait::async_trait;
+use bytes::Bytes;
+use log::warn;
+use pingora::Result;
+use pingora_cache::storage::{HandleHit, HandleMiss, HitHandler, MissFinishType, MissHandler, Storage};
+use pingora_cache::trace::SpanHandle;
+use pingora_cache::{CacheKey, CacheMeta, PurgeOutcome, PurgeTarget, PurgeType};
+use redis::aio::ConnectionManager;
+use redis::{AsyncCommands, Client, RedisResult};
+use std::any::Any;
+use std::borrow::Cow;
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub struct RedisCache {
+    pub client: Arc<ConnectionManager>,
+    pub ttl: u64,
+}
+
+pub struct RedisHitHandler {
+    body: Option<Bytes>,
+}
+
+pub struct RedisMissHandler {
+    client: Arc<ConnectionManager>,
+    body_key: String,
+    body_buffer: Vec<u8>,
+    ttl: u64,
+}
+
+impl RedisCache {
+    pub async fn new(url: &str, ttl: u64) -> RedisResult<Self> {
+        let client = Client::open(url)?;
+        let manager = ConnectionManager::new(client).await?;
+        Ok(Self { client: Arc::new(manager), ttl })
+    }
+    fn make_key(key: &CacheKey) -> Cow<'_, str> {
+        match key.primary_key_str() {
+            Some(s) => Cow::Borrowed(s),
+            None => Cow::Owned(key.primary_key().iter().map(|b| format!("{:02x}", b)).collect()),
+        }
+    }
+}
+
+#[async_trait]
+impl HandleHit for RedisHitHandler {
+    async fn read_body(&mut self) -> Result<Option<Bytes>> {
+        Ok(self.body.take())
+    }
+
+    async fn finish(self: Box<Self>, _storage: &'static (dyn Storage + Sync), _key: &CacheKey, _trace: &SpanHandle) -> Result<()> {
+        Ok(())
+    }
+
+    fn can_seek(&self) -> bool {
+        false
+    }
+
+    fn seek(&mut self, _start: usize, _end: Option<usize>) -> Result<()> {
+        Ok(())
+    }
+
+    fn should_count_access(&self) -> bool {
+        true
+    }
+
+    fn get_eviction_weight(&self) -> usize {
+        self.body.as_ref().map(|b| b.len()).unwrap_or(0)
+    }
+
+    fn entry_id(&self) -> Option<pingora_cache::eviction::CacheEntryId> {
+        None
+    }
+
+    fn as_any(&self) -> &(dyn Any + Send + Sync) {
+        self
+    }
+
+    fn as_any_mut(&mut self) -> &mut (dyn Any + Send + Sync) {
+        self
+    }
+}
+
+#[async_trait]
+impl HandleMiss for RedisMissHandler {
+    async fn write_body(&mut self, data: Bytes, _eof: bool) -> Result<()> {
+        self.body_buffer.extend_from_slice(&data);
+        Ok(())
+    }
+
+    async fn finish(self: Box<Self>) -> Result<MissFinishType> {
+        let len = self.body_buffer.len();
+        let mut conn = (*self.client).clone();
+        if let Err(e) = conn.set_ex::<_, _, ()>(&self.body_key, &self.body_buffer, self.ttl).await {
+            warn!("Failed to write cache entry to Redis: {e}");
+        }
+        Ok(MissFinishType::Appended(len, None))
+    }
+
+    fn entry_id(&self) -> Option<pingora_cache::eviction::CacheEntryId> {
+        None
+    }
+}
+
+#[async_trait]
+impl Storage for RedisCache {
+    async fn lookup(&'static self, key: &CacheKey, _trace: &SpanHandle) -> Result<Option<(CacheMeta, HitHandler)>> {
+        let base = Self::make_key(key);
+
+        // Pre-allocate exact capacities to prevent reallocations
+        let mut meta_key = String::with_capacity(base.len() + 5);
+        meta_key.push_str(&base);
+        meta_key.push_str(":meta");
+
+        let mut body_key = String::with_capacity(base.len() + 5);
+        body_key.push_str(&base);
+        body_key.push_str(":body");
+
+        let mut internal_key = String::with_capacity(meta_key.len() + 9);
+        internal_key.push_str(&meta_key);
+        internal_key.push_str(":internal");
+
+        let mut header_key = String::with_capacity(meta_key.len() + 7);
+        header_key.push_str(&meta_key);
+        header_key.push_str(":header");
+
+        let mut conn = (*self.client).clone();
+
+        let internal: Option<Vec<u8>> = conn.get(&internal_key).await.ok().flatten();
+        let header: Option<Vec<u8>> = conn.get(&header_key).await.ok().flatten();
+        let body: Option<Vec<u8>> = conn.get(&body_key).await.ok().flatten();
+
+        match (internal, header, body) {
+            (Some(internal_bytes), Some(header_bytes), Some(body_bytes)) => match CacheMeta::deserialize(&internal_bytes, &header_bytes) {
+                Ok(meta) => {
+                    let handler: HitHandler = Box::new(RedisHitHandler {
+                        body: Some(Bytes::from(body_bytes)),
+                    });
+                    Ok(Some((meta, handler)))
+                }
+                Err(_) => {
+                    let _: RedisResult<()> = conn.del(&[&internal_key, &header_key, &body_key]).await;
+                    Ok(None)
+                }
+            },
+            _ => Ok(None),
+        }
+    }
+
+    async fn get_miss_handler(&'static self, key: &CacheKey, meta: &CacheMeta, _trace: &SpanHandle) -> Result<MissHandler> {
+        let base = Self::make_key(key);
+
+        let mut meta_key = String::with_capacity(base.len() + 5);
+        meta_key.push_str(&base);
+        meta_key.push_str(":meta");
+
+        let mut body_key = String::with_capacity(base.len() + 5);
+        body_key.push_str(&base);
+        body_key.push_str(":body");
+
+        let mut internal_key = String::with_capacity(meta_key.len() + 9);
+        internal_key.push_str(&meta_key);
+        internal_key.push_str(":internal");
+
+        let mut header_key = String::with_capacity(meta_key.len() + 7);
+        header_key.push_str(&meta_key);
+        header_key.push_str(":header");
+
+        let ttl = self.ttl;
+        let (internal_bytes, header_bytes) = meta.serialize()?;
+        let mut conn = (*self.client).clone();
+
+        let _: RedisResult<()> = conn.set_ex(&internal_key, &internal_bytes, ttl).await;
+        let _: RedisResult<()> = conn.set_ex(&header_key, &header_bytes, ttl).await;
+
+        Ok(Box::new(RedisMissHandler {
+            client: self.client.clone(),
+            body_key,
+            body_buffer: Vec::with_capacity(4096),
+            ttl,
+        }))
+    }
+
+    async fn purge(&'static self, _target: PurgeTarget<'_>, _purge_type: PurgeType, _trace: &SpanHandle) -> Result<PurgeOutcome> {
+        Ok(PurgeOutcome::NotFound)
+    }
+
+    async fn update_meta(&'static self, key: &CacheKey, meta: &CacheMeta, _trace: &SpanHandle) -> Result<bool> {
+        let base = Self::make_key(key);
+
+        let mut meta_key = String::with_capacity(base.len() + 5);
+        meta_key.push_str(&base);
+        meta_key.push_str(":meta");
+
+        let mut internal_key = String::with_capacity(meta_key.len() + 9);
+        internal_key.push_str(&meta_key);
+        internal_key.push_str(":internal");
+
+        let mut header_key = String::with_capacity(meta_key.len() + 7);
+        header_key.push_str(&meta_key);
+        header_key.push_str(":header");
+
+        let ttl = self.ttl;
+        match meta.serialize() {
+            Ok((internal_bytes, header_bytes)) => {
+                let mut conn = (*self.client).clone();
+                let _: RedisResult<()> = conn.set_ex(&internal_key, &internal_bytes, ttl).await;
+                let _: RedisResult<()> = conn.set_ex(&header_key, &header_bytes, ttl).await;
+                Ok(true)
+            }
+            Err(_) => Ok(false),
+        }
+    }
+
+    fn support_streaming_partial_write(&self) -> bool {
+        false
+    }
+
+    fn as_any(&self) -> &(dyn Any + Send + Sync) {
+        self
+    }
+}

src/cache/types.rs

@@ -0,0 +1,41 @@
+use crate::cache::redis::RedisCache;
+use crate::utils::lazylock::{CACHE_TTL, EVICTION};
+use crate::utils::types::AppConfig;
+use log::{info, warn};
+use pingora_cache::eviction::simple_lru::Manager;
+use pingora_cache::{MemCache, Storage};
+use std::sync::{Arc, OnceLock};
+
+pub static GLOBAL_STORAGE: OnceLock<Box<dyn Storage + Send + Sync>> = OnceLock::new();
+
+pub async fn init_cache(config: Arc<AppConfig>) {
+    let cache_storage = config.cache_storage.clone().unwrap_or_else(|| {
+        info!("No cache_storage configured, defaulting to memory");
+        "memory".to_string()
+    });
+    let cache_type = cache_storage.splitn(2, ':').next().unwrap_or("memory");
+    let ttl = config.cache_ttl.unwrap_or(60);
+
+    match cache_type {
+        "memory" => {
+            CACHE_TTL.set(ttl).ok();
+            if let Some(size) = config.cache_size_mb {
+                let cache_size = size * 1024 * 1024;
+                drop(EVICTION.set(Manager::new(cache_size)));
+            }
+            GLOBAL_STORAGE.set(Box::new(MemCache::new())).ok();
+            info!("Cache initialized successfully. Storage: {}", cache_storage);
+        }
+        "redis" => match RedisCache::new(&cache_storage, ttl).await {
+            Ok(cache) => {
+                CACHE_TTL.set(ttl).ok();
+                GLOBAL_STORAGE.set(Box::new(cache)).ok();
+                info!("Cache initialized successfully. Storage: {}", cache_storage);
+            }
+            Err(e) => log::error!("Cache storage initialization failed: {}", e),
+        },
+        _ => {
+            warn!("Unsupported storage, disabling caching: {}", cache_storage);
+        }
+    }
+}

src/core/bgservice.rs

@@ -1,3 +1,4 @@
+use crate::cache::types::init_cache;
 use crate::core::proxyhttp::LB;
 use crate::ingress::{APIUpstreamProvider, ConsulProvider, Discovery, FromFileProvider, KubernetesProvider};
 use crate::logging::core::{init_access_log, init_access_logging, log_builder};
@@ -29,16 +30,15 @@ impl BackgroundService for LB {
         let config = load_configuration(self.config.upstreams_conf.clone().as_str(), "filepath")
             .await
             .0
-            .expect("Failed to load configuration");
+            .die("Failed to load configuration");
 
         match config.typecfg.as_str() {
             "file" => {
                 info!("Running File discovery, requested type is: {}", config.typecfg);
-                tx.send(config).await.unwrap();
+                tx.send(config).await.die("Running File discovery");
                 let file_load = FromFileProvider {
                     path: self.config.upstreams_conf.clone(),
                 };
-                // let _ = tokio::spawn(async move { file_load.start(tx).await });
                 drop(tokio::spawn(async move { file_load.start(tx).await }));
             }
             "kubernetes" => {
@@ -79,11 +79,12 @@ impl BackgroundService for LB {
         let ff = self.ump_full.clone();
         let im = self.ump_byid.clone();
         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 { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().die("Unable to spawn HC worker"))).await });
         tokio::spawn(async move { calc_cache_metrics().await });
         let acme_provider = self.config.acme_dns_provider.clone();
         tokio::spawn(async move { refresh_order(certdir, confdir, acme_provider).await });
 
+        init_cache(self.config.clone()).await;
         init_access_logging(self.config.access_log.clone()).await;
         loop {
             tokio::select! {
@@ -91,7 +92,7 @@ impl BackgroundService for LB {
                     break;
                 }
                 val = rx.recv() => {
-                    let start = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
+                    let start = SystemTime::now().duration_since(UNIX_EPOCH).die(format!("ON {} {}", file!(), line!()).as_str());
                     if let Some(ss) = val {
                         clone_dashmap_into(&ss.upstreams, &self.ump_full);
                         clone_dashmap_into(&ss.upstreams, &self.ump_upst);
@@ -138,7 +139,7 @@ impl BackgroundService for LB {
                             }
                         }
                     }
-                    let end = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
+                    let end = SystemTime::now().duration_since(UNIX_EPOCH).die(format!("ON {} {}", file!(), line!()).as_str());
                     let startup = end - start;
                     info!("Loading upstreams took: {}ms", startup.as_millis());
                 }

src/core/proxyhttp.rs

@@ -1,7 +1,8 @@
 use crate::auth::authenticate;
+use crate::cache::types::GLOBAL_STORAGE;
 use crate::core::gethosts::{GetHost, GetHostsReturHeaders};
 use crate::logging::core::access_log;
-use crate::utils::lazylock::{CACHE_LOCK, CACHE_TTL, EVICTION, LOCALHOST, MEM_CACHE, RATE_LIMITER, REQUESTS_4XX, REVERSE_STORE};
+use crate::utils::lazylock::{CACHE_LOCK, CACHE_TTL, EVICTION, LOCALHOST, RATE_LIMITER, REQUESTS_4XX, REVERSE_STORE};
 use crate::utils::metrics::*;
 use crate::utils::types::{AppConfig, Extraparams, Headers, InnerMap, UpstreamsDashMap, UpstreamsIdMap};
 use arc_swap::ArcSwap;
@@ -34,6 +35,7 @@ pub struct LB {
     pub config: Arc<AppConfig>,
     pub extraparams: Arc<ArcSwap<Extraparams>>,
     pub cache_enabled: bool,
+    pub master_server: bool,
 }
 
 pub struct Context {
@@ -66,8 +68,7 @@ impl ProxyHttp for LB {
 
     fn cache_key_callback(&self, session: &Session, ctx: &mut Self::CTX) -> Result<CacheKey> {
         let host = ctx.hostname.as_deref().unwrap_or("default");
-        let path = session.req_header().uri.path();
+        let path = session.req_header().uri.path_and_query().map_or("/", |pq| pq.as_str());
-
         let primary_key = KEY_BUF.with(|buf| {
             let mut b = buf.borrow_mut();
             b.clear();
@@ -75,15 +76,14 @@ impl ProxyHttp for LB {
             b.push_str(path);
             b.clone()
         });
-
+        Ok(CacheKey::new(primary_key, host /* host is just for debugging */))
-        let secondary_key = session.req_header().uri.path_and_query().map(|pq| pq.as_str()).unwrap_or("/");
-        Ok(CacheKey::new(primary_key, secondary_key))
     }
-
     fn request_cache_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<()> {
-        if self.cache_enabled && session.req_header().method == Method::GET {
+        let has_auth = _ctx.upstream_peer.as_ref().map(|m| m.authorization.is_some()).unwrap_or(false) || _ctx.extraparams.authentication.is_some();
-            if let Some(eviction) = EVICTION.get() {
+        if self.cache_enabled && session.req_header().method == Method::GET && !has_auth {
-                session.cache.enable(&*MEM_CACHE, Some(eviction), None, Some(&*CACHE_LOCK), None);
+            if let Some(storage) = GLOBAL_STORAGE.get() {
+                let eviction_mgr = EVICTION.get().map(|e| e as &(dyn pingora_cache::eviction::EvictionManager + Sync));
+                session.cache.enable(storage.as_ref(), eviction_mgr, None, Some(&*CACHE_LOCK), None);
             }
         }
         Ok(())
@@ -105,6 +105,10 @@ 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;

src/core/start.rs

@@ -29,19 +29,24 @@ use std::time::Duration;
 use std::{fs, thread};
 
 pub fn run() {
-    default_provider().install_default().expect("Failed to install RusTLS crypto provider");
+    default_provider().install_default().die("Failed to install RusTLS crypto provider:");
     let parameters = Opt::parse_args();
-    let file = parameters.conf.clone().unwrap();
+    let file = parameters.conf.clone().die("Failed to clone the config:");
     let maincfg = crate::utils::parceyaml::parce_main_config(file.as_str());
 
-    let cache_enabled = maincfg.cache_size_mb.is_some();
+    let master_server = maincfg.cluster_member.as_deref() == Some("master");
-    CACHE_TTL.set(maincfg.cache_ttl.unwrap_or(60)).unwrap();
-    if let Some(size) = maincfg.cache_size_mb {
-        let cache_size = size * 1024 * 1024;
-        drop(EVICTION.set(Manager::new(cache_size)));
-    }
 
-    let mut server = Server::new(parameters).unwrap();
+    let cache_enabled = maincfg.cache_enabled.unwrap_or(false);
+
+    // ========================================================================== //
+    CACHE_TTL.set(maincfg.cache_ttl.unwrap_or(60)).die("Failed to set cache TTL");
+    if cache_enabled {
+        let cache_size = maincfg.cache_size_mb.unwrap_or(50);
+        let _ = EVICTION.set(Manager::new(cache_size * 1024 * 1024));
+    }
+    // ========================================================================== //
+
+    let mut server = Server::new(parameters).die("Failed to start the server:");
     server.bootstrap();
 
     let uf_config = Arc::new(DashMap::new());
@@ -70,6 +75,7 @@ pub fn run() {
         server_headers: sh_config,
         extraparams: ec_config,
         cache_enabled,
+        master_server,
     };
 
     let grade = cfg.proxy_tls_grade.clone().unwrap_or("medium".to_string());
@@ -104,25 +110,25 @@ pub fn run() {
     if let Some(bind_address_tls) = bind_address_tls {
         check_priv(bind_address_tls.as_str());
         let (tx, rx): (Sender<Vec<CertificateConfig>>, Receiver<Vec<CertificateConfig>>) = channel();
-        let certs_path = cfg.proxy_configs.clone().unwrap() + "/certificates";
+        let certs_path = cfg.proxy_configs.clone().die("Failed certificates path") + "/certificates";
 
         if fs::metadata(certs_path.clone()).is_err() {
-            fs::create_dir_all(certs_path.clone()).unwrap();
+            fs::create_dir_all(certs_path.clone()).die("Failed to create directories");
         }
         thread::spawn(move || {
-            folder_watch(certs_path, tx).unwrap();
+            folder_watch(certs_path, tx).die("Failed to spawn certificates channel");
         });
-        let certificate_configs = rx.recv().unwrap();
+        let certificate_configs = rx.recv().die("Failed to spawn certificates channel");
         let first_set = load::Certificates::new(&certificate_configs, grade.as_str()).unwrap_or_else(|| panic!("Unable to load initial certificate info"));
         let certificates = Arc::new(ArcSwap::from_pointee(first_set));
         let certs_for_callback = certificates.clone();
 
         let certs_for_watcher = certificates.clone();
         let new_certs = load::Certificates::new(&certificate_configs, grade.as_str());
-        certs_for_watcher.store(Arc::new(new_certs.unwrap()));
+        certs_for_watcher.store(Arc::new(new_certs.die("Failed to build the watcher")));
 
         let mut tls_settings =
-            TlsSettings::intermediate(&certs_for_callback.load().default_cert_path, &certs_for_callback.load().default_key_path).expect("unable to load or parse cert/key");
+            TlsSettings::intermediate(&certs_for_callback.load().default_cert_path, &certs_for_callback.load().default_key_path).die("unable to load or parse cert/key");
 
         grades::set_tsl_grade(&mut tls_settings, grade.as_str());
         tls_settings.set_servername_callback(move |ssl_ref: &mut SslRef, ssl_alert: &mut SslAlert| certs_for_callback.load().server_name_callback(ssl_ref, ssl_alert));
@@ -160,7 +166,7 @@ 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();
+    let mut signals = Signals::new([SIGINT, SIGTERM, SIGQUIT]).die("Failed to create signals");
     if let Some(sig) = signals.forever().next() {
         match sig {
             SIGINT => info!("SIGINT received! Exiting..."),

src/core/webserver.rs

@@ -1,11 +1,14 @@
 use crate::auth::jwt::Claims;
 use crate::ingress::APIUpstreamProvider;
 use crate::tls::acme::http01::{acme_create, acme_order, http01_challenge};
+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::tools::{Die, upstreams_liveness_json, upstreams_to_json};
 use crate::utils::types::{Config, Configuration, UpstreamsDashMap};
 use axum::body::Body;
+use axum::extract::ConnectInfo;
 use axum::extract::{Query, State};
+use axum::http::HeaderMap;
 use axum::http::{Response, StatusCode};
 use axum::response::IntoResponse;
 use axum::routing::{any, get, post};
@@ -63,6 +66,7 @@ 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);
@@ -84,13 +88,16 @@ 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.into_make_service_with_connect_info::<SocketAddr>())
+            .await
+            .die("Can't start API server")
+    });
     let (tx, mut rx) = mpsc::channel(1);
     std::thread::spawn(move || {
-        let mut signals = Signals::new([SIGQUIT]).unwrap();
+        let mut signals = Signals::new([SIGQUIT]).die("Can't initialize OS SIGNALS");
         if let Some(sig) = signals.forever().next() {
-            tx.blocking_send(sig).unwrap();
+            tx.blocking_send(sig).die("Can;t send signal to main thread");
         }
     });
     rx.recv().await;
@@ -101,29 +108,79 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
     info!("Exiting...");
 }
 
-async fn conf(State(st): State<AppState>, Query(params): Query<HashMap<String, String>>, content: String) -> impl IntoResponse {
+async fn conf(
+    State(st): State<AppState>,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    Query(params): Query<HashMap<String, String>>,
+    headers: HeaderMap,
+    content: String,
+) -> impl IntoResponse {
     if !st.config_api_enabled {
-        return Response::builder().status(StatusCode::FORBIDDEN).body(Body::from("Config API is disabled !\n")).unwrap();
+        return (StatusCode::FORBIDDEN, "Config API is disabled !\n").into_response();
+    }
+
+    let api_key = headers.get("API-KEY").and_then(|v| v.to_str().ok());
+    let query_key = params.get("key").map(|k| k.as_str());
+    let authorized = match st.master_key.as_deref() {
+        Some(master_key) if !master_key.is_empty() => api_key == Some(master_key) || query_key == Some(master_key),
+        _ => false,
+    };
+    if !authorized {
+        warn!("Unauthorised access to /conf endpoint from {}", addr.ip());
+        return (StatusCode::UNAUTHORIZED, "Unauthorized: Invalid API-KEY\n").into_response();
     }
 
     let strcontent = content.as_str();
     let parsed = noyalib::from_str::<Config>(strcontent);
     match parsed {
         Ok(_) => {
-            if params.contains_key("save") {
+            let save = params.contains_key("save");
-                tokio::spawn(async move { apply_config(content.as_str(), st, true).await });
+            tokio::spawn(async move {
-            } else {
+                apply_config(&content, st, save).await;
-                tokio::spawn(async move { apply_config(content.as_str(), st, false).await });
+            });
-            }
+            (StatusCode::OK, "Accepted! Applying in background\n").into_response()
-            Response::builder().status(StatusCode::OK).body(Body::from("Accepted! Applying in background\n")).unwrap()
         }
         Err(err) => {
             error!("Failed to parse upstreams file: {}", err);
-            Response::builder().status(StatusCode::BAD_GATEWAY).body(Body::from(format!("Failed: {}\n", err))).unwrap()
+            (StatusCode::BAD_REQUEST, format!("Failed: {}\n", err)).into_response()
         }
     }
 }
 
+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 (StatusCode::UNAUTHORIZED, "Unauthorized: Invalid API-KEY\n").into_response();
+    }
+    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 (StatusCode::OK, "Accepted! Applying in background\n").into_response();
+                        }
+                        Err(err) => {
+                            error!("Failed to parse upstreams file: {}", err);
+                            return (StatusCode::BAD_GATEWAY, format!("Failed: {}\n", err)).into_response();
+                        }
+                    }
+                }
+                _ => {
+                    return (StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response();
+                }
+            }
+        }
+        return (StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response();
+    }
+    (StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response()
+}
+
 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 {
@@ -185,17 +242,13 @@ async fn metrics() -> impl IntoResponse {
     let encoder = TextEncoder::new();
     let mut buffer = Vec::new();
     if let Err(e) = encoder.encode(&metric_families, &mut buffer) {
-        // encoding error fallback
+        return (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to encode metrics: {}", e)).into_response();
-        return Response::builder()
-            .status(StatusCode::INTERNAL_SERVER_ERROR)
-            .body(Body::from(format!("Failed to encode metrics: {}", e)))
-            .unwrap();
     }
     Response::builder()
         .status(StatusCode::OK)
         .header("Content-Type", encoder.format_type())
         .body(Body::from(buffer))
-        .unwrap()
+        .unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response())
 }
 
 #[allow(clippy::needless_return)]
@@ -206,7 +259,7 @@ async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String,
             .status(StatusCode::OK)
             .header("Content-Type", "application/json")
             .body(Body::from(format!("{}", r)))
-            .unwrap();
+            .unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
     }
     if params.contains_key("all") {
         let resp = upstreams_to_json(&st.current_upstreams);
@@ -216,29 +269,17 @@ 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_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
-            }
-            Err(e) => {
-                return Response::builder()
-                    .status(StatusCode::INTERNAL_SERVER_ERROR)
-                    .body(Body::from(format!("Failed to get status: {}", e)))
-                    .unwrap();
             }
+            Err(e) => return (StatusCode::BAD_GATEWAY, format!("Internal Server Error {}\n", e)).into_response(),
         }
     }
-    Response::builder()
+    (StatusCode::INTERNAL_SERVER_ERROR, "Parameter mismatch\n").into_response()
-        .status(StatusCode::INTERNAL_SERVER_ERROR)
-        .body(Body::from("Parameter mismatch"))
-        .unwrap()
 }
 
 pub async fn port_is_available(name: &str, address: &str) -> TcpListener {
     let addr = SocketAddr::from_str(address).unwrap_or_else(|e| panic!("{}: Invalid address format: {:?}", name, e));
     let t = Duration::from_secs(2);
-
-    //if addr.ip() == IpAddr::V4(Ipv4Addr::UNSPECIFIED) {
-    //    addr.set_ip(IpAddr::V4(Ipv4Addr::LOCALHOST));
-    //}
     let p = addr.port();
     loop {
         match TcpListener::bind(addr).await {

src/ingress/consul.rs

@@ -1,7 +1,8 @@
 use crate::ingress::kuberconsul;
-use crate::ingress::kuberconsul::{list_to_upstreams, ServiceDiscovery};
+use crate::ingress::kuberconsul::{ServiceDiscovery, list_to_upstreams};
 use crate::utils::httpclient;
 use crate::utils::parceyaml::build_headers;
+use crate::utils::tools::Die;
 use crate::utils::types::{Configuration, GlobalServiceMapping, UpstreamsDashMap};
 use async_trait::async_trait;
 use dashmap::DashMap;
@@ -38,7 +39,7 @@ impl ServiceDiscovery for ConsulDiscovery {
                 });
                 let end = servers.len().saturating_sub(1);
                 let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 };
-                let consul_data = servers.get(num).unwrap().to_string();
+                let consul_data = servers.get(num).die("Reading consul data").to_string();
                 let ss = format!("{}/v1/catalog/service/", consul_data);
                 let dc = format!("{}/v1/catalog/services?filter=%22aralez.service=yes%22+in+ServiceTags", consul_data);
 

src/ingress/kuberconsul.rs

@@ -1,9 +1,9 @@
-use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams};
+use crate::utils::tools::{Die, clone_dashmap_into, compare_dashmaps, print_upstreams};
 use crate::utils::types::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap};
 use async_trait::async_trait;
 use dashmap::DashMap;
-use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
 use tokio::fs::File;
 use tokio::io::AsyncReadExt;
 use tokio::sync::mpsc::Sender;
@@ -31,9 +31,9 @@ pub fn match_path(conf: &GlobalServiceMapping, upstreams: &DashMap<Arc<str>, (Ve
 }
 
 pub async fn read_token(path: &str) -> String {
-    let mut file = File::open(path).await.unwrap();
+    let mut file = File::open(path).await.die("Reading kubernetes token");
     let mut contents = String::new();
-    file.read_to_string(&mut contents).await.unwrap();
+    file.read_to_string(&mut contents).await.die("Kubernetes token to string");
     contents.trim().to_string()
 }
 

src/ingress/kubernetes.rs

@@ -9,10 +9,11 @@ use std::env;
 use std::sync::Arc;
 
 use crate::ingress::kuberconsul;
-use crate::ingress::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery};
+use crate::ingress::kuberconsul::{ServiceDiscovery, clone_compare, list_to_upstreams};
-use crate::ingress::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate};
+use crate::ingress::kubewatcher::{AralezRouteUpdate, start_ingress_watcher_with_config};
-use rustls::crypto::ring::default_provider;
+use crate::utils::tools::Die;
 use rustls::crypto::CryptoProvider;
+use rustls::crypto::ring::default_provider;
 use tokio::sync::mpsc::Sender;
 
 #[derive(Debug, serde::Deserialize)]
@@ -66,12 +67,12 @@ impl ServiceDiscovery for KubernetesDiscovery {
 
             let end = servers.len().saturating_sub(1);
             let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 };
-            let server = servers.get(num).unwrap().to_string();
+            let server = servers.get(num).die("Getting k8s api server").to_string();
             let path = kuber.tokenpath.unwrap_or_else(|| "/var/run/secrets/kubernetes.io/serviceaccount/token".to_string());
             let token = kuberconsul::read_token(path.as_str()).await;
 
             if CryptoProvider::get_default().is_none() {
-                default_provider().install_default().expect("Failed to install rustls crypto provider");
+                default_provider().install_default().die("Failed to install rustls crypto provider");
             }
 
             let (tx, mut rx) = tokio::sync::mpsc::channel::<AralezRouteUpdate>(100);

src/lib.rs

@@ -1,4 +1,5 @@
 pub mod auth;
+pub mod cache;
 pub mod core;
 pub mod ingress;
 pub mod logging;

src/logging/core.rs

@@ -1,13 +1,14 @@
 use crate::logging::types::start_logging_backend;
 use crate::utils::metrics::LOGGING_ERRORS;
+use crate::utils::tools::Die;
 use crate::utils::types::AppConfig;
 use anyhow::Result;
-use log::{error, info, log, Level, LevelFilter, Record};
+use log::{Level, LevelFilter, Record, error, info, log};
+use log4rs::append::Append;
+use log4rs::append::rolling_file::RollingFileAppender;
+use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
 use log4rs::append::rolling_file::policy::compound::roll::fixed_window::FixedWindowRoller;
 use log4rs::append::rolling_file::policy::compound::trigger::size::SizeTrigger;
-use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
-use log4rs::append::rolling_file::RollingFileAppender;
-use log4rs::append::Append;
 use log4rs::config::Logger;
 use log4rs::{
     append::console::ConsoleAppender,
@@ -149,7 +150,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
                 .logger(Logger::builder().build("h2", LevelFilter::Off));
         }
 
-        let config = config_builder.build(Root::builder().appender("stdout").build(log_level)).unwrap();
+        let config = config_builder.build(Root::builder().appender("stdout").build(log_level)).die("Logging config builder");
 
         if let Err(e) = log4rs::init_config(config) {
             eprintln!("Warning: log4rs logger already initialized: {:?}", e);
@@ -177,7 +178,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
             _ => format!("{}.{{}}", path),
         };
 
-        let roller = FixedWindowRoller::builder().build(pattern_str.as_str(), keep).unwrap();
+        let roller = FixedWindowRoller::builder().build(pattern_str.as_str(), keep).die("Logging roller builder");
         let trigger = SizeTrigger::new(size_mb * 1024 * 1024);
         let policy = CompoundPolicy::new(Box::new(trigger), Box::new(roller));
 
@@ -193,7 +194,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
             .appender(Appender::builder().build("file", Box::new(file)))
             .logger(Logger::builder().build("pingora_proxy", LevelFilter::Off))
             .build(Root::builder().appender("file").build(log_level))
-            .unwrap();
+            .die("Parsing Lo4rsConfig");
         if let Err(e) = log4rs::init_config(config) {
             eprintln!("Warning: log4rs logger already initialized: {:?}", e);
         }
@@ -204,7 +205,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
             .appender(Appender::builder().build("stdout", Box::new(stdout)))
             .logger(Logger::builder().build("pingora_proxy", LevelFilter::Off))
             .build(Root::builder().appender("stdout").build(log_level))
-            .unwrap();
+            .die("Parsing Lo4rsConfig");
         if let Err(e) = log4rs::init_config(config) {
             eprintln!("Warning: log4rs logger already initialized: {:?}", e);
         }

src/tls/acme/dns/cloudflare.rs

@@ -1,5 +1,6 @@
 use crate::tls::acme::lookup::lookup_wait;
 use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
+use crate::utils::tools::Die;
 use log::info;
 use reqwest::Client;
 use serde_json::json;
@@ -12,8 +13,8 @@ pub struct CloudflareProvider {
 
 impl CloudflareProvider {
     pub fn new() -> Self {
-        let api_token = std::env::var("CLOUDFLARE_API_TOKEN").expect("Environment variable CLOUDFLARE_API_TOKEN is required");
+        let api_token = std::env::var("CLOUDFLARE_API_TOKEN").die("Environment variable CLOUDFLARE_API_TOKEN is required");
-        let zone_id = std::env::var("CLOUDFLARE_ZONE_ID").expect("Environment variable CLOUDFLARE_ZONE_ID is required");
+        let zone_id = std::env::var("CLOUDFLARE_ZONE_ID").die("Environment variable CLOUDFLARE_ZONE_ID is required");
 
         Self {
             client: Client::new(),

src/tls/acme/dns/digitalocean.rs

@@ -0,0 +1,71 @@
+use crate::tls::acme::lookup::lookup_wait;
+use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
+use crate::utils::tools::Die;
+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").die("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

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

src/tls/acme/dns/route53.rs

@@ -1,26 +1,123 @@
+// ------------------------ //
+// This file is vibecoded !
+// ------------------------ //
+
+use crate::tls::acme::lookup::lookup_wait;
 use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
+use crate::utils::tools::Die;
+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 {
-        Self {
+        let hosted_zone_id = std::env::var("AWS_HOSTED_ZONE_ID").die("Environment variable AWS_HOSTED_ZONE_ID is required");
-            hosted_zone_id: std::env::var("AWS_HOSTED_ZONE_ID").unwrap_or_default(),
+
-        }
+        let config = if let Ok(handle) = Handle::try_current() {
+            // If we're inside a tokio runtime, we can block on it directly
+            handle.block_on(aws_config::load_from_env())
+        } else {
+            // If we're completely outside tokio, spin up a temporary current-thread runtime
+            tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .die("Failed to create current-thread runtime")
+                .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)
     }
 }
 
-// Not working, just a placeholder.
 #[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>> {
+    async fn create_txt_record(&self, _domain: &str, name: &str, value: &str) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
-        println!(" ====> Route53 Created TXT Record for {}: {}: {}", _name, _domain, _value);
+        // AWS Route53 requires TXT strings to be wrapped in double quotes
-        Ok("route53_change_id".to_string())
+        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>> {
+
-        println!(" ====> Route53 Deleted TXT Record for {}: {}", _record_name, _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(())
     }
 }

src/tls/acme/http01.rs

@@ -5,7 +5,6 @@ use axum::body::Body;
 use axum::extract::State;
 use axum::http::{Response, StatusCode};
 use axum::response::IntoResponse;
-
 #[allow(clippy::needless_return)]
 pub async fn acme_create(State(state): State<webserver::AppState>) -> impl IntoResponse {
     match account::load_or_create(state.cert_creds.as_str()).await {
@@ -14,13 +13,10 @@ 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_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
         }
         Err(e) => {
-            return Response::builder()
+            return (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to create account {}\n", e)).into_response();
-                .status(StatusCode::INTERNAL_SERVER_ERROR)
-                .body(Body::from(format!("Failed to create account: {}", e)))
-                .unwrap();
         }
     };
 }
@@ -33,13 +29,10 @@ pub async fn acme_order(State(state): State<webserver::AppState>, axum::extract:
                 .status(StatusCode::OK)
                 .header("Content-Type", "text/plain")
                 .body(Body::from(txt))
-                .unwrap();
+                .unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
         }
         Err(e) => {
-            return Response::builder()
+            return (StatusCode::INTERNAL_SERVER_ERROR, format!("Internal Server Error: {}\n", e)).into_response();
-                .status(StatusCode::INTERNAL_SERVER_ERROR)
-                .body(Body::from(format!("Failed to order a certificate: {}", e)))
-                .unwrap();
         }
     };
 }
@@ -50,12 +43,8 @@ pub async fn http01_challenge(axum::extract::Path(token): axum::extract::Path<St
                 .status(StatusCode::OK)
                 .header("Content-Type", "text/plain")
                 .body(Body::from(key_authorization.clone()))
-                .unwrap();
+                .unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
         }
     }
-    Response::builder()
+    (StatusCode::NOT_FOUND, "Not found\n").into_response()
-        .status(StatusCode::NOT_FOUND)
-        .header("Content-Type", "text/plain")
-        .body(Body::from("Not found"))
-        .unwrap()
 }

src/tls/acme/lookup.rs

@@ -1,3 +1,4 @@
+use crate::utils::tools::Die;
 use hickory_resolver::TokioResolver;
 use hickory_resolver::config::{NameServerConfig, ResolverConfig, ResolverOpts};
 use hickory_resolver::net::runtime::TokioRuntimeProvider;
@@ -17,7 +18,7 @@ pub async fn lookup_wait(record_name: &str, dns_server_ip: Option<&str>, max_wai
     let resolver = TokioResolver::builder_with_config(config, TokioRuntimeProvider::default())
         .with_options(opts)
         .build()
-        .expect("failed to build DNS resolver");
+        .die("failed to build DNS resolver");
 
     let start = std::time::Instant::now();
     let poll_interval = Duration::from_secs(3);

src/tls/acme/order.rs

@@ -2,6 +2,7 @@ use crate::tls::acme::account::get_account;
 use crate::tls::acme::types::get_provider;
 use crate::utils::lazylock::CHALLENGES;
 use crate::utils::lazylock::DOMAINS;
+use crate::utils::tools::Die;
 use base64::Engine;
 use base64::engine::general_purpose::URL_SAFE_NO_PAD;
 use instant_acme::{ChallengeType, Identifier, NewOrder, RetryPolicy};
@@ -13,8 +14,6 @@ use std::fs;
 use std::time::Duration;
 use x509_parser::prelude::*;
 
-static DNS: bool = true;
-
 pub async fn refresh_order(certs_dir: String, autoconf_dir: String, provider: Option<String>) {
     let credsfile = autoconf_dir + "/acme_credentials.json";
     loop {
@@ -25,27 +24,24 @@ pub async fn refresh_order(certs_dir: String, autoconf_dir: String, provider: Op
     }
 }
 pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: Option<String>) -> Result<String, Box<dyn std::error::Error>> {
-    let crt = certs_dir.clone() + "/" + domain + ".crt";
+    let crt = format!("{}/{}.crt", certs_dir, domain);
-    let key = certs_dir.clone() + "/" + domain + ".key";
+    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();
+        let newlist: Vec<String> = DOMAINS.iter().map(|item| item.key().to_string()).collect();
-        for item in DOMAINS.iter() {
-            newlist.push(item.key().to_string());
-        }
         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));
             }
         }
     }
+
     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());
         }
     };
@@ -57,17 +53,21 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
 
     while let Some(auth) = authorizations.next().await {
         let mut auth = auth?;
-        let domain = auth.identifier().to_string();
+        let domain_name = auth.identifier().to_string();
-        let challenge_type = if DNS { ChallengeType::Dns01 } else { ChallengeType::Http01 };
+        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("No challenge found")?;
+        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();
 
-        if let Some(prov) = provider.clone() {
+        if is_dns {
+            let prov = provider.as_ref().die("Failed to get DNS provider");
             let dns_provider = get_provider(prov.clone());
             let dns_value = calculate_dns_value(&key_auth_str);
-            let clean_domain = domain.trim_start_matches("*.").to_string();
+            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 {
@@ -82,8 +82,9 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
                 }
             }
         } else {
-            let token = key_auth_str.split('.').next().ok_or("invalid key authorization")?.to_string();
+            let token = key_auth_str.split('.').next().ok_or("Invalid key authorization")?.to_string();
-            CHALLENGES.write().unwrap().insert(token, key_auth_str);
+            CHALLENGES.write().die("Failed to write challenge").insert(token, key_auth_str);
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
             challenge_handle.set_ready().await?;
         }
     }
@@ -97,12 +98,9 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
     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();
+    CHALLENGES.write().die("Challenges lock poisoned").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())

src/tls/acme/types.rs

@@ -9,10 +9,10 @@ pub struct DnsBackendPlugin {
 
 inventory::collect!(DnsBackendPlugin);
 
-static PROVIDERS: LazyLock<HashMap<&'static str, Box<dyn DnsProvider>>> = LazyLock::new(|| {
+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.insert(plugin.name, plugin.factory);
     }
     map
 });
@@ -24,9 +24,9 @@ pub trait DnsProvider: 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) -> &'static dyn DnsProvider {
+pub fn get_provider(provider: String) -> Box<dyn DnsProvider> {
-    PROVIDERS
+    FACTORIES
         .get(provider.as_str())
-        .map(|boxed| boxed.as_ref())
+        .map(|factory| factory())
-        .unwrap_or_else(|| panic!("Unknown DNS provider '{}'. Available providers: {:?}", provider, PROVIDERS.keys().collect::<Vec<_>>()))
+        .unwrap_or_else(|| panic!("Unknown DNS provider '{}'. Available providers: {:?}", provider, FACTORIES.keys().collect::<Vec<_>>()))
 }

src/tls/load.rs

@@ -1,8 +1,9 @@
 use crate::tls::grades;
+use crate::utils::tools::Die;
 use dashmap::DashMap;
 use log::error;
 use pingora::tls::ssl::{NameType, SniError, SslAlert, SslContext, SslFiletype, SslMethod, SslRef};
-use rustls_pemfile::{read_one, Item};
+use rustls_pemfile::{Item, read_one};
 use serde::Deserialize;
 use std::collections::HashSet;
 use std::fs::File;
@@ -10,6 +11,7 @@ use std::io::BufReader;
 use x509_parser::extensions::GeneralName;
 use x509_parser::nom::Err as NomErr;
 use x509_parser::prelude::*;
+
 #[derive(Clone, Deserialize, Debug)]
 pub struct CertificateConfig {
     pub cert_path: String,
@@ -37,7 +39,7 @@ pub struct Certificates {
 
 impl Certificates {
     pub fn new(configs: &Vec<CertificateConfig>, _grade: &str) -> Option<Self> {
-        let default_cert = configs.first().expect("At least one TLS certificate required");
+        let default_cert = configs.first().die("At least one TLS certificate required");
         let mut cert_infos = Vec::new();
         let name_map: DashMap<String, SslContext> = DashMap::new();
         for config in configs {

src/utils/deser.rs

@@ -0,0 +1,298 @@
+use crate::utils::hcclient::{HttpClientRequest, httpclient};
+use crate::utils::lazylock::{GLOBAL_PARAMS, MASTER_KEY, WORKERS};
+use crate::utils::tools::Die;
+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().die("Failed to read config");
+    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().die("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().die("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/grpc.rs

@@ -1,23 +1,30 @@
 use bytes::Bytes;
 use http_body_util::Full;
 use hyper::Request;
-use hyper_util::client::legacy::connect::HttpConnector;
 use hyper_util::client::legacy::Client;
+use hyper_util::client::legacy::connect::HttpConnector;
 use hyper_util::rt::TokioExecutor;
+use log::error;
 
 pub async fn ping_grpc(addr: &str) -> bool {
     let connector = HttpConnector::new();
     let client = Client::builder(TokioExecutor::new()).http2_only(true).build(connector);
     let uri = format!("{}{}", addr, "/aralez.Probe/Ping");
     let body = Full::new(Bytes::from(vec![0, 0, 0, 0, 0]));
-    let request = Request::post(uri).header("content-type", "application/grpc").header("te", "trailers").body(body).unwrap();
+
-    match client.request(request).await {
+    match Request::post(uri.clone()).header("content-type", "application/grpc").header("te", "trailers").body(body) {
-        Ok(resp) => resp
+        Ok(request) => match client.request(request).await {
-            .headers()
+            Ok(resp) => resp
-            .get("content-type")
+                .headers()
-            .and_then(|v| v.to_str().ok())
+                .get("content-type")
-            .map(|v| v.starts_with("application/grpc"))
+                .and_then(|v| v.to_str().ok())
-            .unwrap_or(false),
+                .map(|v| v.starts_with("application/grpc"))
-        Err(_) => false,
+                .unwrap_or(false),
+            Err(_) => false,
+        },
+        Err(e) => {
+            error!("Failed to create request: {}", e);
+            false
+        }
     }
 }

src/utils/hcclient.rs

@@ -18,9 +18,10 @@ pub struct HttpClientRequest<'a> {
     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) {
+pub async fn httpclient(req: HttpClientRequest<'_>) -> (bool, bool, u16) {
     let HttpClientRequest {
         method,
         tls,
@@ -30,6 +31,7 @@ pub async fn httpclient(req: HttpClientRequest<'_>) -> (bool, bool) {
         port,
         url,
         payload,
+        headers,
     } = req;
     let method = match method {
         "HEAD" => "HEAD",
@@ -57,7 +59,7 @@ pub async fn httpclient(req: HttpClientRequest<'_>) -> (bool, bool) {
             } else {
                 log::warn!("Fail to connect to addr: {}, tls {}, host {}", address, tls, host);
             }
-            return (false, false);
+            return (false, false, 500);
         }
     };
 
@@ -68,7 +70,7 @@ pub async fn httpclient(req: HttpClientRequest<'_>) -> (bool, bool) {
         }
         None => {
             if ping_grpc(url).await {
-                return (true, true);
+                return (true, true, 204);
             }
         }
     }
@@ -77,7 +79,7 @@ pub async fn httpclient(req: HttpClientRequest<'_>) -> (bool, bool) {
         Ok(r) => r,
         Err(e) => {
             log::warn!("Failed to build request: {}", e);
-            return (false, false);
+            return (false, false, 500);
         }
     };
 
@@ -85,113 +87,33 @@ pub async fn httpclient(req: HttpClientRequest<'_>) -> (bool, bool) {
     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::deser::{send_to_worker, serialize_config};
 use crate::utils::hcclient::{HttpClientRequest, httpclient};
-use crate::utils::lazylock::REVERSE_STORE;
+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 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(crate) async fn initiate_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsDashMap {
-pub async fn initiate_upstreams(fullist: UpstreamsDashMap) -> UpstreamsDashMap {
-    build_upstreams(&fullist, "HEAD").await
-}
-
-async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsDashMap {
     let totest: UpstreamsDashMap = DashMap::new();
     let fclone = clone_dashmap(fullist);
     for val in fclone.iter() {
@@ -47,13 +63,13 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsD
                 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,
@@ -77,6 +93,7 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsD
                         port: upstream.port,
                         url: link.as_str(),
                         payload: EMPTY_PAYLOAD,
+                        headers: None,
                     })
                     .await;
                     if resp.0 {
@@ -109,7 +126,7 @@ async fn detect_tls(ip: &str, port: &u16, sni: &str) -> bool {
             .danger_accept_invalid_certs(true)
             .danger_accept_invalid_hostnames(true)
             .build()
-            .unwrap(),
+            .die("Failed to create TLS connector"),
     );
     connector.connect(sni, stream).await.is_ok()
 }

src/utils/lazylock.rs

@@ -1,3 +1,4 @@
+use crate::utils::types::{Configuration, WorkerNodes};
 use dashmap::DashMap;
 use moka::sync::Cache;
 use pingora_cache::MemCache;
@@ -20,3 +21,8 @@ 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/metrics.rs

@@ -1,9 +1,10 @@
 use crate::utils::lazylock::EVICTION;
+use crate::utils::tools::Die;
 use pingora_cache::eviction::EvictionManager;
 use pingora_http::Method;
 use pingora_http::StatusCode;
 use pingora_http::Version;
-use prometheus::{register_histogram, register_int_counter, register_int_counter_vec, register_int_gauge, Histogram, IntCounter, IntCounterVec, IntGauge};
+use prometheus::{Histogram, IntCounter, IntCounterVec, IntGauge, register_histogram, register_int_counter, register_int_counter_vec, register_int_gauge};
 use std::sync::Arc;
 use std::sync::LazyLock;
 use std::time::Duration;
@@ -16,19 +17,28 @@ pub struct MetricTypes {
     pub version: Version,
 }
 
-pub static OPEN_FILES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_open_files", "Number of open file descriptors").unwrap());
+pub static OPEN_FILES: LazyLock<IntGauge> =
-pub static LOGGING_ERRORS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_logging_errors", "Number of log errors").unwrap());
+    LazyLock::new(|| register_int_gauge!("aralez_open_files", "Number of open file descriptors").die("Failed to register OPEN_FILES metric"));
-pub static MEMORY_USAGE: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_memory_bytes", "Total memory allocated in bytes").unwrap());
+pub static LOGGING_ERRORS: LazyLock<IntGauge> =
-pub static ACTIVE_SESSIONS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").unwrap());
+    LazyLock::new(|| register_int_gauge!("aralez_logging_errors", "Number of log errors").die("Failed to register LOGGING_ERRORS metric"));
-pub static REQUEST_COUNT: LazyLock<IntCounter> = LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").unwrap());
+pub static MEMORY_USAGE: LazyLock<IntGauge> =
+    LazyLock::new(|| register_int_gauge!("aralez_memory_bytes", "Total memory allocated in bytes").die("Failed to register MEMORY_USAGE metric"));
+pub static ACTIVE_SESSIONS: LazyLock<IntGauge> =
+    LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").die("Failed to register ACTIVE_SESSIONS metric"));
+pub static REQUEST_COUNT: LazyLock<IntCounter> =
+    LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").die("Failed to register REQUEST_COUNT metric"));
 
-pub static CACHE_SIZE_BYTES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_size_bytes", "Current cache size in bytes").unwrap());
+pub static CACHE_SIZE_BYTES: LazyLock<IntGauge> =
-pub static CACHE_ITEMS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_items", "Current number of cached objects").unwrap());
+    LazyLock::new(|| register_int_gauge!("aralez_cache_size_bytes", "Current cache size in bytes").die("Failed to register CACHE_SIZE_BYTES metric"));
-pub static CACHE_EVICTED_BYTES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_bytes_total", "Total bytes evicted from cache").unwrap());
+pub static CACHE_ITEMS: LazyLock<IntGauge> =
+    LazyLock::new(|| register_int_gauge!("aralez_cache_items", "Current number of cached objects").die("Failed to register CACHE_ITEMS metric"));
+pub static CACHE_EVICTED_BYTES: LazyLock<IntGauge> =
+    LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_bytes_total", "Total bytes evicted from cache").die("Failed to register CACHE_EVICTED_BYTES metric"));
 
-pub static CACHE_EVICTED_ITEMS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_items_total", "Total cache items evicted").unwrap());
+pub static CACHE_EVICTED_ITEMS: LazyLock<IntGauge> =
+    LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_items_total", "Total cache items evicted").die("Failed to register CACHE_EVICTED_ITEMS metric"));
 pub static RESPONSE_CODES: LazyLock<IntCounterVec> =
-    LazyLock::new(|| register_int_counter_vec!("aralez_responses_total", "Responses grouped by status code", &["status"]).unwrap());
+    LazyLock::new(|| register_int_counter_vec!("aralez_responses_total", "Responses grouped by status code", &["status"]).die("Failed to register RESPONSE_CODES metric"));
 
 pub static RESPONSE_LATENCY: LazyLock<Histogram> = LazyLock::new(|| {
     register_histogram!(
@@ -36,17 +46,20 @@ pub static RESPONSE_LATENCY: LazyLock<Histogram> = LazyLock::new(|| {
         "Response latency in seconds",
         vec![0.001, 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0]
     )
-    .unwrap()
+    .die("Failed to register RESPONSE_LATENCY metric")
 });
 
-pub static REQUESTS_BY_METHOD: LazyLock<IntCounterVec> =
+pub static REQUESTS_BY_METHOD: LazyLock<IntCounterVec> = LazyLock::new(|| {
-    LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_method_total", "Number of requests by HTTP method", &["method"]).unwrap());
+    register_int_counter_vec!("aralez_requests_by_method_total", "Number of requests by HTTP method", &["method"]).die("Failed to register REQUESTS_BY_METHOD metric")
+});
 
-pub static REQUESTS_BY_UPSTREAM: LazyLock<IntCounterVec> =
+pub static REQUESTS_BY_UPSTREAM: LazyLock<IntCounterVec> = LazyLock::new(|| {
-    LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_upstream", "Number of requests by UPSTREAM server", &["upstream"]).unwrap());
+    register_int_counter_vec!("aralez_requests_by_upstream", "Number of requests by UPSTREAM server", &["upstream"]).die("Failed to register REQUESTS_BY_UPSTREAM metric")
+});
 
-pub static REQUESTS_BY_VERSION: LazyLock<IntCounterVec> =
+pub static REQUESTS_BY_VERSION: LazyLock<IntCounterVec> = LazyLock::new(|| {
-    LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_version_total", "Number of requests by HTTP versions", &["version"]).unwrap());
+    register_int_counter_vec!("aralez_requests_by_version_total", "Number of requests by HTTP versions", &["version"]).die("Failed to register REQUESTS_BY_VERSION metric")
+});
 pub static HTTP_11_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/1.1"]));
 pub static HTTP_20_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/2.0"]));
 pub static HTTP_30_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/3.0"]));

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,7 +1,8 @@
+use crate::utils::deser::{send_to_worker, serialize_config};
 use crate::utils::healthcheck;
-use crate::utils::lazylock::{DOMAINS, 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::{Die, clone_dashmap_into};
 use crate::utils::types::*;
 use dashmap::DashMap;
 use log::{error, info, warn};
@@ -47,8 +48,8 @@ pub async fn load_configuration(d: &str, kind: &str) -> (Option<Configuration>,
                 return (None, err_msg);
             }
 
-            let mut confdir = Path::new(d).parent().unwrap().to_path_buf();
+            let mut confdir = Path::new(d).parent().die("Invalid configuration path").to_path_buf();
-            let mut autocfg = Path::new(d).parent().unwrap().to_path_buf();
+            let mut autocfg = Path::new(d).parent().die("Invalid configuration path").to_path_buf();
 
             autocfg.push("autoconfigs");
             if fs::metadata(autocfg.clone()).is_err() {
@@ -223,8 +224,8 @@ 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_ssl: path_config.is_ssl.unwrap_or(false),
-                                is_http2: 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.unwrap_or(0),
                                 x4xx_limit: path_config.x4xx_limit.unwrap_or(0),
@@ -239,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().die("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(&imtdashmap, "HEAD").await;
-            let r = healthcheck::initiate_upstreams(y).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 data = fs::read_to_string(path).die("Failed to read main config file");
-    let mut cfo: AppConfig = noyalib::from_str(&data).expect("Failed to parse main config file");
+    let mut cfo: AppConfig = noyalib::from_str(&data).die("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(':') {
@@ -279,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()))).die("Config can only be initialized once");
+        let mk = cfo.master_key.clone().die("In CLuster mode master key must be set");
+        MASTER_KEY.set(Some(Arc::from(mk.as_str()))).die("Config can only be initialized once");
+        if member == "master" {
+            IM_A_MASTER.set(true).die("Master status can only be initialized once");
+            let worker_str = cfo.worker_nodes.as_deref().die("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>().die("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>().die("Invalid worker port number");
+                //     wks.push((host.to_string(), port));
+                // } else {
+                //     panic!("Invalid worker format, expected host:port");
+                // }
+            }
+            WORKERS.set(wks).die("Workers list can only be initialized once");
+        }
+    }
     cfo
 }
 

src/utils/state.rs

@@ -1,3 +1,4 @@
+use crate::utils::tools::Die;
 use std::sync::{LazyLock, RwLock};
 
 #[derive(Debug)]
@@ -8,11 +9,11 @@ pub struct SharedState {
 pub static GLOBAL_STATE: LazyLock<RwLock<SharedState>> = LazyLock::new(|| RwLock::new(SharedState { first_run: true }));
 
 pub fn mark_not_first_run() {
-    let mut state = GLOBAL_STATE.write().unwrap();
+    let mut state = GLOBAL_STATE.write().die("Writing global state failed");
     state.first_run = false;
 }
 
 pub fn is_first_run() -> bool {
-    let state = GLOBAL_STATE.read().unwrap();
+    let state = GLOBAL_STATE.read().die("Writing global state failed");
     state.first_run
 }

src/utils/tools.rs

@@ -7,13 +7,14 @@ use privdrop::PrivDrop;
 use serde_json::{Value, json};
 use sha2::{Digest, Sha256};
 use std::collections::{HashMap, HashSet};
-use std::fmt::Write;
+use std::fmt::{Debug, Write};
 use std::fs::OpenOptions;
 use std::io::Write as IoWrite;
 use std::net::SocketAddr;
 use std::net::TcpListener;
 use std::os::unix::fs::MetadataExt;
 use std::os::unix::fs::OpenOptionsExt;
+use std::panic::Location;
 use std::process::Command;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -23,9 +24,9 @@ use std::{env, fs, process, thread, time};
 pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams) {
     let mut out = String::new();
     for host_entry in upstreams.iter() {
-        writeln!(out, "Hostname: {}", host_entry.key()).unwrap();
+        writeln!(out, "Hostname: {}", host_entry.key()).unwrap_or_default();
         for path_entry in host_entry.value().iter() {
-            writeln!(out, "    Path: {}", path_entry.key()).unwrap();
+            writeln!(out, "    Path: {}", path_entry.key()).unwrap_or_default();
             for f in path_entry.value().0.clone() {
                 writeln!(
                     out,
@@ -38,7 +39,7 @@ pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams)
                     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();
+                .unwrap_or_default();
             }
         }
     }
@@ -161,9 +162,9 @@ pub fn clone_idmap_into(original: &UpstreamsDashMap, cloned: &UpstreamsIdMap) {
 pub fn listdir(dir: String) -> Vec<load::CertificateConfig> {
     let mut f = HashMap::new();
     let mut certificate_configs: Vec<load::CertificateConfig> = vec![];
-    let paths = fs::read_dir(dir).unwrap();
+    let paths = fs::read_dir(dir).die("listing certificates directory failed");
     for path in paths {
-        let path_str = path.unwrap().path().to_str().unwrap().to_owned();
+        let path_str = path.die("getting certs path").path().to_str().die("certs pat t &str").to_owned();
         if path_str.ends_with(".crt") {
             let name = path_str.replace(".crt", "");
             let mut inner = vec![];
@@ -212,9 +213,9 @@ pub fn drop_priv(user: String, group: String, http_addr: String, tls_addr: Optio
 }
 
 pub fn check_priv(addr: &str) {
-    let port = SocketAddr::from_str(addr).map(|sa| sa.port()).expect("Failed to parse address port ");
+    let port = SocketAddr::from_str(addr).map(|sa| sa.port()).die("Failed to parse address port ");
     if port < 1024 {
-        let meta = std::fs::metadata("/proc/self").map(|m| m.uid()).unwrap();
+        let meta = std::fs::metadata("/proc/self").map(|m| m.uid()).die("Reading /proc/self");
         if meta != 0 {
             error!("Running on privileged port requires to start as ROOT");
             process::exit(1)
@@ -364,3 +365,40 @@ pub fn get_hostname() -> String {
         .filter(|s| !s.is_empty())
         .unwrap_or_else(|| "unknown".to_string())
 }
+
+// pub fn die(message: &str) -> ! {
+//     log::error!("{message}");
+//     process::exit(1);
+// }
+
+pub trait Die<T> {
+    fn die(self, message: &str) -> T;
+}
+
+impl<T, E: Debug> Die<T> for Result<T, E> {
+    #[track_caller]
+    fn die(self, message: &str) -> T {
+        match self {
+            Ok(value) => value,
+            Err(e) => {
+                let loc = Location::caller();
+                eprintln!("\x1b[1;31mERROR:\x1b[0m {message}: {e:?}: {}: {}", loc.file(), loc.line());
+                process::exit(1);
+            }
+        }
+    }
+}
+
+impl<T> Die<T> for Option<T> {
+    #[track_caller]
+    fn die(self, message: &str) -> T {
+        match self {
+            Some(value) => value,
+            None => {
+                let loc = Location::caller();
+                eprintln!("\x1b[1;31mERROR:\x1b[0m {message} at {}:{}", loc.file(), loc.line());
+                process::exit(1);
+            }
+        }
+    }
+}

src/utils/types.rs

@@ -9,7 +9,7 @@ pub type UpstreamsDashMap = DashMap<Arc<str>, DashMap<Arc<str>, (Vec<Arc<InnerMa
 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 sticky_sessions: u64,
     pub rate_limit: isize,
@@ -92,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 {
@@ -110,6 +112,8 @@ pub struct AppConfig {
     pub hc_method: String,
     pub upstreams_conf: String,
     pub log_level: String,
+    pub cache_enabled: Option<bool>,
+    pub cache_storage: Option<String>,
     pub cache_size_mb: Option<usize>,
     pub cache_ttl: Option<u64>,
     pub access_log: Option<String>,
@@ -138,9 +142,11 @@ pub struct AppConfig {
     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>,
@@ -201,3 +207,11 @@ 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

@@ -1,6 +1,6 @@
 use crate::tls::CertificateConfig;
 use crate::utils::parceyaml::load_configuration;
-use crate::utils::tools::listdir;
+use crate::utils::tools::{Die, listdir};
 use crate::utils::types::Configuration;
 use log::{error, info};
 use notify::event::ModifyKind;
@@ -15,7 +15,7 @@ use tokio::task;
 pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
     sleep(Duration::from_millis(50)).await; // For having nice logs :-)
     let file_path = fp.as_str();
-    let parent_dir = Path::new(file_path).parent().unwrap();
+    let parent_dir = Path::new(file_path).parent().die("Get parent folder");
     let (local_tx, mut local_rx) = tokio::sync::mpsc::channel::<notify::Result<Event>>(1);
 
     let _watcher_handle = task::spawn_blocking({
@@ -27,8 +27,8 @@ pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
                 },
                 Config::default(),
             )
-            .unwrap();
+            .die("Initialize file watcher");
-            watcher.watch(&parent_dir, RecursiveMode::Recursive).unwrap();
+            watcher.watch(&parent_dir, RecursiveMode::Recursive).die("");
             let (_rtx, mut rrx) = tokio::sync::mpsc::channel::<bool>(1);
             let _ = rrx.blocking_recv();
         }
@@ -42,7 +42,7 @@ pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
                     start = Instant::now();
                     let snd = load_configuration(file_path, "filepath").await.0;
                     if let Some(snd) = snd {
-                        toreturn.send(snd).await.unwrap();
+                        toreturn.send(snd).await.die("Send to file watcher");
                     }
                 }
                 _ => (),