mirror of
https://github.com/sadoyan/aralez.git
synced 2026-04-30 23:08:40 +08:00
Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
68140d0cf0 | ||
|
|
7b9b206c13 | ||
|
|
4706b281bc | ||
|
|
1f8efc6af7 | ||
|
|
9f595b2709 | ||
|
|
ed44516015 | ||
|
|
17da7862e3 | ||
|
|
24d00da855 | ||
|
|
c9422759aa |
4
.github/FUNDING.yml
vendored
4
.github/FUNDING.yml
vendored
@@ -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']
|
||||
|
||||
@@ -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"
|
||||
@@ -3,6 +3,7 @@ mod web;
|
||||
|
||||
#[global_allocator]
|
||||
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
|
||||
// pub static A: CountingAllocator = CountingAllocator;
|
||||
|
||||
fn main() {
|
||||
web::start::run();
|
||||
|
||||
@@ -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(),
|
||||
};
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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())));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user