diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 8a60710a42..12a4f8379a 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -16749,6 +16749,7 @@ dependencies = [ "aes-gcm", "aho-corasick", "anyhow", + "arc-swap", "async-recursion", "async-stream", "async-trait", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index ca820d28bc..2b1a51e2e9 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -404,6 +404,7 @@ argon2 = "^0" quick_cache = "^0" rand = "=0.9.0" rand_core = { version = "^0", features = ["std"] } +arc-swap = "1" ed25519-dalek = { version = "2", features = ["rand_core"] } magic-crypt = "^3" git-version = "^0" diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 5085261744..cca18546c1 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -972893c3870e4c4a70a35748abed282d88904805 +be2f3d4d11bb7110200524d7157caab3aac53996 \ No newline at end of file diff --git a/backend/windmill-api-auth/src/auth.rs b/backend/windmill-api-auth/src/auth.rs index b5b04cb3e8..34474bd35f 100644 --- a/backend/windmill-api-auth/src/auth.rs +++ b/backend/windmill-api-auth/src/auth.rs @@ -225,15 +225,7 @@ impl AuthCache { t_hash, w_id.as_ref(), ) - .map(|x| { - ( - x.owner, - x.email, - x.super_admin, - x.scopes, - x.label, - ) - }) + .map(|x| (x.owner, x.email, x.super_admin, x.scopes, x.label)) .fetch_optional(&self.db) .await .ok() @@ -242,13 +234,7 @@ impl AuthCache { if let Some(user) = user_o { let authed_o = { match user { - ( - Some(owner), - Some(email), - super_admin, - _, - label, - ) if w_id.is_some() => { + (Some(owner), Some(email), super_admin, _, label) if w_id.is_some() => { let username_override = username_override_from_label(label); if let Some((prefix, name)) = owner.split_once('/') { if prefix == "u" { @@ -701,6 +687,21 @@ pub async fn resolve_opt_job_authed( Span::current().record("username", &authed.username.as_str()); Span::current().record("email", &authed.email); + // Mirror into the per-request LogContext so exported OTEL + // LogRecords carry the same identifiers (the log bridge + // doesn't walk span fields — see windmill_common::log_context). + let username_copy = authed.username.clone(); + let email_copy = authed.email.clone(); + let workspace_copy = workspace_id.clone(); + windmill_common::log_context::update_log_context(move |c| { + windmill_common::log_context::LogContext { + username: Some(username_copy), + email: Some(email_copy), + workspace_id: workspace_copy.or_else(|| c.workspace_id.clone()), + ..c.clone() + } + }); + if let Some(workspace_id) = workspace_id { Span::current().record("workspace_id", &workspace_id); } diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 9b02f23cce..24b4574019 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -995,6 +995,14 @@ pub async fn run_server( app }; + // Seed the per-request LogContext task-local. Registered outside + // TraceLayer so MyOnResponse::on_response's `"response"` log fires inside + // the scope and gets method/uri/workspace_id/email attached by the EE + // LogContextBridge. + let app = app.layer(axum::middleware::from_fn( + tracing_init::log_context_middleware, + )); + let app = app.layer(CatchPanicLayer::custom(|err| { tracing::error!("panic in handler, returning 500: {:?}", err); Response::builder() diff --git a/backend/windmill-api/src/tracing_init.rs b/backend/windmill-api/src/tracing_init.rs index de6fba55d2..09a3e5565e 100644 --- a/backend/windmill-api/src/tracing_init.rs +++ b/backend/windmill-api/src/tracing_init.rs @@ -8,9 +8,13 @@ use crate::s3_log_batching::{record_s3_log, S3ProxyRequest}; use ::tracing::{field, Span}; +use axum::extract::Request; +use axum::middleware::Next; +use axum::response::Response as AxumResponse; use hyper::Response; use tower_http::trace::{MakeSpan, OnFailure, OnResponse}; use uuid::Uuid; +use windmill_common::log_context::{with_log_context, LogContext}; lazy_static::lazy_static! { static ref LOG_REQUESTS: bool = std::env::var("LOG_REQUESTS") @@ -86,3 +90,29 @@ impl MakeSpan for MyMakeSpan { ) } } + +/// Axum middleware that seeds a per-request `LogContext` with method/uri/ +/// traceId and wraps the downstream chain in a task-local scope. Auth and +/// workspace-resolution code later mutate this context (via +/// `update_log_context`) as email/username/workspace_id become known. +/// +/// Registered at the top of the router layer stack in `windmill-api/src/lib.rs` +/// so every route — and critically, the `MyOnResponse::on_response` callback +/// that TraceLayer invokes on the way out — runs inside the scope and thus +/// flows through to exported OTEL LogRecords via the EE LogContextBridge. +pub async fn log_context_middleware(request: Request, next: Next) -> AxumResponse { + let trace_id = request + .headers() + .get(TRACING_HEADER.as_str()) + .and_then(|x| x.to_str().ok()) + .map(|s| s.to_string()); + + let ctx = LogContext { + method: Some(request.method().to_string()), + uri: Some(request.uri().to_string()), + trace_id, + ..Default::default() + }; + + with_log_context(ctx, next.run(request)).await +} diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 9e7f4cd42e..c50d537e37 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -115,6 +115,7 @@ aes-gcm = { workspace = true, optional = true } semver.workspace = true croner.workspace = true quick_cache.workspace = true +arc-swap.workspace = true pin-project-lite.workspace = true futures.workspace = true tempfile.workspace = true diff --git a/backend/windmill-common/examples/otlp_smoke.rs b/backend/windmill-common/examples/otlp_smoke.rs new file mode 100644 index 0000000000..0f5f68af5e --- /dev/null +++ b/backend/windmill-common/examples/otlp_smoke.rs @@ -0,0 +1,77 @@ +//! Manual smoke test for the LogContext → OTLP bridge pipeline. +//! +//! Run a local OTEL collector with: +//! docker run -d --rm --name wm-otelcol --network host \ +//! -v /tmp/otel-verify/otelcol-config.yaml:/etc/otelcol-contrib/config.yaml:ro \ +//! -v /tmp/otel-verify:/out \ +//! otel/opentelemetry-collector-contrib:latest +//! +//! Then run this binary: +//! OTEL_LOGS=1 OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \ +//! cargo run --example otlp_smoke --features enterprise,private,otel \ +//! -p windmill-common +//! +//! Inspect the collector's debug output with: +//! docker logs wm-otelcol +//! Or the file sink: +//! cat /tmp/otel-verify/logs.json + +use windmill_common::log_context::{update_log_context, with_log_context, LogContext}; +use windmill_common::utils::Mode; + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + let (_guard, _meter_provider) = + windmill_common::tracing_init::initialize_tracing("otlp-smoke-host", &Mode::Server, "dev"); + + // Simulate an HTTP request: middleware seeds method/uri, then the auth + // chain adds email/username/workspace_id. + let ctx = LogContext { + method: Some("POST".into()), + uri: Some("/api/w/acme/jobs/run/f/foo".into()), + trace_id: Some("smoke-trace-id-abc-123".into()), + ..Default::default() + }; + + with_log_context(ctx, async { + // Simulate auth middleware running: + update_log_context(|c| LogContext { + email: Some("alice@acme.co".into()), + username: Some("alice".into()), + workspace_id: Some("acme".into()), + ..c.clone() + }); + + // This mimics MyOnResponse::on_response for a 500 error: + tracing::error!(status = 500u16, latency = 123u64, "response"); + + // And an ad-hoc handler error: + tracing::error!("database connection refused"); + }) + .await; + + // Simulate a worker job context: + let job_ctx = LogContext { + workspace_id: Some("acme".into()), + job_id: Some("00000000-0000-4000-8000-000000000001".into()), + script_path: Some("f/ingest/run".into()), + job_kind: Some("script".into()), + language: Some("python3".into()), + worker: Some("wk-default-smoke".into()), + hostname: Some("otlp-smoke-host".into()), + tag: Some("deno".into()), + ..Default::default() + }; + with_log_context(job_ctx, async { + tracing::error!("job failed: ValueError: bad input"); + }) + .await; + + // And one event outside any scope to prove it still emits (sans fields): + tracing::error!("background task failure (no context)"); + + // Give the batch exporter time to flush before we exit. + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + + eprintln!("smoke emitted; check `docker logs wm-otelcol` for the debug exporter output"); +} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 6f71090124..045bc8632c 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -66,6 +66,7 @@ pub mod global_settings; pub mod indexer; pub mod instance_config; pub mod job_metrics; +pub mod log_context; pub mod min_version; pub mod notify_events; pub mod runtime_assets; diff --git a/backend/windmill-common/src/log_context.rs b/backend/windmill-common/src/log_context.rs new file mode 100644 index 0000000000..4a517c7afb --- /dev/null +++ b/backend/windmill-common/src/log_context.rs @@ -0,0 +1,244 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +//! Per-task enrichment context for OTEL log export. +//! +//! This is the MDC-equivalent for windmill: a tokio task-local that carries +//! request/job identifiers from the entry point (axum middleware in +//! windmill-api, job execution wrapper in windmill-worker) to log emission +//! sites. The EE OTEL log bridge reads this at event time and attaches the +//! fields to exported LogRecords, so Sentry / OTLP backends can filter logs +//! by workspace_id, email, script_path, etc., without joining against +//! traces. +//! +//! The context is stored as `ArcSwap` inside the task-local, +//! which means reads are effectively free (atomic load of an Arc) and +//! writes are immutable swaps. There is no lock — so no reentrant-deadlock +//! risk, no mutex poisoning, and no stale-read race. The tradeoff is that +//! `update_log_context` takes an `FnOnce(&LogContext) -> LogContext` +//! closure that must return a new context value instead of mutating the +//! existing one in place. + +use arc_swap::ArcSwap; +use std::sync::Arc; + +/// Fields promoted from request/job context onto exported log records. +#[derive(Clone, Debug, Default)] +pub struct LogContext { + // HTTP request span (windmill-api/src/tracing_init.rs) + pub method: Option, + pub uri: Option, + pub trace_id: Option, + + // Auth (windmill-api-auth/src/auth.rs) + pub email: Option, + pub username: Option, + + // Workspace + pub workspace_id: Option, + + // Worker / job span (windmill-worker/src/worker.rs) + pub worker: Option, + pub hostname: Option, + pub tag: Option, + pub job_id: Option, + pub parent_job: Option, + pub root_job: Option, + pub script_path: Option, + pub script_hash: Option, + pub job_kind: Option, + pub language: Option, + pub flow_step_id: Option, + pub trigger_kind: Option, + pub trigger: Option, + pub created_by: Option, +} + +tokio::task_local! { + pub static LOG_CONTEXT: ArcSwap; +} + +/// Run a future inside a freshly-seeded LogContext scope. +/// +/// Use at request/job entry points. Downstream code can read the context +/// via [`current_log_context`] and mutate it via [`update_log_context`]. +pub async fn with_log_context(ctx: LogContext, fut: F) -> F::Output +where + F: std::future::Future, +{ + LOG_CONTEXT.scope(ArcSwap::from_pointee(ctx), fut).await +} + +/// Snapshot the current LogContext. Returns `None` outside any scope. +/// +/// The returned `Arc` is a zero-cost view of the context *at +/// the moment of the call*. Subsequent `update_log_context` calls on the +/// same task will atomically swap in a new Arc — this snapshot is +/// unaffected. +pub fn current_log_context() -> Option> { + LOG_CONTEXT.try_with(|c| c.load_full()).ok() +} + +/// Replace the current LogContext with a new value derived from the +/// previous one. No-op outside a scope. +/// +/// The closure receives the current context by reference and must return +/// a new owned context. Callers typically use functional record update +/// syntax: +/// +/// ```ignore +/// update_log_context(|c| LogContext { +/// email: Some("alice@acme.co".into()), +/// ..(**c).clone() +/// }); +/// ``` +/// +/// There is no lock held while the closure runs, so reentrant calls +/// (`update_log_context` inside the closure) are safe — they'll operate on +/// the intermediate state and may lose the intermediate write if the +/// enclosing call races, but they cannot deadlock. +pub fn update_log_context(f: F) +where + F: FnOnce(&LogContext) -> LogContext, +{ + let _ = LOG_CONTEXT.try_with(|c| { + let current = c.load_full(); + let next = Arc::new(f(¤t)); + c.store(next); + }); +} + +/// Spawn a future with a snapshot of the current LogContext forwarded to +/// the new task. Use in place of `tokio::spawn` when you want the spawned +/// task's logs to inherit the calling task's context. +/// +/// The snapshot is a shared `Arc` captured at spawn time — mutations in +/// either task after that point are independent because each task's +/// `ArcSwap` is its own task-local slot. +pub fn spawn_with_log_context(fut: F) -> tokio::task::JoinHandle +where + F: std::future::Future + Send + 'static, + F::Output: Send + 'static, +{ + let snapshot = current_log_context(); + tokio::spawn(async move { + match snapshot { + Some(arc) => LOG_CONTEXT.scope(ArcSwap::new(arc), fut).await, + None => fut.await, + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn reads_and_mutates_current_context() { + let ctx = LogContext { + workspace_id: Some("acme".into()), + email: Some("alice@acme.co".into()), + ..Default::default() + }; + with_log_context(ctx, async { + let snap = current_log_context().expect("in scope"); + assert_eq!(snap.workspace_id.as_deref(), Some("acme")); + assert_eq!(snap.email.as_deref(), Some("alice@acme.co")); + + update_log_context(|c| LogContext { + script_path: Some("f/ingest/run".into()), + ..c.clone() + }); + + let snap = current_log_context().expect("in scope"); + assert_eq!(snap.script_path.as_deref(), Some("f/ingest/run")); + // Old fields are preserved by functional update: + assert_eq!(snap.workspace_id.as_deref(), Some("acme")); + }) + .await; + } + + #[tokio::test] + async fn outside_scope_returns_none() { + assert!(current_log_context().is_none()); + // update_log_context is a no-op outside a scope, doesn't panic + update_log_context(|c| LogContext { workspace_id: Some("ignored".into()), ..c.clone() }); + assert!(current_log_context().is_none()); + } + + #[tokio::test] + async fn spawn_inherits_snapshot() { + let ctx = LogContext { workspace_id: Some("acme".into()), ..Default::default() }; + with_log_context(ctx, async { + let handle = spawn_with_log_context(async { + current_log_context() + .expect("inherited in spawned task") + .workspace_id + .clone() + }); + let ws = handle.await.unwrap(); + assert_eq!(ws.as_deref(), Some("acme")); + }) + .await; + } + + #[tokio::test] + async fn bare_spawn_has_no_context() { + let ctx = LogContext { workspace_id: Some("acme".into()), ..Default::default() }; + with_log_context(ctx, async { + let handle = tokio::spawn(async { current_log_context() }); + assert!( + handle.await.unwrap().is_none(), + "bare tokio::spawn drops context" + ); + }) + .await; + } + + #[tokio::test] + async fn spawn_snapshot_is_independent_of_parent_mutations() { + let ctx = LogContext { workspace_id: Some("parent".into()), ..Default::default() }; + with_log_context(ctx, async { + let handle = spawn_with_log_context(async { + tokio::task::yield_now().await; + current_log_context().unwrap().workspace_id.clone() + }); + update_log_context(|c| LogContext { + workspace_id: Some("mutated".into()), + ..c.clone() + }); + let child_ws = handle.await.unwrap(); + assert_eq!(child_ws.as_deref(), Some("parent")); + }) + .await; + } + + #[tokio::test] + async fn update_preserves_other_fields() { + let ctx = LogContext { + method: Some("POST".into()), + uri: Some("/api/w/acme/jobs".into()), + workspace_id: Some("acme".into()), + ..Default::default() + }; + with_log_context(ctx, async { + update_log_context(|c| LogContext { + email: Some("alice@acme.co".into()), + username: Some("alice".into()), + ..c.clone() + }); + let snap = current_log_context().unwrap(); + assert_eq!(snap.method.as_deref(), Some("POST")); + assert_eq!(snap.uri.as_deref(), Some("/api/w/acme/jobs")); + assert_eq!(snap.workspace_id.as_deref(), Some("acme")); + assert_eq!(snap.email.as_deref(), Some("alice@acme.co")); + assert_eq!(snap.username.as_deref(), Some("alice")); + }) + .await; + } +} diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 21c4648569..de50abce78 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1289,12 +1289,17 @@ fn add_outstanding_wait_time( if let Some(db) = conn.as_sql() { let db = db.clone(); - tokio::spawn(async move { - match insert_wait_time(job_id, root_job_id, &db, wait_time).await { - Ok(()) => tracing::warn!("job {job_id} waited for an executor for a significant amount of time. Recording value wait_time={}ms", wait_time), - Err(e) => tracing::error!("Failed to insert outstanding wait time: {}", e), + let span = tracing::Span::current(); + windmill_common::log_context::spawn_with_log_context(async move { + async move { + match insert_wait_time(job_id, root_job_id, &db, wait_time).await { + Ok(()) => tracing::warn!("job {job_id} waited for an executor for a significant amount of time. Recording value wait_time={}ms", wait_time), + Err(e) => tracing::error!("Failed to insert outstanding wait time: {}", e), + } } - }.in_current_span()); + .instrument(span) + .await + }); } } @@ -1373,6 +1378,38 @@ pub fn create_span_with_name( span } +/// Build the per-job `LogContext` that gets seeded at the top of job +/// execution. Mirrors the field set recorded on the `"job"` tracing span in +/// `create_span_with_name` so exported log records and traces carry the +/// same identifiers. +pub fn log_context_for_job( + arc_job: &MiniPulledJob, + worker_name: &str, + hostname: Option<&str>, +) -> windmill_common::log_context::LogContext { + let existing = windmill_common::log_context::current_log_context() + .map(|arc| (*arc).clone()) + .unwrap_or_default(); + windmill_common::log_context::LogContext { + job_id: Some(arc_job.id.to_string()), + workspace_id: Some(arc_job.workspace_id.clone()), + worker: Some(worker_name.to_string()), + tag: Some(arc_job.tag.clone()), + job_kind: Some(arc_job.kind.as_str().to_string()), + created_by: Some(arc_job.created_by.clone()), + script_path: arc_job.runnable_path.clone(), + script_hash: arc_job.runnable_id.map(|h| h.to_string()), + language: arc_job.script_lang.map(|l| l.as_str().to_string()), + flow_step_id: arc_job.flow_step_id.clone(), + parent_job: arc_job.parent_job.map(|id| id.to_string()), + root_job: arc_job.flow_innermost_root_job.map(|id| id.to_string()), + trigger_kind: arc_job.trigger_kind.as_ref().map(|k| k.to_string()), + trigger: arc_job.trigger.clone(), + hostname: hostname.map(|h| h.to_string()), + ..existing + } +} + pub async fn handle_all_job_kind_error( conn: &Connection, authed_client: &AuthedClient, @@ -2782,30 +2819,34 @@ pub async fn run_worker( windmill_common::sensitive_log_masks::register_running_job(arc_job.id); let span = create_span_with_name(&arc_job, &worker_name, Some(hostname), "job"); + let log_ctx = log_context_for_job(&arc_job, &worker_name, Some(hostname)); - let job_result = handle_queued_job( - arc_job.clone(), - raw_code, - raw_lock, - raw_flow, - parent_runnable_path, - &conn, - &authed_client, - hostname, - &worker_name, - &worker_dir, - &job_dir, - Some(same_worker_tx.clone()), - base_internal_url, - job_completed_tx.clone(), - &mut occupancy_metrics, - &mut killpill_rx2, - precomputed_bundle, - flow_runners, - #[cfg(feature = "benchmark")] - &mut bench, + let job_result = windmill_common::log_context::with_log_context( + log_ctx, + handle_queued_job( + arc_job.clone(), + raw_code, + raw_lock, + raw_flow, + parent_runnable_path, + &conn, + &authed_client, + hostname, + &worker_name, + &worker_dir, + &job_dir, + Some(same_worker_tx.clone()), + base_internal_url, + job_completed_tx.clone(), + &mut occupancy_metrics, + &mut killpill_rx2, + precomputed_bundle, + flow_runners, + #[cfg(feature = "benchmark")] + &mut bench, + ) + .instrument(span), ) - .instrument(span) .await; match job_result { diff --git a/backend/windmill-worker/src/worker_utils.rs b/backend/windmill-worker/src/worker_utils.rs index 260f03f3a1..4845bc0adc 100644 --- a/backend/windmill-worker/src/worker_utils.rs +++ b/backend/windmill-worker/src/worker_utils.rs @@ -353,8 +353,8 @@ pub(crate) async fn queue_vacuum(conn: &Connection, worker_name: &str, hostname: let current_span = tracing::Span::current(); let worker_name = worker_name.to_string(); let hostname = hostname.to_string(); - tokio::task::spawn( - (async move { + windmill_common::log_context::spawn_with_log_context(async move { + async move { tracing::info!(worker = %worker_name, hostname = %hostname, "vacuuming queue"); if let Err(e) = sqlx::query!("VACUUM (SKIP_LOCKED) v2_job_queue, v2_job_runtime, v2_job_status, job_perms") .execute(&db2) @@ -363,9 +363,10 @@ pub(crate) async fn queue_vacuum(conn: &Connection, worker_name: &str, hostname: tracing::error!(worker = %worker_name, hostname = %hostname, "failed to vacuum queue: {}", e); } tracing::info!(worker = %worker_name, hostname = %hostname, "vacuumed queue"); - }) - .instrument(current_span), - ); + } + .instrument(current_span) + .await + }); } Connection::Http(_) => { // do nothing in http mode