Menu
Public repository / 672363de5c910b74bd75d069248a20f2c78036ac
AkurAI-Monitor / src/alert.rs
Latest change 6bb03beff5ca893c4caf66f9aa5fa00769e53cd3 - Remove pi-bun by Ólafur Búi Ólafsson
use crate::db;
use chrono::Utc;
use mysql::{params, prelude::Queryable};
use serde::Serialize;
use serde_json::json;
use std::io::Write;
struct AlertRule {
id: i64,
host_id: Option<i64>,
application_id: Option<i64>,
name: String,
metric_name: String,
operator: String,
threshold: f64,
duration_secs: i64,
}
#[derive(Clone, Serialize)]
pub struct AlertNotification {
event: &'static str,
rule_id: i64,
rule: String,
metric: String,
value: f64,
threshold: f64,
host_id: Option<i64>,
application_id: Option<i64>,
timestamp: i64,
}
struct Destination {
channel_id: i64,
channel_name: String,
kind: String,
endpoint: String,
token_env: Option<String>,
recipient: String,
target: String,
}
pub async fn evaluate_alerts(alert_log_path: &str, stale_secs: i64) {
let rules = db::with_db(|conn| {
conn.query_map(
"SELECT id, host_id, application_id, name, metric_name, operator, threshold, duration_secs
FROM alert_rules WHERE enabled = 1",
|(id, host_id, application_id, name, metric_name, operator, threshold, duration_secs)| {
AlertRule {
id,
host_id,
application_id,
name,
metric_name,
operator,
threshold,
duration_secs,
}
},
)
.unwrap_or_default()
});
let now = Utc::now().timestamp();
let mut notifications = Vec::new();
for rule in &rules {
let cutoff = now - rule.duration_secs;
if rule.metric_name.contains('%') {
// Pattern rule (SQL LIKE): expand to every distinct concrete
// metric name currently reporting samples in-window — one per
// running process/service/container — and evaluate + track each
// as an independent alert subject. Capped so a runaway process
// count can't turn one rule into an unbounded query fan-out.
let subjects: Vec<String> = db::with_db(|conn| {
conn.exec(
"SELECT DISTINCT name FROM metrics
WHERE name LIKE :pattern AND ts >= :cutoff
AND (:host_id IS NULL OR host_id = :host_id)
AND (:application_id IS NULL OR application_id = :application_id)
LIMIT 200",
params! {
"pattern" => &rule.metric_name,
"cutoff" => cutoff,
"host_id" => rule.host_id,
"application_id" => rule.application_id,
},
)
.unwrap_or_default()
});
for subject in &subjects {
if has_specific_override(&rules, rule, subject) {
close_superseded_events(rule.id, subject, now);
continue;
}
if let Some(notification) =
evaluate_subject(rule, Some(subject.as_str()), cutoff, now, alert_log_path)
{
notifications.push(notification);
}
}
} else if let Some(notification) = evaluate_subject(rule, None, cutoff, now, alert_log_path)
{
notifications.push(notification);
}
}
let stale = db::with_db(|conn| resolve_stale_events(conn, now, stale_secs));
for notification in &stale {
write_alert_log(
alert_log_path,
&format!(
"RESOLVED rule=\"{}\" metric=\"{}\" reason=stale",
notification.rule, notification.metric
),
);
tracing::info!(
rule = notification.rule,
metric = notification.metric,
stale_secs,
"alert resolved: metric went silent"
);
}
notifications.extend(stale);
for notification in notifications {
if let Err(message) = deliver(¬ification, None).await {
tracing::error!(
error = message,
rule = notification.rule,
"alert delivery failed"
);
}
}
}
/// A pattern rule yields to an enabled exact-metric rule for the same host
/// scope and rule name: that per-subject rule carries its own ceiling (e.g. a
/// known-heavy process), so the generic default must not also fire for it.
fn has_specific_override(rules: &[AlertRule], pattern_rule: &AlertRule, subject: &str) -> bool {
rules.iter().any(|other| {
other.id != pattern_rule.id
&& other.metric_name == subject
&& other.name == pattern_rule.name
&& other.host_id == pattern_rule.host_id
&& other.application_id == pattern_rule.application_id
})
}
/// Events opened by the pattern rule before an exact rule took over the
/// subject would otherwise never be evaluated again and stay open forever.
fn close_superseded_events(pattern_rule_id: i64, subject: &str, now: i64) {
db::with_db(|conn| {
conn.exec_drop(
"UPDATE alert_events SET resolved_at = ? WHERE rule_id = ? AND subject <=> ? AND resolved_at IS NULL",
(now, pattern_rule_id, subject),
)
.ok();
});
}
/// An open alert is stale once its metric has had no sample for the stale
/// window. The window never drops below the rule's own evaluation window:
/// resolving earlier would let the same in-window samples re-trigger it.
fn is_stale(last_sample_ts: Option<i64>, now: i64, stale_secs: i64, duration_secs: i64) -> bool {
let window = stale_secs.max(duration_secs);
last_sample_ts.is_none_or(|ts| now - ts > window)
}
/// Resolve every open alert whose metric stopped reporting. Rule evaluation
/// only revisits subjects that still have in-window samples, so without this
/// an alert on a process/metric that vanished would stay open forever.
fn resolve_stale_events(
conn: &mut impl Queryable,
now: i64,
stale_secs: i64,
) -> Vec<AlertNotification> {
let open: Vec<(i64, i64, Option<i64>, AlertNotification)> = conn
.query_map(
"SELECT e.id, r.duration_secs,
(SELECT MAX(m.ts) FROM metrics m
WHERE m.name = COALESCE(e.subject, r.metric_name)
AND (r.host_id IS NULL OR m.host_id = r.host_id)
AND (r.application_id IS NULL OR m.application_id = r.application_id)),
e.rule_id, r.name, COALESCE(e.subject, r.metric_name), e.value, r.threshold,
r.host_id, r.application_id
FROM alert_events e JOIN alert_rules r ON r.id = e.rule_id
WHERE e.resolved_at IS NULL
ORDER BY e.id",
|(
id,
duration_secs,
last_ts,
rule_id,
rule,
metric,
value,
threshold,
host_id,
application_id,
)| {
let notification = AlertNotification {
event: "resolved",
rule_id,
rule,
metric,
value,
threshold,
host_id,
application_id,
timestamp: now,
};
(id, duration_secs, last_ts, notification)
},
)
.unwrap_or_default();
open.into_iter()
.filter(|(_, duration_secs, last_ts, _)| {
is_stale(*last_ts, now, stale_secs, *duration_secs)
})
.filter(|(id, ..)| {
conn.exec_drop(
"UPDATE alert_events SET resolved_at = ? WHERE id = ? AND resolved_at IS NULL",
(now, id),
)
.is_ok()
})
.map(|(.., notification)| notification)
.collect()
}
/// Evaluate one concrete metric stream against a rule and update
/// `alert_events` accordingly. `subject` distinguishes multiple concurrent
/// alert streams expanded from the same pattern rule (e.g. one per process/
/// service/container many-per-rule fan-out); it is `None` for exact
/// (non-pattern) rules, which queries `rule.metric_name` directly and
/// preserves the original single-stream-per-rule behavior — matching
/// existing rows whose `subject` column is NULL. `subject <=> ?` (NULL-safe
/// equality) makes `None` match exactly those legacy/exact-rule rows and
/// nothing else.
fn evaluate_subject(
rule: &AlertRule,
subject: Option<&str>,
cutoff: i64,
now: i64,
alert_log_path: &str,
) -> Option<AlertNotification> {
let query_name = subject.unwrap_or(rule.metric_name.as_str());
let samples: Vec<f64> = db::with_db(|conn| {
conn.exec(
"SELECT value FROM metrics
WHERE name = :name AND ts >= :cutoff
AND (:host_id IS NULL OR host_id = :host_id)
AND (:application_id IS NULL OR application_id = :application_id)
ORDER BY ts",
params! {
"name" => query_name,
"cutoff" => cutoff,
"host_id" => rule.host_id,
"application_id" => rule.application_id,
},
)
.unwrap_or_default()
});
if samples.is_empty() {
return None;
}
let all_violating = samples
.iter()
.all(|value| violates(*value, &rule.operator, rule.threshold));
let latest = *samples.last().unwrap();
let has_open_alert = db::with_db(|conn| {
conn.exec_first::<i64, _, _>(
"SELECT COUNT(*) FROM alert_events WHERE rule_id = ? AND subject <=> ? AND resolved_at IS NULL",
(rule.id, subject),
)
.ok()
.flatten()
.unwrap_or(0)
> 0
});
let event = if all_violating && !has_open_alert {
db::with_db(|conn| {
conn.exec_drop(
"INSERT INTO alert_events (rule_id, triggered_at, value, subject) VALUES (?, ?, ?, ?)",
(rule.id, now, latest, subject),
)
.ok();
});
write_alert_log(
alert_log_path,
&format!(
"ALERT rule=\"{}\" metric=\"{}\" value={:.1} threshold={:.1}",
rule.name, query_name, latest, rule.threshold
),
);
tracing::warn!(
rule = rule.name,
metric = query_name,
value = latest,
threshold = rule.threshold,
"alert triggered"
);
"triggered"
} else if !all_violating && has_open_alert {
db::with_db(|conn| {
conn.exec_drop(
"UPDATE alert_events SET resolved_at = ? WHERE rule_id = ? AND subject <=> ? AND resolved_at IS NULL",
(now, rule.id, subject),
)
.ok();
});
write_alert_log(
alert_log_path,
&format!(
"RESOLVED rule=\"{}\" metric=\"{}\" value={:.1}",
rule.name, query_name, latest
),
);
tracing::info!(rule = rule.name, metric = query_name, "alert resolved");
"resolved"
} else {
return None;
};
Some(AlertNotification {
event,
rule_id: rule.id,
rule: rule.name.clone(),
metric: query_name.to_string(),
value: latest,
threshold: rule.threshold,
host_id: rule.host_id,
application_id: rule.application_id,
timestamp: now,
})
}
pub async fn test_channel(channel_id: i64) -> Result<usize, String> {
let notification = AlertNotification {
event: "test",
rule_id: 0,
rule: "Channel test".to_string(),
metric: "monitor.channel".to_string(),
value: 1.0,
threshold: 1.0,
host_id: None,
application_id: None,
timestamp: Utc::now().timestamp(),
};
deliver(¬ification, Some(channel_id)).await
}
async fn deliver(
notification: &AlertNotification,
channel_id: Option<i64>,
) -> Result<usize, String> {
let destinations = db::with_db(|conn| {
conn.exec_map(
"SELECT c.id, c.name, c.kind, c.endpoint, c.token_env, r.name, r.target
FROM recipients r
JOIN channels c ON c.id = r.channel_id
WHERE c.enabled = 1 AND r.enabled = 1 AND (:channel_id IS NULL OR c.id = :channel_id)
ORDER BY c.id, r.id",
params! { "channel_id" => channel_id },
|(channel_id, channel_name, kind, endpoint, token_env, recipient, target)| {
Destination {
channel_id,
channel_name,
kind,
endpoint,
token_env,
recipient,
target,
}
},
)
.map_err(|_| "failed to query recipients".to_string())
})?;
if channel_id.is_some() && destinations.is_empty() {
return Err("channel has no enabled recipients".to_string());
}
let mut sent = 0;
let mut failures = Vec::new();
for destination in destinations {
match deliver_one(notification, &destination).await {
Ok(()) => sent += 1,
Err(message) => failures.push(format!(
"channel {} ({}) recipient {}: {message}",
destination.channel_id, destination.channel_name, destination.recipient
)),
}
}
if failures.is_empty() {
Ok(sent)
} else {
Err(failures.join("; "))
}
}
async fn deliver_one(
notification: &AlertNotification,
destination: &Destination,
) -> Result<(), String> {
if destination.kind != "webhook" {
return Err(format!("unsupported channel kind {}", destination.kind));
}
let webhook_client = webhook_client(&destination.endpoint).await?;
let mut request = webhook_client
.post(&destination.endpoint)
.json(&webhook_payload(notification, destination));
if let Some(token) = channel_token(destination)? {
request = request.bearer_auth(token);
}
let response = request.send().await.map_err(|error| error.to_string())?;
if response.status().is_success() {
Ok(())
} else {
Err(format!("webhook returned HTTP {}", response.status()))
}
}
fn webhook_payload(
notification: &AlertNotification,
destination: &Destination,
) -> serde_json::Value {
json!({
"recipient": {
"name": destination.recipient,
"target": destination.target,
},
"alert": notification,
})
}
fn channel_token(destination: &Destination) -> Result<Option<String>, String> {
destination
.token_env
.as_deref()
.map(|name| std::env::var(name).map_err(|_| "channel token is unavailable".to_string()))
.transpose()
}
async fn webhook_client(endpoint: &str) -> Result<reqwest::Client, String> {
let url =
reqwest::Url::parse(endpoint).map_err(|_| "webhook endpoint is invalid".to_string())?;
let host = url
.host_str()
.ok_or_else(|| "webhook endpoint has no host".to_string())?;
let port = url
.port_or_known_default()
.ok_or_else(|| "webhook endpoint has no port".to_string())?;
let addresses = tokio::net::lookup_host((host, port))
.await
.map_err(|_| "webhook endpoint could not be resolved".to_string())?
.collect::<Vec<_>>();
if addresses.is_empty() || addresses.iter().any(|address| !public_ip(address.ip())) {
return Err("webhook endpoint resolves to a non-public address".to_string());
}
reqwest::Client::builder()
.connect_timeout(std::time::Duration::from_secs(5))
.timeout(std::time::Duration::from_secs(15))
.resolve_to_addrs(host, &addresses)
.build()
.map_err(|_| "failed to create webhook HTTP client".to_string())
}
fn public_ip(address: std::net::IpAddr) -> bool {
match address {
std::net::IpAddr::V4(address) => {
let octets = address.octets();
!(address.is_private()
|| address.is_loopback()
|| address.is_link_local()
|| address.is_unspecified()
|| address.is_broadcast()
|| address.is_multicast()
|| address.is_documentation()
|| octets[0] == 0
|| octets[0] >= 240
|| (octets[0] == 100 && (64..=127).contains(&octets[1]))
|| (octets[0] == 198 && matches!(octets[1], 18 | 19)))
}
std::net::IpAddr::V6(address) => {
if let Some(address) = address.to_ipv4_mapped() {
return public_ip(std::net::IpAddr::V4(address));
}
!(address.is_loopback()
|| address.is_unspecified()
|| address.is_multicast()
|| address.is_unique_local()
|| address.is_unicast_link_local())
}
}
}
fn violates(value: f64, operator: &str, threshold: f64) -> bool {
match operator {
"gt" => value > threshold,
"lt" => value < threshold,
"eq" => (value - threshold).abs() < f64::EPSILON,
_ => false,
}
}
fn write_alert_log(path: &str, message: &str) {
if let Some(parent) = std::path::Path::new(path).parent() {
std::fs::create_dir_all(parent).ok();
}
let timestamp = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
if let Ok(mut file) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(path)
{
let _ = writeln!(file, "{timestamp} {message}");
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::db::testing::TestDb;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
fn rule(id: i64, metric: &str, threshold: f64) -> AlertRule {
AlertRule {
id,
host_id: Some(8),
application_id: None,
name: "High Process Memory".into(),
metric_name: metric.into(),
operator: "gt".into(),
threshold,
duration_secs: 300,
}
}
#[test]
fn exact_rule_overrides_pattern_rule_for_its_subject_only() {
let generic = rule(1, "proc.%.mem_kb", 2.0);
let rules = vec![generic, rule(2, "proc.node.mem_kb", 4.0)];
assert!(has_specific_override(&rules, &rules[0], "proc.node.mem_kb"));
assert!(!has_specific_override(
&rules,
&rules[0],
"proc.bash.mem_kb"
));
}
#[test]
fn webhook_payload_carries_recipient_and_alert() {
let destination = Destination {
channel_id: 3,
channel_name: "ops".into(),
kind: "webhook".into(),
endpoint: "https://alerts.example.com/hook".into(),
token_env: Some("MONITOR_CHANNEL_TOKEN_OPS".into()),
recipient: "on-call".into(),
target: "pager".into(),
};
let notification = AlertNotification {
event: "triggered",
rule_id: 7,
rule: "High CPU".into(),
metric: "cpu.usage".into(),
value: 97.5,
threshold: 90.0,
host_id: Some(8),
application_id: None,
timestamp: 1_800_000_000,
};
let payload = webhook_payload(¬ification, &destination);
assert_eq!(
payload["recipient"],
json!({"name": "on-call", "target": "pager"})
);
assert_eq!(payload["alert"]["event"], "triggered");
assert_eq!(payload["alert"]["metric"], "cpu.usage");
assert_eq!(payload["alert"]["host_id"], 8);
assert!(
!payload.to_string().contains("MONITOR_CHANNEL_TOKEN"),
"token env names stay server-side"
);
}
#[test]
fn alert_goes_stale_once_its_metric_is_silent_past_the_window() {
assert!(is_stale(None, 1_000, 300, 60), "never sampled");
assert!(is_stale(Some(600), 1_000, 300, 60), "silent 400s > 300s");
assert!(!is_stale(Some(800), 1_000, 300, 60), "silent 200s < 300s");
}
#[test]
fn stale_window_is_never_shorter_than_the_rule_window() {
// Resolving inside the rule window would let the same in-window
// samples re-trigger it on the next tick (resolve/trigger flapping).
assert!(!is_stale(Some(600), 1_000, 300, 600));
assert!(is_stale(Some(300), 1_000, 300, 600));
}
#[test]
fn resolves_open_alerts_whose_metric_went_silent() {
let Some(db) = TestDb::new() else { return };
let mut conn = db.conn();
crate::schema::create_tables(&mut conn);
let host = crate::schema::monitor_host_id(&mut conn);
let now: i64 = 1_800_000_000;
conn.exec_drop(
"INSERT INTO hosts (name, address, kind, enabled, created_at, updated_at)
VALUES ('other', '100.88.0.7', 'linux', 1, ?, ?)",
(now, now),
)
.unwrap();
let other_host = conn.last_insert_id() as i64;
let add_rule = |conn: &mut mysql::PooledConn, name: &str, metric: &str| -> i64 {
conn.exec_drop(
"INSERT INTO alert_rules (host_id, name, metric_name, operator, threshold, duration_secs)
VALUES (?, ?, ?, 'gt', 1, 300)",
(host, name, metric),
)
.unwrap();
conn.last_insert_id() as i64
};
let pattern = add_rule(&mut conn, "High Process Memory", "proc.%.mem_kb");
let exact = add_rule(&mut conn, "High CPU", "cpu.usage");
for (host_id, name, ts) in [
(host, "proc.gone.mem_kb", now - 3_600),
// A fresh sample on another host must not keep a host-scoped alert alive.
(other_host, "proc.gone.mem_kb", now - 10),
(host, "proc.live.mem_kb", now - 30),
] {
conn.exec_drop(
"INSERT INTO metrics (host_id, name, value, ts) VALUES (?, ?, 5, ?)",
(host_id, name, ts),
)
.unwrap();
}
for (rule_id, subject) in [
(pattern, Some("proc.gone.mem_kb")),
(pattern, Some("proc.live.mem_kb")),
(exact, None),
] {
conn.exec_drop(
"INSERT INTO alert_events (rule_id, triggered_at, value, subject) VALUES (?, ?, 5, ?)",
(rule_id, now - 7_200, subject),
)
.unwrap();
}
let resolved = resolve_stale_events(&mut conn, now, 600);
let metrics: Vec<&str> = resolved.iter().map(|n| n.metric.as_str()).collect();
assert_eq!(metrics, ["proc.gone.mem_kb", "cpu.usage"]);
assert!(resolved.iter().all(|n| n.event == "resolved"));
let still_open: Vec<Option<String>> = conn
.query("SELECT subject FROM alert_events WHERE resolved_at IS NULL")
.unwrap();
assert_eq!(still_open, [Some("proc.live.mem_kb".to_string())]);
let resolved_at: Vec<i64> = conn
.query("SELECT resolved_at FROM alert_events WHERE resolved_at IS NOT NULL")
.unwrap();
assert_eq!(resolved_at, [now, now]);
assert!(resolve_stale_events(&mut conn, now, 600).is_empty());
}
#[test]
fn webhook_addresses_must_be_public() {
assert!(public_ip(IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8))));
assert!(public_ip(IpAddr::V6(
"2606:4700:4700::1111".parse::<Ipv6Addr>().unwrap()
)));
for address in [
Ipv4Addr::new(127, 0, 0, 1),
Ipv4Addr::new(10, 0, 0, 1),
Ipv4Addr::new(100, 88, 0, 9),
Ipv4Addr::new(169, 254, 169, 254),
Ipv4Addr::new(192, 0, 2, 1),
] {
assert!(!public_ip(IpAddr::V4(address)));
}
}
}