wangjia/pangolin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity

feat(alert): 统一告警出口 TG bot + runbook [tsk_9YMHMTfWJyNB]

Browse Source 
新增 server/internal/alert 包（15G）：
- 定义 Notifier 接口及 7 种 EventType（判封确认/补新失败/水位低/熔断/
  探针失联/心跳缺失/故障态）
- TGNotifier：Bot API 发送，Critical 事件不去重，Warning/Info 事件
  10min SETNX 去重窗口，失败重试 ≤2 次后降级至 LogNotifier
- LogNotifier：slog 结构化降级实现
- 单测：7 种事件模板 + runbook 锚点正确性；去重窗口内第二条被抑制；
  TG 5xx 重试后 fallback 且 Notify() 返回 nil；
  runbook 文件锚点与枚举一致性

接入 scheduler（替换旧的 NotifyFault 桩）：
- detect/engine.go：故障态（Rule 5）→ EventTypeFault；
  判封确认（Rule 3）→ EventTypeBlockConfirmed
- orchestrate/deps.go：Notifier 类型别名指向 alert.Notifier
- orchestrate/replacer.go：补新失败 → EventTypeReplenishFailed；
  熔断触发 → EventTypeBreakerTripped
- probe/prober_agent.go：failCount ≥3 → EventTypeProbeAgentLost
- probe/store.go：新增 CheckHeartbeats() 供 15H 检测心跳缺失>90s

新增 docs/runbook-scheduler.md：7 节各含含义/先查什么/处置/升级条件，
锚点与代码枚举对应。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-06-16 00:52:26 +08:00
parent cadd527680
commit 5d4b484646
11 changed files with 1387 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/scheduler/detect/engine.go
+19 -26
View File
@@ -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",
server/internal/scheduler/detect/engine_test.go
+21 -10
View File
@@ -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")
}
}
server/internal/scheduler/orchestrate/deps.go
+9 -17
View File
@@ -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)
server/internal/scheduler/orchestrate/replacer.go
+21 -5
View File
@@ -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)
server/internal/scheduler/orchestrate/replacer_test.go
+23 -8
View File
@@ -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.
server/internal/scheduler/probe/prober_agent.go
+32
View File
@@ -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
}
server/internal/scheduler/probe/store.go
+49
View File
@@ -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
}