Menu
Public repository / 6bb03beff5ca893c4caf66f9aa5fa00769e53cd3
AkurAI-Monitor / src/journald.rs
Latest change 6bb03beff5ca893c4caf66f9aa5fa00769e53cd3 - Remove pi-bun by Ólafur Búi Ólafsson
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use crate::{db, dns_metrics, health, schema, 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 host_id = schema::monitor_host_id(conn);
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 (?, ?, ?, ?, ?, 'server', ?, ?)",
(host_id, &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 {
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));
}
}