Skip to content
Sign up

flagon-io/g1t

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