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.