From 84abac7b585b133e91467284049861d8e434358d Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 13 May 2026 10:16:37 -0600 Subject: [PATCH 1/3] feat(ingest): make SeriesIdResolver authoritative; populate series_assignments (PR-1/3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Option B from the sid-uniqueness discussion: collapse the duality between SeriesIdResolver (sequential) and compute_sketch_sid (xxh64 content-addressed) by making the resolver the single mint on the ingest path. xxh64 only gives probabilistic uniqueness; the wire shape (agent omits attrs after caching sid → backend disambiguates by sid) needs uniqueness as a contract, not a property. Changes: - `route_modified_otlp_sketches_to_precompute` returns `IngestOutcome { unknown_series_ids, series_assignments }` instead of bare `Vec`. Both halves of the Phase-4 round trip surface together. - Attrs-bearing DPs resolve via `series_resolver.resolve(metric, fp)` instead of `compute_sketch_sid`. Sender's sid disagreeing with the resolver's binding is signalled via unknown_series_ids; the canonical assignment is always echoed back so the sender refreshes its cache. - gRPC `Export` response now populates `series_assignments`; HTTP response surfaces the count for observability (OTLP/HTTP spec keeps the response shape minimal). - 3 existing tests updated for the new return type + Option B semantics; 1 new round-trip test (`second_emit_with_cached_sid_and_no_attrs_hits_same_instance`) exercises the cache-hit bandwidth-saving path. Trade-off taken on: sids are no longer stable across independent backends or backend restarts. PR-2 of this chain adds WAL-backed resolver persistence; without it, restart-recovery still works via the existing `unknown_series_ids` eviction primitive (agents observe stale sid → evict → re-emit with attrs → fresh assignment). `compute_sketch_sid` is now dead at the ingest layer but stays in the codebase for now — PR-3 deletes it after PR-2 lands. cargo build -p data_plane --lib: clean cargo test -p data_plane --lib drivers::ingest: 14/14 passing Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/drivers/ingest/otel.rs | 308 ++++++++++++++++++++------ 1 file changed, 235 insertions(+), 73 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 24e9e0523..84a6f2b47 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -185,25 +185,27 @@ impl MetricsService for MetricsServiceImpl { if let Some(cache) = &self.shared.probe_cache { capture_freshness_probe_samples(&req, cache); } - let mut unknown_series_ids: Vec = Vec::new(); + let mut outcome = IngestOutcome::default(); if let Some(state) = &self.shared.ingest_state { route_otlp_to_precompute(&req, state).await; - unknown_series_ids = - route_modified_otlp_sketches_to_precompute(&req, state).await; + outcome = route_modified_otlp_sketches_to_precompute(&req, state).await; } debug!("OTLP sending response via gRPC"); Ok(Response::new(ExportMetricsServiceResponse { partial_success: None, - // Modified-OTLP collector hands out stable series descriptors via - // this field; not yet wired (PR B will populate it when the - // backend learns to mint series_ids). - series_assignments: Vec::new(), + // Sid bindings the sender should cache. Each entry maps an + // `attributes_fingerprint` to the canonical sid the backend's + // `SeriesIdResolver` minted (or returned from its cache). The + // sender's local dictionary keys on `fingerprint`, so it can + // refresh stale entries and pick up brand-new ones from this + // field without a separate `ResolveSeriesIDs` round-trip. + series_assignments: outcome.series_assignments, // Phase 4 — backend signals senders to evict cached sids here // when this Export carried a sid the resolver does not // recognize (sid-cache divergence — e.g. after a backend // restart without persistence, or when the sender's sid // disagrees with the resolved sid for the same attrs). - unknown_series_ids, + unknown_series_ids: outcome.unknown_series_ids, })) } @@ -277,15 +279,23 @@ async fn handle_otlp_http( if let Some(cache) = &shared.probe_cache { capture_freshness_probe_samples(&req, cache); } - let mut unknown_series_ids: Vec = Vec::new(); + let mut outcome = IngestOutcome::default(); if let Some(state) = &shared.ingest_state { route_otlp_to_precompute(&req, state).await; - unknown_series_ids = route_modified_otlp_sketches_to_precompute(&req, state).await; + outcome = route_modified_otlp_sketches_to_precompute(&req, state).await; } debug!("OTLP sending response via HTTP"); + // HTTP OTLP exporters don't generally read `series_assignments` + // back the way the gRPC path does (the OTLP/HTTP spec keeps the + // response shape minimal), so we surface assignments only as a + // count for observability. Senders that need the bindings should + // use the gRPC transport. Eviction signals stay first-class — they + // are the universal recovery primitive (see proto comment on + // `ExportMetricsServiceResponse.unknown_series_ids`). Ok(Json(serde_json::json!({ "rejected": 0, - "unknown_series_ids": unknown_series_ids, + "unknown_series_ids": outcome.unknown_series_ids, + "series_assignments_count": outcome.series_assignments.len(), }))) } @@ -672,10 +682,29 @@ async fn route_otlp_to_precompute( /// today (KLL / DDSketch / CountSketch / HLL) fall through to the /// §5.2 fallback path so the user still gets a correct answer; PR C /// (task #8) will close those decoder gaps. +/// Per-Export outcome of the modified-OTLP sketch ingest path. Carries +/// both halves of the Phase-4/B round trip: +/// +/// - `unknown_series_ids` → sids the receiver could not satisfy this +/// Export (sender's cache is stale; sender must evict + re-emit with +/// attrs). +/// - `series_assignments` → newly-minted or cache-resolved sid bindings +/// the receiver wants the sender to cache. Sent back in +/// `ExportMetricsServiceResponse.series_assignments` so the sender +/// omits attrs on subsequent emits keyed by these sids. +/// +/// Empty `series_assignments` is the no-op case (every DP arrived with +/// the right sid already, or every DP was rejected to `unknown_series_ids`). +#[derive(Debug, Default)] +pub(crate) struct IngestOutcome { + pub unknown_series_ids: Vec, + pub series_assignments: Vec, +} + async fn route_modified_otlp_sketches_to_precompute( request: &ExportMetricsServiceRequest, ingest_state: &Arc, -) -> Vec { +) -> IngestOutcome { use asap_otel_proto::tonic::metrics::v1::metric::Data; let ingest_received_at = Instant::now(); @@ -702,6 +731,14 @@ async fn route_modified_otlp_sketches_to_precompute( // these sids and re-emit with attributes; backend re-resolves and // returns fresh `series_assignments`. let mut unknown_sids: Vec = Vec::new(); + // Sid bindings the receiver wants the sender to cache. Populated + // every time an attrs-bearing DP is resolved by the + // `SeriesIdResolver` (either fresh mint or cache hit). Returned + // alongside `unknown_sids` so the gRPC / HTTP handler can stamp + // them into `ExportMetricsServiceResponse.series_assignments`. + let mut new_assignments: Vec< + asap_otel_proto::tonic::collector::metrics::v1::SeriesAssignment, + > = Vec::new(); for resource_metrics in &request.resource_metrics { let resource_attrs = resource_metrics @@ -829,23 +866,49 @@ 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 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) → can't recompute the hash; - // fall back to "is this sid - // registered?" via - // SketchStore. Unknown → push + // Sid resolution — registry-allocated, NOT content- + // addressed. The `SeriesIdResolver` is the single + // authoritative mint for sids in the pipeline: same + // `(metric_name, attrs_fingerprint)` always returns + // the same sid for the lifetime of the resolver's + // cache. Uniqueness is by construction + // (`AtomicU64::fetch_add`); two different identities + // CANNOT share a sid. The content-addressed + // `compute_sketch_sid` path was retired here because + // u64 xxhash gives only probabilistic uniqueness, + // and "unique sid per series" is a contract this + // wire shape needs (the agent omits attrs on + // subsequent emits; the receiver must be able to + // disambiguate `sid → (metric, attrs)`). + // + // Determinism trade-off: sids are NOT stable across + // independent backends, and (without persistence) + // not across backend restarts either. PR-2 of this + // chain wires a WAL-backed `SeriesResolverPersistence` + // trait so the mapping survives restart; without it, + // restart-recovery still works via the + // `unknown_series_ids` eviction primitive (agents + // observe their cached sid is unknown, evict, re-emit + // with attrs, get a fresh assignment). + // + // Four wire cases: + // (sid=0, attrs) → resolve (mint or cache + // hit); always emit a + // `SeriesAssignment` so the + // sender caches the binding + // (sid!=0, attrs) → resolve; if the resolver's + // sid disagrees with the + // sender's, push sender's + // sid to `unknown_sids` + // (sender's cache was stale, + // e.g. after a backend + // restart without + // persistence); always emit a + // fresh assignment + // (sid!=0, no attrs) → can't resolve without + // attrs; accept the sid iff + // the SketchStore has it + // registered. Unknown → push // to `unknown_sids` and drop // this DP (sender will // re-emit with attrs next @@ -854,10 +917,12 @@ async fn route_modified_otlp_sketches_to_precompute( 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 — can't recompute hash. - // Accept the sid iff we've registered it before. + // No attrs on the wire — can't consult the + // resolver (it keys on `(metric, fp)`). Accept + // the sid iff the SketchStore has registered it + // (i.e. a prior Export with this same sid + attrs + // already landed and registered metadata). match dp.series_id { 0 => None, sid => { @@ -870,18 +935,31 @@ async fn route_modified_otlp_sketches_to_precompute( } } } else { - let computed = crate::storage_engines::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. + let assigned = + ingest_state.series_resolver.resolve(&metric.name, &fp); + if dp.series_id != 0 && dp.series_id != assigned { + // Sender's cached sid disagrees with the + // resolver's binding — sender's cache is + // stale, signal eviction. unknown_sids.push(dp.series_id); } - Some(computed) + // Always echo the canonical binding back in + // `series_assignments` so the sender caches it + // (or refreshes a stale entry). The dictionary + // bookkeeping fields beyond + // `(attributes_fingerprint, series_id)` are + // optional today — the patched OTel-Go exporter + // keys its local cache on the fingerprint, not on + // the dictionary metadata. + new_assignments.push( + asap_otel_proto::tonic::collector::metrics::v1::SeriesAssignment { + attributes_fingerprint: fp.as_bytes().to_vec(), + series_id: assigned, + metric_name: metric.name.clone(), + ..Default::default() + }, + ); + Some(assigned) }; let Some(sid) = resolved_sid else { continue; @@ -1112,7 +1190,10 @@ async fn route_modified_otlp_sketches_to_precompute( ); } - unknown_sids + IngestOutcome { + unknown_series_ids: unknown_sids, + series_assignments: new_assignments, + } } /// Phase 5 helper — map a `ModifiedOtlpSketchDp` to the matching @@ -1984,15 +2065,30 @@ mod sid_resolution_tests { }; let req = build_request("http_latency_ms", dp); - let unknown = route_modified_otlp_sketches_to_precompute(&req, &state).await; - assert!(unknown.is_empty(), "no unknown sids on a fresh-attrs DP"); - // M2 — sketch sid is hash-derived; resolver is not consulted on - // the sketch ingest path. SketchStore is the registration set. + let outcome = route_modified_otlp_sketches_to_precompute(&req, &state).await; + assert!( + outcome.unknown_series_ids.is_empty(), + "no unknown sids on a fresh-attrs DP" + ); + // Option B — sid is resolver-allocated; every attrs-bearing DP + // gets a SeriesAssignment echoed back so the sender caches it. + assert_eq!( + outcome.series_assignments.len(), + 1, + "one series_assignment returned for one fresh DP" + ); + let assigned = &outcome.series_assignments[0]; + assert_eq!(assigned.metric_name, "http_latency_ms"); + assert_ne!( + assigned.series_id, 0, + "resolver mints a non-zero sid (zero is reserved on the wire)" + ); assert_eq!( state.sketch_index.instance_count(), 1, - "SketchStore registered one instance" + "SketchStore registered one instance under the resolver-minted sid" ); + assert!(state.sketch_index.instance(assigned.series_id).is_some()); drop(state); let _ = drain.await; @@ -2003,7 +2099,7 @@ mod sid_resolution_tests { let (state, drain) = make_state().await; // sid != 0, no attrs — SketchStore 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. + // registered). The resolver can't be consulted without attrs. let dp = DdSketchDataPoint { attributes: Vec::new(), start_time_unix_nano: 0, @@ -2016,8 +2112,12 @@ mod sid_resolution_tests { }; let req = build_request("http_latency_ms", dp); - let unknown = route_modified_otlp_sketches_to_precompute(&req, &state).await; - assert_eq!(unknown, vec![7777]); + let outcome = route_modified_otlp_sketches_to_precompute(&req, &state).await; + assert_eq!(outcome.unknown_series_ids, vec![7777]); + assert!( + outcome.series_assignments.is_empty(), + "no assignment when attrs are missing" + ); assert_eq!(state.sketch_index.instance_count(), 0); drop(state); @@ -2026,11 +2126,9 @@ mod sid_resolution_tests { #[tokio::test] async fn sid_attrs_disagreement_signals_stale_sid_but_uses_resolved_value() { - use crate::storage_engines::sketch_db::index::{compute_sketch_sid, SketchConfig, SketchKindHandle}; - let (state, drain) = make_state().await; - // First, register the sid by sending sid=0 with attrs. Sketch - // sid is now hash-derived, not resolver-minted. + // First, register the sid by sending sid=0 with attrs. The + // resolver mints a fresh u64 (sequential, NOT content-addressed). let dp_seed = DdSketchDataPoint { attributes: vec![kv("zone", "z0")], start_time_unix_nano: 1_000_000, @@ -2041,26 +2139,21 @@ mod sid_resolution_tests { flags: 0, series_id: 0, }; - let _ = route_modified_otlp_sketches_to_precompute( + let seed_outcome = route_modified_otlp_sketches_to_precompute( &build_request("http_latency_ms", dp_seed), &state, ) .await; - 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, - }, - ); + let assigned_sid = seed_outcome.series_assignments[0].series_id; assert!( - state.sketch_index.instance(expected_sid).is_some(), - "seed registers the hash-derived sid" + state.sketch_index.instance(assigned_sid).is_some(), + "seed registers the resolver-minted sid" ); - // Now arrive with the same attrs but a STALE sid. - let stale = expected_sid.wrapping_add(123); + // Now arrive with the same attrs but a STALE sid (sender's + // local cache was wrong, e.g. survived a backend restart without + // persistence). + let stale = assigned_sid.wrapping_add(123); let dp_disagree = DdSketchDataPoint { attributes: vec![kv("zone", "z0")], start_time_unix_nano: 1_000_000, @@ -2071,15 +2164,84 @@ mod sid_resolution_tests { flags: 0, series_id: stale, }; - let unknown = route_modified_otlp_sketches_to_precompute( + let outcome = route_modified_otlp_sketches_to_precompute( &build_request("http_latency_ms", dp_disagree), &state, ) .await; - assert_eq!(unknown, vec![stale], "stale sid should be signalled"); - // The expected sid stays registered — the second DP routed to - // it via hash recomputation. - assert!(state.sketch_index.instance(expected_sid).is_some()); + assert_eq!( + outcome.unknown_series_ids, + vec![stale], + "stale sid should be signalled for eviction" + ); + assert_eq!( + outcome.series_assignments.len(), + 1, + "fresh assignment echoed so sender can refresh its cache" + ); + assert_eq!( + outcome.series_assignments[0].series_id, assigned_sid, + "resolver returns the canonical sid for this (metric, attrs) — same as seed" + ); + // The assigned sid stays registered — the second DP routed to + // it via the resolver's cache hit. + assert!(state.sketch_index.instance(assigned_sid).is_some()); + + drop(state); + let _ = drain.await; + } + + #[tokio::test] + async fn second_emit_with_cached_sid_and_no_attrs_hits_same_instance() { + // Round-trip: emit DP with attrs → cache the assignment → + // re-emit with (sid, no attrs). Second emit must NOT push + // anything to unknown_sids and MUST land in the same SketchStore + // instance. This is the bandwidth-saving path the registry + // approach is built for. + let (state, drain) = make_state().await; + + let dp_first = DdSketchDataPoint { + attributes: vec![kv("zone", "z0")], + start_time_unix_nano: 1_000_000, + time_unix_nano: 11_000_000, + sketch: vec![1], + encoding: 1, + exemplars: Vec::new(), + flags: 0, + series_id: 0, + }; + let first = route_modified_otlp_sketches_to_precompute( + &build_request("http_latency_ms", dp_first), + &state, + ) + .await; + let cached_sid = first.series_assignments[0].series_id; + + let dp_second = DdSketchDataPoint { + attributes: Vec::new(), // sender omits attrs now that it has the sid + start_time_unix_nano: 1_000_000, + time_unix_nano: 12_000_000, + sketch: vec![2], + encoding: 1, + exemplars: Vec::new(), + flags: 0, + series_id: cached_sid, + }; + let second = route_modified_otlp_sketches_to_precompute( + &build_request("http_latency_ms", dp_second), + &state, + ) + .await; + assert!( + second.unknown_series_ids.is_empty(), + "cached sid + no attrs hits the same SketchStore instance" + ); + assert!( + second.series_assignments.is_empty(), + "no fresh assignment when sender already had a valid binding" + ); + // Both DPs landed against the same sid — no proliferation. + assert_eq!(state.sketch_index.instance_count(), 1); drop(state); let _ = drain.await; From 3ac90cf4b9ea98f059bfe7d2abf31e451fe94815 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 13 May 2026 10:39:16 -0600 Subject: [PATCH 2/3] feat(ingest): WAL-backed persistence for SeriesIdResolver (PR-2/3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolver bindings now survive backend restart under --persistence-enabled. PR-1 made the resolver the authoritative mint; this PR makes those mints durable. WAL format v1 (append-only, single-writer): header: 8 bytes → b"ASAPSRP\x01" record: u64 sid LE + u32 metric_len LE + metric utf8 + u32 fp_len LE + fp utf8 Each `resolve()` mint calls `persistence.append(...)` (fsync before returning) so the caller never observes a sid that isn't on stable storage. Persistence errors log at WARN and don't propagate — the resolver stays in-memory-correct; the next restart pays the eviction cost for the lost mint. Crash recovery: a torn write at EOF (short read on any record field, or out-of-range length prefix) is detected at replay; the file is truncated to the last durable record's offset. Bounds: MAX_METRIC_LEN 16KiB, MAX_FP_LEN 64KiB — well above any realistic input and small enough that a corrupted file can't OOM the replay loop. No CRC for now; add one if bit-rot telemetry ever fires. Trait shape lets tests inject a mock (NoopPersistence) or a failing backend; production constructor `SeriesIdResolver::open(path)` wires FilePersistence + replays before returning a warm resolver. `next_sid` resumes at `max(replayed_sid) + 1`. Wiring in main.rs: under --persistence-enabled, the WAL lives at `{persistence_dir}/series_resolver.wal`; otherwise NoopPersistence preserves the current (in-memory-only) behaviour. Tests (8 new under series_resolver::persistence_tests): - empty_log_replays_empty - append_then_replay_round_trips - header_mismatch_errors_on_open - torn_record_truncated_on_replay - out_of_range_metric_len_treated_as_torn - resolver_open_replays_existing_log - resolver_with_noop_persistence_does_not_persist - append_failure_logs_but_does_not_panic cargo build -p data_plane (lib + bin): clean cargo test -p data_plane --lib drivers::ingest: 22/22 passing Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/drivers/ingest/series_resolver.rs | 578 +++++++++++++++++- data_plane/src/main.rs | 20 +- 2 files changed, 583 insertions(+), 15 deletions(-) diff --git a/data_plane/src/drivers/ingest/series_resolver.rs b/data_plane/src/drivers/ingest/series_resolver.rs index c78bdec48..ee9fb49ab 100644 --- a/data_plane/src/drivers/ingest/series_resolver.rs +++ b/data_plane/src/drivers/ingest/series_resolver.rs @@ -25,7 +25,12 @@ //! at `docs/design-controller-into-backend.md`. use dashmap::DashMap; +use std::fs::{File, OpenOptions}; +use std::io::{Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use tracing::{info, warn}; /// Canonical fingerprint key — `(metric_name, attrs_fingerprint)`. /// @@ -43,36 +48,117 @@ type CacheKey = (String, String); pub struct SeriesIdResolver { cache: DashMap, next_sid: AtomicU64, + /// Durability hook. `NoopPersistence` is the default — fast, no I/O, + /// loses every binding on restart (recovery path is the existing + /// `unknown_series_ids` eviction signal). `FilePersistence` writes a + /// WAL record per fresh mint and replays on construction, so the + /// agent's cached sids stay valid across backend restarts. + persistence: Arc, } impl SeriesIdResolver { + /// Build an in-memory-only resolver. Equivalent to + /// `with_persistence(NoopPersistence)`. Suitable for tests, for the + /// `--persistence-enabled=false` deployment mode, and for any path + /// where the caller doesn't need restart-survival semantics. pub fn new() -> Self { + Self::with_persistence(Arc::new(NoopPersistence)) + } + + /// Build a resolver with an injected persistence backend. Used by + /// tests that want to verify the trait contract against a mock, and + /// by [`Self::open`] under the hood. + pub fn with_persistence(persistence: Arc) -> Self { Self { cache: DashMap::new(), // sid=0 is reserved for "unresolved/uncached"; start minting at 1. next_sid: AtomicU64::new(1), + persistence, + } + } + + /// Open a file-backed resolver at `path`, replay every durable + /// binding into the in-memory cache, resume `next_sid` at + /// `max(replayed_sid) + 1`, and return the warm resolver. The + /// production constructor under `--persistence-enabled`. + /// + /// If the file does not exist, it is created with a fresh header + /// and the resolver starts cold (no bindings, `next_sid = 1`). + /// + /// If a torn record is detected at EOF during replay (e.g. backend + /// crashed mid-append), the file is truncated to the last durable + /// record's offset. Replay returns the durable prefix. + pub fn open(path: PathBuf) -> std::io::Result { + let persistence = Arc::new(FilePersistence::open(path)?); + let records = persistence.replay()?; + info!( + replayed = records.len(), + "series-resolver WAL replayed", + ); + let cache: DashMap = DashMap::new(); + let mut max_sid: u64 = 0; + for r in records { + cache.insert((r.metric, r.attrs_fingerprint), r.sid); + if r.sid > max_sid { + max_sid = r.sid; + } } + // sid=0 is reserved; if the log was empty, max_sid is 0 and we + // start minting at 1 (the same as a cold start). + Ok(Self { + cache, + next_sid: AtomicU64::new(max_sid.saturating_add(1).max(1)), + persistence, + }) } /// Resolve `(metric_name, attrs)` to a series_id. Returns the existing /// sid if this `(metric, attrs)` tuple was already registered; - /// otherwise mints a fresh sid, caches it, and returns the new value. + /// otherwise mints a fresh sid, durably persists the binding (when + /// a non-noop backend is wired), caches it, and returns the new + /// value. /// /// Idempotent: repeated calls with the same input ALWAYS return the - /// same sid for the lifetime of the cache. After a backend restart - /// without persistence, the cache is empty — recovered agents emit - /// with attributes, and this method mints fresh sids (potentially - /// different from the pre-restart values). Old sids the agents had - /// cached are signalled as stale via the response's - /// `unknown_series_ids` field; agents evict and re-resolve. + /// same sid for the lifetime of the cache. With `FilePersistence`, + /// "lifetime of the cache" extends across backend restarts (the WAL + /// is replayed on [`Self::open`]). With `NoopPersistence`, the + /// cache resets per process; agents observe their cached sids as + /// stale via `unknown_series_ids` and re-resolve with attrs. + /// + /// Persistence failures are logged at WARN and do NOT propagate — + /// the resolver stays in-memory-correct. Next restart will not + /// recover the lost mint, and the agent will hit the eviction + /// recovery path (one extra round trip with attrs). pub fn resolve(&self, metric_name: &str, attrs_fingerprint: &str) -> u64 { let key = (metric_name.to_string(), attrs_fingerprint.to_string()); - // DashMap::entry().or_insert_with() is atomic — concurrent - // callers for the same key serialize on the bucket lock. - let entry = self - .cache - .entry(key) - .or_insert_with(|| self.next_sid.fetch_add(1, Ordering::Relaxed)); + // Fast path: read-only check on the cache before taking the + // bucket's write lock. DashMap's `get` takes a shard read lock; + // the common case (a hit on a known identity) never serializes + // against other resolve calls. + if let Some(existing) = self.cache.get(&key) { + return *existing; + } + // Slow path: bucket write lock + mint + persist + insert. + // `entry().or_insert_with` ensures only ONE caller runs the + // closure for a given key, even under concurrent load. The + // persistence append happens inside the closure so the binding + // is durable before any caller observes the sid. + let entry = self.cache.entry(key).or_insert_with(|| { + let sid = self.next_sid.fetch_add(1, Ordering::Relaxed); + if let Err(e) = + self.persistence + .append(sid, metric_name, attrs_fingerprint) + { + warn!( + metric = %metric_name, + sid, + error = %e, + "resolver persistence append failed; binding is \ + in-memory-only and will not survive restart", + ); + } + sid + }); *entry } @@ -106,6 +192,295 @@ impl Default for SeriesIdResolver { } } +// ── Persistence ────────────────────────────────────────────────────────────── +// +// The resolver's `(metric, fp) → sid` cache is in-memory only by default. +// Under `--persistence-enabled`, a `FilePersistence` backend writes a WAL +// record per fresh mint; the resolver replays it on startup so the agent's +// cached sids stay valid across backend restarts. +// +// WAL format v1: +// header: 8 bytes → b"ASAPSRP\x01" +// record: 8 bytes → sid (u64 little-endian) +// 4 bytes → metric_len (u32 LE) +// metric_len bytes → metric utf8 +// 4 bytes → fp_len (u32 LE) +// fp_len bytes → fp utf8 +// +// Append-only; sids are minted once and never rewritten, so the log size +// is proportional to live cardinality. At 100M sids (~5GB) compaction +// becomes worth scheduling; not implemented here. +// +// Crash safety: every `append` calls `fsync` before returning. A torn +// write at EOF (kernel buffered the bytes but the metadata flush was +// interrupted) is detected at replay via short-read on any record field +// — the file is truncated to the last durable record's offset and replay +// returns the durable prefix. No CRC: bit-rot is low-probability for an +// append-only WAL; add a CRC field if telemetry ever shows it firing. + +/// One durable binding row read back from the WAL. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ResolverRecord { + pub sid: u64, + pub metric: String, + pub attrs_fingerprint: String, +} + +/// Durability hook for [`SeriesIdResolver`]. Implementations decide +/// whether mints survive process restart. `NoopPersistence` is fine for +/// tests and stateless deployments; `FilePersistence` is the production +/// answer under `--persistence-enabled`. +pub trait SeriesResolverPersistence: Send + Sync { + /// Durably record a fresh `(sid, metric, fp)` binding. MUST be + /// flushed to stable storage before returning `Ok` — the resolver + /// only returns the sid to its caller after this returns. + /// On error, the binding is in-memory-only; the caller logs and + /// continues. + fn append( + &self, + sid: u64, + metric: &str, + attrs_fingerprint: &str, + ) -> std::io::Result<()>; + + /// Read every durable binding in append order. Called once at + /// resolver construction time. + fn replay(&self) -> std::io::Result>; +} + +/// In-memory-only impl. Every `append` is a no-op; `replay` returns +/// empty. Resolver bindings reset on process restart; agents recover via +/// the `unknown_series_ids` eviction primitive. +pub struct NoopPersistence; + +impl SeriesResolverPersistence for NoopPersistence { + fn append(&self, _sid: u64, _metric: &str, _fp: &str) -> std::io::Result<()> { + Ok(()) + } + + fn replay(&self) -> std::io::Result> { + Ok(Vec::new()) + } +} + +/// File-backed WAL. Single-writer; the `Mutex` serializes appends +/// to keep the on-disk order deterministic and so `fsync` ordering +/// matches mint order. Reads only happen at construction. +#[derive(Debug)] +pub struct FilePersistence { + file: Mutex, + path: PathBuf, +} + +const WAL_MAGIC: &[u8; 8] = b"ASAPSRP\x01"; +/// Reject any single field whose length-prefix exceeds these caps. A +/// corrupted file might claim huge field lengths; without these bounds +/// the replay loop could allocate gigabytes of zeros before discovering +/// the lengths don't match the actual content. The caps are far above +/// any realistic input — metric names are tens of bytes, fingerprints +/// are hundreds. +const MAX_METRIC_LEN: usize = 16 * 1024; +const MAX_FP_LEN: usize = 64 * 1024; + +impl FilePersistence { + /// Open or create the WAL at `path`. On a fresh file, writes the + /// magic header and fsyncs. On an existing file, verifies the + /// header matches and seeks to EOF for future appends. + pub fn open(path: PathBuf) -> std::io::Result { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + let mut file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .open(&path)?; + let len = file.metadata()?.len(); + if len == 0 { + file.write_all(WAL_MAGIC)?; + file.sync_all()?; + } else { + let mut hdr = [0u8; 8]; + file.seek(SeekFrom::Start(0))?; + file.read_exact(&mut hdr)?; + if &hdr != WAL_MAGIC { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!( + "resolver WAL header mismatch at {:?}: expected {:?}, got {:?}", + path, WAL_MAGIC, hdr, + ), + )); + } + } + // Position at EOF — appends start here. + file.seek(SeekFrom::End(0))?; + Ok(Self { + file: Mutex::new(file), + path, + }) + } + + /// Diagnostic accessor — the WAL path. Tests use this to inspect + /// the on-disk file. + pub fn path(&self) -> &Path { + &self.path + } +} + +impl SeriesResolverPersistence for FilePersistence { + fn append( + &self, + sid: u64, + metric: &str, + fp: &str, + ) -> std::io::Result<()> { + let metric_bytes = metric.as_bytes(); + let fp_bytes = fp.as_bytes(); + let metric_len: u32 = metric_bytes + .len() + .try_into() + .map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "metric name longer than u32::MAX bytes", + ) + })?; + let fp_len: u32 = fp_bytes.len().try_into().map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "fingerprint longer than u32::MAX bytes", + ) + })?; + + let mut f = self.file.lock().unwrap(); + f.write_all(&sid.to_le_bytes())?; + f.write_all(&metric_len.to_le_bytes())?; + f.write_all(metric_bytes)?; + f.write_all(&fp_len.to_le_bytes())?; + f.write_all(fp_bytes)?; + // Durability barrier: caller must not observe the sid until the + // record is on stable storage. fsync is the slow part of the + // mint path (a few ms on SSD) but it's amortized — minting is + // once per identity, not per emit. + f.sync_all()?; + Ok(()) + } + + fn replay(&self) -> std::io::Result> { + let mut f = self.file.lock().unwrap(); + f.seek(SeekFrom::Start(0))?; + let mut hdr = [0u8; 8]; + match f.read_exact(&mut hdr) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { + // File exists but is empty — caller likely opened it + // moments ago without writing the header yet. Treat as + // no records. + return Ok(Vec::new()); + } + Err(e) => return Err(e), + } + if &hdr != WAL_MAGIC { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "resolver WAL header mismatch during replay", + )); + } + let mut out = Vec::new(); + // After successful header read, the offset is 8. + let mut safe_offset: u64 = 8; + loop { + match read_one_record(&mut *f) { + ReadOne::Ok(record, new_offset) => { + out.push(record); + safe_offset = new_offset; + } + ReadOne::Eof => break, + ReadOne::Torn => { + warn!( + path = %self.path.display(), + torn_at = safe_offset, + recovered = out.len(), + "resolver WAL: torn record at EOF — truncating to last durable offset", + ); + f.set_len(safe_offset)?; + f.seek(SeekFrom::End(0))?; + return Ok(out); + } + } + } + // Clean EOF — seek back to end for future appends and return. + f.seek(SeekFrom::End(0))?; + Ok(out) + } +} + +/// Outcome of attempting to read a single WAL record. `Torn` means a +/// short read or out-of-range field length was detected mid-record; +/// the caller truncates the file to the last `Ok` offset. +enum ReadOne { + Ok(ResolverRecord, u64), + Eof, + Torn, +} + +fn read_one_record(f: &mut File) -> ReadOne { + let mut sid_buf = [0u8; 8]; + match f.read_exact(&mut sid_buf) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return ReadOne::Eof, + Err(_) => return ReadOne::Torn, + } + let sid = u64::from_le_bytes(sid_buf); + + let mut len_buf = [0u8; 4]; + if f.read_exact(&mut len_buf).is_err() { + return ReadOne::Torn; + } + let metric_len = u32::from_le_bytes(len_buf) as usize; + if metric_len > MAX_METRIC_LEN { + return ReadOne::Torn; + } + let mut metric_bytes = vec![0u8; metric_len]; + if f.read_exact(&mut metric_bytes).is_err() { + return ReadOne::Torn; + } + + if f.read_exact(&mut len_buf).is_err() { + return ReadOne::Torn; + } + let fp_len = u32::from_le_bytes(len_buf) as usize; + if fp_len > MAX_FP_LEN { + return ReadOne::Torn; + } + let mut fp_bytes = vec![0u8; fp_len]; + if f.read_exact(&mut fp_bytes).is_err() { + return ReadOne::Torn; + } + + let metric = match String::from_utf8(metric_bytes) { + Ok(s) => s, + Err(_) => return ReadOne::Torn, + }; + let attrs_fingerprint = match String::from_utf8(fp_bytes) { + Ok(s) => s, + Err(_) => return ReadOne::Torn, + }; + let new_offset = match f.stream_position() { + Ok(p) => p, + Err(_) => return ReadOne::Torn, + }; + ReadOne::Ok( + ResolverRecord { + sid, + metric, + attrs_fingerprint, + }, + new_offset, + ) +} + /// Compute the canonical attributes fingerprint matching the patched /// OTel-Go exporter's `attributesFingerprint`. Both sides MUST produce /// the same string for the same attribute set — sender uses it to look @@ -172,3 +547,180 @@ mod tests { assert_eq!(r.lookup("m", "k=v2;"), None); } } + +#[cfg(test)] +mod persistence_tests { + use super::*; + use tempfile::TempDir; + + fn wal_path(dir: &TempDir) -> PathBuf { + dir.path().join("series_resolver.wal") + } + + #[test] + fn empty_log_replays_empty() { + let dir = TempDir::new().unwrap(); + let p = FilePersistence::open(wal_path(&dir)).unwrap(); + let records = p.replay().unwrap(); + assert!(records.is_empty()); + } + + #[test] + fn append_then_replay_round_trips() { + let dir = TempDir::new().unwrap(); + let p = FilePersistence::open(wal_path(&dir)).unwrap(); + p.append(1, "metric_a", "zone=z0;").unwrap(); + p.append(2, "metric_a", "zone=z1;").unwrap(); + p.append(3, "metric_b", "zone=z0;").unwrap(); + + // Reopen to confirm durability across handle close. + drop(p); + let p2 = FilePersistence::open(wal_path(&dir)).unwrap(); + let records = p2.replay().unwrap(); + assert_eq!(records.len(), 3); + assert_eq!(records[0].sid, 1); + assert_eq!(records[0].metric, "metric_a"); + assert_eq!(records[0].attrs_fingerprint, "zone=z0;"); + assert_eq!(records[1].sid, 2); + assert_eq!(records[2].metric, "metric_b"); + } + + #[test] + fn header_mismatch_errors_on_open() { + let dir = TempDir::new().unwrap(); + let path = wal_path(&dir); + // Write a non-WAL file at the path. + std::fs::write(&path, b"NOTAWAL!extra bytes").unwrap(); + let err = FilePersistence::open(path).unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + } + + #[test] + fn torn_record_truncated_on_replay() { + let dir = TempDir::new().unwrap(); + let path = wal_path(&dir); + // Write two clean records, then a torn third one (sid + len + // header but truncated payload). + { + let p = FilePersistence::open(path.clone()).unwrap(); + p.append(1, "m", "k=v;").unwrap(); + p.append(2, "m", "k=w;").unwrap(); + } + // Manually append a torn record: sid (8B) + metric_len=999 + // (claims 999 bytes of metric but we write 0 bytes after). + { + use std::io::Write; + let mut f = OpenOptions::new().append(true).open(&path).unwrap(); + f.write_all(&3u64.to_le_bytes()).unwrap(); + f.write_all(&999u32.to_le_bytes()).unwrap(); + // No payload bytes — replay reads metric_len=999 then + // hits EOF. + f.sync_all().unwrap(); + } + let pre_size = std::fs::metadata(&path).unwrap().len(); + let p = FilePersistence::open(path.clone()).unwrap(); + let records = p.replay().unwrap(); + assert_eq!(records.len(), 2, "only the two clean records survive"); + let post_size = std::fs::metadata(&path).unwrap().len(); + assert!( + post_size < pre_size, + "torn tail truncated: pre={pre_size} post={post_size}" + ); + // After truncation, subsequent appends pick up from the + // truncated EOF — no gap, no rewrite of historical records. + p.append(3, "m", "k=x;").unwrap(); + drop(p); + let p2 = FilePersistence::open(path).unwrap(); + let records2 = p2.replay().unwrap(); + assert_eq!(records2.len(), 3); + assert_eq!(records2[2].sid, 3); + } + + #[test] + fn out_of_range_metric_len_treated_as_torn() { + // A corrupted file might claim a 4GB metric name. The replay + // must NOT allocate that much; the bounds check rejects it as + // torn instead. + let dir = TempDir::new().unwrap(); + let path = wal_path(&dir); + { + let p = FilePersistence::open(path.clone()).unwrap(); + p.append(1, "m", "k=v;").unwrap(); + } + { + use std::io::Write; + let mut f = OpenOptions::new().append(true).open(&path).unwrap(); + f.write_all(&2u64.to_le_bytes()).unwrap(); + // metric_len = MAX_METRIC_LEN + 1 — over the cap. + f.write_all(&((MAX_METRIC_LEN as u32) + 1).to_le_bytes()) + .unwrap(); + f.sync_all().unwrap(); + } + let p = FilePersistence::open(path).unwrap(); + let records = p.replay().unwrap(); + assert_eq!(records.len(), 1); + } + + #[test] + fn resolver_open_replays_existing_log() { + let dir = TempDir::new().unwrap(); + let path = wal_path(&dir); + // First process: mint three bindings. + { + let r = SeriesIdResolver::open(path.clone()).unwrap(); + let s1 = r.resolve("m", "k=v0;"); + let s2 = r.resolve("m", "k=v1;"); + let s3 = r.resolve("m", "k=v2;"); + assert_eq!(s1, 1); + assert_eq!(s2, 2); + assert_eq!(s3, 3); + } + // Second process: reopen, same inputs return the same sids; + // a fresh input mints sid=4 (max replayed + 1). + { + let r = SeriesIdResolver::open(path).unwrap(); + assert_eq!(r.resolve("m", "k=v0;"), 1); + assert_eq!(r.resolve("m", "k=v1;"), 2); + assert_eq!(r.resolve("m", "k=v2;"), 3); + let fresh = r.resolve("m", "k=v3;"); + assert_eq!(fresh, 4, "next_sid resumes at max(replayed)+1"); + } + } + + #[test] + fn resolver_with_noop_persistence_does_not_persist() { + // Sanity check: NoopPersistence is the back-compat path; resolver + // bindings reset across construction. + let r1 = SeriesIdResolver::new(); + let s1 = r1.resolve("m", "k=v;"); + drop(r1); + let r2 = SeriesIdResolver::new(); + let s2 = r2.resolve("m", "k=v;"); + // Both resolvers start fresh, so both mint sid=1. + assert_eq!(s1, 1); + assert_eq!(s2, 1); + } + + #[test] + fn append_failure_logs_but_does_not_panic() { + // A custom persistence impl that always returns an error. + // The resolver should log + return the sid anyway (in-memory- + // only); subsequent resolves for the same key hit the cache + // and don't re-attempt append. + struct FailingPersistence; + impl SeriesResolverPersistence for FailingPersistence { + fn append(&self, _: u64, _: &str, _: &str) -> std::io::Result<()> { + Err(std::io::Error::other("simulated I/O failure")) + } + fn replay(&self) -> std::io::Result> { + Ok(Vec::new()) + } + } + let r = SeriesIdResolver::with_persistence(Arc::new(FailingPersistence)); + let sid = r.resolve("m", "k=v;"); + assert_eq!(sid, 1, "resolver returns the sid despite persistence error"); + // Second call hits the cache; no second append attempt. + let sid2 = r.resolve("m", "k=v;"); + assert_eq!(sid, sid2); + } +} diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index ef5898e06..3ce89e8ae 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -310,8 +310,24 @@ async fn main() -> Result<()> { // are constructed so both can be wired with a single canonical // instance — even when precompute is disabled, the engine still // needs the index for the Phase 6 archive failover trigger. - let series_resolver = - Arc::new(data_plane::drivers::ingest::series_resolver::SeriesIdResolver::new()); + // Under --persistence-enabled, the resolver replays its WAL on + // startup so the agent's cached sids stay valid across backend + // restarts. Without persistence (tests, stateless deploys), every + // restart drops the cache; agents recover via the existing + // `unknown_series_ids` eviction primitive — one extra round trip + // per identity on the first emit post-restart. + let series_resolver = if args.persistence_enabled { + use data_plane::drivers::ingest::series_resolver::SeriesIdResolver; + let dir = args + .persistence_dir + .as_ref() + .expect("--persistence-enabled requires --persistence-dir"); + let wal_path = std::path::PathBuf::from(dir).join("series_resolver.wal"); + info!("opening series-resolver WAL at {:?}", wal_path); + Arc::new(SeriesIdResolver::open(wal_path)?) + } else { + Arc::new(data_plane::drivers::ingest::series_resolver::SeriesIdResolver::new()) + }; let sketch_index = Arc::new(data_plane::storage_engines::sketch_db::index::SketchStore::new()); From f72b533a903fd8dbf8c3fa80d643cde519d13579 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 13 May 2026 11:00:06 -0600 Subject: [PATCH 3/3] feat(ingest): resolver key = (metric, fp, agg_kind); delete compute_sketch_sid (PR-3/3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the Interpretation-B identity model on the OTel ingest path. Sid identity is now `(metric, attrs_fingerprint, agg_kind_canonical)` — the same 3-tuple `compute_sketch_sid` hashed over, just held as a registry-allocated u64 instead of a content-addressed one. Two aggregations over the same series (e.g. DDSketch and Sum on `http_latency_ms{zone=z0}`) now mint DISTINCT sids, matching the behaviour that existed before PR-1. PR-1 had a latent regression: the resolver key was only `(metric, fp)`, so two sketch kinds over the same series would collapse to one sid and the second's metadata would silently overwrite the first at `SketchStore::register`. This PR fixes that by threading `agg_kind` through to the resolver. Changes: - `AggKind::canonical_string()` — stable string form on `sketch_db::data::AggKind`. Used as the third element of the resolver's cache key and as the new `agg_kind_canonical` field in the WAL. Examples: `"sketch:DDSketch:D:0.01"`, `"precompute:Sum:"`. - `SeriesIdResolver::resolve(metric, fp, agg_kind_canonical)` — added the third arg. `lookup` similarly. Caller passes the canonical string (resolver doesn't depend on `AggKind` type, just on a `&str`). - WAL bumped to v2: header `ASAPSRP\x02`, records gain a 4th length-prefixed field (`agg_kind_len` u32 LE + bytes). No v1 migration: PR-2 hasn't shipped to production, so v1 files don't exist in the wild. v1 headers error on open with a clear message. - OTel modified-OTLP sketch ingest at `otel.rs:879` now builds `AggKind::Sketch { kind, config }`, calls `canonical_string()`, passes it to the resolver. The wire-case comment block rewrites to document the Option-B identity contract. - `ResolveSeriesIDs` gRPC handler stubbed to return empty `assignments` + a one-shot WARN log. The proto's `SeriesQuery` carries only `(metric, fp)`; under the new identity model a pre-resolve here can't produce the right sid. Drop or extend the proto in a follow-up. - `compute_sketch_sid` and its 5 unit tests deleted + the legacy- parity test deleted. `compute_sid` stays alive: still called by `SketchStore::ingest_precompute_for_agg_config` for PRECOMPUTE aggregations. PR-4 migrates that path to the resolver (touching 5 callers: output_sink, eviction, backfill, 2 test sites) and deletes `compute_sid` + `sketch_kind_tag` + `encode_sketch_config`. Tests added: - `distinct_agg_kinds_same_series_distinct_sids` — sid identity contract under Interpretation B. - `resolver_open_distinguishes_agg_kinds_on_replay` — WAL v2 records agg_kind correctly and replay rebuilds the 3-tuple cache. - All existing resolver + persistence tests updated for the new signature. cargo build -p data_plane (lib + bin): clean cargo test -p data_plane --lib: 763/763 passing cargo test -p data_plane --lib drivers::ingest: 24/24 passing Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/drivers/ingest/otel.rs | 72 +++-- .../src/drivers/ingest/series_resolver.rs | 253 +++++++++++++----- .../src/storage_engines/sketch_db/data/mod.rs | 120 ++++++--- .../storage_engines/sketch_db/index/mod.rs | 90 +------ 4 files changed, 332 insertions(+), 203 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 84a6f2b47..391e50bf2 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -218,29 +218,39 @@ impl MetricsService for MetricsServiceImpl { Response, Status, > { - use asap_otel_proto::tonic::collector::metrics::v1::{ - ResolveSeriesIDsResponse, SeriesAssignment, - }; + use asap_otel_proto::tonic::collector::metrics::v1::ResolveSeriesIDsResponse; + + // Pre-resolve handshake is structurally redundant with the + // canonical Export path under Interpretation B (sid identity is + // `(metric, attrs_fingerprint, agg_kind_canonical)` and + // `agg_kind` isn't carried in `SeriesQuery`). On the first + // Export with attrs, the backend resolves the correct sid and + // echoes it back via `ExportMetricsServiceResponse.series_assignments` + // — that's the canonical channel, and it also handles every + // cache-divergence failure mode (proto comment at + // `ExportMetricsServiceResponse.unknown_series_ids:100-104`). + // + // The RPC is kept as a stable surface so older patched + // exporters that still call it don't see `Unimplemented`. The + // returned `assignments` vec is empty; the agent's local cache + // stays cold and the first real Export populates it. + // + // Follow-up: drop the RPC entirely (proto change), OR extend + // `SeriesQuery` to carry `agg_kind_canonical` so this can + // perform a real pre-resolve. Today's stub is the no-harm path. let req = request.into_inner(); - let mut assignments = Vec::with_capacity(req.queries.len()); - if let Some(state) = &self.shared.ingest_state { - for q in req.queries { - // The fingerprint travels as opaque bytes on the wire, but - // the resolver hashes it as a string (the sender's - // canonical fingerprint algorithm matches our - // `canonical_attrs_fingerprint`). UTF-8 is lossy here only - // for malformed inputs — those produce a degraded but - // deterministic key, never a panic. - let fp = String::from_utf8_lossy(&q.attributes_fingerprint).into_owned(); - let sid = state.series_resolver.resolve(&q.metric_name, &fp); - assignments.push(SeriesAssignment { - attributes_fingerprint: q.attributes_fingerprint, - series_id: sid, - ..Default::default() - }); - } + if !req.queries.is_empty() { + warn!( + queries = req.queries.len(), + "ResolveSeriesIDs RPC called with non-empty batch; \ + returning empty assignments — sid identity now \ + includes agg_kind which this RPC doesn't carry. \ + Agent will mint on first Export with attrs.", + ); } - Ok(Response::new(ResolveSeriesIDsResponse { assignments })) + Ok(Response::new(ResolveSeriesIDsResponse { + assignments: Vec::new(), + })) } } @@ -935,8 +945,24 @@ async fn route_modified_otlp_sketches_to_precompute( } } } else { - let assigned = - ingest_state.series_resolver.resolve(&metric.name, &fp); + // Build the canonical AggKind string for this DP so + // the resolver's cache key is `(metric, fp, agg_kind)`. + // Two DPs over the same (metric, attrs) but different + // sketch kinds/configs (e.g. DDSketch vs Kll, or two + // DDSketches at different relative_accuracy) get + // SEPARATE sids — matching the identity model the + // retired `compute_sketch_sid` hashed over. + let kind_for_sid = sketch_kind_handle_for(&dp); + let agg_kind = crate::storage_engines::sketch_db::data::AggKind::Sketch { + kind: kind_for_sid, + config: dp.container_config.clone(), + }; + let agg_kind_canonical = agg_kind.canonical_string(); + let assigned = ingest_state.series_resolver.resolve( + &metric.name, + &fp, + &agg_kind_canonical, + ); if dp.series_id != 0 && dp.series_id != assigned { // Sender's cached sid disagrees with the // resolver's binding — sender's cache is diff --git a/data_plane/src/drivers/ingest/series_resolver.rs b/data_plane/src/drivers/ingest/series_resolver.rs index ee9fb49ab..1b824a37e 100644 --- a/data_plane/src/drivers/ingest/series_resolver.rs +++ b/data_plane/src/drivers/ingest/series_resolver.rs @@ -32,14 +32,20 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use tracing::{info, warn}; -/// Canonical fingerprint key — `(metric_name, attrs_fingerprint)`. +/// Canonical sid identity — `(metric_name, attrs_fingerprint, agg_kind_canonical)`. /// -/// `attrs_fingerprint` is a string produced by canonicalizing the -/// attribute set: keys sorted lexicographically, then `key=value;`-joined. -/// This matches the format the patched OTel-Go exporter writes into -/// `SeriesAssignment.attributes_fingerprint`, so cache hits across the -/// agent's exporter cache and this backend resolver align bit-exactly. -type CacheKey = (String, String); +/// This is the same 3-tuple that the retired `compute_sketch_sid` / +/// `compute_sid` functions hashed over, just held as a string key +/// for the registry-allocated mint path. +/// +/// - `attrs_fingerprint` — keys sorted lexicographically, then +/// `key=value;`-joined. Matches the patched OTel-Go exporter so +/// sender and receiver agree bit-exactly. +/// - `agg_kind_canonical` — the stable string form of `AggKind` (see +/// `sketch_db::data::AggKind::canonical_string`). Distinguishes +/// different aggregations over the same series — e.g. a DDSketch +/// and a Sum on the same `(metric, attrs)` get separate sids. +type CacheKey = (String, String, String); /// Idempotent compute-or-mint resolver. Atomic per-key — concurrent /// `resolve()` calls for the same `(metric, attrs)` from different agents @@ -98,7 +104,10 @@ impl SeriesIdResolver { let cache: DashMap = DashMap::new(); let mut max_sid: u64 = 0; for r in records { - cache.insert((r.metric, r.attrs_fingerprint), r.sid); + cache.insert( + (r.metric, r.attrs_fingerprint, r.agg_kind_canonical), + r.sid, + ); if r.sid > max_sid { max_sid = r.sid; } @@ -112,11 +121,15 @@ impl SeriesIdResolver { }) } - /// Resolve `(metric_name, attrs)` to a series_id. Returns the existing - /// sid if this `(metric, attrs)` tuple was already registered; - /// otherwise mints a fresh sid, durably persists the binding (when - /// a non-noop backend is wired), caches it, and returns the new - /// value. + /// Resolve `(metric, attrs, agg_kind)` to a series_id. Returns the + /// existing sid if this 3-tuple was already registered; otherwise + /// mints a fresh sid, durably persists the binding (when a non-noop + /// backend is wired), caches it, and returns the new value. + /// + /// `agg_kind_canonical` is the stable string form of + /// [`sketch_db::data::AggKind`] (call its `canonical_string()` + /// method at the call site). It's a string here so the resolver + /// doesn't take a dep on storage_engines. /// /// Idempotent: repeated calls with the same input ALWAYS return the /// same sid for the lifetime of the cache. With `FilePersistence`, @@ -129,8 +142,17 @@ impl SeriesIdResolver { /// the resolver stays in-memory-correct. Next restart will not /// recover the lost mint, and the agent will hit the eviction /// recovery path (one extra round trip with attrs). - pub fn resolve(&self, metric_name: &str, attrs_fingerprint: &str) -> u64 { - let key = (metric_name.to_string(), attrs_fingerprint.to_string()); + pub fn resolve( + &self, + metric_name: &str, + attrs_fingerprint: &str, + agg_kind_canonical: &str, + ) -> u64 { + let key = ( + metric_name.to_string(), + attrs_fingerprint.to_string(), + agg_kind_canonical.to_string(), + ); // Fast path: read-only check on the cache before taking the // bucket's write lock. DashMap's `get` takes a shard read lock; // the common case (a hit on a known identity) never serializes @@ -145,10 +167,12 @@ impl SeriesIdResolver { // is durable before any caller observes the sid. let entry = self.cache.entry(key).or_insert_with(|| { let sid = self.next_sid.fetch_add(1, Ordering::Relaxed); - if let Err(e) = - self.persistence - .append(sid, metric_name, attrs_fingerprint) - { + if let Err(e) = self.persistence.append( + sid, + metric_name, + attrs_fingerprint, + agg_kind_canonical, + ) { warn!( metric = %metric_name, sid, @@ -163,12 +187,18 @@ impl SeriesIdResolver { } /// Look up an existing sid without minting. Returns `None` if the - /// `(metric, attrs)` tuple is not in the cache. Used by the OTLP - /// receive path to check whether an incoming sid (without attrs) is - /// recognized — sids the backend doesn't recognize go into the - /// response's `unknown_series_ids` so the sender re-sends with attrs. - pub fn lookup(&self, metric_name: &str, attrs_fingerprint: &str) -> Option { - let key = (metric_name.to_string(), attrs_fingerprint.to_string()); + /// `(metric, attrs, agg_kind)` tuple is not in the cache. + pub fn lookup( + &self, + metric_name: &str, + attrs_fingerprint: &str, + agg_kind_canonical: &str, + ) -> Option { + let key = ( + metric_name.to_string(), + attrs_fingerprint.to_string(), + agg_kind_canonical.to_string(), + ); self.cache.get(&key).map(|v| *v) } @@ -194,18 +224,20 @@ impl Default for SeriesIdResolver { // ── Persistence ────────────────────────────────────────────────────────────── // -// The resolver's `(metric, fp) → sid` cache is in-memory only by default. -// Under `--persistence-enabled`, a `FilePersistence` backend writes a WAL -// record per fresh mint; the resolver replays it on startup so the agent's -// cached sids stay valid across backend restarts. +// The resolver's `(metric, fp, agg_kind) → sid` cache is in-memory only +// by default. Under `--persistence-enabled`, a `FilePersistence` backend +// writes a WAL record per fresh mint; the resolver replays it on startup +// so the agent's cached sids stay valid across backend restarts. // -// WAL format v1: -// header: 8 bytes → b"ASAPSRP\x01" +// WAL format v2 (current; v1 was attrs-only, never shipped to prod): +// header: 8 bytes → b"ASAPSRP\x02" // record: 8 bytes → sid (u64 little-endian) // 4 bytes → metric_len (u32 LE) // metric_len bytes → metric utf8 // 4 bytes → fp_len (u32 LE) // fp_len bytes → fp utf8 +// 4 bytes → agg_kind_len (u32 LE) +// agg_kind_len bytes → agg_kind_canonical utf8 // // Append-only; sids are minted once and never rewritten, so the log size // is proportional to live cardinality. At 100M sids (~5GB) compaction @@ -224,6 +256,7 @@ pub struct ResolverRecord { pub sid: u64, pub metric: String, pub attrs_fingerprint: String, + pub agg_kind_canonical: String, } /// Durability hook for [`SeriesIdResolver`]. Implementations decide @@ -231,9 +264,9 @@ pub struct ResolverRecord { /// tests and stateless deployments; `FilePersistence` is the production /// answer under `--persistence-enabled`. pub trait SeriesResolverPersistence: Send + Sync { - /// Durably record a fresh `(sid, metric, fp)` binding. MUST be - /// flushed to stable storage before returning `Ok` — the resolver - /// only returns the sid to its caller after this returns. + /// Durably record a fresh `(sid, metric, fp, agg_kind)` binding. + /// MUST be flushed to stable storage before returning `Ok` — the + /// resolver only returns the sid to its caller after this returns. /// On error, the binding is in-memory-only; the caller logs and /// continues. fn append( @@ -241,6 +274,7 @@ pub trait SeriesResolverPersistence: Send + Sync { sid: u64, metric: &str, attrs_fingerprint: &str, + agg_kind_canonical: &str, ) -> std::io::Result<()>; /// Read every durable binding in append order. Called once at @@ -254,7 +288,13 @@ pub trait SeriesResolverPersistence: Send + Sync { pub struct NoopPersistence; impl SeriesResolverPersistence for NoopPersistence { - fn append(&self, _sid: u64, _metric: &str, _fp: &str) -> std::io::Result<()> { + fn append( + &self, + _sid: u64, + _metric: &str, + _fp: &str, + _agg_kind_canonical: &str, + ) -> std::io::Result<()> { Ok(()) } @@ -272,15 +312,16 @@ pub struct FilePersistence { path: PathBuf, } -const WAL_MAGIC: &[u8; 8] = b"ASAPSRP\x01"; +const WAL_MAGIC: &[u8; 8] = b"ASAPSRP\x02"; /// Reject any single field whose length-prefix exceeds these caps. A /// corrupted file might claim huge field lengths; without these bounds /// the replay loop could allocate gigabytes of zeros before discovering /// the lengths don't match the actual content. The caps are far above /// any realistic input — metric names are tens of bytes, fingerprints -/// are hundreds. +/// are hundreds, agg_kind canonical strings are tens. const MAX_METRIC_LEN: usize = 16 * 1024; const MAX_FP_LEN: usize = 64 * 1024; +const MAX_AGG_KIND_LEN: usize = 4 * 1024; impl FilePersistence { /// Open or create the WAL at `path`. On a fresh file, writes the @@ -334,13 +375,13 @@ impl SeriesResolverPersistence for FilePersistence { sid: u64, metric: &str, fp: &str, + agg_kind_canonical: &str, ) -> std::io::Result<()> { let metric_bytes = metric.as_bytes(); let fp_bytes = fp.as_bytes(); - let metric_len: u32 = metric_bytes - .len() - .try_into() - .map_err(|_| { + let agg_kind_bytes = agg_kind_canonical.as_bytes(); + let metric_len: u32 = + metric_bytes.len().try_into().map_err(|_| { std::io::Error::new( std::io::ErrorKind::InvalidInput, "metric name longer than u32::MAX bytes", @@ -352,6 +393,13 @@ impl SeriesResolverPersistence for FilePersistence { "fingerprint longer than u32::MAX bytes", ) })?; + let agg_kind_len: u32 = + agg_kind_bytes.len().try_into().map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "agg_kind_canonical longer than u32::MAX bytes", + ) + })?; let mut f = self.file.lock().unwrap(); f.write_all(&sid.to_le_bytes())?; @@ -359,6 +407,8 @@ impl SeriesResolverPersistence for FilePersistence { f.write_all(metric_bytes)?; f.write_all(&fp_len.to_le_bytes())?; f.write_all(fp_bytes)?; + f.write_all(&agg_kind_len.to_le_bytes())?; + f.write_all(agg_kind_bytes)?; // Durability barrier: caller must not observe the sid until the // record is on stable storage. fsync is the slow part of the // mint path (a few ms on SSD) but it's amortized — minting is @@ -459,6 +509,18 @@ fn read_one_record(f: &mut File) -> ReadOne { return ReadOne::Torn; } + if f.read_exact(&mut len_buf).is_err() { + return ReadOne::Torn; + } + let agg_kind_len = u32::from_le_bytes(len_buf) as usize; + if agg_kind_len > MAX_AGG_KIND_LEN { + return ReadOne::Torn; + } + let mut agg_kind_bytes = vec![0u8; agg_kind_len]; + if f.read_exact(&mut agg_kind_bytes).is_err() { + return ReadOne::Torn; + } + let metric = match String::from_utf8(metric_bytes) { Ok(s) => s, Err(_) => return ReadOne::Torn, @@ -467,6 +529,10 @@ fn read_one_record(f: &mut File) -> ReadOne { Ok(s) => s, Err(_) => return ReadOne::Torn, }; + let agg_kind_canonical = match String::from_utf8(agg_kind_bytes) { + Ok(s) => s, + Err(_) => return ReadOne::Torn, + }; let new_offset = match f.stream_position() { Ok(p) => p, Err(_) => return ReadOne::Torn, @@ -476,6 +542,7 @@ fn read_one_record(f: &mut File) -> ReadOne { sid, metric, attrs_fingerprint, + agg_kind_canonical, }, new_offset, ) @@ -507,30 +574,51 @@ pub fn canonical_attrs_fingerprint(attrs: &[(&str, &str)]) -> String { mod tests { use super::*; + /// Stand-in canonical `AggKind` string. Tests don't care about the + /// specific encoding — the resolver only uses the value for key + /// equality. Production callers compute this via + /// `AggKind::canonical_string()`. + const TEST_AGG: &str = "sketch:DDSketch:D:0.01"; + #[test] fn idempotent_same_input_same_sid() { let r = SeriesIdResolver::new(); - let sid1 = r.resolve("http_requests_total", "zone=z0;"); - let sid2 = r.resolve("http_requests_total", "zone=z0;"); + let sid1 = r.resolve("http_requests_total", "zone=z0;", TEST_AGG); + let sid2 = r.resolve("http_requests_total", "zone=z0;", TEST_AGG); assert_eq!(sid1, sid2, "same input must produce same sid"); } #[test] fn distinct_inputs_distinct_sids() { let r = SeriesIdResolver::new(); - let s_z0 = r.resolve("metric_a", "zone=z0;"); - let s_z1 = r.resolve("metric_a", "zone=z1;"); + let s_z0 = r.resolve("metric_a", "zone=z0;", TEST_AGG); + let s_z1 = r.resolve("metric_a", "zone=z1;", TEST_AGG); assert_ne!(s_z0, s_z1); } #[test] fn distinct_metrics_same_attrs_distinct_sids() { let r = SeriesIdResolver::new(); - let s_a = r.resolve("metric_a", "zone=z0;"); - let s_b = r.resolve("metric_b", "zone=z0;"); + let s_a = r.resolve("metric_a", "zone=z0;", TEST_AGG); + let s_b = r.resolve("metric_b", "zone=z0;", TEST_AGG); assert_ne!(s_a, s_b); } + #[test] + fn distinct_agg_kinds_same_series_distinct_sids() { + // Two aggregations over the same (metric, attrs) tuple — e.g. + // a DDSketch and a Sum on `http_latency_ms{zone=z0}` — get + // SEPARATE sids. This is the core property of Interpretation B: + // sid identity is `(metric, attrs, agg_kind)`. + let r = SeriesIdResolver::new(); + let s_dd = r.resolve("http_latency_ms", "zone=z0;", "sketch:DDSketch:D:0.01"); + let s_sum = r.resolve("http_latency_ms", "zone=z0;", "precompute:Sum:"); + assert_ne!( + s_dd, s_sum, + "different agg_kinds over the same series must mint distinct sids", + ); + } + #[test] fn fingerprint_sorts_keys() { let f1 = canonical_attrs_fingerprint(&[("zone", "z0"), ("rack", "r00")]); @@ -542,9 +630,11 @@ mod tests { #[test] fn lookup_returns_existing_without_mint() { let r = SeriesIdResolver::new(); - let sid = r.resolve("m", "k=v;"); - assert_eq!(r.lookup("m", "k=v;"), Some(sid)); - assert_eq!(r.lookup("m", "k=v2;"), None); + let sid = r.resolve("m", "k=v;", TEST_AGG); + assert_eq!(r.lookup("m", "k=v;", TEST_AGG), Some(sid)); + assert_eq!(r.lookup("m", "k=v2;", TEST_AGG), None); + // Same (metric, attrs) but different agg_kind is a miss. + assert_eq!(r.lookup("m", "k=v;", "precompute:Sum:"), None); } } @@ -565,13 +655,16 @@ mod persistence_tests { assert!(records.is_empty()); } + const TEST_AGG: &str = "sketch:DDSketch:D:0.01"; + #[test] fn append_then_replay_round_trips() { let dir = TempDir::new().unwrap(); let p = FilePersistence::open(wal_path(&dir)).unwrap(); - p.append(1, "metric_a", "zone=z0;").unwrap(); - p.append(2, "metric_a", "zone=z1;").unwrap(); - p.append(3, "metric_b", "zone=z0;").unwrap(); + p.append(1, "metric_a", "zone=z0;", TEST_AGG).unwrap(); + p.append(2, "metric_a", "zone=z1;", TEST_AGG).unwrap(); + p.append(3, "metric_b", "zone=z0;", "precompute:Sum:") + .unwrap(); // Reopen to confirm durability across handle close. drop(p); @@ -581,8 +674,10 @@ mod persistence_tests { assert_eq!(records[0].sid, 1); assert_eq!(records[0].metric, "metric_a"); assert_eq!(records[0].attrs_fingerprint, "zone=z0;"); + assert_eq!(records[0].agg_kind_canonical, TEST_AGG); assert_eq!(records[1].sid, 2); assert_eq!(records[2].metric, "metric_b"); + assert_eq!(records[2].agg_kind_canonical, "precompute:Sum:"); } #[test] @@ -603,8 +698,8 @@ mod persistence_tests { // header but truncated payload). { let p = FilePersistence::open(path.clone()).unwrap(); - p.append(1, "m", "k=v;").unwrap(); - p.append(2, "m", "k=w;").unwrap(); + p.append(1, "m", "k=v;", TEST_AGG).unwrap(); + p.append(2, "m", "k=w;", TEST_AGG).unwrap(); } // Manually append a torn record: sid (8B) + metric_len=999 // (claims 999 bytes of metric but we write 0 bytes after). @@ -628,7 +723,7 @@ mod persistence_tests { ); // After truncation, subsequent appends pick up from the // truncated EOF — no gap, no rewrite of historical records. - p.append(3, "m", "k=x;").unwrap(); + p.append(3, "m", "k=x;", TEST_AGG).unwrap(); drop(p); let p2 = FilePersistence::open(path).unwrap(); let records2 = p2.replay().unwrap(); @@ -645,7 +740,7 @@ mod persistence_tests { let path = wal_path(&dir); { let p = FilePersistence::open(path.clone()).unwrap(); - p.append(1, "m", "k=v;").unwrap(); + p.append(1, "m", "k=v;", TEST_AGG).unwrap(); } { use std::io::Write; @@ -668,9 +763,9 @@ mod persistence_tests { // First process: mint three bindings. { let r = SeriesIdResolver::open(path.clone()).unwrap(); - let s1 = r.resolve("m", "k=v0;"); - let s2 = r.resolve("m", "k=v1;"); - let s3 = r.resolve("m", "k=v2;"); + let s1 = r.resolve("m", "k=v0;", TEST_AGG); + let s2 = r.resolve("m", "k=v1;", TEST_AGG); + let s3 = r.resolve("m", "k=v2;", TEST_AGG); assert_eq!(s1, 1); assert_eq!(s2, 2); assert_eq!(s3, 3); @@ -679,23 +774,43 @@ mod persistence_tests { // a fresh input mints sid=4 (max replayed + 1). { let r = SeriesIdResolver::open(path).unwrap(); - assert_eq!(r.resolve("m", "k=v0;"), 1); - assert_eq!(r.resolve("m", "k=v1;"), 2); - assert_eq!(r.resolve("m", "k=v2;"), 3); - let fresh = r.resolve("m", "k=v3;"); + assert_eq!(r.resolve("m", "k=v0;", TEST_AGG), 1); + assert_eq!(r.resolve("m", "k=v1;", TEST_AGG), 2); + assert_eq!(r.resolve("m", "k=v2;", TEST_AGG), 3); + let fresh = r.resolve("m", "k=v3;", TEST_AGG); assert_eq!(fresh, 4, "next_sid resumes at max(replayed)+1"); } } + #[test] + fn resolver_open_distinguishes_agg_kinds_on_replay() { + // Same (metric, attrs) but two agg_kinds — both replay to the + // resolver as distinct keys, and re-resolving each returns its + // original sid. + let dir = TempDir::new().unwrap(); + let path = wal_path(&dir); + { + let r = SeriesIdResolver::open(path.clone()).unwrap(); + let s_dd = r.resolve("m", "k=v0;", "sketch:DDSketch:D:0.01"); + let s_sum = r.resolve("m", "k=v0;", "precompute:Sum:"); + assert_ne!(s_dd, s_sum); + } + { + let r = SeriesIdResolver::open(path).unwrap(); + assert_eq!(r.resolve("m", "k=v0;", "sketch:DDSketch:D:0.01"), 1); + assert_eq!(r.resolve("m", "k=v0;", "precompute:Sum:"), 2); + } + } + #[test] fn resolver_with_noop_persistence_does_not_persist() { // Sanity check: NoopPersistence is the back-compat path; resolver // bindings reset across construction. let r1 = SeriesIdResolver::new(); - let s1 = r1.resolve("m", "k=v;"); + let s1 = r1.resolve("m", "k=v;", TEST_AGG); drop(r1); let r2 = SeriesIdResolver::new(); - let s2 = r2.resolve("m", "k=v;"); + let s2 = r2.resolve("m", "k=v;", TEST_AGG); // Both resolvers start fresh, so both mint sid=1. assert_eq!(s1, 1); assert_eq!(s2, 1); @@ -709,7 +824,7 @@ mod persistence_tests { // and don't re-attempt append. struct FailingPersistence; impl SeriesResolverPersistence for FailingPersistence { - fn append(&self, _: u64, _: &str, _: &str) -> std::io::Result<()> { + fn append(&self, _: u64, _: &str, _: &str, _: &str) -> std::io::Result<()> { Err(std::io::Error::other("simulated I/O failure")) } fn replay(&self) -> std::io::Result> { @@ -717,10 +832,10 @@ mod persistence_tests { } } let r = SeriesIdResolver::with_persistence(Arc::new(FailingPersistence)); - let sid = r.resolve("m", "k=v;"); + let sid = r.resolve("m", "k=v;", TEST_AGG); assert_eq!(sid, 1, "resolver returns the sid despite persistence error"); // Second call hits the cache; no second append attempt. - let sid2 = r.resolve("m", "k=v;"); + let sid2 = r.resolve("m", "k=v;", TEST_AGG); assert_eq!(sid, sid2); } } diff --git a/data_plane/src/storage_engines/sketch_db/data/mod.rs b/data_plane/src/storage_engines/sketch_db/data/mod.rs index d279c301c..27e6c2002 100644 --- a/data_plane/src/storage_engines/sketch_db/data/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/data/mod.rs @@ -26,8 +26,10 @@ //! the read path (sid + label values + per-window samples). //! - [`AccuracyBound`] — `(epsilon, confidence)` derived from a //! `SketchConfig`. Surfaces in HTTP response headers. -//! - [`compute_sid`] / [`compute_sketch_sid`] — the canonical sid -//! hash. Deterministic across hosts; defines the sid identity. +//! - [`compute_sid`] — the canonical sid hash used today only by the +//! precompute ingest path (`SketchStore::ingest_precompute_for_agg_config`). +//! The OTel sketch ingest path moved to `SeriesIdResolver` (Option B +//! — registry-allocated sids); the precompute path follows in PR-4. //! - [`canonical_parameters`] — helper that renders a parameters //! `HashMap` into the canonical string form //! `AggKind::Precompute::parameters_canonical` expects. @@ -120,41 +122,91 @@ pub fn canonical_parameters( buf } -// ── sid hash ──────────────────────────────────────────────────────────────── +impl AggKind { + /// Stable string form of this `AggKind`, used as the third element + /// of the `SeriesIdResolver` cache key and as the `agg_kind_canonical` + /// field in the resolver's WAL. + /// + /// Sid identity is `(metric, attrs_fingerprint, agg_kind_canonical)` + /// — the same canonical tuple that the retired + /// [`compute_sketch_sid`] / [`compute_sid`] functions hashed over, + /// just rendered as a string for the registry-allocated mint path + /// instead of byte-fed to xxh64. Two `AggKind`s that compare equal + /// MUST produce the same canonical string; two that differ in any + /// observable parameter MUST produce different strings. + /// + /// Format: + /// - `Sketch { kind: DDSketch, config: DDSketch{rel_acc: 0.01} }` + /// → `"sketch:DDSketch:D:0.01"` + /// - `Sketch { kind: Kll, config: Kll{k: 200} }` + /// → `"sketch:Kll:K:200"` + /// - `Precompute { agg_type: Sum, parameters_canonical: "" }` + /// → `"precompute:Sum:"` + pub fn canonical_string(&self) -> String { + match self { + AggKind::Sketch { kind, config } => { + format!( + "sketch:{}:{}", + sketch_kind_canonical(*kind), + sketch_config_canonical(config), + ) + } + AggKind::Precompute { + agg_type, + parameters_canonical, + } => { + // `AggregationType`'s `Display` impl is stable + // (matches the snake-case form on the wire) and + // `parameters_canonical` is already canonicalized + // upstream (see [`canonical_parameters`]). + format!("precompute:{}:{}", agg_type, parameters_canonical) + } + } + } +} -/// 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. -/// -/// `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. -/// -/// 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 { - compute_sid( - metric_name, - attrs_fingerprint, - &AggKind::Sketch { - kind: sketch_kind, - config: sketch_config.clone(), - }, - ) +fn sketch_kind_canonical(k: SketchKindHandle) -> &'static str { + match k { + SketchKindHandle::DDSketch => "DDSketch", + SketchKindHandle::Kll => "Kll", + SketchKindHandle::Hll => "Hll", + SketchKindHandle::CountSketch => "CountSketch", + SketchKindHandle::CountMin => "CountMin", + SketchKindHandle::CmsWithHeap => "CmsWithHeap", + SketchKindHandle::CountSketchWithHeap => "CountSketchWithHeap", + // `Any` is the analysis-time wildcard; never reaches the + // ingest path which detects a concrete kind from the OTLP + // wire variant. Mapping it to a unique tag anyway keeps the + // canonical form total. + SketchKindHandle::Any => "Any", + } +} + +fn sketch_config_canonical(cfg: &SketchConfig) -> String { + match cfg { + SketchConfig::DDSketch { relative_accuracy } => { + format!("D:{relative_accuracy}") + } + SketchConfig::Kll { k } => format!("K:{k}"), + SketchConfig::Hll { precision } => format!("H:{precision}"), + SketchConfig::CountSketch { rows, cols } => format!("S:{rows}:{cols}"), + SketchConfig::CountMin { rows, cols } => format!("M:{rows}:{cols}"), + } } -/// Generalized version of [`compute_sketch_sid`] covering both sketch -/// and precompute aggregations. Same `(metric, attrs, agg_kind)` tuple -/// always yields the same sid. +// ── sid hash ──────────────────────────────────────────────────────────────── +// +// `compute_sketch_sid` was retired alongside the OTel ingest path's +// migration to `SeriesIdResolver` (Option B — registry-allocated sids). +// The remaining `compute_sid` function is still called by +// `SketchStore::ingest_precompute_for_agg_config` for PRECOMPUTE +// aggregations; that path migrates to the resolver in PR-4, at which +// point `compute_sid` and its `sketch_kind_tag` / `encode_sketch_config` +// helpers will go away too. Sketch sids today come exclusively from the +// resolver — same authority as precompute sids will after PR-4. + +/// Generalized content-addressed sid hash. Same `(metric, attrs, agg_kind)` +/// tuple always yields the same sid. /// /// The two branches encode disjointly: a `Sketch` payload starts with /// `sketch_kind_tag` (1..=7), while a `Precompute` payload starts with diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 297217b96..dc6cd31e2 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -33,9 +33,9 @@ use crate::storage_engines::sketch_db::lifecycle::AggStatus; // (`crate::storage_engines::sketch_db::index::*`) keep compiling // during the reorg. pub use crate::storage_engines::sketch_db::data::{ - canonical_parameters, compute_sid, compute_sketch_sid, AccuracyBound, AggKind, AggPayload, - AggregationType, Capability, SketchConfig, SketchEncoding, SketchKindHandle, - SketchSampleState, SketchTimeSeries, + canonical_parameters, compute_sid, AccuracyBound, AggKind, AggPayload, AggregationType, + Capability, SketchConfig, SketchEncoding, SketchKindHandle, SketchSampleState, + SketchTimeSeries, }; fn now_ms() -> u64 { @@ -1126,60 +1126,11 @@ mod tests { 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); - } + // `compute_sketch_sid_*` tests removed alongside the function they + // exercised. The same identity properties (metric/attrs/kind/config + // disambiguate sketch sids) are now covered by the resolver's own + // `distinct_agg_kinds_same_series_distinct_sids` test plus the + // round-trip parity captured at the OTel ingest layer. #[test] fn compute_sid_precompute_is_deterministic() { @@ -1456,26 +1407,11 @@ mod tests { assert!(precompute.as_precompute().is_some()); } - #[test] - fn compute_sketch_sid_matches_new_compute_sid_for_sketch_branch() { - // The legacy `compute_sketch_sid` wrapper must produce - // the exact same value as `compute_sid` with an - // `AggKind::Sketch` — otherwise existing ingest sids - // would skew across the migration. - let cfg = SketchConfig::DDSketch { - relative_accuracy: 0.01, - }; - let legacy = compute_sketch_sid("m", "zone=z0;", SketchKindHandle::DDSketch, &cfg); - let new = compute_sid( - "m", - "zone=z0;", - &AggKind::Sketch { - kind: SketchKindHandle::DDSketch, - config: cfg, - }, - ); - assert_eq!(legacy, new); - } + // `compute_sketch_sid_matches_new_compute_sid_for_sketch_branch` + // removed — it tested parity between two hash wrappers, and the + // outer wrapper is now gone. The remaining `compute_sid_*` tests + // exercise the encoding properties that PR-4 will lean on when it + // migrates the precompute path to the resolver. } // 2026-05 reorg: generic epoch-partitioned columnar storage lives