diff --git a/.env.example b/.env.example index 0244045..fb7ed35 100644 --- a/.env.example +++ b/.env.example @@ -29,6 +29,10 @@ OT_PROXY_UPSTREAM_BASE_URL=https://api.openai.com OT_PROXY_DEFAULT_MODEL=gpt-4o-mini OT_PROXY_TIMEOUT=120s +# 渠道健康检查 +OT_PROXY_HEALTH_INTERVAL=60s +OT_PROXY_HEALTH_FAIL_THRESHOLD=2 + # 初始管理员(仅首次创建生效) OT_ADMIN_USERNAME=admin OT_ADMIN_EMAIL=admin@localhost diff --git a/PLANNING.md b/PLANNING.md index db3e057..5854a1a 100644 --- a/PLANNING.md +++ b/PLANNING.md @@ -471,9 +471,12 @@ pending(待审核) ──approve──▶ credited(已入账) - mock 上游新增 Anthropic Messages 端点,端到端验证 8 种组合(三协议 × 直通/转换 × 流式/非流式)。 **验收**:chat 调 Claude、messages 调 OpenAI、responses 调 Claude 均正确,流式逐事件转换,usage 记账准确。 -### M5 渠道体系完善(◻ 规划) -健康检查、负载均衡、重试/故障转移、并发控制、模型自动导入。 -**验收**:杀一个渠道自动切换;连续失败进 cooldown 并恢复。 +### M5 渠道体系完善(✅ 已完成) +- `channel.Candidates` 按模型绑定取候选 + `Pick` 加权随机负载均衡;`TryAcquire` 每渠道并发信号量(满载溢出到其他渠道)。 +- `HealthMonitor` 后台定时探测,连续失败进 cooldown、恢复放回(`OT_PROXY_HEALTH_INTERVAL`/`OT_PROXY_HEALTH_FAIL_THRESHOLD`)。 +- `doProxy` 遍历候选渠道故障转移:网络错误/429/5xx/超时且未写出响应头时安全重试;流式已写出首字节放弃。 +- 单测覆盖候选过滤/加权/并发/健康状态机。 +**验收**:杀一个渠道自动切换;连续失败进 cooldown 并恢复;并发请求溢出不失败。 ### M6 充值 + 审核(◻ 待定,接口/模型已预留) 充值订单、人工审核、流水留痕、前端页面。 diff --git a/README.md b/README.md index bffabbc..19df7e0 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ 自托管的 LLM API 中转网关,功能对标 OpenRouter / one-api:统一 OpenAI 与 Anthropic 协议入口,背后对接多个上游渠道,内置用户体系、API Key 管理与用量计费。 -> 规划文档见 [PLANNING.md](./PLANNING.md)。当前进度:**M0-M2 + M4 已完成**(基建 + 用户/密钥/核心代理 + 前端 MVP + 管理后台基础 + 三协议互转)。 +> 规划文档见 [PLANNING.md](./PLANNING.md)。当前进度:**M0-M2 + M4-M5 已完成**(基建 + 用户/密钥/核心代理 + 前端 MVP + 管理后台基础 + 三协议互转 + 渠道体系)。 ## 功能(当前) @@ -13,6 +13,7 @@ - `GET /v1/models` — 可用模型列表 - **三协议互转**:客户端协议 × 渠道协议不匹配时自动转换(如 Chat 调用 Claude、Messages 调用 OpenAI、Responses 调用 Claude),流式逐事件转换;协议匹配时直通 - 错误按客户端协议返回(OpenAI 格式 / Anthropic 格式) +- **渠道体系**:按模型绑定选渠道 + 加权负载均衡;每渠道并发信号量(满载溢出);后台健康检查(连续失败进 cooldown、恢复放回);可安全重试的失败自动故障转移(网络错误/429/5xx/超时且未写出响应头) - **用户体系**:注册(开放/邀请码可切换,管理后台可改)、登录(JWT access + HttpOnly refresh cookie)、argon2id 密码 - **API Key**:`sk-` 48 位 base62,仅存 SHA-256 哈希,明文一次性展示;支持限额/过期/白名单字段 - **用量计费**:请求级 `usage_logs` 异步批量落库,按模型价格扣减余额,日粒度预聚合(`usage_daily`) @@ -23,7 +24,7 @@ - 管理后台:运营总览、渠道管理、模型与定价、用户管理、系统配置 - **后端**:Go + Gin + GORM,SQLite(开发)/ PostgreSQL(生产) -> 渠道健康检查/负载均衡/重试在 M5(见 PLANNING.md §10)。 +> 剩余:M6 充值(待定)与 M3 收尾(限流、骨架屏)。 ## 快速开始(开发) diff --git a/server/internal/app/app.go b/server/internal/app/app.go index edf5033..167a434 100644 --- a/server/internal/app/app.go +++ b/server/internal/app/app.go @@ -6,6 +6,7 @@ import ( "log" "time" + "github.com/openteam/server/internal/channel" "github.com/openteam/server/internal/config" "github.com/openteam/server/internal/pkg/crypto" "github.com/openteam/server/internal/pkg/jwt" @@ -21,7 +22,10 @@ type App struct { Enc *crypto.Encryptor JWT *jwt.Manager Usage *usage.Recorder + Health *channel.HealthMonitor startedAt time.Time + ctx context.Context + cancel context.CancelFunc } func New(cfg *config.Config) (*App, error) { @@ -38,11 +42,18 @@ func New(cfg *config.Config) (*App, error) { JWT: jwt.NewManager(cfg.JWT.Secret, cfg.JWT.Issuer, cfg.JWT.AccessTTL, cfg.JWT.RefreshTTL), startedAt: time.Now(), } + a.ctx, a.cancel = context.WithCancel(context.Background()) a.Usage = usage.NewRecorder(db) if err := a.Seed(); err != nil { return nil, err } + + a.Health = channel.NewHealthMonitor(db, a.Enc, channel.HealthConfig{ + Interval: cfg.Proxy.HealthInterval, + FailThreshold: cfg.Proxy.HealthFailThreshold, + }) + a.Health.Start(a.ctx) return a, nil } @@ -116,6 +127,7 @@ func (a *App) Seed() error { } func (a *App) Shutdown(ctx context.Context) { + a.cancel() a.Usage.Close() if sqlDB, err := a.DB.DB(); err == nil { _ = sqlDB.Close() diff --git a/server/internal/channel/channel.go b/server/internal/channel/channel.go index bc19e3b..2369126 100644 --- a/server/internal/channel/channel.go +++ b/server/internal/channel/channel.go @@ -1,10 +1,12 @@ -// Package channel 渠道仓储:选择、密钥加解密、模型解析。 -// M1 实现最小选择逻辑(优先级+权重取第一个健康启用的渠道); -// 负载均衡/健康检查/故障转移在 M5 完善。 +// Package channel 渠道仓储:候选选择、加权负载均衡、并发控制、密钥加解密。 +// M5 起支持多候选(负载均衡 + 故障转移)与每渠道并发信号量。 package channel import ( + "crypto/rand" "errors" + "math/big" + "sync" "github.com/openteam/server/internal/pkg/crypto" "github.com/openteam/server/internal/store" @@ -16,47 +18,120 @@ var ErrNoChannel = errors.New("no available channel") type Service struct { db *gorm.DB enc *crypto.Encryptor + + mu sync.Mutex + sems map[uint64]chan struct{} } func NewService(db *gorm.DB, enc *crypto.Encryptor) *Service { - return &Service{db: db, enc: enc} + return &Service{db: db, enc: enc, sems: map[uint64]chan struct{}{}} } -// Select 选择处理请求的渠道:启用 + 健康,按 priority 升序、weight 降序。 -func (s *Service) Select() (*store.Channel, error) { +// Candidates 返回可用渠道候选:健康 + 启用,按优先级、权重降序、id 升序排列。 +// model 非空时优先取绑定该模型的渠道;无绑定则退回全局。 +func (s *Service) Candidates(model string) []*store.Channel { + if model != "" { + var b []store.ChannelModelBinding + var modelIDs []uint64 + s.db.Model(&store.Model{}).Where("name = ? AND enabled = ?", model, true).Pluck("id", &modelIDs) + if len(modelIDs) > 0 { + s.db.Where("model_id IN ?", modelIDs).Find(&b) + chs := s.loadBound(b) + if len(chs) > 0 { + return chs + } + } + } var chs []store.Channel - if err := s.db.Where("enabled = ? AND health_status = ?", true, store.ChannelHealthHealthy). - Order("priority ASC, weight DESC, id ASC").Find(&chs).Error; err != nil { - return nil, err + s.db.Where("enabled = ? AND health_status = ?", true, store.ChannelHealthHealthy). + Order("priority ASC, weight DESC, id ASC").Find(&chs) + out := make([]*store.Channel, 0, len(chs)) + for i := range chs { + out = append(out, &chs[i]) } - if len(chs) == 0 { - return nil, ErrNoChannel + return out +} + +// loadBound 按绑定顺序加载渠道,过滤健康/启用。 +func (s *Service) loadBound(bindings []store.ChannelModelBinding) []*store.Channel { + if len(bindings) == 0 { + return nil + } + ids := make([]uint64, 0, len(bindings)) + seen := map[uint64]bool{} + for _, b := range bindings { + if !seen[b.ChannelID] { + seen[b.ChannelID] = true + ids = append(ids, b.ChannelID) + } + } + var chs []store.Channel + s.db.Where("id IN ? AND enabled = ? AND health_status = ?", ids, true, store.ChannelHealthHealthy). + Order("priority ASC, weight DESC, id ASC").Find(&chs) + // 保持绑定顺序 + byID := map[uint64]*store.Channel{} + for i := range chs { + byID[chs[i].ID] = &chs[i] + } + out := make([]*store.Channel, 0, len(ids)) + for _, id := range ids { + if ch, ok := byID[id]; ok { + out = append(out, ch) + } + } + return out +} + +// Pick 按权重加权随机选一个候选(负载均衡)。 +func (s *Service) Pick(cands []*store.Channel) *store.Channel { + if len(cands) == 0 { + return nil + } + total := 0 + for _, c := range cands { + w := c.Weight + if w <= 0 { + w = 1 + } + total += w + } + n, _ := rand.Int(rand.Reader, big.NewInt(int64(total))) + acc := 0 + for _, c := range cands { + w := c.Weight + if w <= 0 { + w = 1 + } + acc += w + if int(n.Int64()) < acc { + return c + } + } + return cands[len(cands)-1] +} + +// TryAcquire 尝试获取渠道并发槽;渠道满载返回 false(调用方可溢出到其他渠道)。 +// MaxConcurrency<=0 视为不限制。 +func (s *Service) TryAcquire(ch *store.Channel) (func(), bool) { + if ch.MaxConcurrency <= 0 { + return func() {}, true + } + s.mu.Lock() + sem, ok := s.sems[ch.ID] + if !ok { + sem = make(chan struct{}, ch.MaxConcurrency) + s.sems[ch.ID] = sem + } + s.mu.Unlock() + select { + case sem <- struct{}{}: + return func() { <-sem }, true + default: + return nil, false } - return &chs[0], nil } // UpstreamKey 解密渠道上游密钥。 func (s *Service) UpstreamKey(ch *store.Channel) (string, error) { return s.enc.Decrypt(ch.APIKeyEnc) } - -// ResolveModel 按全局模型名找到绑定渠道;M1 简化:返回绑定该模型的第一个健康渠道。 -func (s *Service) ResolveModel(modelName string) (*store.Channel, *store.ChannelModelBinding, error) { - var m store.Model - if err := s.db.Where("name = ? AND enabled = ?", modelName, true).First(&m).Error; err != nil { - return nil, nil, err - } - var b store.ChannelModelBinding - if err := s.db.Where("model_id = ?", m.ID). - Order("weight DESC, id ASC").First(&b).Error; err != nil { - return nil, nil, err - } - var ch store.Channel - if err := s.db.First(&ch, b.ChannelID).Error; err != nil { - return nil, nil, err - } - if !ch.Enabled || ch.HealthStatus != store.ChannelHealthHealthy { - return nil, nil, ErrNoChannel - } - return &ch, &b, nil -} diff --git a/server/internal/channel/channel_test.go b/server/internal/channel/channel_test.go new file mode 100644 index 0000000..d78d2bc --- /dev/null +++ b/server/internal/channel/channel_test.go @@ -0,0 +1,131 @@ +package channel + +import ( + "context" + "os" + "testing" + "time" + + "github.com/openteam/server/internal/pkg/crypto" + "github.com/openteam/server/internal/store" + "gorm.io/gorm" +) + +func newTestDB(t *testing.T) *gorm.DB { + t.Helper() + dsn := "/tmp/channel_test_" + time.Now().Format("150405.000000000") + ".db" + db, err := store.Open("sqlite", dsn) + if err != nil { + t.Fatalf("open db: %v", err) + } + t.Cleanup(func() { + sqlDB, _ := db.DB() + sqlDB.Close() + os.Remove(dsn) + }) + return db +} + +func seedChannel(t *testing.T, db *gorm.DB, name string, weight, priority, maxConc int) uint64 { + t.Helper() + enc := crypto.NewEncryptor("test-master-key-1234567890") + key, _ := enc.Encrypt("upstream-key") + ch := store.Channel{ + Name: name, Provider: store.ChannelProviderOpenAI, BaseURL: "http://localhost:9000", + APIKeyEnc: key, Weight: weight, Priority: priority, MaxConcurrency: maxConc, + HealthStatus: store.ChannelHealthHealthy, Enabled: true, + } + if err := db.Create(&ch).Error; err != nil { + t.Fatalf("create channel: %v", err) + } + return ch.ID +} + +func TestCandidatesFiltersUnhealthy(t *testing.T) { + db := newTestDB(t) + seedChannel(t, db, "a", 1, 0, 4) + seedChannel(t, db, "b", 5, 1, 4) + // 禁用渠道不入候选 + bad := store.Channel{Name: "bad", Provider: store.ChannelProviderOpenAI, BaseURL: "http://x", APIKeyEnc: "x", HealthStatus: store.ChannelHealthCooldown, Enabled: true} + db.Create(&bad) + + s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890")) + cands := s.Candidates("") + if len(cands) != 2 { + t.Fatalf("candidates = %d, want 2", len(cands)) + } + if cands[0].Name != "a" { + t.Fatalf("first by priority should be a, got %s", cands[0].Name) + } +} + +func TestPickWeighted(t *testing.T) { + db := newTestDB(t) + seedChannel(t, db, "only", 1, 0, 4) + s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890")) + cands := s.Candidates("") + chosen := s.Pick(cands) + if chosen == nil || chosen.Name != "only" { + t.Fatalf("pick single: %+v", chosen) + } + // 空候选返回 nil + if p := s.Pick(nil); p != nil { + t.Fatal("expected nil for empty candidates") + } +} + +func TestTryAcquire(t *testing.T) { + db := newTestDB(t) + id := seedChannel(t, db, "c", 1, 0, 1) + var ch store.Channel + db.First(&ch, id) + s := NewService(db, crypto.NewEncryptor("test-master-key-1234567890")) + release, ok := s.TryAcquire(&ch) + if !ok { + t.Fatal("first acquire should succeed") + } + if _, ok := s.TryAcquire(&ch); ok { + t.Fatal("second acquire should fail (capacity 1)") + } + release() + if _, ok := s.TryAcquire(&ch); !ok { + t.Fatal("acquire after release should succeed") + } +} + +func TestHealthCooldownAndRecovery(t *testing.T) { + db := newTestDB(t) + id := seedChannel(t, db, "h", 1, 0, 4) + var ch store.Channel + db.First(&ch, id) + + m := NewHealthMonitor(db, crypto.NewEncryptor("test-master-key-1234567890"), HealthConfig{ + Interval: time.Minute, FailThreshold: 2, Timeout: time.Second, + }) + + // 两次失败 → cooldown + m.record(&ch, false) + if ch.HealthStatus != store.ChannelHealthHealthy { + t.Fatal("one failure should stay healthy") + } + m.record(&ch, false) + if ch.HealthStatus != store.ChannelHealthCooldown { + t.Fatalf("two failures should cooldown, got %s", ch.HealthStatus) + } + // 恢复 + m.record(&ch, true) + if ch.HealthStatus != store.ChannelHealthHealthy { + t.Fatalf("success should recover, got %s", ch.HealthStatus) + } +} + +func TestHealthStartStops(t *testing.T) { + db := newTestDB(t) + seedChannel(t, db, "probe", 1, 0, 4) + m := NewHealthMonitor(db, crypto.NewEncryptor("test-master-key-1234567890"), HealthConfig{ + Interval: time.Hour, FailThreshold: 2, Timeout: time.Second, + }) + ctx, cancel := context.WithCancel(context.Background()) + m.Start(ctx) + cancel() // 不应 panic +} diff --git a/server/internal/channel/health.go b/server/internal/channel/health.go new file mode 100644 index 0000000..9a7a45d --- /dev/null +++ b/server/internal/channel/health.go @@ -0,0 +1,122 @@ +package channel + +import ( + "context" + "log" + "net/http" + "strings" + "sync" + "time" + + "github.com/openteam/server/internal/pkg/crypto" + "github.com/openteam/server/internal/store" + "gorm.io/gorm" +) + +// HealthConfig 健康检查参数。 +type HealthConfig struct { + Interval time.Duration // 探测周期 + FailThreshold int // 连续失败 N 次进入 cooldown + Timeout time.Duration // 单次探测超时 +} + +// HealthMonitor 定时探测渠道健康状态。 +type HealthMonitor struct { + db *gorm.DB + enc *crypto.Encryptor + cfg HealthConfig + hc *http.Client + mu sync.Mutex + fail map[uint64]int // 渠道连续失败次数 +} + +func NewHealthMonitor(db *gorm.DB, enc *crypto.Encryptor, cfg HealthConfig) *HealthMonitor { + if cfg.Interval <= 0 { + cfg.Interval = 60 * time.Second + } + if cfg.FailThreshold <= 0 { + cfg.FailThreshold = 2 + } + if cfg.Timeout <= 0 { + cfg.Timeout = 5 * time.Second + } + return &HealthMonitor{ + db: db, enc: enc, cfg: cfg, + hc: &http.Client{Timeout: cfg.Timeout}, + fail: map[uint64]int{}, + } +} + +// Start 启动后台探测循环(ctx 取消即停止)。 +func (h *HealthMonitor) Start(ctx context.Context) { + go func() { + ticker := time.NewTicker(h.cfg.Interval) + defer ticker.Stop() + h.probeAll() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + h.probeAll() + } + } + }() + log.Printf("health: monitor started (interval=%s, threshold=%d)", h.cfg.Interval, h.cfg.FailThreshold) +} + +// probeAll 探测所有启用渠道并更新健康状态。 +func (h *HealthMonitor) probeAll() { + var chs []store.Channel + if err := h.db.Where("enabled = ?", true).Find(&chs).Error; err != nil { + return + } + for i := range chs { + h.probe(&chs[i]) + } +} + +// probe 单渠道探测:GET {base}/v1/models。 +func (h *HealthMonitor) probe(ch *store.Channel) { + key, err := h.enc.Decrypt(ch.APIKeyEnc) + if err != nil { + h.record(ch, false) + return + } + url := strings.TrimRight(ch.BaseURL, "/") + "/v1/models" + req, _ := http.NewRequest(http.MethodGet, url, nil) + req.Header.Set("Authorization", "Bearer "+key) + req.Header.Set("Accept", "application/json") + resp, err := h.hc.Do(req) + ok := err == nil && resp.StatusCode >= 200 && resp.StatusCode < 300 + if resp != nil { + resp.Body.Close() + } + h.record(ch, ok) +} + +// record 记录一次探测结果,按阈值切换健康状态。 +func (h *HealthMonitor) record(ch *store.Channel, ok bool) { + h.mu.Lock() + defer h.mu.Unlock() + + prev := ch.HealthStatus + if ok { + h.fail[ch.ID] = 0 + if prev != store.ChannelHealthHealthy { + h.setStatus(ch, store.ChannelHealthHealthy) + log.Printf("health: channel %q recovered -> healthy", ch.Name) + } + return + } + h.fail[ch.ID]++ + if h.fail[ch.ID] >= h.cfg.FailThreshold && prev != store.ChannelHealthCooldown { + h.setStatus(ch, store.ChannelHealthCooldown) + log.Printf("health: channel %q -> cooldown (%d consecutive failures)", ch.Name, h.fail[ch.ID]) + } +} + +func (h *HealthMonitor) setStatus(ch *store.Channel, status string) { + h.db.Model(&store.Channel{}).Where("id = ?", ch.ID).Update("health_status", status) + ch.HealthStatus = status +} diff --git a/server/internal/config/config.go b/server/internal/config/config.go index d5c2f48..a44c234 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -50,6 +50,8 @@ type ProxyConfig struct { UpstreamKey string // 渠道上游 key 默认值 DefaultModel string // 渠道模型导入时使用的模型名 Timeout time.Duration + HealthInterval time.Duration // 渠道健康检查周期 + HealthFailThreshold int // 连续失败 N 次进 cooldown } // loadDotEnv 读取 .env 并把 KEY=VALUE 注入环境变量(AutomaticEnv 自动映射 OT_ 前缀)。 @@ -111,6 +113,8 @@ func Load() (*Config, error) { v.SetDefault("proxy.upstream_key", "") v.SetDefault("proxy.default_model", "gpt-4o-mini") v.SetDefault("proxy.timeout", "120s") + v.SetDefault("proxy.health_interval", "60s") + v.SetDefault("proxy.health_fail_threshold", 2) return &Config{ Env: v.GetString("env"), @@ -137,11 +141,13 @@ func Load() (*Config, error) { SaltLen: v.GetInt("auth.salt_len"), }, Proxy: ProxyConfig{ - DefaultChannelName: v.GetString("proxy.default_channel_name"), - UpstreamBaseURL: v.GetString("proxy.upstream_base_url"), - UpstreamKey: v.GetString("proxy.upstream_key"), - DefaultModel: v.GetString("proxy.default_model"), - Timeout: v.GetDuration("proxy.timeout"), + DefaultChannelName: v.GetString("proxy.default_channel_name"), + UpstreamBaseURL: v.GetString("proxy.upstream_base_url"), + UpstreamKey: v.GetString("proxy.upstream_key"), + DefaultModel: v.GetString("proxy.default_model"), + Timeout: v.GetDuration("proxy.timeout"), + HealthInterval: v.GetDuration("proxy.health_interval"), + HealthFailThreshold: v.GetInt("proxy.health_fail_threshold"), }, Master: v.GetString("master_key"), }, nil diff --git a/server/internal/proxy/gateway.go b/server/internal/proxy/gateway.go index e47017e..031e87f 100644 --- a/server/internal/proxy/gateway.go +++ b/server/internal/proxy/gateway.go @@ -123,14 +123,9 @@ func (g *Gateway) models(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"object": "list", "data": data}) } -// selectChannel 选渠道:优先按模型绑定解析,退化为全局选渠道。 -func (g *Gateway) selectChannel(c *gin.Context, model string) (*store.Channel, error) { - if model != "" { - if ch, _, err := g.ch.ResolveModel(model); err == nil { - return ch, nil - } - } - return g.ch.Select() +// candidateChannels 返回可用渠道候选(按模型绑定优先,退化全局)。 +func (g *Gateway) candidateChannels(model string) []*store.Channel { + return g.ch.Candidates(model) } // resolveUser 取当前用户(含余额)。 diff --git a/server/internal/proxy/handlers.go b/server/internal/proxy/handlers.go index f5e7e4a..50267af 100644 --- a/server/internal/proxy/handlers.go +++ b/server/internal/proxy/handlers.go @@ -24,20 +24,15 @@ func (g *Gateway) chatCompletions(c *gin.Context) { c.Set("protocol", convert.ProtoChat) c.Set("model_name", br.Model) - ch, err := g.selectChannel(c, br.Model) - if err != nil { + cands := g.candidateChannels(br.Model) + if len(cands) == 0 { apiError(c, http.StatusServiceUnavailable, "no_channel", "No available upstream channel") g.recordError(c, nil, nil, now(), "no_channel") return } - plan, err := prepareUpstream(ch.Provider, convert.ProtoChat, body) - if err != nil { - apiError(c, http.StatusInternalServerError, "conversion_error", "Failed to convert request: "+err.Error()) - return - } sink := &usageSink{} c.Set("usage_raw", &sinkHolder{sink: sink}) - g.doProxy(c, ch, plan, br.Stream, sink) + g.doProxy(c, cands, convert.ProtoChat, body, br.Stream, sink) } // responses POST /v1/responses(OpenAI Responses API) @@ -57,20 +52,15 @@ func (g *Gateway) responses(c *gin.Context) { c.Set("protocol", convert.ProtoResponses) c.Set("model_name", br.Model) - ch, err := g.selectChannel(c, br.Model) - if err != nil { + cands := g.candidateChannels(br.Model) + if len(cands) == 0 { apiError(c, http.StatusServiceUnavailable, "no_channel", "No available upstream channel") g.recordError(c, nil, nil, now(), "no_channel") return } - plan, err := prepareUpstream(ch.Provider, convert.ProtoResponses, body) - if err != nil { - apiError(c, http.StatusInternalServerError, "conversion_error", "Failed to convert request: "+err.Error()) - return - } sink := &usageSink{} c.Set("usage_raw", &sinkHolder{sink: sink}) - g.doProxy(c, ch, plan, br.Stream, sink) + g.doProxy(c, cands, convert.ProtoResponses, body, br.Stream, sink) } // messages POST /v1/messages(Anthropic Messages API) @@ -90,20 +80,15 @@ func (g *Gateway) messages(c *gin.Context) { c.Set("protocol", convert.ProtoMessages) c.Set("model_name", br.Model) - ch, err := g.selectChannel(c, br.Model) - if err != nil { + cands := g.candidateChannels(br.Model) + if len(cands) == 0 { apiError(c, http.StatusServiceUnavailable, "no_channel", "No available upstream channel") g.recordError(c, nil, nil, now(), "no_channel") return } - plan, err := prepareUpstream(ch.Provider, convert.ProtoMessages, body) - if err != nil { - apiError(c, http.StatusInternalServerError, "conversion_error", "Failed to convert request: "+err.Error()) - return - } sink := &usageSink{} c.Set("usage_raw", &sinkHolder{sink: sink}) - g.doProxy(c, ch, plan, br.Stream, sink) + g.doProxy(c, cands, convert.ProtoMessages, body, br.Stream, sink) } // usageSinkHolder 桥接:gin context 里保存 sink 引用,供 finishUsage 读取最终 usage。 diff --git a/server/internal/proxy/passthrough.go b/server/internal/proxy/passthrough.go index 70dc36b..e440183 100644 --- a/server/internal/proxy/passthrough.go +++ b/server/internal/proxy/passthrough.go @@ -48,12 +48,45 @@ func upstreamURL(ch *store.Channel, path string) string { return strings.TrimRight(ch.BaseURL, "/") + path } -// doProxy 通用代理:替换 Authorization 为渠道密钥,转发请求;按 plan 决定路径与转换。 -func (g *Gateway) doProxy(c *gin.Context, ch *store.Channel, plan *upstreamPlan, stream bool, sink *usageSink) { +// doProxy 通用代理(M5):遍历候选渠道,按需转换;可安全重试的失败自动故障转移。 +func (g *Gateway) doProxy(c *gin.Context, cands []*store.Channel, clientProto string, body []byte, stream bool, sink *usageSink) { + var lastStatus = http.StatusBadGateway + var lastBody = []byte("all upstream channels failed") + for _, ch := range cands { + plan, err := prepareUpstream(ch.Provider, clientProto, body) + if err != nil { + lastStatus, lastBody = http.StatusInternalServerError, []byte("conversion error: "+err.Error()) + continue + } + release, ok := g.ch.TryAcquire(ch) + if !ok { + continue // 渠道满载,溢出到下一个 + } + written, retry, st, b := g.proxyOne(c, ch, plan, stream, sink) + release() + if written { + return + } + if !retry { + // 非重试性失败(如 400):透传上游错误体 + c.Header("Content-Type", "application/json") + c.Data(st, "application/json", b) + g.recordError(c, ch, nil, now(), "upstream_http_"+strconv.Itoa(st)) + return + } + lastStatus, lastBody = st, b + } + // 全部候选重试性失败 + apiError(c, lastStatus, "upstream_error", string(lastBody)) + g.recordError(c, nil, nil, now(), "all_channels_failed") +} + +// proxyOne 对单个渠道执行一次代理。 +// 返回:written=是否已写客户端响应;retry=是否可安全换渠道重试;status+respBody=失败信息。 +func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan, stream bool, sink *usageSink) (written bool, retry bool, status int, respBody []byte) { upKey, err := g.ch.UpstreamKey(ch) if err != nil { - apiError(c, http.StatusInternalServerError, "channel_error", "failed to decrypt channel key") - return + return false, true, http.StatusInternalServerError, []byte("failed to decrypt channel key") } upBody := plan.body @@ -72,8 +105,7 @@ func (g *Gateway) doProxy(c *gin.Context, ch *store.Channel, plan *upstreamPlan, defer cancel() req, err := http.NewRequestWithContext(ctx, http.MethodPost, upstreamURL(ch, plan.path), bytes.NewReader(upBody)) if err != nil { - apiError(c, http.StatusInternalServerError, "internal_error", "failed to build upstream request") - return + return false, false, http.StatusInternalServerError, []byte("failed to build upstream request") } req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+upKey) @@ -84,7 +116,6 @@ func (g *Gateway) doProxy(c *gin.Context, ch *store.Channel, plan *upstreamPlan, if plan.path == "/v1/messages" { req.Header.Set("anthropic-version", "2023-06-01") } - // 透传 OpenAI 生态请求头(组织/项目等) for _, h := range []string{"OpenAI-Organization", "OpenAI-Project", "OpenAI-Beta"} { if v := c.GetHeader(h); v != "" { req.Header.Set(h, v) @@ -94,29 +125,20 @@ func (g *Gateway) doProxy(c *gin.Context, ch *store.Channel, plan *upstreamPlan, start := time.Now() resp, err := g.hc.Do(req) if err != nil { - status := http.StatusBadGateway - msg := "Upstream request failed: " + err.Error() if ctx.Err() == context.DeadlineExceeded { - status = http.StatusGatewayTimeout - msg = "Upstream request timed out" + return false, true, http.StatusGatewayTimeout, []byte("upstream request timed out") } - apiError(c, status, "upstream_error", msg) - g.recordError(c, ch, nil, start, "upstream_error") - return + return false, true, http.StatusBadGateway, []byte("upstream request failed: " + err.Error()) } defer resp.Body.Close() - // 非 2xx:透传上游错误体,并记录 error 用量 + // 非 2xx:429/5xx 可重试;其余透传错误体 if resp.StatusCode < 200 || resp.StatusCode >= 300 { errBody, _ := io.ReadAll(resp.Body) - status := resp.StatusCode - if status >= 500 { - status = http.StatusBadGateway + if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { + return false, true, resp.StatusCode, errBody } - c.Header("Content-Type", "application/json") - c.DataFromReader(status, int64(len(errBody)), "application/json", bytes.NewReader(errBody), nil) - g.recordError(c, ch, resp, start, "upstream_http_"+strconv.Itoa(resp.StatusCode)) - return + return false, false, resp.StatusCode, errBody } c.Header("Content-Type", resp.Header.Get("Content-Type")) @@ -126,6 +148,7 @@ func (g *Gateway) doProxy(c *gin.Context, ch *store.Channel, plan *upstreamPlan, } else { g.copyAndCapture(c, ch, resp.Body, start, plan.bodyConv, sink) } + return true, false, http.StatusOK, nil } // copyAndCapture 非流式:整体转发(可转换)+ 解析 usage + 记账。