From 639365c43b3a8ab26195276b354e434302eaf3ab Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Tue, 22 Sep 2026 18:25:52 +0200 Subject: [PATCH] restructurization of logging subsystem --- Cargo.lock | 1 + Cargo.toml | 1 + src/core/bgservice.rs | 2 +- src/core/mod.rs | 1 - src/core/proxyhttp.rs | 2 +- src/core/start.rs | 2 +- src/lib.rs | 1 + src/{core/logging.rs => logging/core.rs} | 114 ++++++++++++++++++++++- src/logging/elasticsearch.rs | 34 +++++++ src/logging/mod.rs | 3 + src/logging/types.rs | 13 +++ src/utils/parceyaml.rs | 2 +- src/utils/types.rs | 1 + 13 files changed, 168 insertions(+), 9 deletions(-) rename src/{core/logging.rs => logging/core.rs} (61%) create mode 100644 src/logging/elasticsearch.rs create mode 100644 src/logging/mod.rs create mode 100644 src/logging/types.rs diff --git a/Cargo.lock b/Cargo.lock index 2d61ed3..4e18976 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -133,6 +133,7 @@ name = "aralez" version = "0.93.5" dependencies = [ "ahash", + "anyhow", "arc-swap", "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 0d7d64f..823c643 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -64,3 +64,4 @@ kube = { version = "4.2.0", features = ["runtime", "client", "derive"] } k8s-openapi = { version = "0.28.0", features = ["v1_32"] } futures = "0.3.34" lazy_static = "1.5.0" +anyhow = "1.0.103" diff --git a/src/core/bgservice.rs b/src/core/bgservice.rs index 5a66e72..9c0dcdf 100644 --- a/src/core/bgservice.rs +++ b/src/core/bgservice.rs @@ -1,6 +1,6 @@ -use crate::core::logging::init_logging; use crate::core::proxyhttp::LB; use crate::ingress::{APIUpstreamProvider, ConsulProvider, Discovery, FromFileProvider, KubernetesProvider}; +use crate::logging::core::init_logging; use crate::tls::acme::order::refresh_order; use crate::utils::metrics::calc_cache_metrics; use crate::utils::parceyaml::load_configuration; diff --git a/src/core/mod.rs b/src/core/mod.rs index 859d4dd..6172525 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -1,6 +1,5 @@ pub mod bgservice; pub mod gethosts; -pub mod logging; pub mod proxyhttp; pub mod start; pub mod webserver; diff --git a/src/core/proxyhttp.rs b/src/core/proxyhttp.rs index 16f046c..c76d992 100644 --- a/src/core/proxyhttp.rs +++ b/src/core/proxyhttp.rs @@ -1,6 +1,6 @@ use crate::auth::authenticate; use crate::core::gethosts::{GetHost, GetHostsReturHeaders}; -use crate::core::logging::access_log; +use crate::logging::core::access_log; 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::types::{AppConfig, Extraparams, Headers, InnerMap, UpstreamsDashMap, UpstreamsIdMap}; diff --git a/src/core/start.rs b/src/core/start.rs index 697d2a6..a3ceb75 100644 --- a/src/core/start.rs +++ b/src/core/start.rs @@ -1,5 +1,5 @@ -use crate::core::logging::init_access_log; use crate::core::proxyhttp::LB; +use crate::logging::core::init_access_log; use crate::tls::grades; use crate::tls::load; use crate::tls::load::CertificateConfig; diff --git a/src/lib.rs b/src/lib.rs index 14e02d2..7bf3ad1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,6 @@ pub mod auth; pub mod core; pub mod ingress; +pub mod logging; pub mod tls; pub mod utils; diff --git a/src/core/logging.rs b/src/logging/core.rs similarity index 61% rename from src/core/logging.rs rename to src/logging/core.rs index 993acfe..4db96b8 100644 --- a/src/core/logging.rs +++ b/src/logging/core.rs @@ -1,10 +1,13 @@ +use crate::logging::types::sendlog; use crate::utils::metrics::LOGGING_ERRORS; use crate::utils::types::AppConfig; -use log::{error, info, warn, LevelFilter}; +use anyhow::Result; +use log::{error, info, warn, LevelFilter, Record}; 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::Append; use log4rs::config::Logger; use log4rs::{ append::console::ConsoleAppender, @@ -14,6 +17,7 @@ use log4rs::{ use pingora_cache::CachePhase; use pingora_http::Version; use pingora_proxy::Session; +use serde::Serialize; use std::net::{IpAddr, Ipv4Addr}; use std::sync::OnceLock; use tokio::sync::mpsc; @@ -27,10 +31,53 @@ pub struct LogMessage { pub user_agent: String, pub cache_status: CachePhase, } + +#[derive(Debug, Serialize)] +pub struct StructuredSystemLog { + pub target: String, + pub level: log::Level, + pub message: String, +} + static LOG_SENDER: OnceLock> = OnceLock::new(); +pub(crate) static PINGORA_LOG_SENDER: OnceLock> = OnceLock::new(); static ACCESS_LOG: OnceLock = OnceLock::new(); const LOG_BUFFER: usize = 16384; +static LOG_BACKEND: OnceLock<&'static str> = OnceLock::new(); +pub fn set_backend(key: String) -> Result<(), &'static str> { + let static_key: &'static str = Box::leak(key.into_boxed_str()); + LOG_BACKEND.set(static_key).map_err(|_| "LOG_BACKEND was already initialized!") +} + +pub fn get_backend() -> &'static str { + LOG_BACKEND.get().copied().unwrap_or("system") +} + +#[derive(Debug)] +pub struct StructiredChannelAppender { + sender: mpsc::Sender, +} + +impl StructiredChannelAppender { + pub fn new(sender: mpsc::Sender) -> Self { + Self { sender } + } +} + +impl Append for StructiredChannelAppender { + fn append(&self, record: &Record) -> Result<()> { + let msg = StructuredSystemLog { + target: record.target().to_string(), + level: record.level(), + message: format!("{}", record.args()), + }; + let _ = self.sender.try_send(msg); + Ok(()) + } + + fn flush(&self) {} +} pub fn log_builder(conf: &AppConfig, location: &Option) { let log_level = match conf.log_level.as_str() { "info" => LevelFilter::Info, @@ -46,11 +93,53 @@ pub fn log_builder(conf: &AppConfig, location: &Option) { }; let mut pat: String = "{d(%Y-%m-%d %H:%M:%S)} {l} - {m} {n}".to_string(); + if let Some(ptrn) = conf.log_pattern.clone() { pat = ptrn; } let pattern = pat.as_str(); + if let Some(backend) = conf.log_structired.clone() { + let b = backend.split_whitespace().collect::>(); + let s = b[0].to_string(); + let _ = set_backend(s); + init_pingora_syslog(); + + let pingora_appender: Option> = PINGORA_LOG_SENDER + .get() + .cloned() + .map(|sender| Box::new(StructiredChannelAppender::new(sender)) as Box); + + let stdout = ConsoleAppender::builder().encoder(Box::new(PatternEncoder::new(pattern))).build(); + + let mut config_builder = Log4rsConfig::builder().appender(Appender::builder().build("stdout", Box::new(stdout))); + + if let Some(appender) = pingora_appender { + config_builder = config_builder + .appender(Appender::builder().build("pingora_channel", appender)) + .logger(Logger::builder().appender("pingora_channel").additive(false).build("pingora_proxy", log_level)) + .logger(Logger::builder().appender("pingora_channel").additive(false).build("pingora_core", log_level)) + .logger(Logger::builder().appender("pingora_channel").additive(false).build("pingora_pool", log_level)) + .logger(Logger::builder().appender("pingora_channel").additive(false).build("pingora_cache", log_level)) + .logger(Logger::builder().appender("pingora_channel").additive(false).build("hyper_util", log_level)) + .logger(Logger::builder().appender("pingora_channel").additive(false).build("aralez", log_level)) + .logger(Logger::builder().appender("pingora_channel").additive(false).build("h2", log_level)); + } else { + config_builder = config_builder + .logger(Logger::builder().build("pingora_core", LevelFilter::Off)) + .logger(Logger::builder().build("pingora_pool", LevelFilter::Off)) + .logger(Logger::builder().build("pingora_cache", LevelFilter::Off)) + .logger(Logger::builder().build("hyper_util", LevelFilter::Off)) + .logger(Logger::builder().build("aralez", LevelFilter::Off)) + .logger(Logger::builder().build("h2", LevelFilter::Off)); + } + + let config = config_builder.build(Root::builder().appender("stdout").build(log_level)).unwrap(); + + log4rs::init_config(config).unwrap(); + info!("Enabling structured logging with backend : {}", backend); + return; + } if let Some(location) = location { let parts: Vec<&str> = location.splitn(4, ',').map(|s| s.trim()).collect(); @@ -82,7 +171,6 @@ pub fn log_builder(conf: &AppConfig, location: &Option) { 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))) .logger(Logger::builder().build("pingora_proxy", LevelFilter::Off)) @@ -121,6 +209,7 @@ pub enum MatchStatus { Er4xx, Er5xx, } + impl MatchStatus { #[inline] pub fn from_code(code: u16) -> Self { @@ -131,12 +220,15 @@ impl MatchStatus { } } } + pub fn access_log(response_code: u16, summary: &str, session: &Session) { let level = ACCESS_LOG.get().unwrap_or(&LogLevel::None); let status = MatchStatus::from_code(response_code); + let should_log = match level { LogLevel::Access => true, LogLevel::None => false, + // Captures all 4xx and 5xx errors when level is set to Error LogLevel::Error => matches!(status, MatchStatus::Er5xx), }; @@ -151,6 +243,7 @@ pub fn access_log(response_code: u16, summary: &str, session: &Session) { .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 log = LogMessage { response_code, summary: summary.to_owned(), @@ -161,7 +254,6 @@ pub fn access_log(response_code: u16, summary: &str, session: &Session) { }; if let Some(sender) = LOG_SENDER.get() { - let sender = sender; if let Err(_) = sender.try_send(log) { LOGGING_ERRORS.inc(); } @@ -169,7 +261,7 @@ pub fn access_log(response_code: u16, summary: &str, session: &Session) { } pub fn init_logging(enabled: Option) { - if let Some(_) = enabled { + if enabled.is_some() { LOGGING_ERRORS.set(0); info!("Enabling {:?} log, with buffer of {} messages", ACCESS_LOG.get().unwrap_or(&LogLevel::None), LOG_BUFFER); let (ltx, lrx) = mpsc::channel(LOG_BUFFER); @@ -178,6 +270,20 @@ pub fn init_logging(enabled: Option) { } } +pub fn init_pingora_syslog() { + let (tx, mut rx) = mpsc::channel::(LOG_BUFFER); + let _ = PINGORA_LOG_SENDER.set(tx); + let backend = get_backend(); + std::thread::Builder::new() + .name("structured-log-receiver".to_string()) + .spawn(move || { + while let Some(syslog) = rx.blocking_recv() { + sendlog(backend, &syslog); + } + }) + .expect("Failed to spawn log thread"); +} + pub fn log_receiver(mut receiver: mpsc::Receiver) { while let Some(msg) = receiver.blocking_recv() { match MatchStatus::from_code(msg.response_code) { diff --git a/src/logging/elasticsearch.rs b/src/logging/elasticsearch.rs new file mode 100644 index 0000000..3787870 --- /dev/null +++ b/src/logging/elasticsearch.rs @@ -0,0 +1,34 @@ +use crate::logging::core::StructuredSystemLog; +use crate::logging::types::WriteLog; +use std::env; +use std::sync::LazyLock; + +pub struct ElasticSearchConfig { + pub hosts: &'static str, + pub user: &'static str, + pub password: &'static str, +} + +impl ElasticSearchConfig { + pub fn load_from_env() -> Self { + let host = env::var("LOG_ELASTIC_HOSTS").unwrap_or_else(|_| "127.0.0.1".to_string()); + let user = env::var("LOG_ELASTIC_USER").unwrap_or_else(|_| "elastic".to_string()); + let password = env::var("LOG_ELASTIC_PASSWORD").unwrap_or_else(|_| "elastic".to_string()); + + Self { + hosts: Box::leak(host.into_boxed_str()), + user: Box::leak(user.into_boxed_str()), + password: Box::leak(password.into_boxed_str()), + } + } +} +pub static ELASTIC: LazyLock = LazyLock::new(ElasticSearchConfig::load_from_env); +pub struct ElasticSearch; + +impl WriteLog for ElasticSearch { + fn writelog(&self, msg: &StructuredSystemLog) { + if let Ok(jsonmsg) = serde_json::to_string(&msg) { + println!("Connecting to {} - {}:{} => {}", ELASTIC.hosts, ELASTIC.user, ELASTIC.password, jsonmsg); + } + } +} diff --git a/src/logging/mod.rs b/src/logging/mod.rs new file mode 100644 index 0000000..8906b7d --- /dev/null +++ b/src/logging/mod.rs @@ -0,0 +1,3 @@ +pub mod core; +pub mod elasticsearch; +pub mod types; diff --git a/src/logging/types.rs b/src/logging/types.rs new file mode 100644 index 0000000..08dcac4 --- /dev/null +++ b/src/logging/types.rs @@ -0,0 +1,13 @@ +use crate::logging::core::StructuredSystemLog; +use crate::logging::elasticsearch::ElasticSearch; + +pub trait WriteLog { + fn writelog(&self, msg: &StructuredSystemLog); +} +pub fn sendlog(backend: &str, msg: &StructuredSystemLog) { + match backend { + "elasticsearch" => ElasticSearch.writelog(msg), + "somewhereelse" => println!("somewhereelse"), + _ => log::warn!("Unsupported logging mechanism : {}", backend), + } +} diff --git a/src/utils/parceyaml.rs b/src/utils/parceyaml.rs index 37dba8b..03a95ca 100644 --- a/src/utils/parceyaml.rs +++ b/src/utils/parceyaml.rs @@ -1,4 +1,4 @@ -use crate::core::logging::log_builder; +use crate::logging::core::log_builder; use crate::utils::healthcheck; use crate::utils::lazylock::REVERSE_STORE; use crate::utils::state::{is_first_run, mark_not_first_run}; diff --git a/src/utils/types.rs b/src/utils/types.rs index ce45eb3..8d5b2de 100644 --- a/src/utils/types.rs +++ b/src/utils/types.rs @@ -132,6 +132,7 @@ pub struct AppConfig { pub rungroup: Option, pub log_file: Option, pub log_pattern: Option, + pub log_structired: Option, pub tcp_keepalive_idle: Option, pub tcp_keepalive_interval: Option, pub tcp_keepalive_count: Option,