mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-19 08:03:16 +00:00
feat: Add Bun OTel auto-instrumentation loader and EE dependencies
- 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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.5
parent
e38b316aad
commit
f7a8c85b05
Generated
+2
@@ -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",
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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
|
||||
});
|
||||
}
|
||||
@@ -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())
|
||||
|
||||
@@ -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<String> = 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()
|
||||
|
||||
Reference in New Issue
Block a user