From dc85f925754d62a33043efed0df742db31a81d65 Mon Sep 17 00:00:00 2001 From: Oloruntobi Allen Date: Tue, 20 Oct 2026 12:37:25 +0100 Subject: [PATCH 1/4] feat(telemetry): add JSON log format Compact log lines can only be searched by matching their text, so fields cannot be queried directly, multi-line errors split across lines, and any query that matches the line text is tied to how events happen to print. `log_format: json` (DWCTL_LOG_FORMAT) writes one JSON object per event, fields at the top level and the span list under `spans`. `compact` stays the default. Switching formats must not lose error logs, so: - In JSON mode a panic is written to stderr as one JSON error line. The hook writes it directly instead of emitting a tracing event, which a RUST_LOG directive or a panic inside the subscriber could drop. - A config that fails to load is logged in the format it asks for, when log_format can still be read from it, and a server exiting with an error logs it before exiting. - Telemetry now starts before batch config validation, whose error was emitted before any subscriber existed and so was never written. - tracing-log is listed explicitly so log-crate events from dependencies keep reaching the subscriber if default features are ever turned off. The zero-token analytics error now records request_model and response_model as plain strings rather than Debug-formatted Options, so they read as values in either format. Queries that matched the old `Some("...")` form need updating before this is deployed. Refs DW-966 --- Cargo.lock | 13 ++ config.yaml | 3 + docs/src/reference/configuration.md | 17 ++ dwctl/Cargo.toml | 4 +- dwctl/src/config.rs | 17 ++ dwctl/src/db/handlers/api_keys.rs | 1 + dwctl/src/lib.rs | 2 +- dwctl/src/main.rs | 39 +++- .../src/request_logging/analytics_handler.rs | 4 +- dwctl/src/telemetry.rs | 180 ++++++++++++++++-- dwctl/src/test/utils.rs | 1 + 11 files changed, 250 insertions(+), 31 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7404b6667..6e8a1377d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7336,6 +7336,16 @@ dependencies = [ "web-time", ] +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + [[package]] name = "tracing-subscriber" version = "0.3.23" @@ -7346,12 +7356,15 @@ dependencies = [ "nu-ansi-term", "once_cell", "regex-automata", + "serde", + "serde_json", "sharded-slab", "smallvec", "thread_local", "tracing", "tracing-core", "tracing-log", + "tracing-serde", ] [[package]] diff --git a/config.yaml b/config.yaml index 0dd2f4b4e..529e9e74c 100644 --- a/config.yaml +++ b/config.yaml @@ -339,6 +339,9 @@ credits: # Optional feature toggles enable_metrics: true # Enable Prometheus metrics endpoint enable_request_logging: true # Enable request/response logging to database +# Log output format: compact (human-readable lines) or json (one JSON object +# per event, for log aggregators). Override with DWCTL_LOG_FORMAT. +log_format: compact # Opt-in heap profile listener (GET /debug/pprof/heap, gzip pprof). Profiles exist # only if the process also starts with jemalloc sampling on, e.g. diff --git a/docs/src/reference/configuration.md b/docs/src/reference/configuration.md index 580f6e00c..8979cb087 100644 --- a/docs/src/reference/configuration.md +++ b/docs/src/reference/configuration.md @@ -808,6 +808,23 @@ does not itself enable span export: correlation also depends on the connecting spans being sampled, exported, and retained. Stored IDs can therefore refer to unavailable spans, and older analytics records may have no gateway span ID. +### Log Format + +```yaml +log_format: compact +``` + +`compact` writes one human-readable line per event. `json` writes one JSON +object per event, for log aggregators: `timestamp`, `level`, `message` and +`target`, the event's own fields at the top level, the innermost span under +`span` and every enclosing span, root first, under `spans`. A multi-line error +stays on one line. In `json` mode panics are written to stderr in the same +shape, with `target` set to `panic` and a message that starts like Rust's +default panic output (`thread '…' panicked at …`). + +Queries and alerts that match text in compact lines need updating when you +switch, since the same event reads differently in each format. + ## Sample Files Generate sample JSONL files for new users: diff --git a/dwctl/Cargo.toml b/dwctl/Cargo.toml index f99e44660..9777f39d2 100644 --- a/dwctl/Cargo.toml +++ b/dwctl/Cargo.toml @@ -54,7 +54,9 @@ anyhow = "1.0" arc-swap = "1.7" log = "0.4" tracing = "0.1" -tracing-subscriber = { version = "0.3", features = ["env-filter"] } +# `tracing-log` forwards `log`-crate events (from dependencies) into tracing. +# It is a default feature, listed so it survives `default-features = false`. +tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "tracing-log"] } # OpenTelemetry opentelemetry = "0.32.0" opentelemetry_sdk = { version = "0.32", features = ["rt-tokio"] } diff --git a/dwctl/src/config.rs b/dwctl/src/config.rs index f8f815a4f..03f91ad94 100644 --- a/dwctl/src/config.rs +++ b/dwctl/src/config.rs @@ -176,6 +176,9 @@ pub struct Config { pub analytics: AnalyticsConfig, /// Enable OpenTelemetry OTLP export for distributed tracing pub enable_otel_export: bool, + /// Log output format: `compact` (human-readable lines) or `json` (one + /// JSON object per event, for log aggregators) + pub log_format: LogFormat, /// Credit system configuration pub credits: CreditsConfig, /// Sample file generation configuration for new users @@ -2042,6 +2045,19 @@ impl BatchConfig { } } +/// How log events are written to stdout. +#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum LogFormat { + /// One human-readable line per event. + #[default] + Compact, + /// One JSON object per event, with the event's fields at the top level + /// and the spans it occurred in under `span` and `spans`. Panics are + /// written in the same shape. + Json, +} + /// Which fusillade claim loops should run inside this daemon process. #[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] @@ -3463,6 +3479,7 @@ impl Default for Config { enable_analytics: true, analytics: AnalyticsConfig::default(), enable_otel_export: false, + log_format: LogFormat::default(), credits: CreditsConfig::default(), sample_files: SampleFilesConfig::default(), limits: LimitsConfig::default(), diff --git a/dwctl/src/db/handlers/api_keys.rs b/dwctl/src/db/handlers/api_keys.rs index ef146f88e..fa4648e9e 100644 --- a/dwctl/src/db/handlers/api_keys.rs +++ b/dwctl/src/db/handlers/api_keys.rs @@ -2566,6 +2566,7 @@ mod tests { enable_analytics: true, analytics: Default::default(), enable_otel_export: false, + log_format: Default::default(), credits: Default::default(), batches: Default::default(), background_services: crate::config::BackgroundServicesConfig::default(), diff --git a/dwctl/src/lib.rs b/dwctl/src/lib.rs index c08d34919..6cb7a8e70 100644 --- a/dwctl/src/lib.rs +++ b/dwctl/src/lib.rs @@ -99,7 +99,7 @@ //! let config = Config::load(&args)?; //! //! // Initialize telemetry (structured logging and optional OpenTelemetry) -//! let tracer_provider = dwctl::telemetry::init_telemetry(config.enable_otel_export)?; +//! let tracer_provider = dwctl::telemetry::init_telemetry(config.enable_otel_export, config.log_format)?; //! //! // Create and start the application //! let app = Application::new(config, tracer_provider).await?; diff --git a/dwctl/src/main.rs b/dwctl/src/main.rs index 3b6967a17..363447aaa 100644 --- a/dwctl/src/main.rs +++ b/dwctl/src/main.rs @@ -121,21 +121,35 @@ async fn async_main() -> anyhow::Result<()> { let cli = Cli::parse(); let args = cli.args; - // Load configuration - let config = Config::load(&args)?; + // Load configuration. A config that fails to load still says how to log + // the failure, when `log_format` itself can be read from it. + let config = match Config::load(&args) { + Ok(config) => config, + Err(error) => { + let log_format = Config::figment(&args).extract_inner("log_format").unwrap_or_default(); + if telemetry::init_telemetry(false, log_format).is_ok() { + tracing::error!(error = %error, "Failed to load configuration"); + } + return Err(error.into()); + } + }; + + // Initialize telemetry (tracing + optional OpenTelemetry) before anything + // logs: events emitted earlier have no subscriber and are discarded. + let tracer_provider = telemetry::init_telemetry(config.enable_otel_export, config.log_format)?; // Validate config consistency config.batches.validate(); // If --validate flag is set, exit successfully after config validation if args.validate { + if let Some(provider) = tracer_provider { + let _ = provider.shutdown(); + } println!("Configuration is valid."); return Ok(()); } - // Initialize telemetry (tracing + optional OpenTelemetry) - let tracer_provider = telemetry::init_telemetry(config.enable_otel_export)?; - tracing::debug!("{:?}", args); if let Some(Command::Migrate { check }) = cli.command { @@ -157,8 +171,15 @@ async fn async_main() -> anyhow::Result<()> { // Run the application with graceful shutdown on SIGTERM/Ctrl+C let shutdown = shutdown_signal(); - Application::new_with_config_path(config, Some(args.config.clone()), tracer_provider) - .await? - .serve(shutdown) - .await + let result = async { + Application::new_with_config_path(config, Some(args.config.clone()), tracer_provider) + .await? + .serve(shutdown) + .await + } + .await; + if let Err(error) = &result { + tracing::error!(error = format!("{error:#}"), "dwctl exited with an error"); + } + result } diff --git a/dwctl/src/request_logging/analytics_handler.rs b/dwctl/src/request_logging/analytics_handler.rs index e14ef65f3..fe42b89d4 100644 --- a/dwctl/src/request_logging/analytics_handler.rs +++ b/dwctl/src/request_logging/analytics_handler.rs @@ -264,8 +264,8 @@ impl RequestHandler for AnalyticsHandler { correlation_id = correlation_id, uri = %request_data.uri, response_type = %metrics.response_type, - request_model = ?metrics.request_model, - response_model = ?metrics.response_model, + request_model = metrics.request_model.as_deref(), + response_model = metrics.response_model.as_deref(), "Successful generative response recorded zero tokens" ); } diff --git a/dwctl/src/telemetry.rs b/dwctl/src/telemetry.rs index 8db9ed611..123453944 100644 --- a/dwctl/src/telemetry.rs +++ b/dwctl/src/telemetry.rs @@ -12,6 +12,13 @@ //! - `OTEL_EXPORTER_OTLP_HEADERS` - Headers as comma-separated key=value pairs. The values can have their spaces encoded URL style - i.e. replace %20 with space. //! - `OTEL_SERVICE_NAME` - Service name for resource identification //! +//! # Log format +//! +//! `log_format` in the config (`DWCTL_LOG_FORMAT`) selects `compact` lines, +//! the default, or `json`: one object per event, for log aggregators. Alert +//! queries that read log lines depend on the JSON field names described on +//! [`fmt_layer`], and in JSON mode panics are written in the same shape. +//! //! # What gets exported //! //! Logs and traces are filtered separately. `RUST_LOG` decides what reaches the @@ -93,14 +100,20 @@ //! ``` use axum::{extract::Request, middleware::Next, response::Response}; +use chrono::{SecondsFormat, Utc}; use onwards::{ErrorOrigin, RequestIdentity, ServedBy}; use opentelemetry::trace::TracerProvider as _; use opentelemetry_otlp::{Protocol, WithExportConfig, WithHttpConfig}; pub use opentelemetry_sdk::trace::SdkTracerProvider; use opentelemetry_sdk::trace::Tracer; +use serde_json::json; +use std::backtrace::{Backtrace, BacktraceStatus}; use std::collections::HashMap; +use std::io::Write; +use std::panic::PanicHookInfo; use tracing::{Subscriber, info}; use tracing_opentelemetry::OpenTelemetrySpanExt; +use tracing_subscriber::fmt::MakeWriter; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::registry::LookupSpan; use tracing_subscriber::util::SubscriberInitExt; @@ -114,6 +127,7 @@ pub const OTEL_TRACES_FILTER: &str = "OTEL_TRACES_FILTER"; /// and events, no repository spans or debug lines. pub const DEFAULT_OTEL_TRACES_FILTER: &str = "info"; +use crate::config::LogFormat; use crate::inference::rejections::RejectionCode; /// Longest account name recorded on a span. Display names are free text, and @@ -204,40 +218,107 @@ pub fn record_ai_response_attributes(response: &axum::http::Response, span /// Initialize tracing with optional OpenTelemetry support /// /// This function sets up tracing-subscriber with: -/// - Console output (fmt layer) +/// - Log output on stdout, in `log_format` (see [`fmt_layer`]) /// - OpenTelemetry OTLP export (only if `enable_otel_export` is true and configured via environment variables) /// +/// With [`LogFormat::Json`] it also replaces the panic hook, so a panic is +/// written as a JSON error line rather than as plain text (see +/// [`write_panic_as_json`]). +/// /// Parameters: /// - `enable_otel_export`: If true, attempts to configure OTLP export using environment variables +/// - `log_format`: How log events are written /// /// Returns the tracer provider if OTLP export was successfully enabled. The caller should /// store this and call `provider.shutdown()` before application exit to flush pending spans. -pub fn init_telemetry(enable_otel_export: bool) -> anyhow::Result> { +pub fn init_telemetry(enable_otel_export: bool, log_format: LogFormat) -> anyhow::Result> { let env_filter = EnvFilter::new(std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_string())); - if enable_otel_export { + let (otel, provider, traces_filter) = if enable_otel_export { let (tracer, provider) = create_otlp_tracer()?; let traces_filter = std::env::var(OTEL_TRACES_FILTER).unwrap_or_else(|_| DEFAULT_OTEL_TRACES_FILTER.to_string()); + (Some(otel_layer(tracer, &traces_filter)), Some(provider), Some(traces_filter)) + } else { + (None, None, None) + }; - tracing_subscriber::registry() - .with(env_filter) - .with(tracing_subscriber::fmt::layer().compact()) - .with(otel_layer(tracer, &traces_filter)) - .try_init()?; + tracing_subscriber::registry() + .with(env_filter) + .with(fmt_layer(log_format, std::io::stdout)) + .with(otel) + .try_init()?; - info!(traces_filter, "Telemetry initialized with OTLP export enabled"); - return Ok(Some(provider)); - } else { - // OTLP export disabled - use only console logging - tracing_subscriber::registry() - .with(env_filter) - .with(tracing_subscriber::fmt::layer().compact()) - .try_init()?; + if log_format == LogFormat::Json { + std::panic::set_hook(Box::new(write_panic_as_json)); + } + + match traces_filter { + Some(traces_filter) => info!(traces_filter, ?log_format, "Telemetry initialized with OTLP export enabled"), + None => info!(?log_format, "Telemetry initialized (OTLP export disabled)"), + } - info!("Telemetry initialized (OTLP export disabled)"); + Ok(provider) +} + +/// The log output layer, writing to `writer` in `format`. +/// +/// JSON events carry their fields at the top level (`message`, `error`, +/// `request_model`, ...) beside `timestamp`, `level` and `target`, the +/// innermost span under `span` and the full span list, root first, under +/// `spans`. Loki derives `detected_level` from `level`. +fn fmt_layer(format: LogFormat, writer: W) -> Box + Send + Sync> +where + S: Subscriber + for<'span> LookupSpan<'span>, + W: for<'writer> MakeWriter<'writer> + Send + Sync + 'static, +{ + let layer = tracing_subscriber::fmt::layer().with_writer(writer); + match format { + LogFormat::Compact => layer.compact().boxed(), + LogFormat::Json => layer.json().flatten_event(true).boxed(), } +} + +/// Panic hook for JSON logging: writes the panic to stderr as one JSON error +/// line, in the shape [`fmt_layer`] gives events. +/// +/// It writes the line itself rather than emitting a tracing event, because an +/// event can be lost exactly when it matters: `RUST_LOG` directives that don't +/// cover the `panic` target filter it out, and a panic raised while the +/// subscriber is handling an event is dispatched to no subscriber at all. +fn write_panic_as_json(info: &PanicHookInfo<'_>) { + let line = panic_line(info, &Backtrace::capture()); + let _ = writeln!(std::io::stderr().lock(), "{line}"); +} - Ok(None) +/// The JSON line [`write_panic_as_json`] writes. The message reads like +/// Rust's default panic output, so searches for `panicked at` still match. +/// The backtrace is included when `RUST_BACKTRACE` enables capturing one. +fn panic_line(info: &PanicHookInfo<'_>, backtrace: &Backtrace) -> serde_json::Value { + let payload = info + .payload() + .downcast_ref::<&str>() + .copied() + .or_else(|| info.payload().downcast_ref::().map(String::as_str)) + .unwrap_or("Box"); + let thread = std::thread::current(); + let thread = thread.name().unwrap_or(""); + let location = info.location().map(ToString::to_string); + let message = match &location { + Some(location) => format!("thread '{thread}' panicked at {location}: {payload}"), + None => format!("thread '{thread}' panicked: {payload}"), + }; + let mut line = json!({ + "timestamp": Utc::now().to_rfc3339_opts(SecondsFormat::Micros, true), + "level": "ERROR", + "message": message, + "thread": thread, + "location": location, + "target": "panic", + }); + if backtrace.status() == BacktraceStatus::Captured { + line["backtrace"] = backtrace.to_string().into(); + } + line } /// The span export layer: `tracer` behind its own `directives` filter, without @@ -453,4 +534,67 @@ mod tests { result.unwrap_err() ); } + + /// Run by [`json_logging_writes_every_error_as_one_json_line`] in a child + /// process, which owns the global subscriber and panic hook. Does nothing + /// in a normal test run. + #[test] + fn json_logging_child() { + if std::env::var_os(JSON_LOGGING_CHILD).is_none() { + return; + } + init_telemetry(false, LogFormat::Json).unwrap(); + tracing::info_span!("request").in_scope(|| { + tracing::error!(error = "first line\nsecond line", request_model = Some("model-a"), "tracing error"); + }); + log::error!("log crate error"); + let _ = std::thread::spawn(|| panic!("worker panic")).join(); + } + + const JSON_LOGGING_CHILD: &str = "DWCTL_JSON_LOGGING_TEST_CHILD"; + + /// With `log_format: json`, a tracing error (multi-line message included), + /// a `log`-crate error from a dependency and a panic each reach the output + /// as a single JSON line at level `ERROR`. Runs the real initialization in + /// a child process, since it installs the global subscriber and panic hook. + #[test] + fn json_logging_writes_every_error_as_one_json_line() { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "telemetry::tests::json_logging_child", "--nocapture"]) + .env(JSON_LOGGING_CHILD, "1") + .env("RUST_LOG", "info") + .output() + .expect("start json logging child"); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "json logging child failed:\n{stdout}\n{stderr}"); + + let lines: Vec = stdout + .lines() + .chain(stderr.lines()) + .filter_map(|line| serde_json::from_str(line).ok()) + .collect(); + let find = |message: &str| { + lines + .iter() + .find(|line| line["message"].as_str().is_some_and(|m| m.contains(message))) + .unwrap_or_else(|| panic!("no JSON line for {message:?}:\n{stdout}\n{stderr}")) + }; + + let tracing_error = find("tracing error"); + assert_eq!(tracing_error["level"], "ERROR"); + assert_eq!(tracing_error["error"], "first line\nsecond line"); + assert_eq!(tracing_error["request_model"], "model-a"); + assert_eq!(tracing_error["spans"][0]["name"], "request"); + + assert_eq!(find("log crate error")["level"], "ERROR"); + + let panic = find("worker panic"); + assert_eq!(panic["level"], "ERROR"); + assert_eq!(panic["target"], "panic"); + assert!( + panic["message"].as_str().unwrap().contains("panicked at"), + "panic message keeps the default wording: {panic}" + ); + } } diff --git a/dwctl/src/test/utils.rs b/dwctl/src/test/utils.rs index a4a26cf6c..4a702fbb1 100644 --- a/dwctl/src/test/utils.rs +++ b/dwctl/src/test/utils.rs @@ -366,6 +366,7 @@ pub fn create_test_config() -> crate::config::Config { enable_analytics: true, analytics: crate::config::AnalyticsConfig::default(), enable_otel_export: false, + log_format: Default::default(), credits: crate::config::CreditsConfig::default(), batches: BatchConfig { enabled: true, From 08aa52b91ed7e59b20246924ee738d00fe338c58 Mon Sep 17 00:00:00 2001 From: Oloruntobi Allen Date: Tue, 20 Oct 2026 13:05:10 +0100 Subject: [PATCH 2/4] fix(telemetry): keep JSON mode output free of plain-text lines Review follow-ups for the JSON log format: - main reports the error dwctl exits with once, through tracing when logging is up, and exits nonzero. Returning it from main also printed it as plain `Error: ...` lines beside the JSON line. - Logging is installed before the OTLP exporter is built, so an exporter setup error is reported in the configured format, and the `[OTLP]` startup prints become fields on the initialization event. - Panic lines carry the `span` and `spans` the panic happened in, as events do, so a panic can be tied to the request that caused it. - The docs no longer claim compact output is one line per event. Refs DW-966 --- docs/src/reference/configuration.md | 17 ++-- dwctl/src/main.rs | 67 ++++++++------- dwctl/src/telemetry.rs | 126 ++++++++++++++++++++++------ 3 files changed, 148 insertions(+), 62 deletions(-) diff --git a/docs/src/reference/configuration.md b/docs/src/reference/configuration.md index 8979cb087..89cc67999 100644 --- a/docs/src/reference/configuration.md +++ b/docs/src/reference/configuration.md @@ -814,13 +814,16 @@ unavailable spans, and older analytics records may have no gateway span ID. log_format: compact ``` -`compact` writes one human-readable line per event. `json` writes one JSON -object per event, for log aggregators: `timestamp`, `level`, `message` and -`target`, the event's own fields at the top level, the innermost span under -`span` and every enclosing span, root first, under `spans`. A multi-line error -stays on one line. In `json` mode panics are written to stderr in the same -shape, with `target` set to `panic` and a message that starts like Rust's -default panic output (`thread '…' panicked at …`). +`compact` writes human-readable text for each event; a multi-line message or +error spans several lines. `json` writes one JSON object per event, for log +aggregators: `timestamp`, `level`, `message` and `target`, the event's own +fields at the top level, the innermost span under `span` and every enclosing +span, root first, under `spans`. A multi-line error stays on one line. In +`json` mode panics are written to stderr in the same shape, including the +spans the panic happened in, with `target` set to `panic` and a message that +starts like Rust's default panic output (`thread '…' panicked at …`). The +error `dwctl` exits with, including a configuration that fails to load, is +logged once as `dwctl exited with an error`. Queries and alerts that match text in compact lines need updating when you switch, since the same event reads differently in each format. diff --git a/dwctl/src/main.rs b/dwctl/src/main.rs index 363447aaa..9dade3eee 100644 --- a/dwctl/src/main.rs +++ b/dwctl/src/main.rs @@ -1,3 +1,6 @@ +use std::process::ExitCode; + +use anyhow::Context; use clap::{Parser, Subcommand}; use dwctl::config::Args; use dwctl::{Application, Config, migrations, telemetry}; @@ -79,14 +82,38 @@ async fn shutdown_signal() { } } -fn main() -> anyhow::Result<()> { - tokio::runtime::Builder::new_multi_thread() +fn main() -> ExitCode { + let result = tokio::runtime::Builder::new_multi_thread() .enable_all() // 8MB stack per worker thread — the default 2MB overflows with deep // tracing-opentelemetry span nesting during batch request processing .thread_stack_size(8 * 1024 * 1024) - .build()? - .block_on(async_main()) + .build() + .map_err(anyhow::Error::from) + .and_then(|runtime| runtime.block_on(async_main())); + match result { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + report_error(&error); + ExitCode::FAILURE + } + } +} + +/// Reports the error `dwctl` exits with, once: through tracing when logging +/// is installed, so it comes out in the configured format, and as plain text +/// on stderr when it isn't (logging itself failed to start). Returning the +/// error from `main` instead would print it a second time, as plain text, +/// beside the log line. +fn report_error(error: &anyhow::Error) { + if tracing::dispatcher::has_been_set() { + // `{:#}` prints the context chain (target, migration version) ahead + // of the underlying error, which is what an operator reading a failed + // pod or Job needs first. + tracing::error!(error = format!("{error:#}"), "dwctl exited with an error"); + } else { + eprintln!("Error: {error:?}"); + } } /// `dwctl` serves by default; `dwctl migrate` applies schema migrations and @@ -127,10 +154,8 @@ async fn async_main() -> anyhow::Result<()> { Ok(config) => config, Err(error) => { let log_format = Config::figment(&args).extract_inner("log_format").unwrap_or_default(); - if telemetry::init_telemetry(false, log_format).is_ok() { - tracing::error!(error = %error, "Failed to load configuration"); - } - return Err(error.into()); + let _ = telemetry::init_telemetry(false, log_format); + return Err(anyhow::Error::from(error).context("Failed to load configuration")); } }; @@ -157,29 +182,13 @@ async fn async_main() -> anyhow::Result<()> { if let Some(provider) = tracer_provider { let _ = provider.shutdown(); } - return match result { - Ok(_) => Ok(()), - Err(error) => { - // `{:#}` prints the context chain (target, migration version) - // ahead of the driver error, which is what an operator reading - // a failed Job needs first. - tracing::error!(error = format!("{error:#}"), "dwctl migrate failed"); - Err(error) - } - }; + return result.map(|_| ()).context("dwctl migrate failed"); } // Run the application with graceful shutdown on SIGTERM/Ctrl+C let shutdown = shutdown_signal(); - let result = async { - Application::new_with_config_path(config, Some(args.config.clone()), tracer_provider) - .await? - .serve(shutdown) - .await - } - .await; - if let Err(error) = &result { - tracing::error!(error = format!("{error:#}"), "dwctl exited with an error"); - } - result + Application::new_with_config_path(config, Some(args.config.clone()), tracer_provider) + .await? + .serve(shutdown) + .await } diff --git a/dwctl/src/telemetry.rs b/dwctl/src/telemetry.rs index 123453944..23b87c3eb 100644 --- a/dwctl/src/telemetry.rs +++ b/dwctl/src/telemetry.rs @@ -106,16 +106,17 @@ use opentelemetry::trace::TracerProvider as _; use opentelemetry_otlp::{Protocol, WithExportConfig, WithHttpConfig}; pub use opentelemetry_sdk::trace::SdkTracerProvider; use opentelemetry_sdk::trace::Tracer; -use serde_json::json; +use serde_json::{Map, Value, json}; use std::backtrace::{Backtrace, BacktraceStatus}; use std::collections::HashMap; use std::io::Write; use std::panic::PanicHookInfo; use tracing::{Subscriber, info}; use tracing_opentelemetry::OpenTelemetrySpanExt; -use tracing_subscriber::fmt::MakeWriter; +use tracing_subscriber::fmt::format::JsonFields; +use tracing_subscriber::fmt::{FormattedFields, MakeWriter}; use tracing_subscriber::layer::SubscriberExt; -use tracing_subscriber::registry::LookupSpan; +use tracing_subscriber::registry::{LookupSpan, Registry, SpanRef}; use tracing_subscriber::util::SubscriberInitExt; use tracing_subscriber::{EnvFilter, Layer}; @@ -231,15 +232,17 @@ pub fn record_ai_response_attributes(response: &axum::http::Response, span /// /// Returns the tracer provider if OTLP export was successfully enabled. The caller should /// store this and call `provider.shutdown()` before application exit to flush pending spans. +/// +/// If OTLP export fails to set up, logging is still installed (without +/// export) before the error is returned, so the caller can log the error in +/// `log_format`. pub fn init_telemetry(enable_otel_export: bool, log_format: LogFormat) -> anyhow::Result> { let env_filter = EnvFilter::new(std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_string())); - - let (otel, provider, traces_filter) = if enable_otel_export { - let (tracer, provider) = create_otlp_tracer()?; - let traces_filter = std::env::var(OTEL_TRACES_FILTER).unwrap_or_else(|_| DEFAULT_OTEL_TRACES_FILTER.to_string()); - (Some(otel_layer(tracer, &traces_filter)), Some(provider), Some(traces_filter)) - } else { - (None, None, None) + let traces_filter = std::env::var(OTEL_TRACES_FILTER).unwrap_or_else(|_| DEFAULT_OTEL_TRACES_FILTER.to_string()); + let (otel, provider, otlp_settings, otlp_error) = match enable_otel_export.then(create_otlp_tracer) { + Some(Ok((tracer, provider, settings))) => (Some(otel_layer(tracer, &traces_filter)), Some(provider), Some(settings), None), + Some(Err(error)) => (None, None, None, Some(error)), + None => (None, None, None, None), }; tracing_subscriber::registry() @@ -252,8 +255,18 @@ pub fn init_telemetry(enable_otel_export: bool, log_format: LogFormat) -> anyhow std::panic::set_hook(Box::new(write_panic_as_json)); } - match traces_filter { - Some(traces_filter) => info!(traces_filter, ?log_format, "Telemetry initialized with OTLP export enabled"), + if let Some(error) = otlp_error { + return Err(error.context("Failed to set up OTLP export")); + } + match otlp_settings { + Some(settings) => info!( + service_name = settings.service_name, + endpoint = settings.endpoint, + custom_headers = settings.header_count, + traces_filter, + ?log_format, + "Telemetry initialized with OTLP export enabled" + ), None => info!(?log_format, "Telemetry initialized (OTLP export disabled)"), } @@ -286,14 +299,40 @@ where /// cover the `panic` target filter it out, and a panic raised while the /// subscriber is handling an event is dispatched to no subscriber at all. fn write_panic_as_json(info: &PanicHookInfo<'_>) { - let line = panic_line(info, &Backtrace::capture()); + let line = panic_line(info, &Backtrace::capture(), current_spans()); let _ = writeln!(std::io::stderr().lock(), "{line}"); } +/// The span the panicking thread is in and its scope, root first, shaped +/// like the `span` and `spans` of a JSON event. The hook runs on that thread +/// before unwinding, so this is the span of the request or task that +/// panicked. `None` outside any span, or when the panic happened inside the +/// subscriber, which then dispatches to no subscriber. +fn current_spans() -> Option<(Value, Vec)> { + tracing::dispatcher::get_default(|dispatch| { + let registry = dispatch.downcast_ref::()?; + let id = dispatch.current_span().id()?.clone(); + let span = registry.span(&id)?; + let spans = span.scope().from_root().map(|span| span_json(&span)).collect(); + Some((span_json(&span), spans)) + }) +} + +/// A span as the JSON layer writes it: its fields, then its `name`. +fn span_json(span: &SpanRef<'_, Registry>) -> Value { + let mut object = span + .extensions() + .get::>() + .and_then(|fields| serde_json::from_str::>(fields).ok()) + .unwrap_or_default(); + object.insert("name".to_string(), span.name().into()); + Value::Object(object) +} + /// The JSON line [`write_panic_as_json`] writes. The message reads like /// Rust's default panic output, so searches for `panicked at` still match. /// The backtrace is included when `RUST_BACKTRACE` enables capturing one. -fn panic_line(info: &PanicHookInfo<'_>, backtrace: &Backtrace) -> serde_json::Value { +fn panic_line(info: &PanicHookInfo<'_>, backtrace: &Backtrace, spans: Option<(Value, Vec)>) -> Value { let payload = info .payload() .downcast_ref::<&str>() @@ -315,6 +354,10 @@ fn panic_line(info: &PanicHookInfo<'_>, backtrace: &Backtrace) -> serde_json::Va "location": location, "target": "panic", }); + if let Some((span, spans)) = spans { + line["span"] = span; + line["spans"] = spans.into(); + } if backtrace.status() == BacktraceStatus::Captured { line["backtrace"] = backtrace.to_string().into(); } @@ -350,7 +393,7 @@ where /// - OTEL_SERVICE_NAME /// /// Returns both the tracer and provider. The provider must be retained for shutdown. -fn create_otlp_tracer() -> anyhow::Result<(Tracer, SdkTracerProvider)> { +fn create_otlp_tracer() -> anyhow::Result<(Tracer, SdkTracerProvider, OtlpSettings)> { // Get service name from env or use default let service_name = std::env::var("OTEL_SERVICE_NAME").unwrap_or_else(|_| "dwctl".to_string()); @@ -363,9 +406,6 @@ fn create_otlp_tracer() -> anyhow::Result<(Tracer, SdkTracerProvider)> { format!("{}/v1/traces", base.trim_end_matches('/')) }; - eprintln!("[OTLP] Initializing OTLP tracer with the following configuration:"); - eprintln!("[OTLP] Service Name: {}", service_name); - eprintln!("[OTLP] Endpoint: {}", endpoint); // Parse headers from environment variable let mut headers = HashMap::new(); if let Ok(headers_str) = std::env::var("OTEL_EXPORTER_OTLP_HEADERS") { @@ -381,9 +421,10 @@ fn create_otlp_tracer() -> anyhow::Result<(Tracer, SdkTracerProvider)> { headers.insert(key, value); } } - eprintln!("[OTLP] Custom headers, length: {}", headers.len()); } + let headers_len = headers.len(); + // Determine protocol let protocol = match std::env::var("OTEL_EXPORTER_OTLP_PROTOCOL").as_deref().unwrap_or("http/protobuf") { "http/protobuf" => Protocol::HttpBinary, @@ -409,9 +450,21 @@ fn create_otlp_tracer() -> anyhow::Result<(Tracer, SdkTracerProvider)> { .with_resource(resource) .build(); - let tracer = tracer_provider.tracer(service_name); + let tracer = tracer_provider.tracer(service_name.clone()); + let settings = OtlpSettings { + service_name, + endpoint, + header_count: headers_len, + }; + + Ok((tracer, tracer_provider, settings)) +} - Ok((tracer, tracer_provider)) +/// What [`create_otlp_tracer`] configured, logged once logging is up. +struct OtlpSettings { + service_name: String, + endpoint: String, + header_count: usize, } #[cfg(test)] @@ -514,7 +567,7 @@ mod tests { #[test] fn otlp_tracer_builds_with_http_client() { - let (_, provider) = create_otlp_tracer().expect( + let (_, provider, _) = create_otlp_tracer().expect( "OTLP tracer failed to build - likely a feature flag conflict \ (reqwest-client vs reqwest-blocking-client)", ); @@ -543,26 +596,36 @@ mod tests { if std::env::var_os(JSON_LOGGING_CHILD).is_none() { return; } - init_telemetry(false, LogFormat::Json).unwrap(); + // Nothing listens on the endpoint; setting up export doesn't connect. + let provider = init_telemetry(true, LogFormat::Json).unwrap(); + assert!(provider.is_some()); tracing::info_span!("request").in_scope(|| { tracing::error!(error = "first line\nsecond line", request_model = Some("model-a"), "tracing error"); }); log::error!("log crate error"); - let _ = std::thread::spawn(|| panic!("worker panic")).join(); + let _ = std::thread::spawn(|| { + tracing::info_span!("request", request_id = 7).in_scope(|| { + tracing::info_span!("worker").in_scope(|| panic!("worker panic")); + }); + }) + .join(); } const JSON_LOGGING_CHILD: &str = "DWCTL_JSON_LOGGING_TEST_CHILD"; /// With `log_format: json`, a tracing error (multi-line message included), /// a `log`-crate error from a dependency and a panic each reach the output - /// as a single JSON line at level `ERROR`. Runs the real initialization in - /// a child process, since it installs the global subscriber and panic hook. + /// as a single JSON line at level `ERROR`, the panic with the spans it + /// happened in, and setting up OTLP export prints nothing outside the log. + /// Runs the real initialization in a child process, since it installs the + /// global subscriber and panic hook. #[test] fn json_logging_writes_every_error_as_one_json_line() { let output = std::process::Command::new(std::env::current_exe().unwrap()) .args(["--exact", "telemetry::tests::json_logging_child", "--nocapture"]) .env(JSON_LOGGING_CHILD, "1") .env("RUST_LOG", "info") + .env("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:9") .output() .expect("start json logging child"); let stdout = String::from_utf8_lossy(&output.stdout); @@ -596,5 +659,16 @@ mod tests { panic["message"].as_str().unwrap().contains("panicked at"), "panic message keeps the default wording: {panic}" ); + assert_eq!(panic["span"]["name"], "worker"); + assert_eq!(panic["spans"][0]["name"], "request"); + assert_eq!(panic["spans"][0]["request_id"], 7); + assert_eq!(panic["spans"][1]["name"], "worker"); + + let initialized = find("OTLP export enabled"); + assert_eq!(initialized["endpoint"], "http://127.0.0.1:9/v1/traces"); + assert!( + !stdout.contains("[OTLP]") && !stderr.contains("[OTLP]"), + "OTLP setup is logged as an event, not printed: {stderr}" + ); } } From 52885988aafd2ea62ca131e6427a92782e0e4733 Mon Sep 17 00:00:00 2001 From: Oloruntobi Allen Date: Wed, 21 Oct 2026 10:16:14 +0100 Subject: [PATCH 3/4] feat(telemetry): put request IDs at the top level of JSON log lines In JSON mode span fields only appear inside the `spans` array, which Loki's `json` parser does not look into, so the request ID could no longer be used as a key:value lookup. Services downstream of dwctl already log the same ID as a top-level `x_request_id`. A small layer records each span's correlation fields (x_request_id, edge_request_id, provider_request_id, trace_id), and the JSON formatter repeats them at the top level of every event inside that span, the innermost value winning and an event's own field taking precedence. Panic lines carry them too. The analytics handler, which runs after the response and outside the request span, now records the request's x_request_id on its own span so its errors can be found the same way. One query now finds a request across dwctl and the services it calls: `|= "" | json | x_request_id=""`. Refs DW-966 --- Cargo.lock | 2 +- docs/src/reference/configuration.md | 12 + dwctl/src/lib.rs | 6 +- .../src/request_logging/analytics_handler.rs | 9 +- dwctl/src/telemetry.rs | 254 ++++++++++++++++-- dwctl/src/test/responses/request_id.rs | 44 +++ 6 files changed, 308 insertions(+), 19 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1fb536484..ff86536aa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1942,7 +1942,7 @@ dependencies = [ "hex", "hickory-resolver", "hmac 0.13.0", - "http-body 1.1.0", + "http-body", "http-body-util", "humantime", "humantime-serde", diff --git a/docs/src/reference/configuration.md b/docs/src/reference/configuration.md index 89cc67999..d07d68be3 100644 --- a/docs/src/reference/configuration.md +++ b/docs/src/reference/configuration.md @@ -825,6 +825,18 @@ starts like Rust's default panic output (`thread '…' panicked at …`). The error `dwctl` exits with, including a configuration that fails to load, is logged once as `dwctl exited with an error`. +Every JSON line logged while handling a request also carries the request's +IDs at the top level: `x_request_id` (the ID dwctl mints and returns in the +`X-Request-ID` header and forwards to providers), `edge_request_id` (an +inbound `X-Request-ID`, such as one set by an ingress), `provider_request_id` +and `trace_id`. Analytics logs written after the response carry +`x_request_id` too. Find every line of a request, in dwctl and in any +provider that logs `x_request_id` the same way, with one query: + +```logql +{namespace=~""} |= "" | json | x_request_id="" +``` + Queries and alerts that match text in compact lines need updating when you switch, since the same event reads differently in each format. diff --git a/dwctl/src/lib.rs b/dwctl/src/lib.rs index 5f6161929..332983de6 100644 --- a/dwctl/src/lib.rs +++ b/dwctl/src/lib.rs @@ -2776,8 +2776,10 @@ pub async fn build_router( // Add tracing layer with OTel-compatible span names and HTTP semantic conventions. // Tracing span fields appear on every fmt log line in the request, so only the // correlation keys are fields: trace_id, otel.name, x_request_id, - // edge_request_id and provider_request_id (the request IDs are what - // `|= "x_request_id="` log queries match on), plus otel.kind, which the + // edge_request_id and provider_request_id (the request IDs are what log + // queries match on: `|= "x_request_id="` on compact lines, and + // `| json | x_request_id=""` on JSON lines, which repeat them at the + // top level; see `telemetry::CORRELATION_FIELDS`), plus otel.kind, which the // tracing-opentelemetry layer only reads from a field (as a raw attribute it // is ignored and the span exports as `internal`). All other attributes are // set via OpenTelemetrySpanExt::set_attribute() so they're exported to the diff --git a/dwctl/src/request_logging/analytics_handler.rs b/dwctl/src/request_logging/analytics_handler.rs index 7cc6bcbcc..c806b7795 100644 --- a/dwctl/src/request_logging/analytics_handler.rs +++ b/dwctl/src/request_logging/analytics_handler.rs @@ -212,8 +212,15 @@ impl RequestHandler for AnalyticsHandler { let span = info_span!( "dwctl.analytics_handler", correlation_id = correlation_id, - status = %response_data.status + status = %response_data.status, + x_request_id = tracing::field::Empty, ); + // This runs after the response, outside the request span, so it + // carries the request's ID itself for the analytics logs to be found + // by it. `assign_request_id` set the header before outlet captured it. + if let Some(x_request_id) = extract_header_as_string(&request_data, "x-request-id") { + span.record("x_request_id", tracing::field::display(x_request_id)); + } async { // Whether this is a token-bearing generative endpoint. Gates the diff --git a/dwctl/src/telemetry.rs b/dwctl/src/telemetry.rs index 23b87c3eb..5f459138d 100644 --- a/dwctl/src/telemetry.rs +++ b/dwctl/src/telemetry.rs @@ -111,11 +111,13 @@ use std::backtrace::{Backtrace, BacktraceStatus}; use std::collections::HashMap; use std::io::Write; use std::panic::PanicHookInfo; -use tracing::{Subscriber, info}; +use tracing::field::{Field, Visit}; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Event, Subscriber, info}; use tracing_opentelemetry::OpenTelemetrySpanExt; -use tracing_subscriber::fmt::format::JsonFields; -use tracing_subscriber::fmt::{FormattedFields, MakeWriter}; -use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::fmt::format::{Format, Json, JsonFields, Writer}; +use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields, FormattedFields, MakeWriter}; +use tracing_subscriber::layer::{Context, SubscriberExt}; use tracing_subscriber::registry::{LookupSpan, Registry, SpanRef}; use tracing_subscriber::util::SubscriberInitExt; use tracing_subscriber::{EnvFilter, Layer}; @@ -278,8 +280,10 @@ pub fn init_telemetry(enable_otel_export: bool, log_format: LogFormat) -> anyhow /// JSON events carry their fields at the top level (`message`, `error`, /// `request_model`, ...) beside `timestamp`, `level` and `target`, the /// innermost span under `span` and the full span list, root first, under -/// `spans`. Loki derives `detected_level` from `level`. -fn fmt_layer(format: LogFormat, writer: W) -> Box + Send + Sync> +/// `spans`. Loki derives `detected_level` from `level`. The +/// [`CORRELATION_FIELDS`] of the spans an event is in are repeated at the top +/// level, so `| json | x_request_id=""` finds every line of a request. +pub(crate) fn fmt_layer(format: LogFormat, writer: W) -> Box + Send + Sync> where S: Subscriber + for<'span> LookupSpan<'span>, W: for<'writer> MakeWriter<'writer> + Send + Sync + 'static, @@ -287,7 +291,142 @@ where let layer = tracing_subscriber::fmt::layer().with_writer(writer); match format { LogFormat::Compact => layer.compact().boxed(), - LogFormat::Json => layer.json().flatten_event(true).boxed(), + LogFormat::Json => CorrelationLayer + .and_then( + layer + .json() + .event_format(JsonWithCorrelation(tracing_subscriber::fmt::format().json().flatten_event(true))), + ) + .boxed(), + } +} + +/// Span fields that identify a request across services. In JSON output each +/// is repeated at the top level of every event inside a span that recorded +/// it, because Loki's `json` parser does not look inside the `spans` array. +/// `x_request_id` is the key other services in the request path log the same +/// ID under, so one key:value query matches all of them. +const CORRELATION_FIELDS: [&str; 4] = ["x_request_id", "edge_request_id", "provider_request_id", "trace_id"]; + +/// The [`CORRELATION_FIELDS`] a span has recorded, kept in its extensions by +/// [`CorrelationLayer`]. +#[derive(Default)] +struct CorrelationIds(Vec<(&'static str, String)>); + +impl Visit for CorrelationIds { + fn record_str(&mut self, field: &Field, value: &str) { + self.set(field, value.to_string()); + } + + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + // A `display` value's Debug is its Display, so IDs recorded with + // `display` (as the request span's are) are not quoted. + self.set(field, format!("{value:?}")); + } +} + +impl CorrelationIds { + fn set(&mut self, field: &Field, value: String) { + let Some(name) = CORRELATION_FIELDS.into_iter().find(|name| *name == field.name()) else { + return; + }; + match self.0.iter_mut().find(|(existing, _)| *existing == name) { + Some((_, existing)) => *existing = value, + None => self.0.push((name, value)), + } + } +} + +/// Records each span's [`CORRELATION_FIELDS`], at creation and when they are +/// recorded later, for [`JsonWithCorrelation`] and the panic hook to read. +struct CorrelationLayer; + +impl Layer for CorrelationLayer +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) { + let mut ids = CorrelationIds::default(); + attrs.record(&mut ids); + if !ids.0.is_empty() + && let Some(span) = ctx.span(id) + { + span.extensions_mut().insert(ids); + } + } + + fn on_record(&self, id: &Id, values: &Record<'_>, ctx: Context<'_, S>) { + let Some(span) = ctx.span(id) else { return }; + let mut extensions = span.extensions_mut(); + match extensions.get_mut::() { + Some(ids) => values.record(ids), + None => { + let mut ids = CorrelationIds::default(); + values.record(&mut ids); + if !ids.0.is_empty() { + extensions.insert(ids); + } + } + } + } +} + +/// The correlation IDs of a span scope given leaf first: the innermost value +/// of each field wins, and fields `skip` names are left out. In +/// [`CORRELATION_FIELDS`] order, so lines read the same way every time. +fn correlation_ids<'a, R: LookupSpan<'a> + 'a>( + leaf_first: impl Iterator>, + skip: impl Fn(&str) -> bool, +) -> Vec<(&'static str, String)> { + let mut found: Vec<(&'static str, String)> = Vec::new(); + for span in leaf_first { + if let Some(ids) = span.extensions().get::() { + for (name, value) in &ids.0 { + if !skip(name) && !found.iter().any(|(existing, _)| existing == name) { + found.push((name, value.clone())); + } + } + } + } + found.sort_by_key(|(name, _)| CORRELATION_FIELDS.iter().position(|field| field == name)); + found +} + +/// The JSON event format, with the event's correlation IDs (see +/// [`CORRELATION_FIELDS`]) added at the top level. An event that sets one of +/// those fields itself keeps its own value. +struct JsonWithCorrelation(Format); + +impl FormatEvent for JsonWithCorrelation +where + S: Subscriber + for<'span> LookupSpan<'span>, + N: for<'writer> FormatFields<'writer> + 'static, +{ + fn format_event(&self, ctx: &FmtContext<'_, S, N>, mut writer: Writer<'_>, event: &Event<'_>) -> std::fmt::Result { + let fields = event.metadata().fields(); + let ids = ctx + .event_scope() + .map(|scope| correlation_ids(scope, |name| fields.field(name).is_some())) + .unwrap_or_default(); + if ids.is_empty() { + return self.0.format_event(ctx, writer, event); + } + let mut line = String::new(); + self.0.format_event(ctx, Writer::new(&mut line), event)?; + // The JSON format writes one object and a newline; add the IDs as + // the object's last members. + let Some(close) = line.trim_end().strip_suffix('}').map(str::len) else { + return writer.write_str(&line); + }; + let mut members = String::new(); + for (name, value) in ids { + members.push(','); + members.push_str(&Value::from(name).to_string()); + members.push(':'); + members.push_str(&Value::from(value).to_string()); + } + line.insert_str(close, &members); + writer.write_str(&line) } } @@ -308,16 +447,26 @@ fn write_panic_as_json(info: &PanicHookInfo<'_>) { /// before unwinding, so this is the span of the request or task that /// panicked. `None` outside any span, or when the panic happened inside the /// subscriber, which then dispatches to no subscriber. -fn current_spans() -> Option<(Value, Vec)> { +fn current_spans() -> Option { tracing::dispatcher::get_default(|dispatch| { let registry = dispatch.downcast_ref::()?; let id = dispatch.current_span().id()?.clone(); let span = registry.span(&id)?; - let spans = span.scope().from_root().map(|span| span_json(&span)).collect(); - Some((span_json(&span), spans)) + Some(PanicSpans { + span: span_json(&span), + spans: span.scope().from_root().map(|span| span_json(&span)).collect(), + correlation_ids: correlation_ids(span.scope(), |_| false), + }) }) } +/// The span context [`current_spans`] found for a panic. +struct PanicSpans { + span: Value, + spans: Vec, + correlation_ids: Vec<(&'static str, String)>, +} + /// A span as the JSON layer writes it: its fields, then its `name`. fn span_json(span: &SpanRef<'_, Registry>) -> Value { let mut object = span @@ -332,7 +481,7 @@ fn span_json(span: &SpanRef<'_, Registry>) -> Value { /// The JSON line [`write_panic_as_json`] writes. The message reads like /// Rust's default panic output, so searches for `panicked at` still match. /// The backtrace is included when `RUST_BACKTRACE` enables capturing one. -fn panic_line(info: &PanicHookInfo<'_>, backtrace: &Backtrace, spans: Option<(Value, Vec)>) -> Value { +fn panic_line(info: &PanicHookInfo<'_>, backtrace: &Backtrace, spans: Option) -> Value { let payload = info .payload() .downcast_ref::<&str>() @@ -354,9 +503,12 @@ fn panic_line(info: &PanicHookInfo<'_>, backtrace: &Backtrace, spans: Option<(Va "location": location, "target": "panic", }); - if let Some((span, spans)) = spans { - line["span"] = span; - line["spans"] = spans.into(); + if let Some(spans) = spans { + line["span"] = spans.span; + line["spans"] = spans.spans.into(); + for (name, value) in spans.correlation_ids { + line[name] = value.into(); + } } if backtrace.status() == BacktraceStatus::Captured { line["backtrace"] = backtrace.to_string().into(); @@ -588,6 +740,77 @@ mod tests { ); } + /// Writes JSON lines into a buffer, for tests that read them back. + #[derive(Clone, Default)] + struct Captured(Arc>>); + + impl std::io::Write for Captured { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl Captured { + fn lines(&self) -> Vec { + String::from_utf8(self.0.lock().unwrap().clone()) + .unwrap() + .lines() + .map(str::to_string) + .collect() + } + } + + /// In JSON, the request IDs recorded on the spans an event is in are + /// repeated at the top level, the innermost value winning and recorded + /// late (as the request span records them) as well as at creation. An + /// event that sets one itself keeps its own value, and an event outside + /// any span gets none. + #[test] + fn json_events_carry_their_spans_request_ids_at_the_top_level() { + let captured = Captured::default(); + let writer = captured.clone(); + let subscriber = tracing_subscriber::registry() + .with(EnvFilter::new("info")) + .with(fmt_layer(LogFormat::Json, move || writer.clone())); + tracing::subscriber::with_default(subscriber, || { + let request = tracing::info_span!("request", x_request_id = tracing::field::Empty, trace_id = "trace-1"); + request.record("x_request_id", tracing::field::display("req-outer")); + request.in_scope(|| { + tracing::info_span!("handler", edge_request_id = "edge-1").in_scope(|| { + tracing::info!("in request"); + tracing::info_span!("retry", x_request_id = "req-inner").in_scope(|| tracing::info!("in retry")); + tracing::info!(x_request_id = "own", "sets its own"); + }); + }); + tracing::info!("outside"); + }); + + let lines = captured.lines(); + let parsed: Vec = lines.iter().map(|line| serde_json::from_str(line).unwrap()).collect(); + let [in_request, in_retry, own, outside] = parsed.as_slice() else { + panic!("expected four lines: {lines:?}"); + }; + assert_eq!(in_request["x_request_id"], "req-outer"); + assert_eq!(in_request["edge_request_id"], "edge-1"); + assert_eq!(in_request["trace_id"], "trace-1"); + assert_eq!(in_request["spans"][0]["x_request_id"], "req-outer", "the span list is unchanged"); + assert_eq!(in_retry["x_request_id"], "req-inner"); + assert_eq!(own["x_request_id"], "own"); + assert_eq!( + lines[2].matches(r#""x_request_id":"own""#).count(), + 1, + "the event's own field is not repeated: {}", + lines[2] + ); + assert_eq!(own["edge_request_id"], "edge-1"); + assert!(outside.get("x_request_id").is_none(), "{outside}"); + } + /// Run by [`json_logging_writes_every_error_as_one_json_line`] in a child /// process, which owns the global subscriber and panic hook. Does nothing /// in a normal test run. @@ -604,7 +827,7 @@ mod tests { }); log::error!("log crate error"); let _ = std::thread::spawn(|| { - tracing::info_span!("request", request_id = 7).in_scope(|| { + tracing::info_span!("request", request_id = 7, x_request_id = "req-1").in_scope(|| { tracing::info_span!("worker").in_scope(|| panic!("worker panic")); }); }) @@ -663,6 +886,7 @@ mod tests { assert_eq!(panic["spans"][0]["name"], "request"); assert_eq!(panic["spans"][0]["request_id"], 7); assert_eq!(panic["spans"][1]["name"], "worker"); + assert_eq!(panic["x_request_id"], "req-1"); let initialized = find("OTLP export enabled"); assert_eq!(initialized["endpoint"], "http://127.0.0.1:9/v1/traces"); diff --git a/dwctl/src/test/responses/request_id.rs b/dwctl/src/test/responses/request_id.rs index 5928df0dd..17a52e217 100644 --- a/dwctl/src/test/responses/request_id.rs +++ b/dwctl/src/test/responses/request_id.rs @@ -188,6 +188,50 @@ async fn root_span_records_request_ids(pool: sqlx::PgPool) { assert_eq!(attr(&span, "edge_request_id"), None); } +/// In JSON logs every line of a request carries the ID at the top level, +/// under the key other services in the request path use, so a single +/// `| json | x_request_id=""` query finds the whole request. +#[dwctl_test_macros::test] +async fn json_log_lines_carry_request_id_at_top_level(pool: sqlx::PgPool) { + use tracing_subscriber::layer::SubscriberExt; + + let logs = LogBuffer::default(); + let _guard = tracing::subscriber::set_default( + tracing_subscriber::registry() + .with(tracing_subscriber::EnvFilter::new("info")) + .with(crate::telemetry::fmt_layer(crate::config::LogFormat::Json, logs.clone())), + ); + + let mock = wiremock::MockServer::start().await; + super::mount_chat_completions_mock(&mock).await; + let (server, key, _bg) = super::setup_ai_test(pool, &mock, true).await; + let start = logs.0.lock().unwrap().len(); + + let response = server + .post("/ai/v1/chat/completions") + .add_header("Authorization", format!("Bearer {key}")) + .add_header("x-request-id", "edge-123") + .json(&json!({"model":"gpt-4o", "messages":[{"role":"user","content":"hello"}]})) + .await; + response.assert_status_ok(); + let id = response_id(&response).to_string(); + + let text = String::from_utf8_lossy(&logs.0.lock().unwrap()[start..]).into_owned(); + let in_request: Vec = text + .lines() + .map(|line| serde_json::from_str::(line).expect("one JSON object per line")) + .filter(|line| line["spans"][0]["name"] == "request") + .collect(); + assert!( + in_request.iter().any(|line| line["message"] == "finished processing request"), + "request lines logged:\n{text}" + ); + for line in &in_request { + assert_eq!(line["x_request_id"], id.as_str(), "{line}"); + assert_eq!(line["edge_request_id"], "edge-123", "{line}"); + } +} + /// Failures inside the admin AI proxy middleware, which runs before routing, /// still carry our ID. #[dwctl_test_macros::test] From 33d82c5e45f0f79b597303fc21c21bddfd58d94c Mon Sep 17 00:00:00 2001 From: Oloruntobi Allen Date: Wed, 21 Oct 2026 12:18:14 +0100 Subject: [PATCH 4/4] docs(telemetry): say which request IDs and panic spans are conditional Only x_request_id is on every request line. edge_request_id, trace_id and provider_request_id appear once known, and a panic outside every span or inside the subscriber has no span context. Refs DW-966 --- docs/src/reference/configuration.md | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/docs/src/reference/configuration.md b/docs/src/reference/configuration.md index d07d68be3..e897ba7ed 100644 --- a/docs/src/reference/configuration.md +++ b/docs/src/reference/configuration.md @@ -819,18 +819,22 @@ error spans several lines. `json` writes one JSON object per event, for log aggregators: `timestamp`, `level`, `message` and `target`, the event's own fields at the top level, the innermost span under `span` and every enclosing span, root first, under `spans`. A multi-line error stays on one line. In -`json` mode panics are written to stderr in the same shape, including the -spans the panic happened in, with `target` set to `panic` and a message that -starts like Rust's default panic output (`thread '…' panicked at …`). The +`json` mode panics are written to stderr in the same shape, with `target` set +to `panic` and a message that starts like Rust's default panic output +(`thread '…' panicked at …`). They include the spans the panic happened in +when there are any; a panic outside every span, or inside the logging +subscriber itself, has none. The error `dwctl` exits with, including a configuration that fails to load, is logged once as `dwctl exited with an error`. -Every JSON line logged while handling a request also carries the request's -IDs at the top level: `x_request_id` (the ID dwctl mints and returns in the -`X-Request-ID` header and forwards to providers), `edge_request_id` (an -inbound `X-Request-ID`, such as one set by an ingress), `provider_request_id` -and `trace_id`. Analytics logs written after the response carry -`x_request_id` too. Find every line of a request, in dwctl and in any +Every JSON line logged while handling a request carries `x_request_id` at the +top level: the ID dwctl mints, returns in the `X-Request-ID` header and +forwards to providers. Other IDs are added at the top level once they are +known: `edge_request_id` when the request arrived with its own +`X-Request-ID` (for example from an ingress), `trace_id` when the request +has a trace context, and `provider_request_id` on lines logged after the +provider's response returned its own `X-Request-ID`. Analytics logs written +after the response carry `x_request_id` too. Find every line of a request, in dwctl and in any provider that logs `x_request_id` the same way, with one query: ```logql