Making custom Elsticsearch logger.

src/core/bgservice.rs

@@ -1,6 +1,6 @@
 use crate::core::proxyhttp::LB;
 use crate::ingress::{APIUpstreamProvider, ConsulProvider, Discovery, FromFileProvider, KubernetesProvider};
-use crate::logging::core::init_access_logging;
+use crate::logging::core::{init_access_log, init_access_logging, log_builder};
 use crate::tls::acme::order::refresh_order;
 use crate::utils::metrics::calc_cache_metrics;
 use crate::utils::parceyaml::load_configuration;
@@ -19,6 +19,10 @@ use tokio::sync::mpsc;
 #[async_trait]
 impl BackgroundService for LB {
     async fn start(&self, mut shutdown: ShutdownWatch) {
+        log_builder(&self.config, &self.config.log_file).await;
+        let al = &self.config.access_log.clone().unwrap_or("none".to_string());
+        let _ = init_access_log(al.as_str()).await;
+
         info!("Starting background service"); // tx: Sender<Configuration>
         let (tx, mut rx) = mpsc::channel::<Configuration>(1);
         let tx_api = tx.clone();
@@ -80,7 +84,8 @@ impl BackgroundService for LB {
         }));
         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());
+
+        init_access_logging(self.config.access_log.clone()).await;
         loop {
             tokio::select! {
                 _ = shutdown.changed() => {

src/core/proxyhttp.rs

@@ -355,7 +355,7 @@ impl ProxyHttp for LB {
                 }
             }
         }
-        access_log(response_code, &self.request_summary(session, ctx), session);
+        access_log(response_code, &self.request_summary(session, ctx), session).await;
     }
 }
 

src/core/start.rs

@@ -1,5 +1,4 @@
 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;
@@ -71,8 +70,8 @@ pub fn run() {
         extraparams: ec_config,
         cache_enabled: cache_enabled,
     };
-    let al = cfg.access_log.clone().unwrap_or("none".to_string());
-    init_access_log(al.as_str());
+    // let al = cfg.access_log.clone().unwrap_or("none".to_string());
+    // init_access_log(al.as_str());
 
     let grade = cfg.proxy_tls_grade.clone().unwrap_or("medium".to_string());
     info!("TLS grade set to: [ {} ]", grade);

src/logging/core.rs

@@ -94,7 +94,7 @@ impl Append for StructuredChannelAppender {
 
     fn flush(&self) {}
 }
-pub fn log_builder(conf: &AppConfig, location: &Option<String>) {
+pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
     let log_level = match conf.log_level.as_str() {
         "info" => LevelFilter::Info,
         "error" => LevelFilter::Error,
@@ -118,7 +118,7 @@ pub fn log_builder(conf: &AppConfig, location: &Option<String>) {
     if let Some(backend) = conf.log_structured.as_deref() {
         if let Some(backend_name) = backend.split_whitespace().next() {
             let _ = set_backend(backend_name.to_string());
-            init_structured_log();
+            init_structured_log().await;
         }
         let pingora_appender: Option<Box<dyn Append>> = PINGORA_LOG_SENDER
             .get()
@@ -215,7 +215,7 @@ pub fn log_builder(conf: &AppConfig, location: &Option<String>) {
     }
 }
 
-pub fn init_access_log(level_str: &str) {
+pub async fn init_access_log(level_str: &str) {
     let level = LogLevel::from_str(level_str);
     let _ = ACCESS_LOG.set(level);
 }
@@ -255,7 +255,7 @@ impl MatchStatus {
     }
 }
 
-pub fn access_log(response_code: u16, summary: &str, session: &Session) {
+pub async 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);
 
@@ -287,7 +287,7 @@ pub fn access_log(response_code: u16, summary: &str, session: &Session) {
     };
 
     if IS_STRUCTURED.get().is_some() {
-        write_access_log(&msg);
+        write_access_log(&msg).await;
         return;
     }
     if let Some(sender) = LOG_SENDER.get() {
@@ -297,37 +297,44 @@ pub fn access_log(response_code: u16, summary: &str, session: &Session) {
     }
 }
 
-pub fn init_access_logging(enabled: Option<String>) {
+pub async fn init_access_logging(enabled: Option<String>) {
     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);
         let _ = LOG_SENDER.set(ltx);
-        std::thread::spawn(move || access_log_receiver(lrx));
+        drop(tokio::spawn(async move { access_log_receiver(lrx).await }));
     }
 }
 
-pub fn init_structured_log() {
+pub async fn init_structured_log() {
     let (tx, mut rx) = mpsc::channel::<StructuredSystemLog>(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");
+
+    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");
 }
 
-pub fn access_log_receiver(mut receiver: mpsc::Receiver<LogMessage>) {
-    while let Some(msg) = receiver.blocking_recv() {
-        write_access_log(&msg);
+pub async fn access_log_receiver(mut receiver: mpsc::Receiver<LogMessage>) {
+    while let Some(msg) = receiver.recv().await {
+        write_access_log(&msg).await;
     }
 }
 
-fn write_access_log(msg: &LogMessage) {
+async fn write_access_log(msg: &LogMessage) {
     match MatchStatus::from_code(msg.response_code) {
         MatchStatus::Ok2xx => info!(
             "{}, {}, {}, client: {}, version: {:?}, useragent: {}",

src/logging/elasticsearch.rs

@@ -1,22 +1,30 @@
 use crate::logging::core::StructuredSystemLog;
 use crate::logging::types::{LogBackendPlugin, WriteLog};
+use crate::utils::hcclient::httpclient;
+use async_trait::async_trait;
+use bytes::Bytes;
 use std::env;
 use std::sync::LazyLock;
 
 pub struct ElasticSearchConfig {
-    pub hosts: &'static str,
+    pub hosts: &'static [&'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 hosts: Vec<&'static str> = env::var("LOG_ELASTIC_HOSTS")
+            .unwrap_or_else(|_| "127.0.0.1".to_string())
+            .split(',')
+            .map(|s| -> &'static str { Box::leak(s.trim().to_string().into_boxed_str()) })
+            .collect();
+
         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()),
+            hosts: Box::leak(hosts.into_boxed_slice()),
             user: Box::leak(user.into_boxed_str()),
             password: Box::leak(password.into_boxed_str()),
         }
@@ -25,13 +33,18 @@ impl ElasticSearchConfig {
 pub static ELASTIC: LazyLock<ElasticSearchConfig> = LazyLock::new(ElasticSearchConfig::load_from_env);
 pub struct ElasticSearch;
 
+#[async_trait]
 impl WriteLog for ElasticSearch {
-    fn writelog(&self, msg: &StructuredSystemLog) {
-        if let Ok(jsonmsg) = serde_json::to_string(&msg) {
-            // ElasticSearch logic is not yet implemented.
-            println!("{} - {}:{} => {}", ELASTIC.hosts, ELASTIC.user, ELASTIC.password, jsonmsg);
-            // let resp = httpclient(method, tls, host, path, upstream.address.as_ref(), upstream.port, link).await;
-        }
+    async fn writelog(&self, msg: &StructuredSystemLog) {
+        let payload = match serde_json::to_vec(&msg) {
+            Ok(vec) => Bytes::from(vec),
+            Err(e) => {
+                log::warn!("Failed to serialize structured system log to JSON bytes: {}", e);
+                return;
+            }
+        };
+        println!("{} => {} => {}:{:?}", ELASTIC.hosts[0], ELASTIC.hosts[1], ELASTIC.user, ELASTIC.password);
+        let _ = httpclient("POST", false, "127.0.0.1", "/i", "localhost", 8000, "http://127.0.0.1:8000/d".to_string(), payload).await;
     }
 }
 

src/logging/example.rs

@@ -1,10 +1,11 @@
 use crate::logging::core::StructuredSystemLog;
 use crate::logging::types::{LogBackendPlugin, WriteLog};
+use async_trait::async_trait;
 
 pub struct Example;
-
+#[async_trait]
 impl WriteLog for Example {
-    fn writelog(&self, msg: &StructuredSystemLog) {
+    async fn writelog(&self, msg: &StructuredSystemLog) {
         // Here comes the backend logic
         println!("Sending log Example : {:?}", msg);
     }

src/logging/types.rs

@@ -1,9 +1,11 @@
 use crate::logging::core::StructuredSystemLog;
+use async_trait::async_trait;
 use std::collections::HashMap;
 use std::sync::LazyLock;
 
+#[async_trait]
 pub trait WriteLog: Send + Sync {
-    fn writelog(&self, msg: &StructuredSystemLog);
+    async fn writelog(&self, msg: &StructuredSystemLog);
 }
 
 pub struct LogBackendPlugin {
@@ -21,9 +23,9 @@ static BACKENDS: LazyLock<HashMap<&'static str, Box<dyn WriteLog>>> = LazyLock::
     map
 });
 
-pub fn sendlog(backend: &str, msg: &StructuredSystemLog) {
+pub async fn sendlog(backend: &str, msg: &StructuredSystemLog) {
     if let Some(logger) = BACKENDS.get(backend) {
-        logger.writelog(msg);
+        logger.writelog(msg).await;
     } else {
         log::warn!("Unsupported logging mechanism: {}", backend);
     }

src/utils/hcclient.rs

@@ -9,7 +9,7 @@ use std::time::Duration;
 
 pub static HC_CONNECTOR: LazyLock<Connector> = LazyLock::new(|| Connector::new(None));
 
-pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address: &str, port: u16, url: String) -> (bool, bool) {
+pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address: &str, port: u16, url: String, payload: Bytes) -> (bool, bool) {
     let method = match method {
         "HEAD" => "HEAD",
         "GET" => "GET",
@@ -24,6 +24,86 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address
         peer.options.verify_hostname = false;
     }
 
+    let mut is_h2 = false;
+    peer.options.total_connection_timeout = Option::from(Duration::from_secs(5));
+    peer.options.alpn = ALPN::H2H1;
+
+    let (mut http_session, _) = match HC_CONNECTOR.get_http_session(&peer).await {
+        Ok(s) => s,
+        Err(e) => {
+            if let Some(msg) = e.context {
+                log::warn!("{}, type: {}", msg.as_str(), e.etype.as_str());
+            } else {
+                log::warn!("Fail to connect to addr: {}, tls {}, host {}", address, tls, host);
+            }
+            return (false, false);
+        }
+    };
+
+    match http_session.as_http2() {
+        Some(_) => {
+            peer.options.alpn = ALPN::H2;
+            is_h2 = true;
+        }
+        None => {
+            if ping_grpc(url.as_str()).await {
+                return (true, true);
+            }
+        }
+    }
+
+    let mut req = match RequestHeader::build(method, path.as_bytes(), None) {
+        Ok(r) => r,
+        Err(e) => {
+            log::warn!("Failed to build request: {}", e);
+            return (false, false);
+        }
+    };
+
+    req.insert_header("Host", host).ok();
+    let body_bytes = if method == "POST" {
+        req.insert_header("Content-Type", "application/json").ok();
+        req.insert_header("Content-Length", payload.len().to_string()).ok();
+        payload
+    } else {
+        Bytes::new()
+    };
+    if let Err(e) = http_session.write_request_header(Box::new(req)).await {
+        log::warn!("Write header failed: {}", e);
+        return (false, false);
+    }
+    if let Err(e) = http_session.write_request_body(body_bytes, true).await {
+        log::warn!("Write body failed: {}", e);
+        return (false, false);
+    }
+    let status = match http_session.read_response_header().await {
+        Ok(_) => http_session.response_header().map(|r| r.status.as_u16()).unwrap_or(500),
+        Err(e) => {
+            if !is_h2 && ping_grpc(url.as_str()).await {
+                return (true, true);
+            }
+            log::warn!("Health Check read failed ({}) : {} - {}", if is_h2 { "H2" } else { "H1" }, host, e);
+            return (false, false);
+        }
+    };
+    HC_CONNECTOR.release_http_session(http_session, &peer, None).await;
+    ((200..500).contains(&status), is_h2)
+}
+/*
+pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address: &str, port: u16, url: String) -> (bool, bool) {
+    let method = match method {
+        "HEAD" => "HEAD",
+        "GET" => "GET",
+        "POST" => "POST",
+        _ => "GET",
+    };
+    let mut peer = HttpPeer::new((address, port), tls, host.to_string());
+
+    if tls {
+        peer.options.verify_cert = false;
+        peer.options.verify_hostname = false;
+    }
+
     let mut is_h2 = false;
     peer.options.total_connection_timeout = Option::from(Duration::from_secs(5));
     peer.options.alpn = ALPN::H2H1;
@@ -93,3 +173,4 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address
     HC_CONNECTOR.release_http_session(http_session, &peer, None).await;
     ((200..500).contains(&status), is_h2)
 }
+*/

src/utils/healthcheck.rs

@@ -2,6 +2,7 @@ use crate::utils::hcclient::httpclient;
 use crate::utils::lazylock::REVERSE_STORE;
 use crate::utils::tools::*;
 use crate::utils::types::{InnerMap, UpstreamsDashMap, UpstreamsIdMap};
+use bytes::Bytes;
 use dashmap::DashMap;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
@@ -9,6 +10,7 @@ use std::time::Duration;
 use tokio::net::TcpStream;
 use tokio::time::{interval, timeout};
 use tokio_native_tls::native_tls::TlsConnector;
+const EMPTY_PAYLOAD: Bytes = Bytes::from_static(b"");
 
 pub async fn hc2(upslist: Arc<UpstreamsDashMap>, fullist: Arc<UpstreamsDashMap>, idlist: Arc<UpstreamsIdMap>, params: (&str, u64)) {
     let mut period = interval(Duration::from_secs(params.1));
@@ -66,7 +68,7 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsD
                     } else {
                         format!("http://{}:{}{}", upstream.address, upstream.port, path)
                     };
-                    let resp = httpclient(method, tls, host, path, upstream.address.as_ref(), upstream.port, link).await;
+                    let resp = httpclient(method, tls, host, path, upstream.address.as_ref(), upstream.port, link, EMPTY_PAYLOAD).await;
                     if resp.0 {
                         if resp.1 {
                             scheme.is_http2 = resp.1;

src/utils/parceyaml.rs

@@ -1,4 +1,3 @@
-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};
@@ -264,8 +263,7 @@ pub fn parce_main_config(path: &str) -> AppConfig {
     if let Ok(jwt_key) = env::var("JWT_KEY") {
         cfo.master_key = Some(jwt_key);
     };
-
-    log_builder(&cfo, &cfo.log_file);
+    // log_builder(&cfo, &cfo.log_file);
     cfo.hc_method = cfo.hc_method.to_uppercase();
     if let Some((ip, port_str)) = cfo.config_address.rsplit_once(':') {
         if let Ok(port) = port_str.parse::<u16>() {