From 64ca8f97ce030f7eae268756d21942ba17dd2d8e Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 18:54:10 +0100 Subject: [PATCH] fix(observability): dedicated OTLP runtime for sync training binaries The batch span processor needs a tokio runtime for gRPC transport and periodic flush. Async services already have one via #[tokio::main], but sync training binaries (hyperopt, train, evaluate) don't. Previous approach (making binaries async with #[tokio::main]) caused "Cannot start a runtime from within a runtime" panics because the ML crate's internal code creates its own tokio runtimes for block_on(). New approach: build_otel_tracer() detects runtime context via Handle::try_current(). If absent, it creates a dedicated 1-worker multi-thread runtime stored in a process-lifetime OnceLock. The worker thread actively polls the OTLP batch export task. Reverts training binaries to sync fn main() so internal runtime creation (hyperopt adapters, DQN/PPO trainers) continues working as before. Co-Authored-By: Claude Opus 4.6 --- .../src/observability/tracing_config.rs | 35 +++++++++++++++++++ crates/ml/examples/evaluate_baseline.rs | 3 +- crates/ml/examples/evaluate_supervised.rs | 3 +- crates/ml/examples/hyperopt_baseline_rl.rs | 3 +- .../examples/hyperopt_baseline_supervised.rs | 3 +- crates/ml/examples/train_baseline_rl.rs | 3 +- .../ml/examples/train_baseline_supervised.rs | 3 +- 7 files changed, 41 insertions(+), 12 deletions(-) diff --git a/crates/common/src/observability/tracing_config.rs b/crates/common/src/observability/tracing_config.rs index cd37439c8..760ff791d 100644 --- a/crates/common/src/observability/tracing_config.rs +++ b/crates/common/src/observability/tracing_config.rs @@ -27,6 +27,18 @@ use opentelemetry_sdk::{ trace::{Sampler, SdkTracerProvider}, Resource, }; +use std::sync::OnceLock; + +/// Dedicated tokio runtime for the OTLP batch exporter in sync binaries. +/// +/// The batch span processor uses `tokio::spawn` internally for periodic flushing +/// and the tonic gRPC transport needs a reactor. Async services (with +/// `#[tokio::main]`) already provide this; sync training binaries need a +/// dedicated runtime that lives for the process lifetime. +/// +/// Uses a multi-thread runtime with 1 worker thread so the export loop is +/// actively polled without the caller needing to drive it. +static OTEL_RUNTIME: OnceLock> = OnceLock::new(); /// Configuration for OpenTelemetry tracing. /// @@ -82,6 +94,29 @@ pub fn build_otel_tracer( return None; } + // The batch exporter needs a tokio runtime for gRPC transport and periodic + // flush. Async services already have one; sync training binaries get a + // dedicated single-worker runtime that lives for the process lifetime. + let _runtime_guard = if tokio::runtime::Handle::try_current().is_ok() { + None + } else { + let rt_opt = OTEL_RUNTIME.get_or_init(|| { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .thread_name("otel-export") + .enable_all() + .build() + .ok() + }); + match rt_opt.as_ref() { + Some(rt) => Some(rt.enter()), + None => { + eprintln!("OTLP runtime creation failed (non-fatal)"); + return None; + } + } + }; + // Build an explicit tonic Channel that respects the URI scheme (http vs https). // This prevents TLS negotiation on plaintext http:// endpoints like Tempo. let channel = match tonic::transport::Channel::from_shared(config.otlp_endpoint.clone()) { diff --git a/crates/ml/examples/evaluate_baseline.rs b/crates/ml/examples/evaluate_baseline.rs index fd2a34264..5b1953a44 100644 --- a/crates/ml/examples/evaluate_baseline.rs +++ b/crates/ml/examples/evaluate_baseline.rs @@ -639,8 +639,7 @@ fn run_sanity_checks( // --------------------------------------------------------------------------- #[allow(clippy::cognitive_complexity, clippy::too_many_lines)] -#[tokio::main(flavor = "current_thread")] -async fn main() -> Result<()> { +fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/evaluate_supervised.rs b/crates/ml/examples/evaluate_supervised.rs index fa59fc2f0..ae93f43a3 100644 --- a/crates/ml/examples/evaluate_supervised.rs +++ b/crates/ml/examples/evaluate_supervised.rs @@ -532,8 +532,7 @@ fn evaluate_fold( // --------------------------------------------------------------------------- #[allow(clippy::cognitive_complexity, clippy::too_many_lines)] -#[tokio::main(flavor = "current_thread")] -async fn main() -> Result<()> { +fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/hyperopt_baseline_rl.rs b/crates/ml/examples/hyperopt_baseline_rl.rs index ac47b9db8..86ae022a9 100644 --- a/crates/ml/examples/hyperopt_baseline_rl.rs +++ b/crates/ml/examples/hyperopt_baseline_rl.rs @@ -261,8 +261,7 @@ fn run_ppo_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device) } #[allow(clippy::cognitive_complexity)] -#[tokio::main(flavor = "current_thread")] -async fn main() -> Result<()> { +fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/hyperopt_baseline_supervised.rs b/crates/ml/examples/hyperopt_baseline_supervised.rs index 3a76d2692..742338b02 100644 --- a/crates/ml/examples/hyperopt_baseline_supervised.rs +++ b/crates/ml/examples/hyperopt_baseline_supervised.rs @@ -543,8 +543,7 @@ const VALID_MODELS: &[&str] = &[ ]; #[allow(clippy::cognitive_complexity)] -#[tokio::main(flavor = "current_thread")] -async fn main() -> Result<()> { +fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/train_baseline_rl.rs b/crates/ml/examples/train_baseline_rl.rs index 72a912cca..3dac5db35 100644 --- a/crates/ml/examples/train_baseline_rl.rs +++ b/crates/ml/examples/train_baseline_rl.rs @@ -827,8 +827,7 @@ fn run_training(args: &Args) -> Result> { // Main // --------------------------------------------------------------------------- -#[tokio::main(flavor = "current_thread")] -async fn main() -> Result<()> { +fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability( diff --git a/crates/ml/examples/train_baseline_supervised.rs b/crates/ml/examples/train_baseline_supervised.rs index 5203bfad9..ee8c54c16 100644 --- a/crates/ml/examples/train_baseline_supervised.rs +++ b/crates/ml/examples/train_baseline_supervised.rs @@ -796,8 +796,7 @@ fn run_training(args: &Args) -> Result> { // Main // --------------------------------------------------------------------------- -#[tokio::main(flavor = "current_thread")] -async fn main() -> Result<()> { +fn main() -> Result<()> { // Initialize tracing with optional OTLP export to Tempo let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); if let Err(e) = common::observability::init_observability(