diff --git a/Cargo.lock b/Cargo.lock index d538438..8cccf2d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -774,6 +774,12 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "mio" version = "1.2.3" @@ -785,6 +791,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -824,6 +840,7 @@ dependencies = [ "http-body-util", "humantime", "include_dir", + "nom", "pest", "pest_derive", "regex", diff --git a/Cargo.toml b/Cargo.toml index c0b2ca8..d77a9b3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" clap = { version = "4", features = ["derive"] } regex = "1" +nom = "7" pest = "2" pest_derive = "2" chrono = "0.4" diff --git a/src/api.rs b/src/api.rs index 0ca5719..89d7aea 100644 --- a/src/api.rs +++ b/src/api.rs @@ -13,7 +13,7 @@ use crate::engine::{ EngineResponse, }; use crate::alerts::{deliver_ntfy, AlertError}; -use crate::profile::{self, Profile, ProfileError, ProfileStore}; +use crate::profile::{self, ParamDeclarationRecord, Profile, ProfileError, ProfileStore}; use crate::settings::{CssError, SettingsHandle}; use axum::extract::{Path as AxumPath, Query, State}; use axum::http::{header, StatusCode}; @@ -55,13 +55,21 @@ impl ApiState { } } - /// Same as `persist_param_declaration`, for metric declarations. - fn persist_metric_declaration(&self, request: &DeclareMetricRequest) { + /// Same as `persist_param_declaration`, for metric declarations. Also promotes any of the + /// metric's referenced params that only exist as auto-derived engine state (see + /// `promoted_param_records`) into persisted profile params, so a metric built against an + /// auto-detected logfmt key survives a restart instead of being rejected as "undeclared" once + /// declarations replay before any log line re-derives it. + async fn persist_metric_declaration(&self, request: &DeclareMetricRequest) { let Some((store_path, profile_name)) = self.active_profile_location() else { return; }; let mut store = ProfileStore::load(&store_path); - if let Err(err) = store.record_metric_declaration(&store_path, &profile_name, request) { + let existing_params = store.get(&profile_name).map(|p| p.params.clone()).unwrap_or_default(); + let promoted = promoted_param_records(&self.engine, &existing_params, request).await; + if let Err(err) = + store.record_metric_declaration(&store_path, &profile_name, request, &promoted) + { eprintln!( "warning: failed to persist metric declaration '{}' into profile '{profile_name}': {err}", request.name @@ -70,14 +78,20 @@ impl ApiState { } /// Same as `persist_metric_declaration`, for an in-place edit (`PUT /api/metrics/{name}`) — also drops the persisted record under `old_name` when the edit renames the metric, so a rename doesn't leave a stale duplicate declaration in the profile file. - fn persist_metric_declaration_update(&self, old_name: &str, request: &DeclareMetricRequest) { + async fn persist_metric_declaration_update(&self, old_name: &str, request: &DeclareMetricRequest) { let Some((store_path, profile_name)) = self.active_profile_location() else { return; }; let mut store = ProfileStore::load(&store_path); - if let Err(err) = - store.record_metric_declaration_update(&store_path, &profile_name, old_name, request) - { + let existing_params = store.get(&profile_name).map(|p| p.params.clone()).unwrap_or_default(); + let promoted = promoted_param_records(&self.engine, &existing_params, request).await; + if let Err(err) = store.record_metric_declaration_update( + &store_path, + &profile_name, + old_name, + request, + &promoted, + ) { eprintln!( "warning: failed to persist metric declaration update '{}' into profile '{profile_name}': {err}", request.name @@ -177,6 +191,47 @@ async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result Vec { + let missing: Vec<&str> = request + .label_params + .iter() + .map(String::as_str) + .chain(request.value_param.as_deref()) + .filter(|name| !existing_params.iter().any(|p| p.name == **name)) + .collect(); + if missing.is_empty() { + return Vec::new(); + } + let Ok(EngineResponse::ParamDeclarations(declarations)) = + ask(engine, EngineQuery::ParamDeclarations).await + else { + return Vec::new(); + }; + declarations + .into_iter() + .filter(|param| missing.contains(¶m.name.as_str())) + .map(|param| ParamDeclarationRecord { + name: param.name, + message_match_regex: param.message_match_regex, + capture_group: param.capture_group, + json_field_path: param.json_field_path, + kind: param.kind.into(), + elapsed_since_last: param.elapsed_since_last, + }) + .collect() +} + async fn get_logs( State(state): State, Query(params): Query, @@ -245,6 +300,9 @@ async fn declare_param( json_field_path: request.json_field_path, kind: request.kind, elapsed_since_last: request.elapsed_since_last, + // A request that reached here was accepted by `declare_param`, which always stores + // it as non-auto (see `engine::declare_param`). + auto_derived: false, }; (StatusCode::CREATED, Json(dto)).into_response() } @@ -277,7 +335,7 @@ async fn declare_metric( } match response.await { Ok(Ok(())) => { - state.persist_metric_declaration(&request); + state.persist_metric_declaration(&request).await; let dto = MetricDeclarationDto { name: request.name, message_match_regex: request.message_match_regex, @@ -311,7 +369,7 @@ async fn update_metric( } match response.await { Ok(Ok(())) => { - state.persist_metric_declaration_update(&name, &request); + state.persist_metric_declaration_update(&name, &request).await; let dto = MetricDeclarationDto { name: request.name, message_match_regex: request.message_match_regex, @@ -602,11 +660,14 @@ async fn delete_profile( } /// Removes every currently-declared param/metric from the engine, then replays `profile`'s declarations into it. +/// Auto-derived params (`ParamDeclarationDto::auto_derived`) are left alone — they aren't part of +/// any profile, just facts discovered from the log stream, so a profile switch shouldn't discard +/// them (they'd otherwise only reappear once a new line re-introduces that key). async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) { if let Ok(EngineResponse::ParamDeclarations(params)) = ask(engine, EngineQuery::ParamDeclarations).await { - for param in params { + for param in params.into_iter().filter(|param| !param.auto_derived) { let (respond_to, response) = oneshot::channel(); if engine .send(EngineMessage::RemoveParamDeclaration { name: param.name, respond_to }) @@ -1206,6 +1267,75 @@ mod tests { assert_eq!(profile.params[0].name, "level"); } + /// Regression test: declaring a metric against a param that only exists as auto-derived engine + /// state (discovered from an ingested logfmt key, never manually declared via `POST + /// /api/params`) must persist that param into the active profile too, not just the metric — + /// otherwise a restart replays the metric before any log line re-derives the auto param, and the + /// engine rejects it as referencing an undeclared param (see `promoted_param_records`). + #[tokio::test] + async fn declaring_metric_against_auto_derived_param_promotes_it_into_the_profile() { + let (app, engine, settings_handle, dir) = test_app(); + + let settings_path = dir.path().join("settings.json"); + let profiles_dir = { + let settings = settings_handle.lock().unwrap(); + profile::resolve_profiles_dir(&settings, &settings_path) + }; + let store_path = profile::profiles_store_path(&profiles_dir); + let mut store = ProfileStore::load(&store_path); + { + let mut settings = settings_handle.lock().unwrap(); + store + .create(&store_path, &mut settings, &settings_path, "web".to_string()) + .unwrap(); + store + .set_active(&mut settings, &settings_path, Some("web".to_string())) + .unwrap(); + } + + // Ingest a line carrying a logfmt key before any param is manually declared, so "dx" only + // exists as engine-internal auto-derived state (never persisted anywhere). + ingest_lines(&engine, 0, &["dx=1 moved"]).await; + + let (status, _body) = send_json( + &app, + "POST", + "/api/metrics", + json!({ + "name": "direction_x", + "message_match_regex": r"dx=(?P\S+) moved", + "label_params": ["dx"] + }), + ) + .await; + assert_eq!(status, StatusCode::CREATED); + + let reloaded_store = ProfileStore::load(&store_path); + let profile = reloaded_store.get("web").unwrap(); + assert_eq!(profile.metrics.len(), 1); + assert_eq!(profile.metrics[0].name, "direction_x"); + // The auto-derived "dx" param got promoted into the profile alongside the metric. + assert_eq!(profile.params.len(), 1); + assert_eq!(profile.params[0].name, "dx"); + + // Simulate a restart: replay the persisted profile into a fresh engine with no prior ingestion. + let (fresh_engine_tx, fresh_engine_rx) = mpsc::channel(16); + let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); + tokio::spawn(crate::engine::run(fresh_engine_rx, fresh_settings_handle)); + profile::load_into_engine(&fresh_engine_tx, profile).await; + + let (respond_to, response) = oneshot::channel(); + fresh_engine_tx + .send(EngineMessage::Query { query: EngineQuery::MetricDeclarations, respond_to }) + .await + .unwrap(); + let EngineResponse::MetricDeclarations(metrics) = response.await.unwrap() else { + panic!("unexpected response"); + }; + assert_eq!(metrics.len(), 1); + assert_eq!(metrics[0].name, "direction_x"); + } + async fn delete_request(app: &Router, uri: &str) -> StatusCode { let request = Request::builder().method("DELETE").uri(uri).body(Body::empty()).unwrap(); app.clone().oneshot(request).await.unwrap().status() diff --git a/src/data_value.rs b/src/data_value.rs index 1d88ba7..62dab2f 100644 --- a/src/data_value.rs +++ b/src/data_value.rs @@ -33,6 +33,14 @@ pub struct ParamDeclaration { /// regex/JSON-field extraction. #[serde(default)] pub elapsed_since_last: bool, + /// Set when this param was discovered by `crate::engine::auto_declare_params` (a nom-based + /// `key=value` scan run on every ingested line) rather than declared through `POST /api/params`. + /// `declare_param` refuses to overwrite an auto-derived param by name, and it's never persisted + /// to a profile — see `crate::engine::declare_param` and `crate::api::reload_engine_declarations`. + /// `#[serde(default)]` so profiles/requests saved before this field existed still load as + /// ordinary (non-auto) params. + #[serde(default)] + pub auto_derived: bool, } /// Reads a value out of a JSON object by dot-separated field path (e.g. `"foo.bar"`), for diff --git a/src/dto.rs b/src/dto.rs index 6dec5d1..3f010f8 100644 --- a/src/dto.rs +++ b/src/dto.rs @@ -25,6 +25,9 @@ pub struct ParamDeclarationDto { pub kind: MetricPrimitiveKind, /// See `crate::data_value::ParamDeclaration::elapsed_since_last`. pub elapsed_since_last: bool, + /// See `crate::data_value::ParamDeclaration::auto_derived`. The frontend uses this to disable + /// editing — `POST /api/params` also rejects a request naming an auto-derived param. + pub auto_derived: bool, } #[derive(Debug, Clone, Serialize)] diff --git a/src/engine.rs b/src/engine.rs index b1bb33e..5a01237 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -152,6 +152,7 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle DataValue::LogLiteral { text: text.clone() }, ); log_count += 1; + auto_declare_params(&mut index, &text); apply_all_metrics(&mut index, sequence, timestamp, &text); evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); @@ -219,6 +220,16 @@ fn declare_param( index: &mut BTreeMap, request: DeclareParamRequest, ) -> Result<(), String> { + 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 + )); + } + } // An `elapsed_since_last` param is never text-extracted at all, so `message_match_regex` is // irrelevant to it — but the field always compiles as a regex regardless of extractor (an empty // pattern is valid and matches trivially), so it's harmless to still require it to parse. @@ -243,11 +254,78 @@ fn declare_param( index.insert( IndexKey::ParamDeclaration { name: request.name.clone() }, - DataValue::ParamDeclaration(request), + // `auto_derived` is never trusted from the request — only `auto_declare_params` may set it. + DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }), ); Ok(()) } +/// Turns an arbitrary log-line key into a valid regex named-capture-group identifier (ASCII +/// alphanumeric/underscore, not starting with a digit) — the param's own `name` (shown to the user, +/// used as the index key) keeps the raw key text; only the capture group embedded in its generated +/// regex needs sanitizing. Two different keys sanitizing to the same group name is harmless since +/// each param's regex/capture-group pair is private to that one param. +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 +} + +/// Conservative kind inference for an auto-derived param's first sample: only `Number` is guessed +/// (int, falling back to float), everything else stays `String` — `Boolean`/`Timestamp`/`Duration` +/// carry extraction semantics beyond "parses as", so guessing them wrong would silently drop values +/// that don't fit (see `parse_metric_primitive`). +fn infer_kind(sample: &str) -> MetricPrimitiveKind { + if sample.parse::().is_ok() || sample.parse::().is_ok() { + MetricPrimitiveKind::Number + } else { + MetricPrimitiveKind::String + } +} + +/// Builds the regex an auto-derived param uses to re-extract its value from any line on read +/// (`derive_log_params`) — quoted keys always look for a `"..."` value, unquoted keys a +/// whitespace-delimited bareword, matching how `crate::logfmt::extract_pairs` found the sample. +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+)") + } +} + +/// Scans `text` for `key=value`/`key="value"` pairs (`crate::logfmt::extract_pairs`) and declares a +/// standalone param for any key not already declared — auto or manual. A manual declaration for a +/// name always wins and is never touched here; this only fills in names nobody has declared yet. +/// Run on every ingested line (`EngineMessage::Ingest`), same as `apply_all_metrics`, but note param +/// *values* are still recomputed from raw text at read time (`derive_log_params`), not stored here — +/// so a key discovered on line N also back-fills the column for lines before N on the next read. +fn auto_declare_params(index: &mut BTreeMap, text: &str) { + for (key, value, quoted) in crate::logfmt::extract_pairs(text) { + let index_key = IndexKey::ParamDeclaration { name: key.to_string() }; + if index.contains_key(&index_key) { + continue; + } + let group = sanitize_group_name(key); + index.insert( + index_key, + DataValue::ParamDeclaration(ParamDeclaration { + name: key.to_string(), + message_match_regex: build_auto_param_pattern(key, &group, quoted), + capture_group: group, + json_field_path: None, + kind: infer_kind(value), + elapsed_since_last: false, + auto_derived: true, + }), + ); + } +} + /// Validates the metric's own regex plus that every name in `label_params`/`value_param` refers to /// an already-declared standalone param (not just the first bad one, so a rejection reports every /// problem at once). Shared by `declare_metric` and `update_metric` so both reject a bad @@ -407,7 +485,7 @@ struct ResolvedParam { 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)| { + .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, _)| { Regex::new(message_match_regex).ok().map(|regex| { ( name.to_string(), @@ -956,19 +1034,20 @@ fn metric_declarations_range( fn param_declarations_range( index: &BTreeMap, -) -> impl Iterator, MetricPrimitiveKind, bool)> { +) -> impl Iterator, MetricPrimitiveKind, 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, }), ) => Some(( name.as_str(), @@ -977,6 +1056,7 @@ fn param_declarations_range( json_field_path.as_deref(), *kind, *elapsed_since_last, + *auto_derived, )), _ => None, }) @@ -991,8 +1071,8 @@ fn compiled_param_declarations( index: &BTreeMap, ) -> Vec<(String, Regex, String, Option, MetricPrimitiveKind)> { param_declarations_range(index) - .filter(|(.., elapsed_since_last)| !elapsed_since_last) - .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _)| { + .filter(|(_, _, _, _, _, elapsed_since_last, _)| !elapsed_since_last) + .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _, _)| { Regex::new(message_match_regex).ok().map(|re| { ( name.to_string(), @@ -1198,7 +1278,7 @@ fn handle_query( } EngineQuery::ParamDeclarations => { let declarations = param_declarations_range(index) - .map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last)| { + .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(), @@ -1206,6 +1286,7 @@ fn handle_query( json_field_path: json_field_path.map(|path| path.to_string()), kind, elapsed_since_last, + auto_derived, } }) .collect(); @@ -1578,6 +1659,7 @@ mod tests { json_field_path: None, kind: MetricPrimitiveKind::String, elapsed_since_last: false, + auto_derived: false, } } @@ -1594,6 +1676,51 @@ mod tests { 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_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"); + } + #[test] fn rejects_invalid_regex() { let mut index = BTreeMap::new(); @@ -1635,6 +1762,7 @@ mod tests { json_field_path: None, kind: MetricPrimitiveKind::String, elapsed_since_last: true, + auto_derived: false, }, ); assert!(result.is_err()); @@ -1655,6 +1783,7 @@ mod tests { json_field_path: Some("foo.bar".to_string()), kind: MetricPrimitiveKind::String, elapsed_since_last: false, + auto_derived: false, }, ); assert!(result.is_ok()); @@ -1676,6 +1805,7 @@ mod tests { json_field_path: Some("foo.bar".to_string()), kind: MetricPrimitiveKind::Number, elapsed_since_last: false, + auto_derived: false, }, ) .unwrap(); @@ -1792,6 +1922,7 @@ mod tests { json_field_path: None, kind, elapsed_since_last: false, + auto_derived: false, }, ) .unwrap(); @@ -1811,6 +1942,7 @@ mod tests { json_field_path: None, kind: MetricPrimitiveKind::Duration, elapsed_since_last: true, + auto_derived: false, }, ) .unwrap(); @@ -2032,6 +2164,7 @@ mod tests { json_field_path: Some("request.path".to_string()), kind: MetricPrimitiveKind::String, elapsed_since_last: false, + auto_derived: false, }, ) .unwrap(); diff --git a/src/logfmt.rs b/src/logfmt.rs new file mode 100644 index 0000000..7a44a23 --- /dev/null +++ b/src/logfmt.rs @@ -0,0 +1,111 @@ +//! Fast, tolerant `key=value`/`key="value"` tokenizer (logfmt-ish), used by +//! `crate::engine::auto_declare_params` to discover param names as lines are ingested. Scans +//! anywhere in the line rather than requiring the whole line to be logfmt, so it works on lines +//! with an arbitrary non-logfmt prefix/suffix (e.g. `ts=... msg="..." x1=-4 y1=34`). + +use nom::branch::alt; +use nom::bytes::complete::{escaped, take_while1}; +use nom::character::complete::{char, none_of, one_of}; +use nom::sequence::delimited; +use nom::IResult; + +fn key(input: &str) -> IResult<&str, &str> { + take_while1(|c: char| c.is_ascii_alphanumeric() || c == '_' || c == '.' || c == '-')(input) +} + +fn quoted_value(input: &str) -> IResult<&str, &str> { + delimited( + char('"'), + alt((escaped(none_of("\\\""), '\\', one_of("\"\\nt")), nom::combinator::success(""))), + char('"'), + )(input) +} + +fn bareword_value(input: &str) -> IResult<&str, &str> { + take_while1(|c: char| !c.is_whitespace())(input) +} + +/// One `key=value` pair found on a line: the key, the raw captured value (unescaped as-is, quotes +/// stripped but any inner `\"`/`\\` left literal since callers only need it for numeric-kind +/// inference), and whether the value was quoted. +fn pair(input: &str) -> IResult<&str, (&str, &str, bool)> { + let (input, k) = key(input)?; + let (input, _) = char('=')(input)?; + if let Ok((rest, v)) = quoted_value(input) { + Ok((rest, (k, v, true))) + } else { + let (rest, v) = bareword_value(input)?; + Ok((rest, (k, v, false))) + } +} + +/// Finds every `key=value`/`key="value"` pair anywhere in `line`. A position that doesn't start a +/// pair is skipped one char at a time, so this never fails outright — a line with no pairs at all +/// (plain prose, a JSON blob) just returns an empty vec. +pub fn extract_pairs(line: &str) -> Vec<(&str, &str, bool)> { + let mut results = Vec::new(); + let mut rest = line; + while !rest.is_empty() { + match pair(rest) { + Ok((remaining, found)) => { + results.push(found); + rest = remaining; + } + Err(_) => { + let mut chars = rest.chars(); + chars.next(); + rest = chars.as_str(); + } + } + } + results +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extracts_quoted_value() { + let pairs = extract_pairs(r#"msg="entity_moved""#); + assert_eq!(pairs, vec![("msg", "entity_moved", true)]); + } + + #[test] + fn extracts_unquoted_bareword_value() { + let pairs = extract_pairs("x1=-4"); + assert_eq!(pairs, vec![("x1", "-4", false)]); + } + + #[test] + fn extracts_mixed_quoted_and_unquoted_pairs_in_one_line() { + let pairs = extract_pairs(r#"ts=1788648105773 msg="entity_moved" x1=-4 y1=34"#); + assert_eq!( + pairs, + vec![ + ("ts", "1788648105773", false), + ("msg", "entity_moved", true), + ("x1", "-4", false), + ("y1", "34", false), + ] + ); + } + + #[test] + fn returns_empty_for_lines_with_no_pairs() { + assert!(extract_pairs("just some plain prose here").is_empty()); + assert!(extract_pairs(r#"{"foo": {"bar": 42}}"#).is_empty()); + } + + #[test] + fn does_not_panic_on_empty_value_at_end_of_line() { + assert_eq!(extract_pairs("key="), Vec::<(&str, &str, bool)>::new()); + assert_eq!(extract_pairs(r#"key="""#), vec![("key", "", true)]); + } + + #[test] + fn keeps_escaped_quote_inside_quoted_value() { + let pairs = extract_pairs(r#"msg="he said \"hi\"""#); + assert_eq!(pairs, vec![("msg", r#"he said \"hi\""#, true)]); + } +} diff --git a/src/main.rs b/src/main.rs index d01d55e..8823fba 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,6 +7,7 @@ mod filter_datetime; mod filter_parser; mod index_key; mod ingest; +mod logfmt; mod metric_primitive; mod profile; mod promql; diff --git a/src/profile.rs b/src/profile.rs index 6e7b3cc..6096014 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -98,6 +98,8 @@ impl From<&ParamDeclarationRecord> for DeclareParamRequest { json_field_path: record.json_field_path.clone(), kind: record.kind.into(), elapsed_since_last: record.elapsed_since_last, + // A profile only ever stores manually-declared params (see `record_param_declaration`). + auto_derived: false, } } } @@ -682,17 +684,32 @@ impl ProfileStore { } /// Records/updates a declared metric into `profile_name`'s stored declarations (matched by metric name), persisting the store; same integration intent as `record_param_declaration`. + /// + /// `promoted_params` is any of the metric's referenced `label_params`/`value_param` that live + /// only as auto-derived engine state (see `crate::engine::auto_declare_params`) rather than a + /// persisted profile param — the caller (`ApiState::persist_metric_declaration`) resolves these + /// against the live engine before calling in. Without persisting them alongside the metric, a + /// restart would replay the metric before any log line re-derives its auto params, and the + /// engine would reject it as referencing an undeclared param. Each is added only if `profile + /// .params` doesn't already have an entry of that name, so an existing (e.g. manually-declared) + /// param is never clobbered. pub fn record_metric_declaration( &mut self, store_path: &Path, profile_name: &str, request: &DeclareMetricRequest, + promoted_params: &[ParamDeclarationRecord], ) -> Result<(), ProfileError> { let profile = self .profiles .iter_mut() .find(|p| p.name == profile_name) .ok_or(ProfileError::NotFound)?; + for param in promoted_params { + if !profile.params.iter().any(|p| p.name == param.name) { + profile.params.push(param.clone()); + } + } let record: MetricDeclarationRecord = request.into(); match profile.metrics.iter_mut().find(|r| r.name == record.name) { Some(existing) => *existing = record, @@ -702,13 +719,14 @@ impl ProfileStore { Ok(()) } - /// Same as `record_metric_declaration`, for an in-place edit that may rename the metric: drops any persisted record under `old_name` first (when it differs from the new name) so a rename doesn't leave a stale duplicate declaration behind, then upserts by the new name as usual. + /// Same as `record_metric_declaration`, for an in-place edit that may rename the metric: drops any persisted record under `old_name` first (when it differs from the new name) so a rename doesn't leave a stale duplicate declaration behind, then upserts by the new name as usual. `promoted_params` has the same meaning as in `record_metric_declaration`. pub fn record_metric_declaration_update( &mut self, store_path: &Path, profile_name: &str, old_name: &str, request: &DeclareMetricRequest, + promoted_params: &[ParamDeclarationRecord], ) -> Result<(), ProfileError> { let profile = self .profiles @@ -718,6 +736,11 @@ impl ProfileStore { if old_name != request.name { profile.metrics.retain(|r| r.name != old_name); } + for param in promoted_params { + if !profile.params.iter().any(|p| p.name == param.name) { + profile.params.push(param.clone()); + } + } let record: MetricDeclarationRecord = request.into(); match profile.metrics.iter_mut().find(|r| r.name == record.name) { Some(existing) => *existing = record, @@ -815,6 +838,7 @@ mod tests { json_field_path: None, kind: MetricPrimitiveKind::String, elapsed_since_last: false, + auto_derived: false, } } @@ -1080,7 +1104,7 @@ mod tests { store.record_param_declaration(&store_path, "web", ¶m_req).unwrap(); let metric_req = metric_request("requests", r"duration=(?P\d+)", "duration"); - store.record_metric_declaration(&store_path, "web", &metric_req).unwrap(); + store.record_metric_declaration(&store_path, "web", &metric_req, &[]).unwrap(); let profile = store.get("web").unwrap(); assert_eq!(profile.params.len(), 1); @@ -1134,6 +1158,7 @@ mod tests { &store_path, "web", &metric_request("requests", r"request duration=(?P\d+)", "duration"), + &[], ) .unwrap(); store.set_active(&mut settings, &settings_path, Some("web".to_string())).unwrap(); @@ -1465,7 +1490,7 @@ mod tests { ) .unwrap(); store - .record_metric_declaration(&store_path, "web", &metric_request("requests", r"path=(?P\S+)", "path")) + .record_metric_declaration(&store_path, "web", &metric_request("requests", r"path=(?P\S+)", "path"), &[]) .unwrap(); let before = store.get("web").unwrap().clone(); @@ -1473,4 +1498,83 @@ mod tests { let after = reloaded.get("web").unwrap().clone(); assert_eq!(before, after); } + + /// Acceptance criterion: a metric declared against a param that only exists as auto-derived + /// engine state (not yet a persisted profile param) still ingests correctly after a simulated + /// restart, because `record_metric_declaration`'s `promoted_params` persists that param into the + /// profile alongside the metric. Without this, `load_into_engine` would reject the metric on + /// restart since declarations replay before any log line re-derives the auto param. + #[tokio::test] + async fn metric_referencing_auto_derived_param_survives_restart_via_promotion() { + let dir = tempfile::tempdir().unwrap(); + let store_path = dir.path().join("profiles.json"); + let settings_path = dir.path().join("settings.json"); + let mut settings = Settings::default(); + + let mut store = ProfileStore::load(&store_path); + store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap(); + store.set_active(&mut settings, &settings_path, Some("web".to_string())).unwrap(); + + // Simulate the auto-derive path: ingest a line carrying a logfmt key before any param is + // manually declared, so `dx` only exists as engine-internal auto-derived state. + let (engine_tx, engine_rx) = mpsc::channel(16); + let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone())); + tokio::spawn(crate::engine::run(engine_rx, settings_handle)); + engine_tx + .send(EngineMessage::Ingest(IngestedLine { + sequence: 0, + timestamp: Timestamp::from_millis(0), + text: "dx=1 moved".to_string(), + })) + .await + .unwrap(); + + // Declare a metric referencing the auto-derived "dx" param — this succeeds live (the param + // exists in the engine), mirroring the interactive-declare flow that hit this bug. + let metric_req = metric_request("direction_x", r"dx=(?P\S+) moved", "dx"); + let (respond_to, response) = oneshot::channel(); + engine_tx + .send(EngineMessage::DeclareMetric { request: metric_req.clone(), respond_to }) + .await + .unwrap(); + response.await.unwrap().unwrap(); + + // Promote "dx" from its live (auto-derived) engine state, as `ApiState::persist_metric_declaration` does. + let EngineResponse::ParamDeclarations(live_params) = + ask(&engine_tx, EngineQuery::ParamDeclarations).await + else { + panic!("unexpected response"); + }; + let dx = live_params.iter().find(|p| p.name == "dx").unwrap(); + assert!(dx.auto_derived); + let promoted = ParamDeclarationRecord { + name: dx.name.clone(), + message_match_regex: dx.message_match_regex.clone(), + capture_group: dx.capture_group.clone(), + json_field_path: dx.json_field_path.clone(), + kind: dx.kind.into(), + elapsed_since_last: dx.elapsed_since_last, + }; + store.record_metric_declaration(&store_path, "web", &metric_req, &[promoted]).unwrap(); + + // Simulate a restart: drop everything, reload from disk into a *fresh* engine with no prior ingestion. + drop(store); + let reloaded_store = ProfileStore::load(&store_path); + let profile = reloaded_store.get("web").unwrap().clone(); + assert_eq!(profile.params.len(), 1); + assert_eq!(profile.params[0].name, "dx"); + + let (fresh_engine_tx, fresh_engine_rx) = mpsc::channel(16); + let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); + tokio::spawn(crate::engine::run(fresh_engine_rx, fresh_settings_handle)); + load_into_engine(&fresh_engine_tx, &profile).await; + + let EngineResponse::MetricDeclarations(metrics) = + ask(&fresh_engine_tx, EngineQuery::MetricDeclarations).await + else { + panic!("unexpected response"); + }; + assert_eq!(metrics.len(), 1); + assert_eq!(metrics[0].name, "direction_x"); + } } diff --git a/src/promql.rs b/src/promql.rs index 0c3d237..5027113 100644 --- a/src/promql.rs +++ b/src/promql.rs @@ -289,6 +289,10 @@ impl<'a> Parser<'a> { _ => None, }; self.skip_whitespace(); + // `sum by (label) (metric)`: a `by (...)` clause may also precede the aggregation's + // parenthesized argument, not just follow it. + let prefix_by_labels = if agg.is_some() { self.parse_by_clause()? } else { Vec::new() }; + self.skip_whitespace(); if agg.is_some() && self.peek() == Some('(') { self.expect_char('(')?; self.skip_whitespace(); @@ -323,7 +327,15 @@ impl<'a> Parser<'a> { }; self.expect_char(')')?; inner.agg = agg; - inner.by_labels = self.parse_by_clause()?; + let trailing_by_labels = self.parse_by_clause()?; + if !prefix_by_labels.is_empty() && !trailing_by_labels.is_empty() { + return Err( + "by (...) clause may only appear once, either before or after the aggregation's argument" + .to_string(), + ); + } + inner.by_labels = + if prefix_by_labels.is_empty() { trailing_by_labels } else { prefix_by_labels }; Ok(inner) } else { // Rewind and parse as a plain vector selector, covering metric names that collide with an agg_op keyword. @@ -547,4 +559,23 @@ mod tests { fn rejects_unterminated_by_clause() { assert!(parse_query("sum(requests) by (path").is_err()); } + + #[test] + fn parses_sum_with_prefix_by_clause() { + let query = parse_query("sum by (path) (requests)").unwrap(); + assert_eq!(query.agg, Some(AggOp::Sum)); + assert_eq!(query.by_labels, vec!["path".to_string()]); + } + + #[test] + fn parses_sum_with_prefix_by_clause_multiple_labels() { + let query = parse_query("sum by (path, method) (requests)").unwrap(); + assert_eq!(query.by_labels, vec!["path".to_string(), "method".to_string()]); + } + + #[test] + fn rejects_by_clause_both_before_and_after_aggregation() { + let err = parse_query("sum by (path) (requests) by (method)").unwrap_err(); + assert!(err.contains("only appear once")); + } }