Optimize structured logging.

Cargo.lock

@@ -130,7 +130,7 @@ checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3"
 
 [[package]]
 name = "aralez"
-version = "0.93.5"
+version = "0.94.1"
 dependencies = [
  "ahash",
  "anyhow",

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 {
+        tokio::spawn(async move { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await });
-            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 });
-        drop(tokio::spawn(async move { calc_cache_metrics().await }));
-        drop(tokio::spawn(async move { refresh_order(certdir, confdir).await }));
 
         init_access_logging(self.config.access_log.clone()).await;
         loop {

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<AppState>, Query(params): Query<HashMap<String, S
     match parsed {
         Ok(_) => {
             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()
         }

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<String> = RefCell::new(String::with_capacity(512));
 }
 
-static LOG_SENDER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
+static ACCESS_LOG_SENDER: OnceLock<mpsc::Sender<LogMessage>> = OnceLock::new();
-pub(crate) static PINGORA_LOG_SENDER: OnceLock<mpsc::Sender<StructuredSystemLog>> = OnceLock::new();
+pub(crate) static SYSTEM_LOG_SENDER: OnceLock<mpsc::Sender<StructuredSystemLog>> = OnceLock::new();
 static ACCESS_LOG: OnceLock<LogLevel> = OnceLock::new();
 const LOG_BUFFER: usize = 16384;
 static IS_STRUCTURED: OnceLock<bool> = OnceLock::new();
@@ -120,7 +120,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
             let _ = set_backend(backend_name.to_string());
             init_structured_log().await;
         }
-        let pingora_appender: Option<Box<dyn Append>> = PINGORA_LOG_SENDER
+        let syslog_appender: Option<Box<dyn Append>> = SYSTEM_LOG_SENDER
             .get()
             .cloned()
             .map(|sender| Box::new(StructuredChannelAppender::new(sender)) as Box<dyn Append>);
@@ -129,16 +129,16 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
 
         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))
+                .appender(Appender::builder().build("syslog_appender", appender))
-                .logger(Logger::builder().appender("pingora_channel").additive(false).build("pingora_proxy", log_level))
+                .logger(Logger::builder().appender("syslog_appender").additive(false).build("pingora_proxy", log_level))
-                .logger(Logger::builder().appender("pingora_channel").additive(false).build("pingora_core", log_level))
+                .logger(Logger::builder().appender("syslog_appender").additive(false).build("pingora_core", log_level))
-                .logger(Logger::builder().appender("pingora_channel").additive(false).build("pingora_pool", log_level))
+                .logger(Logger::builder().appender("syslog_appender").additive(false).build("pingora_pool", log_level))
-                .logger(Logger::builder().appender("pingora_channel").additive(false).build("pingora_cache", log_level))
+                .logger(Logger::builder().appender("syslog_appender").additive(false).build("pingora_cache", log_level))
-                .logger(Logger::builder().appender("pingora_channel").additive(false).build("hyper_util", log_level))
+                .logger(Logger::builder().appender("syslog_appender").additive(false).build("hyper_util", log_level))
-                .logger(Logger::builder().appender("pingora_channel").additive(false).build("aralez", log_level))
+                .logger(Logger::builder().appender("syslog_appender").additive(false).build("aralez", log_level))
-                .logger(Logger::builder().appender("pingora_channel").additive(false).build("h2", 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<String>) {
         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);
+        let _ = ACCESS_LOG_SENDER.set(ltx);
-        drop(tokio::spawn(async move { access_log_receiver(lrx).await }));
+        tokio::spawn(async move { access_log_receiver(lrx).await });
     }
 }
 
 pub async fn init_structured_log() {
-    let (tx, mut rx) = mpsc::channel::<StructuredSystemLog>(LOG_BUFFER);
+    let (tx, rx) = mpsc::channel::<StructuredSystemLog>(LOG_BUFFER);
-    let _ = PINGORA_LOG_SENDER.set(tx);
+    let _ = SYSTEM_LOG_SENDER.set(tx);
     let backend = get_backend();
-
+    // Start backend receiver directly without double-channel forwarding
-    drop(tokio::spawn(async move {
+    start_logging_backend(backend, rx);
-        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");
 }
 
 pub async fn access_log_receiver(mut receiver: mpsc::Receiver<LogMessage>) {

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::{Duration, SystemTime};
-use std::time::SystemTime;
 use tokio::sync::mpsc;
 use tokio::sync::OnceCell;
 
 static CLIENT: OnceCell<Elasticsearch> = OnceCell::const_new();
+
 #[derive(Debug)]
 struct ElasticHost {
     url: &'static str,
@@ -33,6 +33,7 @@ struct ElasticSearchConfig {
 }
 
 static ELASTIC: LazyLock<ElasticSearchConfig> = 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<mpsc::UnboundedSender<StructuredSystemLog>> = LazyLock::new(|| {
-    let (tx, mut rx) = mpsc::unbounded_channel::<StructuredSystemLog>();
-    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) {
+    async fn run(&self, mut rx: mpsc::Receiver<StructuredSystemLog>) {
-        let _ = LOG_CHANNEL.send(msg.clone());
+        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<StructuredSystemLog>, buffer: &mut BytesMut, hostname: &str) {
+async fn flush_to_es(batch: &mut Vec<StructuredSystemLog>, 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(&index_header);
+
-        // buffer.extend_from_slice(b"{\"index\":{\"_index\":\"logs\"}}\n");
         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<StructuredSystemLog>, 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;
 

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) {
+    async fn run(&self, mut rx: mpsc::Receiver<StructuredSystemLog>) {
-        // Here comes the backend logic
+        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",

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<StructuredSystemLog>);
 }
 
 pub struct LogBackendPlugin {
@@ -23,9 +24,12 @@ static BACKENDS: LazyLock<HashMap<&'static str, Box<dyn WriteLog>>> = LazyLock::
     map
 });
 
-pub async fn sendlog(backend: &str, msg: &StructuredSystemLog) {
+pub fn start_logging_backend(backend: &str, rx: mpsc::Receiver<StructuredSystemLog>) {
     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);
     }