From ef373ced253d593ea283993f00a8ca2a21149d72 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 18 May 2026 05:41:19 -0600 Subject: [PATCH 1/2] fix(emit): cumulative streaming-config across (metric, role) pairs (B2 follow-up) PR #283 made `WorkloadStore` and `PlanStore` (metric, role)-keyed so a single metric can carry multiple aggregation roles (e.g. `http_requests_total` post-B2 has both DDSketch-Quantile from `quantile_over_time(...)` AND ExactAgg-Sum from `sum by (zone) (http_requests_total)`). But the streaming-config emit path in `control_plane/src/main.rs` stayed metric-keyed and emitted the current iteration's single-role `BackendStageConfig` alone. The data plane's `POST /api/v1/streaming-config` handler is an atomic full `handle.swap(new_config)`, so the second per-role POST destroyed the first role's aggregations on the backend and `sum by (zone) (http_requests_total)` failed with `ExactAgg(Sum) capability not satisfied`. This commit: * Rekeys `backend_routing_cache` from `HashMap` to `HashMap<(String, AggRole), _>`. * Inserts with the role derived in `handle_plan`'s existing `derive_agg_role` block (already in scope). * Builds a single cumulative `BackendStageConfig` by concatenating every cache entry's `aggregations` + `readouts` before emitting the streaming-config JSON, so the data plane's swap installs every role's aggregations atomically. * Merges per-(metric, role) cache entries by metric name before passing to `emit_backend_storage_routing` so a metric with both DDSketch + Sum correctly routes both shape families to the ASAP tier (the routing classifier reads `cfg.aggregations` to derive shape routing). * Updates the `backend_routing_cache` doc comment to reflect the (metric, role) keying and the cumulative-across-all-roles emit semantic. Constraints honored: no `data_plane/` changes (swap semantics are correct), no signature changes to `emit_backend_streaming_config_json` or `emit_backend_storage_routing` (merge happens at the call site), fire-and-forget POST contract preserved. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/main.rs | 242 +++++++++++++++++++++++++------------- 1 file changed, 163 insertions(+), 79 deletions(-) diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 09512bb7..804afea4 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -51,6 +51,7 @@ use replan::Replanner; use physical::colored_dag::emitter::BackendStageConfig; use store::{PlanStore, WorkloadStore}; use types::StageResourceBudgets; +use workload::AggRole; // ── Shared state ────────────────────────────────────────────────────────────── @@ -80,31 +81,48 @@ struct AppState { /// `CONTROLLER_BACKEND_ENDPOINT` is unset, matching the /// pre-existing fire-and-forget contract. backend_client: Option>, - /// Per-metric `BackendStageConfig` cache used to emit a - /// **cumulative** `BackendStorageRouting` JSON document on every - /// per-metric replan. + /// Per-`(metric, role)` `BackendStageConfig` cache used to emit + /// **cumulative** `StreamingConfig` AND `BackendStorageRouting` + /// JSON documents on every plan-emit cycle. /// - /// Why this exists: `POST /api/v1/storage_routing` on the backend - /// is an atomic per-tenant SWAP — every push replaces the whole - /// tenant's routing table. The control plane's pre-existing per-metric - /// post path emits a single-element `metrics:[…]` document per - /// `handle_plan` call, so when N metrics replan in sequence only - /// the last metric's entry survives in the backend's routing table. - /// That defaults the other N-1 metrics to `sketch_store`, which - /// has no ASAP-tier sketch state for archive-shape queries - /// (`count`, `topk`, `rate_post_hoc`, `histogram_quantile`, - /// `delta`, `deriv`, `absent`) → the backend returns empty / 404 → - /// the demo's accuracy reducer logs `archive_miss` for those metrics - /// even though gorillas3 wrote their TSDB blocks to MinIO and Thanos - /// has them indexed. + /// **Why this is (metric, role)-keyed** (B2 cumulative-emit follow-up + /// to PR #283): a single metric can carry MULTIPLE [`AggRole`] + /// entries (e.g. post-B2 the workload-registry pre-pop loop registers + /// `http_requests_total` against BOTH a DDSketch-Quantile role from + /// `quantile_over_time(...)` AND an ExactAgg-Sum role from + /// `sum by (zone) (http_requests_total)`). Pre-fix the cache was + /// keyed by metric name alone, so the second role's + /// `BackendStageConfig` overwrote the first. The data plane's + /// `POST /api/v1/streaming-config` handler is an atomic full + /// `handle.swap(new_config)` (see + /// `data_plane/src/drivers/query/servers/http.rs`), so the second + /// per-role POST destroys the first role's aggregations on the + /// backend → `sum by (zone) (http_requests_total)` returns + /// `ExactAgg(Sum) capability not satisfied`. /// - /// The cache is a `HashMap` keyed - /// by metric name. On every plan-emit cycle we update the entry for - /// the metric being planned and re-emit the storage-routing JSON - /// from the union of all currently-known plans, then push the - /// cumulative table. The next cycle's swap then preserves every - /// previously-seen metric's routing entry. - backend_routing_cache: Arc>>, + /// **Why this also matters for storage-routing**: `POST + /// /api/v1/storage_routing` is similarly an atomic per-tenant SWAP + /// — every push replaces the whole tenant's routing table. Pre-fix + /// the per-metric cache emitted a single-element `metrics:[…]` + /// document per `handle_plan` call, so when N metrics replanned in + /// sequence only the last metric's entry survived → archive-shape + /// queries fell to `default_engine: sketch_store` → `archive_miss` + /// for the other N-1 metrics. + /// + /// **Cumulative emit semantics** (post-fix): on every plan-emit + /// cycle the cache entry for the `(metric, role)` being planned is + /// updated, then: + /// * Concatenate `aggregations` + `readouts` across ALL cache + /// entries into a single cumulative `BackendStageConfig`, and + /// post that one config to `/api/v1/streaming-config` so the + /// data plane's swap installs every role's aggregations + /// simultaneously. + /// * Group cache entries by metric name and merge each metric's + /// `BackendStageConfig`s (concat aggregations + readouts) into + /// one entry per metric. Pass that per-metric list to + /// `emit_backend_storage_routing` so a metric carrying both + /// DDSketch + ExactAgg routes both shape families correctly. + backend_routing_cache: Arc>>, } // ── Entry point ─────────────────────────────────────────────────────────────── @@ -408,7 +426,7 @@ async fn main() { ); let runtime_samples_store = runtime_samples::RuntimeSamplesStore::new(1024); - let backend_routing_cache: Arc>> = + let backend_routing_cache: Arc>> = Arc::new(Mutex::new(HashMap::new())); let state = AppState { analyzer: Arc::new(Analyzer::new()), @@ -682,18 +700,92 @@ async fn handle_plan( } agg.grouping = workload.group_by_labels.clone(); } - // Phase C: post the typed L5 streaming-config - // JSON to ASAPQuery-backend via the shared - // BackendClient when configured. Without a - // configured endpoint this still no-ops - // silently — same fire-and-forget contract - // as the existing Replanner path. - match emit::emit_backend_streaming_config_json(&be) { + // B2 cumulative-emit follow-up: update the + // per-(metric, role) cache with THIS + // iteration's `be`, then BOTH the + // streaming-config emit and the + // storage-routing emit below derive their + // payload from the FULL cache. The + // single-iteration `be` is never sent on + // the wire on its own — every post is + // cumulative across all `(metric, role)` + // pairs the control plane has planned. + // + // Why: the data plane's + // `POST /api/v1/streaming-config` handler + // is `handle.swap(new_config)` (an atomic + // full replace) and the storage-routing + // handler is similarly an atomic per-tenant + // swap. Per-iteration posts overwrite + // siblings: + // * streaming-config — drops the prior + // role's `aggregations`, so a metric + // with both DDSketch (Quantile) and + // ExactAgg (Sum) loses one on the + // backend → `sum by (zone) + // (http_requests_total)` returns + // `ExactAgg(Sum) capability not satisfied` + // (the regression that motivates this + // PR — direct follow-up to #283 which + // made the workload/plan stores + // (metric, role)-keyed but left the + // emit path metric-only). + // * storage-routing — drops other + // metrics' entries → default + // `sketch_store` engine → `archive_miss`. + let cumulative_entries: Vec<((String, AggRole), BackendStageConfig)> = { + let mut cache = st.backend_routing_cache.lock().await; + cache.insert((workload.metric_name.clone(), role), be.clone()); + let mut v: Vec<((String, AggRole), BackendStageConfig)> = cache + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + // Deterministic ordering so the emitted + // JSON body is reproducible across runs + // and across test invocations. HashMap + // iteration order would otherwise make + // captured-body regression assertions + // flaky. + v.sort_by(|(a_k, _), (b_k, _)| { + a_k.0 + .cmp(&b_k.0) + .then_with(|| a_k.1.as_str().cmp(b_k.1.as_str())) + }); + v + }; + + // Cumulative streaming-config — one + // `BackendStageConfig` whose `aggregations` + // + `readouts` are the concatenation of + // every cache entry's. The data plane's + // swap installs this single + // multi-aggregation config atomically, so + // ALL roles for ALL metrics survive. + let cumulative_be = BackendStageConfig { + aggregations: cumulative_entries + .iter() + .flat_map(|(_, c)| c.aggregations.iter().cloned()) + .collect(), + readouts: cumulative_entries + .iter() + .flat_map(|(_, c)| c.readouts.iter().cloned()) + .collect(), + }; + + // Phase C: post the cumulative typed L5 + // streaming-config JSON to ASAPQuery-backend + // via the shared BackendClient when + // configured. Without a configured endpoint + // this still no-ops silently — same + // fire-and-forget contract as the existing + // Replanner path. + match emit::emit_backend_streaming_config_json(&cumulative_be) { Ok(json_doc) => { info!( stage = "backend", - aggregations = be.aggregations.len(), - readouts = be.readouts.len(), + aggregations = cumulative_be.aggregations.len(), + readouts = cumulative_be.readouts.len(), + cumulative_pairs = cumulative_entries.len(), "[USE_TYPED_STAGE_SPLIT] posting typed backend JSON" ); if let Some(client) = st.backend_client.as_ref() { @@ -723,58 +815,50 @@ async fn handle_plan( Err(e) => warn!(error = %e, "emit_backend_streaming_config_json failed"), } - // Phase α (MVP): emit per-metric storage - // routing table from the same typed L5 - // BackendStageConfig, and POST it to the - // backend's `/api/v1/storage_routing` - // endpoint via the BackendClient sibling - // method. The classification rules live in - // `config::stage_emit::emit_backend_storage_routing` - // — see that function's doc-comment for the - // sketch-family → query-shape mapping. - // - // CRITICAL: the backend's - // `POST /api/v1/storage_routing` handler is - // an atomic per-tenant SWAP — every push - // replaces the whole tenant's routing - // table. We MUST emit the cumulative - // routing JSON across every metric the - // control plane has planned to date, otherwise - // each per-metric replan erases the routing - // entries for every other metric and the - // backend defaults them to - // `sketch_store` (which has nothing for - // archive-shape queries). That's the - // `archive_miss` failure mode for - // HLL/CountSketch/CountMin/KLL metrics in - // the post-#345 demo runs even though - // gorillas3 writes their TSDB blocks to - // MinIO and Thanos has them indexed. + // Phase α (MVP) cumulative storage-routing. + // The routing classifier + // (`build_routing_entry` in + // `emit/stage_config.rs`) reads + // `cfg.aggregations` to derive shape + // routing, so we MUST merge every role's + // aggregations for one metric into a single + // `BackendStageConfig` before passing it + // through — otherwise a metric with both + // DDSketch (Quantile) and ExactAgg (Sum) + // would emit only the last-cached role's + // shape classifications and route the + // siblings to archive. // - // We thread the per-metric `BackendStageConfig` - // through `state.backend_routing_cache` so - // a metric replan picks up an updated entry - // for itself but preserves every previously - // planned metric's entry. - let cumulative_plans = { - let mut cache = st.backend_routing_cache.lock().await; - cache.insert(workload.metric_name.clone(), be.clone()); - cache - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect::>() - }; - let routing_input: Vec<(String, &BackendStageConfig)> = - cumulative_plans - .iter() - .map(|(k, v)| (k.clone(), v)) - .collect(); + // `emit_backend_storage_routing`'s signature + // is `&[(String, &BackendStageConfig)]` — + // per-metric, NOT per-(metric, role) — so + // the merge happens at the call site (per + // the PR's no-signature-change constraint). + let mut by_metric: std::collections::BTreeMap = + std::collections::BTreeMap::new(); + for ((m, _r), cfg) in &cumulative_entries { + let entry = by_metric.entry(m.clone()).or_insert_with(|| { + BackendStageConfig { + aggregations: Vec::new(), + readouts: Vec::new(), + } + }); + entry.aggregations.extend(cfg.aggregations.iter().cloned()); + entry.readouts.extend(cfg.readouts.iter().cloned()); + } + let routing_owned: Vec<(String, BackendStageConfig)> = + by_metric.into_iter().collect(); + let routing_input: Vec<(String, &BackendStageConfig)> = routing_owned + .iter() + .map(|(k, v)| (k.clone(), v)) + .collect(); match emit::emit_backend_storage_routing(&routing_input) { Ok(routing_doc) => { info!( stage = "backend", metric = %workload.metric_name, - cumulative_metrics = cumulative_plans.len(), + cumulative_metrics = routing_owned.len(), + cumulative_pairs = cumulative_entries.len(), "[USE_TYPED_STAGE_SPLIT] posting cumulative storage-routing JSON" ); if let Some(client) = st.backend_client.as_ref() { From 5e0a9d8f8e097f87b56fdb78316be0d1a4d42395 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 18 May 2026 05:49:09 -0600 Subject: [PATCH 2/2] test(emit): regression test for cumulative streaming-config push MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `streaming_config_cumulative_push_covers_all_planned_metrics` — the streaming-config sibling of the existing `storage_routing_cumulative_push_covers_all_5_sketched_metrics`. Replays the demo's per-metric plan-POST sequence (5 sketched contract metrics) against a mock backend, captures every streaming-config body, and asserts the LAST body (the one the data plane's atomic swap installs) carries aggregations for ALL 5 metrics — proving the per-(metric, role) cache + cumulative concatenation in main.rs's typed Backend-stage block preserves every prior plan-POST's row across subsequent swaps. Pre-fix this would fail with "missing aggregations for metric X" — only the last POST's metric survived the swap. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/main.rs | 156 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 156 insertions(+) diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 804afea4..08aad275 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -2978,4 +2978,160 @@ mod api_tests { ); } } + + // ── Regression: cumulative streaming-config across (metric, role) ───────── + // + // PR #283 made `WorkloadStore` and `PlanStore` (metric, role)-keyed, + // so a single metric can carry MULTIPLE aggregation roles (e.g. + // post-B2 `http_requests_total` has both a DDSketch-Quantile entry + // from `quantile_over_time(...)` AND an ExactAgg-Sum entry from + // `sum by (zone) (...)` in the workload store). + // + // Pre-this-fix the streaming-config emit path in `handle_plan` was + // still metric-keyed and posted the CURRENT iteration's + // `BackendStageConfig` alone. The data plane's + // `POST /api/v1/streaming-config` handler is an atomic full + // `handle.swap(new_config)`, so the second per-(metric, role) plan + // POST destroyed the first one's aggregations on the backend and + // `sum by (zone) (http_requests_total)` lands with + // `ExactAgg(Sum) capability not satisfied`. + // + // This test replays the demo's per-metric plan-POST sequence + // against a mock backend and captures every streaming-config body. + // The LAST body (the one the data plane's swap installs) MUST + // carry aggregations from EVERY prior plan POST, otherwise the + // swap erases the earlier metrics' rows and the data plane can't + // answer queries against them. + // + // The (metric, role) cache key is exercised in tandem by the live + // mvp-workload.yaml pre-pop loop (the workload registry lists 3 + // entries for `http_requests_total`) → see the MVP smoke-test + // pipeline. This in-process test exercises the cumulative-merge + // plumbing in isolation against the same emit path used by both + // the pre-pop loop and per-request replans. + #[tokio::test] + async fn streaming_config_cumulative_push_covers_all_planned_metrics() { + // Activate the typed-stage-split path (the only path that emits + // the typed streaming-config JSON; the legacy emit path no-ops). + let _env = EnvVarGuard::set(physical::stage_split::ENV_USE_TYPED_STAGE_SPLIT, "1"); + + // Mock backend that captures every streaming-config body. Same + // pattern as the sibling `storage_routing_cumulative_push_...` + // test — an axum router that drains the request body into a + // shared sink. Mounted at the canonical + // `/api/v1/streaming-config` path so `BackendClient`'s URL + // forwarding hits it directly. + type SinkInner = std::sync::Mutex>; + let sink: Arc = Arc::new(std::sync::Mutex::new(Vec::new())); + let sink_capture = Arc::clone(&sink); + let mock_app = axum::Router::new() + .route( + "/api/v1/streaming-config", + axum::routing::post(move |body: axum::body::Bytes| { + let sink = Arc::clone(&sink_capture); + async move { + let s = String::from_utf8_lossy(&body).to_string(); + sink.lock().unwrap().push(s); + axum::http::StatusCode::OK + } + }), + ) + // Sibling storage-routing endpoint stubbed so the + // `handle_plan` cycle's second POST doesn't 404 and + // pollute the test log (the assertion only inspects the + // streaming-config sink). + .route( + "/api/v1/storage_routing", + axum::routing::post(|| async { axum::http::StatusCode::OK }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, mock_app).await.unwrap(); + }); + tokio::time::sleep(Duration::from_millis(50)).await; + + // Build an AppState with the backend pointed at the mock URL. + let backend_url = format!("http://{addr}/api/v1/streaming-config"); + let (state, _) = test_app_with_backend(Some(backend_url)); + let app = axum::Router::new() + .route("/api/v1/plan", axum::routing::post(handle_plan)) + .with_state(state.clone()); + + // The 5 sketched contract metrics from MVP §46 — the SAME + // set the sibling `storage_routing_cumulative_push_...` test + // exercises. Each gets a separate `POST /api/v1/plan` with + // the metric-name → classified sketch family from + // `classify_demo_metric`. Pre-fix the metric-only cache + // would have collapsed sequential same-metric POSTs onto one + // slot; this test uses 5 distinct metrics so the assertion + // surfaces the cumulative-merge gap (every metric's row must + // survive every other metric's swap). + let sketched = [ + "http_requests_total_latency_ms", + "request_size_bytes", + "unique_users_per_min", + "top_endpoint_qps", + "endpoint_request_freq", + ]; + + for m in &sketched { + let app = app.clone(); + let req = Request::builder() + .method("POST") + .uri("/api/v1/plan") + .header("content-type", "application/json") + .body(Body::from(plan_spec(m).to_string())) + .unwrap(); + let resp = app.oneshot(req).await.unwrap(); + assert_eq!( + resp.status(), + StatusCode::OK, + "POST /api/v1/plan for `{m}` must return 200", + ); + } + + // Drain the mock sink: every plan-emit must have produced + // exactly one streaming-config body (5 plans → 5 bodies). + let bodies = sink.lock().unwrap().clone(); + assert_eq!( + bodies.len(), + sketched.len(), + "expected one streaming-config POST per plan; got {} bodies", + bodies.len(), + ); + + // The LAST body is the one the data plane's swap installs + // (the swap is destructive — last write wins). It MUST list + // aggregations for ALL 5 sketched metrics, otherwise the + // swap erases the earlier metrics' rows and queries against + // them fail with `…capability not satisfied` — the + // streaming-config analogue of the storage-routing + // `archive_miss` failure documented on the sibling test. + let last: serde_json::Value = + serde_json::from_str(bodies.last().unwrap()).expect("last body is valid JSON"); + let aggs = last["aggregations"] + .as_array() + .expect("aggregations array on cumulative streaming-config body"); + // Wire-format note: `build_backend_aggregation_json` writes + // the field under key `metric` (NOT `metric_name`) — see + // `emit/stage_config.rs::build_backend_aggregation_json`. + let metric_names: std::collections::BTreeSet = aggs + .iter() + .filter_map(|a| { + a.get("metric") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + }) + .collect(); + for m in &sketched { + assert!( + metric_names.contains(*m), + "final cumulative streaming-config missing aggregations \ + for metric `{m}`; contains only {metric_names:?}\n\ + full body: {}", + bodies.last().unwrap(), + ); + } + } }