Menu
Public repository / adf2041cfda7c69377545fe6f587d1fe427fb716
AkurAI-Monitor / src/journald.rs
Latest change adf2041cfda7c69377545fe6f587d1fe427fb716 - Harden log ingestion monitoring 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).
type JournalEntry = (String, String, i64, Option<String>);
/// Follow systemd-journald for the given units and ingest their log lines.
/// Each unit's stdout/stderr (e.g. a Node service's console output, a Rust
/// service's tracing output) is captured by journald, so this needs no changes
/// in the apps themselves. Source is tagged with the systemd unit name.
pub fn spawn(units: Vec<String>) {
if units.is_empty() {
return;
}
// Register each unit for health tracking.
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!(
units = units.join(","),
has_cursor = cursor.is_some(),
"spawning journald follower"
);
loop {
if let Err(e) = follow(&units, &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;
}
});
}
async fn follow(units: &[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}"));
}
for u in units {
cmd.arg("-u").arg(u);
}
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) {
// Derive DNS metrics from akurai-dns query lines.
if entry.0.contains("akurai-dns") {
dns_metrics::observe(&entry.1);
}
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());
Some((source, message, ts, cursor))
}
/// 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| {
let Ok(tx) = conn.unchecked_transaction() 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) in entries {
let inserted = tx.execute(
"INSERT OR IGNORE INTO logs (host_id, source, line, ts, cursor)
VALUES ((SELECT id FROM hosts WHERE name = 'local'), ?1, ?2, ?3, ?4)",
rusqlite::params![source, line, ts, entry_cursor],
);
let Ok(inserted) = inserted else {
tracing::error!(source, "failed to persist journald entry");
return (Vec::new(), None);
};
if let Some(entry_cursor) = entry_cursor.as_deref() {
if tx
.execute(
"INSERT INTO ingestion_state (source, kind, last_cursor, last_ts)
VALUES (?1, 'journald', ?2, ?3)
ON CONFLICT(source) DO UPDATE SET
kind = excluded.kind,
last_cursor = excluded.last_cursor,
last_ts = excluded.last_ts",
rusqlite::params![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));
}
}
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)| serde_json::json!({"source": source, "line": line, "ts": ts}))
.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"}"#,
)
.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"));
}
}