feat(jobs): host-metric sampler captures load + memory every minute, 7d retention

Powers the 2-hour host-load chart Marc asked for above the
/admin/system-status section. The pg-boss `host-metric-sample` queue
runs `* * * * *` and writes one HostMetric row per tick.

Sampler implementation:
  - `os.loadavg()` → load_avg_1/5/15 (rounded to 2 decimals)
  - `os.totalmem() - os.freemem()` → mem_used_bytes / mem_total_bytes
  - On Linux: parses /proc/diskstats and sums sector counters across
    physical block devices (skipping loop/ram/dm-/md/sr/fd/nbd/zram
    pseudo-devices that double-count). Multiplies by the canonical
    512-byte sector size for raw bytes
  - On macOS / Windows / inside containers without /proc: disk fields
    are NULL, the chart gracefully degrades

The disk-stats reader is injected so the unit test can stub it without
touching `node:fs`. `runHostMetricTick()` does the insert + the
`deleteMany` for rows older than HOST_METRIC_RETENTION_DAYS (default 7,
env override capped at >= 1d to guard misconfigs).

Wired into reminder-worker.ts alongside the other recurring queues.
The worker handler swallows errors to a Wide-Event warning rather than
re-queueing, because a single missed tick costs a minute of chart data
— not worth a retry storm.

8 unit tests pin retention env handling, the disk-stats null/throw
fallback paths, and the create + deleteMany call shape.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Marc-André BombeckandClaude Opus 4.7 committed 2026-05-09 23:50:08 +02:00
1 parent 621109cf20
commit f1bd8014d8
3 files changed
+366

No files matched your search

@@ -0,0 +1,162 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import type { PrismaClient } from "@/generated/prisma/client";
import {
captureHostMetric,
getHostMetricRetentionDays,
runHostMetricTick,
DEFAULT_HOST_METRIC_RETENTION_DAYS,
} from "../host-metric-sampler";
function makePrismaMock() {
return {
hostMetric: {
create: vi.fn().mockResolvedValue({}),
deleteMany: vi.fn().mockResolvedValue({ count: 3 }),
},
} as unknown as PrismaClient;
}
const ORIGINAL_ENV = { ...process.env };
beforeEach(() => {
process.env = { ...ORIGINAL_ENV };
delete process.env.HOST_METRIC_RETENTION_DAYS;
});
afterEach(() => {
process.env = ORIGINAL_ENV;
vi.restoreAllMocks();
});
describe("getHostMetricRetentionDays", () => {
it("returns 7 days when env is unset", () => {
expect(getHostMetricRetentionDays()).toBe(
DEFAULT_HOST_METRIC_RETENTION_DAYS,
);
});
it("respects a valid override", () => {
process.env.HOST_METRIC_RETENTION_DAYS = "14";
expect(getHostMetricRetentionDays()).toBe(14);
});
it("ignores nonsensical values (NaN, 0, negative)", () => {
process.env.HOST_METRIC_RETENTION_DAYS = "0";
expect(getHostMetricRetentionDays()).toBe(
DEFAULT_HOST_METRIC_RETENTION_DAYS,
);
process.env.HOST_METRIC_RETENTION_DAYS = "-100";
expect(getHostMetricRetentionDays()).toBe(
DEFAULT_HOST_METRIC_RETENTION_DAYS,
);
process.env.HOST_METRIC_RETENTION_DAYS = "not a number";
expect(getHostMetricRetentionDays()).toBe(
DEFAULT_HOST_METRIC_RETENTION_DAYS,
);
});
});
describe("captureHostMetric", () => {
it("returns load + memory fields and forwards disk-stats output", async () => {
const sample = await captureHostMetric(async () => ({
readBytes: BigInt(12345),
writeBytes: BigInt(67890),
}));
expect(typeof sample.loadAvg1).toBe("number");
expect(typeof sample.loadAvg5).toBe("number");
expect(typeof sample.loadAvg15).toBe("number");
// Two-decimal precision keeps Postgres rows compact.
expect(sample.loadAvg1).toBe(Math.round(sample.loadAvg1 * 100) / 100);
expect(typeof sample.memUsedBytes).toBe("bigint");
expect(typeof sample.memTotalBytes).toBe("bigint");
expect(sample.memTotalBytes).toBeGreaterThan(BigInt(0));
expect(sample.memUsedBytes).toBeLessThanOrEqual(sample.memTotalBytes);
expect(sample.diskReadBytes).toBe(BigInt(12345));
expect(sample.diskWriteBytes).toBe(BigInt(67890));
});
it("returns null disk fields when the diskstats reader returns null", async () => {
const sample = await captureHostMetric(async () => null);
expect(sample.diskReadBytes).toBeNull();
expect(sample.diskWriteBytes).toBeNull();
});
it("returns null disk fields when the diskstats reader throws", async () => {
const sample = await captureHostMetric(async () => {
throw new Error("EACCES /proc/diskstats");
});
expect(sample.diskReadBytes).toBeNull();
expect(sample.diskWriteBytes).toBeNull();
});
});
describe("runHostMetricTick", () => {
it("inserts a sample row and prunes expired rows", async () => {
const prisma = makePrismaMock();
const now = new Date("2026-05-09T12:00:00Z");
const result = await runHostMetricTick(prisma, {
now,
captureFn: async () => ({
loadAvg1: 0.42,
loadAvg5: 0.55,
loadAvg15: 0.61,
memUsedBytes: BigInt(4_000_000_000),
memTotalBytes: BigInt(8_000_000_000),
diskReadBytes: BigInt(1_000),
diskWriteBytes: BigInt(2_000),
}),
});
expect(result).toEqual({ inserted: 1, pruned: 3 });
expect(prisma.hostMetric.create).toHaveBeenCalledTimes(1);
expect(prisma.hostMetric.create).toHaveBeenCalledWith({
data: {
capturedAt: now,
loadAvg1: 0.42,
loadAvg5: 0.55,
loadAvg15: 0.61,
memUsedBytes: BigInt(4_000_000_000),
memTotalBytes: BigInt(8_000_000_000),
diskReadBytes: BigInt(1_000),
diskWriteBytes: BigInt(2_000),
},
});
// 7-day cutoff by default.
const expectedCutoff = new Date(now.getTime() - 7 * 86_400_000);
expect(prisma.hostMetric.deleteMany).toHaveBeenCalledWith({
where: { capturedAt: { lt: expectedCutoff } },
});
});
it("respects HOST_METRIC_RETENTION_DAYS for the prune cutoff", async () => {
process.env.HOST_METRIC_RETENTION_DAYS = "14";
const prisma = makePrismaMock();
const now = new Date("2026-05-09T12:00:00Z");
await runHostMetricTick(prisma, {
now,
captureFn: async () => ({
loadAvg1: 0,
loadAvg5: 0,
loadAvg15: 0,
memUsedBytes: BigInt(1),
memTotalBytes: BigInt(2),
diskReadBytes: null,
diskWriteBytes: null,
}),
});
const expectedCutoff = new Date(now.getTime() - 14 * 86_400_000);
expect(prisma.hostMetric.deleteMany).toHaveBeenCalledWith({
where: { capturedAt: { lt: expectedCutoff } },
});
});
});
+174
View File
@@ -0,0 +1,174 @@
/**
* Per-minute host-load sampler that backs the admin /admin/system-status
* "last 2 hours" chart (v1.4.16 phase B3).
*
* Marc asked for a small at-a-glance graph above the system-status facts
* grid showing CPU + memory + disk-IO so a glance at /admin tells him
* whether the apps-01 host is healthy. We considered scraping Coolify's
* Sentinel collector, but the Coolify HTTP API does not expose a clean
* read endpoint for per-server metrics — every approach ended up either
* scraping the Coolify dashboard HTML or hitting a private RPC. An
* in-process sampler costs us a single 7-column row per minute, never
* leaves the app's own DB, and works identically on dev (macOS) and
* production (Linux).
*
* The sampler uses `os.loadavg()` (1/5/15-minute averages) and
* `os.totalmem() - os.freemem()` for memory. On Linux it also reads the
* cumulative /proc/diskstats counters; non-Linux hosts return null disk
* fields, which the API endpoint then surfaces as "no disk data" instead
* of failing the whole chart.
*
* Retention: rows older than 7 days are deleted at the end of every
* tick. With one insert per minute that's 10,080 rows steady-state,
* indexed by `captured_at` so the 2h query stays under 200 rows.
*/
import { loadavg, totalmem, freemem } from "node:os";
import { readFile } from "node:fs/promises";
import type { PrismaClient } from "@/generated/prisma/client";
export const DEFAULT_HOST_METRIC_RETENTION_DAYS = 7;
export interface HostMetricSample {
loadAvg1: number;
loadAvg5: number;
loadAvg15: number;
memUsedBytes: bigint;
memTotalBytes: bigint;
diskReadBytes: bigint | null;
diskWriteBytes: bigint | null;
}
/**
* Capture a single host-load snapshot. Pure-ish — only `os.*` and
* `/proc/diskstats` are touched, so the unit test can stub
* `readDiskStats` to assert the disk path without mocking `node:os`.
*/
export async function captureHostMetric(
readDiskStats: () => Promise<{
readBytes: bigint;
writeBytes: bigint;
} | null> = readLinuxDiskStats,
): Promise<HostMetricSample> {
const [load1, load5, load15] = loadavg();
const total = totalmem();
const free = freemem();
const used = total - free;
const disk = await readDiskStats().catch(() => null);
return {
loadAvg1: round(load1),
loadAvg5: round(load5),
loadAvg15: round(load15),
memUsedBytes: BigInt(used),
memTotalBytes: BigInt(total),
diskReadBytes: disk?.readBytes ?? null,
diskWriteBytes: disk?.writeBytes ?? null,
};
}
/**
* Linux /proc/diskstats reader. Returns the sum of read + write byte
* counters across all non-virtual block devices (loop/ram/dm-* are
* excluded — they double-count or report meaningless numbers).
*
* Each /proc/diskstats line is whitespace-separated:
* 3:major 4:minor ... 6:sectors_read ... 10:sectors_written ...
* Multiplying sectors by the canonical 512 B sector size gives the
* cumulative byte counters the chart uses to derive bytes-per-second
* between samples.
*/
async function readLinuxDiskStats(): Promise<{
readBytes: bigint;
writeBytes: bigint;
} | null> {
let raw: string;
try {
raw = await readFile("/proc/diskstats", "utf-8");
} catch {
return null;
}
let totalRead = BigInt(0);
let totalWrite = BigInt(0);
const SECTOR_BYTES = BigInt(512);
for (const line of raw.split("\n")) {
const parts = line.trim().split(/\s+/);
if (parts.length < 14) continue;
const device = parts[2];
// Skip virtual / synthetic devices that double-count or report
// bogus numbers. ram*/loop* are pseudo-devices; dm-* and md* are
// device-mapper / RAID overlays whose underlying disks are already
// counted by the physical device line.
if (
!device ||
/^(loop|ram|dm-|md|sr|fd)/.test(device) ||
/^(nbd|zram)/.test(device)
) {
continue;
}
const sectorsRead = BigInt(parts[5] ?? "0");
const sectorsWritten = BigInt(parts[9] ?? "0");
totalRead += sectorsRead * SECTOR_BYTES;
totalWrite += sectorsWritten * SECTOR_BYTES;
}
return { readBytes: totalRead, writeBytes: totalWrite };
}
export function getHostMetricRetentionDays(): number {
const raw = process.env.HOST_METRIC_RETENTION_DAYS;
if (raw === undefined) return DEFAULT_HOST_METRIC_RETENTION_DAYS;
const parsed = parseInt(raw, 10);
if (!Number.isFinite(parsed) || parsed <= 0) {
return DEFAULT_HOST_METRIC_RETENTION_DAYS;
}
// Refuse a misconfigured value below 1 day — the chart needs at least
// 2h of data, but cron drift + missed ticks make 1d the safe floor.
if (parsed < 1) return DEFAULT_HOST_METRIC_RETENTION_DAYS;
return parsed;
}
/**
* Insert a single sample row + delete expired rows. Returns the number
* of rows pruned so the worker can annotate its Wide Event.
*/
export async function runHostMetricTick(
prisma: PrismaClient,
options: {
now?: Date;
captureFn?: () => Promise<HostMetricSample>;
} = {},
): Promise<{ inserted: 1; pruned: number }> {
const now = options.now ?? new Date();
const sample = await (options.captureFn ?? (() => captureHostMetric()))();
await prisma.hostMetric.create({
data: {
capturedAt: now,
loadAvg1: sample.loadAvg1,
loadAvg5: sample.loadAvg5,
loadAvg15: sample.loadAvg15,
memUsedBytes: sample.memUsedBytes,
memTotalBytes: sample.memTotalBytes,
diskReadBytes: sample.diskReadBytes,
diskWriteBytes: sample.diskWriteBytes,
},
});
const cutoff = new Date(
now.getTime() - getHostMetricRetentionDays() * 86_400_000,
);
const { count } = await prisma.hostMetric.deleteMany({
where: { capturedAt: { lt: cutoff } },
});
return { inserted: 1, pruned: count };
}
function round(value: number): number {
// 2-decimal precision is enough for a chart whose y-axis spans 0–4
// load — and it keeps Postgres `DOUBLE PRECISION` rows compact.
return Math.round(value * 100) / 100;
}
+30
View File
@@ -29,6 +29,7 @@ import {
import { setGlobalBoss } from "@/lib/jobs/boss-instance";
import { cleanupExpiredIdempotencyKeys } from "@/lib/jobs/idempotency-cleanup";
import { cleanupOldAuditLogs } from "@/lib/jobs/audit-log-cleanup";
import { runHostMetricTick } from "@/lib/jobs/host-metric-sampler";
import { rotateLegacyMoodLogSecrets } from "@/lib/moodlog-secret";
import { deleteMessage } from "@/lib/telegram";
import { decrypt, encrypt } from "@/lib/crypto";
@@ -105,6 +106,9 @@ const OFFHOST_BACKUP_QUEUE = "data-backup-offhost";
// in-DB DATA_BACKUP at 03:00 (Sundays only) so the off-host copy is
// always at-or-ahead of the local one.
const OFFHOST_BACKUP_CRON = "30 2 * * *";
const HOST_METRIC_QUEUE = "host-metric-sample";
// Per-minute cadence — matches the chart's 60s polling refetchInterval.
const HOST_METRIC_CRON = "* * * * *";
interface ReminderCheckPayload {
triggeredAt: string;
@@ -168,6 +172,10 @@ interface OffhostBackupPayload {
triggeredAt: string;
}
interface HostMetricSamplePayload {
triggeredAt: string;
}
// Re-export timezone utilities under local names for backward compatibility
const getUserTodayBounds = getUserTodayBoundsUtil;
const getDayOfWeekInTz = getDayOfWeekInTzUtil;
@@ -915,6 +923,21 @@ async function handleAuditLogCleanup(jobs: Job<AuditLogCleanupPayload>[]) {
});
}
async function handleHostMetricSample(jobs: Job<HostMetricSamplePayload>[]) {
void jobs;
await withBackgroundEvent("job.host_metric_sample", async (evt) => {
const p = getWorkerPrisma();
try {
const { pruned } = await runHostMetricTick(p);
evt.addMeta("host_metric_pruned", pruned);
} catch (err) {
// The chart degrades gracefully when samples are missing — log
// and move on rather than poisoning the boss queue with retries.
evt.addWarning(`host-metric-sample failed: ${err}`);
}
});
}
async function handleOffhostBackup(jobs: Job<OffhostBackupPayload>[]) {
void jobs;
await withBackgroundEvent("job.offhost_backup", async (evt) => {
@@ -1160,6 +1183,7 @@ export async function startReminderWorker() {
IDEMPOTENCY_CLEANUP_QUEUE,
AUDIT_LOG_CLEANUP_QUEUE,
OFFHOST_BACKUP_QUEUE,
HOST_METRIC_QUEUE,
];
for (const q of allQueues) {
@@ -1182,6 +1206,7 @@ export async function startReminderWorker() {
[IDEMPOTENCY_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_CRON],
[AUDIT_LOG_CLEANUP_QUEUE, AUDIT_LOG_CLEANUP_CRON],
[OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON],
[HOST_METRIC_QUEUE, HOST_METRIC_CRON],
];
for (const [name, cron] of schedules) {
@@ -1264,6 +1289,11 @@ export async function startReminderWorker() {
{ localConcurrency: 1 },
handleOffhostBackup,
);
await boss.work<HostMetricSamplePayload>(
HOST_METRIC_QUEUE,
{ localConcurrency: 1 },
handleHostMetricSample,
);
return boss;
}