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