Skip to content
Sign up

flagon-io/g1t

public
0
1,204 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 { 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 * `overQuota`: the project's workspace is past its day's share of its
277 * free observability (its meter says so, ../meter.ts and ../quota.ts):
278 * OTLP is refused (`quota`), and tail events keep their requests without
279 * logs. Returns the bytes kept, for the workspace's meter.
280 */
281 async ingest(
282 batch: Batch,
283 halves: RequestHalf[] = [],
284 token?: string,
285 overQuota = false,
286 ): Promise<{ accepted: boolean; reason?: string; bytes?: number }> {
287 if (token !== undefined && !(await this.checkToken(token))) return { accepted: false, reason: "token" };
288 if (token === undefined && !this.settings().deployments) return { accepted: false, reason: "off" };
289 if (overQuota) {
290 if (token !== undefined) return { accepted: false, reason: "quota" };
291 batch = { ...batch, logs: [], spans: [], metrics: [], errors: [], bytes: sizeOf(halves) + sizeOf(batch.requests) };
292 }
293
294 const live: LiveItem[] = [];
295 for (const h of halves) this.putHalf(h);
296 for (const r of batch.requests) this.putHalf({ ...r, half: "app" });
297 for (const log of batch.logs) {
298 this.sql.exec(
299 `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)
300 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
301 log.id, log.time_us, log.severity_number, log.body, log.trace_id, log.span_id, log.request_id, JSON.stringify(log.attributes),
302 log.service, log.environment, log.deployment_id, log.commit_sha, log.source,
303 );
304 live.push({ type: "log", log });
305 }
306 for (const s of batch.spans) {
307 this.sql.exec(
308 `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,
309 service, environment, deployment_id, commit_sha, source)
310 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
311 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,
312 JSON.stringify(s.attributes), JSON.stringify(s.resource), JSON.stringify(s.events), JSON.stringify(s.links),
313 s.service, s.environment, s.deployment_id, s.commit_sha, s.source,
314 );
315 live.push({ type: "span", span: s });
316 }
317 for (const m of batch.metrics) {
318 this.sql.exec(
319 `INSERT INTO metric_points (name, kind, time_us, value, count, sum, min, max, bounds, counts, quantiles, monotonic, attributes, service, environment, deployment_id, source)
320 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
321 m.name, m.kind, m.time_us, m.value, m.count, m.sum, m.min, m.max,
322 m.bounds ? JSON.stringify(m.bounds) : null, m.counts ? JSON.stringify(m.counts) : null, m.quantiles ? JSON.stringify(m.quantiles) : null,
323 m.monotonic ? 1 : 0, JSON.stringify(m.attributes), m.service, m.environment, m.deployment_id, m.source,
324 );
325 const known = this.rows("SELECT services FROM metric_names WHERE name = ?", m.name)[0];
326 const services = new Set<string>(json<string[]>(known?.services ?? null, []));
327 services.add(m.service);
328 this.sql.exec(
329 `INSERT INTO metric_names (name, kind, unit, description, services, last_us) VALUES (?, ?, ?, ?, ?, ?)
330 ON CONFLICT (name) DO UPDATE SET kind = excluded.kind, unit = excluded.unit,
331 description = CASE WHEN excluded.description = '' THEN metric_names.description ELSE excluded.description END,
332 services = excluded.services, last_us = MAX(metric_names.last_us, excluded.last_us)`,
333 m.name, m.kind, m.unit, m.description, JSON.stringify([...services].slice(0, 50)), m.time_us,
334 );
335 }
336 for (const e of batch.errors) await this.seeError(e);
337
338 const requestCount = halves.filter((h) => h.half === "edge").length + batch.requests.length + halves.filter((h) => h.half === "app" && h.trigger !== "fetch").length;
339 this.sql.exec(
340 `INSERT INTO usage (month, ingested_bytes, requests, logs, spans, metric_points) VALUES (?, ?, ?, ?, ?, ?)
341 ON CONFLICT (month) DO UPDATE SET ingested_bytes = ingested_bytes + excluded.ingested_bytes, requests = requests + excluded.requests,
342 logs = logs + excluded.logs, spans = spans + excluded.spans, metric_points = metric_points + excluded.metric_points`,
343 month(), batch.bytes, requestCount, batch.logs.length, batch.spans.length, batch.metrics.length,
344 );
345 this.sql.exec(
346 "INSERT INTO usage_days (day, ingested_bytes) VALUES (?, ?) ON CONFLICT (day) DO UPDATE SET ingested_bytes = ingested_bytes + excluded.ingested_bytes",
347 day(), batch.bytes,
348 );
349
350 // Requests go live (and to exports) whole: once both halves are in, or
351 // for OTLP, at once.
352 const ids = [...new Set([...halves.map((h) => h.id), ...batch.requests.map((r) => r.id)])];
353 const requests = ids.length ? this.requestsByIds(ids) : [];
354 for (const r of requests) live.unshift({ type: "request", request: r });
355 this.broadcast(live);
356 // Requests are sent on when they are counted (`roll`), once both halves are in.
357 this.state.waitUntil(this.forward({ ...batch, requests: [] }, []));
358 await this.ensureAlarm();
359 return { accepted: true, bytes: batch.bytes };
360 }
361
362 /** One half of a request (./tail.ts), merged with the other. */
363 putHalf(h: RequestHalf) {
364 const edge = h.half === "edge";
365 this.sql.exec(
366 `INSERT INTO requests (id, time_us, duration_us, trigger, method, url, route, status, outcome, cpu_ms, region, country, trace_id, log_count,
367 error_name, error_message, service, environment, deployment_id, commit_sha, source, edge, app)
368 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
369 ON CONFLICT (id) DO UPDATE SET
370 time_us = MIN(requests.time_us, excluded.time_us),
371 -- The dispatcher's half has the whole request: its time and status win.
372 duration_us = CASE WHEN excluded.edge = 1 OR requests.edge = 0 THEN excluded.duration_us ELSE requests.duration_us END,
373 status = CASE WHEN excluded.edge = 1 OR requests.status IS NULL THEN COALESCE(excluded.status, requests.status) ELSE requests.status END,
374 trigger = COALESCE(excluded.trigger, requests.trigger),
375 method = COALESCE(requests.method, excluded.method),
376 url = COALESCE(requests.url, excluded.url),
377 route = COALESCE(requests.route, excluded.route),
378 region = COALESCE(requests.region, excluded.region),
379 country = COALESCE(requests.country, excluded.country),
380 trace_id = COALESCE(requests.trace_id, excluded.trace_id),
381 outcome = CASE WHEN excluded.app = 1 THEN excluded.outcome ELSE requests.outcome END,
382 cpu_ms = COALESCE(excluded.cpu_ms, requests.cpu_ms),
383 log_count = requests.log_count + excluded.log_count,
384 error_name = COALESCE(requests.error_name, excluded.error_name),
385 error_message = COALESCE(requests.error_message, excluded.error_message),
386 service = CASE WHEN excluded.service != '' THEN excluded.service ELSE requests.service END,
387 environment = COALESCE(excluded.environment, requests.environment),
388 deployment_id = COALESCE(excluded.deployment_id, requests.deployment_id),
389 commit_sha = COALESCE(excluded.commit_sha, requests.commit_sha),
390 edge = MAX(requests.edge, excluded.edge),
391 app = MAX(requests.app, excluded.app)`,
392 h.id, h.time_us ?? Date.now() * 1000, h.duration_us ?? 0, h.trigger ?? "fetch", h.method ?? null, h.url ?? null, h.route ?? null,
393 h.status ?? null, h.outcome ?? "ok", h.cpu_ms ?? null, h.region ?? null, h.country ?? null, h.trace_id ?? null, h.log_count ?? 0,
394 h.error?.name ?? null, h.error?.message ?? null, h.service ?? "", h.environment ?? null, h.deployment_id ?? null, h.commit_sha ?? null,
395 h.source ?? "cloudflare", edge ? 1 : 0, edge ? 0 : 1,
396 );
397 }
398
399 async seeError(e: Batch["errors"][number]) {
400 const fp = await fingerprint(e.name, e.message, e.stack, e.service);
401 const known = this.rows("SELECT state FROM error_groups WHERE fingerprint = ?", fp)[0];
402 if (!known) {
403 this.sql.exec(
404 `INSERT INTO error_groups (fingerprint, name, message, stack, service, environment, first_us, last_us, count, first_deployment_id, last_request_id, last_trace_id)
405 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)`,
406 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,
407 e.deployment_id, e.request_id, e.trace_id,
408 );
409 return;
410 }
411 // A resolved error that happens again is open again; an ignored one stays ignored.
412 this.sql.exec(
413 `UPDATE error_groups SET count = count + 1, last_us = MAX(last_us, ?), last_request_id = COALESCE(?, last_request_id),
414 last_trace_id = COALESCE(?, last_trace_id), stack = COALESCE(stack, ?), state = CASE WHEN state = 'resolved' THEN 'open' ELSE state END
415 WHERE fingerprint = ?`,
416 e.time_us, e.request_id, e.trace_id, e.stack ?? null, fp,
417 );
418 }
419
420 // ---- Live ---------------------------------------------------------------
421
422 async fetch(request: Request): Promise<Response> {
423 if (request.headers.get("upgrade") !== "websocket") return new Response("Expected a WebSocket.\n", { status: 426 });
424 const pair = new WebSocketPair();
425 this.state.acceptWebSocket(pair[1]);
426 pair[1].send(JSON.stringify({ type: "hello" }));
427 return new Response(null, { status: 101, webSocket: pair[0] });
428 }
429
430 async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
431 if (message === "ping") ws.send("pong");
432 }
433
434 async webSocketClose(ws: WebSocket, code: number) {
435 try {
436 ws.close(code === 1005 ? 1000 : code, "closing");
437 } catch {}
438 }
439
440 broadcast(items: LiveItem[]) {
441 const sockets = this.state.getWebSockets();
442 if (!sockets.length || !items.length) return;
443 const shown = items.slice(0, LIVE_PER_BATCH);
444 const dropped = items.length - shown.length;
445 const message = JSON.stringify({ type: "items", items: shown, dropped });
446 for (const ws of sockets) {
447 try {
448 ws.send(message);
449 } catch {}
450 }
451 }
452
453 // ---- Exports ------------------------------------------------------------
454
455 exportRows(): (TelemetryExport & { headers: Record<string, string> })[] {
456 return this.rows("SELECT * FROM exports ORDER BY name").map((r) => {
457 const headers = json<Record<string, string>>(r.headers, {});
458 return {
459 id: String(r.id),
460 name: String(r.name),
461 endpoint: String(r.endpoint),
462 signals: json(r.signals, ["traces", "logs", "metrics"]),
463 header_names: Object.keys(headers),
464 headers,
465 enabled: r.enabled === 1,
466 last_error: strOrNull(r.last_error),
467 last_sent_us: numOrNull(r.last_sent_us),
468 };
469 });
470 }
471
472 exports(): TelemetryExport[] {
473 return this.exportRows().map(({ headers: _, ...rest }) => rest);
474 }
475
476 putExport(input: { id?: string | null; name: string; endpoint: string; signals: TelemetryExport["signals"]; headers?: Record<string, string> | null; enabled: boolean }): TelemetryExport {
477 const problem = exportEndpointProblem(input.endpoint);
478 if (problem) throw new Error(problem);
479 const signals = input.signals.filter((s) => s === "traces" || s === "logs" || s === "metrics");
480 if (!signals.length) throw new Error("Choose at least one signal to send.");
481 const id = input.id || randomId(8);
482 const before = this.rows("SELECT headers FROM exports WHERE id = ?", id)[0];
483 // Header values are written, never read back: keep those not given again.
484 const headers = { ...json<Record<string, string>>(before?.headers ?? null, {}) };
485 for (const [k, v] of Object.entries(input.headers ?? {})) {
486 if (v === "") delete headers[k];
487 else headers[k] = v;
488 }
489 this.sql.exec(
490 `INSERT INTO exports (id, name, endpoint, signals, headers, enabled) VALUES (?, ?, ?, ?, ?, ?)
491 ON CONFLICT (id) DO UPDATE SET name = excluded.name, endpoint = excluded.endpoint, signals = excluded.signals,
492 headers = excluded.headers, enabled = excluded.enabled, last_error = NULL`,
493 id, input.name.trim().slice(0, 80) || "Export", input.endpoint.trim(), JSON.stringify(signals), JSON.stringify(headers), input.enabled ? 1 : 0,
494 );
495 return this.exports().find((e) => e.id === id)!;
496 }
497
498 deleteExport(id: string): boolean {
499 return this.sql.exec("DELETE FROM exports WHERE id = ?", id).rowsWritten > 0;
500 }
501
502 async forward(batch: Batch, requests: TelemetryRequest[]) {
503 const targets = this.exportRows().filter((e) => e.enabled);
504 if (!targets.length || (!requests.length && !batch.logs.length && !batch.spans.length && !batch.metrics.length)) return;
505 const bodies = otlpBodies(batch, requests);
506 for (const target of targets) {
507 let error: string | null = null;
508 for (const signal of target.signals) {
509 const body = bodies[signal];
510 if (body) error = (await send(target, signal, body)) ?? error;
511 }
512 this.sql.exec("UPDATE exports SET last_error = ?, last_sent_us = ? WHERE id = ?", error, Date.now() * 1000, target.id);
513 }
514 }
515
516 // ---- Upkeep -------------------------------------------------------------
517
518 async ensureAlarm() {
519 if ((await this.state.storage.getAlarm()) === null) await this.state.storage.setAlarm(Date.now() + 60_000);
520 }
521
522 async alarm() {
523 const rolled = this.roll();
524 if (rolled.length) await this.forward({ requests: [], logs: [], spans: [], metrics: [], errors: [], bytes: 0 }, rolled);
525 const last = Number(this.rows("SELECT value FROM meta WHERE key = 'pruned'")[0]?.value ?? 0);
526 if (Date.now() - last > 3_600_000) {
527 this.prune();
528 this.sql.exec("INSERT INTO meta (key, value) VALUES ('pruned', ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value", String(Date.now()));
529 }
530 // Keep going while there is anything left to count or keep.
531 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;
532 if (pending) await this.state.storage.setAlarm(Date.now() + 60_000);
533 }
534
535 /** Counts settled requests into per-minute rollups, and sends them to exports. */
536 roll(now = Date.now() * 1000): TelemetryRequest[] {
537 const rolled: TelemetryRequest[] = [];
538 for (;;) {
539 const rows = this.rows("SELECT * FROM requests WHERE rolled = 0 AND time_us < ? ORDER BY time_us, id LIMIT 2000", now - ROLL_AFTER_US);
540 if (!rows.length) break;
541 for (const a of aggregate(rows)) {
542 const before = this.rows(
543 "SELECT latency, cpu FROM rollups WHERE minute = ? AND service = ? AND environment = ? AND deployment_id = ? AND method = ? AND route = ?",
544 a.minute, a.service, a.environment, a.deployment_id, a.method, a.route,
545 )[0];
546 const latency = encode(before ? addInto(decode(String(before.latency)), decode(a.latency)) : decode(a.latency));
547 const cpu = encode(before ? addInto(decode(String(before.cpu)), decode(a.cpu)) : decode(a.cpu));
548 this.sql.exec(
549 `INSERT INTO rollups (minute, service, environment, deployment_id, commit_sha, source, method, route, requests, errors, s2, s3, s4, s5, latency, cpu)
550 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
551 ON CONFLICT (minute, service, environment, deployment_id, method, route) DO UPDATE SET
552 requests = requests + excluded.requests, errors = errors + excluded.errors, s2 = s2 + excluded.s2, s3 = s3 + excluded.s3,
553 s4 = s4 + excluded.s4, s5 = s5 + excluded.s5, latency = excluded.latency, cpu = excluded.cpu,
554 commit_sha = COALESCE(excluded.commit_sha, rollups.commit_sha)`,
555 a.minute, a.service, a.environment, a.deployment_id, a.commit_sha, a.source, a.method, a.route, a.requests, a.errors,
556 a.s2, a.s3, a.s4, a.s5, latency, cpu,
557 );
558 }
559 const last = rows[rows.length - 1];
560 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);
561 rolled.push(...rows.map((r) => this.requestOf(r)));
562 if (rows.length < 2000) break;
563 }
564 return rolled;
565 }
566
567 prune(now = Date.now() * 1000) {
568 const keep = now - this.settings().retention_days * DAY_US;
569 // Requests are counted before they go.
570 this.roll(now);
571 this.sql.exec("DELETE FROM requests WHERE time_us < ?", keep);
572 this.sql.exec("DELETE FROM logs WHERE time_us < ?", keep);
573 this.sql.exec("DELETE FROM spans WHERE start_us < ?", keep);
574 this.sql.exec("DELETE FROM metric_points WHERE time_us < ?", keep);
575 this.sql.exec("DELETE FROM rollups WHERE minute < ?", Math.floor((now - ROLLUP_DAYS * DAY_US) / MINUTE_US));
576 this.sql.exec("DELETE FROM error_groups WHERE last_us < ? AND issue_number IS NULL", now - ROLLUP_DAYS * DAY_US);
577 this.sql.exec("DELETE FROM usage_days WHERE day < ?", day(now - 40 * DAY_US));
578 // Over its size, the store sheds a day at a time from the oldest.
579 for (let i = 0; i < 30 && this.state.storage.sql.databaseSize > MAX_BYTES; i++) {
580 const oldest = Number(this.rows("SELECT MIN(time_us) AS t FROM logs")[0]?.t ?? now);
581 const cut = oldest + DAY_US;
582 this.sql.exec("DELETE FROM logs WHERE time_us < ?", cut);
583 this.sql.exec("DELETE FROM spans WHERE start_us < ?", cut);
584 this.sql.exec("DELETE FROM requests WHERE time_us < ?", cut);
585 this.sql.exec("DELETE FROM metric_points WHERE time_us < ?", cut);
586 }
587 }
588
589 async purge(): Promise<boolean> {
590 for (const ws of this.state.getWebSockets()) {
591 try {
592 ws.close(1000, "gone");
593 } catch {}
594 }
595 await this.state.storage.deleteAlarm();
596 await this.state.storage.deleteAll();
597 return true;
598 }
599
600 // ---- Reading ------------------------------------------------------------
601
602 /** WHERE clauses for what every list narrows by. */
603 filter(f: TelemetryFilter, time: string, prefix = ""): { clauses: string[]; args: SqlStorageValue[] } {
604 const clauses: string[] = [];
605 const args: SqlStorageValue[] = [];
606 const col = (c: string) => `${prefix}${c}`;
607 if (f.from_us) clauses.push(`${col(time)} >= ?`), args.push(f.from_us);
608 if (f.to_us) clauses.push(`${col(time)} <= ?`), args.push(f.to_us);
609 if (f.services?.length) clauses.push(`${col("service")} IN (${f.services.map(() => "?").join(",")})`), args.push(...f.services);
610 if (f.environments?.length) clauses.push(`${col("environment")} IN (${f.environments.map(() => "?").join(",")})`), args.push(...f.environments);
611 if (f.deployment_id) clauses.push(`${col("deployment_id")} = ?`), args.push(f.deployment_id);
612 if (f.source) clauses.push(`${col("source")} = ?`), args.push(f.source);
613 return { clauses, args };
614 }
615
616 /** `key=value` pairs: a column the table has, or an attribute. */
617 pairs(pairs: [string, string][], columns: string[], clauses: string[], args: SqlStorageValue[], attributes: string) {
618 for (const [key, value] of pairs) {
619 if (columns.includes(key)) {
620 clauses.push(`CAST(${attributes.includes(".") ? `${attributes.split(".")[0]}.` : ""}${key} AS TEXT) = ?`);
621 args.push(value);
622 } else if (attributes) {
623 clauses.push(`CAST(json_extract(${attributes}, ?) AS TEXT) = ?`);
624 args.push(`$."${key.replace(/"/g, "")}"`, value);
625 }
626 }
627 }
628
629 requestOf(r: Row): TelemetryRequest {
630 return {
631 id: String(r.id),
632 time_us: Number(r.time_us),
633 duration_us: Number(r.duration_us),
634 trigger: String(r.trigger),
635 method: strOrNull(r.method),
636 url: strOrNull(r.url),
637 route: strOrNull(r.route),
638 status: numOrNull(r.status),
639 outcome: String(r.outcome),
640 cpu_ms: numOrNull(r.cpu_ms),
641 region: strOrNull(r.region),
642 country: strOrNull(r.country),
643 trace_id: strOrNull(r.trace_id),
644 log_count: Number(r.log_count ?? 0),
645 error: r.error_name ? { name: String(r.error_name), message: String(r.error_message ?? "") } : null,
646 service: String(r.service ?? ""),
647 environment: strOrNull(r.environment),
648 deployment_id: strOrNull(r.deployment_id),
649 commit_sha: strOrNull(r.commit_sha),
650 source: r.source === "otlp" ? "otlp" : "cloudflare",
651 };
652 }
653
654 requestsByIds(ids: string[]): TelemetryRequest[] {
655 return this.rows(`SELECT * FROM requests WHERE id IN (${ids.map(() => "?").join(",")})`, ...ids).map((r) => this.requestOf(r));
656 }
657
658 logOf(r: Row): TelemetryLog {
659 const n = Number(r.severity_number);
660 return {
661 id: String(r.id),
662 time_us: Number(r.time_us),
663 severity: severityOf(n),
664 severity_number: n,
665 body: String(r.body),
666 trace_id: strOrNull(r.trace_id),
667 span_id: strOrNull(r.span_id),
668 request_id: strOrNull(r.request_id),
669 attributes: json<Attributes>(r.attributes, {}),
670 service: String(r.service ?? ""),
671 environment: strOrNull(r.environment),
672 deployment_id: strOrNull(r.deployment_id),
673 commit_sha: strOrNull(r.commit_sha),
674 source: r.source === "otlp" ? "otlp" : "cloudflare",
675 };
676 }
677
678 spanOf(r: Row): TelemetrySpan {
679 return {
680 trace_id: String(r.trace_id),
681 span_id: String(r.span_id),
682 parent_span_id: strOrNull(r.parent_span_id),
683 name: String(r.name),
684 kind: String(r.kind) as TelemetrySpan["kind"],
685 start_us: Number(r.start_us),
686 duration_us: Number(r.duration_us),
687 status: String(r.status) as TelemetrySpan["status"],
688 status_message: strOrNull(r.status_message),
689 attributes: json(r.attributes, {}),
690 resource: json(r.resource, {}),
691 events: json(r.events, []),
692 links: json(r.links, []),
693 service: String(r.service ?? ""),
694 environment: strOrNull(r.environment),
695 deployment_id: strOrNull(r.deployment_id),
696 commit_sha: strOrNull(r.commit_sha),
697 source: r.source === "otlp" ? "otlp" : "cloudflare",
698 };
699 }
700
701 requests(q: RequestQuery): Page<TelemetryRequest> {
702 const { clauses, args } = this.filter(q, "time_us");
703 const { words, pairs } = parseSearch(q.text);
704 this.pairs(pairs, ["method", "route", "status", "outcome", "region", "country", "trace_id", "service", "environment", "deployment_id", "commit_sha", "trigger"], clauses, args, "");
705 // Software outside g1t.page often reports a route and no address: both are searched.
706 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}%`);
707 if (q.status?.length) {
708 const parts: string[] = [];
709 for (const s of q.status) {
710 const cls = /^([1-5])xx$/i.exec(s);
711 if (cls) parts.push("(status >= ? AND status < ?)"), args.push(Number(cls[1]) * 100, Number(cls[1]) * 100 + 100);
712 else if (/^\d{3}$/.test(s)) parts.push("status = ?"), args.push(Number(s));
713 }
714 if (parts.length) clauses.push(`(${parts.join(" OR ")})`);
715 }
716 if (q.errors_only) clauses.push(ERROR_SQL);
717 if (q.route) clauses.push("route = ?"), args.push(q.route);
718 if (q.min_duration_ms) clauses.push("duration_us >= ?"), args.push(q.min_duration_ms * 1000);
719 const after = readCursor(q.cursor);
720 if (after) clauses.push("(time_us < ? OR (time_us = ? AND id < ?))"), args.push(after.time, after.time, after.id);
721 const limit = limitOf(q.limit);
722 const rows = this.rows(
723 `SELECT * FROM requests ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""} ORDER BY time_us DESC, id DESC LIMIT ?`,
724 ...args,
725 limit + 1,
726 );
727 const items = rows.slice(0, limit).map((r) => this.requestOf(r));
728 const last = items[items.length - 1];
729 return { items, next: rows.length > limit && last ? cursorOf(last.time_us, last.id) : null };
730 }
731
732 request(id: string): { request: TelemetryRequest; logs: TelemetryLog[] } | null {
733 const row = this.rows("SELECT * FROM requests WHERE id = ?", id)[0];
734 if (!row) return null;
735 const logs = this.rows("SELECT * FROM logs WHERE request_id = ? ORDER BY time_us, rowid LIMIT 1000", id).map((r) => this.logOf(r));
736 return { request: this.requestOf(row), logs };
737 }
738
739 logs(q: LogQuery): Page<TelemetryLog> {
740 const { clauses, args } = this.filter(q, "time_us", "logs.");
741 const { words, pairs } = parseSearch(q.text);
742 const isLevel = ([k]: [string, string]) => k === "level" || k === "severity";
743 this.pairs(pairs.filter((p) => !isLevel(p)), ["trace_id", "span_id", "request_id", "service", "environment", "deployment_id", "commit_sha"], clauses, args, "logs.attributes");
744 const levels = [...(q.severities ?? []), ...pairs.filter(isLevel).map(([, v]) => v.toLowerCase() as LogSeverity)];
745 const ranges = levels.filter((l) => SEVERITY_FLOOR[l]).map((l) => SEVERITY_FLOOR[l]);
746 if (ranges.length) {
747 clauses.push(`(${ranges.map(() => "logs.severity_number BETWEEN ? AND ?").join(" OR ")})`);
748 for (const [lo, hi] of ranges) args.push(lo, hi);
749 }
750 if (q.trace_id) clauses.push("logs.trace_id = ?"), args.push(q.trace_id);
751 if (q.request_id) clauses.push("logs.request_id = ?"), args.push(q.request_id);
752 let from = "logs";
753 if (words.length && this.fts) {
754 from = "logs JOIN logs_fts ON logs_fts.rowid = logs.rowid";
755 clauses.push("logs_fts MATCH ?");
756 args.push(ftsQuery(words));
757 } else {
758 for (const w of words) clauses.push("logs.body LIKE ?"), args.push(`%${w}%`);
759 }
760 const after = readCursor(q.cursor);
761 if (after) clauses.push("(logs.time_us < ? OR (logs.time_us = ? AND logs.id < ?))"), args.push(after.time, after.time, after.id);
762 const limit = limitOf(q.limit, 200, 1000);
763 const rows = this.rows(
764 `SELECT logs.* FROM ${from} ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""} ORDER BY logs.time_us DESC, logs.id DESC LIMIT ?`,
765 ...args,
766 limit + 1,
767 );
768 const items = rows.slice(0, limit).map((r) => this.logOf(r));
769 const last = items[items.length - 1];
770 return { items, next: rows.length > limit && last ? cursorOf(last.time_us, last.id) : null };
771 }
772
773 traces(q: TraceQuery): Page<TraceSummary> {
774 const { clauses, args } = this.filter(q, "start_us");
775 const { words, pairs } = parseSearch(q.text);
776 this.pairs(pairs, ["name", "service", "environment", "deployment_id", "kind", "status"], clauses, args, "attributes");
777 for (const w of words) clauses.push("name LIKE ?"), args.push(`%${w}%`);
778 const after = readCursor(q.cursor);
779 const having: string[] = [];
780 const havingArgs: SqlStorageValue[] = [];
781 if (q.errors_only) having.push("SUM(status = 'error') > 0");
782 if (q.min_duration_ms) having.push("MAX(start_us + duration_us) - MIN(start_us) >= ?"), havingArgs.push(q.min_duration_ms * 1000);
783 if (after) having.push("(MIN(start_us) < ? OR (MIN(start_us) = ? AND trace_id < ?))"), havingArgs.push(after.time, after.time, after.id);
784 const limit = limitOf(q.limit, 50, 200);
785 const rows = this.rows(
786 `SELECT trace_id, MIN(start_us) AS start_us, MAX(start_us + duration_us) - MIN(start_us) AS duration_us, COUNT(*) AS span_count,
787 SUM(status = 'error') AS error_count, GROUP_CONCAT(DISTINCT service) AS services
788 FROM spans ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""}
789 GROUP BY trace_id ${having.length ? `HAVING ${having.join(" AND ")}` : ""}
790 ORDER BY start_us DESC, trace_id DESC LIMIT ?`,
791 ...args,
792 ...havingArgs,
793 limit + 1,
794 );
795 const items: TraceSummary[] = rows.slice(0, limit).map((r) => {
796 const traceId = String(r.trace_id);
797 // The trace's request on g1t.page (its true root), else its earliest span.
798 const request = this.rows("SELECT * FROM requests WHERE trace_id = ? ORDER BY time_us LIMIT 1", traceId)[0];
799 const first = this.rows("SELECT * FROM spans WHERE trace_id = ? ORDER BY parent_span_id IS NOT NULL, start_us LIMIT 1", traceId)[0];
800 const req = request ? this.requestOf(request) : null;
801 const start = Math.min(Number(r.start_us), req?.time_us ?? Infinity);
802 const end = Math.max(Number(r.start_us) + Number(r.duration_us), req ? req.time_us + req.duration_us : 0);
803 return {
804 trace_id: traceId,
805 name: req ? [req.method, req.route].filter(Boolean).join(" ") : String(first?.name ?? ""),
806 start_us: start,
807 duration_us: end - start,
808 span_count: Number(r.span_count) + (req ? 1 : 0),
809 error_count: Number(r.error_count) + (req?.error ? 1 : 0),
810 services: [...new Set([...(req ? [req.service] : []), ...String(r.services ?? "").split(",").filter(Boolean)])],
811 status: req?.status ?? numOrNull(json<Attributes>(first?.attributes ?? null, {})["http.response.status_code"] as SqlStorageValue),
812 service: req?.service ?? String(first?.service ?? ""),
813 environment: req?.environment ?? strOrNull(first?.environment ?? null),
814 deployment_id: req?.deployment_id ?? strOrNull(first?.deployment_id ?? null),
815 commit_sha: req?.commit_sha ?? strOrNull(first?.commit_sha ?? null),
816 source: req?.source ?? (first?.source === "cloudflare" ? "cloudflare" : "otlp"),
817 };
818 });
819 const last = rows[Math.min(rows.length, limit) - 1];
820 return { items, next: rows.length > limit && last ? cursorOf(Number(last.start_us), String(last.trace_id)) : null };
821 }
822
823 /**
824 * A trace: its spans, with each request on g1t.page in it as the server
825 * span it was (the parent the app's own spans name), and every log
826 * written under it.
827 */
828 trace(traceId: string): TraceDetail | null {
829 const spans = this.rows("SELECT * FROM spans WHERE trace_id = ? ORDER BY start_us LIMIT 5000", traceId).map((r) => this.spanOf(r));
830 const requests = this.rows("SELECT * FROM requests WHERE trace_id = ? ORDER BY time_us LIMIT 100", traceId).map((r) => this.requestOf(r));
831 if (!spans.length && !requests.length) return null;
832 const have = new Set(spans.map((s) => s.span_id));
833 for (const r of requests) {
834 if (have.has(r.id)) continue;
835 spans.unshift({
836 trace_id: traceId,
837 span_id: r.id,
838 parent_span_id: null,
839 name: [r.method, r.route].filter(Boolean).join(" ") || r.trigger,
840 kind: "server",
841 start_us: r.time_us,
842 duration_us: r.duration_us,
843 status: r.error || (r.status ?? 0) >= 500 ? "error" : "unset",
844 status_message: r.error?.message ?? null,
845 attributes: {
846 ...(r.method ? { "http.request.method": r.method } : {}),
847 ...(r.url ? { "url.full": `https://${r.url}` } : {}),
848 ...(r.status !== null ? { "http.response.status_code": r.status } : {}),
849 ...(r.cpu_ms !== null ? { "cloudflare.cpu_time_ms": r.cpu_ms } : {}),
850 ...(r.region ? { "cloudflare.colo": r.region } : {}),
851 "cloudflare.outcome": r.outcome,
852 "g1t.request.id": r.id,
853 },
854 resource: { "service.name": r.service },
855 events: [],
856 links: [],
857 service: r.service,
858 environment: r.environment,
859 deployment_id: r.deployment_id,
860 commit_sha: r.commit_sha,
861 source: r.source,
862 });
863 }
864 const ids = requests.map((r) => r.id);
865 const logs = this.rows(
866 `SELECT * FROM logs WHERE trace_id = ? ${ids.length ? `OR request_id IN (${ids.map(() => "?").join(",")})` : ""} ORDER BY time_us LIMIT 2000`,
867 traceId,
868 ...ids,
869 ).map((r) => this.logOf(r));
870 return { trace_id: traceId, spans, logs };
871 }
872
873 errors(f: TelemetryFilter & { state?: ErrorGroup["state"] | null }): ErrorGroup[] {
874 const { clauses, args } = this.filter({ ...f, deployment_id: null, source: null }, "last_us");
875 if (f.state) clauses.push("state = ?"), args.push(f.state);
876 return this.rows(`SELECT * FROM error_groups ${clauses.length ? `WHERE ${clauses.join(" AND ")}` : ""} ORDER BY last_us DESC LIMIT 200`, ...args).map(errorOf);
877 }
878
879 setError(fp: string, change: { state?: ErrorGroup["state"]; issue_number?: number | null }): ErrorGroup | null {
880 if (change.state && !["open", "resolved", "ignored"].includes(change.state)) throw new Error("State is open, resolved or ignored.");
881 if (change.state) this.sql.exec("UPDATE error_groups SET state = ? WHERE fingerprint = ?", change.state, fp);
882 if (change.issue_number !== undefined) this.sql.exec("UPDATE error_groups SET issue_number = ? WHERE fingerprint = ?", change.issue_number, fp);
883 const row = this.rows("SELECT * FROM error_groups WHERE fingerprint = ?", fp)[0];
884 return row ? errorOf(row) : null;
885 }
886
887 metrics(): MetricInfo[] {
888 return this.rows("SELECT * FROM metric_names ORDER BY name").map((r) => ({
889 name: String(r.name),
890 kind: String(r.kind) as MetricInfo["kind"],
891 unit: String(r.unit),
892 description: String(r.description),
893 services: json(r.services, []),
894 last_us: Number(r.last_us),
895 }));
896 }
897
898 /**
899 * A metric over time. Gauges average (or take the min or max) in each
900 * step; sums add (a cumulative sum becomes the increase in the step);
901 * histograms give their mean, or a percentile from their buckets.
902 */
903 metric(q: TelemetryFilter & { name: string; aggregate?: string | null; group_by?: string | null; step_s?: number | null }): MetricSeries {
904 const now = Date.now() * 1000;
905 const from = q.from_us ?? now - DAY_US;
906 const to = q.to_us ?? now;
907 const step = q.step_s ?? stepFor(to - from);
908 const { clauses, args } = this.filter({ ...q, from_us: from, to_us: to }, "time_us");
909 clauses.push("name = ?");
910 args.push(q.name);
911 const info = this.rows("SELECT unit, kind FROM metric_names WHERE name = ?", q.name)[0];
912 const rows = this.rows(`SELECT * FROM metric_points WHERE ${clauses.join(" AND ")} ORDER BY time_us LIMIT 200000`, ...args);
913 const stepUs = step * 1_000_000;
914 const groups = new Map<string | null, Map<number, number[]>>();
915 const lastCumulative = new Map<string, number>();
916 const aggregate = q.aggregate ?? (info?.kind === "sum" ? "sum" : "avg");
917 for (const r of rows) {
918 const attrs = json<Attributes>(r.attributes, {});
919 const group = q.group_by ? String(q.group_by === "service" ? r.service : (attrs[q.group_by] ?? "")) : null;
920 const bucket = Math.floor(Number(r.time_us) / stepUs) * stepUs;
921 let value: number | null = numOrNull(r.value);
922 if (r.kind === "sum" && r.monotonic === 1) {
923 // Each series of a cumulative sum is told apart by its attributes.
924 const key = `${r.service}\u0000${String(r.attributes)}`;
925 const before = lastCumulative.get(key);
926 lastCumulative.set(key, value ?? 0);
927 value = before === undefined || (value ?? 0) < before ? null : (value ?? 0) - before;
928 }
929 if (r.kind === "histogram" && /^p\d+$/.test(aggregate) && r.counts && r.bounds) {
930 value = histogramQuantile(json(r.bounds, []), json(r.counts, []), Number(aggregate.slice(1)) / 100);
931 }
932 if (value === null) continue;
933 let g = groups.get(group);
934 if (!g) groups.set(group, (g = new Map()));
935 g.set(bucket, [...(g.get(bucket) ?? []), value]);
936 }
937 const reduce = (values: number[]): number => {
938 switch (aggregate) {
939 case "sum":
940 return values.reduce((a, b) => a + b, 0);
941 case "min":
942 return Math.min(...values);
943 case "max":
944 return Math.max(...values);
945 default:
946 return values.reduce((a, b) => a + b, 0) / values.length;
947 }
948 };
949 const start = Math.floor(from / stepUs) * stepUs;
950 return {
951 name: q.name,
952 unit: String(info?.unit ?? ""),
953 step_s: step,
954 series: [...groups.entries()].slice(0, 20).map(([group, points]) => {
955 const out: { time_us: number; value: number | null }[] = [];
956 for (let t = start; t <= to; t += stepUs) {
957 const values = points.get(t);
958 out.push({ time_us: t, value: values ? reduce(values) : null });
959 }
960 return { group, points: out };
961 }),
962 };
963 }
964
965 facets(): { services: string[]; environments: string[]; attribute_keys: string[] } {
966 const since = Date.now() * 1000 - 7 * DAY_US;
967 const services = this.rows(
968 `SELECT service FROM rollups WHERE minute >= ? AND service != '' GROUP BY service
969 UNION SELECT service FROM spans WHERE start_us >= ? AND service != '' GROUP BY service
970 UNION SELECT service FROM logs WHERE time_us >= ? AND service != '' GROUP BY service`,
971 Math.floor(since / MINUTE_US), since, since,
972 ).map((r) => String(r.service));
973 const environments = this.rows(
974 `SELECT environment FROM rollups WHERE minute >= ? AND environment != '' GROUP BY environment
975 UNION SELECT environment FROM logs WHERE time_us >= ? AND environment IS NOT NULL GROUP BY environment`,
976 Math.floor(since / MINUTE_US), since,
977 ).map((r) => String(r.environment));
978 const keys = new Set<string>();
979 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);
980 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);
981 return { services: [...new Set(services)].sort(), environments: [...new Set(environments)].sort(), attribute_keys: [...keys].sort().slice(0, 200) };
982 }
983
984 usage(): TelemetryUsage {
985 const m = month();
986 const row = this.rows("SELECT * FROM usage WHERE month = ?", m)[0];
987 return {
988 month: m,
989 ingested_bytes: Number(row?.ingested_bytes ?? 0),
990 requests: Number(row?.requests ?? 0),
991 logs: Number(row?.logs ?? 0),
992 spans: Number(row?.spans ?? 0),
993 metric_points: Number(row?.metric_points ?? 0),
994 stored_bytes: this.state.storage.sql.databaseSize,
995 today_bytes: this.todayBytes(),
996 };
997 }
998
999 overview(f: TelemetryFilter & { step_s?: number | null }): TelemetryOverview {
1000 const now = Date.now() * 1000;
1001 const to = f.to_us ?? now;
1002 const from = f.from_us ?? to - DAY_US;
1003 const step = f.step_s && STEPS.includes(f.step_s) ? f.step_s : stepFor(to - from);
1004 const read = (lo: number, hi: number, detail: boolean) => {
1005 const { clauses, args } = this.filter({ ...f, from_us: null, to_us: null }, "minute");
1006 clauses.push("minute >= ?", "minute <= ?", detail ? "route != '*'" : "route = '*'");
1007 args.push(Math.floor(lo / MINUTE_US), Math.floor(hi / MINUTE_US));
1008 const counted = this.rows(`SELECT * FROM rollups WHERE ${clauses.join(" AND ")}`, ...args);
1009 // The newest requests, not counted yet, are counted here as they stand.
1010 const { clauses: pc, args: pa } = this.filter({ ...f, from_us: lo, to_us: hi }, "time_us");
1011 pc.push("rolled = 0");
1012 const pending = aggregate(this.rows(`SELECT * FROM requests WHERE ${pc.join(" AND ")} LIMIT 20000`, ...pa))
1013 .filter((a) => (detail ? a.route !== "*" : a.route === "*")) as unknown as Row[];
1014 return [...counted, ...pending];
1015 };
1016 const totalsOf = (rows: Row[]): TelemetryTotals => {
1017 let requests = 0;
1018 let errors = 0;
1019 const latency = emptyHistogram();
1020 const cpu = emptyHistogram();
1021 for (const r of rows) {
1022 requests += Number(r.requests);
1023 errors += Number(r.errors);
1024 addInto(latency, decode(String(r.latency)));
1025 addInto(cpu, decode(String(r.cpu)));
1026 }
1027 return { requests, errors, error_rate: requests ? errors / requests : 0, latency: percentiles(latency), cpu: percentiles(cpu) };
1028 };
1029
1030 const rows = read(from, to, false);
1031 const previous = read(from - (to - from), from, false);
1032 const stepMinutes = step / 60;
1033 const points = new Map<number, { rows: Row[] }>();
1034 for (const r of rows) {
1035 const bucket = Math.floor(Number(r.minute) / stepMinutes) * stepMinutes;
1036 const p = points.get(bucket) ?? { rows: [] };
1037 p.rows.push(r);
1038 points.set(bucket, p);
1039 }
1040 const series = [];
1041 for (let m = Math.floor(from / MINUTE_US / stepMinutes) * stepMinutes; m * MINUTE_US <= to; m += stepMinutes) {
1042 const p = points.get(m)?.rows ?? [];
1043 const t = totalsOf(p);
1044 const s = (k: string) => p.reduce((n, r) => n + Number(r[k]), 0);
1045 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 });
1046 }
1047
1048 const routes = new Map<string, Row[]>();
1049 for (const r of read(from, to, true)) {
1050 const key = `${r.method}\u0000${r.route}`;
1051 routes.set(key, [...(routes.get(key) ?? []), r]);
1052 }
1053 const deployments = new Map<string, Row[]>();
1054 const services = new Map<string, Row[]>();
1055 for (const r of rows) {
1056 if (r.deployment_id) deployments.set(String(r.deployment_id), [...(deployments.get(String(r.deployment_id)) ?? []), r]);
1057 const sk = `${r.service}\u0000${r.source}`;
1058 services.set(sk, [...(services.get(sk) ?? []), r]);
1059 }
1060 const { clauses: rc, args: ra } = this.filter({ ...f, from_us: from, to_us: to }, "time_us");
1061 const regions = this.rows(
1062 `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`,
1063 ...ra,
1064 );
1065
1066 return {
1067 from_us: from,
1068 to_us: to,
1069 step_s: step,
1070 totals: totalsOf(rows),
1071 previous: totalsOf(previous),
1072 points: series,
1073 routes: [...routes.values()]
1074 .map((list) => {
1075 const t = totalsOf(list);
1076 return { route: String(list[0].route), method: strOrNull(list[0].method), requests: t.requests, errors: t.errors, latency: t.latency };
1077 })
1078 .sort((a, b) => b.requests - a.requests)
1079 .slice(0, 50),
1080 deployments: [...deployments.entries()]
1081 .map(([id, list]) => {
1082 const t = totalsOf(list);
1083 return {
1084 deployment_id: id,
1085 environment: strOrNull(list[0].environment),
1086 commit_sha: strOrNull(list.find((r) => r.commit_sha)?.commit_sha ?? null),
1087 first_us: Math.min(...list.map((r) => Number(r.minute))) * MINUTE_US,
1088 requests: t.requests,
1089 errors: t.errors,
1090 latency: t.latency,
1091 };
1092 })
1093 .sort((a, b) => b.first_us - a.first_us)
1094 .slice(0, 20),
1095 services: [...services.values()]
1096 .map((list) => ({
1097 service: String(list[0].service),
1098 source: list[0].source === "otlp" ? ("otlp" as const) : ("cloudflare" as const),
1099 requests: list.reduce((n, r) => n + Number(r.requests), 0),
1100 errors: list.reduce((n, r) => n + Number(r.errors), 0),
1101 last_us: Math.max(...list.map((r) => Number(r.minute))) * MINUTE_US,
1102 }))
1103 .sort((a, b) => b.requests - a.requests),
1104 regions: regions.map((r) => ({ region: String(r.region), requests: Number(r.n) })),
1105 };
1106 }
1107}
1108
1109/** A request that failed: a 5xx, an outcome other than ok, or an error the app recorded. */
1110const ERROR_SQL = "(status >= 500 OR outcome NOT IN ('ok', 'canceled', 'responseStreamDisconnected') OR error_name IS NOT NULL)";
1111
1112function isError(status: number | null, outcome: string, errorName: SqlStorageValue): boolean {
1113 return (status !== null && status >= 500) || !["ok", "canceled", "responseStreamDisconnected"].includes(outcome) || (errorName !== null && errorName !== undefined && errorName !== "");
1114}
1115
1116function errorOf(r: Row): ErrorGroup {
1117 return {
1118 fingerprint: String(r.fingerprint),
1119 name: String(r.name),
1120 message: String(r.message),
1121 stack: strOrNull(r.stack),
1122 service: String(r.service),
1123 environment: strOrNull(r.environment),
1124 first_us: Number(r.first_us),
1125 last_us: Number(r.last_us),
1126 count: Number(r.count),
1127 first_deployment_id: strOrNull(r.first_deployment_id),
1128 last_request_id: strOrNull(r.last_request_id),
1129 last_trace_id: strOrNull(r.last_trace_id),
1130 state: String(r.state) as ErrorGroup["state"],
1131 issue_number: numOrNull(r.issue_number),
1132 };
1133}
1134
1135/** The step that gives a chart about a hundred points over `rangeUs`. */
1136export function stepFor(rangeUs: number): number {
1137 const seconds = rangeUs / 1_000_000;
1138 return STEPS.find((s) => seconds / s <= 150) ?? STEPS[STEPS.length - 1];
1139}
1140
1141/** A percentile from an OTLP histogram's explicit buckets. */
1142export function histogramQuantile(bounds: number[], counts: number[], q: number): number | null {
1143 const total = counts.reduce((a, b) => a + b, 0);
1144 if (!total) return null;
1145 const rank = q * total;
1146 let seen = 0;
1147 for (let i = 0; i < counts.length; i++) {
1148 if (seen + counts[i] >= rank) {
1149 const low = i === 0 ? 0 : bounds[i - 1];
1150 const high = i < bounds.length ? bounds[i] : bounds[bounds.length - 1] ?? low;
1151 return low + (high - low) * (counts[i] ? (rank - seen) / counts[i] : 0);
1152 }
1153 seen += counts[i];
1154 }
1155 return bounds[bounds.length - 1] ?? null;
1156}
1157
1158/** Requests' counts by minute, service, environment, deployment and route, with all routes together as `*`. */
1159export type Rollup = {
1160 minute: number;
1161 service: string;
1162 environment: string;
1163 deployment_id: string;
1164 commit_sha: string | null;
1165 source: string;
1166 method: string;
1167 route: string;
1168 requests: number;
1169 errors: number;
1170 s2: number;
1171 s3: number;
1172 s4: number;
1173 s5: number;
1174 latency: string;
1175 cpu: string;
1176};
1177
1178export function aggregate(rows: Row[]): Rollup[] {
1179 const acc = new Map<string, Omit<Rollup, "latency" | "cpu"> & { latency: number[]; cpu: number[] }>();
1180 const add = (key: Pick<Rollup, "minute" | "service" | "environment" | "deployment_id" | "commit_sha" | "source" | "method" | "route">, r: Row) => {
1181 const k = [key.minute, key.service, key.environment, key.deployment_id, key.method, key.route].join("\u0000");
1182 let a = acc.get(k);
1183 if (!a) acc.set(k, (a = { ...key, requests: 0, errors: 0, s2: 0, s3: 0, s4: 0, s5: 0, latency: emptyHistogram(), cpu: emptyHistogram() }));
1184 a.requests++;
1185 const status = numOrNull(r.status);
1186 if (isError(status, String(r.outcome), r.error_name)) a.errors++;
1187 if (status !== null && status >= 200 && status < 600) a[`s${Math.floor(status / 100)}` as "s2" | "s3" | "s4" | "s5"]++;
1188 a.latency[bucketOf(Number(r.duration_us) / 1000)]++;
1189 if (r.cpu_ms !== null && r.cpu_ms !== undefined) a.cpu[bucketOf(Number(r.cpu_ms))]++;
1190 };
1191 for (const r of rows) {
1192 const base = {
1193 minute: Math.floor(Number(r.time_us) / MINUTE_US),
1194 service: String(r.service ?? ""),
1195 environment: String(r.environment ?? ""),
1196 deployment_id: String(r.deployment_id ?? ""),
1197 commit_sha: strOrNull(r.commit_sha),
1198 source: String(r.source),
1199 };
1200 add({ ...base, method: "*", route: "*" }, r);
1201 add({ ...base, method: String(r.method ?? ""), route: String(r.route ?? "") }, r);
1202 }
1203 return [...acc.values()].map((a) => ({ ...a, latency: encode(a.latency), cpu: encode(a.cpu) }));
1204}