LibrePortal/containers/libreportal/backend/utils/metrics-ring.js

// Binary ring buffer for system metrics, on-disk.
//
// Two tiers feed the Admin → System trend charts:
//   - 1m: 1440 points  → 24 h at 1-minute resolution
//   - 5m: 2016 points  → 7  d at 5-minute resolution
//
// Why binary instead of JSON: JSON parse of a 2016-point array is enough work
// (~ms in the libreportal container) that the history endpoint felt sluggish
// at 7-day ranges. A fixed 32-byte-per-point binary file is ~10× smaller, has
// O(range) cost with no parse, and the file is mmap-friendly if we ever need
// even larger windows.
//
// On-disk layout (little-endian):
//   HEADER  (32 B)
//     0   4  magic        "LPMR"
//     4   1  version      0x01
//     5   1  point_size   bytes per point (32)
//     6   1  field_count  metric fields per point, not counting timestamp (7)
//     7   1  flags        reserved
//     8   4  capacity     max points
//    12   4  head         next write index   (0..capacity-1)
//    16   4  count        valid points       (<= capacity)
//    20   4  bucket_sec   seconds per bucket (60 | 300)
//    24   4  last_t       last bucket timestamp, unix seconds
//    28   4  reserved
//
//   POINT   (32 B)
//     0   4  uint32 t       bucket start, unix seconds
//     4   4  float32 cpu    %
//     8   4  float32 mem    %
//    12   4  float32 swap   %
//    16   4  float32 disk   %  (root mount)
//    20   4  float32 load1
//    24   4  float32 net_rx bytes/sec average over bucket
//    28   4  float32 net_tx bytes/sec
//
// File size: 32 + capacity * 32. For our tiers, 46 KB and 64 KB respectively.
// All writes are append-only (no mid-ring rewrites) and atomic at the byte-
// range level: we open with O_RDWR, write a single 32-byte point at the slot
// offset, then patch the header. A torn write is recoverable (count won't
// have advanced; the slot is just garbage that'll be overwritten next tick).

const fs = require('fs');
const path = require('path');

const MAGIC = Buffer.from('LPMR', 'ascii');
const VERSION = 0x01;
const POINT_SIZE = 32;
const HEADER_SIZE = 32;
const FIELD_COUNT = 7;
const FIELDS = ['cpu', 'mem', 'swap', 'disk', 'load1', 'net_rx', 'net_tx'];

// Build a new header buffer for `capacity` points at `bucketSec` resolution.
function newHeader(capacity, bucketSec) {
    const h = Buffer.alloc(HEADER_SIZE);
    MAGIC.copy(h, 0);
    h.writeUInt8(VERSION,     4);
    h.writeUInt8(POINT_SIZE,  5);
    h.writeUInt8(FIELD_COUNT, 6);
    h.writeUInt8(0,           7);
    h.writeUInt32LE(capacity, 8);
    h.writeUInt32LE(0,        12);   // head
    h.writeUInt32LE(0,        16);   // count
    h.writeUInt32LE(bucketSec, 20);
    h.writeUInt32LE(0,        24);   // last_t
    h.writeUInt32LE(0,        28);
    return h;
}

function parseHeader(buf) {
    if (buf.length < HEADER_SIZE) throw new Error('ring: short header');
    if (buf.slice(0, 4).compare(MAGIC) !== 0) throw new Error('ring: bad magic');
    if (buf.readUInt8(4) !== VERSION) throw new Error('ring: unsupported version');
    if (buf.readUInt8(5) !== POINT_SIZE) throw new Error('ring: wrong point size');
    return {
        version:     buf.readUInt8(4),
        pointSize:   buf.readUInt8(5),
        fieldCount:  buf.readUInt8(6),
        capacity:    buf.readUInt32LE(8),
        head:        buf.readUInt32LE(12),
        count:       buf.readUInt32LE(16),
        bucketSec:   buf.readUInt32LE(20),
        lastT:       buf.readUInt32LE(24),
    };
}

function encodePoint(p) {
    const b = Buffer.alloc(POINT_SIZE);
    b.writeUInt32LE(Math.max(0, Math.floor(p.t || 0)), 0);
    b.writeFloatLE(Number(p.cpu)    || 0,  4);
    b.writeFloatLE(Number(p.mem)    || 0,  8);
    b.writeFloatLE(Number(p.swap)   || 0, 12);
    b.writeFloatLE(Number(p.disk)   || 0, 16);
    b.writeFloatLE(Number(p.load1)  || 0, 20);
    b.writeFloatLE(Number(p.net_rx) || 0, 24);
    b.writeFloatLE(Number(p.net_tx) || 0, 28);
    return b;
}

function decodePoint(buf, offset = 0) {
    return {
        t:      buf.readUInt32LE(offset),
        cpu:    +buf.readFloatLE(offset + 4).toFixed(2),
        mem:    +buf.readFloatLE(offset + 8).toFixed(2),
        swap:   +buf.readFloatLE(offset + 12).toFixed(2),
        disk:   +buf.readFloatLE(offset + 16).toFixed(2),
        load1:  +buf.readFloatLE(offset + 20).toFixed(3),
        net_rx: Math.round(buf.readFloatLE(offset + 24)),
        net_tx: Math.round(buf.readFloatLE(offset + 28)),
    };
}

class MetricsRing {
    // file:      absolute path to the .bin
    // capacity:  max points
    // bucketSec: seconds per bucket (60 | 300)
    constructor({ file, capacity, bucketSec }) {
        this.file = file;
        this.capacity = capacity;
        this.bucketSec = bucketSec;
        this.fd = null;
        this.header = null;
    }

    // Open (creating if missing) and validate. Safe to call repeatedly.
    async open() {
        if (this.fd !== null) return;
        const dir = path.dirname(this.file);
        try { await fs.promises.mkdir(dir, { recursive: true }); } catch (_) {}
        let needInit = false;
        try {
            const st = await fs.promises.stat(this.file);
            const expected = HEADER_SIZE + this.capacity * POINT_SIZE;
            if (st.size !== expected) needInit = true;
        } catch (_) {
            needInit = true;
        }
        if (needInit) {
            const full = Buffer.alloc(HEADER_SIZE + this.capacity * POINT_SIZE);
            newHeader(this.capacity, this.bucketSec).copy(full, 0);
            await fs.promises.writeFile(this.file, full);
        }
        this.fd = await fs.promises.open(this.file, 'r+');
        const headBuf = Buffer.alloc(HEADER_SIZE);
        await this.fd.read(headBuf, 0, HEADER_SIZE, 0);
        this.header = parseHeader(headBuf);
        if (this.header.capacity !== this.capacity || this.header.bucketSec !== this.bucketSec) {
            // Capacity / bucket changed (config bump) — start fresh. Old data
            // wouldn't line up with the new grid anyway.
            await this.fd.close();
            this.fd = null;
            await fs.promises.unlink(this.file);
            return this.open();
        }
    }

    async close() {
        if (this.fd) { try { await this.fd.close(); } catch (_) {} this.fd = null; }
    }

    // Append a single point. Caller is responsible for rounding `t` to the
    // bucket boundary (e.g. Math.floor(now / bucketSec) * bucketSec).
    async append(point) {
        await this.open();
        const slot = this.header.head;
        const offset = HEADER_SIZE + slot * POINT_SIZE;
        const buf = encodePoint(point);
        await this.fd.write(buf, 0, POINT_SIZE, offset);
        this.header.head = (slot + 1) % this.capacity;
        this.header.count = Math.min(this.capacity, this.header.count + 1);
        this.header.lastT = point.t >>> 0;
        const h = Buffer.alloc(HEADER_SIZE);
        // Re-encode only the mutable fields; the static prefix stays the same.
        MAGIC.copy(h, 0);
        h.writeUInt8(VERSION,     4);
        h.writeUInt8(POINT_SIZE,  5);
        h.writeUInt8(FIELD_COUNT, 6);
        h.writeUInt8(0,           7);
        h.writeUInt32LE(this.capacity,    8);
        h.writeUInt32LE(this.header.head, 12);
        h.writeUInt32LE(this.header.count, 16);
        h.writeUInt32LE(this.bucketSec,   20);
        h.writeUInt32LE(this.header.lastT, 24);
        h.writeUInt32LE(0,                28);
        await this.fd.write(h, 0, HEADER_SIZE, 0);
        // fdatasync would be safer against power loss but doubles cost; the
        // worst case here is losing the most recent minute or two of metrics
        // — not worth the IO penalty on every append.
    }

    // Read the last `n` points in chronological order (oldest → newest).
    // Returns [] if the ring is empty.
    async readLast(n) {
        await this.open();
        const { head, count, capacity } = this.header;
        if (count === 0) return [];
        const want = Math.max(1, Math.min(n | 0, count));
        // The oldest of `want` lives `want` slots before head (modulo cap).
        const start = (head - want + capacity) % capacity;
        // Two-segment read so we don't span the wrap unnecessarily.
        const out = new Array(want);
        if (start + want <= capacity) {
            const buf = Buffer.alloc(want * POINT_SIZE);
            await this.fd.read(buf, 0, buf.length, HEADER_SIZE + start * POINT_SIZE);
            for (let i = 0; i < want; i++) out[i] = decodePoint(buf, i * POINT_SIZE);
        } else {
            const firstLen = capacity - start;
            const a = Buffer.alloc(firstLen * POINT_SIZE);
            const b = Buffer.alloc((want - firstLen) * POINT_SIZE);
            await this.fd.read(a, 0, a.length, HEADER_SIZE + start * POINT_SIZE);
            await this.fd.read(b, 0, b.length, HEADER_SIZE);
            for (let i = 0; i < firstLen; i++)        out[i] = decodePoint(a, i * POINT_SIZE);
            for (let i = 0; i < want - firstLen; i++) out[firstLen + i] = decodePoint(b, i * POINT_SIZE);
        }
        return out;
    }

    // Convenience: latest bucket timestamp on disk (0 if empty).
    async lastT() {
        await this.open();
        return this.header.lastT || 0;
    }
}

module.exports = { MetricsRing, FIELDS, HEADER_SIZE, POINT_SIZE };