COnfig parameters update

README.md

@@ -101,6 +101,7 @@ A sample `upstreams.yaml` entry:
 
 ```yaml
 provider: "file"
+stickysessions: false
 globals:
   headers:
     - "Access-Control-Allow-Origin:*"
@@ -130,6 +131,7 @@ myhost.mydomain.com:
 
 This means:
 
+- Sticky sessions are disabled globally. This boolean setting applies to all upstreams.
 - Requests to `myhost.mydomain.com/` will be load balanced to `127.0.0.1` and `127.0.0.2` servers via plain http.
 - Requests to `myhost.mydomain.com/foo` will be load balanced to `127.0.0.4` and `127.0.0.5` servers via https.
 - Global headers (CORS for this case) will be injected to all upstreams

etc/main.yaml

@@ -17,4 +17,3 @@ log_level: info # info, warn, error, debug, trace, off
 hc_method: HEAD # Healthcheck method (HEAD, GET, POST are supported) UPPERCASE
 hc_interval: 2 #Interval for health checks in seconds
 master_key: 910517d9-f9a1-48de-8826-dbadacbd84af-cb6f830e-ab16-47ec-9d8f-0090de732774 # Mater key for working with API server and JWT Secret
-sticky_sessions: true # Globally enables/disables Sticky session . Boolean will panic on any other than true/false .

etc/upstreams.yaml

@@ -1,5 +1,6 @@
 # The file is under watch and hot reload , changes are applied immediately, no need to restart or reload
 provider: "file" # consul
+stickysessions: false
 globals:
   headers: # Global headers, appended for all upstreams and all paths.
     - "Access-Control-Allow-Origin:*"

src/utils/consul.rs

@@ -70,6 +70,7 @@ pub async fn start(fp: String, mut toreturn: Sender<Configuration>) {
                                         headers: Default::default(),
                                         consul: None,
                                         typecfg: "".to_string(),
+                                        extraparams: config.extraparams.clone(),
                                         globals: Default::default(),
                                     };
 
@@ -94,62 +95,6 @@ pub async fn start(fp: String, mut toreturn: Sender<Configuration>) {
     }
 }
 
-/*
-async fn http_request(url: String, whitelist: Option<Vec<String>>) -> Option<UpstreamsDashMap> {
-    let client = reqwest::Client::new();
-    let to = Duration::from_secs(1);
-    let upstreams = UpstreamsDashMap::new();
-    let excludes = vec!["consul", "nomad", "nomad-client"];
-    let ss = url.clone() + "/v1/catalog/service";
-    let response = client.get(ss.clone() + "s").timeout(to).send().await;
-    match response {
-        Ok(r) => {
-            let json = r.json::<HashMap<String, Vec<String>>>().await;
-            match json {
-                Ok(_j) => {
-                    for (k, _v) in _j {
-                        match whitelist.clone() {
-                            Some(whitelist) => {
-                                if whitelist.iter().any(|i| *i == k) {
-                                    let mut pref: String = ss.clone() + "/";
-                                    pref.push_str(&k);
-                                    let list = get_by_http(pref).await;
-                                    match list {
-                                        Some(list) => {
-                                            upstreams.insert(k.to_string(), list);
-                                        }
-                                        None => {}
-                                    }
-                                }
-                            }
-                            None => {
-                                if !excludes.iter().any(|&i| i == k) {
-                                    let mut pref: String = ss.clone() + "/";
-                                    pref.push_str(&k);
-                                    let list = get_by_http(pref).await;
-                                    match list {
-                                        Some(list) => {
-                                            upstreams.insert(k.to_string(), list);
-                                        }
-                                        None => {}
-                                    }
-                                }
-                            }
-                        }
-                    }
-                    // print_upstreams(&upstreams);
-                    Some(upstreams)
-                }
-                Err(_) => None,
-            }
-        }
-        Err(e) => {
-            println!("Error: {:?}", e);
-            None
-        }
-    }
-}
-*/
 async fn consul_request(url: String, whitelist: Option<Vec<ServiceMapping>>, token: Option<String>) -> Option<UpstreamsDashMap> {
     let upstreams = UpstreamsDashMap::new();
     let ss = url.clone() + "/v1/catalog/service/";

src/utils/filewatch.rs

@@ -22,7 +22,6 @@ pub async fn start(fp: String, mut toreturn: Sender<Configuration>) {
         }
         None => {}
     }
-
     let _watcher_handle = task::spawn_blocking({
         let parent_dir = parent_dir.to_path_buf(); // Move directory path into the closure
         move || {

src/utils/parceyaml.rs

@@ -13,6 +13,11 @@ pub struct ServiceMapping {
     pub real: String,
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Extraparams {
+    pub stickysessions: bool,
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Consul {
     pub servers: Option<Vec<String>>,
@@ -22,6 +27,7 @@ pub struct Consul {
 #[derive(Debug, Serialize, Deserialize)]
 struct Config {
     provider: String,
+    stickysessions: bool,
     upstreams: Option<HashMap<String, HostConfig>>,
     globals: Option<HashMap<String, Vec<String>>>,
     consul: Option<Consul>,
@@ -43,6 +49,7 @@ pub struct Configuration {
     pub headers: Headers,
     pub consul: Option<Consul>,
     pub typecfg: String,
+    pub extraparams: Extraparams,
     pub globals: Option<DashMap<String, Vec<String>>>,
 }
 
@@ -53,6 +60,7 @@ pub fn load_configuration(d: &str, kind: &str) -> Option<Configuration> {
         headers: Default::default(),
         consul: None,
         typecfg: "".to_string(),
+        extraparams: Extraparams { stickysessions: false },
         globals: Default::default(),
     };
     toreturn.upstreams = UpstreamsDashMap::new();
@@ -97,7 +105,7 @@ pub fn load_configuration(d: &str, kind: &str) -> Option<Configuration> {
                 }
                 global_headers.insert("/".to_string(), hl);
                 toreturn.headers.insert("GLOBAL_HEADERS".to_string(), global_headers);
-
+                toreturn.extraparams.stickysessions = parsed.stickysessions;
                 let cfg = DashMap::new();
                 if let Some(k) = globals.get("authorization") {
                     cfg.insert("authorization".to_string(), k.to_owned());
@@ -166,7 +174,6 @@ pub fn load_configuration(d: &str, kind: &str) -> Option<Configuration> {
 
 #[derive(Debug, Deserialize)]
 pub struct AppConfig {
-    pub sticky_sessions: bool,
     pub hc_interval: u16,
     pub hc_method: String,
     pub upstreams_conf: String,

src/web/bgservice.rs

@@ -54,6 +54,14 @@ impl BackgroundService for LB {
                             clone_dashmap_into(&ss.upstreams, &self.ump_full);
                             clone_dashmap_into(&ss.upstreams, &self.ump_upst);
                             self.proxyconf.clear();
+
+                            // println!(" ====> {:?}", self.extraparams);
+                            {
+                                let mut write_guard = self.extraparams.write().await;
+                                write_guard.stickysessions = ss.extraparams.stickysessions;
+                            }
+                            // println!(" ====> {:?}", self.extraparams);
+
                             match ss.globals {
                                 Some(globals) => {
                                     for (k,v) in globals {

src/web/proxyhttp.rs

@@ -1,5 +1,5 @@
 use crate::utils::auth::authenticate;
-use crate::utils::parceyaml::AppConfig;
+use crate::utils::parceyaml::{AppConfig, Extraparams};
 use crate::utils::tools::*;
 use crate::web::gethosts::GetHost;
 use async_trait::async_trait;
@@ -13,6 +13,7 @@ use pingora_http::ResponseHeader;
 use pingora_proxy::{ProxyHttp, Session};
 use std::ops::Deref;
 use std::sync::Arc;
+use tokio::sync::RwLock;
 
 pub struct LB {
     pub ump_upst: Arc<UpstreamsDashMap>,
@@ -23,6 +24,8 @@ pub struct LB {
     pub config: Arc<AppConfig>,
     pub local: Arc<(String, u16)>,
     pub proxyconf: Arc<DashMap<String, Vec<String>>>,
+    // pub extraparams: Arc<Mutex<Extraparams>>,
+    pub extraparams: Arc<RwLock<Extraparams>>,
 }
 
 pub struct MyCtx {
@@ -53,15 +56,17 @@ impl ProxyHttp for LB {
                 // }
 
                 let mut backend_id = None;
-
+                {
-                if self.config.sticky_sessions {
+                    let read_guard = self.extraparams.read().await;
-                    if let Some(cookies) = session.req_header().headers.get("cookie") {
+                    if read_guard.stickysessions {
-                        if let Ok(cookie_str) = cookies.to_str() {
+                        if let Some(cookies) = session.req_header().headers.get("cookie") {
-                            for cookie in cookie_str.split(';') {
+                            if let Ok(cookie_str) = cookies.to_str() {
-                                let trimmed = cookie.trim();
+                                for cookie in cookie_str.split(';') {
-                                if let Some(value) = trimmed.strip_prefix("backend_id=") {
+                                    let trimmed = cookie.trim();
-                                    backend_id = Some(value);
+                                    if let Some(value) = trimmed.strip_prefix("backend_id=") {
-                                    break;
+                                        backend_id = Some(value);
+                                        break;
+                                    }
                                 }
                             }
                         }
@@ -134,11 +139,14 @@ impl ProxyHttp for LB {
     async fn response_filter(&self, _session: &mut Session, _upstream_response: &mut ResponseHeader, _ctx: &mut Self::CTX) -> Result<()> {
         // _upstream_response.insert_header("X-Proxied-From", "Fooooooooooooooo").unwrap();
 
-        if self.config.sticky_sessions {
+        {
-            let backend_id = _ctx.backend_id.clone();
+            let read_guard = self.extraparams.read().await;
-            if let Some(bid) = self.ump_byid.get(&backend_id) {
+            if read_guard.stickysessions {
-                // let _ = _upstream_response.insert_header("set-cookie", format!("backend {}", bid.0));
+                let backend_id = _ctx.backend_id.clone();
-                let _ = _upstream_response.insert_header("set-cookie", format!("backend_id={}; Path=/; Max-Age=600; HttpOnly; SameSite=Lax", bid.0));
+                if let Some(bid) = self.ump_byid.get(&backend_id) {
+                    // let _ = _upstream_response.insert_header("set-cookie", format!("backend {}", bid.0));
+                    let _ = _upstream_response.insert_header("set-cookie", format!("backend_id={}; Path=/; Max-Age=600; HttpOnly; SameSite=Lax", bid.0));
+                }
             }
         }
 

src/web/start.rs

@@ -1,3 +1,4 @@
+use crate::utils::parceyaml::Extraparams;
 use crate::utils::tools::*;
 use crate::web::proxyhttp::LB;
 use dashmap::DashMap;
@@ -6,6 +7,7 @@ use pingora_core::prelude::{background_service, Opt};
 use pingora_core::server::Server;
 use std::env;
 use std::sync::Arc;
+use tokio::sync::RwLock;
 
 pub fn run() {
     let parameters = Some(Opt::parse_args()).unwrap();
@@ -28,11 +30,13 @@ pub fn run() {
     let ff: UpstreamsDashMap = DashMap::new();
     let im: UpstreamsIdMap = DashMap::new();
     let hh: Headers = DashMap::new();
+    let ec: Extraparams = Extraparams { stickysessions: false };
 
     let uf_config = Arc::new(uf);
     let ff_config = Arc::new(ff);
     let im_config = Arc::new(im);
     let hh_config = Arc::new(hh);
+    let ec_config = Arc::new(RwLock::new(ec));
 
     let cfg = Arc::new(maincfg);
     let local = Arc::new(local_conf);
@@ -48,6 +52,7 @@ pub fn run() {
         local: local.clone(),
         headers: hh_config.clone(),
         proxyconf: pconf.clone(),
+        extraparams: ec_config.clone(),
     };
     let bg = LB {
         ump_upst: uf_config.clone(),
@@ -57,6 +62,7 @@ pub fn run() {
         local: local.clone(),
         headers: hh_config.clone(),
         proxyconf: pconf.clone(),
+        extraparams: ec_config.clone(),
     };
 
     // env_logger::Env::new();