#!/usr/bin/env node /** * systemd 服务异常退出邮件上报器。 * * 两种调用: * service-failure-notify.mjs --report --service pi-mail-bridge.service * service-failure-notify.mjs --flush --service pi-mail-bridge.service * * `--report` 供 ExecStopPost 使用。systemd 会把 SERVICE_RESULT / EXIT_CODE / * EXIT_STATUS / INVOCATION_ID 注入命令环境;正常 stop/restart 不上报,只有 * exit-code、signal、oom-kill、timeout、watchdog 等异常结果才发信。 * * Gateway 暂时不可达时,报告只把**不含密钥**的邮件 payload 落到本地 spool; * 下次服务启动由 `--flush` 补发。relay_key 取 systemd invocation id,立即发送与 * 补发即使竞态也只会入库一封。 */ import { createHash, randomUUID } from 'node:crypto'; import { mkdir, readdir, readFile, rename, rm, stat, writeFile } from 'node:fs/promises'; import { basename, join, resolve } from 'node:path'; import { fileURLToPath } from 'node:url'; const DEFAULT_GATEWAY = 'http://127.0.0.1:8180'; const DEFAULT_RECIPIENT = 'jianf@'; const DEFAULT_SPOOL = '/var/lib/agentmail/failure-spool'; const MAX_SPOOL_FILES = 100; const REQUEST_TIMEOUT_MS = 5_000; function argValue(argv, name) { const i = argv.indexOf(name); return i >= 0 && i + 1 < argv.length ? argv[i + 1] : ''; } export function isAbnormalExit(env = process.env) { const result = String(env.SERVICE_RESULT || '').trim(); if (result) return result !== 'success'; const code = String(env.EXIT_CODE || '').trim(); const status = String(env.EXIT_STATUS || '').trim(); if (!code && !status) return false; return !(code === 'exited' && (status === '' || status === '0')); } function cleanServiceName(value, agentName) { const raw = String(value || '').trim(); if (raw) return basename(raw).replace(/[^A-Za-z0-9_.@-]/g, '-').slice(0, 96); return `${String(agentName || 'agent').replace(/[^A-Za-z0-9_.@-]/g, '-')}-bridge`; } function redact(value, env) { let out = String(value || ''); for (const secret of [env.AGENTMAIL_AGENT_KEY, env.AGENTMAIL_AGENT_SECRET]) { if (secret) out = out.split(String(secret)).join('[REDACTED]'); } return out .replace(/Bearer\s+[A-Za-z0-9._~+\/-]+/gi, 'Bearer [REDACTED]') .replace(/\b(?:ak_[A-Za-z0-9_-]+|sk-[A-Za-z0-9_-]{12,})\b/g, '[REDACTED]') .slice(0, 1_500); } /** * 本次服务启动的短标识,用于让每封故障通知各自成会话(见 subject 处的注释)。 * * 优先用 systemd 的 `INVOCATION_ID`(每次启动唯一)。它缺失时退回随机值 —— * 宁可每次不同,也不要因为拿不到标识而退回"主题相同 → 告警被护栏挡下"。 */ function invocationTag(env) { const invocation = String(env.INVOCATION_ID || '').trim(); if (invocation) return invocation.replace(/[^A-Za-z0-9]/g, '').slice(0, 8); return createHash('sha256') .update(`${Date.now()}:${randomUUID()}`) .digest('hex') .slice(0, 8); } function relayKeyFor(env, serviceName) { const invocation = String(env.INVOCATION_ID || '').trim(); if (invocation) return `service-failure:${invocation}`; const seed = `${serviceName}:${Date.now()}:${randomUUID()}`; return `service-failure:${createHash('sha256').update(seed).digest('hex')}`; } export function buildFailurePayload({ env = process.env, serviceName = '', reason = '', error = null, now = new Date(), } = {}) { const agentName = String(env.AGENTMAIL_AGENT_NAME || 'unknown-agent').trim(); const service = cleanServiceName(serviceName || env.AGENTMAIL_SERVICE_NAME, agentName); const result = String(env.SERVICE_RESULT || reason || 'unexpected-exit'); const exitCode = String(env.EXIT_CODE || 'unknown'); const exitStatus = String(env.EXIT_STATUS || 'unknown'); const detail = redact(error?.stack || error?.message || error || reason, env); const lines = [ `**${agentName}** 的宿主服务 **${service}** 异常终止。`, '', `- 时间:${now.toISOString()}`, `- systemd 结果:${result}`, `- 退出类型:${exitCode}`, `- 退出状态:${exitStatus}`, `- invocation:${String(env.INVOCATION_ID || 'unknown')}`, ]; if (detail) lines.push('', '**异常详情(已脱敏)**:', '```text', detail, '```'); lines.push( '', 'systemd 已按指数退避策略自动重启该服务;如果邮件连续出现,请检查:', `\`journalctl -u ${service} -n 100 --no-pager\``, ); return { to: String(env.AGENTMAIL_FAILURE_RECIPIENT || DEFAULT_RECIPIENT), /* * 主题带上本次启动的 invocation 短号。 * * 服务端按 `AutoAliasFor(收件人, 主题)` 派生会话别名,所以**主题相同 = 同一条会话**。 * 崩溃循环下这一点会咬人:多封告警全落进同一条会话,而网关对会话内**连续**的中继 * 邮件有 `maxRelayHops=5` 的硬上限(防两个 Agent 互相唤醒的正当护栏)→ 第 6 封起 * 返回 403,报告只能落 spool,等下次 ExecStartPost `--flush` 才补投。 * * 实测就是这样:dsh 崩溃循环时的六条告警全部 spooled,而**服务反复崩溃正是最需要 * 告警送达的时刻**。 * * 带 invocation 让每次崩溃各自成会话,因而每次都能投出去 —— 护栏不受影响 * (它针对的是 agent↔agent 的互相唤醒,不是同一个人收多条故障通知)。 * 副作用是崩溃循环会产生多条会话而不是一条线程;对"服务在崩"这件事, * 分开计数比合并成一条更容易发现问题。 */ subject: `[${agentName}] 桥服务异常终止 #${invocationTag(env)}`, body: lines.join('\n'), relay: 'summary', relay_key: relayKeyFor(env, service), }; } async function postPayload(payload, env = process.env) { const key = String(env.AGENTMAIL_AGENT_KEY || '').trim(); if (!key) throw new Error('AGENTMAIL_AGENT_KEY 未配置'); const base = String(env.AGENTMAIL_GATEWAY_URL || DEFAULT_GATEWAY).replace(/\/+$/, ''); const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS); try { const res = await fetch(`${base}/api/v1/mail/send`, { method: 'POST', headers: { Authorization: `Bearer ${key}`, 'X-Agent-Name': String(env.AGENTMAIL_AGENT_NAME || ''), 'Content-Type': 'application/json', }, body: JSON.stringify(payload), signal: controller.signal, }); const responseBody = await res.text(); if (!res.ok) throw new Error(`Gateway HTTP ${res.status}: ${responseBody.slice(0, 300)}`); return responseBody; } finally { clearTimeout(timer); } } function safeAgentName(env) { return String(env.AGENTMAIL_AGENT_NAME || 'unknown-agent') .replace(/[^A-Za-z0-9_.@-]/g, '-') .slice(0, 96); } function spoolDirFor(env) { return join(String(env.AGENTMAIL_FAILURE_SPOOL || DEFAULT_SPOOL), safeAgentName(env)); } function safeSpoolName(payload) { return `${createHash('sha256').update(String(payload.relay_key)).digest('hex')}.json`; } async function pruneSpool(spoolDir) { const names = (await readdir(spoolDir).catch(() => [])) .filter((name) => name.endsWith('.json')); if (names.length <= MAX_SPOOL_FILES) return; const entries = await Promise.all(names.map(async (name) => ({ name, mtime: (await stat(join(spoolDir, name))).mtimeMs, }))); entries.sort((a, b) => a.mtime - b.mtime); for (const entry of entries.slice(0, entries.length - MAX_SPOOL_FILES)) { await rm(join(spoolDir, entry.name), { force: true }); } } export async function spoolPayload(payload, env = process.env) { // 每个 Agent 独立目录:否则 pi 启动时可能拿自己的密钥去补发 dsh 的报告, // 发件身份与正文主体会错位,relay_key 的幂等范围也变了。 const spoolDir = spoolDirFor(env); await mkdir(spoolDir, { recursive: true, mode: 0o700 }); const target = join(spoolDir, safeSpoolName(payload)); const temp = `${target}.${process.pid}.${randomUUID()}.tmp`; await writeFile(temp, `${JSON.stringify({ payload }, null, 2)}\n`, { mode: 0o600 }); await rename(temp, target); await pruneSpool(spoolDir); return target; } export async function notifyCrash(reason, error, options = {}) { const env = options.env || process.env; const payload = buildFailurePayload({ env, serviceName: options.serviceName, reason, error, now: options.now, }); try { await postPayload(payload, env); return { sent: true, spooled: false, payload }; } catch (sendError) { if (options.spool === false) throw sendError; const path = await spoolPayload(payload, env); return { sent: false, spooled: true, path, payload, error: sendError }; } } export async function flushSpool(env = process.env) { const spoolDir = spoolDirFor(env); const names = (await readdir(spoolDir).catch(() => [])) .filter((name) => name.endsWith('.json')) .sort(); let sent = 0; let failed = 0; for (const name of names) { const path = join(spoolDir, name); try { const record = JSON.parse(await readFile(path, 'utf8')); await postPayload(record.payload, env); await rm(path, { force: true }); sent++; } catch { failed++; } } return { total: names.length, sent, failed }; } /** * 把发送失败翻译成一句能定位原因的话。 * * `postPayload` 对非 2xx 抛的是 `Gateway HTTP : ` —— 里面已经带着 * 状态码与响应体,直接打出来就够定位了。只有 fetch 本身抛(DNS/连接/超时)才真的是 * 「不可达」,此时按名字区分开。 */ function describeSendError(error) { const message = String(error?.message || error || ''); const httpMatch = message.match(/^Gateway HTTP (\d{3}): ([\s\S]*)$/); if (httpMatch) { return `网关可达,但返回 HTTP ${httpMatch[1]}:${httpMatch[2].slice(0, 300)}`; } if (error?.name === 'AbortError') return '请求超时(网关未在限定时间内响应)'; if (/fetch failed|ECONNREFUSED|ENOTFOUND|EHOSTUNREACH/i.test(message)) { return `网关不可达:${message}`; } return message || '未知错误'; } async function main(argv = process.argv.slice(2), env = process.env) { const serviceName = argValue(argv, '--service'); if (argv.includes('--flush')) { const result = await flushSpool(env); if (result.sent || result.failed) { console.error(`[agentmail-failure-notify] spool: sent=${result.sent} failed=${result.failed}`); } return; } if (!argv.includes('--report')) throw new Error('需要 --report 或 --flush'); if (!isAbnormalExit(env)) return; if (argv.includes('--dry-run')) { console.log(JSON.stringify(buildFailurePayload({ env, serviceName }), null, 2)); return; } const result = await notifyCrash(env.SERVICE_RESULT || 'unexpected-exit', null, { env, serviceName, }); if (result.sent) { console.error(`[agentmail-failure-notify] ${serviceName} 异常已上报`); return; } /* * 区分「连不上」与「连上了但被拒」。 * * 原先两种情况都打印「Gateway 不可达」。实测代价:切换插件时 dsh 进了崩溃循环, * 通知脚本连打六条「Gateway 不可达」并写进 spool —— 而网关**一直在正常服务** * (NRestarts=0),真实响应是 **HTTP 403**(同一会话连续中继邮件撞上防互相唤醒 * 的跳数上限,`maxRelayHops=5`)。 * * 那句话会把人送去查网络,而问题在策略层。故障通知本身给出误导性诊断, * 是「静默失败」的另一种形态。 */ console.error( `[agentmail-failure-notify] ${serviceName} 异常上报失败,报告已暂存 ${result.path}` ); console.error(`[agentmail-failure-notify] 原因:${describeSendError(result.error)}`); } const isCLI = process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url); if (isCLI) { main().catch((error) => { console.error(`[agentmail-failure-notify] ${error?.message || error}`); process.exitCode = 1; }); }