diff --git a/Cargo.lock b/Cargo.lock index 9eb7cadb..5f1fb4b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -343,7 +343,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" dependencies = [ "asap-types", "serde", @@ -354,7 +354,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" dependencies = [ "asap-types", "promql-parser 0.10.0", @@ -374,7 +374,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=378a7547ede629a64e84c9f7c810226ce196cce9#378a7547ede629a64e84c9f7c810226ce196cce9" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=c3410d14865497758d212e4265ad25c782187de1#c3410d14865497758d212e4265ad25c782187de1" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 0c276d99..52d7961c 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -93,8 +93,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -102,7 +102,7 @@ asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/examples/workload_cost_manifest.rs b/control_plane/examples/workload_cost_manifest.rs new file mode 100644 index 00000000..723fe8ed --- /dev/null +++ b/control_plane/examples/workload_cost_manifest.rs @@ -0,0 +1,28 @@ +//! Emit pricing requirements; never fabricate quotes or publish a plan. +use control_plane::physical::{ + compiler::BackendLocalPlanningSnapshot, compiler::PhysicalCompiler, workload_cost, +}; + +fn main() -> Result<(), Box> { + let path = std::env::args() + .nth(1) + .ok_or("usage: workload_cost_manifest SNAPSHOT.json")?; + let snapshot: BackendLocalPlanningSnapshot = + serde_json::from_str(&std::fs::read_to_string(path)?)?; + let (request, environment) = snapshot.planning_request()?; + let manifests = workload_cost::with_exact_alternative(request)? + .into_iter() + .filter_map(|candidate| { + let queries = candidate.queries.clone(); + PhysicalCompiler + .compile(candidate, environment.clone()) + .and_then(|plan| workload_cost::manifest(&plan, &queries)) + .ok() + }) + .collect::>(); + if manifests.is_empty() { + return Err("no bindable workload cost manifests".into()); + } + println!("{}", serde_json::to_string_pretty(&manifests)?); + Ok(()) +} diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index 230f7095..de50a861 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -396,6 +396,33 @@ impl BackendClient { } } + pub async fn discard_staged_physical_plan( + &self, + plan_id: u64, + plan_version: u64, + ) -> std::result::Result<(), BackendPostError> { + let response = self + .http + .post(format!( + "{}/discard", + derive_physical_plan_url(&self.endpoint) + )) + .json(&serde_json::json!({"plan_id": plan_id, "plan_version": plan_version})) + .send() + .await + .map_err(classify_reqwest_error)?; + let status = response.status(); + if status.is_success() { + Ok(()) + } else { + Err(classify_http_status( + status, + response.text().await.unwrap_or_default(), + "PhysicalPlan discard POST", + )) + } + } + pub async fn activate_physical_plan( &self, plan_id: u64, diff --git a/control_plane/src/backend_plan/from_stage_config.rs b/control_plane/src/backend_plan/from_stage_config.rs index a840420a..aa7079c4 100644 --- a/control_plane/src/backend_plan/from_stage_config.rs +++ b/control_plane/src/backend_plan/from_stage_config.rs @@ -264,6 +264,7 @@ mod tests { spatial_filter: String::new(), grouping, item_label: None, + heap_update_mode: None, aggregation_input: AggregationInput::SketchEnvelope, } } @@ -514,6 +515,7 @@ mod tests { spatial_filter: String::new(), grouping: vec!["zone".to_string()], item_label: None, + heap_update_mode: None, aggregation_input: AggregationInput::Raw, }; let cfg = BackendStageConfig { diff --git a/control_plane/src/emit/backend_push.rs b/control_plane/src/emit/backend_push.rs index a748377d..cdf49b93 100644 --- a/control_plane/src/emit/backend_push.rs +++ b/control_plane/src/emit/backend_push.rs @@ -746,6 +746,7 @@ mod tests { ), grouping: vec![], item_label: None, + heap_update_mode: None, spatial_filter: String::new(), window_secs: 60, aggregation_input: AggregationInput::SketchEnvelope, diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 8155a20e..6582b79b 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -2986,6 +2986,11 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa obj.insert("item_label".to_string(), JsonValue::String(label.clone())); } } + if let Some(mode) = agg.heap_update_mode { + if let Some(obj) = parameters.as_object_mut() { + obj.insert("weight_mode".into(), JsonValue::String(mode.into())); + } + } let aggregation_input = match agg.aggregation_input { AggregationInput::SketchEnvelope => "sketch_envelope", AggregationInput::Raw => "raw", @@ -3206,6 +3211,7 @@ mod tests { spatial_filter: String::new(), grouping: Vec::new(), item_label: None, + heap_update_mode: None, aggregation_input, } } diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 8e0ac885..be6dc44d 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -546,6 +546,10 @@ async fn main() { let app = Router::new() .route("/api/v1/plan", post(handle_plan)) + .route( + "/api/v1/physical-plan/cost-manifests", + post(handle_workload_cost_manifests), + ) .route( "/api/v1/physical-plan/compile-and-publish", post(handle_compile_and_publish_physical_plan), @@ -590,6 +594,8 @@ struct PhysicalPlanQueryRequest { #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct CompileAndPublishPhysicalPlanRequest { + #[serde(default)] + workload_cost_evidence: Option, queries: Vec, collector_ids: Vec, capability_snapshot_id: String, @@ -613,11 +619,13 @@ fn default_physical_plan_timeout_ms() -> u64 { #[derive(Debug, Serialize)] struct CompileAndPublishPhysicalPlanResponse { + cost_comparison: Option, plan_id: u64, plan_version: u64, status: &'static str, generated_at_unix_ms: u64, collector_ids: Vec, + lifecycle_estimates: Vec, } /// Compile one Planner IR decision into matching Collector, Precompute, and Backend views @@ -628,9 +636,18 @@ async fn handle_compile_and_publish_physical_plan( State(st): State, Json(request): Json, ) -> impl IntoResponse { - let (bundle, collector_ids, apply_timeout, adaptation_evidence) = - match compile_physical_plan_request(request) { - Ok(compiled) => compiled, + let (bundle, collector_ids, apply_timeout, adaptation_evidence, _) = + match compile_physical_plan_request(request, false) { + Ok((Some(bundle), ids, timeout, adaptation, manifests)) => { + (bundle, ids, timeout, adaptation, manifests) + } + Ok((None, ..)) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "publication requires a selected plan", + ) + .into_response() + } Err(response) => return response.into_response(), }; @@ -674,9 +691,12 @@ async fn handle_compile_and_publish_physical_plan( .publish_collector_plans(&bundle.collector_plans, apply_timeout) .await { + let cleanup = backend + .discard_staged_physical_plan(bundle.envelope.plan_id, bundle.envelope.plan_version) + .await; return ( StatusCode::BAD_GATEWAY, - format!("collector physical-plan publication failed: {error}"), + format!("collector physical-plan publication failed: {error}; staged backend cleanup: {cleanup:?}"), ) .into_response(); } @@ -707,11 +727,13 @@ async fn handle_compile_and_publish_physical_plan( } Json(CompileAndPublishPhysicalPlanResponse { + cost_comparison: bundle.cost_comparison, plan_id: bundle.envelope.plan_id, plan_version: bundle.envelope.plan_version, status: "active", generated_at_unix_ms: bundle.envelope.generated_at_unix_ms, collector_ids, + lifecycle_estimates: bundle.lifecycle_estimates, }) .into_response() } @@ -720,12 +742,14 @@ 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, + manifests_only: bool, ) -> Result< ( - physical::compiler::PhysicalPlan, + Option, Vec, Duration, Vec, + Vec, ), (StatusCode, String), > { @@ -759,6 +783,7 @@ fn compile_physical_plan_request( .unwrap_or_default() .as_millis() as u64; let mut queries = Vec::with_capacity(request.queries.len()); + let mut canonical_roots = Vec::with_capacity(request.queries.len()); for query in request.queries { if query.query_id.trim().is_empty() || query.metric.trim().is_empty() @@ -773,15 +798,11 @@ fn compile_physical_plan_request( Ok(expr) => expr, Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; - let post_asap = match physical::compiler::select_post_asap( - &expr, - query.accuracy.clone(), - &query.lifecycle, - request.evidence.get(&query.query_id), - ) { + let post_asap = match control_plane::planner_selection::keep_pre_asap(&expr) { Ok(plan) => plan, Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; + canonical_roots.push(std::rc::Rc::new(expr)); queries.push(physical::compiler::PlanningQuery { query_id: query.query_id, query_string: query.query_string, @@ -798,36 +819,91 @@ fn compile_physical_plan_request( }); } - let bundle = match physical::compiler::PhysicalCompiler.compile( - physical::compiler::PlanningRequest { - queries, - evidence: request.evidence, - planner_revision: request.planner_revision, - }, - physical::compiler::DeploymentEnvironment { - target: physical::compiler::PhysicalDeploymentTarget::DistributedCollectors, - collector_ids: request.collector_ids.clone(), - capability_snapshot_id: request.capability_snapshot_id, - observed_at_unix_ms: now, - max_evidence_age_ms: request.max_evidence_age_ms, - plan_version: request.plan_version, - activation_unix_ms: request.activation_unix_ms, - expiry_unix_ms: request.expiry_unix_ms, - backend_compat: request.backend_compat, - }, - ) { + if let Err(error) = + physical::compiler::select_workload_roots(&mut queries, canonical_roots, &request.evidence) + { + return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())); + } + + let planning_request = physical::compiler::PlanningRequest { + queries, + evidence: request.evidence, + planner_revision: request.planner_revision, + }; + let environment = physical::compiler::DeploymentEnvironment { + target: physical::compiler::PhysicalDeploymentTarget::DistributedCollectors, + collector_ids: request.collector_ids.clone(), + capability_snapshot_id: request.capability_snapshot_id, + observed_at_unix_ms: now, + max_evidence_age_ms: request.max_evidence_age_ms, + plan_version: request.plan_version, + activation_unix_ms: request.activation_unix_ms, + expiry_unix_ms: request.expiry_unix_ms, + backend_compat: request.backend_compat, + }; + let candidates = physical::workload_cost::with_exact_alternative(planning_request.clone()) + .map_err(|error| (StatusCode::UNPROCESSABLE_ENTITY, error.to_string()))?; + let manifests: Vec<_> = candidates + .iter() + .filter_map(|candidate| { + physical::compiler::PhysicalCompiler + .compile(candidate.clone(), environment.clone()) + .and_then(|plan| physical::workload_cost::manifest(&plan, &candidate.queries)) + .ok() + }) + .collect(); + let apply_timeout = Duration::from_millis(request.apply_timeout_ms); + // Quote preparation enumerates feasible bindings; it does not select the + // default warm candidate, which may be unavailable while exact is valid. + if manifests_only { + if manifests.is_empty() { + return Err(( + StatusCode::UNPROCESSABLE_ENTITY, + "no bindable workload cost manifests".into(), + )); + } + return Ok(( + None, + request.collector_ids, + apply_timeout, + request.runtime_adaptation_evidence, + manifests, + )); + } + let compiled = match request.workload_cost_evidence { + Some(evidence) => physical::workload_cost::select(candidates, environment, &evidence), + None => physical::compiler::PhysicalCompiler.compile(planning_request, environment), + }; + let bundle = match compiled { Ok(bundle) => bundle, Err(error) => return Err((StatusCode::UNPROCESSABLE_ENTITY, error.to_string())), }; - let apply_timeout = Duration::from_millis(request.apply_timeout_ms); Ok(( - bundle, + Some(bundle), request.collector_ids, apply_timeout, request.runtime_adaptation_evidence, + manifests, )) } +/// Read-only preparation: no OpAMP, staging, activation or data-plane writes. +async fn handle_workload_cost_manifests( + Json(request): Json, +) -> impl IntoResponse { + if request.workload_cost_evidence.is_some() { + return ( + StatusCode::UNPROCESSABLE_ENTITY, + "omit quotes when requesting manifests", + ) + .into_response(); + } + match compile_physical_plan_request(request, true) { + Ok((_, _, _, _, manifests)) => Json(manifests).into_response(), + Err(error) => error.into_response(), + } +} + // ── Handlers ────────────────────────────────────────────────────────────────── async fn handle_plan(State(st): State, Json(spec): Json) -> impl IntoResponse { @@ -2033,6 +2109,39 @@ mod api_tests { use http_body_util::BodyExt; use tower::ServiceExt; + // A missing warm implementation must not hide the executable exact quote. + #[tokio::test] + async fn cost_manifests_survive_unavailable_warm_candidate() { + let snapshot: physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_str( + include_str!("../../docs/examples/asapquery-planning-snapshot.json"), + ) + .unwrap(); + let (planning, _) = snapshot.planning_request().unwrap(); + let query = &planning.queries[0]; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64; + let request = serde_json::from_value(serde_json::json!({ + "queries": [{ + "query_id": query.query_id, "query_string": query.query_string, + "metric": "m", "window_secs": 60, "accuracy": query.accuracy, + "lifecycle": query.lifecycle, "window_implementations": [] + }], + "collector_ids": ["test"], "capability_snapshot_id": "test", + "planner_revision": physical::compiler::PLANNER_REVISION, + "max_evidence_age_ms": 60000, "plan_version": 1, + "activation_unix_ms": now, "backend_compat": control_plane::backend_plan::BACKEND_COMPAT + })) + .unwrap(); + let response = handle_workload_cost_manifests(Json(request)) + .await + .into_response(); + assert_eq!(response.status(), StatusCode::OK); + let manifests = body_json(response).await; + assert_eq!(manifests.as_array().unwrap().len(), 1); + } + async fn body_json(resp: axum::response::Response) -> serde_json::Value { let bytes = resp.into_body().collect().await.unwrap().to_bytes(); serde_json::from_slice(&bytes).unwrap() diff --git a/control_plane/src/physical/colored_dag/emitter.rs b/control_plane/src/physical/colored_dag/emitter.rs index e7966843..787d43b5 100644 --- a/control_plane/src/physical/colored_dag/emitter.rs +++ b/control_plane/src/physical/colored_dag/emitter.rs @@ -702,6 +702,9 @@ pub struct BackendAggregation { /// `parameters["item_label"]` so the data-plane ingest records it on the /// CMS sid and can answer per-item `estimate(key)` (FrequencyEstimate). pub item_label: Option, + /// Runtime accumulator mode derived from SummaryAgg.input.weight, never + /// from the TopK readout. None retains the legacy value-update default. + pub heap_update_mode: Option<&'static str>, /// Phase ε.1 — what shape the backend ingests for this /// aggregation. Mode 1 (sketch at edge) / sketch_envelope is the /// default (the wire payload is a sketch state already). Mode 2 @@ -899,6 +902,7 @@ impl Emitter for ThreeStageEmitter { }); backend_aggregations.push(BackendAggregation { item_label: None, + heap_update_mode: None, aggregation_id, metric_name: edge.source_metric.clone().unwrap_or_default(), family: SummaryFamilyType::Sketch( @@ -983,6 +987,7 @@ impl Emitter for ThreeStageEmitter { next_agg_index += 1; backend_aggregations.push(BackendAggregation { item_label: None, + heap_update_mode: None, aggregation_id: aid, metric_name: edge.source_metric.clone().unwrap_or_default(), family: SummaryFamilyType::Sketch( diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index a764fc4e..e5057ca5 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -42,7 +42,7 @@ use crate::query_plan::{ use crate::types_v2::AccuracyTarget; use planner_types::pre_asap::Source; -pub const PLANNER_REVISION: &str = "378a7547ede629a64e84c9f7c810226ce196cce9"; +pub const PLANNER_REVISION: &str = "c3410d14865497758d212e4265ad25c782187de1"; #[derive(Debug, Clone)] pub struct PlanningQuery { @@ -169,6 +169,8 @@ pub enum PhysicalDeploymentTarget { #[serde(deny_unknown_fields)] pub struct BackendLocalPlanningSnapshot { pub snapshot_version: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workload_cost_evidence: Option, pub query_workload: QueryWorkload, pub data_workload: DataWorkload, pub implementation: BackendLocalImplementation, @@ -185,6 +187,10 @@ pub struct BackendLocalImplementation { pub window_implementation_id: String, pub state_layout: String, pub implementation_cost: ImplementationCostEvidence, + /// Certificates keyed by exact registered PromQL; converted to root IDs + /// before workload selection so one query cannot borrow another's evidence. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub topk_evidence: HashMap, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -618,6 +624,20 @@ pub struct PhysicalPlan { pub transmission_plan: TransmissionPlan, pub backend_plan: BackendPlan, pub query_plan: QueryPlan, + /// Lifecycle component only, not a complete physical-plan comparison. + pub lifecycle_estimates: Vec, + pub cost_comparison: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct MaterializationLifecycleEstimate { + pub materialization: asap_types::PolicyFingerprint, + pub consumer_query_ids: Vec, + pub window_implementation_id: String, + pub horizon_seconds: f64, + pub expected_reads: f64, + pub expected_updates: f64, + pub lifecycle_cost: f64, } #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] @@ -1501,7 +1521,28 @@ impl BackendLocalPlanningSnapshot { /// one backend-local PhysicalPlan. No CollectorPlan is produced and no /// precompiled serving artifact is accepted at this boundary. pub fn compile(self) -> Result { - if self.snapshot_version != 1 { + let evidence = self.workload_cost_evidence.clone(); + if self.snapshot_version == 2 && evidence.is_none() { + return Err(CompileError::Snapshot( + "version 2 requires complete workload cost evidence".into(), + )); + } + let (request, environment) = self.planning_request()?; + match evidence { + Some(evidence) => super::workload_cost::select( + super::workload_cost::with_exact_alternative(request)?, + environment, + &evidence, + ), + None => PhysicalCompiler.compile(request, environment), + } + } + + /// Build Planner-authorized candidates for evidence collection without publishing. + pub fn planning_request( + self, + ) -> Result<(PlanningRequest, DeploymentEnvironment), CompileError> { + if self.snapshot_version != 1 && self.snapshot_version != 2 { return Err(CompileError::Snapshot(format!( "unsupported workload snapshot version {}", self.snapshot_version @@ -1545,6 +1586,7 @@ impl BackendLocalPlanningSnapshot { } let mut queries = Vec::with_capacity(entries.len()); let mut canonical_roots = Vec::with_capacity(entries.len()); + let mut topk_evidence_by_id = HashMap::new(); for (index, entry) in entries.into_iter().enumerate() { let evaluation_interval_ms = match entry.recurrence { QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(interval)) => interval.0, @@ -1597,8 +1639,12 @@ impl BackendLocalPlanningSnapshot { cost.workload_fingerprint = canonical_promql(&query_string).map_err(CompileError::QueryPlan)?; cost.horizon_seconds = self.implementation.horizon_seconds; + let query_id = format!("compat-query-{index}"); + if let Some(evidence) = self.implementation.topk_evidence.get(&query_string) { + topk_evidence_by_id.insert(query_id.clone(), evidence.clone()); + } queries.push(PlanningQuery { - query_id: format!("compat-query-{index}"), + query_id, query_string, post_asap, source: Source::TimeSeries { @@ -1619,33 +1665,15 @@ impl BackendLocalPlanningSnapshot { runtime_policy: RuntimeRulePolicy::default(), }); } - // A cohort shares the same end-to-end requirement, not an inferred - // weakest common accuracy. Different targets are searched separately. - let mut cohorts: Vec<(AccuracyTarget, Vec<(usize, Rc)>)> = Vec::new(); - for (index, root) in canonical_roots.into_iter().enumerate() { - let accuracy = &queries[index].accuracy; - if let Some((_, roots)) = cohorts.iter_mut().find(|(target, _)| target == accuracy) { - roots.push((index, root)); - } else { - cohorts.push((accuracy.clone(), vec![(index, root)])); - } - } - for (accuracy, roots) in cohorts { - let model = ControlPlaneCostModel::new(accuracy.clone()); - let selected = crate::planner_selection::select_workload(roots, accuracy, &model) - .map_err(|error| CompileError::Snapshot(error.to_string()))?; - for (index, node) in selected { - queries[index].post_asap = node; - } - } - PhysicalCompiler.compile( + select_workload_roots(&mut queries, canonical_roots, &topk_evidence_by_id)?; + Ok(( PlanningRequest { queries, - evidence: HashMap::new(), + evidence: topk_evidence_by_id, planner_revision: PLANNER_REVISION.into(), }, self.environment, - ) + )) } } @@ -1690,12 +1718,25 @@ impl PhysicalCompiler { // the planner DAG node identity across the workload; serving never scans // BackendPlan candidates to rediscover this decision. let mut node_bindings = HashMap::::new(); + let consumers = materialization_consumers(&request.queries, environment.target)?; + let mut lifecycle_estimates = BTreeMap::new(); for query in &request.queries { let evidence = request.evidence.get(&query.query_id); if let Some(e) = evidence { validate_evidence(&query.query_id, e, &environment)?; } + let node = query.post_asap.clone(); + let selected = + collect_selected_materializations(&node).map_err(|reason| CompileError::Query { + query_id: query.query_id.clone(), + reason, + })?; + // An exact native fallback has no maintained state and must not + // depend on evidence for unused window/state implementations. + if selected.is_empty() { + continue; + } validate_lifecycle_input(&query.query_id, &query.lifecycle)?; let lifecycle_costs = SummaryMaintenanceLifecycleCostInputs { build_cost: Some(Cost(query.lifecycle.costs.build)), @@ -1717,12 +1758,6 @@ impl PhysicalCompiler { }, ) .with_window_framework_costs(window_costs); - let node = query.post_asap.clone(); - let selected = - collect_selected_materializations(&node).map_err(|reason| CompileError::Query { - query_id: query.query_id.clone(), - reason, - })?; if environment.target == PhysicalDeploymentTarget::DistributedCollectors && selected.iter().any(|state| { matches!( @@ -1740,19 +1775,6 @@ impl PhysicalCompiler { .into(), }); } - if selected.is_empty() { - continue; - } - let planner_selection = select_lifecycle(query, &node, &model, &environment)?; - let window_implementation = query - .window_implementations - .iter() - .filter(|candidate| candidate.framework == planner_selection.window_framework) - .min_by(|left, right| left.cost.weighted_cost.total_cmp(&right.cost.weighted_cost)) - .ok_or_else(|| CompileError::Lifecycle { - query_id: query.query_id.clone(), - reason: "Planner selected a window framework without a retained concrete implementation".into(), - })?; match &query.source { Source::TimeSeries { .. } => {} Source::Table { .. } => { @@ -1773,26 +1795,49 @@ impl PhysicalCompiler { SummaryFamilyType::ExactAggregate(kind, _) => { format!("{kind:?}").to_ascii_lowercase() } - _ => selected.algorithm, - }; - let aggregation = BackendAggregation { - aggregation_id: aggregation_id.clone(), - metric_name: metric.clone(), - family: physical_family, - window_secs: query.window_secs, - spatial_filter: String::new(), - grouping: query.group_by.clone(), - item_label: None, - aggregation_input: match environment.target { - PhysicalDeploymentTarget::DistributedCollectors => { - AggregationInput::SketchEnvelope - } - PhysicalDeploymentTarget::BackendLocalRemoteWrite => AggregationInput::Raw, - }, + _ => selected.algorithm.clone(), }; + let aggregation = physical_aggregation( + query, + &selected, + aggregation_id.clone(), + environment.target, + ); let precompute_materialization = backend_plan::aggregation_config_for_materialization(&aggregation)?; let materialization = precompute_materialization.policy_fingerprint(); + let state_consumers = consumers[&materialization] + .iter() + .map(|index| &request.queries[*index]) + .collect::>(); + let planner_selection = select_lifecycle( + query, + &selected.node, + &model, + &environment, + &state_consumers, + )?; + let window_implementation = query.window_implementations.iter() + .filter(|candidate| candidate.framework == planner_selection.window_framework) + .min_by(|left, right| left.cost.weighted_cost.total_cmp(&right.cost.weighted_cost)) + .ok_or_else(|| CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "Planner selected a window framework without a retained concrete implementation".into(), + })?; + lifecycle_estimates + .entry(materialization) + .or_insert_with(|| MaterializationLifecycleEstimate { + materialization, + consumer_query_ids: state_consumers + .iter() + .map(|query| query.query_id.clone()) + .collect(), + window_implementation_id: window_implementation.implementation_id.clone(), + horizon_seconds: query.lifecycle.horizon_seconds, + expected_reads: planner_selection.expected_reads, + expected_updates: planner_selection.expected_updates, + lifecycle_cost: planner_selection.lifecycle_cost, + }); let binding_key = selected.node_identity; if let Some(existing) = node_bindings.insert(binding_key, materialization) { if existing != materialization { @@ -2029,6 +2074,8 @@ impl PhysicalCompiler { transmission_plan, backend_plan, query_plan, + lifecycle_estimates: lifecycle_estimates.into_values().collect(), + cost_comparison: None, }) } } @@ -2072,6 +2119,52 @@ fn summary_agg_metric(node: &SummaryNode) -> Option { .flatten() } +/// Shared selection boundary for canonical startup and compile-and-publish. +/// Certificate-bearing roots stay isolated: equal certificate values do not +/// establish that the certificate's source scope covers another query. +pub fn select_workload_roots( + queries: &mut [PlanningQuery], + roots: Vec>, + evidence: &HashMap, +) -> Result<(), CompileError> { + if roots.len() != queries.len() { + return Err(CompileError::Snapshot( + "canonical root/query mapping is incomplete".into(), + )); + } + let mut cohorts: Vec<(AccuracyTarget, Option, Vec<(usize, Rc)>)> = + Vec::new(); + for (index, root) in roots.into_iter().enumerate() { + let accuracy = &queries[index].accuracy; + let certificate_scope = evidence + .contains_key(&queries[index].query_id) + .then(|| queries[index].query_id.clone()); + if let Some((_, _, roots)) = cohorts + .iter_mut() + .find(|(target, scope, _)| target == accuracy && scope == &certificate_scope) + { + roots.push((index, root)); + } else { + cohorts.push((accuracy.clone(), certificate_scope, vec![(index, root)])); + } + } + for (accuracy, scope, roots) in cohorts { + let model = ControlPlaneCostModel::new(accuracy.clone()); + let certificate = scope.as_ref().and_then(|id| evidence.get(id)); + let selected = crate::planner_selection::select_workload_with_evidence( + roots, + accuracy, + &model, + &QueryEvidence(certificate), + ) + .map_err(|error| CompileError::Snapshot(error.to_string()))?; + for (index, node) in selected { + queries[index].post_asap = node; + } + } + Ok(()) +} + /// Planner-adapter selection step used before physical compilation. Keeping /// this separate makes the ownership boundary explicit: callers supply the /// selected post-ASAP DAG to [`PhysicalCompiler::compile`]. @@ -2246,6 +2339,9 @@ fn validate_window_implementations( struct PlannerPhysicalSelection { lifecycle: CollectorLifecycle, window_framework: SummaryWindowFramework, + expected_reads: f64, + expected_updates: f64, + lifecycle_cost: f64, } fn select_lifecycle( @@ -2253,28 +2349,49 @@ fn select_lifecycle( node: &SummaryNode, model: &ControlPlaneCostModel, environment: &DeploymentEnvironment, + consumers: &[&PlanningQuery], ) -> Result { + // Current lifecycle evidence is per producer with one unit read cost. + // Conflicting source/rate/horizon/cost snapshots cannot be averaged into + // invented evidence. Only recurrence may differ between consumers. + let mut common = query.lifecycle.clone(); + common.evaluation_interval_ms = 0; + for consumer in consumers { + let mut input = consumer.lifecycle.clone(); + input.evaluation_interval_ms = 0; + if input != common { + return Err(CompileError::Lifecycle { + query_id: consumer.query_id.clone(), + reason: "shared producer consumers have conflicting lifecycle evidence".into(), + }); + } + } let workload = QueryWorkload { language: QueryLanguage::PromQL, query_batch: None, - repeating_queries: Some(vec![RepeatingEntry { - query: Query(query.query_id.clone()), - demand: RepeatedDemand::FixedInterval(RepetitionInterval( - query.lifecycle.evaluation_interval_ms, - )), - requirements: QueryRequirements { - accuracy: AccuracyRequirement::Explicit(query.accuracy.clone()), - ..QueryRequirements::default() - }, - predictability: Predictability::Predictable { known_at: None }, - time_selection: TimeSelection { - // Collector windows are retired as whole states. They do not - // claim deletion support for moving-window retractions. - scope: QueryTimeScope::Unknown, - lookback: Some(DurationMs(query.window_secs.saturating_mul(1_000))), - as_of: None, - }, - }]), + repeating_queries: Some( + consumers + .iter() + .map(|query| RepeatingEntry { + query: Query(query.query_id.clone()), + demand: RepeatedDemand::FixedInterval(RepetitionInterval( + query.lifecycle.evaluation_interval_ms, + )), + requirements: QueryRequirements { + accuracy: AccuracyRequirement::Explicit(query.accuracy.clone()), + ..QueryRequirements::default() + }, + predictability: Predictability::Predictable { known_at: None }, + time_selection: TimeSelection { + // Collector windows are retired as whole states. They do not + // claim deletion support for moving-window retractions. + scope: QueryTimeScope::Unknown, + lookback: Some(DurationMs(query.window_secs.saturating_mul(1_000))), + as_of: None, + }, + }) + .collect(), + ), data_workload: Some(DataWorkload { arrival: DataArrival::ContinuouslyIngesting, ingestion_rate: Evidence { @@ -2288,7 +2405,7 @@ fn select_lifecycle( }; let plan = plan_summary_maintenance_lifecycles( Rc::new(node.clone()), - WorkloadDemand::new(&workload, &[0]), + WorkloadDemand::new(&workload, &(0..consumers.len()).collect::>()), environment.observed_at_unix_ms, Some(Horizon(query.lifecycle.horizon_seconds)), SummaryMaintenanceLifecycleCapabilities { @@ -2320,6 +2437,32 @@ fn select_lifecycle( reason: "latest ASAPPlanner selected no window framework from the supplied physical evidence".into(), })?; Ok(PlannerPhysicalSelection { + expected_reads: plan.expected_reads.ok_or_else(|| CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "missing joint read demand".into(), + })?, + expected_updates: plan + .update_rate + .ok_or_else(|| CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "missing source update demand".into(), + })? + .0 + * query.lifecycle.horizon_seconds, + lifecycle_cost: plan.deployments[0] + .alternatives + .iter() + .find(|alternative| { + alternative.rejection.is_none() + && alternative.summary_maintenance_lifecycle + == guarantee.summary_maintenance_lifecycle + }) + .and_then(|alternative| alternative.total_cost) + .ok_or_else(|| CompileError::Lifecycle { + query_id: query.query_id.clone(), + reason: "missing selected lifecycle cost".into(), + })? + .0, lifecycle: CollectorLifecycle { kind: match guarantee.summary_maintenance_lifecycle { SummaryMaintenanceLifecycle::Ephemeral => "ephemeral", @@ -2351,6 +2494,7 @@ fn select_lifecycle( } struct SelectedMaterialization { + node: Rc, node_identity: usize, metric: String, family: SummaryFamilyType, @@ -2359,6 +2503,59 @@ struct SelectedMaterialization { parameters: Value, } +fn physical_aggregation( + query: &PlanningQuery, + selected: &SelectedMaterialization, + aggregation_id: String, + target: PhysicalDeploymentTarget, +) -> BackendAggregation { + BackendAggregation { + aggregation_id, + metric_name: selected.metric.clone(), + family: physical_materialization_family(&selected.family), + window_secs: query.window_secs, + spatial_filter: String::new(), + grouping: query.group_by.clone(), + item_label: None, + heap_update_mode: selected.parameters.get("weight_mode").and_then(|mode| { + match mode.as_str() { + Some("count") => Some("count"), + Some("value") => Some("value"), + _ => None, + } + }), + aggregation_input: match target { + PhysicalDeploymentTarget::DistributedCollectors => AggregationInput::SketchEnvelope, + PhysicalDeploymentTarget::BackendLocalRemoteWrite => AggregationInput::Raw, + }, + } +} + +fn materialization_consumers( + queries: &[PlanningQuery], + target: PhysicalDeploymentTarget, +) -> Result>, CompileError> { + let mut consumers = BTreeMap::<_, BTreeSet<_>>::new(); + for (index, query) in queries.iter().enumerate() { + let states = collect_selected_materializations(&query.post_asap).map_err(|reason| { + CompileError::Query { + query_id: query.query_id.clone(), + reason, + } + })?; + for state in states { + let config = backend_plan::aggregation_config_for_materialization( + &physical_aggregation(query, &state, query.query_id.clone(), target), + )?; + consumers + .entry(config.policy_fingerprint()) + .or_default() + .insert(index); + } + } + Ok(consumers) +} + /// Collect every executable materialization leaf in the selected post-ASAP /// graph. Readout context flows through merge nodes, so a graph such as /// `Estimate(Merge(Agg(a), Agg(b)))` creates two physical bindings while the @@ -2408,13 +2605,27 @@ fn collect_selected_materializations( } SummaryExpr::SummaryAgg { family: SummaryFamilyType::Sketch(kind, _), + input, .. } => { if let Some(readout) = readout { + let mut parameters = sketch_params_json(kind.params()); + if matches!(readout, SketchQuery::TopK { .. }) { + use planner_types::post_asap::SummaryInputExpr; + let mode = match &input.weight { + SummaryInputExpr::Constant(value) if *value == 1.0 => "count", + SummaryInputExpr::Column( + planner_types::pre_asap::ColumnRef::SampleValue, + ) => "value", + _ => return Err("unsupported TopK SummaryUpdate weight".into()), + }; + parameters["weight_mode"] = mode.into(); + } let metric = summary_agg_metric(node).ok_or_else(|| { "SummaryAgg has no unique time-series source in post-ASAP IR".to_string() })?; selected.push(SelectedMaterialization { + node: Rc::clone(node), node_identity: Rc::as_ptr(node) as usize, metric, family: SummaryFamilyType::Sketch( @@ -2423,7 +2634,7 @@ fn collect_selected_materializations( ), readout: Some(readout.clone()), algorithm: format!("{:?}", kind.algorithm()).to_ascii_lowercase(), - parameters: sketch_params_json(kind.params()), + parameters, }); } } @@ -2435,6 +2646,7 @@ fn collect_selected_materializations( "SummaryAgg has no unique time-series source in post-ASAP IR".to_string() })?; selected.push(SelectedMaterialization { + node: Rc::clone(node), node_identity: Rc::as_ptr(node) as usize, metric, family: SummaryFamilyType::ExactAggregate(kind.clone(), params.clone()), @@ -2526,6 +2738,32 @@ fn stable_workload_plan_id( mod tests { use super::*; + // Count and value rankings must configure different state update contracts. + #[test] + fn temporal_topk_binds_planner_update_weight() { + for (query, mode) in [ + ("topk(1, sum_over_time(m[1m]))", "value"), + ("topk(1, count_over_time(m[1m]))", "count"), + ] { + let evidence = TopKMembershipEvidence { + selected_lower_bound: 101.0, + excluded_upper_bound: 100.0, + interval_failure_probability: 0.001, + observed_at_unix_ms: 9500, + source: "unit-fixture".into(), + }; + let request = request_with_evidence("topk", query, Some(evidence)).unwrap(); + let plan = PhysicalCompiler + .compile(request, environment(10000)) + .unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 1, "{query}"); + assert_eq!( + plan.precompute_plan.materializations[0].parameters["weight_mode"], mode, + "{query}" + ); + } + } + fn environment(now: u64) -> DeploymentEnvironment { DeploymentEnvironment { target: PhysicalDeploymentTarget::DistributedCollectors, @@ -2611,6 +2849,100 @@ mod tests { request_with_evidence(query_id, promql, None).expect("post-ASAP selection") } + // Both production adapters preserve canonical root identity and select the + // whole evidence-free cohort, rather than independently binding roots. + #[test] + fn shared_selection_adapter_preserves_query_mapping() { + let mut workload = request("q90", "quantile_over_time(0.9, m[1m])"); + workload + .queries + .extend(request("q99", "quantile_over_time(0.99, m[1m])").queries); + let roots = workload + .queries + .iter() + .map(|query| { + Rc::new( + crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .unwrap(), + ) + }) + .collect(); + select_workload_roots(&mut workload.queries, roots, &workload.evidence).unwrap(); + let bundle = PhysicalCompiler + .compile(workload, environment(10000)) + .unwrap(); + assert_eq!(bundle.query_plan.entries.len(), 2); + assert_eq!(bundle.collector_plans[0].materializations.len(), 1); + assert_eq!( + bundle + .query_plan + .entries + .values() + .map(|entry| entry.query_id.as_str()) + .collect::>(), + BTreeSet::from(["q90", "q99"]) + ); + } + + // A broken input mapping must be rejected, never silently drop a root. + #[test] + fn shared_selection_rejects_incomplete_root_mapping() { + let mut workload = request("q", "quantile_over_time(0.9, m[1m])"); + assert!(select_workload_roots(&mut workload.queries, vec![], &workload.evidence).is_err()); + } + + // Adding another readout adds recurring reads, not another update stream. + #[test] + fn joint_lifecycle_charges_shared_updates_once() { + let baseline = PhysicalCompiler + .compile( + request("q90", "quantile_over_time(0.9, m[1m])"), + environment(10000), + ) + .unwrap(); + let mut workload = request("q90", "quantile_over_time(0.9, m[1m])"); + let mut second = request("q99", "quantile_over_time(0.99, m[1m])") + .queries + .remove(0); + second.lifecycle.evaluation_interval_ms = 20000; + workload.queries.push(second); + let shared = PhysicalCompiler + .compile(workload, environment(10000)) + .unwrap(); + assert_eq!(shared.lifecycle_estimates.len(), 1); + let estimate = &shared.lifecycle_estimates[0]; + assert_eq!(estimate.consumer_query_ids, vec!["q90", "q99"]); + assert_eq!(estimate.expected_reads, 45.0); + assert_eq!(estimate.expected_updates, 30000.0); + assert!((estimate.lifecycle_cost - 45.8).abs() < 1e-9); + assert_eq!( + estimate.expected_updates, + baseline.lifecycle_estimates[0].expected_updates + ); + assert!( + (estimate.lifecycle_cost - baseline.lifecycle_estimates[0].lifecycle_cost - 1.5).abs() + < 1e-9 + ); + } + + // Unknown joint provenance cannot be replaced by whichever root came first. + #[test] + fn shared_lifecycle_rejects_conflicting_source_evidence() { + let mut workload = request("q90", "quantile_over_time(0.9, m[1m])"); + let mut second = request("q99", "quantile_over_time(0.99, m[1m])") + .queries + .remove(0); + second.lifecycle.ingestion_rate_per_second = 200.0; + workload.queries.push(second); + assert!(matches!( + PhysicalCompiler.compile(workload, environment(10000)), + Err(CompileError::Lifecycle { .. }) + )); + } + #[test] fn shared_materialization_is_emitted_once_for_every_runtime() { for target in [ @@ -3005,6 +3337,7 @@ mod tests { .remove(0); let snapshot = BackendLocalPlanningSnapshot { snapshot_version: 1, + workload_cost_evidence: None, query_workload, data_workload, implementation: BackendLocalImplementation { @@ -3015,6 +3348,7 @@ mod tests { window_implementation_id: "backend-tumbling-v1".into(), state_layout: "anchored-pane-v1".into(), implementation_cost: template.window_implementations[0].cost.clone(), + topk_evidence: HashMap::new(), }, environment, }; @@ -3127,13 +3461,15 @@ mod tests { assert!(plan.collector_plans.is_empty()); assert!(plan.transmission_plan.rules.is_empty()); - assert_eq!(plan.query_plan.entries.len(), 4); - assert_eq!(plan.precompute_plan.materializations.len(), 3); + assert_eq!(plan.query_plan.entries.len(), 6); + assert_eq!(plan.precompute_plan.materializations.len(), 5); for query in [ "rate(asap_demo_counter_total[5s])", "increase(asap_demo_counter_total[5s])", "sum_over_time(asap_demo_gauge[5s])", "quantile_over_time(0.5, asap_demo_latency_ms[5s])", + "topk(1, sum_over_time(asap_demo_gauge[5s]))", + "topk(1, count_over_time(asap_demo_gauge[5s]))", ] { assert!(plan.query_plan.lookup(query).is_ok(), "missing {query}"); } diff --git a/control_plane/src/physical/mod.rs b/control_plane/src/physical/mod.rs index a544efcf..e5d3c0c1 100644 --- a/control_plane/src/physical/mod.rs +++ b/control_plane/src/physical/mod.rs @@ -34,6 +34,7 @@ pub mod sketch_catalog; pub mod stage_split; pub mod topology; pub mod window_fusion; +pub mod workload_cost; pub mod workload_planner; // Convenience re-exports — preserve the surface that consumers of the diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs new file mode 100644 index 00000000..d2baa296 --- /dev/null +++ b/control_plane/src/physical/workload_cost.rs @@ -0,0 +1,715 @@ +//! Complete, provider-priced comparisons of already-bound workload alternatives. +//! +//! This is an evidence manifest over the existing physical projection, not a +//! second semantic DAG. Planner supplies legal alternatives; deployment quotes +//! price every reachable operation, and the backend commits one complete plan. + +use std::collections::{BTreeMap, BTreeSet}; + +use asap_aware_mapping::cost_model::Cost; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use super::compiler::{ + CompileError, DeploymentEnvironment, PhysicalCompiler, PhysicalPlan, PlanningQuery, + PlanningRequest, +}; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct CostDemand { + /// Exact implementation/configuration being priced, not merely a family. + pub implementation: Value, + /// `horizon` includes all work in the manifest's source/time scope; + /// `query_evaluation` is one execution of this bound query operator. + pub unit: String, + pub multiplicity: f64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct WorkloadCostManifest { + pub plan_id: u64, + pub plan_version: u64, + pub planner_revision: String, + pub capability_snapshot_id: String, + pub backend_compat: String, + pub horizon_seconds: f64, + /// Canonical roots, requirements and demand must match across alternatives. + pub workload: BTreeMap, + pub components: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct WorkloadQuote { + pub manifest: WorkloadCostManifest, + /// Provider attests feasibility for this capability/data generation, + /// including an accessible exact backend when the plan has fallback nodes. + pub executable: bool, + /// Calibrated costs in ONE model's units. All keys are required, including + /// explicit zero costs. Horizon quotes include source cardinality, all + /// maintained groups, retention/spill and the stated partition's work. + pub unit_costs: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct WorkloadCostEvidence { + pub data_snapshot_id: String, + pub model_version: String, + pub observed_at_unix_ms: u64, + pub valid_for_ms: u64, + pub quotes: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct AlternativeCost { + pub plan_id: Option, + pub total_cost: Option, + pub unavailable_reason: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct WorkloadCostComparison { + pub data_snapshot_id: String, + pub model_version: String, + pub selected_plan_id: u64, + pub selected_manifest: WorkloadCostManifest, + pub component_costs: BTreeMap, + pub alternatives: Vec, +} + +fn invalid(reason: impl Into) -> CompileError { + CompileError::Snapshot(format!("complete workload cost: {}", reason.into())) +} + +pub fn manifest( + plan: &PhysicalPlan, + queries: &[PlanningQuery], +) -> Result { + let horizon = queries + .first() + .ok_or_else(|| invalid("empty workload"))? + .lifecycle + .horizon_seconds; + if !horizon.is_finite() || horizon <= 0.0 { + return Err(invalid("invalid horizon")); + } + let mut workload = BTreeMap::new(); + let mut reads = BTreeMap::new(); + for query in queries { + if query.lifecycle.horizon_seconds != horizon || query.lifecycle.evaluation_interval_ms == 0 + { + return Err(invalid("mixed horizons or unknown recurrence")); + } + let canonical = crate::query_plan::canonical_promql(&query.query_string)?; + if workload + .insert( + query.query_id.clone(), + json!({ + "query": canonical, "accuracy": query.accuracy, + "evaluation_interval_ms": query.lifecycle.evaluation_interval_ms, + "source": query.source, + }), + ) + .is_some() + { + return Err(invalid("duplicate query ID")); + } + reads.insert( + query.query_id.clone(), + horizon * 1000.0 / f64::from(query.lifecycle.evaluation_interval_ms), + ); + } + let mut components = BTreeMap::new(); + let mut add = |id: String, implementation: Value, unit: &str, multiplicity: f64| { + components.insert( + id, + CostDemand { + implementation, + unit: unit.into(), + multiplicity, + }, + ); + }; + // Backend merge/update, storage, and edge maintenance are separate work. + // The raw input is read once per source partition, not once per consumer. + for schema in &plan.precompute_plan.schemas { + let mut locations = vec!["backend".to_string()]; + locations.extend( + plan.collector_plans + .iter() + .filter(|collector| { + collector + .materializations + .iter() + .any(|m| m.materialization == schema.materialization) + }) + .map(|collector| format!("collector:{}", collector.collector_id)), + ); + for location in locations { + let source = json!({"source": schema.source, "location": location, "ingest": plan.precompute_plan.ingest}); + add(format!("source:{}", source), source, "horizon", 1.0); + let physical = plan + .precompute_plan + .materializations + .iter() + .find(|m| m.policy_fingerprint() == schema.materialization) + .ok_or_else(|| invalid("state has no physical implementation"))?; + let identity = json!({"schema": schema, "location": location, "physical": physical, + "window_implementation": plan.lifecycle_estimates.iter().find(|e| e.materialization == schema.materialization).map(|e| &e.window_implementation_id)}); + for operation in ["build", "update", "residency", "retire"] { + add( + format!("state:{location}:{}:{operation}", schema.materialization.0), + json!({"operation": operation, "binding": identity}), + "horizon", + 1.0, + ); + } + } + } + for rule in &plan.transmission_plan.rules { + add( + format!("transport:{}:{}", rule.producer_id, rule.materialization.0), + json!(rule), + "horizon", + 1.0, + ); + } + for entry in plan.query_plan.entries.values() { + let evaluations = *reads + .get(&entry.query_id) + .ok_or_else(|| invalid("unmapped query root"))?; + if entry + .nodes + .values() + .any(|node| matches!(node, crate::query_plan::QueryPlanNode::ExactFallback { .. })) + { + let query = queries + .iter() + .find(|query| query.query_id == entry.query_id) + .ok_or_else(|| invalid("unmapped exact source"))?; + // Charge the exact service's input upkeep/storage separately from + // per-evaluation native execution. Zero is valid only if explicitly + // quoted as already-provisioned/non-incremental for this decision. + let parsed = crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .map_err(|error| invalid(error.to_string()))?; + for metric in exact_source_metrics(&parsed)? { + let source = json!({"source": planner_types::pre_asap::Source::TimeSeries { metric }, "location": "exact_backend"}); + add(format!("source:{}", source), source, "horizon", 1.0); + } + } + // Reachability comes from QueryPlan, including materialization reads, + // arithmetic, reduction and a complete engine-native exact fallback. + for node_id in entry.topological_order()? { + add( + format!("query:{}:{}", entry.query_id, node_id.0), + json!({"node": entry.nodes[&node_id], "query": entry.canonical_promql, "instant": entry.instant}), + "query_evaluation", + evaluations, + ); + } + add( + format!("result:{}", entry.query_id), + json!({"query_id": entry.query_id, "root": entry.root}), + "query_evaluation", + evaluations, + ); + } + Ok(WorkloadCostManifest { + plan_id: plan.envelope.plan_id, + plan_version: plan.envelope.plan_version, + planner_revision: plan.envelope.planner_revision.clone(), + capability_snapshot_id: plan.envelope.capability_snapshot_id.clone(), + backend_compat: plan.envelope.backend_compat.clone(), + horizon_seconds: horizon, + workload, + components, + }) +} + +/// Walk the canonical relational tree, preserving every input to binary and +/// fan-in operators. Unsupported source discovery must not produce a partial quote. +fn exact_source_metrics( + expr: &planner_types::pre_asap::QueryExpr, +) -> Result, CompileError> { + use planner_types::pre_asap::{QueryExpr, Source}; + fn visit(expr: &QueryExpr, metrics: &mut BTreeSet) -> Result<(), CompileError> { + match expr { + QueryExpr::Scan { + source: Source::TimeSeries { metric }, + .. + } if !metric.is_empty() => { + metrics.insert(metric.clone()); + } + QueryExpr::PromqlScalarBridge(child) + | QueryExpr::PromqlVectorFromScalar(child) + | QueryExpr::PromqlScalarFromVector(child) + | QueryExpr::PromqlRelabel { child, .. } + | QueryExpr::PromqlSeriesSample { child, .. } + | QueryExpr::Filter { child, .. } + | QueryExpr::Project { child, .. } + | QueryExpr::Aggregate { child, .. } + | QueryExpr::Dedup { child, .. } + | QueryExpr::Sort { child, .. } + | QueryExpr::Limit { child, .. } + | QueryExpr::PromqlSubquery { child, .. } + | QueryExpr::TimeRange { child, .. } + | QueryExpr::TimeShift { child, .. } => visit(child, metrics)?, + QueryExpr::BinaryOp { + lhs: left, + rhs: right, + .. + } + | QueryExpr::Join { left, right, .. } + | QueryExpr::SetOp { left, right, .. } => { + visit(left, metrics)?; + visit(right, metrics)?; + } + QueryExpr::Concat { children, .. } => { + for child in children { + visit(child, metrics)?; + } + } + QueryExpr::Literal(_) | QueryExpr::EvalTimestamp => {} + // In particular, info() has an implicit metadata source that is + // not a Scan child, and unnamed selectors require source discovery. + _ => { + return Err(invalid( + "exact-source pricing cannot enumerate this query's sources", + )) + } + } + Ok(()) + } + let mut metrics = BTreeSet::new(); + visit(expr, &mut metrics)?; + Ok(metrics) +} + +impl WorkloadCostEvidence { + fn validate(&self, env: &DeploymentEnvironment) -> Result<(), CompileError> { + if self.data_snapshot_id.trim().is_empty() + || self.model_version.trim().is_empty() + || self.valid_for_ms == 0 + || self.observed_at_unix_ms > env.observed_at_unix_ms + || env.observed_at_unix_ms - self.observed_at_unix_ms + > self.valid_for_ms.min(env.max_evidence_age_ms) + { + return Err(invalid("missing, future or stale evidence generation")); + } + Ok(()) + } + + fn price( + &self, + manifest: &WorkloadCostManifest, + ) -> Result<(Cost, BTreeMap), String> { + let quotes = self + .quotes + .iter() + .filter(|quote| "e.manifest == manifest) + .collect::>(); + if quotes.len() != 1 { + return Err("missing or ambiguous quote for exact manifest".into()); + } + let quote = quotes[0]; + if !quote.executable { + return Err("provider reports unavailable implementation".into()); + } + if !quote.unit_costs.keys().eq(manifest.components.keys()) { + return Err("incomplete or extraneous component evidence".into()); + } + let mut total = 0.0; + let mut components = BTreeMap::new(); + for (id, demand) in &manifest.components { + let unit = quote.unit_costs[id]; + let cost = unit * demand.multiplicity; + if !unit.is_finite() || unit < 0.0 || !cost.is_finite() || cost < 0.0 { + return Err(format!("invalid cost for {id}")); + } + total += cost; + components.insert(id.clone(), cost); + } + if !total.is_finite() { + return Err("cost overflow".into()); + } + Ok((Cost(total), components)) + } +} + +/// Compare complete Planner-authorized forests after binding. Infeasible or +/// uncosted alternatives are retained as unavailable, never assigned zero. +pub fn select( + candidates: Vec, + env: DeploymentEnvironment, + evidence: &WorkloadCostEvidence, +) -> Result { + evidence.validate(&env)?; + if candidates.is_empty() || candidates.len() > 64 { + return Err(invalid( + "candidate inventory must contain 1..=64 alternatives", + )); + } + let mut comparison_workload = None; + let mut alternatives = Vec::new(); + let mut best: Option<( + Cost, + PhysicalPlan, + WorkloadCostManifest, + BTreeMap, + )> = None; + for candidate in candidates { + let queries = candidate.queries.clone(); + let plan = match PhysicalCompiler.compile(candidate, env.clone()) { + Ok(plan) => plan, + Err(error) => { + alternatives.push(AlternativeCost { + plan_id: None, + total_cost: None, + unavailable_reason: Some(error.to_string()), + }); + continue; + } + }; + let manifest = match manifest(&plan, &queries) { + Ok(manifest) => manifest, + Err(error) => { + alternatives.push(AlternativeCost { + plan_id: Some(plan.envelope.plan_id), + total_cost: None, + unavailable_reason: Some(error.to_string()), + }); + continue; + } + }; + let scope = (manifest.workload.clone(), manifest.horizon_seconds); + if comparison_workload + .as_ref() + .is_some_and(|previous| previous != &scope) + { + return Err(invalid( + "alternatives describe different workloads/horizons", + )); + } + comparison_workload = Some(scope); + match evidence.price(&manifest) { + Ok((cost, components)) => { + alternatives.push(AlternativeCost { + plan_id: Some(plan.envelope.plan_id), + total_cost: Some(cost.0), + unavailable_reason: None, + }); + if best.as_ref().is_none_or(|(previous, ..)| cost < *previous) { + best = Some((cost, plan, manifest, components)); + } + } + Err(reason) => alternatives.push(AlternativeCost { + plan_id: Some(plan.envelope.plan_id), + total_cost: None, + unavailable_reason: Some(reason), + }), + } + } + let (_, mut plan, selected_manifest, component_costs) = + best.ok_or_else(|| invalid("no feasible completely costed alternative"))?; + plan.cost_comparison = Some(WorkloadCostComparison { + data_snapshot_id: evidence.data_snapshot_id.clone(), + model_version: evidence.model_version.clone(), + selected_plan_id: plan.envelope.plan_id, + selected_manifest, + component_costs, + alternatives, + }); + Ok(plan) +} + +/// The current executor exposes continuously maintained state and the native +/// exact backend. Additional Planner-produced forests can use `select` directly. +pub fn with_exact_alternative( + request: PlanningRequest, +) -> Result, CompileError> { + let mut exact = request.clone(); + for query in &mut exact.queries { + let parsed = crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .map_err(|error| invalid(error.to_string()))?; + query.post_asap = crate::planner_selection::keep_pre_asap(&parsed) + .map_err(|error| invalid(error.to_string()))?; + } + if request + .queries + .iter() + .zip(&exact.queries) + .all(|(a, b)| a.post_asap == b.post_asap) + { + Ok(vec![request]) + } else { + Ok(vec![request, exact]) + } +} + +#[cfg(test)] +mod tests { + use super::super::compiler::BackendLocalPlanningSnapshot; + use super::*; + + fn fixture() -> BackendLocalPlanningSnapshot { + serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap() + } + + fn quoted() -> ( + Vec, + DeploymentEnvironment, + WorkloadCostEvidence, + ) { + let (request, env) = fixture().planning_request().unwrap(); + let candidates = with_exact_alternative(request).unwrap(); + let quotes = candidates + .iter() + .map(|candidate| { + let plan = PhysicalCompiler + .compile(candidate.clone(), env.clone()) + .unwrap(); + let manifest = manifest(&plan, &candidate.queries).unwrap(); + let unit_costs = manifest + .components + .keys() + .map(|id| (id.clone(), 1.0)) + .collect(); + WorkloadQuote { + manifest, + executable: true, + unit_costs, + } + }) + .collect(); + let evidence = WorkloadCostEvidence { + data_snapshot_id: "fixture-data-v1".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: env.observed_at_unix_ms, + valid_for_ms: env.max_evidence_age_ms, + quotes, + }; + (candidates, env, evidence) + } + + // All input metrics need upkeep quotes; repeated reads share that upkeep. + #[test] + fn exact_manifest_covers_and_deduplicates_query_sources() { + for (query, expected) in [ + ( + "sum_over_time(m[1m]) + sum_over_time(n[1m])", + vec!["m", "n"], + ), + ("sum_over_time(m[1m]) + count_over_time(m[1m])", vec!["m"]), + ] { + let (mut request, env) = fixture().planning_request().unwrap(); + request.queries[0].query_string = query.into(); + let exact = with_exact_alternative(request).unwrap().pop().unwrap(); + let plan = PhysicalCompiler + .compile(exact.clone(), env.clone()) + .unwrap(); + let manifest = manifest(&plan, &exact.queries).unwrap(); + let sources: Vec<_> = manifest + .components + .iter() + .filter(|(id, _)| id.starts_with("source:")) + .map(|(_, demand)| { + demand.implementation["source"]["TimeSeries"]["metric"] + .as_str() + .unwrap() + }) + .collect(); + assert_eq!(sources, expected, "{query}"); + let mut evidence = WorkloadCostEvidence { + data_snapshot_id: "test-data".into(), + model_version: "test-model".into(), + observed_at_unix_ms: env.observed_at_unix_ms, + valid_for_ms: env.max_evidence_age_ms, + quotes: vec![WorkloadQuote { + unit_costs: manifest + .components + .keys() + .map(|id| (id.clone(), 1.0)) + .collect(), + manifest, + executable: true, + }], + }; + assert!(select(vec![exact.clone()], env.clone(), &evidence).is_ok()); + let source_id = evidence.quotes[0] + .unit_costs + .keys() + .rfind(|id| id.starts_with("source:")) + .unwrap() + .clone(); + evidence.quotes[0].unit_costs.remove(&source_id); + assert!( + select(vec![exact], env, &evidence).is_err(), + "missing input upkeep must fail closed" + ); + } + } + + // Hidden or unresolved sources must not yield a partially priced manifest. + #[test] + fn exact_source_discovery_rejects_unresolved_inputs() { + let accuracy = fixture().planning_request().unwrap().0.queries[0] + .accuracy + .clone(); + for query in ["info(m)", "{job=\"api\"}"] { + let parsed = + crate::query_parser::parse_query_expr_canonical(query, accuracy.clone()).unwrap(); + assert!(exact_source_metrics(&parsed).is_err(), "{query}"); + } + } + + #[test] + fn complete_cost_changes_selection_and_reports_shared_work_once() { + let (candidates, env, mut evidence) = quoted(); + assert_ne!(evidence.quotes[0].manifest, evidence.quotes[1].manifest); + assert!(evidence.quotes[0] + .manifest + .components + .keys() + .any(|key| key.starts_with("state:"))); + assert!(!evidence.quotes[1] + .manifest + .components + .keys() + .any(|key| key.starts_with("state:"))); + for cost in evidence.quotes[1].unit_costs.values_mut() { + *cost = 1000.0; + } + let warm = select(candidates.clone(), env.clone(), &evidence).unwrap(); + assert_eq!(warm.envelope.plan_id, evidence.quotes[0].manifest.plan_id); + let report = warm.cost_comparison.unwrap(); + assert_eq!( + report.component_costs.len(), + report.selected_manifest.components.len() + ); + assert_eq!(report.alternatives.len(), 2); + assert!(report.alternatives.iter().all(|a| a.total_cost.is_some())); + for (id, cost) in &mut evidence.quotes[0].unit_costs { + if id.ends_with(":residency") { + *cost = 1e9; + } + } + let raw = select(candidates, env, &evidence).unwrap(); + assert_eq!(raw.envelope.plan_id, evidence.quotes[1].manifest.plan_id); + } + + #[test] + fn incomplete_unavailable_and_wrong_generation_quotes_are_not_free() { + let (candidates, env, mut evidence) = quoted(); + evidence.quotes[0].unit_costs.pop_first(); + let plan = select(candidates.clone(), env.clone(), &evidence).unwrap(); + assert!(plan.cost_comparison.unwrap().alternatives[0] + .unavailable_reason + .is_some()); + evidence.quotes[1].executable = false; + assert!(select(candidates.clone(), env.clone(), &evidence).is_err()); + let (_, _, mut evidence) = quoted(); + evidence + .quotes + .iter_mut() + .for_each(|quote| quote.manifest.capability_snapshot_id.push_str("-wrong")); + assert!(select(candidates.clone(), env.clone(), &evidence).is_err()); + let (_, _, mut evidence) = quoted(); + evidence.observed_at_unix_ms = env.observed_at_unix_ms + 1; + assert!(select(candidates, env, &evidence).is_err()); + } + + #[test] + fn v2_snapshot_requires_quotes_and_roundtrips_selection() { + let mut snapshot = fixture(); + snapshot.snapshot_version = 2; + assert!(snapshot.clone().compile().is_err()); + let (_, _, evidence) = quoted(); + snapshot.workload_cost_evidence = Some(evidence); + let snapshot: BackendLocalPlanningSnapshot = + serde_json::from_str(&serde_json::to_string(&snapshot).unwrap()).unwrap(); + assert!(snapshot.compile().unwrap().cost_comparison.is_some()); + } + + #[test] + fn second_consumer_adds_reads_not_another_shared_state() { + let (request, env) = fixture().planning_request().unwrap(); + let first = manifest( + &PhysicalCompiler + .compile(request.clone(), env.clone()) + .unwrap(), + &request.queries, + ) + .unwrap(); + let mut shared = request.clone(); + let mut second = shared.queries[0].clone(); + second.query_id = "second-consumer".into(); + second.query_string = "quantile_over_time(0.5, m[1m])".into(); + shared.queries.push(second); + let roots = shared + .queries + .iter() + .map(|query| { + std::rc::Rc::new( + crate::query_parser::parse_query_expr_canonical( + &query.query_string, + query.accuracy.clone(), + ) + .unwrap(), + ) + }) + .collect(); + super::super::compiler::select_workload_roots(&mut shared.queries, roots, &shared.evidence) + .unwrap(); + let plan = PhysicalCompiler.compile(shared.clone(), env).unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 1); + let second = manifest(&plan, &shared.queries).unwrap(); + let states = |m: &WorkloadCostManifest| { + m.components + .keys() + .filter(|key| key.starts_with("state:")) + .cloned() + .collect::>() + }; + assert_eq!(states(&first), states(&second)); + assert!(second.components.len() > first.components.len()); + } + + #[test] + fn exact_alternative_does_not_require_unused_state_implementation_evidence() { + let (candidates, env, evidence) = quoted(); + let mut exact = candidates[1].clone(); + assert_eq!(with_exact_alternative(exact.clone()).unwrap().len(), 1); + exact.queries[0].window_implementations.clear(); + assert!(select(vec![exact], env, &evidence).is_ok()); + } + + #[test] + fn altered_horizon_duplicate_and_invalid_costs_are_rejected() { + let (candidates, env, mut evidence) = quoted(); + evidence + .quotes + .iter_mut() + .for_each(|quote| quote.manifest.horizon_seconds += 1.0); + assert!(select(candidates.clone(), env.clone(), &evidence).is_err()); + let (_, _, mut evidence) = quoted(); + evidence.quotes.extend(evidence.quotes.clone()); + assert!(select(candidates.clone(), env.clone(), &evidence).is_err()); + let (_, _, mut evidence) = quoted(); + for quote in &mut evidence.quotes { + *quote.unit_costs.values_mut().next().unwrap() = -1.0; + } + assert!(select(candidates, env, &evidence).is_err()); + } +} diff --git a/control_plane/src/planner_selection.rs b/control_plane/src/planner_selection.rs index a41ef105..c464bfed 100644 --- a/control_plane/src/planner_selection.rs +++ b/control_plane/src/planner_selection.rs @@ -158,12 +158,33 @@ pub fn select_workload( roots: Vec<(usize, Rc)>, accuracy: AccuracyTarget, cost_model: &dyn CostModel, +) -> Result)>, SelectionError> { + select_workload_with_evidence( + roots, + accuracy, + cost_model, + &asap_aware_mapping::NoAccuracyEvidence, + ) +} + +/// The entire cohort uses the same scoped accuracy certificate; callers must +/// not spread one query's evidence to unrelated workload roots. +pub fn select_workload_with_evidence( + roots: Vec<(usize, Rc)>, + accuracy: AccuracyTarget, + cost_model: &dyn CostModel, + evidence: &dyn AccuracyEvidenceProvider, ) -> Result)>, SelectionError> { // Canonical CSE still runs inside search_workload_with_targets. Do not // offer CSE's per-invocation recompute alternative: this runtime currently // provisions continuously maintained, content-addressed state only. let strategies: Vec> = vec![ - Box::new(SketchAlgorithmStrategy::new(cost_model)), + Box::new(SketchAlgorithmStrategy::with_models_and_evidence( + cost_model, + &asap_aware_mapping::DefaultAccuracyModel, + &asap_aware_mapping::EqualSplitAllocator, + evidence, + )), Box::new(asap_aware_mapping::SemanticEquivalentRewriteStrategy), ]; let space = asap_aware_mapping::search_workload_with_targets( diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index 19a86813..aa2b5d73 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -680,6 +680,7 @@ impl Replanner { Some(BackendStageConfig { aggregations: vec![BackendAggregation { item_label: None, + heap_update_mode: None, aggregation_id: format!("exact-{}-{}", workload.metric_name, role), metric_name: workload.metric_name.clone(), family: SummaryFamilyType::ExactAggregate(exact_kind, exact_params), @@ -1310,6 +1311,7 @@ mod tests { BackendStageConfig { aggregations: vec![BackendAggregation { item_label: None, + heap_update_mode: None, aggregation_id: "exact-http_requests_total-sum".to_string(), metric_name: "http_requests_total".to_string(), family: planner_types::post_asap::SummaryFamilyType::ExactAggregate( diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index 2d520078..d093312d 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -30,4 +30,4 @@ xxhash-rust = { version = "0.8", features = ["xxh64"] } # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 50b3a0d9..f8b43fbe 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -38,7 +38,7 @@ control_plane = { path = "../control_plane" } # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "378a7547ede629a64e84c9f7c810226ce196cce9" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c3410d14865497758d212e4265ad25c782187de1" } # Shared external (workspace) serde.workspace = true diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 1af9c00f..12866f28 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -597,13 +597,9 @@ fn flush_barrier_drops(_state: &IngestState, drops: &HashMap, driver_t /// distinct `(rack, node, pod)` tuple under a `grouping_labels=[zone]` /// policy would mint its own sid and never roll up. /// -/// `agg_kind` is `ExactAgg { ... }` for both raw-sample and opaque- -/// envelope sketch paths so the resolver key matches the signature -/// `reconcile_from_streaming_config` derives from the same config; the -/// modified-OTLP first-class sketch path takes a different sid- -/// resolution route inside `route_modified_otlp_sketches_to_precompute` -/// because it carries per-DP `(SketchAlgorithm, SketchConfig)` and -/// must distinguish (e.g.) DDSketch vs Kll over the same series. +/// Configured ingest shares the policy-aware physical identity used by the +/// live storage sink and backfill. Unbound modified-OTLP sketches retain +/// their separate wire-level identity protocol. fn resolve_bucket_sid_for_agg_config( ingest_state: &Arc, config: &asap_types::aggregation_config::AggregationConfig, @@ -619,14 +615,8 @@ fn resolve_bucket_sid_for_agg_config( }) .collect(); let fp = crate::drivers::ingest::canonical_attrs_fingerprint(&grouping_pairs); - let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg { - agg_type: config.aggregation_type, - parameters_canonical: crate::storage_engines::sketch_db::data::canonical_parameters( - &config.parameters, - ), - spatial_filter_canonical: config.spatial_filter_normalized.clone(), - }; - let agg_kind_canonical = agg_kind.canonical_string(); + let agg_kind_canonical = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(config); let sid = ingest_state .series_resolver .resolve(&config.metric, &fp, &agg_kind_canonical); @@ -4656,14 +4646,8 @@ mod sid_bucketing_tests { // regardless of group_key shape — the test pin is on sid // assignment, not on group_key content), then verify the // sid matches the resolver mint for THAT zone. - let agg_kind = crate::storage_engines::sketch_db::data::AggKind::ExactAgg { - agg_type: cfg.aggregation_type, - parameters_canonical: crate::storage_engines::sketch_db::data::canonical_parameters( - &cfg.parameters, - ), - spatial_filter_canonical: cfg.spatial_filter_normalized.clone(), - }; - let agg_kind_canonical = agg_kind.canonical_string(); + let agg_kind_canonical = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(&cfg); for (sid, _, _, samples) in &groups { let mut vals: Vec = samples.iter().map(|(_, _, v)| *v).collect(); vals.sort_by(|a, b| a.partial_cmp(b).unwrap()); diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 07901031..b90cd013 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -461,13 +461,17 @@ fn route_messages( }) .collect(); let attrs_fp = super::canonical_attrs_fingerprint(&grouping_pairs); - let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(config); - let sid = ingest.series_resolver.resolve( - &config.metric, - &attrs_fp, - &agg_kind.canonical_string(), - ); let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64()); + // A sketch family is not a complete physical identity. Two + // materializations may use the same family and grouping while + // differing in update semantics (for example count- versus + // value-weighted Top-K). Keep those states on distinct SIDs. + let materialization_kind = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(config); + let sid = + ingest + .series_resolver + .resolve(&config.metric, &attrs_fp, &materialization_kind); buckets .entry(sid) .or_insert_with(|| ((sid, policy_fp, group_key), Vec::new())) diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 5bdd2a4b..717cb081 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -482,6 +482,10 @@ impl HttpServer { get(handle_get_backend_plan).post(handle_post_backend_plan), ) .route("/api/v1/physical-plan", post(handle_post_physical_plan)) + .route( + "/api/v1/physical-plan/discard", + post(handle_discard_physical_plan), + ) .route( "/api/v1/physical-plan/activate", post(handle_activate_physical_plan), @@ -588,6 +592,10 @@ impl HttpServer { get(handle_get_backend_plan).post(handle_post_backend_plan), ) .route("/api/v1/physical-plan", post(handle_post_physical_plan)) + .route( + "/api/v1/physical-plan/discard", + post(handle_discard_physical_plan), + ) .route( "/api/v1/physical-plan/activate", post(handle_activate_physical_plan), @@ -6076,6 +6084,24 @@ async fn handle_activate_physical_plan( .into_response() } +async fn handle_discard_physical_plan( + State(state): State, + axum::Json(request): axum::Json, +) -> axum::response::Response { + use axum::response::IntoResponse; + let Some(lifecycle) = state.physical_plan_lifecycle.as_ref() else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "physical-plan lifecycle is not attached", + ) + .into_response(); + }; + match lifecycle.discard_staged(request.plan_id, request.plan_version) { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(error) => (StatusCode::CONFLICT, error.to_string()).into_response(), + } +} + async fn handle_physical_plan_status(State(state): State) -> axum::response::Response { use axum::response::IntoResponse; let Some(lifecycle) = state.physical_plan_lifecycle.as_ref() else { diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index d85a1146..0ca072a2 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -484,6 +484,12 @@ async fn main() -> Result<()> { let plan = snapshot .compile() .map_err(|error| format!("startup planning failed for {}: {error}", path.display()))?; + if let Some(comparison) = &plan.cost_comparison { + info!( + "Startup workload cost decision: {}", + serde_json::to_string(comparison)? + ); + } Some( data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest { precompute_plan: plan.precompute_plan, diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 37aa7009..34ce3c56 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -1191,7 +1191,7 @@ pub fn decode_label_value(s: &str) -> std::borrow::Cow<'_, str> { /// the key dimension *inside* the sketch (e.g., which bucket in a CMS, which /// entry in a MultipleSumAccumulator's HashMap). This matches the Arroyo SQL /// pattern: `udf(concat_ws(';', aggregated_labels), value)`. -fn apply_sample( +pub(crate) fn apply_sample( updater: &mut dyn AccumulatorUpdater, series_key: &str, val: f64, @@ -1199,7 +1199,19 @@ fn apply_sample( config: &AggregationConfig, ) { if updater.is_keyed() { - let key = extract_aggregated_key_from_series(series_key, config); + // Planner's PromQL Top-K item is the series identity. When no + // explicit aggregated labels are projected, retain the canonical + // series key instead of collapsing every series onto an empty item. + let key = if config.aggregated_labels.labels.is_empty() + && matches!( + config.aggregation_type, + crate::storage_engines::types::AggregationType::CountMinSketchWithHeap + | crate::storage_engines::types::AggregationType::CountSketchWithHeap + ) { + KeyByLabelValues::new_with_labels(vec![series_key.to_string()]) + } else { + extract_aggregated_key_from_series(series_key, config) + }; updater.update_keyed(&key, val, ts); } else { updater.update_single(val, ts); diff --git a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs index d9ca4b7c..0d843996 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs @@ -72,7 +72,6 @@ use tracing::debug; use crate::drivers::ingest::canonical_attrs_fingerprint; use crate::drivers::ingest::series_resolver::SeriesIdResolver; use crate::precompute_engine::worker::parse_labels_from_series_key; -use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; use crate::storage_engines::types::{AggregateCore, HotReloadStreamingConfig, KeyByLabelValues}; use asap_types::aggregation_config::AggregationConfig; use asap_types::PolicyFingerprint; @@ -119,12 +118,8 @@ fn build_group_key_label_values(group_key: &str) -> KeyByLabelValues { /// embedded in `series_key` (the `metric{k1="v1",k2="v2"}` text shape /// `RawSample::labels` holds), so we parse them out first. /// -/// The sid identity tuple is `(metric, attrs_fp, agg_kind_canonical)` -/// — identical to what the live ingest path computes, so the same -/// `(metric, grouping-values, agg_kind)` produces the SAME sid no -/// matter which path (live or backfill) saw the sample first. That -/// invariant is what lets backfill writes land in the same store -/// row the live ingest already populated for `[created_at, ∞)`. +/// Policy and grouping identity must match live ingestion so historical and +/// live windows occupy the same storage row. fn resolve_backfill_bucket_sid( resolver: &SeriesIdResolver, config: &AggregationConfig, @@ -138,12 +133,8 @@ fn resolve_backfill_bucket_sid( .map(|name| (name.as_str(), *labels.get(name.as_str()).unwrap_or(&""))) .collect(); let attrs_fp = canonical_attrs_fingerprint(&grouping_pairs); - let agg_kind = AggKind::ExactAgg { - agg_type: config.aggregation_type, - parameters_canonical: canonical_parameters(&config.parameters), - spatial_filter_canonical: config.spatial_filter_normalized.clone(), - }; - let agg_kind_canonical = agg_kind.canonical_string(); + let agg_kind_canonical = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(config); resolver.resolve(&config.metric, &attrs_fp, &agg_kind_canonical) } @@ -912,19 +903,10 @@ mod tests { assert_eq!(written, vec![(fp, (0u64, 100u64))]); } - /// B7.7 invariant: the sid the backfill processor mints for a - /// `(config, grouping-values)` tuple is bit-equal to the sid the - /// live ingest path's `resolve_bucket_sid_for_agg_config` would - /// mint via the SAME `SeriesIdResolver`. Locks the "live and - /// backfill share one sid namespace" contract — without it, the - /// `[created_at, ∞)` and `[0, created_at)` halves of the agg's - /// timeline would live under DIFFERENT sids and the query path - /// would only see half the history. + /// Replay and the actual live storage sink must resolve the same row. #[test] fn backfill_sid_matches_live_ingest_sid_for_same_grouping_values() { - use crate::drivers::ingest::canonical_attrs_fingerprint; use crate::drivers::ingest::series_resolver::SeriesIdResolver; - use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; let cfg = sum_config(1, "latency", vec!["svc", "zone"]); let resolver = SeriesIdResolver::new(); @@ -933,20 +915,29 @@ mod tests { let backfill_sid = resolve_backfill_bucket_sid(&resolver, &cfg, "latency{svc=\"a\",zone=\"z0\"}"); - // Live side: mirror what `resolve_bucket_sid_for_agg_config` - // in drivers/ingest/otel.rs does, manually here so the test - // doesn't need to drive the OTLP pipeline. - let live_attrs_fp = canonical_attrs_fingerprint(&[("svc", "a"), ("zone", "z0")]); - let live_agg_kind = AggKind::ExactAgg { - agg_type: cfg.aggregation_type, - parameters_canonical: canonical_parameters(&cfg.parameters), - spatial_filter_canonical: cfg.spatial_filter_normalized.clone(), - }; - let live_sid = resolver.resolve( - &cfg.metric, - &live_attrs_fp, - &live_agg_kind.canonical_string(), + // Exercise the actual live sink instead of duplicating its SID formula. + let store = crate::storage_engines::sketch_db::index::SketchStore::new(); + let output = crate::storage_engines::types::PrecomputedOutput::new( + 100, + 200, + Some( + crate::storage_engines::types::KeyByLabelValues::new_with_labels(vec![ + "a".into(), + "z0".into(), + ]), + ), + cfg.policy_fingerprint(), ); + let acc = + crate::precompute_engine::operators::sum_accumulator::SumAccumulator::with_sum(1.0); + let live_sid = store + .ingest_precompute_for_agg_config( + |metric, attrs, kind| resolver.resolve(metric, attrs, kind), + &cfg, + &output, + &acc, + ) + .expect("live sink write"); assert_eq!( backfill_sid, live_sid, diff --git a/data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs b/data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs index 0d00e57c..55ba2f77 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/window_builder.rs @@ -50,44 +50,16 @@ use crate::precompute_engine::accumulator_factory::{ create_accumulator_updater, AccumulatorUpdater, }; -use crate::precompute_engine::worker::parse_labels_from_series_key; +use crate::precompute_engine::worker::apply_sample; use crate::storage_engines::sketch_db::backfill::raw_sample_reader::RawSample; -use crate::storage_engines::types::{AggregateCore, KeyByLabelValues}; +use crate::storage_engines::types::AggregateCore; use asap_types::aggregation_config::AggregationConfig; -/// Extract the MultipleSubpopulation aggregated-label key from a -/// Prometheus-style series key. Duplicated from -/// `precompute_engine::worker::extract_aggregated_key_from_series` -/// (which is file-private). Kept here so the backfill module -/// doesn't force a `pub(crate)` on a live-path helper — the -/// dependency is one-way: worker does NOT import anything from -/// backfill. -/// -/// The implementation must track the live one exactly; the -/// end-to-end determinism test in `backfill_processor.rs` will -/// fail if they drift. -fn extract_aggregated_key(series_key: &str, config: &AggregationConfig) -> KeyByLabelValues { - let labels = parse_labels_from_series_key(series_key); - let mut values = Vec::new(); - for label_name in &config.aggregated_labels.labels { - if let Some(val) = labels.get(label_name.as_str()) { - values.push(val.to_string()); - } else { - values.push(String::new()); - } - } - KeyByLabelValues::new_with_labels(values) -} - /// Construct the accumulator for one `(agg_id, window)` pair by /// feeding `samples` in order into a fresh `AccumulatorUpdater`. /// -/// Sample format: `samples[i].labels` is the full series key -/// (Prometheus-style `metric{k="v",…}`); the function extracts -/// the MultipleSubpopulation key from the series key using the -/// same helper the live worker uses -/// (`extract_aggregated_key_from_series`), so the keyed dispatch -/// is bit-identical. +/// Samples carry full series keys. Replay uses the live worker's sample +/// dispatch so keyed identity and update semantics remain identical. /// /// Ordering contract: samples are consumed in the iteration order /// of the input `Vec`. §10.5 requires that the caller preserve @@ -101,15 +73,14 @@ pub fn build_backfilled_accumulator( samples: &[RawSample], ) -> Box { let mut updater: Box = create_accumulator_updater(config); - if updater.is_keyed() { - for s in samples { - let key = extract_aggregated_key(&s.labels, config); - updater.update_keyed(&key, s.value, s.timestamp_ms); - } - } else { - for s in samples { - updater.update_single(s.value, s.timestamp_ms); - } + for sample in samples { + apply_sample( + &mut *updater, + &sample.labels, + sample.value, + sample.timestamp_ms, + config, + ); } updater.take_accumulator() } @@ -151,6 +122,59 @@ mod tests { } } + // Replay must preserve each series and rank by the selected update mode. + #[test] + fn backfilled_topk_preserves_series_and_weight_mode() { + use crate::precompute_engine::operators::{ + CountMinSketchWithHeapAccumulator, CountSketchWithHeapAccumulator, + }; + for kind in [ + AggregationType::CountMinSketchWithHeap, + AggregationType::CountSketchWithHeap, + ] { + for mode in ["count", "value"] { + let mut config = sum_config(); + config.aggregation_type = kind; + config.parameters = serde_json::from_value(serde_json::json!({ + "d": 4, "w": 1024, "heap_size": 10, "weight_mode": mode + })) + .unwrap(); + let samples = vec![ + raw("m{svc=\"a\"}", 10, 100.0), + raw("m{svc=\"b\"}", 20, 2.0), + raw("m{svc=\"b\"}", 30, 3.0), + ]; + let acc = build_backfilled_accumulator(&config, &samples); + let mut ranked: Vec<(String, f64)> = if let Some(heap) = + acc.as_any() + .downcast_ref::() + { + heap.inner + .topk_heap_items() + .into_iter() + .map(|i| (i.key, i.value)) + .collect() + } else { + acc.as_any() + .downcast_ref::() + .unwrap() + .inner + .topk_heap_items() + .into_iter() + .map(|i| (i.key, i.value)) + .collect() + }; + ranked.sort_by(|a, b| b.1.total_cmp(&a.1)); + let expected = if mode == "count" { + vec![("m{svc=\"b\"}".into(), 2.0), ("m{svc=\"a\"}".into(), 1.0)] + } else { + vec![("m{svc=\"a\"}".into(), 100.0), ("m{svc=\"b\"}".into(), 5.0)] + }; + assert_eq!(ranked, expected, "{kind:?} {mode}"); + } + } + } + #[test] fn sum_accumulator_sums_all_samples_in_order() { let config = sum_config(); diff --git a/data_plane/src/storage_engines/sketch_db/data/mod.rs b/data_plane/src/storage_engines/sketch_db/data/mod.rs index 1d8aec4d..b150d2c3 100644 --- a/data_plane/src/storage_engines/sketch_db/data/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/data/mod.rs @@ -128,6 +128,18 @@ pub enum AggKind { }, } +/// Complete resolver identity for a configured materialization. All live and +/// replay paths must include policy semantics, not just the sketch family. +pub(crate) fn materialization_kind_for_config( + config: &asap_types::aggregation_config::AggregationConfig, +) -> String { + format!( + "{}|{}", + agg_kind_for_config(config).canonical_string(), + config.policy_fingerprint() + ) +} + /// Resolve the physical state family produced by a precompute policy. This is /// shared by SID minting and store registration so a sketch policy can never /// be minted as `ExactAgg` and later registered as `Sketch` (or vice versa). diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index ffec3adc..80dfbec4 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -1855,14 +1855,8 @@ impl SketchStore { // samples by sid up-front) skip the resolver round-trip by // invoking the sid-direct sibling. let (attrs_fp, _label_values_map) = build_attrs_fp_and_label_map(agg_cfg, output); - let agg_kind = crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg); - // Sid mint delegated to the caller's closure — typically - // `|m, fp, ak| series_resolver.resolve(m, fp, ak)`. Keeps the - // SketchStore free of any layer-inverted dependency on the - // resolver type (which lives in `drivers::ingest`). Tests - // pass either a real local resolver or a counter-mock - // closure. - let agg_kind_canonical = agg_kind.canonical_string(); + let agg_kind_canonical = + crate::storage_engines::sketch_db::data::materialization_kind_for_config(agg_cfg); let sid = mint_sid(&agg_cfg.metric, &attrs_fp, &agg_kind_canonical); self.ingest_precompute_with_sid(sid, agg_cfg, output, accumulator) } diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index 959a123b..5d75f2b6 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -309,6 +309,27 @@ impl PhysicalPlanLifecycle { Ok(()) } + /// Roll back a failed publication without touching active readers or state. + pub fn discard_staged( + &self, + plan_id: u64, + plan_version: u64, + ) -> Result<(), PhysicalPlanLifecycleError> { + let key = (plan_id, plan_version); + let mut state = self + .state + .lock() + .expect("physical-plan lifecycle lock poisoned"); + if state.staged.remove(&key).is_none() { + return Err(PhysicalPlanLifecycleError::NotStaged { + plan_id, + plan_version, + }); + } + state.statuses.remove(&key); + Ok(()) + } + pub fn activate( &self, plan_id: u64, @@ -981,6 +1002,25 @@ mod tests { assert_eq!(lifecycle.statuses()[0].phase, PhysicalPlanPhase::Retired); } + // Failed publication releases only its staging slot; active readers remain valid. + #[test] + fn discard_staged_allows_retry_and_never_discards_active() { + let active = HotReloadActivePhysicalPlan::new(physical_plan(7, 1, 100, None)); + let held_reader = active.snapshot(); + let lifecycle = PhysicalPlanLifecycle::new(active.clone()); + lifecycle + .stage(physical_plan(7, 2, 200, None), 150) + .unwrap(); + lifecycle.discard_staged(7, 2).unwrap(); + lifecycle + .stage(physical_plan(7, 2, 300, None), 250) + .unwrap(); + lifecycle.activate(7, 2, 300).unwrap(); + assert!(lifecycle.discard_staged(7, 2).is_err()); + assert_eq!(active.snapshot().backend_plan.plan_version, 2); + assert_eq!(held_reader.backend_plan.plan_version, 1); + } + #[test] fn materialization_readiness_is_generation_scoped_and_monotonic() { let active = HotReloadActivePhysicalPlan::new(physical_plan(7, 1, 100, None)); diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 02478793..edf36bc3 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -51,11 +51,20 @@ async fn wait_until_ready(client: &reqwest::Client, url: &str, child: &mut Child } fn series(metric: &str, samples: &[(i64, f64)]) -> TimeSeries { + series_with_labels(metric, &[], samples) +} + +fn series_with_labels(metric: &str, labels: &[(&str, &str)], samples: &[(i64, f64)]) -> TimeSeries { + let mut wire_labels = vec![Label { + name: "__name__".into(), + value: metric.into(), + }]; + wire_labels.extend(labels.iter().map(|(name, value)| Label { + name: (*name).into(), + value: (*value).into(), + })); TimeSeries { - labels: vec![Label { - name: "__name__".into(), - value: metric.into(), - }], + labels: wire_labels, samples: samples .iter() .map(|(timestamp, value)| Sample { @@ -138,9 +147,46 @@ async fn shared_exact_dashboard_executes_selected_workload() { }) .collect(), ); - let typed: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + let mut typed: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(snapshot.clone()).unwrap(); + let (request, environment) = typed.clone().planning_request().unwrap(); + let candidates = + control_plane::physical::workload_cost::with_exact_alternative(request).unwrap(); + let quotes = candidates + .into_iter() + .enumerate() + .map(|(index, candidate)| { + let plan = control_plane::physical::compiler::PhysicalCompiler + .compile(candidate.clone(), environment.clone()) + .unwrap(); + let manifest = + control_plane::physical::workload_cost::manifest(&plan, &candidate.queries) + .unwrap(); + let unit_costs = manifest + .components + .keys() + .map(|key| (key.clone(), if index == 0 { 1.0 } else { 1000.0 })) + .collect(); + control_plane::physical::workload_cost::WorkloadQuote { + manifest, + executable: true, + unit_costs, + } + }) + .collect(); + typed.snapshot_version = 2; + typed.workload_cost_evidence = Some( + control_plane::physical::workload_cost::WorkloadCostEvidence { + data_snapshot_id: "process-fixture-v1".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: environment.observed_at_unix_ms, + valid_for_ms: environment.max_evidence_age_ms, + quotes, + }, + ); + snapshot = serde_json::to_value(&typed).unwrap(); let plan = typed.compile().unwrap(); + assert!(plan.cost_comparison.is_some()); assert_eq!(plan.precompute_plan.materializations.len(), 1); assert_eq!(plan.query_plan.entries.len(), 3); let snapshot_path = output_dir.path().join("snapshot.json"); @@ -444,8 +490,9 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() (base + 9_400, 12.0), ], ), - series( + series_with_labels( "asap_demo_gauge", + &[("job", "api")], &[ (base + 500, 1.0), (base + 1_700, 2.0), @@ -457,6 +504,20 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() (base + 9_400, 8.0), ], ), + series_with_labels( + "asap_demo_gauge", + &[("job", "worker")], + &[(base + 700, 100.0), (base + 3_100, 100.0)], + ), + series_with_labels( + "asap_demo_gauge", + &[("job", "cron")], + &[ + (base + 900, 10.0), + (base + 2_100, 10.0), + (base + 3_700, 10.0), + ], + ), series( "asap_demo_latency_ms", &[ @@ -476,7 +537,11 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() let watermark_advance = WriteRequest { timeseries: vec![ series("asap_demo_counter_total", &[(base + 10_500, 15.0)]), - series("asap_demo_gauge", &[(base + 10_500, 9.0)]), + series_with_labels( + "asap_demo_gauge", + &[("job", "api")], + &[(base + 10_500, 9.0)], + ), series("asap_demo_latency_ms", &[(base + 10_500, 55.0)]), ], }; @@ -531,10 +596,49 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() &backend_log, ) .await; + let topk_sum = wait_for_warm_instant( + &client, + &backend, + "topk(1, sum_over_time(asap_demo_gauge[5s]))", + first_eval, + &backend_log, + ) + .await; + let topk_count = wait_for_warm_instant( + &client, + &backend, + "topk(1, count_over_time(asap_demo_gauge[5s]))", + first_eval, + &backend_log, + ) + .await; + assert_eq!( + first_value(&topk_sum, "value"), + Some(200.0), + "value-weighted Top-K must select worker: {topk_sum}" + ); + assert_eq!( + first_value(&topk_count, "value"), + Some(4.0), + "count-weighted Top-K must select api: {topk_count}" + ); + assert_eq!(topk_sum["data"]["result"].as_array().map(Vec::len), Some(1)); + assert_eq!( + topk_count["data"]["result"].as_array().map(Vec::len), + Some(1) + ); + assert_eq!( + topk_sum["data"]["result"][0]["metric"]["item"], + "asap_demo_gauge{job=\"worker\"}" + ); + assert_eq!( + topk_count["data"]["result"][0]["metric"]["item"], + "asap_demo_gauge{job=\"api\"}" + ); let rate_value = first_value(&rate, "value").expect("rate value"); let increase_value = first_value(&increase, "value").expect("increase value"); assert!((rate_value * 5.0 - increase_value).abs() < 1e-9); - assert!((first_value(&sum, "value").expect("sum value") - 10.0).abs() < 1e-9); + assert!((first_value(&sum, "value").expect("sum value") - 240.0).abs() < 1e-9); let quantile_value = first_value(&quantile, "value").expect("quantile value"); assert!( (19.0..=31.0).contains(&quantile_value), @@ -546,6 +650,8 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() "increase(asap_demo_counter_total[5s])", "sum_over_time(asap_demo_gauge[5s])", "quantile_over_time(0.5, asap_demo_latency_ms[5s])", + "topk(1, sum_over_time(asap_demo_gauge[5s]))", + "topk(1, count_over_time(asap_demo_gauge[5s]))", ] { let response: Value = client .get(format!("{backend}/api/v1/query_range")) @@ -566,6 +672,40 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() is_warm(&response), "{query} did not use warm tier: {response}" ); + if query.starts_with("topk(") { + let mut ranked_points = response["data"]["result"] + .as_array() + .unwrap_or_else(|| panic!("missing Top-K range result for {query}: {response}")) + .iter() + .flat_map(|series| { + series["values"] + .as_array() + .into_iter() + .flatten() + .map(|point| { + ( + point[0].as_f64().expect("Top-K timestamp"), + point[1] + .as_str() + .expect("Top-K value") + .parse::() + .expect("numeric Top-K value"), + ) + }) + }) + .collect::>(); + ranked_points.sort_by(|left, right| left.0.total_cmp(&right.0)); + let expected = if query.contains("sum_over_time") { + vec![(first_eval, 200.0), (second_eval, 26.0)] + } else { + vec![(first_eval, 4.0), (second_eval, 4.0)] + }; + assert_eq!( + ranked_points, expected, + "wrong Top-K windows for {query}: {response}" + ); + continue; + } let values = response["data"]["result"][0]["values"] .as_array() .unwrap_or_else(|| panic!("missing range values for {query}: {response}")); @@ -574,6 +714,11 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() assert_eq!(values[1][0], second_eval); } + // Readiness polling may briefly reach the exact fallback before a newly + // closed warm window is visible. Every planned query above was required + // to converge to a warm answer; isolate the explicit fallback assertions. + fallback_calls.lock().await.clear(); + let fallback_instant: Value = client .get(format!("{backend}/api/v1/query")) .header("authorization", "Bearer demo") @@ -646,7 +791,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() let materializations = status["materializations"] .as_array() .expect("materialization statuses"); - assert_eq!(materializations.len(), 3); + assert_eq!(materializations.len(), 5); assert!(materializations .iter() .all(|entry| entry["phase"] == "serving")); @@ -660,7 +805,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly() .await .expect("metrics body"); assert!(metrics.contains("asap_remote_write_requests_total 4")); - assert!(metrics.contains("asap_remote_write_samples_total 27")); - assert!(metrics.contains("asap_remote_write_duplicates_total 24")); + assert!(metrics.contains("asap_remote_write_samples_total 32")); + assert!(metrics.contains("asap_remote_write_duplicates_total 29")); assert!(metrics.contains("asap_remote_write_rejected_requests_total 1")); } diff --git a/data_plane/tests/backend_process_e2e.rs b/data_plane/tests/backend_process_e2e.rs index a0beba65..f6ed75ef 100644 --- a/data_plane/tests/backend_process_e2e.rs +++ b/data_plane/tests/backend_process_e2e.rs @@ -16,7 +16,8 @@ use asap_otel_proto::tonic::metrics::v1::{ metric::Data, DdSketch, DdSketchDataPoint, DdSketchEncoding, Metric, ResourceMetrics, ScopeMetrics, }; -use asap_sketchlib::proto::sketchlib::DdSketchState; +use asap_precompute_rs::Precompute; +use asap_sketchlib::proto::sketchlib::{sketch_envelope, SketchEnvelope as ProtoEnvelope}; use control_plane::opamp::{ opamp_proto, CollectorPlanStatus, CollectorPlanStatusKind, COLLECTOR_PLAN_CAPABILITY, COLLECTOR_PLAN_MESSAGE, PLAN_STATUS_MESSAGE, @@ -67,26 +68,105 @@ async fn wait_http(client: &reqwest::Client, url: &str, child: &mut Child, name: panic!("{name} did not become ready at {url}"); } -fn ddsketch_export(metric: &str, timestamp_ns: u64, values: &[f64], alpha: f64) -> Vec { - let mut sketch = asap_sketchlib::DdSketch::new(alpha); +fn ddsketch_export( + metric: &str, + timestamp_ns: u64, + values: &[f64], + alpha: f64, + plan: &serde_json::Value, + sequence: u64, +) -> Vec { + let decoded = asap_precompute_rs::CollectorPlan::from_json( + &serde_json::to_vec(plan).unwrap(), + "whole-e2e-collector", + ) + .unwrap(); + let mut configs = decoded.to_precompute_config_set().unwrap().configs; + assert_eq!( + configs.len(), + 1, + "two query roots must create only one producer" + ); + let config = configs.remove(0); + assert_eq!(config.sketch_params["relative_accuracy"], alpha); + let runtime = asap_precompute_rs::precompute::PrecomputeImpl::new( + Some(config), + Some(Box::new(move || { + Box::new(asap_precompute_rs::sketches::DDSketchWrapper::new(alpha)) + })), + Some(Box::new(asap_precompute_rs::sketches::DDSketchObserver)), + ); for value in values { - sketch.update(*value); + runtime + .observe(&asap_precompute_rs::Observation::new( + timestamp_ns / 1_000_000 - 500, + metric, + vec![], + vec![asap_precompute_rs::KeyValue::new("service", "whole-e2e")], + asap_precompute_rs::ObservationValue { + kind: asap_precompute_rs::ObservationValueKind::Float, + float: *value, + ..Default::default() + }, + )) + .unwrap(); } - let point = DdSketchDataPoint { - attributes: vec![KeyValue { - key: "service".into(), + let envelopes = runtime.tick(timestamp_ns / 1_000_000); + assert_eq!(runtime.stats().input_observations, values.len() as u64); + assert_eq!(envelopes.len(), 1); + assert_eq!(envelopes[0].count, values.len() as u64); + let wire = ProtoEnvelope::decode(envelopes[0].payload.as_slice()).unwrap(); + let Some(sketch_envelope::SketchState::Ddsketch(state)) = wire.sketch_state else { + panic!("expected actual Collector DDSketch state") + }; + let materialization = plan["materializations"][0]["materialization"] + .as_u64() + .unwrap(); + let mut attributes = vec![KeyValue { + key: "service".into(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue("whole-e2e".into())), + }), + }]; + for (key, value) in [ + ("identity_version", "1".into()), + ("plan_id", plan["envelope"]["plan_id"].to_string()), + ("plan_version", plan["envelope"]["plan_version"].to_string()), + ( + "backend_compat", + control_plane::backend_plan::BACKEND_COMPAT.into(), + ), + ("materialization", materialization.to_string()), + ( + "series_identity", + data_plane::drivers::ingest::canonical_attrs_fingerprint(&[("service", "whole-e2e")]), + ), + ( + "schema_id", + format!( + "{}:summary-state:v1:{materialization}", + control_plane::backend_plan::BACKEND_COMPAT + ), + ), + ("producer_id", "whole-e2e-collector".into()), + ("producer_epoch", "process-e2e".into()), + ("sequence", sequence.to_string()), + ("kind", "full".into()), + ("encoding", "sketchlib_protobuf_v1".into()), + ("checkpoint_id", format!("checkpoint-{sequence}")), + ] { + attributes.push(KeyValue { + key: format!("asap.frame.{key}"), value: Some(AnyValue { - value: Some(any_value::Value::StringValue("whole-e2e".into())), + value: Some(any_value::Value::StringValue(value)), }), - }], + }); + } + let point = DdSketchDataPoint { + attributes, start_time_unix_nano: timestamp_ns.saturating_sub(1_000_000_000), time_unix_nano: timestamp_ns, - sketch: DdSketchState { - alpha: sketch.wire_alpha(), - store_counts: sketch.store_counts, - store_offset: sketch.store_offset, - } - .encode_to_vec(), + sketch: state.encode_to_vec(), encoding: DdSketchEncoding::DdsketchEncodingProto as i32, exemplars: Vec::new(), flags: 0, @@ -178,7 +258,13 @@ async fn send_agent_message( .expect("send OpAMP AgentToServer"); } -async fn apply_next_collector_plan(address: String) -> serde_json::Value { +type CollectorSocket = + tokio_tungstenite::WebSocketStream>; + +async fn apply_next_collector_plan( + address: String, + accept: bool, +) -> (serde_json::Value, CollectorSocket) { let mut socket = connect_collector(&address).await; send_agent_message( &mut socket, @@ -190,7 +276,13 @@ async fn apply_next_collector_plan(address: String) -> serde_json::Value { }, ) .await; + respond_next_collector_plan(socket, accept).await +} +async fn respond_next_collector_plan( + mut socket: CollectorSocket, + accept: bool, +) -> (serde_json::Value, CollectorSocket) { let frame = tokio::time::timeout(Duration::from_secs(10), socket.next()) .await .expect("controller did not publish a collector plan") @@ -205,6 +297,12 @@ async fn apply_next_collector_plan(address: String) -> serde_json::Value { .expect("collector-plan custom message"); assert_eq!(custom.capability, COLLECTOR_PLAN_CAPABILITY); assert_eq!(custom.r#type, COLLECTOR_PLAN_MESSAGE); + let decoded = asap_precompute_rs::collector_plan::CollectorPlan::from_json( + &custom.data, + "whole-e2e-collector", + ) + .expect("actual Collector validator accepts the emitted plan"); + assert_eq!(decoded.to_precompute_config_set().unwrap().configs.len(), 1); let plan: serde_json::Value = serde_json::from_slice(&custom.data).expect("decode collector physical plan"); let plan_id = plan["envelope"]["plan_id"] @@ -217,8 +315,12 @@ async fn apply_next_collector_plan(address: String) -> serde_json::Value { let status = serde_json::to_vec(&CollectorPlanStatus { plan_id, plan_version, - status: CollectorPlanStatusKind::Applied, - error: None, + status: if accept { + CollectorPlanStatusKind::Staged + } else { + CollectorPlanStatusKind::Failed + }, + error: (!accept).then(|| "injected Collector staging failure".into()), }) .expect("encode applied status"); send_agent_message( @@ -233,7 +335,61 @@ async fn apply_next_collector_plan(address: String) -> serde_json::Value { }, ) .await; - plan + (plan, socket) +} + +async fn quote_workload( + client: &reqwest::Client, + control_base: &str, + request: &mut serde_json::Value, +) { + request + .as_object_mut() + .unwrap() + .remove("workload_cost_evidence"); + let manifests: Vec = client + .post(format!( + "{control_base}/api/v1/physical-plan/cost-manifests" + )) + .json(&request) + .send() + .await + .unwrap() + .error_for_status() + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(manifests.len(), 2); + let quotes = manifests + .into_iter() + .enumerate() + .map(|(index, manifest)| { + let unit_costs = manifest + .components + .keys() + .map(|key| (key.clone(), 1.0)) + .collect(); + control_plane::physical::workload_cost::WorkloadQuote { + manifest, + executable: index == 0, + unit_costs, + } + }) + .collect(); + request["workload_cost_evidence"] = serde_json::to_value( + control_plane::physical::workload_cost::WorkloadCostEvidence { + data_snapshot_id: "whole-process-fixture-v1".into(), + model_version: "test-only-unit-costs".into(), + observed_at_unix_ms: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64, + valid_for_ms: 60000, + quotes, + }, + ) + .unwrap(); } #[tokio::test] @@ -312,45 +468,69 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { ) .await; - let collector = tokio::spawn(apply_next_collector_plan(control_opamp.clone())); + let collector = tokio::spawn(apply_next_collector_plan(control_opamp.clone(), true)); let observed_at_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .expect("system clock") .as_millis() as u64; + let mut request = serde_json::json!({ + "queries": [{ + "query_id": "whole-process-e2e-query", + "query_string": "quantile_over_time(0.99, whole_process_e2e_latency_ms[1s])", + "metric": "whole_process_e2e_latency_ms", + "window_secs": 1, + "group_by": ["service"], + "accuracy": {"Epsilon": 0.01}, + "window_implementations": [{ + "implementation_id": "collector-tumbling-v1", "framework": "tumbling", + "window_secs": 1, "pane_secs": 1, "state_layout": "anchored-pane-v1", + "cost": { + "model_version": "process-e2e-v1", "workload_fingerprint": "shared-quantiles", + "observed_at_unix_ms": observed_at_ms, "valid_for_ms": 60000, + "horizon_seconds": 300.0, "cpu_cost": 1.0, "weighted_cost": 1.0, + "peak_memory_bytes": 4096, "network_bytes": 1024, "storage_bytes": 2048, + "source_scan_bytes": 0 + } + }], + "lifecycle": { + "evaluation_interval_ms": 1000, + "ingestion_rate_per_second": 100.0, + "evidence_observed_at_unix_ms": observed_at_ms, + "evidence_valid_for_ms": 60000, + "horizon_seconds": 300.0, + "costs": { + "build": 10.0, + "maintenance_per_update": 0.001, + "read": 0.1, + "retention_per_second": 0.001, + "retirement": 1.0 + } + } + }], + "collector_ids": ["whole-e2e-collector"], + "capability_snapshot_id": "whole-e2e-capabilities", + "evidence": {}, + "planner_revision": PLANNER_REVISION, + "max_evidence_age_ms": 60000, + "plan_version": 1, + "activation_unix_ms": observed_at_ms, + "expiry_unix_ms": null, + "backend_compat": control_plane::backend_plan::BACKEND_COMPAT, + "apply_timeout_ms": 10000 + }); + let mut second = request["queries"][0].clone(); + second["query_id"] = "whole-process-e2e-median".into(); + second["query_string"] = "quantile_over_time(0.5, whole_process_e2e_latency_ms[1s])".into(); + request["queries"].as_array_mut().unwrap().push(second); + // Obtain actual compiler requirements without publishing, then provide + // deterministic test-only quotes. The native fallback is unavailable in + // this deployment; a cost number alone must not make it executable. + quote_workload(&client, &control_base, &mut request).await; let publication_response = client .post(format!( "{control_base}/api/v1/physical-plan/compile-and-publish" )) - .json(&serde_json::json!({ - "queries": [{ - "query_id": "whole-process-e2e-query", - "query_string": "quantile_over_time(0.99, whole_process_e2e_latency_ms[30s])", - "metric": "whole_process_e2e_latency_ms", - "window_secs": 1, - "group_by": ["service"], - "accuracy": {"Epsilon": 0.01}, - "lifecycle": { - "evaluation_interval_ms": 1000, - "ingestion_rate_per_second": 100.0, - "evidence_observed_at_unix_ms": observed_at_ms, - "evidence_valid_for_ms": 60000, - "horizon_seconds": 300.0, - "costs": { - "build": 10.0, - "maintenance_per_update": 0.001, - "read": 0.1, - "retention_per_second": 0.001, - "retirement": 1.0 - } - } - }], - "collector_ids": ["whole-e2e-collector"], - "capability_snapshot_id": "whole-e2e-capabilities", - "evidence": {}, - "planner_revision": PLANNER_REVISION, - "max_evidence_age_ms": 60000, - "apply_timeout_ms": 10000 - })) + .json(&request) .send() .await .expect("request physical-plan publication"); @@ -365,7 +545,15 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { ); let publication: serde_json::Value = serde_json::from_str(&publication_body).expect("decode publication response"); - let collector_plan = collector.await.expect("collector task completed"); + assert_eq!( + publication["cost_comparison"]["alternatives"] + .as_array() + .unwrap() + .len(), + 2 + ); + assert!(publication["cost_comparison"]["alternatives"][1]["unavailable_reason"].is_string()); + let (collector_plan, collector_socket) = collector.await.expect("collector task completed"); assert_eq!( publication["plan_id"], collector_plan["envelope"]["plan_id"] @@ -407,10 +595,14 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .expect("system clock"); - let sample_ns = now.as_nanos() as u64; - let raw_values = (1..=100).map(|value| value as f64).collect::>(); + let window_ms = planned_window_secs * 1000; + let window_end_ms = (now.as_millis() as u64 / window_ms) * window_ms; + let sample_ns = window_end_ms * 1_000_000; + // These ranks are integral under both PromQL interpolation and sketch + // order-statistic readout; interpolation coverage is a separate contract. + let raw_values = (1..=101).map(|value| value as f64).collect::>(); let reference_p99 = exact_quantile(&raw_values, 0.99); - client + let ingestion = client .post(format!("http://{otlp_http}/v1/metrics")) .header("content-type", "application/x-protobuf") .body(ddsketch_export( @@ -418,44 +610,25 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { sample_ns, &raw_values, planned_alpha, + &collector_plan, + 1, )) .send() .await - .expect("POST OTLP to production data plane") - .error_for_status() - .expect("data plane accepted OTLP"); + .expect("POST OTLP to production data plane"); + let status = ingestion.status(); + let body = ingestion.text().await.unwrap(); + assert!(status.is_success(), "OTLP rejected: {status}: {body}"); - // Advance event time after the controller-selected tumbling window has - // really ended. The E2E uses no artificial future timestamp here. - let window_ms = planned_window_secs * 1000; - let now_ms = now.as_millis() as u64; - let window_end_ms = (now_ms / window_ms + 1) * window_ms; - tokio::time::sleep(Duration::from_millis(window_end_ms - now_ms + 100)).await; - let watermark_ns = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("system clock") - .as_nanos() as u64; - client - .post(format!("http://{otlp_http}/v1/metrics")) - .header("content-type", "application/x-protobuf") - .body(ddsketch_export( - "whole_process_e2e_latency_ms", - watermark_ns, - &[], - planned_alpha, - )) - .send() - .await - .expect("POST watermark OTLP to production data plane") - .error_for_status() - .expect("data plane accepted watermark"); - - let query = "quantile_over_time(0.99, whole_process_e2e_latency_ms[30s])"; + let query = "quantile_over_time(0.99, whole_process_e2e_latency_ms[1s])"; let mut last_response = serde_json::Value::Null; for _ in 0..50 { let response: serde_json::Value = client .get(format!("{data_base}/api/v1/query")) - .query(&[("query", query)]) + .query(&[ + ("query", query.to_string()), + ("time", (window_end_ms as f64 / 1000.0).to_string()), + ]) .send() .await .expect("query production data plane") @@ -474,6 +647,198 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { response["data"]["result"][0]["metric"]["service"], "whole-e2e" ); + let median: serde_json::Value = client + .get(format!("{data_base}/api/v1/query")) + .query(&[ + ( + "query", + "quantile_over_time(0.5, whole_process_e2e_latency_ms[1s])".to_string(), + ), + ("time", (window_end_ms as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let median_value = first_scalar(&median).expect("second consumer is warm"); + let reference = exact_quantile(&raw_values, 0.5); + assert!( + (median_value - reference).abs() / reference <= planned_alpha * 1.05, + "{median}" + ); + // A rejected successor must not replace the active generation. + request["plan_version"] = 2.into(); + request["activation_unix_ms"] = (std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64 + + 1000) + .into(); + quote_workload(&client, &control_base, &mut request).await; + let collector = tokio::spawn(respond_next_collector_plan(collector_socket, false)); + let failed = client + .post(format!( + "{control_base}/api/v1/physical-plan/compile-and-publish" + )) + .json(&request) + .send() + .await + .unwrap(); + let failed_status = failed.status().as_u16(); + let failed_body = failed.text().await.unwrap(); + assert_eq!(failed_status, 502, "{failed_body}"); + assert!( + failed_body.contains("injected Collector staging failure"), + "{failed_body}" + ); + let (rejected_plan, collector_socket) = collector.await.unwrap(); + let rejected_frame = client + .post(format!("http://{otlp_http}/v1/metrics")) + .header("content-type", "application/x-protobuf") + .body(ddsketch_export( + "whole_process_e2e_latency_ms", + sample_ns, + &[9999.0], + planned_alpha, + &rejected_plan, + 1, + )) + .send() + .await + .unwrap(); + assert_eq!( + rejected_frame.status().as_u16(), + 422, + "inactive generation frame was accepted" + ); + let still_active: serde_json::Value = client + .get(format!("{data_base}/api/v1/backend-plan")) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!( + still_active, backend_plan, + "failed rollout changed active plan" + ); + let still_warm: serde_json::Value = client + .get(format!("{data_base}/api/v1/query")) + .query(&[ + ("query", query.to_string()), + ("time", (window_end_ms as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(first_scalar(&still_warm), Some(value)); + // Retry the staged successor while queries are in flight. Each + // request must retain a complete active snapshot through cutover. + request["activation_unix_ms"] = (std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as u64 + + 250) + .into(); + let reader_client = client.clone(); + let reader_base = data_base.clone(); + let readers = tokio::spawn(async move { + for _ in 0..40 { + let response: serde_json::Value = reader_client + .get(format!("{reader_base}/api/v1/query")) + .query(&[ + ("query", query.to_string()), + ("time", (window_end_ms as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!( + first_scalar(&response), + Some(value), + "torn serving snapshot: {response}" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + }); + let collector = tokio::spawn(respond_next_collector_plan(collector_socket, true)); + let activated = client + .post(format!( + "{control_base}/api/v1/physical-plan/compile-and-publish" + )) + .json(&request) + .send() + .await + .unwrap(); + let status = activated.status(); + let body = activated.text().await.unwrap(); + assert!( + status.is_success(), + "successful successor rejected: {status}: {body}" + ); + let activated: serde_json::Value = serde_json::from_str(&body).unwrap(); + assert_eq!(activated["plan_version"], 2); + let (successor, _collector_socket) = collector.await.unwrap(); + readers.await.unwrap(); + let old_frame = client + .post(format!("http://{otlp_http}/v1/metrics")) + .header("content-type", "application/x-protobuf") + .body(ddsketch_export( + "whole_process_e2e_latency_ms", + sample_ns, + &[9999.0], + planned_alpha, + &collector_plan, + 2, + )) + .send() + .await + .unwrap(); + assert_eq!( + old_frame.status().as_u16(), + 422, + "retired generation accepted a write" + ); + let new_frame = client + .post(format!("http://{otlp_http}/v1/metrics")) + .header("content-type", "application/x-protobuf") + .body(ddsketch_export( + "whole_process_e2e_latency_ms", + sample_ns + 1_000_000_000, + &[200.0; 101], + planned_alpha, + &successor, + 1, + )) + .send() + .await + .unwrap(); + assert!(new_frame.status().is_success()); + let new_result: serde_json::Value = client + .get(format!("{data_base}/api/v1/query")) + .query(&[ + ("query", query.to_string()), + ("time", ((window_end_ms + 1000) as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!( + (first_scalar(&new_result).unwrap() - 200.0).abs() <= 200.0 * planned_alpha * 1.05, + "{new_result}" + ); return; } last_response = response; diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json index eb62b192..6f255629 100644 --- a/docs/examples/asapquery-compatibility-demo-snapshot.json +++ b/docs/examples/asapquery-compatibility-demo-snapshot.json @@ -34,6 +34,26 @@ }, "predictability": { "predictable": { "known_at": null } }, "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + }, + { + "query": "topk(1, sum_over_time(asap_demo_gauge[5s]))", + "demand": { "fixed_interval": 1000 }, + "requirements": { + "accuracy": { "explicit": { "EpsilonDelta": { "epsilon": 0.01, "delta": 0.01 } } }, + "response_latency": "unspecified" + }, + "predictability": { "predictable": { "known_at": null } }, + "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } + }, + { + "query": "topk(1, count_over_time(asap_demo_gauge[5s]))", + "demand": { "fixed_interval": 1000 }, + "requirements": { + "accuracy": { "explicit": { "EpsilonDelta": { "epsilon": 0.01, "delta": 0.01 } } }, + "response_latency": "unspecified" + }, + "predictability": { "predictable": { "known_at": null } }, + "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } } ], "data_workload": { @@ -76,6 +96,22 @@ "storage_bytes": 2048, "source_scan_bytes": 0, "weighted_cost": 1.0 + }, + "topk_evidence": { + "topk(1, sum_over_time(asap_demo_gauge[5s]))": { + "selected_lower_bound": 101.0, + "excluded_upper_bound": 100.0, + "interval_failure_probability": 0.005, + "observed_at_unix_ms": 9500, + "source": "compatibility-fixture" + }, + "topk(1, count_over_time(asap_demo_gauge[5s]))": { + "selected_lower_bound": 101.0, + "excluded_upper_bound": 100.0, + "interval_failure_probability": 0.005, + "observed_at_unix_ms": 9500, + "source": "compatibility-fixture" + } } }, "environment": { diff --git a/docs/examples/workload-cost-evidence.md b/docs/examples/workload-cost-evidence.md new file mode 100644 index 00000000..612f58e7 --- /dev/null +++ b/docs/examples/workload-cost-evidence.md @@ -0,0 +1,95 @@ +# Complete workload cost evidence + +The migrated, evidence-required startup profile uses `snapshot_version: 2`. +Version 1 and live requests without `workload_cost_evidence` remain compatibility +paths: their lifecycle estimates are not complete workload costs. + +## Workflow + +1. Prepare the canonical workload, deployment capabilities and implementation + evidence as in `asapquery-planning-snapshot.json`. Set version 2. +2. Obtain requirements without deploying: + + ```sh + cargo run -p control_plane --example workload_cost_manifest -- snapshot.json + ``` + + For distributed deployment, POST the normal compile-and-publish request, + without quotes, to `/api/v1/physical-plan/cost-manifests`. This endpoint only + compiles; it does not stage, publish to Collectors, or activate a generation. + It returns manifests for the compilable candidates in this runtime profile. +3. Have the deployment's cost/capability provider price each manifest. Attach a + `workload_cost_evidence` object to the snapshot or publication request: + + ```json + { + "data_snapshot_id": "deployment-input-generation-42", + "model_version": "calibrated-provider-v1", + "observed_at_unix_ms": 1780000000000, + "valid_for_ms": 60000, + "quotes": [{ + "manifest": "REPLACE with the complete returned manifest object", + "executable": true, + "unit_costs": { "REPLACE with every component ID": 1.0 } + }] + } + ``` + + This is a structural illustration, not runnable or calibrated evidence. + `unit_costs` must contain exactly the manifest's component keys. A quote + declares feasibility against the capability generation, including raw-data + access and the exact service when a candidate uses native fallback. +4. Compile/start or publish. The backend selects the cheapest completely quoted, + feasible candidate. The live response includes `cost_comparison`; startup + logs the same report. Preserve it with the deployment's evidence records. + +## What is priced + +This is a flat coverage manifest over the existing physical projection, not +another semantic DAG. Planner supplies the legal post-ASAP computations; costs +can change which bound workload is committed, not rewrite its semantics. + +- Source ingestion once per source/location over the horizon. Exact fallback + includes every named input metric, including both sides of binary expressions; + repeated references to a metric share one upkeep component. Queries whose + sources cannot be fully enumerated are unavailable for complete costing. +- Each shared state's build, update/merge, residency/spill and retirement once + per actual location, independent of the number of result consumers. +- Each Collector transmission rule over the horizon. +- Every reachable query operator, including reads, arithmetic, reductions and + complete engine-native fallback, multiplied by the query's recurrence. +- Each result's output cost. Exact-service input upkeep/storage has a separate + horizon component and is not silently omitted from a raw alternative. + +All quotes use one provider model's common cost units. A horizon quote includes +the complete stated partition's work, data volume/cardinality, maintained +groups and retention. Do not reuse a global ingestion rate as a per-metric rate. +Per-evaluation quotes exclude upkeep already charged in horizon components. +Sunk infrastructure may explicitly cost zero under the provider's documented +decision boundary; unknown costs may not. + +The manifest binds query requirements, demand, horizon, implementation details, +Planner revision, plan identity/version and capability generation. Evidence also +names its input generation, model and validity interval. A provider must refresh +quotes when those data assumptions change; this API does not discover or +authenticate telemetry itself. Freshness is evaluated at the supplied snapshot +decision time for replay, and at server time for live requests. + +Missing, duplicate, negative, nonfinite, stale, mismatched or infeasible quotes +make a candidate unavailable. If no completely quoted candidate remains, the +decision fails closed. Tests use explicit synthetic unit costs, not production +measurements. + +## Supported migration scope + +The default inventory compares the Planner-selected continuously maintained +workload with its whole-workload exact fallback. `workload_cost::select` also +accepts additional Planner-authorized, already-bindable forests. This does not +claim exhaustive search over every lifecycle, engine or Planner algorithm. +An exact alternative without an accessible native backend is unavailable even +if its numeric quote would be cheap. + +This provider-priced binding boundary deliberately avoids inventing physical +statistics to populate Planner's generic physical-formula provider. It completes +cost selection for the supported execution profile; production calibration and +additional provider implementations remain deployment work.