Populate metric buckets from declarations and log ingest (ticket 003)

Runs each declared metric's regex against log text both retroactively
(backfill on DeclareMetric) and going forward (on each Ingest), appending
(sequence, timestamp, value) entries to IndexKey::Bucket/DataValue::BucketValue.

All of a metric's declared params become the Bucket key's labels (the only
structured place for them), and the stored value is a fixed presence marker
(Number::Int(1)) — a bucket's values are an occurrence timeseries for that
label combination, not the captured values themselves. A param whose capture
doesn't match or fails to parse under its declared kind skips the whole match
rather than storing a partially-labeled bucket.

Exposes an internal bucket_values(index, metric_name, labels_filter) helper
for the upcoming metrics-API ticket to build on.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o

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<EngineMessage>) {
     while let Some(message) = rx.recv().await {
         match message {
             EngineMessage::Ingest(line) => {
+                let IngestedLine { sequence, timestamp, text } = line;
                 index.insert(
-                    IndexKey::IngestedLog {
+                    IndexKey::IngestedLog { sequence, timestamp },
-                        sequence: line.sequence,
+                    DataValue::LogLiteral { text: text.clone() },
-                        timestamp: line.timestamp,
-                    },
-                    DataValue::LogLiteral { text: line.text },
                 );
+                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<MetricPrimitive> {
+    match kind {
+        MetricPrimitiveKind::String => Some(MetricPrimitive::String(raw.to_string())),
+        MetricPrimitiveKind::Number => raw
+            .parse::<i64>()
+            .map(Number::Int)
+            .or_else(|_| raw.parse::<f64>().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<IndexKey, DataValue>,
+    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<IndexKey, DataValue>,
+    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<IndexKey, DataValue>,
+    sequence: u64,
+    timestamp: Timestamp,
+    text: &str,
+) {
+    let metrics: Vec<(String, String, Vec<RegexToMetricPrimitiveMatcher>)> =
+        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, &params, 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<IndexKey, DataValue>,
+    metric_name: &'a str,
+    labels_filter: Option<&'a BTreeMap<String, MetricPrimitive>>,
+) -> impl Iterator<Item = (&'a BTreeMap<String, MetricPrimitive>, &'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<IndexKey, DataValue>,
 ) -> impl Iterator<Item = (&str, &str, &[RegexToMetricPrimitiveMatcher])> {
@@ -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<IndexKey, DataValue>, 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<path>\S+) duration=(?P<duration>\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<path>\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<path>\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<path>\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<u64> = 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<path>\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<path>\S+) duration=(?P<duration>\d+)",
+                vec![("path", "path", MetricPrimitiveKind::String)],
+            ),
+        )
+        .unwrap();
+        declare_metric(
+            &mut index,
+            metric_request(
+                "durations",
+                r"request path=(?P<path>\S+) duration=(?P<duration>\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);
+    }
 }