diff --git a/src/data_value.rs b/src/data_value.rs index cd5c89c..2d66651 100644 --- a/src/data_value.rs +++ b/src/data_value.rs @@ -41,6 +41,16 @@ pub struct ParamDeclaration { /// ordinary (non-auto) params. #[serde(default)] pub auto_derived: bool, + /// Set for a `log stream` column/message param declared by + /// `crate::engine::auto_declare_log_stream_params`. Those params' `message_match_regex` is a + /// pure fixed character-offset slice with no content anchor of its own (ticket 076) — it'll + /// happily "match" and extract garbage from any line at least that long, including a wrapped + /// multi-line dump's continuation line or the header row itself. When set, extraction ( + /// `extract_param_raw`/`derive_log_params`) additionally requires + /// `crate::logfmt::is_log_stream_line(text)` before trusting the regex at all. + /// `#[serde(default)]` so params declared before this field existed keep extracting as before. + #[serde(default)] + pub log_stream_gated: bool, } /// Reads a value out of a JSON object by dot-separated field path (e.g. `"foo.bar"`), for diff --git a/src/dto.rs b/src/dto.rs index 167841d..6675a5b 100644 --- a/src/dto.rs +++ b/src/dto.rs @@ -78,6 +78,12 @@ pub struct StatsDto { /// than the fixed per-entry bookkeeping), and needs no per-entry /// bookkeeping beyond what eviction already tracks. pub approx_bytes_total: u64, + /// Ticket 073: the highest ingestion-to-engine-processing lag (`ottyl_ts` at read time vs. when + /// the engine actor actually got around to processing that line) seen within the trailing 5 + /// seconds, in milliseconds — see `crate::engine::record_ingest_lag`'s rolling high-watermark. + /// Stays near zero under normal load; rises under actor backpressure (a slow query in flight, a + /// large backlog) and decays back out of the window once the stall clears. + pub ingest_lag_high_watermark_ms: i64, } #[derive(Debug, Deserialize)] diff --git a/src/engine.rs b/src/engine.rs index c2383a0..df2dc2d 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -171,11 +171,20 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle // 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; + // Ticket 073: rolling high-watermark of ingestion-to-engine-processing lag samples — see + // `record_ingest_lag`/`ingest_lag_high_watermark_ms`. + let mut ingest_lag_samples: Vec<(Timestamp, i64)> = Vec::new(); while let Some(message) = rx.recv().await { match message { EngineMessage::Ingest(line) => { let IngestedLine { sequence, timestamp, text } = line; + let processed_at = Timestamp::now(); + record_ingest_lag( + &mut ingest_lag_samples, + processed_at, + (processed_at.as_millis() - timestamp.as_millis()).max(0), + ); approx_bytes_total += approx_log_bytes(&text); index.insert( IndexKey::IngestedLog { sequence, timestamp }, @@ -256,7 +265,14 @@ 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, &mut sort_helper_indexes); + let ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms(&ingest_lag_samples, Timestamp::now()); + let response = handle_query( + &index, + query, + approx_bytes_total, + ingest_lag_high_watermark_ms, + &mut sort_helper_indexes, + ); let _ = respond_to.send(response); } } @@ -450,6 +466,7 @@ fn auto_declare_params(index: &mut BTreeMap, text: &str) { kind: infer_kind(&candidate.value), elapsed_since_last: false, auto_derived: true, + log_stream_gated: false, }), ); new_this_line += 1; @@ -502,6 +519,7 @@ fn auto_declare_log_stream_params( kind: infer_kind(value.trim()), elapsed_since_last: false, auto_derived: true, + log_stream_gated: true, }), ); } @@ -516,6 +534,7 @@ fn auto_declare_log_stream_params( kind: infer_kind(message.trim()), elapsed_since_last: false, auto_derived: true, + log_stream_gated: true, }) }); } @@ -724,11 +743,13 @@ struct ResolvedParam { /// `capture_group`/`json_field_path` above are unused — resolution goes through /// `resolve_param_value`'s computed path instead of `extract_param_raw`. elapsed_since_last: bool, + /// See `crate::data_value::ParamDeclaration::log_stream_gated`. + log_stream_gated: bool, } fn resolved_param_declarations(index: &BTreeMap) -> HashMap { param_declarations_range(index) - .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, _)| { + .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| { Regex::new(message_match_regex).ok().map(|regex| { ( name.to_string(), @@ -738,6 +759,7 @@ fn resolved_param_declarations(index: &BTreeMap) -> HashMap json_field_path: json_field_path.map(|path| path.to_string()), kind, elapsed_since_last, + log_stream_gated, }, ) }) @@ -749,13 +771,19 @@ fn resolved_param_declarations(index: &BTreeMap) -> HashMap /// precedence when set) — shared by `apply_metric_to_log` (resolving a metric's `label_params`/ /// `value_param`) and `derive_log_params` (a standalone param declaration shown as a logs-table /// column). Returns `None` when the regex doesn't match, or (for a JSON field path) the field can't -/// be resolved. +/// be resolved. `log_stream_gated` (ticket 076) additionally requires `text` to pass +/// `crate::logfmt::is_log_stream_line` before the regex is trusted at all — see +/// `crate::data_value::ParamDeclaration::log_stream_gated`. fn extract_param_raw( text: &str, regex: &Regex, capture_group: &str, json_field_path: Option<&str>, + log_stream_gated: bool, ) -> Option { + if log_stream_gated && !crate::logfmt::is_log_stream_line(text) { + return None; + } match json_field_path.filter(|path| !path.is_empty()) { Some(field_path) => crate::data_value::extract_json_field(text, field_path), None => regex.captures(text)?.name(capture_group).map(|m| m.as_str().to_string()), @@ -795,7 +823,13 @@ fn resolve_param_value( .map(|previous| MetricPrimitive::Duration(timestamp.as_millis() - previous.as_millis())); } - let raw = extract_param_raw(text, &resolved.regex, &resolved.capture_group, resolved.json_field_path.as_deref())?; + let raw = extract_param_raw( + text, + &resolved.regex, + &resolved.capture_group, + resolved.json_field_path.as_deref(), + resolved.log_stream_gated, + )?; parse_metric_primitive(&raw, resolved.kind) } @@ -1296,7 +1330,7 @@ fn metric_declarations_range( fn param_declarations_range( index: &BTreeMap, -) -> impl Iterator, MetricPrimitiveKind, bool, bool)> { +) -> impl Iterator, MetricPrimitiveKind, bool, bool, bool)> { index .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit) .filter_map(|(key, value)| match (key, value) { @@ -1310,6 +1344,7 @@ fn param_declarations_range( kind, elapsed_since_last, auto_derived, + log_stream_gated, }), ) => Some(( name.as_str(), @@ -1319,6 +1354,7 @@ fn param_declarations_range( *kind, *elapsed_since_last, *auto_derived, + *log_stream_gated, )), _ => None, }) @@ -1359,13 +1395,20 @@ const SORT_HELPER_INDEX_PURGE_AFTER: std::time::Duration = std::time::Duration:: /// 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. +/// A compiled param declaration: name, regex, capture group, JSON field path, kind, and whether +/// `crate::data_value::ParamDeclaration::log_stream_gated` was set. +type CompiledParamDeclaration = (String, Regex, String, Option, MetricPrimitiveKind, bool); + fn derive_single_param( text: &str, - declarations: &[(String, Regex, String, Option, MetricPrimitiveKind)], + declarations: &[CompiledParamDeclaration], param_name: &str, ) -> Option { - let (_, regex, capture_group, json_field_path, kind) = + let (_, regex, capture_group, json_field_path, kind, log_stream_gated) = declarations.iter().find(|(name, ..)| name == param_name)?; + if *log_stream_gated && !crate::logfmt::is_log_stream_line(text) { + return None; + } if *kind == MetricPrimitiveKind::Boolean { return Some(MetricPrimitive::Boolean(regex.is_match(text))); } @@ -1426,12 +1469,10 @@ fn purge_stale_sort_helper_indexes(sort_helper_indexes: &mut HashMap, -) -> Vec<(String, Regex, String, Option, MetricPrimitiveKind)> { +fn compiled_param_declarations(index: &BTreeMap) -> Vec { param_declarations_range(index) - .filter(|(_, _, _, _, _, elapsed_since_last, _)| !elapsed_since_last) - .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _, _)| { + .filter(|(_, _, _, _, _, elapsed_since_last, _, _)| !elapsed_since_last) + .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _, _, log_stream_gated)| { Regex::new(message_match_regex).ok().map(|re| { ( name.to_string(), @@ -1439,6 +1480,7 @@ fn compiled_param_declarations( capture_group.to_string(), json_field_path.map(|path| path.to_string()), kind, + log_stream_gated, ) }) }) @@ -1450,12 +1492,12 @@ fn compiled_param_declarations( /// exception is a `Boolean`-kind param, which never uses a capture group and is instead present as /// `true`/`false` on every line based purely on whether `message_match_regex` matches /// (case-insensitively or per the regex's own flags). -fn derive_log_params( - text: &str, - declarations: &[(String, Regex, String, Option, MetricPrimitiveKind)], -) -> BTreeMap { +fn derive_log_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap { let mut params = BTreeMap::new(); - for (name, regex, capture_group, json_field_path, kind) in declarations { + for (name, regex, capture_group, json_field_path, kind, log_stream_gated) in declarations { + if *log_stream_gated && !crate::logfmt::is_log_stream_line(text) { + continue; + } if *kind == MetricPrimitiveKind::Boolean { params.insert(name.clone(), MetricPrimitive::Boolean(regex.is_match(text))); continue; @@ -1481,6 +1523,17 @@ fn derive_log_params( params } +/// `LogEntryDto::params` for one line: `derive_log_params`'s declared-param values, plus the +/// reserved `ottyl_fmt` pseudo-param (ticket 072) — computed line-locally the same way filter-bar +/// `ottyl_fmt` comparisons already do (`compile_ottyl_fmt_comparison`), not stored as a +/// `ParamDeclaration`, so it's merged in here rather than inside `derive_log_params` itself (whose +/// other callers/tests want the plain declared-param set only). +fn log_entry_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap { + let mut params = derive_log_params(text, declarations); + params.insert("ottyl_fmt".to_string(), MetricPrimitive::String(crate::logfmt::detect_format(text).to_string())); + params +} + /// Ticket 011's approximate per-log byte size: see `StatsDto::approx_bytes_total`'s /// doc comment for why the raw UTF-8 text length is the chosen proxy rather than a /// full struct-size estimate. @@ -1488,6 +1541,47 @@ fn approx_log_bytes(text: &str) -> u64 { text.len() as u64 } +/// Ticket 073: how long a rolling ingestion-to-engine-processing lag sample stays eligible to be +/// the reported high watermark. +const INGEST_LAG_WATERMARK_WINDOW_MS: i64 = 5000; +/// Ticket 073: at most this many lag samples are held at once — a fresh sample only displaces the +/// current *lowest*-lag held sample (never appended past this cap), so the collection always +/// reflects the highest few lags seen recently rather than every sample. +const INGEST_LAG_WATERMARK_MAX_ENTRIES: usize = 5; + +/// Records one `(engine processing time) - (ottyl_ts)` lag sample into the rolling high-watermark +/// collection the wishlist asked for (ticket 073): up to `INGEST_LAG_WATERMARK_MAX_ENTRIES` samples, +/// each valid only within the trailing `INGEST_LAG_WATERMARK_WINDOW_MS` — a sample older than that +/// window is dropped outright (an ingest burst that happened seconds ago shouldn't keep the reported +/// watermark artificially high forever), and once the collection is at capacity a new sample only +/// replaces the current *lowest*-lag entry, and only if it's actually larger. Called on every +/// `EngineMessage::Ingest`, so this stays O(`INGEST_LAG_WATERMARK_MAX_ENTRIES`) per ingested line. +fn record_ingest_lag(samples: &mut Vec<(Timestamp, i64)>, now: Timestamp, lag_ms: i64) { + samples.retain(|(recorded_at, _)| now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS); + if samples.len() < INGEST_LAG_WATERMARK_MAX_ENTRIES { + samples.push((now, lag_ms)); + return; + } + let Some((min_index, &(_, min_lag))) = samples.iter().enumerate().min_by_key(|(_, (_, lag))| *lag) else { + return; + }; + if lag_ms > min_lag { + samples[min_index] = (now, lag_ms); + } +} + +/// The current high watermark to report (`StatsDto::ingest_lag_high_watermark_ms`): the highest +/// still-in-window sample, or `0` when nothing's been recorded (or everything's aged out) — a +/// missing signal reads as "no lag", not as an error. +fn ingest_lag_high_watermark_ms(samples: &[(Timestamp, i64)], now: Timestamp) -> i64 { + samples + .iter() + .filter(|(recorded_at, _)| now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS) + .map(|(_, lag)| *lag) + .max() + .unwrap_or(0) +} + /// Evicts the oldest (lowest-`sequence`) ingested logs until `log_count` is /// back at or under `max_logs`. `sequence` numbering is not renumbered on /// eviction, and any `Bucket` values already derived from an evicted log are @@ -1550,6 +1644,7 @@ fn handle_query( index: &BTreeMap, query: EngineQuery, approx_bytes_total: u64, + ingest_lag_high_watermark_ms: i64, sort_helper_indexes: &mut HashMap, ) -> EngineResponse { match query { @@ -1592,7 +1687,7 @@ fn handle_query( sequence, timestamp_ms: timestamp.as_millis(), text: text.to_string(), - params: derive_log_params(text, &compiled_params), + params: log_entry_params(text, &compiled_params), }) .collect(), // Ticket 060: walk the (lazily built-or-reused) helper index for this sort key in @@ -1645,7 +1740,7 @@ fn handle_query( sequence, timestamp_ms: timestamp.as_millis(), text: text.clone(), - params: derive_log_params(text, &compiled_params), + params: log_entry_params(text, &compiled_params), }); if entries.len() >= limit { break; @@ -1706,11 +1801,12 @@ fn handle_query( first_timestamp_ms, last_timestamp_ms, approx_bytes_total, + ingest_lag_high_watermark_ms, }) } EngineQuery::ParamDeclarations => { let declarations = param_declarations_range(index) - .map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, auto_derived)| { + .map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, auto_derived, _)| { ParamDeclarationDto { name: name.to_string(), message_match_regex: message_match_regex.to_string(), @@ -2192,6 +2288,7 @@ mod tests { kind: MetricPrimitiveKind::String, elapsed_since_last: false, auto_derived: false, + log_stream_gated: false, } } @@ -2341,6 +2438,71 @@ mod tests { ); } + #[test] + fn log_stream_columns_are_not_contaminated_by_non_columnar_lines() { + // Ticket 076: a wrapped multi-line dump's continuation line, and the header row itself, must + // not pick up garbage values for the fixed-offset columns once they're declared. + let mut index = BTreeMap::new(); + ingest_log_stream_fixture(&mut index); + let compiled = compiled_param_declarations(&index); + + let continuation_line = "so_gencnt: 4042707 so_state: 0x0"; + let params = derive_log_params(continuation_line, &compiled); + assert!(!params.contains_key("Timestamp"), "continuation line got a Timestamp: {params:?}"); + assert!(!params.contains_key("Activity"), "continuation line got an Activity: {params:?}"); + assert!(!params.contains_key("PID"), "continuation line got a PID: {params:?}"); + + let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); + let header_params = derive_log_params(header_line, &compiled); + assert!(!header_params.contains_key("Timestamp"), "header line got a Timestamp: {header_params:?}"); + assert!(!header_params.contains_key("Activity"), "header line got an Activity: {header_params:?}"); + assert!(!header_params.contains_key("Thread"), "header line got a Thread: {header_params:?}"); + } + + /// Ticket 074/076 end-to-end: sorting a real `log stream` capture by PID (a non-monotonic, + /// declared-via-`auto_declare_log_stream_params` column, exactly what the wishlist reported as + /// "does not work") must actually reorder the response — this is the scenario ticket 076's + /// contamination fix was expected to unblock, verified here via the real `EngineQuery::Logs` + /// sorted path (`handle_query`) rather than just the lower-level param-derivation helpers. + #[test] + fn logs_query_sorted_by_a_log_stream_column_actually_reorders_real_log_stream_output() { + let mut index = BTreeMap::new(); + let mut header = None; + for (i, line) in LOG_STREAM_FIXTURE.lines().enumerate() { + let sequence = i as u64; + let timestamp = Timestamp::from_millis(sequence as i64); + auto_declare_log_stream_params(&mut index, &mut header, line); + auto_declare_params(&mut index, line); + index.insert( + IndexKey::IngestedLog { sequence, timestamp }, + DataValue::LogLiteral { text: line.to_string() }, + ); + } + + let entries = match handle_query(&index, logs_query_sorted("PID", Some(false)), 0, 0, &mut HashMap::new()) { + EngineResponse::Logs(entries) => entries, + _ => panic!("expected Logs response"), + }; + assert!(!entries.is_empty(), "expected at least one line with a PID value"); + + let pids: Vec = entries + .iter() + .map(|entry| match entry.params.get("PID") { + Some(MetricPrimitive::Number(Number::Int(pid))) => *pid, + other => panic!("expected every returned entry to have a numeric PID, got {other:?}"), + }) + .collect(); + let mut sorted_pids = pids.clone(); + sorted_pids.sort_unstable(); + assert_eq!(pids, sorted_pids, "entries were not actually reordered by PID"); + + // Sanity check this isn't a no-op: the fixture's own ingest order is not already PID-sorted. + let original_order: Vec = entries.iter().map(|e| e.sequence).collect(); + let mut sorted_by_sequence = original_order.clone(); + sorted_by_sequence.sort_unstable(); + assert_ne!(original_order, sorted_by_sequence, "PID sort produced the same order as plain ingest order"); + } + #[test] fn non_log_stream_lines_do_not_get_any_columns_declared_without_a_cached_header() { let mut index = BTreeMap::new(); @@ -2415,6 +2577,7 @@ mod tests { kind: MetricPrimitiveKind::String, elapsed_since_last: true, auto_derived: false, + log_stream_gated: false, }, ); assert!(result.is_err()); @@ -2436,6 +2599,7 @@ mod tests { kind: MetricPrimitiveKind::String, elapsed_since_last: false, auto_derived: false, + log_stream_gated: false, }, ); assert!(result.is_ok()); @@ -2458,6 +2622,7 @@ mod tests { kind: MetricPrimitiveKind::Number, elapsed_since_last: false, auto_derived: false, + log_stream_gated: false, }, ) .unwrap(); @@ -2519,6 +2684,63 @@ mod tests { assert_eq!(approx_bytes_total, 3); } + // -- Ticket 073: ingest lag high watermark -- + + #[test] + fn ingest_lag_watermark_reports_the_highest_sample_seen() { + let mut samples = Vec::new(); + let t0 = Timestamp::from_millis(0); + record_ingest_lag(&mut samples, t0, 10); + record_ingest_lag(&mut samples, t0, 50); + record_ingest_lag(&mut samples, t0, 20); + assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 50); + } + + #[test] + fn ingest_lag_watermark_only_replaces_the_current_minimum_once_at_capacity() { + let mut samples = Vec::new(); + let t0 = Timestamp::from_millis(0); + for lag in [10, 20, 30, 40, 50] { + record_ingest_lag(&mut samples, t0, lag); + } + assert_eq!(samples.len(), 5); + assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 50); + + // Smaller than every held sample -- nothing changes, so the max is unaffected. + record_ingest_lag(&mut samples, t0, 5); + assert_eq!(samples.len(), 5); + assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 50); + + // Bigger than the current minimum (10) -- replaces it, so the new min becomes 20 and the max + // stays 50 (this sample didn't beat it). + record_ingest_lag(&mut samples, t0, 25); + assert_eq!(samples.len(), 5); + assert!(samples.iter().all(|(_, lag)| *lag != 10), "the old minimum (10) should have been replaced"); + assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 50); + } + + #[test] + fn ingest_lag_watermark_decays_once_samples_age_out_of_the_window() { + let mut samples = Vec::new(); + let t0 = Timestamp::from_millis(0); + record_ingest_lag(&mut samples, t0, 500); + assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 500); + + // Still within the 5s window just before it closes. + let almost_expired = Timestamp::from_millis(INGEST_LAG_WATERMARK_WINDOW_MS - 1); + assert_eq!(ingest_lag_high_watermark_ms(&samples, almost_expired), 500); + + // Past the 5s window -- the stale spike no longer counts, reads back to the "no lag" default. + let expired = Timestamp::from_millis(INGEST_LAG_WATERMARK_WINDOW_MS); + assert_eq!(ingest_lag_high_watermark_ms(&samples, expired), 0); + + // A fresh low-lag sample recorded at that point evicts the stale one outright rather than + // just being masked by the window filter. + record_ingest_lag(&mut samples, expired, 5); + assert_eq!(samples.len(), 1); + assert_eq!(ingest_lag_high_watermark_ms(&samples, expired), 5); + } + fn insert_ingested_log(index: &mut BTreeMap, sequence: u64, text: &str) { index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, @@ -2642,6 +2864,7 @@ mod tests { kind, elapsed_since_last: false, auto_derived: false, + log_stream_gated: false, }, ) .unwrap(); @@ -2662,6 +2885,7 @@ mod tests { kind: MetricPrimitiveKind::Duration, elapsed_since_last: true, auto_derived: false, + log_stream_gated: false, }, ) .unwrap(); @@ -2942,6 +3166,7 @@ mod tests { kind: MetricPrimitiveKind::String, elapsed_since_last: false, auto_derived: false, + log_stream_gated: false, }, ) .unwrap(); @@ -3761,6 +3986,7 @@ mod tests { end_ms: None, }, 0, + 0, &mut HashMap::new(), ); match response { @@ -3783,6 +4009,7 @@ mod tests { end_ms: None, }, 0, + 0, &mut HashMap::new(), ); match response { @@ -3876,6 +4103,7 @@ mod tests { &index, EngineQuery::Promql { query: "count(ottyl_seq)".to_string(), start_ms: None, end_ms: None }, 0, + 0, &mut HashMap::new(), ); match response { @@ -3931,6 +4159,27 @@ mod tests { }) } + /// Ticket 072: `ottyl_fmt` is filterable in the search bar (`compile_ottyl_fmt_comparison`) but + /// wasn't previously attached to `LogEntryDto::params` at all, so Log Preview's column picker + /// could offer it as a column yet every row showed blank. + #[test] + fn logs_query_response_includes_ottyl_fmt_per_entry() { + let mut index = BTreeMap::new(); + ingest_log(&mut index, 0, r#"{"level":"error"}"#); + ingest_log(&mut index, 1, "level=info msg=hello"); + ingest_log(&mut index, 2, "just plain prose"); + + match handle_query(&index, logs_query(None), 0, 0, &mut HashMap::new()) { + EngineResponse::Logs(entries) => { + let formats: Vec> = entries.iter().map(|e| e.params.get("ottyl_fmt")).collect(); + assert_eq!(formats[0], Some(&MetricPrimitive::String("json".to_string()))); + assert_eq!(formats[1], Some(&MetricPrimitive::String("logfmt".to_string()))); + assert_eq!(formats[2], Some(&MetricPrimitive::String("none".to_string()))); + } + _ => panic!("expected Logs response"), + } + } + #[test] fn logs_query_with_regex_filter_returns_only_matching_lines() { let mut index = BTreeMap::new(); @@ -3939,7 +4188,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, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 3]); @@ -3955,7 +4204,7 @@ mod tests { ingest_log(&mut index, i, &format!("line {i}")); } - match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 1]); @@ -3975,7 +4224,7 @@ mod tests { ); } - match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![3, 4]); @@ -4002,7 +4251,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, &mut HashMap::new()) { + match handle_query(&index, logs_query_sorted("priority", None), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 2, 0]); @@ -4030,7 +4279,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, &mut HashMap::new()) { + match handle_query(&index, logs_query_sorted("priority", Some(true)), 0, 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 @@ -4056,7 +4305,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, &mut HashMap::new()) { + match handle_query(&index, logs_query_sorted("priority", Some(false)), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![2, 0]); @@ -4070,7 +4319,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, &mut HashMap::new()) { + match handle_query(&index, logs_query_sorted("does_not_exist", None), 0, 0, &mut HashMap::new()) { EngineResponse::Error(err) => assert!(err.contains("does_not_exist")), _ => panic!("expected EngineResponse::Error"), } @@ -4094,7 +4343,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, &mut HashMap::new()) { + match handle_query(&index, logs_query(None), 0, 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]); @@ -4124,7 +4373,7 @@ mod tests { // 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) { + match handle_query(&index, logs_query_sorted("priority", None), 0, 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]); @@ -4157,7 +4406,7 @@ mod tests { ); assert_eq!(sort_helper_indexes["priority"].index.len(), 5); - match handle_query(&index, logs_query_sorted("priority", None), 0, &mut sort_helper_indexes) { + match handle_query(&index, logs_query_sorted("priority", None), 0, 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 @@ -4217,13 +4466,13 @@ mod tests { 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); + let _ = handle_query(&index, logs_query_sorted("priority", None), 0, 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) { + match handle_query(&index, logs_query_sorted("priority", None), 0, 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]); @@ -4242,7 +4491,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, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("ottyl_fmt==json")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 3]); @@ -4250,7 +4499,7 @@ mod tests { _ => panic!("expected Logs response"), } - match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1]); @@ -4258,7 +4507,7 @@ mod tests { _ => panic!("expected Logs response"), } - match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![2]); @@ -4274,7 +4523,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, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); @@ -4282,7 +4531,7 @@ mod tests { _ => panic!("expected Logs response"), } - match handle_query(&index, logs_query(Some("json[code]>=300")), 0, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("json[code]>=300")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); @@ -4307,6 +4556,7 @@ mod tests { &index, logs_query(Some(r#"regex("boom") and requests[level]==error"#)), 0, + 0, &mut HashMap::new(), ) { EngineResponse::Logs(entries) => { @@ -4321,6 +4571,7 @@ mod tests { &index, logs_query(Some(r#"regex("boom") or requests[level]==error"#)), 0, + 0, &mut HashMap::new(), ) { EngineResponse::Logs(entries) => { @@ -4338,7 +4589,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, &mut HashMap::new()) else { + let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, 0, &mut HashMap::new()) else { panic!("expected Logs response"); }; @@ -4360,7 +4611,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, &mut HashMap::new()) else { + let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, 0, &mut HashMap::new()) else { panic!("expected Logs response"); }; @@ -4376,7 +4627,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, &mut HashMap::new()) { + match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0, 0, &mut HashMap::new()) { EngineResponse::Timeline(buckets) => { let total: u64 = buckets.iter().map(|b| b.count).sum(); assert_eq!(total, 2); @@ -4390,12 +4641,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, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0, 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, &mut HashMap::new()) { + match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0, 0, &mut HashMap::new()) { EngineResponse::Error(message) => assert!(!message.is_empty()), _ => panic!("expected an Error response for a malformed filter string"), } @@ -4409,7 +4660,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, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("requests==5")), 0, 0, &mut HashMap::new()) { EngineResponse::Error(message) => assert!(message.contains("param")), _ => panic!("expected an Error response for a metric comparison without [param]"), } @@ -4424,7 +4675,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, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("requests[nonexistent_param]==\"x\"")), 0, 0, &mut HashMap::new()) { EngineResponse::Error(message) => { assert!( message.contains("nonexistent_param"), @@ -4454,7 +4705,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, &mut HashMap::new()) { + match handle_query(&index, logs_query(Some("requests[level]==error")), 0, 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 4f24676..c62f1e4 100644 --- a/src/logfmt.rs +++ b/src/logfmt.rs @@ -90,6 +90,13 @@ pub fn extract_pairs(line: &str) -> Vec<(&str, &str, bool)> { /// separately contain real `key=value` pairs (see the WiFi/radio stat lines in the fixture), which /// `crate::engine::auto_declare_params` keeps discovering via `extract_pairs` regardless of this /// line's `ottyl_fmt` label. +/// `crate::data_value::ParamDeclaration::log_stream_gated` params (the columns/`message` +/// `auto_declare_log_stream_params` declares) require this to hold before their own fixed-offset +/// `message_match_regex` is trusted at all (ticket 076) — that regex is pure character-offset +/// arithmetic with no content anchor of its own, so it happily "matches" (and extracts garbage +/// from) any unrelated line at least that long, e.g. a wrapped multi-line dump's continuation line +/// or the header line itself. This can't be baked into `message_match_regex` as a leading lookahead +/// instead: the `regex` crate deliberately doesn't support look-around at all. static LOG_STREAM_RE: LazyLock = LazyLock::new(|| { Regex::new( 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+[^\s:]+:\s", @@ -583,6 +590,23 @@ mod tests { ); } + #[test] + fn is_log_stream_line_rejects_non_columnar_lines_that_a_fixed_offset_column_regex_would_still_match() { + // Ticket 076: this is exactly why `ParamDeclaration::log_stream_gated` params must check + // `is_log_stream_line` before trusting their own offset-based capture at all -- the raw + // column regex has no content anchor and matches plenty on lines that clearly aren't + // columnar `log stream` output. + let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); + let header = detect_log_stream_header(header_line).unwrap(); + let timestamp_re = Regex::new(&log_stream_column_pattern(&header, 0, "Timestamp")).unwrap(); + + let continuation_line = "so_gencnt: 4042707 so_state: 0x0"; + assert!(timestamp_re.captures(continuation_line).is_some(), "sanity: regex alone still matches garbage"); + assert!(!is_log_stream_line(continuation_line)); + assert!(!is_log_stream_line(header_line)); + assert!(!is_log_stream_line("}")); + } + #[test] fn a_wifi_stat_dump_line_is_still_recognized_as_log_stream_despite_embedded_key_value_pairs() { let line = LOG_STREAM_FIXTURE diff --git a/src/profile.rs b/src/profile.rs index 9a0053b..93a95f6 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -103,6 +103,7 @@ impl From<&ParamDeclarationRecord> for DeclareParamRequest { elapsed_since_last: record.elapsed_since_last, // A profile only ever stores manually-declared params (see `record_param_declaration`). auto_derived: false, + log_stream_gated: false, } } } @@ -828,6 +829,7 @@ mod tests { kind: MetricPrimitiveKind::String, elapsed_since_last: false, auto_derived: false, + log_stream_gated: false, } } diff --git a/tickets/071-filter-bar-button-placement-and-popup-clarity.md b/tickets/071-filter-bar-button-placement-and-popup-clarity.md deleted file mode 100644 index 2ffdadd..0000000 --- a/tickets/071-filter-bar-button-placement-and-popup-clarity.md +++ /dev/null @@ -1,49 +0,0 @@ -# 071 — Filter bar's syntax button/popup is ugly and incoherent - -## Summary - -Wishlist feedback, verbatim: "UI filter ugly syntax. The button moved -to the right, away from the existing button cluster. The popup is a -incoherent mess and should be clearer." - -Two distinct complaints bundled together: - -1. Layout regression — some button (the filter-help/syntax button, - presumably introduced or moved by ticket 065's filter-bar help - rewrite) now sits away from the rest of the filter bar's button - cluster in `FilterBar.tsx` (`webapp/src/components/FilterBar.tsx`, - currently 67 lines), instead of grouped with the others. -2. Content/clarity — whatever popup that button opens (the filter - syntax help introduced/reworked in ticket 065, - "filter-bar-help-should-describe-filter-syntax-not-regex") reads as - an incoherent mess rather than a clear reference. - -## Scope - -- Reproduce in the running app: open the filter bar, locate the - syntax/help button, and confirm which button has drifted from the - cluster and what "incoherent" means concretely (wall of text? - disorganized sections? unclear examples?) — the wishlist entry - doesn't specify, so this needs eyes-on before a fix. -- Fix the button's position so it's grouped with the filter bar's other - controls again. -- Rework the popup's content/layout for clarity — likely restructuring - into clearly separated sections (e.g. by filter feature: regex match, - metric comparison, datetime, sort) with short examples per section, - rather than a single undifferentiated block. -- Out of scope: changing the underlying filter grammar itself - (`filter_parser.pest`/`filter_parser.rs`) — this is presentation only. - -## Acceptance criteria - -- The filter syntax button sits back with the filter bar's other - buttons, not isolated to one side. -- The help popup reads as a structured, scannable reference rather than - a single dense block — verified visually in the running app (`make - dev`), not just by inspecting JSX. - -## Depends on - -None. Touches the same area as ticket 065 and ticket 066 -("filter-bar-value-doesnt-reflect-active-preset-or-sort") — check those -haven't already reshuffled this layout before starting. diff --git a/tickets/072-log-preview-format-column.md b/tickets/072-log-preview-format-column.md deleted file mode 100644 index 033cb06..0000000 --- a/tickets/072-log-preview-format-column.md +++ /dev/null @@ -1,44 +0,0 @@ -# 072 — Log Preview needs a format column (`ottyl_fmt`) - -## Summary - -Wishlist feedback: "Log preview needs format column." `ottyl_fmt` -(json/logfmt/none, `crate::logfmt::detect_format`) already exists as a -reserved, filterable pseudo-param (`OTTYL_RESERVED_PARAM_NAMES`, -`src/engine.rs:1960`; filter-bar support via -`compile_ottyl_fmt_comparison`, `src/engine.rs:1067`) — you can already -filter `ottyl_fmt==json` in the search bar. But it isn't surfaced as a -visible column in the Log Preview table -(`webapp/src/components/LogPreview.tsx`), and it can't be, because -`LogEntryDto` (`src/dto.rs:10-16`) doesn't include it: `params` only -holds declared `ParamDeclaration` matches, and format detection is -computed line-locally elsewhere, never attached to the returned log -entry. - -## Scope - -- Add the per-line detected format to `LogEntryDto` (`src/dto.rs:10`), - populated wherever log entries are assembled for the `Logs` query - response in `engine.rs` (reuse `crate::logfmt::detect_format`, the - same detection `ottyl_fmt` filtering already calls). -- Add `ottyl_fmt` as a column option in Log Preview's `ColumnPicker` - (`webapp/src/components/LogPreview.tsx:28-56`) alongside declared - params — it's a reserved pseudo-param like `ottyl_seq`/`ottyl_ts`, so - treat it the way the params list already flags `reserved` - (`dto.rs:38`, `ParamDeclarationDto.reserved`) rather than requiring - it be "declared" to show up as a pickable column. -- Render the column value as the short format label (`json`/`logfmt`/ - `none`) matching what `ottyl_fmt` filter comparisons already produce. - -## Acceptance criteria - -- A "Format" (or `ottyl_fmt`) column is available in the Log Preview's - column picker and, once toggled on, shows the correct detected format - per row — json lines show `json`, logfmt lines show `logfmt`, neither - shows `none`. -- No change to filter-bar `ottyl_fmt` comparison behavior (still works - as before). - -## Depends on - -None. diff --git a/tickets/073-ingestion-to-engine-processing-lag-tracking.md b/tickets/073-ingestion-to-engine-processing-lag-tracking.md deleted file mode 100644 index ee133f8..0000000 --- a/tickets/073-ingestion-to-engine-processing-lag-tracking.md +++ /dev/null @@ -1,64 +0,0 @@ -# 073 — Track and surface ingestion-to-engine-processing lag (high watermark) - -## Summary - -Wishlist feedback, verbatim: - -> There needs to be a ingstion to parsing latency on frontend. -> Basically when a log is read, it gets what would be the `ottyl_ts` -> value. However, the engine can lag on ingestion. That lag needs to -> be tracked as a high watermark. We can have a collection that stores -> say 5 highest values within the last 5 seconds. If a value in that -> collection is older than the 5s window, it gets replaced. If a value -> is higher than the minimum current, it gets replaced. The frontend -> will display the highest single value (humantime duration lag, or ms -> lag) alongside where it displays the logs and allocation size. - -Confirmed the mechanism this is about: `ingest.rs` stamps -`timestamp: Timestamp::now()` at read time (this becomes `ottyl_ts`, -e.g. `ingest.rs:25`/`:107`) *before* sending `EngineMessage::Ingest` -over the mpsc channel to the engine actor task. The engine only -processes it whenever the actor's message loop gets scheduled to it — -under load (large backlog, slow query in flight, etc.) that gap can -grow. Today nothing measures `(engine processing time) - (ottyl_ts)`, -so there's no visibility into ingest backpressure. - -The "allocation size" reference point is `approx_bytes_total` -(`src/dto.rs:80`, ticket 011), currently shown in -`webapp/src/components/NavMenu.tsx:104` in the stats popover — that's -the existing display location the lag figure should sit alongside. - -## Scope - -- In the engine actor's ingest handling (where `EngineMessage::Ingest` - is received, `engine.rs`), compute `lag = Timestamp::now() - - message.timestamp` at the moment of processing. -- Maintain a small rolling high-watermark collection per the spec - above: keep up to 5 entries, each valid only within the last 5 - seconds; a new lag value replaces the current minimum-held entry if - it's larger, and any entry older than the 5s window is evicted - (replaced) regardless of magnitude. Expose the current max of that - collection. -- Add this figure to whatever stats DTO backs the popover - `approx_bytes_total` comes from, and to the corresponding stats - query/response path in `api.rs`. -- Display it in `NavMenu.tsx` next to `approx_bytes_total` (`:104`), - formatted as a humantime-style duration for larger lag and plain ms - for small values — match whatever precedent this codebase already - uses for duration formatting (check `webapp/src/lib` for an existing - humanize helper before writing a new one). - -## Acceptance criteria - -- Under normal (non-lagged) ingestion, the displayed value stays near - zero. -- Artificially delaying engine processing (e.g. a slow query - saturating the actor's message loop, or a synthetic test) causes the - displayed high-watermark to rise and correctly decays out of the - rolling 5s window afterward. -- The value is visible in the same UI location as - `approx_bytes_total`. - -## Depends on - -None. diff --git a/tickets/074-timeline-sort-still-broken.md b/tickets/074-timeline-sort-still-broken.md deleted file mode 100644 index dfa1082..0000000 --- a/tickets/074-timeline-sort-still-broken.md +++ /dev/null @@ -1,68 +0,0 @@ -# 074 — Timeline sort-by-property still doesn't work after tickets 059-061 - -## Summary - -Wishlist feedback, verbatim: "The sorting on the timeline still does -not work. I have tried sorting things that I know are sorted (logs -param timestamp) and things that arent (thread name or pid). It does -not work (change view, index etc). If it's taking time to index, that -needs to be visible." - -This is a regression/bug report against functionality that should -already exist: sort-by-param was added in ticket 045, and tickets -060/061 (resolved in commit `de9ba6f`, "Resolve tickets 059-061: -generic log-stream columns + sort helper indexes") replaced the naive -full-materialize-then-sort with an incremental helper-index scheme -(`SortHelperIndexEntry`, `build_sort_helper_index`, -`update_sort_helper_indexes`, `purge_stale_sort_helper_indexes` in -`engine.rs:1332-1431`), wired into the `EngineQuery::Logs` sorted -branch around `engine.rs:1598-1634`. On the frontend, `sortBy`/ -`sortMissingFallback` are threaded from `LogTimeline.tsx`'s settings -through `App.tsx` (`:349-350`, `:505-506`) into `fetchLogs` -(`webapp/src/lib/api.ts:256-266`). Nothing found in a static read of -this wiring looks obviously broken, so this needs live reproduction, -not just a code read — treat the below as an investigation, not an -assumed root cause. - -## Scope - -- Reproduce with `make dev` against real piped input: declare a param - covering thread name or PID, set the timeline's sort dropdown to it, - and confirm whether results actually reorder. Also test sorting by an - already-monotonic value (`ottyl_ts` or a declared timestamp param) to - isolate whether the bug is "sort does nothing" vs. "sort scrambles an - already-correct order." -- Check whether the helper-index rebuild triggers correctly on first - use per sort key (`engine.rs:1604`, `.entry(sort_by...).or_insert_with`) - — a stale/empty helper index reused across a param rename or - re-declaration (see the ticket-061-era comments at `engine.rs:225-233` - about helper indexes needing invalidation when a declaration changes) - is one plausible cause worth ruling in or out. -- Check whether the frontend is actually sending `sort_by` on every - poll/refresh, not just the initial query — a poll path that omits it - (`webapp/src/lib/polling.ts`) would look exactly like "sorting - doesn't work" despite correct backend logic. -- Add visibility for the "still indexing" case the wishlist explicitly - calls out: if a helper index is being built (or rebuilt) for a - newly-selected sort key and that's slow enough to be user-visible, - surface a loading/building state in the timeline UI rather than - silently returning unsorted or partial results while it builds. -- Write a regression test once the actual defect is identified — - ticket 060's existing tests apparently didn't catch this, so identify - what scenario they miss. - -## Acceptance criteria - -- Sorting the timeline by a declared param (including thread - name/PID-style non-monotonic params) actually reorders displayed log - entries, verified live in the browser. -- Sorting by a param that's already effectively time-ordered doesn't - regress (still displays in the expected order). -- If a helper index build takes long enough to matter, the UI shows - that it's building rather than looking broken or stalled. - -## Depends on - -Builds on tickets 045, 060, 061 — read those and their resolving -commits (`de9ba6f` and its predecessor implementing 045) before -forming a hypothesis. diff --git a/tickets/076-log-stream-whitespace-columnar-parsing-broken.md b/tickets/076-log-stream-whitespace-columnar-parsing-broken.md deleted file mode 100644 index a47dda7..0000000 --- a/tickets/076-log-stream-whitespace-columnar-parsing-broken.md +++ /dev/null @@ -1,104 +0,0 @@ -# 076 — `log stream` declared columns get contaminated with garbage on non-columnar lines - -## Summary - -Wishlist feedback, verbatim: "The `log stream` type whitespace-separated -formatting is not working at all. I have a session `suited-gyrfalcon` -running, please investigate." - -Investigated directly against the live `suited-gyrfalcon` instance -(`ottyl sessions` → port 62059 → `/api/params`, `/api/logs`, -`/api/stats`; 236,069 ingested log lines, real `log stream` output). -Root cause confirmed — this is not a gap in header/column detection -(ticket 059's detector itself works correctly on genuine `log stream` -rows), it's that the *declared params* it produces get applied far too -broadly once they exist. - -**What's actually happening:** `auto_declare_log_stream_params` -(`engine.rs:472-497`) correctly gates *declaration* on -`is_log_stream_line(text)` (`:484`) — so it only fires column -declarations the first time it sees a genuine columnar data row. But -the resulting `ParamDeclaration.message_match_regex` for each column is -a pure **fixed character-offset** pattern with no content anchor at all -— e.g. observed live: `Activity` → `^.{55}(?P\S(?:.{0,19}\S)?)`, -`PID` → `^.{76}(?P\S(?:.{0,5}\S)?)`, `Thread` → -`^.{32}(?P\S(?:.{0,9}\S)?)` (from `log_stream_column_pattern`, -`logfmt.rs:287-303`). Once declared, these are ordinary -`ParamDeclaration`s: the *general* per-line param-derivation path -(`auto_declare_params`/`extract_param_raw`) applies every declared -param's regex to **every** ingested line, with no re-check that the -line is actually log-stream-shaped — there's no gate equivalent to -`is_log_stream_line` at derivation time, only at first-declaration time. - -Confirmed impact from the live instance: -- Sampling ~1000-line windows at several points across the 236k-line - stream, roughly **1-2% of lines get a garbage `Timestamp` value** - that isn't a real timestamp — e.g. line 306: - `"so_gencnt: 4042707 so_state: 0x0"` (a continuation line from a - wrapped multi-line network-stats dump, no relation to the columnar - format at all) got `Timestamp: "so_gencnt: 4042707 so_state: 0x0"`, - `Activity`/`PID`/etc sliced from arbitrary character offsets of that - unrelated text. Same for bracket/brace-only continuation lines - (`"}"`, `"]"`) inside pretty-printed dumps. -- Even the header line itself (sequence 0, - `"Timestamp Thread Type Activity PID TTL"`) ends up with - `Activity: "Activity"`, `Thread: "Thread"`, `Timestamp: "Timestamp"`, - `Type: "Type"` as its own param values — because ticket 052's - retroactive re-derivation (`engine.rs:540`, re-running - `auto_declare_params` over every retained line whenever a new param - is declared) re-applies the newly-declared fixed-offset regexes back - over already-ingested lines, including the header line, with no - exemption for it. -- This directly undermines ticket 074's "sorting is broken" complaint - too: sorting by `Timestamp` mixes ~1-2% garbage string values in among - otherwise lexically-sortable ISO timestamps, which would visibly - scramble ordering — worth re-checking ticket 074 once this is fixed, - the two may partially be the same underlying issue. - -This explains "not working at all" better than a full parse failure -would: the common case (genuine columnar rows) parses fine — it's that -the contamination on the ~1-2% of non-columnar lines interleaved -throughout every real `log stream` capture (multi-line dumps are -common) makes every declared log-stream column look untrustworthy. - -## Scope - -- Make the per-column `message_match_regex` self-verifying, not purely - positional: either (a) prefix each generated column/message pattern - (`log_stream_column_pattern`/`log_stream_message_pattern`, - `logfmt.rs:287-320`) with the same shape check `is_log_stream_line` - uses so the regex only matches on a genuinely columnar line, or (b) - gate the general param-derivation path itself so a param declared via - `auto_declare_log_stream_params` is only evaluated against lines - where `is_log_stream_line` holds — (a) is likely simpler since it - keeps the gate local to the declaration rather than adding a - special case to the generic derivation path. -- Fix ticket 052's retroactive re-derivation (`engine.rs:540`) to - respect that same gate, so the header line (and any other - non-columnar retained line) doesn't retroactively pick up columnar - param values. -- Add a regression test using the real contaminated lines captured - above (continuation lines like `"so_gencnt: ... so_state: ..."` and - the header line itself) alongside the existing - `log_stream_sample.txt` fixture — the current fixture and its tests - don't include a multi-line-dump continuation line, which is why - ticket 059's tests passed despite this. - -## Acceptance criteria - -- Ingesting a `log stream` capture that includes multi-line - continuation content (network stat dumps, bracketed/braced - pretty-printed blocks) does not populate `Timestamp`/`Thread`/`Type`/ - `Activity`/`PID`/`TTL` params on those continuation lines. -- The header line itself never ends up with its own column names as - values. -- Existing correctly-parsed columnar rows are unaffected. -- A regression test using real captured contamination examples (not - just the original fixture) passes. - -## Depends on - -Builds on ticket 059 (commit `de9ba6f`) and ticket 052's retroactive -re-derivation. Related to ticket 074 (timeline sort) — re-verify that -complaint after this fix lands, since garbage `Timestamp` values are a -plausible contributor to it. diff --git a/webapp/src/App.tsx b/webapp/src/App.tsx index 15c00f6..0751f9f 100644 --- a/webapp/src/App.tsx +++ b/webapp/src/App.tsx @@ -69,6 +69,23 @@ function saveLogViewerHeight(px: number) { } } +// Ticket 074: a sort helper index's one-time full-scan build (`build_sort_helper_index`) runs +// inline in the engine actor, so a `sort_by` poll can occasionally take much longer than the usual +// sub-poll-interval round trip — without this, that stall looks exactly like "sorting is broken" +// rather than "still indexing". Only flips `setBuilding(true)` if the fetch is still in flight after +// `thresholdMs`, so an ordinary fast poll (helper index already built/incrementally maintained) +// never flashes the indicator. +const SORT_BUILDING_INDICATOR_THRESHOLD_MS = 400 + +function trackSlowSortedFetch(sortBy: string | null, setBuilding: (building: boolean) => void, promise: Promise): Promise { + if (!sortBy) return promise + const timer = setTimeout(() => setBuilding(true), SORT_BUILDING_INDICATOR_THRESHOLD_MS) + return promise.finally(() => { + clearTimeout(timer) + setBuilding(false) + }) +} + /// Thin drag handle between the log viewer and the sections below it. Reports each pointer-move /// delta via `onResize` (letting the caller clamp/apply it) and fires `onResizeEnd` on release so /// the final size can be persisted once, not on every pixel of movement. @@ -286,6 +303,7 @@ function StandaloneLog() { const [filter, setFilter] = useState('') const [filterError, setFilterError] = useState(null) const [sortSettings, setSortSettings] = useState(loadLogSortSettings) + const [sortIndexBuilding, setSortIndexBuilding] = useState(false) function handleSortSettingsChange(next: LogSortSettings) { setSortSettings(next) @@ -341,14 +359,18 @@ function StandaloneLog() { bucketMs: effectiveRange ? computeBucketMs(effectiveRange.endMs - effectiveRange.startMs) : undefined, filter: filter || undefined, }), - fetchLogs({ - startMs: effectiveRange?.startMs, - endMs: effectiveRange?.endMs, - limit: 1000, - filter: filter || undefined, - sortBy: sortSettings.sortBy ?? undefined, - sortMissingFallback: sortSettings.sortBy ? sortSettings.sortMissingFallback : undefined, - }), + trackSlowSortedFetch( + sortSettings.sortBy, + setSortIndexBuilding, + fetchLogs({ + startMs: effectiveRange?.startMs, + endMs: effectiveRange?.endMs, + limit: 1000, + filter: filter || undefined, + sortBy: sortSettings.sortBy ?? undefined, + sortMissingFallback: sortSettings.sortBy ? sortSettings.sortMissingFallback : undefined, + }), + ), // Ticket 058: params can get auto-declared from freshly ingested lines mid-run, not just from a user's own edit. fetchParamDeclarations(), ]) @@ -393,6 +415,7 @@ function StandaloneLog() { paramNames={params.map((p) => p.name)} sortSettings={sortSettings} onSortSettingsChange={handleSortSettingsChange} + sortIndexBuilding={sortIndexBuilding} /> @@ -423,6 +446,7 @@ function FullApp() { const [logViewerCollapsed, setLogViewerCollapsed] = useSectionCollapsed('log-viewer') const [logViewerHeight, setLogViewerHeight] = useState(() => loadLogViewerHeight()) const [sortSettings, setSortSettings] = useState(loadLogSortSettings) + const [sortIndexBuilding, setSortIndexBuilding] = useState(false) function handleSortSettingsChange(next: LogSortSettings) { setSortSettings(next) @@ -497,14 +521,18 @@ function FullApp() { bucketMs: effectiveRange ? computeBucketMs(effectiveRange.endMs - effectiveRange.startMs) : undefined, filter: filter || undefined, }), - fetchLogs({ - startMs: effectiveRange?.startMs, - endMs: effectiveRange?.endMs, - limit: 1000, - filter: filter || undefined, - sortBy: sortSettings.sortBy ?? undefined, - sortMissingFallback: sortSettings.sortBy ? sortSettings.sortMissingFallback : undefined, - }), + trackSlowSortedFetch( + sortSettings.sortBy, + setSortIndexBuilding, + fetchLogs({ + startMs: effectiveRange?.startMs, + endMs: effectiveRange?.endMs, + limit: 1000, + filter: filter || undefined, + sortBy: sortSettings.sortBy ?? undefined, + sortMissingFallback: sortSettings.sortBy ? sortSettings.sortMissingFallback : undefined, + }), + ), // Ticket 058: params can get auto-declared from freshly ingested lines mid-run, not just from a user's own edit. fetchParamDeclarations(), ]) @@ -566,6 +594,7 @@ function FullApp() { paramNames={params.map((p) => p.name)} sortSettings={sortSettings} onSortSettingsChange={handleSortSettingsChange} + sortIndexBuilding={sortIndexBuilding} /> diff --git a/webapp/src/components/Docs.tsx b/webapp/src/components/Docs.tsx index 41fbdc0..d777e8f 100644 --- a/webapp/src/components/Docs.tsx +++ b/webapp/src/components/Docs.tsx @@ -78,75 +78,108 @@ const SECTIONS: DocSection[] = [ id: 'filtering', title: 'Filtering', body: ( -
+

The filter bar combines regex search over raw log text with comparisons against declared metrics, joined - by and/or. Examples: + by and/or. The server evaluates the filter string; a parse error shows as inline + text under the filter bar.

-
    -
  • - {'regex("error")'} — lines containing "error". -
    • -
  • - {'regex("error") and ottyl_seq<5'} — lines containing "error" with sequence under 5. -
    • -
  • - {'requests[duration]>100'} — the requests metric's captured{' '} - duration param over 100. -
    • -
  • - {'ottyl_ts>-5m'} — lines from the last 5 minutes. -
    • -
  • - {'ottyl_fmt==json'} — lines auto-detected as JSON. -
    • -
-

- ottyl_seq, ottyl_ts, and ottyl_fmt are reserved metric names present - on every log line, so they're always filterable without declaring anything. ottyl_fmt holds the - auto-detected format of the line's text — log_stream (macOS log stream output),{' '} - json, logfmt, or none. A log_stream line also gets{' '} - log_type, process, and pid params derived automatically from its - fixed-column header. All three reserved names also appear (flagged reserved) in{' '} - GET /api/params for discoverability, though there's no UI to edit or remove them — see also{' '} - Dashboards below for using ottyl_seq/ottyl_ts in a panel query. -

-

Date/time formats

-

Timestamp comparisons accept three forms:

-
    -
  • - relative offsets from now — 5h (5 hours from now), -3m (3 minutes ago) -
    • -
  • - absolute clock times — 12:00, 09:30:00 -
    • -
  • - full dates — 2026-01-15, 2026-01-15T09:30:00Z -
    • -
-

- The filter string is evaluated on the server. A parse error shows as inline text under the filter bar. -

-

Worked examples

-
    -
  • - Only lines mentioning "timeout" in the last 15 minutes:{' '} - {'regex("timeout") and ottyl_ts>-15m'} -
    • -
  • - Requests slower than 200ms, once a requests metric has a value-role{' '} - duration param declared: {'requests[duration]>200'} -
    • -
  • - 5xx responses on one path, given a requests metric with a path label param — - note that [param] comparisons only work against a declared metric's labels, not a - bare param declared on its own, and values are unquoted (no surrounding "..."):{' '} - {'requests[path]==/checkout and regex("5[0-9]{2}")'} -
    • -
  • - Everything before noon on a specific date: {'ottyl_ts<2024-06-15T12:00:00Z'} -
    • -
+ +
+

Regex match

+
    +
  • + {'regex("error")'} — lines containing "error". See the @ icon next to the + filter bar for the exact regex syntax supported (Rust's regex crate — no lookaround or + backreferences). +
    • +
+
+ +
+

Metric/param comparisons

+
    +
  • + {'requests[duration]>100'} — the requests metric's captured{' '} + duration param over 100. [param] comparisons only work against a{' '} + declared metric's labels, not a bare param declared on its own, and values are unquoted (no + surrounding "..."). +
    • +
+
+ +
+

Reserved names (always filterable)

+

+ ottyl_seq, ottyl_ts, and ottyl_fmt are present on every log line + without declaring anything, and also appear (flagged reserved) in{' '} + GET /api/params for discoverability, though there's no UI to edit or remove them. +

+
    +
  • + {'ottyl_seq<5'} — ingest sequence number under 5. +
    • +
  • + {'ottyl_ts>-5m'} — lines from the last 5 minutes (see Date/time formats below). +
    • +
  • + {'ottyl_fmt==json'} — the line's auto-detected text format:{' '} + log_stream (macOS log stream output — its columns are also auto-declared as + their own named params, e.g. PID/Thread, from that line's own header), json,{' '} + logfmt, or none. +
    • +
+
+ +
+

Date/time formats

+

Timestamp comparisons (e.g. against ottyl_ts) accept three forms:

+
    +
  • + relative offsets from now — 5h (5 hours from now), -3m (3 minutes ago) +
    • +
  • + absolute clock times — 12:00, 09:30:00 +
    • +
  • + full dates — 2026-01-15, 2026-01-15T09:30:00Z +
    • +
+
+ +
+

Combining clauses

+
    +
  • + {'regex("error") and ottyl_seq<5'} — lines containing "error" with sequence under 5. +
    • +
  • + {'requests[path]==/checkout and regex("5[0-9]{2}")'} — 5xx responses on one path, given a{' '} + requests metric with a path label param. +
    • +
+
+ +
+

Worked examples

+
    +
  • + Only lines mentioning "timeout" in the last 15 minutes:{' '} + {'regex("timeout") and ottyl_ts>-15m'} +
    • +
  • + Requests slower than 200ms, once a requests metric has a value-role{' '} + duration param declared: {'requests[duration]>200'} +
    • +
  • + Everything before noon on a specific date: {'ottyl_ts<2024-06-15T12:00:00Z'} +
    • +
+

+ See also Dashboards below for using ottyl_seq/ottyl_ts in a panel + query. +

+
), }, diff --git a/webapp/src/components/FilterBar.tsx b/webapp/src/components/FilterBar.tsx index b472e37..e0c21fc 100644 --- a/webapp/src/components/FilterBar.tsx +++ b/webapp/src/components/FilterBar.tsx @@ -41,6 +41,11 @@ export function FilterBar({ Filter syntax + {collapseToggle} setDraft(e.target.value)} /> - - - regex(...) -