| 1 | /** |
| 2 | * The telemetry service: observability for projects. What a project's |
| 3 | * running software did, as requests, logs, traces and metrics, kept in |
| 4 | * one store per project (src/store.ts) and read on the project's |
| 5 | * Observability page, the API and MCP. |
| 6 | * |
| 7 | * Doors in: |
| 8 | * - `tail`: the Tail Worker of every app on g1t.page and of the |
| 9 | * dispatcher in front of them (src/tail.ts). Nothing to set up. |
| 10 | * - `POST https://otel.g1t.sh/v1/{traces,logs,metrics}`: OTLP over HTTP, |
| 11 | * JSON or protobuf, gzipped or not, with a project's ingest token |
| 12 | * (`Authorization: Bearer g1to_…`). From g1t's own apps (their SDKs get |
| 13 | * the endpoint and a token as `OTEL_EXPORTER_OTLP_*`) and from software |
| 14 | * hosted anywhere else. |
| 15 | * - `POST /cloudflare/v1/{traces,logs}`: Cloudflare's own OpenTelemetry |
| 16 | * export of the apps on g1t.page (automatic spans for every fetch, |
| 17 | * KV, D1 and the rest), once a destination on the account points here |
| 18 | * with `CLOUDFLARE_EXPORT_KEY`. Off without the key. |
| 19 | * |
| 20 | * Reached by the site and the API through service bindings: |
| 21 | * `POST /rpc/<method>` (snake_case, `telemetryClient` in @g1t/contracts), |
| 22 | * called once they have checked the viewer may read (or manage) the |
| 23 | * project; and `GET /live?project_id=…`, a project's live tail. |
| 24 | */ |
| 25 | import { type RateLimitBinding, clientAddress, isLimited } from "@g1t/contracts/rate-limits"; |
| 26 | import { TELEMETRY_TOKEN_PREFIX, type ErrorGroup } from "@g1t/contracts/telemetry"; |
| 27 | |
| 28 | import { emptyBatch, type Batch } from "./batch.ts"; |
| 29 | import { IngestError, accepted, projectOfToken, readOtlp, refused, tokenOf, type Signal } from "./otlp/ingest.ts"; |
| 30 | import { OTLP_DOOR, type Door } from "./otlp/normalize.ts"; |
| 31 | import type { Store } from "./store.ts"; |
| 32 | import { readTail, type RequestHalf } from "./tail.ts"; |
| 33 | |
| 34 | export { TelemetryStore } from "./durable.ts"; |
| 35 | |
| 36 | type Env = { |
| 37 | STORES: DurableObjectNamespace; |
| 38 | /** Hostname to app, written by the deployments service (as the dispatcher reads it). */ |
| 39 | DOMAINS?: KVNamespace; |
| 40 | /** The secret Cloudflare's OpenTelemetry export sends, as `x-g1t-export-key`. */ |
| 41 | CLOUDFLARE_EXPORT_KEY?: string; |
| 42 | /** OTLP requests per client address (RATE_LIMITS in packages/contracts). Fails open without it. */ |
| 43 | TELEMETRY_INGEST_LIMIT?: RateLimitBinding; |
| 44 | }; |
| 45 | |
| 46 | const DEPLOYMENTS_DOMAIN = "g1t.page"; |
| 47 | |
| 48 | /** |
| 49 | * A project's store. Its methods are called over Durable Object RPC; the |
| 50 | * stub is typed by hand, as the generated type of so many methods is more |
| 51 | * than the compiler will work out. |
| 52 | */ |
| 53 | type StoreStub = { |
| 54 | [K in keyof Store as Store[K] extends (...args: any[]) => any ? K : never]: Store[K] extends (...args: infer A) => infer R |
| 55 | ? (...args: A) => Promise<Awaited<R>> |
| 56 | : never; |
| 57 | }; |
| 58 | |
| 59 | function store(env: Env, projectId: string): StoreStub { |
| 60 | return env.STORES.get(env.STORES.idFromName(projectId)) as unknown as StoreStub; |
| 61 | } |
| 62 | |
| 63 | const CORS = { |
| 64 | "access-control-allow-origin": "*", |
| 65 | "access-control-allow-methods": "POST, OPTIONS", |
| 66 | "access-control-allow-headers": "authorization, content-type, content-encoding, x-g1t-token", |
| 67 | "access-control-max-age": "86400", |
| 68 | }; |
| 69 | |
| 70 | function withCors(response: Response): Response { |
| 71 | const out = new Response(response.body, response); |
| 72 | for (const [k, v] of Object.entries(CORS)) out.headers.set(k, v); |
| 73 | return out; |
| 74 | } |
| 75 | |
| 76 | // ---- Apps on g1t.page, by script ------------------------------------------ |
| 77 | |
| 78 | type AppEntry = { project: string; deployment: string | null; environment: string | null; service: string; commit: string | null }; |
| 79 | |
| 80 | /** What the deployments service noted for each script, kept a minute in this isolate. */ |
| 81 | const apps = new Map<string, { at: number; app: AppEntry | null }>(); |
| 82 | |
| 83 | async function appOf(env: Env, script: string): Promise<AppEntry | null> { |
| 84 | const cached = apps.get(script); |
| 85 | if (cached && Date.now() - cached.at < 60_000) return cached.app; |
| 86 | let app: AppEntry | null = null; |
| 87 | try { |
| 88 | const entry = (await env.DOMAINS?.get(`${script}.${DEPLOYMENTS_DOMAIN}`, { type: "json", cacheTtl: 60 })) as { telemetry?: AppEntry } | null; |
| 89 | if (entry?.telemetry && typeof entry.telemetry.project === "string") app = entry.telemetry; |
| 90 | } catch (error) { |
| 91 | console.error("could not look up app", script, error); |
| 92 | } |
| 93 | apps.set(script, { at: Date.now(), app }); |
| 94 | if (apps.size > 5000) apps.delete(apps.keys().next().value!); |
| 95 | return app; |
| 96 | } |
| 97 | |
| 98 | /** Tail events, into each project's store. */ |
| 99 | async function onTail(events: readonly TraceItem[], env: Env): Promise<void> { |
| 100 | const byProject = new Map<string, { batch: Batch; halves: RequestHalf[] }>(); |
| 101 | for (const group of readTail(events)) { |
| 102 | let project = group.project_id; |
| 103 | let halves = group.halves; |
| 104 | if (!project) { |
| 105 | // The dispatcher's half of a request to an app it had no note for yet. |
| 106 | const app = await appOf(env, group.script); |
| 107 | if (!app) continue; |
| 108 | project = app.project; |
| 109 | halves = halves.map((h) => ({ ...h, service: app.service, environment: app.environment, deployment_id: app.deployment, commit_sha: app.commit, source: "cloudflare" })); |
| 110 | } |
| 111 | const into = byProject.get(project) ?? { batch: emptyBatch(), halves: [] }; |
| 112 | merge(into.batch, group.batch); |
| 113 | into.halves.push(...halves); |
| 114 | byProject.set(project, into); |
| 115 | } |
| 116 | await Promise.all( |
| 117 | [...byProject].map(([project, { batch, halves }]) => |
| 118 | store(env, project) |
| 119 | .ingest(batch, halves) |
| 120 | .catch((error: unknown) => console.error("could not keep tail events", project, error)), |
| 121 | ), |
| 122 | ); |
| 123 | } |
| 124 | |
| 125 | function merge(into: Batch, from: Batch) { |
| 126 | into.requests.push(...from.requests); |
| 127 | into.logs.push(...from.logs); |
| 128 | into.spans.push(...from.spans); |
| 129 | into.metrics.push(...from.metrics); |
| 130 | into.errors.push(...from.errors); |
| 131 | into.bytes += from.bytes; |
| 132 | } |
| 133 | |
| 134 | // ---- OTLP ----------------------------------------------------------------- |
| 135 | |
| 136 | async function otlp(request: Request, env: Env, signal: Signal): Promise<Response> { |
| 137 | const token = tokenOf(request); |
| 138 | if (!token) return refused(401, `No token: send the project's ingest token as "Authorization: Bearer ${TELEMETRY_TOKEN_PREFIX}…". Make one under the project's Observability settings.`); |
| 139 | const project = projectOfToken(token); |
| 140 | if (!project) return refused(401, "That is not a g1t ingest token."); |
| 141 | // Counted before the body is read or the project's store is woken, so a |
| 142 | // flood of requests (with made-up tokens or real ones) costs little. |
| 143 | if (await isLimited(env.TELEMETRY_INGEST_LIMIT, `ip:${clientAddress(request)}`)) { |
| 144 | return refused(429, "Too many requests from this address. Batch more per request, and wait a minute: https://docs.g1t.sh/reference/rate-limits/"); |
| 145 | } |
| 146 | let read: { batch: Batch; protobuf: boolean }; |
| 147 | try { |
| 148 | read = await readOtlp(request, signal, OTLP_DOOR); |
| 149 | } catch (error) { |
| 150 | if (error instanceof IngestError) return refused(error.status, error.message); |
| 151 | throw error; |
| 152 | } |
| 153 | const done = await store(env, project).ingest(read.batch, [], token); |
| 154 | if (!done.accepted) return refused(401, "That token is not one of the project's, or it was revoked."); |
| 155 | return accepted(read.protobuf); |
| 156 | } |
| 157 | |
| 158 | /** |
| 159 | * Cloudflare's own export of an app on g1t.page: its resource names the |
| 160 | * Worker (`cloudflare.script_name`, or the service's name), which is |
| 161 | * filed under the app's project and deployment. |
| 162 | */ |
| 163 | async function cloudflareExport(request: Request, env: Env, signal: Signal): Promise<Response> { |
| 164 | if (!env.CLOUDFLARE_EXPORT_KEY) return refused(404, "Cloudflare's export is not set up here."); |
| 165 | if (request.headers.get("x-g1t-export-key") !== env.CLOUDFLARE_EXPORT_KEY) return refused(401, "Wrong export key."); |
| 166 | if (signal === "metrics") return accepted(false); |
| 167 | // Resources are read twice: first to find the scripts, then filed. |
| 168 | const body = await request.clone().json().catch(() => null) as { resourceSpans?: { resource?: { attributes?: { key: string; value?: { stringValue?: string } }[] } }[]; resourceLogs?: { resource?: { attributes?: { key: string; value?: { stringValue?: string } }[] } }[] } | null; |
| 169 | const scripts = new Set<string>(); |
| 170 | for (const r of [...(body?.resourceSpans ?? []), ...(body?.resourceLogs ?? [])]) { |
| 171 | const attrs = r.resource?.attributes ?? []; |
| 172 | const script = attrs.find((a) => a.key === "cloudflare.script_name")?.value?.stringValue ?? attrs.find((a) => a.key === "service.name")?.value?.stringValue; |
| 173 | if (script) scripts.add(script); |
| 174 | } |
| 175 | const known = new Map<string, AppEntry>(); |
| 176 | for (const script of scripts) { |
| 177 | const app = await appOf(env, script); |
| 178 | if (app) known.set(script, app); |
| 179 | } |
| 180 | if (!known.size) return accepted(false); |
| 181 | for (const [script, app] of known) { |
| 182 | const door: Door = { |
| 183 | source: "cloudflare", |
| 184 | // Cloudflare's spans of a request are under the request the tail |
| 185 | // events already counted: they are not counted again. |
| 186 | countRequests: false, |
| 187 | origin: (resource) => |
| 188 | (resource["cloudflare.script_name"] ?? resource["service.name"]) === script |
| 189 | ? { service: app.service, environment: app.environment, deployment_id: app.deployment, commit_sha: app.commit, source: "cloudflare" } |
| 190 | : null, |
| 191 | }; |
| 192 | const read = await readOtlp(request.clone(), signal, door); |
| 193 | await store(env, app.project).ingest({ ...read.batch, requests: [], metrics: [], bytes: Math.round(read.batch.bytes / known.size) }, []); |
| 194 | } |
| 195 | return accepted(false); |
| 196 | } |
| 197 | |
| 198 | // ---- RPC ------------------------------------------------------------------ |
| 199 | |
| 200 | /** `POST /rpc/<method>`: each is one project's, named by `project_id`. */ |
| 201 | async function rpc(env: Env, method: string, args: Record<string, any>): Promise<unknown> { |
| 202 | const projectId = String(args.project_id ?? ""); |
| 203 | if (!projectId) throw new IngestError(400, "No project_id."); |
| 204 | const s = store(env, projectId); |
| 205 | switch (method) { |
| 206 | case "overview": |
| 207 | return s.overview(args); |
| 208 | case "requests": |
| 209 | return s.requests(args); |
| 210 | case "request": |
| 211 | return s.request(String(args.id)); |
| 212 | case "logs": |
| 213 | return s.logs(args); |
| 214 | case "traces": |
| 215 | return s.traces(args); |
| 216 | case "trace": |
| 217 | return s.trace(String(args.trace_id)); |
| 218 | case "errors": |
| 219 | return s.errors(args); |
| 220 | case "set_error": |
| 221 | return s.setError(String(args.fingerprint), { state: args.state as ErrorGroup["state"] | undefined, issue_number: args.issue_number }); |
| 222 | case "metrics": |
| 223 | return s.metrics(); |
| 224 | case "metric": |
| 225 | return s.metric({ ...args, name: String(args.name) }); |
| 226 | case "facets": |
| 227 | return s.facets(); |
| 228 | case "settings": |
| 229 | return s.settings(); |
| 230 | case "set_settings": |
| 231 | return s.setSettings(args); |
| 232 | case "tokens": |
| 233 | return s.tokens(); |
| 234 | case "create_token": |
| 235 | return s.createToken(projectId, String(args.name ?? ""), args.created_by ?? null); |
| 236 | case "revoke_token": |
| 237 | return s.revokeToken(String(args.id)); |
| 238 | case "deployment_token": |
| 239 | return s.deploymentToken(projectId); |
| 240 | case "exports": |
| 241 | return s.exports(); |
| 242 | case "put_export": |
| 243 | return s.putExport(args as never); |
| 244 | case "delete_export": |
| 245 | return s.deleteExport(String(args.id)); |
| 246 | case "usage": |
| 247 | return s.usage(); |
| 248 | case "purge": |
| 249 | return s.purge(); |
| 250 | default: |
| 251 | return undefined; |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | export default { |
| 256 | async fetch(request: Request, env: Env): Promise<Response> { |
| 257 | const url = new URL(request.url); |
| 258 | const otlpPath = /^\/v1\/(traces|logs|metrics)\/?$/.exec(url.pathname); |
| 259 | if (otlpPath) { |
| 260 | if (request.method === "OPTIONS") return new Response(null, { status: 204, headers: CORS }); |
| 261 | if (request.method !== "POST") return withCors(refused(405, "OTLP is sent with POST.")); |
| 262 | return withCors(await otlp(request, env, otlpPath[1] as Signal)); |
| 263 | } |
| 264 | const cfPath = /^\/cloudflare\/v1\/(traces|logs|metrics)\/?$/.exec(url.pathname); |
| 265 | if (cfPath && request.method === "POST") return cloudflareExport(request, env, cfPath[1] as Signal); |
| 266 | |
| 267 | // A project's live tail: the site checked the viewer may read it. |
| 268 | if (url.pathname === "/live" && request.method === "GET") { |
| 269 | const project = url.searchParams.get("project_id"); |
| 270 | if (!project) return new Response("No project_id.\n", { status: 400 }); |
| 271 | return env.STORES.get(env.STORES.idFromName(project)).fetch(request); |
| 272 | } |
| 273 | |
| 274 | const rpcPath = /^\/rpc\/([a-z_]+)$/.exec(url.pathname); |
| 275 | if (rpcPath && request.method === "POST") { |
| 276 | const args = (await request.json().catch(() => ({}))) as Record<string, unknown>; |
| 277 | try { |
| 278 | const result = await rpc(env, rpcPath[1], args); |
| 279 | return result === undefined ? new Response("Unknown method\n", { status: 404 }) : Response.json(result); |
| 280 | } catch (error) { |
| 281 | const status = error instanceof IngestError ? error.status : 400; |
| 282 | return new Response(String(error).replace(/^Error: /, ""), { status }); |
| 283 | } |
| 284 | } |
| 285 | |
| 286 | if (url.pathname === "/" && request.method === "GET") { |
| 287 | return new Response( |
| 288 | "g1t's OpenTelemetry endpoint. Send OTLP over HTTP to /v1/traces, /v1/logs and /v1/metrics with a project's ingest token.\nhttps://docs.g1t.sh/guides/observability/\n", |
| 289 | { headers: { "content-type": "text/plain; charset=utf-8" } }, |
| 290 | ); |
| 291 | } |
| 292 | return new Response("Not found\n", { status: 404 }); |
| 293 | }, |
| 294 | |
| 295 | async tail(events: TraceItem[], env: Env): Promise<void> { |
| 296 | await onTail(events, env); |
| 297 | }, |
| 298 | } satisfies ExportedHandler<Env>; |