diff --git a/control_plane/proto/control_plane.proto b/control_plane/proto/control_plane.proto index a7fa9aba5..de1bb002f 100644 --- a/control_plane/proto/control_plane.proto +++ b/control_plane/proto/control_plane.proto @@ -80,18 +80,12 @@ message BackendCollectorConfig { repeated string group_by = 2; } -message PrecomputeJob { - string query_expr = 1; - google.protobuf.Duration granularity = 2; - string sketch_source = 3; - string store_path = 4; -} - message CollectionPlan { AgentCollectorConfig agent_config = 1; GatewayCollectorConfig gateway_config = 2; BackendCollectorConfig backend_config = 3; - repeated PrecomputeJob precompute = 4; + reserved 4; + reserved "precompute"; google.protobuf.Timestamp valid_until = 5; } diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index 20c5336ab..60cf6e6a8 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -10,7 +10,6 @@ //! | `config/backend.rs` | *retired — emitted YAML for a "backend-role" OTel merge collector tier that was never deployed; superseded by the typed L5's [`stage_config::emit_backend_streaming_config_json`] which posts to asapquery-backend's precompute engine over HTTP* | //! | `config/asapquery_backend.rs` | *retired — `generate_streaming_config_yaml` was the legacy single-aggregation `CollectionPlan`-shaped emitter for `POST /api/v1/streaming-config`; under the data plane's atomic `handle.swap(new_config)` it would WIPE sibling `(metric, role)` aggregations on every fire. Replaced by [`backend_push::post_typed_backend_for_role`], which posts a cumulative typed `BackendStageConfig` derived from the shared per-`(metric, role)` cache* | //! | *(new)* | [`backend_push`] | -//! | `config/precompute.rs` | [`precompute`] | //! | `config/stage_config.rs` | [`stage_config`] (TODO: split into `opamp` + `streaming_config` + `inference_config` per design.md §5; deferred from refactor 2026-05 because the 3,020-line monolith mixes OTel-collector YAML emit, ASAPQuery-backend JSON emit, and shared internals — clean split needs ownership reorganisation, not file renames) | //! | `config/stage_config_otap.rs` | [`otap`] | //! | `config/stage_config_telegraf.rs` | [`telegraf`] | @@ -20,7 +19,6 @@ pub mod agent; pub mod backend_push; pub mod monitor; pub mod otap; -pub mod precompute; pub mod stage_config; pub mod telegraf; @@ -29,7 +27,6 @@ pub use backend_push::{ post_typed_backend_for_role, repost_cumulative_backend_config, BackendRoutingCache, PushOutcome, }; pub use otap::emit_otap_dag_yaml; -pub use precompute::{build_precompute_engine_jobs, should_precompute, PrecomputeClient}; pub use stage_config::{ emit_backend_storage_routing, emit_backend_storage_routing_for_tenant, emit_backend_storage_routing_with_prometheus, diff --git a/control_plane/src/emit/precompute.rs b/control_plane/src/emit/precompute.rs deleted file mode 100644 index 7a16435b0..000000000 --- a/control_plane/src/emit/precompute.rs +++ /dev/null @@ -1,323 +0,0 @@ -use anyhow::Context; -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; -use std::time::Duration; - -use crate::pipeline::format_duration; -use crate::types::*; - -// ── Scheduling rule ─────────────────────────────────────────────────────────── - -/// Returns true when a query should be precomputed. -/// -/// Rule (design doc): precompute when `repeat_every < latency_sla`, meaning -/// the query fires more often than the system can recompute it on demand. -pub fn should_precompute(w: &QueryWorkload) -> bool { - match (w.repeat_every, w.latency_sla) { - (Some(re), Some(ls)) => re < ls, - _ => false, - } -} - -/// Builds the list of precompute jobs for a workload. Returns an empty Vec -/// when the workload does not meet the precompute eligibility criterion. -/// -/// The job's `query_expr` is the `build_query_expr()` template derived from -/// the workload's aggregation / metric / filter fields. (An earlier SP-9 -/// path took `query_expr` from `StagedPlan.precompute.query_expr` instead — -/// but that field was never populated, so the template was always the -/// effective output; the dead path was dropped with the legacy L5.) -pub fn build_precompute_engine_jobs(w: &QueryWorkload, backend_addr: &str) -> Vec { - if !should_precompute(w) { - return vec![]; - } - let granularity = w.repeat_every.unwrap(); // safe: should_precompute checked it - - vec![PrecomputeJob { - query_expr: build_query_expr(w), - granularity, - sketch_source: backend_addr.to_string(), - store_path: build_store_path(w), - }] -} - -fn build_query_expr(w: &QueryWorkload) -> String { - let agg_fn = match w.aggregations.first() { - Some(AggType::Cardinality) => "count_distinct_over_time", - Some(AggType::Frequency) => "top_k_over_time", - _ => "quantile_over_time", - }; - let filters: Vec = w - .label_filters - .iter() - .map(|(k, v)| format!("{k}=\"{v}\"")) - .collect(); - let selector = if filters.is_empty() { - w.metric_name.clone() - } else { - format!("{}{{{}}}", w.metric_name, filters.join(",")) - }; - format!( - "{agg_fn}(0.99, {selector}[{}])", - format_duration(w.time_window) - ) -} - -fn build_store_path(w: &QueryWorkload) -> String { - format!( - "precomputed/{}/p99/{}", - w.metric_name, - format_duration(w.time_window) - ) -} - -// ── HTTP client for ASAPQuery precompute API ────────────────────────────────── - -#[derive(Debug, Serialize)] -struct JobRequest { - query: String, - granularity: String, - source: String, - sketch_type: String, - store_path: String, -} - -#[derive(Debug, Clone, Deserialize, Serialize)] -pub struct JobResponse { - pub job_id: String, - pub status: String, - pub created_at: Option>, -} - -pub struct PrecomputeClient { - base_url: String, - client: reqwest::Client, -} - -impl PrecomputeClient { - pub fn new(base_url: impl Into) -> Self { - Self { - base_url: base_url.into(), - client: reqwest::Client::builder() - .timeout(Duration::from_secs(10)) - .build() - .expect("reqwest client"), - } - } - - /// Registers a precompute job with the ASAPQuery engine. - pub async fn register(&self, job: &PrecomputeJob) -> anyhow::Result { - let req = JobRequest { - query: job.query_expr.clone(), - granularity: format_duration(job.granularity), - source: job.sketch_source.clone(), - sketch_type: "ddsketch".into(), - store_path: job.store_path.clone(), - }; - let resp = self - .client - .post(format!("{}/api/v1/precompute/jobs", self.base_url)) - .json(&req) - .send() - .await - .context("POST precompute job")?; - - if !resp.status().is_success() { - anyhow::bail!("precompute API returned {}", resp.status()); - } - resp.json::() - .await - .context("decode job response") - } - - /// Removes a precompute job by ID. - pub async fn deregister(&self, job_id: &str) -> anyhow::Result<()> { - let resp = self - .client - .delete(format!("{}/api/v1/precompute/jobs/{job_id}", self.base_url)) - .send() - .await - .context("DELETE precompute job")?; - - if !resp.status().is_success() { - anyhow::bail!("precompute API returned {}", resp.status()); - } - Ok(()) - } -} - -// ── Tests ───────────────────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - use axum::{ - http::StatusCode, - routing::{delete, post}, - Json, Router, - }; - use serde_json::json; - use std::collections::HashMap; - use tokio::net::TcpListener; - - fn w(repeat_every: Option, latency_sla: Option) -> QueryWorkload { - QueryWorkload { - metric_name: "latency".into(), - label_filters: HashMap::new(), - group_by_labels: vec![], - aggregations: vec![AggType::Quantile], - time_window: Duration::from_secs(300), - repeat_every, - accuracy_sla: 0.01, - latency_sla, - sketch_type_override: None, - exact_required: false, - quantiles: vec![], - } - } - - #[test] - fn should_precompute_true_when_repeat_lt_latency() { - assert!(should_precompute(&w( - Some(Duration::from_secs(60)), - Some(Duration::from_secs(300)) - ))); - } - - #[test] - fn should_precompute_false_when_repeat_geq_latency() { - assert!(!should_precompute(&w( - Some(Duration::from_secs(300)), - Some(Duration::from_secs(60)) - ))); - } - - #[test] - fn should_precompute_false_when_missing_fields() { - assert!(!should_precompute(&w(None, Some(Duration::from_secs(300))))); - assert!(!should_precompute(&w(Some(Duration::from_secs(60)), None))); - } - - #[test] - fn build_jobs_returns_job_when_eligible() { - let workload = w( - Some(Duration::from_secs(60)), - Some(Duration::from_secs(600)), - ); - let jobs = build_precompute_engine_jobs(&workload, "data-plane:4317"); - assert_eq!(jobs.len(), 1); - assert_eq!(jobs[0].sketch_source, "data-plane:4317"); - assert_eq!(jobs[0].granularity, Duration::from_secs(60)); - assert!(jobs[0].query_expr.contains("latency")); - } - - #[test] - fn build_jobs_empty_when_not_eligible() { - let workload = w( - Some(Duration::from_secs(300)), - Some(Duration::from_secs(60)), - ); - let jobs = build_precompute_engine_jobs(&workload, "data-plane:4317"); - assert!(jobs.is_empty()); - } - - #[test] - fn store_path_contains_metric_and_window() { - let workload = w( - Some(Duration::from_secs(60)), - Some(Duration::from_secs(600)), - ); - let jobs = build_precompute_engine_jobs(&workload, "data-plane:4317"); - assert!(jobs[0].store_path.contains("latency")); - assert!(jobs[0].store_path.contains("5m")); - } - - #[test] - fn cardinality_uses_count_distinct_expr() { - let mut workload = w( - Some(Duration::from_secs(60)), - Some(Duration::from_secs(600)), - ); - workload.aggregations = vec![AggType::Cardinality]; - let jobs = build_precompute_engine_jobs(&workload, "data-plane:4317"); - assert!( - jobs[0].query_expr.contains("count_distinct_over_time"), - "got: {}", - jobs[0].query_expr - ); - } - - #[tokio::test] - async fn client_register_success() { - let app = Router::new().route( - "/api/v1/precompute/jobs", - post(|| async { - ( - StatusCode::OK, - Json(json!({ - "job_id": "job-123", - "status": "created", - "created_at": null - })), - ) - }), - ); - let listener = 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(); - }); - - let client = PrecomputeClient::new(format!("http://{addr}")); - let resp = client - .register(&PrecomputeJob { - query_expr: "quantile_over_time(0.99, latency[5m])".into(), - granularity: Duration::from_secs(60), - sketch_source: "data-plane:4317".into(), - store_path: "precomputed/latency/p99/5m".into(), - }) - .await - .unwrap(); - assert_eq!(resp.job_id, "job-123"); - } - - #[tokio::test] - async fn client_register_error_on_bad_status() { - let app = Router::new().route( - "/api/v1/precompute/jobs", - post(|| async { StatusCode::INTERNAL_SERVER_ERROR }), - ); - let listener = 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(); - }); - - let client = PrecomputeClient::new(format!("http://{addr}")); - assert!(client - .register(&PrecomputeJob { - query_expr: "q".into(), - granularity: Duration::from_secs(60), - sketch_source: "s".into(), - store_path: "p".into(), - }) - .await - .is_err()); - } - - #[tokio::test] - async fn client_deregister_success() { - let app = Router::new().route( - "/api/v1/precompute/jobs/:id", - delete(|| async { StatusCode::NO_CONTENT }), - ); - let listener = 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(); - }); - - let client = PrecomputeClient::new(format!("http://{addr}")); - assert!(client.deregister("job-abc").await.is_ok()); - } -} diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index a4fdeda7d..eba92f8e5 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -5,7 +5,6 @@ use control_plane::metrics_exposer; use control_plane::monitor; use control_plane::opamp; use control_plane::physical; -use control_plane::physical::post_asap; use control_plane::pipeline; use control_plane::query_parser; use control_plane::replan; @@ -30,10 +29,8 @@ use std::time::Duration; use tokio::sync::Mutex; use tracing::{info, warn}; -use emit::{ - build_precompute_engine_jobs, generate_agent_collector_config, post_typed_backend_for_role, -}; 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 physical::allocator::SketchAllocator; @@ -612,7 +609,7 @@ struct CompileAndPublishPhysicalPlanResponse { collector_ids: Vec, } -/// Compile one Planner IR decision into the matching backend/Collector views +/// Compile one Planner IR decision into matching Collector, Precompute, and Backend views /// and install them in dependency order. Unlike the legacy planning endpoint, /// this MVP boundary is fail-closed: the Collector plan is never published /// unless the backend accepted the exact matching BackendPlan first. @@ -643,6 +640,31 @@ async fn handle_compile_and_publish_physical_plan( ) .into_response(); } + let precompute_stage_config = BackendStageConfig { + aggregations: bundle.precompute_plan.materializations.clone(), + readouts: Vec::new(), + }; + let precompute_config = + match emit::emit_backend_streaming_config_json(&precompute_stage_config, &[]) { + Ok(config) => config.to_string(), + Err(error) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("failed to emit precompute physical plan: {error}"), + ) + .into_response() + } + }; + if let Err(error) = backend + .post_streaming_config_json_typed(precompute_config) + .await + { + return ( + StatusCode::BAD_GATEWAY, + format!("backend rejected precompute physical plan: {error}"), + ) + .into_response(); + } if let Err(error) = backend .post_backend_plan_typed(bundle.backend_plan.encode_to_vec()) .await @@ -677,14 +699,7 @@ async fn handle_compile_and_publish_physical_plan( // Only the Send-safe compiled bundle crosses an await point. fn compile_physical_plan_request( request: CompileAndPublishPhysicalPlanRequest, -) -> Result< - ( - physical::compiler::CompiledPlanBundle, - Vec, - Duration, - ), - (StatusCode, String), -> { +) -> Result<(physical::compiler::PhysicalPlan, Vec, Duration), (StatusCode, String)> { if request.queries.is_empty() || request.collector_ids.is_empty() { return Err(( StatusCode::UNPROCESSABLE_ENTITY, @@ -767,7 +782,7 @@ async fn handle_plan(State(st): State, Json(spec): Json) -> Err(e) => return (StatusCode::UNPROCESSABLE_ENTITY, e.to_string()).into_response(), }; - let mut plan = st.planner.plan(&workload, Some(&wc)); + let plan = st.planner.plan(&workload, Some(&wc)); // ── L1→L5: parse → optimise → bind → stage. One algebra pipeline. ─────── // When the spec carries a `query_string`, this is the single place the @@ -831,7 +846,6 @@ async fn handle_plan(State(st): State, Json(spec): Json) -> (plan_summary, stage_configs) }; - plan.precompute = build_precompute_engine_jobs(&workload, "data-plane:4317"); // 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 @@ -1055,7 +1069,6 @@ async fn handle_plan(State(st): State, Json(spec): Json) -> "aggregate_by": plan.agent_config.aggregate_by, "valid_until": plan.valid_until, "agents_notified": agents.len(), - "precompute_jobs": plan.precompute.len(), "delta_decision": plan.delta_decision, "transmission_costs": { "raw_bytes_per_sec": cost.raw_bytes_per_sec, diff --git a/control_plane/src/metrics_exposer.rs b/control_plane/src/metrics_exposer.rs index c5fbda86c..b9b47cdfc 100644 --- a/control_plane/src/metrics_exposer.rs +++ b/control_plane/src/metrics_exposer.rs @@ -459,7 +459,6 @@ mod tests { data_sink: AgentDataSink::default(), }, gateway_config: GatewayCollectorConfig { passthrough: true }, - precompute: vec![], valid_until: Utc::now() + chrono::Duration::seconds(valid_secs), delta_decision: Default::default(), transmission_cost_summary: Default::default(), diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 622bcf4a7..847816cad 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -132,10 +132,24 @@ pub struct CollectorPlan { pub materializations: Vec, } +/// Backend-side materialization projection consumed by the streaming +/// precompute engine. This is deliberately config-driven: it contains no +/// PromQL string or ad-hoc scheduler job. The aggregation definitions are +/// emitted to `/api/v1/streaming-config`, where the runtime matches incoming +/// series, maintains windows, and writes content-addressed materializations. #[derive(Debug, Clone)] -pub struct CompiledPlanBundle { +pub struct PrecomputePlan { + pub envelope: PlanEnvelope, + pub materializations: Vec, +} + +/// Complete physical projection of one post-ASAP planning decision. +/// All three child plans share the same envelope and are compiled together. +#[derive(Debug, Clone)] +pub struct PhysicalPlan { pub envelope: PlanEnvelope, pub collector_plans: Vec, + pub precompute_plan: PrecomputePlan, pub backend_plan: BackendPlan, } @@ -185,7 +199,7 @@ impl PhysicalCompiler { &self, request: PlanningRequest, environment: DeploymentEnvironment, - ) -> Result { + ) -> Result { if request.planner_revision != PLANNER_REVISION { return Err(CompileError::PlannerRevision { request: request.planner_revision, @@ -284,11 +298,12 @@ impl PhysicalCompiler { planner_revision: PLANNER_REVISION.into(), capability_snapshot_id: environment.capability_snapshot_id, }; + let backend_stage_config = BackendStageConfig { + aggregations: aggregations.clone(), + readouts, + }; let mut backend_plan = backend_plan::from_stage_config( - &BackendStageConfig { - aggregations, - readouts, - }, + &backend_stage_config, &Vec::::new(), plan_id, environment.observed_at_unix_ms, @@ -310,9 +325,14 @@ impl PhysicalCompiler { materializations: collector_materializations.clone(), }) .collect(); - Ok(CompiledPlanBundle { + let precompute_plan = PrecomputePlan { + envelope: envelope.clone(), + materializations: aggregations, + }; + Ok(PhysicalPlan { envelope, collector_plans, + precompute_plan, backend_plan, }) } @@ -576,6 +596,8 @@ mod tests { ) .expect("compile"); assert_eq!(bundle.collector_plans.len(), 2); + assert_eq!(bundle.precompute_plan.envelope, bundle.envelope); + assert_eq!(bundle.precompute_plan.materializations.len(), 1); assert_eq!(bundle.backend_plan.materializations.len(), 1); assert_eq!( bundle diff --git a/control_plane/src/physical/deployment_cost/delta.rs b/control_plane/src/physical/deployment_cost/delta.rs index f09748d23..f4690311f 100644 --- a/control_plane/src/physical/deployment_cost/delta.rs +++ b/control_plane/src/physical/deployment_cost/delta.rs @@ -517,7 +517,6 @@ mod tests { data_sink: AgentDataSink::default(), }, gateway_config: GatewayCollectorConfig { passthrough: true }, - precompute: vec![], valid_until: Utc::now(), delta_decision: DeltaDecision::default(), transmission_cost_summary: TransmissionCostSummary::default(), diff --git a/control_plane/src/physical/deployment_cost/mod.rs b/control_plane/src/physical/deployment_cost/mod.rs index 9f96ff0b8..5e1abb562 100644 --- a/control_plane/src/physical/deployment_cost/mod.rs +++ b/control_plane/src/physical/deployment_cost/mod.rs @@ -376,7 +376,6 @@ mod tests { data_sink: AgentDataSink::default(), }, gateway_config: GatewayCollectorConfig { passthrough: true }, - precompute: vec![], valid_until: Utc::now(), delta_decision: DeltaDecision::default(), transmission_cost_summary: TransmissionCostSummary::default(), diff --git a/control_plane/src/physical/workload_planner.rs b/control_plane/src/physical/workload_planner.rs index 065273fb9..66d0ba9c7 100644 --- a/control_plane/src/physical/workload_planner.rs +++ b/control_plane/src/physical/workload_planner.rs @@ -360,7 +360,6 @@ impl DeploymentPlanCompiler { data_sink: AgentDataSink::default(), }, gateway_config: GatewayCollectorConfig { passthrough: true }, - precompute: vec![], valid_until, delta_decision: DeltaDecision::default(), transmission_cost_summary: TransmissionCostSummary::default(), @@ -400,7 +399,6 @@ impl DeploymentPlanCompiler { data_sink: AgentDataSink::default(), }, gateway_config: GatewayCollectorConfig { passthrough: true }, - precompute: vec![], valid_until, delta_decision: DeltaDecision::default(), transmission_cost_summary: TransmissionCostSummary::default(), diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index 9f72af609..93cfb0746 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -22,9 +22,9 @@ use tracing::{info, warn}; use crate::backend_client::BackendClient; use crate::emit::{ - build_precompute_engine_jobs, collect_metric_to_family, emit_for_runtime, - extend_edge_with_demo_plumbing, generate_agent_collector_config, post_typed_backend_for_role, - repost_cumulative_backend_config, AgentRuntime, PushOutcome, WorkloadRegistry, + collect_metric_to_family, emit_for_runtime, extend_edge_with_demo_plumbing, + generate_agent_collector_config, post_typed_backend_for_role, repost_cumulative_backend_config, + AgentRuntime, PushOutcome, WorkloadRegistry, }; use crate::monitor::Scraper; use crate::opamp::{OpampServer, RemoteConfig}; @@ -447,8 +447,7 @@ impl Replanner { // previously established baseline — the whole point of a re-plan is to // re-optimise with current EMA data. self.planner.reset(metric); - let mut plan = self.planner.plan(&workload, Some(&wc)); - plan.precompute = build_precompute_engine_jobs(&workload, "data-plane:4317"); + let plan = self.planner.plan(&workload, Some(&wc)); self.plan_store.set(metric, role, plan.clone()); // Push agent config only to agents registered for this specific @@ -925,7 +924,6 @@ mod tests { data_sink: AgentDataSink::default(), }, gateway_config: GatewayCollectorConfig { passthrough: true }, - precompute: vec![], valid_until: Utc::now() + chrono::Duration::seconds(3600), delta_decision: DeltaDecision::default(), transmission_cost_summary: TransmissionCostSummary::default(), diff --git a/control_plane/src/store/mod.rs b/control_plane/src/store/mod.rs index b5c3e451b..98eba7b96 100644 --- a/control_plane/src/store/mod.rs +++ b/control_plane/src/store/mod.rs @@ -219,7 +219,6 @@ mod tests { data_sink: AgentDataSink::default(), }, gateway_config: GatewayCollectorConfig { passthrough: true }, - precompute: vec![], valid_until, delta_decision: DeltaDecision::default(), transmission_cost_summary: TransmissionCostSummary::default(), diff --git a/control_plane/src/types.rs b/control_plane/src/types.rs index 81b9de7b0..45dac081f 100644 --- a/control_plane/src/types.rs +++ b/control_plane/src/types.rs @@ -598,14 +598,6 @@ pub struct GatewayCollectorConfig { pub passthrough: bool, } -#[derive(Debug, Clone)] -pub struct PrecomputeJob { - pub query_expr: String, - pub granularity: Duration, - pub sketch_source: String, - pub store_path: String, -} - // ── Per-stage resource budgets ──────────────────────────────────────────────── /// Per-stage resource caps. The agent cap feeds the physical planner's @@ -644,7 +636,6 @@ impl StageResourceBudgets { pub struct CollectionPlan { pub agent_config: AgentCollectorConfig, pub gateway_config: GatewayCollectorConfig, - pub precompute: Vec, pub valid_until: DateTime, /// Resolved delta transmission decision and rationale. pub delta_decision: DeltaDecision, diff --git a/data_plane/src/drivers/control_plane_client/config_fetcher.rs b/data_plane/src/drivers/control_plane_client/config_fetcher.rs index fdccda492..0f86b4d19 100644 --- a/data_plane/src/drivers/control_plane_client/config_fetcher.rs +++ b/data_plane/src/drivers/control_plane_client/config_fetcher.rs @@ -51,8 +51,6 @@ pub struct ControlPlanePlan { #[serde(default)] pub aggregate_by: Vec, #[serde(default)] - pub precompute_jobs: usize, - #[serde(default)] pub delta_decision: serde_json::Value, #[serde(default)] pub staged_plan: Option, diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index f947f9436..7edeb40b3 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -9,7 +9,7 @@ use axum::{ }; use serde_json::Value; use std::collections::HashMap; -use std::sync::{Arc, RwLock}; +use std::sync::Arc; use std::time::Instant; use tokio::net::TcpListener; use tracing::{debug, info, warn}; @@ -23,73 +23,6 @@ use crate::query_engines::ASAPQueryEngine; use crate::storage_engines::types::StorageBackend; use asap_types::Statistic; -// ─── Control-plane-pushed precompute job registry ──────────────────────────── -// -// The control plane's `PrecomputeClient` (control_plane/src/emit/precompute.rs) -// registers / cancels precompute jobs via: -// -// * `POST /api/v1/precompute/jobs` — body -// `{query, granularity, source, sketch_type, store_path}`; response -// `{job_id, status, created_at}`. -// * `DELETE /api/v1/precompute/jobs/{job_id}` — 204 on success, 404 -// when the id is unknown. -// -// The handler is intentionally minimal: it tracks the spec in an -// in-memory map and acknowledges the call. No precompute work is -// scheduled — that's a later wiring. The goal is to make the -// control plane's calls succeed instead of 404 so the control plane can -// progress its plan-push loop end-to-end. - -/// Body shape posted by the control plane's `PrecomputeClient::register`. -/// Matches `control_plane/src/emit/precompute.rs::JobRequest`. -#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] -pub struct PrecomputeJobSpec { - pub query: String, - pub granularity: String, - pub source: String, - pub sketch_type: String, - pub store_path: String, -} - -/// In-memory map of `job_id → spec`, populated by the -/// `POST /api/v1/precompute/jobs` handler and drained by the matching -/// DELETE handler. Wrapped in an `Arc>` so the axum state -/// can clone freely; the lock is held briefly per request and is -/// uncontended in practice (job count is small). -#[derive(Clone, Default)] -pub struct PrecomputeJobRegistry { - inner: Arc>>, -} - -impl PrecomputeJobRegistry { - pub fn new() -> Self { - Self::default() - } - - /// Insert a new job with a fresh UUID. Returns the generated id. - pub fn insert(&self, spec: PrecomputeJobSpec) -> String { - let job_id = uuid::Uuid::new_v4().to_string(); - let mut guard = self.inner.write().expect("poisoned"); - guard.insert(job_id.clone(), spec); - job_id - } - - /// Remove a job. Returns `true` when the id was present. - pub fn remove(&self, job_id: &str) -> bool { - let mut guard = self.inner.write().expect("poisoned"); - guard.remove(job_id).is_some() - } -} - -impl std::fmt::Debug for PrecomputeJobRegistry { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let n = self.inner.read().map(|g| g.len()).unwrap_or(0); - f.debug_struct("PrecomputeJobRegistry") - .field("jobs", &n) - .finish() - } -} - /// Per-query engine override header (Phase-6 accuracy reducer). /// /// When the client sets `X-ASAP-Engine: ` (or the @@ -213,12 +146,6 @@ pub struct HttpServer { /// normal routing-table path (which goes to Thanos for the cold /// archive and observes the 60–90 s flush gap). probe_cache: Option>, - /// In-memory job-spec map populated by the control plane's - /// `POST /api/v1/precompute/jobs` calls. Always present (an empty - /// `Default` registry is fine for binaries that never wire the - /// control plane). Future PRs will plumb this into the precompute - /// engine; today the handler just acks the call. - precompute_jobs: PrecomputeJobRegistry, } #[derive(Clone)] @@ -244,8 +171,6 @@ struct AppState { data_retention_ms: Option, /// See [`HttpServer::probe_cache`]. probe_cache: Option>, - /// See [`HttpServer::precompute_jobs`]. - precompute_jobs: PrecomputeJobRegistry, } impl HttpServer { @@ -270,7 +195,6 @@ impl HttpServer { backfill: None, data_retention_ms: None, probe_cache: None, - precompute_jobs: PrecomputeJobRegistry::new(), } } @@ -406,15 +330,6 @@ impl HttpServer { self } - /// Attach a [`PrecomputeJobRegistry`] used by the control-plane-pushed - /// `POST /api/v1/precompute/jobs` and matching `DELETE` endpoints. - /// Callers that don't override this share the per-server default - /// (an empty in-memory map populated by the registration handler). - pub fn with_precompute_jobs(mut self, registry: PrecomputeJobRegistry) -> Self { - self.precompute_jobs = registry; - self - } - pub async fn run(self) -> Result<(), Box> { srv_metrics::register_all(); @@ -443,7 +358,6 @@ impl HttpServer { backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, probe_cache: self.probe_cache.clone(), - precompute_jobs: self.precompute_jobs.clone(), }; let range_query_endpoint = adapter.get_range_query_endpoint(); @@ -456,20 +370,6 @@ impl HttpServer { .route(runtime_info_path, get(handle_runtime_info)) .route(runtime_info_path, post(handle_runtime_info)) .route("/metrics", get(handle_metrics)) - // Control plane integration endpoints - .route("/api/v1/precompute", post(handle_precompute_job)) - // Control plane's `PrecomputeClient` (control_plane/src/emit/precompute.rs) - // posts to `/jobs` and DELETEs by job_id. Tracks the spec - // in memory; the `/api/v1/precompute` route stays for - // legacy `{query_expr, granularity_secs, start, end}` callers. - .route( - "/api/v1/precompute/jobs", - post(handle_post_precompute_job_register), - ) - .route( - "/api/v1/precompute/jobs/:job_id", - axum::routing::delete(handle_delete_precompute_job), - ) .route("/api/v1/health", get(handle_health)) .route("/api/v1/store/metrics", get(handle_store_metrics)) .route( @@ -543,7 +443,6 @@ impl HttpServer { backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, probe_cache: self.probe_cache.clone(), - precompute_jobs: self.precompute_jobs.clone(), }; let range_query_endpoint = adapter.get_range_query_endpoint(); @@ -588,14 +487,6 @@ impl HttpServer { "/api/v1/db/backfill/jobs/:job_id", get(handle_get_backfill_job).delete(handle_delete_backfill_job), ) - .route( - "/api/v1/precompute/jobs", - post(handle_post_precompute_job_register), - ) - .route( - "/api/v1/precompute/jobs/:job_id", - axum::routing::delete(handle_delete_precompute_job), - ) .with_state(app_state); let listener = TcpListener::bind("127.0.0.1:0").await?; @@ -5230,192 +5121,6 @@ aggregations: saw probe value {now_ms} leaked into response: {body}", ); } - - // ── Control-plane-pushed precompute job registry tests ─────────────────── - - /// `POST /api/v1/precompute/jobs` returns 200 + a `job_id`; the - /// matching DELETE returns 204 the first time and 404 on the - /// second call. - #[tokio::test] - async fn http_precompute_jobs_register_then_delete_roundtrip() { - let port = setup_test_server_with_router(StorageBackend::SketchStore, Vec::new()).await; - let client = Client::new(); - - let body = serde_json::json!({ - "query": "quantile_over_time(0.99, http_requests_total[5m])", - "granularity": "60s", - "source": "backend:4317", - "sketch_type": "ddsketch", - "store_path": "precomputed/http_requests_total/p99/5m"}); - let resp = client - .post(format!("http://127.0.0.1:{port}/api/v1/precompute/jobs")) - .json(&body) - .send() - .await - .expect("send ok"); - assert_eq!(resp.status().as_u16(), 200, "register must 200"); - let resp_body: serde_json::Value = resp.json().await.expect("json"); - let job_id = resp_body["job_id"].as_str().expect("job_id").to_string(); - assert!(!job_id.is_empty(), "job_id must be non-empty"); - assert_eq!(resp_body["status"], "created"); - - let resp = client - .delete(format!( - "http://127.0.0.1:{port}/api/v1/precompute/jobs/{job_id}" - )) - .send() - .await - .expect("send ok"); - assert_eq!( - resp.status().as_u16(), - 204, - "first delete must 204, got {}", - resp.status() - ); - - let resp = client - .delete(format!( - "http://127.0.0.1:{port}/api/v1/precompute/jobs/{job_id}" - )) - .send() - .await - .expect("send ok"); - assert_eq!( - resp.status().as_u16(), - 404, - "second delete must 404 (job already removed)", - ); - } - - /// `DELETE /api/v1/precompute/jobs/{unknown}` returns 404. - #[tokio::test] - async fn http_precompute_jobs_delete_unknown_id_returns_404() { - let port = setup_test_server_with_router(StorageBackend::SketchStore, Vec::new()).await; - let client = Client::new(); - let resp = client - .delete(format!( - "http://127.0.0.1:{port}/api/v1/precompute/jobs/{}", - "never-registered" - )) - .send() - .await - .expect("send ok"); - assert_eq!(resp.status().as_u16(), 404); - } -} - -// ── Controller integration: PrecomputeJob execution ────────────────────────── - -/// Request body from DataCollector controller's PrecomputeJob. -#[derive(serde::Deserialize)] -struct PrecomputeJobRequest { - /// PromQL expression to evaluate against stored sketches. - query_expr: String, - /// Window granularity in seconds. - #[serde(default)] - granularity_secs: u64, - /// Start timestamp (unix seconds). 0 = use earliest available. - #[serde(default)] - start: f64, - /// End timestamp (unix seconds). 0 = use latest available. - #[serde(default)] - end: f64, -} - -/// Execute a precompute job from the DataCollector controller. -/// -/// POST /api/v1/precompute -/// -/// The controller creates PrecomputeJobs when a query's upper sub-tree -/// (e.g., TopK or a histogram-quantile-shaped Aggregate{Quantile(φ)}) -/// requires evaluation on merged sketches. -/// This endpoint receives that job and runs it against the SketchStore. -async fn handle_precompute_job( - State(state): State, - axum::Json(req): axum::Json, -) -> axum::response::Response { - use axum::http::StatusCode; - use axum::response::IntoResponse; - - let time = if req.end > 0.0 { - req.end - } else { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs_f64() - }; - - info!( - query = %req.query_expr, - start = %req.start, - end = %req.end, - granularity_secs = %req.granularity_secs, - time = %time, - "Executing precompute job from controller" - ); - - // B7.5: legacy `handle_query_promql` retired; route through the - // modern `execute(&str)` trait surface. - use crate::query_engines::routing::query_engine_routing::QueryEngine; - let _ = time; - match state.query_engine.execute(&req.query_expr).await { - Ok(result) => { - let body = serde_json::json!({ - "status": "success", - "data": { - "result_type": "precompute", - "key_by": "{}", - "result": format!("{:?}", result)} - }); - (StatusCode::OK, axum::Json(body)).into_response() - } - Err(_) => { - // Query not answerable by sketches — return 404 with hint - let body = serde_json::json!({ - "status": "error", - "error": "query not answerable by stored sketches", - "hint": "ensure the metric has been ingested via OTLP/Kafka and a matching query_config exists" - }); - (StatusCode::NOT_FOUND, axum::Json(body)).into_response() - } - } -} - -/// `POST /api/v1/precompute/jobs` — register a controller-pushed -/// precompute job. Body matches `controller/src/config/precompute.rs`'s -/// `JobRequest` (`{query, granularity, source, sketch_type, store_path}`). -/// Returns `200 OK` with `{job_id, status, created_at}`. -/// -/// The handler is intentionally minimal: it stores the spec in an -/// in-memory map and acknowledges. No precompute work is scheduled — -/// future PRs will plumb this into the precompute engine. -async fn handle_post_precompute_job_register( - State(state): State, - axum::Json(spec): axum::Json, -) -> Response { - let job_id = state.precompute_jobs.insert(spec); - let body = serde_json::json!({ - "job_id": job_id, - "status": "created", - "created_at": chrono::Utc::now().to_rfc3339()}); - (StatusCode::OK, axum::Json(body)).into_response() -} - -/// `DELETE /api/v1/precompute/jobs/:job_id` — drop a registered job. -/// Returns `204 No Content` on success and `404` when unknown. -async fn handle_delete_precompute_job( - State(state): State, - axum::extract::Path(job_id): axum::extract::Path, -) -> Response { - if state.precompute_jobs.remove(&job_id) { - (StatusCode::NO_CONTENT, ()).into_response() - } else { - let body = serde_json::json!({ - "status": "error", - "error": format!("precompute job '{job_id}' not found")}); - (StatusCode::NOT_FOUND, axum::Json(body)).into_response() - } } /// Health check endpoint for DataCollector controller to verify backend is alive. diff --git a/docs/developer_docs/control-plane/physical-compiler.md b/docs/developer_docs/control-plane/physical-compiler.md index 0eaf96562..08814eabf 100644 --- a/docs/developer_docs/control-plane/physical-compiler.md +++ b/docs/developer_docs/control-plane/physical-compiler.md @@ -5,10 +5,12 @@ ## Current implementation boundary -The compiler consumes ASAPPlanner types pinned to revision `3afcba6`, selects +The compiler consumes ASAPPlanner types pinned to the revision exposed as +`physical::compiler::PLANNER_REVISION`, selects from Planner's legal candidate space with backend-owned cost and evidence -inputs, and emits one `CompiledPlanBundle`. The bundle contains a CollectorPlan -for every target collector and the matching BackendPlan. Legacy +inputs, and emits one `PhysicalPlan`. The plan contains CollectorPlan, +PrecomputePlan, and BackendPlan projections compiled from the same decision +for every target collector. Legacy `StageAllocator`/`ThreeStageEmitter` paths remain for older publication flows; they are not a second semantic planner. @@ -30,10 +32,11 @@ PlanningRequest ASAPPlanner candidate selection | v -PhysicalCompiler -------> CompiledPlanBundle - | | - v v - CollectorPlan BackendPlan +PhysicalCompiler -------> PhysicalPlan + | | | + v v v + Collector Precompute Backend + Plan Plan Plan ``` - **Planner selection boundary** is @@ -41,8 +44,8 @@ PhysicalCompiler -------> CompiledPlanBundle candidates and commits only a legal candidate. - **Physical compiler** adds backend-owned placement, windows, transport, and runtime capabilities without changing logical semantics. -- **Plan bundle** is the only output passed to publication. CollectorPlan and - BackendPlan are created together and share identities. +- **PhysicalPlan** is the only output passed to publication. Its CollectorPlan, + PrecomputePlan, and BackendPlan projections are created together and share identities. Logical query parsing, summary alternatives, guarantees, and candidate search remain public ASAPPlanner interfaces. Runtime publication is documented in @@ -96,7 +99,7 @@ impl PhysicalCompiler { &self, request: PlanningRequest, environment: DeploymentEnvironment, - ) -> Result; + ) -> Result; } ``` @@ -108,9 +111,10 @@ pub struct DeploymentEnvironment { pub max_evidence_age_ms: u64, } -pub struct CompiledPlanBundle { +pub struct PhysicalPlan { pub envelope: PlanEnvelope, pub collector_plans: Vec, // complete per-target projections + pub precompute_plan: PrecomputePlan, // backend streaming materializations pub backend_plan: BackendPlan, } ``` @@ -122,6 +126,7 @@ Supporting public types: | `DeploymentEnvironment` | Target collector IDs, capability snapshot identity, planning time, and evidence freshness policy. | | `PlanEnvelope` | Shared deterministic `plan_id`, generation time, capability snapshot, and Planner revision. | | `CollectorPlan` | Serializable execution projection consumed by ASAPCollector. | +| `PrecomputePlan` | Config-driven aggregation/window projection consumed by the backend streaming precompute engine. | | `BackendPlan` | Versioned public data-plane materialization/routing contract defined in this repository. | Current MVP limits are explicit: time-series sources and sketch @@ -135,6 +140,7 @@ Output definitions: | --- | --- | | `envelope` | Shared plan/version/activation/compatibility identity. | | `collector_plans` | One plan per targeted collector, following ASAPCollector's public CollectorPlan schema. | +| `precompute_plan` | Aggregation definitions emitted to `/api/v1/streaming-config`; contains no query-string jobs. | | `backend_plan` | Matching data-plane materialization and routing contract. | ### What the compiler puts in CollectorPlan