mirror of
https://github.com/sadoyan/aralez.git
synced 2026-10-23 06:01:35 +08:00
Making custom Elsticsearch logger 2.
This commit is contained in:
+16
-30
@@ -2,7 +2,7 @@ use crate::logging::types::sendlog;
|
|||||||
use crate::utils::metrics::LOGGING_ERRORS;
|
use crate::utils::metrics::LOGGING_ERRORS;
|
||||||
use crate::utils::types::AppConfig;
|
use crate::utils::types::AppConfig;
|
||||||
use anyhow::Result;
|
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::roll::fixed_window::FixedWindowRoller;
|
||||||
use log4rs::append::rolling_file::policy::compound::trigger::size::SizeTrigger;
|
use log4rs::append::rolling_file::policy::compound::trigger::size::SizeTrigger;
|
||||||
use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
|
use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
|
||||||
@@ -335,33 +335,19 @@ pub async fn access_log_receiver(mut receiver: mpsc::Receiver<LogMessage>) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn write_access_log(msg: &LogMessage) {
|
async fn write_access_log(msg: &LogMessage) {
|
||||||
match MatchStatus::from_code(msg.response_code) {
|
let level = match MatchStatus::from_code(msg.response_code) {
|
||||||
MatchStatus::Ok2xx => info!(
|
MatchStatus::Ok2xx => Level::Info,
|
||||||
"{}, {}, {}, client: {}, version: {:?}, useragent: {}",
|
MatchStatus::Er4xx => Level::Warn,
|
||||||
msg.response_code,
|
MatchStatus::Er5xx => Level::Error,
|
||||||
msg.cache_status.as_str(),
|
};
|
||||||
msg.summary,
|
log!(
|
||||||
msg.client_ip,
|
level,
|
||||||
msg.version,
|
"{}, {}, {}, client: {}, version: {:?}, useragent: {}",
|
||||||
msg.user_agent,
|
msg.response_code,
|
||||||
),
|
msg.cache_status.as_str(),
|
||||||
MatchStatus::Er4xx => warn!(
|
msg.summary,
|
||||||
"{}, {}, {}, client: {}, version: {:?}, useragent: {}",
|
msg.client_ip,
|
||||||
msg.response_code,
|
msg.version,
|
||||||
msg.cache_status.as_str(),
|
msg.user_agent,
|
||||||
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,
|
|
||||||
),
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,32 +6,73 @@ use bytes::Bytes;
|
|||||||
use std::env;
|
use std::env;
|
||||||
use std::sync::LazyLock;
|
use std::sync::LazyLock;
|
||||||
|
|
||||||
pub struct ElasticSearchConfig {
|
static ELASTIC: LazyLock<ElasticSearchConfig> = LazyLock::new(ElasticSearchConfig::load_from_env);
|
||||||
pub hosts: &'static [&'static str],
|
|
||||||
|
#[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 user: &'static str,
|
||||||
pub password: &'static str,
|
pub password: &'static str,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ElasticSearchConfig {
|
impl ElasticSearchConfig {
|
||||||
pub fn load_from_env() -> Self {
|
pub fn load_from_env() -> Self {
|
||||||
let hosts: Vec<&'static str> = env::var("LOG_ELASTIC_HOSTS")
|
let raw_hosts = env::var("LOG_ELASTIC_HOSTS").unwrap_or_else(|_| "http://127.0.0.1:9200".to_string());
|
||||||
.unwrap_or_else(|_| "127.0.0.1".to_string())
|
|
||||||
.split(',')
|
let mut hosts_struct = Vec::new();
|
||||||
.map(|s| -> &'static str { Box::leak(s.trim().to_string().into_boxed_str()) })
|
|
||||||
.collect();
|
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::<u16>().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 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 password = env::var("LOG_ELASTIC_PASSWORD").unwrap_or_else(|_| "elastic".to_string());
|
||||||
|
|
||||||
Self {
|
Self {
|
||||||
hosts: Box::leak(hosts.into_boxed_slice()),
|
hosts: Box::leak(hosts_struct.into_boxed_slice()),
|
||||||
user: Box::leak(user.into_boxed_str()),
|
user: Box::leak(user.into_boxed_str()),
|
||||||
password: Box::leak(password.into_boxed_str()),
|
password: Box::leak(password.into_boxed_str()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
pub static ELASTIC: LazyLock<ElasticSearchConfig> = LazyLock::new(ElasticSearchConfig::load_from_env);
|
|
||||||
pub struct ElasticSearch;
|
struct ElasticSearch;
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
impl WriteLog for ElasticSearch {
|
impl WriteLog for ElasticSearch {
|
||||||
@@ -43,8 +84,10 @@ impl WriteLog for ElasticSearch {
|
|||||||
return;
|
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;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ use std::time::Duration;
|
|||||||
|
|
||||||
pub static HC_CONNECTOR: LazyLock<Connector> = LazyLock::new(|| Connector::new(None));
|
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, 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 {
|
let method = match method {
|
||||||
"HEAD" => "HEAD",
|
"HEAD" => "HEAD",
|
||||||
"GET" => "GET",
|
"GET" => "GET",
|
||||||
@@ -46,7 +46,7 @@ pub async fn httpclient(method: &str, tls: bool, host: &str, path: &str, address
|
|||||||
is_h2 = true;
|
is_h2 = true;
|
||||||
}
|
}
|
||||||
None => {
|
None => {
|
||||||
if ping_grpc(url.as_str()).await {
|
if ping_grpc(url).await {
|
||||||
return (true, true);
|
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 {
|
let status = match http_session.read_response_header().await {
|
||||||
Ok(_) => http_session.response_header().map(|r| r.status.as_u16()).unwrap_or(500),
|
Ok(_) => http_session.response_header().map(|r| r.status.as_u16()).unwrap_or(500),
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
if !is_h2 && ping_grpc(url.as_str()).await {
|
if !is_h2 && ping_grpc(url).await {
|
||||||
return (true, true);
|
return (true, true);
|
||||||
}
|
}
|
||||||
log::warn!("Health Check read failed ({}) : {} - {}", if is_h2 { "H2" } else { "H1" }, host, e);
|
log::warn!("Health Check read failed ({}) : {} - {}", if is_h2 { "H2" } else { "H1" }, host, e);
|
||||||
|
|||||||
@@ -68,7 +68,7 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str) -> UpstreamsD
|
|||||||
} else {
|
} else {
|
||||||
format!("http://{}:{}{}", upstream.address, upstream.port, path)
|
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.0 {
|
||||||
if resp.1 {
|
if resp.1 {
|
||||||
scheme.is_http2 = resp.1;
|
scheme.is_http2 = resp.1;
|
||||||
|
|||||||
Reference in New Issue
Block a user