From 756e054527fafecd56d5c58dc12a408588d2b461 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Mon, 7 Sep 2026 13:27:49 +0100 Subject: [PATCH] Extract param extraction into its own EngineSnapshot::ingest_params stage (ticket 101) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Raw storage, log-stream header derivation, and auto-param detection were inlined directly in the ingest actor's write closure alongside metric computation. Pulling them into their own method gives the ingest arm two clearly separated, independently-callable stages — param extraction (ingest_params) and metric computation (apply_all_metrics) — with no behavior change, in preparation for splitting these onto separate actors (ticket 102). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ --- src/engine/handle.rs | 20 +----- src/engine/snapshot/mod.rs | 30 +++++++++ ...-metric-computation-into-separate-stage.md | 62 ------------------- 3 files changed, 33 insertions(+), 79 deletions(-) delete mode 100644 tickets/101-extract-metric-computation-into-separate-stage.md diff --git a/src/engine/handle.rs b/src/engine/handle.rs index 4e512c4..167e4a4 100644 --- a/src/engine/handle.rs +++ b/src/engine/handle.rs @@ -4,7 +4,6 @@ use crate::config::SettingsHandle; use crate::consts::{INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS}; use crate::engine::alerts::{self, AlertRuntimeState}; -use crate::engine::data_value::DataValue; use crate::engine::index_key::IndexKey; use crate::engine::message::EngineMessage; use crate::engine::metric_primitive::{MetricPrimitive, Number}; @@ -12,7 +11,6 @@ use crate::engine::snapshot::EngineSnapshot; use crate::engine::snapshot::params::log_stream_header_transition; use crate::engine::timestamp::Timestamp; use crate::engine::{ClearParamsResult, IngestedLine}; -use crate::flavour::LogLiteralHelper; use left_right::Absorb; use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; @@ -179,21 +177,9 @@ async fn run( let text_for_op = Arc::clone(&text); write.mutate(move |snap: &mut EngineSnapshot| { - snap.approx_bytes_total += text_for_op.approx_log_bytes(); - snap.index.insert( - IndexKey::IngestedLog { - sequence, - timestamp, - }, - DataValue::LogLiteral { - text: Arc::clone(&text_for_op), - }, - ); - snap.log_count += 1; - if let Some(header) = row_header.as_ref() { - snap.derive_log_stream_params_from_row(header, &text_for_op); - } - snap.auto_declare_params(&text_for_op); + // Two independently-callable stages (ticket 101): primary/param + // extraction, then profile-scoped metric computation. + snap.ingest_params(sequence, timestamp, &text_for_op, row_header.as_ref()); snap.apply_all_metrics(sequence, timestamp, &text_for_op); snap.update_sort_helper_indexes_on_ingest(sequence, timestamp, &text_for_op); snap.purge_stale_sort_helper_indexes(); diff --git a/src/engine/snapshot/mod.rs b/src/engine/snapshot/mod.rs index 3dda21b..a32a8bd 100644 --- a/src/engine/snapshot/mod.rs +++ b/src/engine/snapshot/mod.rs @@ -45,6 +45,7 @@ use sort::{ derive_single_param, }; use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::Arc; /// One row of a metric declaration: name, message-match regex, label param names, value param /// name, and the label-name-to-value-param and static-label maps. @@ -812,6 +813,35 @@ impl EngineSnapshot { .count() } + /// The primary/param-extraction stage of ingesting one line: raw storage, log-stream header + /// derivation, and auto-param detection — deliberately excludes metric computation + /// (`apply_all_metrics`), which is the profile-scoped stage a caller applies separately, so + /// the two remain independently callable (ticket 100/101's precondition for eventually + /// running them on different actors, see `ideas/0001-actor-split.md`). + pub(crate) fn ingest_params( + &mut self, + sequence: u64, + timestamp: Timestamp, + text: &Arc, + row_header: Option<&crate::engine::logfmt::LogStreamHeader>, + ) { + self.approx_bytes_total += text.approx_log_bytes(); + self.index.insert( + IndexKey::IngestedLog { + sequence, + timestamp, + }, + DataValue::LogLiteral { + text: Arc::clone(text), + }, + ); + self.log_count += 1; + if let Some(header) = row_header { + self.derive_log_stream_params_from_row(header, text); + } + self.auto_declare_params(text); + } + pub(crate) fn auto_declare_params(&mut self, text: &str) { let mut total = self.auto_derived_param_count(); if total >= MAX_AUTO_DECLARED_PARAMS { diff --git a/tickets/101-extract-metric-computation-into-separate-stage.md b/tickets/101-extract-metric-computation-into-separate-stage.md deleted file mode 100644 index 81a92eb..0000000 --- a/tickets/101-extract-metric-computation-into-separate-stage.md +++ /dev/null @@ -1,62 +0,0 @@ -# 101 — Metric computation is inlined into the single ingest actor's write closure, not a separately-callable stage - -## Summary - -Per-line ingest currently does ingest-lag bookkeeping, log-stream -header detection, raw-text storage, auto-param detection, *and* every -declared metric's regex evaluation, all synchronously inside one -`write.mutate(...)` closure in the actor's `run()` loop, followed by -alert evaluation against the result in the same message handler. This -is migration step 2 of [[0001-actor-split]]'s resolved plan: pull -metric computation out into a stage that can be called independently -of the ingest closure — still within the single actor for now — so -that step 3 (splitting into reader/primary/profile actors) is a -wiring change rather than also being the first time metric computation -is separated from raw ingest. Depends on [[100-arc-ify-stored-log-and-metric-strings]] -landing first (Arc'd text is what makes passing a line into a separate -stage cheap). - -## Where - -- `src/engine/handle.rs:132-224` (`run()`'s `EngineMessage::Ingest` - arm) — the closure passed to `write.mutate` at :180-198 does header - detection, storage, `auto_declare_params`, and - `snap.apply_all_metrics(sequence, timestamp, &text_for_op)` (:196) - all in one synchronous block; alert evaluation follows immediately - after at :199-219, reading the snapshot metric computation just - produced. -- `src/engine/snapshot/mod.rs:616` (`apply_all_metrics`) — the entry - point that iterates every declared metric and calls - `apply_metric_to_log` (:499-586) per match; currently only reachable - as a method on `EngineSnapshot` invoked from inside the ingest - closure above. -- `src/engine/alerts.rs:211` (`evaluate_alerts`) — currently called - directly from the ingest arm (`handle.rs:199-219`) against the - snapshot mutated by the same message; this is the alert-evaluation - call site that migration step 3 will later move to a profile actor - (per [[0001-actor-split]]'s "Resolved decisions" — not in scope for - this ticket). - -## Scope - -- Extract param-extraction (header detection, `auto_declare_params`, - raw storage) and metric-computation (`apply_all_metrics`) into - separate, independently-callable functions/methods with a clear - boundary between them — e.g. a function that takes a line and - returns/records derived params only, and a second that takes the - declared-metrics set plus a param-resolved line and returns bucket - updates — rather than one closure that does both inline. -- Keep both stages running inside the same actor and the same - `write.mutate` transaction for this ticket — the goal is a clean - internal seam, not concurrency yet. Step 3 (a separate ticket, not - this one) is what actually moves metric computation to a different - task. -- Do not change alert evaluation's behavior or its current call site - relative to metric computation (still immediately after, same - actor) — only prepare the metric-computation code so it's callable - as its own unit; moving alerts to a profile actor is explicitly step - 3's concern per [[0001-actor-split]]'s resolved decisions. -- Out of scope: the reader/primary/profile actor split itself, any new - `tokio::sync::watch` signaling, and any change to `EngineHandle`'s - public API — this ticket is an internal refactor within the existing - single-actor structure.