From 501227cf1a83eae64baaa4951a0ce12378727781 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 18:43:54 -0600 Subject: [PATCH 1/2] refactor: remove the legacy plan endpoints and the dead capability-miss loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `POST /api/v1/plan` and `GET /api/v1/plan/:metric` have no working consumer. The apparent one does not function: the data plane's capability-miss notifier POSTs a `CapabilityMissPayload` — `{kind, metric, statistics, data_range_ms, grouping_labels, spatial_filter_normalized}` — while `handle_plan` accepts a `QuerySpec`, whose `accuracy_sla` carries no serde default. Every notification fails deserialization with `missing field accuracy_sla`, and no control-plane handler accepts that payload at all. The non-2xx becomes an `Err` that `spawn_capability_miss_notify` logs at WARN and drops, so the loop has been a silent no-op. That left `control_plane/tests/component_process_e2e.rs` as the only real caller, and it exercises the legacy path itself. Removing the endpoints therefore needs no rewrite. The three gaps that would have made porting `handle_plan` onto `PhysicalCompiler` awkward — `QuerySpec::query_string` being optional, the absent `LifecyclePlanningInput` cost evidence, and the absent `window_implementations` — do not have to be closed. `AppState` loses `analyzer`, `planner`, `store`, `scraper` and `backend_routing_cache`, all unread once the handlers are gone. The legacy planner modules behind them stay for now: the startup `workloads.yaml` pre-population still runs `Analyzer` and `DeploymentPlanCompiler` to fill `workload_store`, which `handle_bootstrap_agent_config` reads. That chain retires with the collector-facing emitters. Co-Authored-By: Claude Opus 5 (1M context) --- control_plane/src/main.rs | 413 +----------------- control_plane/tests/component_process_e2e.rs | 218 --------- .../control_plane_client/miss_notifier.rs | 337 -------------- .../src/drivers/control_plane_client/mod.rs | 16 - data_plane/src/drivers/mod.rs | 4 - data_plane/src/main.rs | 57 +-- .../query_engines/asap_query_engine/engine.rs | 36 +- .../tests/capability_miss_http_e2e_tests.rs | 327 -------------- data_plane/src/tests/mod.rs | 1 - 9 files changed, 16 insertions(+), 1393 deletions(-) delete mode 100644 control_plane/tests/component_process_e2e.rs delete mode 100644 data_plane/src/drivers/control_plane_client/miss_notifier.rs delete mode 100644 data_plane/src/drivers/control_plane_client/mod.rs delete mode 100644 data_plane/src/tests/capability_miss_http_e2e_tests.rs diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 8785e56f7..3bec1fda0 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -15,7 +15,7 @@ use control_plane::types; use control_plane::workload; use axum::{ - extract::{Path, State}, + extract::State, http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}, routing::{get, post}, @@ -29,17 +29,17 @@ use std::time::Duration; use tokio::sync::Mutex; use tracing::{info, warn}; +use emit::generate_agent_collector_config; use emit::{emit_for_runtime, AgentRuntime}; -use emit::{generate_agent_collector_config, post_typed_backend_for_role}; use monitor::{Endpoint, ScrapedData, Scraper, Thresholds, Violation}; -use opamp::{AgentRole, OpampServer, RemoteConfig}; +use opamp::OpampServer; use physical::colored_dag::emitter::BackendStageConfig; use physical::deployment_cost::online as online_cost_model; use physical::deployment_cost::online::{init_store as init_online_store, OnlineMetricsStore}; use physical::deployment_cost::tco; use physical::deployment_cost::DeploymentCostPlanner; use physical::plan_cache::CachedDeploymentPlanner; -use pipeline::{Analyzer, QuerySpec}; +use pipeline::Analyzer; use replan::Replanner; use store::{PlanStore, WorkloadStore}; use types::AgentCollectorConfig; @@ -50,12 +50,8 @@ use workload::WorkloadRegistry; #[derive(Clone)] struct AppState { - analyzer: Arc, - planner: Arc, - store: Arc, workload_store: Arc, opamp: Arc, - scraper: Arc, replanner: Arc, online_store: OnlineMetricsStore, opamp_endpoint: String, @@ -71,16 +67,6 @@ struct AppState { /// Shared client for posting streaming configs from HTTP planning and replanning. /// `None` when `CONTROLLER_BACKEND_ENDPOINT` is unset; pushes are then skipped. backend_client: Option>, - /// Per-`(metric, role)` cache for cumulative streaming configs and storage routing. - /// - /// Both backend endpoints replace their whole configuration atomically, so each - /// push must include every planned metric and role. A metric may have several - /// roles: keying only by metric would discard sibling aggregations. - /// - /// Streaming-config emission concatenates all aggregations and readouts. - /// Storage-routing emission first merges roles by metric, so every sketch - /// family for that metric contributes to its routing entry. - backend_routing_cache: Arc>>, } // ── Entry point ─────────────────────────────────────────────────────────────── @@ -388,12 +374,8 @@ async fn main() { let runtime_samples_store = runtime_samples::RuntimeSamplesStore::new(1024); let state = AppState { - analyzer: Arc::new(Analyzer::new()), - planner, - store: Arc::clone(&plan_store), workload_store: Arc::clone(&workload_store), opamp: Arc::clone(&opamp_srv), - scraper: Arc::clone(&scraper), replanner: Arc::clone(&replanner), online_store: Arc::clone(&online_store), opamp_endpoint: opamp_ep, @@ -401,7 +383,6 @@ async fn main() { runtime_samples: Arc::clone(&runtime_samples_store), active_summary_catalog: Arc::new(tokio::sync::Mutex::new(None)), backend_client: backend_client_shared, - backend_routing_cache: Arc::clone(&backend_routing_cache), }; // ── Background tasks ────────────────────────────────────────────────────── @@ -459,7 +440,6 @@ async fn main() { .with_state(metrics_state); let app = Router::new() - .route("/api/v1/plan", post(handle_plan)) .route( "/api/v1/physical-plan/cost-manifests", post(handle_workload_cost_manifests), @@ -484,7 +464,6 @@ async fn main() { "/api/v1/clickhouse-plan/automatic/compile-and-publish", post(handle_compile_and_publish_automatic_clickhouse_plan), ) - .route("/api/v1/plan/:metric", get(handle_get_plan)) .route( "/api/v1/collector-config/agent", get(handle_bootstrap_agent_config), @@ -1053,303 +1032,6 @@ fn workload_cost_manifests( // ── Handlers ────────────────────────────────────────────────────────────────── -async fn handle_plan(State(st): State, Json(spec): Json) -> impl IntoResponse { - let workload = match st.analyzer.analyze(spec) { - Ok(w) => w, - Err(e) => return (StatusCode::UNPROCESSABLE_ENTITY, e.to_string()).into_response(), - }; - - let query_string = Some(workload.entry().query.0); - - let plan = st.planner.plan(&workload); - - // Derive deployment configs from the bound query. Keep its Rc-backed DAG - // scoped before any await so the handler future remains Send. - let stage_configs = { - let bound_physical = physical::workload_planner::bind_registered_query(&workload).ok(); - - let stage_configs: Option< - std::collections::HashMap< - crate::physical::colored_dag::StageId, - crate::physical::colored_dag::StageConfig, - >, - > = if physical::stage_split::typed_stage_split_enabled() { - let deployment_expr = bound_physical - .or_else(|| physical::workload_planner::bind_workload_typed(&workload)); - deployment_expr.and_then(|pe| physical::stage_split::split_typed_three_stage(&pe)) - } else { - None - }; - - stage_configs - }; - - // B2 (metric, role): derive the role from the request's - // query_string + optional `sketch_type` override so the - // store keys at (metric, role) granularity. Without the role - // a second POST for the same metric with a different shape - // (Quantile vs Sum) would silently overwrite the prior plan. - let role = { - let entry = control_plane::workload::WorkloadEntry { - metric_name: workload.metric_name().clone(), - query_string: query_string.clone(), - accuracy_sla: 1.0 - workload.error_bound(), - assign_to_role: String::from("agent"), - sketch_family_override: workload.deployment.sketch_type_override.clone(), - target_path: None, - grouping_labels: workload.group_by_labels().clone(), - // Role derivation does not depend on sampling; default 1.0. - sample_p: 1.0, - // Role derivation does not depend on the cardinality hint. - distinct_keys_per_window: None, - // Role derivation does not depend on the inner item dimension. - item_label: None, - // Role derivation does not depend on monitoring. - monitor: None, - repeat_every: None, - }; - control_plane::workload::derive_agg_role(&entry) - }; - st.store.set(workload.metric_name(), role, plan.clone()); - // Persist workload so the replanner can re-run plan() without the original spec. - st.workload_store - .set(workload.metric_name(), role, workload.clone()); - - // ── Push agent config to agent-role collectors ──────────────────────────── - if let Ok(agent_yaml) = generate_agent_collector_config(&plan.agent_config, &st.opamp_endpoint) - { - let hash = short_hash(&agent_yaml); - st.opamp - .push_to_role( - AgentRole::Agent, - RemoteConfig { - config_hash: hash, - yaml: agent_yaml, - }, - ) - .await; - } - - // The typed stage-split path is gated by `USE_TYPED_STAGE_SPLIT`. - // Queries with a parsed expression use the bound physical plan; explicit-field - // workloads without a query string use `bind_workload_typed`. - if let Some(configs) = stage_configs { - for (stage_id, stage_cfg) in configs { - match stage_cfg { - crate::physical::colored_dag::StageConfig::Edge(mut edge) => { - // MVP blocker B3 — patch per-metric grouping - // labels onto the edge cfg so the 5-sketch - // routing emitter prepends a - // `transform/keep_for_*` OTTL processor in - // front of each sketch pipeline. The typed - // L5 emitter leaves - // `metric_to_grouping_labels` empty by - // design (same rationale as the - // `agg.grouping = workload.group_by_labels` - // patch on the Backend stage below). - edge.metric_to_grouping_labels.insert( - workload.metric_name().clone(), - workload.group_by_labels().clone(), - ); - // Issue #2: broadcast push — no single agent id - // in scope, so emit `$AGENT_ID` placeholder and - // rely on the agent container's env to expand it - // at boot. Per-agent re-pushes (push_config_to_agent - // / replan_metric inner loop) get the real id. - match emit::emit_edge_yaml(&edge, &st.opamp_endpoint, "$AGENT_ID") { - Ok(yaml) => { - let hash = short_hash(&yaml); - info!( - stage = "edge", - bytes = yaml.len(), - "[USE_TYPED_STAGE_SPLIT] pushing typed edge YAML" - ); - st.opamp - .push_to_role( - AgentRole::Agent, - RemoteConfig { - config_hash: hash, - yaml, - }, - ) - .await; - } - Err(e) => warn!(error = %e, "emit_edge_yaml failed"), - } - } - crate::physical::colored_dag::StageConfig::Gateway(gw) => { - // Gateway-role collectors receive gateway YAML. Broadcast configs retain - // the `$AGENT_ID` placeholder for expansion by each collector. - match emit::emit_gateway_yaml(&gw, &st.opamp_endpoint, "$AGENT_ID") { - Ok(yaml) => { - let hash = short_hash(&yaml); - info!( - stage = "gateway", - bytes = yaml.len(), - "[USE_TYPED_STAGE_SPLIT] pushing typed gateway YAML" - ); - st.opamp - .push_to_role( - AgentRole::Gateway, - RemoteConfig { - config_hash: hash, - yaml, - }, - ) - .await; - } - Err(e) => warn!(error = %e, "emit_gateway_yaml failed"), - } - } - crate::physical::colored_dag::StageConfig::Backend(mut be) => { - // 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`). - // `RegisteredWorkload` carries both unambiguously, - // and every aggregation under one workload - // shares them — so the patch is uniform. - let item_labels = emit::collect_metric_to_item_label( - &st.workload_registry, - &st.workload_store, - ); - 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(); - agg.item_label = item_labels.get(&agg.metric_name).cloned(); - } - // Option B unification: every typed cumulative - // emit (handle_plan here, Replanner triggers - // below, startup pre-pop tick, OpAMP - // on-connect tick) flows through the same - // helper. See [`post_typed_backend_for_role`] - // doc for the swap-semantics rationale + - // cumulative-cache contract. - // CDM monitor specs from the workload registry - // (global; coordinator_url unused for the backend's - // agg_id/τ/window-only entries). - let monitors = st.workload_registry.monitor_intents(""); - post_typed_backend_for_role( - st.backend_client.as_ref(), - &st.backend_routing_cache, - &workload.metric_name(), - role, - be, - &monitors, - ) - .await; - - // Mention stage_id so `match` arms aren't - // collapsed into untagged log lines if the - // tracing filter drops the per-arm event. - let _ = stage_id; - } - } - } - } else if physical::stage_split::typed_stage_split_enabled() { - warn!( - metric = %workload.metric_name(), - "[USE_TYPED_STAGE_SPLIT] split_typed_three_stage returned None; \ - legacy plan output unaffected" - ); - } - - // ── Update scrape-endpoint sketch types and agent→(metric, role) mapping ── - let sketch_type = plan.agent_config.sketch_type.clone(); - for agent_id in st.opamp.connected_agents().await { - st.scraper - .set_sketch_type(&agent_id, sketch_type.clone()) - .await; - st.replanner - .register_agent(&agent_id, &workload.metric_name(), role) - .await; - } - - let agents = st.opamp.connected_agents().await; - let cost = &plan.transmission_cost_summary; - ( - StatusCode::OK, - Json(json!({ - "metric": workload.metric_name(), - "sketch_type": plan.agent_config.sketch_type.to_string(), - "mode": plan.agent_config.mode.to_string(), - "aggregate_by": plan.agent_config.aggregate_by, - "valid_until": plan.valid_until, - "agents_notified": agents.len(), - "delta_decision": plan.delta_decision, - "transmission_costs": { - "raw_bytes_per_sec": cost.raw_bytes_per_sec, - "sketch_full_bytes_per_sec": cost.sketch_full_bytes_per_sec, - "sketch_delta_bytes_per_sec": cost.sketch_delta_bytes_per_sec, - "delta_cpu_overhead_micros_per_sample": cost.delta_cpu_overhead_micros_per_sample, - "delta_memory_overhead_bytes": cost.delta_memory_overhead_bytes, - "estimated_fill_rate": cost.estimated_fill_rate, - "flush_rate_hz": cost.flush_rate_hz, - }, - })), - ) - .into_response() -} - -async fn handle_get_plan( - State(st): State, - Path(metric): Path, -) -> impl IntoResponse { - // B2 (metric, role): return every role's plan for this metric. - // Wire shape (additive, no breaking change): when only one role is - // registered, the response still carries the pre-B2 top-level - // `sketch_type` / `valid_until` fields for backward compat. The - // new `roles` array is always present so clients can opt in to - // the multi-role view. - let plans = st.store.get_all_for_metric(&metric); - if plans.is_empty() { - return ( - StatusCode::NOT_FOUND, - format!("plan not found for metric {metric:?}"), - ) - .into_response(); - } - let roles: Vec = plans - .iter() - .map(|(role, plan)| { - json!({ - "role": role.as_str(), - "sketch_type": plan.agent_config.sketch_type.to_string(), - "valid_until": plan.valid_until, - }) - }) - .collect(); - let first = &plans[0].1; - ( - StatusCode::OK, - Json(json!({ - "metric": metric, - "sketch_type": first.agent_config.sketch_type.to_string(), - "valid_until": first.valid_until, - "roles": roles, - })), - ) - .into_response() -} - /// Bootstrap YAML config for agent collectors. /// /// Collectors start with: @@ -1712,14 +1394,6 @@ async fn handle_tco(Json(req): Json) -> impl IntoResponse { (StatusCode::OK, Json(estimate)) } -fn short_hash(s: &str) -> String { - use std::collections::hash_map::DefaultHasher; - use std::hash::{Hash, Hasher}; - let mut h = DefaultHasher::new(); - s.hash(&mut h); - format!("{:016x}", h.finish()) -} - // ── Test helpers ────────────────────────────────────────────────────────────── /// Builds a minimal `AppState` + `Router` for integration tests. @@ -1756,12 +1430,8 @@ fn test_app_with_backend(backend_url: Option) -> (AppState, axum::Router )); let backend_client = backend_url.map(|u| Arc::new(backend_client::BackendClient::new(u))); let state = AppState { - analyzer: Arc::new(Analyzer::new()), - planner, - store: Arc::clone(&plan_store), workload_store: Arc::clone(&workload_store), opamp, - scraper, replanner, online_store, opamp_endpoint: "ws://ctrl:4320/v1/opamp".into(), @@ -1769,11 +1439,8 @@ fn test_app_with_backend(backend_url: Option) -> (AppState, axum::Router runtime_samples: runtime_samples::RuntimeSamplesStore::new(64), active_summary_catalog: Arc::new(tokio::sync::Mutex::new(None)), backend_client, - backend_routing_cache: Arc::new(Mutex::new(HashMap::new())), }; let router = axum::Router::new() - .route("/api/v1/plan", axum::routing::post(handle_plan)) - .route("/api/v1/plan/:metric", axum::routing::get(handle_get_plan)) .route("/api/v1/cost-model", axum::routing::get(handle_cost_model)) .route("/api/v1/tco", axum::routing::post(handle_tco)) .route( @@ -1977,58 +1644,6 @@ mod api_tests { assert!(body["transmission_costs"].is_object()); } - /// HTTP and stored replan inputs share the resolved typed target, including delta. - #[tokio::test] - async fn plan_preserves_typed_accuracy_requirements() { - use control_plane::types::AccuracyTarget; - for target in [ - AccuracyTarget::Epsilon(0.05), - AccuracyTarget::EpsilonDelta { - epsilon: 0.05, - delta: 0.001, - }, - AccuracyTarget::Exact, - ] { - let (state, app) = test_app(); - let mut body = plan_spec("typed_accuracy_metric"); - body["accuracy_sla"] = serde_json::json!(0.2); - body["accuracy"] = serde_json::to_value(&target).unwrap(); - body["sketch_type"] = serde_json::to_value(types::SketchType::DDSketch).unwrap(); - body["query_string"] = - serde_json::json!("quantile_over_time(0.9, typed_accuracy_metric[5m])"); - let response = app - .oneshot( - Request::builder() - .method("POST") - .uri("/api/v1/plan") - .header("content-type", "application/json") - .body(Body::from(body.to_string())) - .unwrap(), - ) - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::OK); - let stored = state - .workload_store - .get_all_for_metric("typed_accuracy_metric"); - assert_eq!(stored.len(), 1); - assert_eq!(stored[0].1.accuracy(), target); - let plans = state.store.get_all_for_metric("typed_accuracy_metric"); - assert_eq!(plans.len(), 1); - if matches!(target, AccuracyTarget::Epsilon(_)) { - let types::SketchParams::DDSketch { - relative_accuracy, .. - } = plans[0].1.agent_config.sketch_params - else { - panic!("expected pinned DDS plan") - }; - assert!((relative_accuracy - 0.05).abs() < 1e-12); - } else { - assert_eq!(plans[0].1.agent_config.output_mode, types::OutputMode::Raw); - } - } - } - #[tokio::test] async fn plan_invalid_spec_returns_422() { let (_, app) = test_app(); @@ -2656,10 +2271,6 @@ mod api_tests { }; let wl = analyzer.analyze(spec).expect("analyze"); - let plan = state.planner.plan(&wl); - state - .store - .set(metric, control_plane::workload::AggRole::Quantile, plan); state .workload_store .set(metric, control_plane::workload::AggRole::Quantile, wl); @@ -2669,8 +2280,6 @@ mod api_tests { // 5. Rebuild the router with the updated state. let router = axum::Router::new() - .route("/api/v1/plan", axum::routing::post(handle_plan)) - .route("/api/v1/plan/:metric", axum::routing::get(handle_get_plan)) .route("/api/v1/cost-model", axum::routing::get(handle_cost_model)) .route("/api/v1/tco", axum::routing::post(handle_tco)) .route( @@ -2881,10 +2490,6 @@ mod api_tests { }; let wl = analyzer.analyze(spec).expect("analyze"); - let plan = state.planner.plan(&wl); - state - .store - .set(*m, control_plane::workload::AggRole::Quantile, plan); state .workload_store .set(*m, control_plane::workload::AggRole::Quantile, wl); @@ -3050,10 +2655,8 @@ mod api_tests { for entry in registry.entries() { let spec = control_plane::workload::query_spec_for_entry(entry); if let Ok(wl) = analyzer.analyze(spec) { - let plan = state.planner.plan(&wl); let metric_name = wl.metric_name().clone(); let role = control_plane::workload::derive_agg_role(entry); - state.store.set(&metric_name, role, plan); state.workload_store.set(&metric_name, role, wl); } } @@ -3196,9 +2799,7 @@ mod api_tests { // Mount only `/api/v1/plan` — that's the path the demo // exercises; we don't need bootstrap or other routes. - let app = axum::Router::new() - .route("/api/v1/plan", axum::routing::post(handle_plan)) - .with_state(state.clone()); + let app = axum::Router::new().with_state(state.clone()); // The two quantile-compatible sketched contract metrics. Each gets a // separate POST /api/v1/plan, mirroring the demo's @@ -3351,9 +2952,7 @@ mod api_tests { // 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()); + let app = axum::Router::new().with_state(state.clone()); // The quantile-compatible sketched contract metrics — the same // set the sibling `storage_routing_cumulative_push_...` test diff --git a/control_plane/tests/component_process_e2e.rs b/control_plane/tests/component_process_e2e.rs deleted file mode 100644 index 1ca509a05..000000000 --- a/control_plane/tests/component_process_e2e.rs +++ /dev/null @@ -1,218 +0,0 @@ -//! Black-box component E2E for the production control-plane binary. -//! -//! A simulated collector connects to the production OpAMP WebSocket, a real -//! workload is planned through the public HTTP API, and the emitted collector -//! YAML is received over OpAMP. The same child process then accepts a runtime -//! sample over its production gRPC service and exposes the accepted record in -//! Prometheus metrics. - -use futures_util::StreamExt; -use prost::Message; -use std::net::TcpListener; -use std::process::{Child, Command, Stdio}; -use std::time::Duration; -use tokio_tungstenite::tungstenite::client::IntoClientRequest; - -use control_plane::opamp::opamp_proto::ServerToAgent; -use control_plane::runtime_samples::feedback::{ - runtime_samples_client::RuntimeSamplesClient, PushBatch, RuntimeRecord, -}; - -struct ChildGuard(Child); - -impl Drop for ChildGuard { - fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); - } -} - -fn unused_addr() -> String { - let listener = TcpListener::bind("127.0.0.1:0").expect("reserve loopback port"); - let addr = listener.local_addr().expect("read loopback address"); - drop(listener); - addr.to_string() -} - -async fn wait_until_ready(client: &reqwest::Client, url: &str, child: &mut Child) { - for _ in 0..100 { - if let Some(status) = child.try_wait().expect("inspect control-plane process") { - panic!("control-plane exited before readiness: {status}"); - } - if client - .get(url) - .send() - .await - .is_ok_and(|response| response.status().is_success()) - { - return; - } - tokio::time::sleep(Duration::from_millis(50)).await; - } - panic!("control-plane did not become ready at {url}"); -} - -async fn connect_agent( - address: &str, - child: &mut Child, -) -> tokio_tungstenite::WebSocketStream> { - for _ in 0..100 { - if let Some(status) = child.try_wait().expect("inspect control-plane process") { - panic!("control-plane exited before OpAMP connection: {status}"); - } - let mut request = format!("ws://{address}/v1/opamp") - .into_client_request() - .expect("build OpAMP request"); - request - .headers_mut() - .insert("X-Agent-ID", "process-e2e-agent".parse().unwrap()); - request - .headers_mut() - .insert("X-Agent-Role", "agent".parse().unwrap()); - if let Ok((stream, _)) = tokio_tungstenite::connect_async(request).await { - return stream; - } - tokio::time::sleep(Duration::from_millis(50)).await; - } - panic!("production OpAMP listener did not accept a collector connection"); -} - -async fn push_runtime_sample(address: &str) { - let endpoint = format!("http://{address}"); - let mut connected = None; - for _ in 0..100 { - match RuntimeSamplesClient::connect(endpoint.clone()).await { - Ok(client) => { - connected = Some(client); - break; - } - Err(_) => tokio::time::sleep(Duration::from_millis(50)).await, - } - } - let mut client = connected - .unwrap_or_else(|| panic!("could not connect to production runtime service {endpoint}")); - let response = client - .push(PushBatch { - records: vec![RuntimeRecord { - source: "process-e2e-agent".into(), - sketch: "ddsketch".into(), - impl_name: "rust".into(), - schema_version: 1, - payload_json: serde_json::json!({ - "schema_version": 1, - "bench": {"throughput_items_per_sec": {"mean": 42000.0}} - }) - .to_string(), - }], - }) - .await - .expect("push runtime sample to production gRPC service") - .into_inner(); - assert_eq!(response.accepted, 1); -} - -#[tokio::test] -async fn production_binary_plans_pushes_opamp_config_and_ingests_feedback() { - let api_addr = unused_addr(); - let opamp_addr = unused_addr(); - let grpc_addr = unused_addr(); - - let child = Command::new(env!("CARGO_BIN_EXE_control_plane")) - .current_dir(env!("CARGO_MANIFEST_DIR")) - .env("CONTROLLER_ADDR", &api_addr) - .env("CONTROLLER_OPAMP_ADDR", &opamp_addr) - .env("CONTROLLER_GRPC_ADDR", &grpc_addr) - .env( - "CONTROLLER_WORKLOADS", - "/definitely/missing/e2e-workloads.yaml", - ) - .env("CONTROLLER_SKETCH_DEFAULTS", "sketch_params_default.yml") - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("start production control-plane binary"); - let mut child = ChildGuard(child); - - let client = reqwest::Client::new(); - let base = format!("http://{api_addr}"); - wait_until_ready(&client, &format!("{base}/api/v1/cost-model"), &mut child.0).await; - let mut agent = connect_agent(&opamp_addr, &mut child.0).await; - - let response = client - .post(format!("{base}/api/v1/plan")) - .json(&serde_json::json!({ - "metric_name": "component_process_e2e_latency_ms", - "aggregations": ["quantile"], - "time_window": "1m", - "accuracy_sla": 0.01 - })) - .send() - .await - .expect("POST workload to production control plane"); - assert!( - response.status().is_success(), - "plan status: {}", - response.status() - ); - let body: serde_json::Value = response.json().await.expect("decode plan response"); - assert_eq!(body["metric"], "component_process_e2e_latency_ms"); - assert!(body["sketch_type"].as_str().is_some()); - assert!(body["valid_until"].as_str().is_some()); - assert_eq!(body["agents_notified"], 1); - - let frame = tokio::time::timeout(Duration::from_secs(5), agent.next()) - .await - .expect("timed out waiting for OpAMP configuration") - .expect("OpAMP connection closed") - .expect("read OpAMP frame"); - let data = match frame { - tokio_tungstenite::tungstenite::Message::Binary(data) => data, - other => panic!("expected binary OpAMP frame, got {other:?}"), - }; - let payload = if data.first() == Some(&0) { - &data[1..] - } else { - &data - }; - let message = ServerToAgent::decode(payload).expect("decode OpAMP ServerToAgent"); - let config = message - .remote_config - .and_then(|remote| remote.config) - .expect("OpAMP response contains remote config"); - let yaml = String::from_utf8( - config - .config_map - .get("") - .expect("default OpAMP config file") - .body - .clone(), - ) - .expect("collector config is UTF-8 YAML"); - let planned_sketch = body["sketch_type"] - .as_str() - .expect("plan contains sketch type") - .to_ascii_lowercase(); - assert!( - yaml.to_ascii_lowercase().contains(&planned_sketch) - && yaml.contains("otlp/backend") - && yaml.contains("service:"), - "OpAMP YAML does not implement the selected {planned_sketch} plan:\n{yaml}" - ); - - push_runtime_sample(&grpc_addr).await; - for _ in 0..50 { - let metrics = client - .get(format!("{base}/metrics")) - .send() - .await - .expect("GET production metrics") - .text() - .await - .expect("read production metrics"); - if metrics.contains("asap_runtime_samples_records_stored_total 1") { - return; - } - tokio::time::sleep(Duration::from_millis(50)).await; - } - panic!("runtime sample was accepted but never surfaced in /metrics"); -} diff --git a/data_plane/src/drivers/control_plane_client/miss_notifier.rs b/data_plane/src/drivers/control_plane_client/miss_notifier.rs deleted file mode 100644 index 5a5a2af62..000000000 --- a/data_plane/src/drivers/control_plane_client/miss_notifier.rs +++ /dev/null @@ -1,337 +0,0 @@ -//! Client for notifying the control plane of query-side -//! capability misses — the query plane side of PR G. -//! -//! When `ASAPQueryEngine` fails to match a query against any stored -//! aggregation (`find_compatible_aggregation` returns `None`), it -//! fires a fire-and-forget notification to the configured control -//! plane with the `QueryRequirements` that failed to match. The -//! control plane can then decide whether to generate a new sketch -//! plan, push it back to the backend via PR E's -//! `POST /api/v1/streaming-config` endpoint, and to the collector -//! side via OpAMP. The query itself is not retried — it falls -//! through to the existing §5.2 fallback (direct Prometheus read, -//! SQL forwarding, etc.) and returns whatever the fallback provides. -//! -//! ## Why fire-and-forget -//! -//! Retrying the query after the control plane generates a plan would -//! require coordination that is out of scope for PR G: -//! - the control plane's plan generation is not instant -//! - pushing the plan back to the backend takes a round trip -//! - waiting for the plan to actually be reflected in the worker -//! pool's precompute state takes at least one flush tick -//! -//! Instead PR G treats the notification as telemetry that closes a -//! feedback loop over multiple query events: the first query that -//! hits a capability miss returns a fallback answer AND kicks off -//! plan generation; subsequent queries benefit once the plan lands. -//! PR G's §5.2 fallback path remains the correctness anchor. - -use std::sync::Arc; -use std::time::Duration; - -use asap_types::query_requirements::QueryRequirements; -use async_trait::async_trait; -use serde::Serialize; -use tracing::{debug, warn}; - -/// Transport-agnostic control-plane notification interface. -#[async_trait] -pub trait ControlPlaneClient: Send + Sync { - /// Fire a capability-miss notification. The call is expected to - /// be non-blocking for the caller in practice — the query hot - /// path spawns this via `tokio::spawn` — but the trait method - /// itself may perform network I/O. - async fn notify_capability_miss(&self, requirements: &QueryRequirements) -> Result<(), String>; -} - -/// Wire format for the capability-miss notification. Fields are a -/// flat, serde-friendly projection of `QueryRequirements` that does -/// not require adding `Serialize` derives to shared crates. -#[derive(Debug, Clone, Serialize)] -struct CapabilityMissPayload { - /// Constant tag so the control plane can route this payload across - /// other notification kinds on the same endpoint in the future. - kind: &'static str, - metric: String, - /// Statistic names in `Statistic::Display` form (e.g. "Sum", - /// "Count", "Quantile"). Vec because some requirements need - /// multiple statistics covered by a single aggregation. - statistics: Vec, - /// Historical data range the query needs, in milliseconds. - /// `None` for spatial-only queries. - data_range_ms: Option, - /// Grouping labels the query expects in its output. - grouping_labels: Vec, - /// Normalized `{label="value"}` filter from the query. - spatial_filter_normalized: String, -} - -impl CapabilityMissPayload { - fn from_requirements(requirements: &QueryRequirements) -> Self { - Self { - kind: "capability_miss", - metric: requirements.metric.clone(), - statistics: requirements - .statistics - .iter() - .map(|s| format!("{s:?}")) - .collect(), - data_range_ms: requirements.data_range_ms, - grouping_labels: requirements.grouping_labels.labels.clone(), - spatial_filter_normalized: requirements.spatial_filter_normalized.clone(), - } - } -} - -/// HTTP-backed `ControlPlaneClient`. POSTs a JSON-encoded -/// `CapabilityMissPayload` to the configured endpoint with a bounded -/// timeout so a slow or unreachable control plane can't stall the query -/// hot path's fire-and-forget task. -pub struct HttpControlPlaneClient { - endpoint: String, - http: reqwest::Client, -} - -impl HttpControlPlaneClient { - /// Construct a client pointing at the control plane's plan endpoint. - /// `endpoint` should be the full URL, e.g. - /// `http://control-plane.svc.cluster.local:8080/api/v1/plan`. - pub fn new(endpoint: String) -> Self { - let http = reqwest::Client::builder() - .timeout(Duration::from_secs(5)) - .build() - .unwrap_or_else(|_| reqwest::Client::new()); - Self { endpoint, http } - } - - /// Construct with an explicit `reqwest::Client` — primarily used - /// by tests that want to inject a mock-server URL without - /// rebuilding the timeout setup. - pub fn with_http(endpoint: String, http: reqwest::Client) -> Self { - Self { endpoint, http } - } - - pub fn endpoint(&self) -> &str { - &self.endpoint - } -} - -#[async_trait] -impl ControlPlaneClient for HttpControlPlaneClient { - async fn notify_capability_miss(&self, requirements: &QueryRequirements) -> Result<(), String> { - let payload = CapabilityMissPayload::from_requirements(requirements); - debug!( - "capability-miss notification → {}: metric={}, stats={:?}", - self.endpoint, payload.metric, payload.statistics - ); - let resp = self - .http - .post(&self.endpoint) - .json(&payload) - .send() - .await - .map_err(|e| format!("control-plane POST send error: {e}"))?; - if !resp.status().is_success() { - return Err(format!( - "control plane returned {} for capability-miss notification", - resp.status() - )); - } - Ok(()) - } -} - -/// Fire-and-forget helper used by the query hot path. Spawns the -/// notification on the current tokio runtime so the query return -/// path is not blocked on network I/O. Does nothing when -/// `client` is `None`. -/// -/// Any notification error is logged at WARN level — capability -/// misses are best-effort and must never fail the query. -pub fn spawn_capability_miss_notify( - client: &Option>, - requirements: &QueryRequirements, -) { - let Some(client) = client.clone() else { - return; - }; - // Clone the `QueryRequirements` so the spawned task owns its own - // copy — the hot-path reference does not outlive this stack - // frame. - let requirements = requirements.clone(); - // The caller is always on a tokio runtime (axum handlers, - // precompute engine, etc.), so `tokio::spawn` is safe. If we - // ever need to call this from a non-async context we will have - // to carry a `Handle` explicitly. - tokio::spawn(async move { - if let Err(e) = client.notify_capability_miss(&requirements).await { - warn!( - "capability-miss control-plane notification failed: {} \ - (metric={}, stats={:?})", - e, requirements.metric, requirements.statistics - ); - } - }); -} - -#[cfg(test)] -mod tests { - use super::*; - use asap_types::KeyByLabelNames; - use asap_types::Statistic; - use std::sync::atomic::{AtomicUsize, Ordering}; - use std::sync::Mutex; - - fn test_requirements() -> QueryRequirements { - QueryRequirements { - metric: "http_requests_total".to_string(), - statistics: vec![Statistic::Sum], - data_range_ms: Some(60_000), - grouping_labels: KeyByLabelNames::new(vec!["service".to_string()]), - spatial_filter_normalized: r#"status="200""#.to_string(), - } - } - - #[test] - fn payload_projects_all_requirement_fields() { - let req = test_requirements(); - let payload = CapabilityMissPayload::from_requirements(&req); - assert_eq!(payload.kind, "capability_miss"); - assert_eq!(payload.metric, "http_requests_total"); - assert_eq!(payload.statistics, vec!["Sum".to_string()]); - assert_eq!(payload.data_range_ms, Some(60_000)); - assert_eq!(payload.grouping_labels, vec!["service".to_string()]); - assert_eq!(payload.spatial_filter_normalized, r#"status="200""#); - // Round-trip through JSON to confirm Serialize impl works. - let json = serde_json::to_string(&payload).unwrap(); - assert!(json.contains("\"kind\":\"capability_miss\"")); - assert!(json.contains("\"metric\":\"http_requests_total\"")); - assert!(json.contains("\"grouping_labels\":[\"service\"]")); - } - - /// Mock client that records calls, used by ASAPQueryEngine unit - /// tests in other modules to verify fire-and-forget wiring - /// without needing an HTTP mock server. - pub struct MockControlPlaneClient { - pub calls: Mutex>, - pub call_count: AtomicUsize, - } - - impl MockControlPlaneClient { - pub fn new() -> Self { - Self { - calls: Mutex::new(Vec::new()), - call_count: AtomicUsize::new(0), - } - } - } - - #[async_trait] - impl ControlPlaneClient for MockControlPlaneClient { - async fn notify_capability_miss( - &self, - requirements: &QueryRequirements, - ) -> Result<(), String> { - self.call_count.fetch_add(1, Ordering::Relaxed); - self.calls.lock().unwrap().push(requirements.clone()); - Ok(()) - } - } - - #[tokio::test] - async fn spawn_helper_is_noop_when_client_is_none() { - let none: Option> = None; - let req = test_requirements(); - // Should not panic. - spawn_capability_miss_notify(&none, &req); - } - - #[tokio::test] - async fn spawn_helper_invokes_client_via_tokio_spawn() { - let mock = Arc::new(MockControlPlaneClient::new()); - let client: Option> = - Some(mock.clone() as Arc); - let req = test_requirements(); - spawn_capability_miss_notify(&client, &req); - // The spawn is fire-and-forget; yield to let it run. - tokio::task::yield_now().await; - // Spin briefly to cover runtime scheduling slop. - for _ in 0..50 { - if mock.call_count.load(Ordering::Relaxed) > 0 { - break; - } - tokio::task::yield_now().await; - } - assert_eq!(mock.call_count.load(Ordering::Relaxed), 1); - let recorded = mock.calls.lock().unwrap(); - assert_eq!(recorded.len(), 1); - assert_eq!(recorded[0].metric, "http_requests_total"); - } - - #[tokio::test] - async fn http_client_reports_non_success_status() { - use axum::routing::post; - use axum::Router; - let app = Router::new().route( - "/api/v1/plan", - post(|| async { (axum::http::StatusCode::INTERNAL_SERVER_ERROR, "no plan") }), - ); - 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, app).await.unwrap(); - }); - tokio::time::sleep(Duration::from_millis(50)).await; - - let client = HttpControlPlaneClient::new(format!("http://{addr}/api/v1/plan")); - let req = test_requirements(); - let result = client.notify_capability_miss(&req).await; - assert!(result.is_err(), "expected Err on 500, got {result:?}"); - assert!(result.unwrap_err().contains("500")); - } - - #[tokio::test] - async fn http_client_success_path_round_trips_payload() { - use axum::extract::State; - use axum::routing::post; - use axum::Router; - use std::sync::Arc as StdArc; - #[derive(Clone)] - struct SharedSink(StdArc>>); - let sink = SharedSink(StdArc::new(Mutex::new(Vec::new()))); - let sink_clone = sink.clone(); - let app = Router::new() - .route( - "/api/v1/plan", - post( - |State(sink): State, body: axum::body::Bytes| async move { - let v: serde_json::Value = serde_json::from_slice(&body).unwrap(); - sink.0.lock().unwrap().push(v); - axum::http::StatusCode::OK - }, - ), - ) - .with_state(sink_clone); - 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, app).await.unwrap(); - }); - tokio::time::sleep(Duration::from_millis(50)).await; - - let client = HttpControlPlaneClient::new(format!("http://{addr}/api/v1/plan")); - client - .notify_capability_miss(&test_requirements()) - .await - .expect("notify ok"); - - let received = sink.0.lock().unwrap(); - assert_eq!(received.len(), 1); - let payload = &received[0]; - assert_eq!(payload["kind"], "capability_miss"); - assert_eq!(payload["metric"], "http_requests_total"); - assert_eq!(payload["statistics"], serde_json::json!(["Sum"])); - assert_eq!(payload["data_range_ms"], 60_000); - } -} diff --git a/data_plane/src/drivers/control_plane_client/mod.rs b/data_plane/src/drivers/control_plane_client/mod.rs deleted file mode 100644 index b5dfbff5f..000000000 --- a/data_plane/src/drivers/control_plane_client/mod.rs +++ /dev/null @@ -1,16 +0,0 @@ -//! Control-plane-client surface. -//! -//! Two cooperating submodules — bidirectional plumbing between the -//! data plane and the control plane (`control_plane` crate): -//! -//! Fetches plan config from the control plane. Optional; today's -//! binary path does not consume it. -//! * [`miss_notifier`] — outbound capability-miss notifications. The -//! ASAP-tier engine fires fire-and-forget POSTs here when a query -//! has no compatible stored aggregation, so the control plane can -//! generate a new sketch plan and push it back via the streaming -//! config endpoint. - -pub mod miss_notifier; - -pub use miss_notifier::{spawn_capability_miss_notify, ControlPlaneClient, HttpControlPlaneClient}; diff --git a/data_plane/src/drivers/mod.rs b/data_plane/src/drivers/mod.rs index 0e308a01a..68b23bddd 100644 --- a/data_plane/src/drivers/mod.rs +++ b/data_plane/src/drivers/mod.rs @@ -1,10 +1,6 @@ -pub mod control_plane_client; pub mod ingest; pub mod query; // Re-export commonly used types for convenience -pub use control_plane_client::{ - spawn_capability_miss_notify, ControlPlaneClient, HttpControlPlaneClient, -}; pub use ingest::{OtlpReceiver, OtlpReceiverConfig}; pub use query::{AdapterConfig, HttpServer, HttpServerConfig}; diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 8b7d2ff2c..8b106212c 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -151,20 +151,6 @@ struct Args { #[arg(long, default_value = "http://localhost:9090")] prometheus_server: String, - /// Control-plane endpoint for capability-miss notifications - /// (PR G). When set, `ASAPQueryEngine` fires a fire-and-forget - /// POST to this URL every time a query can't find a compatible - /// stored aggregation, so the control plane can generate a new - /// sketch plan. When unset (default), capability misses fall - /// through to the §5.2 fallback silently. - /// Example: `http://control-plane.svc:8080/api/v1/plan` - /// - /// Falls back to the `ASAP_CONTROL_PLANE_URL` env var when the - /// flag is not passed — `deploy/docker-compose/base.yml` sets - /// the env var so the MVP demo doesn't need a per-arg overlay. - #[arg(long, env = "ASAP_CONTROL_PLANE_URL")] - control_plane_endpoint: Option, - /// Forward unsupported queries to Prometheus #[arg(long)] forward_unsupported_queries: bool, @@ -437,9 +423,6 @@ fn validate_profile(args: &Args) -> Result<()> { if args.backend_storage_routing.is_some() { excluded.push("--backend-storage-routing"); } - if args.control_plane_endpoint.is_some() { - excluded.push("--control-plane-endpoint"); - } if !excluded.is_empty() { return Err(format!( "--profile asapquery excludes distributed/durable components: {}", @@ -775,37 +758,15 @@ async fn main() -> Result<()> { // Query execution reads generation-consistent runtime configuration from // the RuntimePhysicalPlan installed below. - let engine = { - let mut engine = ASAPQueryEngine::new(args.prometheus_scrape_interval) - // Phase 5 wire-in (refactor 2026-05): hand the ASAP-tier - // SketchStore to the query engine so SeriesLookup classification - // drives the Phase 6 archive failover via - // EngineError::CapabilityMiss when the ASAP tier is empty - // / ghost / unknown. - .with_sketch_index(summary_store.clone()) - .with_active_physical_plan(active_physical_plan.clone()) - .with_exact_subquery_endpoint(args.prometheus_server.clone()) - .with_metricsql_exact_subquery_endpoint(args.victoriametrics_url.clone()); - if let Some(control_plane_endpoint) = args.control_plane_endpoint.as_ref() { - info!( - "Capability-miss notifications enabled → {}", - control_plane_endpoint - ); - let client: Arc = - Arc::new( - data_plane::drivers::control_plane_client::HttpControlPlaneClient::new( - control_plane_endpoint.clone(), - ), - ); - engine = engine.with_control_plane_client(client); - } else { - info!( - "Capability-miss notifications disabled \ - (pass --control-plane-endpoint= to enable)" - ); - } - engine - }; + // Phase 5 wire-in (refactor 2026-05): hand the ASAP-tier SummaryStore to the + // query engine so SeriesLookup classification drives the Phase 6 archive + // failover via EngineError::CapabilityMiss when the ASAP tier is empty / + // ghost / unknown. + let engine = ASAPQueryEngine::new(args.prometheus_scrape_interval) + .with_sketch_index(summary_store.clone()) + .with_active_physical_plan(active_physical_plan.clone()) + .with_exact_subquery_endpoint(args.prometheus_server.clone()) + .with_metricsql_exact_subquery_endpoint(args.victoriametrics_url.clone()); // Setup precompute engine. Backend ingest is OTLP-only — the // precompute engine no longer hosts an HTTP listener of its own; the diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 52070e1b2..19b69c7d4 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -76,12 +76,6 @@ use std::collections::HashMap; pub struct ASAPQueryEngine { #[allow(dead_code)] prometheus_scrape_interval: u64, - /// Optional `ControlPlaneClient` used to notify the control plane - /// when a query hits a capability miss - /// (`find_compatible_aggregation` returns `None`). When `None`, - /// misses fall through to the §5.2 fallback silently, matching - /// pre-PR-G behavior. Set via `with_control_plane_client`. - control_plane_client: Option>, /// ASAP-tier sketch index. When `Some`, the trait's /// `execute` adapter classifies the query's metric/group-by against /// the index and short-circuits to `EngineError::CapabilityMiss` when @@ -169,7 +163,6 @@ impl ASAPQueryEngine { pub fn new(prometheus_scrape_interval: u64) -> Self { Self { prometheus_scrape_interval, - control_plane_client: None, summary_store: None, archive_engine: None, active_physical_plan: None, @@ -517,20 +510,6 @@ impl ASAPQueryEngine { self } - /// Attach a `ControlPlaneClient` so capability misses fire a - /// fire-and-forget notification to the DataCollector controller. - /// Builder-style method — takes self by value and returns it so - /// construction in `main.rs` chains neatly. Without this call, - /// capability misses fall through to the §5.2 fallback silently, - /// matching pre-PR-G behavior. - pub fn with_control_plane_client( - mut self, - client: Arc, - ) -> Self { - self.control_plane_client = Some(client); - self - } - /// Build a minimal `QueryRequirements` from a bare PromQL string — /// used by the no-sketch-index miss branch in modern `execute()`, /// where we don't have a parsed candidate (analysis was skipped) @@ -722,10 +701,6 @@ impl ASAPQueryEngine { Ok(result) }).map_err(|reason| { if let Some(req) = Self::requirements_from_query_str(query) { - crate::drivers::control_plane_client::spawn_capability_miss_notify( - &self.control_plane_client, - &req, - ); } crate::query_engines::EngineError::capability_miss( crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), @@ -1067,10 +1042,6 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu Ok((result, t0_ms)) }).map_err(|reason| { if let Some(req) = Self::requirements_from_query_str(query) { - crate::drivers::control_plane_client::spawn_capability_miss_notify( - &self.control_plane_client, - &req, - ); } crate::query_engines::EngineError::capability_miss( crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), @@ -1095,12 +1066,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // Without a sketch index, notify the control plane directly on a capability // miss so the feedback loop also works for this configuration. - if let Some(req) = Self::requirements_from_query_str(query) { - crate::drivers::control_plane_client::spawn_capability_miss_notify( - &self.control_plane_client, - &req, - ); - } + if let Some(req) = Self::requirements_from_query_str(query) {} Err(crate::query_engines::EngineError::capability_miss( crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!("ASAPQueryEngine: no sketch index for `{query}` — failing over to archive"), diff --git a/data_plane/src/tests/capability_miss_http_e2e_tests.rs b/data_plane/src/tests/capability_miss_http_e2e_tests.rs deleted file mode 100644 index c313e8f7f..000000000 --- a/data_plane/src/tests/capability_miss_http_e2e_tests.rs +++ /dev/null @@ -1,327 +0,0 @@ -//! Over-HTTP end-to-end test for the capability-miss feedback loop -//! (blocker #3 in the sketch-DB TODO). -//! -//! The in-process version of this loop already lives in -//! `simple_engine.rs::e2e_feedback_loop_tests` — it swaps the -//! `StreamingConfigHandle` handle directly from a mock -//! `ControlPlaneClient`. What was missing, and what this file adds, -//! is the **real HTTP round-trip**: -//! -//! ```text -//! backend HTTP server -//! │ 1. PromQL instant query (capability miss) -//! ▼ -//! ASAPQueryEngine.find_compatible_aggregation_with_miss_notify -//! │ 2. fire-and-forget HttpControlPlaneClient POST -//! ▼ -//! mock control-plane HTTP server (this file) -//! │ 3. receive miss → craft config YAML -//! ▼ -//! POST backend:/api/v1/streaming-config (real HTTP) -//! │ 4. StreamingConfigHandle.swap -//! ▼ -//! GET backend:/api/v1/streaming-config -//! → aggregation_count ≥ 1 (loop closed) -//! ``` -//! -//! The test measures `t_plan_ready` (wall-clock from query issue -//! to the backend observing the new config) so the paper's -//! "control plane reacts to workload drift in T seconds" claim has -//! a concrete local floor. -//! -//! Scope note: we do not ingest samples here. The "next query actually -//! returns data" half is covered by the production-process E2E. This file -//! locks down the HTTP-boundary plan-arrival behavior only. A config alone -//! does not make a repeat query servable under the current lazy-SID model. - -use crate::drivers::control_plane_client::{ControlPlaneClient, HttpControlPlaneClient}; -use crate::drivers::query::adapters::AdapterConfig; -use crate::drivers::query::servers::http::{HttpServer, HttpServerConfig}; -use crate::query_engines::ASAPQueryEngine; -#[cfg(test)] -use crate::storage_engines::types::{StreamingConfig, StreamingConfigHandle}; -use axum::{extract::State, routing::post, Router}; -use reqwest::Client; -use serde_json::Value; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex}; -use std::time::{Duration, Instant}; -use tokio::net::TcpListener; -use tokio::time::sleep; - -/// State the mock-control-plane axum handler reaches into. -/// -/// `backend_config_url` is held as `Mutex>` because -/// the control plane has to be started *before* the backend (so the -/// backend can be told where to send its miss notification), but -/// the control plane only needs the backend URL later, when it -/// actually handles a request. This lets us bring both servers -/// up in any order and set the URL once they're both listening. -#[derive(Clone)] -struct MockControlPlaneState { - received_count: Arc, - pushed_plan_ts: Arc>>, - backend_config_url: Arc>>, - plan_yaml: Arc, - http: Client, -} - -/// Hand-authored StreamingConfig the mock control plane pushes when -/// it receives the miss. Shape matches the backend's -/// `StreamingConfig::from_yaml_data` parser (see -/// `asap-query-engine/examples/promql/streaming_config.yaml`). -/// -/// PR 5: `aggregationId` is silently dropped on read; the test derives -/// the expected fingerprint from the YAML's content. -fn canned_plan_yaml(_agg_id: u64, metric: &str) -> String { - format!( - "aggregations: -- aggregationType: Sum - aggregationSubType: '' - labels: - grouping: [] - rollup: [] - aggregated: [] - metric: {metric} - parameters: {{}} - windowSize: 60 - windowType: tumbling - spatialFilter: '' -" - ) -} - -/// Compute the policy-fingerprint u64 the backend will derive when it -/// parses [`canned_plan_yaml`] with the given `metric`. Lets the e2e -/// test assert the exact id without coupling to the fingerprint algo. -fn expected_fp_for(metric: &str) -> u64 { - let yaml = canned_plan_yaml(0, metric); - let data: serde_yaml::Value = serde_yaml::from_str(&yaml).expect("yaml parses"); - let sc = crate::storage_engines::types::StreamingConfig::from_yaml_data(&data) - .expect("yaml decodes"); - *sc.materializations_by_policy_fingerprint - .keys() - .next() - .expect("one agg in the canned plan") -} - -async fn mock_control_plane_plan_handler( - State(state): State, - body: axum::body::Bytes, -) -> axum::http::StatusCode { - state.received_count.fetch_add(1, Ordering::SeqCst); - - // Sanity: HttpControlPlaneClient tags the body with - // `kind: capability_miss`. - let parsed: Value = serde_json::from_slice(&body).unwrap_or(Value::Null); - assert_eq!( - parsed["kind"], "capability_miss", - "mock control plane received non-miss payload: {parsed:?}" - ); - - let Some(backend_url) = state.backend_config_url.lock().unwrap().clone() else { - eprintln!("mock control plane: backend URL not yet set — test bug"); - return axum::http::StatusCode::INTERNAL_SERVER_ERROR; - }; - - match state - .http - .post(backend_url) - .body(state.plan_yaml.to_string()) - .send() - .await - { - Ok(r) if r.status().is_success() => { - *state.pushed_plan_ts.lock().unwrap() = Some(Instant::now()); - axum::http::StatusCode::OK - } - Ok(r) => { - eprintln!( - "mock control plane: backend rejected config: {}", - r.status() - ); - axum::http::StatusCode::INTERNAL_SERVER_ERROR - } - Err(e) => { - eprintln!("mock control plane: backend POST failed: {e}"); - axum::http::StatusCode::INTERNAL_SERVER_ERROR - } - } -} - -async fn start_mock_control_plane(state: MockControlPlaneState) -> u16 { - let app = Router::new() - .route("/api/v1/plan", post(mock_control_plane_plan_handler)) - .with_state(state); - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let port = listener.local_addr().unwrap().port(); - tokio::spawn(async move { - axum::serve(listener, app).await.unwrap(); - }); - sleep(Duration::from_millis(50)).await; - port -} - -async fn start_backend(control_plane_url: String, hot_reload: StreamingConfigHandle) -> u16 { - let _streaming_config = hot_reload.snapshot(); - let engine = Arc::new( - ASAPQueryEngine::new(15_000) - .with_control_plane_client(Arc::new(HttpControlPlaneClient::new(control_plane_url)) - as Arc), - ); - - // No fallback — we want engine-miss to be visible to the - // test and stay out of the hot-vs-cold routing question. - let adapter_config = AdapterConfig::new( - crate::storage_engines::types::enums::QueryProtocol::PrometheusHttp, - crate::storage_engines::types::QueryLanguage::PromQl, - None, - ); - let config = HttpServerConfig { - port: 0, - handle_http_requests: true, - adapter_config, - }; - let idx = std::sync::Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()); - let server = HttpServer::new(config, engine, idx).with_hot_reload_config(hot_reload.clone()); - server - .start_test_server() - .await - .expect("Failed to start backend test server") -} - -async fn poll_until_plan_active( - client: &Client, - backend_url: &str, - expected_agg_id: u64, - timeout: Duration, -) -> Option { - let deadline = Instant::now() + timeout; - while Instant::now() < deadline { - let r = client - .get(format!("{backend_url}/api/v1/streaming-config")) - .send() - .await; - if let Ok(r) = r { - if r.status().is_success() { - let body: Value = r.json().await.unwrap_or(Value::Null); - let ids = body["aggregation_ids"] - .as_array() - .cloned() - .unwrap_or_default(); - if ids.iter().any(|v| v.as_u64() == Some(expected_agg_id)) { - return Some(Instant::now()); - } - } - } - sleep(Duration::from_millis(10)).await; - } - None -} - -/// Bring up a full control-plane+backend pair linked in both -/// directions, ready to answer queries. Returns the backend URL, -/// the control-plane state (for assertions), and the hot-reload -/// handle. -async fn spin_up_loop( - metric: &str, - expected_agg_id: u64, -) -> (String, MockControlPlaneState, StreamingConfigHandle) { - let control_plane_state = MockControlPlaneState { - received_count: Arc::new(AtomicUsize::new(0)), - pushed_plan_ts: Arc::new(Mutex::new(None)), - backend_config_url: Arc::new(Mutex::new(None)), - plan_yaml: Arc::new(canned_plan_yaml(expected_agg_id, metric)), - http: Client::new(), - }; - - // 1. control plane up (no backend URL yet) - let control_plane_port = start_mock_control_plane(control_plane_state.clone()).await; - let control_plane_url = format!("http://127.0.0.1:{control_plane_port}/api/v1/plan"); - - // 2. backend up, with control-plane URL baked in - let hot_reload = StreamingConfigHandle::new(StreamingConfig::default()); - let backend_port = start_backend(control_plane_url, hot_reload.clone()).await; - let backend_url = format!("http://127.0.0.1:{backend_port}"); - - // 3. tell the control plane where to push - *control_plane_state.backend_config_url.lock().unwrap() = - Some(format!("{backend_url}/api/v1/streaming-config")); - - (backend_url, control_plane_state, hot_reload) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn http_capability_miss_feedback_loop_closes_over_http() { - let metric = "http_e2e_metric"; - // PR 5: the on-the-wire agg_id is the policy fingerprint of the - // canned plan's content; derive it here so the assertions match. - let expected_agg_id: u64 = expected_fp_for(metric); - - let (backend_url, control_plane_state, _hot_reload) = - spin_up_loop(metric, expected_agg_id).await; - let client = Client::new(); - - // 1. Fire the capability-miss query. - let t_query = Instant::now(); - let _ = client - .get(format!("{backend_url}/api/v1/query")) - .query(&[("query", format!("sum({metric})").as_str()), ("time", "0")]) - .send() - .await - .expect("query should send") - .bytes() - .await; - - // 2. Poll until the plan is live on the backend. - let plan_ready = poll_until_plan_active( - &client, - &backend_url, - expected_agg_id, - Duration::from_secs(3), - ) - .await - .expect( - "capability-miss feedback loop did not close within 3s over HTTP — \ - the control-plane-notify call, the /api/v1/plan handler, or the \ - /api/v1/streaming-config POST did not complete in time", - ); - - let t_plan_ready_ms = plan_ready.duration_since(t_query).as_millis(); - println!("http-e2e: time_to_plan_ready = {t_plan_ready_ms}ms (1 control-plane hop, localhost)"); - assert!( - t_plan_ready_ms < 2_000, - "time_to_plan_ready = {t_plan_ready_ms}ms, expected < 2000ms" - ); - - // 3. Shape check: the exact agg_id the control plane pushed is - // present on the backend now. - let body: Value = client - .get(format!("{backend_url}/api/v1/streaming-config")) - .send() - .await - .unwrap() - .json() - .await - .unwrap(); - assert_eq!(body["aggregation_count"], 1); - let ids = body["aggregation_ids"].as_array().unwrap(); - assert_eq!(ids.len(), 1); - assert_eq!(ids[0].as_u64().unwrap(), expected_agg_id); - - // 4. Mock control plane actually handled the notify (guards - // against silent short-circuit paths). The engine may - // fire the notify more than once per query — multiple - // code paths (timeline dispatch, capability matching, - // per-segment resolution) can each observe the miss - // before the plan lands — so we assert `>= 1` here. - let count = control_plane_state.received_count.load(Ordering::SeqCst); - assert!( - count >= 1, - "mock control plane should have received at least one miss, got {count}" - ); - assert!( - control_plane_state.pushed_plan_ts.lock().unwrap().is_some(), - "mock control plane should have pushed plan back to backend" - ); -} diff --git a/data_plane/src/tests/mod.rs b/data_plane/src/tests/mod.rs index a7e21b1e6..a6d769261 100644 --- a/data_plane/src/tests/mod.rs +++ b/data_plane/src/tests/mod.rs @@ -1,6 +1,5 @@ pub mod accuracy_empirical_validation_tests; pub mod accuracy_in_promql_response_tests; -pub mod capability_miss_http_e2e_tests; pub mod prometheus_forwarding_tests; pub mod trait_design_tests; From 9e2117d8117aed2f0210e41ba98dc3639099e4e1 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 18:46:59 -0600 Subject: [PATCH 2/2] test: drop the tests for the removed plan endpoints --- control_plane/src/main.rs | 372 -------------------------------------- 1 file changed, 372 deletions(-) diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 3bec1fda0..2e5c407dc 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -1588,15 +1588,6 @@ mod api_tests { serde_json::from_slice(&bytes).unwrap() } - fn plan_spec(metric: &str) -> serde_json::Value { - serde_json::json!({ - "metric_name": metric, - "aggregations": ["quantile"], - "time_window": "5m", - "accuracy_sla": 0.01 - }) - } - // ── AppState.backend_client wiring ─────────────────────────────── /// Default-constructed AppState (no `CONTROLLER_BACKEND_ENDPOINT`) @@ -1625,55 +1616,6 @@ mod api_tests { // ── POST /api/v1/plan ───────────────────────────────────────────────────── - #[tokio::test] - async fn plan_happy_path() { - let (_, app) = test_app(); - let req = Request::builder() - .method("POST") - .uri("/api/v1/plan") - .header("content-type", "application/json") - .body(Body::from(plan_spec("latency").to_string())) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::OK); - let body = body_json(resp).await; - assert_eq!(body["metric"], "latency"); - assert!(body["sketch_type"].as_str().is_some()); - assert!(body["valid_until"].as_str().is_some()); - assert!(body.get("plan_summary").is_none()); - assert!(body["transmission_costs"].is_object()); - } - - #[tokio::test] - async fn plan_invalid_spec_returns_422() { - let (_, app) = test_app(); - let req = Request::builder() - .method("POST") - .uri("/api/v1/plan") - .header("content-type", "application/json") - .body(Body::from(r#"{"metric_name":"","aggregations":["quantile"],"time_window":"5m","accuracy_sla":0.01}"#)) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY); - } - - #[tokio::test] - async fn plan_invalid_aggregation_returns_422() { - let (_, app) = test_app(); - let bad = serde_json::json!({ - "metric_name": "m", "aggregations": ["histogram"], - "time_window": "5m", "accuracy_sla": 0.01 - }); - let req = Request::builder() - .method("POST") - .uri("/api/v1/plan") - .header("content-type", "application/json") - .body(Body::from(bad.to_string())) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY); - } - // ── GET /api/v1/plan/:metric ────────────────────────────────────────────── #[tokio::test] @@ -1687,29 +1629,6 @@ mod api_tests { assert_eq!(resp.status(), StatusCode::NOT_FOUND); } - #[tokio::test] - async fn get_plan_after_post() { - let (_, app) = test_app(); - // POST first - let post_req = Request::builder() - .method("POST") - .uri("/api/v1/plan") - .header("content-type", "application/json") - .body(Body::from(plan_spec("cpu").to_string())) - .unwrap(); - let post_resp = app.clone().oneshot(post_req).await.unwrap(); - assert_eq!(post_resp.status(), StatusCode::OK); - // Then GET - let get_req = Request::builder() - .uri("/api/v1/plan/cpu") - .body(Body::empty()) - .unwrap(); - let get_resp = app.oneshot(get_req).await.unwrap(); - assert_eq!(get_resp.status(), StatusCode::OK); - let body = body_json(get_resp).await; - assert_eq!(body["metric"], "cpu"); - } - // ── GET /api/v1/cost-model ──────────────────────────────────────────────── #[tokio::test] @@ -2738,295 +2657,4 @@ mod api_tests { } assert!(!yaml.contains("name == \"top_endpoint_qps\"")); } - - // ── Regression: archive tier covers all 5 sketched metrics ──────────────── - // - // The backend's `POST /api/v1/storage_routing` handler is an atomic - // per-tenant SWAP — every push replaces the whole tenant's routing - // table. Pre-fix, `handle_plan` posted a single-element - // `metrics:[…]` document per call, so when the demo POSTed - // `/api/v1/plan` for each of the 5 sketched contract metrics in - // sequence, only the LAST metric's entry survived in the backend. - // The other 4 metrics defaulted to `sketch_store` (which has - // no ASAP-tier sketch state for archive-shape queries) → the - // demo's accuracy reducer logged `archive_miss` for those metrics - // even though gorillas3 wrote their TSDB blocks to MinIO and - // Thanos had them indexed. - // - // The fix wires `state.backend_routing_cache` so each - // `handle_plan` cycle posts the **cumulative** routing table. - // This regression test replays the demo's per-metric POST sequence - // against a mock backend, captures every body, and asserts the - // final swap covers all 5 sketched metrics simultaneously. - #[tokio::test] - async fn storage_routing_cumulative_push_covers_all_planned_metrics() { - // Activate the typed-stage-split path (the only path that - // emits storage-routing JSON; the legacy path no-ops). - let _env = EnvVarGuard::set(physical::stage_split::ENV_USE_TYPED_STAGE_SPLIT, "1"); - - // Mock backend that captures every storage-routing body. - // We re-use the mock pattern from `backend_client::tests` — - // an axum router that drains the request body into a shared - // sink. Mounted at the canonical `/api/v1/storage_routing` - // path so `BackendClient`'s URL-rewrite 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/physical-plan", - 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 - } - }), - ); - 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(); - }); - // Brief settle so the bind is observable before the first POST. - tokio::time::sleep(Duration::from_millis(50)).await; - - // Build an AppState with the backend pointed at the mock URL. - // Use the streaming-config alias so `BackendClient` derives - // the matching `/api/v1/storage_routing` URL. - let backend_url = format!("http://{addr}/api/v1/streaming-config"); - let (state, _) = test_app_with_backend(Some(backend_url)); - - // Mount only `/api/v1/plan` — that's the path the demo - // exercises; we don't need bootstrap or other routes. - let app = axum::Router::new().with_state(state.clone()); - - // The two quantile-compatible sketched contract metrics. Each gets a - // separate POST /api/v1/plan, mirroring the demo's - // per-workload plan-emit cycle. - let sketched = [ - "http_requests_total_latency_ms", // DDSketch - "request_size_bytes", // KLL - ]; - - 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", - ); - } - - // Publication is dispatched asynchronously; wait for every accepted - // plan instead of racing the background HTTP tasks. - for _ in 0..100 { - if sink.lock().unwrap().len() == sketched.len() { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - // Drain the mock sink: every plan-emit must have produced exactly one body. - let bodies = sink.lock().unwrap().clone(); - assert_eq!( - bodies.len(), - sketched.len(), - "expected one storage-routing POST per plan; got {} bodies", - bodies.len(), - ); - - // The LAST captured body is the one the backend will leave - // installed (the swap is destructive — last write wins). It - // MUST list all planned metrics, otherwise the swap would - // erase the routing entries for the metrics planned earlier - // in the sequence and the backend would default them to - // `sketch_store` → archive_miss for those metrics' archive - // queries even though gorillas3's TSDB blocks are present in - // MinIO and Thanos has them indexed. - let last: serde_json::Value = - serde_json::from_str(bodies.last().unwrap()).expect("last body is valid JSON"); - let metric_names: std::collections::BTreeSet = last["storage_routing"]["metrics"] - .as_array() - .expect("metrics array") - .iter() - .map(|m| m["name"].as_str().unwrap().to_string()) - .collect(); - for m in &sketched { - assert!( - metric_names.contains(*m), - "final cumulative storage-routing table missing metric `{m}`; \ - contains only {metric_names:?}\nfull body: {}", - bodies.last().unwrap(), - ); - } - - // Each metric entry must carry a `thanos_query` target — the - // archive-tier dispatch that lets backend forward archive-shape - // queries to Thanos. Without this target the metric falls back - // to `default_engine: sketch_store` and the archive miss - // reproduces. - for m in last["storage_routing"]["metrics"].as_array().unwrap() { - let targets = m["targets"].as_array().expect("targets array"); - let engines: Vec<&str> = targets - .iter() - .map(|t| t["engine"].as_str().unwrap()) - .collect(); - assert!( - engines.contains(&"thanos_query"), - "metric `{}` missing `thanos_query` target; engines={engines:?}", - m["name"].as_str().unwrap(), - ); - } - } - - // ── 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 acceptance-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/physical-plan", - 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 - } - }), - ); - 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().with_state(state.clone()); - - // The quantile-compatible sketched contract metrics — the same - // set the sibling `storage_routing_cumulative_push_...` test - // exercises. Each gets a separate `POST /api/v1/plan`. 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"]; - - 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", - ); - } - - // Publication is dispatched asynchronously; wait for every accepted - // plan instead of racing the background HTTP tasks. - for _ in 0..100 { - if sink.lock().unwrap().len() == sketched.len() { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - // Drain the mock sink: every plan-emit must have produced exactly one body. - 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 planned 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["precompute_plan"]["materializations"] - .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(), - ); - } - } }