Remove dead parse_log_line public API (DSK-136)

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));
             });
         });

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};

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<DateTime<Utc>>,
 }
 
 impl LogLineTimestampResolver {
-    fn resolve(
+    pub fn resolve(
         &mut self,
         parsed: Option<DateTime<Utc>>,
         msg: Arc<str>,
@@ -213,7 +214,8 @@ impl LogLineTimestampResolver {
     }
 }
 
-fn split_log_line(raw: &[u8]) -> (Option<DateTime<Utc>>, Arc<str>) {
+#[doc(hidden)]
+pub fn split_log_line(raw: &[u8]) -> (Option<DateTime<Utc>>, Arc<str>) {
     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<DateTime<Utc>>, Arc<str>) {
     (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<Utc>, Arc<str>) {
-    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<Utc>, Arc<str>) {
-    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");
     }
 }