apps/glyph Public The code is public; the app answers only to 1 person who can change it and the crew.

Files draw
728 lines · 22.6 KB · ~6.6k tok Raw History
/**
 * OpenTelemetry in one file, with nothing to install: W3C trace context in
 * and out, spans, metrics and logs 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<string, Attr | undefined>;

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<Config>): 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-<trace>-<span>-<flags>`, 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: bigint;
  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,
    startTime?: number,
  ) {
    // A span recorded after the fact (a run that finished) says when it began.
    this.startNs = startTime === undefined ? nowNs() : BigInt(Math.round(startTime * 1e3)) * 1000n;
    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<Span>();

/** 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;
  /** When it began, in epoch milliseconds, if before now. */
  readonly startTime?: number;
}

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 ?? {}, options.startTime);
}

/**
 * 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<T>(
  name: string,
  fn: (span: Span) => T | Promise<T>,
  options: SpanOptions = {},
): Promise<T> {
  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 = <T>(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<R extends Response | undefined>(
  request: Request,
  handle: (span: Span) => R | Promise<R>,
  attributes: Attrs = {},
): Promise<R> {
  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<Response> {
  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<string, { attrs: Attrs; value: number }>();
  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.",
});

// ---- logs -------------------------------------------------------------------

export type Severity = "debug" | "info" | "warn" | "error";
const SEVERITY_NUMBER: Record<Severity, number> = { debug: 5, info: 9, warn: 13, error: 17 };

export interface LogOptions {
  readonly severity?: Severity;
  readonly attributes?: Attrs;
  /**
   * Who is speaking, when it is not this process: the forge files each
   * app's output under the app's own `service.name`. Added to (and wins
   * over) this process's resource.
   */
  readonly resource?: Attrs;
}

interface LogEntry {
  readonly timeNs: bigint;
  readonly body: string;
  readonly severity: Severity;
  readonly attributes: Attrs;
  readonly resource: Attrs | undefined;
  readonly traceId: string | undefined;
  readonly spanId: string | undefined;
}

const MAX_LOG_QUEUE = 4096;
let logQueue: LogEntry[] = [];

/**
 * One log record, tied to the current span when there is one. A no-op
 * without an endpoint; never throws, never blocks.
 */
export function emitLog(body: string, options: LogOptions = {}): void {
  if (!enabled()) return;
  const s = currentSpan();
  logQueue.push({
    timeNs: nowNs(),
    body,
    severity: options.severity ?? "info",
    attributes: options.attributes ?? {},
    resource: options.resource,
    traceId: s?.traceId,
    spanId: s?.spanId,
  });
  // The oldest go first: a flood keeps its newest lines.
  if (logQueue.length > MAX_LOG_QUEUE) logQueue = logQueue.slice(-MAX_LOG_QUEUE);
  startTimer();
}

/** Log records in OTLP/JSON, grouped by who spoke. */
export function logsToOtlp(entries: readonly LogEntry[]): unknown {
  const byResource = new Map<string, { resource: Attrs | undefined; records: unknown[] }>();
  for (const e of entries) {
    const key = e.resource === undefined ? "" : attrKey(e.resource);
    let group = byResource.get(key);
    if (group === undefined) byResource.set(key, (group = { resource: e.resource, records: [] }));
    group.records.push({
      timeUnixNano: e.timeNs.toString(),
      observedTimeUnixNano: e.timeNs.toString(),
      severityNumber: SEVERITY_NUMBER[e.severity],
      severityText: e.severity.toUpperCase(),
      body: { stringValue: e.body },
      attributes: otlpAttrs(e.attributes),
      ...(e.traceId === undefined ? {} : { traceId: e.traceId }),
      ...(e.spanId === undefined ? {} : { spanId: e.spanId }),
    });
  }
  return {
    resourceLogs: [...byResource.values()].map((g) => ({
      resource: resource(g.resource),
      scopeLogs: [{ scope: SCOPE, logRecords: g.records }],
    })),
  };
}

// ---- export -----------------------------------------------------------------

function otlpValue(value: Attr): Record<string, unknown> {
  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 = (over?: Attrs) => ({
  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,
    ...over,
  }),
});

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<typeof setInterval> | undefined;
let lastMetrics = 0;

function enqueue(s: Span): void {
  queue.push(s);
  if (queue.length > MAX_QUEUE) queue = queue.slice(-MAX_QUEUE);
  startTimer();
}

function startTimer(): void {
  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<boolean> {
  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<void> {
  while (queue.length > 0) {
    const batch = queue.splice(0, BATCH);
    await post("/v1/traces", {
      resourceSpans: [
        { resource: resource(), scopeSpans: [{ scope: SCOPE, spans: batch.map(spanToOtlp) }] },
      ],
    });
  }
  while (logQueue.length > 0) {
    const batch = logQueue.splice(0, BATCH);
    await post("/v1/logs", logsToOtlp(batch));
  }
  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<void> {
  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?.();
}