fable/val Public Public: anyone can read this; nobody can change it.

forked from benjivers/val

workbenchthis machine

otel: every request a span (traceServe), to the platform's traces

src/otel.ts is the forge's zero-dependency OpenTelemetry (apps/forge
src/otel/otel.ts). Bun.serve(traceServe({...})) makes each request a
server span parented on the forge's, exported to the collector the forge
hands the app (OTEL_* env). No endpoint, no-op.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Benjivers-bit committed f9684ed Browse files
2 files changed +641 −2
src/otel.ts added +638
@@ -0,0 +1,638 @@
1 /**
2 * OpenTelemetry in one file, with nothing to install: W3C trace context in
3 * and out, spans and metrics exported as OTLP/JSON over plain `fetch`. The
4 * forge traces its own requests with it, and every app the forge scaffolds
5 * carries this same file as `src/otel.ts` — one implementation, read by both.
6 *
7 * Configured by the standard variables, so any OpenTelemetry collector works:
8 *
9 * OTEL_EXPORTER_OTLP_ENDPOINT e.g. http://otel-collector:4318 (unset = off)
10 * OTEL_SERVICE_NAME what this process is called in a trace
11 * OTEL_RESOURCE_ATTRIBUTES k=v,k=v added to every span and metric
12 * OTEL_SDK_DISABLED=true off, whatever else is set
13 *
14 * With no endpoint every call here is a cheap no-op: spans are still made
15 * (so code reads the same) but nothing is kept or sent. Export never throws
16 * and never blocks a request; a collector that is down costs a dropped
17 * batch, not an error page. The queue is bounded — the oldest spans go first.
18 */
19
20 import { AsyncLocalStorage } from "node:async_hooks";
21 import { randomBytes } from "node:crypto";
22
23 // The real fetch, taken before anyone (an app's own setup) can replace the
24 // global with `tracedFetch`: export and tracedFetch's own call use it, so
25 // neither traces itself.
26 const nativeFetch: typeof fetch = globalThis.fetch.bind(globalThis);
27
28 export type Attr = string | number | boolean;
29 export type Attrs = Record<string, Attr | undefined>;
30
31 export const SpanKind = { internal: 1, server: 2, client: 3 } as const;
32 export type SpanKindName = keyof typeof SpanKind;
33
34 export interface SpanContext {
35 readonly traceId: string;
36 readonly spanId: string;
37 readonly sampled: boolean;
38 }
39
40 interface Config {
41 readonly endpoint: string | undefined;
42 readonly service: string;
43 readonly resource: Attrs;
44 }
45
46 const env = (name: string): string | undefined => {
47 const value = process.env[name]?.trim();
48 return value === "" ? undefined : value;
49 };
50
51 /** `k=v,k=v` as the spec writes it (values may be percent-encoded). */
52 export function parseResourceAttributes(raw: string | undefined): Attrs {
53 const out: Attrs = {};
54 for (const pair of (raw ?? "").split(",")) {
55 const at = pair.indexOf("=");
56 if (at <= 0) continue;
57 const key = pair.slice(0, at).trim();
58 let value = pair.slice(at + 1).trim();
59 try {
60 value = decodeURIComponent(value);
61 } catch {
62 // A stray % is kept as written.
63 }
64 if (key !== "") out[key] = value;
65 }
66 return out;
67 }
68
69 function configFromEnv(): Config {
70 const disabled = env("OTEL_SDK_DISABLED") === "true";
71 const resource = parseResourceAttributes(env("OTEL_RESOURCE_ATTRIBUTES"));
72 const service =
73 env("OTEL_SERVICE_NAME") ??
74 (typeof resource["service.name"] === "string"
75 ? resource["service.name"]
76 : "unknown_service:bun");
77 return {
78 endpoint: disabled
79 ? undefined
80 : env("OTEL_EXPORTER_OTLP_ENDPOINT")?.replace(/\/+$/, ""),
81 service,
82 resource,
83 };
84 }
85
86 let config: Config = configFromEnv();
87
88 /** Re-read the environment, or set explicitly (tests, embedders). */
89 export function configure(next?: Partial<Config>): void {
90 config = { ...configFromEnv(), ...next };
91 }
92
93 export const enabled = (): boolean => config.endpoint !== undefined;
94
95 // ---- trace context ----------------------------------------------------------
96
97 const hex = (bytes: number): string => randomBytes(bytes).toString("hex");
98 const ZERO_TRACE = "0".repeat(32);
99 const ZERO_SPAN = "0".repeat(16);
100
101 /** `00-<trace>-<span>-<flags>`, or undefined for anything malformed. */
102 export function parseTraceparent(
103 header: string | null | undefined,
104 ): SpanContext | undefined {
105 const match = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/.exec(
106 (header ?? "").trim().toLowerCase(),
107 );
108 if (match === null) return undefined;
109 const [, version, traceId, spanId, flags] = match as unknown as [
110 string,
111 string,
112 string,
113 string,
114 string,
115 ];
116 if (version === "ff" || traceId === ZERO_TRACE || spanId === ZERO_SPAN)
117 return undefined;
118 return { traceId, spanId, sampled: (Number.parseInt(flags, 16) & 1) === 1 };
119 }
120
121 export const formatTraceparent = (ctx: SpanContext): string =>
122 `00-${ctx.traceId}-${ctx.spanId}-${ctx.sampled ? "01" : "00"}`;
123
124 // ---- spans ------------------------------------------------------------------
125
126 /** Wall-clock nanoseconds, as a decimal string (OTLP/JSON's uint64 form). */
127 const nowNs = (): bigint =>
128 BigInt(Math.round((performance.timeOrigin + performance.now()) * 1e3)) *
129 1000n;
130
131 export class Span implements SpanContext {
132 readonly traceId: string;
133 readonly spanId: string;
134 readonly parentSpanId: string | undefined;
135 readonly sampled: boolean;
136 readonly startNs = nowNs();
137 endNs: bigint | undefined;
138 readonly attributes: Attrs;
139 status: { code: 0 | 1 | 2; message?: string } = { code: 0 };
140 readonly events: { name: string; timeNs: bigint; attributes: Attrs }[] = [];
141
142 constructor(
143 public name: string,
144 readonly kind: SpanKindName,
145 parent: SpanContext | undefined,
146 attributes: Attrs,
147 ) {
148 this.traceId = parent?.traceId ?? hex(16);
149 this.parentSpanId = parent?.spanId;
150 this.sampled = parent?.sampled ?? true;
151 this.spanId = hex(8);
152 this.attributes = { ...attributes };
153 }
154
155 set(key: string, value: Attr | undefined): this {
156 if (value !== undefined) this.attributes[key] = value;
157 return this;
158 }
159
160 event(name: string, attributes: Attrs = {}): this {
161 this.events.push({ name, timeNs: nowNs(), attributes });
162 return this;
163 }
164
165 /** Mark failed. An exception's message is kept; its stack is not. */
166 fail(error: unknown): this {
167 const message = error instanceof Error ? error.message : String(error);
168 this.status = { code: 2, message };
169 this.event("exception", {
170 "exception.type": error instanceof Error ? error.name : typeof error,
171 "exception.message": message,
172 });
173 return this;
174 }
175
176 end(): void {
177 if (this.endNs !== undefined) return;
178 this.endNs = nowNs();
179 if (this.sampled && enabled()) enqueue(this);
180 }
181
182 get durationSeconds(): number {
183 return Number((this.endNs ?? nowNs()) - this.startNs) / 1e9;
184 }
185 }
186
187 const active = new AsyncLocalStorage<Span>();
188
189 /** The span the current code runs inside, if any. */
190 export const currentSpan = (): Span | undefined => active.getStore();
191
192 export interface SpanOptions {
193 readonly kind?: SpanKindName;
194 /** Parent explicitly; default is the current span. `null` = a new trace. */
195 readonly parent?: SpanContext | null;
196 readonly attributes?: Attrs;
197 }
198
199 export function startSpan(name: string, options: SpanOptions = {}): Span {
200 const parent =
201 options.parent === null ? undefined : (options.parent ?? currentSpan());
202 return new Span(name, options.kind ?? "internal", parent, options.attributes ?? {});
203 }
204
205 /**
206 * Run `fn` inside a new span: it is current for everything `fn` awaits, it
207 * ends when `fn` settles, and a throw marks it failed (and still throws).
208 */
209 export async function span<T>(
210 name: string,
211 fn: (span: Span) => T | Promise<T>,
212 options: SpanOptions = {},
213 ): Promise<T> {
214 const s = startSpan(name, options);
215 try {
216 return await active.run(s, () => fn(s));
217 } catch (error) {
218 s.fail(error);
219 throw error;
220 } finally {
221 s.end();
222 }
223 }
224
225 /** Run `fn` with `s` as the current span, without ending it. */
226 export const withSpan = <T>(s: Span, fn: () => T): T => active.run(s, fn);
227
228 // ---- HTTP -------------------------------------------------------------------
229
230 /**
231 * A server span for one request, parented on the caller's `traceparent`.
232 * Records `http.server.request.duration` too. `route` names the matched
233 * route template (`/notes/:id`) once it is known — set it on the span.
234 */
235 export async function serveSpan<R extends Response | undefined>(
236 request: Request,
237 handle: (span: Span) => R | Promise<R>,
238 attributes: Attrs = {},
239 ): Promise<R> {
240 const url = new URL(request.url);
241 const s = startSpan(request.method, {
242 kind: "server",
243 parent: parseTraceparent(request.headers.get("traceparent")) ?? null,
244 attributes: {
245 "http.request.method": request.method,
246 "url.path": url.pathname,
247 "url.scheme":
248 request.headers.get("x-forwarded-proto") ?? url.protocol.slice(0, -1),
249 "server.address": (
250 request.headers.get("x-forwarded-host") ??
251 request.headers.get("host") ??
252 url.hostname
253 ).replace(/:\d+$/, ""),
254 "user_agent.original": request.headers.get("user-agent") ?? undefined,
255 ...attributes,
256 },
257 });
258 let status = 500;
259 try {
260 const response = await active.run(s, () => handle(s));
261 // undefined = the socket was upgraded (a websocket): it answered 101.
262 status = response?.status ?? 101;
263 return response;
264 } catch (error) {
265 s.fail(error);
266 throw error;
267 } finally {
268 endServerSpan(s, status);
269 }
270 }
271
272 /**
273 * Trace a whole `Bun.serve` options object: its `fetch` runs inside
274 * `serveSpan`, everything else (websocket handlers, port, error) untouched.
275 * The one-line retrofit for an app that routes by hand:
276 *
277 * Bun.serve(traceServe({ port, fetch(request, server) { ... } }))
278 *
279 * A route template isn't known here, so spans are named by method only —
280 * set `http.route` on `currentSpan()` from the handler if you route by name.
281 */
282 export function traceServe<
283 T extends { fetch: (request: Request, ...rest: never[]) => unknown },
284 >(options: T): T {
285 const inner = options.fetch;
286 return {
287 ...options,
288 fetch(this: unknown, request: Request, ...rest: never[]) {
289 return serveSpan(
290 request,
291 async () =>
292 (await inner.call(this, request, ...rest)) as Response | undefined,
293 );
294 },
295 };
296 }
297
298 /**
299 * Finish a server span made by hand (a framework's request hooks): status,
300 * the `METHOD /route` name, and the duration histogram — what `serveSpan`
301 * does in its `finally`.
302 */
303 export function endServerSpan(s: Span, status: number): void {
304 const method = s.attributes["http.request.method"];
305 s.set("http.response.status_code", status);
306 if (status >= 500 && s.status.code === 0) s.status = { code: 2 };
307 const route = s.attributes["http.route"];
308 if (typeof route === "string" && typeof method === "string")
309 s.name = `${method} ${route}`;
310 s.end();
311 requestDuration.record(s.durationSeconds, {
312 "http.request.method": typeof method === "string" ? method : undefined,
313 "http.route": typeof route === "string" ? route : undefined,
314 "http.response.status_code": status,
315 });
316 }
317
318 /**
319 * `fetch` that makes a client span and hands the callee our `traceparent`,
320 * so its spans join this trace. Same signature as `fetch`.
321 */
322 export async function tracedFetch(
323 input: string | URL | Request,
324 init: RequestInit = {},
325 ): Promise<Response> {
326 const request =
327 input instanceof Request
328 ? new Request(input, init)
329 : new Request(String(input), init);
330 const url = new URL(request.url);
331 return span(
332 `${request.method}`,
333 async (s) => {
334 const headers = new Headers(request.headers);
335 headers.set("traceparent", formatTraceparent(s));
336 const response = await nativeFetch(new Request(request, { headers }));
337 s.set("http.response.status_code", response.status);
338 if (response.status >= 500) s.status = { code: 2 };
339 return response;
340 },
341 {
342 kind: "client",
343 attributes: {
344 "http.request.method": request.method,
345 "server.address": url.hostname,
346 "server.port": url.port === "" ? undefined : Number(url.port),
347 "url.full": `${url.origin}${url.pathname}`,
348 },
349 },
350 );
351 }
352
353 // ---- metrics ----------------------------------------------------------------
354
355 const DURATION_BUCKETS = [
356 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,
357 ];
358
359 const attrKey = (attrs: Attrs): string =>
360 JSON.stringify(
361 Object.entries(attrs)
362 .filter(([, v]) => v !== undefined)
363 .sort(([a], [b]) => (a < b ? -1 : 1)),
364 );
365
366 interface Instrument {
367 readonly name: string;
368 readonly unit: string;
369 readonly description: string;
370 toOtlp(startNs: string, nowNs: string): unknown | undefined;
371 }
372
373 const instruments: Instrument[] = [];
374 const processStartNs = nowNs().toString();
375
376 export interface Histogram {
377 record(value: number, attributes?: Attrs): void;
378 }
379
380 /** A cumulative explicit-bucket histogram. Keep attribute values few. */
381 export function histogram(
382 name: string,
383 options: { unit?: string; description?: string; buckets?: number[] } = {},
384 ): Histogram {
385 const bounds = options.buckets ?? DURATION_BUCKETS;
386 const points = new Map<
387 string,
388 { attrs: Attrs; count: number; sum: number; buckets: number[] }
389 >();
390 instruments.push({
391 name,
392 unit: options.unit ?? "s",
393 description: options.description ?? "",
394 toOtlp(startNs, timeNs) {
395 if (points.size === 0) return undefined;
396 return {
397 histogram: {
398 aggregationTemporality: 2,
399 dataPoints: [...points.values()].map((p) => ({
400 attributes: otlpAttrs(p.attrs),
401 startTimeUnixNano: startNs,
402 timeUnixNano: timeNs,
403 count: String(p.count),
404 sum: p.sum,
405 bucketCounts: p.buckets.map(String),
406 explicitBounds: bounds,
407 })),
408 },
409 };
410 },
411 });
412 return {
413 record(value, attributes = {}) {
414 if (!enabled() || !Number.isFinite(value)) return;
415 const key = attrKey(attributes);
416 let p = points.get(key);
417 if (p === undefined) {
418 p = {
419 attrs: attributes,
420 count: 0,
421 sum: 0,
422 buckets: new Array(bounds.length + 1).fill(0),
423 };
424 points.set(key, p);
425 }
426 p.count += 1;
427 p.sum += value;
428 let i = bounds.findIndex((b) => value <= b);
429 if (i === -1) i = bounds.length;
430 p.buckets[i] = (p.buckets[i] ?? 0) + 1;
431 },
432 };
433 }
434
435 export interface Counter {
436 add(value?: number, attributes?: Attrs): void;
437 }
438
439 /** A cumulative monotonic sum. */
440 export function counter(
441 name: string,
442 options: { unit?: string; description?: string } = {},
443 ): Counter {
444 const points = new Map<string, { attrs: Attrs; value: number }>();
445 instruments.push({
446 name,
447 unit: options.unit ?? "1",
448 description: options.description ?? "",
449 toOtlp(startNs, timeNs) {
450 if (points.size === 0) return undefined;
451 return {
452 sum: {
453 aggregationTemporality: 2,
454 isMonotonic: true,
455 dataPoints: [...points.values()].map((p) => ({
456 attributes: otlpAttrs(p.attrs),
457 startTimeUnixNano: startNs,
458 timeUnixNano: timeNs,
459 asDouble: p.value,
460 })),
461 },
462 };
463 },
464 });
465 return {
466 add(value = 1, attributes = {}) {
467 if (!enabled() || !(value >= 0)) return;
468 const key = attrKey(attributes);
469 const p = points.get(key);
470 if (p === undefined) points.set(key, { attrs: attributes, value });
471 else p.value += value;
472 },
473 };
474 }
475
476 /** A value read at export time: queue depth, rows in a table, players online. */
477 export function gauge(
478 name: string,
479 read: () => number | { value: number; attributes?: Attrs }[],
480 options: { unit?: string; description?: string } = {},
481 ): void {
482 instruments.push({
483 name,
484 unit: options.unit ?? "1",
485 description: options.description ?? "",
486 toOtlp(_startNs, timeNs) {
487 let values: { value: number; attributes?: Attrs }[];
488 try {
489 const got = read();
490 values = typeof got === "number" ? [{ value: got }] : got;
491 } catch {
492 return undefined;
493 }
494 if (values.length === 0) return undefined;
495 return {
496 gauge: {
497 dataPoints: values.map((v) => ({
498 attributes: otlpAttrs(v.attributes ?? {}),
499 timeUnixNano: timeNs,
500 asDouble: v.value,
501 })),
502 },
503 };
504 },
505 });
506 }
507
508 /** The semantic-convention HTTP server histogram `serveSpan` records into. */
509 const requestDuration = histogram("http.server.request.duration", {
510 unit: "s",
511 description: "Duration of HTTP server requests.",
512 });
513
514 // ---- export -----------------------------------------------------------------
515
516 function otlpValue(value: Attr): Record<string, unknown> {
517 if (typeof value === "boolean") return { boolValue: value };
518 if (typeof value === "number")
519 return Number.isInteger(value)
520 ? { intValue: String(value) }
521 : { doubleValue: value };
522 return { stringValue: value };
523 }
524
525 export function otlpAttrs(attrs: Attrs): unknown[] {
526 return Object.entries(attrs)
527 .filter((e): e is [string, Attr] => e[1] !== undefined)
528 .map(([key, value]) => ({ key, value: otlpValue(value) }));
529 }
530
531 const resource = () => ({
532 attributes: otlpAttrs({
533 "telemetry.sdk.name": "forge-otel",
534 "telemetry.sdk.language": "js",
535 "process.runtime.name": process.versions.bun === undefined ? "node" : "bun",
536 "process.runtime.version": process.versions.bun ?? process.versions.node,
537 ...config.resource,
538 "service.name": config.service,
539 }),
540 });
541
542 const SCOPE = { name: "forge-otel", version: "1" };
543
544 /** One span in OTLP/JSON (ids are lowercase hex, times decimal ns). */
545 export function spanToOtlp(s: Span): unknown {
546 return {
547 traceId: s.traceId,
548 spanId: s.spanId,
549 ...(s.parentSpanId === undefined ? {} : { parentSpanId: s.parentSpanId }),
550 name: s.name,
551 kind: SpanKind[s.kind],
552 startTimeUnixNano: s.startNs.toString(),
553 endTimeUnixNano: (s.endNs ?? s.startNs).toString(),
554 attributes: otlpAttrs(s.attributes),
555 events: s.events.map((e) => ({
556 name: e.name,
557 timeUnixNano: e.timeNs.toString(),
558 attributes: otlpAttrs(e.attributes),
559 })),
560 status: s.status,
561 };
562 }
563
564 const MAX_QUEUE = 2048;
565 const BATCH = 512;
566 let queue: Span[] = [];
567 let timer: ReturnType<typeof setInterval> | undefined;
568 let lastMetrics = 0;
569
570 function enqueue(s: Span): void {
571 queue.push(s);
572 if (queue.length > MAX_QUEUE) queue = queue.slice(-MAX_QUEUE);
573 if (timer === undefined) {
574 timer = setInterval(() => void flush(), 5_000);
575 // Telemetry must never be the reason a process stays alive.
576 (timer as { unref?: () => void }).unref?.();
577 }
578 }
579
580 async function post(path: string, body: unknown): Promise<boolean> {
581 if (config.endpoint === undefined) return false;
582 try {
583 const response = await nativeFetch(`${config.endpoint}${path}`, {
584 method: "POST",
585 headers: { "content-type": "application/json" },
586 body: JSON.stringify(body),
587 signal: AbortSignal.timeout(5_000),
588 });
589 return response.ok;
590 } catch {
591 return false;
592 }
593 }
594
595 /** Send what is queued now. Resolves when sent (or dropped); never throws. */
596 export async function flush(): Promise<void> {
597 while (queue.length > 0) {
598 const batch = queue.splice(0, BATCH);
599 await post("/v1/traces", {
600 resourceSpans: [
601 { resource: resource(), scopeSpans: [{ scope: SCOPE, spans: batch.map(spanToOtlp) }] },
602 ],
603 });
604 }
605 if (Date.now() - lastMetrics >= 30_000) {
606 lastMetrics = Date.now();
607 await flushMetrics();
608 }
609 }
610
611 export function metricsToOtlp(): unknown | undefined {
612 const timeNs = nowNs().toString();
613 const metrics = instruments.flatMap((i) => {
614 const data = i.toOtlp(processStartNs, timeNs);
615 return data === undefined
616 ? []
617 : [{ name: i.name, unit: i.unit, description: i.description, ...(data as object) }];
618 });
619 if (metrics.length === 0) return undefined;
620 return {
621 resourceMetrics: [
622 { resource: resource(), scopeMetrics: [{ scope: SCOPE, metrics }] },
623 ],
624 };
625 }
626
627 export async function flushMetrics(): Promise<void> {
628 if (!enabled()) return;
629 const body = metricsToOtlp();
630 if (body !== undefined) await post("/v1/metrics", body);
631 }
632
633 /** Start the periodic exporter even before the first span (metrics-only apps). */
634 export function start(): void {
635 if (!enabled() || timer !== undefined) return;
636 timer = setInterval(() => void flush(), 5_000);
637 (timer as { unref?: () => void }).unref?.();
638 }
src/server.ts +3 −2
@@ -29,6 +29,7 @@ import {
2929 import { recentResults } from "./db.ts";
3030 import { gridRows, TILE } from "./map.ts";
3131 import { gamePage } from "./pages/game.ts";
32 import { traceServe } from "./otel.ts";
3233
3334 const env = process.env.FORGE_APP_ENV ?? "local";
3435 const pull = process.env.FORGE_APP_PULL;
@@ -164,7 +165,7 @@ type SocketData = { session: string; playerId: string; address: string; token: s
164165
165166 // ---------------------------------------------------------------- routing
166167
167 export const server = Bun.serve<SocketData, Record<string, never>>({
168 export const server = Bun.serve<SocketData, Record<string, never>>(traceServe({
168169 port: Number(process.env.PORT ?? 8080),
169170 hostname: process.env.HOST ?? "127.0.0.1",
170171 async fetch(request, self) {
@@ -330,7 +331,7 @@ export const server = Bun.serve<SocketData, Record<string, never>>({
330331 console.error(error);
331332 return html("<h1>Something went wrong</h1>", 500);
332333 },
333 });
334 }));
334335
335336 startArena();
336337
↑↓ navigate ↵ open esc close

Keyboard shortcuts