MangaCircus/perf/trace.js

'use strict';
/*
 * Switchable performance trace: one JSON object per line in DATA/perf/trace-YYYY-MM-DD.jsonl.
 *
 * On while PERF_TRACE=1 or while the file DATA/perf/trace.on exists (checked every few seconds, so it can be
 * switched on a running server with `touch` / `rm`). Off, every call is a cheap no-op.
 *
 * Event kinds (field `k`): `req` one HTTP request with its phases, `client` timings the reader page reports,
 * `loop` event-loop lag / CPU / memory every SAMPLE_MS, `nas` a periodic read of one uncached library page,
 * `scan` library scans, `scraper` calls to the scraper daemon, `ahead` read-ahead batches, `trace` on/off.
 * The format and the report tool: docs/performance.md, tools/perf-report.js.
 */
const fs = require('fs'), path = require('path'), { monitorEventLoopDelay } = require('perf_hooks');

const CHECK_MS = 5000, FLUSH_MS = 1000, SAMPLE_MS = 5000;
const MAX_DAY_BYTES = 256 * 1024 * 1024; // a runaway trace never fills the data disk
const KEEP_DAYS = 14;

class Tracer {
  constructor({ dir, env = process.env, now = Date.now } = {}) {
    this.dir = dir; this.now = now;
    this.forced = env.PERF_TRACE === '1';
    this.flag = path.join(dir, 'trace.on');
    this.on = false; this.buf = []; this.day = ''; this.dayBytes = 0; this.dropped = 0;
    this.timers = []; this.probe = null; this.probeMs = Math.max(10, +env.PERF_NAS_PROBE_SECONDS || 60) * 1000;
  }

  start() {
    this.check();
    this.timers.push(setInterval(() => this.check(), CHECK_MS).unref());
    this.timers.push(setInterval(() => this.flush(), FLUSH_MS).unref());
    return this;
  }
  stop() { for (const t of this.timers) clearInterval(t); this.timers = []; this.setOn(false); this.flush(); }

  check() {
    let on = this.forced;
    if (!on) try { fs.accessSync(this.flag); on = true; } catch { /* off */ }
    this.setOn(on);
  }
  setOn(on) {
    if (on === this.on) return;
    if (on) {
      this.on = true;
      this.log({ k: 'trace', on: true, pid: process.pid, node: process.version });
      this.startSampler();
    } else {
      this.log({ k: 'trace', on: false });
      this.on = false;
      this.stopSampler();
    }
  }

  log(event) {
    if (!this.on) return;
    this.buf.push(JSON.stringify({ t: this.now(), ...event }));
    if (this.buf.length >= 500) this.flush();
  }

  flush() {
    if (!this.buf.length) return;
    const lines = this.buf.join('\n') + '\n'; this.buf = [];
    const day = new Date(this.now()).toISOString().slice(0, 10);
    try {
      if (day !== this.day) { this.day = day; this.dayBytes = 0; fs.mkdirSync(this.dir, { recursive: true }); this.prune(); }
      if (this.dayBytes + lines.length > MAX_DAY_BYTES) { this.dropped += lines.length; return; }
      fs.appendFileSync(path.join(this.dir, `trace-${day}.jsonl`), lines);
      this.dayBytes += lines.length;
    } catch (error) { console.warn('[perf] trace write failed:', error.message); }
  }
  prune() {
    const cutoff = new Date(this.now() - KEEP_DAYS * 864e5).toISOString().slice(0, 10);
    for (const name of fs.readdirSync(this.dir)) {
      const m = /^trace-(\d{4}-\d\d-\d\d)\.jsonl$/.exec(name);
      if (m && m[1] < cutoff) fs.rmSync(path.join(this.dir, name), { force: true });
    }
  }

  // Event-loop delay shows whether the reader itself (CPU, sync work) holds requests back.
  startSampler() {
    const h = monitorEventLoopDelay({ resolution: 10 }); h.enable();
    let cpu = process.cpuUsage(), at = process.hrtime.bigint();
    this.sampler = { h, timer: setInterval(() => {
      const nowCpu = process.cpuUsage(), nowAt = process.hrtime.bigint(), wall = Number(nowAt - at) / 1000;
      const mem = process.memoryUsage();
      this.log({ k: 'loop', lagP50: ms(h.percentile(50)), lagP99: ms(h.percentile(99)), lagMax: ms(h.max),
        cpu: round(((nowCpu.user - cpu.user) + (nowCpu.system - cpu.system)) / wall * 100), rss: mb(mem.rss), heap: mb(mem.heapUsed) });
      h.reset(); cpu = nowCpu; at = nowAt;
    }, SAMPLE_MS).unref() };
    this.startProbe();
  }
  stopSampler() {
    if (this.sampler) { clearInterval(this.sampler.timer); this.sampler.h.disable(); this.sampler = null; }
    if (this.probe) { clearInterval(this.probe); this.probe = null; }
  }

  // NAS probe: read one random library page now and then. A page picked at random out of a large library is
  // almost never in the OS cache, so this measures the storage itself (and shows disks waking from standby).
  setProbeSource(pick) { this.pick = pick; if (this.on) this.startProbe(); }
  startProbe() {
    if (this.probe || !this.pick) return;
    const run = () => this.probeOnce().catch(() => {});
    this.probe = setInterval(run, this.probeMs).unref();
    setTimeout(run, 1000).unref();
  }
  async probeOnce() {
    const file = this.pick && this.pick();
    if (!file) return;
    const t0 = performance.now();
    let st, bytes = 0, firstMs = null, error = null;
    try {
      st = await fs.promises.stat(file);
      const statMs = performance.now() - t0;
      const fh = await fs.promises.open(file, 'r');
      try {
        const chunk = Buffer.allocUnsafe(256 * 1024);
        for (;;) {
          const { bytesRead } = await fh.read(chunk, 0, chunk.length, bytes);
          if (firstMs === null) firstMs = performance.now() - t0;
          if (!bytesRead) break;
          bytes += bytesRead;
        }
      } finally { await fh.close(); }
      this.log({ k: 'nas', statMs: round(statMs), firstMs: round(firstMs), ms: round(performance.now() - t0), bytes });
    } catch (e) { error = e.code || e.message; this.log({ k: 'nas', ms: round(performance.now() - t0), error }); }
  }
}

// Wraps one HTTP request: `mark(name)` stores the time since the request arrived, `set` adds fields, and the
// event is written when the response is done (or the client went away). Off, returns a shared no-op.
const NOOP = { mark() {}, set() {}, stream() {}, elapsed: () => 0, on: false };
function requestTrace(tracer, req, res) {
  if (!tracer.on) return NOOP;
  const t0 = performance.now(), ev = { k: 'req', m: req.method, path: req.url.split('?')[0].slice(0, 300) };
  const trace = {
    on: true,
    mark(name) { ev[name] = round(performance.now() - t0); },
    elapsed: () => round(performance.now() - t0),
    set(fields) { Object.assign(ev, fields); },
    // Splits a file stream's time into waiting for the disk and waiting for the client (TCP backpressure):
    // pipe() pauses the file stream while the socket buffer is full and resumes it on 'drain'.
    stream(src) {
      let pausedAt = 0, waitNet = 0;
      src.on('open', () => trace.mark('open'));
      src.once('data', () => trace.mark('first'));
      src.on('data', chunk => { ev.bytes = (ev.bytes || 0) + chunk.length; });
      src.on('pause', () => { pausedAt = performance.now(); });
      src.on('resume', () => { if (pausedAt) { waitNet += performance.now() - pausedAt; pausedAt = 0; } });
      src.on('end', () => {
        trace.mark('readEnd');
        ev.netWait = round(waitNet);
        ev.disk = round(Math.max(0, ev.readEnd - (ev.open || 0) - waitNet));
      });
    },
  };
  let done = false;
  const finish = aborted => {
    if (done) return; done = true;
    ev.ms = round(performance.now() - t0); ev.status = res.statusCode;
    if (aborted) ev.aborted = true;
    ev.conn = req.socket._perfRequests = (req.socket._perfRequests || 0) + 1; // >1: keep-alive reuse
    tracer.log(ev);
  };
  res.on('finish', () => finish(false));
  res.on('close', () => finish(!res.writableFinished));
  return trace;
}

const round = v => (v == null || Number.isNaN(v) ? null : Math.round(v * 10) / 10);
const ms = ns => round(ns / 1e6);
const mb = b => Math.round(b / 1048576);

module.exports = { Tracer, requestTrace };