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'