diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 78baaedc..0fe768b2 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -500,8 +500,19 @@ async fn route_otlp_to_precompute( let snap = ingest_state.config_snapshot(); let agg_configs = snap.get_all_aggregation_configs(); // Reconcile schema registry against the snapshot — Phase 2a of - // the sketch DB design (`docs/design-sketch-db.md` §6). + // the sketch DB design (`docs/design-sketch-db.md` §6). Kept + // alongside the new sid-level reconcile below until the schema + // module is fully retired (schema retirement #5): both registries + // run in parallel so the §6.3 ingest barrier on + // `ingest_state.schemas.is_writable(agg_id)` below still sees + // accurate `Active/Retired/Expired` transitions while the sid + // catalog gets the same transitions in its own lifecycle fields. let _ = ingest_state.schemas.reconcile(&snap); + let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config( + ingest_state.sketch_index.as_ref(), + &snap, + ingest_state.schemas.retirement_retention(), + ); // Build (agg_id, group_key) → Vec<(series_key, ts_ms, value)> for raw points. type GroupKey = (u64, String); @@ -677,8 +688,15 @@ async fn route_modified_otlp_sketches_to_precompute( let snap = ingest_state.config_snapshot(); let agg_configs = snap.get_all_aggregation_configs(); // Reconcile schema registry against the snapshot — Phase 2a of - // the sketch DB design (`docs/design-sketch-db.md` §6). + // the sketch DB design (`docs/design-sketch-db.md` §6). Schema + // retirement #4 adds the sid-level reconcile alongside; see the + // raw-OTLP path for the rationale on running both until #5. let _ = ingest_state.schemas.reconcile(&snap); + let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config( + ingest_state.sketch_index.as_ref(), + &snap, + ingest_state.schemas.retirement_retention(), + ); let mut messages: Vec = Vec::new(); let mut routed = 0usize; let mut decoded_failed = 0usize; diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 5d662a6b..b77d5e09 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -4652,12 +4652,22 @@ async fn handle_post_streaming_config( // schemas and retires removed agg_ids (scheduling their data for // expiry after the retirement retention). // + // Schema retirement #4 wires the sid-level reconcile alongside so + // the sid catalog mirrors the same Active/Retired transitions. The + // schema half goes away when retirement #5 deletes the + // `SchemaRegistry`. + // // If `schemas` isn't attached (tests, legacy deployments), the // per-batch reconcile in IngestState still handles it — just // with up to one batch worth of latency. let (schema_added, schema_retired) = if let Some(schemas) = &state.schemas { let snap = handle.snapshot(); let summary = schemas.reconcile(snap.as_ref()); + let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_from_streaming_config( + state.sketch_index.as_ref(), + snap.as_ref(), + schemas.retirement_retention(), + ); (summary.added, summary.retired) } else { (Vec::new(), Vec::new()) diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/mod.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/mod.rs index 45b73c19..4af3369b 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/mod.rs @@ -15,13 +15,15 @@ //! `remove_instances_for_agg_config`. Schedule-driven half of the //! "controller dropped a config → its sids go away" flow. //! -//! Future content: -//! - `reconcile_from_streaming_config` — once schema/ is retired, the -//! "controller-reconcile-driven bulk-retire" entrypoint moves here -//! and operates on sids directly. +//! - [`reconcile::reconcile_from_streaming_config`] — sid-level mirror +//! of [`SchemaRegistry::reconcile`](crate::storage_engines::sketch_db::schema::SchemaRegistry::reconcile), +//! landed in schema-retirement #4 alongside the otel.rs reconcile +//! call-site migration. pub mod eviction; +pub mod reconcile; pub use eviction::{ warn_if_retention_inverted, SchemaEvictionConfig, SchemaEvictionHandle, SchemaEvictionService, }; +pub use reconcile::{reconcile_from_streaming_config, SidReconcileSummary}; diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs new file mode 100644 index 00000000..ca81d9a6 --- /dev/null +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs @@ -0,0 +1,302 @@ +//! Sid-level reconcile against a fresh `StreamingConfig` snapshot — +//! schema retirement #4. +//! +//! Mirrors the "retire orphans" half of the legacy +//! [`SchemaRegistry::reconcile`](crate::storage_engines::sketch_db::schema::SchemaRegistry::reconcile) +//! but reads / writes the sid catalog directly. The "add new ids" +//! half is implicit in the sid model: sids are minted lazily by the +//! ingest path on first write (see +//! `SketchStore::ingest_precompute_for_agg_config` and the modified-OTLP +//! sketch path), so there is nothing to pre-register up front. +//! +//! ## Signature comparison +//! +//! Each sid carries a content signature `(metric_name, agg_kind, +//! group_by_keys)` derived from the agg-config + attrs at ingest time. +//! Each agg-config in the new `StreamingConfig` likewise canonicalizes +//! to a signature. A sid is "orphaned" when its signature does not +//! match any agg-config in the new config — at that point ingest can +//! no longer route to it, and it must be retired so the eviction +//! sweep can later drop it. +//! +//! Sketch-typed agg-configs are not yet covered: the +//! `AggregationConfig` shape doesn't carry a `SketchKindHandle` / +//! `SketchConfig` natively (controller pushes them through a parallel +//! capability-routing channel). For now the reconciler treats every +//! agg-config as a precompute signature; sketch sids never compare +//! equal so they're never retired by this path. That matches the +//! pre-retirement behavior: `SchemaRegistry::reconcile` only retired +//! agg_ids in its own registry, which mirrored the +//! `StreamingConfig.aggregation_configs` map (also precompute-only). +//! Sketch lifecycle stays driven by the controller's eviction RPC +//! until M3 unifies the two. + +use std::collections::{BTreeSet, HashSet}; +use std::time::Duration; + +use asap_types::aggregation_config::AggregationConfig; +use asap_types::streaming_config::StreamingConfig; + +use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; +use crate::storage_engines::sketch_db::index::{SketchInstanceMetadata, SketchStore}; +use crate::storage_engines::sketch_db::schema::AggStatus; + +/// Sids the reconciler force-retired this call. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct SidReconcileSummary { + /// Sids that transitioned `Active → Retired` because their + /// content signature is no longer present in the new + /// `StreamingConfig`. Already-Retired or already-Expired sids + /// are not re-touched. + pub retired: Vec, +} + +/// Iterate every sid in `store`; force-retire any whose signature +/// `(metric_name, agg_kind, group_by_keys)` is not represented in +/// `config`. `retention` is forwarded to +/// [`SketchStore::force_retire`] so the schedule of eventual +/// `Retired → Expired` matches the pre-retirement `SchemaRegistry` +/// behavior. +pub fn reconcile_from_streaming_config( + store: &SketchStore, + config: &StreamingConfig, + retention: Duration, +) -> SidReconcileSummary { + let live_signatures = build_live_signature_set(config); + + let mut retired = Vec::new(); + for meta in store.snapshot_instances() { + if !matches!(meta.status(), AggStatus::Active) { + continue; + } + let sig = signature_from_meta(&meta); + if !live_signatures.contains(&sig) { + if store.force_retire(meta.sid, retention).is_some() { + retired.push(meta.sid); + } + } + } + SidReconcileSummary { retired } +} + +/// Canonical byte encoding of a sid's content signature, used as the +/// HashSet key. Two sids of the same signature produce identical +/// bytes; cross-signature collisions require a 64-bit hash birthday +/// — we don't compress here, we compare the full bytes. +fn signature_bytes(metric_name: &str, agg_kind: &AggKind, group_by_keys: &BTreeSet) -> Vec { + let mut buf: Vec = Vec::new(); + buf.extend_from_slice(metric_name.as_bytes()); + buf.push(0); + encode_agg_kind(agg_kind, &mut buf); + buf.push(0); + for k in group_by_keys { + buf.extend_from_slice(k.as_bytes()); + buf.push(b','); + } + buf +} + +fn signature_from_meta(meta: &SketchInstanceMetadata) -> Vec { + signature_bytes(&meta.metric_name, &meta.agg_kind, &meta.group_by_keys) +} + +fn signature_from_agg_config(cfg: &AggregationConfig) -> Vec { + let agg_kind = AggKind::Precompute { + agg_type: cfg.aggregation_type, + parameters_canonical: canonical_parameters(&cfg.parameters), + }; + let group_by_keys: BTreeSet = cfg.grouping_labels.labels.iter().cloned().collect(); + signature_bytes(&cfg.metric, &agg_kind, &group_by_keys) +} + +fn build_live_signature_set(config: &StreamingConfig) -> HashSet> { + config + .get_all_aggregation_configs() + .values() + .map(signature_from_agg_config) + .collect() +} + +fn encode_agg_kind(agg_kind: &AggKind, buf: &mut Vec) { + use crate::storage_engines::sketch_db::data::{SketchConfig, SketchKindHandle}; + match agg_kind { + AggKind::Sketch { kind, config } => { + buf.push(b'S'); + buf.push(match kind { + SketchKindHandle::DDSketch => 1, + SketchKindHandle::Kll => 2, + SketchKindHandle::Hll => 3, + SketchKindHandle::CountSketch => 4, + SketchKindHandle::CountMin => 5, + SketchKindHandle::CmsWithHeap => 6, + SketchKindHandle::CountSketchWithHeap => 7, + SketchKindHandle::Any => 0, + }); + match config { + SketchConfig::DDSketch { relative_accuracy } => { + buf.push(b'D'); + buf.extend_from_slice(&relative_accuracy.to_le_bytes()); + } + SketchConfig::Kll { k } => { + buf.push(b'K'); + buf.extend_from_slice(&k.to_le_bytes()); + } + SketchConfig::Hll { precision } => { + buf.push(b'H'); + buf.extend_from_slice(&precision.to_le_bytes()); + } + SketchConfig::CountSketch { rows, cols } => { + buf.push(b'S'); + buf.extend_from_slice(&rows.to_le_bytes()); + buf.extend_from_slice(&cols.to_le_bytes()); + } + SketchConfig::CountMin { rows, cols } => { + buf.push(b'M'); + buf.extend_from_slice(&rows.to_le_bytes()); + buf.extend_from_slice(&cols.to_le_bytes()); + } + } + } + AggKind::Precompute { + agg_type, + parameters_canonical, + } => { + buf.push(b'P'); + buf.extend_from_slice(agg_type.as_str().as_bytes()); + buf.push(b';'); + buf.extend_from_slice(parameters_canonical.as_bytes()); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + use asap_types::aggregation_config::AggregationConfig; + use asap_types::enums::{AggregationType, WindowType}; + use promql_utilities::data_model::key_by_label_names::KeyByLabelNames; + + use crate::storage_engines::sketch_db::data::AggKind; + use crate::storage_engines::sketch_db::index::{SketchInstanceMetadata, SketchStore}; + + fn agg_config(metric: &str, agg_type: AggregationType, group_by: Vec<&str>) -> AggregationConfig { + AggregationConfig::new( + 0, + agg_type, + String::new(), + HashMap::new(), + KeyByLabelNames::new(group_by.into_iter().map(|s| s.to_string()).collect()), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 1, + 1, + WindowType::Tumbling, + String::new(), + metric.to_string(), + None, + None, + None, + ) + } + + fn meta( + sid: u64, + metric: &str, + agg_type: AggregationType, + group_by: Vec<&str>, + ) -> SketchInstanceMetadata { + let group_by_keys: BTreeSet = + group_by.into_iter().map(|s| s.to_string()).collect(); + SketchInstanceMetadata { + sid, + metric_name: metric.to_string(), + group_by_keys, + capability: None, + agg_kind: AggKind::Precompute { + agg_type, + parameters_canonical: String::new(), + }, + accuracy: None, + first_seen_unix_ms: 0, + retired_at_ms: None, + expires_at_ms: None, + } + } + + fn streaming(configs: Vec) -> StreamingConfig { + let mut map = HashMap::new(); + for (i, c) in configs.into_iter().enumerate() { + map.insert(i as u64 + 1, c); + } + StreamingConfig::new(map) + } + + #[test] + fn empty_config_retires_every_active_sid() { + let store = SketchStore::new(); + store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); + store.register(meta(2, "mem", AggregationType::Sum, vec!["host"])); + let summary = + reconcile_from_streaming_config(&store, &streaming(vec![]), Duration::from_secs(60)); + let mut retired = summary.retired.clone(); + retired.sort(); + assert_eq!(retired, vec![1, 2]); + } + + #[test] + fn config_with_matching_signature_keeps_sid_active() { + let store = SketchStore::new(); + store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); + let cfg = streaming(vec![agg_config("cpu", AggregationType::Sum, vec!["host"])]); + let summary = reconcile_from_streaming_config(&store, &cfg, Duration::from_secs(60)); + assert!(summary.retired.is_empty()); + } + + #[test] + fn config_with_different_signature_retires_orphan() { + let store = SketchStore::new(); + // sid 1 is Sum-by-host; only Sum-by-region is in the new config. + store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); + let cfg = streaming(vec![agg_config("cpu", AggregationType::Sum, vec!["region"])]); + let summary = reconcile_from_streaming_config(&store, &cfg, Duration::from_secs(60)); + assert_eq!(summary.retired, vec![1]); + } + + #[test] + fn already_retired_sid_is_not_re_retired() { + let store = SketchStore::new(); + store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); + // First reconcile with empty config retires it. + let _ = reconcile_from_streaming_config(&store, &streaming(vec![]), Duration::from_secs(60)); + // Second reconcile reports nothing new. + let summary = reconcile_from_streaming_config( + &store, + &streaming(vec![]), + Duration::from_secs(60), + ); + assert!(summary.retired.is_empty()); + } + + #[test] + fn distinct_metrics_do_not_cross_match() { + let store = SketchStore::new(); + store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); + // Same agg_kind + grouping, different metric — sid 1 must + // still retire because its signature includes the metric. + let cfg = streaming(vec![agg_config("mem", AggregationType::Sum, vec!["host"])]); + let summary = reconcile_from_streaming_config(&store, &cfg, Duration::from_secs(60)); + assert_eq!(summary.retired, vec![1]); + } + + #[test] + fn different_agg_type_does_not_match() { + let store = SketchStore::new(); + store.register(meta(1, "cpu", AggregationType::Sum, vec!["host"])); + let cfg = streaming(vec![agg_config("cpu", AggregationType::Increase, vec!["host"])]); + let summary = reconcile_from_streaming_config(&store, &cfg, Duration::from_secs(60)); + assert_eq!(summary.retired, vec![1]); + } +}