From 82ed77a09869f25d5c022eb7a5a62ffd31f28abe Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Fri, 5 Dec 2025 10:18:39 -0700 Subject: [PATCH] add support for udp over json logging --- Cargo.lock | 3 + signal-gateway-bin/Cargo.toml | 3 + signal-gateway-bin/src/main.rs | 20 +- signal-gateway-bin/src/udp_json/mod.rs | 472 +++++++++++++++++++++++++ 4 files changed, 493 insertions(+), 5 deletions(-) create mode 100644 signal-gateway-bin/src/udp_json/mod.rs diff --git a/Cargo.lock b/Cargo.lock index ac3695e..49fda46 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1672,12 +1672,15 @@ dependencies = [ name = "signal-gateway-bin" version = "0.1.0" dependencies = [ + "chrono", "conf", "conf-extra", "dotenvy", "hyper", "hyper-util", "reqwest", + "serde", + "serde_json", "signal-gateway", "syslog_rfc5424", "tokio", diff --git a/signal-gateway-bin/Cargo.toml b/signal-gateway-bin/Cargo.toml index 8e10fa5..f1f4117 100644 --- a/signal-gateway-bin/Cargo.toml +++ b/signal-gateway-bin/Cargo.toml @@ -18,12 +18,15 @@ rustls-tls = ["signal-gateway/rustls-tls"] [dependencies] signal-gateway = { path = "../signal-gateway" } +chrono = { workspace = true } conf = { workspace = true } conf-extra = { workspace = true } dotenvy = { workspace = true } hyper = { workspace = true } hyper-util = { workspace = true } reqwest = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } syslog_rfc5424 = { workspace = true } tokio = { workspace = true, features = ["rt-multi-thread", "signal"] } tokio-util = { workspace = true } diff --git a/signal-gateway-bin/src/main.rs b/signal-gateway-bin/src/main.rs index b3a1062..4c3c243 100644 --- a/signal-gateway-bin/src/main.rs +++ b/signal-gateway-bin/src/main.rs @@ -18,6 +18,9 @@ use admin_netcat::AdminNetcatConfig; mod syslog; use syslog::SyslogUdpConfig; +mod udp_json; +use udp_json::UdpJsonConfig; + /// Admin message handler configuration - select how non-command messages are handled #[derive(Clone, Debug, Subcommands)] enum AdminHandlerCommand { @@ -30,11 +33,11 @@ enum AdminHandlerCommand { Http(AdminHttpConfig), } -impl AdminHandlerConfig { +impl AdminHandlerCommand { fn into_handler(self) -> MessageHandler { match self { - AdminHandlerConfig::Netcat(config) => config.into_handler(), - AdminHandlerConfig::Http(config) => config.into_handler(), + AdminHandlerCommand::Netcat(config) => config.into_handler(), + AdminHandlerCommand::Http(config) => config.into_handler(), } } } @@ -49,6 +52,8 @@ struct Config { http_listen_addr: SocketAddr, #[conf(flatten, prefix)] syslog_udp: Option, + #[conf(flatten, prefix)] + udp_json: Option, /// Optional admin message handler (netcat or http) #[conf(subcommands)] admin_handler: Option, @@ -112,13 +117,18 @@ async fn main() { thread_token.cancel(); }); - // Start the two server tasks + // Start the server tasks let _http_task = start_http_task(listener, gateway.clone()); - let _udp_task = if let Some(syslog_udp) = &config.syslog_udp { + let _syslog_udp_task = if let Some(syslog_udp) = &config.syslog_udp { Some(syslog_udp.start_udp_task(gateway.clone()).await.unwrap()) } else { None }; + let _udp_json_task = if let Some(udp_json) = &config.udp_json { + Some(udp_json.start_udp_task(gateway.clone()).await.unwrap()) + } else { + None + }; // Run gateway task and block on it returning. Note that it exits if the token is canceled. gateway.run().await; diff --git a/signal-gateway-bin/src/udp_json/mod.rs b/signal-gateway-bin/src/udp_json/mod.rs new file mode 100644 index 0000000..6127375 --- /dev/null +++ b/signal-gateway-bin/src/udp_json/mod.rs @@ -0,0 +1,472 @@ +//! JSON UDP listener for receiving log messages in a logstash-compatible format +//! +//! This module accepts JSON log messages over UDP and converts them to LogMessage. +//! It's designed to be flexible and accept various common formats. + +use chrono::{DateTime, TimeZone, Utc}; +use conf::Conf; +use serde::Deserialize; +use signal_gateway::{Gateway, Level, LogMessage}; +use std::{net::SocketAddr, sync::Arc}; +use tokio::net::UdpSocket; +use tracing::{error, info}; + +/// Configuration for the JSON UDP listener +#[derive(Conf, Debug)] +pub struct UdpJsonConfig { + /// Socket to listen for UDP messages in JSON format + #[conf(long, env)] + pub listen_addr: SocketAddr, +} + +impl UdpJsonConfig { + /// Bind a UDP socket and start a background task to handle incoming JSON log messages. + /// + /// Returns a join handle for the background task. + pub async fn start_udp_task( + &self, + gateway: Arc, + ) -> std::io::Result> { + let udp_socket = UdpSocket::bind(self.listen_addr).await?; + info!("Listening for JSON UDP on {}", self.listen_addr); + + Ok(tokio::task::spawn(async move { + let mut buf = vec![0u8; 65536]; // Larger buffer for JSON + loop { + let Ok((len, _addr)) = udp_socket + .recv_from(&mut buf) + .await + .inspect_err(|err| error!("Error receiving UDP packet: {err}")) + else { + continue; + }; + + let Ok(text) = std::str::from_utf8(&buf[0..len]) + .inspect_err(|err| error!("UDP packet was not utf8: {err}")) + else { + continue; + }; + + let Ok(json_msg) = serde_json::from_str::(text) + .inspect_err(|err| error!("UDP packet was not valid JSON: {err}:\n{text}")) + else { + continue; + }; + + let log_msg = json_msg.into_log_message(); + gateway.handle_log_message(log_msg).await; + } + })) + } +} + +/// A flexible JSON log message format compatible with logstash and similar systems. +/// +/// Supports various field names and formats commonly used in logging systems. +#[derive(Debug, Deserialize)] +pub struct JsonLogMessage { + /// The log message text - accepts "message" or "msg" + #[serde(alias = "msg")] + pub message: String, + + /// Log level - accepts various formats (error, ERROR, err, etc.) + #[serde(default, alias = "severity", deserialize_with = "deserialize_level")] + pub level: Option, + + /// Timestamp - accepts Unix epoch seconds (int or string) or RFC3339 string + #[serde( + default, + alias = "time", + alias = "@timestamp", + deserialize_with = "deserialize_timestamp" + )] + pub timestamp: Option>, + + /// Hostname + #[serde(alias = "host")] + pub hostname: Option, + + /// Application name + #[serde(alias = "app", alias = "application", alias = "service")] + pub appname: Option, + + /// Module path + #[serde(alias = "module", alias = "logger", alias = "logger_name")] + pub module_path: Option, + + /// Source file + #[serde(alias = "filename", alias = "source_file")] + pub file: Option, + + /// Line number - accepts integer or string + #[serde(default, alias = "lineno", deserialize_with = "deserialize_line")] + pub line: Option, +} + +impl JsonLogMessage { + /// Convert to a LogMessage + pub fn into_log_message(self) -> LogMessage { + let level = self.level.unwrap_or(Level::INFO); + let mut builder = LogMessage::builder(level, self.message); + + if let Some(ts) = self.timestamp { + builder = builder.timestamp(ts.timestamp()); + builder = builder.timestamp_nanos(ts.timestamp_subsec_nanos()); + } + if let Some(hostname) = self.hostname { + builder = builder.hostname(hostname); + } + if let Some(appname) = self.appname { + builder = builder.appname(appname); + } + if let Some(module_path) = self.module_path { + builder = builder.module_path(module_path); + } + if let Some(file) = self.file { + builder = builder.file(file); + } + if let Some(line) = self.line { + builder = builder.line(line); + } + + builder.build() + } +} + +/// Deserialize a log level from various string formats +fn deserialize_level<'de, D>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + let opt: Option = Option::deserialize(deserializer)?; + Ok(opt.and_then(|s| parse_level(&s))) +} + +/// Parse a level string into a Level enum +fn parse_level(s: &str) -> Option { + match s.to_lowercase().as_str() { + "emergency" | "emerg" => Some(Level::EMERGENCY), + "alert" => Some(Level::ALERT), + "critical" | "crit" | "fatal" => Some(Level::CRITICAL), + "error" | "err" => Some(Level::ERROR), + "warning" | "warn" => Some(Level::WARNING), + "notice" => Some(Level::NOTICE), + "info" | "information" => Some(Level::INFO), + "debug" => Some(Level::DEBUG), + "trace" => Some(Level::TRACE), + _ => None, + } +} + +/// Deserialize a timestamp from Unix epoch (int or string) or RFC3339 string +fn deserialize_timestamp<'de, D>(deserializer: D) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + use serde::de::Error; + + #[derive(Deserialize)] + #[serde(untagged)] + enum TimestampValue { + Integer(i64), + Float(f64), + String(String), + } + + let opt: Option = Option::deserialize(deserializer)?; + match opt { + None => Ok(None), + Some(TimestampValue::Integer(secs)) => Ok(Utc.timestamp_opt(secs, 0).single()), + Some(TimestampValue::Float(secs)) => { + let whole_secs = secs.trunc() as i64; + let nanos = ((secs.fract()) * 1_000_000_000.0) as u32; + Ok(Utc.timestamp_opt(whole_secs, nanos).single()) + } + Some(TimestampValue::String(s)) => { + // Try parsing as integer first (Unix timestamp as string) + if let Ok(secs) = s.parse::() { + return Ok(Utc.timestamp_opt(secs, 0).single()); + } + // Try parsing as float (Unix timestamp with fractional seconds) + if let Ok(secs) = s.parse::() { + let whole_secs = secs.trunc() as i64; + let nanos = ((secs.fract()) * 1_000_000_000.0) as u32; + return Ok(Utc.timestamp_opt(whole_secs, nanos).single()); + } + // Try parsing as RFC3339 + DateTime::parse_from_rfc3339(&s) + .map(|dt| Some(dt.with_timezone(&Utc))) + .map_err(|e| D::Error::custom(format!("invalid timestamp: {e}"))) + } + } +} + +/// Deserialize a line number from integer or string +fn deserialize_line<'de, D>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + #[derive(Deserialize)] + #[serde(untagged)] + enum LineValue { + Integer(u64), + String(String), + } + + let opt: Option = Option::deserialize(deserializer)?; + Ok(match opt { + None => None, + Some(LineValue::Integer(n)) => Some(n.to_string()), + Some(LineValue::String(s)) => Some(s), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_basic_message_with_message_field() { + let json = r#"{"message": "Hello, world!"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.message, "Hello, world!"); + let log_msg = msg.into_log_message(); + assert_eq!(&*log_msg.msg, "Hello, world!"); + assert_eq!(log_msg.level, Level::INFO); // default + } + + #[test] + fn test_basic_message_with_msg_field() { + let json = r#"{"msg": "Hello from msg!"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.message, "Hello from msg!"); + } + + #[test] + fn test_level_variations() { + // Lowercase + let json = r#"{"message": "test", "level": "error"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.level, Some(Level::ERROR)); + + // Uppercase + let json = r#"{"message": "test", "level": "ERROR"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.level, Some(Level::ERROR)); + + // Short form + let json = r#"{"message": "test", "level": "err"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.level, Some(Level::ERROR)); + + // Warning variations + let json = r#"{"message": "test", "level": "warn"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.level, Some(Level::WARNING)); + + let json = r#"{"message": "test", "level": "warning"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.level, Some(Level::WARNING)); + + // Fatal -> Critical + let json = r#"{"message": "test", "level": "fatal"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.level, Some(Level::CRITICAL)); + + // Severity alias + let json = r#"{"message": "test", "severity": "error"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.level, Some(Level::ERROR)); + } + + #[test] + fn test_timestamp_as_integer() { + let json = r#"{"message": "test", "timestamp": 1733500000}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + let ts = msg.timestamp.unwrap(); + assert_eq!(ts.timestamp(), 1733500000); + } + + #[test] + fn test_timestamp_as_float() { + let json = r#"{"message": "test", "timestamp": 1733500000.123456}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + let ts = msg.timestamp.unwrap(); + assert_eq!(ts.timestamp(), 1733500000); + assert!(ts.timestamp_subsec_nanos() > 123000000); + assert!(ts.timestamp_subsec_nanos() < 124000000); + } + + #[test] + fn test_timestamp_as_string_integer() { + let json = r#"{"message": "test", "timestamp": "1733500000"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + let ts = msg.timestamp.unwrap(); + assert_eq!(ts.timestamp(), 1733500000); + } + + #[test] + fn test_timestamp_as_rfc3339() { + let json = r#"{"message": "test", "timestamp": "2024-12-06T12:00:00Z"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + let ts = msg.timestamp.unwrap(); + assert_eq!(ts.timestamp(), 1733486400); + } + + #[test] + fn test_timestamp_as_rfc3339_with_offset() { + let json = r#"{"message": "test", "timestamp": "2024-12-06T12:00:00+05:30"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + let ts = msg.timestamp.unwrap(); + // 12:00 +05:30 = 06:30 UTC + assert_eq!(ts.timestamp(), 1733486400 - 5 * 3600 - 30 * 60); + } + + #[test] + fn test_timestamp_aliases() { + // @timestamp (logstash style) + let json = r#"{"message": "test", "@timestamp": "2024-12-06T12:00:00Z"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert!(msg.timestamp.is_some()); + + // time + let json = r#"{"message": "test", "time": 1733500000}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert!(msg.timestamp.is_some()); + } + + #[test] + fn test_line_as_integer() { + let json = r#"{"message": "test", "line": 42}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.line, Some("42".to_string())); + } + + #[test] + fn test_line_as_string() { + let json = r#"{"message": "test", "line": "42"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.line, Some("42".to_string())); + } + + #[test] + fn test_line_aliases() { + let json = r#"{"message": "test", "lineno": 123}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.line, Some("123".to_string())); + } + + #[test] + fn test_hostname_aliases() { + let json = r#"{"message": "test", "host": "myserver"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.hostname, Some("myserver".to_string())); + + let json = r#"{"message": "test", "hostname": "myserver2"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.hostname, Some("myserver2".to_string())); + } + + #[test] + fn test_appname_aliases() { + let json = r#"{"message": "test", "app": "myapp"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.appname, Some("myapp".to_string())); + + let json = r#"{"message": "test", "application": "myapp2"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.appname, Some("myapp2".to_string())); + + let json = r#"{"message": "test", "service": "myservice"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.appname, Some("myservice".to_string())); + } + + #[test] + fn test_module_path_aliases() { + let json = r#"{"message": "test", "module": "mymodule"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.module_path, Some("mymodule".to_string())); + + let json = r#"{"message": "test", "logger": "mylogger"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.module_path, Some("mylogger".to_string())); + + let json = r#"{"message": "test", "logger_name": "com.example.MyClass"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.module_path, Some("com.example.MyClass".to_string())); + } + + #[test] + fn test_file_aliases() { + let json = r#"{"message": "test", "file": "main.rs"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.file, Some("main.rs".to_string())); + + let json = r#"{"message": "test", "filename": "app.py"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.file, Some("app.py".to_string())); + + let json = r#"{"message": "test", "source_file": "Handler.java"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.file, Some("Handler.java".to_string())); + } + + #[test] + fn test_full_logstash_style_message() { + let json = r#"{ + "@timestamp": "2024-12-06T12:00:00Z", + "message": "User logged in", + "level": "info", + "host": "web-01", + "service": "auth-service", + "logger_name": "com.example.auth.LoginHandler", + "filename": "LoginHandler.java", + "lineno": 123 + }"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + let log_msg = msg.into_log_message(); + + assert_eq!(&*log_msg.msg, "User logged in"); + assert_eq!(log_msg.level, Level::INFO); + assert_eq!(log_msg.hostname.as_deref(), Some("web-01")); + assert_eq!(log_msg.appname.as_deref(), Some("auth-service")); + assert_eq!( + log_msg.module_path.as_deref(), + Some("com.example.auth.LoginHandler") + ); + assert_eq!(log_msg.file.as_deref(), Some("LoginHandler.java")); + assert_eq!(log_msg.line.as_deref(), Some("123")); + assert!(log_msg.timestamp.is_some()); + } + + #[test] + fn test_minimal_message() { + let json = r#"{"msg": "simple log"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + let log_msg = msg.into_log_message(); + + assert_eq!(&*log_msg.msg, "simple log"); + assert_eq!(log_msg.level, Level::INFO); + assert!(log_msg.hostname.is_none()); + assert!(log_msg.appname.is_none()); + assert!(log_msg.timestamp.is_none()); + } + + #[test] + fn test_unknown_level_defaults_to_none() { + let json = r#"{"message": "test", "level": "unknown_level"}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.level, None); + // Should default to INFO when converted + let log_msg = msg.into_log_message(); + assert_eq!(log_msg.level, Level::INFO); + } + + #[test] + fn test_extra_fields_are_ignored() { + let json = r#"{"message": "test", "extra_field": "ignored", "nested": {"also": "ignored"}}"#; + let msg: JsonLogMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.message, "test"); + } +}