telelog_core/
lib.rs

1//! The log model shared by every Telelogs component.
2
3use std::collections::BTreeMap;
4use std::collections::HashMap;
5use std::time::SystemTime;
6
7use serde::{Deserialize, Serialize};
8
9/// Where a log line came from.
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
11pub enum SourceKind {
12    Docker,
13    Kubernetes,
14    Vm,
15}
16
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
18pub enum Stream {
19    Stdout,
20    Stderr,
21}
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
24pub enum Level {
25    Unknown,
26    Trace,
27    Debug,
28    Info,
29    Warn,
30    Error,
31}
32
33impl SourceKind {
34    /// The inverse of [`SourceKind::as_str`].
35    pub fn parse(s: &str) -> Option<Self> {
36        match s {
37            "docker" => Some(SourceKind::Docker),
38            "kubernetes" => Some(SourceKind::Kubernetes),
39            "vm" => Some(SourceKind::Vm),
40            _ => None,
41        }
42    }
43
44    pub fn as_str(&self) -> &'static str {
45        match self {
46            SourceKind::Docker => "docker",
47            SourceKind::Kubernetes => "kubernetes",
48            SourceKind::Vm => "vm",
49        }
50    }
51}
52
53impl Stream {
54    /// The inverse of [`Stream::as_str`].
55    pub fn parse(s: &str) -> Option<Self> {
56        match s {
57            "stdout" => Some(Stream::Stdout),
58            "stderr" => Some(Stream::Stderr),
59            _ => None,
60        }
61    }
62
63    pub fn as_str(&self) -> &'static str {
64        match self {
65            Stream::Stdout => "stdout",
66            Stream::Stderr => "stderr",
67        }
68    }
69}
70
71impl Level {
72    /// The inverse of [`Level::as_str`].
73    pub fn parse(s: &str) -> Option<Self> {
74        match s {
75            "unknown" => Some(Level::Unknown),
76            "trace" => Some(Level::Trace),
77            "debug" => Some(Level::Debug),
78            "info" => Some(Level::Info),
79            "warn" => Some(Level::Warn),
80            "error" => Some(Level::Error),
81            _ => None,
82        }
83    }
84
85    pub fn as_str(&self) -> &'static str {
86        match self {
87            Level::Unknown => "unknown",
88            Level::Trace => "trace",
89            Level::Debug => "debug",
90            Level::Info => "info",
91            Level::Warn => "warn",
92            Level::Error => "error",
93        }
94    }
95
96    /// Best-effort level detection from an unstructured line.
97    pub fn sniff(line: &str) -> Level {
98        if let Some(level) = klog_level(line) {
99            return level;
100        }
101        // The first 64 bytes, cut on a character boundary so multi-byte text can't split.
102        let mut end = line.len().min(64);
103        while !line.is_char_boundary(end) {
104            end -= 1;
105        }
106        let upper = line[..end].to_ascii_uppercase();
107        if upper.contains("ERROR") || upper.contains("FATAL") || upper.contains("PANIC") {
108            Level::Error
109        } else if upper.contains("WARN") {
110            Level::Warn
111        } else if upper.contains("INFO") {
112            Level::Info
113        } else if upper.contains("DEBUG") {
114            Level::Debug
115        } else if upper.contains("TRACE") {
116            Level::Trace
117        } else {
118            Level::Unknown
119        }
120    }
121
122    /// The level of `line` when `previous` is the level of the line before it from the same
123    /// container. Stack traces arrive one frame per line with no level of their own, so lines
124    /// that continue a trace keep the level of the line that started it.
125    pub fn sniff_after(line: &str, previous: Level) -> Level {
126        if previous != Level::Unknown && is_continuation(line) {
127            previous
128        } else {
129            Level::sniff(line)
130        }
131    }
132}
133
134/// Kubernetes components (klog/glog) start each line with the level's letter and the date:
135/// `I1002 23:47:01.123456 1 controller.go:42] ...`.
136fn klog_level(line: &str) -> Option<Level> {
137    let b = line.as_bytes();
138    if b.len() < 6 || !b[1..5].iter().all(u8::is_ascii_digit) || b[5] != b' ' {
139        return None;
140    }
141    match b[0] {
142        b'I' => Some(Level::Info),
143        b'W' => Some(Level::Warn),
144        b'E' | b'F' => Some(Level::Error),
145        _ => None,
146    }
147}
148
149/// A line that continues the previous one: an indented stack frame, a chained cause, or the
150/// exception line that heads a trace (`java.lang.IllegalStateException: ...`, `ValueError: ...`).
151fn is_continuation(line: &str) -> bool {
152    if line.starts_with([' ', '\t']) {
153        return true;
154    }
155    if ["Caused by:", "Suppressed:", "Traceback (most recent call last):"]
156        .iter()
157        .any(|p| line.starts_with(p))
158    {
159        return true;
160    }
161    let head = line.split(':').next().unwrap_or_default();
162    !head.is_empty()
163        && head.chars().all(|c| c.is_ascii_alphanumeric() || "._$".contains(c))
164        && ["Exception", "Error", "Throwable"].iter().any(|s| head.ends_with(s))
165}
166
167/// One log line plus its metadata.
168#[derive(Debug, Clone, Serialize, Deserialize)]
169pub struct LogRecord {
170    pub timestamp: SystemTime,
171    pub source: SourceKind,
172    /// Human-readable origin, e.g. a container name or `namespace/pod/container`.
173    pub origin: String,
174    pub stream: Stream,
175    pub level: Level,
176    pub body: String,
177    pub labels: BTreeMap<String, String>,
178}
179
180/// A log-producing target that can be tailed (a container, a pod, a VM file).
181#[derive(Debug, Clone, Serialize, Deserialize)]
182pub struct Target {
183    pub id: String,
184    pub name: String,
185    pub source: SourceKind,
186    pub state: String,
187    pub labels: BTreeMap<String, String>,
188}
189
190/// Case-insensitive substring filter applied to records, optionally limited to a time range.
191#[derive(Debug, Clone, Default)]
192pub struct Filter {
193    needle: String,
194    pub min_level: Option<Level>,
195    /// Inclusive lower bound on `LogRecord::timestamp`.
196    pub from: Option<SystemTime>,
197    /// Inclusive upper bound on `LogRecord::timestamp`.
198    pub to: Option<SystemTime>,
199    /// Per origin, excludes records at or after this time, because the caller already holds
200    /// that origin's lines from there on.
201    pub skip_from: HashMap<String, SystemTime>,
202}
203
204impl Filter {
205    pub fn new(text: &str) -> Self {
206        Filter {
207            needle: text.to_lowercase(),
208            ..Default::default()
209        }
210    }
211
212    pub fn is_empty(&self) -> bool {
213        self.needle.is_empty()
214            && self.min_level.is_none()
215            && self.from.is_none()
216            && self.to.is_none()
217            && self.skip_from.is_empty()
218    }
219
220    pub fn matches(&self, record: &LogRecord) -> bool {
221        if let Some(min) = self.min_level
222            && record.level < min
223        {
224            return false;
225        }
226        if self.from.is_some_and(|from| record.timestamp < from) || self.to.is_some_and(|to| record.timestamp > to) {
227            return false;
228        }
229        if self
230            .skip_from
231            .get(&record.origin)
232            .is_some_and(|&skip| record.timestamp >= skip)
233        {
234            return false;
235        }
236        self.needle.is_empty()
237            || record.body.to_lowercase().contains(&self.needle)
238            || record.origin.to_lowercase().contains(&self.needle)
239    }
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245
246    fn record(origin: &str, body: &str) -> LogRecord {
247        LogRecord {
248            timestamp: SystemTime::UNIX_EPOCH,
249            source: SourceKind::Docker,
250            origin: origin.into(),
251            stream: Stream::Stdout,
252            level: Level::sniff(body),
253            body: body.into(),
254            labels: BTreeMap::new(),
255        }
256    }
257
258    #[test]
259    fn sniffs_levels() {
260        assert_eq!(Level::sniff("2026-10-01 ERROR db down"), Level::Error);
261        assert_eq!(Level::sniff("[warn] slow query"), Level::Warn);
262        assert_eq!(Level::sniff("hello"), Level::Unknown);
263    }
264
265    #[test]
266    fn sniffs_klog_levels() {
267        assert_eq!(
268            Level::sniff("I1002 23:47:01.123456 1 leader.go:42] renewed lease"),
269            Level::Info
270        );
271        assert_eq!(
272            Level::sniff("W1002 23:47:01.123456 1 reflector.go:561] watch closed"),
273            Level::Warn
274        );
275        assert_eq!(
276            Level::sniff("E1002 23:47:01.123456 1 controller.go:9] sync failed"),
277            Level::Error
278        );
279        // A klog info line that mentions an error is still info.
280        assert_eq!(
281            Level::sniff("I1002 23:47:01.123456 1 x.go:1] retrying after error"),
282            Level::Info
283        );
284        assert_eq!(Level::sniff("Ingress controller started"), Level::Unknown);
285    }
286
287    #[test]
288    fn sniff_survives_multibyte_text_at_the_cut() {
289        // 63 ASCII bytes then a 3-byte character straddling byte 64.
290        let line = format!("{}€ ERROR", "a".repeat(63));
291        assert_eq!(Level::sniff(&line), Level::Unknown);
292        assert_eq!(Level::sniff("✓ ok"), Level::Unknown);
293    }
294
295    #[test]
296    fn stack_traces_keep_the_level_of_their_first_line() {
297        let lines = [
298            "2026-10-02T23:47:01.123Z ERROR 1 --- [nio-8080-exec-1] o.a.c.c.C.[dispatcherServlet] : Servlet failed",
299            "org.springframework.web.client.HttpServerErrorException$InternalServerError: 500 on POST",
300            "\tat org.springframework.web.client.DefaultResponseErrorHandler.handleError(DefaultResponseErrorHandler.java:89)",
301            "Caused by: java.net.SocketTimeoutException: Read timed out",
302            "\t... 16 more",
303            "2026-10-02T23:47:02.001Z  INFO 1 --- [main] c.d.payment.App : recovered",
304        ];
305        let mut level = Level::Unknown;
306        let levels: Vec<Level> = lines
307            .iter()
308            .map(|line| {
309                level = Level::sniff_after(line, level);
310                level
311            })
312            .collect();
313        assert_eq!(
314            levels,
315            [
316                Level::Error,
317                Level::Error,
318                Level::Error,
319                Level::Error,
320                Level::Error,
321                Level::Info
322            ]
323        );
324        // With nothing before it, an indented line has no level to inherit.
325        assert_eq!(Level::sniff_after("\tat x.y(Z.java:1)", Level::Unknown), Level::Unknown);
326    }
327
328    #[test]
329    fn filter_matches_body_and_origin_case_insensitively() {
330        let r = record("api-1", "GET /Health 200");
331        assert!(Filter::new("health").matches(&r));
332        assert!(Filter::new("API").matches(&r));
333        assert!(!Filter::new("postgres").matches(&r));
334        assert!(Filter::new("").matches(&r));
335    }
336
337    #[test]
338    fn filter_skips_lines_an_origin_already_holds() {
339        let at = |secs| SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(secs);
340        let mut filter = Filter::new("");
341        filter.skip_from.insert("api-1".into(), at(100));
342        let line = |origin, secs| LogRecord {
343            timestamp: at(secs),
344            ..record(origin, "x")
345        };
346        assert!(filter.matches(&line("api-1", 99)));
347        assert!(!filter.matches(&line("api-1", 100)));
348        assert!(!filter.matches(&line("api-1", 500)));
349        // Other origins, such as a removed container, are untouched.
350        assert!(filter.matches(&line("gone", 500)));
351        assert!(!filter.is_empty());
352    }
353
354    #[test]
355    fn enum_names_round_trip() {
356        for kind in [SourceKind::Docker, SourceKind::Kubernetes, SourceKind::Vm] {
357            assert_eq!(SourceKind::parse(kind.as_str()), Some(kind));
358        }
359        for stream in [Stream::Stdout, Stream::Stderr] {
360            assert_eq!(Stream::parse(stream.as_str()), Some(stream));
361        }
362        for level in [
363            Level::Unknown,
364            Level::Trace,
365            Level::Debug,
366            Level::Info,
367            Level::Warn,
368            Level::Error,
369        ] {
370            assert_eq!(Level::parse(level.as_str()), Some(level));
371        }
372        assert_eq!(Level::parse("nope"), None);
373    }
374
375    #[test]
376    fn filter_respects_time_range() {
377        use std::time::Duration;
378        let at = |secs| {
379            let mut r = record("a", "x");
380            r.timestamp = SystemTime::UNIX_EPOCH + Duration::from_secs(secs);
381            r
382        };
383        let mut f = Filter::new("");
384        f.from = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10));
385        f.to = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(20));
386        assert!(!f.is_empty());
387        assert!(!f.matches(&at(9)));
388        assert!(f.matches(&at(10)));
389        assert!(f.matches(&at(20)));
390        assert!(!f.matches(&at(21)));
391    }
392
393    #[test]
394    fn filter_respects_min_level() {
395        let mut f = Filter::new("");
396        f.min_level = Some(Level::Warn);
397        assert!(f.matches(&record("a", "ERROR boom")));
398        assert!(!f.matches(&record("a", "INFO ok")));
399    }
400}
telelog_core/lib.rs

telelog_core/
lib.rs
