diff --git a/crates/rustern-core/benches/hot_path.rs b/crates/rustern-core/benches/hot_path.rs index 79a513c..3f601c0 100644 --- a/crates/rustern-core/benches/hot_path.rs +++ b/crates/rustern-core/benches/hot_path.rs @@ -7,7 +7,6 @@ use criterion::{BenchmarkId, Criterion, Throughput, black_box, criterion_group, use futures::stream::{self, Stream, StreamExt}; use regex::Regex; use rustern_core::format_display::{TimestampStyle, TimestampZone}; -use rustern_core::parse_log_line; use rustern_core::pipeline::{ ColorAssignOpts, ExitWatchState, QueryMode, include_exclude, jq_evaluate, json_annotate, level_classify, validate_filter, @@ -21,6 +20,7 @@ use rustern_core::runtime::PipelineSpecBuilder; use rustern_core::source::{ ContextName, Labels, LogEvent, LogSourceError, SourceKey, SourceKind, SourceMeta, }; +use rustern_core::{LogLineTimestampResolver, split_log_line}; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; @@ -359,15 +359,16 @@ fn bench_parse_log_line(c: &mut Criterion) { ] { let line = kube_log_line(&msg); group.bench_with_input(BenchmarkId::new("parse", name), &line, |b, line| { - b.iter(|| black_box(parse_log_line(black_box(line.as_str())))); + b.iter(|| { + let (parsed, msg) = split_log_line(black_box(line.as_bytes())); + black_box((parsed, msg)); + }); }); } group.finish(); } fn bench_ingest_log_line(c: &mut Criterion) { - use rustern_core::parse_log_line_bytes; - let mut group = c.benchmark_group("ingest_log_line"); for (name, msg) in [ @@ -377,8 +378,10 @@ fn bench_ingest_log_line(c: &mut Criterion) { let mut buf = kube_log_line(&msg).into_bytes(); buf.push(b'\n'); group.bench_with_input(BenchmarkId::new("buffer_to_event", name), &buf, |b, buf| { + let mut resolver = LogLineTimestampResolver::default(); b.iter(|| { - let (ts, message) = parse_log_line_bytes(black_box(buf.as_slice())); + let (parsed, msg) = split_log_line(black_box(buf.as_slice())); + let (ts, message) = resolver.resolve(parsed, msg); black_box((ts, message)); }); }); diff --git a/crates/rustern-core/src/lib.rs b/crates/rustern-core/src/lib.rs index c5b025c..8409a37 100644 --- a/crates/rustern-core/src/lib.rs +++ b/crates/rustern-core/src/lib.rs @@ -25,4 +25,6 @@ pub use runtime::{ }; #[cfg(feature = "bench")] pub use source::ScriptLogSourceOpener; -pub use source::pod_log::{PodLogRequest, parse_log_line, parse_log_line_bytes}; +pub use source::pod_log::PodLogRequest; +#[cfg(feature = "bench")] +pub use source::pod_log::{LogLineTimestampResolver, split_log_line}; diff --git a/crates/rustern-core/src/source/pod_log.rs b/crates/rustern-core/src/source/pod_log.rs index 61e998e..ea04cd0 100644 --- a/crates/rustern-core/src/source/pod_log.rs +++ b/crates/rustern-core/src/source/pod_log.rs @@ -190,12 +190,13 @@ fn trim_line_end(mut raw: &[u8]) -> &[u8] { /// advance +1µs from the previous line on the same stream so ordering and /// `--cursor-reconnect` cursors stay stable instead of jumping to wall clock. #[derive(Debug, Default)] -struct LogLineTimestampResolver { +#[doc(hidden)] +pub struct LogLineTimestampResolver { last: Option>, } impl LogLineTimestampResolver { - fn resolve( + pub fn resolve( &mut self, parsed: Option>, msg: Arc, @@ -213,7 +214,8 @@ impl LogLineTimestampResolver { } } -fn split_log_line(raw: &[u8]) -> (Option>, Arc) { +#[doc(hidden)] +pub fn split_log_line(raw: &[u8]) -> (Option>, Arc) { let raw = trim_line_end(raw); if let Some(sp) = raw.iter().position(|&b| b == b' ') && let Ok(ts_s) = std::str::from_utf8(&raw[..sp]) @@ -233,22 +235,6 @@ fn split_log_line(raw: &[u8]) -> (Option>, Arc) { (None, msg) } -/// Parse a kube log line with an optional RFC3339 prefix into `(timestamp, message)`. -/// -/// Standalone calls without stream context assign [`DateTime::UNIX_EPOCH`] to undated lines. -pub fn parse_log_line(raw: &str) -> (DateTime, Arc) { - let mut resolver = LogLineTimestampResolver::default(); - let (parsed, msg) = split_log_line(raw.as_bytes()); - resolver.resolve(parsed, msg) -} - -/// Parse a newline-terminated kube log line from a reusable read buffer. -pub fn parse_log_line_bytes(raw: &[u8]) -> (DateTime, Arc) { - let mut resolver = LogLineTimestampResolver::default(); - let (parsed, msg) = split_log_line(raw); - resolver.resolve(parsed, msg) -} - impl LogSource for PodLogSource { fn meta(&self) -> &SourceMeta { &self.meta @@ -321,8 +307,9 @@ mod tests { } #[test] - fn parse_log_line_strips_rfc3339_prefix() { - let (ts, msg) = parse_log_line("2024-03-15T10:30:45Z hello world"); + fn split_log_line_strips_rfc3339_prefix() { + let (parsed, msg) = split_log_line(b"2024-03-15T10:30:45Z hello world"); + let ts = parsed.expect("timestamp"); assert_eq!( ts, DateTime::parse_from_rfc3339("2024-03-15T10:30:45Z") @@ -333,10 +320,10 @@ mod tests { } #[test] - fn parse_log_line_fallback_without_timestamp() { - let (ts, msg) = parse_log_line("plain line without prefix"); + fn split_log_line_without_timestamp_prefix() { + let (parsed, msg) = split_log_line(b"plain line without prefix"); + assert!(parsed.is_none()); assert_eq!(&*msg, "plain line without prefix"); - assert_eq!(ts, DateTime::UNIX_EPOCH); } #[test] @@ -355,11 +342,16 @@ mod tests { } #[test] - fn parse_log_line_bytes_matches_str_parser() { + fn split_log_line_trims_trailing_newline() { let line = b"2024-03-15T10:30:45Z hello world\n"; - let (ts_str, msg_str) = parse_log_line("2024-03-15T10:30:45Z hello world"); - let (ts_bytes, msg_bytes) = parse_log_line_bytes(line); - assert_eq!(ts_str, ts_bytes); - assert_eq!(msg_str, msg_bytes); + let (parsed, msg) = split_log_line(line); + let ts = parsed.expect("timestamp"); + assert_eq!( + ts, + DateTime::parse_from_rfc3339("2024-03-15T10:30:45Z") + .unwrap() + .with_timezone(&Utc) + ); + assert_eq!(&*msg, "hello world"); } }