src/ops-otlp.ts

Functions

Imports

It imports src/http.ts, src/log.ts, src/ops-session-auth.ts, src/signed-requests.ts. Imported by src/routes.ts.

Source

1import { readBoundedText } from "./signed-requests";
2import { resolveSessionCaller, type SessionCaller } from "./ops-session-auth";
3import { logEvent } from "./log";
4import { jsonResponse as json, textResponse as plain } from "./http";
5
6// Claude Code's OpenTelemetry, received as OTLP/HTTP JSON (docs/telemetry.md).
7//
8// POST /ops/otlp/v1/metrics keeps per-session totals in session_usage
9// (migrations/0025): cost, tokens, active time, commits, pull requests and lines of
10// code, per metric, type attribute and model. Delta points are added; a cumulative
11// point replaces the stored total. Nothing else in the export is read.
12//
13// POST /ops/otlp/v1/logs keeps one number: how many api_error events each session
14// reported, as the metric claude_code.api_error keyed by the HTTP status. Every other
15// event is counted as seen and dropped; a log record's body and its other attributes
16// (a prompt, when OTEL_LOG_USER_PROMPTS is on; a tool's input) are never read into a
17// value that is stored.
18//
19// ALLOWLIST, NOT DENYLIST. What reaches D1 is: the session id, the job id, the metric
20// name from METRICS below, a type from that metric's list, a model name that matches
21// MODEL_SHAPE, and a number. A new attribute Claude Code starts sending is ignored
22// until it is added here. (test/ops-otlp.test.ts, the PLANT test.)
23
24export const OTLP_METRICS_PATH = "/ops/otlp/v1/metrics";
25export const OTLP_LOGS_PATH = "/ops/otlp/v1/logs";
26
27// 1MB, after decompression. The compressed body is held to the same cap.
28export const OTLP_MAX_BYTES = 1024 * 1024;
29// Bounds on one request, so a runaway exporter costs one bounded batch.
30const OTLP_MAX_SERIES = 200;
31const OTLP_MAX_SESSIONS = 16;
32
33// The metrics kept, and the type attribute values each may carry. An empty list means
34// the metric has no type attribute, and its kind is ''.
35const METRICS: Record<string, readonly string[]> = {
36  "claude_code.cost.usage": [],
37  "claude_code.token.usage": ["input", "output", "cacheRead", "cacheCreation"],
38  "claude_code.active_time.total": ["user", "cli"],
39  "claude_code.commit.count": [],
40  "claude_code.pull_request.count": [],
41  "claude_code.lines_of_code.count": ["added", "removed"],
42};
43const API_ERROR_METRIC = "claude_code.api_error";
44
45const SESSION_SHAPE = /^[A-Za-z0-9._:-]{1,128}$/;
46const MODEL_SHAPE = /^[A-Za-z0-9._:@/[\]-]{1,100}$/;
47const JOB_SHAPE = /^job_[0-9a-f]{12}$/;
48const STATUS_SHAPE = /^[1-5][0-9]{2}$/;
49
50// OTLP's AggregationTemporality: 1 delta, 2 cumulative. JSON carries the number; the
51// enum name is accepted too.
52type Temporality = "delta" | "cumulative";
53
54interface UsagePoint {
55  session_id: string;
56  // The capsid.job_id attribute, when the exporter sent one that is a job id.
57  job_attr: string | null;
58  metric: string;
59  kind: string;
60  model: string;
61  value: number;
62  temporality: Temporality;
63  // For "latest wins" among cumulative points of one series in one request.
64  at: bigint;
65}
66
67interface ParsedExport {
68  points: UsagePoint[];
69  // Points of a kept metric that could not be read: no session id, a value that is
70  // not a non-negative number, an unknown type, a gauge. Reported as partialSuccess.
71  rejected: number;
72  problems: string[];
73}
74
75type Attr = { key?: unknown; value?: Record<string, unknown> };
76
77// Only scalar attribute values are read. An arrayValue or kvlistValue is skipped.
78function attrMap(list: unknown): Map<string, string> {
79  const out = new Map<string, string>();
80  if (!Array.isArray(list)) return out;
81  for (const item of list as Attr[]) {
82    if (typeof item?.key !== "string" || !item.value || typeof item.value !== "object") continue;
83    const v = item.value;
84    if (typeof v.stringValue === "string") out.set(item.key, v.stringValue);
85    else if (typeof v.intValue === "string" || typeof v.intValue === "number") out.set(item.key, String(v.intValue));
86    else if (typeof v.doubleValue === "number") out.set(item.key, String(v.doubleValue));
87    else if (typeof v.boolValue === "boolean") out.set(item.key, String(v.boolValue));
88  }
89  return out;
90}
91
92function temporalityOf(raw: unknown): Temporality | null {
93  if (raw === 1 || raw === "AGGREGATION_TEMPORALITY_DELTA") return "delta";
94  if (raw === 2 || raw === "AGGREGATION_TEMPORALITY_CUMULATIVE") return "cumulative";
95  return null;
96}
97
98/** A data point's value: asDouble, or asInt, which JSON may carry as a decimal string. */
99export function pointValue(point: Record<string, unknown>): number | null {
100  let value: number | null = null;
101  if (typeof point.asDouble === "number") value = point.asDouble;
102  else if (typeof point.asInt === "number") value = point.asInt;
103  else if (typeof point.asInt === "string" && /^-?[0-9]{1,18}$/.test(point.asInt)) value = Number(point.asInt);
104  if (value === null || !Number.isFinite(value) || value < 0) return null;
105  return value;
106}
107
108function timeOf(point: Record<string, unknown>): bigint {
109  const raw = point.timeUnixNano;
110  if (typeof raw === "string" && /^[0-9]{1,20}$/.test(raw)) return BigInt(raw);
111  if (typeof raw === "number" && Number.isSafeInteger(raw) && raw >= 0) return BigInt(raw);
112  return 0n;
113}
114
115const arr = (v: unknown): Record<string, unknown>[] => (Array.isArray(v) ? (v as Record<string, unknown>[]).filter((x) => x && typeof x === "object") : []);
116
117/** The kept points of an ExportMetricsServiceRequest. Throws nothing; a body that is
118 *  not the expected shape yields no points. */
119export function parseMetrics(body: unknown): ParsedExport {
120  const out: ParsedExport = { points: [], rejected: 0, problems: [] };
121  const problem = (message: string) => {
122    out.rejected += 1;
123    if (out.problems.length < 5 && !out.problems.includes(message)) out.problems.push(message);
124  };
125  const root = body && typeof body === "object" ? (body as Record<string, unknown>) : {};
126  for (const rm of arr(root.resourceMetrics)) {
127    const resource = attrMap((rm.resource as Record<string, unknown> | undefined)?.attributes);
128    for (const sm of arr(rm.scopeMetrics)) {
129      for (const metric of arr(sm.metrics)) {
130        const name = metric.name;
131        // Not one of ours: ignored, not rejected. The exporter did nothing wrong.
132        if (typeof name !== "string" || !Object.hasOwn(METRICS, name)) continue;
133        const sum = metric.sum as Record<string, unknown> | undefined;
134        if (!sum || typeof sum !== "object") {
135          const gaugePoints = arr((metric.gauge as Record<string, unknown> | undefined)?.dataPoints).length;
136          for (let i = 0; i < gaugePoints; i += 1) problem(`${name} is not a sum`);
137          continue;
138        }
139        const temporality = temporalityOf(sum.aggregationTemporality);
140        for (const dp of arr(sum.dataPoints)) {
141          if (!temporality) {
142            problem(`${name} has no aggregationTemporality`);
143            continue;
144          }
145          const attrs = new Map([...resource, ...attrMap(dp.attributes)]);
146          const session = attrs.get("session.id");
147          if (!session || !SESSION_SHAPE.test(session)) {
148            problem(`${name} point without a usable session.id`);
149            continue;
150          }
151          const value = pointValue(dp);
152          if (value === null) {
153            problem(`${name} point with a value that is not a non-negative number`);
154            continue;
155          }
156          const kinds = METRICS[name];
157          const kind = kinds.length === 0 ? "" : (attrs.get("type") ?? "");
158          if (kinds.length > 0 && !kinds.includes(kind)) {
159            problem(`${name} point with an unknown type`);
160            continue;
161          }
162          const model = attrs.get("model") ?? "";
163          if (model !== "" && !MODEL_SHAPE.test(model)) {
164            problem(`${name} point with a model name this receiver does not accept`);
165            continue;
166          }
167          const job = attrs.get("capsid.job_id");
168          out.points.push({
169            session_id: session,
170            job_attr: job && JOB_SHAPE.test(job) ? job : null,
171            metric: name,
172            kind,
173            model,
174            value,
175            temporality,
176            at: timeOf(dp),
177          });
178        }
179      }
180    }
181  }
182  return out;
183}
184
185/** api_error events from an ExportLogsServiceRequest, as delta points of
186 *  claude_code.api_error keyed by status. Every other record is skipped. */
187export function parseLogs(body: unknown): ParsedExport {
188  const out: ParsedExport = { points: [], rejected: 0, problems: [] };
189  const root = body && typeof body === "object" ? (body as Record<string, unknown>) : {};
190  for (const rl of arr(root.resourceLogs)) {
191    const resource = attrMap((rl.resource as Record<string, unknown> | undefined)?.attributes);
192    for (const sl of arr(rl.scopeLogs)) {
193      for (const record of arr(sl.logRecords)) {
194        const attrs = new Map([...resource, ...attrMap(record.attributes)]);
195        const event = attrs.get("event.name");
196        // Claude Code names the event in event.name ("api_error") and in the body
197        // ("claude_code.api_error"). The body is compared, never kept.
198        const bodyName = (record.body as Record<string, unknown> | undefined)?.stringValue;
199        const isApiError = event === "api_error" || event === API_ERROR_METRIC || bodyName === API_ERROR_METRIC;
200        if (!isApiError) continue;
201        const session = attrs.get("session.id");
202        if (!session || !SESSION_SHAPE.test(session)) {
203          out.rejected += 1;
204          if (out.problems.length === 0) out.problems.push("api_error record without a usable session.id");
205          continue;
206        }
207        const status = attrs.get("status_code") ?? "";
208        const model = attrs.get("model") ?? "";
209        const job = attrs.get("capsid.job_id");
210        out.points.push({
211          session_id: session,
212          job_attr: job && JOB_SHAPE.test(job) ? job : null,
213          metric: API_ERROR_METRIC,
214          kind: STATUS_SHAPE.test(status) ? status : "",
215          model: MODEL_SHAPE.test(model) ? model : "",
216          value: 1,
217          temporality: "delta",
218          at: timeOf(record),
219        });
220      }
221    }
222  }
223  return out;
224}
225
226interface SeriesWrite {
227  session_id: string;
228  metric: string;
229  kind: string;
230  model: string;
231  value: number;
232  mode: "add" | "replace";
233}
234
235/** One write per series: delta points of a series summed, cumulative points of a
236 *  series reduced to the newest. */
237export function aggregate(points: UsagePoint[]): SeriesWrite[] {
238  const deltas = new Map<string, SeriesWrite>();
239  const latest = new Map<string, { write: SeriesWrite; at: bigint }>();
240  for (const p of points) {
241    const key = JSON.stringify([p.session_id, p.metric, p.kind, p.model]);
242    if (p.temporality === "delta") {
243      const seen = deltas.get(key);
244      if (seen) seen.value += p.value;
245      else deltas.set(key, { session_id: p.session_id, metric: p.metric, kind: p.kind, model: p.model, value: p.value, mode: "add" });
246    } else {
247      const seen = latest.get(key);
248      if (!seen || p.at >= seen.at) {
249        latest.set(key, { write: { session_id: p.session_id, metric: p.metric, kind: p.kind, model: p.model, value: p.value, mode: "replace" }, at: p.at });
250      }
251    }
252  }
253  return [...deltas.values(), ...[...latest.values()].map((l) => l.write)];
254}
255
256/**
257 * The job each session's usage is recorded against. A session already bound (its
258 * agent_sessions row, written by the hook receiver) keeps that job. Otherwise the
259 * caller's binding. The capsid.job_id attribute never picks a job on its own: it is
260 * set by whoever configured the exporter, so it is honoured only where it agrees with
261 * the binding, and a disagreement is reported rather than recorded.
262 */
263export function jobForSession(caller: SessionCaller, bound: string | null | undefined, attr: string | null): { job: string | null; mismatch: boolean } {
264  const job = bound ?? caller.job_id;
265  return { job, mismatch: attr !== null && attr !== job };
266}
267
268// A session belongs to the key that first reported it, by hook or by telemetry. The
269// claim: an agent_sessions row for a session nobody has reported yet, and nothing for
270// one that exists (the hook receiver owns every other column).
271// Each statement is one literal at its prepare, so test-integration/query-plans.test.ts
272// can reconstruct and EXPLAIN it.
273function claimSession(db: D1Database, id: string, actor: string, jobId: string | null, namespace: string | null, at: string): D1PreparedStatement {
274  return db
275    .prepare(
276      `INSERT INTO agent_sessions (session_id, agent, job_id, namespace, started_at, last_event_at, last_event, updated_at)
277       VALUES (?1, ?2, ?3, ?4, ?5, ?5, 'otlp', ?5)
278       ON CONFLICT(session_id) DO NOTHING`
279    )
280    .bind(id, actor, jobId, namespace, at);
281}
282
283// The upsert, conditional on the session being this caller's (?8), in the same batch as
284// the claim, so a session another key owns is never written whatever order two
285// requests commit in. RETURNING says which series landed; never meta.changes
286// (path mutation invariant, test/path-mutation.test.ts). The job a row was first written with is kept (COALESCE).
287// The SELECT carries a WHERE, which SQLite needs to read the ON CONFLICT that follows.
288// Delta points add to the stored total; cumulative points replace it.
289function upsertUsage(
290  db: D1Database,
291  mode: "add" | "replace",
292  w: { session_id: string; metric: string; kind: string; model: string; value: number },
293  jobId: string | null,
294  at: string,
295  actor: string
296): D1PreparedStatement {
297  const statement =
298    mode === "add"
299      ? db.prepare(
300          `INSERT INTO session_usage (session_id, job_id, metric, kind, model, value, updated_at)
301           SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7
302           WHERE EXISTS (SELECT 1 FROM agent_sessions WHERE session_id = ?1 AND agent = ?8)
303           ON CONFLICT(session_id, metric, kind, model) DO UPDATE SET
304             value = session_usage.value + excluded.value,
305             job_id = COALESCE(session_usage.job_id, excluded.job_id),
306             updated_at = excluded.updated_at
307           RETURNING session_id`
308        )
309      : db.prepare(
310          `INSERT INTO session_usage (session_id, job_id, metric, kind, model, value, updated_at)
311           SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7
312           WHERE EXISTS (SELECT 1 FROM agent_sessions WHERE session_id = ?1 AND agent = ?8)
313           ON CONFLICT(session_id, metric, kind, model) DO UPDATE SET
314             value = excluded.value,
315             job_id = COALESCE(session_usage.job_id, excluded.job_id),
316             updated_at = excluded.updated_at
317           RETURNING session_id`
318        );
319  return statement.bind(w.session_id, jobId, w.metric, w.kind, w.model, w.value, at, actor);
320}
321
322/** The request body as text: identity or gzip, at most OTLP_MAX_BYTES either side of
323 *  decompression. */
324export async function readOtlpBody(request: Request): Promise<{ ok: true; text: string } | { ok: false; response: Response }> {
325  const type = (request.headers.get("Content-Type") ?? "").split(";")[0].trim().toLowerCase();
326  if (type !== "application/json") {
327    return { ok: false, response: plain("unsupported media type: this receiver takes OTLP/HTTP JSON (OTEL_EXPORTER_OTLP_PROTOCOL=http/json), not protobuf", 415) };
328  }
329  const length = Number(request.headers.get("Content-Length") ?? "0");
330  if (Number.isFinite(length) && length > OTLP_MAX_BYTES) {
331    return { ok: false, response: plain(`payload too large: the body exceeds ${OTLP_MAX_BYTES} bytes`, 413) };
332  }
333  const encoding = (request.headers.get("Content-Encoding") ?? "identity").trim().toLowerCase();
334  if (encoding !== "identity" && encoding !== "gzip") {
335    return { ok: false, response: plain(`unsupported content encoding '${encoding}': send gzip or none`, 415) };
336  }
337  if (!request.body) return { ok: true, text: "" };
338  let stream: ReadableStream<Uint8Array> = request.body;
339  if (encoding === "gzip") {
340    // The compressed bytes are capped on the way in too, by counting what the
341    // decompressor is fed.
342    let fed = 0;
343    const cap = new TransformStream<Uint8Array, Uint8Array>({
344      transform(chunk, controller) {
345        fed += chunk.byteLength;
346        if (fed > OTLP_MAX_BYTES) controller.error(new Error("compressed body too large"));
347        else controller.enqueue(chunk);
348      },
349    });
350    stream = request.body.pipeThrough(cap).pipeThrough(new DecompressionStream("gzip") as unknown as TransformStream<Uint8Array, Uint8Array>);
351  }
352  try {
353    const read = await readBoundedText({ body: stream }, OTLP_MAX_BYTES);
354    if (!read.ok) return { ok: false, response: plain(`payload too large: the body exceeds ${OTLP_MAX_BYTES} bytes after decompression`, 413) };
355    return { ok: true, text: read.text };
356  } catch (err) {
357    const message = err instanceof Error ? err.message : String(err);
358    if (/too large/.test(message)) return { ok: false, response: plain(`payload too large: the compressed body exceeds ${OTLP_MAX_BYTES} bytes`, 413) };
359    return { ok: false, response: plain(`bad request: the gzip body does not decompress (${message})`, 400) };
360  }
361}
362
363/** Parse, bind and write one export. Returns the OTLP response. */
364export async function recordExport(
365  db: D1Database,
366  caller: SessionCaller,
367  parsed: ParsedExport,
368  now: Date
369): Promise<Response> {
370  let rejected = parsed.rejected;
371  const problems = [...parsed.problems];
372
373  // Bounded: at most OTLP_MAX_SESSIONS sessions per request; points of the rest are
374  // rejected.
375  const sessions = [...new Set(parsed.points.map((p) => p.session_id))];
376  const kept = new Set(sessions.slice(0, OTLP_MAX_SESSIONS));
377  if (sessions.length > kept.size) problems.push(`more than ${OTLP_MAX_SESSIONS} sessions in one request`);
378  const points = parsed.points.filter((p) => {
379    if (kept.has(p.session_id)) return true;
380    rejected += 1;
381    return false;
382  });
383
384  // Who owns each session already. A session another key reported first is refused
385  // here, and the conditional upsert below refuses it again at commit.
386  const actor = caller.agent.actor;
387  const bound = new Map<string, string | null>();
388  const foreign = new Set<string>();
389  if (kept.size > 0) {
390    const { results } = await db
391      .prepare("SELECT session_id, agent, job_id FROM agent_sessions WHERE session_id IN (SELECT value FROM json_each(?1))")
392      .bind(JSON.stringify([...kept]))
393      .all<{ session_id: string; agent: string; job_id: string | null }>();
394    for (const row of results ?? []) {
395      if (row.agent === actor) bound.set(row.session_id, row.job_id);
396      else foreign.add(row.session_id);
397    }
398  }
399  if (foreign.size > 0) {
400    problems.push(`${foreign.size} session(s) were first reported by another key, so their points were not recorded`);
401  }
402  const owned = points.filter((p) => {
403    if (!foreign.has(p.session_id)) return true;
404    rejected += 1;
405    return false;
406  });
407  const jobOf = new Map<string, string | null>();
408  let mismatched = 0;
409  for (const p of owned) {
410    const { job, mismatch } = jobForSession(caller, bound.get(p.session_id), p.job_attr);
411    jobOf.set(p.session_id, job);
412    if (mismatch) mismatched += 1;
413  }
414  if (mismatched > 0) problems.push(`${mismatched} point(s) named a capsid.job_id other than the job this caller is bound to; they were recorded against the bound job`);
415
416  let writes = aggregate(owned);
417  if (writes.length > OTLP_MAX_SERIES) {
418    problems.push(`more than ${OTLP_MAX_SERIES} series in one request`);
419    // The points behind the dropped series are not counted one by one; each dropped
420    // series counts as one.
421    rejected += writes.length - OTLP_MAX_SERIES;
422    writes = writes.slice(0, OTLP_MAX_SERIES);
423  }
424  const at = now.toISOString();
425  if (writes.length > 0) {
426    const claims = [...new Set(writes.map((w) => w.session_id))].filter((id) => !bound.has(id));
427    try {
428      const results = await db.batch([
429        ...claims.map((id) => claimSession(db, id, actor, caller.job_id, caller.namespace, at)),
430        ...writes.map((w) => upsertUsage(db, w.mode, w, jobOf.get(w.session_id) ?? null, at, actor)),
431      ]);
432      // A series whose upsert returned no row was refused by the ownership condition:
433      // another key claimed the session between the read above and this batch.
434      const refused = results.slice(claims.length).filter((r) => (r.results ?? []).length === 0).length;
435      if (refused > 0) {
436        rejected += refused;
437        problems.push(`${refused} series belonged to a session another key reported first, so they were not recorded`);
438      }
439    } catch (err) {
440      // Not swallowed (CLAUDE.md, no swallowed error rule): logged, and answered 503 so
441      // the exporter retries rather than losing the points.
442      const message = err instanceof Error ? err.message : String(err);
443      logEvent("error", "OTLP_WRITE_FAILED", { message: `OTLP_WRITE_FAILED ${caller.agent.actor}: ${message}` });
444      return plain(`service unavailable: the usage could not be written (${message})`, 503);
445    }
446  }
447  if (rejected === 0 && problems.length === 0) return json({});
448  return json({ partialSuccess: { rejectedDataPoints: rejected, errorMessage: problems.join("; ") } });
449}
450
451async function handle(request: Request, env: { DB: D1Database; OPERATOR_KEY_HASH?: string }, parse: (body: unknown) => ParsedExport): Promise<Response> {
452  const caller = await resolveSessionCaller(request, env);
453  if (caller instanceof Response) return caller;
454  const body = await readOtlpBody(request);
455  if (!body.ok) return body.response;
456  let decoded: unknown;
457  try {
458    decoded = JSON.parse(body.text);
459  } catch {
460    return plain("bad request: the body is not JSON", 400);
461  }
462  if (!decoded || typeof decoded !== "object" || Array.isArray(decoded)) return plain("bad request: the body is not an OTLP export object", 400);
463  return recordExport(env.DB, caller, parse(decoded), new Date());
464}
465
466export function handleOtlpMetrics(request: Request, env: { DB: D1Database; OPERATOR_KEY_HASH?: string }): Promise<Response> {
467  return handle(request, env, parseMetrics);
468}
469
470export function handleOtlpLogs(request: Request, env: { DB: D1Database; OPERATOR_KEY_HASH?: string }): Promise<Response> {
471  return handle(request, env, parseLogs);
472}