Merge branch 'worktree-agent-a184a6929a356b613'

# Conflicts:
#	src/engine.rs
This commit is contained in:
Hugh Kaznowski
2026-09-05 13:12:35 +01:00
7 changed files with 478 additions and 17 deletions
+437 -10
View File
@@ -3,13 +3,16 @@ use crate::dto::{
BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto,
PromqlSeriesDto, StatsDto, TimelineBucketDto, TimelineParams,
};
use crate::filter_datetime;
use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
use crate::index_key::IndexKey;
use crate::metric_primitive::{MetricPrimitive, Number};
use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery};
use crate::timestamp::Timestamp;
use regex::Regex;
use serde::Deserialize;
use std::collections::BTreeMap;
use std::cmp::Ordering;
use std::collections::{BTreeMap, HashSet};
use tokio::sync::{mpsc, oneshot};
pub struct IngestedLine {
@@ -110,6 +113,13 @@ pub enum EngineResponse {
/// `Err` holds a human-readable message for a malformed query (see
/// `crate::promql::parse_query`), which the API layer maps to a 400 response.
Promql(Result<Vec<PromqlSeriesDto>, String>),
/// A query-level error, e.g. a malformed `filter` string on
/// `EngineQuery::Logs`/`Timeline` (ticket 007). Kept as a variant on the
/// existing response enum, rather than making `handle_query` return a
/// `Result`, so every query path still funnels through the same
/// `oneshot::Sender<EngineResponse>` — callers (`api.rs`) just add one
/// more match arm to turn it into a 4xx.
Error(String),
}
pub type EngineHandle = mpsc::Sender<EngineMessage>;
@@ -347,6 +357,234 @@ fn bucket_values<'a>(
.flatten()
}
/// A [`FilterExpr`] compiled against a *snapshot* of the current index, ready
/// to be evaluated per-log-line without touching the index again.
///
/// ## Design: how metric comparisons resolve against buckets
///
/// A FilterBar comparison like `my_metric[level]==5` can't be evaluated by
/// re-running `my_metric`'s regex against each log line at query time — that
/// would duplicate work the engine already did at ingest/backfill time
/// (ticket 003), and would require re-parsing the metric's declaration for
/// every query. Instead, since ticket 003 already derived, for every metric,
/// one `Bucket` per distinct label combination holding the `(sequence,
/// timestamp, _)` occurrences that produced it, resolving a metric
/// comparison is a **one-time scan of that metric's buckets per query**
/// (via [`bucket_values`]), rather than a per-log-line regex re-run:
///
/// 1. For `metric[param] <op> value`, scan every bucket for `metric` (this
/// reuses the same ticket-003 range-scan `bucket_values` already uses for
/// `GET /api/metrics/:name`), and keep the buckets whose `labels[param]`
/// satisfies `<op> value`.
/// 2. Every `(sequence, ..)` occurrence recorded in a kept bucket is a log
/// line the comparison matches. Those sequences are collected into a
/// `HashSet<u64>` once, up front — [`CompiledFilter::MetricMatch`].
/// 3. Evaluating the comparison for a given log line is then just an `O(1)`
/// `HashSet::contains(&sequence)` check, reusing the fact that a line's
/// bucket membership (which buckets a line matched) was already computed
/// when the line was ingested/backfilled, rather than recomputed here.
///
/// `sequence`/`timestamp` comparisons skip buckets entirely: they compare
/// straight against the log's own `IndexKey::IngestedLog` fields, so they're
/// compiled into their own lightweight variants ([`CompiledFilter::Sequence`]
/// / [`CompiledFilter::Timestamp`]) instead of [`CompiledFilter::MetricMatch`].
///
/// A metric comparison with no `[param]` isn't meaningful under this model —
/// v1's buckets have no single "the metric's value", only a per-label-combo
/// occurrence marker (see `apply_metric_to_log`'s doc comment) — so
/// `compile_filter` rejects it with a clear error instead of guessing.
enum CompiledFilter {
Regex(Regex),
Sequence { op: CompareOp, target: f64 },
Timestamp { op: CompareOp, target_ms: i64 },
MetricMatch(HashSet<u64>),
And(Box<CompiledFilter>, Box<CompiledFilter>),
Or(Box<CompiledFilter>, Box<CompiledFilter>),
}
/// Parses and compiles a FilterBar query string against `index`, resolving
/// every regex/metric-comparison/datetime-literal up front so evaluating it
/// per log line (see [`eval_compiled`]) never touches the index or re-parses
/// anything. `now` anchors relative datetime literals (e.g. `-5m`) so every
/// comparison in one query is resolved against the same instant.
fn compile_filter_query(
index: &BTreeMap<IndexKey, DataValue>,
filter: &str,
now: Timestamp,
) -> Result<CompiledFilter, String> {
let expr = filter_parser::parse(filter).map_err(|err| err.to_string())?;
compile_filter(index, &expr, now)
}
fn compile_filter(
index: &BTreeMap<IndexKey, DataValue>,
expr: &FilterExpr,
now: Timestamp,
) -> Result<CompiledFilter, String> {
match expr {
FilterExpr::Regex(pattern) => Regex::new(pattern)
.map(CompiledFilter::Regex)
.map_err(|err| format!("invalid regex in filter: {err}")),
FilterExpr::Comparison { metric, param, op, value } => match metric.as_str() {
"sequence" => compile_sequence_comparison(*op, value),
"timestamp" => compile_timestamp_comparison(*op, value, now),
other => compile_metric_comparison(index, other, param.as_deref(), *op, value, now),
},
FilterExpr::And(lhs, rhs) => Ok(CompiledFilter::And(
Box::new(compile_filter(index, lhs, now)?),
Box::new(compile_filter(index, rhs, now)?),
)),
FilterExpr::Or(lhs, rhs) => Ok(CompiledFilter::Or(
Box::new(compile_filter(index, lhs, now)?),
Box::new(compile_filter(index, rhs, now)?),
)),
}
}
/// `sequence <op> value` — compares straight against the log's own sequence
/// number. `value` is normally `FilterValue::Number` (the grammar parses a
/// bare integer as a number); a `Raw` value is accepted too as long as it
/// still parses as a number (defensive - the grammar shouldn't produce this
/// for `sequence`, but nothing stops a caller from writing `sequence<"5"`-ish
/// input that the value token still captures as text).
fn compile_sequence_comparison(op: CompareOp, value: &FilterValue) -> Result<CompiledFilter, String> {
let target = match value {
FilterValue::Number(n) => *n,
FilterValue::Raw(raw) => raw.parse::<f64>().map_err(|_| {
format!("'sequence' comparisons require a numeric value, got {raw:?}")
})?,
};
Ok(CompiledFilter::Sequence { op, target })
}
/// `timestamp <op> value` — compares straight against the log's own
/// timestamp. A `FilterValue::Number` is treated as raw milliseconds (no
/// datetime parsing needed); a `FilterValue::Raw` token (e.g. `12:00`,
/// `-5m`, `2024-01-15`) is resolved via `filter_datetime::parse_datetime`
/// against `now`.
fn compile_timestamp_comparison(
op: CompareOp,
value: &FilterValue,
now: Timestamp,
) -> Result<CompiledFilter, String> {
let target_ms = match value {
FilterValue::Number(n) => *n as i64,
FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, now)
.map_err(|err| err.to_string())?
.as_millis(),
};
Ok(CompiledFilter::Timestamp { op, target_ms })
}
/// `metric[param] <op> value` for any metric name other than the reserved
/// `sequence`/`timestamp` — see [`CompiledFilter`]'s doc comment for the
/// overall approach.
fn compile_metric_comparison(
index: &BTreeMap<IndexKey, DataValue>,
metric: &str,
param: Option<&str>,
op: CompareOp,
value: &FilterValue,
now: Timestamp,
) -> Result<CompiledFilter, String> {
let param = param.ok_or_else(|| {
format!(
"metric comparison on '{metric}' requires a [param] (e.g. '{metric}[some_param]{}...'); \
v1 buckets only carry per-label occurrence data, not a single metric value",
match op {
CompareOp::Lt => "<",
CompareOp::Gt => ">",
CompareOp::Le => "<=",
CompareOp::Ge => ">=",
CompareOp::Eq => "==",
CompareOp::Ne => "!=",
}
)
})?;
let mut matching = HashSet::new();
for (labels, (sequence, _, _)) in bucket_values(index, metric, None) {
let Some(primitive) = labels.get(param) else {
continue;
};
if compare_metric_primitive(primitive, op, value, now)? {
matching.insert(*sequence);
}
}
Ok(CompiledFilter::MetricMatch(matching))
}
/// Compares one bucket label's `MetricPrimitive` against a filter's
/// right-hand side. Numbers compare numerically (a `Raw` value is parsed as
/// a number); timestamps compare against a resolved datetime (a `Number`
/// value is treated as raw milliseconds, a `Raw` value goes through
/// `filter_datetime::parse_datetime`); strings compare lexicographically
/// against the raw token (or a number's textual form). Mismatched kinds that
/// can't be meaningfully compared (e.g. a string label against a bare
/// number) simply never match, rather than erroring.
fn compare_metric_primitive(
primitive: &MetricPrimitive,
op: CompareOp,
value: &FilterValue,
now: Timestamp,
) -> Result<bool, String> {
let ordering = match (primitive, value) {
(MetricPrimitive::Number(n), FilterValue::Number(v)) => n.as_f64().partial_cmp(v),
(MetricPrimitive::Number(n), FilterValue::Raw(raw)) => {
let v: f64 = raw
.parse()
.map_err(|_| format!("expected a numeric value, got {raw:?}"))?;
n.as_f64().partial_cmp(&v)
}
(MetricPrimitive::Timestamp(ts), FilterValue::Number(v)) => {
ts.as_millis().partial_cmp(&(*v as i64))
}
(MetricPrimitive::Timestamp(ts), FilterValue::Raw(raw)) => {
let target = filter_datetime::parse_datetime(raw, now).map_err(|err| err.to_string())?;
ts.as_millis().partial_cmp(&target.as_millis())
}
(MetricPrimitive::String(s), FilterValue::Raw(raw)) => s.as_str().partial_cmp(raw.as_str()),
(MetricPrimitive::String(s), FilterValue::Number(n)) => {
s.as_str().partial_cmp(n.to_string().as_str())
}
};
Ok(ordering.is_some_and(|ordering| apply_compare_op(ordering, op)))
}
fn apply_compare_op(ordering: Ordering, op: CompareOp) -> bool {
match op {
CompareOp::Lt => ordering == Ordering::Less,
CompareOp::Gt => ordering == Ordering::Greater,
CompareOp::Le => ordering != Ordering::Greater,
CompareOp::Ge => ordering != Ordering::Less,
CompareOp::Eq => ordering == Ordering::Equal,
CompareOp::Ne => ordering != Ordering::Equal,
}
}
/// Evaluates a compiled filter against one log line. Cheap: every regex,
/// datetime literal, and bucket scan was already resolved by
/// [`compile_filter_query`], so this is just pattern matching, numeric
/// comparisons, and `HashSet` lookups — no index access.
fn eval_compiled(compiled: &CompiledFilter, sequence: u64, timestamp: Timestamp, text: &str) -> bool {
match compiled {
CompiledFilter::Regex(re) => re.is_match(text),
CompiledFilter::Sequence { op, target } => {
apply_compare_op((sequence as f64).total_cmp(target), *op)
}
CompiledFilter::Timestamp { op, target_ms } => {
apply_compare_op(timestamp.as_millis().cmp(target_ms), *op)
}
CompiledFilter::MetricMatch(sequences) => sequences.contains(&sequence),
CompiledFilter::And(lhs, rhs) => {
eval_compiled(lhs, sequence, timestamp, text) && eval_compiled(rhs, sequence, timestamp, text)
}
CompiledFilter::Or(lhs, rhs) => {
eval_compiled(lhs, sequence, timestamp, text) || eval_compiled(rhs, sequence, timestamp, text)
}
}
}
fn metric_declarations_range(
index: &BTreeMap<IndexKey, DataValue>,
) -> impl Iterator<Item = (&str, &str, &[RegexToMetricPrimitiveMatcher])> {
@@ -432,6 +670,13 @@ fn logs_range(index: &BTreeMap<IndexKey, DataValue>) -> impl Iterator<Item = (u6
fn handle_query(index: &BTreeMap<IndexKey, DataValue>, query: EngineQuery) -> EngineResponse {
match query {
EngineQuery::Logs(params) => {
let compiled_filter = match params.filter.as_deref() {
Some(filter) => match compile_filter_query(index, filter, Timestamp::now()) {
Ok(compiled) => Some(compiled),
Err(err) => return EngineResponse::Error(err),
},
None => None,
};
let limit = params.limit.unwrap_or(500).min(10_000);
// Compile each currently-declared param's regex once up front, then
// apply the compiled set to every *returned* line below (ticket 015).
@@ -450,6 +695,11 @@ fn handle_query(index: &BTreeMap<IndexKey, DataValue>, query: EngineQuery) -> En
&& params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
&& params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
})
.filter(|(sequence, timestamp, text)| {
compiled_filter
.as_ref()
.is_none_or(|filter| eval_compiled(filter, *sequence, *timestamp, text))
})
.take(limit)
.map(|(sequence, timestamp, text)| LogEntryDto {
sequence,
@@ -461,12 +711,25 @@ fn handle_query(index: &BTreeMap<IndexKey, DataValue>, query: EngineQuery) -> En
EngineResponse::Logs(entries)
}
EngineQuery::Timeline(params) => {
let compiled_filter = match params.filter.as_deref() {
Some(filter) => match compile_filter_query(index, filter, Timestamp::now()) {
Ok(compiled) => Some(compiled),
Err(err) => return EngineResponse::Error(err),
},
None => None,
};
let bucket_ms = params.bucket_ms.unwrap_or(1_000).max(1);
let mut buckets: BTreeMap<i64, u64> = BTreeMap::new();
for (_, timestamp, _) in logs_range(index).filter(|(_, timestamp, _)| {
for (sequence, timestamp, text) in logs_range(index).filter(|(_, timestamp, _)| {
params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
&& params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
}) {
if compiled_filter
.as_ref()
.is_some_and(|filter| !eval_compiled(filter, sequence, timestamp, text))
{
continue;
}
let bucket_start = (timestamp.as_millis() / bucket_ms) * bucket_ms;
*buckets.entry(bucket_start).or_insert(0) += 1;
}
@@ -1141,6 +1404,125 @@ mod tests {
}
}
// -- ticket 007: FilterBar query integration -----------------------
fn logs_query(filter: Option<&str>) -> EngineQuery {
EngineQuery::Logs(LogsParams {
since_sequence: None,
start_ms: None,
end_ms: None,
limit: None,
filter: filter.map(str::to_string),
})
}
fn timeline_query(filter: Option<&str>, bucket_ms: i64) -> EngineQuery {
EngineQuery::Timeline(TimelineParams {
start_ms: None,
end_ms: None,
bucket_ms: Some(bucket_ms),
filter: filter.map(str::to_string),
})
}
#[test]
fn logs_query_with_regex_filter_returns_only_matching_lines() {
let mut index = BTreeMap::new();
ingest_log(&mut index, 0, "everything is fine");
ingest_log(&mut index, 1, "an error occurred");
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")"#))) {
EngineResponse::Logs(entries) => {
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
assert_eq!(sequences, vec![1, 3]);
}
_ => panic!("expected Logs response"),
}
}
#[test]
fn logs_query_with_sequence_comparison_filters_by_sequence() {
let mut index = BTreeMap::new();
for i in 0..5u64 {
ingest_log(&mut index, i, &format!("line {i}"));
}
match handle_query(&index, logs_query(Some("sequence<2"))) {
EngineResponse::Logs(entries) => {
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
assert_eq!(sequences, vec![0, 1]);
}
_ => panic!("expected Logs response"),
}
}
#[test]
fn logs_query_with_timestamp_comparison_filters_by_timestamp() {
let mut index = BTreeMap::new();
for i in 0..5u64 {
let timestamp = Timestamp::from_millis(i as i64 * 1000);
index.insert(
IndexKey::IngestedLog { sequence: i, timestamp },
DataValue::LogLiteral { text: format!("line {i}") },
);
}
match handle_query(&index, logs_query(Some("timestamp>2000"))) {
EngineResponse::Logs(entries) => {
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
assert_eq!(sequences, vec![3, 4]);
}
_ => panic!("expected Logs response"),
}
}
#[test]
fn logs_query_combines_regex_and_metric_comparison_with_and() {
let mut index = BTreeMap::new();
declare_metric(
&mut index,
metric_request(
"requests",
r"level=(?P<level>\w+) duration=(?P<duration>\d+)",
vec![("level", "level", MetricPrimitiveKind::String)],
),
)
.unwrap();
ingest_log(&mut index, 0, "level=error duration=1 boom");
ingest_log(&mut index, 1, "level=info duration=2 boom");
ingest_log(&mut index, 2, "level=error duration=3 all good");
// Only sequence 0 has both the "boom" regex match *and* level=error.
// Note: comparison values are bare tokens (see `value` in the
// grammar), not quoted string literals like `regex(...)`'s argument
// — so `level==error`, not `level=="error"`.
match handle_query(
&index,
logs_query(Some(r#"regex("boom") and requests[level]==error"#)),
) {
EngineResponse::Logs(entries) => {
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
assert_eq!(sequences, vec![0]);
}
_ => panic!("expected Logs response"),
}
// With `or`, sequences 0 (boom) and 2 (level=error) both match.
match handle_query(
&index,
logs_query(Some(r#"regex("boom") or requests[level]==error"#)),
) {
EngineResponse::Logs(entries) => {
let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
assert_eq!(sequences, vec![0, 1, 2]);
}
_ => panic!("expected Logs response"),
}
}
#[test]
fn logs_query_includes_declared_params_on_matching_lines_only() {
let mut index = BTreeMap::new();
@@ -1148,14 +1530,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, EngineQuery::Logs(LogsParams {
since_sequence: None,
start_ms: None,
end_ms: None,
limit: None,
}))
else {
let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None)) else {
panic!("expected Logs response");
};
@@ -1166,4 +1541,56 @@ mod tests {
);
assert!(entries[1].params.get("level").is_none());
}
#[test]
fn timeline_query_applies_filter_before_bucketing() {
let mut index = BTreeMap::new();
ingest_log(&mut index, 0, "everything is fine");
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)) {
EngineResponse::Timeline(buckets) => {
let total: u64 = buckets.iter().map(|b| b.count).sum();
assert_eq!(total, 2);
}
_ => panic!("expected Timeline response"),
}
}
#[test]
fn malformed_filter_string_surfaces_as_engine_error() {
let mut index = BTreeMap::new();
ingest_log(&mut index, 0, "hello");
match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@"))) {
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)) {
EngineResponse::Error(message) => assert!(!message.is_empty()),
_ => panic!("expected an Error response for a malformed filter string"),
}
}
#[test]
fn metric_comparison_without_param_is_a_compile_error() {
let mut index = BTreeMap::new();
declare_metric(
&mut index,
metric_request(
"requests",
r"level=(?P<level>\w+)",
vec![("level", "level", MetricPrimitiveKind::String)],
),
)
.unwrap();
ingest_log(&mut index, 0, "level=error");
match handle_query(&index, logs_query(Some("requests==5"))) {
EngineResponse::Error(message) => assert!(message.contains("param")),
_ => panic!("expected an Error response for a metric comparison without [param]"),
}
}
}