diff --git a/Cargo.lock b/Cargo.lock index 1e3e94b..2497e93 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -130,7 +130,7 @@ checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "aralez" -version = "0.93.5" +version = "0.94.1" dependencies = [ "ahash", "anyhow", diff --git a/src/core/bgservice.rs b/src/core/bgservice.rs index 608508d..bce0787 100644 --- a/src/core/bgservice.rs +++ b/src/core/bgservice.rs @@ -73,17 +73,14 @@ impl BackgroundService for LB { current_upstreams: self.ump_upst.clone(), full_upstreams: self.ump_full.clone(), }; - drop(tokio::spawn(async move { api_load.start(tx_api).await })); - + tokio::spawn(async move { api_load.start(tx_api).await }); let uu = self.ump_upst.clone(); let ff = self.ump_full.clone(); let im = self.ump_byid.clone(); let (hc_method, hc_interval) = (self.config.hc_method.clone(), self.config.hc_interval); - drop(tokio::spawn(async move { - healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await - })); - drop(tokio::spawn(async move { calc_cache_metrics().await })); - drop(tokio::spawn(async move { refresh_order(certdir, confdir).await })); + tokio::spawn(async move { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await }); + tokio::spawn(async move { calc_cache_metrics().await }); + tokio::spawn(async move { refresh_order(certdir, confdir).await }); init_access_logging(self.config.access_log.clone()).await; loop { diff --git a/src/core/webserver.rs b/src/core/webserver.rs index c803b55..0eaa216 100644 --- a/src/core/webserver.rs +++ b/src/core/webserver.rs @@ -78,10 +78,6 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende 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; @@ -114,9 +110,9 @@ async fn conf(State(st): State, Query(params): Query { if let Some(_) = params.get("save") { - drop(tokio::spawn(async move { apply_config(content.as_str(), st, true).await })); + tokio::spawn(async move { apply_config(content.as_str(), st, true).await }); } else { - drop(tokio::spawn(async move { apply_config(content.as_str(), st, false).await })); + tokio::spawn(async move { apply_config(content.as_str(), st, false).await }); } Response::builder().status(StatusCode::OK).body(Body::from("Accepted! Applying in background\n")).unwrap() } diff --git a/src/logging/core.rs b/src/logging/core.rs index ec3c54a..7bf67cf 100644 --- a/src/logging/core.rs +++ b/src/logging/core.rs @@ -1,4 +1,4 @@ -use crate::logging::types::sendlog; +use crate::logging::types::start_logging_backend; use crate::utils::metrics::LOGGING_ERRORS; use crate::utils::types::AppConfig; use anyhow::Result; @@ -44,8 +44,8 @@ thread_local! { static LOG_BUF: RefCell = RefCell::new(String::with_capacity(512)); } -static LOG_SENDER: OnceLock> = OnceLock::new(); -pub(crate) static PINGORA_LOG_SENDER: OnceLock> = OnceLock::new(); +static ACCESS_LOG_SENDER: OnceLock> = OnceLock::new(); +pub(crate) static SYSTEM_LOG_SENDER: OnceLock> = OnceLock::new(); static ACCESS_LOG: OnceLock = OnceLock::new(); const LOG_BUFFER: usize = 16384; static IS_STRUCTURED: OnceLock = OnceLock::new(); @@ -120,7 +120,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option) { let _ = set_backend(backend_name.to_string()); init_structured_log().await; } - let pingora_appender: Option> = PINGORA_LOG_SENDER + let syslog_appender: Option> = SYSTEM_LOG_SENDER .get() .cloned() .map(|sender| Box::new(StructuredChannelAppender::new(sender)) as Box); @@ -129,16 +129,16 @@ pub async fn log_builder(conf: &AppConfig, location: &Option) { let mut config_builder = Log4rsConfig::builder().appender(Appender::builder().build("stdout", Box::new(stdout))); - if let Some(appender) = pingora_appender { + if let Some(appender) = syslog_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)); + .appender(Appender::builder().build("syslog_appender", appender)) + .logger(Logger::builder().appender("syslog_appender").additive(false).build("pingora_proxy", log_level)) + .logger(Logger::builder().appender("syslog_appender").additive(false).build("pingora_core", log_level)) + .logger(Logger::builder().appender("syslog_appender").additive(false).build("pingora_pool", log_level)) + .logger(Logger::builder().appender("syslog_appender").additive(false).build("pingora_cache", log_level)) + .logger(Logger::builder().appender("syslog_appender").additive(false).build("hyper_util", log_level)) + .logger(Logger::builder().appender("syslog_appender").additive(false).build("aralez", log_level)) + .logger(Logger::builder().appender("syslog_appender").additive(false).build("h2", log_level)); } else { config_builder = config_builder .logger(Logger::builder().build("pingora_core", LevelFilter::Off)) @@ -290,7 +290,7 @@ pub async fn access_log(response_code: u16, summary: &str, session: &Session) { write_access_log(&msg).await; return; } - if let Some(sender) = LOG_SENDER.get() { + if let Some(sender) = ACCESS_LOG_SENDER.get() { if let Err(_) = sender.try_send(msg) { LOGGING_ERRORS.inc(); } @@ -302,30 +302,17 @@ pub async fn init_access_logging(enabled: Option) { 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); - let _ = LOG_SENDER.set(ltx); - drop(tokio::spawn(async move { access_log_receiver(lrx).await })); + let _ = ACCESS_LOG_SENDER.set(ltx); + tokio::spawn(async move { access_log_receiver(lrx).await }); } } pub async fn init_structured_log() { - let (tx, mut rx) = mpsc::channel::(LOG_BUFFER); - let _ = PINGORA_LOG_SENDER.set(tx); + let (tx, rx) = mpsc::channel::(LOG_BUFFER); + let _ = SYSTEM_LOG_SENDER.set(tx); let backend = get_backend(); - - drop(tokio::spawn(async move { - while let Some(syslog) = rx.recv().await { - sendlog(backend, &syslog).await; - } - })); - - // 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"); + // Start backend receiver directly without double-channel forwarding + start_logging_backend(backend, rx); } pub async fn access_log_receiver(mut receiver: mpsc::Receiver) { diff --git a/src/logging/elasticsearch.rs b/src/logging/elasticsearch.rs index 13a8e6a..9165365 100644 --- a/src/logging/elasticsearch.rs +++ b/src/logging/elasticsearch.rs @@ -10,12 +10,12 @@ use humantime::format_rfc3339; use serde_json::json; use std::env; use std::sync::LazyLock; -use std::time::Duration; -use std::time::SystemTime; +use std::time::{Duration, SystemTime}; use tokio::sync::mpsc; use tokio::sync::OnceCell; static CLIENT: OnceCell = OnceCell::const_new(); + #[derive(Debug)] struct ElasticHost { url: &'static str, @@ -33,6 +33,7 @@ struct ElasticSearchConfig { } static ELASTIC: LazyLock = LazyLock::new(ElasticSearchConfig::load_from_env); + impl ElasticSearchConfig { pub fn load_from_env() -> Self { let raw_hosts = env::var("LOG_ELASTIC_HOSTS").unwrap_or_else(|_| "http://127.0.0.1:9200".to_string()); @@ -90,32 +91,6 @@ impl ElasticSearchConfig { } } -static LOG_CHANNEL: LazyLock> = LazyLock::new(|| { - let (tx, mut rx) = mpsc::unbounded_channel::(); - tokio::spawn(async move { - let mut batch = Vec::with_capacity(ELASTIC.batch_len); - let mut timer = tokio::time::interval(Duration::from_secs(ELASTIC.flush_duration)); - let mut buffer = BytesMut::with_capacity(ELASTIC.buffer_size); - loop { - tokio::select! { - Some(msg) = rx.recv() => { - batch.push(msg); - if batch.len() >= ELASTIC.batch_len { - flush_to_es(&mut batch, &mut buffer, ELASTIC.hostname).await; - } - } - _ = timer.tick() => { - if !batch.is_empty() { - flush_to_es(&mut batch, &mut buffer, ELASTIC.hostname).await; - } - } - } - } - }); - - tx -}); - async fn make_es_pool() -> Elasticsearch { let valid_urls: Vec<&str> = ELASTIC.hosts.iter().map(|target| target.url).collect(); if valid_urls.is_empty() { @@ -138,25 +113,44 @@ async fn make_es_pool() -> Elasticsearch { return client; } } - log::warn!("Elasticsearch pool initialized, but clustser ping returned non-200"); + log::warn!("Elasticsearch pool initialized, but cluster ping returned non-200"); client } + struct ElasticSearch; #[async_trait] impl WriteLog for ElasticSearch { - async fn writelog(&self, msg: &StructuredSystemLog) { - let _ = LOG_CHANNEL.send(msg.clone()); + async fn run(&self, mut rx: mpsc::Receiver) { + let mut batch = Vec::with_capacity(ELASTIC.batch_len); + let mut timer = tokio::time::interval(Duration::from_secs(ELASTIC.flush_duration)); + let mut buffer = BytesMut::with_capacity(ELASTIC.buffer_size); + let index_header = format!("{{\"index\":{{\"_index\":\"{}\"}}}}\n", ELASTIC.index_name).into_bytes(); + + loop { + tokio::select! { + Some(msg) = rx.recv() => { + batch.push(msg); + if batch.len() >= ELASTIC.batch_len { + flush_to_es(&mut batch, &mut buffer, &index_header, ELASTIC.hostname).await; + } + } + _ = timer.tick() => { + if !batch.is_empty() { + flush_to_es(&mut batch, &mut buffer, &index_header, ELASTIC.hostname).await; + } + } + } + } } } -async fn flush_to_es(batch: &mut Vec, buffer: &mut BytesMut, hostname: &str) { +async fn flush_to_es(batch: &mut Vec, buffer: &mut BytesMut, index_header: &[u8], hostname: &str) { buffer.clear(); for msg in batch.drain(..) { - let index_header = format!("{{\"index\":{{\"_index\":\"{}\"}}}}\n", ELASTIC.index_name).into_bytes(); - buffer.extend_from_slice(&index_header); - // buffer.extend_from_slice(b"{\"index\":{\"_index\":\"logs\"}}\n"); + buffer.extend_from_slice(index_header); + let logstash_payload = json!({ "@timestamp": format_rfc3339(SystemTime::now()).to_string(), "@version": "1", @@ -174,9 +168,11 @@ async fn flush_to_es(batch: &mut Vec, buffer: &mut BytesMut buffer.extend_from_slice(b"\n"); } } + if buffer.is_empty() { return; } + let payload = buffer.split().freeze(); let client = CLIENT.get_or_init(make_es_pool).await; diff --git a/src/logging/example.rs b/src/logging/example.rs index 9745b0f..75ceee0 100644 --- a/src/logging/example.rs +++ b/src/logging/example.rs @@ -1,17 +1,19 @@ use crate::logging::core::StructuredSystemLog; use crate::logging::types::{LogBackendPlugin, WriteLog}; use async_trait::async_trait; +use tokio::sync::mpsc; pub struct Example; + #[async_trait] impl WriteLog for Example { - async fn writelog(&self, msg: &StructuredSystemLog) { - // Here comes the backend logic - println!("Sending log Example : {:?}", msg); + async fn run(&self, mut rx: mpsc::Receiver) { + while let Some(msg) = rx.recv().await { + println!("Sending log Example : {:?}", msg); + } } } -// Mandatory with name: matching value of "log_structired" in main.yaml inventory::submit! { LogBackendPlugin { name: "example", diff --git a/src/logging/types.rs b/src/logging/types.rs index b925e43..d3486db 100644 --- a/src/logging/types.rs +++ b/src/logging/types.rs @@ -2,10 +2,11 @@ use crate::logging::core::StructuredSystemLog; use async_trait::async_trait; use std::collections::HashMap; use std::sync::LazyLock; +use tokio::sync::mpsc; #[async_trait] pub trait WriteLog: Send + Sync { - async fn writelog(&self, msg: &StructuredSystemLog); + async fn run(&self, rx: mpsc::Receiver); } pub struct LogBackendPlugin { @@ -23,9 +24,12 @@ static BACKENDS: LazyLock>> = LazyLock:: map }); -pub async fn sendlog(backend: &str, msg: &StructuredSystemLog) { +pub fn start_logging_backend(backend: &str, rx: mpsc::Receiver) { if let Some(logger) = BACKENDS.get(backend) { - logger.writelog(msg).await; + let logger_ref = logger.as_ref(); + tokio::spawn(async move { + logger_ref.run(rx).await; + }); } else { log::warn!("Unsupported logging mechanism: {}", backend); }