System Log

server/core/audit.js

@@ -18,11 +18,16 @@ const MUTATING = new Set([
   'handshake',
 ])
 
-function auditPath() {
+export function getAuditPath() {
   const dir = process.env.PEARDATA_DATA_DIR || path.resolve('data')
   return path.join(dir, 'audit.log')
 }
 
+/** @deprecated use getAuditPath */
+function auditPath() {
+  return getAuditPath()
+}
+
 /**
  * @param {string} method
  */

server/handlers/monitor.js

@@ -28,6 +28,7 @@ import { getCollector } from '../services/collector.js'
 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 {
   listAlerts,
   getAlert,
@@ -150,6 +151,10 @@ export function registerMonitorHandlers(session) {
   session.respond('queryData', async (args) => store.query(args), { hot: true })
   session.respond('getWeights', async (args) => computeWeights(args || {}), { hot: true })
 
+  session.respond('queryLogs', async (args) =>
+    queryLogs({ ...(args || {}), role: session.role })
+  )
+
   session.respond('getDbInfo', async () => {
     const db = getDb()
     if (!db) return { enabled: false }

server/rest/routes.js

@@ -4,7 +4,13 @@
  * Local-agent style GET endpoints for charts/data/contexts/nodes/info/allmetrics.
  */
 import os from 'os'
-import { APP_NAME, APP_VERSION, PROTOCOL, PROTOCOL_VERSION } from '../../shared/protocol.js'
+import {
+  APP_NAME,
+  APP_VERSION,
+  PROTOCOL,
+  PROTOCOL_VERSION,
+  Roles,
+} from '../../shared/protocol.js'
 import {
   getAllChartDefs,
   getContextIds,
@@ -15,6 +21,7 @@ import { getStore } from '../services/store.js'
 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 { listAlerts, getAlert } from '../services/alerts.js'
 import { getServerPublicKeyHex } from '../core/auth-keys.js'
 import { formatAllMetrics } from './formatters.js'
@@ -179,6 +186,27 @@ export async function handleRest(pathname, query) {
     }
     return json(result)
   }
+  // ── logs (anomaly / audit / journal) ─────────────────────
+  if (path === '/api/v3/logs' || path === '/api/v2/logs' || path === '/api/v1/logs') {
+    const result = await queryLogs({
+      source: query.get('source') || 'anomaly',
+      q: query.get('q') || query.get('query') || '',
+      since: query.get('since') || undefined,
+      until: query.get('until') || undefined,
+      priority: query.get('priority') || undefined,
+      unit: query.get('unit') || undefined,
+      limit: query.get('limit') || undefined,
+      cursor: query.get('cursor') || undefined,
+      // REST is localhost-open; treat as admin for audit/journal
+      role: Roles.admin,
+    })
+    if (!result.ok) {
+      const status = result.code === 'PERMISSION_DENIED' ? 403 : 400
+      return { status, contentType: 'application/json', body: JSON.stringify(result) }
+    }
+    return json(result)
+  }
+
   if (path === '/api/v3/q' || path === '/api/v2/q') {
     const q = (query.get('q') || query.get('query') || '').toLowerCase()
     const hits = getAllChartDefs()

server/services/logs.js

@@ -0,0 +1,407 @@
+/**
+ * Unified log query: anomalies, agent audit.log, optional host journalctl.
+ */
+import { spawn } from 'child_process'
+import fs from 'fs'
+import os from 'os'
+import { Roles, roleAllows } from '../../shared/protocol.js'
+import { getAnomalyEngine } from './anomaly.js'
+import { getAuditPath } from '../core/audit.js'
+
+const DEFAULT_LIMIT = 200
+const MAX_LIMIT = 2000
+const JOURNAL_TIMEOUT_MS = 3000
+
+/**
+ * @param {string} role
+ * @param {string} source
+ */
+export function assertLogSourceAllowed(role, source) {
+  const src = String(source || 'anomaly').toLowerCase()
+  if (src === 'anomaly') return { ok: true, source: src }
+  if (src === 'audit' || src === 'journal') {
+    if (!roleAllows(role || Roles.viewer, Roles.admin)) {
+      return {
+        ok: false,
+        source: src,
+        error: 'PERMISSION_DENIED',
+        code: 'PERMISSION_DENIED',
+        hint: 'Audit and journal sources require admin role',
+      }
+    }
+    return { ok: true, source: src }
+  }
+  return { ok: false, source: src, error: `unknown source: ${src}` }
+}
+
+/**
+ * @param {unknown} n
+ * @param {number} fallback
+ */
+export function clampLimit(n, fallback = DEFAULT_LIMIT) {
+  const v = Number(n)
+  if (!Number.isFinite(v) || v < 1) return fallback
+  return Math.min(MAX_LIMIT, Math.floor(v))
+}
+
+/**
+ * @param {string|number|undefined|null} v
+ * @returns {number|null} unix ms
+ */
+export function parseTimeBound(v) {
+  if (v == null || v === '') return null
+  if (typeof v === 'number' && Number.isFinite(v)) {
+    // seconds vs ms heuristic
+    return v < 1e12 ? Math.floor(v * 1000) : Math.floor(v)
+  }
+  const s = String(v).trim()
+  if (!s) return null
+  // relative like -3600 or -1h
+  const rel = s.match(/^(-?\d+)([smhd])?$/i)
+  if (rel) {
+    const n = Number(rel[1])
+    const unit = (rel[2] || 's').toLowerCase()
+    const mult =
+      unit === 'm' ? 60_000 : unit === 'h' ? 3_600_000 : unit === 'd' ? 86_400_000 : 1000
+    return Date.now() + n * mult
+  }
+  const ms = Date.parse(s)
+  return Number.isFinite(ms) ? ms : null
+}
+
+/**
+ * @param {{ q?: string, since?: number|null, until?: number|null }} filters
+ * @param {{ ts: number, message: string, unit?: string, severity?: string }} row
+ */
+export function matchesFilters(filters, row) {
+  const q = String(filters.q || '')
+    .trim()
+    .toLowerCase()
+  if (q) {
+    const hay = `${row.message || ''} ${row.unit || ''} ${row.severity || ''}`.toLowerCase()
+    if (!hay.includes(q)) return false
+  }
+  if (filters.since != null && row.ts < filters.since) return false
+  if (filters.until != null && row.ts > filters.until) return false
+  return true
+}
+
+/**
+ * @param {object} ev
+ * @param {number} index
+ */
+export function normalizeAnomaly(ev, index = 0) {
+  const ts = Number(ev.ts) || Date.now()
+  return {
+    id: `anomaly:${ts}:${ev.chart || index}`,
+    ts,
+    source: 'anomaly',
+    severity: String(ev.severity || (ev.cleared ? 'ok' : 'warning')),
+    unit: String(ev.chart || ev.context || ''),
+    message: String(ev.message || ''),
+    fields: {
+      chart: ev.chart || null,
+      context: ev.context || null,
+      score: ev.score ?? null,
+      cleared: Boolean(ev.cleared),
+    },
+  }
+}
+
+/**
+ * @param {object} entry
+ * @param {number} index
+ */
+export function normalizeAudit(entry, index = 0) {
+  const ts = entry.ts ? Date.parse(entry.ts) || Date.now() : Date.now()
+  const ok = entry.ok !== false
+  const msg = [
+    entry.method || 'rpc',
+    ok ? 'ok' : 'fail',
+    entry.peerId ? `peer=${entry.peerId}` : '',
+    entry.role ? `role=${entry.role}` : '',
+    entry.error ? `error=${entry.error}` : '',
+  ]
+    .filter(Boolean)
+    .join(' ')
+  return {
+    id: `audit:${ts}:${index}`,
+    ts,
+    source: 'audit',
+    severity: ok ? 'info' : 'warning',
+    unit: String(entry.method || 'rpc'),
+    message: msg,
+    fields: {
+      peerId: entry.peerId || null,
+      role: entry.role || null,
+      ok,
+      error: entry.error || null,
+    },
+  }
+}
+
+/**
+ * @param {object} j
+ * @param {number} index
+ */
+export function normalizeJournal(j, index = 0) {
+  const usec = Number(j.__REALTIME_TIMESTAMP || j.TIMESTAMP || 0)
+  const ts = usec > 1e15 ? Math.floor(usec / 1000) : usec > 1e12 ? Math.floor(usec) : Date.now()
+  const priority = j.PRIORITY != null ? String(j.PRIORITY) : ''
+  const sevMap = {
+    0: 'emerg',
+    1: 'alert',
+    2: 'crit',
+    3: 'err',
+    4: 'warning',
+    5: 'notice',
+    6: 'info',
+    7: 'debug',
+  }
+  const severity = sevMap[priority] || priority || 'info'
+  const unit = String(j._SYSTEMD_UNIT || j.SYSLOG_IDENTIFIER || j._COMM || '')
+  const message = String(j.MESSAGE || j.message || '')
+  return {
+    id: `journal:${j.__CURSOR || ts}:${index}`,
+    ts,
+    source: 'journal',
+    severity,
+    unit,
+    message,
+    fields: {
+      priority: priority || null,
+      cursor: j.__CURSOR || null,
+      pid: j._PID || null,
+    },
+  }
+}
+
+/**
+ * Build journalctl argv (no shell). Exported for tests.
+ * @param {{
+ *   limit?: number,
+ *   sinceMs?: number|null,
+ *   untilMs?: number|null,
+ *   priority?: string,
+ *   unit?: string,
+ *   q?: string,
+ * }} opts
+ */
+export function buildJournalArgv(opts = {}) {
+  const limit = clampLimit(opts.limit)
+  const args = ['--output=json', '--no-pager', '-n', String(limit)]
+  if (opts.sinceMs != null) {
+    args.push('--since', new Date(opts.sinceMs).toISOString())
+  }
+  if (opts.untilMs != null) {
+    args.push('--until', new Date(opts.untilMs).toISOString())
+  }
+  if (opts.priority != null && opts.priority !== '') {
+    args.push('-p', String(opts.priority))
+  }
+  if (opts.unit) {
+    args.push('-u', String(opts.unit))
+  }
+  const q = String(opts.q || '').trim()
+  if (q) {
+    args.push('--grep', q)
+  }
+  return args
+}
+
+export function isJournalEnabled() {
+  const v = String(process.env.PEARDATA_JOURNAL || '').trim().toLowerCase()
+  return v === '1' || v === 'true' || v === 'yes' || v === 'on'
+}
+
+/**
+ * @param {string} file
+ * @param {number} maxBytes
+ */
+export function readAuditFileLines(file, maxBytes = 512_000) {
+  if (!fs.existsSync(file)) return []
+  const stat = fs.statSync(file)
+  const start = Math.max(0, stat.size - maxBytes)
+  const buf = Buffer.alloc(stat.size - start)
+  const fd = fs.openSync(file, 'r')
+  try {
+    fs.readSync(fd, buf, 0, buf.length, start)
+  } finally {
+    fs.closeSync(fd)
+  }
+  const text = buf.toString('utf8')
+  const lines = text.split('\n').filter(Boolean)
+  // If we started mid-line, drop first fragment
+  if (start > 0 && lines.length) lines.shift()
+  /** @type {object[]} */
+  const parsed = []
+  for (const line of lines) {
+    try {
+      parsed.push(JSON.parse(line))
+    } catch {
+      // skip corrupt
+    }
+  }
+  return parsed
+}
+
+/**
+ * @param {object} args
+ * @param {{ anomalies?: { listRecent: (n: number) => object[] }, spawnJournal?: Function }} [deps]
+ */
+export async function queryLogs(args = {}, deps = {}) {
+  const gate = assertLogSourceAllowed(args.role || Roles.viewer, args.source || 'anomaly')
+  if (!gate.ok) {
+    return {
+      ok: false,
+      error: gate.error,
+      code: gate.code || undefined,
+      hint: gate.hint || undefined,
+      source: gate.source,
+      entries: [],
+    }
+  }
+  const source = gate.source
+  const limit = clampLimit(args.limit)
+  const cursor = Math.max(0, Math.floor(Number(args.cursor) || 0))
+  const since = parseTimeBound(args.since)
+  const until = parseTimeBound(args.until)
+  const q = String(args.q || '').trim()
+  const filters = { q, since, until }
+
+  if (source === 'anomaly') {
+    const engine = deps.anomalies || getAnomalyEngine()
+    const raw = engine.listRecent(Math.min(MAX_LIMIT, limit + cursor + 50))
+    const all = raw
+      .map((ev, i) => normalizeAnomaly(ev, i))
+      .filter((row) => matchesFilters(filters, row))
+    const slice = all.slice(cursor, cursor + limit)
+    return {
+      ok: true,
+      source,
+      entries: slice,
+      nextCursor: cursor + slice.length < all.length ? cursor + slice.length : null,
+      stats: { matched: all.length, returned: slice.length },
+    }
+  }
+
+  if (source === 'audit') {
+    const file = args.auditPath || getAuditPath()
+    const raw = readAuditFileLines(file).reverse()
+    const all = raw
+      .map((entry, i) => normalizeAudit(entry, i))
+      .filter((row) => matchesFilters(filters, row))
+    const slice = all.slice(cursor, cursor + limit)
+    return {
+      ok: true,
+      source,
+      entries: slice,
+      nextCursor: cursor + slice.length < all.length ? cursor + slice.length : null,
+      stats: { matched: all.length, returned: slice.length },
+    }
+  }
+
+  // journal
+  if (!isJournalEnabled()) {
+    return {
+      ok: false,
+      source,
+      error: 'journal_disabled',
+      hint: 'Set PEARDATA_JOURNAL=1 and ensure the agent user can read the journal (e.g. SupplementaryGroups=systemd-journal)',
+      entries: [],
+    }
+  }
+  if (os.platform() !== 'linux') {
+    return {
+      ok: false,
+      source,
+      error: 'unsupported',
+      hint: 'Host journal is only available on Linux agents',
+      entries: [],
+    }
+  }
+
+  const argv = buildJournalArgv({
+    limit,
+    sinceMs: since,
+    untilMs: until,
+    priority: args.priority,
+    unit: args.unit,
+    q,
+  })
+
+  const spawnFn = deps.spawnJournal || spawnJournalctl
+  try {
+    const { stdout, stderr, code } = await spawnFn(argv)
+    if (code !== 0 && !stdout.trim()) {
+      return {
+        ok: false,
+        source,
+        error: 'journalctl_failed',
+        hint: stderr.trim() || `journalctl exited ${code}`,
+        entries: [],
+      }
+    }
+    const entries = []
+    const lines = stdout.split('\n').filter(Boolean)
+    for (let i = 0; i < lines.length; i++) {
+      try {
+        const j = JSON.parse(lines[i])
+        const row = normalizeJournal(j, i)
+        // --grep already applied when q set; still apply since/until if journal ignored them
+        if (matchesFilters({ q: '', since, until }, row)) entries.push(row)
+      } catch {
+        // skip
+      }
+    }
+    // journalctl returns oldest-first typically; present newest first
+    entries.reverse()
+    return {
+      ok: true,
+      source,
+      entries: entries.slice(0, limit),
+      nextCursor: null,
+      stats: { matched: entries.length, returned: Math.min(limit, entries.length) },
+    }
+  } catch (err) {
+    return {
+      ok: false,
+      source,
+      error: 'journalctl_failed',
+      hint: err?.message || String(err),
+      entries: [],
+    }
+  }
+}
+
+/**
+ * @param {string[]} argv
+ * @returns {Promise<{ stdout: string, stderr: string, code: number }>}
+ */
+function spawnJournalctl(argv) {
+  return new Promise((resolve, reject) => {
+    const child = spawn('journalctl', argv, {
+      stdio: ['ignore', 'pipe', 'pipe'],
+    })
+    let stdout = ''
+    let stderr = ''
+    const timer = setTimeout(() => {
+      child.kill('SIGKILL')
+      reject(new Error(`journalctl timed out after ${JOURNAL_TIMEOUT_MS}ms`))
+    }, JOURNAL_TIMEOUT_MS)
+    child.stdout.on('data', (d) => {
+      stdout += d.toString('utf8')
+    })
+    child.stderr.on('data', (d) => {
+      stderr += d.toString('utf8')
+    })
+    child.on('error', (err) => {
+      clearTimeout(timer)
+      reject(err)
+    })
+    child.on('close', (code) => {
+      clearTimeout(timer)
+      resolve({ stdout, stderr, code: code ?? 1 })
+    })
+  })
+}