Skip to content

Add runtime debug logs across plan publication and query execution #756

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zzylol wants to merge 1 commit into
base: refactor/query-plan-dag-execution
Choose a base branch
from
Open

Add runtime debug logs across plan publication and query execution #756

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions control_plane/src/backend_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)))?;
Expand All @@ -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 {
Expand All @@ -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!(
Expand All @@ -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
Expand All @@ -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 {
Expand Down
84 changes: 80 additions & 4 deletions control_plane/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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 =
Expand Down Expand Up @@ -266,11 +278,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();
Expand All @@ -282,16 +299,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(),
Expand All @@ -303,6 +341,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}"),
Expand All @@ -312,11 +352,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
Expand All @@ -327,17 +370,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;
Expand All @@ -347,12 +400,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(),
Expand All @@ -366,6 +433,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}"),
Expand All @@ -374,6 +443,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,
Expand Down
11 changes: 11 additions & 0 deletions control_plane/src/physical/compiler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -381,11 +381,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);
}
Expand Down Expand Up @@ -957,12 +961,16 @@ impl PhysicalPlanCompiler {
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", "physical plan compiler entered");
if let Some(data) = &request.data_workload {
data.validate()
.map_err(|error| CompileError::Snapshot(error.to_string()))?;
Expand Down Expand Up @@ -2158,13 +2166,16 @@ 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>>,
evidence: &HashMap<String, TopKMembershipEvidence>,
exact_costs: &HashMap<String, Vec<ExactCompositionCostEvidence>>,
erp: Option<&super::erp::ErpPlanningInput>,
) -> Result<Vec<serde_json::Value>, CompileError> {
tracing::debug!(target: "asap_runtime_debug", "logical root selection started");
let mut traces = Vec::new();
if roots.len() != queries.len() {
return Err(CompileError::Snapshot(
Expand Down
Loading
Loading