kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: enrich OTEL log records with per-request LogContext (#8812)

Browse Source 
* fix: enrich OTEL log records with per-request LogContext

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore: add otlp_smoke example for manual OTEL log bridge verification

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore: update ee-repo-ref to 5d6b713b74fc46735807f5c32883002e8d976fbc

This commit updates the EE repository reference after PR #529 was merged in windmill-ee-private.

Previous ee-repo-ref: 45959d063bc941c567488d330b5819601cdd2d3d

New ee-repo-ref: 5d6b713b74fc46735807f5c32883002e8d976fbc

Automated by sync-ee-ref workflow.

* refactor: store LogContext in ArcSwap instead of Mutex

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore: pin ee-repo-ref to ArcSwap branch commit

* chore: update ee-repo-ref to be2f3d4d11bb7110200524d7157caab3aac53996

This commit updates the EE repository reference after PR #530 was merged in windmill-ee-private.

Previous ee-repo-ref: 45b4d7963a9ebcd583d1a87abe7d07d3d521584a

New ee-repo-ref: be2f3d4d11bb7110200524d7157caab3aac53996

Automated by sync-ee-ref workflow.

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
2026-04-13 17:50:50 -04:00
committed by GitHub
parent c889a185d5
commit 42d3e8c789
12 changed files with 455 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/Cargo.lock generated
View File

@@ -16749,6 +16749,7 @@ dependencies = [
"aes-gcm",
"aho-corasick",
"anyhow",
"arc-swap",
"async-recursion",
"async-stream",
"async-trait",

1
backend/Cargo.toml
View File

@@ -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"

2
backend/ee-repo-ref.txt
View File

@@ -1 +1 @@
972893c3870e4c4a70a35748abed282d88904805
be2f3d4d11bb7110200524d7157caab3aac53996

33
backend/windmill-api-auth/src/auth.rs
View File

@@ -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);
}

8
backend/windmill-api/src/lib.rs
View File

@@ -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()

30
backend/windmill-api/src/tracing_init.rs
View File

@@ -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<B> MakeSpan<B> 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
}

1
backend/windmill-common/Cargo.toml
View File

@@ -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

77
backend/windmill-common/examples/otlp_smoke.rs Normal file
View File

@@ -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");
}

1
backend/windmill-common/src/lib.rs
View File

@@ -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;

244
backend/windmill-common/src/log_context.rs Normal file
View File

@@ -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<LogContext>` 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<String>,
pub uri: Option<String>,
pub trace_id: Option<String>,
// Auth (windmill-api-auth/src/auth.rs)
pub email: Option<String>,
pub username: Option<String>,
// Workspace
pub workspace_id: Option<String>,
// Worker / job span (windmill-worker/src/worker.rs)
pub worker: Option<String>,
pub hostname: Option<String>,
pub tag: Option<String>,
pub job_id: Option<String>,
pub parent_job: Option<String>,
pub root_job: Option<String>,
pub script_path: Option<String>,
pub script_hash: Option<String>,
pub job_kind: Option<String>,
pub language: Option<String>,
pub flow_step_id: Option<String>,
pub trigger_kind: Option<String>,
pub trigger: Option<String>,
pub created_by: Option<String>,
}
tokio::task_local! {
pub static LOG_CONTEXT: ArcSwap<LogContext>;
}
/// 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<F>(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<LogContext>` 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<Arc<LogContext>> {
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: F)
where
F: FnOnce(&LogContext) -> LogContext,
{
let _ = LOG_CONTEXT.try_with(|c| {
let current = c.load_full();
let next = Arc::new(f(&current));
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<F>(fut: F) -> tokio::task::JoinHandle<F::Output>
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;
}
}

95
backend/windmill-worker/src/worker.rs
View File

@@ -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 {

11
backend/windmill-worker/src/worker_utils.rs
View File

@@ -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