From d51cab835b069b49085fdf85249d9e979c370942 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 28 May 2026 09:44:48 -0600 Subject: [PATCH 1/4] CMS per-item estimate(key): reducer estimate path + engine item-key gate sketch_reducer: decode_frequency_estimate (calls CountMinSketch/CountSketch::estimate(key); heap path rebuilds via from_legacy_matrix) + thread item_key through evaluate_for_capability/evaluate_core; FrequencyEstimate branch dispatches estimate(key) vs the per-window bucket total. engine + index: sid->item_label side-table (set_item_label/item_label_for); engine extract_filter_value resolves the item value from a hit sid's registered item_label and lifts the keyed-frequency safe-miss only for item_label-mode sids. Tests: cms_per_item_estimate, extract_filter_value. Co-Authored-By: Claude Opus 4.7 --- .../query_engines/asap_query_engine/engine.rs | 73 +++++++++++++ .../storage_engines/sketch_db/index/mod.rs | 32 ++++++ .../sketch_db/query/sketch_reducer.rs | 103 +++++++++++++++++- .../storage_engines/sketch_db/query/tests.rs | 85 +++++++++++++++ 4 files changed, 289 insertions(+), 4 deletions(-) diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 177d5f1a..e8add4b3 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -864,6 +864,12 @@ impl ASAPQueryEngine { &candidate.required_capability, &hit_sids, &candidate.function_args, + // Per-item CMS estimate(key) is wired through the reducer + // but only dispatched once the engine resolves the item + // value against an item_label-mode sid (Phase 2b). Until + // then keyed CMS frequency safe-misses (see below), so the + // bucket-total path is correct here. + None, effective_is_cumulative(candidate), start_ms, end_ms, @@ -1002,6 +1008,24 @@ fn effective_is_cumulative( ) } +/// Extract the VALUE of `label` from a canonical spatial-filter string of +/// the form `{a="1",service="svc-3"}` (the shape produced by +/// `normalize_spatial_filter`). Used by the per-item CMS `estimate(key)` +/// gate to pull the item value a keyed selector targets. Returns `None` +/// when `label` is absent. Exact label match (not substring), so +/// `service` does not match `myservice`. +fn extract_filter_value(canonical: &str, label: &str) -> Option { + let inner = canonical.trim().trim_start_matches('{').trim_end_matches('}'); + for part in inner.split(',') { + if let Some((k, v)) = part.trim().split_once('=') { + if k.trim() == label { + return Some(v.trim().trim_matches('"').to_string()); + } + } + } + None +} + /// Whether the analyzer's `outer_agg` should still be folded over the /// reducer's result, or has already been CONSUMED by the /// capability dispatch. @@ -1523,7 +1547,33 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // left alone; the bare `count_over_time(cms[r])` demo has // an empty filter and is unaffected. Full string-keyed // estimate is a larger follow-up; the safe-miss is enough. + // Phase 2b: resolve a per-item estimate key. If a hit sid is + // registered in item_label mode (item_labels side-table) and + // the candidate's spatial filter selects that exact label, the + // per-item `estimate(key)` path CAN answer the keyed selector — + // so we extract the value and DON'T safe-miss below. + let mut cms_item_key: Option = None; + if matches!( + &candidate.required_capability, + crate::storage_engines::sketch_db::index::Capability::FrequencyEstimate(_) + ) && !candidate.spatial_filter_canonical.is_empty() + { + for sid in &hit_sids { + if let Some(label) = idx.item_label_for(*sid) { + if let Some(val) = + extract_filter_value(&candidate.spatial_filter_canonical, &label) + { + cms_item_key = Some(val); + break; + } + } + } + } + if freq_rate_override.is_none() + // A resolved per-item key means the keyed estimate path + // answers this selector — skip the safe-miss. + && cms_item_key.is_none() && matches!( &candidate.required_capability, crate::storage_engines::sketch_db::index::Capability::FrequencyEstimate(_) @@ -1673,6 +1723,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu &candidate.required_capability, &hit_sids, &candidate.function_args, + cms_item_key.as_deref(), effective_is_cumulative(candidate), t0_ms, now_ms, @@ -2122,6 +2173,28 @@ mod hot_reload_phase2_tests { StreamingConfig, WindowType}; use promql_utilities::data_model::key_by_label_names::KeyByLabelNames; + #[test] + fn extract_filter_value_pulls_item_value() { + // exact-label match, single and multi-matcher canonical forms + assert_eq!( + super::extract_filter_value("{service=\"svc-000003\"}", "service"), + Some("svc-000003".to_string()) + ); + assert_eq!( + super::extract_filter_value("{zone=\"z1\",service=\"svc-000003\"}", "service"), + Some("svc-000003".to_string()) + ); + // absent label -> None + assert_eq!(super::extract_filter_value("{zone=\"z1\"}", "service"), None); + // substring labels must NOT match (service != myservice) + assert_eq!( + super::extract_filter_value("{myservice=\"x\"}", "service"), + None + ); + // empty filter -> None + assert_eq!(super::extract_filter_value("", "service"), None); + } + fn dummy_agg(_id: u64, metric: &str) -> crate::storage_engines::types::AggregationConfig { // `_id` is unused after PR 5 — identity is content-addressed. crate::storage_engines::types::AggregationConfig::new( 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 661fee06..2b5f2764 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -286,6 +286,16 @@ pub struct SketchStore { /// whose state was merged away by an upstream gateway before /// reaching this backend). instances: RwLock>, + /// sid → item_label (the data-point attribute NAME, e.g. "service" + /// or "endpoint") for CountMin/CountSketch sids registered in + /// per-item mode. Its presence is what makes a CMS sid answerable by + /// the per-item `estimate(key)` path (the query engine extracts the + /// matching selector value and gates the safe-miss on it). Absent => + /// per-attribute-set CMS (only the bucket total is meaningful). + /// Kept as a decoupled side-table so recording item_label does not + /// change sid identity (`AggKind` canonical string) or churn the many + /// `SketchInstanceMetadata` / `AggKind::Sketch` literals. + item_labels: RwLock>, /// sid → per-sid columnar storage. Empty `SidStoreData` (or absent /// key) for ghost sids — query path detects this and falls through /// to Thanos archive. @@ -431,6 +441,28 @@ impl SketchStore { metric_idx.entry(metric_name).or_default().insert(sid); } + /// Record that `sid` is a per-item (item_label-mode) frequency sketch + /// keyed by the data-point attribute `label` (e.g. "service"). The + /// query engine consults this to decide whether a keyed selector like + /// `cms_metric{service="X"}` can be answered by the per-item + /// `estimate(key)` path. A no-op `label` (empty) clears it. + pub fn set_item_label(&self, sid: u64, label: &str) { + let mut m = self.item_labels.write().unwrap(); + if label.is_empty() { + m.remove(&sid); + } else { + m.insert(sid, label.to_string()); + } + } + + /// The per-item attribute name recorded for `sid`, if any. `Some` + /// means the sketch hashes that label's VALUE (so `estimate(value)` + /// is meaningful); `None` means per-attribute-set keying (only the + /// bucket total is meaningful — keyed selectors must safe-miss). + pub fn item_label_for(&self, sid: u64) -> Option { + self.item_labels.read().unwrap().get(&sid).cloned() + } + /// Resolve a policy fingerprint to the set of sids it has minted. /// Returns an empty vector when no sid is bound to the fingerprint /// (e.g. fresh policy with no ingest activity yet) or when the diff --git a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs index 505bc57b..6aa045a3 100644 --- a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs +++ b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs @@ -343,7 +343,8 @@ impl<'a> SketchReducer<'a> { function_name, "quantile_over_time" | "count_distinct_over_time" | "topk_over_time" ); - self.evaluate_core(sids, family, is_cumulative, function_name, function_args, t0_ms, t1_ms) + // Legacy string entry: no item-key channel — always bucket-total. + self.evaluate_core(sids, family, is_cumulative, function_name, function_args, None, t0_ms, t1_ms) } /// Typed-dispatch sister of [`Self::evaluate`] (P2-4). Picks the @@ -367,6 +368,10 @@ impl<'a> SketchReducer<'a> { cap: &Capability, sids: &[u64], function_args: &[f64], + // Per-item point-estimate key for FrequencyEstimate (CMS estimate(key)). + // The engine passes `Some` only for an item_label-mode CMS candidate; + // `None` for every other capability/candidate. + item_key: Option<&str>, is_cumulative: bool, t0_ms: u64, t1_ms: u64, @@ -385,7 +390,7 @@ impl<'a> SketchReducer<'a> { QueryFamily::FrequencyTopk => "topk", QueryFamily::FrequencyEstimate => "frequency", }; - self.evaluate_core(sids, family, is_cumulative, function_label, function_args, t0_ms, t1_ms) + self.evaluate_core(sids, family, is_cumulative, function_label, function_args, item_key, t0_ms, t1_ms) } /// Shared evaluation core for the string ([`Self::evaluate`]) and @@ -400,6 +405,10 @@ impl<'a> SketchReducer<'a> { is_cumulative: bool, function_name: &str, function_args: &[f64], + // Per-item point-estimate key (the item_label VALUE, e.g. a service + // name). `Some` triggers the CMS/CountSketch `estimate(key)` path in + // the FrequencyEstimate branch; `None` keeps the bucket-total default. + item_key: Option<&str>, t0_ms: u64, t1_ms: u64, ) -> Result { @@ -436,6 +445,9 @@ impl<'a> SketchReducer<'a> { // plumbing a string-keyed `function_arg` through the reducer // entry point, which the current `&[f64]` signature can't carry. if family == QueryFamily::FrequencyEstimate { + let kind = meta + .sketch_kind() + .expect("ASAP-tier reducer only handles sketch-backed sids"); for ts in series_list { let mut samples_out: Vec<(i64, f64)> = Vec::with_capacity(ts.samples.len()); for (w_end, state) in ts.samples.iter() { @@ -447,8 +459,14 @@ impl<'a> SketchReducer<'a> { if w > cov_hi { cov_hi = w; } - let total = decode_frequency_total(sid, meta.sketch_kind().expect("ASAP-tier reducer only handles sketch-backed sids"), state)?; - samples_out.push((*w_end, total)); + // Per-item point estimate when an item key is supplied + // (and the sid is item_label-mode — gated by the engine); + // otherwise the per-window bucket TOTAL (sum of row 0). + let value = match item_key { + Some(key) => decode_frequency_estimate(sid, kind, state, key)?, + None => decode_frequency_total(sid, kind, state)?, + }; + samples_out.push((*w_end, value)); } out_series.push((ts.series_label_values, samples_out)); } @@ -1337,6 +1355,83 @@ fn decode_frequency_total( } } +/// Per-item frequency POINT estimate: decode the window's CMS / CountSketch +/// and return `estimate(key)` — the keyed analogue of +/// [`decode_frequency_total`]'s row-0 sum (which returns the bucket TOTAL). +/// +/// CMS `estimate` is min-over-rows (non-negative one-sided over-estimate); +/// CountSketch `estimate` is median-of-signed-rows, clamped to >= 0 for the +/// count-frequency surface. Only valid for an item_label-mode sid (the query +/// engine gates this; a per-attribute-set CMS would hash a different key and +/// must NOT be served here). +fn decode_frequency_estimate( + sid: u64, + sketch_kind: SketchKindHandle, + state: &SketchSampleState, + key: &str, +) -> Result { + let to_err = |e: String, encoding: SketchEncoding| ASAPTierError::DeserializeFailure { + sid, + encoding, + reason: e, + }; + match sketch_kind { + SketchKindHandle::CountMin => { + let cms = match state.encoding { + SketchEncoding::ProtoFull => decode_cms_from_proto(&state.bytes) + .map_err(|e| to_err(e, state.encoding))?, + SketchEncoding::MsgpackFull => decode_cms_from_msgpack(&state.bytes) + .map_err(|e| to_err(e, state.encoding))?, + SketchEncoding::ProtoDelta => decode_cms_from_proto_delta(&state.bytes) + .map_err(|e| to_err(e, state.encoding))?, + SketchEncoding::MsgpackDelta => { + return Err(to_err( + "CountMin (heap-less) MSGPACK_DELTA is not a valid producer encoding" + .to_string(), + state.encoding, + )); + } + }; + Ok(cms.estimate(key).max(0.0)) + } + SketchKindHandle::CountSketch => { + let cs = match state.encoding { + SketchEncoding::ProtoFull => decode_cs_from_proto(&state.bytes) + .map_err(|e| to_err(e, state.encoding))?, + SketchEncoding::MsgpackFull => decode_cs_from_msgpack(&state.bytes) + .map_err(|e| to_err(e, state.encoding))?, + SketchEncoding::ProtoDelta => decode_cs_from_proto_delta(&state.bytes) + .map_err(|e| to_err(e, state.encoding))?, + SketchEncoding::MsgpackDelta => { + return Err(to_err( + "CountSketch (heap-less) MSGPACK_DELTA is not a valid producer encoding" + .to_string(), + state.encoding, + )); + } + }; + Ok(cs.estimate(key).max(0.0)) + } + SketchKindHandle::CmsWithHeap | SketchKindHandle::CountSketchWithHeap => { + let heap = match state.encoding { + SketchEncoding::MsgpackDelta => decode_cms_with_heap_from_msgpack_delta(&state.bytes) + .map_err(|e| to_err(e, state.encoding))?, + _ => decode_cms_with_heap_from_msgpack(&state.bytes) + .map_err(|e| to_err(e, state.encoding))?, + }; + let matrix = heap.sketch_matrix(); + let rows = matrix.len(); + let cols = matrix.first().map(|r| r.len()).unwrap_or(0); + let cms = CountMinSketch::from_legacy_matrix(matrix, rows, cols); + Ok(cms.estimate(key).max(0.0)) + } + other => Err(ASAPTierError::UnsupportedCapability { + function: "frequency_estimate".to_string(), + capability: Capability::FrequencyEstimate(other), + }), + } +} + fn row0_sum_cms(cms: &CountMinSketch) -> f64 { let matrix = cms.sketch(); row0_sum_from_matrix(&matrix) diff --git a/data_plane/src/storage_engines/sketch_db/query/tests.rs b/data_plane/src/storage_engines/sketch_db/query/tests.rs index f4910e85..a6d5f6c1 100644 --- a/data_plane/src/storage_engines/sketch_db/query/tests.rs +++ b/data_plane/src/storage_engines/sketch_db/query/tests.rs @@ -462,6 +462,7 @@ fn multi_series_one_per_label_value() { // --------------------------------------------------------------------------- use asap_sketchlib::CountMinSketchWithHeap; +use asap_sketchlib::CountMinSketch; fn cms_heap_meta(sid: u64) -> SketchInstanceMetadata { let cfg = SketchConfig::CountMin { rows: 4, cols: 256 }; @@ -593,6 +594,90 @@ fn cms_without_heap_returns_missing_heap() { } } +#[test] +fn cms_per_item_estimate_returns_keyed_count() { + let idx = SketchStore::new(); + let sid = 320; + // A FrequencyEstimate-capable plain CountMin sid. + let cfg = SketchConfig::CountMin { rows: 4, cols: 256 }; + idx.register(SketchInstanceMetadata { + sid, + metric_name: "endpoint_request_freq".to_string(), + group_by_keys: BTreeSet::new(), + capability: Some(Capability::FrequencyEstimate(SketchKindHandle::CountMin)), + agg_kind: AggKind::Sketch { + kind: SketchKindHandle::CountMin, + 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, + }); + + // One CMS keyed by item value: /checkout x50, /cart x20. + let mut cms = CountMinSketch::new(4, 256); + for _ in 0..50 { + cms.update("/checkout", 1.0); + } + for _ in 0..20 { + cms.update("/cart", 1.0); + } + let bytes = cms.to_msgpack().expect("serialize cms"); + idx.append_sample(sid, BTreeMap::new(), (1000, 1010), msgpack_full(bytes)); + + let reducer = SketchReducer::new(&idx); + + // Per-item estimate path (Some key): one-sided over-estimate of the + // inserted count (50), tight band given 256 cols / 2 keys. + let keyed = reducer + .evaluate_for_capability( + &Capability::FrequencyEstimate(SketchKindHandle::CountMin), + &[sid], + &[], + Some("/checkout"), + false, + 1000, + 1010, + ) + .expect("keyed frequency estimate should succeed"); + let est = keyed + .series + .first() + .and_then(|s| s.1.first()) + .map(|s| s.1) + .expect("a keyed estimate sample"); + assert!( + (50.0..=55.0).contains(&est), + "per-item estimate(/checkout) = {est}, expected one-sided ~50" + ); + + // No key: the per-window bucket TOTAL (row-0 sum = all inserts = 70). + let total = reducer + .evaluate_for_capability( + &Capability::FrequencyEstimate(SketchKindHandle::CountMin), + &[sid], + &[], + None, + false, + 1000, + 1010, + ) + .expect("bucket total should succeed"); + let tot = total + .series + .first() + .and_then(|s| s.1.first()) + .map(|s| s.1) + .expect("a bucket-total sample"); + assert!( + (tot - 70.0).abs() <= 1.0, + "bucket total = {tot}, expected ~70 (50 + 20)" + ); +} + // --------------------------------------------------------------------------- // TODO-2 tests — delta encoding stitching. // From b9da087eefca0cd1bcabe4dee999d5697a4eaf75 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 28 May 2026 09:44:48 -0600 Subject: [PATCH 2/4] Sum AggregationType envelope ingest (SumAgg -> ExactAgg(Sum)) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit asap_otel_proto: add SumAgg/SumAggDataPoint/SumAggEncoding + Data::SumAgg (sum_agg=18, matching the collector pdata oneof tag). SumAccumulator::from_sum_bytes decodes the self-contained 16-byte Sum payload (float64 sum LE || uint64 count LE) the edge emits — Sum is an aggregation, NOT a sketch, so the payload deliberately does not use the sketchlib envelope proto. otel.rs ingest routes Data::SumAgg through SumAccumulator into the existing ExactAgg(Sum) MetricPoint path (no new SketchKindHandle; sid stays exact_agg:Sum). Cross-language golden test (Go SumWrapper bytes -> sum=100). Co-Authored-By: Claude Opus 4.7 --- .../proto/metrics/v1/metrics.proto | 34 ++++++++++++++ data_plane/src/drivers/ingest/otel.rs | 24 ++++++++++ .../operators/sum_accumulator.rs | 44 ++++++++++++++++++- 3 files changed, 101 insertions(+), 1 deletion(-) diff --git a/crates/asap_otel_proto/proto/opentelemetry/proto/metrics/v1/metrics.proto b/crates/asap_otel_proto/proto/opentelemetry/proto/metrics/v1/metrics.proto index cf5ee296..8aa6e443 100644 --- a/crates/asap_otel_proto/proto/opentelemetry/proto/metrics/v1/metrics.proto +++ b/crates/asap_otel_proto/proto/opentelemetry/proto/metrics/v1/metrics.proto @@ -213,6 +213,9 @@ message Metric { CountSketch countsketch = 15; CountMinSketch countminsketch = 16; HLLSketch hllsketch = 17; + // Scalar Sum aggregate (AggregationKind = Sum); field 18 mirrors the + // collector pdata SumAgg oneof tag. Decoded into ExactAgg(Sum). + SumAgg sum_agg = 18; } // Additional metadata attributes that describe the metric. [Optional]. @@ -298,6 +301,37 @@ message DDSketch { double relative_accuracy = 3; } +// SumAgg represents a first-class scalar Sum aggregate (AggregationKind = Sum), +// carried as a portable {sum,count} envelope in the sketch bytes. Not a sketch; +// shares the modified-OTLP metric data oneof so the backend decodes it via the +// same envelope path (into the ExactAgg(Sum) accumulator). +message SumAgg { + repeated SumAggDataPoint data_points = 1; + AggregationTemporality aggregation_temporality = 2; +} + +// SumAggDataPoint carries a scalar Sum aggregate as a {sum,count} envelope. +message SumAggDataPoint { + repeated opentelemetry.proto.common.v1.KeyValue attributes = 9; + fixed64 start_time_unix_nano = 2; + fixed64 time_unix_nano = 3; + // Serialized SumState envelope (sketchlib SketchEnvelope{sum:SumState}). + bytes sketch = 8; + SumAggEncoding encoding = 10; + repeated Exemplar exemplars = 11; + uint32 flags = 15; + uint64 series_id = 16; +} + +// SumAggEncoding identifies how the SumAgg payload bytes are encoded. +enum SumAggEncoding { + SUM_AGG_ENCODING_UNSPECIFIED = 0; + SUM_AGG_ENCODING_PROTO = 1; + SUM_AGG_ENCODING_PROTO_DELTA = 2; + SUM_AGG_ENCODING_MSGPACK = 3; + SUM_AGG_ENCODING_MSGPACK_DELTA = 4; +} + // Summary metric data are used to convey quantile summaries, // a Prometheus (see: https://prometheus.io/docs/concepts/metric_types/#summary) // and OpenMetrics (see: https://github.com/prometheus/OpenMetrics/blob/4dbf6075567ab43296eed941037c12951faafb92/protos/prometheus.proto#L45) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 253b5612..5b1cd8a1 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -2433,6 +2433,7 @@ fn otlp_to_record_count(request: &ExportMetricsServiceRequest) -> usize { Some(Data::Countsketch(c)) => count += c.data_points.len(), Some(Data::Countminsketch(c)) => count += c.data_points.len(), Some(Data::Hllsketch(h)) => count += h.data_points.len(), + Some(Data::SumAgg(sa)) => count += sa.data_points.len(), None => {} } } @@ -2522,6 +2523,29 @@ fn otlp_to_metric_points_and_sketches(request: &ExportMetricsServiceRequest) -> }); } } + Some(Data::SumAgg(sa)) => { + // First-class Sum AggregationType: each data point carries + // a SumState envelope ({sum,count}) in `sketch`. Decode it + // and feed the sum as a MetricPoint into the SAME + // ExactAgg(Sum) path as a plain delta Sum — the backend sums + // the per-window/per-shard partials for the same sid. + for dp in &sa.data_points { + let value = match crate::precompute_engine::operators::sum_accumulator::SumAccumulator::from_sum_bytes(&dp.sketch) { + Ok(acc) => acc.sum, + Err(e) => { + debug!("asap_edge: SumAgg data point decode failed (skipping): {e}"); + continue; + } + }; + let labels = merge_point_attributes(&base_labels, &dp.attributes); + points.push(MetricPoint { + name: metric.name.clone(), + labels, + timestamp_nanos: dp.time_unix_nano, + value, + }); + } + } Some(Data::Histogram(hist)) => { for dp in &hist.data_points { if let Some((attr_name, payload)) = diff --git a/data_plane/src/precompute_engine/operators/sum_accumulator.rs b/data_plane/src/precompute_engine/operators/sum_accumulator.rs index 6fbc583d..6c6fa0c4 100644 --- a/data_plane/src/precompute_engine/operators/sum_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/sum_accumulator.rs @@ -42,7 +42,27 @@ impl SumAccumulator { Ok(Self::with_sum(sum)) } - + /// Decode the fixed Sum payload produced by the first-class Sum + /// AggregationType path (asap-precompute-go's SumWrapper): float64 sum + /// (little-endian) followed by uint64 count (little-endian), 16 bytes. + /// + /// Sum is an aggregation, NOT a sketch, so this deliberately does NOT + /// depend on the sketchlib sketch-envelope proto — the payload is a small + /// self-contained fixed layout. It decodes into the SAME + /// `AggregationType::Sum` accumulator as a plain-OTLP Sum, so the SumAgg + /// envelope and a plain Sum land on one identity (`exact_agg:Sum`) with no + /// new SketchKindHandle. `count` is decoded but not retained + /// (SumAccumulator tracks the scalar sum only; Sum is never sample_p-thinned + /// so no 1/p rescale is needed). + pub fn from_sum_bytes(buffer: &[u8]) -> Result> { + if buffer.len() < 16 { + return Err(format!("Sum payload too short: {} bytes (want 16)", buffer.len()).into()); + } + let sum = f64::from_le_bytes(buffer[0..8].try_into().unwrap()); + // count = u64::from_le_bytes(buffer[8..16]) — decoded position documented + // but not retained by the scalar-sum accumulator. + Ok(Self::with_sum(sum)) + } } impl Default for SumAccumulator { @@ -275,6 +295,28 @@ mod tests { assert_eq!(acc.type_name(), "SumAccumulator"); } + #[test] + fn from_sum_bytes_decodes_go_sum_payload() { + // GOLDEN: the 16-byte payload asap-precompute-go's + // SumWrapper{10,20,30,40}.Snapshot() emits — float64 sum (LE) followed + // by uint64 count (LE), sum=100, count=4. Proves the Rust backend + // decodes the first-class Sum payload the Go agent produces + // (cross-language wire parity, no sketchlib proto dependency). + let go_bytes: &[u8] = &[ + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x59, 0x40, // 100.0 f64 LE + 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // 4 u64 LE + ]; + let acc = SumAccumulator::from_sum_bytes(go_bytes).expect("decode Go Sum payload"); + assert_eq!(acc.sum, 100.0, "decoded Go SumWrapper payload sum"); + } + + #[test] + fn from_sum_bytes_rejects_short_payload() { + // A short buffer is rejected (the ingest path then skips the point). + assert!(SumAccumulator::from_sum_bytes(&[]).is_err()); + assert!(SumAccumulator::from_sum_bytes(&[0u8; 8]).is_err()); + } + #[test] fn aux_stats_exposes_sum_only() { let acc = SumAccumulator::with_sum(123.5); From 52b5fa949c5c808bb05b94f2f0878415d0ff4801 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 28 May 2026 10:36:21 -0600 Subject: [PATCH 3/4] Wire item_label controller->data-plane so CMS estimate(key) is reachable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit control_plane: thread the per-metric item_label into the backend streaming-config. BackendAggregation gains an item_label field (patched from collect_metric_to_item_label in both plan-handler loops), emitted into the policy parameters["item_label"] by build_backend_aggregation_json. data_plane: at sketch ingest registration, read the matched policy's parameters["item_label"] and record it on the sid via set_item_label — so the FrequencyEstimate safe-miss gate resolves the item key and serves per-item estimate(key). Previously the gate was inert (set_item_label was never called). control_plane 794 lib tests pass; data_plane Sum/CMS tests green. Co-Authored-By: Claude Opus 4.7 --- control_plane/src/emit/backend_push.rs | 1 + control_plane/src/emit/stage_config.rs | 22 ++++++++++++++++++- control_plane/src/emit/trait_def.rs | 1 + control_plane/src/main.rs | 5 +++++ .../src/physical/colored_dag/emitter.rs | 10 +++++++++ control_plane/src/replan.rs | 11 ++++++++++ data_plane/src/drivers/ingest/otel.rs | 20 +++++++++++++++++ 7 files changed, 69 insertions(+), 1 deletion(-) diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index b73c4f88..ff340f00 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -638,6 +638,7 @@ mod tests { sketch_kind: SketchKind::DDSketch, sketch_params: SketchParams::DDSketch(DDSketchParams { alpha: 0.01 }), grouping: vec![], + item_label: None, spatial_filter: String::new(), window_secs: 60, aggregation_input: AggregationInput::SketchEnvelope, diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 02b45bb4..785a15a6 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -2694,13 +2694,23 @@ fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonValue { // synthesizes for non-sketch (Sum-shaped) workloads. The // `sketch_kind` / `sketch_params` fields carry sentinel values // in this case and are not emitted on the wire. - let (aggregation_type, parameters) = match &agg.agg_type_override { + let (aggregation_type, mut parameters) = match &agg.agg_type_override { Some(s) => (s.clone(), json!({})), None => ( sketch_kind_to_backend_type(&agg.sketch_kind, &agg.sketch_params).to_string(), sketch_params_to_json(&agg.sketch_params), ), }; + // Carry the per-item dimension (e.g. "endpoint"/"service") into the + // policy parameters so the data-plane ingest can record it on the CMS + // sid and answer per-item estimate(key). Only set for item_label-mode + // frequency sketches; a subset content-match keeps policy resolution + // working for sketches that don't carry it. + if let Some(label) = &agg.item_label { + if let Some(obj) = parameters.as_object_mut() { + obj.insert("item_label".to_string(), JsonValue::String(label.clone())); + } + } let aggregation_input = match agg.aggregation_input { AggregationInput::SketchEnvelope => "sketch_envelope", AggregationInput::Raw => "raw", @@ -3097,6 +3107,7 @@ mod tests { let cfg = BackendStageConfig { aggregations: vec![ BackendAggregation { + item_label: None, aggregation_id: "agg0".into(), metric_name: "http_latency_ms".into(), sketch_kind: SketchKind::DDSketch, @@ -3108,6 +3119,7 @@ mod tests { agg_type_override: None, }, BackendAggregation { + item_label: None, aggregation_id: "agg1".into(), metric_name: "http_requests_total".into(), sketch_kind: SketchKind::Hll, @@ -3161,6 +3173,7 @@ mod tests { let cfg = BackendStageConfig { aggregations: vec![ BackendAggregation { + item_label: None, aggregation_id: "agg0".into(), metric_name: "endpoint_count".into(), sketch_kind: SketchKind::CountSketch, @@ -3176,6 +3189,7 @@ mod tests { agg_type_override: None, }, BackendAggregation { + item_label: None, aggregation_id: "agg1".into(), metric_name: "endpoint_hits".into(), sketch_kind: SketchKind::Cms, @@ -3247,6 +3261,7 @@ mod tests { }; BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: "agg0".into(), metric_name: "test_metric".into(), sketch_kind: kind.clone(), @@ -3627,6 +3642,7 @@ mod tests { fn backend_json_emits_grouping_under_labels() { let cfg = BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: "agg0".into(), metric_name: "http_latency_ms".into(), sketch_kind: SketchKind::DDSketch, @@ -3672,6 +3688,7 @@ mod tests { fn phase_b_backend_json_aggregation_readout_alias_snapshot() { let cfg = BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: "phase_b_agg0".into(), metric_name: "phase_b_metric".into(), sketch_kind: SketchKind::Kll, @@ -3719,6 +3736,7 @@ mod tests { fn phase_eps1_mode1_aggregation_input_is_sketch_envelope() { let cfg = BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: "agg0".into(), metric_name: "test_metric".into(), sketch_kind: SketchKind::DDSketch, @@ -3743,6 +3761,7 @@ mod tests { fn phase_eps1_mode2_aggregation_input_is_raw() { let cfg = BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: "agg0".into(), metric_name: "test_metric".into(), sketch_kind: SketchKind::DDSketch, @@ -5429,6 +5448,7 @@ mod tests { let cfg = BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: "agg0".to_string(), metric_name: "http_requests_total_latency_ms".to_string(), sketch_kind: SketchKind::DDSketch, diff --git a/control_plane/src/emit/trait_def.rs b/control_plane/src/emit/trait_def.rs index abeb4f60..9a2ca419 100644 --- a/control_plane/src/emit/trait_def.rs +++ b/control_plane/src/emit/trait_def.rs @@ -234,6 +234,7 @@ mod tests { fn empty_backend_cfg() -> BackendStageConfig { BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: "agg0".to_string(), metric_name: "test_metric".to_string(), sketch_kind: SketchKind::DDSketch, diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index c99a8079..5a784d9d 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -752,6 +752,10 @@ async fn handle_plan( // `QueryWorkload` carries both unambiguously, // and every aggregation under one workload // shares them — so the patch is uniform. + let item_labels = emit::collect_metric_to_item_label( + &st.workload_registry, + &st.workload_store, + ); for agg in &mut be.aggregations { if agg.metric_name.is_empty() { agg.metric_name = workload.metric_name.clone(); @@ -760,6 +764,7 @@ async fn handle_plan( agg.window_secs = workload.time_window.as_secs(); } agg.grouping = workload.group_by_labels.clone(); + agg.item_label = item_labels.get(&agg.metric_name).cloned(); } // Option B unification: every typed cumulative // emit (handle_plan here, Replanner triggers diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index 41f960ab..82299470 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -571,6 +571,14 @@ pub struct BackendAggregation { /// strings are the only reliable source of the names today. #[serde(default)] pub grouping: Vec, + /// Per-item dimension (the data-point attribute NAME, e.g. "endpoint" + /// or "service") for an item_label-mode frequency sketch. Like + /// `grouping`, the L5 emitter leaves this `None`; `handle_plan` patches + /// it from the workload's `item_label`. Emitted into the aggregation's + /// `parameters["item_label"]` so the data-plane ingest records it on the + /// CMS sid and can answer per-item `estimate(key)` (FrequencyEstimate). + #[serde(default)] + pub item_label: Option, /// Phase ε.1 — what shape the backend ingests for this /// aggregation. Mode 1 (sketch at edge) / sketch_envelope is the /// default (the wire payload is a sketch state already). Mode 2 @@ -786,6 +794,7 @@ impl Emitter for ThreeStageEmitter { aggregation_id: aggregation_id.clone(), }); backend_aggregations.push(BackendAggregation { + item_label: None, aggregation_id, metric_name: edge.source_metric.clone().unwrap_or_default(), sketch_kind: sketch_type.clone(), @@ -869,6 +878,7 @@ impl Emitter for ThreeStageEmitter { let aid = format!("agg{next_agg_index}"); next_agg_index += 1; backend_aggregations.push(BackendAggregation { + item_label: None, aggregation_id: aid, metric_name: edge.source_metric.clone().unwrap_or_default(), sketch_kind: family.clone(), diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index 5c497103..a9e7f76a 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -598,6 +598,14 @@ impl Replanner { // `ColumnId`s with no label-name resolution today). The // `QueryWorkload` carries both unambiguously, and every // aggregation under one workload shares them. + // Per-metric item_label (the high-card dimension a CMS/CountSketch + // hashes): threaded into the policy params so the data-plane ingest + // records it on the sid and can answer per-item estimate(key). + let item_labels = self + .workload_registry + .as_ref() + .map(|reg| crate::emit::collect_metric_to_item_label(reg, &self.workload_store)) + .unwrap_or_default(); for agg in &mut be.aggregations { if agg.metric_name.is_empty() { agg.metric_name = workload.metric_name.clone(); @@ -606,6 +614,7 @@ impl Replanner { agg.window_secs = workload.time_window.as_secs(); } agg.grouping = workload.group_by_labels.clone(); + agg.item_label = item_labels.get(&agg.metric_name).cloned(); } return Some(be); } @@ -633,6 +642,7 @@ impl Replanner { let window_secs = workload.time_window.as_secs().max(1); Some(BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: format!("exact-{}-{}", workload.metric_name, role), metric_name: workload.metric_name.clone(), // Sentinel sketch_kind / sketch_params — `agg_type_override` @@ -1242,6 +1252,7 @@ mod tests { ("http_requests_total".to_string(), AggRole::Sum), BackendStageConfig { aggregations: vec![BackendAggregation { + item_label: None, aggregation_id: "exact-http_requests_total-sum".to_string(), metric_name: "http_requests_total".to_string(), sketch_kind: SketchKind::DDSketch, diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 5b1cd8a1..f909eea6 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -1289,6 +1289,23 @@ async fn route_modified_otlp_sketches_to_precompute( &cfg, &group_by_keys, ); + // Per-item dimension (item_label) the controller threaded + // into the matched policy's parameters — recorded on the sid + // below so the query engine can answer per-item estimate(key) + // (the CMS/CountSketch FrequencyEstimate gate consults it). + let item_label_for_sid: Option = { + let snap = ingest_state.config_snapshot(); + snap.get_aggregation_config(policy_fp.as_u64()) + .or_else(|| { + snap.get_all_aggregation_configs() + .values() + .find(|c| c.metric == canonical_name) + }) + .and_then(|c| c.parameters.get("item_label")) + .and_then(|v| v.as_str()) + .filter(|s| !s.is_empty()) + .map(|s| s.to_string()) + }; ingest_state.sketch_index.register(SketchInstanceMetadata { sid, metric_name: canonical_name.clone(), @@ -1305,6 +1322,9 @@ async fn route_modified_otlp_sketches_to_precompute( expires_at_ms: None, policy_fp, }); + if let Some(label) = &item_label_for_sid { + ingest_state.sketch_index.set_item_label(sid, label); + } } else if let Some(existing) = ingest_state.sketch_index.instance(sid) { // P1-4 (a) — one-way capability UPGRADE. The sid // was first registered from a non-heap frame From 6485f814818c9267c73265533618914d151ee87c Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 28 May 2026 15:39:26 -0600 Subject: [PATCH 4/4] data-plane: honest SketchStore memory diagnostic + idle-sid eviction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `SketchStore::approx_memory_bytes` is the flusher's evictable-payload gauge (current_epoch + sealed payloads). At idle it correctly reads ~0 because payloads are flushed to disk — but the per-sid registry (`instances` metadata + per-series `InternTable` label caches) stays resident, so the MEMORY_DIAG line read "0.00 KB" while process RSS sat in the hundreds of MB, and there was no relief valve for that residue (stale sketch sids are pinned until config-driven retirement). Two cohesive changes, both scoped to the SketchStore: 1. Honest memory diagnostic. Adds `InternTable::approx_heap_bytes` and `SketchStore::approx_resident_bytes` (registry metadata + intern caches + live payload) and `process_resident_bytes()` (/proc/self/statm). The 30s MEMORY_DIAG line now logs payload (evictable) / registry+intern (resident, not flushable) / process RSS, so the real footprint is visible. The flusher's payload-pressure trigger is intentionally left on the payload gauge — feeding non-flushable registry memory into it would livelock the flush loop (it can only evict payload). 2. Idle-sid eviction (opt-in --idle-sid-evict-secs / ASAP_IDLE_SID_EVICT_SECS, default 0 = off). A periodic sweep drops the in-memory `SidStoreData` (epoch columns + intern cache + series slot) for sketch sids that are persistence-backed, write-idle past the threshold, AND fully durable on disk (sealed_epochs + current_epoch empty), while KEEPING the queryable `SketchInstanceMetadata`. `union_disk_parts_into` reads the durable tier via the retained metadata only (never `self.series`), so an evicted series stays answerable from disk; the append path's `entry().or_insert_with` rehydrates a fresh store on the next write. Bounds resident registry memory under series churn. Effective idle horizon is max(threshold, persistence_hot_window) since eviction waits for a sid's windows to seal+flush first. Adds `SidStoreData::last_write_unix_ms`, stamped under the append write lock the hot path already holds (no extra cost). 5 unit tests cover resident accounting and the eviction predicate / rehydration / durability guard. Validated live on the Google-cluster trace: 31.5k stale sids reclaimed, in-memory sid state held to a few thousand vs pinning all 36k; MEMORY_DIAG now shows payload≈KB / registry≈MB / RSS≈MB instead of a misleading 0.00 KB. Co-Authored-By: Claude Opus 4.8 (1M context) --- data_plane/src/main.rs | 79 +++++- .../sketch_db/index/epoch_columnar.rs | 41 +++ .../storage_engines/sketch_db/index/mod.rs | 268 ++++++++++++++++++ 3 files changed, 385 insertions(+), 3 deletions(-) diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index edbb3034..deaee547 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -196,6 +196,18 @@ struct Args { #[arg(long)] schema_eviction_dry_run: bool, + /// Idle-sid eviction (memory reclaim). Drop the in-memory state of any + /// sketch sid with no writes for this many seconds AND whose state is + /// fully flushed to disk, keeping its queryable metadata — the series + /// stays answerable from the durable tier and rehydrates on the next + /// write. Bounds resident registry memory when series churn / go stale + /// (without it, stale sketch sids are pinned in RAM until config-driven + /// retirement). 0 disables. Effective horizon is + /// max(this, --persistence-hot-window-secs), since eviction waits for + /// the sid's windows to seal+flush first. + #[arg(long, env = "ASAP_IDLE_SID_EVICT_SECS", default_value = "0")] + idle_sid_evict_secs: u64, + // ---- SketchStore persistence ---- // // When --persistence-enabled is set, the store is constructed via @@ -502,6 +514,36 @@ async fn main() -> Result<()> { // via the shared `SketchStore` (already passed in above). let engine = Arc::new(engine); + // Idle-sid eviction sweep (memory reclaim) — opt-in via + // --idle-sid-evict-secs. Drops the in-memory `SidStoreData` for + // write-idle, fully-flushed sketch sids while keeping their queryable + // metadata, bounding resident registry memory under series churn. + if args.idle_sid_evict_secs > 0 { + let evict_index = sketch_index.clone(); + let idle_ms = args.idle_sid_evict_secs.saturating_mul(1000); + // Sweep a few times per idle horizon, clamped to a sane cadence. + let sweep = std::time::Duration::from_secs(args.idle_sid_evict_secs.clamp(10, 60)); + info!( + "Idle-sid eviction enabled: idle threshold {}s, sweep every {}s", + args.idle_sid_evict_secs, + sweep.as_secs() + ); + tokio::spawn(async move { + let mut interval = tokio::time::interval(sweep); + loop { + interval.tick().await; + let n = evict_index.evict_idle_series(idle_ms); + if n > 0 { + info!( + "[IDLE_EVICT] evicted {} idle sid(s) from memory \ + (still queryable from disk; rehydrate on next write)", + n + ); + } + } + }); + } + // Setup OTLP receiver (after precompute engine so it can share the ingest state) // Issue #46 ⑥ — freshness-probe last-value cache. Shared between // the OTLP receiver (write path) and the HTTP query handler (read @@ -833,6 +875,23 @@ async fn main() -> Result<()> { Ok(()) } +/// Best-effort process resident-set size (RSS) in bytes, read from +/// `/proc/self/statm` (field 2 = resident pages × page size). Returns 0 if +/// unreadable (non-Linux / sandboxed) so the diagnostic degrades gracefully +/// rather than failing. This is the ground-truth counterpart to the +/// store's structural estimates in the memory diagnostic. +fn process_resident_bytes() -> usize { + let Ok(statm) = std::fs::read_to_string("/proc/self/statm") else { + return 0; + }; + let Some(resident_pages) = statm.split_whitespace().nth(1) else { + return 0; + }; + let pages: usize = resident_pages.parse().unwrap_or(0); + // `sysconf(_SC_PAGESIZE)` is 4 KiB on every platform this runs on. + pages * 4096 +} + /// Periodic memory diagnostics logger — runs every 30 seconds. async fn spawn_memory_diagnostics( sketch_index: Arc, @@ -849,12 +908,26 @@ async fn spawn_memory_diagnostics( // pre-M2.3 per-agg_id SketchStore::diagnostic_info). let instance_count = sketch_index.instance_count(); let series_count = sketch_index.series_len(); - let approx_bytes = sketch_index.approx_memory_bytes(); + // `approx_memory_bytes` is the flusher's EVICTABLE-payload gauge: + // it counts only live sketch payloads (current_epoch + sealed), so + // it correctly reads ~0 once everything has been flushed to disk. + // On its own it badly misrepresents the store's footprint — the + // per-sid registry + intern caches stay resident and are not + // flushable. Report all three: evictable payload, the structural + // resident estimate, and the process RSS ground truth. + let payload_bytes = sketch_index.approx_memory_bytes(); + let resident_bytes = sketch_index.approx_resident_bytes(); + let rss_bytes = process_resident_bytes(); info!( - "[MEMORY_DIAG] SketchStore: {} instance(s), {} sid(s) with state, {:.2} KB approx in-memory bytes (hot current_epoch + sealed)", + "[MEMORY_DIAG] SketchStore: {} instance(s), {} sid(s) with state, \ + payload={:.2} KB (evictable, flusher gauge), \ + registry+intern\u{2248}{:.2} MB (resident, not flushable), \ + process RSS={:.1} MB", instance_count, series_count, - approx_bytes as f64 / 1024.0, + payload_bytes as f64 / 1024.0, + resident_bytes as f64 / (1024.0 * 1024.0), + rss_bytes as f64 / (1024.0 * 1024.0), ); // 2. Worker diagnostics (precompute engine only) diff --git a/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs index 8e613afc..33c5cfa3 100644 --- a/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs +++ b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs @@ -104,6 +104,37 @@ impl Default for InternTable { } } +impl InternTable> { + /// Approximate RESIDENT heap bytes of the interned label maps for the + /// concrete `BTreeMap` key the SketchStore uses. + /// + /// Both `id_to_label` (the `Vec`) and `label_to_id` (the `HashMap`) + /// retain a clone of every interned key, so each distinct label map is + /// counted twice, plus backing-store slot capacity. This is the + /// dominant per-sid resident cost once sketch payloads have been + /// flushed to disk — and it is exactly the cost the payload-only + /// [`crate::storage_engines::sketch_db::index::persistence::EpochSource::approx_memory_bytes`] + /// (the flusher's eviction gauge) does NOT see, which is why the memory + /// diagnostic read ~0 KB while RSS sat in the hundreds of MB. + pub fn approx_heap_bytes(&self) -> usize { + let mut key_bytes = 0usize; + for m in &self.id_to_label { + for (k, v) in m.iter() { + // string bytes + the two `String` headers + a BTree node. + key_bytes += k.len() + v.len() + 2 * std::mem::size_of::() + 32; + } + key_bytes += std::mem::size_of::>(); + } + // ×2 for the cloned copy held by `label_to_id`, plus the backing + // Vec / HashMap slot capacity. + key_bytes * 2 + + self.id_to_label.capacity() * std::mem::size_of::>() + + self.label_to_id.capacity() + * (std::mem::size_of::>() + + std::mem::size_of::()) + } +} + /// Active (mutable) epoch: append-only insert, O(1) amortized. /// /// Three parallel columns (Opt 5) — windows / label-id / payload — @@ -816,6 +847,15 @@ pub struct SidStoreData { /// So when this is `true`, `enforce_retention` is a no-op. When /// `false` (the default), #327 retention is the memory bound. pub persistence_enabled: bool, + /// Wall-clock millis of the most recent write (append) into this sid. + /// `0` means "never written" / freshly (re)hydrated. Drives idle-sid + /// eviction: a sid with no writes for the idle threshold whose state + /// is fully durable on disk can have this whole `SidStoreData` dropped + /// from memory while its queryable `SketchInstanceMetadata` is kept + /// (the series stays answerable from the disk tier and rehydrates on + /// the next write). Updated under the per-sid write lock the append + /// path already holds, so it costs nothing extra on the hot path. + pub last_write_unix_ms: u64, } /// Default in-memory retention horizon (ms) for the WARM sketch store. @@ -858,6 +898,7 @@ impl SidStoreData { retention_horizon_ms: default_retention_horizon_ms(), seal_window_count: None, persistence_enabled: false, + last_write_unix_ms: 0, } } 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 2b5f2764..6d6034d8 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -536,6 +536,7 @@ impl SketchStore { .clone(); let mut guard = store.write().unwrap(); guard.insert(window, series_label_values, AggPayload::Sketch(sample)); + guard.last_write_unix_ms = now_ms(); } /// Build a `SidStoreData` pre-configured for the store's current @@ -577,6 +578,7 @@ impl SketchStore { .clone(); let mut guard = store.write().unwrap(); guard.insert(window, series_label_values, AggPayload::ExactAgg(payload)); + guard.last_write_unix_ms = now_ms(); } /// Range-query the ASAP-tier state for one sid. Window-end-keyed @@ -1403,6 +1405,140 @@ impl SketchStore { self.instances.read().unwrap().len() } + /// True when a sid's in-memory `SidStoreData` may be dropped to reclaim + /// resident memory: persistence owns its durability, NOTHING is pending + /// in memory (so dropping it loses no un-flushed data), and it has been + /// write-idle for at least `idle_threshold_ms`. `last_write_unix_ms == 0` + /// (never written / freshly rehydrated) is never evictable. + fn is_idle_evictable( + data: &SidStoreData, AggPayload>, + now: u64, + idle_threshold_ms: u64, + ) -> bool { + data.persistence_enabled + && data.sealed_epochs.is_empty() + && data.current_epoch.is_empty() + && data.last_write_unix_ms != 0 + && now.saturating_sub(data.last_write_unix_ms) >= idle_threshold_ms + } + + /// Idle-sid eviction (memory reclaim). Drops the in-memory + /// `SidStoreData` (epoch columns + intern-table label cache + the + /// `series` slot) for every sketch sid that has gone write-idle past + /// `idle_threshold_ms` AND whose state is fully durable on disk, while + /// KEEPING its [`SketchInstanceMetadata`] in `instances`. + /// + /// Why keep the metadata: the query path's disk union + /// ([`Self::query_range`] → `union_disk_parts_into`) needs + /// `sid_group_by_keys(sid)` and `instances_matching` needs the + /// metric/keys entry — drop those and the series silently stops + /// resolving warm and falls through to the archive. So only the heavy, + /// reconstructable part is evicted; the series stays queryable from the + /// durable tier and the append path + /// ([`Self::append_sample`]/[`Self::append_precompute`], both + /// `entry(..).or_insert_with(..)`) transparently rehydrates a fresh + /// store on the next write. + /// + /// Returns the number of sids evicted. `idle_threshold_ms == 0` is a + /// no-op (feature disabled). O(N sids); meant for a periodic sweep, not + /// the hot path. NOTE: because eviction requires `current_epoch` to be + /// empty (all windows sealed+flushed), the effective idle horizon is + /// `max(idle_threshold_ms, persistence_hot_window)`. + pub fn evict_idle_series(&self, idle_threshold_ms: u64) -> usize { + if idle_threshold_ms == 0 { + return 0; + } + let now = now_ms(); + + // Pass 1: collect candidates under read-only iteration. Removing + // during `iter()` can deadlock against our own shard guards, so we + // only gather here and remove afterwards. + let mut candidates = Vec::new(); + for entry in self.series.iter() { + if let Ok(data) = entry.value().read() { + if Self::is_idle_evictable(&data, now, idle_threshold_ms) { + candidates.push(*entry.key()); + } + } + } + if candidates.is_empty() { + return 0; + } + + // Pass 2: remove each, RE-CHECKING under the per-sid write lock so a + // concurrent write that rehydrated/appended between the two passes + // is not dropped. `remove_if` only deletes when the closure returns + // true; taking the write lock there serializes with the append + // path's `store.write()`. (No lock-order inversion: no path holds a + // per-sid lock while acquiring a `series` shard lock.) + let mut evicted = 0usize; + for sid in candidates { + let removed = self.series.remove_if(&sid, |_, store| { + store + .write() + .map(|d| Self::is_idle_evictable(&d, now, idle_threshold_ms)) + .unwrap_or(false) + }); + if removed.is_some() { + evicted += 1; + } + } + evicted + } + + /// Approximate TOTAL resident bytes held by the store — the honest + /// counterpart to the [`persistence::EpochSource::approx_memory_bytes`] + /// payload gauge. + /// + /// `approx_memory_bytes` (used by the flusher's pressure trigger) + /// counts ONLY live sketch payloads in `current_epoch` + `sealed_epochs` + /// — which is correct for deciding what to FLUSH, because flushing only + /// relieves payload. But once payloads are sealed to disk it reads ~0, + /// even while the per-sid registry (`instances` metadata, the secondary + /// indexes, and the per-series `InternTable` label caches) keeps + /// hundreds of MB resident. That residue is NOT evictable by flushing — + /// it is only released by retiring/evicting the sid itself — so it must + /// not feed the flush trigger, but the memory DIAGNOSTIC must surface it + /// or operators are blind to the real footprint. This method is that + /// surface; it is O(N sids) and meant for the 30 s diagnostic tick, not + /// the hot path. + pub fn approx_resident_bytes(&self) -> usize { + let mut total = 0usize; + + // 1. Registry metadata (instances map + its string heaps). + if let Ok(insts) = self.instances.read() { + for m in insts.values() { + total += std::mem::size_of::(); + total += m.metric_name.len(); + for k in &m.group_by_keys { + total += k.len() + std::mem::size_of::(); + } + } + total += insts.capacity() + * (std::mem::size_of::() + std::mem::size_of::()); + } + + // 2. Per-sid series storage: live payloads + interned label maps + + // the `Arc>` container slot. + for entry in self.series.iter() { + let Ok(data) = entry.value().read() else { + continue; + }; + for (_, _, payload) in data.current_epoch.iter_entries() { + total += payload.approx_bytes(); + } + for epoch in data.sealed_epochs.values() { + for (_, _, payload) in &epoch.entries { + total += payload.approx_bytes(); + } + } + total += data.intern.approx_heap_bytes(); + total += std::mem::size_of::(); + } + + total + } + /// Clone every registered `SketchInstanceMetadata` into a snapshot /// vec. Used by read-side primitives that need to scan the whole /// catalog without holding the registry lock across user code @@ -3618,6 +3754,138 @@ mod tests { windows — the diagnostic under-reports and the flusher's memory trigger is blind" ); } + + /// The idle under-report this fix targets: once a sid's payload has + /// been flushed/evicted to disk, `approx_memory_bytes` (the flusher's + /// evictable gauge) reads 0 — but the registered sid still costs + /// resident registry/metadata memory. `approx_resident_bytes` must + /// surface that so the memory diagnostic isn't blind (the live + /// "0.00 KB while 600 MB RSS" symptom). + #[test] + fn approx_resident_bytes_counts_registry_when_payload_is_zero() { + let idx = SketchStore::new(); + idx.register(meta_with_host_key(9001)); + // No samples appended → no live payload (models the idle sid whose + // epochs were sealed+flushed to disk). + assert_eq!( + idx.approx_memory_bytes(), + 0, + "precondition: no resident payload" + ); + assert!( + idx.approx_resident_bytes() > 0, + "approx_resident_bytes must account for the registered sid's \ + metadata even when no payload is resident" + ); + } + + /// Resident accounting must include the per-series intern-table label + /// cache, which grows with the number of distinct label-value maps a + /// sid has seen — the dominant per-sid resident cost at scale. + #[test] + fn approx_resident_bytes_grows_with_interned_label_cardinality() { + let idx = SketchStore::new(); + idx.register(meta_with_host_key(9100)); + for i in 0..50u64 { + let s = i * 30_000; + idx.append_sample( + 9100, + lv_host(&format!("host-{i}")), + (s, s + 30_000), + sample((i + 1) as u8), + ); + } + let many = idx.approx_resident_bytes(); + + let idx2 = SketchStore::new(); + idx2.register(meta_with_host_key(9101)); + idx2.append_sample(9101, lv_host("host-0"), (0, 30_000), sample(1)); + let few = idx2.approx_resident_bytes(); + + assert!( + many > few, + "resident bytes should grow with interned label cardinality: \ + many={many} few={few}" + ); + } + + #[test] + fn is_idle_evictable_predicate() { + let now = now_ms(); + let mut d = SidStoreData::, AggPayload>::new(); + d.persistence_enabled = true; + d.last_write_unix_ms = now.saturating_sub(120_000); + assert!( + SketchStore::is_idle_evictable(&d, now, 60_000), + "idle 120s past a 60s threshold, durable + empty → evictable" + ); + assert!( + !SketchStore::is_idle_evictable(&d, now, 300_000), + "idle 120s under a 300s threshold → spared" + ); + d.last_write_unix_ms = 0; + assert!( + !SketchStore::is_idle_evictable(&d, now, 1), + "never-written (0) is never evictable" + ); + // In-memory-only (no persistence) sids are never idle-evicted — there + // is no durable copy to serve them from. + let mut d2 = SidStoreData::, AggPayload>::new(); + d2.persistence_enabled = false; + d2.last_write_unix_ms = now.saturating_sub(120_000); + assert!(!SketchStore::is_idle_evictable(&d2, now, 1)); + } + + #[test] + fn evict_idle_series_drops_state_but_keeps_metadata() { + let idx = SketchStore::new(); + idx.register(meta_with_host_key(7001)); + // Durable, write-idle, empty-in-memory sid (models a series whose + // windows have all sealed+flushed to disk and then gone quiet). + let mut d = SidStoreData::, AggPayload>::new(); + d.persistence_enabled = true; + d.last_write_unix_ms = now_ms().saturating_sub(120_000); + idx.series.insert(7001, Arc::new(RwLock::new(d))); + assert_eq!(idx.series.len(), 1); + + let evicted = idx.evict_idle_series(60_000); + assert_eq!(evicted, 1, "the idle sid is evicted"); + assert_eq!(idx.series.len(), 0, "in-memory state dropped"); + assert!( + idx.instances.read().unwrap().contains_key(&7001), + "metadata retained → series stays queryable from disk + rehydrates" + ); + + // A subsequent append rehydrates the series entry transparently. + idx.append_sample(7001, lv_host("h"), (0, 30_000), sample(1)); + assert_eq!(idx.series.len(), 1, "append rehydrated the evicted sid"); + } + + #[test] + fn evict_idle_series_spares_recent_and_pending_sids() { + let idx = SketchStore::new(); + // Recently written → not idle. + idx.register(meta_with_host_key(7101)); + let mut recent = SidStoreData::, AggPayload>::new(); + recent.persistence_enabled = true; + recent.last_write_unix_ms = now_ms(); + idx.series.insert(7101, Arc::new(RwLock::new(recent))); + // Idle, but still holds un-flushed data in current_epoch → dropping it + // would lose data, so it MUST be spared. + idx.register(meta_with_host_key(7102)); + let mut pending = SidStoreData::, AggPayload>::new(); + pending.persistence_enabled = true; + pending.last_write_unix_ms = now_ms().saturating_sub(120_000); + pending.insert((0, 30_000), lv_host("h"), AggPayload::Sketch(sample(1))); + idx.series.insert(7102, Arc::new(RwLock::new(pending))); + + assert_eq!( + idx.evict_idle_series(60_000), + 0, + "recent + pending-data sids are spared" + ); + assert_eq!(idx.series.len(), 2); + } } // 2026-05 reorg: generic epoch-partitioned columnar storage lives