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

Move approx_log_bytes into a flavour::LogLiteralHelper trait

Browse Source 
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ
This commit is contained in:
Hugh Kaznowski
2026-09-07 01:24:06 +01:00
co-authored by Claude Sonnet 5
parent 82fa0dfeb4
commit d0ce524e95
5 changed files with 148 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/engine/handle.rs
+129 -37
View File
@@ -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::<EngineSnapshot, EngineOp>();
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<EngineMessage>) {
pub(crate) fn test_handle(
channel_capacity: usize,
) -> (EngineHandle, mpsc::Receiver<EngineMessage>) {
let (tx, rx) = mpsc::channel(channel_capacity);
let (_write, read) = left_right::new::<EngineSnapshot, EngineOp>();
(EngineHandle { tx, reader: IndexReader { factory: read.factory() } }, rx)
(
EngineHandle {
tx,
reader: IndexReader {
factory: read.factory(),
},
},
rx,
)
}
async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle, mut write: IndexWriter) {
async fn run(
mut rx: mpsc::Receiver<EngineMessage>,
settings: SettingsHandle,
mut write: IndexWriter,
) {
let mut max_logs: Option<usize> = None;
let mut metric_sample_sequence: u64 = 0;
let mut alert_states: HashMap<String, AlertRuntimeState> = HashMap::new();
@@ -116,7 +141,11 @@ async fn run(mut rx: mpsc::Receiver<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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(&param_name);
});
@@ -279,12 +365,16 @@ async fn run(mut rx: mpsc::Receiver<EngineMessage>, 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)