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

Consolidate EngineSnapshot's impl blocks into snapshot/mod.rs

Browse Source 
They were scattered across filter/metrics/params/query/sort.rs by an
earlier file-splitting pass. Files can be split as much as needed, but
a single type's impl should never be split across files — so all 15
impl EngineSnapshot blocks now live in mod.rs, while each submodule
keeps its own types, free helper functions, and tests.

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 00:43:38 +01:00
co-authored by Claude Sonnet 5
parent 7df6026a21
commit 1b2f690278
6 changed files with 944 additions and 940 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/engine/snapshot/filter.rs
+5 -88
View File
@@ -3,11 +3,8 @@
//! metric-bucket comparisons, and ad-hoc JSON field comparisons. //! metric-bucket comparisons, and ad-hoc JSON field comparisons.
use crate::engine::filter_datetime; use crate::engine::filter_datetime;
use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; use crate::engine::filter_parser::{CompareOp, FilterValue};
use crate::engine::index_key::IndexKey;
use crate::engine::data_value::DataValue;
use crate::engine::metric_primitive::MetricPrimitive; use crate::engine::metric_primitive::MetricPrimitive;
use crate::engine::snapshot::EngineSnapshot;
use crate::engine::timestamp::Timestamp; use crate::engine::timestamp::Timestamp;
use regex::Regex; use regex::Regex;
use std::cmp::Ordering; use std::cmp::Ordering;
@@ -24,88 +21,8 @@ pub(super) enum CompiledFilter {
Or(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 { impl CompiledFilter {
fn from_seq_comparison(op: CompareOp, value: &FilterValue) -> Result<Self, String> { pub(super) fn from_seq_comparison(op: CompareOp, value: &FilterValue) -> Result<Self, String> {
let target = match value { let target = match value {
FilterValue::Number(n) => *n, FilterValue::Number(n) => *n,
FilterValue::Raw(raw) => raw.parse::<f64>().map_err(|_| { FilterValue::Raw(raw) => raw.parse::<f64>().map_err(|_| {
@@ -115,7 +32,7 @@ impl CompiledFilter {
Ok(CompiledFilter::Sequence { op, target }) Ok(CompiledFilter::Sequence { op, target })
} }
fn from_ts_comparison(op: CompareOp, value: &FilterValue, now: Timestamp) -> Result<Self, String> { pub(super) fn from_ts_comparison(op: CompareOp, value: &FilterValue, now: Timestamp) -> Result<Self, String> {
let target_ms = match value { let target_ms = match value {
FilterValue::Number(n) => *n as i64, FilterValue::Number(n) => *n as i64,
FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, now) FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, now)
@@ -125,7 +42,7 @@ impl CompiledFilter {
Ok(CompiledFilter::Timestamp { op, target_ms }) Ok(CompiledFilter::Timestamp { op, target_ms })
} }
fn from_fmt_comparison(op: CompareOp, value: &FilterValue) -> Self { pub(super) fn from_fmt_comparison(op: CompareOp, value: &FilterValue) -> Self {
let target = match value { let target = match value {
FilterValue::Number(n) => n.to_string(), FilterValue::Number(n) => n.to_string(),
FilterValue::Raw(raw) => raw.clone(), FilterValue::Raw(raw) => raw.clone(),
@@ -135,7 +52,7 @@ impl CompiledFilter {
} }
impl MetricPrimitive { impl MetricPrimitive {
fn compare(&self, op: CompareOp, value: &FilterValue, now: Timestamp) -> Result<bool, String> { pub(super) fn compare(&self, op: CompareOp, value: &FilterValue, now: Timestamp) -> Result<bool, String> {
let ordering = match (self, value) { let ordering = match (self, value) {
(MetricPrimitive::Number(n), FilterValue::Number(v)) => n.as_f64().partial_cmp(v), (MetricPrimitive::Number(n), FilterValue::Number(v)) => n.as_f64().partial_cmp(v),
(MetricPrimitive::Number(n), FilterValue::Raw(raw)) => { (MetricPrimitive::Number(n), FilterValue::Raw(raw)) => {
src/engine/snapshot/metrics.rs
+1 -317
View File
@@ -2,133 +2,9 @@
//! ingested lines to populate metric buckets, and backfilling a newly (re)declared metric //! ingested lines to populate metric buckets, and backfilling a newly (re)declared metric
//! against already-ingested logs. //! against already-ingested logs.
use crate::engine::data_value::{DataValue, MetricPrimitiveKind}; use crate::engine::data_value::MetricPrimitiveKind;
use crate::engine::index_key::IndexKey;
use crate::engine::metric_primitive::{MetricPrimitive, Number}; 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::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)?;
self.metric_regex_cache.insert(request.name.clone(), re.clone());
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.metric_regex_cache.remove(existing_name);
}
self.metric_regex_cache.insert(request.name.clone(), re.clone());
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.metric_regex_cache.remove(name);
self.clear_metric_data(name);
existed
}
}
pub(super) fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Option<MetricPrimitive> { pub(super) fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Option<MetricPrimitive> {
match kind { match kind {
@@ -149,195 +25,3 @@ pub(super) fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Op
} }
} }
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,
Vec<String>,
Option<String>,
Option<BTreeMap<String, String>>,
Option<BTreeMap<String, String>>,
);
let metrics: Vec<SeededMetric> = self
.metric_declarations_range()
.map(|(name, _, label_params, value_param, matched_labels, unmatched_labels)| {
(
name.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, label_params, value_param, matched_labels, unmatched_labels) in metrics {
if let Some(re) = self.metric_regex_cache.get(&name).cloned() {
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
{
// IndexKey::Bucket sorts by (metric_name, labels), so bound the range to just this
// metric's buckets instead of ranging MetricBucketSplit..End over every declared
// metric. '\u{10FFFF}' is the maximum Unicode scalar value, so `metric_name` followed
// by it sorts after every real key for this metric (any labels) and before any key
// for a lexicographically greater metric name.
let lower = IndexKey::Bucket { metric_name: metric_name.to_string(), labels: BTreeMap::new() };
let upper = IndexKey::Bucket {
metric_name: format!("{metric_name}\u{10FFFF}"),
labels: BTreeMap::new(),
};
self.index
.range(lower..upper)
.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,
}
}
}
src/engine/snapshot/mod.rs
+907 -5
View File
@@ -1,8 +1,9 @@
//! `EngineSnapshot` is the actor's live state: ingested log lines, declared params/metrics, //! `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 //! 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 //! 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 //! `super::handle`. The submodules below hold everything that isn't part of `EngineSnapshot`'s
//! the struct itself plus the accessors every submodule builds on. //! own impl (types, free functions, tests) split out by concern; every `impl EngineSnapshot`
//! block itself lives here, in one file, rather than being scattered across those submodules.
pub(crate) mod filter; pub(crate) mod filter;
pub(crate) mod metrics; pub(crate) mod metrics;
@@ -12,14 +13,29 @@ pub(crate) mod sort;
pub(crate) use filter::eval_json_comparison; pub(crate) use filter::eval_json_comparison;
use crate::api::dto::PromqlSeriesDto;
use crate::consts::{
MAX_AUTO_DECLARED_PARAMS, MAX_NEW_AUTO_PARAMS_PER_LINE, OTTYL_RESERVED_PARAM_NAMES,
SORT_HELPER_INDEX_PURGE_AFTER,
};
use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration}; use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration};
use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
use crate::engine::index_key::IndexKey; use crate::engine::index_key::IndexKey;
use crate::engine::message::{EngineQuery, EngineResponse}; use crate::engine::message::{EngineQuery, EngineResponse};
use crate::engine::metric_primitive::{MetricPrimitive, Number};
use crate::engine::promql::{AggOp, PromqlQuery};
use crate::engine::timestamp::Timestamp; use crate::engine::timestamp::Timestamp;
use query::handle_query_inner; use crate::engine::{DeclareMetricRequest, DeclareParamRequest};
use filter::CompiledFilter;
use metrics::parse_metric_primitive;
use params::{build_auto_param_pattern, extract_param_raw, infer_kind, sanitize_group_name, AutoParamCandidate, ResolvedParam};
use query::{
coerce_to_f64, handle_query_inner, matches_label_matchers, merge_series_locf,
merge_series_windowed, ottyl_reserved_numeric_value_fn, project_labels, windowed_rate_series,
};
use regex::Regex; use regex::Regex;
use sort::{SortHelperAccess, SortHelperIndexEntry}; use sort::{derive_single_param, CompiledParamDeclaration, SortHelperAccess, SortHelperIndexEntry, SortHelperKey};
use std::collections::{BTreeMap, HashMap}; use std::collections::{BTreeMap, HashMap, HashSet};
#[derive(Debug, Clone, Default)] #[derive(Debug, Clone, Default)]
pub(crate) struct EngineSnapshot { pub(crate) struct EngineSnapshot {
@@ -216,3 +232,889 @@ impl EngineSnapshot {
self.sort_helper_indexes.contains_key(sort_by) self.sort_helper_indexes.contains_key(sort_by)
} }
} }
impl EngineSnapshot {
pub(in crate::engine::snapshot) 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 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)?;
self.metric_regex_cache.insert(request.name.clone(), re.clone());
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.metric_regex_cache.remove(existing_name);
}
self.metric_regex_cache.insert(request.name.clone(), re.clone());
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.metric_regex_cache.remove(name);
self.clear_metric_data(name);
existed
}
}
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,
Vec<String>,
Option<String>,
Option<BTreeMap<String, String>>,
Option<BTreeMap<String, String>>,
);
let metrics: Vec<SeededMetric> = self
.metric_declarations_range()
.map(|(name, _, label_params, value_param, matched_labels, unmatched_labels)| {
(
name.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, label_params, value_param, matched_labels, unmatched_labels) in metrics {
if let Some(re) = self.metric_regex_cache.get(&name).cloned() {
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
{
// IndexKey::Bucket sorts by (metric_name, labels), so bound the range to just this
// metric's buckets instead of ranging MetricBucketSplit..End over every declared
// metric. '\u{10FFFF}' is the maximum Unicode scalar value, so `metric_name` followed
// by it sorts after every real key for this metric (any labels) and before any key
// for a lexicographically greater metric name.
let lower = IndexKey::Bucket { metric_name: metric_name.to_string(), labels: BTreeMap::new() };
let upper = IndexKey::Bucket {
metric_name: format!("{metric_name}\u{10FFFF}"),
labels: BTreeMap::new(),
};
self.index
.range(lower..upper)
.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,
}
}
}
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) {
if let Ok(re) = Regex::new(&request.message_match_regex) {
self.param_regex_cache.insert(request.name.clone(), re);
}
self.index.insert(
IndexKey::ParamDeclaration { name: request.name.clone() },
DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }),
);
}
}
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;
}
if let Ok(re) = Regex::new(&candidate.message_match_regex) {
self.param_regex_cache.insert(candidate.name.clone(), re);
}
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;
}
}
}
impl EngineSnapshot {
pub(crate) fn derive_log_stream_params_from_row(
&mut self,
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 self.index.contains_key(&index_key) {
continue;
}
let message_match_regex = header.column_pattern(i, &column.name);
if let Ok(re) = Regex::new(&message_match_regex) {
self.param_regex_cache.insert(column.name.clone(), re);
}
self.index.insert(
index_key,
DataValue::ParamDeclaration(ParamDeclaration {
name: column.name.clone(),
message_match_regex,
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()) {
let index_key = IndexKey::ParamDeclaration { name: "message".to_string() };
if !self.index.contains_key(&index_key) {
let message_match_regex = header.message_pattern("message");
if let Ok(re) = Regex::new(&message_match_regex) {
self.param_regex_cache.insert("message".to_string(), re);
}
self.index.insert(
index_key,
DataValue::ParamDeclaration(ParamDeclaration {
name: "message".to_string(),
message_match_regex,
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,
}),
);
}
}
}
}
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: name.clone() });
self.param_regex_cache.remove(&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)
}
}
impl EngineSnapshot {
pub(in crate::engine::snapshot) fn resolved_param_declarations(&self) -> HashMap<String, ResolvedParam> {
self.param_declarations_range()
.filter_map(|(name, _, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| {
self.param_regex_cache.get(name).map(|regex| {
(
name.to_string(),
ResolvedParam {
regex: regex.clone(),
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()
}
}
impl EngineSnapshot {
pub(in crate::engine::snapshot) 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)
}
}
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()
}
})
}
}
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, _, capture_group, json_field_path, kind, _, _, log_stream_gated)| {
self.param_regex_cache.get(name).map(|re| {
(
name.to_string(),
re.clone(),
capture_group.to_string(),
json_field_path.map(|path| path.to_string()),
kind,
log_stream_gated,
)
})
})
.collect()
}
}
src/engine/snapshot/params.rs
+19 -280
View File
@@ -1,65 +1,12 @@
//! Param declaration: validating and storing explicit declarations, and the auto-detection //! 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. //! 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::MetricPrimitiveKind;
use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration}; #[cfg(test)]
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::snapshot::EngineSnapshot;
use crate::engine::timestamp::Timestamp; #[cfg(test)]
use crate::engine::DeclareParamRequest; use crate::engine::DeclareParamRequest;
use regex::Regex; 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) {
if let Ok(re) = Regex::new(&request.message_match_regex) {
self.param_regex_cache.insert(request.name.clone(), re);
}
self.index.insert(
IndexKey::ParamDeclaration { name: request.name.clone() },
DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }),
);
}
}
#[cfg(test)] #[cfg(test)]
pub(crate) fn declare_param(snapshot: &mut EngineSnapshot, request: DeclareParamRequest) -> Result<(), String> { pub(crate) fn declare_param(snapshot: &mut EngineSnapshot, request: DeclareParamRequest) -> Result<(), String> {
@@ -68,7 +15,7 @@ pub(crate) fn declare_param(snapshot: &mut EngineSnapshot, request: DeclareParam
Ok(()) Ok(())
} }
fn sanitize_group_name(key: &str) -> String { pub(crate) fn sanitize_group_name(key: &str) -> String {
let mut group: String = let mut group: String =
key.chars().map(|c| if c.is_ascii_alphanumeric() || c == '_' { c } else { '_' }).collect(); 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()) { if group.chars().next().is_none_or(|c| c.is_ascii_digit()) {
@@ -77,7 +24,7 @@ fn sanitize_group_name(key: &str) -> String {
group group
} }
fn infer_kind(sample: &str) -> MetricPrimitiveKind { pub(crate) fn infer_kind(sample: &str) -> MetricPrimitiveKind {
if sample.parse::<i64>().is_ok() || sample.parse::<f64>().is_ok() { if sample.parse::<i64>().is_ok() || sample.parse::<f64>().is_ok() {
MetricPrimitiveKind::Number MetricPrimitiveKind::Number
} else { } else {
@@ -85,7 +32,7 @@ fn infer_kind(sample: &str) -> MetricPrimitiveKind {
} }
} }
fn build_auto_param_pattern(key: &str, group: &str, quoted: bool) -> String { pub(crate) fn build_auto_param_pattern(key: &str, group: &str, quoted: bool) -> String {
let escaped_key = regex::escape(key); let escaped_key = regex::escape(key);
if quoted { if quoted {
format!(r#"{escaped_key}="(?P<{group}>[^"]*)""#) format!(r#"{escaped_key}="(?P<{group}>[^"]*)""#)
@@ -95,80 +42,12 @@ fn build_auto_param_pattern(key: &str, group: &str, quoted: bool) -> String {
} }
struct AutoParamCandidate { pub(crate) struct AutoParamCandidate {
name: String, pub(crate) name: String,
value: String, pub(crate) value: String,
message_match_regex: String, pub(crate) message_match_regex: String,
capture_group: String, pub(crate) capture_group: String,
json_field_path: Option<String>, pub(crate) 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;
}
if let Ok(re) = Regex::new(&candidate.message_match_regex) {
self.param_regex_cache.insert(candidate.name.clone(), re);
}
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( pub(crate) fn log_stream_header_transition(
@@ -178,67 +57,6 @@ pub(crate) fn log_stream_header_transition(
crate::engine::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone()) crate::engine::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone())
} }
impl EngineSnapshot {
pub(crate) fn derive_log_stream_params_from_row(
&mut self,
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 self.index.contains_key(&index_key) {
continue;
}
let message_match_regex = header.column_pattern(i, &column.name);
if let Ok(re) = Regex::new(&message_match_regex) {
self.param_regex_cache.insert(column.name.clone(), re);
}
self.index.insert(
index_key,
DataValue::ParamDeclaration(ParamDeclaration {
name: column.name.clone(),
message_match_regex,
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()) {
let index_key = IndexKey::ParamDeclaration { name: "message".to_string() };
if !self.index.contains_key(&index_key) {
let message_match_regex = header.message_pattern("message");
if let Ok(re) = Regex::new(&message_match_regex) {
self.param_regex_cache.insert("message".to_string(), re);
}
self.index.insert(
index_key,
DataValue::ParamDeclaration(ParamDeclaration {
name: "message".to_string(),
message_match_regex,
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)] #[cfg(test)]
pub(crate) fn auto_declare_log_stream_params( pub(crate) fn auto_declare_log_stream_params(
snapshot: &mut EngineSnapshot, snapshot: &mut EngineSnapshot,
@@ -255,61 +73,16 @@ pub(crate) fn auto_declare_log_stream_params(
} }
} }
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: name.clone() });
self.param_regex_cache.remove(&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 { pub(super) struct ResolvedParam {
regex: Regex, pub(super) regex: Regex,
capture_group: String, pub(super) capture_group: String,
json_field_path: Option<String>, pub(super) json_field_path: Option<String>,
kind: MetricPrimitiveKind, pub(super) kind: MetricPrimitiveKind,
elapsed_since_last: bool, pub(super) elapsed_since_last: bool,
log_stream_gated: bool, pub(super) log_stream_gated: bool,
} }
impl EngineSnapshot { pub(crate) fn extract_param_raw(
pub(super) fn resolved_param_declarations(&self) -> HashMap<String, ResolvedParam> {
self.param_declarations_range()
.filter_map(|(name, _, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| {
self.param_regex_cache.get(name).map(|regex| {
(
name.to_string(),
ResolvedParam {
regex: regex.clone(),
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, text: &str,
regex: &Regex, regex: &Regex,
capture_group: &str, capture_group: &str,
@@ -324,37 +97,3 @@ fn extract_param_raw(
None => regex.captures(text)?.name(capture_group).map(|m| m.as_str().to_string()), 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)
}
}
src/engine/snapshot/query.rs
+10 -184
View File
@@ -2,15 +2,15 @@
//! stats, declaration listing, bucket-value listing, and PromQL evaluation. //! stats, declaration listing, bucket-value listing, and PromQL evaluation.
use crate::api::dto::{ use crate::api::dto::{
BucketValueDto, LogEntryDto, MetricDeclarationDto, ParamDeclarationDto, PromqlSeriesDto, BucketValueDto, LogEntryDto, MetricDeclarationDto, ParamDeclarationDto,
StatsDto, TimelineBucketDto, StatsDto, TimelineBucketDto,
}; };
use crate::consts::{DEFAULT_TIMELINE_BUCKET_MS, MERGE_WINDOW_MS}; use crate::consts::{DEFAULT_TIMELINE_BUCKET_MS, MERGE_WINDOW_MS};
use crate::engine::data_value::{DataValue, MetricPrimitiveKind}; use crate::engine::data_value::DataValue;
use crate::engine::index_key::IndexKey; use crate::engine::index_key::IndexKey;
use crate::engine::message::{EngineQuery, EngineResponse}; use crate::engine::message::{EngineQuery, EngineResponse};
use crate::engine::metric_primitive::{MetricPrimitive, Number}; use crate::engine::metric_primitive::{MetricPrimitive, Number};
use crate::engine::promql::{AggOp, LabelMatcher, PromqlQuery}; use crate::engine::promql::{LabelMatcher, PromqlQuery};
use crate::engine::rate; use crate::engine::rate;
#[cfg(test)] #[cfg(test)]
use crate::engine::snapshot::sort::SortHelperIndexEntry; use crate::engine::snapshot::sort::SortHelperIndexEntry;
@@ -263,7 +263,7 @@ pub(super) fn handle_query_inner(
} }
} }
fn coerce_to_f64(value: &MetricPrimitive) -> f64 { pub(crate) fn coerce_to_f64(value: &MetricPrimitive) -> f64 {
match value { match value {
MetricPrimitive::Number(n) => match n { MetricPrimitive::Number(n) => match n {
Number::Int(v) => *v as f64, Number::Int(v) => *v as f64,
@@ -282,7 +282,7 @@ fn coerce_to_f64(value: &MetricPrimitive) -> f64 {
} }
} }
fn matches_label_matchers( pub(crate) fn matches_label_matchers(
labels: &BTreeMap<String, MetricPrimitive>, labels: &BTreeMap<String, MetricPrimitive>,
matchers: &[LabelMatcher], matchers: &[LabelMatcher],
) -> bool { ) -> bool {
@@ -291,7 +291,7 @@ fn matches_label_matchers(
}) })
} }
fn project_labels( pub(crate) fn project_labels(
labels: &BTreeMap<String, String>, labels: &BTreeMap<String, String>,
by_labels: &[String], by_labels: &[String],
) -> BTreeMap<String, String> { ) -> BTreeMap<String, String> {
@@ -301,7 +301,7 @@ fn project_labels(
.collect() .collect()
} }
fn windowed_rate_series( pub(crate) fn windowed_rate_series(
points: &[(i64, f64)], points: &[(i64, f64)],
range_ms: i64, range_ms: i64,
start_ms: Option<i64>, start_ms: Option<i64>,
@@ -329,7 +329,7 @@ fn windowed_rate_series(
series series
} }
fn merge_series_windowed(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { pub(crate) fn merge_series_windowed(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> {
if sub_series.len() <= 1 { if sub_series.len() <= 1 {
return sub_series.first().cloned().unwrap_or_default(); return sub_series.first().cloned().unwrap_or_default();
} }
@@ -350,7 +350,7 @@ fn merge_series_windowed(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i
.collect() .collect()
} }
fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { pub(crate) 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(); let mut all_timestamps: std::collections::BTreeSet<i64> = std::collections::BTreeSet::new();
for points in sub_series { for points in sub_series {
all_timestamps.extend(points.iter().map(|(ts, _)| *ts)); all_timestamps.extend(points.iter().map(|(ts, _)| *ts));
@@ -373,184 +373,10 @@ fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64,
.collect() .collect()
} }
fn ottyl_reserved_numeric_value_fn(metric_name: &str) -> Option<fn(u64, i64) -> f64> { pub(crate) fn ottyl_reserved_numeric_value_fn(metric_name: &str) -> Option<fn(u64, i64) -> f64> {
match metric_name { match metric_name {
"ottyl_seq" => Some(|sequence, _timestamp_ms| sequence as f64), "ottyl_seq" => Some(|sequence, _timestamp_ms| sequence as f64),
"ottyl_ts" => Some(|_sequence, timestamp_ms| timestamp_ms as f64), "ottyl_ts" => Some(|_sequence, timestamp_ms| timestamp_ms as f64),
_ => None, _ => 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()
}
})
}
}
src/engine/snapshot/sort.rs
+2 -66
View File
@@ -3,12 +3,12 @@
//! O(n log n) over every log, so it's memoized in `EngineSnapshot::sort_helper_indexes` and //! 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. //! 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::data_value::DataValue;
use crate::engine::index_key::IndexKey; use crate::engine::index_key::IndexKey;
use crate::engine::data_value::MetricPrimitiveKind; use crate::engine::data_value::MetricPrimitiveKind;
use crate::engine::metric_primitive::MetricPrimitive; use crate::engine::metric_primitive::MetricPrimitive;
use crate::engine::snapshot::metrics::parse_metric_primitive; use crate::engine::snapshot::metrics::parse_metric_primitive;
#[cfg(test)]
use crate::engine::snapshot::EngineSnapshot; use crate::engine::snapshot::EngineSnapshot;
use crate::engine::timestamp::Timestamp; use crate::engine::timestamp::Timestamp;
use regex::Regex; use regex::Regex;
@@ -28,7 +28,7 @@ pub(crate) struct SortHelperIndexEntry {
pub(super) type CompiledParamDeclaration = (String, Regex, String, Option<String>, MetricPrimitiveKind, bool); pub(super) type CompiledParamDeclaration = (String, Regex, String, Option<String>, MetricPrimitiveKind, bool);
fn derive_single_param( pub(crate) fn derive_single_param(
text: &str, text: &str,
declarations: &[CompiledParamDeclaration], declarations: &[CompiledParamDeclaration],
param_name: &str, param_name: &str,
@@ -49,70 +49,6 @@ fn derive_single_param(
parse_metric_primitive(&raw, *kind) 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, _, capture_group, json_field_path, kind, _, _, log_stream_gated)| {
self.param_regex_cache.get(name).map(|re| {
(
name.to_string(),
re.clone(),
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> { pub(crate) fn derive_log_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap<String, MetricPrimitive> {
let mut params = BTreeMap::new(); let mut params = BTreeMap::new();
for (name, regex, capture_group, json_field_path, kind, log_stream_gated) in declarations { for (name, regex, capture_group, json_field_path, kind, log_stream_gated) in declarations {