diff --git a/docs/runbook-scheduler.md b/docs/runbook-scheduler.md new file mode 100644 index 0000000..c900347 --- /dev/null +++ b/docs/runbook-scheduler.md @@ -0,0 +1,402 @@ +# Pangolin Scheduler Runbook + +> **适用范围**：本文档面向 Pangolin 内部运维人员，描述 scheduler 七种告警事件的排查与处置流程。 +> **保密提示**：告警消息含内部节点 ID，不得转发至外部渠道。节点 ID 是不透明内部标识符，不含域名。 + +--- + +## 目录 + +1. [判封确认 (block_confirmed)](#block-confirmed) +2. [补新连续失败≥3 (replenish_failed)](#replenish-failed) +3. [水位<70% (watermark_low)](#watermark-low) +4. [熔断触发 (breaker_tripped)](#breaker-tripped) +5. [探针失联 (probe_agent_lost)](#probe-agent-lost) +6. [心跳缺失>90s (heartbeat_missing)](#heartbeat-missing) +7. [故障态 (fault)](#node-fault) + +--- + + +## 1. 判封确认 (block_confirmed) + +### 含义 + +15D（检测引擎）确认某节点被 GFW 封锁：该节点在 `blocked_suspect` 状态下连续经历 ≥6 个探测周期（约 30 分钟），国内三大运营商中 ≥2/3 探测失败而境外探测正常，判定为确认封锁（`blocked_confirmed`）。封锁确认后节点立即转入 `down` 状态，并进入 15E 补充队列。 + +**正常处置**：15E 自动补充新节点，通常无需人工干预。若随后出现 [`replenish_failed`](#replenish-failed) 告警则升级处理。 + +### 先查什么 + +```sql +-- 查看 node_events 最近的封锁确认事件（需替换 ） +SELECT created_at, actor, action, detail +FROM node_events +WHERE target = 'node:' + AND action IN ('transition', 'block_confirmed') +ORDER BY created_at DESC +LIMIT 10; +``` + +```bash +# 查看 15E 补充记录（在 ec2 上） +redis-cli GET sched:replace: +redis-cli SMEMBERS sched:replace:index +``` + +```bash +# 查看该节点最新探针快照 +redis-cli KEYS "probe::*" +redis-cli GET "probe::CN::3rd-ChinaTelecom" +``` + +### 处置步骤 + +1. **确认补充进行中**：检查 `sched:replace:index` 是否有该节点的记录，以及记录的 `phase` 字段（应为 `pending` / `creating` / `probing` 之一）。若存在则等待 15E 自动完成。 +2. **若补充卡住**：参见 [`replenish_failed`](#replenish-failed)。 +3. **记录 IP**：从节点数据库查取该节点旧 IP，记录到封锁 IP 备案。 +4. **可选 SNI 轮换**：如池内同 SNI 多节点均被封，通知基础设施团队轮换 REALITY SNI。 + +### 升级条件 + +- 同一池（tier/region）内 30 分钟内出现 ≥3 次 `block_confirmed` → 触发 [`breaker_tripped`](#breaker-tripped)。 +- 补充失败 → 升级至 [`replenish_failed`](#replenish-failed)。 + +--- + + +## 2. 补新连续失败≥3 (replenish_failed) + +### 含义 + +15E（编排引擎）在为某个已封锁节点补充新节点时，连续尝试 3 次（默认 `MaxAttempts = 3`）均失败（探针验证超时 15 min 或 Provider API 报错），替换记录进入 `failed` 状态。旧节点仍处于 `down` 状态，**池容量已减少**。此为 Critical 级别告警，需立即人工介入。 + +### 先查什么 + +```bash +# 读取失败的替换记录 +redis-cli GET sched:replace: +# 字段说明：attempts（尝试次数）、providerTried（已试过的 Provider） +``` + +```bash +# 查看目标 Pool 当前容量 +redis-cli SMEMBERS sched:replace:index # 进行中的替换数量 +``` + +```bash +# 查看新节点的探针快照（每次 attempt 生成的 newNode） +redis-cli KEYS "probe::*" +``` + +```sql +-- 查看对应 node_events 日志 +SELECT created_at, actor, action, detail +FROM node_events +WHERE target = 'node:' +ORDER BY created_at DESC +LIMIT 20; +``` + +### 处置步骤 + +1. **判断失败原因**： + - 若 `providerTried` 列出了所有可用 Provider → Provider 全面故障，联系 IaaS 供应商。 + - 若探针超时 → 检查新节点 IP 是否立即被封（该 IP 段被 GFW 封锁），换 Provider 或换 IP 段。 + - 若 Provider API 报错 → 检查 API 凭证（`ALIYUN_PROBE_ACCESS_KEY_*` 等）是否有效。 + +2. **手动重入队列**（谨慎操作）： + ```bash + # 删除失败记录并重新推入队列，让 15E 重试 + redis-cli DEL sched:replace: + redis-cli SREM sched:replace:index + # 生成新 UUID，推送新条目 + redis-cli LPUSH detect:replace:queue '{"nodeId":"","replacementUuid":""}' + ``` + +3. **临时扩容**：若容量持续不足，通知用户停止新订阅，并从健康 Pool 临时调配权重。 + +4. **根因修复**：修复 Provider 问题或更换 IP 段后，15E 下次 Tick（30s）自动恢复。 + +### 升级条件 + +- 同一 Pool 内 ≥2 个节点同时 `replenish_failed` → 极端容量危机，启动灾备预案。 +- Provider 全部不可用超过 1 小时 → 升级到架构层面（新增 Provider）。 + +--- + + +## 3. 水位<70% (watermark_low) + +### 含义 + +某节点池的有效路由权重之和低于满容量的 70%。通常由节点封锁（判封确认后进入 `down`）积累引起，可能影响用户体验（延迟上升、连接失败率增加）。 + +**触发条件**：池内活跃节点（`up` 状态）权重总和 < 池满容量 × 70%。 + +### 先查什么 + +```sql +-- 查看池内各节点状态和权重 +SELECT id, status, weight, tier, region, updated_at +FROM nodes +WHERE tier = '' AND region = '' +ORDER BY status, weight DESC; +``` + +```bash +# 查看进行中的补充任务 +redis-cli SMEMBERS sched:replace:index +for uuid in $(redis-cli SMEMBERS sched:replace:index); do + redis-cli GET sched:replace:$uuid | python3 -m json.tool +done +``` + +```bash +# 查看灰度坡道（新节点权重爬坡） +redis-cli KEYS "sched:gray:*" +``` + +### 处置步骤 + +1. **确认补充进行中**：若有多个 `replenish` 记录在 `probing` / `creating` 阶段，15E 正在恢复，通常等待即可。 +2. **若补充全部卡住**：参见 [`replenish_failed`](#replenish-failed)。 +3. **加速灰度爬坡**（临时措施）：若新节点已在灰度但权重还低，可手动将其权重推高： + ```bash + # 通过管理 API 设置节点权重（参见 #8 管理操作） + curl -X POST https://api.internal/admin/nodes//weight -d '{"weight":100}' + ``` +4. **容量预警通知**：若水位持续 <50% 超过 30 分钟，通知运营团队评估影响面。 + +### 升级条件 + +- 水位降至 <50% → 紧急：启动备用容量或降级流控。 +- 水位持续 <70% 超过 2 小时且无自动恢复 → 升级为容量规划问题。 + +--- + + +## 4. 熔断触发 (breaker_tripped) + +### 含义 + +15E 的熔断器（circuit breaker）阻止了新的替换操作。熔断触发表明同一 tier/region 池在短时间内有过多确认封锁，系统认为继续补充可能造成新节点也立即被封（IP 段整体被墙），因此暂停补充以避免浪费资源。此为 Critical 级别告警，需立即人工研判。 + +熔断后受影响节点的替换记录停留在 `pending` 阶段直到熔断解除。 + +### 先查什么 + +```bash +# 查看熔断器计数器（按 tier:region 键） +redis-cli KEYS "breaker:*" +redis-cli GET "breaker:::count" +redis-cli TTL "breaker:::count" +``` + +```bash +# 查看待处理的替换任务 +redis-cli SMEMBERS sched:replace:index +``` + +```sql +-- 查看近期封锁事件数量 +SELECT DATE_TRUNC('hour', created_at) as hour, COUNT(*) as cnt +FROM node_events +WHERE action = 'transition' + AND detail->>'to' = 'blocked_confirmed' + AND detail->>'tier' = '' + AND created_at > NOW() - INTERVAL '2 hours' +GROUP BY 1 +ORDER BY 1 DESC; +``` + +### 处置步骤 + +1. **评估封锁模式**： + - 若仅个别节点封锁 → 正常 GFW 例行扫描，等待熔断自动超时（通常 30 min）恢复。 + - 若批量封锁（>5 节点/小时）→ IP 段整体被封，需更换 IP 段或切换 Provider。 + +2. **手动解除熔断**（#8 管理操作）： + ```bash + # 通过管理 API 清除熔断计数器 + curl -X DELETE https://api.internal/admin/breaker// + # 或直接在 Redis 删除计数键 + redis-cli DEL "breaker:::count" + ``` + +3. **IP 段评估**：联系 IaaS 供应商，确认当前 IP 范围是否已进入 GFW 黑名单，必要时申请新 IP 段。 + +4. **降级保障**：若熔断超过 4 小时，通知运营团队考虑临时迁移到其他 Provider。 + +### 升级条件 + +- 多个 region 同时熔断 → 全球性 GFW 扫描事件，启动应急响应。 +- 手动解除熔断后立即再次触发 → IP 段问题未解决，升级到基础设施团队。 + +--- + + +## 5. 探针失联 (probe_agent_lost) + +### 含义 + +15F（探针子系统）的第三方拨测 Agent（阿里云云监控）连续 ≥3 次 API 调用失败。这不代表被测节点本身有问题，而是**探测能力本身丧失**：15D 将无法获取国内运营商探测数据，可能导致判封灵敏度下降（漏判）。 + +**注意**：探针失联不触发节点状态变更，只是减少探测数据的覆盖范围。 + +### 先查什么 + +```bash +# 检查阿里云 API 凭证是否有效（在 ec2 上） +curl -s "https://cloudmonitor.cn-hangzhou.aliyuncs.com/" | head -20 +# 预期返回 403/401（证明网络可达），而非 connection refused + +# 检查 scheduler 进程日志 +journalctl -u pangolin-scheduler -n 100 --no-pager | grep "prober_agent" +``` + +```bash +# 检查阿里云 RAM 子账号配额 +# （需在阿里云控制台或通过 aliyun CLI 查询） +``` + +### 处置步骤 + +1. **网络连通性**：确认 ec2 可访问 `cloudmonitor.cn-hangzhou.aliyuncs.com`（国内端点需确保没有出口限制）。 +2. **API 凭证**：检查 `ALIYUN_PROBE_ACCESS_KEY_ID` / `ALIYUN_PROBE_ACCESS_KEY_SECRET` 环境变量是否正确且未过期。 +3. **配额耗尽**：阿里云云拨测按次计费，检查当月用量是否超限。若超限，临时降低探测频率或充值。 +4. **服务故障**：访问阿里云状态页确认云监控服务是否有故障。 +5. **降级运行**：探针失联期间 15D 仅依赖已有的历史快照（TTL 30 min），封锁判断会有所延迟，可接受短期（<30 min）降级。 + +### 升级条件 + +- 探针失联超过 30 分钟 → 15D 的历史快照开始过期，判封能力严重受损，需立即恢复。 +- 凭证问题无法快速解决 → 临时切换到自建探针 Agent（参见 probe 包文档）。 + +--- + + +## 6. 心跳缺失>90s (heartbeat_missing) + +### 含义 + +某个**自建（first-party）探针 Agent** 超过 90 秒未向 `/probe/report` 发送任何心跳数据。与探针失联（[`probe_agent_lost`](#probe-agent-lost)）不同，此告警针对自建 Agent，不是第三方拨测服务。 + +自建 Agent 心跳缺失意味着来自该 Agent 所在网络位置（特定 ISP/省份）的 L3 数据将中断，影响判封精确度。 + +### 先查什么 + +```bash +# 查看对应探针最后一次心跳时间 +redis-cli GET "probe:hb:" +# 值为 Unix 时间戳，与当前时间差即为失联时长 + +redis-cli TTL "probe:hb:" +# 剩余 TTL（15min = 900s），若已到期则 key 不存在 +``` + +```bash +# 在对应探针机器上检查 probe agent 进程状态 +ssh "systemctl status pangolin-probe-agent" +ssh "journalctl -u pangolin-probe-agent -n 50 --no-pager" +``` + +```bash +# 检查探针机器与 scheduler 的网络连通性 +ssh "curl -v https:///probe/report" +``` + +### 处置步骤 + +1. **检查 Agent 进程**： + - 若进程未运行 → `systemctl restart pangolin-probe-agent`。 + - 若进程运行但报错 → 查看日志，常见原因：HMAC 密钥错误、Scheduler 地址配置错误、TLS 证书问题。 +2. **网络连通性**：确认探针机器出网正常，且 Scheduler 的 `/probe/report` 端口可达。 +3. **HMAC 密钥轮换**：若密钥过期或被更新，更新 Agent 配置文件后重启。 +4. **探针机器故障**：若机器故障，从备用位置部署新探针 Agent。 + +### 升级条件 + +- 某 ISP / 省份所有探针均失联 → 该区域探测盲区，节点封锁可能被漏判，升级处理。 +- 失联超过 2 小时且无法恢复 → 考虑临时增加第三方拨测覆盖（阿里云）弥补缺口。 + +--- + + +## 7. 故障态 (fault) + +### 含义 + +15D 检测到某节点**国内与境外探测同时失败**，判定为节点级别的网络故障（不是 GFW 封锁）。故障态节点**不进行状态转换、不触发补充流程**，也不消耗补充配额，仅通知人工研判。 + +典型场景：节点主机宕机、网卡故障、IDC 网络中断等基础设施问题。 + +### 先查什么 + +```bash +# 查看该节点当前探针快照 +redis-cli KEYS "probe::*" +# 检查各运营商和境外的探测结果 +redis-cli GET "probe::CN::3rd-ChinaTelecom" +redis-cli GET "probe::SG::" +``` + +```bash +# SSH 登录节点进行基础诊断（若可达） +ssh "systemctl status singbox xray" +ssh "ss -tlnp | grep -E '443|8080'" +``` + +```bash +# 从控制平面 ping / traceroute（境外节点） +ping -c 5 +traceroute +``` + +```sql +-- 查看节点历史状态变化 +SELECT created_at, action, detail +FROM node_events +WHERE target = 'node:' +ORDER BY created_at DESC +LIMIT 20; +``` + +### 处置步骤 + +1. **区分 GFW 封锁与真实故障**： + - 仅国内失败 + 境外正常 → GFW 封锁（此路径不应触发 fault，由 15D Rule 1-3 处理）。 + - 国内 + 境外均失败 → 节点级故障（此告警场景）。 +2. **IaaS 控制台确认**：登录 Provider 控制台，检查节点（EC2/VPS）运行状态。 +3. **若节点可 SSH**：检查服务进程是否崩溃，查看系统日志（`dmesg`, `journalctl`）。 +4. **若节点不可 SSH**：通过 Provider 控制台进行 VNC/串口连接或强制重启。 +5. **若需下线**： + - 手动将节点状态改为 `down`（通过管理 API #8）。 + - 手动推入补充队列： + ```bash + redis-cli LPUSH detect:replace:queue '{"nodeId":"","replacementUuid":""}' + ``` +6. **根因记录**：在节点事件表记录故障原因（通过 WriteAuditLog API）。 + +### 升级条件 + +- 同一 IDC / 可用区多节点同时故障 → IDC 事件，联系 IaaS 供应商。 +- 故障节点无法通过控制台恢复 → 放弃该节点，补充新节点（手动推入队列）。 +- 故障持续 > 1 小时且涉及 >10% 池容量 → 进入水位<70% 处置流程。 + +--- + +## 附录：关键 Redis Key 速查 + +| Key 前缀 | 含义 | +|---|---| +| `probe::::` | 节点探针快照（30min TTL） | +| `probe:hb:` | 自建探针心跳时间戳（15min TTL） | +| `probe:freq:` | 节点提频标记（进入 suspect 时写入，45min TTL） | +| `detect:replace:queue` | 15D → 15E 补充任务队列（LPUSH/RPOP） | +| `sched:replace:` | 单次替换编排记录（7天审计保留） | +| `sched:replace:index` | 进行中替换任务 UUID 集合 | +| `sched:gray:` | 新节点灰度爬坡记录 | +| `alert:dedup::` | 告警去重令牌（10min TTL，仅非 Critical 事件） | +| `breaker:::count` | 熔断器计数器 | +| `streak:` | 节点连续失败/恢复计数（Redis JSON） | diff --git a/server/internal/alert/notifier.go b/server/internal/alert/notifier.go new file mode 100644 index 0000000..cfdcf53 --- /dev/null +++ b/server/internal/alert/notifier.go @@ -0,0 +1,430 @@ +// Package alert implements the unified alert exit channel for Pangolin's +// scheduler subsystem (task 15G). +// +// # Event types +// +// Seven fixed event types cover all failure modes detected by 15D (detect), +// 15E (orchestrate), and 15F (probe): +// +// EventTypeBlockConfirmed — node confirmed GFW-censored after ≥6 cycles +// EventTypeReplenishFailed — replacement provisioning exhausted all retries +// EventTypeWatermarkLow — pool node-weight drops below 70 % capacity +// EventTypeBreakerTripped — circuit breaker blocked a replacement attempt +// EventTypeProbeAgentLost — third-party synthetic probe failing ≥3 times +// EventTypeHeartbeatMissing — first-party probe silent for >90 s +// EventTypeFault — node-level outage (domestic + overseas both fail) +// +// # Deduplication +// +// Non-critical events are throttled: the same (Type, NodeID) pair fires at +// most once per 10-minute window (Redis SETNX + TTL). Critical events always +// fire — they are never suppressed. +// +// # Alert channel safety +// +// Notify must never return a blocking error. TG API failures are retried up +// to maxRetries times, then fall back to the slog-based LogNotifier. The +// calling goroutine (scheduler tick) is never stalled by alert-channel faults. +// +// # Identity isolation +// +// The TG bot token and chat ID are injected via environment variables only. +// The bot belongs to a dedicated anonymous operator account; see +// infra/identity-isolation.md and red line 06 §2. +// +// # Privacy +// +// Event messages use internal node IDs (opaque), never public domain names. +// This prevents domain leakage if the alert message is forwarded. +package alert + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "strings" + "time" + + "github.com/redis/go-redis/v9" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Event types +// ───────────────────────────────────────────────────────────────────────────── + +// EventType identifies the class of alert event. +type EventType string + +const ( + // EventTypeBlockConfirmed fires when detect (15D) transitions a node from + // blocked_suspect to blocked_confirmed, confirming GFW censorship after + // ≥6 consecutive failing probe cycles. + // Runbook anchor: #block-confirmed + EventTypeBlockConfirmed EventType = "block_confirmed" + + // EventTypeReplenishFailed fires when orchestrate (15E) exhausts all + // MaxAttempts replacement attempts for a blocked node. + // Runbook anchor: #replenish-failed + EventTypeReplenishFailed EventType = "replenish_failed" + + // EventTypeWatermarkLow fires when the effective capacity of a node pool + // (combined routing weight of live nodes) drops below 70 % of full capacity. + // Runbook anchor: #watermark-low + EventTypeWatermarkLow EventType = "watermark_low" + + // EventTypeBreakerTripped fires when the circuit breaker (15F) blocks a + // replacement attempt due to a burst of confirmed blocks in the same pool. + // Runbook anchor: #breaker-tripped + EventTypeBreakerTripped EventType = "breaker_tripped" + + // EventTypeProbeAgentLost fires when the third-party (Aliyun) synthetic + // probe agent encounters ≥3 consecutive API failures. + // Runbook anchor: #probe-agent-lost + EventTypeProbeAgentLost EventType = "probe_agent_lost" + + // EventTypeHeartbeatMissing fires when a first-party probe agent has not + // delivered any heartbeat for more than 90 seconds. + // Runbook anchor: #heartbeat-missing + EventTypeHeartbeatMissing EventType = "heartbeat_missing" + + // EventTypeFault fires when both domestic AND overseas probes are failing + // for a node, indicating a node-level outage rather than GFW censorship. + // Runbook anchor: #node-fault + EventTypeFault EventType = "fault" +) + +// Severity indicates the urgency of an event. +type Severity string + +const ( + SeverityCritical Severity = "critical" // requires immediate human action + SeverityWarning Severity = "warning" // review within the hour + SeverityInfo Severity = "info" // informational only +) + +// typeMeta bundles display properties for one event type. +type typeMeta struct { + Severity Severity + TitleZH string // human-readable Chinese label + Emoji string // prefixed to the TG message + RunbookAnchor string // HTML id anchor in docs/runbook-scheduler.md +} + +// typeMetaMap is indexed by EventType. +var typeMetaMap = map[EventType]typeMeta{ + EventTypeBlockConfirmed: {SeverityWarning, "判封确认", "⚠️", "#block-confirmed"}, + EventTypeReplenishFailed: {SeverityCritical, "补新连续失败≥3", "🔴", "#replenish-failed"}, + EventTypeWatermarkLow: {SeverityWarning, "水位<70%", "⚠️", "#watermark-low"}, + EventTypeBreakerTripped: {SeverityCritical, "熔断触发", "🔴", "#breaker-tripped"}, + EventTypeProbeAgentLost: {SeverityWarning, "探针失联", "⚠️", "#probe-agent-lost"}, + EventTypeHeartbeatMissing: {SeverityWarning, "心跳缺失>90s", "⚠️", "#heartbeat-missing"}, + EventTypeFault: {SeverityCritical, "故障态", "🔴", "#node-fault"}, +} + +// ───────────────────────────────────────────────────────────────────────────── +// Event +// ───────────────────────────────────────────────────────────────────────────── + +// Event carries all data required to render one alert notification. +type Event struct { + // Type is one of the seven EventType constants. + Type EventType + + // Severity controls dedup and display urgency. + // Derived from Type via NewEvent; may be overridden by the caller. + Severity Severity + + // NodeID is the internal opaque node or probe agent identifier. + // MUST be an internal ID — never a public domain name (privacy). + NodeID string + + // Pool optionally identifies the operational pool (e.g. "free/hkg"). + Pool string + + // Detail holds supplementary key-value data for the notification. + // Values must not contain personal user data. + Detail map[string]string + + // RunbookAnchor is the #fragment anchor in docs/runbook-scheduler.md. + // Overrides the type default when non-empty. + RunbookAnchor string +} + +// NewEvent constructs an Event with type-derived Severity and RunbookAnchor. +// Callers may override any field after construction. +func NewEvent(t EventType, nodeID string, detail map[string]string) Event { + m := typeMetaMap[t] + return Event{ + Type: t, + Severity: m.Severity, + NodeID: nodeID, + Detail: detail, + RunbookAnchor: m.RunbookAnchor, + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Notifier interface +// ───────────────────────────────────────────────────────────────────────────── + +// Notifier is the unified alert outlet for scheduler events. +// Implementations must be safe for concurrent use and must never return a +// blocking error — alert-channel failures must not halt the scheduler. +type Notifier interface { + Notify(ctx context.Context, event Event) error +} + +// ───────────────────────────────────────────────────────────────────────────── +// LogNotifier +// ───────────────────────────────────────────────────────────────────────────── + +// LogNotifier writes every event to the structured logger. +// It is used when TG_BOT_TOKEN is not configured, or as the fallback when the +// TGNotifier exhausts its retries. +type LogNotifier struct{} + +// Notify implements Notifier. +func (LogNotifier) Notify(_ context.Context, e Event) error { + m := typeMetaMap[e.Type] + slog.Warn("alert: scheduler event", + "type", string(e.Type), + "title", m.TitleZH, + "severity", string(e.Severity), + "node_id", e.NodeID, + "pool", e.Pool, + "detail", e.Detail, + "runbook", e.RunbookAnchor, + ) + return nil +} + +// ───────────────────────────────────────────────────────────────────────────── +// TGNotifier +// ───────────────────────────────────────────────────────────────────────────── + +const ( + // dedupTTL is the dedup suppression window for non-critical events. + dedupTTL = 10 * time.Minute + + // dedupKeyPrefix is the Redis key namespace for dedup tokens. + // Full key: alert:dedup:{type}:{nodeID} + dedupKeyPrefix = "alert:dedup:" + + // maxRetries is the number of additional TG send attempts after the first + // failure. Exhausting retries falls back to LogNotifier silently. + maxRetries = 2 + + // tgSendMessageURL is the Telegram Bot API pattern for sendMessage. + tgSendMessageURL = "https://api.telegram.org/bot%s/sendMessage" +) + +// TGConfig holds the runtime configuration for TGNotifier. +// All sensitive values come from environment variables; see cmd/ bootstrap. +type TGConfig struct { + // BotToken is the Telegram bot token (env: TG_BOT_TOKEN). + // The bot must belong to a dedicated anonymous operator account (red line §2). + BotToken string + + // ChatID is the target Telegram group or channel ID (env: TG_ALERT_CHAT_ID). + ChatID string + + // RunbookBaseURL is prepended to the anchor to form the full runbook URL. + // Leave empty to embed only the #anchor fragment. + RunbookBaseURL string + + // HTTPTimeout overrides the per-call HTTP timeout (default 10 s). + HTTPTimeout time.Duration + + // BaseURL overrides the TG API base URL for testing. + // Leave empty in production. + BaseURL string +} + +// TGNotifier sends events to a Telegram group via the Bot API. +// +// Deduplication: +// - Non-critical events: one message per (Type, NodeID) per 10-minute window +// (Redis SETNX + TTL). The second identical message within the window is +// silently dropped. +// - Critical events: always delivered, no dedup. +// +// Retry + fallback: +// - On TG API failure, retries up to maxRetries (2) times before delegating +// to the fallback Notifier. Notify always returns nil. +type TGNotifier struct { + cfg TGConfig + rdb *redis.Client + httpClient *http.Client + fallback Notifier + apiURL string // base send URL, pre-formatted with token +} + +// NewTGNotifier constructs a TGNotifier. +// rdb is used for the dedup SETNX gate. +// fallback is used when TG API calls are exhausted; nil defaults to LogNotifier. +func NewTGNotifier(cfg TGConfig, rdb *redis.Client, fallback Notifier) *TGNotifier { + if fallback == nil { + fallback = LogNotifier{} + } + httpTimeout := 10 * time.Second + if cfg.HTTPTimeout > 0 { + httpTimeout = cfg.HTTPTimeout + } + baseURL := fmt.Sprintf(tgSendMessageURL, cfg.BotToken) + if cfg.BaseURL != "" { + // Test hook: override the entire API base URL (token appended separately). + baseURL = cfg.BaseURL + } + return &TGNotifier{ + cfg: cfg, + rdb: rdb, + httpClient: &http.Client{Timeout: httpTimeout}, + fallback: fallback, + apiURL: baseURL, + } +} + +// Notify implements Notifier. +func (n *TGNotifier) Notify(ctx context.Context, e Event) error { + // Critical events bypass the dedup gate entirely. + if e.Severity != SeverityCritical { + allowed, err := n.dedupGate(ctx, e) + if err != nil { + // Redis error → allow through (fail-open) to avoid losing alerts. + slog.Warn("alert: dedup Redis error; allowing through", "error", err) + } + if !allowed { + return nil // duplicate within the 10-min window; silently dropped + } + } + + text := n.renderMessage(e) + + var lastErr error + for attempt := 0; attempt <= maxRetries; attempt++ { + if err := n.callTG(ctx, text); err != nil { + lastErr = err + slog.Warn("alert: TG send failed", + "attempt", attempt+1, "of", maxRetries+1, + "type", string(e.Type), "node_id", e.NodeID, + "error", err, + ) + continue + } + return nil // delivered + } + + // Retries exhausted: degrade to log. Never return an error. + slog.Error("alert: TG delivery exhausted retries; falling back to log", + "type", string(e.Type), "node_id", e.NodeID, "last_error", lastErr) + _ = n.fallback.Notify(ctx, e) + return nil +} + +// dedupGate returns (true, nil) if the event should be sent (key newly set), +// or (false, nil) if it is a duplicate within the 10-min window. +func (n *TGNotifier) dedupGate(ctx context.Context, e Event) (bool, error) { + key := dedupKeyPrefix + string(e.Type) + ":" + e.NodeID + set, err := n.rdb.SetNX(ctx, key, "1", dedupTTL).Result() + if err != nil { + return true, fmt.Errorf("alert: dedup SETNX: %w", err) + } + return set, nil +} + +// renderMessage builds the Telegram HTML message body. +// Node IDs are used as-is (internal opaque IDs, not domain names). +func (n *TGNotifier) renderMessage(e Event) string { + m := typeMetaMap[e.Type] + + var sb strings.Builder + + // Line 1: emoji + bold event title + node id + sb.WriteString(m.Emoji) + sb.WriteString(" ") + sb.WriteString(htmlEscape(m.TitleZH)) + sb.WriteString(" — 节点 ") + sb.WriteString(htmlEscape(e.NodeID)) + sb.WriteString("") + if e.Pool != "" { + sb.WriteString(" 池 ") + sb.WriteString(htmlEscape(e.Pool)) + sb.WriteString("") + } + sb.WriteString("\n") + + // Line 2: severity label + sb.WriteString("严重度: ") + sb.WriteString(string(e.Severity)) + sb.WriteString("\n") + + // Detail pairs (sorted-ish; map iteration is random, determinism not required) + for k, v := range e.Detail { + sb.WriteString(htmlEscape(k)) + sb.WriteString(": ") + sb.WriteString(htmlEscape(v)) + sb.WriteString("\n") + } + + // Runbook link + anchor := e.RunbookAnchor + if anchor == "" { + anchor = m.RunbookAnchor + } + link := anchor + if n.cfg.RunbookBaseURL != "" { + link = n.cfg.RunbookBaseURL + anchor + } + sb.WriteString("📖 处置手册") + + return sb.String() +} + +// callTG posts one sendMessage request to the Telegram Bot API. +func (n *TGNotifier) callTG(ctx context.Context, text string) error { + payload, err := json.Marshal(map[string]string{ + "chat_id": n.cfg.ChatID, + "text": text, + "parse_mode": "HTML", + "disable_web_page_preview": "true", + }) + if err != nil { + return fmt.Errorf("alert: marshal TG payload: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, n.apiURL, bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("alert: build TG request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := n.httpClient.Do(req) + if err != nil { + return fmt.Errorf("alert: TG http: %w", err) + } + defer resp.Body.Close() + io.Copy(io.Discard, resp.Body) //nolint:errcheck // response body drained for keep-alive + + if resp.StatusCode >= 500 { + return fmt.Errorf("alert: TG server error HTTP %d", resp.StatusCode) + } + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("alert: TG unexpected HTTP %d", resp.StatusCode) + } + return nil +} + +// htmlEscape escapes the five HTML special characters relevant to Telegram HTML mode. +func htmlEscape(s string) string { + s = strings.ReplaceAll(s, "&", "&") + s = strings.ReplaceAll(s, "<", "<") + s = strings.ReplaceAll(s, ">", ">") + s = strings.ReplaceAll(s, "\"", """) + return s +} diff --git a/server/internal/alert/notifier_test.go b/server/internal/alert/notifier_test.go new file mode 100644 index 0000000..19b519c --- /dev/null +++ b/server/internal/alert/notifier_test.go @@ -0,0 +1,308 @@ +package alert_test + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + + "github.com/wangjia/pangolin/server/internal/alert" +) + +// ───────────────────────────────────────────────────────────────────────────── +// Test helpers +// ───────────────────────────────────────────────────────────────────────────── + +func newTestRedis(t *testing.T) *redis.Client { + t.Helper() + mr := miniredis.RunT(t) + return redis.NewClient(&redis.Options{Addr: mr.Addr()}) +} + +// tgServer is a tiny Telegram Bot API mock. +type tgServer struct { + statusCode atomic.Int32 // HTTP status to return; default 200 + callCount atomic.Int32 // total calls received + lastBody atomic.Value // last []byte body received + srv *httptest.Server +} + +func newTGServer(t *testing.T) *tgServer { + t.Helper() + ts := &tgServer{} + ts.statusCode.Store(http.StatusOK) + ts.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ts.callCount.Add(1) + body, _ := io.ReadAll(r.Body) + ts.lastBody.Store(body) + w.WriteHeader(int(ts.statusCode.Load())) + w.Write([]byte(`{"ok":true}`)) //nolint:errcheck + })) + t.Cleanup(ts.srv.Close) + return ts +} + +// parsedBody parses the last received request body as a map. +func (ts *tgServer) parsedBody(t *testing.T) map[string]string { + t.Helper() + raw, _ := ts.lastBody.Load().([]byte) + if len(raw) == 0 { + t.Fatal("tgServer: no body received yet") + } + var m map[string]string + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("tgServer: unmarshal body: %v", err) + } + return m +} + +// newNotifier creates a TGNotifier pointing at the mock server. +func newNotifier(t *testing.T, ts *tgServer, rdb *redis.Client) *alert.TGNotifier { + t.Helper() + return alert.NewTGNotifier(alert.TGConfig{ + BotToken: "test-token", + ChatID: "-1001234567", + RunbookBaseURL: "https://example.internal/runbook", + BaseURL: ts.srv.URL, + }, rdb, nil) +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: all seven event types render correct templates with runbook anchors +// ───────────────────────────────────────────────────────────────────────────── + +func TestAllEventTypesRendered(t *testing.T) { + tests := []struct { + eventType alert.EventType + wantAnchor string + wantTitlePart string // substring expected in the TG message text (HTML-encoded if needed) + }{ + {alert.EventTypeBlockConfirmed, "#block-confirmed", "判封确认"}, + {alert.EventTypeReplenishFailed, "#replenish-failed", "补新连续失败≥3"}, + // "<" is HTML-escaped to "<" in TG HTML mode — check escaped form. + {alert.EventTypeWatermarkLow, "#watermark-low", "水位<70%"}, + {alert.EventTypeBreakerTripped, "#breaker-tripped", "熔断触发"}, + {alert.EventTypeProbeAgentLost, "#probe-agent-lost", "探针失联"}, + // ">" is HTML-escaped to ">" in TG HTML mode — check escaped form. + {alert.EventTypeHeartbeatMissing, "#heartbeat-missing", "心跳缺失>90s"}, + {alert.EventTypeFault, "#node-fault", "故障态"}, + } + + for _, tc := range tests { + t.Run(string(tc.eventType), func(t *testing.T) { + rdb := newTestRedis(t) + ts := newTGServer(t) + n := newNotifier(t, ts, rdb) + + e := alert.NewEvent(tc.eventType, "node-abc", map[string]string{ + "reason": "test reason", + }) + + ctx := context.Background() + if err := n.Notify(ctx, e); err != nil { + t.Fatalf("Notify() returned error: %v", err) + } + + // Exactly one TG call must have been made. + if got := ts.callCount.Load(); got != 1 { + t.Fatalf("TG call count = %d; want 1", got) + } + + body := ts.parsedBody(t) + text := body["text"] + + // Message must contain the Chinese event title. + if !strings.Contains(text, tc.wantTitlePart) { + t.Errorf("message does not contain %q:\n%s", tc.wantTitlePart, text) + } + + // Message must contain the runbook anchor. + if !strings.Contains(text, tc.wantAnchor) { + t.Errorf("message does not contain runbook anchor %q:\n%s", tc.wantAnchor, text) + } + + // Message must contain the node ID (internal only — no domain). + if !strings.Contains(text, "node-abc") { + t.Errorf("message does not contain node ID:\n%s", text) + } + + // parse_mode must be HTML. + if body["parse_mode"] != "HTML" { + t.Errorf("parse_mode = %q; want HTML", body["parse_mode"]) + } + }) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: dedup — second message for same (Type, NodeID) within 10 min is suppressed +// ───────────────────────────────────────────────────────────────────────────── + +func TestDedupSuppressesSecondMessage(t *testing.T) { + rdb := newTestRedis(t) + ts := newTGServer(t) + n := newNotifier(t, ts, rdb) + + ctx := context.Background() + // Use a Warning-severity event (WatermarkLow) so dedup applies. + // Note: WatermarkLow has Warning severity → dedup applies. + e := alert.NewEvent(alert.EventTypeBlockConfirmed, "node-dup", nil) + + // First call — must go through. + if err := n.Notify(ctx, e); err != nil { + t.Fatalf("first Notify() error: %v", err) + } + if got := ts.callCount.Load(); got != 1 { + t.Fatalf("after first call: TG count = %d; want 1", got) + } + + // Second call with same (Type, NodeID) — must be suppressed (no TG call). + if err := n.Notify(ctx, e); err != nil { + t.Fatalf("second Notify() error: %v", err) + } + if got := ts.callCount.Load(); got != 1 { + t.Errorf("after second call: TG count = %d; want still 1 (dedup)", got) + } + + // Third call with different NodeID — must go through (distinct dedup key). + e2 := alert.NewEvent(alert.EventTypeBlockConfirmed, "node-other", nil) + if err := n.Notify(ctx, e2); err != nil { + t.Fatalf("third Notify() error: %v", err) + } + if got := ts.callCount.Load(); got != 2 { + t.Errorf("after third call (different node): TG count = %d; want 2", got) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: critical events are never deduplicated — each call goes through +// ───────────────────────────────────────────────────────────────────────────── + +func TestCriticalEventsNotDeduped(t *testing.T) { + rdb := newTestRedis(t) + ts := newTGServer(t) + n := newNotifier(t, ts, rdb) + + ctx := context.Background() + // BreakerTripped is Critical severity — must bypass dedup. + e := alert.NewEvent(alert.EventTypeBreakerTripped, "node-crit", nil) + + for i := 1; i <= 3; i++ { + if err := n.Notify(ctx, e); err != nil { + t.Fatalf("call %d: Notify() error: %v", i, err) + } + if got := ts.callCount.Load(); int(got) != i { + t.Errorf("after call %d: TG count = %d; want %d (no dedup for critical)", i, got, i) + } + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: TG 5xx triggers retries; after exhausting retries falls back to log; +// Notify() never returns an error. +// ───────────────────────────────────────────────────────────────────────────── + +func TestTG5xxRetriesThenFallsBack(t *testing.T) { + rdb := newTestRedis(t) + ts := newTGServer(t) + + // Capture fallback calls. + var fallbackCalls atomic.Int32 + fallback := &capturingNotifier{ + fn: func(alert.Event) { fallbackCalls.Add(1) }, + } + + n := alert.NewTGNotifier(alert.TGConfig{ + BotToken: "test-token", + ChatID: "-1001234567", + BaseURL: ts.srv.URL, + }, rdb, fallback) + + // Configure mock TG server to return 500. + ts.statusCode.Store(http.StatusInternalServerError) + + ctx := context.Background() + e := alert.NewEvent(alert.EventTypeFault, "node-fail", map[string]string{"reason": "both fail"}) + + // Notify must return nil (not block the scheduler). + if err := n.Notify(ctx, e); err != nil { + t.Fatalf("Notify() must not return error; got %v", err) + } + + // TG must have been called 3 times (1 initial + 2 retries = maxRetries+1). + if got := ts.callCount.Load(); got != 3 { + t.Errorf("TG call count = %d; want 3 (1 + 2 retries)", got) + } + + // Fallback must have been called exactly once. + if got := fallbackCalls.Load(); got != 1 { + t.Errorf("fallback call count = %d; want 1", got) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: LogNotifier always succeeds (used in offline/dev mode) +// ───────────────────────────────────────────────────────────────────────────── + +func TestLogNotifier(t *testing.T) { + n := alert.LogNotifier{} + ctx := context.Background() + + for _, et := range []alert.EventType{ + alert.EventTypeBlockConfirmed, + alert.EventTypeReplenishFailed, + alert.EventTypeWatermarkLow, + alert.EventTypeBreakerTripped, + alert.EventTypeProbeAgentLost, + alert.EventTypeHeartbeatMissing, + alert.EventTypeFault, + } { + e := alert.NewEvent(et, "node-log", nil) + if err := n.Notify(ctx, e); err != nil { + t.Errorf("LogNotifier.Notify(%s) error: %v", et, err) + } + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Test: RunbookAnchor override takes precedence over type default +// ───────────────────────────────────────────────────────────────────────────── + +func TestRunbookAnchorOverride(t *testing.T) { + rdb := newTestRedis(t) + ts := newTGServer(t) + n := newNotifier(t, ts, rdb) + + e := alert.NewEvent(alert.EventTypeFault, "node-anch", nil) + e.RunbookAnchor = "#custom-anchor" + + if err := n.Notify(context.Background(), e); err != nil { + t.Fatal(err) + } + + body := ts.parsedBody(t) + if !strings.Contains(body["text"], "#custom-anchor") { + t.Errorf("expected #custom-anchor in message:\n%s", body["text"]) + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Helpers +// ───────────────────────────────────────────────────────────────────────────── + +// capturingNotifier is a Notifier that calls fn on each Notify. +type capturingNotifier struct { + fn func(alert.Event) +} + +func (c *capturingNotifier) Notify(_ context.Context, e alert.Event) error { + c.fn(e) + return nil +} diff --git a/server/internal/alert/runbook_test.go b/server/internal/alert/runbook_test.go new file mode 100644 index 0000000..07a8647 --- /dev/null +++ b/server/internal/alert/runbook_test.go @@ -0,0 +1,73 @@ +package alert_test + +import ( + "os" + "strings" + "testing" + + "github.com/wangjia/pangolin/server/internal/alert" +) + +// allEventTypes lists every EventType constant for completeness checks. +var allEventTypes = []alert.EventType{ + alert.EventTypeBlockConfirmed, + alert.EventTypeReplenishFailed, + alert.EventTypeWatermarkLow, + alert.EventTypeBreakerTripped, + alert.EventTypeProbeAgentLost, + alert.EventTypeHeartbeatMissing, + alert.EventTypeFault, +} + +// expectedAnchors maps each EventType to the expected in the runbook. +var expectedAnchors = map[alert.EventType]string{ + alert.EventTypeBlockConfirmed: "block-confirmed", + alert.EventTypeReplenishFailed: "replenish-failed", + alert.EventTypeWatermarkLow: "watermark-low", + alert.EventTypeBreakerTripped: "breaker-tripped", + alert.EventTypeProbeAgentLost: "probe-agent-lost", + alert.EventTypeHeartbeatMissing: "heartbeat-missing", + alert.EventTypeFault: "node-fault", +} + +// TestRunbookAnchorsInCode verifies that every EventType produces an Event +// whose RunbookAnchor matches the expected anchor. +func TestRunbookAnchorsInCode(t *testing.T) { + for _, et := range allEventTypes { + e := alert.NewEvent(et, "node-x", nil) + wantAnchor := "#" + expectedAnchors[et] + if e.RunbookAnchor != wantAnchor { + t.Errorf("EventType %q: RunbookAnchor = %q; want %q", + et, e.RunbookAnchor, wantAnchor) + } + } +} + +// TestRunbookFileContainsAllAnchors verifies that docs/runbook-scheduler.md +// contains an tag for every event type. +// +// The test is skipped when the file does not exist yet (so it never blocks CI +// while the runbook is being drafted) and fails once the file is present but +// an anchor is missing. +func TestRunbookFileContainsAllAnchors(t *testing.T) { + // Navigate up from server/internal/alert to the repo root, then to docs/. + // Go test sets cwd to the package directory (server/internal/alert/), + // so three levels up reaches the worktree root (where docs/ lives). + runbookPath := "../../../docs/runbook-scheduler.md" + + data, err := os.ReadFile(runbookPath) + if os.IsNotExist(err) { + t.Skip("docs/runbook-scheduler.md not found; skipping anchor check") + } + if err != nil { + t.Fatalf("read runbook: %v", err) + } + content := string(data) + + for et, anchor := range expectedAnchors { + tag := `id="` + anchor + `"` + if !strings.Contains(content, tag) { + t.Errorf("runbook missing anchor for EventType %q: expected ", et, tag) + } + } +} diff --git a/server/internal/scheduler/detect/engine.go b/server/internal/scheduler/detect/engine.go index d8a98d3..e095031 100644 --- a/server/internal/scheduler/detect/engine.go +++ b/server/internal/scheduler/detect/engine.go @@ -9,32 +9,14 @@ import ( "github.com/redis/go-redis/v9" + "github.com/wangjia/pangolin/server/internal/alert" "github.com/wangjia/pangolin/server/internal/idgen" ) -// ───────────────────────────────────────────────────────────────────────────── -// Notifier — 15G interface stub -// ───────────────────────────────────────────────────────────────────────────── - -// Notifier is the 15G event sink for fault notifications. -// The real implementation (15G) sends a Telegram/alerting message; the stub -// below writes to the structured logger and is used until 15G is ready. -type Notifier interface { - NotifyFault(ctx context.Context, nodeID, reason string) error -} - -// LogNotifier is a Notifier stub that logs via slog. -// It is used when no real Notifier is wired up. -type LogNotifier struct{} - -// NotifyFault implements Notifier. -func (LogNotifier) NotifyFault(_ context.Context, nodeID, reason string) error { - slog.Warn("node fault detected — manual review required", - "node_id", nodeID, - "reason", reason, - ) - return nil -} +// Notifier is the 15G alert outlet used by the detection engine. +// It is satisfied by alert.Notifier (the real TG implementation) and by +// alert.LogNotifier (the fallback / development stub). +type Notifier = alert.Notifier // ───────────────────────────────────────────────────────────────────────────── // Redis key constants @@ -91,7 +73,7 @@ func NewEngine( cfg = &d } if notifier == nil { - notifier = LogNotifier{} + notifier = alert.LogNotifier{} } return &Engine{ probeStore: probeStore, @@ -150,8 +132,11 @@ func (e *Engine) processNode(ctx context.Context, node NodeInfo) error { // Streaks are left unchanged so that when the node recovers the engine // resumes from its current position rather than re-triggering immediately. if isFault(sig) { - reason := fmt.Sprintf("domestic_fail_isps=%d overseas_ok=false", sig.DomesticFailISPs) - if notifyErr := e.notifier.NotifyFault(ctx, node.ID, reason); notifyErr != nil { + ev := alert.NewEvent(alert.EventTypeFault, node.ID, map[string]string{ + "domestic_fail_isps": fmt.Sprintf("%d", sig.DomesticFailISPs), + "overseas_ok": "false", + }) + if notifyErr := e.notifier.Notify(ctx, ev); notifyErr != nil { slog.Error("detect: notify fault", "node_id", node.ID, "error", notifyErr) } return nil // do not persist streak changes @@ -254,6 +239,14 @@ func (e *Engine) processSuspect(ctx context.Context, node NodeInfo, sig NodeSign return e.streaks.Save(ctx, node.ID, sk) } + // Emit 判封确认 alert (15G exit channel). + confirmedEv := alert.NewEvent(alert.EventTypeBlockConfirmed, node.ID, map[string]string{ + "suspect_streak": fmt.Sprintf("%d", sk.SuspectStreak), + }) + if notifyErr := e.notifier.Notify(ctx, confirmedEv); notifyErr != nil { + slog.Error("detect: notify block confirmed", "node_id", node.ID, "error", notifyErr) + } + // Immediately mark down — skip draining per lifecycle policy. downDetail := map[string]any{ "from": "blocked_confirmed", diff --git a/server/internal/scheduler/detect/engine_test.go b/server/internal/scheduler/detect/engine_test.go index c0decc1..69af7d6 100644 --- a/server/internal/scheduler/detect/engine_test.go +++ b/server/internal/scheduler/detect/engine_test.go @@ -9,6 +9,7 @@ import ( "github.com/alicebob/miniredis/v2" "github.com/redis/go-redis/v9" + "github.com/wangjia/pangolin/server/internal/alert" "github.com/wangjia/pangolin/server/internal/scheduler/detect" "github.com/wangjia/pangolin/server/internal/scheduler/probe" ) @@ -29,16 +30,26 @@ func (m *mockSnapshotter) SnapshotsByNode(_ context.Context, nodeID string) (map return nil, nil } -// recordingNotifier records NotifyFault calls for assertion. +// recordingNotifier records Notify calls for assertion. type recordingNotifier struct { - calls []string // nodeID values + events []alert.Event } -func (r *recordingNotifier) NotifyFault(_ context.Context, nodeID, _ string) error { - r.calls = append(r.calls, nodeID) +func (r *recordingNotifier) Notify(_ context.Context, e alert.Event) error { + r.events = append(r.events, e) return nil } +// hasFaultEvent returns true if any recorded event has type EventTypeFault. +func (r *recordingNotifier) hasFaultEvent() bool { + for _, e := range r.events { + if e.Type == alert.EventTypeFault { + return true + } + } + return false +} + // newTestRedis creates an in-process Redis (miniredis) and returns a connected // client plus a cleanup function. Tests must call cleanup() at the end. func newTestRedis(t *testing.T) (*redis.Client, *miniredis.Miniredis) { @@ -345,11 +356,11 @@ func TestRules(t *testing.T) { } // Verify fault notification. - if tc.wantFaultNotified && len(notifier.calls) == 0 { - t.Error("expected NotifyFault to be called, but it was not") + if tc.wantFaultNotified && !notifier.hasFaultEvent() { + t.Error("expected fault Notify event to be recorded, but it was not") } - if !tc.wantFaultNotified && len(notifier.calls) > 0 { - t.Errorf("unexpected NotifyFault calls: %v", notifier.calls) + if !tc.wantFaultNotified && notifier.hasFaultEvent() { + t.Errorf("unexpected fault Notify events: %v", notifier.events) } }) } @@ -594,8 +605,8 @@ func TestFaultNoTransitionNoStreak(t *testing.T) { if events := lc.Events(); len(events) != 0 { t.Errorf("unexpected events: %v", events) } - if len(notifier.calls) == 0 { - t.Error("expected NotifyFault to be called at least once") + if !notifier.hasFaultEvent() { + t.Error("expected fault Notify event to be recorded at least once") } } diff --git a/server/internal/scheduler/orchestrate/deps.go b/server/internal/scheduler/orchestrate/deps.go index 2dba0cb..933b903 100644 --- a/server/internal/scheduler/orchestrate/deps.go +++ b/server/internal/scheduler/orchestrate/deps.go @@ -8,9 +8,9 @@ package orchestrate import ( "context" - "log/slog" "time" + "github.com/wangjia/pangolin/server/internal/alert" "github.com/wangjia/pangolin/server/internal/scheduler/probe" ) @@ -164,25 +164,17 @@ func (StubBreaker) Allow(_, _ string) bool { return true } func (StubBreaker) Record(_, _ string) {} // ───────────────────────────────────────────────────────────────────────────── -// Notifier (15G stub) +// Notifier (15G) // ───────────────────────────────────────────────────────────────────────────── -// Notifier is the 15G alerting interface. -type Notifier interface { - NotifyFault(ctx context.Context, nodeID, reason string) error -} +// Notifier is the 15G unified alert outlet used by the orchestrator. +// It is satisfied by alert.Notifier (TG implementation) and alert.LogNotifier +// (fallback / development stub). +type Notifier = alert.Notifier -// LogNotifier logs faults via slog. Used when no real notifier is wired. -type LogNotifier struct{} - -// NotifyFault implements Notifier. -func (LogNotifier) NotifyFault(_ context.Context, nodeID, reason string) error { - slog.Warn("orchestrate: replacement failed — manual review required", - "node_id", nodeID, - "reason", reason, - ) - return nil -} +// LogNotifier is re-exported for callers that need a no-op Notifier without +// importing the alert package directly. +type LogNotifier = alert.LogNotifier // ───────────────────────────────────────────────────────────────────────────── // Clock (for testability) diff --git a/server/internal/scheduler/orchestrate/replacer.go b/server/internal/scheduler/orchestrate/replacer.go index ff97352..ed0c3c3 100644 --- a/server/internal/scheduler/orchestrate/replacer.go +++ b/server/internal/scheduler/orchestrate/replacer.go @@ -10,6 +10,7 @@ import ( "github.com/redis/go-redis/v9" + "github.com/wangjia/pangolin/server/internal/alert" "github.com/wangjia/pangolin/server/internal/scheduler/probe" ) @@ -95,7 +96,7 @@ func NewReplacer(cfg Config) *Replacer { cfg.Breaker = StubBreaker{} } if cfg.Notifier == nil { - cfg.Notifier = LogNotifier{} + cfg.Notifier = alert.LogNotifier{} } if cfg.Clock == nil { cfg.Clock = RealClock{} @@ -237,11 +238,21 @@ func (r *Replacer) stepPending(ctx context.Context, uuid string, rec *ReplaceRec if !r.breaker.Allow(nodeInfo.Tier, nodeInfo.Region) { slog.Info("orchestrate: breaker blocked replacement", "uuid", uuid, "tier", nodeInfo.Tier, "region", nodeInfo.Region) + // Emit 熔断触发 alert (15G exit channel). + ev := alert.NewEvent(alert.EventTypeBreakerTripped, rec.OldNode, map[string]string{ + "tier": nodeInfo.Tier, + "region": nodeInfo.Region, + "replacement_uuid": uuid, + }) + ev.Pool = nodeInfo.Tier + "/" + nodeInfo.Region + if notifyErr := r.notifier.Notify(ctx, ev); notifyErr != nil { + slog.Error("orchestrate: notify breaker tripped", "uuid", uuid, "error", notifyErr) + } return nil // stay pending; retry next Tick } // Watermark / quota check — stub (always passes). - // TODO(15F): implement real capacity-quota guard here. + // TODO(15F): emit EventTypeWatermarkLow when real capacity guard is wired. rec.Phase = PhaseCreating rec.PhaseStartedAt = r.clock.Now() @@ -371,9 +382,14 @@ func (r *Replacer) failProbeAttempt(ctx context.Context, uuid string, rec *Repla _ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err() _ = r.rdb.Expire(ctx, replaceKeyPrefix+uuid, replaceTTL).Err() - alertReason := fmt.Sprintf("probing failed after %d attempts: %s", rec.Attempts, reason) - if notifyErr := r.notifier.NotifyFault(ctx, rec.OldNode, alertReason); notifyErr != nil { - slog.Error("orchestrate: notify fault", "uuid", uuid, "error", notifyErr) + // Emit 补新连续失败≥3 alert (15G exit channel). + ev := alert.NewEvent(alert.EventTypeReplenishFailed, rec.OldNode, map[string]string{ + "attempts": fmt.Sprintf("%d", rec.Attempts), + "last_reason": reason, + "replacement_uuid": uuid, + }) + if notifyErr := r.notifier.Notify(ctx, ev); notifyErr != nil { + slog.Error("orchestrate: notify replenish failed", "uuid", uuid, "error", notifyErr) } slog.Error("orchestrate: replacement permanently failed — manual review required", "uuid", uuid, "old_node", rec.OldNode, "attempts", rec.Attempts) diff --git a/server/internal/scheduler/orchestrate/replacer_test.go b/server/internal/scheduler/orchestrate/replacer_test.go index 7fcaf4d..25ab112 100644 --- a/server/internal/scheduler/orchestrate/replacer_test.go +++ b/server/internal/scheduler/orchestrate/replacer_test.go @@ -11,6 +11,7 @@ import ( "github.com/alicebob/miniredis/v2" "github.com/redis/go-redis/v9" + "github.com/wangjia/pangolin/server/internal/alert" "github.com/wangjia/pangolin/server/internal/scheduler/orchestrate" "github.com/wangjia/pangolin/server/internal/scheduler/probe" ) @@ -267,21 +268,35 @@ func passingSnapshots() map[string]probe.ProbeSnapshot { // ───────────────────────────────────────────────────────────────────────────── type mockNotifier struct { - mu sync.Mutex - calls []string + mu sync.Mutex + events []alert.Event } -func (n *mockNotifier) NotifyFault(_ context.Context, nodeID, _ string) error { +func (n *mockNotifier) Notify(_ context.Context, e alert.Event) error { n.mu.Lock() defer n.mu.Unlock() - n.calls = append(n.calls, nodeID) + n.events = append(n.events, e) return nil } +// count returns the number of Notify calls received. func (n *mockNotifier) count() int { n.mu.Lock() defer n.mu.Unlock() - return len(n.calls) + return len(n.events) +} + +// countByType returns the number of Notify calls with the given EventType. +func (n *mockNotifier) countByType(t alert.EventType) int { + n.mu.Lock() + defer n.mu.Unlock() + c := 0 + for _, e := range n.events { + if e.Type == t { + c++ + } + } + return c } // ───────────────────────────────────────────────────────────────────────────── @@ -557,9 +572,9 @@ func TestProbeFailMaxAttempts(t *testing.T) { } } - // NotifyFault must be called exactly once. - if n := h.notifier.count(); n != 1 { - t.Errorf("NotifyFault calls = %d; want 1", n) + // Notify(EventTypeReplenishFailed) must be called exactly once. + if n := h.notifier.countByType(alert.EventTypeReplenishFailed); n != 1 { + t.Errorf("Notify(ReplenishFailed) calls = %d; want 1", n) } // Record must be in failed phase. diff --git a/server/internal/scheduler/probe/prober_agent.go b/server/internal/scheduler/probe/prober_agent.go index c0079b2..d09aa5a 100644 --- a/server/internal/scheduler/probe/prober_agent.go +++ b/server/internal/scheduler/probe/prober_agent.go @@ -34,9 +34,12 @@ import ( "log/slog" "net/http" "sort" + "strconv" "strings" "sync/atomic" "time" + + "github.com/wangjia/pangolin/server/internal/alert" ) // -------------------------------------------------------------------------- @@ -113,6 +116,11 @@ var aliyunISP = []struct { // Overridable in tests via AliyunSyntheticAgent.baseURL. const aliyunEndpoint = "https://cloudmonitor.cn-hangzhou.aliyuncs.com/" +// probeAgentLostThreshold is the number of consecutive per-ISP API failures +// that must accumulate before an EventTypeProbeAgentLost alert is emitted. +// This mirrors the "连续失败≥3" policy documented in the package comments. +const probeAgentLostThreshold = 3 + // AliyunSyntheticAgentConfig holds configuration for AliyunSyntheticAgent. // The AccessKeyID and AccessKeySecret must belong to a RAM sub-account with // minimal permissions (cloudmonitor:CreateSiteMonitor + @@ -158,6 +166,16 @@ type AliyunSyntheticAgent struct { baseURL string // overridable in tests failCount atomic.Int64 logger *slog.Logger + // notifier is the 15G exit channel for EventTypeProbeAgentLost events. + // Nil means no alerting (development / test with no TG configured). + notifier alert.Notifier +} + +// SetNotifier injects the 15G alert outlet into the agent. +// When not set, no EventTypeProbeAgentLost alerts are emitted (dev/test mode). +// Call before RunOnce / Probe. +func (a *AliyunSyntheticAgent) SetNotifier(n alert.Notifier) { + a.notifier = n } // NewAliyunSyntheticAgent creates an AliyunSyntheticAgent. @@ -255,6 +273,9 @@ func (a *AliyunSyntheticAgent) RunOnce(ctx context.Context, targets []ProbeTarge // // On any API or polling error the ISP vantage is skipped (no result returned, // no Redis write) per the degradation contract. +// +// When consecutive per-ISP API failures reach probeAgentLostThreshold the +// 探针失联 (EventTypeProbeAgentLost) alert is emitted via the injected Notifier. func (a *AliyunSyntheticAgent) Probe(ctx context.Context, target ProbeTarget) ([]VantageResult, error) { var out []VantageResult for _, isp := range aliyunISP { @@ -264,6 +285,17 @@ func (a *AliyunSyntheticAgent) Probe(ctx context.Context, target ProbeTarget) ([ a.logger.Warn("prober_agent: ISP probe failed (degraded, no data written)", "node", target.NodeID, "isp", isp.name, "error", err, "consecutive_failures", cnt) + // Emit 探针失联 alert when threshold is crossed (15G exit channel). + if cnt >= probeAgentLostThreshold && a.notifier != nil { + ev := alert.NewEvent(alert.EventTypeProbeAgentLost, "aliyun-synthetic", map[string]string{ + "consecutive_failures": strconv.FormatInt(cnt, 10), + "last_isp": isp.name, + "last_node": target.NodeID, + }) + if notifyErr := a.notifier.Notify(ctx, ev); notifyErr != nil { + a.logger.Warn("prober_agent: notify probe agent lost", "error", notifyErr) + } + } // Degradation: skip this vantage this cycle. continue } diff --git a/server/internal/scheduler/probe/store.go b/server/internal/scheduler/probe/store.go index 018a26b..626e5fb 100644 --- a/server/internal/scheduler/probe/store.go +++ b/server/internal/scheduler/probe/store.go @@ -9,6 +9,8 @@ import ( "time" "github.com/redis/go-redis/v9" + + "github.com/wangjia/pangolin/server/internal/alert" ) // Redis TTL constants for the probe subsystem. @@ -210,3 +212,50 @@ func (s *Store) AliveProbes(ctx context.Context) ([]string, error) { } return ids, nil } + +// CheckHeartbeats scans all known probe heartbeat keys and fires an +// EventTypeHeartbeatMissing alert via notifier for every probe whose last +// heartbeat timestamp is older than threshold. +// +// This is called by 15H (DetectLoop / assembly) on a periodic basis. +// Missing-key semantics apply: a key that expired (TTL elapsed) is not seen +// at all — only keys that exist but carry a stale timestamp are reported. +// +// threshold should be ≥90 s per the operational SLO. +func (s *Store) CheckHeartbeats(ctx context.Context, threshold time.Duration, notifier alert.Notifier) error { + now := time.Now().Unix() + cutoff := now - int64(threshold.Seconds()) + + var scanErr error + iter := s.rdb.Scan(ctx, 0, "probe:hb:*", 0).Iterator() + for iter.Next(ctx) { + k := iter.Val() + probeID := strings.TrimPrefix(k, "probe:hb:") + + val, err := s.rdb.Get(ctx, k).Result() + if err != nil { + // Key may have expired between SCAN and GET; skip. + continue + } + ts, err := strconv.ParseInt(val, 10, 64) + if err != nil { + continue // corrupt value; ignore + } + if ts < cutoff { + // Heartbeat is stale: emit alert. + staleSecs := now - ts + ev := alert.NewEvent(alert.EventTypeHeartbeatMissing, probeID, map[string]string{ + "stale_seconds": strconv.FormatInt(staleSecs, 10), + "threshold_s": strconv.FormatInt(int64(threshold.Seconds()), 10), + }) + if notifyErr := notifier.Notify(ctx, ev); notifyErr != nil { + // Log but continue checking other probes. + scanErr = fmt.Errorf("probe: notify heartbeat missing for %s: %w", probeID, notifyErr) + } + } + } + if err := iter.Err(); err != nil { + return fmt.Errorf("probe: scan heartbeats: %w", err) + } + return scanErr +}