diff --git a/src/core/bgservice.rs b/src/core/bgservice.rs index 1489410..608508d 100644 --- a/src/core/bgservice.rs +++ b/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 let (tx, mut rx) = mpsc::channel::(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() => { diff --git a/src/core/proxyhttp.rs b/src/core/proxyhttp.rs index c76d992..cd73790 100644 --- a/src/core/proxyhttp.rs +++ b/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; } } diff --git a/src/core/start.rs b/src/core/start.rs index a3ceb75..0c82545 100644 --- a/src/core/start.rs +++ b/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); diff --git a/src/logging/core.rs b/src/logging/core.rs index 49f33d1..1ff7ddc 100644 --- a/src/logging/core.rs +++ b/src/logging/core.rs @@ -94,7 +94,7 @@ impl Append for StructuredChannelAppender { fn flush(&self) {} } -pub fn log_builder(conf: &AppConfig, location: &Option) { +pub async fn log_builder(conf: &AppConfig, location: &Option) { 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) { 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> = PINGORA_LOG_SENDER .get() @@ -215,7 +215,7 @@ pub fn log_builder(conf: &AppConfig, location: &Option) { } } -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) { +pub async fn init_access_logging(enabled: Option) { 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::(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) { - while let Some(msg) = receiver.blocking_recv() { - write_access_log(&msg); +pub async fn access_log_receiver(mut receiver: mpsc::Receiver) { + 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: {}", diff --git a/src/logging/elasticsearch.rs b/src/logging/elasticsearch.rs index 2a5f439..bacd16f 100644 --- a/src/logging/elasticsearch.rs +++ b/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 = 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; } } diff --git a/src/logging/example.rs b/src/logging/example.rs index d978ef9..9745b0f 100644 --- a/src/logging/example.rs +++ b/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); } diff --git a/src/logging/types.rs b/src/logging/types.rs index 2fbaee3..b925e43 100644 --- a/src/logging/types.rs +++ b/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>> = 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); } diff --git a/src/utils/hcclient.rs b/src/utils/hcclient.rs index 4d5a4d8..fc0d106 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) -> (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) } +*/ diff --git a/src/utils/healthcheck.rs b/src/utils/healthcheck.rs index 72e5efb..5a8c65c 100644 --- a/src/utils/healthcheck.rs +++ b/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, fullist: Arc, idlist: Arc, 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; diff --git a/src/utils/parceyaml.rs b/src/utils/parceyaml.rs index 03a95ca..a9b9bb5 100644 --- a/src/utils/parceyaml.rs +++ b/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::() {