Add process view

server/handlers/monitor.js

@@ -29,6 +29,7 @@ import { getStore } from '../services/store.js'
 import { getAnomalyEngine } from '../services/anomaly.js'
 import { computeWeights } from '../services/weights.js'
 import { queryLogs } from '../services/logs.js'
+import { listProcesses } from '../services/processes.js'
 import {
   listAlerts,
   getAlert,
@@ -161,6 +162,10 @@ export function registerMonitorHandlers(session) {
     queryLogs({ ...(args || {}), role: session.role })
   )
 
+  session.respond('listProcesses', async (args) => listProcesses(args || {}), {
+    hot: true,
+  })
+
   session.respond('getDbInfo', async () => {
     const db = getDb()
     if (!db) return { enabled: false }

server/rest/routes.js

@@ -22,6 +22,7 @@ import { getCollector } from '../services/collector.js'
 import { getAnomalyEngine } from '../services/anomaly.js'
 import { computeWeights } from '../services/weights.js'
 import { queryLogs } from '../services/logs.js'
+import { listProcesses } from '../services/processes.js'
 import { listAlerts, getAlert } from '../services/alerts.js'
 import { getServerPublicKeyHex } from '../core/auth-keys.js'
 import { formatAllMetrics } from './formatters.js'
@@ -186,6 +187,24 @@ export async function handleRest(pathname, query) {
     }
     return json(result)
   }
+  // ── processes (live /proc table) ─────────────────────────
+  if (
+    path === '/api/v3/processes' ||
+    path === '/api/v2/processes' ||
+    path === '/api/v1/processes'
+  ) {
+    const result = listProcesses({
+      q: query.get('q') || query.get('query') || '',
+      sort: query.get('sort') || undefined,
+      order: query.get('order') || undefined,
+      filter: query.get('filter') || undefined,
+      limit: query.get('limit') || undefined,
+      offset: query.get('offset') || undefined,
+      pid: query.get('pid') || undefined,
+    })
+    return json(result)
+  }
+
   // ── logs (anomaly / audit / journal) ─────────────────────
   if (path === '/api/v3/logs' || path === '/api/v2/logs' || path === '/api/v1/logs') {
     const result = await queryLogs({

server/services/collectors/processes.js

@@ -23,7 +23,10 @@ const log = logger.child('processes')
 
 export function isProcessCollectorEnabled() {
   const v = process.env.PEARDATA_PROCESSES
-  return v === '1' || v === 'on' || v === 'true'
+  if (v === '0' || v === 'off' || v === 'false') return false
+  if (v === '1' || v === 'on' || v === 'true') return true
+  // Default on for Linux — powers Charts top-* series alongside the Processes tab
+  return os.platform() === 'linux'
 }
 
 function topN() {

server/services/processes.js

@@ -0,0 +1,672 @@
+/**
+ * Live process monitor — Linux /proc snapshot for the Processes tab + RPC.
+ *
+ * Always available on Linux (independent of PEARDATA_PROCESSES chart collector).
+ * Caches successive samples so CPU / I/O rates stay meaningful across polls.
+ */
+import fs from 'fs'
+import path from 'path'
+import os from 'os'
+
+function chartsCollectorEnabled() {
+  const v = process.env.PEARDATA_PROCESSES
+  if (v === '0' || v === 'off' || v === 'false') return false
+  if (v === '1' || v === 'on' || v === 'true') return true
+  return os.platform() === 'linux'
+}
+
+const USER_HZ = 100
+const LIST_CACHE_MS = 450
+const PASSWD_CACHE_MS = 60_000
+const MAX_LIMIT = 2000
+const DEFAULT_LIMIT = 250
+
+const SORTS = new Set(['cpu', 'rss', 'pid', 'name', 'io', 'threads', 'age', 'fds', 'state'])
+const FILTERS = new Set([
+  'all',
+  'running',
+  'sleeping',
+  'zombie',
+  'stopped',
+  'highcpu',
+  'highmem',
+  'hasio',
+  'kernel',
+  'user',
+])
+
+/** @type {Map<string, string>} */
+let passwdMap = new Map()
+let passwdAt = 0
+
+/** @type {{ ts: number, wallMs: number, byPid: Map<number, SamplePrev>, rows: object[], summary: object }|null} */
+let cache = null
+/** @type {Map<number, SamplePrev>} */
+let prevByPid = new Map()
+let prevWallMs = 0
+
+/**
+ * @typedef {{
+ *   ticks: number,
+ *   utime: number,
+ *   stime: number,
+ *   readBytes: number,
+ *   writeBytes: number,
+ *   wallMs: number,
+ * }} SamplePrev
+ */
+
+function pageSize() {
+  try {
+    return os.constants?.os?.PAGE_SIZE || 4096
+  } catch {
+    return 4096
+  }
+}
+
+function hostCpus() {
+  try {
+    return os.cpus().length || 1
+  } catch {
+    return 1
+  }
+}
+
+function loadavg1() {
+  try {
+    return os.loadavg()?.[0] ?? null
+  } catch {
+    return null
+  }
+}
+
+function readFile(p) {
+  try {
+    return fs.readFileSync(p, 'utf8')
+  } catch {
+    return null
+  }
+}
+
+function readLink(p) {
+  try {
+    return fs.readlinkSync(p)
+  } catch {
+    return null
+  }
+}
+
+function refreshPasswd() {
+  const now = Date.now()
+  if (now - passwdAt < PASSWD_CACHE_MS && passwdMap.size) return
+  passwdAt = now
+  /** @type {Map<string, string>} */
+  const map = new Map()
+  const raw = readFile('/etc/passwd')
+  if (raw) {
+    for (const line of raw.split('\n')) {
+      if (!line || line.startsWith('#')) continue
+      const parts = line.split(':')
+      if (parts.length < 3) continue
+      map.set(String(parts[2]), parts[0])
+    }
+  }
+  passwdMap = map
+}
+
+function uidName(uid) {
+  if (uid == null) return ''
+  refreshPasswd()
+  return passwdMap.get(String(uid)) || String(uid)
+}
+
+function bootTimeMs() {
+  const up = readFile('/proc/uptime')
+  if (!up) return Date.now() - os.uptime() * 1000
+  const sec = Number(up.trim().split(/\s+/)[0])
+  if (!Number.isFinite(sec)) return Date.now() - os.uptime() * 1000
+  return Date.now() - sec * 1000
+}
+
+function shortCgroup(raw) {
+  if (!raw) return ''
+  const line = raw.trim().split('\n').pop() || ''
+  // cgroup v2: "0::/system.slice/foo.service"
+  const pathPart = line.includes('::') ? line.split('::')[1] : line.split(':').pop()
+  const s = String(pathPart || '').replace(/^\//, '')
+  if (!s || s === '.') return ''
+  const bits = s.split('/')
+  return bits.slice(-2).join('/').slice(0, 80)
+}
+
+function stateLabel(ch) {
+  switch (ch) {
+    case 'R':
+      return 'running'
+    case 'S':
+      return 'sleeping'
+    case 'D':
+      return 'disk-sleep'
+    case 'Z':
+      return 'zombie'
+    case 'T':
+    case 't':
+      return 'stopped'
+    case 'I':
+      return 'idle'
+    default:
+      return ch ? String(ch) : 'unknown'
+  }
+}
+
+/**
+ * Parse one /proc/[pid]/stat line.
+ * @param {string} raw
+ */
+export function parseProcStat(raw) {
+  const open = raw.indexOf('(')
+  const close = raw.lastIndexOf(')')
+  if (open < 0 || close < open) return null
+  const name = raw.slice(open + 1, close)
+  const rest = raw.slice(close + 2).split(/\s+/)
+  // After comm: state ppid … utime stime … priority nice num_threads … starttime vsize rss pin …
+  return {
+    name,
+    state: rest[0] || '?',
+    ppid: Number(rest[1]) || 0,
+    utime: Number(rest[11]) || 0,
+    stime: Number(rest[12]) || 0,
+    priority: Number(rest[15]) || 0,
+    nice: Number(rest[16]) || 0,
+    numThreads: Number(rest[17]) || 0,
+    starttime: Number(rest[19]) || 0,
+    vsize: Number(rest[20]) || 0,
+    rssPages: Number(rest[21]) || 0,
+    processor: Number(rest[36]) || 0,
+  }
+}
+
+/**
+ * @param {string} statusRaw
+ */
+export function parseProcStatus(statusRaw) {
+  /** @type {Record<string, string>} */
+  const m = {}
+  for (const line of String(statusRaw || '').split('\n')) {
+    const i = line.indexOf(':')
+    if (i < 0) continue
+    m[line.slice(0, i)] = line.slice(i + 1).trim()
+  }
+  const uid = (m.Uid || '').split(/\s+/)[0]
+  const gid = (m.Gid || '').split(/\s+/)[0]
+  return {
+    uid: uid != null && uid !== '' ? Number(uid) : null,
+    gid: gid != null && gid !== '' ? Number(gid) : null,
+    threads: Number(m.Threads) || 0,
+    vmRssKb: Number((m.VmRSS || '').split(/\s+/)[0]) || 0,
+    vmSizeKb: Number((m.VmSize || '').split(/\s+/)[0]) || 0,
+    vmSwapKb: Number((m.VmSwap || '').split(/\s+/)[0]) || 0,
+    voluntaryCtxt: Number(m.voluntary_ctxt_switches) || 0,
+    nonvoluntaryCtxt: Number(m.nonvoluntary_ctxt_switches) || 0,
+  }
+}
+
+/**
+ * Full /proc scrape with rates (relative to previous sample).
+ * @param {{ detailPid?: number|null }} [opts]
+ */
+export function scanProcesses(opts = {}) {
+  if (os.platform() !== 'linux') {
+    return {
+      supported: false,
+      platform: os.platform(),
+      ts: Date.now(),
+      rows: [],
+      summary: emptySummary(),
+    }
+  }
+
+  let dirs
+  try {
+    dirs = fs.readdirSync('/proc')
+  } catch {
+    return {
+      supported: false,
+      platform: 'linux',
+      ts: Date.now(),
+      rows: [],
+      summary: emptySummary(),
+      error: 'cannot read /proc',
+    }
+  }
+
+  const pageBytes = pageSize()
+  const ncpu = hostCpus()
+  const bootMs = bootTimeMs()
+  const wallMs = Date.now()
+  const detailPid = opts.detailPid != null ? Number(opts.detailPid) : null
+  const dWallSec = prevWallMs && wallMs > prevWallMs ? (wallMs - prevWallMs) / 1000 : 0
+
+  /** @type {object[]} */
+  const rows = []
+  /** @type {Map<number, SamplePrev>} */
+  const nextPrev = new Map()
+
+  let running = 0
+  let sleeping = 0
+  let zombie = 0
+  let stopped = 0
+  let idle = 0
+  let diskSleep = 0
+  let totalCpu = 0
+  let totalRss = 0
+  let totalThreads = 0
+
+  for (const ent of dirs) {
+    if (!/^\d+$/.test(ent)) continue
+    const pid = Number(ent)
+    const base = path.join('/proc', ent)
+    const statRaw = readFile(path.join(base, 'stat'))
+    if (!statRaw) continue
+    const st = parseProcStat(statRaw)
+    if (!st) continue
+
+    const statusRaw = readFile(path.join(base, 'status'))
+    const status = statusRaw ? parseProcStatus(statusRaw) : null
+
+    let readBytes = null
+    let writeBytes = null
+    const ioRaw = readFile(path.join(base, 'io'))
+    if (ioRaw) {
+      const rb = ioRaw.match(/^read_bytes:\s*(\d+)/m)
+      const wb = ioRaw.match(/^write_bytes:\s*(\d+)/m)
+      if (rb) readBytes = Number(rb[1]) || 0
+      if (wb) writeBytes = Number(wb[1]) || 0
+    }
+
+    const ticks = st.utime + st.stime
+    const prev = prevByPid.get(pid)
+    let cpuPct = 0
+    let cpuUserPct = 0
+    let cpuSystemPct = 0
+    let ioReadKiBs = 0
+    let ioWriteKiBs = 0
+    if (prev && dWallSec > 0) {
+      const dTicks = ticks - prev.ticks
+      const dUser = st.utime - prev.utime
+      const dSys = st.stime - prev.stime
+      if (dTicks >= 0) {
+        cpuPct = Math.min(100 * ncpu, (dTicks / USER_HZ / dWallSec) * 100)
+        cpuUserPct = Math.min(100 * ncpu, (dUser / USER_HZ / dWallSec) * 100)
+        cpuSystemPct = Math.min(100 * ncpu, (dSys / USER_HZ / dWallSec) * 100)
+      }
+      if (readBytes != null && writeBytes != null) {
+        ioReadKiBs = Math.max(0, readBytes - prev.readBytes) / dWallSec / 1024
+        ioWriteKiBs = Math.max(0, writeBytes - prev.writeBytes) / dWallSec / 1024
+      }
+    }
+
+    nextPrev.set(pid, {
+      ticks,
+      utime: st.utime,
+      stime: st.stime,
+      readBytes: readBytes ?? 0,
+      writeBytes: writeBytes ?? 0,
+      wallMs,
+    })
+
+    const rssMiB =
+      status?.vmRssKb != null && status.vmRssKb > 0
+        ? status.vmRssKb / 1024
+        : (st.rssPages * pageBytes) / (1024 * 1024)
+    const vszMiB =
+      status?.vmSizeKb != null && status.vmSizeKb > 0
+        ? status.vmSizeKb / 1024
+        : st.vsize / (1024 * 1024)
+    const startMs = bootMs + (st.starttime / USER_HZ) * 1000
+    const ageSec = Math.max(0, (wallMs - startMs) / 1000)
+    const state = stateLabel(st.state)
+    const threads = status?.threads || st.numThreads || 0
+    const uid = status?.uid
+    const user = uidName(uid)
+
+    let cmdline = ''
+    const cmdRaw = readFile(path.join(base, 'cmdline'))
+    if (cmdRaw) {
+      cmdline = cmdRaw.replace(/\0/g, ' ').trim()
+    }
+    if (!cmdline) cmdline = st.name
+
+    const cgroup = shortCgroup(readFile(path.join(base, 'cgroup')) || '')
+
+    /** @type {object} */
+    const row = {
+      pid,
+      ppid: st.ppid,
+      name: st.name,
+      cmdline: cmdline.slice(0, 256),
+      state: st.state,
+      stateLabel: state,
+      user,
+      uid,
+      gid: status?.gid ?? null,
+      cpu: round(cpuPct, 2),
+      cpuUser: round(cpuUserPct, 2),
+      cpuSystem: round(cpuSystemPct, 2),
+      mem: round(rssMiB, 2),
+      vsz: round(vszMiB, 2),
+      swap: round((status?.vmSwapKb || 0) / 1024, 2),
+      threads,
+      nice: st.nice,
+      priority: st.priority,
+      ioRead: round(ioReadKiBs, 2),
+      ioWrite: round(ioWriteKiBs, 2),
+      io: round(ioReadKiBs + ioWriteKiBs, 2),
+      fds: null,
+      startMs: Math.round(startMs),
+      ageSec: Math.round(ageSec),
+      cgroup,
+      processor: st.processor,
+      kernel:
+        pid === 2 ||
+        st.ppid === 2 ||
+        /^(kthreadd|kworker|ksoftirq|kswapd|migration|rcu_|cpuhp|kdevtmpfs)/.test(st.name),
+    }
+
+    if (detailPid === pid) {
+      row.cmdline = cmdline.slice(0, 4096)
+      row.exe = readLink(path.join(base, 'exe'))
+      row.cwd = readLink(path.join(base, 'cwd'))
+      row.wchan = (readFile(path.join(base, 'wchan')) || '').trim().slice(0, 64)
+      try {
+        row.fds = fs.readdirSync(path.join(base, 'fd')).length
+      } catch {
+        row.fds = null
+      }
+      row.voluntaryCtxt = status?.voluntaryCtxt ?? 0
+      row.nonvoluntaryCtxt = status?.nonvoluntaryCtxt ?? 0
+      row.detail = true
+    }
+
+    rows.push(row)
+
+    if (state === 'running') running++
+    else if (state === 'sleeping') sleeping++
+    else if (state === 'zombie') zombie++
+    else if (state === 'stopped') stopped++
+    else if (state === 'idle') idle++
+    else if (state === 'disk-sleep') diskSleep++
+
+    totalCpu += cpuPct
+    totalRss += rssMiB
+    totalThreads += threads
+  }
+
+  prevByPid = nextPrev
+  prevWallMs = wallMs
+
+  const summary = {
+    total: rows.length,
+    running,
+    sleeping,
+    idle,
+    zombie,
+    stopped,
+    diskSleep,
+    totalCpuPct: round(Math.min(100 * ncpu, totalCpu), 2),
+    totalRssMiB: round(totalRss, 1),
+    totalThreads,
+    ncpu,
+    load1: loadavg1() != null ? round(loadavg1(), 2) : null,
+    pageBytes,
+  }
+
+  return {
+    supported: true,
+    platform: 'linux',
+    ts: wallMs,
+    rows,
+    summary,
+  }
+}
+
+function emptySummary() {
+  return {
+    total: 0,
+    running: 0,
+    sleeping: 0,
+    idle: 0,
+    zombie: 0,
+    stopped: 0,
+    diskSleep: 0,
+    totalCpuPct: 0,
+    totalRssMiB: 0,
+    totalThreads: 0,
+    ncpu: hostCpus(),
+    load1: loadavg1() != null ? round(loadavg1(), 2) : null,
+    pageBytes: pageSize(),
+  }
+}
+
+function round(n, d) {
+  const f = 10 ** d
+  return Math.round(Number(n) * f) / f
+}
+
+/**
+ * @param {object} row
+ * @param {string} q
+ */
+export function processMatchesQuery(row, q) {
+  const needle = String(q || '')
+    .trim()
+    .toLowerCase()
+  if (!needle) return true
+  const hay = [
+    row.pid,
+    row.ppid,
+    row.name,
+    row.cmdline,
+    row.user,
+    row.uid,
+    row.stateLabel,
+    row.cgroup,
+    row.exe,
+  ]
+    .filter((x) => x != null && x !== '')
+    .join(' ')
+    .toLowerCase()
+  return hay.includes(needle)
+}
+
+/**
+ * @param {object} row
+ * @param {string} filter
+ */
+export function processMatchesFilter(row, filter) {
+  switch (filter) {
+    case 'running':
+      return row.stateLabel === 'running'
+    case 'sleeping':
+      return row.stateLabel === 'sleeping' || row.stateLabel === 'idle'
+    case 'zombie':
+      return row.stateLabel === 'zombie'
+    case 'stopped':
+      return row.stateLabel === 'stopped'
+    case 'highcpu':
+      return Number(row.cpu) >= 5
+    case 'highmem':
+      return Number(row.mem) >= 100
+    case 'hasio':
+      return Number(row.io) > 0.5
+    case 'kernel':
+      return Boolean(row.kernel)
+    case 'user':
+      return !row.kernel
+    default:
+      return true
+  }
+}
+
+/**
+ * @param {object[]} rows
+ * @param {string} sort
+ * @param {string} order
+ */
+export function sortProcesses(rows, sort, order) {
+  const key = SORTS.has(sort) ? sort : 'cpu'
+  const dir = order === 'asc' ? 1 : -1
+  const val = (r) => {
+    switch (key) {
+      case 'pid':
+        return r.pid
+      case 'name':
+        return String(r.name || '').toLowerCase()
+      case 'rss':
+        return Number(r.mem) || 0
+      case 'io':
+        return Number(r.io) || 0
+      case 'threads':
+        return Number(r.threads) || 0
+      case 'age':
+        return Number(r.ageSec) || 0
+      case 'fds':
+        return Number(r.fds) || 0
+      case 'state':
+        return String(r.stateLabel || '')
+      case 'cpu':
+      default:
+        return Number(r.cpu) || 0
+    }
+  }
+  return [...rows].sort((a, b) => {
+    const av = val(a)
+    const bv = val(b)
+    if (typeof av === 'string' && typeof bv === 'string') {
+      return av.localeCompare(bv) * dir
+    }
+    return (Number(av) - Number(bv)) * dir
+  })
+}
+
+/**
+ * Public RPC / REST entry.
+ * @param {object} [args]
+ */
+export function listProcesses(args = {}) {
+  const sort = SORTS.has(String(args.sort || '').toLowerCase())
+    ? String(args.sort).toLowerCase()
+    : 'cpu'
+  const order = String(args.order || 'desc').toLowerCase() === 'asc' ? 'asc' : 'desc'
+  let limit = args.limit == null ? DEFAULT_LIMIT : Number(args.limit)
+  if (!Number.isFinite(limit) || limit < 1) limit = DEFAULT_LIMIT
+  limit = Math.min(MAX_LIMIT, Math.floor(limit))
+  let offset = args.offset == null ? 0 : Number(args.offset)
+  if (!Number.isFinite(offset) || offset < 0) offset = 0
+  offset = Math.floor(offset)
+  const filter = FILTERS.has(String(args.filter || '').toLowerCase())
+    ? String(args.filter).toLowerCase()
+    : 'all'
+  const q = args.q != null ? String(args.q) : ''
+  const detailPid =
+    args.pid != null && Number.isFinite(Number(args.pid)) ? Number(args.pid) : null
+
+  const now = Date.now()
+  const needDetail = detailPid != null
+  const cacheOk =
+    cache &&
+    !needDetail &&
+    now - cache.ts < LIST_CACHE_MS &&
+    cache.rows?.length
+
+  let snap
+  if (cacheOk) {
+    snap = cache
+  } else {
+    const scanned = scanProcesses({ detailPid })
+    // Warm rates: first scan after boot has zeros — keep prev and allow second call
+    snap = {
+      ts: scanned.ts,
+      wallMs: scanned.ts,
+      byPid: prevByPid,
+      rows: scanned.rows,
+      summary: scanned.summary,
+      supported: scanned.supported,
+      platform: scanned.platform,
+      error: scanned.error,
+    }
+    if (!needDetail) cache = snap
+  }
+
+  if (!snap.supported) {
+    return {
+      ok: true,
+      supported: false,
+      platform: snap.platform || os.platform(),
+      ts: Date.now(),
+      summary: snap.summary || emptySummary(),
+      processes: [],
+      totalMatched: 0,
+      truncated: false,
+      chartsEnabled: chartsCollectorEnabled(),
+      error: snap.error || 'Process listing requires Linux /proc',
+    }
+  }
+
+  let matched = snap.rows.filter(
+    (r) => processMatchesFilter(r, filter) && processMatchesQuery(r, q)
+  )
+  if (detailPid != null) {
+    matched = matched.filter((r) => r.pid === detailPid)
+    // If detail pid missing from cache path, force rescan
+    if (!matched.length || !matched[0].detail) {
+      const scanned = scanProcesses({ detailPid })
+      matched = scanned.rows.filter((r) => r.pid === detailPid)
+      snap = {
+        ...snap,
+        rows: scanned.rows,
+        summary: scanned.summary,
+        ts: scanned.ts,
+      }
+    }
+  }
+
+  const sorted = sortProcesses(matched, sort, order)
+  const slice = sorted.slice(offset, offset + limit)
+  const topCpu = sortProcesses(snap.rows, 'cpu', 'desc')
+    .slice(0, 5)
+    .map((r) => ({ pid: r.pid, name: r.name, cpu: r.cpu, mem: r.mem }))
+  const topMem = sortProcesses(snap.rows, 'rss', 'desc')
+    .slice(0, 5)
+    .map((r) => ({ pid: r.pid, name: r.name, cpu: r.cpu, mem: r.mem }))
+
+  return {
+    ok: true,
+    supported: true,
+    platform: 'linux',
+    ts: snap.ts,
+    summary: snap.summary,
+    processes: slice,
+    totalMatched: matched.length,
+    total: snap.summary.total,
+    offset,
+    limit,
+    sort,
+    order,
+    filter,
+    q,
+    truncated: offset + slice.length < matched.length,
+    chartsEnabled: chartsCollectorEnabled(),
+    topCpu,
+    topMem,
+  }
+}
+
+/** Test helper — reset rate / cache state. */
+export function _resetProcessMonitorForTests() {
+  cache = null
+  prevByPid = new Map()
+  prevWallMs = 0
+  passwdMap = new Map()
+  passwdAt = 0
+}