peardata/server/services/collectors/ebpf.js

/**
 * Embedded eBPF-family collector.
 *
 * Prefer the bundled `peardata-ebpf` helper (extracted from the server
 * binary). Falls back to the NDJSON file bridge, then to an in-process
 * JS implementation of the same charts.
 *
 * Enable: PEARDATA_EBPF=1 (default on when helper is available)
 * Disable: PEARDATA_EBPF=0
 */
import { spawn } from 'child_process'
import { EventEmitter } from 'events'
import fs from 'fs'
import os from 'os'
import {
  SAMPLE_INTERVAL_MS,
  registerChart,
} from '../../../shared/metrics.js'
import { readFileCached, readFileBuf } from '../../utils/fd-cache.js'
import { extractHelper, hasEmbeddedHelper } from '../../native/extract-helper.js'
import logger from '../../utils/logger.js'

const log = logger.child('ebpf')

export function isEbpfEnabled() {
  const v = process.env.PEARDATA_EBPF
  if (v === '0' || v === 'off' || v === 'false') return false
  if (v === '1' || v === 'on' || v === 'true') return true
  // Auto: on Linux when helper exists or can be extracted
  return os.platform() === 'linux'
}

function chartDef(id, title, units, dims) {
  return {
    id,
    name: id,
    context: id,
    title,
    units,
    family: 'ebpf',
    chartType: 'line',
    priority: 9500,
    plugin: 'ebpf',
    dimensions: dims.map((d) => ({ id: d, name: d, algorithm: 'absolute' })),
  }
}

function ensureCharts() {
  registerChart(
    chartDef('ebpf.cachestat', 'Page cache efficiency', 'events/s', [
      'hits',
      'misses',
      'ratio',
      'pgpgin',
      'pgpgout',
    ])
  )
  registerChart(chartDef('ebpf.fd', 'Open file descriptors', 'files', ['open', 'max']))
  registerChart(chartDef('ebpf.oom', 'OOM kills', 'kills/s', ['kills', 'total']))
  registerChart(
    chartDef('ebpf.process', 'Process lifecycle', 'events/s', ['forks', 'ctxt'])
  )
  registerChart(chartDef('ebpf.shm', 'System V shared memory', 'bytes', ['segments', 'bytes']))
  registerChart(chartDef('ebpf.swap', 'Swap I/O (eBPF family)', 'pages/s', ['in', 'out']))
  registerChart(
    chartDef('ebpf.vfs', 'VFS page I/O', 'pages/s', ['read_pages', 'write_pages'])
  )
  registerChart(
    chartDef('ebpf.socket', 'Socket tracking', 'sockets', ['tcp_inuse', 'tcp_delta'])
  )
}

export class EbpfCollector extends EventEmitter {
  constructor(opts = {}) {
    super()
    this.intervalMs =
      opts.intervalMs ?? (Number(process.env.PEARDATA_SAMPLE_MS) || SAMPLE_INTERVAL_MS)
    this.child = null
    this.timer = null
    this.running = false
    this.buf = ''
    this.mode = 'none'
    /** JS fallback state */
    this.prev = {}
    this.lastTs = 0
    this.bridgeOffset = 0
  }

  start() {
    if (this.running) return
    this.running = true
    ensureCharts()

    const helper = extractHelper('peardata-ebpf')
    if (helper) {
      this.mode = 'helper'
      this._startHelper(helper)
      log.info('eBPF helper started', { helper, embedded: hasEmbeddedHelper('peardata-ebpf') })
      return
    }

    const bridge = process.env.PEARDATA_EBPF_PATH
    if (bridge) {
      this.mode = 'bridge'
      this.timer = setInterval(() => this._tickBridge(bridge), this.intervalMs)
      if (this.timer.unref) this.timer.unref()
      log.info('eBPF bridge mode', { path: bridge })
      return
    }

    this.mode = 'js'
    this.timer = setInterval(() => this._tickJs(), this.intervalMs)
    if (this.timer.unref) this.timer.unref()
    this._tickJs()
    log.info('eBPF JS fallback collector started')
  }

  stop() {
    this.running = false
    if (this.timer) clearInterval(this.timer)
    this.timer = null
    if (this.child) {
      try {
        this.child.kill('SIGTERM')
      } catch {
        // ignore
      }
      this.child = null
    }
  }

  _startHelper(bin) {
    this.child = spawn(bin, ['--interval', String(this.intervalMs)], {
      stdio: ['ignore', 'pipe', 'pipe'],
    })
    this.child.stdout.setEncoding('utf8')
    this.child.stdout.on('data', (chunk) => this._onStdout(chunk))
    this.child.stderr.on('data', (chunk) => {
      const s = String(chunk).trim()
      if (s) log.info('helper', { msg: s })
    })
    this.child.on('exit', (code) => {
      log.warn('eBPF helper exited', { code })
      this.child = null
      if (this.running) {
        this.mode = 'js'
        this.timer = setInterval(() => this._tickJs(), this.intervalMs)
        if (this.timer.unref) this.timer.unref()
      }
    })
  }

  _onStdout(chunk) {
    this.buf += chunk
    const parts = this.buf.split('\n')
    this.buf = parts.pop() || ''
    const ts = Date.now()
    /** @type {Array<{ chart: string, context: string, ts: number, values: object }>} */
    const batch = []
    for (const line of parts) {
      if (!line.trim()) continue
      try {
        const row = JSON.parse(line)
        if (!row.chart || !row.values) continue
        const chart = String(row.chart).startsWith('ebpf.')
          ? String(row.chart)
          : `ebpf.${row.chart}`
        batch.push({
          chart,
          context: chart,
          ts: row.ts || ts,
          values: row.values,
        })
      } catch {
        // ignore bad lines
      }
    }
    if (batch.length) this.emit('samples', batch)
  }

  _tickBridge(p) {
    let st
    try {
      st = fs.statSync(p)
    } catch {
      return
    }
    if (st.size < this.bridgeOffset) this.bridgeOffset = 0
    if (st.size === this.bridgeOffset) return
    try {
      const fd = fs.openSync(p, 'r')
      const len = Math.min(st.size - this.bridgeOffset, 256 * 1024)
      const buf = Buffer.alloc(len)
      const n = fs.readSync(fd, buf, 0, len, this.bridgeOffset)
      fs.closeSync(fd)
      this.bridgeOffset += n
      this._onStdout(buf.slice(0, n).toString('utf8') + '\n')
    } catch {
      // ignore
    }
  }

  _readVm() {
    /** @type {Record<string, number>} */
    const out = {}
    const raw = readFileCached('/proc/vmstat')
    if (raw) {
      for (const line of raw.split('\n')) {
        const [k, v] = line.trim().split(/\s+/)
        if (k) out[k] = Number(v) || 0
      }
    }
    return out
  }

  _tickJs() {
    const ts = Date.now()
    const dt = this.lastTs ? (ts - this.lastTs) / 1000 : this.intervalMs / 1000
    this.lastTs = ts
    const vm = this._readVm()
    const prev = this.prev
    const rate = (k) => (prev[k] != null ? Math.max(0, (vm[k] || 0) - prev[k]) / dt : 0)

    const hit = Math.max(0, (vm.pgfault || 0) - (vm.pgmajfault || 0))
    const miss = vm.pgmajfault || 0
    const dHit = prev.hit != null ? Math.max(0, hit - prev.hit) / dt : 0
    const dMiss = prev.miss != null ? Math.max(0, miss - prev.miss) / dt : 0
    const den = dHit + dMiss

    let fileNr = [0, 0]
    const fileNrRaw = readFileCached('/proc/sys/fs/file-nr')
    if (fileNrRaw) {
      const parts = fileNrRaw.trim().split(/\s+/)
      fileNr = [Number(parts[0]) || 0, Number(parts[2]) || 0]
    }

    let forks = 0
    let ctxt = 0
    const statRaw = readFileCached('/proc/stat')
    if (statRaw) {
      for (const line of statRaw.split('\n')) {
        if (line.startsWith('processes ')) forks = Number(line.slice(10)) || 0
        if (line.startsWith('ctxt ')) ctxt = Number(line.slice(5)) || 0
      }
    }

    let tcp = 0
    const sockRaw = readFileCached('/proc/net/sockstat')
    if (sockRaw) {
      const m = sockRaw.match(/TCP:\s+inuse\s+(\d+)/)
      if (m) tcp = Number(m[1]) || 0
    }

    const batch = [
      {
        chart: 'ebpf.cachestat',
        context: 'ebpf.cachestat',
        ts,
        values: {
          hits: dHit,
          misses: dMiss,
          ratio: den > 0 ? (dHit / den) * 100 : 0,
          pgpgin: rate('pgpgin'),
          pgpgout: rate('pgpgout'),
        },
      },
      {
        chart: 'ebpf.fd',
        context: 'ebpf.fd',
        ts,
        values: { open: fileNr[0], max: fileNr[1] },
      },
      {
        chart: 'ebpf.oom',
        context: 'ebpf.oom',
        ts,
        values: { kills: rate('oom_kill'), total: vm.oom_kill || 0 },
      },
      {
        chart: 'ebpf.process',
        context: 'ebpf.process',
        ts,
        values: {
          forks: prev.forks != null ? Math.max(0, forks - prev.forks) / dt : 0,
          ctxt: prev.ctxt != null ? Math.max(0, ctxt - prev.ctxt) / dt : 0,
        },
      },
      {
        chart: 'ebpf.swap',
        context: 'ebpf.swap',
        ts,
        values: { in: rate('pswpin'), out: rate('pswpout') },
      },
      {
        chart: 'ebpf.vfs',
        context: 'ebpf.vfs',
        ts,
        // pgpgin = pages read from block devices; pgpgout = pages written out
        values: { read_pages: rate('pgpgin'), write_pages: rate('pgpgout') },
      },
      {
        chart: 'ebpf.socket',
        context: 'ebpf.socket',
        ts,
        values: {
          tcp_inuse: tcp,
          tcp_delta: prev.tcp != null ? (tcp - prev.tcp) / dt : 0,
        },
      },
      {
        chart: 'ebpf.shm',
        context: 'ebpf.shm',
        ts,
        values: { segments: 0, bytes: 0 },
      },
    ]

    this.prev = { ...vm, hit, miss, forks, ctxt, tcp }
    this.emit('samples', batch)
  }
}

let singleton = null
export function getEbpfCollector() {
  if (!singleton) singleton = new EbpfCollector()
  return singleton
}