| 1 | /** |
| 2 | * Tail events from g1t.page, as rows. Two Workers name this service as |
| 3 | * their Tail Worker, and each tells half of a request: |
| 4 | * |
| 5 | * - **The dispatcher** (`services/pages`), in front of every app: it |
| 6 | * writes one `g1t-request` line per request (`REQUEST_MARK`) with the |
| 7 | * request's id, the app's script and the trace it began or continued, |
| 8 | * and its tail event has the status, the time taken, the method, the |
| 9 | * address and where in the world it was served. Every request is here, |
| 10 | * including those a site's files answered without running its code. |
| 11 | * - **The app itself**, when its code ran: the logs it wrote, the |
| 12 | * exceptions it threw, its CPU time and outcome, and its tags |
| 13 | * (`TELEMETRY_TAG`: the project and the deployment). The dispatcher |
| 14 | * handed it the request's id (`x-g1t-request-id`) and a `traceparent`, |
| 15 | * which is how the halves meet, and how spans an OpenTelemetry SDK in |
| 16 | * the app sends join the request's trace. |
| 17 | * |
| 18 | * Halves arrive apart, in either order; the project's store merges them |
| 19 | * by id (./store.ts). The dispatcher's half names the project the |
| 20 | * deployments service noted for the app's hostname; when it had none yet, |
| 21 | * the service looks the script up itself (./index.ts). |
| 22 | */ |
| 23 | import { TELEMETRY_TAG, type Attributes, type TelemetryLog, type TelemetryRequest } from "@g1t/contracts/telemetry"; |
| 24 | |
| 25 | import { emptyBatch, type Batch } from "./batch.ts"; |
| 26 | import { clip, hostPath, randomId, route, severityNumber, severityOf, text } from "./shape.ts"; |
| 27 | |
| 28 | /** The first argument of the dispatcher's line for each request. */ |
| 29 | export const REQUEST_MARK = "g1t-request"; |
| 30 | /** The header the dispatcher hands each app with the request's id. */ |
| 31 | export const REQUEST_ID_HEADER = "x-g1t-request-id"; |
| 32 | |
| 33 | /** |
| 34 | * What the dispatcher's line says: the request, its app, and what the |
| 35 | * deployments service noted for the app's hostname (`AppTelemetry` there), |
| 36 | * when the dispatcher had it. |
| 37 | */ |
| 38 | export type RequestMark = { |
| 39 | id: string; |
| 40 | script: string; |
| 41 | trace_id: string; |
| 42 | app?: { project: string; deployment: string | null; environment: string | null; service: string; commit: string | null } | null; |
| 43 | }; |
| 44 | |
| 45 | /** A request's half, merged with the other half by id in the store. */ |
| 46 | export type RequestHalf = Partial<TelemetryRequest> & { id: string; half: "edge" | "app" }; |
| 47 | |
| 48 | /** Rows for one project, or for one script whose project is not known yet. */ |
| 49 | export type TailGroup = { project_id: string | null; script: string; batch: Batch; halves: RequestHalf[] }; |
| 50 | |
| 51 | /** What a tail event is about: its project and deployment from its tags. */ |
| 52 | function tagged(tags: readonly string[] | undefined): { project_id: string | null; deployment_id: string | null; environment: string | null; service: string } { |
| 53 | let service = ""; |
| 54 | let project_id: string | null = null; |
| 55 | let deployment_id: string | null = null; |
| 56 | let environment: string | null = null; |
| 57 | for (const tag of tags ?? []) { |
| 58 | if (tag.startsWith(TELEMETRY_TAG.project)) project_id = tag.slice(TELEMETRY_TAG.project.length); |
| 59 | else if (tag.startsWith(TELEMETRY_TAG.deployment)) deployment_id = tag.slice(TELEMETRY_TAG.deployment.length); |
| 60 | else if (tag === "production" || tag === "preview") environment = tag; |
| 61 | // `project:<workspace>/<slug>`: the service is the project's name. |
| 62 | else if (tag.startsWith("project:")) service = tag.slice(tag.indexOf("/") + 1); |
| 63 | } |
| 64 | return { project_id, deployment_id, environment, service }; |
| 65 | } |
| 66 | |
| 67 | function markOf(item: TraceItem): RequestMark | null { |
| 68 | for (const log of item.logs) { |
| 69 | const message = log.message as unknown; |
| 70 | if (!Array.isArray(message) || message[0] !== REQUEST_MARK) continue; |
| 71 | const mark = message[1] as Partial<RequestMark> | undefined; |
| 72 | if (mark && typeof mark.id === "string" && typeof mark.script === "string") { |
| 73 | const app = mark.app && typeof mark.app.project === "string" ? mark.app : null; |
| 74 | return { id: mark.id, script: mark.script, trace_id: typeof mark.trace_id === "string" ? mark.trace_id : "", app }; |
| 75 | } |
| 76 | } |
| 77 | return null; |
| 78 | } |
| 79 | |
| 80 | function fetchInfo(item: TraceItem): TraceItemFetchEventInfo | null { |
| 81 | const event = item.event as TraceItemFetchEventInfo | null; |
| 82 | return event && typeof event === "object" && "request" in event ? event : null; |
| 83 | } |
| 84 | |
| 85 | function triggerOf(item: TraceItem): string { |
| 86 | const event = item.event as Record<string, unknown> | null; |
| 87 | if (!event) return "other"; |
| 88 | if ("request" in event) return "fetch"; |
| 89 | if ("cron" in event) return "scheduled"; |
| 90 | if ("queue" in event) return "queue"; |
| 91 | if ("mailFrom" in event) return "email"; |
| 92 | if ("rpcMethod" in event) return "rpc"; |
| 93 | if ("scheduledTime" in event) return "alarm"; |
| 94 | return "other"; |
| 95 | } |
| 96 | |
| 97 | /** The `traceparent` header's trace id, if it is a valid one. */ |
| 98 | export function traceIdOf(traceparent: string | undefined | null): string | null { |
| 99 | const m = /^[0-9a-f]{2}-([0-9a-f]{32})-([0-9a-f]{16})-[0-9a-f]{2}$/.exec((traceparent ?? "").trim().toLowerCase()); |
| 100 | return m && !/^0+$/.test(m[1]) ? m[1] : null; |
| 101 | } |
| 102 | |
| 103 | /** The status code of an event that failed before answering: what the visitor got. */ |
| 104 | function failedStatus(outcome: string): number | null { |
| 105 | if (outcome === "ok" || outcome === "canceled" || outcome === "responseStreamDisconnected") return null; |
| 106 | return outcome === "exceededCpu" || outcome === "exceededMemory" ? 503 : 500; |
| 107 | } |
| 108 | |
| 109 | /** Tail events as rows, grouped by the project (or, for the dispatcher's half, the script) they are of. */ |
| 110 | export function readTail(events: readonly TraceItem[]): TailGroup[] { |
| 111 | const groups = new Map<string, TailGroup>(); |
| 112 | const group = (project_id: string | null, script: string): TailGroup => { |
| 113 | const key = project_id ?? `script:${script}`; |
| 114 | let g = groups.get(key); |
| 115 | if (!g) groups.set(key, (g = { project_id, script, batch: emptyBatch(), halves: [] })); |
| 116 | return g; |
| 117 | }; |
| 118 | |
| 119 | for (const item of events) { |
| 120 | const time = (item.eventTimestamp ?? Date.now()) * 1000; |
| 121 | const info = fetchInfo(item); |
| 122 | const mark = markOf(item); |
| 123 | |
| 124 | if (mark) { |
| 125 | // The dispatcher's half: every request, whatever answered it. |
| 126 | const g = group(mark.app?.project ?? null, mark.script); |
| 127 | const where = info ? hostPath(info.request.url) : null; |
| 128 | const cf = (info?.request.cf ?? {}) as { colo?: string; country?: string }; |
| 129 | const status = info?.response?.status ?? failedStatus(item.outcome); |
| 130 | g.halves.push({ |
| 131 | id: mark.id, |
| 132 | half: "edge", |
| 133 | ...(mark.app |
| 134 | ? { service: mark.app.service, environment: mark.app.environment, deployment_id: mark.app.deployment, commit_sha: mark.app.commit, source: "cloudflare" as const } |
| 135 | : {}), |
| 136 | time_us: time, |
| 137 | duration_us: Math.round(item.wallTime * 1000), |
| 138 | trigger: "fetch", |
| 139 | method: info?.request.method ?? null, |
| 140 | url: where?.url ?? null, |
| 141 | route: where ? route(where.path) : null, |
| 142 | status, |
| 143 | region: cf.colo ?? null, |
| 144 | country: cf.country ?? null, |
| 145 | trace_id: mark.trace_id || null, |
| 146 | }); |
| 147 | g.batch.bytes += 256; |
| 148 | continue; |
| 149 | } |
| 150 | |
| 151 | const tags = tagged(item.scriptTags); |
| 152 | if (!tags.project_id) continue; |
| 153 | const g = group(tags.project_id, item.scriptName ?? ""); |
| 154 | const headers = info?.request.headers ?? {}; |
| 155 | const id = headers[REQUEST_ID_HEADER] || randomId(); |
| 156 | const trace_id = traceIdOf(headers.traceparent); |
| 157 | const origin = { |
| 158 | service: tags.service, |
| 159 | environment: tags.environment, |
| 160 | deployment_id: tags.deployment_id, |
| 161 | commit_sha: null, |
| 162 | source: "cloudflare" as const, |
| 163 | }; |
| 164 | |
| 165 | const logs: TelemetryLog[] = item.logs.map((log) => { |
| 166 | const number = severityNumber(log.level); |
| 167 | return { |
| 168 | ...origin, |
| 169 | id: randomId(), |
| 170 | time_us: log.timestamp * 1000, |
| 171 | severity: severityOf(number), |
| 172 | severity_number: number, |
| 173 | body: clip(text(log.message)), |
| 174 | trace_id, |
| 175 | span_id: null, |
| 176 | request_id: id, |
| 177 | attributes: (log.level === "log" ? {} : { "log.level": log.level }) as Attributes, |
| 178 | }; |
| 179 | }); |
| 180 | for (const e of item.exceptions) { |
| 181 | logs.push({ |
| 182 | ...origin, |
| 183 | id: randomId(), |
| 184 | time_us: e.timestamp * 1000, |
| 185 | severity: "error", |
| 186 | severity_number: 17, |
| 187 | body: clip(`${e.name}: ${e.message}${e.stack ? `\n${e.stack}` : ""}`), |
| 188 | trace_id, |
| 189 | span_id: null, |
| 190 | request_id: id, |
| 191 | attributes: { "exception.type": e.name, "exception.message": clip(e.message, 4096) }, |
| 192 | }); |
| 193 | g.batch.errors.push({ ...origin, time_us: e.timestamp * 1000, name: e.name, message: e.message, stack: e.stack ?? null, request_id: id, trace_id }); |
| 194 | } |
| 195 | if (item.truncated) { |
| 196 | logs.push({ |
| 197 | ...origin, |
| 198 | id: randomId(), |
| 199 | time_us: time, |
| 200 | severity: "warn", |
| 201 | severity_number: 13, |
| 202 | body: "Cloudflare cut this request's logs short: it wrote more than a tail event holds.", |
| 203 | trace_id, |
| 204 | span_id: null, |
| 205 | request_id: id, |
| 206 | attributes: {}, |
| 207 | }); |
| 208 | } |
| 209 | g.batch.logs.push(...logs); |
| 210 | |
| 211 | const where = info ? hostPath(info.request.url) : null; |
| 212 | const status = info?.response?.status ?? failedStatus(item.outcome); |
| 213 | const first = item.exceptions[0]; |
| 214 | g.halves.push({ |
| 215 | ...origin, |
| 216 | id, |
| 217 | half: "app", |
| 218 | time_us: time, |
| 219 | duration_us: Math.round(item.wallTime * 1000), |
| 220 | trigger: triggerOf(item), |
| 221 | method: info?.request.method ?? null, |
| 222 | url: where?.url ?? null, |
| 223 | route: where ? route(where.path) : null, |
| 224 | status, |
| 225 | outcome: item.outcome, |
| 226 | cpu_ms: item.cpuTime, |
| 227 | trace_id, |
| 228 | log_count: logs.length, |
| 229 | error: first ? { name: first.name, message: clip(first.message, 2048) } : item.outcome !== "ok" && item.outcome !== "canceled" ? { name: item.outcome, message: outcomeText(item.outcome) } : null, |
| 230 | }); |
| 231 | g.batch.bytes += logs.reduce((n, l) => n + l.body.length, 256); |
| 232 | } |
| 233 | return [...groups.values()]; |
| 234 | } |
| 235 | |
| 236 | /** What Cloudflare's outcome means, for the request's error line. */ |
| 237 | export function outcomeText(outcome: string): string { |
| 238 | switch (outcome) { |
| 239 | case "exception": |
| 240 | return "The code threw before it answered."; |
| 241 | case "exceededCpu": |
| 242 | return "The request used more CPU time than it is allowed."; |
| 243 | case "exceededMemory": |
| 244 | return "The isolate ran out of memory."; |
| 245 | case "exceededWallTime": |
| 246 | return "The request ran longer than it is allowed."; |
| 247 | case "scriptNotFound": |
| 248 | return "No app was there to answer."; |
| 249 | case "loadShed": |
| 250 | return "Cloudflare shed the request under load."; |
| 251 | default: |
| 252 | return `The request ended with ${outcome}.`; |
| 253 | } |
| 254 | } |