From f7a8c85b05c604a327cd853392309b9567c72edc Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Wed, 7 Jan 2026 04:47:54 +0000 Subject: [PATCH] feat: Add Bun OTel auto-instrumentation loader and EE dependencies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add otel_bun_loader.js for automatic fetch span tracing in Bun scripts - Inject OTel loader via -r flag when WINDMILL_OTEL_AUTO_INSTRUMENTATION is set - Add prost and axum as optional enterprise dependencies for OTLP parsing - Fix ansible_executor.rs compilation warnings for OSS build 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- backend/Cargo.lock | 2 + backend/Cargo.toml | 1 + backend/windmill-worker/Cargo.toml | 4 +- backend/windmill-worker/otel_bun_loader.js | 191 ++++++++++++++++++ .../windmill-worker/src/ansible_executor.rs | 7 +- backend/windmill-worker/src/bun_executor.rs | 65 ++++-- 6 files changed, 251 insertions(+), 19 deletions(-) create mode 100644 backend/windmill-worker/otel_bun_loader.js diff --git a/backend/Cargo.lock b/backend/Cargo.lock index fdc816ccfa..e103c2ab86 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15814,6 +15814,7 @@ dependencies = [ "aws-credential-types", "aws-sdk-bedrockruntime", "aws-smithy-types", + "axum", "backon", "base64 0.22.1", "bit-vec 0.6.3", @@ -15863,6 +15864,7 @@ dependencies = [ "postgres-native-tls 0.5.1", "process-wrap", "prometheus", + "prost", "rand 0.9.0", "regex", "reqwest 0.12.24", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 6fea14e376..c3cc041417 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -450,3 +450,4 @@ oracle = { version = "0.6.3", features = ["chrono"] } rumqttc = { version = "0.24.0", features = ["use-native-tls"]} strum = { version = "0.27", features = ["derive"] } strum_macros = "^0" +prost = "0.13.5" diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 19ebb88db5..8316a25bb0 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -12,7 +12,7 @@ path = "src/lib.rs" default = [] private = [] prometheus = ["dep:prometheus", "windmill-common/prometheus"] -enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:pem", "dep:tokio-util"] +enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:pem", "dep:tokio-util", "dep:prost", "dep:axum"] mssql = ["dep:tiberius"] bigquery = ["dep:gcp_auth"] benchmark = ["windmill-queue/benchmark", "windmill-common/benchmark"] @@ -138,6 +138,8 @@ libloading = { workspace = true, optional = true } opentelemetry = { workspace = true, optional = true } bollard = { workspace = true, optional = true } oracle = { workspace = true, optional = true } +prost = { workspace = true, optional = true } +axum = { workspace = true, optional = true } [build-dependencies] deno_fetch = { workspace = true, optional = true } diff --git a/backend/windmill-worker/otel_bun_loader.js b/backend/windmill-worker/otel_bun_loader.js new file mode 100644 index 0000000000..cb4b420d75 --- /dev/null +++ b/backend/windmill-worker/otel_bun_loader.js @@ -0,0 +1,191 @@ +// OpenTelemetry Auto-Instrumentation Loader for Bun/Node.js +// This file is loaded via -r flag when WINDMILL_OTEL_AUTO_INSTRUMENTATION=true +// It wraps the global fetch to automatically create OTel spans + +const OTEL_ENDPOINT = process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT; +const JOB_ID = process.env.WINDMILL_JOB_ID; +const WORKSPACE_ID = process.env.WINDMILL_WORKSPACE_ID; +const SERVICE_NAME = process.env.OTEL_SERVICE_NAME || 'windmill-script'; + +if (!OTEL_ENDPOINT || !JOB_ID || !WORKSPACE_ID) { + // Skip instrumentation if env vars not set + console.log('[OTel] Missing environment variables, skipping instrumentation'); +} else { + console.log(`[OTel] Auto-instrumentation enabled, sending traces to ${OTEL_ENDPOINT}`); + + // Simple trace context generator + function generateTraceId() { + const bytes = new Uint8Array(16); + crypto.getRandomValues(bytes); + return Array.from(bytes).map(b => b.toString(16).padStart(2, '0')).join(''); + } + + function generateSpanId() { + const bytes = new Uint8Array(8); + crypto.getRandomValues(bytes); + return Array.from(bytes).map(b => b.toString(16).padStart(2, '0')).join(''); + } + + // Store for pending spans + const pendingSpans = []; + let flushTimeout = null; + + // Encode spans to OTLP protobuf format (simplified JSON export for now) + async function flushSpans() { + if (pendingSpans.length === 0) return; + + const spansToSend = pendingSpans.splice(0, pendingSpans.length); + + // Build OTLP JSON format (will be converted to protobuf by collector or use JSON endpoint) + const jsonEndpoint = OTEL_ENDPOINT.replace('/v1/traces', '/v1/traces'); + + const exportRequest = { + resourceSpans: [{ + resource: { + attributes: [ + { key: 'service.name', value: { stringValue: SERVICE_NAME } }, + { key: 'windmill.job_id', value: { stringValue: JOB_ID } }, + { key: 'windmill.workspace_id', value: { stringValue: WORKSPACE_ID } } + ] + }, + scopeSpans: [{ + scope: { + name: 'windmill-otel-bun-loader', + version: '1.0.0' + }, + spans: spansToSend + }] + }] + }; + + try { + // Use the original fetch to avoid recursion + await originalFetch(jsonEndpoint, { + method: 'POST', + headers: { + 'Content-Type': 'application/json' + }, + body: JSON.stringify(exportRequest) + }); + } catch (e) { + // Silently ignore export errors to not affect the script + console.error('[OTel] Failed to export spans:', e.message); + } + } + + function scheduleFlush() { + if (flushTimeout) return; + flushTimeout = setTimeout(() => { + flushTimeout = null; + flushSpans(); + }, 100); + } + + // Wrap the global fetch + const originalFetch = globalThis.fetch; + + globalThis.fetch = async function instrumentedFetch(input, init) { + const url = typeof input === 'string' ? input : input.url; + const method = init?.method || (typeof input === 'object' ? input.method : 'GET') || 'GET'; + + // Don't instrument calls to the OTel endpoint to avoid recursion + if (url.includes(OTEL_ENDPOINT) || url.includes('/v1/traces')) { + return originalFetch(input, init); + } + + const traceId = generateTraceId(); + const spanId = generateSpanId(); + const startTimeNano = BigInt(Date.now()) * 1000000n; + + let statusCode = 0; // UNSET + let statusMessage = ''; + let responseStatus = 0; + let error = null; + + try { + const response = await originalFetch(input, init); + responseStatus = response.status; + + if (response.ok) { + statusCode = 1; // OK + } else { + statusCode = 2; // ERROR + statusMessage = `HTTP ${response.status}`; + } + + return response; + } catch (e) { + statusCode = 2; // ERROR + statusMessage = e.message; + error = e; + throw e; + } finally { + const endTimeNano = BigInt(Date.now()) * 1000000n; + + // Parse URL for attributes + let parsedUrl; + try { + parsedUrl = new URL(url); + } catch { + parsedUrl = { hostname: '', pathname: url, protocol: '' }; + } + + const span = { + traceId: hexToBase64(traceId), + spanId: hexToBase64(spanId), + name: `HTTP ${method}`, + kind: 3, // CLIENT + startTimeUnixNano: startTimeNano.toString(), + endTimeUnixNano: endTimeNano.toString(), + attributes: [ + { key: 'http.method', value: { stringValue: method } }, + { key: 'http.url', value: { stringValue: url } }, + { key: 'http.host', value: { stringValue: parsedUrl.hostname } }, + { key: 'http.target', value: { stringValue: parsedUrl.pathname } }, + { key: 'http.scheme', value: { stringValue: parsedUrl.protocol?.replace(':', '') || 'https' } } + ], + status: { + code: statusCode, + message: statusMessage + } + }; + + if (responseStatus) { + span.attributes.push({ key: 'http.status_code', value: { intValue: responseStatus.toString() } }); + } + + if (error) { + span.events = [{ + timeUnixNano: endTimeNano.toString(), + name: 'exception', + attributes: [ + { key: 'exception.message', value: { stringValue: error.message } }, + { key: 'exception.type', value: { stringValue: error.name } } + ] + }]; + } + + pendingSpans.push(span); + scheduleFlush(); + } + }; + + // Helper to convert hex to base64 for OTLP JSON format + function hexToBase64(hex) { + const bytes = new Uint8Array(hex.match(/.{1,2}/g).map(byte => parseInt(byte, 16))); + return btoa(String.fromCharCode(...bytes)); + } + + // Ensure spans are flushed before process exits + process.on('beforeExit', async () => { + if (flushTimeout) { + clearTimeout(flushTimeout); + flushTimeout = null; + } + await flushSpans(); + }); + + process.on('exit', () => { + // Synchronous - can't do async here, but beforeExit should have handled it + }); +} diff --git a/backend/windmill-worker/src/ansible_executor.rs b/backend/windmill-worker/src/ansible_executor.rs index 3da3fa619d..c43b575ce5 100644 --- a/backend/windmill-worker/src/ansible_executor.rs +++ b/backend/windmill-worker/src/ansible_executor.rs @@ -13,12 +13,13 @@ use tokio::process::Command; use uuid::Uuid; use windmill_common::{ error, - git_sync_oss::prepend_token_to_github_url, worker::{ is_allowed_file_location, split_python_requirements, to_raw_value, write_file, write_file_at_user_defined_location, Connection, PyVAlias, WORKER_CONFIG, }, }; +#[cfg(feature = "enterprise")] +use windmill_common::git_sync_oss::prepend_token_to_github_url; use windmill_queue::MiniPulledJob; use windmill_parser_yaml::{ @@ -948,8 +949,12 @@ pub async fn handle_ansible_job( )); }; + #[cfg(feature = "enterprise")] let mut secret_url = git_repo_resource.get("url").and_then(|s| s.as_str()).map(|s| s.to_string()) .ok_or(anyhow!("Failed to get url from git repo resource, please check that the resource has the correct type (git_repository)"))?; + #[cfg(not(feature = "enterprise"))] + let secret_url = git_repo_resource.get("url").and_then(|s| s.as_str()).map(|s| s.to_string()) + .ok_or(anyhow!("Failed to get url from git repo resource, please check that the resource has the correct type (git_repository)"))?; #[cfg(feature = "enterprise")] let is_github_app = git_repo_resource.get("is_github_app").and_then(|s| s.as_bool()) diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 229101a683..b0690f440c 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -64,6 +64,8 @@ const RELATIVE_BUN_LOADER: &str = include_str!("../loader.bun.js"); const RELATIVE_BUN_BUILDER: &str = include_str!("../loader_builder.bun.js"); +const OTEL_BUN_LOADER: &str = include_str!("../otel_bun_loader.js"); + const NSJAIL_CONFIG_RUN_BUN_CONTENT: &str = include_str!("../nsjail/run.bun.config.proto"); pub const BUN_LOCK_SPLIT: &str = "\n//bun.lock\n"; @@ -1377,6 +1379,12 @@ try {{ vec![] }; + // Write OTel loader file if auto-instrumentation is enabled + let otel_enabled = !otel_envs.is_empty(); + if otel_enabled { + write_file(job_dir, "otel_bun_loader.js", OTEL_BUN_LOADER)?; + } + //do not cache local dependencies let child = if !*DISABLE_NSJAIL { let _ = write_file( @@ -1409,17 +1417,22 @@ try {{ "/tmp/nodejs/wrapper.mjs", ] } else if codebase.is_some() || has_bundle_cache { - vec![ + let mut base_args = vec![ "--config", "run.config.proto", "--", &BUN_PATH, "run", "--preserve-symlinks", - "/tmp/bun/wrapper.mjs", - ] + ]; + // Add OTel loader if enabled + if otel_enabled { + base_args.extend_from_slice(&["-r", "/tmp/bun/otel_bun_loader.js"]); + } + base_args.push("/tmp/bun/wrapper.mjs"); + base_args } else { - vec![ + let mut base_args = vec![ "--config", "run.config.proto", "--", @@ -1429,8 +1442,13 @@ try {{ "--prefer-offline", "-r", "/tmp/bun/loader.bun.js", - "/tmp/bun/wrapper.mjs", - ] + ]; + // Add OTel loader if enabled + if otel_enabled { + base_args.extend_from_slice(&["-r", "/tmp/bun/otel_bun_loader.js"]); + } + base_args.push("/tmp/bun/wrapper.mjs"); + base_args }; nsjail_cmd .current_dir(job_dir) @@ -1467,20 +1485,33 @@ try {{ bun_cmd } else { let script_path = format!("{job_dir}/wrapper.mjs"); + let otel_loader_path = format!("{job_dir}/otel_bun_loader.js"); - let args: Vec<&str> = if codebase.is_some() || has_bundle_cache { - vec!["run", &script_path] + let args: Vec = if codebase.is_some() || has_bundle_cache { + let mut base_args = vec!["run".to_string()]; + // Add OTel loader if enabled + if otel_enabled { + base_args.extend_from_slice(&["-r".to_string(), otel_loader_path.clone()]); + } + base_args.push(script_path); + base_args } else { - vec![ - "run", - "-i", - "--prefer-offline", - "-r", - "./loader.bun.js", - &script_path, - ] + let mut base_args = vec![ + "run".to_string(), + "-i".to_string(), + "--prefer-offline".to_string(), + "-r".to_string(), + "./loader.bun.js".to_string(), + ]; + // Add OTel loader if enabled + if otel_enabled { + base_args.extend_from_slice(&["-r".to_string(), otel_loader_path.clone()]); + } + base_args.push(script_path); + base_args }; - let mut bun_cmd = build_command_with_isolation(&*BUN_PATH, &args); + let args_refs: Vec<&str> = args.iter().map(|s| s.as_str()).collect(); + let mut bun_cmd = build_command_with_isolation(&*BUN_PATH, &args_refs); bun_cmd .current_dir(job_dir) .env_clear()