← back to Cncp Failure Collector
index.js
426 lines
#!/usr/bin/env node
// CNCP failure collector — tick 1 (data only, no Hermes classifier).
//
// What it does on each cycle (every POLL_INTERVAL_MS):
// 1. `pm2 jlist` -> any process that flipped to status != online since last cycle,
// or whose restart_time bumped (= crash + auto-restart loop), is reported.
// 2. Scans ~/.pm2/logs/<name>-error.log + ~/Library/Logs/com.steve.*.err.log
// for new bytes since last cycle, posts the new chunk if non-empty.
// 3. POSTs to CNCP at http://127.0.0.1:3333/api/failures. CNCP dedupes within 60s.
//
// State (last restart counts, last error-log byte offsets) lives in state.json
// in this directory, so a daemon restart doesn't flood CNCP with re-reported events.
//
// Best-practices flags addressed:
// - pm2 poll, not pm2 logs --json stream (drops lines under volume, schema-volatile)
// - per-process dedup (server-side 60s window + here: skip if restart_time unchanged)
// - fail-soft: every step in try/catch; one bad cycle doesn't kill the daemon
// - no Anthropic SDK; no ANTHROPIC_API_KEY env var read; CNCP is local
const fs = require('fs');
const path = require('path');
const os = require('os');
const { execSync, exec } = require('child_process');
const HOST = process.env.COLLECTOR_HOST || os.hostname();
const PLATFORM = os.platform(); // 'darwin' on Mac2, 'linux' on Kamatera
const IS_LINUX = PLATFORM === 'linux';
const CNCP_URL = process.env.CNCP_URL || 'http://127.0.0.1:3333';
const POLL_INTERVAL_MS = parseInt(process.env.POLL_INTERVAL_MS, 10) || 30_000;
const CLASSIFIER_ENABLED = (process.env.CLASSIFIER_ENABLED || '1') !== '0';
const CLASSIFY_RATE_LIMIT_MS = 5 * 60 * 1000;
const CLASSIFY_MAP_TTL_MS = 24 * 60 * 60 * 1000; // tick 7: prune entries older than 24h
const lastClassify = new Map(); // processName -> timestamp
function pruneClassifyMap() {
const cutoff = Date.now() - CLASSIFY_MAP_TTL_MS;
let dropped = 0;
for (const [name, ts] of lastClassify.entries()) {
if (ts < cutoff) { lastClassify.delete(name); dropped++; }
}
if (dropped > 0) log(`pruned ${dropped} stale entries from classifier rate-limit map`);
}
const STATE_FILE = path.join(__dirname, 'state.json');
const ONCE = process.argv.includes('--once');
const PM2_LOG_DIR = path.join(os.homedir(), '.pm2', 'logs');
function log(...a) { console.log(new Date().toISOString(), ...a); }
// Per-process patterns that match expected, non-actionable log noise.
// If EVERY non-empty line in a new log chunk matches the regex, we skip posting.
const BENIGN_NOISE_PATTERNS = {
// cloudflared rotates edge connections every ~16s; the reconnect cycle logs
// ERR but it's normal heartbeat behavior, not a failure.
'big-red-tunnel': /Tunnel connection curve preferences|failed to run the datagram handler.*context canceled|failed to serve tunnel connection.*control stream encountered|Serve tunnel error.*control stream encountered|Retrying connection in up to|Failed to dial a quic connection.*handshake did not complete in time|^\{.*"level":"(info|error)".*\}$/,
// microsite-hawk emits cycle-complete on success — should be stdout, but
// belt-and-suspenders in case the source fix regresses.
'microsite-hawk': /^\[hawk\] Cycle complete .* failures: 0/,
// abrams-report scrapers log vendor HTTP 4xx/timeouts as warnings; these are
// vendor-side issues, not actionable code bugs.
'abrams-report-scrape': /^\[(html|rss)\] [a-z0-9-]+ failed: (HTTP 4\d\d|Request timed out|network timeout)/,
'abrams-report-wallpaper': /^\[(html|rss)\] [a-z0-9-]+ failed: (HTTP 4\d\d|Request timed out|network timeout)/,
// nginx error.log gets constant noise from bots probing TLS — bad key share,
// unsupported protocol, etc. Not actionable on our side.
'error': /SSL_do_handshake\(\) failed.*(bad key share|unsupported protocol|wrong version number|no shared cipher|tlsv1 alert|peer closed connection)/i,
};
function isBenignNoise(processName, chunk) {
const pat = BENIGN_NOISE_PATTERNS[processName];
if (!pat) return false;
const lines = chunk.split('\n').map(l => l.trim()).filter(Boolean);
if (lines.length === 0) return true;
return lines.every(l => pat.test(l));
}
function loadState() {
try {
if (fs.existsSync(STATE_FILE)) return JSON.parse(fs.readFileSync(STATE_FILE, 'utf-8'));
} catch (e) { log('state.json read failed, starting fresh:', e.message); }
return { pm2: {}, logOffsets: {}, startedAt: Date.now() };
}
function saveState(s) {
try { fs.writeFileSync(STATE_FILE, JSON.stringify(s, null, 2), 'utf-8'); }
catch (e) { log('state.json write failed:', e.message); }
}
async function postTask(payload) {
try {
const r = await fetch(`${CNCP_URL}/api/failures`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ ...payload, host: HOST })
});
if (!r.ok) { log('POST /api/failures failed', r.status, await r.text().catch(()=>'')); return null; }
const result = await r.json();
// Fire-and-forget classification on NEW rows only. Per-process 5-min rate limit + server-side single-flight.
if (CLASSIFIER_ENABLED && result.deduped === false && result.task && result.task.id) {
const name = payload.processName;
const last = lastClassify.get(name) || 0;
if (Date.now() - last >= CLASSIFY_RATE_LIMIT_MS) {
lastClassify.set(name, Date.now());
fetch(`${CNCP_URL}/api/failures/${result.task.id}/classify`, { method: 'POST' })
.then(rc => rc.json())
.then(d => log(`classify ${name}: ${d.classified ? d.task.severity + '/' + d.task.category : (d.busy ? 'busy' : 'failed: ' + (d.error || '?'))}`))
.catch(e => log(`classify ${name} threw:`, e.message));
}
}
return result;
} catch (e) {
log('POST /api/failures threw:', e.message);
return null;
}
}
function readPm2() {
try {
// maxBuffer: pm2 jlist embeds full pm2_env per process; on a busy box the
// JSON blows past execSync's 1MB default -> ENOBUFS -> collector dies
// (root cause of the 2026-05-15 mac2 collector death). 64MB = ample headroom.
const out = execSync('pm2 jlist', { encoding: 'utf-8', timeout: 8_000, maxBuffer: 64 * 1024 * 1024 });
return JSON.parse(out);
} catch (e) {
log('pm2 jlist failed:', e.message);
return [];
}
}
function tailFile(p, lines = 20) {
try {
return execSync(`tail -n ${lines} ${JSON.stringify(p)}`, { encoding: 'utf-8', timeout: 3_000, maxBuffer: 16 * 1024 * 1024 });
} catch { return ''; }
}
async function scanPm2(state) {
const procs = readPm2();
const seenNames = new Set();
const isBootstrap = Object.keys(state.pm2).length === 0;
for (const p of procs) {
const name = p.name;
seenNames.add(name);
// Skip ourselves — a restart of the collector (intentional deploy) shouldn't flood the ledger.
if (name === 'cncp-failure-collector') {
state.pm2[name] = { restart_time: p.pm2_env?.restart_time || 0, status: p.pm2_env?.status || 'unknown' };
continue;
}
const prior = state.pm2[name];
const restartTime = p.pm2_env?.restart_time || 0;
const status = p.pm2_env?.status || 'unknown';
const exitedAt = p.pm2_env?.created_at;
// On bootstrap (no prior state), seed baseline silently. Don't post historical restarts.
if (!prior || isBootstrap) {
state.pm2[name] = { restart_time: restartTime, status, exitedAt };
continue;
}
const restartedSinceLast = restartTime > prior.restart_time;
const flippedDown = prior.status === 'online' && status !== 'online';
const persistentlyDown = status !== 'online' && status !== 'launching' && prior.status === status;
// Report on either a fresh crash-and-restart, OR a fresh flip-to-non-online.
// Skip persistently-down (already reported in a prior cycle).
if (restartedSinceLast || flippedDown) {
const errLog = p.pm2_env?.pm_err_log_path || path.join(PM2_LOG_DIR, `${name}-error.log`);
const lastLines = fs.existsSync(errLog) ? tailFile(errLog, 30) : '';
// Skip "phantom" restarts — process bumped restart_time but is online again
// with no error log content. Usually a `pm2 restart` from deploy or a
// one-shot scraper finishing under autorestart=true. flippedDown posts
// always go through even with empty logs (those are real downtime signals).
const restartCountDelta = restartTime - (prior.restart_time || 0);
if (restartedSinceLast && !flippedDown && status === 'online' &&
lastLines.trim().length === 0 && restartCountDelta <= 1) {
state.pm2[name] = { restart_time: restartTime, status, exitedAt };
continue;
}
const posted = await postTask({
source: 'pm2',
processName: name,
exitCode: p.pm2_env?.exit_code ?? null,
lastLines,
restartCount: restartTime
});
if (posted) {
log(`reported pm2 failure: ${name} (status=${status}, restarts=${restartTime})`);
state.pm2[name] = { restart_time: restartTime, status, exitedAt };
} else {
log(`pm2 post failed, will retry next cycle: ${name}`);
// Don't advance state — next cycle will re-attempt.
continue;
}
} else {
state.pm2[name] = { restart_time: restartTime, status, exitedAt };
}
}
// Clean up state for processes that no longer exist (deleted from pm2).
for (const k of Object.keys(state.pm2)) {
if (!seenNames.has(k)) delete state.pm2[k];
}
}
const LAUNCH_AGENTS_DIR = path.join(os.homedir(), 'Library', 'LaunchAgents');
const PLIST_SCAN_INTERVAL_MS = 10 * 60 * 1000;
const MAX_WATCHLIST = 200;
const MAX_LOG_SIZE = 100 * 1024 * 1024;
let cachedWatchlist = null;
function buildLaunchdWatchlist() {
// No-op on Linux — launchd is macOS-only.
if (IS_LINUX) return [];
// Scan ~/Library/LaunchAgents/com.steve.*.plist; extract Label + StandardErrorPath.
// Steve's logs live at per-project paths (e.g. ~/Projects/animals/logs/hawk.stderr.log),
// NOT in ~/Library/Logs/. plist is the only authoritative source.
let plists = [];
try {
plists = fs.readdirSync(LAUNCH_AGENTS_DIR)
.filter(f => f.startsWith('com.steve.') && f.endsWith('.plist'))
.map(f => path.join(LAUNCH_AGENTS_DIR, f));
} catch (e) {
log('buildLaunchdWatchlist: readdir failed', e.message);
return [];
}
const out = [];
for (const p of plists) {
let label = null, errPath = null, outPath = null;
try {
const json = execSync(`plutil -convert json -o - ${JSON.stringify(p)}`, { encoding: 'utf-8', timeout: 2000 });
const parsed = JSON.parse(json);
label = parsed.Label;
errPath = parsed.StandardErrorPath;
outPath = parsed.StandardOutPath;
} catch {
// Regex fallback for malformed plists.
try {
const raw = fs.readFileSync(p, 'utf-8');
const lm = raw.match(/<key>Label<\/key>\s*<string>([^<]+)<\/string>/);
const em = raw.match(/<key>StandardErrorPath<\/key>\s*<string>([^<]+)<\/string>/);
if (lm) label = lm[1];
if (em) errPath = em[1];
} catch {}
}
if (!label) continue;
// Prefer err path; fall back to stdout (some scripts only log to stdout).
const watchPath = errPath || outPath;
if (!watchPath) continue;
if (!fs.existsSync(watchPath)) continue;
out.push({ label, errPath: watchPath });
}
if (out.length > MAX_WATCHLIST) {
log(`launchd watchlist capped: ${out.length} -> ${MAX_WATCHLIST}`);
out.length = MAX_WATCHLIST;
}
return out;
}
function getWatchlist(state) {
const now = Date.now();
if (cachedWatchlist && state.lastPlistScan && (now - state.lastPlistScan) < PLIST_SCAN_INTERVAL_MS) {
return cachedWatchlist;
}
cachedWatchlist = buildLaunchdWatchlist();
state.lastPlistScan = now;
log(`building launchd watchlist: ${cachedWatchlist.length} entries`);
return cachedWatchlist;
}
async function scanLaunchdErrLogs(state) {
const watchlist = getWatchlist(state);
const isBootstrap = Object.keys(state.logOffsets).length === 0;
for (const { label, errPath } of watchlist) {
let stat;
try { stat = fs.statSync(errPath); } catch { continue; }
if (stat.size > MAX_LOG_SIZE) {
// Runaway log — skip read but seed offset at EOF so we resume cleanly if it shrinks.
state.logOffsets[errPath] = stat.size;
continue;
}
const prior = state.logOffsets[errPath];
const cur = stat.size;
if (prior === undefined || isBootstrap) {
state.logOffsets[errPath] = cur;
continue;
}
if (cur <= prior) {
state.logOffsets[errPath] = cur;
continue;
}
let chunk = '';
try {
const fd = fs.openSync(errPath, 'r');
const len = Math.min(cur - prior, 16_000);
const buf = Buffer.alloc(len);
fs.readSync(fd, buf, 0, len, prior);
fs.closeSync(fd);
chunk = buf.toString('utf-8');
} catch (e) { log('read err log failed', errPath, e.message); continue; }
const meaningful = chunk.replace(/\s+/g, '').length > 5;
if (!meaningful) { state.logOffsets[errPath] = cur; continue; }
const processName = label.replace(/^com\.steve\./, '');
// Per-process known-benign-noise filter. Each entry: if EVERY non-empty line
// in the chunk matches the regex, advance the offset without posting.
if (isBenignNoise(processName, chunk)) {
state.logOffsets[errPath] = cur;
continue;
}
const posted = await postTask({
source: 'launchd',
processName,
exitCode: null,
lastLines: chunk.slice(-4000)
});
if (posted) {
log(`reported launchd error chunk: ${processName} (${chunk.length}B new)`);
state.logOffsets[errPath] = cur;
} else {
log(`launchd post failed, will retry: ${processName} (offset stays at ${prior})`);
}
}
}
// ---------------- Linux-only watch paths ----------------
// On Kamatera / any Linux host: tail the configured log files (nginx error log,
// syslog) for new error lines.
// All paths are configurable via env so Steve can swap them out per host without code edits.
const LINUX_LOG_PATHS = (process.env.LINUX_LOG_PATHS || '/var/log/nginx/error.log,/var/log/syslog')
.split(',').map(s => s.trim()).filter(Boolean);
async function scanLinuxLogs(state) {
for (const errPath of LINUX_LOG_PATHS) {
let stat;
try { stat = fs.statSync(errPath); } catch { continue; }
if (stat.size > MAX_LOG_SIZE) { state.logOffsets[errPath] = stat.size; continue; }
const prior = state.logOffsets[errPath];
const cur = stat.size;
const isBootstrap = Object.keys(state.logOffsets).length === 0;
if (prior === undefined || isBootstrap) { state.logOffsets[errPath] = cur; continue; }
if (cur <= prior) { state.logOffsets[errPath] = cur; continue; }
let chunk = '';
try {
const fd = fs.openSync(errPath, 'r');
const len = Math.min(cur - prior, 16_000);
const buf = Buffer.alloc(len);
fs.readSync(fd, buf, 0, len, prior);
fs.closeSync(fd);
chunk = buf.toString('utf-8');
} catch (e) { log('read linux log failed', errPath, e.message); continue; }
// Skip purely benign noise — only post on lines with ERROR / FATAL / crit / [crit] / nginx error markers.
const meaningful = /\b(error|fatal|crit|emerg|alert|panic|segfault|core dumped)\b/i.test(chunk);
if (!meaningful) { state.logOffsets[errPath] = cur; continue; }
const processName = path.basename(errPath).replace(/\.log$/, '');
// Per-process known-benign-noise filter (same as the launchd path). nginx
// error.log (processName === 'error') is mostly TLS-probe noise from bots —
// SSL_do_handshake bad key share / unsupported protocol / etc. If EVERY line
// in the new chunk is that noise, advance the offset without posting. A chunk
// mixing real errors with TLS noise still posts (isBenignNoise requires ALL
// lines match). Without this, a single inbound bad-key-share crit became a
// high-severity ledger row (2026-05-30).
if (isBenignNoise(processName, chunk)) { state.logOffsets[errPath] = cur; continue; }
const posted = await postTask({
source: 'linux',
processName,
exitCode: null,
lastLines: chunk.slice(-4000)
});
if (posted) {
log(`reported linux log chunk: ${processName} (${chunk.length}B new)`);
state.logOffsets[errPath] = cur;
} else {
log(`linux post failed, will retry: ${processName} (offset stays at ${prior})`);
}
}
}
let cyclesSinceLastHeartbeat = 0;
const HEARTBEAT_EVERY = 10; // log once every 10 cycles (~5 min at 30s poll) when quiet
async function cycle() {
// Prune classifier rate-limit Map — drop entries older than 24h so it doesn't grow unbounded over weeks.
pruneClassifyMap();
const state = loadState();
const before = loadCurrentFailuresCount();
try { await scanPm2(state); } catch (e) { log('scanPm2 threw:', e.message); }
try { await scanLaunchdErrLogs(state); } catch (e) { log('scanLaunchdErrLogs threw:', e.message); }
if (IS_LINUX) {
try { await scanLinuxLogs(state); } catch (e) { log('scanLinuxLogs threw:', e.message); }
}
saveState(state);
const after = loadCurrentFailuresCount();
cyclesSinceLastHeartbeat++;
if (after === before && cyclesSinceLastHeartbeat >= HEARTBEAT_EVERY) {
log(`heartbeat — ${cyclesSinceLastHeartbeat} quiet cycles, ${Object.keys(state.pm2).length} pm2 tracked, ${Object.keys(state.logOffsets).length} logs tracked`);
cyclesSinceLastHeartbeat = 0;
} else if (after !== before) {
cyclesSinceLastHeartbeat = 0;
}
}
function loadCurrentFailuresCount() {
// Read directly from CNCP's file (cheaper than HTTP). Returns 0 if unreadable.
try {
const p = path.join(os.homedir(), 'cncp-starter', 'cncp-failures.json');
if (fs.existsSync(p)) return JSON.parse(fs.readFileSync(p, 'utf-8')).length;
} catch {}
return 0;
}
async function main() {
log(`cncp-failure-collector starting. host=${HOST} platform=${PLATFORM} CNCP=${CNCP_URL} POLL=${POLL_INTERVAL_MS}ms ONCE=${ONCE}`);
// Health probe: confirm CNCP is reachable before claiming we're up.
try {
const r = await fetch(`${CNCP_URL}/api/failures?status=triage`);
log('CNCP reachable, current open-tasks count:', (await r.json()).length);
} catch (e) {
log('WARNING: CNCP not reachable at startup — will retry each cycle. err=', e.message);
}
if (ONCE) { await cycle(); return; }
while (true) {
try { await cycle(); } catch (e) { log('cycle threw:', e.message); }
await new Promise(r => setTimeout(r, POLL_INTERVAL_MS));
}
}
main().catch(e => { console.error('fatal:', e); process.exit(1); });