From ee436b69db33b9773e7a489b32c3a3c99c94e721 Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Fri, 5 Dec 2025 02:59:01 -0700 Subject: [PATCH] break syslog-rfc5424 dependency in signal-gateway, now it's only in the bin --- Cargo.lock | 1 - signal-gateway-bin/src/main.rs | 70 ++++++++- signal-gateway/Cargo.toml | 1 - signal-gateway/src/gateway/log_handler.rs | 127 ++++++---------- signal-gateway/src/gateway/mod.rs | 60 ++------ signal-gateway/src/lib.rs | 2 + signal-gateway/src/log_message.rs | 173 ++++++++++++++++++++++ 7 files changed, 297 insertions(+), 137 deletions(-) create mode 100644 signal-gateway/src/log_message.rs diff --git a/Cargo.lock b/Cargo.lock index ad15d1d..771edb5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1660,7 +1660,6 @@ dependencies = [ "rand", "serde", "serde_json", - "syslog_rfc5424", "thiserror", "tokio", "tokio-util", diff --git a/signal-gateway-bin/src/main.rs b/signal-gateway-bin/src/main.rs index 44ff867..fafabae 100644 --- a/signal-gateway-bin/src/main.rs +++ b/signal-gateway-bin/src/main.rs @@ -2,14 +2,62 @@ use conf::Conf; use hyper::service::service_fn; use hyper_util::rt::TokioIo; use hyper_util::server::conn::auto; -use signal_gateway::{Gateway, GatewayConfig}; +use signal_gateway::{Gateway, GatewayConfig, Level, LogMessage}; use std::{net::SocketAddr, str::FromStr, sync::Arc, time::Duration}; -use syslog_rfc5424::SyslogMessage; +use syslog_rfc5424::{SyslogMessage, SyslogSeverity}; use tokio::net::{TcpListener, UdpSocket}; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; use tracing_subscriber::EnvFilter; +/// Convert SyslogSeverity to our Level enum +fn severity_to_level(sev: SyslogSeverity) -> Level { + match sev { + SyslogSeverity::SEV_EMERG => Level::EMERGENCY, + SyslogSeverity::SEV_ALERT => Level::ALERT, + SyslogSeverity::SEV_CRIT => Level::CRITICAL, + SyslogSeverity::SEV_ERR => Level::ERROR, + SyslogSeverity::SEV_WARNING => Level::WARNING, + SyslogSeverity::SEV_NOTICE => Level::NOTICE, + SyslogSeverity::SEV_INFO => Level::INFO, + SyslogSeverity::SEV_DEBUG => Level::DEBUG, + } +} + +/// Convert a SyslogMessage to a LogMessage, extracting structured data for tracing metadata +fn syslog_to_log_message(msg: SyslogMessage, sd_id: &str) -> LogMessage { + let level = severity_to_level(msg.severity); + let mut builder = LogMessage::builder(level, msg.msg); + + if let Some(ts) = msg.timestamp { + builder = builder.timestamp(ts); + } + if let Some(nanos) = msg.timestamp_nanos { + builder = builder.timestamp_nanos(nanos); + } + if let Some(hostname) = msg.hostname { + builder = builder.hostname(hostname); + } + if let Some(appname) = msg.appname { + builder = builder.appname(appname); + } + + // Extract tracing metadata from structured data + if let Some(sd_element) = msg.sd.find_sdid(sd_id) { + if let Some(module) = sd_element.get("module") { + builder = builder.module_path(module.clone()); + } + if let Some(file) = sd_element.get("file") { + builder = builder.file(file.clone()); + } + if let Some(line) = sd_element.get("line") { + builder = builder.line(line.clone()); + } + } + + builder.build() +} + #[derive(Conf, Debug)] struct Config { /// If true, just validate config and don't start @@ -21,6 +69,9 @@ struct Config { /// Socket to listen for UDP messages, in syslog RFC 5424 format #[conf(long, env, default_value = "0.0.0.0:5424")] udp_listen_addr: SocketAddr, + /// Structured data ID for tracing metadata (module, file, line) in syslog messages + #[conf(long, env, default_value = "tracing-meta@64700")] + sd_id: String, #[conf(flatten)] gateway: GatewayConfig, } @@ -85,7 +136,7 @@ async fn main() { // Start the two server tasks let _http_task = start_http_task(listener, gateway.clone()); - let _udp_task = start_udp_task(udp_socket, gateway.clone()); + let _udp_task = start_udp_task(udp_socket, gateway.clone(), config.sd_id); // Run gateway task and block on it returning. Note that it exits if the token is canceled. gateway.run().await; @@ -128,8 +179,12 @@ fn start_http_task(listener: TcpListener, gateway: Arc) -> tokio::task: }) } -fn start_udp_task(udp_socket: UdpSocket, gateway: Arc) -> tokio::task::JoinHandle<()> { - // Loop waiting for http incoming connections, and pass them to gateway +fn start_udp_task( + udp_socket: UdpSocket, + gateway: Arc, + sd_id: String, +) -> tokio::task::JoinHandle<()> { + // Loop waiting for UDP syslog messages tokio::task::spawn(async move { let mut buf = vec![0u8; 8192]; loop { @@ -147,13 +202,14 @@ fn start_udp_task(udp_socket: UdpSocket, gateway: Arc) -> tokio::task:: continue; }; - let Ok(msg) = SyslogMessage::from_str(text) + let Ok(syslog_msg) = SyslogMessage::from_str(text) .inspect_err(|err| error!("UDP packet was not valid syslog: {err}:\n{text}")) else { continue; }; - gateway.handle_syslog_message(msg).await; + let log_msg = syslog_to_log_message(syslog_msg, &sd_id); + gateway.handle_log_message(log_msg).await; } }) } diff --git a/signal-gateway/Cargo.toml b/signal-gateway/Cargo.toml index 2c209ee..0a3ae67 100644 --- a/signal-gateway/Cargo.toml +++ b/signal-gateway/Cargo.toml @@ -26,7 +26,6 @@ humantime = { workspace = true } jsonrpsee = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } -syslog_rfc5424 = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tokio-util = { workspace = true } diff --git a/signal-gateway/src/gateway/log_handler.rs b/signal-gateway/src/gateway/log_handler.rs index 9e2fa2e..c8148a2 100644 --- a/signal-gateway/src/gateway/log_handler.rs +++ b/signal-gateway/src/gateway/log_handler.rs @@ -1,11 +1,13 @@ use super::circular_buffer::CircularBuffer; -use super::{AdminMessage, MultiRateLimiter, Origin, RateThreshold, SourceLocationRateLimiter}; -use crate::human_duration::HumanTMinus; +use super::{AdminMessage, MultiRateLimiter, RateThreshold, SourceLocationRateLimiter}; +use crate::{ + human_duration::HumanTMinus, + log_message::{Level, LogMessage, Origin}, +}; use chrono::{TimeDelta, Utc}; use conf::Conf; use serde::Deserialize; use std::{fmt, time::Duration}; -use syslog_rfc5424::{SyslogMessage, SyslogSeverity}; use tokio::sync::{Mutex, mpsc::UnboundedSender}; use tracing::{error, info, warn}; @@ -14,7 +16,7 @@ enum SuppressionReason { /// Suppressed by a configured alert rule (with 0-based rule index) Rule(usize), /// Suppressed by the source-location rate limiter - SourceLocation { file: String, line: String }, + SourceLocation { file: Box, line: Box }, } impl fmt::Display for SuppressionReason { @@ -39,9 +41,6 @@ pub struct LogHandlerConfig { pub format_module: bool, #[conf(long, env)] pub format_source_location: bool, - /// Structured data ID for tracing metadata (module, file, line) in syslog messages - #[conf(long, env, default_value = "tracing-meta@64700")] - pub sd_id: String, /// Number of recent log messages to buffer per origin #[conf(long, env, default_value = "64")] pub log_buffer_size: usize, @@ -63,28 +62,28 @@ pub struct AlertRule { } impl AlertRule { - /// Check if a syslog message passes the filter defined by this rule - fn eval_filter(&self, syslog_msg: &SyslogMessage, sd_id: &str) -> bool { - if !self.msg_contains.is_empty() && !syslog_msg.msg.contains(&self.msg_contains) { + /// Check if a log message passes the filter defined by this rule + fn eval_filter(&self, log_msg: &LogMessage) -> bool { + if !self.msg_contains.is_empty() && !log_msg.msg.contains(&self.msg_contains) { return false; } if !self.module_equals.is_empty() { - match syslog_msg.sd.find_tuple(sd_id, "module") { + match log_msg.module_path.as_deref() { Some(module) if module == self.module_equals.as_str() => {} _ => return false, } } if !self.file_equals.is_empty() { - match syslog_msg.sd.find_tuple(sd_id, "file") { + match log_msg.file.as_deref() { Some(file) if file == self.file_equals.as_str() => {} _ => return false, } } if !self.line_equals.is_empty() { - match syslog_msg.sd.find_tuple(sd_id, "line") { + match log_msg.line.as_deref() { Some(line) if line == self.line_equals.as_str() => {} _ => return false, } @@ -110,7 +109,7 @@ impl AlertRule { pub struct LogHandler { config: LogHandlerConfig, admin_mq_tx: UnboundedSender, - syslog_buffer: Mutex>, + log_buffer: Mutex>, rate_limiters: Vec<(AlertRule, Mutex)>, /// Rate limiter keyed by source location (file:line), so different error locations /// can alert independently without suppressing each other. @@ -156,12 +155,12 @@ impl LogHandler { OVERALL_LIMITER_MAX_ENTRIES, )); - let syslog_buffer = Mutex::new(CircularBuffer::new(config.log_buffer_size)); + let log_buffer = Mutex::new(CircularBuffer::new(config.log_buffer_size)); Self { config, admin_mq_tx, - syslog_buffer, + log_buffer, rate_limiters, overall_limiter, any_rule_uses_module, @@ -172,7 +171,7 @@ impl LogHandler { /// Format recent logs into a string pub async fn format_logs(&self) -> String { - let lk = self.syslog_buffer.lock().await; + let lk = self.log_buffer.lock().await; let mut text = format!("{} log messages (newest first):\n", lk.len()); @@ -181,28 +180,28 @@ impl LogHandler { // Collect and reverse to show newest first let messages: Vec<_> = lk.iter().collect(); - for syslog_msg in messages.into_iter().rev() { - self.write_syslog_msg(&mut text, syslog_msg, now); + for log_msg in messages.into_iter().rev() { + self.write_log_msg(&mut text, log_msg, now); } text } - /// Consume a new syslog message from the given origin - pub async fn handle_syslog_message(&self, mut syslog_msg: SyslogMessage, origin: Origin) { - let suppression_reason = self.check_suppression(&mut syslog_msg).await; + /// Consume a new log message from the given origin + pub async fn handle_log_message(&self, mut log_msg: LogMessage, origin: Origin) { + let suppression_reason = self.check_suppression(&mut log_msg).await; if let Some(reason) = &suppression_reason - && syslog_msg.severity <= SyslogSeverity::SEV_ERR + && log_msg.level <= Level::ERROR { - let sev = Self::severity_to_str(syslog_msg.severity); - info!("Suppressed {sev} ({reason}):\n{}", syslog_msg.msg); + let sev = log_msg.level.to_str(); + info!("Suppressed {sev} ({reason}):\n{}", log_msg.msg); } // Record this new message. // Then, if we should alert now, also format the whole buffer to a string, // and then release the lock. let formatted_text = { - let mut lk = self.syslog_buffer.lock().await; - lk.push_back(syslog_msg); + let mut lk = self.log_buffer.lock().await; + lk.push_back(log_msg); if suppression_reason.is_some() { return; } @@ -213,8 +212,8 @@ impl LogHandler { let now = Utc::now().timestamp(); // Iterate in reverse (newest first) without copying - for syslog_msg in lk.iter().rev() { - self.write_syslog_msg(&mut text, syslog_msg, now); + for log_msg in lk.iter().rev() { + self.write_log_msg(&mut text, log_msg, now); } lk.clear(); @@ -231,32 +230,13 @@ impl LogHandler { } } - // Convert SyslogSeverity to our own all-caps string that fits in 5 chars - fn severity_to_str(severity: SyslogSeverity) -> &'static str { - match severity { - SyslogSeverity::SEV_EMERG => "EMERG", - SyslogSeverity::SEV_ALERT => "ALERT", - SyslogSeverity::SEV_CRIT => "CRIT", - SyslogSeverity::SEV_ERR => "ERROR", - SyslogSeverity::SEV_WARNING => "WARN", - SyslogSeverity::SEV_NOTICE => "NOTE", - SyslogSeverity::SEV_INFO => "INFO", - SyslogSeverity::SEV_DEBUG => "DEBUG", - } - } - - // Format a syslog message into a Writer, followed by \n, and using any config options to do so - fn write_syslog_msg( - &self, - mut writer: impl std::fmt::Write, - syslog_msg: &SyslogMessage, - now: i64, - ) { - let sev = Self::severity_to_str(syslog_msg.severity); - let msg = &syslog_msg.msg; + // Format a log message into a Writer, followed by \n, and using any config options to do so + fn write_log_msg(&self, mut writer: impl std::fmt::Write, log_msg: &LogMessage, now: i64) { + let sev = log_msg.level.to_str(); + let msg = &log_msg.msg; // Format relative timestamp if available - let time_str = if let Some(ts) = syslog_msg.timestamp { + let time_str = if let Some(ts) = log_msg.timestamp { let diff_secs = now.saturating_sub(ts); HumanTMinus(TimeDelta::seconds(diff_secs)).to_string() } else { @@ -264,18 +244,17 @@ impl LogHandler { }; // Extract metadata from structured data if configured - let sd_id = &self.config.sd_id; let mut metadata_parts = Vec::new(); if self.config.format_module - && let Some(module) = syslog_msg.sd.find_tuple(sd_id, "module") + && let Some(module) = log_msg.module_path.as_ref() { metadata_parts.push(module.to_string()); } if self.config.format_source_location { - let file_opt = syslog_msg.sd.find_tuple(sd_id, "file"); - let line_opt = syslog_msg.sd.find_tuple(sd_id, "line"); + let file_opt = log_msg.file.as_ref(); + let line_opt = log_msg.line.as_ref(); let location = if let Some(file) = file_opt { // Strip /home/{username}/ prefix if present @@ -306,37 +285,35 @@ impl LogHandler { }; if let Err(err) = result { - error!("Couldn't write syslog message ({err}): {sev}: {msg}"); + error!("Couldn't write log message ({err}): {sev}: {msg}"); } } /// Check if an alert should be suppressed for a given error message. /// /// Returns `None` if the alert should fire, or `Some(reason)` if suppressed. - async fn check_suppression(&self, syslog_msg: &mut SyslogMessage) -> Option { - let ts_sec = *syslog_msg + async fn check_suppression(&self, log_msg: &mut LogMessage) -> Option { + let ts_sec = *log_msg .timestamp .get_or_insert_with(|| Utc::now().timestamp()); - let high_severity = syslog_msg.severity <= SyslogSeverity::SEV_ERR; + let high_severity = log_msg.level <= Level::ERROR; if !high_severity { // Low severity messages are always "suppressed" (not alerted on) // but we don't need to log a reason for this return Some(SuppressionReason::Rule(usize::MAX)); } - let sd_id = &self.config.sd_id; - // Warn if rules expect structured data but the message doesn't have it - let has_module = syslog_msg.sd.find_tuple(sd_id, "module").is_some(); - let has_file = syslog_msg.sd.find_tuple(sd_id, "file").is_some(); - let has_line = syslog_msg.sd.find_tuple(sd_id, "line").is_some(); + let has_module = log_msg.module_path.is_some(); + let has_file = log_msg.file.is_some(); + let has_line = log_msg.line.is_some(); if (self.any_rule_uses_module && !has_module) || (self.any_rule_uses_file && !has_file) || (self.any_rule_uses_line && !has_line) { warn!( - "Error message missing structured data (sd_id={sd_id}), filtering rules may not work: {syslog_msg:#?}" + "Log message missing source location data, filtering rules may not work: {log_msg:#?}" ); } @@ -344,7 +321,7 @@ impl LogHandler { // Note: we check all rules even if one already suppressed, to update all rate limiters let mut suppressed_by_rule: Option = None; for (idx, (filter, limiter)) in self.rate_limiters.iter().enumerate() { - if filter.eval_filter(syslog_msg, sd_id) && !limiter.lock().await.evaluate(ts_sec) { + if filter.eval_filter(log_msg) && !limiter.lock().await.evaluate(ts_sec) { suppressed_by_rule.get_or_insert(idx); } } @@ -353,14 +330,8 @@ impl LogHandler { } // Extract source location for per-location rate limiting - let file = syslog_msg - .sd - .find_tuple(sd_id, "file") - .map_or("?", |s| s.as_str()); - let line = syslog_msg - .sd - .find_tuple(sd_id, "line") - .map_or("?", |s| s.as_str()); + let file = log_msg.file.as_deref().unwrap_or("?"); + let line = log_msg.line.as_deref().unwrap_or("?"); if !self .overall_limiter @@ -369,8 +340,8 @@ impl LogHandler { .evaluate(file, line, ts_sec) { return Some(SuppressionReason::SourceLocation { - file: file.to_owned(), - line: line.to_owned(), + file: file.into(), + line: line.into(), }); } diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index 909549d..c60f0b1 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -1,6 +1,7 @@ use crate::{ alertmanager::AlertPost, jsonrpc::{Envelope, RpcClient, RpcClientError, SignalMessage, connect_tcp}, + log_message::{LogMessage, Origin}, prometheus::{Prometheus, PrometheusConfig}, }; use chrono::Utc; @@ -13,7 +14,6 @@ use prometheus_http_client::{AlertStatus, ExtractLabels}; use std::{ collections::HashMap, error::Error, fmt::Write, net::SocketAddr, path::PathBuf, time::Duration, }; -use syslog_rfc5424::SyslogMessage; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, join, @@ -127,47 +127,6 @@ fn parse_gateway_command(s: &str) -> Result { .map_err(|e| e.to_string()) } -/// Identifies the source of log messages (app name + host). -/// Used to separate log buffers and rate limiters per source. -#[derive(Clone, Debug, Default, PartialEq, Eq, Hash)] -pub struct Origin { - pub app: String, - pub host: String, -} - -impl std::fmt::Display for Origin { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}@{}", self.app, self.host) - } -} - -impl From<&SyslogMessage> for Origin { - fn from(msg: &SyslogMessage) -> Self { - Self { - app: msg.appname.clone().unwrap_or_default(), - host: msg.hostname.clone().unwrap_or_default(), - } - } -} - -impl Origin { - /// Check if this origin matches a filter string. - /// - /// If the filter contains '@', it is split on the first '@': - /// - The part before '@' must be a substring of `app` - /// - The part after '@' must be a substring of `host` - /// - /// If the filter does not contain '@', it matches if either `app` or `host` - /// contains the filter string. - pub fn matches_filter(&self, filter: &str) -> bool { - if let Some((app_filter, host_filter)) = filter.split_once('@') { - self.app.contains(app_filter) && self.host.contains(host_filter) - } else { - self.app.contains(filter) || self.host.contains(filter) - } - } -} - /// A message queued to be sent to all admins. /// This is generally an alert message, which may have attached images. #[derive(Clone, Debug, Default)] @@ -725,14 +684,15 @@ impl Gateway { Ok(text) } - pub async fn handle_syslog_message(&self, syslog_msg: SyslogMessage) { - let origin = Origin::from(&syslog_msg); + pub async fn handle_log_message(&self, log_msg: impl Into) { + let log_msg = log_msg.into(); + let origin = Origin::from(&log_msg); // Try to get existing handler with read lock first { let handlers = self.log_handlers.read().await; if let Some(handler) = handlers.get(&origin) { - handler.handle_syslog_message(syslog_msg, origin).await; + handler.handle_log_message(log_msg, origin).await; return; } } @@ -744,7 +704,7 @@ impl Gateway { info!("Creating new log handler for origin: {origin}"); LogHandler::new(self.config.log_handler.clone(), self.admin_mq_tx.clone()) }); - handler.handle_syslog_message(syslog_msg, origin).await; + handler.handle_log_message(log_msg, origin).await; } } @@ -854,8 +814,8 @@ mod tests { #[test] fn test_origin_matches_filter() { let origin = Origin { - app: "muad-dib".to_string(), - host: "tokyo-server".to_string(), + app: "muad-dib".into(), + host: "tokyo-server".into(), }; // Without @: matches if app OR host contains the string @@ -879,8 +839,8 @@ mod tests { // Edge case: filter matches the @ in the format but origin has no @ let origin2 = Origin { - app: "app".to_string(), - host: "host".to_string(), + app: "app".into(), + host: "host".into(), }; assert!(origin2.matches_filter("app@host")); assert!(!origin2.matches_filter("app@other")); diff --git a/signal-gateway/src/lib.rs b/signal-gateway/src/lib.rs index d7a01cc..0f24177 100644 --- a/signal-gateway/src/lib.rs +++ b/signal-gateway/src/lib.rs @@ -3,7 +3,9 @@ pub mod gateway; pub(crate) mod human_duration; pub(crate) mod jsonrpc; +pub(crate) mod log_message; pub(crate) mod prometheus; pub(crate) mod transports; pub use gateway::{Gateway, GatewayConfig}; +pub use log_message::{Level, LogMessage, LogMessageBuilder}; diff --git a/signal-gateway/src/log_message.rs b/signal-gateway/src/log_message.rs new file mode 100644 index 0000000..5d98a9c --- /dev/null +++ b/signal-gateway/src/log_message.rs @@ -0,0 +1,173 @@ +//! Log message schema used by this crate + +#[non_exhaustive] +#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)] +pub enum Level { + EMERGENCY = 0, + ALERT = 1, + CRITICAL = 2, + ERROR = 3, + WARNING = 4, + NOTICE = 5, + INFO = 6, + DEBUG = 7, + TRACE = 8, +} + +impl Level { + // Convert to our own all-caps string that fits in 5 chars + pub(crate) fn to_str(self) -> &'static str { + match self { + Self::EMERGENCY => "EMERG", + Self::ALERT => "ALERT", + Self::CRITICAL => "CRIT", + Self::ERROR => "ERROR", + Self::WARNING => "WARN", + Self::NOTICE => "NOTE", + Self::INFO => "INFO", + Self::DEBUG => "DEBUG", + Self::TRACE => "TRACE", + } + } +} + +#[non_exhaustive] +#[derive(Clone, Debug)] +pub struct LogMessage { + pub level: Level, + pub timestamp: Option, + pub timestamp_nanos: u32, + pub hostname: Option>, + pub appname: Option>, + pub msg: Box, + pub module_path: Option>, + pub file: Option>, + pub line: Option>, +} + +impl LogMessage { + pub fn builder(level: Level, msg: impl Into>) -> LogMessageBuilder { + LogMessageBuilder { + level, + msg: msg.into(), + timestamp: None, + timestamp_nanos: 0, + hostname: None, + appname: None, + module_path: None, + file: None, + line: None, + } + } +} + +#[derive(Clone, Debug)] +pub struct LogMessageBuilder { + level: Level, + msg: Box, + timestamp: Option, + timestamp_nanos: u32, + hostname: Option>, + appname: Option>, + module_path: Option>, + file: Option>, + line: Option>, +} + +impl LogMessageBuilder { + pub fn timestamp(mut self, ts: i64) -> Self { + self.timestamp = Some(ts); + self + } + + pub fn timestamp_nanos(mut self, nanos: u32) -> Self { + self.timestamp_nanos = nanos; + self + } + + pub fn hostname(mut self, hostname: impl Into>) -> Self { + self.hostname = Some(hostname.into()); + self + } + + pub fn appname(mut self, appname: impl Into>) -> Self { + self.appname = Some(appname.into()); + self + } + + pub fn module_path(mut self, module_path: impl Into>) -> Self { + self.module_path = Some(module_path.into()); + self + } + + pub fn file(mut self, file: impl Into>) -> Self { + self.file = Some(file.into()); + self + } + + pub fn line(mut self, line: impl Into>) -> Self { + self.line = Some(line.into()); + self + } + + pub fn build(self) -> LogMessage { + LogMessage { + level: self.level, + timestamp: self.timestamp, + timestamp_nanos: self.timestamp_nanos, + hostname: self.hostname, + appname: self.appname, + msg: self.msg, + module_path: self.module_path, + file: self.file, + line: self.line, + } + } +} + +impl From for LogMessage { + fn from(builder: LogMessageBuilder) -> Self { + builder.build() + } +} + +/// Identifies the source of log messages (app name + host). +/// Used to separate log buffers and rate limiters per source. +#[derive(Clone, Debug, Default, PartialEq, Eq, Hash)] +pub struct Origin { + pub app: Box, + pub host: Box, +} + +impl std::fmt::Display for Origin { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}@{}", self.app, self.host) + } +} + +impl From<&LogMessage> for Origin { + fn from(msg: &LogMessage) -> Self { + Self { + app: msg.appname.clone().unwrap_or_default(), + host: msg.hostname.clone().unwrap_or_default(), + } + } +} + +impl Origin { + /// Check if this origin matches a filter string. + /// + /// If the filter contains '@', it is split on the first '@': + /// - The part before '@' must be a substring of `app` + /// - The part after '@' must be a substring of `host` + /// + /// If the filter does not contain '@', it matches if either `app` or `host` + /// contains the filter string. + pub fn matches_filter(&self, filter: &str) -> bool { + if let Some((app_filter, host_filter)) = filter.split_once('@') { + self.app.contains(app_filter) && self.host.contains(host_filter) + } else { + self.app.contains(filter) || self.host.contains(filter) + } + } +}