Split src/engine/mod.rs into handle/message/snapshot submodules
Pure code-motion refactor, no logic changes. Follows the repo convention of one struct/enum + its impl per module file: - engine/handle.rs: the left-right write/read plumbing (EngineOp, IndexWriter, IndexReadHandle, IndexReader, EngineHandle) and the actor loop (spawn/run) that owns EngineSnapshot. - engine/message.rs: EngineQuery/EngineMessage/EngineResponse. - engine/snapshot/: EngineSnapshot itself, split by concern into params.rs (declaration + auto-detect), metrics.rs (declare/apply/ backfill), filter.rs (search-bar filter compilation/eval), sort.rs (sort-helper indexes), and query.rs (handle_query_inner + PromQL evaluation), with mod.rs holding the struct and the accessors shared across all of them. - engine/tests.rs: the inline test module, kept as one file (per the task's own escape hatch) since most tests exercise ingest -> declare -> query across several of the new submodules at once rather than one in isolation. Visibility was narrowed per call site: pub(super) where only sibling snapshot submodules need an item, pub(crate) where engine::handle or engine::tests (outside the snapshot subtree) also need it. cargo build --all-targets and cargo test are green (291 passed, 1 ignored, same as before the split). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS
This commit is contained in:
Hugh Kaznowski
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
3b3ace146a
commit
f19fa50a25
@@ -0,0 +1,301 @@
|
||||
//! 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::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::params::{derive_log_stream_params_from_row, 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 left_right::Absorb;
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
|
||||
struct EngineOp(Box<dyn FnMut(&mut EngineSnapshot) + Send + 'static>);
|
||||
|
||||
impl Absorb<EngineOp> for EngineSnapshot {
|
||||
fn absorb_first(&mut self, operation: &mut EngineOp, _other: &Self) {
|
||||
(operation.0)(self);
|
||||
}
|
||||
|
||||
fn drop_first(self: Box<Self>) {}
|
||||
|
||||
fn sync_with(&mut self, first: &Self) {
|
||||
*self = first.clone();
|
||||
}
|
||||
}
|
||||
|
||||
struct IndexWriter {
|
||||
handle: left_right::WriteHandle<EngineSnapshot, EngineOp>,
|
||||
}
|
||||
|
||||
impl IndexWriter {
|
||||
fn mutate(&mut self, op: impl FnMut(&mut EngineSnapshot) + Send + 'static) {
|
||||
self.handle.append(EngineOp(Box::new(op)));
|
||||
self.handle.publish();
|
||||
}
|
||||
|
||||
fn peek(&mut self) -> Option<left_right::ReadGuard<'_, EngineSnapshot>> {
|
||||
self.handle.enter()
|
||||
}
|
||||
}
|
||||
|
||||
pub struct IndexReadHandle(left_right::ReadHandle<EngineSnapshot>);
|
||||
|
||||
impl IndexReadHandle {
|
||||
pub(crate) fn enter(&self) -> Option<left_right::ReadGuard<'_, EngineSnapshot>> {
|
||||
self.0.enter()
|
||||
}
|
||||
}
|
||||
|
||||
pub struct IndexReader {
|
||||
factory: left_right::ReadHandleFactory<EngineSnapshot>,
|
||||
}
|
||||
|
||||
impl Clone for IndexReader {
|
||||
fn clone(&self) -> Self {
|
||||
IndexReader { factory: self.factory.clone() }
|
||||
}
|
||||
}
|
||||
|
||||
impl IndexReader {
|
||||
pub(crate) fn handle(&self) -> IndexReadHandle {
|
||||
IndexReadHandle(self.factory.handle())
|
||||
}
|
||||
}
|
||||
|
||||
pub struct EngineHandle {
|
||||
pub tx: mpsc::Sender<EngineMessage>,
|
||||
pub reader: IndexReader,
|
||||
}
|
||||
|
||||
impl Clone for EngineHandle {
|
||||
fn clone(&self) -> Self {
|
||||
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() {
|
||||
let _ = response.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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() };
|
||||
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>) {
|
||||
let (tx, rx) = mpsc::channel(channel_capacity);
|
||||
let (_write, read) = left_right::new::<EngineSnapshot, EngineOp>();
|
||||
(EngineHandle { tx, reader: IndexReader { factory: read.factory() } }, rx)
|
||||
}
|
||||
|
||||
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();
|
||||
let mut log_stream_header: Option<crate::engine::logfmt::LogStreamHeader> = None;
|
||||
let mut ingest_lag_samples: Vec<(Timestamp, i64)> = Vec::new();
|
||||
|
||||
while let Some(message) = rx.recv().await {
|
||||
match message {
|
||||
EngineMessage::Ingest(line) => {
|
||||
let IngestedLine { sequence, timestamp, text } = line;
|
||||
let processed_at = Timestamp::now();
|
||||
record_ingest_lag(
|
||||
&mut ingest_lag_samples,
|
||||
processed_at,
|
||||
(processed_at.as_millis() - timestamp.as_millis()).max(0),
|
||||
);
|
||||
let ingest_lag_high_watermark_ms =
|
||||
ingest_lag_high_watermark_ms(&ingest_lag_samples, processed_at);
|
||||
|
||||
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 (rules, ntfy_topic, ntfy_server) = {
|
||||
let settings = settings.lock().unwrap();
|
||||
(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.index.insert(
|
||||
IndexKey::IngestedLog { sequence, timestamp },
|
||||
DataValue::LogLiteral { text: text_for_op.clone() },
|
||||
);
|
||||
snap.log_count += 1;
|
||||
if let Some(header) = row_header.as_ref() {
|
||||
derive_log_stream_params_from_row(&mut snap.index, header, &text_for_op);
|
||||
}
|
||||
snap.auto_declare_params(&text_for_op);
|
||||
snap.apply_all_metrics(sequence, timestamp, &text_for_op);
|
||||
snap.update_sort_helper_indexes_on_ingest(sequence, timestamp, &text_for_op);
|
||||
snap.purge_stale_sort_helper_indexes();
|
||||
snap.evict_excess_logs(max_logs);
|
||||
snap.ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms;
|
||||
});
|
||||
|
||||
let fired = write
|
||||
.peek()
|
||||
.map(|guard| {
|
||||
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()));
|
||||
}
|
||||
}
|
||||
EngineMessage::IngestMetricSample(sample) => {
|
||||
let labels: BTreeMap<String, MetricPrimitive> = sample
|
||||
.labels
|
||||
.into_iter()
|
||||
.map(|(name, value)| (name, MetricPrimitive::String(value)))
|
||||
.collect();
|
||||
let metric_name = sample.metric_name;
|
||||
let sequence = metric_sample_sequence;
|
||||
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());
|
||||
});
|
||||
metric_sample_sequence += 1;
|
||||
}
|
||||
EngineMessage::SetMaxLogs(new_max) => {
|
||||
max_logs = new_max;
|
||||
write.mutate(move |snap: &mut EngineSnapshot| {
|
||||
snap.evict_excess_logs(new_max);
|
||||
});
|
||||
}
|
||||
EngineMessage::DeclareParam { request, respond_to } => {
|
||||
let result = write
|
||||
.peek()
|
||||
.map(|guard| guard.validate_param_declaration(&request))
|
||||
.unwrap_or_else(|| Err("engine is shutting down".to_string()));
|
||||
let succeeded = result.is_ok();
|
||||
let name = request.name.clone();
|
||||
let request_for_op = request.clone();
|
||||
write.mutate(move |snap: &mut EngineSnapshot| {
|
||||
snap.sort_helper_indexes.remove(&name);
|
||||
if succeeded {
|
||||
snap.insert_param_declaration(request_for_op.clone());
|
||||
}
|
||||
});
|
||||
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 name_for_op = name.clone();
|
||||
write.mutate(move |snap: &mut EngineSnapshot| {
|
||||
snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() });
|
||||
snap.sort_helper_indexes.remove(&name_for_op);
|
||||
});
|
||||
let _ = respond_to.send(existed);
|
||||
}
|
||||
EngineMessage::ClearParams { respond_to } => {
|
||||
let mut scratch = EngineSnapshot::from_index(
|
||||
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();
|
||||
write.mutate(move |snap: &mut EngineSnapshot| {
|
||||
snap.clear_all_param_declarations();
|
||||
snap.reextract_params_from_retained_logs();
|
||||
snap.sort_helper_indexes.clear();
|
||||
});
|
||||
let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered });
|
||||
}
|
||||
EngineMessage::DeclareMetric { request, respond_to } => {
|
||||
let result = write
|
||||
.peek()
|
||||
.map(|guard| guard.validate_metric_request(&request).map(|_| ()))
|
||||
.unwrap_or_else(|| Err("engine is shutting down".to_string()));
|
||||
let succeeded = result.is_ok();
|
||||
let request_for_op = request.clone();
|
||||
write.mutate(move |snap: &mut EngineSnapshot| {
|
||||
if succeeded {
|
||||
let _ = snap.declare_metric(request_for_op.clone());
|
||||
}
|
||||
});
|
||||
let _ = respond_to.send(result);
|
||||
}
|
||||
EngineMessage::UpdateMetric { name, request, respond_to } => {
|
||||
let result = write
|
||||
.peek()
|
||||
.map(|guard| guard.validate_update_metric(&name, &request).map(|_| ()))
|
||||
.unwrap_or_else(|| Err("engine is shutting down".to_string()));
|
||||
let succeeded = result.is_ok();
|
||||
let existing_name = name.clone();
|
||||
let request_for_op = request.clone();
|
||||
write.mutate(move |snap: &mut EngineSnapshot| {
|
||||
if succeeded {
|
||||
let _ = snap.update_metric(&existing_name, request_for_op.clone());
|
||||
}
|
||||
});
|
||||
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 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 } => {
|
||||
write.mutate(move |snap: &mut EngineSnapshot| {
|
||||
snap.ensure_sort_helper_index_built(¶m_name);
|
||||
});
|
||||
let _ = respond_to.send(());
|
||||
}
|
||||
EngineMessage::Barrier(respond_to) => {
|
||||
let _ = respond_to.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
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 {
|
||||
return;
|
||||
};
|
||||
if lag_ms > min_lag {
|
||||
samples[min_index] = (now, lag_ms);
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
.map(|(_, lag)| *lag)
|
||||
.max()
|
||||
.unwrap_or(0)
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
//! The engine actor's inbound message/query types and its outbound response type.
|
||||
|
||||
use crate::api::dto::{
|
||||
BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto,
|
||||
PromqlSeriesDto, StatsDto, TimelineBucketDto, TimelineParams,
|
||||
};
|
||||
use crate::engine::{ClearParamsResult, DeclareMetricRequest, DeclareParamRequest, IngestedLine, MetricSample};
|
||||
use std::collections::BTreeMap;
|
||||
use tokio::sync::oneshot;
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub enum EngineQuery {
|
||||
Logs(LogsParams),
|
||||
Timeline(TimelineParams),
|
||||
Stats,
|
||||
ParamDeclarations,
|
||||
MetricDeclarations,
|
||||
BucketValues {
|
||||
metric_name: String,
|
||||
labels_filter: Option<BTreeMap<String, String>>,
|
||||
start_ms: Option<i64>,
|
||||
end_ms: Option<i64>,
|
||||
since_sequence: Option<u64>,
|
||||
limit: Option<usize>,
|
||||
},
|
||||
Promql { query: String, start_ms: Option<i64>, end_ms: Option<i64> },
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub enum EngineMessage {
|
||||
Ingest(IngestedLine),
|
||||
IngestMetricSample(MetricSample),
|
||||
DeclareParam {
|
||||
request: DeclareParamRequest,
|
||||
respond_to: oneshot::Sender<Result<(), String>>,
|
||||
},
|
||||
RemoveParamDeclaration {
|
||||
name: String,
|
||||
respond_to: oneshot::Sender<bool>,
|
||||
},
|
||||
ClearParams {
|
||||
respond_to: oneshot::Sender<ClearParamsResult>,
|
||||
},
|
||||
DeclareMetric {
|
||||
request: DeclareMetricRequest,
|
||||
respond_to: oneshot::Sender<Result<(), String>>,
|
||||
},
|
||||
UpdateMetric {
|
||||
name: String,
|
||||
request: DeclareMetricRequest,
|
||||
respond_to: oneshot::Sender<Result<(), String>>,
|
||||
},
|
||||
RemoveMetricDeclaration {
|
||||
name: String,
|
||||
respond_to: oneshot::Sender<bool>,
|
||||
},
|
||||
SetMaxLogs(Option<usize>),
|
||||
EnsureSortHelperIndex {
|
||||
param_name: String,
|
||||
respond_to: oneshot::Sender<()>,
|
||||
},
|
||||
Barrier(oneshot::Sender<()>),
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub enum EngineResponse {
|
||||
Logs(Vec<LogEntryDto>),
|
||||
Timeline(Vec<TimelineBucketDto>),
|
||||
Stats(StatsDto),
|
||||
ParamDeclarations(Vec<ParamDeclarationDto>),
|
||||
MetricDeclarations(Vec<MetricDeclarationDto>),
|
||||
BucketValues(Vec<BucketValueDto>),
|
||||
Promql(Result<Vec<PromqlSeriesDto>, String>),
|
||||
Error(String),
|
||||
}
|
||||
|
||||
impl EngineQuery {
|
||||
pub(crate) fn sort_by(&self) -> Option<&str> {
|
||||
match self {
|
||||
EngineQuery::Logs(params) => params.sort_by.as_deref(),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
+14
-4383
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,261 @@
|
||||
//! Compiling and evaluating the search-bar filter syntax (`filter_parser`) against log lines
|
||||
//! and buckets: regex matches, the `ottyl_seq`/`ottyl_ts`/`ottyl_fmt` pseudo-metrics, per-param
|
||||
//! metric-bucket comparisons, and ad-hoc JSON field comparisons.
|
||||
|
||||
use crate::engine::filter_datetime;
|
||||
use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
|
||||
use crate::engine::index_key::IndexKey;
|
||||
use crate::engine::data_value::DataValue;
|
||||
use crate::engine::metric_primitive::MetricPrimitive;
|
||||
use crate::engine::snapshot::EngineSnapshot;
|
||||
use crate::engine::timestamp::Timestamp;
|
||||
use regex::Regex;
|
||||
use std::cmp::Ordering;
|
||||
use std::collections::HashSet;
|
||||
|
||||
pub(super) enum CompiledFilter {
|
||||
Regex(Regex),
|
||||
Sequence { op: CompareOp, target: f64 },
|
||||
Timestamp { op: CompareOp, target_ms: i64 },
|
||||
Fmt { op: CompareOp, target: String },
|
||||
MetricMatch(HashSet<u64>),
|
||||
Json { field: String, op: CompareOp, value: FilterValue },
|
||||
And(Box<CompiledFilter>, Box<CompiledFilter>),
|
||||
Or(Box<CompiledFilter>, Box<CompiledFilter>),
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(super) fn compile_filter_query(&self, filter: &str, now: Timestamp) -> Result<CompiledFilter, String> {
|
||||
let expr = filter_parser::parse(filter).map_err(|err| err.to_string())?;
|
||||
self.compile_filter(&expr, now)
|
||||
}
|
||||
|
||||
fn compile_filter(&self, expr: &FilterExpr, now: Timestamp) -> Result<CompiledFilter, String> {
|
||||
match expr {
|
||||
FilterExpr::Regex(pattern) => Regex::new(pattern)
|
||||
.map(CompiledFilter::Regex)
|
||||
.map_err(|err| format!("invalid regex in filter: {err}")),
|
||||
FilterExpr::Comparison { metric, param, op, value } => match metric.as_str() {
|
||||
"ottyl_seq" => CompiledFilter::from_seq_comparison(*op, value),
|
||||
"ottyl_ts" => CompiledFilter::from_ts_comparison(*op, value, now),
|
||||
"ottyl_fmt" => Ok(CompiledFilter::from_fmt_comparison(*op, value)),
|
||||
other => self.compile_metric_comparison(other, param.as_deref(), *op, value, now),
|
||||
},
|
||||
FilterExpr::JsonComparison { field, op, value } => Ok(CompiledFilter::Json {
|
||||
field: field.clone(),
|
||||
op: *op,
|
||||
value: value.clone(),
|
||||
}),
|
||||
FilterExpr::And(lhs, rhs) => Ok(CompiledFilter::And(
|
||||
Box::new(self.compile_filter(lhs, now)?),
|
||||
Box::new(self.compile_filter(rhs, now)?),
|
||||
)),
|
||||
FilterExpr::Or(lhs, rhs) => Ok(CompiledFilter::Or(
|
||||
Box::new(self.compile_filter(lhs, now)?),
|
||||
Box::new(self.compile_filter(rhs, now)?),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn compile_metric_comparison(
|
||||
&self,
|
||||
metric: &str,
|
||||
param: Option<&str>,
|
||||
op: CompareOp,
|
||||
value: &FilterValue,
|
||||
now: Timestamp,
|
||||
) -> Result<CompiledFilter, String> {
|
||||
let param = param.ok_or_else(|| {
|
||||
format!(
|
||||
"metric comparison on '{metric}' requires a [param] (e.g. '{metric}[some_param]{}...'); \
|
||||
v1 buckets only carry per-label occurrence data, not a single metric value",
|
||||
match op {
|
||||
CompareOp::Lt => "<",
|
||||
CompareOp::Gt => ">",
|
||||
CompareOp::Le => "<=",
|
||||
CompareOp::Ge => ">=",
|
||||
CompareOp::Eq => "==",
|
||||
CompareOp::Ne => "!=",
|
||||
}
|
||||
)
|
||||
})?;
|
||||
|
||||
if let Some(DataValue::MetricDeclaration { label_params, .. }) =
|
||||
self.index.get(&IndexKey::MetricDeclaration { name: metric.to_string() })
|
||||
{
|
||||
if !label_params.iter().any(|p| p == param) {
|
||||
return Err(format!(
|
||||
"param '{param}' is not a label param of metric '{metric}'; declared label params: [{}]",
|
||||
label_params.join(", ")
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
let mut matching = HashSet::new();
|
||||
for (labels, (sequence, _, _)) in self.bucket_values(metric, None) {
|
||||
let Some(primitive) = labels.get(param) else {
|
||||
continue;
|
||||
};
|
||||
if primitive.compare(op, value, now)? {
|
||||
matching.insert(*sequence);
|
||||
}
|
||||
}
|
||||
Ok(CompiledFilter::MetricMatch(matching))
|
||||
}
|
||||
}
|
||||
|
||||
impl CompiledFilter {
|
||||
fn from_seq_comparison(op: CompareOp, value: &FilterValue) -> Result<Self, String> {
|
||||
let target = match value {
|
||||
FilterValue::Number(n) => *n,
|
||||
FilterValue::Raw(raw) => raw.parse::<f64>().map_err(|_| {
|
||||
format!("'ottyl_seq' comparisons require a numeric value, got {raw:?}")
|
||||
})?,
|
||||
};
|
||||
Ok(CompiledFilter::Sequence { op, target })
|
||||
}
|
||||
|
||||
fn from_ts_comparison(op: CompareOp, value: &FilterValue, now: Timestamp) -> Result<Self, String> {
|
||||
let target_ms = match value {
|
||||
FilterValue::Number(n) => *n as i64,
|
||||
FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, now)
|
||||
.map_err(|err| err.to_string())?
|
||||
.as_millis(),
|
||||
};
|
||||
Ok(CompiledFilter::Timestamp { op, target_ms })
|
||||
}
|
||||
|
||||
fn from_fmt_comparison(op: CompareOp, value: &FilterValue) -> Self {
|
||||
let target = match value {
|
||||
FilterValue::Number(n) => n.to_string(),
|
||||
FilterValue::Raw(raw) => raw.clone(),
|
||||
};
|
||||
CompiledFilter::Fmt { op, target }
|
||||
}
|
||||
}
|
||||
|
||||
impl MetricPrimitive {
|
||||
fn compare(&self, op: CompareOp, value: &FilterValue, now: Timestamp) -> Result<bool, String> {
|
||||
let ordering = match (self, value) {
|
||||
(MetricPrimitive::Number(n), FilterValue::Number(v)) => n.as_f64().partial_cmp(v),
|
||||
(MetricPrimitive::Number(n), FilterValue::Raw(raw)) => {
|
||||
let v: f64 = raw
|
||||
.parse()
|
||||
.map_err(|_| format!("expected a numeric value, got {raw:?}"))?;
|
||||
n.as_f64().partial_cmp(&v)
|
||||
}
|
||||
(MetricPrimitive::Timestamp(ts), FilterValue::Number(v)) => {
|
||||
ts.as_millis().partial_cmp(&(*v as i64))
|
||||
}
|
||||
(MetricPrimitive::Timestamp(ts), FilterValue::Raw(raw)) => {
|
||||
let target = filter_datetime::parse_datetime(raw, now).map_err(|err| err.to_string())?;
|
||||
ts.as_millis().partial_cmp(&target.as_millis())
|
||||
}
|
||||
(MetricPrimitive::String(s), FilterValue::Raw(raw)) => s.as_str().partial_cmp(raw.as_str()),
|
||||
(MetricPrimitive::String(s), FilterValue::Number(n)) => {
|
||||
s.as_str().partial_cmp(n.to_string().as_str())
|
||||
}
|
||||
(MetricPrimitive::Duration(ms), FilterValue::Number(v)) => (*ms as f64).partial_cmp(v),
|
||||
(MetricPrimitive::Duration(ms), FilterValue::Raw(raw)) => {
|
||||
let v: f64 = raw
|
||||
.parse()
|
||||
.map_err(|_| format!("expected a numeric value, got {raw:?}"))?;
|
||||
(*ms as f64).partial_cmp(&v)
|
||||
}
|
||||
(MetricPrimitive::Boolean(b), FilterValue::Raw(raw)) => {
|
||||
let v: bool = raw
|
||||
.parse()
|
||||
.map_err(|_| format!("expected true or false, got {raw:?}"))?;
|
||||
b.partial_cmp(&v)
|
||||
}
|
||||
(MetricPrimitive::Boolean(_), FilterValue::Number(_)) => None,
|
||||
};
|
||||
Ok(ordering.is_some_and(|ordering| op.matches(ordering)))
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn eval_json_comparison(text: &str, field: &str, op: CompareOp, value: &FilterValue) -> bool {
|
||||
let Ok(parsed) = serde_json::from_str::<serde_json::Value>(text) else {
|
||||
return false;
|
||||
};
|
||||
let Some(field_value) = parsed.as_object().and_then(|obj| obj.get(field)) else {
|
||||
return false;
|
||||
};
|
||||
compare_json_value(field_value, op, value)
|
||||
}
|
||||
|
||||
fn compare_json_value(field_value: &serde_json::Value, op: CompareOp, value: &FilterValue) -> bool {
|
||||
match field_value {
|
||||
serde_json::Value::Number(n) => {
|
||||
let target = match value {
|
||||
FilterValue::Number(v) => Some(*v),
|
||||
FilterValue::Raw(raw) => raw.parse::<f64>().ok(),
|
||||
};
|
||||
match (n.as_f64(), target) {
|
||||
(Some(f), Some(v)) => f.partial_cmp(&v).is_some_and(|ordering| op.matches(ordering)),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
serde_json::Value::String(s) => {
|
||||
let target = match value {
|
||||
FilterValue::Raw(raw) => raw.clone(),
|
||||
FilterValue::Number(n) => n.to_string(),
|
||||
};
|
||||
s.as_str()
|
||||
.partial_cmp(target.as_str())
|
||||
.is_some_and(|ordering| op.matches(ordering))
|
||||
}
|
||||
serde_json::Value::Bool(b) => {
|
||||
let FilterValue::Raw(raw) = value else {
|
||||
return false;
|
||||
};
|
||||
match raw.parse::<bool>() {
|
||||
Ok(v) => match op {
|
||||
CompareOp::Eq => *b == v,
|
||||
CompareOp::Ne => *b != v,
|
||||
_ => false,
|
||||
},
|
||||
Err(_) => false,
|
||||
}
|
||||
}
|
||||
serde_json::Value::Null | serde_json::Value::Array(_) | serde_json::Value::Object(_) => false,
|
||||
}
|
||||
}
|
||||
|
||||
impl CompareOp {
|
||||
pub(crate) fn matches(&self, ordering: Ordering) -> bool {
|
||||
match self {
|
||||
CompareOp::Lt => ordering == Ordering::Less,
|
||||
CompareOp::Gt => ordering == Ordering::Greater,
|
||||
CompareOp::Le => ordering != Ordering::Greater,
|
||||
CompareOp::Ge => ordering != Ordering::Less,
|
||||
CompareOp::Eq => ordering == Ordering::Equal,
|
||||
CompareOp::Ne => ordering != Ordering::Equal,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl CompiledFilter {
|
||||
pub(super) fn eval(&self, sequence: u64, timestamp: Timestamp, text: &str) -> bool {
|
||||
match self {
|
||||
CompiledFilter::Regex(re) => re.is_match(text),
|
||||
CompiledFilter::Sequence { op, target } => {
|
||||
op.matches((sequence as f64).total_cmp(target))
|
||||
}
|
||||
CompiledFilter::Timestamp { op, target_ms } => {
|
||||
op.matches(timestamp.as_millis().cmp(target_ms))
|
||||
}
|
||||
CompiledFilter::Fmt { op, target } => {
|
||||
op.matches(crate::engine::logfmt::detect_format(text).cmp(target.as_str()))
|
||||
}
|
||||
CompiledFilter::MetricMatch(sequences) => sequences.contains(&sequence),
|
||||
CompiledFilter::Json { field, op, value } => eval_json_comparison(text, field, *op, value),
|
||||
CompiledFilter::And(lhs, rhs) => {
|
||||
lhs.eval(sequence, timestamp, text) && rhs.eval(sequence, timestamp, text)
|
||||
}
|
||||
CompiledFilter::Or(lhs, rhs) => {
|
||||
lhs.eval(sequence, timestamp, text) || rhs.eval(sequence, timestamp, text)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,331 @@
|
||||
//! Metric declaration and application: validating/storing declarations, applying them to
|
||||
//! ingested lines to populate metric buckets, and backfilling a newly (re)declared metric
|
||||
//! against already-ingested logs.
|
||||
|
||||
use crate::engine::data_value::{DataValue, MetricPrimitiveKind};
|
||||
use crate::engine::index_key::IndexKey;
|
||||
use crate::engine::metric_primitive::{MetricPrimitive, Number};
|
||||
use crate::engine::snapshot::params::ResolvedParam;
|
||||
use crate::engine::snapshot::EngineSnapshot;
|
||||
use crate::engine::timestamp::Timestamp;
|
||||
use crate::engine::DeclareMetricRequest;
|
||||
use regex::Regex;
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(crate) fn validate_metric_request(&self, request: &DeclareMetricRequest) -> Result<Regex, String> {
|
||||
let re = Regex::new(&request.message_match_regex)
|
||||
.map_err(|err| format!("invalid regex: {err}"))?;
|
||||
|
||||
let declared =
|
||||
|name: &str| self.index.contains_key(&IndexKey::ParamDeclaration { name: name.to_string() });
|
||||
|
||||
let mut errors: Vec<String> = request
|
||||
.label_params
|
||||
.iter()
|
||||
.filter(|name| !declared(name))
|
||||
.map(|name| format!("label param '{name}' references an undeclared param"))
|
||||
.collect();
|
||||
if let Some(value_param) = &request.value_param {
|
||||
if !declared(value_param) {
|
||||
errors.push(format!("value param '{value_param}' references an undeclared param"));
|
||||
}
|
||||
}
|
||||
if !errors.is_empty() {
|
||||
return Err(errors.join("; "));
|
||||
}
|
||||
|
||||
Ok(re)
|
||||
}
|
||||
|
||||
pub(crate) fn declare_metric(&mut self, request: DeclareMetricRequest) -> Result<(), String> {
|
||||
let re = self.validate_metric_request(&request)?;
|
||||
let resolved_params = self.resolved_param_declarations();
|
||||
|
||||
self.backfill_metric(
|
||||
&request.name,
|
||||
&re,
|
||||
&request.label_params,
|
||||
request.value_param.as_deref(),
|
||||
&resolved_params,
|
||||
request.matched_labels.as_ref(),
|
||||
request.unmatched_labels.as_ref(),
|
||||
);
|
||||
self.index.insert(
|
||||
IndexKey::MetricDeclaration { name: request.name },
|
||||
DataValue::MetricDeclaration {
|
||||
message_match_regex: request.message_match_regex,
|
||||
label_params: request.label_params,
|
||||
value_param: request.value_param,
|
||||
matched_labels: request.matched_labels,
|
||||
unmatched_labels: request.unmatched_labels,
|
||||
},
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn validate_update_metric(
|
||||
&self,
|
||||
existing_name: &str,
|
||||
request: &DeclareMetricRequest,
|
||||
) -> Result<Regex, String> {
|
||||
if !self.index.contains_key(&IndexKey::MetricDeclaration { name: existing_name.to_string() }) {
|
||||
return Err(format!("no metric declaration named '{existing_name}'"));
|
||||
}
|
||||
if request.name != existing_name
|
||||
&& self.index.contains_key(&IndexKey::MetricDeclaration { name: request.name.clone() })
|
||||
{
|
||||
return Err(format!("a metric named '{}' already exists", request.name));
|
||||
}
|
||||
self.validate_metric_request(request)
|
||||
}
|
||||
|
||||
pub(crate) fn update_metric(&mut self, existing_name: &str, request: DeclareMetricRequest) -> Result<(), String> {
|
||||
let re = self.validate_update_metric(existing_name, &request)?;
|
||||
let resolved_params = self.resolved_param_declarations();
|
||||
|
||||
self.clear_metric_data(existing_name);
|
||||
if request.name != existing_name {
|
||||
self.index.remove(&IndexKey::MetricDeclaration { name: existing_name.to_string() });
|
||||
}
|
||||
|
||||
self.backfill_metric(
|
||||
&request.name,
|
||||
&re,
|
||||
&request.label_params,
|
||||
request.value_param.as_deref(),
|
||||
&resolved_params,
|
||||
request.matched_labels.as_ref(),
|
||||
request.unmatched_labels.as_ref(),
|
||||
);
|
||||
self.index.insert(
|
||||
IndexKey::MetricDeclaration { name: request.name.clone() },
|
||||
DataValue::MetricDeclaration {
|
||||
message_match_regex: request.message_match_regex,
|
||||
label_params: request.label_params,
|
||||
value_param: request.value_param,
|
||||
matched_labels: request.matched_labels,
|
||||
unmatched_labels: request.unmatched_labels,
|
||||
},
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn clear_metric_data(&mut self, metric_name: &str) {
|
||||
self.index.retain(|key, _| match key {
|
||||
IndexKey::Bucket { metric_name: name, .. } => name != metric_name,
|
||||
IndexKey::LastMetricEvent { metric_name: name, .. } => name != metric_name,
|
||||
_ => true,
|
||||
});
|
||||
}
|
||||
|
||||
pub(crate) fn remove_metric_declaration(&mut self, name: &str) -> bool {
|
||||
let existed = self.index.remove(&IndexKey::MetricDeclaration { name: name.to_string() }).is_some();
|
||||
self.clear_metric_data(name);
|
||||
existed
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Option<MetricPrimitive> {
|
||||
match kind {
|
||||
MetricPrimitiveKind::String => Some(MetricPrimitive::String(raw.to_string())),
|
||||
MetricPrimitiveKind::Number => raw
|
||||
.parse::<i64>()
|
||||
.map(Number::Int)
|
||||
.or_else(|_| raw.parse::<f64>().map(Number::Float))
|
||||
.ok()
|
||||
.map(MetricPrimitive::Number),
|
||||
MetricPrimitiveKind::Timestamp => {
|
||||
crate::engine::filter_datetime::parse_datetime(raw, Timestamp::now())
|
||||
.ok()
|
||||
.map(MetricPrimitive::Timestamp)
|
||||
}
|
||||
MetricPrimitiveKind::Duration => raw.parse::<i64>().ok().map(MetricPrimitive::Duration),
|
||||
MetricPrimitiveKind::Boolean => raw.parse::<bool>().ok().map(MetricPrimitive::Boolean),
|
||||
}
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn apply_metric_to_log(
|
||||
&mut self,
|
||||
metric_name: &str,
|
||||
regex: &Regex,
|
||||
label_params: &[String],
|
||||
value_param: Option<&str>,
|
||||
resolved_params: &HashMap<String, ResolvedParam>,
|
||||
matched_labels: Option<&BTreeMap<String, String>>,
|
||||
unmatched_labels: Option<&BTreeMap<String, String>>,
|
||||
sequence: u64,
|
||||
timestamp: Timestamp,
|
||||
text: &str,
|
||||
) {
|
||||
if regex.captures(text).is_none() {
|
||||
if let Some(unmatched_labels) = unmatched_labels {
|
||||
let labels = unmatched_labels
|
||||
.iter()
|
||||
.map(|(name, value)| (name.clone(), MetricPrimitive::String(value.clone())))
|
||||
.collect();
|
||||
self.push_bucket_value(metric_name, labels, sequence, timestamp, MetricPrimitive::Number(Number::Int(1)));
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
let mut labels = BTreeMap::new();
|
||||
for param_name in label_params {
|
||||
let Some(resolved) = resolved_params.get(param_name) else { return };
|
||||
let Some(parsed) = self.resolve_param_value(metric_name, resolved, &labels, text, timestamp) else {
|
||||
return;
|
||||
};
|
||||
labels.insert(param_name.clone(), parsed);
|
||||
}
|
||||
|
||||
if let Some(matched_labels) = matched_labels {
|
||||
for (name, value) in matched_labels {
|
||||
labels.insert(name.clone(), MetricPrimitive::String(value.clone()));
|
||||
}
|
||||
}
|
||||
|
||||
let value = match value_param {
|
||||
Some(value_param) => {
|
||||
let Some(resolved) = resolved_params.get(value_param) else { return };
|
||||
match self.resolve_param_value(metric_name, resolved, &labels, text, timestamp) {
|
||||
Some(parsed) => parsed,
|
||||
None => return,
|
||||
}
|
||||
}
|
||||
None => MetricPrimitive::Number(Number::Int(1)),
|
||||
};
|
||||
|
||||
self.push_bucket_value(metric_name, labels, sequence, timestamp, value);
|
||||
}
|
||||
|
||||
pub(crate) fn push_bucket_value(
|
||||
&mut self,
|
||||
metric_name: &str,
|
||||
labels: BTreeMap<String, MetricPrimitive>,
|
||||
sequence: u64,
|
||||
timestamp: Timestamp,
|
||||
value: MetricPrimitive,
|
||||
) {
|
||||
let key = IndexKey::Bucket { metric_name: metric_name.to_string(), labels };
|
||||
let entry = self
|
||||
.index
|
||||
.entry(key)
|
||||
.or_insert_with(|| DataValue::BucketValue { values: Vec::new() });
|
||||
if let DataValue::BucketValue { values } = entry {
|
||||
values.push((sequence, timestamp, value));
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn backfill_metric(
|
||||
&mut self,
|
||||
metric_name: &str,
|
||||
regex: &Regex,
|
||||
label_params: &[String],
|
||||
value_param: Option<&str>,
|
||||
resolved_params: &HashMap<String, ResolvedParam>,
|
||||
matched_labels: Option<&BTreeMap<String, String>>,
|
||||
unmatched_labels: Option<&BTreeMap<String, String>>,
|
||||
) {
|
||||
let lines: Vec<(u64, Timestamp, String)> = self
|
||||
.logs_range()
|
||||
.map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string()))
|
||||
.collect();
|
||||
for (sequence, timestamp, text) in lines {
|
||||
self.apply_metric_to_log(
|
||||
metric_name,
|
||||
regex,
|
||||
label_params,
|
||||
value_param,
|
||||
resolved_params,
|
||||
matched_labels,
|
||||
unmatched_labels,
|
||||
sequence,
|
||||
timestamp,
|
||||
&text,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn apply_all_metrics(&mut self, sequence: u64, timestamp: Timestamp, text: &str) {
|
||||
type SeededMetric = (
|
||||
String,
|
||||
String,
|
||||
Vec<String>,
|
||||
Option<String>,
|
||||
Option<BTreeMap<String, String>>,
|
||||
Option<BTreeMap<String, String>>,
|
||||
);
|
||||
let metrics: Vec<SeededMetric> = self
|
||||
.metric_declarations_range()
|
||||
.map(|(name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels)| {
|
||||
(
|
||||
name.to_string(),
|
||||
message_match_regex.to_string(),
|
||||
label_params.to_vec(),
|
||||
value_param.map(|s| s.to_string()),
|
||||
matched_labels.cloned(),
|
||||
unmatched_labels.cloned(),
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
let resolved_params = self.resolved_param_declarations();
|
||||
for (name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels) in metrics {
|
||||
if let Ok(re) = Regex::new(&message_match_regex) {
|
||||
self.apply_metric_to_log(
|
||||
&name,
|
||||
&re,
|
||||
&label_params,
|
||||
value_param.as_deref(),
|
||||
&resolved_params,
|
||||
matched_labels.as_ref(),
|
||||
unmatched_labels.as_ref(),
|
||||
sequence,
|
||||
timestamp,
|
||||
text,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn bucket_values<'a>(
|
||||
&'a self,
|
||||
metric_name: &'a str,
|
||||
labels_filter: Option<&'a BTreeMap<String, MetricPrimitive>>,
|
||||
) -> impl Iterator<Item = (&'a BTreeMap<String, MetricPrimitive>, &'a (u64, Timestamp, MetricPrimitive))> + 'a
|
||||
{
|
||||
self.index
|
||||
.range(IndexKey::MetricBucketSplit..IndexKey::End)
|
||||
.filter_map(move |(key, value)| match (key, value) {
|
||||
(IndexKey::Bucket { metric_name: name, labels }, DataValue::BucketValue { values })
|
||||
if name == metric_name && labels_filter.is_none_or(|filter| filter == labels) =>
|
||||
{
|
||||
Some(values.iter().map(move |entry| (labels, entry)))
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.flatten()
|
||||
}
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(crate) fn declared_value_kind(&self, metric_name: &str) -> Option<MetricPrimitiveKind> {
|
||||
match self.index.get(&IndexKey::MetricDeclaration { name: metric_name.to_string() }) {
|
||||
Some(DataValue::MetricDeclaration { value_param, .. }) => Some(
|
||||
value_param
|
||||
.as_ref()
|
||||
.and_then(|name| {
|
||||
self.index.get(&IndexKey::ParamDeclaration { name: name.clone() })
|
||||
})
|
||||
.map(|value| match value {
|
||||
DataValue::ParamDeclaration(param) => param.kind,
|
||||
_ => MetricPrimitiveKind::Number,
|
||||
})
|
||||
.unwrap_or(MetricPrimitiveKind::Number),
|
||||
),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
//! `EngineSnapshot` is the actor's live state: ingested log lines, declared params/metrics,
|
||||
//! computed metric buckets, and the sort-helper indexes built on demand — all stored in one
|
||||
//! flat `BTreeMap<IndexKey, DataValue>` and read/written through the left-right index in
|
||||
//! `super::handle`. Its impl is split across the submodules below by concern; this file holds
|
||||
//! the struct itself plus the accessors every submodule builds on.
|
||||
|
||||
pub(crate) mod filter;
|
||||
pub(crate) mod metrics;
|
||||
pub(crate) mod params;
|
||||
pub(crate) mod query;
|
||||
pub(crate) mod sort;
|
||||
|
||||
pub(crate) use filter::eval_json_comparison;
|
||||
|
||||
use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration};
|
||||
use crate::engine::index_key::IndexKey;
|
||||
use crate::engine::message::{EngineQuery, EngineResponse};
|
||||
use crate::engine::timestamp::Timestamp;
|
||||
use query::handle_query_inner;
|
||||
use sort::{SortHelperAccess, SortHelperIndexEntry};
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub(crate) struct EngineSnapshot {
|
||||
pub(crate) index: BTreeMap<IndexKey, DataValue>,
|
||||
pub(crate) sort_helper_indexes: HashMap<String, SortHelperIndexEntry>,
|
||||
pub(crate) log_count: usize,
|
||||
pub(crate) approx_bytes_total: u64,
|
||||
pub(crate) ingest_lag_high_watermark_ms: i64,
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(crate) fn metric_declarations_range(
|
||||
&self,
|
||||
) -> impl Iterator<
|
||||
Item = (
|
||||
&str,
|
||||
&str,
|
||||
&[String],
|
||||
Option<&str>,
|
||||
Option<&BTreeMap<String, String>>,
|
||||
Option<&BTreeMap<String, String>>,
|
||||
),
|
||||
> {
|
||||
self.index
|
||||
.range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit)
|
||||
.filter_map(|(key, value)| match (key, value) {
|
||||
(
|
||||
IndexKey::MetricDeclaration { name },
|
||||
DataValue::MetricDeclaration {
|
||||
message_match_regex,
|
||||
label_params,
|
||||
value_param,
|
||||
matched_labels,
|
||||
unmatched_labels,
|
||||
},
|
||||
) => Some((
|
||||
name.as_str(),
|
||||
message_match_regex.as_str(),
|
||||
label_params.as_slice(),
|
||||
value_param.as_deref(),
|
||||
matched_labels.as_ref(),
|
||||
unmatched_labels.as_ref(),
|
||||
)),
|
||||
_ => None,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn param_declarations_range(
|
||||
&self,
|
||||
) -> impl Iterator<Item = (&str, &str, &str, Option<&str>, MetricPrimitiveKind, bool, bool, bool)> {
|
||||
self.index
|
||||
.range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit)
|
||||
.filter_map(|(key, value)| match (key, value) {
|
||||
(
|
||||
IndexKey::ParamDeclaration { name },
|
||||
DataValue::ParamDeclaration(ParamDeclaration {
|
||||
name: _,
|
||||
message_match_regex,
|
||||
capture_group,
|
||||
json_field_path,
|
||||
kind,
|
||||
elapsed_since_last,
|
||||
auto_derived,
|
||||
log_stream_gated,
|
||||
}),
|
||||
) => Some((
|
||||
name.as_str(),
|
||||
message_match_regex.as_str(),
|
||||
capture_group.as_str(),
|
||||
json_field_path.as_deref(),
|
||||
*kind,
|
||||
*elapsed_since_last,
|
||||
*auto_derived,
|
||||
*log_stream_gated,
|
||||
)),
|
||||
_ => None,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn approx_log_bytes(text: &str) -> u64 {
|
||||
text.len() as u64
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(crate) fn from_index(index: BTreeMap<IndexKey, DataValue>) -> Self {
|
||||
EngineSnapshot { index, ..Default::default() }
|
||||
}
|
||||
|
||||
pub(crate) fn evict_excess_logs(&mut self, max_logs: Option<usize>) {
|
||||
let Some(max_logs) = max_logs else { return };
|
||||
while self.log_count > max_logs {
|
||||
let Some(oldest_key) = self
|
||||
.index
|
||||
.range(IndexKey::Start..IndexKey::LogParamSplit)
|
||||
.next()
|
||||
.map(|(key, _)| key.clone())
|
||||
else {
|
||||
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.index.remove(&oldest_key);
|
||||
self.log_count -= 1;
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn logs_range(&self) -> impl Iterator<Item = (u64, Timestamp, &str)> {
|
||||
self.index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| {
|
||||
match (key, value) {
|
||||
(IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => {
|
||||
Some((*sequence, *timestamp, text.as_str()))
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn logs_since(
|
||||
&self,
|
||||
before_sequence: u64,
|
||||
since_ms: i64,
|
||||
now_ms: i64,
|
||||
) -> impl Iterator<Item = (u64, Timestamp, &str)> {
|
||||
self.logs_range().filter(move |(sequence, timestamp, _)| {
|
||||
*sequence < before_sequence
|
||||
&& timestamp.as_millis() > since_ms
|
||||
&& timestamp.as_millis() <= now_ms
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(crate) fn handle_query_ro(&self, query: EngineQuery) -> EngineResponse {
|
||||
handle_query_inner(self, query, SortHelperAccess::Immutable(&self.sort_helper_indexes))
|
||||
}
|
||||
|
||||
pub(crate) fn has_sort_helper_index(&self, sort_by: &str) -> bool {
|
||||
self.sort_helper_indexes.contains_key(sort_by)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,339 @@
|
||||
//! Param declaration: validating and storing explicit declarations, and the auto-detection
|
||||
//! logic that derives declarations from logfmt/json/log-stream-shaped lines as they're ingested.
|
||||
|
||||
use crate::consts::{MAX_AUTO_DECLARED_PARAMS, MAX_NEW_AUTO_PARAMS_PER_LINE, OTTYL_RESERVED_PARAM_NAMES};
|
||||
use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration};
|
||||
use crate::engine::index_key::IndexKey;
|
||||
use crate::engine::metric_primitive::MetricPrimitive;
|
||||
use crate::engine::snapshot::metrics::parse_metric_primitive;
|
||||
use crate::engine::snapshot::EngineSnapshot;
|
||||
use crate::engine::timestamp::Timestamp;
|
||||
use crate::engine::DeclareParamRequest;
|
||||
use regex::Regex;
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(crate) fn validate_param_declaration(&self, request: &DeclareParamRequest) -> Result<(), String> {
|
||||
if OTTYL_RESERVED_PARAM_NAMES.contains(&request.name.as_str()) {
|
||||
return Err(format!(
|
||||
"'{}' is a reserved name, present on every log line without declaring anything; \
|
||||
it can't be redeclared as a param",
|
||||
request.name
|
||||
));
|
||||
}
|
||||
if let Some(DataValue::ParamDeclaration(existing)) =
|
||||
self.index.get(&IndexKey::ParamDeclaration { name: request.name.clone() })
|
||||
{
|
||||
if existing.auto_derived {
|
||||
return Err(format!(
|
||||
"param '{}' was auto-detected from the log stream and can't be edited",
|
||||
request.name
|
||||
));
|
||||
}
|
||||
}
|
||||
let re = Regex::new(&request.message_match_regex)
|
||||
.map_err(|err| format!("invalid regex: {err}"))?;
|
||||
if request.elapsed_since_last && request.kind != MetricPrimitiveKind::Duration {
|
||||
return Err("elapsed_since_last params must declare kind 'duration'".to_string());
|
||||
}
|
||||
let uses_json_field = request.json_field_path.as_deref().is_some_and(|path| !path.is_empty());
|
||||
if request.kind != MetricPrimitiveKind::Boolean
|
||||
&& !request.elapsed_since_last
|
||||
&& !uses_json_field
|
||||
&& !re.capture_names().flatten().any(|name| name == request.capture_group)
|
||||
{
|
||||
return Err(format!(
|
||||
"regex has no capture group named '{}'",
|
||||
request.capture_group
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn insert_param_declaration(&mut self, request: DeclareParamRequest) {
|
||||
self.index.insert(
|
||||
IndexKey::ParamDeclaration { name: request.name.clone() },
|
||||
DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn declare_param(snapshot: &mut EngineSnapshot, request: DeclareParamRequest) -> Result<(), String> {
|
||||
snapshot.validate_param_declaration(&request)?;
|
||||
snapshot.insert_param_declaration(request);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn sanitize_group_name(key: &str) -> String {
|
||||
let mut group: String =
|
||||
key.chars().map(|c| if c.is_ascii_alphanumeric() || c == '_' { c } else { '_' }).collect();
|
||||
if group.chars().next().is_none_or(|c| c.is_ascii_digit()) {
|
||||
group.insert(0, '_');
|
||||
}
|
||||
group
|
||||
}
|
||||
|
||||
fn infer_kind(sample: &str) -> MetricPrimitiveKind {
|
||||
if sample.parse::<i64>().is_ok() || sample.parse::<f64>().is_ok() {
|
||||
MetricPrimitiveKind::Number
|
||||
} else {
|
||||
MetricPrimitiveKind::String
|
||||
}
|
||||
}
|
||||
|
||||
fn build_auto_param_pattern(key: &str, group: &str, quoted: bool) -> String {
|
||||
let escaped_key = regex::escape(key);
|
||||
if quoted {
|
||||
format!(r#"{escaped_key}="(?P<{group}>[^"]*)""#)
|
||||
} else {
|
||||
format!(r"{escaped_key}=(?P<{group}>\S+)")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
struct AutoParamCandidate {
|
||||
name: String,
|
||||
value: String,
|
||||
message_match_regex: String,
|
||||
capture_group: String,
|
||||
json_field_path: Option<String>,
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
fn auto_derived_param_count(&self) -> usize {
|
||||
self.param_declarations_range().filter(|declaration| declaration.6).count()
|
||||
}
|
||||
|
||||
pub(crate) fn auto_declare_params(&mut self, text: &str) {
|
||||
let mut total = self.auto_derived_param_count();
|
||||
if total >= MAX_AUTO_DECLARED_PARAMS {
|
||||
return;
|
||||
}
|
||||
let candidates: Vec<AutoParamCandidate> = if crate::engine::logfmt::detect_format(text) == "json" {
|
||||
crate::engine::data_value::flatten_json_object(text)
|
||||
.into_iter()
|
||||
.map(|(path, value)| AutoParamCandidate {
|
||||
name: path.clone(),
|
||||
value,
|
||||
message_match_regex: String::new(),
|
||||
capture_group: String::new(),
|
||||
json_field_path: Some(path),
|
||||
})
|
||||
.collect()
|
||||
} else {
|
||||
crate::engine::logfmt::extract_pairs(text)
|
||||
.into_iter()
|
||||
.map(|(key, value, quoted)| {
|
||||
let group = sanitize_group_name(key);
|
||||
AutoParamCandidate {
|
||||
name: key.to_string(),
|
||||
value: value.to_string(),
|
||||
message_match_regex: build_auto_param_pattern(key, &group, quoted),
|
||||
capture_group: group,
|
||||
json_field_path: None,
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
|
||||
let mut new_this_line = 0;
|
||||
for candidate in candidates {
|
||||
if new_this_line >= MAX_NEW_AUTO_PARAMS_PER_LINE || total >= MAX_AUTO_DECLARED_PARAMS {
|
||||
break;
|
||||
}
|
||||
let index_key = IndexKey::ParamDeclaration { name: candidate.name.clone() };
|
||||
if self.index.contains_key(&index_key) {
|
||||
continue;
|
||||
}
|
||||
self.index.insert(
|
||||
index_key,
|
||||
DataValue::ParamDeclaration(ParamDeclaration {
|
||||
name: candidate.name,
|
||||
message_match_regex: candidate.message_match_regex,
|
||||
capture_group: candidate.capture_group,
|
||||
json_field_path: candidate.json_field_path,
|
||||
kind: infer_kind(&candidate.value),
|
||||
elapsed_since_last: false,
|
||||
auto_derived: true,
|
||||
log_stream_gated: false,
|
||||
}),
|
||||
);
|
||||
new_this_line += 1;
|
||||
total += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn log_stream_header_transition(
|
||||
current: &Option<crate::engine::logfmt::LogStreamHeader>,
|
||||
text: &str,
|
||||
) -> Option<crate::engine::logfmt::LogStreamHeader> {
|
||||
crate::engine::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone())
|
||||
}
|
||||
|
||||
pub(crate) fn derive_log_stream_params_from_row(
|
||||
index: &mut BTreeMap<IndexKey, DataValue>,
|
||||
header: &crate::engine::logfmt::LogStreamHeader,
|
||||
text: &str,
|
||||
) {
|
||||
if !crate::engine::logfmt::is_log_stream_line(text) {
|
||||
return;
|
||||
}
|
||||
let row = header.split_row(text);
|
||||
|
||||
for (i, column) in header.columns.iter().enumerate() {
|
||||
let Some(value) = row.get(i) else { continue };
|
||||
let index_key = IndexKey::ParamDeclaration { name: column.name.clone() };
|
||||
if index.contains_key(&index_key) {
|
||||
continue;
|
||||
}
|
||||
index.insert(
|
||||
index_key,
|
||||
DataValue::ParamDeclaration(ParamDeclaration {
|
||||
name: column.name.clone(),
|
||||
message_match_regex: header.column_pattern(i, &column.name),
|
||||
capture_group: column.name.clone(),
|
||||
json_field_path: None,
|
||||
kind: infer_kind(value.trim()),
|
||||
elapsed_since_last: false,
|
||||
auto_derived: true,
|
||||
log_stream_gated: true,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) {
|
||||
index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| {
|
||||
DataValue::ParamDeclaration(ParamDeclaration {
|
||||
name: "message".to_string(),
|
||||
message_match_regex: header.message_pattern("message"),
|
||||
capture_group: "message".to_string(),
|
||||
json_field_path: None,
|
||||
kind: infer_kind(message.trim()),
|
||||
elapsed_since_last: false,
|
||||
auto_derived: true,
|
||||
log_stream_gated: true,
|
||||
})
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn auto_declare_log_stream_params(
|
||||
index: &mut BTreeMap<IndexKey, DataValue>,
|
||||
header: &mut Option<crate::engine::logfmt::LogStreamHeader>,
|
||||
text: &str,
|
||||
) {
|
||||
let is_new_header = crate::engine::logfmt::LogStreamHeader::detect(text).is_some();
|
||||
*header = log_stream_header_transition(header, text);
|
||||
if is_new_header {
|
||||
return;
|
||||
}
|
||||
if let Some(header) = header.as_ref() {
|
||||
derive_log_stream_params_from_row(index, header, text);
|
||||
}
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(crate) fn clear_all_param_declarations(&mut self) -> usize {
|
||||
let names: Vec<String> =
|
||||
self.param_declarations_range().map(|(name, ..)| name.to_string()).collect();
|
||||
let removed = names.len();
|
||||
for name in names {
|
||||
self.index.remove(&IndexKey::ParamDeclaration { name });
|
||||
}
|
||||
removed
|
||||
}
|
||||
|
||||
pub(crate) fn reextract_params_from_retained_logs(&mut self) -> (usize, usize) {
|
||||
let texts: Vec<String> = self.logs_range().map(|(_, _, text)| text.to_string()).collect();
|
||||
let logs_considered = texts.len();
|
||||
for text in &texts {
|
||||
self.auto_declare_params(text);
|
||||
}
|
||||
let reextracted = self.param_declarations_range().count();
|
||||
(reextracted, logs_considered)
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) struct ResolvedParam {
|
||||
regex: Regex,
|
||||
capture_group: String,
|
||||
json_field_path: Option<String>,
|
||||
kind: MetricPrimitiveKind,
|
||||
elapsed_since_last: bool,
|
||||
log_stream_gated: bool,
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(super) fn resolved_param_declarations(&self) -> HashMap<String, ResolvedParam> {
|
||||
self.param_declarations_range()
|
||||
.filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| {
|
||||
Regex::new(message_match_regex).ok().map(|regex| {
|
||||
(
|
||||
name.to_string(),
|
||||
ResolvedParam {
|
||||
regex,
|
||||
capture_group: capture_group.to_string(),
|
||||
json_field_path: json_field_path.map(|path| path.to_string()),
|
||||
kind,
|
||||
elapsed_since_last,
|
||||
log_stream_gated,
|
||||
},
|
||||
)
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
fn extract_param_raw(
|
||||
text: &str,
|
||||
regex: &Regex,
|
||||
capture_group: &str,
|
||||
json_field_path: Option<&str>,
|
||||
log_stream_gated: bool,
|
||||
) -> Option<String> {
|
||||
if log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) {
|
||||
return None;
|
||||
}
|
||||
match json_field_path.filter(|path| !path.is_empty()) {
|
||||
Some(field_path) => crate::engine::data_value::extract_json_field(text, field_path),
|
||||
None => regex.captures(text)?.name(capture_group).map(|m| m.as_str().to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(super) fn resolve_param_value(
|
||||
&mut self,
|
||||
metric_name: &str,
|
||||
resolved: &ResolvedParam,
|
||||
labels_so_far: &BTreeMap<String, MetricPrimitive>,
|
||||
text: &str,
|
||||
timestamp: Timestamp,
|
||||
) -> Option<MetricPrimitive> {
|
||||
if resolved.elapsed_since_last {
|
||||
let last_event_key = IndexKey::LastMetricEvent {
|
||||
metric_name: metric_name.to_string(),
|
||||
labels: labels_so_far.clone(),
|
||||
};
|
||||
let previous_timestamp = match self.index.get(&last_event_key) {
|
||||
Some(DataValue::LastEventTimestamp { timestamp: previous }) => Some(*previous),
|
||||
_ => None,
|
||||
};
|
||||
self.index.insert(last_event_key, DataValue::LastEventTimestamp { timestamp });
|
||||
return previous_timestamp
|
||||
.map(|previous| MetricPrimitive::Duration(timestamp.as_millis() - previous.as_millis()));
|
||||
}
|
||||
|
||||
let raw = extract_param_raw(
|
||||
text,
|
||||
&resolved.regex,
|
||||
&resolved.capture_group,
|
||||
resolved.json_field_path.as_deref(),
|
||||
resolved.log_stream_gated,
|
||||
)?;
|
||||
parse_metric_primitive(&raw, resolved.kind)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,549 @@
|
||||
//! Answering `EngineQuery`s: log/timeline listing (including the sort-helper-index path),
|
||||
//! stats, declaration listing, bucket-value listing, and PromQL evaluation.
|
||||
|
||||
use crate::api::dto::{
|
||||
BucketValueDto, LogEntryDto, MetricDeclarationDto, ParamDeclarationDto, PromqlSeriesDto,
|
||||
StatsDto, TimelineBucketDto,
|
||||
};
|
||||
use crate::consts::{DEFAULT_TIMELINE_BUCKET_MS, MERGE_WINDOW_MS};
|
||||
use crate::engine::data_value::{DataValue, MetricPrimitiveKind};
|
||||
use crate::engine::index_key::IndexKey;
|
||||
use crate::engine::message::{EngineQuery, EngineResponse};
|
||||
use crate::engine::metric_primitive::{MetricPrimitive, Number};
|
||||
use crate::engine::promql::{AggOp, LabelMatcher, PromqlQuery};
|
||||
use crate::engine::rate;
|
||||
#[cfg(test)]
|
||||
use crate::engine::snapshot::sort::SortHelperIndexEntry;
|
||||
use crate::engine::snapshot::sort::{log_entry_params, SortHelperAccess, SortHelperKey};
|
||||
use crate::engine::snapshot::EngineSnapshot;
|
||||
use crate::engine::timestamp::Timestamp;
|
||||
#[cfg(test)]
|
||||
use std::collections::HashMap;
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn handle_query(
|
||||
snapshot: &EngineSnapshot,
|
||||
query: EngineQuery,
|
||||
_approx_bytes_total: u64,
|
||||
_ingest_lag_high_watermark_ms: i64,
|
||||
sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
|
||||
) -> EngineResponse {
|
||||
handle_query_inner(snapshot, query, SortHelperAccess::Mutable(sort_helper_indexes))
|
||||
}
|
||||
|
||||
pub(super) fn handle_query_inner(
|
||||
snapshot: &EngineSnapshot,
|
||||
query: EngineQuery,
|
||||
sort_helper_access: SortHelperAccess<'_>,
|
||||
) -> EngineResponse {
|
||||
let index = &snapshot.index;
|
||||
let approx_bytes_total = snapshot.approx_bytes_total;
|
||||
let ingest_lag_high_watermark_ms = snapshot.ingest_lag_high_watermark_ms;
|
||||
match query {
|
||||
EngineQuery::Logs(params) => {
|
||||
let compiled_filter = match params.filter.as_deref() {
|
||||
Some(filter) => match snapshot.compile_filter_query(filter, Timestamp::now()) {
|
||||
Ok(compiled) => Some(compiled),
|
||||
Err(err) => return EngineResponse::Error(err),
|
||||
},
|
||||
None => None,
|
||||
};
|
||||
if let Some(sort_by) = params.sort_by.as_deref() {
|
||||
if !snapshot.param_declarations_range().any(|(name, ..)| name == sort_by) {
|
||||
return EngineResponse::Error(format!("unknown param '{sort_by}' for sort_by"));
|
||||
}
|
||||
}
|
||||
let limit = params.limit.unwrap_or(500).min(10_000);
|
||||
let compiled_params = snapshot.compiled_param_declarations();
|
||||
let filtered = snapshot
|
||||
.logs_range()
|
||||
.filter(|(sequence, timestamp, _)| {
|
||||
params.since_sequence.is_none_or(|since| *sequence > since)
|
||||
&& params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
|
||||
&& params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
|
||||
})
|
||||
.filter(|(sequence, timestamp, text)| {
|
||||
compiled_filter
|
||||
.as_ref()
|
||||
.is_none_or(|filter| filter.eval(*sequence, *timestamp, text))
|
||||
});
|
||||
|
||||
let entries: Vec<LogEntryDto> = match params.sort_by.as_deref() {
|
||||
None => filtered
|
||||
.take(limit)
|
||||
.map(|(sequence, timestamp, text)| LogEntryDto {
|
||||
sequence,
|
||||
timestamp_ms: timestamp.as_millis(),
|
||||
text: text.to_string(),
|
||||
params: log_entry_params(text, &compiled_params),
|
||||
})
|
||||
.collect(),
|
||||
Some(sort_by) => {
|
||||
static EMPTY_HELPER: std::sync::LazyLock<BTreeMap<SortHelperKey, Timestamp>> =
|
||||
std::sync::LazyLock::new(BTreeMap::new);
|
||||
let helper_index =
|
||||
sort_helper_access.resolve(index, sort_by).unwrap_or(&EMPTY_HELPER);
|
||||
|
||||
let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true);
|
||||
let mut entries = Vec::new();
|
||||
for (helper_key, timestamp) in helper_index.iter() {
|
||||
let sequence = match helper_key {
|
||||
SortHelperKey::WithParam(_, sequence) => *sequence,
|
||||
SortHelperKey::WithoutParam(sequence) => {
|
||||
if !sort_missing_fallback {
|
||||
continue;
|
||||
}
|
||||
*sequence
|
||||
}
|
||||
};
|
||||
if !(params.since_sequence.is_none_or(|since| sequence > since)
|
||||
&& params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
|
||||
&& params.end_ms.is_none_or(|end| timestamp.as_millis() <= end))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
let Some(DataValue::LogLiteral { text }) =
|
||||
index.get(&IndexKey::IngestedLog { sequence, timestamp: *timestamp })
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
if !compiled_filter
|
||||
.as_ref()
|
||||
.is_none_or(|filter| filter.eval(sequence, *timestamp, text))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
entries.push(LogEntryDto {
|
||||
sequence,
|
||||
timestamp_ms: timestamp.as_millis(),
|
||||
text: text.clone(),
|
||||
params: log_entry_params(text, &compiled_params),
|
||||
});
|
||||
if entries.len() >= limit {
|
||||
break;
|
||||
}
|
||||
}
|
||||
entries
|
||||
}
|
||||
};
|
||||
let entries: Vec<LogEntryDto> = entries.into_iter().take(limit).collect();
|
||||
EngineResponse::Logs(entries)
|
||||
}
|
||||
EngineQuery::Timeline(params) => {
|
||||
let compiled_filter = match params.filter.as_deref() {
|
||||
Some(filter) => match snapshot.compile_filter_query(filter, Timestamp::now()) {
|
||||
Ok(compiled) => Some(compiled),
|
||||
Err(err) => return EngineResponse::Error(err),
|
||||
},
|
||||
None => None,
|
||||
};
|
||||
let bucket_ms = params.bucket_ms.unwrap_or(DEFAULT_TIMELINE_BUCKET_MS).max(1);
|
||||
let mut buckets: BTreeMap<i64, u64> = BTreeMap::new();
|
||||
for (sequence, timestamp, text) in snapshot.logs_range().filter(|(_, timestamp, _)| {
|
||||
params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
|
||||
&& params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
|
||||
}) {
|
||||
if compiled_filter
|
||||
.as_ref()
|
||||
.is_some_and(|filter| !filter.eval(sequence, timestamp, text))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
let bucket_start = (timestamp.as_millis() / bucket_ms) * bucket_ms;
|
||||
*buckets.entry(bucket_start).or_insert(0) += 1;
|
||||
}
|
||||
let out = buckets
|
||||
.into_iter()
|
||||
.map(|(start_ms, count)| TimelineBucketDto { start_ms, count })
|
||||
.collect();
|
||||
EngineResponse::Timeline(out)
|
||||
}
|
||||
EngineQuery::Stats => {
|
||||
let mut log_count = 0u64;
|
||||
let mut first_sequence = None;
|
||||
let mut last_sequence = None;
|
||||
let mut first_timestamp_ms = None;
|
||||
let mut last_timestamp_ms = None;
|
||||
for (sequence, timestamp, _) in snapshot.logs_range() {
|
||||
log_count += 1;
|
||||
first_sequence.get_or_insert(sequence);
|
||||
last_sequence = Some(sequence);
|
||||
first_timestamp_ms.get_or_insert(timestamp.as_millis());
|
||||
last_timestamp_ms = Some(timestamp.as_millis());
|
||||
}
|
||||
EngineResponse::Stats(StatsDto {
|
||||
log_count,
|
||||
first_sequence,
|
||||
last_sequence,
|
||||
first_timestamp_ms,
|
||||
last_timestamp_ms,
|
||||
approx_bytes_total,
|
||||
ingest_lag_high_watermark_ms,
|
||||
})
|
||||
}
|
||||
EngineQuery::ParamDeclarations => {
|
||||
let declarations = snapshot.param_declarations_range()
|
||||
.map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, auto_derived, _)| {
|
||||
ParamDeclarationDto {
|
||||
name: name.to_string(),
|
||||
message_match_regex: message_match_regex.to_string(),
|
||||
capture_group: capture_group.to_string(),
|
||||
json_field_path: json_field_path.map(|path| path.to_string()),
|
||||
kind,
|
||||
elapsed_since_last,
|
||||
auto_derived,
|
||||
reserved: false,
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
EngineResponse::ParamDeclarations(declarations)
|
||||
}
|
||||
EngineQuery::MetricDeclarations => {
|
||||
let declarations = snapshot.metric_declarations_range()
|
||||
.map(|(name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels)| {
|
||||
MetricDeclarationDto {
|
||||
name: name.to_string(),
|
||||
message_match_regex: message_match_regex.to_string(),
|
||||
label_params: label_params.to_vec(),
|
||||
value_param: value_param.map(|s| s.to_string()),
|
||||
matched_labels: matched_labels.cloned(),
|
||||
unmatched_labels: unmatched_labels.cloned(),
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
EngineResponse::MetricDeclarations(declarations)
|
||||
}
|
||||
EngineQuery::BucketValues {
|
||||
metric_name,
|
||||
labels_filter,
|
||||
start_ms,
|
||||
end_ms,
|
||||
since_sequence,
|
||||
limit,
|
||||
} => {
|
||||
let limit = limit.unwrap_or(500).min(10_000);
|
||||
let values: Vec<BucketValueDto> = snapshot.bucket_values(&metric_name, None)
|
||||
.filter(|(labels, _)| {
|
||||
labels_filter.as_ref().is_none_or(|filter| {
|
||||
filter.iter().all(|(name, expected)| {
|
||||
labels
|
||||
.get(name)
|
||||
.is_some_and(|value| &value.to_string() == expected)
|
||||
})
|
||||
})
|
||||
})
|
||||
.filter(|(_, (sequence, timestamp, _))| {
|
||||
since_sequence.is_none_or(|since| *sequence > since)
|
||||
&& start_ms.is_none_or(|start| timestamp.as_millis() >= start)
|
||||
&& end_ms.is_none_or(|end| timestamp.as_millis() <= end)
|
||||
})
|
||||
.take(limit)
|
||||
.map(|(labels, (sequence, timestamp, value))| BucketValueDto {
|
||||
labels: labels.clone(),
|
||||
sequence: *sequence,
|
||||
timestamp_ms: timestamp.as_millis(),
|
||||
value: value.clone(),
|
||||
})
|
||||
.collect();
|
||||
EngineResponse::BucketValues(values)
|
||||
}
|
||||
EngineQuery::Promql { query, start_ms, end_ms } => {
|
||||
let response = PromqlQuery::parse(&query)
|
||||
.and_then(|parsed| snapshot.evaluate_promql(&parsed, start_ms, end_ms));
|
||||
EngineResponse::Promql(response)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn coerce_to_f64(value: &MetricPrimitive) -> f64 {
|
||||
match value {
|
||||
MetricPrimitive::Number(n) => match n {
|
||||
Number::Int(v) => *v as f64,
|
||||
Number::Float(v) => *v,
|
||||
},
|
||||
MetricPrimitive::Timestamp(ts) => ts.as_millis() as f64,
|
||||
MetricPrimitive::Duration(ms) => *ms as f64,
|
||||
MetricPrimitive::String(_) => 1.0,
|
||||
MetricPrimitive::Boolean(b) => {
|
||||
if *b {
|
||||
1.0
|
||||
} else {
|
||||
0.0
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn matches_label_matchers(
|
||||
labels: &BTreeMap<String, MetricPrimitive>,
|
||||
matchers: &[LabelMatcher],
|
||||
) -> bool {
|
||||
matchers.iter().all(|matcher| {
|
||||
labels.get(&matcher.name).is_some_and(|value| value.to_string() == matcher.value)
|
||||
})
|
||||
}
|
||||
|
||||
fn project_labels(
|
||||
labels: &BTreeMap<String, String>,
|
||||
by_labels: &[String],
|
||||
) -> BTreeMap<String, String> {
|
||||
by_labels
|
||||
.iter()
|
||||
.filter_map(|name| labels.get(name).map(|value| (name.clone(), value.clone())))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn windowed_rate_series(
|
||||
points: &[(i64, f64)],
|
||||
range_ms: i64,
|
||||
start_ms: Option<i64>,
|
||||
end_ms: Option<i64>,
|
||||
) -> Vec<(i64, f64)> {
|
||||
let ts_points: Vec<(Timestamp, f64)> =
|
||||
points.iter().map(|(ms, value)| (Timestamp::from_millis(*ms), *value)).collect();
|
||||
let mut series: Vec<(i64, f64)> = rate::windowed_rate_series(&ts_points, range_ms)
|
||||
.into_iter()
|
||||
.map(|(ts, value)| (ts.as_millis(), value))
|
||||
.collect();
|
||||
|
||||
if let Some(end) = end_ms {
|
||||
if series.last().is_none_or(|(ts, _)| *ts < end) {
|
||||
let value = rate::rate_at(&ts_points, Timestamp::from_millis(end), range_ms);
|
||||
series.push((end, value));
|
||||
}
|
||||
}
|
||||
if let Some(start) = start_ms {
|
||||
if series.first().is_none_or(|(ts, _)| *ts > start) {
|
||||
let value = rate::rate_at(&ts_points, Timestamp::from_millis(start), range_ms);
|
||||
series.insert(0, (start, value));
|
||||
}
|
||||
}
|
||||
series
|
||||
}
|
||||
|
||||
fn merge_series_windowed(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> {
|
||||
if sub_series.len() <= 1 {
|
||||
return sub_series.first().cloned().unwrap_or_default();
|
||||
}
|
||||
let mut grouped: BTreeMap<i64, Vec<f64>> = BTreeMap::new();
|
||||
for points in sub_series {
|
||||
for (timestamp_ms, value) in points {
|
||||
let window_start = timestamp_ms.div_euclid(MERGE_WINDOW_MS) * MERGE_WINDOW_MS;
|
||||
grouped.entry(window_start).or_default().push(*value);
|
||||
}
|
||||
}
|
||||
grouped
|
||||
.into_iter()
|
||||
.map(|(timestamp_ms, values)| {
|
||||
let sum: f64 = values.iter().sum();
|
||||
let value = if is_avg { sum / values.len() as f64 } else { sum };
|
||||
(timestamp_ms, value)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> {
|
||||
let mut all_timestamps: std::collections::BTreeSet<i64> = std::collections::BTreeSet::new();
|
||||
for points in sub_series {
|
||||
all_timestamps.extend(points.iter().map(|(ts, _)| *ts));
|
||||
}
|
||||
let mut cursors = vec![0usize; sub_series.len()];
|
||||
let mut last_values = vec![0.0f64; sub_series.len()];
|
||||
all_timestamps
|
||||
.into_iter()
|
||||
.map(|ts| {
|
||||
for (i, points) in sub_series.iter().enumerate() {
|
||||
while cursors[i] < points.len() && points[cursors[i]].0 <= ts {
|
||||
last_values[i] = points[cursors[i]].1;
|
||||
cursors[i] += 1;
|
||||
}
|
||||
}
|
||||
let sum: f64 = last_values.iter().sum();
|
||||
let value = if is_avg { sum / last_values.len() as f64 } else { sum };
|
||||
(ts, value)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn ottyl_reserved_numeric_value_fn(metric_name: &str) -> Option<fn(u64, i64) -> f64> {
|
||||
match metric_name {
|
||||
"ottyl_seq" => Some(|sequence, _timestamp_ms| sequence as f64),
|
||||
"ottyl_ts" => Some(|_sequence, timestamp_ms| timestamp_ms as f64),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(crate) fn evaluate_promql(
|
||||
&self,
|
||||
parsed: &PromqlQuery,
|
||||
start_ms: Option<i64>,
|
||||
end_ms: Option<i64>,
|
||||
) -> Result<Vec<PromqlSeriesDto>, String> {
|
||||
if parsed.metric_name == "ottyl_fmt" {
|
||||
return Err(
|
||||
"'ottyl_fmt' is a categorical value (json/logfmt/none), not a number, so it can't be \
|
||||
used in a dashboard PromQL query; filter on it in the search bar instead, e.g. \
|
||||
ottyl_fmt==json"
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
|
||||
if ottyl_reserved_numeric_value_fn(&parsed.metric_name).is_none()
|
||||
&& !self.index.contains_key(&IndexKey::MetricDeclaration { name: parsed.metric_name.clone() })
|
||||
&& self.bucket_values(&parsed.metric_name, None).next().is_none()
|
||||
{
|
||||
return Err(format!(
|
||||
"unknown metric \"{}\": no declaration or recorded data for it (was it deleted?)",
|
||||
parsed.metric_name
|
||||
));
|
||||
}
|
||||
if matches!(parsed.agg, Some(AggOp::Sum) | Some(AggOp::Avg))
|
||||
&& !parsed.inner_rate
|
||||
&& self.declared_value_kind(&parsed.metric_name) == Some(MetricPrimitiveKind::String)
|
||||
{
|
||||
return Err(format!(
|
||||
"cannot sum/avg a string-typed metric \"{}\"; use count() or a numeric-typed metric",
|
||||
parsed.metric_name
|
||||
));
|
||||
}
|
||||
|
||||
let mut series_points: BTreeMap<BTreeMap<String, String>, Vec<(i64, f64)>> = BTreeMap::new();
|
||||
if let Some(reserved_value) = ottyl_reserved_numeric_value_fn(&parsed.metric_name) {
|
||||
let no_labels: BTreeMap<String, MetricPrimitive> = BTreeMap::new();
|
||||
if matches_label_matchers(&no_labels, &parsed.matchers) {
|
||||
for (sequence, timestamp, _text) in self.logs_range() {
|
||||
let timestamp_ms = timestamp.as_millis();
|
||||
if start_ms.is_none_or(|start| timestamp_ms >= start)
|
||||
&& end_ms.is_none_or(|end| timestamp_ms <= end)
|
||||
{
|
||||
series_points
|
||||
.entry(BTreeMap::new())
|
||||
.or_default()
|
||||
.push((timestamp_ms, reserved_value(sequence, timestamp_ms)));
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
for (labels, (_, timestamp, value)) in self.bucket_values(&parsed.metric_name, None)
|
||||
.filter(|(labels, _)| matches_label_matchers(labels, &parsed.matchers))
|
||||
.filter(|(_, (_, timestamp, _))| {
|
||||
start_ms.is_none_or(|start| timestamp.as_millis() >= start)
|
||||
&& end_ms.is_none_or(|end| timestamp.as_millis() <= end)
|
||||
})
|
||||
{
|
||||
let display_labels: BTreeMap<String, String> =
|
||||
labels.iter().map(|(name, value)| (name.clone(), value.to_string())).collect();
|
||||
series_points
|
||||
.entry(display_labels)
|
||||
.or_default()
|
||||
.push((timestamp.as_millis(), coerce_to_f64(value)));
|
||||
}
|
||||
}
|
||||
for points in series_points.values_mut() {
|
||||
points.sort_by_key(|(timestamp_ms, _)| *timestamp_ms);
|
||||
}
|
||||
|
||||
Ok(match parsed.agg {
|
||||
None => series_points
|
||||
.into_iter()
|
||||
.map(|(labels, points)| PromqlSeriesDto { labels, points })
|
||||
.collect(),
|
||||
Some(AggOp::Count) => {
|
||||
let mut groups: BTreeMap<BTreeMap<String, String>, Vec<Vec<(i64, f64)>>> = BTreeMap::new();
|
||||
for (labels, points) in series_points {
|
||||
let count_points: Vec<(i64, f64)> =
|
||||
points.iter().map(|(timestamp_ms, _)| (*timestamp_ms, 1.0)).collect();
|
||||
let group_key = project_labels(&labels, &parsed.by_labels);
|
||||
groups.entry(group_key).or_default().push(count_points);
|
||||
}
|
||||
groups
|
||||
.into_iter()
|
||||
.map(|(group_labels, sub_series)| {
|
||||
let points = merge_series_windowed(&sub_series, false);
|
||||
PromqlSeriesDto { labels: group_labels, points }
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
Some(AggOp::Sum) | Some(AggOp::Avg) => {
|
||||
let is_avg = parsed.agg == Some(AggOp::Avg);
|
||||
let rate_range_ms = parsed
|
||||
.inner_rate
|
||||
.then(|| parsed.range.expect("inner rate() always carries a range").as_millis() as i64);
|
||||
|
||||
let mut groups: BTreeMap<BTreeMap<String, String>, Vec<Vec<(i64, f64)>>> = BTreeMap::new();
|
||||
for (labels, points) in series_points {
|
||||
let points = match rate_range_ms {
|
||||
Some(range_ms) => windowed_rate_series(&points, range_ms, start_ms, end_ms),
|
||||
None => points,
|
||||
};
|
||||
let group_key = project_labels(&labels, &parsed.by_labels);
|
||||
groups.entry(group_key).or_default().push(points);
|
||||
}
|
||||
|
||||
groups
|
||||
.into_iter()
|
||||
.map(|(group_labels, sub_series)| {
|
||||
let points = if rate_range_ms.is_some() {
|
||||
merge_series_locf(&sub_series, is_avg)
|
||||
} else {
|
||||
merge_series_windowed(&sub_series, is_avg)
|
||||
};
|
||||
PromqlSeriesDto { labels: group_labels, points }
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
Some(AggOp::Rate) => {
|
||||
let range_ms = parsed.range.expect("rate() always carries a range").as_millis() as i64;
|
||||
series_points
|
||||
.into_iter()
|
||||
.map(|(labels, points)| PromqlSeriesDto {
|
||||
labels,
|
||||
points: windowed_rate_series(&points, range_ms, start_ms, end_ms),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
Some(AggOp::Gauge) => {
|
||||
series_points
|
||||
.into_iter()
|
||||
.filter_map(|(labels, points)| {
|
||||
let last = *points.last()?;
|
||||
Some(PromqlSeriesDto { labels, points: vec![last] })
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
Some(AggOp::Histogram) => {
|
||||
let bucket_width =
|
||||
parsed.bucket_width.expect("histogram() always carries a bucket width");
|
||||
let last_ts = series_points
|
||||
.values()
|
||||
.flat_map(|points| points.iter().map(|(ts, _)| *ts))
|
||||
.max()
|
||||
.unwrap_or(0);
|
||||
let mut groups: BTreeMap<BTreeMap<String, String>, Vec<&(i64, f64)>> = BTreeMap::new();
|
||||
for (labels, points) in &series_points {
|
||||
let group_key = project_labels(labels, &parsed.by_labels);
|
||||
groups.entry(group_key).or_default().extend(points.iter());
|
||||
}
|
||||
|
||||
groups
|
||||
.into_iter()
|
||||
.flat_map(|(group_labels, points)| {
|
||||
let mut counts: BTreeMap<i64, u64> = BTreeMap::new();
|
||||
for (_, value) in points {
|
||||
let bucket_index = (value / bucket_width).floor() as i64;
|
||||
*counts.entry(bucket_index).or_insert(0) += 1;
|
||||
}
|
||||
counts.into_iter().map(move |(bucket_index, count)| {
|
||||
let upper_bound = (bucket_index + 1) as f64 * bucket_width;
|
||||
let mut labels = group_labels.clone();
|
||||
labels.insert("le".to_string(), upper_bound.to_string());
|
||||
PromqlSeriesDto { labels, points: vec![(last_ts, count as f64)] }
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
//! Sort-helper indexes: a per-param `sort_by` lets the logs view return results ordered by a
|
||||
//! declared param's value instead of ingest order. Building that ordering from scratch is
|
||||
//! O(n log n) over every log, so it's memoized in `EngineSnapshot::sort_helper_indexes` and
|
||||
//! kept incrementally up to date on ingest, rather than rebuilt per query.
|
||||
|
||||
use crate::consts::SORT_HELPER_INDEX_PURGE_AFTER;
|
||||
use crate::engine::data_value::DataValue;
|
||||
use crate::engine::index_key::IndexKey;
|
||||
use crate::engine::data_value::MetricPrimitiveKind;
|
||||
use crate::engine::metric_primitive::MetricPrimitive;
|
||||
use crate::engine::snapshot::metrics::parse_metric_primitive;
|
||||
use crate::engine::snapshot::EngineSnapshot;
|
||||
use crate::engine::timestamp::Timestamp;
|
||||
use regex::Regex;
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub(crate) enum SortHelperKey {
|
||||
WithParam(MetricPrimitive, u64),
|
||||
WithoutParam(u64),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub(crate) struct SortHelperIndexEntry {
|
||||
pub(crate) index: BTreeMap<SortHelperKey, Timestamp>,
|
||||
pub(crate) last_accessed: std::time::Instant,
|
||||
}
|
||||
|
||||
pub(super) type CompiledParamDeclaration = (String, Regex, String, Option<String>, MetricPrimitiveKind, bool);
|
||||
|
||||
fn derive_single_param(
|
||||
text: &str,
|
||||
declarations: &[CompiledParamDeclaration],
|
||||
param_name: &str,
|
||||
) -> Option<MetricPrimitive> {
|
||||
let (_, regex, capture_group, json_field_path, kind, log_stream_gated) =
|
||||
declarations.iter().find(|(name, ..)| name == param_name)?;
|
||||
if *log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) {
|
||||
return None;
|
||||
}
|
||||
if *kind == MetricPrimitiveKind::Boolean {
|
||||
return Some(MetricPrimitive::Boolean(regex.is_match(text)));
|
||||
}
|
||||
let captures = regex.captures(text)?;
|
||||
let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) {
|
||||
Some(field_path) => crate::engine::data_value::extract_json_field(text, field_path)?,
|
||||
None => captures.name(capture_group)?.as_str().to_string(),
|
||||
};
|
||||
parse_metric_primitive(&raw, *kind)
|
||||
}
|
||||
|
||||
impl EngineSnapshot {
|
||||
pub(super) fn build_sort_helper_index(&self, sort_by: &str) -> BTreeMap<SortHelperKey, Timestamp> {
|
||||
let compiled_params = self.compiled_param_declarations();
|
||||
self.logs_range()
|
||||
.map(|(sequence, timestamp, text)| {
|
||||
let key = match derive_single_param(text, &compiled_params, sort_by) {
|
||||
Some(value) => SortHelperKey::WithParam(value, sequence),
|
||||
None => SortHelperKey::WithoutParam(sequence),
|
||||
};
|
||||
(key, timestamp)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub(crate) fn update_sort_helper_indexes_on_ingest(&mut self, sequence: u64, timestamp: Timestamp, text: &str) {
|
||||
if self.sort_helper_indexes.is_empty() {
|
||||
return;
|
||||
}
|
||||
let compiled_params = self.compiled_param_declarations();
|
||||
for (sort_by, entry) in self.sort_helper_indexes.iter_mut() {
|
||||
let key = match derive_single_param(text, &compiled_params, sort_by) {
|
||||
Some(value) => SortHelperKey::WithParam(value, sequence),
|
||||
None => SortHelperKey::WithoutParam(sequence),
|
||||
};
|
||||
entry.index.insert(key, timestamp);
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn purge_stale_sort_helper_indexes(&mut self) {
|
||||
self.sort_helper_indexes
|
||||
.retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER);
|
||||
}
|
||||
|
||||
pub(crate) fn ensure_sort_helper_index_built(&mut self, sort_by: &str) {
|
||||
if !self.sort_helper_indexes.contains_key(sort_by) {
|
||||
let built = self.build_sort_helper_index(sort_by);
|
||||
self.sort_helper_indexes.insert(
|
||||
sort_by.to_string(),
|
||||
SortHelperIndexEntry { index: built, last_accessed: std::time::Instant::now() },
|
||||
);
|
||||
}
|
||||
let entry = self.sort_helper_indexes.get_mut(sort_by).expect("just inserted above if missing");
|
||||
entry.last_accessed = std::time::Instant::now();
|
||||
}
|
||||
|
||||
pub(crate) fn compiled_param_declarations(&self) -> Vec<CompiledParamDeclaration> {
|
||||
self.param_declarations_range()
|
||||
.filter(|(_, _, _, _, _, elapsed_since_last, _, _)| !elapsed_since_last)
|
||||
.filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _, _, log_stream_gated)| {
|
||||
Regex::new(message_match_regex).ok().map(|re| {
|
||||
(
|
||||
name.to_string(),
|
||||
re,
|
||||
capture_group.to_string(),
|
||||
json_field_path.map(|path| path.to_string()),
|
||||
kind,
|
||||
log_stream_gated,
|
||||
)
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn derive_log_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap<String, MetricPrimitive> {
|
||||
let mut params = BTreeMap::new();
|
||||
for (name, regex, capture_group, json_field_path, kind, log_stream_gated) in declarations {
|
||||
if *log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) {
|
||||
continue;
|
||||
}
|
||||
if *kind == MetricPrimitiveKind::Boolean {
|
||||
params.insert(name.clone(), MetricPrimitive::Boolean(regex.is_match(text)));
|
||||
continue;
|
||||
}
|
||||
let Some(captures) = regex.captures(text) else {
|
||||
continue;
|
||||
};
|
||||
let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) {
|
||||
Some(field_path) => match crate::engine::data_value::extract_json_field(text, field_path) {
|
||||
Some(value) => value,
|
||||
None => continue,
|
||||
},
|
||||
None => match captures.name(capture_group) {
|
||||
Some(m) => m.as_str().to_string(),
|
||||
None => continue,
|
||||
},
|
||||
};
|
||||
let Some(value) = parse_metric_primitive(&raw, *kind) else {
|
||||
continue;
|
||||
};
|
||||
params.insert(name.clone(), value);
|
||||
}
|
||||
params
|
||||
}
|
||||
|
||||
pub(super) fn log_entry_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap<String, MetricPrimitive> {
|
||||
let mut params = derive_log_params(text, declarations);
|
||||
params.insert("ottyl_fmt".to_string(), MetricPrimitive::String(crate::engine::logfmt::detect_format(text).to_string()));
|
||||
params
|
||||
}
|
||||
|
||||
pub(super) enum SortHelperAccess<'a> {
|
||||
#[cfg(test)]
|
||||
Mutable(&'a mut HashMap<String, SortHelperIndexEntry>),
|
||||
Immutable(&'a HashMap<String, SortHelperIndexEntry>),
|
||||
}
|
||||
|
||||
impl<'a> SortHelperAccess<'a> {
|
||||
#[cfg_attr(not(test), allow(unused_variables))]
|
||||
pub(super) fn resolve(
|
||||
self,
|
||||
index: &BTreeMap<IndexKey, DataValue>,
|
||||
sort_by: &str,
|
||||
) -> Option<&'a BTreeMap<SortHelperKey, Timestamp>> {
|
||||
match self {
|
||||
#[cfg(test)]
|
||||
SortHelperAccess::Mutable(map) => {
|
||||
if !map.contains_key(sort_by) {
|
||||
let built = EngineSnapshot::from_index(index.clone()).build_sort_helper_index(sort_by);
|
||||
map.insert(
|
||||
sort_by.to_string(),
|
||||
SortHelperIndexEntry { index: built, last_accessed: std::time::Instant::now() },
|
||||
);
|
||||
}
|
||||
let entry = map.get_mut(sort_by).expect("just inserted above if missing");
|
||||
entry.last_accessed = std::time::Instant::now();
|
||||
map.get(sort_by).map(|entry| &entry.index)
|
||||
}
|
||||
SortHelperAccess::Immutable(map) => map.get(sort_by).map(|entry| &entry.index),
|
||||
}
|
||||
}
|
||||
}
|
||||
+2292
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user