use crate::alerts::{self, AlertRuntimeState}; use crate::data_value::{DataValue, MetricPrimitiveKind, ParamRole, RegexToMetricPrimitiveMatcher}; 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::settings::SettingsHandle; use crate::timestamp::Timestamp; use regex::Regex; use serde::Deserialize; use std::cmp::Ordering; use std::collections::{BTreeMap, HashMap, HashSet}; use tokio::sync::{mpsc, oneshot}; pub struct IngestedLine { pub sequence: u64, pub timestamp: Timestamp, pub text: String, } /// One sample scraped from a Prometheus/OpenMetrics endpoint, pushed straight into a metric's bucket /// without going through log-line regex matching — see `crate::ingest::run_metrics_poll`. pub struct MetricSample { pub metric_name: String, pub labels: BTreeMap, pub value: f64, pub timestamp: Timestamp, } /// Also the request body for `POST /api/params` — `Deserialize` is derived directly here rather than adding a separate DTO, since the shape already matches. #[derive(Debug, Clone, Deserialize)] pub struct DeclareParamRequest { pub name: String, pub message_match_regex: String, pub capture_group: String, /// See `RegexToMetricPrimitiveMatcher::json_field_path`; `#[serde(default)]` so existing /// regex-only callers are unaffected. #[serde(default)] pub json_field_path: Option, pub kind: MetricPrimitiveKind, } /// Params are derived purely from inline named capture groups on this metric's own regex — referencing an already-declared `ParamDeclaration` by name isn't supported yet. Also the request body for `POST /api/metrics` (hence `Deserialize`). #[derive(Debug, Clone, Deserialize)] pub struct DeclareMetricRequest { pub name: String, pub message_match_regex: String, pub params: Vec, /// Fixed labels layered on top of what `params` captured, so e.g. a case-insensitive match doesn't fragment into "ERROR"/"Error" series; `#[serde(default)]` so existing declarations are unaffected. #[serde(default)] pub matched_labels: Option>, /// When a line doesn't match, registers one occurrence labeled with these fixed values instead of being invisible to this metric — plain regex alternation can't express "didn't match" without lookaround; `#[serde(default)]` so existing declarations keep matching-only behavior. #[serde(default)] pub unmatched_labels: Option>, } #[allow(dead_code)] pub enum EngineQuery { Logs(LogsParams), Timeline(TimelineParams), Stats, ParamDeclarations, MetricDeclarations, /// `labels_filter` compares against each label's `Display` form (not a typed value), since the filter arrives as plain strings off the query string. BucketValues { metric_name: String, labels_filter: Option>, start_ms: Option, end_ms: Option, since_sequence: Option, limit: Option, }, /// `query` is parsed inside the engine (not the API handler) so it stays in the single-owner loop with no copying; a malformed query is only discovered here, hence `EngineResponse::Promql` carries a `Result`. Promql { query: String, start_ms: Option, end_ms: Option }, } #[allow(dead_code)] pub enum EngineMessage { Ingest(IngestedLine), IngestMetricSample(MetricSample), DeclareParam { request: DeclareParamRequest, respond_to: oneshot::Sender>, }, RemoveParamDeclaration { name: String, respond_to: oneshot::Sender, }, DeclareMetric { request: DeclareMetricRequest, respond_to: oneshot::Sender>, }, /// Updates an existing metric declaration named `name` in place — same validation as `DeclareMetric`, but fails if no declaration named `name` exists (rather than silently creating one), and re-backfills bucket data from scratch so edited params/regex don't leave stale buckets from the old declaration mixed in. UpdateMetric { name: String, request: DeclareMetricRequest, respond_to: oneshot::Sender>, }, RemoveMetricDeclaration { name: String, respond_to: oneshot::Sender, }, /// `None` means unlimited; sent on startup and whenever the setting changes so the cap applies without a restart. SetMaxLogs(Option), Query { query: EngineQuery, respond_to: oneshot::Sender, }, } #[allow(dead_code)] pub enum EngineResponse { Logs(Vec), Timeline(Vec), Stats(StatsDto), ParamDeclarations(Vec), MetricDeclarations(Vec), BucketValues(Vec), /// `Err` holds a human-readable message for a malformed query, which the API layer maps to a 400 response. Promql(Result, String>), /// Kept as a variant (rather than making `handle_query` return a `Result`) so every query path still funnels through the same `oneshot::Sender`. Error(String), } pub type EngineHandle = mpsc::Sender; /// Sole owner of the primary index; runs single-threaded so the map never needs a lock, and builds DTOs itself so callers never clone the index. pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle) { let mut index: BTreeMap = BTreeMap::new(); let mut max_logs: Option = None; let mut log_count: usize = 0; // Its own counter, independent of log `sequence` — scraped samples aren't tied to any ingested line. let mut metric_sample_sequence: u64 = 0; // Per-rule rolling windows/seen-value sets, keyed by rule id; lives only as long as this loop (no restart persistence). let mut alert_states: HashMap = HashMap::new(); // Ticket 011: running total of `approx_log_bytes` (see that function) across // every currently-ingested log, kept in lockstep with `log_count` — incremented // here on ingest, decremented in `evict_excess_logs` when a log is dropped. let mut approx_bytes_total: u64 = 0; while let Some(message) = rx.recv().await { match message { EngineMessage::Ingest(line) => { let IngestedLine { sequence, timestamp, text } = line; approx_bytes_total += approx_log_bytes(&text); index.insert( IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text: text.clone() }, ); log_count += 1; apply_all_metrics(&mut index, sequence, timestamp, &text); evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); // Evaluate alert rules against just this line, then hand any fired rule off to delivery via tokio::spawn so it stays off the hot path. let (rules, ntfy_topic, ntfy_server) = { let settings = settings.lock().unwrap(); (settings.alert_rules.clone(), settings.ntfy_topic.clone(), settings.ntfy_server.clone()) }; let fired = alerts::evaluate_alerts(&rules, &mut alert_states, sequence, timestamp, &text); for (rule, message) in fired { tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone())); } } EngineMessage::IngestMetricSample(sample) => { let labels = sample .labels .into_iter() .map(|(name, value)| (name, MetricPrimitive::String(value))) .collect(); push_bucket_value( &mut index, &sample.metric_name, labels, metric_sample_sequence, sample.timestamp, MetricPrimitive::Number(Number::Float(sample.value)), ); metric_sample_sequence += 1; } EngineMessage::SetMaxLogs(new_max) => { max_logs = new_max; evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); } EngineMessage::DeclareParam { request, respond_to } => { let result = declare_param(&mut index, request); let _ = respond_to.send(result); } EngineMessage::RemoveParamDeclaration { name, respond_to } => { let existed = index.remove(&IndexKey::ParamDeclaration { name }).is_some(); let _ = respond_to.send(existed); } EngineMessage::DeclareMetric { request, respond_to } => { let result = declare_metric(&mut index, request); let _ = respond_to.send(result); } EngineMessage::UpdateMetric { name, request, respond_to } => { let result = update_metric(&mut index, &name, request); let _ = respond_to.send(result); } EngineMessage::RemoveMetricDeclaration { name, respond_to } => { let existed = index.remove(&IndexKey::MetricDeclaration { name }).is_some(); let _ = respond_to.send(existed); } EngineMessage::Query { query, respond_to } => { let response = handle_query(&index, query, approx_bytes_total); let _ = respond_to.send(response); } } } } /// Validates the regex and capture group before storing, so a bad declaration is rejected up front rather than discovered later. fn declare_param( index: &mut BTreeMap, request: DeclareParamRequest, ) -> Result<(), String> { let re = Regex::new(&request.message_match_regex) .map_err(|err| format!("invalid regex: {err}"))?; // A Boolean-kind param never uses a capture group — its value is just whether the regex matched. let uses_json_field = request.json_field_path.as_deref().is_some_and(|path| !path.is_empty()); if request.kind != MetricPrimitiveKind::Boolean && !uses_json_field && !re.capture_names().flatten().any(|name| name == request.capture_group) { return Err(format!( "regex has no capture group named '{}'", request.capture_group )); } index.insert( IndexKey::ParamDeclaration { name: request.name }, DataValue::ParamDeclaration { message_match_regex: request.message_match_regex, capture_group: request.capture_group, json_field_path: request.json_field_path, kind: request.kind, }, ); Ok(()) } /// Validates every param matcher's capture-group reference (not just the first bad one) so a rejection reports every problem at once, and the at-most-one-Value-param rule. Shared by `declare_metric` and `update_metric` so both reject a bad declaration the same way. fn validate_metric_request(request: &DeclareMetricRequest) -> Result { let re = Regex::new(&request.message_match_regex) .map_err(|err| format!("invalid regex: {err}"))?; let errors: Vec = request .params .iter() .filter(|matcher| { let uses_json_field = matcher.json_field_path.as_deref().is_some_and(|path| !path.is_empty()); !uses_json_field && !re.capture_names().flatten().any(|name| name == matcher.capture_group) }) .map(|matcher| { format!( "param '{}' references capture group '{}', which the regex does not have", matcher.param_name, matcher.capture_group ) }) .collect(); if !errors.is_empty() { return Err(errors.join("; ")); } let value_params = request.params.iter().filter(|matcher| matcher.role == ParamRole::Value).count(); if value_params > 1 { return Err(format!( "a metric may declare at most one Value-role param, found {value_params}" )); } Ok(re) } fn declare_metric( index: &mut BTreeMap, request: DeclareMetricRequest, ) -> Result<(), String> { let re = validate_metric_request(&request)?; backfill_metric( index, &request.name, &re, &request.params, request.matched_labels.as_ref(), request.unmatched_labels.as_ref(), ); index.insert( IndexKey::MetricDeclaration { name: request.name }, DataValue::MetricDeclaration { message_match_regex: request.message_match_regex, params: request.params, matched_labels: request.matched_labels, unmatched_labels: request.unmatched_labels, }, ); Ok(()) } /// Updates the metric declaration named `existing_name` in place, reusing `declare_metric`'s validation so an edit is rejected the same way a bad new declaration would be. Fails (rather than creating a fresh declaration) if no metric named `existing_name` exists yet. Existing bucket/last-event data for this metric is cleared first and rebuilt from the full log history under the new regex/params, so edited matchers don't leave stale buckets from the previous declaration mixed in with fresh ones. Supports renaming (when `request.name` differs from `existing_name`) as long as the new name isn't already taken by another declaration. fn update_metric( index: &mut BTreeMap, existing_name: &str, request: DeclareMetricRequest, ) -> Result<(), String> { if !index.contains_key(&IndexKey::MetricDeclaration { name: existing_name.to_string() }) { return Err(format!("no metric declaration named '{existing_name}'")); } if request.name != existing_name && index.contains_key(&IndexKey::MetricDeclaration { name: request.name.clone() }) { return Err(format!("a metric named '{}' already exists", request.name)); } let re = validate_metric_request(&request)?; clear_metric_data(index, existing_name); if request.name != existing_name { index.remove(&IndexKey::MetricDeclaration { name: existing_name.to_string() }); } backfill_metric( index, &request.name, &re, &request.params, request.matched_labels.as_ref(), request.unmatched_labels.as_ref(), ); index.insert( IndexKey::MetricDeclaration { name: request.name.clone() }, DataValue::MetricDeclaration { message_match_regex: request.message_match_regex, params: request.params, matched_labels: request.matched_labels, unmatched_labels: request.unmatched_labels, }, ); Ok(()) } /// Removes every `Bucket` and `LastMetricEvent` entry belonging to `metric_name`, ahead of a full re-backfill in `update_metric` — otherwise the old declaration's buckets would sit alongside freshly backfilled ones under the same metric name. fn clear_metric_data(index: &mut BTreeMap, metric_name: &str) { index.retain(|key, _| match key { IndexKey::Bucket { metric_name: name, .. } => name != metric_name, IndexKey::LastMetricEvent { metric_name: name, .. } => name != metric_name, _ => true, }); } /// Returns `None` on a parse failure, which the caller treats as "this match doesn't apply". fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Option { match kind { MetricPrimitiveKind::String => Some(MetricPrimitive::String(raw.to_string())), MetricPrimitiveKind::Number => raw .parse::() .map(Number::Int) .or_else(|_| raw.parse::().map(Number::Float)) .ok() .map(MetricPrimitive::Number), MetricPrimitiveKind::Timestamp => { crate::filter_datetime::parse_datetime(raw, Timestamp::now()) .ok() .map(MetricPrimitive::Timestamp) } // A Value-role param never reaches this branch (see apply_metric_to_log); reached only for a Label-role param or standalone ParamDeclaration, parsed as a literal millisecond count. MetricPrimitiveKind::Duration => raw.parse::().ok().map(MetricPrimitive::Duration), // A standalone Boolean param never reaches this branch (see derive_log_params, which derives its // value from whether the regex matched rather than parsing captured text); reached only if a metric // matcher references a Boolean-kind param, where it's parsed as a literal "true"/"false" capture. MetricPrimitiveKind::Boolean => raw.parse::().ok().map(MetricPrimitive::Boolean), } } /// At most one param may be `Value`-role (its primitive becomes the event's value, otherwise a fixed `1` marker is used); any param failing to match/parse skips the whole match rather than storing a partial one. A `Value`-role `Duration` param is elapsed time since the previous same-labeled event, so the first occurrence pushes no bucket (nothing to diff against) rather than a bogus zero. When `regex` doesn't match and `unmatched_labels` is `Some`, an occurrence is still registered with those fixed labels. fn apply_metric_to_log( index: &mut BTreeMap, metric_name: &str, regex: &Regex, params: &[RegexToMetricPrimitiveMatcher], matched_labels: Option<&BTreeMap>, unmatched_labels: Option<&BTreeMap>, sequence: u64, timestamp: Timestamp, text: &str, ) { let Some(captures) = regex.captures(text) else { if let Some(unmatched_labels) = unmatched_labels { let labels = unmatched_labels .iter() .map(|(name, value)| (name.clone(), MetricPrimitive::String(value.clone()))) .collect(); push_bucket_value(index, metric_name, labels, sequence, timestamp, MetricPrimitive::Number(Number::Int(1))); } return; }; let mut labels = BTreeMap::new(); let mut value = None; let mut is_duration = false; for matcher in params { let raw: String = match matcher.json_field_path.as_deref().filter(|path| !path.is_empty()) { Some(field_path) => match crate::data_value::extract_json_field(text, field_path) { Some(value) => value, None => return, }, None => match captures.name(&matcher.capture_group) { Some(m) => m.as_str().to_string(), None => return, }, }; if matcher.role == ParamRole::Value && matcher.kind == MetricPrimitiveKind::Duration { is_duration = true; continue; } let Some(parsed) = parse_metric_primitive(&raw, matcher.kind) else { return; }; match matcher.role { ParamRole::Label => { labels.insert(matcher.param_name.clone(), parsed); } ParamRole::Value => { value = Some(parsed); } } } if let Some(matched_labels) = matched_labels { for (name, value) in matched_labels { labels.insert(name.clone(), MetricPrimitive::String(value.clone())); } } let value = if is_duration { // Tracked separately from Bucket: a Duration's first occurrence pushes no Bucket entry, so the last-event timestamp can't live in Bucket.values. let last_event_key = IndexKey::LastMetricEvent { metric_name: metric_name.to_string(), labels: labels.clone() }; let previous_timestamp = match index.get(&last_event_key) { Some(DataValue::LastEventTimestamp { timestamp: previous }) => Some(*previous), _ => None, }; index.insert(last_event_key, DataValue::LastEventTimestamp { timestamp }); match previous_timestamp { Some(previous) => { MetricPrimitive::Duration(timestamp.as_millis() - previous.as_millis()) } None => return, } } else { value.unwrap_or(MetricPrimitive::Number(Number::Int(1))) }; push_bucket_value(index, metric_name, labels, sequence, timestamp, value); } fn push_bucket_value( index: &mut BTreeMap, metric_name: &str, labels: BTreeMap, sequence: u64, timestamp: Timestamp, value: MetricPrimitive, ) { let key = IndexKey::Bucket { metric_name: metric_name.to_string(), labels }; let entry = index .entry(key) .or_insert_with(|| DataValue::BucketValue { values: Vec::new() }); if let DataValue::BucketValue { values } = entry { values.push((sequence, timestamp, value)); } } /// Backfills buckets for a newly-declared metric against every already-ingested log, not just future ones. fn backfill_metric( index: &mut BTreeMap, metric_name: &str, regex: &Regex, params: &[RegexToMetricPrimitiveMatcher], matched_labels: Option<&BTreeMap>, unmatched_labels: Option<&BTreeMap>, ) { let lines: Vec<(u64, Timestamp, String)> = logs_range(index) .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string())) .collect(); for (sequence, timestamp, text) in lines { apply_metric_to_log( index, metric_name, regex, params, matched_labels, unmatched_labels, sequence, timestamp, &text, ); } } fn apply_all_metrics( index: &mut BTreeMap, sequence: u64, timestamp: Timestamp, text: &str, ) { type SeededMetric = ( String, String, Vec, Option>, Option>, ); let metrics: Vec = metric_declarations_range(index) .map(|(name, message_match_regex, params, matched_labels, unmatched_labels)| { ( name.to_string(), message_match_regex.to_string(), params.to_vec(), matched_labels.cloned(), unmatched_labels.cloned(), ) }) .collect(); for (name, message_match_regex, params, matched_labels, unmatched_labels) in metrics { if let Ok(re) = Regex::new(&message_match_regex) { apply_metric_to_log( index, &name, &re, ¶ms, matched_labels.as_ref(), unmatched_labels.as_ref(), sequence, timestamp, text, ); } } } /// The metrics API layers its own textual label-filter and range filtering on top rather than using this function's `labels_filter` param (used directly by tests). fn bucket_values<'a>( index: &'a BTreeMap, metric_name: &'a str, labels_filter: Option<&'a BTreeMap>, ) -> impl Iterator, &'a (u64, Timestamp, MetricPrimitive))> + 'a { index .range(IndexKey::MetricBucketSplit..IndexKey::End) .filter_map(move |(key, value)| match (key, value) { (IndexKey::Bucket { metric_name: name, labels }, DataValue::BucketValue { values }) if name == metric_name && labels_filter.is_none_or(|filter| filter == labels) => { Some(values.iter().map(move |entry| (labels, entry))) } _ => None, }) .flatten() } /// A [`FilterExpr`] compiled against a snapshot of the index, so evaluating per log line touches the index only once: a `metric[param]` comparison is pre-resolved into a `HashSet` of matching sequences ([`CompiledFilter::MetricMatch`]) for O(1) per-line lookup instead of a regex re-run; a comparison with no `[param]` is rejected since v1 buckets have no single "the metric's value", only a per-label-combo occurrence marker. enum CompiledFilter { Regex(Regex), Sequence { op: CompareOp, target: f64 }, Timestamp { op: CompareOp, target_ms: i64 }, MetricMatch(HashSet), /// `json[field] 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, Box), Or(Box, Box), } /// Resolves every regex/metric-comparison/datetime-literal up front so per-line evaluation ([`eval_compiled`]) never touches the index; `now` anchors relative literals so every comparison in one query resolves against the same instant. fn compile_filter_query( index: &BTreeMap, filter: &str, now: Timestamp, ) -> Result { let expr = filter_parser::parse(filter).map_err(|err| err.to_string())?; compile_filter(index, &expr, now) } fn compile_filter( index: &BTreeMap, expr: &FilterExpr, now: Timestamp, ) -> Result { 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::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)?), )), FilterExpr::Or(lhs, rhs) => Ok(CompiledFilter::Or( Box::new(compile_filter(index, lhs, now)?), Box::new(compile_filter(index, rhs, now)?), )), } } /// A `Raw` value is accepted defensively (parsed as a number) even though the grammar shouldn't produce one for `sequence`. fn compile_sequence_comparison(op: CompareOp, value: &FilterValue) -> Result { let target = match value { FilterValue::Number(n) => *n, FilterValue::Raw(raw) => raw.parse::().map_err(|_| { format!("'sequence' comparisons require a numeric value, got {raw:?}") })?, }; Ok(CompiledFilter::Sequence { op, target }) } /// A `Number` is treated as raw milliseconds; a `Raw` token is resolved via `filter_datetime::parse_datetime` against `now`. fn compile_timestamp_comparison( op: CompareOp, value: &FilterValue, now: Timestamp, ) -> Result { 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] value` for any metric name other than `sequence`/`timestamp` — see [`CompiledFilter`]'s doc comment. fn compile_metric_comparison( index: &BTreeMap, metric: &str, param: Option<&str>, op: CompareOp, value: &FilterValue, now: Timestamp, ) -> Result { 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)) } /// 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 { 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()) } (MetricPrimitive::Duration(ms), FilterValue::Number(v)) => (*ms as f64).partial_cmp(v), (MetricPrimitive::Duration(ms), FilterValue::Raw(raw)) => { let v: f64 = raw .parse() .map_err(|_| format!("expected a numeric value, got {raw:?}"))?; (*ms as f64).partial_cmp(&v) } (MetricPrimitive::Boolean(b), FilterValue::Raw(raw)) => { let v: bool = raw .parse() .map_err(|_| format!("expected true or false, got {raw:?}"))?; b.partial_cmp(&v) } (MetricPrimitive::Boolean(_), FilterValue::Number(_)) => None, }; Ok(ordering.is_some_and(|ordering| apply_compare_op(ordering, op))) } /// `json[field] 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::(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::().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::() { 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 { 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, } } /// Cheap: everything was already resolved by [`compile_filter_query`], so this is just pattern matching 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::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) } CompiledFilter::Or(lhs, rhs) => { eval_compiled(lhs, sequence, timestamp, text) || eval_compiled(rhs, sequence, timestamp, text) } } } fn metric_declarations_range( index: &BTreeMap, ) -> impl Iterator< Item = ( &str, &str, &[RegexToMetricPrimitiveMatcher], Option<&BTreeMap>, Option<&BTreeMap>, ), > { index .range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit) .filter_map(|(key, value)| match (key, value) { ( IndexKey::MetricDeclaration { name }, DataValue::MetricDeclaration { message_match_regex, params, matched_labels, unmatched_labels }, ) => Some(( name.as_str(), message_match_regex.as_str(), params.as_slice(), matched_labels.as_ref(), unmatched_labels.as_ref(), )), _ => None, }) } fn param_declarations_range( index: &BTreeMap, ) -> impl Iterator, MetricPrimitiveKind)> { index .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit) .filter_map(|(key, value)| match (key, value) { ( IndexKey::ParamDeclaration { name }, DataValue::ParamDeclaration { message_match_regex, capture_group, json_field_path, kind }, ) => Some(( name.as_str(), message_match_regex.as_str(), capture_group.as_str(), json_field_path.as_deref(), *kind, )), _ => None, }) } /// A param whose stored regex fails to compile (shouldn't happen — `declare_param` validates it) is skipped rather than panicking. fn compiled_param_declarations( index: &BTreeMap, ) -> Vec<(String, Regex, String, Option, MetricPrimitiveKind)> { param_declarations_range(index) .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind)| { Regex::new(message_match_regex).ok().map(|re| { ( name.to_string(), re, capture_group.to_string(), json_field_path.map(|path| path.to_string()), kind, ) }) }) .collect() } /// A param is omitted (not included as null/empty) when its regex/capture group (or JSON field /// path) doesn't match, or the extracted text fails to parse — see `LogEntryDto::params`. The one /// 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, MetricPrimitiveKind)], ) -> BTreeMap { let mut params = BTreeMap::new(); for (name, regex, capture_group, json_field_path, kind) in declarations { if *kind == MetricPrimitiveKind::Boolean { params.insert(name.clone(), MetricPrimitive::Boolean(regex.is_match(text))); continue; } let Some(captures) = regex.captures(text) else { continue; }; let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) { Some(field_path) => match crate::data_value::extract_json_field(text, field_path) { Some(value) => value, None => continue, }, None => match captures.name(capture_group) { Some(m) => m.as_str().to_string(), None => continue, }, }; let Some(value) = parse_metric_primitive(&raw, *kind) else { continue; }; params.insert(name.clone(), value); } 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. fn approx_log_bytes(text: &str) -> u64 { text.len() as u64 } /// 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 /// left in place — buckets are aggregate data, not raw logs, so they're not /// retroactively pruned. `approx_bytes_total` (ticket 011) is decremented by /// each evicted log's approximate size, mirroring the increment on ingest. fn evict_excess_logs( index: &mut BTreeMap, log_count: &mut usize, max_logs: Option, approx_bytes_total: &mut u64, ) { let Some(max_logs) = max_logs else { return }; while *log_count > max_logs { let Some(oldest_key) = index .range(IndexKey::Start..IndexKey::LogParamSplit) .next() .map(|(key, _)| key.clone()) else { break; }; if let Some(DataValue::LogLiteral { text }) = index.get(&oldest_key) { *approx_bytes_total = approx_bytes_total.saturating_sub(approx_log_bytes(text)); } index.remove(&oldest_key); *log_count -= 1; } } fn logs_range(index: &BTreeMap) -> impl Iterator { index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| { match (key, value) { (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => { Some((*sequence, *timestamp, text.as_str())) } _ => None, } }) } fn handle_query( index: &BTreeMap, query: EngineQuery, approx_bytes_total: u64, ) -> 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); // O(lines_returned * declared_params) per query — fine at v1 scale; if volume/params grow, move to incremental derivation at ingest time like apply_all_metrics does for buckets. let compiled_params = compiled_param_declarations(index); let entries: Vec = logs_range(index) .filter(|(sequence, timestamp, _)| { params.since_sequence.is_none_or(|since| *sequence > since) && 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, timestamp_ms: timestamp.as_millis(), text: text.to_string(), params: derive_log_params(text, &compiled_params), }) .collect(); 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 = BTreeMap::new(); 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; } let out = buckets .into_iter() .map(|(start_ms, count)| TimelineBucketDto { start_ms, count }) .collect(); EngineResponse::Timeline(out) } EngineQuery::Stats => { let mut log_count = 0u64; let mut first_sequence = None; let mut last_sequence = None; let mut first_timestamp_ms = None; let mut last_timestamp_ms = None; for (sequence, timestamp, _) in logs_range(index) { log_count += 1; first_sequence.get_or_insert(sequence); last_sequence = Some(sequence); first_timestamp_ms.get_or_insert(timestamp.as_millis()); last_timestamp_ms = Some(timestamp.as_millis()); } EngineResponse::Stats(StatsDto { log_count, first_sequence, last_sequence, first_timestamp_ms, last_timestamp_ms, approx_bytes_total, }) } EngineQuery::ParamDeclarations => { let declarations = param_declarations_range(index) .map(|(name, message_match_regex, capture_group, json_field_path, kind)| ParamDeclarationDto { name: name.to_string(), message_match_regex: message_match_regex.to_string(), capture_group: capture_group.to_string(), json_field_path: json_field_path.map(|path| path.to_string()), kind, }) .collect(); EngineResponse::ParamDeclarations(declarations) } EngineQuery::MetricDeclarations => { let declarations = metric_declarations_range(index) .map(|(name, message_match_regex, params, matched_labels, unmatched_labels)| MetricDeclarationDto { name: name.to_string(), message_match_regex: message_match_regex.to_string(), params: params.to_vec(), matched_labels: matched_labels.cloned(), unmatched_labels: unmatched_labels.cloned(), }) .collect(); EngineResponse::MetricDeclarations(declarations) } EngineQuery::BucketValues { metric_name, labels_filter, start_ms, end_ms, since_sequence, limit, } => { let limit = limit.unwrap_or(500).min(10_000); let values: Vec = bucket_values(index, &metric_name, None) .filter(|(labels, _)| { labels_filter.as_ref().is_none_or(|filter| { filter.iter().all(|(name, expected)| { labels .get(name) .is_some_and(|value| &value.to_string() == expected) }) }) }) .filter(|(_, (sequence, timestamp, _))| { since_sequence.is_none_or(|since| *sequence > since) && start_ms.is_none_or(|start| timestamp.as_millis() >= start) && end_ms.is_none_or(|end| timestamp.as_millis() <= end) }) .take(limit) .map(|(labels, (sequence, timestamp, value))| BucketValueDto { labels: labels.clone(), sequence: *sequence, timestamp_ms: timestamp.as_millis(), value: value.clone(), }) .collect(); EngineResponse::BucketValues(values) } EngineQuery::Promql { query, start_ms, end_ms } => { let response = promql::parse_query(&query) .map(|parsed| evaluate_promql(index, &parsed, start_ms, end_ms)); EngineResponse::Promql(response) } } } /// Coerces a bucket's `MetricPrimitive` into a plottable `f64`: Number/Timestamp/Duration use their natural numeric value; `String` has no numeric reading, so it falls back to `1.0` (matching the presence/occurrence marker used for every bucket) rather than being skipped, keeping every declared metric queryable. fn coerce_to_f64(value: &MetricPrimitive) -> f64 { match value { MetricPrimitive::Number(n) => match n { Number::Int(v) => *v as f64, Number::Float(v) => *v, }, MetricPrimitive::Timestamp(ts) => ts.as_millis() as f64, MetricPrimitive::Duration(ms) => *ms as f64, MetricPrimitive::String(_) => 1.0, MetricPrimitive::Boolean(b) => { if *b { 1.0 } else { 0.0 } } } } /// Compared against each label's `Display` form; extra labels on the bucket are ignored — the matcher list only needs to be a subset. fn matches_label_matchers( labels: &BTreeMap, matchers: &[LabelMatcher], ) -> bool { matchers.iter().all(|matcher| { labels.get(&matcher.name).is_some_and(|value| value.to_string() == matcher.value) }) } /// Projects a full label map down to just the names in `by_labels`; an empty `by_labels` collapses /// everything to the empty map, i.e. "one group total" (the historical `sum()`/`avg()` behavior). fn project_labels( labels: &BTreeMap, by_labels: &[String], ) -> BTreeMap { by_labels .iter() .filter_map(|name| labels.get(name).map(|value| (name.clone(), value.clone()))) .collect() } /// Turns one series' occurrence points into a windowed rate: at each of the series' own timestamps, /// counts how many points fall in the trailing `(ts - range_ms, ts]` window and divides by the window /// width in seconds. Points are occurrence markers rather than cumulative counters, so this reads as /// "occurrences per second over the trailing window", evaluated at every point instead of just the last. fn windowed_rate_series(points: &[(i64, f64)], range_ms: i64) -> Vec<(i64, f64)> { let range_seconds = range_ms as f64 / 1000.0; points .iter() .map(|(ts, _)| { let window_start = ts - range_ms; let count_in_window = points.iter().filter(|(t, _)| *t > window_start && *t <= *ts).count(); (*ts, count_in_window as f64 / range_seconds) }) .collect() } /// Merges several series by exact timestamp match: values sharing a timestamp are summed (or averaged). /// This is the plain `sum()`/`avg()` behavior — appropriate when the underlying points are occurrence /// markers rather than a continuous reading, so there's nothing to carry forward between timestamps. fn merge_series_exact(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { let mut grouped: BTreeMap> = BTreeMap::new(); for points in sub_series { for (timestamp_ms, value) in points { grouped.entry(*timestamp_ms).or_default().push(*value); } } grouped .into_iter() .map(|(timestamp_ms, values)| { let sum: f64 = values.iter().sum(); let value = if is_avg { sum / values.len() as f64 } else { sum }; (timestamp_ms, value) }) .collect() } /// Merges several already-windowed rate series by sweeping the union of their timestamps, carrying each /// sub-series' last-known value forward (0.0 before its first point) so every step reflects every /// sub-series' most recent reading — unlike `merge_series_exact`, which only combines exact-timestamp matches. fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { let mut all_timestamps: std::collections::BTreeSet = std::collections::BTreeSet::new(); for points in sub_series { all_timestamps.extend(points.iter().map(|(ts, _)| *ts)); } let mut cursors = vec![0usize; sub_series.len()]; let mut last_values = vec![0.0f64; sub_series.len()]; all_timestamps .into_iter() .map(|ts| { for (i, points) in sub_series.iter().enumerate() { while cursors[i] < points.len() && points[cursors[i]].0 <= ts { last_values[i] = points[cursors[i]].1; cursors[i] += 1; } } let sum: f64 = last_values.iter().sum(); let value = if is_avg { sum / last_values.len() as f64 } else { sum }; (ts, value) }) .collect() } /// Evaluates a parsed PromQL query against `index`, producing chart-ready series: baseline is one series per label combination; `sum()`/`avg()` merge series into one series per `by_labels` group (the whole result collapses to a single label-less series when `by_labels` is empty); `rate()` — bare or wrapped in `sum()`/`avg()` — turns each series into a windowed occurrences-per-second reading via `windowed_rate_series`, so it plots as a real line rather than one trailing point; `histogram()` groups by `by_labels` the same way sum()/avg() do, then buckets each group's values independently, so `histogram(m) by (label)` yields one histogram (a family of `le`-labeled series carrying the group's own labels) per label group instead of collapsing all groups together. fn evaluate_promql( index: &BTreeMap, parsed: &PromqlQuery, start_ms: Option, end_ms: Option, ) -> Vec { let mut series_points: BTreeMap, Vec<(i64, f64)>> = BTreeMap::new(); for (labels, (_, timestamp, value)) in bucket_values(index, &parsed.metric_name, None) .filter(|(labels, _)| matches_label_matchers(labels, &parsed.matchers)) .filter(|(_, (_, timestamp, _))| { start_ms.is_none_or(|start| timestamp.as_millis() >= start) && end_ms.is_none_or(|end| timestamp.as_millis() <= end) }) { let display_labels: BTreeMap = labels.iter().map(|(name, value)| (name.clone(), value.to_string())).collect(); series_points .entry(display_labels) .or_default() .push((timestamp.as_millis(), coerce_to_f64(value))); } for points in series_points.values_mut() { points.sort_by_key(|(timestamp_ms, _)| *timestamp_ms); } match parsed.agg { None => series_points .into_iter() .map(|(labels, points)| PromqlSeriesDto { labels, points }) .collect(), Some(AggOp::Sum) | Some(AggOp::Avg) => { let is_avg = parsed.agg == Some(AggOp::Avg); // Validated by `promql::parse_query`: an inner rate() always carries a range. let rate_range_ms = parsed .inner_rate .then(|| parsed.range.expect("inner rate() always carries a range").as_millis() as i64); let mut groups: BTreeMap, Vec>> = BTreeMap::new(); for (labels, points) in series_points { let points = match rate_range_ms { Some(range_ms) => windowed_rate_series(&points, range_ms), None => points, }; let group_key = project_labels(&labels, &parsed.by_labels); groups.entry(group_key).or_default().push(points); } groups .into_iter() .map(|(group_labels, sub_series)| { let points = if rate_range_ms.is_some() { merge_series_locf(&sub_series, is_avg) } else { merge_series_exact(&sub_series, is_avg) }; PromqlSeriesDto { labels: group_labels, points } }) .collect() } Some(AggOp::Rate) => { // Validated by `promql::parse_query`: `rate()` never parses without one. let range_ms = parsed.range.expect("rate() always carries a range").as_millis() as i64; series_points .into_iter() .map(|(labels, points)| PromqlSeriesDto { labels, points: windowed_rate_series(&points, range_ms), }) .collect() } Some(AggOp::Gauge) => { // Collapse each label combination's events down to their most-recent value in range — "value at time T", not a distinct stored gauge kind. series_points .into_iter() .filter_map(|(labels, points)| { let last = *points.last()?; Some(PromqlSeriesDto { labels, points: vec![last] }) }) .collect() } Some(AggOp::Histogram) => { // Validated by `promql::parse_query`: `histogram()` never parses // without one. let bucket_width = parsed.bucket_width.expect("histogram() always carries a bucket width"); let last_ts = series_points .values() .flat_map(|points| points.iter().map(|(ts, _)| *ts)) .max() .unwrap_or(0); // Groups series by `by_labels` first (empty `by_labels` collapses to a single group, // matching sum()/avg()'s behavior), then buckets each group's values independently into // fixed-width buckets, labeled `le` (Prometheus's less-than-or-equal convention) alongside // the group's own by_labels values. let mut groups: BTreeMap, Vec<&(i64, f64)>> = BTreeMap::new(); for (labels, points) in &series_points { let group_key = project_labels(labels, &parsed.by_labels); groups.entry(group_key).or_default().extend(points.iter()); } groups .into_iter() .flat_map(|(group_labels, points)| { let mut counts: BTreeMap = BTreeMap::new(); for (_, value) in points { let bucket_index = (value / bucket_width).floor() as i64; *counts.entry(bucket_index).or_insert(0) += 1; } counts.into_iter().map(move |(bucket_index, count)| { let upper_bound = (bucket_index + 1) as f64 * bucket_width; let mut labels = group_labels.clone(); labels.insert("le".to_string(), upper_bound.to_string()); PromqlSeriesDto { labels, points: vec![(last_ts, count as f64)] } }) }) .collect() } } } #[cfg(test)] mod tests { use super::*; fn request(name: &str, pattern: &str, capture_group: &str) -> DeclareParamRequest { DeclareParamRequest { name: name.to_string(), message_match_regex: pattern.to_string(), capture_group: capture_group.to_string(), json_field_path: None, kind: MetricPrimitiveKind::String, } } #[test] fn declares_valid_param() { let mut index = BTreeMap::new(); let result = declare_param(&mut index, request("level", r"level=(?P\w+)", "level")); assert!(result.is_ok()); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "level"); assert_eq!(declarations[0].1, r"level=(?P\w+)"); assert_eq!(declarations[0].2, "level"); } #[test] fn rejects_invalid_regex() { let mut index = BTreeMap::new(); let result = declare_param(&mut index, request("level", "level=(", "level")); assert!(result.is_err()); assert!(index.is_empty()); } #[test] fn rejects_unknown_capture_group() { let mut index = BTreeMap::new(); let result = declare_param( &mut index, request("level", r"level=(?P\w+)", "missing"), ); assert!(result.is_err()); assert!(index.is_empty()); } #[test] fn declares_valid_param_from_json_field_path() { let mut index = BTreeMap::new(); let result = declare_param( &mut index, DeclareParamRequest { name: "level".to_string(), // No named capture group needed for JSON-field extraction, just something that // matches the lines this param applies to. message_match_regex: r".".to_string(), capture_group: String::new(), json_field_path: Some("foo.bar".to_string()), kind: MetricPrimitiveKind::String, }, ); assert!(result.is_ok()); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].3, Some("foo.bar")); } #[test] fn derives_log_param_from_json_field_path() { let mut index = BTreeMap::new(); declare_param( &mut index, DeclareParamRequest { name: "bar".to_string(), message_match_regex: r".".to_string(), capture_group: String::new(), json_field_path: Some("foo.bar".to_string()), kind: MetricPrimitiveKind::Number, }, ) .unwrap(); let declarations = compiled_param_declarations(&index); let params = derive_log_params(r#"{"foo":{"bar":42}}"#, &declarations); assert_eq!(params.get("bar"), Some(&MetricPrimitive::Number(Number::Int(42)))); // A line that doesn't parse as JSON, or is missing the field, yields no param. assert!(derive_log_params("not json", &declarations).is_empty()); assert!(derive_log_params(r#"{"foo":{}}"#, &declarations).is_empty()); } #[test] fn evict_excess_logs_drops_oldest_sequences_first() { let mut index = BTreeMap::new(); let mut log_count = 0; let mut approx_bytes_total = 0u64; for sequence in 0..5u64 { let text = format!("line {sequence}"); approx_bytes_total += approx_log_bytes(&text); index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, DataValue::LogLiteral { text }, ); log_count += 1; } evict_excess_logs(&mut index, &mut log_count, Some(3), &mut approx_bytes_total); assert_eq!(log_count, 3); let remaining: Vec<_> = logs_range(&index).map(|(seq, _, _)| seq).collect(); assert_eq!(remaining, vec![2, 3, 4]); let expected_bytes: u64 = remaining.iter().map(|seq| approx_log_bytes(&format!("line {seq}"))).sum(); assert_eq!(approx_bytes_total, expected_bytes); } #[test] fn evict_excess_logs_is_noop_when_unlimited_or_under_cap() { let mut index = BTreeMap::new(); let mut log_count = 0; let mut approx_bytes_total = 0u64; for sequence in 0..3u64 { approx_bytes_total += approx_log_bytes("x"); index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, DataValue::LogLiteral { text: "x".to_string() }, ); log_count += 1; } evict_excess_logs(&mut index, &mut log_count, None, &mut approx_bytes_total); assert_eq!(log_count, 3); assert_eq!(approx_bytes_total, 3); evict_excess_logs(&mut index, &mut log_count, Some(10), &mut approx_bytes_total); assert_eq!(log_count, 3); assert_eq!(approx_bytes_total, 3); } #[test] fn declarations_stay_within_their_range() { let mut index = BTreeMap::new(); index.insert( IndexKey::IngestedLog { sequence: 0, timestamp: Timestamp::from_millis(0) }, DataValue::LogLiteral { text: "hello".to_string() }, ); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); index.insert( IndexKey::MetricDeclaration { name: "requests".to_string() }, DataValue::MetricDeclaration { message_match_regex: "x".to_string(), params: vec![], matched_labels: None, unmatched_labels: None, }, ); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "level"); for key in index.range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit).map(|(k, _)| k) { assert!(matches!(key, IndexKey::ParamDeclaration { .. })); } } #[test] fn removes_declaration() { let mut index = BTreeMap::new(); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); assert!(index.remove(&IndexKey::ParamDeclaration { name: "level".to_string() }).is_some()); assert!(param_declarations_range(&index).next().is_none()); } fn metric_request( name: &str, pattern: &str, params: Vec<(&str, &str, MetricPrimitiveKind)>, ) -> DeclareMetricRequest { metric_request_with_roles( name, pattern, params .into_iter() .map(|(param_name, capture_group, kind)| (param_name, capture_group, kind, ParamRole::Label)) .collect(), ) } fn metric_request_with_roles( name: &str, pattern: &str, params: Vec<(&str, &str, MetricPrimitiveKind, ParamRole)>, ) -> DeclareMetricRequest { DeclareMetricRequest { name: name.to_string(), message_match_regex: pattern.to_string(), matched_labels: None, unmatched_labels: None, params: params .into_iter() .map(|(param_name, capture_group, kind, role)| RegexToMetricPrimitiveMatcher { param_name: param_name.to_string(), capture_group: capture_group.to_string(), json_field_path: None, kind, role, }) .collect(), } } #[test] fn declares_valid_metric() { let mut index = BTreeMap::new(); let result = declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+) duration=(?P\d+)", vec![ ("path", "path", MetricPrimitiveKind::String), ("duration", "duration", MetricPrimitiveKind::Number), ], ), ); assert!(result.is_ok()); let declarations: Vec<_> = metric_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "requests"); assert_eq!( declarations[0].1, r"request path=(?P\S+) duration=(?P\d+)" ); assert_eq!(declarations[0].2.len(), 2); } #[test] fn rejects_invalid_metric_regex() { let mut index = BTreeMap::new(); let result = declare_metric(&mut index, metric_request("requests", "path=(", vec![])); assert!(result.is_err()); assert!(index.is_empty()); } #[test] fn rejects_metric_matcher_with_unknown_capture_group() { let mut index = BTreeMap::new(); let result = declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+)", vec![ ("path", "path", MetricPrimitiveKind::String), ("duration", "duration", MetricPrimitiveKind::Number), ], ), ); assert!(result.is_err()); // Both matchers are validated: the error mentions the one that's actually bad. assert!(result.unwrap_err().contains("duration")); assert!(index.is_empty()); } #[test] fn metric_declarations_stay_within_their_range() { let mut index = BTreeMap::new(); index.insert( IndexKey::IngestedLog { sequence: 0, timestamp: Timestamp::from_millis(0) }, DataValue::LogLiteral { text: "hello".to_string() }, ); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); declare_metric( &mut index, metric_request( "requests", r"duration=(?P\d+)", vec![("duration", "duration", MetricPrimitiveKind::Number)], ), ) .unwrap(); let declarations: Vec<_> = metric_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "requests"); for key in index.range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit).map(|(k, _)| k) { assert!(matches!(key, IndexKey::MetricDeclaration { .. })); } } #[test] fn removes_metric_declaration() { let mut index = BTreeMap::new(); declare_metric( &mut index, metric_request( "requests", r"duration=(?P\d+)", vec![("duration", "duration", MetricPrimitiveKind::Number)], ), ) .unwrap(); assert!(index.remove(&IndexKey::MetricDeclaration { name: "requests".to_string() }).is_some()); assert!(metric_declarations_range(&index).next().is_none()); } #[test] fn update_metric_rejects_when_no_declaration_exists_yet() { let mut index = BTreeMap::new(); let result = update_metric( &mut index, "requests", metric_request("requests", r"request path=(?P\S+)", vec![]), ); assert!(result.is_err()); assert!(result.unwrap_err().contains("requests")); } #[test] fn update_metric_reuses_declare_metric_validation() { let mut index = BTreeMap::new(); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec![])).unwrap(); let result = update_metric( &mut index, "requests", metric_request( "requests", r"request path=(?P\S+)", vec![("duration", "duration", MetricPrimitiveKind::Number)], ), ); assert!(result.is_err()); assert!(result.unwrap_err().contains("duration")); } #[test] fn update_metric_replaces_declaration_in_place_without_duplicating() { let mut index = BTreeMap::new(); declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); let result = update_metric( &mut index, "requests", metric_request( "requests", r"request path=(?P\S+) duration=(?P\d+)", vec![ ("path", "path", MetricPrimitiveKind::String), ("duration", "duration", MetricPrimitiveKind::Number), ], ), ); assert!(result.is_ok()); // Still one declaration named "requests", now with the updated regex/params. let declarations: Vec<_> = metric_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "requests"); assert_eq!(declarations[0].2.len(), 2); } #[test] fn update_metric_rebackfills_bucket_data_from_the_new_declaration() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/b"); declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); assert_eq!(bucket_values(&index, "requests", None).count(), 2); // Editing to a narrower regex should leave exactly the buckets matching the *new* declaration — not the old ones layered underneath. update_metric( &mut index, "requests", metric_request( "requests", r"request path=(?P/a)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); assert_eq!(bucket_values(&index, "requests", None).count(), 1); } #[test] fn update_metric_can_rename_and_rejects_a_rename_onto_an_existing_name() { let mut index = BTreeMap::new(); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec![])).unwrap(); declare_metric(&mut index, metric_request("errors", r"error code=(?P\d+)", vec![])).unwrap(); // Renaming onto an already-taken name is rejected. let result = update_metric( &mut index, "requests", metric_request("errors", r"request path=(?P\S+)", vec![]), ); assert!(result.is_err()); // Renaming onto a fresh name succeeds and drops the old declaration. update_metric( &mut index, "requests", metric_request("http_requests", r"request path=(?P\S+)", vec![]), ) .unwrap(); let names: Vec<_> = metric_declarations_range(&index).map(|(name, ..)| name.to_string()).collect(); assert!(names.contains(&"http_requests".to_string())); assert!(!names.contains(&"requests".to_string())); } fn ingest_log(index: &mut BTreeMap, sequence: u64, text: &str) { let timestamp = Timestamp::from_millis(sequence as i64); index.insert( IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text: text.to_string() }, ); apply_all_metrics(index, sequence, timestamp, text); } #[test] fn declaring_metric_backfills_matching_logs() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "request path=/a duration=1"); ingest_log(&mut index, 1, "request path=/b duration=2"); ingest_log(&mut index, 2, "unrelated line"); declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+) duration=(?P\d+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); let values: Vec<_> = bucket_values(&index, "requests", None).collect(); assert_eq!(values.len(), 2); } #[test] fn metric_param_extracts_from_json_field_path() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, r#"{"request":{"path":"/a"}}"#); // Doesn't parse as JSON at all — should be skipped, not panic. ingest_log(&mut index, 1, "not json"); // Parses as JSON but is missing the field — also skipped. ingest_log(&mut index, 2, r#"{"request":{}}"#); declare_metric( &mut index, DeclareMetricRequest { name: "requests".to_string(), message_match_regex: r".".to_string(), matched_labels: None, unmatched_labels: None, params: vec![RegexToMetricPrimitiveMatcher { param_name: "path".to_string(), capture_group: String::new(), json_field_path: Some("request.path".to_string()), kind: MetricPrimitiveKind::String, role: ParamRole::Label, }], }, ) .unwrap(); let values: Vec<_> = bucket_values(&index, "requests", None).collect(); assert_eq!(values.len(), 1); assert_eq!( values[0].0.get("path"), Some(&MetricPrimitive::String("/a".to_string())) ); } #[test] fn ingesting_after_declaration_updates_buckets_live() { let mut index = BTreeMap::new(); declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); ingest_log(&mut index, 0, "request path=/a"); let values: Vec<_> = bucket_values(&index, "requests", None).collect(); assert_eq!(values.len(), 1); assert_eq!(values[0].1.0, 0); } #[test] fn non_matching_log_leaves_bucket_range_untouched() { let mut index = BTreeMap::new(); declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); ingest_log(&mut index, 0, "totally unrelated line"); assert!( index .range(IndexKey::MetricBucketSplit..IndexKey::End) .find(|(key, _)| matches!(key, IndexKey::Bucket { .. })) .is_none() ); } #[test] fn bucket_values_preserve_insertion_order() { let mut index = BTreeMap::new(); declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); ingest_log(&mut index, 5, "request path=/a"); ingest_log(&mut index, 2, "request path=/a"); ingest_log(&mut index, 9, "request path=/a"); let sequences: Vec = bucket_values(&index, "requests", None) .map(|(_, (sequence, _, _))| *sequence) .collect(); assert_eq!(sequences, vec![5, 2, 9]); } #[test] fn differing_label_combinations_produce_distinct_buckets() { let mut index = BTreeMap::new(); declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/b"); let bucket_keys: Vec<_> = index .range(IndexKey::MetricBucketSplit..IndexKey::End) .filter(|(key, _)| matches!(key, IndexKey::Bucket { .. })) .collect(); assert_eq!(bucket_keys.len(), 2); for (_, value) in &bucket_keys { match value { DataValue::BucketValue { values } => assert_eq!(values.len(), 1), _ => panic!("expected BucketValue"), } } } #[test] fn single_log_line_can_match_multiple_metrics() { let mut index = BTreeMap::new(); declare_metric( &mut index, metric_request( "requests", r"request path=(?P\S+) duration=(?P\d+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); declare_metric( &mut index, metric_request( "durations", r"request path=(?P\S+) duration=(?P\d+)", vec![("duration", "duration", MetricPrimitiveKind::Number)], ), ) .unwrap(); ingest_log(&mut index, 0, "request path=/a duration=5"); assert_eq!(bucket_values(&index, "requests", None).count(), 1); assert_eq!(bucket_values(&index, "durations", None).count(), 1); } fn declare_requests_metric(index: &mut BTreeMap) { declare_metric( index, metric_request( "requests", r"request path=(?P\S+)", vec![("path", "path", MetricPrimitiveKind::String)], ), ) .unwrap(); } #[test] fn promql_query_selects_matching_label_combination() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/b"); ingest_log(&mut index, 2, "request path=/a"); let parsed = promql::parse_query(r#"requests{path="/a"}"#).unwrap(); let series = evaluate_promql(&index, &parsed, None, None); assert_eq!(series.len(), 1); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); // Occurrence markers coerce to 1.0 (see coerce_to_f64); timestamps come from ingest_log's sequence-as-millis helper. assert_eq!(series[0].points, vec![(0, 1.0), (2, 1.0)]); } #[test] fn promql_query_with_no_matching_labels_returns_no_series() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); let parsed = promql::parse_query(r#"requests{path="/nope"}"#).unwrap(); let series = evaluate_promql(&index, &parsed, None, None); assert!(series.is_empty()); } #[test] fn promql_sum_aggregates_across_label_combinations() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(requests)").unwrap(); let series = evaluate_promql(&index, &parsed, None, None); assert_eq!(series.len(), 1); assert!(series[0].labels.is_empty()); assert_eq!(series[0].points, vec![(0, 2.0)]); } #[test] fn promql_sum_by_preserves_the_named_label_dimension() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); ingest_log(&mut index, 1, "request path=/a"); let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0)]); assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/b")); assert_eq!(series[1].points, vec![(0, 1.0)]); } #[test] fn bare_rate_produces_a_windowed_point_per_timestamp_not_a_single_point() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1000, "request path=/a"); let parsed = promql::parse_query("rate(requests[2s])").unwrap(); let series = evaluate_promql(&index, &parsed, None, None); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]); } #[test] fn sum_of_rate_by_label_gives_each_label_value_its_own_line() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1000, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(rate(requests[2s])) by (path)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]); assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/b")); assert_eq!(series[1].points, vec![(0, 0.5)]); } #[test] fn sum_of_rate_without_by_clause_still_collapses_to_one_series() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(rate(requests[2s]))").unwrap(); let series = evaluate_promql(&index, &parsed, None, None); assert_eq!(series.len(), 1); assert!(series[0].labels.is_empty()); assert_eq!(series[0].points, vec![(0, 1.0)]); } #[test] fn promql_range_bounds_filter_points_via_start_end_ms() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 5, "request path=/a"); ingest_log(&mut index, 10, "request path=/a"); let parsed = promql::parse_query("requests").unwrap(); let series = evaluate_promql(&index, &parsed, Some(1), Some(9)); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(5, 1.0)]); } #[test] fn declares_metric_with_value_role_param() { let mut index = BTreeMap::new(); let result = declare_metric( &mut index, metric_request_with_roles( "durations", r"request path=(?P\S+) duration=(?P\d+)", vec![ ("path", "path", MetricPrimitiveKind::String, ParamRole::Label), ("duration", "duration", MetricPrimitiveKind::Number, ParamRole::Value), ], ), ); assert!(result.is_ok()); } #[test] fn rejects_metric_with_multiple_value_role_params() { let mut index = BTreeMap::new(); let result = declare_metric( &mut index, metric_request_with_roles( "bad", r"a=(?P\d+) b=(?P\d+)", vec![ ("a", "a", MetricPrimitiveKind::Number, ParamRole::Value), ("b", "b", MetricPrimitiveKind::Number, ParamRole::Value), ], ), ); assert!(result.is_err()); assert!(result.unwrap_err().contains("at most one Value-role param")); } #[test] fn value_role_param_becomes_the_bucket_value_not_a_label() { let mut index = BTreeMap::new(); declare_metric( &mut index, metric_request_with_roles( "durations", r"request path=(?P\S+) duration=(?P\d+)", vec![ ("path", "path", MetricPrimitiveKind::String, ParamRole::Label), ("duration", "duration", MetricPrimitiveKind::Number, ParamRole::Value), ], ), ) .unwrap(); ingest_log(&mut index, 0, "request path=/a duration=5"); let values: Vec<_> = bucket_values(&index, "durations", None).collect(); assert_eq!(values.len(), 1); let (labels, (_, _, value)) = &values[0]; assert!(!labels.contains_key("duration")); assert_eq!(labels.get("path").map(|v| v.to_string()), Some("/a".to_string())); assert_eq!(*value, MetricPrimitive::Number(Number::Int(5))); } fn declare_durations_metric(index: &mut BTreeMap) { declare_metric( index, metric_request_with_roles( "durations", r"request path=(?P\S+) duration=(?P\d+)", vec![ ("path", "path", MetricPrimitiveKind::String, ParamRole::Label), ("duration", "duration", MetricPrimitiveKind::Number, ParamRole::Value), ], ), ) .unwrap(); } #[test] fn gauge_returns_most_recent_value_per_label_combination() { let mut index = BTreeMap::new(); declare_durations_metric(&mut index); ingest_log(&mut index, 0, "request path=/a duration=5"); ingest_log(&mut index, 1, "request path=/a duration=9"); ingest_log(&mut index, 2, "request path=/b duration=42"); let parsed = promql::parse_query("gauge(durations)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].points, vec![(1, 9.0)]); assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/b")); assert_eq!(series[1].points, vec![(2, 42.0)]); } #[test] fn gauge_respects_time_range_bounds() { let mut index = BTreeMap::new(); declare_durations_metric(&mut index); ingest_log(&mut index, 0, "request path=/a duration=5"); ingest_log(&mut index, 5, "request path=/a duration=9"); ingest_log(&mut index, 10, "request path=/a duration=42"); let parsed = promql::parse_query("gauge(durations)").unwrap(); let series = evaluate_promql(&index, &parsed, None, Some(9)); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(5, 9.0)]); } #[test] fn histogram_buckets_captured_values_by_fixed_width() { let mut index = BTreeMap::new(); declare_durations_metric(&mut index); for (seq, duration) in [(0, 2), (1, 4), (2, 12), (3, 15), (4, 21)] { ingest_log(&mut index, seq, &format!("request path=/a duration={duration}")); } let parsed = promql::parse_query("histogram(durations[10])").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None); series.sort_by(|a, b| a.labels.get("le").cmp(&b.labels.get("le"))); // durations 2,4 -> le=10 (count 2); 12,15 -> le=20 (count 2); 21 -> le=30 (count 1). assert_eq!(series.len(), 3); assert_eq!(series[0].labels.get("le").map(String::as_str), Some("10")); assert_eq!(series[0].points[0].1, 2.0); assert_eq!(series[1].labels.get("le").map(String::as_str), Some("20")); assert_eq!(series[1].points[0].1, 2.0); assert_eq!(series[2].labels.get("le").map(String::as_str), Some("30")); assert_eq!(series[2].points[0].1, 1.0); } #[test] fn histogram_with_by_clause_buckets_each_label_group_separately() { let mut index = BTreeMap::new(); declare_durations_metric(&mut index); // path=/a: 2,4 -> le=10 (count 2); 12 -> le=20 (count 1). // path=/b: 21 -> le=30 (count 1). for (seq, path, duration) in [ (0, "/a", 2), (1, "/a", 4), (2, "/a", 12), (3, "/b", 21), ] { ingest_log(&mut index, seq, &format!("request path={path} duration={duration}")); } let parsed = promql::parse_query("histogram(durations[10]) by (path)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None); series.sort_by(|a, b| { a.labels.get("path").cmp(&b.labels.get("path")).then(a.labels.get("le").cmp(&b.labels.get("le"))) }); assert_eq!(series.len(), 3); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].labels.get("le").map(String::as_str), Some("10")); assert_eq!(series[0].points[0].1, 2.0); assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[1].labels.get("le").map(String::as_str), Some("20")); assert_eq!(series[1].points[0].1, 1.0); assert_eq!(series[2].labels.get("path").map(String::as_str), Some("/b")); assert_eq!(series[2].labels.get("le").map(String::as_str), Some("30")); assert_eq!(series[2].points[0].1, 1.0); } fn declare_login_duration_metric(index: &mut BTreeMap) { declare_metric( index, metric_request_with_roles( "login_gap", r"user=(?P\S+) logged in", vec![ ("user", "user", MetricPrimitiveKind::String, ParamRole::Label), ("gap", "user", MetricPrimitiveKind::Duration, ParamRole::Value), ], ), ) .unwrap(); } #[test] fn duration_value_param_computes_elapsed_time_since_previous_same_labels_event() { let mut index = BTreeMap::new(); declare_login_duration_metric(&mut index); ingest_log(&mut index, 0, "user=alice logged in"); ingest_log(&mut index, 1, "user=bob logged in"); ingest_log(&mut index, 2, "user=alice logged in"); let values: Vec<_> = bucket_values(&index, "login_gap", None).collect(); // First occurrences (alice@0, bob@1) produce no duration value — only alice's second occurrence (seq 2) has something to diff against. assert_eq!(values.len(), 1); let (labels, (sequence, timestamp, value)) = &values[0]; assert_eq!(labels.get("user").map(|v| v.to_string()), Some("alice".to_string())); assert_eq!(*sequence, 2); // `ingest_log`'s helper timestamps by `sequence as millis`. assert_eq!(timestamp.as_millis(), 2); assert_eq!(*value, MetricPrimitive::Duration(2)); } #[test] fn duration_diffs_against_the_immediately_preceding_event_not_the_first() { let mut index = BTreeMap::new(); declare_login_duration_metric(&mut index); ingest_log(&mut index, 0, "user=alice logged in"); ingest_log(&mut index, 3, "user=alice logged in"); ingest_log(&mut index, 9, "user=alice logged in"); let values: Vec<_> = bucket_values(&index, "login_gap", None).collect(); assert_eq!(values.len(), 2); assert_eq!(values[0].1.2, MetricPrimitive::Duration(3)); assert_eq!(values[1].1.2, MetricPrimitive::Duration(6)); } #[test] fn duration_values_are_queryable_via_promql_like_any_other_metric() { let mut index = BTreeMap::new(); declare_login_duration_metric(&mut index); ingest_log(&mut index, 0, "user=alice logged in"); ingest_log(&mut index, 10, "user=alice logged in"); let parsed = promql::parse_query("login_gap").unwrap(); let series = evaluate_promql(&index, &parsed, None, None); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(10, 10.0)]); } // Fixed labels (not a captured param) so a case-insensitive match collapses into one clean "error" series regardless of source-text case. fn declare_error_category_metric(index: &mut BTreeMap) { declare_metric( index, DeclareMetricRequest { name: "log_category".to_string(), message_match_regex: r"(?i)error".to_string(), params: vec![], matched_labels: Some(BTreeMap::from([("category".to_string(), "error".to_string())])), unmatched_labels: Some(BTreeMap::from([("category".to_string(), "other".to_string())])), }, ) .unwrap(); } #[test] fn unmatched_labels_registers_occurrence_for_non_matching_lines() { let mut index = BTreeMap::new(); declare_error_category_metric(&mut index); ingest_log(&mut index, 0, "everything is fine"); ingest_log(&mut index, 1, "an ERROR occurred"); ingest_log(&mut index, 2, "still fine"); let error_labels = BTreeMap::from([("category".to_string(), MetricPrimitive::String("error".to_string()))]); let error_values: Vec<_> = bucket_values(&index, "log_category", Some(&error_labels)).collect(); assert_eq!(error_values.len(), 1); assert_eq!(error_values[0].1.0, 1); let other_labels = BTreeMap::from([("category".to_string(), MetricPrimitive::String("other".to_string()))]); let other_values: Vec<_> = bucket_values(&index, "log_category", Some(&other_labels)).collect(); assert_eq!(other_values.len(), 2); } #[test] fn unmatched_labels_produces_two_series_from_one_bare_promql_query() { let mut index = BTreeMap::new(); declare_error_category_metric(&mut index); ingest_log(&mut index, 0, "an ERROR occurred"); ingest_log(&mut index, 1, "all good"); ingest_log(&mut index, 2, "all good again"); let parsed = promql::parse_query("log_category").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None); series.sort_by(|a, b| a.labels.get("category").cmp(&b.labels.get("category"))); assert_eq!(series.len(), 2); assert_eq!(series[0].labels.get("category").map(String::as_str), Some("error")); assert_eq!(series[0].points.len(), 1); assert_eq!(series[1].labels.get("category").map(String::as_str), Some("other")); assert_eq!(series[1].points.len(), 2); } #[test] fn no_unmatched_labels_means_non_matching_lines_are_untouched() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "not a request line at all"); assert_eq!(bucket_values(&index, "requests", None).count(), 0); } #[test] fn handle_query_promql_rejects_malformed_query() { let index = BTreeMap::new(); let response = handle_query( &index, EngineQuery::Promql { query: "not a valid promql query(".to_string(), start_ms: None, end_ms: None, }, 0, ); match response { EngineResponse::Promql(Err(message)) => assert!(!message.is_empty()), _ => panic!("expected EngineResponse::Promql(Err(_))"), } } #[test] fn handle_query_promql_returns_series_for_valid_query() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); let response = handle_query( &index, EngineQuery::Promql { query: r#"requests{path="/a"}"#.to_string(), start_ms: None, end_ms: None, }, 0, ); match response { EngineResponse::Promql(Ok(series)) => { assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 1.0)]); } _ => panic!("expected EngineResponse::Promql(Ok(_))"), } } // -- 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")"#)), 0) { EngineResponse::Logs(entries) => { let sequences: Vec = 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")), 0) { EngineResponse::Logs(entries) => { let sequences: Vec = 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")), 0) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![3, 4]); } _ => panic!("expected Logs response"), } } #[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 = 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 = 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(); declare_metric( &mut index, metric_request( "requests", r"level=(?P\w+) duration=(?P\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; note comparison values are bare tokens (level==error), not quoted string literals like regex()'s argument. match handle_query( &index, logs_query(Some(r#"regex("boom") and requests[level]==error"#)), 0, ) { EngineResponse::Logs(entries) => { let sequences: Vec = 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"#)), 0, ) { EngineResponse::Logs(entries) => { let sequences: Vec = 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(); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); 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) else { panic!("expected Logs response"); }; assert_eq!(entries.len(), 2); assert_eq!( entries[0].params.get("level"), Some(&MetricPrimitive::String("info".to_string())) ); assert!(entries[1].params.get("level").is_none()); } #[test] fn boolean_param_declares_without_capture_group_and_is_always_present() { let mut index = BTreeMap::new(); let mut req = request("IsError", r"(?i)error", ""); req.kind = MetricPrimitiveKind::Boolean; assert!(declare_param(&mut index, req).is_ok()); 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) else { panic!("expected Logs response"); }; assert_eq!(entries.len(), 2); assert_eq!(entries[0].params.get("IsError"), Some(&MetricPrimitive::Boolean(true))); assert_eq!(entries[1].params.get("IsError"), Some(&MetricPrimitive::Boolean(false))); } #[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), 0) { 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 @@@")), 0) { 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) { 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\w+)", vec![("level", "level", MetricPrimitiveKind::String)], ), ) .unwrap(); ingest_log(&mut index, 0, "level=error"); match handle_query(&index, logs_query(Some("requests==5")), 0) { EngineResponse::Error(message) => assert!(message.contains("param")), _ => panic!("expected an Error response for a metric comparison without [param]"), } } }