feat(command-center): U10 — OpenTelemetry (OTLP) metrics export
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.
This commit is contained in:
gsxdsm
14
.changeset/u10-otel-export.md
Normal file
14
.changeset/u10-otel-export.md
Normal file
@@ -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.
|
||||
207
packages/core/src/__tests__/otel-metrics.test.ts
Normal file
207
packages/core/src/__tests__/otel-metrics.test.ts
Normal file
@@ -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<typeof mapAnalyticsToOtlp>, 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");
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
|
||||
284
packages/core/src/otel-metrics.ts
Normal file
284
packages/core/src/otel-metrics.ts
Normal file
@@ -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<string, string>;
|
||||
/** 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<string, string>): 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,
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
};
|
||||
}
|
||||
250
packages/dashboard/src/__tests__/otel-exporter.test.ts
Normal file
250
packages/dashboard/src/__tests__/otel-exporter.test.ts
Normal file
@@ -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<string, unknown>;
|
||||
}
|
||||
|
||||
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> = {}): 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<string, string> | 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<string, string>)["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<FetchLike>(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);
|
||||
});
|
||||
});
|
||||
365
packages/dashboard/src/otel-exporter.ts
Normal file
365
packages/dashboard/src/otel-exporter.ts
Normal file
@@ -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<string, string>;
|
||||
/** 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<string, string>;
|
||||
}
|
||||
|
||||
/** 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<string, string> {
|
||||
const out: Record<string, string> = {};
|
||||
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<string, string>,
|
||||
): Record<string, string> {
|
||||
const out: Record<string, string> = {};
|
||||
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<string, string>;
|
||||
body: string;
|
||||
signal?: AbortSignal;
|
||||
},
|
||||
) => Promise<{ ok: boolean; status: number; text?: () => Promise<string> }>;
|
||||
|
||||
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<void>;
|
||||
/** 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<FetchLike>);
|
||||
const now = deps.now ?? Date.now;
|
||||
|
||||
let stopped = false;
|
||||
let timer: ReturnType<typeof setTimeout> | 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<void> {
|
||||
// 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,
|
||||
});
|
||||
}
|
||||
@@ -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<typeof dashboardApp.listen>) => {
|
||||
const normalizedArgs = normalizeListenArgsForTests(args) as Parameters<typeof originalListen>;
|
||||
|
||||
@@ -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", {
|
||||
|
||||
Reference in New Issue
Block a user