phughk/ottyl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Add approximate heap/memory tracker for ingested logs (ticket 011)

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o
This commit is contained in:
Hugh Kaznowski
2026-09-05 18:38:08 +01:00
co-authored by Claude Sonnet 5
parent b1e509efbd
commit 0834c246d6
4 changed files with 73 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/engine.rs
+57 -17
View File
@@ -128,18 +128,23 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, 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<String, AlertRuntimeState> = 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<EngineMessage>, 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<EngineMessage>, 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<IndexKey, DataValue>,
log_count: &mut usize,
max_logs: Option<usize>,
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<IndexKey, DataValue>) -> impl Iterator<Item = (u6
})
}
fn handle_query(index: &BTreeMap<IndexKey, DataValue>, query: EngineQuery) -> EngineResponse {
fn handle_query(
index: &BTreeMap<IndexKey, DataValue>,
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<IndexKey, DataValue>, 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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<u64> = 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]"),
}