use crate::alerts::{self, AlertRuntimeState}; use crate::consts::{ DEFAULT_TIMELINE_BUCKET_MS, INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS, MAX_AUTO_DECLARED_PARAMS, MAX_NEW_AUTO_PARAMS_PER_LINE, MERGE_WINDOW_MS, OTTYL_RESERVED_PARAM_NAMES, SORT_HELPER_INDEX_PURGE_AFTER, }; use crate::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration}; use crate::dto::{ BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, PromqlSeriesDto, StatsDto, TimelineBucketDto, TimelineParams, }; use crate::filter_datetime; use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; use crate::index_key::IndexKey; use crate::metric_primitive::{MetricPrimitive, Number}; use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery}; use crate::rate; use crate::settings::SettingsHandle; use crate::timestamp::Timestamp; use left_right::Absorb; use regex::Regex; use serde::Deserialize; use std::cmp::Ordering; use std::collections::{BTreeMap, HashMap, HashSet}; use tokio::sync::{mpsc, oneshot}; pub struct IngestedLine { pub sequence: u64, pub timestamp: Timestamp, pub text: String, } pub struct MetricSample { pub metric_name: String, pub labels: BTreeMap, pub value: f64, pub timestamp: Timestamp, } pub type DeclareParamRequest = ParamDeclaration; #[derive(Debug, Clone, serde::Serialize)] pub struct ClearParamsResult { pub removed: usize, pub reextracted: usize, pub logs_considered: usize, } #[derive(Debug, Clone, Deserialize)] pub struct DeclareMetricRequest { pub name: String, pub message_match_regex: String, #[serde(default)] pub label_params: Vec, #[serde(default)] pub value_param: Option, #[serde(default)] pub matched_labels: Option>, #[serde(default)] pub unmatched_labels: Option>, } #[allow(dead_code)] pub enum EngineQuery { Logs(LogsParams), Timeline(TimelineParams), Stats, ParamDeclarations, MetricDeclarations, BucketValues { metric_name: String, labels_filter: Option>, start_ms: Option, end_ms: Option, since_sequence: Option, limit: Option, }, Promql { query: String, start_ms: Option, end_ms: Option }, } #[allow(dead_code)] pub enum EngineMessage { Ingest(IngestedLine), IngestMetricSample(MetricSample), DeclareParam { request: DeclareParamRequest, respond_to: oneshot::Sender>, }, RemoveParamDeclaration { name: String, respond_to: oneshot::Sender, }, ClearParams { respond_to: oneshot::Sender, }, DeclareMetric { request: DeclareMetricRequest, respond_to: oneshot::Sender>, }, UpdateMetric { name: String, request: DeclareMetricRequest, respond_to: oneshot::Sender>, }, RemoveMetricDeclaration { name: String, respond_to: oneshot::Sender, }, SetMaxLogs(Option), EnsureSortHelperIndex { param_name: String, respond_to: oneshot::Sender<()>, }, Barrier(oneshot::Sender<()>), } #[allow(dead_code)] pub enum EngineResponse { Logs(Vec), Timeline(Vec), Stats(StatsDto), ParamDeclarations(Vec), MetricDeclarations(Vec), BucketValues(Vec), Promql(Result, String>), Error(String), } #[derive(Debug, Clone, Default)] pub(crate) struct EngineSnapshot { index: BTreeMap, sort_helper_indexes: HashMap, log_count: usize, approx_bytes_total: u64, ingest_lag_high_watermark_ms: i64, } struct EngineOp(Box); impl Absorb for EngineSnapshot { fn absorb_first(&mut self, operation: &mut EngineOp, _other: &Self) { (operation.0)(self); } fn drop_first(self: Box) {} fn sync_with(&mut self, first: &Self) { *self = first.clone(); } } pub struct EngineHandle { pub tx: mpsc::Sender, pub reader_factory: left_right::ReadHandleFactory, } impl Clone for EngineHandle { fn clone(&self) -> Self { EngineHandle { tx: self.tx.clone(), reader_factory: self.reader_factory.clone() } } } impl EngineHandle { pub async fn barrier(&self) { let (respond_to, response) = oneshot::channel(); if self.tx.send(EngineMessage::Barrier(respond_to)).await.is_ok() { let _ = response.await; } } } pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle { let (tx, rx) = mpsc::channel(channel_capacity); let (write, read) = left_right::new::(); let reader_factory = read.factory(); tokio::spawn(run(rx, settings, write)); EngineHandle { tx, reader_factory } } #[cfg(test)] pub(crate) fn test_handle(channel_capacity: usize) -> (EngineHandle, mpsc::Receiver) { let (tx, rx) = mpsc::channel(channel_capacity); let (_write, read) = left_right::new::(); (EngineHandle { tx, reader_factory: read.factory() }, rx) } async fn run( mut rx: mpsc::Receiver, settings: SettingsHandle, mut write: left_right::WriteHandle, ) { let mut max_logs: Option = None; let mut metric_sample_sequence: u64 = 0; let mut alert_states: HashMap = HashMap::new(); let mut log_stream_header: Option = None; let mut ingest_lag_samples: Vec<(Timestamp, i64)> = Vec::new(); while let Some(message) = rx.recv().await { match message { EngineMessage::Ingest(line) => { let IngestedLine { sequence, timestamp, text } = line; let processed_at = Timestamp::now(); record_ingest_lag( &mut ingest_lag_samples, processed_at, (processed_at.as_millis() - timestamp.as_millis()).max(0), ); let ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms(&ingest_lag_samples, processed_at); let is_new_header = crate::logfmt::detect_log_stream_header(&text).is_some(); log_stream_header = log_stream_header_transition(&log_stream_header, &text); let row_header = if is_new_header { None } else { log_stream_header.clone() }; let (rules, ntfy_topic, ntfy_server) = { let settings = settings.lock().unwrap(); (settings.alert_rules.clone(), settings.ntfy_topic.clone(), settings.ntfy_server.clone()) }; let text_for_op = text.clone(); write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { snap.approx_bytes_total += approx_log_bytes(&text_for_op); snap.index.insert( IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text: text_for_op.clone() }, ); snap.log_count += 1; if let Some(header) = row_header.as_ref() { derive_log_stream_params_from_row(&mut snap.index, header, &text_for_op); } auto_declare_params(&mut snap.index, &text_for_op); apply_all_metrics(&mut snap.index, sequence, timestamp, &text_for_op); update_sort_helper_indexes_on_ingest( &mut snap.sort_helper_indexes, &snap.index, sequence, timestamp, &text_for_op, ); purge_stale_sort_helper_indexes(&mut snap.sort_helper_indexes); evict_excess_logs(&mut snap.index, &mut snap.log_count, max_logs, &mut snap.approx_bytes_total); snap.ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms; }))); write.publish(); let fired = write .enter() .map(|guard| { alerts::evaluate_alerts(&rules, &mut alert_states, &guard.index, sequence, timestamp, &text) }) .unwrap_or_default(); for (rule, message) in fired { tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone())); } } EngineMessage::IngestMetricSample(sample) => { let labels: BTreeMap = sample .labels .into_iter() .map(|(name, value)| (name, MetricPrimitive::String(value))) .collect(); let metric_name = sample.metric_name; let sequence = metric_sample_sequence; let timestamp = sample.timestamp; let value = MetricPrimitive::Number(Number::Float(sample.value)); write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { push_bucket_value(&mut snap.index, &metric_name, labels.clone(), sequence, timestamp, value.clone()); }))); write.publish(); metric_sample_sequence += 1; } EngineMessage::SetMaxLogs(new_max) => { max_logs = new_max; write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { evict_excess_logs(&mut snap.index, &mut snap.log_count, new_max, &mut snap.approx_bytes_total); }))); write.publish(); } EngineMessage::DeclareParam { request, respond_to } => { let result = write .enter() .map(|guard| validate_param_declaration(&guard.index, &request)) .unwrap_or_else(|| Err("engine is shutting down".to_string())); let succeeded = result.is_ok(); let name = request.name.clone(); let request_for_op = request.clone(); write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { snap.sort_helper_indexes.remove(&name); if succeeded { insert_param_declaration(&mut snap.index, request_for_op.clone()); } }))); write.publish(); let _ = respond_to.send(result); } EngineMessage::RemoveParamDeclaration { name, respond_to } => { let existed = write .enter() .is_some_and(|guard| guard.index.contains_key(&IndexKey::ParamDeclaration { name: name.clone() })); let name_for_op = name.clone(); write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() }); snap.sort_helper_indexes.remove(&name_for_op); }))); write.publish(); let _ = respond_to.send(existed); } EngineMessage::ClearParams { respond_to } => { let mut scratch = write.enter().map(|guard| guard.index.clone()).unwrap_or_default(); let removed = clear_all_param_declarations(&mut scratch); let (reextracted, logs_considered) = reextract_params_from_retained_logs(&mut scratch); write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { clear_all_param_declarations(&mut snap.index); reextract_params_from_retained_logs(&mut snap.index); snap.sort_helper_indexes.clear(); }))); write.publish(); let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered }); } EngineMessage::DeclareMetric { request, respond_to } => { let result = write .enter() .map(|guard| validate_metric_request(&guard.index, &request).map(|_| ())) .unwrap_or_else(|| Err("engine is shutting down".to_string())); let succeeded = result.is_ok(); let request_for_op = request.clone(); write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { if succeeded { let _ = declare_metric(&mut snap.index, request_for_op.clone()); } }))); write.publish(); let _ = respond_to.send(result); } EngineMessage::UpdateMetric { name, request, respond_to } => { let result = write .enter() .map(|guard| validate_update_metric(&guard.index, &name, &request).map(|_| ())) .unwrap_or_else(|| Err("engine is shutting down".to_string())); let succeeded = result.is_ok(); let existing_name = name.clone(); let request_for_op = request.clone(); write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { if succeeded { let _ = update_metric(&mut snap.index, &existing_name, request_for_op.clone()); } }))); write.publish(); let _ = respond_to.send(result); } EngineMessage::RemoveMetricDeclaration { name, respond_to } => { let existed = write .enter() .is_some_and(|guard| guard.index.contains_key(&IndexKey::MetricDeclaration { name: name.clone() })); let name_for_op = name.clone(); write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { remove_metric_declaration(&mut snap.index, &name_for_op); }))); write.publish(); let _ = respond_to.send(existed); } EngineMessage::EnsureSortHelperIndex { param_name, respond_to } => { write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { ensure_sort_helper_index_built(&mut snap.sort_helper_indexes, &snap.index, ¶m_name); }))); write.publish(); let _ = respond_to.send(()); } EngineMessage::Barrier(respond_to) => { let _ = respond_to.send(()); } } } } fn validate_param_declaration( index: &BTreeMap, 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)) = 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(()) } fn insert_param_declaration(index: &mut BTreeMap, request: DeclareParamRequest) { index.insert( IndexKey::ParamDeclaration { name: request.name.clone() }, DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }), ); } #[cfg(test)] fn declare_param( index: &mut BTreeMap, request: DeclareParamRequest, ) -> Result<(), String> { validate_param_declaration(index, &request)?; insert_param_declaration(index, request); Ok(()) } fn sanitize_group_name(key: &str) -> String { let mut group: String = key.chars().map(|c| if c.is_ascii_alphanumeric() || c == '_' { c } else { '_' }).collect(); if group.chars().next().is_none_or(|c| c.is_ascii_digit()) { group.insert(0, '_'); } group } fn infer_kind(sample: &str) -> MetricPrimitiveKind { if sample.parse::().is_ok() || sample.parse::().is_ok() { MetricPrimitiveKind::Number } else { MetricPrimitiveKind::String } } fn build_auto_param_pattern(key: &str, group: &str, quoted: bool) -> String { let escaped_key = regex::escape(key); if quoted { format!(r#"{escaped_key}="(?P<{group}>[^"]*)""#) } else { format!(r"{escaped_key}=(?P<{group}>\S+)") } } fn auto_derived_param_count(index: &BTreeMap) -> usize { param_declarations_range(index).filter(|declaration| declaration.6).count() } struct AutoParamCandidate { name: String, value: String, message_match_regex: String, capture_group: String, json_field_path: Option, } fn auto_declare_params(index: &mut BTreeMap, text: &str) { let mut total = auto_derived_param_count(index); if total >= MAX_AUTO_DECLARED_PARAMS { return; } let candidates: Vec = if crate::logfmt::detect_format(text) == "json" { crate::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::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 index.contains_key(&index_key) { continue; } 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; } } fn log_stream_header_transition( current: &Option, text: &str, ) -> Option { crate::logfmt::detect_log_stream_header(text).or_else(|| current.clone()) } fn derive_log_stream_params_from_row( index: &mut BTreeMap, header: &crate::logfmt::LogStreamHeader, text: &str, ) { if !crate::logfmt::is_log_stream_line(text) { return; } let row = crate::logfmt::split_log_stream_row(text, header); for (i, column) in header.columns.iter().enumerate() { let Some(value) = row.get(i) else { continue }; let index_key = IndexKey::ParamDeclaration { name: column.name.clone() }; if index.contains_key(&index_key) { continue; } index.insert( index_key, DataValue::ParamDeclaration(ParamDeclaration { name: column.name.clone(), message_match_regex: crate::logfmt::log_stream_column_pattern(header, i, &column.name), capture_group: column.name.clone(), json_field_path: None, kind: infer_kind(value.trim()), elapsed_since_last: false, auto_derived: true, log_stream_gated: true, }), ); } if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) { index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| { DataValue::ParamDeclaration(ParamDeclaration { name: "message".to_string(), message_match_regex: crate::logfmt::log_stream_message_pattern(header, "message"), capture_group: "message".to_string(), json_field_path: None, kind: infer_kind(message.trim()), elapsed_since_last: false, auto_derived: true, log_stream_gated: true, }) }); } } #[cfg(test)] fn auto_declare_log_stream_params( index: &mut BTreeMap, header: &mut Option, text: &str, ) { let is_new_header = crate::logfmt::detect_log_stream_header(text).is_some(); *header = log_stream_header_transition(header, text); if is_new_header { return; } if let Some(header) = header.as_ref() { derive_log_stream_params_from_row(index, header, text); } } fn clear_all_param_declarations(index: &mut BTreeMap) -> usize { let names: Vec = param_declarations_range(index).map(|(name, ..)| name.to_string()).collect(); let removed = names.len(); for name in names { index.remove(&IndexKey::ParamDeclaration { name }); } removed } fn reextract_params_from_retained_logs(index: &mut BTreeMap) -> (usize, usize) { let texts: Vec = logs_range(index).map(|(_, _, text)| text.to_string()).collect(); let logs_considered = texts.len(); for text in &texts { auto_declare_params(index, text); } let reextracted = param_declarations_range(index).count(); (reextracted, logs_considered) } fn validate_metric_request( index: &BTreeMap, request: &DeclareMetricRequest, ) -> Result { let re = Regex::new(&request.message_match_regex) .map_err(|err| format!("invalid regex: {err}"))?; let declared = |name: &str| 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) } fn declare_metric( index: &mut BTreeMap, request: DeclareMetricRequest, ) -> Result<(), String> { let re = validate_metric_request(index, &request)?; let resolved_params = resolved_param_declarations(index); backfill_metric( index, &request.name, &re, &request.label_params, request.value_param.as_deref(), &resolved_params, request.matched_labels.as_ref(), request.unmatched_labels.as_ref(), ); 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(()) } fn validate_update_metric( index: &BTreeMap, existing_name: &str, request: &DeclareMetricRequest, ) -> Result { if !index.contains_key(&IndexKey::MetricDeclaration { name: existing_name.to_string() }) { return Err(format!("no metric declaration named '{existing_name}'")); } if request.name != existing_name && index.contains_key(&IndexKey::MetricDeclaration { name: request.name.clone() }) { return Err(format!("a metric named '{}' already exists", request.name)); } validate_metric_request(index, request) } fn update_metric( index: &mut BTreeMap, existing_name: &str, request: DeclareMetricRequest, ) -> Result<(), String> { let re = validate_update_metric(index, existing_name, &request)?; let resolved_params = resolved_param_declarations(index); clear_metric_data(index, existing_name); if request.name != existing_name { index.remove(&IndexKey::MetricDeclaration { name: existing_name.to_string() }); } backfill_metric( index, &request.name, &re, &request.label_params, request.value_param.as_deref(), &resolved_params, request.matched_labels.as_ref(), request.unmatched_labels.as_ref(), ); 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(index: &mut BTreeMap, metric_name: &str) { index.retain(|key, _| match key { IndexKey::Bucket { metric_name: name, .. } => name != metric_name, IndexKey::LastMetricEvent { metric_name: name, .. } => name != metric_name, _ => true, }); } fn remove_metric_declaration(index: &mut BTreeMap, name: &str) -> bool { let existed = index.remove(&IndexKey::MetricDeclaration { name: name.to_string() }).is_some(); clear_metric_data(index, name); existed } fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Option { match kind { MetricPrimitiveKind::String => Some(MetricPrimitive::String(raw.to_string())), MetricPrimitiveKind::Number => raw .parse::() .map(Number::Int) .or_else(|_| raw.parse::().map(Number::Float)) .ok() .map(MetricPrimitive::Number), MetricPrimitiveKind::Timestamp => { crate::filter_datetime::parse_datetime(raw, Timestamp::now()) .ok() .map(MetricPrimitive::Timestamp) } MetricPrimitiveKind::Duration => raw.parse::().ok().map(MetricPrimitive::Duration), MetricPrimitiveKind::Boolean => raw.parse::().ok().map(MetricPrimitive::Boolean), } } struct ResolvedParam { regex: Regex, capture_group: String, json_field_path: Option, kind: MetricPrimitiveKind, elapsed_since_last: bool, log_stream_gated: bool, } fn resolved_param_declarations(index: &BTreeMap) -> HashMap { param_declarations_range(index) .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| { Regex::new(message_match_regex).ok().map(|regex| { ( name.to_string(), ResolvedParam { regex, capture_group: capture_group.to_string(), json_field_path: json_field_path.map(|path| path.to_string()), kind, elapsed_since_last, log_stream_gated, }, ) }) }) .collect() } fn extract_param_raw( text: &str, regex: &Regex, capture_group: &str, json_field_path: Option<&str>, log_stream_gated: bool, ) -> Option { if log_stream_gated && !crate::logfmt::is_log_stream_line(text) { return None; } match json_field_path.filter(|path| !path.is_empty()) { Some(field_path) => crate::data_value::extract_json_field(text, field_path), None => regex.captures(text)?.name(capture_group).map(|m| m.as_str().to_string()), } } fn resolve_param_value( index: &mut BTreeMap, 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 index.get(&last_event_key) { Some(DataValue::LastEventTimestamp { timestamp: previous }) => Some(*previous), _ => None, }; 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) } #[allow(clippy::too_many_arguments)] fn apply_metric_to_log( index: &mut BTreeMap, 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(); push_bucket_value(index, 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) = resolve_param_value(index, 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 resolve_param_value(index, metric_name, resolved, &labels, text, timestamp) { Some(parsed) => parsed, None => return, } } None => MetricPrimitive::Number(Number::Int(1)), }; push_bucket_value(index, metric_name, labels, sequence, timestamp, value); } fn push_bucket_value( index: &mut BTreeMap, metric_name: &str, labels: BTreeMap, sequence: u64, timestamp: Timestamp, value: MetricPrimitive, ) { let key = IndexKey::Bucket { metric_name: metric_name.to_string(), labels }; let entry = 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( index: &mut BTreeMap, 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)> = logs_range(index) .map(|(sequence, timestamp, text)| (sequence, timestamp, text.to_string())) .collect(); for (sequence, timestamp, text) in lines { apply_metric_to_log( index, metric_name, regex, label_params, value_param, resolved_params, matched_labels, unmatched_labels, sequence, timestamp, &text, ); } } fn apply_all_metrics( index: &mut BTreeMap, sequence: u64, timestamp: Timestamp, text: &str, ) { type SeededMetric = ( String, String, Vec, Option, Option>, Option>, ); let metrics: Vec = metric_declarations_range(index) .map(|(name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels)| { ( name.to_string(), message_match_regex.to_string(), label_params.to_vec(), value_param.map(|s| s.to_string()), matched_labels.cloned(), unmatched_labels.cloned(), ) }) .collect(); let resolved_params = resolved_param_declarations(index); for (name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels) in metrics { if let Ok(re) = Regex::new(&message_match_regex) { apply_metric_to_log( index, &name, &re, &label_params, value_param.as_deref(), &resolved_params, matched_labels.as_ref(), unmatched_labels.as_ref(), sequence, timestamp, text, ); } } } fn bucket_values<'a>( index: &'a BTreeMap, metric_name: &'a str, labels_filter: Option<&'a BTreeMap>, ) -> impl Iterator, &'a (u64, Timestamp, MetricPrimitive))> + 'a { index .range(IndexKey::MetricBucketSplit..IndexKey::End) .filter_map(move |(key, value)| match (key, value) { (IndexKey::Bucket { metric_name: name, labels }, DataValue::BucketValue { values }) if name == metric_name && labels_filter.is_none_or(|filter| filter == labels) => { Some(values.iter().map(move |entry| (labels, entry))) } _ => None, }) .flatten() } enum CompiledFilter { Regex(Regex), Sequence { op: CompareOp, target: f64 }, Timestamp { op: CompareOp, target_ms: i64 }, Fmt { op: CompareOp, target: String }, MetricMatch(HashSet), Json { field: String, op: CompareOp, value: FilterValue }, And(Box, Box), Or(Box, Box), } fn compile_filter_query( index: &BTreeMap, filter: &str, now: Timestamp, ) -> Result { let expr = filter_parser::parse(filter).map_err(|err| err.to_string())?; compile_filter(index, &expr, now) } fn compile_filter( index: &BTreeMap, 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" => compile_ottyl_seq_comparison(*op, value), "ottyl_ts" => compile_ottyl_ts_comparison(*op, value, now), "ottyl_fmt" => Ok(compile_ottyl_fmt_comparison(*op, value)), other => compile_metric_comparison(index, 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(compile_filter(index, lhs, now)?), Box::new(compile_filter(index, rhs, now)?), )), FilterExpr::Or(lhs, rhs) => Ok(CompiledFilter::Or( Box::new(compile_filter(index, lhs, now)?), Box::new(compile_filter(index, rhs, now)?), )), } } fn compile_ottyl_seq_comparison(op: CompareOp, value: &FilterValue) -> Result { let target = match value { FilterValue::Number(n) => *n, FilterValue::Raw(raw) => raw.parse::().map_err(|_| { format!("'ottyl_seq' comparisons require a numeric value, got {raw:?}") })?, }; Ok(CompiledFilter::Sequence { op, target }) } fn compile_ottyl_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) .map_err(|err| err.to_string())? .as_millis(), }; Ok(CompiledFilter::Timestamp { op, target_ms }) } fn compile_ottyl_fmt_comparison(op: CompareOp, value: &FilterValue) -> CompiledFilter { let target = match value { FilterValue::Number(n) => n.to_string(), FilterValue::Raw(raw) => raw.clone(), }; CompiledFilter::Fmt { op, target } } fn compile_metric_comparison( index: &BTreeMap, 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, .. }) = 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 bucket_values(index, metric, None) { let Some(primitive) = labels.get(param) else { continue; }; if compare_metric_primitive(primitive, op, value, now)? { matching.insert(*sequence); } } Ok(CompiledFilter::MetricMatch(matching)) } fn compare_metric_primitive( primitive: &MetricPrimitive, op: CompareOp, value: &FilterValue, now: Timestamp, ) -> Result { let ordering = match (primitive, value) { (MetricPrimitive::Number(n), FilterValue::Number(v)) => n.as_f64().partial_cmp(v), (MetricPrimitive::Number(n), FilterValue::Raw(raw)) => { let v: f64 = raw .parse() .map_err(|_| format!("expected a numeric value, got {raw:?}"))?; n.as_f64().partial_cmp(&v) } (MetricPrimitive::Timestamp(ts), FilterValue::Number(v)) => { ts.as_millis().partial_cmp(&(*v as i64)) } (MetricPrimitive::Timestamp(ts), FilterValue::Raw(raw)) => { let target = filter_datetime::parse_datetime(raw, now).map_err(|err| err.to_string())?; ts.as_millis().partial_cmp(&target.as_millis()) } (MetricPrimitive::String(s), FilterValue::Raw(raw)) => s.as_str().partial_cmp(raw.as_str()), (MetricPrimitive::String(s), FilterValue::Number(n)) => { s.as_str().partial_cmp(n.to_string().as_str()) } (MetricPrimitive::Duration(ms), FilterValue::Number(v)) => (*ms as f64).partial_cmp(v), (MetricPrimitive::Duration(ms), FilterValue::Raw(raw)) => { let v: f64 = raw .parse() .map_err(|_| format!("expected a numeric value, got {raw:?}"))?; (*ms as f64).partial_cmp(&v) } (MetricPrimitive::Boolean(b), FilterValue::Raw(raw)) => { let v: bool = raw .parse() .map_err(|_| format!("expected true or false, got {raw:?}"))?; b.partial_cmp(&v) } (MetricPrimitive::Boolean(_), FilterValue::Number(_)) => None, }; Ok(ordering.is_some_and(|ordering| apply_compare_op(ordering, op))) } pub(crate) fn eval_json_comparison(text: &str, field: &str, op: CompareOp, value: &FilterValue) -> bool { let Ok(parsed) = serde_json::from_str::(text) else { return false; }; let Some(field_value) = parsed.as_object().and_then(|obj| obj.get(field)) else { return false; }; compare_json_value(field_value, op, value) } fn compare_json_value(field_value: &serde_json::Value, op: CompareOp, value: &FilterValue) -> bool { match field_value { serde_json::Value::Number(n) => { let target = match value { FilterValue::Number(v) => Some(*v), FilterValue::Raw(raw) => raw.parse::().ok(), }; match (n.as_f64(), target) { (Some(f), Some(v)) => f.partial_cmp(&v).is_some_and(|ordering| apply_compare_op(ordering, op)), _ => false, } } serde_json::Value::String(s) => { let target = match value { FilterValue::Raw(raw) => raw.clone(), FilterValue::Number(n) => n.to_string(), }; s.as_str() .partial_cmp(target.as_str()) .is_some_and(|ordering| apply_compare_op(ordering, op)) } serde_json::Value::Bool(b) => { let FilterValue::Raw(raw) = value else { return false; }; match raw.parse::() { Ok(v) => match op { CompareOp::Eq => *b == v, CompareOp::Ne => *b != v, _ => false, }, Err(_) => false, } } serde_json::Value::Null | serde_json::Value::Array(_) | serde_json::Value::Object(_) => false, } } pub(crate) fn apply_compare_op(ordering: Ordering, op: CompareOp) -> bool { match op { CompareOp::Lt => ordering == Ordering::Less, CompareOp::Gt => ordering == Ordering::Greater, CompareOp::Le => ordering != Ordering::Greater, CompareOp::Ge => ordering != Ordering::Less, CompareOp::Eq => ordering == Ordering::Equal, CompareOp::Ne => ordering != Ordering::Equal, } } fn eval_compiled(compiled: &CompiledFilter, sequence: u64, timestamp: Timestamp, text: &str) -> bool { match compiled { CompiledFilter::Regex(re) => re.is_match(text), CompiledFilter::Sequence { op, target } => { apply_compare_op((sequence as f64).total_cmp(target), *op) } CompiledFilter::Timestamp { op, target_ms } => { apply_compare_op(timestamp.as_millis().cmp(target_ms), *op) } CompiledFilter::Fmt { op, target } => { apply_compare_op(crate::logfmt::detect_format(text).cmp(target.as_str()), *op) } CompiledFilter::MetricMatch(sequences) => sequences.contains(&sequence), CompiledFilter::Json { field, op, value } => eval_json_comparison(text, field, *op, value), CompiledFilter::And(lhs, rhs) => { eval_compiled(lhs, sequence, timestamp, text) && eval_compiled(rhs, sequence, timestamp, text) } CompiledFilter::Or(lhs, rhs) => { eval_compiled(lhs, sequence, timestamp, text) || eval_compiled(rhs, sequence, timestamp, text) } } } fn metric_declarations_range( index: &BTreeMap, ) -> impl Iterator< Item = ( &str, &str, &[String], Option<&str>, Option<&BTreeMap>, Option<&BTreeMap>, ), > { index .range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit) .filter_map(|(key, value)| match (key, value) { ( IndexKey::MetricDeclaration { name }, DataValue::MetricDeclaration { message_match_regex, label_params, value_param, matched_labels, unmatched_labels, }, ) => Some(( name.as_str(), message_match_regex.as_str(), label_params.as_slice(), value_param.as_deref(), matched_labels.as_ref(), unmatched_labels.as_ref(), )), _ => None, }) } fn param_declarations_range( index: &BTreeMap, ) -> impl Iterator, MetricPrimitiveKind, bool, bool, bool)> { 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, }) } #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] enum SortHelperKey { WithParam(MetricPrimitive, u64), WithoutParam(u64), } #[derive(Debug, Clone)] struct SortHelperIndexEntry { index: BTreeMap, last_accessed: std::time::Instant, } type CompiledParamDeclaration = (String, Regex, String, Option, MetricPrimitiveKind, bool); fn derive_single_param( text: &str, declarations: &[CompiledParamDeclaration], param_name: &str, ) -> Option { let (_, regex, capture_group, json_field_path, kind, log_stream_gated) = declarations.iter().find(|(name, ..)| name == param_name)?; if *log_stream_gated && !crate::logfmt::is_log_stream_line(text) { return None; } if *kind == MetricPrimitiveKind::Boolean { return Some(MetricPrimitive::Boolean(regex.is_match(text))); } let captures = regex.captures(text)?; let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) { Some(field_path) => crate::data_value::extract_json_field(text, field_path)?, None => captures.name(capture_group)?.as_str().to_string(), }; parse_metric_primitive(&raw, *kind) } fn build_sort_helper_index(index: &BTreeMap, sort_by: &str) -> BTreeMap { let compiled_params = compiled_param_declarations(index); logs_range(index) .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() } fn update_sort_helper_indexes_on_ingest( sort_helper_indexes: &mut HashMap, index: &BTreeMap, sequence: u64, timestamp: Timestamp, text: &str, ) { if sort_helper_indexes.is_empty() { return; } let compiled_params = compiled_param_declarations(index); for (sort_by, entry) in 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); } } fn purge_stale_sort_helper_indexes(sort_helper_indexes: &mut HashMap) { sort_helper_indexes.retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER); } fn ensure_sort_helper_index_built( sort_helper_indexes: &mut HashMap, index: &BTreeMap, sort_by: &str, ) { let entry = sort_helper_indexes.entry(sort_by.to_string()).or_insert_with(|| SortHelperIndexEntry { index: build_sort_helper_index(index, sort_by), last_accessed: std::time::Instant::now(), }); entry.last_accessed = std::time::Instant::now(); } fn compiled_param_declarations(index: &BTreeMap) -> Vec { param_declarations_range(index) .filter(|(_, _, _, _, _, elapsed_since_last, _, _)| !elapsed_since_last) .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _, _, log_stream_gated)| { Regex::new(message_match_regex).ok().map(|re| { ( name.to_string(), re, capture_group.to_string(), json_field_path.map(|path| path.to_string()), kind, log_stream_gated, ) }) }) .collect() } 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 { if *log_stream_gated && !crate::logfmt::is_log_stream_line(text) { continue; } if *kind == MetricPrimitiveKind::Boolean { params.insert(name.clone(), MetricPrimitive::Boolean(regex.is_match(text))); continue; } let Some(captures) = regex.captures(text) else { continue; }; let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) { Some(field_path) => match crate::data_value::extract_json_field(text, field_path) { Some(value) => value, None => continue, }, None => match captures.name(capture_group) { Some(m) => m.as_str().to_string(), None => continue, }, }; let Some(value) = parse_metric_primitive(&raw, *kind) else { continue; }; params.insert(name.clone(), value); } params } fn log_entry_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap { let mut params = derive_log_params(text, declarations); params.insert("ottyl_fmt".to_string(), MetricPrimitive::String(crate::logfmt::detect_format(text).to_string())); params } fn approx_log_bytes(text: &str) -> u64 { text.len() as u64 } fn record_ingest_lag(samples: &mut Vec<(Timestamp, i64)>, now: Timestamp, lag_ms: i64) { samples.retain(|(recorded_at, _)| now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS); if samples.len() < INGEST_LAG_WATERMARK_MAX_ENTRIES { samples.push((now, lag_ms)); return; } let Some((min_index, &(_, min_lag))) = samples.iter().enumerate().min_by_key(|(_, (_, lag))| *lag) else { return; }; if lag_ms > min_lag { samples[min_index] = (now, lag_ms); } } fn ingest_lag_high_watermark_ms(samples: &[(Timestamp, i64)], now: Timestamp) -> i64 { samples .iter() .filter(|(recorded_at, _)| now.as_millis() - recorded_at.as_millis() < INGEST_LAG_WATERMARK_WINDOW_MS) .map(|(_, lag)| *lag) .max() .unwrap_or(0) } fn evict_excess_logs( index: &mut BTreeMap, log_count: &mut usize, max_logs: Option, approx_bytes_total: &mut u64, ) { let Some(max_logs) = max_logs else { return }; while *log_count > max_logs { let Some(oldest_key) = index .range(IndexKey::Start..IndexKey::LogParamSplit) .next() .map(|(key, _)| key.clone()) else { break; }; if let Some(DataValue::LogLiteral { text }) = index.get(&oldest_key) { *approx_bytes_total = approx_bytes_total.saturating_sub(approx_log_bytes(text)); } index.remove(&oldest_key); *log_count -= 1; } } fn logs_range(index: &BTreeMap) -> impl Iterator { index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| { match (key, value) { (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => { Some((*sequence, *timestamp, text.as_str())) } _ => None, } }) } pub(crate) fn logs_since( index: &BTreeMap, before_sequence: u64, since_ms: i64, now_ms: i64, ) -> impl Iterator { logs_range(index).filter(move |(sequence, timestamp, _)| { *sequence < before_sequence && timestamp.as_millis() > since_ms && timestamp.as_millis() <= now_ms }) } enum SortHelperAccess<'a> { #[cfg(test)] Mutable(&'a mut HashMap), Immutable(&'a HashMap), } impl<'a> SortHelperAccess<'a> { #[cfg_attr(not(test), allow(unused_variables))] fn resolve( self, index: &BTreeMap, sort_by: &str, ) -> Option<&'a BTreeMap> { match self { #[cfg(test)] SortHelperAccess::Mutable(map) => { ensure_sort_helper_index_built(map, index, sort_by); map.get(sort_by).map(|entry| &entry.index) } SortHelperAccess::Immutable(map) => map.get(sort_by).map(|entry| &entry.index), } } } #[cfg(test)] fn handle_query( index: &BTreeMap, query: EngineQuery, approx_bytes_total: u64, ingest_lag_high_watermark_ms: i64, sort_helper_indexes: &mut HashMap, ) -> EngineResponse { handle_query_inner( index, query, approx_bytes_total, ingest_lag_high_watermark_ms, SortHelperAccess::Mutable(sort_helper_indexes), ) } pub(crate) fn handle_query_ro(snapshot: &EngineSnapshot, query: EngineQuery) -> EngineResponse { handle_query_inner( &snapshot.index, query, snapshot.approx_bytes_total, snapshot.ingest_lag_high_watermark_ms, SortHelperAccess::Immutable(&snapshot.sort_helper_indexes), ) } pub(crate) fn query_sort_by(query: &EngineQuery) -> Option<&str> { match query { EngineQuery::Logs(params) => params.sort_by.as_deref(), _ => None, } } pub(crate) fn has_sort_helper_index(snapshot: &EngineSnapshot, sort_by: &str) -> bool { snapshot.sort_helper_indexes.contains_key(sort_by) } fn handle_query_inner( index: &BTreeMap, query: EngineQuery, approx_bytes_total: u64, ingest_lag_high_watermark_ms: i64, sort_helper_access: SortHelperAccess<'_>, ) -> EngineResponse { match query { EngineQuery::Logs(params) => { let compiled_filter = match params.filter.as_deref() { Some(filter) => match compile_filter_query(index, filter, Timestamp::now()) { Ok(compiled) => Some(compiled), Err(err) => return EngineResponse::Error(err), }, None => None, }; if let Some(sort_by) = params.sort_by.as_deref() { if !param_declarations_range(index).any(|(name, ..)| name == sort_by) { return EngineResponse::Error(format!("unknown param '{sort_by}' for sort_by")); } } let limit = params.limit.unwrap_or(500).min(10_000); let compiled_params = compiled_param_declarations(index); let filtered = logs_range(index) .filter(|(sequence, timestamp, _)| { params.since_sequence.is_none_or(|since| *sequence > since) && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start) && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end) }) .filter(|(sequence, timestamp, text)| { compiled_filter .as_ref() .is_none_or(|filter| eval_compiled(filter, *sequence, *timestamp, text)) }); let entries: Vec = match params.sort_by.as_deref() { None => filtered .take(limit) .map(|(sequence, timestamp, text)| LogEntryDto { sequence, timestamp_ms: timestamp.as_millis(), text: text.to_string(), params: log_entry_params(text, &compiled_params), }) .collect(), Some(sort_by) => { static EMPTY_HELPER: std::sync::LazyLock> = std::sync::LazyLock::new(BTreeMap::new); let helper_index = sort_helper_access.resolve(index, sort_by).unwrap_or(&EMPTY_HELPER); let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true); let mut entries = Vec::new(); for (helper_key, timestamp) in helper_index.iter() { let sequence = match helper_key { SortHelperKey::WithParam(_, sequence) => *sequence, SortHelperKey::WithoutParam(sequence) => { if !sort_missing_fallback { continue; } *sequence } }; if !(params.since_sequence.is_none_or(|since| sequence > since) && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start) && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)) { continue; } let Some(DataValue::LogLiteral { text }) = index.get(&IndexKey::IngestedLog { sequence, timestamp: *timestamp }) else { continue; }; if !compiled_filter .as_ref() .is_none_or(|filter| eval_compiled(filter, sequence, *timestamp, text)) { continue; } entries.push(LogEntryDto { sequence, timestamp_ms: timestamp.as_millis(), text: text.clone(), params: log_entry_params(text, &compiled_params), }); if entries.len() >= limit { break; } } entries } }; let entries: Vec = entries.into_iter().take(limit).collect(); EngineResponse::Logs(entries) } EngineQuery::Timeline(params) => { let compiled_filter = match params.filter.as_deref() { Some(filter) => match compile_filter_query(index, filter, Timestamp::now()) { Ok(compiled) => Some(compiled), Err(err) => return EngineResponse::Error(err), }, None => None, }; let bucket_ms = params.bucket_ms.unwrap_or(DEFAULT_TIMELINE_BUCKET_MS).max(1); let mut buckets: BTreeMap = BTreeMap::new(); for (sequence, timestamp, text) in logs_range(index).filter(|(_, timestamp, _)| { params.start_ms.is_none_or(|start| timestamp.as_millis() >= start) && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end) }) { if compiled_filter .as_ref() .is_some_and(|filter| !eval_compiled(filter, sequence, timestamp, text)) { continue; } let bucket_start = (timestamp.as_millis() / bucket_ms) * bucket_ms; *buckets.entry(bucket_start).or_insert(0) += 1; } let out = buckets .into_iter() .map(|(start_ms, count)| TimelineBucketDto { start_ms, count }) .collect(); EngineResponse::Timeline(out) } EngineQuery::Stats => { let mut log_count = 0u64; let mut first_sequence = None; let mut last_sequence = None; let mut first_timestamp_ms = None; let mut last_timestamp_ms = None; for (sequence, timestamp, _) in logs_range(index) { log_count += 1; first_sequence.get_or_insert(sequence); last_sequence = Some(sequence); first_timestamp_ms.get_or_insert(timestamp.as_millis()); last_timestamp_ms = Some(timestamp.as_millis()); } EngineResponse::Stats(StatsDto { log_count, first_sequence, last_sequence, first_timestamp_ms, last_timestamp_ms, approx_bytes_total, ingest_lag_high_watermark_ms, }) } EngineQuery::ParamDeclarations => { let declarations = param_declarations_range(index) .map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, auto_derived, _)| { ParamDeclarationDto { name: name.to_string(), message_match_regex: message_match_regex.to_string(), capture_group: capture_group.to_string(), json_field_path: json_field_path.map(|path| path.to_string()), kind, elapsed_since_last, auto_derived, reserved: false, } }) .collect(); EngineResponse::ParamDeclarations(declarations) } EngineQuery::MetricDeclarations => { let declarations = metric_declarations_range(index) .map(|(name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels)| { MetricDeclarationDto { name: name.to_string(), message_match_regex: message_match_regex.to_string(), label_params: label_params.to_vec(), value_param: value_param.map(|s| s.to_string()), matched_labels: matched_labels.cloned(), unmatched_labels: unmatched_labels.cloned(), } }) .collect(); EngineResponse::MetricDeclarations(declarations) } EngineQuery::BucketValues { metric_name, labels_filter, start_ms, end_ms, since_sequence, limit, } => { let limit = limit.unwrap_or(500).min(10_000); let values: Vec = bucket_values(index, &metric_name, None) .filter(|(labels, _)| { labels_filter.as_ref().is_none_or(|filter| { filter.iter().all(|(name, expected)| { labels .get(name) .is_some_and(|value| &value.to_string() == expected) }) }) }) .filter(|(_, (sequence, timestamp, _))| { since_sequence.is_none_or(|since| *sequence > since) && start_ms.is_none_or(|start| timestamp.as_millis() >= start) && end_ms.is_none_or(|end| timestamp.as_millis() <= end) }) .take(limit) .map(|(labels, (sequence, timestamp, value))| BucketValueDto { labels: labels.clone(), sequence: *sequence, timestamp_ms: timestamp.as_millis(), value: value.clone(), }) .collect(); EngineResponse::BucketValues(values) } EngineQuery::Promql { query, start_ms, end_ms } => { let response = promql::parse_query(&query) .and_then(|parsed| evaluate_promql(index, &parsed, start_ms, end_ms)); EngineResponse::Promql(response) } } } fn coerce_to_f64(value: &MetricPrimitive) -> f64 { match value { MetricPrimitive::Number(n) => match n { Number::Int(v) => *v as f64, Number::Float(v) => *v, }, MetricPrimitive::Timestamp(ts) => ts.as_millis() as f64, MetricPrimitive::Duration(ms) => *ms as f64, MetricPrimitive::String(_) => 1.0, MetricPrimitive::Boolean(b) => { if *b { 1.0 } else { 0.0 } } } } fn matches_label_matchers( labels: &BTreeMap, matchers: &[LabelMatcher], ) -> bool { matchers.iter().all(|matcher| { labels.get(&matcher.name).is_some_and(|value| value.to_string() == matcher.value) }) } fn project_labels( labels: &BTreeMap, by_labels: &[String], ) -> BTreeMap { by_labels .iter() .filter_map(|name| labels.get(name).map(|value| (name.clone(), value.clone()))) .collect() } fn windowed_rate_series( points: &[(i64, f64)], range_ms: i64, start_ms: Option, end_ms: Option, ) -> Vec<(i64, f64)> { let ts_points: Vec<(Timestamp, f64)> = points.iter().map(|(ms, value)| (Timestamp::from_millis(*ms), *value)).collect(); let mut series: Vec<(i64, f64)> = rate::windowed_rate_series(&ts_points, range_ms) .into_iter() .map(|(ts, value)| (ts.as_millis(), value)) .collect(); if let Some(end) = end_ms { if series.last().is_none_or(|(ts, _)| *ts < end) { let value = rate::rate_at(&ts_points, Timestamp::from_millis(end), range_ms); series.push((end, value)); } } if let Some(start) = start_ms { if series.first().is_none_or(|(ts, _)| *ts > start) { let value = rate::rate_at(&ts_points, Timestamp::from_millis(start), range_ms); series.insert(0, (start, value)); } } series } fn merge_series_windowed(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { if sub_series.len() <= 1 { return sub_series.first().cloned().unwrap_or_default(); } let mut grouped: BTreeMap> = BTreeMap::new(); for points in sub_series { for (timestamp_ms, value) in points { let window_start = timestamp_ms.div_euclid(MERGE_WINDOW_MS) * MERGE_WINDOW_MS; grouped.entry(window_start).or_default().push(*value); } } grouped .into_iter() .map(|(timestamp_ms, values)| { let sum: f64 = values.iter().sum(); let value = if is_avg { sum / values.len() as f64 } else { sum }; (timestamp_ms, value) }) .collect() } fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { let mut all_timestamps: std::collections::BTreeSet = std::collections::BTreeSet::new(); for points in sub_series { all_timestamps.extend(points.iter().map(|(ts, _)| *ts)); } let mut cursors = vec![0usize; sub_series.len()]; let mut last_values = vec![0.0f64; sub_series.len()]; all_timestamps .into_iter() .map(|ts| { for (i, points) in sub_series.iter().enumerate() { while cursors[i] < points.len() && points[cursors[i]].0 <= ts { last_values[i] = points[cursors[i]].1; cursors[i] += 1; } } let sum: f64 = last_values.iter().sum(); let value = if is_avg { sum / last_values.len() as f64 } else { sum }; (ts, value) }) .collect() } fn declared_value_kind( index: &BTreeMap, metric_name: &str, ) -> Option { match index.get(&IndexKey::MetricDeclaration { name: metric_name.to_string() }) { Some(DataValue::MetricDeclaration { value_param, .. }) => Some( value_param .as_ref() .and_then(|name| { index.get(&IndexKey::ParamDeclaration { name: name.clone() }) }) .map(|value| match value { DataValue::ParamDeclaration(param) => param.kind, _ => MetricPrimitiveKind::Number, }) .unwrap_or(MetricPrimitiveKind::Number), ), _ => None, } } 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, } } fn evaluate_promql( index: &BTreeMap, 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() && !index.contains_key(&IndexKey::MetricDeclaration { name: parsed.metric_name.clone() }) && bucket_values(index, &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 && declared_value_kind(index, &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 logs_range(index) { 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 bucket_values(index, &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() } }) } #[cfg(test)] mod tests { use super::*; fn request(name: &str, pattern: &str, capture_group: &str) -> DeclareParamRequest { DeclareParamRequest { name: name.to_string(), message_match_regex: pattern.to_string(), capture_group: capture_group.to_string(), json_field_path: None, kind: MetricPrimitiveKind::String, elapsed_since_last: false, auto_derived: false, log_stream_gated: false, } } #[test] fn declares_valid_param() { let mut index = BTreeMap::new(); let result = declare_param(&mut index, request("level", r"level=(?P\w+)", "level")); assert!(result.is_ok()); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "level"); assert_eq!(declarations[0].1, r"level=(?P\w+)"); assert_eq!(declarations[0].2, "level"); } #[test] fn auto_declare_params_discovers_new_keys_with_inferred_kinds_and_flags_them_auto() { let mut index = BTreeMap::new(); auto_declare_params(&mut index, r#"ts=1788648105773 msg="entity_moved" x1=-4 y1=34"#); let declarations: HashMap<&str, _> = param_declarations_range(&index).map(|d| (d.0, d)).collect(); assert_eq!(declarations.len(), 4); assert_eq!(declarations["x1"].4, MetricPrimitiveKind::Number); assert!(declarations["x1"].6, "expected x1 to be flagged auto_derived"); assert_eq!(declarations["msg"].4, MetricPrimitiveKind::String); let compiled = compiled_param_declarations(&index); let params = derive_log_params(r#"ts=1788648105773 msg="entity_moved" x1=-4 y1=34"#, &compiled); assert_eq!(params.get("x1"), Some(&MetricPrimitive::Number(Number::Int(-4)))); assert_eq!(params.get("msg"), Some(&MetricPrimitive::String("entity_moved".to_string()))); } #[test] fn auto_declare_params_discovers_json_leaf_fields_via_json_field_path() { let mut index = BTreeMap::new(); auto_declare_params(&mut index, r#"{"level":"error","code":500,"request":{"path":"/x"}}"#); let declarations: HashMap<&str, _> = param_declarations_range(&index).map(|d| (d.0, d)).collect(); assert_eq!(declarations.len(), 3); assert_eq!(declarations["code"].4, MetricPrimitiveKind::Number); assert_eq!(declarations["code"].3, Some("code")); assert!(declarations["level"].6, "expected level to be flagged auto_derived"); assert_eq!(declarations["request.path"].3, Some("request.path")); let compiled = compiled_param_declarations(&index); let params = derive_log_params(r#"{"level":"error","code":500,"request":{"path":"/x"}}"#, &compiled); assert_eq!(params.get("level"), Some(&MetricPrimitive::String("error".to_string()))); assert_eq!(params.get("code"), Some(&MetricPrimitive::Number(Number::Int(500)))); assert_eq!(params.get("request.path"), Some(&MetricPrimitive::String("/x".to_string()))); } #[test] fn auto_declare_params_never_overwrites_an_existing_declaration() { let mut index = BTreeMap::new(); declare_param(&mut index, request("x1", r#"x1="(?P[^"]+)""#, "x1")).unwrap(); auto_declare_params(&mut index, "x1=-4"); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].1, r#"x1="(?P[^"]+)""#, "manual declaration must survive untouched"); assert!(!declarations[0].6, "a manual declaration must never be flagged auto_derived"); } #[test] fn declare_param_rejects_editing_an_auto_derived_param() { let mut index = BTreeMap::new(); auto_declare_params(&mut index, "x1=-4"); let result = declare_param(&mut index, request("x1", r#"x1="(?P[^"]+)""#, "x1")); let err = result.unwrap_err(); assert!(err.contains("auto-detected"), "unexpected error message: {err}"); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert!(declarations[0].6, "the original auto-derived declaration must be untouched"); } const LOG_STREAM_FIXTURE: &str = include_str!("fixtures/log_stream_sample.txt"); fn ingest_log_stream_fixture(index: &mut BTreeMap) { let mut header = None; for line in LOG_STREAM_FIXTURE.lines() { auto_declare_log_stream_params(index, &mut header, line); auto_declare_params(index, line); } } #[test] fn ingesting_log_stream_fixture_does_not_blow_up_declared_param_count() { let mut index = BTreeMap::new(); ingest_log_stream_fixture(&mut index); let declared = param_declarations_range(&index).count(); assert!( declared <= MAX_AUTO_DECLARED_PARAMS, "expected at most {MAX_AUTO_DECLARED_PARAMS} declared params, got {declared}" ); assert!(declared < 100, "declared param count ({declared}) should stay well under the old ~220"); } #[test] fn base64_padding_fragments_in_log_stream_fixture_are_never_auto_declared() { let mut index = BTreeMap::new(); ingest_log_stream_fixture(&mut index); let names: HashSet<&str> = param_declarations_range(&index).map(|d| d.0).collect(); assert!(!names.contains("rKTEtG7JmnThhoKPbZiVrA")); assert!(!names.contains("gEdA8N4g")); } #[test] fn log_stream_shaped_lines_expose_header_named_columns_as_queryable_params() { let mut index = BTreeMap::new(); ingest_log_stream_fixture(&mut index); let declared: HashSet<&str> = param_declarations_range(&index).map(|d| d.0).collect(); for name in ["Timestamp", "Thread", "Type", "Activity", "PID", "TTL", "message"] { assert!(declared.contains(name), "expected {name} to be declared, got {declared:?}"); } let compiled = compiled_param_declarations(&index); let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap(); let params = derive_log_params(data_line, &compiled); assert_eq!(params.get("Type"), Some(&MetricPrimitive::String("Default".to_string()))); assert_eq!(params.get("PID"), Some(&MetricPrimitive::Number(Number::Int(1343)))); assert_eq!(params.get("TTL"), Some(&MetricPrimitive::Number(Number::Int(7)))); assert_eq!( params.get("message"), Some(&MetricPrimitive::String( "kitty: (SkyLight) [com.apple.SkyLight:default] SLSGetDisplaysWithOpenGLDisplayMask, 3614: Display 0x00000001: Mode 54, (0 x 0) [1728 x 1117], Depth 32 BPP, active".to_string() )) ); } #[test] fn log_stream_columns_are_not_contaminated_by_non_columnar_lines() { let mut index = BTreeMap::new(); ingest_log_stream_fixture(&mut index); let compiled = compiled_param_declarations(&index); let continuation_line = "so_gencnt: 4042707 so_state: 0x0"; let params = derive_log_params(continuation_line, &compiled); assert!(!params.contains_key("Timestamp"), "continuation line got a Timestamp: {params:?}"); assert!(!params.contains_key("Activity"), "continuation line got an Activity: {params:?}"); assert!(!params.contains_key("PID"), "continuation line got a PID: {params:?}"); let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); let header_params = derive_log_params(header_line, &compiled); assert!(!header_params.contains_key("Timestamp"), "header line got a Timestamp: {header_params:?}"); assert!(!header_params.contains_key("Activity"), "header line got an Activity: {header_params:?}"); assert!(!header_params.contains_key("Thread"), "header line got a Thread: {header_params:?}"); } #[test] fn logs_query_sorted_by_a_log_stream_column_actually_reorders_real_log_stream_output() { let mut index = BTreeMap::new(); let mut header = None; for (i, line) in LOG_STREAM_FIXTURE.lines().enumerate() { let sequence = i as u64; let timestamp = Timestamp::from_millis(sequence as i64); auto_declare_log_stream_params(&mut index, &mut header, line); auto_declare_params(&mut index, line); index.insert( IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text: line.to_string() }, ); } let entries = match handle_query(&index, logs_query_sorted("PID", Some(false)), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => entries, _ => panic!("expected Logs response"), }; assert!(!entries.is_empty(), "expected at least one line with a PID value"); let pids: Vec = entries .iter() .map(|entry| match entry.params.get("PID") { Some(MetricPrimitive::Number(Number::Int(pid))) => *pid, other => panic!("expected every returned entry to have a numeric PID, got {other:?}"), }) .collect(); let mut sorted_pids = pids.clone(); sorted_pids.sort_unstable(); assert_eq!(pids, sorted_pids, "entries were not actually reordered by PID"); let original_order: Vec = entries.iter().map(|e| e.sequence).collect(); let mut sorted_by_sequence = original_order.clone(); sorted_by_sequence.sort_unstable(); assert_ne!(original_order, sorted_by_sequence, "PID sort produced the same order as plain ingest order"); } #[test] fn non_log_stream_lines_do_not_get_any_columns_declared_without_a_cached_header() { let mut index = BTreeMap::new(); let mut header = None; auto_declare_log_stream_params(&mut index, &mut header, r#"ts=1788648105773 msg="entity_moved" x1=-4"#); assert!(param_declarations_range(&index).next().is_none()); assert!(header.is_none()); } #[test] #[ignore] fn bench_ingest_fixture_throughput() { let repeats = 50; let mut index = BTreeMap::new(); let start = std::time::Instant::now(); for _ in 0..repeats { let mut header = None; for line in LOG_STREAM_FIXTURE.lines() { auto_declare_log_stream_params(&mut index, &mut header, line); auto_declare_params(&mut index, line); } } let elapsed = start.elapsed(); let total_lines = repeats * LOG_STREAM_FIXTURE.lines().count(); eprintln!( "ingested {total_lines} lines in {elapsed:?} ({:.0} lines/sec), {} params declared", total_lines as f64 / elapsed.as_secs_f64(), param_declarations_range(&index).count(), ); } #[test] fn rejects_invalid_regex() { let mut index = BTreeMap::new(); let result = declare_param(&mut index, request("level", "level=(", "level")); assert!(result.is_err()); assert!(index.is_empty()); } #[test] fn rejects_unknown_capture_group() { let mut index = BTreeMap::new(); let result = declare_param( &mut index, request("level", r"level=(?P\w+)", "missing"), ); assert!(result.is_err()); assert!(index.is_empty()); } #[test] fn declares_valid_elapsed_since_last_param_without_capture_group() { let mut index = BTreeMap::new(); declare_elapsed_since_last_param(&mut index, "gap"); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert!(declarations[0].5, "expected elapsed_since_last to be true"); } #[test] fn rejects_elapsed_since_last_param_with_non_duration_kind() { let mut index = BTreeMap::new(); let result = declare_param( &mut index, DeclareParamRequest { name: "gap".to_string(), message_match_regex: ".".to_string(), capture_group: String::new(), json_field_path: None, kind: MetricPrimitiveKind::String, elapsed_since_last: true, auto_derived: false, log_stream_gated: false, }, ); assert!(result.is_err()); assert!(index.is_empty()); } #[test] fn declares_valid_param_from_json_field_path() { let mut index = BTreeMap::new(); let result = declare_param( &mut index, DeclareParamRequest { name: "level".to_string(), message_match_regex: r".".to_string(), capture_group: String::new(), json_field_path: Some("foo.bar".to_string()), kind: MetricPrimitiveKind::String, elapsed_since_last: false, auto_derived: false, log_stream_gated: false, }, ); assert!(result.is_ok()); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].3, Some("foo.bar")); } #[test] fn derives_log_param_from_json_field_path() { let mut index = BTreeMap::new(); declare_param( &mut index, DeclareParamRequest { name: "bar".to_string(), message_match_regex: r".".to_string(), capture_group: String::new(), json_field_path: Some("foo.bar".to_string()), kind: MetricPrimitiveKind::Number, elapsed_since_last: false, auto_derived: false, log_stream_gated: false, }, ) .unwrap(); let declarations = compiled_param_declarations(&index); let params = derive_log_params(r#"{"foo":{"bar":42}}"#, &declarations); assert_eq!(params.get("bar"), Some(&MetricPrimitive::Number(Number::Int(42)))); assert!(derive_log_params("not json", &declarations).is_empty()); assert!(derive_log_params(r#"{"foo":{}}"#, &declarations).is_empty()); } #[test] fn evict_excess_logs_drops_oldest_sequences_first() { let mut index = BTreeMap::new(); let mut log_count = 0; let mut approx_bytes_total = 0u64; for sequence in 0..5u64 { let text = format!("line {sequence}"); approx_bytes_total += approx_log_bytes(&text); index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, DataValue::LogLiteral { text }, ); log_count += 1; } evict_excess_logs(&mut index, &mut log_count, Some(3), &mut approx_bytes_total); assert_eq!(log_count, 3); let remaining: Vec<_> = logs_range(&index).map(|(seq, _, _)| seq).collect(); assert_eq!(remaining, vec![2, 3, 4]); let expected_bytes: u64 = remaining.iter().map(|seq| approx_log_bytes(&format!("line {seq}"))).sum(); assert_eq!(approx_bytes_total, expected_bytes); } #[test] fn evict_excess_logs_is_noop_when_unlimited_or_under_cap() { let mut index = BTreeMap::new(); let mut log_count = 0; let mut approx_bytes_total = 0u64; for sequence in 0..3u64 { approx_bytes_total += approx_log_bytes("x"); index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, DataValue::LogLiteral { text: "x".to_string() }, ); log_count += 1; } evict_excess_logs(&mut index, &mut log_count, None, &mut approx_bytes_total); assert_eq!(log_count, 3); assert_eq!(approx_bytes_total, 3); evict_excess_logs(&mut index, &mut log_count, Some(10), &mut approx_bytes_total); assert_eq!(log_count, 3); assert_eq!(approx_bytes_total, 3); } #[test] fn ingest_lag_watermark_reports_the_highest_sample_seen() { let mut samples = Vec::new(); let t0 = Timestamp::from_millis(0); record_ingest_lag(&mut samples, t0, 10); record_ingest_lag(&mut samples, t0, 50); record_ingest_lag(&mut samples, t0, 20); assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 50); } #[test] fn ingest_lag_watermark_only_replaces_the_current_minimum_once_at_capacity() { let mut samples = Vec::new(); let t0 = Timestamp::from_millis(0); for lag in [10, 20, 30, 40, 50] { record_ingest_lag(&mut samples, t0, lag); } assert_eq!(samples.len(), 5); assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 50); record_ingest_lag(&mut samples, t0, 5); assert_eq!(samples.len(), 5); assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 50); record_ingest_lag(&mut samples, t0, 25); assert_eq!(samples.len(), 5); assert!(samples.iter().all(|(_, lag)| *lag != 10), "the old minimum (10) should have been replaced"); assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 50); } #[test] fn ingest_lag_watermark_decays_once_samples_age_out_of_the_window() { let mut samples = Vec::new(); let t0 = Timestamp::from_millis(0); record_ingest_lag(&mut samples, t0, 500); assert_eq!(ingest_lag_high_watermark_ms(&samples, t0), 500); let almost_expired = Timestamp::from_millis(INGEST_LAG_WATERMARK_WINDOW_MS - 1); assert_eq!(ingest_lag_high_watermark_ms(&samples, almost_expired), 500); let expired = Timestamp::from_millis(INGEST_LAG_WATERMARK_WINDOW_MS); assert_eq!(ingest_lag_high_watermark_ms(&samples, expired), 0); record_ingest_lag(&mut samples, expired, 5); assert_eq!(samples.len(), 1); assert_eq!(ingest_lag_high_watermark_ms(&samples, expired), 5); } fn insert_ingested_log(index: &mut BTreeMap, sequence: u64, text: &str) { index.insert( IndexKey::IngestedLog { sequence, timestamp: Timestamp::from_millis(sequence as i64) }, DataValue::LogLiteral { text: text.to_string() }, ); } #[test] fn clear_all_param_declarations_removes_manual_and_auto_derived() { let mut index = BTreeMap::new(); auto_declare_params(&mut index, "x1=-4"); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); assert_eq!(param_declarations_range(&index).count(), 2); let removed = clear_all_param_declarations(&mut index); assert_eq!(removed, 2); assert_eq!(param_declarations_range(&index).count(), 0); } #[test] fn reextract_params_from_retained_logs_rebuilds_from_ingested_text_without_new_ingest() { let mut index = BTreeMap::new(); insert_ingested_log(&mut index, 0, r#"level=info msg="started""#); insert_ingested_log(&mut index, 1, "x1=-4 y1=34"); auto_declare_params(&mut index, r#"level=info msg="started""#); auto_declare_params(&mut index, "x1=-4 y1=34"); assert_eq!(param_declarations_range(&index).count(), 4); clear_all_param_declarations(&mut index); assert_eq!(param_declarations_range(&index).count(), 0); let (reextracted, logs_considered) = reextract_params_from_retained_logs(&mut index); assert_eq!(logs_considered, 2); assert_eq!(reextracted, 4); let declarations: HashMap<&str, _> = param_declarations_range(&index).map(|d| (d.0, d)).collect(); assert!(declarations.contains_key("level")); assert!(declarations.contains_key("msg")); assert!(declarations.contains_key("x1")); assert!(declarations.contains_key("y1")); assert!(declarations.values().all(|d| d.6), "re-extracted params must be flagged auto_derived"); } #[test] fn reextract_params_from_retained_logs_only_covers_lines_still_in_the_index() { let mut index = BTreeMap::new(); insert_ingested_log(&mut index, 1, "x1=-4"); auto_declare_params(&mut index, "evicted_only_key=1"); auto_declare_params(&mut index, "x1=-4"); assert_eq!(param_declarations_range(&index).count(), 2); clear_all_param_declarations(&mut index); let (reextracted, logs_considered) = reextract_params_from_retained_logs(&mut index); assert_eq!(logs_considered, 1, "only the retained log line should be re-scanned"); assert_eq!(reextracted, 1, "the evicted line's key can't come back"); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations[0].0, "x1"); } #[test] fn declarations_stay_within_their_range() { let mut index = BTreeMap::new(); index.insert( IndexKey::IngestedLog { sequence: 0, timestamp: Timestamp::from_millis(0) }, DataValue::LogLiteral { text: "hello".to_string() }, ); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); index.insert( IndexKey::MetricDeclaration { name: "requests".to_string() }, DataValue::MetricDeclaration { message_match_regex: "x".to_string(), label_params: vec![], value_param: None, matched_labels: None, unmatched_labels: None, }, ); let declarations: Vec<_> = param_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "level"); for key in index.range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit).map(|(k, _)| k) { assert!(matches!(key, IndexKey::ParamDeclaration { .. })); } } #[test] fn removes_declaration() { let mut index = BTreeMap::new(); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); assert!(index.remove(&IndexKey::ParamDeclaration { name: "level".to_string() }).is_some()); assert!(param_declarations_range(&index).next().is_none()); } fn declare_test_param( index: &mut BTreeMap, name: &str, pattern: &str, capture_group: &str, kind: MetricPrimitiveKind, ) { declare_param( index, DeclareParamRequest { name: name.to_string(), message_match_regex: pattern.to_string(), capture_group: capture_group.to_string(), json_field_path: None, kind, elapsed_since_last: false, auto_derived: false, log_stream_gated: false, }, ) .unwrap(); } fn declare_elapsed_since_last_param(index: &mut BTreeMap, name: &str) { declare_param( index, DeclareParamRequest { name: name.to_string(), message_match_regex: ".".to_string(), capture_group: String::new(), json_field_path: None, kind: MetricPrimitiveKind::Duration, elapsed_since_last: true, auto_derived: false, log_stream_gated: false, }, ) .unwrap(); } fn metric_request(name: &str, pattern: &str, label_params: Vec<&str>) -> DeclareMetricRequest { metric_request_with_value(name, pattern, label_params, None) } fn metric_request_with_value( name: &str, pattern: &str, label_params: Vec<&str>, value_param: Option<&str>, ) -> DeclareMetricRequest { DeclareMetricRequest { name: name.to_string(), message_match_regex: pattern.to_string(), label_params: label_params.into_iter().map(str::to_string).collect(), value_param: value_param.map(str::to_string), matched_labels: None, unmatched_labels: None, } } #[test] fn declares_valid_metric() { let mut index = BTreeMap::new(); let pattern = r"request path=(?P\S+) duration=(?P\d+)"; declare_test_param(&mut index, "path", pattern, "path", MetricPrimitiveKind::String); declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); let result = declare_metric(&mut index, metric_request("requests", pattern, vec!["path", "duration"])); assert!(result.is_ok()); let declarations: Vec<_> = metric_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "requests"); assert_eq!(declarations[0].1, pattern); assert_eq!(declarations[0].2.len(), 2); } #[test] fn rejects_invalid_metric_regex() { let mut index = BTreeMap::new(); let result = declare_metric(&mut index, metric_request("requests", "path=(", vec![])); assert!(result.is_err()); assert!(index.is_empty()); } #[test] fn rejects_metric_referencing_undeclared_param() { let mut index = BTreeMap::new(); let pattern = r"request path=(?P\S+)"; declare_test_param(&mut index, "path", pattern, "path", MetricPrimitiveKind::String); let result = declare_metric(&mut index, metric_request("requests", pattern, vec!["path", "duration"])); assert!(result.is_err()); assert!(result.unwrap_err().contains("duration")); assert!(!index.contains_key(&IndexKey::MetricDeclaration { name: "requests".to_string() })); } #[test] fn metric_declarations_stay_within_their_range() { let mut index = BTreeMap::new(); index.insert( IndexKey::IngestedLog { sequence: 0, timestamp: Timestamp::from_millis(0) }, DataValue::LogLiteral { text: "hello".to_string() }, ); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); let pattern = r"duration=(?P\d+)"; declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); declare_metric(&mut index, metric_request("requests", pattern, vec!["duration"])).unwrap(); let declarations: Vec<_> = metric_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "requests"); for key in index.range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit).map(|(k, _)| k) { assert!(matches!(key, IndexKey::MetricDeclaration { .. })); } } #[test] fn removes_metric_declaration() { let mut index = BTreeMap::new(); let pattern = r"duration=(?P\d+)"; declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); declare_metric(&mut index, metric_request("requests", pattern, vec!["duration"])).unwrap(); assert!(index.remove(&IndexKey::MetricDeclaration { name: "requests".to_string() }).is_some()); assert!(metric_declarations_range(&index).next().is_none()); } #[test] fn remove_metric_declaration_reports_whether_it_existed() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); assert!(remove_metric_declaration(&mut index, "requests")); assert!(!remove_metric_declaration(&mut index, "requests")); assert!(!remove_metric_declaration(&mut index, "never-declared")); } #[test] fn remove_metric_declaration_purges_bucket_history_so_promql_reports_unknown_metric() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); let parsed = promql::parse_query("requests").unwrap(); assert!(evaluate_promql(&index, &parsed, None, None).unwrap().len() == 1); assert!(remove_metric_declaration(&mut index, "requests")); assert!(bucket_values(&index, "requests", None).next().is_none()); let err = evaluate_promql(&index, &parsed, None, None).unwrap_err(); assert!(err.contains("unknown metric"), "unexpected error message: {err}"); } #[test] fn undeclared_metric_with_bucket_data_is_still_queryable_via_promql() { let mut index = BTreeMap::new(); push_bucket_value( &mut index, "scraped_metric", BTreeMap::new(), 0, Timestamp::from_millis(0), MetricPrimitive::Number(Number::Float(1.0)), ); let parsed = promql::parse_query("scraped_metric").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); } #[test] fn promql_rejects_metric_with_no_declaration_and_no_data() { let index = BTreeMap::new(); let parsed = promql::parse_query("never_declared").unwrap(); let err = evaluate_promql(&index, &parsed, None, None).unwrap_err(); assert!(err.contains("unknown metric"), "unexpected error message: {err}"); } #[test] fn update_metric_rejects_when_no_declaration_exists_yet() { let mut index = BTreeMap::new(); let result = update_metric( &mut index, "requests", metric_request("requests", r"request path=(?P\S+)", vec![]), ); assert!(result.is_err()); assert!(result.unwrap_err().contains("requests")); } #[test] fn update_metric_reuses_declare_metric_validation() { let mut index = BTreeMap::new(); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec![])).unwrap(); let result = update_metric( &mut index, "requests", metric_request("requests", r"request path=(?P\S+)", vec!["duration"]), ); assert!(result.is_err()); assert!(result.unwrap_err().contains("duration")); } #[test] fn update_metric_replaces_declaration_in_place_without_duplicating() { let mut index = BTreeMap::new(); let pattern = r"request path=(?P\S+) duration=(?P\d+)"; declare_test_param(&mut index, "path", pattern, "path", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec!["path"])).unwrap(); declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); let result = update_metric(&mut index, "requests", metric_request("requests", pattern, vec!["path", "duration"])); assert!(result.is_ok()); let declarations: Vec<_> = metric_declarations_range(&index).collect(); assert_eq!(declarations.len(), 1); assert_eq!(declarations[0].0, "requests"); assert_eq!(declarations[0].2.len(), 2); } #[test] fn update_metric_rebackfills_bucket_data_from_the_new_declaration() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/b"); declare_test_param(&mut index, "path", r"request path=(?P\S+)", "path", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec!["path"])).unwrap(); assert_eq!(bucket_values(&index, "requests", None).count(), 2); update_metric(&mut index, "requests", metric_request("requests", r"request path=(?P/a)", vec!["path"])).unwrap(); assert_eq!(bucket_values(&index, "requests", None).count(), 1); } #[test] fn update_metric_can_rename_and_rejects_a_rename_onto_an_existing_name() { let mut index = BTreeMap::new(); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec![])).unwrap(); declare_metric(&mut index, metric_request("errors", r"error code=(?P\d+)", vec![])).unwrap(); let result = update_metric( &mut index, "requests", metric_request("errors", r"request path=(?P\S+)", vec![]), ); assert!(result.is_err()); update_metric( &mut index, "requests", metric_request("http_requests", r"request path=(?P\S+)", vec![]), ) .unwrap(); let names: Vec<_> = metric_declarations_range(&index).map(|(name, ..)| name.to_string()).collect(); assert!(names.contains(&"http_requests".to_string())); assert!(!names.contains(&"requests".to_string())); } fn ingest_log(index: &mut BTreeMap, sequence: u64, text: &str) { let timestamp = Timestamp::from_millis(sequence as i64); index.insert( IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text: text.to_string() }, ); apply_all_metrics(index, sequence, timestamp, text); } #[test] fn declaring_metric_backfills_matching_logs() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "request path=/a duration=1"); ingest_log(&mut index, 1, "request path=/b duration=2"); ingest_log(&mut index, 2, "unrelated line"); let pattern = r"request path=(?P\S+) duration=(?P\d+)"; declare_test_param(&mut index, "path", pattern, "path", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", pattern, vec!["path"])).unwrap(); let values: Vec<_> = bucket_values(&index, "requests", None).collect(); assert_eq!(values.len(), 2); } #[test] fn metric_param_extracts_from_json_field_path() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, r#"{"request":{"path":"/a"}}"#); ingest_log(&mut index, 1, "not json"); ingest_log(&mut index, 2, r#"{"request":{}}"#); declare_param( &mut index, DeclareParamRequest { name: "path".to_string(), message_match_regex: r".".to_string(), capture_group: String::new(), json_field_path: Some("request.path".to_string()), kind: MetricPrimitiveKind::String, elapsed_since_last: false, auto_derived: false, log_stream_gated: false, }, ) .unwrap(); declare_metric(&mut index, metric_request("requests", r".", vec!["path"])).unwrap(); let values: Vec<_> = bucket_values(&index, "requests", None).collect(); assert_eq!(values.len(), 1); assert_eq!( values[0].0.get("path"), Some(&MetricPrimitive::String("/a".to_string())) ); } #[test] fn ingesting_after_declaration_updates_buckets_live() { let mut index = BTreeMap::new(); declare_test_param(&mut index, "path", r"request path=(?P\S+)", "path", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec!["path"])).unwrap(); ingest_log(&mut index, 0, "request path=/a"); let values: Vec<_> = bucket_values(&index, "requests", None).collect(); assert_eq!(values.len(), 1); assert_eq!(values[0].1.0, 0); } #[test] fn non_matching_log_leaves_bucket_range_untouched() { let mut index = BTreeMap::new(); declare_test_param(&mut index, "path", r"request path=(?P\S+)", "path", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec!["path"])).unwrap(); ingest_log(&mut index, 0, "totally unrelated line"); assert!( index .range(IndexKey::MetricBucketSplit..IndexKey::End) .find(|(key, _)| matches!(key, IndexKey::Bucket { .. })) .is_none() ); } #[test] fn bucket_values_preserve_insertion_order() { let mut index = BTreeMap::new(); declare_test_param(&mut index, "path", r"request path=(?P\S+)", "path", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec!["path"])).unwrap(); ingest_log(&mut index, 5, "request path=/a"); ingest_log(&mut index, 2, "request path=/a"); ingest_log(&mut index, 9, "request path=/a"); let sequences: Vec = bucket_values(&index, "requests", None) .map(|(_, (sequence, _, _))| *sequence) .collect(); assert_eq!(sequences, vec![5, 2, 9]); } #[test] fn differing_label_combinations_produce_distinct_buckets() { let mut index = BTreeMap::new(); declare_test_param(&mut index, "path", r"request path=(?P\S+)", "path", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", r"request path=(?P\S+)", vec!["path"])).unwrap(); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/b"); let bucket_keys: Vec<_> = index .range(IndexKey::MetricBucketSplit..IndexKey::End) .filter(|(key, _)| matches!(key, IndexKey::Bucket { .. })) .collect(); assert_eq!(bucket_keys.len(), 2); for (_, value) in &bucket_keys { match value { DataValue::BucketValue { values } => assert_eq!(values.len(), 1), _ => panic!("expected BucketValue"), } } } #[test] fn single_log_line_can_match_multiple_metrics() { let mut index = BTreeMap::new(); let pattern = r"request path=(?P\S+) duration=(?P\d+)"; declare_test_param(&mut index, "path", pattern, "path", MetricPrimitiveKind::String); declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); declare_metric(&mut index, metric_request("requests", pattern, vec!["path"])).unwrap(); declare_metric(&mut index, metric_request("durations", pattern, vec!["duration"])).unwrap(); ingest_log(&mut index, 0, "request path=/a duration=5"); assert_eq!(bucket_values(&index, "requests", None).count(), 1); assert_eq!(bucket_values(&index, "durations", None).count(), 1); } fn declare_requests_metric(index: &mut BTreeMap) { declare_test_param(index, "path", r"request path=(?P\S+)", "path", MetricPrimitiveKind::String); declare_metric(index, metric_request("requests", r"request path=(?P\S+)", vec!["path"])).unwrap(); } #[test] fn promql_query_selects_matching_label_combination() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/b"); ingest_log(&mut index, 2, "request path=/a"); let parsed = promql::parse_query(r#"requests{path="/a"}"#).unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].points, vec![(0, 1.0), (2, 1.0)]); } #[test] fn promql_query_with_no_matching_labels_returns_no_series() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); let parsed = promql::parse_query(r#"requests{path="/nope"}"#).unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert!(series.is_empty()); } #[test] fn promql_sum_aggregates_across_label_combinations() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(requests)").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert!(series[0].labels.is_empty()); assert_eq!(series[0].points, vec![(0, 2.0)]); } #[test] fn promql_sum_by_preserves_the_named_label_dimension() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); ingest_log(&mut index, 1, "request path=/a"); let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0)]); assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/b")); assert_eq!(series[1].points, vec![(0, 1.0)]); } fn declare_string_valued_metric(index: &mut BTreeMap) { declare_test_param( index, "status", r"request status=(?P\S+)", "status", MetricPrimitiveKind::String, ); declare_metric( index, metric_request_with_value( "statuses", r"request status=(?P\S+)", vec![], Some("status"), ), ) .unwrap(); } fn declare_requests_by_host_metric(index: &mut BTreeMap) { declare_test_param( index, "path", r"request path=(?P\S+) host=(?P\S+)", "path", MetricPrimitiveKind::String, ); declare_test_param( index, "host", r"request path=(?P\S+) host=(?P\S+)", "host", MetricPrimitiveKind::String, ); declare_metric( index, metric_request( "requests", r"request path=(?P\S+) host=(?P\S+)", vec!["path", "host"], ), ) .unwrap(); } #[test] fn sum_over_string_kind_metric_is_rejected_with_an_actionable_error() { let mut index = BTreeMap::new(); declare_string_valued_metric(&mut index); ingest_log(&mut index, 0, "request status=ok"); let parsed = promql::parse_query("sum(statuses)").unwrap(); let err = evaluate_promql(&index, &parsed, None, None).unwrap_err(); assert!(err.contains("statuses"), "error should name the offending metric: {err}"); assert!(err.to_lowercase().contains("string"), "error should explain why: {err}"); } #[test] fn avg_over_string_kind_metric_is_rejected_with_an_actionable_error() { let mut index = BTreeMap::new(); declare_string_valued_metric(&mut index); ingest_log(&mut index, 0, "request status=ok"); let parsed = promql::parse_query("avg(statuses)").unwrap(); let err = evaluate_promql(&index, &parsed, None, None).unwrap_err(); assert!(err.contains("statuses")); } #[test] fn count_over_string_kind_metric_counts_occurrences_by_name_not_by_accident() { let mut index = BTreeMap::new(); declare_string_valued_metric(&mut index); ingest_log(&mut index, 0, "request status=ok"); ingest_log(&mut index, 1, "request status=error"); let parsed = promql::parse_query("count(statuses)").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert!(series[0].labels.is_empty()); assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0)]); } #[test] fn sum_over_numeric_kind_metric_is_unaffected_by_the_string_kind_guard() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(requests)").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 2.0)]); } #[test] fn sum_of_rate_over_string_kind_metric_is_still_allowed_since_rate_only_counts_occurrences() { let mut index = BTreeMap::new(); declare_string_valued_metric(&mut index); ingest_log(&mut index, 0, "request status=ok"); ingest_log(&mut index, 1000, "request status=error"); let parsed = promql::parse_query("sum(rate(statuses[2s]))").unwrap(); let series = evaluate_promql(&index, &parsed, None, None); assert!(series.is_ok(), "sum(rate(...)) never reads the coerced value, so it should not be rejected: {series:?}"); } #[test] fn bare_query_over_string_kind_metric_is_still_allowed() { let mut index = BTreeMap::new(); declare_string_valued_metric(&mut index); ingest_log(&mut index, 0, "request status=ok"); let parsed = promql::parse_query("statuses").unwrap(); let series = evaluate_promql(&index, &parsed, None, None); assert!(series.is_ok()); } #[test] fn sum_by_merges_staggered_raw_timestamps_across_label_groups_instead_of_going_sparse() { let mut index = BTreeMap::new(); declare_requests_by_host_metric(&mut index); ingest_log(&mut index, 0, "request path=/a host=web1"); ingest_log(&mut index, 1, "request path=/a host=web2"); ingest_log(&mut index, 2, "request path=/a host=web3"); let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); let total: f64 = series[0].points.iter().map(|(_, v)| v).sum(); assert_eq!(total, 3.0, "expected all three staggered occurrences to sum to 3.0, got {:?}", series[0].points); assert!( series[0].points.len() < 3, "expected staggered timestamps to be bucketed together rather than staying sparse/near-empty, got {:?}", series[0].points ); } #[test] fn avg_by_merges_staggered_raw_timestamps_across_label_groups() { let mut index = BTreeMap::new(); declare_requests_by_host_metric(&mut index); ingest_log(&mut index, 0, "request path=/a host=web1"); ingest_log(&mut index, 1, "request path=/a host=web2"); let parsed = promql::parse_query("avg(requests) by (path)").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].points, vec![(0, 1.0)]); } #[test] fn sum_by_still_gives_a_single_label_group_its_own_line_at_full_resolution() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/a"); let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0)]); } #[test] fn bare_rate_produces_a_windowed_point_per_timestamp_not_a_single_point() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1000, "request path=/a"); let parsed = promql::parse_query("rate(requests[2s])").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]); } #[test] fn sum_of_rate_by_label_gives_each_label_value_its_own_line() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1000, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(rate(requests[2s])) by (path)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]); assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/b")); assert_eq!(series[1].points, vec![(0, 0.5)]); } #[test] fn rate_series_gets_a_trailing_zero_point_once_occurrences_stop_early_in_a_wider_range() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/a"); let parsed = promql::parse_query("rate(requests[2s])").unwrap(); let series = evaluate_promql(&index, &parsed, None, Some(100_000)).unwrap(); assert_eq!(series.len(), 1); assert_eq!( series[0].points, vec![(0, 0.5), (1, 1.0), (100_000, 0.0)], "the window (2s) is long since aged out by end_ms, so the appended boundary point must read 0" ); } #[test] fn rate_series_gets_a_leading_zero_point_when_occurrences_start_late_in_a_wider_range() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 50_000, "request path=/a"); let parsed = promql::parse_query("rate(requests[2s])").unwrap(); let series = evaluate_promql(&index, &parsed, Some(0), None).unwrap(); assert_eq!(series[0].points, vec![(0, 0.0), (50_000, 0.5)]); } #[test] fn sum_by_rate_gets_boundary_points_per_group_across_a_wider_range() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(rate(requests[2s])) by (path)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, Some(100_000)).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); assert_eq!(series[0].points, vec![(0, 0.5), (1, 1.0), (100_000, 0.0)]); assert_eq!(series[1].points, vec![(0, 0.5), (100_000, 0.0)]); } #[test] fn rate_series_skips_boundary_points_already_covered_by_real_data() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1000, "request path=/a"); let parsed = promql::parse_query("rate(requests[2s])").unwrap(); let series = evaluate_promql(&index, &parsed, Some(0), Some(1000)).unwrap(); assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]); } #[test] fn dashboard_rate_and_alert_threshold_agree_on_same_windowed_data() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); let lines: Vec<(u64, &str)> = vec![ (0, "request path=/a"), (500, "request path=/a"), (1000, "request path=/a"), (1500, "request path=/a"), (2000, "request path=/a"), ]; for (ms, text) in &lines { ingest_log(&mut index, *ms, text); } let parsed = promql::parse_query("rate(requests[2s])").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); let (_, dashboard_rate) = *series[0].points.last().unwrap(); let rule = alerts::AlertRule { id: "r1".to_string(), name: "spike".to_string(), enabled: true, condition: alerts::AlertCondition::Threshold { query: r#"regex("request path=/a")"#.to_string(), op: CompareOp::Ge, threshold: 0.0, window_ms: 2_000, }, }; let mut alert_states = HashMap::new(); let mut last_message = String::new(); for (ms, text) in &lines { let timestamp = Timestamp::from_millis(*ms as i64); let fired = alerts::evaluate_alerts(&[rule.clone()], &mut alert_states, &index, *ms, timestamp, text); assert_eq!(fired.len(), 1); last_message = fired[0].1.clone(); } let alert_count: f64 = last_message .split("match count ") .nth(1) .and_then(|rest| rest.split_whitespace().next()) .and_then(|token| token.parse().ok()) .expect("fired message should report a match count"); let window_seconds = 2.0; assert_eq!(dashboard_rate, 2.0); assert_eq!(alert_count, 4.0); assert_eq!(alert_count, dashboard_rate * window_seconds); } #[test] fn sum_of_rate_without_by_clause_still_collapses_to_one_series() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(rate(requests[2s]))").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert!(series[0].labels.is_empty()); assert_eq!(series[0].points, vec![(0, 1.0)]); } #[test] fn promql_range_bounds_filter_points_via_start_end_ms() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 5, "request path=/a"); ingest_log(&mut index, 10, "request path=/a"); let parsed = promql::parse_query("requests").unwrap(); let series = evaluate_promql(&index, &parsed, Some(1), Some(9)).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(5, 1.0)]); } #[test] fn declares_metric_with_value_param() { let mut index = BTreeMap::new(); let pattern = r"request path=(?P\S+) duration=(?P\d+)"; declare_test_param(&mut index, "path", pattern, "path", MetricPrimitiveKind::String); declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); let result = declare_metric( &mut index, metric_request_with_value("durations", pattern, vec!["path"], Some("duration")), ); assert!(result.is_ok()); } #[test] fn value_param_becomes_the_bucket_value_not_a_label() { let mut index = BTreeMap::new(); let pattern = r"request path=(?P\S+) duration=(?P\d+)"; declare_test_param(&mut index, "path", pattern, "path", MetricPrimitiveKind::String); declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); declare_metric( &mut index, metric_request_with_value("durations", pattern, vec!["path"], Some("duration")), ) .unwrap(); ingest_log(&mut index, 0, "request path=/a duration=5"); let values: Vec<_> = bucket_values(&index, "durations", None).collect(); assert_eq!(values.len(), 1); let (labels, (_, _, value)) = &values[0]; assert!(!labels.contains_key("duration")); assert_eq!(labels.get("path").map(|v| v.to_string()), Some("/a".to_string())); assert_eq!(*value, MetricPrimitive::Number(Number::Int(5))); } fn declare_durations_metric(index: &mut BTreeMap) { let pattern = r"request path=(?P\S+) duration=(?P\d+)"; declare_test_param(index, "path", pattern, "path", MetricPrimitiveKind::String); declare_test_param(index, "duration", pattern, "duration", MetricPrimitiveKind::Number); declare_metric( index, metric_request_with_value("durations", pattern, vec!["path"], Some("duration")), ) .unwrap(); } #[test] fn gauge_returns_most_recent_value_per_label_combination() { let mut index = BTreeMap::new(); declare_durations_metric(&mut index); ingest_log(&mut index, 0, "request path=/a duration=5"); ingest_log(&mut index, 1, "request path=/a duration=9"); ingest_log(&mut index, 2, "request path=/b duration=42"); let parsed = promql::parse_query("gauge(durations)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].points, vec![(1, 9.0)]); assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/b")); assert_eq!(series[1].points, vec![(2, 42.0)]); } #[test] fn gauge_respects_time_range_bounds() { let mut index = BTreeMap::new(); declare_durations_metric(&mut index); ingest_log(&mut index, 0, "request path=/a duration=5"); ingest_log(&mut index, 5, "request path=/a duration=9"); ingest_log(&mut index, 10, "request path=/a duration=42"); let parsed = promql::parse_query("gauge(durations)").unwrap(); let series = evaluate_promql(&index, &parsed, None, Some(9)).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(5, 9.0)]); } #[test] fn histogram_buckets_captured_values_by_fixed_width() { let mut index = BTreeMap::new(); declare_durations_metric(&mut index); for (seq, duration) in [(0, 2), (1, 4), (2, 12), (3, 15), (4, 21)] { ingest_log(&mut index, seq, &format!("request path=/a duration={duration}")); } let parsed = promql::parse_query("histogram(durations[10])").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("le").cmp(&b.labels.get("le"))); assert_eq!(series.len(), 3); assert_eq!(series[0].labels.get("le").map(String::as_str), Some("10")); assert_eq!(series[0].points[0].1, 2.0); assert_eq!(series[1].labels.get("le").map(String::as_str), Some("20")); assert_eq!(series[1].points[0].1, 2.0); assert_eq!(series[2].labels.get("le").map(String::as_str), Some("30")); assert_eq!(series[2].points[0].1, 1.0); } #[test] fn histogram_with_by_clause_buckets_each_label_group_separately() { let mut index = BTreeMap::new(); declare_durations_metric(&mut index); for (seq, path, duration) in [ (0, "/a", 2), (1, "/a", 4), (2, "/a", 12), (3, "/b", 21), ] { ingest_log(&mut index, seq, &format!("request path={path} duration={duration}")); } let parsed = promql::parse_query("histogram(durations[10]) by (path)").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| { a.labels.get("path").cmp(&b.labels.get("path")).then(a.labels.get("le").cmp(&b.labels.get("le"))) }); assert_eq!(series.len(), 3); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[0].labels.get("le").map(String::as_str), Some("10")); assert_eq!(series[0].points[0].1, 2.0); assert_eq!(series[1].labels.get("path").map(String::as_str), Some("/a")); assert_eq!(series[1].labels.get("le").map(String::as_str), Some("20")); assert_eq!(series[1].points[0].1, 1.0); assert_eq!(series[2].labels.get("path").map(String::as_str), Some("/b")); assert_eq!(series[2].labels.get("le").map(String::as_str), Some("30")); assert_eq!(series[2].points[0].1, 1.0); } fn declare_login_duration_metric(index: &mut BTreeMap) { let pattern = r"user=(?P\S+) logged in"; declare_test_param(index, "user", pattern, "user", MetricPrimitiveKind::String); declare_elapsed_since_last_param(index, "gap"); declare_metric(index, metric_request_with_value("login_gap", pattern, vec!["user"], Some("gap"))).unwrap(); } #[test] fn duration_value_param_computes_elapsed_time_since_previous_same_labels_event() { let mut index = BTreeMap::new(); declare_login_duration_metric(&mut index); ingest_log(&mut index, 0, "user=alice logged in"); ingest_log(&mut index, 1, "user=bob logged in"); ingest_log(&mut index, 2, "user=alice logged in"); let values: Vec<_> = bucket_values(&index, "login_gap", None).collect(); assert_eq!(values.len(), 1); let (labels, (sequence, timestamp, value)) = &values[0]; assert_eq!(labels.get("user").map(|v| v.to_string()), Some("alice".to_string())); assert_eq!(*sequence, 2); assert_eq!(timestamp.as_millis(), 2); assert_eq!(*value, MetricPrimitive::Duration(2)); } #[test] fn duration_diffs_against_the_immediately_preceding_event_not_the_first() { let mut index = BTreeMap::new(); declare_login_duration_metric(&mut index); ingest_log(&mut index, 0, "user=alice logged in"); ingest_log(&mut index, 3, "user=alice logged in"); ingest_log(&mut index, 9, "user=alice logged in"); let values: Vec<_> = bucket_values(&index, "login_gap", None).collect(); assert_eq!(values.len(), 2); assert_eq!(values[0].1.2, MetricPrimitive::Duration(3)); assert_eq!(values[1].1.2, MetricPrimitive::Duration(6)); } #[test] fn duration_values_are_queryable_via_promql_like_any_other_metric() { let mut index = BTreeMap::new(); declare_login_duration_metric(&mut index); ingest_log(&mut index, 0, "user=alice logged in"); ingest_log(&mut index, 10, "user=alice logged in"); let parsed = promql::parse_query("login_gap").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(10, 10.0)]); } fn declare_error_category_metric(index: &mut BTreeMap) { declare_metric( index, DeclareMetricRequest { name: "log_category".to_string(), message_match_regex: r"(?i)error".to_string(), label_params: vec![], value_param: None, matched_labels: Some(BTreeMap::from([("category".to_string(), "error".to_string())])), unmatched_labels: Some(BTreeMap::from([("category".to_string(), "other".to_string())])), }, ) .unwrap(); } #[test] fn unmatched_labels_registers_occurrence_for_non_matching_lines() { let mut index = BTreeMap::new(); declare_error_category_metric(&mut index); ingest_log(&mut index, 0, "everything is fine"); ingest_log(&mut index, 1, "an ERROR occurred"); ingest_log(&mut index, 2, "still fine"); let error_labels = BTreeMap::from([("category".to_string(), MetricPrimitive::String("error".to_string()))]); let error_values: Vec<_> = bucket_values(&index, "log_category", Some(&error_labels)).collect(); assert_eq!(error_values.len(), 1); assert_eq!(error_values[0].1.0, 1); let other_labels = BTreeMap::from([("category".to_string(), MetricPrimitive::String("other".to_string()))]); let other_values: Vec<_> = bucket_values(&index, "log_category", Some(&other_labels)).collect(); assert_eq!(other_values.len(), 2); } #[test] fn unmatched_labels_produces_two_series_from_one_bare_promql_query() { let mut index = BTreeMap::new(); declare_error_category_metric(&mut index); ingest_log(&mut index, 0, "an ERROR occurred"); ingest_log(&mut index, 1, "all good"); ingest_log(&mut index, 2, "all good again"); let parsed = promql::parse_query("log_category").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("category").cmp(&b.labels.get("category"))); assert_eq!(series.len(), 2); assert_eq!(series[0].labels.get("category").map(String::as_str), Some("error")); assert_eq!(series[0].points.len(), 1); assert_eq!(series[1].labels.get("category").map(String::as_str), Some("other")); assert_eq!(series[1].points.len(), 2); } #[test] fn no_unmatched_labels_means_non_matching_lines_are_untouched() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "not a request line at all"); assert_eq!(bucket_values(&index, "requests", None).count(), 0); } #[test] fn handle_query_promql_rejects_malformed_query() { let index = BTreeMap::new(); let response = handle_query( &index, EngineQuery::Promql { query: "not a valid promql query(".to_string(), start_ms: None, end_ms: None, }, 0, 0, &mut HashMap::new(), ); match response { EngineResponse::Promql(Err(message)) => assert!(!message.is_empty()), _ => panic!("expected EngineResponse::Promql(Err(_))"), } } #[test] fn handle_query_promql_returns_series_for_valid_query() { let mut index = BTreeMap::new(); declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); let response = handle_query( &index, EngineQuery::Promql { query: r#"requests{path="/a"}"#.to_string(), start_ms: None, end_ms: None, }, 0, 0, &mut HashMap::new(), ); match response { EngineResponse::Promql(Ok(series)) => { assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 1.0)]); } _ => panic!("expected EngineResponse::Promql(Ok(_))"), } } #[test] fn promql_count_over_ottyl_seq_gives_one_point_per_ingested_log_with_no_metric_declaration() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "line one"); ingest_log(&mut index, 1, "line two"); ingest_log(&mut index, 2, "line three"); let parsed = promql::parse_query("count(ottyl_seq)").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0), (2, 1.0)]); } #[test] fn promql_bare_ottyl_seq_returns_the_actual_sequence_numbers() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "line zero"); ingest_log(&mut index, 5, "line five"); let parsed = promql::parse_query("ottyl_seq").unwrap(); let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); series[0].points.sort_by_key(|(ts, _)| *ts); assert_eq!(series[0].points, vec![(0, 0.0), (5, 5.0)]); } #[test] fn promql_bare_ottyl_ts_returns_ingest_timestamps() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "a"); ingest_log(&mut index, 1, "b"); let parsed = promql::parse_query("ottyl_ts").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 0.0), (1, 1.0)]); } #[test] fn promql_rate_over_ottyl_ts_produces_a_windowed_series_not_an_empty_one() { let mut index = BTreeMap::new(); for sequence in 0..5u64 { ingest_log(&mut index, sequence, "line"); } let parsed = promql::parse_query("rate(ottyl_ts[5s])").unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert!(!series[0].points.is_empty(), "rate(ottyl_ts[...]) must not silently return no data"); } #[test] fn promql_ottyl_seq_with_label_matcher_matches_nothing_since_it_has_no_labels() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "line"); let parsed = promql::parse_query(r#"ottyl_seq{path="/a"}"#).unwrap(); let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert!(series.is_empty()); } #[test] fn promql_over_ottyl_fmt_is_explicitly_rejected_not_silently_empty() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, r#"{"level":"error"}"#); let parsed = promql::parse_query("count(ottyl_fmt)").unwrap(); let err = evaluate_promql(&index, &parsed, None, None).unwrap_err(); assert!(err.contains("ottyl_fmt"), "error should name the rejected metric: {err}"); assert!(err.contains("categorical") || err.contains("filter"), "error should explain why: {err}"); } #[test] fn handle_query_promql_returns_series_for_ottyl_seq_end_to_end() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "a"); ingest_log(&mut index, 1, "b"); let response = handle_query( &index, EngineQuery::Promql { query: "count(ottyl_seq)".to_string(), start_ms: None, end_ms: None }, 0, 0, &mut HashMap::new(), ); match response { EngineResponse::Promql(Ok(series)) => { assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0)]); } _ => panic!("expected EngineResponse::Promql(Ok(_))"), } } #[test] fn declare_param_rejects_reserved_ottyl_names() { let mut index = BTreeMap::new(); for name in OTTYL_RESERVED_PARAM_NAMES { let err = declare_param(&mut index, request(name, "x", "x")).unwrap_err(); assert!(err.contains("reserved"), "expected a 'reserved' error for '{name}', got: {err}"); } } fn logs_query(filter: Option<&str>) -> EngineQuery { EngineQuery::Logs(LogsParams { since_sequence: None, start_ms: None, end_ms: None, limit: None, filter: filter.map(str::to_string), sort_by: None, sort_missing_fallback: None, }) } fn logs_query_sorted(sort_by: &str, sort_missing_fallback: Option) -> EngineQuery { EngineQuery::Logs(LogsParams { since_sequence: None, start_ms: None, end_ms: None, limit: None, filter: None, sort_by: Some(sort_by.to_string()), sort_missing_fallback, }) } fn timeline_query(filter: Option<&str>, bucket_ms: i64) -> EngineQuery { EngineQuery::Timeline(TimelineParams { start_ms: None, end_ms: None, bucket_ms: Some(bucket_ms), filter: filter.map(str::to_string), }) } #[test] fn logs_query_response_includes_ottyl_fmt_per_entry() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, r#"{"level":"error"}"#); ingest_log(&mut index, 1, "level=info msg=hello"); ingest_log(&mut index, 2, "just plain prose"); match handle_query(&index, logs_query(None), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let formats: Vec> = entries.iter().map(|e| e.params.get("ottyl_fmt")).collect(); assert_eq!(formats[0], Some(&MetricPrimitive::String("json".to_string()))); assert_eq!(formats[1], Some(&MetricPrimitive::String("logfmt".to_string()))); assert_eq!(formats[2], Some(&MetricPrimitive::String("none".to_string()))); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_with_regex_filter_returns_only_matching_lines() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "everything is fine"); ingest_log(&mut index, 1, "an error occurred"); ingest_log(&mut index, 2, "still fine"); ingest_log(&mut index, 3, "another error here"); match handle_query(&index, logs_query(Some(r#"regex("error")"#)), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 3]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_with_ottyl_seq_comparison_filters_by_sequence() { let mut index = BTreeMap::new(); for i in 0..5u64 { ingest_log(&mut index, i, &format!("line {i}")); } match handle_query(&index, logs_query(Some("ottyl_seq<2")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 1]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_with_ottyl_ts_comparison_filters_by_timestamp() { let mut index = BTreeMap::new(); for i in 0..5u64 { let timestamp = Timestamp::from_millis(i as i64 * 1000); index.insert( IndexKey::IngestedLog { sequence: i, timestamp }, DataValue::LogLiteral { text: format!("line {i}") }, ); } match handle_query(&index, logs_query(Some("ottyl_ts>2000")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![3, 4]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_sort_by_orders_lines_by_declared_param_value() { let mut index = BTreeMap::new(); declare_test_param( &mut index, "priority", r"priority=(?P\d+)", "priority", MetricPrimitiveKind::Number, ); ingest_log(&mut index, 0, "priority=3 boom"); ingest_log(&mut index, 1, "priority=1 boom"); ingest_log(&mut index, 2, "priority=2 boom"); match handle_query(&index, logs_query_sorted("priority", None), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 2, 0]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_sort_by_missing_param_fallback_orders_by_sequence_and_appends() { let mut index = BTreeMap::new(); declare_test_param( &mut index, "priority", r"priority=(?P\d+)", "priority", MetricPrimitiveKind::Number, ); ingest_log(&mut index, 0, "priority=3 boom"); ingest_log(&mut index, 1, "no param here"); ingest_log(&mut index, 2, "priority=1 boom"); ingest_log(&mut index, 3, "also missing"); match handle_query(&index, logs_query_sorted("priority", Some(true)), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![2, 0, 1, 3]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_sort_by_missing_param_drop_excludes_them() { let mut index = BTreeMap::new(); declare_test_param( &mut index, "priority", r"priority=(?P\d+)", "priority", MetricPrimitiveKind::Number, ); ingest_log(&mut index, 0, "priority=3 boom"); ingest_log(&mut index, 1, "no param here"); ingest_log(&mut index, 2, "priority=1 boom"); ingest_log(&mut index, 3, "also missing"); match handle_query(&index, logs_query_sorted("priority", Some(false)), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![2, 0]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_sort_by_unknown_param_returns_error() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "hello"); match handle_query(&index, logs_query_sorted("does_not_exist", None), 0, 0, &mut HashMap::new()) { EngineResponse::Error(err) => assert!(err.contains("does_not_exist")), _ => panic!("expected EngineResponse::Error"), } } #[test] fn logs_query_without_sort_by_keeps_default_ottyl_ts_order() { let mut index = BTreeMap::new(); declare_test_param( &mut index, "priority", r"priority=(?P\d+)", "priority", MetricPrimitiveKind::Number, ); ingest_log(&mut index, 0, "priority=3 boom"); ingest_log(&mut index, 1, "no param here"); ingest_log(&mut index, 2, "priority=1 boom"); ingest_log(&mut index, 3, "also missing"); match handle_query(&index, logs_query(None), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 1, 2, 3]); } _ => panic!("expected Logs response"), } } #[test] fn sort_helper_index_is_built_lazily_and_reused_across_queries() { let mut index = BTreeMap::new(); declare_test_param( &mut index, "priority", r"priority=(?P\d+)", "priority", MetricPrimitiveKind::Number, ); ingest_log(&mut index, 0, "priority=3 boom"); ingest_log(&mut index, 1, "priority=1 boom"); ingest_log(&mut index, 2, "priority=2 boom"); let mut sort_helper_indexes = HashMap::new(); assert!(!sort_helper_indexes.contains_key("priority")); match handle_query(&index, logs_query_sorted("priority", None), 0, 0, &mut sort_helper_indexes) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 2, 0]); } _ => panic!("expected Logs response"), } assert!(sort_helper_indexes.contains_key("priority")); assert_eq!(sort_helper_indexes["priority"].index.len(), 3); ingest_log(&mut index, 3, "priority=0 boom"); update_sort_helper_indexes_on_ingest( &mut sort_helper_indexes, &index, 3, Timestamp::from_millis(3), "priority=0 boom", ); ingest_log(&mut index, 4, "no priority here"); update_sort_helper_indexes_on_ingest( &mut sort_helper_indexes, &index, 4, Timestamp::from_millis(4), "no priority here", ); assert_eq!(sort_helper_indexes["priority"].index.len(), 5); match handle_query(&index, logs_query_sorted("priority", None), 0, 0, &mut sort_helper_indexes) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![3, 1, 2, 0, 4]); } _ => panic!("expected Logs response"), } } #[test] fn sort_helper_index_incremental_update_is_a_noop_when_no_index_is_live() { let mut sort_helper_indexes: HashMap = HashMap::new(); let index = BTreeMap::new(); update_sort_helper_indexes_on_ingest(&mut sort_helper_indexes, &index, 0, Timestamp::from_millis(0), "hello"); assert!(sort_helper_indexes.is_empty()); } #[test] fn purge_stale_sort_helper_indexes_drops_only_entries_past_the_purge_duration() { let mut sort_helper_indexes: HashMap = HashMap::new(); sort_helper_indexes.insert( "stale".to_string(), SortHelperIndexEntry { index: BTreeMap::new(), last_accessed: std::time::Instant::now() - SORT_HELPER_INDEX_PURGE_AFTER - std::time::Duration::from_secs(1), }, ); sort_helper_indexes.insert( "fresh".to_string(), SortHelperIndexEntry { index: BTreeMap::new(), last_accessed: std::time::Instant::now() }, ); purge_stale_sort_helper_indexes(&mut sort_helper_indexes); assert!(!sort_helper_indexes.contains_key("stale")); assert!(sort_helper_indexes.contains_key("fresh")); } #[test] fn sorted_query_transparently_rebuilds_a_purged_helper_index() { let mut index = BTreeMap::new(); declare_test_param( &mut index, "priority", r"priority=(?P\d+)", "priority", MetricPrimitiveKind::Number, ); ingest_log(&mut index, 0, "priority=3 boom"); ingest_log(&mut index, 1, "priority=1 boom"); ingest_log(&mut index, 2, "priority=2 boom"); let mut sort_helper_indexes = HashMap::new(); let _ = handle_query(&index, logs_query_sorted("priority", None), 0, 0, &mut sort_helper_indexes); assert!(sort_helper_indexes.contains_key("priority")); sort_helper_indexes.remove("priority"); match handle_query(&index, logs_query_sorted("priority", None), 0, 0, &mut sort_helper_indexes) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1, 2, 0]); } _ => panic!("expected Logs response"), } assert!(sort_helper_indexes.contains_key("priority")); } #[test] fn logs_query_with_ottyl_fmt_comparison_filters_by_detected_format_for_mixed_format_lines() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, r#"{"level":"error","code":500}"#); ingest_log(&mut index, 1, r#"ts=1788648105773 msg="entity_moved" x1=-4"#); ingest_log(&mut index, 2, "just some plain prose here"); ingest_log(&mut index, 3, r#"{"level":"info","code":200}"#); match handle_query(&index, logs_query(Some("ottyl_fmt==json")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 3]); } _ => panic!("expected Logs response"), } match handle_query(&index, logs_query(Some("ottyl_fmt==logfmt")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![1]); } _ => panic!("expected Logs response"), } match handle_query(&index, logs_query(Some("ottyl_fmt==none")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![2]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_with_json_field_comparison_filters_by_json_field() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, r#"{"level":"error","code":500}"#); ingest_log(&mut index, 1, r#"{"level":"info","code":200}"#); ingest_log(&mut index, 2, "not json at all"); match handle_query(&index, logs_query(Some(r#"json[level]==error"#)), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); } _ => panic!("expected Logs response"), } match handle_query(&index, logs_query(Some("json[code]>=300")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_combines_regex_and_metric_comparison_with_and() { let mut index = BTreeMap::new(); let pattern = r"level=(?P\w+) duration=(?P\d+)"; declare_test_param(&mut index, "level", pattern, "level", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", pattern, vec!["level"])).unwrap(); ingest_log(&mut index, 0, "level=error duration=1 boom"); ingest_log(&mut index, 1, "level=info duration=2 boom"); ingest_log(&mut index, 2, "level=error duration=3 all good"); match handle_query( &index, logs_query(Some(r#"regex("boom") and requests[level]==error"#)), 0, 0, &mut HashMap::new(), ) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); } _ => panic!("expected Logs response"), } match handle_query( &index, logs_query(Some(r#"regex("boom") or requests[level]==error"#)), 0, 0, &mut HashMap::new(), ) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0, 1, 2]); } _ => panic!("expected Logs response"), } } #[test] fn logs_query_includes_declared_params_on_matching_lines_only() { let mut index = BTreeMap::new(); declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); ingest_log(&mut index, 0, "level=info hello"); ingest_log(&mut index, 1, "no match here"); let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, 0, &mut HashMap::new()) else { panic!("expected Logs response"); }; assert_eq!(entries.len(), 2); assert_eq!( entries[0].params.get("level"), Some(&MetricPrimitive::String("info".to_string())) ); assert!(entries[1].params.get("level").is_none()); } #[test] fn boolean_param_declares_without_capture_group_and_is_always_present() { let mut index = BTreeMap::new(); let mut req = request("IsError", r"(?i)error", ""); req.kind = MetricPrimitiveKind::Boolean; assert!(declare_param(&mut index, req).is_ok()); ingest_log(&mut index, 0, "an ERROR occurred"); ingest_log(&mut index, 1, "everything is fine"); let EngineResponse::Logs(entries) = handle_query(&index, logs_query(None), 0, 0, &mut HashMap::new()) else { panic!("expected Logs response"); }; assert_eq!(entries.len(), 2); assert_eq!(entries[0].params.get("IsError"), Some(&MetricPrimitive::Boolean(true))); assert_eq!(entries[1].params.get("IsError"), Some(&MetricPrimitive::Boolean(false))); } #[test] fn timeline_query_applies_filter_before_bucketing() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "everything is fine"); ingest_log(&mut index, 1, "an error occurred"); ingest_log(&mut index, 2, "another error here"); match handle_query(&index, timeline_query(Some(r#"regex("error")"#), 1_000_000), 0, 0, &mut HashMap::new()) { EngineResponse::Timeline(buckets) => { let total: u64 = buckets.iter().map(|b| b.count).sum(); assert_eq!(total, 2); } _ => panic!("expected Timeline response"), } } #[test] fn malformed_filter_string_surfaces_as_engine_error() { let mut index = BTreeMap::new(); ingest_log(&mut index, 0, "hello"); match handle_query(&index, logs_query(Some("@@@ not a valid filter @@@")), 0, 0, &mut HashMap::new()) { EngineResponse::Error(message) => assert!(!message.is_empty()), _ => panic!("expected an Error response for a malformed filter string"), } match handle_query(&index, timeline_query(Some("@@@ not a valid filter @@@"), 1000), 0, 0, &mut HashMap::new()) { EngineResponse::Error(message) => assert!(!message.is_empty()), _ => panic!("expected an Error response for a malformed filter string"), } } #[test] fn metric_comparison_without_param_is_a_compile_error() { let mut index = BTreeMap::new(); let pattern = r"level=(?P\w+)"; declare_test_param(&mut index, "level", pattern, "level", MetricPrimitiveKind::String); declare_metric(&mut index, metric_request("requests", pattern, vec!["level"])).unwrap(); ingest_log(&mut index, 0, "level=error"); match handle_query(&index, logs_query(Some("requests==5")), 0, 0, &mut HashMap::new()) { EngineResponse::Error(message) => assert!(message.contains("param")), _ => panic!("expected an Error response for a metric comparison without [param]"), } } #[test] fn metric_comparison_with_undeclared_param_is_rejected_naming_real_label_params() { let mut index = BTreeMap::new(); let pattern = r"level=(?P\w+) duration=(?P\d+)"; declare_test_param(&mut index, "level", pattern, "level", MetricPrimitiveKind::String); declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); declare_metric(&mut index, metric_request("requests", pattern, vec!["level", "duration"])).unwrap(); ingest_log(&mut index, 0, "level=error duration=5"); match handle_query(&index, logs_query(Some("requests[nonexistent_param]==\"x\"")), 0, 0, &mut HashMap::new()) { EngineResponse::Error(message) => { assert!( message.contains("nonexistent_param"), "error should name the offending param, got: {message}" ); assert!( message.contains("requests"), "error should name the metric, got: {message}" ); assert!( message.contains("level") && message.contains("duration"), "error should list the metric's real declared label params, got: {message}" ); } _ => panic!("expected an Error response for an undeclared param"), } } #[test] fn metric_comparison_with_valid_label_param_still_matches() { let mut index = BTreeMap::new(); let pattern = r"level=(?P\w+) duration=(?P\d+)"; declare_test_param(&mut index, "level", pattern, "level", MetricPrimitiveKind::String); declare_test_param(&mut index, "duration", pattern, "duration", MetricPrimitiveKind::Number); declare_metric(&mut index, metric_request("requests", pattern, vec!["level", "duration"])).unwrap(); ingest_log(&mut index, 0, "level=error duration=5"); ingest_log(&mut index, 1, "level=info duration=2"); match handle_query(&index, logs_query(Some("requests[level]==error")), 0, 0, &mut HashMap::new()) { EngineResponse::Logs(entries) => { let sequences: Vec = entries.iter().map(|e| e.sequence).collect(); assert_eq!(sequences, vec![0]); } _ => panic!("expected Logs response"), } } }