package app import ( "context" "net/http" "strconv" "sync" "time" "ccLoad/internal/model" "ccLoad/internal/util" "ccLoad/internal/version" "github.com/gin-gonic/gin" ) // ==================== 统计和监控 ==================== // 从admin.go拆分统计监控,遵循SRP原则 // HandleErrors 获取日志列表 // GET /admin/logs?range=today&limit=100&offset=0 func (s *Server) HandleErrors(c *gin.Context) { params := ParsePaginationParams(c) lf := BuildLogFilter(c) since, until := params.GetTimeRange() logs, total, err := s.store.ListLogsRangeWithCount(c.Request.Context(), since, until, params.Limit, params.Offset, &lf) if err != nil { RespondError(c, http.StatusInternalServerError, err) return } RespondJSONWithCount(c, http.StatusOK, logs, total) } // HandleMetrics 获取聚合指标数据 // GET /admin/metrics?range=today&bucket_min=5&channel_type=anthropic&model=claude-3-5-sonnet-20241022&channel_id=1&channel_name_like=xxx func (s *Server) HandleMetrics(c *gin.Context) { params := ParsePaginationParams(c) bucketMin, _ := strconv.Atoi(c.DefaultQuery("bucket_min", "5")) if bucketMin <= 0 { bucketMin = 5 } // 使用统一的筛选参数构建器(支持 channel_type、channel_id、channel_name_like、model、auth_token_id) lf := BuildLogFilter(c) lf.LogSource = model.LogSourceProxy since, until := params.GetTimeRange() pts, err := s.store.AggregateRangeWithFilter(c.Request.Context(), since, until, time.Duration(bucketMin)*time.Minute, &lf) if err != nil { RespondError(c, http.StatusInternalServerError, err) return } RespondJSON(c, http.StatusOK, pts) } // HandleStats 获取渠道和模型统计 // GET /admin/stats?range=today&channel_name_like=xxx&model_like=xxx func (s *Server) HandleStats(c *gin.Context) { params := ParsePaginationParams(c) lf := BuildLogFilter(c) lf.LogSource = model.LogSourceProxy startTime, endTime := params.GetTimeRange() // 判断是否为本日(本日才计算最近一分钟) isToday := params.Range == "today" || params.Range == "" stats, err := s.statsCache.GetStats(c.Request.Context(), startTime, endTime, &lf, isToday) if err != nil { RespondError(c, http.StatusInternalServerError, err) return } // 计算时间跨度(秒),用于前端计算RPM和QPS durationSeconds := endTime.Sub(startTime).Seconds() if durationSeconds < 1 { durationSeconds = 1 // 防止除零 } // 获取RPM统计(峰值、平均、最近一分钟) rpmStats, err := s.statsCache.GetRPMStats(c.Request.Context(), startTime, endTime, &lf, isToday) if err != nil { RespondError(c, http.StatusInternalServerError, err) return } // 计算健康时间线(固定48个时间点,当日显示最近4小时) channelHealth := s.fillHealthTimeline(c.Request.Context(), stats, startTime, endTime, &lf, isToday) RespondJSON(c, http.StatusOK, gin.H{ "stats": stats, "channel_health": channelHealth, "duration_seconds": durationSeconds, "rpm_stats": rpmStats, "is_today": isToday, }) } // HandlePublicSummary 获取基础统计摘要(公开端点,无需认证) // GET /public/summary?range=today // 按渠道类型分组统计,Claude和Codex类型包含Token和成本信息 // // [SECURITY NOTE] 该端点故意设计为公开访问,用于首页仪表盘展示。 // 如需隐藏运营数据,可在 server.go:SetupRoutes 中添加 RequireTokenAuth 中间件。 func (s *Server) HandlePublicSummary(c *gin.Context) { params := ParsePaginationParams(c) startTime, endTime := params.GetTimeRange() // 判断是否为本日(本日才计算最近一分钟) isToday := params.Range == "today" || params.Range == "" ctx := c.Request.Context() // [OPT] P1: 并行执行三个独立查询 var ( stats []model.StatsEntry rpmStats *model.RPMStats channelTypes map[int64]string statsErr error rpmErr error typesErr error wg sync.WaitGroup ) wg.Add(3) // 查询1: 基础统计(使用 Lite 版本跳过 fillStatsRPM) go func() { defer wg.Done() stats, statsErr = s.statsCache.GetStatsLite(ctx, startTime, endTime, nil) }() // 查询2: RPM统计 go func() { defer wg.Done() rpmStats, rpmErr = s.statsCache.GetRPMStats(ctx, startTime, endTime, nil, isToday) }() // 查询3: 渠道类型映射(带缓存) go func() { defer wg.Done() channelTypes, typesErr = s.getChannelTypesMapCached(ctx) }() wg.Wait() // 错误处理 if statsErr != nil { RespondError(c, http.StatusInternalServerError, statsErr) return } if rpmErr != nil { RespondError(c, http.StatusInternalServerError, rpmErr) return } if typesErr != nil { RespondError(c, http.StatusInternalServerError, typesErr) return } // 计算时间跨度(秒),用于前端计算RPM和QPS durationSeconds := endTime.Sub(startTime).Seconds() if durationSeconds < 1 { durationSeconds = 1 // 防止除零 } // 按渠道类型分组统计 typeStats := make(map[string]*TypeSummary) totalSuccess := 0 totalError := 0 for _, stat := range stats { // 获取渠道类型,跳过无法确定类型的记录(已删除的渠道) var channelType string if stat.ChannelID != nil { if ct, ok := channelTypes[int64(*stat.ChannelID)]; ok { channelType = ct } } if channelType == "" { // 渠道已删除或类型未知,不计入按类型统计(与 /admin/stats 保持一致) continue } totalSuccess += stat.Success totalError += stat.Error // 初始化类型统计 if _, exists := typeStats[channelType]; !exists { typeStats[channelType] = &TypeSummary{ ChannelType: channelType, TotalRequests: 0, SuccessRequests: 0, ErrorRequests: 0, } } ts := typeStats[channelType] ts.TotalRequests += stat.Success + stat.Error ts.SuccessRequests += stat.Success ts.ErrorRequests += stat.Error // 所有渠道类型都统计Token和成本 if stat.TotalInputTokens != nil { ts.TotalInputTokens += *stat.TotalInputTokens } if stat.TotalOutputTokens != nil { ts.TotalOutputTokens += *stat.TotalOutputTokens } if stat.TotalCost != nil { ts.TotalCost += *stat.TotalCost } if stat.EffectiveCost != nil { if ts.EffectiveCost == nil { ts.EffectiveCost = new(float64) } *ts.EffectiveCost += *stat.EffectiveCost } else if stat.TotalCost != nil { if ts.EffectiveCost == nil { ts.EffectiveCost = new(float64) } *ts.EffectiveCost += *stat.TotalCost } // Claude和Codex类型额外统计缓存(其他类型不支持prompt caching) if channelType == "anthropic" || channelType == "codex" { if stat.TotalCacheReadInputTokens != nil { ts.TotalCacheReadTokens += *stat.TotalCacheReadInputTokens } if stat.TotalCacheCreationInputTokens != nil { ts.TotalCacheCreationTokens += *stat.TotalCacheCreationInputTokens } } } response := gin.H{ "total_requests": totalSuccess + totalError, "success_requests": totalSuccess, "error_requests": totalError, "range": params.Range, "duration_seconds": durationSeconds, "rpm_stats": rpmStats, "is_today": isToday, "by_type": typeStats, // 按渠道类型分组的统计 } RespondJSON(c, http.StatusOK, response) } // TypeSummary 按渠道类型的统计摘要 type TypeSummary struct { ChannelType string `json:"channel_type"` TotalRequests int `json:"total_requests"` SuccessRequests int `json:"success_requests"` ErrorRequests int `json:"error_requests"` TotalInputTokens int64 `json:"total_input_tokens,omitempty"` // 所有类型 TotalOutputTokens int64 `json:"total_output_tokens,omitempty"` // 所有类型 TotalCacheReadTokens int64 `json:"total_cache_read_tokens,omitempty"` // Claude/Codex专用(prompt caching) TotalCacheCreationTokens int64 `json:"total_cache_creation_tokens,omitempty"` // Claude/Codex专用(prompt caching) TotalCost float64 `json:"total_cost,omitempty"` // 标准成本 EffectiveCost *float64 `json:"effective_cost,omitempty"` // 倍率后成本 } // fetchChannelTypesMap 查询所有渠道的类型映射 func (s *Server) fetchChannelTypesMap(ctx context.Context) (map[int64]string, error) { configs, err := s.store.ListConfigs(ctx) if err != nil { return nil, err } channelTypes := make(map[int64]string, len(configs)) for _, cfg := range configs { channelTypes[cfg.ID] = cfg.ChannelType } return channelTypes, nil } // getChannelTypesMapCached 带 TTL 缓存的渠道类型映射查询 // [OPT] P3: 渠道类型变化频率极低,使用 60 秒缓存减少数据库查询 const channelTypesCacheTTL = 60 * time.Second func (s *Server) getChannelTypesMapCached(ctx context.Context) (map[int64]string, error) { // 读锁检查缓存 s.channelTypesCacheMu.RLock() if s.channelTypesCache != nil && time.Since(s.channelTypesCacheTime) < channelTypesCacheTTL { result := s.channelTypesCache s.channelTypesCacheMu.RUnlock() return result, nil } s.channelTypesCacheMu.RUnlock() // 写锁更新缓存 s.channelTypesCacheMu.Lock() defer s.channelTypesCacheMu.Unlock() // 双重检查:可能其他 goroutine 已更新 if s.channelTypesCache != nil && time.Since(s.channelTypesCacheTime) < channelTypesCacheTTL { return s.channelTypesCache, nil } channelTypes, err := s.fetchChannelTypesMap(ctx) if err != nil { return nil, err } s.channelTypesCache = channelTypes s.channelTypesCacheTime = time.Now() return channelTypes, nil } // HandleCooldownStats 获取当前冷却状态监控指标 // GET /admin/cooldown/stats func (s *Server) HandleCooldownStats(c *gin.Context) { // 优先走缓存层,缓存不可用时自动降级到数据库查询 channelCooldowns, _ := s.getAllChannelCooldowns(c.Request.Context()) keyCooldowns, _ := s.getAllKeyCooldowns(c.Request.Context()) var keyCount int for _, m := range keyCooldowns { keyCount += len(m) } response := gin.H{ "channel_cooldowns": len(channelCooldowns), "key_cooldowns": keyCount, } RespondJSON(c, http.StatusOK, response) } // HandleGetChannelTypes 获取渠道类型配置(公开端点,前端动态加载) // GET /public/channel-types // 编译时常量,浏览器缓存24小时减少HF Spaces等高延迟环境的网络往返 func (s *Server) HandleGetChannelTypes(c *gin.Context) { c.Header("Cache-Control", "public, max-age=86400") RespondJSON(c, http.StatusOK, util.ChannelTypes) } // HandlePublicVersion 获取当前版本信息(公开端点,前端显示版本) // GET /public/version // 版本信息变化频率极低(后台每4小时检查一次),缓存5分钟 func (s *Server) HandlePublicVersion(c *gin.Context) { c.Header("Cache-Control", "public, max-age=300") hasUpdate, latestVersion, releaseURL := version.GetUpdateInfo() RespondJSON(c, http.StatusOK, gin.H{ "version": version.Version, "has_update": hasUpdate, "latest_version": latestVersion, "release_url": releaseURL, }) } // ModelsChannelsResponse 模型和渠道列表响应 type ModelsChannelsResponse struct { Models []string `json:"models"` Channels []model.ChannelNameID `json:"channels"` } // HandleGetModels 获取数据库中有日志的模型和渠道列表(去重) // GET /admin/models // 支持参数:range(时间范围)、channel_type(渠道类型筛选) func (s *Server) HandleGetModels(c *gin.Context) { rangeParam := c.DefaultQuery("range", "this_month") params := ParsePaginationParams(c) params.Range = rangeParam since, until := params.GetTimeRange() channelType := c.Query("channel_type") logFilter := &model.LogFilter{LogSource: model.LogSourceProxy} models, err := s.store.GetDistinctModels(c.Request.Context(), since, until, channelType, logFilter) if err != nil { RespondError(c, http.StatusInternalServerError, err) return } channels, err := s.store.GetDistinctChannels(c.Request.Context(), since, until, channelType, logFilter) if err != nil { RespondError(c, http.StatusInternalServerError, err) return } RespondJSON(c, http.StatusOK, ModelsChannelsResponse{Models: models, Channels: channels}) } // HandleHealth 健康检查端点(公开访问,无需认证) // GET /health // 仅检查数据库连接是否活跃(适用于K8s liveness/readiness probe) func (s *Server) HandleHealth(c *gin.Context) { // 设置100ms超时,避免慢查询阻塞healthcheck ctx, cancel := context.WithTimeout(c.Request.Context(), 100*time.Millisecond) defer cancel() if err := s.store.Ping(ctx); err != nil { RespondError(c, http.StatusServiceUnavailable, err) return } RespondJSON(c, http.StatusOK, gin.H{"status": "ok"}) } // fillHealthTimeline 为每个统计条目填充健康时间线 // isToday=true: 显示最近4小时,每5分钟一个状态(48个) // isToday=false: 按总时间跨度/48计算时间桶 func (s *Server) fillHealthTimeline(ctx context.Context, stats []model.StatsEntry, startTime, endTime time.Time, filter *model.LogFilter, isToday bool) map[int][]model.HealthPoint { if len(stats) == 0 { return nil } const numBuckets = 48 // 计算健康指示器的时间范围和桶大小 var healthStart time.Time var bucketSeconds int64 if isToday { // 当日:最近4小时,每5分钟一个桶 bucketSeconds = 5 * 60 // 5分钟 healthStart = endTime.Add(-4 * time.Hour) // 确保不早于查询开始时间 if healthStart.Before(startTime) { healthStart = startTime } } else { // 其他时间范围:按总时长/48计算 duration := endTime.Sub(startTime) bucketSeconds = int64(duration.Seconds() / numBuckets) if bucketSeconds < 1 { bucketSeconds = 1 } healthStart = startTime } // 转换为毫秒,直接与 logs.time 比较,避免索引失效 sinceMs := healthStart.UnixMilli() untilMs := endTime.UnixMilli() bucketMs := bucketSeconds * 1000 // 构建结构化查询参数(SQL 构建已下沉到存储层) params := model.HealthTimelineParams{ SinceMs: sinceMs, UntilMs: untilMs, BucketMs: bucketMs, Filter: filter, } rows, err := s.store.GetHealthTimeline(ctx, params) if err != nil { // 静默失败,不影响主流程 return nil } // 构建映射:(channel_id, model) -> StatsEntry索引 type channelModelKey struct { channelID int model string } statsMap := make(map[channelModelKey]int) for i := range stats { if stats[i].ChannelID != nil { key := channelModelKey{ channelID: *stats[i].ChannelID, model: stats[i].Model, } statsMap[key] = i } } // 解析查询结果 - 按时间桶索引位置填充 timeline := make(map[channelModelKey][]model.HealthPoint) sinceUnix := healthStart.Unix() // 为每个渠道初始化48个空时间点 for key := range statsMap { points := make([]model.HealthPoint, numBuckets) for i := 0; i < numBuckets; i++ { points[i] = model.HealthPoint{ Ts: time.Unix(sinceUnix+int64(i)*bucketSeconds, 0), SuccessRate: -1, // -1 表示无数据 } } timeline[key] = points } for _, row := range rows { key := channelModelKey{channelID: row.ChannelID, model: row.Model} // 只处理 stats 中存在的组合 if _, exists := statsMap[key]; !exists { continue } // 计算该时间桶对应的索引位置(BucketTs 是毫秒,需转换为秒再计算) bucketIndex := int((row.BucketTs/1000 - sinceUnix) / bucketSeconds) if bucketIndex < 0 || bucketIndex >= numBuckets { continue } total := row.Success + row.ErrorCount successRate := 0.0 if total > 0 { successRate = float64(row.Success) / float64(total) } // 更新数据字段,保留初始化时的 Ts(Go 计算的桶起始时间) // 不能用 SQL 的 FLOOR 桶边界覆盖 Ts,否则同一桶索引在不同模型间 // 产生不同时间戳,导致前端按 ts 合并时出现幽灵条目 p := &timeline[key][bucketIndex] p.SuccessRate = successRate p.SuccessCount = row.Success p.ErrorCount = row.ErrorCount p.AvgFirstByteTime = row.AvgFirstByteTime p.AvgDuration = row.AvgDuration p.TotalInputTokens = row.InputTokens p.TotalOutputTokens = row.OutputTokens p.TotalCacheReadTokens = row.CacheReadTokens p.TotalCacheCreationTokens = row.CacheCreationTokens p.TotalCost = row.TotalCost p.EffectiveCost = row.EffectiveCost } // 填充到 stats 中(per-model,供 stats 页面使用) for key, idx := range statsMap { if points, exists := timeline[key]; exists { stats[idx].HealthTimeline = points } } // 按渠道聚合健康时间线(供渠道管理页面使用) // 用桶索引合并,不依赖时间戳字符串,彻底避免前端 merge 的对齐问题 channelHealth := make(map[int][]model.HealthPoint) for key, points := range timeline { ch, exists := channelHealth[key.channelID] if !exists { ch = make([]model.HealthPoint, numBuckets) for i := range ch { ch[i] = model.HealthPoint{ Ts: points[i].Ts, SuccessRate: -1, } } channelHealth[key.channelID] = ch } for i, pt := range points { if pt.SuccessRate < 0 { continue } if ch[i].SuccessRate < 0 { ch[i] = pt continue } // 加权合并平均值(用 SuccessCount 做权重,比前端用 total 更准确) oldSucc := ch[i].SuccessCount newSucc := pt.SuccessCount if totalSucc := oldSucc + newSucc; totalSucc > 0 { w := float64(totalSucc) ch[i].AvgFirstByteTime = (ch[i].AvgFirstByteTime*float64(oldSucc) + pt.AvgFirstByteTime*float64(newSucc)) / w ch[i].AvgDuration = (ch[i].AvgDuration*float64(oldSucc) + pt.AvgDuration*float64(newSucc)) / w } ch[i].SuccessCount += pt.SuccessCount ch[i].ErrorCount += pt.ErrorCount if total := ch[i].SuccessCount + ch[i].ErrorCount; total > 0 { ch[i].SuccessRate = float64(ch[i].SuccessCount) / float64(total) } ch[i].TotalInputTokens += pt.TotalInputTokens ch[i].TotalOutputTokens += pt.TotalOutputTokens ch[i].TotalCacheReadTokens += pt.TotalCacheReadTokens ch[i].TotalCacheCreationTokens += pt.TotalCacheCreationTokens ch[i].TotalCost += pt.TotalCost ch[i].EffectiveCost += pt.EffectiveCost } } return channelHealth } // HandleStatsFilterOptions 返回统计页筛选下拉的全集(渠道名/模型), // 从指定时间范围内的日志记录中提取,与表格数据解耦。 // GET /admin/stats/filter-options?range=today&channel_type= func (s *Server) HandleStatsFilterOptions(c *gin.Context) { params := ParsePaginationParams(c) startTime, endTime := params.GetTimeRange() lf := BuildLogFilter(c) lf.LogSource = model.LogSourceProxy channelType := c.Query("channel_type") if channelType == "all" { channelType = "" } channels, err := s.store.GetDistinctChannels(c.Request.Context(), startTime, endTime, channelType, &lf) if err != nil { RespondError(c, http.StatusInternalServerError, err) return } models, err := s.store.GetDistinctModels(c.Request.Context(), startTime, endTime, channelType, &lf) if err != nil { RespondError(c, http.StatusInternalServerError, err) return } channelNames := make([]string, 0, len(channels)) for _, ch := range channels { if ch.Name != "" { channelNames = append(channelNames, ch.Name) } } RespondJSON(c, http.StatusOK, gin.H{ "channel_names": channelNames, "models": models, }) }