Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
428 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) => ({
204 ...h,
205 service: app.service,
206 environment: app.environment,
207 deployment_id: app.deployment,
208 commit_sha: app.commit,
209 source: "cloudflare",
210 }));
211 }
212 const into = byProject.get(project) ?? { batch: emptyBatch(), halves: [] };
213 merge(into.batch, group.batch);
214 into.halves.push(...halves);
215 byProject.set(project, into);
216 }
217 await Promise.all(
218 [...byProject].map(async ([project, { batch, halves }]) => {
219 const quota = await quotaFor(env, project);
220 const done = await store(env, project)
221 .ingest(batch, halves, undefined, await overQuota(env, quota, batch.bytes))
222 .catch((error: unknown) => {
223 console.error("could not keep tail events", project, error);
224 return null;
225 });
226 if (done?.accepted) await meterKept(env, quota, done.bytes);
227 }),
228 );
229}
230
231function merge(into: Batch, from: Batch) {
232 into.requests.push(...from.requests);
233 into.logs.push(...from.logs);
234 into.spans.push(...from.spans);
235 into.metrics.push(...from.metrics);
236 into.errors.push(...from.errors);
237 into.bytes += from.bytes;
238}
239
240// ---- OTLP -----------------------------------------------------------------
241
242async function otlp(request: Request, env: Env, signal: Signal): Promise<Response> {
243 const token = tokenOf(request);
244 if (!token)
245 return refused(
246 401,
247 `No token: send the project's ingest token as "Authorization: Bearer ${TELEMETRY_TOKEN_PREFIX}…". Make one under the project's Observability settings.`,
248 );
249 const project = projectOfToken(token);
250 if (!project) return refused(401, "That is not a g1t ingest token.");
251 // Counted before the body is read or the project's store is woken, so a
252 // flood of requests (with made-up tokens or real ones) costs little.
253 if (await isLimited(env.TELEMETRY_INGEST_LIMIT, `ip:${clientAddress(request)}`)) {
254 return refused(429, "Too many requests from this address. Batch more per request, and wait a minute: https://docs.g1t.sh/reference/rate-limits/");
255 }
256 let read: { batch: Batch; protobuf: boolean };
257 try {
258 read = await readOtlp(request, signal, OTLP_DOOR);
259 } catch (error) {
260 if (error instanceof IngestError) return refused(error.status, error.message);
261 throw error;
262 }
263 const quota = await quotaFor(env, project);
264 const done = await store(env, project).ingest(read.batch, [], token, await overQuota(env, quota, read.batch.bytes));
265 if (!done.accepted && done.reason === "quota") return refused(429, overCapMessage(quota.cap ?? 0));
266 if (!done.accepted) return refused(401, "That token is not one of the project's, or it was revoked.");
267 await meterKept(env, quota, done.bytes);
268 return accepted(read.protobuf);
269}
270
271/**
272 * Cloudflare's own export of an app on g1t.page: its resource names the
273 * Worker (`cloudflare.script_name`, or the service's name), which is
274 * filed under the app's project and deployment.
275 */
276async function cloudflareExport(request: Request, env: Env, signal: Signal): Promise<Response> {
277 if (!env.CLOUDFLARE_EXPORT_KEY) return refused(404, "Cloudflare's export is not set up here.");
278 if (request.headers.get("x-g1t-export-key") !== env.CLOUDFLARE_EXPORT_KEY) return refused(401, "Wrong export key.");
279 if (signal === "metrics") return accepted(false);
280 // Resources are read twice: first to find the scripts, then filed.
281 const body = (await request
282 .clone()
283 .json()
284 .catch(() => null)) as {
285 resourceSpans?: { resource?: { attributes?: { key: string; value?: { stringValue?: string } }[] } }[];
286 resourceLogs?: { resource?: { attributes?: { key: string; value?: { stringValue?: string } }[] } }[];
287 } | null;
288 const scripts = new Set<string>();
289 for (const r of [...(body?.resourceSpans ?? []), ...(body?.resourceLogs ?? [])]) {
290 const attrs = r.resource?.attributes ?? [];
291 const script = attrs.find((a) => a.key === "cloudflare.script_name")?.value?.stringValue ?? attrs.find((a) => a.key === "service.name")?.value?.stringValue;
292 if (script) scripts.add(script);
293 }
294 const known = new Map<string, AppEntry>();
295 for (const script of scripts) {
296 const app = await appOf(env, script);
297 if (app) known.set(script, app);
298 }
299 if (!known.size) return accepted(false);
300 for (const [script, app] of known) {
301 const door: Door = {
302 source: "cloudflare",
303 // Cloudflare's spans of a request are under the request the tail
304 // events already counted: they are not counted again.
305 countRequests: false,
306 origin: (resource) =>
307 (resource["cloudflare.script_name"] ?? resource["service.name"]) === script
308 ? { service: app.service, environment: app.environment, deployment_id: app.deployment, commit_sha: app.commit, source: "cloudflare" }
309 : null,
310 };
311 const read = await readOtlp(request.clone(), signal, door);
312 // Not capped (Cloudflare sends it whatever g1t answers), but counted
313 // on the workspace's meter like everything kept.
314 const done = await store(env, app.project).ingest({ ...read.batch, requests: [], metrics: [], bytes: Math.round(read.batch.bytes / known.size) }, []);
315 if (done.accepted) await meterKept(env, await quotaFor(env, app.project), done.bytes);
316 }
317 return accepted(false);
318}
319
320// ---- RPC ------------------------------------------------------------------
321
322/** `POST /rpc/<method>`: each is one project's, named by `project_id`. */
323async function rpc(env: Env, method: string, args: Record<string, any>): Promise<unknown> {
324 const projectId = String(args.project_id ?? "");
325 if (!projectId) throw new IngestError(400, "No project_id.");
326 const s = store(env, projectId);
327 // The site says which workspace the project is in when it shows its
328 // settings and makes a token, and the deployments service when it deploys
329 // one of its apps: the store keeps it, for the workspace's cap and meter.
330 const tellsWorkspace = ["settings", "create_token", "usage", "deployment_token"].includes(method);
331 if (typeof args.workspace === "string" && tellsWorkspace) {
332 await s.setWorkspace(args.workspace);
333 caps.delete(projectId);
334 }
335 switch (method) {
336 case "overview":
337 return s.overview(args);
338 case "requests":
339 return s.requests(args);
340 case "request":
341 return s.request(String(args.id));
342 case "logs":
343 return s.logs(args);
344 case "traces":
345 return s.traces(args);
346 case "trace":
347 return s.trace(String(args.trace_id));
348 case "errors":
349 return s.errors(args);
350 case "set_error":
351 return s.setError(String(args.fingerprint), { state: args.state as ErrorGroup["state"] | undefined, item: args.item });
352 case "metrics":
353 return s.metrics();
354 case "metric":
355 return s.metric({ ...args, name: String(args.name) });
356 case "facets":
357 return s.facets();
358 case "settings":
359 return s.settings();
360 case "set_settings":
361 return s.setSettings(args);
362 case "tokens":
363 return s.tokens();
364 case "create_token":
365 return s.createToken(projectId, String(args.name ?? ""), args.created_by ?? null);
366 case "revoke_token":
367 return s.revokeToken(String(args.id));
368 case "deployment_token":
369 return s.deploymentToken(projectId);
370 case "exports":
371 return s.exports();
372 case "put_export":
373 return s.putExport(args as never);
374 case "delete_export":
375 return s.deleteExport(String(args.id));
376 case "usage":
377 return s.usage();
378 case "purge":
379 return s.purge();
380 default:
381 return undefined;
382 }
383}
384
385export default {
386 async fetch(request: Request, env: Env): Promise<Response> {
387 const url = new URL(request.url);
388 const otlpPath = /^\/v1\/(traces|logs|metrics)\/?$/.exec(url.pathname);
389 if (otlpPath) {
390 if (request.method === "OPTIONS") return new Response(null, { status: 204, headers: CORS });
391 if (request.method !== "POST") return withCors(refused(405, "OTLP is sent with POST."));
392 return withCors(await otlp(request, env, otlpPath[1] as Signal));
393 }
394 const cfPath = /^\/cloudflare\/v1\/(traces|logs|metrics)\/?$/.exec(url.pathname);
395 if (cfPath && request.method === "POST") return cloudflareExport(request, env, cfPath[1] as Signal);
396
397 // A project's live tail: the site checked the viewer may read it.
398 if (url.pathname === "/live" && request.method === "GET") {
399 const project = url.searchParams.get("project_id");
400 if (!project) return new Response("No project_id.\n", { status: 400 });
401 return env.STORES.get(env.STORES.idFromName(project)).fetch(request);
402 }
403
404 const rpcPath = /^\/rpc\/([a-z_]+)$/.exec(url.pathname);
405 if (rpcPath && request.method === "POST") {
406 const args = (await request.json().catch(() => ({}))) as Record<string, unknown>;
407 try {
408 const result = await rpc(env, rpcPath[1], args);
409 return result === undefined ? new Response("Unknown method\n", { status: 404 }) : Response.json(result);
410 } catch (error) {
411 const status = error instanceof IngestError ? error.status : 400;
412 return new Response(String(error).replace(/^Error: /, ""), { status });
413 }
414 }
415
416 if (url.pathname === "/" && request.method === "GET") {
417 return new Response(
418 "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",
419 { headers: { "content-type": "text/plain; charset=utf-8" } },
420 );
421 }
422 return new Response("Not found\n", { status: 404 });
423 },
424
425 async tail(events: TraceItem[], env: Env): Promise<void> {
426 await onTail(events, env);
427 },
428} satisfies ExportedHandler<Env>;