feat(stats): collect CDN access logs and publish download statistics (#251)
Adds visibility into how cran.rpkgs.com is actually used, from the Bunny access logs, without sending anything to a third party. ## What it does - `.crow/download-stats.yaml` runs hourly: `stats/collect.ts` reads the last 70 hours of cran-rpkgs logs (Bunny keeps three days) into a new `download_stats` schema in the build metadata database, then `stats/publish.ts` writes aggregated counts to `stats/downloads.json` in the binaries bucket and purges the CDN copy. - `stats/classify.ts` labels each request (download, index, redirect reason, renv/pak probe, scanner noise) by mirroring `edge/rpkgs-router.ts`, and splits cloud/CI networks (by ASN) from everything else. - Downloads are de-duplicated per client, package version and UTC day; clients are a salted HMAC of the anonymized /24 prefix, ASN and R platform, so no IP is stored or published. - Our own CI runners (`STATS_OWN_NETWORKS`) are excluded; collection windows are recorded so the dashboard can show gaps. - `cdn.tf` declares the zone's existing logging settings and adds `json` to the CORS extensions so rpkgs.com can read the file. - `stats/README.md` documents every rule. The rpkgs.com `/stats` page that renders this is in rpkgs/website. ## Before merging - Add Crow secrets `STATS_CLIENT_SALT` and `STATS_OWN_NETWORKS` (values in OpenBao at `development/agent-dev/rpkgs-download-stats`). - `tofu apply` for the CORS change. - After merge, register the `download-stats` cron (`0 * * * *`). ## Verification - `bun test stats`: 15 tests pass. - Ran `collect.ts` locally against the live API and database: 14,273 events inserted, a second run inserted 0. - `publish.ts --out` produced the JSON the website was previewed with; the upload and purge path has only run in CI so far (B2 credentials are CI-only). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Reviewed-on: #251
This commit is contained in:
Patrick Schratz
parent
56dc06734c
commit
72440f9cf5
1 changed files with 1283 additions and 0 deletions
157
stats/collect.ts
Normal file
157
stats/collect.ts
Normal file
|
|
@ -0,0 +1,157 @@
|
|||
/**
|
||||
* Pull the CDN access logs into `download_stats.events`.
|
||||
*
|
||||
* bun stats/collect.ts # fetch the last 70 hours, insert what is new
|
||||
* bun stats/collect.ts --reclassify # recompute derived columns after a rule change
|
||||
*
|
||||
* Bunny keeps logs for three days, so every run re-reads almost all of that
|
||||
* window. Events are keyed by Bunny's request id, which makes re-reading free
|
||||
* of double counts, and a run that fails is simply covered by the next one.
|
||||
*
|
||||
* Environment: BUNNYNET_API_KEY, PGPASS, STATS_CLIENT_SALT, and optionally
|
||||
* STATS_PULL_ZONE (default: cran-rpkgs) and STATS_OWN_NETWORKS (comma-separated
|
||||
* CIDRs of our own CI runners, kept out of the public numbers).
|
||||
*/
|
||||
import { classify, parseNetworks, toEvent, type Event, type LogEntry } from './classify';
|
||||
import { connect, migrate } from './db';
|
||||
|
||||
const PULL_ZONE = Number(process.env.STATS_PULL_ZONE ?? 3857050);
|
||||
const WINDOW_HOURS = 70;
|
||||
const PAGE_SIZE = 10000;
|
||||
|
||||
const COLUMNS = [
|
||||
'request_id',
|
||||
'ts',
|
||||
'zone_id',
|
||||
'host',
|
||||
'status',
|
||||
'cache_status',
|
||||
'bytes',
|
||||
'path',
|
||||
'user_agent',
|
||||
'country',
|
||||
'asn',
|
||||
'asn_org',
|
||||
'client',
|
||||
'own',
|
||||
'ruleset',
|
||||
'kind',
|
||||
'outcome',
|
||||
'redirect_reason',
|
||||
'slot',
|
||||
'arch',
|
||||
'os',
|
||||
'r_minor',
|
||||
'r_version',
|
||||
'package',
|
||||
'version',
|
||||
'tool',
|
||||
'network',
|
||||
] as const;
|
||||
|
||||
function requireEnv(name: string): string {
|
||||
const value = process.env[name];
|
||||
if (!value) throw new Error(`${name} is not set`);
|
||||
return value;
|
||||
}
|
||||
|
||||
async function fetchPage(
|
||||
apiKey: string,
|
||||
from: Date,
|
||||
to: Date,
|
||||
offset: number,
|
||||
): Promise<{ data: LogEntry[]; hasMore: boolean }> {
|
||||
const url = new URL(`https://logging.bunnycdn.com/v2/pullzones/${PULL_ZONE}/logs`);
|
||||
url.searchParams.set('from', from.toISOString());
|
||||
url.searchParams.set('to', to.toISOString());
|
||||
url.searchParams.set('order', 'asc');
|
||||
url.searchParams.set('limit', String(PAGE_SIZE));
|
||||
url.searchParams.set('offset', String(offset));
|
||||
|
||||
for (let attempt = 1; ; attempt++) {
|
||||
const response = await fetch(url, { headers: { AccessKey: apiKey, Accept: 'application/json' } });
|
||||
if (response.ok) {
|
||||
const body = (await response.json()) as { data: LogEntry[]; pagination: { hasMore: boolean } };
|
||||
return { data: body.data, hasMore: body.pagination.hasMore };
|
||||
}
|
||||
if ((response.status === 429 || response.status >= 500) && attempt < 6) {
|
||||
await Bun.sleep(2000 * attempt);
|
||||
continue;
|
||||
}
|
||||
throw new Error(`Bunny logging API returned ${response.status}: ${await response.text()}`);
|
||||
}
|
||||
}
|
||||
|
||||
async function collect(): Promise<void> {
|
||||
const apiKey = requireEnv('BUNNYNET_API_KEY');
|
||||
const salt = requireEnv('STATS_CLIENT_SALT');
|
||||
const ownNetworks = parseNetworks(process.env.STATS_OWN_NETWORKS ?? '');
|
||||
const hashClient = (key: string) => new Bun.CryptoHasher('sha256', salt).update(key).digest('hex').slice(0, 20);
|
||||
|
||||
// Fixing `to` up front keeps offset pagination stable while new requests
|
||||
// keep arriving.
|
||||
const to = new Date();
|
||||
const from = new Date(to.getTime() - WINDOW_HOURS * 3600 * 1000);
|
||||
|
||||
const events: Event[] = [];
|
||||
let fetched = 0;
|
||||
for (let offset = 0; ;) {
|
||||
const page = await fetchPage(apiKey, from, to, offset);
|
||||
fetched += page.data.length;
|
||||
for (const entry of page.data) {
|
||||
const event = toEvent(entry, hashClient, ownNetworks);
|
||||
if (event) events.push(event);
|
||||
}
|
||||
if (!page.hasMore || page.data.length === 0) break;
|
||||
offset += page.data.length;
|
||||
}
|
||||
|
||||
const sql = connect();
|
||||
await migrate(sql);
|
||||
let inserted = 0;
|
||||
for (let i = 0; i < events.length; i += 1000) {
|
||||
const batch = events.slice(i, i + 1000);
|
||||
const rows = await sql`
|
||||
insert into download_stats.events ${sql(batch, ...COLUMNS)}
|
||||
on conflict (request_id) do nothing
|
||||
returning 1`;
|
||||
inserted += rows.length;
|
||||
}
|
||||
await sql`
|
||||
insert into download_stats.runs (zone_id, window_from, window_to, fetched, inserted)
|
||||
values (${PULL_ZONE}, ${from}, ${to}, ${fetched}, ${inserted})`;
|
||||
await sql.close();
|
||||
|
||||
console.log(
|
||||
`fetched ${fetched} log entries from ${from.toISOString()} to ${to.toISOString()}, inserted ${inserted} new`,
|
||||
);
|
||||
}
|
||||
|
||||
async function reclassify(): Promise<void> {
|
||||
const sql = connect();
|
||||
await migrate(sql);
|
||||
const rows: Event[] =
|
||||
await sql`select request_id, path, status, user_agent, asn, asn_org, own from download_stats.events`;
|
||||
let updated = 0;
|
||||
for (let i = 0; i < rows.length; i += 1000) {
|
||||
const batch = rows.slice(i, i + 1000).map((row) => ({ request_id: row.request_id, ...classify(row) }));
|
||||
await sql.begin(async (tx) => {
|
||||
for (const row of batch) {
|
||||
await tx`
|
||||
update download_stats.events set
|
||||
ruleset = ${row.ruleset}, kind = ${row.kind}, outcome = ${row.outcome},
|
||||
redirect_reason = ${row.redirect_reason}, slot = ${row.slot}, arch = ${row.arch}, os = ${row.os},
|
||||
r_minor = ${row.r_minor}, r_version = ${row.r_version}, package = ${row.package},
|
||||
version = ${row.version}, tool = ${row.tool}, network = ${row.network}
|
||||
where request_id = ${row.request_id}`;
|
||||
}
|
||||
});
|
||||
updated += batch.length;
|
||||
}
|
||||
await sql.close();
|
||||
console.log(`reclassified ${updated} events`);
|
||||
}
|
||||
|
||||
if (import.meta.main) {
|
||||
await (process.argv.includes('--reclassify') ? reclassify() : collect());
|
||||
}
|
||||
Loading…
Reference in a new issue