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