//! Alert rules: definitions, trigger evaluation, and ntfy delivery. //! //! Alert queries are restricted to regex/sequence/timestamp/json[field] — a metric[param] comparison would require a full bucket rescan per line, breaking the proportional-cost guarantee. json[field] is fine here since, like regex/sequence/timestamp, it only inspects the line's own text/fields, no index lookup. use std::collections::hash_map::DefaultHasher; use std::collections::{HashMap, HashSet, VecDeque}; use std::fmt; use std::hash::{Hash, Hasher}; use std::io; use std::path::Path; use std::time::{SystemTime, UNIX_EPOCH}; use regex::Regex; use serde::{Deserialize, Serialize}; use crate::engine::{apply_compare_op, eval_json_comparison}; use crate::filter_datetime; use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; use crate::settings::Settings; use crate::timestamp::Timestamp; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct AlertRule { pub id: String, pub name: String, pub enabled: bool, pub condition: AlertCondition, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "lowercase")] pub enum AlertCondition { Immediate { query: String }, /// Fires when `query`'s match count within a rolling `window_ms` crosses `threshold` (via `op`). Threshold { query: String, op: CompareOp, threshold: f64, window_ms: i64, }, /// Seen values live only in memory — a restart forgets them, so previously-seen values can fire again; a deliberate simplicity tradeoff, not an oversight. NewValue { field_regex: String, capture_group: String, }, /// A zero `count_b` skips evaluation for that line rather than treating the ratio as infinite. Ratio { query_a: String, query_b: String, op: CompareOp, ratio: f64, window_ms: i64, }, } #[derive(Debug)] pub enum AlertError { NotFound, Io(io::Error), } impl fmt::Display for AlertError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { AlertError::NotFound => write!(f, "no alert rule with that id"), AlertError::Io(err) => write!(f, "I/O error: {err}"), } } } impl std::error::Error for AlertError {} impl From for AlertError { fn from(err: io::Error) -> Self { AlertError::Io(err) } } /// Hashes the name and current time rather than pulling in a `uuid` dependency. fn generate_alert_rule_id(name: &str) -> String { let mut hasher = DefaultHasher::new(); name.hash(&mut hasher); let nanos = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_nanos()) .unwrap_or(0); nanos.hash(&mut hasher); format!("alert-{:016x}", hasher.finish()) } impl Settings { pub fn create_alert_rule( &mut self, settings_path: &Path, name: String, enabled: bool, condition: AlertCondition, ) -> Result { let rule = AlertRule { id: generate_alert_rule_id(&name), name, enabled, condition }; self.alert_rules.push(rule.clone()); self.save(settings_path)?; Ok(rule) } pub fn update_alert_rule( &mut self, settings_path: &Path, id: &str, name: String, enabled: bool, condition: AlertCondition, ) -> Result { let rule = self .alert_rules .iter_mut() .find(|rule| rule.id == id) .ok_or(AlertError::NotFound)?; rule.name = name; rule.enabled = enabled; rule.condition = condition; let updated = rule.clone(); self.save(settings_path)?; Ok(updated) } pub fn remove_alert_rule(&mut self, settings_path: &Path, id: &str) -> Result<(), AlertError> { let index = self .alert_rules .iter() .position(|rule| rule.id == id) .ok_or(AlertError::NotFound)?; self.alert_rules.remove(index); self.save(settings_path)?; Ok(()) } } /// Per-rule in-memory evaluation state; lives only for the process's lifetime — not persisted, not backfilled against already-ingested logs. #[derive(Debug, Default)] pub struct AlertRuntimeState { threshold_matches: VecDeque, ratio_a_matches: VecDeque, ratio_b_matches: VecDeque, seen_values: HashSet, } /// `timestamps` is push_back-ordered, so once the front entry is within the window, every entry behind it is too. fn prune_window(timestamps: &mut VecDeque, now_ms: i64, window_ms: i64) { while let Some(&front) = timestamps.front() { if now_ms - front > window_ms { timestamps.pop_front(); } else { break; } } } /// Evaluates a FilterBar query against a single line's own fields only (metric[param] comparisons aren't supported — see module doc; json[field] is supported since it's also line-local). fn eval_alert_query(query: &str, sequence: u64, timestamp: Timestamp, text: &str) -> Result { let expr = filter_parser::parse(query).map_err(|err| err.to_string())?; eval_alert_expr(&expr, sequence, timestamp, text) } fn eval_alert_expr(expr: &FilterExpr, sequence: u64, timestamp: Timestamp, text: &str) -> Result { match expr { FilterExpr::Regex(pattern) => { let re = Regex::new(pattern).map_err(|err| format!("invalid regex in alert query: {err}"))?; Ok(re.is_match(text)) } FilterExpr::Comparison { metric, param: None, op, value } if metric == "sequence" => { let target = match value { FilterValue::Number(n) => *n, FilterValue::Raw(raw) => raw .parse::() .map_err(|_| format!("'sequence' comparisons require a numeric value, got {raw:?}"))?, }; Ok(apply_compare_op((sequence as f64).total_cmp(&target), *op)) } FilterExpr::Comparison { metric, param: None, op, value } if metric == "timestamp" => { let target_ms = match value { FilterValue::Number(n) => *n as i64, FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, Timestamp::now()) .map_err(|err| err.to_string())? .as_millis(), }; Ok(apply_compare_op(timestamp.as_millis().cmp(&target_ms), *op)) } FilterExpr::Comparison { metric, .. } => Err(format!( "alert queries only support regex()/sequence/timestamp (not metric comparisons like \ '{metric}[param]'): metric-based alert conditions aren't supported yet" )), FilterExpr::JsonComparison { field, op, value } => { Ok(eval_json_comparison(text, field, *op, value)) } FilterExpr::And(lhs, rhs) => { Ok(eval_alert_expr(lhs, sequence, timestamp, text)? && eval_alert_expr(rhs, sequence, timestamp, text)?) } FilterExpr::Or(lhs, rhs) => { Ok(eval_alert_expr(lhs, sequence, timestamp, text)? || eval_alert_expr(rhs, sequence, timestamp, text)?) } } } /// A malformed query (bad regex, unsupported metric comparison) is logged and simply never matches, rather than panicking. pub fn evaluate_alerts( rules: &[AlertRule], states: &mut HashMap, sequence: u64, timestamp: Timestamp, text: &str, ) -> Vec<(AlertRule, String)> { let mut fired = Vec::new(); for rule in rules { if !rule.enabled { continue; } let state = states.entry(rule.id.clone()).or_default(); if let Some(message) = evaluate_condition(&rule.condition, state, sequence, timestamp, text, &rule.name) { fired.push((rule.clone(), message)); } } fired } fn evaluate_condition( condition: &AlertCondition, state: &mut AlertRuntimeState, sequence: u64, timestamp: Timestamp, text: &str, rule_name: &str, ) -> Option { match condition { AlertCondition::Immediate { query } => match eval_alert_query(query, sequence, timestamp, text) { Ok(true) => Some(format!("{rule_name}: matched line: {text}")), Ok(false) => None, Err(err) => { eprintln!("warning: alert '{rule_name}' query error: {err}"); None } }, AlertCondition::Threshold { query, op, threshold, window_ms } => { match eval_alert_query(query, sequence, timestamp, text) { Ok(true) => { let now_ms = timestamp.as_millis(); state.threshold_matches.push_back(now_ms); prune_window(&mut state.threshold_matches, now_ms, *window_ms); let count = state.threshold_matches.len() as f64; apply_compare_op(count.total_cmp(threshold), *op).then(|| { format!("{rule_name}: match count {count} {op:?} threshold {threshold} (window {window_ms}ms)") }) } Ok(false) => None, Err(err) => { eprintln!("warning: alert '{rule_name}' query error: {err}"); None } } } AlertCondition::NewValue { field_regex, capture_group } => { let re = match Regex::new(field_regex) { Ok(re) => re, Err(err) => { eprintln!("warning: alert '{rule_name}' has an invalid field_regex: {err}"); return None; } }; let captures = re.captures(text)?; let value = captures.name(capture_group)?.as_str().to_string(); state .seen_values .insert(value.clone()) .then(|| format!("{rule_name}: new value seen: {value} (line: {text})")) } AlertCondition::Ratio { query_a, query_b, op, ratio, window_ms } => { let matched_a = eval_alert_query(query_a, sequence, timestamp, text); let matched_b = eval_alert_query(query_b, sequence, timestamp, text); let (matched_a, matched_b) = match (matched_a, matched_b) { (Ok(a), Ok(b)) => (a, b), (Err(err), _) | (_, Err(err)) => { eprintln!("warning: alert '{rule_name}' query error: {err}"); return None; } }; if !matched_a && !matched_b { return None; } let now_ms = timestamp.as_millis(); if matched_a { state.ratio_a_matches.push_back(now_ms); prune_window(&mut state.ratio_a_matches, now_ms, *window_ms); } if matched_b { state.ratio_b_matches.push_back(now_ms); prune_window(&mut state.ratio_b_matches, now_ms, *window_ms); } let count_a = state.ratio_a_matches.len() as f64; let count_b = state.ratio_b_matches.len() as f64; if count_b == 0.0 { // Zero denominator: skip rather than treat the ratio as infinite. return None; } let observed = count_a / count_b; apply_compare_op(observed.total_cmp(ratio), *op).then(|| { format!( "{rule_name}: ratio {observed:.3} ({count_a}/{count_b}) {op:?} {ratio} (window {window_ms}ms)" ) }) } } } /// Delivers to the single, process-wide ntfy sink configured in Settings. Errors are only logged /// (no retry queue); spawned off the ingest hot path so a slow/unreachable sink never blocks ingestion. /// A silently no-op'd delivery (no topic configured) is expected — a rule can be defined before the sink is. pub async fn deliver_for_rule(rule_name: String, message: String, topic: Option, server: Option) { let Some(topic) = topic else { return; }; let server = server.unwrap_or_else(|| "https://ntfy.sh".to_string()); if let Err(err) = deliver_ntfy(&server, &topic, &message).await { eprintln!("warning: failed to deliver alert '{rule_name}' via ntfy: {err}"); } } /// Standalone and `AlertRule`-free so a "send test notification" endpoint can call it directly. pub async fn deliver_ntfy(server: &str, topic: &str, message: &str) -> Result<(), String> { let url = format!("{}/{}", server.trim_end_matches('/'), topic); let response = reqwest::Client::new() .post(&url) .body(message.to_string()) .send() .await .map_err(|err| err.to_string())?; if response.status().is_success() { Ok(()) } else { Err(format!("ntfy responded with status {}", response.status())) } } #[cfg(test)] mod tests { use super::*; fn line(text: &str, sequence: u64, timestamp_ms: i64) -> (u64, Timestamp, &str) { (sequence, Timestamp::from_millis(timestamp_ms), text) } fn immediate_rule(query: &str) -> AlertRule { AlertRule { id: "r1".to_string(), name: "test rule".to_string(), enabled: true, condition: AlertCondition::Immediate { query: query.to_string() }, } } #[test] fn alert_rule_crud_round_trips_through_save_and_load() { let dir = tempfile::tempdir().unwrap(); let settings_path = dir.path().join("settings.json"); let mut settings = Settings::load(&settings_path); let created = settings .create_alert_rule( &settings_path, "spike".to_string(), true, AlertCondition::Immediate { query: r#"regex("error")"#.to_string() }, ) .unwrap(); assert_eq!(settings.alert_rules.len(), 1); let reloaded = Settings::load(&settings_path); assert_eq!(reloaded.alert_rules, settings.alert_rules); let updated = settings .update_alert_rule( &settings_path, &created.id, "renamed".to_string(), false, AlertCondition::Immediate { query: r#"regex("warn")"#.to_string() }, ) .unwrap(); assert_eq!(updated.id, created.id); assert_eq!(updated.name, "renamed"); assert!(!updated.enabled); settings.remove_alert_rule(&settings_path, &created.id).unwrap(); assert!(settings.alert_rules.is_empty()); let reloaded = Settings::load(&settings_path); assert!(reloaded.alert_rules.is_empty()); } #[test] fn alert_rule_mutations_reject_unknown_id() { let dir = tempfile::tempdir().unwrap(); let settings_path = dir.path().join("settings.json"); let mut settings = Settings::load(&settings_path); let result = settings.update_alert_rule( &settings_path, "does-not-exist", "x".to_string(), true, AlertCondition::Immediate { query: r#"regex("x")"#.to_string() }, ); assert!(matches!(result, Err(AlertError::NotFound))); let result = settings.remove_alert_rule(&settings_path, "does-not-exist"); assert!(matches!(result, Err(AlertError::NotFound))); } #[test] fn immediate_fires_on_matching_line_only() { let rules = vec![immediate_rule(r#"regex("error")"#)]; let mut states = HashMap::new(); let (seq, ts, text) = line("all good", 0, 0); assert!(evaluate_alerts(&rules, &mut states, seq, ts, text).is_empty()); let (seq, ts, text) = line("an error happened", 1, 1); let fired = evaluate_alerts(&rules, &mut states, seq, ts, text); assert_eq!(fired.len(), 1); assert!(fired[0].1.contains("an error happened")); } #[test] fn disabled_rule_never_fires() { let mut rule = immediate_rule(r#"regex("error")"#); rule.enabled = false; let mut states = HashMap::new(); let (seq, ts, text) = line("error!", 0, 0); assert!(evaluate_alerts(&[rule], &mut states, seq, ts, text).is_empty()); } #[test] fn threshold_fires_once_count_crosses_and_not_before() { let rule = AlertRule { id: "r1".to_string(), name: "spike".to_string(), enabled: true, condition: AlertCondition::Threshold { query: r#"regex("error")"#.to_string(), op: CompareOp::Ge, threshold: 3.0, window_ms: 10_000, }, }; let mut states = HashMap::new(); for (seq, ms) in [(0u64, 0i64), (1, 1000)] { let fired = evaluate_alerts(&[rule.clone()], &mut states, seq, Timestamp::from_millis(ms), "error"); assert!(fired.is_empty(), "should not fire before threshold is crossed"); } let fired = evaluate_alerts(&[rule.clone()], &mut states, 2, Timestamp::from_millis(2000), "error"); assert_eq!(fired.len(), 1); // Stays quiet for a non-matching line even though the window is full. let fired = evaluate_alerts(&[rule], &mut states, 3, Timestamp::from_millis(2100), "fine"); assert!(fired.is_empty()); } #[test] fn threshold_window_expires_old_matches() { let rule = AlertRule { id: "r1".to_string(), name: "spike".to_string(), enabled: true, condition: AlertCondition::Threshold { query: r#"regex("error")"#.to_string(), op: CompareOp::Ge, threshold: 2.0, window_ms: 1_000, }, }; let mut states = HashMap::new(); assert!(evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "error").is_empty()); // Second match arrives well after the first has aged out of the window. assert!(evaluate_alerts(&[rule], &mut states, 1, Timestamp::from_millis(5_000), "error").is_empty()); } #[test] fn new_value_fires_only_on_first_sighting() { let rule = AlertRule { id: "r1".to_string(), name: "new user agent".to_string(), enabled: true, condition: AlertCondition::NewValue { field_regex: r"agent=(?P\S+)".to_string(), capture_group: "agent".to_string(), }, }; let mut states = HashMap::new(); let fired = evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "agent=curl"); assert_eq!(fired.len(), 1); let fired = evaluate_alerts(&[rule.clone()], &mut states, 1, Timestamp::from_millis(1), "agent=curl"); assert!(fired.is_empty()); let fired = evaluate_alerts(&[rule], &mut states, 2, Timestamp::from_millis(2), "agent=wget"); assert_eq!(fired.len(), 1); } #[test] fn ratio_fires_when_crossing_and_skips_zero_denominator() { let rule = AlertRule { id: "r1".to_string(), name: "error ratio".to_string(), enabled: true, condition: AlertCondition::Ratio { query_a: r#"regex("error")"#.to_string(), query_b: r#"regex("request")"#.to_string(), op: CompareOp::Ge, ratio: 0.5, window_ms: 10_000, }, }; let mut states = HashMap::new(); // Only an "error" so far; no "request" yet, so denominator is 0 -> skip. assert!(evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "error").is_empty()); // One request: ratio 1/1 = 1.0 >= 0.5 -> fires. let fired = evaluate_alerts(&[rule.clone()], &mut states, 1, Timestamp::from_millis(1), "request"); assert_eq!(fired.len(), 1); // A second request: ratio 1/2 = 0.5 still >= 0.5 -> fires again. let fired = evaluate_alerts(&[rule.clone()], &mut states, 2, Timestamp::from_millis(2), "request"); assert_eq!(fired.len(), 1); // A third request with no more errors: ratio drops to 1/3 -> no longer fires. let fired = evaluate_alerts(&[rule], &mut states, 3, Timestamp::from_millis(3), "request"); assert!(fired.is_empty()); } #[test] fn metric_comparison_in_alert_query_is_rejected_not_panicking() { let rule = immediate_rule("some_metric[param]==5"); let mut states = HashMap::new(); // Should log a warning and simply not fire, never panic. let fired = evaluate_alerts(&[rule], &mut states, 0, Timestamp::from_millis(0), "anything"); assert!(fired.is_empty()); } #[tokio::test] async fn deliver_ntfy_posts_message_to_server_topic_path() { use axum::extract::State; use axum::routing::post; use axum::Router; use std::sync::{Arc, Mutex}; use tokio::net::TcpListener; #[derive(Clone, Default)] struct Captured(Arc>>); async fn capture( State(captured): State, axum::extract::Path(topic): axum::extract::Path, body: String, ) -> &'static str { *captured.0.lock().unwrap() = Some((topic, body)); "ok" } let captured = Captured::default(); let app = Router::new().route("/{topic}", post(capture)).with_state(captured.clone()); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); let server = format!("http://{addr}"); deliver_ntfy(&server, "alerts", "hello world").await.unwrap(); let (topic, body) = captured.0.lock().unwrap().clone().unwrap(); assert_eq!(topic, "alerts"); assert_eq!(body, "hello world"); } }