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.

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