From 80f1611de880531104f6c97dfc243d79b7258055 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 18 May 2026 05:38:24 -0600 Subject: [PATCH] fix(data_plane): make format_series_key + parse_labels_from_series_key roundtrip cleanly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `format_series_key` (in `drivers/ingest/otel.rs`) emitted unquoted `key=value` pairs, but `parse_labels_from_series_key` (in `precompute_engine/worker.rs`) required quoted `key="value"`. The mismatch meant `IngestState::extract_group_key_for` returned `""` for every OTLP wire-format input, and any other consumer that roundtripped a freshly-formatted series key through the parser got an empty label map back. Discovery: PR #284 (B7.6 ingest sid rekey) noticed the bug while implementing the bucketing rekey, side-stepped it by having the new helper read `point.labels` directly via HashMap lookup, and left a TODO-style note in the test docstring. Bucketing was unaffected because B7.6 routes by sid; emit-time `KeyByLabelValues` content elsewhere was silently broken. Decision: option (B) — bend `format_series_key`. The quoted PromQL- style `metric{k="v",...}` form is the canonical shape every other producer / consumer in the data plane already uses: * `render_series_key` in `storage_engines/sketch_db/backfill/prometheus_reader.rs` emits quoted-with-escapes (matches the Prometheus wire format) * `RawSample.labels`' rustdoc documents the quoted form * `sample_matches` in `storage_engines/sketch_db/backfill/raw_sample_reader.rs` strips `"` from values when parsing — expects quoted * Every existing parser test passes the quoted form * Nothing persists `format_series_key`'s output to disk (it flows into in-memory `WorkerMessage::GroupSamples` payloads and debug log lines only) `format_series_key` now escapes embedded `"`, `\`, `\n` per the PromQL lexer rules (matching `render_series_key`'s `escape_label_value`). The parser walks past `\` escape pairs when scanning for the closing quote so values containing embedded `"` no longer terminate early. The returned `&str` is still the raw (un-decoded) slice — a new `decode_label_value(&str) -> Cow` helper unescapes when needed. Returning the un-decoded slice keeps the existing `HashMap<&str, &str>` API (and its `processor.rs` caller, which is off-limits this PR for the B7.7 parallel agent) working without change; most live callers compare against literal config values that never contain escapable characters, so the borrow is fine. Regression coverage (new `series_key_roundtrip_tests` module in `otel.rs` + new unit tests in `worker.rs`): * canonical PromQL quoted shape pinned * roundtrip with commas in value * roundtrip with equals in value * roundtrip with embedded `"` (exercises the escape pair scan + `decode_label_value`) * roundtrip with `\` in value * roundtrip with `\n` in value * roundtrip with all metacharacters in one value * empty-labels bare-braces case * `decode_label_value` borrows when no escapes, unescapes when present, passes unknown escapes through verbatim All 725 `cargo test -p data_plane --lib` tests pass (2 pre-existing ignored, unchanged). Updated the stale "this returns empty" note in B7.6's `raw_otlp_buckets_by_sid_with_distinct_group_keys` test docstring to point at this fix. Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/drivers/ingest/otel.rs | 187 +++++++++++++++++++-- data_plane/src/precompute_engine/worker.rs | 120 ++++++++++++- 2 files changed, 288 insertions(+), 19 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 05914e53..015418f7 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -333,17 +333,61 @@ pub struct SketchPoint { type OtlpParseResult = (Vec, Vec); +/// Render OTLP `(name, labels)` into the canonical PromQL-style series +/// key `metric{k1="v1",k2="v2"}` used everywhere in the data plane. +/// +/// Values are wrapped in double quotes and embedded `"`, `\`, `\n` are +/// escaped per the PromQL lexer rules. This matches: +/// +/// * `parse_labels_from_series_key` in `precompute_engine/worker.rs` +/// (the inverse — expects `key="value"`) +/// * `render_series_key` in `storage_engines/sketch_db/backfill/ +/// prometheus_reader.rs` (the other producer of this shape) +/// * `RawSample.labels`' documented shape (`metric{k="v",...}`) +/// * `sample_matches` in the backfill raw reader (strips `"` when +/// parsing) +/// +/// Pre-fix this helper emitted **unquoted** values (`k=v`), which the +/// parser silently rejected → `IngestState::extract_group_key_for` +/// returned `""` for every OTLP wire-format input, and the keyed +/// dispatch path inside `apply_sample` saw empty aggregated keys for +/// every OTLP sample. The bug was discovered while implementing PR +/// #284 (B7.6 ingest sid rekey); that PR side-stepped it by reading +/// `point.labels` directly, but emit-time `KeyByLabelValues` +/// content elsewhere depended on the roundtrip working — hence this +/// fix. See the regression test +/// `format_series_key_roundtrips_through_parse_labels` in +/// `precompute_engine/worker.rs`. fn format_series_key(name: &str, labels: &HashMap) -> String { let mut pairs: Vec<_> = labels.iter().collect(); pairs.sort_by_key(|(k, _)| *k); let labels_str = pairs .iter() - .map(|(k, v)| format!("{}={}", k, v)) + .map(|(k, v)| format!("{}=\"{}\"", k, escape_label_value(v))) .collect::>() .join(","); format!("{}{{{}}}", name, labels_str) } +/// Escape a label value for the PromQL series-key format. Mirrors +/// `storage_engines::sketch_db::backfill::prometheus_reader:: +/// escape_label_value` — both producers must stay in lockstep so the +/// parser in `precompute_engine::worker::parse_labels_from_series_key` +/// sees a consistent shape regardless of which ingest path emitted +/// the series key. +fn escape_label_value(v: &str) -> String { + let mut out = String::with_capacity(v.len()); + for c in v.chars() { + match c { + '\\' => out.push_str("\\\\"), + '"' => out.push_str("\\\""), + '\n' => out.push_str("\\n"), + other => out.push(other), + } + } + out +} + fn get_sketch_payload_from_attrs( attrs: &[asap_otel_proto::tonic::common::v1::KeyValue], ) -> Option<(String, Vec)> { @@ -2144,6 +2188,118 @@ fn attributes_to_map( m } +#[cfg(test)] +mod series_key_roundtrip_tests { + //! Regression coverage for the `format_series_key` ↔ + //! `parse_labels_from_series_key` roundtrip bug discovered while + //! shipping PR #284 (B7.6 ingest sid rekey). The formatter + //! emitted unquoted `k=v` pairs but the parser required + //! `k="v"`, which silently produced empty group keys for every + //! OTLP wire-format input. These tests pin the canonical + //! PromQL-style quoted format the data plane now uses + //! throughout (see also `render_series_key` in + //! `storage_engines/sketch_db/backfill/prometheus_reader.rs`). + use super::*; + use crate::precompute_engine::worker::{decode_label_value, parse_labels_from_series_key}; + + fn roundtrip(name: &str, input: &[(&str, &str)]) { + let labels: HashMap = input + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + let key = format_series_key(name, &labels); + let parsed = parse_labels_from_series_key(&key); + for (k, v) in input { + let got = parsed + .get(*k) + .map(|raw| decode_label_value(raw).into_owned()) + .unwrap_or_else(|| panic!("label '{}' missing after roundtrip; key={}", k, key)); + assert_eq!( + got, *v, + "label '{}' value mismatch after roundtrip; key={}", + k, key + ); + } + assert_eq!( + parsed.len(), + input.len(), + "label count mismatch after roundtrip; key={} parsed={:?}", + key, + parsed + ); + } + + #[test] + fn format_series_key_emits_promql_quoted_form() { + // The canonical shape every downstream parser + // (`parse_labels_from_series_key`, `sample_matches`) expects. + let mut labels = HashMap::new(); + labels.insert("svc".to_string(), "auth".to_string()); + labels.insert("env".to_string(), "prod".to_string()); + let key = format_series_key("latency", &labels); + // Keys are sorted lexicographically so the formatted output + // is deterministic regardless of HashMap iteration order. + assert_eq!(key, r#"latency{env="prod",svc="auth"}"#); + } + + #[test] + fn roundtrip_simple_alphanumeric() { + roundtrip("metric", &[("svc", "auth"), ("env", "prod")]); + } + + #[test] + fn roundtrip_value_with_comma() { + // Comma is the pair delimiter — quoting must keep it inside + // the value. Pre-fix the unquoted format would split mid-value. + roundtrip("metric", &[("tag", "a,b,c"), ("svc", "auth")]); + } + + #[test] + fn roundtrip_value_with_equals() { + // Equals is the key/value delimiter — quoting must protect it. + roundtrip("metric", &[("expr", "x=y"), ("svc", "auth")]); + } + + #[test] + fn roundtrip_value_with_embedded_quote() { + // `"` must be escaped as `\"` on emit and decoded back on + // read. The parser's closing-quote scan must walk past the + // escape; `decode_label_value` un-escapes the slice. + roundtrip("metric", &[("msg", r#"hello "world""#), ("svc", "auth")]); + } + + #[test] + fn roundtrip_value_with_backslash() { + roundtrip("metric", &[("path", r"C:\Users\app"), ("svc", "auth")]); + } + + #[test] + fn roundtrip_value_with_newline() { + // `\n` round-trips through the `\n` escape; verifies the + // decoder handles all three escape body variants. + roundtrip("metric", &[("multi", "line1\nline2"), ("svc", "auth")]); + } + + #[test] + fn roundtrip_value_with_all_metacharacters() { + // One stress case combining every escape body and every + // pair-delimiter character in a single value. + roundtrip( + "metric", + &[("payload", "a,b=c\"d\\e\nf"), ("svc", "auth")], + ); + } + + #[test] + fn empty_labels_yield_bare_braces() { + let labels: HashMap = HashMap::new(); + let key = format_series_key("metric", &labels); + assert_eq!(key, "metric{}"); + let parsed = parse_labels_from_series_key(&key); + assert!(parsed.is_empty()); + } +} + #[cfg(test)] mod policy_fp_lookup_tests { use super::*; @@ -2722,16 +2878,17 @@ mod sid_bucketing_tests { /// - samples in each bucket are exactly the DPs whose `zone` /// attribute matches that bucket (the GROUP-BY semantic) /// - /// Note: the `group_key` field on the message currently comes from - /// `IngestState::extract_group_key_for(series_key, config)`, and - /// that helper has a pre-existing label-parsing inconsistency with - /// `format_series_key` (one quotes values, the other doesn't), so - /// it currently returns the empty string for OTLP wire-format - /// inputs. Bucketing is unaffected because B7.6 routes by sid (read - /// directly from `point.labels`, not the joined series_key); - /// fixing the group_key-extraction bug is a separate task and - /// would update emit-time label rendering, not the routing - /// contract this test pins. + /// Note: the `group_key` field on the message comes from + /// `IngestState::extract_group_key_for(series_key, config)`. Pre- + /// PR-after-#284 a quoting mismatch between `format_series_key` + /// (unquoted) and `parse_labels_from_series_key` (quoted) caused + /// this to return the empty string for OTLP wire inputs; the + /// follow-up PR fixed the formatter to emit the canonical + /// PromQL `k="v"` form and added a roundtrip regression in + /// `series_key_roundtrip_tests`. Bucketing was always correct + /// here because B7.6 routes by sid (read directly from + /// `point.labels`, not the joined series_key) — the + /// group_key value is informational only for this test. #[tokio::test] async fn raw_otlp_buckets_by_sid_with_distinct_group_keys() { // Channel large enough to capture all routed messages without @@ -2813,10 +2970,10 @@ mod sid_bucketing_tests { // Each sid must match what the resolver records for its bucket // identity: (metric, "zone=;", ExactAgg-canonical). Use // the bucket's sample values to identify which zone it - // represents (group_key is currently empty due to the - // unrelated extract_group_key_for inconsistency — see the - // test-level doc above), then verify the sid matches the - // resolver mint for THAT zone. + // represents (sample-value-based identification is robust + // regardless of group_key shape — the test pin is on sid + // assignment, not on group_key content), then verify the + // sid matches the resolver mint for THAT zone. let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg { agg_type: cfg.aggregation_type, parameters_canonical: diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 2d9efc28..d4680ffb 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -929,6 +929,19 @@ pub fn extract_key_from_series(series_key: &str, config: &AggregationConfig) -> /// Parse label key-value pairs from a series key string. /// `"metric{a=\"b\",c=\"d\"}"` → `{("a", "b"), ("c", "d")}` +/// +/// The returned `&str` value is the **raw, still-escaped** slice +/// between the opening and closing quote — e.g. for `k="a\"b"` the +/// value is the four bytes `a\"b`, not the decoded `a"b`. Call +/// [`decode_label_value`] if you need the decoded form. Most live +/// callers compare against literal config values that never contain +/// escapable characters (`"`, `\`, `\n`), so the un-decoded slice +/// suffices and saves an allocation per label per sample. +/// +/// The closing-quote scan walks past `\\`, `\"`, `\n` escape pairs +/// emitted by [`format_series_key`] / `render_series_key`, so a +/// value containing embedded `"` no longer terminates parsing +/// prematurely (pre-fix bug — see PR following #284). pub fn parse_labels_from_series_key(series_key: &str) -> HashMap<&str, &str> { let mut labels = HashMap::new(); @@ -945,7 +958,7 @@ pub fn parse_labels_from_series_key(series_key: &str) -> HashMap<&str, &str> { let label_str = &series_key[start..end]; - // Parse comma-separated key="value" pairs + // Parse comma-separated key="value" pairs. let mut remaining = label_str; while !remaining.is_empty() { let eq_pos = match remaining.find('=') { @@ -958,10 +971,30 @@ pub fn parse_labels_from_series_key(series_key: &str) -> HashMap<&str, &str> { break; } + // Walk after the opening quote looking for the closing quote, + // skipping over `\` escape pairs so that values containing + // embedded `"` (escaped as `\"`) don't terminate early. ASCII- + // byte scan; safe because `\` and `"` are single-byte UTF-8 + // and never appear as continuation bytes inside a multi-byte + // scalar — so byte indexing into a `&str` always lands on a + // char boundary at the chosen positions. let value_start = 1; // skip opening quote - let value_end = match after_eq[value_start..].find('"') { - Some(pos) => value_start + pos, - None => break}; + let bytes = after_eq.as_bytes(); + let mut i = value_start; + let value_end = loop { + if i >= bytes.len() { + // No closing quote — malformed input, abandon parse. + return labels; + } + match bytes[i] { + b'\\' if i + 1 < bytes.len() => { + // Skip the escape body byte (\", \\, \n, …). + i += 2; + } + b'"' => break i, + _ => i += 1, + } + }; let value = &after_eq[value_start..value_end]; labels.insert(key, value); @@ -976,6 +1009,49 @@ pub fn parse_labels_from_series_key(series_key: &str) -> HashMap<&str, &str> { labels } +/// Decode a `parse_labels_from_series_key` value slice into its +/// original textual form by undoing the `\"`, `\\`, `\n` escapes +/// emitted by `format_series_key` / `render_series_key`. +/// +/// Returns a borrowed `Cow` when the slice has no `\` byte (the +/// common case — most label values are alphanumeric / dotted / +/// dashed), avoiding allocation. Only allocates when an escape is +/// present. +pub fn decode_label_value(s: &str) -> std::borrow::Cow<'_, str> { + if !s.contains('\\') { + return std::borrow::Cow::Borrowed(s); + } + let mut out = String::with_capacity(s.len()); + // Walk by `char` boundaries so multi-byte UTF-8 scalars round- + // trip intact. Escape recognition operates on ASCII metas (`\`, + // `"`, `n`) which are always single-byte chars in UTF-8. + let mut it = s.chars().peekable(); + while let Some(c) = it.next() { + if c == '\\' { + match it.next() { + Some('"') => out.push('"'), + Some('\\') => out.push('\\'), + Some('n') => out.push('\n'), + Some(other) => { + // Unknown escape — pass the backslash + body + // through verbatim so we don't silently drop data. + out.push('\\'); + out.push(other); + } + None => { + // Trailing backslash with no escape body — keep + // it so round-tripping is lossless even for + // malformed input. + out.push('\\'); + } + } + } else { + out.push(c); + } + } + std::borrow::Cow::Owned(out) +} + /// Route a single sample to `updater`, dispatching keyed vs. non-keyed based on config. /// /// For keyed accumulators (MultipleSum, CMS, HydraKLL), the key is extracted @@ -1126,6 +1202,42 @@ mod tests { assert!(labels.is_empty()); } + #[test] + fn test_parse_labels_skips_escaped_closing_quote() { + // The closing-quote scan must walk past `\"` rather than + // terminating the value early. Regression for the + // `format_series_key` ↔ `parse_labels_from_series_key` + // roundtrip bug — see PR #284's discovery and the + // `series_key_roundtrip_tests` module in + // `drivers/ingest/otel.rs`. + let labels = parse_labels_from_series_key(r#"metric{msg="a\"b",svc="x"}"#); + // Raw (un-decoded) values are returned; `decode_label_value` + // un-escapes them. + assert_eq!(labels.get("msg"), Some(&r#"a\"b"#)); + assert_eq!(labels.get("svc"), Some(&"x")); + } + + #[test] + fn test_decode_label_value_unescapes_known_pairs() { + assert_eq!(decode_label_value("plain"), "plain"); + assert_eq!(decode_label_value(r#"a\"b"#), r#"a"b"#); + assert_eq!(decode_label_value(r"a\\b"), r"a\b"); + assert_eq!(decode_label_value(r"line1\nline2"), "line1\nline2"); + // Unknown escapes pass through unchanged so we don't silently + // drop producer-side data. + assert_eq!(decode_label_value(r"a\xb"), r"a\xb"); + } + + #[test] + fn test_decode_label_value_borrows_when_no_escapes() { + // Borrowed for the common case — no allocation. + let s = "no_escapes_here"; + match decode_label_value(s) { + std::borrow::Cow::Borrowed(b) => assert_eq!(b, s), + std::borrow::Cow::Owned(_) => panic!("expected borrowed, no `\\` in input"), + } + } + // ----------------------------------------------------------------------- // Helpers // -----------------------------------------------------------------------