Hugh KaznowskiandClaude Sonnet 5
f43445b280
DataValue::LogLiteral and IngestedLine now carry an Arc<str> instead of an owned String, so the two clones the ingest path paid per line (once to move text into the write closure, once to store it) become cheap refcount bumps instead of full string copies. Also cuts the per-string-content cost of left_right's full-EngineSnapshot deep clone on each reconciliation sync. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ
1440 lines
55 KiB
Rust
1440 lines
55 KiB
Rust
//! `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`. The submodules below hold everything that isn't part of `EngineSnapshot`'s
|
|
//! 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 metrics;
|
|
pub(crate) mod params;
|
|
pub(crate) mod query;
|
|
pub(crate) mod sort;
|
|
|
|
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, MetricSemantics, ParamDeclaration,
|
|
};
|
|
use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
|
|
use crate::engine::index_key::IndexKey;
|
|
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::{DeclareMetricRequest, DeclareParamRequest};
|
|
use crate::flavour::LogLiteralHelper;
|
|
use filter::CompiledFilter;
|
|
use metrics::parse_metric_primitive;
|
|
use params::{
|
|
AutoParamCandidate, ResolvedParam, build_auto_param_pattern, extract_param_raw, infer_kind,
|
|
sanitize_group_name,
|
|
};
|
|
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 sort::{
|
|
CompiledParamDeclaration, SortHelperAccess, SortHelperIndexEntry, SortHelperKey,
|
|
derive_single_param,
|
|
};
|
|
use std::collections::{BTreeMap, HashMap, HashSet};
|
|
|
|
/// One row of a metric declaration: name, message-match regex, label param names, value param
|
|
/// name, and the label-name-to-value-param and static-label maps.
|
|
type MetricDeclarationRow<'a> = (
|
|
&'a str,
|
|
&'a str,
|
|
&'a [String],
|
|
Option<&'a str>,
|
|
Option<&'a BTreeMap<String, String>>,
|
|
Option<&'a BTreeMap<String, String>>,
|
|
MetricSemantics,
|
|
);
|
|
|
|
/// Per-label-group point series, keyed by the group's label set, before merging into one
|
|
/// `PromqlSeriesDto` per group.
|
|
type GroupedPointSeries = BTreeMap<BTreeMap<String, String>, Vec<Vec<(i64, f64)>>>;
|
|
|
|
#[derive(Debug, Clone, Default)]
|
|
pub(crate) struct EngineSnapshot {
|
|
pub(crate) index: BTreeMap<IndexKey, DataValue>,
|
|
pub(crate) sort_helper_indexes: HashMap<String, SortHelperIndexEntry>,
|
|
/// Compiled regexes for param/metric declarations, keyed by declaration name. Compiling a
|
|
/// regex is far more expensive than matching with one, and `apply_all_metrics`/
|
|
/// `update_sort_helper_indexes_on_ingest` run per ingested line, so every insertion or
|
|
/// removal of a declaration (see `params.rs`/`metrics.rs`) must keep these in sync rather
|
|
/// than recompiling from the stored pattern string on every line.
|
|
pub(crate) param_regex_cache: HashMap<String, Regex>,
|
|
pub(crate) metric_regex_cache: HashMap<String, Regex>,
|
|
pub(crate) log_count: usize,
|
|
pub(crate) approx_bytes_total: u64,
|
|
pub(crate) ingest_lag_high_watermark_ms: i64,
|
|
}
|
|
|
|
// --- core: snapshot construction and log range access ---
|
|
impl EngineSnapshot {
|
|
pub(crate) fn metric_declarations_range(
|
|
&self,
|
|
) -> impl Iterator<Item = MetricDeclarationRow<'_>> {
|
|
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,
|
|
semantics,
|
|
},
|
|
) => 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(),
|
|
*semantics,
|
|
)),
|
|
_ => 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 from_index(index: BTreeMap<IndexKey, DataValue>) -> Self {
|
|
let mut snapshot = EngineSnapshot {
|
|
index,
|
|
..Default::default()
|
|
};
|
|
snapshot.rebuild_regex_caches();
|
|
snapshot
|
|
}
|
|
|
|
/// Recompiles `param_regex_cache`/`metric_regex_cache` from the declarations currently in
|
|
/// `index`. Needed whenever a snapshot's declarations are populated by inserting straight
|
|
/// into `index` rather than through the normal declare/insert methods that keep the caches
|
|
/// in sync incrementally (e.g. `from_index` building a snapshot from a bare index).
|
|
pub(crate) fn rebuild_regex_caches(&mut self) {
|
|
self.param_regex_cache = self
|
|
.param_declarations_range()
|
|
.filter_map(|(name, message_match_regex, ..)| {
|
|
regex::Regex::new(message_match_regex)
|
|
.ok()
|
|
.map(|re| (name.to_string(), re))
|
|
})
|
|
.collect();
|
|
self.metric_regex_cache = self
|
|
.metric_declarations_range()
|
|
.filter_map(|(name, message_match_regex, ..)| {
|
|
regex::Regex::new(message_match_regex)
|
|
.ok()
|
|
.map(|re| (name.to_string(), re))
|
|
})
|
|
.collect();
|
|
}
|
|
|
|
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(text.approx_log_bytes());
|
|
}
|
|
self.index.remove(&oldest_key);
|
|
self.log_count -= 1;
|
|
}
|
|
}
|
|
|
|
pub(crate) fn logs_range(&self) -> impl Iterator<Item = (u64, Timestamp, &str)> {
|
|
self.logs_in_sequence_range(..)
|
|
}
|
|
|
|
/// Seeks directly to `sequence_range` within the `IngestedLog` key range instead of
|
|
/// scanning from `IndexKey::Start` and filtering after the fact. `IngestedLog` sorts by
|
|
/// `(sequence, timestamp)`, and `sequence` is assigned monotonically at ingest
|
|
/// (`handle.rs`), so a bound on `sequence` alone is an exact seek — `Timestamp::MIN` is
|
|
/// used as the synthetic sub-key since it sorts below any real timestamp for that sequence.
|
|
pub(crate) fn logs_in_sequence_range(
|
|
&self,
|
|
sequence_range: impl std::ops::RangeBounds<u64>,
|
|
) -> impl Iterator<Item = (u64, Timestamp, &str)> {
|
|
let lower = match sequence_range.start_bound() {
|
|
std::ops::Bound::Included(&sequence) => IndexKey::IngestedLog {
|
|
sequence,
|
|
timestamp: Timestamp::MIN,
|
|
},
|
|
std::ops::Bound::Excluded(&sequence) => IndexKey::IngestedLog {
|
|
sequence: sequence.saturating_add(1),
|
|
timestamp: Timestamp::MIN,
|
|
},
|
|
std::ops::Bound::Unbounded => IndexKey::Start,
|
|
};
|
|
let upper = match sequence_range.end_bound() {
|
|
std::ops::Bound::Included(&sequence) => IndexKey::IngestedLog {
|
|
sequence: sequence.saturating_add(1),
|
|
timestamp: Timestamp::MIN,
|
|
},
|
|
std::ops::Bound::Excluded(&sequence) => IndexKey::IngestedLog {
|
|
sequence,
|
|
timestamp: Timestamp::MIN,
|
|
},
|
|
std::ops::Bound::Unbounded => IndexKey::LogParamSplit,
|
|
};
|
|
self.index
|
|
.range(lower..upper)
|
|
.filter_map(|(key, value)| match (key, value) {
|
|
(
|
|
IndexKey::IngestedLog {
|
|
sequence,
|
|
timestamp,
|
|
},
|
|
DataValue::LogLiteral { text },
|
|
) => Some((*sequence, *timestamp, text.as_ref())),
|
|
_ => None,
|
|
})
|
|
}
|
|
|
|
pub(crate) fn logs_since(
|
|
&self,
|
|
before_sequence: u64,
|
|
since_ms: i64,
|
|
now_ms: i64,
|
|
) -> impl Iterator<Item = (u64, Timestamp, &str)> {
|
|
self.logs_in_sequence_range(..before_sequence)
|
|
.filter(move |(_, timestamp, _)| {
|
|
timestamp.as_millis() > since_ms && timestamp.as_millis() <= now_ms
|
|
})
|
|
}
|
|
}
|
|
|
|
// --- filter: compiling search-bar filter expressions ---
|
|
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(),
|
|
})
|
|
&& !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))
|
|
}
|
|
}
|
|
|
|
// --- metrics: declaring, applying, and querying metric buckets ---
|
|
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
|
|
&& !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());
|
|
self.resolve_backfill_and_store_metric(request, &re);
|
|
Ok(())
|
|
}
|
|
|
|
/// Shared by `declare_metric` and `update_metric`: resolves param declarations, backfills
|
|
/// the metric's buckets from existing log history, then writes the `MetricDeclaration` entry.
|
|
/// Callers differ only in what happens before this (rename bookkeeping for updates).
|
|
fn resolve_backfill_and_store_metric(&mut self, request: DeclareMetricRequest, re: &Regex) {
|
|
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.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,
|
|
semantics: request.semantics,
|
|
},
|
|
);
|
|
}
|
|
|
|
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)?;
|
|
|
|
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.resolve_backfill_and_store_metric(request, &re);
|
|
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
|
|
}
|
|
|
|
#[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()
|
|
}
|
|
|
|
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,
|
|
}
|
|
}
|
|
|
|
pub(crate) fn declared_semantics(&self, metric_name: &str) -> Option<MetricSemantics> {
|
|
match self.index.get(&IndexKey::MetricDeclaration {
|
|
name: metric_name.to_string(),
|
|
}) {
|
|
Some(DataValue::MetricDeclaration { semantics, .. }) => Some(*semantics),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
/// Whether `metric_name` captures a per-event value (rather than just counting matching log
|
|
/// lines). The gauge/counter semantics guard in `evaluate_promql` only applies to metrics
|
|
/// with a captured value: a bare occurrence count is always safe to both `rate()` (events per
|
|
/// window) and `sum()`/`avg()` (total/average occurrences) since each event contributes
|
|
/// exactly one point, unlike a captured reading that might itself be a cumulative total.
|
|
fn declares_value_param(&self, metric_name: &str) -> bool {
|
|
matches!(
|
|
self.index.get(&IndexKey::MetricDeclaration {
|
|
name: metric_name.to_string(),
|
|
}),
|
|
Some(DataValue::MetricDeclaration {
|
|
value_param: Some(_),
|
|
..
|
|
})
|
|
)
|
|
}
|
|
}
|
|
|
|
// --- params: declaring and resolving param values ---
|
|
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(),
|
|
})
|
|
&& 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
|
|
}),
|
|
);
|
|
}
|
|
|
|
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;
|
|
}
|
|
if self.try_insert_auto_param(candidate, &mut total) {
|
|
new_this_line += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Inserts `candidate` as an auto-derived `ParamDeclaration` unless a param with that name
|
|
/// is already declared, bumping `*total` and returning whether it inserted.
|
|
fn try_insert_auto_param(&mut self, candidate: AutoParamCandidate, total: &mut usize) -> bool {
|
|
let index_key = IndexKey::ParamDeclaration {
|
|
name: candidate.name.clone(),
|
|
};
|
|
if self.index.contains_key(&index_key) {
|
|
return false;
|
|
}
|
|
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,
|
|
}),
|
|
);
|
|
*total += 1;
|
|
true
|
|
}
|
|
|
|
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,
|
|
}),
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
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(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()
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|
|
|
|
// --- query: EngineQuery/PromQL evaluation ---
|
|
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)
|
|
}
|
|
|
|
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
|
|
));
|
|
}
|
|
// A metric with no captured value (just occurrence counts of matching log lines) is
|
|
// always safe under both rate() and sum()/avg(): each event contributes exactly one
|
|
// point, so there's no cumulative-total to misinterpret. The guard below only matters
|
|
// once a metric captures a per-event value via `value_param`.
|
|
let declared_semantics = self
|
|
.declares_value_param(&parsed.metric_name)
|
|
.then(|| self.declared_semantics(&parsed.metric_name))
|
|
.flatten();
|
|
if matches!(parsed.agg, Some(AggOp::Rate))
|
|
&& declared_semantics == Some(MetricSemantics::Gauge)
|
|
{
|
|
return Err(format!(
|
|
"cannot rate() \"{}\": it's declared as a gauge, not a counter; gauges are point-in-time \
|
|
readings, so rate() over them isn't meaningful — use it bare or wrap it in gauge()",
|
|
parsed.metric_name
|
|
));
|
|
}
|
|
if matches!(parsed.agg, Some(AggOp::Sum) | Some(AggOp::Avg))
|
|
&& !parsed.inner_rate
|
|
&& declared_semantics == Some(MetricSemantics::Counter)
|
|
{
|
|
return Err(format!(
|
|
"cannot sum/avg \"{}\" directly: it's declared as a counter, so summing/averaging its raw \
|
|
values double-counts the running total — wrap it in rate() first, e.g. sum(rate({}[5m]))",
|
|
parsed.metric_name, 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: GroupedPointSeries = 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: GroupedPointSeries = 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()
|
|
}
|
|
Some(AggOp::Quantile) => {
|
|
let quantile = parsed
|
|
.quantile
|
|
.expect("histogram_quantile() always carries a quantile");
|
|
// The bracketed range, if any, narrows each series to its trailing window
|
|
// (mirroring rate()'s range selector) rather than the full panel window —
|
|
// interpolating over whatever raw values are retained for that series, since
|
|
// this engine doesn't keep a separate pre-aggregated bucketed-histogram
|
|
// representation to compute quantiles from.
|
|
let range_ms = parsed.range.map(|d| d.as_millis() as i64);
|
|
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<f64>> = BTreeMap::new();
|
|
for (labels, points) in &series_points {
|
|
let group_key = project_labels(labels, &parsed.by_labels);
|
|
let window_start = match range_ms {
|
|
Some(range_ms) => {
|
|
let window_end = points.last().map(|(ts, _)| *ts).unwrap_or(0);
|
|
Some(window_end - range_ms)
|
|
}
|
|
None => None,
|
|
};
|
|
groups.entry(group_key).or_default().extend(
|
|
points
|
|
.iter()
|
|
.filter(|(ts, _)| window_start.is_none_or(|start| *ts >= start))
|
|
.map(|(_, value)| *value),
|
|
);
|
|
}
|
|
|
|
groups
|
|
.into_iter()
|
|
.filter_map(|(labels, mut values)| {
|
|
let quantile_value = interpolated_quantile(&mut values, quantile)?;
|
|
Some(PromqlSeriesDto {
|
|
labels,
|
|
points: vec![(last_ts, quantile_value)],
|
|
})
|
|
})
|
|
.collect()
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
/// Linear interpolation between closest ranks over the raw retained sample values (the "R-7"
|
|
/// method, matching numpy/Excel's default `PERCENTILE` behavior) — exact for the data actually
|
|
/// kept, unlike a bucketed-histogram approximation.
|
|
fn interpolated_quantile(values: &mut [f64], quantile: f64) -> Option<f64> {
|
|
if values.is_empty() {
|
|
return None;
|
|
}
|
|
values.sort_by(|a, b| a.partial_cmp(b).unwrap());
|
|
if values.len() == 1 {
|
|
return Some(values[0]);
|
|
}
|
|
let rank = quantile * (values.len() - 1) as f64;
|
|
let lower = rank.floor() as usize;
|
|
let upper = rank.ceil() as usize;
|
|
if lower == upper {
|
|
return Some(values[lower]);
|
|
}
|
|
let fraction = rank - lower as f64;
|
|
Some(values[lower] + (values[upper] - values[lower]) * fraction)
|
|
}
|
|
|
|
// --- sort: sort-helper index maintenance ---
|
|
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()
|
|
}
|
|
}
|