From d9025a5f5e27def5a1f5b200a26dd34054c97535 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Mon, 19 Oct 2026 16:50:11 +0000 Subject: [PATCH] feat: SQLite telemetry store in Chord, shared TelemetryStore and trace renderer - pi-telemetry: StoredSpan/StoredTrace/TelemetryStore (findTraces with name, attribute, error, time-window filters and cursor paging; getTrace; findMetrics), trace ids carried in carriers, renderTrace in the /render subpath; the in-memory adapter implements the store; the adapter contract trusts typed inputs (malformed-input conformance cases removed) - chord: @earendil-works/chord/telemetry-sqlite over chord/sqlite: settled spans with inline events and links, metrics aggregated per series per 10 s, 1 s batched flush with retention, bounded queue (dropped and counted), store queries --- packages/chord/CHANGELOG.md | 1 + packages/chord/README.md | 29 ++ packages/chord/package.json | 5 + packages/chord/src/telemetry-sqlite/index.ts | 363 ++++++++++++++++ packages/chord/src/telemetry-sqlite/store.ts | 171 ++++++++ .../chord/test/telemetry-sqlite-store.test.ts | 103 +++++ packages/chord/test/telemetry-sqlite.test.ts | 390 ++++++++++++++++++ packages/telemetry/CHANGELOG.md | 2 + packages/telemetry/README.md | 61 ++- packages/telemetry/package.json | 5 + packages/telemetry/src/index.ts | 67 +++ packages/telemetry/src/memory.ts | 352 ++++++++-------- packages/telemetry/src/metrics.ts | 7 +- packages/telemetry/src/noop.ts | 8 +- packages/telemetry/src/render.ts | 49 +++ packages/telemetry/src/testing/conformance.ts | 212 ++-------- packages/telemetry/test/metrics.test.ts | 43 +- packages/telemetry/test/propagation.test.ts | 30 +- packages/telemetry/test/render.test.ts | 147 +++++++ packages/telemetry/test/store-conformance.ts | 225 ++++++++++ packages/telemetry/test/store.test.ts | 82 ++++ packages/telemetry/test/telemetry.test.ts | 41 +- tsconfig.json | 2 + 23 files changed, 1958 insertions(+), 437 deletions(-) create mode 100644 packages/chord/src/telemetry-sqlite/index.ts create mode 100644 packages/chord/src/telemetry-sqlite/store.ts create mode 100644 packages/chord/test/telemetry-sqlite-store.test.ts create mode 100644 packages/chord/test/telemetry-sqlite.test.ts create mode 100644 packages/telemetry/src/render.ts create mode 100644 packages/telemetry/test/render.test.ts create mode 100644 packages/telemetry/test/store-conformance.ts create mode 100644 packages/telemetry/test/store.test.ts diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 8bd28808e..59ea796d3 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -8,6 +8,7 @@ ### Added +- `@earendil-works/chord/telemetry-sqlite`: portable SQLite telemetry recording with bounded settled-span queues, ten-second metric aggregates, periodic flushes, and retention cleanup. - `Context.telemetry` is an explicit, non-optional telemetry parent (no-op by default). `withTelemetry(telemetry, context)` derives it without changing values or cancellation; `traceIn(context, options, fn)` starts a child span and passes its derived context to the callback. - `@earendil-works/chord/sqlite`: portable SQLite interfaces, atomic migration runner, and schema-prefix validation, with Node (`/sqlite/node`) and Cloudflare Durable Object (`/sqlite/cloudflare`) drivers. Moved from Durable, which keeps its storage implementations and re-exports the moved public names. - `@earendil-works/chord/output`: `Progress` takes an optional `ProgressClock` (`now`, `setTimer`) to pace commits on, by default `Date.now` and `setTimeout`. diff --git a/packages/chord/README.md b/packages/chord/README.md index b3d2dc906..20f595ccf 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -19,6 +19,7 @@ Contracts are runtime values described with [TypeBox](https://github.com/sinclai | `@earendil-works/chord/context` | Explicit cancellation, telemetry, and invocation values: `BACKGROUND_CONTEXT`, `withCancel`, `withAbortSignal`, `withoutAbortSignal`, `withTelemetry`, `traceIn`, `withContextValue`, `awaitWithContext`. | Node and browsers | | `@earendil-works/chord/delta` | The transactional JSON tracker, structural diff, immutable apply, and the wire encoder/decoder for operation batches. | Node and browsers | | `@earendil-works/chord/output` | Bounded output streams and paced progress: `OutputBuffer` (head or tail window of a stream, `OutputLimits`, `OutputSkip`), `boundOutput`, `sanitizeOutput`, `utf8ByteLength`, and `Progress` (adaptive commits of a changing value). | Node and browsers | +| `@earendil-works/chord/telemetry-sqlite` | `createSqliteTelemetry(db, { retentionMs?, maxQueuedSpans? })`: buffered recording and backend-independent `TelemetryStore` queries. | Any SQLite adapter | | `@earendil-works/chord/sqlite` | Portable `SqliteDatabase`, `SqliteExecutor`, and `SqliteValue`, atomic `applySqliteMigrations(database, migrations, { table })`, and `validateSqliteSchemaPrefix`. The metadata table must be a valid, non-reserved SQLite identifier; migration versions start at 1 and are contiguous. | Any SQLite adapter | | `@earendil-works/chord/sqlite/node` | `NodeSqliteDatabase`, `openNodeSqliteDatabase`, and `NodeSqliteDatabaseOptions`. | Node | | `@earendil-works/chord/sqlite/cloudflare` | `DurableObjectSqliteDatabase` and the structural `DurableObjectSqliteStorage` binding type. | Cloudflare Durable Objects | @@ -91,6 +92,34 @@ await telemetry.startSpan({ name: 'app.request' }, async (span) => { These helpers operate on the standalone `Context` API. They do not yet instrument Chord hosts, methods, or network frames. +## SQLite telemetry + +```ts +import { createSqliteTelemetry } from '@earendil-works/chord/telemetry-sqlite'; +import { openNodeSqliteDatabase } from '@earendil-works/chord/sqlite/node'; +import { renderTrace } from '@earendil-works/pi-telemetry/render'; + +const db = await openNodeSqliteDatabase('telemetry.db'); +const telemetry = await createSqliteTelemetry(db); +try { + await telemetry.startSpan({ name: 'app.request' }, async (span) => { + span.addEvent('request.accepted'); + await span.startSpan({ name: 'app.read' }, () => readAccount()); + }); + await telemetry.flush(); + const { traces } = await telemetry.findTraces({ name: 'app.request', limit: 10 }); + for (const trace of traces) console.log(renderTrace(trace.spans)); +} finally { + try { await telemetry.close(); } finally { await db.close(); } +} +``` + +The adapter is portable over `SqliteDatabase` and owns no connection. Recording performs no SQL. A one-second timer flushes settled spans and completed ten-second metric intervals in one transaction, including retention cleanup. Metrics retain count, sum, minimum, and maximum per name/kind/unit/attribute series, not raw samples, buckets, or exemplars. `close()` stops recording and drains partial metric intervals without waiting for business callbacks or closing the database. Late span completions are ignored. + +Retention defaults to seven days. `maxQueuedSpans` defaults to 10,000 and counts settled spans awaiting commit, including an in-flight batch. The same limit independently bounds pending metric interval/series pairs. Overflow increments `droppedSpans` or `droppedMeasurements`; the limits bound counts, not bytes or active spans. A write failure stops recording permanently: `flush()` and `close()` reject with the original error without retrying. + +The adapter implements `TelemetryStore`: `findTraces()`, `getTrace()`, and `findMetrics()` read committed rows, including after adapter close while the database remains open. See [stored telemetry](../telemetry/README.md#stored-telemetry) for filters, pagination, and cross-process trace assembly. Schema tables are `telemetry_spans`, `telemetry_metrics`, and `telemetry_schema`. + ## Tracking JSON deltas Import the standalone transactional tracker from `@earendil-works/chord/delta`: diff --git a/packages/chord/package.json b/packages/chord/package.json index f88036b93..99878e741 100644 --- a/packages/chord/package.json +++ b/packages/chord/package.json @@ -31,6 +31,11 @@ "types": "./dist/node.d.ts", "import": "./dist/node.js" }, + "./telemetry-sqlite": { + "source": "./src/telemetry-sqlite/index.ts", + "types": "./dist/telemetry-sqlite/index.d.ts", + "import": "./dist/telemetry-sqlite/index.js" + }, "./sqlite": { "source": "./src/sqlite/index.ts", "types": "./dist/sqlite/index.d.ts", diff --git a/packages/chord/src/telemetry-sqlite/index.ts b/packages/chord/src/telemetry-sqlite/index.ts new file mode 100644 index 000000000..0f6ba6677 --- /dev/null +++ b/packages/chord/src/telemetry-sqlite/index.ts @@ -0,0 +1,363 @@ +import { + type InstrumentOptions, + NOOP_TELEMETRY_CONTEXT, + type SpanAttributes, + type SpanLink, + type SpanOptions, + type SpanStatus, + type TelemetryCarrier, + type TelemetryContext, + type TelemetryMetrics, + type TelemetryPropagation, + type TelemetrySpan, + type TelemetryStore, +} from "@earendil-works/pi-telemetry"; +import { applySqliteMigrations, type SqliteDatabase, type SqliteValue } from "../sqlite/index.ts"; +import { findMetrics, findTraces, getTrace } from "./store.ts"; + +export interface SqliteTelemetryOptions { + /** Retain seven days by default. Cleanup runs in the flush transaction. */ + readonly retentionMs?: number; + /** Settled spans awaiting commit, including the in-flight batch. Defaults to 10,000. + * Also bounds pending metric interval/series pairs independently. */ + readonly maxQueuedSpans?: number; +} + +export interface SqliteTelemetry extends TelemetryContext, TelemetryStore { + readonly metrics: TelemetryMetrics; + readonly propagation: TelemetryPropagation; + readonly droppedSpans: number; + readonly droppedMeasurements: number; + /** Persist settled spans and completed ten-second metric intervals. */ + flush(): Promise; + /** Stop recording and drain, including partial metric intervals. Does not close the database. */ + close(): Promise; +} + +type Identity = { id: string; traceId: string }; +type Link = { trace_id: string; span_id: string; attributes?: SpanAttributes }; +type Metric = { + interval: number; + key: string; + name: string; + kind: "counter" | "upDownCounter" | "histogram"; + unit: string | null; + attributes: string; + count: number; + sum: number; + min: number; + max: number; +}; +const identities = new WeakMap(); +const carrierKey = "pi-sqlite-span"; +const traceKey = "pi-sqlite-trace"; +const intervalMs = 10_000; + +function copyAttributes(attributes?: SpanAttributes): SpanAttributes { + return Object.fromEntries( + Object.entries(attributes ?? {}) + .filter(([, value]) => value !== undefined) + .map(([key, value]) => [key, Array.isArray(value) ? [...value] : value]), + ); +} + +function copyLink(link: SpanLink): Link | undefined { + const identity = identities.get(link.context); + return identity + ? { trace_id: identity.traceId, span_id: identity.id, attributes: copyAttributes(link.attributes) } + : undefined; +} + +class Recorder implements SqliteTelemetry, TelemetryPropagation { + readonly propagation: TelemetryPropagation = this; + readonly metrics: TelemetryMetrics = { + counter: (name, options) => ({ add: this.measurement("counter", name, options) }), + upDownCounter: (name, options) => ({ add: this.measurement("upDownCounter", name, options) }), + histogram: (name, options) => ({ record: this.measurement("histogram", name, options) }), + }; + droppedSpans = 0; + droppedMeasurements = 0; + private readonly db: SqliteDatabase; + private readonly retention: number; + private readonly capacity: number; + private readonly scope = crypto.randomUUID(); + private sequence = 0; + private readonly spans: SqliteValue[][] = []; + private readonly measurements = new Map(); + private inFlightMeasurements = 0; + private readonly timer: ReturnType; + private stopped = false; + private failed = false; + private failure: unknown; + private writing?: Promise; + private closing?: Promise; + + constructor(db: SqliteDatabase, retention: number, capacity: number) { + this.db = db; + this.retention = retention; + this.capacity = capacity; + this.timer = setInterval(() => { + if (!this.writing) void this.flush().catch(() => {}); + }, 1000); + this.timer.unref?.(); + } + + startSpan(options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { + return this.start(undefined, options, callback); + } + + inject(context: TelemetryContext): TelemetryCarrier | undefined { + const identity = identities.get(context); + return identity ? { [carrierKey]: identity.id, [traceKey]: identity.traceId } : undefined; + } + + extract(carrier: TelemetryCarrier | undefined): TelemetryContext { + const id = carrier?.[carrierKey]; + const traceId = carrier?.[traceKey]; + if (typeof id !== "string" || typeof traceId !== "string") return this; + const identity = { id, traceId }; + const parent: TelemetryContext = { + metrics: this.metrics, + propagation: this, + startSpan: (options, callback) => this.start(identity, options, callback), + }; + identities.set(parent, identity); + return parent; + } + + private inert(options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { + return NOOP_TELEMETRY_CONTEXT.startSpan(options, (noop) => + callback({ + ...noop, + metrics: this.metrics, + propagation: this, + startSpan: (options, callback) => this.inert(options, callback), + }), + ); + } + + private async start( + parent: Identity | undefined, + options: SpanOptions, + callback: (span: TelemetrySpan) => T | Promise, + ): Promise { + if (this.stopped) return this.inert(options, callback); + const name = options.name; + let attributes = copyAttributes(options.attributes); + const links: Link[] = []; + for (const link of options.links ?? []) { + const copied = copyLink(link); + if (copied) links.push(copied); + } + const id = `${this.scope}:${++this.sequence}`; + const identity = { id, traceId: parent?.traceId ?? id }; + const started = Date.now(); + const clock = performance.now(); + const events: { name: string; time_ms: number; attributes: SpanAttributes }[] = []; + let status: SpanStatus = { status: "ok" }; + let explicit = false; + let settled = false; + const record = (action: () => void): void => { + if (!settled && !this.stopped) action(); + }; + const span: TelemetrySpan = { + metrics: this.metrics, + propagation: this, + startSpan: (options, callback) => + settled ? this.inert(options, callback) : this.start(identity, options, callback), + addLink: (link) => + record(() => { + const copied = copyLink(link); + if (copied) links.push(copied); + }), + addEvent: (name, attrs) => + record(() => { + events.push({ name, time_ms: performance.now() - clock, attributes: copyAttributes(attrs) }); + }), + setAttributes: (attrs) => + record(() => { + attributes = { ...attributes, ...copyAttributes(attrs) }; + }), + setStatus: (value) => + record(() => { + status = + value.status === "error" && value.error + ? { status: "error", error: { ...value.error } } + : { status: value.status }; + explicit = true; + }), + }; + identities.set(span, identity); + try { + return await callback(span); + } catch (error) { + if (!explicit) + status = + error instanceof Error + ? { status: "error", error: { name: error.name, message: error.message } } + : { status: "error" }; + throw error; + } finally { + settled = true; + if (!this.stopped) { + if (this.spans.length >= this.capacity) this.droppedSpans++; + else + this.spans.push([ + id, + identity.traceId, + parent?.id ?? null, + name, + started, + started + performance.now() - clock, + JSON.stringify(attributes), + JSON.stringify(events), + JSON.stringify(links), + JSON.stringify(status), + ]); + } + } + } + + private measurement( + kind: Metric["kind"], + name: string, + options?: InstrumentOptions, + ): (value: number, attributes?: SpanAttributes) => void { + const unit = options?.unit ?? null; + return (value, attributes) => { + if (this.stopped) return; + const copied = copyAttributes(attributes); + const encoded = JSON.stringify( + Object.fromEntries( + Object.keys(copied) + .sort() + .map((key) => [key, copied[key]]), + ), + ); + const key = JSON.stringify([name, kind, unit, encoded]); + const interval = Math.floor(Date.now() / intervalMs) * intervalMs; + const pendingKey = `${interval}:${key}`; + const previous = this.measurements.get(pendingKey); + if (!previous && this.measurements.size + this.inFlightMeasurements >= this.capacity) { + this.droppedMeasurements++; + return; + } + this.measurements.set(pendingKey, { + interval, + key, + name, + kind, + unit, + attributes: encoded, + count: (previous?.count ?? 0) + 1, + sum: (previous?.sum ?? 0) + value, + min: Math.min(previous?.min ?? value, value), + max: Math.max(previous?.max ?? value, value), + }); + }; + } + + /** Readers use committed rows only; flush first to include newly settled spans. */ + readonly getTrace: TelemetryStore["getTrace"] = (traceId) => getTrace(this.db, traceId); + readonly findTraces: TelemetryStore["findTraces"] = (query) => findTraces(this.db, query); + readonly findMetrics: TelemetryStore["findMetrics"] = (query) => findMetrics(this.db, query); + + flush(): Promise { + return this.closing ?? this.drain(false); + } + + close(): Promise { + if (!this.closing) { + this.stopped = true; + clearInterval(this.timer); + this.closing = this.drain(true); + } + return this.closing; + } + + private async drain(partial: boolean): Promise { + while (this.writing) await this.writing; + if (this.failed) throw this.failure; + const spans = this.spans.slice(); + const metrics: Metric[] = []; + const completed = Math.floor(Date.now() / intervalMs) * intervalMs; + for (const [key, metric] of this.measurements) { + if (!partial && metric.interval >= completed) continue; + metrics.push(metric); + this.measurements.delete(key); + } + this.inFlightMeasurements = metrics.length; + this.writing = Promise.resolve() + .then(() => + this.db.transaction(async (tx) => { + for (const span of spans) + await tx.run("INSERT INTO telemetry_spans VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ...span); + for (const metric of metrics) + await tx.run( + `INSERT INTO telemetry_metrics VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(interval_ms, series_key) DO UPDATE SET count = count + excluded.count, + sum = sum + excluded.sum, min = min(min, excluded.min), max = max(max, excluded.max)`, + metric.interval, + metric.key, + metric.name, + metric.kind, + metric.unit, + metric.attributes, + metric.count, + metric.sum, + metric.min, + metric.max, + ); + const cutoff = Date.now() - this.retention; + await tx.run("DELETE FROM telemetry_spans WHERE ended_ms < ?", cutoff); + await tx.run("DELETE FROM telemetry_metrics WHERE interval_ms < ?", cutoff); + }), + ) + .then(() => { + this.spans.splice(0, spans.length); + }) + .catch((error: unknown) => { + this.failed = this.stopped = true; + this.failure = error; + this.spans.length = 0; + this.measurements.clear(); + clearInterval(this.timer); + throw error; + }) + .finally(() => { + this.writing = undefined; + this.inFlightMeasurements = 0; + }); + await this.writing; + } +} + +/** Portable, buffered SQLite recorder. Recording never performs SQL; flush runs every second. + * Only settled spans are stored. Metrics aggregate count/sum/min/max per ten-second interval. + * A write failure permanently stops recording and is reported by flush and close. */ +export async function createSqliteTelemetry( + db: SqliteDatabase, + options: SqliteTelemetryOptions = {}, +): Promise { + const retention = options.retentionMs ?? 7 * 24 * 60 * 60 * 1000; + const capacity = options.maxQueuedSpans ?? 10_000; + await applySqliteMigrations( + db, + [ + { + version: 1, + statements: [ + `CREATE TABLE telemetry_spans (id TEXT PRIMARY KEY, trace_id TEXT NOT NULL, parent_id TEXT, name TEXT, started_ms REAL, + ended_ms REAL NOT NULL, attributes TEXT, events TEXT, links TEXT, status TEXT)`, + "CREATE INDEX telemetry_spans_ended_ms ON telemetry_spans(ended_ms)", + "CREATE INDEX telemetry_spans_trace_started ON telemetry_spans(trace_id, started_ms)", + "CREATE INDEX telemetry_spans_started_ms ON telemetry_spans(started_ms)", + `CREATE TABLE telemetry_metrics (interval_ms INTEGER, series_key TEXT, name TEXT, kind TEXT, + unit TEXT, attributes TEXT, count INTEGER, sum REAL, min REAL, max REAL, PRIMARY KEY(interval_ms, series_key))`, + ], + }, + ], + { table: "telemetry_schema" }, + ); + return new Recorder(db, retention, capacity); +} diff --git a/packages/chord/src/telemetry-sqlite/store.ts b/packages/chord/src/telemetry-sqlite/store.ts new file mode 100644 index 000000000..6ca0b4493 --- /dev/null +++ b/packages/chord/src/telemetry-sqlite/store.ts @@ -0,0 +1,171 @@ +import type { + MetricQuery, + SpanAttributes, + SpanStatus, + StoredMetric, + StoredSpan, + StoredTrace, + TraceQuery, +} from "@earendil-works/pi-telemetry"; +import type { SqliteDatabase, SqliteValue } from "../sqlite/index.ts"; + +type SpanRow = { + id: string; + trace_id: string; + parent_id: string | null; + name: string; + started_ms: number; + ended_ms: number; + attributes: string; + events: string; + links: string; + status: string; +}; + +function storedSpan(row: SpanRow): StoredSpan { + return { + id: row.id, + traceId: row.trace_id, + ...(row.parent_id === null ? {} : { parentId: row.parent_id }), + name: row.name, + startedMs: row.started_ms, + endedMs: row.ended_ms, + attributes: JSON.parse(row.attributes) as SpanAttributes, + events: (JSON.parse(row.events) as { name: string; time_ms: number; attributes: SpanAttributes }[]).map( + (event) => ({ name: event.name, timeMs: event.time_ms, attributes: event.attributes }), + ), + links: (JSON.parse(row.links) as { trace_id: string; span_id: string; attributes?: SpanAttributes }[]).map( + (link) => ({ + traceId: link.trace_id, + spanId: link.span_id, + ...(link.attributes === undefined ? {} : { attributes: link.attributes }), + }), + ), + status: JSON.parse(row.status) as SpanStatus, + }; +} + +// JSON equality must distinguish boolean true from numeric 1. json_each also handles arbitrary attribute keys. +function attributesSql(attributes: TraceQuery["attributes"], column: string, params: SqliteValue[]): string { + return ( + Object.entries(attributes ?? {}) + .map(([key, value]) => { + const type = + typeof value === "number" + ? "entry.type IN ('integer', 'real')" + : typeof value === "boolean" + ? `entry.type = '${value ? "true" : "false"}'` + : "entry.type = 'text'"; + params.push(key, typeof value === "boolean" ? Number(value) : value); + return `EXISTS (SELECT 1 FROM json_each(${column}) entry WHERE entry.key = ? AND ${type} AND entry.value = ?)`; + }) + .join(" AND ") || "1" + ); +} + +export async function getTrace(db: SqliteDatabase, traceId: string): Promise { + const rows = await db.all( + "SELECT * FROM telemetry_spans WHERE trace_id = ? ORDER BY started_ms, id", + traceId, + ); + return rows.length ? { traceId, spans: rows.map(storedSpan) } : undefined; +} + +export async function findTraces( + db: SqliteDatabase, + query: TraceQuery = {}, +): Promise<{ traces: readonly StoredTrace[]; cursor?: string }> { + const limit = query.limit ?? 20; + const conditions: string[] = []; + const params: SqliteValue[] = []; + if (query.name !== undefined) { + conditions.push( + "(SELECT name FROM telemetry_spans WHERE trace_id = heads.trace_id ORDER BY started_ms, id LIMIT 1) = ?", + ); + params.push(query.name); + } + if (query.startMin !== undefined) { + conditions.push("started_ms >= ?"); + params.push(query.startMin); + } + if (query.startMax !== undefined) { + conditions.push("started_ms <= ?"); + params.push(query.startMax); + } + if (query.errors) + conditions.push( + "EXISTS (SELECT 1 FROM telemetry_spans WHERE trace_id = heads.trace_id AND json_extract(status, '$.status') = 'error')", + ); + if (query.attributes) + conditions.push( + `EXISTS (SELECT 1 FROM telemetry_spans candidate WHERE candidate.trace_id = heads.trace_id AND ${attributesSql(query.attributes, "candidate.attributes", params)})`, + ); + if (query.cursor !== undefined) { + const cursor = JSON.parse(query.cursor) as [number, string]; + conditions.push("(started_ms < ? OR (started_ms = ? AND trace_id > ?))"); + params.push(cursor[0], cursor[0], cursor[1]); + } + return db.transaction(async (tx) => { + const heads = await tx.all<{ trace_id: string; started_ms: number }>( + `WITH heads AS ( + SELECT trace_id, min(started_ms) started_ms FROM telemetry_spans GROUP BY trace_id + ) SELECT * FROM heads WHERE ${conditions.join(" AND ") || "1"} ORDER BY started_ms DESC, trace_id LIMIT ?`, + ...params, + limit + 1, + ); + const more = heads.length > limit; + if (more) heads.pop(); + if (!heads.length) return { traces: [] }; + const rows = await tx.all( + "SELECT * FROM telemetry_spans WHERE trace_id IN (SELECT value FROM json_each(?)) ORDER BY started_ms, id", + JSON.stringify(heads.map((head) => head.trace_id)), + ); + const grouped = new Map(); + for (const row of rows) { + const spans = grouped.get(row.trace_id) ?? []; + spans.push(storedSpan(row)); + grouped.set(row.trace_id, spans); + } + const last = heads.at(-1)!; + return { + traces: heads.map((head) => ({ traceId: head.trace_id, spans: grouped.get(head.trace_id)! })), + ...(more ? { cursor: JSON.stringify([last.started_ms, last.trace_id]) } : {}), + }; + }); +} + +export async function findMetrics(db: SqliteDatabase, query: MetricQuery): Promise { + const params: SqliteValue[] = [query.name]; + const conditions = ["name = ?"]; + if (query.startMin !== undefined) { + conditions.push("interval_ms >= ?"); + params.push(query.startMin); + } + if (query.startMax !== undefined) { + conditions.push("interval_ms <= ?"); + params.push(query.startMax); + } + conditions.push(attributesSql(query.attributes, "attributes", params)); + const rows = await db.all<{ + interval_ms: number; + name: string; + kind: StoredMetric["kind"]; + unit: string | null; + attributes: string; + count: number; + sum: number; + min: number; + max: number; + }>(`SELECT * FROM telemetry_metrics WHERE ${conditions.join(" AND ")} ORDER BY interval_ms, series_key`, ...params); + return rows.map((row) => ({ + intervalMs: row.interval_ms, + name: row.name, + kind: row.kind, + ...(row.unit === null ? {} : { unit: row.unit }), + attributes: JSON.parse(row.attributes) as SpanAttributes, + count: row.count, + sum: row.sum, + min: row.min, + max: row.max, + })); +} diff --git a/packages/chord/test/telemetry-sqlite-store.test.ts b/packages/chord/test/telemetry-sqlite-store.test.ts new file mode 100644 index 000000000..4da872546 --- /dev/null +++ b/packages/chord/test/telemetry-sqlite-store.test.ts @@ -0,0 +1,103 @@ +import { DatabaseSync } from "node:sqlite"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { telemetryStoreConformance } from "../../telemetry/test/store-conformance.ts"; +import { DurableObjectSqliteDatabase } from "../src/sqlite/cloudflare.ts"; +import type { SqliteExecutor, SqliteValue } from "../src/sqlite/index.ts"; +import { openNodeSqliteDatabase } from "../src/sqlite/node.ts"; +import { createSqliteTelemetry } from "../src/telemetry-sqlite/index.ts"; +import { durableObjectStorage } from "./sqlite/helpers.ts"; + +afterEach(() => vi.restoreAllMocks()); + +async function fixture() { + const db = await openNodeSqliteDatabase(":memory:"); + const context = await createSqliteTelemetry(db); + return { + db, + context, + store: context, + flush: () => context.flush(), + async [Symbol.asyncDispose]() { + try { + await context.close(); + } finally { + await db.close(); + } + }, + }; +} + +telemetryStoreConformance("SQLite telemetry store conformance", fixture); +telemetryStoreConformance("Durable Object SQLite telemetry store conformance", async () => { + const sqlite = new DatabaseSync(":memory:"); + const db = new DurableObjectSqliteDatabase(durableObjectStorage(sqlite)); + const context = await createSqliteTelemetry(db); + return { + context, + store: context, + flush: () => context.flush(), + async [Symbol.asyncDispose]() { + try { + await context.close(); + await db.close(); + } finally { + sqlite.close(); + } + }, + }; +}); + +describe("SQLite telemetry reads", () => { + it("uses two span queries for a page, not one query per trace", async () => { + await using f = await fixture(); + for (let index = 0; index < 25; index++) await f.context.startSpan({ name: "root" }, () => {}); + await f.flush(); + let queries = 0; + const original = f.db.transaction.bind(f.db); + vi.spyOn(f.db, "transaction").mockImplementation((callback) => + original((tx) => { + const counted: SqliteExecutor = { + exec: (sql) => tx.exec(sql), + run: (sql, ...params) => tx.run(sql, ...params), + get: (sql: string, ...params: SqliteValue[]) => tx.get(sql, ...params), + all: (sql: string, ...params: SqliteValue[]) => { + queries++; + return tx.all(sql, ...params); + }, + }; + return callback(counted); + }), + ); + expect((await f.context.findTraces({ limit: 20 })).traces).toHaveLength(20); + expect(queries).toBe(2); + queries = 0; + expect((await f.context.findTraces({ limit: 1 })).traces).toHaveLength(1); + expect(queries).toBe(2); + }); + + it("reads committed rows without implicitly flushing, including after close", async () => { + await using f = await fixture(); + await f.context.startSpan({ name: "pending" }, () => {}); + expect((await f.context.findTraces()).traces).toEqual([]); + f.context.metrics.counter("partial").add(2); + expect(await f.context.findMetrics({ name: "partial" })).toEqual([]); + await f.context.close(); + const trace = (await f.context.findTraces()).traces[0]!; + expect((await f.context.getTrace(trace.traceId))?.spans[0]?.name).toBe("pending"); + expect(await f.context.findMetrics({ name: "partial" })).toMatchObject([{ count: 1, sum: 2 }]); + }); + + it("keeps surviving descendants discoverable after a stored parent is removed", async () => { + await using f = await fixture(); + await f.context.startSpan({ name: "root" }, (root) => root.startSpan({ name: "child" }, () => {})); + await f.flush(); + const original = (await f.context.findTraces()).traces[0]!; + const parent = original.spans.find((span) => span.name === "root")!; + await f.db.run("DELETE FROM telemetry_spans WHERE id = ?", parent.id); + const trace = (await f.context.findTraces()).traces[0]!; + expect(trace.traceId).toBe(original.traceId); + expect(trace.spans).toHaveLength(1); + expect(trace.spans[0]).toMatchObject({ name: "child", parentId: parent.id }); + expect((await f.context.getTrace(original.traceId))?.spans).toHaveLength(1); + }); +}); diff --git a/packages/chord/test/telemetry-sqlite.test.ts b/packages/chord/test/telemetry-sqlite.test.ts new file mode 100644 index 000000000..a74b0316f --- /dev/null +++ b/packages/chord/test/telemetry-sqlite.test.ts @@ -0,0 +1,390 @@ +import type { + RecordedTelemetrySpan, + SpanAttributes, + SpanStatus, + TelemetryCarrier, + TelemetrySpan, +} from "@earendil-works/pi-telemetry"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createTelemetryAdapterConformance } from "../../telemetry/src/testing/index.ts"; +import type { SqliteDatabase } from "../src/sqlite/index.ts"; +import { openNodeSqliteDatabase } from "../src/sqlite/node.ts"; +import { createSqliteTelemetry, type SqliteTelemetryOptions } from "../src/telemetry-sqlite/index.ts"; + +type SpanRow = { + id: string; + trace_id: string; + parent_id: string | null; + name: string; + started_ms: number; + ended_ms: number; + attributes: string; + events: string; + links: string; + status: string; + rowid: number; +}; +type MetricRow = { + interval_ms: number; + series_key: string; + name: string; + kind: string; + unit: string | null; + attributes: string; + count: number; + sum: number; + min: number; + max: number; +}; + +async function fixture(options?: SqliteTelemetryOptions) { + const db = await openNodeSqliteDatabase(":memory:"); + const telemetry = await createSqliteTelemetry(db, options); + return { + db, + telemetry, + context: telemetry, + propagation: telemetry.propagation, + async getSpans(): Promise { + await telemetry.flush(); + const rows = await db.all("SELECT rowid, * FROM telemetry_spans"); + return rows + .map((row) => ({ + id: Number(row.id.split(":")[1]), + parentId: + row.parent_id?.split(":")[0] === row.id.split(":")[0] ? Number(row.parent_id?.split(":")[1]) : null, + context: { "pi-sqlite-span": row.id, "pi-sqlite-trace": row.trace_id }, + parentContext: row.parent_id + ? { "pi-sqlite-span": row.parent_id, "pi-sqlite-trace": row.trace_id } + : null, + name: row.name, + attributes: JSON.parse(row.attributes) as SpanAttributes, + events: (JSON.parse(row.events) as { name: string; attributes: SpanAttributes }[]).map( + ({ name, attributes }) => ({ name, attributes }), + ), + links: ( + JSON.parse(row.links) as { trace_id: string; span_id: string; attributes: SpanAttributes }[] + ).map((link) => ({ + context: { "pi-sqlite-span": link.span_id, "pi-sqlite-trace": link.trace_id }, + attributes: link.attributes, + })), + status: JSON.parse(row.status) as SpanStatus, + settled: true, + endSequence: row.rowid, + })) + .sort((a, b) => a.id - b.id); + }, + async [Symbol.asyncDispose]() { + try { + await telemetry.close(); + } finally { + await db.close(); + } + }, + }; +} + +afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); +}); + +describe("SQLite telemetry span and propagation conformance", () => { + for (const test of createTelemetryAdapterConformance(fixture).filter((test) => test.group !== "metrics")) + it(`${test.group}: ${test.name}`, test.run); +}); + +describe("SQLite telemetry buffering and aggregation", () => { + it("does not execute SQL while recording and stores only settled spans", async () => { + await using f = await fixture(); + const transaction = vi.spyOn(f.db, "transaction"); + const gate = Promise.withResolvers(); + let active!: TelemetrySpan; + const pending = f.telemetry.startSpan({ name: "active" }, (span) => { + active = span; + span.addEvent("begin"); + f.telemetry.metrics.counter("calls").add(1); + return gate.promise; + }); + expect(transaction).not.toHaveBeenCalled(); + await f.telemetry.flush(); + expect(await f.db.all("SELECT * FROM telemetry_spans")).toEqual([]); + gate.resolve(); + await pending; + expect(transaction).toHaveBeenCalledTimes(1); + active.addEvent("late"); + await f.telemetry.flush(); + const row = await f.db.get("SELECT * FROM telemetry_spans"); + expect(row?.ended_ms).toBeGreaterThanOrEqual(row?.started_ms ?? Infinity); + expect(JSON.parse(row?.events ?? "[]")).toEqual([{ name: "begin", time_ms: expect.any(Number), attributes: {} }]); + }); + + it("aggregates canonical series by interval and writes partial intervals only on close", async () => { + vi.spyOn(Date, "now").mockReturnValue(20_001); + await using f = await fixture(); + const metric = f.telemetry.metrics.histogram("duration", { unit: "ms", boundaries: [1, 10] }); + const attrs = { b: ["original"], a: true }; + metric.record(2, attrs); + attrs.b.push("changed"); + metric.record(8, { a: true, b: ["original"] }); + f.telemetry.metrics.upDownCounter("active").add(-1); + await f.telemetry.flush(); + expect(await f.db.all("SELECT * FROM telemetry_metrics")).toEqual([]); + vi.mocked(Date.now).mockReturnValue(30_001); + metric.record(4, { b: ["original"], a: true }); + await f.telemetry.flush(); + const rows = await f.db.all("SELECT * FROM telemetry_metrics ORDER BY name"); + expect(rows).toHaveLength(2); + expect(rows[0]).toMatchObject({ + name: "active", + kind: "upDownCounter", + unit: null, + count: 1, + sum: -1, + min: -1, + max: -1, + }); + expect(rows[1]).toMatchObject({ + interval_ms: 20_000, + name: "duration", + kind: "histogram", + unit: "ms", + attributes: '{"a":true,"b":["original"]}', + count: 2, + sum: 10, + min: 2, + max: 8, + }); + await f.telemetry.close(); + expect(await f.db.get("SELECT count, sum FROM telemetry_metrics WHERE interval_ms = 30000")).toEqual({ + count: 1, + sum: 4, + }); + }); + + it("separates kind, name, unit, and caller attributes without retaining descriptions or boundaries", async () => { + await using f = await fixture(); + f.telemetry.metrics.counter("same").add(1); + f.telemetry.metrics.counter("same", { unit: "By" }).add(2); + f.telemetry.metrics.upDownCounter("same").add(3); + f.telemetry.metrics.counter("other").add(4); + f.telemetry.metrics.counter("same").add(5, { method: "read" }); + f.telemetry.metrics.histogram("same", { description: "first", boundaries: [1] }).record(6); + f.telemetry.metrics.histogram("same", { description: "second", boundaries: [2] }).record(7); + await f.telemetry.close(); + const rows = await f.db.all("SELECT * FROM telemetry_metrics"); + expect(rows).toHaveLength(6); + expect(rows.find((row) => row.kind === "histogram")).toMatchObject({ count: 2, sum: 13, min: 6, max: 7 }); + }); + + it("strictly bounds queued spans including the in-flight batch and bounds pending metric series", async () => { + await using f = await fixture({ maxQueuedSpans: 1 }); + await f.telemetry.startSpan({ name: "first" }, () => {}); + f.telemetry.metrics.counter("first").add(1); + f.telemetry.metrics.counter("first").add(2); + f.telemetry.metrics.counter("overflow").add(3); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const original = f.db.transaction.bind(f.db); + vi.spyOn(f.db, "transaction").mockImplementation(async (callback) => { + entered.resolve(); + await release.promise; + return original(callback); + }); + const flushing = f.telemetry.flush(); + await entered.promise; + await f.telemetry.startSpan({ name: "overflow" }, () => {}); + expect(f.telemetry.droppedSpans).toBe(1); + expect(f.telemetry.droppedMeasurements).toBe(1); + release.resolve(); + await flushing; + await f.telemetry.startSpan({ name: "after" }, () => {}); + await f.telemetry.close(); + expect(await f.db.all("SELECT name FROM telemetry_spans ORDER BY rowid")).toEqual([ + { name: "first" }, + { name: "after" }, + ]); + expect(await f.db.all("SELECT name, count, sum FROM telemetry_metrics")).toEqual([ + { name: "first", count: 2, sum: 3 }, + ]); + }); + + it("counts in-flight metric intervals against capacity until commit", async () => { + vi.spyOn(Date, "now").mockReturnValue(20_001); + await using f = await fixture({ maxQueuedSpans: 1 }); + f.telemetry.metrics.counter("calls").add(1); + vi.mocked(Date.now).mockReturnValue(30_001); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const original = f.db.transaction.bind(f.db); + vi.spyOn(f.db, "transaction").mockImplementation(async (callback) => { + entered.resolve(); + await release.promise; + return original(callback); + }); + const flushing = f.telemetry.flush(); + await entered.promise; + f.telemetry.metrics.counter("calls").add(2); + expect(f.telemetry.droppedMeasurements).toBe(1); + release.resolve(); + await flushing; + f.telemetry.metrics.counter("calls").add(3); + await f.telemetry.close(); + expect(await f.db.all("SELECT interval_ms, sum FROM telemetry_metrics ORDER BY interval_ms")).toEqual([ + { interval_ms: 20_000, sum: 1 }, + { interval_ms: 30_000, sum: 3 }, + ]); + }); + + it("flushes once per second and retains only rows inside the retention window", async () => { + vi.useFakeTimers(); + vi.setSystemTime(20_001); + await using f = await fixture({ retentionMs: 20_000 }); + await f.telemetry.startSpan({ name: "old" }, () => {}); + f.telemetry.metrics.counter("old").add(1); + await vi.advanceTimersByTimeAsync(1000); + expect(await f.db.all("SELECT name FROM telemetry_spans")).toEqual([{ name: "old" }]); + vi.setSystemTime(50_001); + await f.telemetry.startSpan({ name: "recent" }, () => {}); + f.telemetry.metrics.counter("recent").add(1); + await vi.advanceTimersByTimeAsync(1000); + expect(await f.db.all("SELECT name FROM telemetry_spans")).toEqual([{ name: "recent" }]); + expect(await f.db.all("SELECT name FROM telemetry_metrics")).toEqual([]); + await f.telemetry.close(); + expect(await f.db.all("SELECT name FROM telemetry_metrics")).toEqual([{ name: "recent" }]); + }); + + it("serializes concurrent flush and close without losing measurements recorded during writes", async () => { + vi.spyOn(Date, "now").mockReturnValue(20_001); + await using f = await fixture(); + f.telemetry.metrics.counter("calls").add(1); + vi.mocked(Date.now).mockReturnValue(30_001); + const original = f.db.transaction.bind(f.db); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + let active = 0; + let maximum = 0; + vi.spyOn(f.db, "transaction").mockImplementation(async (callback) => { + maximum = Math.max(maximum, ++active); + entered.resolve(); + await release.promise; + try { + return await original(callback); + } finally { + active--; + } + }); + const first = f.telemetry.flush(); + await entered.promise; + f.telemetry.metrics.counter("calls").add(2); + const second = f.telemetry.flush(); + const close = f.telemetry.close(); + expect(f.telemetry.close()).toBe(close); + release.resolve(); + await Promise.all([first, second, close]); + expect(maximum).toBe(1); + expect(await f.db.all("SELECT interval_ms, sum FROM telemetry_metrics ORDER BY interval_ms")).toEqual([ + { interval_ms: 20_000, sum: 1 }, + { interval_ms: 30_000, sum: 2 }, + ]); + }); + + it("close does not wait for active business callbacks, ignores late completions, and leaves the database open", async () => { + await using f = await fixture(); + const release = Promise.withResolvers(); + let active!: TelemetrySpan; + const pending = f.telemetry.startSpan({ name: "active" }, (span) => { + active = span; + return release.promise; + }); + await f.telemetry.startSpan({ name: "settled" }, () => {}); + await f.telemetry.close(); + expect(await f.db.all("SELECT name FROM telemetry_spans")).toEqual([{ name: "settled" }]); + active.addEvent("late"); + await active.startSpan({ name: "late-child" }, () => {}); + f.telemetry.metrics.counter("late").add(1); + release.resolve(); + await pending; + await f.telemetry.flush(); + expect(await f.db.all("SELECT name FROM telemetry_spans")).toEqual([{ name: "settled" }]); + expect(await f.db.all("SELECT * FROM telemetry_metrics")).toEqual([]); + }); + + it("stops permanently after a write failure and reports the identical error without retry", async () => { + const db = await openNodeSqliteDatabase(":memory:"); + try { + const telemetry = await createSqliteTelemetry(db); + await telemetry.startSpan({ name: "queued" }, () => {}); + const error = new Error("write failure"); + const transaction = vi.spyOn(db, "transaction").mockRejectedValue(error); + await expect(telemetry.flush()).rejects.toBe(error); + await telemetry.startSpan({ name: "ignored" }, () => 1); + telemetry.metrics.counter("ignored").add(1); + await expect(telemetry.flush()).rejects.toBe(error); + const close = telemetry.close(); + expect(telemetry.close()).toBe(close); + await expect(close).rejects.toBe(error); + expect(transaction).toHaveBeenCalledTimes(1); + expect(await db.all("SELECT * FROM telemetry_spans")).toEqual([]); + } finally { + await db.close(); + } + }); + + it("remembers an undefined background write failure and stops its timer", async () => { + vi.useFakeTimers(); + const db = await openNodeSqliteDatabase(":memory:"); + try { + const telemetry = await createSqliteTelemetry(db); + const transaction = vi.spyOn(db, "transaction").mockRejectedValue(undefined); + await vi.advanceTimersByTimeAsync(1000); + await expect(telemetry.flush()).rejects.toBeUndefined(); + await expect(telemetry.close()).rejects.toBeUndefined(); + await vi.advanceTimersByTimeAsync(5000); + expect(transaction).toHaveBeenCalledTimes(1); + } finally { + await db.close(); + } + }); + + it("rolls back spans and metric aggregates together on SQL failure", async () => { + const db = await openNodeSqliteDatabase(":memory:"); + try { + const telemetry = await createSqliteTelemetry(db); + await telemetry.startSpan({ name: "rollback" }, () => {}); + telemetry.metrics.counter("rollback").add(1); + await db.exec("DROP TABLE telemetry_metrics"); + await expect(telemetry.close()).rejects.toThrow(); + expect(await db.all("SELECT * FROM telemetry_spans")).toEqual([]); + } finally { + await db.close(); + } + }); + + it("reuses migrations, preserves identities across reopen, and merges aggregates from multiple adapters", async () => { + const db: SqliteDatabase = await openNodeSqliteDatabase(":memory:"); + try { + const first = await createSqliteTelemetry(db); + let carrier: TelemetryCarrier | undefined; + await first.startSpan({ name: "parent" }, (span) => { + carrier = first.propagation.inject(span); + }); + first.metrics.counter("shared").add(2); + await first.close(); + const second = await createSqliteTelemetry(db); + await second.propagation.extract(carrier).startSpan({ name: "child" }, () => {}); + second.metrics.counter("shared").add(4); + await second.close(); + const rows = await db.all("SELECT * FROM telemetry_spans ORDER BY rowid"); + expect(rows[1]?.parent_id).toBe(rows[0]?.id); + expect(rows[1]?.trace_id).toBe(rows[0]?.trace_id); + expect(rows[0]?.trace_id).toBe(rows[0]?.id); + expect(rows[1]?.id).not.toBe(rows[0]?.id); + expect(await db.all("SELECT count, sum, min, max FROM telemetry_metrics")).toEqual([ + { count: 2, sum: 6, min: 2, max: 4 }, + ]); + } finally { + await db.close(); + } + }); +}); diff --git a/packages/telemetry/CHANGELOG.md b/packages/telemetry/CHANGELOG.md index 37ae6ad52..47c7bdc48 100644 --- a/packages/telemetry/CHANGELOG.md +++ b/packages/telemetry/CHANGELOG.md @@ -4,6 +4,7 @@ ### Breaking Changes +- Adapters trust typed instrumentation inputs; behavior for malformed values is unspecified. Carrier extraction checks required string identities, without validating their format. - Adapters must implement `TelemetrySpan.addLink()`. Normalized span snapshots include carrier-shaped context and parent identities and ordered links. - Adapters with optional metric or propagation capabilities must carry them through `TelemetryContext.metrics` and `TelemetryContext.propagation` on roots, spans, and extracted parents. @@ -16,6 +17,7 @@ - `NOOP_TELEMETRY_CONTEXT` is typed as `TelemetrySpan`, matching the shared inert span it provides. - Metric schema definitions and `createTypedMetrics()` with exact instrument names, kinds, and required/optional low-cardinality attributes. - In-memory propagation across processes, link and measurement snapshots, and adapter conformance cases for propagation, links, and optional metrics. +- `StoredSpan`, `StoredTrace`, and `TelemetryStore` define backend-independent trace discovery, trace retrieval, and metric queries. `@earendil-works/pi-telemetry/render` exports `renderTrace()` for plain-text span trees with attributes, durations, errors, events, and cross-trace links; merged spans are deduplicated by ID and links are never followed. ## [1.3.0] - 2026-10-15 diff --git a/packages/telemetry/README.md b/packages/telemetry/README.md index a8cd21426..206849ab9 100644 --- a/packages/telemetry/README.md +++ b/packages/telemetry/README.md @@ -24,6 +24,7 @@ Applications can use the in-memory reference or provide an adapter for OpenTelem - [Metrics](#metrics) - [No-op Context](#no-op-context) - [In-Memory Reference Adapter](#in-memory-reference-adapter) +- [Stored Telemetry](#stored-telemetry) - [Adapter Conformance](#adapter-conformance) - [Typed Schemas](#typed-schemas) - [Start and Completion Attributes](#start-and-completion-attributes) @@ -63,7 +64,7 @@ example.account.load span └─ status: ok final outcome ``` -A span is diagnostic data, not business state. Recording it must not change whether the account load runs, succeeds, fails, or is persisted. An adapter translates these generic concepts into the corresponding concepts used by OpenTelemetry, Sentry, logs, or another backend. +A span is diagnostic data, not business state. An adapter translates these generic concepts into the corresponding concepts used by OpenTelemetry, Sentry, logs, or another backend. ## Core Context API @@ -121,7 +122,9 @@ return telemetryContext.startSpan({ name: 'example.save' }, async (span) => { ## Adapter Contract -An adapter implements `TelemetryContext` and bridges the generic API to its backend. It must: +An adapter implements `TelemetryContext` and bridges the generic API to its backend. Typed inputs are trusted; malformed inputs have unspecified behavior. Adapters are not required to validate inputs or suppress adapter/backend failures. + +The callback and recording contract is: - create a child span and invoke the callback synchronously, exactly once; - expose optional metric and propagation capabilities on its contexts, child spans, and extracted parents, including inert fallback spans; @@ -130,9 +133,8 @@ An adapter implements `TelemetryContext` and bridges the generic API to its back - treat normal completion as `ok` and throws/rejections as errors unless an explicit status was set; - make repeated `setStatus()` calls last-write-wins; - merge `setAttributes()` calls, with later defined values replacing earlier values and `undefined` ignored; -- make recording methods synchronous, passive, and non-throwing; -- ignore calls made after settlement; -- ignore a failed recording call atomically, suppress backend failures, and still execute the business callback exactly once. +- make recording methods synchronous; +- ignore calls made after settlement. Adapters may activate backend-native ambient context internally for automatic instrumentation, but pi code always propagates the parent through `TelemetryContext` arguments. Exporter buffering, flushing, sampling, backend IDs, and backend-specific context objects belong to the adapter. Use the [adapter conformance suite](#adapter-conformance) to check these observable semantics. @@ -162,7 +164,7 @@ await caller.startSpan({ name: 'rpc.client' }, async (span) => { }); ``` -Injection returns a detached carrier or `undefined` for roots, no-op contexts, and unsupported contexts. Extraction returns a receiver-owned parent. Absent, invalid, unknown, or unreadable carriers fall back to the adapter's root, so incoming work is still recorded. Both methods are synchronous, passive, and non-throwing. Trace metadata is untrusted diagnostics, never authorization. Transport boundaries own carrier size limits. +Injection returns a detached carrier or `undefined` for roots, no-op contexts, and unsupported contexts. Extraction returns a receiver-owned parent. Absent or unknown carriers fall back to the adapter's root. Extraction checks only that required carrier keys contain strings; identifiers are opaque and need no format validation. Both methods are synchronous. Trace metadata is diagnostics, never authorization. Transport boundaries own carrier size limits. A link relates work without changing its parent. Use `SpanOptions.links` when the target is known at creation, or `span.addLink()` after the span has started: @@ -176,9 +178,9 @@ await telemetry.startSpan( ); ``` -Link targets are `TelemetryContext` handles, including spans and extracted remote parents. For a remote link, first call `adapter.extract(carrier)`. A settled target remains linkable and injectable; a settled source ignores `addLink()`. Links append in order, copy attributes, and ignore root, no-op, unsupported, or malformed targets. Failed link calls are atomic and never throw. The typed span starter's callback also supports `addLink()`; initial links use the low-level `startSpan()` API. +Link targets are `TelemetryContext` handles, including spans and extracted remote parents. For a remote link, first call `adapter.extract(carrier)`. A settled target remains linkable and injectable; a settled source ignores `addLink()`. Links append in order, copy attributes, and ignore root, no-op, or unsupported contexts. The typed span starter's callback also supports `addLink()`; initial links use the low-level `startSpan()` API. -`NOOP_TELEMETRY_PROPAGATION` never inspects its arguments: injection returns `undefined`, extraction returns `NOOP_TELEMETRY_CONTEXT`. +`NOOP_TELEMETRY_PROPAGATION` ignores its arguments: injection returns `undefined`, extraction returns `NOOP_TELEMETRY_CONTEXT`. ## Metrics @@ -200,15 +202,15 @@ const duration = telemetry.histogram('call.duration', { duration.record(elapsedMs, { method: 'read' }, currentSpan); ``` -Factories take an instrument name and optional `InstrumentOptions` (`description`, `unit`). Histograms additionally accept `HistogramOptions.boundaries`: finite, strictly increasing bucket hints. Backends own aggregation and may translate these hints. Counters accept non-negative finite increments; up-down counters and histograms accept finite signed values. Histogram `record(value, attributes?, context?)` optionally correlates a measurement with a span, called an exemplar. Counter `add(value, attributes?)` has no context argument. +Factories take an instrument name and optional `InstrumentOptions` (`description`, `unit`). Histograms additionally accept `HistogramOptions.boundaries`: callers supply finite, strictly increasing bucket hints. Backends own aggregation and may translate these hints. Callers are responsible for finite measurements and non-negative counter increments; up-down counters and histograms may use signed values. Histogram `record(value, attributes?, context?)` optionally correlates a measurement with a span, called an exemplar. Counter `add(value, attributes?)` has no context argument. -Factories and recording methods are synchronous, passive, and non-throwing. Invalid measurements, unreadable metadata/attributes, or invalid boundaries are ignored atomically. `NOOP_TELEMETRY_METRICS` returns shared frozen instruments without inspecting inputs. Record only values already available to business code; do not serialize or measure extra data for telemetry. +Factories and recording methods are synchronous and trust typed inputs. Adapters need not validate measurement values, metadata, attributes, or boundaries; malformed inputs have unspecified behavior. `NOOP_TELEMETRY_METRICS` returns shared frozen instruments that discard measurements. Record only values already available to business code; do not serialize or measure extra data for telemetry. Use only low-cardinality attributes: bounded values such as direction or method, not request IDs, user IDs, prompts, or arbitrary paths. The low-level attribute bag remains open; instrumentation owns this policy, not backend-specific heuristics. ### Typed metric schemas -`TelemetrySchemaDefinition.metrics` optionally declares instrument names, kinds, metadata, and required/optional attributes. `createTypedMetrics()` combines one or more schemas and rejects duplicate metric names, incorrect instrument kinds, unknown attributes, and missing required attributes: +`TelemetrySchemaDefinition.metrics` optionally declares instrument names, kinds, metadata, and required/optional attributes. `createTypedMetrics()` combines one or more schemas and rejects duplicate metric names, incorrect instrument kinds, unknown attributes, and missing required attributes at compile time: ```typescript const schema = defineTelemetrySchema({ @@ -233,7 +235,7 @@ metrics.counter('network.bytes').add(encodedSize, { direction: 'out' }); Metric definitions use `kind: 'counter' | 'upDownCounter' | 'histogram'`; only histograms may declare `boundaries`. Metric attributes use the same scalar/array definitions as spans, but reject `cardinality: 'high'`. Omitting the cardinality annotation does not waive the low-cardinality policy. Unlike the typed span starter, the metric helper reads and retains schema instrument metadata at runtime. It performs no runtime attribute validation. -`InMemoryTelemetryContext.getMetrics()` returns ordered, detached `RecordedTelemetryMeasurement` snapshots: `kind`, `name`, optional `description`/`unit`/`boundaries`, `value`, copied `attributes`, and a carrier-shaped exemplar `context` or `null`. This is a measurement log, not aggregated counters or buckets; storage is unbounded and records no timestamps. +`InMemoryTelemetryContext.getMetrics()` returns ordered, detached `RecordedTelemetryMeasurement` snapshots: `kind`, `name`, optional `description`/`unit`/`boundaries`, `value`, copied `attributes`, and a carrier-shaped exemplar `context` or `null`. This is a measurement log, not aggregated counters or buckets; storage is unbounded. Measurement times are retained privately for store queries, not exposed in these snapshots. ## No-op Context @@ -276,12 +278,35 @@ await telemetry.startSpan( console.log(telemetry.getSpans()); ``` -`getSpans()` returns detached snapshots in span-start order. Each `RecordedTelemetrySpan` contains a deterministic numeric ID, local parent ID, carrier-shaped `context` and `parentContext` snapshots, ordered `links`, merged attributes, ordered events, final status, settlement state, and deterministic end sequence. It records no timestamps. A remote parent has `parentId: null` but a non-null `parentContext`; a root has both null. A `RecordedTelemetryLink` contains a target carrier and copied attributes. +`getSpans()` returns detached snapshots in span-start order. Each `RecordedTelemetrySpan` contains a deterministic numeric ID, local parent ID, carrier-shaped `context` and `parentContext` snapshots, ordered `links`, merged attributes, ordered events, final status, settlement state, and deterministic end sequence. These normalized snapshots expose no timestamps. A remote parent has `parentId: null` but a non-null `parentContext`; a root has both null. A `RecordedTelemetryLink` contains a target carrier and copied attributes. -The adapter implements `TelemetryPropagation` with its own small carrier codec: a unique recorder scope and a deterministic numeric span ID. The carrier survives JSON round trips and works across independent processes without a shared registry. Its keys are an implementation detail, not a public wire format for other adapters. +The adapter implements `TelemetryPropagation` with its own small carrier codec: a unique recorder scope, a deterministic numeric span ID, and an inherited trace ID. The carrier survives JSON round trips and works across independent processes without a shared registry. Its keys are an implementation detail, not a public wire format for other adapters. The adapter is safe to use as an ordinary `TelemetryContext`, but storage is unbounded and process-local. Create a fresh instance to isolate tests or recording scopes, and do not capture sensitive attributes unless the caller's data policy allows them. +## Stored Telemetry + +`TelemetryStore` separates querying from recording. Both `InMemoryTelemetryContext` and `@earendil-works/chord/telemetry-sqlite` implement it: + +- `findTraces(query?)` returns complete locally stored traces, newest trace start first. Filters include the earliest stored span's `name`, inclusive `startMin`/`startMax`, `errors: true`, and scalar `attributes`. All requested attributes must match one span, anywhere in the trace. Pages default to 20 traces; pass the returned opaque `cursor` with the same filters to continue. Equal start times are ordered by trace ID. +- `getTrace(traceId)` returns a trace or `undefined`. +- `findMetrics({ name, startMin?, startMax?, attributes? })` returns metric rows with count, sum, minimum, and maximum. Time bounds are inclusive interval-start times. The in-memory adapter maps each raw measurement to a count-one row in its ten-second interval; SQLite aggregates measurements per series and interval. + +Only settled spans are returned. A trace ID is the root's globally unique span ID (`:`) and propagates with the parent identity. Stores return the spans available locally: remote or expired parents can be absent, and a child can appear before its parent settles. Pagination is not a snapshot across requests; late settlements and retention can change a trace's earliest stored span. SQLite reads committed rows without implicitly flushing. + +`StoredSpan` includes epoch-millisecond `startedMs`/`endedMs`, event offsets in `timeMs`, and links carrying both target `traceId` and `spanId`. Duration uses a monotonic clock; wall clocks across processes may disagree. Merge spans from several stores by trace ID; parent relationships, not timestamp containment, define the tree. + +Rendering is a separate subpath, kept out of the root entry graph: + +```typescript +import { renderTrace } from '@earendil-works/pi-telemetry/render'; + +const page = await telemetry.findTraces({ errors: true, limit: 10 }); +for (const trace of page.traces) console.log(renderTrace(trace.spans)); +``` + +`renderTrace()` accepts stored spans, deduplicates IDs, treats absent parents as roots, and prints attributes, durations, statuses, event offsets, and cross-trace links. It never follows links or queries a store. + ## Adapter Conformance `@earendil-works/pi-telemetry/testing` exports a runner-independent conformance suite modeled as grouped cases. A fixture supplies a fresh context and converts its backend's finished spans into normalized `RecordedTelemetrySpan` snapshots: @@ -313,7 +338,7 @@ for (const group of new Set(conformance.map((testCase) => testCase.group))) { } ``` -The suite checks capability identity on roots, child spans, extracted parents and inert spans, synchronous single admission, result and rejection identity, automatic and explicit status, attribute merging, event ordering, inert post-settlement calls, nested and concurrent parentage, links, and suppression of unreadable telemetry payload failures. Fixtures may supply `propagation` to additionally check JSON round trips, multiple adapter hops, and safe fallback for malformed carriers. Fixtures may also supply `metrics` and `getMetrics()` to check ordered measurements, histogram exemplars, input copying, and atomic suppression of invalid metadata or measurements. `getSpans()` may flush an asynchronous exporter before returning. The testing subpath uses Node's assertion APIs; the root telemetry package remains runtime-neutral. +The suite checks capability identity on roots, child spans, extracted parents and inert spans, synchronous single admission, result and rejection identity, automatic and explicit status, attribute merging, event ordering, inert post-settlement calls, nested and concurrent parentage, and links. Fixtures may supply `propagation` to additionally check JSON round trips, multiple adapter hops, and root fallback for absent or unknown carriers. Fixtures may also supply `metrics` and `getMetrics()` to check ordered measurements, histogram exemplars, and input copying. It does not test malformed inputs or require suppression of adapter/backend failures. `getSpans()` may flush an asynchronous exporter before returning. The testing subpath uses Node's assertion APIs; the root telemetry package remains runtime-neutral. ## Typed Schemas @@ -487,7 +512,7 @@ The AI and Durable schemas use pi-owned `pi.ai.*` and `pi.durable.*` names. Adap ## Security and Portability -Telemetry is diagnostics, not durable application state. Do not persist a `TelemetryContext`, `TelemetrySpan`, or backend-native trace object in records, messages, snapshots, or deferred handles. An injected `TelemetryCarrier` may cross a transport or be stored as optional diagnostic correlation data. It is best effort: an adapter change may lose correlation, but must not prevent work. +Telemetry is diagnostics, not durable application state. Do not persist a `TelemetryContext`, `TelemetrySpan`, or backend-native trace object in records, messages, snapshots, or deferred handles. An injected `TelemetryCarrier` may cross a transport or be stored as optional diagnostic correlation data. Changing adapters may lose correlation; carriers are not application state. Attribute values are intentionally limited to primitive scalars and arrays. Domain instrumentation should avoid prompts, completions, tool arguments or output, file contents, provider payloads, headers, credentials, and free-form error details unless its schema and data policy explicitly allow them. @@ -519,6 +544,10 @@ The package does not use `AsyncLocalStorage` or another runtime-specific ambient | `RecordedTelemetrySpan` | Normalized captured span snapshot | | `RecordedTelemetryEvent` | Normalized captured event snapshot | | `RecordedTelemetryLink` | Captured target carrier and link attributes | +| `TelemetryStore` | Settled trace discovery, trace retrieval, and metric queries | +| `StoredSpan` / `StoredTrace` | Backend-independent timed span and trace snapshots | +| `TraceQuery` / `MetricQuery` | Trace filters, pagination, and metric filters | +| `StoredMetric` | Metric interval row with count, sum, minimum, and maximum | ### Schema definitions and inference diff --git a/packages/telemetry/package.json b/packages/telemetry/package.json index b12cfe1f2..1fb122650 100644 --- a/packages/telemetry/package.json +++ b/packages/telemetry/package.json @@ -10,6 +10,11 @@ "types": "./dist/index.d.ts", "import": "./dist/index.js" }, + "./render": { + "source": "./src/render.ts", + "types": "./dist/render.d.ts", + "import": "./dist/render.js" + }, "./testing": { "types": "./dist/testing/index.d.ts", "import": "./dist/testing/index.js" diff --git a/packages/telemetry/src/index.ts b/packages/telemetry/src/index.ts index fbea18135..6432d095a 100644 --- a/packages/telemetry/src/index.ts +++ b/packages/telemetry/src/index.ts @@ -12,6 +12,7 @@ export type TelemetryCarrier = Record; /** Optional transport capability implemented by the application's telemetry adapter. */ export interface TelemetryPropagation { inject(context: TelemetryContext): TelemetryCarrier | undefined; + /** Absent or unknown carriers use the root; recognized identifiers are opaque strings. */ extract(carrier: TelemetryCarrier | undefined): TelemetryContext; } @@ -28,6 +29,72 @@ export interface SpanOptions { export type SpanStatus = { status: "ok" } | { status: "error"; error?: { name: string; message: string } }; +/** A settled span snapshot, independent of the recording adapter. Event times are relative to span start. */ +export interface StoredSpan { + readonly id: string; + readonly traceId: string; + readonly parentId?: string; + readonly name: string; + readonly startedMs: number; + readonly endedMs: number; + readonly attributes: SpanAttributes; + readonly events: readonly { readonly name: string; readonly timeMs: number; readonly attributes: SpanAttributes }[]; + readonly links: readonly { + readonly traceId: string; + readonly spanId: string; + readonly attributes?: SpanAttributes; + }[]; + readonly status: SpanStatus; +} + +export interface StoredTrace { + readonly traceId: string; + readonly spans: readonly StoredSpan[]; +} + +export interface TraceQuery { + /** Name of the earliest stored span, with span ID breaking timestamp ties. */ + readonly name?: string; + /** All equalities must match one span; that span may be anywhere in the trace. */ + readonly attributes?: Readonly>; + readonly errors?: true; + /** Inclusive bounds on the earliest stored span's start time. */ + readonly startMin?: number; + readonly startMax?: number; + /** Positive page size; defaults to 20. */ + readonly limit?: number; + /** Opaque continuation from the previous page; retain the same filters. */ + readonly cursor?: string; +} + +export interface MetricQuery { + readonly name: string; + /** Inclusive bounds on interval start time. */ + readonly startMin?: number; + readonly startMax?: number; + readonly attributes?: Readonly>; +} + +export interface StoredMetric { + readonly intervalMs: number; + readonly name: string; + readonly kind: "counter" | "upDownCounter" | "histogram"; + readonly unit?: string; + readonly attributes: SpanAttributes; + readonly count: number; + readonly sum: number; + readonly min: number; + readonly max: number; +} + +/** Read settled traces without depending on a storage backend. */ +export interface TelemetryStore { + findTraces(query?: TraceQuery): Promise<{ traces: readonly StoredTrace[]; cursor?: string }>; + getTrace(traceId: string): Promise; + findMetrics(query: MetricQuery): Promise; +} + +/** Adapter contract for trusted typed inputs; malformed input behavior is unspecified. */ export interface TelemetryContext { /** Adapter capabilities remain available on child spans and extracted parents. */ readonly metrics?: TelemetryMetrics; diff --git a/packages/telemetry/src/memory.ts b/packages/telemetry/src/memory.ts index dd9934297..b59cb1639 100644 --- a/packages/telemetry/src/memory.ts +++ b/packages/telemetry/src/memory.ts @@ -1,13 +1,18 @@ import type { - AttributeValue, + MetricQuery, SpanAttributes, SpanLink, SpanOptions, SpanStatus, + StoredMetric, + StoredSpan, + StoredTrace, TelemetryCarrier, TelemetryContext, TelemetryPropagation, TelemetrySpan, + TelemetryStore, + TraceQuery, } from "./index.ts"; import type { HistogramOptions, @@ -56,11 +61,15 @@ export interface RecordedTelemetryMeasurement { } interface MutableRecordedTelemetryEvent { + timeMs: number; name: string; attributes: SpanAttributes; } interface MutableRecordedTelemetrySpan { + startedMs: number; + clock: number; + endedMs?: number; id: number; parentId: number | null; context: TelemetryCarrier; @@ -81,6 +90,10 @@ interface InMemoryParent { span?: MutableRecordedTelemetrySpan; } +function carrierId(carrier: Readonly): string { + return `${carrier["pi-memory-scope"]}:${carrier["pi-memory-span"]}`; +} + // Identity survives settlement without exposing backend IDs on the generic context contract. const contextCarriers = new WeakMap(); @@ -89,40 +102,22 @@ interface InMemoryTelemetryState { propagation: TelemetryPropagation; scope: string; spans: MutableRecordedTelemetrySpan[]; - measurements: RecordedTelemetryMeasurement[]; + measurements: { intervalMs: number; measurement: RecordedTelemetryMeasurement }[]; nextSpanId: number; nextEndSequence: number; } -function copyAttributeValue(value: AttributeValue): AttributeValue { - return Array.isArray(value) ? ([...value] as AttributeValue) : value; -} - function copyAttributes(attributes?: SpanAttributes): SpanAttributes { - const copy: SpanAttributes = {}; - if (!attributes) return copy; - for (const [name, value] of Object.entries(attributes)) { - if (value !== undefined) copy[name] = copyAttributeValue(value); - } - return copy; -} - -function mergeAttributes(current: SpanAttributes, attributes: SpanAttributes): SpanAttributes { - const merged = copyAttributes(current); - for (const [name, value] of Object.entries(attributes)) { - if (value !== undefined) merged[name] = copyAttributeValue(value); - } - return merged; + return Object.fromEntries( + Object.entries(attributes ?? {}) + .filter(([, value]) => value !== undefined) + .map(([name, value]) => [name, Array.isArray(value) ? [...value] : value]), + ); } function copyLink(link: SpanLink): RecordedTelemetryLink | undefined { - try { - const context = contextCarriers.get(link.context); - if (!context) return undefined; - return { context: { ...context }, attributes: copyAttributes(link.attributes) }; - } catch { - return undefined; - } + const context = contextCarriers.get(link.context); + return context ? { context: { ...context }, attributes: copyAttributes(link.attributes) } : undefined; } function copyStatus(status: SpanStatus): SpanStatus { @@ -132,29 +127,6 @@ function copyStatus(status: SpanStatus): SpanStatus { : { status: "error" }; } -function automaticErrorStatus(error: unknown): SpanStatus { - try { - if (error instanceof Error) { - return { status: "error", error: { name: error.name, message: error.message } }; - } - } catch { - // Error inspection is passive. Fall through to an error status without details. - } - return { status: "error" }; -} - -function settleSpan( - state: InMemoryTelemetryState, - span: MutableRecordedTelemetrySpan, - failed: boolean, - error?: unknown, -): void { - if (span.settled) return; - if (failed && !span.explicitStatus) span.status = automaticErrorStatus(error); - span.settled = true; - span.endSequence = state.nextEndSequence++; -} - function createSpan( state: InMemoryTelemetryState, parent: InMemoryParent | undefined, @@ -163,19 +135,21 @@ function createSpan( const name = options.name; const attributes = copyAttributes(options.attributes); const links: RecordedTelemetryLink[] = []; - try { - for (const link of options.links ?? []) { - const copy = copyLink(link); - if (copy) links.push(copy); - } - } catch { - // An unreadable link list must not suppress the operation's span. + for (const link of options.links ?? []) { + const copy = copyLink(link); + if (copy) links.push(copy); } const id = state.nextSpanId++; return { + startedMs: Date.now(), + clock: performance.now(), id, parentId: parent?.localId ?? null, - context: { "pi-memory-scope": state.scope, "pi-memory-span": String(id) }, + context: { + "pi-memory-scope": state.scope, + "pi-memory-span": String(id), + "pi-memory-trace": parent?.context["pi-memory-trace"] ?? `${state.scope}:${id}`, + }, parentContext: parent ? { ...parent.context } : null, links, name, @@ -205,7 +179,7 @@ function startInertSpan( ); } -function startInMemorySpan( +async function startInMemorySpan( state: InMemoryTelemetryState, parent: InMemoryParent | undefined, options: SpanOptions, @@ -213,13 +187,8 @@ function startInMemorySpan( ): Promise { if (parent?.span?.settled) return startInertSpan(state, options, callback); - let recordedSpan: MutableRecordedTelemetrySpan; - try { - recordedSpan = createSpan(state, parent, options); - state.spans.push(recordedSpan); - } catch { - return startInertSpan(state, options, callback); - } + const recordedSpan = createSpan(state, parent, options); + state.spans.push(recordedSpan); const span: TelemetrySpan = { metrics: state.metrics, @@ -241,51 +210,39 @@ function startInMemorySpan( }, addEvent(name, attributes) { if (recordedSpan.settled) return; - try { - recordedSpan.events.push({ name, attributes: copyAttributes(attributes) }); - } catch { - // Recording is passive. Ignore malformed or unreadable telemetry payloads. - } + recordedSpan.events.push({ + name, + timeMs: performance.now() - recordedSpan.clock, + attributes: copyAttributes(attributes), + }); }, setAttributes(attributes) { if (recordedSpan.settled) return; - try { - recordedSpan.attributes = mergeAttributes(recordedSpan.attributes, attributes); - } catch { - // Recording is passive. Ignore malformed or unreadable telemetry payloads. - } + recordedSpan.attributes = { ...recordedSpan.attributes, ...copyAttributes(attributes) }; }, setStatus(status) { if (recordedSpan.settled) return; - try { - recordedSpan.status = copyStatus(status); - recordedSpan.explicitStatus = true; - } catch { - // Recording is passive. Ignore malformed or unreadable telemetry payloads. - } + recordedSpan.status = copyStatus(status); + recordedSpan.explicitStatus = true; }, }; contextCarriers.set(span, recordedSpan.context); - let result: T | Promise; try { - result = callback(span); + return await callback(span); } catch (error) { - settleSpan(state, recordedSpan, true, error); - return Promise.reject(error); + if (!recordedSpan.explicitStatus) + recordedSpan.status = + error instanceof Error + ? { status: "error", error: { name: error.name, message: error.message } } + : { status: "error" }; + throw error; + } finally { + recordedSpan.settled = true; + recordedSpan.endedMs = recordedSpan.startedMs + performance.now() - recordedSpan.clock; + recordedSpan.endSequence = state.nextEndSequence++; } - - return Promise.resolve(result).then( - (value) => { - settleSpan(state, recordedSpan, false); - return value; - }, - (error: unknown) => { - settleSpan(state, recordedSpan, true, error); - throw error; - }, - ); } function createMeasurementRecorder( @@ -294,50 +251,37 @@ function createMeasurementRecorder( name: string, options?: HistogramOptions, ): (value: number, attributes?: SpanAttributes, context?: TelemetryContext) => void { - try { - const description = options?.description; - const unit = options?.unit; - const boundaries = - kind === "histogram" && options?.boundaries !== undefined ? [...options.boundaries] : undefined; - if ( - typeof name !== "string" || - (description !== undefined && typeof description !== "string") || - (unit !== undefined && typeof unit !== "string") - ) - return () => {}; - if (boundaries?.some((value, index) => !Number.isFinite(value) || (index > 0 && value <= boundaries[index - 1]!))) - return () => {}; - const metadata = { - kind, - name, - ...(description === undefined ? {} : { description }), - ...(unit === undefined ? {} : { unit }), - ...(boundaries === undefined ? {} : { boundaries }), - }; - return (value, attributes, context) => { - if (!Number.isFinite(value) || (kind === "counter" && value < 0)) return; - try { - const carrier = context === undefined ? undefined : contextCarriers.get(context); - state.measurements.push({ - ...metadata, - value, - attributes: copyAttributes(attributes), - context: carrier ? { ...carrier } : null, - }); - } catch { - // Measurements are passive; discard the whole failed call. - } - }; - } catch { - return () => {}; - } + const description = options?.description; + const unit = options?.unit; + const boundaries = kind === "histogram" && options?.boundaries !== undefined ? [...options.boundaries] : undefined; + const metadata = { + kind, + name, + ...(description === undefined ? {} : { description }), + ...(unit === undefined ? {} : { unit }), + ...(boundaries === undefined ? {} : { boundaries }), + }; + return (value, attributes, context) => { + const carrier = context === undefined ? undefined : contextCarriers.get(context); + state.measurements.push({ + intervalMs: Math.floor(Date.now() / 10_000) * 10_000, + measurement: { + ...metadata, + value, + attributes: copyAttributes(attributes), + context: carrier ? { ...carrier } : null, + }, + }); + }; } /** * Backend-neutral reference implementation that records spans in process memory. * Create a fresh instance to isolate tests or independent recording scopes. */ -export class InMemoryTelemetryContext implements TelemetryContext, TelemetryPropagation, TelemetryMetrics { +export class InMemoryTelemetryContext + implements TelemetryContext, TelemetryPropagation, TelemetryMetrics, TelemetryStore +{ readonly metrics: TelemetryMetrics = this; readonly propagation: TelemetryPropagation = this; private readonly state: InMemoryTelemetryState = { @@ -360,34 +304,22 @@ export class InMemoryTelemetryContext implements TelemetryContext, TelemetryProp return carrier ? { ...carrier } : undefined; } - /** Unrecognized or unreadable carriers fall back to this recorder's root context. */ + /** Missing or non-string carrier identities fall back to this recorder's root context. */ extract(carrier: TelemetryCarrier | undefined): TelemetryContext { - try { - if (!carrier || typeof carrier !== "object" || Array.isArray(carrier)) return this; - if (!Object.hasOwn(carrier, "pi-memory-scope") || !Object.hasOwn(carrier, "pi-memory-span")) return this; - if (Object.values(carrier).some((value) => typeof value !== "string")) return this; - const scope = carrier["pi-memory-scope"]; - const span = carrier["pi-memory-span"]; - if (typeof scope !== "string" || !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/.test(scope)) - return this; - if (typeof span !== "string" || !/^[1-9][0-9]*$/.test(span) || !Number.isSafeInteger(Number(span))) - return this; - const context = { "pi-memory-scope": scope, "pi-memory-span": span }; - const localId = - scope === this.state.scope && this.state.spans.some((candidate) => candidate.id === Number(span)) - ? Number(span) - : null; - const parent: TelemetryContext = { - metrics: this, - propagation: this, - startSpan: (options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise) => - startInMemorySpan(this.state, { context, localId }, options, callback), - }; - contextCarriers.set(parent, context); - return parent; - } catch { - return this; - } + const scope = carrier?.["pi-memory-scope"]; + const span = carrier?.["pi-memory-span"]; + const traceId = carrier?.["pi-memory-trace"]; + if (typeof scope !== "string" || typeof span !== "string" || typeof traceId !== "string") return this; + const context = { "pi-memory-scope": scope, "pi-memory-span": span, "pi-memory-trace": traceId }; + const localId = scope === this.state.scope ? Number(span) : null; + const parent: TelemetryContext = { + metrics: this, + propagation: this, + startSpan: (options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise) => + startInMemorySpan(this.state, { context, localId }, options, callback), + }; + contextCarriers.set(parent, context); + return parent; } counter(name: string, options?: InstrumentOptions): TelemetryCounter { @@ -402,9 +334,103 @@ export class InMemoryTelemetryContext implements TelemetryContext, TelemetryProp return { record: createMeasurementRecorder(this.state, "histogram", name, options) }; } + private storedSpans(): StoredSpan[] { + return this.state.spans + .filter((span) => span.settled) + .map((span) => ({ + id: carrierId(span.context), + traceId: span.context["pi-memory-trace"]!, + ...(span.parentContext ? { parentId: carrierId(span.parentContext) } : {}), + name: span.name, + startedMs: span.startedMs, + endedMs: span.endedMs!, + attributes: copyAttributes(span.attributes), + events: span.events.map((event) => ({ + name: event.name, + timeMs: event.timeMs, + attributes: copyAttributes(event.attributes), + })), + links: span.links.map((link) => ({ + traceId: link.context["pi-memory-trace"]!, + spanId: carrierId(link.context), + attributes: copyAttributes(link.attributes), + })), + status: copyStatus(span.status), + })) + .sort((a, b) => a.startedMs - b.startedMs || (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)); + } + + async getTrace(traceId: string): Promise { + const spans = this.storedSpans().filter((span) => span.traceId === traceId); + return spans.length ? { traceId, spans } : undefined; + } + + async findTraces(query: TraceQuery = {}): Promise<{ traces: readonly StoredTrace[]; cursor?: string }> { + const limit = query.limit ?? 20; + const after = query.cursor === undefined ? undefined : (JSON.parse(query.cursor) as [number, string]); + const grouped = new Map(); + for (const span of this.storedSpans()) { + let trace = grouped.get(span.traceId); + if (!trace) { + trace = { traceId: span.traceId, spans: [] }; + grouped.set(span.traceId, trace); + } + trace.spans.push(span); + } + const matches = [...grouped.values()] + .filter((trace) => { + const first = trace.spans[0]!; + return ( + (query.name === undefined || first.name === query.name) && + (query.startMin === undefined || first.startedMs >= query.startMin) && + (query.startMax === undefined || first.startedMs <= query.startMax) && + (!after || first.startedMs < after[0] || (first.startedMs === after[0] && trace.traceId > after[1])) && + (!query.errors || trace.spans.some((span) => span.status.status === "error")) && + (!query.attributes || + trace.spans.some((span) => + Object.entries(query.attributes!).every(([key, value]) => span.attributes[key] === value), + )) + ); + }) + .sort( + (a, b) => + b.spans[0]!.startedMs - a.spans[0]!.startedMs || + (a.traceId < b.traceId ? -1 : a.traceId > b.traceId ? 1 : 0), + ); + const traces = matches.slice(0, limit); + const last = traces.at(-1)!; + return { + traces, + ...(matches.length > limit ? { cursor: JSON.stringify([last.spans[0]!.startedMs, last.traceId]) } : {}), + }; + } + + /** Read raw measurements as count-one rows, without changing getMetrics() snapshots. */ + async findMetrics(query: MetricQuery): Promise { + return this.state.measurements + .filter( + ({ intervalMs, measurement }) => + measurement.name === query.name && + (query.startMin === undefined || intervalMs >= query.startMin) && + (query.startMax === undefined || intervalMs <= query.startMax) && + Object.entries(query.attributes ?? {}).every(([key, value]) => measurement.attributes[key] === value), + ) + .map(({ intervalMs, measurement }) => ({ + intervalMs, + name: measurement.name, + kind: measurement.kind, + ...(measurement.unit === undefined ? {} : { unit: measurement.unit }), + attributes: copyAttributes(measurement.attributes), + count: 1, + sum: measurement.value, + min: measurement.value, + max: measurement.value, + })); + } + /** Returns detached measurements in recording order, without aggregation. */ getMetrics(): readonly RecordedTelemetryMeasurement[] { - return this.state.measurements.map((measurement) => ({ + return this.state.measurements.map(({ measurement }) => ({ ...measurement, ...(measurement.boundaries === undefined ? {} : { boundaries: [...measurement.boundaries] }), attributes: copyAttributes(measurement.attributes), diff --git a/packages/telemetry/src/metrics.ts b/packages/telemetry/src/metrics.ts index 428b8a6bd..5d6e4fd79 100644 --- a/packages/telemetry/src/metrics.ts +++ b/packages/telemetry/src/metrics.ts @@ -13,6 +13,7 @@ export interface InstrumentOptions { } export interface HistogramOptions extends InstrumentOptions { + /** Bucket hints; callers supply finite, strictly increasing values. */ boundaries?: readonly number[]; } @@ -28,7 +29,11 @@ export interface TelemetryHistogram { record(value: number, attributes?: SpanAttributes, context?: TelemetryContext): void; } -/** Optional adapter capability. Factories and recording methods are synchronous and non-throwing. */ +/** + * Optional adapter capability with synchronous factories and recording methods. + * Typed inputs are trusted; callers supply finite measurements and nonnegative counter increments. + * Malformed input behavior is unspecified. + */ export interface TelemetryMetrics { counter(name: string, options?: InstrumentOptions): TelemetryCounter; upDownCounter(name: string, options?: InstrumentOptions): TelemetryUpDownCounter; diff --git a/packages/telemetry/src/noop.ts b/packages/telemetry/src/noop.ts index 0e39acb11..51cf07497 100644 --- a/packages/telemetry/src/noop.ts +++ b/packages/telemetry/src/noop.ts @@ -1,12 +1,8 @@ import type { SpanOptions, TelemetryPropagation, TelemetrySpan } from "./index.ts"; import type { TelemetryCounter, TelemetryHistogram, TelemetryMetrics, TelemetryUpDownCounter } from "./metrics.ts"; -function startNoopSpan(_options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { - try { - return Promise.resolve(callback(noopTelemetrySpan)); - } catch (error) { - return Promise.reject(error); - } +async function startNoopSpan(_options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { + return callback(noopTelemetrySpan); } const noopCounter: TelemetryCounter = Object.freeze({ add: () => {} }); diff --git a/packages/telemetry/src/render.ts b/packages/telemetry/src/render.ts new file mode 100644 index 000000000..a3ccb7108 --- /dev/null +++ b/packages/telemetry/src/render.ts @@ -0,0 +1,49 @@ +import type { SpanAttributes, StoredSpan } from "./index.ts"; + +function formatAttributes(attributes: Readonly = {}): string { + return Object.entries(attributes) + .map(([key, value]) => ` ${key}=${JSON.stringify(value)}`) + .join(""); +} + +/** Render stored spans as a chronological forest, treating absent parents as roots. */ +export function renderTrace(spans: readonly StoredSpan[]): string { + const byId = new Map(spans.map((span) => [span.id, span])); + const roots: StoredSpan[] = []; + const children = new Map(); + for (const span of [...byId.values()].sort((left, right) => left.startedMs - right.startedMs)) { + if (span.parentId === undefined || !byId.has(span.parentId)) { + roots.push(span); + } else { + const siblings = children.get(span.parentId); + if (siblings === undefined) children.set(span.parentId, [span]); + else siblings.push(span); + } + } + + const lines: string[] = []; + const pending = roots.reverse().map((span) => ({ span, depth: 0 })); + while (pending.length > 0) { + const { span, depth } = pending.pop()!; + const indent = " ".repeat(depth); + let error = ""; + if (span.status.status === "error") { + error = " error"; + if (span.status.error !== undefined) error += ` ${span.status.error.name}: ${span.status.error.message}`; + } + lines.push( + `${indent}${span.name}${formatAttributes(span.attributes)} (${span.endedMs - span.startedMs}ms)${error}`, + ); + for (const event of span.events) { + lines.push(`${indent} · ${event.name}${formatAttributes(event.attributes)} (+${event.timeMs}ms)`); + } + for (const link of span.links) { + lines.push(`${indent} → link trace ${link.traceId} span ${link.spanId}${formatAttributes(link.attributes)}`); + } + const descendants = children.get(span.id) ?? []; + for (let index = descendants.length - 1; index >= 0; index--) { + pending.push({ span: descendants[index]!, depth: depth + 1 }); + } + } + return lines.join("\n"); +} diff --git a/packages/telemetry/src/testing/conformance.ts b/packages/telemetry/src/testing/conformance.ts index a8b16e46a..71630c0c4 100644 --- a/packages/telemetry/src/testing/conformance.ts +++ b/packages/telemetry/src/testing/conformance.ts @@ -1,14 +1,5 @@ -import { deepStrictEqual, doesNotThrow, fail, ok, strictEqual } from "node:assert/strict"; -import { - NOOP_TELEMETRY_CONTEXT, - type SpanAttributes, - type SpanLink, - type SpanOptions, - type SpanStatus, - type TelemetryCarrier, - type TelemetryContext, - type TelemetrySpan, -} from "../index.ts"; +import { deepStrictEqual, fail, ok, strictEqual } from "node:assert/strict"; +import { NOOP_TELEMETRY_CONTEXT, type TelemetryCarrier, type TelemetryContext, type TelemetrySpan } from "../index.ts"; import type { RecordedTelemetrySpan } from "../memory.ts"; import type { TelemetryAdapterConformanceCase, @@ -49,23 +40,6 @@ async function rejectsWithSameValue(operation: Promise, expected: unkno } } -function unreadable(value: T): T { - return new Proxy(value, { - get: () => { - throw new Error("read"); - }, - getOwnPropertyDescriptor: () => { - throw new Error("inspect"); - }, - getPrototypeOf: () => { - throw new Error("prototype"); - }, - ownKeys: () => { - throw new Error("enumerate"); - }, - }); -} - /** Creates runner-independent cases for the callback telemetry adapter contract. */ export function createTelemetryAdapterConformance( factory: TelemetryAdapterFixtureFactory, @@ -132,34 +106,24 @@ export function createTelemetryAdapterConformance( deepStrictEqual(measurements[3]?.context, findSpan(await fixture.getSpans(), "measured").context); }, ), - createCase( - factory, - "metrics", - "copies inputs and suppresses invalid measurements atomically", - async (fixture) => { - if (!fixture.metrics) return; - ok(fixture.getMetrics, "Metrics fixtures require a normalized measurement reader"); - const boundaries = [1, 10]; - const options = { description: "Original", boundaries }; - const histogram = fixture.metrics.histogram("snapshot", options); - boundaries.push(100); - options.description = "Changed"; - const attributes = { tags: ["original"] }; - histogram.record(-1, attributes); - attributes.tags.push("changed"); - const counter = fixture.metrics.counter("invalid"); - for (const value of [NaN, Infinity, -Infinity, -1]) doesNotThrow(() => counter.add(value)); - doesNotThrow(() => counter.add(1, { partial: true, broken: unreadable(["value"]) })); - doesNotThrow(() => fixture.metrics?.histogram("bad-options", unreadable({ unit: "ms" })).record(1)); - doesNotThrow(() => fixture.metrics?.histogram("bad-boundaries", { boundaries: [2, 1] }).record(1)); - const measurements = await fixture.getMetrics(); - strictEqual(measurements.length, 1); - deepStrictEqual(measurements[0]?.attributes, { tags: ["original"] }); - deepStrictEqual(measurements[0]?.boundaries, [1, 10]); - strictEqual(measurements[0]?.description, "Original"); - strictEqual(measurements[0]?.context, null); - }, - ), + createCase(factory, "metrics", "copies instrument metadata and measurement attributes", async (fixture) => { + if (!fixture.metrics) return; + ok(fixture.getMetrics, "Metrics fixtures require a normalized measurement reader"); + const boundaries = [1, 10]; + const options = { description: "Original", boundaries }; + const histogram = fixture.metrics.histogram("snapshot", options); + boundaries.push(100); + options.description = "Changed"; + const attributes = { tags: ["original"] }; + histogram.record(-1, attributes); + attributes.tags.push("changed"); + const measurements = await fixture.getMetrics(); + strictEqual(measurements.length, 1); + deepStrictEqual(measurements[0]?.attributes, { tags: ["original"] }); + deepStrictEqual(measurements[0]?.boundaries, [1, 10]); + strictEqual(measurements[0]?.description, "Original"); + strictEqual(measurements[0]?.context, null); + }), createCase(factory, "links", "records initial and late links without changing parents", async (fixture) => { let target: TelemetrySpan | undefined; await fixture.context.startSpan({ name: "target" }, (span) => { @@ -182,25 +146,16 @@ export function createTelemetryAdapterConformance( { context: findSpan(spans, "target").context, attributes: { reason: "late" } }, ]); }), - createCase( - factory, - "links", - "suppresses malformed links atomically and ignores settled sources", - async (fixture) => { - let source: TelemetrySpan | undefined; - await fixture.context.startSpan({ name: "link-source" }, (span) => { - source = span; - doesNotThrow(() => span.addLink(unreadable({ context: span }))); - doesNotThrow(() => - span.addLink({ context: span, attributes: { partial: true, broken: unreadable(["value"]) } }), - ); - }); - if (!source) throw new Error("Expected source span"); - const sourceSpan = source; - doesNotThrow(() => sourceSpan.addLink({ context: sourceSpan })); - deepStrictEqual(findSpan(await fixture.getSpans(), "link-source").links, []); - }, - ), + createCase(factory, "links", "ignores links added after source settlement", async (fixture) => { + let source: TelemetrySpan | undefined; + await fixture.context.startSpan({ name: "link-source" }, (span) => { + source = span; + }); + if (!source) throw new Error("Expected source span"); + source.addLink({ context: source }); + deepStrictEqual(findSpan(await fixture.getSpans(), "link-source").links, []); + }), + createCase( factory, "propagation", @@ -221,27 +176,17 @@ export function createTelemetryAdapterConformance( deepStrictEqual(findSpan(spans, "receiver").parentContext, findSpan(spans, "sender").context); }, ), - createCase(factory, "propagation", "unknown and unreadable carriers produce recorded roots", async (fixture) => { + createCase(factory, "propagation", "absent and unknown carriers produce recorded roots", async (fixture) => { if (!fixture.propagation) return; strictEqual(fixture.propagation.inject(NOOP_TELEMETRY_CONTEXT), undefined); - for (const [index, carrier] of [ - undefined, - {}, - { unknown: "format" }, - unreadable({ bad: "value" }), - ].entries()) { - let result = 0; - doesNotThrow(() => { - const context = fixture.propagation?.extract(carrier); - if (!context) throw new Error("Expected root context"); - void context.startSpan({ name: `fallback-${index}` }, () => { - result = 42; - }); - }); - strictEqual(result, 42); + const carriers: readonly (TelemetryCarrier | undefined)[] = [undefined, {}, { unknown: "format" }]; + for (const [index, carrier] of carriers.entries()) { + const context = fixture.propagation.extract(carrier); + strictEqual(await context.startSpan({ name: `fallback-${index}` }, () => 42), 42); } for (const span of await fixture.getSpans()) strictEqual(span.parentContext, null); }), + createCase(factory, "propagation", "propagates across fresh adapters and multiple hops", async (fixture) => { if (!fixture.propagation) return; await using receiver = await factory(); @@ -313,30 +258,8 @@ export function createTelemetryAdapterConformance( undefined, ); - const unreadableError = unreadable({ kind: "unreadable" }); - await rejectsWithSameValue( - fixture.context.startSpan({ name: "unreadable-error" }, () => { - throw unreadableError; - }), - unreadableError, - ); - - const asyncUnreadableError = unreadable({ kind: "async-unreadable" }); - await rejectsWithSameValue( - fixture.context.startSpan({ name: "async-unreadable-error" }, () => - Promise.reject(asyncUnreadableError), - ), - asyncUnreadableError, - ); - const spans = await fixture.getSpans(); - for (const name of [ - "sync-error", - "async-error", - "undefined-error", - "unreadable-error", - "async-unreadable-error", - ]) { + for (const name of ["sync-error", "async-error", "undefined-error"]) { strictEqual(findSpan(spans, name).status.status, "error"); } }, @@ -406,20 +329,6 @@ export function createTelemetryAdapterConformance( ]); }), - createCase(factory, "recording", "ignores failed attribute calls atomically", async (fixture) => { - await fixture.context.startSpan({ name: "atomic-attributes", attributes: { retained: "value" } }, (span) => { - const attributes: SpanAttributes = { - partial: "must not survive", - unreadable: unreadable(["value"]), - }; - doesNotThrow(() => span.setAttributes(attributes)); - }); - - deepStrictEqual(findSpan(await fixture.getSpans(), "atomic-attributes").attributes, { - retained: "value", - }); - }), - createCase(factory, "recording", "makes calls after settlement inert", async (fixture) => { let settledSpan: TelemetrySpan | undefined; await fixture.context.startSpan({ name: "settled", attributes: { value: "initial" } }, (span) => { @@ -428,9 +337,9 @@ export function createTelemetryAdapterConformance( const capturedSpan = settledSpan; if (!capturedSpan) throw new Error("Expected callback span"); - doesNotThrow(() => capturedSpan.setAttributes({ value: "late" })); - doesNotThrow(() => capturedSpan.addEvent("late", { value: true })); - doesNotThrow(() => capturedSpan.setStatus({ status: "error" })); + capturedSpan.setAttributes({ value: "late" }); + capturedSpan.addEvent("late", { value: true }); + capturedSpan.setStatus({ status: "error" }); let childAdmitted = false; const childResult = capturedSpan.startSpan({ name: "late-child" }, () => { childAdmitted = true; @@ -473,46 +382,5 @@ export function createTelemetryAdapterConformance( ok(second.endSequence < first.endSequence); ok(first.endSequence < parent.endSequence); }), - - createCase(factory, "passivity", "suppresses unreadable telemetry payload failures", async (fixture) => { - let calls = 0; - const options = unreadable({ name: "unreadable-options", attributes: { secret: "value" } }); - const result = fixture.context.startSpan(options, () => { - calls++; - return 9; - }); - - strictEqual(calls, 1); - strictEqual(await result, 9); - deepStrictEqual(await fixture.getSpans(), []); - - await fixture.context.startSpan({ name: "unreadable-recording" }, (span) => { - const attributes = unreadable({ secret: "value" }); - const status = unreadable({ status: "ok" }); - doesNotThrow(() => span.setAttributes(attributes)); - doesNotThrow(() => span.addEvent("unreadable-event", attributes)); - doesNotThrow(() => span.setStatus(status)); - }); - - const recorded = await fixture.getSpans(); - strictEqual(recorded.length, 1); - deepStrictEqual(recorded[0]?.attributes, {}); - deepStrictEqual(recorded[0]?.events, []); - deepStrictEqual(recorded[0]?.status, { status: "ok" }); - }), - - createCase(factory, "passivity", "ignores failed status calls atomically", async (fixture) => { - const rejection = new Error("rejected after unreadable status"); - await rejectsWithSameValue( - fixture.context.startSpan({ name: "unreadable-status" }, (span) => { - const status = unreadable({ status: "ok" }); - doesNotThrow(() => span.setStatus(status)); - return Promise.reject(rejection); - }), - rejection, - ); - - strictEqual(findSpan(await fixture.getSpans(), "unreadable-status").status.status, "error"); - }), ]; } diff --git a/packages/telemetry/test/metrics.test.ts b/packages/telemetry/test/metrics.test.ts index 11e691f17..d21d7bdb3 100644 --- a/packages/telemetry/test/metrics.test.ts +++ b/packages/telemetry/test/metrics.test.ts @@ -2,9 +2,7 @@ import { describe, expect, expectTypeOf, it } from "vitest"; import { createTypedMetrics, defineTelemetrySchema, - type HistogramOptions, InMemoryTelemetryContext, - type InstrumentOptions, NOOP_TELEMETRY_CONTEXT, NOOP_TELEMETRY_METRICS, type SpanAttributes, @@ -12,17 +10,6 @@ import { type TelemetryMetricDefinition, } from "../src/index.ts"; -function unreadable(value: T): T { - return new Proxy(value, { - get() { - throw new Error("read"); - }, - ownKeys() { - throw new Error("enumerate"); - }, - }); -} - const traffic = defineTelemetrySchema({ version: 1, spans: {}, @@ -161,27 +148,9 @@ describe("metrics", () => { }); }); - it("ignores malformed metadata, nonfinite values, and unreadable attributes", () => { - const telemetry = new InMemoryTelemetryContext(); - telemetry.counter("bad", unreadable({ unit: "By" })).add(1); - telemetry.histogram("bad", unreadable({ boundaries: [1] })).record(1); - for (const boundaries of [[1, 1], [2, 1], [NaN], [Infinity]]) - telemetry.histogram("bad", { boundaries }).record(1); - for (const value of [NaN, Infinity, -Infinity]) { - telemetry.counter("bad").add(value); - telemetry.upDownCounter("bad").add(value); - telemetry.histogram("bad").record(value); - } - telemetry.counter("bad").add(-1); - telemetry.histogram("bad").record(1, unreadable({ tag: "value" })); - expect(telemetry.getMetrics()).toEqual([]); - telemetry.histogram("valid").record(-1, {}, NOOP_TELEMETRY_CONTEXT); - expect(telemetry.getMetrics()[0]?.context).toBeNull(); - }); - - it("shares frozen no-op instruments without inspecting arguments", () => { - const options = unreadable({ unit: "By" }); - const attributes = unreadable({ secret: "value" }); + it("shares frozen no-op instruments", () => { + const options = { unit: "By" }; + const attributes = { direction: "out" }; const counter = NOOP_TELEMETRY_METRICS.counter("bytes", options); const active = NOOP_TELEMETRY_METRICS.upDownCounter("active", options); const histogram = NOOP_TELEMETRY_METRICS.histogram("duration", options); @@ -189,8 +158,8 @@ describe("metrics", () => { expect(Object.isFrozen(counter)).toBe(true); expect(Object.isFrozen(active)).toBe(true); expect(Object.isFrozen(histogram)).toBe(true); - expect(() => counter.add(1, attributes)).not.toThrow(); - expect(() => active.add(-1, attributes)).not.toThrow(); - expect(() => histogram.record(1, attributes, unreadable(NOOP_TELEMETRY_CONTEXT))).not.toThrow(); + counter.add(1, attributes); + active.add(-1, attributes); + histogram.record(1, attributes, NOOP_TELEMETRY_CONTEXT); }); }); diff --git a/packages/telemetry/test/propagation.test.ts b/packages/telemetry/test/propagation.test.ts index 4707a4a2f..fb7c6006e 100644 --- a/packages/telemetry/test/propagation.test.ts +++ b/packages/telemetry/test/propagation.test.ts @@ -11,14 +11,6 @@ import { type TelemetrySpan, } from "../src/index.ts"; -function unreadable(value: T): T { - return new Proxy(value, { - get() { - throw new Error("unreadable"); - }, - }); -} - describe("propagation and links", () => { it("continues a parent in another Node process without shared recorder state", async () => { const telemetry = new InMemoryTelemetryContext(); @@ -121,15 +113,27 @@ describe("propagation and links", () => { ]); }); - it("rejects malformed memory carriers and never inspects no-op payloads", async () => { + it("uses the root for absent and unknown carriers", async () => { const telemetry = new InMemoryTelemetryContext(); - for (const carrier of [{ "pi-memory-scope": "bad", "pi-memory-span": "1" }, unreadable({ bad: "value" })]) { + const carriers: readonly (TelemetryCarrier | undefined)[] = [undefined, {}, { unknown: "format" }]; + for (const carrier of carriers) { expect(telemetry.extract(carrier)).toBe(telemetry); } - expect(NOOP_TELEMETRY_PROPAGATION.inject(unreadable(NOOP_TELEMETRY_CONTEXT))).toBeUndefined(); - expect(NOOP_TELEMETRY_PROPAGATION.extract(unreadable({ bad: "value" }))).toBe(NOOP_TELEMETRY_CONTEXT); + expect(NOOP_TELEMETRY_PROPAGATION.inject(NOOP_TELEMETRY_CONTEXT)).toBeUndefined(); + expect(NOOP_TELEMETRY_PROPAGATION.extract({ unknown: "format" })).toBe(NOOP_TELEMETRY_CONTEXT); await NOOP_TELEMETRY_CONTEXT.startSpan({ name: "noop" }, (span) => { - expect(() => span.addLink(unreadable({ context: span }))).not.toThrow(); + span.addLink({ context: span }); }); }); + + it("treats recognized carrier identifiers as opaque strings", async () => { + const telemetry = new InMemoryTelemetryContext(); + const carrier = { + "pi-memory-scope": "external-recorder", + "pi-memory-span": "external-span", + "pi-memory-trace": "external-trace", + }; + await telemetry.extract(carrier).startSpan({ name: "receiver" }, () => {}); + expect(telemetry.getSpans()[0]?.parentContext).toEqual(carrier); + }); }); diff --git a/packages/telemetry/test/render.test.ts b/packages/telemetry/test/render.test.ts new file mode 100644 index 000000000..720d0f985 --- /dev/null +++ b/packages/telemetry/test/render.test.ts @@ -0,0 +1,147 @@ +import { describe, expect, it } from "vitest"; +import type { StoredSpan } from "../src/index.ts"; +import { renderTrace } from "../src/render.ts"; + +function span(id: string, overrides: Partial = {}): StoredSpan { + return { + id, + traceId: "trace", + name: id, + startedMs: 0, + endedMs: 10, + attributes: {}, + events: [], + links: [], + status: { status: "ok" }, + ...overrides, + }; +} + +describe("renderTrace", () => { + it("renders an unordered trace with attributes, durations, errors, events, and links", () => { + const root = span("root", { + startedMs: 1000, + endedMs: 1020, + attributes: { + request: "a b", + attempt: 2, + cached: false, + tags: ["a", "b"], + sizes: [1, 2], + flags: [true, false], + }, + events: [ + { name: "checkpoint", timeMs: 2, attributes: { phase: "tools" } }, + { name: "tick", timeMs: 3.5, attributes: {} }, + ], + links: [ + { traceId: "remote-trace", spanId: "remote", attributes: { reason: "steer" } }, + { traceId: "other-trace", spanId: "other" }, + ], + }); + const first = span("first", { + parentId: "root", + startedMs: 1005, + endedMs: 1009.5, + status: { status: "error", error: { name: "Error", message: "failed call" } }, + }); + const nested = span("nested", { parentId: "first", startedMs: 1006, endedMs: 1007 }); + const second = span("second", { + parentId: "root", + startedMs: 1010, + endedMs: 1012, + status: { status: "error" }, + }); + const spans = Object.freeze([second, nested, first, root]); + const before = structuredClone(spans); + expect(renderTrace(spans)).toBe( + [ + 'root request="a b" attempt=2 cached=false tags=["a","b"] sizes=[1,2] flags=[true,false] (20ms)', + ' · checkpoint phase="tools" (+2ms)', + " · tick (+3.5ms)", + ' → link trace remote-trace span remote reason="steer"', + " → link trace other-trace span other", + " first (4.5ms) error Error: failed call", + " nested (1ms)", + " second (2ms) error", + ].join("\n"), + ); + expect(spans).toEqual(before); + }); + + it("treats missing parents as roots and orders roots and children by start time", () => { + expect( + renderTrace([ + span("later", { parentId: "external", startedMs: 20, endedMs: 30 }), + span("child", { parentId: "later", startedMs: 25, endedMs: 30 }), + span("earlier", { startedMs: 10, endedMs: 15, links: [{ traceId: "trace", spanId: "later" }] }), + ]), + ).toBe(["earlier (5ms)", " → link trace trace span later", "later (10ms)", " child (5ms)"].join("\n")); + }); + + it("prints cross-trace links without following them, including cycles", () => { + expect( + renderTrace([ + span("server:1", { + name: "server", + traceId: "server-trace", + links: [{ traceId: "worker-trace", spanId: "worker:1" }], + }), + span("worker:1", { + name: "worker", + traceId: "worker-trace", + links: [{ traceId: "server-trace", spanId: "server:1" }], + }), + ]), + ).toBe( + [ + "server (10ms)", + " → link trace worker-trace span worker:1", + "worker (10ms)", + " → link trace server-trace span server:1", + ].join("\n"), + ); + }); + + it("deduplicates spans merged from stores by global ID", () => { + const root = span("server:1", { name: "root" }); + const child = span("worker:1", { + name: "child", + parentId: "server:1", + events: [{ name: "done", timeMs: 3, attributes: {} }], + }); + const grandchild = span("worker:2", { name: "grandchild", parentId: "worker:1" }); + expect(renderTrace([child, root, grandchild, structuredClone(root), structuredClone(child)])).toBe( + "root (10ms)\n child (10ms)\n · done (+3ms)\n grandchild (10ms)", + ); + }); + + it("preserves input order for equal start times without mutating the input", () => { + const spans = Object.freeze([ + span("root"), + span("z", { parentId: "root" }), + span("a", { parentId: "root" }), + span("other"), + ]); + expect(renderTrace(spans)).toBe("root (10ms)\n z (10ms)\n a (10ms)\nother (10ms)"); + }); + + it("escapes string attribute values without hiding their contents", () => { + expect(renderTrace([span("root", { attributes: { text: 'line one\n"line two"' } })])).toBe( + 'root text="line one\\n\\"line two\\"" (10ms)', + ); + }); + + it("renders deep trees without recursive calls", () => { + const spans = Array.from({ length: 2000 }, (_, index) => + span(String(index), index === 0 ? {} : { parentId: String(index - 1) }), + ); + const rendered = renderTrace(spans); + expect(rendered.split("\n")).toHaveLength(spans.length); + expect(rendered.endsWith(`${" ".repeat(1999)}1999 (10ms)`)).toBe(true); + }); + + it("renders an empty input as an empty string", () => { + expect(renderTrace([])).toBe(""); + }); +}); diff --git a/packages/telemetry/test/store-conformance.ts b/packages/telemetry/test/store-conformance.ts new file mode 100644 index 000000000..9b12a71c1 --- /dev/null +++ b/packages/telemetry/test/store-conformance.ts @@ -0,0 +1,225 @@ +import { describe, expect, it, vi } from "vitest"; +import type { + TelemetryCarrier, + TelemetryContext, + TelemetryMetrics, + TelemetryPropagation, + TelemetryStore, +} from "../src/index.ts"; + +export interface StoreFixture extends AsyncDisposable { + readonly context: TelemetryContext & { + readonly metrics: TelemetryMetrics; + readonly propagation: TelemetryPropagation; + }; + readonly store: TelemetryStore; + flush(): Promise; +} + +/** Shared read-store checks; no raw-measurement or backend-specific persistence assumptions. */ +export function telemetryStoreConformance(name: string, factory: () => Promise): void { + describe(name, () => { + it("returns only settled traces, complete span snapshots, and detached results", async () => { + vi.spyOn(Date, "now").mockReturnValue(20_000); + await using f = await factory(); + const gate = Promise.withResolvers(); + const active = f.context.startSpan({ name: "root", attributes: { tags: ["original"] } }, async (root) => { + root.addEvent("event", { tags: ["original"] }); + await gate.promise; + await root.startSpan({ name: "child" }, () => {}); + }); + await f.flush(); + expect((await f.store.findTraces()).traces).toEqual([]); + expect(await f.store.getTrace("missing")).toBeUndefined(); + gate.resolve(); + await active; + await f.flush(); + const trace = (await f.store.findTraces()).traces[0]!; + const root = trace.spans.find((span) => span.name === "root")!; + const child = trace.spans.find((span) => span.name === "child")!; + expect(root.id).toMatch(/^[0-9a-f-]{36}:[1-9][0-9]*$/); + expect(root.traceId).toBe(root.id); + expect(root.parentId).toBeUndefined(); + expect(child.parentId).toBe(root.id); + expect(child.traceId).toBe(root.traceId); + expect(root.startedMs).toBe(20_000); + expect(root.endedMs).toBeGreaterThanOrEqual(root.startedMs); + expect(root.events[0]?.timeMs).toBeGreaterThanOrEqual(0); + root.attributes.tags = ["changed"]; + root.events[0]!.attributes.tags = ["changed"]; + const fresh = await f.store.getTrace(trace.traceId); + expect(fresh?.spans.find((span) => span.id === root.id)?.attributes.tags).toEqual(["original"]); + expect(fresh?.spans.find((span) => span.id === root.id)?.events[0]?.attributes.tags).toEqual(["original"]); + }); + + it("filters by first-span name and start, any error, and all attributes on one span", async () => { + const now = vi.spyOn(Date, "now").mockReturnValue(10_000); + await using f = await factory(); + await f.context.startSpan({ name: "old" }, async (root) => { + now.mockReturnValue(30_000); + await root.startSpan({ name: "recent-child", attributes: { left: 1 } }, () => {}); + await root.startSpan({ name: "split", attributes: { right: 2 } }, () => {}); + }); + now.mockReturnValue(20_000); + await f.context.startSpan({ name: "wanted" }, async (root) => { + now.mockReturnValue(20_001); + await root.startSpan( + { name: "failure", attributes: { left: 1, right: 2, valid: true, 'odd."key': "value" } }, + (child) => child.setStatus({ status: "error" }), + ); + }); + now.mockReturnValue(40_000); + await f.context.startSpan({ name: "new", attributes: { valid: 1 } }, () => {}); + await f.flush(); + const names = async (query: Parameters[0]) => + (await f.store.findTraces(query)).traces.map((trace) => trace.spans[0]?.name); + expect( + await names({ + name: "wanted", + attributes: { left: 1, right: 2 }, + errors: true, + startMin: 20_000, + startMax: 20_000, + }), + ).toEqual(["wanted"]); + expect(await names({ name: "failure" })).toEqual([]); + expect(await names({ attributes: { left: 1, right: 2 } })).toEqual(["wanted"]); + expect(await names({ attributes: { valid: true, 'odd."key': "value" } })).toEqual(["wanted"]); + expect(await names({ attributes: { valid: 1 } })).toEqual(["new"]); + expect(await names({ errors: true })).toEqual(["wanted"]); + expect(await names({ startMin: 20_000 })).toEqual(["new", "wanted"]); + expect(await names({ startMax: 10_000 })).toEqual(["old"]); + }); + + it("pages complete traces without duplicates when trace starts tie", async () => { + const now = vi.spyOn(Date, "now").mockReturnValue(20_000); + await using f = await factory(); + for (let index = 0; index < 25; index++) { + now.mockReturnValue(20_000); + await f.context.startSpan({ name: "root", attributes: { index } }, async (root) => { + now.mockReturnValue(20_001); + await root.startSpan({ name: "child" }, () => {}); + }); + } + await f.flush(); + const first = await f.store.findTraces(); + expect(first.traces).toHaveLength(20); + expect(first.cursor).toBeDefined(); + const rest = await f.store.findTraces({ cursor: first.cursor }); + expect(rest.traces).toHaveLength(5); + expect(rest.cursor).toBeUndefined(); + const all = (await f.store.findTraces({ limit: 100 })).traces; + expect([...first.traces, ...rest.traces]).toEqual(all); + expect(all.every((trace) => trace.spans.length === 2)).toBe(true); + const ids: string[] = []; + let cursor: string | undefined; + do { + const page = await f.store.findTraces({ name: "root", limit: 3, cursor }); + ids.push(...page.traces.map((trace) => trace.traceId)); + cursor = page.cursor; + } while (cursor); + expect(ids).toEqual(all.map((trace) => trace.traceId)); + }); + + it("inherits remote trace IDs and preserves remote parents and cross-trace links", async () => { + vi.spyOn(Date, "now").mockReturnValue(20_000); + await using sender = await factory(); + await using receiver = await factory(); + let carrier!: TelemetryCarrier; + await sender.context.startSpan({ name: "sender" }, (span) => { + carrier = sender.context.propagation.inject(span)!; + }); + await sender.flush(); + const sent = (await sender.store.findTraces()).traces[0]!; + const remote = receiver.context.propagation.extract(JSON.parse(JSON.stringify(carrier)) as TelemetryCarrier); + await remote.startSpan({ name: "receiver" }, () => {}); + await receiver.context.startSpan( + { name: "linked", links: [{ context: remote, attributes: { reason: "background" } }] }, + () => {}, + ); + await receiver.flush(); + const continued = await receiver.store.getTrace(sent.traceId); + expect(continued?.spans).toHaveLength(1); + expect(continued?.spans[0]).toMatchObject({ + traceId: sent.traceId, + parentId: sent.spans[0]!.id, + name: "receiver", + }); + expect(continued?.spans[0]?.id).not.toBe(sent.spans[0]!.id); + const linked = (await receiver.store.findTraces({ name: "linked" })).traces[0]!; + expect(linked.traceId).not.toBe(sent.traceId); + expect(linked.spans[0]?.links).toEqual([ + { traceId: sent.traceId, spanId: sent.spans[0]!.id, attributes: { reason: "background" } }, + ]); + for (const key of Object.keys(carrier)) { + const incomplete = { ...carrier }; + delete incomplete[key]; + await receiver.context.propagation.extract(incomplete).startSpan({ name: "fallback" }, () => {}); + const nonString = { ...carrier, [key]: 42 } as unknown as TelemetryCarrier; + await receiver.context.propagation.extract(nonString).startSpan({ name: "fallback" }, () => {}); + } + await receiver.flush(); + const fallbacks = (await receiver.store.findTraces({ name: "fallback" })).traces; + expect(fallbacks).toHaveLength(Object.keys(carrier).length * 2); + expect( + fallbacks.every((trace) => trace.spans[0]?.parentId === undefined && trace.traceId === trace.spans[0]?.id), + ).toBe(true); + }); + + it("accepts external string identities without imposing an ID format", async () => { + await using f = await factory(); + let carrier!: TelemetryCarrier; + await f.context.startSpan({ name: "source" }, (span) => { + carrier = f.context.propagation.inject(span)!; + }); + const external = Object.fromEntries(Object.keys(carrier).map((key) => [key, `external-${key}`])); + const parent = f.context.propagation.extract(external); + expect(f.context.propagation.inject(parent)).toEqual(external); + await parent.startSpan({ name: "continued" }, () => {}); + await f.flush(); + const trace = (await f.store.findTraces({ name: "continued" })).traces[0]!; + expect(trace.traceId).toMatch(/^external-/); + expect(trace.spans[0]?.parentId).toMatch(/^external-/); + }); + + it("queries metric names, inclusive interval bounds, typed attribute equality, and detached values", async () => { + const now = vi.spyOn(Date, "now").mockReturnValue(20_001); + await using f = await factory(); + const counter = f.context.metrics.counter("hits", { unit: "By" }); + counter.add(2, { method: "read", valid: true }); + counter.add(3, { valid: true, method: "read" }); + counter.add(11, { method: "write", valid: 1 }); + f.context.metrics.counter("other").add(100, { method: "read", valid: true }); + now.mockReturnValue(30_001); + counter.add(7, { method: "read", valid: true }); + now.mockReturnValue(40_001); + await f.flush(); + const query = { + name: "hits", + startMin: 20_000, + startMax: 20_000, + attributes: { method: "read", valid: true }, + }; + const rows = await f.store.findMetrics(query); + expect(rows.reduce((count, row) => count + row.count, 0)).toBe(2); + expect(rows.reduce((sum, row) => sum + row.sum, 0)).toBe(5); + expect(Math.min(...rows.map((row) => row.min))).toBe(2); + expect(Math.max(...rows.map((row) => row.max))).toBe(3); + expect(rows.every((row) => row.kind === "counter" && row.unit === "By" && row.intervalMs === 20_000)).toBe( + true, + ); + rows[0]!.attributes.method = "changed"; + expect((await f.store.findMetrics(query))[0]?.attributes.method).toBe("read"); + expect(await f.store.findMetrics({ name: "missing" })).toEqual([]); + expect( + (await f.store.findMetrics({ name: "hits", attributes: { valid: 1 } })).reduce( + (sum, row) => sum + row.sum, + 0, + ), + ).toBe(11); + expect( + (await f.store.findMetrics({ name: "hits", startMin: 30_000 })).reduce((sum, row) => sum + row.sum, 0), + ).toBe(7); + }); + }); +} diff --git a/packages/telemetry/test/store.test.ts b/packages/telemetry/test/store.test.ts new file mode 100644 index 000000000..d59d45cad --- /dev/null +++ b/packages/telemetry/test/store.test.ts @@ -0,0 +1,82 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { InMemoryTelemetryContext, type TelemetrySpan } from "../src/index.ts"; +import { telemetryStoreConformance } from "./store-conformance.ts"; + +afterEach(() => vi.restoreAllMocks()); + +telemetryStoreConformance("in-memory telemetry store conformance", async () => { + const context = new InMemoryTelemetryContext(); + return { context, store: context, flush: async () => {}, [Symbol.asyncDispose]: async () => {} }; +}); + +describe("in-memory store snapshots", () => { + it("records wall-clock boundaries and relative event times without changing normalized event snapshots", async () => { + vi.spyOn(Date, "now").mockReturnValue(1000); + const clock = vi.spyOn(performance, "now").mockReturnValue(100); + const telemetry = new InMemoryTelemetryContext(); + let id = ""; + await telemetry.startSpan({ name: "timed" }, (span) => { + const carrier = telemetry.inject(span)!; + id = `${carrier["pi-memory-scope"]}:${carrier["pi-memory-span"]}`; + clock.mockReturnValue(110); + span.addEvent("event", { count: 1 }); + clock.mockReturnValue(125); + }); + expect(await telemetry.getTrace(id)).toEqual({ + traceId: id, + spans: [ + { + id, + traceId: id, + name: "timed", + startedMs: 1000, + endedMs: 1025, + attributes: {}, + events: [{ name: "event", timeMs: 10, attributes: { count: 1 } }], + links: [], + status: { status: "ok" }, + }, + ], + }); + expect(telemetry.getSpans()[0]?.events).toEqual([{ name: "event", attributes: { count: 1 } }]); + expect(telemetry.getSpans()[0]).not.toHaveProperty("startedMs"); + }); + + it("maps raw metric measurements to count-one rows without exposing private timing in getMetrics", async () => { + vi.spyOn(Date, "now").mockReturnValue(20_001); + const telemetry = new InMemoryTelemetryContext(); + telemetry.counter("calls").add(2); + telemetry.counter("calls").add(3); + expect( + (await telemetry.findMetrics({ name: "calls" })).map((row) => [row.intervalMs, row.count, row.sum]), + ).toEqual([ + [20_000, 1, 2], + [20_000, 1, 3], + ]); + expect(telemetry.getMetrics()).toEqual([ + { kind: "counter", name: "calls", value: 2, attributes: {}, context: null }, + { kind: "counter", name: "calls", value: 3, attributes: {}, context: null }, + ]); + }); + + it("detaches link attributes and error details", async () => { + const telemetry = new InMemoryTelemetryContext(); + let target!: TelemetrySpan; + await telemetry.startSpan({ name: "target" }, (span) => { + target = span; + }); + await telemetry.startSpan( + { name: "source", links: [{ context: target, attributes: { tags: ["original"] } }] }, + (span) => { + span.setStatus({ status: "error", error: { name: "Expected", message: "original" } }); + }, + ); + const trace = (await telemetry.findTraces({ name: "source" })).traces[0]!; + const span = trace.spans[0]!; + span.links[0]!.attributes!.tags = ["changed"]; + if (span.status.status === "error" && span.status.error) span.status.error.message = "changed"; + const fresh = (await telemetry.getTrace(trace.traceId))!.spans[0]!; + expect(fresh.links[0]?.attributes?.tags).toEqual(["original"]); + expect(fresh.status).toEqual({ status: "error", error: { name: "Expected", message: "original" } }); + }); +}); diff --git a/packages/telemetry/test/telemetry.test.ts b/packages/telemetry/test/telemetry.test.ts index bcd31fe32..0a0ee7e42 100644 --- a/packages/telemetry/test/telemetry.test.ts +++ b/packages/telemetry/test/telemetry.test.ts @@ -7,27 +7,10 @@ import { NOOP_TELEMETRY_METRICS, NOOP_TELEMETRY_PROPAGATION, type SchemaTelemetrySpan, - type SpanAttributes, - type SpanOptions, - type SpanStatus, type TelemetrySchemaSpanStartAttributes, type TelemetrySpan, } from "../src/index.ts"; -function unreadable(value: T): T { - return new Proxy(value, { - get: () => { - throw new Error("read"); - }, - getOwnPropertyDescriptor: () => { - throw new Error("inspect"); - }, - ownKeys: () => { - throw new Error("enumerate"); - }, - }); -} - describe("telemetry schemas", () => { it("preserves serializable definitions and infers exact attributes", () => { const definition = { @@ -121,9 +104,6 @@ describe("telemetry schemas", () => { const requestSpan = spans.find((span) => span.name === "request"); expect(operationSpan?.parentId).toBeNull(); expect(requestSpan?.parentId).toBe(operationSpan?.id); - expect(() => - createTypedSpanStarter(telemetryContext, unreadable([operationSchema, requestSchema] as const)), - ).not.toThrow(); const syncError = { kind: "sync" }; const syncResult = startSpan("operation", { kind: "write" }, () => { @@ -190,14 +170,17 @@ describe("NOOP_TELEMETRY_CONTEXT", () => { await expect(asyncResult).rejects.toBe(asyncError); }); - it("does not inspect or retain telemetry payloads", async () => { - const options = unreadable({ name: "operation", attributes: { secret: "prompt content" } }); - await NOOP_TELEMETRY_CONTEXT.startSpan(options, (span) => { - const attributes = unreadable({ secret: "content" }); - const status = unreadable({ status: "ok" }); - expect(() => span.addEvent("event", attributes)).not.toThrow(); - expect(() => span.setAttributes(attributes)).not.toThrow(); - expect(() => span.setStatus(status)).not.toThrow(); - }); + it("ignores typed recording payloads", async () => { + const result = await NOOP_TELEMETRY_CONTEXT.startSpan( + { name: "operation", attributes: { kind: "test" } }, + (span) => { + span.addEvent("event", { count: 1 }); + span.setAttributes({ complete: true }); + span.setStatus({ status: "error", error: { name: "Expected", message: "test" } }); + span.addLink({ context: span }); + return 42; + }, + ); + expect(result).toBe(42); }); }); diff --git a/tsconfig.json b/tsconfig.json index fe67ca64c..b530a76e4 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -10,11 +10,13 @@ "@earendil-works/chord/delta": ["./packages/chord/src/delta/index.ts"], "@earendil-works/chord/node": ["./packages/chord/src/node.ts"], "@earendil-works/chord/output": ["./packages/chord/src/output.ts"], + "@earendil-works/chord/telemetry-sqlite": ["./packages/chord/src/telemetry-sqlite/index.ts"], "@earendil-works/chord/sqlite": ["./packages/chord/src/sqlite/index.ts"], "@earendil-works/chord/sqlite/node": ["./packages/chord/src/sqlite/node.ts"], "@earendil-works/chord/sqlite/cloudflare": ["./packages/chord/src/sqlite/cloudflare.ts"], "@earendil-works/pi-telemetry": ["./packages/telemetry/src/index.ts"], "@earendil-works/pi-telemetry/testing": ["./packages/telemetry/src/testing/index.ts"], + "@earendil-works/pi-telemetry/render": ["./packages/telemetry/src/render.ts"], "@earendil-works/pi-codemode": ["./packages/codemode/src/index.ts"], "@earendil-works/pi-codemode/cloudflare": ["./packages/codemode/src/cloudflare.ts"], "@earendil-works/pi-codemode/declarations": ["./packages/codemode/src/declarations.ts"],