phughk/ottyl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Resolve tickets 071-074, 076: log-stream contamination, sort, ottyl_fmt column, ingest lag

Browse Source 
- 076: gate auto-declared log-stream column/message params
  (`ParamDeclaration::log_stream_gated`) behind `is_log_stream_line` at
  extraction time, since their fixed-offset regex has no content anchor
  and previously extracted garbage from continuation lines and the
  header row itself (the regex crate doesn't support lookaround, so
  this couldn't be baked into the regex text).
- 074: end-to-end regression test confirms sorting a real `log stream`
  capture by a log-stream column (e.g. PID) now actually reorders
  results, since 076's fix removes the garbage values that were
  corrupting the sort helper index; also add a "Building sort
  index…" indicator in the timeline UI for a slow first-time sort.
- 072: attach the reserved `ottyl_fmt` pseudo-param to each
  `LogEntryDto`'s params so it can be picked as a Log Preview column
  (previously listed as pickable but always blank).
- 073: track a rolling 5s high-watermark of ingestion-to-engine-processing
  lag and surface it in the nav stats popover next to approx_bytes_total.
- 071: group the regex(...) help button back with the filter bar's other
  controls, and restructure the Filtering docs section into scannable
  per-feature subsections (also fixes stale log_type/process/pid mention).

Tickets 070 (CLI scoping) and 075 (API unresponsiveness investigation)
are left in place per their own scope — both are explicitly
conversation/investigation tasks, not implementation tickets.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS
This commit is contained in:
Hugh Kaznowski
2026-09-06 15:12:07 +01:00
co-authored by Claude Sonnet 5
parent f34dbba6c7
commit 9fc7540a5b
16 changed files with 511 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/engine.rs
+296 -45
View File
@@ -171,11 +171,20 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, 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<crate::logfmt::LogStreamHeader> = 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<EngineMessage>, 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<IndexKey, DataValue>, 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<IndexKey, DataValue>) -> HashMap<String, ResolvedParam> {
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<IndexKey, DataValue>) -> 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<IndexKey, DataValue>) -> 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<String> {
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<IndexKey, DataValue>,
) -> impl Iterator<Item = (&str, &str, &str, Option<&str>, MetricPrimitiveKind, bool, bool)> {
) -> impl Iterator<Item = (&str, &str, &str, Option<&str>, 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<String>, MetricPrimitiveKind, bool);
fn derive_single_param(
text: &str,
declarations: &[(String, Regex, String, Option<String>, MetricPrimitiveKind)],
declarations: &[CompiledParamDeclaration],
param_name: &str,
) -> Option<MetricPrimitive> {
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<String, Sor
sort_helper_indexes.retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER);
}
fn compiled_param_declarations(
index: &BTreeMap<IndexKey, DataValue>,
) -> Vec<(String, Regex, String, Option<String>, MetricPrimitiveKind)> {
fn compiled_param_declarations(index: &BTreeMap<IndexKey, DataValue>) -> Vec<CompiledParamDeclaration> {
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<String>, MetricPrimitiveKind)],
) -> BTreeMap<String, MetricPrimitive> {
fn derive_log_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap<String, MetricPrimitive> {
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<String, MetricPrimitive> {
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<IndexKey, DataValue>,
query: EngineQuery,
approx_bytes_total: u64,
ingest_lag_high_watermark_ms: i64,
sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
) -> 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<i64> = 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<u64> = 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<IndexKey, DataValue>, 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<Option<&MetricPrimitive>> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = entries.iter().map(|e| e.sequence).collect();
assert_eq!(sequences, vec![0]);