mirror of
https://github.com/Buriburizaem0n/nezha_domains.git
synced 2026-11-04 16:40:21 +00:00
fix(alert): prevent persisted rule crash loops
This commit is contained in:
naiba committed
1 parent
8f1147c800
commit
19daf4ad3c
8 files changed
+479
-71
No files matched your search
@@ -33,7 +33,7 @@ var (
|
||||
|
||||
// addCycleTransferStatsInfo 向AlertsCycleTransferStatsStore中添加周期流量报警统计信息
|
||||
func addCycleTransferStatsInfo(alert *model.AlertRule) {
|
||||
if !alert.Enabled() {
|
||||
if alert == nil || !alert.Enabled() || !alert.IsSafeToEvaluate() {
|
||||
return
|
||||
}
|
||||
for _, rule := range alert.Rules {
|
||||
@@ -67,8 +67,16 @@ func AlertSentinelStart() {
|
||||
panic(err)
|
||||
}
|
||||
for _, alert := range Alerts {
|
||||
if alert == nil {
|
||||
log.Printf("NEZHA>> Skipping invalid nil alert rule loaded from database")
|
||||
continue
|
||||
}
|
||||
alertsStore[alert.ID] = make(map[uint64][][]bool)
|
||||
alertsPrevState[alert.ID] = make(map[uint64]uint8)
|
||||
if !alert.IsSafeToEvaluate() {
|
||||
log.Printf("NEZHA>> Skipping invalid alert rule %d loaded from database", alert.ID)
|
||||
continue
|
||||
}
|
||||
addCycleTransferStatsInfo(alert)
|
||||
}
|
||||
AlertsLock.Unlock()
|
||||
@@ -136,7 +144,7 @@ func checkStatus() {
|
||||
|
||||
for _, alert := range Alerts {
|
||||
// 跳过未启用
|
||||
if !alert.Enabled() {
|
||||
if alert == nil || !alert.Enabled() || !alert.IsSafeToEvaluate() {
|
||||
continue
|
||||
}
|
||||
for _, server := range m {
|
||||
@@ -152,48 +160,62 @@ func checkStatus() {
|
||||
if alert.UserID != server.GetUserID() && !role.IsAdmin() {
|
||||
continue
|
||||
}
|
||||
alertsStore[alert.ID][server.ID] = append(alertsStore[alert.
|
||||
ID][server.ID], alert.Snapshot(AlertsCycleTransferStatsStore[alert.ID], server, DB))
|
||||
// 发送通知，分为触发报警和恢复通知
|
||||
_, passed := alert.Check(alertsStore[alert.ID][server.ID])
|
||||
// 保存当前服务器状态信息
|
||||
curServer := model.Server{}
|
||||
copier.Copy(&curServer, server)
|
||||
|
||||
// 本次未通过检查
|
||||
if !passed {
|
||||
// 始终触发模式或上次检查不为失败时触发报警（跳过单次触发+上次失败的情况）
|
||||
if alert.TriggerMode == model.ModeAlwaysTrigger || alertsPrevState[alert.ID][server.ID] != _RuleCheckFail {
|
||||
alertsPrevState[alert.ID][server.ID] = _RuleCheckFail
|
||||
message := fmt.Sprintf("[%s] %s(%s) %s", Localizer.T("Incident"),
|
||||
server.Name, IPDesensitize(server.GeoIP.IP.Join()), alert.Name)
|
||||
go CronShared.SendTriggerTasks(alert.FailTriggerTasks, curServer.ID, alert.UserID)
|
||||
go NotificationShared.SendNotification(alert.NotificationGroupID, message, NotificationMuteLabel.ServerIncident(server.ID, alert.ID), &curServer)
|
||||
// 清除恢复通知的静音缓存
|
||||
NotificationShared.UnMuteNotification(alert.NotificationGroupID, NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID))
|
||||
}
|
||||
} else {
|
||||
// 本次通过检查但上一次的状态为失败，则发送恢复通知
|
||||
if alertsPrevState[alert.ID][server.ID] == _RuleCheckFail {
|
||||
message := fmt.Sprintf("[%s] %s(%s) %s", Localizer.T("Resolved"),
|
||||
server.Name, IPDesensitize(server.GeoIP.IP.Join()), alert.Name)
|
||||
go CronShared.SendTriggerTasks(alert.RecoverTriggerTasks, curServer.ID, alert.UserID)
|
||||
go NotificationShared.SendNotification(alert.NotificationGroupID, message, NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID), &curServer)
|
||||
// 清除失败通知的静音缓存
|
||||
NotificationShared.UnMuteNotification(alert.NotificationGroupID, NotificationMuteLabel.ServerIncident(server.ID, alert.ID))
|
||||
}
|
||||
alertsPrevState[alert.ID][server.ID] = _RuleCheckPass
|
||||
}
|
||||
// 清理旧数据：保留窗口由规则定义决定（各规则 Duration 的最大值），
|
||||
// 而非 Check 的判定结果。window==0 表示没有任何有效规则需要回看历史
|
||||
// （例如全部 Duration<=0），此时清空采样避免切片无限增长。
|
||||
window := alert.RetentionWindow()
|
||||
samples := alertsStore[alert.ID][server.ID]
|
||||
if window <= 0 {
|
||||
alertsStore[alert.ID][server.ID] = samples[:0]
|
||||
} else if window < len(samples) {
|
||||
alertsStore[alert.ID][server.ID] = samples[len(samples)-window:]
|
||||
}
|
||||
checkStatusForServer(alert, server)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// checkStatusForServer isolates each alert/server evaluation. Input validation
|
||||
// and model-level guards handle known malformed states; this recovery boundary
|
||||
// ensures an unexpected evaluator panic cannot terminate the dashboard process
|
||||
// or prevent unrelated servers from being checked on the same tick.
|
||||
func checkStatusForServer(alert *model.AlertRule, server *model.Server) {
|
||||
defer func() {
|
||||
if recovered := recover(); recovered != nil {
|
||||
log.Printf("NEZHA>> Recovered panic evaluating alert rule %d for server %d: %v", alert.ID, server.ID, recovered)
|
||||
}
|
||||
}()
|
||||
|
||||
alertsStore[alert.ID][server.ID] = append(alertsStore[alert.
|
||||
ID][server.ID], alert.Snapshot(AlertsCycleTransferStatsStore[alert.ID], server, DB))
|
||||
// 发送通知，分为触发报警和恢复通知
|
||||
_, passed := alert.Check(alertsStore[alert.ID][server.ID])
|
||||
// 保存当前服务器状态信息
|
||||
curServer := model.Server{}
|
||||
copier.Copy(&curServer, server)
|
||||
|
||||
// 本次未通过检查
|
||||
if !passed {
|
||||
// 始终触发模式或上次检查不为失败时触发报警（跳过单次触发+上次失败的情况）
|
||||
if alert.TriggerMode == model.ModeAlwaysTrigger || alertsPrevState[alert.ID][server.ID] != _RuleCheckFail {
|
||||
alertsPrevState[alert.ID][server.ID] = _RuleCheckFail
|
||||
message := fmt.Sprintf("[%s] %s(%s) %s", Localizer.T("Incident"),
|
||||
server.Name, IPDesensitize(server.GeoIP.IP.Join()), alert.Name)
|
||||
go CronShared.SendTriggerTasks(alert.FailTriggerTasks, curServer.ID, alert.UserID)
|
||||
go NotificationShared.SendNotification(alert.NotificationGroupID, message, NotificationMuteLabel.ServerIncident(server.ID, alert.ID), &curServer)
|
||||
// 清除恢复通知的静音缓存
|
||||
NotificationShared.UnMuteNotification(alert.NotificationGroupID, NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID))
|
||||
}
|
||||
} else {
|
||||
// 本次通过检查但上一次的状态为失败，则发送恢复通知
|
||||
if alertsPrevState[alert.ID][server.ID] == _RuleCheckFail {
|
||||
message := fmt.Sprintf("[%s] %s(%s) %s", Localizer.T("Resolved"),
|
||||
server.Name, IPDesensitize(server.GeoIP.IP.Join()), alert.Name)
|
||||
go CronShared.SendTriggerTasks(alert.RecoverTriggerTasks, curServer.ID, alert.UserID)
|
||||
go NotificationShared.SendNotification(alert.NotificationGroupID, message, NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID), &curServer)
|
||||
// 清除失败通知的静音缓存
|
||||
NotificationShared.UnMuteNotification(alert.NotificationGroupID, NotificationMuteLabel.ServerIncident(server.ID, alert.ID))
|
||||
}
|
||||
alertsPrevState[alert.ID][server.ID] = _RuleCheckPass
|
||||
}
|
||||
// 清理旧数据：保留窗口由规则定义决定（各规则 Duration 的最大值），
|
||||
// 而非 Check 的判定结果。window==0 表示没有任何有效规则需要回看历史
|
||||
// （例如全部 Duration<=0），此时清空采样避免切片无限增长。
|
||||
window := alert.RetentionWindow()
|
||||
samples := alertsStore[alert.ID][server.ID]
|
||||
if window <= 0 {
|
||||
alertsStore[alert.ID][server.ID] = samples[:0]
|
||||
} else if window < len(samples) {
|
||||
alertsStore[alert.ID][server.ID] = samples[len(samples)-window:]
|
||||
}
|
||||
}
|
||||
@@ -117,3 +117,53 @@ func TestCheckStatus_SampleMemoryBounded(t *testing.T) {
|
||||
t.Fatalf("sample capacity grew unbounded: peak cap %d exceeds 4x window %d", peakCap, duration*4)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckStatusForServerRecoversUnexpectedEvaluatorPanic(t *testing.T) {
|
||||
originalStore := alertsStore
|
||||
originalPrevState := alertsPrevState
|
||||
originalCycleStore := AlertsCycleTransferStatsStore
|
||||
alertsStore = map[uint64]map[uint64][][]bool{}
|
||||
alertsPrevState = map[uint64]map[uint64]uint8{}
|
||||
AlertsCycleTransferStatsStore = map[uint64]*model.CycleTransferStats{}
|
||||
t.Cleanup(func() {
|
||||
alertsStore = originalStore
|
||||
alertsPrevState = originalPrevState
|
||||
AlertsCycleTransferStatsStore = originalCycleStore
|
||||
})
|
||||
|
||||
enabled := true
|
||||
alert := &model.AlertRule{
|
||||
Common: model.Common{ID: 77, UserID: 2},
|
||||
Enable: &enabled,
|
||||
Rules: []*model.Rule{{Type: "cpu", Duration: 3, Cover: model.RuleCoverAll}},
|
||||
}
|
||||
server := &model.Server{Common: model.Common{ID: 1, UserID: 2}}
|
||||
model.InitServer(server)
|
||||
|
||||
// The deliberately missing alertsStore[alert.ID] map panics at assignment
|
||||
// after evaluation. The per-alert/server recovery boundary must contain it.
|
||||
checkStatusForServer(alert, server)
|
||||
}
|
||||
|
||||
func TestAddCycleTransferStatsSkipsPoisonedPersistedRule(t *testing.T) {
|
||||
original := AlertsCycleTransferStatsStore
|
||||
AlertsCycleTransferStatsStore = map[uint64]*model.CycleTransferStats{}
|
||||
t.Cleanup(func() { AlertsCycleTransferStatsStore = original })
|
||||
|
||||
enabled := true
|
||||
alert := &model.AlertRule{
|
||||
Common: model.Common{ID: 88},
|
||||
Enable: &enabled,
|
||||
Rules: []*model.Rule{{
|
||||
Type: "transfer_in_cycle",
|
||||
CycleInterval: 1,
|
||||
Cover: model.RuleCoverAll,
|
||||
// CycleStart intentionally nil, as it may be in a poisoned legacy DB row.
|
||||
}},
|
||||
}
|
||||
|
||||
addCycleTransferStatsInfo(alert)
|
||||
if _, exists := AlertsCycleTransferStatsStore[alert.ID]; exists {
|
||||
t.Fatal("invalid persisted cycle rule must not be scheduled")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user