fixed closed #52

Cached headers bug fix

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "aralez"
-version = "0.92.18"
+version = "0.93.3"
 edition = "2021"
 license = "Apache-2.0"
 description = "Reverse proxy built on top of Cloudflare's Pingora"
@@ -23,12 +23,12 @@ pingora-http = "0.8.1"
 pingora-limits = "0.8.1"
 pingora-cache = "0.8.1"
 dashmap = "7.0.0-rc2"
-async-trait = "0.1.91"
-log = "0.4.33"
+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"] }
-noyalib = { version = "0.0.18", features = ["compat-serde-yaml"] }
+noyalib = { version = "0.0.32", features = ["compat-serde-yaml"] }
 rand = "0.10.2"
 base64 = "0.23.1"
 jsonwebtoken = { version = "11.0.0", default-features = false, features = ["use_pem", "rust_crypto"] }
@@ -39,22 +39,26 @@ arc-swap = "1.9.2"
 prometheus = "0.14.0"
 x509-parser = "0.18.1"
 rustls-pemfile = "2.2.0"
-tower-http = { version = "0.7.0", features = ["fs"] }
+tower-http = { version = "0.7.1", features = ["fs"] }
 privdrop = "0.5.6"
 serde_json = "1.0.151"
 subtle = "2.6.1"
-moka = { version = "0.12.15", features = ["sync"] }
+moka = { version = "0.12.16", features = ["sync"] }
 ahash = "0.8.12"
 instant-acme = "0.8.5"
-rcgen = "0.14.8"
+rcgen = "0.14.10"
 log4rs = { version = "1.4.0", features = ["rolling_file_appender", "compound_policy", "size_trigger", "fixed_window_roller", "gzip"] }
 mimalloc = { version = "0.1.52", default-features = false }
 signal-hook = "0.4.4"
 sd-notify = "0.5.0"
 libc = "0.2.186"
 tokio-native-tls = "0.3.1"
-hyper = "1.11.0"
+hyper = "1.11.1"
 hyper-util = "0.1.20"
-http-body-util = "0.1.4"
+http-body-util = "0.1.5"
 bytes = "1.12.1"
 
+rustls = { version = "0.23", default-features = false, features = ["ring"] }
+kube = { version = "4.2.0", features = ["runtime", "client", "derive"] }
+k8s-openapi = { version = "0.28.0", features = ["v1_32"] }
+futures = "0.3.34"

etc/consul/apache.hcl.example

@@ -0,0 +1,66 @@
+job "apache" {
+  meta {
+    uuid = uuidv4()
+  }
+  datacenters = ["eu-ams-1"]
+  type = "service"
+  constraint {
+    attribute = node.unique.name
+    operator  = "regexp"
+    value     = "consul[0-9]"
+  }
+  update {
+    max_parallel      = 1
+    min_healthy_time  = "10s"
+    healthy_deadline  = "3m"
+    progress_deadline = "10m"
+    auto_revert       = false
+    canary            = 0
+  }
+  migrate {
+    max_parallel     = 1
+    health_check     = "checks"
+    min_healthy_time = "10s"
+    healthy_deadline = "5m"
+  }
+  group "Apache" {
+    network {
+      port "apache" { to = 80 }
+    }
+    constraint {
+      operator = "distinct_hosts"
+      value    = "true"
+    }
+    count = 3
+    restart {
+      attempts = 2
+      interval = "30m"
+      delay    = "15s"
+      mode     = "fail"
+    }
+    task "health" {
+      service {
+        port = "apache"
+        tags = [
+          "aralez.service=yes",
+          "aralez.host=apache.example.com",
+          "aralez.path=/",
+          "aralez.redirect=http://checkip.amazonaws.com"
+        ]
+        check {
+          type     = "http"
+          port     = "apache"
+          path     = "/"
+          interval = "5s"
+          timeout  = "2s"
+        }
+      }
+      driver = "docker"
+      config {
+        image      = "httpd:latest"
+        force_pull = true
+        ports = ["apache"]
+      }
+    }
+  }
+}

etc/consul/echo.hcl.example

@@ -0,0 +1,70 @@
+job "echo" {
+  meta {
+    uuid = uuidv4()
+  }
+  datacenters = ["eu-ams-1"]
+  type = "service"
+  constraint {
+    attribute = node.unique.name
+    operator  = "regexp"
+    value     = "consul[0-9]"
+  }
+  update {
+    max_parallel      = 1
+    min_healthy_time  = "10s"
+    healthy_deadline  = "3m"
+    progress_deadline = "10m"
+    auto_revert       = false
+    canary            = 0
+  }
+  migrate {
+    max_parallel     = 1
+    health_check     = "checks"
+    min_healthy_time = "10s"
+    healthy_deadline = "5m"
+  }
+  group "Echo" {
+    network {
+      port "echo" { to = 80 }
+    }
+    constraint {
+      operator = "distinct_hosts"
+      value    = "true"
+    }
+    count = 3
+    restart {
+      attempts = 2
+      interval = "30m"
+      delay    = "15s"
+      mode     = "fail"
+    }
+    task "server" {
+      service {
+        port = "echo"
+        tags = [
+          "aralez.service=yes",
+          "aralez.host=echo.example.com",
+          "aralez.path=/",
+          "aralez.rate=20",
+          "aralez.4xx_rate=10",
+          "aralez.to_https=true",
+          "aralez.client_header=X-Client-Header:Some simple header",
+          "aralez.server_header=X-Server-Header:Some simple header",
+        ]
+        check {
+          type     = "http"
+          port     = "echo"
+          path     = "/"
+          interval = "5s"
+          timeout  = "2s"
+        }
+      }
+      driver = "docker"
+      config {
+        image      = "ealen/echo-server:latest"
+        force_pull = true
+        ports = ["echo"]
+      }
+    }
+  }
+}

etc/consul/nginx.hcl.example

@@ -0,0 +1,72 @@
+job "nginx" {
+  meta {
+    uuid = uuidv4()
+  }
+  datacenters = ["eu-ams-1"]
+  type = "service"
+  constraint {
+    attribute = node.unique.name
+    operator  = "regexp"
+    value     = "consul[0-9]"
+  }
+  update {
+    max_parallel      = 1
+    min_healthy_time  = "10s"
+    healthy_deadline  = "3m"
+    progress_deadline = "10m"
+    auto_revert       = false
+    canary            = 0
+  }
+  migrate {
+    max_parallel     = 1
+    health_check     = "checks"
+    min_healthy_time = "10s"
+    healthy_deadline = "5m"
+  }
+  group "NginX" {
+    network {
+      port "nginx" { to = 80 }
+    }
+    constraint {
+      operator = "distinct_hosts"
+      value    = "true"
+    }
+    count = 3
+    restart {
+      attempts = 2
+      interval = "30m"
+      delay    = "15s"
+      mode     = "fail"
+    }
+    task "server" {
+      service {
+        port = "nginx"
+        tags = [
+          "aralez.service=yes",
+          "aralez.host=nginx.example.com",
+          "aralez.path=/",
+          "aralez.rate=1",
+          "aralez.4xx_rate=10",
+          "aralez.to_https=true",
+          "aralez.client_header=X-Some:Some Random Header",
+          "aralez.client_header=X-Emos:Redaeh Modnar Emos",
+          "aralez.server_header=X-From:Aralez & Consul",
+          "aralez.server_header=X-Mrof: Lusnoc & Zelara",
+        ]
+        check {
+          type     = "http"
+          port     = "nginx"
+          path     = "/"
+          interval = "5s"
+          timeout  = "2s"
+        }
+      }
+      driver = "docker"
+      config {
+        image      = "nginx:latest"
+        force_pull = true
+        ports = ["nginx"]
+      }
+    }
+  }
+}

etc/kubernetes/aralez-certificate-secret.yml

@@ -0,0 +1,9 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: aralez-tls
+type: kubernetes.io/tls
+data:
+  tls.crt: <base64-cert>
+  tls.key: <base64-key>
+

etc/kubernetes/aralez-example-deployment.yml

@@ -0,0 +1,43 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: aralez
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: aralez
+  template:
+    metadata:
+      labels:
+        app: aralez
+    spec:
+      serviceAccountName: aralez-sa
+      containers:
+        - name: aralez
+          image: sadoyan/aralez:latest
+          ports:
+            - containerPort: 80
+            - containerPort: 443
+          volumeMounts:
+            - name: main-config
+              mountPath: /etc/aralez/main.yaml
+              subPath: main.yaml
+              readOnly: true
+            - name: upstreams-config
+              mountPath: /etc/aralez/upstreams.yml
+              subPath: upstreams.yml
+              readOnly: true
+            - name: tls-certs
+              mountPath: /etc/aralez/certificates
+              readOnly: true
+      volumes:
+        - name: main-config
+          configMap:
+            name: aralez-main-config
+        - name: upstreams-config
+          configMap:
+            name: aralez-upstreams-config
+        - name: tls-certs
+          secret:
+            secretName: aralez-tls

etc/kubernetes/aralez-example-main.yml

@@ -0,0 +1,16 @@
+threads: 12
+daemon: false
+upstream_keepalive_pool_size: 500
+pid_file: /tmp/aralez.pid
+error_log: /tmp/aralez_err.log
+upgrade_sock: /tmp/aralez.sock
+config_api_enabled: false
+config_address: 127.0.0.1:3000
+proxy_address_http: 0.0.0.0:80
+proxy_address_tls: 0.0.0.0:443
+proxy_configs: /etc/aralez
+proxy_tls_grade: high
+upstreams_conf: /etc/aralez/upstreams.yml
+log_level: info
+hc_method: HEAD
+hc_interval: 2

etc/kubernetes/aralez-example-upstreams.yml

@@ -0,0 +1,14 @@
+# The file under watch and hot reload, changes are applied immediately, no need to restart or reload.
+provider: "kubernetes" # "file" "consul" "kubernetes"
+sticky_sessions: 172000
+to_https: false
+rate_limit: 500000
+x4xx_limit: 100000
+server_headers:
+  - "X-Forwarded-Proto:https"
+  - "X-Forwarded-Port:443"
+client_headers:
+  - "X-Global-Client:Yooooooo"
+kubernetes:
+  servers:
+    - "127.0.0.1:6443" # Gets KUBERNETES_SERVICE_HOST : KUBERNETES_SERVICE_PORT_HTTPS env variables.

etc/kubernetes/nginx-example-deployment.yml

@@ -0,0 +1,21 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: svc-nginx
+  labels:
+    app: svc-nginx
+spec:
+  selector:
+    matchLabels:
+      app: svc-nginx
+  replicas: 3
+  template:
+    metadata:
+      labels:
+        app: svc-nginx
+    spec:
+      containers:
+      - name: svc-nginx
+        image: nginx:latest
+        ports:
+        - containerPort: 80

etc/kubernetes/nginx-example-ingress.yml

@@ -0,0 +1,24 @@
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: wss-service-v2
+  namespace: default
+  annotations:
+    # These custom annotations are parsed by Aralez
+    aralez.rs/rate_limit: "50"
+    aralez.rs/x4xx_limit: "10"
+    aralez.rs/client_headers: '["X-Some-Client:Some Custom Header", "X-Example-Client:An Example Header"]'
+    aralez.rs/server_headers: '["X-Some-Server:Some Custom Header", "X-Example-Server:An Example Header"]'
+spec:
+  ingressClassName: aralez  # Matches the IngressClass metadata
+  rules:
+    - host: wss-service-v2.blabla.com
+      http:
+        paths:
+          - path: /ws
+            pathType: Prefix
+            backend:
+              service:
+                name: svc-nginx
+                port:
+                  number: 80

etc/kubernetes/nginx-example-service.yml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: svc-nginx
+spec:
+  selector:
+    app: svc-nginx
+  ports:
+    - protocol: TCP
+      port: 80
+      targetPort: 80

src/utils.rs

@@ -1,4 +1,5 @@
 pub mod auth;
+pub mod consul;
 pub mod discovery;
 mod filewatch;
 pub mod fordebug;
@@ -8,6 +9,8 @@ pub mod healthcheck;
 pub mod httpclient;
 pub mod jwt;
 pub mod kuberconsul;
+pub mod kubernetes;
+pub mod kubewatcher;
 pub mod lazylock;
 pub mod metrics;
 pub mod parceyaml;

src/utils/consul.rs

@@ -0,0 +1,91 @@
+use crate::utils::httpclient;
+use crate::utils::kuberconsul::{list_to_upstreams, ServiceDiscovery};
+use crate::utils::parceyaml::build_headers;
+use crate::utils::structs::{Configuration, GlobalServiceMapping, UpstreamsDashMap};
+use async_trait::async_trait;
+use dashmap::DashMap;
+use pingora::prelude::sleep;
+use rand::RngExt;
+use serde::Deserialize;
+use std::env;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::mpsc::Sender;
+
+#[derive(Debug, Deserialize)]
+pub struct ConsulService {
+    #[serde(rename = "Address")]
+    pub address: String,
+    #[serde(rename = "ServicePort")]
+    pub port: u16,
+}
+pub struct ConsulDiscovery;
+
+#[async_trait]
+impl ServiceDiscovery for ConsulDiscovery {
+    async fn fetch_upstreams(&self, config: Arc<Configuration>, toreturn: Sender<Configuration>) {
+        loop {
+            let upstreams = UpstreamsDashMap::new();
+
+            if let Some(consul) = config.consul.clone() {
+                let servers = consul.servers.unwrap_or_else(|| {
+                    vec![format!(
+                        "{}:{}",
+                        env::var("CONSUL_SERVICE_HOST").unwrap_or_else(|_| "0.0.0.0".to_string()),
+                        env::var("CONSUL_SERVICE_PORT").unwrap_or_else(|_| "0".to_string())
+                    )]
+                });
+                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 ss = format!("{}/v1/catalog/service/", consul_data);
+                let dc = format!("{}/v1/catalog/services?filter=%22aralez.service=yes%22+in+ServiceTags", consul_data);
+
+                if let Some(infolist) = httpclient::for_consul_list(dc.as_str(), consul.token.clone()).await {
+                    for (k, v) in infolist {
+                        let client_header_map: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
+                        let mut client_header_list = Vec::new();
+                        build_headers(&v.client_headers, config.as_ref(), &mut client_header_list);
+
+                        let sever_header_map: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
+                        let mut server_header_list = Vec::new();
+                        build_headers(&v.server_headers, config.as_ref(), &mut server_header_list);
+
+                        if !client_header_list.is_empty() {
+                            let path_key = v.path.clone();
+                            client_header_map.insert(Arc::from(path_key), client_header_list);
+                            config.client_headers.insert(Arc::from(v.host.clone()), client_header_map);
+                        }
+                        if !server_header_list.is_empty() {
+                            let path_key = v.path.clone();
+                            sever_header_map.insert(Arc::from(path_key), server_header_list);
+                            config.server_headers.insert(Arc::from(v.host.clone()), sever_header_map);
+                        }
+                        let pref = format!("{}{}", ss, k);
+                        let gsm = &GlobalServiceMapping {
+                            upstream: k.clone(),
+                            hostname: v.host.clone(),
+                            path: Some(v.path),
+                            to_https: v.to_https,
+                            redirect_to: v.redirect,
+                            authorization: v.auth,
+                            rate_limit: v.rate,
+                            x4xx_limit: v.xrate,
+                            client_headers: v.client_headers,
+                            server_headers: v.server_headers,
+                        };
+
+                        let list = httpclient::for_consul(pref.as_str(), consul.token.clone(), gsm).await;
+                        list_to_upstreams(list, &upstreams, gsm);
+                    }
+                }
+            }
+
+            if let Some(lt) = crate::utils::kuberconsul::clone_compare(&upstreams, &config).await {
+                let _ = toreturn.send(lt).await;
+            }
+
+            sleep(Duration::from_secs(5)).await;
+        }
+    }
+}

src/utils/discovery.rs

@@ -1,5 +1,7 @@
+use crate::utils::consul::ConsulDiscovery;
 use crate::utils::filewatch;
-use crate::utils::kuberconsul::{ConsulDiscovery, KubernetesDiscovery, ServiceDiscovery};
+use crate::utils::kuberconsul::ServiceDiscovery;
+use crate::utils::kubernetes::KubernetesDiscovery;
 use crate::utils::structs::{Configuration, UpstreamsDashMap};
 use crate::web::webserver;
 use async_trait::async_trait;

src/utils/healthcheck.rs

@@ -7,7 +7,7 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::net::TcpStream;
-use tokio::time::interval;
+use tokio::time::{interval, timeout};
 use tokio_native_tls::native_tls::TlsConnector;
 
 pub async fn hc2(upslist: Arc<UpstreamsDashMap>, fullist: Arc<UpstreamsDashMap>, idlist: Arc<UpstreamsIdMap>, params: (&str, u64)) {
@@ -86,12 +86,12 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsD
 
 async fn detect_tls(ip: &str, port: &u16, sni: &str) -> bool {
     let addr = format!("{ip}:{port}");
-
-    let stream = match TcpStream::connect(&addr).await {
-        Ok(s) => s,
+    let timeout_duration = Duration::from_secs(2); // THINK !!!
+    let stream = match timeout(timeout_duration, TcpStream::connect(&addr)).await {
+        Ok(Ok(s)) => s,
+        Ok(Err(_)) => return false,
         Err(_) => return false,
     };
-
     let connector = tokio_native_tls::TlsConnector::from(
         TlsConnector::builder()
             .danger_accept_invalid_certs(true)

src/utils/httpclient.rs

@@ -1,33 +1,95 @@
-use crate::utils::kuberconsul::{match_path, ConsulService, KubeEndpoints};
+use crate::utils::consul::ConsulService;
+use crate::utils::kuberconsul::match_path;
+use crate::utils::kubernetes::KubeEndpoints;
 use crate::utils::structs::{GlobalServiceMapping, InnerMap};
+use ahash::HashMap;
 use dashmap::DashMap;
 use pingora_core::connectors::http::Connector;
 use pingora_core::listeners::ALPN;
 use pingora_core::prelude::HttpPeer;
 use pingora_http::RequestHeader;
+use serde::{Deserialize, Serialize};
 use std::sync::atomic::AtomicUsize;
 use std::sync::{Arc, LazyLock};
 use std::time::Duration;
 
 pub static CONNECTOR: LazyLock<Connector> = LazyLock::new(|| Connector::new(None));
 
+#[derive(Debug, Default, Serialize, Deserialize, Clone)]
+pub struct ConsulServicesInternal {
+    #[serde(rename = "aralez.host")]
+    pub host: String,
+    #[serde(rename = "aralez.path")]
+    pub path: String,
+    #[serde(rename = "aralez.auth")]
+    pub auth: Option<String>,
+    #[serde(rename = "aralez.redirect")]
+    pub redirect: Option<String>,
+    #[serde(rename = "aralez.rate")]
+    pub rate: Option<isize>,
+    #[serde(rename = "aralez.4xx_rate")]
+    pub xrate: Option<u32>,
+    #[serde(rename = "aralez.client_headers")]
+    pub client_headers: Option<Vec<String>>,
+    #[serde(rename = "aralez.server_headers")]
+    pub server_headers: Option<Vec<String>>,
+    #[serde(rename = "aralez.to_https")]
+    pub to_https: Option<bool>,
+}
+pub type ConsulServices = HashMap<String, ConsulServicesInternal>;
+pub async fn for_consul_list(url: &str, token: Option<String>) -> Option<ConsulServices> {
+    if let Some(data) = getfromapi(url, token, "consul").await {
+        let yo = parse_services(data);
+        if let Ok(y) = yo {
+            return Some(y);
+        }
+        return None;
+    }
+    None
+}
+
+fn parse_services(json: Vec<u8>) -> Result<ConsulServices, serde_json::Error> {
+    let raw: HashMap<String, Vec<String>> = serde_json::from_slice(&json)?;
+    Ok(raw
+        .into_iter()
+        .map(|(service, tags)| {
+            let mut internal = ConsulServicesInternal::default();
+            for tag in tags {
+                if let Some((key, value)) = tag.split_once('=') {
+                    match key {
+                        "aralez.host" => internal.host = value.to_string(),
+                        "aralez.path" => internal.path = value.to_string(),
+                        "aralez.rate" => internal.rate = value.parse::<isize>().ok(),
+                        "aralez.4xx_rate" => internal.xrate = value.parse::<u32>().ok(),
+                        "aralez.to_https" => internal.to_https = value.parse::<bool>().ok(),
+                        "aralez.auth" => internal.auth = Option::from(value.to_string()),
+                        "aralez.redirect" => internal.redirect = Option::from(value.to_string()),
+                        "aralez.client_header" => internal.client_headers.get_or_insert_default().push(value.to_string()),
+                        "aralez.server_header" => internal.server_headers.get_or_insert_default().push(value.to_string()),
+                        _ => {}
+                    }
+                }
+            }
+
+            (service, internal)
+        })
+        .collect())
+}
 pub async fn for_consul(url: &str, token: Option<String>, conf: &GlobalServiceMapping) -> Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>> {
     if let Some(data) = getfromapi(url, token, "consul").await {
         let endpoints: Vec<ConsulService> = serde_json::from_slice(&data).ok()?;
         let mut inner_vec = Vec::new();
         let upstreams: DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)> = DashMap::new();
         for subsets in endpoints {
-            let addr = subsets.tagged_addresses.get("lan_ipv4").unwrap().address.clone();
-            let prt = subsets.tagged_addresses.get("lan_ipv4").unwrap().port;
             let to_add = Arc::from(InnerMap {
-                address: Arc::from(&*addr),
-                port: prt,
+                address: Arc::from(&*subsets.address),
+                port: subsets.port,
                 is_ssl: false,
                 is_http2: false,
                 to_https: conf.to_https.unwrap_or(false),
                 rate_limit: conf.rate_limit,
                 x4xx_limit: conf.x4xx_limit,
-                redirect_to: None,
+                redirect_to: conf.redirect_to.clone().map(Arc::<str>::from),
                 healthcheck: None,
                 authorization: None,
             });
@@ -164,3 +226,10 @@ fn parse_url(url: &str) -> Result<(&str, u16, &str, bool), &'static str> {
 
     Ok((host, port, uri, is_https))
 }
+
+// fn base64_decode(encoded: &str) -> Option<String> {
+//     let decoded_bytes = general_purpose::STANDARD.decode(encoded).unwrap_or_default();
+//     let decoded: Result<HashMap<String, String>, serde_json::Error> = serde_json::from_slice(&decoded_bytes);
+//     println!("Decoded: {:?}", decoded);
+//     None
+// }

src/utils/kuberconsul.rs

@@ -1,216 +1,50 @@
-use crate::utils::httpclient;
-use crate::utils::parceyaml::build_headers;
 use crate::utils::structs::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap};
 use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams};
 use async_trait::async_trait;
 use dashmap::DashMap;
-use pingora::prelude::sleep;
-use rand::RngExt;
-use serde::Deserialize;
-use std::collections::HashMap;
-use std::env;
-use std::fs;
-use std::path::Path;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
-use std::time::Duration;
 use tokio::fs::File;
 use tokio::io::AsyncReadExt;
 use tokio::sync::mpsc::Sender;
 
-#[derive(Debug, serde::Deserialize)]
-pub struct KubeEndpoints {
-    pub subsets: Option<Vec<KubeSubset>>,
-}
-#[derive(Debug, serde::Deserialize)]
-pub struct KubeSubset {
-    pub addresses: Option<Vec<KubeAddress>>,
-    pub ports: Option<Vec<KubePort>>,
-}
-
-#[derive(Debug, serde::Deserialize)]
-pub struct KubeAddress {
-    pub ip: String,
-}
-
-#[derive(Debug, serde::Deserialize)]
-pub struct KubePort {
-    pub port: u16,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct ConsulService {
-    #[serde(rename = "ServiceTaggedAddresses")]
-    pub tagged_addresses: HashMap<String, ConsulTaggedAddress>,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct ConsulTaggedAddress {
-    #[serde(rename = "Address")]
-    pub address: String,
-    #[serde(rename = "Port")]
-    pub port: u16,
-}
 #[allow(clippy::type_complexity)]
 pub fn list_to_upstreams(lt: Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>>, upstreams: &UpstreamsDashMap, i: &GlobalServiceMapping) {
     if let Some(list) = lt {
-        match upstreams.get(&*i.hostname.clone()) {
+        let key: Arc<str> = Arc::from(i.hostname.as_str());
+        match upstreams.get(&key) {
             Some(upstr) => {
                 for (k, v) in list {
-                    upstr.value().insert(k.to_owned(), v);
+                    upstr.value().insert(k, v);
                 }
             }
             None => {
-                upstreams.insert(Arc::from(i.hostname.clone()), list);
+                upstreams.insert(key, list);
             }
         };
     }
 }
 
 pub fn match_path(conf: &GlobalServiceMapping, upstreams: &DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>, values: Vec<Arc<InnerMap>>) {
-    match conf.path {
-        Some(ref p) => {
-            upstreams.insert(Arc::from(p.clone()), (values, AtomicUsize::new(0)));
-        }
-        None => {
-            upstreams.insert(Arc::from("/"), (values, AtomicUsize::new(0)));
-        }
-    }
+    let path = conf.path.as_deref().unwrap_or("/");
+    upstreams.insert(Arc::from(path), (values, AtomicUsize::new(0)));
 }
 
-async fn read_token(path: &str) -> String {
+pub async fn read_token(path: &str) -> String {
     let mut file = File::open(path).await.unwrap();
     let mut contents = String::new();
     file.read_to_string(&mut contents).await.unwrap();
     contents.trim().to_string()
 }
+
 #[async_trait]
 pub trait ServiceDiscovery {
     async fn fetch_upstreams(&self, config: Arc<Configuration>, toreturn: Sender<Configuration>);
 }
 
-pub struct KubernetesDiscovery;
-pub struct ConsulDiscovery;
-
-#[async_trait]
-impl ServiceDiscovery for KubernetesDiscovery {
-    async fn fetch_upstreams(&self, config: Arc<Configuration>, toreturn: Sender<Configuration>) {
-        let prev_upstreams = UpstreamsDashMap::new();
-
-        if let Some(kuber) = config.kubernetes.clone() {
-            let servers = kuber.servers.unwrap_or(vec![format!(
-                "{}:{}",
-                env::var("KUBERNETES_SERVICE_HOST").unwrap_or("0.0.0.0".to_string()),
-                env::var("KUBERNETES_SERVICE_PORT_HTTPS").unwrap_or("0".to_string())
-            )]);
-
-            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 path = kuber.tokenpath.unwrap_or("/var/run/secrets/kubernetes.io/serviceaccount/token".to_string());
-            let namespace = get_current_namespace().unwrap_or_else(|| "default".to_string());
-            let token = read_token(path.as_str()).await;
-            loop {
-                let upstreams = UpstreamsDashMap::new();
-                if let Some(kuber) = config.kubernetes.clone() {
-                    if let Some(svc) = kuber.services {
-                        for service in svc {
-                            let header_list: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
-                            let mut hl = Vec::new();
-                            build_headers(&service.client_headers, config.as_ref(), &mut hl);
-                            if !hl.is_empty() {
-                                match service.path.clone() {
-                                    Some(path) => {
-                                        header_list.insert(Arc::from(path.as_str()), hl);
-                                    }
-                                    None => {
-                                        header_list.insert(Arc::from("/"), hl);
-                                    }
-                                }
-
-                                // header_list.insert(Arc::from(path.as_str()), hl);
-                                // header_list.insert(Arc::from(i.path).unwrap_or(Arc::from("/")).as_str(), hl);
-                                config.client_headers.insert(Arc::from(service.hostname.clone()), header_list);
-                            }
-                            let url = format!("https://{}/api/v1/namespaces/{}/endpoints/{}", server, namespace, service.hostname);
-                            // let url = format!("https://{}/api/v1/namespaces/{}/endpoints?labelSelector=app", server, namespace);
-                            let list = httpclient::for_kuber(&url, &token, &service).await;
-                            // println!("{:?}", list);
-                            list_to_upstreams(list, &upstreams, &service);
-                        }
-                    }
-                    if let Some(lt) = clone_compare(&upstreams, &prev_upstreams, &config).await {
-                        toreturn.send(lt).await.unwrap();
-                    }
-                }
-                sleep(Duration::from_secs(5)).await;
-            }
-        }
-    }
-}
-
-fn get_current_namespace() -> Option<String> {
-    let ns_path = "/var/run/secrets/kubernetes.io/serviceaccount/namespace";
-    if Path::new(ns_path).exists() {
-        if let Ok(contents) = fs::read_to_string(ns_path) {
-            return Some(contents.trim().to_string());
-        }
-    }
-    std::env::var("POD_NAMESPACE").ok()
-}
-
-#[async_trait]
-impl ServiceDiscovery for ConsulDiscovery {
-    async fn fetch_upstreams(&self, config: Arc<Configuration>, toreturn: Sender<Configuration>) {
-        let prev_upstreams = UpstreamsDashMap::new();
-        loop {
-            let upstreams = UpstreamsDashMap::new();
-
-            if let Some(consul) = config.consul.clone() {
-                let servers = consul.servers.unwrap_or(vec![format!(
-                    "{}:{}",
-                    env::var("CONSUL_SERVICE_HOST").unwrap_or("0.0.0.0".to_string()),
-                    env::var("CONSUL_SERVICE_PORT").unwrap_or("0".to_string())
-                )]);
-                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 ss = consul_data + "/v1/catalog/service/";
-
-                if let Some(svc) = consul.services {
-                    for i in svc {
-                        let header_list = DashMap::new();
-                        let mut hl = Vec::new();
-                        build_headers(&i.client_headers, config.as_ref(), &mut hl);
-                        if !hl.is_empty() {
-                            match i.path.clone() {
-                                Some(path) => {
-                                    header_list.insert(Arc::from(path.as_str()), hl);
-                                }
-                                None => {
-                                    header_list.insert(Arc::from("/"), hl);
-                                }
-                            }
-                            // header_list.insert(i.path.clone().unwrap_or("/".to_string()), hl);
-                            config.client_headers.insert(Arc::from(i.hostname.clone()), header_list);
-                        }
-
-                        let pref = ss.clone() + &i.upstream;
-                        let list = httpclient::for_consul(pref.as_str(), consul.token.clone(), &i).await;
-                        list_to_upstreams(list, &upstreams, &i);
-                    }
-                }
-            }
-            if let Some(lt) = clone_compare(&upstreams, &prev_upstreams, &config).await {
-                toreturn.send(lt).await.unwrap();
-            }
-            sleep(Duration::from_secs(5)).await;
-        }
-    }
-}
-async fn clone_compare(upstreams: &UpstreamsDashMap, prev_upstreams: &UpstreamsDashMap, config: &Arc<Configuration>) -> Option<Configuration> {
-    if !compare_dashmaps(upstreams, prev_upstreams) {
-        let tosend: Configuration = Configuration {
+pub async fn clone_compare(upstreams: &UpstreamsDashMap, config: &Arc<Configuration>) -> Option<Configuration> {
+    if !compare_dashmaps(upstreams, &config.upstreams) {
+        let tosend = Configuration {
             upstreams: Default::default(),
             client_headers: config.client_headers.clone(),
             server_headers: config.server_headers.clone(),
@@ -219,10 +53,11 @@ async fn clone_compare(upstreams: &UpstreamsDashMap, prev_upstreams: &UpstreamsD
             typecfg: config.typecfg.clone(),
             extraparams: config.extraparams.clone(),
         };
-        clone_dashmap_into(upstreams, prev_upstreams);
+        clone_dashmap_into(upstreams, &config.upstreams);
         clone_dashmap_into(upstreams, &tosend.upstreams);
+
         print_upstreams(&tosend.upstreams, &tosend.extraparams);
         return Some(tosend);
-    };
+    }
     None
 }

src/utils/kubernetes.rs

@@ -0,0 +1,157 @@
+use crate::utils::httpclient;
+use crate::utils::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate};
+use crate::utils::parceyaml::build_headers;
+use crate::utils::structs::{Configuration, GlobalServiceMapping, UpstreamsDashMap};
+use async_trait::async_trait;
+use dashmap::DashMap;
+use log::error;
+use rand::RngExt;
+use std::env;
+use std::fs;
+use std::path::Path;
+use std::sync::Arc;
+
+use crate::utils::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery};
+use rustls::crypto::ring::default_provider;
+use rustls::crypto::CryptoProvider;
+use tokio::sync::mpsc::Sender;
+
+#[derive(Debug, serde::Deserialize)]
+pub struct KubeEndpoints {
+    pub subsets: Option<Vec<KubeSubset>>,
+}
+
+#[derive(Debug, serde::Deserialize)]
+pub struct KubeSubset {
+    pub addresses: Option<Vec<KubeAddress>>,
+    pub ports: Option<Vec<KubePort>>,
+}
+
+#[derive(Debug, serde::Deserialize)]
+pub struct KubeAddress {
+    pub ip: String,
+}
+
+#[derive(Debug, serde::Deserialize)]
+pub struct KubePort {
+    pub port: u16,
+}
+pub struct KubernetesDiscovery;
+
+#[async_trait]
+impl ServiceDiscovery for KubernetesDiscovery {
+    async fn fetch_upstreams(&self, config: Arc<Configuration>, toreturn: Sender<Configuration>) {
+        if let Some(kuber) = config.kubernetes.clone() {
+            let servers = kuber.servers.unwrap_or_else(|| {
+                vec![format!(
+                    "{}:{}",
+                    env::var("KUBERNETES_SERVICE_HOST").unwrap_or_else(|_| "0.0.0.0".to_string()),
+                    env::var("KUBERNETES_SERVICE_PORT_HTTPS").unwrap_or_else(|_| "0".to_string())
+                )]
+            });
+
+            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 path = kuber.tokenpath.unwrap_or_else(|| "/var/run/secrets/kubernetes.io/serviceaccount/token".to_string());
+            let namespace = get_current_namespace().unwrap_or_else(|| "default".to_string());
+            let token = crate::utils::kuberconsul::read_token(path.as_str()).await;
+
+            if CryptoProvider::get_default().is_none() {
+                default_provider().install_default().expect("Failed to install rustls crypto provider");
+            }
+
+            let (tx, mut rx) = tokio::sync::mpsc::channel::<AralezRouteUpdate>(100);
+            let srv = server.clone();
+            let tok = token.clone();
+            tokio::spawn(async move {
+                if let Err(e) = start_ingress_watcher_with_config(srv.as_str(), tok.as_str(), tx, "aralez").await {
+                    error!("Fatal watcher error: {:?}", e);
+                }
+            });
+
+            let upstreams = UpstreamsDashMap::new();
+
+            while let Some(update) = rx.recv().await {
+                let host_key: Arc<str> = Arc::from(update.host.as_str());
+
+                if update.is_deleted {
+                    upstreams.remove(&host_key);
+                    config.client_headers.remove(&host_key);
+                } else {
+                    let service = GlobalServiceMapping {
+                        upstream: update.service_name.clone(),
+                        hostname: update.host.clone(),
+                        path: Some(update.path.clone()),
+                        to_https: None,
+                        redirect_to: None,
+                        authorization: None,
+                        // sticky_sessions: update.sticky_sessions,
+                        rate_limit: update.rate_limit,
+                        x4xx_limit: update.x4xx_limit,
+                        client_headers: None,
+                        server_headers: None,
+                    };
+
+                    let cheader_list: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
+                    let sheader_list: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
+
+                    let mut client_headers_list = Vec::new();
+                    let mut server_headers_list = Vec::new();
+
+                    let mut chl: Option<Vec<String>> = None;
+                    let mut shl: Option<Vec<String>> = None;
+
+                    if let Some(hdr) = service.client_headers.as_ref() {
+                        chl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned());
+                    }
+                    if let Some(hdr) = service.server_headers.as_ref() {
+                        shl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned());
+                    }
+
+                    if let Some(ch) = update.client_headers {
+                        chl.get_or_insert_with(Vec::new).extend(ch);
+                    }
+                    if let Some(hdr) = update.server_headers.as_ref() {
+                        shl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned());
+                    }
+
+                    build_headers(&chl, config.as_ref(), &mut client_headers_list);
+                    if !client_headers_list.is_empty() {
+                        let path_key = service.path.as_deref().unwrap_or("/");
+                        cheader_list.insert(Arc::from(path_key), client_headers_list);
+                        config.client_headers.insert(host_key.clone(), cheader_list);
+                    }
+
+                    build_headers(&shl, config.as_ref(), &mut server_headers_list);
+                    if !server_headers_list.is_empty() {
+                        let path_key = service.path.as_deref().unwrap_or("/");
+                        sheader_list.insert(Arc::from(path_key), server_headers_list);
+                        config.server_headers.insert(host_key.clone(), sheader_list);
+                    }
+                    let url = format!("https://{}/api/v1/namespaces/{}/endpoints/{}", server, namespace, update.service_name);
+                    let list = httpclient::for_kuber(&url, &token, &service).await;
+
+                    if list.is_none() {
+                        error!("Endpoint URL: {} returned empty/invalid response", url);
+                    }
+
+                    list_to_upstreams(list, &upstreams, &service);
+                }
+                if let Some(lt) = clone_compare(&upstreams, &config).await {
+                    let _ = toreturn.send(lt).await;
+                }
+            }
+        }
+    }
+}
+
+fn get_current_namespace() -> Option<String> {
+    let ns_path = "/var/run/secrets/kubernetes.io/serviceaccount/namespace";
+    if Path::new(ns_path).exists() {
+        if let Ok(contents) = fs::read_to_string(ns_path) {
+            return Some(contents.trim().to_string());
+        }
+    }
+    env::var("POD_NAMESPACE").ok()
+}

src/utils/kubewatcher.rs

@@ -0,0 +1,271 @@
+use futures::StreamExt;
+use hyper::Uri;
+use k8s_openapi::api::core::v1::{Node, Service};
+use k8s_openapi::api::networking::v1::Ingress;
+use kube::api::{Patch, PatchParams};
+use kube::{
+    api::Api,
+    client::Client,
+    config::{AuthInfo, Config},
+    runtime::watcher::{self, Event},
+};
+use log::{error, info};
+use serde_json::json;
+use std::time::Duration;
+use tokio::sync::mpsc;
+
+pub struct AralezRouteUpdate {
+    pub host: String,
+    pub path: String,
+    pub service_name: String,
+    pub is_deleted: bool,
+    pub rate_limit: Option<isize>,
+    pub x4xx_limit: Option<u32>,
+    pub sticky_sessions: Option<u64>,
+    pub client_headers: Option<Vec<String>>,
+    pub server_headers: Option<Vec<String>>,
+}
+
+#[derive(Debug, Clone)]
+struct AralezAnnotation {
+    rate_limit: Option<isize>,
+    x4xx_limit: Option<u32>,
+    sticky_sessions: Option<u64>,
+    pub client_headers: Option<Vec<String>>,
+    pub server_headers: Option<Vec<String>>,
+}
+
+pub async fn start_ingress_watcher_with_config(
+    api_server: &str,
+    token: &str,
+    tx: mpsc::Sender<AralezRouteUpdate>,
+    target_class: &'static str,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let client = create_custom_k8s_client(api_server, token, true).await?;
+    let ingresses: Api<Ingress> = Api::all(client.clone());
+
+    let watcher_config = watcher::Config::default().any_semantic();
+    let mut stream = watcher::watcher(ingresses, watcher_config).boxed();
+
+    while let Some(event) = stream.next().await {
+        match event {
+            Ok(Event::Apply(ing)) => {
+                process_ingress_change(&client, &ing, target_class, false, &tx).await;
+            }
+            Ok(Event::Delete(ing)) => {
+                process_ingress_change(&client, &ing, target_class, true, &tx).await;
+            }
+            Ok(Event::InitApply(ing)) => {
+                process_ingress_change(&client, &ing, target_class, false, &tx).await;
+            }
+            Ok(Event::InitDone) => {
+                info!("Ingress watcher initial sync complete");
+            }
+            Err(err) => {
+                error!("Kubernetes Ingress watch error: {:?}", err);
+            }
+            _ => {}
+        }
+    }
+    Ok(())
+}
+
+async fn process_ingress_change(client: &Client, ing: &Ingress, target_class: &str, is_deleted: bool, tx: &mpsc::Sender<AralezRouteUpdate>) {
+    let spec = match &ing.spec {
+        Some(s) => s,
+        None => return,
+    };
+    let class_name = spec.ingress_class_name.as_deref().unwrap_or_default();
+    if class_name != target_class {
+        return;
+    }
+
+    if !is_deleted {
+        if let (Some(name), Some(ns)) = (ing.metadata.name.as_deref(), ing.metadata.namespace.as_deref()) {
+            let svc_name = std::env::var("ARALEZ_SERVICE_NAME").unwrap_or_else(|_| "aralez-service".to_string());
+            let external_ip = match discover_controller_ip(&client, ns, &svc_name).await {
+                Some(ip) => ip,
+                None => discover_node_ip(&client)
+                    .await
+                    .unwrap_or_else(|| std::env::var("ARALEZ_EXTERNAL_IP").unwrap_or_else(|_| "127.0.0.1".to_string())),
+            };
+            let client_clone = client.clone();
+            let name_clone = name.to_string();
+            let ns_clone = ns.to_string();
+            let status = ing.status.clone();
+            tokio::spawn(async move {
+                patch_ingress_status_address(&client_clone, &ns_clone, &name_clone, status.as_ref(), &external_ip).await;
+            });
+        }
+    }
+
+    if let Some(rules) = &spec.rules {
+        for rule in rules {
+            let host = match &rule.host {
+                Some(h) => h.clone(),
+                None => "*".to_string(),
+            };
+
+            if let Some(http) = &rule.http {
+                for path_entry in &http.paths {
+                    let path = path_entry.path.clone().unwrap_or_else(|| "/".to_string());
+                    if let Some(service) = &path_entry.backend.service {
+                        let service_name = service.name.clone();
+                        let mut update = AralezRouteUpdate {
+                            host: host.clone(),
+                            path,
+                            service_name,
+                            is_deleted,
+                            rate_limit: None,
+                            x4xx_limit: None,
+                            sticky_sessions: None,
+                            client_headers: None,
+                            server_headers: None,
+                        };
+                        if let Some(annot) = parse_ingress_config(ing) {
+                            update.rate_limit = annot.rate_limit;
+                            update.x4xx_limit = annot.x4xx_limit;
+                            update.sticky_sessions = annot.sticky_sessions;
+                            update.client_headers = annot.client_headers;
+                            update.server_headers = annot.server_headers;
+                        }
+                        if let Err(e) = tx.send(update).await {
+                            error!("Failed to send route update to channel: {:?}", e);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+pub async fn create_custom_k8s_client(api_server_url: &str, bearer_token: &str, insecure_tls: bool) -> Result<Client, Box<dyn std::error::Error>> {
+    let full_url = if !api_server_url.starts_with("http://") && !api_server_url.starts_with("https://") {
+        format!("https://{}", api_server_url)
+    } else {
+        api_server_url.to_string()
+    };
+    let uri: Uri = full_url.parse()?;
+    let mut config = Config::new(uri);
+    config.auth_info = AuthInfo {
+        token: Some(bearer_token.trim().to_string().into()),
+        ..Default::default()
+    };
+    config.connect_timeout = Some(Duration::from_secs(10));
+    config.read_timeout = None;
+    if insecure_tls {
+        config.accept_invalid_certs = true;
+    }
+    let client = Client::try_from(config)?;
+    Ok(client)
+}
+
+fn parse_ingress_config(ing: &Ingress) -> Option<AralezAnnotation> {
+    if let Some(annotations) = &ing.metadata.annotations {
+        let mut clh: Option<Vec<String>> = None;
+        let mut srh: Option<Vec<String>> = None;
+
+        if let Some(ch) = annotations.get("aralez.rs/client_headers") {
+            let data: Vec<String> = serde_json::from_str(ch).unwrap_or_default();
+            clh = Some(data);
+        }
+        if let Some(sh) = annotations.get("aralez.rs/server_headers") {
+            let data: Vec<String> = serde_json::from_str(sh).unwrap_or_default();
+            srh = Some(data);
+        };
+
+        let annots = AralezAnnotation {
+            rate_limit: annotations.get("aralez.rs/rate_limit").and_then(|v| v.parse().ok()),
+            x4xx_limit: annotations.get("aralez.rs/x4xx_limit").and_then(|v| v.parse().ok()),
+            sticky_sessions: annotations.get("aralez.rs/sticky_sessions").and_then(|v| v.parse().ok()),
+            client_headers: clh,
+            server_headers: srh,
+        };
+        return Some(annots);
+    }
+    None
+}
+
+pub async fn patch_ingress_status_address(
+    client: &Client,
+    namespace: &str,
+    name: &str,
+    current_status: Option<&k8s_openapi::api::networking::v1::IngressStatus>,
+    external_ip: &str,
+) {
+    if let Some(status) = current_status {
+        if let Some(lb) = &status.load_balancer {
+            if let Some(ing_list) = &lb.ingress {
+                if ing_list.iter().any(|i| i.ip.as_deref() == Some(external_ip)) {
+                    return;
+                }
+            }
+        }
+    }
+
+    let ingresses: Api<Ingress> = Api::namespaced(client.clone(), namespace);
+
+    let patch = json!({
+        "apiVersion": "networking.k8s.io/v1",
+        "kind": "Ingress",
+        "status": {
+            "loadBalancer": {
+                "ingress": [
+                    {
+                        "ip": external_ip
+                    }
+                ]
+            }
+        }
+    });
+
+    let params = PatchParams::apply("aralez-ingress-controller").force();
+
+    match ingresses.patch_status(name, &params, &Patch::Apply(patch)).await {
+        Ok(_) => info!("Patched Ingress {}/{} ADDRESS to {}", namespace, name, external_ip),
+        Err(e) => error!("Failed to patch status for Ingress {}/{}: {:?}", namespace, name, e),
+    }
+}
+
+pub async fn discover_node_ip(client: &Client) -> Option<String> {
+    let nodes: Api<Node> = Api::all(client.clone());
+    if let Ok(node_list) = nodes.list(&Default::default()).await {
+        for node in node_list {
+            if let Some(status) = node.status {
+                if let Some(addresses) = status.addresses {
+                    for addr in &addresses {
+                        if addr.type_ == "ExternalIP" {
+                            return Some(addr.address.clone());
+                        }
+                    }
+                    for addr in &addresses {
+                        if addr.type_ == "InternalIP" {
+                            return Some(addr.address.clone());
+                        }
+                    }
+                }
+            }
+        }
+    }
+    None
+}
+pub async fn discover_controller_ip(client: &Client, namespace: &str, service_name: &str) -> Option<String> {
+    let services: Api<Service> = Api::namespaced(client.clone(), namespace);
+    if let Ok(svc) = services.get(service_name).await {
+        if let Some(status) = svc.status {
+            if let Some(lb) = status.load_balancer {
+                if let Some(ingress_list) = lb.ingress {
+                    for ing in ingress_list {
+                        if let Some(ip) = ing.ip {
+                            return Some(ip);
+                        }
+                        if let Some(hostname) = ing.hostname {
+                            return Some(hostname);
+                        }
+                    }
+                }
+            }
+        }
+    }
+    None
+}

src/utils/metrics.rs

@@ -47,28 +47,83 @@ pub static REQUESTS_BY_UPSTREAM: LazyLock<IntCounterVec> =
 
 pub static REQUESTS_BY_VERSION: LazyLock<IntCounterVec> =
     LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_version_total", "Number of requests by HTTP versions", &["version"]).unwrap());
+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"]));
+pub static HTTP_10_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/1.0"]));
+pub static HTTP_UNK_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["Unknown"]));
+
+pub static METHOD_GET: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_METHOD.with_label_values(&["GET"]));
+pub static METHOD_POST: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_METHOD.with_label_values(&["POST"]));
+pub static METHOD_PUT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_METHOD.with_label_values(&["PUT"]));
+pub static METHOD_DELETE: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_METHOD.with_label_values(&["DELETE"]));
+pub static METHOD_HEAD: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_METHOD.with_label_values(&["HEAD"]));
+pub static METHOD_OPTIONS: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_METHOD.with_label_values(&["OPTIONS"]));
+pub static METHOD_PATCH: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_METHOD.with_label_values(&["PATCH"]));
+pub static METHOD_OTHER: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_METHOD.with_label_values(&["OTHER"]));
+
+pub static STATUS_100: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["100"]));
+pub static STATUS_101: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["101"]));
+pub static STATUS_200: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["200"]));
+pub static STATUS_204: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["204"]));
+pub static STATUS_301: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["301"]));
+pub static STATUS_302: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["302"]));
+pub static STATUS_304: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["304"]));
+pub static STATUS_400: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["400"]));
+pub static STATUS_403: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["403"]));
+pub static STATUS_404: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["404"]));
+pub static STATUS_500: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["500"]));
+pub static STATUS_501: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["501"]));
+pub static STATUS_502: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["502"]));
+pub static STATUS_503: LazyLock<IntCounter> = LazyLock::new(|| RESPONSE_CODES.with_label_values(&["503"]));
 
 pub fn calc_metrics(metric_types: &MetricTypes) {
     REQUEST_COUNT.inc();
-    let version_str = match metric_types.version {
-        Version::HTTP_11 => "HTTP/1.1",
-        Version::HTTP_2 => "HTTP/2.0",
-        Version::HTTP_3 => "HTTP/3.0",
-        Version::HTTP_10 => "HTTP/1.0",
-        _ => "Unknown",
+    match metric_types.version {
+        Version::HTTP_11 => HTTP_11_COUNT.inc(),
+        Version::HTTP_2 => HTTP_20_COUNT.inc(),
+        Version::HTTP_3 => HTTP_30_COUNT.inc(),
+        Version::HTTP_10 => HTTP_10_COUNT.inc(),
+        _ => HTTP_UNK_COUNT.inc(),
     };
-    REQUESTS_BY_VERSION.with_label_values(&[version_str]).inc();
-    RESPONSE_CODES.with_label_values(&[metric_types.code.unwrap_or(StatusCode::GONE).as_str()]).inc();
-    REQUESTS_BY_METHOD.with_label_values(&[metric_types.method.as_str()]).inc();
+    match metric_types.method.as_str() {
+        "GET" => METHOD_GET.inc(),
+        "POST" => METHOD_POST.inc(),
+        "PUT" => METHOD_PUT.inc(),
+        "DELETE" => METHOD_DELETE.inc(),
+        "HEAD" => METHOD_HEAD.inc(),
+        "OPTIONS" => METHOD_OPTIONS.inc(),
+        "PATCH" => METHOD_PATCH.inc(),
+        _ => METHOD_OTHER.inc(),
+    }
+    let code = metric_types.code.unwrap_or(StatusCode::GONE);
+    match code.as_u16() {
+        100 => STATUS_100.inc(),
+        101 => STATUS_101.inc(),
+        200 => STATUS_200.inc(),
+        204 => STATUS_204.inc(),
+        301 => STATUS_301.inc(),
+        302 => STATUS_302.inc(),
+        304 => STATUS_304.inc(),
+        400 => STATUS_400.inc(),
+        403 => STATUS_403.inc(),
+        404 => STATUS_404.inc(),
+        500 => STATUS_500.inc(),
+        501 => STATUS_501.inc(),
+        502 => STATUS_502.inc(),
+        503 => STATUS_503.inc(),
+        _ => RESPONSE_CODES.with_label_values(&[code.as_str()]).inc(), // Fallback for rare status codes
+    }
+
     REQUESTS_BY_UPSTREAM.with_label_values(&[metric_types.upstream.as_ref()]).inc();
     RESPONSE_LATENCY.observe(metric_types.latency.as_secs_f64());
 
-    if let Some(eviction) = EVICTION.get() {
-        CACHE_SIZE_BYTES.set(eviction.total_size() as i64);
-        CACHE_ITEMS.set(eviction.total_items() as i64);
-        CACHE_EVICTED_BYTES.set(eviction.evicted_size() as i64);
-        CACHE_EVICTED_ITEMS.set(eviction.evicted_items() as i64);
-    }
+    // if let Some(eviction) = EVICTION.get() {
+    //     CACHE_SIZE_BYTES.set(eviction.total_size() as i64);
+    //     CACHE_ITEMS.set(eviction.total_items() as i64);
+    //     CACHE_EVICTED_BYTES.set(eviction.evicted_size() as i64);
+    //     CACHE_EVICTED_ITEMS.set(eviction.evicted_items() as i64);
+    // }
 }
 
 pub(crate) fn get_memory_usage() -> usize {
@@ -87,3 +142,15 @@ pub(crate) fn get_memory_usage() -> usize {
 pub fn get_open_files() -> usize {
     std::fs::read_dir("/proc/self/fd").map(|dir| dir.count()).unwrap_or(0)
 }
+
+pub async fn calc_cache_metrics() {
+    loop {
+        if let Some(eviction) = EVICTION.get() {
+            CACHE_SIZE_BYTES.set(eviction.total_size() as i64);
+            CACHE_ITEMS.set(eviction.total_items() as i64);
+            CACHE_EVICTED_BYTES.set(eviction.evicted_size() as i64);
+            CACHE_EVICTED_ITEMS.set(eviction.evicted_items() as i64);
+        }
+        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+    }
+}

src/utils/state.rs

@@ -16,14 +16,3 @@ pub fn is_first_run() -> bool {
     let state = GLOBAL_STATE.read().unwrap();
     state.first_run
 }
-
-/*
-impl SharedState {
-    pub fn mark_first_run(&mut self) {
-        self.first_run = false;
-    }
-    pub fn is_first_run(&self) -> bool {
-        self.first_run
-    }
-}
-*/

src/utils/structs.rs

@@ -20,11 +20,12 @@ pub struct Extraparams {
 
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
 pub struct GlobalServiceMapping {
+    pub redirect_to: Option<String>,
+    pub authorization: Option<String>,
     pub upstream: String,
     pub hostname: String,
     pub path: Option<String>,
     pub to_https: Option<bool>,
-    pub sticky_sessions: Option<u64>,
     pub rate_limit: Option<isize>,
     pub x4xx_limit: Option<u32>,
     pub client_headers: Option<Vec<String>>,
@@ -34,14 +35,12 @@ pub struct GlobalServiceMapping {
 #[derive(Clone, Default, Debug, Serialize, Deserialize)]
 pub struct Kubernetes {
     pub servers: Option<Vec<String>>,
-    pub services: Option<Vec<GlobalServiceMapping>>,
     pub tokenpath: Option<String>,
 }
 
 #[derive(Clone, Default, Debug, Serialize, Deserialize)]
 pub struct Consul {
     pub servers: Option<Vec<String>>,
-    pub services: Option<Vec<GlobalServiceMapping>>,
     pub token: Option<String>,
 }
 #[derive(Debug, Default, Serialize, Deserialize)]

src/web/bgservice.rs

@@ -1,5 +1,6 @@
 use crate::tls::acme::order::refresh_order;
 use crate::utils::discovery::{APIUpstreamProvider, ConsulProvider, Discovery, FromFileProvider, KubernetesProvider};
+use crate::utils::metrics::calc_cache_metrics;
 use crate::utils::parceyaml::load_configuration;
 use crate::utils::structs::Configuration;
 use crate::utils::tools::*;
@@ -76,6 +77,7 @@ impl BackgroundService for LB {
         drop(tokio::spawn(async move {
             healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await
         }));
+        drop(tokio::spawn(async move { calc_cache_metrics().await }));
         drop(tokio::spawn(async move { refresh_order(certdir, confdir).await }));
         init_logging(self.config.access_log.clone());
         loop {

src/web/proxyhttp.rs

@@ -310,15 +310,16 @@ impl ProxyHttp for LB {
                 let _ = _upstream_response.append_header("set-cookie", buf.as_str());
             }
         }
-
+        Ok(())
+    }
+    async fn upstream_response_filter(&self, _session: &mut Session, upstream_response: &mut ResponseHeader, ctx: &mut Self::CTX) -> Result<()> {
         if let Some(client_headers) = &ctx.client_headers {
             for (k, v) in client_headers.iter() {
-                _upstream_response.append_header(k.clone(), v.as_ref())?;
+                upstream_response.append_header(k.clone(), v.as_ref())?;
             }
         }
         Ok(())
     }
-
     async fn logging(&self, session: &mut Session, _e: Option<&pingora::Error>, ctx: &mut Self::CTX) {
         let response_code = session.response_written().map_or(0, |resp| resp.status.as_u16());
         let m = &MetricTypes {

src/web/start.rs

@@ -1,4 +1,3 @@
-// use rustls::crypto::ring::default_provider;
 use crate::tls::grades;
 use crate::tls::load;
 use crate::tls::load::CertificateConfig;
@@ -18,6 +17,7 @@ use pingora_core::prelude::{background_service, Opt};
 use pingora_core::protocols::TcpKeepalive;
 use pingora_core::server::Server;
 use privdrop::reexports::libc::SIGQUIT;
+use rustls::crypto::ring::default_provider;
 use sd_notify::NotifyState;
 use signal_hook::{
     consts::{SIGINT, SIGTERM},
@@ -27,9 +27,8 @@ use std::sync::mpsc::{channel, Receiver, Sender};
 use std::sync::Arc;
 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().expect("Failed to install RusTLS crypto provider");
     let parameters = Opt::parse_args();
     let file = parameters.conf.clone().unwrap();
     let maincfg = crate::utils::parceyaml::parce_main_config(file.as_str());