feat: 补齐上游细节、渠道观测与模型定价元数据
- 上游 429 的 Retry-After 转成渠道冷却时长（夹在 10 分钟内；比默认值短则不缩短， 否则 1s 就把坏渠道放回轮转）。失败计数改由 proxyOne 就地报——只有它读得到响应头。 - 故障转移时间预算 OT_PROXY_FAILOVER_BUDGET（默认 30s）：只约束第二个及以后的渠道， 首个不受限（长回答本就要跑几分钟），但坏渠道不再把客户端拖到 N×timeout。 - 流式空闲心跳 OT_PROXY_STREAM_HEARTBEAT（默认 15s）：补一行 SSE 注释 `: ping`， 防止 nginx/Cloudflare 的空闲超时掐掉长思考。心跳与正文共用一把写的锁， gin 的 ResponseWriter 不是并发安全的；注释行不进转换也不计费。 - 渠道观测：GET /api/v1/admin/channels/stats?window=5m|1h|6h|24h|7d 聚合请求量、 成功率、P50/P95/max 延迟、成本与 top_errors；分位只取成功样本，canceled（客户端 主动断开）不计入渠道错误率。渠道页桌面表格加「近N小时」列、移动端加同内容条， 顶部可切窗口。 - 修一个让统计瞎掉的归因问题：全部渠道失败时 recordError 传的是 nil， usage_logs 里 channel_id=0，这类失败在渠道维度完全看不见。改为归到最后一个 真正试过的渠道。 - 模型定价与元数据：Model 新增 cache_write_price（0 则回退输入价×1.25，可显式配 Anthropic 1h 的 ×2）、context_length 与 supports_tools/vision/reasoning； finishUsage 走 CacheWritePriceOr()。/v1/models 叠加用户与密钥白名单过滤 （原来会把限制外的模型名直接泄漏在列表里）并返回元数据，价格不外泄。 - /v1/models 的 Anthropic 分支原是死代码：协议只按路径判，而两种客户端共用这个路径。 改为按 x-api-key / anthropic-version 头判，OpenAI SDK 从不发这两个头，互不影响。 - NewGateway 的十个位置参数换成 Options 结构体（三个 Duration 两个 int，传错编译期发现不了）。 新增 6 个用例：Retry-After 两种写法与夹顶、重试预算截止、空闲心跳出帧、 /v1/models 过滤与双形状、缓存写价两条路径、失败归因。
This commit is contained in:
Sakurasan committed
1 parent
ad10bdc400
commit
c39c3e7da4
16 files changed
+984
-125
No files matched your search
@@ -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 // 无定价模型时按示例价
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user