diff --git a/control_plane/src/backend_client.rs b/control_plane/src/backend_client.rs index dc356a00..4329695f 100644 --- a/control_plane/src/backend_client.rs +++ b/control_plane/src/backend_client.rs @@ -268,12 +268,16 @@ impl BackendClient { } /// Publish one authoritative catalog generation and all plans that reference it. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(plan_id = publication.transmission_plan.envelope.plan_id, + plan_version = publication.transmission_plan.envelope.plan_version))] pub async fn post_catalog_plan_typed( &self, publication: &crate::physical::publication::PhysicalPlanPublication, storage_routing: Option, adaptation_evidence: &[crate::physical::compiler::RuntimeAdaptationEvidence], ) -> std::result::Result<(), BackendPostError> { + tracing::debug!(target: "asap_runtime_debug", "backend plan stage request started"); let body = publication .install_request(storage_routing, adaptation_evidence.to_vec()) .map_err(|error| BackendPostError::Permanent(anyhow::anyhow!(error)))?; @@ -285,6 +289,8 @@ impl BackendClient { .await .map_err(classify_reqwest_error)?; let status = response.status(); + tracing::debug!(target: "asap_runtime_debug", http_status = %status, + "backend plan stage response received"); if status.is_success() { Ok(()) } else { @@ -297,11 +303,18 @@ impl BackendClient { } } + #[tracing::instrument( + level = "debug", + target = "asap_runtime_debug", + skip_all, + fields(plan_id, plan_version) + )] pub async fn discard_staged_physical_plan( &self, plan_id: u64, plan_version: u64, ) -> std::result::Result<(), BackendPostError> { + tracing::debug!(target: "asap_runtime_debug", "staged backend plan cleanup requested"); let response = self .http .post(format!( @@ -324,11 +337,18 @@ impl BackendClient { } } + #[tracing::instrument( + level = "debug", + target = "asap_runtime_debug", + skip_all, + fields(plan_id, plan_version) + )] pub async fn activate_physical_plan( &self, plan_id: u64, plan_version: u64, ) -> std::result::Result<(), BackendPostError> { + tracing::debug!(target: "asap_runtime_debug", "backend plan activation request started"); let url = format!("{}/activate", derive_physical_plan_url(&self.endpoint)); let response = self .http @@ -341,6 +361,8 @@ impl BackendClient { .await .map_err(classify_reqwest_error)?; let status = response.status(); + tracing::debug!(target: "asap_runtime_debug", http_status = %status, + "backend plan activation response received"); if status.is_success() { Ok(()) } else { diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 8464ec17..51b584e4 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -18,10 +18,13 @@ use axum::{ use serde::{Deserialize, Serialize}; use serde_json::json; use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::Duration; use tracing::info; +static NEXT_PLAN_CALL_ID: AtomicU64 = AtomicU64::new(1); + use opamp::OpampServer; use physical::deployment_cost::online as online_cost_model; use physical::deployment_cost::online::{init_store as init_online_store, OnlineMetricsStore}; @@ -50,7 +53,16 @@ struct AppState { #[tokio::main] async fn main() { - tracing_subscriber::fmt::init(); + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with_file(true) + .with_line_number(true) + .with_target(true) + .with_writer(std::io::stdout) + .init(); let api_addr = std::env::var("CONTROLLER_ADDR").unwrap_or_else(|_| "0.0.0.0:8080".into()); let opamp_addr = @@ -271,11 +283,16 @@ async fn compile_and_publish_physical_plan( mut request: CompileAndPublishPhysicalPlanRequest, frontend: QueryFrontend, ) -> Response { + let call_id = NEXT_PLAN_CALL_ID.fetch_add(1, Ordering::Relaxed); + let started = std::time::Instant::now(); + tracing::debug!(target: "asap_runtime_debug", call_id, ?frontend, + "physical plan compilation requested"); // Serialize typed activations so an older response cannot overwrite the // catalog recorded after a newer backend activation. let mut active_catalog = st.active_summary_catalog.lock().await; if let Some(erp) = &mut request.erp { if let Err(error) = erp.hydrate_observed_shape(&st.runtime_samples) { + tracing::warn!(call_id, %error, "ERP observation hydration failed"); return (StatusCode::UNPROCESSABLE_ENTITY, error).into_response(); } let catalog = active_catalog.clone(); @@ -287,16 +304,37 @@ async fn compile_and_publish_physical_plan( (bundle, ids, timeout, adaptation, manifests) } Ok((None, ..)) => { + tracing::error!( + call_id, + "physical plan compilation produced no deployable plan" + ); return ( StatusCode::INTERNAL_SERVER_ERROR, "publication requires a selected plan", ) - .into_response() + .into_response(); + } + Err(response) => { + tracing::warn!(call_id, status = %response.0, error = %response.1, + "physical plan compilation failed"); + return physical_compile_failure(response); } - Err(response) => return physical_compile_failure(response), }; + info!( + call_id, + plan_id = bundle.envelope.plan_id, + plan_version = bundle.envelope.plan_version, + collector_count = bundle.collector_plans.len(), + "physical plan compiled" + ); let Some(backend) = st.backend_client.as_ref() else { + tracing::warn!( + call_id, + plan_id = bundle.envelope.plan_id, + plan_version = bundle.envelope.plan_version, + "backend endpoint is not configured" + ); return ( StatusCode::SERVICE_UNAVAILABLE, "CONTROLLER_BACKEND_ENDPOINT is required for physical-plan publication".to_string(), @@ -308,6 +346,8 @@ async fn compile_and_publish_physical_plan( .ensure_collector_plan_targets(&bundle.collector_plans, apply_timeout) .await { + tracing::warn!(call_id, plan_id = bundle.envelope.plan_id, plan_version = bundle.envelope.plan_version, + %error, "collector plan preflight failed"); return ( StatusCode::BAD_GATEWAY, format!("collector physical-plan preflight failed: {error}"), @@ -317,11 +357,14 @@ async fn compile_and_publish_physical_plan( let publication = match bundle.to_publication_artifact() { Ok(publication) => publication, Err(error) => { + tracing::error!(call_id, plan_id = bundle.envelope.plan_id, + plan_version = bundle.envelope.plan_version, %error, + "catalog publication artifact failed"); return ( StatusCode::INTERNAL_SERVER_ERROR, format!("invalid catalog publication: {error}"), ) - .into_response() + .into_response(); } }; if let Err(error) = backend @@ -332,17 +375,27 @@ async fn compile_and_publish_physical_plan( ) .await { + tracing::warn!(call_id, plan_id = bundle.envelope.plan_id, plan_version = bundle.envelope.plan_version, + %error, "backend plan staging failed"); return ( StatusCode::BAD_GATEWAY, format!("backend rejected physical plan: {error}"), ) .into_response(); } + info!( + call_id, + plan_id = bundle.envelope.plan_id, + plan_version = bundle.envelope.plan_version, + "backend physical plan staged" + ); if let Err(error) = st .opamp .publish_collector_plans(&bundle.collector_plans, apply_timeout) .await { + tracing::warn!(call_id, plan_id = bundle.envelope.plan_id, plan_version = bundle.envelope.plan_version, + %error, "collector plan publication failed"); let cleanup = backend .discard_staged_physical_plan(bundle.envelope.plan_id, bundle.envelope.plan_version) .await; @@ -352,12 +405,26 @@ async fn compile_and_publish_physical_plan( ) .into_response(); } + info!( + call_id, + plan_id = bundle.envelope.plan_id, + plan_version = bundle.envelope.plan_version, + "collector plans published" + ); let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_millis() as u64; let activation_wait = bundle.envelope.activation_unix_ms.saturating_sub(now); if activation_wait > apply_timeout.as_millis() as u64 { + tracing::warn!( + call_id, + plan_id = bundle.envelope.plan_id, + plan_version = bundle.envelope.plan_version, + activation_wait_ms = activation_wait, + timeout_ms = apply_timeout.as_millis() as u64, + "scheduled activation exceeds apply timeout; backend plan remains staged" + ); return ( StatusCode::GATEWAY_TIMEOUT, "activation time exceeds apply_timeout_ms; backend remains staged".to_string(), @@ -371,6 +438,8 @@ async fn compile_and_publish_physical_plan( .activate_physical_plan(bundle.envelope.plan_id, bundle.envelope.plan_version) .await { + tracing::warn!(call_id, plan_id = bundle.envelope.plan_id, plan_version = bundle.envelope.plan_version, + %error, "backend plan activation failed"); return ( StatusCode::BAD_GATEWAY, format!("backend physical-plan activation failed: {error}"), @@ -379,6 +448,13 @@ async fn compile_and_publish_physical_plan( } *active_catalog = Some(Arc::new(bundle.summary_catalog)); + info!( + call_id, + plan_id = bundle.envelope.plan_id, + plan_version = bundle.envelope.plan_version, + elapsed_ms = started.elapsed().as_millis() as u64, + "physical plan active" + ); Json(CompileAndPublishPhysicalPlanResponse { cost_comparison: bundle.cost_comparison, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 1abb2016..5460c99b 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -532,11 +532,15 @@ pub fn gos_policy_from_accuracy_budget( }) } +#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(plan_id = envelope.plan_id, plan_version = envelope.plan_version, + producer_count = precompute.producers.len()))] pub fn build_transmission_plan( envelope: PlanEnvelope, precompute: &PrecomputePlan, runtime_policies: &BTreeMap, ) -> Result { + tracing::debug!(target: "asap_runtime_debug", "transmission plan construction started"); if envelope != precompute.envelope { return Err(TransmissionPlanError::EnvelopeMismatch); } @@ -1000,6 +1004,9 @@ impl BackendLocalPlanningInput { let asap_aware_mapping::Replacement::Summary(root) = candidate.replacement else { continue; }; + let _physical = tracing::debug_span!(target: "asap_runtime_debug", "physical_candidate_compile", + stage = "planner.physical_candidate", query_id = %query.query_id, + input_kind = "bound_promql_vector").entered(); let root = asap_aware_mapping::replacement::finalize_query_candidate(root, &typed) .map_err(|error| CompileError::Snapshot(error.to_string()))?; let compiled = asap_physical_operators::physical_planner::promql_rows::compile_current_series_readout(&root) @@ -1310,12 +1317,16 @@ impl DeploymentPlanCompiler { self.compile_for_frontend(request, environment, QueryFrontend::MetricsQl) } + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(frontend = ?frontend, plan_version = environment.plan_version, + query_count = request.queries.len()))] pub fn compile_for_frontend( &self, mut request: PhysicalCompilationRequest, environment: PhysicalDeploymentContext, frontend: QueryFrontend, ) -> Result { + tracing::debug!(target: "asap_runtime_debug", stage = "deployment.bind", "backend deployment compiler entered"); environment .dataset_identity .validate() @@ -2730,6 +2741,8 @@ pub fn select_logical_roots_with_error_resource_profiles( select_logical_roots_with_trace(queries, roots, evidence, exact_costs, erp).map(|_| ()) } +#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(query_count = queries.len(), root_count = roots.len()))] pub fn select_logical_roots_with_trace( queries: &mut [QueryCompilationInput], roots: Vec>, @@ -2779,6 +2792,7 @@ fn logical_roots_and_candidates( now_ms: u64, mut candidates_out: Option<&mut Vec)>>>, ) -> Result, CompileError> { + tracing::debug!(target: "asap_runtime_debug", stage = "planner.select", "Planner selection entered"); let mut traces = Vec::new(); if roots.len() != queries.len() { return Err(CompileError::Snapshot( diff --git a/control_plane/src/physical/maintained_population.rs b/control_plane/src/physical/maintained_population.rs index 7844c566..00c07412 100644 --- a/control_plane/src/physical/maintained_population.rs +++ b/control_plane/src/physical/maintained_population.rs @@ -159,6 +159,8 @@ pub(super) fn operator( /// The maintained population is a deployment source; ranking is compiled by /// Planner before this candidate is priced or installed. +#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.compile_install", query_id = %entry.query_id, input_kind = "current_series_snapshot"), err)] pub(super) fn install_native_topk( entry: &mut asap_types::query_plan::QueryPlanEntry, selected: &std::rc::Rc, diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index eea4d353..521e21ae 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/control_plane/src/physical/workload_cost.rs @@ -795,6 +795,17 @@ pub fn select_candidates( } } } + tracing::debug!(target: "asap_runtime_debug", stage = "deployment.candidate_inventory", + candidate_count = candidate_evaluations.len(), + search_coverage = ?materialization_search_coverage, + "bounded candidate evaluation completed; absence is not a cost rejection"); + for candidate in &candidate_evaluations { + tracing::debug!(target: "asap_runtime_debug", stage = "deployment.candidate_evaluation", + candidate_id = ?candidate.candidate_id, + physical_candidate_id = ?candidate.physical_candidate_id, + status = ?candidate.status, total_cost = candidate.total_cost, + reason = ?candidate.unavailable_reason, "candidate evaluated"); + } let selected = asap_physical_operators::physical_planner::select_candidate( priced_candidates, |(cost, _, manifest, _, _)| { @@ -817,6 +828,11 @@ pub fn select_candidates( })?; let (_, mut plan, selected_manifest, component_costs, best_index) = selected.candidate; candidate_evaluations[best_index].status = CandidateEvaluationStatus::Selected; + tracing::debug!(target: "asap_runtime_debug", stage = "deployment.candidate_selected", + plan_id = plan.envelope.plan_id, plan_version = plan.envelope.plan_version, + candidate_id = ?candidate_evaluations[best_index].candidate_id, + total_cost = candidate_evaluations[best_index].total_cost, + "lowest quoted cost selected within the admitted inventory"); plan.cost_comparison = Some(CandidatePlanSelectionReport { planner_selection_trace, materialization_search_coverage, diff --git a/crates/asap_types/src/query_plan.rs b/crates/asap_types/src/query_plan.rs index 1e214b67..8924aaa8 100644 --- a/crates/asap_types/src/query_plan.rs +++ b/crates/asap_types/src/query_plan.rs @@ -760,6 +760,152 @@ pub enum QueryPlanNode { } impl QueryPlanNode { + /// Operator label for logs: the serialized `op` tag, plus the residual + /// `kind` for logical nodes, including the operation where applicable + /// (e.g. `logical/aggregate/sum`). + pub fn op_label(&self) -> &'static str { + use residual::ResidualQueryOperator as R; + match self { + Self::RelationalJoin { .. } => "relational_join", + Self::Relational { .. } => "relational", + Self::Logical { operator, .. } => match operator { + R::CurrentSeries { .. } => "logical/current_series", + R::ExactSubquery { .. } => "logical/exact_subquery", + R::CandidateExactSubquery { .. } => "logical/candidate_exact_subquery", + R::Scan { .. } => "logical/scan", + R::UnaryNegate => "logical/unary_negate", + R::VectorToScalar => "logical/vector_to_scalar", + R::Aggregate { operation, .. } => match operation { + residual::Aggregation::Sum => "logical/aggregate/sum", + residual::Aggregation::Max => "logical/aggregate/max", + residual::Aggregation::Min => "logical/aggregate/min", + residual::Aggregation::Avg => "logical/aggregate/avg", + residual::Aggregation::Count => "logical/aggregate/count", + }, + R::Limit { .. } => "logical/limit", + R::Binary { .. } => "logical/binary", + R::Temporal { .. } => "logical/temporal", + R::Sort { .. } => "logical/sort", + R::HistogramQuantile => "logical/histogram_quantile", + R::Subquery { .. } => "logical/subquery", + }, + Self::Physical { .. } => "physical_dag", + Self::PhysicalFragment { .. } => "physical_fragment", + Self::Scalar { .. } => "scalar", + Self::Binary { .. } => "binary", + Self::ReduceSum { .. } => "reduce_sum", + Self::ReadMaterialization { .. } => "read_materialization", + Self::SummaryEstimate { .. } => "summary_estimate", + Self::ExactReadout { readout, .. } => match readout { + ExactReadout::Sum => "exact_readout/sum", + ExactReadout::Count => "exact_readout/count", + ExactReadout::Increase => "exact_readout/increase", + ExactReadout::Rate => "exact_readout/rate", + ExactReadout::Min => "exact_readout/min", + ExactReadout::Max => "exact_readout/max", + }, + Self::SummaryMerge { .. } => "summary_merge", + Self::ExternalExact { .. } => "external_exact", + Self::ExactFallback { .. } => "exact_fallback", + } + } + + /// Bounded, query-text-free operator arguments for execution logs. + /// The query ID links these details to the full installed plan when needed. + pub fn log_syntax(&self) -> String { + use residual::ResidualQueryOperator as R; + match self { + Self::RelationalJoin { join_kind, .. } => format!("join_kind={join_kind:?}"), + Self::Relational { .. } => String::new(), + Self::Logical { operator, .. } => match operator { + R::CurrentSeries { readout, .. } => format!("readout={readout:?}"), + R::ExactSubquery { .. } => String::new(), + R::CandidateExactSubquery { .. } => String::new(), + R::Scan { + metric, + matchers, + range_ms, + offset_ms, + } => format!( + "metric={} matcher_count={} range_ms={range_ms:?} offset_ms={offset_ms}", + metric + .as_deref() + .map(|name| name.chars().take(64).collect::()) + .unwrap_or_default(), + matchers.len(), + ), + R::UnaryNegate | R::VectorToScalar | R::HistogramQuantile => String::new(), + R::Aggregate { + operation, + grouping, + } => format!( + "operation={operation:?} grouping={}", + log_grouping(&grouping.labels, grouping.without) + ), + R::Limit { + n, + offset, + grouping, + } => format!( + "n={n} offset={offset} grouping={}", + log_grouping(&grouping.labels, grouping.without) + ), + R::Binary { + operation, + return_bool, + } => format!("operation={operation:?} return_bool={return_bool}"), + R::Temporal { operation } => format!("operation={operation:?}"), + R::Sort { + descending, + grouping, + } => format!( + "descending={descending} grouping={}", + log_grouping(&grouping.labels, grouping.without) + ), + R::Subquery { + range_ms, + step_ms, + offset_ms, + } => format!("range_ms={range_ms} step_ms={step_ms} offset_ms={offset_ms}"), + }, + Self::PhysicalFragment { + inputs, row_input, .. + } => format!("input_count={} row_input={row_input}", inputs.len()), + Self::Physical { + inputs, + source_nodes, + max_bytes, + } => format!( + "input_count={} source_count={} max_bytes={max_bytes}", + inputs.len(), + source_nodes.len() + ), + Self::Scalar { value } => format!("value={value}"), + Self::Binary { operator, .. } => format!("operation={operator:?}"), + Self::ReduceSum { grouping, .. } => match grouping { + PhysicalGrouping::PerEntity => "grouping=per_entity".into(), + PhysicalGrouping::Reduce(labels) => { + format!("grouping=reduce({})", log_labels(labels)) + } + }, + Self::ReadMaterialization { binding } => format!( + "window_ms={} lookback_ms={:?}", + binding.window_ms, binding.readout_lookback_ms + ), + Self::SummaryEstimate { query, .. } => match query { + QueryReadout::FrequencyL2 => "readout=frequency_l2".into(), + QueryReadout::FrequencyEntropy => "readout=frequency_entropy".into(), + QueryReadout::Quantile { q } => format!("readout=quantile q={q}"), + QueryReadout::PointCount { .. } => "readout=point_count".into(), + QueryReadout::Cardinality => "readout=cardinality".into(), + QueryReadout::TopK { k } => format!("readout=top_k k={k}"), + }, + Self::ExactReadout { readout, .. } => format!("readout={readout:?}"), + Self::SummaryMerge { .. } => String::new(), + Self::ExternalExact { .. } | Self::ExactFallback { .. } => String::new(), + } + } + pub fn inputs(&self) -> &[QueryNodeId] { match self { Self::Scalar { .. } | Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => { @@ -779,6 +925,26 @@ impl QueryPlanNode { } } +fn log_labels(labels: &[String]) -> String { + let mut names = labels + .iter() + .take(8) + .map(|label| label.chars().take(64).collect::()) + .collect::>(); + if labels.len() > 8 { + names.push("...".into()); + } + names.join(",") +} + +fn log_grouping(labels: &[String], without: bool) -> String { + format!( + "{}({})", + if without { "without" } else { "by" }, + log_labels(labels) + ) +} + pub use planner_types::post_asap::CandidateCompleteness; #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] @@ -872,6 +1038,8 @@ pub fn canonical_promql(query: &str) -> Result { #[cfg(test)] mod contract_tests { + use super::{residual, QueryPlanNode}; + // Installed plans cross producer/query threads without Planner Rc state. #[test] fn installed_query_contract_is_send_sync() { @@ -879,6 +1047,66 @@ mod contract_tests { assert_send_sync::(); assert_send_sync::(); } + + #[test] + fn aggregate_log_labels_identify_the_operation() { + for (operation, expected) in [ + (residual::Aggregation::Sum, "logical/aggregate/sum"), + (residual::Aggregation::Count, "logical/aggregate/count"), + (residual::Aggregation::Avg, "logical/aggregate/avg"), + ] { + let node = QueryPlanNode::Logical { + operator: residual::ResidualQueryOperator::Aggregate { + operation, + grouping: residual::Grouping { + labels: vec!["service".into()], + without: false, + }, + }, + inputs: vec![], + }; + assert_eq!(node.op_label(), expected); + assert!(node.log_syntax().contains("grouping=by(service)")); + } + for (readout, expected) in [ + (super::ExactReadout::Sum, "exact_readout/sum"), + (super::ExactReadout::Count, "exact_readout/count"), + ] { + assert_eq!( + QueryPlanNode::ExactReadout { + input: super::QueryNodeId(1), + readout, + } + .op_label(), + expected + ); + } + } + + #[test] + fn execution_log_syntax_identifies_operator_without_query_text() { + let binary = QueryPlanNode::Logical { + operator: residual::ResidualQueryOperator::Binary { + operation: residual::BinaryOperation::CheckedDiv, + return_bool: false, + }, + inputs: vec![super::QueryNodeId(1), super::QueryNodeId(2)], + }; + assert_eq!(binary.op_label(), "logical/binary"); + assert_eq!( + binary.log_syntax(), + "operation=CheckedDiv return_bool=false" + ); + + let exact = QueryPlanNode::Logical { + operator: residual::ResidualQueryOperator::ExactSubquery { + query: "secret_metric{credential=\"secret\"}".into(), + }, + inputs: vec![], + }; + assert_eq!(exact.op_label(), "logical/exact_subquery"); + assert!(exact.log_syntax().is_empty()); + } } /// Bind portable relation semantics before an installed plan can access its sources. diff --git a/crates/asap_types/src/query_plan/native.rs b/crates/asap_types/src/query_plan/native.rs index 06ec4029..d92aeea4 100644 --- a/crates/asap_types/src/query_plan/native.rs +++ b/crates/asap_types/src/query_plan/native.rs @@ -9,6 +9,8 @@ use std::sync::Arc; impl QueryPlanEntry { /// Invoke during plan compilation, after external/storage boundaries are fixed. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.compile_install", query_id = %self.query_id), err)] pub fn compile_relational_physical_dag(&mut self) -> Result<(), QueryPlanError> { let Some(expected) = self.relation_output_schema()? else { self.physical_dag = None; @@ -159,6 +161,8 @@ impl QueryPlanEntry { } /// Validate persisted computation and boundary schemas without logical lowering. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.recover_validate", query_id = %self.query_id), err)] pub fn recover_relational_physical_dag(&self) -> Result { let invalid = |message: String| QueryPlanError::Invalid(message); let value = self @@ -265,6 +269,8 @@ impl QueryPlanEntry { /// Recover an installed population readout; the source binds the complete /// maintained vector, while the physical program owns ranking and limiting. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.recover_validate", query_id = %self.query_id, input_kind = "current_series_snapshot"), err)] pub fn recover_population_physical_dag(&self) -> Result { let invalid = |message: &str| QueryPlanError::Invalid(message.into()); let population = self @@ -340,6 +346,8 @@ impl QueryPlanEntry { /// Validate bound counter vectors or stored aggregate batches against the /// retained physical program; recovery never lowers logical operators. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.recover_validate", query_id = %self.query_id, input_kind = "bound_native_input"), err)] pub fn recover_vector_physical_dag(&self) -> Result { let invalid = |message: &str| QueryPlanError::Invalid(message.into()); let (inputs, source_nodes, max_bytes) = self diff --git a/data_plane/docker-compose.yml.j2 b/data_plane/docker-compose.yml.j2 index cacf95ef..7a650240 100644 --- a/data_plane/docker-compose.yml.j2 +++ b/data_plane/docker-compose.yml.j2 @@ -27,7 +27,6 @@ services: "--prometheus-server", "http://{{ prometheus_host }}:{{ prometheus_port }}", "--prometheus-scrape-interval", "{{ prometheus_scrape_interval }}", "--delete-existing-db", - "--log-level", "{{ log_level }}", "--output-dir", "/app/outputs", "--streaming-engine", "{{ streaming_engine }}", "--query-language", "{{ query_language }}", diff --git a/data_plane/query-engine-rust-cli-compose.yml.j2 b/data_plane/query-engine-rust-cli-compose.yml.j2 index 32fe8f96..d54bb112 100644 --- a/data_plane/query-engine-rust-cli-compose.yml.j2 +++ b/data_plane/query-engine-rust-cli-compose.yml.j2 @@ -25,7 +25,6 @@ services: "--prometheus-server", "http://prometheus:9090", "--prometheus-scrape-interval", "{{ prometheus_scrape_interval }}", "--delete-existing-db", - "--log-level", "{{ log_level }}", "--output-dir", "/app/outputs", "--streaming-engine", "{{ streaming_engine }}", "--query-language", "{{ query_language }}", diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index a8294d09..2a808bf0 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -9,11 +9,14 @@ use axum::{ }; use serde_json::Value; use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::net::TcpListener; use tracing::{debug, info, warn}; +static NEXT_PLAN_REQUEST_ID: AtomicU64 = AtomicU64::new(1); + use crate::drivers::query::adapters::{AdapterConfig, HttpProtocolAdapter, PrometheusHttpAdapter}; use crate::drivers::query::servers::metrics as srv_metrics; use crate::query_engines::routing::{ @@ -1999,7 +2002,7 @@ async fn handle_runtime_info( let mut forwarding_headers = HashMap::new(); if let Some(auth) = headers.get(axum::http::header::AUTHORIZATION) { if let Ok(auth_str) = auth.to_str() { - debug!("Found Authorization header for runtime info: {}", auth_str); + debug!("Found Authorization header for runtime info"); forwarding_headers.insert("Authorization".to_string(), auth_str.to_string()); } } else { @@ -5366,10 +5369,14 @@ pub use asap_types::plan_publication::PhysicalPlanInstallRequest; /// Decode and cross-validate every backend view before it can become visible. /// Used by both startup artifact loading and the staged HTTP install path. +#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(plan_id = request.transmission_plan.envelope.plan_id, + plan_version = request.transmission_plan.envelope.plan_version))] pub fn validate_and_build_runtime_plan( request: PhysicalPlanInstallRequest, default_routing: Arc, ) -> Result { + tracing::debug!(target: "asap_runtime_debug", "backend runtime plan validation started"); use std::collections::BTreeSet; request .precompute_plan @@ -5453,8 +5460,25 @@ async fn handle_post_physical_plan( ) -> axum::response::Response { use axum::http::StatusCode; use axum::response::IntoResponse; + let call_id = NEXT_PLAN_REQUEST_ID.fetch_add(1, Ordering::Relaxed); + let started = Instant::now(); + let requested_plan_id = request.transmission_plan.envelope.plan_id; + let requested_plan_version = request.transmission_plan.envelope.plan_version; + tracing::debug!(target: "asap_runtime_debug", + call_id, + plan_id = requested_plan_id, + plan_version = requested_plan_version, + "physical plan staging requested" + ); let Some(active_handle) = state.active_physical_plan.as_ref() else { + tracing::warn!( + call_id, + plan_id = requested_plan_id, + plan_version = requested_plan_version, + error = "physical-plan hot-reload handles are not attached", + "physical plan staging failed" + ); return ( StatusCode::SERVICE_UNAVAILABLE, axum::Json(serde_json::json!({ @@ -5464,6 +5488,13 @@ async fn handle_post_physical_plan( .into_response(); }; let Some(lifecycle) = state.physical_plan_lifecycle.as_ref() else { + tracing::warn!( + call_id, + plan_id = requested_plan_id, + plan_version = requested_plan_version, + error = "physical-plan lifecycle is not attached", + "physical plan staging failed" + ); return ( StatusCode::SERVICE_UNAVAILABLE, axum::Json(serde_json::json!({ @@ -5479,6 +5510,9 @@ async fn handle_post_physical_plan( &request.adaptation_evidence, unix_time_ms(), ) { + tracing::warn!(call_id, plan_id = requested_plan_id, + plan_version = requested_plan_version, + %error, "physical plan successor rejected"); return ( StatusCode::UNPROCESSABLE_ENTITY, axum::Json(serde_json::json!({ @@ -5493,6 +5527,9 @@ async fn handle_post_physical_plan( let active = match validate_and_build_runtime_plan(request, current.storage_routing.clone()) { Ok(active) => active, Err(error) => { + tracing::warn!(call_id, plan_id = requested_plan_id, + plan_version = requested_plan_version, %error, + "physical plan validation failed"); return ( StatusCode::UNPROCESSABLE_ENTITY, axum::Json(serde_json::json!({"status": "error", "error": error})), @@ -5508,6 +5545,13 @@ async fn handle_post_physical_plan( ) || active.precompute_plan.ingest.endpoint_path != "/api/v1/write") { + tracing::warn!( + call_id, + plan_id = requested_plan_id, + plan_version = requested_plan_version, + error = "Remote Write listener requires a non-bootstrap prometheus_remote_write_v1 plan at /api/v1/write", + "physical plan staging failed" + ); return ( StatusCode::UNPROCESSABLE_ENTITY, axum::Json(serde_json::json!({ @@ -5534,12 +5578,23 @@ async fn handle_post_physical_plan( let plan_version = active.plan_version(); let now = unix_time_ms(); if let Err(error) = lifecycle.stage(active, now) { + tracing::warn!(call_id, plan_id, plan_version, %error, "physical plan staging failed"); return ( StatusCode::CONFLICT, axum::Json(serde_json::json!({"status": "error", "error": error.to_string()})), ) .into_response(); } + tracing::info!( + call_id, + plan_id, + plan_version, + elapsed_ms = started.elapsed().as_millis() as u64, + materialization_count, + metricsql_query_count, + clickhouse_plan_count, + "physical plan staged" + ); ( StatusCode::ACCEPTED, axum::Json(serde_json::json!({ @@ -5564,10 +5619,25 @@ async fn handle_activate_physical_plan( axum::Json(request): axum::Json, ) -> axum::response::Response { use axum::response::IntoResponse; + let call_id = NEXT_PLAN_REQUEST_ID.fetch_add(1, Ordering::Relaxed); + let started = Instant::now(); + tracing::debug!(target: "asap_runtime_debug", + call_id, + plan_id = request.plan_id, + plan_version = request.plan_version, + "physical plan activation requested" + ); let (Some(lifecycle), Some(active_handle)) = ( state.physical_plan_lifecycle.as_ref(), state.active_physical_plan.as_ref(), ) else { + tracing::warn!( + call_id, + plan_id = request.plan_id, + plan_version = request.plan_version, + error = "physical-plan lifecycle is not attached", + "physical plan activation failed" + ); return ( StatusCode::SERVICE_UNAVAILABLE, axum::Json(serde_json::json!({ @@ -5600,16 +5670,27 @@ async fn handle_activate_physical_plan( ) { Ok(old) => old, Err(error) => { + tracing::warn!(call_id, plan_id = request.plan_id, plan_version = request.plan_version, + %error, "physical plan activation failed"); return ( StatusCode::CONFLICT, axum::Json(serde_json::json!({ "status": "error", "error": error.to_string() })), ) - .into_response() + .into_response(); } }; let activated = active_handle.active_snapshot(); + tracing::info!( + call_id, + plan_id = request.plan_id, + plan_version = request.plan_version, + elapsed_ms = started.elapsed().as_millis() as u64, + previous_plan_id = old.plan_id(), + previous_plan_version = old.plan_version(), + "physical plan activated; summary catalog installed" + ); let clickhouse_plan_count = activated .query_plan .entries diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 7fcefac3..3003bbf6 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -171,10 +171,6 @@ struct Args { #[arg(long, default_value = "/var/log/asap")] output_dir: String, - /// Log level - #[arg(long, default_value = "INFO")] - log_level: String, - /// Enable profiling (currently unused, kept for compatibility) #[arg(long)] do_profiling: bool, @@ -519,7 +515,7 @@ async fn main() -> Result<()> { // Initialize logging similar to Python's create_loggers function // Keep the guard alive for the entire lifetime of the application - let _log_guard = setup_logging(&args.output_dir, &args.log_level)?; + let _log_guard = setup_logging(&args.output_dir)?; info!("Starting Query Engine Rust"); info!("Output directory: {}", args.output_dir); @@ -1325,16 +1321,10 @@ async fn spawn_memory_diagnostics( } } -fn setup_logging( - output_dir: &str, - log_level: &str, -) -> Result { +fn setup_logging(output_dir: &str) -> Result { use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; - // Create env filter that respects RUST_LOG, with fallback to command line arg - let env_filter = EnvFilter::try_from_default_env() - .or_else(|_| EnvFilter::try_new(log_level)) - .unwrap_or_else(|_| EnvFilter::new("info")); + let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); // Create file appender for logging to file let file_appender = tracing_appender::rolling::never(output_dir, "query_engine.log"); diff --git a/data_plane/src/precompute_engine/native_maintenance.rs b/data_plane/src/precompute_engine/native_maintenance.rs index dba59db0..0e7ed0d0 100644 --- a/data_plane/src/precompute_engine/native_maintenance.rs +++ b/data_plane/src/precompute_engine/native_maintenance.rs @@ -14,6 +14,8 @@ use planner_types::{ pre_asap::DataType, }; +#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.maintenance.execute", query_id = %installed.document.query_id, window_start_ms = window.0, window_end_ms = window.1, max_bytes, input_populations = inputs.len()), err)] pub(super) fn execute( installed: &InstalledPostAsapDag, program: &CompiledPhysicalDag, diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index 42f2562c..d2cd54d1 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -5,7 +5,7 @@ use crate::storage_engines::types::hot_reload_config::InstalledPrecomputePlanHan use crate::storage_engines::types::{AggregateCore, PrecomputedOutput}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; -use tracing::{debug_span, warn}; +use tracing::{debug, debug_span, warn}; /// CQ-6 — process-global fallback for the output-sink policy-miss /// counter, used when a `SketchStoreSink` was constructed without an @@ -233,11 +233,15 @@ impl OutputSink for SketchStoreSink { if outputs.is_empty() { return Ok(()); } - let _span = debug_span!("sketch_index_insert", batch_size = outputs.len()).entered(); + let _span = debug_span!(target: "asap_runtime_debug", "sketch_index_insert", + batch_size = outputs.len()) + .entered(); let output_count = outputs.len(); let failed = consume_in_order(outputs, |(output, accumulator)| { self.append_to_index(output, accumulator.as_ref()) }); + debug!(target: "asap_runtime_debug", output_count, failed, + "precompute output batch stored"); if failed > 0 { return Err(format!( "SketchStore rejected {failed} of {} completed outputs", diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs index 76e809af..dc091b1d 100644 --- a/data_plane/src/precompute_engine/subdag_scheduler.rs +++ b/data_plane/src/precompute_engine/subdag_scheduler.rs @@ -82,6 +82,33 @@ where Ok(outputs.remove(0)) } +fn node_syntax(payload: &ExecutableOperatorPayload) -> String { + let details = match payload { + ExecutableOperatorPayload::Fallback { .. } => String::new(), + ExecutableOperatorPayload::Binary { operator } => { + format!("operator={operator:?}") + } + ExecutableOperatorPayload::Value { operation } => { + format!("operation={operation:?}") + } + ExecutableOperatorPayload::RelationalJoin { join_kind, .. } => { + format!("join_kind={join_kind:?}") + } + ExecutableOperatorPayload::SummaryAgg { + family, + input, + reduction, + .. + } => format!("family={family:?} input={input:?} reduction={reduction:?}"), + ExecutableOperatorPayload::SummaryJoin { family, .. } => format!("family={family:?}"), + ExecutableOperatorPayload::SummarySubtract => String::new(), + ExecutableOperatorPayload::SummaryDelete { .. } => String::new(), + ExecutableOperatorPayload::SummaryEstimate { query } => format!("readout={query:?}"), + ExecutableOperatorPayload::SummaryMerge => String::new(), + }; + details.chars().take(256).collect() +} + /// Evaluate all selected stored outputs with one dependency cache. Keys must /// describe the same input revision and window; only their output identity may /// differ. Validation finishes before executing or committing any output. @@ -270,10 +297,12 @@ where } }; if committed.contains(&node.0) { + tracing::debug!(target: "asap_runtime_debug", sink_node_id = node.0, "precompute DAG reused committed sink"); Ok(value) } else { - sink.commit_if_absent(key.clone(), value) - .map_err(ScheduleError::Sink) + let value = sink.commit_if_absent(key.clone(), value).map_err(ScheduleError::Sink)?; + tracing::debug!(target: "asap_runtime_debug", sink_node_id = node.0, "precompute DAG sink commit completed"); + Ok(value) } } }), @@ -310,6 +339,7 @@ impl> ) -> Result>, execution::Error> { Ok(futures::stream::once(async move { if let Some(source) = &self.source { + tracing::debug!(target: "asap_runtime_debug", node_id = self.node.id.0, "precompute node used materialized input"); return Ok(Arc::clone(source)); } let inputs = @@ -323,10 +353,16 @@ impl> .iter() .map(|value| Arc::clone(value.value())) .collect::>(); + let started = std::time::Instant::now(); + tracing::debug!(target: "asap_runtime_debug", node_id = self.node.id.0, phase = ?self.node.output_state.timing, syntax = %node_syntax(&self.node.payload), "precompute node started"); self.registry .execute(self.node, &values, context) - .map(Arc::new) + .map(|value| { + tracing::debug!(target: "asap_runtime_debug", node_id = self.node.id.0, elapsed_us = started.elapsed().as_micros() as u64, "precompute node completed"); + Arc::new(value) + }) .map_err(|e| { + tracing::warn!(node_id = self.node.id.0, elapsed_us = started.elapsed().as_micros() as u64, "precompute node failed"); *self.error.borrow_mut() = Some(e); execution::Error::Operator(format!("ingestion node {} failed", self.node.id.0)) }) diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 0ed985c3..70f712c5 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -335,8 +335,11 @@ impl Worker { } => { let sample_count = samples.len(); let _span = debug_span!( + target: "asap_runtime_debug", "worker_process_group", worker_id = self.id, + plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id), + plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version), sid, policy_fp = %policy_fp, group = %group_key, @@ -346,12 +349,13 @@ impl Worker { if let Err(e) = self.process_group_samples(sid, policy_fp, &group_key, samples) { processing_error = Some(e.to_string()); - warn!( - "Worker {} error processing sid={} (policy_fp={}, group={}): {}", - self.id, sid, policy_fp, group_key, e - ); + warn!(worker_id = self.id, sid, policy_fp = %policy_fp, + plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id), + plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version), + error = %e, "worker group processing failed"); } debug!( + target: "asap_runtime_debug", e2e_latency_us = ingest_received_at.elapsed().as_micros() as u64, "e2e: ingest->worker complete" ); @@ -362,17 +366,24 @@ impl Worker { ingest_received_at, } => { let _span = debug_span!( + target: "asap_runtime_debug", "worker_process_raw", worker_id = self.id, + plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id), + plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version), series = %series_key, sample_count = samples.len(), ) .entered(); if let Err(e) = self.process_samples_raw(&series_key, samples) { processing_error = Some(e.to_string()); - warn!("Worker {} raw error for {}: {}", self.id, series_key, e); + warn!(worker_id = self.id, series = %series_key, + plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id), + plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version), + error = %e, "worker raw processing failed"); } debug!( + target: "asap_runtime_debug", e2e_latency_us = ingest_received_at.elapsed().as_micros() as u64, "e2e: ingest->worker complete (raw)" ); @@ -386,8 +397,11 @@ impl Worker { ingest_received_at, } => { let _span = debug_span!( + target: "asap_runtime_debug", "worker_process_accumulator", worker_id = self.id, + plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id), + plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version), sid, policy_fp = %policy_fp, group = %group_key, @@ -403,12 +417,13 @@ impl Worker { accumulator, ) { processing_error = Some(e.to_string()); - warn!( - "Worker {} accumulator input error for sid={} (policy_fp={}, group={}): {}", - self.id, sid, policy_fp, group_key, e - ); + warn!(worker_id = self.id, sid, policy_fp = %policy_fp, + plan_id = ?self.current_catalog_generation.as_ref().map(|g| g.plan_id), + plan_version = ?self.current_catalog_generation.as_ref().map(|g| g.plan_version), + error = %e, "worker accumulator processing failed"); } debug!( + target: "asap_runtime_debug", e2e_latency_us = ingest_received_at.elapsed().as_micros() as u64, "e2e: ingest->worker complete (accumulator)" ); @@ -528,6 +543,8 @@ impl Worker { /// `PrecomputeMaterialization` fingerprint used to resolve the bucket's /// config on first sight; `group_key` is held on the resulting /// `GroupState` for emit-time label rendering. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(worker_id = self.id, sid, policy_fp = %policy_fp, sample_count = samples.len()))] pub fn process_group_samples( &mut self, sid: u64, @@ -535,6 +552,7 @@ impl Worker { group_key: &Arc, samples: Vec<(String, i64, f64)>, // (series_key, timestamp_ms, value) ) -> Result<(), Box> { + debug!(target: "asap_runtime_debug", "worker group processing started"); let input_revision = self.current_input_revision.clone(); let worker_id = self.id; let allowed_lateness_ms = self.allowed_lateness_ms; @@ -552,6 +570,8 @@ impl Worker { return Ok(()); } let state = self.group_states.get_mut(&sid).unwrap(); + tracing::debug!(target: "asap_runtime_debug", worker_id = self.id, policy_fp = %policy_fp, + "precompute update route resolved"); #[cfg(not(test))] if state.program.is_none() { return Err("raw precompute requires an installed post-ASAP DAG producer".into()); @@ -851,6 +871,8 @@ impl Worker { /// /// `policy_fp` / `group_key` carry the same semantics as on /// `process_group_samples` — policy lookup + emit-time label rendering. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(worker_id = self.id, sid, policy_fp = %policy_fp, timestamp_ms))] pub fn process_accumulator_input( &mut self, sid: u64, @@ -859,6 +881,7 @@ impl Worker { timestamp_ms: i64, incoming: Box, ) -> Result<(), Box> { + debug!(target: "asap_runtime_debug", "worker accumulator processing started"); let worker_id = self.id; let allowed_lateness_ms = self.allowed_lateness_ms; let late_data_policy = self.late_data_policy; @@ -877,6 +900,10 @@ impl Worker { let state = self.group_states.get_mut(&sid).unwrap(); let previous_event_time = state.max_event_time_ms; + debug!(target: "asap_runtime_debug", aggregation_type = ?state.config.aggregation_type, + window_secs = state.config.window_size, slide_secs = state.config.slide_interval, + layout = ?state.config.window_layout, + "worker accumulator maintenance configuration selected"); let current_event_time = if timestamp_ms > previous_event_time { timestamp_ms } else { @@ -1021,11 +1048,14 @@ impl Worker { } /// Raw fast-path: emit each sample as a standalone `SumAccumulator`. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(worker_id = self.id, sample_count = samples.len()))] pub fn process_samples_raw( &self, series_key: &str, samples: Vec<(i64, f64)>, ) -> Result<(), Box> { + debug!(target: "asap_runtime_debug", "worker raw processing started"); let mut emit_batch: Vec<(PrecomputedOutput, Box)> = Vec::with_capacity(samples.len()); @@ -1097,7 +1127,10 @@ impl Worker { } } + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(worker_id = self.id, group_count = self.group_states.len()))] fn flush_all(&mut self) -> Result<(), Box> { + debug!(target: "asap_runtime_debug", "worker flush started"); if self.pass_raw_samples { return Ok(()); } diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs index c8583c0e..a28fd250 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs @@ -77,6 +77,8 @@ impl RelationDagExecutor<'_> { #[cfg(not(test))] fn record_evaluation(&mut self, _id: QueryNodeId) {} + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.execute", query_id = %self.entry.query_id, root = root.0), err)] fn execute_graph( &mut self, root: QueryNodeId, @@ -166,9 +168,15 @@ impl RelationDagExecutor<'_> { ) as asap_physical_operators::physical_planner::Source<'_>, ); } - let graph = compiled - .instantiate(resolved_inputs) - .map_err(|e| e.to_string())?; + let graph = { + let _binding = + tracing::debug_span!(target: "asap_runtime_debug", "physical_input_binding", + stage = "physical.bind_inputs", input_count = resolved_inputs.len()) + .entered(); + compiled + .instantiate(resolved_inputs) + .map_err(|e| e.to_string())? + }; let mut output = graph .execute(&[root.0], context) .map_err(|e| e.to_string())? diff --git a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs index c227e5eb..a5ac9d4d 100644 --- a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs +++ b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs @@ -119,40 +119,48 @@ pub(crate) fn validate_payload( let mut resolved = BTreeMap::new(); for id in entry.topological_order().map_err(|e| miss(e.to_string()))? { let node = &entry.nodes[&id]; - let state_ids = match node { - QueryPlanNode::ReadMaterialization { binding } => { - if let std::collections::btree_map::Entry::Vacant(entry) = - resolved.entry(binding.materialization) - { - entry.insert(resolve(catalog, binding.materialization)?); + let state_ids = (|| -> Result, EngineError> { + Ok(match node { + QueryPlanNode::ReadMaterialization { binding } => { + if let std::collections::btree_map::Entry::Vacant(entry) = + resolved.entry(binding.materialization) + { + entry.insert(resolve(catalog, binding.materialization)?); + } + BTreeSet::from([binding.materialization]) } - BTreeSet::from([binding.materialization]) - } - QueryPlanNode::SummaryMerge { inputs } => { - let mut ids = BTreeSet::new(); - for input in inputs { - let children: &BTreeSet = states - .get(input) - .ok_or_else(|| miss("summary merge has no state input"))?; - if children.is_empty() { - return Err(miss("summary merge has value input")); + QueryPlanNode::SummaryMerge { inputs } => { + let mut ids = BTreeSet::new(); + for input in inputs { + let children: &BTreeSet = states + .get(input) + .ok_or_else(|| miss("summary merge has no state input"))?; + if children.is_empty() { + return Err(miss("summary merge has value input")); + } + ids.extend(children); } - ids.extend(children); + ids } - ids - } - QueryPlanNode::SummaryEstimate { input, .. } - | QueryPlanNode::ExactReadout { input, .. } => { - let ids: &BTreeSet = states - .get(input) - .ok_or_else(|| miss("readout has no state input"))?; - if ids.is_empty() || ids.iter().any(|id| !resolved[id].supports(node)) { - return Err(miss("catalog descriptor cannot satisfy installed readout")); + QueryPlanNode::SummaryEstimate { input, .. } + | QueryPlanNode::ExactReadout { input, .. } => { + let ids: &BTreeSet = states + .get(input) + .ok_or_else(|| miss("readout has no state input"))?; + if ids.is_empty() || ids.iter().any(|id| !resolved[id].supports(node)) { + return Err(miss("catalog descriptor cannot satisfy installed readout")); + } + BTreeSet::new() } - BTreeSet::new() - } - _ => BTreeSet::new(), - }; + _ => BTreeSet::new(), + }) + })() + .map_err(|error| { + tracing::debug!(target: "asap_runtime_debug", query_id = %entry.query_id, + node_id = ?id, op = node.op_label(), %error, + "installed query node failed catalog validation"); + error + })?; states.insert(id, state_ids); } Ok(()) diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index df980fcb..eedb39c3 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -1,5 +1,40 @@ +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; -use tracing::debug; +use std::time::Instant; +use tracing::{debug, Instrument}; + +static NEXT_QUERY_CALL_ID: AtomicU64 = AtomicU64::new(1); + +fn query_call_id() -> u64 { + NEXT_QUERY_CALL_ID.fetch_add(1, Ordering::Relaxed) +} + +fn log_query_outcome( + call_id: u64, + operation: &'static str, + started: Instant, + remote_stats: Option<(usize, usize)>, + result: &Result, +) { + match result { + Ok(_) => match remote_stats { + Some((remote_evaluations, remote_rpcs)) => debug!(target: "asap_runtime_debug", + call_id, operation, elapsed_ms = started.elapsed().as_millis() as u64, + remote_evaluations, remote_rpcs, "query call completed"), + None => debug!(target: "asap_runtime_debug", + call_id, operation, elapsed_ms = started.elapsed().as_millis() as u64, + "query call completed"), + }, + Err(error @ crate::query_engines::EngineError::CapabilityMiss { .. }) => { + debug!(target: "asap_runtime_debug", + call_id, operation, elapsed_ms = started.elapsed().as_millis() as u64, + %error, "query call could not be served by ASAP tier") + } + Err(error) => tracing::warn!(call_id, operation, + elapsed_ms = started.elapsed().as_millis() as u64, %error, + "query call failed"), + } +} use asap_types::query_requirements::QueryRequirements; use asap_types::KeyByLabelNames; @@ -191,27 +226,49 @@ impl ASAPQueryEngine { now_ms: u64, ) -> Result { - let physical = self.active_physical_plan_snapshot().ok_or_else(|| { - crate::query_engines::EngineError::capability_miss( - "query_plan", - "no active physical plan", - ) - })?; - let planned = physical - .query_plan - .lookup_canonical(asap_types::query_plan::QueryLanguage::MetricsQl, identity) - .map_err(|error| { - crate::query_engines::EngineError::capability_miss("query_plan", error.to_string()) + let call_id = query_call_id(); + let started = Instant::now(); + let mut remote_stats = None; + debug!(target: "asap_runtime_debug", + call_id, + operation = "metricsql_instant", + evaluation_ms = now_ms, + "query call started" + ); + let result = async { + let physical = self.active_physical_plan_snapshot().ok_or_else(|| { + crate::query_engines::EngineError::capability_miss( + "query_plan", + "no active physical plan", + ) })?; - let leaves = self - .prepare_query_inputs(&physical, planned, &[now_ms]) - .await?; - let (mut result, mut stats) = - self.execute_logical_entry(&physical, planned, &leaves, now_ms)?; - stats.remote_evaluations = leaves.values().map(|leaf| leaf.remote_evaluations).sum(); - stats.remote_rpcs = leaves.values().map(|leaf| leaf.remote_rpcs).sum(); - annotate_logical_execution(&mut result, &stats); - Ok(result) + let planned = physical + .query_plan + .lookup_canonical(asap_types::query_plan::QueryLanguage::MetricsQl, identity) + .map_err(|error| { + crate::query_engines::EngineError::capability_miss( + "query_plan", + error.to_string(), + ) + })?; + debug!(target: "asap_runtime_debug", plan_id = physical.plan_id(), plan_version = physical.plan_version(), + query_id = %planned.query_id, evaluation_ms = now_ms, + "installed MetricsQL instant query selected"); + let leaves = self + .prepare_query_inputs(&physical, planned, &[now_ms]) + .await?; + let (mut result, mut stats) = + self.execute_logical_entry(&physical, planned, &leaves, now_ms)?; + stats.remote_evaluations = leaves.values().map(|leaf| leaf.remote_evaluations).sum(); + stats.remote_rpcs = leaves.values().map(|leaf| leaf.remote_rpcs).sum(); + annotate_logical_execution(&mut result, &stats); + remote_stats = Some((stats.remote_evaluations, stats.remote_rpcs)); + Ok(result) + } + .instrument(tracing::info_span!("query_call", call_id)) + .await; + log_query_outcome(call_id, "metricsql_instant", started, remote_stats, &result); + result } pub async fn execute_metricsql_range( @@ -222,20 +279,42 @@ impl ASAPQueryEngine { step_ms: u64, ) -> Result { - let physical = self.active_physical_plan_snapshot().ok_or_else(|| { - crate::query_engines::EngineError::capability_miss( - "query_plan", - "no active physical plan", - ) - })?; - let planned = physical - .query_plan - .lookup_canonical(asap_types::query_plan::QueryLanguage::MetricsQl, identity) - .map_err(|error| { - crate::query_engines::EngineError::capability_miss("query_plan", error.to_string()) + let call_id = query_call_id(); + let started = Instant::now(); + debug!(target: "asap_runtime_debug", + call_id, + operation = "metricsql_range", + start_ms, + end_ms, + step_ms, + "query call started" + ); + let result = async { + let physical = self.active_physical_plan_snapshot().ok_or_else(|| { + crate::query_engines::EngineError::capability_miss( + "query_plan", + "no active physical plan", + ) })?; - self.execute_logical_range(&physical, planned, start_ms, end_ms, step_ms) - .await + let planned = physical + .query_plan + .lookup_canonical(asap_types::query_plan::QueryLanguage::MetricsQl, identity) + .map_err(|error| { + crate::query_engines::EngineError::capability_miss( + "query_plan", + error.to_string(), + ) + })?; + debug!(target: "asap_runtime_debug", plan_id = physical.plan_id(), plan_version = physical.plan_version(), + query_id = %planned.query_id, start_ms, end_ms, step_ms, + "installed MetricsQL range query selected"); + self.execute_logical_range(&physical, planned, start_ms, end_ms, step_ms) + .await + } + .instrument(tracing::info_span!("query_call", call_id)) + .await; + log_query_outcome(call_id, "metricsql_range", started, None, &result); + result } /// Construct the query executor. Runtime configuration is read only from @@ -272,12 +351,16 @@ impl ASAPQueryEngine { self.query_forwarding_policy = policy; self } + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(plan_id = physical.plan_id(), plan_version = physical.plan_version(), + query_id = %entry.query_id, evaluation_count = times.len()))] async fn prepare_query_inputs( &self, physical: &crate::storage_engines::types::RuntimePhysicalPlan, entry: &asap_types::query_plan::QueryPlanEntry, times: &[u64], ) -> Result { + debug!(target: "asap_runtime_debug", "installed query input preparation started"); super::catalog_resolver::validate_entry( physical.summary_catalog.as_deref(), entry, @@ -350,7 +433,7 @@ impl ASAPQueryEngine { ) }) { - debug!( + debug!(target: "asap_runtime_debug", language = ?entry.language, query_id = %entry.query_id, "query forwarding disabled; external exact subquery blocked" @@ -382,6 +465,9 @@ impl ASAPQueryEngine { .await } + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(plan_id = physical.plan_id(), plan_version = physical.plan_version(), + query_id = %entry.query_id, evaluation_ms = at))] fn execute_logical_entry( &self, physical: &crate::storage_engines::types::RuntimePhysicalPlan, @@ -552,6 +638,9 @@ impl ASAPQueryEngine { result } + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(plan_id = physical.plan_id(), plan_version = physical.plan_version(), + query_id = %entry.query_id, start_ms = start, end_ms = end, step_ms = step))] async fn execute_logical_range( &self, physical: &crate::storage_engines::types::RuntimePhysicalPlan, @@ -565,6 +654,7 @@ impl ASAPQueryEngine { query_result::{QueryResult, RangeVectorElement}, EngineError, }; + debug!(target: "asap_runtime_debug", "installed query range execution started"); if step == 0 || start > end || (end - start) / step >= 11_000 { return Err(EngineError::capability_miss( "installed_logical_dag", @@ -755,8 +845,21 @@ impl ASAPQueryEngine { step_ms: u64, ) -> Result { + let call_id = query_call_id(); + let started = Instant::now(); + debug!(target: "asap_runtime_debug", + call_id, + operation = "promql_range", + start_ms, + end_ms, + step_ms, + "query call started" + ); + let result = async { if let Some(physical) = self.active_physical_plan_snapshot() { if let Ok(entry) = physical.query_plan.lookup(query) { + debug!(target: "asap_runtime_debug", plan_id = physical.plan_id(), plan_version = physical.plan_version(), + query_id = %entry.query_id, "installed query DAG selected"); return self .execute_logical_range(&physical, entry, start_ms, end_ms, step_ms) .await; @@ -862,6 +965,9 @@ impl ASAPQueryEngine { // Complete coverage is a prerequisite above. Hybrid stitching is // retained for legacy/test callers without an active QueryPlan only. Ok(warm_qr) + }.instrument(tracing::info_span!("query_call", call_id)).await; + log_query_outcome(call_id, "promql_range", started, None, &result); + result } } @@ -1005,25 +1111,43 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu now_ms: u64, ) -> Result { + let call_id = query_call_id(); + let started = Instant::now(); + let mut remote_stats = None; + debug!(target: "asap_runtime_debug", + call_id, + operation = "promql_instant", + evaluation_ms = now_ms, + "query call started" + ); + let result = async { if let Some(physical) = self.active_physical_plan_snapshot() { if let Ok(entry) = physical.query_plan.lookup(query) { + debug!(target: "asap_runtime_debug", plan_id = physical.plan_id(), plan_version = physical.plan_version(), + query_id = %entry.query_id, evaluation_ms = now_ms, + "installed query DAG selected"); let leaves = self .prepare_query_inputs(&physical, entry, &[now_ms]) .await .map_err(|error| { - tracing::warn!(query, error = %error, "installed query DAG preparation failed"); + tracing::warn!(plan_id = physical.plan_id(), + plan_version = physical.plan_version(), query_id = %entry.query_id, + error = %error, "installed query DAG preparation failed"); error })?; let (mut result, mut stats) = self .execute_logical_entry(&physical, entry, &leaves, now_ms) .map_err(|error| { - tracing::warn!(query, error = %error, "installed query DAG execution failed"); + tracing::warn!(plan_id = physical.plan_id(), + plan_version = physical.plan_version(), query_id = %entry.query_id, + error = %error, "installed query DAG execution failed"); error })?; stats.remote_evaluations = leaves.values().map(|leaf| leaf.remote_evaluations).sum(); stats.remote_rpcs = leaves.values().map(|leaf| leaf.remote_rpcs).sum(); annotate_logical_execution(&mut result, &stats); + remote_stats = Some((stats.remote_evaluations, stats.remote_rpcs)); return Ok(result); } } @@ -1121,6 +1245,9 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!("ASAPQueryEngine: no sketch index for `{query}` — failing over to archive"), )) + }.instrument(tracing::info_span!("query_call", call_id)).await; + log_query_outcome(call_id, "promql_instant", started, remote_stats, &result); + result } /// Range-query entry point for the [`EngineRouter`] failover loop. diff --git a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs index a782eb93..be46457a 100644 --- a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs +++ b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs @@ -300,147 +300,154 @@ pub(super) async fn prepare_external( // Equivalent exact cuts at the same time share one actual remote request. let mut remote_cache = HashMap::<(QueryLanguage, String, i64), Value>::new(); for ((id, at), leaf) in leaves(entry, times)? { - u64::try_from(at).map_err(|_| miss("subquery predates epoch"))?; - let (language, query, candidate_input) = match &leaf { - ExactLeaf::Legacy(ResidualQueryOperator::ExactSubquery { query }) => { - (QueryLanguage::PromQl, query.clone(), None) - } - ExactLeaf::Legacy(ResidualQueryOperator::CandidateExactSubquery { - query, - item_label, - }) => ( - QueryLanguage::PromQl, - query.clone(), - Some((entry.nodes[&id].inputs()[0], item_label.as_str())), - ), - ExactLeaf::External(request) => { - if !matches!( - request.language, - QueryLanguage::PromQl | QueryLanguage::MetricsQl - ) { - return Err(miss(format!( - "external exact language {:?} has no installed adapter", - request.language - ))); + let result = async { + u64::try_from(at).map_err(|_| miss("subquery predates epoch"))?; + let (language, query, candidate_input) = match &leaf { + ExactLeaf::Legacy(ResidualQueryOperator::ExactSubquery { query }) => { + (QueryLanguage::PromQl, query.clone(), None) } - let candidate = match request.input_contracts.as_slice() { - [] => None, - [ExternalExactInput::CandidateMembership { item_label }] => { - Some((entry.nodes[&id].inputs()[0], item_label.as_str())) + ExactLeaf::Legacy(ResidualQueryOperator::CandidateExactSubquery { + query, + item_label, + }) => ( + QueryLanguage::PromQl, + query.clone(), + Some((entry.nodes[&id].inputs()[0], item_label.as_str())), + ), + ExactLeaf::External(request) => { + if !matches!( + request.language, + QueryLanguage::PromQl | QueryLanguage::MetricsQl + ) { + return Err(miss(format!( + "external exact language {:?} has no installed adapter", + request.language + ))); } - _ => return Err(miss("unsupported external exact input contract")), - }; - (request.language, request.expression.clone(), candidate) - } - _ => return Err(miss("prepared leaf is not an exact subtree")), - }; - let (query, candidate_filtered) = if let Some((candidate_input, item_label)) = - candidate_input - { - if language == QueryLanguage::MetricsQl { - return Err(miss( - "candidate-filtered MetricsQL exact subqueries are not implemented", - )); - } - let candidate = prepared - .get(&(candidate_input, at)) - .ok_or_else(|| miss("candidate membership was not prepared"))?; - let Value::Vector(rows) = &candidate.value else { - return Err(miss("candidate membership is not an instant vector")); + let candidate = match request.input_contracts.as_slice() { + [] => None, + [ExternalExactInput::CandidateMembership { item_label }] => { + Some((entry.nodes[&id].inputs()[0], item_label.as_str())) + } + _ => return Err(miss("unsupported external exact input contract")), + }; + (request.language, request.expression.clone(), candidate) + } + _ => return Err(miss("prepared leaf is not an exact subtree")), }; - let mut values = rows - .iter() - .map(|(labels, _)| { - labels.get(item_label).cloned().ok_or_else(|| { - miss(format!( - "candidate membership is missing item label {item_label}" - )) + let (query, candidate_filtered) = if let Some((candidate_input, item_label)) = + candidate_input + { + if language == QueryLanguage::MetricsQl { + return Err(miss( + "candidate-filtered MetricsQL exact subqueries are not implemented", + )); + } + let candidate = prepared + .get(&(candidate_input, at)) + .ok_or_else(|| miss("candidate membership was not prepared"))?; + let Value::Vector(rows) = &candidate.value else { + return Err(miss("candidate membership is not an instant vector")); + }; + let mut values = rows + .iter() + .map(|(labels, _)| { + labels.get(item_label).cloned().ok_or_else(|| { + miss(format!( + "candidate membership is missing item label {item_label}" + )) + }) }) - }) - .collect::, _>>()?; - values.sort(); - values.dedup(); - if values.is_empty() { - prepared.insert( - (id, at), - PreparedLeaf { + .collect::, _>>()?; + values.sort(); + values.dedup(); + if values.is_empty() { + return Ok(PreparedLeaf { value: Value::Vector(Vec::new()), remote: true, remote_evaluations: 0, remote_rpcs: 0, - }, - ); - continue; - } - if values.len() > MAX_CANDIDATE_VALUES { - return Err(miss(format!( - "candidate set has {} values, exceeding limit {MAX_CANDIDATE_VALUES}", - values.len() - ))); - } - let restricted = inject_candidate_matcher(&query, item_label, &values)?; - if restricted.len() > MAX_CANDIDATE_QUERY_BYTES { - return Err(miss(format!( - "candidate-filtered exact query has {} bytes, exceeding limit {MAX_CANDIDATE_QUERY_BYTES}", - restricted.len() + }); + } + if values.len() > MAX_CANDIDATE_VALUES { + return Err(miss(format!( + "candidate set has {} values, exceeding limit {MAX_CANDIDATE_VALUES}", + values.len() ))); - } - (restricted, true) - } else { - (query, false) - }; - let key = (language, query.clone(), at); - let cached = remote_cache.contains_key(&key); - let value = if let Some(value) = remote_cache.get(&key) { - value.clone() - } else { - let endpoint = match language { - QueryLanguage::PromQl => prometheus_endpoint - .ok_or_else(|| miss("Prometheus exact endpoint unavailable"))?, - QueryLanguage::MetricsQl => metricsql_endpoint - .ok_or_else(|| miss("VictoriaMetrics exact endpoint unavailable"))?, - QueryLanguage::ClickHouseSql => { - return Err(miss("ClickHouse exact subquery needs its SQL adapter")) } + let restricted = inject_candidate_matcher(&query, item_label, &values)?; + if restricted.len() > MAX_CANDIDATE_QUERY_BYTES { + return Err(miss(format!( + "candidate-filtered exact query has {} bytes, exceeding limit {MAX_CANDIDATE_QUERY_BYTES}", + restricted.len() + ))); + } + (restricted, true) + } else { + (query, false) }; - let url = format!("{}/api/v1/query", endpoint.trim_end_matches('/')); - let time = format!("{:.3}", at as f64 / 1000.0); - // Candidate sets can be large enough to exceed proxy URL limits; - // Prometheus accepts the instant-query parameters as an encoded - // form body. Static exact cuts keep their existing GET contract. - let request = if candidate_filtered { - client - .post(url) - .form(&[("query", query.as_str()), ("time", time.as_str())]) + let key = (language, query.clone(), at); + let cached = remote_cache.contains_key(&key); + let value = if let Some(value) = remote_cache.get(&key) { + value.clone() } else { - client - .get(url) - .query(&[("query", query.as_str()), ("time", time.as_str())]) + let endpoint = match language { + QueryLanguage::PromQl => prometheus_endpoint + .ok_or_else(|| miss("Prometheus exact endpoint unavailable"))?, + QueryLanguage::MetricsQl => metricsql_endpoint + .ok_or_else(|| miss("VictoriaMetrics exact endpoint unavailable"))?, + QueryLanguage::ClickHouseSql => { + return Err(miss("ClickHouse exact subquery needs its SQL adapter")) + } + }; + let url = format!("{}/api/v1/query", endpoint.trim_end_matches('/')); + let time = format!("{:.3}", at as f64 / 1000.0); + // Candidate sets can be large enough to exceed proxy URL limits; + // Prometheus accepts the instant-query parameters as an encoded + // form body. Static exact cuts keep their existing GET contract. + let request = if candidate_filtered { + client + .post(url) + .form(&[("query", query.as_str()), ("time", time.as_str())]) + } else { + client + .get(url) + .query(&[("query", query.as_str()), ("time", time.as_str())]) + }; + let response = request + .send() + .await + .map_err(|e| miss(format!("exact request failed: {e}")))?; + if !response.status().is_success() { + return Err(miss(format!("exact endpoint HTTP {}", response.status()))); + } + let body: serde_json::Value = response + .json() + .await + .map_err(|e| miss(format!("invalid exact response: {e}")))?; + let value = parse_result(&body, at)?; + remote_cache.insert(key, value.clone()); + value }; - let response = request - .send() - .await - .map_err(|e| miss(format!("exact request failed: {e}")))?; - if !response.status().is_success() { - return Err(miss(format!("exact endpoint HTTP {}", response.status()))); - } - let body: serde_json::Value = response - .json() - .await - .map_err(|e| miss(format!("invalid exact response: {e}")))?; - let value = parse_result(&body, at)?; - remote_cache.insert(key, value.clone()); - value - }; - prepared.insert( - (id, at), - PreparedLeaf { + Ok(PreparedLeaf { value, remote: true, remote_evaluations: usize::from(!cached), remote_rpcs: usize::from(!cached), - }, - ); + }) + } + .await; + match result { + Ok(leaf) => { + prepared.insert((id, at), leaf); + } + Err(error) => { + tracing::debug!(target: "asap_runtime_debug", query_id = %entry.query_id, + node_id = ?id, op = entry.nodes[&id].op_label(), evaluation_ms = at, %error, + "installed query exact leaf could not be prepared"); + return Err(error); + } + } } Ok(prepared) } diff --git a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs index 6c74edc7..ac9d970c 100644 --- a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs +++ b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs @@ -143,6 +143,7 @@ where for dependency in &dependencies { if let Some(id) = identities.get(dependency) { runtime.borrow_mut().stats.memo_hits += 1; + tracing::debug!(target: "asap_runtime_debug", query_id = %entry.query_id, node_id = ?dependency.0, evaluation_ms = dependency.1, "installed query node reused within request"); input_ids.push(*id); } else { if identities.len() >= 200_000 { @@ -623,6 +624,8 @@ impl Result> })) .await?; let values = values.iter().map(|v| v.value()).collect::>(); + let started = std::time::Instant::now(); + tracing::debug!(target: "asap_runtime_debug", node_id = ?self.id, evaluation_ms = self.time, op = self.node.op_label(), syntax = %self.node.log_syntax(), "installed query node started"); self.runtime .borrow_mut() .execute_node( @@ -633,7 +636,11 @@ impl Result> &self.dependencies, &context, ) + .inspect(|_| { + tracing::debug!(target: "asap_runtime_debug", node_id = ?self.id, op = self.node.op_label(), elapsed_us = started.elapsed().as_micros() as u64, "installed query node completed"); + }) .map_err(|error| { + tracing::warn!(node_id = ?self.id, op = self.node.op_label(), elapsed_us = started.elapsed().as_micros() as u64, %error, "installed query node failed"); *self.error.borrow_mut() = Some(error); physical::Error::Operator(format!( "query node {} at {} failed", diff --git a/data_plane/src/query_engines/asap_query_engine/logical_dag/native_values.rs b/data_plane/src/query_engines/asap_query_engine/logical_dag/native_values.rs index b0dd8feb..bda5f42e 100644 --- a/data_plane/src/query_engines/asap_query_engine/logical_dag/native_values.rs +++ b/data_plane/src/query_engines/asap_query_engine/logical_dag/native_values.rs @@ -584,6 +584,9 @@ mod tests { } } +#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.bind_vectors", query_id = %entry.query_id, evaluation_time_ms = at, + input_kind = "bound_promql_vector", bound_counter_state = entry.physical_vector_binding().is_some()), err)] pub(super) fn execute_vectors( entry: &asap_types::query_plan::QueryPlanEntry, at: u64, @@ -651,6 +654,8 @@ where ) } +#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.bind_stored_summary", query_id = %entry.query_id, evaluation_time_ms = at, plan_id, plan_version), err)] pub(in crate::query_engines::asap_query_engine) fn execute_stored( entry: &asap_types::query_plan::QueryPlanEntry, plan_id: u64, @@ -705,6 +710,8 @@ pub(in crate::query_engines::asap_query_engine) fn execute_stored( }) } +#[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "physical.execute", input_count, evaluation_time_ms = at, max_bytes), err)] fn execute_batches( program: &asap_physical_operators::physical_planner::CompiledPhysicalDag, max_bytes: u64, @@ -743,7 +750,11 @@ fn execute_batches( Operator::source(input.schema.clone(), vec![batch]).map_err(EngineError::from)?; sources.insert(input_id, Box::new(source) as Source<'_>); } - let graph = program.instantiate(sources).map_err(EngineError::from)?; + let graph = { + let _binding = tracing::debug_span!(target: "asap_runtime_debug", "physical_input_binding", + stage = "physical.bind_inputs", input_count = sources.len(), input_bytes, input_kind = "native_batch").entered(); + program.instantiate(sources).map_err(EngineError::from)? + }; let context = dag::RunContext::new( dag::Scope::Query { evaluation_time_ms: at_signed, diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index 0511e1aa..f0f65c42 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs @@ -237,6 +237,11 @@ impl QueryExecutionContext<'_> { /// Resolve exactly one compiler-bound materialization. This is the formal /// QueryPlan path: the validated stored output resolves to its definition's /// generation-scoped SID index. Metadata checks never broaden that set. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "sds.bound_read", + stored_output_id = binding.stored_output_reference.stored_output_id.as_u64(), + definition_id = ?binding.stored_output_reference.definition_id, + window_start_ms = self.t0_ms, window_end_ms = self.t1_ms), err(Debug, level = "debug"))] pub fn read_bound_materialization( &self, binding: &asap_types::query_plan::MaterializationBinding, @@ -275,6 +280,9 @@ impl QueryExecutionContext<'_> { )); } + tracing::debug!(target: "asap_runtime_debug", stage = "sds.validate_binding", + plan_id = catalog.plan_id, plan_version = catalog.plan_version, + "installed output and semantic identity validated"); let inventory_revision = self.index.summary_update_revision(); let query_range = asap_types::sds::HalfOpenTimeRange { start_ms: i64::try_from(self.t0_ms).map_err(|_| { @@ -342,6 +350,8 @@ impl QueryExecutionContext<'_> { .storage_handles_for_output(&binding.stored_output_reference); sids.sort_unstable(); sids.dedup(); + tracing::debug!(target: "asap_runtime_debug", stage = "sds.locate_records", + storage_handle_count = sids.len(), "bound output lookup completed"); let mut matched_metadata = 0usize; let mut by_group: BTreeMap, Vec> = BTreeMap::new(); let mut source_order = BTreeMap::new(); @@ -545,6 +555,9 @@ impl QueryExecutionContext<'_> { "summary input changed during read", )); } + tracing::debug!(target: "asap_runtime_debug", stage = "sds.validate_records", + group_count = result.len(), + "state format, applicable coverage and stable revision checks completed"); Ok(result) } 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 bd33de29..722cc676 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -985,11 +985,14 @@ impl SketchStore { self.install_catalog_outputs(catalog, outputs) } + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(plan_id = catalog.plan_id, plan_version = catalog.plan_version))] fn install_catalog_outputs( &self, catalog: Arc, outputs: BTreeMap, ) -> Result<(), String> { + tracing::debug!(target: "asap_runtime_debug", "summary store catalog installation started"); for (definition, output) in &outputs { output.validate().map_err(|e| e.to_string())?; if output.stored_output_id != *definition @@ -1084,12 +1087,24 @@ impl SketchStore { // producer cannot mutate a new generation before its receipt is rejected. let mut inventory = self.admission.write().unwrap(); if inventory.validate_publication(generation, coordinate, first_revision, revision)? { + tracing::debug!(target: "asap_runtime_debug", plan_id = generation.plan_id, + plan_version = generation.plan_version, + stage = "sds.publish", stored_output_id = coordinate.stored_output_id.as_u64(), + window_start_ms = coordinate.time_range.start_ms, + window_end_ms = coordinate.time_range.end_ms, revision, + "SDS summary publication replay acknowledged"); return Ok(()); } let series_id = persist(&SummaryPublicationWriter(self)).ok_or("summary state publication failed")?; inventory.record_series(generation, coordinate, series_id)?; inventory.acknowledge(generation, coordinate, revision)?; + tracing::debug!(target: "asap_runtime_debug", plan_id = generation.plan_id, + plan_version = generation.plan_version, sid = series_id, + stage = "sds.publish", stored_output_id = coordinate.stored_output_id.as_u64(), + window_start_ms = coordinate.time_range.start_ms, + window_end_ms = coordinate.time_range.end_ms, revision, + "SDS summary publication acknowledged"); self.admitted_mutations .fetch_add(1, std::sync::atomic::Ordering::SeqCst); let floor = coordinate @@ -1773,6 +1788,7 @@ impl SketchStore { start_unix_ms: u64, end_unix_ms: u64, ) -> Vec { + let started = std::time::Instant::now(); if self .incomplete_summary_lineages .get(&sid) @@ -1783,6 +1799,8 @@ impl SketchStore { }) }) { + tracing::debug!(target: "asap_runtime_debug", sid, start_unix_ms, end_unix_ms, + reason = "incomplete_summary_lineage", "SketchStore sketch range read skipped"); return Vec::new(); } // Result is keyed by the resolved label MAP so the in-memory tier @@ -1976,14 +1994,18 @@ impl SketchStore { // state; the guard is belt-and-suspenders). self.union_disk_parts_into(sid, start_unix_ms, end_unix_ms, &mut by_label_map); - by_label_map + let result: Vec<_> = by_label_map .into_iter() .map(|(label_values, samples)| SketchTimeSeries { sid, series_label_values: label_values, samples, }) - .collect() + .collect(); + tracing::debug!(target: "asap_runtime_debug", sid, start_unix_ms, end_unix_ms, + series_count = result.len(), elapsed_us = started.elapsed().as_micros() as u64, + "SketchStore sketch range read completed"); + result } /// Resolve the sorted group-by KEYS for a sid from its instance @@ -2218,6 +2240,7 @@ impl SketchStore { BTreeMap, BTreeMap>, )> { + let started = std::time::Instant::now(); // Key by the resolved label MAP (not `LabelValuesId`) so the // in-memory tier and the durable disk tier — which carry // independent intern spaces — union by label identity. Mirrors @@ -2278,7 +2301,11 @@ impl SketchStore { // of the range. In-memory wins on a window-end collision. self.union_disk_exact_agg_into(sid, start_unix_ms, end_unix_ms, &mut by_label_map); - by_label_map.into_iter().collect() + let result: Vec<_> = by_label_map.into_iter().collect(); + tracing::debug!(target: "asap_runtime_debug", sid, start_unix_ms, end_unix_ms, + series_count = result.len(), elapsed_us = started.elapsed().as_micros() as u64, + "SketchStore exact range read completed"); + result } /// Union the durable disk tier's exact-aggregation entries into @@ -3406,6 +3433,13 @@ impl SketchStore { accumulator.clone_boxed_core(), ), }; + tracing::debug!(target: "asap_runtime_debug", sid, + policy_fp = %output.policy_fp, + aggregation_type = ?agg_cfg.aggregation_type, + window_start_ms = output.start_timestamp, + window_end_ms = output.end_timestamp, + accepted, + "SketchStore precompute window append completed"); accepted.then_some(sid) } } @@ -3555,6 +3589,12 @@ impl SketchStore { /// /// Persisted metadata errors propagate to startup. Only an absent file is /// treated as a fresh store. + #[tracing::instrument( + level = "debug", + target = "asap_runtime_debug", + skip_all, + fields(stage = "sds.recover") + )] pub fn register_recovered_disk_series( &self, disk_path: &std::path::Path, @@ -3644,6 +3684,8 @@ impl SketchStore { registered += 1; } } + tracing::debug!(target: "asap_runtime_debug", registered, + "durable identities validated and eligible storage handles restored"); Ok(registered) } diff --git a/data_plane/src/storage_engines/sketch_db/index/native.rs b/data_plane/src/storage_engines/sketch_db/index/native.rs index a93bef09..71c28267 100644 --- a/data_plane/src/storage_engines/sketch_db/index/native.rs +++ b/data_plane/src/storage_engines/sketch_db/index/native.rs @@ -141,6 +141,9 @@ impl AggregateCore for NativeSummaryOutput { impl SketchStore { /// Publish a finalized native result only after the complete raw input /// cohort is durable. Existing publication fences prevent duplicate commits. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "sds.publish_native", start_ms = output.start_timestamp, + end_ms = output.end_timestamp, max_bytes), err)] pub fn publish_native_summary_output( &self, resolver: &SeriesIdResolver, @@ -219,6 +222,10 @@ impl SketchStore { /// Resolve the installed output/group prefix without minting a new handle, /// then read exactly the requested immutable window and validate its format. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "sds.read_native", plan_version = address.plan_version, + start_ms = address.window.start_ms, end_ms = address.window.end_ms, + max_bytes), err)] pub fn read_native_summary_output( &self, resolver: &SeriesIdResolver, @@ -262,6 +269,8 @@ impl SketchStore { /// A complete native batch is stored atomically under one global address. /// Logical grouping remains in the batch; reads never allocate a resolver ID. + #[tracing::instrument(level = "debug", target = "asap_runtime_debug", skip_all, + fields(stage = "sds.read_bound_native", plan_version = address.plan_version, stored_output_id = ?reference.stored_output_id, start_ms = address.window.start_ms, end_ms = address.window.end_ms, max_bytes), err)] pub fn read_bound_native_summary( &self, address: &asap_types::sds::StoredSummaryKey, diff --git a/data_plane/src/storage_engines/sketch_db/sds.rs b/data_plane/src/storage_engines/sketch_db/sds.rs index 58ea8eb6..911c4dbf 100644 --- a/data_plane/src/storage_engines/sketch_db/sds.rs +++ b/data_plane/src/storage_engines/sketch_db/sds.rs @@ -201,6 +201,11 @@ impl SummaryDescriptorRegistry { snapshot_sha256: reference.snapshot_sha256, }); *self.authoritative_catalog.write().unwrap() = Some((catalog, generation, outputs)); + tracing::info!( + plan_id = reference.plan_id, + plan_version = reference.plan_version, + "storage descriptor catalog installed" + ); Ok(()) } @@ -311,6 +316,19 @@ impl SummaryDescriptorRegistry { } }; + let catalog_generation = authoritative + .as_ref() + .map(|(_, generation, _)| generation.clone()); + tracing::debug!(target: "asap_runtime_debug", sid = metadata.storage_handle, + stage = "sds.bind_storage_handle", + stored_output_id = stored_output_reference.as_ref().map(|r| r.stored_output_id.as_u64()), + definition_id = ?stored_output_reference.as_ref().map(|r| &r.definition_id), + policy_fp = %metadata.policy_fp, + summary_descriptor_hash = format_args!("{:016x}", xxhash_rust::xxh64::xxh64(summary_descriptor.id().canonical().as_bytes(), 0)), + data_descriptor_hash = format_args!("{:016x}", xxhash_rust::xxh64::xxh64(data_id.canonical().as_bytes(), 0)), + plan_id = catalog_generation.as_ref().map(|g| g.plan_id), + plan_version = catalog_generation.as_ref().map(|g| g.plan_version), + "SDS descriptor binding resolved"); Ok(SdsBinding { stored_output_reference, metadata: Arc::new(metadata), diff --git a/docs/runtime-debugging.md b/docs/runtime-debugging.md new file mode 100644 index 00000000..731ca368 --- /dev/null +++ b/docs/runtime-debugging.md @@ -0,0 +1,125 @@ +# Runtime diagnostics along the execution path + +Audience: developers tracing an installed plan from selection to a result. +The [architecture](design_docs/asapplanner-integration.md) and +[SDS contract](design_docs/summary-catalog-sds-architecture.md) define ownership; +diagnostics observe these decisions and never select replacement computations. + +## Enable and correlate + +Set `RUST_LOG=info,asap_runtime_debug=debug` on controller and backend. Use +`RUST_LOG=info` to hide these events. Logs contain timestamps, target, file/line, +span context and event fields. Controller output goes to stdout; backend output +also goes to `/query_engine.log` without terminal colors. + +Correlate by `plan_id` and `plan_version`, then `query_id` and the process-local +`call_id`. A call ID is not a distributed trace ID. For state reads and writes, +use `stored_output_id` for the deployed producer and `definition_id` for its +meaning. A `sid`/storage handle is only a local row locator. Descriptor hashes +are diagnostics, not semantic IDs or authorization tokens. + +## Follow the architecture in order + +| Step | Owner and operation | Diagnostic evidence | +| --- | --- | --- | +| 1 | Planner selects the Logical Post-ASAP DAG and maintenance requirements | `planner.select` entry and selection errors. `deployment.candidate_inventory`, `deployment.candidate_evaluation` and `deployment.candidate_selected` distinguish compilation/pricing failures from selection within the bounded inventory. Backend spans bracket Planner calls; they do not instrument every internal Planner optimization. | +| 2 | Planner compiles physical operators, dependencies and typed boundaries | Inspect the selected DAG artifact alongside compilation spans. Do not interpret a backend adapter node as a second Planner physical node. | +| 3 | Backend Deployment Plan Compiler binds sources, stored outputs and deployment policy | `deployment.bind`, transmission construction and publication spans; plan generation and query counts. The backend does not reselect operators. | +| 4 | Backend validates, stages and activates one generation | HTTP staging/activation and summary-catalog installation events. Activation does not imply state readiness. | +| 5 | Precompute engine resolves inputs and invokes the shared executor | Worker and precompute DAG spans: node/dependency IDs, execution timing, window, sink and inclusive node duration. Reused inputs and already-committed sinks have separate events. | +| 6 | SummaryStore admits and publishes state | `sds.bind_storage_handle` includes both bound identities; `sds.publish` includes output, plan generation, window and revision. Replay acknowledgment is distinguished from a new publication. | +| 7 | Query engine selects the installed QueryPlan | Query call/preparation events and the selected query DAG. A bound query does not search for semantically similar outputs. | +| 8 | Bound read validates identity, locates records and validates eligibility | `sds.bound_read` carries output ID, definition ID and requested range. `sds.validate_binding` checks the installed reference; `sds.locate_records` reports candidate storage handles; `sds.validate_records` follows format, applicable coverage and stable-revision checks. Failures terminate the read span. | +| 9 | Shared operators compute the result; backend serves or applies installed fallback | Node start/completion/failure, memo hits, query completion and remote/RPC counts. Match fallback evidence to the request rather than assuming a successful response ran locally. | +| 10 | SummaryStore recovers persisted state on restart | `sds.recover` brackets metadata replay and reports eligible restored handles. Existing warnings distinguish provenance mismatch, missing identity and unsupported state. A successful disk read alone does not establish semantic eligibility. | + +Step 8 follows the actual implementation: it validates the installed reference +before lookup, then validates concrete state. Lookup is scoped to the selected +output; equal definition IDs never authorize switching to another deployed +output. Coverage rules depend on the selected state family: additive panes need +contiguous coverage, while supported counter state carries sample endpoints. +No diagnostic introduces a universal no-gap rule for every operator. + +## Timing and node identity + +Node durations are inclusive. Parent and child intervals can overlap and shared +producers can serve multiple consumers, so do not sum them as CPU time. These +logs do not provide exclusive kernel, allocation or lock timing. Pair them with +#766's profiling workflow for overhead attribution. + +Backend QueryPlan adapter IDs and Planner DAG IDs occupy different namespaces. +The installed bindings connect their sinks; retain the plan artifact when +following an edge across that boundary. Build, merge and readout are operators, +not fixed precompute/query phases. Interpret placement from the selected DAG. + +## Detail and limitations + +For storage and worker details, add +`data_plane::precompute_engine=debug,data_plane::storage_engines=debug`. +The diagnostic target omits samples, complete query text and full fallback +expressions from its new structured fields. Existing errors or worker fields can +still contain query text or group labels. Logs are text events, not a versioned +JSON API. Missing a completion event can indicate an interrupted process; it +is not proof that work committed. Publication receipts and recovered state remain +the source of truth. + +A missing candidate is not evidence that its placement costs more. The #728 +inventory now includes grouped Rate with query-time Sum and precomputed Sum; +compare their admission and resource costs before interpreting selection. Search +coverage is scoped to the declared inventory, never a claim of exhaustive +physical optimization. + +### Native Rate aggregation stages + +`physical.maintenance.execute` runs the retained Planner maintenance program over +complete bound counter windows. It covers per-series Rate followed by grouped +Sum or CMS/CountSketch heap construction. `sds.publish_native` publishes one +atomic grouped batch for that window. Overlapping complete windows are separate +executions and records. + +On a bound query, `sds.read_bound_native` resolves the installed stored output; +`physical.bind_stored_summary` binds the recovered batch. Query-time candidates +instead use `physical.bind_vectors` for the per-series Rate vector. Both call +`physical.execute` for the retained query graph. These spans expose the selected +placement; they never move an operator between query and maintenance phases. +The finite complete-input barrier is required for precomputed grouped outputs; +an elapsed-time event does not prove the population is complete. + +### Installed SQL physical DAG stages + +SQL plans compile native operators before installation (`physical.compile_install`). +Activation and serving validate the persisted program (`physical.recover_validate`); +this stage decodes physical operators and does not lower logical expressions. +`physical.bind_inputs` attaches the resolved external/SDS batches, and +`physical.execute` covers the request's physical execution path. These debug spans +include the query identity, root or input count, and errors without logging batches +or sketch payloads. They use the existing `asap_runtime_debug` target. + +These stages describe the installed SQL path. They do not establish that the +remaining PromQL candidate-selection and maintenance paths have migrated to the +same physical handoff. + +### Native SDS publication and recovery + +`sds.publish_native` covers complete-cohort validation and immutable publication; +`sds.read_native` covers the installed output lookup and native decoding. Their +fields include the requested window and byte budget; read spans also include the +plan version. Group labels and summary payloads are not logged. Successful return +from an idempotent publication can mean the output was already committed; the +publication receipt remains authoritative. + +These spans instrument the native storage API, including its recovery E2E. They +do not imply that all PromQL maintenance candidates use that API yet. + +Current-series TopK uses these same stages: installation compiles the Planner +readout program, recovery validates the persisted program, binding supplies the +complete eligible population, and native execution performs ranking and limiting. +These spans use `input_kind=current_series_snapshot`; label sets and sample +values are not recorded. They describe the exact ranking path, not heap storage +or request-time physical compilation. + +Snapshot heap discovery records `planner.physical_candidate` while Planner +compiles the candidate above its population boundary. Deployment admission and +pricing follow separately; an emitted physical graph does not imply that its +accuracy evidence was admitted. The same recovery, binding and execution spans +cover the installed CountSketch heap and exact ranking programs.