- 上游 429 的 Retry-After 转成渠道冷却时长(夹在 10 分钟内;比默认值短则不缩短, 否则 1s 就把坏渠道放回轮转)。失败计数改由 proxyOne 就地报——只有它读得到响应头。 - 故障转移时间预算 OT_PROXY_FAILOVER_BUDGET(默认 30s):只约束第二个及以后的渠道, 首个不受限(长回答本就要跑几分钟),但坏渠道不再把客户端拖到 N×timeout。 - 流式空闲心跳 OT_PROXY_STREAM_HEARTBEAT(默认 15s):补一行 SSE 注释 `: ping`, 防止 nginx/Cloudflare 的空闲超时掐掉长思考。心跳与正文共用一把写的锁, gin 的 ResponseWriter 不是并发安全的;注释行不进转换也不计费。 - 渠道观测:GET /api/v1/admin/channels/stats?window=5m|1h|6h|24h|7d 聚合请求量、 成功率、P50/P95/max 延迟、成本与 top_errors;分位只取成功样本,canceled(客户端 主动断开)不计入渠道错误率。渠道页桌面表格加「近N小时」列、移动端加同内容条, 顶部可切窗口。 - 修一个让统计瞎掉的归因问题:全部渠道失败时 recordError 传的是 nil, usage_logs 里 channel_id=0,这类失败在渠道维度完全看不见。改为归到最后一个 真正试过的渠道。 - 模型定价与元数据:Model 新增 cache_write_price(0 则回退输入价×1.25,可显式配 Anthropic 1h 的 ×2)、context_length 与 supports_tools/vision/reasoning; finishUsage 走 CacheWritePriceOr()。/v1/models 叠加用户与密钥白名单过滤 (原来会把限制外的模型名直接泄漏在列表里)并返回元数据,价格不外泄。 - /v1/models 的 Anthropic 分支原是死代码:协议只按路径判,而两种客户端共用这个路径。 改为按 x-api-key / anthropic-version 头判,OpenAI SDK 从不发这两个头,互不影响。 - NewGateway 的十个位置参数换成 Options 结构体(三个 Duration 两个 int,传错编译期发现不了)。 新增 6 个用例:Retry-After 两种写法与夹顶、重试预算截止、空闲心跳出帧、 /v1/models 过滤与双形状、缓存写价两条路径、失败归因。
249 lines
7.4 KiB
Go
249 lines
7.4 KiB
Go
package api
|
|
|
|
import (
|
|
"math"
|
|
"sort"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/openteam/server/internal/pkg/resp"
|
|
"github.com/openteam/server/internal/store"
|
|
)
|
|
|
|
// AdminStatsOverview GET /api/v1/admin/stats/overview — 运营总览。
|
|
func (h *Handler) AdminStatsOverview(c *gin.Context) {
|
|
now := time.Now().UTC()
|
|
today := now.Format("2006-01-02")
|
|
month := now.Format("2006-01")
|
|
|
|
var totalUsers int64
|
|
h.a.DB.Model(&store.User{}).Count(&totalUsers)
|
|
var totalKeys int64
|
|
h.a.DB.Model(&store.APIKey{}).Count(&totalKeys)
|
|
var totalChannels int64
|
|
h.a.DB.Model(&store.Channel{}).Count(&totalChannels)
|
|
var totalModels int64
|
|
h.a.DB.Model(&store.Model{}).Count(&totalModels)
|
|
|
|
// 全局今日/本月汇总(跨用户)
|
|
var todayReq, monthReq int64
|
|
var todayCost, monthCost float64
|
|
var todayTokens, monthTokens int64
|
|
h.a.DB.Model(&store.UsageDaily{}).Where("date = ?", today).
|
|
Select("COALESCE(SUM(requests),0), COALESCE(SUM(cost),0), COALESCE(SUM(input_tokens+output_tokens+cache_read_tokens),0)").
|
|
Row().Scan(&todayReq, &todayCost, &todayTokens)
|
|
h.a.DB.Model(&store.UsageDaily{}).Where("date LIKE ?", month+"%").
|
|
Select("COALESCE(SUM(requests),0), COALESCE(SUM(cost),0), COALESCE(SUM(input_tokens+output_tokens+cache_read_tokens),0)").
|
|
Row().Scan(&monthReq, &monthCost, &monthTokens)
|
|
|
|
// 近 14 天趋势(日粒度)
|
|
var days []struct {
|
|
Date string
|
|
Req int64
|
|
Cost float64
|
|
}
|
|
h.a.DB.Model(&store.UsageDaily{}).
|
|
Select("date, SUM(requests) req, SUM(cost) cost").
|
|
Where("date >= ?", now.AddDate(0, 0, -13).Format("2006-01-02")).
|
|
Group("date").Order("date").Scan(&days)
|
|
trend := make([]gin.H, 0, len(days))
|
|
for _, d := range days {
|
|
trend = append(trend, gin.H{"date": d.Date, "requests": d.Req, "cost": d.Cost})
|
|
}
|
|
|
|
resp.OK(c, gin.H{
|
|
"total_users": totalUsers, "total_keys": totalKeys,
|
|
"total_channels": totalChannels, "total_models": totalModels,
|
|
"today": gin.H{"requests": todayReq, "cost": todayCost, "tokens": todayTokens},
|
|
"month": gin.H{"requests": monthReq, "cost": monthCost, "tokens": monthTokens},
|
|
"trend_14d": trend,
|
|
})
|
|
}
|
|
|
|
// AdminChannelStats GET /api/v1/admin/channels/stats?window=1h — 按渠道聚合近一段的
|
|
// 请求量、成功率、错误码分布与延迟分位。定位"哪条渠道在掉"不该靠翻日志。
|
|
func (h *Handler) AdminChannelStats(c *gin.Context) {
|
|
win := c.DefaultQuery("window", "1h")
|
|
d, ok := statWindows[win]
|
|
if !ok {
|
|
win, d = "1h", statWindows["1h"]
|
|
}
|
|
since := time.Now().UTC().Add(-d)
|
|
|
|
// 只取窗口内最近这么多行,避免大窗口把内存吃光(超出部分不参与分位计算)
|
|
const maxRows = 50000
|
|
var rows []channelStatRow
|
|
h.a.DB.Model(&store.UsageLog{}).
|
|
Select("channel_id, status, latency_ms, cost, "+
|
|
"(input_tokens + output_tokens + cache_read_tokens) AS tokens, error_code, created_at").
|
|
Where("created_at >= ? AND channel_id > 0", since).
|
|
Order("id DESC").Limit(maxRows).Scan(&rows)
|
|
|
|
agg := map[uint64]*chanAgg{}
|
|
for _, r := range rows {
|
|
a := agg[r.ChannelID]
|
|
if a == nil {
|
|
a = &chanAgg{lat: make([]int, 0, 64), errs: map[string]int{}}
|
|
agg[r.ChannelID] = a
|
|
}
|
|
a.requests++
|
|
a.cost += r.Cost
|
|
a.tokens += r.Tokens
|
|
switch r.Status {
|
|
case store.UsageStatusSuccess:
|
|
a.success++
|
|
a.lat = append(a.lat, r.LatencyMS) // 分位只按正常完成的请求算,中断/失败的延迟不代表体感
|
|
case store.UsageStatusCanceled:
|
|
a.canceled++
|
|
default:
|
|
a.err++
|
|
if r.ErrorCode != nil && *r.ErrorCode != "" {
|
|
a.errs[*r.ErrorCode]++
|
|
}
|
|
}
|
|
if r.CreatedAt.After(a.lastAt) {
|
|
a.lastAt = r.CreatedAt
|
|
}
|
|
}
|
|
|
|
var chans []store.Channel
|
|
h.a.DB.Find(&chans)
|
|
names := make(map[uint64]string, len(chans))
|
|
for _, ch := range chans {
|
|
names[ch.ID] = ch.Name
|
|
}
|
|
|
|
out := make([]gin.H, 0, len(agg))
|
|
for id, a := range agg {
|
|
out = append(out, gin.H{
|
|
"channel_id": id, "channel_name": names[id],
|
|
"requests": a.requests, "success": a.success,
|
|
"error": a.err, "canceled": a.canceled,
|
|
"err_rate": rate(a.err, a.requests), // canceled 是客户端主动断开,不算渠道的错
|
|
"p50_ms": percentile(a.lat, 0.50),
|
|
"p95_ms": percentile(a.lat, 0.95),
|
|
"max_ms": percentile(a.lat, 1.0),
|
|
"tokens": a.tokens,
|
|
"cost": a.cost,
|
|
"top_errors": topErrors(a.errs, 3),
|
|
"last_at": a.lastAt,
|
|
})
|
|
}
|
|
sort.Slice(out, func(i, j int) bool {
|
|
return out[i]["requests"].(int64) > out[j]["requests"].(int64)
|
|
})
|
|
resp.OK(c, gin.H{"window": win, "since": since, "items": out})
|
|
}
|
|
|
|
// statWindows 支持的统计窗口。
|
|
var statWindows = map[string]time.Duration{
|
|
"5m": 5 * time.Minute,
|
|
"1h": time.Hour,
|
|
"6h": 6 * time.Hour,
|
|
"24h": 24 * time.Hour,
|
|
"7d": 7 * 24 * time.Hour,
|
|
}
|
|
|
|
type channelStatRow struct {
|
|
ChannelID uint64
|
|
Status string
|
|
LatencyMS int
|
|
Cost float64
|
|
Tokens int64
|
|
ErrorCode *string
|
|
CreatedAt time.Time
|
|
}
|
|
|
|
type chanAgg struct {
|
|
requests, success, err, canceled int64
|
|
cost float64
|
|
tokens int64
|
|
lat []int
|
|
errs map[string]int
|
|
lastAt time.Time
|
|
}
|
|
|
|
func rate(part, total int64) float64 {
|
|
if total <= 0 {
|
|
return 0
|
|
}
|
|
return float64(part) / float64(total)
|
|
}
|
|
|
|
// percentile 最近秩法取分位(q=1 为最大值);空样本返回 0。
|
|
func percentile(xs []int, q float64) int {
|
|
if len(xs) == 0 {
|
|
return 0
|
|
}
|
|
sort.Ints(xs)
|
|
idx := int(math.Ceil(q*float64(len(xs)))) - 1
|
|
if idx < 0 {
|
|
idx = 0
|
|
}
|
|
if idx >= len(xs) {
|
|
idx = len(xs) - 1
|
|
}
|
|
return xs[idx]
|
|
}
|
|
|
|
func topErrors(m map[string]int, n int) []gin.H {
|
|
list := make([]gin.H, 0, len(m))
|
|
for code, cnt := range m {
|
|
list = append(list, gin.H{"code": code, "count": cnt})
|
|
}
|
|
sort.Slice(list, func(i, j int) bool { return list[i]["count"].(int) > list[j]["count"].(int) })
|
|
if len(list) > n {
|
|
list = list[:n]
|
|
}
|
|
return list
|
|
}
|
|
|
|
// AdminUsage GET /api/v1/admin/usage — 全局用量日志(分页 + 过滤)。
|
|
func (h *Handler) AdminUsage(c *gin.Context) {
|
|
page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
|
|
size, _ := strconv.Atoi(c.DefaultQuery("page_size", "20"))
|
|
if page < 1 {
|
|
page = 1
|
|
}
|
|
if size < 1 || size > 100 {
|
|
size = 20
|
|
}
|
|
q := h.a.DB.Model(&store.UsageLog{})
|
|
if from := c.Query("from"); from != "" {
|
|
q = q.Where("created_at >= ?", from+" 00:00:00")
|
|
}
|
|
if to := c.Query("to"); to != "" {
|
|
q = q.Where("created_at <= ?", to+" 23:59:59")
|
|
}
|
|
if model := c.Query("model"); model != "" {
|
|
q = q.Where("model_name = ?", model)
|
|
}
|
|
if user := c.Query("user_id"); user != "" {
|
|
q = q.Where("user_id = ?", user)
|
|
}
|
|
var total int64
|
|
q.Count(&total)
|
|
var logs []store.UsageLog
|
|
q.Order("id DESC").Offset((page - 1) * size).Limit(size).Find(&logs)
|
|
out := make([]gin.H, 0, len(logs))
|
|
for _, l := range logs {
|
|
var uname string
|
|
h.a.DB.Model(&store.User{}).Where("id = ?", l.UserID).Pluck("username", &uname)
|
|
var chName string
|
|
if l.ChannelID > 0 {
|
|
h.a.DB.Model(&store.Channel{}).Where("id = ?", l.ChannelID).Pluck("name", &chName)
|
|
}
|
|
out = append(out, gin.H{
|
|
"id": l.ID, "user": uname, "user_id": l.UserID, "model": l.ModelName, "protocol": l.Protocol,
|
|
"channel_id": l.ChannelID, "channel": chName,
|
|
"input_tokens": l.InputTokens, "output_tokens": l.OutputTokens,
|
|
"cache_read_tokens": l.CacheReadTokens, "cost": l.Cost,
|
|
"latency_ms": l.LatencyMS, "status": l.Status, "error_code": l.ErrorCode,
|
|
"raw_request": l.RawRequest, "raw_response": l.RawResponse,
|
|
"created_at": l.CreatedAt,
|
|
})
|
|
}
|
|
resp.OK(c, gin.H{"items": out, "total": total, "page": page, "page_size": size})
|
|
}
|