From f43445b2800c3dbc3fc6bb836bc9d5357b776d77 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Mon, 7 Sep 2026 13:26:03 +0100 Subject: [PATCH] Arc-ify stored log text to remove per-line double clones (ticket 100) DataValue::LogLiteral and IngestedLine now carry an Arc instead of an owned String, so the two clones the ingest path paid per line (once to move text into the write closure, once to store it) become cheap refcount bumps instead of full string copies. Also cuts the per-string-content cost of left_right's full-EngineSnapshot deep clone on each reconciliation sync. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ --- src/api/mod.rs | 8 +- src/config/profile.rs | 8 +- src/engine/data_value.rs | 3 +- src/engine/handle.rs | 5 +- src/engine/ingest.rs | 6 +- src/engine/mod.rs | 3 +- src/engine/snapshot/mod.rs | 2 +- src/engine/snapshot/query.rs | 2 +- src/engine/tests.rs | 26 ++--- ...-metric-computation-into-separate-stage.md | 62 ++++++++++ ...gine-into-reader-primary-profile-actors.md | 107 ++++++++++++++++++ 11 files changed, 198 insertions(+), 34 deletions(-) create mode 100644 tickets/101-extract-metric-computation-into-separate-stage.md create mode 100644 tickets/102-split-engine-into-reader-primary-profile-actors.md diff --git a/src/api/mod.rs b/src/api/mod.rs index a863965..ad2c1a3 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1440,7 +1440,7 @@ mod tests { .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), - text: "request path=/a".to_string(), + text: "request path=/a".into(), })) .await .unwrap(); @@ -1494,7 +1494,7 @@ mod tests { .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), - text: text.to_string(), + text: text.into(), })) .await .unwrap(); @@ -1554,7 +1554,7 @@ mod tests { .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), - text: text.to_string(), + text: text.into(), })) .await .unwrap(); @@ -1822,7 +1822,7 @@ mod tests { .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), - text: text.to_string(), + text: (*text).into(), })) .await .unwrap(); diff --git a/src/config/profile.rs b/src/config/profile.rs index 48bd9e7..f9ef68f 100644 --- a/src/config/profile.rs +++ b/src/config/profile.rs @@ -894,7 +894,7 @@ mod tests { .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64), - text: text.to_string(), + text: text.into(), })) .await .unwrap(); @@ -1144,7 +1144,7 @@ mod tests { .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), - text: "level=info request duration=42".to_string(), + text: "level=info request duration=42".into(), })) .await .unwrap(); @@ -1327,7 +1327,7 @@ mod tests { .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), - text: "request path=/a duration=42".to_string(), + text: "request path=/a duration=42".into(), })) .await .unwrap(); @@ -1511,7 +1511,7 @@ mod tests { .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), - text: "dx=1 moved".to_string(), + text: "dx=1 moved".into(), })) .await .unwrap(); diff --git a/src/engine/data_value.rs b/src/engine/data_value.rs index af190d8..063db50 100644 --- a/src/engine/data_value.rs +++ b/src/engine/data_value.rs @@ -2,6 +2,7 @@ use crate::engine::metric_primitive::MetricPrimitive; use crate::engine::timestamp::Timestamp; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; +use std::sync::Arc; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct ParamDeclaration { @@ -98,7 +99,7 @@ pub enum MetricSemantics { #[derive(Debug, Clone, Serialize)] pub enum DataValue { LogLiteral { - text: String, + text: Arc, }, ParamDeclaration(ParamDeclaration), MetricDeclaration { diff --git a/src/engine/handle.rs b/src/engine/handle.rs index 3070835..4e512c4 100644 --- a/src/engine/handle.rs +++ b/src/engine/handle.rs @@ -15,6 +15,7 @@ use crate::engine::{ClearParamsResult, IngestedLine}; use crate::flavour::LogLiteralHelper; use left_right::Absorb; use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; use tokio::sync::{mpsc, oneshot}; struct EngineOp(Box); @@ -176,7 +177,7 @@ async fn run( ) }; - let text_for_op = text.clone(); + 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( @@ -185,7 +186,7 @@ async fn run( timestamp, }, DataValue::LogLiteral { - text: text_for_op.clone(), + text: Arc::clone(&text_for_op), }, ); snap.log_count += 1; diff --git a/src/engine/ingest.rs b/src/engine/ingest.rs index 4da4459..069c4c1 100644 --- a/src/engine/ingest.rs +++ b/src/engine/ingest.rs @@ -16,7 +16,7 @@ pub async fn run(engine: EngineHandle) { let line = IngestedLine { sequence, timestamp: Timestamp::now(), - text, + text: text.into(), }; sequence += 1; @@ -86,7 +86,7 @@ pub async fn run_file(engine: EngineHandle, path: PathBuf) { let line = IngestedLine { sequence, timestamp: Timestamp::now(), - text, + text: text.into(), }; sequence += 1; @@ -247,7 +247,7 @@ mod tests { .expect("timed out waiting for ingested line") .expect("engine channel closed") { - EngineMessage::Ingest(line) => line.text, + EngineMessage::Ingest(line) => line.text.to_string(), _ => panic!("expected Ingest"), } } diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 54fa7ab..5aa305d 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -26,11 +26,12 @@ use crate::engine::data_value::{MetricSemantics, ParamDeclaration}; use crate::engine::timestamp::Timestamp; use serde::Deserialize; use std::collections::BTreeMap; +use std::sync::Arc; pub struct IngestedLine { pub sequence: u64, pub timestamp: Timestamp, - pub text: String, + pub text: Arc, } pub struct MetricSample { diff --git a/src/engine/snapshot/mod.rs b/src/engine/snapshot/mod.rs index 0a072d3..3dda21b 100644 --- a/src/engine/snapshot/mod.rs +++ b/src/engine/snapshot/mod.rs @@ -249,7 +249,7 @@ impl EngineSnapshot { timestamp, }, DataValue::LogLiteral { text }, - ) => Some((*sequence, *timestamp, text.as_str())), + ) => Some((*sequence, *timestamp, text.as_ref())), _ => None, }) } diff --git a/src/engine/snapshot/query.rs b/src/engine/snapshot/query.rs index 9321738..e63cc84 100644 --- a/src/engine/snapshot/query.rs +++ b/src/engine/snapshot/query.rs @@ -106,7 +106,7 @@ fn logs_sorted_by_param( entries.push(LogEntryDto { sequence, timestamp_ms: timestamp.as_millis(), - text: text.clone(), + text: text.to_string(), params: log_entry_params(text, compiled_params), }); if entries.len() >= limit { diff --git a/src/engine/tests.rs b/src/engine/tests.rs index bd5a281..06db12e 100644 --- a/src/engine/tests.rs +++ b/src/engine/tests.rs @@ -287,9 +287,7 @@ fn logs_query_sorted_by_a_log_stream_column_actually_reorders_real_log_stream_ou sequence, timestamp, }, - DataValue::LogLiteral { - text: line.to_string(), - }, + DataValue::LogLiteral { text: line.into() }, ); } @@ -477,7 +475,7 @@ fn evict_excess_logs_drops_oldest_sequences_first() { sequence, timestamp: Timestamp::from_millis(sequence as i64), }, - DataValue::LogLiteral { text }, + DataValue::LogLiteral { text: text.into() }, ); index.log_count += 1; } @@ -504,7 +502,7 @@ fn logs_in_sequence_range_seeks_instead_of_scanning_from_start() { timestamp: Timestamp::from_millis(sequence as i64), }, DataValue::LogLiteral { - text: format!("line {sequence}"), + text: format!("line {sequence}").into(), }, ); } @@ -548,9 +546,7 @@ fn evict_excess_logs_is_noop_when_unlimited_or_under_cap() { sequence, timestamp: Timestamp::from_millis(sequence as i64), }, - DataValue::LogLiteral { - text: "x".to_string(), - }, + DataValue::LogLiteral { text: "x".into() }, ); index.log_count += 1; } @@ -621,9 +617,7 @@ fn insert_ingested_log(index: &mut EngineSnapshot, sequence: u64, text: &str) { sequence, timestamp: Timestamp::from_millis(sequence as i64), }, - DataValue::LogLiteral { - text: text.to_string(), - }, + DataValue::LogLiteral { text: text.into() }, ); } @@ -701,7 +695,7 @@ fn declarations_stay_within_their_range() { timestamp: Timestamp::from_millis(0), }, DataValue::LogLiteral { - text: "hello".to_string(), + text: "hello".into(), }, ); declare_param( @@ -906,7 +900,7 @@ fn metric_declarations_stay_within_their_range() { timestamp: Timestamp::from_millis(0), }, DataValue::LogLiteral { - text: "hello".to_string(), + text: "hello".into(), }, ); declare_param( @@ -1161,9 +1155,7 @@ fn ingest_log(index: &mut EngineSnapshot, sequence: u64, text: &str) { sequence, timestamp, }, - DataValue::LogLiteral { - text: text.to_string(), - }, + DataValue::LogLiteral { text: text.into() }, ); index.apply_all_metrics(sequence, timestamp, text); } @@ -2677,7 +2669,7 @@ fn logs_query_with_ottyl_ts_comparison_filters_by_timestamp() { timestamp, }, DataValue::LogLiteral { - text: format!("line {i}"), + text: format!("line {i}").into(), }, ); } diff --git a/tickets/101-extract-metric-computation-into-separate-stage.md b/tickets/101-extract-metric-computation-into-separate-stage.md new file mode 100644 index 0000000..81a92eb --- /dev/null +++ b/tickets/101-extract-metric-computation-into-separate-stage.md @@ -0,0 +1,62 @@ +# 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. diff --git a/tickets/102-split-engine-into-reader-primary-profile-actors.md b/tickets/102-split-engine-into-reader-primary-profile-actors.md new file mode 100644 index 0000000..63ebac5 --- /dev/null +++ b/tickets/102-split-engine-into-reader-primary-profile-actors.md @@ -0,0 +1,107 @@ +# 102 — Split the single engine actor into reader, primary, and profile actors + +## Summary + +Today, one actor (`run()` in `src/engine/handle.rs`) does everything: +tailing/ingest bookkeeping, raw log storage, param extraction, *every* +declared metric's regex evaluation, sort-helper indexing, eviction, and +alert evaluation, all serialized through one `mpsc` channel. This is +migration step 3 (the actual split) of [[0001-actor-split]]'s resolved +plan, and depends on [[100-arc-ify-stored-log-and-metric-strings]] and +[[101-extract-metric-computation-into-separate-stage]] landing first — +this ticket is the wiring/topology change those two steps prepare for, +not a from-scratch redesign. See [[0001-actor-split]]'s "Resolved +decisions" section for the full rationale and answers to the open +questions below; this ticket's scope is that design, made concrete. + +## Where + +- `src/engine/handle.rs:130-365` (`run()`) — the single actor loop to + be split. The `EngineMessage::Ingest` arm (:143-219) is the part that + becomes the reader→primary boundary; `apply_all_metrics` / + `apply_metric_to_log` (`src/engine/snapshot/mod.rs:616`, :499-586) + and alert evaluation (`src/engine/alerts.rs:211`, called from + `handle.rs:199-219`) are what moves to the new profile actor. +- `src/cli/mod.rs:95` (`spawn_engine`) and `:105-124` + (`load_active_profile`) — the sole production call site of + `engine::spawn`; `load_active_profile`'s existing comment ("there's + no way to reprocess earlier lines later, so this must happen before + `spawn_ingest`", `:104-106`) documents a constraint the new topology + must preserve for the primary actor's startup ordering. +- `src/engine/ingest.rs` (stdin/file-tailing + Prometheus-polling + tasks, spawned at `src/cli/mod.rs:136-145`) — today's only + multi-task precedent, producer-side only (multiple producers into one + actor's channel); the new reader actor sits upstream of this or + subsumes part of it, per [[0001-actor-split]]'s three-actor design. +- `src/api/mod.rs:712-768` (`activate_profile` / + `reload_engine_declarations`) — the runtime profile-switch path that + currently removes-then-redeclares against the single actor; needs to + target the new profile actor instead. See [[0004-profile-switch-reindexing-cost]] + for the existing cost analysis of this path, which doesn't disappear + with this split but moves and stops blocking the primary. +- `src/engine/tests.rs` (2348 lines) and the `test_handle` pattern + (see ticket 099's removal of the duplicate `test_handle` stub) — + built around one actor with one channel; will need real rework, not + just extension, to cover primary alone, profile actor alone, and + primary+profile interaction. + +## Scope + +Implement the three-actor topology exactly as resolved in +[[0001-actor-split]]: + +- **Reader actor**: tails stdin/file, splits into lines, timestamps + them, wraps each line in an `Arc` (per [[100-arc-ify-stored-log-and-metric-strings]]), + sends downstream. +- **Primary actor**: ingests timestamped `Arc` lines, extracts declared + + auto params only (no metrics, no profile logic), maintains the + `param=value` index. Owns retention/eviction (`SetMaxLogs`) on its + own schedule — a lagging profile actor does not block primary + eviction; it silently misses data for the evicted range (accepted + tradeoff, not a bug to fix here). +- **Profile actor**: holds its own `left_right` reader over the + primary's index; builds metrics, custom params, timeseries, and + PromQL evaluation for the one active profile. Owns alert evaluation + (moved from the primary per the resolved decision that alerts are + profile-scoped). Still exactly one active profile at a time — no + concurrent multi-profile indexing; this changes *how* the active + profile's data is computed, not the single-active-profile user model. +- **Primary→profile signaling**: a `tokio::sync::watch` (or equivalent) + that the primary notifies right after each `left_right` sync + completes; the profile actor wakes on that signal and re-reads + through its own `left_right` reader — no busy-polling, no second data + copy in transit. +- **Query fan-out**: a layer for requests that need both raw-log and + profile-derived data in one response (e.g. `Logs`/`Timeline` + queries, which today read both from a single `EngineSnapshot` in one + pass — see `query.rs`'s `log_entry_params` call against + `compiled_params`) — this becomes a real join across two + snapshots/actors. +- **`EngineHandle` API**: split into two handle types (primary vs. + profile), or one combined handle that internally fans out — decide + during implementation which `api.rs`'s `ApiState` should hold; either + is acceptable per [[0001-actor-split]], this ticket doesn't mandate + one. +- Rework `src/engine/tests.rs` and the `test_handle` pattern to cover + the new topology (primary alone, profile actor alone, + primary+profile interaction) — explicitly in scope, not deferred, + since the existing single-actor test harness cannot exercise the new + split at all. +- Update `src/api/mod.rs:712-768`'s `activate_profile` / + `reload_engine_declarations` to address the profile actor. + +Out of scope (per [[0001-actor-split]]'s resolved decisions — +explicitly deferred, not silently dropped): + +- System param suppressions — dropped from scope entirely; revisit as + a separate idea if it becomes a real requirement. +- Full-text/regex indexing on the primary — a materially different + investment ([[0005-additional-read-indexes]]), doesn't depend on + actor topology. +- Profile-actor crash/restart checkpointing beyond "restart and + rebuild from whatever the primary currently retains" — no persisted + checkpoint required for this ticket. +- Diff-based or async profile-switch reload + ([[0004-profile-switch-reindexing-cost]]'s options) — this ticket + only needs `activate_profile` to target the right actor, not to + change the switch's synchronous-rebuild cost model.