Daan MeijerandClaude Fable 5.1
c18c320927
The sampling ranges in other years are capture too, at history priorities: they made a band each. The database size walked the HDD partitions for minutes and the probe counts took 10 s on 11 M rows: dropped, and the partial indexes' reltuples instead. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
337 lines
14 KiB
TypeScript
337 lines
14 KiB
TypeScript
import db from '@adonisjs/lucid/services/db'
|
||
import { readdir, stat, statfs } from 'node:fs/promises'
|
||
import { join } from 'node:path'
|
||
import coordinatorConfig from '#config/coordinator'
|
||
import GroupLeases from '#services/group_leases'
|
||
import type { Lane } from '#lib/edge/index'
|
||
|
||
export interface RangeBand {
|
||
band: string
|
||
planned: number
|
||
plannedArticles: number
|
||
leased: number
|
||
done: number
|
||
doneArticles: number
|
||
failed: number
|
||
}
|
||
|
||
export interface HeldGroup {
|
||
node: string
|
||
group: string
|
||
lane: Lane
|
||
range: { start: number; end: number; next: number } | null
|
||
heartbeatSeconds: number
|
||
snapshotNext: number | null
|
||
snapshotSeconds: number | null
|
||
snapshotMb: number | null
|
||
rangesDone: number
|
||
headers: number
|
||
}
|
||
|
||
export interface SnapshotFile {
|
||
group: string
|
||
lane: Lane
|
||
mb: number
|
||
hours: number
|
||
}
|
||
|
||
export interface Status {
|
||
at: string
|
||
catalog: {
|
||
tablespace: string
|
||
gb: number
|
||
capGb: number
|
||
full: boolean
|
||
freeGb: number
|
||
}
|
||
bands: RangeBand[]
|
||
lastHour: {
|
||
indexedArticles: number
|
||
doneRanges: number
|
||
storedArticles: number
|
||
storedRanges: number
|
||
}
|
||
corpus: { pendingRanges: number; pendingArticles: number }
|
||
spool: { ranges: number; chunks: number }
|
||
tail: { groups: number; ranges: number; articles: number }
|
||
held: HeldGroup[]
|
||
snapshots: { count: number; gb: number; oldestHours: number | null; largest: SnapshotFile[] }
|
||
releases: { estimated: number; provisional: number }
|
||
/** `ready` and `waiting` are the planner's estimates (the last analyze), the others counts. */
|
||
probes: { ready: number; waiting: number; pendingWaiting: number; pendingNamed: number }
|
||
judge: { dirty: number }
|
||
nodes: { name: string; lastSeenSeconds: number | null }[]
|
||
}
|
||
|
||
const seconds = (since: Date | string | null) =>
|
||
since ? Math.round((Date.now() - new Date(since).getTime()) / 1000) : null
|
||
|
||
/**
|
||
* What the indexer is doing, in one object (commands/coordinator_status.ts prints it): the catalog's room on the
|
||
* NVMe, the ranges by band (the tail, the capture windows, the history), the last hour's rates, the held groups
|
||
* with their snapshots, the phase-2 backlogs. Every number comes from the database, the snapshot directory or the
|
||
* filesystem; nothing is kept between calls.
|
||
*/
|
||
class StatusService {
|
||
async collect(): Promise<Status> {
|
||
const tail = coordinatorConfig.tail.priority
|
||
const cap = coordinatorConfig.catalogMaxNvmeGb
|
||
// The catalog's tablespace only (CatalogSpace measures it every minute anyway): pg_database_size walks the
|
||
// chunk partitions and the probe cache on the HDD, minutes on homelab.
|
||
const { rows: space } = await db.rawQuery(
|
||
`SELECT coalesce(nullif(current_setting('default_tablespace'), ''), 'pg_default') AS tablespace,
|
||
pg_tablespace_size(coalesce(nullif(current_setting('default_tablespace'), ''), 'pg_default')) AS bytes`
|
||
)
|
||
const tablespaceGb = Number(space[0].bytes) / 1e9
|
||
const fs = await statfs('/').catch(() => null)
|
||
const freeGb = fs ? (fs.bavail * fs.bsize) / 1e9 : 0
|
||
|
||
// The ranges by band: the tail, the capture windows (priorities 0–2; the sampling ranges in other years are
|
||
// capture too, at their history priorities), the history by its thousand.
|
||
const { rows: bands } = await db.rawQuery(
|
||
`SELECT CASE WHEN priority <= ? THEN 'tail'
|
||
WHEN priority < 0 THEN 'raised'
|
||
WHEN capture AND priority <= 2 THEN 'window ' || priority
|
||
WHEN priority < 1000 THEN 'history 0-999'
|
||
ELSE 'history ' || (priority / 1000) * 1000 || '+' END AS band,
|
||
count(*) FILTER (WHERE status = 'planned') AS planned,
|
||
coalesce(sum(end_id - next_id + 1) FILTER (WHERE status = 'planned'), 0) AS planned_articles,
|
||
count(*) FILTER (WHERE status = 'leased') AS leased,
|
||
count(*) FILTER (WHERE status = 'done') AS done,
|
||
coalesce(sum(end_id - start_id + 1) FILTER (WHERE status = 'done'), 0) AS done_articles,
|
||
count(*) FILTER (WHERE status = 'failed') AS failed
|
||
FROM ranges GROUP BY 1 ORDER BY min(priority)`,
|
||
[tail]
|
||
)
|
||
const { rows: hour } = await db.rawQuery(
|
||
`SELECT coalesce(sum(end_id - start_id + 1) FILTER (WHERE done_at > now() - interval '1 hour'), 0) AS indexed,
|
||
count(*) FILTER (WHERE done_at > now() - interval '1 hour') AS done_ranges,
|
||
coalesce(sum(end_id - start_id + 1) FILTER (WHERE stored_at > now() - interval '1 hour'), 0) AS stored,
|
||
count(*) FILTER (WHERE stored_at > now() - interval '1 hour') AS stored_ranges,
|
||
count(*) FILTER (WHERE capture AND stored_at IS NULL AND status <> 'done') AS corpus_pending,
|
||
coalesce(sum(end_id - start_id + 1) FILTER (WHERE capture AND stored_at IS NULL AND status <> 'done'), 0) AS corpus_pending_articles,
|
||
count(*) FILTER (WHERE status IN ('planned', 'leased') AND spooled_chunks > 0) AS spooled_ranges,
|
||
coalesce(sum(spooled_chunks) FILTER (WHERE status IN ('planned', 'leased')), 0) AS spooled_chunks,
|
||
count(DISTINCT group_id) FILTER (WHERE status = 'planned' AND priority <= ?) AS tail_groups,
|
||
count(*) FILTER (WHERE status = 'planned' AND priority <= ?) AS tail_ranges,
|
||
coalesce(sum(end_id - next_id + 1) FILTER (WHERE status = 'planned' AND priority <= ?), 0) AS tail_articles
|
||
FROM ranges`,
|
||
[tail, tail, tail]
|
||
)
|
||
const h = hour[0]
|
||
|
||
const ttl = coordinatorConfig.leaseTtlSeconds
|
||
const { rows: held } = await db.rawQuery(
|
||
`SELECT n.name AS node, g.name AS "group", l.lane, l.heartbeat_at, l.snapshot_next, l.snapshot_at, l.ranges_done, l.headers,
|
||
r.start_id, r.end_id, r.next_id, l.group_id
|
||
FROM group_leases l JOIN groups g ON g.id = l.group_id JOIN nodes n ON n.id = l.node_id
|
||
LEFT JOIN ranges r ON r.id = l.range_id
|
||
WHERE l.node_id IS NOT NULL AND l.heartbeat_at >= now() - make_interval(secs => ?)
|
||
ORDER BY l.lane, g.name`,
|
||
[ttl]
|
||
)
|
||
const snapshotSizes = new Map<string, number>()
|
||
const files: SnapshotFile[] = []
|
||
const names = new Map<number, string>(
|
||
(await db.from('groups').select('id', 'name')).map((g) => [Number(g.id), g.name as string])
|
||
)
|
||
for (const file of await readdir(GroupLeases.dir).catch(() => [] as string[])) {
|
||
const m = /^(\d+)(-tail)?\.gz$/.exec(file)
|
||
if (!m) continue
|
||
const s = await stat(join(GroupLeases.dir, file)).catch(() => null)
|
||
if (!s) continue
|
||
const lane: Lane = m[2] ? 'tail' : 'history'
|
||
snapshotSizes.set(`${m[1]}:${lane}`, s.size)
|
||
files.push({
|
||
group: names.get(Number(m[1])) ?? m[1],
|
||
lane,
|
||
mb: s.size / 1e6,
|
||
hours: (Date.now() - s.mtimeMs) / 3600_000,
|
||
})
|
||
}
|
||
files.sort((a, b) => b.mb - a.mb)
|
||
|
||
// The probe backlog from the planner's estimates of the partial indexes (counting 11 M rows took 10 s on
|
||
// homelab); the rest are small partial indexes.
|
||
const { rows: releases } = await db.rawQuery(
|
||
`SELECT (SELECT reltuples::bigint FROM pg_class WHERE oid = 'releases'::regclass) AS estimated,
|
||
(SELECT count(*) FROM releases WHERE settled_at IS NULL) AS provisional,
|
||
(SELECT greatest(0, reltuples)::bigint FROM pg_class WHERE oid = 'probe_candidates_ready_idx'::regclass) AS ready,
|
||
(SELECT greatest(0, reltuples)::bigint FROM pg_class WHERE oid = 'probe_candidates_waiting_idx'::regclass) AS waiting,
|
||
(SELECT count(*) FROM pending_releases WHERE status = 0) AS pending_waiting,
|
||
(SELECT count(*) FROM pending_releases WHERE status = 4) AS pending_named,
|
||
(SELECT count(*) FROM judge_dirty) AS dirty`
|
||
)
|
||
const r = releases[0]
|
||
const nodes = await db.from('nodes').select('name', 'last_seen_at').orderBy('name')
|
||
|
||
return {
|
||
at: new Date().toISOString(),
|
||
catalog: {
|
||
tablespace: space[0].tablespace,
|
||
gb: tablespaceGb,
|
||
capGb: cap,
|
||
full: cap > 0 && tablespaceGb > cap,
|
||
freeGb,
|
||
},
|
||
bands: bands.map((b: any) => ({
|
||
band: b.band,
|
||
planned: Number(b.planned),
|
||
plannedArticles: Number(b.planned_articles),
|
||
leased: Number(b.leased),
|
||
done: Number(b.done),
|
||
doneArticles: Number(b.done_articles),
|
||
failed: Number(b.failed),
|
||
})),
|
||
lastHour: {
|
||
indexedArticles: Number(h.indexed),
|
||
doneRanges: Number(h.done_ranges),
|
||
storedArticles: Number(h.stored),
|
||
storedRanges: Number(h.stored_ranges),
|
||
},
|
||
corpus: {
|
||
pendingRanges: Number(h.corpus_pending),
|
||
pendingArticles: Number(h.corpus_pending_articles),
|
||
},
|
||
spool: { ranges: Number(h.spooled_ranges), chunks: Number(h.spooled_chunks) },
|
||
tail: {
|
||
groups: Number(h.tail_groups),
|
||
ranges: Number(h.tail_ranges),
|
||
articles: Number(h.tail_articles),
|
||
},
|
||
held: held.map((l: any) => {
|
||
const size = snapshotSizes.get(`${l.group_id}:${l.lane}`)
|
||
return {
|
||
node: l.node,
|
||
group: l.group,
|
||
lane: l.lane,
|
||
range:
|
||
l.start_id !== null
|
||
? { start: Number(l.start_id), end: Number(l.end_id), next: Number(l.next_id) }
|
||
: null,
|
||
heartbeatSeconds: seconds(l.heartbeat_at) ?? 0,
|
||
snapshotNext: l.snapshot_next === null ? null : Number(l.snapshot_next),
|
||
snapshotSeconds: seconds(l.snapshot_at),
|
||
snapshotMb: size === undefined ? null : size / 1e6,
|
||
rangesDone: Number(l.ranges_done),
|
||
headers: Number(l.headers),
|
||
}
|
||
}),
|
||
snapshots: {
|
||
count: files.length,
|
||
gb: files.reduce((sum, f) => sum + f.mb, 0) / 1000,
|
||
oldestHours: files.length > 0 ? Math.max(...files.map((f) => f.hours)) : null,
|
||
largest: files.slice(0, 5),
|
||
},
|
||
releases: { estimated: Math.max(0, Number(r.estimated)), provisional: Number(r.provisional) },
|
||
probes: {
|
||
ready: Number(r.ready),
|
||
waiting: Number(r.waiting),
|
||
pendingWaiting: Number(r.pending_waiting),
|
||
pendingNamed: Number(r.pending_named),
|
||
},
|
||
judge: { dirty: Number(r.dirty) },
|
||
nodes: nodes.map((n) => ({ name: n.name, lastSeenSeconds: seconds(n.last_seen_at) })),
|
||
}
|
||
}
|
||
|
||
/** The status as a text panel. */
|
||
render(s: Status): string {
|
||
const n = (v: number) => Math.round(v).toLocaleString('en-US')
|
||
const b = (v: number) =>
|
||
v >= 1e9
|
||
? `${(v / 1e9).toFixed(2)} B`
|
||
: v >= 1e6
|
||
? `${(v / 1e6).toFixed(1)} M`
|
||
: v >= 1e3
|
||
? `${(v / 1e3).toFixed(0)} k`
|
||
: String(v)
|
||
const age = (sec: number | null) =>
|
||
sec === null
|
||
? '-'
|
||
: sec < 90
|
||
? `${sec} s`
|
||
: sec < 5400
|
||
? `${Math.round(sec / 60)} min`
|
||
: `${(sec / 3600).toFixed(1)} h`
|
||
const lines: string[] = []
|
||
lines.push(`Status at ${s.at}`)
|
||
lines.push('')
|
||
lines.push(
|
||
`Catalog: ${s.catalog.gb.toFixed(1)} GB in tablespace ${s.catalog.tablespace}${s.catalog.capGb ? ` of ${s.catalog.capGb} GB${s.catalog.full ? ' (FULL: history paused)' : ''}` : ' (no cap)'}; ${s.catalog.freeGb.toFixed(1)} GB free on /`
|
||
)
|
||
lines.push(
|
||
`Last hour: ${b(s.lastHour.indexedArticles)} articles indexed in ${n(s.lastHour.doneRanges)} ranges (${n(s.lastHour.indexedArticles / 3600)}/s); corpus fill ${b(s.lastHour.storedArticles)} in ${n(s.lastHour.storedRanges)} ranges, ${n(s.corpus.pendingRanges)} capture ranges (${b(s.corpus.pendingArticles)}) still to store`
|
||
)
|
||
lines.push(
|
||
`Spool: ${n(s.spool.chunks)} chunks in ${n(s.spool.ranges)} ranges. Tail: ${b(s.tail.articles)} planned in ${n(s.tail.ranges)} ranges of ${n(s.tail.groups)} groups`
|
||
)
|
||
lines.push('')
|
||
const table = (header: string[], rows: string[][]) => {
|
||
const widths = header.map((h, i) => Math.max(h.length, ...rows.map((r) => r[i].length)))
|
||
const line = (r: string[]) =>
|
||
r.map((c, i) => (i === 0 ? c.padEnd(widths[i]) : c.padStart(widths[i]))).join(' ')
|
||
lines.push(line(header))
|
||
for (const r of rows) lines.push(line(r))
|
||
}
|
||
lines.push('Ranges by band:')
|
||
table(
|
||
['band', 'planned', 'articles', 'leased', 'done', 'articles', 'failed'],
|
||
s.bands.map((x) => [
|
||
x.band,
|
||
n(x.planned),
|
||
b(x.plannedArticles),
|
||
n(x.leased),
|
||
n(x.done),
|
||
b(x.doneArticles),
|
||
n(x.failed),
|
||
])
|
||
)
|
||
lines.push('')
|
||
lines.push(`Held groups (${s.held.length}):`)
|
||
if (s.held.length > 0)
|
||
table(
|
||
[
|
||
'group',
|
||
'lane',
|
||
'node',
|
||
'range',
|
||
'at',
|
||
'heartbeat',
|
||
'snapshot',
|
||
'size',
|
||
'done',
|
||
'headers',
|
||
],
|
||
s.held.map((x) => [
|
||
x.group,
|
||
x.lane,
|
||
x.node,
|
||
x.range ? `${x.range.start}-${x.range.end}` : '-',
|
||
x.range ? String(x.range.next) : '-',
|
||
age(x.heartbeatSeconds),
|
||
x.snapshotNext === null ? '-' : `${x.snapshotNext} (${age(x.snapshotSeconds)})`,
|
||
x.snapshotMb === null ? '-' : `${x.snapshotMb.toFixed(0)} MB`,
|
||
n(x.rangesDone),
|
||
b(x.headers),
|
||
])
|
||
)
|
||
lines.push('')
|
||
lines.push(
|
||
`Snapshots: ${n(s.snapshots.count)} files, ${s.snapshots.gb.toFixed(2)} GB${s.snapshots.oldestHours === null ? '' : `, oldest ${s.snapshots.oldestHours.toFixed(1)} h`}${s.snapshots.largest.length > 0 ? `; largest ${s.snapshots.largest.map((f) => `${f.group}${f.lane === 'tail' ? ' (tail)' : ''} ${f.mb.toFixed(0)} MB`).join(', ')}` : ''}`
|
||
)
|
||
lines.push(
|
||
`Releases: ~${n(s.releases.estimated)}, ${n(s.releases.provisional)} provisional. Judge: ${n(s.judge.dirty)} dirty`
|
||
)
|
||
lines.push(
|
||
`Probes: ~${n(s.probes.ready)} ready of ~${n(s.probes.waiting)} waiting; pending ladder ${n(s.probes.pendingWaiting)} waiting, ${n(s.probes.pendingNamed)} named`
|
||
)
|
||
lines.push(
|
||
`Nodes: ${s.nodes.map((x) => `${x.name} (${age(x.lastSeenSeconds)})`).join(', ') || 'none'}`
|
||
)
|
||
return lines.join('\n')
|
||
}
|
||
}
|
||
|
||
export default new StatusService()
|