break syslog-rfc5424 dependency in signal-gateway, now it's only in the bin

Cargo.lock

@@ -1660,7 +1660,6 @@ dependencies = [
  "rand",
  "serde",
  "serde_json",
- "syslog_rfc5424",
  "thiserror",
  "tokio",
  "tokio-util",

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<Gateway>) -> tokio::task:
     })
 }
 
-fn start_udp_task(udp_socket: UdpSocket, gateway: Arc<Gateway>) -> tokio::task::JoinHandle<()> {
+fn start_udp_task(
-    // Loop waiting for http incoming connections, and pass them to gateway
+    udp_socket: UdpSocket,
+    gateway: Arc<Gateway>,
+    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<Gateway>) -> 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;
         }
     })
 }

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 }

signal-gateway/src/gateway/log_handler.rs

@@ -1,11 +1,13 @@
 use super::circular_buffer::CircularBuffer;
-use super::{AdminMessage, MultiRateLimiter, Origin, RateThreshold, SourceLocationRateLimiter};
+use super::{AdminMessage, MultiRateLimiter, RateThreshold, SourceLocationRateLimiter};
-use crate::human_duration::HumanTMinus;
+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<str>, line: Box<str> },
 }
 
 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
+    /// Check if a log message passes the filter defined by this rule
-    fn eval_filter(&self, syslog_msg: &SyslogMessage, sd_id: &str) -> bool {
+    fn eval_filter(&self, log_msg: &LogMessage) -> bool {
-        if !self.msg_contains.is_empty() && !syslog_msg.msg.contains(&self.msg_contains) {
+        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<AdminMessage>,
-    syslog_buffer: Mutex<CircularBuffer<SyslogMessage>>,
+    log_buffer: Mutex<CircularBuffer<LogMessage>>,
     rate_limiters: Vec<(AlertRule, Mutex<MultiRateLimiter>)>,
     /// 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() {
+        for log_msg in messages.into_iter().rev() {
-            self.write_syslog_msg(&mut text, syslog_msg, now);
+            self.write_log_msg(&mut text, log_msg, now);
         }
         text
     }
 
-    /// Consume a new syslog message from the given origin
+    /// Consume a new log message from the given origin
-    pub async fn handle_syslog_message(&self, mut syslog_msg: SyslogMessage, origin: Origin) {
+    pub async fn handle_log_message(&self, mut log_msg: LogMessage, origin: Origin) {
-        let suppression_reason = self.check_suppression(&mut syslog_msg).await;
+        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);
+            let sev = log_msg.level.to_str();
-            info!("Suppressed {sev} ({reason}):\n{}", syslog_msg.msg);
+            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;
+            let mut lk = self.log_buffer.lock().await;
-            lk.push_back(syslog_msg);
+            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() {
+            for log_msg in lk.iter().rev() {
-                self.write_syslog_msg(&mut text, syslog_msg, now);
+                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
+    // Format a log message into a Writer, followed by \n, and using any config options to do so
-    fn severity_to_str(severity: SyslogSeverity) -> &'static str {
+    fn write_log_msg(&self, mut writer: impl std::fmt::Write, log_msg: &LogMessage, now: i64) {
-        match severity {
+        let sev = log_msg.level.to_str();
-            SyslogSeverity::SEV_EMERG => "EMERG",
+        let msg = &log_msg.msg;
-            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 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 file_opt = log_msg.file.as_ref();
-            let line_opt = syslog_msg.sd.find_tuple(sd_id, "line");
+            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<SuppressionReason> {
+    async fn check_suppression(&self, log_msg: &mut LogMessage) -> Option<SuppressionReason> {
-        let ts_sec = *syslog_msg
+        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_module = log_msg.module_path.is_some();
-        let has_file = syslog_msg.sd.find_tuple(sd_id, "file").is_some();
+        let has_file = log_msg.file.is_some();
-        let has_line = syslog_msg.sd.find_tuple(sd_id, "line").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<usize> = 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
+        let file = log_msg.file.as_deref().unwrap_or("?");
-            .sd
+        let line = log_msg.line.as_deref().unwrap_or("?");
-            .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());
 
         if !self
             .overall_limiter
@@ -369,8 +340,8 @@ impl LogHandler {
             .evaluate(file, line, ts_sec)
         {
             return Some(SuppressionReason::SourceLocation {
-                file: file.to_owned(),
+                file: file.into(),
-                line: line.to_owned(),
+                line: line.into(),
             });
         }
 

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<GatewayCommand, String> {
         .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) {
+    pub async fn handle_log_message(&self, log_msg: impl Into<LogMessage>) {
-        let origin = Origin::from(&syslog_msg);
+        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(),
+            app: "muad-dib".into(),
-            host: "tokyo-server".to_string(),
+            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(),
+            app: "app".into(),
-            host: "host".to_string(),
+            host: "host".into(),
         };
         assert!(origin2.matches_filter("app@host"));
         assert!(!origin2.matches_filter("app@other"));

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};

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<i64>,
+    pub timestamp_nanos: u32,
+    pub hostname: Option<Box<str>>,
+    pub appname: Option<Box<str>>,
+    pub msg: Box<str>,
+    pub module_path: Option<Box<str>>,
+    pub file: Option<Box<str>>,
+    pub line: Option<Box<str>>,
+}
+
+impl LogMessage {
+    pub fn builder(level: Level, msg: impl Into<Box<str>>) -> 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<str>,
+    timestamp: Option<i64>,
+    timestamp_nanos: u32,
+    hostname: Option<Box<str>>,
+    appname: Option<Box<str>>,
+    module_path: Option<Box<str>>,
+    file: Option<Box<str>>,
+    line: Option<Box<str>>,
+}
+
+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<Box<str>>) -> Self {
+        self.hostname = Some(hostname.into());
+        self
+    }
+
+    pub fn appname(mut self, appname: impl Into<Box<str>>) -> Self {
+        self.appname = Some(appname.into());
+        self
+    }
+
+    pub fn module_path(mut self, module_path: impl Into<Box<str>>) -> Self {
+        self.module_path = Some(module_path.into());
+        self
+    }
+
+    pub fn file(mut self, file: impl Into<Box<str>>) -> Self {
+        self.file = Some(file.into());
+        self
+    }
+
+    pub fn line(mut self, line: impl Into<Box<str>>) -> 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<LogMessageBuilder> 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<str>,
+    pub host: Box<str>,
+}
+
+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)
+        }
+    }
+}