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
305 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Projects have Observability: their requests, logs, traces, metrics and errors, from apps on g1t.page with nothing to set up, and from software anywhere else over OpenTelemetry.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 const list = store.traces({});
212 assert.equal(list.items[0].trace_id, TRACE);
213 assert.equal(list.items[0].name, "GET /users/:id");
214 assert.equal(list.items[0].span_count, 2);
215});
216
217test("logs are searched by words, levels and attributes, and paged", async () => {
218 const store = new Store(state());
219 const batch = emptyBatch();
220 for (let i = 0; i < 30; i++) {
221 batch.logs.push({
222 id: `log${String(i).padStart(3, "0")}`,
223 time_us: (NOW_MS + i) * 1000,
224 severity: i % 3 === 0 ? "error" : "info",
225 severity_number: i % 3 === 0 ? 17 : 9,
226 body: i % 2 ? `payment captured for order ${i}` : `cache miss on key ${i}`,
227 trace_id: null,
228 span_id: null,
229 request_id: null,
230 attributes: { "user.id": `u${i % 5}` },
231 service: "api",
232 environment: "production",
233 deployment_id: null,
234 commit_sha: null,
235 source: "otlp",
236 });
237 }
238 const { secret } = await store.createToken("prj_1", "x", null);
239 await store.ingest(batch, [], secret);
240 assert.equal(store.logs({ text: "payment" }).items.length, 15);
241 assert.equal(store.logs({ text: "payment captured" }).items.length, 15);
242 assert.equal(store.logs({ text: "pay" }).items.length, 15, "the last word is a prefix");
243 assert.equal(store.logs({ text: "level:error" }).items.length, 10);
244 assert.equal(store.logs({ severities: ["error"], text: "cache" }).items.length, 5);
245 assert.equal(store.logs({ text: "user.id=u1" }).items.length, 6);
246 const first = store.logs({ limit: 20 });
247 assert.equal(first.items.length, 20);
248 const second = store.logs({ limit: 20, cursor: first.next });
249 assert.equal(second.items.length, 10);
250 assert.equal(second.next, null);
251 assert.equal(new Set([...first.items, ...second.items].map((l) => l.id)).size, 30);
252});
253
254test("tokens: the deployments' one is kept and the same each time; others can be revoked", async () => {
255 const store = new Store(state());
256 const a = await store.deploymentToken("prj_1");
257 const b = await store.deploymentToken("prj_1");
258 assert.equal(a.secret, b.secret);
259 assert.match(a.secret, /^g1to_prj_1_[0-9a-f]{48}$/);
260 const made = await store.createToken("prj_1", "laptop", "u1");
261 const listed = store.tokens();
262 assert.equal(listed.length, 2);
263 assert.ok(!JSON.stringify(listed).includes(made.secret), "a token's secret is never listed");
264 assert.equal(store.revokeToken(listed.find((t) => t.managed)!.id), false);
265 assert.equal(store.revokeToken(made.token.id), true);
266 assert.equal(await store.checkToken(made.secret), false);
267});
268
269test("retention removes what is past it, and turning deployments off stops tail events", async () => {
270 const store = new Store(state());
271 await ingestTail(store, [appEvent("fffffffffffffff1", { logs: ["old"] }), edgeEvent("fffffffffffffff1", 200, 5)]);
272 store.prune((NOW_MS + 8 * 86_400_000) * 1000);
273 assert.equal(store.requests({}).items.length, 0);
274 assert.equal(store.logs({}).items.length, 0);
275 // Counted before they went.
276 assert.equal(store.overview({ from_us: (NOW_MS - 60_000) * 1000, to_us: (NOW_MS + 60_000) * 1000 }).totals.requests, 1);
277 store.setSettings({ deployments: false });
278 const [g] = readTail([appEvent("fffffffffffffff2")]);
279 assert.equal((await store.ingest(g.batch, g.halves)).accepted, false);
280 assert.throws(() => store.setSettings({ retention_days: 365 }));
281});
282
283test("exports refuse addresses that loop or stay inside a network", () => {
284 const store = new Store(state());
285 assert.throws(() => store.putExport({ name: "x", endpoint: "https://otel.g1t.sh", signals: ["traces"], enabled: true }));
286 assert.throws(() => store.putExport({ name: "x", endpoint: "http://collector.example.com", signals: ["traces"], enabled: true }));
287 assert.throws(() => store.putExport({ name: "x", endpoint: "https://10.0.0.4", signals: ["traces"], enabled: true }));
288 const made = store.putExport({ name: "Honeycomb", endpoint: "https://api.honeycomb.io", signals: ["traces", "logs"], headers: { "x-honeycomb-team": "secret" }, enabled: true });
289 assert.deepEqual(made.header_names, ["x-honeycomb-team"]);
290 assert.ok(!JSON.stringify(store.exports()).includes("secret"));
291 // Changing it without giving the header again keeps it.
292 store.putExport({ id: made.id, name: "Honeycomb", endpoint: "https://api.honeycomb.io", signals: ["traces"], enabled: false });
293 assert.deepEqual(store.exports()[0].header_names, ["x-honeycomb-team"]);
294});
295
296test("search text, FTS queries and chart steps", () => {
297 assert.deepEqual(parseSearch('status:500 "two words" user.id=7 timeout https://x.y'), {
298 words: ["two words", "timeout", "https://x.y"],
299 pairs: [["status", "500"], ["user.id", "7"]],
300 });
301 assert.equal(ftsQuery(["a", 'b"c']), '"a" AND "b""c"*');
302 assert.equal(stepFor(3_600_000_000), 60);
303 assert.equal(stepFor(86_400_000_000), 900);
304 assert.equal(aggregate([]).length, 0);
305});