From c893aa57867d8a3d608e2f2f16d1594c9766cdf3 Mon Sep 17 00:00:00 2001 From: naiba Date: Tue, 21 Jul 2026 13:11:05 +0000 Subject: [PATCH] fix(service): serialize reports with service lifecycle Co-authored-by: naiba/CloudCode --- service/singleton/servicesentinel.go | 70 ++-- .../servicesentinel_lifecycle_test.go | 301 ++++++++++++++++++ 2 files changed, 347 insertions(+), 24 deletions(-) create mode 100644 service/singleton/servicesentinel_lifecycle_test.go diff --git a/service/singleton/servicesentinel.go b/service/singleton/servicesentinel.go index a8300176..2a98aab6 100644 --- a/service/singleton/servicesentinel.go +++ b/service/singleton/servicesentinel.go @@ -74,8 +74,11 @@ type ServiceSentinel struct { serviceCurrentStatusData map[uint64]*serviceTaskStatus // 当前任务结果缓存 serviceResponseDataStore map[uint64]serviceResponseData // 当前数据 - serviceResponsePing map[uint64]map[uint64]*pingStore // [service_id] -> ClientID -> delay - tlsCertCache map[uint64]string + serviceResponsePing map[uint64]map[uint64]*pingStore // guarded by serviceResponseDataStoreLock; [service_id] -> ClientID -> delay + tlsCertCache map[uint64]string // guarded by serviceResponseDataStoreLock + serviceReportValidatedHook func(uint64) + loadStatsResponseLockedHook func() + serviceReportBeforeTLSSideEffectsHook func(uint64) servicesLock sync.RWMutex serviceListLock sync.RWMutex @@ -386,6 +389,7 @@ func (ss *ServiceSentinel) Delete(ids []uint64) { for _, id := range ids { delete(ss.serviceCurrentStatusData, id) delete(ss.serviceResponseDataStore, id) + delete(ss.serviceResponsePing, id) delete(ss.tlsCertCache, id) delete(ss.serviceStatusToday, id) @@ -398,12 +402,15 @@ func (ss *ServiceSentinel) Delete(ids []uint64) { } func (ss *ServiceSentinel) LoadStats() map[uint64]*serviceResponseItem { - ss.servicesLock.RLock() - defer ss.servicesLock.RUnlock() ss.serviceResponseDataStoreLock.RLock() defer ss.serviceResponseDataStoreLock.RUnlock() + if ss.loadStatsResponseLockedHook != nil { + ss.loadStatsResponseLockedHook() + } ss.monthlyStatusLock.Lock() defer ss.monthlyStatusLock.Unlock() + ss.servicesLock.RLock() + defer ss.servicesLock.RUnlock() // 刷新最新一天的数据 for k := range ss.services { @@ -536,8 +543,23 @@ func (ss *ServiceSentinel) worker() { log.Printf("NEZHA>> Incorrect service monitor report %+v", r) continue } + if ss.serviceReportValidatedHook != nil { + ss.serviceReportValidatedHook(r.Data.GetId()) + } mh := r.Data + // Serialize Delete and Update before this accepted report causes any side effect. + ss.serviceResponseDataStoreLock.Lock() + serviceStatusToday := ss.serviceStatusToday[mh.GetId()] + serviceCurrentStatusData := ss.serviceCurrentStatusData[mh.GetId()] + currentService, serviceExists := ss.Get(mh.GetId()) + if serviceStatusToday == nil || serviceCurrentStatusData == nil || !serviceExists || + !canReportServiceResult(currentService, reporter, mh.GetType()) { + ss.serviceResponseDataStoreLock.Unlock() + continue + } + cs = currentService + if mh.Type == model.TaskTypeTCPPing || mh.Type == model.TaskTypeICMPPing { // TCP/ICMP Ping 使用平均值计算后再写入 serviceTcpMap, ok := ss.serviceResponsePing[mh.GetId()] @@ -594,33 +616,31 @@ func (ss *ServiceSentinel) worker() { } } - ss.serviceResponseDataStoreLock.Lock() // 写入当天状态 if mh.Successful { - ss.serviceStatusToday[mh.GetId()].Delay = (ss.serviceStatusToday[mh. - GetId()].Delay*float64(ss.serviceStatusToday[mh.GetId()].Up) + - float64(mh.Delay)) / float64(ss.serviceStatusToday[mh.GetId()].Up+1) - ss.serviceStatusToday[mh.GetId()].Up++ + serviceStatusToday.Delay = (serviceStatusToday.Delay*float64(serviceStatusToday.Up) + + float64(mh.Delay)) / float64(serviceStatusToday.Up+1) + serviceStatusToday.Up++ } else { - ss.serviceStatusToday[mh.GetId()].Down++ + serviceStatusToday.Down++ } currentTime := time.Now() - if ss.serviceCurrentStatusData[mh.GetId()].t.IsZero() { - ss.serviceCurrentStatusData[mh.GetId()].t = currentTime + if serviceCurrentStatusData.t.IsZero() { + serviceCurrentStatusData.t = currentTime } // 写入当前数据 - if ss.serviceCurrentStatusData[mh.GetId()].t.Before(currentTime) { - ss.serviceCurrentStatusData[mh.GetId()].t = currentTime.Add(30 * time.Second) - ss.serviceCurrentStatusData[mh.GetId()].result = append(ss.serviceCurrentStatusData[mh.GetId()].result, mh) + if serviceCurrentStatusData.t.Before(currentTime) { + serviceCurrentStatusData.t = currentTime.Add(30 * time.Second) + serviceCurrentStatusData.result = append(serviceCurrentStatusData.result, mh) } // 更新当前状态 ss.serviceResponseDataStore[mh.GetId()] = serviceResponseData{} // 永远是最新的 30 个数据的状态 [01:00, 02:00, 03:00] -> [04:00, 02:00, 03: 00] - for _, cs := range ss.serviceCurrentStatusData[mh.GetId()].result { + for _, cs := range serviceCurrentStatusData.result { if cs.GetId() > 0 { rd := ss.serviceResponseDataStore[mh.GetId()] if cs.Successful { @@ -644,8 +664,8 @@ func (ss *ServiceSentinel) worker() { stateCode = GetStatusCode(upPercent) } - if len(ss.serviceCurrentStatusData[mh.GetId()].result) == _CurrentStatusSize { - ss.serviceCurrentStatusData[mh.GetId()].t = currentTime + if len(serviceCurrentStatusData.result) == _CurrentStatusSize { + serviceCurrentStatusData.t = currentTime if !TSDBEnabled() { rd := ss.serviceResponseDataStore[mh.GetId()] if err := DB.Create(&model.ServiceHistory{ @@ -658,10 +678,9 @@ func (ss *ServiceSentinel) worker() { log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err) } } - ss.serviceCurrentStatusData[mh.GetId()].result = ss.serviceCurrentStatusData[mh.GetId()].result[:0] + serviceCurrentStatusData.result = serviceCurrentStatusData.result[:0] } - cs, _ = ss.Get(mh.GetId()) m := ServerShared.GetList() // 延迟报警 if mh.Delay > 0 { @@ -669,16 +688,18 @@ func (ss *ServiceSentinel) worker() { } // 状态变更报警+触发任务执行 - if stateCode == StatusDown || stateCode != ss.serviceCurrentStatusData[mh.GetId()].lastStatus { - lastStatus := ss.serviceCurrentStatusData[mh.GetId()].lastStatus + if stateCode == StatusDown || stateCode != serviceCurrentStatusData.lastStatus { + lastStatus := serviceCurrentStatusData.lastStatus // 存储新的状态值 - ss.serviceCurrentStatusData[mh.GetId()].lastStatus = stateCode + serviceCurrentStatusData.lastStatus = stateCode notifyCheck(&r, m, cs, mh, lastStatus, stateCode) } - ss.serviceResponseDataStoreLock.Unlock() // TLS 证书报警 + if ss.serviceReportBeforeTLSSideEffectsHook != nil { + ss.serviceReportBeforeTLSSideEffectsHook(mh.GetId()) + } var errMsg string if strings.HasPrefix(mh.Data, "SSL证书错误：") { // i/o timeout、connection timeout、EOF 错误 @@ -746,6 +767,7 @@ func (ss *ServiceSentinel) worker() { } } } + ss.serviceResponseDataStoreLock.Unlock() } } diff --git a/service/singleton/servicesentinel_lifecycle_test.go b/service/singleton/servicesentinel_lifecycle_test.go new file mode 100644 index 00000000..0941b453 --- /dev/null +++ b/service/singleton/servicesentinel_lifecycle_test.go @@ -0,0 +1,301 @@ +package singleton + +import ( + "context" + "fmt" + "os" + "os/exec" + "strings" + "sync" + "testing" + "time" + + "github.com/nezhahq/nezha/model" +) + +const serviceSentinelLifecycleSuccessMarker = "service-sentinel-stale-report-lifecycle-success" + +func TestServiceSentinelWorkerIgnoresStaleReportAfterDeletion(t *testing.T) { + if os.Getenv("NEZHA_SERVICE_SENTINEL_LIFECYCLE_CHILD") == "1" { + testServiceSentinelWorkerIgnoresStaleReportAfterDeletionChild(t) + return + } + + // Given + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + child := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestServiceSentinelWorkerIgnoresStaleReportAfterDeletion$") + child.Env = append(os.Environ(), "NEZHA_SERVICE_SENTINEL_LIFECYCLE_CHILD=1") + + // When + output, err := child.CombinedOutput() + + // Then + if ctx.Err() != nil { + t.Fatalf("service sentinel lifecycle child timed out: %v\n%s", ctx.Err(), output) + } + if err != nil { + t.Fatalf("service sentinel lifecycle child failed: %v\n%s", err, output) + } + if !strings.Contains(string(output), serviceSentinelLifecycleSuccessMarker) { + t.Fatalf("service sentinel lifecycle child did not report success:\n%s", output) + } +} + +func testServiceSentinelWorkerIgnoresStaleReportAfterDeletionChild(t *testing.T) { + // Given + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + ss := newServiceMonitorSecurityHarness(t, + &model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"}, + ) + for _, service := range []*model.Service{ + { + Common: model.Common{ID: 10, UserID: 1}, + Name: "stale-service", + Type: model.TaskTypeTCPPing, + Target: "stale.example.invalid:443", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + }, + { + Common: model.Common{ID: 20, UserID: 1}, + Name: "valid-service", + Type: model.TaskTypeTCPPing, + Target: "valid.example.invalid:443", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + }, + } { + addServiceMonitorSecurityService(t, ss, service) + } + acceptedStaleReport := make(chan struct{}) + releaseWorker := make(chan struct{}) + var releaseOnce sync.Once + releaseWorkerHook := func() { + releaseOnce.Do(func() { close(releaseWorker) }) + } + ss.serviceReportValidatedHook = func(serviceID uint64) { + if serviceID == 10 { + close(acceptedStaleReport) + <-releaseWorker + } + } + t.Cleanup(func() { + releaseWorkerHook() + ss.Close() + }) + + // When + ss.Dispatch(serviceMonitorResult(1, 10, model.TaskTypeTCPPing, true)) + select { + case <-acceptedStaleReport: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + ss.Delete([]uint64{10}) + releaseWorkerHook() + ss.Dispatch(serviceMonitorResult(1, 20, model.TaskTypeTCPPing, true)) + ss.Close() + + // Then + var staleHistoryCount int64 + if err := DB.Model(&model.ServiceHistory{}). + Where("service_id = ? AND server_id = ?", 10, 1). + Count(&staleHistoryCount).Error; err != nil { + t.Fatal(err) + } + if staleHistoryCount != 0 { + t.Fatalf("expected stale service to write zero per-reporter history rows, got %d", staleHistoryCount) + } + var validHistoryCount int64 + if err := DB.Model(&model.ServiceHistory{}). + Where("service_id = ? AND server_id = ?", 20, 1). + Count(&validHistoryCount).Error; err != nil { + t.Fatal(err) + } + if validHistoryCount != 1 { + t.Fatalf("expected exactly one valid service history row, got %d", validHistoryCount) + } + ss.serviceResponseDataStoreLock.RLock() + _, stalePingCached := ss.serviceResponsePing[10] + validStats := ss.serviceStatusToday[20] + ss.serviceResponseDataStoreLock.RUnlock() + if stalePingCached { + t.Fatal("expected stale service ping cache to be deleted") + } + if validStats == nil || validStats.Up != 1 || validStats.Down != 0 { + t.Fatalf("expected valid service stats up=1 down=0, got %+v", validStats) + } + if _, err := fmt.Fprintln(os.Stdout, serviceSentinelLifecycleSuccessMarker); err != nil { + t.Fatal(err) + } +} + +func TestServiceSentinelWorkerRevalidatesReportAfterUpdate(t *testing.T) { + // Given + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + ss := newServiceMonitorSecurityHarness(t, + &model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"}, + ) + service := &model.Service{ + Common: model.Common{ID: 10, UserID: 1}, + Name: "updatable-service", + Type: model.TaskTypeTCPPing, + Target: "updatable.example.invalid:443", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + } + addServiceMonitorSecurityService(t, ss, service) + ss.serviceResponseDataStoreLock.Lock() + ss.serviceStatusToday[service.ID] = &_TodayStatsOfService{Up: 7, Down: 3, Delay: 12.5} + ss.serviceResponseDataStoreLock.Unlock() + acceptedReport := make(chan struct{}) + releaseWorker := make(chan struct{}) + var releaseOnce sync.Once + releaseWorkerHook := func() { + releaseOnce.Do(func() { close(releaseWorker) }) + } + ss.serviceReportValidatedHook = func(serviceID uint64) { + if serviceID == service.ID { + close(acceptedReport) + <-releaseWorker + } + } + t.Cleanup(func() { + releaseWorkerHook() + ss.Close() + }) + + // When + ss.Dispatch(serviceMonitorResult(1, service.ID, model.TaskTypeTCPPing, true)) + select { + case <-acceptedReport: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + updatedService := *service + updatedService.Name = "updated-service" + updatedService.SkipServers = map[uint64]bool{} + if err := ss.Update(&updatedService); err != nil { + t.Fatal(err) + } + releaseWorkerHook() + ss.Close() + + // Then + var historyCount int64 + if err := DB.Model(&model.ServiceHistory{}). + Where("service_id = ? AND server_id = ?", service.ID, 1). + Count(&historyCount).Error; err != nil { + t.Fatal(err) + } + if historyCount != 0 { + t.Fatalf("expected updated service to write zero per-reporter history rows, got %d", historyCount) + } + ss.serviceResponseDataStoreLock.RLock() + _, pingCached := ss.serviceResponsePing[service.ID] + stats := ss.serviceStatusToday[service.ID] + ss.serviceResponseDataStoreLock.RUnlock() + if pingCached { + t.Fatal("expected updated service report to leave no ping cache entry") + } + if stats == nil || stats.Up != 7 || stats.Down != 3 || stats.Delay != 12.5 { + t.Fatalf("expected existing service stats to remain unchanged, got %+v", stats) + } + currentService, ok := ss.Get(service.ID) + if !ok || currentService.Name != updatedService.Name || currentService.SkipServers[1] { + t.Fatalf("expected updated service configuration, got %+v", currentService) + } +} + +func TestServiceSentinelLoadStatsFollowsLifecycleLockOrder(t *testing.T) { + // Given + ss := &ServiceSentinel{ + serviceStatusToday: make(map[uint64]*_TodayStatsOfService), + serviceResponseDataStore: make(map[uint64]serviceResponseData), + services: make(map[uint64]*model.Service), + monthlyStatus: make(map[uint64]*serviceResponseItem), + } + ss.loadStatsResponseLockedHook = func() { + if ss.serviceResponseDataStoreLock.TryLock() { + ss.serviceResponseDataStoreLock.Unlock() + t.Fatal("LoadStats invoked the hook before acquiring the response read lock") + } + if !ss.monthlyStatusLock.TryLock() { + t.Fatal("LoadStats acquired monthlyStatusLock before the response lock hook") + } + ss.monthlyStatusLock.Unlock() + if !ss.servicesLock.TryLock() { + t.Fatal("LoadStats acquired servicesLock before the response lock hook") + } + ss.servicesLock.Unlock() + } + + // When / Then + ss.LoadStats() +} + +func TestServiceSentinelWorkerHoldsResponseLockDuringTLSSideEffects(t *testing.T) { + // Given + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + ss := newServiceMonitorSecurityHarness(t, + &model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"}, + ) + service := &model.Service{ + Common: model.Common{ID: 10, UserID: 1}, + Name: "tls-service", + Type: model.TaskTypeHTTPGet, + Target: "https://tls.example.invalid", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + } + addServiceMonitorSecurityService(t, ss, service) + tlsSideEffectsReady := make(chan struct{}) + releaseTLSSideEffects := make(chan struct{}) + var releaseOnce sync.Once + releaseTLSSideEffectsHook := func() { + releaseOnce.Do(func() { close(releaseTLSSideEffects) }) + } + ss.serviceReportBeforeTLSSideEffectsHook = func(serviceID uint64) { + if serviceID == service.ID { + close(tlsSideEffectsReady) + <-releaseTLSSideEffects + } + } + t.Cleanup(func() { + releaseTLSSideEffectsHook() + ss.Close() + }) + report := serviceMonitorResult(1, service.ID, model.TaskTypeHTTPGet, true) + report.Data.Data = "issuer|2030-01-02 15:04:05 +0000 UTC" + + // When + ss.Dispatch(report) + select { + case <-tlsSideEffectsReady: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + responseLockAcquired := ss.serviceResponseDataStoreLock.TryLock() + if responseLockAcquired { + ss.serviceResponseDataStoreLock.Unlock() + t.Fatal("worker released the response lock before TLS side effects") + } + releaseTLSSideEffectsHook() + ss.Close() + + // Then + ss.serviceResponseDataStoreLock.RLock() + cachedCertificate := ss.tlsCertCache[service.ID] + ss.serviceResponseDataStoreLock.RUnlock() + if cachedCertificate != report.Data.Data { + t.Fatalf("expected TLS cache %q, got %q", report.Data.Data, cachedCertificate) + } +}