From ae117eb638fa6d63c97191e9a191fe2130e60138 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 3 Sep 2026 14:42:28 -0600 Subject: [PATCH] feat: install strict backend-local physical plans --- control_plane/src/physical/compiler.rs | 88 ++++- .../examples/asapquery/streaming-config.yaml | 35 -- .../drivers/ingest/prometheus_remote_write.rs | 138 +++++++- data_plane/src/drivers/query/servers/http.rs | 330 +++++++++++------- data_plane/src/main.rs | 127 ++++++- docs/user_guide/asapquery-profile.md | 30 +- 6 files changed, 541 insertions(+), 207 deletions(-) delete mode 100644 data_plane/examples/asapquery/streaming-config.yaml diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index f78da9f2b..8983ef4ab 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -216,12 +216,14 @@ pub struct PrecomputePlan { #[serde(rename_all = "snake_case")] pub enum IngestProtocol { ModifiedOtlpMetricsV1, + PrometheusRemoteWriteV1, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum TimestampUnit { UnixNanoseconds, + UnixMilliseconds, } /// Backend ingress semantics installed with the precompute projection. This @@ -331,7 +333,7 @@ pub struct ProducerContract { pub enum PrecomputePlanError { #[error("PrecomputePlan envelope does not match BackendPlan identity/lifecycle")] PlanIdentityMismatch, - #[error("precompute ingest endpoint must be /v1/metrics")] + #[error("unsupported precompute ingest protocol/endpoint/identity contract")] UnsupportedIngestEndpoint, #[error("duplicate materialization {0}")] DuplicateMaterialization(u64), @@ -417,6 +419,32 @@ impl PrecomputePlan { Ok(plan) } + /// Build the backend-local projection used when raw Prometheus samples + /// are precomputed inside ASAPQuery rather than by ASAPCollector. + pub fn build_backend_local( + envelope: PlanEnvelope, + materializations: Vec, + backend_plan: &BackendPlan, + ) -> Result { + let mut plan = Self::build( + envelope, + materializations, + backend_plan, + &["backend-local".into()], + )?; + plan.ingest = IngestContract { + protocol: IngestProtocol::PrometheusRemoteWriteV1, + endpoint_path: "/api/v1/write".into(), + timestamp_unit: TimestampUnit::UnixMilliseconds, + require_plan_identity: false, + require_materialization_identity: false, + require_registered_producer: false, + }; + plan.producers.clear(); + plan.validate_against_backend(backend_plan)?; + Ok(plan) + } + pub fn runtime_materializations( &self, ) -> Result, PrecomputePlanError> { @@ -430,11 +458,23 @@ impl PrecomputePlan { } pub fn validate(&self) -> Result<(), PrecomputePlanError> { - if self.ingest.endpoint_path != "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/v1/metrics" - || !self.ingest.require_plan_identity - || !self.ingest.require_materialization_identity - || !self.ingest.require_registered_producer - { + let valid_ingest = match self.ingest.protocol { + IngestProtocol::ModifiedOtlpMetricsV1 => { + self.ingest.endpoint_path == "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/v1/metrics" + && self.ingest.timestamp_unit == TimestampUnit::UnixNanoseconds + && self.ingest.require_plan_identity + && self.ingest.require_materialization_identity + && self.ingest.require_registered_producer + } + IngestProtocol::PrometheusRemoteWriteV1 => { + self.ingest.endpoint_path == "/api/v1/write" + && self.ingest.timestamp_unit == TimestampUnit::UnixMilliseconds + && !self.ingest.require_plan_identity + && !self.ingest.require_materialization_identity + && !self.ingest.require_registered_producer + } + }; + if !valid_ingest { return Err(PrecomputePlanError::UnsupportedIngestEndpoint); } let mut materializations = BTreeSet::new(); @@ -495,8 +535,10 @@ impl PrecomputePlan { } produced.insert(producer.materialization); } - if let Some(missing) = materializations.difference(&produced).next() { - return Err(PrecomputePlanError::MissingProducer(missing.0)); + if self.ingest.require_registered_producer { + if let Some(missing) = materializations.difference(&produced).next() { + return Err(PrecomputePlanError::MissingProducer(missing.0)); + } } Ok(()) } @@ -2333,6 +2375,36 @@ mod tests { } } + #[test] + fn backend_local_precompute_contract_has_no_collector_producers() { + let bundle = PhysicalCompiler + .compile( + request("q-quantile", "quantile_over_time(0.99, m[1m])"), + environment(10_000), + ) + .expect("compile"); + let plan = PrecomputePlan::build_backend_local( + bundle.envelope.clone(), + bundle.precompute_plan.materializations.clone(), + &bundle.backend_plan, + ) + .expect("backend-local contract"); + + assert_eq!( + plan.ingest.protocol, + IngestProtocol::PrometheusRemoteWriteV1 + ); + assert_eq!(plan.ingest.endpoint_path, "/api/v1/write"); + assert_eq!(plan.ingest.timestamp_unit, TimestampUnit::UnixMilliseconds); + assert!(plan.producers.is_empty()); + plan.validate_against_backend(&bundle.backend_plan) + .expect("valid backend-local projection"); + TransmissionPlan::build(bundle.envelope, &plan, &BTreeMap::new()) + .expect("empty backend-local transmission contract") + .validate(&plan) + .expect("valid empty transmission plan"); + } + #[test] fn multiple_readouts_share_one_precompute_materialization() { let mut planning_request = request("q-p90", "quantile_over_time(0.90, m[1m])"); diff --git a/data_plane/examples/asapquery/streaming-config.yaml b/data_plane/examples/asapquery/streaming-config.yaml deleted file mode 100644 index c2174a06c..000000000 --- a/data_plane/examples/asapquery/streaming-config.yaml +++ /dev/null @@ -1,35 +0,0 @@ -aggregations: - - aggregationType: Sum - aggregationSubType: '' - metric: compat_value - labels: - grouping: [job] - rollup: [] - aggregated: [] - parameters: {} - windowSize: 30 - windowType: tumbling - spatialFilter: '' - - aggregationType: Increase - aggregationSubType: '' - metric: compat_counter_total - labels: - grouping: [job] - rollup: [] - aggregated: [] - parameters: {} - windowSize: 30 - windowType: tumbling - spatialFilter: '' - - aggregationType: DDSketch - aggregationSubType: '' - metric: compat_latency_seconds - labels: - grouping: [job] - rollup: [] - aggregated: [] - parameters: - relative_accuracy: 0.01 - windowSize: 30 - windowType: tumbling - spatialFilter: '' diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 55d1a3767..b04fd2885 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -100,8 +100,8 @@ struct ReceiverInner { #[derive(Default)] struct DedupState { - values: HashMap<(String, i64), DedupValue>, - expiry: VecDeque<(Instant, String, i64)>, + values: HashMap<(u64, u64, String, i64), DedupValue>, + expiry: VecDeque<(Instant, u64, u64, String, i64)>, } #[derive(Clone, Copy, PartialEq, Eq)] @@ -136,6 +136,8 @@ pub enum RemoteWriteError { Conflict { series: String, timestamp: i64 }, #[error("deduplication capacity ({0}) is exhausted")] DedupCapacity(usize), + #[error("Remote Write requires an active PrometheusRemoteWriteV1 PhysicalPlan")] + InactivePhysicalPlan, #[error(transparent)] Backpressure(#[from] TryRouteError), } @@ -208,6 +210,24 @@ impl PrometheusRemoteWriteReceiver { let request = WriteRequest::decode(decoded.as_slice()) .map_err(|error| RemoteWriteError::Protobuf(error.to_string()))?; let samples = canonicalize_request(&request, config)?; + let physical_plan = self + .inner + .ingest + .physical_plan_snapshot() + .ok_or(RemoteWriteError::InactivePhysicalPlan)?; + if physical_plan.precompute_plan.envelope.plan_id == 0 + || !matches!( + physical_plan.precompute_plan.ingest.protocol, + control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1 + ) + || physical_plan.precompute_plan.ingest.endpoint_path != "/api/v1/write" + { + return Err(RemoteWriteError::InactivePhysicalPlan); + } + let plan_identity = ( + physical_plan.precompute_plan.envelope.plan_id, + physical_plan.precompute_plan.envelope.plan_version, + ); let now = Instant::now(); let mut dedup = self @@ -219,11 +239,16 @@ impl PrometheusRemoteWriteReceiver { // Validate conflicts both against committed history and inside this // request before reserving any worker capacity. - let mut batch_values: HashMap<(String, i64), DedupValue> = HashMap::new(); + let mut batch_values: HashMap<(u64, u64, String, i64), DedupValue> = HashMap::new(); let mut new_samples = Vec::with_capacity(samples.len()); let mut duplicates = 0u64; for sample in samples { - let key = (sample.series_key.clone(), sample.timestamp_ms); + let key = ( + plan_identity.0, + plan_identity.1, + sample.series_key.clone(), + sample.timestamp_ms, + ); let value = sample .value .map(|v| DedupValue::Number(v.to_bits())) @@ -249,15 +274,19 @@ impl PrometheusRemoteWriteReceiver { return Err(RemoteWriteError::DedupCapacity(config.max_dedup_entries)); } - let messages = route_messages(&new_samples, &self.inner.ingest); + let messages = route_messages(&new_samples, &self.inner.ingest, &physical_plan); self.inner .ingest .router .try_route_group_batch_atomic(messages)?; - for ((series, timestamp), value) in batch_values { - dedup.values.insert((series.clone(), timestamp), value); - dedup.expiry.push_back((now, series, timestamp)); + for ((plan_id, plan_version, series, timestamp), value) in batch_values { + dedup + .values + .insert((plan_id, plan_version, series.clone(), timestamp), value); + dedup + .expiry + .push_back((now, plan_id, plan_version, series, timestamp)); } let stale_count = new_samples .iter() @@ -288,10 +317,11 @@ impl DedupState { while self .expiry .front() - .is_some_and(|(accepted_at, _, _)| *accepted_at < cutoff) + .is_some_and(|(accepted_at, _, _, _, _)| *accepted_at < cutoff) { - if let Some((_, series, timestamp)) = self.expiry.pop_front() { - self.values.remove(&(series, timestamp)); + if let Some((_, plan_id, plan_version, series, timestamp)) = self.expiry.pop_front() { + self.values + .remove(&(plan_id, plan_version, series, timestamp)); } } } @@ -392,10 +422,14 @@ fn canonicalize_labels( Ok((metric, attrs, series_key)) } -fn route_messages(samples: &[CanonicalSample], ingest: &Arc) -> Vec { +fn route_messages( + samples: &[CanonicalSample], + ingest: &Arc, + physical_plan: &crate::storage_engines::types::ActivePhysicalPlan, +) -> Vec { type Bucket = (u64, asap_types::PolicyFingerprint, String); type RoutedSample = (String, i64, f64); - let snapshot = ingest.config_snapshot(); + let snapshot = physical_plan.runtime_config.clone(); let _ = crate::storage_engines::sketch_db::lifecycle::reconcile_if_config_changed( ingest.sketch_index.as_ref(), &snapshot, @@ -492,7 +526,10 @@ mod tests { use super::*; use crate::precompute_engine::ingest_handler::IngestObservability; use crate::precompute_engine::series_router::SeriesRouter; - use crate::storage_engines::types::{HotReloadStreamingConfig, StreamingConfig}; + use crate::storage_engines::types::{ + ActivePhysicalPlan, BackendStorageRouting, HotReloadActivePhysicalPlan, + HotReloadStreamingConfig, StreamingConfig, + }; use tokio::sync::mpsc; fn compressed(request: WriteRequest) -> Vec { @@ -501,13 +538,61 @@ mod tests { .unwrap() } + fn physical_config(streaming: StreamingConfig) -> HotReloadStreamingConfig { + use control_plane::physical::compiler::{ + FrameIdentityContract, IngestContract, IngestProtocol, PlanEnvelope, PrecomputePlan, + SequenceScope, TimestampUnit, TransmissionPlan, PLANNER_REVISION, + }; + let envelope = PlanEnvelope { + plan_id: 7, + plan_version: 3, + generated_at_unix_ms: 1, + activation_unix_ms: 1, + expiry_unix_ms: None, + backend_compat: control_plane::backend_plan::BACKEND_COMPAT.into(), + planner_revision: PLANNER_REVISION.into(), + capability_snapshot_id: "test".into(), + }; + let active = ActivePhysicalPlan { + precompute_plan: PrecomputePlan { + envelope: envelope.clone(), + ingest: IngestContract { + protocol: IngestProtocol::PrometheusRemoteWriteV1, + endpoint_path: "/api/v1/write".into(), + timestamp_unit: TimestampUnit::UnixMilliseconds, + require_plan_identity: false, + require_materialization_identity: false, + require_registered_producer: false, + }, + schemas: Vec::new(), + producers: Vec::new(), + materializations: streaming.aggregation_configs.values().cloned().collect(), + }, + transmission_plan: TransmissionPlan { + envelope, + frame_identity: FrameIdentityContract { + identity_version: 1, + sequence_scope: SequenceScope::MaterializationSeriesProducerEpoch, + require_checkpoint_for_full: true, + require_base_checkpoint_for_delta: true, + }, + rules: Vec::new(), + }, + runtime_config: Arc::new(streaming), + backend_plan: Arc::new(control_plane::backend_plan::BackendPlan::default()), + query_plan: Arc::new(control_plane::query_plan::QueryPlan::empty()), + storage_routing: Arc::new(BackendStorageRouting::empty()), + }; + HotReloadStreamingConfig::from_active(HotReloadActivePhysicalPlan::new(active)) + } + fn receiver(config: PrometheusRemoteWriteConfig) -> PrometheusRemoteWriteReceiver { let (sender, _receiver) = mpsc::channel(8); let ingest = Arc::new(IngestState { router: SeriesRouter::new(vec![sender]), samples_ingested: AtomicU64::new(0), samples_blocked_by_schema_barrier: AtomicU64::new(0), - hot_reload_config: HotReloadStreamingConfig::new(StreamingConfig::default()), + hot_reload_config: physical_config(StreamingConfig::default()), pass_raw_samples: false, sketch_snapshots: dashmap::DashMap::new(), series_resolver: Arc::new(super::super::SeriesIdResolver::new()), @@ -545,7 +630,7 @@ mod tests { router: SeriesRouter::new(vec![sender]), samples_ingested: AtomicU64::new(0), samples_blocked_by_schema_barrier: AtomicU64::new(0), - hot_reload_config: HotReloadStreamingConfig::new(streaming), + hot_reload_config: physical_config(streaming), pass_raw_samples: false, sketch_snapshots: dashmap::DashMap::new(), series_resolver: Arc::new(super::super::SeriesIdResolver::new()), @@ -581,6 +666,27 @@ mod tests { }) } + #[test] + fn rejects_writes_without_an_active_physical_plan() { + let (sender, _worker) = mpsc::channel(1); + let ingest = Arc::new(IngestState { + router: SeriesRouter::new(vec![sender]), + samples_ingested: AtomicU64::new(0), + samples_blocked_by_schema_barrier: AtomicU64::new(0), + hot_reload_config: HotReloadStreamingConfig::new(StreamingConfig::default()), + pass_raw_samples: false, + sketch_snapshots: dashmap::DashMap::new(), + series_resolver: Arc::new(super::super::SeriesIdResolver::new()), + sketch_index: Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()), + observability: IngestObservability::default(), + }); + let receiver = PrometheusRemoteWriteReceiver::new(Default::default(), ingest); + assert!(matches!( + receiver.accept(&one_sample(1.0)), + Err(RemoteWriteError::InactivePhysicalPlan) + )); + } + #[test] fn canonicalizes_labels_and_recognizes_staleness() { let request = WriteRequest { diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index ce33f9792..7b0ad9a82 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -519,6 +519,8 @@ impl HttpServer { .route(range_query_endpoint, get(handle_range_query)) .route(range_query_endpoint, post(handle_range_query_post)) .route(runtime_info_path, get(handle_runtime_info)) + .route("/metrics", get(handle_metrics)) + .route("/api/v1/health", get(handle_health)) .route( "/api/v1/write", post(handle_prometheus_remote_write) @@ -2223,8 +2225,66 @@ mod tests { } async fn setup_remote_write_test_server() -> (u16, PrometheusRemoteWriteReceiver) { + use control_plane::physical::compiler::{ + FrameIdentityContract, IngestContract, IngestProtocol, PlanEnvelope, PrecomputePlan, + SequenceScope, TimestampUnit, TransmissionPlan, PLANNER_REVISION, + }; let streaming_config = Arc::new(StreamingConfig::default()); - let hot_reload = HotReloadStreamingConfig::new((*streaming_config).clone()); + let envelope = PlanEnvelope { + plan_id: 7, + plan_version: 1, + generated_at_unix_ms: 0, + activation_unix_ms: 1, + expiry_unix_ms: None, + backend_compat: control_plane::backend_plan::BACKEND_COMPAT.into(), + planner_revision: PLANNER_REVISION.into(), + capability_snapshot_id: "test".into(), + }; + let active = crate::storage_engines::types::HotReloadActivePhysicalPlan::new( + crate::storage_engines::types::ActivePhysicalPlan { + precompute_plan: PrecomputePlan { + envelope: envelope.clone(), + ingest: IngestContract { + protocol: IngestProtocol::PrometheusRemoteWriteV1, + endpoint_path: "/api/v1/write".into(), + timestamp_unit: TimestampUnit::UnixMilliseconds, + require_plan_identity: false, + require_materialization_identity: false, + require_registered_producer: false, + }, + schemas: Vec::new(), + producers: Vec::new(), + materializations: Vec::new(), + }, + transmission_plan: TransmissionPlan { + envelope, + frame_identity: FrameIdentityContract { + identity_version: 1, + sequence_scope: SequenceScope::MaterializationSeriesProducerEpoch, + require_checkpoint_for_full: true, + require_base_checkpoint_for_delta: true, + }, + rules: Vec::new(), + }, + runtime_config: streaming_config.clone(), + backend_plan: Arc::new(control_plane::backend_plan::BackendPlan { + plan_id: 7, + plan_version: 1, + activation_unix_ms: 1, + backend_compat: control_plane::backend_plan::BACKEND_COMPAT.into(), + ..Default::default() + }), + query_plan: Arc::new(control_plane::query_plan::QueryPlan { + plan_id: 7, + plan_version: 1, + entries: Default::default(), + }), + storage_routing: Arc::new( + crate::storage_engines::types::BackendStorageRouting::empty(), + ), + }, + ); + let hot_reload = HotReloadStreamingConfig::from_active(active.clone()); let (sender, _worker) = mpsc::channel(8); let ingest = Arc::new(IngestState { router: SeriesRouter::new(vec![sender]), @@ -2249,6 +2309,7 @@ mod tests { Arc::new(ASAPQueryEngine::new(streaming_config, 15_000)), Arc::new(crate::storage_engines::sketch_db::index::SketchStore::new()), ) + .with_active_physical_plan(active) .with_remote_write(receiver.clone()); (server.start_test_server().await.unwrap(), receiver) } @@ -2284,6 +2345,12 @@ mod tests { .await .unwrap(); assert_eq!(accepted.status().as_u16(), StatusCode::NO_CONTENT.as_u16()); + let ready = client + .get(format!("http://127.0.0.1:{port}/api/v1/health")) + .send() + .await + .unwrap(); + assert_eq!(ready.status().as_u16(), StatusCode::OK.as_u16()); assert_eq!( receiver .stats() @@ -5303,14 +5370,44 @@ async fn handle_prometheus_remote_write( error @ (crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::Backpressure( _, ) - | crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::DedupCapacity(_)), + | crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::DedupCapacity(_) + | crate::drivers::ingest::prometheus_remote_write::RemoteWriteError::InactivePhysicalPlan), ) => (StatusCode::SERVICE_UNAVAILABLE, error.to_string()).into_response(), Err(error) => (StatusCode::BAD_REQUEST, error.to_string()).into_response(), } } -async fn handle_health() -> &'static str { - "ok" +async fn handle_health(State(state): State) -> axum::response::Response { + use axum::response::IntoResponse; + + if state.remote_write.is_some() { + let Some(active) = state + .active_physical_plan + .as_ref() + .map(|handle| handle.snapshot()) + else { + return (StatusCode::SERVICE_UNAVAILABLE, "no active PhysicalPlan").into_response(); + }; + let now = unix_time_ms(); + let lifecycle_ready = active.backend_plan.plan_id != 0 + && active.backend_plan.activation_unix_ms <= now + && active + .backend_plan + .expiry_unix_ms + .is_none_or(|expiry| now < expiry); + let ingest_ready = matches!( + active.precompute_plan.ingest.protocol, + control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1 + ) && active.precompute_plan.ingest.endpoint_path == "/api/v1/write"; + if !lifecycle_ready || !ingest_ready { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "active PhysicalPlan is not ready for Remote Write", + ) + .into_response(); + } + } + (StatusCode::OK, "ok").into_response() } /// Return list of metrics currently in the store. @@ -5520,15 +5617,85 @@ async fn handle_post_backend_plan( (StatusCode::OK, axum::Json(body)).into_response() } -#[derive(serde::Deserialize)] -struct PhysicalPlanInstallRequest { - precompute_plan: control_plane::physical::compiler::PrecomputePlan, - transmission_plan: control_plane::physical::compiler::TransmissionPlan, - backend_plan: Vec, - query_plan: control_plane::query_plan::QueryPlan, - storage_routing: Option, +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PhysicalPlanInstallRequest { + pub precompute_plan: control_plane::physical::compiler::PrecomputePlan, + pub transmission_plan: control_plane::physical::compiler::TransmissionPlan, + pub backend_plan: Vec, + pub query_plan: control_plane::query_plan::QueryPlan, + pub storage_routing: Option, #[serde(default)] - adaptation_evidence: Vec, + pub adaptation_evidence: Vec, +} + +/// Decode and cross-validate every backend view before it can become visible. +/// Used by both startup artifact loading and the staged HTTP install path. +pub fn build_active_physical_plan( + request: PhysicalPlanInstallRequest, + default_routing: Arc, +) -> Result { + use std::collections::BTreeSet; + let runtime_materializations = request + .precompute_plan + .runtime_materializations() + .map_err(|error| format!("PrecomputePlan validation error: {error}"))?; + request + .transmission_plan + .validate(&request.precompute_plan) + .map_err(|error| format!("TransmissionPlan validation error: {error}"))?; + let backend_plan = control_plane::backend_plan::BackendPlan::decode(&request.backend_plan) + .map_err(|error| format!("BackendPlan decode error: {error}"))?; + backend_plan + .validate() + .map_err(|error| format!("BackendPlan validation error: {error}"))?; + request + .precompute_plan + .validate_against_backend(&backend_plan) + .map_err(|error| format!("PrecomputePlan validation error: {error}"))?; + let envelope = &request.precompute_plan.envelope; + if request.query_plan.plan_id != backend_plan.plan_id + || request.query_plan.plan_version != backend_plan.plan_version + || envelope.plan_id != backend_plan.plan_id + || envelope.plan_version != backend_plan.plan_version + || envelope.activation_unix_ms != backend_plan.activation_unix_ms + || envelope.expiry_unix_ms != backend_plan.expiry_unix_ms + || envelope.backend_compat != backend_plan.backend_compat + || request.transmission_plan.envelope != *envelope + { + return Err("physical subplans have different plan identity/version".into()); + } + let runtime_config = + crate::storage_engines::types::StreamingConfig::new(runtime_materializations); + let config_fps: BTreeSet = runtime_config.aggregation_configs.keys().copied().collect(); + let plan_fps: BTreeSet = backend_plan + .materializations + .keys() + .map(|fp| fp.0) + .collect(); + if config_fps != plan_fps { + return Err("PrecomputePlan and BackendPlan materialization fingerprints differ".into()); + } + let typed_fps: BTreeSet<_> = backend_plan.materializations.keys().copied().collect(); + request + .query_plan + .validate(&typed_fps) + .map_err(|error| format!("QueryPlan validation error: {error}"))?; + let storage_routing = match request.storage_routing.as_ref() { + Some(value) => Arc::new( + crate::storage_engines::types::BackendStorageRouting::from_json_payload(value) + .map_err(|error| format!("BackendStorageRouting build error: {error:#}"))?, + ), + None => default_routing, + }; + Ok(crate::storage_engines::types::ActivePhysicalPlan { + precompute_plan: request.precompute_plan, + transmission_plan: request.transmission_plan, + runtime_config: Arc::new(runtime_config), + backend_plan: Arc::new(backend_plan), + query_plan: Arc::new(request.query_plan), + storage_routing, + }) } /// Validate and stage all backend views. Staging never changes query routing; @@ -5539,7 +5706,6 @@ async fn handle_post_physical_plan( ) -> axum::response::Response { use axum::http::StatusCode; use axum::response::IntoResponse; - use std::collections::BTreeSet; let Some(active_handle) = state.active_physical_plan.as_ref() else { return ( @@ -5559,26 +5725,6 @@ async fn handle_post_physical_plan( ) .into_response(); }; - let runtime_materializations = - match request.precompute_plan.runtime_materializations() { - Ok(materializations) => materializations, - Err(error) => return ( - StatusCode::UNPROCESSABLE_ENTITY, - axum::Json(serde_json::json!({ - "status": "error", "error": format!("PrecomputePlan validation error: {error}") - })), - ) - .into_response(), - }; - if let Err(error) = request.transmission_plan.validate(&request.precompute_plan) { - return ( - StatusCode::UNPROCESSABLE_ENTITY, - axum::Json(serde_json::json!({ - "status": "error", "error": format!("TransmissionPlan validation error: {error}") - })), - ) - .into_response(); - } let current = active_handle.snapshot(); if current.transmission_plan.envelope.plan_id != 0 { if let Err(error) = current.transmission_plan.authorize_successor( @@ -5596,120 +5742,36 @@ async fn handle_post_physical_plan( .into_response(); } } - let new_config = crate::storage_engines::types::StreamingConfig::new(runtime_materializations); - let new_plan = match control_plane::backend_plan::BackendPlan::decode(&request.backend_plan) { - Ok(plan) => plan, + let _guard = state.physical_plan_lock.lock().await; + let active = match build_active_physical_plan(request, current.storage_routing.clone()) { + Ok(active) => active, Err(error) => { return ( - StatusCode::BAD_REQUEST, - axum::Json(serde_json::json!({ - "status": "error", "error": format!("BackendPlan decode error: {error}") - })), + StatusCode::UNPROCESSABLE_ENTITY, + axum::Json(serde_json::json!({"status": "error", "error": error})), ) - .into_response() + .into_response(); } }; - if let Err(error) = new_plan.validate() { - return ( - StatusCode::UNPROCESSABLE_ENTITY, - axum::Json(serde_json::json!({ - "status": "error", "error": format!("BackendPlan validation error: {error}") - })), - ) - .into_response(); - } - if let Err(error) = request.precompute_plan.validate_against_backend(&new_plan) { - return ( - StatusCode::UNPROCESSABLE_ENTITY, - axum::Json(serde_json::json!({ - "status": "error", "error": format!("PrecomputePlan validation error: {error}") - })), - ) - .into_response(); - } - if request.query_plan.plan_id != new_plan.plan_id - || request.query_plan.plan_version != new_plan.plan_version - || request.precompute_plan.envelope.plan_id != new_plan.plan_id - || request.precompute_plan.envelope.plan_version != new_plan.plan_version - || request.precompute_plan.envelope.activation_unix_ms != new_plan.activation_unix_ms - || request.precompute_plan.envelope.expiry_unix_ms != new_plan.expiry_unix_ms - || request.precompute_plan.envelope.backend_compat != new_plan.backend_compat + if state.remote_write.is_some() + && (active.backend_plan.plan_id == 0 + || !matches!( + active.precompute_plan.ingest.protocol, + control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1 + ) + || active.precompute_plan.ingest.endpoint_path != "/api/v1/write") { - return ( - StatusCode::UNPROCESSABLE_ENTITY, - axum::Json(serde_json::json!({ - "status": "error", "error": "physical subplans have different plan identity/version" - })), - ) - .into_response(); - } - let config_fps: BTreeSet = new_config.aggregation_configs.keys().copied().collect(); - let plan_fps: BTreeSet = new_plan.materializations.keys().map(|fp| fp.0).collect(); - if config_fps != plan_fps { return ( StatusCode::UNPROCESSABLE_ENTITY, axum::Json(serde_json::json!({ "status": "error", - "error": "streaming-config and BackendPlan materialization fingerprints differ" + "error": "Remote Write listener requires a non-bootstrap prometheus_remote_write_v1 plan at /api/v1/write" })), ) .into_response(); } - for schema in &request.precompute_plan.schemas { - let Some(materialization) = new_plan.materializations.get(&schema.materialization) else { - return ( - StatusCode::UNPROCESSABLE_ENTITY, - axum::Json(serde_json::json!({ - "status": "error", "error": "PrecomputePlan schema is absent from BackendPlan" - })), - ) - .into_response(); - }; - if control_plane::physical::compiler::StateFamilyContract::try_from(&materialization.family) - != Ok(schema.family.clone()) - { - return ( - StatusCode::UNPROCESSABLE_ENTITY, - axum::Json(serde_json::json!({ - "status": "error", "error": "PrecomputePlan and BackendPlan schema semantics differ" - })), - ) - .into_response(); - } - } - let typed_plan_fps: BTreeSet<_> = new_plan.materializations.keys().copied().collect(); - if let Err(error) = request.query_plan.validate(&typed_plan_fps) { - return ( - StatusCode::UNPROCESSABLE_ENTITY, - axum::Json(serde_json::json!({ - "status": "error", "error": format!("QueryPlan validation error: {error}") - })), - ) - .into_response(); - } - let new_routing = match request.storage_routing.as_ref() { - Some(value) => match crate::storage_engines::types::BackendStorageRouting::from_json_payload(value) { - Ok(routing) => Some(routing), - Err(error) => return (StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({ - "status": "error", "error": format!("BackendStorageRouting build error: {error:#}") - }))).into_response(), - }, - None => None, - }; - let new_routing = new_routing - .map(Arc::new) - .unwrap_or_else(|| active_handle.snapshot().storage_routing.clone()); - - let _guard = state.physical_plan_lock.lock().await; - let plan_id = new_plan.plan_id; - let active = crate::storage_engines::types::ActivePhysicalPlan { - precompute_plan: request.precompute_plan, - transmission_plan: request.transmission_plan, - runtime_config: Arc::new(new_config), - backend_plan: Arc::new(new_plan), - query_plan: Arc::new(request.query_plan), - storage_routing: new_routing, - }; + let plan_id = active.backend_plan.plan_id; + let materialization_count = active.precompute_plan.materializations.len(); let plan_version = active.backend_plan.plan_version; let now = unix_time_ms(); if let Err(error) = lifecycle.stage(active, now) { @@ -5723,7 +5785,7 @@ async fn handle_post_physical_plan( StatusCode::ACCEPTED, axum::Json(serde_json::json!({ "status": "staged", "plan_id": plan_id, "plan_version": plan_version, - "materialization_count": plan_fps.len() + "materialization_count": materialization_count })), ) .into_response() diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index e534b7dc2..792c12613 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -34,6 +34,13 @@ enum RuntimeProfile { Asapquery, } +fn unix_time_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_millis() as u64) + .unwrap_or(0) +} + #[derive(Parser, Debug)] #[command(author, version, about, long_about = None)] struct Args { @@ -42,9 +49,14 @@ struct Args { #[arg(long, value_enum, default_value = "distributed")] profile: RuntimeProfile, - /// File path for streaming_config + /// Legacy bootstrap streaming config (distributed profile only). + #[arg(long)] + streaming_config: Option, + + /// JSON physical-plan artifact. Required by the backend-local profile; + /// all runtime/query views are validated and installed as one snapshot. #[arg(long)] - streaming_config: String, + physical_plan: Option, /// Cleanup policy for SketchStore retention. /// `circular_buffer`: keep the N most recent windows per agg @@ -330,8 +342,17 @@ struct Args { fn validate_profile(args: &Args) -> Result<()> { if args.profile != RuntimeProfile::Asapquery { + if args.streaming_config.is_none() { + return Err("the distributed profile requires --streaming-config".into()); + } return Ok(()); } + if args.streaming_config.is_some() { + return Err("--profile asapquery rejects --streaming-config; use --physical-plan".into()); + } + if args.physical_plan.is_none() { + return Err("--profile asapquery requires --physical-plan".into()); + } let mut excluded = Vec::new(); if args.enable_otel_ingest { excluded.push("--enable-otel-ingest"); @@ -436,7 +457,57 @@ async fn main() -> Result<()> { ); } - let streaming_config = Arc::new(read_streaming_config(&args.streaming_config)?); + let startup_physical_plan = if let Some(path) = args.physical_plan.as_ref() { + let bytes = fs::read(path)?; + let artifact: data_plane::drivers::query::servers::http::PhysicalPlanInstallRequest = + serde_json::from_slice(&bytes).map_err(|error| { + format!( + "failed to decode physical plan artifact {}: {error}", + path.display() + ) + })?; + let active = data_plane::drivers::query::servers::http::build_active_physical_plan( + artifact, + Arc::new(data_plane::storage_engines::types::BackendStorageRouting::empty()), + ) + .map_err(|error| format!("invalid physical plan artifact {}: {error}", path.display()))?; + if args.profile == RuntimeProfile::Asapquery + && (active.backend_plan.plan_id == 0 + || !matches!( + active.precompute_plan.ingest.protocol, + control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1 + ) + || active.precompute_plan.ingest.endpoint_path != "/api/v1/write") + { + return Err("the asapquery profile requires a non-bootstrap physical plan declaring prometheus_remote_write_v1 at /api/v1/write".into()); + } + let now = unix_time_ms(); + if active.backend_plan.activation_unix_ms > now { + return Err(format!( + "physical plan activation {} is later than startup time {now}", + active.backend_plan.activation_unix_ms + ) + .into()); + } + if active + .backend_plan + .expiry_unix_ms + .is_some_and(|expiry| expiry <= now) + { + return Err("physical plan artifact is expired".into()); + } + Some(active) + } else { + None + }; + let streaming_config = match startup_physical_plan.as_ref() { + Some(active) => active.runtime_config.clone(), + None => Arc::new(read_streaming_config( + args.streaming_config + .as_deref() + .expect("validated distributed streaming config"), + )?), + }; info!( "Loaded streaming config with {} entries", streaming_config.get_all_aggregation_configs().len() @@ -594,7 +665,7 @@ async fn main() -> Result<()> { }, rules: Vec::new(), }; - let active_physical_plan = data_plane::storage_engines::types::HotReloadActivePhysicalPlan::new( + let initial_active_plan = startup_physical_plan.unwrap_or_else(|| { data_plane::storage_engines::types::ActivePhysicalPlan { precompute_plan: initial_precompute_plan, transmission_plan: initial_transmission_plan, @@ -604,8 +675,10 @@ async fn main() -> Result<()> { storage_routing: Arc::new( data_plane::storage_engines::types::BackendStorageRouting::empty(), ), - }, - ); + } + }); + let active_physical_plan = + data_plane::storage_engines::types::HotReloadActivePhysicalPlan::new(initial_active_plan); let hot_reload_config = data_plane::storage_engines::types::HotReloadStreamingConfig::from_active( active_physical_plan.clone(), @@ -914,10 +987,16 @@ async fn main() -> Result<()> { // lifecycle transitions at the sid level via the shared // `SketchStore` (already passed in below). let mut server = HttpServer::new(http_config, engine, sketch_index.clone()) - .with_hot_reload_config(hot_reload_config.clone()) - .with_hot_reload_backend_plan(hot_reload_backend_plan.clone()) .with_active_physical_plan(active_physical_plan.clone()) .with_probe_cache(probe_cache.clone()); + if args.profile == RuntimeProfile::Distributed { + // Legacy partial-document endpoints remain available to distributed + // deployments. The compatibility profile deliberately exposes only + // the atomic PhysicalPlan stage/activate lifecycle. + server = server + .with_hot_reload_config(hot_reload_config.clone()) + .with_hot_reload_backend_plan(hot_reload_backend_plan.clone()); + } if args.enable_remote_write || args.profile == RuntimeProfile::Asapquery { let receiver = PrometheusRemoteWriteReceiver::new( @@ -1338,8 +1417,40 @@ fn setup_logging( #[cfg(test)] mod tests { + use super::{validate_profile, Args}; + use clap::Parser; use data_plane::drivers::AdapterConfig; + #[test] + fn asapquery_requires_atomic_physical_plan_not_streaming_config() { + let valid = Args::try_parse_from([ + "data_plane", + "--profile", + "asapquery", + "--physical-plan", + "plan.json", + "--forward-unsupported-queries", + ]) + .unwrap(); + assert!(validate_profile(&valid).is_ok()); + + let legacy = Args::try_parse_from([ + "data_plane", + "--profile", + "asapquery", + "--streaming-config", + "streaming.yaml", + "--physical-plan", + "plan.json", + "--forward-unsupported-queries", + ]) + .unwrap(); + assert!(validate_profile(&legacy) + .unwrap_err() + .to_string() + .contains("rejects --streaming-config")); + } + // Step-1 of the JSONL deprecation refactor deleted the // ยง5.2 `ColdFallback` adapter and the // `from_prom_with_optional_cold` constructor. The surviving diff --git a/docs/user_guide/asapquery-profile.md b/docs/user_guide/asapquery-profile.md index e8adf5076..1ded9be78 100644 --- a/docs/user_guide/asapquery-profile.md +++ b/docs/user_guide/asapquery-profile.md @@ -17,7 +17,7 @@ Then start the backend: ```bash cargo run -p data_plane -- \ --profile asapquery \ - --streaming-config data_plane/examples/asapquery/streaming-config.yaml \ + --physical-plan /etc/asapquery/physical-plan.json \ --prometheus-server http://127.0.0.1:9090 \ --forward-unsupported-queries \ --http-port 9091 \ @@ -26,7 +26,8 @@ cargo run -p data_plane -- \ The profile checks Prometheus's `/-/healthy` endpoint before opening its public listener. It rejects OTLP ingest, monitor coordination, persistence, backfill, -schema eviction, and archive routing flags. Remote Write and PromQL share the +schema eviction, archive routing flags, and the legacy `--streaming-config` +bootstrap. Remote Write and PromQL share the backend listener: - `POST /api/v1/write` @@ -53,7 +54,24 @@ Receiver evidence is exported from `/metrics` as `asap_remote_write_rejected_requests_total`, and `asap_remote_write_bytes_total`. -The example streaming config is a checked-in precompute-plan fixture. Startup -loading of canonical `QueryWorkload` and `DataWorkload` snapshots and automatic -backend-only compilation remain required before the profile meets the complete -compatibility contract in the design document. +`physical-plan.json` is the JSON representation accepted by +`POST /api/v1/physical-plan`: a `PrecomputePlan`, `TransmissionPlan`, encoded +`BackendPlan`, and authoritative `QueryPlan` DAG with one shared plan identity +and version. For this profile, the precompute ingest contract must be +`prometheus_remote_write_v1` at `/api/v1/write`; raw writes are rejected if a +different plan is active. Startup validates all fingerprints, schemas, query +bindings, lifecycle fields, and transmission rules before constructing the +single active snapshot. Runtime replacement uses `POST /api/v1/physical-plan` +to stage the complete successor and `POST /api/v1/physical-plan/activate` for +the atomic cutover. The partial `/streaming-config` and `/backend-plan` install +handles are not attached in this profile. + +Query serving uses only the installed `QueryPlan` DAG. A query absent from that +DAG is a capability miss and goes to the exact Prometheus fallback; the backend +does not search materialization candidates while serving. + +Automatic startup compilation from canonical `QueryWorkload` and +`DataWorkload` snapshots is not part of this phase. Until the workload types +have a stable serialized contract and the physical compiler supports a +backend-only deployment target, the artifact must be produced offline by the +planner/control-plane pipeline.