Convert engine.rs index-manipulating functions into EngineSnapshot methods

Free functions that took the engine's index/sort-helper-index state as
a parameter now live on EngineSnapshot (or, for a handful of clean
single-receiver cases, on MetricPrimitive/CompareOp/CompiledFilter/
EngineQuery). Route handlers in api.rs and alert evaluation in
alerts.rs are updated to call the new methods; alerts.rs now threads
an &EngineSnapshot through instead of a raw &BTreeMap so the per-line
alert backfill path doesn't clone the whole index just to call a method.

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 17:03:19 +01:00
co-authored by Claude Sonnet 5
parent b6401c1c3c
commit 529f79e9da
4 changed files with 1241 additions and 1289 deletions
+23 -25
View File
@@ -1,5 +1,5 @@
use std::collections::hash_map::DefaultHasher; use std::collections::hash_map::DefaultHasher;
use std::collections::{BTreeMap, HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::fmt; use std::fmt;
use std::hash::{Hash, Hasher}; use std::hash::{Hash, Hasher};
use std::io; use std::io;
@@ -9,11 +9,9 @@ use std::time::{SystemTime, UNIX_EPOCH};
use regex::Regex; use regex::Regex;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use crate::data_value::DataValue; use crate::engine::{self, eval_json_comparison};
use crate::engine::{self, apply_compare_op, eval_json_comparison};
use crate::filter_datetime; use crate::filter_datetime;
use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
use crate::index_key::IndexKey;
use crate::rate::RateWindow; use crate::rate::RateWindow;
use crate::settings::Settings; use crate::settings::Settings;
use crate::timestamp::Timestamp; use crate::timestamp::Timestamp;
@@ -142,7 +140,7 @@ pub struct AlertRuntimeState {
} }
fn backfill_window( fn backfill_window(
index: &BTreeMap<IndexKey, DataValue>, snapshot: &engine::EngineSnapshot,
query: &str, query: &str,
before_sequence: u64, before_sequence: u64,
now: Timestamp, now: Timestamp,
@@ -150,7 +148,7 @@ fn backfill_window(
) -> RateWindow { ) -> RateWindow {
let since_ms = now.as_millis() - window_ms; let since_ms = now.as_millis() - window_ms;
let matching_timestamps = let matching_timestamps =
engine::logs_since(index, before_sequence, since_ms, now.as_millis()).filter_map( snapshot.logs_since(before_sequence, since_ms, now.as_millis()).filter_map(
|(sequence, timestamp, text)| { |(sequence, timestamp, text)| {
(eval_alert_query(query, sequence, timestamp, text) == Ok(true)).then_some(timestamp) (eval_alert_query(query, sequence, timestamp, text) == Ok(true)).then_some(timestamp)
}, },
@@ -176,7 +174,7 @@ fn eval_alert_expr(expr: &FilterExpr, sequence: u64, timestamp: Timestamp, text:
.parse::<f64>() .parse::<f64>()
.map_err(|_| format!("'ottyl_seq' comparisons require a numeric value, got {raw:?}"))?, .map_err(|_| format!("'ottyl_seq' comparisons require a numeric value, got {raw:?}"))?,
}; };
Ok(apply_compare_op((sequence as f64).total_cmp(&target), *op)) Ok(op.matches((sequence as f64).total_cmp(&target)))
} }
FilterExpr::Comparison { metric, param: None, op, value } if metric == "ottyl_ts" => { FilterExpr::Comparison { metric, param: None, op, value } if metric == "ottyl_ts" => {
let target_ms = match value { let target_ms = match value {
@@ -185,14 +183,14 @@ fn eval_alert_expr(expr: &FilterExpr, sequence: u64, timestamp: Timestamp, text:
.map_err(|err| err.to_string())? .map_err(|err| err.to_string())?
.as_millis(), .as_millis(),
}; };
Ok(apply_compare_op(timestamp.as_millis().cmp(&target_ms), *op)) Ok(op.matches(timestamp.as_millis().cmp(&target_ms)))
} }
FilterExpr::Comparison { metric, param: None, op, value } if metric == "ottyl_fmt" => { FilterExpr::Comparison { metric, param: None, op, value } if metric == "ottyl_fmt" => {
let target = match value { let target = match value {
FilterValue::Number(n) => n.to_string(), FilterValue::Number(n) => n.to_string(),
FilterValue::Raw(raw) => raw.clone(), FilterValue::Raw(raw) => raw.clone(),
}; };
Ok(apply_compare_op(crate::logfmt::detect_format(text).cmp(target.as_str()), *op)) Ok(op.matches(crate::logfmt::detect_format(text).cmp(target.as_str())))
} }
FilterExpr::Comparison { metric, .. } => Err(format!( FilterExpr::Comparison { metric, .. } => Err(format!(
"alert queries only support regex()/ottyl_seq/ottyl_ts/ottyl_fmt (not metric comparisons like \ "alert queries only support regex()/ottyl_seq/ottyl_ts/ottyl_fmt (not metric comparisons like \
@@ -213,7 +211,7 @@ fn eval_alert_expr(expr: &FilterExpr, sequence: u64, timestamp: Timestamp, text:
pub fn evaluate_alerts( pub fn evaluate_alerts(
rules: &[AlertRule], rules: &[AlertRule],
states: &mut HashMap<String, AlertRuntimeState>, states: &mut HashMap<String, AlertRuntimeState>,
index: &BTreeMap<IndexKey, DataValue>, snapshot: &engine::EngineSnapshot,
sequence: u64, sequence: u64,
timestamp: Timestamp, timestamp: Timestamp,
text: &str, text: &str,
@@ -225,7 +223,7 @@ pub fn evaluate_alerts(
} }
let state = states.entry(rule.id.clone()).or_default(); let state = states.entry(rule.id.clone()).or_default();
if let Some(message) = if let Some(message) =
evaluate_condition(&rule.condition, state, index, sequence, timestamp, text, &rule.name) evaluate_condition(&rule.condition, state, snapshot, sequence, timestamp, text, &rule.name)
{ {
fired.push((rule.clone(), message)); fired.push((rule.clone(), message));
} }
@@ -236,7 +234,7 @@ pub fn evaluate_alerts(
fn evaluate_condition( fn evaluate_condition(
condition: &AlertCondition, condition: &AlertCondition,
state: &mut AlertRuntimeState, state: &mut AlertRuntimeState,
index: &BTreeMap<IndexKey, DataValue>, snapshot: &engine::EngineSnapshot,
sequence: u64, sequence: u64,
timestamp: Timestamp, timestamp: Timestamp,
text: &str, text: &str,
@@ -253,14 +251,14 @@ fn evaluate_condition(
}, },
AlertCondition::Threshold { query, op, threshold, window_ms } => { AlertCondition::Threshold { query, op, threshold, window_ms } => {
if !state.threshold_backfilled { if !state.threshold_backfilled {
state.threshold_window = backfill_window(index, query, sequence, timestamp, *window_ms); state.threshold_window = backfill_window(snapshot, query, sequence, timestamp, *window_ms);
state.threshold_backfilled = true; state.threshold_backfilled = true;
} }
match eval_alert_query(query, sequence, timestamp, text) { match eval_alert_query(query, sequence, timestamp, text) {
Ok(true) => { Ok(true) => {
state.threshold_window.push(timestamp, *window_ms); state.threshold_window.push(timestamp, *window_ms);
let count = state.threshold_window.count() as f64; let count = state.threshold_window.count() as f64;
apply_compare_op(count.total_cmp(threshold), *op).then(|| { op.matches(count.total_cmp(threshold)).then(|| {
format!("{rule_name}: match count {count} {op:?} threshold {threshold} (window {window_ms}ms)") format!("{rule_name}: match count {count} {op:?} threshold {threshold} (window {window_ms}ms)")
}) })
} }
@@ -300,11 +298,11 @@ fn evaluate_condition(
return None; return None;
} }
if !state.ratio_a_backfilled { if !state.ratio_a_backfilled {
state.ratio_a_window = backfill_window(index, query_a, sequence, timestamp, *window_ms); state.ratio_a_window = backfill_window(snapshot, query_a, sequence, timestamp, *window_ms);
state.ratio_a_backfilled = true; state.ratio_a_backfilled = true;
} }
if !state.ratio_b_backfilled { if !state.ratio_b_backfilled {
state.ratio_b_window = backfill_window(index, query_b, sequence, timestamp, *window_ms); state.ratio_b_window = backfill_window(snapshot, query_b, sequence, timestamp, *window_ms);
state.ratio_b_backfilled = true; state.ratio_b_backfilled = true;
} }
if matched_a { if matched_a {
@@ -319,7 +317,7 @@ fn evaluate_condition(
return None; return None;
} }
let observed = count_a / count_b; let observed = count_a / count_b;
apply_compare_op(observed.total_cmp(ratio), *op).then(|| { op.matches(observed.total_cmp(ratio)).then(|| {
format!( format!(
"{rule_name}: ratio {observed:.3} ({count_a}/{count_b}) {op:?} {ratio} (window {window_ms}ms)" "{rule_name}: ratio {observed:.3} ({count_a}/{count_b}) {op:?} {ratio} (window {window_ms}ms)"
) )
@@ -431,7 +429,7 @@ mod tests {
fn immediate_fires_on_matching_line_only() { fn immediate_fires_on_matching_line_only() {
let rules = vec![immediate_rule(r#"regex("error")"#)]; let rules = vec![immediate_rule(r#"regex("error")"#)];
let mut states = HashMap::new(); let mut states = HashMap::new();
let index: BTreeMap<IndexKey, DataValue> = BTreeMap::new(); let index = engine::EngineSnapshot::default();
let (seq, ts, text) = line("all good", 0, 0); let (seq, ts, text) = line("all good", 0, 0);
assert!(evaluate_alerts(&rules, &mut states, &index, seq, ts, text).is_empty()); assert!(evaluate_alerts(&rules, &mut states, &index, seq, ts, text).is_empty());
@@ -447,7 +445,7 @@ mod tests {
let mut rule = immediate_rule(r#"regex("error")"#); let mut rule = immediate_rule(r#"regex("error")"#);
rule.enabled = false; rule.enabled = false;
let mut states = HashMap::new(); let mut states = HashMap::new();
let index: BTreeMap<IndexKey, DataValue> = BTreeMap::new(); let index = engine::EngineSnapshot::default();
let (seq, ts, text) = line("error!", 0, 0); let (seq, ts, text) = line("error!", 0, 0);
assert!(evaluate_alerts(&[rule], &mut states, &index, seq, ts, text).is_empty()); assert!(evaluate_alerts(&[rule], &mut states, &index, seq, ts, text).is_empty());
} }
@@ -466,7 +464,7 @@ mod tests {
}, },
}; };
let mut states = HashMap::new(); let mut states = HashMap::new();
let index: BTreeMap<IndexKey, DataValue> = BTreeMap::new(); let index = engine::EngineSnapshot::default();
for (seq, ms) in [(0u64, 0i64), (1, 1000)] { for (seq, ms) in [(0u64, 0i64), (1, 1000)] {
let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, seq, Timestamp::from_millis(ms), "error"); let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, seq, Timestamp::from_millis(ms), "error");
assert!(fired.is_empty(), "should not fire before threshold is crossed"); assert!(fired.is_empty(), "should not fire before threshold is crossed");
@@ -491,7 +489,7 @@ mod tests {
}, },
}; };
let mut states = HashMap::new(); let mut states = HashMap::new();
let index: BTreeMap<IndexKey, DataValue> = BTreeMap::new(); let index = engine::EngineSnapshot::default();
assert!(evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "error").is_empty()); assert!(evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "error").is_empty());
assert!(evaluate_alerts(&[rule], &mut states, &index, 1, Timestamp::from_millis(5_000), "error").is_empty()); assert!(evaluate_alerts(&[rule], &mut states, &index, 1, Timestamp::from_millis(5_000), "error").is_empty());
} }
@@ -508,7 +506,7 @@ mod tests {
}, },
}; };
let mut states = HashMap::new(); let mut states = HashMap::new();
let index: BTreeMap<IndexKey, DataValue> = BTreeMap::new(); let index = engine::EngineSnapshot::default();
let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "agent=curl"); let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "agent=curl");
assert_eq!(fired.len(), 1); assert_eq!(fired.len(), 1);
let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 1, Timestamp::from_millis(1), "agent=curl"); let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 1, Timestamp::from_millis(1), "agent=curl");
@@ -532,7 +530,7 @@ mod tests {
}, },
}; };
let mut states = HashMap::new(); let mut states = HashMap::new();
let index: BTreeMap<IndexKey, DataValue> = BTreeMap::new(); let index = engine::EngineSnapshot::default();
assert!(evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "error").is_empty()); assert!(evaluate_alerts(&[rule.clone()], &mut states, &index, 0, Timestamp::from_millis(0), "error").is_empty());
let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 1, Timestamp::from_millis(1), "request"); let fired = evaluate_alerts(&[rule.clone()], &mut states, &index, 1, Timestamp::from_millis(1), "request");
assert_eq!(fired.len(), 1); assert_eq!(fired.len(), 1);
@@ -546,7 +544,7 @@ mod tests {
fn ottyl_fmt_alert_query_matches_detected_format() { fn ottyl_fmt_alert_query_matches_detected_format() {
let rule = immediate_rule("ottyl_fmt==json"); let rule = immediate_rule("ottyl_fmt==json");
let mut states = HashMap::new(); let mut states = HashMap::new();
let index: BTreeMap<IndexKey, DataValue> = BTreeMap::new(); let index = engine::EngineSnapshot::default();
let fired = evaluate_alerts( let fired = evaluate_alerts(
&[rule.clone()], &[rule.clone()],
@@ -566,7 +564,7 @@ mod tests {
fn metric_comparison_in_alert_query_is_rejected_not_panicking() { fn metric_comparison_in_alert_query_is_rejected_not_panicking() {
let rule = immediate_rule("some_metric[param]==5"); let rule = immediate_rule("some_metric[param]==5");
let mut states = HashMap::new(); let mut states = HashMap::new();
let index: BTreeMap<IndexKey, DataValue> = BTreeMap::new(); let index = engine::EngineSnapshot::default();
let fired = evaluate_alerts(&[rule], &mut states, &index, 0, Timestamp::from_millis(0), "anything"); let fired = evaluate_alerts(&[rule], &mut states, &index, 0, Timestamp::from_millis(0), "anything");
assert!(fired.is_empty()); assert!(fired.is_empty());
} }
+3 -3
View File
@@ -217,9 +217,9 @@ fn mime_for_extension(extension: &str) -> &'static str {
async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse, StatusCode> { async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse, StatusCode> {
let read_handle = engine.reader_factory.handle(); let read_handle = engine.reader_factory.handle();
if let Some(sort_by) = crate::engine::query_sort_by(&query) { if let Some(sort_by) = query.sort_by() {
let needs_build = match read_handle.enter() { let needs_build = match read_handle.enter() {
Some(guard) => !crate::engine::has_sort_helper_index(&guard, sort_by), Some(guard) => !guard.has_sort_helper_index(sort_by),
None => return Err(StatusCode::SERVICE_UNAVAILABLE), None => return Err(StatusCode::SERVICE_UNAVAILABLE),
}; };
if needs_build { if needs_build {
@@ -233,7 +233,7 @@ async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse
} }
} }
let guard = read_handle.enter().ok_or(StatusCode::SERVICE_UNAVAILABLE)?; let guard = read_handle.enter().ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
Ok(crate::engine::handle_query_ro(&guard, query)) Ok(guard.handle_query_ro(query))
} }
async fn promoted_param_records( async fn promoted_param_records(
+643 -689
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -700,7 +700,7 @@ mod tests {
async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse { async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse {
let read_handle = engine.reader_factory.handle(); let read_handle = engine.reader_factory.handle();
let guard = read_handle.enter().expect("engine unavailable"); let guard = read_handle.enter().expect("engine unavailable");
crate::engine::handle_query_ro(&guard, query) guard.handle_query_ro(query)
} }
#[test] #[test]