/** * OpenTelemetry in one file, with nothing to install: W3C trace context in * and out, spans and metrics exported as OTLP/JSON over plain `fetch`. The * forge traces its own requests with it, and every app the forge scaffolds * carries this same file as `src/otel.ts` — one implementation, read by both. * * Configured by the standard variables, so any OpenTelemetry collector works: * * OTEL_EXPORTER_OTLP_ENDPOINT e.g. http://otel-collector:4318 (unset = off) * OTEL_SERVICE_NAME what this process is called in a trace * OTEL_RESOURCE_ATTRIBUTES k=v,k=v added to every span and metric * OTEL_SDK_DISABLED=true off, whatever else is set * * With no endpoint every call here is a cheap no-op: spans are still made * (so code reads the same) but nothing is kept or sent. Export never throws * and never blocks a request; a collector that is down costs a dropped * batch, not an error page. The queue is bounded — the oldest spans go first. */ import { AsyncLocalStorage } from "node:async_hooks"; import { randomBytes } from "node:crypto"; // The real fetch, taken before anyone (an app's own setup) can replace the // global with `tracedFetch`: export and tracedFetch's own call use it, so // neither traces itself. const nativeFetch: typeof fetch = globalThis.fetch.bind(globalThis); export type Attr = string | number | boolean; export type Attrs = Record; export const SpanKind = { internal: 1, server: 2, client: 3 } as const; export type SpanKindName = keyof typeof SpanKind; export interface SpanContext { readonly traceId: string; readonly spanId: string; readonly sampled: boolean; } interface Config { readonly endpoint: string | undefined; readonly service: string; readonly resource: Attrs; } const env = (name: string): string | undefined => { const value = process.env[name]?.trim(); return value === "" ? undefined : value; }; /** `k=v,k=v` as the spec writes it (values may be percent-encoded). */ export function parseResourceAttributes(raw: string | undefined): Attrs { const out: Attrs = {}; for (const pair of (raw ?? "").split(",")) { const at = pair.indexOf("="); if (at <= 0) continue; const key = pair.slice(0, at).trim(); let value = pair.slice(at + 1).trim(); try { value = decodeURIComponent(value); } catch { // A stray % is kept as written. } if (key !== "") out[key] = value; } return out; } function configFromEnv(): Config { const disabled = env("OTEL_SDK_DISABLED") === "true"; const resource = parseResourceAttributes(env("OTEL_RESOURCE_ATTRIBUTES")); const service = env("OTEL_SERVICE_NAME") ?? (typeof resource["service.name"] === "string" ? resource["service.name"] : "unknown_service:bun"); return { endpoint: disabled ? undefined : env("OTEL_EXPORTER_OTLP_ENDPOINT")?.replace(/\/+$/, ""), service, resource, }; } let config: Config = configFromEnv(); /** Re-read the environment, or set explicitly (tests, embedders). */ export function configure(next?: Partial): void { config = { ...configFromEnv(), ...next }; } export const enabled = (): boolean => config.endpoint !== undefined; // ---- trace context ---------------------------------------------------------- const hex = (bytes: number): string => randomBytes(bytes).toString("hex"); const ZERO_TRACE = "0".repeat(32); const ZERO_SPAN = "0".repeat(16); /** `00---`, or undefined for anything malformed. */ export function parseTraceparent( header: string | null | undefined, ): SpanContext | undefined { const match = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/.exec( (header ?? "").trim().toLowerCase(), ); if (match === null) return undefined; const [, version, traceId, spanId, flags] = match as unknown as [ string, string, string, string, string, ]; if (version === "ff" || traceId === ZERO_TRACE || spanId === ZERO_SPAN) return undefined; return { traceId, spanId, sampled: (Number.parseInt(flags, 16) & 1) === 1 }; } export const formatTraceparent = (ctx: SpanContext): string => `00-${ctx.traceId}-${ctx.spanId}-${ctx.sampled ? "01" : "00"}`; // ---- spans ------------------------------------------------------------------ /** Wall-clock nanoseconds, as a decimal string (OTLP/JSON's uint64 form). */ const nowNs = (): bigint => BigInt(Math.round((performance.timeOrigin + performance.now()) * 1e3)) * 1000n; export class Span implements SpanContext { readonly traceId: string; readonly spanId: string; readonly parentSpanId: string | undefined; readonly sampled: boolean; readonly startNs = nowNs(); endNs: bigint | undefined; readonly attributes: Attrs; status: { code: 0 | 1 | 2; message?: string } = { code: 0 }; readonly events: { name: string; timeNs: bigint; attributes: Attrs }[] = []; constructor( public name: string, readonly kind: SpanKindName, parent: SpanContext | undefined, attributes: Attrs, ) { this.traceId = parent?.traceId ?? hex(16); this.parentSpanId = parent?.spanId; this.sampled = parent?.sampled ?? true; this.spanId = hex(8); this.attributes = { ...attributes }; } set(key: string, value: Attr | undefined): this { if (value !== undefined) this.attributes[key] = value; return this; } event(name: string, attributes: Attrs = {}): this { this.events.push({ name, timeNs: nowNs(), attributes }); return this; } /** Mark failed. An exception's message is kept; its stack is not. */ fail(error: unknown): this { const message = error instanceof Error ? error.message : String(error); this.status = { code: 2, message }; this.event("exception", { "exception.type": error instanceof Error ? error.name : typeof error, "exception.message": message, }); return this; } end(): void { if (this.endNs !== undefined) return; this.endNs = nowNs(); if (this.sampled && enabled()) enqueue(this); } get durationSeconds(): number { return Number((this.endNs ?? nowNs()) - this.startNs) / 1e9; } } const active = new AsyncLocalStorage(); /** The span the current code runs inside, if any. */ export const currentSpan = (): Span | undefined => active.getStore(); export interface SpanOptions { readonly kind?: SpanKindName; /** Parent explicitly; default is the current span. `null` = a new trace. */ readonly parent?: SpanContext | null; readonly attributes?: Attrs; } export function startSpan(name: string, options: SpanOptions = {}): Span { const parent = options.parent === null ? undefined : (options.parent ?? currentSpan()); return new Span(name, options.kind ?? "internal", parent, options.attributes ?? {}); } /** * Run `fn` inside a new span: it is current for everything `fn` awaits, it * ends when `fn` settles, and a throw marks it failed (and still throws). */ export async function span( name: string, fn: (span: Span) => T | Promise, options: SpanOptions = {}, ): Promise { const s = startSpan(name, options); try { return await active.run(s, () => fn(s)); } catch (error) { s.fail(error); throw error; } finally { s.end(); } } /** Run `fn` with `s` as the current span, without ending it. */ export const withSpan = (s: Span, fn: () => T): T => active.run(s, fn); // ---- HTTP ------------------------------------------------------------------- /** * A server span for one request, parented on the caller's `traceparent`. * Records `http.server.request.duration` too. `route` names the matched * route template (`/notes/:id`) once it is known — set it on the span. */ export async function serveSpan( request: Request, handle: (span: Span) => R | Promise, attributes: Attrs = {}, ): Promise { const url = new URL(request.url); const s = startSpan(request.method, { kind: "server", parent: parseTraceparent(request.headers.get("traceparent")) ?? null, attributes: { "http.request.method": request.method, "url.path": url.pathname, "url.scheme": request.headers.get("x-forwarded-proto") ?? url.protocol.slice(0, -1), "server.address": ( request.headers.get("x-forwarded-host") ?? request.headers.get("host") ?? url.hostname ).replace(/:\d+$/, ""), "user_agent.original": request.headers.get("user-agent") ?? undefined, ...attributes, }, }); let status = 500; try { const response = await active.run(s, () => handle(s)); // undefined = the socket was upgraded (a websocket): it answered 101. status = response?.status ?? 101; return response; } catch (error) { s.fail(error); throw error; } finally { endServerSpan(s, status); } } /** * Trace a whole `Bun.serve` options object: its `fetch` runs inside * `serveSpan`, everything else (websocket handlers, port, error) untouched. * The one-line retrofit for an app that routes by hand: * * Bun.serve(traceServe({ port, fetch(request, server) { ... } })) * * A route template isn't known here, so spans are named by method only — * set `http.route` on `currentSpan()` from the handler if you route by name. */ export function traceServe< T extends { fetch: (request: Request, ...rest: never[]) => unknown }, >(options: T): T { const inner = options.fetch; return { ...options, fetch(this: unknown, request: Request, ...rest: never[]) { return serveSpan( request, async () => (await inner.call(this, request, ...rest)) as Response | undefined, ); }, }; } /** * Finish a server span made by hand (a framework's request hooks): status, * the `METHOD /route` name, and the duration histogram — what `serveSpan` * does in its `finally`. */ export function endServerSpan(s: Span, status: number): void { const method = s.attributes["http.request.method"]; s.set("http.response.status_code", status); if (status >= 500 && s.status.code === 0) s.status = { code: 2 }; const route = s.attributes["http.route"]; if (typeof route === "string" && typeof method === "string") s.name = `${method} ${route}`; s.end(); requestDuration.record(s.durationSeconds, { "http.request.method": typeof method === "string" ? method : undefined, "http.route": typeof route === "string" ? route : undefined, "http.response.status_code": status, }); } /** * `fetch` that makes a client span and hands the callee our `traceparent`, * so its spans join this trace. Same signature as `fetch`. */ export async function tracedFetch( input: string | URL | Request, init: RequestInit = {}, ): Promise { const request = input instanceof Request ? new Request(input, init) : new Request(String(input), init); const url = new URL(request.url); return span( `${request.method}`, async (s) => { const headers = new Headers(request.headers); headers.set("traceparent", formatTraceparent(s)); const response = await nativeFetch(new Request(request, { headers })); s.set("http.response.status_code", response.status); if (response.status >= 500) s.status = { code: 2 }; return response; }, { kind: "client", attributes: { "http.request.method": request.method, "server.address": url.hostname, "server.port": url.port === "" ? undefined : Number(url.port), "url.full": `${url.origin}${url.pathname}`, }, }, ); } // ---- metrics ---------------------------------------------------------------- const DURATION_BUCKETS = [ 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10, ]; const attrKey = (attrs: Attrs): string => JSON.stringify( Object.entries(attrs) .filter(([, v]) => v !== undefined) .sort(([a], [b]) => (a < b ? -1 : 1)), ); interface Instrument { readonly name: string; readonly unit: string; readonly description: string; toOtlp(startNs: string, nowNs: string): unknown | undefined; } const instruments: Instrument[] = []; const processStartNs = nowNs().toString(); export interface Histogram { record(value: number, attributes?: Attrs): void; } /** A cumulative explicit-bucket histogram. Keep attribute values few. */ export function histogram( name: string, options: { unit?: string; description?: string; buckets?: number[] } = {}, ): Histogram { const bounds = options.buckets ?? DURATION_BUCKETS; const points = new Map< string, { attrs: Attrs; count: number; sum: number; buckets: number[] } >(); instruments.push({ name, unit: options.unit ?? "s", description: options.description ?? "", toOtlp(startNs, timeNs) { if (points.size === 0) return undefined; return { histogram: { aggregationTemporality: 2, dataPoints: [...points.values()].map((p) => ({ attributes: otlpAttrs(p.attrs), startTimeUnixNano: startNs, timeUnixNano: timeNs, count: String(p.count), sum: p.sum, bucketCounts: p.buckets.map(String), explicitBounds: bounds, })), }, }; }, }); return { record(value, attributes = {}) { if (!enabled() || !Number.isFinite(value)) return; const key = attrKey(attributes); let p = points.get(key); if (p === undefined) { p = { attrs: attributes, count: 0, sum: 0, buckets: new Array(bounds.length + 1).fill(0), }; points.set(key, p); } p.count += 1; p.sum += value; let i = bounds.findIndex((b) => value <= b); if (i === -1) i = bounds.length; p.buckets[i] = (p.buckets[i] ?? 0) + 1; }, }; } export interface Counter { add(value?: number, attributes?: Attrs): void; } /** A cumulative monotonic sum. */ export function counter( name: string, options: { unit?: string; description?: string } = {}, ): Counter { const points = new Map(); instruments.push({ name, unit: options.unit ?? "1", description: options.description ?? "", toOtlp(startNs, timeNs) { if (points.size === 0) return undefined; return { sum: { aggregationTemporality: 2, isMonotonic: true, dataPoints: [...points.values()].map((p) => ({ attributes: otlpAttrs(p.attrs), startTimeUnixNano: startNs, timeUnixNano: timeNs, asDouble: p.value, })), }, }; }, }); return { add(value = 1, attributes = {}) { if (!enabled() || !(value >= 0)) return; const key = attrKey(attributes); const p = points.get(key); if (p === undefined) points.set(key, { attrs: attributes, value }); else p.value += value; }, }; } /** A value read at export time: queue depth, rows in a table, players online. */ export function gauge( name: string, read: () => number | { value: number; attributes?: Attrs }[], options: { unit?: string; description?: string } = {}, ): void { instruments.push({ name, unit: options.unit ?? "1", description: options.description ?? "", toOtlp(_startNs, timeNs) { let values: { value: number; attributes?: Attrs }[]; try { const got = read(); values = typeof got === "number" ? [{ value: got }] : got; } catch { return undefined; } if (values.length === 0) return undefined; return { gauge: { dataPoints: values.map((v) => ({ attributes: otlpAttrs(v.attributes ?? {}), timeUnixNano: timeNs, asDouble: v.value, })), }, }; }, }); } /** The semantic-convention HTTP server histogram `serveSpan` records into. */ const requestDuration = histogram("http.server.request.duration", { unit: "s", description: "Duration of HTTP server requests.", }); // ---- export ----------------------------------------------------------------- function otlpValue(value: Attr): Record { if (typeof value === "boolean") return { boolValue: value }; if (typeof value === "number") return Number.isInteger(value) ? { intValue: String(value) } : { doubleValue: value }; return { stringValue: value }; } export function otlpAttrs(attrs: Attrs): unknown[] { return Object.entries(attrs) .filter((e): e is [string, Attr] => e[1] !== undefined) .map(([key, value]) => ({ key, value: otlpValue(value) })); } const resource = () => ({ attributes: otlpAttrs({ "telemetry.sdk.name": "forge-otel", "telemetry.sdk.language": "js", "process.runtime.name": process.versions.bun === undefined ? "node" : "bun", "process.runtime.version": process.versions.bun ?? process.versions.node, ...config.resource, "service.name": config.service, }), }); const SCOPE = { name: "forge-otel", version: "1" }; /** One span in OTLP/JSON (ids are lowercase hex, times decimal ns). */ export function spanToOtlp(s: Span): unknown { return { traceId: s.traceId, spanId: s.spanId, ...(s.parentSpanId === undefined ? {} : { parentSpanId: s.parentSpanId }), name: s.name, kind: SpanKind[s.kind], startTimeUnixNano: s.startNs.toString(), endTimeUnixNano: (s.endNs ?? s.startNs).toString(), attributes: otlpAttrs(s.attributes), events: s.events.map((e) => ({ name: e.name, timeUnixNano: e.timeNs.toString(), attributes: otlpAttrs(e.attributes), })), status: s.status, }; } const MAX_QUEUE = 2048; const BATCH = 512; let queue: Span[] = []; let timer: ReturnType | undefined; let lastMetrics = 0; function enqueue(s: Span): void { queue.push(s); if (queue.length > MAX_QUEUE) queue = queue.slice(-MAX_QUEUE); if (timer === undefined) { timer = setInterval(() => void flush(), 5_000); // Telemetry must never be the reason a process stays alive. (timer as { unref?: () => void }).unref?.(); } } async function post(path: string, body: unknown): Promise { if (config.endpoint === undefined) return false; try { const response = await nativeFetch(`${config.endpoint}${path}`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body), signal: AbortSignal.timeout(5_000), }); return response.ok; } catch { return false; } } /** Send what is queued now. Resolves when sent (or dropped); never throws. */ export async function flush(): Promise { while (queue.length > 0) { const batch = queue.splice(0, BATCH); await post("/v1/traces", { resourceSpans: [ { resource: resource(), scopeSpans: [{ scope: SCOPE, spans: batch.map(spanToOtlp) }] }, ], }); } if (Date.now() - lastMetrics >= 30_000) { lastMetrics = Date.now(); await flushMetrics(); } } export function metricsToOtlp(): unknown | undefined { const timeNs = nowNs().toString(); const metrics = instruments.flatMap((i) => { const data = i.toOtlp(processStartNs, timeNs); return data === undefined ? [] : [{ name: i.name, unit: i.unit, description: i.description, ...(data as object) }]; }); if (metrics.length === 0) return undefined; return { resourceMetrics: [ { resource: resource(), scopeMetrics: [{ scope: SCOPE, metrics }] }, ], }; } export async function flushMetrics(): Promise { if (!enabled()) return; const body = metricsToOtlp(); if (body !== undefined) await post("/v1/metrics", body); } /** Start the periodic exporter even before the first span (metrics-only apps). */ export function start(): void { if (!enabled() || timer !== undefined) return; timer = setInterval(() => void flush(), 5_000); (timer as { unref?: () => void }).unref?.(); }