diff --git a/src/engine.rs b/src/engine.rs index 10cbe55..544cb14 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -4,6 +4,7 @@ use crate::dto::{ TimelineBucketDto, TimelineParams, }; use crate::index_key::IndexKey; +use crate::metric_primitive::{MetricPrimitive, Number}; use crate::timestamp::Timestamp; use regex::Regex; use std::collections::BTreeMap; @@ -87,13 +88,12 @@ pub async fn run(mut rx: mpsc::Receiver) { while let Some(message) = rx.recv().await { match message { EngineMessage::Ingest(line) => { + let IngestedLine { sequence, timestamp, text } = line; index.insert( - IndexKey::IngestedLog { - sequence: line.sequence, - timestamp: line.timestamp, - }, - DataValue::LogLiteral { text: line.text }, + 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); @@ -173,6 +173,7 @@ fn declare_metric( return Err(errors.join("; ")); } + backfill_metric(index, &request.name, &re, &request.params); index.insert( IndexKey::MetricDeclaration { name: request.name }, DataValue::MetricDeclaration { @@ -183,6 +184,134 @@ fn declare_metric( 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 +/// label combination. Internal helper for now — the metrics-API ticket ([004]) will +/// wrap this (or an `EngineQuery` built on it) with time/sequence-range filtering and +/// HTTP plumbing. +#[allow(dead_code)] +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 { @@ -495,4 +624,157 @@ mod tests { 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); + } }