From 3fe037de5a3f864c10aaed3638eefed76ce50cb6 Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Thu, 24 Sep 2026 18:27:53 +0200 Subject: [PATCH] Making custom Elsticsearch logger 2. --- src/logging/core.rs | 46 +++++++++---------------- src/logging/elasticsearch.rs | 67 +++++++++++++++++++++++++++++------- src/utils/hcclient.rs | 6 ++-- src/utils/healthcheck.rs | 2 +- 4 files changed, 75 insertions(+), 46 deletions(-) diff --git a/src/logging/core.rs b/src/logging/core.rs index 1ff7ddc..e077023 100644 --- a/src/logging/core.rs +++ b/src/logging/core.rs @@ -2,7 +2,7 @@ use crate::logging::types::sendlog; use crate::utils::metrics::LOGGING_ERRORS; use crate::utils::types::AppConfig; use anyhow::Result; -use log::{error, info, warn, LevelFilter, Record}; +use log::{error, info, log, Level, 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; @@ -335,33 +335,19 @@ pub async fn access_log_receiver(mut receiver: mpsc::Receiver) { } async fn write_access_log(msg: &LogMessage) { - match MatchStatus::from_code(msg.response_code) { - MatchStatus::Ok2xx => info!( - "{}, {}, {}, client: {}, version: {:?}, useragent: {}", - msg.response_code, - msg.cache_status.as_str(), - msg.summary, - msg.client_ip, - msg.version, - msg.user_agent, - ), - MatchStatus::Er4xx => warn!( - "{}, {}, {}, client: {}, version: {:?}, useragent: {}", - msg.response_code, - msg.cache_status.as_str(), - msg.summary, - msg.client_ip, - msg.version, - msg.user_agent, - ), - MatchStatus::Er5xx => error!( - "{}, {}, {}, client: {}, version: {:?}, useragent: {}", - msg.response_code, - msg.cache_status.as_str(), - msg.summary, - msg.client_ip, - msg.version, - msg.user_agent, - ), - } + let level = match MatchStatus::from_code(msg.response_code) { + MatchStatus::Ok2xx => Level::Info, + MatchStatus::Er4xx => Level::Warn, + MatchStatus::Er5xx => Level::Error, + }; + log!( + level, + "{}, {}, {}, client: {}, version: {:?}, useragent: {}", + msg.response_code, + msg.cache_status.as_str(), + msg.summary, + msg.client_ip, + msg.version, + msg.user_agent, + ); } diff --git a/src/logging/elasticsearch.rs b/src/logging/elasticsearch.rs index bacd16f..ac6e577 100644 --- a/src/logging/elasticsearch.rs +++ b/src/logging/elasticsearch.rs @@ -6,32 +6,73 @@ use bytes::Bytes; use std::env; use std::sync::LazyLock; -pub struct ElasticSearchConfig { - pub hosts: &'static [&'static str], +static ELASTIC: LazyLock = LazyLock::new(ElasticSearchConfig::load_from_env); + +#[derive(Debug)] +struct ElasticHost { + address: &'static str, + port: u16, + tls: bool, + url: &'static str, +} + +struct ElasticSearchConfig { + pub hosts: &'static [ElasticHost], pub user: &'static str, pub password: &'static str, } impl ElasticSearchConfig { pub fn load_from_env() -> Self { - 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 raw_hosts = env::var("LOG_ELASTIC_HOSTS").unwrap_or_else(|_| "http://127.0.0.1:9200".to_string()); + + let mut hosts_struct = Vec::new(); + + for entry in raw_hosts.split(',') { + let entry = entry.trim(); + if entry.is_empty() { + continue; + } + + let (scheme, rest) = if let Some(stripped) = entry.strip_prefix("https://") { + ("https", stripped) + } else if let Some(stripped) = entry.strip_prefix("http://") { + ("http", stripped) + } else { + ("http", entry) + }; + + let (address, port) = match rest.split_once(':') { + Some((host_part, port_part)) => { + let port = port_part.parse::().unwrap_or(9200); + (host_part, port) + } + None => (rest, 9200), + }; + + let tls = scheme == "https"; + let full_url = format!("{}://{}:{}", scheme, address, port); + + hosts_struct.push(ElasticHost { + address: Box::leak(address.to_string().into_boxed_str()), + port, + tls, + url: Box::leak(full_url.into_boxed_str()), // Clean static URL + }); + } 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(hosts.into_boxed_slice()), + hosts: Box::leak(hosts_struct.into_boxed_slice()), 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; + +struct ElasticSearch; #[async_trait] impl WriteLog for ElasticSearch { @@ -43,8 +84,10 @@ impl WriteLog for ElasticSearch { 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; + + if let Some(target) = ELASTIC.hosts.first() { + let _ = httpclient("POST", target.tls, target.address, "/i", target.address, target.port, target.url, payload).await; + } } } diff --git a/src/utils/hcclient.rs b/src/utils/hcclient.rs index fc0d106..ec4995f 100644 --- a/src/utils/hcclient.rs +++ b/src/utils/hcclient.rs @@ -9,7 +9,7 @@ use std::time::Duration; pub static HC_CONNECTOR: LazyLock = LazyLock::new(|| Connector::new(None)); -pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address: &str, port: u16, url: String, payload: Bytes) -> (bool, bool) { +pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address: &str, port: u16, url: &str, payload: Bytes) -> (bool, bool) { let method = match method { "HEAD" => "HEAD", "GET" => "GET", @@ -46,7 +46,7 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address is_h2 = true; } None => { - if ping_grpc(url.as_str()).await { + if ping_grpc(url).await { return (true, true); } } @@ -79,7 +79,7 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address 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 { + if !is_h2 && ping_grpc(url).await { return (true, true); } log::warn!("Health Check read failed ({}) : {} - {}", if is_h2 { "H2" } else { "H1" }, host, e); diff --git a/src/utils/healthcheck.rs b/src/utils/healthcheck.rs index 5a8c65c..6e9cef4 100644 --- a/src/utils/healthcheck.rs +++ b/src/utils/healthcheck.rs @@ -68,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, EMPTY_PAYLOAD).await; + let resp = httpclient(method, tls, host, path, upstream.address.as_ref(), upstream.port, link.as_str(), EMPTY_PAYLOAD).await; if resp.0 { if resp.1 { scheme.is_http2 = resp.1;