diff --git a/Cargo.lock b/Cargo.lock index 3b4a3fb81..4662f685e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7386,6 +7386,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" @@ -7396,12 +7406,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..e897ba7ed 100644 --- a/docs/src/reference/configuration.md +++ b/docs/src/reference/configuration.md @@ -808,6 +808,42 @@ 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 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, 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 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 +{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. + ## Sample Files Generate sample JSONL files for new users: diff --git a/dwctl/Cargo.toml b/dwctl/Cargo.toml index e13738af6..2fe26c7b7 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 bd257c46d..b9d085863 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?; @@ -2782,8 +2782,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/main.rs b/dwctl/src/main.rs index 3b6967a17..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 @@ -121,21 +148,33 @@ 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(); + let _ = telemetry::init_telemetry(false, log_format); + return Err(anyhow::Error::from(error).context("Failed to load configuration")); + } + }; + + // 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 { @@ -143,16 +182,7 @@ 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 diff --git a/dwctl/src/request_logging/analytics_handler.rs b/dwctl/src/request_logging/analytics_handler.rs index 55e7f525c..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 @@ -265,8 +272,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(), client_disconnected = client_disconnected(&response_data), "Successful generative response recorded zero tokens" ); diff --git a/dwctl/src/telemetry.rs b/dwctl/src/telemetry.rs index 052840ada..d290ea7c1 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 @@ -97,16 +104,25 @@ //! ``` 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::{Map, Value, json}; +use std::backtrace::{Backtrace, BacktraceStatus}; use std::collections::HashMap; -use tracing::{Subscriber, info}; +use std::io::Write; +use std::panic::PanicHookInfo; +use tracing::field::{Field, Visit}; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Event, Subscriber, info}; use tracing_opentelemetry::OpenTelemetrySpanExt; -use tracing_subscriber::layer::SubscriberExt; -use tracing_subscriber::registry::LookupSpan; +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}; @@ -118,6 +134,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 @@ -208,40 +225,299 @@ 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> { +/// +/// 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 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), + }; - 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()); + tracing_subscriber::registry() + .with(env_filter) + .with(fmt_layer(log_format, std::io::stdout)) + .with(otel) + .try_init()?; - tracing_subscriber::registry() - .with(env_filter) - .with(tracing_subscriber::fmt::layer().compact()) - .with(otel_layer(tracer, &traces_filter)) - .try_init()?; + if log_format == LogFormat::Json { + std::panic::set_hook(Box::new(write_panic_as_json)); + } - 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 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)"), + } + + 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`. 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, +{ + let layer = tracing_subscriber::fmt::layer().with_writer(writer); + match format { + LogFormat::Compact => layer.compact().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); - info!("Telemetry initialized (OTLP export disabled)"); +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) } +} + +/// 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(), current_spans()); + let _ = writeln!(std::io::stderr().lock(), "{line}"); +} - Ok(None) +/// 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 { + tracing::dispatcher::get_default(|dispatch| { + let registry = dispatch.downcast_ref::()?; + let id = dispatch.current_span().id()?.clone(); + let span = registry.span(&id)?; + 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 + .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, spans: Option) -> 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 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(); + } + line } /// The span export layer: `tracer` behind its own `directives` filter, without @@ -273,7 +549,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()); @@ -286,9 +562,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") { @@ -304,9 +577,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, @@ -332,9 +606,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)) + Ok((tracer, tracer_provider, settings)) +} + +/// What [`create_otlp_tracer`] configured, logged once logging is up. +struct OtlpSettings { + service_name: String, + endpoint: String, + header_count: usize, } #[cfg(test)] @@ -437,7 +723,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)", ); @@ -457,4 +743,160 @@ mod tests { result.unwrap_err() ); } + + /// 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. + #[test] + fn json_logging_child() { + if std::env::var_os(JSON_LOGGING_CHILD).is_none() { + return; + } + // 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(|| { + tracing::info_span!("request", request_id = 7, x_request_id = "req-1").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`, 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); + 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}" + ); + 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"); + 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"); + assert!( + !stdout.contains("[OTLP]") && !stderr.contains("[OTLP]"), + "OTLP setup is logged as an event, not printed: {stderr}" + ); + } } 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] 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,