Labyricorn/signal-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

listen on udp and tcp for json logs as well

Browse Source
This commit is contained in:
Chris Beck
2025-12-05 15:12:20 -07:00
parent a42aad48f4
commit ffa1246e51
7 changed files with 74 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
signal-gateway-bin/src/admin_http/mod.rs
+6 -4
View File
@@ -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()));
signal-gateway-bin/src/admin_netcat/mod.rs
+7 -5
View File
@@ -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);
signal-gateway-bin/src/udp_json/json_lines.rs → signal-gateway-bin/src/json/json_lines.rs
+1 -3
View File
@@ -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<R>(
reader: &mut BufReader<R>,
) -> std::io::Result<Option<Vec<u8>>>
pub async fn read_json_lines_value<R>(reader: &mut BufReader<R>) -> std::io::Result<Option<Vec<u8>>>
where
R: AsyncRead + Unpin,
{
signal-gateway-bin/src/udp_json/mod.rs → signal-gateway-bin/src/json/mod.rs
+37 -18
View File
@@ -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");
}
signal-gateway-bin/src/main.rs
+5 -5
View File
@@ -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<SyslogConfig>,
#[conf(flatten, prefix)]
udp_json: Option<UdpJsonConfig>,
json: Option<JsonConfig>,
/// Optional admin message handler (netcat or http)
#[conf(subcommands)]
admin_handler: Option<AdminHandlerCommand>,
@@ -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
};
signal-gateway-bin/src/syslog/mod.rs
+15 -8
View File
@@ -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,
signal-gateway/src/gateway/mod.rs
+3 -1
View File
@@ -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()))
}