9 Commits

Author SHA1 Message Date
Ara Sadoyan
68140d0cf0 tye changes, optimization 2026-03-26 17:40:22 +01:00
Ara Sadoyan
7b9b206c13 optimization & cleanup 2026-03-26 16:58:53 +01:00
Ara Sadoyan
4706b281bc cleanup 2026-03-26 14:17:59 +01:00
Ara Sadoyan
1f8efc6af7 FUNDING.yml 2026-03-25 15:16:47 +01:00
Ara Sadoyan
9f595b2709 example config file update 2026-03-25 11:15:55 +01:00
Ara Sadoyan
ed44516015 added redirect_to directive for upstreams 2026-03-24 16:08:14 +01:00
Ara Sadoyan
17da7862e3 upstreams ID hashing update 2026-03-18 20:06:50 +01:00
Ara Sadoyan
24d00da855 performance improvement, sticky session minor bug fix 2026-03-17 19:21:05 +01:00
Ara Sadoyan
c9422759aa Minor performance improvement 2026-03-17 13:54:42 +01:00
11 changed files with 191 additions and 118 deletions

4
.github/FUNDING.yml vendored
View File

@@ -1,6 +1,6 @@
# These are supported funding model platforms
github: # Replace with up to 4 GitHub Sponsors-enabled usernames e.g., [user1, user2]
github: sadoyan
patreon: # Replace with a single Patreon username
open_collective: # Replace with a single Open Collective username
ko_fi: # Replace with a single Ko-fi username
@@ -10,6 +10,6 @@ liberapay: # Replace with a single Liberapay username
issuehunt: # Replace with a single IssueHunt username
lfx_crowdfunding: # Replace with a single LFX Crowdfunding project-name e.g., cloud-foundry
polar: # Replace with a single Polar username
buy_me_a_coffee: sadoyan
buy_me_a_coffee: sadoyan
thanks_dev: # Replace with a single thanks.dev username
custom: # Replace with up to 4 custom sponsorship URLs e.g., ['link1', 'link2']

View File

@@ -43,7 +43,7 @@ kubernetes:
path: "/"
upstream: "webapi-service"
- hostname: "webapi-service"
upstream: "vt-console-service"
upstream: "console-service"
path: "/one"
client_headers:
- "X-Some-Thing:Yaaaaaaaaaaaaaaa"
@@ -51,7 +51,7 @@ kubernetes:
rate_limit: 100
to_https: false
- hostname: "webapi-service"
upstream: "vt-rambulik-service"
upstream: "rambul-service"
path: "/two"
- hostname: "websocket-service"
upstream: "websocket-service"
@@ -72,6 +72,9 @@ upstreams:
- "127.0.0.4:8000"
- "127.0.0.5:8000"
"/ping":
authorization: # Will be ignored if global authentication is enabled.
type: "basic"
creds: "admin:admin"
to_https: false
server_headers:
- "X-Forwarded-Proto:https"
@@ -107,9 +110,10 @@ upstreams:
healthcheck: false
servers:
- "127.0.0.1:8001"
localpost:
rdr.mydomain.com:
paths:
"/":
to_https: false
redirect_to: "https://som.other.domain:6194"
healthcheck: false
servers:
- "127.0.0.1:9000"
- "127.0.0.1:8080"

View File

@@ -3,6 +3,7 @@ mod web;
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
// pub static A: CountingAllocator = CountingAllocator;
fn main() {
web::start::run();

View File

@@ -70,6 +70,7 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str, client: &Clie
to_https: upstream.to_https,
rate_limit: upstream.rate_limit,
healthcheck: upstream.healthcheck,
redirect_to: upstream.redirect_to.clone(),
authorization: upstream.authorization.clone(),
};

View File

@@ -1,5 +1,5 @@
use crate::utils::kuberconsul::{match_path, ConsulService, KubeEndpoints};
use crate::utils::structs::{InnerMap, ServiceMapping};
use crate::utils::structs::{GlobalServiceMapping, InnerMap};
use axum::http::{HeaderMap, HeaderValue};
use dashmap::DashMap;
use reqwest::Client;
@@ -7,7 +7,7 @@ use std::sync::atomic::AtomicUsize;
use std::sync::Arc;
use std::time::Duration;
pub async fn for_consul(url: String, token: Option<String>, conf: &ServiceMapping) -> Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>> {
pub async fn for_consul(url: String, token: Option<String>, conf: &GlobalServiceMapping) -> Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>> {
let client = Client::builder().timeout(Duration::from_secs(2)).danger_accept_invalid_certs(true).build().ok()?;
let mut headers = HeaderMap::new();
if let Some(token) = token {
@@ -27,6 +27,7 @@ pub async fn for_consul(url: String, token: Option<String>, conf: &ServiceMappin
// let prt = subsets.tagged_addresses.get("lan_ipv4").unwrap().port.clone();
let addr = subsets.tagged_addresses.get("lan_ipv4").unwrap().address.clone();
let prt = subsets.tagged_addresses.get("lan_ipv4").unwrap().port.clone();
// let redirect_link = conf.redirect_to.as_ref().map(|www| Arc::from(www.as_str()));
let to_add = Arc::from(InnerMap {
address: Arc::from(&*addr),
port: prt,
@@ -34,6 +35,7 @@ pub async fn for_consul(url: String, token: Option<String>, conf: &ServiceMappin
is_http2: false,
to_https: conf.to_https.unwrap_or(false),
rate_limit: conf.rate_limit,
redirect_to: None,
healthcheck: None,
authorization: None,
});
@@ -43,7 +45,7 @@ pub async fn for_consul(url: String, token: Option<String>, conf: &ServiceMappin
Some(upstreams)
}
pub async fn for_kuber(url: &str, token: &str, conf: &ServiceMapping) -> Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>> {
pub async fn for_kuber(url: &str, token: &str, conf: &GlobalServiceMapping) -> Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>> {
let to = Duration::from_secs(10);
let client = Client::builder().timeout(Duration::from_secs(10)).danger_accept_invalid_certs(true).build().ok()?;
let resp = client.get(url).timeout(to).bearer_auth(token).send().await.ok()?;
@@ -61,6 +63,7 @@ pub async fn for_kuber(url: &str, token: &str, conf: &ServiceMapping) -> Option<
let mut inner_vec = Vec::new();
for addr in addresses {
for port in &ports {
// let redirect_link = conf.redirect_to.as_ref().map(|www| Arc::from(www.as_str()));
let to_add = Arc::from(InnerMap {
address: Arc::from(addr.ip.clone()),
port: port.port.clone(),
@@ -69,6 +72,7 @@ pub async fn for_kuber(url: &str, token: &str, conf: &ServiceMapping) -> Option<
to_https: conf.to_https.unwrap_or(false),
rate_limit: conf.rate_limit,
healthcheck: None,
redirect_to: None,
authorization: None,
});
inner_vec.push(to_add);

View File

@@ -1,6 +1,6 @@
use crate::utils::httpclient;
use crate::utils::parceyaml::build_headers;
use crate::utils::structs::{Configuration, InnerMap, ServiceMapping, UpstreamsDashMap};
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;
@@ -52,7 +52,7 @@ pub struct ConsulTaggedAddress {
#[serde(rename = "Port")]
pub port: u16,
}
pub fn list_to_upstreams(lt: Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>>, upstreams: &UpstreamsDashMap, i: &ServiceMapping) {
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()) {
Some(upstr) => {
@@ -67,7 +67,7 @@ pub fn list_to_upstreams(lt: Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, Atomi
}
}
pub fn match_path(conf: &ServiceMapping, upstreams: &DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>, values: Vec<Arc<InnerMap>>) {
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)));
@@ -115,7 +115,7 @@ impl ServiceDiscovery for KubernetesDiscovery {
if let Some(kuber) = config.kubernetes.clone() {
if let Some(svc) = kuber.services {
for service in svc {
let header_list: DashMap<Arc<str>, Vec<(Arc<str>, Arc<str>)>> = DashMap::new();
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() {

View File

@@ -68,27 +68,27 @@ pub async fn load_configuration(d: &str, kind: &str) -> (Option<Configuration>,
}
async fn populate_headers_and_auth(config: &mut Configuration, parsed: &Config) {
let mut ch: Vec<(Arc<str>, Arc<str>)> = Vec::new();
let mut ch: Vec<(String, Arc<str>)> = Vec::new();
if let Some(headers) = &parsed.client_headers {
for header in headers {
if let Some((key, val)) = header.split_once(':') {
ch.push((Arc::from(key), Arc::from(val)));
ch.push((key.to_string(), Arc::from(val)));
}
}
}
let global_headers: DashMap<Arc<str>, Vec<(Arc<str>, Arc<str>)>> = DashMap::new();
let global_headers: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
global_headers.insert(Arc::from("/"), ch);
config.client_headers.insert(Arc::from("GLOBAL_CLIENT_HEADERS"), global_headers);
let mut sh: Vec<(Arc<str>, Arc<str>)> = Vec::new();
let mut sh: Vec<(String, Arc<str>)> = Vec::new();
if let Some(headers) = &parsed.server_headers {
for header in headers {
if let Some((key, val)) = header.split_once(':') {
sh.push((Arc::from(key.trim()), Arc::from(val.trim())));
sh.push((key.to_string(), Arc::from(val.trim())));
}
}
}
let server_global_headers: DashMap<Arc<str>, Vec<(Arc<str>, Arc<str>)>> = DashMap::new();
let server_global_headers: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
server_global_headers.insert(Arc::from("/"), sh);
config.server_headers.insert(Arc::from("GLOBAL_SERVER_HEADERS"), server_global_headers);
config.extraparams.to_https = parsed.to_https;
@@ -104,7 +104,7 @@ async fn populate_headers_and_auth(config: &mut Configuration, parsed: &Config)
auth_type: Arc::from(pa.auth_type.clone()),
auth_cred: Arc::from(pa.auth_cred.clone()),
};
config.extraparams.authentication = Some(y);
config.extraparams.authentication = Some(Arc::from(y));
}
}
@@ -119,8 +119,8 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) {
if let Some(rate) = &path_config.rate_limit {
info!("Applied Rate Limit for {} : {} request per second", hostname, rate);
}
let mut hl: Vec<(Arc<str>, Arc<str>)> = Vec::new();
let mut sl: Vec<(Arc<str>, Arc<str>)> = Vec::new();
let mut hl: Vec<(String, Arc<str>)> = Vec::new();
let mut sl: Vec<(String, Arc<str>)> = Vec::new();
build_headers(&path_config.client_headers, config, &mut hl);
build_headers(&path_config.server_headers, config, &mut sl);
client_header_list.insert(Arc::from(path.as_str()), hl);
@@ -135,7 +135,7 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) {
};
path_auth = Some(Arc::from(y));
}
let redirect_link = path_config.redirect_to.as_ref().map(|www| Arc::from(www.as_str()));
if let Some((ip, port_str)) = server.split_once(':') {
if let Ok(port) = port_str.parse::<u16>() {
server_list.push(Arc::from(InnerMap {
@@ -146,6 +146,7 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) {
to_https: path_config.to_https.unwrap_or(false),
rate_limit: path_config.rate_limit,
healthcheck: path_config.healthcheck,
redirect_to: redirect_link,
authorization: path_auth,
}));
}
@@ -185,13 +186,24 @@ pub fn parce_main_config(path: &str) -> AppConfig {
cfo.local_server = Option::from((ip.to_string(), port));
}
}
// if let Some(tlsport_cfg) = cfo.proxy_address_tls.clone() {
// if let Some((_, port_str)) = tlsport_cfg.split_once(':') {
// if let Ok(port) = port_str.parse::<u16>() {
// cfo.proxy_port_tls = Some(port);
// }
// }
// };
if let Some(tlsport_cfg) = cfo.proxy_address_tls.clone() {
if let Some((_, port_str)) = tlsport_cfg.split_once(':') {
if let Ok(port) = port_str.parse::<u16>() {
cfo.proxy_port_tls = Some(port);
}
cfo.proxy_port_tls = Some(port_str.to_string());
}
};
if let Some((_, port_str)) = cfo.proxy_address_http.split_once(':') {
cfo.proxy_port = Some(port_str.to_string());
}
cfo.proxy_tls_grade = parce_tls_grades(cfo.proxy_tls_grade.clone());
cfo
}
@@ -242,11 +254,11 @@ fn log_builder(conf: &AppConfig) {
env_logger::builder().init();
}
pub fn build_headers(path_config: &Option<Vec<String>>, _config: &Configuration, hl: &mut Vec<(Arc<str>, Arc<str>)>) {
pub fn build_headers(path_config: &Option<Vec<String>>, _config: &Configuration, hl: &mut Vec<(String, Arc<str>)>) {
if let Some(headers) = &path_config {
for header in headers {
if let Some((key, val)) = header.split_once(':') {
hl.push((Arc::from(key.trim()), Arc::from(val.trim())));
hl.push((key.trim().to_string(), Arc::from(val.trim())));
}
}
}

View File

@@ -7,18 +7,18 @@ use std::sync::Arc;
pub type UpstreamsDashMap = DashMap<Arc<str>, DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>>;
pub type UpstreamsIdMap = DashMap<String, Arc<InnerMap>>;
pub type Headers = DashMap<Arc<str>, DashMap<Arc<str>, Vec<(Arc<str>, Arc<str>)>>>;
pub type Headers = DashMap<Arc<str>, DashMap<Arc<str>, Vec<(String, Arc<str>)>>>;
#[derive(Clone, Debug, Default)]
pub struct Extraparams {
pub to_https: Option<bool>,
pub sticky_sessions: bool,
pub authentication: Option<InnerAuth>,
pub authentication: Option<Arc<InnerAuth>>,
pub rate_limit: Option<isize>,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct ServiceMapping {
pub struct GlobalServiceMapping {
pub upstream: String,
pub hostname: String,
pub path: Option<String>,
@@ -32,14 +32,14 @@ pub struct ServiceMapping {
#[derive(Clone, Default, Debug, Serialize, Deserialize)]
pub struct Kubernetes {
pub servers: Option<Vec<String>>,
pub services: Option<Vec<ServiceMapping>>,
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<ServiceMapping>>,
pub services: Option<Vec<GlobalServiceMapping>>,
pub token: Option<String>,
}
#[derive(Debug, Default, Serialize, Deserialize)]
@@ -86,6 +86,7 @@ pub struct PathConfig {
pub server_headers: Option<Vec<String>>,
pub rate_limit: Option<isize>,
pub healthcheck: Option<bool>,
pub redirect_to: Option<String>,
pub authorization: Option<Auth>,
}
#[derive(Debug, Default)]
@@ -113,7 +114,8 @@ pub struct AppConfig {
pub config_tls_certificate: Option<String>,
pub config_tls_key_file: Option<String>,
pub proxy_address_tls: Option<String>,
pub proxy_port_tls: Option<u16>,
pub proxy_port_tls: Option<String>,
pub proxy_port: Option<String>,
pub local_server: Option<(String, u16)>,
pub proxy_certificates: Option<String>,
pub proxy_tls_grade: Option<String>,
@@ -138,6 +140,7 @@ pub struct InnerMap {
pub to_https: bool,
pub rate_limit: Option<isize>,
pub healthcheck: Option<bool>,
pub redirect_to: Option<Arc<str>>,
pub authorization: Option<Arc<InnerAuth>>,
}
@@ -152,6 +155,7 @@ impl InnerMap {
to_https: Default::default(),
rate_limit: Default::default(),
healthcheck: Default::default(),
redirect_to: Default::default(),
authorization: Default::default(),
}
}

View File

@@ -146,7 +146,7 @@ pub fn compare_dashmaps(map1: &UpstreamsDashMap, map2: &UpstreamsDashMap) -> boo
true
}
pub fn merge_headers(target: &DashMap<Arc<str>, Vec<(Arc<str>, Arc<str>)>>, source: &DashMap<Arc<str>, Vec<(Arc<str>, Arc<str>)>>) {
pub fn merge_headers(target: &DashMap<Arc<str>, Vec<(String, Arc<str>)>>, source: &DashMap<Arc<str>, Vec<(String, Arc<str>)>>) {
for entry in source.iter() {
let global_key = entry.key().clone();
let global_values = entry.value().clone();
@@ -166,8 +166,22 @@ pub fn clone_idmap_into(original: &UpstreamsDashMap, cloned: &UpstreamsIdMap) {
let new_vec = vec.clone();
for x in vec.iter() {
let mut id = String::new();
write!(&mut id, "{}:{}:{}", x.address, x.port, x.is_ssl).unwrap();
write!(
&mut id,
"{}:{}:{}:{}:{}:{}:{}:{:?}",
outer_entry.key(),
x.address,
x.port,
x.is_http2,
x.to_https,
x.rate_limit.unwrap_or_default(),
x.healthcheck.unwrap_or_default(),
x.authorization
)
.unwrap_or(());
let mut hasher = Sha256::new();
// address: "127.0.0.3", port: 8000, is_ssl: false, is_http2: false, to_https: false, rate_limit: Some(200), healthcheck: None, authorization: None } }
hasher.update(id.clone().into_bytes());
let hash = hasher.finalize();
let hex_hash = base16ct::lower::encode_string(&hash);
@@ -180,9 +194,9 @@ pub fn clone_idmap_into(original: &UpstreamsDashMap, cloned: &UpstreamsIdMap) {
to_https: false,
rate_limit: None,
healthcheck: None,
redirect_to: None,
authorization: None,
};
cloned.insert(id, Arc::from(to_add));
cloned.insert(hh, Arc::from(x.to_owned()));
// println!("CLONNED :===========> {:?}", cloned);
@@ -190,6 +204,7 @@ pub fn clone_idmap_into(original: &UpstreamsDashMap, cloned: &UpstreamsIdMap) {
new_inner_map.insert(path.clone(), new_vec);
}
}
info!("Upstreams are fully populated. Ready to server requests");
}
pub fn listdir(dir: String) -> Vec<tls::CertificateConfig> {
@@ -368,3 +383,16 @@ pub fn upstreams_liveness_json(configured: &UpstreamsDashMap, current: &Upstream
}
Value::Object(result)
}
#[allow(dead_code)]
pub fn prepend(prefix: &str, val: &Option<Arc<str>>, uri: &str, port: &str) -> Option<String> {
val.as_ref().map(|s| {
let mut buf = String::with_capacity(32);
buf.push_str(prefix);
buf.push_str(s);
buf.push_str(":");
buf.push_str(port);
buf.push_str(uri);
buf
})
}

View File

@@ -6,8 +6,8 @@ use std::sync::Arc;
#[derive(Debug, Clone)]
pub struct GetHostsReturHeaders {
pub client_headers: Option<Vec<(Arc<str>, Arc<str>)>>,
pub server_headers: Option<Vec<(Arc<str>, Arc<str>)>>,
pub client_headers: Option<Vec<(String, Arc<str>)>>,
pub server_headers: Option<Vec<(String, Arc<str>)>>,
}
#[async_trait]

View File

@@ -1,6 +1,7 @@
use crate::utils::auth::authenticate;
use crate::utils::metrics::*;
use crate::utils::structs::{AppConfig, Extraparams, Headers, InnerMap, UpstreamsDashMap, UpstreamsIdMap};
// use std::collections::BTreeMap;
use crate::web::gethosts::{GetHost, GetHostsReturHeaders};
use arc_swap::ArcSwap;
use async_trait::async_trait;
@@ -20,7 +21,6 @@ use pingora_limits::rate::Rate;
use pingora_proxy::{ProxyHttp, Session};
use sha2::{Digest, Sha256};
use std::cell::RefCell;
// use std::collections::BTreeMap;
use std::fmt::Write;
use std::sync::Arc;
use std::time::Duration;
@@ -43,14 +43,13 @@ pub struct LB {
pub struct Context {
backend_id: Option<String>,
to_https: bool,
sticky_sessions: bool,
redirect_to: Option<String>,
// redirect_to: Option<String>,
start_time: Instant,
hostname: Option<Arc<str>>,
upstream_peer: Option<Arc<InnerMap>>,
extraparams: arc_swap::Guard<Arc<Extraparams>>,
client_headers: Option<Arc<Vec<(Arc<str>, Arc<str>)>>>,
client_headers: Option<Vec<(String, Arc<str>)>>,
}
#[async_trait]
@@ -59,9 +58,8 @@ impl ProxyHttp for LB {
fn new_ctx(&self) -> Self::CTX {
Context {
backend_id: None,
to_https: false,
sticky_sessions: false,
redirect_to: None,
// redirect_to: None,
start_time: Instant::now(),
hostname: None,
upstream_peer: None,
@@ -70,15 +68,6 @@ impl ProxyHttp for LB {
}
}
async fn request_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<bool> {
// let ep = _ctx.extraparams.as_ref();
if let Some(auth) = &_ctx.extraparams.authentication {
let authenticated = authenticate(&auth.auth_type, &auth.auth_cred, &session);
if !authenticated {
let _ = session.respond_error(401).await;
warn!("Forbidden: {:?}, {}", session.client_addr(), session.req_header().uri.path());
return Ok(true);
}
}
let hostname = return_header_host_from_upstream(session, &self.ump_upst);
_ctx.hostname = hostname;
let mut backend_id = None;
@@ -98,19 +87,14 @@ impl ProxyHttp for LB {
None => return Ok(false),
Some(host) => {
let optioninnermap = self.get_host(host, session.req_header().uri.path(), backend_id);
match optioninnermap {
None => return Ok(false),
Some(ref innermap) => {
// Inner auth works only if global is disabled.
if let Some(auth) = &innermap.authorization {
if _ctx.extraparams.authentication.is_none() {
let authenticated = authenticate(&auth.auth_type, &auth.auth_cred, &session);
if !authenticated {
let _ = session.respond_error(401).await;
warn!("Forbidden: {:?}, {}", session.client_addr(), session.req_header().uri.path());
return Ok(true);
}
if let Some(auth) = _ctx.extraparams.authentication.as_ref().or(innermap.authorization.as_ref()) {
if !authenticate(&auth.auth_type, &auth.auth_cred, &session) {
let _ = session.respond_error(401).await;
warn!("Forbidden: {:?}, {}", session.client_addr(), session.req_header().uri.path());
return Ok(true);
}
}
@@ -130,6 +114,44 @@ impl ProxyHttp for LB {
return Ok(true);
}
}
if let Some(redirect_to) = &innermap.redirect_to {
let uri = session.req_header().uri.path();
let capacity = redirect_to.len() + uri.len();
let mut s = String::with_capacity(capacity);
s.push_str(redirect_to);
s.push_str(uri);
let mut resp = ResponseHeader::build(StatusCode::MOVED_PERMANENTLY, None)?;
resp.insert_header("Location", s)?;
resp.insert_header("Content-Length", "0")?;
session.write_response_header(Box::new(resp), true).await?;
return Ok(true);
}
if _ctx.extraparams.to_https.unwrap_or(false) || innermap.to_https {
if let Some(stream) = session.stream() {
if stream.get_ssl().is_none() {
if let Some(host) = _ctx.hostname.as_ref() {
let port = self.config.proxy_port_tls.as_deref().unwrap_or("443");
let uri = session.req_header().uri.path();
let capacity = host.len() + uri.len() + 8;
let mut s = String::with_capacity(capacity);
s.push_str("https://");
s.push_str(host);
if port != "443" {
s.push_str(":");
s.push_str(&port);
}
s.push_str(uri);
let mut resp = ResponseHeader::build(StatusCode::MOVED_PERMANENTLY, None)?;
resp.insert_header("Location", s)?;
resp.insert_header("Content-Length", "0")?;
session.write_response_header(Box::new(resp), true).await?;
return Ok(true);
}
}
}
}
}
}
_ctx.upstream_peer = optioninnermap;
@@ -150,38 +172,36 @@ impl ProxyHttp for LB {
peer.options.verify_cert = false;
peer.options.verify_hostname = false;
}
/*
Experimental optionsv
The following TCP optimizations were tested but caused performance degrade under heavy load:
peer.options.tcp_keepalive = Some(TcpKeepalive {
idle: Duration::from_secs(60),
interval: Duration::from_secs(10),
count: 5,
user_timeout: Duration::from_secs(30),
});
// Experimental optionsv
// The following TCP optimizations were tested but caused performance degrade under heavy load:
// peer.options.tcp_keepalive = Some(TcpKeepalive {
// idle: Duration::from_secs(60),
// interval: Duration::from_secs(10),
// count: 5,
// user_timeout: Duration::from_secs(30),
// });
//
// peer.options.idle_timeout = Some(Duration::from_secs(300));
// peer.options.tcp_recv_buf = Some(128 * 1024);
// End of experimental options
if ctx.extraparams.to_https.unwrap_or(false) || innermap.to_https {
if let Some(stream) = session.stream() {
if stream.get_ssl().is_none() {
if let Some(host) = ctx.hostname.as_ref() {
let uri = session.req_header().uri.path_and_query().map_or("/", |pq| pq.as_str());
let port = self.config.proxy_port_tls.unwrap_or(443);
ctx.to_https = true;
let mut s = String::with_capacity(64);
write!(&mut s, "https://{}:{}{}", host, port, uri).unwrap_or_default();
ctx.redirect_to = Some(s);
}
}
}
}
peer.options.idle_timeout = Some(Duration::from_secs(300));
peer.options.tcp_recv_buf = Some(128 * 1024);
End of experimental options
*/
if ctx.extraparams.sticky_sessions {
let mut s = String::with_capacity(64);
write!(&mut s, "{}:{}:{}", innermap.address, innermap.port, innermap.is_ssl).unwrap();
write!(
&mut s,
"{}:{}:{}:{}:{}:{}:{}:{:?}",
hostname,
innermap.address,
innermap.port,
innermap.is_http2,
innermap.to_https,
innermap.rate_limit.unwrap_or_default(),
innermap.healthcheck.unwrap_or_default(),
innermap.authorization
)
.unwrap_or(());
ctx.backend_id = Some(s);
ctx.sticky_sessions = true;
}
@@ -216,18 +236,19 @@ impl ProxyHttp for LB {
}
async fn upstream_request_filter(&self, session: &mut Session, upstream_request: &mut RequestHeader, ctx: &mut Self::CTX) -> Result<()> {
if let Some(hostname) = ctx.hostname.as_deref() {
upstream_request.insert_header("Host", hostname)?;
}
// if let Some(hostname) = ctx.hostname.as_deref() {
// upstream_request.insert_header("Host", hostname)?;
// }
if let Some(client_ip) = session.client_addr() {
IP_BUFFER.with(|buffer| {
let mut buf = buffer.borrow_mut();
buf.clear();
write!(buf, "{}", client_ip).unwrap_or(());
upstream_request.append_header("x-forward-for", buf.as_str()).unwrap_or(false);
upstream_request.append_header("X-Forward-For", buf.as_str()).unwrap_or(false);
});
}
let hostname = ctx.hostname.as_deref().unwrap_or("localhost");
let path = session.req_header().uri.path();
let GetHostsReturHeaders { server_headers, client_headers } = match self.get_header(hostname, path) {
@@ -237,46 +258,43 @@ impl ProxyHttp for LB {
if let Some(sh) = server_headers {
for (k, v) in sh {
upstream_request.insert_header(k.to_string(), v.as_ref())?;
upstream_request.insert_header(k, v.as_ref())?;
}
}
if let Some(ch) = client_headers {
ctx.client_headers = Some(Arc::new(ch));
ctx.client_headers = Some(ch);
}
Ok(())
}
async fn response_filter(&self, session: &mut Session, _upstream_response: &mut ResponseHeader, ctx: &mut Self::CTX) -> Result<()> {
async fn response_filter(&self, _session: &mut Session, _upstream_response: &mut ResponseHeader, ctx: &mut Self::CTX) -> Result<()> {
if ctx.sticky_sessions {
if let Some(bid) = ctx.backend_id.clone() {
if REVERSE_STORE.get(&*bid).is_none() {
if let Some(bid) = &ctx.backend_id {
let tt = if let Some(existing) = REVERSE_STORE.get(bid) {
existing.value().clone()
} else {
let mut hasher = Sha256::new();
hasher.update(bid.clone().into_bytes());
hasher.update(bid.as_bytes());
let hash = hasher.finalize();
let hex_hash = base16ct::lower::encode_string(&hash);
let hh = hex_hash[0..50].to_string();
REVERSE_STORE.insert(bid.clone(), hh.clone());
REVERSE_STORE.insert(hh.clone(), bid.clone());
}
if let Some(tt) = REVERSE_STORE.get(&*bid) {
let _ = _upstream_response.insert_header("set-cookie", format!("backend_id={}; Path=/; Max-Age=600; HttpOnly; SameSite=Lax", tt.value()));
}
hh
};
// let _ = _upstream_response.insert_header("set-cookie", format!("backend_id={}; Path=/; Max-Age=600; HttpOnly; SameSite=Lax", tt));
let mut buf = String::with_capacity(80);
buf.push_str("backend_id=");
buf.push_str(&tt);
buf.push_str("; Path=/; Max-Age=600; HttpOnly; SameSite=Lax");
let _ = _upstream_response.insert_header("set-cookie", buf.as_str());
}
}
if ctx.to_https {
let mut redirect_response = ResponseHeader::build(StatusCode::MOVED_PERMANENTLY, None)?;
redirect_response.insert_header("Location", ctx.redirect_to.clone().unwrap_or(String::from("/")))?;
redirect_response.insert_header("Content-Length", "0")?;
session.write_response_header(Box::new(redirect_response), false).await?;
}
// ALLOCATIONS !
if let Some(client_headers) = &ctx.client_headers {
for (k, v) in client_headers.iter() {
_upstream_response.append_header(k.to_string(), v.as_ref())?;
_upstream_response.append_header(k.clone(), v.as_ref())?;
}
}
// END ALLOCATIONS !
// session.set_keepalive(Some(300));
// println!("session.get_keepalive: {:?}", session.get_keepalive());
@@ -291,7 +309,8 @@ impl ProxyHttp for LB {
code: session.response_written().map(|resp| resp.status),
latency: ctx.start_time.elapsed(),
version: session.req_header().version,
upstream: ctx.hostname.clone().unwrap_or(Arc::from("localhost")),
// upstream: ctx.hostname.clone().unwrap_or(Arc::from("localhost")),
upstream: ctx.hostname.take().unwrap_or_else(|| Arc::from("localhost")),
};
calc_metrics(m);
}