AkurAI-Monitor / src/tailer.rs

Latest change adf2041cfda7c69377545fe6f587d1fe427fb716 - Harden log ingestion monitoring by Ólafur Búi Ólafsson

use std::collections::HashMap;
use std::io::{Read, Seek, SeekFrom};

use crate::health;

/// Tracks per-file state: byte offset and inode for rotation detection.
struct FileState {
    position: u64,
    inode: u64,
}

pub struct LogTailer {
    files: HashMap<String, FileState>,
    /// Per-file consecutive error counts for observability.
    error_counts: HashMap<String, u64>,
}

impl LogTailer {
    pub fn new(paths: &[String]) -> Self {
        let mut files = HashMap::new();
        for path in paths {
            let (position, inode) = file_position_and_inode(path);
            files.insert(path.clone(), FileState { position, inode });
        }
        LogTailer {
            files,
            error_counts: HashMap::new(),
        }
    }

    /// Read new lines from each tracked file. Handles rotation (inode change or
    /// truncation), appends, and reports errors to the shared health tracker.
    pub fn read_new_lines(&mut self) -> Vec<(String, String)> {
        let mut entries = Vec::new();
        let now = chrono::Utc::now().timestamp();

        let paths: Vec<String> = self.files.keys().cloned().collect();
        for path in &paths {
            match self.read_file(path) {
                Ok(file_entries) => {
                    health::clear_error(path);
                    if !file_entries.is_empty() {
                        health::record_activity(path, now);
                        entries.extend(file_entries);
                    }
                    self.error_counts.remove(path);
                }
                Err(e) => {
                    let count = self
                        .error_counts
                        .entry(path.clone())
                        .and_modify(|c| *c += 1)
                        .or_insert(1);
                    let message = format!("{e} (consecutive error #{count})");
                    health::record_error(path, now, &message);
                    if *count == 1 || (*count).is_multiple_of(60) {
                        tracing::warn!(path, error = message, "log tail read failed");
                    }
                }
            }
        }

        entries
    }

    fn read_file(&mut self, path: &str) -> Result<Vec<(String, String)>, String> {
        let meta = std::fs::metadata(path).map_err(|e| format!("stat: {e}"))?;

        let current_inode = inode_from_metadata(&meta);
        let current_len = meta.len();

        let state = self
            .files
            .get_mut(path)
            .ok_or_else(|| "state missing".to_string())?;

        // Detect rotation: file was replaced (new inode) or truncated in place.
        let rotated = current_inode != state.inode || current_len < state.position;

        let seek_pos = if rotated {
            tracing::info!(
                path,
                old_inode = state.inode,
                new_inode = current_inode,
                old_len = state.position,
                new_len = current_len,
                "detected log rotation"
            );
            // Reset to start of the new file.
            state.inode = current_inode;
            0
        } else {
            state.position
        };

        let mut file = std::fs::File::open(path).map_err(|e| format!("open: {e}"))?;

        file.seek(SeekFrom::Start(seek_pos))
            .map_err(|e| format!("seek: {e}"))?;

        let mut buf = String::new();
        file.read_to_string(&mut buf)
            .map_err(|e| format!("read: {e}"))?;

        let complete_len = buf.rfind('\n').map_or(0, |index| index + 1);
        let mut file_entries = Vec::new();
        for line in buf[..complete_len].lines() {
            let line = line.trim();
            if !line.is_empty() {
                file_entries.push((path.to_string(), line.to_string()));
            }
        }

        // Leave an unterminated final line unread until the writer completes it.
        state.position = seek_pos + complete_len as u64;

        Ok(file_entries)
    }
}

/// Snapshot the file's current length and inode so we can detect rotation later.
fn file_position_and_inode(path: &str) -> (u64, u64) {
    match std::fs::metadata(path) {
        Ok(meta) => (meta.len(), inode_from_metadata(&meta)),
        Err(_) => (0, 0),
    }
}

#[cfg(unix)]
fn inode_from_metadata(meta: &std::fs::Metadata) -> u64 {
    use std::os::unix::fs::MetadataExt;
    meta.ino()
}

#[cfg(not(unix))]
fn inode_from_metadata(_meta: &std::fs::Metadata) -> u64 {
    0
}

#[cfg(test)]
mod tests {
    use std::io::Write;

    use super::*;

    fn test_path(name: &str) -> std::path::PathBuf {
        std::env::temp_dir().join(format!(
            "rust-monitor-{name}-{}-{}",
            std::process::id(),
            chrono::Utc::now().timestamp_nanos_opt().unwrap_or_default()
        ))
    }

    #[test]
    fn waits_for_complete_lines() {
        let path = test_path("partial");
        std::fs::write(&path, "existing\n").expect("create test log");
        let source = path.to_string_lossy().into_owned();
        let mut tailer = LogTailer::new(std::slice::from_ref(&source));

        let mut file = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .expect("open test log");
        write!(file, "partial").expect("write partial line");
        assert!(tailer.read_new_lines().is_empty());
        writeln!(file).expect("finish partial line");
        writeln!(file, "complete").expect("write complete line");

        let lines = tailer.read_new_lines();
        assert_eq!(lines.len(), 2);
        assert_eq!(lines[0].1, "partial");
        assert_eq!(lines[1].1, "complete");
        std::fs::remove_file(path).expect("remove test log");
    }

    #[test]
    fn reads_replacement_file_from_start() {
        let path = test_path("rotation");
        let rotated = path.with_extension("old");
        std::fs::write(&path, "existing\n").expect("create test log");
        let source = path.to_string_lossy().into_owned();
        let mut tailer = LogTailer::new(std::slice::from_ref(&source));

        std::fs::rename(&path, &rotated).expect("rotate test log");
        std::fs::write(&path, "replacement\n").expect("create replacement log");

        let lines = tailer.read_new_lines();
        assert_eq!(lines.len(), 1);
        assert_eq!(lines[0].1, "replacement");
        std::fs::remove_file(path).expect("remove replacement log");
        std::fs::remove_file(rotated).expect("remove rotated log");
    }
}