From de9ba6f01b663c792eb152cf832cee03b7fdd1c4 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Sun, 6 Sep 2026 13:23:10 +0100 Subject: [PATCH] Resolve tickets 059-061: generic log-stream columns + sort helper indexes - 059: log stream header/column detection is now generic whitespace-columnar splitting (2+ spaces, single tab) instead of one fixed six-column regex. Declared params come from whatever columns the header actually names, plus a trailing unlabelled message field that's never swallowed by the last named column. - 060: timeline sort-by-property now walks an incrementally-maintained per-sort-key helper index (BTreeMap keyed so "has the param, ordered by value" sorts before "missing it, ordered by ottyl_seq") instead of materializing and re-sorting the full filtered range on every query. - 061: helper indexes are swept for staleness on ingest and dropped after 5 minutes unused; a purged (or never-built) index transparently rebuilds via the same lazy one-time scan on the next sorted query. - Also invalidate a param's helper index on manual re-declaration, removal, or ClearParams, so a redefined regex can't leave a sort helper index serving values derived under the old declaration. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS --- src/engine.rs | 482 +++++++++++++++--- src/logfmt.rs | 316 +++++++++++- ...g-should-be-generic-whitespace-columnar.md | 62 --- ...custom-sort-needs-helper-btreemap-index.md | 57 --- .../061-purge-unused-sort-helper-indexes.md | 43 -- 5 files changed, 699 insertions(+), 261 deletions(-) delete mode 100644 tickets/059-log-stream-parsing-should-be-generic-whitespace-columnar.md delete mode 100644 tickets/060-timeline-custom-sort-needs-helper-btreemap-index.md delete mode 100644 tickets/061-purge-unused-sort-helper-indexes.md diff --git a/src/engine.rs b/src/engine.rs index 8992f1d..c2383a0 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -164,6 +164,13 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle // every currently-ingested log, kept in lockstep with `log_count` — incremented // here on ingest, decremented in `evict_excess_logs` when a log is dropped. let mut approx_bytes_total: u64 = 0; + // Ticket 060/061: one entry per param name that's ever been used as a `sort_by`, populated + // lazily on first sorted query and kept incrementally up to date on every ingest thereafter; + // swept for staleness on ingest too (see `purge_stale_sort_helper_indexes`). + let mut sort_helper_indexes: HashMap = HashMap::new(); + // Ticket 059: the most recently seen `log stream` header line's column names, cached across + // ingested lines since a `log stream` pipe only prints its header once, before any data line. + let mut log_stream_header: Option = None; while let Some(message) = rx.recv().await { match message { @@ -175,9 +182,11 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle DataValue::LogLiteral { text: text.clone() }, ); log_count += 1; - auto_declare_log_stream_params(&mut index, &text); + auto_declare_log_stream_params(&mut index, &mut log_stream_header, &text); auto_declare_params(&mut index, &text); apply_all_metrics(&mut index, sequence, timestamp, &text); + update_sort_helper_indexes_on_ingest(&mut sort_helper_indexes, &index, sequence, timestamp, &text); + purge_stale_sort_helper_indexes(&mut sort_helper_indexes); evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); // Evaluate alert rules against just this line, then hand any fired rule off to delivery via tokio::spawn so it stays off the hot path. @@ -212,16 +221,26 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); } EngineMessage::DeclareParam { request, respond_to } => { + // A manual re-declaration can change the regex/kind under the same name — drop any + // helper index built under the old declaration so a sorted query rebuilds it fresh + // rather than serving values derived from the now-stale regex. + sort_helper_indexes.remove(&request.name); let result = declare_param(&mut index, request); let _ = respond_to.send(result); } EngineMessage::RemoveParamDeclaration { name, respond_to } => { - let existed = index.remove(&IndexKey::ParamDeclaration { name }).is_some(); + let existed = index.remove(&IndexKey::ParamDeclaration { name: name.clone() }).is_some(); + // A helper index built under the old declaration (or a same-named one re-declared + // with a different regex/kind later) would otherwise keep serving stale derived + // values until ticket 061's sweep happens to purge it — dropping it here instead + // means the next sorted query for this param rebuilds it fresh. + sort_helper_indexes.remove(&name); let _ = respond_to.send(existed); } EngineMessage::ClearParams { respond_to } => { let removed = clear_all_param_declarations(&mut index); let (reextracted, logs_considered) = reextract_params_from_retained_logs(&mut index); + sort_helper_indexes.clear(); let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered }); } EngineMessage::DeclareMetric { request, respond_to } => { @@ -237,7 +256,7 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle let _ = respond_to.send(existed); } EngineMessage::Query { query, respond_to } => { - let response = handle_query(&index, query, approx_bytes_total); + let response = handle_query(&index, query, approx_bytes_total, &mut sort_helper_indexes); let _ = respond_to.send(response); } } @@ -438,43 +457,68 @@ fn auto_declare_params(index: &mut BTreeMap, text: &str) { } } -/// Declares `log_type`/`process`/`pid` as standalone params the first time a `log stream`-shaped line -/// is seen (ticket 050/054) — `Type`, the process name, and `PID` are low-cardinality, genuinely -/// useful columns derivable directly from `log stream`'s fixed-column shape -/// (`crate::logfmt::is_log_stream_line`), with none of the `=`-scanning ambiguity `auto_declare_params` -/// has to guard against. Gated on that shape so ordinary logfmt/JSON/prose input never gets these three -/// params. Unlike `auto_declare_params`, adding these is never *skipped* on account of -/// `MAX_AUTO_DECLARED_PARAMS` — it can only ever add exactly these three names, once each — though -/// since they're stored as `auto_derived` too, once present they do count against that shared ceiling, -/// tightening (slightly) the budget left for generic `key=value` pairs. Like `auto_declare_params`, -/// never overwrites an existing declaration (manual or auto) for any of the three names. -fn auto_declare_log_stream_params(index: &mut BTreeMap, text: &str) { +/// Declares one param per column of the most-recently-seen `log stream` header line, plus a trailing +/// unlabelled `message` param for whatever text follows the last named column (ticket 059) — replacing +/// the old fixed three (`log_type`/`process`/`pid`) with whatever columns the header actually names. +/// `header` is threaded through by the caller (`run`'s ingest loop) rather than owned here: a +/// `log stream` pipe prints its header once, before any data line, and this function only ever sees +/// one ingested line at a time, so the header has to be cached across calls somewhere. A later header +/// line (e.g. a fresh `log stream` invocation piped into the same run) simply replaces the cached one. +/// Unlike `auto_declare_params`, adding these is never *skipped* on account of +/// `MAX_AUTO_DECLARED_PARAMS` — though since they're stored as `auto_derived` too, once present they do +/// count against that shared ceiling, tightening (slightly) the budget left for generic `key=value` +/// pairs. Like `auto_declare_params`, never overwrites an existing declaration (manual or auto) for +/// any of these names. +fn auto_declare_log_stream_params( + index: &mut BTreeMap, + header: &mut Option, + text: &str, +) { + if let Some(detected) = crate::logfmt::detect_log_stream_header(text) { + *header = Some(detected); + return; + } + let Some(header) = header.as_ref() else { + return; + }; if !crate::logfmt::is_log_stream_line(text) { return; } - let candidates: [(&str, &str, &str, MetricPrimitiveKind); 3] = [ - ("log_type", crate::logfmt::LOG_STREAM_TYPE_PATTERN, "log_type", MetricPrimitiveKind::String), - ("process", crate::logfmt::LOG_STREAM_PROCESS_PATTERN, "process", MetricPrimitiveKind::String), - ("pid", crate::logfmt::LOG_STREAM_PID_PATTERN, "pid", MetricPrimitiveKind::Number), - ]; - for (name, pattern, group, kind) in candidates { - let index_key = IndexKey::ParamDeclaration { name: name.to_string() }; + let row = crate::logfmt::split_log_stream_row(text, header); + + for (i, column) in header.columns.iter().enumerate() { + let Some(value) = row.get(i) else { continue }; + let index_key = IndexKey::ParamDeclaration { name: column.name.clone() }; if index.contains_key(&index_key) { continue; } index.insert( index_key, DataValue::ParamDeclaration(ParamDeclaration { - name: name.to_string(), - message_match_regex: pattern.to_string(), - capture_group: group.to_string(), + name: column.name.clone(), + message_match_regex: crate::logfmt::log_stream_column_pattern(header, i, &column.name), + capture_group: column.name.clone(), json_field_path: None, - kind, + kind: infer_kind(value.trim()), elapsed_since_last: false, auto_derived: true, }), ); } + + if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) { + index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| { + DataValue::ParamDeclaration(ParamDeclaration { + name: "message".to_string(), + message_match_regex: crate::logfmt::log_stream_message_pattern(header, "message"), + capture_group: "message".to_string(), + json_field_path: None, + kind: infer_kind(message.trim()), + elapsed_since_last: false, + auto_derived: true, + }) + }); + } } /// Ticket 052: removes every `IndexKey::ParamDeclaration` entry — manual and auto-derived alike. @@ -1285,6 +1329,103 @@ fn param_declarations_range( /// meaningful reading as a standalone logs-table column (its value depends on which metric's label /// set it's grouped under, which only exists once it's referenced by a metric's `label_params`/ /// `value_param`). +/// Ticket 060: key type for a sort helper index — one such index exists per param name that's ever +/// been used as a `sort_by`. Declaration order matters for the derived `Ord`: every `WithParam` key +/// sorts before every `WithoutParam` key, and within each variant ordering falls out of the tuple +/// fields (value then sequence; sequence alone), which is exactly `sort_missing_fallback`'s "sorted +/// lines first, then lines missing the param ordered by `ottyl_seq`" semantics — expressed as part of +/// the index's own ordering instead of a separate partition-and-append step per query. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +enum SortHelperKey { + WithParam(MetricPrimitive, u64), + WithoutParam(u64), +} + +/// One param's helper index plus when a query last actually read from it — ticket 061 sweeps entries +/// whose `last_accessed` is stale. The map's value is that line's `Timestamp`, not `()`, purely so a +/// query can reconstruct the exact `IndexKey::IngestedLog { sequence, timestamp }` primary-index key +/// and look its text up in O(log n) instead of re-scanning `logs_range` to find it by sequence alone. +struct SortHelperIndexEntry { + index: BTreeMap, + last_accessed: std::time::Instant, +} + +/// How long a sort helper index may sit unread before ticket 061's sweep drops it. Not exposed via +/// `Settings` (ticket 061 explicitly scopes that out) — a dropped index is always transparently +/// rebuildable on the next query for that sort key, so this is purely a memory-bound knob, not a +/// user-visible behavior change. +const SORT_HELPER_INDEX_PURGE_AFTER: std::time::Duration = std::time::Duration::from_secs(300); + +/// Derives a single named param's value from `text`, for helper-index maintenance where deriving +/// every declared param (`derive_log_params`) per ingested line would be wasted work when only one +/// sort key's helper index actually needs updating. +fn derive_single_param( + text: &str, + declarations: &[(String, Regex, String, Option, MetricPrimitiveKind)], + param_name: &str, +) -> Option { + let (_, regex, capture_group, json_field_path, kind) = + declarations.iter().find(|(name, ..)| name == param_name)?; + if *kind == MetricPrimitiveKind::Boolean { + return Some(MetricPrimitive::Boolean(regex.is_match(text))); + } + let captures = regex.captures(text)?; + let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) { + Some(field_path) => crate::data_value::extract_json_field(text, field_path)?, + None => captures.name(capture_group)?.as_str().to_string(), + }; + parse_metric_primitive(&raw, *kind) +} + +/// One-time full scan building a sort helper index for `sort_by` from every currently-ingested log — +/// used both the first time a param is sorted by (no helper index exists yet) and to transparently +/// rebuild one ticket 061's sweep purged. +fn build_sort_helper_index(index: &BTreeMap, sort_by: &str) -> BTreeMap { + let compiled_params = compiled_param_declarations(index); + logs_range(index) + .map(|(sequence, timestamp, text)| { + let key = match derive_single_param(text, &compiled_params, sort_by) { + Some(value) => SortHelperKey::WithParam(value, sequence), + None => SortHelperKey::WithoutParam(sequence), + }; + (key, timestamp) + }) + .collect() +} + +/// Incrementally extends every currently-live helper index with one freshly-ingested line, called +/// from `EngineMessage::Ingest` right alongside the primary index insert (ticket 060) — this is what +/// lets a sorted query avoid re-deriving params for the whole timeline on every call once a helper +/// index for that sort key exists. A no-op when no helper index has been built yet (the common case +/// for a param nobody has sorted by), so ingest pays nothing for sort keys that are never queried. +fn update_sort_helper_indexes_on_ingest( + sort_helper_indexes: &mut HashMap, + index: &BTreeMap, + sequence: u64, + timestamp: Timestamp, + text: &str, +) { + if sort_helper_indexes.is_empty() { + return; + } + let compiled_params = compiled_param_declarations(index); + for (sort_by, entry) in sort_helper_indexes.iter_mut() { + let key = match derive_single_param(text, &compiled_params, sort_by) { + Some(value) => SortHelperKey::WithParam(value, sequence), + None => SortHelperKey::WithoutParam(sequence), + }; + entry.index.insert(key, timestamp); + } +} + +/// Ticket 061: drops any helper index whose `last_accessed` exceeds `SORT_HELPER_INDEX_PURGE_AFTER`. +/// Cheap to call opportunistically on every ingest (`HashMap::retain` is a no-op scan when nothing's +/// stale) rather than needing a dedicated timer/task — a purged index rebuilds transparently via +/// `build_sort_helper_index` the next time something sorts by it. +fn purge_stale_sort_helper_indexes(sort_helper_indexes: &mut HashMap) { + sort_helper_indexes.retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER); +} + fn compiled_param_declarations( index: &BTreeMap, ) -> Vec<(String, Regex, String, Option, MetricPrimitiveKind)> { @@ -1409,6 +1550,7 @@ fn handle_query( index: &BTreeMap, query: EngineQuery, approx_bytes_total: u64, + sort_helper_indexes: &mut HashMap, ) -> EngineResponse { match query { EngineQuery::Logs(params) => { @@ -1453,33 +1595,63 @@ fn handle_query( params: derive_log_params(text, &compiled_params), }) .collect(), - // An alternate sort key needs every matching line's params derived up front (there's - // no way to know which lines will end up in the first `limit` before sorting), then - // reordered before the limit is finally applied. + // Ticket 060: walk the (lazily built-or-reused) helper index for this sort key in + // key order instead of materializing and sorting the full filtered set — the index's + // own key ordering already expresses "sorted-by-value lines, then lines missing the + // param ordered by ottyl_seq" (`sort_missing_fallback`'s semantics), so this just + // filters/limits as it walks rather than partition-sort-concat afterward. Some(sort_by) => { - let mut all: Vec = filtered - .map(|(sequence, timestamp, text)| LogEntryDto { + let helper = sort_helper_indexes.entry(sort_by.to_string()).or_insert_with(|| { + SortHelperIndexEntry { + index: build_sort_helper_index(index, sort_by), + last_accessed: std::time::Instant::now(), + } + }); + helper.last_accessed = std::time::Instant::now(); + + let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true); + let mut entries = Vec::new(); + for (helper_key, timestamp) in helper.index.iter() { + let sequence = match helper_key { + SortHelperKey::WithParam(_, sequence) => *sequence, + SortHelperKey::WithoutParam(sequence) => { + if !sort_missing_fallback { + continue; + } + *sequence + } + }; + if !(params.since_sequence.is_none_or(|since| sequence > since) + && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start) + && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)) + { + continue; + } + // A line evicted (by `max_logs`) since this key was recorded is simply + // skipped, same as buckets/other derived data aren't retroactively pruned + // on eviction elsewhere in this file. + let Some(DataValue::LogLiteral { text }) = + index.get(&IndexKey::IngestedLog { sequence, timestamp: *timestamp }) + else { + continue; + }; + if !compiled_filter + .as_ref() + .is_none_or(|filter| eval_compiled(filter, sequence, *timestamp, text)) + { + continue; + } + entries.push(LogEntryDto { sequence, timestamp_ms: timestamp.as_millis(), - text: text.to_string(), + text: text.clone(), params: derive_log_params(text, &compiled_params), - }) - .collect(); - let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true); - if sort_missing_fallback { - let (mut with_param, mut without_param): (Vec<_>, Vec<_>) = - all.drain(..).partition(|entry| entry.params.contains_key(sort_by)); - with_param.sort_by(|a, b| a.params[sort_by].cmp(&b.params[sort_by])); - // Lines missing the sort param are only ordered relative to each other (by - // `ottyl_seq`) and appended after the sorted lines, rather than interleaved - // into the overall ordering — see ticket 045. - without_param.sort_by_key(|entry| entry.sequence); - with_param.into_iter().chain(without_param).collect() - } else { - all.retain(|entry| entry.params.contains_key(sort_by)); - all.sort_by(|a, b| a.params[sort_by].cmp(&b.params[sort_by])); - all + }); + if entries.len() >= limit { + break; + } } + entries } }; let entries: Vec = entries.into_iter().take(limit).collect(); @@ -2106,8 +2278,9 @@ mod tests { /// Runs both auto-declaration passes over every fixture line, same order as `EngineMessage::Ingest`. fn ingest_log_stream_fixture(index: &mut BTreeMap) { + let mut header = None; for line in LOG_STREAM_FIXTURE.lines() { - auto_declare_log_stream_params(index, line); + auto_declare_log_stream_params(index, &mut header, line); auto_declare_params(index, line); } } @@ -2115,13 +2288,13 @@ mod tests { #[test] fn ingesting_log_stream_fixture_does_not_blow_up_declared_param_count() { // Ticket 050: unbounded auto-declaration on this real `log stream` dump produced 220 distinct - // params. The bound (MAX_AUTO_DECLARED_PARAMS + the 3 fixed log_stream params) must keep the - // total in the tens, nowhere near "hundreds". + // params. The bound (MAX_AUTO_DECLARED_PARAMS + the header-derived log_stream columns) must + // keep the total in the tens, nowhere near "hundreds". let mut index = BTreeMap::new(); ingest_log_stream_fixture(&mut index); let declared = param_declarations_range(&index).count(); - // `log_type`/`process`/`pid` are stored as `auto_derived` too, so the shared + // The header-derived columns (ticket 059) are stored as `auto_derived` too, so the shared // `MAX_AUTO_DECLARED_PARAMS` ceiling covers all of them together (see // `auto_declare_log_stream_params`'s doc comment). assert!( @@ -2144,29 +2317,38 @@ mod tests { } #[test] - fn log_stream_shaped_lines_expose_type_process_and_pid_as_queryable_params() { + fn log_stream_shaped_lines_expose_header_named_columns_as_queryable_params() { let mut index = BTreeMap::new(); ingest_log_stream_fixture(&mut index); let declared: HashSet<&str> = param_declarations_range(&index).map(|d| d.0).collect(); - assert!(declared.contains("log_type")); - assert!(declared.contains("process")); - assert!(declared.contains("pid")); + // Derived generically from the fixture's own header row (ticket 059), not a fixed set of three. + for name in ["Timestamp", "Thread", "Type", "Activity", "PID", "TTL", "message"] { + assert!(declared.contains(name), "expected {name} to be declared, got {declared:?}"); + } let compiled = compiled_param_declarations(&index); let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap(); let params = derive_log_params(data_line, &compiled); - assert_eq!(params.get("log_type"), Some(&MetricPrimitive::String("Default".to_string()))); - assert_eq!(params.get("process"), Some(&MetricPrimitive::String("kitty".to_string()))); - assert_eq!(params.get("pid"), Some(&MetricPrimitive::Number(Number::Int(1343)))); + assert_eq!(params.get("Type"), Some(&MetricPrimitive::String("Default".to_string()))); + assert_eq!(params.get("PID"), Some(&MetricPrimitive::Number(Number::Int(1343)))); + assert_eq!(params.get("TTL"), Some(&MetricPrimitive::Number(Number::Int(7)))); + assert_eq!( + params.get("message"), + Some(&MetricPrimitive::String( + "kitty: (SkyLight) [com.apple.SkyLight:default] SLSGetDisplaysWithOpenGLDisplayMask, 3614: Display 0x00000001: Mode 54, (0 x 0) [1728 x 1117], Depth 32 BPP, active".to_string() + )) + ); } #[test] - fn non_log_stream_lines_do_not_get_log_type_process_or_pid_declared() { + fn non_log_stream_lines_do_not_get_any_columns_declared_without_a_cached_header() { let mut index = BTreeMap::new(); - auto_declare_log_stream_params(&mut index, r#"ts=1788648105773 msg="entity_moved" x1=-4"#); + let mut header = None; + auto_declare_log_stream_params(&mut index, &mut header, r#"ts=1788648105773 msg="entity_moved" x1=-4"#); assert!(param_declarations_range(&index).next().is_none()); + assert!(header.is_none()); } #[test] @@ -2176,8 +2358,9 @@ mod tests { let mut index = BTreeMap::new(); let start = std::time::Instant::now(); for _ in 0..repeats { + let mut header = None; for line in LOG_STREAM_FIXTURE.lines() { - auto_declare_log_stream_params(&mut index, line); + auto_declare_log_stream_params(&mut index, &mut header, line); auto_declare_params(&mut index, line); } } @@ -3578,6 +3761,7 @@ mod tests { end_ms: None, }, 0, + &mut HashMap::new(), ); match response { EngineResponse::Promql(Err(message)) => assert!(!message.is_empty()), @@ -3599,6 +3783,7 @@ mod tests { end_ms: None, }, 0, + &mut HashMap::new(), ); match response { EngineResponse::Promql(Ok(series)) => { @@ -3691,6 +3876,7 @@ mod tests { &index, EngineQuery::Promql { query: "count(ottyl_seq)".to_string(), start_ms: None, end_ms: None }, 0, + &mut HashMap::new(), ); match response { EngineResponse::Promql(Ok(series)) => { @@ -3753,7 +3939,7 @@ mod tests { ingest_log(&mut index, 2, "still fine"); ingest_log(&mut index, 3, "another error here"); - match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0) { + match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 3]); @@ -3769,7 +3955,7 @@ mod tests { ingest_log(&mut index, i, &format!("line {i}")); } - match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0) { + match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 1]); @@ -3789,7 +3975,7 @@ mod tests { ); } - match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0) { + match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![3, 4]); @@ -3816,7 +4002,7 @@ mod tests { ingest_log(&mut index, 1, "priority=1 boom"); ingest_log(&mut index, 2, "priority=2 boom"); - match handle_query(&index, logs_query_sorted("priority", None), 0) { + match handle_query(&index, logs_query_sorted("priority", None), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 2, 0]); @@ -3844,7 +4030,7 @@ mod tests { ingest_log(&mut index, 2, "priority=1 boom"); // has param ingest_log(&mut index, 3, "also missing"); // missing - match handle_query(&index, logs_query_sorted("priority", Some(true)), 0) { + match handle_query(&index, logs_query_sorted("priority", Some(true)), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); // Lines with the param sorted ascending by value (2 then 0), followed by the @@ -3870,7 +4056,7 @@ mod tests { ingest_log(&mut index, 2, "priority=1 boom"); ingest_log(&mut index, 3, "also missing"); - match handle_query(&index, logs_query_sorted("priority", Some(false)), 0) { + match handle_query(&index, logs_query_sorted("priority", Some(false)), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![2, 0]); @@ -3884,7 +4070,7 @@ mod tests { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "hello"); - match handle_query(&index, logs_query_sorted("does_not_exist", None), 0) { + match handle_query(&index, logs_query_sorted("does_not_exist", None), 0, &mut HashMap::new()) { EngineResponse::Error(err) => assert!(err.contains("does_not_exist")), _ => panic!("expected EngineResponse::Error"), } @@ -3908,7 +4094,7 @@ mod tests { ingest_log(&mut index, 2, "priority=1 boom"); ingest_log(&mut index, 3, "also missing"); - match handle_query(&index, logs_query(None), 0) { + match handle_query(&index, logs_query(None), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 1, 2, 3]); @@ -3917,6 +4103,136 @@ mod tests { } } + // -- Ticket 060: sort helper index -- + + #[test] + fn sort_helper_index_is_built_lazily_and_reused_across_queries() { + let mut index = BTreeMap::new(); + declare_test_param( + &mut index, + "priority", + r"priority=(?P\d+)", + "priority", + MetricPrimitiveKind::Number, + ); + ingest_log(&mut index, 0, "priority=3 boom"); + ingest_log(&mut index, 1, "priority=1 boom"); + ingest_log(&mut index, 2, "priority=2 boom"); + + let mut sort_helper_indexes = HashMap::new(); + assert!(!sort_helper_indexes.contains_key("priority")); + + // First query has no helper index yet — must lazily build one rather than erroring or + // silently returning unsorted results. + match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) { + EngineResponse::Logs(entries) => { + let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); + assert_eq!(sequences, vec![1, 2, 0]); + } + _ => panic!("expected Logs response"), + } + assert!(sort_helper_indexes.contains_key("priority")); + assert_eq!(sort_helper_indexes["priority"].index.len(), 3); + + // Ingest more lines directly into the already-built helper index (bypassing the full + // `EngineMessage::Ingest` actor loop, same as every other test in this file exercises + // `handle_query` directly) via the same incremental-update path `run()` calls on ingest, + // then requery — the new lines must appear correctly ordered without the helper index + // having been thrown away and rebuilt from scratch. + ingest_log(&mut index, 3, "priority=0 boom"); + update_sort_helper_indexes_on_ingest( + &mut sort_helper_indexes, + &index, + 3, + Timestamp::from_millis(3), + "priority=0 boom", + ); + ingest_log(&mut index, 4, "no priority here"); + update_sort_helper_indexes_on_ingest( + &mut sort_helper_indexes, + &index, + 4, + Timestamp::from_millis(4), + "no priority here", + ); + assert_eq!(sort_helper_indexes["priority"].index.len(), 5); + + match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) { + EngineResponse::Logs(entries) => { + let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); + // 3 (priority=0) sorts before the original three; 4 (missing the param) falls back + // to ottyl_seq order, appended after every line that has the param. + assert_eq!(sequences, vec![3, 1, 2, 0, 4]); + } + _ => panic!("expected Logs response"), + } + } + + #[test] + fn sort_helper_index_incremental_update_is_a_noop_when_no_index_is_live() { + // Guards `update_sort_helper_indexes_on_ingest`'s early return: ingest must not pay to + // derive a param nobody has ever sorted by. + let mut sort_helper_indexes: HashMap = HashMap::new(); + let index = BTreeMap::new(); + update_sort_helper_indexes_on_ingest(&mut sort_helper_indexes, &index, 0, Timestamp::from_millis(0), "hello"); + assert!(sort_helper_indexes.is_empty()); + } + + // -- Ticket 061: purging stale sort helper indexes -- + + #[test] + fn purge_stale_sort_helper_indexes_drops_only_entries_past_the_purge_duration() { + let mut sort_helper_indexes: HashMap = HashMap::new(); + sort_helper_indexes.insert( + "stale".to_string(), + SortHelperIndexEntry { + index: BTreeMap::new(), + last_accessed: std::time::Instant::now() - SORT_HELPER_INDEX_PURGE_AFTER + - std::time::Duration::from_secs(1), + }, + ); + sort_helper_indexes.insert( + "fresh".to_string(), + SortHelperIndexEntry { index: BTreeMap::new(), last_accessed: std::time::Instant::now() }, + ); + + purge_stale_sort_helper_indexes(&mut sort_helper_indexes); + + assert!(!sort_helper_indexes.contains_key("stale")); + assert!(sort_helper_indexes.contains_key("fresh")); + } + + #[test] + fn sorted_query_transparently_rebuilds_a_purged_helper_index() { + let mut index = BTreeMap::new(); + declare_test_param( + &mut index, + "priority", + r"priority=(?P\d+)", + "priority", + MetricPrimitiveKind::Number, + ); + ingest_log(&mut index, 0, "priority=3 boom"); + ingest_log(&mut index, 1, "priority=1 boom"); + ingest_log(&mut index, 2, "priority=2 boom"); + + let mut sort_helper_indexes = HashMap::new(); + let _ = handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes); + assert!(sort_helper_indexes.contains_key("priority")); + + // Simulate ticket 061's sweep having purged this entry for staleness. + sort_helper_indexes.remove("priority"); + + match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) { + EngineResponse::Logs(entries) => { + let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); + assert_eq!(sequences, vec![1, 2, 0]); + } + _ => panic!("expected Logs response"), + } + assert!(sort_helper_indexes.contains_key("priority")); + } + #[test] fn logs_query_with_ottyl_fmt_comparison_filters_by_detected_format_for_mixed_format_lines() { let mut index = BTreeMap::new(); @@ -3926,7 +4242,7 @@ mod tests { ingest_log(&mut index, 2, "just some plain prose here"); ingest_log(&mut index, 3, r#"{"level":"info","code":200}"#); - match handle_query(&index, logs_query(Some("ottyl_fmt==json")), 0) { + match handle_query(&index, logs_query(Some("ottyl_fmt==json")), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 3]); @@ -3934,7 +4250,7 @@ mod tests { _ => panic!("expected Logs response"), } - match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0) { + match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1]); @@ -3942,7 +4258,7 @@ mod tests { _ => panic!("expected Logs response"), } - match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0) { + match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![2]); @@ -3958,7 +4274,7 @@ mod tests { ingest_log(&mut index, 1, r#"{"level":"info","code":200}"#); ingest_log(&mut index, 2, "not json at all"); - match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0) { + match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); @@ -3966,7 +4282,7 @@ mod tests { _ => panic!("expected Logs response"), } - match handle_query(&index, logs_query(Some("json[code]>=300")), 0) { + match handle_query(&index, logs_query(Some("json[code]>=300")), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); @@ -3991,6 +4307,7 @@ mod tests { &index, logs_query(Some(r#"regex("boom") and requests[level]==error"#)), 0, + &mut HashMap::new(), ) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); @@ -4004,6 +4321,7 @@ mod tests { &index, logs_query(Some(r#"regex("boom") or requests[level]==error"#)), 0, + &mut HashMap::new(), ) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); @@ -4020,7 +4338,7 @@ mod tests { ingest_log(&mut index, 0, "level=info hello"); ingest_log(&mut index, 1, "no match here"); - let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else { + let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, &mut HashMap::new()) else { panic!("expected Logs response"); }; @@ -4042,7 +4360,7 @@ mod tests { ingest_log(&mut index, 0, "an ERROR occurred"); ingest_log(&mut index, 1, "everything is fine"); - let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else { + let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, &mut HashMap::new()) else { panic!("expected Logs response"); }; @@ -4058,7 +4376,7 @@ mod tests { ingest_log(&mut index, 1, "an error occurred"); ingest_log(&mut index, 2, "another error here"); - match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0) { + match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0, &mut HashMap::new()) { EngineResponse::Timeline(buckets) => { let total: u64 = buckets.iter().map(|b| b.count).sum(); assert_eq!(total, 2); @@ -4072,12 +4390,12 @@ mod tests { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "hello"); - match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0) { + match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0, &mut HashMap::new()) { EngineResponse::Error(message) => assert!(!message.is_empty()), _ => panic!("expected an Error response for a malformed filter string"), } - match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0) { + match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0, &mut HashMap::new()) { EngineResponse::Error(message) => assert!(!message.is_empty()), _ => panic!("expected an Error response for a malformed filter string"), } @@ -4091,7 +4409,7 @@ mod tests { declare_metric(&mut index, metric_request("requests", pattern, vec!["level"])).unwrap(); ingest_log(&mut index, 0, "level=error"); - match handle_query(&index, logs_query(Some("requests==5")), 0) { + match handle_query(&index, logs_query(Some("requests==5")), 0, &mut HashMap::new()) { EngineResponse::Error(message) => assert!(message.contains("param")), _ => panic!("expected an Error response for a metric comparison without [param]"), } @@ -4106,7 +4424,7 @@ mod tests { declare_metric(&mut index, metric_request("requests", pattern, vec!["level", "duration"])).unwrap(); ingest_log(&mut index, 0, "level=error duration=5"); - match handle_query(&index, logs_query(Some("requests[nonexistent_param]==\"x\"")), 0) { + match handle_query(&index, logs_query(Some("requests[nonexistent_param]==\"x\"")), 0, &mut HashMap::new()) { EngineResponse::Error(message) => { assert!( message.contains("nonexistent_param"), @@ -4136,7 +4454,7 @@ mod tests { ingest_log(&mut index, 0, "level=error duration=5"); ingest_log(&mut index, 1, "level=info duration=2"); - match handle_query(&index, logs_query(Some("requests[level]==error")), 0) { + match handle_query(&index, logs_query(Some("requests[level]==error")), 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); diff --git a/src/logfmt.rs b/src/logfmt.rs index fe44d68..4f24676 100644 --- a/src/logfmt.rs +++ b/src/logfmt.rs @@ -101,16 +101,223 @@ pub fn is_log_stream_line(line: &str) -> bool { LOG_STREAM_RE.is_match(line) } -/// `message_match_regex` patterns `crate::engine::auto_declare_log_stream_params` uses to declare -/// `log_type`/`process`/`pid` as standalone params the first time a `log stream`-shaped line is seen. -/// Each isolates exactly one named capture group (matching `derive_log_params`'s one-group-per-param -/// convention) over the same column layout `LOG_STREAM_RE` detects. -pub const LOG_STREAM_TYPE_PATTERN: &str = - r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+(?P\S+)\s+0x[0-9a-fA-F]+\s+\d+\s+\d+\s+[^\s:]+:\s"; -pub const LOG_STREAM_PROCESS_PATTERN: &str = - r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+\S+\s+0x[0-9a-fA-F]+\s+\d+\s+\d+\s+(?P[^\s:]+):\s"; -pub const LOG_STREAM_PID_PATTERN: &str = - r"^\d{4}-\d{2}-\d{2}\s+\S+\s+0x[0-9a-fA-F]+\s+\S+\s+0x[0-9a-fA-F]+\s+(?P\d+)\s+\d+\s+[^\s:]+:\s"; +/// Regex fragment for one column separator under the generic whitespace-columnar rule ticket 059 +/// asks for: a run of 2+ whitespace chars, or a lone tab, separates columns; a single plain space does +/// not (so a value that legitimately contains one -- e.g. `log stream`'s `date time+tz` timestamp -- +/// stays intact within its own column). `is_separator_run` is the same rule applied at split time; +/// this is that rule re-expressed as a regex fragment for the trailing message field's +/// `message_match_regex`, which re-derives its value the same way whenever a caller queries by it +/// later (declarations don't cache extracted values -- see `crate::engine::extract_param_raw`). +const COLUMN_SEPARATOR_PATTERN: &str = r"(?:\t|[ \t]{2,})"; + +/// A column's value: one or more non-whitespace tokens joined by single spaces, with no upper bound +/// on length. Used for the trailing (unbounded-width) message field and the last named column, where +/// there's no next column's start position to bound against. +const COLUMN_VALUE_PATTERN: &str = r"\S+(?: \S+)*"; + +/// Same idea as `COLUMN_VALUE_PATTERN`, but bounded to a known column `width` (in chars) -- needed for +/// every column except the last, since two adjacent `log stream` values can legitimately be separated +/// by only a single space when the first one nearly fills its column's on-screen width (e.g. the +/// `Timestamp` column: its own `date time+tz` value already contains one internal space, so a +/// content-only pattern can't tell that boundary apart from the one-space gap before the next column +/// starts -- seed this from `LogStreamColumn::start` positions instead). Greedily matches up to +/// `width` chars while requiring the match to end on a non-whitespace char, which is what excludes the +/// column's own trailing padding spaces from the capture. +fn bounded_column_value_pattern(width: usize) -> String { + if width <= 1 { + r"\S".to_string() + } else { + format!(r"\S(?:.{{0,{}}}\S)?", width - 2) + } +} + +fn is_separator_run(run: &str) -> bool { + run.len() >= 2 || run.contains('\t') +} + +static WHITESPACE_RUN: LazyLock = LazyLock::new(|| Regex::new(r"[ \t]+").expect("static regex must compile")); + +/// Splits `line` at whitespace runs that qualify as a column separator (`is_separator_run`), returning +/// each piece alongside its byte offset in `line`; a lone plain space never splits. Consumes at most +/// `max_columns.map(|n| n - 1)` separators -- the final piece returned is the untouched remainder of +/// the line, including any further whitespace runs inside it. That's what lets a trailing unlabelled +/// `log stream` message tail (whose own internal spacing can be arbitrary) come back as one field +/// instead of being chopped up by it. +fn split_columns_with_offsets(line: &str, max_columns: Option) -> Vec<(usize, &str)> { + let mut columns = Vec::new(); + let mut start = 0usize; + let mut consumed = 0usize; + for m in WHITESPACE_RUN.find_iter(line) { + if max_columns.is_some_and(|max| consumed + 1 >= max) { + break; + } + let run = &line[m.start()..m.end()]; + if is_separator_run(run) { + columns.push((start, &line[start..m.start()])); + start = m.end(); + consumed += 1; + } + } + columns.push((start, &line[start..])); + columns +} + +fn split_columns(line: &str, max_columns: Option) -> Vec<&str> { + split_columns_with_offsets(line, max_columns).into_iter().map(|(_, s)| s).collect() +} + +/// A column-name identifier a generated `message_match_regex`'s named capture group can use as-is +/// (the regex crate requires `[A-Za-z_][0-9A-Za-z_]*`). +static COLUMN_NAME: LazyLock = + LazyLock::new(|| Regex::new(r"^[A-Za-z][A-Za-z0-9_]*$").expect("static regex must compile")); + +/// One column named by a detected `log stream` header (ticket 059). `start` is that column's position +/// in `char`s (not bytes -- see `char_slice`) within the header line -- meaningful only when +/// `LogStreamHeader::tab_delimited` is `false` (see that field's doc comment). +pub struct LogStreamColumn { + pub name: String, + start: usize, +} + +/// A detected `log stream` header: its named columns, plus whether they were tab- or space-separated. +/// `log stream`'s real output right-pads each header name to a fixed on-screen width using runs of +/// 2+ spaces, and every data line's values land in that same fixed alignment regardless of how much +/// of that width they individually fill -- so for that shape, a data line's column boundaries are +/// best found at the header's own fixed character positions, not by re-splitting the data line's own +/// whitespace (see `split_log_stream_row`'s doc comment for why that alone isn't reliable there). +/// A tab-separated header carries no such alignment guarantee (values aren't padded to a shared +/// width), but a lone tab is itself an unambiguous separator, so that case instead re-splits each data +/// line directly. +pub struct LogStreamHeader { + pub columns: Vec, + tab_delimited: bool, +} + +/// Byte-indexed `str` slicing doesn't work directly with `LogStreamColumn::start` (a *char* offset), +/// since `log stream` output is expected to stay ASCII in practice but callers shouldn't panic if a +/// message contains multi-byte chars before the slice point. `end` of `None` slices to end of string. +fn char_slice(s: &str, start: usize, end: Option) -> &str { + if end.is_some_and(|end| end <= start) { + return ""; + } + let mut indices = s.char_indices(); + let byte_start = indices.nth(start).map(|(i, _)| i).unwrap_or(s.len()); + let byte_end = match end { + Some(end) => s.char_indices().nth(end).map(|(i, _)| i).unwrap_or(s.len()), + None => s.len(), + }; + &s[byte_start..byte_end.max(byte_start)] +} + +/// Recognizes a `log stream`-style header row generically (ticket 059), instead of assuming the fixed +/// `Timestamp Thread Type Activity PID TTL` names: a short run of bareword column names separated by +/// qualifying whitespace runs (`split_columns_with_offsets`) that is, critically, *not itself* a +/// `log stream` data line (`is_log_stream_line`) -- so a header is never mistaken for the data line +/// immediately following it. `crate::engine::auto_declare_log_stream_params` caches the result across +/// ingested lines, since `log stream` prints its header once, before any data line, and this only ever +/// sees one line at a time. +pub fn detect_log_stream_header(line: &str) -> Option { + if is_log_stream_line(line) { + return None; + } + let mut columns = Vec::new(); + for (byte_start, slice) in split_columns_with_offsets(line, None) { + let trimmed = slice.trim(); + if trimmed.is_empty() { + continue; + } + let leading_ws = slice.len() - slice.trim_start().len(); + let start = line[..byte_start + leading_ws].chars().count(); + columns.push(LogStreamColumn { name: trimmed.to_string(), start }); + } + if columns.len() < 2 || !columns.iter().all(|c| COLUMN_NAME.is_match(&c.name)) { + return None; + } + let tab_delimited = WHITESPACE_RUN.find_iter(line).any(|m| m.as_str().contains('\t')); + Some(LogStreamHeader { columns, tab_delimited }) +} + +/// Splits a `log stream` data line into `header`'s named columns plus, when present, one trailing +/// unlabelled tail -- `crate::engine::auto_declare_log_stream_params` maps the first +/// `header.columns.len()` pieces onto the header's names and the optional last piece onto the +/// `message` param. +/// +/// For a space-padded fixed-width header, every column but the last is sliced at the header's own +/// fixed character positions (`LogStreamColumn::start`) rather than by re-splitting the data line's +/// own whitespace, since two adjacent values can legitimately be joined by only a single space when +/// the first nearly fills its column's width (e.g. `log stream`'s `Timestamp` column, whose value +/// already contains one internal space). Only the last named column's end is whitespace-run-detected +/// (`split_columns`), since that's what distinguishes its own value from the free-text message tail +/// `log stream` appends after it. +/// +/// For a tab-delimited header, there's no shared on-screen width to align against, but a lone tab is +/// itself unambiguous, so this instead re-splits the data line directly on qualifying separators. +pub fn split_log_stream_row<'a>(line: &'a str, header: &LogStreamHeader) -> Vec<&'a str> { + if header.tab_delimited { + return split_columns(line, Some(header.columns.len() + 1)); + } + let mut result = Vec::with_capacity(header.columns.len() + 1); + for (i, column) in header.columns.iter().enumerate() { + match header.columns.get(i + 1) { + Some(next) => result.push(char_slice(line, column.start, Some(next.start)).trim()), + None => { + let tail = char_slice(line, column.start, None); + let pieces = split_columns(tail, Some(2)); + result.push(pieces[0].trim()); + if let Some(message) = pieces.get(1) { + result.push(message); + } + } + } + } + result +} + +/// Builds the `message_match_regex` for the `index`-th (0-based) of `header`'s named columns. +/// +/// Fixed-width case: skips exactly `header.columns[index].start` chars, then captures this column's +/// own value bounded to its known width (`bounded_column_value_pattern`) -- unbounded for the last +/// column, which has no next column to bound against. +/// +/// Tab-delimited case: skips `index` preceding columns' own value+separator pairs instead, since +/// there's no shared fixed width to anchor a character-position skip against. +/// +/// Anchored at the start of the line since a `log stream` line never carries this fixed-per-run +/// column prefix anywhere but the very start. +pub fn log_stream_column_pattern(header: &LogStreamHeader, index: usize, capture_group: &str) -> String { + if header.tab_delimited { + let mut pattern = String::from("^"); + for _ in 0..index { + pattern.push_str(COLUMN_VALUE_PATTERN); + pattern.push_str(COLUMN_SEPARATOR_PATTERN); + } + pattern.push_str(&format!("(?P<{capture_group}>{COLUMN_VALUE_PATTERN})")); + return pattern; + } + let start = header.columns[index].start; + let value_pattern = match header.columns.get(index + 1) { + Some(next) => bounded_column_value_pattern(next.start.saturating_sub(start)), + None => COLUMN_VALUE_PATTERN.to_string(), + }; + format!("^.{{{start}}}(?P<{capture_group}>{value_pattern})") +} + +/// Builds the `message_match_regex` for the trailing unlabelled tail after `header`'s last named +/// column: everything from the end of that column's own separator to end of line, verbatim -- its own +/// internal whitespace, however irregular, is never treated as a further separator. +pub fn log_stream_message_pattern(header: &LogStreamHeader, capture_group: &str) -> String { + if header.tab_delimited { + let mut pattern = String::from("^"); + for _ in 0..header.columns.len() { + pattern.push_str(COLUMN_VALUE_PATTERN); + pattern.push_str(COLUMN_SEPARATOR_PATTERN); + } + pattern.push_str(&format!("(?P<{capture_group}>.*)$")); + return pattern; + } + let start = header.columns.last().map(|c| c.start).unwrap_or(0); + format!("^.{{{start}}}{COLUMN_VALUE_PATTERN}{COLUMN_SEPARATOR_PATTERN}(?P<{capture_group}>.*)$") +} /// Go `log/slog` `NewJSONHandler` output (the `go-log` label): a single-line JSON object whose first /// three fields are always `time`, `level`, `msg` in that fixed order (slog's `JSONHandler` writes the @@ -290,15 +497,90 @@ mod tests { assert!(!is_log_stream_line(header)); } + fn column_names(header: &LogStreamHeader) -> Vec<&str> { + header.columns.iter().map(|c| c.name.as_str()).collect() + } + #[test] - fn log_stream_patterns_capture_type_process_and_pid() { + fn detects_log_stream_header_columns_generically() { + let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); + let header = detect_log_stream_header(header_line).expect("fixture header must be detected"); + assert_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type", "Activity", "PID", "TTL"]); + assert!(!header.tab_delimited); + } + + #[test] + fn a_log_stream_data_line_is_never_detected_as_its_own_header() { + let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap(); + assert!(detect_log_stream_header(data_line).is_none()); + } + + #[test] + fn plain_prose_is_not_detected_as_a_log_stream_header() { + assert!(detect_log_stream_header("just some plain prose here").is_none()); + assert!(detect_log_stream_header("ts=1788648105773 msg=\"entity_moved\"").is_none()); + } + + #[test] + fn splits_log_stream_row_into_named_columns_plus_trailing_message() { + let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); + let header = detect_log_stream_header(header_line).unwrap(); + let data_line = "2026-09-06 01:48:54.518117+0100 0x4dfed2 Default 0x0 606 0 WindowServer: (HSTouchHIDService) something happened"; + let row = split_log_stream_row(data_line, &header); + assert_eq!(row.len(), header.columns.len() + 1); + assert_eq!(row[0], "2026-09-06 01:48:54.518117+0100"); + assert_eq!(row[1], "0x4dfed2"); + assert_eq!(row[2], "Default"); + assert_eq!(row[3], "0x0"); + assert_eq!(row[4], "606"); + assert_eq!(row[5], "0"); + // The message tail is never chopped up by its own internal single/double spacing. + assert_eq!(row[6], "WindowServer: (HSTouchHIDService) something happened"); + } + + #[test] + fn a_single_tab_separates_columns_exactly_like_a_run_of_spaces() { + let header = detect_log_stream_header("Timestamp\tThread\tType").unwrap(); + assert_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type"]); + assert!(header.tab_delimited); + + // Two tabs in a row still qualify as a single separator (like a run of 2+ spaces), so this + // still cleanly separates Timestamp/Thread/Type despite the doubled tab before "Default". + let spaced = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\t\tDefault trailing message", &header); + let tabbed = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\tDefault trailing message", &header); + assert_eq!(spaced[0], "2026-09-06 01:48:54"); + assert_eq!(spaced[1], "0x4dfed2"); + assert_eq!(spaced[2], "Default trailing message"); + assert_eq!(tabbed[0], "2026-09-06 01:48:54"); + assert_eq!(tabbed[1], "0x4dfed2"); + assert_eq!(tabbed[2], "Default trailing message"); + } + + #[test] + fn column_and_message_patterns_capture_the_same_fields_split_log_stream_row_finds() { + let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); + let header = detect_log_stream_header(header_line).unwrap(); let line = "2026-09-06 01:48:54.518117+0100 0x4dfed2 Default 0x0 606 0 WindowServer: (HSTouchHIDService) something happened"; - let type_re = Regex::new(LOG_STREAM_TYPE_PATTERN).unwrap(); - let process_re = Regex::new(LOG_STREAM_PROCESS_PATTERN).unwrap(); - let pid_re = Regex::new(LOG_STREAM_PID_PATTERN).unwrap(); - assert_eq!(type_re.captures(line).unwrap().name("log_type").unwrap().as_str(), "Default"); - assert_eq!(process_re.captures(line).unwrap().name("process").unwrap().as_str(), "WindowServer"); - assert_eq!(pid_re.captures(line).unwrap().name("pid").unwrap().as_str(), "606"); + + let type_re = Regex::new(&log_stream_column_pattern(&header, 2, "Type")).unwrap(); + assert_eq!(type_re.captures(line).unwrap().name("Type").unwrap().as_str(), "Default"); + + let pid_re = Regex::new(&log_stream_column_pattern(&header, 4, "PID")).unwrap(); + assert_eq!(pid_re.captures(line).unwrap().name("PID").unwrap().as_str(), "606"); + + let message_re = Regex::new(&log_stream_message_pattern(&header, "message")).unwrap(); + assert_eq!( + message_re.captures(line).unwrap().name("message").unwrap().as_str(), + "WindowServer: (HSTouchHIDService) something happened" + ); + + // Also exercise the Timestamp column specifically -- its own value contains one internal + // space, so a naive content-only pattern would previously over-consume into the next column. + let timestamp_re = Regex::new(&log_stream_column_pattern(&header, 0, "Timestamp")).unwrap(); + assert_eq!( + timestamp_re.captures(line).unwrap().name("Timestamp").unwrap().as_str(), + "2026-09-06 01:48:54.518117+0100" + ); } #[test] diff --git a/tickets/059-log-stream-parsing-should-be-generic-whitespace-columnar.md b/tickets/059-log-stream-parsing-should-be-generic-whitespace-columnar.md deleted file mode 100644 index 9ec591e..0000000 --- a/tickets/059-log-stream-parsing-should-be-generic-whitespace-columnar.md +++ /dev/null @@ -1,62 +0,0 @@ -# 059 — `log stream` parsing is one hardcoded regex, not generic whitespace-columnar detection - -## Summary - -Current `log stream` support is a single hardcoded regex, -`LOG_STREAM_RE` (`src/logfmt.rs:93-113`), matching exactly six fixed -columns (`Timestamp Thread Type Activity PID TTL`) by content shape -(dates/hex/decimals) rather than by parsing the header row generically. -`auto_declare_log_stream_params` (`src/engine.rs:451-472`) then extracts -only three hardcoded fields (`log_type`, `process`, `pid`) via three more -bespoke regexes (`LOG_STREAM_TYPE_PATTERN` etc., `logfmt.rs:108-113`). - -The fixture `src/fixtures/log_stream_sample.txt` has a real header row -(`Timestamp Thread Type Activity PID TTL `) -whose columns are separated by runs of 2+ spaces — the general CSV-like -shape the wishlist asks for — plus a trailing unlabelled `process: -message` tail that must not get merged into the preceding column. Today -nothing derives column names or positions from that header; every field -is matched by a fixed, hand-tuned regex, so any deviation from this -exact `log stream` output shape silently fails to parse. - -## Scope - -- Replace `LOG_STREAM_RE`/`LOG_STREAM_*_PATTERN`'s fixed-shape matching - with a generic whitespace-columnar detector: given a header line, - split on runs of duplicated whitespace (2+ spaces, or a single tab, or - " \t " — any whitespace repeated) to determine column names and - positions, then apply the same column boundaries to subsequent data - lines. - - The header's last named column's value must still capture through - to end-of-line without swallowing the trailing unlabelled message - column that `log stream` appends after the last header-named field - (i.e. that trailing text becomes its own unlabelled/message param, - not merged into the last header column). - - Single tabs are acceptable separators even where surrounding spaces - aren't doubled. -- Wire the new column-driven extraction into - `auto_declare_log_stream_params` in place of the three bespoke - per-field regexes, so declared params are derived from whatever - columns the header actually names, not a fixed list of three. -- Use `src/fixtures/log_stream_sample.txt` as the reference fixture and - add/extend tests in `src/logfmt.rs`/`src/engine.rs` asserting header - columns are correctly detected and the trailing message field is kept - separate. -- Out of scope: changing param-cardinality bounding introduced in ticket - 050/054 (commit f48d3c6f) — this is purely about how columns are - detected/split, not about pruning noisy auto-declarations. - -## Acceptance criteria - -- Piping real `log stream` output (as in `log_stream_sample.txt`) - through ottyl declares params matching the header's actual column - names, not the previous fixed three. -- The trailing unlabelled message text on each `log stream` line is - captured as its own field and never merged into the last named - header column. -- A header/data line whose columns are separated by a single tab is - parsed identically to one separated by 2+ spaces. - -## Depends on - -None. diff --git a/tickets/060-timeline-custom-sort-needs-helper-btreemap-index.md b/tickets/060-timeline-custom-sort-needs-helper-btreemap-index.md deleted file mode 100644 index e57a5b8..0000000 --- a/tickets/060-timeline-custom-sort-needs-helper-btreemap-index.md +++ /dev/null @@ -1,57 +0,0 @@ -# 060 — Timeline sort-by-property does a full materialize+sort per query instead of using a helper index - -## Summary - -Sorting the timeline by an arbitrary property does work today (from -ticket 045): in the `EngineQuery::Logs` branch (`src/engine.rs:1408-1487`), -when `sort_by` is set, every filtered line is materialized into a -`Vec` (deriving params for all of them), then sorted -in-memory with `.sort_by(...)`, with `sort_missing_fallback` (default -true) partitioning entries that have the sort param from those that -don't and ordering the latter by `ottyl_seq` (`:1468-1482`). - -The problem is architectural: there is no incremental secondary index -keyed by the sort property, only the primary `BTreeMap` keyed by -`IndexKey::IngestedLog { sequence, timestamp }` (`src/index_key.rs`, -inserted at `engine.rs:156-181`). Every sorted query re-derives params -for and re-sorts the *entire* filtered range from scratch, and range -selection (e.g. "give me the next page in this order") has no way to -seek directly into a sorted position — it can only sort everything then -slice. - -## Scope - -- Introduce a "helper index": a secondary `BTreeMap` keyed by - `(sort-param-value, ottyl_seq)` (or equivalent composite key so - `ottyl_seq` is the tie-break/fallback ordering baked into the key - itself, not a post-hoc partition step) per distinct param name that's - been sorted by. -- Populate a helper index incrementally as lines are ingested (near - `engine.rs:156-181`, alongside the primary index insert), for any - param name that currently has a live helper index, rather than - rebuilding from scratch per query. -- Lines missing the sort param must still be captured in the helper - index's ordering, falling back to `ottyl_seq` ordering — matching - current `sort_missing_fallback` semantics but expressed as part of the - index's key/ordering rather than a separate partition-and-append step. -- Adapt the `EngineQuery::Logs` sorted branch (`engine.rs:1459-1483`) to - do a range walk over the relevant helper index instead of - materializing and sorting the full filtered set every time. -- Out of scope: the eviction/purge policy for helper indexes that - haven't been used recently — that's ticket 061. - -## Acceptance criteria - -- Querying the timeline sorted by a declared param returns results - ordered correctly, including lines missing that param falling back to - `ottyl_seq` order, matching current behavior. -- A sorted query no longer re-sorts the full filtered range from scratch - on every call once a helper index for that sort key exists — verified - by a test/benchmark showing incremental maintenance rather than - full materialize-then-sort. -- Existing sort-related tests (from ticket 045) continue to pass - unmodified in their assertions. - -## Depends on - -None. diff --git a/tickets/061-purge-unused-sort-helper-indexes.md b/tickets/061-purge-unused-sort-helper-indexes.md deleted file mode 100644 index 199507a..0000000 --- a/tickets/061-purge-unused-sort-helper-indexes.md +++ /dev/null @@ -1,43 +0,0 @@ -# 061 — Sort helper indexes need to be purged after a period of no access - -## Summary - -Ticket 060 introduces per-sort-key helper `BTreeMap` indexes. Left -unbounded, a user (or dashboard) that sorts by many distinct params over -a session accumulates a helper index per param forever, each duplicating -the full timeline's worth of keys. There's no existing time-since-access -eviction pattern in `engine.rs` to model this on: the only eviction today -is `evict_excess_logs` (`src/engine.rs:1356-1363`, called from ingest at -`:181` and from `SetMaxLogs` at `:212`), which is a hard count-cap -("drop oldest once over `max_logs`"), not an LRU/last-accessed scheme. - -## Scope - -- Track a last-accessed timestamp per helper index (e.g. - `HashMap, Instant)>` or equivalent), updated - whenever a query actually reads from that index. -- Add a sweep (triggered periodically, or opportunistically on - ingest/query — match whichever fits the existing engine actor's - message-loop structure) that drops any helper index whose - last-accessed time exceeds a fixed duration. -- A dropped helper index must be transparently rebuildable: the next - query sorted by that param should lazily rebuild the index (a one-time - full scan) exactly as if it had never existed, rather than erroring or - falling back permanently to unsorted behavior. -- Out of scope: making the purge duration user-configurable via - `Settings` — pick a reasonable fixed constant unless the ticket - implementer finds a strong reason to expose it. - -## Acceptance criteria - -- A helper index untouched for longer than the purge duration is - removed from engine memory. -- Querying a sort key whose helper index was purged transparently - rebuilds it and returns correct results, with no user-visible error. -- Actively-used helper indexes (queried more recently than the purge - duration) are never evicted. - -## Depends on - -Ticket 060 (timeline custom sort needs helper BTreeMap index) — this -ticket assumes helper indexes already exist.