feat(telemetry): added otlp log and metric export
Extended the OTLP export bootstrap beyond traces so omp emits the full OpenTelemetry signal set from a single session. - Registered a LoggerProvider + BatchLogRecordProcessor and a MeterProvider + PeriodicExportingMetricReader when their OTLP endpoints (or the shared endpoint) are set, each gated independently by OTEL_*_EXPORTER=none, OTEL_SDK_DISABLED, and http/protobuf protocol checks. - Bridged the centralized logger through a new registerLogSink API so every log event also becomes an OTLP log record with severity, attributes, and active span context for log-trace correlation (min level via OTEL_LOG_LEVEL). - Recorded gen_ai.client.token.usage and pi.omp.agent.* metrics from the agent run summary and per-chat usage hooks, and emitted a structured run summary log event. - Added an out-of-process logs+metrics probe and gating tests covering the per-signal kill switches. Fixes #4604
This commit is contained in:
roboomp
@@ -89,9 +89,14 @@
|
||||
"@oh-my-pi/pi-wire": "catalog:",
|
||||
"@oh-my-pi/snapcompact": "catalog:",
|
||||
"@opentelemetry/api": "catalog:",
|
||||
"@opentelemetry/api-logs": "catalog:",
|
||||
"@opentelemetry/context-async-hooks": "catalog:",
|
||||
"@opentelemetry/exporter-logs-otlp-proto": "catalog:",
|
||||
"@opentelemetry/exporter-metrics-otlp-proto": "catalog:",
|
||||
"@opentelemetry/exporter-trace-otlp-proto": "catalog:",
|
||||
"@opentelemetry/resources": "catalog:",
|
||||
"@opentelemetry/sdk-logs": "catalog:",
|
||||
"@opentelemetry/sdk-metrics": "catalog:",
|
||||
"@opentelemetry/sdk-trace-base": "catalog:",
|
||||
"@opentelemetry/sdk-trace-node": "catalog:",
|
||||
"@puppeteer/browsers": "catalog:",
|
||||
@@ -350,9 +355,14 @@
|
||||
"@oh-my-pi/pi-wire": "16.3.6",
|
||||
"@oh-my-pi/snapcompact": "16.3.6",
|
||||
"@opentelemetry/api": "^1.9.1",
|
||||
"@opentelemetry/api-logs": "^0.218.0",
|
||||
"@opentelemetry/context-async-hooks": "^2.7.1",
|
||||
"@opentelemetry/exporter-logs-otlp-proto": "^0.218.0",
|
||||
"@opentelemetry/exporter-metrics-otlp-proto": "^0.218.0",
|
||||
"@opentelemetry/exporter-trace-otlp-proto": "^0.218.0",
|
||||
"@opentelemetry/resources": "^2.7.1",
|
||||
"@opentelemetry/sdk-logs": "^0.218.0",
|
||||
"@opentelemetry/sdk-metrics": "^2.7.1",
|
||||
"@opentelemetry/sdk-trace-base": "^2.7.1",
|
||||
"@opentelemetry/sdk-trace-node": "^2.7.1",
|
||||
"@puppeteer/browsers": "^3.0.4",
|
||||
@@ -772,6 +782,12 @@
|
||||
|
||||
"@opentelemetry/core": ["@opentelemetry/core@2.8.0", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-hd1Lfh8p545nNz+jq1Ejfz+Mn1hyLuxYn1YzTfFNrxr8urEWMNQLPf1Th8kjOH+HxwawCrtgBp8JpBUR4ZSgww=="],
|
||||
|
||||
"@opentelemetry/exporter-logs-otlp-proto": ["@opentelemetry/exporter-logs-otlp-proto@0.218.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.218.0", "@opentelemetry/core": "2.7.1", "@opentelemetry/otlp-exporter-base": "0.218.0", "@opentelemetry/otlp-transformer": "0.218.0", "@opentelemetry/resources": "2.7.1", "@opentelemetry/sdk-logs": "0.218.0", "@opentelemetry/sdk-trace-base": "2.7.1" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-1/noQNsp9gXD75HPzgjBrcF1+XTtry7pFAUfxVEJgg7mPv2AawKQuYkhMmJ8qjxz4Ubc3Y8bwvfxevXsKTq4cg=="],
|
||||
|
||||
"@opentelemetry/exporter-metrics-otlp-http": ["@opentelemetry/exporter-metrics-otlp-http@0.218.0", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/otlp-exporter-base": "0.218.0", "@opentelemetry/otlp-transformer": "0.218.0", "@opentelemetry/resources": "2.7.1", "@opentelemetry/sdk-metrics": "2.7.1" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-bV7d2OuMpZu2+gAaxUAhzfZ0h3WVZk8ETQUEE3DNSntbTaMpuITjtm8I0rNyHFdm7Ax57K6ty7SgFXlBmOLIvQ=="],
|
||||
|
||||
"@opentelemetry/exporter-metrics-otlp-proto": ["@opentelemetry/exporter-metrics-otlp-proto@0.218.0", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/exporter-metrics-otlp-http": "0.218.0", "@opentelemetry/otlp-exporter-base": "0.218.0", "@opentelemetry/otlp-transformer": "0.218.0", "@opentelemetry/resources": "2.7.1", "@opentelemetry/sdk-metrics": "2.7.1" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-ubLddKjWULhla9YZRCj/rTBeppjJYE4e9w0icx5mTu3eFhWjQzbV75NYjXuIlEG+NJsBl6d+sTFw5Qu+oej4oQ=="],
|
||||
|
||||
"@opentelemetry/exporter-trace-otlp-proto": ["@opentelemetry/exporter-trace-otlp-proto@0.218.0", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/otlp-exporter-base": "0.218.0", "@opentelemetry/otlp-transformer": "0.218.0", "@opentelemetry/resources": "2.7.1", "@opentelemetry/sdk-trace-base": "2.7.1" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-r1Msf8SNLRmwh9J6XQ5uh82D7CdDWMNHnPB7LAVHjzut0TkSeKc5KcIvr4SvHvfk/xwN5gxC+VLKQ1k0o8PSPw=="],
|
||||
|
||||
"@opentelemetry/otlp-exporter-base": ["@opentelemetry/otlp-exporter-base@0.218.0", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/otlp-transformer": "0.218.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-ZwqpkNL5W7RyGJPDZ9g06DvKp8KFTWPJPN12anpMQYSKpTSU0z3EIZuPq9vPGpS8siFyOqDYDAuCwlNO9FqgbA=="],
|
||||
@@ -1468,6 +1484,20 @@
|
||||
|
||||
"@isaacs/fs-minipass/minipass": ["minipass@7.1.3", "", {}, "sha512-tEBHqDnIoM/1rXME1zgka9g6Q2lcoCkxHLuc7ODJ5BxbP5d4c2Z5cGgtXAku59200Cx7diuHTOYfSBD8n6mm8A=="],
|
||||
|
||||
"@opentelemetry/exporter-logs-otlp-proto/@opentelemetry/core": ["@opentelemetry/core@2.7.1", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-QAqIj32AtK6+pEVNG7EOVxHdE06RP+FM5qpiEJ4RtDcFIqKUZHYhl7/7UY5efhwmwNAg7j8QbJVBLxMerc0+gw=="],
|
||||
|
||||
"@opentelemetry/exporter-logs-otlp-proto/@opentelemetry/resources": ["@opentelemetry/resources@2.7.1", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-DeT6KKolmC4e/dRQvMQ/RwlnzhaqeiFOXY5ngoOPJ07GgVVKxZOg9EcrNZb5aTzUn+iCrJldAgOfQm1O/QfPAQ=="],
|
||||
|
||||
"@opentelemetry/exporter-logs-otlp-proto/@opentelemetry/sdk-trace-base": ["@opentelemetry/sdk-trace-base@2.7.1", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/resources": "2.7.1", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-NAYIlsF8MPUsKqJMiDQJTMPOmlbawC1Iz/omMLygZ1C9am8fTKYjTaI+OZM+WTY3t3Glo0wnOg/6/pac6RGPPw=="],
|
||||
|
||||
"@opentelemetry/exporter-metrics-otlp-http/@opentelemetry/core": ["@opentelemetry/core@2.7.1", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-QAqIj32AtK6+pEVNG7EOVxHdE06RP+FM5qpiEJ4RtDcFIqKUZHYhl7/7UY5efhwmwNAg7j8QbJVBLxMerc0+gw=="],
|
||||
|
||||
"@opentelemetry/exporter-metrics-otlp-http/@opentelemetry/resources": ["@opentelemetry/resources@2.7.1", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-DeT6KKolmC4e/dRQvMQ/RwlnzhaqeiFOXY5ngoOPJ07GgVVKxZOg9EcrNZb5aTzUn+iCrJldAgOfQm1O/QfPAQ=="],
|
||||
|
||||
"@opentelemetry/exporter-metrics-otlp-proto/@opentelemetry/core": ["@opentelemetry/core@2.7.1", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-QAqIj32AtK6+pEVNG7EOVxHdE06RP+FM5qpiEJ4RtDcFIqKUZHYhl7/7UY5efhwmwNAg7j8QbJVBLxMerc0+gw=="],
|
||||
|
||||
"@opentelemetry/exporter-metrics-otlp-proto/@opentelemetry/resources": ["@opentelemetry/resources@2.7.1", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-DeT6KKolmC4e/dRQvMQ/RwlnzhaqeiFOXY5ngoOPJ07GgVVKxZOg9EcrNZb5aTzUn+iCrJldAgOfQm1O/QfPAQ=="],
|
||||
|
||||
"@opentelemetry/exporter-trace-otlp-proto/@opentelemetry/core": ["@opentelemetry/core@2.7.1", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-QAqIj32AtK6+pEVNG7EOVxHdE06RP+FM5qpiEJ4RtDcFIqKUZHYhl7/7UY5efhwmwNAg7j8QbJVBLxMerc0+gw=="],
|
||||
|
||||
"@opentelemetry/exporter-trace-otlp-proto/@opentelemetry/resources": ["@opentelemetry/resources@2.7.1", "", { "dependencies": { "@opentelemetry/core": "2.7.1", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-DeT6KKolmC4e/dRQvMQ/RwlnzhaqeiFOXY5ngoOPJ07GgVVKxZOg9EcrNZb5aTzUn+iCrJldAgOfQm1O/QfPAQ=="],
|
||||
|
||||
@@ -38,9 +38,14 @@
|
||||
"@oh-my-pi/pi-wire": "16.3.6",
|
||||
"@oh-my-pi/snapcompact": "16.3.6",
|
||||
"@opentelemetry/api": "^1.9.1",
|
||||
"@opentelemetry/api-logs": "^0.218.0",
|
||||
"@opentelemetry/context-async-hooks": "^2.7.1",
|
||||
"@opentelemetry/exporter-logs-otlp-proto": "^0.218.0",
|
||||
"@opentelemetry/exporter-metrics-otlp-proto": "^0.218.0",
|
||||
"@opentelemetry/exporter-trace-otlp-proto": "^0.218.0",
|
||||
"@opentelemetry/resources": "^2.7.1",
|
||||
"@opentelemetry/sdk-logs": "^0.218.0",
|
||||
"@opentelemetry/sdk-metrics": "^2.7.1",
|
||||
"@opentelemetry/sdk-trace-base": "^2.7.1",
|
||||
"@opentelemetry/sdk-trace-node": "^2.7.1",
|
||||
"@puppeteer/browsers": "^3.0.4",
|
||||
|
||||
@@ -2,6 +2,10 @@
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
|
||||
- Added OpenTelemetry log and metric export alongside the existing trace export. When `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` (or the shared `OTEL_EXPORTER_OTLP_ENDPOINT`) is set, `omp` registers a `LoggerProvider` and forwards every centralized-logger event as an OTLP log record (severity + attributes + active span context for log↔trace correlation, min level via `OTEL_LOG_LEVEL`, plus a structured `agent run completed` summary event). When `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` (or the shared endpoint) is set, it registers a `MeterProvider` with a `PeriodicExportingMetricReader` and records GenAI-semconv `gen_ai.client.token.usage` plus `pi.omp.agent.*` counters/histograms (runs, steps, chat/tool calls by name+status+finish reason, latencies, estimated cost, errors) from the agent run summary and per-chat usage hooks. Each signal honors its own `OTEL_*_EXPORTER=none` kill switch, the global `OTEL_SDK_DISABLED`, and declines non-`http/protobuf` protocols independently ([#4604](https://github.com/can1357/oh-my-pi/issues/4604)).
|
||||
|
||||
## [16.3.6] - 2026-07-04
|
||||
|
||||
### Changed
|
||||
|
||||
@@ -68,9 +68,14 @@
|
||||
"@oh-my-pi/pi-wire": "catalog:",
|
||||
"@oh-my-pi/snapcompact": "catalog:",
|
||||
"@opentelemetry/api": "catalog:",
|
||||
"@opentelemetry/api-logs": "catalog:",
|
||||
"@opentelemetry/context-async-hooks": "catalog:",
|
||||
"@opentelemetry/exporter-logs-otlp-proto": "catalog:",
|
||||
"@opentelemetry/exporter-metrics-otlp-proto": "catalog:",
|
||||
"@opentelemetry/exporter-trace-otlp-proto": "catalog:",
|
||||
"@opentelemetry/resources": "catalog:",
|
||||
"@opentelemetry/sdk-logs": "catalog:",
|
||||
"@opentelemetry/sdk-metrics": "catalog:",
|
||||
"@opentelemetry/sdk-trace-base": "catalog:",
|
||||
"@opentelemetry/sdk-trace-node": "catalog:",
|
||||
"@puppeteer/browsers": "catalog:",
|
||||
|
||||
@@ -73,7 +73,7 @@ import { executeBuiltinSlashCommand } from "./slash-commands/builtin-registry";
|
||||
import { shouldShowStartupSplash } from "./startup-splash";
|
||||
import { discoverTitleSystemPromptFile, resolvePromptInput } from "./system-prompt";
|
||||
import { createPersistedSubagentReviverFactory } from "./task/persisted-revive";
|
||||
import { initTelemetryExport, isTelemetryExportEnabled } from "./telemetry-export";
|
||||
import { createTelemetryExportConfig, initTelemetryExport, isTelemetryExportEnabled } from "./telemetry-export";
|
||||
import { concreteThinkingLevel, parseConfiguredThinkingLevel } from "./thinking";
|
||||
import type { LspStartupServerInfo } from "./tools";
|
||||
import {
|
||||
@@ -1249,15 +1249,13 @@ export async function runRootCommand(
|
||||
sessionOptions.hasUI = isInteractive || mode === "rpc-ui";
|
||||
sessionOptions.settings = settingsInstance;
|
||||
|
||||
// OTEL: register the global OTLP trace exporter when an OTLP endpoint is
|
||||
// configured via env, then switch on the agent loop's telemetry so its
|
||||
// GenAI spans (invoke_agent / chat / execute_tool) are actually emitted.
|
||||
// Both are no-ops when OTEL_EXPORTER_OTLP_ENDPOINT is unset. An empty config
|
||||
// is enough to enable telemetry — content capture is governed by the
|
||||
// standard OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT env var.
|
||||
// OTEL: register global OTLP exporters when an endpoint is configured via
|
||||
// env, then switch on the agent loop's telemetry hooks so traces, run-level
|
||||
// metrics, and structured logs have source events to export. Content capture
|
||||
// remains governed by OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT.
|
||||
await logger.time("initTelemetryExport", initTelemetryExport);
|
||||
if (isTelemetryExportEnabled()) {
|
||||
sessionOptions.telemetry = {};
|
||||
sessionOptions.telemetry = createTelemetryExportConfig(sessionOptions.telemetry);
|
||||
}
|
||||
|
||||
// Handle CLI --api-key as runtime override (not persisted)
|
||||
|
||||
@@ -1,144 +1,505 @@
|
||||
/**
|
||||
* OTLP trace export bootstrap.
|
||||
* OTLP telemetry export bootstrap.
|
||||
*
|
||||
* oh-my-pi's agent core (`@oh-my-pi/pi-agent-core`) emits OpenTelemetry GenAI
|
||||
* spans through the global `@opentelemetry/api` tracer, but only when a
|
||||
* TracerProvider is registered in the process — otherwise the API returns a
|
||||
* no-op tracer and the spans are silently dropped. The shipped CLI never
|
||||
* registered one, so headless / embedded hosts (e.g. an ACP harness that
|
||||
* spawns `omp` as a child process) had no way to collect omp's internal traces.
|
||||
* spans through the global `@opentelemetry/api` tracer, and exposes run-level
|
||||
* callbacks for metrics/log pipelines. This module registers the OTLP/proto
|
||||
* trace, log, and metric SDK providers when the standard `OTEL_*` endpoint env
|
||||
* vars are set so `omp` can be observed by any OTLP collector without vendor
|
||||
* coupling.
|
||||
*
|
||||
* This module registers a NodeTracerProvider with an OTLP/proto exporter when
|
||||
* the standard `OTEL_EXPORTER_OTLP_ENDPOINT` (or `..._TRACES_ENDPOINT`) env var
|
||||
* is set, following the zero-code OTEL env contract: the exporter reads its
|
||||
* endpoint, headers, and timeout from `OTEL_EXPORTER_OTLP_*` itself. The
|
||||
* consuming process configures the destination entirely through env; omp stays
|
||||
* provider-agnostic and ships no vendor coupling. Only the `http/protobuf`
|
||||
* transport is supported — an `OTEL_EXPORTER_OTLP*_PROTOCOL` of `grpc` or
|
||||
* `http/json` declines rather than misrouting spans.
|
||||
*
|
||||
* The OTLP/proto exporter on the 2.x line is used deliberately: the 1.x line
|
||||
* deadlocks under Bun — its `req.on('close')` handler fires a spurious failure
|
||||
* after the success path. `exporter-trace-otlp-proto@0.218` paired with
|
||||
* `sdk-trace-base@2.7` exports cleanly on Bun.
|
||||
* Only the `http/protobuf` transport is supported — an
|
||||
* `OTEL_EXPORTER_OTLP*_PROTOCOL` of `grpc` or `http/json` declines rather than
|
||||
* misrouting protobuf payloads. The exporter line is pinned to the 0.218/2.7
|
||||
* family validated under Bun; the 1.x OTLP line deadlocks when its
|
||||
* `req.on("close")` handler fires after a successful export.
|
||||
*/
|
||||
import type {
|
||||
AgentRunCoverage,
|
||||
AgentRunSummary,
|
||||
AgentTelemetryConfig,
|
||||
AgentTelemetryWarning,
|
||||
ChatUsageEvent,
|
||||
ToolStatus,
|
||||
} from "@oh-my-pi/pi-agent-core";
|
||||
import { logger, postmortem } from "@oh-my-pi/pi-utils";
|
||||
import type * as TraceNode from "@opentelemetry/sdk-trace-node";
|
||||
import {
|
||||
type Attributes,
|
||||
type AttributeValue,
|
||||
type Counter,
|
||||
context,
|
||||
type Histogram,
|
||||
type Meter,
|
||||
metrics,
|
||||
} from "@opentelemetry/api";
|
||||
import { type LogAttributes, logs, type Logger as OtelLogger, SeverityNumber } from "@opentelemetry/api-logs";
|
||||
import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks";
|
||||
import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-proto";
|
||||
import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-proto";
|
||||
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";
|
||||
import { resourceFromAttributes } from "@opentelemetry/resources";
|
||||
import { BatchLogRecordProcessor, LoggerProvider } from "@opentelemetry/sdk-logs";
|
||||
import { MeterProvider, PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
|
||||
import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-base";
|
||||
import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
|
||||
|
||||
/**
|
||||
* Periodic flush interval. A long-lived `omp` process (the ACP server is
|
||||
* spawned once and reused across many turns) would otherwise hold finished
|
||||
* spans until the batch window elapses or the process exits.
|
||||
* telemetry until a batch window elapses or the process exits.
|
||||
*/
|
||||
const FLUSH_INTERVAL_MS = 30_000;
|
||||
|
||||
let provider: TraceNode.NodeTracerProvider | undefined;
|
||||
const SERVICE_NAME = "oh-my-pi";
|
||||
|
||||
type TelemetrySignal = "trace" | "log" | "metric";
|
||||
type OtelLogLevel = "none" | logger.LogLevel;
|
||||
|
||||
interface SignalConfig {
|
||||
readonly trace: boolean;
|
||||
readonly log: boolean;
|
||||
readonly metric: boolean;
|
||||
}
|
||||
|
||||
const LOG_SEVERITY: Record<logger.LogLevel, SeverityNumber> = {
|
||||
error: SeverityNumber.ERROR,
|
||||
warn: SeverityNumber.WARN,
|
||||
info: SeverityNumber.INFO,
|
||||
debug: SeverityNumber.DEBUG,
|
||||
};
|
||||
|
||||
const LOG_LEVEL_WEIGHT: Record<logger.LogLevel, number> = {
|
||||
error: 0,
|
||||
warn: 1,
|
||||
info: 2,
|
||||
debug: 3,
|
||||
};
|
||||
|
||||
const TOOL_STATUSES = ["ok", "error", "skipped", "blocked", "timeout", "aborted"] satisfies readonly ToolStatus[];
|
||||
|
||||
let traceProvider: NodeTracerProvider | undefined;
|
||||
let logProvider: LoggerProvider | undefined;
|
||||
let meterProvider: MeterProvider | undefined;
|
||||
let metricRecorder: AgentMetricRecorder | undefined;
|
||||
let otelLogger: OtelLogger | undefined;
|
||||
let unregisterLogSink: (() => void) | undefined;
|
||||
let initPromise: Promise<void> | undefined;
|
||||
|
||||
/**
|
||||
* Whether {@link initTelemetryExport} registered a real provider. The CLI uses
|
||||
* this to decide whether to switch on the agent loop's telemetry config — there
|
||||
* is no point emitting spans into a no-op tracer.
|
||||
* Whether {@link initTelemetryExport} registered any real OTLP signal provider.
|
||||
* The CLI uses this to decide whether to switch on the agent loop's telemetry
|
||||
* hooks; metrics and structured logs need those callbacks even when traces are
|
||||
* disabled.
|
||||
*/
|
||||
export function isTelemetryExportEnabled(): boolean {
|
||||
return provider !== undefined;
|
||||
if (traceProvider) return true;
|
||||
if (logProvider) return true;
|
||||
if (meterProvider) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Register the global TracerProvider + OTLP exporter when an OTLP endpoint is
|
||||
* configured via env. Idempotent, and a no-op when no endpoint is set (or when
|
||||
* the OTEL kill-switches are engaged), so it is safe to call unconditionally at
|
||||
* startup.
|
||||
* Merge OTLP metrics/log hooks into an existing agent telemetry config.
|
||||
*
|
||||
* The caller still owns content-capture policy, cost estimation, and custom
|
||||
* attributes. This only appends host-level metrics/log forwarding for the
|
||||
* providers registered by {@link initTelemetryExport}.
|
||||
*/
|
||||
export function createTelemetryExportConfig(
|
||||
config: AgentTelemetryConfig | undefined,
|
||||
): AgentTelemetryConfig | undefined {
|
||||
if (!isTelemetryExportEnabled()) return config;
|
||||
return {
|
||||
...config,
|
||||
onChatUsage: event => {
|
||||
config?.onChatUsage?.(event);
|
||||
metricRecorder?.recordChatUsage(event);
|
||||
},
|
||||
onRunEnd: (summary, coverage) => {
|
||||
config?.onRunEnd?.(summary, coverage);
|
||||
metricRecorder?.recordRun(summary, coverage);
|
||||
emitRunSummaryLog(summary, coverage);
|
||||
},
|
||||
onTelemetryWarning: warning => {
|
||||
config?.onTelemetryWarning?.(warning);
|
||||
emitTelemetryWarningLog(warning);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Register global trace/log/meter providers when OTLP endpoints are configured
|
||||
* through env. Idempotent, and a no-op when no signal has an endpoint (or when
|
||||
* the OTEL kill-switches are engaged), so startup can call it unconditionally.
|
||||
*/
|
||||
export async function initTelemetryExport(): Promise<void> {
|
||||
if (provider) return;
|
||||
if (isTelemetryExportEnabled()) return;
|
||||
if (initPromise) return initPromise;
|
||||
|
||||
// The OTEL env contract parses booleans and enum lists case-insensitively, so
|
||||
// OTEL_SDK_DISABLED=TRUE and OTEL_TRACES_EXPORTER=None must also disable export.
|
||||
if (process.env.OTEL_SDK_DISABLED?.trim().toLowerCase() === "true") return;
|
||||
if (tracesExporterDisabled(process.env.OTEL_TRACES_EXPORTER)) return;
|
||||
|
||||
const endpoint = process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
|
||||
if (!endpoint) return;
|
||||
const signalConfig = resolveSignalConfig();
|
||||
if (!signalConfig.trace && !signalConfig.log && !signalConfig.metric) return;
|
||||
|
||||
// We only ship the http/protobuf transport (the line validated on Bun). The
|
||||
// OTEL contract lets OTEL_EXPORTER_OTLP*_PROTOCOL select grpc / http/json;
|
||||
// rather than silently send protobuf-over-HTTP to a grpc :4317 port and lose
|
||||
// every span, decline when an unsupported protocol is requested.
|
||||
const protocol = (process.env.OTEL_EXPORTER_OTLP_TRACES_PROTOCOL ?? process.env.OTEL_EXPORTER_OTLP_PROTOCOL)
|
||||
?.trim()
|
||||
.toLowerCase();
|
||||
if (protocol && protocol !== "http/protobuf") {
|
||||
logger.warn(
|
||||
`OTEL trace export disabled: OTEL_EXPORTER_OTLP_PROTOCOL=${protocol} is unsupported (only http/protobuf)`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
initPromise = registerProvider();
|
||||
initPromise = registerProviders(signalConfig);
|
||||
return initPromise;
|
||||
}
|
||||
|
||||
async function registerProvider(): Promise<void> {
|
||||
const [
|
||||
{ AsyncLocalStorageContextManager },
|
||||
{ OTLPTraceExporter },
|
||||
{ resourceFromAttributes },
|
||||
{ BatchSpanProcessor },
|
||||
{ NodeTracerProvider },
|
||||
] = await Promise.all([
|
||||
import("@opentelemetry/context-async-hooks"),
|
||||
import("@opentelemetry/exporter-trace-otlp-proto"),
|
||||
import("@opentelemetry/resources"),
|
||||
import("@opentelemetry/sdk-trace-base"),
|
||||
import("@opentelemetry/sdk-trace-node"),
|
||||
]);
|
||||
|
||||
// The exporter reads endpoint/headers/timeout from OTEL_EXPORTER_OTLP_* itself,
|
||||
// so there is nothing to thread through here.
|
||||
const exporter = new OTLPTraceExporter();
|
||||
const tracerProvider = new NodeTracerProvider({
|
||||
resource: resourceFromAttributes({
|
||||
"service.name": process.env.OTEL_SERVICE_NAME ?? "oh-my-pi",
|
||||
}),
|
||||
spanProcessors: [new BatchSpanProcessor(exporter)],
|
||||
async function registerProviders(signalConfig: SignalConfig): Promise<void> {
|
||||
const resource = resourceFromAttributes({
|
||||
"service.name": process.env.OTEL_SERVICE_NAME ?? SERVICE_NAME,
|
||||
});
|
||||
// register() installs the global tracer provider and the W3C trace-context +
|
||||
// baggage propagators; the explicit AsyncLocalStorage context manager keeps
|
||||
// parent/child span linkage working under Bun.
|
||||
tracerProvider.register({ contextManager: new AsyncLocalStorageContextManager().enable() });
|
||||
provider = tracerProvider;
|
||||
|
||||
if (signalConfig.trace) {
|
||||
const exporter = new OTLPTraceExporter();
|
||||
traceProvider = new NodeTracerProvider({
|
||||
resource,
|
||||
spanProcessors: [new BatchSpanProcessor(exporter)],
|
||||
});
|
||||
traceProvider.register({ contextManager: new AsyncLocalStorageContextManager().enable() });
|
||||
}
|
||||
|
||||
if (signalConfig.metric) {
|
||||
const exporter = new OTLPMetricExporter();
|
||||
meterProvider = new MeterProvider({
|
||||
resource,
|
||||
readers: [new PeriodicExportingMetricReader({ exporter })],
|
||||
});
|
||||
metrics.setGlobalMeterProvider(meterProvider);
|
||||
metricRecorder = new AgentMetricRecorder(metrics.getMeter("@oh-my-pi/pi-coding-agent"));
|
||||
}
|
||||
|
||||
if (signalConfig.log) {
|
||||
const exporter = new OTLPLogExporter();
|
||||
logProvider = new LoggerProvider({
|
||||
resource,
|
||||
processors: [new BatchLogRecordProcessor(exporter)],
|
||||
});
|
||||
logs.setGlobalLoggerProvider(logProvider);
|
||||
otelLogger = logs.getLogger("@oh-my-pi/pi-coding-agent");
|
||||
unregisterLogSink = logger.registerLogSink(event => {
|
||||
emitOtelLog(
|
||||
event.level,
|
||||
event.message,
|
||||
logAttributesFromContext(event.context),
|
||||
"pi.omp.log",
|
||||
event.timestamp,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
const flushTimer = setInterval(() => {
|
||||
provider?.forceFlush().catch(() => {});
|
||||
flushTelemetryExport().catch(() => {});
|
||||
}, FLUSH_INTERVAL_MS);
|
||||
flushTimer.unref();
|
||||
|
||||
// Shut down through postmortem rather than a bare signal listener. postmortem
|
||||
// owns SIGINT/SIGTERM/SIGHUP/exit and quit(), and awaits registered cleanups
|
||||
// before calling process.exit — so the batch processor's final OTLP export
|
||||
// completes instead of being cut off mid-flight on the shutdown path.
|
||||
postmortem.register("otel-trace-export", async () => {
|
||||
postmortem.register("otel-export", async () => {
|
||||
clearInterval(flushTimer);
|
||||
await provider?.shutdown();
|
||||
unregisterLogSink?.();
|
||||
unregisterLogSink = undefined;
|
||||
const shutdowns: Promise<void>[] = [];
|
||||
if (traceProvider) shutdowns.push(traceProvider.shutdown());
|
||||
if (logProvider) shutdowns.push(logProvider.shutdown());
|
||||
if (meterProvider) shutdowns.push(meterProvider.shutdown());
|
||||
await Promise.all(shutdowns);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse the `OTEL_TRACES_EXPORTER` selection. The value is a case-insensitive,
|
||||
* comma-separated list; the literal `none` disables span export entirely.
|
||||
*/
|
||||
function tracesExporterDisabled(raw: string | undefined): boolean {
|
||||
if (!raw) return false;
|
||||
return raw.split(",").some(entry => entry.trim().toLowerCase() === "none");
|
||||
function resolveSignalConfig(): SignalConfig {
|
||||
const signalConfig: SignalConfig = {
|
||||
trace: signalEnabled(
|
||||
"trace",
|
||||
process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT,
|
||||
process.env.OTEL_TRACES_EXPORTER,
|
||||
process.env.OTEL_EXPORTER_OTLP_TRACES_PROTOCOL ?? process.env.OTEL_EXPORTER_OTLP_PROTOCOL,
|
||||
),
|
||||
log: signalEnabled(
|
||||
"log",
|
||||
process.env.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT,
|
||||
process.env.OTEL_LOGS_EXPORTER,
|
||||
process.env.OTEL_EXPORTER_OTLP_LOGS_PROTOCOL ?? process.env.OTEL_EXPORTER_OTLP_PROTOCOL,
|
||||
),
|
||||
metric: signalEnabled(
|
||||
"metric",
|
||||
process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT ?? process.env.OTEL_EXPORTER_OTLP_ENDPOINT,
|
||||
process.env.OTEL_METRICS_EXPORTER,
|
||||
process.env.OTEL_EXPORTER_OTLP_METRICS_PROTOCOL ?? process.env.OTEL_EXPORTER_OTLP_PROTOCOL,
|
||||
),
|
||||
};
|
||||
return signalConfig;
|
||||
}
|
||||
|
||||
function signalEnabled(
|
||||
signal: TelemetrySignal,
|
||||
endpoint: string | undefined,
|
||||
exporterSelection: string | undefined,
|
||||
protocolSelection: string | undefined,
|
||||
): boolean {
|
||||
if (exporterSelection) {
|
||||
for (const entry of exporterSelection.split(",")) {
|
||||
if (entry.trim().toLowerCase() === "none") return false;
|
||||
}
|
||||
}
|
||||
if (!endpoint) return false;
|
||||
|
||||
const protocol = protocolSelection?.trim().toLowerCase();
|
||||
if (protocol && protocol !== "http/protobuf") {
|
||||
logger.warn(`OTEL ${signal} export disabled: OTEL_EXPORTER_OTLP_PROTOCOL=${protocol} is unsupported`, {
|
||||
supported: "http/protobuf",
|
||||
});
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
class AgentMetricRecorder {
|
||||
readonly #tokenUsage: Histogram<Attributes>;
|
||||
readonly #chatCostUsd: Counter<Attributes>;
|
||||
readonly #runs: Counter<Attributes>;
|
||||
readonly #steps: Counter<Attributes>;
|
||||
readonly #chatCalls: Counter<Attributes>;
|
||||
readonly #chatDurationMs: Histogram<Attributes>;
|
||||
readonly #toolCalls: Counter<Attributes>;
|
||||
readonly #toolDurationMs: Histogram<Attributes>;
|
||||
readonly #errors: Counter<Attributes>;
|
||||
|
||||
constructor(meter: Meter) {
|
||||
this.#tokenUsage = meter.createHistogram("gen_ai.client.token.usage", {
|
||||
description: "Token usage reported by GenAI chat calls.",
|
||||
unit: "{token}",
|
||||
});
|
||||
this.#chatCostUsd = meter.createCounter("pi.omp.agent.chat.cost.estimated_usd", {
|
||||
description: "Estimated USD cost for completed chat calls.",
|
||||
unit: "USD",
|
||||
});
|
||||
this.#runs = meter.createCounter("pi.omp.agent.runs", {
|
||||
description: "Completed agent runs.",
|
||||
unit: "{run}",
|
||||
});
|
||||
this.#steps = meter.createCounter("pi.omp.agent.steps", {
|
||||
description: "Agent loop steps completed inside a run.",
|
||||
unit: "{step}",
|
||||
});
|
||||
this.#chatCalls = meter.createCounter("pi.omp.agent.chat.calls", {
|
||||
description: "Chat calls completed inside agent runs.",
|
||||
unit: "{call}",
|
||||
});
|
||||
this.#chatDurationMs = meter.createHistogram("pi.omp.agent.chat.duration", {
|
||||
description: "Total chat latency observed in an agent run.",
|
||||
unit: "ms",
|
||||
});
|
||||
this.#toolCalls = meter.createCounter("pi.omp.agent.tool.calls", {
|
||||
description: "Tool calls completed inside agent runs.",
|
||||
unit: "{call}",
|
||||
});
|
||||
this.#toolDurationMs = meter.createHistogram("pi.omp.agent.tool.duration", {
|
||||
description: "Total tool latency observed in an agent run.",
|
||||
unit: "ms",
|
||||
});
|
||||
this.#errors = meter.createCounter("pi.omp.agent.errors", {
|
||||
description: "Errors observed in chat and tool execution.",
|
||||
unit: "{error}",
|
||||
});
|
||||
}
|
||||
|
||||
recordChatUsage(event: ChatUsageEvent): void {
|
||||
const baseAttrs = metricAttributes({
|
||||
"gen_ai.operation.name": "chat",
|
||||
"gen_ai.provider.name": event.provider,
|
||||
"gen_ai.request.model": event.model,
|
||||
"gen_ai.response.service_tier": event.serviceTier,
|
||||
"pi.gen_ai.agent.id": event.agent?.id,
|
||||
"pi.gen_ai.agent.name": event.agent?.name,
|
||||
});
|
||||
|
||||
this.#recordToken(event.usage.inputTokens, baseAttrs, "input");
|
||||
this.#recordToken(event.usage.outputTokens, baseAttrs, "output");
|
||||
this.#recordToken(event.usage.totalTokens, baseAttrs, "total");
|
||||
this.#recordToken(event.usage.cachedInputTokens, baseAttrs, "cache_read_input");
|
||||
this.#recordToken(event.usage.cacheWriteTokens, baseAttrs, "cache_write_input");
|
||||
this.#recordToken(event.usage.reasoningOutputTokens, baseAttrs, "reasoning_output");
|
||||
|
||||
if (event.cost && "usd" in event.cost && event.cost.usd > 0) {
|
||||
this.#chatCostUsd.add(event.cost.usd, baseAttrs);
|
||||
}
|
||||
}
|
||||
|
||||
recordRun(summary: AgentRunSummary, coverage: AgentRunCoverage): void {
|
||||
const runAttrs = metricAttributes({
|
||||
"pi.omp.agent.models_used.count": coverage.modelsUsed.length,
|
||||
"pi.omp.agent.providers_used.count": coverage.providersUsed.length,
|
||||
"pi.omp.agent.tools_available.count": coverage.toolsAvailable.length,
|
||||
"pi.omp.agent.tools_invoked.count": coverage.toolsInvoked.length,
|
||||
"pi.omp.agent.tools_unused.count": coverage.toolsUnused.length,
|
||||
});
|
||||
|
||||
this.#runs.add(1, runAttrs);
|
||||
if (summary.stepCount > 0) this.#steps.add(summary.stepCount, runAttrs);
|
||||
if (summary.chats.total > 0) this.#chatCalls.add(summary.chats.total, runAttrs);
|
||||
if (summary.chats.totalLatencyMs > 0) this.#chatDurationMs.record(summary.chats.totalLatencyMs, runAttrs);
|
||||
if (summary.tools.total > 0) this.#toolCalls.add(summary.tools.total, runAttrs);
|
||||
if (summary.tools.totalLatencyMs > 0) this.#toolDurationMs.record(summary.tools.totalLatencyMs, runAttrs);
|
||||
if (summary.errors.total > 0) this.#errors.add(summary.errors.total, runAttrs);
|
||||
|
||||
for (const reason in summary.chats.byStopReason) {
|
||||
const count = summary.chats.byStopReason[reason];
|
||||
if (count > 0)
|
||||
this.#chatCalls.add(count, metricAttributes({ ...runAttrs, "gen_ai.response.finish_reason": reason }));
|
||||
}
|
||||
for (const toolName in summary.tools.byName) {
|
||||
const counters = summary.tools.byName[toolName];
|
||||
const toolAttrs = metricAttributes({ ...runAttrs, "gen_ai.tool.name": toolName });
|
||||
if (counters.total > 0) this.#toolCalls.add(counters.total, toolAttrs);
|
||||
if (counters.totalLatencyMs > 0) this.#toolDurationMs.record(counters.totalLatencyMs, toolAttrs);
|
||||
for (const status of TOOL_STATUSES) {
|
||||
const count = counters[status];
|
||||
if (count > 0) this.#toolCalls.add(count, metricAttributes({ ...toolAttrs, "pi.omp.tool.status": status }));
|
||||
}
|
||||
}
|
||||
for (const errorType in summary.errors.byType) {
|
||||
const count = summary.errors.byType[errorType];
|
||||
if (count > 0) this.#errors.add(count, metricAttributes({ ...runAttrs, "error.type": errorType }));
|
||||
}
|
||||
}
|
||||
|
||||
#recordToken(value: number | undefined, baseAttrs: Attributes, tokenType: string): void {
|
||||
if (!value || value <= 0) return;
|
||||
this.#tokenUsage.record(value, metricAttributes({ ...baseAttrs, "gen_ai.token.type": tokenType }));
|
||||
}
|
||||
}
|
||||
|
||||
function metricAttributes(fields: Readonly<Record<string, unknown>>): Attributes {
|
||||
const out: Attributes = {};
|
||||
for (const key in fields) {
|
||||
const value = fields[key];
|
||||
if (value === undefined || value === null) continue;
|
||||
if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") {
|
||||
out[key] = value;
|
||||
continue;
|
||||
}
|
||||
const text = String(value);
|
||||
if (text.length > 0) out[key] = text;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function emitRunSummaryLog(summary: AgentRunSummary, coverage: AgentRunCoverage): void {
|
||||
emitOtelLog(
|
||||
"info",
|
||||
"agent run completed",
|
||||
{
|
||||
"pi.omp.agent.step_count": summary.stepCount,
|
||||
"pi.omp.agent.chats.total": summary.chats.total,
|
||||
"pi.omp.agent.chats.total_latency_ms": summary.chats.totalLatencyMs,
|
||||
"pi.omp.agent.tools.total": summary.tools.total,
|
||||
"pi.omp.agent.tools.ok": summary.tools.ok,
|
||||
"pi.omp.agent.tools.error": summary.tools.error,
|
||||
"pi.omp.agent.tools.skipped": summary.tools.skipped,
|
||||
"pi.omp.agent.tools.blocked": summary.tools.blocked,
|
||||
"pi.omp.agent.tools.timeout": summary.tools.timeout,
|
||||
"pi.omp.agent.tools.aborted": summary.tools.aborted,
|
||||
"pi.omp.agent.tools.total_latency_ms": summary.tools.totalLatencyMs,
|
||||
"pi.omp.agent.usage.input_tokens": summary.usage.inputTokens,
|
||||
"pi.omp.agent.usage.output_tokens": summary.usage.outputTokens,
|
||||
"pi.omp.agent.usage.cached_input_tokens": summary.usage.cachedInputTokens,
|
||||
"pi.omp.agent.usage.cache_write_tokens": summary.usage.cacheWriteTokens,
|
||||
"pi.omp.agent.usage.reasoning_output_tokens": summary.usage.reasoningOutputTokens,
|
||||
"pi.omp.agent.usage.total_tokens": summary.usage.totalTokens,
|
||||
"pi.omp.agent.cost.estimated_usd": summary.cost.estimatedUsd,
|
||||
"pi.omp.agent.cost.unavailable_reasons": summary.cost.unavailableReasons.join(","),
|
||||
"pi.omp.agent.errors.total": summary.errors.total,
|
||||
"pi.omp.agent.coverage.tools_available": coverage.toolsAvailable.join(","),
|
||||
"pi.omp.agent.coverage.tools_invoked": coverage.toolsInvoked.join(","),
|
||||
"pi.omp.agent.coverage.tools_unused": coverage.toolsUnused.join(","),
|
||||
"pi.omp.agent.coverage.models_used": coverage.modelsUsed.join(","),
|
||||
"pi.omp.agent.coverage.providers_used": coverage.providersUsed.join(","),
|
||||
},
|
||||
"pi.omp.agent.run.completed",
|
||||
);
|
||||
}
|
||||
|
||||
function emitTelemetryWarningLog(warning: AgentTelemetryWarning): void {
|
||||
const attrs = logAttributesFromContext({
|
||||
code: warning.code,
|
||||
error: warning.error,
|
||||
});
|
||||
emitOtelLog("warn", warning.message, attrs, "pi.omp.telemetry.warning");
|
||||
}
|
||||
|
||||
function emitOtelLog(
|
||||
level: logger.LogLevel,
|
||||
body: string,
|
||||
attributes: LogAttributes,
|
||||
eventName: string,
|
||||
timestamp = new Date(),
|
||||
): void {
|
||||
if (!otelLogger) return;
|
||||
const minLevel = parseOtelLogLevel(process.env.OTEL_LOG_LEVEL);
|
||||
if (minLevel === "none") return;
|
||||
if (LOG_LEVEL_WEIGHT[level] > LOG_LEVEL_WEIGHT[minLevel]) return;
|
||||
otelLogger.emit({
|
||||
eventName,
|
||||
timestamp,
|
||||
observedTimestamp: new Date(),
|
||||
severityNumber: LOG_SEVERITY[level],
|
||||
severityText: level.toUpperCase(),
|
||||
body,
|
||||
attributes,
|
||||
context: context.active(),
|
||||
});
|
||||
}
|
||||
|
||||
function parseOtelLogLevel(raw: string | undefined): OtelLogLevel {
|
||||
if (!raw) return "info";
|
||||
switch (raw.trim().toLowerCase()) {
|
||||
case "none":
|
||||
return "none";
|
||||
case "error":
|
||||
return "error";
|
||||
case "warn":
|
||||
case "warning":
|
||||
return "warn";
|
||||
case "debug":
|
||||
return "debug";
|
||||
default:
|
||||
return "info";
|
||||
}
|
||||
}
|
||||
|
||||
function logAttributesFromContext(input: Record<string, unknown> | undefined): LogAttributes {
|
||||
const out: LogAttributes = { "process.pid": process.pid };
|
||||
if (!input) return out;
|
||||
for (const key in input) {
|
||||
const attr = logAttributeValue(input[key]);
|
||||
if (attr !== undefined) out[key] = attr;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function logAttributeValue(value: unknown): AttributeValue | undefined {
|
||||
if (value === undefined || value === null) return undefined;
|
||||
if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") return value;
|
||||
if (value instanceof Error) {
|
||||
return `${value.name}: ${value.message}`;
|
||||
}
|
||||
try {
|
||||
const text = JSON.stringify(value);
|
||||
if (text && text.length > 0) return text;
|
||||
} catch {
|
||||
return String(value);
|
||||
}
|
||||
return String(value);
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush any buffered spans to the exporter. No-op when export is disabled.
|
||||
* Flush buffered spans, log records, and metrics. No-op when export is disabled.
|
||||
* Hosts embedding the agent can call this at natural boundaries (e.g. the end
|
||||
* of a turn) so traces surface promptly rather than on the batch interval.
|
||||
* of a turn) so telemetry surfaces promptly rather than on the batch interval.
|
||||
*/
|
||||
export async function flushTelemetryExport(): Promise<void> {
|
||||
await provider?.forceFlush();
|
||||
const flushes: Promise<void>[] = [];
|
||||
if (traceProvider) flushes.push(traceProvider.forceFlush());
|
||||
if (logProvider) flushes.push(logProvider.forceFlush());
|
||||
if (meterProvider) flushes.push(meterProvider.forceFlush());
|
||||
await Promise.all(flushes);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
/**
|
||||
* Positive-path probe for the OTLP log + metric exporters, run as a subprocess
|
||||
* by telemetry-export.test.ts. Keeping it out-of-process means the global
|
||||
* LoggerProvider / MeterProvider singletons that initTelemetryExport() registers
|
||||
* never leak into the test runner.
|
||||
*
|
||||
* Stands up a loopback OTLP/proto receiver, points the standard env vars at it,
|
||||
* registers the providers, drives a log record through the bridged
|
||||
* `@oh-my-pi/pi-utils` logger and metric instruments through the agent
|
||||
* telemetry hooks, flushes, and exits 0 only if the receiver got a non-empty
|
||||
* protobuf POST at both /v1/logs and /v1/metrics.
|
||||
*/
|
||||
|
||||
import type { AgentRunCoverage, AgentRunSummary, ChatUsageEvent } from "@oh-my-pi/pi-agent-core";
|
||||
import { emptyAgentRunCoverage, emptyAgentRunSummary } from "@oh-my-pi/pi-agent-core";
|
||||
import {
|
||||
createTelemetryExportConfig,
|
||||
flushTelemetryExport,
|
||||
initTelemetryExport,
|
||||
isTelemetryExportEnabled,
|
||||
} from "@oh-my-pi/pi-coding-agent/telemetry-export";
|
||||
import { logger } from "@oh-my-pi/pi-utils";
|
||||
|
||||
const seen = new Set<string>();
|
||||
|
||||
const server = Bun.serve({
|
||||
port: 0,
|
||||
async fetch(req) {
|
||||
const path = new URL(req.url).pathname;
|
||||
if (req.method === "POST" && req.headers.get("content-type") === "application/x-protobuf") {
|
||||
const body = await req.arrayBuffer();
|
||||
if (body.byteLength > 0) {
|
||||
if (path.endsWith("/v1/logs")) seen.add("logs");
|
||||
if (path.endsWith("/v1/metrics")) seen.add("metrics");
|
||||
}
|
||||
}
|
||||
return new Response('{"partialSuccess":{}}', {
|
||||
status: 200,
|
||||
headers: { "content-type": "application/json" },
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
const base = `http://localhost:${server.port}`;
|
||||
process.env.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT = `${base}/v1/logs`;
|
||||
process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT = `${base}/v1/metrics`;
|
||||
process.env.OTEL_SERVICE_NAME = "oh-my-pi-signals-probe";
|
||||
// Force a short metric export interval so the periodic reader flushes fast.
|
||||
process.env.OTEL_METRIC_EXPORT_INTERVAL = "500";
|
||||
|
||||
await initTelemetryExport();
|
||||
if (!isTelemetryExportEnabled()) {
|
||||
console.error("PROBE: providers did not register");
|
||||
await server.stop(true);
|
||||
process.exit(2);
|
||||
}
|
||||
|
||||
const config = createTelemetryExportConfig(undefined);
|
||||
if (!config) {
|
||||
console.error("PROBE: export config not produced");
|
||||
await server.stop(true);
|
||||
process.exit(2);
|
||||
}
|
||||
|
||||
// Bridged utility logger -> OTel log record.
|
||||
logger.error("probe error", { code: "probe" });
|
||||
|
||||
// Metric instruments via the agent telemetry hooks.
|
||||
const usage: ChatUsageEvent = {
|
||||
span: undefined as never,
|
||||
agent: { id: "main", name: "Main" },
|
||||
conversationId: "probe-session",
|
||||
stepNumber: 0,
|
||||
model: "claude-haiku-4-5",
|
||||
provider: "anthropic",
|
||||
serviceTier: undefined,
|
||||
usage: {
|
||||
inputTokens: 1000,
|
||||
outputTokens: 200,
|
||||
totalTokens: 1200,
|
||||
cachedInputTokens: 0,
|
||||
cacheWriteTokens: 0,
|
||||
reasoningOutputTokens: 0,
|
||||
},
|
||||
cost: { usd: 0.01 },
|
||||
attributes: undefined,
|
||||
headers: undefined,
|
||||
};
|
||||
config.onChatUsage?.(usage);
|
||||
|
||||
const summary: AgentRunSummary = {
|
||||
...emptyAgentRunSummary(),
|
||||
chats: { total: 1, byStopReason: { end_turn: 1 }, totalLatencyMs: 1500 },
|
||||
tools: {
|
||||
total: 1,
|
||||
ok: 1,
|
||||
error: 0,
|
||||
skipped: 0,
|
||||
blocked: 0,
|
||||
timeout: 0,
|
||||
aborted: 0,
|
||||
totalLatencyMs: 42,
|
||||
byName: {
|
||||
read: { total: 1, ok: 1, error: 0, skipped: 0, blocked: 0, timeout: 0, aborted: 0, totalLatencyMs: 42 },
|
||||
},
|
||||
},
|
||||
stepCount: 1,
|
||||
};
|
||||
const coverage: AgentRunCoverage = {
|
||||
...emptyAgentRunCoverage(),
|
||||
toolsAvailable: ["read", "write"],
|
||||
toolsInvoked: ["read"],
|
||||
toolsUnused: ["write"],
|
||||
modelsUsed: ["claude-haiku-4-5"],
|
||||
providersUsed: ["anthropic"],
|
||||
};
|
||||
config.onRunEnd?.(summary, coverage);
|
||||
|
||||
await flushTelemetryExport();
|
||||
// The metric reader exports on its own interval; wait one cycle then flush.
|
||||
await Bun.sleep(700);
|
||||
await flushTelemetryExport();
|
||||
await server.stop(true);
|
||||
|
||||
const ok = seen.has("logs") && seen.has("metrics");
|
||||
console.log(ok ? "PROBE: RECEIVED" : `PROBE: MISSING ${["logs", "metrics"].filter(s => !seen.has(s)).join(",")}`);
|
||||
process.exit(ok ? 0 : 1);
|
||||
@@ -11,10 +11,16 @@ import { initTelemetryExport, isTelemetryExportEnabled } from "@oh-my-pi/pi-codi
|
||||
const OTEL_KEYS = [
|
||||
"OTEL_EXPORTER_OTLP_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_LOGS_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
|
||||
"OTEL_EXPORTER_OTLP_PROTOCOL",
|
||||
"OTEL_EXPORTER_OTLP_TRACES_PROTOCOL",
|
||||
"OTEL_EXPORTER_OTLP_LOGS_PROTOCOL",
|
||||
"OTEL_EXPORTER_OTLP_METRICS_PROTOCOL",
|
||||
"OTEL_SDK_DISABLED",
|
||||
"OTEL_TRACES_EXPORTER",
|
||||
"OTEL_LOGS_EXPORTER",
|
||||
"OTEL_METRICS_EXPORTER",
|
||||
] as const;
|
||||
|
||||
let saved: Record<string, string | undefined>;
|
||||
@@ -45,8 +51,8 @@ describe("initTelemetryExport gating", () => {
|
||||
expect(isTelemetryExportEnabled()).toBe(false);
|
||||
});
|
||||
|
||||
it("stays disabled when OTEL_TRACES_EXPORTER=none even with an endpoint", async () => {
|
||||
process.env.OTEL_EXPORTER_OTLP_ENDPOINT = "http://localhost:4318";
|
||||
it("stays disabled when OTEL_TRACES_EXPORTER=none and only the traces endpoint is set", async () => {
|
||||
process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT = "http://localhost:4318";
|
||||
process.env.OTEL_TRACES_EXPORTER = "none";
|
||||
await initTelemetryExport();
|
||||
expect(isTelemetryExportEnabled()).toBe(false);
|
||||
@@ -71,12 +77,23 @@ describe("initTelemetryExport gating", () => {
|
||||
|
||||
delete process.env.OTEL_SDK_DISABLED;
|
||||
process.env.OTEL_TRACES_EXPORTER = "otlp,None";
|
||||
process.env.OTEL_LOGS_EXPORTER = "none";
|
||||
process.env.OTEL_METRICS_EXPORTER = "none";
|
||||
await initTelemetryExport();
|
||||
expect(isTelemetryExportEnabled()).toBe(false);
|
||||
});
|
||||
|
||||
it("stays disabled when every signal exporter is set to none", async () => {
|
||||
process.env.OTEL_EXPORTER_OTLP_ENDPOINT = "http://localhost:4318";
|
||||
process.env.OTEL_TRACES_EXPORTER = "none";
|
||||
process.env.OTEL_LOGS_EXPORTER = "none";
|
||||
process.env.OTEL_METRICS_EXPORTER = "none";
|
||||
await initTelemetryExport();
|
||||
expect(isTelemetryExportEnabled()).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("initTelemetryExport export path", () => {
|
||||
describe("initTelemetryExport signals export path", () => {
|
||||
it("registers a provider and exports spans to an OTLP/proto receiver", async () => {
|
||||
// Run in a subprocess: initTelemetryExport() registers a process-global
|
||||
// provider, so exercising the positive path in-process would leak that
|
||||
@@ -88,4 +105,15 @@ describe("initTelemetryExport export path", () => {
|
||||
expect(stdout).toContain("PROBE: RECEIVED");
|
||||
expect(code).toBe(0);
|
||||
}, 20_000);
|
||||
|
||||
it("exports log records and metrics to OTLP/proto receivers", async () => {
|
||||
// Same subprocess isolation as the trace probe: the logs/metrics probe
|
||||
// drives the bridged logger and the agent telemetry metric hooks, then
|
||||
// asserts protobuf POSTs landed at both /v1/logs and /v1/metrics.
|
||||
const probe = fileURLToPath(new URL("./otel-signals-probe.ts", import.meta.url));
|
||||
const proc = Bun.spawn(["bun", probe], { stdout: "pipe", stderr: "pipe" });
|
||||
const [code, stdout] = await Promise.all([proc.exited, new Response(proc.stdout).text()]);
|
||||
expect(stdout).toContain("PROBE: RECEIVED");
|
||||
expect(code).toBe(0);
|
||||
}, 20_000);
|
||||
});
|
||||
|
||||
@@ -2,6 +2,10 @@
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Added
|
||||
|
||||
- Added a structured log sink API to the centralized logger (`registerLogSink`, `LogEvent`, `LogLevel`) so out-of-band consumers (e.g. OpenTelemetry log export) receive every `error`/`warn`/`info`/`debug` event after the local transport path runs, without disturbing existing file/console logging ([#4604](https://github.com/can1357/oh-my-pi/issues/4604)).
|
||||
|
||||
## [16.3.1] - 2026-07-02
|
||||
|
||||
### Fixed
|
||||
|
||||
@@ -17,6 +17,42 @@ import DailyRotateFile from "winston-daily-rotate-file";
|
||||
import { getLogsDir } from "./dirs";
|
||||
import { drainModuleLoadEvents } from "./timing-buffer";
|
||||
|
||||
/** Severity names accepted by the centralized logger. */
|
||||
export type LogLevel = "error" | "warn" | "info" | "debug";
|
||||
|
||||
/** Structured log event forwarded to out-of-band sinks such as OpenTelemetry. */
|
||||
export interface LogEvent {
|
||||
readonly level: LogLevel;
|
||||
readonly message: string;
|
||||
readonly context: Record<string, unknown> | undefined;
|
||||
readonly timestamp: Date;
|
||||
}
|
||||
|
||||
/** Receives each structured log event after the local transport path runs. */
|
||||
export type LogSink = (event: LogEvent) => void;
|
||||
|
||||
const logSinks = new Set<LogSink>();
|
||||
|
||||
/** Register an out-of-band log sink and return a disposer. */
|
||||
export function registerLogSink(sink: LogSink): () => void {
|
||||
logSinks.add(sink);
|
||||
return () => {
|
||||
logSinks.delete(sink);
|
||||
};
|
||||
}
|
||||
|
||||
function emitToSinks(level: LogLevel, message: string, context: Record<string, unknown> | undefined): void {
|
||||
if (logSinks.size === 0) return;
|
||||
const event: LogEvent = { level, message, context, timestamp: new Date() };
|
||||
for (const sink of logSinks) {
|
||||
try {
|
||||
sink(event);
|
||||
} catch {
|
||||
// Sinks are side channels; they must never break local logging.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Ensure a logs directory exists; return the resolved path. */
|
||||
function ensureDir(dir: string): string {
|
||||
if (!fs.existsSync(dir)) {
|
||||
@@ -148,6 +184,7 @@ export function error(message: string, context?: Record<string, unknown>): void
|
||||
} catch {
|
||||
// Silently ignore logging failures
|
||||
}
|
||||
emitToSinks("error", message, context);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -161,6 +198,7 @@ export function warn(message: string, context?: Record<string, unknown>): void {
|
||||
} catch {
|
||||
// Silently ignore logging failures
|
||||
}
|
||||
emitToSinks("warn", message, context);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -174,6 +212,7 @@ export function info(message: string, context?: Record<string, unknown>): void {
|
||||
} catch {
|
||||
// Silently ignore logging failures
|
||||
}
|
||||
emitToSinks("info", message, context);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -187,6 +226,7 @@ export function debug(message: string, context?: Record<string, unknown>): void
|
||||
} catch {
|
||||
// Silently ignore logging failures
|
||||
}
|
||||
emitToSinks("debug", message, context);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user