AkurAI-Monitor / src/journald.rs

Latest change d0b8d14afb524ab72435e5abd0bbd6049a205349 - Migrate storage layer from SQLite to MariaDB by Ólafur Búi Ólafsson

use std::time::Duration;

use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;

use crate::{db, dns_metrics, health, stream};

/// A parsed journald entry:
/// (source_unit, message_line, timestamp, cursor, level, pid).
type JournalEntry = (
    String,
    String,
    i64,
    Option<String>,
    Option<String>,
    Option<i64>,
);

/// Follow systemd-journald for the whole journal and ingest app log lines.
/// Rather than a hand-maintained unit list (brittle, needs env changes to grow),
/// we tail the entire journal and store entries whose `_SYSTEMD_UNIT` matches an
/// app allowlist (`MONITOR_JOURNAL_MATCH` prefixes, default `akurai-`,
/// `golfsetrid`, `rust-monitor`) or any explicitly-configured unit. New apps are
/// captured automatically with no configuration change. Each unit's stdout/stderr
/// is captured by journald, so this needs no changes in the apps themselves;
/// severity comes from `PRIORITY`, pid from `_PID`, process from the unit name.
pub fn spawn(units: Vec<String>) {
    let prefixes = crate::config::get().journal_match.clone();

    // Register explicitly-configured units for health tracking up front; units
    // matched only by prefix register lazily on first sighting.
    for u in &units {
        health::register_source(u, "journald");
    }

    tokio::spawn(async move {
        // Resume from the last committed cursor. A missing or invalid cursor
        // falls back to a bounded replay window and cursor-based deduplication.
        let mut cursor = health::load_any_cursor();
        tracing::info!(
            explicit_units = units.join(","),
            match_prefixes = prefixes.join(","),
            has_cursor = cursor.is_some(),
            "spawning journald follower (whole-journal, allowlist-filtered)"
        );

        loop {
            if let Err(e) = follow(&units, &prefixes, &mut cursor).await {
                let now = chrono::Utc::now().timestamp();
                let err_msg = e.to_string();
                tracing::warn!(error = %e, "journald follower stopped; restarting in 5s");
                if cursor.take().is_some() {
                    health::clear_journal_cursors();
                }
                for u in &units {
                    health::record_error(u, now, &err_msg);
                }
            }
            tokio::time::sleep(Duration::from_secs(5)).await;
        }
    });
}

/// True if a journal entry's unit should be stored: either an explicitly
/// configured unit, or one matching an app allowlist prefix.
fn should_store(unit: &str, explicit: &[String], prefixes: &[String]) -> bool {
    explicit.iter().any(|u| u == unit) || prefixes.iter().any(|p| unit.starts_with(p.as_str()))
}

async fn follow(
    units: &[String],
    prefixes: &[String],
    cursor: &mut Option<String>,
) -> std::io::Result<()> {
    let mut cmd = Command::new("journalctl");
    cmd.args(["-f", "-o", "json", "--no-pager"]);

    if let Some(c) = cursor {
        // Resume from the last persisted cursor to avoid gaps on restart.
        cmd.args(["--after-cursor", c]);
    } else {
        let since =
            chrono::Utc::now().timestamp() - crate::config::get().journal_replay_secs.max(0);
        cmd.arg("--since").arg(format!("@{since}"));
    }

    // No `-u`: follow the whole journal so newly-deployed app units are captured
    // automatically; entries are filtered to the app allowlist below.
    cmd.stdout(std::process::Stdio::piped());

    let mut child = cmd.spawn()?;
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| std::io::Error::other("journalctl produced no stdout"))?;

    for unit in units {
        health::clear_error(unit);
    }

    tracing::info!(
        units = units.join(","),
        cursor = cursor.as_deref(),
        "following journald"
    );

    let mut lines = BufReader::new(stdout).lines();
    let mut buf: Vec<JournalEntry> = Vec::new();
    let mut flush = tokio::time::interval(Duration::from_secs(1));

    loop {
        tokio::select! {
            line = lines.next_line() => {
                match line {
                    Ok(Some(l)) => {
                        if let Some(entry) = parse_line(&l)
                            && should_store(&entry.0, units, prefixes)
                        {
                                if entry.0.contains("akurai-dns") {
                                    dns_metrics::observe(&entry.1);
                                }
                                // Register newly-seen units for health tracking.
                                health::register_source(&entry.0, "journald");
                                buf.push(entry);
                            }
                    }
                    Ok(None) => break,        // journalctl exited
                    Err(e) => return Err(e),
                }
            }
            _ = flush.tick() => {
                if !buf.is_empty() {
                    persist_and_broadcast(std::mem::take(&mut buf), cursor);
                }
            }
        }
    }

    if !buf.is_empty() {
        persist_and_broadcast(buf, cursor);
    }
    let status = child.wait().await?;
    if !status.success() {
        return Err(std::io::Error::other(format!(
            "journalctl exited with {status}"
        )));
    }
    Ok(())
}

/// Extract (source, line, ts, cursor) from one journald JSON record.
fn parse_line(raw: &str) -> Option<JournalEntry> {
    let v: serde_json::Value = serde_json::from_str(raw).ok()?;

    // MESSAGE is a string when clean UTF-8, but journald encodes it as an array
    // of byte values when it contains control chars (e.g. ANSI colour codes from
    // a `tracing` subscriber). Handle both, then strip ANSI escapes.
    let message = match v.get("MESSAGE")? {
        serde_json::Value::String(s) => s.clone(),
        serde_json::Value::Array(arr) => {
            let bytes: Vec<u8> = arr
                .iter()
                .filter_map(|n| n.as_u64().map(|b| b as u8))
                .collect();
            String::from_utf8_lossy(&bytes).into_owned()
        }
        _ => return None,
    };
    let message = strip_ansi(&message);
    if message.is_empty() {
        return None;
    }

    let source = v
        .get("_SYSTEMD_UNIT")
        .and_then(|s| s.as_str())
        .or_else(|| v.get("SYSLOG_IDENTIFIER").and_then(|s| s.as_str()))
        .unwrap_or("journal")
        .to_string();

    // __REALTIME_TIMESTAMP is microseconds since the epoch, as a string.
    let ts = v
        .get("__REALTIME_TIMESTAMP")
        .and_then(|s| s.as_str())
        .and_then(|s| s.parse::<i64>().ok())
        .map(|us| us / 1_000_000)
        .unwrap_or_else(|| chrono::Utc::now().timestamp());

    // __CURSOR is the journald cursor — unique per entry, used for dedup and replay.
    let cursor = v
        .get("__CURSOR")
        .and_then(|s| s.as_str())
        .map(|s| s.to_string());

    // PRIORITY is the syslog severity (0-7) as a string; map to a log level.
    let level = v
        .get("PRIORITY")
        .and_then(|s| s.as_str())
        .and_then(|s| s.parse::<u8>().ok())
        .map(|p| match p {
            0..=3 => "error",
            4 => "warn",
            5 | 6 => "info",
            _ => "debug",
        })
        .map(|s| s.to_string());

    let pid = v
        .get("_PID")
        .and_then(|s| s.as_str())
        .and_then(|s| s.parse::<i64>().ok());

    Some((source, message, ts, cursor, level, pid))
}

/// Remove ANSI CSI escape sequences (ESC `[` … final-byte) from a log line.
fn strip_ansi(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    let mut chars = s.chars().peekable();
    while let Some(c) = chars.next() {
        if c == '\u{1b}' {
            if chars.peek() == Some(&'[') {
                chars.next();
                for n in chars.by_ref() {
                    if n.is_ascii_alphabetic() {
                        break;
                    }
                }
            }
        } else {
            out.push(c);
        }
    }
    out.trim().to_string()
}

fn persist_and_broadcast(entries: Vec<JournalEntry>, cursor: &mut Option<String>) {
    let now = chrono::Utc::now().timestamp();
    let (stored, committed_cursor) = db::with_db(|conn| {
        use mysql::prelude::Queryable;
        let Ok(mut tx) = conn.start_transaction(mysql::TxOpts::default()) else {
            tracing::error!("failed to start journald persistence transaction");
            return (Vec::new(), None);
        };
        let mut stored = Vec::new();
        let mut committed_cursor = None;

        for (source, line, ts, entry_cursor, level, pid) in entries {
            if tx
                .exec_drop(
                    "INSERT IGNORE INTO logs
                       (host_id, source, line, level, pid, origin, ts, `cursor`)
                     VALUES ((SELECT id FROM hosts WHERE name = 'local'), ?, ?, ?, ?, 'server', ?, ?)",
                    (&source, &line, &level, pid, ts, &entry_cursor),
                )
                .is_err()
            {
                tracing::error!(source, "failed to persist journald entry");
                return (Vec::new(), None);
            }
            let inserted = tx.affected_rows();

            if let Some(entry_cursor) = entry_cursor.as_deref() {
                if tx
                    .exec_drop(
                        "INSERT INTO ingestion_state (source, kind, last_cursor, last_ts)
                         VALUES (?, 'journald', ?, ?)
                         ON DUPLICATE KEY UPDATE
                           kind = VALUES(kind),
                           last_cursor = VALUES(last_cursor),
                           last_ts = VALUES(last_ts)",
                        (&source, entry_cursor, ts),
                    )
                    .is_err()
                {
                    tracing::error!(source, "failed to persist journald cursor");
                    return (Vec::new(), None);
                }
                committed_cursor = Some(entry_cursor.to_string());
            }

            if inserted == 1 {
                crate::pibun_push::push_error(&source, &line, level.as_deref());
                stored.push((source, line, ts, level, pid));
            }
        }

        if tx.commit().is_err() {
            tracing::error!("failed to commit journald persistence transaction");
            return (Vec::new(), None);
        }
        (stored, committed_cursor)
    });

    if let Some(committed_cursor) = committed_cursor {
        *cursor = Some(committed_cursor);
    }
    if stored.is_empty() {
        return;
    }

    let mut seen = std::collections::HashSet::new();
    for (source, _, _, _, _) in &stored {
        if seen.insert(source.as_str()) {
            health::record_activity(source, now);
        }
    }

    let logs: Vec<_> = stored
        .iter()
        .map(|(source, line, ts, level, pid)| {
            serde_json::json!({
                "source": source, "line": line, "ts": ts,
                "level": level, "pid": pid, "origin": "server"
            })
        })
        .collect();
    stream::publish("log", serde_json::json!({"logs": logs}).to_string());
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_cursor_timestamp_and_strips_ansi() {
        let entry = parse_line(
            r#"{"MESSAGE":"\u001b[31mfailed\u001b[0m","_SYSTEMD_UNIT":"demo.service","__REALTIME_TIMESTAMP":"1700000000123456","__CURSOR":"s=cursor","PRIORITY":"3","_PID":"4242"}"#,
        )
        .expect("parse journald record");

        assert_eq!(entry.0, "demo.service");
        assert_eq!(entry.1, "failed");
        assert_eq!(entry.2, 1_700_000_000);
        assert_eq!(entry.3.as_deref(), Some("s=cursor"));
        assert_eq!(entry.4.as_deref(), Some("error"));
        assert_eq!(entry.5, Some(4242));
    }

    #[test]
    fn should_store_matches_allowlist_and_explicit() {
        let prefixes = vec!["akurai-".to_string(), "rust-monitor".to_string()];
        let explicit = vec!["custom.service".to_string()];
        assert!(should_store("akurai-notes.service", &explicit, &prefixes));
        assert!(should_store("rust-monitor.service", &explicit, &prefixes));
        assert!(should_store("custom.service", &explicit, &prefixes));
        assert!(!should_store("nginx.service", &explicit, &prefixes));
        assert!(!should_store("ssh.service", &explicit, &prefixes));
    }
}