Merge ticket/016-filter-by-json-field

src/alerts.rs

@@ -1,6 +1,6 @@
 //! Alert rules: definitions, trigger evaluation, and ntfy delivery.
 //!
-//! Alert queries are restricted to regex/sequence/timestamp — a metric[param] comparison would require a full bucket rescan per line, breaking the proportional-cost guarantee.
+//! Alert queries are restricted to regex/sequence/timestamp/json[field] — a metric[param] comparison would require a full bucket rescan per line, breaking the proportional-cost guarantee. json[field] is fine here since, like regex/sequence/timestamp, it only inspects the line's own text/fields, no index lookup.
 
 use std::collections::hash_map::DefaultHasher;
 use std::collections::{HashMap, HashSet, VecDeque};
@@ -13,7 +13,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use regex::Regex;
 use serde::{Deserialize, Serialize};
 
-use crate::engine::apply_compare_op;
+use crate::engine::{apply_compare_op, eval_json_comparison};
 use crate::filter_datetime;
 use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
 use crate::settings::Settings;
@@ -155,7 +155,7 @@ fn prune_window(timestamps: &mut VecDeque<i64>, now_ms: i64, window_ms: i64) {
     }
 }
 
-/// Evaluates a FilterBar query against a single line's own fields only (metric[param] comparisons aren't supported — see module doc).
+/// Evaluates a FilterBar query against a single line's own fields only (metric[param] comparisons aren't supported — see module doc; json[field] is supported since it's also line-local).
 fn eval_alert_query(query: &str, sequence: u64, timestamp: Timestamp, text: &str) -> Result<bool, String> {
     let expr = filter_parser::parse(query).map_err(|err| err.to_string())?;
     eval_alert_expr(&expr, sequence, timestamp, text)
@@ -189,6 +189,9 @@ fn eval_alert_expr(expr: &FilterExpr, sequence: u64, timestamp: Timestamp, text:
             "alert queries only support regex()/sequence/timestamp (not metric comparisons like \
              '{metric}[param]'): metric-based alert conditions aren't supported yet"
         )),
+        FilterExpr::JsonComparison { field, op, value } => {
+            Ok(eval_json_comparison(text, field, *op, value))
+        }
         FilterExpr::And(lhs, rhs) => {
             Ok(eval_alert_expr(lhs, sequence, timestamp, text)? && eval_alert_expr(rhs, sequence, timestamp, text)?)
         }

src/engine.rs

@@ -505,6 +505,10 @@ enum CompiledFilter {
     Sequence { op: CompareOp, target: f64 },
     Timestamp { op: CompareOp, target_ms: i64 },
     MetricMatch(HashSet<u64>),
+    /// `json[field] <op> value` — unlike `MetricMatch`, this can't be pre-resolved against the
+    /// index (the JSON lives in the log line itself, not a declared metric/param), so it's
+    /// evaluated per line in `eval_compiled` by parsing `text` as JSON.
+    Json { field: String, op: CompareOp, value: FilterValue },
     And(Box<CompiledFilter>, Box<CompiledFilter>),
     Or(Box<CompiledFilter>, Box<CompiledFilter>),
 }
@@ -533,6 +537,11 @@ fn compile_filter(
             "timestamp" => compile_timestamp_comparison(*op, value, now),
             other => compile_metric_comparison(index, other, param.as_deref(), *op, value, now),
         },
+        FilterExpr::JsonComparison { field, op, value } => Ok(CompiledFilter::Json {
+            field: field.clone(),
+            op: *op,
+            value: value.clone(),
+        }),
         FilterExpr::And(lhs, rhs) => Ok(CompiledFilter::And(
             Box::new(compile_filter(index, lhs, now)?),
             Box::new(compile_filter(index, rhs, now)?),
@@ -650,6 +659,59 @@ fn compare_metric_primitive(
     Ok(ordering.is_some_and(|ordering| apply_compare_op(ordering, op)))
 }
 
+/// `json[field] <op> value`: the log line is parsed as JSON fresh on every call (no per-line
+/// cache), since this doesn't go through the index-backed pre-resolution `MetricMatch` uses.
+/// Any line that isn't a JSON object, or lacks `field`, simply doesn't match — no error.
+pub(crate) fn eval_json_comparison(text: &str, field: &str, op: CompareOp, value: &FilterValue) -> bool {
+    let Ok(parsed) = serde_json::from_str::<serde_json::Value>(text) else {
+        return false;
+    };
+    let Some(field_value) = parsed.as_object().and_then(|obj| obj.get(field)) else {
+        return false;
+    };
+    compare_json_value(field_value, op, value)
+}
+
+/// Mirrors `compare_metric_primitive`'s "mismatched kinds never match" behavior. Only `Eq`/`Ne`
+/// are meaningful for booleans, and `null`/array/object fields aren't scalar-comparable at all.
+fn compare_json_value(field_value: &serde_json::Value, op: CompareOp, value: &FilterValue) -> bool {
+    match field_value {
+        serde_json::Value::Number(n) => {
+            let target = match value {
+                FilterValue::Number(v) => Some(*v),
+                FilterValue::Raw(raw) => raw.parse::<f64>().ok(),
+            };
+            match (n.as_f64(), target) {
+                (Some(f), Some(v)) => f.partial_cmp(&v).is_some_and(|ordering| apply_compare_op(ordering, op)),
+                _ => false,
+            }
+        }
+        serde_json::Value::String(s) => {
+            let target = match value {
+                FilterValue::Raw(raw) => raw.clone(),
+                FilterValue::Number(n) => n.to_string(),
+            };
+            s.as_str()
+                .partial_cmp(target.as_str())
+                .is_some_and(|ordering| apply_compare_op(ordering, op))
+        }
+        serde_json::Value::Bool(b) => {
+            let FilterValue::Raw(raw) = value else {
+                return false;
+            };
+            match raw.parse::<bool>() {
+                Ok(v) => match op {
+                    CompareOp::Eq => *b == v,
+                    CompareOp::Ne => *b != v,
+                    _ => false,
+                },
+                Err(_) => false,
+            }
+        }
+        serde_json::Value::Null | serde_json::Value::Array(_) | serde_json::Value::Object(_) => false,
+    }
+}
+
 /// `pub(crate)` so `crate::alerts` can reuse it for `Threshold`/`Ratio` comparisons instead of duplicating the match.
 pub(crate) fn apply_compare_op(ordering: Ordering, op: CompareOp) -> bool {
     match op {
@@ -673,6 +735,7 @@ fn eval_compiled(compiled: &CompiledFilter, sequence: u64, timestamp: Timestamp,
             apply_compare_op(timestamp.as_millis().cmp(target_ms), *op)
         }
         CompiledFilter::MetricMatch(sequences) => sequences.contains(&sequence),
+        CompiledFilter::Json { field, op, value } => eval_json_comparison(text, field, *op, value),
         CompiledFilter::And(lhs, rhs) => {
             eval_compiled(lhs, sequence, timestamp, text) && eval_compiled(rhs, sequence, timestamp, text)
         }
@@ -2063,6 +2126,30 @@ mod tests {
         }
     }
 
+    #[test]
+    fn logs_query_with_json_field_comparison_filters_by_json_field() {
+        let mut index = BTreeMap::new();
+        ingest_log(&mut index, 0, r#"{"level":"error","code":500}"#);
+        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) {
+            EngineResponse::Logs(entries) => {
+                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
+                assert_eq!(sequences, vec![0]);
+            }
+            _ => panic!("expected Logs response"),
+        }
+
+        match handle_query(&index, logs_query(Some("json[code]>=300")), 0) {
+            EngineResponse::Logs(entries) => {
+                let sequences: Vec<u64> = entries.iter().map(|e| e.sequence).collect();
+                assert_eq!(sequences, vec![0]);
+            }
+            _ => panic!("expected Logs response"),
+        }
+    }
+
     #[test]
     fn logs_query_combines_regex_and_metric_comparison_with_and() {
         let mut index = BTreeMap::new();

src/filter_parser.pest

@@ -3,6 +3,7 @@
 // Supports:
 //   - regex("...")                      raw-text regex search over log lines
 //   - metric_name[param] <op> value     metric comparisons
+//   - json[field] <op> value            JSON field comparisons (line parsed as JSON)
 //   - sequence <op> value               reserved metric name (no [param])
 //   - timestamp <op> value              reserved metric name (no [param])
 //   - `and` / `or` boolean combinators, with `and` binding tighter than `or`
@@ -31,7 +32,12 @@ string_literal = ${ "\"" ~ inner_str ~ "\"" }
 inner_str = @{ (escaped_char | !("\"" | "\\") ~ ANY)* }
 escaped_char = @{ "\\" ~ ANY }
 
-comparison = { metric ~ ("[" ~ param ~ "]")? ~ op ~ value }
+comparison = _{ json_comparison | metric_comparison }
+
+// `json[field] <op> value` must be tried before `metric_comparison` since "json" is
+// otherwise a syntactically valid (if never declared) metric name.
+json_comparison = { "json" ~ "[" ~ param ~ "]" ~ op ~ value }
+metric_comparison = { metric ~ ("[" ~ param ~ "]")? ~ op ~ value }
 
 metric = @{ ident }
 param = @{ ident }

src/filter_parser.rs

@@ -23,6 +23,13 @@ pub enum FilterExpr {
         op: CompareOp,
         value: FilterValue,
     },
+    /// `json[field] <op> value` — a comparison against a field of the line's parsed JSON.
+    /// Lines that aren't valid JSON (or lack the field) simply don't match.
+    JsonComparison {
+        field: String,
+        op: CompareOp,
+        value: FilterValue,
+    },
     And(Box<FilterExpr>, Box<FilterExpr>),
     Or(Box<FilterExpr>, Box<FilterExpr>),
 }
@@ -130,7 +137,8 @@ fn build_and_expr(pair: Pair<Rule>) -> Result<FilterExpr, FilterParseError> {
 fn build_term(pair: Pair<Rule>) -> Result<FilterExpr, FilterParseError> {
     match pair.as_rule() {
         Rule::regex_expr => build_regex(pair),
-        Rule::comparison => build_comparison(pair),
+        Rule::metric_comparison => build_comparison(pair),
+        Rule::json_comparison => build_json_comparison(pair),
         Rule::or_expr => build_or_expr(pair),
         other => Err(FilterParseError::new(format!(
             "unexpected grammar element: {other:?}"
@@ -187,6 +195,27 @@ fn build_comparison(pair: Pair<Rule>) -> Result<FilterExpr, FilterParseError> {
     })
 }
 
+fn build_json_comparison(pair: Pair<Rule>) -> Result<FilterExpr, FilterParseError> {
+    let mut inner = pair.into_inner();
+
+    let field_pair = inner
+        .next()
+        .ok_or_else(|| FilterParseError::new("malformed json[...] comparison: missing field name"))?;
+    let field = field_pair.as_str().to_string();
+
+    let op_pair = inner
+        .next()
+        .ok_or_else(|| FilterParseError::new("malformed json[...] comparison: missing operator"))?;
+    let op = parse_op(op_pair.as_str())?;
+
+    let value_pair = inner
+        .next()
+        .ok_or_else(|| FilterParseError::new("malformed json[...] comparison: missing value"))?;
+    let value = parse_value(value_pair.as_str());
+
+    Ok(FilterExpr::JsonComparison { field, op, value })
+}
+
 fn parse_op(s: &str) -> Result<CompareOp, FilterParseError> {
     match s {
         "<" => Ok(CompareOp::Lt),
@@ -282,6 +311,51 @@ mod tests {
         );
     }
 
+    #[test]
+    fn parses_json_field_comparison() {
+        let expr = parse("json[level]==error").expect("should parse");
+        assert_eq!(
+            expr,
+            FilterExpr::JsonComparison {
+                field: "level".to_string(),
+                op: CompareOp::Eq,
+                value: FilterValue::Raw("error".to_string()),
+            }
+        );
+    }
+
+    #[test]
+    fn parses_json_field_numeric_comparison() {
+        let expr = parse("json[status]>=500").expect("should parse");
+        assert_eq!(
+            expr,
+            FilterExpr::JsonComparison {
+                field: "status".to_string(),
+                op: CompareOp::Ge,
+                value: FilterValue::Number(500.0),
+            }
+        );
+    }
+
+    #[test]
+    fn json_field_composes_with_and_or() {
+        let expr = parse(r#"json[level]==error and sequence<5"#).expect("should parse");
+        let expected = FilterExpr::And(
+            Box::new(FilterExpr::JsonComparison {
+                field: "level".to_string(),
+                op: CompareOp::Eq,
+                value: FilterValue::Raw("error".to_string()),
+            }),
+            Box::new(FilterExpr::Comparison {
+                metric: "sequence".to_string(),
+                param: None,
+                op: CompareOp::Lt,
+                value: FilterValue::Number(5.0),
+            }),
+        );
+        assert_eq!(expr, expected);
+    }
+
     #[test]
     fn parses_and_or_with_and_binding_tighter() {
         let expr = parse(r#"regex("error") and sequence<5 or timestamp>12:00"#)