perf(usage): 优化 /admin/usage 打开速度与刷新响应

根因:页面 mount 并发 6 个请求,其中 5 个在原始 usage_logs 上 live 聚合,
且有 1 个重复 getModelStats。优化(不引入预聚合):

前端
- UsageView 经 :model-options 下传 model 列表,移除 UsageFilters 重复的
  getModelStats(mount 请求 6→5,少一次 usage_logs 全表 GROUP BY model)
- 刷新/换筛选保留旧模型数据(invalidateModelStatsCache 只失效标记不清空数据),
  图表不再闪空,刷新期间页面保持可交互

后端
- GetStatsWithFilters 4 条聚合 errgroup 并行(仅 *sql.DB 连接池路径,ent.Tx
  顺序回退以保事务内不并发),endpoint 明细 best-effort;抑制取消级联噪声日志
- /admin/usage/stats 复用 dashboard 的 newSnapshotCache 30s 处理器层缓存,按
  filters+窗口为 key;前端手动刷新带 nocache=1 强制回源(刷新=最新)

注:自合并提交 4c8396c 迁移而来,仅取"列表打开速度与刷新响应"部分;原提交的
"审计查看弹窗大 body 渲染"改动(AuditLogModal / audit-log-format / 审计 i18n)
依赖尚未迁移的审计功能(d8389ade),本次已排除。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
DaydreamCoding
2026-05-30 18:56:08 +08:00
committed by QTom
co-authored by Claude Opus 4.8
parent b60d8bb4cc
commit bf24b61139
10 changed files with 325 additions and 74 deletions
@@ -325,10 +325,24 @@ func (h *UsageHandler) Stats(c *gin.Context) {
EndTime: &endTime,
}
stats, err := h.usageService.GetStatsWithFilters(c.Request.Context(), filters)
if err != nil {
response.ErrorFrom(c, err)
return
var stats *usagestats.UsageStats
// nocache: 绕过缓存直接回源,刷新者本人拿最新;不回写缓存(管理台"我刷新我自己拿最新"语义,非全局失效)。
if parseBoolQueryWithDefault(c.Query("nocache"), false) {
s, err := h.usageService.GetStatsWithFilters(c.Request.Context(), filters)
if err != nil {
response.ErrorFrom(c, err)
return
}
stats = s
c.Header("X-Usage-Stats-Cache", "bypass")
} else {
s, hit, err := h.getStatsCached(c.Request.Context(), filters)
if err != nil {
response.ErrorFrom(c, err)
return
}
stats = s
c.Header("X-Usage-Stats-Cache", cacheStatusValue(hit))
}
response.Success(c, stats)
@@ -0,0 +1,62 @@
package admin
import (
"context"
"time"
"github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
)
// 与 dashboard 查询缓存同款:30s TTL 进程内缓存,仅服务 /admin/usage/stats 读路径。
var usageStatsCache = newSnapshotCache(30 * time.Second)
type usageStatsCacheKeyData struct {
StartTime string `json:"start_time"`
EndTime string `json:"end_time"`
UserID int64 `json:"user_id"`
APIKeyID int64 `json:"api_key_id"`
AccountID int64 `json:"account_id"`
GroupID int64 `json:"group_id"`
Model string `json:"model"`
BillingMode string `json:"billing_mode"`
RequestType *int16 `json:"request_type"`
Stream *bool `json:"stream"`
BillingType *int8 `json:"billing_type"`
}
func usageStatsCacheKey(filters usagestats.UsageLogFilters) string {
start := ""
if filters.StartTime != nil {
start = filters.StartTime.UTC().Format(time.RFC3339)
}
end := ""
if filters.EndTime != nil {
end = filters.EndTime.UTC().Format(time.RFC3339)
}
return mustMarshalDashboardCacheKey(usageStatsCacheKeyData{
StartTime: start,
EndTime: end,
UserID: filters.UserID,
APIKeyID: filters.APIKeyID,
AccountID: filters.AccountID,
GroupID: filters.GroupID,
Model: filters.Model,
BillingMode: filters.BillingMode,
RequestType: filters.RequestType,
Stream: filters.Stream,
BillingType: filters.BillingType,
})
}
// getStatsCached 命中则返回缓存,未命中则回源 usageService 并写缓存。
func (h *UsageHandler) getStatsCached(ctx context.Context, filters usagestats.UsageLogFilters) (*usagestats.UsageStats, bool, error) {
key := usageStatsCacheKey(filters)
entry, hit, err := usageStatsCache.GetOrLoad(key, func() (any, error) {
return h.usageService.GetStatsWithFilters(ctx, filters)
})
if err != nil {
return nil, hit, err
}
stats, err := snapshotPayloadAs[*usagestats.UsageStats](entry.Payload)
return stats, hit, err
}
@@ -0,0 +1,28 @@
package admin
import (
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
"github.com/stretchr/testify/require"
)
func TestUsageStatsCacheKey_StableAndDistinct(t *testing.T) {
start := time.Date(2026, 5, 29, 0, 0, 0, 0, time.UTC)
end := time.Date(2026, 5, 31, 0, 0, 0, 0, time.UTC)
base := usagestats.UsageLogFilters{StartTime: &start, EndTime: &end, Model: "claude-3"}
k1 := usageStatsCacheKey(base)
k2 := usageStatsCacheKey(base)
require.NotEmpty(t, k1)
require.Equal(t, k1, k2, "same filters must produce same key")
other := base
other.Model = "gpt-4o"
require.NotEqual(t, k1, usageStatsCacheKey(other), "different model must change key")
withUser := base
withUser.UserID = 7
require.NotEqual(t, k1, usageStatsCacheKey(withUser), "different user must change key")
}
+68 -30
View File
@@ -27,6 +27,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/lib/pq"
gocache "github.com/patrickmn/go-cache"
"golang.org/x/sync/errgroup"
)
const usageLogSelectColumns = "id, user_id, api_key_id, account_id, request_id, model, requested_model, upstream_model, group_id, subscription_id, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, image_output_tokens, image_output_cost, input_cost, output_cost, cache_creation_cost, cache_read_cost, total_cost, actual_cost, rate_multiplier, account_rate_multiplier, billing_type, request_type, stream, openai_ws_mode, duration_ms, first_token_ms, user_agent, ip_address, image_count, image_size, image_input_size, image_output_size, image_size_source, image_size_breakdown, service_tier, reasoning_effort, inbound_endpoint, upstream_endpoint, cache_ttl_overridden, channel_id, model_mapping_chain, billing_tier, billing_mode, account_stats_cost, created_at"
@@ -3538,24 +3539,6 @@ func (r *usageLogRepository) GetStatsWithFilters(ctx context.Context, filters Us
stats := &UsageStats{}
var totalAccountCost float64
if err := scanSingleRow(
ctx,
r.sql,
query,
args,
&stats.TotalRequests,
&stats.TotalInputTokens,
&stats.TotalOutputTokens,
&stats.TotalCacheTokens,
&stats.TotalCost,
&stats.TotalActualCost,
&totalAccountCost,
&stats.AverageDurationMs,
); err != nil {
return nil, err
}
stats.TotalAccountCost = &totalAccountCost
stats.TotalTokens = stats.TotalInputTokens + stats.TotalOutputTokens + stats.TotalCacheTokens
start := time.Unix(0, 0).UTC()
if filters.StartTime != nil {
@@ -3566,21 +3549,76 @@ func (r *usageLogRepository) GetStatsWithFilters(ctx context.Context, filters Us
end = *filters.EndTime
}
endpoints, endpointErr := r.GetEndpointStatsWithFilters(ctx, start, end, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType)
if endpointErr != nil {
logger.LegacyPrintf("repository.usage_log", "GetEndpointStatsWithFilters failed in GetStatsWithFilters: %v", endpointErr)
endpoints = []EndpointStat{}
var endpoints, upstreamEndpoints, endpointPaths []EndpointStat
// 汇总查询:失败即致命。
runSummary := func(c context.Context) error {
return scanSingleRow(
c, r.sql, query, args,
&stats.TotalRequests,
&stats.TotalInputTokens,
&stats.TotalOutputTokens,
&stats.TotalCacheTokens,
&stats.TotalCost,
&stats.TotalActualCost,
&totalAccountCost,
&stats.AverageDurationMs,
)
}
upstreamEndpoints, upstreamEndpointErr := r.GetUpstreamEndpointStatsWithFilters(ctx, start, end, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType)
if upstreamEndpointErr != nil {
logger.LegacyPrintf("repository.usage_log", "GetUpstreamEndpointStatsWithFilters failed in GetStatsWithFilters: %v", upstreamEndpointErr)
upstreamEndpoints = []EndpointStat{}
// endpoint 明细:best-effort(失败 log + 返空),不致命。
runEndpoints := func(c context.Context) {
res, err := r.GetEndpointStatsWithFilters(c, start, end, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType)
if err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.LegacyPrintf("repository.usage_log", "GetEndpointStatsWithFilters failed in GetStatsWithFilters: %v", err)
}
res = []EndpointStat{}
}
endpoints = res
}
endpointPaths, endpointPathErr := r.getEndpointPathStatsWithFilters(ctx, start, end, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType)
if endpointPathErr != nil {
logger.LegacyPrintf("repository.usage_log", "getEndpointPathStatsWithFilters failed in GetStatsWithFilters: %v", endpointPathErr)
endpointPaths = []EndpointStat{}
runUpstream := func(c context.Context) {
res, err := r.GetUpstreamEndpointStatsWithFilters(c, start, end, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType)
if err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.LegacyPrintf("repository.usage_log", "GetUpstreamEndpointStatsWithFilters failed in GetStatsWithFilters: %v", err)
}
res = []EndpointStat{}
}
upstreamEndpoints = res
}
runPaths := func(c context.Context) {
res, err := r.getEndpointPathStatsWithFilters(c, start, end, filters.UserID, filters.APIKeyID, filters.AccountID, filters.GroupID, filters.Model, filters.RequestType, filters.Stream, filters.BillingType)
if err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.LegacyPrintf("repository.usage_log", "getEndpointPathStatsWithFilters failed in GetStatsWithFilters: %v", err)
}
res = []EndpointStat{}
}
endpointPaths = res
}
if r.db != nil {
// 生产路径:r.sql 是 *sql.DB 连接池,可并发。4 条查询并行,延迟取最大值。
g, gctx := errgroup.WithContext(ctx)
g.Go(func() error { return runSummary(gctx) })
g.Go(func() error { runEndpoints(gctx); return nil })
g.Go(func() error { runUpstream(gctx); return nil })
g.Go(func() error { runPaths(gctx); return nil })
if err := g.Wait(); err != nil {
return nil, err
}
} else {
// 事务路径(ent.Tx 不能并发查询):顺序执行,行为与重构前一致。
if err := runSummary(ctx); err != nil {
return nil, err
}
runEndpoints(ctx)
runUpstream(ctx)
runPaths(ctx)
}
stats.TotalAccountCost = &totalAccountCost
stats.TotalTokens = stats.TotalInputTokens + stats.TotalOutputTokens + stats.TotalCacheTokens
stats.Endpoints = endpoints
stats.UpstreamEndpoints = upstreamEndpoints
stats.EndpointPaths = endpointPaths
@@ -0,0 +1,51 @@
//go:build integration
package repository
import (
"context"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/stretchr/testify/require"
)
func TestUsageLog_GetStatsWithFilters_AggregatesAndEndpoints(t *testing.T) {
ctx := context.Background()
tx := testEntTx(t)
client := tx.Client()
repo := newUsageLogRepositoryWithSQL(client, tx)
user := mustCreateUser(t, client, &service.User{Email: "stats@test.com"})
apiKey := mustCreateApiKey(t, client, &service.APIKey{UserID: user.ID, Key: "sk-stats-1", Name: "k"})
account := mustCreateAccount(t, client, &service.Account{Name: "acc-stats"})
now := time.Now().UTC()
inboundEndpoint := "/v1/messages"
upstreamEndpoint := "/v1/responses"
for i := 0; i < 3; i++ {
_, err := repo.Create(ctx, &service.UsageLog{
UserID: user.ID, APIKeyID: apiKey.ID, AccountID: account.ID,
Model: "claude-3", InputTokens: 2, OutputTokens: 3,
TotalCost: 0.5, ActualCost: 0.4, CreatedAt: now,
InboundEndpoint: &inboundEndpoint, UpstreamEndpoint: &upstreamEndpoint,
})
require.NoError(t, err)
}
start := now.Add(-1 * time.Hour)
end := now.Add(1 * time.Hour)
// 按本测试创建的 user 维度过滤:集成库为共享实例,其它用 testEntClient 的兄弟测试会留下
// 已提交的 usage_log 行(含零 token 的失败请求),不限定 user 会把它们计入 TotalRequests。
stats, err := repo.GetStatsWithFilters(ctx, usagestats.UsageLogFilters{UserID: user.ID, StartTime: &start, EndTime: &end})
require.NoError(t, err)
require.Equal(t, int64(3), stats.TotalRequests)
require.Equal(t, int64(6), stats.TotalInputTokens)
require.Equal(t, int64(9), stats.TotalOutputTokens)
require.InDelta(t, 1.2, stats.TotalActualCost, 1e-9)
require.NotEmpty(t, stats.Endpoints)
require.NotEmpty(t, stats.UpstreamEndpoints)
require.NotEmpty(t, stats.EndpointPaths)
}