From 531a03c836948ac64574bae7286cfdbf6421b830 Mon Sep 17 00:00:00 2001 From: Sakurasan <26715255+Sakurasan@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:45:57 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E6=B5=81=E5=BC=8F=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E5=88=A4=E5=AE=9A=E9=A2=84=E8=BF=87=E6=BB=A4=E3=80=81=E5=86=8D?= =?UTF-8?q?=E5=87=8F=E4=B8=80=E4=B8=AA=E5=86=99=E7=B4=A2=E5=BC=95=EF=BC=8C?= =?UTF-8?q?=E5=8A=A0=E5=8F=AF=E9=80=89=E6=98=8E=E7=BB=86=E4=BF=9D=E7=95=99?= =?UTF-8?q?=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 流式放行前的上游错误判定加廉价子串预过滤（"error" / response.failed）： 长思考阶段几十上百条事件不再逐条 json.Unmarshal。 - usage_logs 去掉 idx_channel_created：没有任何查询按 channel_id 过滤该表 （渠道统计走 created_at 时间窗后在内存分组），复合索引只是每行一次多余的 B-tree 插入；schema v9 迁移里显式 DROP 旧库索引。至此该表只剩 (user_id, created_at)、created_at 两个真被用到的索引。 - 新增可选保留期（OT_USAGE_RETENTION_DAYS，默认 0=永久保留）：每天分批 （2000 行/批、批间 50ms）清理过期 usage_logs，避免长期运行后时间窗查询 与备份随表增长变慢；余额流水 balance_logs 不在清理范围。 参数复测（记账稳态吞吐，3 次取中位）： - batch 64（当前）30.2k rows/s；batch 128 19.0k、batch 256 14.3k（更差，保持 64） - wal_autocheckpoint(4000)、cache_size(-64MB) 与当前配置无差别，不引入 验证：go vet、go test ./...、-race（usage/proxy/store）全绿；新增保留期三项测试 （只删明细不动流水、未开启不动数据、跨分批删完）。 --- .env.example | 4 ++ README.md | 1 + server/internal/app/app.go | 32 +++++---- server/internal/config/config.go | 12 ++++ server/internal/proxy/hold.go | 5 ++ server/internal/store/db.go | 5 +- server/internal/store/models.go | 15 +++-- server/internal/usage/retention.go | 86 +++++++++++++++++++++++++ server/internal/usage/retention_test.go | 84 ++++++++++++++++++++++++ 9 files changed, 224 insertions(+), 20 deletions(-) create mode 100644 server/internal/usage/retention.go create mode 100644 server/internal/usage/retention_test.go diff --git a/.env.example b/.env.example index 7c3041d..a06ad34 100644 --- a/.env.example +++ b/.env.example @@ -64,6 +64,10 @@ OT_PROXY_HEALTH_FAIL_THRESHOLD=2 # 原始体会截断：请求体 256 KiB、响应体 1 MiB，避免单条日志几十 MB 拖慢查询与备份。 OT_PROXY_LOG_RAW=false +# 用量明细保留天数（0=永久保留，默认）。开启后每天清理超过该天数的 usage_logs， +# 分批删除（每批 2000 行），余额流水 balance_logs 不受影响。 +OT_USAGE_RETENTION_DAYS=0 + # 限流（内存计数，Redis 后置）：用户级每秒请求数上限（0=不限制） OT_RATELIMIT_USER_RPS=20 # 认证端点限速（每分钟，0=不限制）：登录是密码爆破面，注册是批量建号面 diff --git a/README.md b/README.md index d44d545..139add8 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,7 @@ - **渠道体系**：按模型绑定选渠道 + 加权负载均衡；模型已有绑定时只用绑定渠道（都不可用则 503，不会把请求发给不认识它的渠道）；每渠道并发信号量（满载排队 `OT_PROXY_CHANNEL_ACQUIRE_WAIT`，默认 150ms，仍满则 503 + Retry-After 而不是 502）；后台健康检查（探测 `/models`，连续失败进 cooldown、恢复放回）+ 运行期健康（按失败原因分类冷却：网络错误/5xx 连续 3 次进冷却且指数退避，余额/配额类第一次即冷却 30 分钟，成功清零退避计数）；可安全重试的失败自动故障转移（网络错误/429/5xx/超时/流式首个内容事件之前的任何失败）；**会话亲和**（`OT_PROXY_AFFINITY_TTL`，默认 5m）：同一会话粘住上次应答的渠道，命中上游提示词缓存而不是全价重算，冷却/下线自动回退加权随机 - **上游超时**：渠道 `timeout_ms` 生效——非流式为整段请求上限，流式仅约束首包（正文不限时长）；上游请求绑定客户端 context，断连即取消 - **故障转移与流式细节**（借鉴 magpie 的 holdWriter）：流式请求在**首个内容事件**到达前由网关缓冲——上游的错误事件、空流、断连、协议形态不符（要流给了整包，didNotStream）都在放行前换渠道消化，客户端无感知；`OT_PROXY_FIRST_TOKEN_WAIT` 限定放行前等内容的最长时间（0=不限）；429 不一定是限流——错误体带余额/配额特征时按配额类长冷却；上游 `Retry-After` 会延长该渠道冷却（夹在 10 分钟内）；重试受 `OT_PROXY_FAILOVER_BUDGET` 时间预算约束（首个渠道不受限，避免坏渠道把客户端拖到 N×timeout）；流式空闲超过 `OT_PROXY_STREAM_HEARTBEAT` 补一行 SSE 注释心跳（扣住期间也照发），防止 nginx/Cloudflare 的空闲超时掐掉长思考；响应头已被心跳先行写出时，全部候选失败以 SSE 错误事件收尾 +- **数据保留**：`OT_USAGE_RETENTION_DAYS`（默认 0=永久保留）开启后每天分批清理过期用量明细；余额流水不清理 - **渠道观测**：`GET /api/v1/admin/channels/stats?window=5m|1h|6h|24h|7d` 按渠道聚合请求量、成功率、错误码分布与 P50/P95 延迟；渠道页直接展示近况与运行期冷却剩余时间，可一键恢复 - **限流/配额**（内存计数）：密钥级每日请求数 / Token 数配额、用户级每秒速率（`OT_RATELIMIT_USER_RPS`）；超限返回 429 - **用量计费**：请求级 `usage_logs` 异步批量落库，按模型价格扣减余额，日粒度预聚合（`usage_daily`）；模型可单独配置缓存写价（留空按输入价 ×1.25）；记录思考 token（`reasoning_tokens`，展示口径不重复计费）、Anthropic 1h 缓存写（计价 2× 输入）与上游实际应答的模型名（`served_model`，映射/聚合商场景对账用） diff --git a/server/internal/app/app.go b/server/internal/app/app.go index 8117d8d..ba8590a 100644 --- a/server/internal/app/app.go +++ b/server/internal/app/app.go @@ -19,18 +19,20 @@ import ( ) type App struct { - Cfg *config.Config - DB *gorm.DB - Hasher *crypto.PasswordHasher - Enc *crypto.Encryptor - JWT *jwt.Manager - Usage *usage.Recorder - Health *channel.HealthMonitor - Limit *ratelimit.Limiter - Passkeys *passkey.Service - startedAt time.Time - ctx context.Context - cancel context.CancelFunc + Cfg *config.Config + DB *gorm.DB + Hasher *crypto.PasswordHasher + Enc *crypto.Encryptor + JWT *jwt.Manager + Usage *usage.Recorder + Health *channel.HealthMonitor + Limit *ratelimit.Limiter + Passkeys *passkey.Service + // usageRetention 用量明细保留期任务（未开启时为 no-op） + usageRetention func() + startedAt time.Time + ctx context.Context + cancel context.CancelFunc } func New(cfg *config.Config) (*App, error) { @@ -61,6 +63,9 @@ func New(cfg *config.Config) (*App, error) { }) a.Health.Start(a.ctx) + // 用量明细保留期（默认 0=永久保留，只清明细不动余额流水） + a.usageRetention = usage.StartRetention(a.ctx, db, cfg.Usage.RetentionDays) + a.Passkeys, err = passkey.New(db, passkey.Config{ RPID: cfg.WebAuthn.RPID, Origin: cfg.WebAuthn.RPOrigin, Name: cfg.WebAuthn.RPName, }) @@ -71,6 +76,9 @@ func New(cfg *config.Config) (*App, error) { } func (a *App) Close() { + if a.usageRetention != nil { + a.usageRetention() + } a.Usage.Close() } diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 3e5eb5e..8738c23 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -18,6 +18,7 @@ type Config struct { JWT JWTConfig Auth AuthConfig Proxy ProxyConfig + Usage UsageConfig RateLimit RateLimitConfig WebAuthn WebAuthnConfig Master string // 渠道密钥 AES-GCM 主密钥（来自环境变量） @@ -32,6 +33,13 @@ type Config struct { TrustedProxies []string } +// UsageConfig 用量数据的保留策略。 +type UsageConfig struct { + // RetentionDays 用量明细保留天数；0（默认）= 永久保留。 + // 只清 usage_logs，不动余额流水。 + RetentionDays int +} + // WebAuthnConfig Passkey（WebAuthn）配置。 type WebAuthnConfig struct { RPID string // Relying Party ID（域名，如 localhost） @@ -156,6 +164,7 @@ func Load() (*Config, error) { v.SetDefault("proxy.affinity_ttl", "5m") v.SetDefault("proxy.channel_acquire_wait", "150ms") v.SetDefault("proxy.auth_cache_ttl", "2s") + v.SetDefault("usage.retention_days", 0) v.SetDefault("proxy.health_interval", "60s") v.SetDefault("proxy.health_fail_threshold", 2) v.SetDefault("proxy.log_raw", false) @@ -213,6 +222,9 @@ func Load() (*Config, error) { HealthFailThreshold: v.GetInt("proxy.health_fail_threshold"), LogRaw: v.GetBool("proxy.log_raw"), }, + Usage: UsageConfig{ + RetentionDays: v.GetInt("usage.retention_days"), + }, RateLimit: RateLimitConfig{ UserRPS: v.GetInt("ratelimit.user_rps"), LoginIPPerMin: v.GetInt("ratelimit.login_ip_per_min"), diff --git a/server/internal/proxy/hold.go b/server/internal/proxy/hold.go index f67ae57..3bdafae 100644 --- a/server/internal/proxy/hold.go +++ b/server/internal/proxy/hold.go @@ -242,6 +242,11 @@ func sseEventHasContent(line []byte, proto string) bool { // 返回错误消息文本；不是错误事件时 ok=false。只在放行前判定——放行后错误 // 属于客户端可见的回复内容，不能拿去换渠道。 func sseUpstreamError(line []byte) (msg string, ok bool) { + // 廉价的子串预过滤：绝大多数事件与错误无关，先排除再谈解析。 + // 放行前每条事件都会过这里，长思考阶段的几十上百条事件不值得逐条反序列化。 + if !bytes.Contains(line, []byte(`"error"`)) && !bytes.Contains(line, []byte("response.failed")) { + return "", false + } s, ok := sseDataPayload(line) if !ok { return "", false diff --git a/server/internal/store/db.go b/server/internal/store/db.go index f090841..0a5bc81 100644 --- a/server/internal/store/db.go +++ b/server/internal/store/db.go @@ -27,7 +27,8 @@ import ( // idx_user_created 是 user_id 单列，光改 struct tag 不会重建，需删掉重迁一次。 // v7：usage_logs 加 reasoning_tokens / cache_write_1h_tokens / served_model 三列。 // v8：去掉 usage_logs 上没人查询的 key_id / trace_id 索引（每行两次多余的 B-tree 插入）。 -const schemaVersion = 8 +// v9：同样去掉没人查询的 idx_channel_created。 +const schemaVersion = 9 // Open 打开数据库连接并自动迁移。 // 开发默认 SQLite（自动追加 WAL 与 busy_timeout，见 withSQLitePragmas），生产可切 postgres。 @@ -137,7 +138,7 @@ func withSQLitePragmas(dsn string) string { // dropUnusedUsageLogIndexes 删除历史库里已经没用的索引：AutoMigrate 不会自动删索引， // 去掉 struct tag 只是不再新建，旧库上它们还会继续拖慢每次插入。 func dropUnusedUsageLogIndexes(db *gorm.DB) { - for _, name := range []string{"idx_usage_logs_key_id", "idx_usage_logs_trace_id"} { + for _, name := range []string{"idx_usage_logs_key_id", "idx_usage_logs_trace_id", "idx_channel_created"} { if err := db.Exec("DROP INDEX IF EXISTS " + name).Error; err != nil { log.Printf("store: drop index %s: %v", name, err) } diff --git a/server/internal/store/models.go b/server/internal/store/models.go index e987592..ac7b190 100644 --- a/server/internal/store/models.go +++ b/server/internal/store/models.go @@ -187,10 +187,12 @@ type UsageLog struct { RequestID string `gorm:"size:128" json:"request_id"` // 上游 request id // TraceID/KeyID 不建索引：没有查询按它们过滤（幂等检查走 balance_logs.ref_id）， // 每个索引都是每行一次 B-tree 插入，在记账这条最热的写路径上纯属写放大。 - TraceID string `gorm:"size:64" json:"trace_id"` - UserID uint64 `gorm:"index:idx_user_created,priority:1;not null" json:"user_id"` - KeyID uint64 `json:"key_id"` - ChannelID uint64 `gorm:"index:idx_channel_created,priority:1" json:"channel_id"` + TraceID string `gorm:"size:64" json:"trace_id"` + UserID uint64 `gorm:"index:idx_user_created,priority:1;not null" json:"user_id"` + KeyID uint64 `json:"key_id"` + // 与 key_id/trace_id 同理：没有任何查询按 channel_id 过滤 usage_logs + // （渠道统计走 created_at 时间窗后在内存里分组），复合索引纯属写放大。 + ChannelID uint64 `json:"channel_id"` ModelID uint64 `json:"model_id"` ModelName string `gorm:"size:128" json:"model_name"` Protocol string `gorm:"size:32" json:"protocol"` // responses|chat|messages @@ -210,8 +212,9 @@ type UsageLog struct { ErrorCode *string `json:"error_code,omitempty"` RawRequest string `gorm:"type:text" json:"raw_request"` // 客户端原始请求体（未转换） RawResponse string `gorm:"type:text" json:"raw_response"` // 上游原始响应（未转换；流式为全部 SSE 事件） - // CreatedAt 同时参与 user/channel 两个复合索引（时间窗过滤），自身单列索引保留给全表时间扫 - CreatedAt time.Time `gorm:"index;index:idx_user_created,priority:2;index:idx_channel_created,priority:2" json:"created_at"` + // CreatedAt 参与 (user, created) 复合索引（用户侧时间窗过滤），单列索引服务 + // 全表时间窗扫描（渠道统计、运营总览）。 + CreatedAt time.Time `gorm:"index;index:idx_user_created,priority:2" json:"created_at"` } // UsageDaily 日粒度预聚合（PLANNING §6.6） diff --git a/server/internal/usage/retention.go b/server/internal/usage/retention.go new file mode 100644 index 0000000..d86c69e --- /dev/null +++ b/server/internal/usage/retention.go @@ -0,0 +1,86 @@ +package usage + +import ( + "context" + "log" + "time" + + "gorm.io/gorm" +) + +// 记账明细的保留期清理。 +// +// usage_logs 只增不减：长期运行的实例上，渠道统计这类「按时间窗扫描」的查询会 +// 随表增长线性变慢，备份也会越来越大。余额流水（balance_logs）不在清理范围内—— +// 那是财务记录，需要长期保留。 +// +// 默认关闭（retention_days=0）：删数据是不可逆操作，由部署者显式开启。 +// 清理分批进行（每批 2000 行、批间小睡），避免长时间持锁影响在线请求。 + +const ( + retentionBatch = 2000 + retentionTick = 24 * time.Hour +) + +// StartRetention 周期性删除超过 days 天的用量明细；days<=0 时不启动。 +// 返回停止函数（进程退出时调用）。 +func StartRetention(ctx context.Context, db *gorm.DB, days int) (stop func()) { + if days <= 0 || db == nil { + return func() {} + } + done := make(chan struct{}) + go func() { + purgeOlderThan(ctx, db, days) + t := time.NewTicker(retentionTick) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-done: + return + case <-t.C: + purgeOlderThan(ctx, db, days) + } + } + }() + return func() { + select { + case <-done: + default: + close(done) + } + } +} + +// purgeOlderThan 分批删除 cutoff 之前的明细，返回删除行数。 +func purgeOlderThan(ctx context.Context, db *gorm.DB, days int) int64 { + cutoff := time.Now().UTC().AddDate(0, 0, -days) + var total int64 + for { + if ctx.Err() != nil { + return total + } + res := db.Exec( + "DELETE FROM usage_logs WHERE id IN (SELECT id FROM usage_logs WHERE created_at < ? LIMIT ?)", + cutoff, retentionBatch) + if res.Error != nil { + log.Printf("usage: retention purge failed: %v", res.Error) + return total + } + n := res.RowsAffected + total += n + if n < retentionBatch { + break + } + select { + case <-ctx.Done(): + return total + case <-time.After(50 * time.Millisecond): + } + } + if total > 0 { + log.Printf("usage: retention removed %d rows older than %s", total, cutoff.Format("2006-01-02")) + } + return total +} diff --git a/server/internal/usage/retention_test.go b/server/internal/usage/retention_test.go new file mode 100644 index 0000000..c151938 --- /dev/null +++ b/server/internal/usage/retention_test.go @@ -0,0 +1,84 @@ +package usage + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/openteam/server/internal/store" +) + +// 保留期只删过期明细，余额流水一条不动；未开启（0）时什么都不做。 +func TestRetentionPurgesOldLogsOnly(t *testing.T) { + db := openTestDB(t) + u := seedUser(t, db, 100) + + old := time.Now().UTC().AddDate(0, 0, -40) + fresh := time.Now().UTC() + for i := 0; i < 5; i++ { + db.Create(&store.UsageLog{TraceID: fmt.Sprintf("old-%d", i), UserID: u, ModelName: "m1", + Cost: 0.01, Status: store.UsageStatusSuccess, CreatedAt: old}) + } + for i := 0; i < 3; i++ { + db.Create(&store.UsageLog{TraceID: fmt.Sprintf("new-%d", i), UserID: u, ModelName: "m1", + Cost: 0.01, Status: store.UsageStatusSuccess, CreatedAt: fresh}) + } + db.Create(&store.BalanceLog{UserID: u, Change: -0.05, BalanceAfter: 99.95, + Type: store.BalanceTypeUsage, RefID: "usage:old-0"}) + + n := purgeOlderThan(context.Background(), db, 30) + if n != 5 { + t.Fatalf("应删除 5 条过期明细，实际 %d", n) + } + var logs, ledgers int64 + db.Model(&store.UsageLog{}).Count(&logs) + db.Model(&store.BalanceLog{}).Count(&ledgers) + if logs != 3 { + t.Fatalf("应保留 3 条新明细，实际 %d", logs) + } + if ledgers != 1 { + t.Fatalf("余额流水不应被清理，实际 %d 条", ledgers) + } +} + +// 未配置保留期时不启动、不删数据。 +func TestRetentionDisabled(t *testing.T) { + db := openTestDB(t) + u := seedUser(t, db, 100) + db.Create(&store.UsageLog{TraceID: "x", UserID: u, ModelName: "m1", + Cost: 0.01, Status: store.UsageStatusSuccess, + CreatedAt: time.Now().UTC().AddDate(0, 0, -400)}) + + stop := StartRetention(context.Background(), db, 0) + stop() + var n int64 + db.Model(&store.UsageLog{}).Count(&n) + if n != 1 { + t.Fatalf("未开启保留期时不应删除数据，实际剩 %d", n) + } +} + +// 分批删除：行数超过一批时应全部删完。 +func TestRetentionBatches(t *testing.T) { + db := openTestDB(t) + u := seedUser(t, db, 100) + old := time.Now().UTC().AddDate(0, 0, -90) + logs := make([]store.UsageLog, 0, retentionBatch+50) + for i := 0; i < retentionBatch+50; i++ { + logs = append(logs, store.UsageLog{TraceID: fmt.Sprintf("b-%d", i), UserID: u, + ModelName: "m1", Cost: 0.001, Status: store.UsageStatusSuccess, CreatedAt: old}) + } + if err := db.CreateInBatches(logs, 500).Error; err != nil { + t.Fatal(err) + } + n := purgeOlderThan(context.Background(), db, 30) + if n != int64(len(logs)) { + t.Fatalf("应删除 %d 条，实际 %d", len(logs), n) + } + var left int64 + db.Model(&store.UsageLog{}).Count(&left) + if left != 0 { + t.Fatalf("应全部删完，实际剩 %d", left) + } +}