Skip to content
Sign up

flagon-io/g1t

public
0
411 linesCodeBlameRaw
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 */
25import { type RateLimitBinding, clientAddress, isLimited } from "@g1t/contracts/rate-limits";
26import { type Standing, capOf, dailyCap, freeMonthlyBytes, overCapMessage } from "./quota.ts";
27import type { WorkspaceMeter } from "./meter.ts";
28import { TELEMETRY_TOKEN_PREFIX, type ErrorGroup } from "@g1t/contracts/telemetry";
29
30import { emptyBatch, type Batch } from "./batch.ts";
31import { IngestError, accepted, projectOfToken, readOtlp, refused, tokenOf, type Signal } from "./otlp/ingest.ts";
32import { OTLP_DOOR, type Door } from "./otlp/normalize.ts";
33import type { Store } from "./store.ts";
34import { readTail, type RequestHalf } from "./tail.ts";
35
36export { TelemetryMeter, TelemetryStore } from "./durable.ts";
37
38type Env = {
39 STORES: DurableObjectNamespace;
40 /** One meter per workspace, by slug (./meter.ts): its daily share, month total and last known plan. */
41 METERS?: DurableObjectNamespace;
42 /** Hostname to app, written by the deployments service (as the dispatcher reads it). */
43 DOMAINS?: KVNamespace;
44 /** The secret Cloudflare's OpenTelemetry export sends, as `x-g1t-export-key`. */
45 CLOUDFLARE_EXPORT_KEY?: string;
46 /** OTLP requests per client address (RATE_LIMITS in packages/contracts). Fails open without it. */
47 TELEMETRY_INGEST_LIMIT?: RateLimitBinding;
48 /**
49 * Whether a project's workspace has the plan, for the free cap
50 * (./quota.ts), and where the plan's month total goes. Without it, no cap.
51 */
52 BILLING?: { fetch: typeof fetch };
53 /** What a workspace's projects send free a month between them, in bytes (./quota.ts); 0 is no cap. */
54 FREE_TELEMETRY_BYTES_PER_MONTH?: string;
55};
56
57/** A project's workspace and its daily cap (null: none). */
58type Quota = { workspace: string | null; cap: number | null };
59
60/** Each project's quota, kept ten minutes in this isolate. */
61const caps = new Map<string, { at: number; quota: Quota }>();
62const CAP_TTL_MS = 10 * 60_000;
63
64/**
65 * A project's workspace (as the site last told its store) and the day's
66 * share the workspace's projects may send between them: none on the plan,
67 * a thirtieth of the free amount on the free plan, and 0 for a workspace
68 * g1t staff put a hold on (or suspended). When billing does not answer,
69 * the plan it last said holds (the workspace's meter keeps it), and a
70 * workspace it never said anything about is free. A project whose
71 * workspace is not known yet is neither capped nor metered; deployed apps'
72 * stores learn theirs from the deployments service.
73 */
74async function quotaFor(env: Env, projectId: string): Promise<Quota> {
75 const cached = caps.get(projectId);
76 if (cached && Date.now() - cached.at < CAP_TTL_MS) return cached.quota;
77 let quota: Quota = { workspace: null, cap: null };
78 const workspace = await store(env, projectId)
79 .workspaceOf()
80 .catch((error: unknown) => {
81 console.error("could not read the project's workspace", projectId, error);
82 return null;
83 });
84 // Without billing (self-hosted without it) there is no plan, and no cap.
85 if (workspace && env.BILLING) {
86 const meter = env.METERS ? meterOf(env, workspace) : null;
87 let standing: Standing | null = null;
88 try {
89 const { billingClient } = await import("@g1t/contracts");
90 const entitlements = await billingClient(env.BILLING as never).entitlements(workspace);
91 standing = entitlements.held ? "held" : entitlements.plan === "free" ? "free" : "paid";
92 await meter?.setStanding(standing).catch((error: unknown) => console.error("could not remember the workspace's plan", workspace, error));
93 } catch (error) {
94 console.error("could not tell whether the project's workspace is free; keeping the plan it was last on", projectId, error);
95 standing = (await meter?.standing().catch(() => null)) ?? null;
96 }
97 quota = { workspace, cap: capOf(standing, dailyCap(freeMonthlyBytes(env.FREE_TELEMETRY_BYTES_PER_MONTH))) };
98 } else if (workspace) {
99 quota = { workspace, cap: null };
100 }
101 caps.set(projectId, { at: Date.now(), quota });
102 if (caps.size > 5000) caps.delete(caps.keys().next().value!);
103 return quota;
104}
105
106/** Whether the workspace's projects are past today's share with `adding` more. Never over without a cap. */
107async function overQuota(env: Env, quota: Quota, adding: number): Promise<boolean> {
108 if (!quota.workspace || quota.cap == null) return false;
109 // A held workspace takes nothing, whatever its meter says.
110 if (quota.cap === 0) return true;
111 if (!env.METERS) return adding > quota.cap;
112 return meterOf(env, quota.workspace)
113 .over(adding, quota.cap)
114 .catch((error: unknown) => {
115 // The meter could not say: the cap still holds, on this request alone.
116 console.error("could not read the workspace's meter", quota.workspace, error);
117 return adding > (quota.cap ?? 0);
118 });
119}
120
121/** Counts what a project kept on its workspace's meter. */
122async function meterKept(env: Env, quota: Quota, bytes: number | undefined): Promise<void> {
123 if (!quota.workspace || !env.METERS || !bytes) return;
124 await meterOf(env, quota.workspace)
125 .add(quota.workspace, bytes)
126 .catch((error: unknown) => console.error("could not meter the workspace's telemetry", quota.workspace, bytes, error));
127}
128
129type MeterStub = {
130 [K in keyof WorkspaceMeter as WorkspaceMeter[K] extends (...args: any[]) => any ? K : never]: WorkspaceMeter[K] extends (...args: infer A) => infer R
131 ? (...args: A) => Promise<Awaited<R>>
132 : never;
133};
134
135function meterOf(env: Env, workspace: string): MeterStub {
136 if (!env.METERS) throw new Error("No METERS binding.");
137 return env.METERS.get(env.METERS.idFromName(workspace)) as unknown as MeterStub;
138}
139
140const DEPLOYMENTS_DOMAIN = "g1t.page";
141
142/**
143 * A project's store. Its methods are called over Durable Object RPC; the
144 * stub is typed by hand, as the generated type of so many methods is more
145 * than the compiler will work out.
146 */
147type StoreStub = {
148 [K in keyof Store as Store[K] extends (...args: any[]) => any ? K : never]: Store[K] extends (...args: infer A) => infer R
149 ? (...args: A) => Promise<Awaited<R>>
150 : never;
151};
152
153function store(env: Env, projectId: string): StoreStub {
154 return env.STORES.get(env.STORES.idFromName(projectId)) as unknown as StoreStub;
155}
156
157const CORS = {
158 "access-control-allow-origin": "*",
159 "access-control-allow-methods": "POST, OPTIONS",
160 "access-control-allow-headers": "authorization, content-type, content-encoding, x-g1t-token",
161 "access-control-max-age": "86400",
162};
163
164function withCors(response: Response): Response {
165 const out = new Response(response.body, response);
166 for (const [k, v] of Object.entries(CORS)) out.headers.set(k, v);
167 return out;
168}
169
170// ---- Apps on g1t.page, by script ------------------------------------------
171
172type AppEntry = { project: string; deployment: string | null; environment: string | null; service: string; commit: string | null };
173
174/** What the deployments service noted for each script, kept a minute in this isolate. */
175const apps = new Map<string, { at: number; app: AppEntry | null }>();
176
177async function appOf(env: Env, script: string): Promise<AppEntry | null> {
178 const cached = apps.get(script);
179 if (cached && Date.now() - cached.at < 60_000) return cached.app;
180 let app: AppEntry | null = null;
181 try {
182 const entry = (await env.DOMAINS?.get(`${script}.${DEPLOYMENTS_DOMAIN}`, { type: "json", cacheTtl: 60 })) as { telemetry?: AppEntry } | null;
183 if (entry?.telemetry && typeof entry.telemetry.project === "string") app = entry.telemetry;
184 } catch (error) {
185 console.error("could not look up app", script, error);
186 }
187 apps.set(script, { at: Date.now(), app });
188 if (apps.size > 5000) apps.delete(apps.keys().next().value!);
189 return app;
190}
191
192/** Tail events, into each project's store. */
193async function onTail(events: readonly TraceItem[], env: Env): Promise<void> {
194 const byProject = new Map<string, { batch: Batch; halves: RequestHalf[] }>();
195 for (const group of readTail(events)) {
196 let project = group.project_id;
197 let halves = group.halves;
198 if (!project) {
199 // The dispatcher's half of a request to an app it had no note for yet.
200 const app = await appOf(env, group.script);
201 if (!app) continue;
202 project = app.project;
203 halves = halves.map((h) => ({ ...h, service: app.service, environment: app.environment, deployment_id: app.deployment, commit_sha: app.commit, source: "cloudflare" }));
204 }
205 const into = byProject.get(project) ?? { batch: emptyBatch(), halves: [] };
206 merge(into.batch, group.batch);
207 into.halves.push(...halves);
208 byProject.set(project, into);
209 }
210 await Promise.all(
211 [...byProject].map(async ([project, { batch, halves }]) => {
212 const quota = await quotaFor(env, project);
213 const done = await store(env, project)
214 .ingest(batch, halves, undefined, await overQuota(env, quota, batch.bytes))
215 .catch((error: unknown) => {
216 console.error("could not keep tail events", project, error);
217 return null;
218 });
219 if (done?.accepted) await meterKept(env, quota, done.bytes);
220 }),
221 );
222}
223
224function merge(into: Batch, from: Batch) {
225 into.requests.push(...from.requests);
226 into.logs.push(...from.logs);
227 into.spans.push(...from.spans);
228 into.metrics.push(...from.metrics);
229 into.errors.push(...from.errors);
230 into.bytes += from.bytes;
231}
232
233// ---- OTLP -----------------------------------------------------------------
234
235async function otlp(request: Request, env: Env, signal: Signal): Promise<Response> {
236 const token = tokenOf(request);
237 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.`);
238 const project = projectOfToken(token);
239 if (!project) return refused(401, "That is not a g1t ingest token.");
240 // Counted before the body is read or the project's store is woken, so a
241 // flood of requests (with made-up tokens or real ones) costs little.
242 if (await isLimited(env.TELEMETRY_INGEST_LIMIT, `ip:${clientAddress(request)}`)) {
243 return refused(429, "Too many requests from this address. Batch more per request, and wait a minute: https://docs.g1t.sh/reference/rate-limits/");
244 }
245 let read: { batch: Batch; protobuf: boolean };
246 try {
247 read = await readOtlp(request, signal, OTLP_DOOR);
248 } catch (error) {
249 if (error instanceof IngestError) return refused(error.status, error.message);
250 throw error;
251 }
252 const quota = await quotaFor(env, project);
253 const done = await store(env, project).ingest(read.batch, [], token, await overQuota(env, quota, read.batch.bytes));
254 if (!done.accepted && done.reason === "quota") return refused(429, overCapMessage(quota.cap ?? 0));
255 if (!done.accepted) return refused(401, "That token is not one of the project's, or it was revoked.");
256 await meterKept(env, quota, done.bytes);
257 return accepted(read.protobuf);
258}
259
260/**
261 * Cloudflare's own export of an app on g1t.page: its resource names the
262 * Worker (`cloudflare.script_name`, or the service's name), which is
263 * filed under the app's project and deployment.
264 */
265async function cloudflareExport(request: Request, env: Env, signal: Signal): Promise<Response> {
266 if (!env.CLOUDFLARE_EXPORT_KEY) return refused(404, "Cloudflare's export is not set up here.");
267 if (request.headers.get("x-g1t-export-key") !== env.CLOUDFLARE_EXPORT_KEY) return refused(401, "Wrong export key.");
268 if (signal === "metrics") return accepted(false);
269 // Resources are read twice: first to find the scripts, then filed.
270 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;
271 const scripts = new Set<string>();
272 for (const r of [...(body?.resourceSpans ?? []), ...(body?.resourceLogs ?? [])]) {
273 const attrs = r.resource?.attributes ?? [];
274 const script = attrs.find((a) => a.key === "cloudflare.script_name")?.value?.stringValue ?? attrs.find((a) => a.key === "service.name")?.value?.stringValue;
275 if (script) scripts.add(script);
276 }
277 const known = new Map<string, AppEntry>();
278 for (const script of scripts) {
279 const app = await appOf(env, script);
280 if (app) known.set(script, app);
281 }
282 if (!known.size) return accepted(false);
283 for (const [script, app] of known) {
284 const door: Door = {
285 source: "cloudflare",
286 // Cloudflare's spans of a request are under the request the tail
287 // events already counted: they are not counted again.
288 countRequests: false,
289 origin: (resource) =>
290 (resource["cloudflare.script_name"] ?? resource["service.name"]) === script
291 ? { service: app.service, environment: app.environment, deployment_id: app.deployment, commit_sha: app.commit, source: "cloudflare" }
292 : null,
293 };
294 const read = await readOtlp(request.clone(), signal, door);
295 // Not capped (Cloudflare sends it whatever g1t answers), but counted
296 // on the workspace's meter like everything kept.
297 const done = await store(env, app.project).ingest({ ...read.batch, requests: [], metrics: [], bytes: Math.round(read.batch.bytes / known.size) }, []);
298 if (done.accepted) await meterKept(env, await quotaFor(env, app.project), done.bytes);
299 }
300 return accepted(false);
301}
302
303// ---- RPC ------------------------------------------------------------------
304
305/** `POST /rpc/<method>`: each is one project's, named by `project_id`. */
306async function rpc(env: Env, method: string, args: Record<string, any>): Promise<unknown> {
307 const projectId = String(args.project_id ?? "");
308 if (!projectId) throw new IngestError(400, "No project_id.");
309 const s = store(env, projectId);
310 // The site says which workspace the project is in when it shows its
311 // settings and makes a token, and the deployments service when it deploys
312 // one of its apps: the store keeps it, for the workspace's cap and meter.
313 const tellsWorkspace = ["settings", "create_token", "usage", "deployment_token"].includes(method);
314 if (typeof args.workspace === "string" && tellsWorkspace) {
315 await s.setWorkspace(args.workspace);
316 caps.delete(projectId);
317 }
318 switch (method) {
319 case "overview":
320 return s.overview(args);
321 case "requests":
322 return s.requests(args);
323 case "request":
324 return s.request(String(args.id));
325 case "logs":
326 return s.logs(args);
327 case "traces":
328 return s.traces(args);
329 case "trace":
330 return s.trace(String(args.trace_id));
331 case "errors":
332 return s.errors(args);
333 case "set_error":
334 return s.setError(String(args.fingerprint), { state: args.state as ErrorGroup["state"] | undefined, issue_number: args.issue_number });
335 case "metrics":
336 return s.metrics();
337 case "metric":
338 return s.metric({ ...args, name: String(args.name) });
339 case "facets":
340 return s.facets();
341 case "settings":
342 return s.settings();
343 case "set_settings":
344 return s.setSettings(args);
345 case "tokens":
346 return s.tokens();
347 case "create_token":
348 return s.createToken(projectId, String(args.name ?? ""), args.created_by ?? null);
349 case "revoke_token":
350 return s.revokeToken(String(args.id));
351 case "deployment_token":
352 return s.deploymentToken(projectId);
353 case "exports":
354 return s.exports();
355 case "put_export":
356 return s.putExport(args as never);
357 case "delete_export":
358 return s.deleteExport(String(args.id));
359 case "usage":
360 return s.usage();
361 case "purge":
362 return s.purge();
363 default:
364 return undefined;
365 }
366}
367
368export default {
369 async fetch(request: Request, env: Env): Promise<Response> {
370 const url = new URL(request.url);
371 const otlpPath = /^\/v1\/(traces|logs|metrics)\/?$/.exec(url.pathname);
372 if (otlpPath) {
373 if (request.method === "OPTIONS") return new Response(null, { status: 204, headers: CORS });
374 if (request.method !== "POST") return withCors(refused(405, "OTLP is sent with POST."));
375 return withCors(await otlp(request, env, otlpPath[1] as Signal));
376 }
377 const cfPath = /^\/cloudflare\/v1\/(traces|logs|metrics)\/?$/.exec(url.pathname);
378 if (cfPath && request.method === "POST") return cloudflareExport(request, env, cfPath[1] as Signal);
379
380 // A project's live tail: the site checked the viewer may read it.
381 if (url.pathname === "/live" && request.method === "GET") {
382 const project = url.searchParams.get("project_id");
383 if (!project) return new Response("No project_id.\n", { status: 400 });
384 return env.STORES.get(env.STORES.idFromName(project)).fetch(request);
385 }
386
387 const rpcPath = /^\/rpc\/([a-z_]+)$/.exec(url.pathname);
388 if (rpcPath && request.method === "POST") {
389 const args = (await request.json().catch(() => ({}))) as Record<string, unknown>;
390 try {
391 const result = await rpc(env, rpcPath[1], args);
392 return result === undefined ? new Response("Unknown method\n", { status: 404 }) : Response.json(result);
393 } catch (error) {
394 const status = error instanceof IngestError ? error.status : 400;
395 return new Response(String(error).replace(/^Error: /, ""), { status });
396 }
397 }
398
399 if (url.pathname === "/" && request.method === "GET") {
400 return new Response(
401 "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",
402 { headers: { "content-type": "text/plain; charset=utf-8" } },
403 );
404 }
405 return new Response("Not found\n", { status: 404 });
406 },
407
408 async tail(events: TraceItem[], env: Env): Promise<void> {
409 await onTail(events, env);
410 },
411} satisfies ExportedHandler<Env>;