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
438 linesCodeBlameRaw
1/**
2 * The store end to end, on Node's SQLite standing in for the Durable
3 * Object's: tail events from the dispatcher and the app meet as one
4 * request, are counted into the overview, and read back as requests,
5 * logs, traces and errors.
6 */
7import assert from "node:assert/strict";
8import { DatabaseSync } from "node:sqlite";
9import test from "node:test";
10
11import { TELEMETRY_TAG } from "@g1t/contracts/telemetry";
12
13import { emptyBatch } from "./batch.ts";
14import { OTLP_DOOR, readTraces } from "./otlp/normalize.ts";
15import { Store, aggregate, ftsQuery, parseSearch, stepFor } from "./store.ts";
16import { REQUEST_ID_HEADER, REQUEST_MARK, readTail } from "./tail.ts";
17
18/** A Durable Object's state, as far as the store uses it, on an in-memory database. */
19function state() {
20 const db = new DatabaseSync(":memory:");
21 let alarm: number | null = null;
22 const sql = {
23 exec(query: string, ...args: unknown[]) {
24 if (
25 !args.length &&
26 query
27 .trim()
28 .split(";")
29 .filter((q) => q.trim()).length > 1
30 ) {
31 db.exec(query);
32 return { toArray: () => [], rowsWritten: 0 };
33 }
34 const rows = db.prepare(query).all(...(args as never[]));
35 const written = Number((db.prepare("SELECT changes() AS n").get() as { n: number }).n);
36 return { toArray: () => rows, rowsWritten: written };
37 },
38 get databaseSize() {
39 return 1024;
40 },
41 };
42 const waits: Promise<unknown>[] = [];
43 return {
44 storage: {
45 sql,
46 getAlarm: async () => alarm,
47 setAlarm: async (t: number) => void (alarm = t),
48 deleteAlarm: async () => void (alarm = null),
49 deleteAll: async () => {
50 for (const { name } of db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'logs_fts%'").all() as { name: string }[]) {
51 db.exec(`DELETE FROM "${name}"`);
52 }
53 },
54 },
55 getWebSockets: () => [],
56 acceptWebSocket: () => {},
57 waitUntil: (p: Promise<unknown>) => void waits.push(p),
58 blockConcurrencyWhile: async (f: () => Promise<unknown>) => f(),
59 } as unknown as DurableObjectState;
60}
61
62const NOW_MS = 1_760_000_000_000;
63const TRACE = "4bf92f3577b34da6a3ce929d0e0e4736";
64
65function edgeEvent(id: string, status: number, wallMs: number, at = NOW_MS): TraceItem {
66 return {
67 scriptName: "g1t-pages",
68 eventTimestamp: at,
69 outcome: "ok",
70 wallTime: wallMs,
71 cpuTime: 1,
72 truncated: false,
73 exceptions: [],
74 diagnosticsChannelEvents: [],
75 executionModel: "stateless",
76 logs: [
77 {
78 timestamp: at,
79 level: "log",
80 message: [
81 REQUEST_MARK,
82 { id, script: "web-acme", trace_id: TRACE, app: { project: "prj_1", deployment: "dep_1", environment: "production", service: "web", commit: "abc" } },
83 ],
84 },
85 ],
86 event: {
87 request: { url: `https://web-acme.g1t.page/users/${id.length}?q=1`, method: "GET", headers: {}, cf: { colo: "AMS", country: "NL" } },
88 response: { status },
89 },
90 } as unknown as TraceItem;
91}
92
93function appEvent(id: string, opts: { outcome?: string; exception?: string; logs?: string[]; at?: number } = {}): TraceItem {
94 const at = opts.at ?? NOW_MS;
95 return {
96 scriptName: "web-acme",
97 scriptTags: ["workspace:acme", "project:acme/web", "production", `${TELEMETRY_TAG.project}prj_1`, `${TELEMETRY_TAG.deployment}dep_1`],
98 eventTimestamp: at,
99 outcome: opts.outcome ?? "ok",
100 wallTime: 40,
101 cpuTime: 7,
102 truncated: false,
103 diagnosticsChannelEvents: [],
104 executionModel: "stateless",
105 exceptions: opts.exception
106 ? [{ timestamp: at + 1, name: "TypeError", message: opts.exception, stack: "TypeError\n at handler (src/index.ts:3:1)" }]
107 : [],
108 logs: (opts.logs ?? []).map((m, i) => ({ timestamp: at + i, level: i === 0 ? "log" : "warn", message: [m] })),
109 event: {
110 request: {
111 url: "https://web-acme.g1t.page/users/1",
112 method: "GET",
113 headers: { [REQUEST_ID_HEADER]: id, traceparent: `00-${TRACE}-${id}-01` },
114 },
115 response: opts.outcome && opts.outcome !== "ok" ? undefined : { status: 200 },
116 },
117 } as unknown as TraceItem;
118}
119
120async function ingestTail(store: Store, events: TraceItem[]) {
121 for (const g of readTail(events)) {
122 assert.equal(g.project_id, "prj_1");
123 await store.ingest(g.batch, g.halves);
124 }
125}
126
127test("the dispatcher's half and the app's meet as one request, in either order", async () => {
128 const store = new Store(state());
129 await ingestTail(store, [appEvent("aaaaaaaaaaaaaaa1", { logs: ["hello", "careful"] })]);
130 await ingestTail(store, [edgeEvent("aaaaaaaaaaaaaaa1", 200, 55)]);
131 await ingestTail(store, [edgeEvent("aaaaaaaaaaaaaaa2", 304, 3)]); // a site's file: no app half
132
133 const page = store.requests({ limit: 10 });
134 assert.equal(page.items.length, 2);
135 const one = store.request("aaaaaaaaaaaaaaa1")!;
136 assert.equal(one.request.status, 200);
137 assert.equal(one.request.duration_us, 55_000, "the dispatcher's time is the whole request");
138 assert.equal(one.request.cpu_ms, 7);
139 assert.equal(one.request.region, "AMS");
140 assert.equal(one.request.country, "NL");
141 assert.equal(one.request.service, "web");
142 assert.equal(one.request.deployment_id, "dep_1");
143 assert.equal(one.request.trace_id, TRACE);
144 assert.equal(one.request.route, "/users/:id");
145 assert.deepEqual(
146 one.logs.map((l) => [l.severity, l.body]),
147 [
148 ["info", "hello"],
149 ["warn", "careful"],
150 ],
151 );
152 const site = store.request("aaaaaaaaaaaaaaa2")!;
153 assert.equal(site.request.status, 304);
154 assert.equal(site.request.service, "web");
155});
156
157test("errors are grouped, counted, and open again when they come back", async () => {
158 const store = new Store(state());
159 await ingestTail(store, [
160 appEvent("bbbbbbbbbbbbbbb1", { outcome: "exception", exception: "Cannot read 'x' of user 1" }),
161 edgeEvent("bbbbbbbbbbbbbbb1", 502, 30),
162 ]);
163 await ingestTail(store, [appEvent("bbbbbbbbbbbbbbb2", { outcome: "exception", exception: "Cannot read 'y' of user 2" })]);
164 let groups = store.errors({});
165 assert.equal(groups.length, 1);
166 assert.equal(groups[0].count, 2);
167 assert.equal(groups[0].first_deployment_id, "dep_1");
168 store.setError(groups[0].fingerprint, { state: "resolved", item: 12 });
169 await ingestTail(store, [appEvent("bbbbbbbbbbbbbbb3", { outcome: "exception", exception: "Cannot read 'z' of user 3" })]);
170 groups = store.errors({});
171 assert.equal(groups[0].state, "open");
172 assert.equal(groups[0].item, 12);
173 const failed = store.requests({ errors_only: true });
174 assert.equal(failed.items.length, 3);
175 // An app that threw answered with an error page, whichever half said so.
176 assert.equal(store.requests({ status: ["5xx"] }).items.length, 3);
177 assert.equal(store.requests({ status: ["2xx"] }).items.length, 0);
178});
179
180test("the overview counts requests, by minute and route, whether rolled up yet or not", async () => {
181 const store = new Store(state());
182 for (let i = 0; i < 20; i++) {
183 const id = `c${String(i).padStart(15, "0")}`;
184 await ingestTail(store, [edgeEvent(id, i < 2 ? 500 : 200, 10 + i * 10, NOW_MS + i * 1000)]);
185 }
186 const range = { from_us: (NOW_MS - 3_600_000) * 1000, to_us: (NOW_MS + 3_600_000) * 1000 };
187 const before = store.overview(range);
188 assert.equal(before.totals.requests, 20);
189 assert.equal(before.totals.errors, 2);
190 // Counted into rollups, the numbers are the same.
191 store.roll((NOW_MS + 600_000) * 1000);
192 const after = store.overview(range);
193 assert.equal(after.totals.requests, 20);
194 assert.equal(after.totals.errors, 2);
195 assert.deepEqual(after.totals.latency, before.totals.latency);
196 assert.equal(after.routes[0].route, "/users/:id");
197 assert.equal(after.deployments[0].deployment_id, "dep_1");
198 assert.equal(after.regions[0].region, "AMS");
199 assert.equal(
200 after.points.reduce((n, p) => n + p.requests, 0),
201 20,
202 );
203 // Rolling again counts nothing twice.
204 store.roll((NOW_MS + 600_000) * 1000);
205 assert.equal(store.overview(range).totals.requests, 20);
206});
207
208test("a trace joins the request on g1t.page with the app's own spans and logs", async () => {
209 const store = new Store(state());
210 await ingestTail(store, [appEvent("ddddddddddddddd1", { logs: ["in handler"] }), edgeEvent("ddddddddddddddd1", 200, 80)]);
211 const spans = readTraces(
212 {
213 resourceSpans: [
214 {
215 resource: {
216 attributes: [
217 { key: "service.name", value: { stringValue: "web" } },
218 { key: "g1t.deployment.id", value: { stringValue: "dep_1" } },
219 ],
220 },
221 scopeSpans: [
222 {
223 spans: [
224 {
225 traceId: TRACE,
226 spanId: "eeeeeeeeeeeeeee1",
227 parentSpanId: "ddddddddddddddd1",
228 name: "db.query",
229 kind: 3,
230 startTimeUnixNano: `${NOW_MS}010000000`.slice(0, 19),
231 endTimeUnixNano: `${NOW_MS}050000000`.slice(0, 19),
232 },
233 ],
234 },
235 ],
236 },
237 ],
238 },
239 OTLP_DOOR,
240 10,
241 );
242 const ok = await store.ingest(spans, [], "wrong-token");
243 assert.equal(ok.accepted, false, "a token that is not the project's is refused");
244 const { secret } = await store.createToken("prj_1", "ci", "u1");
245 assert.equal((await store.ingest(spans, [], secret)).accepted, true);
246
247 const trace = store.trace(TRACE)!;
248 assert.deepEqual(
249 trace.spans.map((s) => [s.name, s.parent_span_id]),
250 [
251 ["GET /users/:id", null],
252 ["db.query", "ddddddddddddddd1"],
253 ],
254 );
255 assert.equal(trace.logs[0].body, "in handler");
256 // Found by its route too: an OTLP server span often has no address.
257 const { secret: other } = await store.createToken("prj_1", "api", null);
258 await store.ingest(
259 readTraces(
260 {
261 resourceSpans: [
262 {
263 resource: { attributes: [{ key: "service.name", value: { stringValue: "billing" } }] },
264 scopeSpans: [
265 {
266 spans: [
267 {
268 traceId: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
269 spanId: "bbbbbbbbbbbbbbbb",
270 name: "POST",
271 kind: 2,
272 startTimeUnixNano: `${NOW_MS}000000`,
273 attributes: [{ key: "http.route", value: { stringValue: "/invoices" } }],
274 },
275 ],
276 },
277 ],
278 },
279 ],
280 },
281 OTLP_DOOR,
282 10,
283 ),
284 [],
285 other,
286 );
287 assert.equal(store.requests({ text: "invoices" }).items.length, 1);
288 const list = store.traces({});
289 assert.equal(list.items[0].trace_id, TRACE);
290 assert.equal(list.items[0].name, "GET /users/:id");
291 assert.equal(list.items[0].span_count, 2);
292});
293
294test("logs are searched by words, levels and attributes, and paged", async () => {
295 const store = new Store(state());
296 const batch = emptyBatch();
297 for (let i = 0; i < 30; i++) {
298 batch.logs.push({
299 id: `log${String(i).padStart(3, "0")}`,
300 time_us: (NOW_MS + i) * 1000,
301 severity: i % 3 === 0 ? "error" : "info",
302 severity_number: i % 3 === 0 ? 17 : 9,
303 body: i % 2 ? `payment captured for order ${i}` : `cache miss on key ${i}`,
304 trace_id: null,
305 span_id: null,
306 request_id: null,
307 attributes: { "user.id": `u${i % 5}` },
308 service: "api",
309 environment: "production",
310 deployment_id: null,
311 commit_sha: null,
312 source: "otlp",
313 });
314 }
315 const { secret } = await store.createToken("prj_1", "x", null);
316 await store.ingest(batch, [], secret);
317 assert.equal(store.logs({ text: "payment" }).items.length, 15);
318 assert.equal(store.logs({ text: "payment captured" }).items.length, 15);
319 assert.equal(store.logs({ text: "pay" }).items.length, 15, "the last word is a prefix");
320 assert.equal(store.logs({ text: "level:error" }).items.length, 10);
321 assert.equal(store.logs({ severities: ["error"], text: "cache" }).items.length, 5);
322 assert.equal(store.logs({ text: "user.id=u1" }).items.length, 6);
323 const first = store.logs({ limit: 20 });
324 assert.equal(first.items.length, 20);
325 const second = store.logs({ limit: 20, cursor: first.next });
326 assert.equal(second.items.length, 10);
327 assert.equal(second.next, null);
328 assert.equal(new Set([...first.items, ...second.items].map((l) => l.id)).size, 30);
329});
330
331test("tokens: the deployments' one is kept and the same each time; others can be revoked", async () => {
332 const store = new Store(state());
333 const a = await store.deploymentToken("prj_1");
334 const b = await store.deploymentToken("prj_1");
335 assert.equal(a.secret, b.secret);
336 assert.match(a.secret, /^g1to_prj_1_[0-9a-f]{48}$/);
337 const made = await store.createToken("prj_1", "laptop", "u1");
338 const listed = store.tokens();
339 assert.deepEqual(
340 listed.map((t) => t.name),
341 ["laptop"],
342 "the deployments' token is g1t's, not listed",
343 );
344 assert.ok(!JSON.stringify(listed).includes(made.secret), "a token's secret is never listed");
345 const managedId = String((store.sql.exec("SELECT id FROM tokens WHERE managed = 1").toArray()[0] as { id: string }).id);
346 assert.equal(store.revokeToken(managedId), false);
347 assert.equal(store.revokeToken(made.token.id), true);
348 assert.equal(await store.checkToken(made.secret), false);
349});
350
351test("retention removes what is past it, and turning deployments off stops tail events", async () => {
352 const store = new Store(state());
353 await ingestTail(store, [appEvent("fffffffffffffff1", { logs: ["old"] }), edgeEvent("fffffffffffffff1", 200, 5)]);
354 store.prune((NOW_MS + 8 * 86_400_000) * 1000);
355 assert.equal(store.requests({}).items.length, 0);
356 assert.equal(store.logs({}).items.length, 0);
357 // Counted before they went.
358 assert.equal(store.overview({ from_us: (NOW_MS - 60_000) * 1000, to_us: (NOW_MS + 60_000) * 1000 }).totals.requests, 1);
359 store.setSettings({ deployments: false });
360 const [g] = readTail([appEvent("fffffffffffffff2")]);
361 assert.equal((await store.ingest(g.batch, g.halves)).accepted, false);
362 assert.throws(() => store.setSettings({ retention_days: 365 }));
363});
364
365test("exports refuse addresses that loop or stay inside a network", () => {
366 const store = new Store(state());
367 assert.throws(() => store.putExport({ name: "x", endpoint: "https://otel.g1t.sh", signals: ["traces"], enabled: true }));
368 assert.throws(() => store.putExport({ name: "x", endpoint: "http://collector.example.com", signals: ["traces"], enabled: true }));
369 assert.throws(() => store.putExport({ name: "x", endpoint: "https://10.0.0.4", signals: ["traces"], enabled: true }));
370 const made = store.putExport({
371 name: "Honeycomb",
372 endpoint: "https://api.honeycomb.io",
373 signals: ["traces", "logs"],
374 headers: { "x-honeycomb-team": "secret" },
375 enabled: true,
376 });
377 assert.deepEqual(made.header_names, ["x-honeycomb-team"]);
378 assert.ok(!JSON.stringify(store.exports()).includes("secret"));
379 // Changing it without giving the header again keeps it.
380 store.putExport({ id: made.id, name: "Honeycomb", endpoint: "https://api.honeycomb.io", signals: ["traces"], enabled: false });
381 assert.deepEqual(store.exports()[0].header_names, ["x-honeycomb-team"]);
382});
383
384test("search text, FTS queries and chart steps", () => {
385 assert.deepEqual(parseSearch('status:500 "two words" user.id=7 timeout https://x.y'), {
386 words: ["two words", "timeout", "https://x.y"],
387 pairs: [
388 ["status", "500"],
389 ["user.id", "7"],
390 ],
391 });
392 assert.equal(ftsQuery(["a", 'b"c']), '"a" AND "b""c"*');
393 assert.equal(stepFor(3_600_000_000), 60);
394 assert.equal(stepFor(86_400_000_000), 900);
395 assert.equal(aggregate([]).length, 0);
396});
397
398test("a project of a workspace past its day's share is refused, and tail events keep only their requests", async () => {
399 const store = new Store(state());
400 const { secret } = await store.createToken("prj_1", "ci", "u1");
401 const batch = (bytes: number) => ({ ...emptyBatch(), bytes });
402 assert.deepEqual(await store.ingest(batch(60), [], secret, false), { accepted: true, bytes: 60 });
403 assert.equal(store.todayBytes(), 60);
404 const refused = await store.ingest(batch(50), [], secret, true);
405 assert.deepEqual(refused, { accepted: false, reason: "quota" });
406 assert.equal(store.todayBytes(), 60, "nothing refused is counted");
407 // Not over (the plan, or a workspace not known yet): nothing is refused.
408 assert.equal((await store.ingest(batch(5000), [], secret)).accepted, true);
409 // Tail events past the share keep their requests, not their logs, and count only what they kept.
410 const tail = { ...emptyBatch(), bytes: 10_000 };
411 tail.logs.push({
412 id: "l1",
413 time_us: 1,
414 severity: "info",
415 severity_number: 9,
416 body: "x".repeat(9000),
417 trace_id: null,
418 span_id: null,
419 request_id: null,
420 attributes: {},
421 service: "web",
422 environment: null,
423 deployment_id: null,
424 commit_sha: null,
425 source: "cloudflare",
426 });
427 const kept = await store.ingest(tail, [], undefined, true);
428 assert.equal(kept.accepted, true);
429 assert.ok((kept.bytes ?? 0) < 100);
430 assert.equal(store.logs({}).items.length, 0);
431 assert.equal(store.usage().today_bytes, store.todayBytes());
432 // The workspace is remembered as the site says it.
433 assert.equal(store.workspaceOf(), null);
434 store.setWorkspace(" Acme ");
435 assert.equal(store.workspaceOf(), "acme");
436 store.setWorkspace("not a slug!");
437 assert.equal(store.workspaceOf(), "acme");
438});