From ad83efdfc8b3403eacbc5e474e8a1e652da2ffc0 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 12 May 2026 18:46:55 -0600 Subject: [PATCH] =?UTF-8?q?feat(sketch=5Fdb):=20Phase=205=20M2=20=E2=80=94?= =?UTF-8?q?=20derive=20sketch=20sid=20from=20content=20hash?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backend now computes `series_id` for a sketch DataPoint as xxh64(metric_name, attrs_fingerprint, sketch_kind, sketch_config) instead of minting a monotonic counter value through SeriesIdResolver. Same four inputs → same sid across restarts and across hosts, so the controller no longer needs to emit `aggregationId` for each (metric, agg-type, params) tuple — backend derives it. `SketchIndex.instance(sid).is_some()` replaces `SeriesIdResolver::is_known(sid)` as the "have we seen this sid before?" check for the (sid!=0, no-attrs) wire case; the (sid!=0, attrs) case recomputes the hash and signals stale on disagreement. Raw-sample ResolveSeriesIDs RPC and non-sketch paths keep the existing counter-based resolver (no agent-side coordination change needed yet). Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/drivers/ingest/otel.rs | 101 +++++++------- data_plane/src/stores/sketch_db/index/mod.rs | 137 +++++++++++++++++++ 2 files changed, 192 insertions(+), 46 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 2851695b..48d360dd 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -817,29 +817,39 @@ async fn route_modified_otlp_sketches_to_precompute( let series_key = format_series_key(&metric.name, &dp.attrs); let ts_ms = (dp.time_unix_nano / 1_000_000) as i64; - // Phase 4 — sid resolution gate. The four cases mirror - // the design doc §5.4 invariant: - // (sid=0, attrs) → mint a fresh sid and use it - // (sid!=0, attrs) → trust attrs; if cached value - // disagrees, the sender's sid - // is stale → push to + // Phase 5 M2 — sid is content-addressed: a 64-bit + // xxhash of `(metric_name, attrs, sketch_kind, + // sketch_config)`. Same inputs → same sid across + // restarts and across hosts, so the controller no + // longer needs to emit `aggregationId`; the backend + // derives it. Four wire cases: + // (sid=0, attrs) → compute hash, use it + // (sid!=0, attrs) → compute hash; if it + // disagrees with the sender's + // sid the sender's sid is + // stale → push to // `unknown_sids` so the // response evicts it - // (sid!=0, no attrs) → reverse-lookup; if unknown, - // push to `unknown_sids` and - // drop this DP (sender will - // re-emit with attrs next pass) + // (sid!=0, no attrs) → can't recompute the hash; + // fall back to "is this sid + // registered?" via + // SketchIndex. Unknown → push + // to `unknown_sids` and drop + // this DP (sender will + // re-emit with attrs next + // pass) // (sid=0, no attrs) → invalid wire shape, drop let attrs_pairs: Vec<(&str, &str)> = dp.attrs.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&attrs_pairs); + let kind_for_sid = sketch_kind_handle_for(&dp); let resolved_sid: Option = if attrs_pairs.is_empty() { - // No attrs on the wire — sid alone must be - // recognized, otherwise signal stale. + // No attrs on the wire — can't recompute hash. + // Accept the sid iff we've registered it before. match dp.series_id { 0 => None, sid => { - if ingest_state.series_resolver.is_known(sid) { + if ingest_state.sketch_index.instance(sid).is_some() { Some(sid) } else { unknown_sids.push(sid); @@ -847,24 +857,19 @@ async fn route_modified_otlp_sketches_to_precompute( } } } - } else if dp.series_id == 0 { - // Attrs present, no sid yet → mint or fetch. - Some( - ingest_state - .series_resolver - .resolve(&metric.name, &fp), - ) } else { - // Both populated: attrs are the source of truth. - // Backend-resolved value wins; mismatched sender - // sid is signalled stale. - let cached = ingest_state - .series_resolver - .resolve(&metric.name, &fp); - if cached != dp.series_id { + let computed = crate::stores::sketch_db::index::compute_sketch_sid( + &metric.name, + &fp, + kind_for_sid, + &dp.container_config, + ); + if dp.series_id != 0 && dp.series_id != computed { + // Sender's cached sid disagrees with what the + // backend would now compute — signal stale. unknown_sids.push(dp.series_id); } - Some(cached) + Some(computed) }; let Some(sid) = resolved_sid else { continue; @@ -1975,7 +1980,8 @@ mod sid_resolution_tests { let unknown = route_modified_otlp_sketches_to_precompute(&req, &state).await; assert!(unknown.is_empty(), "no unknown sids on a fresh-attrs DP"); - assert_eq!(state.series_resolver.len(), 1, "resolver minted one sid"); + // M2 — sketch sid is hash-derived; resolver is not consulted on + // the sketch ingest path. SketchIndex is the registration set. assert_eq!( state.sketch_index.instance_count(), 1, @@ -1989,8 +1995,9 @@ mod sid_resolution_tests { #[tokio::test] async fn unknown_sid_with_empty_attrs_is_returned_in_response() { let (state, drain) = make_state().await; - // sid != 0, no attrs — resolver doesn't know it; should land in - // unknown_sids and the DP must be dropped (no instance registered). + // sid != 0, no attrs — SketchIndex doesn't know it; should land + // in unknown_sids and the DP must be dropped (no instance + // registered). Hash recomputation isn't possible without attrs. let dp = DdSketchDataPoint { attributes: Vec::new(), start_time_unix_nano: 0, @@ -2005,7 +2012,6 @@ mod sid_resolution_tests { let unknown = route_modified_otlp_sketches_to_precompute(&req, &state).await; assert_eq!(unknown, vec![7777]); - assert_eq!(state.series_resolver.len(), 0); assert_eq!(state.sketch_index.instance_count(), 0); drop(state); @@ -2014,8 +2020,11 @@ mod sid_resolution_tests { #[tokio::test] async fn sid_attrs_disagreement_signals_stale_sid_but_uses_resolved_value() { + use crate::stores::sketch_db::index::{compute_sketch_sid, SketchConfig, SketchKindHandle}; + let (state, drain) = make_state().await; - // First, mint the resolver's view by sending sid=0 with attrs. + // First, register the sid by sending sid=0 with attrs. Sketch + // sid is now hash-derived, not resolver-minted. let dp_seed = DdSketchDataPoint { attributes: vec![kv("zone", "z0")], start_time_unix_nano: 1_000_000, @@ -2031,14 +2040,21 @@ mod sid_resolution_tests { &state, ) .await; - let resolved_sid = state.series_resolver.lookup( + let expected_sid = compute_sketch_sid( "http_latency_ms", &crate::drivers::ingest::canonical_attrs_fingerprint(&[("zone", "z0")]), + SketchKindHandle::DDSketch, + &SketchConfig::DDSketch { + relative_accuracy: 0.01, + }, + ); + assert!( + state.sketch_index.instance(expected_sid).is_some(), + "seed registers the hash-derived sid" ); - let resolved_sid = resolved_sid.expect("seed mints"); // Now arrive with the same attrs but a STALE sid. - let stale = resolved_sid.wrapping_add(123); + let stale = expected_sid.wrapping_add(123); let dp_disagree = DdSketchDataPoint { attributes: vec![kv("zone", "z0")], start_time_unix_nano: 1_000_000, @@ -2055,16 +2071,9 @@ mod sid_resolution_tests { ) .await; assert_eq!(unknown, vec![stale], "stale sid should be signalled"); - // Cache stays at the originally-resolved value — the second call - // returns the same sid via the canonical fingerprint. - let still_resolved = state - .series_resolver - .lookup( - "http_latency_ms", - &crate::drivers::ingest::canonical_attrs_fingerprint(&[("zone", "z0")]), - ) - .expect("still cached"); - assert_eq!(still_resolved, resolved_sid); + // The expected sid stays registered — the second DP routed to + // it via hash recomputation. + assert!(state.sketch_index.instance(expected_sid).is_some()); drop(state); let _ = drain.await; diff --git a/data_plane/src/stores/sketch_db/index/mod.rs b/data_plane/src/stores/sketch_db/index/mod.rs index dbed1fdb..5be8ca79 100644 --- a/data_plane/src/stores/sketch_db/index/mod.rs +++ b/data_plane/src/stores/sketch_db/index/mod.rs @@ -24,6 +24,7 @@ use std::sync::{Arc, RwLock}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use dashmap::DashMap; +use xxhash_rust::xxh64::xxh64; use self::epoch_columnar::{LabelValuesId, SidStoreData, TimestampRange}; use crate::stores::sketch_db::schema::AggStatus; @@ -59,6 +60,87 @@ pub enum SketchConfig { CountMin { rows: i32, cols: i32 }, } +/// Compute a deterministic `series_id` (sid) for one sketch instance. +/// +/// Folds the four DataPoint inputs the backend has at ingest time — +/// `metric_name`, `attrs` (keys + values, canonicalized), `sketch_kind`, +/// and `sketch_config` — into a 64-bit xxhash. Same inputs always +/// produce the same sid across restarts and across hosts, so the +/// controller no longer needs to mint and emit an `aggregation_id` for +/// each (metric, agg-type, params) tuple. Phase 5 M2 directive. +/// +/// `attrs_fingerprint` MUST be the canonical fingerprint string +/// (`canonical_attrs_fingerprint` — keys sorted, joined `k=v;`); the +/// hash is sensitive to whitespace, ordering, and trailing separator, +/// so callers must round-trip through that one function for the +/// pre-flight ResolveSeriesIDs RPC and the ingest path to agree. +/// +/// sid=0 is reserved on the wire (means "unresolved"); if a real input +/// hashes to 0 (vanishingly unlikely with 64-bit xxhash), we perturb to +/// 1. +pub fn compute_sketch_sid( + metric_name: &str, + attrs_fingerprint: &str, + sketch_kind: SketchKindHandle, + sketch_config: &SketchConfig, +) -> u64 { + let mut buf: Vec = + Vec::with_capacity(metric_name.len() + attrs_fingerprint.len() + 24); + buf.extend_from_slice(metric_name.as_bytes()); + buf.push(0); + buf.extend_from_slice(attrs_fingerprint.as_bytes()); + buf.push(0); + buf.push(sketch_kind_tag(sketch_kind)); + buf.push(0); + encode_sketch_config(sketch_config, &mut buf); + let h = xxh64(&buf, 0); + if h == 0 { + 1 + } else { + h + } +} + +fn sketch_kind_tag(k: SketchKindHandle) -> u8 { + match k { + SketchKindHandle::DDSketch => 1, + SketchKindHandle::Kll => 2, + SketchKindHandle::Hll => 3, + SketchKindHandle::CountSketch => 4, + SketchKindHandle::CountMin => 5, + SketchKindHandle::CmsWithHeap => 6, + SketchKindHandle::CountSketchWithHeap => 7, + SketchKindHandle::Any => 0, + } +} + +fn encode_sketch_config(cfg: &SketchConfig, buf: &mut Vec) { + match cfg { + SketchConfig::DDSketch { relative_accuracy } => { + buf.push(b'D'); + buf.extend_from_slice(&relative_accuracy.to_le_bytes()); + } + SketchConfig::Kll { k } => { + buf.push(b'K'); + buf.extend_from_slice(&k.to_le_bytes()); + } + SketchConfig::Hll { precision } => { + buf.push(b'H'); + buf.extend_from_slice(&precision.to_le_bytes()); + } + SketchConfig::CountSketch { rows, cols } => { + buf.push(b'S'); + buf.extend_from_slice(&rows.to_le_bytes()); + buf.extend_from_slice(&cols.to_le_bytes()); + } + SketchConfig::CountMin { rows, cols } => { + buf.push(b'M'); + buf.extend_from_slice(&rows.to_le_bytes()); + buf.extend_from_slice(&cols.to_le_bytes()); + } + } +} + /// Accuracy bound derived from `SketchConfig`. Surfaced to the user via /// query response metadata so they know the precision / confidence of /// each result. @@ -697,6 +779,61 @@ mod tests { assert_eq!(series.len(), 1); assert_eq!(series[0].samples.len(), 4); } + + #[test] + fn compute_sketch_sid_is_deterministic() { + let cfg = SketchConfig::DDSketch { + relative_accuracy: 0.01, + }; + let a = compute_sketch_sid("http_requests_total", "zone=z0;", SketchKindHandle::DDSketch, &cfg); + let b = compute_sketch_sid("http_requests_total", "zone=z0;", SketchKindHandle::DDSketch, &cfg); + assert_eq!(a, b); + assert_ne!(a, 0); + } + + #[test] + fn compute_sketch_sid_distinguishes_metric() { + let cfg = SketchConfig::DDSketch { + relative_accuracy: 0.01, + }; + let a = compute_sketch_sid("metric_a", "zone=z0;", SketchKindHandle::DDSketch, &cfg); + let b = compute_sketch_sid("metric_b", "zone=z0;", SketchKindHandle::DDSketch, &cfg); + assert_ne!(a, b); + } + + #[test] + fn compute_sketch_sid_distinguishes_attrs_values() { + let cfg = SketchConfig::DDSketch { + relative_accuracy: 0.01, + }; + let a = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg); + let b = compute_sketch_sid("m", "zone=z1;", SketchKindHandle::DDSketch, &cfg); + assert_ne!(a, b); + } + + #[test] + fn compute_sketch_sid_distinguishes_sketch_kind() { + let cfg_dd = SketchConfig::DDSketch { + relative_accuracy: 0.01, + }; + let cfg_kll = SketchConfig::Kll { k: 200 }; + let a = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg_dd); + let b = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::Kll, &cfg_kll); + assert_ne!(a, b); + } + + #[test] + fn compute_sketch_sid_distinguishes_container_config() { + let cfg_a = SketchConfig::DDSketch { + relative_accuracy: 0.01, + }; + let cfg_b = SketchConfig::DDSketch { + relative_accuracy: 0.005, + }; + let a = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg_a); + let b = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg_b); + assert_ne!(a, b); + } } // 2026-05 reorg: generic epoch-partitioned columnar storage lives