feat(detect): 判定引擎 signals+rules+streak (tsk_OYEiDCzM9_0Y)
新建 server/internal/scheduler/detect/ 包，实现任务 15D： ## 核心模块 **lifecycle.go** - 定义 LifecycleService Go interface（ListNodes / TransitionStatus / SetWeight / BumpVersion / GetLoad / GetLoadHistory），供 #5 真实实现，内含乐观锁语义（0 行 受影响即本周期放弃，幂等可重入） - 提供 MockLifecycle 内存 stub（SetConflict / Events 等测试辅助方法） **signals.go** - computeSignals：读 SnapshotsByNode，归一化为 NodeSignal - domesticFailISPs：按 ISP 聚合（stripPrefix "3rd-" 合并 15B/15C 同 ISP 视角）， L3 失败权重最高（L3 挂即记该 ISP 失败，即使 L1/L2 通），无数据 vantage 不计分母 - overseasOK：境外对照点全部 L1 通为真 - trafficDropPct：15min 前后半段在线数降幅百分比 - ProbeSnapshotter interface（解耦 probe.Store，便于 mock） **rules.go** - DetectConfig：全量阈值常量（DomesticFailNumerator/Denominator=2/3、 SuspectStreakMin=2、TrafficDropThreshold=80%、TrafficBaselineMin=20、 ConfirmedStreakMin=6、RecoverStreakMin=2、SuspectWeight=10） - isSuspectTriggered：整数算术避免浮点误差（fail×den >= total×num） - isFault / effectiveSuspectStreakMin / isTrafficWarning 规则助手 **streak.go** - StreakStore：Redis hash detect:streak:{node}（fail_streak / suspect_streak / recover_streak），TTL 2h（宽于 6×5min=30min），进程重启后自动恢复 **engine.go** - Engine.Tick(ctx)：列举 up/blocked_suspect 节点，按节点依序运行五条规则 - Rule 1 suspect：≥2/3 ISP 失败 + 境外正常 → 连续 2 周期 → up→blocked_suspect， weight→10，写 probe:freq:{node}=60（1min 提频标记），BumpVersion - Rule 2 流量预警：trafficDrop≥80% + 基线≥20 → suspect 门槛放宽至 1 周期 - Rule 3 confirmed：blocked_suspect 连续 6 周期 → blocked_confirmed → down（跳过 draining），入队 detect:replace:queue（JSON: nodeId + replacementUuid） - Rule 4 recover：suspect 期间境内连续 2 周期恢复 → up（weight 交 15E warmup） - Rule 5 fault：境内+境外同时失败 → 不迁移，调 Notifier.NotifyFault（log stub） - Notifier interface + LogNotifier stub（15G 就绪前输出 slog.Warn） **engine_test.go**（19 个测试，全部通过） - 表驱动覆盖：恰好 2/3 失败、1/3 失败不触发、9min vs 10min streak、 流量预警 1 周期即 suspect、suspect 第 6 周期 confirmed、境内外同挂 fault 不迁移、 恢复 2 周期回 up、3rd- 前缀合并、L3 覆盖 L1/L2 - 乐观锁冲突：SetConflict → 放弃本周期 → 无副作用 → 解冲突后正常重试 - 进程重启：同一 miniredis，新 Engine 读取已存在 streak 继续计数不误清零 - 产出验证：replace queue JSON 结构、probe:freq 标记、weight 变更、事件 detail 字段 **migrations/000010_node_blocked_statuses.{up,down}.sql** - 扩展 nodes.status ENUM 加入 blocked_suspect / blocked_confirmed Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
wangjia
@@ -0,0 +1,335 @@
|
||||
package detect
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"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
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Redis key constants
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
const (
|
||||
// replaceQueueKey is the Redis list that 15E (Replenisher) drains to
|
||||
// provision replacement nodes. The queue entry shape is replaceRequest JSON.
|
||||
replaceQueueKey = "detect:replace:queue"
|
||||
|
||||
// probeFreqKeyPrefix is the per-node Redis key prefix written when a node
|
||||
// enters "blocked_suspect". The value "60" (seconds) tells 15B to probe
|
||||
// at 1-minute instead of the default 5-minute interval.
|
||||
probeFreqKeyPrefix = "probe:freq:"
|
||||
|
||||
// probeFreqTTL keeps the elevated-frequency marker alive for 45 minutes —
|
||||
// long enough to cover a full suspect → confirmed window (6 × 5 min + buffer).
|
||||
probeFreqTTL = 45 * time.Minute
|
||||
)
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Engine
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// Engine executes one detection cycle per Tick call.
|
||||
//
|
||||
// 15H (DetectLoop) is responsible for:
|
||||
// - scheduling Ticks at 5-minute intervals, and
|
||||
// - leader election (only the elected leader should call Tick).
|
||||
//
|
||||
// Engine itself holds no goroutines and no in-process state beyond its
|
||||
// constructor arguments; all persistent state lives in Redis (via StreakStore).
|
||||
type Engine struct {
|
||||
probeStore ProbeSnapshotter
|
||||
lc LifecycleService
|
||||
streaks *StreakStore
|
||||
rdb *redis.Client
|
||||
notifier Notifier
|
||||
cfg *DetectConfig
|
||||
}
|
||||
|
||||
// NewEngine creates an Engine. If cfg is nil, DefaultConfig() is used.
|
||||
// If notifier is nil, LogNotifier is used.
|
||||
func NewEngine(
|
||||
probeStore ProbeSnapshotter,
|
||||
lc LifecycleService,
|
||||
streaks *StreakStore,
|
||||
rdb *redis.Client,
|
||||
notifier Notifier,
|
||||
cfg *DetectConfig,
|
||||
) *Engine {
|
||||
if cfg == nil {
|
||||
d := DefaultConfig()
|
||||
cfg = &d
|
||||
}
|
||||
if notifier == nil {
|
||||
notifier = LogNotifier{}
|
||||
}
|
||||
return &Engine{
|
||||
probeStore: probeStore,
|
||||
lc: lc,
|
||||
streaks: streaks,
|
||||
rdb: rdb,
|
||||
notifier: notifier,
|
||||
cfg: cfg,
|
||||
}
|
||||
}
|
||||
|
||||
// Tick executes one full detection cycle.
|
||||
//
|
||||
// It lists nodes in "up" and "blocked_suspect" states, reads their probe
|
||||
// snapshots, computes signals, and applies the five classification rules.
|
||||
// Errors from individual nodes are logged but do not halt the cycle for other
|
||||
// nodes.
|
||||
func (e *Engine) Tick(ctx context.Context) error {
|
||||
nodes, err := e.lc.ListNodes(ctx, NodeFilter{
|
||||
Statuses: []NodeStatus{StatusUp, StatusBlockedSuspect},
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("detect: list nodes: %w", err)
|
||||
}
|
||||
|
||||
for _, node := range nodes {
|
||||
if err := e.processNode(ctx, node); err != nil {
|
||||
slog.Error("detect: process node",
|
||||
"node_id", node.ID,
|
||||
"status", node.Status,
|
||||
"error", err,
|
||||
)
|
||||
// Continue — one node failure must not halt the full cycle.
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// processNode applies all five rules to a single node.
|
||||
func (e *Engine) processNode(ctx context.Context, node NodeInfo) error {
|
||||
snapshots, err := e.probeStore.SnapshotsByNode(ctx, node.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("snapshots: %w", err)
|
||||
}
|
||||
|
||||
sig := computeSignals(ctx, node.ID, snapshots, e.lc)
|
||||
|
||||
sk, err := e.streaks.Load(ctx, node.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("streak load: %w", err)
|
||||
}
|
||||
|
||||
// ── Rule 5: fault ─────────────────────────────────────────────────────────
|
||||
// Both domestic AND overseas probes failing → node-level outage, not GFW.
|
||||
// Do NOT transition status; send an alert event for human review.
|
||||
// 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 {
|
||||
slog.Error("detect: notify fault", "node_id", node.ID, "error", notifyErr)
|
||||
}
|
||||
return nil // do not persist streak changes
|
||||
}
|
||||
|
||||
switch node.Status {
|
||||
case StatusUp:
|
||||
return e.processUp(ctx, node, sig, sk)
|
||||
case StatusBlockedSuspect:
|
||||
return e.processSuspect(ctx, node, sig, sk)
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// processUp applies the suspect and traffic-warning rules to an "up" node.
|
||||
func (e *Engine) processUp(ctx context.Context, node NodeInfo, sig NodeSignal, sk Streak) error {
|
||||
domesticFailing := e.cfg.isSuspectTriggered(sig)
|
||||
overseasBarrier := sig.OverseasHasData && sig.OverseasOK
|
||||
|
||||
if domesticFailing && overseasBarrier {
|
||||
// Domestic probes failing, overseas OK → potential GFW block.
|
||||
threshold := e.cfg.effectiveSuspectStreakMin(sig)
|
||||
|
||||
if sk.FailStreak+1 >= threshold {
|
||||
// ── Rule 1: suspect ───────────────────────────────────────────────
|
||||
detail := map[string]any{
|
||||
"from": "up",
|
||||
"to": "blocked_suspect",
|
||||
"domestic_fail_isps": sig.DomesticFailISPs,
|
||||
"domestic_total": sig.DomesticTotalISPs,
|
||||
"fail_streak": sk.FailStreak + 1,
|
||||
}
|
||||
affected, err := e.lc.TransitionStatus(ctx, node.ID, StatusUp, StatusBlockedSuspect, detail)
|
||||
if err != nil {
|
||||
return fmt.Errorf("transition up→suspect: %w", err)
|
||||
}
|
||||
if affected == 0 {
|
||||
// Optimistic lock conflict: another writer changed the state
|
||||
// concurrently. Leave streak unchanged; the engine is idempotent.
|
||||
slog.Info("detect: suspect transition skipped (lock conflict)", "node_id", node.ID)
|
||||
return e.streaks.Save(ctx, node.ID, sk)
|
||||
}
|
||||
|
||||
// Transition succeeded: apply side-effects.
|
||||
if err := e.lc.SetWeight(ctx, node.ID, e.cfg.SuspectWeight); err != nil {
|
||||
slog.Error("detect: set suspect weight", "node_id", node.ID, "error", err)
|
||||
}
|
||||
if err := e.writeProbeFreqMark(ctx, node.ID); err != nil {
|
||||
slog.Error("detect: write probe freq mark", "node_id", node.ID, "error", err)
|
||||
}
|
||||
if err := e.lc.BumpVersion(ctx); err != nil {
|
||||
slog.Error("detect: bump version after suspect", "node_id", node.ID, "error", err)
|
||||
}
|
||||
|
||||
// Reset fail streak; suspect streak begins on the next call to
|
||||
// processSuspect.
|
||||
sk.FailStreak = 0
|
||||
sk.SuspectStreak = 0
|
||||
sk.RecoverStreak = 0
|
||||
} else {
|
||||
// Below the threshold: increment and wait for next cycle.
|
||||
sk.FailStreak++
|
||||
sk.RecoverStreak = 0
|
||||
}
|
||||
} else {
|
||||
// Domestic probes OK (or overseas data absent / also failing — fault is
|
||||
// handled above before this function is called).
|
||||
sk.FailStreak = 0
|
||||
sk.RecoverStreak = 0
|
||||
}
|
||||
|
||||
return e.streaks.Save(ctx, node.ID, sk)
|
||||
}
|
||||
|
||||
// processSuspect applies the confirmed and recover rules to a "blocked_suspect" node.
|
||||
func (e *Engine) processSuspect(ctx context.Context, node NodeInfo, sig NodeSignal, sk Streak) error {
|
||||
domesticFailing := e.cfg.isSuspectTriggered(sig)
|
||||
overseasBarrier := sig.OverseasHasData && sig.OverseasOK
|
||||
|
||||
if domesticFailing && overseasBarrier {
|
||||
// Still looks like a GFW block: advance the suspect streak.
|
||||
sk.SuspectStreak++
|
||||
sk.RecoverStreak = 0
|
||||
sk.FailStreak = 0
|
||||
|
||||
if sk.SuspectStreak >= e.cfg.ConfirmedStreakMin {
|
||||
// ── Rule 3: confirmed ─────────────────────────────────────────────
|
||||
confirmedDetail := map[string]any{
|
||||
"from": "blocked_suspect",
|
||||
"to": "blocked_confirmed",
|
||||
"suspect_streak": sk.SuspectStreak,
|
||||
}
|
||||
n1, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedSuspect, StatusBlockedConfirmed, confirmedDetail)
|
||||
if err != nil {
|
||||
return fmt.Errorf("transition suspect→confirmed: %w", err)
|
||||
}
|
||||
if n1 == 0 {
|
||||
slog.Info("detect: confirmed transition skipped (lock conflict)", "node_id", node.ID)
|
||||
return e.streaks.Save(ctx, node.ID, sk)
|
||||
}
|
||||
|
||||
// Immediately mark down — skip draining per lifecycle policy.
|
||||
downDetail := map[string]any{
|
||||
"from": "blocked_confirmed",
|
||||
"to": "down",
|
||||
"note": "skipped draining (confirmed block policy)",
|
||||
}
|
||||
if _, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedConfirmed, StatusDown, downDetail); err != nil {
|
||||
return fmt.Errorf("transition confirmed→down: %w", err)
|
||||
}
|
||||
|
||||
// Push to 15E replace queue.
|
||||
if err := e.pushReplaceQueue(ctx, node.ID); err != nil {
|
||||
slog.Error("detect: push replace queue", "node_id", node.ID, "error", err)
|
||||
}
|
||||
|
||||
// Reset all streaks; the node is now down and will not be processed
|
||||
// again (ListNodes only returns "up" and "blocked_suspect").
|
||||
sk = Streak{}
|
||||
}
|
||||
} else {
|
||||
// Domestic probes passing (or overseas data absent — treat as recovery).
|
||||
// ── Rule 4: recover ───────────────────────────────────────────────────
|
||||
sk.RecoverStreak++
|
||||
sk.SuspectStreak = 0
|
||||
sk.FailStreak = 0
|
||||
|
||||
if sk.RecoverStreak >= e.cfg.RecoverStreakMin {
|
||||
recoverDetail := map[string]any{
|
||||
"from": "blocked_suspect",
|
||||
"to": "up",
|
||||
"recover_streak": sk.RecoverStreak,
|
||||
}
|
||||
affected, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedSuspect, StatusUp, recoverDetail)
|
||||
if err != nil {
|
||||
return fmt.Errorf("transition suspect→up: %w", err)
|
||||
}
|
||||
if affected == 0 {
|
||||
slog.Info("detect: recover transition skipped (lock conflict)", "node_id", node.ID)
|
||||
return e.streaks.Save(ctx, node.ID, sk)
|
||||
}
|
||||
// Weight recovery is delegated to 15E's warmup/gradual-ramp channel.
|
||||
// The engine intentionally does NOT call SetWeight here.
|
||||
sk = Streak{}
|
||||
}
|
||||
}
|
||||
|
||||
return e.streaks.Save(ctx, node.ID, sk)
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Helpers
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// writeProbeFreqMark sets a Redis key that tells 15B to probe this node at
|
||||
// 1-minute instead of the default 5-minute interval while it is suspect.
|
||||
func (e *Engine) writeProbeFreqMark(ctx context.Context, nodeID string) error {
|
||||
key := probeFreqKeyPrefix + nodeID
|
||||
return e.rdb.Set(ctx, key, "60", probeFreqTTL).Err()
|
||||
}
|
||||
|
||||
// replaceRequest is the payload pushed to the 15E replace queue.
|
||||
type replaceRequest struct {
|
||||
NodeID string `json:"nodeId"`
|
||||
ReplacementUUID string `json:"replacementUuid"`
|
||||
}
|
||||
|
||||
// pushReplaceQueue enqueues a replacement request for 15E (Replenisher).
|
||||
// The replacement UUID is pre-allocated here so 15E can reference it without
|
||||
// generating its own ID.
|
||||
func (e *Engine) pushReplaceQueue(ctx context.Context, nodeID string) error {
|
||||
payload, err := json.Marshal(replaceRequest{
|
||||
NodeID: nodeID,
|
||||
ReplacementUUID: idgen.NewString(),
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("detect: marshal replace request: %w", err)
|
||||
}
|
||||
return e.rdb.LPush(ctx, replaceQueueKey, payload).Err()
|
||||
}
|
||||
Reference in New Issue
Block a user