diff --git a/signal-gateway-bin/src/json/mod.rs b/signal-gateway-bin/src/json/mod.rs index e4226b0..0a543fa 100644 --- a/signal-gateway-bin/src/json/mod.rs +++ b/signal-gateway-bin/src/json/mod.rs @@ -156,6 +156,7 @@ async fn handle_tcp_connection(stream: TcpStream, gateway: &Gateway) -> std::io: /// A flexible JSON log message format compatible with logstash and similar systems. /// /// Supports various field names and formats commonly used in logging systems. +#[non_exhaustive] #[derive(Debug, Deserialize)] pub struct JsonLogMessage { /// The log message text - accepts "message" or "msg" @@ -163,7 +164,7 @@ pub struct JsonLogMessage { pub message: String, /// Log level - accepts various formats (error, ERROR, err, etc.) - #[serde(default, alias = "severity", deserialize_with = "deserialize_level")] + #[serde(default, alias = "severity", deserialize_with = "deserialize_opt_level")] pub level: Option, /// Timestamp - accepts Unix epoch seconds (int or string) or RFC3339 string @@ -228,29 +229,13 @@ impl JsonLogMessage { } } -/// Deserialize a log level from various string formats -fn deserialize_level<'de, D>(deserializer: D) -> Result, D::Error> +/// Deserialize an optional log level, returning None for unknown values. +fn deserialize_opt_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, - } + Ok(opt.and_then(|s| Level::from_str(&s))) } /// Deserialize a timestamp from Unix epoch (int or string) or RFC3339 string diff --git a/signal-gateway-bin/src/main.rs b/signal-gateway-bin/src/main.rs index 2daeb4c..49188fe 100644 --- a/signal-gateway-bin/src/main.rs +++ b/signal-gateway-bin/src/main.rs @@ -1,3 +1,7 @@ +//! Signal Gateway binary - receives alerts and logs, forwards to Signal messenger. + +#![deny(missing_docs)] + use conf::{Conf, Subcommands}; use hyper::service::service_fn; use hyper_util::rt::TokioIo; @@ -18,7 +22,7 @@ use admin_netcat::AdminNetcatConfig; mod syslog; use syslog::SyslogConfig; -mod json; +pub mod json; use json::JsonConfig; /// Admin message handler configuration - select how non-command messages are handled diff --git a/signal-gateway/src/concurrent_map.rs b/signal-gateway/src/concurrent_map.rs new file mode 100644 index 0000000..62c3f27 --- /dev/null +++ b/signal-gateway/src/concurrent_map.rs @@ -0,0 +1,146 @@ +//! A concurrent map with read-preferring access pattern. +//! +//! This module provides a concurrent hash map that optimizes for the common case +//! where values already exist, using a read lock first before falling back to a +//! write lock for insertions. + +use std::collections::HashMap; +use std::hash::Hash; +use tokio::sync::RwLock; + +/// A concurrent hash map that uses read-preferring locking. +/// +/// When accessing a value, it first tries to acquire a read lock. If the key +/// exists, it uses the value immediately. If the key doesn't exist, it upgrades +/// to a write lock and inserts a new value. +#[derive(Debug)] +pub struct ConcurrentMap { + inner: RwLock>, +} + +impl ConcurrentMap +where + K: Eq + Hash + Clone, +{ + /// Create a new empty concurrent map. + pub fn new() -> Self { + Self { + inner: RwLock::new(HashMap::new()), + } + } + + /// Get or insert a value, then access it. + /// + /// This method uses a read-preferring pattern: + /// 1. First acquires a read lock and looks for the key + /// 2. If found, calls `access` with a reference to the value + /// 3. If not found, acquires a write lock, inserts using `create`, then calls `access` + /// + /// The lock is held while `access` runs, so `access` can safely use the reference. + /// For async operations on the value, consider having `access` return a future + /// that owns any data it needs. + pub async fn get_or_insert_with(&self, key: K, create: F, access: A) -> R + where + F: FnOnce() -> V, + A: FnOnce(&V) -> R, + { + // Try to get existing value with read lock first + { + let guard = self.inner.read().await; + if let Some(value) = guard.get(&key) { + return access(value); + } + } + + // Value doesn't exist, need to create with write lock + let mut guard = self.inner.write().await; + // Use entry API - handles the race where another task inserted while we waited + let value = guard.entry(key).or_insert_with(create); + access(value) + } + + /// Access all entries in the map with a read lock. + /// + /// Acquires a read lock and calls `access` with a reference to the underlying HashMap. + /// The lock is held while `access` runs. + pub async fn read_all(&self, access: A) -> R + where + A: FnOnce(&HashMap) -> R, + { + let guard = self.inner.read().await; + access(&guard) + } +} + +impl Default for ConcurrentMap +where + K: Eq + Hash + Clone, +{ + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_get_or_insert_new_key() { + let map: ConcurrentMap = ConcurrentMap::new(); + + let result = map + .get_or_insert_with("key1".to_string(), || 42, |v| *v) + .await; + + assert_eq!(result, 42); + } + + #[tokio::test] + async fn test_get_or_insert_existing_key() { + let map: ConcurrentMap = ConcurrentMap::new(); + + // Insert first time + map.get_or_insert_with("key1".to_string(), || 42, |_| ()) + .await; + + // Access again - should get existing value, not call create + let mut create_called = false; + let result = map + .get_or_insert_with( + "key1".to_string(), + || { + create_called = true; + 100 + }, + |v| *v, + ) + .await; + + assert_eq!(result, 42); + assert!(!create_called); + } + + #[tokio::test] + async fn test_get_or_insert_multiple_keys() { + let map: ConcurrentMap = ConcurrentMap::new(); + + map.get_or_insert_with("a".to_string(), || 1, |_| ()).await; + map.get_or_insert_with("b".to_string(), || 2, |_| ()).await; + map.get_or_insert_with("c".to_string(), || 3, |_| ()).await; + + let a = map + .get_or_insert_with("a".to_string(), || 0, |v| *v) + .await; + let b = map + .get_or_insert_with("b".to_string(), || 0, |v| *v) + .await; + let c = map + .get_or_insert_with("c".to_string(), || 0, |v| *v) + .await; + + assert_eq!(a, 1); + assert_eq!(b, 2); + assert_eq!(c, 3); + } +} diff --git a/signal-gateway/src/gateway/log_handler.rs b/signal-gateway/src/gateway/log_handler.rs index 33fc1f5..43e21cb 100644 --- a/signal-gateway/src/gateway/log_handler.rs +++ b/signal-gateway/src/gateway/log_handler.rs @@ -1,30 +1,45 @@ use super::circular_buffer::CircularBuffer; -use super::{AdminMessage, MultiRateLimiter, RateThreshold, SourceLocationRateLimiter}; +use super::route::{Destination, Limit, Route}; +use super::{AdminMessage, LimitResult, Limiter, LimiterSet}; use crate::{ + concurrent_map::ConcurrentMap, human_duration::HumanTMinus, - log_message::{Level, LogFilter, LogMessage, Origin}, + log_message::{LogMessage, Origin}, }; use chrono::{TimeDelta, Utc}; use conf::Conf; -use serde::Deserialize; -use std::{fmt, time::Duration}; +use std::fmt; use tokio::sync::{Mutex, mpsc::UnboundedSender}; -use tracing::{error, info, warn}; +use tracing::{error, info}; -/// Reason why an alert was suppressed -enum SuppressionReason { - /// Suppressed by a configured alert rule (with 0-based rule index) - Rule(usize), - /// Suppressed by the source-location rate limiter - SourceLocation { file: Box, line: Box }, +/// Reason why an alert was suppressed by rate limiting. +#[derive(Debug)] +pub enum SuppressionReason { + /// No route's filter matched the message. + NoRoutes, + /// Suppressed by route limiters. Contains the index and result for each + /// route whose filter matched but whose limiter blocked the message. + Routes(Vec<(usize, LimitResult)>), + /// Suppressed by an overall limiter. Contains the limiter index and result. + Overall(usize, LimitResult), } impl fmt::Display for SuppressionReason { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - SuppressionReason::Rule(idx) => write!(f, "rule[{idx}]"), - SuppressionReason::SourceLocation { file, line } => { - write!(f, "source-location({file}:{line})") + SuppressionReason::NoRoutes => write!(f, "no-routes"), + SuppressionReason::Routes(failures) => { + write!(f, "routes[")?; + for (i, (idx, result)) in failures.iter().enumerate() { + if i > 0 { + write!(f, ", ")?; + } + write!(f, "{idx}:{result:?}")?; + } + write!(f, "]") + } + SuppressionReason::Overall(idx, result) => { + write!(f, "overall[{idx}]:{result:?}") } } } @@ -33,10 +48,12 @@ impl fmt::Display for SuppressionReason { /// Config options related to the log handler, and what log messages it chooses to alert on. #[derive(Clone, Conf, Debug)] pub struct LogHandlerConfig { - #[conf(long, env, value_parser = serde_json::from_str)] - pub alert_rate_limits: Vec, - #[conf(long, env, default_value = "10m", value_parser = conf_extra::parse_duration)] - pub overall_alert_limit: Duration, + /// Routes for matching and rate-limiting log messages. + #[conf(long, env, value_parser = serde_json::from_str, default_value = "[]")] + pub routes: Vec, + /// Overall rate limits applied after route checks pass. + #[conf(long, env, value_parser = serde_json::from_str, default_value = "[]")] + pub overall_limits: Vec, #[conf(long, env)] pub format_module: bool, #[conf(long, env)] @@ -46,149 +63,154 @@ pub struct LogHandlerConfig { pub log_buffer_size: usize, } -/// Specifies both a rate limiting threshold, and criteria for the threshold to apply -#[derive(Clone, Debug, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct AlertRule { - #[serde(flatten)] - pub filter: LogFilter, - pub threshold: RateThreshold, -} - -/// The log handler takes log messages from a single origin and decides what -/// to do with them. +/// The log handler takes log messages and decides what to do with them. /// -/// 1. Store them in a small circular buffer +/// 1. Store them in a small circular buffer (per origin) /// 2. If it is an error, and meets other criteria, trigger an alert, /// i.e. send a message to admins containing this log and other recent logs. /// 3. The maximum rate of alerts can also be configured. /// /// Additionally, the log handler can format the buffer of recent logs into a string, /// if requested. -/// -/// Each origin (app + host pair) gets its own LogHandler instance, managed by the Gateway. #[derive(Debug)] pub struct LogHandler { config: LogHandlerConfig, admin_mq_tx: UnboundedSender, - 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. - overall_limiter: Mutex, - /// True if any configured rule uses the module structured data field. - any_rule_uses_module: bool, - /// True if any configured rule uses the file structured data field. - any_rule_uses_file: bool, - /// True if any configured rule uses the line structured data field. - any_rule_uses_line: bool, + /// Log buffers keyed by origin (app + host). Lazily created. + log_buffers: ConcurrentMap>>, + /// Routes with their associated limiter sets. + routes: Vec<(Route, Mutex)>, + /// Overall rate limiters applied after route checks pass. + overall_limits: Vec>, } -/// Maximum entries in the source-location rate limiter before triggering cleanup -const OVERALL_LIMITER_MAX_ENTRIES: usize = 2000; - impl LogHandler { /// Initialize a new log handler pub fn new(config: LogHandlerConfig, admin_mq_tx: UnboundedSender) -> Self { - let any_rule_uses_module = config - .alert_rate_limits + let routes = config + .routes .iter() - .any(|r| r.filter.uses_module()); - let any_rule_uses_file = config - .alert_rate_limits - .iter() - .any(|r| r.filter.uses_file()); - let any_rule_uses_line = config - .alert_rate_limits - .iter() - .any(|r| r.filter.uses_line()); - let rate_limiters = config - .alert_rate_limits - .iter() - .map(|rule| { - ( - rule.clone(), - Mutex::new(MultiRateLimiter::from(rule.threshold)), - ) - }) - .collect::>(); - let overall_limiter = Mutex::new(SourceLocationRateLimiter::new( - config.overall_alert_limit, - OVERALL_LIMITER_MAX_ENTRIES, - )); + .map(|route| (route.clone(), Mutex::new(route.make_limiter_set()))) + .collect(); - let log_buffer = Mutex::new(CircularBuffer::new(config.log_buffer_size)); + let overall_limits = config + .overall_limits + .iter() + .map(|limit| Mutex::new(limit.make_limiter())) + .collect(); Self { config, admin_mq_tx, - log_buffer, - rate_limiters, - overall_limiter, - any_rule_uses_module, - any_rule_uses_file, - any_rule_uses_line, + log_buffers: ConcurrentMap::new(), + routes, + overall_limits, } } - /// Format recent logs into a string - pub async fn format_logs(&self) -> String { - let lk = self.log_buffer.lock().await; + /// Format recent logs into a string for all origins, optionally filtered. + /// + /// If `filter` is provided, only origins matching the filter are included. + pub async fn format_logs(&self, filter: Option<&str>) -> String { + self.log_buffers + .read_all(|buffers| { + if buffers.is_empty() { + return "No log sources registered yet".to_string(); + } - let mut text = format!("{} log messages (newest first):\n", lk.len()); + let mut text = String::new(); + let now = Utc::now().timestamp(); - // Calculate now once for consistent relative timestamps - let now = Utc::now().timestamp(); + for (origin, buffer_mutex) in buffers.iter() { + // Apply filter if present + if let Some(f) = filter { + if !origin.matches_filter(f) { + continue; + } + } - // Collect and reverse to show newest first - let messages: Vec<_> = lk.iter().collect(); - for log_msg in messages.into_iter().rev() { - self.write_log_msg(&mut text, log_msg, now); - } - text + // We can't await inside read_all, so use try_lock + // If the buffer is locked, skip it (rare case) + let Some(buffer) = buffer_mutex.try_lock().ok() else { + continue; + }; + + use std::fmt::Write; + writeln!(&mut text, "=== [{origin}] ===").unwrap(); + writeln!(&mut text, "{} log messages (newest first):", buffer.len()).unwrap(); + + // Collect and reverse to show newest first + let messages: Vec<_> = buffer.iter().collect(); + for log_msg in messages.into_iter().rev() { + self.write_log_msg(&mut text, log_msg, now); + } + text.push('\n'); + } + + if text.is_empty() { + "No matching log sources".to_string() + } else { + text + } + }) + .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 - && log_msg.level <= Level::ERROR - { + let ts_sec = *log_msg + .timestamp + .get_or_insert_with(|| Utc::now().timestamp()); + + let rate_limit_result = self.check_rate_limiters(&log_msg, &origin, ts_sec).await; + + if let Err(reason) = &rate_limit_result { 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.log_buffer.lock().await; - lk.push_back(log_msg); - if suppression_reason.is_some() { - return; + let buffer_size = self.config.log_buffer_size; + + // Get or create the buffer for this origin, then record the message + let formatted_text = self + .log_buffers + .get_or_insert_with( + origin.clone(), + || Mutex::new(CircularBuffer::new(buffer_size)), + |buffer_mutex| { + // Lock the buffer and process the message + let mut buffer = buffer_mutex.blocking_lock(); + buffer.push_back(log_msg); + + if rate_limit_result.is_err() { + return None; + } + + let mut text = String::default(); + let now = Utc::now().timestamp(); + + // Iterate in reverse (newest first) without copying + for log_msg in buffer.iter().rev() { + self.write_log_msg(&mut text, log_msg, now); + } + buffer.clear(); + + Some(text) + }, + ) + .await; + + // Send alert if we have formatted text + if let Some(text) = formatted_text { + // TODO: Use destination override from rate_limit_result.ok() if present + if let Err(_err) = self.admin_mq_tx.send(AdminMessage { + origin: Some(origin), + text, + attachment_paths: Default::default(), + summary: None, + }) { + error!("Could not send alert message, queue is closed"); } - - let mut text = String::default(); - - // Calculate now once for consistent relative timestamps - let now = Utc::now().timestamp(); - - // Iterate in reverse (newest first) without copying - for log_msg in lk.iter().rev() { - self.write_log_msg(&mut text, log_msg, now); - } - lk.clear(); - - text - }; - - if let Err(_err) = self.admin_mq_tx.send(AdminMessage { - origin: Some(origin), - text: formatted_text, - attachment_paths: Default::default(), - summary: None, - }) { - error!("Could not send alert message, queue is closed"); } } @@ -251,63 +273,75 @@ impl LogHandler { } } - /// Check if an alert should be suppressed for a given error message. + /// Check if a log message passes all rate limiters. /// - /// Returns `None` if the alert should fire, or `Some(reason)` if suppressed. - async fn check_suppression(&self, log_msg: &mut LogMessage) -> Option { - let ts_sec = *log_msg - .timestamp - .get_or_insert_with(|| Utc::now().timestamp()); + /// Tests the message against each route's filter in succession (no early return). + /// For routes where the filter matches, evaluates the limiter set. + /// + /// Returns: + /// - `Ok(Some(destination))` if passed and a route specified a destination override + /// - `Ok(None)` if passed with no destination override + /// - `Err(SuppressionReason::Routes(...))` if no route's limiter passed + /// - `Err(SuppressionReason::Overall(...))` if routes passed but overall limiter failed + async fn check_rate_limiters( + &self, + log_msg: &LogMessage, + origin: &Origin, + ts_sec: i64, + ) -> Result, SuppressionReason> { + let mut route_failures: Vec<(usize, LimitResult)> = Vec::new(); + let mut first_passed_destination: Option> = None; - 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)); - } + // Test against each route's filter and limiter + for (idx, (route, limiter_set)) in self.routes.iter().enumerate() { + // Check if message level meets route's alert threshold + if log_msg.level > route.alert_level { + continue; + } - // Warn if rules expect structured data but the message doesn't have it - 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!( - "Log message missing source location data, filtering rules may not work: {log_msg:#?}" - ); - } + // Check if message matches route's filter (if any) + let filter_matches = route + .filter + .as_ref() + .map_or(true, |f| f.matches(log_msg)); - // Check each configured rule - track which rule suppressed the alert - // Note: we check all rules even if one already suppressed, to update all rate limiters - let mut suppressed_by_rule: Option = None; - for (idx, (rule, limiter)) in self.rate_limiters.iter().enumerate() { - if rule.filter.matches(log_msg) && !limiter.lock().await.evaluate(ts_sec) { - suppressed_by_rule.get_or_insert(idx); + if !filter_matches { + continue; + } + + // Filter matched, evaluate the limiter set + let result = limiter_set.lock().await.evaluate(log_msg, origin, ts_sec); + + match result { + LimitResult::Passed => { + // Remember the first route that passed + if first_passed_destination.is_none() { + first_passed_destination = Some(route.destination.clone()); + } + } + _ => { + // Record the failure + route_failures.push((idx, result)); + } } } - if let Some(idx) = suppressed_by_rule { - return Some(SuppressionReason::Rule(idx)); + + // If no route passed, return the appropriate error + let first_destination = match first_passed_destination { + Some(dest) => dest, + None if route_failures.is_empty() => return Err(SuppressionReason::NoRoutes), + None => return Err(SuppressionReason::Routes(route_failures)), + }; + + // At least one route passed, now check overall limits + for (idx, limiter) in self.overall_limits.iter().enumerate() { + if !limiter.lock().await.evaluate(log_msg, ts_sec) { + return Err(SuppressionReason::Overall(idx, LimitResult::Limiter(0))); + } } - // Extract source location for per-location rate limiting - let file = log_msg.file.as_deref().unwrap_or("?"); - let line = log_msg.line.as_deref().unwrap_or("?"); - - if !self - .overall_limiter - .lock() - .await - .evaluate(file, line, ts_sec) - { - return Some(SuppressionReason::SourceLocation { - file: file.into(), - line: line.into(), - }); - } - - None // Alert should fire + // All checks passed + Ok(first_destination) } } diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index af6ea72..cd39ef8 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -25,7 +25,7 @@ use std::{collections::HashMap, fmt::Write, net::SocketAddr, path::PathBuf}; use tokio::{ join, sync::{ - Mutex, RwLock, + Mutex, mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}, }, }; @@ -37,8 +37,11 @@ mod circular_buffer; mod log_handler; use log_handler::{LogHandler, LogHandlerConfig}; +mod route; +pub use route::{Destination, Limit, RateThreshold, Route}; + mod rate_limiter; -use rate_limiter::{MultiRateLimiter, RateThreshold, SourceLocationRateLimiter}; +pub use rate_limiter::{LimitResult, Limiter, LimiterSet}; /// Configuration for the gateway. #[derive(Conf, Debug)] @@ -186,8 +189,8 @@ pub struct Gateway { admin_mq_rx: Mutex>, token: CancellationToken, prometheus: Option, - /// Log handlers keyed by origin (app + host). Lazily created when first message from an origin arrives. - log_handlers: RwLock>, + /// Log handler for processing log messages from all origins. + log_handler: LogHandler, /// Handler for admin messages that don't start with `/` message_handler: Option>, } @@ -208,13 +211,15 @@ impl Gateway { .transpose() .expect("Invalid prometheus config"); + let log_handler = LogHandler::new(config.log_handler.clone(), admin_mq_tx.clone()); + Self { config, admin_mq_tx, admin_mq_rx: Mutex::new(admin_mq_rx), token, prometheus, - log_handlers: RwLock::new(HashMap::new()), + log_handler, message_handler, } } @@ -558,25 +563,10 @@ impl Gateway { async fn handle_gateway_command(&self, cmd: GatewayCommand) -> MessageHandlerResult { match cmd { GatewayCommand::Log { filter } => { - let handlers = self.log_handlers.read().await; - if handlers.is_empty() { - return Ok(AdminMessageResponse::new("No log sources registered yet")); - } - let mut text = String::new(); - for (origin, handler) in handlers.iter() { - // Apply filter if present - if let Some(ref f) = filter - && !origin.matches_filter(f) - { - continue; - } - writeln!(&mut text, "=== [{origin}] ===").unwrap(); - text.push_str(&handler.format_logs().await); - text.push('\n'); - } - if text.is_empty() { - return Ok(AdminMessageResponse::new("No matching log sources")); - } + let text = self + .log_handler + .format_logs(filter.as_deref()) + .await; Ok(AdminMessageResponse::new(text)) } GatewayCommand::Query { query } => { @@ -826,24 +816,7 @@ impl Gateway { 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_log_message(log_msg, origin).await; - return; - } - } - - // Handler doesn't exist, need to create one with write lock - let mut handlers = self.log_handlers.write().await; - // Double-check in case another task created it while we were waiting for the write lock - let handler = handlers.entry(origin.clone()).or_insert_with(|| { - info!("Creating new log handler for origin: {origin}"); - LogHandler::new(self.config.log_handler.clone(), self.admin_mq_tx.clone()) - }); - handler.handle_log_message(log_msg, origin).await; + self.log_handler.handle_log_message(log_msg, origin).await; } } diff --git a/signal-gateway/src/gateway/rate_limiter.rs b/signal-gateway/src/gateway/rate_limiter.rs index 794fbda..fcedc66 100644 --- a/signal-gateway/src/gateway/rate_limiter.rs +++ b/signal-gateway/src/gateway/rate_limiter.rs @@ -1,72 +1,113 @@ -use serde::Deserialize; +use super::route::{Limit, RateThreshold}; +use crate::log_message::{LogMessage, Origin}; use std::{ collections::HashMap, - str::FromStr, sync::atomic::{AtomicI64, Ordering}, time::Duration, }; -/// Represents a rate threshold, expressed as a string in the format: -/// -/// * `1 / 10s` -/// * `2 / 5m` -/// * `3 / 1h` -/// * `> 1 / 10s` -/// * `>= 2 / 10s` -/// -/// When the comparator is omitted, it is treated as `>=` -#[derive(Clone, Copy, Debug, Deserialize)] -#[serde(try_from = "String")] -pub struct RateThreshold { - pub times: usize, - pub duration: Duration, +/// Maximum entries in a source-location rate limiter before triggering cleanup. +const SOURCE_LOCATION_MAX_ENTRIES: usize = 2000; + +/// A rate limiter that can be either a multi-rate limiter or a source-location limiter. +#[derive(Debug)] +pub enum Limiter { + /// Counts events regardless of source location. + Multi(MultiRateLimiter), + /// Tracks events independently per source location (file:line). + SourceLocation(SourceLocationRateLimiter), } -impl FromStr for RateThreshold { - type Err = String; +/// Result of evaluating a limiter set. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum LimitResult { + /// The event passed all limits (not rate-limited). + Passed, + /// The event was blocked by a per-origin limiter at the given index. + Limiter(usize), + /// The event was blocked by a global limiter at the given index. + GlobalLimiter(usize), +} - fn from_str(s: &str) -> Result { - let Some((first, second)) = s.trim().split_once('/') else { - return Err("missing '/' character in rate threshold".into()); - }; +/// A set of limiters for a route, containing both per-origin and global limiters. +#[derive(Debug)] +pub struct LimiterSet { + /// Limit configurations (used to create limiters for new origins). + limits: Vec, + /// Per-origin rate limiters, keyed by origin. Lazily created. + limiters: HashMap>, + /// Global rate limiters (shared across all origins). + global_limiters: Vec, +} - let duration = conf_extra::parse_duration(second.trim())?; +impl LimiterSet { + /// Create a new limiter set from limit configurations. + pub fn new(limits: Vec, global_limits: Vec) -> Self { + Self { + limits, + limiters: HashMap::new(), + global_limiters: global_limits.iter().map(|l| l.make_limiter()).collect(), + } + } - let first = first.trim(); - let maybe_mid = first.as_bytes().iter().position(|b| b.is_ascii_digit()); - let (comparator, num) = if let Some(mid) = maybe_mid { - first.split_at(mid) - } else { - ("", first) - }; + /// Evaluate whether an event should pass all rate limits. + /// + /// Returns [`LimitResult::Passed`] if the event passes all limits. + /// Returns [`LimitResult::Limiter(i)`] if blocked by per-origin limiter at index `i`. + /// Returns [`LimitResult::GlobalLimiter(i)`] if blocked by global limiter at index `i`. + pub fn evaluate(&mut self, log_msg: &LogMessage, origin: &Origin, ts_sec: i64) -> LimitResult { + // Get or create limiters for this origin + let origin_limiters = self + .limiters + .entry(origin.clone()) + .or_insert_with(|| self.limits.iter().map(|l| l.make_limiter()).collect()); - let is_greater_equal = match comparator.trim() { - ">" => false, - ">=" | "=>" | "" => true, - _ => return Err(format!("Unexpected comparator format: {comparator}")), - }; - - let num = num.trim(); - let mut times: usize = num - .parse() - .map_err(|err| format!("invalid number {num}: {err}"))?; - - if !is_greater_equal { - times += 1; + for (i, limiter) in origin_limiters.iter_mut().enumerate() { + if !limiter.evaluate(log_msg, ts_sec) { + return LimitResult::Limiter(i); + } } - if times == 0 { - return Err("Invalid threshold, times must be > 0".into()); + for (i, limiter) in self.global_limiters.iter_mut().enumerate() { + if !limiter.evaluate(log_msg, ts_sec) { + return LimitResult::GlobalLimiter(i); + } } - Ok(RateThreshold { times, duration }) + LimitResult::Passed } } -impl TryFrom for RateThreshold { - type Error = ::Err; - fn try_from(s: String) -> Result { - RateThreshold::from_str(&s) +impl Limiter { + /// Evaluate whether an event should pass the rate limit. + /// + /// Returns `true` if the event should be allowed (not rate-limited), + /// `false` if it should be suppressed. + pub fn evaluate(&mut self, log_msg: &LogMessage, ts_sec: i64) -> bool { + match self { + Limiter::Multi(limiter) => limiter.evaluate(ts_sec), + Limiter::SourceLocation(limiter) => { + let file = log_msg.file.as_deref().unwrap_or("?"); + let line = log_msg.line.as_deref().unwrap_or("?"); + limiter.evaluate(file, line, ts_sec) + } + } + } + + /// Create a multi-rate limiter from a threshold. + pub fn multi(threshold: RateThreshold) -> Self { + Limiter::Multi(MultiRateLimiter::from(threshold)) + } + + /// Create a source-location limiter from a threshold. + /// + /// Note: Only the duration is used; source-location limiters allow one event + /// per location per window. + pub fn source_location(threshold: RateThreshold) -> Self { + Limiter::SourceLocation(SourceLocationRateLimiter::new( + threshold.duration, + SOURCE_LOCATION_MAX_ENTRIES, + )) } } @@ -210,6 +251,7 @@ impl From for MultiRateLimiter { #[cfg(test)] mod tests { use super::*; + use std::str::FromStr; #[test] fn parse_rate_threshold() { diff --git a/signal-gateway/src/gateway/route.rs b/signal-gateway/src/gateway/route.rs new file mode 100644 index 0000000..2928c7e --- /dev/null +++ b/signal-gateway/src/gateway/route.rs @@ -0,0 +1,170 @@ +//! Route configuration for log message handling. +//! +//! Routes define how log messages are processed based on filters, severity levels, +//! and destination overrides. + +use crate::log_message::{Level, LogFilter}; +use serde::Deserialize; +use std::{str::FromStr, time::Duration}; + +/// Represents a rate threshold, expressed as a string in the format: +/// +/// * `1 / 10s` +/// * `2 / 5m` +/// * `3 / 1h` +/// * `> 1 / 10s` +/// * `>= 2 / 10s` +/// +/// When the comparator is omitted, it is treated as `>=` +#[derive(Clone, Copy, Debug, Deserialize)] +#[serde(try_from = "String")] +pub struct RateThreshold { + /// Number of events required to trigger the threshold. + pub times: usize, + /// Time window for counting events. + pub duration: Duration, +} + +impl FromStr for RateThreshold { + type Err = String; + + fn from_str(s: &str) -> Result { + let Some((first, second)) = s.trim().split_once('/') else { + return Err("missing '/' character in rate threshold".into()); + }; + + let duration = conf_extra::parse_duration(second.trim())?; + + let first = first.trim(); + let maybe_mid = first.as_bytes().iter().position(|b| b.is_ascii_digit()); + let (comparator, num) = if let Some(mid) = maybe_mid { + first.split_at(mid) + } else { + ("", first) + }; + + let is_greater_equal = match comparator.trim() { + ">" => false, + ">=" | "=>" | "" => true, + _ => return Err(format!("Unexpected comparator format: {comparator}")), + }; + + let num = num.trim(); + let mut times: usize = num + .parse() + .map_err(|err| format!("invalid number {num}: {err}"))?; + + if !is_greater_equal { + times += 1; + } + + if times == 0 { + return Err("Invalid threshold, times must be > 0".into()); + } + + Ok(RateThreshold { times, duration }) + } +} + +impl TryFrom for RateThreshold { + type Error = ::Err; + fn try_from(s: String) -> Result { + RateThreshold::from_str(&s) + } +} + +/// A route configuration for processing log messages. +/// +/// Routes match incoming log messages based on an optional filter, then apply +/// the configured alert level threshold. Each route can optionally override +/// the default destination and define rate limits. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Route { + /// Minimum severity level for messages to trigger an alert. + /// Messages at this level or higher (lower numeric value) will alert. + /// Default: ERROR + #[serde(default = "default_alert_level")] + pub alert_level: Level, + + /// Optional filter to match log messages for this route. + /// If not specified, the route matches all messages. + #[serde(default)] + pub filter: Option, + + /// Optional destination override for alerts from this route. + /// If not specified, alerts go to the default admin destination. + #[serde(default)] + pub destination: Option, + + /// Rate limits applied per-origin for messages matching this route. + /// Each limit specifies a filter and threshold for suppressing repeated alerts. + #[serde(default)] + pub limit: Vec, + + /// Global rate limits applied across all origins for this route. + /// Each limit specifies a filter and threshold for suppressing repeated alerts. + #[serde(default)] + pub global_limit: Vec, +} + +fn default_alert_level() -> Level { + Level::ERROR +} + +/// Destination override for alert messages. +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Destination { + /// Send to specific recipient UUIDs. + Recipients(Vec), + /// Send to a Signal group by group ID. + Group(String), +} + +/// A rate limit rule for suppressing repeated alerts. +/// +/// Combines a filter to match specific log messages with a rate threshold. +#[derive(Clone, Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Limit { + /// Filter criteria for messages this limit applies to. + #[serde(flatten)] + pub filter: LogFilter, + /// Rate threshold for suppressing alerts. + pub threshold: RateThreshold, + /// If true, rate limit independently per source location (file:line). + /// If false (default), count all matching events together. + #[serde(default)] + pub by_source_location: bool, +} + +impl Limit { + /// Create the appropriate limiter for this limit configuration. + pub fn make_limiter(&self) -> super::rate_limiter::Limiter { + if self.by_source_location { + super::rate_limiter::Limiter::source_location(self.threshold) + } else { + super::rate_limiter::Limiter::multi(self.threshold) + } + } +} + +impl Route { + /// Create a limiter set from this route's limit configurations. + pub fn make_limiter_set(&self) -> super::rate_limiter::LimiterSet { + super::rate_limiter::LimiterSet::new(self.limit.clone(), self.global_limit.clone()) + } +} + +impl Default for Route { + fn default() -> Self { + Self { + alert_level: default_alert_level(), + filter: None, + destination: None, + limit: Vec::new(), + global_limit: Vec::new(), + } + } +} diff --git a/signal-gateway/src/lib.rs b/signal-gateway/src/lib.rs index 0c4e6b2..ed13fad 100644 --- a/signal-gateway/src/lib.rs +++ b/signal-gateway/src/lib.rs @@ -9,6 +9,7 @@ pub mod alertmanager; pub mod gateway; pub mod message_handler; +pub(crate) mod concurrent_map; pub(crate) mod human_duration; pub(crate) mod jsonrpc; pub(crate) mod log_message; diff --git a/signal-gateway/src/log_message.rs b/signal-gateway/src/log_message.rs index f3f3345..4c452fc 100644 --- a/signal-gateway/src/log_message.rs +++ b/signal-gateway/src/log_message.rs @@ -1,6 +1,6 @@ //! Log message schema and types. -use serde::Deserialize; +use serde::{Deserialize, de}; /// Log severity level, following syslog conventions. /// @@ -44,6 +44,48 @@ impl Level { Self::TRACE => "TRACE", } } + + /// Parse a level from a string (case-insensitive). + /// + /// Accepts various common aliases: + /// - emergency, emerg + /// - alert + /// - critical, crit, fatal + /// - error, err + /// - warning, warn + /// - notice, note + /// - info, information + /// - debug + /// - trace + pub fn from_str(s: &str) -> Option { + match s.to_ascii_uppercase().as_str() { + "EMERGENCY" | "EMERG" => Some(Self::EMERGENCY), + "ALERT" => Some(Self::ALERT), + "CRITICAL" | "CRIT" | "FATAL" => Some(Self::CRITICAL), + "ERROR" | "ERR" => Some(Self::ERROR), + "WARNING" | "WARN" => Some(Self::WARNING), + "NOTICE" | "NOTE" => Some(Self::NOTICE), + "INFO" | "INFORMATION" => Some(Self::INFO), + "DEBUG" => Some(Self::DEBUG), + "TRACE" => Some(Self::TRACE), + _ => None, + } + } +} + +impl<'de> Deserialize<'de> for Level { + fn deserialize(deserializer: D) -> Result + where + D: de::Deserializer<'de>, + { + let s = String::deserialize(deserializer)?; + Level::from_str(&s).ok_or_else(|| { + de::Error::custom(format!( + "unknown log level '{}', expected one of: emergency, alert, critical, error, warning, notice, info, debug, trace", + s + )) + }) + } } /// A structured log message.