diff --git a/packages/omp/README.md b/packages/omp/README.md index 237c7b8..64fab20 100644 --- a/packages/omp/README.md +++ b/packages/omp/README.md @@ -58,6 +58,19 @@ export default telemetryDevExtension({ Initialization is one-shot; the first extension factory initialized in a process supplies the SDK options. +## Errors + +The integration does not write to the terminal, because omp draws its TUI there. It sends SDK and +extension errors to the omp log through `pi.logger`, with the message `telemetry.dev error`. The +default location of the omp log is `~/.omp/logs`. + +A series of failed exports, for example during an ingest outage, is one entry. The next failure +after an export reaches the ingest is a new entry. + +To receive the errors in your code, pass `onError` to `telemetryDevExtension()`. The callback +replaces the omp log. To also write SDK diagnostics to the console, pass `logLevel`, for example +`"warn"`. + ## Trace shape A typical prompt produces this hierarchy: diff --git a/packages/omp/src/extension.ts b/packages/omp/src/extension.ts index 3d40018..f76bd18 100644 --- a/packages/omp/src/extension.ts +++ b/packages/omp/src/extension.ts @@ -1,8 +1,11 @@ +import { inspect } from "node:util"; + import { flush, log, startSpan, type LogLevel, + type SdkLogLevel, type SpanHandle, type TokenUsage, } from "@telemetry-dev/sdk"; @@ -13,6 +16,10 @@ import { ensureInit, type ClientOverrides, type TelemetryDevOmpOptions } from ". export interface TelemetryDevExtensionOptions extends TelemetryDevOmpOptions { /** Value for `gen_ai.agent.name` on spans and logs. Defaults to `"omp"`. */ agentName?: string; + /** SDK console output. Defaults to `"silent"`, because omp draws its TUI on the terminal. */ + logLevel?: SdkLogLevel; + /** Receives SDK and extension errors. Defaults to omp's log file, through `pi.logger`. */ + onError?: (cause: Error) => void; } /** @@ -38,6 +45,8 @@ export interface TelemetryDevExtensionContext { */ export interface TelemetryDevExtensionHost { on(event: never, handler: never): void; + /** omp's file logger. */ + logger?: { error(message: string, context?: Record): void }; } /** Structural stand-in for the host `ExtensionFactory` type. */ @@ -180,11 +189,16 @@ export function telemetryDevExtension( overrides?: ClientOverrides, ): TelemetryDevExtension { const { agentName = "omp", ...sdkOptions } = options; - const onError = sdkOptions.onError; return (pi) => { + // omp draws its TUI on the terminal, so errors go to omp's log file and not to the console. + const onError = + sdkOptions.onError ?? + ((error: Error) => pi.logger?.error("telemetry.dev error", { error: inspect(error) })); + const logLevel = sdkOptions.logLevel ?? "silent"; + try { - ensureInit(sdkOptions, overrides); + ensureInit({ ...sdkOptions, logLevel, onError }, overrides); } catch (error) { reportError(onError, error); } diff --git a/packages/omp/tests/omp.test.ts b/packages/omp/tests/omp.test.ts index efddf8d..16972a2 100644 --- a/packages/omp/tests/omp.test.ts +++ b/packages/omp/tests/omp.test.ts @@ -3,7 +3,7 @@ import { InMemoryLogRecordExporter } from "@opentelemetry/sdk-logs"; import { InMemorySpanExporter, type ReadableSpan } from "@opentelemetry/sdk-trace-base"; import { flush, type ClientOverrides } from "@telemetry-dev/sdk"; import type { ExtensionContext, ExtensionFactory } from "@oh-my-pi/pi-coding-agent"; -import { afterEach, expect, test } from "vitest"; +import { afterEach, expect, test, vi } from "vitest"; import { resetForTesting } from "../src/config.ts"; import { telemetryDevExtension, type TelemetryDevExtensionOptions } from "../src/extension.ts"; @@ -543,6 +543,46 @@ test("malformed events never throw and report through onError", async () => { expect(harness.exporters.spans.getFinishedSpans().length).toBeGreaterThan(0); }); +test("errors go to omp's file logger, not to the console", async () => { + const consoleSpies = (["error", "warn", "info", "debug", "log"] as const).map((method) => + vi.spyOn(console, method).mockImplementation(() => {}), + ); + const logged: [string, Record | undefined][] = []; + const handlers = new Map unknown>(); + + try { + telemetryDevExtension({ + apiKey: "td_live_test", + exportMode: "immediate", + fetch: async () => new Response(null, { status: 400 }), + })({ + on: (type: string, handler: never) => handlers.set(type, handler), + logger: { error: (message, context) => logged.push([message, context]) }, + }); + + await handlers.get("session_start")?.({ type: "session_start" }, makeContext()); + await flush(); + await handlers.get("message_end")?.( + { + get message(): never { + throw new Error("poisoned event"); + }, + }, + makeContext(), + ); + + expect(logged.map(([message]) => message)).toEqual([ + "telemetry.dev error", + "telemetry.dev error", + ]); + expect(String(logged[0]?.[1]?.error)).toContain("telemetry.dev log ingest failed: 400"); + expect(String(logged[1]?.[1]?.error)).toContain("poisoned event"); + expect(consoleSpies.flatMap((spy) => spy.mock.calls)).toEqual([]); + } finally { + vi.restoreAllMocks(); + } +}); + test("each event reads the current session id", async () => { const harness = makeHarness(); await harness.emit({ type: "session_start" }, makeContext("session-2")); diff --git a/packages/opencode/README.md b/packages/opencode/README.md index 1c2d236..1b5e460 100644 --- a/packages/opencode/README.md +++ b/packages/opencode/README.md @@ -63,6 +63,14 @@ Use one route, not both. If the package is in `opencode.json` and also loaded fr All telemetry SDK options can also be passed to `telemetryDevPlugin`. The `agentName` integration option controls `gen_ai.agent.name` and defaults to `opencode`. +## Errors + +The integration does not write to the console. opencode 1.x runs plugins in the process that draws its TUI, and console output from a plugin shows over the TUI. The integration sends SDK and plugin errors to the opencode log through `client.app.log`, with the service name `telemetry-dev`. The default location of the opencode log is `~/.local/share/opencode/log`. + +A series of failed exports, for example during an ingest outage, is one entry. The next failure after an export reaches the ingest is a new entry. + +To receive the errors in your code, pass `onError` to `telemetryDevPlugin()` in a plugin file. The callback replaces the opencode log. To also write SDK diagnostics to the console, set `logLevel`, for example `"warn"`. + ## Trace shape Each opencode turn produces this hierarchy, correlated by `gen_ai.conversation.id`: diff --git a/packages/opencode/src/plugin.ts b/packages/opencode/src/plugin.ts index 87942e8..66d7217 100644 --- a/packages/opencode/src/plugin.ts +++ b/packages/opencode/src/plugin.ts @@ -1,9 +1,12 @@ +import { inspect } from "node:util"; + import { captureEnabled, flush, log, startSpan, type LogLevel, + type SdkLogLevel, type SpanHandle, type TokenUsage, } from "@telemetry-dev/sdk"; @@ -15,6 +18,10 @@ import { asJsonObject, type JsonObject, type JsonValue } from "./type-guards.ts" export interface TelemetryDevPluginOptions extends TelemetryDevOpencodeOptions { /** Value for `gen_ai.agent.name` on spans and logs. Defaults to `"opencode"`. */ agentName?: string; + /** SDK console output. Defaults to `"silent"`, because opencode can draw its TUI on the plugin's terminal. */ + logLevel?: SdkLogLevel; + /** Receives SDK and plugin errors. Defaults to opencode's log, through `client.app.log`. */ + onError?: (cause: Error) => void; } /** @@ -73,6 +80,20 @@ function numberField(record: JsonObject | undefined, key: string): number | unde return parsed === value && Number.isFinite(parsed) ? parsed : undefined; } +type AppLog = (options: { body: object }) => Promise; + +/** Writes errors to opencode's log file through the plugin client, when the client has one. */ +function opencodeLog(input: unknown): ((error: Error) => void) | undefined { + const app = (input as { client?: { app?: { log?: AppLog } } } | undefined)?.client?.app; + + if (typeof app?.log !== "function") return undefined; + + return (error) => { + const body = { service: "telemetry-dev", level: "error", message: inspect(error) }; + void app.log?.({ body }).catch(() => undefined); + }; +} + function reportError(onError: ((cause: Error) => void) | undefined, cause: unknown): void { try { onError?.(cause instanceof Error ? cause : new Error(String(cause))); @@ -172,7 +193,7 @@ export function telemetryDevPlugin( options: TelemetryDevPluginOptions = {}, overrides?: ClientOverrides, ): TelemetryDevPlugin { - return async (_input: Input, configuredOptions?: Options) => { + return async (input: Input, configuredOptions?: Options) => { const mergedOptions = { ...options, ...configuredOptions, @@ -185,10 +206,13 @@ export function telemetryDevPlugin( ? configuredAgentName : "opencode"; - const onError = sdkOptions.onError; + // opencode 1.x runs plugins in the process that draws its TUI, so errors go to opencode's + // log and not to the console. + const onError = sdkOptions.onError ?? opencodeLog(input); + const logLevel = sdkOptions.logLevel ?? "silent"; try { - ensureInit(sdkOptions, overrides); + ensureInit({ ...sdkOptions, logLevel, onError }, overrides); } catch (error) { reportError(onError, error); } diff --git a/packages/opencode/tests/opencode.test.ts b/packages/opencode/tests/opencode.test.ts index 69851b4..5a2f30e 100644 --- a/packages/opencode/tests/opencode.test.ts +++ b/packages/opencode/tests/opencode.test.ts @@ -2,6 +2,7 @@ import { SpanStatusCode, trace } from "@opentelemetry/api"; import { InMemoryLogRecordExporter } from "@opentelemetry/sdk-logs"; import { InMemorySpanExporter, type ReadableSpan } from "@opentelemetry/sdk-trace-base"; import type { Hooks, Plugin, PluginModule } from "@opencode-ai/plugin"; +import { createOpencodeClient } from "@opencode-ai/sdk/client"; import { flush, type ClientOverrides } from "@telemetry-dev/sdk"; import { afterEach, describe, expect, test, vi } from "vitest"; @@ -420,6 +421,49 @@ test("poisoned events report through onError without throwing", async () => { expect(spansByName(harness.exporters, "invoke_agent")).toHaveLength(1); }); +test("errors go to opencode's log through the plugin client, not to the console", async () => { + const consoleSpies = (["error", "warn", "info", "debug", "log"] as const).map((method) => + vi.spyOn(console, method).mockImplementation(() => {}), + ); + const logged: { method: string; path: string; body: Record }[] = []; + + const client = createOpencodeClient({ + baseUrl: "http://localhost:4096", + fetch: async (request: Request) => { + const { method, url } = request; + const body = (await request.json()) as Record; + logged.push({ method, path: new URL(url).pathname, body }); + + return Response.json(true); + }, + }); + + const hooks = await telemetryDevPlugin({ + apiKey: "td_live_test", + exportMode: "immediate", + fetch: async () => new Response(null, { status: 400 }), + })({ ...pluginInput, client }); + + await event(hooks, "session.created", { info: { id: "session-1" } }); + await flush(); + await hooks.event?.({ + event: { + get type(): never { + throw new Error("poisoned event"); + }, + }, + }); + + await vi.waitFor(() => expect(logged).toHaveLength(2)); + expect(logged.map(({ method, path, body }) => [method, path, body.service, body.level])).toEqual([ + ["POST", "/log", "telemetry-dev", "error"], + ["POST", "/log", "telemetry-dev", "error"], + ]); + expect(logged[0]?.body.message).toContain("telemetry.dev log ingest failed: 400"); + expect(logged[1]?.body.message).toContain("poisoned event"); + expect(consoleSpies.flatMap((spy) => spy.mock.calls)).toEqual([]); +}); + test("tool spans nest under the session's open chat span", async () => { const harness = await makeHarness(); await chatMessage(harness.hooks); diff --git a/packages/otel/README.md b/packages/otel/README.md index f48c6b4..e43886f 100644 --- a/packages/otel/README.md +++ b/packages/otel/README.md @@ -121,7 +121,7 @@ When the SDK and the Claude API send the spans of a request to the same telemetr | `serviceName` | `OTEL_SERVICE_NAME` | `unknown_service`; used on the metrics resource. | | `environment` | `TELEMETRY_DEV_ENVIRONMENT` | `production`; used as `deployment.environment.name` on the metrics resource. | | `fetch` | — | `globalThis.fetch`; override for tests or custom runtimes. | -| `onError` | — | Receives export and processor errors; errors are also reported through diagnostics. | +| `onError` | — | Receives export and processor errors, which diagnostics also get. A series of failed exports is one report. | | `spanExporter` | — | Advanced/test seam replacing the built-in OTLP trace exporter. | ## Missing API key behavior diff --git a/packages/otel/src/index.ts b/packages/otel/src/index.ts index 03ef691..3cc2dd6 100644 --- a/packages/otel/src/index.ts +++ b/packages/otel/src/index.ts @@ -60,9 +60,11 @@ export { } from "./session.ts"; export { + createExportFailures, createLogExporter, createMetricExporter, createTraceExporter, + type ExportFailures, maybeGzip, otlpHeaders, type OtlpTarget, diff --git a/packages/otel/src/otel.ts b/packages/otel/src/otel.ts index bd62375..e53c260 100644 --- a/packages/otel/src/otel.ts +++ b/packages/otel/src/otel.ts @@ -24,6 +24,7 @@ import { } from "./metrics.ts"; import { StampingSpanProcessor } from "./processor.ts"; import { + createExportFailures, createMetricExporter, createTraceExporter, otlpHeaders, @@ -74,6 +75,7 @@ export class TelemetrySpanProcessor implements SpanProcessor { fetchImpl: options.fetch ?? globalThis.fetch, onError: options.onError, exportTimeoutMillis: options.batch?.exportTimeoutMillis, + failures: createExportFailures(options.onError), }; const exporter = diff --git a/packages/otel/src/transport.ts b/packages/otel/src/transport.ts index 2fde519..12c1325 100644 --- a/packages/otel/src/transport.ts +++ b/packages/otel/src/transport.ts @@ -19,6 +19,38 @@ export interface Transport { fetchImpl: typeof fetch; onError?: (error: Error) => void; exportTimeoutMillis?: number; + /** + * Failure state shared by the exporters of one SDK client or `TelemetrySpanProcessor`, so that + * an outage that stops traces, logs, and metrics is one report. Without it, each exporter keeps + * its own state. + */ + failures?: ExportFailures; +} + +export interface ExportFailures { + /** Sends the failure to onError if it is the first failure since an export reached the ingest. */ + report(cause: unknown): void; + /** Records that an export reached the ingest, so that report() sends the next failure. */ + clear(): void; +} + +/** + * Reports the first failed export, then stays quiet until an export reaches the ingest again. + * During an outage every batch fails, and a report for each batch floods the console or onError. + */ +export function createExportFailures(onError?: (error: Error) => void): ExportFailures { + let reported = false; + + return { + report(cause) { + if (reported) return; + reported = true; + reportError(onError, cause); + }, + clear() { + reported = false; + }, + }; } export interface OtlpTarget { @@ -281,6 +313,7 @@ async function postSerialized( const createExportLifecycle = (transport: Transport) => { const inFlight = new Set>(); + const failures = transport.failures ?? createExportFailures(transport.onError); let shutDown = false; const configuredTimeoutMillis = @@ -302,7 +335,7 @@ const createExportLifecycle = (transport: Transport) => { }; const exportBatch = ( - send: (signal: AbortSignal) => Promise, + send: (signal: AbortSignal) => Promise, resultCallback: (result: ExportResult) => void, ) => { if (shutDown) { @@ -325,10 +358,13 @@ const createExportLifecycle = (transport: Transport) => { let tracked: Promise; tracked = Promise.race([Promise.resolve().then(() => send(controller.signal)), timeout]) .then( - () => callback(resultCallback, { code: ExportResultCode.SUCCESS }), + (delivered) => { + if (delivered) failures.clear(); + callback(resultCallback, { code: ExportResultCode.SUCCESS }); + }, (cause: unknown) => { const error = cause instanceof Error ? cause : new Error(String(cause)); - reportError(transport.onError, error); + failures.report(error); callback(resultCallback, { code: ExportResultCode.FAILED, error }); }, ) @@ -357,24 +393,25 @@ interface Serializer { serializeRequest(items: T[]): Uint8Array | undefined; } +/** The sender resolves to true when at least one request reached the ingest. */ function createOtlpBatchSender( serializer: Serializer, target: OtlpTarget, transport: Transport, label: string, -): (items: T[], signal: AbortSignal) => Promise { - const send = async (items: T[], signal: AbortSignal): Promise => { +): (items: T[], signal: AbortSignal) => Promise { + const send = async (items: T[], signal: AbortSignal): Promise => { const body = serializer.serializeRequest(items); - if (!body || body.byteLength === 0) return; + if (!body || body.byteLength === 0) return false; if (body.byteLength > MAX_BODY_BYTES) { if (items.length > 1) { const mid = Math.ceil(items.length / 2); - await send(items.slice(0, mid), signal); - await send(items.slice(mid), signal); + const first = await send(items.slice(0, mid), signal); + const second = await send(items.slice(mid), signal); - return; + return first || second; } // A single record beyond the limit can never be accepted; drop it instead of wedging the batch. @@ -383,10 +420,12 @@ function createOtlpBatchSender( new Error(`telemetry.dev: ${label} record exceeds max export size, dropped`), ); - return; + return false; } await postSerialized(body, target, transport, label, signal); + + return true; }; return send; @@ -444,9 +483,10 @@ export function createMetricExporter(target: OtlpTarget, transport: Transport): ? ProtobufMetricsSerializer.serializeRequest(resourceMetrics) : undefined; - if (body !== undefined && body.byteLength > 0) { - await postSerialized(body, target, transport, "metric", signal); - } + if (body === undefined || body.byteLength === 0) return false; + await postSerialized(body, target, transport, "metric", signal); + + return true; }, resultCallback); }, selectAggregationTemporality: () => AggregationTemporality.DELTA, diff --git a/packages/otel/tests/otel-byo.test.ts b/packages/otel/tests/otel-byo.test.ts index 346d581..a2839a9 100644 --- a/packages/otel/tests/otel-byo.test.ts +++ b/packages/otel/tests/otel-byo.test.ts @@ -124,6 +124,36 @@ test("without an api key or exporter override the processor is a silent no-op", } }); +test("an outage that stops traces and metrics reaches onError once", async () => { + const errors: Error[] = []; + const urls: string[] = []; + + const processor = new TelemetrySpanProcessor({ + apiKey: "td_live_x", + baseUrl: "https://ingest.example", + exportMode: "immediate", + onError: (error) => errors.push(error), + fetch: (input) => { + urls.push(input instanceof URL ? input.href : input instanceof Request ? input.url : input); + + return Promise.resolve(new Response(null, { status: 400 })); + }, + }); + + const provider = new BasicTracerProvider({ spanProcessors: [processor] }); + const span = provider.getTracer("user-app").startSpan("chat"); + span.setAttribute("gen_ai.operation.name", "chat"); + span.end(); + // Shutdown waits for the span export and sends the metrics. + await provider.shutdown(); + + expect(urls.sort()).toEqual([ + "https://ingest.example/v1/metrics", + "https://ingest.example/v1/traces", + ]); + expect(errors).toHaveLength(1); +}); + test("createTelemetrySpanExporter is a no-op without a key and real otherwise", async () => { vi.stubEnv("TELEMETRY_DEV_API_KEY", undefined); const calls: Array<{ url: string; method?: string }> = []; diff --git a/packages/otel/tests/transport.test.ts b/packages/otel/tests/transport.test.ts index a93e1ee..44de745 100644 --- a/packages/otel/tests/transport.test.ts +++ b/packages/otel/tests/transport.test.ts @@ -137,9 +137,9 @@ test("retries on 429/503 then succeeds", async () => { expect(calls[0]!.headers["x-telemetry-dev-sdk"]).toBe("@telemetry-dev/otel"); }); -test("does not retry non-retryable statuses and reports the failure", async () => { +test("does not retry non-retryable statuses and reports a series of failures once", async () => { const errors: unknown[] = []; - const { calls, fetchImpl } = makeFetch([400]); + const { calls, fetchImpl } = makeFetch([400, 400, 200, 400]); const exporter = createTraceExporter( { url: "https://ingest.example/v1/traces", headers: otlpHeaders("td_live_x") }, @@ -149,7 +149,11 @@ test("does not retry non-retryable statuses and reports the failure", async () = const code = await exportSpans(exporter, makeSpans(1)); expect(code).toBe(ExportResultCode.FAILED); expect(calls).toHaveLength(1); + expect(await exportSpans(exporter, makeSpans(1))).toBe(ExportResultCode.FAILED); expect(errors).toHaveLength(1); + expect(await exportSpans(exporter, makeSpans(1))).toBe(ExportResultCode.SUCCESS); + expect(await exportSpans(exporter, makeSpans(1))).toBe(ExportResultCode.FAILED); + expect(errors).toHaveLength(2); }); test("exhausted retries surface as a failed export with onError", async () => { diff --git a/packages/pi/README.md b/packages/pi/README.md index 398561f..1ce1f83 100644 --- a/packages/pi/README.md +++ b/packages/pi/README.md @@ -55,6 +55,20 @@ export default telemetryDevExtension({ Initialization is one-shot; the first extension factory initialized in a process supplies the SDK options. +## Errors + +The integration does not write to the terminal, because pi draws its TUI there. It appends SDK and +extension errors to `telemetry-dev.log` in the pi agent directory: `~/.pi/agent`, or +`PI_CODING_AGENT_DIR` when you set it. The file has mode `0600`. When the file is more than 5 MB, +the integration moves it to `telemetry-dev.log.1`. + +A series of failed exports, for example during an ingest outage, is one entry. The next failure +after an export reaches the ingest is a new entry. + +To receive the errors in your code, pass `onError` to `telemetryDevExtension()`. The callback +replaces the log file. To also write SDK diagnostics to the console, pass `logLevel`, for example +`"warn"`. + ## Trace shape A typical session produces this hierarchy: diff --git a/packages/pi/src/error-log.ts b/packages/pi/src/error-log.ts new file mode 100644 index 0000000..203e368 --- /dev/null +++ b/packages/pi/src/error-log.ts @@ -0,0 +1,35 @@ +import { appendFileSync, renameSync, statSync } from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { inspect } from "node:util"; + +/** pi moves its own `mcp.log` to `mcp.log.1` when the file is more than 5 MB. */ +const MAX_LOG_BYTES = 5 * 1024 * 1024; + +/** + * Appends an error to `telemetry-dev.log` in pi's agent directory. pi draws its TUI on the + * terminal, so an error written to stderr garbles the screen and is not kept. + */ +export function writeErrorLog(error: unknown): void { + const path = join(agentDir(), "telemetry-dev.log"); + const entry = `${new Date().toISOString()} ${inspect(error).replaceAll("\n", "\n ")}\n`; + + try { + if ((statSync(path, { throwIfNoEntry: false })?.size ?? 0) > MAX_LOG_BYTES) { + renameSync(path, `${path}.1`); + } + + appendFileSync(path, entry, { mode: 0o600 }); + } catch { + // The log is best effort; telemetry must never fail a pi session. + } +} + +/** pi's agent directory: `PI_CODING_AGENT_DIR`, else `~/.pi/agent`. */ +function agentDir(): string { + const dir = process.env.PI_CODING_AGENT_DIR; + + if (!dir) return join(homedir(), ".pi", "agent"); + + return dir === "~" || dir.startsWith("~/") ? join(homedir(), dir.slice(1)) : dir; +} diff --git a/packages/pi/src/extension.ts b/packages/pi/src/extension.ts index fb35c0f..d6de419 100644 --- a/packages/pi/src/extension.ts +++ b/packages/pi/src/extension.ts @@ -4,16 +4,22 @@ import { startSpan, type LogLevel, type LogOptions, + type SdkLogLevel, type SpanHandle, type TokenUsage, } from "@telemetry-dev/sdk"; import { ensureInit, type ClientOverrides, type TelemetryDevPiOptions } from "./config.ts"; +import { writeErrorLog } from "./error-log.ts"; /** Options for {@link telemetryDevExtension}. */ export interface TelemetryDevExtensionOptions extends TelemetryDevPiOptions { /** Value for `gen_ai.agent.name` on spans and logs. Defaults to `"pi"`. */ agentName?: string; + /** SDK console output. Defaults to `"silent"`, because pi draws its TUI on the terminal. */ + logLevel?: SdkLogLevel; + /** Receives SDK and extension errors. By default, the extension appends them to `telemetry-dev.log` in pi's agent directory. */ + onError?: (cause: Error) => void; } /** @@ -148,11 +154,13 @@ export function telemetryDevExtension( overrides?: ClientOverrides, ): TelemetryDevExtension { const { agentName = "pi", ...sdkOptions } = options; - const onError = sdkOptions.onError; + // pi draws its TUI on the terminal, so errors go to a log file and not to the console. + const onError = sdkOptions.onError ?? writeErrorLog; + const logLevel = sdkOptions.logLevel ?? "silent"; return (pi) => { try { - ensureInit(sdkOptions, overrides); + ensureInit({ ...sdkOptions, logLevel, onError }, overrides); } catch (error) { reportError(onError, error); } diff --git a/packages/pi/tests/pi.test.ts b/packages/pi/tests/pi.test.ts index b957b0f..a0dddf6 100644 --- a/packages/pi/tests/pi.test.ts +++ b/packages/pi/tests/pi.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -737,6 +737,48 @@ test("malformed events never throw and report through onError", async () => { expect(harness.exporters.spans.getFinishedSpans().length).toBeGreaterThan(0); }); +test("errors go to a private, rotated telemetry-dev.log in the agent directory, not to the console", async () => { + const directory = await mkdtemp(join(tmpdir(), "telemetry-pi-agent-")); + const path = join(directory, "telemetry-dev.log"); + vi.stubEnv("PI_CODING_AGENT_DIR", directory); + const consoleSpies = (["error", "warn", "info", "debug", "log"] as const).map((method) => + vi.spyOn(console, method).mockImplementation(() => {}), + ); + + try { + // An earlier log of more than 5 MB moves to telemetry-dev.log.1. + await writeFile(path, "x".repeat(5 * 1024 * 1024 + 1)); + const handlers = new Map unknown>(); + telemetryDevExtension({ + apiKey: "td_live_test", + exportMode: "immediate", + fetch: async () => new Response(null, { status: 400 }), + })({ on: (type: string, handler: never) => handlers.set(type, handler) }); + + await handlers.get("session_start")?.({ reason: "startup" }, makeContext()); + await flush(); + await handlers.get("message_end")?.( + { + get message(): never { + throw new Error("poisoned event"); + }, + }, + makeContext(), + ); + await resetForTesting(); + + const log = await readFile(path, "utf8"); + expect(log).toContain("telemetry.dev log ingest failed: 400"); + expect(log).toContain("poisoned event"); + expect((await stat(`${path}.1`)).size).toBe(5 * 1024 * 1024 + 1); + expect((await stat(path)).mode & 0o777).toBe(0o600); + expect(consoleSpies.flatMap((spy) => spy.mock.calls)).toEqual([]); + } finally { + vi.unstubAllEnvs(); + await rm(directory, { recursive: true, force: true }); + } +}); + test("each event reads the current session id", async () => { const harness = makeHarness(); await harness.emit({ type: "session_start", reason: "startup" }, makeContext("session-2")); diff --git a/packages/sdk/src/client.ts b/packages/sdk/src/client.ts index 11b2bed..14e65b5 100644 --- a/packages/sdk/src/client.ts +++ b/packages/sdk/src/client.ts @@ -19,6 +19,7 @@ import { als, AlsContextManager, BATCHED_METRIC_INTERVAL_MS, + createExportFailures, createLogExporter, createMetricExporter, createMetricsPipeline, @@ -156,7 +157,10 @@ function initInner(options?: TelemetryOptions, overrides?: ClientOverrides): Tel ...config.resourceAttributes, }); - const transport: Transport = { fetchImpl: config.fetchImpl, onError: config.onError }; + // A failed flush repeats the failure of its export, so the flush and the exporters share one + // failure state. + const failures = createExportFailures(config.onError); + const transport: Transport = { fetchImpl: config.fetchImpl, onError: config.onError, failures }; const headers = config.apiKey ? otlpHeaders(config.apiKey, config.sdkName) : undefined; const spanExporter = @@ -274,7 +278,7 @@ function initInner(options?: TelemetryOptions, overrides?: ClientOverrides): Tel return Promise.all(parts) .then(() => undefined) - .catch((error: Error) => reportError(config.onError, error)); + .catch((error: Error) => failures.report(error)); }; const handoff = (p: Promise): Promise => { diff --git a/packages/sdk/src/config.ts b/packages/sdk/src/config.ts index 1b5ce94..d1378c2 100644 --- a/packages/sdk/src/config.ts +++ b/packages/sdk/src/config.ts @@ -63,7 +63,7 @@ export interface TelemetryOptions { fetch?: typeof fetch; /** Serverless extender (e.g. Cloudflare `ctx.waitUntil`). When provided, `flush()` does not await exports. */ waitUntil?: (p: Promise) => void; - /** Receives errors raised while emitting telemetry; repeated instrumentation failures from one stream may be coalesced into one report. The SDK never throws into user code. */ + /** Receives errors raised while emitting telemetry; repeated instrumentation failures from one stream may be coalesced into one report. A series of failed exports is reported once, and the next failure after an export reaches the ingest is reported again. The SDK never throws into user code. */ onError?: (cause: Error) => void; /** Header value for a custom telemetry.dev integration. */ sdkName?: string; diff --git a/packages/sdk/tests/client.test.ts b/packages/sdk/tests/client.test.ts index 095b4b8..644075f 100644 --- a/packages/sdk/tests/client.test.ts +++ b/packages/sdk/tests/client.test.ts @@ -348,6 +348,57 @@ test("OTLP requests identify the calling SDK package", async () => { expect(requests[0]?.headers.get("x-telemetry-dev-sdk")).toBe("@telemetry-dev/cursor"); }); +test("an ingest outage reaches onError once until an export gets through again", async () => { + vi.useFakeTimers(); + + try { + let ingestUp = false; + const requests: string[] = []; + const errors: Error[] = []; + + init({ + apiKey: "td_live_test", + baseUrl: "https://ingest.test", + logLevel: "silent", + onError: (error) => errors.push(error), + fetch: async (input) => { + requests.push(new URL(new Request(input).url).pathname); + + if (!ingestUp) throw new TypeError("fetch failed"); + + return new Response(null, { status: 200 }); + }, + }); + + // The transport tries a failed request three times, with 600 ms of delays in total. + const flushThroughRetries = async () => { + const flushed = flush(); + await vi.advanceTimersByTimeAsync(1_000); + await flushed; + }; + + startSpan("first outage", { type: "tool" }).end(); + log("first outage"); + await flushThroughRetries(); + log("still in the first outage"); + await flushThroughRetries(); + + expect(new Set(requests)).toEqual(new Set(["/v1/traces", "/v1/logs", "/v1/metrics"])); + expect(errors.map((error) => error.message)).toEqual(["fetch failed"]); + + ingestUp = true; + startSpan("ingest is back", { type: "tool" }).end(); + await flushThroughRetries(); + ingestUp = false; + startSpan("second outage", { type: "tool" }).end(); + await flushThroughRetries(); + + expect(errors.map((error) => error.message)).toEqual(["fetch failed", "fetch failed"]); + } finally { + vi.useRealTimers(); + } +}); + test("captureMasked reports whether the active client has a mask", () => { expect(captureMasked()).toBe(false); diff --git a/packages/sdk/tests/logs.test.ts b/packages/sdk/tests/logs.test.ts index e0310c9..c8aacbd 100644 --- a/packages/sdk/tests/logs.test.ts +++ b/packages/sdk/tests/logs.test.ts @@ -1,6 +1,6 @@ -import { afterEach, expect, test } from "vitest"; +import { afterEach, expect, test, vi } from "vitest"; -import { flush, log, shutdown, startActiveSpan } from "../src/index.ts"; +import { flush, init, log, shutdown, startActiveSpan } from "../src/index.ts"; import { setup } from "./helpers.ts"; afterEach(async () => { @@ -93,3 +93,50 @@ test("log records carry the resource", async () => { expect(record.resource.attributes["service.name"]).toBe("svc"); expect(record.resource.attributes["deployment.environment.name"]).toBe("test"); }); + +test("logs keep reaching /v1/logs after a flush starts during a scheduled export", async () => { + vi.useFakeTimers(); + + try { + const bodies: string[] = []; + const errors: Error[] = []; + let release: (() => void) | undefined; + + init({ + apiKey: "td_live_test", + baseUrl: "https://ingest.test", + logLevel: "silent", + onError: (error) => errors.push(error), + fetch: async (input, request) => { + if (!new Request(input).url.endsWith("/v1/logs")) + return new Response(null, { status: 200 }); + bodies.push(new TextDecoder().decode(request?.body as Uint8Array)); + + if (bodies.length === 1) { + await new Promise((resolve) => { + release = resolve; + }); + } + + return new Response(null, { status: 200 }); + }, + }); + + log("scheduled record"); + // The batch log processor exports 1000 ms after the first record. + await vi.advanceTimersByTimeAsync(1000); + expect(release).toBeDefined(); + const overlapping = flush(); + release?.(); + await overlapping; + log("record after the flush"); + await flush(); + + expect(errors).toEqual([]); + expect(bodies).toHaveLength(2); + expect(bodies[0]).toContain("scheduled record"); + expect(bodies[1]).toContain("record after the flush"); + } finally { + vi.useRealTimers(); + } +});