Wire FilterBar query AST into logs/timeline queries (ticket 007)
Adds a filter evaluator in engine.rs that compiles a parsed FilterExpr against the current index once per query (regex compiled once, sequence/timestamp compared directly against IngestedLog fields, other metric comparisons resolved to a HashSet<u64> of matching sequences via a bucket scan) and evaluates it per log line with no further index access. Wires this into EngineQuery::Logs/Timeline via a new LogsParams::filter/TimelineParams::filter field, and adds EngineResponse::Error so a malformed filter string surfaces as a query-level error (mapped to 400 in api.rs). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
49e8590cf3
commit
86a1a40243
+13
@@ -122,23 +122,36 @@ async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse
|
||||
response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)
|
||||
}
|
||||
|
||||
/// `GET /api/logs` — also accepts a `filter` query param (ticket 007): a
|
||||
/// FilterBar query string, parsed and evaluated inside the engine. A
|
||||
/// malformed filter string comes back as `EngineResponse::Error` and is
|
||||
/// surfaced here as `400 Bad Request` with `{ "error": "..." }`, matching
|
||||
/// ticket 004's `declare_param`/`declare_metric` error-mapping pattern.
|
||||
async fn get_logs(
|
||||
State(state): State<ApiState>,
|
||||
Query(params): Query<LogsParams>,
|
||||
) -> impl IntoResponse {
|
||||
match ask(&state.engine, EngineQuery::Logs(params)).await {
|
||||
Ok(EngineResponse::Logs(entries)) => Json(entries).into_response(),
|
||||
Ok(EngineResponse::Error(err)) => {
|
||||
(StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response()
|
||||
}
|
||||
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|
||||
Err(status) => status.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// `GET /api/timeline` — same `filter` query param and error mapping as
|
||||
/// `get_logs` above.
|
||||
async fn get_timeline(
|
||||
State(state): State<ApiState>,
|
||||
Query(params): Query<TimelineParams>,
|
||||
) -> impl IntoResponse {
|
||||
match ask(&state.engine, EngineQuery::Timeline(params)).await {
|
||||
Ok(EngineResponse::Timeline(buckets)) => Json(buckets).into_response(),
|
||||
Ok(EngineResponse::Error(err)) => {
|
||||
(StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response()
|
||||
}
|
||||
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|
||||
Err(status) => status.into_response(),
|
||||
}
|
||||
|
||||
@@ -47,6 +47,10 @@ pub struct LogsParams {
|
||||
pub start_ms: Option<i64>,
|
||||
pub end_ms: Option<i64>,
|
||||
pub limit: Option<usize>,
|
||||
/// Raw FilterBar query string (ticket 007), e.g. `regex("error") and
|
||||
/// sequence<5`. Parsed with `filter_parser::parse` and evaluated in the
|
||||
/// engine; a malformed string surfaces as `EngineResponse::Error`.
|
||||
pub filter: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
@@ -54,6 +58,9 @@ pub struct TimelineParams {
|
||||
pub start_ms: Option<i64>,
|
||||
pub end_ms: Option<i64>,
|
||||
pub bucket_ms: Option<i64>,
|
||||
/// Same FilterBar query string as `LogsParams::filter`: only logs
|
||||
/// matching it are counted into timeline buckets.
|
||||
pub filter: Option<String>,
|
||||
}
|
||||
|
||||
/// One entry from a metric's derived bucket timeseries: the label combination
|
||||
|
||||
+436
-2
@@ -3,12 +3,15 @@ use crate::dto::{
|
||||
BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, 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::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 {
|
||||
@@ -99,6 +102,13 @@ pub enum EngineResponse {
|
||||
ParamDeclarations(Vec<ParamDeclarationDto>),
|
||||
MetricDeclarations(Vec<MetricDeclarationDto>),
|
||||
BucketValues(Vec<BucketValueDto>),
|
||||
/// 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>;
|
||||
@@ -336,6 +346,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])> {
|
||||
@@ -378,6 +616,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);
|
||||
let entries: Vec<LogEntryDto> = logs_range(index)
|
||||
.filter(|(sequence, timestamp, _)| {
|
||||
@@ -385,6 +630,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,
|
||||
@@ -395,12 +645,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;
|
||||
}
|
||||
@@ -835,4 +1098,175 @@ mod tests {
|
||||
assert_eq!(bucket_values(&index, "requests", None).count(), 1);
|
||||
assert_eq!(bucket_values(&index, "durations", None).count(), 1);
|
||||
}
|
||||
|
||||
// -- 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 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]"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,8 +15,10 @@
|
||||
//! 3. Full date/datetime literals, e.g. `2024-01-15` or `2024-01-15T12:00:00Z`
|
||||
//! (RFC3339-ish). A bare date is interpreted as midnight UTC on that date.
|
||||
|
||||
// Not yet wired into the filter grammar/engine (a separate integration ticket); allow
|
||||
// dead code so this standalone module builds warning-free until that lands.
|
||||
// `engine.rs` (ticket 007) calls `parse_datetime` to resolve `FilterValue::Raw`
|
||||
// tokens (e.g. `-5m`, `12:00`, a full date) in `timestamp`/metric-comparison
|
||||
// filter clauses; `#![allow(dead_code)]` stays since most of this module's
|
||||
// surface is still only exercised by its own unit tests below.
|
||||
#![allow(dead_code)]
|
||||
|
||||
use chrono::{DateTime, NaiveDate, NaiveTime, TimeZone, Utc};
|
||||
|
||||
@@ -16,9 +16,11 @@
|
||||
//! - `and` / `or` boolean combinators, with `and` binding tighter than `or`.
|
||||
//! - Parenthesized sub-expressions.
|
||||
//!
|
||||
//! Nothing in `main.rs` calls into this module yet — it is wired up by a
|
||||
//! later integration ticket — so the compiler's dead-code lint is silenced
|
||||
//! for the whole module rather than for each individual item.
|
||||
//! `engine.rs` (ticket 007) is what actually calls into this module: it
|
||||
//! compiles the returned [`FilterExpr`] against the index (resolving
|
||||
//! [`FilterValue::Raw`] tokens via `filter_datetime` and metric comparisons
|
||||
//! via bucket data) and evaluates it per log line while answering
|
||||
//! `EngineQuery::Logs`/`Timeline`.
|
||||
#![allow(dead_code)]
|
||||
|
||||
use pest::Parser;
|
||||
|
||||
@@ -67,7 +67,11 @@ pub enum Number {
|
||||
}
|
||||
|
||||
impl Number {
|
||||
fn as_f64(&self) -> f64 {
|
||||
/// Widens to `f64` for cross-kind numeric comparisons (e.g. a filter
|
||||
/// query's `FilterValue::Number` against a declared metric's captured
|
||||
/// `Number`). `pub(crate)` rather than private since ticket 007's filter
|
||||
/// evaluator (in `engine.rs`) needs it and lives in a different module.
|
||||
pub(crate) fn as_f64(&self) -> f64 {
|
||||
match self {
|
||||
Number::Int(v) => *v as f64,
|
||||
Number::Float(v) => *v,
|
||||
|
||||
+7
-1
@@ -748,7 +748,13 @@ mod tests {
|
||||
|
||||
let logs = match ask(
|
||||
&engine_tx,
|
||||
EngineQuery::Logs(LogsParams { since_sequence: None, start_ms: None, end_ms: None, limit: None }),
|
||||
EngineQuery::Logs(LogsParams {
|
||||
since_sequence: None,
|
||||
start_ms: None,
|
||||
end_ms: None,
|
||||
limit: None,
|
||||
filter: None,
|
||||
}),
|
||||
)
|
||||
.await
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user