From ffa1246e51212d76f767f6ac2a2109b52bf982e4 Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Fri, 5 Dec 2025 15:12:20 -0700 Subject: [PATCH] listen on udp and tcp for json logs as well --- signal-gateway-bin/src/admin_http/mod.rs | 10 ++-- signal-gateway-bin/src/admin_netcat/mod.rs | 12 ++-- .../src/{udp_json => json}/json_lines.rs | 4 +- .../src/{udp_json => json}/mod.rs | 55 +++++++++++++------ signal-gateway-bin/src/main.rs | 10 ++-- signal-gateway-bin/src/syslog/mod.rs | 23 +++++--- signal-gateway/src/gateway/mod.rs | 4 +- 7 files changed, 74 insertions(+), 44 deletions(-) rename signal-gateway-bin/src/{udp_json => json}/json_lines.rs (98%) rename signal-gateway-bin/src/{udp_json => json}/mod.rs (92%) diff --git a/signal-gateway-bin/src/admin_http/mod.rs b/signal-gateway-bin/src/admin_http/mod.rs index 2a8409b..1202cd7 100644 --- a/signal-gateway-bin/src/admin_http/mod.rs +++ b/signal-gateway-bin/src/admin_http/mod.rs @@ -61,10 +61,12 @@ impl MessageHandler for AdminHttpHandler { .map_err(|err| (502u16, format!("HTTP request failed: {err}").into()))?; let status = response.status(); - let body = response - .text() - .await - .map_err(|err| (502u16, format!("Failed to read response body: {err}").into()))?; + let body = response.text().await.map_err(|err| { + ( + 502u16, + format!("Failed to read response body: {err}").into(), + ) + })?; if !status.is_success() { return Err((status.as_u16(), body.into())); diff --git a/signal-gateway-bin/src/admin_netcat/mod.rs b/signal-gateway-bin/src/admin_netcat/mod.rs index a7cd835..479349e 100644 --- a/signal-gateway-bin/src/admin_netcat/mod.rs +++ b/signal-gateway-bin/src/admin_netcat/mod.rs @@ -49,11 +49,13 @@ impl MessageHandler for AdminNetcatHandler { _context: &dyn Context, ) -> MessageHandlerResult { // Connect to server - let mut stream = - timeout(self.config.timeout, TcpStream::connect(&self.config.tcp_addr)) - .await - .map_err(|_| (504u16, "connecting: timeout".into()))? - .map_err(|err| (502u16, format!("connecting: {err}").into()))?; + let mut stream = timeout( + self.config.timeout, + TcpStream::connect(&self.config.tcp_addr), + ) + .await + .map_err(|_| (504u16, "connecting: timeout".into()))? + .map_err(|err| (502u16, format!("connecting: {err}").into()))?; // Write message with CRLF terminator let message = format!("{}\r\n", msg.message); diff --git a/signal-gateway-bin/src/udp_json/json_lines.rs b/signal-gateway-bin/src/json/json_lines.rs similarity index 98% rename from signal-gateway-bin/src/udp_json/json_lines.rs rename to signal-gateway-bin/src/json/json_lines.rs index f478b1c..41c9978 100644 --- a/signal-gateway-bin/src/udp_json/json_lines.rs +++ b/signal-gateway-bin/src/json/json_lines.rs @@ -17,9 +17,7 @@ use tokio::io::{AsyncRead, AsyncReadExt, BufReader}; /// This parser tracks brace and bracket depth to allow newlines within /// JSON objects and arrays. A newline only ends the value when we're at /// depth 0 (outside any object or array). -pub async fn read_json_lines_value( - reader: &mut BufReader, -) -> std::io::Result>> +pub async fn read_json_lines_value(reader: &mut BufReader) -> std::io::Result>> where R: AsyncRead + Unpin, { diff --git a/signal-gateway-bin/src/udp_json/mod.rs b/signal-gateway-bin/src/json/mod.rs similarity index 92% rename from signal-gateway-bin/src/udp_json/mod.rs rename to signal-gateway-bin/src/json/mod.rs index e11a19e..e4226b0 100644 --- a/signal-gateway-bin/src/udp_json/mod.rs +++ b/signal-gateway-bin/src/json/mod.rs @@ -20,17 +20,37 @@ use tracing::{error, info, trace}; mod json_lines; use json_lines::read_json_lines_value; -/// Configuration for the JSON listener (UDP and TCP). +/// Configuration for the JSON log message listener. +/// +/// Listens for JSON log messages on both UDP and TCP using the same address. +/// +/// See [`JsonLogMessage`] for the JSON schema. #[derive(Clone, Conf, Debug)] -pub struct UdpJsonConfig { - /// Socket to listen for JSON log messages. +pub struct JsonConfig { + /// Socket address to listen for JSON log messages. /// Both UDP and TCP listeners are started on this address. - /// TCP uses relaxed JSON Lines format (newlines allowed within objects). + /// + /// - **UDP**: Each datagram should contain a single JSON object. + /// - **TCP**: Uses a relaxed JSON Lines format where each JSON object is + /// separated by newlines. + /// + /// See [`JsonLogMessage`] for the JSON schema. It is roughly compatible + /// with logstash, graylog, etc. + /// + /// * "message", "Message", "msg" for the log mesage string itself + /// * "timestamp", "@timestamp", "time" for the timestamp, which can be a numeric unix typestamp, + /// or a string in RFC3339 form + /// * "level", "severity" for the log level string + /// * "host" or "hostname" for the originating host + /// * "app" or "appname" or "application" or "service" for the originating program + /// * "file" or "filename" or "source_file" for the source file that wrote the log line + /// * "line" or "lineno" for the source file line number that wrote the log line + /// * "module" or "module_path" or "logger" or "logger_name" for the source module that wrote the log line #[conf(long, env)] pub listen_addr: SocketAddr, } -impl UdpJsonConfig { +impl JsonConfig { /// Bind UDP and TCP sockets and start background tasks to handle incoming JSON log messages. /// /// Returns join handles for the background tasks. @@ -51,7 +71,7 @@ impl UdpJsonConfig { 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 + let mut buf = vec![0u8; 8192]; loop { let Ok((len, _addr)) = udp_socket .recv_from(&mut buf) @@ -122,13 +142,11 @@ async fn handle_tcp_connection(stream: TcpStream, gateway: &Gateway) -> std::io: return Ok(()); // Clean EOF }; - let text = std::str::from_utf8(&msg_bytes).map_err(|err| { - std::io::Error::new(std::io::ErrorKind::InvalidData, err) - })?; + let text = std::str::from_utf8(&msg_bytes) + .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; - let json_msg: JsonLogMessage = serde_json::from_str(text).map_err(|err| { - std::io::Error::new(std::io::ErrorKind::InvalidData, err) - })?; + let json_msg: JsonLogMessage = serde_json::from_str(text) + .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; let log_msg = json_msg.into_log_message(); gateway.handle_log_message(log_msg).await; @@ -183,7 +201,7 @@ impl JsonLogMessage { /// /// TODO: Allow this to take configuration options to customize how fields are mapped pub fn into_log_message(self) -> LogMessage { - let level = self.level.unwrap_or(Level::INFO); + let level = self.level.unwrap_or(Level::ERROR); let mut builder = LogMessage::builder(level, self.message); if let Some(ts) = self.timestamp { @@ -309,7 +327,7 @@ mod tests { 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 + assert_eq!(log_msg.level, Level::ERROR); // default } #[test] @@ -524,7 +542,7 @@ mod tests { let log_msg = msg.into_log_message(); assert_eq!(&*log_msg.msg, "simple log"); - assert_eq!(log_msg.level, Level::INFO); + assert_eq!(log_msg.level, Level::ERROR); assert!(log_msg.hostname.is_none()); assert!(log_msg.appname.is_none()); assert!(log_msg.timestamp.is_none()); @@ -535,14 +553,15 @@ mod tests { 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 + // Should default to ERROR when converted let log_msg = msg.into_log_message(); - assert_eq!(log_msg.level, Level::INFO); + assert_eq!(log_msg.level, Level::ERROR); } #[test] fn test_extra_fields_are_ignored() { - let json = r#"{"message": "test", "extra_field": "ignored", "nested": {"also": "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"); } diff --git a/signal-gateway-bin/src/main.rs b/signal-gateway-bin/src/main.rs index efe1856..2daeb4c 100644 --- a/signal-gateway-bin/src/main.rs +++ b/signal-gateway-bin/src/main.rs @@ -18,8 +18,8 @@ use admin_netcat::AdminNetcatConfig; mod syslog; use syslog::SyslogConfig; -mod udp_json; -use udp_json::UdpJsonConfig; +mod json; +use json::JsonConfig; /// Admin message handler configuration - select how non-command messages are handled #[derive(Clone, Debug, Subcommands)] @@ -53,7 +53,7 @@ struct Config { #[conf(flatten, prefix)] syslog: Option, #[conf(flatten, prefix)] - udp_json: Option, + json: Option, /// Optional admin message handler (netcat or http) #[conf(subcommands)] admin_handler: Option, @@ -124,8 +124,8 @@ async fn main() { } else { None }; - let _udp_json_tasks = if let Some(udp_json) = &config.udp_json { - Some(udp_json.start_tasks(gateway.clone()).await.unwrap()) + let _json_tasks = if let Some(json) = &config.json { + Some(json.start_tasks(gateway.clone()).await.unwrap()) } else { None }; diff --git a/signal-gateway-bin/src/syslog/mod.rs b/signal-gateway-bin/src/syslog/mod.rs index 3c5ef69..dd87fd8 100644 --- a/signal-gateway-bin/src/syslog/mod.rs +++ b/signal-gateway-bin/src/syslog/mod.rs @@ -70,9 +70,9 @@ impl SyslogConfig { continue; }; - let Ok(syslog_msg) = SyslogMessage::from_str(text) - .inspect_err(|err| error!("Syslog UDP packet was not valid syslog: {err}:\n{text}")) - else { + let Ok(syslog_msg) = SyslogMessage::from_str(text).inspect_err(|err| { + error!("Syslog UDP packet was not valid syslog: {err}:\n{text}") + }) else { continue; }; @@ -128,7 +128,9 @@ async fn handle_tcp_connection( let mut reader = BufReader::new(stream); loop { - let Some(msg_bytes) = read_framed_syslog_bytes(&mut reader, config.tcp_cr_is_delimiter).await? else { + let Some(msg_bytes) = + read_framed_syslog_bytes(&mut reader, config.tcp_cr_is_delimiter).await? + else { return Ok(()); // Clean EOF }; @@ -178,14 +180,16 @@ where // Octet counting: starts with non-zero digit b'1'..=b'9' => { - return read_octet_counted_message(reader, first_byte).await.map(Some) + return read_octet_counted_message(reader, first_byte) + .await + .map(Some); } // Non-transparent framing: starts with '<' (beginning of syslog PRI) b'<' => { return read_non_transparent_message(reader, first_byte, cr_is_delimiter) .await - .map(Some) + .map(Some); } // Invalid first byte @@ -197,7 +201,7 @@ where first_byte, char::from(first_byte) ), - )) + )); } } } @@ -223,7 +227,10 @@ where .checked_mul(10) .and_then(|l| l.checked_add((b - b'0') as usize)) .ok_or_else(|| { - std::io::Error::new(std::io::ErrorKind::InvalidData, "message length overflow") + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "message length overflow", + ) })?; } b' ' => break, diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index a2f3f4b..af6ea72 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -485,7 +485,9 @@ impl Gateway { sender_uuid: msg.source_uuid.clone(), group_id: data.group_info.as_ref().map(|g| g.group_id.clone()), }; - handler.handle_verified_signal_message(msg, &GatewayContext).await + handler + .handle_verified_signal_message(msg, &GatewayContext) + .await } else { Err((501u16, "No message handler configured".into())) }