From 279248e0f22b5c1a4584963421677949c8f14d8d Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 4 Aug 2026 04:57:42 -0700 Subject: [PATCH] Add OTEL export sink with OTLP HTTP JSON flush --- docs/PERFTRACE.md | 11 +- src/index.ts | 6 + src/perf/index.ts | 43 ++++++ src/perf/otel-sink.test.ts | 287 +++++++++++++++++++++++++++++++++++++ src/perf/otel-sink.ts | 287 +++++++++++++++++++++++++++++++++++++ src/tui/runner.tsx | 1 + 6 files changed, 630 insertions(+), 5 deletions(-) create mode 100644 src/perf/otel-sink.test.ts create mode 100644 src/perf/otel-sink.ts diff --git a/docs/PERFTRACE.md b/docs/PERFTRACE.md index a2ff1188f..bb10d18e3 100644 --- a/docs/PERFTRACE.md +++ b/docs/PERFTRACE.md @@ -21,10 +21,10 @@ Local measurement does not require any settings or env vars. Export is **off** until an OTLP endpoint is configured. When enabled, traces go to the operator-owned backend you point at — not Corbits product analytics. -The settings/env surface is implemented now (`src/perf/otel-config.ts`). The -actual OTLP transport lands in a follow-up (CL-5173). Invalid config fails -closed with a stable error code `OTEL_CONFIG_INVALID` and does not half-enable -export. +The settings/env surface lives in `src/perf/otel-config.ts`. OTLP/HTTP JSON +export (`src/perf/otel-sink.ts`, CL-5173) flushes the PerfSpan tree once at +process exit when export is enabled. Invalid config fails closed with a stable +error code `OTEL_CONFIG_INVALID` and does not half-enable export. ### Configuration @@ -97,7 +97,8 @@ No endpoint and no half-config → export stays disabled (not an error). ### Targeting common collectors Examples assume the OTLP HTTP base URL your collector documents. Paths such as -`/v1/traces` are appended by the exporter (CL-5173), not by this settings layer. +`/v1/traces` are appended by the exporter unless the endpoint already ends with +`/v1/traces`. #### Arize Phoenix diff --git a/src/index.ts b/src/index.ts index 5709a9891..ce67e7cbb 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,6 +2,7 @@ import { getLogger } from "@intx/log"; import { LOG_NAMESPACE_ROOT } from "./branding.js"; import { loadConfig } from "./config/index.js"; import { ensureTelemetrySettings, globalSettingsPath } from "./config/settings.js"; +import { flushPerfToOtel } from "./perf/index.js"; import { createTelemetry, telemetryDisabledByEnv } from "./telemetry/index.js"; import { getTelemetry, setTelemetry } from "./telemetry/singleton.js"; import { runExec } from "./exec/runner.js"; @@ -67,6 +68,11 @@ export async function mainWithRunners( exitCode = await runners.runTUI(config); } + // Opt-in OTEL export of the PerfSpan tree (session/process boundary). + // No-op when OTEL is disabled — zero network on the export path. + const otelSettings = config.configured ? config.settings : null; + await flushPerfToOtel(otelSettings); + // Bound against process.exit dropping in-flight captures for short // sessions; flush itself is deadline-capped so exit stays snappy. await getTelemetry().flush(); diff --git a/src/perf/index.ts b/src/perf/index.ts index 011bdfea2..694f44e52 100644 --- a/src/perf/index.ts +++ b/src/perf/index.ts @@ -42,6 +42,49 @@ export { type OtelSettings, } from "./otel-config.js"; +export { + buildOtlpPayload, + flushToOtel, + monoToUnixNano, + newOtelTraceId, + otelSpanId, + otlpTracesUrl, + tagsToOtlpAttributes, + type FlushPerfToOtelOptions, + type FlushToOtelOptions, + type OtlpExportPayload, + type OtlpKeyValue, + type OtlpSpan, +} from "./otel-sink.js"; + +import type { Settings } from "../config/settings.js"; +import { + flushPerfToOtel as flushPerfToOtelImpl, + type FlushPerfToOtelOptions, +} from "./otel-sink.js"; + +/** + * Snapshot the process-wide ring and POST to the operator OTLP collector when + * export is enabled. Zero network when disabled. Never throws. + * Cadence: call on session/process exit (wired from main). + */ +export async function flushPerfToOtel( + settings?: Settings | null, + env: NodeJS.ProcessEnv = process.env, + options: FlushPerfToOtelOptions = {}, +): Promise { + const { spans, getSpans, ...rest } = options; + if (spans !== undefined) { + await flushPerfToOtelImpl(settings, env, { ...rest, spans }); + return; + } + if (getSpans !== undefined) { + await flushPerfToOtelImpl(settings, env, { ...rest, getSpans }); + return; + } + await flushPerfToOtelImpl(settings, env, { ...rest, getSpans: snapshot }); +} + /** Core + adapter phase names. Adapters extend; they do not invent new sinks. */ export const SPAN_NAMES = [ "session", diff --git a/src/perf/otel-sink.test.ts b/src/perf/otel-sink.test.ts new file mode 100644 index 000000000..cb315cd9b --- /dev/null +++ b/src/perf/otel-sink.test.ts @@ -0,0 +1,287 @@ +import { afterEach, describe, expect, test } from "bun:test"; + +import type { Settings } from "../config/settings.js"; +import { clear, end, snapshot, start, type PerfSpan } from "./index.js"; +import { + buildOtlpPayload, + flushPerfToOtel, + flushToOtel, + monoToUnixNano, + otelSpanId, + otlpTracesUrl, + tagsToOtlpAttributes, +} from "./otel-sink.js"; +import type { EnabledOtelExportConfig } from "./otel-config.js"; + +afterEach(() => { + clear(); +}); + +const enabledConfig = ( + overrides: Partial = {}, +): EnabledOtelExportConfig => ({ + enabled: true, + endpoint: "http://localhost:4318", + headers: {}, + serviceName: "corbits-code", + resourceAttributes: { "service.name": "corbits-code" }, + ...overrides, +}); + +const baseSettings = (otel?: Settings["otel"]): Settings => ({ + providers: {}, + ...(otel !== undefined ? { otel } : {}), +}); + +const mockFetch = (impl: (input: RequestInfo | URL, init?: RequestInit) => Promise) => + impl as unknown as typeof fetch; + +describe("otlpTracesUrl", () => { + test("appends /v1/traces to base endpoint", () => { + expect(otlpTracesUrl("http://localhost:4318")).toBe("http://localhost:4318/v1/traces"); + }); + + test("does not double-append when path already ends with /v1/traces", () => { + expect(otlpTracesUrl("https://app.phoenix.arize.com/v1/traces")).toBe( + "https://app.phoenix.arize.com/v1/traces", + ); + }); + + test("strips trailing slash before appending", () => { + expect(otlpTracesUrl("http://localhost:4318/")).toBe("http://localhost:4318/v1/traces"); + }); +}); + +describe("otelSpanId", () => { + test("is 16 hex chars and stable", () => { + const a = otelSpanId("1"); + const b = otelSpanId("1"); + expect(a).toMatch(/^[0-9a-f]{16}$/); + expect(a).toBe(b); + expect(otelSpanId("2")).not.toBe(a); + }); +}); + +describe("tagsToOtlpAttributes", () => { + test("maps string and integer tags", () => { + const attrs = tagsToOtlpAttributes({ + provider_id: "openai", + count: 3, + transport: "ws", + }); + expect(attrs).toEqual([ + { key: "count", value: { intValue: "3" } }, + { key: "provider_id", value: { stringValue: "openai" } }, + { key: "transport", value: { stringValue: "ws" } }, + ]); + }); + + test("re-sanitizes forbidden keys at export", () => { + const attrs = tagsToOtlpAttributes({ + provider_id: "xai", + prompt: "secret", + path: "/tmp/x", + } as never); + expect(attrs).toEqual([{ key: "provider_id", value: { stringValue: "xai" } }]); + }); +}); + +describe("buildOtlpPayload", () => { + test("maps parent links, names, and times", () => { + const turnId = start("turn"); + const infId = start("inference", { parentId: turnId, tags: { model_id: "m1" } }); + end(infId); + end(turnId); + const spans: PerfSpan[] = [ + { + id: turnId, + name: "turn", + startNs: 1000n, + endNs: 5000n, + }, + { + id: infId, + name: "inference", + parentId: turnId, + startNs: 2000n, + endNs: 4000n, + tags: { model_id: "m1" }, + }, + ]; + + const anchor = { monoNs: 0n, unixNs: 1_000_000_000_000n }; + const payload = buildOtlpPayload(spans, enabledConfig(), { + wallAnchor: anchor, + traceId: "a".repeat(32), + }); + + const otlpSpans = payload.resourceSpans[0]!.scopeSpans[0]!.spans; + expect(otlpSpans).toHaveLength(2); + + const turn = otlpSpans.find((s) => s.name === "turn")!; + const inf = otlpSpans.find((s) => s.name === "inference")!; + expect(turn.traceId).toBe("a".repeat(32)); + expect(turn.spanId).toBe(otelSpanId(turnId)); + expect(turn.parentSpanId).toBeUndefined(); + expect(turn.startTimeUnixNano).toBe(monoToUnixNano(1000n, anchor).toString()); + expect(turn.endTimeUnixNano).toBe(monoToUnixNano(5000n, anchor).toString()); + + expect(inf.parentSpanId).toBe(otelSpanId(turnId)); + expect(inf.attributes).toEqual([{ key: "model_id", value: { stringValue: "m1" } }]); + + const resource = payload.resourceSpans[0]!.resource.attributes; + expect( + resource.some( + (a) => a.key === "service.name" && "stringValue" in a.value && a.value.stringValue === "corbits-code", + ), + ).toBe(true); + }); + + test("open spans use nowMonoNs as end", () => { + const spans: PerfSpan[] = [{ id: "open1", name: "session", startNs: 10n }]; + const anchor = { monoNs: 0n, unixNs: 0n }; + const payload = buildOtlpPayload(spans, enabledConfig(), { + wallAnchor: anchor, + nowMonoNs: () => 99n, + traceId: "b".repeat(32), + }); + const span = payload.resourceSpans[0]!.scopeSpans[0]!.spans[0]!; + expect(span.startTimeUnixNano).toBe("10"); + expect(span.endTimeUnixNano).toBe("99"); + }); +}); + +describe("flushToOtel", () => { + test("POSTs OTLP JSON with headers to /v1/traces", async () => { + const calls: Array<{ url: string; init: RequestInit }> = []; + const fetchFn = mockFetch(async (input, init) => { + calls.push({ url: String(input), init: init ?? {} }); + return new Response(null, { status: 200 }); + }); + + const spans: PerfSpan[] = [{ id: "1", name: "turn", startNs: 1n, endNs: 2n }]; + await flushToOtel( + spans, + enabledConfig({ + endpoint: "https://collector.example", + headers: { Authorization: "Bearer secret" }, + }), + { fetchFn, traceId: "c".repeat(32), wallAnchor: { monoNs: 0n, unixNs: 0n } }, + ); + + expect(calls).toHaveLength(1); + expect(calls[0]!.url).toBe("https://collector.example/v1/traces"); + const headers = calls[0]!.init.headers as Record; + expect(headers["content-type"]).toBe("application/json"); + expect(headers.Authorization).toBe("Bearer secret"); + + const body = JSON.parse(String(calls[0]!.init.body)) as { + resourceSpans: unknown[]; + }; + expect(body.resourceSpans).toHaveLength(1); + }); + + test("empty spans do not call fetch", async () => { + let called = 0; + const fetchFn = mockFetch(async () => { + called += 1; + return new Response(null, { status: 200 }); + }); + await flushToOtel([], enabledConfig(), { fetchFn }); + expect(called).toBe(0); + }); + + test("network errors are swallowed", async () => { + const fetchFn = mockFetch(async () => { + throw new Error("ECONNREFUSED"); + }); + await expect( + flushToOtel([{ id: "1", name: "turn", startNs: 1n, endNs: 2n }], enabledConfig(), { + fetchFn, + }), + ).resolves.toBeUndefined(); + }); + + test("non-2xx responses are swallowed", async () => { + const fetchFn = mockFetch(async () => new Response("nope", { status: 503 })); + await expect( + flushToOtel([{ id: "1", name: "turn", startNs: 1n, endNs: 2n }], enabledConfig(), { + fetchFn, + }), + ).resolves.toBeUndefined(); + }); +}); + +describe("flushPerfToOtel", () => { + test("disabled config performs zero network", async () => { + let called = 0; + const fetchFn = mockFetch(async () => { + called += 1; + return new Response(null, { status: 200 }); + }); + + const id = start("turn"); + end(id); + await flushPerfToOtel(baseSettings(), {}, { fetchFn }); + expect(called).toBe(0); + }); + + test("enabled config snapshots and POSTs", async () => { + const calls: string[] = []; + const fetchFn = mockFetch(async (input) => { + calls.push(String(input)); + return new Response(null, { status: 200 }); + }); + + const id = start("session"); + end(id); + + await flushPerfToOtel(baseSettings({ endpoint: "http://127.0.0.1:4318" }), {}, { fetchFn, getSpans: snapshot }); + expect(calls).toEqual(["http://127.0.0.1:4318/v1/traces"]); + }); + + test("invalid config does not throw and does not fetch", async () => { + let called = 0; + const fetchFn = mockFetch(async () => { + called += 1; + return new Response(null, { status: 200 }); + }); + + await expect( + flushPerfToOtel(baseSettings({ enabled: true }), {}, { fetchFn }), + ).resolves.toBeUndefined(); + expect(called).toBe(0); + }); + + test("explicit spans override ring snapshot", async () => { + const bodies: string[] = []; + const fetchFn = mockFetch(async (_input, init) => { + bodies.push(String(init?.body ?? "")); + return new Response(null, { status: 200 }); + }); + + // Ring has a session span — should be ignored when spans is provided. + start("session"); + const only: PerfSpan[] = [ + { id: "only", name: "tool", startNs: 1n, endNs: 2n, tags: { tool_id: "t1" } }, + ]; + await flushPerfToOtel( + baseSettings({ endpoint: "http://localhost:4318" }), + {}, + { + fetchFn, + spans: only, + traceId: "d".repeat(32), + wallAnchor: { monoNs: 0n, unixNs: 0n }, + }, + ); + expect(bodies).toHaveLength(1); + const parsed = JSON.parse(bodies[0]!) as { + resourceSpans: Array<{ + scopeSpans: Array<{ spans: Array<{ name: string }> }>; + }>; + }; + const names = parsed.resourceSpans[0]!.scopeSpans[0]!.spans.map((s) => s.name); + expect(names).toEqual(["tool"]); + }); +}); diff --git a/src/perf/otel-sink.ts b/src/perf/otel-sink.ts new file mode 100644 index 000000000..4c273b746 --- /dev/null +++ b/src/perf/otel-sink.ts @@ -0,0 +1,287 @@ +/** + * Opt-in OTLP/HTTP JSON export for PerfTrace (CL-5173). + * + * Maps the in-process PerfSpan tree to OTEL spans and POSTs to the operator's + * collector. Disabled config paths do no network. Network/config failures are + * logged and swallowed — never thrown to callers. + * + * No @opentelemetry/sdk dependency: hand-rolled OTLP HTTP JSON only. + */ + +import { createHash, randomBytes } from "node:crypto"; +import { getLogger } from "@intx/log"; + +import { LOG_NAMESPACE_ROOT } from "../branding.js"; +import type { Settings } from "../config/settings.js"; +import type { PerfSpan } from "./index.js"; +import { + resolveOtelExportConfig, + type EnabledOtelExportConfig, + type OtelExportConfig, +} from "./otel-config.js"; +import { sanitizeTags } from "./sanitize.js"; + +const log = getLogger([LOG_NAMESPACE_ROOT, "perf", "otel"]); + +/** Upper bound so a slow collector cannot hold process exit indefinitely. */ +const EXPORT_TIMEOUT_MS = 3000; + +/** OTLP span kind: INTERNAL (1). */ +const SPAN_KIND_INTERNAL = 1; + +export type FlushToOtelOptions = { + /** Injectable fetch for tests. Defaults to global fetch. */ + fetchFn?: typeof fetch; + /** + * Wall-clock anchor for converting monotonic hrtime to unix nano. + * Defaults to Date.now() * 1e6 ns aligned with process.hrtime.bigint(). + */ + wallAnchor?: { monoNs: bigint; unixNs: bigint }; + /** Override open-span end time (defaults to now). */ + nowMonoNs?: () => bigint; + /** Fixed trace id for deterministic tests (32 hex chars). */ + traceId?: string; +}; + +export type FlushPerfToOtelOptions = FlushToOtelOptions & { + /** Spans to export; when omitted, caller supplies via getSpans. */ + spans?: readonly PerfSpan[]; + /** Lazy snapshot provider — defaults supplied by perf/index flushPerfToOtel. */ + getSpans?: () => readonly PerfSpan[]; +}; + +/** OTLP JSON attribute value (subset we emit). */ +export type OtlpAnyValue = + | { stringValue: string } + | { intValue: string } + | { doubleValue: number } + | { boolValue: boolean }; + +export type OtlpKeyValue = { key: string; value: OtlpAnyValue }; + +export type OtlpSpan = { + traceId: string; + spanId: string; + parentSpanId?: string; + name: string; + kind: number; + startTimeUnixNano: string; + endTimeUnixNano: string; + attributes: OtlpKeyValue[]; + status: { code: number }; +}; + +export type OtlpExportPayload = { + resourceSpans: Array<{ + resource: { attributes: OtlpKeyValue[] }; + scopeSpans: Array<{ + scope: { name: string; version: string }; + spans: OtlpSpan[]; + }>; + }>; +}; + +function defaultWallAnchor(): { monoNs: bigint; unixNs: bigint } { + return { + monoNs: process.hrtime.bigint(), + unixNs: BigInt(Date.now()) * 1_000_000n, + }; +} + +/** Convert monotonic ns to unix epoch ns using a shared wall/mono anchor. */ +export function monoToUnixNano( + monoNs: bigint, + anchor: { monoNs: bigint; unixNs: bigint }, +): bigint { + return anchor.unixNs + (monoNs - anchor.monoNs); +} + +/** + * Stable 16-hex-char OTEL span id from an opaque PerfSpan id. + * Must be non-all-zero; hash guarantees a full 64-bit space. + */ +export function otelSpanId(perfId: string): string { + return createHash("sha256").update(`span:${perfId}`).digest("hex").slice(0, 16); +} + +export function newOtelTraceId(): string { + return randomBytes(16).toString("hex"); +} + +/** + * Resolve the OTLP traces URL. Appends `/v1/traces` unless the endpoint already + * ends with that path (operators sometimes paste full collector URLs). + */ +export function otlpTracesUrl(endpoint: string): string { + const base = endpoint.replace(/\/+$/, ""); + if (base.endsWith("/v1/traces")) return base; + return `${base}/v1/traces`; +} + +export function tagsToOtlpAttributes( + tags: PerfSpan["tags"] | undefined, +): OtlpKeyValue[] { + // Defense in depth: re-sanitize even though start/end/mark already did. + const safe = sanitizeTags(tags as Record | undefined); + if (safe === undefined) return []; + + const out: OtlpKeyValue[] = []; + for (const [key, value] of Object.entries(safe)) { + if (value === undefined) continue; + if (typeof value === "string") { + out.push({ key, value: { stringValue: value } }); + continue; + } + if (typeof value === "number") { + if (!Number.isFinite(value)) continue; + if (Number.isInteger(value)) { + out.push({ key, value: { intValue: String(value) } }); + } else { + out.push({ key, value: { doubleValue: value } }); + } + } + } + // Stable order for tests and collector diffs. + out.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0)); + return out; +} + +function resourceAttributes(config: EnabledOtelExportConfig): OtlpKeyValue[] { + const attrs: OtlpKeyValue[] = []; + for (const [key, value] of Object.entries(config.resourceAttributes)) { + attrs.push({ key, value: { stringValue: value } }); + } + // service.name is guaranteed by resolveOtelExportConfig, but keep explicit. + if (!attrs.some((a) => a.key === "service.name")) { + attrs.push({ key: "service.name", value: { stringValue: config.serviceName } }); + } + attrs.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0)); + return attrs; +} + +/** + * Map PerfSpan[] → OTLP HTTP JSON payload (one resource, one scope, one trace). + * Open spans (no endNs) use `nowMonoNs` as end so partial trees still export. + */ +export function buildOtlpPayload( + spans: readonly PerfSpan[], + config: EnabledOtelExportConfig, + options: FlushToOtelOptions = {}, +): OtlpExportPayload { + const anchor = options.wallAnchor ?? defaultWallAnchor(); + const nowMono = options.nowMonoNs ?? (() => process.hrtime.bigint()); + const traceId = options.traceId ?? newOtelTraceId(); + const endFallback = nowMono(); + + const otlpSpans: OtlpSpan[] = spans.map((span) => { + const endMono = span.endNs ?? endFallback; + const startUnix = monoToUnixNano(span.startNs, anchor); + const endUnix = monoToUnixNano(endMono, anchor); + // Guard inverted times if clock anchor is weird in tests. + const startTimeUnixNano = startUnix <= endUnix ? startUnix : endUnix; + const endTimeUnixNano = endUnix >= startTimeUnixNano ? endUnix : startTimeUnixNano; + + const otlp: OtlpSpan = { + traceId, + spanId: otelSpanId(span.id), + name: span.name, + kind: SPAN_KIND_INTERNAL, + startTimeUnixNano: startTimeUnixNano.toString(), + endTimeUnixNano: endTimeUnixNano.toString(), + attributes: tagsToOtlpAttributes(span.tags), + status: { code: 0 }, // UNSET + }; + if (span.parentId !== undefined && span.parentId.length > 0) { + otlp.parentSpanId = otelSpanId(span.parentId); + } + return otlp; + }); + + return { + resourceSpans: [ + { + resource: { attributes: resourceAttributes(config) }, + scopeSpans: [ + { + scope: { name: "corbits-code.perf", version: "1" }, + spans: otlpSpans, + }, + ], + }, + ], + }; +} + +/** + * POST spans to the operator's OTLP HTTP/JSON collector. + * Never throws. No-op when spans is empty. + */ +export async function flushToOtel( + spans: readonly PerfSpan[], + config: EnabledOtelExportConfig, + options: FlushToOtelOptions = {}, +): Promise { + if (spans.length === 0) return; + + const fetchFn = options.fetchFn ?? fetch; + const url = otlpTracesUrl(config.endpoint); + const body = JSON.stringify(buildOtlpPayload(spans, config, options)); + + const headers: Record = { + "content-type": "application/json", + ...config.headers, + }; + + try { + const response = await fetchFn(url, { + method: "POST", + headers, + body, + signal: AbortSignal.timeout(EXPORT_TIMEOUT_MS), + }); + if (!response.ok) { + log.warn("OTEL export failed: HTTP {status} from {url}", { + status: response.status, + url, + }); + } + } catch (err) { + log.warn("OTEL export failed: {error}", { + error: err instanceof Error ? err.message : String(err), + }); + } +} + +/** + * Resolve settings/env and flush when export is enabled. + * Zero network when disabled or config is invalid (invalid is logged once). + * Never throws. + * + * Provide spans via `options.spans` or `options.getSpans`. When neither is set, + * this is a no-op (perf/index `flushPerfToOtel` wires snapshot()). + */ +export async function flushPerfToOtel( + settings?: Settings | null, + env: NodeJS.ProcessEnv = process.env, + options: FlushPerfToOtelOptions = {}, +): Promise { + let config: OtelExportConfig; + try { + const resolved = resolveOtelExportConfig(settings, env); + if (!resolved.ok) { + log.warn("OTEL export skipped: {message}", { message: resolved.message }); + return; + } + config = resolved.config; + } catch (err) { + log.warn("OTEL export config resolution failed: {error}", { + error: err instanceof Error ? err.message : String(err), + }); + return; + } + + if (!config.enabled) return; + + const spans = options.spans ?? options.getSpans?.() ?? []; + await flushToOtel(spans, config, options); +} diff --git a/src/tui/runner.tsx b/src/tui/runner.tsx index fafc31035..5b9912b82 100644 --- a/src/tui/runner.tsx +++ b/src/tui/runner.tsx @@ -1564,6 +1564,7 @@ export async function runTUI(initialConfig: Config): Promise { }); // Bound against process.exit dropping the session_end capture for short // sessions; flush itself is deadline-capped so exit stays snappy. + // PerfTrace OTEL export runs once at process exit in main (flushPerfToOtel). await getTelemetry().flush(); await sessionOps.awaitTail();