From f4d4292a1a078e1a101d3a9adcce88c1798abb08 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 15 May 2026 15:12:47 -0600 Subject: [PATCH] =?UTF-8?q?test(e2e):=20controller=20=E2=86=92=20backend?= =?UTF-8?q?=20streaming-config=20round-trip=20+=20grouping=20plumb?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First end-to-end test of the gateway-less control-plane ↔ data-plane contract. Lives in `data_plane/tests/` since data_plane already owns the OTLP/precompute/HTTP wiring; control_plane is already a workspace dep so no Cargo.toml changes were needed. ## What it tests Two `#[tokio::test]`s: 1. `controller_plans_ddsketch_quantile_and_backend_parses_streaming_config` — workload `{ metric: http_latency_ms, agg: Quantile, sla: 0.01, window: 60s }` flows through `bind_workload_typed` → `split_typed_three_stage` → `emit_backend_streaming_config_json`, the JSON is POSTed to the in-process backend's `/api/v1/streaming-config`, and the GET endpoint reflects the parsed aggregation. Asserts the emitted JSON's shape (#244 content fields present, #244/#246 `aggregationId` absent). 2. `controller_plans_with_grouping_and_backend_parses_grouping_labels` — same shape but with `group_by_labels: ["zone"]`. Asserts the emitter surfaces `labels.grouping: ["zone"]` (#245), the POST succeeds, and the active-config snapshot contains `"zone"` in the parsed `AggregationConfig.grouping_labels`. ## Findings the test caught `bind_workload_typed`-produced PhysicalExpr doesn't surface `metric_name` or `window_secs` to `BackendAggregation` via `extract_edge_facts`. The underlying cause is likely that the allocator paints the `Logical(Scan{...})` chain at a stage other than Edge in some binder outputs, so the walk's `(Logical, Edge)` match arm doesn't fire — Step γ open-set label work (also blocking grouping fidelity) probably needs a coordinated fix. Pragmatic workaround mirroring the #245 grouping patch: control_plane/src/main.rs (handle_plan, Backend stage match arm) for agg in &mut be.aggregations { if agg.metric_name.is_empty() { agg.metric_name = workload.metric_name.clone(); } if agg.window_secs == 0 { agg.window_secs = workload.time_window.as_secs(); } agg.grouping = workload.group_by_labels.clone(); } This is belt-and-braces — if the L5 walk DID surface the field, the conditional `if … is_empty()` / `== 0` checks preserve it; otherwise the workload spec wins. Closes a real bug where fresh-deploy `handle_plan` POSTs would have failed the backend parser on `Missing metric` / `Missing windowSize` (PR #244 caught the JSON shape gap; this test caught the value-population gap). ## Other change `HttpServer::start_test_server` had `#[cfg(test)]` gating it to the lib's own unit tests — integration tests under `data_plane/tests/` are compiled separately and couldn't see it. Dropped the gate; the method's name + doc-comment make the test-only intent explicit, and production code uses `start()` regardless. ## Test plan - [x] `cargo test --test e2e_controller_plans_and_backend_serves`: **2 passed**. - [x] `cargo test --lib -p control_plane`: **687 passed**. - [x] `cargo test --tests --bins -p control_plane`: **27 passed**. Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/main.rs | 41 ++- data_plane/src/drivers/query/servers/http.rs | 11 +- ...e2e_controller_plans_and_backend_serves.rs | 303 ++++++++++++++++++ 3 files changed, 337 insertions(+), 18 deletions(-) create mode 100644 data_plane/tests/e2e_controller_plans_and_backend_serves.rs diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index b4a08fe5f..b6615a2a1 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -559,22 +559,33 @@ async fn handle_plan( } } crate::physical::colored_dag::StageConfig::Backend(mut be) => { - // Patch grouping label names from the - // workload spec. The typed L5 emitter - // produces `BackendAggregation` with - // `grouping: vec![]` because the canonical - // L3 `QueryExpr::Aggregate.by` is - // positional `ColumnId`s against a - // synthesized schema that has no label - // columns (open-set label naming is a - // Step γ TODO in - // `intent_algebra::column_resolution`). - // `QueryWorkload.group_by_labels` carries - // the names unambiguously, so we patch - // them in here — every aggregation under - // the same workload shares the same - // grouping today. + // Patch metric_name + grouping from the + // workload spec. The typed L5 emitter: + // * sets `metric_name` from + // `edge.source_metric`, which is + // populated by `extract_edge_facts` + // walking the `Logical(Scan{...})` + // chain. The path-recovery isn't + // guaranteed across every binder + // output shape, so we belt-and-brace + // it with `workload.metric_name`. + // * leaves `grouping` empty because the + // canonical L3 `QueryExpr::Aggregate.by` + // is positional `ColumnId`s against a + // synthesized schema with no label + // columns (open-set label naming is + // a Step γ TODO in + // `intent_algebra::column_resolution`). + // `QueryWorkload` carries both unambiguously, + // and every aggregation under one workload + // shares them — so the patch is uniform. for agg in &mut be.aggregations { + if agg.metric_name.is_empty() { + agg.metric_name = workload.metric_name.clone(); + } + if agg.window_secs == 0 { + agg.window_secs = workload.time_window.as_secs(); + } agg.grouping = workload.group_by_labels.clone(); } // Phase C: post the typed L5 streaming-config diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index aba1db530..3640572ca 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -479,9 +479,14 @@ impl HttpServer { Ok(()) } - /// Start server for testing on a random available port - /// Returns the actual port number used - #[cfg(test)] + /// Start server for testing on a random available port. Returns the + /// actual port number used. + /// + /// Intentionally not gated behind `#[cfg(test)]` — integration tests + /// under `data_plane/tests/` are compiled separately from the lib's + /// own unit tests and need this entry point. The name + doc-comment + /// make the testing intent explicit; production callers should use + /// the regular `start()` method. pub async fn start_test_server(&self) -> Result> { // Create adapter using factory let adapter = create_http_adapter(self.config.adapter_config.clone()); diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs new file mode 100644 index 000000000..bef4409d6 --- /dev/null +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -0,0 +1,303 @@ +//! End-to-end test of the gateway-less data path: +//! +//! fake-exporter (test harness) +//! ─OTLP sketches─► +//! asap-otel agent (test harness — sketches built locally) +//! ─OTLP/HTTP /v1/metrics─► +//! asapquery-backend (in-process: OtlpReceiver + PrecomputeEngine + HttpServer) +//! ─PromQL /api/v1/query─► +//! query answer +//! +//! The control plane drives the streaming-config: a `QueryWorkload` +//! goes through `bind_workload_typed` → `split_typed_three_stage` → +//! `emit_backend_streaming_config_json`, the resulting JSON is posted +//! to the backend's `/api/v1/streaming-config` endpoint, and the +//! backend's `AggregationConfig::from_yaml_data` parses it. +//! +//! Out of scope (per the task spec): Thanos / Gorilla / MinIO cold +//! path; the gateway tier (retired in #241/#243/#377); the real +//! `fake-exporter` and `asap-otel` binaries (the test harness plays +//! both roles in-process). +//! +//! Covered: +//! * Test 1 — controller emits a streaming-config JSON for a single +//! DDSketch-quantile workload; the backend's parser accepts it and +//! the GET endpoint reflects the registered aggregation. +//! * Test 2 — same shape with `group_by_labels: ["zone"]`; verifies +//! #245's grouping plumb survives the round-trip into the backend's +//! `AggregationConfig.grouping_labels`. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use control_plane::types::{AggType, QueryWorkload, WorkloadCharacteristics}; +use data_plane::storage_engines::types::HotReloadStreamingConfig; +use serde_json::Value as JsonValue; + +// ── Helpers ───────────────────────────────────────────────────────────────── + +/// Build a `QueryWorkload` with the given parameters. Mirrors the +/// `WorkloadAnalyzer` output shape but constructed directly for tests. +fn build_workload( + metric_name: &str, + aggregations: Vec, + accuracy_sla: f64, + time_window: Duration, + group_by_labels: Vec, + quantiles: Vec, +) -> QueryWorkload { + QueryWorkload { + metric_name: metric_name.to_string(), + label_filters: HashMap::new(), + group_by_labels, + aggregations, + time_window, + repeat_every: None, + accuracy_sla, + latency_sla: None, + sketch_type_override: None, + exact_required: false, + quantiles, + } +} + +/// 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. +/// +/// 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 { + 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) + .expect("split_typed_three_stage produced per-stage configs"); + let mut backend_cfg = configs + .into_iter() + .find_map(|(_stage, cfg)| match cfg { + control_plane::physical::colored_dag::StageConfig::Backend(be) => Some(be), + _ => None, + }) + .expect("typed-L5 emit must include a Backend stage for this workload"); + + // Mirror handle_plan: patch grouping (and metric_name when the L5 + // walk didn't surface it) from the workload spec before posting. + // The typed L5's `extract_edge_facts` populates `source_metric` + // when the `Logical(Scan{...})` chain is painted at Edge — but + // depending on the binder path the path-recovery isn't guaranteed, + // so the safe-belt-and-braces patch is to set both from the + // workload directly, the same way #245 patches grouping. + for agg in &mut backend_cfg.aggregations { + if agg.metric_name.is_empty() { + agg.metric_name = workload.metric_name.clone(); + } + if agg.window_secs == 0 { + agg.window_secs = workload.time_window.as_secs(); + } + agg.grouping = workload.group_by_labels.clone(); + } + + control_plane::emit::emit_backend_streaming_config_json(&backend_cfg) + .expect("emit_backend_streaming_config_json must succeed") +} + +/// Spin up an in-process backend HTTP server with `HotReloadStreamingConfig` +/// wired through both the query engine and the POST `/api/v1/streaming-config` +/// handler. Returns `(port, hot_reload_handle)` — the latter so tests can +/// also inspect the current config from the controller's side. +async fn start_backend_http_server() -> (u16, HotReloadStreamingConfig) { + use data_plane::drivers::query::adapters::config::AdapterConfig; + use data_plane::drivers::query::servers::{HttpServer, HttpServerConfig}; + use data_plane::query_engines::asap_query_engine::engine::ASAPQueryEngine; + use data_plane::storage_engines::sketch_db::index::SketchStore; + use data_plane::storage_engines::types::StreamingConfig; + + let hot_reload = HotReloadStreamingConfig::new(StreamingConfig::default()); + let query_engine = Arc::new(ASAPQueryEngine::new_with_hot_reload( + hot_reload.clone(), + 15_000, + )); + + let adapter_config = AdapterConfig::prometheus_promql( + "http://127.0.0.1:9999".to_string(), // unused — no forwarding in this test + false, + ); + let http_config = HttpServerConfig { + port: 0, + handle_http_requests: true, + adapter_config, + }; + + let sketch_index = Arc::new(SketchStore::new()); + let server = HttpServer::new(http_config, query_engine, sketch_index) + .with_hot_reload_config(hot_reload.clone()); + + let port = server + .start_test_server() + .await + .expect("start_test_server must succeed"); + + (port, hot_reload) +} + +/// POST a serde_json `Value` to `/api/v1/streaming-config` on the +/// in-process backend. Panics on non-2xx (the test wants to verify the +/// controller's emit is parseable). +async fn post_streaming_config(client: &reqwest::Client, port: u16, json: &JsonValue) { + let resp = client + .post(format!("http://127.0.0.1:{port}/api/v1/streaming-config")) + .header("content-type", "application/json") + .body(serde_json::to_vec(json).expect("serialize streaming-config JSON")) + .send() + .await + .expect("POST /api/v1/streaming-config"); + + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + assert!( + status.is_success(), + "POST /api/v1/streaming-config returned {status}: {body}\n\ + (controller-emitted JSON must round-trip through \ + AggregationConfig::from_yaml_data without errors)\n\ + body sent: {}", + serde_json::to_string_pretty(json).unwrap_or_default() + ); +} + +/// GET `/api/v1/streaming-config` and return the active-config snapshot +/// as a `serde_json::Value`. Verifies the active state after a POST. +async fn get_streaming_config(client: &reqwest::Client, port: u16) -> JsonValue { + let resp = client + .get(format!("http://127.0.0.1:{port}/api/v1/streaming-config")) + .send() + .await + .expect("GET /api/v1/streaming-config"); + 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)] +fn _wc_anchor() -> WorkloadCharacteristics { + WorkloadCharacteristics::default() +} + +// ── Test 1 — single DDSketch-quantile workload, no grouping ───────────────── +// +// Smoke test: the controller emits a streaming-config JSON for a +// workload that resolves to DDSketch. The backend's parser accepts it +// (POST returns 2xx) and the registered aggregation surfaces on the +// GET endpoint with the expected metric / sketch family. + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn controller_plans_ddsketch_quantile_and_backend_parses_streaming_config() { + let (port, _hot_reload) = start_backend_http_server().await; + let client = reqwest::Client::new(); + + let workload = build_workload( + "http_latency_ms", + vec![AggType::Quantile], + 0.01, + Duration::from_secs(60), + Vec::new(), + vec![0.99], + ); + let streaming_config_json = plan_streaming_config_json(&workload); + + // Sanity check on the emitted shape before we push it: the + // controller MUST emit content fields (#244) and MUST NOT emit a + // controller-allocated `aggregationId` (#244 / #246). + let aggs = streaming_config_json["aggregations"] + .as_array() + .expect("aggregations array"); + assert_eq!(aggs.len(), 1, "expected exactly one BackendAggregation\n{streaming_config_json}"); + let agg = &aggs[0]; + assert!( + agg.get("aggregationId").is_none(), + "controller must not emit aggregationId\n{agg}" + ); + assert_eq!(agg["metric"], "http_latency_ms"); + assert_eq!(agg["aggregationType"], "DDSketch"); + assert_eq!(agg["windowType"], "tumbling"); + assert!( + agg["windowSize"].as_u64().expect("windowSize u64") > 0, + "windowSize must be > 0\n{agg}" + ); + + post_streaming_config(&client, port, &streaming_config_json).await; + + // Verify the parsed config is visible via GET. + let active = get_streaming_config(&client, port).await; + assert_eq!( + active["aggregation_count"], 1, + "after POST, exactly one aggregation must be registered\n{active}" + ); +} + +// ── Test 2 — cross-host grouping (sum by zone) ────────────────────────────── +// +// Verifies #245's grouping plumb survives the controller → backend +// round-trip. The workload carries `group_by_labels: ["zone"]`; the +// emitted JSON must surface `["zone"]` in `labels.grouping`, the +// backend's parser must materialise it into `AggregationConfig. +// grouping_labels`, and the active-config snapshot must reflect that. + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn controller_plans_with_grouping_and_backend_parses_grouping_labels() { + let (port, _hot_reload) = start_backend_http_server().await; + let client = reqwest::Client::new(); + + let workload = build_workload( + "http_latency_ms", + vec![AggType::Quantile], + 0.01, + Duration::from_secs(60), + vec!["zone".to_string()], + vec![0.99], + ); + let streaming_config_json = plan_streaming_config_json(&workload); + + // Pre-check: the emitter must surface `labels.grouping = ["zone"]`. + let agg = &streaming_config_json["aggregations"][0]; + let grouping = agg["labels"]["grouping"] + .as_array() + .expect("labels.grouping array"); + let names: Vec<&str> = grouping.iter().filter_map(|s| s.as_str()).collect(); + assert_eq!( + names, + vec!["zone"], + "controller must thread workload.group_by_labels → labels.grouping (#245)\n\ + {streaming_config_json}" + ); + + post_streaming_config(&client, port, &streaming_config_json).await; + + // After POST, the active-config snapshot should reflect the + // grouping label was parsed into AggregationConfig. + let active = get_streaming_config(&client, port).await; + assert_eq!(active["aggregation_count"], 1); + + // Walk the streaming_config object to find the registered grouping + // labels. The snapshot path is `streaming_config.aggregation_configs. + // .grouping_labels.`. + let cfgs = active["streaming_config"]["aggregation_configs"] + .as_object() + .expect("aggregation_configs object in snapshot"); + assert_eq!( + cfgs.len(), + 1, + "expected exactly one parsed aggregation_config\n{active}" + ); + let (_fp, cfg) = cfgs.iter().next().expect("first cfg"); + // `grouping_labels` is `KeyByLabelNames` — its JSON shape is + // implementation-defined (likely `{labels: ["zone"]}` or just an + // array). Find "zone" anywhere inside. + let cfg_str = cfg.to_string(); + assert!( + cfg_str.contains("zone"), + "parsed AggregationConfig must contain `zone` in its grouping labels\n{cfg}" + ); +}