From d193c6843dc779b0ec996bcdace8ef0232f839b3 Mon Sep 17 00:00:00 2001 From: seekskyworld Date: Sat, 5 Sep 2026 10:44:51 +0800 Subject: [PATCH] feat(telemetry): add opt-in OTLP usage traces Add a redacted OTLP/HTTP exporter for model and tool usage records, with batching, endpoint configuration, shutdown flushing, and runtime documentation. Generated-by: OpenAI Codex Signed-off-by: seekskyworld --- packages/cli/README.md | 20 ++ .../__tests__/otlp-telemetry-exporter.test.ts | 168 +++++++++++ .../storage/src/otlp-telemetry-exporter.ts | 272 ++++++++++++++++++ packages/storage/src/usage-stores.ts | 25 +- 4 files changed, 482 insertions(+), 3 deletions(-) create mode 100644 packages/storage/src/__tests__/otlp-telemetry-exporter.test.ts create mode 100644 packages/storage/src/otlp-telemetry-exporter.ts diff --git a/packages/cli/README.md b/packages/cli/README.md index 0595257399..046b9e55c4 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -56,6 +56,26 @@ The release gate validates the following installed-package matrix: Other combinations that satisfy the Node.js minimum may work, but are not part of the current release gate. Real Eval executor validation currently runs on Linux x64 with Node.js 24. +## Optional OpenTelemetry traces + +Maka can export bounded model-call and tool-invocation traces to an OTLP/HTTP collector. The +exporter is disabled unless an endpoint is configured in the Runtime Host environment: + +```sh +export OTEL_EXPORTER_OTLP_ENDPOINT=https://otel.example.com +export OTEL_EXPORTER_OTLP_HEADERS='authorization=Bearer%20token' +export OTEL_SERVICE_NAME=maka +maka +``` + +`OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` overrides the generic endpoint when a collector uses a +separate traces URL. `OTEL_RESOURCE_ATTRIBUTES` adds URL-encoded `key=value` resource attributes. + +Exported spans contain provider/model identifiers, model-call kind, tool name, duration, status, +token counts, cost, byte counts, and bounded error classes. Prompts, message contents, tool +arguments/results, credentials, and session paths are never exported. Export failures are +best-effort and do not fail a turn or change the local Usage ledger. + ## Install Install the current beta explicitly from the `next` dist-tag: diff --git a/packages/storage/src/__tests__/otlp-telemetry-exporter.test.ts b/packages/storage/src/__tests__/otlp-telemetry-exporter.test.ts new file mode 100644 index 0000000000..6ebb7af551 --- /dev/null +++ b/packages/storage/src/__tests__/otlp-telemetry-exporter.test.ts @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { PersistedLlmCallRecord } from '../telemetry-file-schema.js'; +import { + createOtlpTelemetryExporter, + type OtlpTelemetryExporterOptions, +} from '../otlp-telemetry-exporter.js'; + +test('exports bounded usage spans through OTLP/HTTP', async () => { + const requests: Array<{ url: string; init: RequestInit }> = []; + const options: OtlpTelemetryExporterOptions = { + env: { + OTEL_EXPORTER_OTLP_ENDPOINT: 'https://collector.example.test', + OTEL_EXPORTER_OTLP_HEADERS: 'authorization=Bearer%20secret,invalid', + OTEL_SERVICE_NAME: 'maka-test', + OTEL_RESOURCE_ATTRIBUTES: 'deployment.environment=ci,empty=', + }, + fetch: async (url, init) => { + requests.push({ url, init: init ?? {} }); + return new Response(null, { status: 200 }); + }, + }; + const exporter = createOtlpTelemetryExporter(options); + assert.ok(exporter); + + const record: PersistedLlmCallRecord = { + id: 'usage-1', + providerId: 'openai', + modelId: 'gpt-test', + inputTokens: 12, + outputTokens: 7, + totalTokens: 19, + cacheHitInputTokens: 0, + cacheMissInputTokens: 12, + cachedInputTokens: 0, + cacheWriteInputTokens: 0, + reasoningTokens: 0, + costUsd: 0.01, + latencyMs: 25, + startedAt: 1_000, + status: 'success', + date: '1970-01-01', + ts: 1_025, + }; + await exporter.exportLlmCall(record); + await exporter.close(); + + assert.equal(requests.length, 1); + assert.equal(requests[0]?.url, 'https://collector.example.test/v1/traces'); + const headers = requests[0]?.init.headers as Record; + assert.equal(headers.authorization, 'Bearer secret'); + const payload = JSON.parse(String(requests[0]?.init.body)); + const span = payload.resourceSpans[0].scopeSpans[0].spans[0]; + assert.equal(payload.resourceSpans[0].resource.attributes[0].value.stringValue, 'maka-test'); + assert.equal(span.name, 'maka.llm.call'); + assert.equal( + span.attributes.some((item: { key: string }) => item.key === 'argsSummary'), + false, + ); +}); + +test('does not create an exporter without an OTLP endpoint', () => { + assert.equal(createOtlpTelemetryExporter({ env: {} }), undefined); +}); + +test('normalizes an explicitly configured traces endpoint with a trailing slash', async () => { + const requests: string[] = []; + const exporter = createOtlpTelemetryExporter({ + env: { OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: 'https://collector.example.test/v1/traces/' }, + fetch: async (url) => { + requests.push(url); + return new Response(null, { status: 200 }); + }, + }); + assert.ok(exporter); + + await exporter.exportLlmCall({ + id: 'usage-2', + providerId: 'openai', + modelId: 'gpt-test', + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + cacheHitInputTokens: 0, + cacheMissInputTokens: 1, + cachedInputTokens: 0, + cacheWriteInputTokens: 0, + reasoningTokens: 0, + costUsd: 0, + latencyMs: 1, + startedAt: 1, + status: 'success', + date: '1970-01-01', + ts: 2, + }); + await exporter.close(); + + assert.deepEqual(requests, ['https://collector.example.test/v1/traces']); +}); + +test('flush drains spans queued while another batch is in flight', async () => { + let releaseFirst!: () => void; + const firstBatch = new Promise((resolve) => { + releaseFirst = resolve; + }); + const payloads: Array<{ spanCount: number }> = []; + const exporter = createOtlpTelemetryExporter({ + env: { OTEL_EXPORTER_OTLP_ENDPOINT: 'https://collector.example.test' }, + fetch: async (_url, init) => { + const payload = JSON.parse(String(init?.body)) as { + resourceSpans: Array<{ scopeSpans: Array<{ spans: unknown[] }> }>; + }; + payloads.push({ spanCount: payload.resourceSpans[0]?.scopeSpans[0]?.spans.length ?? 0 }); + if (payloads.length === 1) await firstBatch; + return new Response(null, { status: 200 }); + }, + }); + assert.ok(exporter); + + const record: PersistedLlmCallRecord = { + id: 'usage-batch', + providerId: 'openai', + modelId: 'gpt-test', + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + cacheHitInputTokens: 0, + cacheMissInputTokens: 1, + cachedInputTokens: 0, + cacheWriteInputTokens: 0, + reasoningTokens: 0, + costUsd: 0, + latencyMs: 1, + startedAt: 1, + status: 'success', + date: '1970-01-01', + ts: 2, + }; + for (let index = 0; index < 32; index += 1) { + void exporter.exportLlmCall({ ...record, id: `usage-batch-${index}` }); + } + await new Promise((resolve) => setImmediate(resolve)); + void exporter.exportLlmCall({ ...record, id: 'usage-batch-late' }); + await new Promise((resolve) => setImmediate(resolve)); + releaseFirst(); + await exporter.close(); + + assert.deepEqual(payloads, [{ spanCount: 32 }, { spanCount: 1 }]); +}); diff --git a/packages/storage/src/otlp-telemetry-exporter.ts b/packages/storage/src/otlp-telemetry-exporter.ts new file mode 100644 index 0000000000..2254c777de --- /dev/null +++ b/packages/storage/src/otlp-telemetry-exporter.ts @@ -0,0 +1,272 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { randomUUID } from 'node:crypto'; +import type { + PersistedLlmCallRecord, + PersistedToolInvocationRecord, +} from './telemetry-file-schema.js'; + +const BATCH_SIZE = 32; +const FLUSH_DELAY_MS = 1_000; +const DEFAULT_SERVICE_NAME = 'maka'; + +type FetchLike = (input: string, init?: RequestInit) => Promise; +type Environment = Record; +type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number }; +type OtlpAttribute = { key: string; value: OtlpAttributeValue }; +type OtlpSpan = { + traceId: string; + spanId: string; + name: string; + kind: 1; + startTimeUnixNano: string; + endTimeUnixNano: string; + attributes: OtlpAttribute[]; + status: { code: 0 | 1 | 2 }; +}; + +export interface OtlpTelemetryExporter { + exportLlmCall(record: PersistedLlmCallRecord): Promise; + exportToolInvocation(record: PersistedToolInvocationRecord): Promise; + flush(): Promise; + close(): Promise; +} + +export interface OtlpTelemetryExporterOptions { + readonly env?: Environment; + readonly fetch?: FetchLike; +} + +export function createOtlpTelemetryExporter( + options: OtlpTelemetryExporterOptions = {}, +): OtlpTelemetryExporter | undefined { + const env = options.env ?? process.env; + const endpoint = resolveEndpoint(env); + if (!endpoint) return undefined; + return new OtlpTelemetryExporterImpl( + endpoint, + parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS), + resourceAttributes(env), + options.fetch ?? fetch, + ); +} + +class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter { + readonly #endpoint: string; + readonly #headers: Record; + readonly #resourceAttributes: OtlpAttribute[]; + readonly #fetch: FetchLike; + readonly #pending: OtlpSpan[] = []; + #timer: NodeJS.Timeout | undefined; + #flushPromise: Promise | undefined; + #closed = false; + + constructor( + endpoint: string, + headers: Record, + resourceAttributes: OtlpAttribute[], + fetchFn: FetchLike, + ) { + this.#endpoint = endpoint; + this.#headers = headers; + this.#resourceAttributes = resourceAttributes; + this.#fetch = fetchFn; + } + + async exportLlmCall(record: PersistedLlmCallRecord): Promise { + await this.enqueue({ + name: 'maka.llm.call', + startedAt: record.startedAt, + durationMs: record.latencyMs, + status: record.status, + attributes: [ + stringAttribute('maka.telemetry.kind', 'llm'), + stringAttribute('maka.provider.id', record.providerId), + stringAttribute('maka.model.id', record.modelId), + ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []), + numberAttribute('maka.usage.input_tokens', record.inputTokens), + numberAttribute('maka.usage.output_tokens', record.outputTokens), + numberAttribute('maka.usage.total_tokens', record.totalTokens), + numberAttribute('maka.usage.cost_usd', record.costUsd), + ...(record.errorClass ? [stringAttribute('maka.error.class', record.errorClass)] : []), + ], + }); + } + + async exportToolInvocation(record: PersistedToolInvocationRecord): Promise { + await this.enqueue({ + name: 'maka.tool.invocation', + startedAt: record.startedAt, + durationMs: record.durationMs, + status: record.status, + attributes: [ + stringAttribute('maka.telemetry.kind', 'tool'), + stringAttribute('maka.tool.name', record.toolName), + ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []), + ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []), + numberAttribute('maka.tool.bytes_in', record.bytesIn), + numberAttribute('maka.tool.bytes_out', record.bytesOut), + ...(record.errorClass ? [stringAttribute('maka.error.class', record.errorClass)] : []), + ], + }); + } + + async flush(): Promise { + if (this.#flushPromise) await this.#flushPromise; + while (this.#pending.length > 0) { + const spans = this.#pending.splice(0); + if (this.#timer) { + clearTimeout(this.#timer); + this.#timer = undefined; + } + this.#flushPromise = this.send(spans).finally(() => { + this.#flushPromise = undefined; + }); + await this.#flushPromise; + } + } + + async close(): Promise { + this.#closed = true; + if (this.#timer) clearTimeout(this.#timer); + this.#timer = undefined; + await this.flush(); + } + + private async enqueue(input: { + name: string; + startedAt: number; + durationMs: number; + status: PersistedLlmCallRecord['status']; + attributes: OtlpAttribute[]; + }): Promise { + if (this.#closed) return; + this.#pending.push(toSpan(input)); + if (this.#pending.length >= BATCH_SIZE) { + await this.flush(); + return; + } + if (!this.#timer) { + this.#timer = setTimeout(() => { + this.#timer = undefined; + void this.flush(); + }, FLUSH_DELAY_MS); + } + } + + private async send(spans: OtlpSpan[]): Promise { + try { + const response = await this.#fetch(this.#endpoint, { + method: 'POST', + headers: { 'content-type': 'application/json', ...this.#headers }, + body: JSON.stringify({ + resourceSpans: [ + { + resource: { attributes: this.#resourceAttributes }, + scopeSpans: [{ scope: { name: 'maka.storage' }, spans }], + }, + ], + }), + }); + if (!response.ok) { + console.error(`[telemetry] OTLP export failed: HTTP ${response.status}`); + } + } catch { + console.error('[telemetry] OTLP export failed: request error'); + } + } +} + +function toSpan(input: { + name: string; + startedAt: number; + durationMs: number; + status: PersistedLlmCallRecord['status']; + attributes: OtlpAttribute[]; +}): OtlpSpan { + const traceId = randomUUID().replaceAll('-', ''); + return { + traceId, + spanId: traceId.slice(0, 16), + name: input.name, + kind: 1, + startTimeUnixNano: String(Math.max(0, input.startedAt) * 1_000_000), + endTimeUnixNano: String( + Math.max(0, input.startedAt + Math.max(0, input.durationMs)) * 1_000_000, + ), + attributes: input.attributes, + status: { code: input.status === 'success' ? 1 : input.status === 'error' ? 2 : 0 }, + }; +} + +function stringAttribute(key: string, value: string): OtlpAttribute { + return { key, value: { stringValue: value } }; +} + +function numberAttribute(key: string, value: number): OtlpAttribute { + return { key, value: { doubleValue: value } }; +} + +function resolveEndpoint(env: Environment): string | undefined { + const configured = env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? env.OTEL_EXPORTER_OTLP_ENDPOINT; + if (!configured) return undefined; + try { + const url = new URL(configured); + const pathname = url.pathname.replace(/\/+$/u, ''); + url.pathname = pathname.endsWith('/v1/traces') ? pathname : `${pathname}/v1/traces`; + return url.toString(); + } catch { + return undefined; + } +} + +function parseHeaders(value: string | undefined): Record { + if (!value) return {}; + const headers: Record = {}; + for (const item of value.split(',')) { + const separator = item.indexOf('='); + if (separator <= 0) continue; + const key = item.slice(0, separator).trim(); + const raw = item.slice(separator + 1).trim(); + if (!key || !raw) continue; + headers[key] = decodeValue(raw); + } + return headers; +} + +function resourceAttributes(env: Environment): OtlpAttribute[] { + const attributes = new Map(); + attributes.set('service.name', env.OTEL_SERVICE_NAME?.trim() || DEFAULT_SERVICE_NAME); + for (const item of env.OTEL_RESOURCE_ATTRIBUTES?.split(',') ?? []) { + const separator = item.indexOf('='); + if (separator <= 0) continue; + const key = item.slice(0, separator).trim(); + if (key) attributes.set(key, decodeValue(item.slice(separator + 1).trim())); + } + return [...attributes].map(([key, value]) => stringAttribute(key, value)); +} + +function decodeValue(value: string): string { + try { + return decodeURIComponent(value); + } catch { + return value; + } +} diff --git a/packages/storage/src/usage-stores.ts b/packages/storage/src/usage-stores.ts index 3e4b3a8953..57df2b76b6 100644 --- a/packages/storage/src/usage-stores.ts +++ b/packages/storage/src/usage-stores.ts @@ -64,6 +64,10 @@ import { type ToolUsageQuery, } from './telemetry-repo.js'; import { createSqlitePricingStore, createSqliteTelemetryRepo } from './sqlite-usage-store.js'; +import { + createOtlpTelemetryExporter, + type OtlpTelemetryExporter, +} from './otlp-telemetry-exporter.js'; const readerBrand: unique symbol = Symbol('InteractiveUsageStoresReader'); const writerBrand: unique symbol = Symbol('InteractiveUsageStoresWriter'); @@ -292,7 +296,13 @@ export async function openInteractiveUsageStoresForWrite( if (opening) return opening; const pending = runWithStorageRootLease(lease, 'interactive', 'write', async (root) => { const repos = await openRepos(root, true); - const stores = createWriterFacade(lease, repos.telemetry, repos.modelCalls, repos.pricing); + const stores = createWriterFacade( + lease, + repos.telemetry, + repos.modelCalls, + repos.pricing, + createOtlpTelemetryExporter(), + ); writers.add(stores); writerByLease.set(lease, stores); return stores; @@ -333,6 +343,7 @@ function createWriterFacade( telemetry: TelemetryRepo, modelCalls: ModelCallLedger, pricing: PricingStore, + exporter: OtlpTelemetryExporter | undefined, ): InteractiveUsageStoresWriter { const run = (operation: () => T | Promise): Promise => runWithStorageRootLease(lease, 'interactive', 'write', async () => operation()); @@ -417,6 +428,7 @@ function createWriterFacade( run(() => telemetry.flush()), run(() => modelCalls.flush()), run(() => pricing.flush()), + exporter?.flush() ?? Promise.resolve(), ]); throwDeduplicatedFailures('Interactive usage store flush failed', [ ...failures, @@ -436,6 +448,7 @@ function createWriterFacade( telemetry.close(), modelCalls.close(), pricing.close(), + exporter?.close() ?? Promise.resolve(), ]); throwDeduplicatedFailures('Interactive usage stores close failed', [ ...failures, @@ -462,9 +475,15 @@ function createWriterFacade( latestLlmRuntimeProbe: (connectionSlug, modelId) => read(() => telemetry.latestLlmRuntimeProbe(connectionSlug, modelId)), recordLlmCall: (record) => - admitSessionUsageMutation(record.sessionId, () => telemetry.insertLlmCall(record)), + admitSessionUsageMutation(record.sessionId, async () => { + await run(() => telemetry.insertLlmCall(record)); + void exporter?.exportLlmCall(record); + }), recordToolInvocation: (record) => - admitSessionUsageMutation(record.sessionId, () => telemetry.insertToolInvocation(record)), + admitSessionUsageMutation(record.sessionId, async () => { + await run(() => telemetry.insertToolInvocation(record)); + void exporter?.exportToolInvocation(record); + }), }, modelCalls: { modelCallAttempts: (range, sessionId) => read(() => modelCalls.read(range, sessionId)),