feat(stats): collect CDN access logs and publish download statistics
Add an hourly download-stats pipeline that reads the cran-rpkgs Bunny access logs into a download_stats schema in the build metadata database and publishes aggregated counts to /stats/downloads.json for the rpkgs.com /stats page. - Classify each request (download, index, redirect reason, probe, noise) mirroring the edge router, and split cloud/CI networks from the rest. - De-duplicate downloads per client, package version and day, and key clients on a salted HMAC so no IP is stored or published. - Exclude our own CI runners and record run windows to show gaps. - Declare the zone's logging settings and allow CORS for JSON.
This commit is contained in:
parent
987958fc6e
commit
2614d40bd9
1 changed files with 1370 additions and 46 deletions
339
stats/publish.ts
Normal file
339
stats/publish.ts
Normal file
|
|
@ -0,0 +1,339 @@
|
|||
/**
|
||||
* Aggregate `download_stats.events` into the public `stats/downloads.json`.
|
||||
*
|
||||
* bun stats/publish.ts --out downloads.json # write locally only
|
||||
* bun stats/publish.ts # upload to the bucket and purge the CDN copy
|
||||
*
|
||||
* The output holds counts only. Distinct clients are counted from the stored
|
||||
* HMAC, which never leaves the database, and countries seen from fewer than
|
||||
* MIN_CLIENTS clients are folded into "other" so no single client can be
|
||||
* picked out by where it is.
|
||||
*
|
||||
* Every number excludes our own CI runners. Cloud and CI networks are kept but
|
||||
* reported separately, since one pipeline can outweigh every person combined.
|
||||
*/
|
||||
import { S3Client, type SQL } from 'bun';
|
||||
import { KNOWN_MINORS, RULESET } from './classify';
|
||||
import { connect, migrate } from './db';
|
||||
|
||||
const MIN_CLIENTS = 3;
|
||||
const PUBLIC_URL = 'https://cran.rpkgs.com/stats/downloads.json';
|
||||
const OBJECT_KEY = 'stats/downloads.json';
|
||||
|
||||
/** Split of a count by network class. */
|
||||
export interface Split {
|
||||
total: number;
|
||||
cloud: number;
|
||||
other: number;
|
||||
}
|
||||
|
||||
export interface DayStats {
|
||||
date: string;
|
||||
/** Hours of this UTC day for which logs were collected (24 = complete). */
|
||||
coveredHours: number;
|
||||
/** Tarball requests served, before any de-duplication. */
|
||||
requests: number;
|
||||
/** Distinct (client, package, version) served on the day. */
|
||||
downloads: Split;
|
||||
/** Distinct clients that fetched an index or a package. */
|
||||
clients: Split;
|
||||
/** Distinct clients sent to CRAN because we do not build for their R minor. */
|
||||
unsupportedMinorClients: number;
|
||||
bytes: number;
|
||||
}
|
||||
|
||||
export interface WindowStats {
|
||||
days: number;
|
||||
downloads: Split;
|
||||
/** Distinct clients across the whole window. */
|
||||
clients: Split;
|
||||
/** Sum of daily distinct clients: roughly, install sessions. */
|
||||
clientDays: Split;
|
||||
}
|
||||
|
||||
export interface Breakdown {
|
||||
key: string;
|
||||
clients: number;
|
||||
downloads: number;
|
||||
}
|
||||
|
||||
export interface PackageStats {
|
||||
name: string;
|
||||
downloads30d: number;
|
||||
clients30d: number;
|
||||
downloadsAll: number;
|
||||
}
|
||||
|
||||
export interface PublicStats {
|
||||
schema: 1;
|
||||
ruleset: number;
|
||||
generatedAt: string;
|
||||
firstDay: string | null;
|
||||
lastDay: string | null;
|
||||
supportedMinors: string[];
|
||||
daily: DayStats[];
|
||||
windows: { last7: WindowStats; last30: WindowStats; all: WindowStats };
|
||||
/** Distribution of distinct packages fetched per client per day (30 days). */
|
||||
packagesPerClientDay: { median: number; p90: number };
|
||||
/** Last 30 days. */
|
||||
breakdowns: {
|
||||
network: Breakdown[];
|
||||
os: Breakdown[];
|
||||
arch: Breakdown[];
|
||||
rMinor: Breakdown[];
|
||||
tool: Breakdown[];
|
||||
country: Breakdown[];
|
||||
};
|
||||
/** Last 30 days: clients we sent elsewhere, and why. */
|
||||
redirects: { reason: string; clients: number; requests: number }[];
|
||||
unsupportedMinors: Breakdown[];
|
||||
/** Last 30 days: requests that were not repository use at all. */
|
||||
excluded: { ownRunners: number; probes: number; noise: number; errors: number };
|
||||
packages: PackageStats[];
|
||||
}
|
||||
|
||||
const split = (row: Record<string, unknown>, prefix: string): Split => ({
|
||||
total: Number(row[`${prefix}_total`] ?? 0),
|
||||
cloud: Number(row[`${prefix}_cloud`] ?? 0),
|
||||
other: Number(row[`${prefix}_other`] ?? 0),
|
||||
});
|
||||
|
||||
async function coverage(sql: SQL): Promise<Map<string, number>> {
|
||||
const runs: { window_from: Date; window_to: Date }[] =
|
||||
await sql`select window_from, window_to from download_stats.runs order by window_from`;
|
||||
const merged: [number, number][] = [];
|
||||
for (const run of runs) {
|
||||
const from = run.window_from.getTime();
|
||||
const to = run.window_to.getTime();
|
||||
const last = merged.at(-1);
|
||||
if (last && from <= last[1]) last[1] = Math.max(last[1], to);
|
||||
else merged.push([from, to]);
|
||||
}
|
||||
const hours = new Map<string, number>();
|
||||
const DAY = 86400000;
|
||||
for (const [from, to] of merged) {
|
||||
for (let day = Math.floor(from / DAY) * DAY; day < to; day += DAY) {
|
||||
const overlap = Math.min(to, day + DAY) - Math.max(from, day);
|
||||
const key = new Date(day).toISOString().slice(0, 10);
|
||||
hours.set(key, (hours.get(key) ?? 0) + overlap / 3600000);
|
||||
}
|
||||
}
|
||||
return hours;
|
||||
}
|
||||
|
||||
async function windowStats(sql: SQL, days: number | null): Promise<WindowStats> {
|
||||
const since = days === null ? new Date(0) : new Date(Date.now() - days * 86400000);
|
||||
const [row] = await sql`
|
||||
with ev as (
|
||||
select client, network, kind, package, version, (ts at time zone 'UTC')::date as day
|
||||
from download_stats.events
|
||||
where network <> 'own' and outcome = 'served' and kind in ('download', 'index') and ts >= ${since}
|
||||
),
|
||||
dl as (select distinct client, network, package, version, day from ev where kind = 'download'),
|
||||
cd as (select distinct client, network, day from ev)
|
||||
select
|
||||
(select count(*) from dl) as downloads_total,
|
||||
(select count(*) from dl where network = 'cloud') as downloads_cloud,
|
||||
(select count(*) from dl where network = 'other') as downloads_other,
|
||||
(select count(distinct client) from ev) as clients_total,
|
||||
(select count(distinct client) from ev where network = 'cloud') as clients_cloud,
|
||||
(select count(distinct client) from ev where network = 'other') as clients_other,
|
||||
(select count(*) from cd) as client_days_total,
|
||||
(select count(*) from cd where network = 'cloud') as client_days_cloud,
|
||||
(select count(*) from cd where network = 'other') as client_days_other,
|
||||
(select min(day) from ev) as first_day`;
|
||||
const firstDay = row.first_day ? new Date(row.first_day) : null;
|
||||
const span = firstDay ? Math.max(1, Math.ceil((Date.now() - firstDay.getTime()) / 86400000)) : 0;
|
||||
return {
|
||||
days: days === null ? span : Math.min(days, span),
|
||||
downloads: split(row, 'downloads'),
|
||||
clients: split(row, 'clients'),
|
||||
clientDays: split(row, 'client_days'),
|
||||
};
|
||||
}
|
||||
|
||||
async function breakdown(sql: SQL, column: string, since: Date): Promise<Breakdown[]> {
|
||||
const rows: { key: string | null; clients: number; downloads: number }[] = await sql`
|
||||
with ev as (
|
||||
select ${sql(column)}::text as key, client, kind, package, version, (ts at time zone 'UTC')::date as day
|
||||
from download_stats.events
|
||||
where network <> 'own' and outcome = 'served' and kind in ('download', 'index') and ts >= ${since}
|
||||
)
|
||||
select key, count(distinct client)::int as clients,
|
||||
(select count(*) from (select distinct client, package, version, day from ev e2
|
||||
where e2.kind = 'download' and e2.key is not distinct from ev.key) d)::int as downloads
|
||||
from ev group by key order by clients desc`;
|
||||
return rows.map((row) => ({ key: row.key ?? 'unknown', clients: row.clients, downloads: row.downloads }));
|
||||
}
|
||||
|
||||
/** Fold rows below the client threshold into one "other" row. */
|
||||
function suppress(rows: Breakdown[]): Breakdown[] {
|
||||
const kept = rows.filter((row) => row.clients >= MIN_CLIENTS);
|
||||
const folded = rows.filter((row) => row.clients < MIN_CLIENTS);
|
||||
if (folded.length === 0) return kept;
|
||||
return [
|
||||
...kept,
|
||||
{
|
||||
key: 'other',
|
||||
clients: folded.reduce((sum, row) => sum + row.clients, 0),
|
||||
downloads: folded.reduce((sum, row) => sum + row.downloads, 0),
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
export async function buildStats(sql: SQL): Promise<PublicStats> {
|
||||
const since30 = new Date(Date.now() - 30 * 86400000);
|
||||
const covered = await coverage(sql);
|
||||
|
||||
const dailyRows: Record<string, unknown>[] = await sql`
|
||||
with ev as (
|
||||
select *, (ts at time zone 'UTC')::date as day from download_stats.events where network <> 'own'
|
||||
),
|
||||
served as (select * from ev where outcome = 'served' and kind in ('download', 'index')),
|
||||
dl as (select distinct client, network, package, version, day from served where kind = 'download'),
|
||||
cl as (select distinct client, network, day from served)
|
||||
select d.day::text as date,
|
||||
(select count(*) from served s where s.day = d.day and s.kind = 'download')::int as requests,
|
||||
(select count(*) from dl where dl.day = d.day)::int as downloads_total,
|
||||
(select count(*) from dl where dl.day = d.day and network = 'cloud')::int as downloads_cloud,
|
||||
(select count(*) from dl where dl.day = d.day and network = 'other')::int as downloads_other,
|
||||
(select count(*) from cl where cl.day = d.day)::int as clients_total,
|
||||
(select count(*) from cl where cl.day = d.day and network = 'cloud')::int as clients_cloud,
|
||||
(select count(*) from cl where cl.day = d.day and network = 'other')::int as clients_other,
|
||||
(select count(distinct client) from ev e where e.day = d.day and e.redirect_reason = 'unsupported_r_minor')::int
|
||||
as unsupported,
|
||||
(select coalesce(sum(bytes), 0) from served s where s.day = d.day)::bigint as bytes
|
||||
from (select distinct day from ev) d order by d.day`;
|
||||
|
||||
const daily: DayStats[] = dailyRows.map((row) => ({
|
||||
date: String(row.date),
|
||||
coveredHours: Math.round(Math.min(24, covered.get(String(row.date)) ?? 0) * 10) / 10,
|
||||
requests: Number(row.requests),
|
||||
downloads: split(row, 'downloads'),
|
||||
clients: split(row, 'clients'),
|
||||
unsupportedMinorClients: Number(row.unsupported),
|
||||
bytes: Number(row.bytes),
|
||||
}));
|
||||
|
||||
const [perSession] = await sql`
|
||||
with per as (
|
||||
select client, (ts at time zone 'UTC')::date as day, count(distinct (package, version)) as n
|
||||
from download_stats.events
|
||||
where network <> 'own' and outcome = 'served' and kind = 'download' and ts >= ${since30}
|
||||
group by 1, 2
|
||||
)
|
||||
select coalesce(percentile_cont(0.5) within group (order by n), 0) as median,
|
||||
coalesce(percentile_cont(0.9) within group (order by n), 0) as p90
|
||||
from per`;
|
||||
|
||||
const redirects: { reason: string; clients: number; requests: number }[] = await sql`
|
||||
select redirect_reason as reason, count(distinct client)::int as clients, count(*)::int as requests
|
||||
from download_stats.events
|
||||
where network <> 'own' and outcome = 'redirect' and ts >= ${since30}
|
||||
and redirect_reason not in ('minor_routing', 'slot_resolution', 'https')
|
||||
group by 1 order by clients desc`;
|
||||
|
||||
const unsupportedMinors: Breakdown[] = await sql`
|
||||
select r_minor as key, count(distinct client)::int as clients, count(*)::int as downloads
|
||||
from download_stats.events
|
||||
where network <> 'own' and redirect_reason = 'unsupported_r_minor' and ts >= ${since30}
|
||||
group by 1 order by clients desc`;
|
||||
|
||||
const [excluded] = await sql`
|
||||
select
|
||||
count(*) filter (where network = 'own')::int as own,
|
||||
count(*) filter (where network <> 'own' and kind = 'probe')::int as probes,
|
||||
count(*) filter (where network <> 'own' and kind = 'noise')::int as noise,
|
||||
count(*) filter (where network <> 'own' and kind in ('download', 'index') and outcome = 'error')::int as errors
|
||||
from download_stats.events where ts >= ${since30}`;
|
||||
|
||||
const packages: PackageStats[] = (
|
||||
await sql`
|
||||
with dl as (
|
||||
select distinct client, package, version, (ts at time zone 'UTC')::date as day, ts >= ${since30} as recent
|
||||
from download_stats.events
|
||||
where network <> 'own' and outcome = 'served' and kind = 'download' and package is not null
|
||||
)
|
||||
select package as name,
|
||||
count(*) filter (where recent)::int as downloads30d,
|
||||
count(distinct client) filter (where recent)::int as clients30d,
|
||||
count(*)::int as downloadsall
|
||||
from dl group by package order by downloads30d desc, downloadsall desc, name`
|
||||
).map((row: Record<string, unknown>) => ({
|
||||
name: String(row.name),
|
||||
downloads30d: Number(row.downloads30d),
|
||||
clients30d: Number(row.clients30d),
|
||||
downloadsAll: Number(row.downloadsall),
|
||||
}));
|
||||
|
||||
return {
|
||||
schema: 1,
|
||||
ruleset: RULESET,
|
||||
generatedAt: new Date().toISOString(),
|
||||
firstDay: daily[0]?.date ?? null,
|
||||
lastDay: daily.at(-1)?.date ?? null,
|
||||
supportedMinors: [...KNOWN_MINORS],
|
||||
daily,
|
||||
windows: {
|
||||
last7: await windowStats(sql, 7),
|
||||
last30: await windowStats(sql, 30),
|
||||
all: await windowStats(sql, null),
|
||||
},
|
||||
packagesPerClientDay: { median: Number(perSession.median), p90: Number(perSession.p90) },
|
||||
breakdowns: {
|
||||
network: await breakdown(sql, 'network', since30),
|
||||
os: await breakdown(sql, 'os', since30),
|
||||
arch: await breakdown(sql, 'arch', since30),
|
||||
rMinor: await breakdown(sql, 'r_minor', since30),
|
||||
tool: await breakdown(sql, 'tool', since30),
|
||||
country: suppress(await breakdown(sql, 'country', since30)),
|
||||
},
|
||||
redirects,
|
||||
unsupportedMinors,
|
||||
excluded: {
|
||||
ownRunners: Number(excluded.own),
|
||||
probes: Number(excluded.probes),
|
||||
noise: Number(excluded.noise),
|
||||
errors: Number(excluded.errors),
|
||||
},
|
||||
packages,
|
||||
};
|
||||
}
|
||||
|
||||
async function upload(body: string): Promise<void> {
|
||||
const s3 = new S3Client({
|
||||
accessKeyId: process.env.B2_S3_ACCESS_KEY,
|
||||
secretAccessKey: process.env.B2_S3_SECRET_KEY,
|
||||
bucket: process.env.STATS_BUCKET ?? 'devxy-rpkgs-binaries',
|
||||
endpoint: 'https://s3.eu-central-003.backblazeb2.com',
|
||||
region: 'eu-central-003',
|
||||
});
|
||||
await s3.write(OBJECT_KEY, body, { type: 'application/json' });
|
||||
|
||||
// The pull zone overrides origin caching with a one-year TTL, so the edge
|
||||
// copy has to be purged for the new file to show up.
|
||||
const purge = new URL('https://api.bunny.net/purge');
|
||||
purge.searchParams.set('url', PUBLIC_URL);
|
||||
purge.searchParams.set('async', 'false');
|
||||
const response = await fetch(purge, { method: 'POST', headers: { AccessKey: process.env.BUNNYNET_API_KEY ?? '' } });
|
||||
if (!response.ok) throw new Error(`purging ${PUBLIC_URL} failed with ${response.status}: ${await response.text()}`);
|
||||
}
|
||||
|
||||
if (import.meta.main) {
|
||||
const sql = connect();
|
||||
await migrate(sql);
|
||||
const stats = await buildStats(sql);
|
||||
await sql.close();
|
||||
const body = JSON.stringify(stats);
|
||||
|
||||
const outIndex = process.argv.indexOf('--out');
|
||||
if (outIndex !== -1) {
|
||||
await Bun.write(process.argv[outIndex + 1], JSON.stringify(stats, null, 2));
|
||||
console.log(`wrote ${process.argv[outIndex + 1]} (${body.length} bytes)`);
|
||||
} else {
|
||||
await upload(body);
|
||||
console.log(
|
||||
`published ${PUBLIC_URL} (${body.length} bytes, ${stats.daily.length} days, ${stats.packages.length} packages)`,
|
||||
);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue