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
526 linesCodeBlameRaw
1#!/usr/bin/env node
2// What is the platform using on Cloudflare? Asks Cloudflare's GraphQL
3// Analytics API, for the month so far (UTC, from the 1st) and the last 24
4// hours, how much each Worker, D1 database, queue, Durable Object namespace,
5// KV namespace and Artifacts namespace did, and prints the top of each with
6// totals. Workers Logs are added when the account's schema has a dataset for
7// them.
8//
9// Read-only: one GraphQL query per dataset and window, all at once, and a few
10// REST listings to put names on database, queue and namespace ids. A dataset
11// or field Cloudflare refuses is a note in the report, never the end of it.
12//
13// CLOUDFLARE_API_TOKEN=<token with Account Analytics: Read> \
14// node scripts/ops/platform-usage.mjs [--top 10] [--json] [--account <id>]
15//
16// --account (or CLOUDFLARE_ACCOUNT_ID) reads another account than g1t's.
17// --json prints one object with snake_case keys and every row, not just the top.
18// Names come from the D1, Queues, KV and Durable Objects listings when the
19// token may read them (D1: Read, Queues: Read, Workers KV Storage: Read,
20// Workers Scripts: Read); otherwise the ids are printed.
21//
22// It is also the check of billing's hourly watcher (services/billing/src/
23// platform.rs): it asks Cloudflare billing's own queries, field for field,
24// over the last full hour, and exits 1 naming every dataset that errored,
25// fell back to fewer fields, or (all of them) answered with no rows. Run it
26// once after deploying a change to the watcher's queries.
27
28import { ACCOUNT_ID, cloudflareAuth } from "../deploy/cloudflare.mjs";
29
30const API = "https://api.cloudflare.com/client/v4";
31
32const HELP = `node scripts/ops/platform-usage.mjs [--top N] [--json] [--account <id>]
33
34Cloudflare usage per Worker, D1 database, queue, Durable Object namespace,
35KV namespace and Artifacts namespace, month to date (UTC) and the last 24 hours.
36
37 --top N rows shown per dataset in the tables (default 10)
38 --json one JSON object, snake_case keys, every row
39 --account <id> the Cloudflare account (default CLOUDFLARE_ACCOUNT_ID, else g1t's)
40 --help this text
41
42Needs CLOUDFLARE_API_TOKEN (Account Analytics: Read), or CLOUDFLARE_API_KEY
43with CLOUDFLARE_EMAIL.
44
45Exits 1 when any dataset errored or fell back to fewer fields, when any of
46billing's watcher queries errored, or when every dataset answered with no
47rows (the wrong account, or a token that cannot see it).`;
48
49/**
50 * Billing's hourly watcher queries, as `QUERIES` in
51 * services/billing/src/platform.rs has them (a test keeps the two the same):
52 * the dataset, what it selects, what it groups by, and the filter field for
53 * an hour.
54 */
55export const WATCHER_QUERIES = [
56 { key: "workers", dataset: "workersInvocationsAdaptive", select: "sum { requests }", dimensions: "scriptName", hourFilter: "datetime" },
57 { key: "workers_cpu", dataset: "workersInvocationsAdaptive", select: "sum { cpuTimeUs }", dimensions: "scriptName", hourFilter: "datetime" },
58 { key: "d1", dataset: "d1AnalyticsAdaptiveGroups", select: "sum { rowsRead rowsWritten }", dimensions: "databaseId", hourFilter: "datetimeHour" },
59 { key: "queues", dataset: "queueMessageOperationsAdaptiveGroups", select: "sum { billableOperations }", dimensions: "queueId", hourFilter: "datetime" },
60 { key: "do_invocations", dataset: "durableObjectsInvocationsAdaptiveGroups", select: "sum { requests }", dimensions: "scriptName", hourFilter: "datetime" },
61 {
62 key: "do_periodic",
63 dataset: "durableObjectsPeriodicGroups",
64 select: "sum { activeTime storageWriteUnits }",
65 dimensions: "namespaceId",
66 hourFilter: "datetime",
67 },
68 { key: "do_sql", dataset: "durableObjectsPeriodicGroups", select: "sum { rowsWritten }", dimensions: "namespaceId", hourFilter: "datetime" },
69 { key: "kv", dataset: "kvOperationsAdaptiveGroups", select: "sum { requests }", dimensions: "namespaceId actionType", hourFilter: "datetime" },
70 { key: "artifacts", dataset: "artifactsEventsAdaptiveGroups", select: "count", dimensions: "repositoryName", hourFilter: "datetime" },
71];
72
73/** The last full hour (UTC) before `now`, as the watcher reads it. */
74export function lastFullHour(now) {
75 const until = new Date(Math.floor(now.getTime() / 3_600_000) * 3_600_000);
76 const since = new Date(until.getTime() - 3_600_000);
77 const label = (date) => `${date.toISOString().slice(0, 13)}:00:00Z`;
78 return { since: label(since), until: label(until) };
79}
80
81/** One watcher query over an hour, as billing sends it (its account variable named as this script names it). */
82export function watcherQuery(query) {
83 return `query ($accountTag: String!, $since: Time!, $until: Time!) {
84 viewer { accounts(filter: { accountTag: $accountTag }) {
85 rows: ${query.dataset}(limit: 10000, filter: { ${query.hourFilter}_geq: $since, ${query.hourFilter}_lt: $until }) {
86 ${query.select}
87 dimensions { ${query.dimensions} }
88 }
89 } }
90}`;
91}
92
93/**
94 * Whether the schema checks out: every problem in the report's datasets and
95 * in the watcher's queries (`watcher`: key to `{ error }` or `{ rows }`).
96 * An empty list is a pass.
97 */
98export function verdict(report, watcher = {}) {
99 const problems = [];
100 for (const window of report.windows) {
101 for (const error of window.errors ?? []) problems.push(`${window.label}: ${error}`);
102 }
103 let answered = 0;
104 let rows = 0;
105 for (const query of WATCHER_QUERIES) {
106 const outcome = watcher[query.key];
107 if (!outcome) continue;
108 if (outcome.error) problems.push(`watcher query ${query.key} (${query.dataset}): ${outcome.error}`);
109 else {
110 answered += 1;
111 rows += outcome.rows;
112 }
113 }
114 const reportRows = report.windows.flatMap((w) => w.datasets).reduce((sum, set) => sum + set.rows.length, 0);
115 const reportAnswered = report.windows.flatMap((w) => w.datasets).length;
116 if (answered + reportAnswered > 0 && rows + reportRows === 0) {
117 problems.push("every dataset answered with no rows: likely the wrong account, or a token that cannot see its analytics");
118 }
119 return problems;
120}
121
122/**
123 * The datasets the report asks for. Each has one or more variants, tried in
124 * order: when Cloudflare refuses a field, the next variant asks for less.
125 * `names` says which REST listing puts a name on the first dimension's ids.
126 */
127export const DATASETS = [
128 {
129 key: "workers",
130 label: "Workers invocations, by script",
131 dataset: "workersInvocationsAdaptive",
132 variants: [
133 { sum: ["requests", "errors", "cpuTimeUs"], dims: ["scriptName"] },
134 { sum: ["requests", "errors"], dims: ["scriptName"] },
135 ],
136 },
137 {
138 key: "d1",
139 label: "D1 rows, by database",
140 dataset: "d1AnalyticsAdaptiveGroups",
141 names: "d1",
142 variants: [
143 { sum: ["rowsRead", "rowsWritten", "readQueries", "writeQueries"], dims: ["databaseId"] },
144 { sum: ["rowsRead", "rowsWritten"], dims: ["databaseId"] },
145 ],
146 },
147 {
148 key: "queues",
149 label: "Queue operations, by queue",
150 dataset: "queueMessageOperationsAdaptiveGroups",
151 names: "queues",
152 variants: [{ sum: ["billableOperations"], dims: ["queueId"] }],
153 },
154 {
155 key: "durable_objects",
156 label: "Durable Object requests, by script",
157 dataset: "durableObjectsInvocationsAdaptiveGroups",
158 variants: [
159 { sum: ["requests", "errors"], dims: ["scriptName"] },
160 { sum: ["requests"], dims: ["scriptName"] },
161 ],
162 },
163 {
164 key: "durable_objects_periodic",
165 label: "Durable Object time and storage, by namespace",
166 dataset: "durableObjectsPeriodicGroups",
167 names: "durable_objects",
168 variants: [
169 { sum: ["activeTime", "cpuTime", "storageReadUnits", "storageWriteUnits"], dims: ["namespaceId"] },
170 { sum: ["activeTime", "storageWriteUnits"], dims: ["namespaceId"] },
171 // Periodic groups may filter by the minute rather than by datetime.
172 { sum: ["activeTime"], dims: ["namespaceId"], time: "datetimeMinute" },
173 ],
174 },
175 {
176 key: "kv",
177 label: "KV operations, by namespace and action",
178 dataset: "kvOperationsAdaptiveGroups",
179 names: "kv",
180 variants: [{ sum: ["requests"], dims: ["namespaceId", "actionType"] }],
181 },
182 {
183 key: "artifacts",
184 label: "Artifacts events, by namespace and type",
185 dataset: "artifactsEventsAdaptiveGroups",
186 variants: [{ count: true, sum: ["durationMs"], dims: ["repositoryNamespace", "eventType"] }],
187 },
188 {
189 key: "workers_logs",
190 label: "Workers Logs events, by script",
191 // Which dataset holds Workers Logs is read from the schema (logsDataset).
192 dataset: null,
193 optional: true,
194 variants: [
195 { count: true, dims: ["scriptName"] },
196 { count: true, dims: [] },
197 ],
198 },
199];
200
201/** The two windows: the UTC month so far, and the last 24 hours. */
202export function windows(now = new Date()) {
203 const end = new Date(now.getTime());
204 const monthStart = new Date(Date.UTC(end.getUTCFullYear(), end.getUTCMonth(), 1));
205 return [
206 { key: "month_to_date", label: "Month to date (UTC)", start: monthStart.toISOString(), end: end.toISOString() },
207 { key: "last_24h", label: "Last 24 hours", start: new Date(end.getTime() - 24 * 3600 * 1000).toISOString(), end: end.toISOString() },
208 ];
209}
210
211/** Which account field holds Workers Logs, from the account type's field names; null when none does. */
212export function logsDataset(fieldNames) {
213 const known = ["workersObservabilityEventsAdaptiveGroups", "workersLogsEventsAdaptiveGroups", "workersLogsAdaptiveGroups"];
214 return known.find((name) => fieldNames.includes(name)) ?? fieldNames.find((name) => /^workers.*(logs|observability).*groups$/i.test(name)) ?? null;
215}
216
217/** One dataset's query, for one variant. The rows come back under `rows`. */
218export function buildQuery(dataset, variant) {
219 const time = variant.time ?? "datetime";
220 const fields = [
221 variant.count ? "count" : "",
222 variant.sum?.length ? `sum { ${variant.sum.join(" ")} }` : "",
223 variant.dims.length ? `dimensions { ${variant.dims.join(" ")} }` : "",
224 ].filter(Boolean);
225 return `query PlatformUsage($accountTag: String!, $start: Time!, $end: Time!) {
226 viewer {
227 accounts(filter: { accountTag: $accountTag }) {
228 rows: ${dataset}(limit: 10000, filter: { ${time}_geq: $start, ${time}_leq: $end }) {
229 ${fields.join("\n ")}
230 }
231 }
232 }
233}`;
234}
235
236/** camelCase to snake_case, for the JSON report's keys. */
237export const snake = (name) => name.replace(/[A-Z]/g, (letter) => `_${letter.toLowerCase()}`);
238
239/** The metric names a variant reports, snake_case: `count` first, then its sums. */
240export const metricsOf = (variant) => [...(variant.count ? ["count"] : []), ...(variant.sum ?? []).map(snake)];
241
242/**
243 * A GraphQL answer to one dataset's query as rows, one per name, summed and
244 * sorted by the first metric (largest first), with totals per metric. Throws
245 * with Cloudflare's message when the answer has errors or no such dataset.
246 */
247export function parseGroups(body, variant, names = {}) {
248 if (body?.errors?.length)
249 throw new Error(
250 body.errors
251 .map((error) => error.message)
252 .join("; ")
253 .slice(0, 400),
254 );
255 const groups = body?.data?.viewer?.accounts?.[0]?.rows;
256 if (!Array.isArray(groups)) throw new Error("no rows in the answer");
257 const metrics = metricsOf(variant);
258 const byName = new Map();
259 for (const group of groups) {
260 const parts = variant.dims.map((dim, at) => {
261 const value = group.dimensions?.[dim];
262 const text = value == null || value === "" ? "(none)" : String(value);
263 return at === 0 ? (names[text] ?? text) : text;
264 });
265 const name = parts.join(" / ") || "(all)";
266 const row = byName.get(name) ?? { name, ...Object.fromEntries(metrics.map((metric) => [metric, 0])) };
267 if (variant.count) row.count += Number(group.count ?? 0);
268 for (const field of variant.sum ?? []) row[snake(field)] += Number(group.sum?.[field] ?? 0);
269 byName.set(name, row);
270 }
271 const rows = sortRows([...byName.values()], metrics[0]);
272 return { metrics, rows, totals: totalsOf(rows, metrics) };
273}
274
275/** Rows largest first by one metric, ties by name. */
276export function sortRows(rows, metric) {
277 return [...rows].sort((a, b) => (b[metric] ?? 0) - (a[metric] ?? 0) || a.name.localeCompare(b.name));
278}
279
280/** Each metric summed over all rows. */
281export function totalsOf(rows, metrics) {
282 return Object.fromEntries(metrics.map((metric) => [metric, rows.reduce((total, row) => total + (row[metric] ?? 0), 0)]));
283}
284
285/**
286 * The report from every dataset's outcome in every window. An outcome is
287 * `{ body, variant }` (a GraphQL answer) or `{ error }` or `{ skipped }`;
288 * whatever cannot be read becomes a note and the other datasets still count.
289 */
290export function assemble({ account, now, windowList, outcomes, names = {} }) {
291 return {
292 account,
293 generated_at: now.toISOString(),
294 windows: windowList.map((window) => {
295 const notes = [];
296 const errors = [];
297 const datasets = [];
298 for (const spec of DATASETS) {
299 const outcome = outcomes[window.key]?.[spec.key];
300 if (!outcome) continue;
301 if (outcome.skipped) {
302 notes.push(`${spec.label}: ${outcome.skipped}`);
303 continue;
304 }
305 if (outcome.fellBack) {
306 const note = `${spec.label} (${outcome.dataset ?? spec.dataset}): fell back to fewer fields: ${outcome.fellBack}`;
307 notes.push(note);
308 errors.push(note);
309 }
310 try {
311 if (outcome.error) throw outcome.error;
312 const parsed = parseGroups(outcome.body, outcome.variant, names[spec.names] ?? {});
313 datasets.push({ key: spec.key, label: spec.label, dataset: outcome.dataset ?? spec.dataset, ...parsed });
314 } catch (error) {
315 const note = `${spec.label} (${outcome.dataset ?? spec.dataset ?? "no dataset"}): ${String(error.message ?? error).split("\n")[0]}`;
316 notes.push(note);
317 errors.push(note);
318 }
319 }
320 return { key: window.key, label: window.label, start: window.start, end: window.end, datasets, notes, errors };
321 }),
322 };
323}
324
325const number = (value) => (Number.isInteger(value) ? value.toLocaleString("en-US") : value.toLocaleString("en-US", { maximumFractionDigits: 2 }));
326
327/** The report as plain tables: the top rows of each dataset, and a totals line. */
328export function format(report, top = 10) {
329 const lines = [`Cloudflare usage for account ${report.account}, ${report.generated_at}`];
330 for (const window of report.windows) {
331 lines.push("", `== ${window.label}: ${window.start} to ${window.end}`);
332 for (const set of window.datasets) {
333 lines.push("", `${set.label} (${set.dataset})`);
334 const shown = set.rows.slice(0, top);
335 const cells = [
336 ["name", ...set.metrics],
337 ...shown.map((row) => [row.name, ...set.metrics.map((m) => number(row[m]))]),
338 ["total", ...set.metrics.map((m) => number(set.totals[m]))],
339 ];
340 const widths = cells[0].map((_, at) => Math.max(...cells.map((row) => String(row[at]).length)));
341 const render = (row) => " " + row.map((cell, at) => (at === 0 ? String(cell).padEnd(widths[at]) : String(cell).padStart(widths[at]))).join(" ");
342 lines.push(render(cells[0]));
343 if (!shown.length) lines.push(" (nothing in this window)");
344 for (const row of cells.slice(1, -1)) lines.push(render(row));
345 if (set.rows.length > shown.length) lines.push(` ... and ${set.rows.length - shown.length} more`);
346 lines.push(render(cells.at(-1)));
347 }
348 if (window.notes.length) {
349 lines.push("", "Notes:");
350 for (const note of window.notes) lines.push(` ${note}`);
351 }
352 }
353 return lines.join("\n");
354}
355
356/** The command line: flags and the account. */
357export function parseArgs(argv, env = process.env) {
358 const option = (name) => {
359 const at = argv.indexOf(name);
360 return at >= 0 && argv[at + 1] && !argv[at + 1].startsWith("--") ? argv[at + 1] : null;
361 };
362 return {
363 help: argv.includes("--help") || argv.includes("-h"),
364 json: argv.includes("--json"),
365 top: Math.max(1, Number(option("--top")) || 10),
366 account: option("--account") || env.CLOUDFLARE_ACCOUNT_ID || ACCOUNT_ID,
367 };
368}
369
370async function graphql(auth, account, query, variables = {}) {
371 const response = await fetch(`${API}/graphql`, {
372 method: "POST",
373 headers: { ...auth, "content-type": "application/json", "user-agent": "g1t-ops" },
374 body: JSON.stringify({ query, variables: { accountTag: account, ...variables } }),
375 });
376 const body = await response.json().catch(() => ({ errors: [{ message: `HTTP ${response.status}, not JSON` }] }));
377 if (!response.ok && !body.errors?.length) body.errors = [{ message: `HTTP ${response.status}` }];
378 return body;
379}
380
381/** The account type's field names, to skip datasets the schema lacks; null when it cannot be read. */
382async function schemaFields(auth, account) {
383 try {
384 const body = await graphql(
385 auth,
386 account,
387 `
388 {
389 __type(name: "account") {
390 fields {
391 name
392 }
393 }
394 }
395 `,
396 );
397 const fields = body.data?.__type?.fields;
398 return Array.isArray(fields) ? fields.map((field) => field.name) : null;
399 } catch {
400 return null;
401 }
402}
403
404/** One dataset in one window: each variant in turn until one is answered. */
405async function ask(auth, account, dataset, spec, window) {
406 let last = null;
407 let first = null;
408 for (const variant of spec.variants) {
409 try {
410 const body = await graphql(auth, account, buildQuery(dataset, variant), { start: window.start, end: window.end });
411 if (!body.errors?.length) return first ? { body, variant, dataset, fellBack: first } : { body, variant, dataset };
412 first ??= body.errors.map((e) => e.message).join("; ");
413 last = { body, variant, dataset };
414 } catch (error) {
415 last = { error, dataset };
416 }
417 }
418 return last;
419}
420
421/** A REST listing as id to name; empty when the token may not read it. */
422async function listing(auth, path, id, name) {
423 const out = {};
424 try {
425 for (let page = 1; page <= 10; page++) {
426 const response = await fetch(`${API}${path}${path.includes("?") ? "&" : "?"}per_page=100&page=${page}`, {
427 headers: { ...auth, "user-agent": "g1t-ops" },
428 });
429 const body = await response.json();
430 if (!response.ok || !Array.isArray(body.result)) break;
431 for (const item of body.result) if (item[id]) out[item[id]] = item[name] ?? item[id];
432 if (body.result.length < 100) break;
433 }
434 } catch {
435 // Ids stand in for names.
436 }
437 return out;
438}
439
440async function main() {
441 const options = parseArgs(process.argv.slice(2));
442 if (options.help) {
443 console.log(HELP);
444 return 0;
445 }
446 const auth = cloudflareAuth();
447 if (!auth) {
448 console.error(
449 `Set CLOUDFLARE_API_TOKEN to a token with Account Analytics: Read on account ${options.account}, or CLOUDFLARE_API_KEY and CLOUDFLARE_EMAIL.`,
450 );
451 return 2;
452 }
453 const now = new Date();
454 const windowList = windows(now);
455 const account = options.account;
456 const base = `/accounts/${account}`;
457 const [fields, d1, queues, kv, durable] = await Promise.all([
458 schemaFields(auth, account),
459 listing(auth, `${base}/d1/database`, "uuid", "name"),
460 listing(auth, `${base}/queues`, "queue_id", "queue_name"),
461 listing(auth, `${base}/storage/kv/namespaces`, "id", "title"),
462 listing(auth, `${base}/workers/durable_objects/namespaces`, "id", "name"),
463 ]);
464 const names = { d1, queues, kv, durable_objects: durable };
465 const outcomes = {};
466 const jobs = [];
467 for (const window of windowList) {
468 outcomes[window.key] = {};
469 for (const spec of DATASETS) {
470 const dataset = spec.dataset ?? (fields ? logsDataset(fields) : null);
471 if (!dataset) {
472 if (!spec.optional) outcomes[window.key][spec.key] = { skipped: "no dataset" };
473 continue;
474 }
475 if (fields && !fields.includes(dataset)) {
476 if (!spec.optional) outcomes[window.key][spec.key] = { skipped: `${dataset} is not in this account's schema` };
477 continue;
478 }
479 jobs.push(ask(auth, account, dataset, spec, window).then((outcome) => (outcomes[window.key][spec.key] = outcome)));
480 }
481 }
482 // Billing's own watcher queries, over the last full hour.
483 const hour = lastFullHour(now);
484 const watcher = {};
485 jobs.push(
486 ...WATCHER_QUERIES.map(async (query) => {
487 try {
488 const body = await graphql(auth, account, watcherQuery(query), { since: hour.since, until: hour.until });
489 watcher[query.key] = body.errors?.length
490 ? { error: body.errors.map((e) => e.message).join("; ") }
491 : { rows: body.data?.viewer?.accounts?.[0]?.rows?.length ?? 0 };
492 } catch (error) {
493 watcher[query.key] = { error: String(error.message ?? error) };
494 }
495 }),
496 );
497 await Promise.all(jobs);
498 const report = assemble({ account, now, windowList, outcomes, names });
499 const problems = verdict(report, watcher);
500 if (options.json) {
501 console.log(JSON.stringify({ ...report, watcher_hour: hour, watcher, problems }, null, 2));
502 } else {
503 console.log(format(report, options.top));
504 console.log("", `== Billing's watcher queries, ${hour.since} to ${hour.until}`);
505 for (const query of WATCHER_QUERIES) {
506 const outcome = watcher[query.key];
507 console.log(` ${query.key.padEnd(15)} ${outcome?.error ? `ERROR ${outcome.error}` : `${outcome?.rows ?? 0} rows`}`);
508 }
509 }
510 if (problems.length) {
511 console.error("", `platform-usage: ${problems.length} problem(s); the watcher may be blind on these:`);
512 for (const problem of problems) console.error(` ${problem}`);
513 return 1;
514 }
515 return 0;
516}
517
518if (process.argv[1]?.replaceAll("\\", "/").endsWith("scripts/ops/platform-usage.mjs")) {
519 main().then(
520 (code) => process.exit(code),
521 (error) => {
522 console.error(`platform-usage: ${error.message}`);
523 process.exit(1);
524 },
525 );
526}