Compare commits

...
8 Commits
12 changed files with 492 additions and 617 deletions
Generated
+286 -540
View File
File diff suppressed because it is too large Load Diff
+8 -6
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "aralez" name = "aralez"
version = "0.92.13" version = "0.92.15"
edition = "2021" edition = "2021"
license = "Apache-2.0" license = "Apache-2.0"
description = "Reverse proxy built on top of Cloudflare's Pingora" description = "Reverse proxy built on top of Cloudflare's Pingora"
@@ -21,22 +21,23 @@ pingora-core = "0.8.1"
pingora-proxy = "0.8.1" pingora-proxy = "0.8.1"
pingora-http = "0.8.1" pingora-http = "0.8.1"
pingora-limits = "0.8.1" pingora-limits = "0.8.1"
pingora-cache = "0.8.1"
dashmap = "7.0.0-rc2" dashmap = "7.0.0-rc2"
async-trait = "0.1.89" async-trait = "0.1.89"
log = "0.4.30" log = "0.4.33"
notify = "9.0.0-rc.4" notify = "9.0.0-rc.4"
axum = { version = "0.8.9" } axum = { version = "0.8.9" }
reqwest = { version = "0.13.4", features = ["json", "stream", "blocking"] } reqwest = { version = "0.13.4", features = ["json", "stream", "blocking"] }
serde = { version = "1.0.228", features = ["derive"] } serde = { version = "1.0.228", features = ["derive"] }
noyalib = { version = "0.0.8", features = ["compat-serde-yaml"] } noyalib = { version = "0.0.15", features = ["compat-serde-yaml"] }
rand = "0.10.1" rand = "0.10.2"
base64 = "0.22.1" base64 = "0.22.1"
jsonwebtoken = { version = "10.4.0", default-features = false, features = ["use_pem", "rust_crypto"] } jsonwebtoken = { version = "10.4.0", default-features = false, features = ["use_pem", "rust_crypto"] }
tonic = "0.14.6" tonic = "0.14.6"
sha2 = { version = "0.11.0-rc.5", default-features = false } sha2 = { version = "0.11.0-rc.5", default-features = false }
base16ct = { version = "1.0.0", features = ["alloc"] } base16ct = { version = "1.0.0", features = ["alloc"] }
urlencoding = "2.1.3" urlencoding = "2.1.3"
arc-swap = "1.9.1" arc-swap = "1.9.2"
prometheus = "0.14.0" prometheus = "0.14.0"
x509-parser = "0.18.1" x509-parser = "0.18.1"
rustls-pemfile = "2.2.0" rustls-pemfile = "2.2.0"
@@ -48,8 +49,9 @@ moka = { version = "0.12.15", features = ["sync"] }
ahash = "0.8.12" ahash = "0.8.12"
instant-acme = "0.8.5" instant-acme = "0.8.5"
rcgen = "0.14.8" rcgen = "0.14.8"
log4rs = "1.4.0" 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 } mimalloc = { version = "0.1.52", default-features = false }
signal-hook = "0.4.4" signal-hook = "0.4.4"
sd-notify = "0.5.0" sd-notify = "0.5.0"
libc = "0.2.186" libc = "0.2.186"
+4 -2
View File
@@ -61,7 +61,7 @@ Built on Rust, on top of **Cloudflare’s Pingora engine**, **Aralez** delivers
### `main.yaml` ### `main.yaml`
| Key | Example Value | Description | | Key | Example Value | Description |
|----------------------------------|----------------------------|-------------------------------------------------------------------------------------------------| |----------------------------------|----------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| **threads** | 12 | Number of running daemon threads. Optional, defaults to 1 | | **threads** | 12 | Number of running daemon threads. Optional, defaults to 1 |
| **runuser** | aralez | Optional. Username for running aralez after dropping root privileges (requires launch as root) | | **runuser** | aralez | Optional. Username for running aralez after dropping root privileges (requires launch as root) |
| **rungroup** | aralez | Optional. Group for running aralez after dropping root privileges (requires launch as root) | | **rungroup** | aralez | Optional. Group for running aralez after dropping root privileges (requires launch as root) |
@@ -76,12 +76,14 @@ Built on Rust, on top of **Cloudflare’s Pingora engine**, **Aralez** delivers
| **upstreams_conf** | /etc/aralez/upstreams.yaml | Location of the upstreams file | | **upstreams_conf** | /etc/aralez/upstreams.yaml | Location of the upstreams file |
| **access_log** | access | Configure access logging. Values: `access, error` | | **access_log** | access | Configure access logging. Values: `access, error` |
| **log_level** | info | Log level: `info`, `warn`, `error`, `debug`, `trace`, `off` | | **log_level** | info | Log level: `info`, `warn`, `error`, `debug`, `trace`, `off` |
| **log_file** | /full/path/to/aralez.log | Optional, the location of log file. If thi entry does not exist logs will be emitted to stdout. | | **log_file** | /tmp/aralez.log, 20, 5, compress | Optional, the location of log file, file size in megabytes, number of log files to keep, compress files .valueas are separated by comma. Defaults to None, 100, 5, No |
| **hc_method** | HEAD | Healthcheck method: HEAD, GET, POST (UPPERCASE) | | **hc_method** | HEAD | Healthcheck method: HEAD, GET, POST (UPPERCASE) |
| **hc_interval** | 2 | Interval for health checks in seconds | | **hc_interval** | 2 | Interval for health checks in seconds |
| **file_server_folder** | /some/local/folder | Optional. Local folder to serve | | **file_server_folder** | /some/local/folder | Optional. Local folder to serve |
| **file_server_address** | 127.0.0.1:3002 | Optional. Local address for file server | | **file_server_address** | 127.0.0.1:3002 | Optional. Local address for file server |
| **config_api_enabled** | true | Enable/disable remote config push capability | | **config_api_enabled** | true | Enable/disable remote config push capability |
| **cache_size_mb** | 50 | Optional. Enable internal cache with size in megabytes of value. Cache is entirely disabled when this line is commented out. |
| **cache_ttl** | 10 | Optional. Set Default TTL in seconds for cached items, works if cache control headers are not set via upstream. Defaults to 60 if cache_size_mb is set. |
--- ---
+3 -1
View File
@@ -15,11 +15,13 @@ upstreams_conf: /opt/aralez/etc/upstreams.yaml # the location of upstreams file
file_server_folder: /opt/aralez/public # Optional, local folder to serve file_server_folder: /opt/aralez/public # Optional, local folder to serve
file_server_address: 0.0.0.0:3002 # Optional, Local address for file server. Can set as upstream for public access. file_server_address: 0.0.0.0:3002 # Optional, Local address for file server. Can set as upstream for public access.
log_level: info # info, warn, error, debug, trace, off log_level: info # info, warn, error, debug, trace, off
log_file: /tmp/aralez.log # Optional, the location of log file. If this entry does not exist logs will be emitted to stdout. log_file: /tmp/aralez.log, 20, 5, compress # Optional, the location of log file, file size in megabytes, number of log files to keep, compress files .valueas are separated by comma.
access_log: error # all, error, (Off if commented) access_log: error # all, error, (Off if commented)
hc_method: HEAD # Healthcheck method (HEAD, GET, POST are supported) UPPERCASE hc_method: HEAD # Healthcheck method (HEAD, GET, POST are supported) UPPERCASE
hc_interval: 2 #Interval for health checks in seconds hc_interval: 2 #Interval for health checks in seconds
tcp_keepalive_idle: 60 # Seconds of inactivity before the kernel starts sending keepalive probes to a downstream client tcp_keepalive_idle: 60 # Seconds of inactivity before the kernel starts sending keepalive probes to a downstream client
tcp_keepalive_interval: 10 # Seconds between individual keepalive probes if the client does not respond tcp_keepalive_interval: 10 # Seconds between individual keepalive probes if the client does not respond
tcp_keepalive_count: 5 # Number of unanswered probes before the kernel declares the connection dead and closes it tcp_keepalive_count: 5 # Number of unanswered probes before the kernel declares the connection dead and closes it
cache_size_mb: 50 # Optional. Enable internal cache with size in megabytes of value. Cache is entirely disabled when this line is commented out
cache_ttl: 120 # Optional. Set Default TTL in seconds for cached items, works if cache control headers are not set via upstream. Defaults to 60 if cache_size_mb is set
+8
View File
@@ -1,7 +1,11 @@
use dashmap::DashMap; use dashmap::DashMap;
use moka::sync::Cache; use moka::sync::Cache;
use pingora_cache::eviction::simple_lru::Manager;
use pingora_cache::lock::CacheLock;
use pingora_cache::MemCache;
use pingora_limits::rate::Rate; use pingora_limits::rate::Rate;
use std::net::IpAddr; use std::net::IpAddr;
use std::sync::OnceLock;
use std::sync::{Arc, LazyLock}; use std::sync::{Arc, LazyLock};
use std::time::Duration; use std::time::Duration;
@@ -9,3 +13,7 @@ pub static REVERSE_STORE: LazyLock<DashMap<String, String>> = LazyLock::new(Dash
pub static RATE_LIMITER: LazyLock<Rate> = LazyLock::new(|| Rate::new(Duration::from_secs(1))); pub static RATE_LIMITER: LazyLock<Rate> = LazyLock::new(|| Rate::new(Duration::from_secs(1)));
pub static REQUESTS_4XX: LazyLock<Cache<IpAddr, u32>> = LazyLock::new(|| Cache::builder().time_to_live(Duration::from_secs(1)).build()); pub static REQUESTS_4XX: LazyLock<Cache<IpAddr, u32>> = LazyLock::new(|| Cache::builder().time_to_live(Duration::from_secs(1)).build());
pub static LOCALHOST: LazyLock<Arc<str>> = LazyLock::new(|| Arc::from("localhost")); pub static LOCALHOST: LazyLock<Arc<str>> = LazyLock::new(|| Arc::from("localhost"));
pub static MEM_CACHE: LazyLock<MemCache> = LazyLock::new(MemCache::new);
pub static CACHE_LOCK: LazyLock<CacheLock> = LazyLock::new(|| CacheLock::new(Duration::from_secs(30)));
pub static EVICTION: OnceLock<Manager> = OnceLock::new();
pub static CACHE_TTL: OnceLock<u64> = OnceLock::new();
+14
View File
@@ -1,3 +1,5 @@
use crate::utils::lazylock::EVICTION;
use pingora_cache::eviction::EvictionManager;
use pingora_http::Method; use pingora_http::Method;
use pingora_http::StatusCode; use pingora_http::StatusCode;
use pingora_http::Version; use pingora_http::Version;
@@ -20,6 +22,11 @@ pub static MEMORY_USAGE: LazyLock<IntGauge> = LazyLock::new(|| register_int_gaug
pub static ACTIVE_SESSIONS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").unwrap()); pub static ACTIVE_SESSIONS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").unwrap());
pub static REQUEST_COUNT: LazyLock<IntCounter> = LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").unwrap()); pub static REQUEST_COUNT: LazyLock<IntCounter> = LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").unwrap());
pub static CACHE_SIZE_BYTES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_size_bytes", "Current cache size in bytes").unwrap());
pub static CACHE_ITEMS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_items", "Current number of cached objects").unwrap());
pub static CACHE_EVICTED_BYTES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_bytes_total", "Total bytes evicted from cache").unwrap());
pub static CACHE_EVICTED_ITEMS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_items_total", "Total cache items evicted").unwrap());
pub static RESPONSE_CODES: LazyLock<IntCounterVec> = pub static RESPONSE_CODES: LazyLock<IntCounterVec> =
LazyLock::new(|| register_int_counter_vec!("aralez_responses_total", "Responses grouped by status code", &["status"]).unwrap()); LazyLock::new(|| register_int_counter_vec!("aralez_responses_total", "Responses grouped by status code", &["status"]).unwrap());
@@ -55,6 +62,13 @@ pub fn calc_metrics(metric_types: &MetricTypes) {
REQUESTS_BY_METHOD.with_label_values(&[metric_types.method.as_str()]).inc(); REQUESTS_BY_METHOD.with_label_values(&[metric_types.method.as_str()]).inc();
REQUESTS_BY_UPSTREAM.with_label_values(&[metric_types.upstream.as_ref()]).inc(); REQUESTS_BY_UPSTREAM.with_label_values(&[metric_types.upstream.as_ref()]).inc();
RESPONSE_LATENCY.observe(metric_types.latency.as_secs_f64()); 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);
}
} }
pub(crate) fn get_memory_usage() -> usize { pub(crate) fn get_memory_usage() -> usize {
+1 -38
View File
@@ -3,14 +3,9 @@ use crate::utils::lazylock::REVERSE_STORE;
use crate::utils::state::{is_first_run, mark_not_first_run}; use crate::utils::state::{is_first_run, mark_not_first_run};
use crate::utils::structs::*; use crate::utils::structs::*;
use crate::utils::tools::{clone_dashmap, clone_dashmap_into, print_upstreams}; use crate::utils::tools::{clone_dashmap, clone_dashmap_into, print_upstreams};
use crate::web::logging::log_builder;
use dashmap::DashMap; use dashmap::DashMap;
use log::LevelFilter;
use log::{error, info, warn}; use log::{error, info, warn};
use log4rs::{
append::{console::ConsoleAppender, file::FileAppender},
config::{Appender, Config as Log4rsConfig, Root},
encode::pattern::PatternEncoder,
};
use std::collections::HashMap; use std::collections::HashMap;
use std::path::Path; use std::path::Path;
use std::sync::atomic::AtomicUsize; use std::sync::atomic::AtomicUsize;
@@ -327,35 +322,3 @@ pub fn build_headers(path_config: &Option<Vec<String>>, _config: &Configuration,
} }
} }
} }
fn log_builder(conf: &AppConfig, location: &Option<String>) {
let log_level = match conf.log_level.as_str() {
"info" => LevelFilter::Info,
"error" => LevelFilter::Error,
"warn" => LevelFilter::Warn,
"debug" => LevelFilter::Debug,
"trace" => LevelFilter::Trace,
"off" => LevelFilter::Off,
_ => {
println!("Error reading log level, defaulting to: INFO");
LevelFilter::Info
}
};
// let pattern = "{d(%Y-%m-%d %H:%M:%S)} {l} {t} - {m}{n}";
let pattern = "{d(%Y-%m-%d %H:%M:%S)} {l} {t} - {m}\n";
if let Some(location) = location {
let file = FileAppender::builder().encoder(Box::new(PatternEncoder::new(pattern))).build(location).unwrap();
let config = Log4rsConfig::builder()
.appender(Appender::builder().build("file", Box::new(file)))
.build(Root::builder().appender("file").build(log_level))
.unwrap();
log4rs::init_config(config).unwrap();
} else {
let stdout = ConsoleAppender::builder().encoder(Box::new(PatternEncoder::new(pattern))).build();
let config = Log4rsConfig::builder()
.appender(Appender::builder().build("stdout", Box::new(stdout)))
.build(Root::builder().appender("stdout").build(log_level))
.unwrap();
log4rs::init_config(config).unwrap();
}
}
+2
View File
@@ -110,6 +110,8 @@ pub struct AppConfig {
pub hc_method: String, pub hc_method: String,
pub upstreams_conf: String, pub upstreams_conf: String,
pub log_level: String, pub log_level: String,
pub cache_size_mb: Option<usize>,
pub cache_ttl: Option<u64>,
pub access_log: Option<String>, pub access_log: Option<String>,
pub pid_file: Option<String>, pub pid_file: Option<String>,
pub master_key: Option<String>, pub master_key: Option<String>,
+83 -4
View File
@@ -1,5 +1,16 @@
use crate::utils::metrics::LOGGING_ERRORS; use crate::utils::metrics::LOGGING_ERRORS;
use log::info; use crate::utils::structs::AppConfig;
use log::{info, LevelFilter};
use log4rs::append::rolling_file::policy::compound::roll::fixed_window::FixedWindowRoller;
use log4rs::append::rolling_file::policy::compound::trigger::size::SizeTrigger;
use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
use log4rs::append::rolling_file::RollingFileAppender;
use log4rs::{
append::console::ConsoleAppender,
config::{Appender, Config as Log4rsConfig, Root},
encode::pattern::PatternEncoder,
};
use pingora_cache::CachePhase;
use pingora_http::Version; use pingora_http::Version;
use pingora_proxy::Session; use pingora_proxy::Session;
use std::net::{IpAddr, Ipv4Addr}; use std::net::{IpAddr, Ipv4Addr};
@@ -13,11 +24,68 @@ pub struct LogMessage {
pub client_ip: IpAddr, pub client_ip: IpAddr,
pub version: Version, pub version: Version,
pub user_agent: String, pub user_agent: String,
pub cache_status: CachePhase,
} }
static LOG_SENDER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new(); static LOG_SENDER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
static ACCESS_LOG: OnceLock<LogLevel> = OnceLock::new(); static ACCESS_LOG: OnceLock<LogLevel> = OnceLock::new();
const LOG_BUFFER: usize = 16384; const LOG_BUFFER: usize = 16384;
pub fn log_builder(conf: &AppConfig, location: &Option<String>) {
let log_level = match conf.log_level.as_str() {
"info" => LevelFilter::Info,
"error" => LevelFilter::Error,
"warn" => LevelFilter::Warn,
"debug" => LevelFilter::Debug,
"trace" => LevelFilter::Trace,
"off" => LevelFilter::Off,
_ => {
println!("Error reading log level, defaulting to: INFO");
LevelFilter::Info
}
};
let pattern = "{d(%Y-%m-%d %H:%M:%S)} {l} {t} - {m}\n";
if let Some(location) = location {
let parts: Vec<&str> = location.splitn(4, ',').map(|s| s.trim()).collect();
let path = parts.get(0).expect("Syntax error, could not get path for log files");
let compress = parts.get(3).unwrap_or(&"No");
let size_mb: u64 = parts.get(1).and_then(|s| s.parse().ok()).unwrap_or(100);
let keep: u32 = parts.get(2).and_then(|s| s.parse().ok()).unwrap_or(5);
let pattern_str = match compress {
&"compress" => format!("{}.{{}}.gz", path),
_ => format!("{}.{{}}", path),
};
let roller = FixedWindowRoller::builder().build(pattern_str.as_str(), keep).unwrap();
let trigger = SizeTrigger::new(size_mb * 1024 * 1024);
let policy = CompoundPolicy::new(Box::new(trigger), Box::new(roller));
let file = RollingFileAppender::builder()
.encoder(Box::new(PatternEncoder::new(pattern)))
.build(path, Box::new(policy))
.unwrap();
let config = Log4rsConfig::builder()
.appender(Appender::builder().build("file", Box::new(file)))
.build(Root::builder().appender("file").build(log_level))
.unwrap();
log4rs::init_config(config).unwrap();
info!("Logging to: {}, Max file size: {}mb, Files to keep: {}, compression: {} ", path, size_mb, keep, compress);
} else {
let stdout = ConsoleAppender::builder().encoder(Box::new(PatternEncoder::new(pattern))).build();
let config = Log4rsConfig::builder()
.appender(Appender::builder().build("stdout", Box::new(stdout)))
.build(Root::builder().appender("stdout").build(log_level))
.unwrap();
log4rs::init_config(config).unwrap();
info!("No files are configured, logging to stdout");
}
}
pub fn init_access_log(level_str: &str) { pub fn init_access_log(level_str: &str) {
let level = LogLevel::from_str(level_str); let level = LogLevel::from_str(level_str);
let _ = ACCESS_LOG.set(level); let _ = ACCESS_LOG.set(level);
@@ -60,13 +128,19 @@ pub fn access_log(response_code: u16, summary: &str, session: &Session) {
.unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST)); .unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST));
let user_agent = session.req_header().headers.get("user-agent").and_then(|v| v.to_str().ok()).unwrap_or("-"); let user_agent = session.req_header().headers.get("user-agent").and_then(|v| v.to_str().ok()).unwrap_or("-");
// let cache_status = match session.cache.phase() {
// CachePhase::Hit => "Hit",
// CachePhase::Miss => "Miss",
// CachePhase::Expired => "Expired",
// _ => "Disabled",
// };
let log = LogMessage { let log = LogMessage {
response_code, response_code,
summary: summary.to_owned(), summary: summary.to_owned(),
client_ip: ip, client_ip: ip,
version: session.req_header().version, version: session.req_header().version,
user_agent: user_agent.to_owned(), user_agent: user_agent.to_owned(),
cache_status: session.cache.phase(),
}; };
if let Some(sender) = LOG_SENDER.get() { if let Some(sender) = LOG_SENDER.get() {
@@ -90,8 +164,13 @@ pub fn init_logging(enabled: Option<String>) {
pub fn log_receiver(mut receiver: mpsc::Receiver<LogMessage>) { pub fn log_receiver(mut receiver: mpsc::Receiver<LogMessage>) {
while let Some(msg) = receiver.blocking_recv() { while let Some(msg) = receiver.blocking_recv() {
info!( info!(
"{}, {}, client: {}, version: {:?}, useragent: {}", "{}, {}, {}, client: {}, version: {:?}, useragent: {}",
msg.response_code, msg.summary, msg.client_ip, msg.version, msg.user_agent, msg.response_code,
msg.cache_status.as_str(),
msg.summary,
msg.client_ip,
msg.version,
msg.user_agent,
); );
} }
} }
+41 -4
View File
@@ -1,5 +1,5 @@
use crate::utils::auth::authenticate; use crate::utils::auth::authenticate;
use crate::utils::lazylock::{LOCALHOST, RATE_LIMITER, REQUESTS_4XX, REVERSE_STORE}; use crate::utils::lazylock::{CACHE_LOCK, CACHE_TTL, EVICTION, LOCALHOST, MEM_CACHE, RATE_LIMITER, REQUESTS_4XX, REVERSE_STORE};
use crate::utils::metrics::*; use crate::utils::metrics::*;
use crate::utils::structs::{AppConfig, Extraparams, Headers, InnerMap, UpstreamsDashMap, UpstreamsIdMap}; use crate::utils::structs::{AppConfig, Extraparams, Headers, InnerMap, UpstreamsDashMap, UpstreamsIdMap};
use crate::web::gethosts::{GetHost, GetHostsReturHeaders}; use crate::web::gethosts::{GetHost, GetHostsReturHeaders};
@@ -8,9 +8,11 @@ use arc_swap::ArcSwap;
use async_trait::async_trait; use async_trait::async_trait;
use axum::body::Bytes; use axum::body::Bytes;
use log::error; use log::error;
use pingora::http::{RequestHeader, ResponseHeader, StatusCode}; use pingora::http::{Method, RequestHeader, ResponseHeader, StatusCode};
use pingora::prelude::*; use pingora::prelude::*;
use pingora::ErrorSource::Upstream; use pingora::ErrorSource::Upstream;
use pingora_cache::cache_control::CacheControl;
use pingora_cache::{CacheKey, CacheMetaDefaults, RespCacheable};
use pingora_core::listeners::ALPN; use pingora_core::listeners::ALPN;
use pingora_core::prelude::HttpPeer; use pingora_core::prelude::HttpPeer;
use pingora_proxy::{ProxyHttp, Session}; use pingora_proxy::{ProxyHttp, Session};
@@ -18,10 +20,10 @@ use sha2::{Digest, Sha256};
use std::cell::RefCell; use std::cell::RefCell;
use std::fmt::Write; use std::fmt::Write;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration;
use tokio::time::Instant; use tokio::time::Instant;
thread_local! {static IP_BUFFER: RefCell<String> = RefCell::new(String::with_capacity(50));} thread_local! {static IP_BUFFER: RefCell<String> = RefCell::new(String::with_capacity(50));}
#[derive(Clone)] #[derive(Clone)]
pub struct LB { pub struct LB {
pub ump_upst: Arc<UpstreamsDashMap>, pub ump_upst: Arc<UpstreamsDashMap>,
@@ -31,6 +33,7 @@ pub struct LB {
pub server_headers: Arc<Headers>, pub server_headers: Arc<Headers>,
pub config: Arc<AppConfig>, pub config: Arc<AppConfig>,
pub extraparams: Arc<ArcSwap<Extraparams>>, pub extraparams: Arc<ArcSwap<Extraparams>>,
pub cache_enabled: bool,
} }
pub struct Context { pub struct Context {
@@ -57,6 +60,34 @@ impl ProxyHttp for LB {
x4xx_limit: None, x4xx_limit: None,
} }
} }
fn cache_key_callback(&self, session: &Session, _ctx: &mut Self::CTX) -> Result<CacheKey> {
Ok(CacheKey::new(_ctx.hostname.clone().unwrap_or_default().as_ref(), session.req_header().uri.to_string(), ""))
}
fn request_cache_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<()> {
if self.cache_enabled && session.req_header().method == Method::GET {
if let Some(eviction) = EVICTION.get() {
session.cache.enable(&*MEM_CACHE, Some(eviction), None, Some(&*CACHE_LOCK), None);
}
}
Ok(())
}
fn response_cache_filter(&self, _session: &Session, resp: &ResponseHeader, _ctx: &mut Self::CTX) -> Result<RespCacheable> {
let cc = CacheControl::from_resp_headers(resp);
let defaults = CacheMetaDefaults::new(
|status| {
if status == StatusCode::OK {
Some(Duration::from_secs(*CACHE_TTL.get().unwrap_or(&60)))
} else {
None
}
},
0,
0,
);
Ok(pingora_cache::filters::resp_cacheable(cc.as_ref(), resp.clone(), false, &defaults))
}
async fn request_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<bool> { async fn request_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<bool> {
ACTIVE_SESSIONS.inc(); ACTIVE_SESSIONS.inc();
let hostname = return_header_host_from_upstream(session, &self.ump_upst); let hostname = return_header_host_from_upstream(session, &self.ump_upst);
@@ -159,7 +190,6 @@ impl ProxyHttp for LB {
Some(hostname) => match ctx.upstream_peer.as_ref() { Some(hostname) => match ctx.upstream_peer.as_ref() {
Some(innermap) => { Some(innermap) => {
let mut peer = Box::new(HttpPeer::new((&*innermap.address, innermap.port), innermap.is_ssl, hostname.to_string())); let mut peer = Box::new(HttpPeer::new((&*innermap.address, innermap.port), innermap.is_ssl, hostname.to_string()));
if innermap.is_http2 { if innermap.is_http2 {
peer.options.alpn = ALPN::H2; peer.options.alpn = ALPN::H2;
} }
@@ -308,6 +338,13 @@ impl ProxyHttp for LB {
} }
} }
} }
// match session.cache.phase() {
// CachePhase::Hit => info!("Cache Status: HIT (Served from Memory)"),
// CachePhase::Miss => info!("Cache Status: MISS (Fetched from Upstream)"),
// CachePhase::Expired => info!("Cache Status: EXPIRED (Revalidating)"),
// _ => {}
// }
access_log(response_code, &self.request_summary(session, ctx), session); access_log(response_code, &self.request_summary(session, ctx), session);
} }
} }
+10
View File
@@ -2,6 +2,7 @@
use crate::tls::grades; use crate::tls::grades;
use crate::tls::load; use crate::tls::load;
use crate::tls::load::CertificateConfig; use crate::tls::load::CertificateConfig;
use crate::utils::lazylock::{CACHE_TTL, EVICTION};
use crate::utils::structs::Extraparams; use crate::utils::structs::Extraparams;
use crate::utils::tools::*; use crate::utils::tools::*;
use crate::web::logging::init_access_log; use crate::web::logging::init_access_log;
@@ -10,6 +11,7 @@ use arc_swap::ArcSwap;
use dashmap::DashMap; use dashmap::DashMap;
use log::info; use log::info;
use pingora::tls::ssl::{SslAlert, SslRef}; use pingora::tls::ssl::{SslAlert, SslRef};
use pingora_cache::eviction::simple_lru::Manager;
use pingora_core::listeners::tls::TlsSettings; use pingora_core::listeners::tls::TlsSettings;
use pingora_core::listeners::TcpSocketOptions; use pingora_core::listeners::TcpSocketOptions;
use pingora_core::prelude::{background_service, Opt}; use pingora_core::prelude::{background_service, Opt};
@@ -32,6 +34,13 @@ pub fn run() {
let file = parameters.conf.clone().unwrap(); let file = parameters.conf.clone().unwrap();
let maincfg = crate::utils::parceyaml::parce_main_config(file.as_str()); let maincfg = crate::utils::parceyaml::parce_main_config(file.as_str());
let cache_enabled = maincfg.cache_size_mb.is_some();
CACHE_TTL.set(maincfg.cache_ttl.unwrap_or(60)).unwrap();
if let Some(size) = maincfg.cache_size_mb {
let cache_size = size * 1024 * 1024;
drop(EVICTION.set(Manager::new(cache_size)));
}
let mut server = Server::new(parameters).unwrap(); let mut server = Server::new(parameters).unwrap();
server.bootstrap(); server.bootstrap();
@@ -59,6 +68,7 @@ pub fn run() {
client_headers: ch_config, client_headers: ch_config,
server_headers: sh_config, server_headers: sh_config,
extraparams: ec_config, extraparams: ec_config,
cache_enabled: cache_enabled,
}; };
let al = cfg.access_log.clone().unwrap_or("none".to_string()); let al = cfg.access_log.clone().unwrap_or("none".to_string());
init_access_log(al.as_str()); init_access_log(al.as_str());
+12 -2
View File
@@ -70,8 +70,18 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
let static_listen = port_is_available("File Server", &address).await; let static_listen = port_is_available("File Server", &address).await;
let static_files = ServeDir::new(folder); let static_files = ServeDir::new(folder);
let static_serve: Router = Router::new().fallback_service(static_files); let static_serve: Router = Router::new().fallback_service(static_files);
// drop(tokio::spawn(async move { axum::serve(static_listen, static_serve).await.unwrap() })); info!("Starting file server on: {}", address);
static_handle = Some(tokio::spawn(async move { axum::serve(static_listen, static_serve).await.unwrap() }))
static_handle = Some(tokio::spawn(async move {
match axum::serve(static_listen, static_serve).await {
Ok(o) => info!("File server exited normally: {:?}", o),
Err(e) => error!("File server exited: {}", e),
}
}))
// let a = config.file_server_address.to_owned();
// let f = config.file_server_folder.to_owned();
// drop(tokio::spawn(async move { run_file_server(a, f).await }));
} }
let listener = port_is_available("Config API", &config.address).await; let listener = port_is_available("Config API", &config.address).await;