From dd85fce06a5647cd475c199265047112fe37dc30 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 16 May 2026 08:23:17 -0600 Subject: [PATCH 1/2] refactor(backend): drop row_num/col_num legacy fallback from cms_params MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-user direction: now that the asapcollector configs migrated to canonical `w` / `d` keys (sibling PR `sync-config-canonical-w-d`), the `row_num` / `col_num` fallback PR #268 added has no remaining caller. Drop it. Three sweeps: - `accumulator_factory::cms_params` — canonical keys only; `cms_params_reads_canonical_w_d_keys` test updated to drop the legacy half. - `storage_engines::sketch_db::accuracy::cms_params` — same. - `tests::accuracy_empirical_validation_tests` — six test fixtures migrated from `row_num` / `col_num` to `d` / `w` to match the helper's new contract. Remaining `row_num` / `col_num` mentions in the codebase are all internal Rust identifiers (function arguments / struct fields / locals describing the sketch matrix dimensions) — not config- parameter keys. Those are idiomatic and stay. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../precompute_engine/accumulator_factory.rs | 46 +++++------------ .../src/storage_engines/sketch_db/accuracy.rs | 49 +++++++++---------- .../accuracy_empirical_validation_tests.rs | 32 ++++++------ 3 files changed, 49 insertions(+), 78 deletions(-) diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index 0b38f727..5d7b6b30 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -650,23 +650,21 @@ fn kll_k_param(config: &AggregationConfig) -> u16 { /// Extract `(row_num, col_num)` for CMS / HydraKLL configs. /// -/// Reads from the canonical `d` (depth = rows) / `w` (width = cols) -/// keys first — these match what the control plane's -/// `sketch_params_to_json` emits and what `sketch_config_to_params` -/// uses for OTLP policy_fp content matching. Falls back to the -/// legacy `row_num` / `col_num` keys so older asapcollector -/// configs continue to work. +/// Reads canonical `d` (depth = rows) / `w` (width = cols) keys — +/// matches what the control plane's `sketch_params_to_json` emits +/// and what `sketch_config_to_params` uses for OTLP policy_fp +/// content matching. The legacy `row_num` / `col_num` form (the +/// only pre-PR-268 reader) was retired in lock-step with the +/// asapcollector migration to canonical keys. fn cms_params(config: &AggregationConfig) -> (usize, usize) { let row_num = config .parameters .get("d") - .or_else(|| config.parameters.get("row_num")) .and_then(|v| v.as_u64()) .unwrap_or(4) as usize; let col_num = config .parameters .get("w") - .or_else(|| config.parameters.get("col_num")) .and_then(|v| v.as_u64()) .unwrap_or(1000) as usize; (row_num, col_num) @@ -979,10 +977,13 @@ mod tests { } #[test] - fn cms_params_accepts_w_d_canonical_and_row_col_num_legacy() { + fn cms_params_reads_canonical_w_d_keys() { use std::collections::HashMap; // Canonical `w`/`d` form — what the control plane's - // `sketch_params_to_json` emits today. + // `sketch_params_to_json` emits and what asapcollector + // streaming-config YAMLs ship (asapcollector PR + // `sync-config-canonical-w-d` migrated them in lock-step + // with the legacy-fallback removal). let mut params = HashMap::new(); params.insert("d".to_string(), serde_json::Value::from(7_u64)); params.insert("w".to_string(), serde_json::Value::from(2048_u64)); @@ -1005,31 +1006,6 @@ mod tests { ); assert_eq!(super::cms_params(&config), (7, 2048)); - // Legacy `row_num`/`col_num` form — older asapcollector - // configs still ship these; the helper must keep accepting - // them so existing Docker e2e setups don't silently degrade. - let mut legacy = HashMap::new(); - legacy.insert("row_num".to_string(), serde_json::Value::from(7_u64)); - legacy.insert("col_num".to_string(), serde_json::Value::from(2048_u64)); - let legacy_config = AggregationConfig::new( - AggregationType::CountMinSketch, - String::new(), - legacy, - promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), - promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), - promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), - String::new(), - 60, - 0, - WindowType::Tumbling, - "m".to_string(), - "m".to_string(), - None, - None, - None, - ); - assert_eq!(super::cms_params(&legacy_config), (7, 2048)); - // Empty params — defaults `(4, 1000)`. let empty_config = AggregationConfig::new( AggregationType::CountMinSketch, diff --git a/data_plane/src/storage_engines/sketch_db/accuracy.rs b/data_plane/src/storage_engines/sketch_db/accuracy.rs index 6b79c115..74e1d7c1 100644 --- a/data_plane/src/storage_engines/sketch_db/accuracy.rs +++ b/data_plane/src/storage_engines/sketch_db/accuracy.rs @@ -138,11 +138,10 @@ impl AccuracyProfile { // CountMinSketch: classic Cormode-Muthukrishnan bound. // ε = e/w, δ = 1/2^d with w = width, d = depth. We - // pull w from `parameters["col_num"]` and d from - // `parameters["row_num"]` because that's how the - // existing `accumulator_factory::cms_params` names - // them; fall back to the factory's (rows=4, cols=1000) - // defaults if absent. + // pull w from `parameters["w"]` and d from + // `parameters["d"]` — the canonical keys the controller + // emits and `accumulator_factory::cms_params` reads. + // Defaults to (rows=4, cols=1000) when absent. AggregationType::CountMinSketch => { let (rows, cols) = cms_params(config); // Using natural e ≈ 2.71828 for tighter bound. @@ -301,23 +300,19 @@ impl AccuracyProfile { // says it's OK for them to drift — this module is the single // authority on *accuracy*, not on *construction*. -/// Reads from the canonical `d` (depth = rows) / `w` (width = cols) -/// keys first — these match what the control plane's -/// `sketch_params_to_json` emits and what -/// `accumulator_factory::cms_params` reads. Falls back to the legacy -/// `row_num` / `col_num` keys so older asapcollector configs -/// continue to work. +/// Reads canonical `d` (depth = rows) / `w` (width = cols) keys. +/// The legacy `row_num` / `col_num` form was retired in lock-step +/// with the asapcollector migration to canonical keys — see +/// `accumulator_factory::cms_params` for the matching change. fn cms_params(config: &AggregationConfig) -> (u64, u64) { let rows = config .parameters .get("d") - .or_else(|| config.parameters.get("row_num")) .and_then(|v| v.as_u64()) .unwrap_or(4); let cols = config .parameters .get("w") - .or_else(|| config.parameters.get("col_num")) .and_then(|v| v.as_u64()) .unwrap_or(1000); (rows, cols) @@ -499,8 +494,8 @@ mod tests { #[test] fn cms_epsilon_is_e_over_w() { let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(5)); - params.insert("col_num".to_string(), json!(2718)); + params.insert("d".to_string(), json!(5)); + params.insert("w".to_string(), json!(2718)); let p = AccuracyProfile::derive(&base_config(AggregationType::CountMinSketch, params)); // e / 2718 ≈ 0.0010001 — very close to 0.001. assert_eq!(p.kind, AccuracyKind::AdditiveFrequency); @@ -525,8 +520,8 @@ mod tests { // Large heap: 1/heap_size (= 1e-4) dominates the e/w CMS // bound (e/1e6 ≈ 2.72e-6). Expect ε = 1/heap. let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(5)); - params.insert("col_num".to_string(), json!(1_000_000)); + params.insert("d".to_string(), json!(5)); + params.insert("w".to_string(), json!(1_000_000)); params.insert("heap_size".to_string(), json!(10_000)); let p = AccuracyProfile::derive(&base_config( AggregationType::CountMinSketchWithHeap, @@ -542,8 +537,8 @@ mod tests { // Generously-sized heap (1e6) + narrow CMS (w=100) → // 1/heap (1e-6) ≪ e/w (2.7e-2), so the CMS bound dominates. let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4)); - params.insert("col_num".to_string(), json!(100)); + params.insert("d".to_string(), json!(4)); + params.insert("w".to_string(), json!(100)); params.insert("heap_size".to_string(), json!(1_000_000)); let p = AccuracyProfile::derive(&base_config( AggregationType::CountMinSketchWithHeap, @@ -556,8 +551,8 @@ mod tests { #[test] fn cms_with_heap_uses_default_heap_size_100() { let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4)); - params.insert("col_num".to_string(), json!(1000)); + params.insert("d".to_string(), json!(4)); + params.insert("w".to_string(), json!(1000)); // heap_size absent → default 100 → 1/100 = 0.01 dominates // e/1000 ≈ 0.00272. let p = AccuracyProfile::derive(&base_config( @@ -574,8 +569,8 @@ mod tests { // "heap_size" — all three should work. for alias in ["heap_size", "topk", "k"] { let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4)); - params.insert("col_num".to_string(), json!(1_000_000)); + params.insert("d".to_string(), json!(4)); + params.insert("w".to_string(), json!(1_000_000)); params.insert(alias.to_string(), json!(500)); let p = AccuracyProfile::derive(&base_config( AggregationType::CountMinSketchWithHeap, @@ -592,8 +587,8 @@ mod tests { #[test] fn countsketch_epsilon_is_one_over_sqrt_w() { let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4)); - params.insert("col_num".to_string(), json!(100)); + params.insert("d".to_string(), json!(4)); + params.insert("w".to_string(), json!(100)); let p = AccuracyProfile::derive(&base_config(AggregationType::CountSketch, params)); assert_eq!(p.kind, AccuracyKind::AdditiveFrequency); assert!((p.epsilon - 0.1).abs() < 1e-9); // 1/√100 = 0.1 @@ -691,8 +686,8 @@ mod tests { // i.e. CountSketch is a factor ~1.65 tighter than CMS on // epsilon alone. Confirms the bounds are not copy-pasted. let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4)); - params.insert("col_num".to_string(), json!(10000)); + params.insert("d".to_string(), json!(4)); + params.insert("w".to_string(), json!(10000)); let cms = AccuracyProfile::derive(&base_config( AggregationType::CountMinSketch, params.clone(), diff --git a/data_plane/src/tests/accuracy_empirical_validation_tests.rs b/data_plane/src/tests/accuracy_empirical_validation_tests.rs index 64306d18..72466b5c 100644 --- a/data_plane/src/tests/accuracy_empirical_validation_tests.rs +++ b/data_plane/src/tests/accuracy_empirical_validation_tests.rs @@ -73,8 +73,8 @@ fn cms_3x1000_matches_cormode_muthukrishnan_e_over_w() { // Cormode-Muthukrishnan 2005: ε = e/w, δ = 1/2^d. // For (w=1000, d=3): ε = e/1000 ≈ 2.71828e-3, δ = 0.125. let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(3u64)); - params.insert("col_num".to_string(), json!(1000u64)); + params.insert("d".to_string(), json!(3u64)); + params.insert("w".to_string(), json!(1000u64)); let p = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketch, params)); assert_eq!(p.kind, AccuracyKind::AdditiveFrequency); assert!((p.epsilon - std::f64::consts::E / 1000.0).abs() < 1e-12); @@ -86,8 +86,8 @@ fn count_sketch_4x10000_matches_charikar_one_over_sqrt_w() { // Charikar-Chen-Farach-Colton 2002: ε = 1/√w for signed // counter sketch. let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4u64)); - params.insert("col_num".to_string(), json!(10_000u64)); + params.insert("d".to_string(), json!(4u64)); + params.insert("w".to_string(), json!(10_000u64)); let p = AccuracyProfile::derive(&cfg(AggregationType::CountSketch, params)); assert_eq!(p.kind, AccuracyKind::AdditiveFrequency); assert!((p.epsilon - 0.01).abs() < 1e-12); @@ -123,8 +123,8 @@ fn cms_with_heap_top_k_combines_cms_and_retention_bounds() { // (w=1000, d=3, heap=50): CMS bound e/w ≈ 2.72e-3, heap // bound 1/50 = 0.02. Heap dominates → ε = 0.02. let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(3u64)); - params.insert("col_num".to_string(), json!(1000u64)); + params.insert("d".to_string(), json!(3u64)); + params.insert("w".to_string(), json!(1000u64)); params.insert("heap_size".to_string(), json!(50u64)); let p = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketchWithHeap, params)); assert_eq!(p.kind, AccuracyKind::TopK); @@ -154,8 +154,8 @@ fn cms_epsilon_shrinks_monotonically_with_width() { let mut last = f64::INFINITY; for w in [100u64, 500, 2000, 10_000, 100_000] { let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4u64)); - params.insert("col_num".to_string(), json!(w)); + params.insert("d".to_string(), json!(4u64)); + params.insert("w".to_string(), json!(w)); let eps = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketch, params)).epsilon; assert!(eps < last, "CMS ε at w={w}: {eps} should be < {last}"); last = eps; @@ -167,8 +167,8 @@ fn cms_delta_shrinks_monotonically_with_depth() { let mut last = f64::INFINITY; for d in [2u64, 3, 4, 5, 6, 8] { let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(d)); - params.insert("col_num".to_string(), json!(1000u64)); + params.insert("d".to_string(), json!(d)); + params.insert("w".to_string(), json!(1000u64)); let delta = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketch, params)).delta; assert!(delta < last, "CMS δ at d={d}: {delta} should be < {last}"); last = delta; @@ -180,8 +180,8 @@ fn countsketch_epsilon_shrinks_monotonically_with_width() { let mut last = f64::INFINITY; for w in [100u64, 500, 2000, 10_000, 100_000] { let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4u64)); - params.insert("col_num".to_string(), json!(w)); + params.insert("d".to_string(), json!(4u64)); + params.insert("w".to_string(), json!(w)); let eps = AccuracyProfile::derive(&cfg(AggregationType::CountSketch, params)).epsilon; assert!( eps < last, @@ -229,8 +229,8 @@ fn cms_with_heap_epsilon_shrinks_monotonically_with_heap_size_when_heap_dominate let mut last = f64::INFINITY; for heap in [10u64, 100, 1000, 10_000, 100_000] { let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(5u64)); - params.insert("col_num".to_string(), json!(1_000_000u64)); + params.insert("d".to_string(), json!(5u64)); + params.insert("w".to_string(), json!(1_000_000u64)); params.insert("heap_size".to_string(), json!(heap)); let eps = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketchWithHeap, params)).epsilon; @@ -252,8 +252,8 @@ fn relative_ordering_of_bounds_matches_published_intuition() { // `countsketch_oxide_matches_cms_oxide` sanity in // sketch-bench. let mut params = HashMap::new(); - params.insert("row_num".to_string(), json!(4u64)); - params.insert("col_num".to_string(), json!(10_000u64)); + params.insert("d".to_string(), json!(4u64)); + params.insert("w".to_string(), json!(10_000u64)); let cms = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketch, params.clone())); let cs = AccuracyProfile::derive(&cfg(AggregationType::CountSketch, params)); // CMS at w=10_000: e/10_000 ≈ 0.000272 From 84efa968431bd6eaba898d8951d779c5b3fb9303 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 17 May 2026 10:03:54 -0600 Subject: [PATCH 2/2] =?UTF-8?q?fix(deploy):=20MVP=20smoke-test=20blockers?= =?UTF-8?q?=20=E2=80=94=20dockerignore=20+=20OpAMP=20endpoint=20default?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two deploy-side bugs found while bringing up the single-node MVP demo: 1. .dockerignore: top-level `target/` rule doesn't match subdirectory `target/` paths, so per-crate caches under `data_plane/target/` and `control_plane/target/` (~700 MB after any host-side `cargo build`) get shipped into the BuildKit context. The first `docker build -f Dockerfile.backend` hit over 7 GB of `transferring backend-src:` before BuildKit canceled. Add `**/target/`, `**/.git/`, editor caches, and `**/eval-results/` so the context shrinks back to source-only. 2. control_plane: default `CONTROLLER_OPAMP_ENDPOINT` is `ws://control_plane:4320/v1/opamp` (the post-reorg crate name), but the canonical compose stack in `ASAPCollector/deploy/mvp-singlenode/docker-compose/base.yml` still names the service `controller`. The endpoint string gets baked into every emitted agent yaml under `extensions.opamp.server.ws.endpoint`, so the broken default tears down OpAMP the moment the controller pushes a fresh config. Switch the default to `ws://controller:4320/v1/opamp`. Co-Authored-By: Claude Opus 4.7 (1M context) --- .dockerignore | 17 +++++++++++++++++ control_plane/src/main.rs | 11 ++++++++++- 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/.dockerignore b/.dockerignore index 3ea08526..38560d2c 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,2 +1,19 @@ target/ +**/target/ .git/ +**/.git/ +# Workspace per-crate caches (control_plane/target, data_plane/target, +# benchmarks/.../target) that aren't covered by the top-level `target/` +# pattern. Excluding them shrinks the BuildKit context transfer for +# `docker build -f deploy/docker/Dockerfile.backend` from multi-GB +# (every host-side cargo build leaves a few-hundred-MB target there) +# back down to the source-only size that the in-container `cargo +# build` needs. + +# Editor + tooling state +.idea/ +.vscode/ +*.swp + +# CI / eval artifacts that occasionally land in subdirs +**/eval-results/ diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index b6615a2a..1b3ee38b 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -117,8 +117,17 @@ async fn main() { .unwrap_or_else(|_| "0.0.0.0:8080".into()); let opamp_addr = std::env::var("CONTROLLER_OPAMP_ADDR") .unwrap_or_else(|_| "0.0.0.0:4320".into()); + // The default tracks the compose service name in + // `ASAPCollector/deploy/mvp-singlenode/docker-compose/base.yml`, + // which is still `controller:` post Phase-9 single-binary + // refactor (both controller and backend ship from + // `asap/query-backend:dev`, but they bind separate listeners + // under separate compose service names). Using the post-reorg + // crate name `control_plane` here doesn't resolve under the + // canonical compose stack and bakes a broken endpoint into every + // agent yaml the controller emits. let opamp_ep = std::env::var("CONTROLLER_OPAMP_ENDPOINT") - .unwrap_or_else(|_| "ws://control_plane:4320/v1/opamp".into()); + .unwrap_or_else(|_| "ws://controller:4320/v1/opamp".into()); let scrape_interval = Duration::from_secs( std::env::var("CONTROLLER_SCRAPE_INTERVAL_SECS") .ok()