Skip to content
Sign up

flagon-io/g1t

public
0
1,198 linesCodeBlameRaw
1/**
2 * One project's telemetry: a Durable Object named by the project's id,
3 * holding its requests, logs, spans, metrics, error groups, ingest tokens
4 * and exports in its own SQLite (./schema.ts). One object per project
5 * means each project's store grows, and is queried, on its own.
6 *
7 * Writes arrive from the service's two doors (`ingest`); reads are the
8 * service's RPC methods, called once the site or the API has checked the
9 * viewer may read the project. Open live tails are hibernating WebSockets
10 * here, each handed what arrives as it arrives.
11 *
12 * An alarm runs while there is work: each minute it counts requests into
13 * per-minute rollups and sends to exports what is waiting; each hour it
14 * removes what is past the project's retention.
15 */
16import {
17 DEFAULT_TELEMETRY_SETTINGS,
18 RETENTION_CHOICES,
19 TELEMETRY_TOKEN_PREFIX,
20 type Attributes,
21 type ErrorGroup,
22 type IngestToken,
23 type LiveItem,
24 type LogQuery,
25 type MetricInfo,
26 type MetricSeries,
27 type TelemetryOverview,
28 type TelemetryTotals,
29 type Page,
30 type RequestQuery,
31 type LogSeverity,
32 type TelemetryExport,
33 type TelemetryFilter,
34 type TelemetryLog,
35 type TelemetryRequest,
36 type TelemetrySettings,
37 type TelemetrySpan,
38 type TelemetryUsage,
39 type TraceDetail,
40 type TraceQuery,
41 type TraceSummary,
42} from "@g1t/contracts/telemetry";
43
44import type { Batch } from "./batch.ts";
45import { fingerprint } from "./errors.ts";
46import { exportEndpointProblem, otlpBodies, send } from "./export.ts";
47import { addInto, bucketOf, decode, emptyHistogram, encode, percentiles } from "./histogram.ts";
48import { FTS, MIGRATIONS, ROLLUP_DAYS } from "./schema.ts";
49import { overCap, sizeOf } from "./quota.ts";
50import { SEVERITY_FLOOR, randomId, severityOf } from "./shape.ts";
51import type { RequestHalf } from "./tail.ts";
52
53const MINUTE_US = 60_000_000;
54const DAY_US = 86_400_000_000;
55/** Requests are counted once they are this old, so both halves have arrived. */
56const ROLL_AFTER_US = 90_000_000;
57/** A store larger than this sheds its oldest day until it is not. */
58const MAX_BYTES = 4 * 1024 ** 3;
59/** At most this many items a second go to each live tail; the rest are counted as dropped. */
60const LIVE_PER_BATCH = 200;
61const STEPS = [60, 300, 900, 1800, 3600, 3 * 3600, 6 * 3600, 86400];
62
63type Row = Record<string, SqlStorageValue>;
64
65const json = <T>(text: SqlStorageValue, fallback: T): T => {
66 if (typeof text !== "string" || text === "") return fallback;
67 try {
68 return JSON.parse(text) as T;
69 } catch {
70 return fallback;
71 }
72};
73const strOrNull = (v: SqlStorageValue) => (v === null || v === undefined || v === "" ? null : String(v));
74const numOrNull = (v: SqlStorageValue) => (v === null || v === undefined ? null : Number(v));
75
76async function sha256(text: string): Promise<string> {
77 const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(text));
78 return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, "0")).join("");
79}
80
81function month(us = Date.now() * 1000): string {
82 return new Date(us / 1000).toISOString().slice(0, 7);
83}
84
85/** `2026-10-12`: the UTC day, for a free workspace's daily share. */
86function day(us = Date.now() * 1000): string {
87 return new Date(us / 1000).toISOString().slice(0, 10);
88}
89
90/** A page's cursor: the last row's time and id, so the next page starts after it. */
91function cursorOf(time: number, id: string): string {
92 return btoa(`${time}:${id}`);
93}
94function readCursor(cursor: string | null | undefined): { time: number; id: string } | null {
95 if (!cursor) return null;
96 try {
97 const [time, id] = atob(cursor).split(":");
98 return Number.isFinite(Number(time)) && id ? { time: Number(time), id } : null;
99 } catch {
100 return null;
101 }
102}
103
104/** A limit within reason. */
105const limitOf = (n: number | null | undefined, fallback = 100, max = 500) => Math.min(max, Math.max(1, Math.trunc(n ?? fallback)));
106
107/**
108 * Search text split into what it says: `key=value` and `key:value` pairs
109 * filter by attribute (or by a column the row has), the rest are words to find.
110 */
111export function parseSearch(text: string | null | undefined): { words: string[]; pairs: [string, string][] } {
112 const words: string[] = [];
113 const pairs: [string, string][] = [];
114 for (const m of (text ?? "").matchAll(/"([^"]+)"|(\S+)/g)) {
115 const token = m[1] ?? m[2];
116 const pair = m[1] ? null : /^([\w.@/-]+)[=:](.+)$/.exec(token);
117 if (pair && !/^https?$/.test(pair[1])) pairs.push([pair[1], pair[2]]);
118 else words.push(token);
119 }
120 return { words, pairs };
121}
122
123/** An FTS5 query in which every word must appear (as a prefix, for the last). */
124export function ftsQuery(words: string[]): string {
125 return words.map((w, i) => `"${w.replace(/"/g, '""')}"${i === words.length - 1 && !/\s/.test(w) ? "*" : ""}`).join(" AND ");
126}
127
128/**
129 * The store's logic, apart from the Durable Object it runs in
130 * (./durable.ts), so tests run it on Node's SQLite.
131 */
132export class Store {
133 state!: DurableObjectState;
134 sql!: SqlStorage;
135 fts = false;
136
137 constructor(state: DurableObjectState) {
138 this.open(state);
139 }
140
141 open(state: DurableObjectState) {
142 this.state = state;
143 this.sql = state.storage.sql;
144 this.migrate();
145 }
146
147 migrate() {
148 this.sql.exec("CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)");
149 const done = Number(this.sql.exec("SELECT value FROM meta WHERE key = 'migrations'").toArray()[0]?.value ?? 0);
150 for (let i = done; i < MIGRATIONS.length; i++) {
151 this.sql.exec(MIGRATIONS[i]);
152 this.sql.exec("INSERT INTO meta (key, value) VALUES ('migrations', ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value", String(i + 1));
153 }
154 try {
155 this.sql.exec(FTS);
156 this.fts = true;
157 } catch {
158 this.fts = false;
159 }
160 }
161
162 rows(query: string, ...args: SqlStorageValue[]): Row[] {
163 return this.sql.exec(query, ...args).toArray() as Row[];
164 }
165
166 // ---- Settings -----------------------------------------------------------
167
168 settings(): TelemetrySettings {
169 const out: TelemetrySettings = { ...DEFAULT_TELEMETRY_SETTINGS };
170 for (const row of this.rows("SELECT key, value FROM settings")) {
171 if (row.key === "retention_days") out.retention_days = Number(row.value);
172 if (row.key === "deployments") out.deployments = row.value === "1";
173 }
174 return out;
175 }
176
177 setSettings(change: Partial<TelemetrySettings>): TelemetrySettings {
178 const put = (key: string, value: string) =>
179 this.sql.exec("INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value", key, value);
180 if (change.retention_days !== undefined) {
181 if (!(RETENTION_CHOICES as readonly number[]).includes(change.retention_days)) throw new Error(`Retention is one of ${RETENTION_CHOICES.join(", ")} days.`);
182 put("retention_days", String(change.retention_days));
183 }
184 if (change.deployments !== undefined) put("deployments", change.deployments ? "1" : "0");
185 return this.settings();
186 }
187
188 /** The workspace the project is in, by slug, as the site last said; null before it has. */
189 workspaceOf(): string | null {
190 const row = this.rows("SELECT value FROM settings WHERE key = 'workspace'")[0];
191 return row ? String(row.value) : null;
192 }
193
194 setWorkspace(slug: string): void {
195 const clean = slug.trim().toLowerCase();
196 if (!/^[a-z0-9][a-z0-9-]{0,38}$/.test(clean) || clean === this.workspaceOf()) return;
197 this.sql.exec("INSERT INTO settings (key, value) VALUES ('workspace', ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value", clean);
198 }
199
200 /** Bytes that arrived today (UTC). */
201 todayBytes(): number {
202 return Number(this.rows("SELECT ingested_bytes FROM usage_days WHERE day = ?", day())[0]?.ingested_bytes ?? 0);
203 }
204
205 // ---- Tokens -------------------------------------------------------------
206
207 tokenRow(row: Row): IngestToken {
208 return {
209 id: String(row.id),
210 name: String(row.name),
211 hint: String(row.hint),
212 created_by: strOrNull(row.created_by),
213 created_us: Number(row.created_us),
214 last_used_us: numOrNull(row.last_used_us),
215 managed: row.managed === 1,
216 };
217 }
218
219 /**
220 * The tokens people made. The one g1t's deployments send with is kept
221 * out of sight: it is g1t's, given to the project's apps on their own.
222 */
223 tokens(): IngestToken[] {
224 return this.rows("SELECT * FROM tokens WHERE managed = 0 ORDER BY created_us DESC").map((r) => this.tokenRow(r));
225 }
226
227 /** A new token: `g1to_<project>_<secret>`, so the ingest door knows whose store to ask. */
228 async createToken(projectId: string, name: string, createdBy: string | null, managed = false): Promise<{ token: IngestToken; secret: string }> {
229 const secret = `${TELEMETRY_TOKEN_PREFIX}${projectId}_${randomId(24)}`;
230 const id = randomId(8);
231 const created = Date.now() * 1000;
232 this.sql.exec(
233 "INSERT INTO tokens (id, name, hash, hint, created_by, created_us, managed, secret) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
234 id,
235 name.trim().slice(0, 80) || "Ingest token",
236 await sha256(secret),
237 `${secret.slice(0, TELEMETRY_TOKEN_PREFIX.length + 4)}…${secret.slice(-4)}`,
238 createdBy,
239 created,
240 managed ? 1 : 0,
241 managed ? secret : null,
242 );
243 return { token: this.tokenRow(this.rows("SELECT * FROM tokens WHERE id = ?", id)[0]), secret };
244 }
245
246 revokeToken(id: string): boolean {
247 return this.sql.exec("DELETE FROM tokens WHERE id = ? AND managed = 0", id).rowsWritten > 0;
248 }
249
250 async deploymentToken(projectId: string): Promise<{ secret: string }> {
251 const row = this.rows("SELECT secret FROM tokens WHERE managed = 1 LIMIT 1")[0];
252 if (row?.secret) return { secret: String(row.secret) };
253 const made = await this.createToken(projectId, "Deployments on g1t.page", null, true);
254 return { secret: made.secret };
255 }
256
257 /** Whether `secret` is one of this project's tokens; marks it used. */
258 async checkToken(secret: string): Promise<boolean> {
259 const hash = await sha256(secret);
260 const row = this.rows("SELECT id, last_used_us FROM tokens WHERE hash = ?", hash)[0];
261 if (!row) return false;
262 const now = Date.now() * 1000;
263 if (!row.last_used_us || now - Number(row.last_used_us) > MINUTE_US) {
264 this.sql.exec("UPDATE tokens SET last_used_us = ? WHERE id = ?", now, row.id);
265 }
266 return true;
267 }
268
269 // ---- Ingest -------------------------------------------------------------
270
271 /**
272 * Adds what arrived. `token` is checked first when given (the OTLP door);
273 * tail events come from Cloudflare and need none. Returns false for a
274 * token that is not this project's, or when this kind of telemetry is off.
275 *
276 * `dailyCap`, for a project of a workspace without the plan, is the
277 * day's share of its free observability (../quota.ts): past it, OTLP is
278 * refused (`quota`), and tail events keep their requests without logs.
279 */
280 async ingest(batch: Batch, halves: RequestHalf[] = [], token?: string, dailyCap?: number | null): Promise<{ accepted: boolean; reason?: string }> {
281 if (token !== undefined && !(await this.checkToken(token))) return { accepted: false, reason: "token" };
282 if (token === undefined && !this.settings().deployments) return { accepted: false, reason: "off" };
283 if (overCap(this.todayBytes(), batch.bytes, dailyCap)) {
284 if (token !== undefined) return { accepted: false, reason: "quota" };
285 batch = { ...batch, logs: [], spans: [], metrics: [], errors: [], bytes: sizeOf(halves) + sizeOf(batch.requests) };
286 }
287
288 const live: LiveItem[] = [];
289 for (const h of halves) this.putHalf(h);
290 for (const r of batch.requests) this.putHalf({ ...r, half: "app" });
291 for (const log of batch.logs) {
292 this.sql.exec(
293 `INSERT OR IGNORE INTO logs (id, time_us, severity_number, body, trace_id, span_id, request_id, attributes, service, environment, deployment_id, commit_sha, source)
294 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
295 log.id, log.time_us, log.severity_number, log.body, log.trace_id, log.span_id, log.request_id, JSON.stringify(log.attributes),
296 log.service, log.environment, log.deployment_id, log.commit_sha, log.source,
297 );
298 live.push({ type: "log", log });
299 }
300 for (const s of batch.spans) {
301 this.sql.exec(
302 `INSERT OR REPLACE INTO spans (trace_id, span_id, parent_span_id, name, kind, start_us, duration_us, status, status_message, attributes, resource, events, links,
303 service, environment, deployment_id, commit_sha, source)
304 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
305 s.trace_id, s.span_id, s.parent_span_id, s.name, s.kind, s.start_us, s.duration_us, s.status, s.status_message,
306 JSON.stringify(s.attributes), JSON.stringify(s.resource), JSON.stringify(s.events), JSON.stringify(s.links),
307 s.service, s.environment, s.deployment_id, s.commit_sha, s.source,
308 );
309 live.push({ type: "span", span: s });
310 }
311 for (const m of batch.metrics) {
312 this.sql.exec(
313 `INSERT INTO metric_points (name, kind, time_us, value, count, sum, min, max, bounds, counts, quantiles, monotonic, attributes, service, environment, deployment_id, source)
314 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
315 m.name, m.kind, m.time_us, m.value, m.count, m.sum, m.min, m.max,
316 m.bounds ? JSON.stringify(m.bounds) : null, m.counts ? JSON.stringify(m.counts) : null, m.quantiles ? JSON.stringify(m.quantiles) : null,
317 m.monotonic ? 1 : 0, JSON.stringify(m.attributes), m.service, m.environment, m.deployment_id, m.source,
318 );
319 const known = this.rows("SELECT services FROM metric_names WHERE name = ?", m.name)[0];
320 const services = new Set<string>(json<string[]>(known?.services ?? null, []));
321 services.add(m.service);
322 this.sql.exec(
323 `INSERT INTO metric_names (name, kind, unit, description, services, last_us) VALUES (?, ?, ?, ?, ?, ?)
324 ON CONFLICT (name) DO UPDATE SET kind = excluded.kind, unit = excluded.unit,
325 description = CASE WHEN excluded.description = '' THEN metric_names.description ELSE excluded.description END,
326 services = excluded.services, last_us = MAX(metric_names.last_us, excluded.last_us)`,
327 m.name, m.kind, m.unit, m.description, JSON.stringify([...services].slice(0, 50)), m.time_us,
328 );
329 }
330 for (const e of batch.errors) await this.seeError(e);
331
332 const requestCount = halves.filter((h) => h.half === "edge").length + batch.requests.length + halves.filter((h) => h.half === "app" && h.trigger !== "fetch").length;
333 this.sql.exec(
334 `INSERT INTO usage (month, ingested_bytes, requests, logs, spans, metric_points) VALUES (?, ?, ?, ?, ?, ?)
335 ON CONFLICT (month) DO UPDATE SET ingested_bytes = ingested_bytes + excluded.ingested_bytes, requests = requests + excluded.requests,
336 logs = logs + excluded.logs, spans = spans + excluded.spans, metric_points = metric_points + excluded.metric_points`,
337 month(), batch.bytes, requestCount, batch.logs.length, batch.spans.length, batch.metrics.length,
338 );
339 this.sql.exec(
340 "INSERT INTO usage_days (day, ingested_bytes) VALUES (?, ?) ON CONFLICT (day) DO UPDATE SET ingested_bytes = ingested_bytes + excluded.ingested_bytes",
341 day(), batch.bytes,
342 );
343
344 // Requests go live (and to exports) whole: once both halves are in, or
345 // for OTLP, at once.
346 const ids = [...new Set([...halves.map((h) => h.id), ...batch.requests.map((r) => r.id)])];
347 const requests = ids.length ? this.requestsByIds(ids) : [];
348 for (const r of requests) live.unshift({ type: "request", request: r });
349 this.broadcast(live);
350 // Requests are sent on when they are counted (`roll`), once both halves are in.
351 this.state.waitUntil(this.forward({ ...batch, requests: [] }, []));
352 await this.ensureAlarm();
353 return { accepted: true };
354 }
355
356 /** One half of a request (./tail.ts), merged with the other. */
357 putHalf(h: RequestHalf) {
358 const edge = h.half === "edge";
359 this.sql.exec(
360 `INSERT INTO requests (id, time_us, duration_us, trigger, method, url, route, status, outcome, cpu_ms, region, country, trace_id, log_count,
361 error_name, error_message, service, environment, deployment_id, commit_sha, source, edge, app)
362 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
363 ON CONFLICT (id) DO UPDATE SET
364 time_us = MIN(requests.time_us, excluded.time_us),
365 -- The dispatcher's half has the whole request: its time and status win.
366 duration_us = CASE WHEN excluded.edge = 1 OR requests.edge = 0 THEN excluded.duration_us ELSE requests.duration_us END,
367 status = CASE WHEN excluded.edge = 1 OR requests.status IS NULL THEN COALESCE(excluded.status, requests.status) ELSE requests.status END,
368 trigger = COALESCE(excluded.trigger, requests.trigger),
369 method = COALESCE(requests.method, excluded.method),
370 url = COALESCE(requests.url, excluded.url),
371 route = COALESCE(requests.route, excluded.route),
372 region = COALESCE(requests.region, excluded.region),
373 country = COALESCE(requests.country, excluded.country),
374 trace_id = COALESCE(requests.trace_id, excluded.trace_id),
375 outcome = CASE WHEN excluded.app = 1 THEN excluded.outcome ELSE requests.outcome END,
376 cpu_ms = COALESCE(excluded.cpu_ms, requests.cpu_ms),
377 log_count = requests.log_count + excluded.log_count,
378 error_name = COALESCE(requests.error_name, excluded.error_name),
379 error_message = COALESCE(requests.error_message, excluded.error_message),
380 service = CASE WHEN excluded.service != '' THEN excluded.service ELSE requests.service END,
381 environment = COALESCE(excluded.environment, requests.environment),
382 deployment_id = COALESCE(excluded.deployment_id, requests.deployment_id),
383 commit_sha = COALESCE(excluded.commit_sha, requests.commit_sha),
384 edge = MAX(requests.edge, excluded.edge),
385 app = MAX(requests.app, excluded.app)`,
386 h.id, h.time_us ?? Date.now() * 1000, h.duration_us ?? 0, h.trigger ?? "fetch", h.method ?? null, h.url ?? null, h.route ?? null,
387 h.status ?? null, h.outcome ?? "ok", h.cpu_ms ?? null, h.region ?? null, h.country ?? null, h.trace_id ?? null, h.log_count ?? 0,
388 h.error?.name ?? null, h.error?.message ?? null, h.service ?? "", h.environment ?? null, h.deployment_id ?? null, h.commit_sha ?? null,
389 h.source ?? "cloudflare", edge ? 1 : 0, edge ? 0 : 1,
390 );
391 }
392
393 async seeError(e: Batch["errors"][number]) {
394 const fp = await fingerprint(e.name, e.message, e.stack, e.service);
395 const known = this.rows("SELECT state FROM error_groups WHERE fingerprint = ?", fp)[0];
396 if (!known) {
397 this.sql.exec(
398 `INSERT INTO error_groups (fingerprint, name, message, stack, service, environment, first_us, last_us, count, first_deployment_id, last_request_id, last_trace_id)
399 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)`,
400 fp, e.name.slice(0, 200), e.message.slice(0, 4096), e.stack?.slice(0, 16384) ?? null, e.service, e.environment, e.time_us, e.time_us,
401 e.deployment_id, e.request_id, e.trace_id,
402 );
403 return;
404 }
405 // A resolved error that happens again is open again; an ignored one stays ignored.
406 this.sql.exec(
407 `UPDATE error_groups SET count = count + 1, last_us = MAX(last_us, ?), last_request_id = COALESCE(?, last_request_id),
408 last_trace_id = COALESCE(?, last_trace_id), stack = COALESCE(stack, ?), state = CASE WHEN state = 'resolved' THEN 'open' ELSE state END
409 WHERE fingerprint = ?`,
410 e.time_us, e.request_id, e.trace_id, e.stack ?? null, fp,
411 );
412 }
413
414 // ---- Live ---------------------------------------------------------------
415
416 async fetch(request: Request): Promise<Response> {
417 if (request.headers.get("upgrade") !== "websocket") return new Response("Expected a WebSocket.\n", { status: 426 });
418 const pair = new WebSocketPair();
419 this.state.acceptWebSocket(pair[1]);
420 pair[1].send(JSON.stringify({ type: "hello" }));
421 return new Response(null, { status: 101, webSocket: pair[0] });
422 }
423
424 async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
425 if (message === "ping") ws.send("pong");
426 }
427
428 async webSocketClose(ws: WebSocket, code: number) {
429 try {
430 ws.close(code === 1005 ? 1000 : code, "closing");
431 } catch {}
432 }
433
434 broadcast(items: LiveItem[]) {
435 const sockets = this.state.getWebSockets();
436 if (!sockets.length || !items.length) return;
437 const shown = items.slice(0, LIVE_PER_BATCH);
438 const dropped = items.length - shown.length;
439 const message = JSON.stringify({ type: "items", items: shown, dropped });
440 for (const ws of sockets) {
441 try {
442 ws.send(message);
443 } catch {}
444 }
445 }
446
447 // ---- Exports ------------------------------------------------------------
448
449 exportRows(): (TelemetryExport & { headers: Record<string, string> })[] {
450 return this.rows("SELECT * FROM exports ORDER BY name").map((r) => {
451 const headers = json<Record<string, string>>(r.headers, {});
452 return {
453 id: String(r.id),
454 name: String(r.name),
455 endpoint: String(r.endpoint),
456 signals: json(r.signals, ["traces", "logs", "metrics"]),
457 header_names: Object.keys(headers),
458 headers,
459 enabled: r.enabled === 1,
460 last_error: strOrNull(r.last_error),
461 last_sent_us: numOrNull(r.last_sent_us),
462 };
463 });
464 }
465
466 exports(): TelemetryExport[] {
467 return this.exportRows().map(({ headers: _, ...rest }) => rest);
468 }
469
470 putExport(input: { id?: string | null; name: string; endpoint: string; signals: TelemetryExport["signals"]; headers?: Record<string, string> | null; enabled: boolean }): TelemetryExport {
471 const problem = exportEndpointProblem(input.endpoint);
472 if (problem) throw new Error(problem);
473 const signals = input.signals.filter((s) => s === "traces" || s === "logs" || s === "metrics");
474 if (!signals.length) throw new Error("Choose at least one signal to send.");
475 const id = input.id || randomId(8);
476 const before = this.rows("SELECT headers FROM exports WHERE id = ?", id)[0];
477 // Header values are written, never read back: keep those not given again.
478 const headers = { ...json<Record<string, string>>(before?.headers ?? null, {}) };
479 for (const [k, v] of Object.entries(input.headers ?? {})) {
480 if (v === "") delete headers[k];
481 else headers[k] = v;
482 }
483 this.sql.exec(
484 `INSERT INTO exports (id, name, endpoint, signals, headers, enabled) VALUES (?, ?, ?, ?, ?, ?)
485 ON CONFLICT (id) DO UPDATE SET name = excluded.name, endpoint = excluded.endpoint, signals = excluded.signals,
486 headers = excluded.headers, enabled = excluded.enabled, last_error = NULL`,
487 id, input.name.trim().slice(0, 80) || "Export", input.endpoint.trim(), JSON.stringify(signals), JSON.stringify(headers), input.enabled ? 1 : 0,
488 );
489 return this.exports().find((e) => e.id === id)!;
490 }
491
492 deleteExport(id: string): boolean {
493 return this.sql.exec("DELETE FROM exports WHERE id = ?", id).rowsWritten > 0;
494 }
495
496 async forward(batch: Batch, requests: TelemetryRequest[]) {
497 const targets = this.exportRows().filter((e) => e.enabled);
498 if (!targets.length || (!requests.length && !batch.logs.length && !batch.spans.length && !batch.metrics.length)) return;
499 const bodies = otlpBodies(batch, requests);
500 for (const target of targets) {
501 let error: string | null = null;
502 for (const signal of target.signals) {
503 const body = bodies[signal];
504 if (body) error = (await send(target, signal, body)) ?? error;
505 }
506 this.sql.exec("UPDATE exports SET last_error = ?, last_sent_us = ? WHERE id = ?", error, Date.now() * 1000, target.id);
507 }
508 }
509
510 // ---- Upkeep -------------------------------------------------------------
511
512 async ensureAlarm() {
513 if ((await this.state.storage.getAlarm()) === null) await this.state.storage.setAlarm(Date.now() + 60_000);
514 }
515
516 async alarm() {
517 const rolled = this.roll();
518 if (rolled.length) await this.forward({ requests: [], logs: [], spans: [], metrics: [], errors: [], bytes: 0 }, rolled);
519 const last = Number(this.rows("SELECT value FROM meta WHERE key = 'pruned'")[0]?.value ?? 0);
520 if (Date.now() - last > 3_600_000) {
521 this.prune();
522 this.sql.exec("INSERT INTO meta (key, value) VALUES ('pruned', ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value", String(Date.now()));
523 }
524 // Keep going while there is anything left to count or keep.
525 const pending = this.rows("SELECT 1 FROM requests LIMIT 1").length || this.rows("SELECT 1 FROM logs LIMIT 1").length || this.rows("SELECT 1 FROM spans LIMIT 1").length;
526 if (pending) await this.state.storage.setAlarm(Date.now() + 60_000);
527 }
528
529 /** Counts settled requests into per-minute rollups, and sends them to exports. */
530 roll(now = Date.now() * 1000): TelemetryRequest[] {
531 const rolled: TelemetryRequest[] = [];
532 for (;;) {
533 const rows = this.rows("SELECT * FROM requests WHERE rolled = 0 AND time_us < ? ORDER BY time_us, id LIMIT 2000", now - ROLL_AFTER_US);
534 if (!rows.length) break;
535 for (const a of aggregate(rows)) {
536 const before = this.rows(
537 "SELECT latency, cpu FROM rollups WHERE minute = ? AND service = ? AND environment = ? AND deployment_id = ? AND method = ? AND route = ?",
538 a.minute, a.service, a.environment, a.deployment_id, a.method, a.route,
539 )[0];
540 const latency = encode(before ? addInto(decode(String(before.latency)), decode(a.latency)) : decode(a.latency));
541 const cpu = encode(before ? addInto(decode(String(before.cpu)), decode(a.cpu)) : decode(a.cpu));
542 this.sql.exec(
543 `INSERT INTO rollups (minute, service, environment, deployment_id, commit_sha, source, method, route, requests, errors, s2, s3, s4, s5, latency, cpu)
544 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
545 ON CONFLICT (minute, service, environment, deployment_id, method, route) DO UPDATE SET
546 requests = requests + excluded.requests, errors = errors + excluded.errors, s2 = s2 + excluded.s2, s3 = s3 + excluded.s3,
547 s4 = s4 + excluded.s4, s5 = s5 + excluded.s5, latency = excluded.latency, cpu = excluded.cpu,
548 commit_sha = COALESCE(excluded.commit_sha, rollups.commit_sha)`,
549 a.minute, a.service, a.environment, a.deployment_id, a.commit_sha, a.source, a.method, a.route, a.requests, a.errors,
550 a.s2, a.s3, a.s4, a.s5, latency, cpu,
551 );
552 }
553 const last = rows[rows.length - 1];
554 this.sql.exec("UPDATE requests SET rolled = 1 WHERE rolled = 0 AND (time_us < ? OR (time_us = ? AND id <= ?))", last.time_us, last.time_us, last.id);
555 rolled.push(...rows.map((r) => this.requestOf(r)));
556 if (rows.length < 2000) break;
557 }
558 return rolled;
559 }
560
561 prune(now = Date.now() * 1000) {
562 const keep = now - this.settings().retention_days * DAY_US;
563 // Requests are counted before they go.
564 this.roll(now);
565 this.sql.exec("DELETE FROM requests WHERE time_us < ?", keep);
566 this.sql.exec("DELETE FROM logs WHERE time_us < ?", keep);
567 this.sql.exec("DELETE FROM spans WHERE start_us < ?", keep);
568 this.sql.exec("DELETE FROM metric_points WHERE time_us < ?", keep);
569 this.sql.exec("DELETE FROM rollups WHERE minute < ?", Math.floor((now - ROLLUP_DAYS * DAY_US) / MINUTE_US));
570 this.sql.exec("DELETE FROM error_groups WHERE last_us < ? AND issue_number IS NULL", now - ROLLUP_DAYS * DAY_US);
571 this.sql.exec("DELETE FROM usage_days WHERE day < ?", day(now - 40 * DAY_US));
572 // Over its size, the store sheds a day at a time from the oldest.
573 for (let i = 0; i < 30 && this.state.storage.sql.databaseSize > MAX_BYTES; i++) {
574 const oldest = Number(this.rows("SELECT MIN(time_us) AS t FROM logs")[0]?.t ?? now);
575 const cut = oldest + DAY_US;
576 this.sql.exec("DELETE FROM logs WHERE time_us < ?", cut);
577 this.sql.exec("DELETE FROM spans WHERE start_us < ?", cut);
578 this.sql.exec("DELETE FROM requests WHERE time_us < ?", cut);
579 this.sql.exec("DELETE FROM metric_points WHERE time_us < ?", cut);
580 }
581 }
582
583 async purge(): Promise<boolean> {
584 for (const ws of this.state.getWebSockets()) {
585 try {
586 ws.close(1000, "gone");
587 } catch {}
588 }
589 await this.state.storage.deleteAlarm();
590 await this.state.storage.deleteAll();
591 return true;
592 }
593
594 // ---- Reading ------------------------------------------------------------
595
596 /** WHERE clauses for what every list narrows by. */
597 filter(f: TelemetryFilter, time: string, prefix = ""): { clauses: string[]; args: SqlStorageValue[] } {
598 const clauses: string[] = [];
599 const args: SqlStorageValue[] = [];
600 const col = (c: string) => `${prefix}${c}`;
601 if (f.from_us) clauses.push(`${col(time)} >= ?`), args.push(f.from_us);
602 if (f.to_us) clauses.push(`${col(time)} <= ?`), args.push(f.to_us);
603 if (f.services?.length) clauses.push(`${col("service")} IN (${f.services.map(() => "?").join(",")})`), args.push(...f.services);
604 if (f.environments?.length) clauses.push(`${col("environment")} IN (${f.environments.map(() => "?").join(",")})`), args.push(...f.environments);
605 if (f.deployment_id) clauses.push(`${col("deployment_id")} = ?`), args.push(f.deployment_id);
606 if (f.source) clauses.push(`${col("source")} = ?`), args.push(f.source);
607 return { clauses, args };
608 }
609
610 /** `key=value` pairs: a column the table has, or an attribute. */
611 pairs(pairs: [string, string][], columns: string[], clauses: string[], args: SqlStorageValue[], attributes: string) {
612 for (const [key, value] of pairs) {
613 if (columns.includes(key)) {
614 clauses.push(`CAST(${attributes.includes(".") ? `${attributes.split(".")[0]}.` : ""}${key} AS TEXT) = ?`);
615 args.push(value);
616 } else if (attributes) {
617 clauses.push(`CAST(json_extract(${attributes}, ?) AS TEXT) = ?`);
618 args.push(`$."${key.replace(/"/g, "")}"`, value);
619 }
620 }
621 }
622
623 requestOf(r: Row): TelemetryRequest {
624 return {
625 id: String(r.id),
626 time_us: Number(r.time_us),
627 duration_us: Number(r.duration_us),
628 trigger: String(r.trigger),
629 method: strOrNull(r.method),
630 url: strOrNull(r.url),
631 route: strOrNull(r.route),
632 status: numOrNull(r.status),
633 outcome: String(r.outcome),
634 cpu_ms: numOrNull(r.cpu_ms),
635 region: strOrNull(r.region),
636 country: strOrNull(r.country),
637 trace_id: strOrNull(r.trace_id),
638 log_count: Number(r.log_count ?? 0),
639 error: r.error_name ? { name: String(r.error_name), message: String(r.error_message ?? "") } : null,
640 service: String(r.service ?? ""),
641 environment: strOrNull(r.environment),
642 deployment_id: strOrNull(r.deployment_id),
643 commit_sha: strOrNull(r.commit_sha),
644 source: r.source === "otlp" ? "otlp" : "cloudflare",
645 };
646 }
647
648 requestsByIds(ids: string[]): TelemetryRequest[] {
649 return this.rows(`SELECT * FROM requests WHERE id IN (${ids.map(() => "?").join(",")})`, ...ids).map((r) => this.requestOf(r));
650 }
651
652 logOf(r: Row): TelemetryLog {
653 const n = Number(r.severity_number);
654 return {
655 id: String(r.id),
656 time_us: Number(r.time_us),
657 severity: severityOf(n),
658 severity_number: n,
659 body: String(r.body),
660 trace_id: strOrNull(r.trace_id),
661 span_id: strOrNull(r.span_id),
662 request_id: strOrNull(r.request_id),
663 attributes: json<Attributes>(r.attributes, {}),
664 service: String(r.service ?? ""),
665 environment: strOrNull(r.environment),
666 deployment_id: strOrNull(r.deployment_id),
667 commit_sha: strOrNull(r.commit_sha),
668 source: r.source === "otlp" ? "otlp" : "cloudflare",
669 };
670 }
671
672 spanOf(r: Row): TelemetrySpan {
673 return {
674 trace_id: String(r.trace_id),
675 span_id: String(r.span_id),
676 parent_span_id: strOrNull(r.parent_span_id),
677 name: String(r.name),
678 kind: String(r.kind) as TelemetrySpan["kind"],
679 start_us: Number(r.start_us),
680 duration_us: Number(r.duration_us),
681 status: String(r.status) as TelemetrySpan["status"],
682 status_message: strOrNull(r.status_message),
683 attributes: json(r.attributes, {}),
684 resource: json(r.resource, {}),
685 events: json(r.events, []),
686 links: json(r.links, []),
687 service: String(r.service ?? ""),
688 environment: strOrNull(r.environment),
689 deployment_id: strOrNull(r.deployment_id),
690 commit_sha: strOrNull(r.commit_sha),
691 source: r.source === "otlp" ? "otlp" : "cloudflare",
692 };
693 }
694
695 requests(q: RequestQuery): Page<TelemetryRequest> {
696 const { clauses, args } = this.filter(q, "time_us");
697 const { words, pairs } = parseSearch(q.text);
698 this.pairs(pairs, ["method", "route", "status", "outcome", "region", "country", "trace_id", "service", "environment", "deployment_id", "commit_sha", "trigger"], clauses, args, "");
699 // Software outside g1t.page often reports a route and no address: both are searched.
700 for (const w of words) clauses.push("(url LIKE ? OR route LIKE ? OR error_message LIKE ? OR error_name LIKE ?)"), args.push(`%${w}%`, `%${w}%`, `%${w}%`, `%${w}%`);
701 if (q.status?.length) {
702 const parts: string[] = [];
703 for (const s of q.status) {
704 const cls = /^([1-5])xx$/i.exec(s);
705 if (cls) parts.push("(status >= ? AND status < ?)"), args.push(Number(cls[1]) * 100, Number(cls[1]) * 100 + 100);
706 else if (/^\d{3}$/.test(s)) parts.push("status = ?"), args.push(Number(s));
707 }
708 if (parts.length) clauses.push(`(${parts.join(" OR ")})`);
709 }
710 if (q.errors_only) clauses.push(ERROR_SQL);
711 if (q.route) clauses.push("route = ?"), args.push(q.route);
712 if (q.min_duration_ms) clauses.push("duration_us >= ?"), args.push(q.min_duration_ms * 1000);
713 const after = readCursor(q.cursor);
714 if (after) clauses.push("(time_us < ? OR (time_us = ? AND id < ?))"), args.push(after.time, after.time, after.id);
715 const limit = limitOf(q.limit);
716 const rows = this.rows(
717 `SELECT * FROM requests ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""} ORDER BY time_us DESC, id DESC LIMIT ?`,
718 ...args,
719 limit + 1,
720 );
721 const items = rows.slice(0, limit).map((r) => this.requestOf(r));
722 const last = items[items.length - 1];
723 return { items, next: rows.length > limit && last ? cursorOf(last.time_us, last.id) : null };
724 }
725
726 request(id: string): { request: TelemetryRequest; logs: TelemetryLog[] } | null {
727 const row = this.rows("SELECT * FROM requests WHERE id = ?", id)[0];
728 if (!row) return null;
729 const logs = this.rows("SELECT * FROM logs WHERE request_id = ? ORDER BY time_us, rowid LIMIT 1000", id).map((r) => this.logOf(r));
730 return { request: this.requestOf(row), logs };
731 }
732
733 logs(q: LogQuery): Page<TelemetryLog> {
734 const { clauses, args } = this.filter(q, "time_us", "logs.");
735 const { words, pairs } = parseSearch(q.text);
736 const isLevel = ([k]: [string, string]) => k === "level" || k === "severity";
737 this.pairs(pairs.filter((p) => !isLevel(p)), ["trace_id", "span_id", "request_id", "service", "environment", "deployment_id", "commit_sha"], clauses, args, "logs.attributes");
738 const levels = [...(q.severities ?? []), ...pairs.filter(isLevel).map(([, v]) => v.toLowerCase() as LogSeverity)];
739 const ranges = levels.filter((l) => SEVERITY_FLOOR[l]).map((l) => SEVERITY_FLOOR[l]);
740 if (ranges.length) {
741 clauses.push(`(${ranges.map(() => "logs.severity_number BETWEEN ? AND ?").join(" OR ")})`);
742 for (const [lo, hi] of ranges) args.push(lo, hi);
743 }
744 if (q.trace_id) clauses.push("logs.trace_id = ?"), args.push(q.trace_id);
745 if (q.request_id) clauses.push("logs.request_id = ?"), args.push(q.request_id);
746 let from = "logs";
747 if (words.length && this.fts) {
748 from = "logs JOIN logs_fts ON logs_fts.rowid = logs.rowid";
749 clauses.push("logs_fts MATCH ?");
750 args.push(ftsQuery(words));
751 } else {
752 for (const w of words) clauses.push("logs.body LIKE ?"), args.push(`%${w}%`);
753 }
754 const after = readCursor(q.cursor);
755 if (after) clauses.push("(logs.time_us < ? OR (logs.time_us = ? AND logs.id < ?))"), args.push(after.time, after.time, after.id);
756 const limit = limitOf(q.limit, 200, 1000);
757 const rows = this.rows(
758 `SELECT logs.* FROM ${from} ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""} ORDER BY logs.time_us DESC, logs.id DESC LIMIT ?`,
759 ...args,
760 limit + 1,
761 );
762 const items = rows.slice(0, limit).map((r) => this.logOf(r));
763 const last = items[items.length - 1];
764 return { items, next: rows.length > limit && last ? cursorOf(last.time_us, last.id) : null };
765 }
766
767 traces(q: TraceQuery): Page<TraceSummary> {
768 const { clauses, args } = this.filter(q, "start_us");
769 const { words, pairs } = parseSearch(q.text);
770 this.pairs(pairs, ["name", "service", "environment", "deployment_id", "kind", "status"], clauses, args, "attributes");
771 for (const w of words) clauses.push("name LIKE ?"), args.push(`%${w}%`);
772 const after = readCursor(q.cursor);
773 const having: string[] = [];
774 const havingArgs: SqlStorageValue[] = [];
775 if (q.errors_only) having.push("SUM(status = 'error') > 0");
776 if (q.min_duration_ms) having.push("MAX(start_us + duration_us) - MIN(start_us) >= ?"), havingArgs.push(q.min_duration_ms * 1000);
777 if (after) having.push("(MIN(start_us) < ? OR (MIN(start_us) = ? AND trace_id < ?))"), havingArgs.push(after.time, after.time, after.id);
778 const limit = limitOf(q.limit, 50, 200);
779 const rows = this.rows(
780 `SELECT trace_id, MIN(start_us) AS start_us, MAX(start_us + duration_us) - MIN(start_us) AS duration_us, COUNT(*) AS span_count,
781 SUM(status = 'error') AS error_count, GROUP_CONCAT(DISTINCT service) AS services
782 FROM spans ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""}
783 GROUP BY trace_id ${having.length ? `HAVING ${having.join(" AND ")}` : ""}
784 ORDER BY start_us DESC, trace_id DESC LIMIT ?`,
785 ...args,
786 ...havingArgs,
787 limit + 1,
788 );
789 const items: TraceSummary[] = rows.slice(0, limit).map((r) => {
790 const traceId = String(r.trace_id);
791 // The trace's request on g1t.page (its true root), else its earliest span.
792 const request = this.rows("SELECT * FROM requests WHERE trace_id = ? ORDER BY time_us LIMIT 1", traceId)[0];
793 const first = this.rows("SELECT * FROM spans WHERE trace_id = ? ORDER BY parent_span_id IS NOT NULL, start_us LIMIT 1", traceId)[0];
794 const req = request ? this.requestOf(request) : null;
795 const start = Math.min(Number(r.start_us), req?.time_us ?? Infinity);
796 const end = Math.max(Number(r.start_us) + Number(r.duration_us), req ? req.time_us + req.duration_us : 0);
797 return {
798 trace_id: traceId,
799 name: req ? [req.method, req.route].filter(Boolean).join(" ") : String(first?.name ?? ""),
800 start_us: start,
801 duration_us: end - start,
802 span_count: Number(r.span_count) + (req ? 1 : 0),
803 error_count: Number(r.error_count) + (req?.error ? 1 : 0),
804 services: [...new Set([...(req ? [req.service] : []), ...String(r.services ?? "").split(",").filter(Boolean)])],
805 status: req?.status ?? numOrNull(json<Attributes>(first?.attributes ?? null, {})["http.response.status_code"] as SqlStorageValue),
806 service: req?.service ?? String(first?.service ?? ""),
807 environment: req?.environment ?? strOrNull(first?.environment ?? null),
808 deployment_id: req?.deployment_id ?? strOrNull(first?.deployment_id ?? null),
809 commit_sha: req?.commit_sha ?? strOrNull(first?.commit_sha ?? null),
810 source: req?.source ?? (first?.source === "cloudflare" ? "cloudflare" : "otlp"),
811 };
812 });
813 const last = rows[Math.min(rows.length, limit) - 1];
814 return { items, next: rows.length > limit && last ? cursorOf(Number(last.start_us), String(last.trace_id)) : null };
815 }
816
817 /**
818 * A trace: its spans, with each request on g1t.page in it as the server
819 * span it was (the parent the app's own spans name), and every log
820 * written under it.
821 */
822 trace(traceId: string): TraceDetail | null {
823 const spans = this.rows("SELECT * FROM spans WHERE trace_id = ? ORDER BY start_us LIMIT 5000", traceId).map((r) => this.spanOf(r));
824 const requests = this.rows("SELECT * FROM requests WHERE trace_id = ? ORDER BY time_us LIMIT 100", traceId).map((r) => this.requestOf(r));
825 if (!spans.length && !requests.length) return null;
826 const have = new Set(spans.map((s) => s.span_id));
827 for (const r of requests) {
828 if (have.has(r.id)) continue;
829 spans.unshift({
830 trace_id: traceId,
831 span_id: r.id,
832 parent_span_id: null,
833 name: [r.method, r.route].filter(Boolean).join(" ") || r.trigger,
834 kind: "server",
835 start_us: r.time_us,
836 duration_us: r.duration_us,
837 status: r.error || (r.status ?? 0) >= 500 ? "error" : "unset",
838 status_message: r.error?.message ?? null,
839 attributes: {
840 ...(r.method ? { "http.request.method": r.method } : {}),
841 ...(r.url ? { "url.full": `https://${r.url}` } : {}),
842 ...(r.status !== null ? { "http.response.status_code": r.status } : {}),
843 ...(r.cpu_ms !== null ? { "cloudflare.cpu_time_ms": r.cpu_ms } : {}),
844 ...(r.region ? { "cloudflare.colo": r.region } : {}),
845 "cloudflare.outcome": r.outcome,
846 "g1t.request.id": r.id,
847 },
848 resource: { "service.name": r.service },
849 events: [],
850 links: [],
851 service: r.service,
852 environment: r.environment,
853 deployment_id: r.deployment_id,
854 commit_sha: r.commit_sha,
855 source: r.source,
856 });
857 }
858 const ids = requests.map((r) => r.id);
859 const logs = this.rows(
860 `SELECT * FROM logs WHERE trace_id = ? ${ids.length ? `OR request_id IN (${ids.map(() => "?").join(",")})` : ""} ORDER BY time_us LIMIT 2000`,
861 traceId,
862 ...ids,
863 ).map((r) => this.logOf(r));
864 return { trace_id: traceId, spans, logs };
865 }
866
867 errors(f: TelemetryFilter & { state?: ErrorGroup["state"] | null }): ErrorGroup[] {
868 const { clauses, args } = this.filter({ ...f, deployment_id: null, source: null }, "last_us");
869 if (f.state) clauses.push("state = ?"), args.push(f.state);
870 return this.rows(`SELECT * FROM error_groups ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""} ORDER BY last_us DESC LIMIT 200`, ...args).map(errorOf);
871 }
872
873 setError(fp: string, change: { state?: ErrorGroup["state"]; issue_number?: number | null }): ErrorGroup | null {
874 if (change.state && !["open", "resolved", "ignored"].includes(change.state)) throw new Error("State is open, resolved or ignored.");
875 if (change.state) this.sql.exec("UPDATE error_groups SET state = ? WHERE fingerprint = ?", change.state, fp);
876 if (change.issue_number !== undefined) this.sql.exec("UPDATE error_groups SET issue_number = ? WHERE fingerprint = ?", change.issue_number, fp);
877 const row = this.rows("SELECT * FROM error_groups WHERE fingerprint = ?", fp)[0];
878 return row ? errorOf(row) : null;
879 }
880
881 metrics(): MetricInfo[] {
882 return this.rows("SELECT * FROM metric_names ORDER BY name").map((r) => ({
883 name: String(r.name),
884 kind: String(r.kind) as MetricInfo["kind"],
885 unit: String(r.unit),
886 description: String(r.description),
887 services: json(r.services, []),
888 last_us: Number(r.last_us),
889 }));
890 }
891
892 /**
893 * A metric over time. Gauges average (or take the min or max) in each
894 * step; sums add (a cumulative sum becomes the increase in the step);
895 * histograms give their mean, or a percentile from their buckets.
896 */
897 metric(q: TelemetryFilter & { name: string; aggregate?: string | null; group_by?: string | null; step_s?: number | null }): MetricSeries {
898 const now = Date.now() * 1000;
899 const from = q.from_us ?? now - DAY_US;
900 const to = q.to_us ?? now;
901 const step = q.step_s ?? stepFor(to - from);
902 const { clauses, args } = this.filter({ ...q, from_us: from, to_us: to }, "time_us");
903 clauses.push("name = ?");
904 args.push(q.name);
905 const info = this.rows("SELECT unit, kind FROM metric_names WHERE name = ?", q.name)[0];
906 const rows = this.rows(`SELECT * FROM metric_points WHERE ${clauses.join(" AND ")} ORDER BY time_us LIMIT 200000`, ...args);
907 const stepUs = step * 1_000_000;
908 const groups = new Map<string | null, Map<number, number[]>>();
909 const lastCumulative = new Map<string, number>();
910 const aggregate = q.aggregate ?? (info?.kind === "sum" ? "sum" : "avg");
911 for (const r of rows) {
912 const attrs = json<Attributes>(r.attributes, {});
913 const group = q.group_by ? String(q.group_by === "service" ? r.service : (attrs[q.group_by] ?? "")) : null;
914 const bucket = Math.floor(Number(r.time_us) / stepUs) * stepUs;
915 let value: number | null = numOrNull(r.value);
916 if (r.kind === "sum" && r.monotonic === 1) {
917 // Each series of a cumulative sum is told apart by its attributes.
918 const key = `${r.service}\u0000${String(r.attributes)}`;
919 const before = lastCumulative.get(key);
920 lastCumulative.set(key, value ?? 0);
921 value = before === undefined || (value ?? 0) < before ? null : (value ?? 0) - before;
922 }
923 if (r.kind === "histogram" && /^p\d+$/.test(aggregate) && r.counts && r.bounds) {
924 value = histogramQuantile(json(r.bounds, []), json(r.counts, []), Number(aggregate.slice(1)) / 100);
925 }
926 if (value === null) continue;
927 let g = groups.get(group);
928 if (!g) groups.set(group, (g = new Map()));
929 g.set(bucket, [...(g.get(bucket) ?? []), value]);
930 }
931 const reduce = (values: number[]): number => {
932 switch (aggregate) {
933 case "sum":
934 return values.reduce((a, b) => a + b, 0);
935 case "min":
936 return Math.min(...values);
937 case "max":
938 return Math.max(...values);
939 default:
940 return values.reduce((a, b) => a + b, 0) / values.length;
941 }
942 };
943 const start = Math.floor(from / stepUs) * stepUs;
944 return {
945 name: q.name,
946 unit: String(info?.unit ?? ""),
947 step_s: step,
948 series: [...groups.entries()].slice(0, 20).map(([group, points]) => {
949 const out: { time_us: number; value: number | null }[] = [];
950 for (let t = start; t <= to; t += stepUs) {
951 const values = points.get(t);
952 out.push({ time_us: t, value: values ? reduce(values) : null });
953 }
954 return { group, points: out };
955 }),
956 };
957 }
958
959 facets(): { services: string[]; environments: string[]; attribute_keys: string[] } {
960 const since = Date.now() * 1000 - 7 * DAY_US;
961 const services = this.rows(
962 `SELECT service FROM rollups WHERE minute >= ? AND service != '' GROUP BY service
963 UNION SELECT service FROM spans WHERE start_us >= ? AND service != '' GROUP BY service
964 UNION SELECT service FROM logs WHERE time_us >= ? AND service != '' GROUP BY service`,
965 Math.floor(since / MINUTE_US), since, since,
966 ).map((r) => String(r.service));
967 const environments = this.rows(
968 `SELECT environment FROM rollups WHERE minute >= ? AND environment != '' GROUP BY environment
969 UNION SELECT environment FROM logs WHERE time_us >= ? AND environment IS NOT NULL GROUP BY environment`,
970 Math.floor(since / MINUTE_US), since,
971 ).map((r) => String(r.environment));
972 const keys = new Set<string>();
973 for (const r of this.rows("SELECT attributes FROM logs ORDER BY time_us DESC LIMIT 500")) for (const k of Object.keys(json(r.attributes, {}))) keys.add(k);
974 for (const r of this.rows("SELECT attributes FROM spans ORDER BY start_us DESC LIMIT 500")) for (const k of Object.keys(json(r.attributes, {}))) keys.add(k);
975 return { services: [...new Set(services)].sort(), environments: [...new Set(environments)].sort(), attribute_keys: [...keys].sort().slice(0, 200) };
976 }
977
978 usage(): TelemetryUsage {
979 const m = month();
980 const row = this.rows("SELECT * FROM usage WHERE month = ?", m)[0];
981 return {
982 month: m,
983 ingested_bytes: Number(row?.ingested_bytes ?? 0),
984 requests: Number(row?.requests ?? 0),
985 logs: Number(row?.logs ?? 0),
986 spans: Number(row?.spans ?? 0),
987 metric_points: Number(row?.metric_points ?? 0),
988 stored_bytes: this.state.storage.sql.databaseSize,
989 today_bytes: this.todayBytes(),
990 };
991 }
992
993 overview(f: TelemetryFilter & { step_s?: number | null }): TelemetryOverview {
994 const now = Date.now() * 1000;
995 const to = f.to_us ?? now;
996 const from = f.from_us ?? to - DAY_US;
997 const step = f.step_s && STEPS.includes(f.step_s) ? f.step_s : stepFor(to - from);
998 const read = (lo: number, hi: number, detail: boolean) => {
999 const { clauses, args } = this.filter({ ...f, from_us: null, to_us: null }, "minute");
1000 clauses.push("minute >= ?", "minute <= ?", detail ? "route != '*'" : "route = '*'");
1001 args.push(Math.floor(lo / MINUTE_US), Math.floor(hi / MINUTE_US));
1002 const counted = this.rows(`SELECT * FROM rollups WHERE ${clauses.join(" AND ")}`, ...args);
1003 // The newest requests, not counted yet, are counted here as they stand.
1004 const { clauses: pc, args: pa } = this.filter({ ...f, from_us: lo, to_us: hi }, "time_us");
1005 pc.push("rolled = 0");
1006 const pending = aggregate(this.rows(`SELECT * FROM requests WHERE ${pc.join(" AND ")} LIMIT 20000`, ...pa))
1007 .filter((a) => (detail ? a.route !== "*" : a.route === "*")) as unknown as Row[];
1008 return [...counted, ...pending];
1009 };
1010 const totalsOf = (rows: Row[]): TelemetryTotals => {
1011 let requests = 0;
1012 let errors = 0;
1013 const latency = emptyHistogram();
1014 const cpu = emptyHistogram();
1015 for (const r of rows) {
1016 requests += Number(r.requests);
1017 errors += Number(r.errors);
1018 addInto(latency, decode(String(r.latency)));
1019 addInto(cpu, decode(String(r.cpu)));
1020 }
1021 return { requests, errors, error_rate: requests ? errors / requests : 0, latency: percentiles(latency), cpu: percentiles(cpu) };
1022 };
1023
1024 const rows = read(from, to, false);
1025 const previous = read(from - (to - from), from, false);
1026 const stepMinutes = step / 60;
1027 const points = new Map<number, { rows: Row[] }>();
1028 for (const r of rows) {
1029 const bucket = Math.floor(Number(r.minute) / stepMinutes) * stepMinutes;
1030 const p = points.get(bucket) ?? { rows: [] };
1031 p.rows.push(r);
1032 points.set(bucket, p);
1033 }
1034 const series = [];
1035 for (let m = Math.floor(from / MINUTE_US / stepMinutes) * stepMinutes; m * MINUTE_US <= to; m += stepMinutes) {
1036 const p = points.get(m)?.rows ?? [];
1037 const t = totalsOf(p);
1038 const s = (k: string) => p.reduce((n, r) => n + Number(r[k]), 0);
1039 series.push({ time_us: m * MINUTE_US, requests: t.requests, errors: t.errors, status_2xx: s("s2"), status_3xx: s("s3"), status_4xx: s("s4"), status_5xx: s("s5"), latency: t.latency, cpu: t.cpu });
1040 }
1041
1042 const routes = new Map<string, Row[]>();
1043 for (const r of read(from, to, true)) {
1044 const key = `${r.method}\u0000${r.route}`;
1045 routes.set(key, [...(routes.get(key) ?? []), r]);
1046 }
1047 const deployments = new Map<string, Row[]>();
1048 const services = new Map<string, Row[]>();
1049 for (const r of rows) {
1050 if (r.deployment_id) deployments.set(String(r.deployment_id), [...(deployments.get(String(r.deployment_id)) ?? []), r]);
1051 const sk = `${r.service}\u0000${r.source}`;
1052 services.set(sk, [...(services.get(sk) ?? []), r]);
1053 }
1054 const { clauses: rc, args: ra } = this.filter({ ...f, from_us: from, to_us: to }, "time_us");
1055 const regions = this.rows(
1056 `SELECT region, COUNT(*) AS n FROM requests WHERE region IS NOT NULL ${rc.length ? `AND ${rc.join(" AND ")}` : ""} GROUP BY region ORDER BY n DESC LIMIT 20`,
1057 ...ra,
1058 );
1059
1060 return {
1061 from_us: from,
1062 to_us: to,
1063 step_s: step,
1064 totals: totalsOf(rows),
1065 previous: totalsOf(previous),
1066 points: series,
1067 routes: [...routes.values()]
1068 .map((list) => {
1069 const t = totalsOf(list);
1070 return { route: String(list[0].route), method: strOrNull(list[0].method), requests: t.requests, errors: t.errors, latency: t.latency };
1071 })
1072 .sort((a, b) => b.requests - a.requests)
1073 .slice(0, 50),
1074 deployments: [...deployments.entries()]
1075 .map(([id, list]) => {
1076 const t = totalsOf(list);
1077 return {
1078 deployment_id: id,
1079 environment: strOrNull(list[0].environment),
1080 commit_sha: strOrNull(list.find((r) => r.commit_sha)?.commit_sha ?? null),
1081 first_us: Math.min(...list.map((r) => Number(r.minute))) * MINUTE_US,
1082 requests: t.requests,
1083 errors: t.errors,
1084 latency: t.latency,
1085 };
1086 })
1087 .sort((a, b) => b.first_us - a.first_us)
1088 .slice(0, 20),
1089 services: [...services.values()]
1090 .map((list) => ({
1091 service: String(list[0].service),
1092 source: list[0].source === "otlp" ? ("otlp" as const) : ("cloudflare" as const),
1093 requests: list.reduce((n, r) => n + Number(r.requests), 0),
1094 errors: list.reduce((n, r) => n + Number(r.errors), 0),
1095 last_us: Math.max(...list.map((r) => Number(r.minute))) * MINUTE_US,
1096 }))
1097 .sort((a, b) => b.requests - a.requests),
1098 regions: regions.map((r) => ({ region: String(r.region), requests: Number(r.n) })),
1099 };
1100 }
1101}
1102
1103/** A request that failed: a 5xx, an outcome other than ok, or an error the app recorded. */
1104const ERROR_SQL = "(status >= 500 OR outcome NOT IN ('ok', 'canceled', 'responseStreamDisconnected') OR error_name IS NOT NULL)";
1105
1106function isError(status: number | null, outcome: string, errorName: SqlStorageValue): boolean {
1107 return (status !== null && status >= 500) || !["ok", "canceled", "responseStreamDisconnected"].includes(outcome) || (errorName !== null && errorName !== undefined && errorName !== "");
1108}
1109
1110function errorOf(r: Row): ErrorGroup {
1111 return {
1112 fingerprint: String(r.fingerprint),
1113 name: String(r.name),
1114 message: String(r.message),
1115 stack: strOrNull(r.stack),
1116 service: String(r.service),
1117 environment: strOrNull(r.environment),
1118 first_us: Number(r.first_us),
1119 last_us: Number(r.last_us),
1120 count: Number(r.count),
1121 first_deployment_id: strOrNull(r.first_deployment_id),
1122 last_request_id: strOrNull(r.last_request_id),
1123 last_trace_id: strOrNull(r.last_trace_id),
1124 state: String(r.state) as ErrorGroup["state"],
1125 issue_number: numOrNull(r.issue_number),
1126 };
1127}
1128
1129/** The step that gives a chart about a hundred points over `rangeUs`. */
1130export function stepFor(rangeUs: number): number {
1131 const seconds = rangeUs / 1_000_000;
1132 return STEPS.find((s) => seconds / s <= 150) ?? STEPS[STEPS.length - 1];
1133}
1134
1135/** A percentile from an OTLP histogram's explicit buckets. */
1136export function histogramQuantile(bounds: number[], counts: number[], q: number): number | null {
1137 const total = counts.reduce((a, b) => a + b, 0);
1138 if (!total) return null;
1139 const rank = q * total;
1140 let seen = 0;
1141 for (let i = 0; i < counts.length; i++) {
1142 if (seen + counts[i] >= rank) {
1143 const low = i === 0 ? 0 : bounds[i - 1];
1144 const high = i < bounds.length ? bounds[i] : bounds[bounds.length - 1] ?? low;
1145 return low + (high - low) * (counts[i] ? (rank - seen) / counts[i] : 0);
1146 }
1147 seen += counts[i];
1148 }
1149 return bounds[bounds.length - 1] ?? null;
1150}
1151
1152/** Requests' counts by minute, service, environment, deployment and route, with all routes together as `*`. */
1153export type Rollup = {
1154 minute: number;
1155 service: string;
1156 environment: string;
1157 deployment_id: string;
1158 commit_sha: string | null;
1159 source: string;
1160 method: string;
1161 route: string;
1162 requests: number;
1163 errors: number;
1164 s2: number;
1165 s3: number;
1166 s4: number;
1167 s5: number;
1168 latency: string;
1169 cpu: string;
1170};
1171
1172export function aggregate(rows: Row[]): Rollup[] {
1173 const acc = new Map<string, Omit<Rollup, "latency" | "cpu"> & { latency: number[]; cpu: number[] }>();
1174 const add = (key: Pick<Rollup, "minute" | "service" | "environment" | "deployment_id" | "commit_sha" | "source" | "method" | "route">, r: Row) => {
1175 const k = [key.minute, key.service, key.environment, key.deployment_id, key.method, key.route].join("\u0000");
1176 let a = acc.get(k);
1177 if (!a) acc.set(k, (a = { ...key, requests: 0, errors: 0, s2: 0, s3: 0, s4: 0, s5: 0, latency: emptyHistogram(), cpu: emptyHistogram() }));
1178 a.requests++;
1179 const status = numOrNull(r.status);
1180 if (isError(status, String(r.outcome), r.error_name)) a.errors++;
1181 if (status !== null && status >= 200 && status < 600) a[`s${Math.floor(status / 100)}` as "s2" | "s3" | "s4" | "s5"]++;
1182 a.latency[bucketOf(Number(r.duration_us) / 1000)]++;
1183 if (r.cpu_ms !== null && r.cpu_ms !== undefined) a.cpu[bucketOf(Number(r.cpu_ms))]++;
1184 };
1185 for (const r of rows) {
1186 const base = {
1187 minute: Math.floor(Number(r.time_us) / MINUTE_US),
1188 service: String(r.service ?? ""),
1189 environment: String(r.environment ?? ""),
1190 deployment_id: String(r.deployment_id ?? ""),
1191 commit_sha: strOrNull(r.commit_sha),
1192 source: String(r.source),
1193 };
1194 add({ ...base, method: "*", route: "*" }, r);
1195 add({ ...base, method: String(r.method ?? ""), route: String(r.route ?? "") }, r);
1196 }
1197 return [...acc.values()].map((a) => ({ ...a, latency: encode(a.latency), cpu: encode(a.cpu) }));
1198}