From 168dc2f796b4fbb768a0c3a5d638c5f0fdf378d5 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 15 Jun 2026 20:19:16 -0700 Subject: [PATCH] =?UTF-8?q?feat(command-center):=20U10=20=E2=80=94=20OpenT?= =?UTF-8?q?elemetry=20(OTLP)=20metrics=20export?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mapAnalyticsToOtlp maps token/cost/activity aggregates to the OTLP/HTTP JSON envelope (counters + gauges, model/provider/node/agent attributes); a periodic dashboard exporter is opt-in via FUSION_OTEL_METRICS_ENDPOINT, https-validated, header-redacted, backs off on failure, and never blocks startup/shutdown. Minimal exporter (no SDK dep) — real @opentelemetry SDK is a follow-up. --- .changeset/u10-otel-export.md | 14 + .../core/src/__tests__/otel-metrics.test.ts | 207 ++++++++++ packages/core/src/index.ts | 8 + packages/core/src/otel-metrics.ts | 284 ++++++++++++++ .../src/__tests__/otel-exporter.test.ts | 250 ++++++++++++ packages/dashboard/src/otel-exporter.ts | 365 ++++++++++++++++++ packages/dashboard/src/server.ts | 18 + 7 files changed, 1146 insertions(+) create mode 100644 .changeset/u10-otel-export.md create mode 100644 packages/core/src/__tests__/otel-metrics.test.ts create mode 100644 packages/core/src/otel-metrics.ts create mode 100644 packages/dashboard/src/__tests__/otel-exporter.test.ts create mode 100644 packages/dashboard/src/otel-exporter.ts diff --git a/.changeset/u10-otel-export.md b/.changeset/u10-otel-export.md new file mode 100644 index 0000000000..a58f8d43b9 --- /dev/null +++ b/.changeset/u10-otel-export.md @@ -0,0 +1,14 @@ +--- +"@runfusion/fusion": minor +--- + +Export Command Center analytics over OpenTelemetry (OTLP) so teams can ship token / cost / activity metrics to Datadog / Grafana / etc. **Disabled by default** (U10, R4). + +- New pure mapping `mapAnalyticsToOtlp` in `@fusion/core` (`otel-metrics.ts`) turns the token/cost/activity aggregator outputs into the OTLP/HTTP JSON wire shape (`resourceMetrics`) — counters for token/cost, gauges for activity — with `model` / `provider` / `node.id` / `agent.id` attributes per data point. Fully testable without a live collector; no SDK dependency in core. +- Dashboard exporter (`otel-exporter.ts`) periodically maps current analytics and POSTs them to a configured collector, wired into `server.ts` startup/shutdown. + +**SDK choice:** ships a **minimal OTLP/HTTP JSON exporter rather than the official `@opentelemetry/*` SDK** — and therefore adds **no new runtime dependency**. The OTLP/HTTP JSON protocol is a single, stable `POST /v1/metrics` of a well-defined JSON envelope (built in core), so for a default-disabled feature we avoid pulling the multi-package SDK (sdk-metrics + exporter-metrics-otlp-http + resources + api). The wire shape is collector-compatible; swapping in the official SDK later is mechanical. (If maintainers prefer the real SDK, that is a follow-up changeset + dependency add.) + +**Enabled only via env** (none set ⇒ nothing starts): `FUSION_OTEL_METRICS_ENDPOINT` (full `/v1/metrics` URL, required to enable), `FUSION_OTEL_METRICS_HEADERS` (`k=v,k2=v2` auth headers), `FUSION_OTEL_METRICS_INTERVAL_MS`, `FUSION_OTEL_METRICS_TIMEOUT_MS`, `FUSION_OTEL_RESOURCE_ATTRIBUTES`. + +**Security:** endpoint validated on write — `http://` is rejected in production (exporter does not start) and warns loudly otherwise; auth header (Datadog/Grafana token) VALUES are never logged and are masked in diagnostics; a collector-unreachable failure logs (redacted) and backs off exponentially without crashing the server or blocking requests. diff --git a/packages/core/src/__tests__/otel-metrics.test.ts b/packages/core/src/__tests__/otel-metrics.test.ts new file mode 100644 index 0000000000..e366f2a2d9 --- /dev/null +++ b/packages/core/src/__tests__/otel-metrics.test.ts @@ -0,0 +1,207 @@ +import { describe, it, expect } from "vitest"; + +import { mapAnalyticsToOtlp, OTEL_METRIC_PREFIX } from "../otel-metrics.js"; +import type { TokenAnalytics } from "../token-analytics.js"; +import type { ActivityAnalytics } from "../activity-analytics.js"; + +const TIME_NANO = "1700000000000000000"; + +function tokenFixture(): TokenAnalytics { + return { + from: null, + to: null, + groupBy: "model", + totals: { + inputTokens: 1000, + outputTokens: 500, + cachedTokens: 200, + cacheWriteTokens: 50, + totalTokens: 1750, + nTasks: 3, + }, + cost: { usd: 12.34, unavailable: false, stale: false }, + groups: [ + { + key: "claude-opus-4-8", + inputTokens: 600, + outputTokens: 300, + cachedTokens: 100, + cacheWriteTokens: 25, + totalTokens: 1025, + nTasks: 2, + cost: { usd: 9.0, unavailable: false, stale: false }, + }, + { + key: "gpt-5", + inputTokens: 400, + outputTokens: 200, + cachedTokens: 100, + cacheWriteTokens: 25, + totalTokens: 725, + nTasks: 1, + // Unpriced group → cost must be omitted, not reported as $0. + cost: { usd: null, unavailable: true, stale: false }, + }, + ], + }; +} + +function activityFixture(): ActivityAnalytics { + return { + from: null, + to: null, + sessions: 7, + messages: 42, + activeNodes: 3, + activeAgents: 5, + daily: [], + stickiness: 0.6, + mttr: { value: null, unavailable: true }, + }; +} + +function findMetric(payload: ReturnType, name: string) { + const metrics = payload.resourceMetrics[0].scopeMetrics[0].metrics; + const m = metrics.find((x) => x.name === name); + expect(m, `metric ${name} present`).toBeDefined(); + return m!; +} + +describe("mapAnalyticsToOtlp", () => { + it("maps token totals to a monotonic Sum counter with a grand-total point", () => { + const payload = mapAnalyticsToOtlp({ + tokens: tokenFixture(), + activity: activityFixture(), + timeUnixNano: TIME_NANO, + }); + const total = findMetric(payload, `${OTEL_METRIC_PREFIX}.tokens.total`); + expect(total.sum?.isMonotonic).toBe(true); + expect(total.sum?.aggregationTemporality).toBe(2); + // Grand total point (no attributes) carries the totals value. + const grand = total.sum?.dataPoints.find((p) => p.attributes.length === 0); + expect(grand?.asInt).toBe("1750"); + }); + + it("emits one attributed data point per group (model/provider/node/agent)", () => { + const payload = mapAnalyticsToOtlp({ + tokens: tokenFixture(), + activity: activityFixture(), + timeUnixNano: TIME_NANO, + }); + const input = findMetric(payload, `${OTEL_METRIC_PREFIX}.tokens.input`); + const modelPoints = input.sum!.dataPoints.filter((p) => + p.attributes.some((a) => a.key === "model"), + ); + const models = modelPoints + .map((p) => p.attributes.find((a) => a.key === "model")!.value.stringValue) + .sort(); + expect(models).toEqual(["claude-opus-4-8", "gpt-5"]); + const opus = modelPoints.find( + (p) => + p.attributes.find((a) => a.key === "model")!.value.stringValue === + "claude-opus-4-8", + ); + expect(opus?.asInt).toBe("600"); + }); + + it("uses provider/node/agent attribute keys per groupBy", () => { + const base = tokenFixture(); + for (const [groupBy, attrKey] of [ + ["provider", "provider"], + ["node", "node.id"], + ["agent", "agent.id"], + ] as const) { + const payload = mapAnalyticsToOtlp({ + tokens: { ...base, groupBy, groups: [{ ...base.groups[0], key: "k" }] }, + activity: activityFixture(), + timeUnixNano: TIME_NANO, + }); + const input = findMetric(payload, `${OTEL_METRIC_PREFIX}.tokens.input`); + const attributed = input.sum!.dataPoints.find((p) => p.attributes.length > 0); + expect(attributed?.attributes[0].key).toBe(attrKey); + } + }); + + it("omits cost data points for unpriced groups (never reports $0)", () => { + const payload = mapAnalyticsToOtlp({ + tokens: tokenFixture(), + activity: activityFixture(), + timeUnixNano: TIME_NANO, + }); + const cost = findMetric(payload, `${OTEL_METRIC_PREFIX}.cost.usd`); + // Grand total (12.34) + opus (9.0); gpt-5 (null) omitted ⇒ 2 points. + expect(cost.sum?.dataPoints.length).toBe(2); + const grand = cost.sum?.dataPoints.find((p) => p.attributes.length === 0); + expect(grand?.asDouble).toBeCloseTo(12.34, 5); + const hasGpt5 = cost.sum?.dataPoints.some((p) => + p.attributes.some((a) => a.value.stringValue === "gpt-5"), + ); + expect(hasGpt5).toBe(false); + }); + + it("maps activity to gauges (active nodes/agents/sessions/messages/stickiness)", () => { + const payload = mapAnalyticsToOtlp({ + tokens: tokenFixture(), + activity: activityFixture(), + timeUnixNano: TIME_NANO, + }); + expect( + findMetric(payload, `${OTEL_METRIC_PREFIX}.activity.active_nodes`).gauge + ?.dataPoints[0].asInt, + ).toBe("3"); + expect( + findMetric(payload, `${OTEL_METRIC_PREFIX}.activity.active_agents`).gauge + ?.dataPoints[0].asInt, + ).toBe("5"); + expect( + findMetric(payload, `${OTEL_METRIC_PREFIX}.activity.sessions`).gauge + ?.dataPoints[0].asInt, + ).toBe("7"); + expect( + findMetric(payload, `${OTEL_METRIC_PREFIX}.activity.messages`).gauge + ?.dataPoints[0].asInt, + ).toBe("42"); + expect( + findMetric(payload, `${OTEL_METRIC_PREFIX}.activity.stickiness`).gauge + ?.dataPoints[0].asDouble, + ).toBeCloseTo(0.6, 5); + }); + + it("applies resource attributes and a default service.name", () => { + const dflt = mapAnalyticsToOtlp({ + tokens: tokenFixture(), + activity: activityFixture(), + timeUnixNano: TIME_NANO, + }); + const defaultAttrs = dflt.resourceMetrics[0].resource.attributes; + expect( + defaultAttrs.find((a) => a.key === "service.name")?.value.stringValue, + ).toBe("fusion-dashboard"); + + const custom = mapAnalyticsToOtlp({ + tokens: tokenFixture(), + activity: activityFixture(), + timeUnixNano: TIME_NANO, + resourceAttributes: { "service.name": "my-svc", env: "staging" }, + }); + const attrs = custom.resourceMetrics[0].resource.attributes; + expect(attrs.find((a) => a.key === "env")?.value.stringValue).toBe("staging"); + }); + + it("coerces non-finite / negative counts to 0 (no NaN on the wire)", () => { + const bad = tokenFixture(); + bad.totals.inputTokens = Number.NaN; + bad.totals.outputTokens = -5; + const payload = mapAnalyticsToOtlp({ + tokens: bad, + activity: activityFixture(), + timeUnixNano: TIME_NANO, + }); + const input = findMetric(payload, `${OTEL_METRIC_PREFIX}.tokens.input`); + const grand = input.sum!.dataPoints.find((p) => p.attributes.length === 0); + expect(grand?.asInt).toBe("0"); + const output = findMetric(payload, `${OTEL_METRIC_PREFIX}.tokens.output`); + const grandOut = output.sum!.dataPoints.find((p) => p.attributes.length === 0); + expect(grandOut?.asInt).toBe("0"); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index d336658c3d..98e038aa3f 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -579,6 +579,14 @@ export type { LiveRun, ColumnCount, } from "./command-center-live.js"; +export { mapAnalyticsToOtlp, OTEL_METRIC_PREFIX } from "./otel-metrics.js"; +export type { + OtelMappingInput, + OtlpExportPayload, + OtlpMetric, + OtlpNumberDataPoint, + OtlpAttribute, +} from "./otel-metrics.js"; export { STALLED_REVIEW_REENQUEUE_THRESHOLD, STALLED_REVIEW_INVALID_TRANSITION_THRESHOLD, diff --git a/packages/core/src/otel-metrics.ts b/packages/core/src/otel-metrics.ts new file mode 100644 index 0000000000..43990bbbd8 --- /dev/null +++ b/packages/core/src/otel-metrics.ts @@ -0,0 +1,284 @@ +/** + * OpenTelemetry (OTLP) metric mapping (U10). + * + * Pure mapping of the Command Center aggregator outputs (tokens / cost / activity) + * to OTLP metric instruments. This module produces the **OTLP/HTTP JSON wire + * shape** (`{ resourceMetrics: [...] }`) directly — the exact body an OTLP/HTTP + * collector accepts at `/v1/metrics` — so it is testable without a live collector + * and without pulling the full `@opentelemetry/*` SDK into `@fusion/core`. + * + * Design (KTD2): the MAPPING lives in core (reusable, side-effect-free); the + * network export (endpoint validation, auth headers, periodic scheduling, back + * off) lives in the dashboard exporter. This module never reads the clock, the + * network, or env — callers pass an explicit `timeUnixNano`. + * + * Instrument choices: + * - Token counts and USD cost are **monotonic counters** (`Sum`, cumulative, + * monotonic) — they only grow over a fixed range and aggregate cleanly. + * - Activity "current state" figures (active nodes/agents, sessions, stickiness) + * are **gauges** — point-in-time values that should not be summed across + * series. + * + * Attributes (model / provider / node / agent) are attached per data point from + * the aggregator's group keys, so a collector can break metrics down by any of + * them. We emit one data point per group plus an unattributed grand-total point. + */ + +import type { TokenAnalytics } from "./token-analytics.js"; +import type { ActivityAnalytics } from "./activity-analytics.js"; + +/** Instrument namespace prefix for every metric this module emits. */ +export const OTEL_METRIC_PREFIX = "fusion.command_center"; + +/** A single OTLP attribute key/value (string-valued; numbers are stringified). */ +export interface OtlpAttribute { + key: string; + value: { stringValue: string }; +} + +/** An OTLP number data point (used for both Sum and Gauge). */ +export interface OtlpNumberDataPoint { + /** Group attributes (model/provider/node/agent), empty for grand totals. */ + attributes: OtlpAttribute[]; + /** Nanoseconds since epoch; the start of the measurement window. */ + startTimeUnixNano: string; + /** Nanoseconds since epoch; when the value was observed. */ + timeUnixNano: string; + /** Integer counts use asInt; fractional values (cost, ratios) use asDouble. */ + asInt?: string; + asDouble?: number; +} + +/** An OTLP metric (one instrument), either a Sum (counter) or a Gauge. */ +export interface OtlpMetric { + name: string; + description: string; + unit: string; + sum?: { + dataPoints: OtlpNumberDataPoint[]; + /** 2 = CUMULATIVE in the OTLP AggregationTemporality enum. */ + aggregationTemporality: 2; + isMonotonic: boolean; + }; + gauge?: { + dataPoints: OtlpNumberDataPoint[]; + }; +} + +/** The OTLP/HTTP JSON export envelope sent to a collector's `/v1/metrics`. */ +export interface OtlpExportPayload { + resourceMetrics: Array<{ + resource: { attributes: OtlpAttribute[] }; + scopeMetrics: Array<{ + scope: { name: string; version: string }; + metrics: OtlpMetric[]; + }>; + }>; +} + +/** Inputs to {@link mapAnalyticsToOtlp}. */ +export interface OtelMappingInput { + tokens: TokenAnalytics; + activity: ActivityAnalytics; + /** Observation time in nanoseconds since the Unix epoch (caller-supplied). */ + timeUnixNano: string; + /** + * Start of the measurement window in nanoseconds since the Unix epoch. Used + * for the Sum start time so a collector treats the counters as a fresh + * cumulative series. Defaults to {@link OtelMappingInput.timeUnixNano}. + */ + startTimeUnixNano?: string; + /** + * Resource attributes describing the emitting service (e.g. + * `{ "service.name": "fusion-dashboard" }`). Defaults to a minimal + * `service.name`. + */ + resourceAttributes?: Record; + /** OTLP scope (instrumentation library) version. Defaults to `"1"`. */ + scopeVersion?: string; +} + +function attr(key: string, value: string): OtlpAttribute { + return { key, value: { stringValue: value } }; +} + +function toAttributes(record: Record): OtlpAttribute[] { + return Object.entries(record).map(([k, v]) => attr(k, v)); +} + +function intPoint( + value: number, + attributes: OtlpAttribute[], + startTimeUnixNano: string, + timeUnixNano: string, +): OtlpNumberDataPoint { + return { + attributes, + startTimeUnixNano, + timeUnixNano, + // OTLP ints are wire-encoded as strings. Coerce non-finite/negative to 0. + asInt: String(Math.max(0, Math.trunc(Number.isFinite(value) ? value : 0))), + }; +} + +function doublePoint( + value: number, + attributes: OtlpAttribute[], + startTimeUnixNano: string, + timeUnixNano: string, +): OtlpNumberDataPoint { + return { + attributes, + startTimeUnixNano, + timeUnixNano, + asDouble: Number.isFinite(value) ? value : 0, + }; +} + +function counter( + name: string, + description: string, + unit: string, + dataPoints: OtlpNumberDataPoint[], +): OtlpMetric { + return { + name, + description, + unit, + sum: { dataPoints, aggregationTemporality: 2, isMonotonic: true }, + }; +} + +function gauge( + name: string, + description: string, + unit: string, + dataPoints: OtlpNumberDataPoint[], +): OtlpMetric { + return { name, description, unit, gauge: { dataPoints } }; +} + +/** + * Attributes for a token group. The grouped dimension is reflected by the key + * the aggregator chose (`groupBy`); we tag it with the matching attribute name + * so a collector sees `model` / `provider` / `node.id` / `agent.id`. + */ +function groupAttributes( + groupBy: TokenAnalytics["groupBy"], + key: string | null, +): OtlpAttribute[] { + if (!groupBy || key === null) return []; + switch (groupBy) { + case "model": + return [attr("model", key)]; + case "provider": + return [attr("provider", key)]; + case "node": + return [attr("node.id", key)]; + case "agent": + return [attr("agent.id", key)]; + } +} + +/** + * Map token + activity analytics to an OTLP/HTTP JSON export payload. + * + * Token/cost metrics emit one data point per group (carrying the group's + * model/provider/node/agent attribute) plus an unattributed grand-total point. + * Cost is omitted from a data point when `cost.usd` is null (unpriced models) so + * an unavailable cost never reports as `$0`. Activity metrics are gauges with no + * group attributes (the activity aggregator is range-scoped, not grouped). + * + * Pure: no I/O, no clock. Returns a fresh payload every call. + */ +export function mapAnalyticsToOtlp(input: OtelMappingInput): OtlpExportPayload { + const { tokens, activity, timeUnixNano } = input; + const start = input.startTimeUnixNano ?? timeUnixNano; + const resourceAttributes = input.resourceAttributes ?? { + "service.name": "fusion-dashboard", + }; + const scopeVersion = input.scopeVersion ?? "1"; + + const p = OTEL_METRIC_PREFIX; + + // ── Token counters (one data point per group + a grand total) ────────── + const inputTokenPoints: OtlpNumberDataPoint[] = []; + const outputTokenPoints: OtlpNumberDataPoint[] = []; + const cachedTokenPoints: OtlpNumberDataPoint[] = []; + const totalTokenPoints: OtlpNumberDataPoint[] = []; + const costPoints: OtlpNumberDataPoint[] = []; + + // Grand totals (unattributed). + inputTokenPoints.push(intPoint(tokens.totals.inputTokens, [], start, timeUnixNano)); + outputTokenPoints.push(intPoint(tokens.totals.outputTokens, [], start, timeUnixNano)); + cachedTokenPoints.push(intPoint(tokens.totals.cachedTokens, [], start, timeUnixNano)); + totalTokenPoints.push(intPoint(tokens.totals.totalTokens, [], start, timeUnixNano)); + if (tokens.cost.usd !== null) { + costPoints.push(doublePoint(tokens.cost.usd, [], start, timeUnixNano)); + } + + // Per-group points. + for (const group of tokens.groups) { + const attrs = groupAttributes(tokens.groupBy, group.key); + inputTokenPoints.push(intPoint(group.inputTokens, attrs, start, timeUnixNano)); + outputTokenPoints.push(intPoint(group.outputTokens, attrs, start, timeUnixNano)); + cachedTokenPoints.push(intPoint(group.cachedTokens, attrs, start, timeUnixNano)); + totalTokenPoints.push(intPoint(group.totalTokens, attrs, start, timeUnixNano)); + if (group.cost.usd !== null) { + costPoints.push(doublePoint(group.cost.usd, attrs, start, timeUnixNano)); + } + } + + const metrics: OtlpMetric[] = [ + counter(`${p}.tokens.input`, "Input (uncached) tokens consumed", "{token}", inputTokenPoints), + counter(`${p}.tokens.output`, "Output tokens generated", "{token}", outputTokenPoints), + counter(`${p}.tokens.cached`, "Cache-read (cached input) tokens", "{token}", cachedTokenPoints), + counter(`${p}.tokens.total`, "Total tokens consumed", "{token}", totalTokenPoints), + counter(`${p}.cost.usd`, "Derived USD cost from token usage", "USD", costPoints), + // ── Activity gauges (point-in-time) ────────────────────────────────── + gauge( + `${p}.activity.active_nodes`, + "Distinct active nodes over the range", + "{node}", + [intPoint(activity.activeNodes, [], start, timeUnixNano)], + ), + gauge( + `${p}.activity.active_agents`, + "Distinct active agents over the range", + "{agent}", + [intPoint(activity.activeAgents, [], start, timeUnixNano)], + ), + gauge( + `${p}.activity.sessions`, + "CLI/chat sessions over the range", + "{session}", + [intPoint(activity.sessions, [], start, timeUnixNano)], + ), + gauge( + `${p}.activity.messages`, + "User messages over the range", + "{message}", + [intPoint(activity.messages, [], start, timeUnixNano)], + ), + gauge( + `${p}.activity.stickiness`, + "Stickiness ratio (DAU/MAU)", + "1", + [doublePoint(activity.stickiness, [], start, timeUnixNano)], + ), + ]; + + return { + resourceMetrics: [ + { + resource: { attributes: toAttributes(resourceAttributes) }, + scopeMetrics: [ + { + scope: { name: p, version: scopeVersion }, + metrics, + }, + ], + }, + ], + }; +} diff --git a/packages/dashboard/src/__tests__/otel-exporter.test.ts b/packages/dashboard/src/__tests__/otel-exporter.test.ts new file mode 100644 index 0000000000..290ec2b10e --- /dev/null +++ b/packages/dashboard/src/__tests__/otel-exporter.test.ts @@ -0,0 +1,250 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; + +import { Database } from "@fusion/core"; +import type { TaskStore } from "@fusion/core"; +import { + resolveOtelExporterConfig, + redactHeadersForDiagnostics, + parseKeyValueList, + startOtelExporter, + maybeStartOtelExporter, + type FetchLike, + type OtelExporterConfig, +} from "../otel-exporter.js"; +import type { RuntimeLogger } from "../runtime-logger.js"; + +interface CapturedLog { + level: "info" | "warn" | "error"; + message: string; + context?: Record; +} + +function makeLogger(): { logger: RuntimeLogger; logs: CapturedLog[] } { + const logs: CapturedLog[] = []; + const mk = (): RuntimeLogger => ({ + info: (message, context) => logs.push({ level: "info", message, context }), + warn: (message, context) => logs.push({ level: "warn", message, context }), + error: (message, context) => logs.push({ level: "error", message, context }), + child: () => mk(), + }); + return { logger: mk(), logs }; +} + +function seedDb(db: Database): void { + db.prepare( + `INSERT INTO tasks + (id, description, "column", createdAt, updatedAt, + tokenUsageInputTokens, tokenUsageOutputTokens, tokenUsageCachedTokens, + tokenUsageCacheWriteTokens, tokenUsageTotalTokens, tokenUsageLastUsedAt, + modelProvider, modelId) + VALUES ('t1', 'd', 'todo', '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z', + 100, 50, 10, 5, 165, '2026-03-01T00:00:00.000Z', 'anthropic', 'claude-opus-4-8')`, + ).run(); +} + +function configFor(overrides: Partial = {}): OtelExporterConfig { + return { + endpoint: "https://collector.example/v1/metrics", + headers: { "DD-API-KEY": "super-secret-token-value" }, + intervalMs: 60_000 as OtelExporterConfig["intervalMs"], + timeoutMs: 5_000, + resourceAttributes: { "service.name": "fusion-dashboard" }, + ...overrides, + }; +} + +describe("resolveOtelExporterConfig (disabled by default)", () => { + it("returns disabled when no endpoint is configured", () => { + expect(resolveOtelExporterConfig({}).kind).toBe("disabled"); + }); + + it("enables when an https endpoint is set", () => { + const r = resolveOtelExporterConfig({ + FUSION_OTEL_METRICS_ENDPOINT: "https://collector:4318/v1/metrics", + FUSION_OTEL_METRICS_HEADERS: "DD-API-KEY=abc,X-Other=1", + }); + expect(r.kind).toBe("enabled"); + if (r.kind !== "enabled") return; + expect(r.warnHttp).toBe(false); + expect(r.config.headers["DD-API-KEY"]).toBe("abc"); + }); + + it("rejects http:// in production", () => { + const r = resolveOtelExporterConfig({ + NODE_ENV: "production", + FUSION_OTEL_METRICS_ENDPOINT: "http://collector:4318/v1/metrics", + }); + expect(r.kind).toBe("rejected"); + }); + + it("allows http:// outside production but flags warnHttp", () => { + const r = resolveOtelExporterConfig({ + FUSION_OTEL_METRICS_ENDPOINT: "http://localhost:4318/v1/metrics", + }); + expect(r.kind).toBe("enabled"); + if (r.kind !== "enabled") return; + expect(r.warnHttp).toBe(true); + }); + + it("rejects a malformed endpoint URL", () => { + const r = resolveOtelExporterConfig({ FUSION_OTEL_METRICS_ENDPOINT: "not a url" }); + expect(r.kind).toBe("rejected"); + }); +}); + +describe("parseKeyValueList / redactHeadersForDiagnostics", () => { + it("parses key=value lists and skips malformed pairs", () => { + expect(parseKeyValueList("a=1, b=2,bad,c=3")).toEqual({ a: "1", b: "2", c: "3" }); + }); + + it("masks all header values, preserving keys", () => { + const r = redactHeadersForDiagnostics({ "DD-API-KEY": "secret", Authorization: "Bearer x" }); + expect(r).toEqual({ "DD-API-KEY": "[REDACTED]", Authorization: "[REDACTED]" }); + }); +}); + +describe("startOtelExporter (with a collector stub)", () => { + let tmpDir: string; + let db: Database; + let store: TaskStore; + + beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), "kb-otel-exporter-")); + db = new Database(join(tmpDir, ".fusion")); + db.init(); + seedDb(db); + store = { getDatabase: () => db } as unknown as TaskStore; + }); + + afterEach(async () => { + db.close(); + await rm(tmpDir, { recursive: true, force: true }); + }); + + it("exports token/cost/activity metrics with expected names + attributes", async () => { + let capturedBody: string | undefined; + const fetchImpl: FetchLike = async (_url, init) => { + capturedBody = init.body; + return { ok: true, status: 200 }; + }; + const { logger } = makeLogger(); + const handle = startOtelExporter({ store, config: configFor(), logger, fetchImpl }); + await handle.exportOnce(); + handle.stop(); + + expect(capturedBody).toBeDefined(); + const payload = JSON.parse(capturedBody!); + const metrics = payload.resourceMetrics[0].scopeMetrics[0].metrics; + const names = metrics.map((m: { name: string }) => m.name); + expect(names).toContain("fusion.command_center.tokens.total"); + expect(names).toContain("fusion.command_center.cost.usd"); + expect(names).toContain("fusion.command_center.activity.active_nodes"); + // model attribute present on a token data point. + const total = metrics.find( + (m: { name: string }) => m.name === "fusion.command_center.tokens.total", + ); + const attributed = total.sum.dataPoints.find( + (p: { attributes: Array<{ key: string }> }) => p.attributes.length > 0, + ); + expect(attributed.attributes[0].key).toBe("model"); + }); + + it("sends configured auth headers but never logs their values", async () => { + let sentHeaders: Record | undefined; + const fetchImpl: FetchLike = async (_url, init) => { + sentHeaders = init.headers; + return { ok: true, status: 200 }; + }; + const { logger, logs } = makeLogger(); + const handle = startOtelExporter({ store, config: configFor(), logger, fetchImpl }); + await handle.exportOnce(); + handle.stop(); + + // The secret IS sent on the wire. + expect(sentHeaders?.["DD-API-KEY"]).toBe("super-secret-token-value"); + // ...but never appears in any log line. + const serialized = JSON.stringify(logs); + expect(serialized).not.toContain("super-secret-token-value"); + }); + + it("backs off and logs (redacted) when the collector is unreachable; never throws", async () => { + const fetchImpl: FetchLike = async () => { + throw new Error("ECONNREFUSED collector down token=super-secret-token-value"); + }; + const { logger, logs } = makeLogger(); + const handle = startOtelExporter({ store, config: configFor(), logger, fetchImpl }); + // Must not throw out of the export. + await expect(handle.exportOnce()).resolves.toBeUndefined(); + handle.stop(); + + const warn = logs.find((l) => l.level === "warn" && l.message.includes("unreachable")); + expect(warn).toBeDefined(); + // The secret embedded in the error message is redacted. + expect(JSON.stringify(logs)).not.toContain("super-secret-token-value"); + // Header values masked in the warn context. + expect((warn?.context?.headers as Record)["DD-API-KEY"]).toBe("[REDACTED]"); + }); + + it("treats a non-2xx response as a failure and backs off, without throwing", async () => { + const fetchImpl: FetchLike = async () => ({ ok: false, status: 503 }); + const { logger, logs } = makeLogger(); + const handle = startOtelExporter({ store, config: configFor(), logger, fetchImpl }); + await expect(handle.exportOnce()).resolves.toBeUndefined(); + handle.stop(); + expect(logs.some((l) => l.level === "warn" && l.context?.status === 503)).toBe(true); + }); +}); + +describe("maybeStartOtelExporter (disabled-by-default gate)", () => { + let tmpDir: string; + let db: Database; + let store: TaskStore; + + beforeEach(() => { + tmpDir = mkdtempSync(join(tmpdir(), "kb-otel-maybe-")); + db = new Database(join(tmpDir, ".fusion")); + db.init(); + store = { getDatabase: () => db } as unknown as TaskStore; + }); + + afterEach(async () => { + db.close(); + await rm(tmpDir, { recursive: true, force: true }); + }); + + it("does NOT start an exporter when no endpoint env is set", () => { + const fetchImpl = vi.fn(async () => ({ ok: true, status: 200 })); + const { logger } = makeLogger(); + const handle = maybeStartOtelExporter({ store, logger, env: {}, fetchImpl }); + expect(handle).toBeNull(); + expect(fetchImpl).not.toHaveBeenCalled(); + }); + + it("logs a warning and does not start when the endpoint is rejected", () => { + const { logger, logs } = makeLogger(); + const handle = maybeStartOtelExporter({ + store, + logger, + env: { NODE_ENV: "production", FUSION_OTEL_METRICS_ENDPOINT: "http://x/v1/metrics" }, + }); + expect(handle).toBeNull(); + expect(logs.some((l) => l.level === "warn" && l.message.includes("NOT started"))).toBe(true); + }); + + it("starts and warns loudly for an http:// endpoint outside production", () => { + const { logger, logs } = makeLogger(); + const handle = maybeStartOtelExporter({ + store, + logger, + env: { FUSION_OTEL_METRICS_ENDPOINT: "http://localhost:4318/v1/metrics" }, + fetchImpl: async () => ({ ok: true, status: 200 }), + }); + expect(handle).not.toBeNull(); + handle?.stop(); + expect(logs.some((l) => l.level === "warn" && l.message.includes("UNENCRYPTED"))).toBe(true); + }); +}); diff --git a/packages/dashboard/src/otel-exporter.ts b/packages/dashboard/src/otel-exporter.ts new file mode 100644 index 0000000000..7c2e586956 --- /dev/null +++ b/packages/dashboard/src/otel-exporter.ts @@ -0,0 +1,365 @@ +/** + * OpenTelemetry (OTLP) metrics exporter wiring (U10) — dashboard side. + * + * Periodically maps the Command Center analytics (tokens / cost / activity) to + * OTLP/HTTP JSON via the pure `mapAnalyticsToOtlp` mapping in `@fusion/core`, + * then POSTs them to a configured collector. **Disabled by default** — nothing + * starts unless an endpoint is explicitly configured. + * + * SDK choice (changeset note): this is a **minimal OTLP/HTTP JSON exporter**, not + * the full `@opentelemetry/*` SDK. The OTLP/HTTP JSON protocol is a single, + * stable `POST /v1/metrics` of a well-defined JSON envelope (produced in core), + * so for a default-disabled feature we avoid pulling the multi-package SDK + * (sdk-metrics + exporter-metrics-otlp-http + resources + api). The wire shape is + * collector-compatible; swapping in the official SDK later is mechanical. + * + * Security: + * - Endpoint is validated on write. In production (`NODE_ENV === "production"`) + * a non-`https:` endpoint is rejected (exporter does not start). Outside + * production an `http://` endpoint is allowed but warns loudly. + * - Auth headers (Datadog/Grafana/etc. tokens) are held in memory only; their + * values are NEVER logged. Header NAMES may appear in diagnostics; header + * VALUES are redacted via `redactSecrets` + an explicit value mask. + * - Collector-unreachable failures log (redacted) and back off exponentially; + * they never throw out of the interval, never crash the server, never block + * a request (the export runs on its own timer). + */ + +import type { TaskStore } from "@fusion/core"; +import { aggregateTokenAnalytics, aggregateActivityAnalytics, mapAnalyticsToOtlp } from "@fusion/core"; +import { redactSecrets } from "@fusion/core"; +import type { RuntimeLogger } from "./runtime-logger.js"; + +/** Resolved, validated exporter configuration. */ +export interface OtelExporterConfig { + /** Full OTLP/HTTP metrics endpoint, e.g. `https://collector:4318/v1/metrics`. */ + endpoint: string; + /** Auth + other headers to send (values are secret-class — never logged). */ + headers: Record; + /** Export interval in ms. */ + intervalMs: string extends never ? never : number; + /** Per-request timeout in ms. */ + timeoutMs: number; + /** Resource attributes (e.g. service.name). */ + resourceAttributes: Record; +} + +/** Minimum/maximum bounds for the export interval (ms). */ +const MIN_INTERVAL_MS = 5_000; +const MAX_INTERVAL_MS = 60 * 60 * 1000; +const DEFAULT_INTERVAL_MS = 60_000; +const DEFAULT_TIMEOUT_MS = 10_000; + +/** Backoff bounds for an unreachable collector. */ +const BACKOFF_BASE_MS = 30_000; +const BACKOFF_MAX_MS = 15 * 60 * 1000; + +/** A single redacted header key (value masked) for diagnostics. */ +const HEADER_VALUE_MASK = "[REDACTED]"; + +function isProduction(env: NodeJS.ProcessEnv): boolean { + return env.NODE_ENV === "production"; +} + +/** + * Parse `key=value,key2=value2` header / attribute lists (the OTEL convention). + * Whitespace around keys/values is trimmed; malformed pairs are skipped. + */ +export function parseKeyValueList(raw: string | undefined): Record { + const out: Record = {}; + if (!raw) return out; + for (const pair of raw.split(",")) { + const eq = pair.indexOf("="); + if (eq <= 0) continue; + const key = pair.slice(0, eq).trim(); + const value = pair.slice(eq + 1).trim(); + if (key) out[key] = value; + } + return out; +} + +function clampInterval(value: number | undefined): number { + if (typeof value !== "number" || !Number.isFinite(value)) return DEFAULT_INTERVAL_MS; + return Math.min(MAX_INTERVAL_MS, Math.max(MIN_INTERVAL_MS, value)); +} + +/** + * Resolve exporter config from environment. Returns `null` (disabled) when no + * endpoint is configured, OR when the endpoint fails production https validation + * (the caller logs the rejection). This is the **disabled-by-default** gate: + * `FUSION_OTEL_METRICS_ENDPOINT` must be explicitly set to enable. + * + * Recognized env: + * - `FUSION_OTEL_METRICS_ENDPOINT` — full `/v1/metrics` URL (required to enable) + * - `FUSION_OTEL_METRICS_HEADERS` — `key=value,key2=value2` auth headers + * - `FUSION_OTEL_METRICS_INTERVAL_MS` — export interval (default 60_000) + * - `FUSION_OTEL_METRICS_TIMEOUT_MS` — per-request timeout (default 10_000) + * - `FUSION_OTEL_RESOURCE_ATTRIBUTES` — `key=value,...` resource attributes + */ +export function resolveOtelExporterConfig( + env: NodeJS.ProcessEnv = process.env, +): + | { kind: "disabled" } + | { kind: "rejected"; reason: string; endpoint: string } + | { kind: "enabled"; config: OtelExporterConfig; warnHttp: boolean } { + const endpoint = env.FUSION_OTEL_METRICS_ENDPOINT?.trim(); + if (!endpoint) return { kind: "disabled" }; + + let url: URL; + try { + url = new URL(endpoint); + } catch { + return { kind: "rejected", reason: "endpoint is not a valid URL", endpoint }; + } + + if (url.protocol !== "https:" && url.protocol !== "http:") { + return { + kind: "rejected", + reason: `unsupported protocol "${url.protocol}" (only http/https)`, + endpoint, + }; + } + + const isHttp = url.protocol === "http:"; + if (isHttp && isProduction(env)) { + return { + kind: "rejected", + reason: "http:// endpoints are not allowed in production (use https://)", + endpoint, + }; + } + + const intervalMs = clampInterval( + env.FUSION_OTEL_METRICS_INTERVAL_MS + ? Number.parseInt(env.FUSION_OTEL_METRICS_INTERVAL_MS, 10) + : undefined, + ); + const timeoutRaw = env.FUSION_OTEL_METRICS_TIMEOUT_MS + ? Number.parseInt(env.FUSION_OTEL_METRICS_TIMEOUT_MS, 10) + : undefined; + const timeoutMs = + typeof timeoutRaw === "number" && Number.isFinite(timeoutRaw) && timeoutRaw > 0 + ? timeoutRaw + : DEFAULT_TIMEOUT_MS; + + const headers = parseKeyValueList(env.FUSION_OTEL_METRICS_HEADERS); + const resourceAttributes = { + "service.name": "fusion-dashboard", + ...parseKeyValueList(env.FUSION_OTEL_RESOURCE_ATTRIBUTES), + }; + + return { + kind: "enabled", + warnHttp: isHttp, + config: { + endpoint, + headers, + intervalMs: intervalMs as OtelExporterConfig["intervalMs"], + timeoutMs, + resourceAttributes, + }, + }; +} + +/** Diagnostic-safe view of headers: keys preserved, values masked + redacted. */ +export function redactHeadersForDiagnostics( + headers: Record, +): Record { + const out: Record = {}; + for (const key of Object.keys(headers)) { + // Never surface the value; mask it. The key alone (e.g. "DD-API-KEY") is + // safe and useful for debugging which auth scheme is configured. + out[key] = HEADER_VALUE_MASK; + } + return out; +} + +/** Minimal fetch-like signature so tests can inject a collector stub. */ +export type FetchLike = ( + url: string, + init: { + method: string; + headers: Record; + body: string; + signal?: AbortSignal; + }, +) => Promise<{ ok: boolean; status: number; text?: () => Promise }>; + +export interface OtelExporterDeps { + store: TaskStore; + config: OtelExporterConfig; + logger: RuntimeLogger; + /** Injectable fetch (defaults to global `fetch`). */ + fetchImpl?: FetchLike; + /** Injectable clock for `timeUnixNano` (defaults to `Date.now`). */ + now?: () => number; +} + +/** + * A running OTLP metrics exporter. Holds an interval that maps current analytics + * and POSTs them. `stop()` clears the timer and any in-flight backoff. + */ +export interface OtelExporterHandle { + /** Run a single export now (used by tests; the timer calls this internally). */ + exportOnce(): Promise; + /** Stop the periodic exporter and release the timer. */ + stop(): void; +} + +/** + * Start the periodic OTLP metrics exporter. The caller is responsible for only + * invoking this when {@link resolveOtelExporterConfig} returned `enabled`. + * + * The export is wrapped so a collector failure logs (redacted) and backs off + * exponentially without ever throwing out of the timer. + */ +export function startOtelExporter(deps: OtelExporterDeps): OtelExporterHandle { + const { store, config, logger } = deps; + const fetchImpl: FetchLike = + deps.fetchImpl ?? ((url, init) => fetch(url, init) as unknown as ReturnType); + const now = deps.now ?? Date.now; + + let stopped = false; + let timer: ReturnType | undefined; + let consecutiveFailures = 0; + + const log = logger.child("otel-exporter"); + + function backoffMs(): number { + if (consecutiveFailures === 0) return config.intervalMs; + const backoff = Math.min( + BACKOFF_MAX_MS, + BACKOFF_BASE_MS * 2 ** (consecutiveFailures - 1), + ); + // Back off, but never poll faster than the configured interval. + return Math.max(config.intervalMs, backoff); + } + + async function exportOnce(): Promise { + // Mapping + DB read are guarded so a malformed snapshot never throws out. + let body: string; + try { + const db = store.getDatabase(); + const tokens = aggregateTokenAnalytics(db, { groupBy: "model", now: now() }); + const activity = aggregateActivityAnalytics(db, {}); + const nowMs = now(); + const payload = mapAnalyticsToOtlp({ + tokens, + activity, + timeUnixNano: String(nowMs * 1_000_000), + resourceAttributes: config.resourceAttributes, + }); + body = JSON.stringify(payload); + } catch (err) { + log.error("Failed to compose OTLP metrics payload", { + message: redactSecrets(err instanceof Error ? err.message : String(err)), + }); + return; + } + + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), config.timeoutMs); + timeout.unref?.(); + try { + const res = await fetchImpl(config.endpoint, { + method: "POST", + headers: { "Content-Type": "application/json", ...config.headers }, + body, + signal: controller.signal, + }); + if (!res.ok) { + consecutiveFailures += 1; + log.warn("OTLP collector returned a non-2xx status; backing off", { + status: res.status, + consecutiveFailures, + // Never log header values; keys only. + headers: redactHeadersForDiagnostics(config.headers), + }); + return; + } + if (consecutiveFailures > 0) { + log.info("OTLP collector reachable again; resuming normal interval", { + afterFailures: consecutiveFailures, + }); + } + consecutiveFailures = 0; + } catch (err) { + consecutiveFailures += 1; + log.warn("OTLP collector unreachable; backing off", { + message: redactSecrets(err instanceof Error ? err.message : String(err)), + consecutiveFailures, + headers: redactHeadersForDiagnostics(config.headers), + }); + } finally { + clearTimeout(timeout); + } + } + + function scheduleNext(): void { + if (stopped) return; + timer = setTimeout(() => { + void exportOnce().finally(scheduleNext); + }, backoffMs()); + timer.unref?.(); + } + + log.info("OTLP metrics exporter started", { + // Endpoint is config (not a secret); headers are masked. + endpoint: config.endpoint, + intervalMs: config.intervalMs, + headers: redactHeadersForDiagnostics(config.headers), + }); + + // First export after one interval (don't block startup). + scheduleNext(); + + return { + exportOnce, + stop() { + stopped = true; + if (timer) clearTimeout(timer); + timer = undefined; + }, + }; +} + +/** + * Convenience wrapper: resolve config from env and, when enabled+valid, start + * the exporter. Returns the handle, or `null` when disabled/rejected (logging + * the rejection). Safe to call unconditionally from server startup — it is a + * no-op unless `FUSION_OTEL_METRICS_ENDPOINT` is set. + */ +export function maybeStartOtelExporter(args: { + store: TaskStore; + logger: RuntimeLogger; + env?: NodeJS.ProcessEnv; + fetchImpl?: FetchLike; + now?: () => number; +}): OtelExporterHandle | null { + const log = args.logger.child("otel-exporter"); + const resolved = resolveOtelExporterConfig(args.env ?? process.env); + if (resolved.kind === "disabled") { + return null; + } + if (resolved.kind === "rejected") { + log.warn("OTLP metrics exporter NOT started (invalid endpoint)", { + endpoint: resolved.endpoint, + reason: resolved.reason, + }); + return null; + } + if (resolved.warnHttp) { + log.warn( + "OTLP metrics endpoint uses http:// — auth tokens will be sent UNENCRYPTED. " + + "Use https:// in any non-local deployment.", + { endpoint: resolved.config.endpoint }, + ); + } + return startOtelExporter({ + store: args.store, + config: resolved.config, + logger: args.logger, + fetchImpl: args.fetchImpl, + now: args.now, + }); +} diff --git a/packages/dashboard/src/server.ts b/packages/dashboard/src/server.ts index 7a95bf7982..f5aca4e322 100644 --- a/packages/dashboard/src/server.ts +++ b/packages/dashboard/src/server.ts @@ -75,6 +75,7 @@ import { recoverAlreadyMergedReviewTasksRecoveriesPerDay, } from "./reliability-metrics.js"; import { loadViewChunkManifest } from "./view-chunk-manifest.js"; +import { maybeStartOtelExporter, type OtelExporterHandle } from "./otel-exporter.js"; const __dirname = dirname(fileURLToPath(import.meta.url)); @@ -1707,6 +1708,10 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT const originalListen = dashboardApp.listen.bind(dashboardApp); const httpsCreds = options?.https; + // U10: OTLP metrics exporter. Disabled by default — only started when + // FUSION_OTEL_METRICS_ENDPOINT is explicitly configured. Held here so the + // server "close" handler can stop its timer. + let otelExporter: OtelExporterHandle | null = null; dashboardApp.listen = ((...args: Parameters) => { const normalizedArgs = normalizeListenArgsForTests(args) as Parameters; @@ -1731,9 +1736,22 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT server = originalListen(...normalizedArgs); } + // U10: start the OTLP exporter (no-op unless FUSION_OTEL_METRICS_ENDPOINT + // is set). Failures here must never break server startup. + try { + otelExporter = maybeStartOtelExporter({ store, logger: runtimeLogger }); + } catch (error) { + runtimeLogger.warn("OTLP metrics exporter failed to start", { + message: "OTLP metrics exporter failed to start", + ...normalizeErrorForLog(error), + }); + } + server.once("close", () => { clearAiSessionCleanupInterval(); aiSessionStore.stopScheduledCleanup(); + otelExporter?.stop(); + otelExporter = null; (apiRouter as Router & { dispose?: () => void }).dispose?.(); void stopAllDevServers().catch((error) => { runtimeLogger.warn("Failed to shutdown dev-server managers", {