From d0ce524e9526430556d794e9d29f9250a46e7c89 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Mon, 7 Sep 2026 01:24:02 +0100 Subject: [PATCH] Move approx_log_bytes into a flavour::LogLiteralHelper trait Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ --- src/engine/handle.rs | 166 ++++++++++++++++++++++++++++--------- src/engine/snapshot/mod.rs | 7 +- src/engine/tests.rs | 7 +- src/flavour/mod.rs | 12 +++ src/main.rs | 1 + 5 files changed, 148 insertions(+), 45 deletions(-) create mode 100644 src/flavour/mod.rs diff --git a/src/engine/handle.rs b/src/engine/handle.rs index f80ef97..0469c4d 100644 --- a/src/engine/handle.rs +++ b/src/engine/handle.rs @@ -1,19 +1,18 @@ //! The left-right write/read plumbing around `EngineSnapshot`, the actor loop that owns it, //! and the `EngineHandle` other tasks use to talk to that actor. -use crate::consts::{ - INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS, -}; +use crate::config::SettingsHandle; +use crate::consts::{INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS}; use crate::engine::alerts::{self, AlertRuntimeState}; use crate::engine::data_value::DataValue; use crate::engine::index_key::IndexKey; use crate::engine::message::EngineMessage; use crate::engine::metric_primitive::{MetricPrimitive, Number}; +use crate::engine::snapshot::EngineSnapshot; use crate::engine::snapshot::params::log_stream_header_transition; -use crate::engine::snapshot::{approx_log_bytes, EngineSnapshot}; use crate::engine::timestamp::Timestamp; use crate::engine::{ClearParamsResult, IngestedLine}; -use crate::config::SettingsHandle; +use crate::flavour::LogLiteralHelper; use left_right::Absorb; use std::collections::{BTreeMap, HashMap}; use tokio::sync::{mpsc, oneshot}; @@ -61,7 +60,9 @@ pub struct IndexReader { impl Clone for IndexReader { fn clone(&self) -> Self { - IndexReader { factory: self.factory.clone() } + IndexReader { + factory: self.factory.clone(), + } } } @@ -78,14 +79,22 @@ pub struct EngineHandle { impl Clone for EngineHandle { fn clone(&self) -> Self { - EngineHandle { tx: self.tx.clone(), reader: self.reader.clone() } + EngineHandle { + tx: self.tx.clone(), + reader: self.reader.clone(), + } } } impl EngineHandle { pub async fn barrier(&self) { let (respond_to, response) = oneshot::channel(); - if self.tx.send(EngineMessage::Barrier(respond_to)).await.is_ok() { + if self + .tx + .send(EngineMessage::Barrier(respond_to)) + .await + .is_ok() + { let _ = response.await; } } @@ -94,19 +103,35 @@ impl EngineHandle { pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle { let (tx, rx) = mpsc::channel(channel_capacity); let (write, read) = left_right::new::(); - let reader = IndexReader { factory: read.factory() }; + let reader = IndexReader { + factory: read.factory(), + }; tokio::spawn(run(rx, settings, IndexWriter { handle: write })); EngineHandle { tx, reader } } #[cfg(test)] -pub(crate) fn test_handle(channel_capacity: usize) -> (EngineHandle, mpsc::Receiver) { +pub(crate) fn test_handle( + channel_capacity: usize, +) -> (EngineHandle, mpsc::Receiver) { let (tx, rx) = mpsc::channel(channel_capacity); let (_write, read) = left_right::new::(); - (EngineHandle { tx, reader: IndexReader { factory: read.factory() } }, rx) + ( + EngineHandle { + tx, + reader: IndexReader { + factory: read.factory(), + }, + }, + rx, + ) } -async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mut write: IndexWriter) { +async fn run( + mut rx: mpsc::Receiver, + settings: SettingsHandle, + mut write: IndexWriter, +) { let mut max_logs: Option = None; let mut metric_sample_sequence: u64 = 0; let mut alert_states: HashMap = HashMap::new(); @@ -116,7 +141,11 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu while let Some(message) = rx.recv().await { match message { EngineMessage::Ingest(line) => { - let IngestedLine { sequence, timestamp, text } = line; + let IngestedLine { + sequence, + timestamp, + text, + } = line; let processed_at = Timestamp::now(); record_ingest_lag( &mut ingest_lag_samples, @@ -128,19 +157,32 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu let is_new_header = crate::engine::logfmt::LogStreamHeader::detect(&text).is_some(); log_stream_header = log_stream_header_transition(&log_stream_header, &text); - let row_header = if is_new_header { None } else { log_stream_header.clone() }; + let row_header = if is_new_header { + None + } else { + log_stream_header.clone() + }; let (rules, ntfy_topic, ntfy_server) = { let settings = settings.lock().unwrap(); - (settings.alert_rules.clone(), settings.ntfy_topic.clone(), settings.ntfy_server.clone()) + ( + settings.alert_rules.clone(), + settings.ntfy_topic.clone(), + settings.ntfy_server.clone(), + ) }; let text_for_op = text.clone(); write.mutate(move |snap: &mut EngineSnapshot| { - snap.approx_bytes_total += approx_log_bytes(&text_for_op); + snap.approx_bytes_total += text_for_op.approx_log_bytes(); snap.index.insert( - IndexKey::IngestedLog { sequence, timestamp }, - DataValue::LogLiteral { text: text_for_op.clone() }, + IndexKey::IngestedLog { + sequence, + timestamp, + }, + DataValue::LogLiteral { + text: text_for_op.clone(), + }, ); snap.log_count += 1; if let Some(header) = row_header.as_ref() { @@ -157,11 +199,23 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu let fired = write .peek() .map(|guard| { - alerts::evaluate_alerts(&rules, &mut alert_states, &guard, sequence, timestamp, &text) + alerts::evaluate_alerts( + &rules, + &mut alert_states, + &guard, + sequence, + timestamp, + &text, + ) }) .unwrap_or_default(); for (rule, message) in fired { - tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone())); + tokio::spawn(alerts::deliver_for_rule( + rule.name, + message, + ntfy_topic.clone(), + ntfy_server.clone(), + )); } } EngineMessage::IngestMetricSample(sample) => { @@ -175,7 +229,13 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu let timestamp = sample.timestamp; let value = MetricPrimitive::Number(Number::Float(sample.value)); write.mutate(move |snap: &mut EngineSnapshot| { - snap.push_bucket_value(&metric_name, labels.clone(), sequence, timestamp, value.clone()); + snap.push_bucket_value( + &metric_name, + labels.clone(), + sequence, + timestamp, + value.clone(), + ); }); metric_sample_sequence += 1; } @@ -185,7 +245,10 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu snap.evict_excess_logs(new_max); }); } - EngineMessage::DeclareParam { request, respond_to } => { + EngineMessage::DeclareParam { + request, + respond_to, + } => { let result = write .peek() .map(|guard| guard.validate_param_declaration(&request)) @@ -202,12 +265,16 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu let _ = respond_to.send(result); } EngineMessage::RemoveParamDeclaration { name, respond_to } => { - let existed = write - .peek() - .is_some_and(|guard| guard.index.contains_key(&IndexKey::ParamDeclaration { name: name.clone() })); + let existed = write.peek().is_some_and(|guard| { + guard + .index + .contains_key(&IndexKey::ParamDeclaration { name: name.clone() }) + }); let name_for_op = name.clone(); write.mutate(move |snap: &mut EngineSnapshot| { - snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() }); + snap.index.remove(&IndexKey::ParamDeclaration { + name: name_for_op.clone(), + }); snap.sort_helper_indexes.remove(&name_for_op); snap.param_regex_cache.remove(&name_for_op); }); @@ -215,7 +282,10 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu } EngineMessage::ClearParams { respond_to } => { let mut scratch = EngineSnapshot::from_index( - write.peek().map(|guard| guard.index.clone()).unwrap_or_default(), + write + .peek() + .map(|guard| guard.index.clone()) + .unwrap_or_default(), ); let removed = scratch.clear_all_param_declarations(); let (reextracted, logs_considered) = scratch.reextract_params_from_retained_logs(); @@ -224,9 +294,16 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu snap.reextract_params_from_retained_logs(); snap.sort_helper_indexes.clear(); }); - let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered }); + let _ = respond_to.send(ClearParamsResult { + removed, + reextracted, + logs_considered, + }); } - EngineMessage::DeclareMetric { request, respond_to } => { + EngineMessage::DeclareMetric { + request, + respond_to, + } => { let result = write .peek() .map(|guard| guard.validate_metric_request(&request).map(|_| ())) @@ -240,7 +317,11 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu }); let _ = respond_to.send(result); } - EngineMessage::UpdateMetric { name, request, respond_to } => { + EngineMessage::UpdateMetric { + name, + request, + respond_to, + } => { let result = write .peek() .map(|guard| guard.validate_update_metric(&name, &request).map(|_| ())) @@ -256,16 +337,21 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu let _ = respond_to.send(result); } EngineMessage::RemoveMetricDeclaration { name, respond_to } => { - let existed = write - .peek() - .is_some_and(|guard| guard.index.contains_key(&IndexKey::MetricDeclaration { name: name.clone() })); + let existed = write.peek().is_some_and(|guard| { + guard + .index + .contains_key(&IndexKey::MetricDeclaration { name: name.clone() }) + }); let name_for_op = name.clone(); write.mutate(move |snap: &mut EngineSnapshot| { snap.remove_metric_declaration(&name_for_op); }); let _ = respond_to.send(existed); } - EngineMessage::EnsureSortHelperIndex { param_name, respond_to } => { + EngineMessage::EnsureSortHelperIndex { + param_name, + respond_to, + } => { write.mutate(move |snap: &mut EngineSnapshot| { snap.ensure_sort_helper_index_built(¶m_name); }); @@ -279,12 +365,16 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu } pub(crate) fn record_ingest_lag(samples: &mut Vec<(Timestamp, i64)>, now: Timestamp, lag_ms: i64) { - samples.retain(|(recorded_at, _)| now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS); + samples.retain(|(recorded_at, _)| { + now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS + }); if samples.len() < INGEST_LAG_WATERMARK_MAX_ENTRIES { samples.push((now, lag_ms)); return; } - let Some((min_index, &(_, min_lag))) = samples.iter().enumerate().min_by_key(|(_, (_, lag))| *lag) else { + let Some((min_index, &(_, min_lag))) = + samples.iter().enumerate().min_by_key(|(_, (_, lag))| *lag) + else { return; }; if lag_ms > min_lag { @@ -295,7 +385,9 @@ pub(crate) fn record_ingest_lag(samples: &mut Vec<(Timestamp, i64)>, now: Timest pub(crate) fn ingest_lag_high_watermark_ms(samples: &[(Timestamp, i64)], now: Timestamp) -> i64 { samples .iter() - .filter(|(recorded_at, _)| now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS) + .filter(|(recorded_at, _)| { + now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS + }) .map(|(_, lag)| *lag) .max() .unwrap_or(0) diff --git a/src/engine/snapshot/mod.rs b/src/engine/snapshot/mod.rs index cfa6351..e91e9e7 100644 --- a/src/engine/snapshot/mod.rs +++ b/src/engine/snapshot/mod.rs @@ -18,6 +18,7 @@ use crate::consts::{ MAX_AUTO_DECLARED_PARAMS, MAX_NEW_AUTO_PARAMS_PER_LINE, OTTYL_RESERVED_PARAM_NAMES, SORT_HELPER_INDEX_PURGE_AFTER, }; +use crate::flavour::LogLiteralHelper; use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration}; use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; use crate::engine::index_key::IndexKey; @@ -123,10 +124,6 @@ impl EngineSnapshot { } } -pub(crate) fn approx_log_bytes(text: &str) -> u64 { - text.len() as u64 -} - impl EngineSnapshot { pub(crate) fn from_index(index: BTreeMap) -> Self { let mut snapshot = EngineSnapshot { index, ..Default::default() }; @@ -165,7 +162,7 @@ impl EngineSnapshot { break; }; if let Some(DataValue::LogLiteral { text }) = self.index.get(&oldest_key) { - self.approx_bytes_total = self.approx_bytes_total.saturating_sub(approx_log_bytes(text)); + self.approx_bytes_total = self.approx_bytes_total.saturating_sub(text.approx_log_bytes()); } self.index.remove(&oldest_key); self.log_count -= 1; diff --git a/src/engine/tests.rs b/src/engine/tests.rs index 61a5103..815af66 100644 --- a/src/engine/tests.rs +++ b/src/engine/tests.rs @@ -19,6 +19,7 @@ use crate::engine::snapshot::query::*; use crate::engine::snapshot::sort::*; use crate::engine::snapshot::*; use crate::engine::timestamp::*; +use crate::flavour::LogLiteralHelper; use std::collections::{BTreeMap, HashMap, HashSet}; use super::*; @@ -363,7 +364,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; let mut index = EngineSnapshot::default(); for sequence in 0..5u64 { let text = format!("line {sequence}"); - index.approx_bytes_total += approx_log_bytes(&text); + index.approx_bytes_total += text.approx_log_bytes(); index.index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, DataValue::LogLiteral { text }, @@ -377,7 +378,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; let remaining: Vec<_> = index.logs_range().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(); + remaining.iter().map(|seq| format!("line {seq}").approx_log_bytes()).sum(); assert_eq!(index.approx_bytes_total, expected_bytes); } @@ -413,7 +414,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; fn evict_excess_logs_is_noop_when_unlimited_or_under_cap() { let mut index = EngineSnapshot::default(); for sequence in 0..3u64 { - index.approx_bytes_total += approx_log_bytes("x"); + index.approx_bytes_total += "x".approx_log_bytes(); index.index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, DataValue::LogLiteral { text: "x".to_string() }, diff --git a/src/flavour/mod.rs b/src/flavour/mod.rs new file mode 100644 index 0000000..27723a6 --- /dev/null +++ b/src/flavour/mod.rs @@ -0,0 +1,12 @@ +pub trait LogLiteralHelper { + fn approx_log_bytes(&self) -> u64; +} + +impl LogLiteralHelper for T +where + T: AsRef, +{ + fn approx_log_bytes(&self) -> u64 { + self.as_ref().len() as u64 + } +} diff --git a/src/main.rs b/src/main.rs index cad58b9..b4ef5ee 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,7 @@ mod cli; mod config; mod consts; mod engine; +pub(crate) mod flavour; #[tokio::main] async fn main() {