diff --git a/internal/logic/monitor/collector.go b/internal/logic/monitor/collector.go index 15b553c..3b460d5 100644 --- a/internal/logic/monitor/collector.go +++ b/internal/logic/monitor/collector.go @@ -417,7 +417,11 @@ func writeMetricsToDB(s SystemMetricsSnapshot) { } // GetLatestMetrics returns the most recent metrics snapshot. +// 采集器未初始化时返回 nil（而非 panic）：调用方本就按 nil 处理缺失指标。 func GetLatestMetrics() *SystemMetricsSnapshot { + if systemBuffer == nil { + return nil + } return systemBuffer.Latest() } diff --git a/internal/logic/monitor/request_tracker.go b/internal/logic/monitor/request_tracker.go index 512bc91..2169965 100644 --- a/internal/logic/monitor/request_tracker.go +++ b/internal/logic/monitor/request_tracker.go @@ -4,11 +4,23 @@ import ( "sync" "sync/atomic" "time" + + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gctx" ) const ( maxActiveRequestsInResponse = 200 - historyCapacity = 100 // ~5 min at 3s polling + + // 采样周期与保留窗口：后台采样器固定 5s 采一次，环形缓冲保留 120 个点 = 10 分钟。 + samplerInterval = 5 * time.Second + historyCapacity = 120 + historyWindow = time.Duration(historyCapacity) * samplerInterval + + // 悬挂条目清扫阈值：普通 relay 请求受 HTTP 超时约束，异步任务由轮询器 30min 超时兜底。 + // 超过该时长仍留在 map 中的一定是漏调 Unregister（如任务被另一实例的轮询器终结）。 + staleRequestTTL = 30 * time.Minute + staleTaskTTL = 60 * time.Minute ) // TrackedRequest represents a single active relay request. @@ -99,6 +111,29 @@ func (r *snapshotRing[T]) All() []T { return result } +// Last 返回最新一个样本（无样本时 ok=false）。 +func (r *snapshotRing[T]) Last() (T, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + var zero T + if r.size == 0 { + return zero, false + } + return r.items[(r.head-1+r.cap)%r.cap], true +} + +// filterRecent 裁掉早于 cutoff 的样本。环形缓冲按时间升序返回，命中首个在窗口内的点即可整体切片。 +// 采样器正常运行时不会有过期点，此处是采样器停摆（如进程挂起）后的兜底， +// 避免把陈旧曲线当成当前流量画出来。 +func filterRecent[T any](items []T, cutoff time.Time, at func(T) time.Time) []T { + for i := range items { + if !at(items[i]).Before(cutoff) { + return items[i:] + } + } + return nil +} + // requestTracker is the global singleton for tracking active requests. type requestTracker struct { mu sync.RWMutex @@ -118,15 +153,99 @@ type requestTracker struct { bwHistory *snapshotRing[BandwidthSnapshot] } -var tracker *requestTracker +var ( + tracker *requestTracker + trackerStop chan struct{} +) -// InitRequestTracker initializes the global request tracker. +// InitRequestTracker initializes the global request tracker and starts its sampler. +// +// 采样必须由后台定时器驱动、与管理页轮询解耦：早期实现只在 /monitor/realtime 被调用时 +// 才往环形缓冲写点，页面关闭期间既不采样也不过期，再次打开时读到的仍是上一次会话遗留的 +// 旧曲线（前端 x 轴只显示 mm:ss，看起来就像"刚刚还有并发/带宽"）。 func InitRequestTracker() { - tracker = &requestTracker{ + StopRequestTracker() + + t := &requestTracker{ activeRequests: make(map[string]*TrackedRequest), concHistory: newSnapshotRing[ConcurrencySnapshot](historyCapacity), bwHistory: newSnapshotRing[BandwidthSnapshot](historyCapacity), } + // 带宽基线置为启动时刻，避免首个样本把「进程启动至今累计字节」算成瞬时速率 + t.lastBwTime = time.Now() + + tracker = t + trackerStop = make(chan struct{}) + go t.runSampler(trackerStop) +} + +// StopRequestTracker stops the background sampler (idempotent). +func StopRequestTracker() { + if trackerStop != nil { + close(trackerStop) + trackerStop = nil + } +} + +func (t *requestTracker) runSampler(stop chan struct{}) { + ticker := time.NewTicker(samplerInterval) + defer ticker.Stop() + for { + select { + case <-stop: + return + case now := <-ticker.C: + t.sample(now) + } + } +} + +// sample 采一次并发 + 带宽快照。带宽差值只在这里计算（固定 5s 间隔）， +// 不再由 HTTP 轮询触发——否则多个管理页同时打开会互相偷走增量， +// 两次轮询间隔极短时还会算出「字节数 / 几毫秒」的假尖峰。 +func (t *requestTracker) sample(now time.Time) { + total, streaming := t.sweepStale(now) + t.concHistory.Push(ConcurrencySnapshot{ + Timestamp: now, + TotalActive: total, + StreamingActive: streaming, + NonStreamingActive: total - streaming, + }) + t.bwHistory.Push(t.snapshotBandwidth(now)) +} + +// sweepStale 清扫超时仍未反注册的悬挂条目，并以 map 为唯一真相重算计数。 +// 漏减场景确实存在：异步任务由轮询器终结，多实例部署下终结它的实例未必是当初受理提交的实例， +// 受理实例内存里的条目就再也没人删——表现为并发数永远回不到 0。 +func (t *requestTracker) sweepStale(now time.Time) (total, streaming int) { + var swept int + + t.mu.Lock() + for key, req := range t.activeRequests { + ttl := staleRequestTTL + if req.IsAsyncTask { + ttl = staleTaskTTL + } + if now.Sub(req.StartTime) > ttl { + delete(t.activeRequests, key) + swept++ + continue + } + total++ + if req.IsStream { + streaming++ + } + } + t.mu.Unlock() + + // 计数以 map 为准回写，顺带修正任何漏减/多减造成的漂移 + t.totalActive.Store(int64(total)) + t.streamingActive.Store(int64(streaming)) + + if swept > 0 { + g.Log().Warningf(gctx.New(), "monitor: 清扫 %d 条悬挂的活跃请求记录（注册后未反注册）", swept) + } + return total, streaming } // RegisterRequest adds a new request to the tracker. @@ -191,9 +310,12 @@ func UnregisterRequestByTaskID(taskID string) { } tracker.mu.Lock() defer tracker.mu.Unlock() - if _, ok := tracker.activeRequests[taskID]; ok { + if req, ok := tracker.activeRequests[taskID]; ok { delete(tracker.activeRequests, taskID) tracker.totalActive.Add(-1) + if req.IsStream { + tracker.streamingActive.Add(-1) + } } } @@ -214,14 +336,17 @@ func RecordBytesOut(n int) { } // GetRealtimeData builds the combined realtime snapshot. +// 纯读：历史点与带宽速率均由后台采样器产出，此处不再推进任何采样状态， +// 因此多个管理页同时打开互不干扰，页面不打开时曲线也照常滚动过期。 func GetRealtimeData() *RealtimeData { if tracker == nil { return &RealtimeData{} } now := time.Now() + cutoff := now.Add(-historyWindow) - // Concurrency snapshot + // 当前并发：取实时计数（比最近一个样本更新） total := int(tracker.totalActive.Load()) streaming := int(tracker.streamingActive.Load()) concSnap := ConcurrencySnapshot{ @@ -230,11 +355,12 @@ func GetRealtimeData() *RealtimeData { StreamingActive: streaming, NonStreamingActive: total - streaming, } - tracker.concHistory.Push(concSnap) - // Bandwidth snapshot (delta between calls) - bwSnap := tracker.snapshotBandwidth(now) - tracker.bwHistory.Push(bwSnap) + // 当前带宽：取最近一个样本；样本过旧（采样器停摆）则视为 0，不展示陈旧速率 + bwSnap := BandwidthSnapshot{Timestamp: now} + if last, ok := tracker.bwHistory.Last(); ok && !last.Timestamp.Before(now.Add(-2*samplerInterval)) { + bwSnap = last + } // Build breakdowns tracker.mu.RLock() @@ -260,17 +386,19 @@ func GetRealtimeData() *RealtimeData { } return &RealtimeData{ - Concurrency: concSnap, - Bandwidth: bwSnap, - Runtime: rt, - GoRuntime: GetGoRuntimeInfo(), - History: tracker.concHistory.All(), - BandwidthHistory: tracker.bwHistory.All(), - ActiveRequests: activeReqs, - ByModel: byModel, - ByChannel: byChannel, - ByTenant: byTenant, - SyncImagePool: GetSyncImagePoolSnapshot(), + Concurrency: concSnap, + Bandwidth: bwSnap, + Runtime: rt, + GoRuntime: GetGoRuntimeInfo(), + History: filterRecent(tracker.concHistory.All(), cutoff, + func(s ConcurrencySnapshot) time.Time { return s.Timestamp }), + BandwidthHistory: filterRecent(tracker.bwHistory.All(), cutoff, + func(s BandwidthSnapshot) time.Time { return s.Timestamp }), + ActiveRequests: activeReqs, + ByModel: byModel, + ByChannel: byChannel, + ByTenant: byTenant, + SyncImagePool: GetSyncImagePoolSnapshot(), } } diff --git a/internal/logic/monitor/request_tracker_test.go b/internal/logic/monitor/request_tracker_test.go index cdaa797..201344e 100644 --- a/internal/logic/monitor/request_tracker_test.go +++ b/internal/logic/monitor/request_tracker_test.go @@ -29,3 +29,60 @@ func TestSnapshotBandwidthConcurrent(t *testing.T) { t.Fatalf("bandwidth rates must not be negative: %+v", snapshot) } } + +// 读接口不得推进采样：历史点只能来自后台采样器， +// 否则页面关闭期间不采样、旧曲线不过期，再打开时会把上次会话的并发/带宽当成"刚刚"画出来。 +func TestGetRealtimeDataDoesNotSample(t *testing.T) { + InitRequestTracker() + defer StopRequestTracker() + + for range 3 { + if data := GetRealtimeData(); len(data.History) != 0 || len(data.BandwidthHistory) != 0 { + t.Fatalf("读接口不应写入历史点: conc=%d bw=%d", len(data.History), len(data.BandwidthHistory)) + } + } + + tracker.sample(time.Now()) + data := GetRealtimeData() + if len(data.History) != 1 || len(data.BandwidthHistory) != 1 { + t.Fatalf("采样器应产出 1 个点: conc=%d bw=%d", len(data.History), len(data.BandwidthHistory)) + } +} + +// 超时未反注册的悬挂条目必须被清扫，且计数以 map 为准重算（修正漏减漂移）。 +func TestSweepStaleDropsHangingRequests(t *testing.T) { + tr := &requestTracker{activeRequests: make(map[string]*TrackedRequest)} + now := time.Now() + + tr.activeRequests["fresh"] = &TrackedRequest{RequestID: "fresh", StartTime: now.Add(-time.Minute), IsStream: true} + tr.activeRequests["hung-req"] = &TrackedRequest{RequestID: "hung-req", StartTime: now.Add(-staleRequestTTL - time.Minute)} + tr.activeRequests["hung-task"] = &TrackedRequest{RequestID: "hung-task", StartTime: now.Add(-staleTaskTTL - time.Minute), IsAsyncTask: true} + // 未到异步任务 TTL 的长任务应保留 + tr.activeRequests["live-task"] = &TrackedRequest{RequestID: "live-task", StartTime: now.Add(-staleRequestTTL - time.Minute), IsAsyncTask: true} + // 计数故意置成漂移值，验证重算能自愈 + tr.totalActive.Store(99) + + total, streaming := tr.sweepStale(now) + if total != 2 || streaming != 1 { + t.Fatalf("清扫后应剩 2 条（1 条流式），实际 total=%d streaming=%d", total, streaming) + } + if got := tr.totalActive.Load(); got != 2 { + t.Fatalf("计数应按 map 重算为 2，实际 %d", got) + } + if _, ok := tr.activeRequests["live-task"]; !ok { + t.Fatal("未超时的异步任务被误清扫") + } +} + +func TestFilterRecentDropsStalePoints(t *testing.T) { + now := time.Now() + items := []ConcurrencySnapshot{ + {Timestamp: now.Add(-2 * historyWindow), TotalActive: 7}, // 上次会话遗留 + {Timestamp: now.Add(-time.Minute), TotalActive: 1}, + {Timestamp: now, TotalActive: 2}, + } + got := filterRecent(items, now.Add(-historyWindow), func(s ConcurrencySnapshot) time.Time { return s.Timestamp }) + if len(got) != 2 || got[0].TotalActive != 1 { + t.Fatalf("应只保留窗口内的 2 个点，实际 %+v", got) + } +}