telelog-core 0.3.0

//! The log model shared by every Telelogs component.

use std::collections::BTreeMap;
use std::collections::HashMap;
use std::time::SystemTime;

use serde::{Deserialize, Serialize};

/// Where a log line came from.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum SourceKind {
    Docker,
    Kubernetes,
    Vm,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum Stream {
    Stdout,
    Stderr,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum Level {
    Unknown,
    Trace,
    Debug,
    Info,
    Warn,
    Error,
}

impl SourceKind {
    /// The inverse of [`SourceKind::as_str`].
    pub fn parse(s: &str) -> Option<Self> {
        match s {
            "docker" => Some(SourceKind::Docker),
            "kubernetes" => Some(SourceKind::Kubernetes),
            "vm" => Some(SourceKind::Vm),
            _ => None,
        }
    }

    pub fn as_str(&self) -> &'static str {
        match self {
            SourceKind::Docker => "docker",
            SourceKind::Kubernetes => "kubernetes",
            SourceKind::Vm => "vm",
        }
    }
}

impl Stream {
    /// The inverse of [`Stream::as_str`].
    pub fn parse(s: &str) -> Option<Self> {
        match s {
            "stdout" => Some(Stream::Stdout),
            "stderr" => Some(Stream::Stderr),
            _ => None,
        }
    }

    pub fn as_str(&self) -> &'static str {
        match self {
            Stream::Stdout => "stdout",
            Stream::Stderr => "stderr",
        }
    }
}

impl Level {
    /// The inverse of [`Level::as_str`].
    pub fn parse(s: &str) -> Option<Self> {
        match s {
            "unknown" => Some(Level::Unknown),
            "trace" => Some(Level::Trace),
            "debug" => Some(Level::Debug),
            "info" => Some(Level::Info),
            "warn" => Some(Level::Warn),
            "error" => Some(Level::Error),
            _ => None,
        }
    }

    pub fn as_str(&self) -> &'static str {
        match self {
            Level::Unknown => "unknown",
            Level::Trace => "trace",
            Level::Debug => "debug",
            Level::Info => "info",
            Level::Warn => "warn",
            Level::Error => "error",
        }
    }

    /// Best-effort level detection from an unstructured line.
    pub fn sniff(line: &str) -> Level {
        if let Some(level) = klog_level(line) {
            return level;
        }
        // The first 64 bytes, cut on a character boundary so multi-byte text can't split.
        let mut end = line.len().min(64);
        while !line.is_char_boundary(end) {
            end -= 1;
        }
        let upper = line[..end].to_ascii_uppercase();
        if upper.contains("ERROR") || upper.contains("FATAL") || upper.contains("PANIC") {
            Level::Error
        } else if upper.contains("WARN") {
            Level::Warn
        } else if upper.contains("INFO") {
            Level::Info
        } else if upper.contains("DEBUG") {
            Level::Debug
        } else if upper.contains("TRACE") {
            Level::Trace
        } else {
            Level::Unknown
        }
    }

    /// The level of `line` when `previous` is the level of the line before it from the same
    /// container. Stack traces arrive one frame per line with no level of their own, so lines
    /// that continue a trace keep the level of the line that started it.
    pub fn sniff_after(line: &str, previous: Level) -> Level {
        if previous != Level::Unknown && is_continuation(line) {
            previous
        } else {
            Level::sniff(line)
        }
    }
}

/// Kubernetes components (klog/glog) start each line with the level's letter and the date:
/// `I1002 23:47:01.123456 1 controller.go:42] ...`.
fn klog_level(line: &str) -> Option<Level> {
    let b = line.as_bytes();
    if b.len() < 6 || !b[1..5].iter().all(u8::is_ascii_digit) || b[5] != b' ' {
        return None;
    }
    match b[0] {
        b'I' => Some(Level::Info),
        b'W' => Some(Level::Warn),
        b'E' | b'F' => Some(Level::Error),
        _ => None,
    }
}

/// A line that continues the previous one: an indented stack frame, a chained cause, or the
/// exception line that heads a trace (`java.lang.IllegalStateException: ...`, `ValueError: ...`).
fn is_continuation(line: &str) -> bool {
    if line.starts_with([' ', '\t']) {
        return true;
    }
    if ["Caused by:", "Suppressed:", "Traceback (most recent call last):"]
        .iter()
        .any(|p| line.starts_with(p))
    {
        return true;
    }
    let head = line.split(':').next().unwrap_or_default();
    !head.is_empty()
        && head.chars().all(|c| c.is_ascii_alphanumeric() || "._$".contains(c))
        && ["Exception", "Error", "Throwable"].iter().any(|s| head.ends_with(s))
}

/// One log line plus its metadata.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LogRecord {
    pub timestamp: SystemTime,
    pub source: SourceKind,
    /// Human-readable origin, e.g. a container name or `namespace/pod/container`.
    pub origin: String,
    pub stream: Stream,
    pub level: Level,
    pub body: String,
    pub labels: BTreeMap<String, String>,
}

/// A log-producing target that can be tailed (a container, a pod, a VM file).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Target {
    pub id: String,
    pub name: String,
    pub source: SourceKind,
    pub state: String,
    pub labels: BTreeMap<String, String>,
}

/// Case-insensitive substring filter applied to records, optionally limited to a time range.
#[derive(Debug, Clone, Default)]
pub struct Filter {
    needle: String,
    pub min_level: Option<Level>,
    /// Inclusive lower bound on `LogRecord::timestamp`.
    pub from: Option<SystemTime>,
    /// Inclusive upper bound on `LogRecord::timestamp`.
    pub to: Option<SystemTime>,
    /// Per origin, excludes records at or after this time, because the caller already holds
    /// that origin's lines from there on.
    pub skip_from: HashMap<String, SystemTime>,
}

impl Filter {
    pub fn new(text: &str) -> Self {
        Filter {
            needle: text.to_lowercase(),
            ..Default::default()
        }
    }

    pub fn is_empty(&self) -> bool {
        self.needle.is_empty()
            && self.min_level.is_none()
            && self.from.is_none()
            && self.to.is_none()
            && self.skip_from.is_empty()
    }

    pub fn matches(&self, record: &LogRecord) -> bool {
        if let Some(min) = self.min_level
            && record.level < min
        {
            return false;
        }
        if self.from.is_some_and(|from| record.timestamp < from) || self.to.is_some_and(|to| record.timestamp > to) {
            return false;
        }
        if self
            .skip_from
            .get(&record.origin)
            .is_some_and(|&skip| record.timestamp >= skip)
        {
            return false;
        }
        self.needle.is_empty()
            || record.body.to_lowercase().contains(&self.needle)
            || record.origin.to_lowercase().contains(&self.needle)
    }
}

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

    fn record(origin: &str, body: &str) -> LogRecord {
        LogRecord {
            timestamp: SystemTime::UNIX_EPOCH,
            source: SourceKind::Docker,
            origin: origin.into(),
            stream: Stream::Stdout,
            level: Level::sniff(body),
            body: body.into(),
            labels: BTreeMap::new(),
        }
    }

    #[test]
    fn sniffs_levels() {
        assert_eq!(Level::sniff("2026-10-01 ERROR db down"), Level::Error);
        assert_eq!(Level::sniff("[warn] slow query"), Level::Warn);
        assert_eq!(Level::sniff("hello"), Level::Unknown);
    }

    #[test]
    fn sniffs_klog_levels() {
        assert_eq!(
            Level::sniff("I1002 23:47:01.123456 1 leader.go:42] renewed lease"),
            Level::Info
        );
        assert_eq!(
            Level::sniff("W1002 23:47:01.123456 1 reflector.go:561] watch closed"),
            Level::Warn
        );
        assert_eq!(
            Level::sniff("E1002 23:47:01.123456 1 controller.go:9] sync failed"),
            Level::Error
        );
        // A klog info line that mentions an error is still info.
        assert_eq!(
            Level::sniff("I1002 23:47:01.123456 1 x.go:1] retrying after error"),
            Level::Info
        );
        assert_eq!(Level::sniff("Ingress controller started"), Level::Unknown);
    }

    #[test]
    fn sniff_survives_multibyte_text_at_the_cut() {
        // 63 ASCII bytes then a 3-byte character straddling byte 64.
        let line = format!("{}€ ERROR", "a".repeat(63));
        assert_eq!(Level::sniff(&line), Level::Unknown);
        assert_eq!(Level::sniff("✓ ok"), Level::Unknown);
    }

    #[test]
    fn stack_traces_keep_the_level_of_their_first_line() {
        let lines = [
            "2026-10-02T23:47:01.123Z ERROR 1 --- [nio-8080-exec-1] o.a.c.c.C.[dispatcherServlet] : Servlet failed",
            "org.springframework.web.client.HttpServerErrorException$InternalServerError: 500 on POST",
            "\tat org.springframework.web.client.DefaultResponseErrorHandler.handleError(DefaultResponseErrorHandler.java:89)",
            "Caused by: java.net.SocketTimeoutException: Read timed out",
            "\t... 16 more",
            "2026-10-02T23:47:02.001Z  INFO 1 --- [main] c.d.payment.App : recovered",
        ];
        let mut level = Level::Unknown;
        let levels: Vec<Level> = lines
            .iter()
            .map(|line| {
                level = Level::sniff_after(line, level);
                level
            })
            .collect();
        assert_eq!(
            levels,
            [
                Level::Error,
                Level::Error,
                Level::Error,
                Level::Error,
                Level::Error,
                Level::Info
            ]
        );
        // With nothing before it, an indented line has no level to inherit.
        assert_eq!(Level::sniff_after("\tat x.y(Z.java:1)", Level::Unknown), Level::Unknown);
    }

    #[test]
    fn filter_matches_body_and_origin_case_insensitively() {
        let r = record("api-1", "GET /Health 200");
        assert!(Filter::new("health").matches(&r));
        assert!(Filter::new("API").matches(&r));
        assert!(!Filter::new("postgres").matches(&r));
        assert!(Filter::new("").matches(&r));
    }

    #[test]
    fn filter_skips_lines_an_origin_already_holds() {
        let at = |secs| SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(secs);
        let mut filter = Filter::new("");
        filter.skip_from.insert("api-1".into(), at(100));
        let line = |origin, secs| LogRecord {
            timestamp: at(secs),
            ..record(origin, "x")
        };
        assert!(filter.matches(&line("api-1", 99)));
        assert!(!filter.matches(&line("api-1", 100)));
        assert!(!filter.matches(&line("api-1", 500)));
        // Other origins, such as a removed container, are untouched.
        assert!(filter.matches(&line("gone", 500)));
        assert!(!filter.is_empty());
    }

    #[test]
    fn enum_names_round_trip() {
        for kind in [SourceKind::Docker, SourceKind::Kubernetes, SourceKind::Vm] {
            assert_eq!(SourceKind::parse(kind.as_str()), Some(kind));
        }
        for stream in [Stream::Stdout, Stream::Stderr] {
            assert_eq!(Stream::parse(stream.as_str()), Some(stream));
        }
        for level in [
            Level::Unknown,
            Level::Trace,
            Level::Debug,
            Level::Info,
            Level::Warn,
            Level::Error,
        ] {
            assert_eq!(Level::parse(level.as_str()), Some(level));
        }
        assert_eq!(Level::parse("nope"), None);
    }

    #[test]
    fn filter_respects_time_range() {
        use std::time::Duration;
        let at = |secs| {
            let mut r = record("a", "x");
            r.timestamp = SystemTime::UNIX_EPOCH + Duration::from_secs(secs);
            r
        };
        let mut f = Filter::new("");
        f.from = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10));
        f.to = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(20));
        assert!(!f.is_empty());
        assert!(!f.matches(&at(9)));
        assert!(f.matches(&at(10)));
        assert!(f.matches(&at(20)));
        assert!(!f.matches(&at(21)));
    }

    #[test]
    fn filter_respects_min_level() {
        let mut f = Filter::new("");
        f.min_level = Some(Level::Warn);
        assert!(f.matches(&record("a", "ERROR boom")));
        assert!(!f.matches(&record("a", "INFO ok")));
    }
}