fix(monitor):修复监控数据未及时更新的问题

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()
 }
 

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(),
 	}
 }
 

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)
+	}
+}