From 080579d0e683c4c7562218b501b517dfebc907c3 Mon Sep 17 00:00:00 2001 From: dushixiang Date: Sun, 23 Aug 2026 23:05:41 +0800 Subject: [PATCH] refactor: decouple public monitor sparklines --- internal/app.go | 1 + internal/handler/monitor_handler.go | 11 ++ internal/metric/monitor.go | 9 +- internal/service/metric_service.go | 128 ++++++++++++++++-- .../service/metric_service_sparkline_test.go | 105 ++++++++++++++ internal/service/monitor_service.go | 33 +++-- .../service/monitor_service_sparkline_test.go | 100 ++++++++++++++ 7 files changed, 364 insertions(+), 23 deletions(-) create mode 100644 internal/service/monitor_service_sparkline_test.go diff --git a/internal/app.go b/internal/app.go index 56b44e4..166e7f6 100644 --- a/internal/app.go +++ b/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) diff --git a/internal/handler/monitor_handler.go b/internal/handler/monitor_handler.go index 6935914..f777a3a 100644 --- a/internal/handler/monitor_handler.go +++ b/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") diff --git a/internal/metric/monitor.go b/internal/metric/monitor.go index c81e6ec..182bdbf 100644 --- a/internal/metric/monitor.go +++ b/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 监控详情响应（整合版） diff --git a/internal/service/metric_service.go b/internal/service/metric_service.go index e169386..d8bcdd1 100644 --- a/internal/service/metric_service.go +++ b/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 { diff --git a/internal/service/metric_service_sparkline_test.go b/internal/service/metric_service_sparkline_test.go index 1def3ba..e39e6bb 100644 --- a/internal/service/metric_service_sparkline_test.go +++ b/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) + } +} diff --git a/internal/service/monitor_service.go b/internal/service/monitor_service.go index 27f3408..cbc5e8d 100644 --- a/internal/service/monitor_service.go +++ b/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 构建监控概览对象 diff --git a/internal/service/monitor_service_sparkline_test.go b/internal/service/monitor_service_sparkline_test.go new file mode 100644 index 0000000..2f8d0e7 --- /dev/null +++ b/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) + } +}