From a8b3f804f42d8f414e2403b95c76d7ccd3ea01b7 Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Fri, 25 Sep 2026 17:52:01 +0200 Subject: [PATCH] Done with logging to ElasticSearch. --- Cargo.lock | 1 + Cargo.toml | 1 + src/logging/core.rs | 2 +- src/logging/elasticsearch.rs | 99 +++++++++++++++++++++++++++++++----- src/utils/tools.rs | 18 ++++++- 5 files changed, 105 insertions(+), 16 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 65b787b..1e3e94b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -144,6 +144,7 @@ dependencies = [ "elasticsearch", "futures", "http-body-util", + "humantime", "hyper", "hyper-util", "instant-acme", diff --git a/Cargo.toml b/Cargo.toml index 5b67fd3..546a80d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -67,4 +67,5 @@ lazy_static = "1.5.0" anyhow = "1.0.103" inventory = "0.3.24" elasticsearch = "9.1.0-alpha.1" +humantime = "2.4.0" diff --git a/src/logging/core.rs b/src/logging/core.rs index 0103100..ec3c54a 100644 --- a/src/logging/core.rs +++ b/src/logging/core.rs @@ -34,7 +34,7 @@ pub struct LogMessage { pub cache_status: CachePhase, } -#[derive(Debug, Serialize)] +#[derive(Debug, Serialize, Clone)] pub struct StructuredSystemLog { pub target: &'static str, pub level: Level, diff --git a/src/logging/elasticsearch.rs b/src/logging/elasticsearch.rs index 84d9059..13a8e6a 100644 --- a/src/logging/elasticsearch.rs +++ b/src/logging/elasticsearch.rs @@ -1,17 +1,21 @@ use crate::logging::core::StructuredSystemLog; use crate::logging::types::{LogBackendPlugin, WriteLog}; +use crate::utils::tools::get_hostname; use async_trait::async_trait; +use bytes::{BufMut, BytesMut}; use elasticsearch::auth::Credentials; use elasticsearch::http::transport::Transport; -use elasticsearch::{Elasticsearch, IndexParts}; +use elasticsearch::{BulkParts, Elasticsearch}; +use humantime::format_rfc3339; +use serde_json::json; use std::env; use std::sync::LazyLock; use std::time::Duration; +use std::time::SystemTime; +use tokio::sync::mpsc; use tokio::sync::OnceCell; -static ELASTIC: LazyLock = LazyLock::new(ElasticSearchConfig::load_from_env); static CLIENT: OnceCell = OnceCell::const_new(); - #[derive(Debug)] struct ElasticHost { url: &'static str, @@ -21,8 +25,14 @@ struct ElasticSearchConfig { pub hosts: &'static [ElasticHost], pub user: &'static str, pub password: &'static str, + pub hostname: &'static str, + pub index_name: &'static str, + pub flush_duration: u64, + pub batch_len: usize, + pub buffer_size: usize, } +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()); @@ -60,15 +70,52 @@ impl ElasticSearchConfig { 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()); + let index_name = env::var("LOG_ELASTIC_INDEX").unwrap_or_else(|_| "aralez-logs".to_string()); + let s_flush_duration = env::var("LOG_ELASTIC_FLUSH_DURATION").unwrap_or_else(|_| "2".to_string()); + let flush_duration = s_flush_duration.parse::().unwrap_or(2); + let s_batch_len = env::var("LOG_ELASTIC_BATCH_LEN").unwrap_or_else(|_| "200".to_string()); + let batch_len = s_batch_len.parse::().unwrap_or(200); + let buffer_size: usize = 100 * 512; Self { hosts: Box::leak(hosts_struct.into_boxed_slice()), user: Box::leak(user.into_boxed_str()), password: Box::leak(password.into_boxed_str()), + flush_duration, + batch_len, + buffer_size, + hostname: Box::leak(get_hostname().into_boxed_str()), + index_name: Box::leak(index_name.into_boxed_str()), } } } +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() { @@ -95,23 +142,47 @@ async fn make_es_pool() -> Elasticsearch { client } struct ElasticSearch; -/* -#[derive(Debug, Serialize)] -pub struct StructuredSystemLog { - pub target: &'static str, - pub level: Level, - pub message: String, -} -*/ + #[async_trait] impl WriteLog for ElasticSearch { async fn writelog(&self, msg: &StructuredSystemLog) { - let index_name = "logs"; - let response = CLIENT.get_or_init(make_es_pool).await.index(IndexParts::Index(index_name)).body(msg).send().await; - println!("{:?}", response); + let _ = LOG_CHANNEL.send(msg.clone()); } } +async fn flush_to_es(batch: &mut Vec, buffer: &mut BytesMut, 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"); + let logstash_payload = json!({ + "@timestamp": format_rfc3339(SystemTime::now()).to_string(), + "@version": "1", + "log": { + "level": msg.level, + "logger": msg.target, + }, + "message": msg.message, + "host": { + "hostname": hostname, + } + }); + + if serde_json::to_writer((buffer).writer(), &logstash_payload).is_ok() { + 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; + + let _ = client.bulk(BulkParts::None).body(vec![payload]).send().await; +} + inventory::submit! { LogBackendPlugin { name: "elasticsearch", diff --git a/src/utils/tools.rs b/src/utils/tools.rs index cb82d93..0dfc746 100644 --- a/src/utils/tools.rs +++ b/src/utils/tools.rs @@ -14,10 +14,11 @@ use std::net::SocketAddr; use std::net::TcpListener; use std::os::unix::fs::MetadataExt; use std::os::unix::fs::OpenOptionsExt; +use std::process::Command; use std::str::FromStr; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; -use std::{fs, process, thread, time}; +use std::{env, fs, process, thread, time}; pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams) { let mut out = String::new(); @@ -348,3 +349,18 @@ pub fn split_host_port(addr: &str, tls: bool) -> Option<(&str, u16, bool, &str)> } }; } + +pub fn get_hostname() -> String { + if let Ok(host) = env::var("HOSTNAME").or_else(|_| env::var("COMPUTERNAME")) { + if !host.trim().is_empty() { + return host; + } + } + Command::new("hostname") + .output() + .ok() + .and_then(|output| String::from_utf8(output.stdout).ok()) + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "unknown".to_string()) +}