Skip to content
Sign up

flagon-io/g1t

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