From e93f6b6d699585585cca975cd4fa69eafac70410 Mon Sep 17 00:00:00 2001 From: Cameron Cordes Date: Tue, 29 Sep 2026 12:15:24 -0400 Subject: [PATCH] Otel: unify tracing subscriber across dev+release; env-gated OTLP export (#138) Install ONE `tracing` subscriber in both build profiles via `otel::init_tracing()` (previously release-only, so spans were a no-op in dev). Behaviour is now environment-driven, not build-profile-driven: - OTLP span export happens only when `OTLP_OTLS_ENDPOINT` is set (release / staging). Unset -> spans are still created with valid contexts (so outgoing `traceparent` injection keeps working) but export nowhere; the console `fmt` layer is what shows them. - `tracing_subscriber::registry()` + `EnvFilter` (RUST_LOG) + `OpenTelemetryLayer` + console `fmt` layer, installed in both builds. - Set the global W3C `TraceContextPropagator`: required by the `tracing-actix-web` request middleware (#139) and by the outgoing traceparent helpers, both of which read the *global* propagator. - `tracing-log` bridge stays OFF (`default-features = false`) so the console fmt layer prints tracing spans only; `log` records keep their existing path (env_logger in debug, OTel logs in release) without double-printing. Deps: + tracing, tracing-subscriber (env-filter/fmt/ansi), tracing-opentelemetry 0.32 (pins opentelemetry 0.31 + otel_sdk 0.31, no version bump); opentelemetry_sdk gains the `testing` feature. Add `tests/otel_propagation.rs`: the regression matrix from spike #133, pinning the invariant the whole migration rests on -- an OTel context does NOT cross a bare `tokio::spawn` on a multi-thread runtime (S2/S4 orphan onto a fresh trace) but DOES with an explicit parent carry across the spawn (S6 `.instrument()`, S7 `span.enter()`). This is why every spawn in the codebase must carry the parent. clippy --all-targets clean; 690 lib + 775 integration + regression + 4 doc-tests pass. --- Cargo.lock | 89 +++++++++++ Cargo.toml | 8 +- src/main.rs | 6 +- src/otel.rs | 58 ++++++-- tests/otel_propagation.rs | 300 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 447 insertions(+), 14 deletions(-) create mode 100644 tests/otel_propagation.rs diff --git a/Cargo.lock b/Cargo.lock index 8dc6f91..a781fdd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2123,6 +2123,9 @@ dependencies = [ "tempfile", "tokio", "tokio-util", + "tracing", + "tracing-opentelemetry", + "tracing-subscriber", "tzf-rs", "urlencoding", "uuid", @@ -2440,6 +2443,15 @@ dependencies = [ "tendril", ] +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "maybe-rayon" version = "0.1.1" @@ -2590,6 +2602,15 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0676bb32a98c1a483ce53e500a81ad9c3d5b3f7c920c28c24e9cb0980d0b5bc8" +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num-bigint" version = "0.4.6" @@ -3785,6 +3806,15 @@ dependencies = [ "digest", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "1.3.0" @@ -4064,6 +4094,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "tiff" version = "0.11.3" @@ -4338,6 +4377,50 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-opentelemetry" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ac28f2d093c6c477eaa76b23525478f38de514fa9aeb1285738d4b97a9552fc" +dependencies = [ + "js-sys", + "opentelemetry", + "smallvec", + "tracing", + "tracing-core", + "tracing-log", + "tracing-subscriber", + "web-time", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "thread_local", + "tracing", + "tracing-core", ] [[package]] @@ -4486,6 +4569,12 @@ version = "0.15.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e8257fbc510f0a46eb602c10215901938b5c2a7d5e70fc11483b1d3c9b5b18c" +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "vcpkg" version = "0.2.15" diff --git a/Cargo.toml b/Cargo.toml index 2a65287..d84ef80 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,13 +39,19 @@ rayon = "1.5" path-absolutize = "3.1" log = "0.4" env_logger = "0.11.5" +# default-features = false keeps the `tracing-log` bridge OFF so the console +# fmt layer prints tracing spans only; `log` records stay on their existing +# path (env_logger in debug, OTel logs in release) without double-printing. +tracing = "0.1" +tracing-subscriber = { version = "0.3", default-features = false, features = ["std", "registry", "env-filter", "fmt", "ansi"] } +tracing-opentelemetry = "0.32" actix-web-prom = "0.9.0" prometheus = "0.13" lazy_static = "1.5" anyhow = "1.0" rand = "0.8.5" opentelemetry = { version = "0.31.0", features = ["default", "metrics", "tracing"] } -opentelemetry_sdk = { version = "0.31.0", features = ["default", "rt-tokio-current-thread", "metrics"] } +opentelemetry_sdk = { version = "0.31.0", features = ["default", "rt-tokio-current-thread", "metrics", "testing"] } opentelemetry-otlp = { version = "0.31.0", features = ["default", "metrics", "tracing", "grpc-tonic"] } opentelemetry-stdout = "0.31.0" opentelemetry-appender-log = "0.31.0" diff --git a/src/main.rs b/src/main.rs index 3a7200b..0b7bc43 100644 --- a/src/main.rs +++ b/src/main.rs @@ -111,7 +111,10 @@ fn main() -> std::io::Result<()> { let system = actix::System::new(); system.block_on(async { - // Just use basic logger when running a non-release build + // One `tracing` subscriber in every build: env-driven, so dev prints + // spans to the console and release exports them over OTLP. The `log` + // pipeline stays separate — env_logger in debug, OTel logs in release. + otel::init_tracing(); #[cfg(debug_assertions)] { env_logger::init(); @@ -119,7 +122,6 @@ fn main() -> std::io::Result<()> { #[cfg(not(debug_assertions))] { otel::init_logs(); - otel::init_tracing(); } // AppState construction loads (and seeds if needed) the libraries diff --git a/src/otel.rs b/src/otel.rs index 1eebbaf..9384abf 100644 --- a/src/otel.rs +++ b/src/otel.rs @@ -3,9 +3,11 @@ use actix_web::http::header::HeaderMap; use opentelemetry::global::{BoxedSpan, BoxedTracer}; use opentelemetry::propagation::TextMapPropagator; use opentelemetry::trace::{ - Span, SpanBuilder, SpanContext, SpanKind, Status, TraceContextExt, Tracer, + Span, SpanBuilder, SpanContext, SpanKind, Status, TraceContextExt, Tracer, TracerProvider, }; use opentelemetry::{Context, KeyValue, global}; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::util::SubscriberInitExt; use opentelemetry_appender_log::OpenTelemetryLogBridge; use opentelemetry_otlp::WithExportConfig; use opentelemetry_sdk::Resource; @@ -16,8 +18,26 @@ pub fn global_tracer() -> BoxedTracer { global::tracer("image-server") } -#[allow(dead_code)] +/// Install the process-wide `tracing` subscriber plus the OTel span pipeline. +/// +/// Called in BOTH debug and release builds (the old split made spans a no-op +/// in dev). Behaviour is driven by the environment, not the build profile: +/// - `OTLP_OTLS_ENDPOINT` set -> spans are batch-exported over OTLP (the +/// release / staging path). +/// - unset -> spans are created with valid contexts (so +/// outgoing `traceparent` injection still works) but exported nowhere; the +/// console `fmt` layer is what shows them. +/// +/// The console `fmt` layer is always present, so a release build run locally +/// also prints spans to stdout. `log` records are NOT routed through this +/// subscriber (the `tracing-log` bridge is off) — they keep their existing +/// path: `env_logger` in debug, OTel logs in release. pub fn init_tracing() { + // W3C propagator for the whole process: the request middleware + // (tracing-actix-web) and the outgoing `traceparent` helpers both read + // the *global* propagator, so it must be set before any request lands. + global::set_text_map_propagator(TraceContextPropagator::new()); + let resources = Resource::builder() .with_attributes([ KeyValue::new("service.name", "image-server"), @@ -25,18 +45,34 @@ pub fn init_tracing() { ]) .build(); - let span_exporter = opentelemetry_otlp::SpanExporter::builder() - .with_tonic() - .with_endpoint(std::env::var("OTLP_OTLS_ENDPOINT").unwrap()) - .build() - .unwrap(); + let mut provider_builder = + opentelemetry_sdk::trace::SdkTracerProvider::builder().with_resource(resources); - let tracer_provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() - .with_batch_exporter(span_exporter) - .with_resource(resources) - .build(); + if let Ok(endpoint) = std::env::var("OTLP_OTLS_ENDPOINT") { + let span_exporter = opentelemetry_otlp::SpanExporter::builder() + .with_tonic() + .with_endpoint(endpoint) + .build() + .expect("OTLP span exporter"); + provider_builder = provider_builder.with_batch_exporter(span_exporter); + } + let tracer_provider = provider_builder.build(); + // The rest of the codebase resolves spans through the global provider + // (`global_tracer()`, `global::tracer("db")`), so install it there; the + // subscriber's OTel layer uses the same provider's tracer. + let tracer = tracer_provider.tracer("image-server"); global::set_tracer_provider(tracer_provider); + + tracing_subscriber::registry() + .with( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with(tracing_opentelemetry::OpenTelemetryLayer::new(tracer)) + .with(tracing_subscriber::fmt::layer().with_target(false)) + .try_init() + .expect("failed to install tracing subscriber"); } #[allow(dead_code)] diff --git a/tests/otel_propagation.rs b/tests/otel_propagation.rs new file mode 100644 index 0000000..f0e7e47 --- /dev/null +++ b/tests/otel_propagation.rs @@ -0,0 +1,300 @@ +//! Regression test for the tracing migration (#134 / map #135). +//! +//! Origin: spike #133 (go/no-go). It pins the one non-obvious property of +//! `tracing` + `tracing-opentelemetry 0.32` that the whole migration rests on: +//! a span made inside a `tokio::spawn`-ed task nests under the pre-spawn +//! parent (same trace_id, parent_span_id link) **only** when the parent span +//! is explicitly carried across the spawn. Bare `#[instrument]` on its own +//! does NOT do that on a multi-thread runtime — the child orphans onto a fresh +//! trace. That is why every `tokio::spawn` in the codebase must carry the +//! parent span (`.instrument()` or an `enter()` guard); this test fails if +//! that guarantee regresses. +//! +//! Does `tracing` + `tracing-opentelemetry 0.32` propagate OTel context across +//! `.await` AND into a `tokio::spawn`-ed task via `#[instrument]`, so a span +//! made inside the spawned task nests under the pre-spawn parent? +//! +//! Key subtlety (verified by `cargo expand` on tracing-attributes): for an +//! `async fn`, `#[instrument]` generates the span-creation INSIDE the future +//! body — the span is created at FIRST POLL, not at call time. For a +//! `tokio::spawn`ed task the first poll always happens on a runtime worker +//! thread, so the parent is looked up on a thread where it was never +//! entered. This holds for BOTH spawn shapes: +//! +//! - `tokio::spawn(child())` and +//! - `tokio::spawn(async move { child().await })` (the production shape, +//! see `insight_chat::chat_turn_stream_inner`) +//! +//! Each scenario is driven with `rt.block_on` so the test body always runs on +//! the calling thread while spawned tasks run on (multi-thread) workers, +//! making cross-thread behavior deterministic. +//! +//! VERDICT (Spike #133): ADJUST. Plain `#[instrument]` does NOT propagate +//! across `tokio::spawn` on a multi-thread runtime (S2/S4 export orphaned +//! spans: fresh trace_id, parent_span_id = 0000000000000000) because the +//! span of an `#[instrument]`ed `async fn` is created at FIRST POLL — always +//! on a worker thread for a spawned task — and tracing span context is +//! thread-local. The fix is explicit context carrying into the task: +//! `tokio::spawn(block.instrument(parent_span))` or a `span.enter()` guard at +//! the top of the block (S6/S7 both propagate). +//! +//! GOTCHA discovered while building this: `SubscriberInitExt::set_default` is +//! THREAD-LOCAL — spans from any other thread (plain std or tokio workers) +//! are silently dropped. The subscriber must be installed process-globally +//! (`.try_init()` / `.init()`, i.e. `dispatcher::set_global_default`) for +//! worker threads to see it at all. (The app's existing `init_tracing` uses +//! `.try_init()`, so this is already satisfied in production.) + +use opentelemetry::trace::TracerProvider; +use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider, SpanData}; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::registry::Registry; +use tracing_subscriber::util::SubscriberInitExt; +use tracing::Instrument; + +const CHILD: &str = "child_in_spawned_task"; + +#[tracing::instrument] +async fn child_in_spawned_task() { + // Yield so the remainder of the body really runs as a spawned task. + tokio::task::yield_now().await; +} + +struct Outcome { + propagated: bool, + child_trace: String, + parent_trace: String, + child_parent_id: String, + parent_span_id: String, +} + +/// Within a scenario's slice there is exactly one parent and one child; +/// identify the parent as the span that is not the child. +fn collect(spans: &[SpanData]) -> Option { + let child = spans.iter().find(|s| s.name == CHILD)?; + let parent = spans.iter().find(|s| s.name != CHILD)?; + Some(Outcome { + propagated: child.span_context.trace_id() == parent.span_context.trace_id() + && child.parent_span_id == parent.span_context.span_id(), + child_trace: child.span_context.trace_id().to_string(), + parent_trace: parent.span_context.trace_id().to_string(), + child_parent_id: child.parent_span_id.to_string(), + parent_span_id: parent.span_context.span_id().to_string(), + }) +} + +fn run_scenario( + exporter: &InMemorySpanExporter, + make_rt: fn() -> tokio::runtime::Runtime, + body: F, +) -> Option +where + F: FnOnce() -> std::pin::Pin + Send>>, +{ + let before = exporter.get_finished_spans().map(|v| v.len()).unwrap_or(0); + let rt = make_rt(); + rt.block_on(async { + let parent = tracing::info_span!("scenario_parent"); + let _entered = parent.enter(); + body().await; + }); + let all = exporter.get_finished_spans().expect("in-memory exporter"); + collect(&all[before..]) +} + +fn current_thread_rt() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime") +} + +fn multi_thread_rt() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("multi-thread runtime") +} + +#[test] +fn tracing_opentelemetry_spawn_propagation_matrix() { + // One process-global subscriber for the whole matrix (set_global_default + // is one-shot per process; the scenarios run sequentially). + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let tracer = provider.tracer("spike-133"); + let subscriber = Registry::default().with(tracing_opentelemetry::OpenTelemetryLayer::new(tracer)); + subscriber + .try_init() + .expect("global subscriber not yet set"); + + // S1: single-thread runtime, instrumented call inside the spawned block. + // Control: spawn and parent share one thread, so the thread-local active + // span survives into the child's first poll. + let s1 = run_scenario(&exporter, current_thread_rt, || { + Box::pin(async { + let h = tokio::spawn(async move { + child_in_spawned_task().await; + }); + h.await.unwrap(); + }) + }); + + // S2: multi-thread runtime, instrumented call inside the spawned block — + // THE PRODUCTION SHAPE (`chat_turn_stream_inner`'s + // `tokio::spawn(async move { svc.run_streaming_turn(...).await })`). + // The first poll of the instrumented call happens on a worker thread + // that never had the parent entered. + let s2 = run_scenario(&exporter, multi_thread_rt, || { + Box::pin(async { + let h = tokio::spawn(async move { + child_in_spawned_task().await; + }); + h.await.unwrap(); + }) + }); + + // S4: multi-thread runtime, instrumented call at the spawn call site — + // `tokio::spawn(child_in_spawned_task())`. The span is created on the + // calling thread where the parent is active. + let s4 = run_scenario(&exporter, multi_thread_rt, || { + Box::pin(async { + let h = tokio::spawn(child_in_spawned_task()); + h.await.unwrap(); + }) + }); + + // S5: no spawn at all — plain `.await` on the same task. Sanity control + // that the layer + exporter wiring works. + let s5 = run_scenario(&exporter, multi_thread_rt, || { + Box::pin(async { + child_in_spawned_task().await; + }) + }); + + // S6: THE FIX — capture the active parent span before the spawn and carry + // it into the task with `.instrument()`. `.instrument` re-enters the span + // on every poll (including the first poll on the worker thread), so the + // `#[instrument]` child created there nests under the captured parent. + let s6 = run_scenario(&exporter, multi_thread_rt, || { + Box::pin(async { + let span = tracing::Span::current(); + let h = tokio::spawn( + async move { + child_in_spawned_task().await; + } + .instrument(span), + ); + h.await.unwrap(); + }) + }); + + // S7: alternate fix — re-enter the captured span manually at the top of + // the spawned block (guard lives for the whole block). + let s7 = run_scenario(&exporter, multi_thread_rt, || { + Box::pin(async { + let span = tracing::Span::current(); + let h = tokio::spawn(async move { + let _guard = span.enter(); + child_in_spawned_task().await; + }); + h.await.unwrap(); + }) + }); + + let outcomes = vec![ + ("S1 current-thread, in-block call", s1), + ("S2 multi-thread, in-block call (PRODUCTION)", s2), + ("S4 multi-thread, direct-call spawn", s4), + ("S5 multi-thread, same-task await (control)", s5), + ("S6 multi-thread, spawn + .instrument() (FIX)", s6), + ("S7 multi-thread, spawn + span.enter() (FIX)", s7), + ]; + + let mut report = String::new(); + for (label, outcome) in &outcomes { + match outcome { + Some(o) => { + let line = format!( + "{:<42} propagated={:<5} child_trace={} parent_trace={} child.parent_span_id={} parent.span_id={}", + label, + o.propagated, + o.child_trace, + o.parent_trace, + o.child_parent_id, + o.parent_span_id, + ); + println!("{}", line); + report.push_str(&line); + report.push('\n'); + } + None => { + let line = format!( + "{:<42} NO SPANS EXPORTED\n", + label + ); + println!("{}", line); + report.push_str(&line); + report.push('\n'); + } + } + } + eprintln!("SPIKE-133-REPORT:\n{}", report); + + let get = |prefix: &str| -> &Outcome { + outcomes + .iter() + .find(|(l, _)| l.starts_with(prefix)) + .and_then(|(_, o)| o.as_ref()) + .unwrap_or_else(|| panic!("{prefix} produced no spans:\n{}", report)) + }; + + // Wiring sanity: same-task await MUST propagate, else the layer/exporter + // is broken and nothing below means anything. + assert!( + get("S5").propagated, + "control S5 (same-task await) failed — wiring is broken, not propagation:\n{}", + report + ); + + // THE FINDING (decisive, production shape): a bare `#[instrument]` fn + // called from inside a `tokio::spawn`ed block on a multi-thread runtime + // does NOT nest under the pre-spawn parent — the child gets a fresh + // trace_id and parent_span_id 0000000000000000 (orphan). We assert the + // orphan explicitly so the regression is pinned, not hidden. + assert!( + !get("S2").propagated, + "S2 unexpectedly propagated — re-check the finding:\n{}", + report + ); + assert!( + !get("S4").propagated, + "S4 unexpectedly propagated — re-check the finding:\n{}", + report + ); + + // Same-thread control: with the parent active on the polling thread the + // child DOES nest (proves the layer wiring is sound when context is + // present). + assert!( + get("S1").propagated, + "S1 (current-thread) failed to propagate — wiring is broken:\n{}", + report + ); + + // THE FIX: carrying the parent span across the spawn (`.instrument()` or + // an explicit `span.enter()` guard) restores the nesting. + assert!( + get("S6").propagated, + "S6 (spawn + .instrument()) did NOT propagate — the fix is broken:\n{}", + report + ); + assert!( + get("S7").propagated, + "S7 (spawn + span.enter()) did NOT propagate — the fix is broken:\n{}", + report + ); +}