From 8328047aa8aeca0a2ec19c00c930e968eeddee58 Mon Sep 17 00:00:00 2001 From: Kris <116849421+cantoblanco@users.noreply.github.com> Date: Thu, 1 Oct 2026 18:31:32 +0200 Subject: [PATCH] fix(alerts): evaluate fresh reports and order notification delivery (#1242) * fix(alerts): evaluate fresh reports and make notification transitions reliable * fix(alerts): identify equal-timestamp reports by sequence * fix(alerts): preserve notification transition semantics --------- Co-authored-by: naiba --- cmd/dashboard/controller/alertrule.go | 6 + model/alert_evaluation_test.go | 205 +++++++++ model/alert_timed_test.go | 104 +++++ model/alertrule.go | 149 ++++++- model/notification.go | 102 ++--- model/rule.go | 262 +++++++----- model/server.go | 24 +- service/singleton/alertsentinel.go | 211 ++++++++-- .../singleton/alertsentinel_freshness_test.go | 281 +++++++++++++ service/singleton/alertsentinel_test.go | 2 +- service/singleton/notification.go | 359 +++++++++++++--- service/singleton/notification_test.go | 393 ++++++++++++++++++ service/singleton/server.go | 1 + service/singleton/servicesentinel.go | 330 +++++++++------ .../singleton/servicesentinel_window_test.go | 63 +++ service/singleton/singleton.go | 2 +- 16 files changed, 2122 insertions(+), 372 deletions(-) create mode 100644 model/alert_evaluation_test.go create mode 100644 model/alert_timed_test.go create mode 100644 service/singleton/alertsentinel_freshness_test.go create mode 100644 service/singleton/servicesentinel_window_test.go diff --git a/cmd/dashboard/controller/alertrule.go b/cmd/dashboard/controller/alertrule.go index 85021a47..5e5ce2d0 100644 --- a/cmd/dashboard/controller/alertrule.go +++ b/cmd/dashboard/controller/alertrule.go @@ -178,6 +178,9 @@ func validateRule(c *gin.Context, r *model.AlertRule) error { if !rule.IsSupportedType() { return singleton.Localizer.ErrorT("unsupported rule type") } + if !rule.HasSafeThresholds() { + return singleton.Localizer.ErrorT("invalid rule thresholds") + } switch rule.Cover { case model.RuleCoverAll, model.RuleCoverIgnoreAll: default: @@ -192,6 +195,9 @@ func validateRule(c *gin.Context, r *model.AlertRule) error { return singleton.Localizer.ErrorT("duration is too large") } } else { + if !rule.HasSafeCycleUnit() { + return singleton.Localizer.ErrorT("invalid cycle unit") + } if rule.CycleInterval < 1 { return singleton.Localizer.ErrorT("cycle_interval need to be at least 1") } diff --git a/model/alert_evaluation_test.go b/model/alert_evaluation_test.go new file mode 100644 index 00000000..3267685c --- /dev/null +++ b/model/alert_evaluation_test.go @@ -0,0 +1,205 @@ +package model + +import ( + "math" + "testing" + "time" + + "gorm.io/driver/sqlite" + "gorm.io/gorm" +) + +func TestAlertEvaluationDistinguishesUnknownFromBreach(t *testing.T) { + server := &Server{Common: Common{ID: 1}} + rule := &Rule{Type: "cpu", Max: 80, Cover: RuleCoverAll} + if passed, known := rule.Evaluate(nil, server, nil); !passed || known { + t.Fatalf("missing state = (%v, %v), want unknown", passed, known) + } + lease := server.AttachStateStream(runtimeOwnershipStream{}) + if !lease.UpdateState(&HostState{CPU: 99}, time.Now()) { + t.Fatal("state update rejected") + } + if passed, known := rule.Evaluate(nil, server, nil); passed || !known { + t.Fatalf("fresh high CPU = (%v, %v), want known breach", passed, known) + } + if !lease.Clear() { + t.Fatal("stream clear rejected") + } + // The raw metric remains available for historical display. The sentinel + // enforces freshness and must not count this value after disconnect. + if !server.RuntimeSnapshot().LastActive.IsZero() { + t.Fatal("disconnect did not clear online visibility") + } +} + +func newCycleTestDB(t *testing.T, migrate bool) *gorm.DB { + t.Helper() + db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if migrate { + if err := db.AutoMigrate(&Transfer{}); err != nil { + t.Fatal(err) + } + } + return db +} + +func newCycleTestStats() *CycleTransferStats { + return &CycleTransferStats{ + ServerName: make(map[uint64]string), + Transfer: make(map[uint64]uint64), + NextUpdate: make(map[uint64]time.Time), + } +} + +func TestCycleRuleRejectsInvalidConfiguration(t *testing.T) { + start := time.Now().Add(-time.Hour) + rule := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Max: 100} + if !rule.HasSafeCycleConfiguration() { + t.Fatal("normal hourly cycle rejected") + } + for _, tc := range []struct { + name string + mutate func(*Rule) + }{ + {"unknown unit", func(r *Rule) { r.CycleUnit = "fortnight" }}, + {"negative limit", func(r *Rule) { r.Max = -1 }}, + {"NaN limit", func(r *Rule) { r.Max = math.NaN() }}, + {"reversed limits", func(r *Rule) { r.Min, r.Max = 200, 100 }}, + {"future start", func(r *Rule) { future := time.Now().Add(time.Hour); r.CycleStart = &future }}, + } { + t.Run(tc.name, func(t *testing.T) { + copy := *rule + tc.mutate(©) + if copy.HasSafeCycleConfiguration() { + t.Fatal("unsafe cycle accepted") + } + }) + } +} + +func TestCycleBoundsUseOneClockAtBoundary(t *testing.T) { + start := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) + rule := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Max: 100} + before := start.Add(59*time.Minute + 59*time.Second) + from, to := rule.transferDurationBounds(before) + if !from.Equal(start) || !to.Equal(start.Add(time.Hour)) { + t.Fatalf("before boundary = %v..%v", from, to) + } + from, to = rule.transferDurationBounds(start.Add(time.Hour)) + if !from.Equal(start.Add(time.Hour)) || !to.Equal(start.Add(2*time.Hour)) { + t.Fatalf("at boundary = %v..%v", from, to) + } +} + +func TestCycleQueryFailureIsUnknownAndDoesNotCache(t *testing.T) { + start := time.Now().Add(-time.Hour) + rule := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Max: 50} + server := &Server{Common: Common{ID: 1}, State: &HostState{}} + passed, known := rule.Evaluate(newCycleTestStats(), server, newCycleTestDB(t, false)) + if !passed || known || len(rule.NextTransferAt) != 0 { + t.Fatalf("SQL error yielded passed=%v known=%v cache=%v", passed, known, rule.NextTransferAt) + } +} + +func TestCycleQueryExcludesFutureRowsAndHandlesMinOnly(t *testing.T) { + start := time.Now().Add(-90 * time.Minute) + db := newCycleTestDB(t, true) + server := &Server{Common: Common{ID: 1}, State: &HostState{}} + for _, row := range []Transfer{ + {Common: Common{CreatedAt: time.Now().Add(-time.Minute)}, ServerID: 1, Out: 10}, + {Common: Common{CreatedAt: time.Now().Add(10 * time.Minute)}, ServerID: 1, Out: 100}, + } { + if err := db.Create(&row).Error; err != nil { + t.Fatal(err) + } + } + rule := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Max: 50} + stats := newCycleTestStats() + passed, known := rule.Evaluate(stats, server, db) + if !passed || !known || stats.Transfer[1] != 10 { + t.Fatalf("future row included: passed=%v known=%v amount=%d", passed, known, stats.Transfer[1]) + } + minOnly := &Rule{Type: "transfer_out_cycle", CycleStart: &start, CycleInterval: 1, CycleUnit: "hour", Min: 20} + passed, known = minOnly.Evaluate(newCycleTestStats(), server, db) + if passed || !known { + t.Fatalf("min-only cycle = (%v,%v), want known breach", passed, known) + } +} + +func TestAlertEvaluationAllNetworkSpeedUsesBothDirections(t *testing.T) { + for _, tc := range []struct { + name string + in, out uint64 + breach bool + }{ + {"inbound only", 100, 0, true}, + {"outbound below combined limit", 0, 30, false}, + {"both directions", 25, 30, true}, + } { + t.Run(tc.name, func(t *testing.T) { + server := &Server{Common: Common{ID: 1}, State: &HostState{NetInSpeed: tc.in, NetOutSpeed: tc.out}} + rule := &Rule{Type: "net_all_speed", Max: 50, Cover: RuleCoverAll} + passed, known := rule.Evaluate(nil, server, nil) + if !known || passed == tc.breach { + t.Fatalf("got passed=%v known=%v, want breach=%v", passed, known, tc.breach) + } + }) + } +} + +func TestAlertEvaluationUnavailableSensorsAreUnknown(t *testing.T) { + tests := []struct { + name string + rule string + state *HostState + host *Host + }{ + {"missing memory host", "memory", &HostState{MemUsed: 1}, nil}, + {"zero memory total", "memory", &HostState{MemUsed: 1}, &Host{}}, + {"missing GPU", "gpu_max", &HostState{}, &Host{}}, + {"missing temperature", "temperature_max", &HostState{}, &Host{}}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + server := &Server{Common: Common{ID: 1}, State: tc.state, Host: tc.host} + passed, known := (&Rule{Type: tc.rule, Max: 50, Cover: RuleCoverAll}).Evaluate(nil, server, nil) + if !passed || known { + t.Fatalf("got passed=%v known=%v, want unknown", passed, known) + } + }) + } +} + +func TestOfflineRuleDoesNotRequireMetricState(t *testing.T) { + server := &Server{Common: Common{ID: 1}} + InitServer(server) + passed, known := (&Rule{Type: "offline", Cover: RuleCoverAll}).Evaluate(nil, server, nil) + if passed || !known { + t.Fatalf("offline without state = (%v, %v), want known breach", passed, known) + } +} + +func TestCompoundAlertUsesOneImmutableAgentReport(t *testing.T) { + server := &Server{Common: Common{ID: 1}} + alert := &AlertRule{Rules: []*Rule{ + {Type: "cpu", Max: 80, Cover: RuleCoverAll}, + {Type: "memory", Max: 80, Cover: RuleCoverAll}, + }} + oldReport := RuntimeSnapshot{ + State: &HostState{CPU: 99, MemUsed: 0}, + Host: &Host{MemTotal: 100}, + } + newReport := RuntimeSnapshot{ + State: &HostState{CPU: 0, MemUsed: 99}, + Host: &Host{MemTotal: 100}, + } + for _, report := range []RuntimeSnapshot{oldReport, newReport} { + point, known := alert.SnapshotStatusWithRuntime(nil, server, report, nil) + if !known || len(point) != 2 || point[0] == point[1] { + t.Fatalf("compound point combined incompatible reports: known=%v point=%v", known, point) + } + } +} diff --git a/model/alert_timed_test.go b/model/alert_timed_test.go new file mode 100644 index 00000000..9a3eb16e --- /dev/null +++ b/model/alert_timed_test.go @@ -0,0 +1,104 @@ +package model + +import ( + "testing" + "time" +) + +func timedRule(kind string, duration uint64) *AlertRule { + return &AlertRule{Rules: []*Rule{{Type: kind, Duration: duration, Max: 80, Cover: RuleCoverAll}}} +} + +func timedPoints(start time.Time, values []bool, spacing time.Duration) []TimedAlertPoint { + points := make([]TimedAlertPoint, len(values)) + for i, value := range values { + points[i] = TimedAlertPoint{At: start.Add(time.Duration(i) * spacing), Values: []bool{value}} + } + return points +} + +func TestTimedAlertNeedsElapsedWindowAndDistinctReports(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + rule := timedRule("cpu", 5) + points := timedPoints(start, []bool{false, false, false, false}, AlertSampleInterval) + if known, _ := rule.CheckTimed(points, points[len(points)-1].At); known { + t.Fatal("four reports decided a five-tick window") + } + points = append(points, TimedAlertPoint{At: start.Add(12 * time.Second), Values: []bool{false}}) + if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || passed { + t.Fatalf("complete high-CPU window: known=%v passed=%v", known, passed) + } + // Repeating the same observation does not advance wall-clock coverage. + if known, _ := rule.CheckTimed(points, start.Add(30*time.Second)); known { + t.Fatal("stale sample created a new complete window") + } +} + +func TestTimedAlertGapIsUnknownAndCannotResolve(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + rule := timedRule("cpu", 5) + points := []TimedAlertPoint{ + {At: start, Values: []bool{false}}, + {At: start.Add(30 * time.Second), Values: []bool{true}}, + } + if known, _ := rule.CheckTimed(points, start.Add(30*time.Second)); known { + t.Fatal("telemetry gap must not be interpreted as healthy") + } +} + +func TestTimedAlertIrregularReportsUseElapsedTime(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + rule := timedRule("cpu", 6) + points := timedPoints(start, []bool{false, false, false, false}, 5*time.Second) + if known, passed := rule.CheckTimed(points, start.Add(15*time.Second)); !known || passed { + t.Fatalf("irregular reports cover the wall-clock window: known=%v passed=%v", known, passed) + } +} + +func TestTimedAlertExactSeventyPercentIsNotIncident(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + rule := timedRule("cpu", 10) + values := []bool{false, false, false, false, false, false, false, true, true, true} + points := timedPoints(start, values, AlertSampleInterval) + if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || !passed { + t.Fatalf("70%% failed is below the >70%% trigger: known=%v passed=%v", known, passed) + } +} + +func TestTimedAlertTenTickWindowDoesNotDecideAtNine(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + rule := timedRule("cpu", 10) + points := timedPoints(start, []bool{false, false, false, false, false, false, false, false, false}, AlertSampleInterval) + if known, _ := rule.CheckTimed(points, points[len(points)-1].At); known { + t.Fatal("nine reports decided a ten-tick window one tick early") + } + points = append(points, TimedAlertPoint{At: start.Add(27 * time.Second), Values: []bool{false}}) + if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || passed { + t.Fatalf("full ten-tick breach: known=%v passed=%v", known, passed) + } +} + +func TestTimedOfflineRequiresSustainedAbsence(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + rule := timedRule("offline", 5) + points := timedPoints(start, []bool{false, false, false, false, false}, AlertSampleInterval) + if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || passed { + t.Fatalf("continuous offline interval: known=%v passed=%v", known, passed) + } + points = append(points, TimedAlertPoint{At: start.Add(15 * time.Second), Values: []bool{true}}) + if known, passed := rule.CheckTimed(points, points[len(points)-1].At); !known || !passed { + t.Fatalf("online observation should resolve offline: known=%v passed=%v", known, passed) + } +} + +func TestTimedCompoundRuleWaitsForAllKnown(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + rule := &AlertRule{Rules: []*Rule{ + {Type: "cpu", Duration: 5, Max: 80, Cover: RuleCoverAll}, + {Type: "memory", Duration: 5, Max: 80, Cover: RuleCoverAll}, + }} + points := timedPoints(start, []bool{false, false, false, false, false}, AlertSampleInterval) + if known, _ := rule.CheckTimed(points, points[len(points)-1].At); known { + t.Fatal("missing second condition treated as decisive") + } +} diff --git a/model/alertrule.go b/model/alertrule.go index 886ea0ce..c4f65dae 100644 --- a/model/alertrule.go +++ b/model/alertrule.go @@ -2,12 +2,134 @@ package model import ( "slices" + "time" "github.com/gin-gonic/gin" "github.com/goccy/go-json" "gorm.io/gorm" ) +const ( + AlertSampleInterval = 3 * time.Second + AlertSampleMaxAge = 6 * time.Second +) + +// TimedAlertPoint records one distinct observation. Its timestamp is the +// dashboard receipt time of the Agent report, or the sentinel tick for an +// offline-only rule. +type TimedAlertPoint struct { + At time.Time + Values []bool +} + +// CheckTimed evaluates the legacy Duration as Duration * 3 seconds, avoiding +// both repeated stale samples and cadence-dependent early incidents. Unknown +// coverage cannot create an incident or a recovery. +func (r *AlertRule) CheckTimed(points []TimedAlertPoint, now time.Time) (known, passed bool) { + if r == nil || len(r.Rules) == 0 { + return false, false + } + allFailed := true + for index, rule := range r.Rules { + if rule == nil || !rule.IsSupportedType() { + return false, false + } + ruleKnown, rulePassed := rule.checkTimed(points, index, now) + if !ruleKnown { + return false, false + } + if rulePassed { + allFailed = false + } + } + return true, !allFailed +} + +func (rule *Rule) checkTimed(points []TimedAlertPoint, index int, now time.Time) (known, passed bool) { + if len(points) == 0 { + return false, false + } + if rule.IsTransferDurationRule() { + last := points[len(points)-1] + if index >= len(last.Values) || now.Sub(last.At) > AlertSampleMaxAge { + return false, false + } + return true, last.Values[index] + } + window := time.Duration(rule.Duration) * AlertSampleInterval + if window <= 0 { + return false, false + } + end := now.Add(AlertSampleInterval) // the current observation represents one tick + start := end.Add(-window) + if points[0].At.After(start) { + return false, false // the configured wall-clock window has not elapsed + } + var covered, failed time.Duration + for i, point := range points { + if index >= len(point.Values) { + return false, false + } + segmentStart := point.At + if start.After(segmentStart) { + segmentStart = start + } + segmentEnd := point.At.Add(AlertSampleMaxAge) + if end.Before(segmentEnd) { + segmentEnd = end + } + if i+1 < len(points) { + if points[i+1].At.Before(segmentEnd) { + segmentEnd = points[i+1].At + } + } else { + if tickEnd := point.At.Add(AlertSampleInterval); tickEnd.Before(segmentEnd) { + segmentEnd = tickEnd + } + } + if !segmentEnd.After(segmentStart) { + continue + } + span := segmentEnd.Sub(segmentStart) + covered += span + if !point.Values[index] { + failed += span + } + } + if covered < window-window/10 { + return false, false + } + if rule.IsOfflineRule() { + if covered-failed >= AlertSampleInterval { + return true, true + } + if failed >= window-AlertSampleInterval && covered >= window-AlertSampleInterval { + return true, false + } + return false, false + } + if float64(failed) > float64(window)*0.70 { + return true, false + } + if float64(failed+window-covered) <= float64(window)*0.70 { + return true, true + } + return false, false +} + +func (r *AlertRule) TimedRetention() time.Duration { + longest := AlertSampleMaxAge + for _, rule := range r.Rules { + if rule == nil || rule.IsTransferDurationRule() { + continue + } + if duration, ok := rule.DurationInt(); ok { + longest = max(longest, time.Duration(duration)*AlertSampleInterval+AlertSampleMaxAge) + } + } + return longest +} + const ( ModeAlwaysTrigger = 0 ModeOnetimeTrigger = 1 @@ -75,6 +197,9 @@ func (r *AlertRule) IsSafeToEvaluate() bool { if rule == nil || !rule.IsSupportedType() { return false } + if !rule.HasSafeThresholds() { + return false + } switch rule.Cover { case RuleCoverAll, RuleCoverIgnoreAll: default: @@ -152,18 +277,38 @@ func (r *AlertRule) HasPermission(ctx *gin.Context) bool { // Snapshot 对传入的Server进行该报警规则下所有type的检查 返回每项检查结果 func (r *AlertRule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) []bool { + point, _ := r.SnapshotStatus(cycleTransferStats, server, db) + return point +} + +// SnapshotStatus preserves the distinction between a healthy measurement and +// telemetry that is unavailable. A compound alert cannot be decided while any +// of its applicable conditions is unknown. +func (r *AlertRule) SnapshotStatus(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) ([]bool, bool) { + if server == nil { + return nil, false + } + return r.SnapshotStatusWithRuntime(cycleTransferStats, server, server.RuntimeSnapshot(), db) +} + +// SnapshotStatusWithRuntime uses one immutable report for every condition. +func (r *AlertRule) SnapshotStatusWithRuntime(cycleTransferStats *CycleTransferStats, server *Server, runtime RuntimeSnapshot, db *gorm.DB) ([]bool, bool) { point := make([]bool, len(r.Rules)) + known := true for i, rule := range r.Rules { if rule == nil || !rule.IsSupportedType() { // Invalid persisted rules are ignored instead of being interpreted as // a failed condition or allowed to panic the sentinel. point[i] = true + known = false continue } - point[i] = rule.Snapshot(cycleTransferStats, server, db) + var valid bool + point[i], valid = rule.EvaluateRuntime(cycleTransferStats, server, runtime, db) + known = known && valid } - return point + return point, known } // Check 传入包含当前报警规则下所有type检查结果 返回报警持续时间与是否通过报警检查(通过则返回true) diff --git a/model/notification.go b/model/notification.go index 635eae93..db151433 100644 --- a/model/notification.go +++ b/model/notification.go @@ -180,63 +180,65 @@ func (ns *NotificationServerBundle) replaceParamsInString(str string, message st if ns.Server != nil { runtime := ns.Server.RuntimeSnapshot() - if runtime.State == nil || runtime.Host == nil { - return str - } - state := runtime.State - host := runtime.Host replacements = append(replacements, "#SERVER.NAME#", mod(ns.Server.Name), "#SERVER.ID#", mod(fmt.Sprintf("%d", ns.Server.ID)), - - // Converted metrics - "#SERVER.CPU#", mod(ns.formatUsage(false, state.CPU)), - "#SERVER.MEM#", mod(ns.formatUsage(true, float64(state.MemUsed)/float64(host.MemTotal))), - "#SERVER.SWAP#", mod(ns.formatUsage(true, float64(state.SwapUsed)/float64(host.SwapTotal))), - "#SERVER.DISK#", mod(ns.formatUsage(true, float64(state.DiskUsed)/float64(host.DiskTotal))), - "#SERVER.SPEEDIN#", mod(fmt.Sprintf("%s/s", ns.formatSize(state.NetInSpeed))), - "#SERVER.SPEEDOUT#", mod(fmt.Sprintf("%s/s", ns.formatSize(state.NetOutSpeed))), - "#SERVER.TRANSFERIN#", mod(ns.formatSize(state.NetInTransfer)), - "#SERVER.TRANSFEROUT#", mod(ns.formatSize(state.NetOutTransfer)), - - // Raw metrics - "#SERVER.CPUUSED#", mod(fmt.Sprintf("%f", state.CPU)), - "#SERVER.MEMUSED#", mod(fmt.Sprintf("%d", state.MemUsed)), - "#SERVER.SWAPUSED#", mod(fmt.Sprintf("%d", state.SwapUsed)), - "#SERVER.DISKUSED#", mod(fmt.Sprintf("%d", state.DiskUsed)), - "#SERVER.MEMTOTAL#", mod(fmt.Sprintf("%d", host.MemTotal)), - "#SERVER.SWAPTOTAL#", mod(fmt.Sprintf("%d", host.SwapTotal)), - "#SERVER.DISKTOTAL#", mod(fmt.Sprintf("%d", host.DiskTotal)), - "#SERVER.NETINSPEED#", mod(fmt.Sprintf("%d", state.NetInSpeed)), - "#SERVER.NETOUTSPEED#", mod(fmt.Sprintf("%d", state.NetOutSpeed)), - "#SERVER.NETINTRANSFER#", mod(fmt.Sprintf("%d", state.NetInTransfer)), - "#SERVER.NETOUTTRANSFER#", mod(fmt.Sprintf("%d", state.NetOutTransfer)), - "#SERVER.LOAD1#", mod(fmt.Sprintf("%f", state.Load1)), - "#SERVER.LOAD5#", mod(fmt.Sprintf("%f", state.Load5)), - "#SERVER.LOAD15#", mod(fmt.Sprintf("%f", state.Load15)), - "#SERVER.TCPCONNCOUNT#", mod(fmt.Sprintf("%d", state.TcpConnCount)), - "#SERVER.UDPCONNCOUNT#", mod(fmt.Sprintf("%d", state.UdpConnCount)), ) + if runtime.State != nil && runtime.Host != nil { + state := runtime.State + host := runtime.Host + replacements = append(replacements, + // Converted metrics + "#SERVER.CPU#", mod(ns.formatUsage(false, state.CPU)), + "#SERVER.MEM#", mod(ns.formatUsage(true, float64(state.MemUsed)/float64(host.MemTotal))), + "#SERVER.SWAP#", mod(ns.formatUsage(true, float64(state.SwapUsed)/float64(host.SwapTotal))), + "#SERVER.DISK#", mod(ns.formatUsage(true, float64(state.DiskUsed)/float64(host.DiskTotal))), + "#SERVER.SPEEDIN#", mod(fmt.Sprintf("%s/s", ns.formatSize(state.NetInSpeed))), + "#SERVER.SPEEDOUT#", mod(fmt.Sprintf("%s/s", ns.formatSize(state.NetOutSpeed))), + "#SERVER.TRANSFERIN#", mod(ns.formatSize(state.NetInTransfer)), + "#SERVER.TRANSFEROUT#", mod(ns.formatSize(state.NetOutTransfer)), - var ipv4, ipv6, validIP string - ip := ns.Server.GeoIP.IP - if ip.IPv4Addr != "" && ip.IPv6Addr != "" { - ipv4 = ip.IPv4Addr - ipv6 = ip.IPv6Addr - validIP = ipv4 - } else if ip.IPv4Addr != "" { - ipv4 = ip.IPv4Addr - validIP = ipv4 - } else { - ipv6 = ip.IPv6Addr - validIP = ipv6 + // Raw metrics + "#SERVER.CPUUSED#", mod(fmt.Sprintf("%f", state.CPU)), + "#SERVER.MEMUSED#", mod(fmt.Sprintf("%d", state.MemUsed)), + "#SERVER.SWAPUSED#", mod(fmt.Sprintf("%d", state.SwapUsed)), + "#SERVER.DISKUSED#", mod(fmt.Sprintf("%d", state.DiskUsed)), + "#SERVER.MEMTOTAL#", mod(fmt.Sprintf("%d", host.MemTotal)), + "#SERVER.SWAPTOTAL#", mod(fmt.Sprintf("%d", host.SwapTotal)), + "#SERVER.DISKTOTAL#", mod(fmt.Sprintf("%d", host.DiskTotal)), + "#SERVER.NETINSPEED#", mod(fmt.Sprintf("%d", state.NetInSpeed)), + "#SERVER.NETOUTSPEED#", mod(fmt.Sprintf("%d", state.NetOutSpeed)), + "#SERVER.NETINTRANSFER#", mod(fmt.Sprintf("%d", state.NetInTransfer)), + "#SERVER.NETOUTTRANSFER#", mod(fmt.Sprintf("%d", state.NetOutTransfer)), + "#SERVER.LOAD1#", mod(fmt.Sprintf("%f", state.Load1)), + "#SERVER.LOAD5#", mod(fmt.Sprintf("%f", state.Load5)), + "#SERVER.LOAD15#", mod(fmt.Sprintf("%f", state.Load15)), + "#SERVER.TCPCONNCOUNT#", mod(fmt.Sprintf("%d", state.TcpConnCount)), + "#SERVER.UDPCONNCOUNT#", mod(fmt.Sprintf("%d", state.UdpConnCount)), + ) } - replacements = append(replacements, - "#SERVER.IP#", mod(validIP), - "#SERVER.IPV4#", mod(ipv4), - "#SERVER.IPV6#", mod(ipv6), - ) + var ipv4, ipv6, validIP string + if ns.Server.GeoIP != nil { + ip := ns.Server.GeoIP.IP + if ip.IPv4Addr != "" && ip.IPv6Addr != "" { + ipv4 = ip.IPv4Addr + ipv6 = ip.IPv6Addr + validIP = ipv4 + } else if ip.IPv4Addr != "" { + ipv4 = ip.IPv4Addr + validIP = ipv4 + } else { + ipv6 = ip.IPv6Addr + validIP = ipv6 + } + + replacements = append(replacements, + "#SERVER.IP#", mod(validIP), + "#SERVER.IPV4#", mod(ipv4), + "#SERVER.IPV6#", mod(ipv6), + ) + } } replacer := strings.NewReplacer(replacements...) diff --git a/model/rule.go b/model/rule.go index 2d492b10..1155ce71 100644 --- a/model/rule.go +++ b/model/rule.go @@ -1,6 +1,8 @@ package model import ( + "log" + "math" "slices" "strings" "time" @@ -90,36 +92,80 @@ func (u *Rule) DurationInt() (duration int, ok bool) { // which is the documented legacy spelling for hours. func (u *Rule) HasSafeCycleConfiguration() bool { return u != nil && u.IsTransferDurationRule() && u.CycleStart != nil && - u.CycleInterval > 0 && u.CycleInterval <= MaxAlertRuleCycleInterval + !u.CycleStart.After(time.Now()) && u.CycleInterval > 0 && u.CycleInterval <= MaxAlertRuleCycleInterval && + u.HasSafeThresholds() && u.HasSafeCycleUnit() } -// Snapshot 未通过规则返回 false, 通过返回 true -func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) bool { - if u == nil || server == nil || !u.IsSupportedType() { +func (u *Rule) HasSafeCycleUnit() bool { + if u == nil { + return false + } + switch strings.ToLower(u.CycleUnit) { + case "", "hour", "day", "week", "month", "year": return true + default: + return false + } +} + +func (u *Rule) HasSafeThresholds() bool { + if u == nil || math.IsNaN(u.Min) || math.IsNaN(u.Max) || math.IsInf(u.Min, 0) || math.IsInf(u.Max, 0) || u.Min < 0 || u.Max < 0 { + return false + } + return u.Min == 0 || u.Max == 0 || u.Min <= u.Max +} + +// Snapshot is retained for callers which only need a boolean result. The alert +// sentinel uses Evaluate so missing telemetry is not mistaken for a breach. +func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) bool { + passed, _ := u.Evaluate(cycleTransferStats, server, db) + return passed +} + +// Evaluate returns known=false when a metric cannot be evaluated from the +// current report. Unknown must neither trigger nor resolve an incident. +func (u *Rule) Evaluate(cycleTransferStats *CycleTransferStats, server *Server, db *gorm.DB) (passed, known bool) { + if server == nil { + return true, false + } + return u.EvaluateRuntime(cycleTransferStats, server, server.RuntimeSnapshot(), db) +} + +// EvaluateRuntime evaluates all conditions against the same immutable Agent +// report. Callers evaluating a compound rule must share this snapshot. +func (u *Rule) EvaluateRuntime(cycleTransferStats *CycleTransferStats, server *Server, runtime RuntimeSnapshot, db *gorm.DB) (passed, known bool) { + if u == nil || server == nil || !u.IsSupportedType() { + return true, false } if u.IsTransferDurationRule() && (!u.HasSafeCycleConfiguration() || cycleTransferStats == nil) { - return true + return true, false } // 监控全部但是排除了此服务器 if u.Cover == RuleCoverAll && u.Ignore[server.ID] { - return true + return true, true } // 忽略全部但是指定监控了此服务器 if u.Cover == RuleCoverIgnoreAll && !u.Ignore[server.ID] { - return true + return true, true } + evalNow := time.Now() + var cycleStart, cycleEnd time.Time + if u.IsTransferDurationRule() { + cycleStart, cycleEnd = u.transferDurationBounds(evalNow) + } // 循环区间流量检测 · 短期无需重复检测 - if u.IsTransferDurationRule() && u.NextTransferAt[server.ID].After(time.Now()) { - return u.LastCycleStatus[server.ID] + if u.IsTransferDurationRule() && u.NextTransferAt[server.ID].After(evalNow) { + return u.LastCycleStatus[server.ID], true } var src float64 - runtime := server.RuntimeSnapshot() + if u.IsOfflineRule() { + return !runtime.LastActive.IsZero() && time.Since(runtime.LastActive) <= 6*time.Second, true + } if runtime.State == nil { - return false + return true, false } state := runtime.State @@ -128,22 +174,31 @@ func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, src = float64(state.CPU) case "gpu", "gpu_max": if len(state.GPU) == 0 { - return true + return true, false } src = slices.Max(state.GPU) case "memory": if runtime.Host == nil { - return false + return true, false + } + if runtime.Host.MemTotal == 0 { + return true, false } src = percentage(state.MemUsed, runtime.Host.MemTotal) case "swap": if runtime.Host == nil { - return false + return true, false + } + if runtime.Host.SwapTotal == 0 { + return true, false } src = percentage(state.SwapUsed, runtime.Host.SwapTotal) case "disk": if runtime.Host == nil { - return false + return true, false + } + if runtime.Host.DiskTotal == 0 { + return true, false } src = percentage(state.DiskUsed, runtime.Host.DiskTotal) case "net_in_speed": @@ -151,38 +206,50 @@ func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, case "net_out_speed": src = float64(state.NetOutSpeed) case "net_all_speed": - src = float64(state.NetOutSpeed + state.NetOutSpeed) + src = float64(state.NetInSpeed) + float64(state.NetOutSpeed) case "transfer_in": src = float64(state.NetInTransfer) case "transfer_out": src = float64(state.NetOutTransfer) case "transfer_all": src = float64(state.NetOutTransfer + state.NetInTransfer) - case "offline": - if runtime.LastActive.IsZero() { - src = 0 - } else { - src = float64(runtime.LastActive.Unix()) - } case "transfer_in_cycle": + if db == nil { + return true, false + } src = float64(utils.SubUintChecked(state.NetInTransfer, runtime.PrevTransferInSnapshot)) if u.CycleInterval != 0 { var res NResult - db.Model(&Transfer{}).Select("SUM(`in`) AS n").Where("datetime(`created_at`) >= datetime(?) AND server_id = ?", u.GetTransferDurationStart().UTC(), server.ID).Scan(&res) + if err := db.Model(&Transfer{}).Select("SUM(`in`) AS n").Where("datetime(`created_at`) >= datetime(?) AND datetime(`created_at`) < datetime(?) AND datetime(`created_at`) <= datetime(?) AND server_id = ?", cycleStart.UTC(), cycleEnd.UTC(), evalNow.UTC(), server.ID).Scan(&res).Error; err != nil { + log.Printf("NEZHA>> Alert cycle transfer query failed for rule %s server %d: %v", u.Type, server.ID, err) + return true, false + } src += float64(res.N) } case "transfer_out_cycle": + if db == nil { + return true, false + } src = float64(utils.SubUintChecked(state.NetOutTransfer, runtime.PrevTransferOutSnapshot)) if u.CycleInterval != 0 { var res NResult - db.Model(&Transfer{}).Select("SUM(`out`) AS n").Where("datetime(`created_at`) >= datetime(?) AND server_id = ?", u.GetTransferDurationStart().UTC(), server.ID).Scan(&res) + if err := db.Model(&Transfer{}).Select("SUM(`out`) AS n").Where("datetime(`created_at`) >= datetime(?) AND datetime(`created_at`) < datetime(?) AND datetime(`created_at`) <= datetime(?) AND server_id = ?", cycleStart.UTC(), cycleEnd.UTC(), evalNow.UTC(), server.ID).Scan(&res).Error; err != nil { + log.Printf("NEZHA>> Alert cycle transfer query failed for rule %s server %d: %v", u.Type, server.ID, err) + return true, false + } src += float64(res.N) } case "transfer_all_cycle": + if db == nil { + return true, false + } src = float64(utils.SubUintChecked(state.NetOutTransfer, runtime.PrevTransferOutSnapshot) + utils.SubUintChecked(state.NetInTransfer, runtime.PrevTransferInSnapshot)) if u.CycleInterval != 0 { var res NResult - db.Model(&Transfer{}).Select("SUM(`in`+`out`) AS n").Where("datetime(`created_at`) >= datetime(?) AND server_id = ?", u.GetTransferDurationStart().UTC(), server.ID).Scan(&res) + if err := db.Model(&Transfer{}).Select("SUM(`in`+`out`) AS n").Where("datetime(`created_at`) >= datetime(?) AND datetime(`created_at`) < datetime(?) AND datetime(`created_at`) <= datetime(?) AND server_id = ?", cycleStart.UTC(), cycleEnd.UTC(), evalNow.UTC(), server.ID).Scan(&res).Error; err != nil { + log.Printf("NEZHA>> Alert cycle transfer query failed for rule %s server %d: %v", u.Type, server.ID, err) + return true, false + } src += float64(res.N) } case "load1": @@ -205,23 +272,32 @@ func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, } } if len(temp) == 0 { - return true + return true, false } src = slices.Max(temp) default: - return true + return true, false } // 循环区间流量检测 · 更新下次需要检测时间 if u.IsTransferDurationRule() { - seconds := max(1800*((u.Max-src)/u.Max), 180) + seconds := float64(180) + if u.Max > 0 { + seconds = max(1800*((u.Max-src)/u.Max), 180) + } else if u.Min > 0 { + seconds = 1800 + } if u.NextTransferAt == nil { u.NextTransferAt = make(map[uint64]time.Time) } if u.LastCycleStatus == nil { u.LastCycleStatus = make(map[uint64]bool) } - u.NextTransferAt[server.ID] = time.Now().Add(time.Second * time.Duration(seconds)) + nextAt := evalNow.Add(time.Second * time.Duration(seconds)) + if nextAt.After(cycleEnd) { + nextAt = cycleEnd + } + u.NextTransferAt[server.ID] = nextAt if (u.Max > 0 && src > u.Max) || (u.Min > 0 && src < u.Min) { u.LastCycleStatus[server.ID] = false } else { @@ -233,17 +309,15 @@ func (u *Rule) Snapshot(cycleTransferStats *CycleTransferStats, server *Server, cycleTransferStats.Transfer[server.ID] = uint64(src) cycleTransferStats.NextUpdate[server.ID] = u.NextTransferAt[server.ID] // 自动更新周期流量展示起止时间 - cycleTransferStats.From = u.GetTransferDurationStart() - cycleTransferStats.To = u.GetTransferDurationEnd() + cycleTransferStats.From = cycleStart + cycleTransferStats.To = cycleEnd } - if u.Type == "offline" && float64(time.Now().Unix())-src > 6 { - return false - } else if (u.Max > 0 && src > u.Max) || (u.Min > 0 && src < u.Min) { - return false + if (u.Max > 0 && src > u.Max) || (u.Min > 0 && src < u.Min) { + return false, true } - return true + return true, true } // IsTransferDurationRule 判断该规则是否属于周期流量规则 属于则返回true @@ -263,83 +337,49 @@ func (u *Rule) IsOfflineRule() bool { return u != nil && u.Type == "offline" } -// GetTransferDurationStart 获取周期流量的起始时间 +// transferDurationBounds computes both edges from the same clock reading. +// Separate time.Now calls can otherwise straddle a cycle boundary. +func (u *Rule) transferDurationBounds(now time.Time) (time.Time, time.Time) { + if !u.HasSafeCycleConfiguration() { + return time.Time{}, time.Time{} + } + start := *u.CycleStart + var end time.Time + switch strings.ToLower(u.CycleUnit) { + case "year": + end = start.AddDate(int(u.CycleInterval), 0, 0) + for !now.Before(end) { + start, end = end, end.AddDate(int(u.CycleInterval), 0, 0) + } + case "month": + end = start.AddDate(0, int(u.CycleInterval), 0) + for !now.Before(end) { + start, end = end, end.AddDate(0, int(u.CycleInterval), 0) + } + case "week": + end = start.AddDate(0, 0, 7*int(u.CycleInterval)) + for !now.Before(end) { + start, end = end, end.AddDate(0, 0, 7*int(u.CycleInterval)) + } + case "day": + end = start.AddDate(0, 0, int(u.CycleInterval)) + for !now.Before(end) { + start, end = end, end.AddDate(0, 0, int(u.CycleInterval)) + } + default: // empty is the legacy spelling for hour + interval := 3600 * int64(u.CycleInterval) + start = time.Unix(u.CycleStart.Unix()+(now.Unix()-u.CycleStart.Unix())/interval*interval, 0) + end = time.Unix(start.Unix()+interval, 0) + } + return start, end +} + func (u *Rule) GetTransferDurationStart() time.Time { - // Accept uppercase and lowercase - unit := strings.ToLower(u.CycleUnit) - startTime := *u.CycleStart - var nextTime time.Time - switch unit { - case "year": - nextTime = startTime.AddDate(int(u.CycleInterval), 0, 0) - for time.Now().After(nextTime) { - startTime = nextTime - nextTime = nextTime.AddDate(int(u.CycleInterval), 0, 0) - } - case "month": - nextTime = startTime.AddDate(0, int(u.CycleInterval), 0) - for time.Now().After(nextTime) { - startTime = nextTime - nextTime = nextTime.AddDate(0, int(u.CycleInterval), 0) - } - case "week": - nextTime = startTime.AddDate(0, 0, 7*int(u.CycleInterval)) - for time.Now().After(nextTime) { - startTime = nextTime - nextTime = nextTime.AddDate(0, 0, 7*int(u.CycleInterval)) - } - case "day": - nextTime = startTime.AddDate(0, 0, int(u.CycleInterval)) - for time.Now().After(nextTime) { - startTime = nextTime - nextTime = nextTime.AddDate(0, 0, int(u.CycleInterval)) - } - default: - // For hour unit or not set. - interval := 3600 * int64(u.CycleInterval) - startTime = time.Unix(u.CycleStart.Unix()+(time.Now().Unix()-u.CycleStart.Unix())/interval*interval, 0) - } - - return startTime + start, _ := u.transferDurationBounds(time.Now()) + return start } -// GetTransferDurationEnd 获取周期流量结束时间 func (u *Rule) GetTransferDurationEnd() time.Time { - // Accept uppercase and lowercase - unit := strings.ToLower(u.CycleUnit) - startTime := *u.CycleStart - var nextTime time.Time - switch unit { - case "year": - nextTime = startTime.AddDate(int(u.CycleInterval), 0, 0) - for time.Now().After(nextTime) { - startTime = nextTime - nextTime = nextTime.AddDate(int(u.CycleInterval), 0, 0) - } - case "month": - nextTime = startTime.AddDate(0, int(u.CycleInterval), 0) - for time.Now().After(nextTime) { - startTime = nextTime - nextTime = nextTime.AddDate(0, int(u.CycleInterval), 0) - } - case "week": - nextTime = startTime.AddDate(0, 0, 7*int(u.CycleInterval)) - for time.Now().After(nextTime) { - startTime = nextTime - nextTime = nextTime.AddDate(0, 0, 7*int(u.CycleInterval)) - } - case "day": - nextTime = startTime.AddDate(0, 0, int(u.CycleInterval)) - for time.Now().After(nextTime) { - startTime = nextTime - nextTime = nextTime.AddDate(0, 0, int(u.CycleInterval)) - } - default: - // For hour unit or not set. - interval := 3600 * int64(u.CycleInterval) - startTime = time.Unix(u.CycleStart.Unix()+(time.Now().Unix()-u.CycleStart.Unix())/interval*interval, 0) - nextTime = time.Unix(startTime.Unix()+interval, 0) - } - - return nextTime + _, end := u.transferDurationBounds(time.Now()) + return end } diff --git a/model/server.go b/model/server.go index fd566199..ada61db9 100644 --- a/model/server.go +++ b/model/server.go @@ -16,6 +16,7 @@ import ( ) var runtimeHolderInitMu sync.Mutex +var nextServerReportSequence atomic.Uint64 type Server struct { Common @@ -73,15 +74,16 @@ type taskStreamHolder struct { } type serverRuntimeHolder struct { - mu sync.Mutex - canonical *Server - stream pb.NezhaService_ReportSystemStateServer - generation uint64 - state *HostState - host *Host - lastActive time.Time - prevIn uint64 - prevOut uint64 + mu sync.Mutex + canonical *Server + stream pb.NezhaService_ReportSystemStateServer + generation uint64 + state *HostState + host *Host + lastActive time.Time + reportSequence uint64 + prevIn uint64 + prevOut uint64 } type StateStreamLease struct { @@ -204,6 +206,7 @@ func (lease StateStreamLease) updateState(receiver *Server, state *HostState, la return false } } + lease.holder.reportSequence = nextServerReportSequence.Add(1) return true } @@ -334,6 +337,7 @@ type RuntimeSnapshot struct { State *HostState Host *Host LastActive time.Time + ReportSequence uint64 PrevTransferInSnapshot uint64 PrevTransferOutSnapshot uint64 } @@ -360,7 +364,7 @@ func (s *Server) RuntimeSnapshot() RuntimeSnapshot { holder.host = cloneHost(s.Host) } } - return RuntimeSnapshot{State: cloneHostState(holder.state), Host: cloneHost(holder.host), LastActive: holder.lastActive, PrevTransferInSnapshot: holder.prevIn, PrevTransferOutSnapshot: holder.prevOut} + return RuntimeSnapshot{State: cloneHostState(holder.state), Host: cloneHost(holder.host), LastActive: holder.lastActive, ReportSequence: holder.reportSequence, PrevTransferInSnapshot: holder.prevIn, PrevTransferOutSnapshot: holder.prevOut} } func (s *Server) SetTransferSnapshots(inbound, outbound uint64) bool { diff --git a/service/singleton/alertsentinel.go b/service/singleton/alertsentinel.go index 82f56846..838f936c 100644 --- a/service/singleton/alertsentinel.go +++ b/service/singleton/alertsentinel.go @@ -4,6 +4,7 @@ import ( "fmt" "log" "sync" + "sync/atomic" "time" "github.com/jinzhu/copier" @@ -26,11 +27,122 @@ type NotificationHistory struct { var ( AlertsLock sync.RWMutex Alerts []*model.AlertRule - alertsStore map[uint64]map[uint64][][]bool // [alert_id][server_id] -> [timeTick][ruleId] 时间点对应的rule的检查结果 - alertsPrevState map[uint64]map[uint64]uint8 // [alert_id][server_id] -> 对应报警规则的上一次报警状态 - AlertsCycleTransferStatsStore map[uint64]*model.CycleTransferStats // [alert_id] -> 对应报警规则的周期流量统计 + alertsStore map[uint64]map[uint64][]model.TimedAlertPoint // distinct reports per alert/server + alertsPrevState map[uint64]map[uint64]uint8 // [alert_id][server_id] -> 对应报警规则的上一次报警状态 + alertsLastMetricSeq map[uint64]map[uint64]uint64 // last distinct Agent metric report per alert/server + AlertsCycleTransferStatsStore map[uint64]*model.CycleTransferStats // [alert_id] -> 对应报警规则的周期流量统计 + alertDeliveryMu sync.Mutex + alertDeliveries map[uint64]map[uint64]*alertDeliveryEntry + alertEventSequence atomic.Uint64 ) +type alertDeliveryEntry struct { + cancel chan struct{} + done chan struct{} + sendMu *sync.Mutex +} + +type alertDeliveryPhase uint8 + +const ( + alertDeliveryIncident alertDeliveryPhase = iota + alertDeliveryRecovery +) + +func cancelAlertDeliveries(alertID uint64) { + alertDeliveryMu.Lock() + defer alertDeliveryMu.Unlock() + for _, entry := range alertDeliveries[alertID] { + close(entry.cancel) + } + delete(alertDeliveries, alertID) +} + +func cancelAlertDeliveriesForServers(serverIDs []uint64) { + alertDeliveryMu.Lock() + defer alertDeliveryMu.Unlock() + for alertID, byServer := range alertDeliveries { + for _, serverID := range serverIDs { + if entry := byServer[serverID]; entry != nil { + close(entry.cancel) + delete(byServer, serverID) + } + } + if len(byServer) == 0 { + delete(alertDeliveries, alertID) + } + } +} + +func runAlertDelivery(cancel <-chan struct{}, send func() bool, always bool, retryDelay time.Duration) { + for { + select { + case <-cancel: + return + default: + } + delivered := send() + if delivered && !always { + return + } + delay := retryDelay + if delivered { + delay = firstNotificationDelay + } + timer := time.NewTimer(delay) + select { + case <-cancel: + timer.Stop() + return + case <-timer.C: + } + } +} + +func startAlertDelivery(alert *model.AlertRule, server *model.Server, message, muteLabel string, phase alertDeliveryPhase) { + // A mute entry belongs to one transition, not all later incidents of the + // same phase. Concurrent in-flight sends may finish after a recovery. + muteLabel = fmt.Sprintf("%s:event-%d", muteLabel, alertEventSequence.Add(1)) + alertDeliveryMu.Lock() + if alertDeliveries == nil { + alertDeliveries = make(map[uint64]map[uint64]*alertDeliveryEntry) + } + if alertDeliveries[alert.ID] == nil { + alertDeliveries[alert.ID] = make(map[uint64]*alertDeliveryEntry) + } + sendMu := new(sync.Mutex) + if old := alertDeliveries[alert.ID][server.ID]; old != nil { + close(old.cancel) + sendMu = old.sendMu + } + entry := &alertDeliveryEntry{cancel: make(chan struct{}), done: make(chan struct{}), sendMu: sendMu} + alertDeliveries[alert.ID][server.ID] = entry + alertDeliveryMu.Unlock() + + groupID := alert.NotificationGroupID + always := phase == alertDeliveryIncident && alert.TriggerMode == model.ModeAlwaysTrigger + go func() { + defer close(entry.done) + runAlertDelivery(entry.cancel, func() bool { + // Serialize incident/recovery delivery for this alert/server pair. + // A recovery can cancel a pending incident, but never overtake an + // already-running network request. + entry.sendMu.Lock() + defer entry.sendMu.Unlock() + select { + case <-entry.cancel: + return true + default: + } + return NotificationShared.SendNotification(groupID, message, muteLabel, server) + }, always, 30*time.Second) + }() +} + +func shouldRunAlertFailTasks(triggerMode uint8, newIncident bool) bool { + return newIncident || triggerMode == model.ModeAlwaysTrigger +} + // addCycleTransferStatsInfo 向AlertsCycleTransferStatsStore中添加周期流量报警统计信息 func addCycleTransferStatsInfo(alert *model.AlertRule) { if alert == nil || !alert.Enabled() || !alert.IsSafeToEvaluate() { @@ -59,8 +171,17 @@ func addCycleTransferStatsInfo(alert *model.AlertRule) { // AlertSentinelStart 报警器启动 func AlertSentinelStart() { - alertsStore = make(map[uint64]map[uint64][][]bool) + alertDeliveryMu.Lock() + for _, byServer := range alertDeliveries { + for _, entry := range byServer { + close(entry.cancel) + } + } + alertDeliveries = make(map[uint64]map[uint64]*alertDeliveryEntry) + alertDeliveryMu.Unlock() + alertsStore = make(map[uint64]map[uint64][]model.TimedAlertPoint) alertsPrevState = make(map[uint64]map[uint64]uint8) + alertsLastMetricSeq = make(map[uint64]map[uint64]uint64) AlertsCycleTransferStatsStore = make(map[uint64]*model.CycleTransferStats) AlertsLock.Lock() if err := DB.Find(&Alerts).Error; err != nil { @@ -71,8 +192,9 @@ func AlertSentinelStart() { log.Printf("NEZHA>> Skipping invalid nil alert rule loaded from database") continue } - alertsStore[alert.ID] = make(map[uint64][][]bool) + alertsStore[alert.ID] = make(map[uint64][]model.TimedAlertPoint) alertsPrevState[alert.ID] = make(map[uint64]uint8) + alertsLastMetricSeq[alert.ID] = make(map[uint64]uint64) if !alert.IsSafeToEvaluate() { log.Printf("NEZHA>> Skipping invalid alert rule %d loaded from database", alert.ID) continue @@ -101,8 +223,10 @@ func AlertSentinelStart() { func OnRefreshOrAddAlert(alert *model.AlertRule) { AlertsLock.Lock() defer AlertsLock.Unlock() + cancelAlertDeliveries(alert.ID) delete(alertsStore, alert.ID) delete(alertsPrevState, alert.ID) + delete(alertsLastMetricSeq, alert.ID) var isEdit bool for i := range Alerts { if Alerts[i].ID == alert.ID { @@ -113,8 +237,9 @@ func OnRefreshOrAddAlert(alert *model.AlertRule) { if !isEdit { Alerts = append(Alerts, alert) } - alertsStore[alert.ID] = make(map[uint64][][]bool) + alertsStore[alert.ID] = make(map[uint64][]model.TimedAlertPoint) alertsPrevState[alert.ID] = make(map[uint64]uint8) + alertsLastMetricSeq[alert.ID] = make(map[uint64]uint64) delete(AlertsCycleTransferStatsStore, alert.ID) addCycleTransferStatsInfo(alert) } @@ -123,8 +248,10 @@ func OnDeleteAlert(id []uint64) { AlertsLock.Lock() defer AlertsLock.Unlock() for _, i := range id { + cancelAlertDeliveries(i) delete(alertsStore, i) delete(alertsPrevState, i) + delete(alertsLastMetricSeq, i) currentAlerts := Alerts[:0] for _, alert := range Alerts { if alert.ID != i { @@ -176,25 +303,67 @@ func checkStatusForServer(alert *model.AlertRule, server *model.Server) { } }() - alertsStore[alert.ID][server.ID] = append(alertsStore[alert. - ID][server.ID], alert.Snapshot(AlertsCycleTransferStatsStore[alert.ID], server, DB)) + // Offline-only rules deliberately observe elapsed time on every tick. Metric + // rules, including compound rules, must never count the same Agent report + // repeatedly or treat a disconnected Agent's retained State as fresh data. + metricRule := false + runtime := server.RuntimeSnapshot() + for _, rule := range alert.Rules { + if !rule.IsOfflineRule() { + metricRule = true + break + } + } + if metricRule { + if runtime.LastActive.IsZero() || runtime.State == nil || time.Since(runtime.LastActive) > model.AlertSampleMaxAge || time.Since(runtime.LastActive) < 0 { + alertsStore[alert.ID][server.ID] = nil + return + } + if runtime.ReportSequence == 0 || runtime.ReportSequence == alertsLastMetricSeq[alert.ID][server.ID] { + return + } + alertsLastMetricSeq[alert.ID][server.ID] = runtime.ReportSequence + } + point, known := alert.SnapshotStatusWithRuntime(AlertsCycleTransferStatsStore[alert.ID], server, runtime, DB) + if !known { + alertsStore[alert.ID][server.ID] = nil + return + } + sampledAt := time.Now() + if metricRule { + sampledAt = runtime.LastActive + } + alertsStore[alert.ID][server.ID] = append(alertsStore[alert.ID][server.ID], model.TimedAlertPoint{At: sampledAt, Values: point}) + // Bound memory even while coverage is insufficient to decide a state. + cutoff := time.Now().Add(-alert.TimedRetention()) + samples := alertsStore[alert.ID][server.ID] + first := 0 + for first < len(samples) && samples[first].At.Before(cutoff) { + first++ + } + if first > 0 { + alertsStore[alert.ID][server.ID] = append([]model.TimedAlertPoint(nil), samples[first:]...) + } // 发送通知，分为触发报警和恢复通知 - _, passed := alert.Check(alertsStore[alert.ID][server.ID]) + known, passed := alert.CheckTimed(alertsStore[alert.ID][server.ID], time.Now()) + if !known { + return + } // 保存当前服务器状态信息 curServer := model.Server{} copier.Copy(&curServer, server) // 本次未通过检查 if !passed { - // 始终触发模式或上次检查不为失败时触发报警（跳过单次触发+上次失败的情况） - if alert.TriggerMode == model.ModeAlwaysTrigger || alertsPrevState[alert.ID][server.ID] != _RuleCheckFail { + newIncident := alertsPrevState[alert.ID][server.ID] != _RuleCheckFail + if newIncident { 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) + startAlertDelivery(alert, &curServer, message, NotificationMuteLabel.ServerIncident(server.ID, alert.ID), alertDeliveryIncident) + } + if shouldRunAlertFailTasks(alert.TriggerMode, newIncident) { 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 { // 本次通过检查但上一次的状态为失败，则发送恢复通知 @@ -202,20 +371,8 @@ func checkStatusForServer(alert *model.AlertRule, server *model.Server) { 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)) + startAlertDelivery(alert, &curServer, message, NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID), alertDeliveryRecovery) } 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:] - } } diff --git a/service/singleton/alertsentinel_freshness_test.go b/service/singleton/alertsentinel_freshness_test.go new file mode 100644 index 00000000..b68aeae5 --- /dev/null +++ b/service/singleton/alertsentinel_freshness_test.go @@ -0,0 +1,281 @@ +package singleton + +import ( + "context" + "testing" + "time" + + "github.com/patrickmn/go-cache" + "google.golang.org/grpc/metadata" + + "github.com/nezhahq/nezha/model" + pb "github.com/nezhahq/nezha/proto" +) + +type alertTestStateStream struct{} + +func (alertTestStateStream) Send(*pb.Receipt) error { return nil } +func (alertTestStateStream) Recv() (*pb.State, error) { return nil, context.Canceled } +func (alertTestStateStream) SetHeader(metadata.MD) error { return nil } +func (alertTestStateStream) SendHeader(metadata.MD) error { return nil } +func (alertTestStateStream) SetTrailer(metadata.MD) {} +func (alertTestStateStream) Context() context.Context { return context.Background() } +func (alertTestStateStream) SendMsg(any) error { return nil } +func (alertTestStateStream) RecvMsg(any) error { return nil } + +func TestAlertSentinelCountsOnlyFreshDistinctMetricReports(t *testing.T) { + oldStore, oldPrev, oldLast, oldCycle := alertsStore, alertsPrevState, alertsLastMetricSeq, AlertsCycleTransferStatsStore + alertsStore = map[uint64]map[uint64][]model.TimedAlertPoint{7: {1: nil}} + alertsPrevState = map[uint64]map[uint64]uint8{7: {}} + alertsLastMetricSeq = map[uint64]map[uint64]uint64{7: {}} + AlertsCycleTransferStatsStore = map[uint64]*model.CycleTransferStats{} + t.Cleanup(func() { + alertsStore, alertsPrevState, alertsLastMetricSeq, AlertsCycleTransferStatsStore = oldStore, oldPrev, oldLast, oldCycle + }) + + alert := &model.AlertRule{ + Common: model.Common{ID: 7}, + Rules: []*model.Rule{{Type: "cpu", Max: 80, Duration: 100, Cover: model.RuleCoverAll}}, + } + server := &model.Server{Common: model.Common{ID: 1}} + model.InitServer(server) + lease := server.AttachStateStream(alertTestStateStream{}) + fixedReportTime := time.Now() + if !lease.UpdateState(&model.HostState{CPU: 99}, fixedReportTime) { + t.Fatal("state update rejected") + } + checkStatusForServer(alert, server) + if got := len(alertsStore[7][1]); got != 1 { + t.Fatalf("first report produced %d samples, want 1", got) + } + for i := 0; i < 100; i++ { + checkStatusForServer(alert, server) + } + if got := len(alertsStore[7][1]); got != 1 { + t.Fatalf("same report was counted %d times, want 1", got) + } + if !lease.Clear() { + t.Fatal("stream clear rejected") + } + checkStatusForServer(alert, server) + if got := len(alertsStore[7][1]); got != 0 { + t.Fatalf("disconnect retained %d high samples, want 0", got) + } + alertsPrevState[7][1] = _RuleCheckFail // incident was active before disconnect + newLease := server.AttachStateStream(alertTestStateStream{}) + // Distinct reports may share a clock timestamp on coarse-resolution hosts. + if !newLease.UpdateState(&model.HostState{CPU: 99}, fixedReportTime) { + t.Fatal("reconnect state update rejected") + } + checkStatusForServer(alert, server) + if got := len(alertsStore[7][1]); got != 1 || alertsStore[7][1][0].Values[0] { + t.Fatalf("reconnect window = %v, want one high sample", alertsStore[7][1]) + } + if alertsPrevState[7][1] != _RuleCheckFail { + t.Fatal("incomplete warm-up falsely resolved active incident") + } + replacement := &model.Server{Common: model.Common{ID: 1}} + model.InitServer(replacement) + replacementLease := replacement.AttachStateStream(alertTestStateStream{}) + if !replacementLease.UpdateState(&model.HostState{CPU: 99}, fixedReportTime) { + t.Fatal("replacement server state update rejected") + } + checkStatusForServer(alert, replacement) + if got := len(alertsStore[7][1]); got != 2 { + t.Fatalf("replacement holder reused a report identity; samples=%d", got) + } +} + +func TestAlertNewIncidentNotMutedByEarlierInflightIncident(t *testing.T) { + oldCache, oldNotifications := Cache, NotificationShared + Cache = cache.New(time.Minute, time.Minute) + n := &model.Notification{Common: model.Common{ID: 601}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 603}, Name: "alert-test"}, []uint64{n.ID}) + NotificationShared = nc + t.Cleanup(func() { + cancelAlertDeliveries(602) + NotificationShared, Cache = oldNotifications, oldCache + }) + started := make(chan struct{}) + release := make(chan struct{}) + delivered := make(chan string, 3) + nc.sendForTest = func(_ *model.Notification, desc string, _ *model.Server) error { + if desc == "first incident" { + close(started) + <-release + } + delivered <- desc + return nil + } + alert := &model.AlertRule{ + Common: model.Common{ID: 602}, + NotificationGroupID: 603, + TriggerMode: model.ModeOnetimeTrigger, + } + server := &model.Server{Common: model.Common{ID: 604}} + label := NotificationMuteLabel.ServerIncident(server.ID, alert.ID) + startAlertDelivery(alert, server, "first incident", label, alertDeliveryIncident) + <-started + startAlertDelivery(alert, server, "recovered", NotificationMuteLabel.ServerIncidentResolved(server.ID, alert.ID), alertDeliveryRecovery) + startAlertDelivery(alert, server, "second incident", label, alertDeliveryIncident) + close(release) + timer := time.NewTimer(time.Second) + defer timer.Stop() + seen := map[string]bool{} + for len(seen) < 2 { + select { + case message := <-delivered: + seen[message] = true + case <-timer.C: + t.Fatalf("new incident was suppressed after old send completed: %v", seen) + } + } + if !seen["first incident"] || !seen["second incident"] || seen["recovered"] { + t.Fatalf("unexpected transition deliveries: %v", seen) + } + alertDeliveryMu.Lock() + done := alertDeliveries[alert.ID][server.ID].done + alertDeliveryMu.Unlock() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("new incident delivery did not finish") + } +} + +func TestAlertRecoveryDeliveryStopsAfterSuccessInAlwaysMode(t *testing.T) { + oldCache, oldNotifications := Cache, NotificationShared + Cache = cache.New(time.Minute, time.Minute) + n := &model.Notification{Common: model.Common{ID: 611}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 613}, Name: "recovery-test"}, []uint64{n.ID}) + NotificationShared = nc + t.Cleanup(func() { + cancelAlertDeliveries(612) + NotificationShared, Cache = oldNotifications, oldCache + }) + + delivered := make(chan struct{}, 2) + nc.sendForTest = func(*model.Notification, string, *model.Server) error { + delivered <- struct{}{} + return nil + } + alert := &model.AlertRule{ + Common: model.Common{ID: 612}, + NotificationGroupID: 613, + TriggerMode: model.ModeAlwaysTrigger, + } + server := &model.Server{Common: model.Common{ID: 614}} + startAlertDelivery(alert, server, "recovered", "recovery-event", alertDeliveryRecovery) + + select { + case <-delivered: + case <-time.After(time.Second): + t.Fatal("recovery notification was not delivered") + } + alertDeliveryMu.Lock() + done := alertDeliveries[alert.ID][server.ID].done + alertDeliveryMu.Unlock() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("successful recovery delivery kept repeating in always mode") + } + select { + case <-delivered: + t.Fatal("recovery notification was delivered more than once") + default: + } +} + +func TestServerDeleteCancelsAlertDelivery(t *testing.T) { + alertDeliveryMu.Lock() + oldDeliveries := alertDeliveries + cancel := make(chan struct{}) + alertDeliveries = map[uint64]map[uint64]*alertDeliveryEntry{ + 621: {622: {cancel: cancel, done: make(chan struct{})}}, + } + alertDeliveryMu.Unlock() + t.Cleanup(func() { + alertDeliveryMu.Lock() + alertDeliveries = oldDeliveries + alertDeliveryMu.Unlock() + }) + + servers := &ServerClass{ + class: class[uint64, *model.Server]{ + list: map[uint64]*model.Server{ + 622: {Common: model.Common{ID: 622}, UUID: "delete-alert-worker"}, + }, + }, + uuidToID: map[string]uint64{"delete-alert-worker": 622}, + } + servers.Delete([]uint64{622}) + + select { + case <-cancel: + default: + t.Fatal("server deletion did not cancel its alert delivery") + } + alertDeliveryMu.Lock() + _, retained := alertDeliveries[621] + alertDeliveryMu.Unlock() + if retained { + t.Fatal("server deletion retained an empty alert delivery map") + } +} + +func TestAlwaysAlertContinuesFailureTriggerTasks(t *testing.T) { + if !shouldRunAlertFailTasks(model.ModeAlwaysTrigger, false) { + t.Fatal("always mode stopped failure trigger tasks after the initial incident") + } + if shouldRunAlertFailTasks(model.ModeOnetimeTrigger, false) { + t.Fatal("one-time mode repeated failure trigger tasks") + } + if !shouldRunAlertFailTasks(model.ModeOnetimeTrigger, true) { + t.Fatal("one-time mode skipped failure trigger tasks for a new incident") + } +} + +func TestAlertDeliveryRetriesFailureAndStopsAfterSuccess(t *testing.T) { + cancel := make(chan struct{}) + done := make(chan struct{}) + attempts := 0 + go func() { + runAlertDelivery(cancel, func() bool { + attempts++ + return attempts == 2 + }, false, time.Millisecond) + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + close(cancel) + t.Fatal("delivery did not retry and finish") + } + if attempts != 2 { + t.Fatalf("got %d attempts, want 2", attempts) + } +} + +func TestAlertDeliveryCancelsAfterRecovery(t *testing.T) { + cancel := make(chan struct{}) + done := make(chan struct{}) + started := make(chan struct{}) + go func() { + runAlertDelivery(cancel, func() bool { + close(started) + return true + }, true, time.Millisecond) + close(done) + }() + <-started + close(cancel) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("recovered incident delivery was not cancelled") + } +} diff --git a/service/singleton/alertsentinel_test.go b/service/singleton/alertsentinel_test.go index e227e57b..a78bf334 100644 --- a/service/singleton/alertsentinel_test.go +++ b/service/singleton/alertsentinel_test.go @@ -122,7 +122,7 @@ func TestCheckStatusForServerRecoversUnexpectedEvaluatorPanic(t *testing.T) { originalStore := alertsStore originalPrevState := alertsPrevState originalCycleStore := AlertsCycleTransferStatsStore - alertsStore = map[uint64]map[uint64][][]bool{} + alertsStore = map[uint64]map[uint64][]model.TimedAlertPoint{} alertsPrevState = map[uint64]map[uint64]uint8{} AlertsCycleTransferStatsStore = map[uint64]*model.CycleTransferStats{} t.Cleanup(func() { diff --git a/service/singleton/notification.go b/service/singleton/notification.go index 31bf3fe2..a5a53bfa 100644 --- a/service/singleton/notification.go +++ b/service/singleton/notification.go @@ -22,8 +22,228 @@ type NotificationClass struct { groupToIDList map[uint64]map[uint64]*model.Notification idToGroupList map[uint64]map[uint64]struct{} - groupList map[uint64]string - groupMu sync.RWMutex + groupList map[uint64]string + groupMu sync.RWMutex + deliveryMu sync.Mutex + deliveryKeys map[string]*notificationKeyLock + asyncInFlight map[string]struct{} + asyncRetryAfter map[string]time.Time + asyncRetryDelay map[string]time.Duration + muteGeneration map[string]uint64 + serviceDelivery map[uint64]*orderedServiceDelivery + sendForTest func(*model.Notification, string, *model.Server) error +} + +type serviceDeliveryRequest struct { + groupID uint64 + desc string + label string +} + +type orderedServiceDelivery struct { + current serviceDeliveryRequest + pending *serviceDeliveryRequest + update chan struct{} + cancel chan struct{} + done chan struct{} +} + +// SendServiceState preserves incident→recovery order without allowing a +// slow webhook to accumulate one goroutine per probe result. Only the latest +// pending phase is retained; a failed current phase retries with backoff. +func (c *NotificationClass) SendServiceState(serviceID, groupID uint64, desc, label string) { + req := serviceDeliveryRequest{groupID: groupID, desc: desc, label: label} + c.deliveryMu.Lock() + if c.serviceDelivery == nil { + c.serviceDelivery = make(map[uint64]*orderedServiceDelivery) + } + state := c.serviceDelivery[serviceID] + if state == nil { + state = &orderedServiceDelivery{current: req, update: make(chan struct{}, 1), cancel: make(chan struct{}), done: make(chan struct{})} + c.serviceDelivery[serviceID] = state + c.deliveryMu.Unlock() + go c.runServiceDelivery(serviceID, state) + return + } + if state.pending == nil && state.current.label == label { + c.deliveryMu.Unlock() + return + } + if state.pending == nil || state.pending.label != label { + state.pending = &req + select { + case state.update <- struct{}{}: + default: + } + } + c.deliveryMu.Unlock() +} + +func (c *NotificationClass) runServiceDelivery(serviceID uint64, state *orderedServiceDelivery) { + defer close(state.done) + delay := 30 * time.Second + for { + select { + case <-state.cancel: + return + default: + } + delivered := c.SendNotification(state.current.groupID, state.current.desc, state.current.label) + c.deliveryMu.Lock() + // Do not advance to a later phase until every recipient has received the + // current one. Otherwise a failed incident can be discarded as soon as a + // recovery is queued, leaving that recipient with a recovery but no incident. + if delivered && state.pending != nil { + next := state.pending + state.current = *next + state.pending = nil + delay = 30 * time.Second + select { + case <-state.update: + default: + } + c.deliveryMu.Unlock() + continue + } + if delivered { + if c.serviceDelivery[serviceID] == state { + delete(c.serviceDelivery, serviceID) + } + c.deliveryMu.Unlock() + return + } + c.deliveryMu.Unlock() + timer := time.NewTimer(delay) + select { + case <-state.cancel: + timer.Stop() + return + case <-state.update: + timer.Stop() + case <-timer.C: + delay = min(delay*2, 15*time.Minute) + } + } +} + +func (c *NotificationClass) CancelServiceState(serviceID uint64) { + if c == nil { + return + } + c.deliveryMu.Lock() + if state := c.serviceDelivery[serviceID]; state != nil { + close(state.cancel) + delete(c.serviceDelivery, serviceID) + } + c.deliveryMu.Unlock() +} + +// SendNotificationAsync coalesces concurrent service-probe reports for the +// same event before allocating a webhook goroutine. Alert transitions use the +// synchronous SendNotification path so they retain their retry semantics. +func (c *NotificationClass) SendNotificationAsync(groupID uint64, desc, muteLabel string, ext ...*model.Server) { + c.sendNotificationAsync(groupID, desc, muteLabel, nil, ext...) +} + +func (c *NotificationClass) sendNotificationAsync(groupID uint64, desc, muteLabel string, onSuccess func(), ext ...*model.Server) { + key := fmt.Sprintf("%d:%s", groupID, muteLabel) + c.deliveryMu.Lock() + if c.asyncInFlight == nil { + c.asyncInFlight = make(map[string]struct{}) + } + if _, busy := c.asyncInFlight[key]; busy { + c.deliveryMu.Unlock() + return + } + if time.Now().Before(c.asyncRetryAfter[key]) { + c.deliveryMu.Unlock() + return + } + c.asyncInFlight[key] = struct{}{} + c.deliveryMu.Unlock() + go func() { + delivered := c.SendNotification(groupID, desc, muteLabel, ext...) + if delivered && onSuccess != nil { + onSuccess() + } + c.deliveryMu.Lock() + if delivered { + delete(c.asyncRetryAfter, key) + delete(c.asyncRetryDelay, key) + } else { + if c.asyncRetryAfter == nil { + c.asyncRetryAfter = make(map[string]time.Time) + c.asyncRetryDelay = make(map[string]time.Duration) + } + delay := c.asyncRetryDelay[key] + if delay == 0 { + delay = 30 * time.Second + } else { + delay = min(delay*2, 15*time.Minute) + } + c.asyncRetryDelay[key] = delay + c.asyncRetryAfter[key] = time.Now().Add(delay) + c.pruneAsyncBackoffLocked() + } + delete(c.asyncInFlight, key) + c.deliveryMu.Unlock() + }() +} + +const maxAsyncBackoffKeys = 4096 + +// Bound memory when a failing webhook sees many short-lived event labels. +// Evicted old event keys are no longer active after their phase changes. +func (c *NotificationClass) pruneAsyncBackoffLocked() { + if len(c.asyncRetryAfter) <= maxAsyncBackoffKeys { + return + } + now := time.Now() + for key, until := range c.asyncRetryAfter { + if !now.Before(until) { + delete(c.asyncRetryAfter, key) + delete(c.asyncRetryDelay, key) + } + } + for key := range c.asyncRetryAfter { + if len(c.asyncRetryAfter) <= maxAsyncBackoffKeys { + break + } + delete(c.asyncRetryAfter, key) + delete(c.asyncRetryDelay, key) + } +} + +type notificationKeyLock struct { + mu sync.Mutex + refs int +} + +func (c *NotificationClass) lockDeliveryKey(key string) func() { + if key == "" { + return func() {} + } + c.deliveryMu.Lock() + if c.deliveryKeys == nil { + c.deliveryKeys = make(map[string]*notificationKeyLock) + } + entry := c.deliveryKeys[key] + if entry == nil { + entry = new(notificationKeyLock) + c.deliveryKeys[key] = entry + } + entry.refs++ + c.deliveryMu.Unlock() + entry.mu.Lock() + return func() { + entry.mu.Unlock() + c.deliveryMu.Lock() + entry.refs-- + if entry.refs == 0 { + delete(c.deliveryKeys, key) + } + c.deliveryMu.Unlock() + } } func NewNotificationClass() *NotificationClass { @@ -202,44 +422,29 @@ func (c *NotificationClass) sortList() { func (c *NotificationClass) UnMuteNotification(notificationGroupID uint64, muteLabel string) { fullMuteLabel := NotificationMuteLabel.AppendNotificationGroupName(muteLabel, c.GetGroupName(notificationGroupID)) - Cache.Delete(fullMuteLabel) + c.listMu.RLock() + for id := range c.groupToIDList[notificationGroupID] { + key := notificationMethodMuteLabel(fullMuteLabel, id) + c.deliveryMu.Lock() + if c.muteGeneration == nil { + c.muteGeneration = make(map[string]uint64) + } + c.muteGeneration[key]++ + Cache.Delete(key) + c.deliveryMu.Unlock() + } + c.listMu.RUnlock() +} + +func notificationMethodMuteLabel(label string, id uint64) string { + return fmt.Sprintf("%s:method-%d", label, id) } // SendNotification 向指定的通知方式组的所有通知方式发送通知 -func (c *NotificationClass) SendNotification(notificationGroupID uint64, desc string, muteLabel string, ext ...*model.Server) { +func (c *NotificationClass) SendNotification(notificationGroupID uint64, desc string, muteLabel string, ext ...*model.Server) bool { + fullMuteLabel := "" if muteLabel != "" { - // 将通知方式组名称加入静音标志 - muteLabel := NotificationMuteLabel.AppendNotificationGroupName(muteLabel, c.GetGroupName(notificationGroupID)) - // 通知防骚扰策略 - var flag bool - if cacheN, has := Cache.Get(muteLabel); has { - nHistory := cacheN.(NotificationHistory) - // 每次提醒都增加一倍等待时间，最后每天最多提醒一次 - if time.Now().After(nHistory.Until) { - flag = true - nHistory.Duration *= 2 - if nHistory.Duration > time.Hour*24 { - nHistory.Duration = time.Hour * 24 - } - nHistory.Until = time.Now().Add(nHistory.Duration) - // 缓存有效期加 10 分钟 - Cache.Set(muteLabel, nHistory, nHistory.Duration+time.Minute*10) - } - } else { - // 新提醒直接通知 - flag = true - Cache.Set(muteLabel, NotificationHistory{ - Duration: firstNotificationDelay, - Until: time.Now().Add(firstNotificationDelay), - }, firstNotificationDelay+time.Minute*10) - } - - if !flag { - if Conf.Debug { - log.Println("NEZHA>> Muted repeated notification", desc, muteLabel) - } - return - } + fullMuteLabel = NotificationMuteLabel.AppendNotificationGroupName(muteLabel, c.GetGroupName(notificationGroupID)) } // Copy the group under the lock. Webhook delivery can take minutes and must // not block notification or notification-group updates while it is in flight. @@ -250,24 +455,82 @@ func (c *NotificationClass) SendNotification(notificationGroupID uint64, desc st } c.listMu.RUnlock() - for _, n := range notifications { - log.Printf("NEZHA>> Try to notify %s", n.Name) + if len(notifications) == 0 { + log.Printf("NEZHA>> Notification group %d has no recipients", notificationGroupID) + return false } + allDelivered := true for _, n := range notifications { - ns := model.NotificationServerBundle{ - Notification: n, - Server: nil, - Loc: Loc, - } + var server *model.Server if len(ext) > 0 { - ns.Server = ext[0] + server = ext[0] } - if err := ns.Send(desc); err != nil { - log.Printf("NEZHA>> Sending notification to %s failed: %v", n.Name, err) - } else { - log.Printf("NEZHA>> Sending notification to %s succeeded", n.Name) + if !c.sendRecipient(n, desc, fullMuteLabel, server) { + allDelivered = false } } + return allDelivered +} + +func (c *NotificationClass) sendRecipient(n *model.Notification, desc, fullMuteLabel string, server *model.Server) bool { + if n == nil { + return false + } + cacheKey := "" + if fullMuteLabel != "" { + cacheKey = notificationMethodMuteLabel(fullMuteLabel, n.ID) + } + unlock := c.lockDeliveryKey(cacheKey) + defer unlock() + var generation uint64 + if cacheKey != "" { + c.deliveryMu.Lock() + generation = c.muteGeneration[cacheKey] + c.deliveryMu.Unlock() + } + var history NotificationHistory + if cacheKey != "" { + if value, exists := Cache.Get(cacheKey); exists { + history = value.(NotificationHistory) + if time.Now().Before(history.Until) { + return true // this recipient already received this phase + } + } + } + log.Printf("NEZHA>> Try to notify %s", n.Name) + var err error + if c.sendForTest != nil { + err = c.sendForTest(n, desc, server) + } else { + err = (&model.NotificationServerBundle{Notification: n, Server: server, Loc: Loc}).Send(desc) + } + if err != nil { + // Webhook errors can contain a credential-bearing request URL. Keep + // failure type without ever writing the URL/token to dashboard logs. + log.Printf("NEZHA>> Sending notification to %s failed (%s)", n.Name, notificationFailureSummary(err)) + return false + } + log.Printf("NEZHA>> Sending notification to %s succeeded", n.Name) + if cacheKey != "" { + if history.Duration == 0 { + history.Duration = firstNotificationDelay + } else { + history.Duration = min(history.Duration*2, 24*time.Hour) + } + history.Until = time.Now().Add(history.Duration) + c.deliveryMu.Lock() + if c.muteGeneration[cacheKey] == generation { + Cache.Set(cacheKey, history, history.Duration+10*time.Minute) + } + c.deliveryMu.Unlock() + } + return true +} + +func notificationFailureSummary(err error) string { + // Even wrapped URL errors can include credentials in path/query. Log only + // the error type; the HTTP status path already returns a dedicated type. + return fmt.Sprintf("%T", err) } type _NotificationMuteLabel struct{} diff --git a/service/singleton/notification_test.go b/service/singleton/notification_test.go index 5e08de31..39ea4102 100644 --- a/service/singleton/notification_test.go +++ b/service/singleton/notification_test.go @@ -1,12 +1,405 @@ package singleton import ( + "errors" + "fmt" + "net/url" + "strings" + "sync/atomic" "testing" "time" + "github.com/patrickmn/go-cache" + "github.com/nezhahq/nezha/model" ) +func TestNotificationFailureDoesNotMuteRetryOrResendSuccessfulRecipient(t *testing.T) { + oldCache := Cache + Cache = cache.New(time.Minute, time.Minute) + t.Cleanup(func() { Cache = oldCache }) + first := &model.Notification{Common: model.Common{ID: 301}, Name: "first"} + second := &model.Notification{Common: model.Common{ID: 302}, Name: "second"} + nc := newNotificationClassWithItems(first, second) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 303}, Name: "retry-test"}, []uint64{first.ID, second.ID}) + counts := map[uint64]int{} + nc.sendForTest = func(n *model.Notification, _ string, _ *model.Server) error { + counts[n.ID]++ + if n.ID == second.ID && counts[n.ID] == 1 { + return errors.New("temporary failure") + } + return nil + } + label := "alert-retry-test" + nc.UnMuteNotification(303, label) + if nc.SendNotification(303, "incident", label) { + t.Fatal("partial delivery must report failure") + } + if !nc.SendNotification(303, "incident", label) { + t.Fatal("failed recipient should be retried successfully") + } + if counts[first.ID] != 1 || counts[second.ID] != 2 { + t.Fatalf("delivery attempts = %v, want first=1 second=2", counts) + } + nc.UnMuteNotification(303, label) +} + +func TestNotificationConcurrentSameRecipientSendsOnce(t *testing.T) { + oldCache := Cache + Cache = cache.New(time.Minute, time.Minute) + t.Cleanup(func() { Cache = oldCache }) + n := &model.Notification{Common: model.Common{ID: 401}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 402}, Name: "concurrent-test"}, []uint64{n.ID}) + entered := make(chan struct{}) + release := make(chan struct{}) + var calls atomic.Int32 + nc.sendForTest = func(*model.Notification, string, *model.Server) error { + if calls.Add(1) == 1 { + close(entered) + <-release + } + return nil + } + done := make(chan bool, 2) + go func() { done <- nc.SendNotification(402, "incident", "same-event") }() + <-entered + go func() { done <- nc.SendNotification(402, "incident", "same-event") }() + close(release) + if !<-done || !<-done { + t.Fatal("concurrent delivery returned failure") + } + if calls.Load() != 1 { + t.Fatalf("same event delivered %d times, want 1", calls.Load()) + } +} + +func TestNotificationAsyncCoalescesBlockedServiceReports(t *testing.T) { + oldCache := Cache + Cache = cache.New(time.Minute, time.Minute) + t.Cleanup(func() { Cache = oldCache }) + n := &model.Notification{Common: model.Common{ID: 501}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 502}, Name: "service-test"}, []uint64{n.ID}) + entered := make(chan struct{}) + release := make(chan struct{}) + var calls atomic.Int32 + nc.sendForTest = func(*model.Notification, string, *model.Server) error { + if calls.Add(1) == 1 { + close(entered) + <-release + } + return nil + } + nc.SendNotificationAsync(502, "down", "service-down") + <-entered + for i := 0; i < 1000; i++ { + nc.SendNotificationAsync(502, "down", "service-down") + } + nc.deliveryMu.Lock() + inFlight := len(nc.asyncInFlight) + nc.deliveryMu.Unlock() + if inFlight != 1 || calls.Load() != 1 { + t.Fatalf("in-flight sends=%d attempts=%d, want one each", inFlight, calls.Load()) + } + close(release) + deadline := time.After(time.Second) + for { + nc.deliveryMu.Lock() + remaining := len(nc.asyncInFlight) + nc.deliveryMu.Unlock() + if remaining == 0 { + break + } + select { + case <-deadline: + t.Fatal("async notification did not finish") + case <-time.After(time.Millisecond): + } + } +} + +func TestNotificationFailureSummaryOmitsCredentialURL(t *testing.T) { + err := &url.Error{Op: "Post", URL: "https://example.test/bot-secret-token/send?key=secret-query", Err: errors.New("timeout")} + if summary := notificationFailureSummary(err); strings.Contains(summary, "secret") || strings.Contains(summary, "example.test") { + t.Fatalf("credential-bearing URL leaked: %s", summary) + } +} + +func TestNotificationAsyncBacksOffFastFailures(t *testing.T) { + oldCache := Cache + Cache = cache.New(time.Minute, time.Minute) + t.Cleanup(func() { Cache = oldCache }) + n := &model.Notification{Common: model.Common{ID: 511}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 512}, Name: "backoff-test"}, []uint64{n.ID}) + var calls atomic.Int32 + nc.sendForTest = func(*model.Notification, string, *model.Server) error { + calls.Add(1) + return errors.New("fast failure") + } + key := "512:service-down" + nc.SendNotificationAsync(512, "down", "service-down") + deadline := time.After(time.Second) + for { + nc.deliveryMu.Lock() + busy := len(nc.asyncInFlight) != 0 + after := nc.asyncRetryAfter[key] + nc.deliveryMu.Unlock() + if !busy && !after.IsZero() { + break + } + select { + case <-deadline: + t.Fatal("first failed send did not record backoff") + case <-time.After(time.Millisecond): + } + } + for i := 0; i < 1000; i++ { + nc.SendNotificationAsync(512, "down", "service-down") + } + if calls.Load() != 1 { + t.Fatalf("fast failures caused %d attempts before retry-after", calls.Load()) + } +} + +func TestUnMuteDuringInflightSendCannotRemuteRecovery(t *testing.T) { + oldCache := Cache + Cache = cache.New(time.Minute, time.Minute) + t.Cleanup(func() { Cache = oldCache }) + n := &model.Notification{Common: model.Common{ID: 521}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 522}, Name: "generation-test"}, []uint64{n.ID}) + entered := make(chan struct{}) + release := make(chan struct{}) + var calls atomic.Int32 + nc.sendForTest = func(*model.Notification, string, *model.Server) error { + if calls.Add(1) == 1 { + close(entered) + <-release + } + return nil + } + done := make(chan bool, 1) + go func() { done <- nc.SendNotification(522, "first failure", "static-label") }() + <-entered + nc.UnMuteNotification(522, "static-label") + close(release) + if !<-done { + t.Fatal("first send failed unexpectedly") + } + if !nc.SendNotification(522, "next failure", "static-label") || calls.Load() != 2 { + t.Fatalf("old in-flight send re-muted next failure: calls=%d", calls.Load()) + } +} + +func TestAsyncBackoffMemoryIsBounded(t *testing.T) { + nc := &NotificationClass{ + asyncRetryAfter: make(map[string]time.Time), + asyncRetryDelay: make(map[string]time.Duration), + } + for i := 0; i < maxAsyncBackoffKeys+100; i++ { + key := fmt.Sprintf("old-event-%d", i) + nc.asyncRetryAfter[key] = time.Now().Add(time.Minute) + nc.asyncRetryDelay[key] = time.Minute + } + nc.pruneAsyncBackoffLocked() + if len(nc.asyncRetryAfter) > maxAsyncBackoffKeys || len(nc.asyncRetryDelay) != len(nc.asyncRetryAfter) { + t.Fatalf("unbounded backoff maps: after=%d delay=%d", len(nc.asyncRetryAfter), len(nc.asyncRetryDelay)) + } +} + +func TestServiceStateNotificationsKeepIncidentRecoveryOrder(t *testing.T) { + oldCache := Cache + Cache = cache.New(time.Minute, time.Minute) + t.Cleanup(func() { Cache = oldCache }) + n := &model.Notification{Common: model.Common{ID: 531}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 532}, Name: "ordered-test"}, []uint64{n.ID}) + entered := make(chan struct{}) + release := make(chan struct{}) + delivered := make(chan string, 3) + nc.sendForTest = func(_ *model.Notification, desc string, _ *model.Server) error { + if desc == "down" { + close(entered) + <-release + } + delivered <- desc + return nil + } + nc.SendServiceState(533, 532, "down", "state-down-1") + <-entered + nc.SendServiceState(533, 532, "good", "state-good-2") + select { + case message := <-delivered: + t.Fatalf("service transition overtook blocked incident: %s", message) + default: + } + close(release) + for _, want := range []string{"down", "good"} { + select { + case got := <-delivered: + if got != want { + t.Fatalf("delivery order got %s, want %s", got, want) + } + case <-time.After(time.Second): + t.Fatalf("missing %s delivery", want) + } + } + deadline := time.After(time.Second) + for { + nc.deliveryMu.Lock() + remaining := len(nc.serviceDelivery) + nc.deliveryMu.Unlock() + if remaining == 0 { + break + } + select { + case <-deadline: + t.Fatal("ordered delivery worker did not exit") + case <-time.After(time.Millisecond): + } + } +} + +func TestServiceStateRetriesFailedIncidentBeforePendingRecovery(t *testing.T) { + oldCache := Cache + Cache = cache.New(time.Minute, time.Minute) + t.Cleanup(func() { Cache = oldCache }) + n := &model.Notification{Common: model.Common{ID: 535}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 536}, Name: "retry-order-test"}, []uint64{n.ID}) + t.Cleanup(func() { nc.CancelServiceState(537) }) + + attempts := make(chan string, 3) + releaseFirst := make(chan struct{}) + var downAttempts atomic.Int32 + nc.sendForTest = func(_ *model.Notification, desc string, _ *model.Server) error { + attempts <- desc + if desc == "down" && downAttempts.Add(1) == 1 { + <-releaseFirst + return errors.New("temporary incident failure") + } + return nil + } + + nc.SendServiceState(537, 536, "down", "state-down-1") + nc.deliveryMu.Lock() + state := nc.serviceDelivery[537] + nc.deliveryMu.Unlock() + select { + case got := <-attempts: + if got != "down" { + t.Fatalf("first delivery = %q, want down", got) + } + case <-time.After(time.Second): + t.Fatal("incident delivery did not start") + } + nc.SendServiceState(537, 536, "good", "state-good-2") + close(releaseFirst) + + for _, want := range []string{"down", "good"} { + select { + case got := <-attempts: + if got != want { + t.Fatalf("delivery after failed incident = %q, want %q", got, want) + } + case <-time.After(time.Second): + t.Fatalf("missing %s delivery", want) + } + } + select { + case <-state.done: + case <-time.After(time.Second): + t.Fatal("ordered delivery worker did not finish") + } +} + +func TestServiceStateCancelStopsOldConfiguration(t *testing.T) { + oldCache := Cache + Cache = cache.New(time.Minute, time.Minute) + t.Cleanup(func() { Cache = oldCache }) + n := &model.Notification{Common: model.Common{ID: 541}, Name: "hook"} + nc := newNotificationClassWithItems(n) + nc.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 542}, Name: "old-group"}, []uint64{n.ID}) + entered := make(chan struct{}) + release := make(chan struct{}) + var calls atomic.Int32 + nc.sendForTest = func(*model.Notification, string, *model.Server) error { + if calls.Add(1) == 1 { + close(entered) + <-release + } + return errors.New("old webhook failed") + } + nc.SendServiceState(543, 542, "down", "old-event") + <-entered + nc.deliveryMu.Lock() + state := nc.serviceDelivery[543] + nc.deliveryMu.Unlock() + nc.CancelServiceState(543) + nc.deliveryMu.Lock() + remaining := len(nc.serviceDelivery) + nc.deliveryMu.Unlock() + if remaining != 0 { + t.Fatal("old service delivery worker remained registered after cancellation") + } + close(release) + select { + case <-state.done: + case <-time.After(time.Second): + t.Fatal("canceled old delivery worker did not exit") + } +} + +func TestServiceUpdateCancelsOldRecipientRetry(t *testing.T) { + ss := newServiceMonitorSecurityHarness(t) + service := &model.Service{ + Common: model.Common{ID: 551, UserID: 1}, + Name: "generic-probe", + Type: model.TaskTypeTCPPing, + Target: "example.invalid:443", + Duration: 3600, + Notify: true, + NotificationGroupID: 552, + Cover: model.ServiceCoverIgnoreAll, + } + addServiceMonitorSecurityService(t, ss, service) + n := &model.Notification{Common: model.Common{ID: 553}, Name: "old-hook"} + NotificationShared.InsertForTest(n) + NotificationShared.UpdateGroup(&model.NotificationGroup{Common: model.Common{ID: 552}, Name: "old-group"}, []uint64{n.ID}) + entered := make(chan struct{}) + release := make(chan struct{}) + NotificationShared.sendForTest = func(*model.Notification, string, *model.Server) error { + close(entered) + <-release + return errors.New("failed old recipient") + } + NotificationShared.SendServiceState(service.ID, 552, "old down", "old-state") + <-entered + NotificationShared.deliveryMu.Lock() + state := NotificationShared.serviceDelivery[service.ID] + NotificationShared.deliveryMu.Unlock() + updated := *service + updated.Notify = false + updated.NotificationGroupID = 554 + if err := ss.Update(&updated); err != nil { + t.Fatal(err) + } + close(release) + select { + case <-state.done: + case <-time.After(time.Second): + t.Fatal("old recipient retry did not stop after service update") + } + NotificationShared.deliveryMu.Lock() + remaining := len(NotificationShared.serviceDelivery) + NotificationShared.deliveryMu.Unlock() + if remaining != 0 { + t.Fatal("old recipient retained after service update") + } +} + func newNotificationClassWithItems(items ...*model.Notification) *NotificationClass { nc := NewEmptyNotificationClassForTest() for _, item := range items { diff --git a/service/singleton/server.go b/service/singleton/server.go index a74a7f62..64ffd645 100644 --- a/service/singleton/server.go +++ b/service/singleton/server.go @@ -117,6 +117,7 @@ func (c *ServerClass) Update(s *model.Server, uuid string) { func (c *ServerClass) Delete(idList []uint64) { c.lockLifecycleWrite() defer c.unlockLifecycleWrite() + cancelAlertDeliveriesForServers(idList) c.listMu.Lock() diff --git a/service/singleton/servicesentinel.go b/service/singleton/servicesentinel.go index a2c6ef8d..3745c915 100644 --- a/service/singleton/servicesentinel.go +++ b/service/singleton/servicesentinel.go @@ -2,6 +2,7 @@ package singleton import ( "cmp" + "crypto/sha256" "fmt" "iter" "log" @@ -9,6 +10,7 @@ import ( "slices" "strings" "sync" + "sync/atomic" "time" "github.com/gin-gonic/gin" @@ -22,9 +24,11 @@ import ( ) const ( - _CurrentStatusSize = 30 // 统计 15 分钟内的数据为当前状态 + _CurrentStatusSize = 30 // initial capacity for the rolling service window ) +var serviceEventSequence atomic.Uint64 + type serviceResponseItem struct { model.ServiceResponseItem @@ -46,9 +50,101 @@ type _TodayStatsOfService struct { type serviceResponseData = _TodayStatsOfService type serviceTaskStatus struct { - lastStatus uint8 - t time.Time - result []*pb.TaskResult + lastStatus uint8 + stateEpoch uint64 + lastHistoryAt time.Time + history serviceResponseData + lastData string + result []serviceWindowBucket +} + +type serviceProbeSample struct { + successful bool + failed bool + delay float64 +} + +type serviceWindowBucket struct { + at time.Time + reporters map[uint64]serviceProbeSample +} + +func (status *serviceTaskStatus) appendResult(now time.Time, reporter uint64, result *pb.TaskResult) { + bucketAt := now.Truncate(30 * time.Second) + if len(status.result) > 0 && bucketAt.Before(status.result[len(status.result)-1].at) { + status.result = nil // wall-clock moved backwards; old buckets are not comparable + } + if len(status.result) == 0 || !status.result[len(status.result)-1].at.Equal(bucketAt) { + status.result = append(status.result, serviceWindowBucket{at: bucketAt, reporters: make(map[uint64]serviceProbeSample)}) + } + bucket := &status.result[len(status.result)-1] + sample := bucket.reporters[reporter] + if result.Successful { + sample.successful = true + sample.delay = float64(result.Delay) + } else { + sample.failed = true + } + bucket.reporters[reporter] = sample + cutoff := now.Add(-15 * time.Minute) + first := 0 + for first < len(status.result) && status.result[first].at.Before(cutoff) { + first++ + } + if first > 0 { + status.result = append([]serviceWindowBucket(nil), status.result[first:]...) + } + if result.Successful { + status.history.Up++ + status.history.Delay = (status.history.Delay*float64(status.history.Up-1) + float64(result.Delay)) / float64(status.history.Up) + } else { + status.history.Down++ + } + status.lastData = result.Data +} + +// flushServiceHistory persists the previous non-overlapping reporting period +// before the next observation is added, so long probe gaps cannot lose or +// double-count samples in the 15-minute availability history. +func (ss *ServiceSentinel) flushServiceHistory(serviceID uint64, status *serviceTaskStatus, now time.Time) { + if status.lastHistoryAt.IsZero() || now.Before(status.lastHistoryAt.Add(15*time.Minute)) { + return + } + if status.history.Up+status.history.Down == 0 { + status.lastHistoryAt = now + return + } + if !TSDBEnabled() { + if err := DB.Create(&model.ServiceHistory{ + CreatedAt: status.lastHistoryAt, + ServiceID: serviceID, + AvgDelay: status.history.Delay, + Data: status.lastData, + Up: status.history.Up, + Down: status.history.Down, + }).Error; err != nil { + log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err) + return + } + } + status.history = serviceResponseData{} + status.lastHistoryAt = now +} + +func (status *serviceTaskStatus) responseData() serviceResponseData { + var data serviceResponseData + for _, bucket := range status.result { + for _, sample := range bucket.reporters { + if sample.successful { + data.Up++ + data.Delay = (data.Delay*float64(data.Up-1) + sample.delay) / float64(data.Up) + } + if sample.failed { + data.Down++ + } + } + } + return data } type pingStore struct { @@ -228,7 +324,7 @@ func (ss *ServiceSentinel) loadServiceHistory() error { } ss.services[service.ID] = service ss.serviceCurrentStatusData[service.ID] = new(serviceTaskStatus) - ss.serviceCurrentStatusData[service.ID].result = make([]*pb.TaskResult, 0, _CurrentStatusSize) + ss.serviceCurrentStatusData[service.ID].result = make([]serviceWindowBucket, 0, _CurrentStatusSize) ss.serviceStatusToday[service.ID] = &_TodayStatsOfService{} validServices = append(validServices, service) } @@ -371,6 +467,9 @@ func (ss *ServiceSentinel) Update(m *model.Service) error { if err != nil { return err } + // A service edit may disable notifications or change their recipients. + // Cancel any retry worker holding the previous service configuration. + NotificationShared.CancelServiceState(m.ID) if ss.services[m.ID] != nil { // 停掉旧任务 CronShared.Remove(ss.services[m.ID].CronJobID) @@ -387,7 +486,7 @@ func (ss *ServiceSentinel) Update(m *model.Service) error { if ss.serviceCurrentStatusData[m.ID] == nil { ss.serviceCurrentStatusData[m.ID] = new(serviceTaskStatus) } - ss.serviceCurrentStatusData[m.ID].result = make([]*pb.TaskResult, 0, _CurrentStatusSize) + ss.serviceCurrentStatusData[m.ID].result = make([]serviceWindowBucket, 0, _CurrentStatusSize) ss.serviceStatusToday[m.ID] = &_TodayStatsOfService{} } // 更新这个任务 @@ -404,6 +503,7 @@ func (ss *ServiceSentinel) Delete(ids []uint64) { defer ss.servicesLock.Unlock() for _, id := range ids { + NotificationShared.CancelServiceState(id) delete(ss.serviceCurrentStatusData, id) delete(ss.serviceResponseDataStore, id) delete(ss.serviceResponsePing, id) @@ -669,32 +769,16 @@ func (ss *ServiceSentinel) processReport(r ReportData, serverShared *ServerClass } currentTime := time.Now() - if serviceCurrentStatusData.t.IsZero() { - serviceCurrentStatusData.t = currentTime - } - - // 写入当前数据 - if serviceCurrentStatusData.t.Before(currentTime) { - serviceCurrentStatusData.t = currentTime.Add(30 * time.Second) - serviceCurrentStatusData.result = append(serviceCurrentStatusData.result, mh) + ss.flushServiceHistory(mh.GetId(), serviceCurrentStatusData, currentTime) + // Include every accepted reporter's result. A fixed count of samples is + // not a 15-minute window when probes have different intervals or reporters. + serviceCurrentStatusData.appendResult(currentTime, r.Reporter, mh) + if serviceCurrentStatusData.lastHistoryAt.IsZero() { + serviceCurrentStatusData.lastHistoryAt = currentTime } // 更新当前状态 - ss.serviceResponseDataStore[mh.GetId()] = serviceResponseData{} - - // 永远是最新的 30 个数据的状态 [01:00, 02:00, 03:00] -> [04:00, 02:00, 03: 00] - for _, cs := range serviceCurrentStatusData.result { - if cs.GetId() > 0 { - rd := ss.serviceResponseDataStore[mh.GetId()] - if cs.Successful { - rd.Up++ - rd.Delay = (rd.Delay*float64(rd.Up-1) + float64(cs.Delay)) / float64(rd.Up) - } else { - rd.Down++ - } - ss.serviceResponseDataStore[mh.GetId()] = rd - } - } + ss.serviceResponseDataStore[mh.GetId()] = serviceCurrentStatusData.responseData() // 计算在线率， var stateCode uint8 @@ -704,24 +788,11 @@ func (ss *ServiceSentinel) processReport(r ReportData, serverShared *ServerClass if rd.Down+rd.Up > 0 { upPercent = rd.Up * 100 / (rd.Down + rd.Up) } - stateCode = GetStatusCode(upPercent) - } - - if len(serviceCurrentStatusData.result) == _CurrentStatusSize { - serviceCurrentStatusData.t = currentTime - if !TSDBEnabled() { - rd := ss.serviceResponseDataStore[mh.GetId()] - if err := DB.Create(&model.ServiceHistory{ - ServiceID: mh.GetId(), - AvgDelay: rd.Delay, - Data: mh.Data, - Up: rd.Up, - Down: rd.Down, - }).Error; err != nil { - log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err) - } + if rd.Down+rd.Up == 0 { + stateCode = StatusNoData + } else { + stateCode = GetStatusCode(upPercent) } - serviceCurrentStatusData.result = serviceCurrentStatusData.result[:0] } // 延迟报警 @@ -734,77 +805,100 @@ func (ss *ServiceSentinel) processReport(r ReportData, serverShared *ServerClass lastStatus := serviceCurrentStatusData.lastStatus // 存储新的状态值 serviceCurrentStatusData.lastStatus = stateCode - - notifyCheck(&r, m, cs, mh, lastStatus, stateCode) - } - - // TLS 证书报警 - if ss.serviceReportBeforeTLSSideEffectsHook != nil { - ss.serviceReportBeforeTLSSideEffectsHook(mh.GetId()) - } - var errMsg string - if strings.HasPrefix(mh.Data, "SSL证书错误：") { - // i/o timeout、connection timeout、EOF 错误 - if !strings.HasSuffix(mh.Data, "timeout") && - !strings.HasSuffix(mh.Data, "EOF") && - !strings.HasSuffix(mh.Data, "timed out") { - errMsg = mh.Data - if cs.Notify { - muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), "network") - go NotificationShared.SendNotification(cs.NotificationGroupID, Localizer.Tf("[TLS] Fetch cert info failed, Reporter: %s, Error: %s", cs.Name, errMsg), muteLabel) - } + if stateCode != lastStatus { + serviceCurrentStatusData.stateEpoch = serviceEventSequence.Add(1) } - } else { - // 清除网络错误静音缓存 - NotificationShared.UnMuteNotification(cs.NotificationGroupID, NotificationMuteLabel.ServiceTLS(mh.GetId(), "network")) - var newCert = strings.Split(mh.Data, "|") - if len(newCert) > 1 { - enableNotify := cs.Notify + notifyCheck(&r, m, cs, mh, lastStatus, stateCode, serviceCurrentStatusData.stateEpoch) + } - // 首次获取证书信息时，缓存证书信息 - if ss.tlsCertCache[mh.GetId()] == "" { - ss.tlsCertCache[mh.GetId()] = mh.Data + // TLS 证书报警仅适用于 HTTPS 探针。其它探针的任意响应文本 + // 不得被当作证书资料解析。 + if mh.Type == model.TaskTypeHTTPGet && strings.HasPrefix(strings.ToLower(cs.Target), "https://") { + if ss.serviceReportBeforeTLSSideEffectsHook != nil { + ss.serviceReportBeforeTLSSideEffectsHook(mh.GetId()) + } + var errMsg string + if strings.HasPrefix(mh.Data, "SSL证书错误：") { + // i/o timeout、connection timeout、EOF 错误 + if !strings.HasSuffix(mh.Data, "timeout") && + !strings.HasSuffix(mh.Data, "EOF") && + !strings.HasSuffix(mh.Data, "timed out") { + errMsg = mh.Data + if cs.Notify { + muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), "network") + NotificationShared.SendNotificationAsync(cs.NotificationGroupID, Localizer.Tf("[TLS] Fetch cert info failed, Reporter: %s, Error: %s", cs.Name, errMsg), muteLabel) + } } + } else { + // 清除网络错误静音缓存 + NotificationShared.UnMuteNotification(cs.NotificationGroupID, NotificationMuteLabel.ServiceTLS(mh.GetId(), "network")) - oldCert := strings.Split(ss.tlsCertCache[mh.GetId()], "|") - isCertChanged := false - expiresOld, _ := time.Parse("2006-01-02 15:04:05 -0700 MST", oldCert[1]) - expiresNew, _ := time.Parse("2006-01-02 15:04:05 -0700 MST", newCert[1]) - - // 证书变更时，更新缓存 - if oldCert[0] != newCert[0] && !expiresNew.Equal(expiresOld) { - isCertChanged = true - ss.tlsCertCache[mh.GetId()] = mh.Data - } - - notificationGroupID := cs.NotificationGroupID - serviceName := cs.Name - - // 需要发送提醒 - if enableNotify { - // 证书过期提醒 - if expiresNew.Before(time.Now().AddDate(0, 0, 7)) { - expiresTimeStr := expiresNew.Format("2006-01-02 15:04:05") - errMsg = Localizer.Tf( - "The TLS certificate will expire within seven days. Expiration time: %s", - expiresTimeStr, - ) - - // 静音规则： 服务id+证书过期时间 - // 用于避免多个监测点对相同证书同时报警 - muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), fmt.Sprintf("expire_%s", expiresTimeStr)) - go NotificationShared.SendNotification(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), muteLabel) + var newCert = strings.Split(mh.Data, "|") + if len(newCert) > 1 { + enableNotify := cs.Notify + expiresNew, newErr := time.Parse("2006-01-02 15:04:05 -0700 MST", newCert[1]) + if newErr != nil { + log.Printf("NEZHA>> Ignoring malformed TLS certificate expiry for service %d", mh.GetId()) + return } - // 证书变更提醒 - if isCertChanged { - errMsg = Localizer.Tf( - "TLS certificate changed, old: issuer %s, expires at %s; new: issuer %s, expires at %s", - oldCert[0], expiresOld.Format("2006-01-02 15:04:05"), newCert[0], expiresNew.Format("2006-01-02 15:04:05")) + // 首次获取证书信息时，缓存证书信息 + if ss.tlsCertCache[mh.GetId()] == "" { + ss.tlsCertCache[mh.GetId()] = mh.Data + } - // 证书变更后会自动更新缓存，所以不需要静音 - go NotificationShared.SendNotification(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), "") + oldCert := strings.Split(ss.tlsCertCache[mh.GetId()], "|") + isCertChanged := false + expiresOld, oldErr := time.Parse("2006-01-02 15:04:05 -0700 MST", oldCert[1]) + if oldErr != nil { + log.Printf("NEZHA>> Ignoring malformed TLS certificate expiry for service %d", mh.GetId()) + return + } + + // 只有通知成功后才推进基线；失败时下一次探针结果会重试。 + if oldCert[0] != newCert[0] || !expiresNew.Equal(expiresOld) { + isCertChanged = true + if !enableNotify { + ss.tlsCertCache[mh.GetId()] = mh.Data + } + } + + notificationGroupID := cs.NotificationGroupID + serviceName := cs.Name + + // 需要发送提醒 + if enableNotify { + // 证书过期提醒 + if expiresNew.Before(time.Now().AddDate(0, 0, 7)) { + expiresTimeStr := expiresNew.Format("2006-01-02 15:04:05") + errMsg = Localizer.Tf( + "The TLS certificate will expire within seven days. Expiration time: %s", + expiresTimeStr, + ) + + // 静音规则： 服务id+证书过期时间 + // 用于避免多个监测点对相同证书同时报警 + muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), fmt.Sprintf("expire_%s", expiresTimeStr)) + NotificationShared.SendNotificationAsync(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), muteLabel) + } + + // 证书变更提醒 + if isCertChanged { + errMsg = Localizer.Tf( + "TLS certificate changed, old: issuer %s, expires at %s; new: issuer %s, expires at %s", + oldCert[0], expiresOld.Format("2006-01-02 15:04:05"), newCert[0], expiresNew.Format("2006-01-02 15:04:05")) + + oldRaw, newRaw, serviceID := ss.tlsCertCache[mh.GetId()], mh.Data, mh.GetId() + muteLabel := NotificationMuteLabel.ServiceTLS(serviceID, fmt.Sprintf("change_%x", sha256.Sum256([]byte(newRaw)))) + NotificationShared.sendNotificationAsync(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), muteLabel, func() { + ss.serviceResponseDataStoreLock.Lock() + if ss.serviceCurrentStatusData[serviceID] != nil && ss.tlsCertCache[serviceID] == oldRaw { + ss.tlsCertCache[serviceID] = newRaw + } + ss.serviceResponseDataStoreLock.Unlock() + }) + } } } } @@ -832,11 +926,11 @@ func delayCheck(r *ReportData, m map[uint64]*model.Server, ss *model.Service, mh if mh.Delay > ss.MaxLatency { // 延迟超过最大值 msg := Localizer.Tf("[Latency] %s %2f > %2f, Reporter: %s", ss.Name, mh.Delay, ss.MaxLatency, reporterServer.Name) - go NotificationShared.SendNotification(notificationGroupID, msg, minMuteLabel) + NotificationShared.SendNotificationAsync(notificationGroupID, msg, minMuteLabel) } else if mh.Delay < ss.MinLatency { // 延迟低于最小值 msg := Localizer.Tf("[Latency] %s %2f < %2f, Reporter: %s", ss.Name, mh.Delay, ss.MinLatency, reporterServer.Name) - go NotificationShared.SendNotification(notificationGroupID, msg, maxMuteLabel) + NotificationShared.SendNotificationAsync(notificationGroupID, msg, maxMuteLabel) } else { // 正常延迟， 清除静音缓存 NotificationShared.UnMuteNotification(notificationGroupID, minMuteLabel) @@ -845,7 +939,7 @@ func delayCheck(r *ReportData, m map[uint64]*model.Server, ss *model.Service, mh } func notifyCheck(r *ReportData, m map[uint64]*model.Server, - ss *model.Service, mh *pb.TaskResult, lastStatus, stateCode uint8) { + ss *model.Service, mh *pb.TaskResult, lastStatus, stateCode uint8, stateEpoch uint64) { // GHSA-jx78-55p5-rwv5: guard against concurrent server deletion (same TOCTOU // class as the 2026-07-21 fix, a few dozen lines lower in the same worker). // ServerShared has its own lock; m is a snapshot taken outside @@ -858,14 +952,9 @@ func notifyCheck(r *ReportData, m map[uint64]*model.Server, if isNeedSendNotification && reporterServer != nil { notificationGroupID := ss.NotificationGroupID notificationMsg := Localizer.Tf("[%s] %s Reporter: %s, Error: %s", StatusCodeToString(stateCode), ss.Name, reporterServer.Name, mh.Data) - muteLabel := NotificationMuteLabel.ServiceStateChanged(mh.GetId()) + muteLabel := fmt.Sprintf("%s:%d:%d", NotificationMuteLabel.ServiceStateChanged(mh.GetId()), stateCode, stateEpoch) - // 状态变更时，清除静音缓存 - if stateCode != lastStatus { - NotificationShared.UnMuteNotification(notificationGroupID, muteLabel) - } - - go NotificationShared.SendNotification(notificationGroupID, notificationMsg, muteLabel) + NotificationShared.SendServiceState(mh.GetId(), notificationGroupID, notificationMsg, muteLabel) } // 判断是否需要触发任务 @@ -890,9 +979,6 @@ const ( ) func GetStatusCode[T constraints.Float | constraints.Integer](percent T) uint8 { - if percent == 0 { - return StatusNoData - } if percent > 95 { return StatusGood } diff --git a/service/singleton/servicesentinel_window_test.go b/service/singleton/servicesentinel_window_test.go new file mode 100644 index 00000000..e5c85147 --- /dev/null +++ b/service/singleton/servicesentinel_window_test.go @@ -0,0 +1,63 @@ +package singleton + +import ( + "testing" + "time" + + pb "github.com/nezhahq/nezha/proto" +) + +func TestServiceWindowCountsFirstFailureAsDown(t *testing.T) { + status := new(serviceTaskStatus) + status.appendResult(time.Unix(1000, 0), 1, &pb.TaskResult{Successful: false}) + data := status.responseData() + if data.Down != 1 || data.Up != 0 || GetStatusCode(uint64(0)) != StatusDown { + t.Fatalf("first failed report = %+v, status %d; want Down", data, GetStatusCode(uint64(0))) + } +} + +func TestServiceWindowExpiresOldReportsAndDoesNotResetAtThirty(t *testing.T) { + status := new(serviceTaskStatus) + start := time.Unix(1000, 0) + for i := 0; i < 30; i++ { + status.appendResult(start.Add(time.Duration(i)*30*time.Second), 1, &pb.TaskResult{Successful: true}) + } + if data := status.responseData(); data.Up != 30 { + t.Fatalf("30th report reset window: %+v", data) + } + status.appendResult(start.Add(29*30*time.Second+time.Second), 2, &pb.TaskResult{Successful: false}) + if data := status.responseData(); data.Up != 30 || data.Down != 1 { + t.Fatalf("31st report corrupted window: %+v", data) + } + status.appendResult(start.Add(31*time.Minute), 1, &pb.TaskResult{Successful: false}) + if data := status.responseData(); data.Up != 0 || data.Down != 1 { + t.Fatalf("expired reports still influence status: %+v", data) + } +} + +func TestServiceWindowDeduplicatesFloodWithinBucket(t *testing.T) { + status := new(serviceTaskStatus) + now := time.Unix(1000, 0) + for i := 0; i < 10000; i++ { + status.appendResult(now, 1, &pb.TaskResult{Successful: false, Data: "ignored payload"}) + } + if len(status.result) != 1 || len(status.result[0].reporters) != 1 { + t.Fatalf("flood grew live window: buckets=%d", len(status.result)) + } + if data := status.responseData(); data.Down != 1 { + t.Fatalf("duplicate reporter skewed status: %+v", data) + } + if status.history.Down != 10000 { + t.Fatalf("history lost raw reports: %+v", status.history) + } +} + +func TestServiceWindowDoesNotEraseFailureWithLaterSuccess(t *testing.T) { + status := new(serviceTaskStatus) + now := time.Unix(1000, 0) + status.appendResult(now, 1, &pb.TaskResult{Successful: false}) + status.appendResult(now.Add(time.Second), 1, &pb.TaskResult{Successful: true, Delay: 20}) + if data := status.responseData(); data.Down != 1 || data.Up != 1 { + t.Fatalf("later success erased failure in same bucket: %+v", data) + } +} diff --git a/service/singleton/singleton.go b/service/singleton/singleton.go index bc906afd..8d41b70b 100644 --- a/service/singleton/singleton.go +++ b/service/singleton/singleton.go @@ -167,7 +167,7 @@ func CleanMonitorHistory() { for _, alert := range alerts { for _, rule := range alert.Rules { // 是不是流量记录规则 - if !rule.IsTransferDurationRule() { + if !rule.HasSafeCycleConfiguration() { continue } dataCouldRemoveBefore := rule.GetTransferDurationStart().UTC()