diff --git a/src/core/dev/otel/store.test.ts b/src/core/dev/otel/store.test.ts new file mode 100644 index 000000000..53d04cc2a --- /dev/null +++ b/src/core/dev/otel/store.test.ts @@ -0,0 +1,164 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { TraceStore } from "./store"; +import type { OtlpPayload } from "./types"; + +const TRACE_A = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; +const TRACE_B = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + +function payload( + traceId: string, + options: { serviceName?: string; startNano?: string; name?: string } = {}, +): OtlpPayload { + return { + resourceSpans: [ + { + resource: { + attributes: [ + { key: "service.name", value: { stringValue: options.serviceName ?? "agent-1" } }, + ], + }, + scopeSpans: [ + { + scope: { name: "test" }, + spans: [ + { + traceId, + spanId: "0123456789abcdef", + name: options.name ?? "invoke_agent strands", + kind: 1, + startTimeUnixNano: options.startNano ?? `${BigInt(Date.now()) * 1_000_000n}`, + endTimeUnixNano: options.startNano ?? `${BigInt(Date.now()) * 1_000_000n}`, + }, + ], + }, + ], + }, + ], + }; +} + +let directory: string; +let store: TraceStore; + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), "trace-store-")); + store = new TraceStore(directory); +}); + +afterEach(async () => { + await rm(directory, { recursive: true, force: true }); +}); + +describe("TraceStore", () => { + test("append then list returns the trace with metadata", async () => { + await store.append(payload(TRACE_A)); + const traces = await store.list(); + expect(traces).toHaveLength(1); + expect(traces[0]!.traceId).toBe(TRACE_A); + expect(traces[0]!.spanCount).toBe("1"); + expect(traces[0]!.resourceSpans).toBeDefined(); + }); + + test("appends to the same trace accumulate spans", async () => { + await store.append(payload(TRACE_A)); + await store.append(payload(TRACE_A, { name: "tool_use" })); + const traces = await store.list(); + expect(traces).toHaveLength(1); + expect(traces[0]!.spanCount).toBe("2"); + }); + + test("payloads without a trace id are dropped", async () => { + await store.append({ resourceSpans: [] }); + expect(await store.list()).toEqual([]); + }); + + test("a batch carrying several traces lands in each trace's own file", async () => { + const batch = payload(TRACE_A); + batch.resourceSpans![0]!.scopeSpans![0]!.spans!.push({ + ...batch.resourceSpans![0]!.scopeSpans![0]!.spans![0]!, + traceId: TRACE_B, + name: "tool_use", + }); + await store.append(batch); + + const traces = await store.list(); + expect(traces.map((trace) => trace.traceId).sort()).toEqual([TRACE_A, TRACE_B]); + expect(traces.every((trace) => trace.spanCount === "1")).toBe(true); + expect(await store.get(TRACE_B)).toBeDefined(); + }); + + test("list filters by service name, matching every participant of a distributed trace", async () => { + await store.append(payload(TRACE_A, { serviceName: "agent-1" })); + // agent-2 contributes spans to the SAME trace (distributed) and owns its own trace. + await store.append(payload(TRACE_A, { serviceName: "agent-2", name: "tool_use" })); + await store.append(payload(TRACE_B, { serviceName: "agent-2" })); + + expect((await store.list({ serviceName: "agent-2" })).map((t) => t.traceId).sort()).toEqual([ + TRACE_A, + TRACE_B, + ]); + expect((await store.list({ serviceName: "agent-1" })).map((t) => t.traceId)).toEqual([TRACE_A]); + expect(await store.list({ serviceName: "agent-3" })).toEqual([]); + }); + + test("list filters by time window, sorts newest first, and caps to limit", async () => { + const oldNano = `${BigInt(Date.now() - 24 * 60 * 60 * 1000) * 1_000_000n}`; + await store.append(payload(TRACE_A, { startNano: oldNano })); + await store.append(payload(TRACE_B)); + + expect((await store.list()).map((trace) => trace.traceId)).toEqual([TRACE_B]); + + const all = await store.list({ startTime: 0 }); + expect(all.map((trace) => trace.traceId)).toEqual([TRACE_B, TRACE_A]); + + // limit keeps the newest N after sorting. + expect((await store.list({ startTime: 0, limit: 1 })).map((trace) => trace.traceId)).toEqual([ + TRACE_B, + ]); + }); + + test("get merges spans across appends and is undefined for unknown ids", async () => { + await store.append(payload(TRACE_A)); + await store.append(payload(TRACE_A, { name: "tool_use" })); + + const detail = await store.get(TRACE_A); + const spans = (detail!.resourceSpans as { scopeSpans: { spans: { name: string }[] }[] }[]) + .flatMap((resourceSpan) => resourceSpan.scopeSpans) + .flatMap((scopeSpan) => scopeSpan.spans); + expect(spans.map((span) => span.name).sort()).toEqual(["invoke_agent strands", "tool_use"]); + + expect(await store.get(TRACE_B)).toBeUndefined(); + }); + + test("skips malformed lines and files without failing", async () => { + await store.append(payload(TRACE_A)); + await writeFile(join(directory, `${TRACE_A}.otlp.jsonl`), "{not json}\n", { + flag: "a", + }); + await writeFile(join(directory, "garbage.otlp.jsonl"), "also not json\n"); + + const traces = await store.list(); + expect(traces).toHaveLength(1); + expect(traces[0]!.spanCount).toBe("1"); + }); + + test("list on a directory that does not exist returns empty", async () => { + const empty = new TraceStore(join(directory, "missing")); + expect(await empty.list()).toEqual([]); + expect(await empty.get(TRACE_A)).toBeUndefined(); + }); + + test("non-ENOENT fs errors bubble up rather than reading as empty", async () => { + // readdir on a path that is a file, not a directory -> ENOTDIR must throw. + const asFile = join(directory, "file"); + await writeFile(asFile, "x"); + expect(new TraceStore(asFile).list()).rejects.toThrow(); + + // readFile on a trace path that is a directory -> EISDIR must throw. + await mkdir(join(directory, `${TRACE_A}.otlp.jsonl`)); + expect(store.list()).rejects.toThrow(); + }); +}); diff --git a/src/core/dev/otel/store.ts b/src/core/dev/otel/store.ts new file mode 100644 index 000000000..2d6c0ea00 --- /dev/null +++ b/src/core/dev/otel/store.ts @@ -0,0 +1,153 @@ +import { appendFile, mkdir, readFile, readdir } from "node:fs/promises"; +import { join } from "node:path"; +import { buildTraceDetail, extractTraceMeta, partitionByTraceId } from "./transforms"; +import type { OtlpPayload, OtlpResourceLog, OtlpResourceSpan } from "./types"; + +const OTLP_EXT = ".otlp.jsonl"; +const DEFAULT_LIST_WINDOW_MS = 12 * 60 * 60 * 1000; + +export interface TraceSummary { + traceId: string; + timestamp: string; + sessionId?: string; + spanCount: string; + resourceSpans?: unknown[]; + resourceLogs?: unknown[]; +} + +export interface TraceDetail { + resourceSpans?: unknown[]; + resourceLogs?: unknown[]; +} + +export interface ListTracesOptions { + serviceName?: string; + startTime?: number; + endTime?: number; + /** Keep only the newest N traces — the inspector re-polls this on every invocation. */ + limit?: number; +} + +/** + * Append-only local trace storage: one JSON Lines file per trace (named by its + * trace id), each line a per-trace slice of an OTLP export payload. No in-memory + * state — reads go to disk on demand, which is fine because the inspector only + * fetches traces on user actions. Malformed files and lines are skipped, never fatal. + */ +export class TraceStore { + constructor(private readonly directory: string) {} + + /** + * Persist one OTLP export payload, partitioned by trace id so a batch that + * carries several traces lands in each trace's own file. Spans and log + * records without a trace id are dropped. + */ + public async append(payload: OtlpPayload): Promise { + const partitions = partitionByTraceId(payload); + if (partitions.size === 0) return; + + await mkdir(this.directory, { recursive: true }); + await Promise.all( + [...partitions].map(([traceId, partition]) => + appendFile( + join(this.directory, `${sanitize(traceId)}${OTLP_EXT}`), + JSON.stringify(partition) + "\n", + ), + ), + ); + } + + /** List traces newest-first, filtered by service name and time range (default: last 12 hours). */ + public async list(options: ListTracesOptions = {}): Promise { + const now = Date.now(); + const start = options.startTime ?? now - DEFAULT_LIST_WINDOW_MS; + const end = options.endTime ?? now; + + const summaries: TraceSummary[] = []; + for (const file of await this.traceFiles()) { + const trace = await this.readTraceFile(file); + if (!trace) continue; + + const meta = extractTraceMeta(trace.resourceSpans, trace.resourceLogs); + // No id means every line failed to parse (empty/corrupt file), not a real trace. + if (!meta.traceId) continue; + if (meta.lastSeen < start || meta.firstSeen > end) continue; + if (options.serviceName && !meta.serviceNames.includes(options.serviceName)) continue; + + const detail = buildTraceDetail(trace.resourceSpans, trace.resourceLogs); + summaries.push({ + traceId: meta.traceId, + timestamp: new Date(meta.lastSeen).toISOString(), + sessionId: meta.sessionId, + // Count the spans the UI actually renders (post noise-filter), not raw records. + spanCount: String(countRenderedSpans(detail.resourceSpans)), + ...detail, + }); + } + + summaries.sort((a, b) => b.timestamp.localeCompare(a.timestamp)); + return options.limit === undefined ? summaries : summaries.slice(0, options.limit); + } + + /** All spans and logs for one trace, or undefined when the trace is unknown. */ + public async get(traceId: string): Promise { + const trace = await this.readTraceFile(`${sanitize(traceId)}${OTLP_EXT}`); + if (!trace) return undefined; + return buildTraceDetail(trace.resourceSpans, trace.resourceLogs); + } + + private async traceFiles(): Promise { + try { + return (await readdir(this.directory)).filter((file) => file.endsWith(OTLP_EXT)); + } catch (error) { + if (isNotFound(error)) return []; // No traces persisted yet — the dir is created on first append. + throw error; + } + } + + private async readTraceFile( + fileName: string, + ): Promise<{ resourceSpans: OtlpResourceSpan[]; resourceLogs: OtlpResourceLog[] } | undefined> { + let content: string; + try { + content = await readFile(join(this.directory, fileName), "utf8"); + } catch (error) { + // Unknown trace (get) or a file removed between listing and read; any other + // fault (permissions, bad path) is real and must not read as "no trace". + if (isNotFound(error)) return undefined; + throw error; + } + + const resourceSpans: OtlpResourceSpan[] = []; + const resourceLogs: OtlpResourceLog[] = []; + for (const line of content.split("\n")) { + if (!line.trim()) continue; + try { + const payload = JSON.parse(line) as OtlpPayload; + if (payload.resourceSpans) resourceSpans.push(...payload.resourceSpans); + if (payload.resourceLogs) resourceLogs.push(...payload.resourceLogs); + } catch { + // Skip malformed lines — a partially written line must not break reads. + } + } + return { resourceSpans, resourceLogs }; + } +} + +function sanitize(value: string): string { + return value.replace(/[^a-zA-Z0-9_-]/g, "_"); +} + +/** Number of spans in a built trace detail — what the inspector's waterfall shows. */ +function countRenderedSpans(resourceSpans: TraceDetail["resourceSpans"]): number { + let count = 0; + for (const resourceSpan of (resourceSpans ?? []) as OtlpResourceSpan[]) { + for (const scopeSpan of resourceSpan.scopeSpans ?? []) count += scopeSpan.spans?.length ?? 0; + } + return count; +} + +/** A missing directory or file — the only fs error reads should treat as "empty". */ +function isNotFound(error: unknown): boolean { + return (error as NodeJS.ErrnoException | null)?.code === "ENOENT"; +} diff --git a/src/core/dev/otel/transforms.test.ts b/src/core/dev/otel/transforms.test.ts new file mode 100644 index 000000000..180391284 --- /dev/null +++ b/src/core/dev/otel/transforms.test.ts @@ -0,0 +1,244 @@ +import { describe, expect, test } from "bun:test"; +import { + buildTraceDetail, + extractAnyValue, + extractTraceMeta, + flattenAttributes, + hexFromB64OrString, + nanoToMs, + partitionByTraceId, +} from "./transforms"; +import type { OtlpResourceLog, OtlpResourceSpan } from "./types"; + +const TRACE_ID_HEX = "0123456789abcdef0123456789abcdef"; +const TRACE_ID_B64 = Buffer.from(TRACE_ID_HEX, "hex").toString("base64"); +const SPAN_ID_HEX = "0123456789abcdef"; + +function resourceSpan(overrides: { serviceName?: string; spans: object[] }): OtlpResourceSpan { + return { + resource: overrides.serviceName + ? { attributes: [{ key: "service.name", value: { stringValue: overrides.serviceName } }] } + : undefined, + scopeSpans: [{ scope: { name: "test-scope" }, spans: overrides.spans }], + }; +} + +const agentSpan = { + traceId: TRACE_ID_B64, + spanId: SPAN_ID_HEX, + name: "invoke_agent strands", + kind: 1, + startTimeUnixNano: "1700000000000000000", + endTimeUnixNano: "1700000001500000000", + attributes: [ + { key: "gen_ai.prompt", value: { stringValue: "hello" } }, + { key: "session.id", value: { stringValue: "session-1" } }, + ], +}; + +describe("extractTraceMeta", () => { + test("collects trace id, time bounds, session, and service", () => { + const meta = extractTraceMeta( + [resourceSpan({ serviceName: "my-agent", spans: [agentSpan] })], + [], + ); + expect(meta).toEqual({ + traceId: TRACE_ID_HEX, + firstSeen: 1700000000000, + lastSeen: 1700000001500, + sessionId: "session-1", + serviceNames: ["my-agent"], + }); + }); + + test("reads trace id, service, and observed time from logs alone", () => { + const logs: OtlpResourceLog[] = [ + { + resource: { attributes: [{ key: "service.name", value: { stringValue: "log-agent" } }] }, + scopeLogs: [ + { + scope: {}, + logRecords: [{ traceId: TRACE_ID_HEX, observedTimeUnixNano: "1700000002000000000" }], + }, + ], + }, + ]; + const meta = extractTraceMeta([], logs); + expect(meta.traceId).toBe(TRACE_ID_HEX); + expect(meta.serviceNames).toEqual(["log-agent"]); + expect(meta.firstSeen).toBe(1700000002000); + expect(meta.lastSeen).toBe(1700000002000); + }); + + test("defaults time bounds to now when no timestamps exist", () => { + const before = Date.now(); + const meta = extractTraceMeta([], []); + expect(meta.firstSeen).toBeGreaterThanOrEqual(before); + expect(meta.lastSeen).toBeGreaterThanOrEqual(before); + expect(meta.traceId).toBeUndefined(); + }); +}); + +describe("partitionByTraceId", () => { + const OTHER_TRACE_HEX = "ffffffffffffffffffffffffffffffff"; + + test("splits a batch carrying several traces into per-trace payloads", () => { + const otherSpan = { ...agentSpan, traceId: OTHER_TRACE_HEX, name: "tool_use" }; + const partitions = partitionByTraceId({ + resourceSpans: [resourceSpan({ serviceName: "svc", spans: [agentSpan, otherSpan] })], + }); + + expect([...partitions.keys()].sort()).toEqual([TRACE_ID_HEX, OTHER_TRACE_HEX]); + const first = partitions.get(TRACE_ID_HEX)!.resourceSpans![0] as OtlpResourceSpan; + expect(first.scopeSpans![0]!.spans).toEqual([agentSpan]); + expect(first.resource).toBeDefined(); + const second = partitions.get(OTHER_TRACE_HEX)!.resourceSpans![0] as OtlpResourceSpan; + expect(second.scopeSpans![0]!.spans).toEqual([otherSpan]); + }); + + test("partitions log records by trace id and keys base64 ids as hex", () => { + const partitions = partitionByTraceId({ + resourceLogs: [ + { + scopeLogs: [ + { + scope: {}, + logRecords: [{ traceId: TRACE_ID_B64 }, { traceId: OTHER_TRACE_HEX }], + }, + ], + }, + ], + }); + + expect([...partitions.keys()].sort()).toEqual([TRACE_ID_HEX, OTHER_TRACE_HEX]); + }); + + test("drops spans without a trace id and returns empty for empty payloads", () => { + expect(partitionByTraceId({}).size).toBe(0); + const partitions = partitionByTraceId({ + resourceSpans: [resourceSpan({ spans: [{ name: "orphan" }] })], + }); + expect(partitions.size).toBe(0); + }); +}); + +describe("buildTraceDetail", () => { + test("hexes ids, flattens attributes, and unwraps log bodies", () => { + const detail = buildTraceDetail( + [resourceSpan({ serviceName: "svc", spans: [agentSpan] })], + [ + { + resource: { attributes: [{ key: "service.name", value: { stringValue: "svc" } }] }, + scopeLogs: [ + { + scope: {}, + logRecords: [ + { traceId: TRACE_ID_B64, spanId: SPAN_ID_HEX, body: { stringValue: "log line" } }, + ], + }, + ], + }, + ], + ); + + const spans = detail.resourceSpans as { + resource: { attributes: Record }; + scopeSpans: { spans: { traceId: string; attributes: Record }[] }[]; + }[]; + expect(spans[0]!.resource.attributes).toEqual({ "service.name": "svc" }); + expect(spans[0]!.scopeSpans[0]!.spans[0]!.traceId).toBe(TRACE_ID_HEX); + expect(spans[0]!.scopeSpans[0]!.spans[0]!.attributes).toEqual({ + "gen_ai.prompt": "hello", + "session.id": "session-1", + }); + + const logs = detail.resourceLogs as { + scopeLogs: { logRecords: { traceId: string; body: unknown }[] }[]; + }[]; + expect(logs[0]!.scopeLogs[0]!.logRecords[0]!.traceId).toBe(TRACE_ID_HEX); + expect(logs[0]!.scopeLogs[0]!.logRecords[0]!.body).toBe("log line"); + }); + + test("filters transport noise but keeps meaningful spans", () => { + const noiseSpans = [ + { name: "GET / http send", attributes: [] }, + { + name: "http.request", + attributes: [{ key: "asgi.event.type", value: { stringValue: "http.request" } }], + }, + { name: "POST", kind: 3, attributes: [] }, + { + name: "POST /invocations", + kind: 2, + attributes: [{ key: "http.method", value: { stringValue: "POST" } }], + }, + ]; + const detail = buildTraceDetail([resourceSpan({ spans: [...noiseSpans, agentSpan] })], []); + const spans = detail.resourceSpans as { scopeSpans: { spans: { name: string }[] }[] }[]; + expect(spans[0]!.scopeSpans[0]!.spans.map((span) => span.name)).toEqual([ + "invoke_agent strands", + ]); + }); + + test("string span kinds from JSON ingest are normalized before filtering", () => { + const detail = buildTraceDetail( + [resourceSpan({ spans: [{ name: "POST", kind: "SPAN_KIND_CLIENT", attributes: [] }] })], + [], + ); + expect(detail.resourceSpans).toBeUndefined(); + }); + + test("returns undefined sections when everything is filtered or empty", () => { + expect(buildTraceDetail([], [])).toEqual({ resourceSpans: undefined, resourceLogs: undefined }); + }); +}); + +describe("helpers", () => { + test("nanoToMs converts and handles absence", () => { + expect(nanoToMs("1700000000123456789")).toBe(1700000000123); + expect(nanoToMs(undefined)).toBe(0); + }); + + test("hexFromB64OrString accepts hex, base64, and empty", () => { + expect(hexFromB64OrString(TRACE_ID_HEX.toUpperCase())).toBe(TRACE_ID_HEX); + expect(hexFromB64OrString(TRACE_ID_B64)).toBe(TRACE_ID_HEX); + expect(hexFromB64OrString(undefined)).toBe(""); + }); + + test("flattenAttributes handles typed values, arrays, and kvlist, empty for none", () => { + expect( + flattenAttributes([ + { key: "s", value: { stringValue: "x" } }, + { key: "i", value: { intValue: "42" } }, + { key: "d", value: { doubleValue: 1.5 } }, + { key: "b", value: { boolValue: true } }, + { key: "a", value: { arrayValue: { values: [{ stringValue: "y" }, { intValue: "7" }] } } }, + { + key: "kv", + value: { kvlistValue: { values: [{ key: "inner", value: { intValue: "3" } }] } }, + }, + { key: "skipped" }, + ]), + ).toEqual({ s: "x", i: 42, d: 1.5, b: true, a: ["y", 7], kv: { inner: 3 } }); + expect(flattenAttributes([])).toBeUndefined(); + expect(flattenAttributes(undefined)).toBeUndefined(); + }); + + test("extractAnyValue unwraps nested kvlist and array values", () => { + expect( + extractAnyValue({ + kvlistValue: { + values: [ + { + key: "nested", + value: { arrayValue: { values: [{ intValue: "1" }, { boolValue: false }] } }, + }, + { key: "plain", value: { stringValue: "v" } }, + ], + }, + }), + ).toEqual({ nested: [1, false], plain: "v" }); + expect(extractAnyValue("passthrough")).toBe("passthrough"); + expect(extractAnyValue(null)).toBeNull(); + }); +}); diff --git a/src/core/dev/otel/transforms.ts b/src/core/dev/otel/transforms.ts new file mode 100644 index 000000000..cef0f5ca1 --- /dev/null +++ b/src/core/dev/otel/transforms.ts @@ -0,0 +1,298 @@ +import type { + OtlpAttributes, + OtlpPayload, + OtlpResource, + OtlpResourceLog, + OtlpResourceSpan, +} from "./types"; + +export interface TraceMeta { + traceId?: string; + firstSeen: number; + lastSeen: number; + sessionId?: string; + /** Every service participating in the trace — a distributed trace spans several local agents. */ + serviceNames: string[]; +} + +/** Extract listing metadata (trace id, time bounds, session, service) from raw OTLP arrays. */ +export function extractTraceMeta( + resourceSpans: OtlpResourceSpan[], + resourceLogs: OtlpResourceLog[], +): TraceMeta { + const meta: TraceMeta = { firstSeen: Infinity, lastSeen: 0, serviceNames: [] }; + const services = new Set(); + + for (const resourceSpan of resourceSpans) { + const service = getResourceAttribute(resourceSpan.resource, "service.name"); + if (service) services.add(service); + for (const scopeSpan of resourceSpan.scopeSpans ?? []) { + for (const span of scopeSpan.spans ?? []) { + meta.traceId ??= hexFromB64OrString(span.traceId) || undefined; + widenTimeBounds(meta, nanoToMs(span.startTimeUnixNano)); + widenTimeBounds(meta, nanoToMs(span.endTimeUnixNano)); + meta.sessionId ??= + getAttributeValue(span.attributes, "session.id") ?? + getAttributeValue(span.attributes, "attributes.session.id"); + } + } + } + + for (const resourceLog of resourceLogs) { + const service = getResourceAttribute(resourceLog.resource, "service.name"); + if (service) services.add(service); + for (const scopeLog of resourceLog.scopeLogs ?? []) { + for (const record of scopeLog.logRecords ?? []) { + meta.traceId ??= hexFromB64OrString(record.traceId) || undefined; + widenTimeBounds( + meta, + nanoToMs(record.timeUnixNano) || nanoToMs(record.observedTimeUnixNano), + ); + } + } + } + + const now = Date.now(); + if (meta.firstSeen === Infinity) meta.firstSeen = now; + if (meta.lastSeen === 0) meta.lastSeen = now; + meta.serviceNames = [...services]; + return meta; +} + +/** + * Split one OTLP export payload into per-trace payloads, keyed by hex trace id. + * A single export batch routinely carries spans from several traces (SDKs batch + * by time, not by trace), so persistence must not attribute a whole batch to + * the first trace id it sees. Spans and log records without a trace id are dropped. + * Resource and scope structure is preserved within each partition. + */ +export function partitionByTraceId(payload: OtlpPayload): Map { + const partitions = new Map(); + const partition = (traceId: string): OtlpPayload => { + let entry = partitions.get(traceId); + if (!entry) { + entry = {}; + partitions.set(traceId, entry); + } + return entry; + }; + + for (const resourceSpan of payload.resourceSpans ?? []) { + for (const scopeSpan of resourceSpan.scopeSpans ?? []) { + const byTrace = groupBy(scopeSpan.spans ?? [], (span) => hexFromB64OrString(span.traceId)); + for (const [traceId, spans] of byTrace) { + (partition(traceId).resourceSpans ??= []).push({ + resource: resourceSpan.resource, + scopeSpans: [{ scope: scopeSpan.scope, spans }], + }); + } + } + } + + for (const resourceLog of payload.resourceLogs ?? []) { + for (const scopeLog of resourceLog.scopeLogs ?? []) { + const byTrace = groupBy(scopeLog.logRecords ?? [], (record) => + hexFromB64OrString(record.traceId), + ); + for (const [traceId, logRecords] of byTrace) { + (partition(traceId).resourceLogs ??= []).push({ + resource: resourceLog.resource, + scopeLogs: [{ scope: scopeLog.scope, logRecords }], + }); + } + } + } + + return partitions; +} + +function groupBy(items: T[], key: (item: T) => string): Map { + const groups = new Map(); + for (const item of items) { + const groupKey = key(item); + if (!groupKey) continue; + const group = groups.get(groupKey); + if (group) group.push(item); + else groups.set(groupKey, [item]); + } + return groups; +} + +/** + * Build frontend-ready trace detail from raw OTLP arrays: ids to hex, attributes + * flattened to plain records, transport-noise spans dropped, log bodies unwrapped. + */ +export function buildTraceDetail( + resourceSpans: OtlpResourceSpan[], + resourceLogs: OtlpResourceLog[], +): { resourceSpans?: unknown[]; resourceLogs?: unknown[] } { + const spans = resourceSpans + .map((resourceSpan) => ({ + resource: resourceSpan.resource + ? { attributes: flattenAttributes(resourceSpan.resource.attributes) } + : undefined, + scopeSpans: resourceSpan.scopeSpans + ?.map((scopeSpan) => ({ + scope: scopeSpan.scope, + spans: scopeSpan.spans + ?.map((span) => ({ + ...span, + traceId: hexFromB64OrString(span.traceId), + spanId: hexFromB64OrString(span.spanId), + parentSpanId: hexFromB64OrString(span.parentSpanId), + attributes: flattenAttributes(span.attributes), + })) + .filter((span) => isMeaningfulSpan(span)), + })) + .filter((scopeSpan) => scopeSpan.spans && scopeSpan.spans.length > 0), + })) + .filter((resourceSpan) => resourceSpan.scopeSpans && resourceSpan.scopeSpans.length > 0); + + const logs = resourceLogs + .map((resourceLog) => ({ + resource: resourceLog.resource + ? { attributes: flattenAttributes(resourceLog.resource.attributes) } + : undefined, + scopeLogs: resourceLog.scopeLogs?.map((scopeLog) => ({ + scope: scopeLog.scope, + logRecords: scopeLog.logRecords?.map((record) => ({ + ...record, + traceId: hexFromB64OrString(record.traceId), + spanId: hexFromB64OrString(record.spanId), + body: record.body === undefined ? undefined : extractAnyValue(record.body), + attributes: flattenAttributes(record.attributes), + })), + })), + })) + .filter((resourceLog) => resourceLog.scopeLogs && resourceLog.scopeLogs.length > 0); + + return { + resourceSpans: spans.length > 0 ? spans : undefined, + resourceLogs: logs.length > 0 ? logs : undefined, + }; +} + +/** + * Whether a span carries application-level signal. Filters ASGI transport events, + * bare HTTP client/server noise, and other framework spans that add nothing in the UI. + */ +function isMeaningfulSpan(span: { + name?: string; + kind?: number | string; + attributes?: Record; +}): boolean { + const name = span.name ?? ""; + const attributes = span.attributes ?? {}; + const kind = normalizeSpanKind(span.kind); + + if (name.endsWith(" http send") || name.endsWith(" http receive")) return false; + if (attributes["asgi.event.type"]) return false; + if (Object.keys(attributes).some((key) => key.startsWith("gen_ai."))) return true; + if (attributes["rpc.system"] || attributes["rpc.method"]) return true; + + const scopeHints = ["strands", "bedrock", "langchain", "crewai", "autogen", "google_adk"]; + if (scopeHints.some((hint) => name.toLowerCase().includes(hint))) return true; + if (name === "tool_use" || name === "tool_call" || attributes["tool.name"]) return true; + + if (kind === SPAN_KIND.CLIENT && (name === "POST" || name === "GET" || name.startsWith("HTTP "))) + return false; + if (kind === SPAN_KIND.SERVER && name.startsWith("POST /") && attributes["http.method"]) + return false; + + return true; +} + +const SPAN_KIND = { INTERNAL: 1, SERVER: 2, CLIENT: 3, PRODUCER: 4, CONSUMER: 5 } as const; + +/** Normalize a span kind from its protobuf enum name or number to the numeric value. */ +function normalizeSpanKind(kind: number | string | undefined): number { + if (typeof kind === "number") return kind; + if (typeof kind === "string") { + const name = kind.replace(/^SPAN_KIND_/, "") as keyof typeof SPAN_KIND; + return SPAN_KIND[name] ?? 0; + } + return 0; +} + +/** Convert a nanosecond timestamp string to milliseconds (0 when absent). */ +export function nanoToMs(nano: string | undefined): number { + if (!nano) return 0; + return Math.floor(Number(nano) / 1_000_000); +} + +/** + * Normalize a trace/span id that may be base64 (protobuf JSON conversion) or + * already hex (JSON ingest) into lowercase hex. + */ +export function hexFromB64OrString(value: string | undefined): string { + if (!value) return ""; + if (/^[0-9a-f]+$/i.test(value) && (value.length === 32 || value.length === 16)) + return value.toLowerCase(); + try { + return Buffer.from(value, "base64").toString("hex"); + } catch { + return value; + } +} + +/** Flatten an OTLP key/value attribute array into a plain record. */ +export function flattenAttributes( + attributes: OtlpAttributes | undefined, +): Record | undefined { + if (!attributes || attributes.length === 0) return undefined; + + const flat: Record = {}; + for (const attribute of attributes) { + if (!attribute.value) continue; + // One unwrap for every AnyValue variant — string/int/double/bool/array/kvlist — + // so nested and kvlist-valued attributes flatten instead of silently dropping. + flat[attribute.key] = extractAnyValue(attribute.value); + } + return flat; +} + +/** Unwrap an OTLP AnyValue (string/int/double/bool/array/kvlist) into a plain value. */ +export function extractAnyValue(value: unknown): unknown { + if (!value || typeof value !== "object") return value; + const anyValue = value as Record; + if (anyValue.stringValue !== undefined) return anyValue.stringValue; + if (anyValue.intValue !== undefined) return Number(anyValue.intValue); + if (anyValue.doubleValue !== undefined) return anyValue.doubleValue; + if (anyValue.boolValue !== undefined) return anyValue.boolValue; + if (anyValue.arrayValue && typeof anyValue.arrayValue === "object") { + const { values } = anyValue.arrayValue as { values?: unknown[] }; + return (values ?? []).map(extractAnyValue); + } + if (anyValue.kvlistValue && typeof anyValue.kvlistValue === "object") { + const { values } = anyValue.kvlistValue as { values?: { key: string; value?: unknown }[] }; + const record: Record = {}; + for (const entry of values ?? []) { + record[entry.key] = entry.value === undefined ? undefined : extractAnyValue(entry.value); + } + return record; + } + return value; +} + +function getResourceAttribute(resource: OtlpResource | undefined, key: string): string | undefined { + return getAttributeValue(resource?.attributes, key); +} + +function getAttributeValue( + attributes: OtlpAttributes | undefined, + key: string, +): string | undefined { + if (!attributes) return undefined; + const attribute = attributes.find((entry) => entry.key === key); + if (!attribute?.value) return undefined; + return ( + attribute.value.stringValue ?? + (attribute.value.intValue != null ? String(attribute.value.intValue) : undefined) + ); +} + +function widenTimeBounds(meta: TraceMeta, timeMs: number): void { + if (!timeMs) return; + if (timeMs < meta.firstSeen) meta.firstSeen = timeMs; + if (timeMs > meta.lastSeen) meta.lastSeen = timeMs; +} diff --git a/src/core/dev/otel/types.ts b/src/core/dev/otel/types.ts new file mode 100644 index 000000000..2b0375445 --- /dev/null +++ b/src/core/dev/otel/types.ts @@ -0,0 +1,65 @@ +/** + * Wire shapes for OTLP/HTTP payloads after protobuf JSON conversion or JSON ingest. + * Attributes always arrive as OTLP key/value arrays; we flatten them to plain + * records only for display output, which is never read back through these types. + */ + +export interface OtlpAttributeValue { + stringValue?: string; + intValue?: string; + doubleValue?: number; + boolValue?: boolean; + arrayValue?: { values?: OtlpAttributeValue[] }; + kvlistValue?: { values?: OtlpAttribute[] }; +} + +export interface OtlpAttribute { + key: string; + value?: OtlpAttributeValue; +} + +export type OtlpAttributes = OtlpAttribute[]; + +export interface OtlpResource { + attributes?: OtlpAttributes; +} + +export interface OtlpSpan { + traceId?: string; + spanId?: string; + parentSpanId?: string; + name?: string; + kind?: number | string; + startTimeUnixNano?: string; + endTimeUnixNano?: string; + attributes?: OtlpAttributes; + status?: { code?: number; message?: string }; + events?: unknown[]; +} + +export interface OtlpResourceSpan { + resource?: OtlpResource; + scopeSpans?: { scope?: { name?: string; version?: string }; spans?: OtlpSpan[] }[]; +} + +export interface OtlpLogRecord { + timeUnixNano?: string; + observedTimeUnixNano?: string; + severityNumber?: number; + severityText?: string; + body?: unknown; + attributes?: OtlpAttributes; + traceId?: string; + spanId?: string; +} + +export interface OtlpResourceLog { + resource?: OtlpResource; + scopeLogs?: { scope?: { name?: string; version?: string }; logRecords?: OtlpLogRecord[] }[]; +} + +/** One OTLP export payload: what a single POST /v1/traces or /v1/logs carries. */ +export interface OtlpPayload { + resourceSpans?: OtlpResourceSpan[]; + resourceLogs?: OtlpResourceLog[]; +}