diff --git a/src/api/mod.rs b/src/api/mod.rs index ad2c1a3..c602376 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -249,6 +249,21 @@ fn mime_for_extension(extension: &str) -> &'static str { impl EngineHandle { async fn ask(&self, query: EngineQuery) -> Result { + // Metric declarations and bucket data live on the profile actor (ticket 104), not the + // primary -- route those query variants through its reader instead. + if matches!( + query, + EngineQuery::MetricDeclarations + | EngineQuery::BucketValues { .. } + | EngineQuery::Promql { .. } + ) { + let profile_handle = self.profile_reader.handle(); + let guard = profile_handle + .enter() + .ok_or(StatusCode::SERVICE_UNAVAILABLE)?; + return Ok(guard.handle_query_ro(query)); + } + let read_handle = self.reader.handle(); if let Some(sort_by) = query.sort_by() { let needs_build = match read_handle.enter() { diff --git a/src/config/profile.rs b/src/config/profile.rs index f9ef68f..9bccdc7 100644 --- a/src/config/profile.rs +++ b/src/config/profile.rs @@ -751,7 +751,19 @@ mod tests { } async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse { - let read_handle = engine.reader.handle(); + // Metric declarations and bucket data live on the profile actor (ticket 104), not the + // primary -- mirrors the routing in `api::EngineHandle::ask`. + let profile_query = matches!( + query, + EngineQuery::MetricDeclarations + | EngineQuery::BucketValues { .. } + | EngineQuery::Promql { .. } + ); + let read_handle = if profile_query { + engine.profile_reader.handle() + } else { + engine.reader.handle() + }; let guard = read_handle.enter().expect("engine unavailable"); guard.handle_query_ro(query) } diff --git a/src/engine/handle.rs b/src/engine/handle.rs index 1f75c8c..a1f9b75 100644 --- a/src/engine/handle.rs +++ b/src/engine/handle.rs @@ -61,7 +61,7 @@ impl IndexReadHandle { } pub struct IndexReader { - factory: left_right::ReadHandleFactory, + pub(crate) factory: left_right::ReadHandleFactory, } impl Clone for IndexReader { @@ -81,6 +81,11 @@ impl IndexReader { pub struct EngineHandle { pub tx: mpsc::Sender, pub reader: IndexReader, + /// Read-only view of the profile actor's snapshot (`super::profile`), which owns metric + /// declarations and bucket data (ticket 104) — `MetricDeclarations`/`BucketValues`/`Promql` + /// queries read through this instead of `reader`. Not part of the write path; the profile + /// actor's own `mpsc` channel (proxied through `EngineMessage`'s metric variants) is that. + pub(crate) profile_reader: IndexReader, } impl Clone for EngineHandle { @@ -88,6 +93,7 @@ impl Clone for EngineHandle { EngineHandle { tx: self.tx.clone(), reader: self.reader.clone(), + profile_reader: self.profile_reader.clone(), } } } @@ -106,13 +112,16 @@ impl EngineHandle { } } -/// Builds the handle/receiver/writer/version-receiver quadruple shared by `spawn` and, in -/// tests, by callers that want the raw receiver without an actor loop draining it. +/// Builds the primary actor's sender/receiver/reader/writer/version-receiver, shared by `spawn` +/// and, in tests, by callers that want the raw receiver without an actor loop draining it. +/// Stops short of building `EngineHandle` itself, since that also needs a profile-actor reader +/// that doesn't exist until the caller spawns (or fakes, see `test_handle`) the profile actor. fn new_channel_and_handle( channel_capacity: usize, ) -> ( - EngineHandle, + mpsc::Sender, mpsc::Receiver, + IndexReader, IndexWriter, watch::Receiver, ) { @@ -123,8 +132,9 @@ fn new_channel_and_handle( }; let (version_tx, version_rx) = watch::channel(0u64); ( - EngineHandle { tx, reader }, + tx, rx, + reader, IndexWriter { handle: write, version: 0, @@ -135,9 +145,14 @@ fn new_channel_and_handle( } pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle { - let (handle, rx, write, version_rx) = new_channel_and_handle(channel_capacity); - crate::engine::profile::spawn(handle.reader.clone(), version_rx); - tokio::spawn(run(rx, settings, write)); + let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity); + let profile = crate::engine::profile::spawn(reader.clone(), version_rx); + let handle = EngineHandle { + tx, + reader, + profile_reader: profile.reader.clone(), + }; + tokio::spawn(run(rx, settings, write, profile)); handle } @@ -145,7 +160,14 @@ pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle pub(crate) fn test_handle( channel_capacity: usize, ) -> (EngineHandle, mpsc::Receiver) { - let (handle, rx, _write, _version_rx) = new_channel_and_handle(channel_capacity); + let (tx, rx, reader, _write, _version_rx) = new_channel_and_handle(channel_capacity); + let handle = EngineHandle { + tx, + reader, + // No actor loop is spawned here at all, so nothing ever publishes -- an inert reader + // over an always-empty snapshot is all a query through this handle could see anyway. + profile_reader: crate::engine::profile::inert_reader(), + }; (handle, rx) } @@ -156,9 +178,14 @@ pub(crate) fn spawn_with_profile_observer( settings: SettingsHandle, channel_capacity: usize, ) -> (EngineHandle, tokio::sync::mpsc::UnboundedReceiver) { - let (handle, rx, write, version_rx) = new_channel_and_handle(channel_capacity); - let observed = crate::engine::profile::spawn_observed(handle.reader.clone(), version_rx); - tokio::spawn(run(rx, settings, write)); + let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity); + let (profile, observed) = crate::engine::profile::spawn_observed(reader.clone(), version_rx); + let handle = EngineHandle { + tx, + reader, + profile_reader: profile.reader.clone(), + }; + tokio::spawn(run(rx, settings, write, profile)); (handle, observed) } @@ -166,6 +193,7 @@ async fn run( mut rx: mpsc::Receiver, settings: SettingsHandle, mut write: IndexWriter, + profile: crate::engine::profile::ProfileHandle, ) { let mut max_logs: Option = None; let mut metric_sample_sequence: u64 = 0; @@ -209,10 +237,10 @@ async fn run( let text_for_op = Arc::clone(&text); write.mutate(move |snap: &mut EngineSnapshot| { - // Two independently-callable stages (ticket 101): primary/param - // extraction, then profile-scoped metric computation. + // Primary/param extraction only (ticket 101) -- metric computation + // (ticket 104) now happens on the profile actor, which wakes on this + // publish and reads the newly-visible line itself. 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(); snap.evict_excess_logs(max_logs); @@ -327,49 +355,40 @@ async fn run( request, respond_to, } => { - let result = write - .peek() - .map(|guard| guard.validate_metric_request(&request).map(|_| ())) - .unwrap_or_else(|| Err("engine is shutting down".to_string())); - let succeeded = result.is_ok(); - let request_for_op = request.clone(); - write.mutate(move |snap: &mut EngineSnapshot| { - if succeeded { - let _ = snap.declare_metric(request_for_op.clone()); - } - }); - let _ = respond_to.send(result); + // Metric declarations live on the profile actor (ticket 104), not here -- + // proxy the message rather than mutating the primary's snapshot. + let _ = profile + .tx + .send(crate::engine::profile::ProfileMessage::DeclareMetric { + request, + respond_to, + }) + .await; } EngineMessage::UpdateMetric { name, request, respond_to, } => { - let result = write - .peek() - .map(|guard| guard.validate_update_metric(&name, &request).map(|_| ())) - .unwrap_or_else(|| Err("engine is shutting down".to_string())); - let succeeded = result.is_ok(); - let existing_name = name.clone(); - let request_for_op = request.clone(); - write.mutate(move |snap: &mut EngineSnapshot| { - if succeeded { - let _ = snap.update_metric(&existing_name, request_for_op.clone()); - } - }); - let _ = respond_to.send(result); + let _ = profile + .tx + .send(crate::engine::profile::ProfileMessage::UpdateMetric { + name, + request, + respond_to, + }) + .await; } EngineMessage::RemoveMetricDeclaration { name, respond_to } => { - let existed = write.peek().is_some_and(|guard| { - guard - .index - .contains_key(&IndexKey::MetricDeclaration { name: name.clone() }) - }); - let name_for_op = name.clone(); - write.mutate(move |snap: &mut EngineSnapshot| { - snap.remove_metric_declaration(&name_for_op); - }); - let _ = respond_to.send(existed); + let _ = profile + .tx + .send( + crate::engine::profile::ProfileMessage::RemoveMetricDeclaration { + name, + respond_to, + }, + ) + .await; } EngineMessage::EnsureSortHelperIndex { param_name, diff --git a/src/engine/profile.rs b/src/engine/profile.rs index 47bddd2..2b78857 100644 --- a/src/engine/profile.rs +++ b/src/engine/profile.rs @@ -1,18 +1,30 @@ -//! The profile actor (ticket 103, first slice of the split described in -//! `ideas/0001-actor-split.md`). It wakes whenever the primary actor's `IndexWriter` -//! (`super::handle`) publishes a `left_right` change, reads whatever primary log lines have -//! become newly visible since it last looked, and — for now — does nothing with them beyond -//! tracking how far it's read. Metric computation (ticket 104) and alert evaluation -//! (ticket 105) land in this wake loop next; nothing outside `engine::handle::spawn` talks to -//! this actor yet, so it has no public handle of its own. +//! The profile actor (ticket 104, building on ticket 103's skeleton). Owns metric declarations +//! and bucket computation, moved off the primary per `ideas/0001-actor-split.md`'s resolved +//! decision that the primary "extracts declared + auto params only... no metrics, no profile +//! logic." Reuses `EngineSnapshot` itself as this actor's own snapshot type — `declare_metric`/ +//! `update_metric`/`remove_metric_declaration`/`apply_all_metrics` all work unchanged — rather +//! than inventing a parallel type, since metric evaluation needs param declarations present in +//! the same snapshot it reads (`resolved_param_declarations`); this actor mirrors the primary's +//! param declarations into its own snapshot on every wake so that stays true, but stores no raw +//! log text of its own, reading it through `primary_reader` (a clone of the primary's +//! `IndexReader`) instead. `EngineHandle`'s public API is unchanged by this ticket — +//! `DeclareMetric`/`UpdateMetric`/`RemoveMetricDeclaration` are still sent to the primary's +//! `mpsc` channel and proxied here by `handle.rs`'s `run()` loop; ticket 106 gives API handlers +//! a direct route. +use crate::engine::DeclareMetricRequest; +use crate::engine::data_value::{DataValue, ParamDeclaration}; use crate::engine::handle::IndexReader; +use crate::engine::index_key::IndexKey; +use crate::engine::snapshot::EngineSnapshot; +use crate::engine::timestamp::Timestamp; use left_right::Absorb; -use tokio::sync::watch; +use regex::Regex; +use tokio::sync::{mpsc, oneshot, watch}; -struct ProfileOp(Box); +struct ProfileOp(Box); -impl Absorb for ProfileSnapshot { +impl Absorb for EngineSnapshot { fn absorb_first(&mut self, operation: &mut ProfileOp, _other: &Self) { (operation.0)(self); } @@ -24,70 +36,281 @@ impl Absorb for ProfileSnapshot { } } -/// Profile-actor-owned derived state. Empty besides a read watermark until ticket 104 moves -/// metric declarations/buckets here and ticket 105 moves alert runtime state here. -#[derive(Debug, Clone, Default)] -struct ProfileSnapshot { - /// Highest primary log sequence this actor has read so far, so the wake loop resumes from - /// there instead of rescanning the primary's whole retained range on every wake. - last_seen_sequence: Option, +/// Metric-declaration messages proxied to the profile actor by the primary's `run()` loop. +pub(crate) enum ProfileMessage { + DeclareMetric { + request: DeclareMetricRequest, + respond_to: oneshot::Sender>, + }, + UpdateMetric { + name: String, + request: DeclareMetricRequest, + respond_to: oneshot::Sender>, + }, + RemoveMetricDeclaration { + name: String, + respond_to: oneshot::Sender, + }, +} + +pub(crate) struct ProfileHandle { + pub(crate) tx: mpsc::Sender, + /// Read-only view of this actor's own snapshot (metric declarations, bucket data) -- + /// `MetricDeclarations`/`BucketValues`/`Promql` queries read through this (see + /// `EngineHandle::profile_reader` in `handle.rs`) instead of the primary's reader. + pub(crate) reader: IndexReader, +} + +/// A read-only view over a fresh, never-written `EngineSnapshot` -- used by `handle::test_handle`, +/// which spawns no actor loop at all (so nothing would ever publish to a real profile reader +/// anyway), just to give `EngineHandle::profile_reader` a valid value. +#[cfg(test)] +pub(crate) fn inert_reader() -> IndexReader { + let (_write, read) = left_right::new::(); + IndexReader { + factory: read.factory(), + } } /// Spawns the profile actor. `primary_reader` is a clone of the primary's `IndexReader`, giving -/// this actor its own read-only view of the primary's index; `primary_version` is bumped by the -/// primary's `IndexWriter::mutate` after every `left_right` publish, so this actor blocks on -/// `changed()` between primary writes rather than polling. -pub(crate) fn spawn(primary_reader: IndexReader, primary_version: watch::Receiver) { - spawn_inner(primary_reader, primary_version, None); +/// this actor its own read-only view of the primary's index (raw log text, param declarations); +/// `primary_version` is bumped by the primary's `IndexWriter::mutate` after every `left_right` +/// publish, so this actor blocks on `changed()` between primary writes rather than polling. +pub(crate) fn spawn( + primary_reader: IndexReader, + primary_version: watch::Receiver, +) -> ProfileHandle { + let (tx, rx) = mpsc::channel(crate::consts::ENGINE_CHANNEL_CAPACITY); + let (write, read) = left_right::new::(); + let reader = IndexReader { + factory: read.factory(), + }; + tokio::spawn(run(primary_reader, primary_version, rx, write, None)); + ProfileHandle { tx, reader } } -/// Test-only variant of `spawn` that reports every watermark advance on an unbounded channel, -/// so a test can assert the wake/read loop actually ran without reaching into actor-private -/// state (`ProfileSnapshot` has no public read API yet — this ticket is plumbing only). +/// Test-only variant of `spawn` that reports every ingest-driven watermark advance on an +/// unbounded channel, so a test can assert the wake/read loop actually ran without reaching +/// into actor-private state. #[cfg(test)] pub(crate) fn spawn_observed( primary_reader: IndexReader, primary_version: watch::Receiver, -) -> tokio::sync::mpsc::UnboundedReceiver { - let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); - spawn_inner(primary_reader, primary_version, Some(tx)); - rx +) -> (ProfileHandle, mpsc::UnboundedReceiver) { + let (tx, rx) = mpsc::channel(crate::consts::ENGINE_CHANNEL_CAPACITY); + let (write, read) = left_right::new::(); + let reader = IndexReader { + factory: read.factory(), + }; + let (advance_tx, advance_rx) = mpsc::unbounded_channel(); + tokio::spawn(run( + primary_reader, + primary_version, + rx, + write, + Some(advance_tx), + )); + (ProfileHandle { tx, reader }, advance_rx) } -fn spawn_inner( +async fn run( primary_reader: IndexReader, mut primary_version: watch::Receiver, - #[cfg_attr(not(test), allow(unused_variables))] on_advance: Option< - tokio::sync::mpsc::UnboundedSender, - >, + mut rx: mpsc::Receiver, + mut write: left_right::WriteHandle, + on_advance: Option>, ) { - tokio::spawn(async move { - let (mut write, _read) = left_right::new::(); + let mut last_seen_sequence: Option = None; - while primary_version.changed().await.is_ok() { - let primary_handle = primary_reader.handle(); - let Some(primary) = primary_handle.enter() else { - continue; - }; - let last_seen = write.enter().and_then(|guard| guard.last_seen_sequence); - let range_start = last_seen.map_or(0, |sequence| sequence.saturating_add(1)); + loop { + tokio::select! { + changed = primary_version.changed() => { + if changed.is_err() { + break; // primary actor gone, shut down + } + let primary_handle = primary_reader.handle(); + let Some(primary) = primary_handle.enter() else { + continue; + }; - let mut new_last_seen = last_seen; - for (sequence, _timestamp, _text) in primary.logs_in_sequence_range(range_start..) { - new_last_seen = - Some(new_last_seen.map_or(sequence, |current| current.max(sequence))); - } - drop(primary); + mirror_param_declarations(&primary, &mut write); - if new_last_seen != last_seen { - write.append(ProfileOp(Box::new(move |snap: &mut ProfileSnapshot| { - snap.last_seen_sequence = new_last_seen; - }))); + let range_start = last_seen_sequence.map_or(0, |sequence| sequence.saturating_add(1)); + let new_lines: Vec<(u64, Timestamp, String)> = primary + .logs_in_sequence_range(range_start..) + .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string())) + .collect(); + drop(primary); + + for (sequence, timestamp, text) in new_lines { + write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| { + snap.apply_all_metrics(sequence, timestamp, &text); + }))); + last_seen_sequence = Some(sequence); + } write.publish(); - if let (Some(sequence), Some(tx)) = (new_last_seen, on_advance.as_ref()) { + if let (Some(sequence), Some(tx)) = (last_seen_sequence, on_advance.as_ref()) { let _ = tx.send(sequence); } } + message = rx.recv() => { + let Some(message) = message else { + break; // primary actor dropped its ProfileHandle, shut down + }; + handle_message(message, &primary_reader, &mut write); + } } - }); + } +} + +fn handle_message( + message: ProfileMessage, + primary_reader: &IndexReader, + write: &mut left_right::WriteHandle, +) { + match message { + ProfileMessage::DeclareMetric { + request, + respond_to, + } => { + // Freshen the param mirror synchronously rather than relying on the watch-wake + // loop having already caught up -- a param declared just before this metric (e.g. + // `Profile::load_into_engine` declaring params then metrics in sequence) may not + // have been mirrored yet when this message is handled, since there's no ordering + // guarantee between the wake branch and this one in `tokio::select!`. + if let Some(primary) = primary_reader.handle().enter() { + mirror_param_declarations(&primary, write); + } + let result = write + .enter() + .map(|guard| guard.validate_metric_request(&request).map(|_| ())) + .unwrap_or_else(|| Err("engine is shutting down".to_string())); + let succeeded = result.is_ok(); + let lines = primary_owned_lines(primary_reader); + let request_for_op = request.clone(); + write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| { + if succeeded { + let _ = snap.declare_metric_with_lines(request_for_op.clone(), &lines); + } + }))); + write.publish(); + let _ = respond_to.send(result); + } + ProfileMessage::UpdateMetric { + name, + request, + respond_to, + } => { + if let Some(primary) = primary_reader.handle().enter() { + mirror_param_declarations(&primary, write); + } + let result = write + .enter() + .map(|guard| guard.validate_update_metric(&name, &request).map(|_| ())) + .unwrap_or_else(|| Err("engine is shutting down".to_string())); + let succeeded = result.is_ok(); + let lines = primary_owned_lines(primary_reader); + let existing_name = name.clone(); + let request_for_op = request.clone(); + write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| { + if succeeded { + let _ = snap.update_metric_with_lines( + &existing_name, + request_for_op.clone(), + &lines, + ); + } + }))); + write.publish(); + let _ = respond_to.send(result); + } + ProfileMessage::RemoveMetricDeclaration { name, respond_to } => { + let existed = write.enter().is_some_and(|guard| { + guard + .index + .contains_key(&IndexKey::MetricDeclaration { name: name.clone() }) + }); + let name_for_op = name.clone(); + write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| { + snap.remove_metric_declaration(&name_for_op); + }))); + write.publish(); + let _ = respond_to.send(existed); + } + } +} + +/// Reads all currently-retained primary lines through `primary_reader`, for backfilling a +/// newly declared/updated metric — the profile actor's own snapshot holds no raw log text, so +/// this is its only source of backfill history (see `EngineSnapshot::declare_metric_with_lines`). +fn primary_owned_lines(primary_reader: &IndexReader) -> Vec<(u64, Timestamp, String)> { + let primary_handle = primary_reader.handle(); + primary_handle + .enter() + .map(|guard| { + guard + .logs_range() + .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string())) + .collect() + }) + .unwrap_or_default() +} + +/// Replaces the profile snapshot's mirrored `ParamDeclaration` entries and `param_regex_cache` +/// with the primary's current ones, so metric label/value resolution +/// (`EngineSnapshot::resolved_param_declarations`) sees the same params the primary has — +/// cheap enough to do unconditionally on every wake since declared params are capped +/// (`MAX_AUTO_DECLARED_PARAMS`). +fn mirror_param_declarations( + primary: &EngineSnapshot, + write: &mut left_right::WriteHandle, +) { + let declarations: Vec = primary + .param_declarations_range() + .map( + |( + name, + message_match_regex, + capture_group, + json_field_path, + kind, + elapsed_since_last, + auto_derived, + log_stream_gated, + )| ParamDeclaration { + 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(str::to_string), + kind, + elapsed_since_last, + auto_derived, + log_stream_gated, + }, + ) + .collect(); + + write.append(ProfileOp(Box::new(move |snap: &mut EngineSnapshot| { + let stale: Vec = snap + .index + .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit) + .map(|(key, _)| key.clone()) + .collect(); + for key in stale { + snap.index.remove(&key); + } + snap.param_regex_cache.clear(); + for declaration in &declarations { + if let Ok(re) = Regex::new(&declaration.message_match_regex) { + snap.param_regex_cache.insert(declaration.name.clone(), re); + } + snap.index.insert( + IndexKey::ParamDeclaration { + name: declaration.name.clone(), + }, + DataValue::ParamDeclaration(declaration.clone()), + ); + } + }))); + write.publish(); } diff --git a/src/engine/snapshot/mod.rs b/src/engine/snapshot/mod.rs index a32a8bd..069a4c3 100644 --- a/src/engine/snapshot/mod.rs +++ b/src/engine/snapshot/mod.rs @@ -394,18 +394,52 @@ impl EngineSnapshot { Ok(re) } + /// Self-contained declare, backfilling from `self.logs_range()` — only used by tests that + /// build a standalone `EngineSnapshot` without the actor split; production code always goes + /// through the profile actor, which calls `declare_metric_with_lines` directly (its own + /// snapshot holds no raw log text of its own to backfill from). + #[cfg_attr(not(test), allow(dead_code))] pub(crate) fn declare_metric(&mut self, request: DeclareMetricRequest) -> Result<(), String> { + let lines = self.owned_logs_range(); + self.declare_metric_with_lines(request, &lines) + } + + /// Like `declare_metric`, but backfills from `lines` instead of `self.logs_range()` — the + /// profile actor (`engine::profile`) calls this, since its own snapshot holds no raw log + /// text (only mirrored param declarations plus its own metric declarations/buckets); it + /// sources backfill lines by reading through its cloned primary `IndexReader` instead. + pub(crate) fn declare_metric_with_lines( + &mut self, + request: DeclareMetricRequest, + lines: &[(u64, Timestamp, String)], + ) -> Result<(), String> { let re = self.validate_metric_request(&request)?; self.metric_regex_cache .insert(request.name.clone(), re.clone()); - self.resolve_backfill_and_store_metric(request, &re); + self.resolve_backfill_and_store_metric(request, &re, lines); Ok(()) } - /// Shared by `declare_metric` and `update_metric`: resolves param declarations, backfills - /// the metric's buckets from existing log history, then writes the `MetricDeclaration` entry. - /// Callers differ only in what happens before this (rename bookkeeping for updates). - fn resolve_backfill_and_store_metric(&mut self, request: DeclareMetricRequest, re: &Regex) { + /// `logs_range` collected into owned `(sequence, timestamp, text)` triples, for callers + /// (like `declare_metric`) that need to hand backfill lines to a method taking them by + /// value rather than borrowing `self` for the duration. + #[cfg_attr(not(test), allow(dead_code))] + fn owned_logs_range(&self) -> Vec<(u64, Timestamp, String)> { + self.logs_range() + .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string())) + .collect() + } + + /// Shared by `declare_metric`/`update_metric` (and their `_with_lines` variants): resolves + /// param declarations, backfills the metric's buckets from `lines`, then writes the + /// `MetricDeclaration` entry. Callers differ only in what happens before this (rename + /// bookkeeping for updates) and where `lines` comes from. + fn resolve_backfill_and_store_metric( + &mut self, + request: DeclareMetricRequest, + re: &Regex, + lines: &[(u64, Timestamp, String)], + ) { let resolved_params = self.resolved_param_declarations(); self.backfill_metric( &request.name, @@ -415,6 +449,7 @@ impl EngineSnapshot { &resolved_params, request.matched_labels.as_ref(), request.unmatched_labels.as_ref(), + lines, ); self.index.insert( IndexKey::MetricDeclaration { @@ -451,10 +486,25 @@ impl EngineSnapshot { self.validate_metric_request(request) } + /// Self-contained update, backfilling from `self.logs_range()` — see `declare_metric`'s doc + /// comment; only used by tests that build a standalone `EngineSnapshot`. + #[cfg_attr(not(test), allow(dead_code))] pub(crate) fn update_metric( &mut self, existing_name: &str, request: DeclareMetricRequest, + ) -> Result<(), String> { + let lines = self.owned_logs_range(); + self.update_metric_with_lines(existing_name, request, &lines) + } + + /// Like `update_metric`, but backfills from `lines` instead of `self.logs_range()` — see + /// `declare_metric_with_lines`. + pub(crate) fn update_metric_with_lines( + &mut self, + existing_name: &str, + request: DeclareMetricRequest, + lines: &[(u64, Timestamp, String)], ) -> Result<(), String> { let re = self.validate_update_metric(existing_name, &request)?; @@ -468,7 +518,7 @@ impl EngineSnapshot { self.metric_regex_cache .insert(request.name.clone(), re.clone()); - self.resolve_backfill_and_store_metric(request, &re); + self.resolve_backfill_and_store_metric(request, &re, lines); Ok(()) } @@ -593,11 +643,8 @@ impl EngineSnapshot { resolved_params: &HashMap, matched_labels: Option<&BTreeMap>, unmatched_labels: Option<&BTreeMap>, + lines: &[(u64, Timestamp, String)], ) { - let lines: Vec<(u64, Timestamp, String)> = self - .logs_range() - .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string())) - .collect(); for (sequence, timestamp, text) in lines { self.apply_metric_to_log( metric_name, @@ -607,9 +654,9 @@ impl EngineSnapshot { resolved_params, matched_labels, unmatched_labels, - sequence, - timestamp, - &text, + *sequence, + *timestamp, + text, ); } } diff --git a/tickets/104-move-metric-computation-to-profile-actor.md b/tickets/104-move-metric-computation-to-profile-actor.md deleted file mode 100644 index 74a739f..0000000 --- a/tickets/104-move-metric-computation-to-profile-actor.md +++ /dev/null @@ -1,73 +0,0 @@ -# 104 — Move metric declarations and bucket computation into the profile actor - -## Summary - -Second slice of the split: metric declarations, `apply_all_metrics`, -and the declare/update/remove/backfill machinery move off the primary -`EngineSnapshot` and onto the profile actor's `ProfileSnapshot` -(stood up in [[103-profile-actor-skeleton-and-primary-signaling]]). -The primary keeps raw log storage and param declarations only, per -[[0001-actor-split]]'s resolved decision that the primary "extracts -declared + auto params only... no metrics, no profile logic." The -profile actor consumes newly-visible primary lines (via the -watch-wake loop from ticket 103) and runs metric regexes against them -itself, building its own bucket data independent of the primary. - -## Where - -- `src/engine/snapshot/mod.rs:364-434` (`validate_metric_request`, - `declare_metric`, `resolve_backfill_and_store_metric`) and - `:434-487` (`validate_update_metric`, `update_metric`, - `remove_metric_declaration`) — the declare/update/remove machinery - to move onto `ProfileSnapshot`. -- `src/engine/snapshot/mod.rs:587-617` (`backfill_metric`, - `apply_all_metrics`) — backfill replays `self.logs_range()` (a - primary-side accessor) through `apply_metric_to_log`; once metrics - live on the profile actor, backfill needs to read logs through the - profile actor's cloned primary `IndexReader` instead of `self`. -- `src/engine/handle.rs` — `EngineMessage::DeclareMetric`/ - `UpdateMetric`/`RemoveMetricDeclaration` arms (see the message list - at `src/engine/message.rs:30-63`) currently route through the - single actor's `write.mutate` against `EngineSnapshot`; these need - to route to the profile actor's `ProfileSnapshot` instead. -- `src/engine/handle.rs`'s `Ingest` arm — `snap.apply_all_metrics(...)` - is called directly after `ingest_params` today - (see ticket 101's change); this call moves out of the primary's - ingest closure entirely once the profile actor computes metrics - itself from the lines it observes via the watch/read loop, rather - than the primary computing them inline. - -## Scope - -- Move `MetricDeclaration`/bucket-related `IndexKey`/`DataValue` - variants' *usage* (not necessarily their type definitions, which are - shared) onto `ProfileSnapshot`'s own index/map, separate from the - primary's. -- Move `declare_metric`/`update_metric`/`remove_metric_declaration`/ - `backfill_metric`/`apply_all_metrics` (and their validation - counterparts) to operate on `ProfileSnapshot`, reading raw log text - via the profile actor's cloned primary `IndexReader` wherever they - need `logs_range`-style access, instead of `self.index` on the - primary. -- Remove the `apply_all_metrics` call from the primary's `Ingest` arm - in `handle.rs` — the primary no longer computes metrics inline; the - profile actor's watch-wake loop (from ticket 103) now calls - `apply_all_metrics`-equivalent logic itself per newly-visible line. -- `DeclareMetric`/`UpdateMetric`/`RemoveMetricDeclaration` messages - need a new route to the profile actor (a second `mpsc` channel, or - folding into the profile actor's own message enum) — decide the - concrete mechanism during implementation; `EngineHandle`'s public - API doesn't need to change yet if this ticket keeps routing internal - to `handle.rs` (full API surface changes are - [[106-split-enginehandle-api-for-primary-and-profile]]'s scope). -- Out of scope: alert evaluation (still reads the primary's snapshot - directly today — [[105-move-alert-evaluation-to-profile-actor]] - moves it), and the profile-switch (`activate_profile`) reload path - (still targets whatever this ticket lands as the metric-owning - actor, but retargeting `src/api/mod.rs:812-852` explicitly is - [[106-split-enginehandle-api-for-primary-and-profile]]'s scope). -- Backfill cost and blocking behavior: per [[0001-actor-split]]'s - performance rationale, this move is what stops a slow backfill from - blocking the primary's ingest/query-serving — confirm with a test - that ingesting new lines on the primary keeps working while the - profile actor is mid-backfill on a newly declared metric.