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

Merge ticket/011-approximate-heap-tracker

Browse Source 
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:32 +01:00
co-authored by Claude Sonnet 5
parent b1e509efbd 2c509b51fc
commit 5e2e556ea1
5 changed files with 73 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/dto.rs
+13
View File
@@ -47,6 +47,19 @@ pub struct StatsDto {
pub last_sequence: Option<u64>,
pub first_timestamp_ms: Option<i64>,
pub last_timestamp_ms: Option<i64>,
/// 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)]
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]"),
}
tickets/011-approximate-heap-tracker.md
-8
View File
@@ -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").
webapp/src/components/NavMenu.tsx
+1
View File
@@ -87,6 +87,7 @@ export function NavMenu({ stats }: { stats: Stats | null }) {
<div>
seq {stats.first_sequence ?? 0}–{stats.last_sequence ?? 0}
</div>
<div>{stats.approx_bytes_total.toLocaleString()} bytes</div>
</>
) : (
<div>connecting…</div>
webapp/src/lib/api.ts
+2
View File
@@ -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'