From 61f1c48ce346cc545d0dc3a7252160a9a4448bc2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 29 Jul 2026 23:13:31 -0600 Subject: [PATCH] test(e2e): prove control_plane really pushes BackendPlan and data_plane really serves from it New end-to-end test, real wire round trip (no mocking): control_plane's actual backend_plan::from_stage_config + BackendClient::post_backend_plan_typed build and push a real encoded BackendPlan; data_plane really decodes and installs it (verified via GET /api/v1/backend-plan reflecting the installed plan_id/materialization_count, and directly via the shared HotReloadBackendPlan handle); a live PromQL query against real ingested DDSketch data returns a correct, finite p99 value with the plan installed. Also extracts plan_backend_stage_config() out of plan_streaming_config_json() so both the legacy JSON emit and the new BackendPlan construction share the exact same BackendStageConfig -- no behavior change to existing tests. Co-Authored-By: Claude Sonnet 5 --- ...e2e_controller_plans_and_backend_serves.rs | 186 +++++++++++++++++- 1 file changed, 179 insertions(+), 7 deletions(-) diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index 2c3653bf..b2b8e03e 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -106,13 +106,17 @@ fn build_workload( } /// Run the controller's planning pipeline end-to-end on a `QueryWorkload` -/// and return the streaming-config JSON document the controller would -/// POST to the backend's `/api/v1/streaming-config` endpoint. +/// and return the `BackendStageConfig` the controller would emit from +/// for it — the same object both `emit_backend_streaming_config_json` +/// (legacy JSON) and `backend_plan::from_stage_config` (`BackendPlan`) +/// consume. /// /// Mirrors the `handle_plan` flow's `StageConfig::Backend(mut be)` -/// branch — including the post-emit grouping patch (#245) so the JSON -/// carries `labels.grouping` from `workload.group_by_labels`. -fn plan_streaming_config_json(workload: &QueryWorkload) -> JsonValue { +/// branch — including the post-emit grouping patch (#245) so the config +/// carries `grouping` from `workload.group_by_labels`. +fn plan_backend_stage_config( + workload: &QueryWorkload, +) -> control_plane::physical::colored_dag::BackendStageConfig { let physical_expr = control_plane::optimizer::rules::bind_workload_typed(workload) .expect("bind_workload_typed produced a PhysicalExpr"); let configs = control_plane::physical::stage_split::split_typed_three_stage(&physical_expr) @@ -141,7 +145,14 @@ fn plan_streaming_config_json(workload: &QueryWorkload) -> JsonValue { } agg.grouping = workload.group_by_labels.clone(); } + backend_cfg +} +/// Run the controller's planning pipeline end-to-end on a `QueryWorkload` +/// and return the streaming-config JSON document the controller would +/// POST to the backend's `/api/v1/streaming-config` endpoint. +fn plan_streaming_config_json(workload: &QueryWorkload) -> JsonValue { + let backend_cfg = plan_backend_stage_config(workload); // No continuous-monitoring (CDM) intents in these tests — pass an empty // slice (the `&[MonitorIntent]` arg added when CDM monitor specs landed). control_plane::emit::emit_backend_streaming_config_json(&backend_cfg, &[]) @@ -223,6 +234,31 @@ async fn get_streaming_config(client: &reqwest::Client, port: u16) -> JsonValue resp.json().await.expect("parse GET response as JSON") } +/// POST an encoded `BackendPlan` to `/api/v1/backend-plan` on the +/// in-process backend, using the control plane's real +/// `BackendClient::post_backend_plan_typed` — the exact code path a live +/// control plane process uses, not a hand-rolled request. +async fn post_backend_plan(port: u16, plan: &control_plane::backend_plan::BackendPlan) { + let endpoint = format!("http://127.0.0.1:{port}/api/v1/streaming-config"); + let client = control_plane::backend_client::BackendClient::new(endpoint); + client + .post_backend_plan_typed(plan.encode_to_vec()) + .await + .expect("POST /api/v1/backend-plan via BackendClient must succeed"); +} + +/// GET `/api/v1/backend-plan` and return the active-plan snapshot as a +/// `serde_json::Value`. Verifies the active state after a POST. +async fn get_backend_plan(client: &reqwest::Client, port: u16) -> JsonValue { + let resp = client + .get(format!("http://127.0.0.1:{port}/api/v1/backend-plan")) + .send() + .await + .expect("GET /api/v1/backend-plan"); + assert!(resp.status().is_success(), "GET returned {}", resp.status()); + resp.json().await.expect("parse GET response as JSON") +} + /// Suppress the `WorkloadCharacteristics` unused warning — kept around /// in case future tests need to pass per-workload resource caps. #[allow(dead_code)] @@ -241,6 +277,7 @@ fn _wc_anchor() -> WorkloadCharacteristics { struct FullStack { backend_port: u16, otlp_http_port: u16, + hot_reload_backend_plan: data_plane::storage_engines::types::HotReloadBackendPlan, } async fn start_full_stack(otlp_http_port: u16, otlp_grpc_port: u16) -> FullStack { @@ -304,6 +341,9 @@ async fn start_full_stack(otlp_http_port: u16, otlp_grpc_port: u16) -> FullStack }); // HTTP query server sharing the same SketchStore + hot-reload handle. + let hot_reload_backend_plan = data_plane::storage_engines::types::HotReloadBackendPlan::new( + control_plane::backend_plan::BackendPlan::default(), + ); let adapter_config = AdapterConfig::prometheus_promql("http://127.0.0.1:9999".to_string(), false); let http_config = HttpServerConfig { @@ -319,10 +359,12 @@ async fn start_full_stack(otlp_http_port: u16, otlp_grpc_port: u16) -> FullStack // `sketch_index` via OTLP ingest (the engine's // `precompute_engine` shares the Arc), but the query // path can't see them without this binding. - .with_sketch_index(sketch_index.clone()), + .with_sketch_index(sketch_index.clone()) + .with_hot_reload_backend_plan(hot_reload_backend_plan.clone()), ); let server = HttpServer::new(http_config, query_engine, sketch_index) - .with_hot_reload_config(hot_reload.clone()); + .with_hot_reload_config(hot_reload.clone()) + .with_hot_reload_backend_plan(hot_reload_backend_plan.clone()); let backend_port = server .start_test_server() .await @@ -334,6 +376,7 @@ async fn start_full_stack(otlp_http_port: u16, otlp_grpc_port: u16) -> FullStack FullStack { backend_port, otlp_http_port, + hot_reload_backend_plan, } } @@ -1023,6 +1066,135 @@ async fn controller_plan_to_query_full_roundtrip_ddsketch() { ); } +// ── Test — BackendPlan wire format: real push, real install, real serve ──── +// +// Same fixture as `controller_plan_to_query_full_roundtrip_ddsketch`, but +// this time the controller ALSO builds a real `BackendPlan` (via +// `control_plane::backend_plan::from_stage_config`, the exact function a +// live control plane process calls) and pushes it through +// `control_plane::backend_client::BackendClient::post_backend_plan_typed` +// — the exact HTTP client code a live control plane process uses, not a +// hand-rolled request. This proves the whole wire is real end to end: +// control-plane-side construction → protobuf encode → HTTP POST → +// backend-side decode → `ArcSwap` install → a live PromQL query answered +// correctly with the plan installed. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn controller_really_pushes_backend_plan_and_query_really_serves() { + let stack = start_full_stack(19_597, 19_598).await; + let client = reqwest::Client::new(); + + // ── 1. Controller plans the workload, POSTs the legacy streaming-config + // (still required: it's what drives sid registration on ingest) + // AND a real BackendPlan built from the SAME BackendStageConfig. ── + let workload = build_workload( + "http_latency_ms", + vec![AggType::Quantile], + 0.01, + Duration::from_secs(1), + vec!["service".to_string()], + vec![0.99], + ); + let backend_cfg = plan_backend_stage_config(&workload); + let streaming_config_json = + control_plane::emit::emit_backend_streaming_config_json(&backend_cfg, &[]) + .expect("emit_backend_streaming_config_json must succeed"); + post_streaming_config(&client, stack.backend_port, &streaming_config_json).await; + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system time before UNIX epoch") + .as_millis() as u64; + let plan = control_plane::backend_plan::from_stage_config(&backend_cfg, &[], 1, now_ms) + .expect("backend_plan::from_stage_config must succeed"); + assert_eq!( + plan.materializations.len(), + 1, + "the DDSketch quantile workload must produce exactly one materialization" + ); + post_backend_plan(stack.backend_port, &plan).await; + + // ── 2. Confirm the backend really installed it (not just accepted the + // POST) — read it back via GET, the same handle + // `l4_lowering.rs`'s serving-time lookup reads from. ── + let installed = get_backend_plan(&client, stack.backend_port).await; + assert_eq!(installed["status"], "success"); + assert_eq!(installed["plan_id"], 1); + assert_eq!( + installed["materialization_count"], 1, + "GET /api/v1/backend-plan must reflect the just-installed plan, not a stale/empty one:\n{}", + serde_json::to_string_pretty(&installed).unwrap_or_default() + ); + assert_eq!( + stack.hot_reload_backend_plan.snapshot().plan_id, + 1, + "the ASAPQueryEngine-side handle (shared with the HTTP server, per main.rs's wiring) \ + must observe the same installed plan the GET endpoint just reported" + ); + + // ── 3. Ingest a real DDSketch via OTLP (same fixture as Test 3). ── + let alpha = 0.01; + let store_counts = vec![5u64, 10, 15, 20]; + let dd_state = build_dd_sketch_state(alpha, store_counts, -1); + let sketch_bytes = dd_state.encode_to_vec(); + + let now_ns = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system time before UNIX epoch") + .as_nanos() as u64; + let sketch_t_ns = now_ns.saturating_sub(3_000_000_000); + let watermark_t_ns = now_ns.saturating_sub(1_000_000_000); + + let req = build_dd_sketch_export( + "http_latency_ms", + &[("service", "e2e-test")], + sketch_t_ns, + sketch_bytes, + alpha, + ); + post_otlp_http(&client, stack.otlp_http_port, req).await; + + let watermark_state = build_dd_sketch_state(alpha, Vec::new(), 0); + let watermark_req = build_dd_sketch_export( + "http_latency_ms", + &[("service", "e2e-test")], + watermark_t_ns, + watermark_state.encode_to_vec(), + alpha, + ); + post_otlp_http(&client, stack.otlp_http_port, watermark_req).await; + + tokio::time::sleep(Duration::from_millis(800)).await; + + // ── 4. Query via PromQL — this must actually succeed AND return a + // real, sane numeric quantile value, with the BackendPlan + // installed (not just "some code path returned success"). ── + let query_url = format!("http://127.0.0.1:{}/api/v1/query", stack.backend_port); + let response: JsonValue = client + .get(&query_url) + .query(&[("query", "quantile_over_time(0.99, http_latency_ms[10s])")]) + .send() + .await + .expect("PromQL query failed to send") + .json() + .await + .expect("PromQL response was not JSON"); + + assert_eq!( + response["status"].as_str().unwrap_or("(missing)"), + "success", + "query must succeed with a real BackendPlan installed. Response:\n{}", + serde_json::to_string_pretty(&response).unwrap_or_default() + ); + let value = response["data"]["result"] + .as_array() + .and_then(|r| extract_first_scalar(&JsonValue::Array(r.clone()))) + .expect("expected a scalar quantile result"); + assert!( + value.is_finite() && value > 0.0, + "expected a real, finite p99 value from the DDSketch fixture, got {value}" + ); +} + // ── Test 4 — full roundtrip with KLL ──────────────────────────────────────── // // Same shape as Test 3 but the workload pins KLL via