AkurAI-Monitor / src/alert.rs

Latest change 8bade644cef8ddaa358cc88eb6017c8e64d5a6e4 - AKURAI-MONITOR-7 monitor processes, docker, systemd; fix #errors alert delivery by Ólafur Búi Ólafsson

use crate::db;
use chrono::Utc;
use mysql::{params, prelude::Queryable};
use serde::{Deserialize, 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,
}

#[derive(Serialize)]
pub struct ReceiverTarget {
    pub value: String,
    pub label: String,
}

#[derive(Deserialize)]
struct PiBunChannels {
    channels: Vec<PiBunChannel>,
}

#[derive(Deserialize)]
struct PiBunChannel {
    name: String,
}

pub async fn evaluate_alerts(alert_log_path: &str) {
    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 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);
        }
    }

    for notification in notifications {
        if let Err(message) = deliver(&notification, None).await {
            tracing::error!(
                error = message,
                rule = notification.rule,
                "alert delivery failed"
            );
        }
    }
}

/// 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(&notification, Some(channel_id)).await
}

pub async fn channel_targets(channel_id: i64) -> Result<Vec<ReceiverTarget>, String> {
    let (kind, endpoint, token_env) = db::with_db(|conn| {
        conn.exec_first::<(String, String, Option<String>), _, _>(
            "SELECT kind, endpoint, token_env FROM channels WHERE id = ?",
            (channel_id,),
        )
    })
    .ok()
    .flatten()
    .ok_or_else(|| "channel was not found".to_string())?;
    if kind != "pi-bun" {
        return Err("channel does not provide selectable receivers".to_string());
    }

    let destination = Destination {
        channel_id,
        channel_name: String::new(),
        kind,
        endpoint,
        token_env,
        recipient: String::new(),
        target: String::new(),
    };
    let client = reqwest::Client::builder()
        .connect_timeout(std::time::Duration::from_secs(5))
        .timeout(std::time::Duration::from_secs(15))
        .build()
        .map_err(|_| "failed to create receiver discovery client".to_string())?;
    let cookie = pi_bun_session(&client, &destination).await?;
    let response = client
        .get(format!(
            "{}/api/channels",
            pi_bun_base(&destination.endpoint)
        ))
        .header(reqwest::header::COOKIE, cookie)
        .send()
        .await
        .map_err(|error| error.to_string())?;
    if !response.status().is_success() {
        return Err(format!(
            "pi-bun channel discovery returned HTTP {}",
            response.status()
        ));
    }
    let channels = response
        .json::<PiBunChannels>()
        .await
        .map_err(|_| "pi-bun returned invalid channel data".to_string())?;
    Ok(channels
        .channels
        .into_iter()
        .map(|channel| ReceiverTarget {
            label: format!("#{}", channel.name),
            value: channel.name,
        })
        .collect())
}

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 client = reqwest::Client::builder()
        .connect_timeout(std::time::Duration::from_secs(5))
        .timeout(std::time::Duration::from_secs(15))
        .build()
        .map_err(|_| "failed to create alert HTTP client".to_string())?;
    let mut sent = 0;
    let mut failures = Vec::new();
    for destination in destinations {
        match deliver_one(&client, 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(
    client: &reqwest::Client,
    notification: &AlertNotification,
    destination: &Destination,
) -> Result<(), String> {
    match destination.kind.as_str() {
        "webhook" => {
            let webhook_client = webhook_client(&destination.endpoint).await?;
            let mut request = webhook_client.post(&destination.endpoint).json(&json!({
                "recipient": {
                    "name": destination.recipient,
                    "target": destination.target,
                },
                "alert": notification,
            }));
            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()))
            }
        }
        "pi-bun" => deliver_pi_bun(client, notification, destination).await,
        _ => Err("unsupported channel kind".to_string()),
    }
}

async fn deliver_pi_bun(
    client: &reqwest::Client,
    notification: &AlertNotification,
    destination: &Destination,
) -> Result<(), String> {
    let endpoint = reqwest::Url::parse(&destination.endpoint)
        .map_err(|_| "pi-bun endpoint is invalid".to_string())?;
    if !matches!(endpoint.scheme(), "http" | "https") || endpoint.host_str() != Some("100.88.0.9") {
        return Err("pi-bun endpoint is outside the Titan mesh".to_string());
    }
    let token = channel_token(destination)?
        .ok_or_else(|| "pi-bun channel requires token_env".to_string())?;
    let message = format!(
        "**{} alert · {}**\n\n{} = {:.2}; threshold {:.2}.{}{}",
        notification.event.to_uppercase(),
        notification.rule,
        notification.metric,
        notification.value,
        notification.threshold,
        notification
            .host_id
            .map(|id| format!(" Host {id}."))
            .unwrap_or_default(),
        notification
            .application_id
            .map(|id| format!(" Application {id}."))
            .unwrap_or_default()
    );
    let response = client
        .post(format!(
            "{}/api/ingest-error",
            pi_bun_base(&destination.endpoint)
        ))
        .bearer_auth(token)
        .json(&json!({
            "channel": destination.target,
            "body": message,
            "level": if notification.event == "triggered" { "error" } else { "info" },
            "source": "AkurAI Monitor",
        }))
        .send()
        .await
        .map_err(|error| error.to_string())?;
    if response.status().is_success() {
        Ok(())
    } else {
        Err(format!("pi-bun ingest returned HTTP {}", response.status()))
    }
}

async fn pi_bun_session(
    client: &reqwest::Client,
    destination: &Destination,
) -> Result<String, String> {
    let token = channel_token(destination)?
        .ok_or_else(|| "pi-bun channel requires token_env".to_string())?;
    let login = client
        .post(format!("{}/api/login", pi_bun_base(&destination.endpoint)))
        .json(&json!({"token": token}))
        .send()
        .await
        .map_err(|error| error.to_string())?;
    if !login.status().is_success() {
        return Err(format!("pi-bun login returned HTTP {}", login.status()));
    }
    login
        .headers()
        .get(reqwest::header::SET_COOKIE)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.split(';').next())
        .map(str::to_string)
        .ok_or_else(|| "pi-bun login returned no session cookie".to_string())
}

fn pi_bun_base(endpoint: &str) -> &str {
    endpoint
        .trim_end_matches('/')
        .strip_suffix("/api/ingest-error")
        .unwrap_or_else(|| endpoint.trim_end_matches('/'))
}

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 std::net::{IpAddr, Ipv4Addr, Ipv6Addr};

    #[test]
    fn pi_bun_endpoint_accepts_base_or_ingest_url() {
        assert_eq!(
            pi_bun_base("http://100.88.0.9:4173/api/ingest-error"),
            "http://100.88.0.9:4173"
        );
        assert_eq!(
            pi_bun_base("http://100.88.0.9:4173/"),
            "http://100.88.0.9:4173"
        );
    }

    #[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)));
        }
    }
}