From 0834c246d6ea6f564ff8ef22bbe4da493656aa20 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Sat, 5 Sep 2026 18:37:14 +0100 Subject: [PATCH 1/2] Add approximate heap/memory tracker for ingested logs (ticket 011) Tracks a running total of ingested log bytes (approximated as each log's raw UTF-8 text length), incremented on ingest and decremented in evict_excess_logs when a log is dropped. Exposed via StatsDto's new approx_bytes_total field on GET /api/stats, and displayed in NavMenu's stats panel formatted with toLocaleString() (comma separators). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o --- src/dto.rs | 13 ++++++ src/engine.rs | 74 ++++++++++++++++++++++++------- webapp/src/components/NavMenu.tsx | 1 + webapp/src/lib/api.ts | 2 + 4 files changed, 73 insertions(+), 17 deletions(-) diff --git a/src/dto.rs b/src/dto.rs index 7a89a2b..05171af 100644 --- a/src/dto.rs +++ b/src/dto.rs @@ -47,6 +47,19 @@ pub struct StatsDto { pub last_sequence: Option, pub first_timestamp_ms: Option, pub last_timestamp_ms: Option, + /// Ticket 011: a running, *approximate* estimate of the in-memory heap + /// held by currently-ingested logs, in bytes. "Approximate" because it + /// only counts each `IngestedLine::text`'s UTF-8 byte length — it does + /// not account for `String`'s allocator overhead/capacity slack, the + /// `BTreeMap`/`IndexKey` node overhead the text is stored under, or any + /// derived data (bucket values, params) the line produces elsewhere in + /// the index. That's judged the best cheap proxy for "how much memory is + /// this log's text costing us": it's O(1) to compute per log (`text.len()`, + /// already an interned byte count for UTF-8 `String`), moves in lockstep + /// with the dominant cost (raw log text, which is usually much larger + /// than the fixed per-entry bookkeeping), and needs no per-entry + /// bookkeeping beyond what eviction already tracks. + pub approx_bytes_total: u64, } #[derive(Debug, Deserialize)] diff --git a/src/engine.rs b/src/engine.rs index 741cf0c..d9b11ee 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -128,18 +128,23 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle let mut metric_sample_sequence: u64 = 0; // Per-rule rolling windows/seen-value sets, keyed by rule id; lives only as long as this loop (no restart persistence). let mut alert_states: HashMap = HashMap::new(); + // Ticket 011: running total of `approx_log_bytes` (see that function) across + // every currently-ingested log, kept in lockstep with `log_count` — incremented + // here on ingest, decremented in `evict_excess_logs` when a log is dropped. + let mut approx_bytes_total: u64 = 0; while let Some(message) = rx.recv().await { match message { EngineMessage::Ingest(line) => { let IngestedLine { sequence, timestamp, text } = line; + approx_bytes_total += approx_log_bytes(&text); index.insert( IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text: text.clone() }, ); log_count += 1; apply_all_metrics(&mut index, sequence, timestamp, &text); - evict_excess_logs(&mut index, &mut log_count, max_logs); + evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); // Evaluate alert rules against just this line, then hand any fired rule off to delivery via tokio::spawn so it stays off the hot path. let (rules, ntfy_topic, ntfy_server) = { @@ -169,7 +174,7 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle } EngineMessage::SetMaxLogs(new_max) => { max_logs = new_max; - evict_excess_logs(&mut index, &mut log_count, max_logs); + evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); } EngineMessage::DeclareParam { request, respond_to } => { let result = declare_param(&mut index, request); @@ -188,7 +193,7 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle let _ = respond_to.send(existed); } EngineMessage::Query { query, respond_to } => { - let response = handle_query(&index, query); + let response = handle_query(&index, query, approx_bytes_total); let _ = respond_to.send(response); } } @@ -723,11 +728,24 @@ fn derive_log_params( params } -/// Bucket values already derived from an evicted log are left in place — buckets are aggregate data, not raw logs, so they're not retroactively pruned. +/// Ticket 011's approximate per-log byte size: see `StatsDto::approx_bytes_total`'s +/// doc comment for why the raw UTF-8 text length is the chosen proxy rather than a +/// full struct-size estimate. +fn approx_log_bytes(text: &str) -> u64 { + text.len() as u64 +} + +/// Evicts the oldest (lowest-`sequence`) ingested logs until `log_count` is +/// back at or under `max_logs`. `sequence` numbering is not renumbered on +/// eviction, and any `Bucket` values already derived from an evicted log are +/// left in place — buckets are aggregate data, not raw logs, so they're not +/// retroactively pruned. `approx_bytes_total` (ticket 011) is decremented by +/// each evicted log's approximate size, mirroring the increment on ingest. fn evict_excess_logs( index: &mut BTreeMap, log_count: &mut usize, max_logs: Option, + approx_bytes_total: &mut u64, ) { let Some(max_logs) = max_logs else { return }; while *log_count > max_logs { @@ -738,6 +756,9 @@ fn evict_excess_logs( else { break; }; + if let Some(DataValue::LogLiteral { text }) = index.get(&oldest_key) { + *approx_bytes_total = approx_bytes_total.saturating_sub(approx_log_bytes(text)); + } index.remove(&oldest_key); *log_count -= 1; } @@ -754,7 +775,11 @@ fn logs_range(index: &BTreeMap) -> impl Iterator, query: EngineQuery) -> EngineResponse { +fn handle_query( + index: &BTreeMap, + query: EngineQuery, + approx_bytes_total: u64, +) -> EngineResponse { match query { EngineQuery::Logs(params) => { let compiled_filter = match params.filter.as_deref() { @@ -836,6 +861,7 @@ fn handle_query(index: &BTreeMap, query: EngineQuery) -> En last_sequence, first_timestamp_ms, last_timestamp_ms, + approx_bytes_total, }) } EngineQuery::ParamDeclarations => { @@ -1080,26 +1106,34 @@ mod tests { fn evict_excess_logs_drops_oldest_sequences_first() { let mut index = BTreeMap::new(); let mut log_count = 0; + let mut approx_bytes_total = 0u64; for sequence in 0..5u64 { + let text = format!("line {sequence}"); + approx_bytes_total += approx_log_bytes(&text); index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, - DataValue::LogLiteral { text: format!("line {sequence}") }, + DataValue::LogLiteral { text }, ); log_count += 1; } - evict_excess_logs(&mut index, &mut log_count, Some(3)); + evict_excess_logs(&mut index, &mut log_count, Some(3), &mut approx_bytes_total); assert_eq!(log_count, 3); let remaining: Vec<_> = logs_range(&index).map(|(seq, _, _)| seq).collect(); assert_eq!(remaining, vec![2, 3, 4]); + let expected_bytes: u64 = + remaining.iter().map(|seq| approx_log_bytes(&format!("line {seq}"))).sum(); + assert_eq!(approx_bytes_total, expected_bytes); } #[test] fn evict_excess_logs_is_noop_when_unlimited_or_under_cap() { let mut index = BTreeMap::new(); let mut log_count = 0; + let mut approx_bytes_total = 0u64; for sequence in 0..3u64 { + approx_bytes_total += approx_log_bytes("x"); index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, DataValue::LogLiteral { text: "x".to_string() }, @@ -1107,11 +1141,13 @@ mod tests { log_count += 1; } - evict_excess_logs(&mut index, &mut log_count, None); + evict_excess_logs(&mut index, &mut log_count, None, &mut approx_bytes_total); assert_eq!(log_count, 3); + assert_eq!(approx_bytes_total, 3); - evict_excess_logs(&mut index, &mut log_count, Some(10)); + evict_excess_logs(&mut index, &mut log_count, Some(10), &mut approx_bytes_total); assert_eq!(log_count, 3); + assert_eq!(approx_bytes_total, 3); } #[test] @@ -1773,6 +1809,7 @@ mod tests { start_ms: None, end_ms: None, }, + 0, ); match response { EngineResponse::Promql(Err(message)) => assert!(!message.is_empty()), @@ -1793,6 +1830,7 @@ mod tests { start_ms: None, end_ms: None, }, + 0, ); match response { EngineResponse::Promql(Ok(series)) => { @@ -1832,7 +1870,7 @@ mod tests { ingest_log(&mut index, 2, "still fine"); ingest_log(&mut index, 3, "another error here"); - match handle_query(&index, logs_query(Some(r#"regex("error")"#))) { + match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 3]); @@ -1848,7 +1886,7 @@ mod tests { ingest_log(&mut index, i, &format!("line {i}")); } - match handle_query(&index, logs_query(Some("sequence<2"))) { + match handle_query(&index, logs_query(Some("sequence<2")), 0) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 1]); @@ -1868,7 +1906,7 @@ mod tests { ); } - match handle_query(&index, logs_query(Some("timestamp>2000"))) { + match handle_query(&index, logs_query(Some("timestamp>2000")), 0) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![3, 4]); @@ -1898,6 +1936,7 @@ mod tests { match handle_query( &index, logs_query(Some(r#"regex("boom") and requests[level]==error"#)), + 0, ) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); @@ -1910,6 +1949,7 @@ mod tests { match handle_query( &index, logs_query(Some(r#"regex("boom") or requests[level]==error"#)), + 0, ) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); @@ -1926,7 +1966,7 @@ mod tests { ingest_log(&mut index, 0, "level=info hello"); ingest_log(&mut index, 1, "no match here"); - let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None)) else { + let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0) else { panic!("expected Logs response"); }; @@ -1945,7 +1985,7 @@ mod tests { ingest_log(&mut index, 1, "an error occurred"); ingest_log(&mut index, 2, "another error here"); - match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000)) { + match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0) { EngineResponse::Timeline(buckets) => { let total: u64 = buckets.iter().map(|b| b.count).sum(); assert_eq!(total, 2); @@ -1959,12 +1999,12 @@ mod tests { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "hello"); - match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@"))) { + match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0) { EngineResponse::Error(message) => assert!(!message.is_empty()), _ => panic!("expected an Error response for a malformed filter string"), } - match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000)) { + match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0) { EngineResponse::Error(message) => assert!(!message.is_empty()), _ => panic!("expected an Error response for a malformed filter string"), } @@ -1984,7 +2024,7 @@ mod tests { .unwrap(); ingest_log(&mut index, 0, "level=error"); - match handle_query(&index, logs_query(Some("requests==5"))) { + match handle_query(&index, logs_query(Some("requests==5")), 0) { EngineResponse::Error(message) => assert!(message.contains("param")), _ => panic!("expected an Error response for a metric comparison without [param]"), } diff --git a/webapp/src/components/NavMenu.tsx b/webapp/src/components/NavMenu.tsx index c7bd83f..7618cad 100644 --- a/webapp/src/components/NavMenu.tsx +++ b/webapp/src/components/NavMenu.tsx @@ -87,6 +87,7 @@ export function NavMenu({ stats }: { stats: Stats | null }) {
seq {stats.first_sequence ?? 0}–{stats.last_sequence ?? 0}
+
{stats.approx_bytes_total.toLocaleString()} bytes
) : (
connecting…
diff --git a/webapp/src/lib/api.ts b/webapp/src/lib/api.ts index 63ead58..91c32ee 100644 --- a/webapp/src/lib/api.ts +++ b/webapp/src/lib/api.ts @@ -39,6 +39,8 @@ export interface Stats { last_sequence: number | null first_timestamp_ms: number | null last_timestamp_ms: number | null + /** Ticket 011: running approximate byte size of all currently-ingested logs. */ + approx_bytes_total: number } export type ParamRole = 'label' | 'value' From 2c509b51fc9462c43d0e12a6afaa039d52c7335a Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Sat, 5 Sep 2026 18:38:29 +0100 Subject: [PATCH 2/2] Remove tickets/011-approximate-heap-tracker.md (implemented) Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o --- tickets/011-approximate-heap-tracker.md | 8 -------- 1 file changed, 8 deletions(-) delete mode 100644 tickets/011-approximate-heap-tracker.md diff --git a/tickets/011-approximate-heap-tracker.md b/tickets/011-approximate-heap-tracker.md deleted file mode 100644 index c4d403e..0000000 --- a/tickets/011-approximate-heap-tracker.md +++ /dev/null @@ -1,8 +0,0 @@ -# Add approximate heap/memory tracker for ingested logs - -An approximate heap tracker should exist. For each log ingested, its bytes are counted. If it's ever dropped, its bytes are subtracted. The size can be displayed on the frontend as comma-separated bytes. - -## Acceptance criteria -- Ingestion path increments a running byte counter by each log's approximate size when ingested. -- Eviction/drop path decrements the counter by that log's size when it's dropped. -- Current total is exposed via the API and displayed on the frontend, formatted with comma separators (e.g. "12,345,678 bytes").