From 376764f76c54a8b4986635a0373b1fecbc5c4320 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 17 May 2026 21:22:40 -0600 Subject: [PATCH] refactor(data_plane): rekey ingest bucketing from (agg_id, group_key) to sid (B7.6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Schema retirement #5 step 6: the precompute engine's per-bucket state used to be keyed by `(agg_id, group_key)`. The grouping label values already fold into the sid via `SeriesIdResolver`'s `(metric, attrs_fingerprint, agg_kind_canonical)` identity contract, so the tuple collapses to a single u64 — `sid` is now the bucket key throughout the ingest → router → worker path. Scope: - `WorkerMessage::GroupSamples` / `AccumulatorInput` carry `sid: u64, policy_fp: PolicyFingerprint, group_key: String` instead of `agg_id + group_key`. `policy_fp` is the source-config handle the worker uses to look up `AggregationConfig` from the hot-reload snapshot; `group_key` still travels for emit-time `KeyByLabelValues` rendering. This is a BREAKING change to `WorkerMessage`, but the enum is private to the data_plane crate. - `SeriesRouter::route_group_batch` hashes by `sid` alone (`worker_for_sid`). Same sid always lands on the same worker; bucket state stays single-owner. - `Worker::group_states: HashMap` (was `HashMap<(u64, String), GroupState>`). `GroupState` gains `policy_fp` and `group_key` fields so `evict_orphaned_groups` can check policy liveness and the emit path can render labels without re-keying the bucket. - `process_group_samples` / `process_accumulator_input` take `(sid, policy_fp, group_key, ...)`. All in-crate call sites updated. - OTLP ingest helper `resolve_bucket_sid_for_agg_config(state, config, point_labels) → (sid, policy_fp)`: derives the bucket sid for one `(config, DP)` pair by resolving against `(metric, grouping-label- values, ExactAgg-of-config)`. Used by all three OTLP dispatch paths (raw points, opaque SketchEnvelope, modified-OTLP first-class sketches). Crucially, "attrs" for sid purposes is the GROUPING-LABEL projection of wire labels — not the full label set — so distinct `(rack, node, pod)` tuples under a `grouping_labels=[zone]` policy still roll up into one bucket per zone (the GROUP-BY semantic). - Regression test added: `drivers::ingest::otel::sid_bucketing_tests:: raw_otlp_buckets_by_sid_with_distinct_group_keys` drives `route_otlp_to_precompute` end-to-end with two `zone` values × two DPs each, asserts exactly two `GroupSamples` are emitted with distinct non-zero sids that round-trip through `SeriesIdResolver::lookup`. The test docstring documents a pre-existing `format_series_key`/`parse_labels_from_series_key` inconsistency that makes `extract_group_key_for` return "" for OTLP inputs; B7.6 bucketing is unaffected because it reads `point.labels` directly (HashMap lookup), not the joined series_key. Files touched (3): - `data_plane/src/precompute_engine/series_router.rs` — message shape + routing hash + test rename. - `data_plane/src/precompute_engine/worker.rs` — `GroupState` / `Worker.group_states` retyped, `get_or_create_group_state` / `process_group_samples` / `process_accumulator_input` / `evict_orphaned_groups` / `flush_all` reworked, 30+ test call sites updated to pass `(sid, PolicyFingerprint, group_key)`. - `data_plane/src/drivers/ingest/otel.rs` — three OTLP dispatch paths switched to sid-bucketing via new `resolve_bucket_sid_for_agg_config` helper; added `sid_bucketing_tests` mod with the regression test. Test plan: - `cargo build -p data_plane` — clean. - `cargo test -p data_plane --lib` — 713 passed / 2 ignored, no new failures vs. main. - `cargo test -p data_plane` integration suite — same 2 pre-existing failures as origin/main (`controller_plan_to_query_full_roundtrip_ ddsketch` / `_kll`); verified by re-running on origin/main HEAD. Unrelated to B7.6. NOT in scope (left for B7.7): - `output_sink.rs` already reads `output.policy_fp` (no `agg_id`); no changes needed there. - `backfill/processor.rs` still uses `(agg_id, group_key)` internally — retired by B7.7. - `AggregationConfig::aggregation_id()` accessor remains; retiring it is deferred until B7.6 + B7.7 both land (per task brief). Co-Authored-By: Claude Opus 4.7 (1M context) --- data_plane/src/drivers/ingest/otel.rs | 372 +++++++++++++++++- .../src/precompute_engine/series_router.rs | 105 +++-- data_plane/src/precompute_engine/worker.rs | 329 ++++++++++------ 3 files changed, 636 insertions(+), 170 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 6c9cf673..05914e53 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -506,6 +506,58 @@ fn flush_barrier_drops(_state: &IngestState, drops: &HashMap, driver_t ); } +/// Resolve the bucket sid (and `policy_fp`) for a single data point +/// against a single matching `AggregationConfig`. +/// +/// B7.6 — sid is the bucket identity in the precompute engine; this +/// helper folds `(config, grouping-label-values)` into a single u64 via +/// `SeriesIdResolver`. Same `(metric, grouping-label-values, agg_kind)` +/// always returns the same sid, so distinct data points that share the +/// same group bucket land in the same `GroupSamples` / +/// `AccumulatorInput` message — the GROUP-BY semantics the legacy +/// `(agg_id, group_key)` tuple expressed. +/// +/// The "attrs" passed to the resolver are the GROUPING-LABEL projection +/// of the wire labels (NOT the full label set) — otherwise every +/// distinct `(rack, node, pod)` tuple under a `grouping_labels=[zone]` +/// policy would mint its own sid and never roll up. +/// +/// `agg_kind` is `ExactAgg { ... }` for both raw-sample and opaque- +/// envelope sketch paths so the resolver key matches the signature +/// `reconcile_from_streaming_config` derives from the same config; the +/// modified-OTLP first-class sketch path takes a different sid- +/// resolution route inside `route_modified_otlp_sketches_to_precompute` +/// because it carries per-DP `(SketchKindHandle, SketchConfig)` and +/// must distinguish (e.g.) DDSketch vs Kll over the same series. +fn resolve_bucket_sid_for_agg_config( + ingest_state: &Arc, + config: &asap_types::aggregation_config::AggregationConfig, + point_labels: &HashMap, +) -> (u64, asap_types::PolicyFingerprint) { + let grouping_pairs: Vec<(&str, &str)> = config + .grouping_labels + .labels + .iter() + .map(|name| { + let v = point_labels.get(name).map(|s| s.as_str()).unwrap_or(""); + (name.as_str(), v) + }) + .collect(); + let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&grouping_pairs); + let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg { + agg_type: config.aggregation_type, + parameters_canonical: + crate::storage_engines::sketch_db::data::canonical_parameters(&config.parameters), + spatial_filter_canonical: config.spatial_filter_normalized.clone(), + }; + let agg_kind_canonical = agg_kind.canonical_string(); + let sid = ingest_state + .series_resolver + .resolve(&config.metric, &fp, &agg_kind_canonical); + let policy_fp = asap_types::PolicyFingerprint(config.aggregation_id()); + (sid, policy_fp) +} + async fn route_otlp_to_precompute( request: &ExportMetricsServiceRequest, ingest_state: &Arc, @@ -529,10 +581,17 @@ async fn route_otlp_to_precompute( crate::storage_engines::sketch_db::DEFAULT_RETIREMENT_RETENTION, ); - // Build (agg_id, group_key) → Vec<(series_key, ts_ms, value)> for raw points. - type GroupKey = (u64, String); + // B7.6 — bucket by `sid` instead of `(agg_id, group_key)`. The + // grouping label values fold into the sid via the + // `(metric, attrs_fp, agg_kind_canonical)` identity contract on + // `SeriesIdResolver`: same (config, grouping-label-values) → same + // sid → same bucket. We still carry `policy_fp` and `group_key` + // alongside the sid in the WorkerMessage so the worker can resolve + // the source config and render emit-time labels without consulting + // the sid → attrs reverse mapping. + type BucketTuple = (u64, asap_types::PolicyFingerprint, String); // (sid, policy_fp, group_key) type SampleTuple = (String, i64, f64); - let mut by_group: HashMap> = HashMap::new(); + let mut by_bucket: HashMap)> = HashMap::new(); let mut raw_matched = 0usize; let mut raw_unmatched = 0usize; // Schema retirement #3 (plan step #3) dropped the agg_id-keyed @@ -557,9 +616,15 @@ async fn route_otlp_to_precompute( continue; } let group_key = IngestState::extract_group_key_for(&series_key, config); - by_group - .entry((config.aggregation_id(), group_key)) - .or_default() + let (sid, policy_fp) = resolve_bucket_sid_for_agg_config( + ingest_state, + config, + &point.labels, + ); + by_bucket + .entry(sid) + .or_insert_with(|| ((sid, policy_fp, group_key.clone()), Vec::new())) + .1 .push((series_key.clone(), ts_ms, point.value)); matched = true; } @@ -571,11 +636,12 @@ async fn route_otlp_to_precompute( } flush_barrier_drops(ingest_state, &raw_barrier_drops, "otlp-raw"); - let raw_messages: Vec = by_group + let raw_messages: Vec = by_bucket .into_iter() .map( - |((agg_id, group_key), samples)| WorkerMessage::GroupSamples { - agg_id, + |(_sid, ((sid, policy_fp, group_key), samples))| WorkerMessage::GroupSamples { + sid, + policy_fp, group_key, samples, ingest_received_at, @@ -636,8 +702,22 @@ async fn route_otlp_to_precompute( continue; } }; + // B7.6 — same sid-resolution scheme as the raw path: bucket + // identity is (metric, grouping-label-values, agg_kind). + // The opaque SketchEnvelope path predates per-variant + // sketch-kind plumbing here; treat the bucket as ExactAgg + // so the resolver key matches what + // `reconcile_from_streaming_config` derives from the same + // config (otherwise the bucket would be reachable but never + // reconciled). + let (sid, policy_fp) = resolve_bucket_sid_for_agg_config( + ingest_state, + config, + &point.labels, + ); sketch_messages.push(WorkerMessage::AccumulatorInput { - agg_id: config.aggregation_id(), + sid, + policy_fp, group_key, timestamp_ms: ts_ms, accumulator, @@ -1200,15 +1280,36 @@ async fn route_modified_otlp_sketches_to_precompute( continue; } let group_key = IngestState::extract_group_key_for(&series_key, config); - // DEPRECATED: aggregation_id-keyed write — remove - // after ASAP-tier validation. The Phase 5 - // SketchStore above is the new write path; this - // legacy router push stays in tandem until the - // query path's ASAP-tier reducer is wired - // end-to-end and the streaming-config / - // SketchStore call sites can be deleted. + // B7.6 — bucket key is the per-config bucket sid + // (folds in `(metric, grouping-label-values, + // ExactAgg-of-config)`), NOT the per-DP `sid` + // resolved above. The per-DP `sid` keys the + // `SketchStore::register/append_sample` lane + // (which uses `AggKind::Sketch` to distinguish + // sketch shapes); the worker's + // group_states are keyed per-aggregation-policy + // bucket, which matches what + // `reconcile_from_streaming_config` derives + // from the same config (so retirement / orphan + // eviction stays consistent). + // + // DEPRECATED routing-side write — remove after + // ASAP-tier validation. The Phase 5 SketchStore + // above is the new write path; this legacy + // router push stays in tandem until the query + // path's ASAP-tier reducer is wired end-to-end + // and the streaming-config / SketchStore call + // sites can be deleted. + let attrs_map: HashMap = dp + .attrs + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + let (bucket_sid, policy_fp) = + resolve_bucket_sid_for_agg_config(ingest_state, config, &attrs_map); messages.push(WorkerMessage::AccumulatorInput { - agg_id: config.aggregation_id(), + sid: bucket_sid, + policy_fp, group_key, timestamp_ms: ts_ms, accumulator: accumulator.clone_boxed_core(), @@ -2511,3 +2612,238 @@ mod sid_resolution_tests { let _ = drain.await; } } + +// ── B7.6 regression: ingest bucketing is keyed by sid, not (agg_id, group_key) ── +#[cfg(test)] +mod sid_bucketing_tests { + //! Pin the B7.6 contract: ingest dispatches to the precompute engine + //! with `(sid, policy_fp, group_key)` on every `GroupSamples` / + //! `AccumulatorInput`, and distinct group_key values mint distinct + //! sids that round-trip through `SeriesIdResolver::lookup`. Tests + //! the actual `route_otlp_to_precompute` path end-to-end so a + //! refactor that drops the per-DP sid-resolve call (or routes by + //! agg_id) breaks here. + use super::*; + use crate::drivers::ingest::series_resolver::SeriesIdResolver; + use crate::precompute_engine::series_router::{SeriesRouter, WorkerMessage}; + use crate::storage_engines::sketch_db::index::SketchStore; + use crate::storage_engines::types::{HotReloadStreamingConfig, StreamingConfig}; + use asap_otel_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest; + use asap_otel_proto::tonic::common::v1::{any_value::Value as AnyVal, AnyValue, KeyValue}; + use asap_otel_proto::tonic::metrics::v1::{ + metric::Data, number_data_point::Value as NumberValue, Gauge as PbGauge, Metric as PbMetric, + NumberDataPoint, ResourceMetrics, ScopeMetrics, + }; + use asap_types::aggregation_config::AggregationConfig; + use asap_types::enums::{AggregationType, WindowType}; + use promql_utilities::data_model::key_by_label_names::KeyByLabelNames; + use std::collections::HashMap; + use std::sync::Arc; + use tokio::sync::mpsc; + + fn kv(k: &str, v: &str) -> KeyValue { + KeyValue { + key: k.to_string(), + value: Some(AnyValue { + value: Some(AnyVal::StringValue(v.to_string())), + }), + } + } + + fn sum_agg_config(metric: &str, grouping: &[&str]) -> AggregationConfig { + AggregationConfig::new( + AggregationType::SingleSubpopulation, + "Sum".to_string(), + HashMap::new(), + KeyByLabelNames::new(grouping.iter().map(|s| s.to_string()).collect()), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 10, + 10, + WindowType::Tumbling, + String::new(), + metric.to_string(), + None, + None, + None, + ) + } + + /// Build a Gauge request with one DataPoint per (zone, value) entry. + /// Distinct `zone` values are the two grouping-label buckets the + /// test inspects. + fn build_gauge_request(metric: &str, points: &[(&str, f64)]) -> ExportMetricsServiceRequest { + let data_points = points + .iter() + .map(|(zone, val)| NumberDataPoint { + attributes: vec![kv("zone", zone)], + start_time_unix_nano: 1_000_000, + time_unix_nano: 11_000_000, + value: Some(NumberValue::AsDouble(*val)), + exemplars: Vec::new(), + flags: 0, + series_id: 0, + }) + .collect(); + ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + resource: None, + scope_metrics: vec![ScopeMetrics { + scope: None, + metrics: vec![PbMetric { + name: metric.to_string(), + description: String::new(), + unit: String::new(), + metadata: Vec::new(), + data: Some(Data::Gauge(PbGauge { data_points })), + }], + schema_url: String::new(), + }], + schema_url: String::new(), + }], + } + } + + /// Two distinct `zone` values under one policy → two distinct + /// `GroupSamples` messages, each keyed by the sid the resolver + /// minted for the corresponding `(metric, zone-only-attrs, + /// ExactAgg-of-config)` tuple. + /// + /// Pre-B7.6 this path dispatched `WorkerMessage::GroupSamples { + /// agg_id, group_key, ... }` and the worker bucketed by + /// `(agg_id, group_key)`. The contract this test pins is: + /// - exactly two `WorkerMessage::GroupSamples` are emitted + /// - their sids are non-zero and distinct + /// - each sid equals what `SeriesIdResolver::lookup` records for + /// `(metric, "zone=;", ExactAgg-canonical)` — i.e. the + /// bucket identity is folded into sid via the resolver + /// - policy_fp = config.aggregation_id() on every message + /// - samples in each bucket are exactly the DPs whose `zone` + /// attribute matches that bucket (the GROUP-BY semantic) + /// + /// Note: the `group_key` field on the message currently comes from + /// `IngestState::extract_group_key_for(series_key, config)`, and + /// that helper has a pre-existing label-parsing inconsistency with + /// `format_series_key` (one quotes values, the other doesn't), so + /// it currently returns the empty string for OTLP wire-format + /// inputs. Bucketing is unaffected because B7.6 routes by sid (read + /// directly from `point.labels`, not the joined series_key); + /// fixing the group_key-extraction bug is a separate task and + /// would update emit-time label rendering, not the routing + /// contract this test pins. + #[tokio::test] + async fn raw_otlp_buckets_by_sid_with_distinct_group_keys() { + // Channel large enough to capture all routed messages without + // blocking the dispatch loop. + let (tx, mut rx) = mpsc::channel::(64); + let router = SeriesRouter::new(vec![tx]); + + let metric = "cpu_seconds"; + let cfg = sum_agg_config(metric, &["zone"]); + let policy_fp = asap_types::PolicyFingerprint(cfg.aggregation_id()); + let mut configs = HashMap::new(); + configs.insert(cfg.aggregation_id(), cfg.clone()); + let streaming = StreamingConfig::new(configs); + let hot_reload = HotReloadStreamingConfig::new(streaming); + + let resolver = Arc::new(SeriesIdResolver::new()); + let state = Arc::new(IngestState { + router, + samples_ingested: std::sync::atomic::AtomicU64::new(0), + samples_blocked_by_schema_barrier: std::sync::atomic::AtomicU64::new(0), + hot_reload_config: hot_reload, + pass_raw_samples: false, + sketch_snapshots: dashmap::DashMap::new(), + series_resolver: resolver.clone(), + sketch_index: Arc::new(SketchStore::new()), + }); + + // Two zones × two DPs each. The two zones must produce two + // separate buckets; the two DPs within a zone must accumulate + // into the same bucket. + let req = build_gauge_request( + metric, + &[("z0", 1.0), ("z0", 2.0), ("z1", 10.0), ("z1", 20.0)], + ); + + route_otlp_to_precompute(&req, &state).await; + + // Drain the messages the dispatcher emitted (one per bucket). + let mut messages: Vec = Vec::new(); + while let Ok(msg) = rx.try_recv() { + messages.push(msg); + } + + // Filter to GroupSamples — the only variant raw OTLP emits. + let groups: Vec<(u64, asap_types::PolicyFingerprint, String, Vec<(String, i64, f64)>)> = + messages + .into_iter() + .filter_map(|m| match m { + WorkerMessage::GroupSamples { + sid, + policy_fp, + group_key, + samples, + .. + } => Some((sid, policy_fp, group_key, samples)), + _ => None, + }) + .collect(); + + assert_eq!( + groups.len(), + 2, + "exactly two buckets (one per zone) — observed {} messages", + groups.len() + ); + + // Both buckets carry the same policy_fp (one source config). + for (_, pf, _, _) in &groups { + assert_eq!(*pf, policy_fp, "policy_fp must equal config.aggregation_id()"); + } + + // sids must be non-zero (zero is reserved on the wire) and distinct. + let mut sids: Vec = groups.iter().map(|(s, _, _, _)| *s).collect(); + sids.sort(); + sids.dedup(); + assert_eq!(sids.len(), 2, "two distinct sids — one per group_key"); + assert!(sids.iter().all(|&s| s != 0), "sid 0 is reserved"); + + // Each sid must match what the resolver records for its bucket + // identity: (metric, "zone=;", ExactAgg-canonical). Use + // the bucket's sample values to identify which zone it + // represents (group_key is currently empty due to the + // unrelated extract_group_key_for inconsistency — see the + // test-level doc above), then verify the sid matches the + // resolver mint for THAT zone. + let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg { + agg_type: cfg.aggregation_type, + parameters_canonical: + crate::storage_engines::sketch_db::data::canonical_parameters(&cfg.parameters), + spatial_filter_canonical: cfg.spatial_filter_normalized.clone(), + }; + let agg_kind_canonical = agg_kind.canonical_string(); + for (sid, _, _, samples) in &groups { + let mut vals: Vec = samples.iter().map(|(_, _, v)| *v).collect(); + vals.sort_by(|a, b| a.partial_cmp(b).unwrap()); + let zone_for_bucket: &str = match vals.as_slice() { + [1.0, 2.0] => "z0", + [10.0, 20.0] => "z1", + other => panic!("unexpected bucket sample values: {other:?}"), + }; + let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&[( + "zone", + zone_for_bucket, + )]); + let resolved = resolver.lookup(metric, &fp, &agg_kind_canonical); + assert_eq!( + resolved, + Some(*sid), + "sid for zone={zone_for_bucket} (inferred from sample values) must equal \ + resolver mint for (metric={metric}, fp={fp}, agg_kind={agg_kind_canonical})", + ); + } + } +} + diff --git a/data_plane/src/precompute_engine/series_router.rs b/data_plane/src/precompute_engine/series_router.rs index e4122d82..85a63d0b 100644 --- a/data_plane/src/precompute_engine/series_router.rs +++ b/data_plane/src/precompute_engine/series_router.rs @@ -1,4 +1,5 @@ use crate::storage_engines::types::AggregateCore; +use asap_types::PolicyFingerprint; use futures::future::try_join_all; use std::collections::HashMap; use std::fmt; @@ -7,6 +8,20 @@ use tokio::sync::mpsc; use xxhash_rust::xxh64::xxh64; /// A message sent from the router to a worker. +/// +/// B7.6 (schema-retirement #5): the per-group bucket key on `GroupSamples` +/// and `AccumulatorInput` is now a single `sid` (registry-allocated by +/// `SeriesIdResolver`), not the `(agg_id, group_key)` tuple. The grouping +/// label values are already folded into the sid via the +/// `(metric, attrs_fingerprint, agg_kind)` identity contract — so one sid +/// uniquely names one bucket, with no extra discriminator needed for +/// hashing or pane lookup. `group_key` and `policy_fp` still travel +/// alongside the sid: `group_key` is consumed at emit-time to render the +/// output label vector; `policy_fp` is the handle the worker uses to fetch +/// the source `AggregationConfig` from the hot-reload snapshot (window +/// shape, late-data policy, etc.). Together they let the worker key state +/// by sid without losing the data the legacy `(agg_id, group_key)` shape +/// carried. pub enum WorkerMessage { /// A batch of samples for the same series, routed by series key. /// Used in `pass_raw_samples` mode where no aggregation is needed. @@ -15,13 +30,24 @@ pub enum WorkerMessage { samples: Vec<(i64, f64)>, // (timestamp_ms, value) ingest_received_at: Instant, }, - /// A batch of samples destined for a specific aggregation group. - /// All samples share the same (agg_id, group_key) and are fed into - /// a single shared accumulator (like Arroyo's GROUP BY). + /// A batch of samples destined for a specific sid (group bucket). + /// All samples share the same `sid` and are fed into a single shared + /// accumulator (like Arroyo's GROUP BY). `sid` is the registry- + /// allocated identity for `(metric, attrs, agg_kind)`; `policy_fp` is + /// the source config's content-addressed fingerprint; `group_key` is + /// kept for emit-time label rendering. GroupSamples { - agg_id: u64, + /// Registry-allocated bucket identity. Folds in + /// `(metric, attrs_fingerprint, agg_kind_canonical)` — see + /// `SeriesIdResolver::resolve`. Worker keys `group_states` on this. + sid: u64, + /// Source `AggregationConfig` fingerprint. Worker looks up its + /// `AggregationConfig` (window size, sketch kind/config, late + /// data policy, etc.) via `snap.get_aggregation_config(policy_fp.as_u64())`. + policy_fp: PolicyFingerprint, /// Grouping label values joined by semicolons (e.g. "constant"). - /// Empty string if the aggregation has no grouping labels. + /// Empty string if the aggregation has no grouping labels. Used + /// at emit time to render the output's `KeyByLabelValues`. group_key: String, /// Each entry: (series_key, timestamp_ms, value). /// series_key is needed for keyed (MultipleSubpopulation) accumulators @@ -29,19 +55,26 @@ pub enum WorkerMessage { samples: Vec<(String, i64, f64)>, ingest_received_at: Instant, }, - /// A pre-built accumulator destined for a specific (agg_id, group_key) - /// pane. The worker merges it into that pane's existing accumulator - /// (or inserts it if the pane is empty) via `AggregateCore::merge_with`. + /// A pre-built accumulator destined for a specific sid's pane. The + /// worker merges it into that pane's existing accumulator (or + /// inserts it if the pane is empty) via `AggregateCore::merge_with`. /// /// Produced by ingest sources that deliver pre-aggregated sketches — /// e.g. the OTLP receiver when DataCollector emits KLL / CountMin / /// CountSketch payloads on a `SketchEnvelope`. Lets the precompute /// engine perform further window-aligned aggregation on sketches the /// same way it does on raw samples. + /// + /// Same sid / policy_fp / group_key contract as `GroupSamples`. AccumulatorInput { - agg_id: u64, + /// Registry-allocated bucket identity; see `GroupSamples::sid`. + sid: u64, + /// Source `AggregationConfig` fingerprint; see + /// `GroupSamples::policy_fp`. + policy_fp: PolicyFingerprint, /// Grouping label values joined by semicolons, matching the /// format produced by `IngestState::extract_group_key_for`. + /// Used at emit time to render the output's `KeyByLabelValues`. group_key: String, /// Wall-clock timestamp the sketch refers to (millis since epoch). /// Used to place the sketch into the correct pane. @@ -69,25 +102,25 @@ impl fmt::Debug for WorkerMessage { .field("sample_count", &samples.len()) .finish(), Self::GroupSamples { - agg_id, + sid, group_key, samples, .. } => f .debug_struct("GroupSamples") - .field("agg_id", agg_id) + .field("sid", sid) .field("group_key", group_key) .field("sample_count", &samples.len()) .finish(), Self::AccumulatorInput { - agg_id, + sid, group_key, timestamp_ms, accumulator, .. } => f .debug_struct("AccumulatorInput") - .field("agg_id", agg_id) + .field("sid", sid) .field("group_key", group_key) .field("timestamp_ms", timestamp_ms) .field("accumulator_type", &accumulator.type_name()) @@ -115,8 +148,10 @@ impl SeriesRouter { /// Route a pre-grouped batch of group messages to workers concurrently. /// - /// Each `GroupSamples` message is routed by `hash(agg_id, group_key)`. - /// Messages within a single worker are sent sequentially to preserve ordering. + /// Each `GroupSamples` / `AccumulatorInput` message is routed by + /// `worker_for_sid(sid)` — same `sid` always lands on the same worker, + /// so per-bucket state stays single-owner. Messages within a single + /// worker are sent sequentially to preserve ordering. pub async fn route_group_batch( &self, messages: Vec, @@ -126,12 +161,8 @@ impl SeriesRouter { let mut per_worker: HashMap> = HashMap::new(); for msg in messages { let worker_idx = match &msg { - WorkerMessage::GroupSamples { - agg_id, group_key, .. - } => self.worker_for_group(*agg_id, group_key), - WorkerMessage::AccumulatorInput { - agg_id, group_key, .. - } => self.worker_for_group(*agg_id, group_key), + WorkerMessage::GroupSamples { sid, .. } => self.worker_for_sid(*sid), + WorkerMessage::AccumulatorInput { sid, .. } => self.worker_for_sid(*sid), WorkerMessage::RawSamples { series_key, .. } => self.worker_for(series_key), _ => 0, }; @@ -181,12 +212,13 @@ impl SeriesRouter { Ok(()) } - /// Determine which worker handles a given group key. - fn worker_for_group(&self, agg_id: u64, group_key: &str) -> usize { - // Hash both agg_id and group_key together for consistent routing - let mut hash_input = agg_id.to_le_bytes().to_vec(); - hash_input.extend_from_slice(group_key.as_bytes()); - let hash = xxh64(&hash_input, 0); + /// Determine which worker handles a given sid bucket. + /// + /// Hashes the sid alone — the legacy `(agg_id, group_key)` tuple folded + /// into one u64 by `SeriesIdResolver`, so a single xxh64 over the sid + /// gives the same per-bucket sharding the tuple-hash produced. + fn worker_for_sid(&self, sid: u64) -> usize { + let hash = xxh64(&sid.to_le_bytes(), 0); (hash as usize) % self.num_workers } @@ -202,24 +234,21 @@ mod tests { use super::*; #[test] - fn test_consistent_group_routing() { + fn test_consistent_sid_routing() { let (senders, _receivers): (Vec<_>, Vec<_>) = (0..4).map(|_| mpsc::channel::(10)).unzip(); let router = SeriesRouter::new(senders); - // Same (agg_id, group_key) should always go to the same worker - let w1 = router.worker_for_group(1, "constant"); - let w2 = router.worker_for_group(1, "constant"); + // Same sid should always go to the same worker. + let w1 = router.worker_for_sid(42); + let w2 = router.worker_for_sid(42); assert_eq!(w1, w2); - // Different group keys may go to different workers - let _ = router.worker_for_group(1, "sine"); - assert!(router.worker_for_group(1, "linear-up") < 4); - - // Different agg_ids with same group key may go to different workers - let _ = router.worker_for_group(2, "constant"); - assert!(router.worker_for_group(2, "constant") < 4); + // All resolved buckets land within the worker count. + assert!(router.worker_for_sid(7) < 4); + assert!(router.worker_for_sid(99) < 4); + assert!(router.worker_for_sid(0) < 4); } #[test] diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 8c8b03d4..2d9efc28 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -10,19 +10,41 @@ use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; use asap_types::aggregation_config::AggregationConfig; use asap_types::PolicyFingerprint; use std::collections::{BTreeMap, HashMap}; +// (PolicyFingerprint is used for both `PolicyFingerprint::from_config(...)` +// on the emit path and the `policy_fp` field of GroupState below.) use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; use std::sync::Arc; use tokio::sync::mpsc; use tracing::{debug, debug_span, info, warn}; -/// Per-group aggregation state: window manager + active pane accumulators. -/// This is the equivalent of one (agg_id, group_key) in Arroyo's GROUP BY. +/// Per-bucket aggregation state: window manager + active pane accumulators. /// -/// All raw series sharing the same grouping label values feed into the same -/// accumulator, producing one output per (group_key, window) — exactly like -/// Arroyo's `GROUP BY window, key`. +/// B7.6 (schema-retirement #5): one `GroupState` per `sid`, where `sid` is +/// the registry-allocated identity for `(metric, attrs, agg_kind)`. The +/// legacy `(agg_id, group_key)` tuple folds into this single u64 — the +/// grouping label values participate in `attrs`, and the source policy +/// participates in `agg_kind`, so distinct buckets always carry distinct +/// sids. `policy_fp` and `group_key` are held here so the worker can +/// recover the source config (for window shape / late-data policy) and +/// the emit-time `KeyByLabelValues` without re-parsing the sid. +/// +/// All raw series sharing the same sid feed into the same accumulator, +/// producing one output per (sid, window) — exactly like Arroyo's +/// `GROUP BY window, key`. struct GroupState { config: Arc, + /// Source policy fingerprint that minted this sid. Held so + /// `evict_orphaned_groups` can check liveness against the streaming + /// config snapshot (a sid stays alive only while its source policy is + /// still configured), and so the worker can re-derive the + /// `PolicyFingerprint` on the emit path without a second config + /// fingerprint pass. + policy_fp: PolicyFingerprint, + /// Grouping label values joined by semicolons. Held so the emit path + /// can render the output's `KeyByLabelValues` without consulting the + /// sid → attrs reverse mapping. Format matches the input messages' + /// `group_key` field. + group_key: String, window_manager: WindowManager, /// Active panes for raw-sample accumulation, keyed by pane_start_ms. active_panes: BTreeMap>, @@ -70,18 +92,22 @@ pub struct WorkerRuntimeConfig { /// (event-time-only behaviour, matching pre-fix semantics). pub wall_clock_grace_period_ms: i64} -/// Worker that processes samples for a shard of the group space. +/// Worker that processes samples for a shard of the sid space. /// /// Unlike the old per-series design, this worker maintains accumulators -/// keyed by `(agg_id, group_key)`. Multiple raw series with the same -/// grouping label values share a single accumulator, producing one merged -/// output per window — matching Arroyo's `GROUP BY` semantics. +/// keyed by `sid` (B7.6 — was `(agg_id, group_key)`). Multiple raw series +/// with the same grouping label values share a single accumulator, +/// producing one merged output per window — matching Arroyo's `GROUP BY` +/// semantics. The grouping label values participate in the sid via the +/// `(metric, attrs_fingerprint, agg_kind_canonical)` identity contract on +/// `SeriesIdResolver`, so one sid uniquely names one bucket. pub struct Worker { id: usize, receiver: mpsc::Receiver, output_sink: Arc, - /// Map from (agg_id, group_key) to per-group state. - group_states: HashMap<(u64, String), GroupState>, + /// Map from sid to per-bucket state. One entry per active sid this + /// worker shard owns. + group_states: HashMap, /// Hot-reload handle — workers read config directly from ArcSwap /// instead of holding a local copy. All components see the same /// config at the same time. @@ -165,7 +191,8 @@ impl Worker { while let Some(msg) = self.receiver.recv().await { match msg { WorkerMessage::GroupSamples { - agg_id, + sid, + policy_fp, group_key, samples, ingest_received_at} => { @@ -173,15 +200,18 @@ impl Worker { let _span = debug_span!( "worker_process_group", worker_id = self.id, - agg_id, + sid, + policy_fp = %policy_fp, group = %group_key, sample_count, ) .entered(); - if let Err(e) = self.process_group_samples(agg_id, &group_key, samples) { + if let Err(e) = + self.process_group_samples(sid, policy_fp, &group_key, samples) + { warn!( - "Worker {} error processing group ({}, {}): {}", - self.id, agg_id, group_key, e + "Worker {} error processing sid={} (policy_fp={}, group={}): {}", + self.id, sid, policy_fp, group_key, e ); } debug!( @@ -209,7 +239,8 @@ impl Worker { ); } WorkerMessage::AccumulatorInput { - agg_id, + sid, + policy_fp, group_key, timestamp_ms, accumulator, @@ -217,21 +248,23 @@ impl Worker { let _span = debug_span!( "worker_process_accumulator", worker_id = self.id, - agg_id, + sid, + policy_fp = %policy_fp, group = %group_key, timestamp_ms, accumulator_type = accumulator.type_name(), ) .entered(); if let Err(e) = self.process_accumulator_input( - agg_id, + sid, + policy_fp, &group_key, timestamp_ms, accumulator, ) { warn!( - "Worker {} accumulator input error for ({}, {}): {}", - self.id, agg_id, group_key, e + "Worker {} accumulator input error for sid={} (policy_fp={}, group={}): {}", + self.id, sid, policy_fp, group_key, e ); } debug!( @@ -266,42 +299,57 @@ impl Worker { ); } - /// Get or create the GroupState for a (agg_id, group_key) pair. + /// Get or create the GroupState for a sid. + /// + /// B7.6 — buckets are now keyed by `sid` (a single u64) rather than + /// `(agg_id, group_key)`. `policy_fp` is the source config's + /// fingerprint, used to fetch the `AggregationConfig` from the + /// hot-reload snapshot the first time we see this sid; `group_key` is + /// remembered on the `GroupState` for emit-time label rendering. + /// /// Reads config directly from the `HotReloadStreamingConfig` - /// ArcSwap handle, so new agg_ids from a config swap are visible + /// ArcSwap handle, so new policies from a config swap are visible /// immediately — no message passing, no delay. - /// Returns None if agg_id has no matching config. + /// Returns None if `policy_fp` has no matching config (e.g. arrived + /// after the policy was retired). fn get_or_create_group_state( &mut self, - agg_id: u64, + sid: u64, + policy_fp: PolicyFingerprint, group_key: &str, ) -> Option<&mut GroupState> { - let key = (agg_id, group_key.to_string()); - if !self.group_states.contains_key(&key) { + if !self.group_states.contains_key(&sid) { let snap = self.hot_reload.snapshot(); - let cfg = snap.get_aggregation_config(agg_id)?; + let cfg = snap.get_aggregation_config(policy_fp.as_u64())?; let config = Arc::new(cfg.clone()); let gs = GroupState { window_manager: WindowManager::new(config.window_size, config.slide_interval), config, + policy_fp, + group_key: group_key.to_string(), active_panes: BTreeMap::new(), sketch_panes: BTreeMap::new(), previous_watermark_ms: i64::MIN, pane_wall_clock_starts_ms: BTreeMap::new()}; - self.group_states.insert(key.clone(), gs); + self.group_states.insert(sid, gs); self.group_count .store(self.group_states.len(), Ordering::Relaxed); } - self.group_states.get_mut(&key) + self.group_states.get_mut(&sid) } - /// Process a batch of samples for a specific (agg_id, group_key). + /// Process a batch of samples for a specific sid bucket. /// All samples in the batch feed into the same shared accumulator. /// /// This is the core of the Arroyo-equivalent GROUP BY logic. + /// B7.6 — buckets are keyed by `sid`; `policy_fp` is the source + /// `AggregationConfig` fingerprint used to resolve the bucket's + /// config on first sight; `group_key` is held on the resulting + /// `GroupState` for emit-time label rendering. pub fn process_group_samples( &mut self, - agg_id: u64, + sid: u64, + policy_fp: PolicyFingerprint, group_key: &str, samples: Vec<(String, i64, f64)>, // (series_key, timestamp_ms, value) ) -> Result<(), Box> { @@ -310,17 +358,17 @@ impl Worker { let late_data_policy = self.late_data_policy; let now_ms = (self.now_ms_fn)(); - if self.get_or_create_group_state(agg_id, group_key).is_none() { + if self + .get_or_create_group_state(sid, policy_fp, group_key) + .is_none() + { warn!( - "Worker {} skipping samples for unknown agg_id={}, group_key={}", - self.id, agg_id, group_key + "Worker {} skipping samples for unknown policy_fp={} (sid={}, group_key={})", + self.id, policy_fp, sid, group_key ); return Ok(()); } - let state = self - .group_states - .get_mut(&(agg_id, group_key.to_string())) - .unwrap(); + let state = self.group_states.get_mut(&sid).unwrap(); // Find the max timestamp in this batch to advance the watermark let batch_max_ts = samples @@ -342,8 +390,8 @@ impl Worker { // Drop late samples if previous_wm != i64::MIN && *ts < previous_wm - allowed_lateness_ms { debug!( - "Worker {} dropping late sample for group ({}, {}): ts={} watermark={}", - worker_id, agg_id, group_key, ts, previous_wm + "Worker {} dropping late sample for sid={} (group={}): ts={} watermark={}", + worker_id, sid, group_key, ts, previous_wm ); continue; } @@ -427,10 +475,10 @@ impl Worker { // Emit to output sink if !emit_batch.is_empty() { debug!( - "Worker {} emitting {} outputs for group ({}, {})", + "Worker {} emitting {} outputs for sid={} (group={})", worker_id, emit_batch.len(), - agg_id, + sid, group_key ); self.output_sink.emit_batch(emit_batch)?; @@ -440,7 +488,7 @@ impl Worker { } /// Process a pre-built accumulator (e.g. an OTLP-delivered sketch) for a - /// specific (agg_id, group_key) pane. + /// specific sid bucket's pane. /// /// The incoming accumulator is merged into `sketch_panes[pane_start]` via /// `AggregateCore::merge_with`. If the pane is empty the accumulator is @@ -451,9 +499,13 @@ impl Worker { /// Unlike `process_group_samples`, this path does not touch /// `active_panes` — sketches live in their own pane map and get merged /// at window close (see `merge_sketch_panes_for_window`). + /// + /// `policy_fp` / `group_key` carry the same semantics as on + /// `process_group_samples` — policy lookup + emit-time label rendering. pub fn process_accumulator_input( &mut self, - agg_id: u64, + sid: u64, + policy_fp: PolicyFingerprint, group_key: &str, timestamp_ms: i64, incoming: Box, @@ -463,17 +515,17 @@ impl Worker { let late_data_policy = self.late_data_policy; let now_ms = (self.now_ms_fn)(); - if self.get_or_create_group_state(agg_id, group_key).is_none() { + if self + .get_or_create_group_state(sid, policy_fp, group_key) + .is_none() + { warn!( - "Worker {} skipping accumulator input for unknown agg_id={}, group_key={}", - self.id, agg_id, group_key + "Worker {} skipping accumulator input for unknown policy_fp={} (sid={}, group_key={})", + self.id, policy_fp, sid, group_key ); return Ok(()); } - let state = self - .group_states - .get_mut(&(agg_id, group_key.to_string())) - .unwrap(); + let state = self.group_states.get_mut(&sid).unwrap(); let previous_wm = state.previous_watermark_ms; let current_wm = if timestamp_ms > previous_wm { @@ -494,8 +546,8 @@ impl Worker { match late_data_policy { LateDataPolicy::Drop => { debug!( - "Worker {} dropping late accumulator input for group ({}, {}): ts={} watermark={}", - worker_id, agg_id, group_key, timestamp_ms, previous_wm + "Worker {} dropping late accumulator input for sid={} (group={}): ts={} watermark={}", + worker_id, sid, group_key, timestamp_ms, previous_wm ); } LateDataPolicy::ForwardToStore => { @@ -582,10 +634,10 @@ impl Worker { if !emit_batch.is_empty() { debug!( - "Worker {} emitting {} sketch outputs for group ({}, {})", + "Worker {} emitting {} sketch outputs for sid={} (group={})", worker_id, emit_batch.len(), - agg_id, + sid, group_key ); self.output_sink.emit_batch(emit_batch)?; @@ -644,30 +696,35 @@ impl Worker { /// 3. Compute global watermark = min(all worker watermarks) /// 4. Advance idle groups to the global watermark, closing due windows /// - /// Remove GroupStates whose agg_id is no longer in the current - /// config (i.e. the control plane removed the aggregation). Groups - /// with non-empty panes are kept until flush_all closes their - /// windows; once both pane maps are empty, the GroupState shell - /// is freed. + /// Remove GroupStates whose source policy is no longer in the + /// current config (i.e. the control plane removed the + /// aggregation). Liveness is checked against each bucket's stored + /// `policy_fp` — a sid stays alive only while its minting policy is + /// still configured. Buckets with non-empty panes are kept until + /// flush_all closes their windows; once both pane maps are empty, + /// the GroupState shell is freed. fn evict_orphaned_groups(&mut self) { let snap = self.hot_reload.snapshot(); let before = self.group_states.len(); - self.group_states.retain(|&(agg_id, _), gs| { - if snap.contains(agg_id) { - return true; // still in config, keep + self.group_states.retain(|&sid, gs| { + if snap.contains(gs.policy_fp.as_u64()) { + return true; // policy still in config, keep } - // Not in config — keep only if there's residual data + // Policy retired — keep only if there's residual data // that flush_all hasn't drained yet. let has_data = !gs.active_panes.is_empty() || !gs.sketch_panes.is_empty(); if !has_data { - debug!("evicting orphaned group (agg_id={})", agg_id); + debug!( + "evicting orphaned bucket (sid={}, policy_fp={})", + sid, gs.policy_fp + ); } has_data }); let after = self.group_states.len(); if before != after { info!( - "Worker {} evicted {} orphaned groups ({} → {})", + "Worker {} evicted {} orphaned buckets ({} → {})", self.id, before - after, before, @@ -700,13 +757,19 @@ impl Worker { // Step 3: Compute global watermark = min(all worker watermarks). let global_wm = self.compute_global_watermark(); - // Step 4: For each group, advance watermark and close due windows. + // Step 4: For each bucket, advance watermark and close due windows. let mut emit_batch: Vec<(PrecomputedOutput, Box)> = Vec::new(); - for ((agg_id, group_key), state) in &mut self.group_states { + for (&sid, state) in &mut self.group_states { + let _ = sid; // sid is the bucket key; group_key/policy_fp live on `state` if state.previous_watermark_ms == i64::MIN { continue; // No samples received yet — no panes to close. } + // group_key/policy_fp travelled in on the message and are + // stored on `state` so the emit path can reach them without + // re-keying the bucket. Clone so the body below can borrow + // `state` mutably for pane drains. + let group_key = state.group_key.clone(); // Effective watermark: max(group's own, global) + 1ms for boundary. let propagated_wm = if global_wm != i64::MIN { @@ -750,7 +813,7 @@ impl Worker { if let Some(accumulator) = merge_panes_for_window(&mut state.active_panes, &pane_starts) { - let key = build_group_key_label_values(group_key); + let key = build_group_key_label_values(&group_key); let output = PrecomputedOutput::new( *window_start as u64, window_end as u64, @@ -763,7 +826,7 @@ impl Worker { if let Some(accumulator) = merge_sketch_panes_for_window(&mut state.sketch_panes, &pane_starts) { - let key = build_group_key_label_values(group_key); + let key = build_group_key_label_values(&group_key); let output = PrecomputedOutput::new( *window_start as u64, window_end as u64, @@ -1247,21 +1310,22 @@ mod tests { let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); // Samples in window [0, 10000ms): sum should be 1+2+3=6. - // All go to the same group (agg_id=1, group_key="") + // All go to the same bucket (sid=1, group_key="") + let pf = PolicyFingerprint(1); worker - .process_group_samples(1, "", group_samples("cpu", vec![(1000, 1.0)])) + .process_group_samples(1, pf, "", group_samples("cpu", vec![(1000, 1.0)])) .unwrap(); worker - .process_group_samples(1, "", group_samples("cpu", vec![(5000, 2.0)])) + .process_group_samples(1, pf, "", group_samples("cpu", vec![(5000, 2.0)])) .unwrap(); worker - .process_group_samples(1, "", group_samples("cpu", vec![(9000, 3.0)])) + .process_group_samples(1, pf, "", group_samples("cpu", vec![(9000, 3.0)])) .unwrap(); assert_eq!(sink.len(), 0); // Sample at t=10000ms closes [0, 10000) worker - .process_group_samples(1, "", group_samples("cpu", vec![(10000, 100.0)])) + .process_group_samples(1, pf, "", group_samples("cpu", vec![(10000, 100.0)])) .unwrap(); let captured = sink.drain(); @@ -1309,11 +1373,13 @@ mod tests { let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); - // Two different series, same group (agg_id=1, group_key="") + // Two different series, same bucket (sid=1, group_key="") // Both feed into the same accumulator + let pf = PolicyFingerprint(1); worker .process_group_samples( 1, + pf, "", vec![ ("cpu{host=\"A\"}".to_string(), 1000, 10.0), @@ -1325,7 +1391,7 @@ mod tests { // Close the window worker - .process_group_samples(1, "", group_samples("cpu{host=\"A\"}", vec![(10000, 0.0)])) + .process_group_samples(1, pf, "", group_samples("cpu{host=\"A\"}", vec![(10000, 0.0)])) .unwrap(); let captured = sink.drain(); @@ -1371,34 +1437,43 @@ mod tests { let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); - // Group "constant" gets samples + // Two distinct group_keys → two distinct sids (sid IS the bucket + // identity; the legacy `(agg_id, group_key)` tuple folds in). + let pf = PolicyFingerprint(1); + let sid_constant = 11_u64; + let sid_sine = 12_u64; + // Bucket sid_constant gets samples worker .process_group_samples( - 1, + sid_constant, + pf, "constant", group_samples("cpu{pattern=\"constant\"}", vec![(1000, 5.0)]), ) .unwrap(); - // Group "sine" gets samples + // Bucket sid_sine gets samples worker .process_group_samples( - 1, + sid_sine, + pf, "sine", group_samples("cpu{pattern=\"sine\"}", vec![(2000, 7.0)]), ) .unwrap(); - // Close both groups' windows + // Close both buckets' windows worker .process_group_samples( - 1, + sid_constant, + pf, "constant", group_samples("cpu{pattern=\"constant\"}", vec![(10000, 0.0)]), ) .unwrap(); worker .process_group_samples( - 1, + sid_sine, + pf, "sine", group_samples("cpu{pattern=\"sine\"}", vec![(10000, 0.0)]), ) @@ -1442,9 +1517,11 @@ mod tests { let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); // Three different series all in group "constant" — all feed one KLL + let pf = PolicyFingerprint(1); worker .process_group_samples( 1, + pf, "constant", vec![ ( @@ -1470,6 +1547,7 @@ mod tests { worker .process_group_samples( 1, + pf, "constant", group_samples( "latency{pattern=\"constant\",host=\"a\"}", @@ -1520,15 +1598,16 @@ mod tests { let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); // Sample at t=15000ms → goes to pane 10000ms + let pf = PolicyFingerprint(2); worker - .process_group_samples(2, "", group_samples("cpu", vec![(15_000, 42.0)])) + .process_group_samples(2, pf, "", group_samples("cpu", vec![(15_000, 42.0)])) .unwrap(); assert_eq!(sink.len(), 0); // Sample at t=45000ms → advances watermark to 45000ms // Closes windows [0, 30000) and [10000, 40000) worker - .process_group_samples(2, "", group_samples("cpu", vec![(45_000, 0.0)])) + .process_group_samples(2, pf, "", group_samples("cpu", vec![(45_000, 0.0)])) .unwrap(); let captured = sink.drain(); @@ -1581,11 +1660,13 @@ mod tests { let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); - // Both series go to the SAME group (group_key="" since grouping is empty). + // Both series go to the SAME bucket (group_key="" since grouping is empty). // The host label is extracted as the aggregated key inside the accumulator. + let pf = PolicyFingerprint(3); worker .process_group_samples( 3, + pf, "", vec![ ("cpu{host=\"A\"}".to_string(), 1000, 10.0), @@ -1594,9 +1675,9 @@ mod tests { ) .unwrap(); - // Close the single group's window + // Close the single bucket's window worker - .process_group_samples(3, "", group_samples("cpu{host=\"A\"}", vec![(10000, 0.0)])) + .process_group_samples(3, pf, "", group_samples("cpu{host=\"A\"}", vec![(10000, 0.0)])) .unwrap(); let captured = sink.drain(); @@ -1667,14 +1748,15 @@ mod tests { ); // Establish watermark at t=20000ms + let pf = PolicyFingerprint(4); worker - .process_group_samples(4, "", group_samples("cpu", vec![(20_000, 1.0)])) + .process_group_samples(4, pf, "", group_samples("cpu", vec![(20_000, 1.0)])) .unwrap(); let _ = sink.drain(); // Send a late sample worker - .process_group_samples(4, "", group_samples("cpu", vec![(5_000, 99.0)])) + .process_group_samples(4, pf, "", group_samples("cpu", vec![(5_000, 99.0)])) .unwrap(); assert_eq!(sink.len(), 0, "late sample should be dropped"); @@ -1719,17 +1801,18 @@ mod tests { ); // Seed then advance watermark to 20000 + let pf = PolicyFingerprint(5); worker - .process_group_samples(5, "", group_samples("cpu", vec![(500, 1.0)])) + .process_group_samples(5, pf, "", group_samples("cpu", vec![(500, 1.0)])) .unwrap(); worker - .process_group_samples(5, "", group_samples("cpu", vec![(20_000, 0.0)])) + .process_group_samples(5, pf, "", group_samples("cpu", vec![(20_000, 0.0)])) .unwrap(); let _ = sink.drain(); // Send late sample for evicted pane worker - .process_group_samples(5, "", group_samples("cpu", vec![(8_000, 55.0)])) + .process_group_samples(5, pf, "", group_samples("cpu", vec![(8_000, 55.0)])) .unwrap(); let captured = sink.drain(); @@ -1790,19 +1873,21 @@ aggregations: let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); + let pf = PolicyFingerprint(agg_id); + let sid = 1_u64; worker - .process_group_samples(agg_id, "", group_samples("requests_total", vec![(1_000, 3.0)])) + .process_group_samples(sid, pf, "", group_samples("requests_total", vec![(1_000, 3.0)])) .unwrap(); worker - .process_group_samples(agg_id, "", group_samples("requests_total", vec![(5_000, 4.0)])) + .process_group_samples(sid, pf, "", group_samples("requests_total", vec![(5_000, 4.0)])) .unwrap(); worker - .process_group_samples(agg_id, "", group_samples("requests_total", vec![(9_000, 5.0)])) + .process_group_samples(sid, pf, "", group_samples("requests_total", vec![(9_000, 5.0)])) .unwrap(); assert_eq!(sink.len(), 0); worker - .process_group_samples(agg_id, "", group_samples("requests_total", vec![(10_000, 0.0)])) + .process_group_samples(sid, pf, "", group_samples("requests_total", vec![(10_000, 0.0)])) .unwrap(); let captured = sink.drain(); @@ -1889,19 +1974,25 @@ aggregations: let sink = Arc::new(CapturingOutputSink::new()); let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); + // Two distinct group_keys ("groupA" / "groupB") under the same + // policy → two distinct sids (each sid is one bucket; the legacy + // `(agg_id, group_key)` tuple folded into a single u64). + let pf = PolicyFingerprint(1); + let sid_a = 21_u64; + let sid_b = 22_u64; // Group A: send sample at t=5s (within window [0, 10s)) worker - .process_group_samples(1, "groupA", group_samples("cpu", vec![(5_000, 1.0)])) + .process_group_samples(sid_a, pf, "groupA", group_samples("cpu", vec![(5_000, 1.0)])) .unwrap(); // Group B: send sample at t=5s (within window [0, 10s)) worker - .process_group_samples(1, "groupB", group_samples("cpu", vec![(5_000, 2.0)])) + .process_group_samples(sid_b, pf, "groupB", group_samples("cpu", vec![(5_000, 2.0)])) .unwrap(); let _ = sink.drain(); // Advance group A's watermark to t=100s (closes many windows). worker - .process_group_samples(1, "groupA", group_samples("cpu", vec![(100_000, 3.0)])) + .process_group_samples(sid_a, pf, "groupA", group_samples("cpu", vec![(100_000, 3.0)])) .unwrap(); let _ = sink.drain(); @@ -2052,8 +2143,9 @@ aggregations: assert_eq!(wm.load(Ordering::Acquire), i64::MIN); // Send data at t=50s + let pf = PolicyFingerprint(1); worker - .process_group_samples(1, "", group_samples("cpu", vec![(50_000, 1.0)])) + .process_group_samples(1, pf, "", group_samples("cpu", vec![(50_000, 1.0)])) .unwrap(); // Flush should publish worker watermark @@ -2118,12 +2210,15 @@ aggregations: let mut worker = make_worker(agg_configs, sink.clone(), false, 0, LateDataPolicy::Drop); // First batch: 10 sketches at t=60_000 ms, all under the same - // group_key="us-east" — mirrors the agent emitting one sketch per - // (zone,rack,node,pod) tuple while the backend rolls them up by zone. + // bucket (group_key="us-east") — mirrors the agent emitting one + // sketch per (zone,rack,node,pod) tuple while the backend rolls + // them up by zone. + let pf = PolicyFingerprint(1); + let sid = 31_u64; for i in 0..10 { let s = make_ddsketch(0.01, &[1.0 + i as f64, 2.0, 3.0]); worker - .process_accumulator_input(1, "us-east", 60_000, Box::new(s)) + .process_accumulator_input(sid, pf, "us-east", 60_000, Box::new(s)) .expect("first batch must process"); } assert_eq!( @@ -2139,7 +2234,7 @@ aggregations: // and the output is emitted. let s2 = make_ddsketch(0.01, &[5.0, 6.0]); worker - .process_accumulator_input(1, "us-east", 120_000, Box::new(s2)) + .process_accumulator_input(sid, pf, "us-east", 120_000, Box::new(s2)) .expect("second batch must process"); let captured = sink.drain(); @@ -2209,29 +2304,33 @@ aggregations: // Three sketches in the SAME zone but different (rack,node,pod) // tuples — emulating what the agent ships. Group key the ingest - // path computes is the zone value alone. + // path computes is the zone value alone; distinct group_keys + // (us-east vs us-west) get distinct sids. + let pf = PolicyFingerprint(1); + let sid_east = 41_u64; + let sid_west = 42_u64; for i in 0..3 { let s = make_ddsketch(0.01, &[100.0 + i as f64]); worker - .process_accumulator_input(1, "us-east", 60_000, Box::new(s)) + .process_accumulator_input(sid_east, pf, "us-east", 60_000, Box::new(s)) .unwrap(); } // Two sketches in a different zone. for i in 0..2 { let s = make_ddsketch(0.01, &[200.0 + i as f64]); worker - .process_accumulator_input(1, "us-west", 60_000, Box::new(s)) + .process_accumulator_input(sid_west, pf, "us-west", 60_000, Box::new(s)) .unwrap(); } // Advance the watermark past 90_000 to close window [60_000, 90_000). let s = make_ddsketch(0.01, &[1.0]); worker - .process_accumulator_input(1, "us-east", 120_000, Box::new(s)) + .process_accumulator_input(sid_east, pf, "us-east", 120_000, Box::new(s)) .unwrap(); let s = make_ddsketch(0.01, &[1.0]); worker - .process_accumulator_input(1, "us-west", 120_000, Box::new(s)) + .process_accumulator_input(sid_west, pf, "us-west", 120_000, Box::new(s)) .unwrap(); let captured = sink.drain(); @@ -2348,10 +2447,12 @@ aggregations: worker.set_now_ms_fn(Box::new(move || wc_clone.load(Ordering::Relaxed))); // Ingest 10 sketches all stamped at frozen event-time t_event=0. + let pf = PolicyFingerprint(1); + let sid = 51_u64; for i in 0..10 { let s = make_ddsketch(0.01, &[1.0 + i as f64]); worker - .process_accumulator_input(1, "us-east", 0, Box::new(s)) + .process_accumulator_input(sid, pf, "us-east", 0, Box::new(s)) .expect("ingest must accept frozen-event-time sketches"); } assert_eq!( @@ -2452,7 +2553,7 @@ aggregations: let s = make_ddsketch(0.01, &[42.0]); worker - .process_accumulator_input(1, "us-east", 0, Box::new(s)) + .process_accumulator_input(1, PolicyFingerprint(1), "us-east", 0, Box::new(s)) .unwrap(); // Even after a wall-clock eternity, no emit happens with