use crate::data_value::{DataValue, MetricPrimitiveKind, RegexToMetricPrimitiveMatcher}; use crate::dto::{ BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, PromqlSeriesDto, StatsDto, TimelineBucketDto, TimelineParams, }; use crate::index_key::IndexKey; use crate::metric_primitive::{MetricPrimitive, Number}; use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery}; use crate::timestamp::Timestamp; use regex::Regex; use serde::Deserialize; use std::collections::BTreeMap; use tokio::sync::{mpsc, oneshot}; pub struct IngestedLine { pub sequence: u64, pub timestamp: Timestamp, pub text: String, } /// Also the request body for `POST /api/params` (ticket 004): `Deserialize` is /// derived directly on this engine type rather than adding a separate DTO, /// since its shape is already exactly what the API needs. #[derive(Debug, Clone, Deserialize)] pub struct DeclareParamRequest { pub name: String, pub message_match_regex: String, pub capture_group: String, pub kind: MetricPrimitiveKind, } /// v1 mechanism: a metric declaration's params are derived purely from inline named /// capture groups on the metric's own `message_match_regex` (each /// `RegexToMetricPrimitiveMatcher::capture_group` must name a capture group on that /// regex). Referencing an already-declared `ParamDeclaration` (001) by name is not /// supported yet — that's left for a future ticket. /// /// Also the request body for `POST /api/metrics` (ticket 004) — see /// `DeclareParamRequest`'s doc comment for why `Deserialize` lives here rather /// than on a separate DTO. #[derive(Debug, Clone, Deserialize)] pub struct DeclareMetricRequest { pub name: String, pub message_match_regex: String, pub params: Vec, } #[allow(dead_code)] pub enum EngineQuery { Logs(LogsParams), Timeline(TimelineParams), Stats, ParamDeclarations, MetricDeclarations, /// Bucket values for one metric (ticket 004's `GET /api/metrics/:name`). /// `labels_filter`, when present, keeps only buckets where every listed /// label matches — compared against each label value's `Display` form /// (see `MetricPrimitive`) rather than 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, }, /// PromQL query (ticket 013's `GET /api/query`). `query` is parsed here (rather /// than by the API handler) so parsing and evaluation both happen inside the /// single-owner engine loop, matching every other `EngineQuery` variant's /// pattern of evaluating in place against `index` with no copying. A malformed /// query is therefore only discovered once this reaches `handle_query`, hence /// `EngineResponse::Promql` carrying a `Result` rather than a bare `Vec`. Promql { query: String, start_ms: Option, end_ms: Option }, } #[allow(dead_code)] pub enum EngineMessage { Ingest(IngestedLine), DeclareParam { request: DeclareParamRequest, respond_to: oneshot::Sender>, }, RemoveParamDeclaration { name: String, respond_to: oneshot::Sender, }, DeclareMetric { request: DeclareMetricRequest, respond_to: oneshot::Sender>, }, RemoveMetricDeclaration { name: String, respond_to: oneshot::Sender, }, 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 (see /// `crate::promql::parse_query`), which the API layer maps to a 400 response. Promql(Result, String>), } pub type EngineHandle = mpsc::Sender; /// Sole owner of the primary index. Runs single-threaded so the map never needs a /// lock, and does its own response serialization/DTO construction so callers never /// have to clone the index to answer a query. pub async fn run(mut rx: mpsc::Receiver) { let mut index: BTreeMap = BTreeMap::new(); while let Some(message) = rx.recv().await { match message { EngineMessage::Ingest(line) => { let IngestedLine { sequence, timestamp, text } = line; index.insert( IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text: text.clone() }, ); apply_all_metrics(&mut index, sequence, timestamp, &text); } 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::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); let _ = respond_to.send(response); } } } } /// Validates the regex and capture group before storing, so a bad declaration is /// rejected up front rather than silently stored (or discovered later at derivation /// time). 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}"))?; if !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, kind: request.kind, }, ); Ok(()) } /// Validates the regex and every param matcher's capture-group reference before /// storing. All matchers are checked (not just the first bad one) so a rejection can /// report every problem at once. fn declare_metric( index: &mut BTreeMap, request: DeclareMetricRequest, ) -> Result<(), String> { let re = Regex::new(&request.message_match_regex) .map_err(|err| format!("invalid regex: {err}"))?; let errors: Vec = request .params .iter() .filter(|matcher| { !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("; ")); } backfill_metric(index, &request.name, &re, &request.params); index.insert( IndexKey::MetricDeclaration { name: request.name }, DataValue::MetricDeclaration { message_match_regex: request.message_match_regex, params: request.params, }, ); Ok(()) } /// Parses a single captured substring into the `MetricPrimitive` its param declares. /// Returns `None` on a parse failure (e.g. a `Number` param whose capture isn't /// numeric), 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) } } } /// Runs one metric's regex against a single log line and, on a match, appends a /// bucket value entry. /// /// Design decision: `Bucket.labels` is the only structured place a `Bucket` key has /// for a metric's declared params, so v1 folds ALL of a metric's params into /// `labels` rather than splitting them into "label" vs "value" params. The stored /// value is therefore a fixed presence marker (`MetricPrimitive::Number(Number::Int(1))`) /// — a bucket's `values` is the timeseries of *when that label combination occurred* /// (a count/occurrence series), not a record of the captured values themselves. If a /// param's capture group doesn't match, or its captured text fails to parse as its /// declared `kind`, the whole match is skipped for this line rather than storing a /// partially-labeled bucket. fn apply_metric_to_log( index: &mut BTreeMap, metric_name: &str, regex: &Regex, params: &[RegexToMetricPrimitiveMatcher], sequence: u64, timestamp: Timestamp, text: &str, ) { let Some(captures) = regex.captures(text) else { return; }; let mut labels = BTreeMap::new(); for matcher in params { let Some(raw) = captures.name(&matcher.capture_group) else { return; }; let Some(value) = parse_metric_primitive(raw.as_str(), matcher.kind) else { return; }; labels.insert(matcher.param_name.clone(), value); } 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, MetricPrimitive::Number(Number::Int(1)))); } } /// Applies a newly-declared metric to every already-ingested log, so a metric /// declared after logs have already arrived still gets populated buckets for them /// (backfill), rather than only reflecting logs ingested from that point on. fn backfill_metric( index: &mut BTreeMap, metric_name: &str, regex: &Regex, params: &[RegexToMetricPrimitiveMatcher], ) { 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, sequence, timestamp, &text); } } /// Runs every currently-declared metric's regex against one newly-ingested log line, /// updating whichever buckets match. A line matching zero metrics leaves the bucket /// range untouched. fn apply_all_metrics( index: &mut BTreeMap, sequence: u64, timestamp: Timestamp, text: &str, ) { let metrics: Vec<(String, String, Vec)> = metric_declarations_range(index) .map(|(name, message_match_regex, params)| { (name.to_string(), message_match_regex.to_string(), params.to_vec()) }) .collect(); for (name, message_match_regex, params) in metrics { if let Ok(re) = Regex::new(&message_match_regex) { apply_metric_to_log(index, &name, &re, ¶ms, sequence, timestamp, text); } } } /// Bucket values for one metric, in insertion order, optionally filtered to a single /// exact label combination. Used directly by tests (exact-match filtering) and, via /// `EngineQuery::BucketValues` in `handle_query` below, by the metrics API (ticket /// 004) — which layers its own textual label-filter and time/sequence-range /// filtering on top rather than using this function's `labels_filter` parameter. 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() } fn metric_declarations_range( index: &BTreeMap, ) -> impl Iterator { index .range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit) .filter_map(|(key, value)| match (key, value) { ( IndexKey::MetricDeclaration { name }, DataValue::MetricDeclaration { message_match_regex, params }, ) => Some((name.as_str(), message_match_regex.as_str(), params.as_slice())), _ => None, }) } fn param_declarations_range( index: &BTreeMap, ) -> impl Iterator { index .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit) .filter_map(|(key, value)| match (key, value) { ( IndexKey::ParamDeclaration { name }, DataValue::ParamDeclaration { message_match_regex, capture_group, kind }, ) => Some((name.as_str(), message_match_regex.as_str(), capture_group.as_str(), *kind)), _ => None, }) } 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) -> EngineResponse { match query { EngineQuery::Logs(params) => { let limit = params.limit.unwrap_or(500).min(10_000); 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) }) .take(limit) .map(|(sequence, timestamp, text)| LogEntryDto { sequence, timestamp_ms: timestamp.as_millis(), text: text.to_string(), }) .collect(); EngineResponse::Logs(entries) } EngineQuery::Timeline(params) => { let bucket_ms = params.bucket_ms.unwrap_or(1_000).max(1); let mut buckets: BTreeMap = BTreeMap::new(); for (_, timestamp, _) 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) }) { 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, }) } EngineQuery::ParamDeclarations => { let declarations = param_declarations_range(index) .map(|(name, message_match_regex, capture_group, kind)| ParamDeclarationDto { name: name.to_string(), message_match_regex: message_match_regex.to_string(), capture_group: capture_group.to_string(), kind, }) .collect(); EngineResponse::ParamDeclarations(declarations) } EngineQuery::MetricDeclarations => { let declarations = metric_declarations_range(index) .map(|(name, message_match_regex, params)| MetricDeclarationDto { name: name.to_string(), message_match_regex: message_match_regex.to_string(), params: params.to_vec(), }) .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 stored `MetricPrimitive` into a plottable `f64` for /// `PromqlSeriesDto`: /// - `Number` uses its underlying numeric value. /// - `Timestamp` uses its millisecond value (arbitrary but at least meaningful/ /// orderable, since there's no other sensible single number for a point in time). /// - `String` has no numeric reading at all, so it falls back to `1.0` — i.e. it is /// treated the same as the presence/occurrence marker `apply_metric_to_log` already /// stores for every bucket value today (ticket 003), meaning a PromQL series over a /// `String`-valued bucket becomes an occurrence-count series rather than being /// skipped. This keeps every declared metric queryable (matching the ticket's /// "documented fallback" requirement) rather than silently omitting non-numeric /// series. 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::String(_) => 1.0, } } /// True when every matcher in `matchers` is satisfied by `labels` (compared against /// each label value's `Display` form, same convention as `EngineQuery::BucketValues`'s /// `labels_filter`). Unmatched/extra labels on the bucket are ignored — a 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) }) } /// Evaluates a parsed PromQL query against `index`, producing chart-ready series. /// /// Baseline (no aggregation wrapper): one output series per distinct label /// combination that matches `parsed.metric_name` + `parsed.matchers`, with that /// combination's bucket values as `(timestamp_ms, value)` points (see /// `coerce_to_f64`), sorted by timestamp. /// /// `sum()`/`avg()`: merges every matching series' points by exact millisecond /// timestamp (summing, or averaging, the values that land on the same timestamp /// across series) into a single output series with empty labels — the aggregation /// collapses the label dimension, same as PromQL's `sum by ()`/`avg by ()` default. /// /// `rate()`: requires (and, per `promql::parse_query`, is only ever given) a range /// selector. This is a simplified, single-point-per-series rate rather than PromQL's /// true sliding-window-per-evaluation-step semantics: for each matching series, it /// takes the window `(last_point_ts - range, last_point_ts]` and divides the number /// of points falling in that window by the range's duration in seconds, producing one /// `(last_point_ts, rate)` point per series. Since bucket values are occurrence /// markers rather than cumulative counters (ticket 003), an occurrence-count-per- /// second reading is the meaningful analogue of PromQL's counter-increase rate here. 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 mut grouped: BTreeMap> = BTreeMap::new(); for points in series_points.values() { for (timestamp_ms, value) in points { grouped.entry(*timestamp_ms).or_default().push(*value); } } let is_avg = parsed.agg == Some(AggOp::Avg); let points = 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(); vec![PromqlSeriesDto { labels: BTreeMap::new(), points }] } 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; let range_seconds = range_ms as f64 / 1000.0; series_points .into_iter() .filter_map(|(labels, points)| { let last_ts = points.last()?.0; let window_start = last_ts - range_ms; let count_in_window = points.iter().filter(|(ts, _)| *ts > window_start && *ts <= last_ts).count(); let rate = count_in_window as f64 / range_seconds; Some(PromqlSeriesDto { labels, points: vec![(last_ts, rate)] }) }) .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(), 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 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![], }, ); 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 { DeclareMetricRequest { name: name.to_string(), message_match_regex: pattern.to_string(), params: params .into_iter() .map(|(param_name, capture_group, kind)| RegexToMetricPrimitiveMatcher { param_name: param_name.to_string(), capture_group: capture_group.to_string(), kind, }) .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()); } 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 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_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 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, }, ); 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, }, ); 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(_))"), } } }