Add live server performance monitoring
This commit is contained in:
85
server/request-monitor.js
Normal file
85
server/request-monitor.js
Normal file
@@ -0,0 +1,85 @@
|
||||
import { appendFile, mkdir, readFile } from 'node:fs/promises';
|
||||
import { dirname } from 'node:path';
|
||||
import { statfs } from 'node:fs/promises';
|
||||
|
||||
const routeKey = (path) => path
|
||||
.replace(/\/api\/media\/[^/]+/g, '/api/media/:id')
|
||||
.replace(/\/api\/download\/[^/]+/g, '/api/download/:id')
|
||||
.replace(/\/api\/admin\/uploads\/[^/]+/g, '/api/admin/uploads/:id')
|
||||
.replace(/\?.*$/, '');
|
||||
const percentile = (values, p) => values.length ? values[Math.min(values.length - 1, Math.ceil(values.length * p) - 1)] : 0;
|
||||
|
||||
export function createRequestMonitor({ snapshotFile = null, diskPaths = [], adapters = {}, now = () => Date.now(), snapshotIntervalMs = 60_000 } = {}) {
|
||||
const routes = new Map();
|
||||
const snapshots = [];
|
||||
let inflight = 0;
|
||||
let lastCpu = process.cpuUsage();
|
||||
let lastCpuAt = now();
|
||||
let snapshotTimer;
|
||||
|
||||
async function sample() {
|
||||
const cpuNow = process.cpuUsage();
|
||||
const elapsed = Math.max(1, now() - lastCpuAt) * 1000;
|
||||
const cpuPct = Math.min(100 * (cpuNow.user + cpuNow.system - lastCpu.user - lastCpu.system) / elapsed, 100);
|
||||
lastCpu = cpuNow; lastCpuAt = now();
|
||||
const volumes = await Promise.all(diskPaths.map(async ({ name, path }) => {
|
||||
try { const fs = await statfs(path); return { name, freeBytes: Number(fs.bavail) * Number(fs.bsize), totalBytes: Number(fs.blocks) * Number(fs.bsize) }; }
|
||||
catch { return { name, unavailable: true }; }
|
||||
}));
|
||||
const routeMetrics = Object.fromEntries([...routes].map(([key, v]) => {
|
||||
const sorted = [...v.latencies].sort((a, b) => a - b);
|
||||
return [key, { p50: percentile(sorted, .5), p95: percentile(sorted, .95), p99: percentile(sorted, .99), requests: v.requests, errors: v.errors, errorRate: v.requests ? v.errors / v.requests : 0 }];
|
||||
}));
|
||||
const [ytdlp, media, transcription, cache] = await Promise.all([
|
||||
adapters.ytdlp?.() ?? null, adapters.media?.() ?? null, adapters.transcription?.() ?? null, adapters.cache?.() ?? null,
|
||||
]);
|
||||
return {
|
||||
at: now(), routes: routeMetrics, inflight, rssBytes: process.memoryUsage().rss, cpuPercent: cpuPct,
|
||||
ytdlp, media, transcription, cache, volumes,
|
||||
};
|
||||
}
|
||||
|
||||
async function snapshot() {
|
||||
const value = await sample(); snapshots.push(value);
|
||||
while (snapshots.length && now() - snapshots[0].at > 24 * 3600_000) snapshots.shift();
|
||||
if (snapshotFile) {
|
||||
try { await mkdir(dirname(snapshotFile), { recursive: true }); await appendFile(snapshotFile, JSON.stringify(value) + '\n'); }
|
||||
catch (error) { console.warn('[monitor] snapshot write failed:', error.message); }
|
||||
}
|
||||
return value;
|
||||
}
|
||||
async function history() {
|
||||
const cutoff = now() - 24 * 3600_000;
|
||||
const out = snapshots.filter(x => x.at >= cutoff);
|
||||
if (snapshotFile) {
|
||||
try {
|
||||
const persisted = (await readFile(snapshotFile, 'utf8')).split('\n').filter(Boolean).map(line => { try { return JSON.parse(line); } catch { return null; } }).filter(x => x?.at >= cutoff);
|
||||
return [...persisted, ...out].slice(-1440);
|
||||
} catch { /* no file yet */ }
|
||||
}
|
||||
return out.slice(-1440);
|
||||
}
|
||||
|
||||
function middleware() {
|
||||
return async (c, next) => {
|
||||
const started = now(); inflight++;
|
||||
let status = 500;
|
||||
try { await next(); status = c.res.status; }
|
||||
finally {
|
||||
inflight--;
|
||||
const key = routeKey(c.req.path);
|
||||
let row = routes.get(key);
|
||||
if (!row) { row = { latencies: [], requests: 0, errors: 0 }; routes.set(key, row); }
|
||||
row.requests++; if (status >= 500) row.errors++;
|
||||
row.latencies.push(Math.max(0, now() - started)); if (row.latencies.length > 1000) row.latencies.shift();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if (snapshotIntervalMs > 0) { snapshotTimer = setInterval(() => snapshot().catch(() => {}), snapshotIntervalMs); snapshotTimer.unref?.(); }
|
||||
return { middleware, sample, snapshot, history, close: () => clearInterval(snapshotTimer) };
|
||||
}
|
||||
|
||||
export function registerRequestMonitorRoutes(app, { adminAuth, monitor }) {
|
||||
app.get('/api/admin/ops/metrics', adminAuth, async c => c.json({ ok: true, current: await monitor.sample(), history: await monitor.history() }, 200, { 'Cache-Control': 'no-store' }));
|
||||
}
|
||||
19
server/request-monitor.test.js
Normal file
19
server/request-monitor.test.js
Normal file
@@ -0,0 +1,19 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createRequestMonitor } from './request-monitor.js';
|
||||
|
||||
test('request monitor records route latency, status errors, and current process health', async () => {
|
||||
let t = 100;
|
||||
const monitor = createRequestMonitor({ now: () => t, snapshotIntervalMs: 0, adapters: { media: () => ({ queued: 2 }) } });
|
||||
const middleware = monitor.middleware();
|
||||
const ctx = (path, status) => ({ req: { path }, res: { status } });
|
||||
await middleware(ctx('/api/media/abc/status', 200), async () => { t += 8; });
|
||||
await middleware(ctx('/api/media/xyz/status', 503), async () => { t += 12; });
|
||||
const sample = await monitor.sample();
|
||||
assert.equal(sample.routes['/api/media/:id/status'].requests, 2);
|
||||
assert.equal(sample.routes['/api/media/:id/status'].errors, 1);
|
||||
assert.equal(sample.routes['/api/media/:id/status'].p50, 8);
|
||||
assert.equal(sample.media.queued, 2);
|
||||
assert.ok(sample.rssBytes > 0);
|
||||
monitor.close();
|
||||
});
|
||||
@@ -71,6 +71,7 @@ import { createYtdlpPool } from './ytdlp-pool.js';
|
||||
import { createWarmQueue } from './warm-queue.js';
|
||||
import { registerDeepSearchRoutes } from './deep-search.js';
|
||||
import { dirname, join as pathJoin } from 'node:path';
|
||||
import { createRequestMonitor, registerRequestMonitorRoutes } from './request-monitor.js';
|
||||
|
||||
// A media proxy must not die because one client's stream hit an edge case
|
||||
// (see /api/play cancel()): log and keep serving instead of crash-looping.
|
||||
@@ -346,6 +347,17 @@ function channelToUrl(c) {
|
||||
|
||||
const app = new Hono();
|
||||
app.use('*', logger());
|
||||
const requestMonitor = createRequestMonitor({
|
||||
snapshotFile: pathJoin(dirname(process.env.DB_PATH || './data/ytplayer.db'), 'ops-metrics.jsonl'),
|
||||
diskPaths: [{ name: 'app-data', path: dirname(process.env.DB_PATH || './data/ytplayer.db') }, { name: 'media', path: process.env.MEDIA_DIR || './data/media' }],
|
||||
adapters: {
|
||||
ytdlp: () => ytdlpPool ? ytdlpPool.stats() : null,
|
||||
media: () => media?.stats ? media.stats() : null,
|
||||
transcription: () => notesDb.transcriptionStats?.() || null,
|
||||
cache: () => ({ ...searchCacheCounters, memoryEntries: searchCache.size }),
|
||||
},
|
||||
});
|
||||
app.use('*', requestMonitor.middleware());
|
||||
registerCatalogRoutes(app, { resolveListener: recommendationListener });
|
||||
|
||||
// ============================================================================
|
||||
@@ -383,6 +395,7 @@ const SEARCH_CACHE_TTL_MS = 10 * 60_000; // fresh: answered from memory
|
||||
const SEARCH_CACHE_STALE_MS = 15 * 24 * 3600_000; // persistent copy: stale-while-revalidate for 15 days
|
||||
const searchCache = new Map(); // lowercased q -> { yt, fetchedAt } (YouTube cards only)
|
||||
const inflightSearches = new Map(); // lowercased q -> Promise<{ yt, youtubeError? }>
|
||||
const searchCacheCounters = { hits: 0, misses: 0 };
|
||||
|
||||
// The YouTube side of one search (up to 200 cards), shared by every caller asking
|
||||
// the same thing at the same time, and written to memory + the persistent cache.
|
||||
@@ -452,11 +465,13 @@ app.get('/api/search', async (c) => {
|
||||
if (row) { hit = { yt: row.results, fetchedAt: row.fetchedAt }; searchCache.set(key, hit); }
|
||||
}
|
||||
const age = hit ? Date.now() - hit.fetchedAt : Infinity;
|
||||
if (hit && age < SEARCH_CACHE_TTL_MS) return answer(hit.yt, { fetchedAt: hit.fetchedAt });
|
||||
if (hit && age < SEARCH_CACHE_TTL_MS) { searchCacheCounters.hits++; return answer(hit.yt, { fetchedAt: hit.fetchedAt }); }
|
||||
if (hit && age < SEARCH_CACHE_STALE_MS) {
|
||||
searchCacheCounters.hits++;
|
||||
fetchYoutube(q).catch(() => { /* keep serving the stale copy */ });
|
||||
return answer(hit.yt, { stale: true, fetchedAt: hit.fetchedAt });
|
||||
}
|
||||
searchCacheCounters.misses++;
|
||||
}
|
||||
try {
|
||||
const { yt, fetchedAt } = await fetchYoutube(q);
|
||||
@@ -2340,6 +2355,7 @@ const uploads = registerUploadRoutes(app, {
|
||||
parseLrc,
|
||||
sanitizeLyrics,
|
||||
});
|
||||
registerRequestMonitorRoutes(app, { adminAuth: notes.requireAdminOrToken, monitor: requestMonitor });
|
||||
const isUpload = (id) => uploads.isUploadId(id);
|
||||
|
||||
// ============================================================================
|
||||
|
||||
@@ -108,5 +108,5 @@ export function createYtdlpPool({ ytdlpPath = '', size = 2, timeoutMs = 60_000,
|
||||
for (const w of workers) if (!w.busy) retire(w);
|
||||
}
|
||||
|
||||
return { run, close, recycle, get disabled() { return disabled; } };
|
||||
return { run, close, recycle, stats: () => ({ queued: queue.length, workers: workers.length, busy: workers.filter(w => !!w.busy).length, disabled }), get disabled() { return disabled; } };
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user