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 6820ce9a..30d0c0af 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -1667,6 +1667,23 @@ impl SketchStore { "SketchStore persistence recovery complete" ); + // Re-register every disk-resident sid from the metadata sidecar so + // the query path can find and serve recovered series. Without this + // a freshly-reopened store recovers the parts manifest/cache but + // has an EMPTY `instances` registry (registration only happens on + // the live ingest path), so `instances_matching` enumerates nothing + // for the recovered metrics and `query_range`'s disk-union + // early-returns on the missing `sid_group_by_keys` → "No result" + // cluster-wide even though the data is durable on disk. Idempotent: + // sids already registered (e.g. by an in-flight DataPoint) are kept. + let recovered = self.register_recovered_disk_series(&cfg.disk_path); + if recovered > 0 { + tracing::info!( + recovered_sids = recovered, + "SketchStore: re-registered disk-resident sids from metadata sidecar" + ); + } + let manifest = Arc::new(Manifest::open_or_init(&cfg.disk_path)?); let parts_root = crate::storage_engines::sketch_db::index::persistence::flusher::parts_root( &cfg.disk_path, @@ -1696,6 +1713,66 @@ impl SketchStore { }) } + /// Replay the per-sid metadata sidecar at `disk_path` and register + /// each disk-resident sid as a queryable instance, UNLESS the sid is + /// already registered (a live DataPoint won the race — its in-memory + /// metadata is authoritative, so we don't clobber it). Returns the + /// number of sids freshly registered from disk. + /// + /// `capability` / `accuracy` are re-derived from the persisted + /// `agg_kind` exactly as the ingest path derives them. The sidecar is + /// missing only for parts written before this feature landed (or a + /// fresh dir) — those sids stay invisible until a live DataPoint + /// re-registers them, the same as pre-fix behavior. + pub fn register_recovered_disk_series(&self, disk_path: &std::path::Path) -> usize { + use crate::storage_engines::sketch_db::index::persistence::metadata::SidMetadataStore; + + let store = SidMetadataStore::new(disk_path); + let records = match store.load() { + Ok(r) => r, + Err(e) => { + tracing::warn!(error = %e, "failed to load sid metadata sidecar on recovery"); + return 0; + } + }; + + let mut registered = 0usize; + for rec in records { + // Don't clobber a live-registered instance. + if self.instance(rec.sid).is_some() { + continue; + } + let Some(agg_kind) = rec.agg_kind() else { + tracing::warn!( + sid = rec.sid, + "skipping recovered sid: unrecognized agg_kind in sidecar" + ); + continue; + }; + let capability = rec.capability(); + let accuracy = rec.accuracy(); + self.register(SketchInstanceMetadata { + sid: rec.sid, + metric_name: rec.metric_name, + group_by_keys: rec.group_by_keys.into_iter().collect(), + capability, + agg_kind, + accuracy, + first_seen_unix_ms: rec.first_seen_unix_ms, + retired_at_ms: None, + expires_at_ms: None, + // The sidecar doesn't carry the policy fingerprint; the + // recovered sid is reachable through the + // `instances_matching(metric, gbk)` walk regardless (the + // policy_fp reverse index is an optimization, not a + // correctness requirement for the query path). + policy_fp: PolicyFingerprint::UNSET, + }); + registered += 1; + } + registered + } + /// Switch the store into durable-tier mode: install the read handle /// the query path uses to consult disk parts, set the per-sid seal /// cadence, and retro-fit any already-created `SidStoreData` so they @@ -1757,6 +1834,23 @@ impl crate::storage_engines::sketch_db::index::persistence::EpochSource for Sket out } + fn instance_metadata_for_persist( + &self, + sid: u64, + ) -> Option { + let g = self.instances.read().ok()?; + let m = g.get(&sid)?; + Some( + crate::storage_engines::sketch_db::index::persistence::metadata::SidMetaRecord::new( + m.sid, + m.metric_name.clone(), + m.group_by_keys.iter().cloned().collect(), + &m.agg_kind, + m.first_seen_unix_ms, + ), + ) + } + fn snapshot_sealed_epoch( &self, sid: u64, @@ -2865,6 +2959,212 @@ mod tests { drop(p2); } + // ── query-from-recovered-disk (fix/query-from-recovered-disk) ─────── + // + // The #330 tests `restart_recovery_makes_flushed_data_queryable` and + // `live_aged_unsealed_panes_flush_and_survive_restart` both call + // `idx2.register(...)` on the FRESH store BEFORE querying (they even + // comment "here we re-register to model that"). That masks the real + // restart bug: in production NOBODY calls `SketchStore::register` on + // restart — registration only happens on the LIVE INGEST path when a + // fresh DataPoint arrives. The SeriesIdResolver WAL recovers + // `(metric, attrs_fp, agg_kind) → sid` but does NOT push identities + // into the SketchStore's `instances` registry. So after a true restart + // the `instances` map is EMPTY for the recovered metrics: + // * `instances_matching(metric, gbk)` enumerates nothing → the engine + // returns "No result" before reading any window, and + // * even if a sid were enumerated, `query_range`'s `union_disk_parts_into` + // early-returns on the missing `sid_group_by_keys(sid)`. + // → the cluster-wide "No result" + collapsed-SketchStore symptom. + // + // These two tests do a GENUINE fresh reopen (no `register`) for BOTH + // the sketch (KLL quantile) and exact-agg (Sum) shapes. On origin/main + // they FAIL ("No result"); with the metadata-sidecar fix they pass + // because recovery re-registers the disk-resident sids. + + fn meta_kll_host(sid: u64) -> SketchInstanceMetadata { + let cfg = SketchConfig::Kll { k: 200 }; + SketchInstanceMetadata { + sid, + metric_name: "http_latency".into(), + group_by_keys: ["host".to_string()].into_iter().collect(), + capability: Some(Capability::QuantileApprox(SketchKindHandle::Kll)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::Kll, + config: cfg.clone(), + spatial_filter_canonical: String::new(), + }, + accuracy: Some(AccuracyBound::from_config(&cfg)), + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + policy_fp: asap_types::PolicyFingerprint::UNSET, + } + } + + #[test] + fn recovered_sketch_series_queryable_after_fresh_reopen_without_register() { + let tmp = tempfile::TempDir::new().unwrap(); + let disk = tmp.path().to_path_buf(); + // ---- session 1: ingest → seal → flush → EVICT (disk-only) ---- + { + let idx = Arc::new(SketchStore::new()); + idx.register(meta_kll_host(7100)); + let p = idx.start_persistence(durable_cfg(disk.clone())).unwrap(); + for i in 0..10u64 { + let s = i * 30_000; + idx.append_sample(7100, lv_host("a"), (s, s + 30_000), sample((i + 1) as u8)); + } + assert!( + wait_until( + || !p.manifest.live_parts().is_empty() + && idx.approx_memory_bytes() == 0 + && idx.list_sealed_epochs_len() == 0, + std::time::Duration::from_secs(5), + ), + "data never flushed+evicted before restart" + ); + let mut p = p; + p.shutdown(); + } + + // ---- session 2: TRUE fresh reopen — NO register() ---- + let idx2 = Arc::new(SketchStore::new()); + // Sanity: before recovery the registry is empty (mirrors prod). + assert_eq!(idx2.instance_count(), 0, "precondition: empty registry"); + let p2 = idx2.start_persistence(durable_cfg(disk.clone())).unwrap(); + assert!( + !p2.manifest.live_parts().is_empty(), + "recovery did not reload any parts" + ); + + // (a) instances_matching must find the recovered series WITHOUT a + // re-register. On origin/main this is empty → "No result". + let gbk = ["host".to_string()].into_iter().collect(); + let sids = idx2.instances_matching("http_latency", &gbk); + assert_eq!( + sids, + vec![7100], + "instances_matching blind to disk-only series after fresh reopen \ + (registry={})", + idx2.instance_count(), + ); + // The recovered sid's metadata must be query-routable. + let meta = idx2.instance(7100).expect("recovered sid metadata present"); + assert_eq!(meta.metric_name, "http_latency"); + assert!(matches!( + meta.capability, + Some(Capability::QuantileApprox(SketchKindHandle::Kll)) + )); + + // (b) a range query over the EVICTED window returns the data. + let series = idx2.query_range(7100, 0, 150_000); + assert_eq!(series.len(), 1, "recovered KLL series not queryable"); + let s = &series[0]; + assert_eq!( + s.series_label_values, + lv_host("a"), + "label map rebuilt from recovered group_by_keys + disk values" + ); + assert!( + s.samples.contains_key(&30_000), + "recovered disk window (0,30000) missing: {:?}", + s.samples.keys().collect::>() + ); + drop(p2); + } + + #[test] + fn recovered_exact_agg_series_queryable_after_fresh_reopen_without_register() { + use crate::storage_engines::types::AggregationType; + let tmp = tempfile::TempDir::new().unwrap(); + let disk = tmp.path().to_path_buf(); + + let lv_zone = |v: &str| { + let mut x = BTreeMap::new(); + x.insert("zone".to_string(), v.to_string()); + x + }; + + // ---- session 1: ExactAgg(Sum) by (zone), flush+evict ---- + { + let idx = Arc::new(SketchStore::new()); + let mut m = meta(8100); + m.metric_name = "http_requests_total".into(); + m.group_by_keys = ["zone".to_string()].into_iter().collect(); + m.capability = Some(Capability::ExactAgg(AggregationType::Sum)); + m.agg_kind = AggKind::ExactAgg { + agg_type: AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }; + m.accuracy = None; + idx.register(m); + let p = idx.start_persistence(durable_cfg(disk.clone())).unwrap(); + for i in 0..10u64 { + let s = i * 30_000; + idx.append_precompute( + 8100, + lv_zone("z0"), + (s, s + 30_000), + Box::new( + crate::precompute_engine::operators::SumAccumulator::with_sum( + (i + 1) as f64, + ), + ), + ); + } + assert!( + wait_until( + || !p.manifest.live_parts().is_empty() + && idx.approx_memory_bytes() == 0 + && idx.list_sealed_epochs_len() == 0, + std::time::Duration::from_secs(5), + ), + "exact-agg windows never flushed+evicted" + ); + let mut p = p; + p.shutdown(); + } + + // ---- session 2: TRUE fresh reopen — NO register() ---- + let idx2 = Arc::new(SketchStore::new()); + assert_eq!(idx2.instance_count(), 0, "precondition: empty registry"); + let p2 = idx2.start_persistence(durable_cfg(disk.clone())).unwrap(); + + // instances_matching must surface the exact-agg sid. + let gbk = ["zone".to_string()].into_iter().collect(); + assert_eq!( + idx2.instances_matching("http_requests_total", &gbk), + vec![8100], + "exact-agg sid blind to instances_matching after fresh reopen" + ); + let meta = idx2.instance(8100).expect("recovered exact-agg metadata"); + assert!(matches!( + meta.agg_kind, + AggKind::ExactAgg { agg_type: AggregationType::Sum, .. } + )); + + // The Sum exact-agg range query must resolve from disk. + let series = idx2.query_exact_agg_range(8100, 0, 150_000); + assert!( + !series.is_empty(), + "recovered exact-agg query returned No result after fresh reopen" + ); + let (label, samples) = &series[0]; + assert_eq!(label.get("zone").map(String::as_str), Some("z0")); + assert!( + samples.contains_key(&30_000), + "recovered exact-agg window (0,30000) missing from disk read-back" + ); + // Rate divisor helper must also see the recovered disk windows. + assert!( + idx2.exact_agg_coverage_bounds(8100, 0, 150_000).is_some(), + "exact_agg_coverage_bounds blind to recovered disk after fresh reopen" + ); + drop(p2); + } + #[test] fn persistence_disabled_keeps_327_retention_behavior() { // Non-regression: with persistence OFF, query_range reads diff --git a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs index 6963168f..299522ba 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs @@ -39,6 +39,10 @@ pub struct FlusherHandle { pub(crate) struct FlusherShared { pub cfg: SketchStorePersistenceConfig, pub manifest: Arc, + /// Per-sid metadata sidecar — upserted on every flush so recovery can + /// re-register disk-resident sids as queryable instances. See + /// [`super::metadata`]. + pub sid_metadata: super::metadata::SidMetadataStore, pub next_part_id: AtomicU64, pub shutdown: AtomicBool, /// Woken by the insert path when it hits `hard_cap_bytes` and by @@ -75,9 +79,12 @@ impl FlusherHandle { .map(|m| m + 1) .unwrap_or(1); + let sid_metadata = super::metadata::SidMetadataStore::new(&cfg.disk_path); + let shared = Arc::new(FlusherShared { cfg: cfg.clone(), manifest: manifest.clone(), + sid_metadata, next_part_id: AtomicU64::new(next_id), shutdown: AtomicBool::new(false), pressure_cv: Condvar::new(), @@ -358,6 +365,32 @@ fn run_tick(shared: &Arc, source: &S) -> PersistR size_bytes: report.data_len + report.index_len + size_bytes_estimate, })?; + // Persist the per-sid metadata sidecar for every sid this part + // makes durable, BEFORE eviction. The on-disk part carries only + // label VALUES + sketch_type_name; the sidecar carries the + // metric name, group-by KEYS, and structured `AggKind` that the + // query path's `instances_matching` / disk-union need to serve + // the series after a restart. Done before evict so the metadata + // is durable whenever the part it describes is. A sidecar write + // failure must NOT abort the flush (the part is already durable) + // — log and continue; recovery degrades to live-ingest re-register. + let sid_meta: Vec = { + let mut seen = std::collections::HashSet::new(); + snapshots + .iter() + .filter(|s| seen.insert(s.agg_id)) + .filter_map(|s| source.instance_metadata_for_persist(s.agg_id)) + .collect() + }; + if let Err(e) = shared.sid_metadata.upsert_all(&sid_meta) { + warn!( + part_id, + error = %e, + "flusher: sid-metadata sidecar upsert failed; recovered series \ + for these sids may need a live DataPoint to become queryable" + ); + } + // Now that the part is durable and referenced, evict the // source epochs. for s in &snapshots { diff --git a/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs new file mode 100644 index 00000000..464bd9c9 --- /dev/null +++ b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs @@ -0,0 +1,497 @@ +//! Per-sid metadata sidecar for the durable warm-sketch tier. +//! +//! ## Why this exists +//! +//! On-disk parts (`part.rs`) store, per entry, only: the `sid` (as +//! `agg_id`), the label *values* (`KeyByLabelValues`), the +//! `sketch_type_name`, the encoding tag, the time bounds, and the +//! opaque sketch bytes. They do NOT carry the pieces of +//! [`SketchInstanceMetadata`](crate::storage_engines::sketch_db::index::SketchInstanceMetadata) +//! that the QUERY path needs to find and serve a series: +//! +//! * `metric_name` — the analyzer's +//! [`instances_matching`](crate::storage_engines::sketch_db::index::SketchStore::instances_matching) +//! filters on it. +//! * `group_by_keys` (label KEYS, not values) — needed both by +//! `instances_matching` AND by the disk-union read path +//! (`sid_group_by_keys` → `rebuild_label_map`), which zips the sorted +//! keys against the stored value vector to reconstruct the label map. +//! * `capability` / `agg_kind` — the sketch reducer's +//! `require_capability` gate. +//! +//! Without these, a `SketchStore` that is freshly reopened after a +//! restart recovers the parts manifest + part cache but registers NO +//! sids in its in-memory `instances` map (registration only ever happens +//! on the live ingest path, when a fresh DataPoint arrives). So +//! `instances_matching` enumerates nothing for the recovered metrics and +//! `query_range`'s disk-union early-returns on the missing +//! `sid_group_by_keys` → the query returns "No result" cluster-wide even +//! though the data is durable on disk. +//! +//! This sidecar closes that gap: the flusher upserts a compact, +//! self-describing record per flushed sid into `sid_metadata.json`, and +//! recovery replays it to re-register every disk-resident sid as a +//! queryable instance. +//! +//! ## Format +//! +//! A single JSON object `{ "": SidMetaRecord, ... }` written +//! atomically (tmp + rename) on every upsert. JSON (not the custom +//! binary part format) because the record count equals live sid +//! cardinality (small) and the schema is human-inspectable for +//! diagnosis. The record stores serializable PRIMITIVES — the +//! `Capability` / `AccuracyBound` are DERIVED on load from `agg_kind` +//! exactly as the ingest path derives them, so this module needs no +//! serde on the control-plane `Capability` / `SketchKindHandle` enums. + +use std::collections::HashMap; +use std::fs::{self, File, OpenOptions}; +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +use crate::storage_engines::sketch_db::data::{ + AccuracyBound, AggKind, Capability, SketchConfig, SketchKindHandle, +}; +use crate::storage_engines::types::AggregationType; + +use super::{PersistError, PersistResult}; + +/// File name of the sid-metadata sidecar under the persistence dir. +pub const SID_METADATA_FILE: &str = "sid_metadata.json"; + +/// Serializable mirror of [`SketchConfig`]. Kept local (rather than +/// deriving serde on the control-plane `SketchConfig`) so the sidecar +/// schema is owned by the persistence layer and changes here can't +/// silently shift the on-disk format from an unrelated edit. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "type", rename_all = "snake_case")] +enum SketchConfigRec { + DdSketch { relative_accuracy: f64 }, + Kll { k: u32 }, + Hll { precision: u32 }, + CountSketch { rows: i32, cols: i32 }, + CountMin { rows: i32, cols: i32 }, +} + +impl From<&SketchConfig> for SketchConfigRec { + fn from(c: &SketchConfig) -> Self { + match c { + SketchConfig::DDSketch { relative_accuracy } => SketchConfigRec::DdSketch { + relative_accuracy: *relative_accuracy, + }, + SketchConfig::Kll { k } => SketchConfigRec::Kll { k: *k }, + SketchConfig::Hll { precision } => SketchConfigRec::Hll { + precision: *precision, + }, + SketchConfig::CountSketch { rows, cols } => SketchConfigRec::CountSketch { + rows: *rows, + cols: *cols, + }, + SketchConfig::CountMin { rows, cols } => SketchConfigRec::CountMin { + rows: *rows, + cols: *cols, + }, + } + } +} + +impl From<&SketchConfigRec> for SketchConfig { + fn from(c: &SketchConfigRec) -> Self { + match c { + SketchConfigRec::DdSketch { relative_accuracy } => SketchConfig::DDSketch { + relative_accuracy: *relative_accuracy, + }, + SketchConfigRec::Kll { k } => SketchConfig::Kll { k: *k }, + SketchConfigRec::Hll { precision } => SketchConfig::Hll { + precision: *precision, + }, + SketchConfigRec::CountSketch { rows, cols } => SketchConfig::CountSketch { + rows: *rows, + cols: *cols, + }, + SketchConfigRec::CountMin { rows, cols } => SketchConfig::CountMin { + rows: *rows, + cols: *cols, + }, + } + } +} + +/// Stable string form of a [`SketchKindHandle`] for the sidecar. Mirrors +/// `sketch_kind_canonical` but is owned by the persistence layer so the +/// on-disk vocabulary is stable independent of any upstream rename. +fn sketch_kind_to_str(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", + SketchKindHandle::Any => "Any", + } +} + +fn sketch_kind_from_str(s: &str) -> Option { + Some(match s { + "DDSketch" => SketchKindHandle::DDSketch, + "Kll" => SketchKindHandle::Kll, + "Hll" => SketchKindHandle::Hll, + "CountSketch" => SketchKindHandle::CountSketch, + "CountMin" => SketchKindHandle::CountMin, + "CmsWithHeap" => SketchKindHandle::CmsWithHeap, + "CountSketchWithHeap" => SketchKindHandle::CountSketchWithHeap, + "Any" => SketchKindHandle::Any, + _ => return None, + }) +} + +/// Serializable mirror of the two [`AggKind`] branches. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum AggKindRec { + Sketch { + sketch_kind: String, + config: SketchConfigRec, + spatial_filter_canonical: String, + }, + ExactAgg { + agg_type: AggregationType, + parameters_canonical: String, + spatial_filter_canonical: String, + }, +} + +impl From<&AggKind> for AggKindRec { + fn from(a: &AggKind) -> Self { + match a { + AggKind::Sketch { + kind, + config, + spatial_filter_canonical, + } => AggKindRec::Sketch { + sketch_kind: sketch_kind_to_str(*kind).to_string(), + config: config.into(), + spatial_filter_canonical: spatial_filter_canonical.clone(), + }, + AggKind::ExactAgg { + agg_type, + parameters_canonical, + spatial_filter_canonical, + } => AggKindRec::ExactAgg { + agg_type: *agg_type, + parameters_canonical: parameters_canonical.clone(), + spatial_filter_canonical: spatial_filter_canonical.clone(), + }, + } + } +} + +impl AggKindRec { + /// Rebuild the structured [`AggKind`]. Returns `None` if a sketch + /// kind string is unrecognized (forward-compat: a newer writer added + /// a kind this reader doesn't know — skip the sid rather than panic). + fn to_agg_kind(&self) -> Option { + Some(match self { + AggKindRec::Sketch { + sketch_kind, + config, + spatial_filter_canonical, + } => AggKind::Sketch { + kind: sketch_kind_from_str(sketch_kind)?, + config: config.into(), + spatial_filter_canonical: spatial_filter_canonical.clone(), + }, + AggKindRec::ExactAgg { + agg_type, + parameters_canonical, + spatial_filter_canonical, + } => AggKind::ExactAgg { + agg_type: *agg_type, + parameters_canonical: parameters_canonical.clone(), + spatial_filter_canonical: spatial_filter_canonical.clone(), + }, + }) + } +} + +/// One durable sid-metadata row. The query-critical fields +/// (`metric_name`, `group_by_keys`, `agg_kind`) are stored explicitly; +/// `capability` and `accuracy` are DERIVED from `agg_kind` on load, the +/// same way the ingest path derives them, so the record stays minimal. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct SidMetaRecord { + pub sid: u64, + pub metric_name: String, + /// Label KEY set, sorted (a `Vec` so the JSON stays compact; the + /// store side rebuilds the `BTreeSet`). + pub group_by_keys: Vec, + agg_kind: AggKindRec, + pub first_seen_unix_ms: i64, +} + +impl SidMetaRecord { + /// Build a record from the live store-side fields. `agg_kind` is the + /// structured `AggKind`; `capability`/`accuracy` are intentionally + /// NOT stored (re-derived on load). + pub fn new( + sid: u64, + metric_name: String, + group_by_keys: Vec, + agg_kind: &AggKind, + first_seen_unix_ms: i64, + ) -> Self { + Self { + sid, + metric_name, + group_by_keys, + agg_kind: agg_kind.into(), + first_seen_unix_ms, + } + } + + /// Reconstruct the structured [`AggKind`], or `None` for an + /// unrecognized sketch kind. + pub fn agg_kind(&self) -> Option { + self.agg_kind.to_agg_kind() + } + + /// Derive the warm-tier [`Capability`] from `agg_kind`, mirroring the + /// ingest path (`otel.rs`) and `ingest_precompute_with_sid`. Returns + /// `None` only when `agg_kind` itself fails to reconstruct. + pub fn capability(&self) -> Option { + let agg_kind = self.agg_kind()?; + Some(match agg_kind { + AggKind::Sketch { kind, .. } => match kind { + SketchKindHandle::DDSketch | SketchKindHandle::Kll => { + Capability::QuantileApprox(kind) + } + SketchKindHandle::Hll => Capability::CardinalityApprox, + SketchKindHandle::CountSketch | SketchKindHandle::CountMin => { + Capability::FrequencyEstimate(kind) + } + SketchKindHandle::CmsWithHeap | SketchKindHandle::CountSketchWithHeap => { + Capability::FrequencyTopk(kind) + } + // Defensive: `Any` is a control-plane wildcard that should + // never reach the index; mirror the ingest fallback. + SketchKindHandle::Any => Capability::QuantileApprox(kind), + }, + AggKind::ExactAgg { agg_type, .. } => Capability::ExactAgg(agg_type), + }) + } + + /// Derive the [`AccuracyBound`] — `Some` for sketch-backed sids, + /// `None` for exact-agg sids (mirrors the ingest registration). + pub fn accuracy(&self) -> Option { + match self.agg_kind()? { + AggKind::Sketch { config, .. } => Some(AccuracyBound::from_config(&config)), + AggKind::ExactAgg { .. } => None, + } + } +} + +/// File-backed sid-metadata sidecar. The whole map is rewritten on every +/// upsert (atomic tmp + rename). Live sid cardinality is small, so a full +/// rewrite per flush tick is cheap and keeps the on-disk file always +/// consistent with no log-replay machinery. +#[derive(Debug)] +pub struct SidMetadataStore { + path: PathBuf, +} + +impl SidMetadataStore { + /// Open (or lazily create on first write) the sidecar at + /// `/sid_metadata.json`. + pub fn new(disk_path: &Path) -> Self { + Self { + path: disk_path.join(SID_METADATA_FILE), + } + } + + pub fn path(&self) -> &Path { + &self.path + } + + /// Load every durable record. Returns an empty vec when the sidecar + /// doesn't exist yet (fresh dir, or parts written before this feature + /// landed) or when it is unparsable (treated as "no recoverable + /// metadata" — the live ingest path still re-registers on first DP). + pub fn load(&self) -> PersistResult> { + let mut f = match File::open(&self.path) { + Ok(f) => f, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(e) => return Err(PersistError::Io(e)), + }; + let mut buf = String::new(); + f.read_to_string(&mut buf)?; + if buf.trim().is_empty() { + return Ok(Vec::new()); + } + let map: HashMap = match serde_json::from_str(&buf) { + Ok(m) => m, + Err(e) => { + tracing::warn!( + path = %self.path.display(), + error = %e, + "sid metadata sidecar unparsable; ignoring (live ingest will re-register)" + ); + return Ok(Vec::new()); + } + }; + Ok(map.into_values().collect()) + } + + /// Upsert a batch of records, merging with whatever is already on + /// disk (last write wins per sid). Atomic via tmp + rename + dir + /// fsync, matching the manifest's durability discipline. + pub fn upsert_all(&self, records: &[SidMetaRecord]) -> PersistResult<()> { + if records.is_empty() { + return Ok(()); + } + let mut map: HashMap = self + .load()? + .into_iter() + .map(|r| (r.sid.to_string(), r)) + .collect(); + let mut changed = false; + for r in records { + let key = r.sid.to_string(); + match map.get(&key) { + Some(existing) if existing == r => {} + _ => { + map.insert(key, r.clone()); + changed = true; + } + } + } + if !changed { + return Ok(()); + } + let json = serde_json::to_string(&map) + .map_err(|e| PersistError::Serialize(format!("sid metadata: {e}")))?; + self.write_atomic(json.as_bytes()) + } + + fn write_atomic(&self, bytes: &[u8]) -> PersistResult<()> { + if let Some(parent) = self.path.parent() { + fs::create_dir_all(parent)?; + } + let tmp = self.path.with_extension("json.tmp"); + { + let mut f = OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(&tmp)?; + f.write_all(bytes)?; + f.sync_all()?; + } + fs::rename(&tmp, &self.path)?; + if let Some(parent) = self.path.parent() { + if let Ok(dir) = File::open(parent) { + let _ = dir.sync_all(); + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn sketch_meta(sid: u64) -> SidMetaRecord { + SidMetaRecord::new( + sid, + "http_latency".into(), + vec!["host".into(), "zone".into()], + &AggKind::Sketch { + kind: SketchKindHandle::Kll, + config: SketchConfig::Kll { k: 200 }, + spatial_filter_canonical: String::new(), + }, + 1234, + ) + } + + fn exact_meta(sid: u64) -> SidMetaRecord { + SidMetaRecord::new( + sid, + "http_requests_total".into(), + vec!["zone".into()], + &AggKind::ExactAgg { + agg_type: AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }, + 5678, + ) + } + + #[test] + fn load_on_missing_file_is_empty() { + let tmp = TempDir::new().unwrap(); + let s = SidMetadataStore::new(tmp.path()); + assert!(s.load().unwrap().is_empty()); + } + + #[test] + fn upsert_then_load_round_trips() { + let tmp = TempDir::new().unwrap(); + let s = SidMetadataStore::new(tmp.path()); + s.upsert_all(&[sketch_meta(1), exact_meta(2)]).unwrap(); + + let mut got = s.load().unwrap(); + got.sort_by_key(|r| r.sid); + assert_eq!(got.len(), 2); + assert_eq!(got[0], sketch_meta(1)); + assert_eq!(got[1], exact_meta(2)); + } + + #[test] + fn upsert_merges_and_overwrites_per_sid() { + let tmp = TempDir::new().unwrap(); + let s = SidMetadataStore::new(tmp.path()); + s.upsert_all(&[sketch_meta(1)]).unwrap(); + // New sid + updated metric for sid 1. + let mut updated = sketch_meta(1); + updated.metric_name = "http_latency_v2".into(); + s.upsert_all(&[updated.clone(), exact_meta(2)]).unwrap(); + + let mut got = s.load().unwrap(); + got.sort_by_key(|r| r.sid); + assert_eq!(got.len(), 2); + assert_eq!(got[0].metric_name, "http_latency_v2"); + assert_eq!(got[1], exact_meta(2)); + } + + #[test] + fn derives_capability_and_accuracy_from_agg_kind() { + let kll = sketch_meta(1); + assert!(matches!( + kll.capability(), + Some(Capability::QuantileApprox(SketchKindHandle::Kll)) + )); + assert!(kll.accuracy().is_some()); + + let sum = exact_meta(2); + assert!(matches!( + sum.capability(), + Some(Capability::ExactAgg(AggregationType::Sum)) + )); + assert!(sum.accuracy().is_none()); + } + + #[test] + fn unparsable_file_loads_as_empty() { + let tmp = TempDir::new().unwrap(); + let s = SidMetadataStore::new(tmp.path()); + std::fs::write(s.path(), b"{not json").unwrap(); + assert!(s.load().unwrap().is_empty()); + } +} diff --git a/data_plane/src/storage_engines/sketch_db/persistence/mod.rs b/data_plane/src/storage_engines/sketch_db/persistence/mod.rs index a0381142..83613fa8 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/mod.rs @@ -21,6 +21,7 @@ pub mod config; pub mod manifest; +pub mod metadata; pub mod part; pub mod source; @@ -30,6 +31,7 @@ pub mod recovery; pub use config::SketchStorePersistenceConfig; pub use manifest::{Manifest, PartEntry}; +pub use metadata::{SidMetaRecord, SidMetadataStore}; pub use part::{PartId, PartReader, PartWriter, SnapshotEntry}; pub use source::{EpochSource, SealedEpochRef}; diff --git a/data_plane/src/storage_engines/sketch_db/persistence/source.rs b/data_plane/src/storage_engines/sketch_db/persistence/source.rs index edb304fd..8242a04e 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/source.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/source.rs @@ -125,4 +125,22 @@ pub trait EpochSource: Send + Sync { fn evict_sealed_epoch(&self, agg_id: u64, epoch_id: u64); fn approx_memory_bytes(&self) -> usize; + + /// Persistable instance metadata for `sid` — the pieces of the store's + /// `SketchInstanceMetadata` the QUERY path needs but the on-disk part + /// format does NOT carry (metric name, group-by KEYS, structured + /// `AggKind`). Called by the flusher right before it makes a part + /// durable so recovery can re-register the sid as a queryable instance + /// after a restart (otherwise `instances_matching` enumerates nothing + /// for disk-only series → "No result" cluster-wide). + /// + /// Returns `None` when the sid is unknown to the source (e.g. a test + /// fake, or a sid whose instance metadata was evicted). Default impl + /// returns `None` so existing/test sources need not implement it. + fn instance_metadata_for_persist( + &self, + _sid: u64, + ) -> Option { + None + } }