refactor: decouple public monitor sparklines

internal/app.go

@@ -162,6 +162,7 @@ func setupApi(app *orz.App, components *AppComponents) error {
 
 		// 监控统计数据（公开访问，支持可选认证）- 用于公共展示页面
 		publicApiWithOptionalAuth.GET("/monitors", components.MonitorHandler.GetMonitors)
+		publicApiWithOptionalAuth.GET("/monitors/sparklines", components.MonitorHandler.GetSparklines)
 		publicApiWithOptionalAuth.GET("/monitors/:id/stats", components.MonitorHandler.GetStatsByID)
 		publicApiWithOptionalAuth.GET("/monitors/:id/agents", components.MonitorHandler.GetAgentStatsByID)
 		publicApiWithOptionalAuth.GET("/monitors/:id/history", components.MonitorHandler.GetHistoryByID)

internal/handler/monitor_handler.go

@@ -150,6 +150,17 @@ func (h *MonitorHandler) GetMonitors(c *echo.Context) error {
 	return orz.Ok(c, stats)
 }
 
+// GetSparklines 获取公开服务列表使用的最近一小时响应时间趋势。
+func (h *MonitorHandler) GetSparklines(c *echo.Context) error {
+	ctx := c.Request().Context()
+	result, err := h.monitorService.GetMonitorSparklinesByAuth(ctx, utils.IsAuthenticated(c))
+	if err != nil {
+		return err
+	}
+
+	return orz.Ok(c, result)
+}
+
 // GetStatsByID 获取指定监控任务的统计数据（公开接口，已登录返回全部，未登录返回公开可见）
 func (h *MonitorHandler) GetStatsByID(c *echo.Context) error {
 	id := c.Param("id")

internal/metric/monitor.go

@@ -36,6 +36,12 @@ type MonitorSparklinePoint struct {
 	Max       float64 `json:"max"`
 }
 
+// PublicMonitorSparklinesResponse 是公开服务列表的独立趋势响应。
+type PublicMonitorSparklinesResponse struct {
+	GeneratedAt int64                              `json:"generatedAt"`
+	Items       map[string][]MonitorSparklinePoint `json:"items"`
+}
+
 // PublicMonitorOverview 用于公开展示的监控配置及汇总数据
 type PublicMonitorOverview struct {
 	ID               string `json:"id"`
@@ -58,8 +64,7 @@ type PublicMonitorOverview struct {
 		Down    int `json:"down"`    // 异常探针数量
 		Unknown int `json:"unknown"` // 未知状态探针数量
 	} `json:"agentStats"` // 探针状态分布
-	LastCheckTime int64                   `json:"lastCheckTime"` // 最后检测时间
-	Sparkline     []MonitorSparklinePoint `json:"sparkline,omitempty"`
+	LastCheckTime int64 `json:"lastCheckTime"` // 最后检测时间
 }
 
 // MonitorDetailResponse 监控详情响应（整合版）

internal/service/metric_service.go

@@ -10,6 +10,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/go-orz/toolkit/syncx"
@@ -35,19 +36,25 @@ type MetricService struct {
 	latestCache cache.Cache[string, *metric.LatestMetrics] // Agent 最新指标缓存
 
 	monitorLatestCache cache.Cache[string, *metric.LatestMonitorMetrics] // 监控最新指标缓存
+
+	monitorSparklineMu       sync.Mutex
+	monitorSparklineCache    map[string]monitorSparklineCacheEntry
+	monitorSparklineInflight map[string]*monitorSparklineCall
 }
 
 // NewMetricService 创建指标服务
 func NewMetricService(logger *zap.Logger, db *gorm.DB, propertyService *PropertyService, trafficService *TrafficService, vmClient *vmclient.VMClient) *MetricService {
 	return &MetricService{
-		logger:             logger,
-		agentRepo:          repo.NewAgentRepo(db),
-		monitorRepo:        repo.NewMonitorRepo(db),
-		propertyService:    propertyService,
-		trafficService:     trafficService,
-		vmClient:           vmClient,
-		latestCache:        cache.New[string, *metric.LatestMetrics](time.Minute),
-		monitorLatestCache: cache.New[string, *metric.LatestMonitorMetrics](5 * time.Minute), // 监控数据缓存 5 分钟
+		logger:                   logger,
+		agentRepo:                repo.NewAgentRepo(db),
+		monitorRepo:              repo.NewMonitorRepo(db),
+		propertyService:          propertyService,
+		trafficService:           trafficService,
+		vmClient:                 vmClient,
+		latestCache:              cache.New[string, *metric.LatestMetrics](time.Minute),
+		monitorLatestCache:       cache.New[string, *metric.LatestMonitorMetrics](5 * time.Minute), // 监控数据缓存 5 分钟
+		monitorSparklineCache:    make(map[string]monitorSparklineCacheEntry),
+		monitorSparklineInflight: make(map[string]*monitorSparklineCall),
 	}
 }
 
@@ -898,6 +905,111 @@ func aggregateMonitorSparklines(result *vmclient.QueryResult, allowedMonitorIDs
 	return resultByMonitor
 }
 
+type monitorSparklineCacheEntry struct {
+	generatedAt int64
+	expiresAt   time.Time
+	items       map[string][]metric.MonitorSparklinePoint
+}
+
+type monitorSparklineCall struct {
+	done        chan struct{}
+	generatedAt int64
+	items       map[string][]metric.MonitorSparklinePoint
+	err         error
+}
+
+func normalizeMonitorSparklineIDs(monitorIDs []string) []string {
+	unique := make(map[string]struct{}, len(monitorIDs))
+	for _, monitorID := range monitorIDs {
+		if monitorID != "" {
+			unique[monitorID] = struct{}{}
+		}
+	}
+
+	normalized := make([]string, 0, len(unique))
+	for monitorID := range unique {
+		normalized = append(normalized, monitorID)
+	}
+	sort.Strings(normalized)
+	return normalized
+}
+
+func cloneMonitorSparklines(items map[string][]metric.MonitorSparklinePoint) map[string][]metric.MonitorSparklinePoint {
+	cloned := make(map[string][]metric.MonitorSparklinePoint, len(items))
+	for monitorID, points := range items {
+		cloned[monitorID] = append([]metric.MonitorSparklinePoint(nil), points...)
+	}
+	return cloned
+}
+
+// GetCachedMonitorSparklines 返回带短期缓存的批量走势图，并合并同一查询的并发请求。
+func (s *MetricService) GetCachedMonitorSparklines(ctx context.Context, monitorIDs []string, window, ttl time.Duration) (int64, map[string][]metric.MonitorSparklinePoint, error) {
+	normalizedIDs := normalizeMonitorSparklineIDs(monitorIDs)
+	if len(normalizedIDs) == 0 {
+		return time.Now().UnixMilli(), map[string][]metric.MonitorSparklinePoint{}, nil
+	}
+
+	key := strconv.FormatInt(window.Milliseconds(), 10) + ":" + strings.Join(normalizedIDs, "\x00")
+	now := time.Now()
+
+	s.monitorSparklineMu.Lock()
+	if s.monitorSparklineCache == nil {
+		s.monitorSparklineCache = make(map[string]monitorSparklineCacheEntry)
+	}
+	if s.monitorSparklineInflight == nil {
+		s.monitorSparklineInflight = make(map[string]*monitorSparklineCall)
+	}
+	if cached, ok := s.monitorSparklineCache[key]; ok {
+		if now.Before(cached.expiresAt) {
+			s.monitorSparklineMu.Unlock()
+			return cached.generatedAt, cloneMonitorSparklines(cached.items), nil
+		}
+		delete(s.monitorSparklineCache, key)
+	}
+	if call, ok := s.monitorSparklineInflight[key]; ok {
+		s.monitorSparklineMu.Unlock()
+		select {
+		case <-call.done:
+			return call.generatedAt, cloneMonitorSparklines(call.items), call.err
+		case <-ctx.Done():
+			return 0, nil, ctx.Err()
+		}
+	}
+
+	call := &monitorSparklineCall{done: make(chan struct{})}
+	s.monitorSparklineInflight[key] = call
+	s.monitorSparklineMu.Unlock()
+
+	end := time.Now().UnixMilli()
+	items, err := s.GetMonitorSparklines(ctx, normalizedIDs, end-window.Milliseconds(), end)
+	if err == nil && items == nil {
+		items = map[string][]metric.MonitorSparklinePoint{}
+	}
+
+	s.monitorSparklineMu.Lock()
+	call.generatedAt = end
+	call.items = items
+	call.err = err
+	if err == nil && ttl > 0 {
+		cacheNow := time.Now()
+		for cacheKey, cached := range s.monitorSparklineCache {
+			if !cacheNow.Before(cached.expiresAt) {
+				delete(s.monitorSparklineCache, cacheKey)
+			}
+		}
+		s.monitorSparklineCache[key] = monitorSparklineCacheEntry{
+			generatedAt: end,
+			expiresAt:   cacheNow.Add(ttl),
+			items:       cloneMonitorSparklines(items),
+		}
+	}
+	delete(s.monitorSparklineInflight, key)
+	close(call.done)
+	s.monitorSparklineMu.Unlock()
+
+	return end, cloneMonitorSparklines(items), err
+}
+
 // GetMonitorSparklines 使用一次 VictoriaMetrics 查询返回多个监控项的列表走势图。
 func (s *MetricService) GetMonitorSparklines(ctx context.Context, monitorIDs []string, start, end int64) (map[string][]metric.MonitorSparklinePoint, error) {
 	if len(monitorIDs) == 0 {

internal/service/metric_service_sparkline_test.go

@@ -1,7 +1,15 @@
 package service
 
 import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/pika-monitor/pika/internal/vmclient"
 )
@@ -63,3 +71,100 @@ func TestAggregateMonitorSparklinesSkipsMalformedValues(t *testing.T) {
 		t.Fatalf("unexpected valid point: %+v", points[0])
 	}
 }
+
+func TestGetCachedMonitorSparklinesCoalescesAndCachesNormalizedIDs(t *testing.T) {
+	var calls atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls.Add(1)
+		query := r.URL.Query().Get("query")
+		if !strings.Contains(query, "monitor-a") || !strings.Contains(query, "monitor-b") {
+			t.Errorf("query does not contain both monitor IDs: %s", query)
+		}
+		time.Sleep(40 * time.Millisecond)
+		if err := json.NewEncoder(w).Encode(vmclient.QueryResult{
+			Status: "success",
+			Data: vmclient.ResultData{ResultType: "matrix", Result: []vmclient.Result{
+				{
+					Metric: map[string]string{"monitor_id": "monitor-a", "agent_id": "agent-1"},
+					Values: [][]interface{}{{float64(100), "125"}},
+				},
+			}},
+		}); err != nil {
+			t.Errorf("encode response: %v", err)
+		}
+	}))
+	defer server.Close()
+
+	service := &MetricService{vmClient: vmclient.NewVMClient(server.URL, time.Second, time.Second)}
+	monitorIDSets := [][]string{
+		{"monitor-b", "monitor-a", "monitor-a", ""},
+		{"monitor-a", "monitor-b"},
+	}
+
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	errs := make(chan error, len(monitorIDSets))
+	for _, monitorIDs := range monitorIDSets {
+		wg.Add(1)
+		go func(ids []string) {
+			defer wg.Done()
+			<-start
+			_, items, err := service.GetCachedMonitorSparklines(context.Background(), ids, time.Hour, 20*time.Second)
+			if err == nil && len(items["monitor-a"]) != 1 {
+				t.Errorf("expected one point for monitor-a, got %+v", items)
+			}
+			errs <- err
+		}(monitorIDs)
+	}
+	close(start)
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		if err != nil {
+			t.Fatalf("GetCachedMonitorSparklines returned error: %v", err)
+		}
+	}
+
+	if got := calls.Load(); got != 1 {
+		t.Fatalf("expected one coalesced VM query, got %d", got)
+	}
+
+	_, cached, err := service.GetCachedMonitorSparklines(context.Background(), []string{"monitor-b", "monitor-a"}, time.Hour, 20*time.Second)
+	if err != nil {
+		t.Fatalf("cached query returned error: %v", err)
+	}
+	cached["monitor-a"][0].Avg = 999
+	_, cloned, err := service.GetCachedMonitorSparklines(context.Background(), []string{"monitor-a", "monitor-b"}, time.Hour, 20*time.Second)
+	if err != nil {
+		t.Fatalf("second cached query returned error: %v", err)
+	}
+	if cloned["monitor-a"][0].Avg != 125 {
+		t.Fatalf("cached data was mutated through caller response: %+v", cloned["monitor-a"][0])
+	}
+	if got := calls.Load(); got != 1 {
+		t.Fatalf("expected cache hits to avoid extra VM queries, got %d", got)
+	}
+}
+
+func TestGetCachedMonitorSparklinesDoesNotCacheFailures(t *testing.T) {
+	var calls atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		if calls.Add(1) == 1 {
+			http.Error(w, "temporary failure", http.StatusServiceUnavailable)
+			return
+		}
+		_ = json.NewEncoder(w).Encode(vmclient.QueryResult{Status: "success"})
+	}))
+	defer server.Close()
+
+	service := &MetricService{vmClient: vmclient.NewVMClient(server.URL, time.Second, time.Second)}
+	if _, _, err := service.GetCachedMonitorSparklines(context.Background(), []string{"monitor-a"}, time.Hour, 20*time.Second); err == nil {
+		t.Fatal("expected the first VM query to fail")
+	}
+	if _, _, err := service.GetCachedMonitorSparklines(context.Background(), []string{"monitor-a"}, time.Hour, 20*time.Second); err != nil {
+		t.Fatalf("expected retry after failure to succeed: %v", err)
+	}
+	if got := calls.Load(); got != 2 {
+		t.Fatalf("expected failed query not to be cached, got %d calls", got)
+	}
+}

internal/service/monitor_service.go

@@ -229,22 +229,29 @@ func (s *MonitorService) ListByAuth(ctx context.Context, isAuthenticated bool) (
 		items = append(items, item)
 	}
 
-	monitorIDs := make([]string, 0, len(items))
-	for _, item := range items {
-		monitorIDs = append(monitorIDs, item.ID)
-	}
-	end := time.Now().UnixMilli()
-	sparklines, err := s.metricService.GetMonitorSparklines(ctx, monitorIDs, end-time.Hour.Milliseconds(), end)
+	return items, nil
+}
+
+// GetMonitorSparklinesByAuth 返回当前访问者可见监控项的批量走势图。
+func (s *MonitorService) GetMonitorSparklinesByAuth(ctx context.Context, isAuthenticated bool) (*metric.PublicMonitorSparklinesResponse, error) {
+	monitors, err := s.FindByAuth(ctx, isAuthenticated)
 	if err != nil {
-		// 趋势图是列表的增强信息，VictoriaMetrics 暂时不可用时仍返回当前状态。
-		s.logger.Warn("查询公开监控列表趋势失败", zap.Error(err))
-	} else {
-		for i := range items {
-			items[i].Sparkline = sparklines[items[i].ID]
-		}
+		return nil, err
 	}
 
-	return items, nil
+	monitorIDs := make([]string, 0, len(monitors))
+	for _, monitor := range monitors {
+		monitorIDs = append(monitorIDs, monitor.ID)
+	}
+
+	queryCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+	generatedAt, items, err := s.metricService.GetCachedMonitorSparklines(queryCtx, monitorIDs, time.Hour, 20*time.Second)
+	if err != nil {
+		return nil, err
+	}
+
+	return &metric.PublicMonitorSparklinesResponse{GeneratedAt: generatedAt, Items: items}, nil
 }
 
 // buildMonitorOverview 构建监控概览对象

internal/service/monitor_service_sparkline_test.go

@@ -0,0 +1,100 @@
+package service
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/glebarez/sqlite"
+	"github.com/pika-monitor/pika/internal/models"
+	"github.com/pika-monitor/pika/internal/vmclient"
+	"go.uber.org/zap"
+	"gorm.io/gorm"
+)
+
+func TestGetMonitorSparklinesByAuthExcludesPrivateMonitors(t *testing.T) {
+	database, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatalf("open database: %v", err)
+	}
+	if err := database.AutoMigrate(&models.MonitorTask{}); err != nil {
+		t.Fatalf("migrate monitor table: %v", err)
+	}
+	for _, monitor := range []models.MonitorTask{
+		{ID: "public-monitor", Name: "Public", Enabled: true, Visibility: "public"},
+		{ID: "private-monitor", Name: "Private", Enabled: true, Visibility: "private"},
+	} {
+		if err := database.Create(&monitor).Error; err != nil {
+			t.Fatalf("create monitor %s: %v", monitor.ID, err)
+		}
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		query := r.URL.Query().Get("query")
+		if !strings.Contains(query, "public-monitor") {
+			t.Errorf("query does not contain public monitor: %s", query)
+		}
+		if strings.Contains(query, "private-monitor") {
+			t.Errorf("anonymous query contains private monitor: %s", query)
+		}
+		_ = json.NewEncoder(w).Encode(vmclient.QueryResult{
+			Status: "success",
+			Data: vmclient.ResultData{ResultType: "matrix", Result: []vmclient.Result{
+				{
+					Metric: map[string]string{"monitor_id": "public-monitor"},
+					Values: [][]interface{}{{float64(100), "100"}},
+				},
+				{
+					Metric: map[string]string{"monitor_id": "private-monitor"},
+					Values: [][]interface{}{{float64(100), "999"}},
+				},
+			}},
+		})
+	}))
+	defer server.Close()
+
+	metricService := &MetricService{vmClient: vmclient.NewVMClient(server.URL, time.Second, time.Second)}
+	monitorService := NewMonitorService(zap.NewNop(), database, metricService, nil)
+	result, err := monitorService.GetMonitorSparklinesByAuth(t.Context(), false)
+	if err != nil {
+		t.Fatalf("GetMonitorSparklinesByAuth returned error: %v", err)
+	}
+	if len(result.Items["public-monitor"]) != 1 {
+		t.Fatalf("expected public monitor sparkline, got %+v", result.Items)
+	}
+	if _, ok := result.Items["private-monitor"]; ok {
+		t.Fatalf("private monitor leaked into anonymous response: %+v", result.Items)
+	}
+}
+
+func TestListByAuthDoesNotRequireVictoriaMetrics(t *testing.T) {
+	database, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatalf("open database: %v", err)
+	}
+	if err := database.AutoMigrate(&models.MonitorTask{}); err != nil {
+		t.Fatalf("migrate monitor table: %v", err)
+	}
+	monitor := models.MonitorTask{
+		ID:         "public-monitor",
+		Name:       "Public",
+		Enabled:    true,
+		Visibility: "public",
+	}
+	if err := database.Create(&monitor).Error; err != nil {
+		t.Fatalf("create monitor: %v", err)
+	}
+
+	metricService := NewMetricService(zap.NewNop(), database, nil, nil, nil)
+	monitorService := NewMonitorService(zap.NewNop(), database, metricService, nil)
+	items, err := monitorService.ListByAuth(t.Context(), false)
+	if err != nil {
+		t.Fatalf("ListByAuth returned error: %v", err)
+	}
+	if len(items) != 1 || items[0].ID != monitor.ID {
+		t.Fatalf("unexpected public monitor list: %+v", items)
+	}
+}