Sakurasan
9f4d631fc4
路由与故障转移（参考 openteam 语义） - channel.Candidates：绑定模型优先（携带 upstream_model 映射）， 未绑定模型回退到权重最低的健康备用渠道；新增 Pick 加权随机与 FilterHealthy 内存健康过滤 - gateway.Dispatch：遍历候选渠道，可重试失败（连接错误/429/5xx）自动故障转移， 4xx 透传；不再使用单一 SelectChannel - 修复 gorm default 标签把渠道 weight=0 静默改写为 1 的问题（去掉 default， 权重 0 语义 = 不参与加权选择，仅作备用承接 unbound 流量） - RecordFailure 连续 2 次进入 degraded 快速熔断，健康检查成功或冷却过期后复位 网关功能补全 - /v1/models 返回 DB 中启用的模型列表（替换 TODO 存根） - 请求级 request_id 生成与用量记录接入：流式 SSE 逐块累计 usage、 非流式从响应提取，按模型定价计算成本后经 usage.Recorder 异步落库 - 流式结束检测：chat 的 [DONE]、messages 的 message_stop、responses 的 response.completed，避免 keep-alive 上游发完不关连接导致读阻塞到超时 - ResponsesRequest.input 兼容字符串与条目数组两种客户端写法 测试 - 修复 convert_test 对新 input 形态的断言 - 网关 e2e（/tmp/test_gateway.py + mock upstream）72/72 全部通过，连续 3 次稳定
130 lines
3.3 KiB
Go
130 lines
3.3 KiB
Go
package proxy
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"log"
|
|
"net/http"
|
|
"net/url"
|
|
"opencatd-open/internal/channel"
|
|
"opencatd-open/internal/dao"
|
|
"opencatd-open/internal/store"
|
|
"opencatd-open/pkg/config"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type Proxy struct {
|
|
ctx context.Context
|
|
cfg *config.Config
|
|
db *gorm.DB
|
|
wg *sync.WaitGroup
|
|
httpClient *http.Client
|
|
|
|
userDAO *dao.UserDAO
|
|
apiKeyDAO *dao.ApiKeyDAO
|
|
usageDAO *dao.UsageDAO
|
|
dailyDAO *dao.DailyUsageDAO
|
|
channelSvc *channel.Service
|
|
}
|
|
|
|
func NewProxy(ctx context.Context, cfg *config.Config, db *gorm.DB, wg *sync.WaitGroup, userDAO *dao.UserDAO, apiKeyDAO *dao.ApiKeyDAO, usageDAO *dao.UsageDAO, dailyDAO *dao.DailyUsageDAO) *Proxy {
|
|
client := http.DefaultClient
|
|
if os.Getenv("LOCAL_PROXY") != "" {
|
|
proxyUrl, err := url.Parse(os.Getenv("LOCAL_PROXY"))
|
|
if err == nil {
|
|
tr := &http.Transport{
|
|
Proxy: http.ProxyURL(proxyUrl),
|
|
}
|
|
client.Transport = tr
|
|
}
|
|
}
|
|
|
|
np := &Proxy{
|
|
ctx: ctx,
|
|
cfg: cfg,
|
|
db: db,
|
|
wg: wg,
|
|
httpClient: client,
|
|
userDAO: userDAO,
|
|
apiKeyDAO: apiKeyDAO,
|
|
usageDAO: usageDAO,
|
|
dailyDAO: dailyDAO,
|
|
}
|
|
|
|
return np
|
|
}
|
|
|
|
// SetChannelService sets the channel service (called after construction)
|
|
func (p *Proxy) SetChannelService(svc *channel.Service) {
|
|
p.channelSvc = svc
|
|
}
|
|
|
|
func (p *Proxy) HandleProxy(c *gin.Context) {
|
|
path := c.Request.URL.Path
|
|
switch {
|
|
case path == "/v1/chat/completions":
|
|
// TODO: Phase 3 - implement chat completions handler
|
|
c.JSON(http.StatusNotImplemented, gin.H{"error": "chat completions not yet implemented"})
|
|
case strings.HasPrefix(path, "/v1/messages"):
|
|
// TODO: Phase 3 - implement messages handler
|
|
c.JSON(http.StatusNotImplemented, gin.H{"error": "messages not yet implemented"})
|
|
case path == "/v1/responses":
|
|
// TODO: Phase 3 - implement responses handler
|
|
c.JSON(http.StatusNotImplemented, gin.H{"error": "responses not yet implemented"})
|
|
default:
|
|
c.JSON(http.StatusNotFound, gin.H{"error": "unknown endpoint"})
|
|
}
|
|
}
|
|
|
|
func (p *Proxy) HandleModels(c *gin.Context) {
|
|
// TODO: Phase 3 - implement models list
|
|
c.JSON(http.StatusOK, gin.H{"object": "list", "data": []interface{}{}})
|
|
}
|
|
|
|
func (p *Proxy) GetDB() *gorm.DB {
|
|
return p.db
|
|
}
|
|
|
|
// SelectChannel selects the best channel for a model
|
|
func (p *Proxy) SelectChannel(modelName string) (*store.Channel, error) {
|
|
if p.channelSvc == nil {
|
|
return nil, fmt.Errorf("channel service not initialized")
|
|
}
|
|
cands := p.channelSvc.Candidates(modelName)
|
|
picked := p.channelSvc.Pick(cands)
|
|
if picked == nil {
|
|
return nil, fmt.Errorf("no enabled channels for model: %s", modelName)
|
|
}
|
|
return picked.Channel, nil
|
|
}
|
|
|
|
// RecordSuccess records a successful request
|
|
func (p *Proxy) RecordSuccess(channelID uint64) {
|
|
if p.channelSvc != nil {
|
|
p.channelSvc.RecordSuccess(channelID)
|
|
}
|
|
}
|
|
|
|
// RecordFailure records a failed request
|
|
func (p *Proxy) RecordFailure(channelID uint64) {
|
|
if p.channelSvc != nil {
|
|
p.channelSvc.RecordFailure(channelID)
|
|
}
|
|
}
|
|
|
|
// SendUsagePlaceholder placeholder for usage processing
|
|
func (p *Proxy) SendUsagePlaceholder(model string, userID uint64, promptTokens, completionTokens int) {
|
|
log.Printf("Usage: model=%s user=%d prompt=%d completion=%d", model, userID, promptTokens, completionTokens)
|
|
}
|
|
|
|
// Placeholder to keep the file compilable
|
|
var _ = json.Marshal
|
|
var _ = io.ReadAll
|