diff --git a/src/engine/snapshot/filter.rs b/src/engine/snapshot/filter.rs index 191c348..ab9ad22 100644 --- a/src/engine/snapshot/filter.rs +++ b/src/engine/snapshot/filter.rs @@ -3,11 +3,8 @@ //! metric-bucket comparisons, and ad-hoc JSON field comparisons. use crate::engine::filter_datetime; -use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; -use crate::engine::index_key::IndexKey; -use crate::engine::data_value::DataValue; +use crate::engine::filter_parser::{CompareOp, FilterValue}; use crate::engine::metric_primitive::MetricPrimitive; -use crate::engine::snapshot::EngineSnapshot; use crate::engine::timestamp::Timestamp; use regex::Regex; use std::cmp::Ordering; @@ -24,88 +21,8 @@ pub(super) enum CompiledFilter { Or(Box, Box), } -impl EngineSnapshot { - pub(super) fn compile_filter_query(&self, filter: &str, now: Timestamp) -> Result { - 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 { - 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 { - let param = param.ok_or_else(|| { - format!( - "metric comparison on '{metric}' requires a [param] (e.g. '{metric}[some_param]{}...'); \ - v1 buckets only carry per-label occurrence data, not a single metric value", - match op { - CompareOp::Lt => "<", - CompareOp::Gt => ">", - CompareOp::Le => "<=", - CompareOp::Ge => ">=", - CompareOp::Eq => "==", - CompareOp::Ne => "!=", - } - ) - })?; - - if let Some(DataValue::MetricDeclaration { label_params, .. }) = - self.index.get(&IndexKey::MetricDeclaration { name: metric.to_string() }) - { - if !label_params.iter().any(|p| p == param) { - return Err(format!( - "param '{param}' is not a label param of metric '{metric}'; declared label params: [{}]", - label_params.join(", ") - )); - } - } - - let mut matching = HashSet::new(); - for (labels, (sequence, _, _)) in self.bucket_values(metric, None) { - let Some(primitive) = labels.get(param) else { - continue; - }; - if primitive.compare(op, value, now)? { - matching.insert(*sequence); - } - } - Ok(CompiledFilter::MetricMatch(matching)) - } -} - impl CompiledFilter { - fn from_seq_comparison(op: CompareOp, value: &FilterValue) -> Result { + pub(super) fn from_seq_comparison(op: CompareOp, value: &FilterValue) -> Result { let target = match value { FilterValue::Number(n) => *n, FilterValue::Raw(raw) => raw.parse::().map_err(|_| { @@ -115,7 +32,7 @@ impl CompiledFilter { Ok(CompiledFilter::Sequence { op, target }) } - fn from_ts_comparison(op: CompareOp, value: &FilterValue, now: Timestamp) -> Result { + pub(super) fn from_ts_comparison(op: CompareOp, value: &FilterValue, now: Timestamp) -> Result { let target_ms = match value { FilterValue::Number(n) => *n as i64, FilterValue::Raw(raw) => filter_datetime::parse_datetime(raw, now) @@ -125,7 +42,7 @@ impl CompiledFilter { 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 { FilterValue::Number(n) => n.to_string(), FilterValue::Raw(raw) => raw.clone(), @@ -135,7 +52,7 @@ impl CompiledFilter { } impl MetricPrimitive { - fn compare(&self, op: CompareOp, value: &FilterValue, now: Timestamp) -> Result { + pub(super) fn compare(&self, op: CompareOp, value: &FilterValue, now: Timestamp) -> Result { let ordering = match (self, value) { (MetricPrimitive::Number(n), FilterValue::Number(v)) => n.as_f64().partial_cmp(v), (MetricPrimitive::Number(n), FilterValue::Raw(raw)) => { diff --git a/src/engine/snapshot/metrics.rs b/src/engine/snapshot/metrics.rs index b288eda..63b712d 100644 --- a/src/engine/snapshot/metrics.rs +++ b/src/engine/snapshot/metrics.rs @@ -2,133 +2,9 @@ //! ingested lines to populate metric buckets, and backfilling a newly (re)declared metric //! against already-ingested logs. -use crate::engine::data_value::{DataValue, MetricPrimitiveKind}; -use crate::engine::index_key::IndexKey; +use crate::engine::data_value::MetricPrimitiveKind; use crate::engine::metric_primitive::{MetricPrimitive, Number}; -use crate::engine::snapshot::params::ResolvedParam; -use crate::engine::snapshot::EngineSnapshot; use crate::engine::timestamp::Timestamp; -use crate::engine::DeclareMetricRequest; -use regex::Regex; -use std::collections::{BTreeMap, HashMap}; - -impl EngineSnapshot { - pub(crate) fn validate_metric_request(&self, request: &DeclareMetricRequest) -> Result { - 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 = 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 { - 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 { 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, - matched_labels: Option<&BTreeMap>, - unmatched_labels: Option<&BTreeMap>, - 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, - 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, - matched_labels: Option<&BTreeMap>, - unmatched_labels: Option<&BTreeMap>, - ) { - 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, - Option, - Option>, - Option>, - ); - let metrics: Vec = 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>, - ) -> impl Iterator, &'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 { - 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, - } - } -} diff --git a/src/engine/snapshot/mod.rs b/src/engine/snapshot/mod.rs index c33d8c9..cfa6351 100644 --- a/src/engine/snapshot/mod.rs +++ b/src/engine/snapshot/mod.rs @@ -1,8 +1,9 @@ //! `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` and read/written through the left-right index in -//! `super::handle`. Its impl is split across the submodules below by concern; this file holds -//! the struct itself plus the accessors every submodule builds on. +//! `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; @@ -12,14 +13,29 @@ 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, 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 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 sort::{SortHelperAccess, SortHelperIndexEntry}; -use std::collections::{BTreeMap, HashMap}; +use sort::{derive_single_param, CompiledParamDeclaration, SortHelperAccess, SortHelperIndexEntry, SortHelperKey}; +use std::collections::{BTreeMap, HashMap, HashSet}; #[derive(Debug, Clone, Default)] pub(crate) struct EngineSnapshot { @@ -216,3 +232,889 @@ impl EngineSnapshot { 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 { + 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 { + 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 { + 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 { + 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 = 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 { + 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, + matched_labels: Option<&BTreeMap>, + unmatched_labels: Option<&BTreeMap>, + 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, + 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, + matched_labels: Option<&BTreeMap>, + unmatched_labels: Option<&BTreeMap>, + ) { + 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, + Option, + Option>, + Option>, + ); + let metrics: Vec = 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>, + ) -> impl Iterator, &'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 { + 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 = 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 = + 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 = 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 { + 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, + text: &str, + timestamp: Timestamp, + ) -> Option { + 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, + end_ms: Option, + ) -> Result, 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, Vec<(i64, f64)>> = BTreeMap::new(); + if let Some(reserved_value) = ottyl_reserved_numeric_value_fn(&parsed.metric_name) { + let no_labels: BTreeMap = 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 = + 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, Vec>> = 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, Vec>> = 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, 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 = 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 { + 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 { + 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() + } +} diff --git a/src/engine/snapshot/params.rs b/src/engine/snapshot/params.rs index 31e1e68..c92b380 100644 --- a/src/engine/snapshot/params.rs +++ b/src/engine/snapshot/params.rs @@ -1,65 +1,12 @@ //! Param declaration: validating and storing explicit declarations, and the auto-detection //! logic that derives declarations from logfmt/json/log-stream-shaped lines as they're ingested. -use crate::consts::{MAX_AUTO_DECLARED_PARAMS, MAX_NEW_AUTO_PARAMS_PER_LINE, OTTYL_RESERVED_PARAM_NAMES}; -use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration}; -use crate::engine::index_key::IndexKey; -use crate::engine::metric_primitive::MetricPrimitive; -use crate::engine::snapshot::metrics::parse_metric_primitive; +use crate::engine::data_value::MetricPrimitiveKind; +#[cfg(test)] use crate::engine::snapshot::EngineSnapshot; -use crate::engine::timestamp::Timestamp; +#[cfg(test)] use crate::engine::DeclareParamRequest; use regex::Regex; -use std::collections::{BTreeMap, HashMap}; - -impl EngineSnapshot { - pub(crate) fn validate_param_declaration(&self, request: &DeclareParamRequest) -> Result<(), String> { - if OTTYL_RESERVED_PARAM_NAMES.contains(&request.name.as_str()) { - return Err(format!( - "'{}' is a reserved name, present on every log line without declaring anything; \ - it can't be redeclared as a param", - request.name - )); - } - if let Some(DataValue::ParamDeclaration(existing)) = - self.index.get(&IndexKey::ParamDeclaration { name: request.name.clone() }) - { - if existing.auto_derived { - return Err(format!( - "param '{}' was auto-detected from the log stream and can't be edited", - request.name - )); - } - } - let re = Regex::new(&request.message_match_regex) - .map_err(|err| format!("invalid regex: {err}"))?; - if request.elapsed_since_last && request.kind != MetricPrimitiveKind::Duration { - return Err("elapsed_since_last params must declare kind 'duration'".to_string()); - } - let uses_json_field = request.json_field_path.as_deref().is_some_and(|path| !path.is_empty()); - if request.kind != MetricPrimitiveKind::Boolean - && !request.elapsed_since_last - && !uses_json_field - && !re.capture_names().flatten().any(|name| name == request.capture_group) - { - return Err(format!( - "regex has no capture group named '{}'", - request.capture_group - )); - } - Ok(()) - } - - pub(crate) fn insert_param_declaration(&mut self, request: DeclareParamRequest) { - 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)] 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(()) } -fn sanitize_group_name(key: &str) -> String { +pub(crate) fn sanitize_group_name(key: &str) -> String { let mut group: String = key.chars().map(|c| if c.is_ascii_alphanumeric() || c == '_' { c } else { '_' }).collect(); if group.chars().next().is_none_or(|c| c.is_ascii_digit()) { @@ -77,7 +24,7 @@ fn sanitize_group_name(key: &str) -> String { group } -fn infer_kind(sample: &str) -> MetricPrimitiveKind { +pub(crate) fn infer_kind(sample: &str) -> MetricPrimitiveKind { if sample.parse::().is_ok() || sample.parse::().is_ok() { MetricPrimitiveKind::Number } 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); if quoted { format!(r#"{escaped_key}="(?P<{group}>[^"]*)""#) @@ -95,80 +42,12 @@ fn build_auto_param_pattern(key: &str, group: &str, quoted: bool) -> String { } -struct AutoParamCandidate { - name: String, - value: String, - message_match_regex: String, - capture_group: String, - json_field_path: Option, -} - -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 = 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) struct AutoParamCandidate { + pub(crate) name: String, + pub(crate) value: String, + pub(crate) message_match_regex: String, + pub(crate) capture_group: String, + pub(crate) json_field_path: Option, } 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()) } -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)] pub(crate) fn auto_declare_log_stream_params( 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 = - 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 = self.logs_range().map(|(_, _, text)| text.to_string()).collect(); - let logs_considered = texts.len(); - for text in &texts { - self.auto_declare_params(text); - } - let reextracted = self.param_declarations_range().count(); - (reextracted, logs_considered) - } -} - pub(super) struct ResolvedParam { - regex: Regex, - capture_group: String, - json_field_path: Option, - kind: MetricPrimitiveKind, - elapsed_since_last: bool, - log_stream_gated: bool, + pub(super) regex: Regex, + pub(super) capture_group: String, + pub(super) json_field_path: Option, + pub(super) kind: MetricPrimitiveKind, + pub(super) elapsed_since_last: bool, + pub(super) log_stream_gated: bool, } -impl EngineSnapshot { - pub(super) fn resolved_param_declarations(&self) -> HashMap { - 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( +pub(crate) fn extract_param_raw( text: &str, regex: &Regex, 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()), } } - -impl EngineSnapshot { - pub(super) fn resolve_param_value( - &mut self, - metric_name: &str, - resolved: &ResolvedParam, - labels_so_far: &BTreeMap, - text: &str, - timestamp: Timestamp, - ) -> Option { - 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) - } -} diff --git a/src/engine/snapshot/query.rs b/src/engine/snapshot/query.rs index bfa76c2..80d01d5 100644 --- a/src/engine/snapshot/query.rs +++ b/src/engine/snapshot/query.rs @@ -2,15 +2,15 @@ //! stats, declaration listing, bucket-value listing, and PromQL evaluation. use crate::api::dto::{ - BucketValueDto, LogEntryDto, MetricDeclarationDto, ParamDeclarationDto, PromqlSeriesDto, + BucketValueDto, LogEntryDto, MetricDeclarationDto, ParamDeclarationDto, StatsDto, TimelineBucketDto, }; 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::message::{EngineQuery, EngineResponse}; use crate::engine::metric_primitive::{MetricPrimitive, Number}; -use crate::engine::promql::{AggOp, LabelMatcher, PromqlQuery}; +use crate::engine::promql::{LabelMatcher, PromqlQuery}; use crate::engine::rate; #[cfg(test)] 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 { MetricPrimitive::Number(n) => match n { 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, matchers: &[LabelMatcher], ) -> bool { @@ -291,7 +291,7 @@ fn matches_label_matchers( }) } -fn project_labels( +pub(crate) fn project_labels( labels: &BTreeMap, by_labels: &[String], ) -> BTreeMap { @@ -301,7 +301,7 @@ fn project_labels( .collect() } -fn windowed_rate_series( +pub(crate) fn windowed_rate_series( points: &[(i64, f64)], range_ms: i64, start_ms: Option, @@ -329,7 +329,7 @@ fn windowed_rate_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 { 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() } -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 = std::collections::BTreeSet::new(); for points in sub_series { 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() } -fn ottyl_reserved_numeric_value_fn(metric_name: &str) -> Option f64> { +pub(crate) fn ottyl_reserved_numeric_value_fn(metric_name: &str) -> Option f64> { match metric_name { "ottyl_seq" => Some(|sequence, _timestamp_ms| sequence as f64), "ottyl_ts" => Some(|_sequence, timestamp_ms| timestamp_ms as f64), _ => None, } } - -impl EngineSnapshot { - pub(crate) fn evaluate_promql( - &self, - parsed: &PromqlQuery, - start_ms: Option, - end_ms: Option, - ) -> Result, 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, Vec<(i64, f64)>> = BTreeMap::new(); - if let Some(reserved_value) = ottyl_reserved_numeric_value_fn(&parsed.metric_name) { - let no_labels: BTreeMap = 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 = - 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, Vec>> = 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, Vec>> = 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, 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 = 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() - } - }) - } -} diff --git a/src/engine/snapshot/sort.rs b/src/engine/snapshot/sort.rs index bca2753..b80eb24 100644 --- a/src/engine/snapshot/sort.rs +++ b/src/engine/snapshot/sort.rs @@ -3,12 +3,12 @@ //! O(n log n) over every log, so it's memoized in `EngineSnapshot::sort_helper_indexes` and //! kept incrementally up to date on ingest, rather than rebuilt per query. -use crate::consts::SORT_HELPER_INDEX_PURGE_AFTER; use crate::engine::data_value::DataValue; use crate::engine::index_key::IndexKey; use crate::engine::data_value::MetricPrimitiveKind; use crate::engine::metric_primitive::MetricPrimitive; use crate::engine::snapshot::metrics::parse_metric_primitive; +#[cfg(test)] use crate::engine::snapshot::EngineSnapshot; use crate::engine::timestamp::Timestamp; use regex::Regex; @@ -28,7 +28,7 @@ pub(crate) struct SortHelperIndexEntry { pub(super) type CompiledParamDeclaration = (String, Regex, String, Option, MetricPrimitiveKind, bool); -fn derive_single_param( +pub(crate) fn derive_single_param( text: &str, declarations: &[CompiledParamDeclaration], param_name: &str, @@ -49,70 +49,6 @@ fn derive_single_param( parse_metric_primitive(&raw, *kind) } -impl EngineSnapshot { - pub(super) fn build_sort_helper_index(&self, sort_by: &str) -> BTreeMap { - 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 { - 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 { let mut params = BTreeMap::new(); for (name, regex, capture_group, json_field_path, kind, log_stream_gated) in declarations {