Add runtime debug logs across plan publication and query execution

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<serde_json::Value>,
         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 {

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,

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<asap_types::PolicyFingerprint, RuntimeRulePolicy>,
 ) -> Result<TransmissionPlan, TransmissionPlanError> {
+    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 compiled = asap_physical_operators::physical_planner::promql_rows::compile_current_series_readout(&root)
                     .or_else(|_| asap_physical_operators::physical_planner::promql_rows::compile_rate_ranking(&root).map(|(_, program)| program));
                 if let Ok(physical) = asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_aggregation(&root) {
@@ -1308,12 +1315,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<CompiledPhysicalPlan, CompileError> {
+        tracing::debug!(target: "asap_runtime_debug", stage = "deployment.bind", "backend deployment compiler entered");
         environment
             .dataset_identity
             .validate()
@@ -2718,6 +2729,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<Rc<QueryExpr>>,
@@ -2767,6 +2780,7 @@ fn logical_roots_and_candidates(
     now_ms: u64,
     mut candidates_out: Option<&mut Vec<Vec<(usize, Rc<SummaryNode>)>>>,
 ) -> Result<Vec<serde_json::Value>, 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(

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<SummaryNode>,

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,