Skip to content
Sign up

flagon-io/g1t

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