diff --git a/.env.example b/.env.example index 74f6e59..87e7494 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,13 @@ OT_PROXY_TIMEOUT=300s # /v1 请求体上限（MB）：超限按客户端协议返回 413 OT_PROXY_MAX_BODY_MB=32 +# 故障转移给"第二个及以后的渠道"留的时间预算（0=不限）。 +# 首个渠道不受它约束——长回答本就要跑几分钟，但不能让坏渠道把客户端拖到 N×timeout。 +OT_PROXY_FAILOVER_BUDGET=30s +# 流式空闲多久补一行 SSE 注释心跳（0=关闭）。挂在 nginx/Cloudflare 后面时， +# 上游长时间不出字（长思考）会被对端的空闲超时掐掉，心跳用来续住这条流。 +OT_PROXY_STREAM_HEARTBEAT=15s + # 渠道健康检查 OT_PROXY_HEALTH_INTERVAL=60s OT_PROXY_HEALTH_FAIL_THRESHOLD=2 diff --git a/README.md b/README.md index ed1e574..7885a1d 100644 --- a/README.md +++ b/README.md @@ -10,12 +10,15 @@ - `POST /v1/chat/completions` — OpenAI Chat（非流式 + 流式 SSE） - `POST /v1/responses` — OpenAI Responses API（非流式 + 流式事件） - `POST /v1/messages` — Anthropic Messages API（非流式 + 流式事件） - - `GET /v1/models` — 可用模型列表 + - `GET /v1/models` — 可用模型列表：只列调用方真能用的模型（叠加用户与密钥白名单），带上下文长度与能力位；Anthropic 客户端（带 `x-api-key`/`anthropic-version`）返回 Anthropic 形状 - **三协议互转**：客户端协议 × 渠道协议不匹配时自动转换（如 Chat 调用 Claude、Messages 调用 OpenAI、Responses 调用 Claude），流式逐事件转换；协议匹配时直通 - 错误按客户端协议返回（OpenAI 格式 / Anthropic 格式） - **渠道体系**：按模型绑定选渠道 + 加权负载均衡；模型已有绑定时只用绑定渠道（都不可用则 503，不会把请求发给不认识它的渠道）；每渠道并发信号量（满载溢出）；后台健康检查（探测 `/models`，连续失败进 cooldown、恢复放回）+ 运行期健康（连续 3 次网络错误/超时/429/5xx 进 30s 冷却，成功即清零）；可安全重试的失败自动故障转移（网络错误/429/5xx/超时且未写出响应头） - **上游超时**：渠道 `timeout_ms` 生效——非流式为整段请求上限，流式仅约束首包（正文不限时长）；上游请求绑定客户端 context，断连即取消 +- **故障转移与流式细节**：上游 429 的 `Retry-After` 会延长该渠道冷却（夹在 10 分钟内）；重试受 `OT_PROXY_FAILOVER_BUDGET` 时间预算约束（首个渠道不受限，避免坏渠道把客户端拖到 N×timeout）；流式空闲超过 `OT_PROXY_STREAM_HEARTBEAT` 补一行 SSE 注释心跳，防止 nginx/Cloudflare 的空闲超时掐掉长思考 +- **渠道观测**：`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） - **认证面防护**：登录按「单 IP」+「单 IP×账号」双桶滑动窗口限速、注册按 IP 限速（429 + `Retry-After`）；`/v1` 请求体上限 `OT_PROXY_MAX_BODY_MB`（超限按客户端协议返回 413）；CORS 改为显式 origin 白名单（默认不发跨域头）；`env=production` 启动时断言 JWT secret / 主密钥 / 管理员密码不能是仓库默认值，否则拒绝启动 - **前端**：Dashboard / 管理总览骨架屏加载态 - **用户体系**：注册（开放/邀请码可切换，管理后台可改）、登录（JWT access + HttpOnly refresh cookie）、argon2id 密码 diff --git a/server/internal/api/admin_models.go b/server/internal/api/admin_models.go index 755d469..e2b949a 100644 --- a/server/internal/api/admin_models.go +++ b/server/internal/api/admin_models.go @@ -43,7 +43,10 @@ func (h *Handler) AdminModels(c *gin.Context) { out = append(out, gin.H{ "id": m.ID, "name": m.Name, "input_price": m.InputPrice, "output_price": m.OutputPrice, "cache_read_price": m.CacheReadPrice, - "enabled": m.Enabled, "sort": m.Sort, "channels": chs, + "cache_write_price": m.CacheWritePrice, "context_length": m.ContextLength, + "supports_tools": m.SupportsTools, "supports_vision": m.SupportsVision, + "supports_reasoning": m.SupportsReasoning, + "enabled": m.Enabled, "sort": m.Sort, "channels": chs, "used": used, "needs_pricing": needsPricing, "denied": denied, }) } @@ -118,20 +121,28 @@ func containsStr(list []string, s string) bool { // AdminCreateModel POST /api/v1/admin/models func (h *Handler) AdminCreateModel(c *gin.Context) { var req struct { - Name string `json:"name" binding:"required,min=1,max=128"` - InputPrice float64 `json:"input_price"` - OutputPrice float64 `json:"output_price"` - CacheReadPrice float64 `json:"cache_read_price"` - Enabled *bool `json:"enabled"` + Name string `json:"name" binding:"required,min=1,max=128"` + InputPrice float64 `json:"input_price"` + OutputPrice float64 `json:"output_price"` + CacheReadPrice float64 `json:"cache_read_price"` + CacheWritePrice float64 `json:"cache_write_price"` + ContextLength int `json:"context_length"` + SupportsTools bool `json:"supports_tools"` + SupportsVision bool `json:"supports_vision"` + SupportsReasoning bool `json:"supports_reasoning"` + Enabled *bool `json:"enabled"` } if err := c.ShouldBindJSON(&req); err != nil { resp.Fail(c, http.StatusBadRequest, "invalid input: "+err.Error()) return } m := store.Model{ - Name: req.Name, + Name: req.Name, InputPrice: req.InputPrice, OutputPrice: req.OutputPrice, CacheReadPrice: req.CacheReadPrice, - Enabled: boolOr(req.Enabled, true), + CacheWritePrice: req.CacheWritePrice, ContextLength: req.ContextLength, + SupportsTools: req.SupportsTools, SupportsVision: req.SupportsVision, + SupportsReasoning: req.SupportsReasoning, + Enabled: boolOr(req.Enabled, true), } if err := h.a.DB.Create(&m).Error; err != nil { resp.Fail(c, http.StatusConflict, "failed to create model (name may already exist)") @@ -148,11 +159,16 @@ func (h *Handler) AdminUpdateModel(c *gin.Context) { return } var req struct { - InputPrice *float64 `json:"input_price"` - OutputPrice *float64 `json:"output_price"` - CacheReadPrice *float64 `json:"cache_read_price"` - Enabled *bool `json:"enabled"` - Sort *int `json:"sort"` + InputPrice *float64 `json:"input_price"` + OutputPrice *float64 `json:"output_price"` + CacheReadPrice *float64 `json:"cache_read_price"` + CacheWritePrice *float64 `json:"cache_write_price"` + ContextLength *int `json:"context_length"` + SupportsTools *bool `json:"supports_tools"` + SupportsVision *bool `json:"supports_vision"` + SupportsReasoning *bool `json:"supports_reasoning"` + Enabled *bool `json:"enabled"` + Sort *int `json:"sort"` } if err := c.ShouldBindJSON(&req); err != nil { resp.Fail(c, http.StatusBadRequest, "invalid input") @@ -173,6 +189,21 @@ func (h *Handler) AdminUpdateModel(c *gin.Context) { if req.CacheReadPrice != nil { updates["cache_read_price"] = *req.CacheReadPrice } + if req.CacheWritePrice != nil { + updates["cache_write_price"] = *req.CacheWritePrice + } + if req.ContextLength != nil { + updates["context_length"] = *req.ContextLength + } + if req.SupportsTools != nil { + updates["supports_tools"] = *req.SupportsTools + } + if req.SupportsVision != nil { + updates["supports_vision"] = *req.SupportsVision + } + if req.SupportsReasoning != nil { + updates["supports_reasoning"] = *req.SupportsReasoning + } if req.Enabled != nil { updates["enabled"] = *req.Enabled } @@ -184,8 +215,20 @@ func (h *Handler) AdminUpdateModel(c *gin.Context) { resp.Fail(c, http.StatusInternalServerError, "failed to update model") return } + if err := h.a.DB.First(&m, id).Error; err != nil { + resp.Fail(c, http.StatusInternalServerError, "failed to reload model") + return + } } - resp.OK(c, gin.H{"ok": true}) + resp.OK(c, gin.H{ + "id": m.ID, "name": m.Name, + "input_price": m.InputPrice, "output_price": m.OutputPrice, + "cache_read_price": m.CacheReadPrice, "cache_write_price": m.CacheWritePrice, + "context_length": m.ContextLength, + "supports_tools": m.SupportsTools, "supports_vision": m.SupportsVision, + "supports_reasoning": m.SupportsReasoning, + "enabled": m.Enabled, "sort": m.Sort, + }) } // AdminDeleteModel DELETE /api/v1/admin/models/:id @@ -288,4 +331,4 @@ func (h *Handler) AdminDeleteModelBinding(c *gin.Context) { } var _ = errors.Is -var _ = gorm.ErrRecordNotFound \ No newline at end of file +var _ = gorm.ErrRecordNotFound diff --git a/server/internal/api/admin_stats.go b/server/internal/api/admin_stats.go index 4eb1ed0..c4804ea 100644 --- a/server/internal/api/admin_stats.go +++ b/server/internal/api/admin_stats.go @@ -1,6 +1,8 @@ package api import ( + "math" + "sort" "strconv" "time" @@ -53,12 +55,150 @@ func (h *Handler) AdminStatsOverview(c *gin.Context) { 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}, + "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")) diff --git a/server/internal/api/router.go b/server/internal/api/router.go index 7a5f4fd..72db43c 100644 --- a/server/internal/api/router.go +++ b/server/internal/api/router.go @@ -114,6 +114,7 @@ func NewRouter(a *app.App, gw *proxy.Gateway) *gin.Engine { admin.POST("/users/:id/balance", h.AdminAdjustBalance) // 渠道 admin.GET("/channels", h.AdminChannels) + admin.GET("/channels/stats", h.AdminChannelStats) admin.POST("/channels", h.AdminCreateChannel) admin.PUT("/channels/:id", h.AdminUpdateChannel) admin.DELETE("/channels/:id", h.AdminDeleteChannel) diff --git a/server/internal/channel/channel.go b/server/internal/channel/channel.go index 7726877..a6cd9d8 100644 --- a/server/internal/channel/channel.go +++ b/server/internal/channel/channel.go @@ -17,6 +17,9 @@ import ( var ErrNoChannel = errors.New("no available channel") +// maxCooldown 运行期冷却上限：上游 Retry-After 再长也不让渠道缺席超过这个时间。 +const maxCooldown = 10 * time.Minute + type Service struct { db *gorm.DB enc *crypto.Encryptor @@ -186,7 +189,16 @@ func (s *Service) UpstreamKey(ch *store.Channel) (string, error) { // ReportFailure 记一次可重试的上游失败；连续达 failThreshold 次则该渠道进入冷却， // 期间不出现在候选集里。冷却到期后需再攒满阈值才会被剔除。 -func (s *Service) ReportFailure(ch *store.Channel) { +// cooldown>0 时用它覆盖默认时长（上游 429 带的 Retry-After），并夹到 maxCooldown +// ——不能让一个写错的重试头把渠道钉死几小时。 +func (s *Service) ReportFailure(ch *store.Channel, cooldown time.Duration) { + if cooldown > s.cooldownFor { + if cooldown > maxCooldown { + cooldown = maxCooldown + } + } else { + cooldown = s.cooldownFor + } s.mu.Lock() st := s.runt[ch.ID] if st == nil { @@ -197,12 +209,12 @@ func (s *Service) ReportFailure(ch *store.Channel) { tripped := false if st.fails >= s.failThreshold { st.fails = 0 - st.until = time.Now().Add(s.cooldownFor) + st.until = time.Now().Add(cooldown) tripped = true } s.mu.Unlock() if tripped { - log.Printf("channel: %q entered runtime cooldown (%s)", ch.Name, s.cooldownFor) + log.Printf("channel: %q entered runtime cooldown (%s)", ch.Name, cooldown) } } diff --git a/server/internal/channel/channel_test.go b/server/internal/channel/channel_test.go index 7f7bbdc..45d5e8b 100644 --- a/server/internal/channel/channel_test.go +++ b/server/internal/channel/channel_test.go @@ -174,12 +174,12 @@ func TestRuntimeCooldownThresholdAndExpiry(t *testing.T) { s.cooldownFor = 60 * time.Millisecond // 默认阈值 3：前两次失败不该踢，第三次才进冷却 - s.ReportFailure(&ch) - s.ReportFailure(&ch) + s.ReportFailure(&ch, 0) + s.ReportFailure(&ch, 0) if s.InCooldown(id) { t.Fatal("未达阈值（3 次连续）不应进冷却") } - s.ReportFailure(&ch) + s.ReportFailure(&ch, 0) if !s.InCooldown(id) { t.Fatal("连续达阈值应进冷却") } @@ -198,9 +198,9 @@ func TestRuntimeCooldownThresholdAndExpiry(t *testing.T) { t.Fatalf("恢复后应重新进候选，实际 %d", len(got)) } - s.ReportFailure(&ch) + s.ReportFailure(&ch, 0) s.ReportSuccess(&ch) - s.ReportFailure(&ch) + s.ReportFailure(&ch, 0) if s.InCooldown(id) { t.Fatal("非连续失败不应触发冷却") } @@ -217,12 +217,12 @@ func TestRuntimeStateAndClear(t *testing.T) { if fails, rem := s.RuntimeState(id); fails != 0 || rem != 0 { t.Fatalf("初始状态 = %d/%s，期望 0/0", fails, rem) } - s.ReportFailure(&ch) - s.ReportFailure(&ch) + s.ReportFailure(&ch, 0) + s.ReportFailure(&ch, 0) if fails, rem := s.RuntimeState(id); fails != 2 || rem != 0 { t.Fatalf("两次失败后 = %d/%s，期望 2/0", fails, rem) } - s.ReportFailure(&ch) + s.ReportFailure(&ch, 0) fails, rem := s.RuntimeState(id) if fails != 0 || rem <= 0 || rem > s.cooldownFor { t.Fatalf("冷却中 = %d/%s，期望 0/(0,%s]", fails, rem, s.cooldownFor) @@ -242,3 +242,38 @@ func TestRuntimeStateAndClear(t *testing.T) { t.Fatalf("恢复后应重新进候选，实际 %d", got) } } + +// 上游 429 带的 Retry-After 可以拉长冷却，但要有上限，不能被钉死几小时。 +func TestReportFailureCooldownOverrideAndClamp(t *testing.T) { + db := newTestDB(t) + id := seedChannel(t, db, "ra", 1, 0, 4) + var ch store.Channel + db.First(&ch, id) + + s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890")) + s.cooldownFor = 30 * time.Second + + for i := 0; i < 3; i++ { + s.ReportFailure(&ch, 2*time.Minute) + } + if _, rem := s.RuntimeState(id); rem < 115*time.Second || rem > 2*time.Minute { + t.Fatalf("应按 Retry-After 冷却 2min，实际剩余 %s", rem) + } + + s.ClearRuntime(id) + for i := 0; i < 3; i++ { + s.ReportFailure(&ch, time.Hour) + } + if _, rem := s.RuntimeState(id); rem > maxCooldown { + t.Fatalf("冷却应夹到上限 %s，实际 %s", maxCooldown, rem) + } + + // 比默认值短的 Retry-After 不缩短冷却（否则 1s 就把坏渠道放回轮转） + s.ClearRuntime(id) + for i := 0; i < 3; i++ { + s.ReportFailure(&ch, 5*time.Second) + } + if _, rem := s.RuntimeState(id); rem < 25*time.Second { + t.Fatalf("短 Retry-After 不应缩短默认冷却，实际 %s", rem) + } +} diff --git a/server/internal/cli/serve.go b/server/internal/cli/serve.go index 1c8defd..661dfa4 100644 --- a/server/internal/cli/serve.go +++ b/server/internal/cli/serve.go @@ -31,8 +31,14 @@ func serve() error { } defer a.Shutdown(context.Background()) - gw := proxy.NewGateway(a.DB, a.Enc, a.Usage, a.Limit, cfg.RateLimit.UserRPS, cfg.Proxy.LogRaw, - cfg.Proxy.Timeout, cfg.Proxy.MaxBodyMB) + gw := proxy.NewGateway(a.DB, a.Enc, a.Usage, a.Limit, proxy.Options{ + UserRPS: cfg.RateLimit.UserRPS, + LogRaw: cfg.Proxy.LogRaw, + GWTimeout: cfg.Proxy.Timeout, + MaxBodyMB: cfg.Proxy.MaxBodyMB, + FailoverBudget: cfg.Proxy.FailoverBudget, + StreamHeartbeat: cfg.Proxy.StreamHeartbeat, + }) router := api.NewRouter(a, gw) srv := &http.Server{ diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 467a046..bd99e84 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -79,6 +79,8 @@ type ProxyConfig struct { DefaultModel string // 渠道模型导入时使用的模型名 Timeout time.Duration MaxBodyMB int // /v1 请求体上限（MB），超限按客户端协议返回 413 + FailoverBudget time.Duration // 故障转移给"第二个及以后的渠道"留的时间预算，0=不限 + StreamHeartbeat time.Duration // 流式空闲多久发一次 SSE 注释心跳，0=关闭 HealthInterval time.Duration // 渠道健康检查周期 HealthFailThreshold int // 连续失败 N 次进 cooldown LogRaw bool // 记录管理员原始请求体+响应到 usage_logs（调试用，默认关） @@ -144,6 +146,8 @@ func Load() (*Config, error) { v.SetDefault("proxy.default_model", "gpt-4o-mini") v.SetDefault("proxy.timeout", "300s") v.SetDefault("proxy.max_body_mb", 32) + v.SetDefault("proxy.failover_budget", "30s") + v.SetDefault("proxy.stream_heartbeat", "15s") v.SetDefault("proxy.health_interval", "60s") v.SetDefault("proxy.health_fail_threshold", 2) v.SetDefault("proxy.log_raw", false) @@ -191,6 +195,8 @@ func Load() (*Config, error) { DefaultModel: v.GetString("proxy.default_model"), Timeout: v.GetDuration("proxy.timeout"), MaxBodyMB: v.GetInt("proxy.max_body_mb"), + FailoverBudget: v.GetDuration("proxy.failover_budget"), + StreamHeartbeat: v.GetDuration("proxy.stream_heartbeat"), HealthInterval: v.GetDuration("proxy.health_interval"), HealthFailThreshold: v.GetInt("proxy.health_fail_threshold"), LogRaw: v.GetBool("proxy.log_raw"), diff --git a/server/internal/proxy/gateway.go b/server/internal/proxy/gateway.go index 9b368cb..9dff553 100644 --- a/server/internal/proxy/gateway.go +++ b/server/internal/proxy/gateway.go @@ -42,8 +42,13 @@ type Gateway struct { // 上游超时兜底值（渠道 timeout_ms 非法时使用）；按超时值缓存 Transport 复用连接池。 gwTimeout time.Duration maxBodyMB int // /v1 请求体上限（MB），仅用于超限报错文案 - tpMu sync.Mutex - tps map[time.Duration]*http.Transport + // 故障转移给"第二个及以后的渠道"留的时间预算；超了就直接返回最后一个错误， + // 不限制首个渠道（长回答本来就可能跑很久）。 + failoverBudget time.Duration + // 流式空闲多久补一行 SSE 注释心跳（0=关闭）。反向代理的空闲超时会掐掉长思考的流。 + streamHeartbeat time.Duration + tpMu sync.Mutex + tps map[time.Duration]*http.Transport policyMu sync.Mutex policy modelPolicy @@ -132,18 +137,31 @@ func contains(list []string, s string) bool { return false } -func NewGateway(db *gorm.DB, enc *crypto.Encryptor, rec *usage.Recorder, lim *ratelimit.Limiter, userRPS int, logRaw bool, gw_timeout time.Duration, maxBodyMB int) *Gateway { +// Options 网关配置项。用结构体而不是十个位置参数：里面三个 time.Duration、 +// 两个 int，传错的编译期发现不了。 +type Options struct { + UserRPS int + LogRaw bool + GWTimeout time.Duration // 上游超时兜底（渠道 timeout_ms 优先） + MaxBodyMB int // /v1 请求体上限（MB） + FailoverBudget time.Duration // 故障转移重试的时间预算 + StreamHeartbeat time.Duration // 流式空闲心跳间隔 +} + +func NewGateway(db *gorm.DB, enc *crypto.Encryptor, rec *usage.Recorder, lim *ratelimit.Limiter, o Options) *Gateway { return &Gateway{ - db: db, - ch: channel.NewService(db, enc), - rec: rec, - enc: enc, - lim: lim, - userRPS: userRPS, - logRaw: logRaw, - gwTimeout: gw_timeout, - maxBodyMB: maxBodyMB, - tps: map[time.Duration]*http.Transport{}, + db: db, + ch: channel.NewService(db, enc), + rec: rec, + enc: enc, + lim: lim, + userRPS: o.UserRPS, + logRaw: o.LogRaw, + gwTimeout: o.GWTimeout, + maxBodyMB: o.MaxBodyMB, + failoverBudget: o.FailoverBudget, + streamHeartbeat: o.StreamHeartbeat, + tps: map[time.Duration]*http.Transport{}, } } @@ -194,6 +212,14 @@ func (g *Gateway) Auth(c *gin.Context) { c.Set("protocol", convert.ProtoMessages) case "/v1/responses": c.Set("protocol", convert.ProtoResponses) + case "/v1/models": + // 模型目录两条协议共用同一路径，只能靠头区分：Anthropic 客户端带 + // x-api-key / anthropic-version，OpenAI 客户端从不发这两个头。 + if c.GetHeader("x-api-key") != "" || c.GetHeader("anthropic-version") != "" { + c.Set("protocol", convert.ProtoMessages) + } else { + c.Set("protocol", convert.ProtoChat) + } default: c.Set("protocol", convert.ProtoChat) } @@ -272,11 +298,18 @@ func (g *Gateway) Handle(c *gin.Context) { } } -// models GET /v1/models：返回对外可见的模型（启用且至少绑定到一个启用且健康的渠道；OpenAI 风格）。 +// models GET /v1/models：返回调用方真正能用的模型。 +// 口径与选路一致（启用 + 至少绑定一个启用且健康的渠道），再叠加该用户/密钥的模型白名单—— +// 否则被限制的模型名照样泄漏在列表里，客户端试了才知道 403。 +// 按客户端协议返回对应形状，并带上目录元数据（上下文长度、能力位）；价格不外泄。 func (g *Gateway) models(c *gin.Context) { + u, ok := g.resolveUser(c) + if !ok { + return + } ids := g.ch.AvailableModelIDs() if len(ids) == 0 { - c.JSON(http.StatusOK, gin.H{"object": "list", "data": []gin.H{}}) + g.writeModelList(c, nil) return } var ms []store.Model @@ -284,14 +317,48 @@ func (g *Gateway) models(c *gin.Context) { apiError(c, http.StatusInternalServerError, "internal_error", "failed to load models") return } + out := make([]store.Model, 0, len(ms)) + for _, m := range ms { + if g.checkModelAllowed(c, u, m.Name) { + out = append(out, m) + } + } + g.writeModelList(c, out) +} + +func (g *Gateway) writeModelList(c *gin.Context, ms []store.Model) { + if p, _ := c.Get("protocol"); p == convert.ProtoMessages { + data := make([]gin.H, 0, len(ms)) + for _, m := range ms { + data = append(data, gin.H{ + "type": "model", + "id": m.Name, + "display_name": m.Name, + "created_at": m.CreatedAt.UTC().Format(time.RFC3339), + "context_length": m.ContextLength, + "capabilities": gin.H{ + "tools": m.SupportsTools, "vision": m.SupportsVision, "reasoning": m.SupportsReasoning, + }, + }) + } + c.JSON(http.StatusOK, gin.H{"data": data, "has_more": false}) + return + } data := make([]gin.H, 0, len(ms)) for _, m := range ms { - data = append(data, gin.H{ - "id": m.Name, - "object": "model", - "created": m.CreatedAt.Unix(), - "owned_by": "openteam", - }) + item := gin.H{ + "id": m.Name, "object": "model", + "created": m.CreatedAt.Unix(), "owned_by": "openteam", + } + if m.ContextLength > 0 { + item["context_length"] = m.ContextLength + } + if m.SupportsTools || m.SupportsVision || m.SupportsReasoning { + item["capabilities"] = gin.H{ + "tools": m.SupportsTools, "vision": m.SupportsVision, "reasoning": m.SupportsReasoning, + } + } + data = append(data, item) } c.JSON(http.StatusOK, gin.H{"object": "list", "data": data}) } diff --git a/server/internal/proxy/passthrough.go b/server/internal/proxy/passthrough.go index 1938948..fc59134 100644 --- a/server/internal/proxy/passthrough.go +++ b/server/internal/proxy/passthrough.go @@ -13,6 +13,8 @@ import ( "net/http" "strconv" "strings" + "sync" + "sync/atomic" "time" "github.com/gin-gonic/gin" @@ -160,6 +162,27 @@ func sseContentText(line []byte) string { return "" } +// retryAfter 解析上游 Retry-After（delta-秒或 HTTP 日期），用于按对端意愿延长渠道冷却。 +// 解析不出或为非正返回 0，交给 channel.Service 用默认时长（并夹到上限）。 +func retryAfter(h http.Header) time.Duration { + v := strings.TrimSpace(h.Get("Retry-After")) + if v == "" { + return 0 + } + if secs, err := strconv.Atoi(v); err == nil { + if secs <= 0 { + return 0 + } + return time.Duration(secs) * time.Second + } + if t, err := http.ParseTime(v); err == nil { + if d := time.Until(t); d > 0 { + return d + } + } + return 0 +} + // upstreamURL 组装上游地址：按协议选 base_url 再拼资源路径（见 store.Channel.UpstreamURL）。 func upstreamURL(ch *store.Channel, proto, path string) string { return ch.UpstreamURL(proto, path) @@ -184,8 +207,18 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto var lastStatus = http.StatusBadGateway var lastBody = []byte("all upstream channels failed") + // 最后真正试过的渠道：全部失败时把错误归到它头上，否则 usage_logs 里 + // channel_id=0，渠道维度的统计与排障就看不到这类失败。 + var lastCh *store.Channel + startAll := time.Now() // 从选中的渠道开始遍历，到末尾后再从头遍历到选中渠道之前 for offset := 0; offset < len(cands); offset++ { + // 重试预算只约束第二个及以后的渠道：首个渠道不受限（长回答本就要跑几分钟）， + // 但坏渠道一个接一个地等下去会让客户端挂到 N×timeout。 + if offset > 0 && g.failoverBudget > 0 && time.Since(startAll) > g.failoverBudget { + lastBody = []byte("failover budget exhausted: " + string(lastBody)) + break + } idx := (startIdx + offset) % len(cands) cand := cands[idx] ch := cand.Channel @@ -207,6 +240,8 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto if !ok { continue // 渠道满载，溢出到下一个 } + // 失败/成功计数由 proxyOne 就地报：只有它读得到上游响应头（429 的 Retry-After） + lastCh = ch done, retry, st, b := g.proxyOne(c, ch, plan, stream, sink) release() if done { @@ -219,12 +254,11 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto g.recordError(c, ch, nil, now(), "upstream_http_"+strconv.Itoa(st)) return } - g.ch.ReportFailure(ch) // 网络错误/超时/429/5xx：累计运行期失败 lastStatus, lastBody = st, b } // 全部候选重试性失败 apiError(c, lastStatus, "upstream_error", string(lastBody)) - g.recordError(c, nil, nil, now(), "all_channels_failed") + g.recordError(c, lastCh, nil, now(), "all_channels_failed") } // proxyOne 对单个渠道执行一次代理。 @@ -233,6 +267,7 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan, stream bool, sink *usageSink) (done bool, retry bool, status int, respBody []byte) { upKey, err := g.ch.UpstreamKey(ch) if err != nil { + g.ch.ReportFailure(ch, 0) return false, true, http.StatusInternalServerError, []byte("failed to decrypt channel key") } @@ -286,6 +321,7 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan g.finishUsage(c, ch, start, store.UsageStatusCanceled, "client_disconnect") return true, false, 0, nil } + g.ch.ReportFailure(ch, 0) // 网络错误/超时 return false, true, http.StatusBadGateway, []byte("upstream request failed: " + err.Error()) } defer resp.Body.Close() @@ -294,6 +330,7 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan if resp.StatusCode < 200 || resp.StatusCode >= 300 { errBody, _ := io.ReadAll(resp.Body) if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { + g.ch.ReportFailure(ch, retryAfter(resp.Header)) return false, true, resp.StatusCode, errBody } return false, false, resp.StatusCode, errBody @@ -360,6 +397,11 @@ func (g *Gateway) streamCopy(c *gin.Context, ch *store.Channel, r io.Reader, sta } } + lw := &lockedWriter{w: w, f: flusher} + lw.lastN.Store(time.Now().UnixNano()) + stopHB := lw.heartbeat(g.streamHeartbeat) + defer stopHB() + scanner := newSSEScanner(r) for { line, err := scanner.Next() @@ -372,13 +414,12 @@ func (g *Gateway) streamCopy(c *gin.Context, ch *store.Channel, r io.Reader, sta out = lineConv(line) } if out != nil { - if _, werr := w.Write(out); werr != nil { + if _, werr := lw.write(out); werr != nil { // 客户端意外断开：按已生成部分收费（canceled） commitRaw() g.finishUsage(c, ch, start, store.UsageStatusCanceled, "client_disconnect") return } - flusher.Flush() if sink != nil { sink.outputText += sseContentText(out) } @@ -406,6 +447,54 @@ type nopFlusher struct{} func (nopFlusher) Flush() {} +// lockedWriter 给流式写出加锁：心跳在独立 goroutine 里写，而 gin 的 ResponseWriter +// 不是并发安全的，两路同时写会坏帧。 +type lockedWriter struct { + mu sync.Mutex + w io.Writer + f http.Flusher + lastN atomic.Int64 // 最近一次写出时间（unix nano），心跳据此判断空闲 +} + +func (lw *lockedWriter) write(b []byte) (int, error) { + lw.mu.Lock() + defer lw.mu.Unlock() + n, err := lw.w.Write(b) + lw.lastN.Store(time.Now().UnixNano()) + if err == nil { + lw.f.Flush() + } + return n, err +} + +// heartbeat 距上次写出超过 interval 时补一行 SSE 注释（`: ping`），返回停止函数。 +// 挂在 nginx/Cloudflare 后面时长思考（首包后上游安静几十秒）会被空闲超时掐掉， +// 客户端只看到流断在半路。注释行按 SSE 规范被解析器忽略，不影响计费与转换。 +func (lw *lockedWriter) heartbeat(interval time.Duration) func() { + if interval <= 0 { + return func() {} + } + done := make(chan struct{}) + var once sync.Once + go func() { + tk := time.NewTicker(interval) + defer tk.Stop() + for { + select { + case <-done: + return + case <-tk.C: + if time.Since(time.Unix(0, lw.lastN.Load())) < interval { + continue + } + // 心跳写失败说明客户端已断开，交给主循环下一次写出时统一记账 + _, _ = lw.write([]byte(": ping\n\n")) + } + } + }() + return func() { once.Do(func() { close(done) }) } +} + // --------------------------------------------------------------------------- // usage 提取 @@ -627,11 +716,11 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time modelID = model.ID // 计价口径：in=非缓存输入、cacheRead=缓存读、cacheCreate=缓存写（Anthropic 语义， // messages 流式 message_delta 的 input_tokens 即非缓存部分）。 - // 缓存写按 1.25× 输入价（Anthropic 5m 口径）。 + // 缓存写价未单独配置时按输入价 1.25×（Anthropic 5m 口径）。 cost = float64(in)/1e6*model.InputPrice + float64(out)/1e6*model.OutputPrice + float64(cacheRead)/1e6*model.CacheReadPrice + - float64(cacheCreate)/1e6*model.InputPrice*1.25 + float64(cacheCreate)/1e6*model.CacheWritePriceOr() } else { cost = float64(in+cacheRead+cacheCreate)/1e6*0.15 + float64(out)/1e6*0.60 // 无定价模型时按示例价 } diff --git a/server/internal/proxy/proxy_test.go b/server/internal/proxy/proxy_test.go index 93b2843..68e9c1c 100644 --- a/server/internal/proxy/proxy_test.go +++ b/server/internal/proxy/proxy_test.go @@ -76,7 +76,12 @@ func newHarness(t *testing.T, baseURL string, timeoutMS int, bind bool, keyModel rec := usage.NewRecorder(db) t.Cleanup(rec.Close) return &harness{ - gw: NewGateway(db, enc, rec, ratelimit.New(), 0, false, 300*time.Second, 32), + gw: NewGateway(db, enc, rec, ratelimit.New(), Options{ + GWTimeout: 300 * time.Second, + MaxBodyMB: 32, + FailoverBudget: 30 * time.Second, + StreamHeartbeat: 15 * time.Second, + }), db: db, rec: rec, key: plain, } } @@ -294,7 +299,7 @@ func TestRuntimeFailureCounterResetsOnSuccess(t *testing.T) { // 回归：全局兜底超时值不再被二次换算。 func TestGatewayTimeoutIsNotMultiplied(t *testing.T) { - gw := NewGateway(nil, nil, nil, nil, 0, false, 30*time.Second, 8) + gw := NewGateway(nil, nil, nil, nil, Options{GWTimeout: 30 * time.Second, MaxBodyMB: 8}) if got := gw.channelTimeout(&store.Channel{TimeoutMS: 0}); got != 30*time.Second { t.Fatalf("兜底超时 = %s，期望 30s", got) } @@ -333,3 +338,270 @@ func TestRequestBodyTooLarge(t *testing.T) { t.Fatalf("报错应点名上限，实际：%s", w.Body.String()) } } + +// 429 的 Retry-After 要能读出来（数字与 HTTP 日期两种写法）。 +func TestRetryAfterParsing(t *testing.T) { + cases := []struct { + val string + want time.Duration + }{ + {"", 0}, + {"7", 7 * time.Second}, + {" 12 ", 12 * time.Second}, + {"0", 0}, + {"-5", 0}, + {"not-a-number", 0}, + } + for _, tc := range cases { + h := http.Header{} + if tc.val != "" { + h.Set("Retry-After", tc.val) + } + if got := retryAfter(h); got != tc.want { + t.Fatalf("Retry-After=%q 解析成 %s，期望 %s", tc.val, got, tc.want) + } + } + h := http.Header{} + h.Set("Retry-After", time.Now().UTC().Add(20*time.Second).Format(http.TimeFormat)) + if got := retryAfter(h); got < 15*time.Second || got > 25*time.Second { + t.Fatalf("HTTP 日期写法解析异常：%s", got) + } +} + +// 故障转移的时间预算只约束第二个及以后的渠道：坏渠道把预算吃光后不再拖着客户端等下一条。 +func TestFailoverBudgetStopsRetrying(t *testing.T) { + var hitFast atomic.Int64 + slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(1200 * time.Millisecond) // 慢且最终失败，专门吃预算 + w.WriteHeader(http.StatusBadGateway) + _, _ = io.WriteString(w, `{"error":{"message":"boom"}}`) + })) + defer slow.Close() + fast := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hitFast.Add(1) + _, _ = io.WriteString(w, `{"id":"ok","choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`) + })) + defer fast.Close() + + h := newHarness(t, slow.URL, 0, true, nil) // 渠道1：慢坏 + // 加权随机决定起点，把权重压到一边，否则起点可能直接落在健康渠道上、测不到重试预算 + if err := h.db.Model(&store.Channel{}).Where("name = ?", "c1").Update("weight", 1000).Error; err != nil { + t.Fatal(err) + } + // 渠道2：健康，绑定同一个模型 + encKey, _ := crypto.NewEncryptor("test-master-key-1234567890").Encrypt("upstream-key") + var m store.Model + h.db.Where("name = ?", "m1").First(&m) + var ch2 store.Channel + ch2 = store.Channel{Name: "c2", Provider: store.ChannelProviderOpenAI, + Formats: []string{store.FormatChat}, BaseURL: fast.URL, APIKeyEnc: encKey, + Weight: 1, TimeoutMS: 0, MaxConcurrency: 4, + HealthStatus: store.ChannelHealthHealthy, Enabled: true} + if err := h.db.Create(&ch2).Error; err != nil { + t.Fatal(err) + } + if err := h.db.Create(&store.ChannelModelBinding{ChannelID: ch2.ID, ModelID: m.ID, UpstreamModel: "m1"}).Error; err != nil { + t.Fatal(err) + } + body := `{"model":"m1","messages":[{"role":"user","content":"hi"}]}` + + // 预算 300ms：渠道1 耗掉 1.2s 后就不再试渠道2，直接回最后一个错误 + h.gw.failoverBudget = 300 * time.Millisecond + w := h.run(t, nil, body) + if hitFast.Load() != 0 { + t.Fatal("重试预算耗尽后不应再去试第二条渠道") + } + if w.Code != http.StatusBadGateway { + t.Fatalf("应返回 502，实际 %d: %s", w.Code, w.Body.String()) + } + + // 关掉预算：正常故障转移到渠道2 并成功 + h.gw.failoverBudget = 0 + w2 := h.run(t, nil, body) + if hitFast.Load() == 0 { + t.Fatalf("不限预算时应故障转移到渠道2，响应 %d: %s", w2.Code, w2.Body.String()) + } + if w2.Code != http.StatusOK { + t.Fatalf("转移到健康渠道应成功，实际 %d: %s", w2.Code, w2.Body.String()) + } +} + +// 流式空闲时补 SSE 注释心跳：上游长时间不出字（长思考）不能被反代空闲超时掐掉。 +func TestStreamHeartbeatWhileIdle(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + time.Sleep(700 * time.Millisecond) // 首包之后上游沉默 + _, _ = io.WriteString(w, `data: {"choices":[{"delta":{"content":"久等了"}}]}`+"\n\n") + _, _ = io.WriteString(w, "data: [DONE]\n\n") + })) + defer srv.Close() + + h := newHarness(t, srv.URL, 0, true, nil) + h.gw.streamHeartbeat = 120 * time.Millisecond + w := h.run(t, nil, `{"model":"m1","stream":true,"messages":[{"role":"user","content":"hi"}]}`) + + if w.Code != http.StatusOK { + t.Fatalf("code=%d body=%s", w.Code, w.Body.String()) + } + out := w.Body.String() + if !strings.Contains(out, ": ping") { + t.Fatalf("空闲期没有发出心跳：%q", out) + } + if !strings.Contains(out, "久等了") { + t.Fatalf("心跳把正文挤掉了：%q", out) + } + // 心跳是注释行，不该被当成输出去计费 + var logs []store.UsageLog + h.db.Find(&logs) + for _, l := range logs { + if strings.Contains(l.RawResponse, ": ping") && l.Status == store.UsageStatusSuccess && l.OutputTokens == 0 { + t.Fatalf("心跳被当作输出了：%+v", l) + } + } +} + +// GET /v1/models 只返回调用方真能用的模型，并带目录元数据。 +func TestModelsListRespectsCallerAllowlist(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = io.WriteString(w, `{}`) })) + defer srv.Close() + h := newHarness(t, srv.URL, 0, true, nil) + + // 给 m1 补上元数据 + if err := h.db.Model(&store.Model{}).Where("name = ?", "m1"). + Updates(map[string]any{"context_length": 200000, "supports_tools": true, "supports_vision": true}).Error; err != nil { + t.Fatal(err) + } + + listAs := func(key string) (int, string) { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + req := httptest.NewRequest(http.MethodGet, "/v1/models", nil) + req.Header.Set("Authorization", "Bearer "+key) + c.Request = req + h.gw.Auth(c) + if !c.IsAborted() { + h.gw.Handle(c) + } + var got struct { + Data []map[string]any `json:"data"` + } + _ = json.Unmarshal(w.Body.Bytes(), &got) + n := 0 + if len(got.Data) > 0 { + n = len(got.Data) + } + return n, w.Body.String() + } + + // unrestricted key：m1 可见且带元数据 + n, body := listAs(h.key) + if n != 1 || !strings.Contains(body, `"context_length":200000`) || !strings.Contains(body, `"tools":true`) { + t.Fatalf(" unrestricted 列表不对：%d %s", n, body) + } + + // Anthropic 客户端（只有 x-api-key）应拿到它自己的形状 + { + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + req := httptest.NewRequest(http.MethodGet, "/v1/models", nil) + req.Header.Set("x-api-key", h.key) + c.Request = req + h.gw.Auth(c) + h.gw.Handle(c) + if !strings.Contains(w.Body.String(), `"type":"model"`) || !strings.Contains(w.Body.String(), "display_name") { + t.Fatalf("x-api-key 客户端应得到 Anthropic 形状：%s", w.Body.String()) + } + if strings.Contains(w.Body.String(), `"owned_by"`) { + t.Fatalf("Anthropic 形状里不该有 OpenAI 字段：%s", w.Body.String()) + } + } + + // 受限密钥：模型名都不该泄漏出去 + token := "sk-ot-" + strings.Repeat("a", 44) + var uid uint64 + h.db.Model(&store.User{}).Select("id").Row().Scan(&uid) + if err := h.db.Create(&store.APIKey{UserID: uid, Name: "limited", + KeyHash: apikey.Hash(token), KeyPrefix: apikey.Prefix(token), + Status: store.KeyStatusActive, AllowedModels: []string{"another-model"}}).Error; err != nil { + t.Fatal(err) + } + if n, body := listAs(token); n != 0 { + t.Fatalf("密钥白名单外的模型不应出现在 /v1/models：%s", body) + } +} + +// 缓存写价单独配置后必须按它计费，而不是继续按输入价×1.25。 +func TestCacheWritePriceBilledSeparately(t *testing.T) { + body := `{"id":"x","choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],` + + `"usage":{"prompt_tokens":10,"completion_tokens":5,"cache_creation_input_tokens":1000000}}` + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, body) + })) + defer srv.Close() + + reqBody := `{"model":"m1","messages":[{"role":"user","content":"hi"}]}` + costOf := func(h *harness) float64 { + if w := h.run(t, nil, reqBody); w.Code != http.StatusOK { + t.Fatalf("code=%d body=%s", w.Code, w.Body.String()) + } + h.rec.Close() + var l store.UsageLog + if err := h.db.Order("id DESC").First(&l).Error; err != nil { + t.Fatal(err) + } + if l.CacheCreationTokens != 1000000 { + t.Fatalf("缓存写 token 没记上：%+v", l) + } + return l.Cost + } + + // 默认未配置：回退输入价 0.15 × 1.25 = 0.1875，加 in/out 的零头 + h1 := newHarness(t, srv.URL, 0, true, nil) + fallback := costOf(h1) + if fallback < 0.18 || fallback > 0.20 { + t.Fatalf("未配置缓存写价应按输入价×1.25，实际 cost=%v", fallback) + } + + // 显式配置 2.0/百万：应显著高于回退值 + h2 := newHarness(t, srv.URL, 0, true, nil) + if err := h2.db.Model(&store.Model{}).Where("name = ?", "m1"). + Update("cache_write_price", 2.0).Error; err != nil { + t.Fatal(err) + } + explicit := costOf(h2) + if explicit < 1.99 || explicit > 2.02 { + t.Fatalf("按 cache_write_price=2.0 计费应为 ~2.0，实际 %v", explicit) + } + if !(explicit > fallback) { + t.Fatalf("显式价应高于回退价：%v vs %v", explicit, fallback) + } +} + +// 全部渠道都失败时，错误行要归到最后一个试过的渠道上， +// 否则 channel_id=0，渠道维度的成功率统计永远看不到这类失败。 +func TestAllChannelsFailedAttributesLastChannel(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + _, _ = io.WriteString(w, `{"error":{"message":"boom"}}`) + })) + defer srv.Close() + + h := newHarness(t, srv.URL, 0, true, nil) + if w := h.run(t, nil, `{"model":"m1","messages":[{"role":"user","content":"hi"}]}`); w.Code != http.StatusInternalServerError { + t.Fatalf("上游 500 应透出 500，实际 %d: %s", w.Code, w.Body.String()) + } + h.rec.Close() + + var l store.UsageLog + if err := h.db.Where("error_code = ?", "all_channels_failed").First(&l).Error; err != nil { + t.Fatalf("没有 all_channels_failed 记录：%v", err) + } + if l.ChannelID == 0 { + t.Fatalf("失败记录没归到渠道上：%+v", l) + } + if l.Status != store.UsageStatusError || l.Cost != 0 { + t.Fatalf("失败不应产生扣费：%+v", l) + } +} diff --git a/server/internal/store/models.go b/server/internal/store/models.go index 55c36c5..cf6ba7c 100644 --- a/server/internal/store/models.go +++ b/server/internal/store/models.go @@ -49,55 +49,55 @@ const ( // User 用户（PLANNING §6.1） type User struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - Username string `gorm:"uniqueIndex;size:64;not null" json:"username"` - Email string `gorm:"uniqueIndex;size:255;not null" json:"email"` - PasswordHash string `gorm:"size:255;not null" json:"-"` - Role string `gorm:"size:16;not null;default:user" json:"role"` - Balance float64 `gorm:"type:numeric(20,8);not null;default:0" json:"balance"` - Status string `gorm:"size:16;not null;default:active" json:"status"` - AllowedModels []string `gorm:"type:jsonb;serializer:json" json:"allowed_models,omitempty"` // 用户级模型白名单(空=不限制) - DeniedModels []string `gorm:"type:jsonb;serializer:json" json:"denied_models,omitempty"` // 用户级模型黑名单 - InviteCode *string `json:"invite_code,omitempty"` - LastLoginAt *time.Time `json:"last_login_at,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + Username string `gorm:"uniqueIndex;size:64;not null" json:"username"` + Email string `gorm:"uniqueIndex;size:255;not null" json:"email"` + PasswordHash string `gorm:"size:255;not null" json:"-"` + Role string `gorm:"size:16;not null;default:user" json:"role"` + Balance float64 `gorm:"type:numeric(20,8);not null;default:0" json:"balance"` + Status string `gorm:"size:16;not null;default:active" json:"status"` + AllowedModels []string `gorm:"type:jsonb;serializer:json" json:"allowed_models,omitempty"` // 用户级模型白名单(空=不限制) + DeniedModels []string `gorm:"type:jsonb;serializer:json" json:"denied_models,omitempty"` // 用户级模型黑名单 + InviteCode *string `json:"invite_code,omitempty"` + LastLoginAt *time.Time `json:"last_login_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // APIKey 密钥（PLANNING §6.2）：库中只存 SHA-256 哈希 + 展示前缀 type APIKey struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - UserID uint64 `gorm:"index;not null" json:"user_id"` - Name string `gorm:"size:64;not null" json:"name"` - KeyHash string `gorm:"uniqueIndex;size:64;not null" json:"-"` - KeyPrefix string `gorm:"size:32;not null" json:"key_prefix"` - QuotaTokensPerDay *int64 `json:"quota_tokens_per_day,omitempty"` - QuotaRequestsPerDay *int `json:"quota_requests_per_day,omitempty"` - AllowedModels []string `gorm:"type:jsonb;serializer:json" json:"allowed_models,omitempty"` - ExpiresAt *time.Time `json:"expires_at,omitempty"` - Status string `gorm:"size:16;not null;default:active" json:"status"` - LastUsedAt *time.Time `json:"last_used_at,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + UserID uint64 `gorm:"index;not null" json:"user_id"` + Name string `gorm:"size:64;not null" json:"name"` + KeyHash string `gorm:"uniqueIndex;size:64;not null" json:"-"` + KeyPrefix string `gorm:"size:32;not null" json:"key_prefix"` + QuotaTokensPerDay *int64 `json:"quota_tokens_per_day,omitempty"` + QuotaRequestsPerDay *int `json:"quota_requests_per_day,omitempty"` + AllowedModels []string `gorm:"type:jsonb;serializer:json" json:"allowed_models,omitempty"` + ExpiresAt *time.Time `json:"expires_at,omitempty"` + Status string `gorm:"size:16;not null;default:active" json:"status"` + LastUsedAt *time.Time `json:"last_used_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // Channel 上游渠道（PLANNING §6.3） type Channel struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - Name string `gorm:"uniqueIndex;size:64;not null" json:"name"` - Provider string `gorm:"size:16;not null" json:"provider"` // openai|anthropic|compatible（供应商/默认格式） + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + Name string `gorm:"uniqueIndex;size:64;not null" json:"name"` + Provider string `gorm:"size:16;not null" json:"provider"` // openai|anthropic|compatible（供应商/默认格式） Formats []string `gorm:"type:jsonb;serializer:json" json:"formats,omitempty"` // 原生支持的协议格式 chat|responses|messages BaseURL string `gorm:"size:255;not null" json:"base_url"` BaseURLs map[string]string `gorm:"type:jsonb;serializer:json" json:"base_urls,omitempty"` // 分协议 base_url 覆盖（chat/responses/messages） - APIKeyEnc string `gorm:"size:1024;not null" json:"-"` // AES-GCM 密文 - Weight int `gorm:"not null;default:1" json:"weight"` - Priority int `gorm:"not null;default:0" json:"priority"` // 数值小优先 - TimeoutMS int `gorm:"not null;default:300000" json:"timeout_ms"` - MaxConcurrency int `gorm:"not null;default:16" json:"max_concurrency"` - HealthStatus string `gorm:"size:16;not null;default:healthy" json:"health_status"` - Enabled bool `gorm:"not null;default:true" json:"enabled"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + APIKeyEnc string `gorm:"size:1024;not null" json:"-"` // AES-GCM 密文 + Weight int `gorm:"not null;default:1" json:"weight"` + Priority int `gorm:"not null;default:0" json:"priority"` // 数值小优先 + TimeoutMS int `gorm:"not null;default:300000" json:"timeout_ms"` + MaxConcurrency int `gorm:"not null;default:16" json:"max_concurrency"` + HealthStatus string `gorm:"size:16;not null;default:healthy" json:"health_status"` + Enabled bool `gorm:"not null;default:true" json:"enabled"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // FormatsEffective 返回渠道实际支持的原生协议；未显式配置时按 provider 推断。 @@ -119,10 +119,10 @@ func (c *Channel) FormatsEffective() []string { var versionSegRe = regexp.MustCompile(`/v[0-9]+/?$`) // UpstreamURL 按协议选 base_url（分协议覆盖优先），再按版本前缀拼资源路径（path 不含 /v1）。 -// - proto 有 BaseURLs 覆盖时用覆盖值，否则用主 BaseURL -// - base 已以资源路径结尾 → 原样 -// - base 含版本前缀(如 /v1、/v4) → base + path -// - 否则 → base + /v1 + path（默认补 OpenAI/Anthropic 的 /v1） +// - proto 有 BaseURLs 覆盖时用覆盖值，否则用主 BaseURL +// - base 已以资源路径结尾 → 原样 +// - base 含版本前缀(如 /v1、/v4) → base + path +// - 否则 → base + /v1 + path（默认补 OpenAI/Anthropic 的 /v1） func (c *Channel) UpstreamURL(proto, path string) string { base := c.BaseURL if len(c.BaseURLs) > 0 && c.BaseURLs[proto] != "" { @@ -143,24 +143,39 @@ func (c *Channel) UpstreamURL(proto, path string) string { // Model 全局模型 + 定价（PLANNING §6.4，价格按每百万 token，USD） type Model struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - Name string `gorm:"uniqueIndex;size:128;not null" json:"name"` - InputPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"input_price"` - OutputPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"output_price"` - CacheReadPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"cache_read_price"` - Enabled bool `gorm:"not null;default:true" json:"enabled"` - Sort int `gorm:"not null;default:0" json:"sort"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + Name string `gorm:"uniqueIndex;size:128;not null" json:"name"` + InputPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"input_price"` + OutputPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"output_price"` + CacheReadPrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"cache_read_price"` + // 缓存写价（每百万 token）。为 0 时按输入价 ×1.25 计（Anthropic 5m 口径）， + // 显式配置它才能支持 1h 缓存（×2）这类不同倍率。 + CacheWritePrice float64 `gorm:"type:numeric(20,8);not null;default:0" json:"cache_write_price"` + ContextLength int `gorm:"not null;default:0" json:"context_length"` // 上下文窗口（token），0=未知 + SupportsTools bool `gorm:"not null;default:false" json:"supports_tools"` + SupportsVision bool `gorm:"not null;default:false" json:"supports_vision"` + SupportsReasoning bool `gorm:"not null;default:false" json:"supports_reasoning"` + Enabled bool `gorm:"not null;default:true" json:"enabled"` + Sort int `gorm:"not null;default:0" json:"sort"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// CacheWritePriceOr 返回缓存写价；未配置时回退到输入价×1.25（Anthropic 5m 口径）。 +func (m *Model) CacheWritePriceOr() float64 { + if m.CacheWritePrice > 0 { + return m.CacheWritePrice + } + return m.InputPrice * 1.25 } // ChannelModelBinding 渠道↔模型绑定（多对多，PLANNING §6.4） type ChannelModelBinding struct { - ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - ChannelID uint64 `gorm:"index:idx_ch_model,unique;not null" json:"channel_id"` - ModelID uint64 `gorm:"index:idx_ch_model,unique;not null" json:"model_id"` - UpstreamModel string `gorm:"size:255;not null" json:"upstream_model"` - Weight int `gorm:"not null;default:1" json:"weight"` + ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` + ChannelID uint64 `gorm:"index:idx_ch_model,unique;not null" json:"channel_id"` + ModelID uint64 `gorm:"index:idx_ch_model,unique;not null" json:"model_id"` + UpstreamModel string `gorm:"size:255;not null" json:"upstream_model"` + Weight int `gorm:"not null;default:1" json:"weight"` Channel Channel `gorm:"foreignKey:ChannelID" json:"-"` Model Model `gorm:"foreignKey:ModelID" json:"-"` } @@ -187,8 +202,8 @@ type UsageLog struct { LatencyMS int `json:"latency_ms"` Status string `gorm:"size:16;not null" json:"status"` ErrorCode *string `json:"error_code,omitempty"` - RawRequest string `gorm:"type:text" json:"raw_request"` // 客户端原始请求体（未转换） - RawResponse string `gorm:"type:text" json:"raw_response"` // 上游原始响应（未转换；流式为全部 SSE 事件） + RawRequest string `gorm:"type:text" json:"raw_request"` // 客户端原始请求体（未转换） + RawResponse string `gorm:"type:text" json:"raw_response"` // 上游原始响应（未转换；流式为全部 SSE 事件） CreatedAt time.Time `gorm:"index" json:"created_at"` } diff --git a/web/src/types.ts b/web/src/types.ts index 37ea18d..85c9f44 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -64,6 +64,11 @@ export interface Model { input_price: number output_price: number cache_read_price: number + cache_write_price: number // 0 = 按输入价 ×1.25 计 + context_length: number // 0 = 未填 + supports_tools: boolean + supports_vision: boolean + supports_reasoning: boolean enabled: boolean sort: number channels: ModelBinding[] @@ -72,6 +77,23 @@ export interface Model { denied?: boolean } +export interface ChannelStat { + channel_id: number + channel_name: string + requests: number + success: number + error: number + canceled: number + err_rate: number + p50_ms: number + p95_ms: number + max_ms: number + tokens: number + cost: number + top_errors: { code: string; count: number }[] + last_at: string +} + export interface ModelSummary { total: number unpriced: number diff --git a/web/src/views/admin/ChannelsView.vue b/web/src/views/admin/ChannelsView.vue index ee776a7..f2f33d2 100644 --- a/web/src/views/admin/ChannelsView.vue +++ b/web/src/views/admin/ChannelsView.vue @@ -1,15 +1,16 @@