Merge pull request #22 from qianfree/dev

修复渠道健康度问题
This commit is contained in:
Qian authored and GitHub committed 2026-08-27 23:50:26 +08:00
commit d9c520401f
42 files changed
+1546 -234

No files matched your search

+8 -6
View File
@@ -47,6 +47,8 @@ type ChannelItem struct {
HealthScore *float64 `json:"health_score"`
BreakerState int `json:"breaker_state"` // 调度熔断状态:0=正常 1=熔断中 2=半开探活
BreakerModels int `json:"breaker_models"` // 处于熔断/半开的模型数量(渠道×模型级汇总)
WorstModel string `json:"worst_model"` // 健康度最低的模型名(空=渠道不在调度目录)
WorstModelScore float64 `json:"worst_model_score"`
}
// ChannelCreateReq 创建渠道请求
@@ -143,6 +145,8 @@ type ChannelDetailRes struct {
HealthScore *float64 `json:"health_score"`
BreakerState int `json:"breaker_state"` // 调度熔断状态:0=正常 1=熔断中 2=半开探活
BreakerModels int `json:"breaker_models"` // 处于熔断/半开的模型数量(渠道×模型级汇总)
WorstModel string `json:"worst_model"` // 健康度最低的模型名(空=渠道不在调度目录)
WorstModelScore float64 `json:"worst_model_score"`
KeyType string `json:"key_type"`
KeyStatus string `json:"key_status"`
KeyName string `json:"key_name"`
@@ -258,12 +262,10 @@ type ChannelHealthTrendRes struct {
// HealthTrendPoint 健康趋势数据点(健康度、延迟取整展示,不含小数)
type HealthTrendPoint struct {
SnapshotAt string `json:"snapshot_at"`
HealthScore int64 `json:"health_score"`
SuccessRate float64 `json:"success_rate"`
LatencyMs int64 `json:"latency_ms"`
StabilityScore float64 `json:"stability_score"`
ConsecutiveFailures int `json:"consecutive_failures"`
SnapshotAt string `json:"snapshot_at"`
HealthScore int64 `json:"health_score"`
SuccessRate float64 `json:"success_rate"`
LatencyMs int64 `json:"latency_ms"`
}
// ChannelExportReq 导出渠道列表请求
+26
View File
@@ -789,3 +789,29 @@ err := g.DB().Transaction(dbCtx, func(txCtx context.Context, tx gdb.TX) error {
**修复**:`internal/dispatchadapter/state_redis.go` `ReadRuntime`/`readBreaker` 改为 `vals[i].String() != ""` 判有无数据;空串一律按无数据处理保留默认值。回归测试 `Test缺失健康键读默认值`。
**正确做法(通用规则)**:判断 gf redis `HMGET`/批量读的字段是否存在,**不要依赖 `IsNil()`**(驱动转换后 nil 会变空字符串);用 `String() != ""` 判断,或改用 Lua 脚本内 `HGET ... or '默认值'` 在 Redis 侧兜底。单 key `HGET` 缺失返回的是真 nil(`IsNil()==true`),两者行为不一致,混用时尤其注意。
### 2026-08-27:Redis HMGET 缺失字段的 gvar 判空失效(`IsNil()` 恒为 false),默认值兜底全线失守
**问题**:渠道探测一切正常(Redis succ_ewma=1.0),管理后台渠道健康度却长期停在 25,其余渠道全是 0。排查发现维护快照读到的几乎所有「从未被真实流量访问过」的渠道×模型健康值都是 `0.000`,而这些模型的 Redis key **根本不存在**——按设计应回落到默认满分 1.0。
**原因**:`RedisState.ReadRuntime` 用如下写法读健康 EWMA:
```go
v, _ := g.Redis().Do(ctx, "HMGET", key, "succ_ewma", "lat_ewma")
vals := v.Vars()
if !vals[0].IsNil() { out.SuccEwma = vals[0].Float64() } // ← 判空失效
```
HMGET 对不存在的 key/字段返回 Redis nil,但 gredis 把它们包装成**非 nil 的空 gvar**:实测 `IsNil()=false`、`IsEmpty()=true`、`String()=""`、`Float64()=0`。于是判空分支永远进入,`Float64("")` 静默返回 **0**,把「无数据」变成了「健康分 0 分」。影响远超展示:`ReadRuntime` 同时供调度目录构建 `healthFactor`,所有冷模型的路由权重被压成 0。
两个读取方对同一份「无数据」表现还不一致,极大增加了排查难度——管理后台 `batchGetModelHealth` 用 `fmt.Sscanf` 解析,解析失败就不赋值,UI 显示"无数据";维护循环用 gconv 读成 0。同一个模型,UI 说无数据、健康分算 0。
**修复**:`state_redis.go` 的 `ReadRuntime` 改为显式字符串校验——`TrimSpace(String())` 为空则保持默认值,非空再 `strconv.ParseFloat` 并做值域校验(succ 取 `0 < f <= 1`;EWMA 由正数衰减而来,数学上不会精确到达 0,精确 0 视为异常数据)。回归测试 `TestReadRuntime脏数据防御`。
**正确做法(通用规则)**:
- **禁止用 `gvar.IsNil()` 判断 Redis 读取结果是否存在**——gredis 的空回复是非 nil 空 gvar,`IsNil()` 恒为 false。用 `IsEmpty()`,或取 `String()` 后判空串。
- **禁止把 `gvar.Float64()`/`Int()` 直接用于「缺失应回落默认值」的场景**:gconv 对空串/非法值静默返回零值,会把「无数据」变成「0 分/0 次」这类**具有强业务含义的极端值**。默认值不是 0 的字段(健康分、成功率、权重、配额),必须显式校验后再赋值。
- 同一份状态有多个读取方时(调度用 gconv、管理后台用 Sscanf),**解析方式必须统一**,否则同一数据在不同页面表现不同,故障期误导排查方向。
排查信号:某项指标「本该回落默认值却全是 0」;同一数据 UI 显示"无数据"而后台计算按 0 处理;`grep -n 'IsNil()' ` 命中 Redis 读取路径。
@@ -9,6 +9,14 @@
>
> **分析时间**:2026-07-30。
> ⚠️ **team-api 部分已过时(2026-08-27 标注)**:本文描述的是渠道调度重写**之前**的实现,
> 文中引用的 `relay/scheduler/scheduler.go`、`internal/logic/relay/health.go` 均已删除。
> 特别注意第 5 节的「4 因子加权健康度」公式(成功率×0.40 + 延迟分×0.25 + 稳定性×0.20 +
> 连续失败分×0.15)**已废止**,现行健康体系为:Redis 渠道×模型 EWMA 为唯一真相,
> 健康分 = `avg(succ_ewma)^α × 100`,调度权重见 `relaykit/dispatch/score.go` 的
> `EffectiveWeight`,落盘见 `internal/dispatchadapter/maintenance.go`。
> new-api / sub2api 两侧的分析不受影响,横向对比结论仍有参考价值。
---
## 目录
+117 -8
View File
@@ -2,6 +2,8 @@ package cmd
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io/fs"
"net/http"
@@ -11,6 +13,7 @@ import (
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/net/ghttp"
"github.com/gogf/gf/v2/os/gcmd"
"github.com/gogf/gf/v2/os/gctx"
"github.com/qianfree/team-api/internal/consts"
adminController "github.com/qianfree/team-api/internal/controller/admin"
@@ -333,9 +336,37 @@ func printBanner() {
fmt.Println()
}
// channelAutoTestSchedule 按配置的探测间隔生成 cron 表达式。
//
// 用 @every 而非 */N:*/N 只在 N < 60 时有意义(*/90 在分钟位上等价于 */30,
// 静默给出与配置不符的频率),而 @every 支持任意时长。副作用是触发时刻相对进程
// 启动时间而非整点对齐——多实例部署时反而更好,不会所有实例卡在同一分钟同时探测。
func channelAutoTestSchedule(ctx context.Context) string {
minutes := common.Config().GetInt(ctx, "channel_auto_test_interval_minutes")
// 配置缺失(GetInt 返回 0)或越界时回落默认值,避免非法表达式让任务注册失败
if minutes < 1 || minutes > 1440 {
minutes = 5
}
return fmt.Sprintf("@every %dm", minutes)
}
// watchChannelAutoTestInterval 探测间隔配置变更时重排 cron,无需重启。
// 回调经 Redis pub/sub 触发,多实例会各自重排自己的调度器。
func watchChannelAutoTestInterval(cs *common.CronScheduler) {
common.OnSettingsChanged(func(ctx context.Context, key string) {
if key != "channel_auto_test_interval_minutes" {
return
}
if err := cs.Reschedule(ctx, "channel_auto_test", channelAutoTestSchedule(ctx)); err != nil {
g.Log().Errorf(ctx, "渠道自动探测重排失败,沿用原间隔: %v", err)
}
})
}
// registerCronJobs 集中注册所有定时任务,避免散落在主启动流程中。
// 新增 cron 任务时在此追加一项即可;任务名(用于分布式锁 key)需保持唯一且稳定。
func registerCronJobs(cs *common.CronScheduler) {
watchChannelAutoTestInterval(cs)
cs.Register("ops_system_collector", "系统指标采集", "* * * * *", func(ctx context.Context) error {
return monitor.CollectSystemMetrics(ctx)
})
@@ -351,7 +382,7 @@ func registerCronJobs(cs *common.CronScheduler) {
cs.Register("health_snapshot", "渠道健康快照", "*/5 * * * *", func(ctx context.Context) error {
return task.SnapshotHealthScores(ctx)
})
cs.Register("channel_auto_test", "渠道自动测试", "*/5 * * * *", func(ctx context.Context) error {
cs.Register("channel_auto_test", "渠道自动测试", channelAutoTestSchedule(gctx.New()), func(ctx context.Context) error {
if common.Config().GetBool(ctx, "channel_auto_test_enabled") {
task.AutoTestChannels(ctx)
}
@@ -521,6 +552,54 @@ func registerFrontendRoutes(s *ghttp.Server) {
})
}
// staticAssetExts 视为静态资源的扩展名集合:此类请求未命中文件时应返回 404
// 而非回退 index.html,避免把 HTML 当 JS/JSON 返回触发浏览器严格 MIME 校验报错。
var staticAssetExts = map[string]struct{}{
".js": {}, ".mjs": {}, ".css": {}, ".wasm": {}, ".map": {},
".jpg": {}, ".jpeg": {}, ".png": {}, ".gif": {}, ".webp": {}, ".avif": {},
".svg": {}, ".ico": {}, ".woff": {}, ".woff2": {}, ".ttf": {}, ".otf": {}, ".eot": {},
".mp4": {}, ".webm": {}, ".mp3": {}, ".wav": {}, ".json": {}, ".xml": {}, ".txt": {},
}
// isAssetPath 判断请求路径是否具有静态资源特征(assets 目录或带资源扩展名)。
func isAssetPath(path string) bool {
if strings.HasPrefix(path, "assets/") {
return true
}
ext := strings.ToLower(path[strings.LastIndex(path, "/")+1:])
dot := strings.LastIndex(ext, ".")
if dot <= 0 {
return false
}
_, ok := staticAssetExts[ext[dot:]]
return ok
}
// setStaticCacheHeader 按资源类型设置缓存策略:
// Vite 构建产物(assets/ 下 hash 文件名)永久强缓存;入口文件(index.html 等)要求每次进站重新校验,
// 保证发版后用户总能拿到新入口。
func setStaticCacheHeader(w http.ResponseWriter, path string) {
if strings.HasPrefix(path, "assets/") {
w.Header().Set("Cache-Control", "public, max-age=31536000, immutable")
return
}
w.Header().Set("Cache-Control", "no-cache")
}
// matchesEtag 判断请求头 If-None-Match 是否包含当前指纹(值可能是逗号分隔列表或通配 *)。
func matchesEtag(header, tag string) bool {
if header == "" {
return false
}
for _, part := range strings.Split(header, ",") {
part = strings.TrimSpace(part)
if part == "*" || part == tag {
return true
}
}
return false
}
// spaHandler returns an http.HandlerFunc that serves static files from the
// given filesystem, falling back to index.html for SPA client-side routing.
func spaHandler(root fs.FS, prefix string) http.HandlerFunc {
@@ -529,28 +608,58 @@ func spaHandler(root fs.FS, prefix string) http.HandlerFunc {
fileServer = http.StripPrefix(prefix, fileServer)
}
// 启动时预读入口文件并预计算 SHA-256 指纹作为 ETag:
// index.html 走 no-cache 协商缓存,未重新发布的重复访问返回 304 空响应,
// 既保证发版即生效、又不产生重复下载开销。
indexBytes, indexErr := fs.ReadFile(root, "index.html")
indexTag := ""
if indexErr == nil {
sum := sha256.Sum256(indexBytes)
indexTag = `"` + hex.EncodeToString(sum[:]) + `"`
}
serveIndex := func(w http.ResponseWriter, r *http.Request) {
if indexErr != nil {
http.NotFound(w, r)
return
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("ETag", indexTag)
if matchesEtag(r.Header.Get("If-None-Match"), indexTag) {
w.WriteHeader(http.StatusNotModified)
return
}
w.Write(indexBytes)
}
return func(w http.ResponseWriter, r *http.Request) {
// Resolve the file path within the embedded FS
path := strings.TrimPrefix(r.URL.Path, prefix)
path = strings.TrimPrefix(path, "/")
if path == "" {
path = "index.html"
// 入口文件统一走自建响应(fileServer 无法为 embed FS 提供 Last-Modified/ETag)
if path == "" || path == "index.html" {
serveIndex(w, r)
return
}
// Try to open the file; if it exists, serve it directly
if f, err := root.Open(path); err == nil {
f.Close()
setStaticCacheHeader(w, path)
fileServer.ServeHTTP(w, r)
return
}
// File not found — serve index.html (SPA fallback)
indexBytes, err := fs.ReadFile(root, "index.html")
if err != nil {
// 带静态资源特征的请求(如发版后旧 hash chunk)缺失时返回 404,
// 不回退 SPA,方便前端自愈逻辑识别并整页刷新到新版本。
if isAssetPath(path) {
http.NotFound(w, r)
return
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
w.Write(indexBytes)
// Route not found — serve index.html (SPA fallback)
serveIndex(w, r)
}
}
+141
View File
@@ -0,0 +1,141 @@
package cmd
import (
"io/fs"
"net/http"
"net/http/httptest"
"strings"
"testing"
"testing/fstest"
)
// 构造模拟的前端构建产物:入口文件 + 一个带 hash 的静态 chunk
func newSpaTestFS() fs.FS {
return fstest.MapFS{
"index.html": &fstest.MapFile{
Data: []byte("<!doctype html><html><head><script src=\"/assets/index-x9y8z7.js\"></script></head><body></body></html>"),
},
"assets/index-x9y8z7.js": &fstest.MapFile{Data: []byte("console.log('chunk')")},
"favicon.ico": &fstest.MapFile{Data: []byte("ico")},
"screenshots/page-a.webp": &fstest.MapFile{Data: []byte("img")},
}
}
func get(t *testing.T, h http.HandlerFunc, target string, headers map[string]string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodGet, target, nil)
for k, v := range headers {
req.Header.Set(k, v)
}
rec := httptest.NewRecorder()
h(rec, req)
return rec
}
func assertContains(t *testing.T, rec *httptest.ResponseRecorder, key, want string) {
t.Helper()
if got := rec.Header().Get(key); got != want {
t.Errorf("%s = %q, 期望 %q", key, got, want)
}
}
// 路由请求回退 index.html,且带 no-cache 与 ETag
func TestSpaHandlerRouteFallback(t *testing.T) {
h := spaHandler(newSpaTestFS(), "")
rec := get(t, h, "/dashboard", nil)
if rec.Code != http.StatusOK {
t.Fatalf("状态码 = %d, 期望 200", rec.Code)
}
if !strings.Contains(rec.Body.String(), "index-x9y8z7.js") {
t.Error("响应体应为 index.html 内容")
}
assertContains(t, rec, "Cache-Control", "no-cache")
if rec.Header().Get("ETag") == "" {
t.Error("index.html 应携带 ETag")
}
}
// 根路径与显式 /index.html 都走同一入口响应
func TestSpaHandlerIndexPaths(t *testing.T) {
h := spaHandler(newSpaTestFS(), "")
for _, target := range []string{"/", "/index.html"} {
rec := get(t, h, target, nil)
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "<!doctype html>") {
t.Errorf("%s 应返回 200 的 index.html,实际 %d", target, rec.Code)
}
}
}
// 未重新发布时重复访问命中 ETag 协商缓存,返回 304 空响应
func TestSpaHandlerEtagNegotiation(t *testing.T) {
h := spaHandler(newSpaTestFS(), "")
first := get(t, h, "/", nil)
tag := first.Header().Get("ETag")
rec := get(t, h, "/dashboard", map[string]string{"If-None-Match": tag})
if rec.Code != http.StatusNotModified {
t.Fatalf("状态码 = %d, 期望 304 Not Modified", rec.Code)
}
if rec.Body.Len() != 0 {
t.Error("304 响应不应携带 body")
}
}
// 带静态资源特征的缺失文件返回 404 而非回退 SPA HTML(发版后旧 hash chunk 场景)
func TestSpaHandlerMissingAssetReturns404(t *testing.T) {
h := spaHandler(newSpaTestFS(), "")
cases := []string{
"/assets/stale-old-hash.js",
"/assets/some.css",
"/assets/vendor.wasm",
"/images/missing.png", // 不存在的资源扩展名
"/manifest.json",
}
for _, target := range cases {
rec := get(t, h, target, nil)
if rec.Code != http.StatusNotFound {
t.Errorf("%s 应返回 404,实际 %d", target, rec.Code)
}
if ct := rec.Header().Get("Content-Type"); strings.Contains(ct, "text/html") {
t.Errorf("%s 的 404 不应是 text/html", target)
}
}
}
// 无扩展名的未知路径仍回退 index.html(深层路由刷新场景)
func TestSpaHandlerUnknownRouteStillFallbacks(t *testing.T) {
h := spaHandler(newSpaTestFS(), "")
rec := get(t, h, "/team/member-detail/42", nil)
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "index") {
t.Fatalf("深层路由应回退 index.html,实际 %d", rec.Code)
}
}
// 存在的 assets 文件命中后强缓存一年;根级非 hash 文件要求协商校验
func TestSpaHandlerCacheHeaders(t *testing.T) {
h := spaHandler(newSpaTestFS(), "")
rec := get(t, h, "/assets/index-x9y8z7.js", nil)
assertContains(t, rec, "Cache-Control", "public, max-age=31536000, immutable")
rec = get(t, h, "/favicon.ico", nil)
assertContains(t, rec, "Cache-Control", "no-cache")
}
// 带 URL 前缀的 admin 子应用:前缀剥离、前缀下的 404 与缓存规则同样生效
func TestSpaHandlerWithPrefix(t *testing.T) {
h := spaHandler(newSpaTestFS(), "/admin")
rec := get(t, h, "/admin/", nil)
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "<!doctype html>") {
t.Fatalf("/admin/ 应返回 index.html,实际 %d", rec.Code)
}
rec = get(t, h, "/admin/assets/missing.js", nil)
if rec.Code != http.StatusNotFound {
t.Errorf("/admin/assets/missing.js 应返回 404,实际 %d", rec.Code)
}
rec = get(t, h, "/admin/assets/index-x9y8z7.js", nil)
assertContains(t, rec, "Cache-Control", "public, max-age=31536000, immutable")
}
+6 -6
View File
@@ -23,12 +23,12 @@ type ChnHealthScoresDao struct {
type ChnHealthScoresColumns struct {
Id string // 主键ID
ChannelId string // 关联渠道ID
SuccessRate string // 请求成功率(0-100)
LatencyMs string // 平均延迟(毫秒)
StabilityScore string // 稳定性评分(0-100,基于延迟波动计算)
ConsecutiveFailures string // 连续失败次数(成功后归零)
HealthScore string // 综合健康度(0-100)= 成功率×0.40 + 延迟分×0.25 + 稳定性×0.20 + 连续失败分×0.15
CalculatedAt string // 最近一次计算时间
SuccessRate string // 成功率(0-100)= 该渠道各模型 succ_ewma 均值×100,仅统计有真实上报的模型
LatencyMs string // 平均延迟(毫秒)= 该渠道各模型 lat_ewma 均值,仅统计有真实上报的模型;仅展示用,不参与健康分计算
StabilityScore string // 【已废弃】稳定性评分,健康体系重写后不再写入,保留列仅为兼容历史数据
ConsecutiveFailures string // 【已废弃】连续失败次数,已由 Redis 熔断器的滑动窗口计数取代(dispatch:v1:breaker:*),保留列仅为兼容历史数据
HealthScore string // 综合健康度(0-100)= avg(succ_ewma)^α × 100,α 取路由策略 health.alpha(默认 2),与调度 healthFactor 同源;每 5 分钟由维护任务聚合落盘,另可由渠道测试/重置健康度即时触发;Redis 读失败或全部模型无真实上报时保留旧值不覆盖。调度决策不读此表,仅供管理后台展示
CalculatedAt string // 最近一次聚合落盘时间
CreatedAt string // 创建时间
UpdatedAt string // 更新时间
}
@@ -23,11 +23,11 @@ type ChnHealthSnapshotsDao struct {
type ChnHealthSnapshotsColumns struct {
Id string // 主键ID
ChannelId string // 关联渠道ID
HealthScore string // 综合健康度(0-100)
HealthScore string // 综合健康度(0-100)快照,取自 chn_health_scores.health_score
SuccessRate string // 请求成功率(0-100)
LatencyMs string // 平均延迟(毫秒)
StabilityScore string // 稳定性评分(0-100)
ConsecutiveFailures string // 连续失败次数
StabilityScore string // 【已废弃】稳定性评分,不再写入,新快照为 0
ConsecutiveFailures string // 【已废弃】连续失败次数,不再写入,新快照为 0
SnapshotAt string // 快照时间
}
+1 -1
View File
@@ -27,7 +27,7 @@ type MdlTenantModelsColumns struct {
Enabled string // 是否启用(禁用后该租户无法调用此模型)
CustomInputPrice string // 租户自定义输入价格(NULL 表示使用默认定价)
CustomOutputPrice string // 租户自定义输出价格(NULL 表示使用默认定价)
Multiplier string // 租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与最终价格计算:最终价格 = 基础价格 × 租户倍率。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier
Multiplier string // 租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与计费,当前实际生效公式 = 基础价格 × 租户乘数 × 时段乘数;模型乘数为预留能力(mdl_models 无 multiplier 字段、computeCost 未接入),bil_records.model_multiplier 快照恒为 1.0。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier(级别兜底仅在前两者均未设置、且级别倍率 <1.0 时生效)
CreatedAt string // 创建时间
UpdatedAt string // 更新时间
BillingMode string // 覆盖模型计费方式(NULL表示跟随模型默认)
+66 -6
View File
@@ -2,6 +2,7 @@ package dispatchadapter
import (
"context"
"math"
"slices"
"sync/atomic"
"time"
@@ -157,6 +158,13 @@ func (c *Catalog) ChannelModels() map[int64][]string {
type ChannelRuntimeState struct {
Breaker dispatch.BreakerState // 渠道级熔断状态(同一渠道所有模型取值相同)
BreakerModels int // 处于 OPEN/HALF_OPEN 的模型数量(渠道×模型级汇总)
// WorstModel / WorstModelScore 健康度最低的模型及其健康分(succ_ewma^α×100,
// 与渠道聚合分同口径)。渠道分是均值,单个模型全挂会被其余健康模型摊薄——10 个
// 健康模型 + 1 个挂掉仍能算出 90 分绿灯,最差项让这种情况在列表页直接可见。
// 取最小值天然免疫冷模型问题:无上报的模型读作满分 1.0,永远不会成为最差项。
WorstModel string
WorstModelScore float64
}
// ChannelRuntimeStates 返回目录快照中各渠道的熔断聚合视图。
@@ -170,19 +178,41 @@ func (c *Catalog) ChannelRuntimeStates() map[int64]ChannelRuntimeState {
if idx == nil {
return nil
}
out := make(map[int64]ChannelRuntimeState)
for _, chans := range idx.byModel {
type accEntry struct {
st ChannelRuntimeState
worstSucc float64
found bool
}
acc := make(map[int64]*accEntry)
for model, chans := range idx.byModel {
for _, ch := range chans {
agg, ok := out[ch.ID]
a, ok := acc[ch.ID]
if !ok {
agg = ChannelRuntimeState{Breaker: ch.Breaker}
a = &accEntry{st: ChannelRuntimeState{Breaker: ch.Breaker}}
acc[ch.ID] = a
}
if ch.ModelBreaker != dispatch.BreakerClosed {
agg.BreakerModels++
a.st.BreakerModels++
}
// 同分时按模型名取小者:byModel 是 map,遍历顺序随机,不做确定性
// 打破平局的话,全部模型同分(例如都满分)时每次请求返回的模型名都会跳变。
if !a.found || ch.SuccEwma < a.worstSucc ||
(ch.SuccEwma == a.worstSucc && model < a.st.WorstModel) {
a.found = true
a.worstSucc = ch.SuccEwma
a.st.WorstModel = model
}
out[ch.ID] = agg
}
}
alpha := HealthAlpha()
out := make(map[int64]ChannelRuntimeState, len(acc))
for id, a := range acc {
if a.found {
a.st.WorstModelScore = math.Pow(clampFloat(a.worstSucc, 0, 1), alpha) * 100
}
out[id] = a.st
}
return out
}
@@ -259,6 +289,20 @@ func (c *Catalog) subscribeInvalidate(ctx context.Context) {
}
}
// healthOf 从快照中查渠道×模型的健康读值(Redis 降级时沿用 last-known)。
// byModel 每个模型下同一渠道至多一行,线性扫描只发生在降级路径,不影响正常刷新。
func (idx *snapshotIndex) healthOf(channelID int64, model string) (succ, lat float64, ok bool) {
if idx == nil {
return 0, 0, false
}
for _, ch := range idx.byModel[model] {
if ch.ID == channelID {
return ch.SuccEwma, ch.LatEwmaMs, true
}
}
return 0, 0, false
}
// Rebuild 全量重建快照:DB 目录 + Redis 运行时读值合并。
// 加载失败保留上一份快照(last-known,基线方案 §13)。
func (c *Catalog) Rebuild(ctx context.Context) {
@@ -268,6 +312,7 @@ func (c *Catalog) Rebuild(ctx context.Context) {
return
}
prev := c.current.Load()
pol := c.policy()
now := time.Now().UnixMilli()
rampWindowMs := int64(pol.Ramp.WindowSeconds) * 1000
@@ -284,6 +329,7 @@ func (c *Catalog) Rebuild(ctx context.Context) {
model string
}
rtCache := make(map[rtKey]RuntimeReadout)
degraded, carried := 0, 0
for _, row := range data.rows {
rk := rtKey{row.ChannelID, row.ModelName}
@@ -294,6 +340,16 @@ func (c *Catalog) Rebuild(ctx context.Context) {
} else {
rt = RuntimeReadout{SuccEwma: 1}
}
// Redis 健康读取失败:沿用上一份快照的健康读值。若让乐观默认值 succ=1 落进
// 新快照,Redis 抖一下就会把全渠道的健康记忆清零(healthFactor 集体回满分),
// 与注释声称的 last-known 语义相悖。
if rt.Degraded {
degraded++
if succ, lat, found := prev.healthOf(row.ChannelID, row.ModelName); found {
rt.SuccEwma, rt.LatEwmaMs = succ, lat
carried++
}
}
rtCache[rk] = rt
}
@@ -344,6 +400,10 @@ func (c *Catalog) Rebuild(ctx context.Context) {
}
}
c.current.Store(idx)
if degraded > 0 {
g.Log().Warningf(ctx, "[Dispatch] 目录刷新时 Redis 健康读取降级: %d 个渠道×模型读失败,其中 %d 个已沿用上一份快照的健康读值(其余无历史,按满分处理)",
degraded, carried)
}
}
// effectiveSoftLimit softLimit 双来源(基线方案 §8.2):
+102
View File
@@ -133,6 +133,48 @@ func TestCatalog_加载失败保留上一份快照(t *testing.T) {
assert.Len(t, c.Snapshot(ctx, 0, "m", nil), 1, "last-known 快照必须保留")
}
// TestCatalog_Redis降级沿用上一份健康读值 Redis 读失败时若让乐观默认值 succ=1 落进
// 新快照,一次 Redis 抖动就会把全渠道的健康记忆清零、healthFactor 集体回满分,
// 与 Rebuild 声称的 last-known 语义相悖。
func TestCatalog_Redis降级沿用上一份健康读值(t *testing.T) {
rows := []catalogRow{
{ChannelID: 1, ChannelName: "A", ModelName: "gpt-4o", Weight: 10, Tier: "primary", CostRatio: 1},
}
degraded := false
c := testCatalog(rows, nil, func(_ context.Context, _ int64, _ string) RuntimeReadout {
if degraded {
return RuntimeReadout{SuccEwma: 1, Degraded: true}
}
return RuntimeReadout{SuccEwma: 0.3, LatEwmaMs: 4000, HasHealth: true}
})
got := c.Snapshot(context.Background(), 0, "gpt-4o", nil)
require.Len(t, got, 1)
assert.Equal(t, 0.3, got[0].SuccEwma)
degraded = true
c.Rebuild(context.Background())
got = c.Snapshot(context.Background(), 0, "gpt-4o", nil)
require.Len(t, got, 1)
assert.Equal(t, 0.3, got[0].SuccEwma, "Redis 降级必须沿用 last-known,不得回落满分")
assert.Equal(t, 4000.0, got[0].LatEwmaMs, "延迟读值同样沿用")
}
// TestCatalog_首次刷新即降级无历史可沿用 新实例启动时 Redis 就不可用:没有 last-known
// 可沿用,保持乐观默认满分(不能让新实例启动即把全渠道判死)。
func TestCatalog_首次刷新即降级无历史可沿用(t *testing.T) {
rows := []catalogRow{
{ChannelID: 1, ChannelName: "A", ModelName: "gpt-4o", Weight: 10, Tier: "primary", CostRatio: 1},
}
c := testCatalog(rows, nil, func(_ context.Context, _ int64, _ string) RuntimeReadout {
return RuntimeReadout{SuccEwma: 1, Degraded: true}
})
got := c.Snapshot(context.Background(), 0, "gpt-4o", nil)
require.Len(t, got, 1)
assert.Equal(t, 1.0, got[0].SuccEwma, "无历史时保持乐观默认")
}
func TestRampElapsed(t *testing.T) {
now := time.Now().UnixMilli()
window := int64(120_000)
@@ -191,3 +233,63 @@ func TestCatalog_运行状态_未Rebuild返回空(t *testing.T) {
c := NewCatalog(func() *dispatch.RoutingPolicy { return dispatch.DefaultRoutingPolicy() }, nil, nil)
assert.Empty(t, c.ChannelRuntimeStates())
}
// TestCatalog_最差模型_单模型故障不被均值摊薄 渠道健康分是均值,10 个健康模型 + 1 个
// 全挂仍能算出 90 分绿灯;最差项要如实报出那个挂掉的模型。
func TestCatalog_最差模型_单模型故障不被均值摊薄(t *testing.T) {
rows := []catalogRow{
{ChannelID: 1, ChannelName: "A", ModelName: "healthy-a", Weight: 10, Tier: "primary"},
{ChannelID: 1, ChannelName: "A", ModelName: "healthy-b", Weight: 10, Tier: "primary"},
{ChannelID: 1, ChannelName: "A", ModelName: "broken", Weight: 10, Tier: "primary"},
}
c := testCatalog(rows, nil, func(_ context.Context, _ int64, model string) RuntimeReadout {
if model == "broken" {
return RuntimeReadout{SuccEwma: 0.1, HasHealth: true}
}
return RuntimeReadout{SuccEwma: 1.0, HasHealth: true}
})
got := c.ChannelRuntimeStates()
require.Len(t, got, 1)
assert.Equal(t, "broken", got[1].WorstModel)
// α 默认 2:0.1² × 100 = 1
assert.InDelta(t, 1.0, got[1].WorstModelScore, 1e-6)
}
// TestCatalog_最差模型_冷模型不被误报 无真实上报的模型读作满分 1.0,取最小值天然
// 不会把它选成最差项——这正是拖垮渠道均值的缺陷在 min 聚合下不存在的原因。
func TestCatalog_最差模型_冷模型不被误报(t *testing.T) {
rows := []catalogRow{
{ChannelID: 1, ChannelName: "A", ModelName: "cold", Weight: 10, Tier: "primary"},
{ChannelID: 1, ChannelName: "A", ModelName: "warm", Weight: 10, Tier: "primary"},
}
c := testCatalog(rows, nil, func(_ context.Context, _ int64, model string) RuntimeReadout {
if model == "cold" {
return RuntimeReadout{SuccEwma: 1} // HasHealth=false,从未上报
}
return RuntimeReadout{SuccEwma: 0.8, HasHealth: true}
})
got := c.ChannelRuntimeStates()
assert.Equal(t, "warm", got[1].WorstModel, "冷模型读作满分,不应被选为最差")
assert.InDelta(t, 64.0, got[1].WorstModelScore, 1e-6) // 0.8² × 100
}
// TestCatalog_最差模型_同分时结果稳定 byModel 是 map,遍历顺序随机;同分不做确定性
// 打破平局的话,全部模型满分时每次请求返回的模型名都会跳变。
func TestCatalog_最差模型_同分时结果稳定(t *testing.T) {
rows := []catalogRow{
{ChannelID: 1, ChannelName: "A", ModelName: "zeta", Weight: 10, Tier: "primary"},
{ChannelID: 1, ChannelName: "A", ModelName: "alpha", Weight: 10, Tier: "primary"},
{ChannelID: 1, ChannelName: "A", ModelName: "mid", Weight: 10, Tier: "primary"},
}
c := testCatalog(rows, nil, func(_ context.Context, _ int64, _ string) RuntimeReadout {
return RuntimeReadout{SuccEwma: 1, HasHealth: true}
})
for range 20 {
got := c.ChannelRuntimeStates()
require.Equal(t, "alpha", got[1].WorstModel, "同分必须稳定取模型名最小者")
assert.InDelta(t, 100.0, got[1].WorstModelScore, 1e-6)
}
}
+177 -26
View File
@@ -2,7 +2,9 @@ package dispatchadapter
import (
"context"
"fmt"
"math"
"strings"
"time"
"github.com/gogf/gf/v2/frame/g"
@@ -11,17 +13,41 @@ import (
"github.com/qianfree/team-api/internal/dao"
lcommon "github.com/qianfree/team-api/internal/logic/common"
do "github.com/qianfree/team-api/internal/model/do"
"github.com/qianfree/team-api/internal/model/entity"
)
// 维护任务(阶段 3):
// 1. 健康快照落盘:Redis EWMA → chn_health_scores 每 5min 一次,仅供管理后台
// 仪表盘/审计展示,调度决策不读此表(替代旧体系的实时写入)。
// 2. 自动禁用:渠道级熔断本轮故障期(first_opened_ms)持续超过
// policy.breaker.autoDisableAfterSeconds → 落库禁用 + 清绑定 + 目录失效。
// 受现有开关 channel_auto_disable_enabled 控制(默认关)。
// 1. 健康快照落盘:Redis EWMA → chn_health_scores 每 5min 一次,另可由
// RequestHealthSnapshot 按渠道即时触发(渠道测试 / 重置健康度后),仅供管理后台
// 仪表盘/审计展示,调度决策不读此表(替代旧体系的实时写入)。
// 两种情况保留库中旧值而不落盘:Redis 读失败(读值不可信)、渠道全部模型无真实
// 上报(默认满分不是结论)——都不能拿乐观默认值覆盖历史分数。
// 2. 自动禁用:渠道级熔断本轮故障期(first_opened_ms)持续超过
// policy.breaker.autoDisableAfterSeconds → 落库禁用 + 清绑定 + 目录失效。
// 受现有开关 channel_auto_disable_enabled 控制(默认关)。
// Redis 读值降级时同样跳过——自动禁用不可逆,不能基于不可信读值触发。
const maintenanceInterval = 5 * time.Minute
// healthRefreshDelay 按需重算前的等待:健康上报是 fire-and-forget(异步 worker 消费),
// 渠道测试刚投递的探测结果可能尚未写入 Redis,立刻重算会读到上报前的旧值。
const healthRefreshDelay = 500 * time.Millisecond
// healthRefresh 单渠道健康快照的按需重算信号。缓冲队列满则丢弃——下一轮定时维护兜底。
var healthRefresh = make(chan int64, 256)
// RequestHealthSnapshot 请求立即重算某渠道的健康快照,供渠道测试 / 重置健康度后调用。
//
// 维护循环是纯 ticker(5 分钟一轮),此前管理员测试渠道成功后 Redis 立刻变、模型级能力
// 列表(直读 Redis)立刻变,唯独渠道健康分要等下一轮才动,看起来像"测试没生效"。
// 不阻塞调用方;调度引擎尚未组装时信号留在队列里,组装后被消费。
func RequestHealthSnapshot(channelID int64) {
select {
case healthRefresh <- channelID:
default:
}
}
// startMaintenance 启动维护循环(wire 组装时调用一次)。
func startMaintenance(ctx context.Context, stop <-chan struct{}) {
go func() {
@@ -31,6 +57,9 @@ func startMaintenance(ctx context.Context, stop <-chan struct{}) {
select {
case <-ticker.C:
runMaintenance(ctx)
case channelID := <-healthRefresh:
time.Sleep(healthRefreshDelay)
refreshChannelHealth(ctx, channelID)
case <-stop:
return
}
@@ -38,6 +67,32 @@ func startMaintenance(ctx context.Context, stop <-chan struct{}) {
}()
}
// refreshChannelHealth 按需重算单个渠道的健康快照。
// 与 runMaintenance 共用聚合与落盘逻辑,跳过条件也一致(降级 / 全无数据保留旧值)。
func refreshChannelHealth(ctx context.Context, channelID int64) {
defer func() {
if r := recover(); r != nil {
g.Log().Warningf(ctx, "[Dispatch] 健康快照按需重算 panic: %v", r)
}
}()
if catalog == nil || redisState == nil {
return
}
models := catalog.ChannelModels()[channelID]
if len(models) == 0 {
// 渠道不在调度目录(禁用 / 无启用能力行):无健康数据可算,
// TestChannel 侧已有 CatalogHasModel 告警提示该情形
return
}
agg := aggregateChannelHealth(models, func(model string) RuntimeReadout {
return redisState.ReadRuntime(ctx, channelID, model)
})
if agg.Degraded || agg.Counted == 0 {
return
}
snapshotHealthScore(ctx, channelID, agg.AvgSucc, agg.AvgLat, HealthAlpha(), agg.Detail)
}
// runMaintenance 执行一轮健康快照 + 自动禁用检查。
func runMaintenance(ctx context.Context) {
defer func() {
@@ -53,41 +108,125 @@ func runMaintenance(ctx context.Context) {
autoDisable := lcommon.Config().GetBool(ctx, "channel_auto_disable_enabled")
now := time.Now().UnixMilli()
for channelID, models := range catalog.ChannelModels() {
channelModels := catalog.ChannelModels()
covered := make([]int64, 0, len(channelModels))
skippedDegraded := make([]int64, 0)
skippedNoData := make([]int64, 0)
for channelID, models := range channelModels {
if len(models) == 0 {
continue
}
// 按渠道聚合各模型的健康读值
var succSum, latSum float64
var channelRt RuntimeReadout
for i, model := range models {
rt := redisState.ReadRuntime(ctx, channelID, model)
succSum += rt.SuccEwma
latSum += rt.LatEwmaMs
if i == 0 {
channelRt = rt // 渠道级熔断状态各模型读值相同,取第一个
}
}
avgSucc := succSum / float64(len(models))
avgLat := latSum / float64(len(models))
agg := aggregateChannelHealth(models, func(model string) RuntimeReadout {
return redisState.ReadRuntime(ctx, channelID, model)
})
snapshotHealthScore(ctx, channelID, avgSucc, avgLat, pol.Health.Alpha)
// Redis 读失败:本轮读值不可信,跳过落盘与自动禁用判定。
// 若照常落盘,乐观默认值会把全渠道健康分刷成 100,事故期间的历史趋势反而全绿。
if agg.Degraded {
skippedDegraded = append(skippedDegraded, channelID)
continue
}
covered = append(covered, channelID)
// 全部模型都没有真实上报(新渠道 / 长期无流量 / EWMA key 已过 TTL):
// 保留库中上一次的分数,不要用默认满分覆盖掉历史结论。
if agg.Counted == 0 {
skippedNoData = append(skippedNoData, channelID)
} else {
snapshotHealthScore(ctx, channelID, agg.AvgSucc, agg.AvgLat, pol.Health.Alpha, agg.Detail)
}
// 自动禁用:本轮故障期持续超过阈值(first_opened_ms 只在故障期存在,
// 探测失败不重置——探测成功即被清除,见 luaBreakerFail/luaBreakerSuccess)
if autoDisable && channelRt.FirstOpenedMs > 0 &&
now-channelRt.FirstOpenedMs >= int64(pol.Breaker.AutoDisableAfterSeconds)*1000 {
autoDisableChannel(ctx, channelID, now-channelRt.FirstOpenedMs)
if autoDisable && agg.ChannelRt.FirstOpenedMs > 0 &&
now-agg.ChannelRt.FirstOpenedMs >= int64(pol.Breaker.AutoDisableAfterSeconds)*1000 {
autoDisableChannel(ctx, channelID, now-agg.ChannelRt.FirstOpenedMs)
}
}
// 覆盖摘要:目录只装载 active × 有启用能力行的渠道,未覆盖渠道的健康分不会落盘
// (探测却正常的渠道长期停在历史分数时,先看这里是否被跳过)
g.Log().Infof(ctx, "[ChannelHealth] 健康快照维护轮次完成: 覆盖渠道 %d 个 %v", len(covered), covered)
if len(skippedDegraded) > 0 {
g.Log().Warningf(ctx, "[ChannelHealth] 本轮 %d 个渠道因 Redis 健康读取失败跳过落盘(保留库中旧值): %v",
len(skippedDegraded), skippedDegraded)
}
if len(skippedNoData) > 0 {
g.Log().Infof(ctx, "[ChannelHealth] 本轮 %d 个渠道全部模型无真实上报,保留库中旧值: %v",
len(skippedNoData), skippedNoData)
}
}
// channelHealthAggregate 单个渠道的健康聚合结果。
type channelHealthAggregate struct {
AvgSucc float64 // 有真实上报模型的成功率 EWMA 均值
AvgLat float64 // 同上,延迟均值
Counted int // 参与均值的模型数;0 = 该渠道全部模型无真实上报
Degraded bool // 任一模型读失败 → 本轮读值整体不可信
Detail string // 各模型明细,供排查"是哪个模型拉低了均值"
ChannelRt RuntimeReadout // 首个模型的读值(渠道级熔断状态各模型相同)
}
// aggregateChannelHealth 聚合渠道下各模型的健康读值。
//
// 只统计确有真实上报的模型(HasHealth):无流量模型的读值是乐观默认满分,计入均值
// 会把真正故障的模型稀释掉——20 个模型的渠道死掉 1 个(succ→0.05)仍能算出 95 分绿灯,
// 而这恰恰是模型级熔断要隔离的典型场景。
//
// 任一模型读失败即整体标记 Degraded:Redis 不可用时读到的满分既非实测也非"确实无数据",
// 拿它去覆盖库中的历史分数会销毁事故现场。
func aggregateChannelHealth(models []string, read func(model string) RuntimeReadout) channelHealthAggregate {
var agg channelHealthAggregate
var succSum, latSum float64
parts := make([]string, 0, len(models))
for i, model := range models {
rt := read(model)
if i == 0 {
agg.ChannelRt = rt
}
if rt.Degraded {
// 中断前已累加的部分一并作废:降级的聚合结果只保留 Degraded 与
// ChannelRt,不留下会被误读为"有效均值"的半成品计数。
return channelHealthAggregate{Degraded: true, ChannelRt: agg.ChannelRt}
}
if !rt.HasHealth {
parts = append(parts, model+"=无数据")
continue
}
succSum += rt.SuccEwma
latSum += rt.LatEwmaMs
agg.Counted++
parts = append(parts, fmt.Sprintf("%s=%.3f", model, rt.SuccEwma))
}
agg.Detail = strings.Join(parts, " ")
if agg.Counted > 0 {
agg.AvgSucc = succSum / float64(agg.Counted)
agg.AvgLat = latSum / float64(agg.Counted)
}
return agg
}
// snapshotHealthScore 健康快照落盘(upsert)。
// health_score 用 succ^alpha 的百分制近似,与调度用的 healthFactor 同源。
func snapshotHealthScore(ctx context.Context, channelID int64, succ, latMs, alpha float64) {
// 每轮固定输出一行明细(计算值 vs 库中值 + 各模型 succ):健康分不动时可直接看出
// 是「算出来就是这个值」(某模型 succ 被拖低)还是「写入失败」(库中值持续落后于计算值)。
func snapshotHealthScore(ctx context.Context, channelID int64, succ, latMs, alpha float64, modelDetail string) {
healthScore := math.Pow(clampFloat(succ, 0, 1), alpha) * 100
// 先读旧值:updated_at 每轮都会变(affected 恒为 1),比对必须基于 health_score
var prev *entity.ChnHealthScores
if err := dao.ChnHealthScores.Ctx(ctx).Where("channel_id", channelID).Scan(&prev); err != nil {
g.Log().Debugf(ctx, "[Dispatch] 健康快照旧值读取失败: channel=%d err=%v", channelID, err)
}
prevScore := -1.0 // 无记录
if prev != nil {
prevScore = prev.HealthScore
}
affected, err := dao.ChnHealthScores.Ctx(ctx).
Where("channel_id", channelID).
Data(do.ChnHealthScores{
@@ -98,7 +237,8 @@ func snapshotHealthScore(ctx context.Context, channelID int64, succ, latMs, alph
}).
UpdateAndGetAffected()
if err != nil {
g.Log().Debugf(ctx, "[Dispatch] 健康快照更新失败: channel=%d err=%v", channelID, err)
// 写入失败必须可见(此前 Debug 级静默,健康分冻结难以定位)
g.Log().Warningf(ctx, "[ChannelHealth] 渠道 %d 健康快照更新失败: %v", channelID, err)
return
}
if affected == 0 {
@@ -110,9 +250,20 @@ func snapshotHealthScore(ctx context.Context, channelID int64, succ, latMs, alph
CalculatedAt: gtime.Now(),
}).Insert()
if err != nil {
g.Log().Debugf(ctx, "[Dispatch] 健康快照插入失败: channel=%d err=%v", channelID, err)
g.Log().Warningf(ctx, "[ChannelHealth] 渠道 %d 健康快照插入失败: %v", channelID, err)
return
}
}
// 分数有实际变化才记 Info;稳态渠道降为 Debug(避免每 5 分钟 × 渠道数的固定噪音)。
// 两种情况都带各模型 succ 明细:排查"健康分不符合预期"时可直接定位是哪个模型拉低了均值。
if math.Abs(healthScore-prevScore) >= 0.05 {
g.Log().Infof(ctx, "[ChannelHealth] 渠道 %d 健康分落盘: %.1f → %.1f | avg_succ %.4f | 模型: %s",
channelID, prevScore, healthScore, succ, modelDetail)
} else {
g.Log().Debugf(ctx, "[ChannelHealth] 渠道 %d 健康快照无变化: %.1f | avg_succ %.4f | 模型: %s",
channelID, healthScore, succ, modelDetail)
}
}
// autoDisableChannel 熔断长期不恢复 → 落库禁用(标记 auto_disabled 供自动恢复探测识别)
@@ -0,0 +1,127 @@
package dispatchadapter
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// readoutOf 构造一份"有真实上报"的读值。
func readoutOf(succ, lat float64) RuntimeReadout {
return RuntimeReadout{SuccEwma: succ, LatEwmaMs: lat, HasHealth: true}
}
// staticReader 按模型名返回预置读值;未预置的模型返回"无数据"默认满分。
func staticReader(m map[string]RuntimeReadout) func(string) RuntimeReadout {
return func(model string) RuntimeReadout {
if rt, ok := m[model]; ok {
return rt
}
return RuntimeReadout{SuccEwma: 1} // HasHealth=false,模拟从未被真实流量访问过
}
}
func TestAggregateChannelHealth_只统计有真实上报的模型(t *testing.T) {
// 20 个模型的渠道死掉 1 个:旧实现把 19 个无数据模型按满分计入均值,
// avgSucc≈0.95 → 91 分绿灯,完全掩盖故障。
models := make([]string, 0, 20)
for i := range 20 {
models = append(models, string(rune('a'+i)))
}
agg := aggregateChannelHealth(models, staticReader(map[string]RuntimeReadout{
"a": readoutOf(0.05, 8000),
}))
assert.False(t, agg.Degraded)
assert.Equal(t, 1, agg.Counted, "只有 1 个模型有真实上报")
assert.InDelta(t, 0.05, agg.AvgSucc, 1e-9, "无数据模型不得稀释故障模型的分数")
assert.InDelta(t, 8000, agg.AvgLat, 1e-9)
assert.Contains(t, agg.Detail, "a=0.050")
assert.Contains(t, agg.Detail, "b=无数据")
}
func TestAggregateChannelHealth_多模型均值(t *testing.T) {
agg := aggregateChannelHealth([]string{"m1", "m2", "m3"}, staticReader(map[string]RuntimeReadout{
"m1": readoutOf(1.0, 100),
"m2": readoutOf(0.6, 300),
"m3": readoutOf(0.8, 200),
}))
assert.Equal(t, 3, agg.Counted)
assert.InDelta(t, 0.8, agg.AvgSucc, 1e-9)
assert.InDelta(t, 200, agg.AvgLat, 1e-9)
}
func TestAggregateChannelHealth_全部无数据不产生均值(t *testing.T) {
agg := aggregateChannelHealth([]string{"m1", "m2"}, staticReader(nil))
assert.False(t, agg.Degraded)
assert.Zero(t, agg.Counted, "Counted=0 时调用方保留库中旧值,不落盘")
assert.Zero(t, agg.AvgSucc, "不得回落成默认满分 1.0 造成 100 分落盘")
}
func TestAggregateChannelHealth_读失败整体降级(t *testing.T) {
// Redis 故障时读值既非实测也非"确实无数据",必须整体作废而非按满分聚合
agg := aggregateChannelHealth([]string{"m1", "m2", "m3"}, staticReader(map[string]RuntimeReadout{
"m1": readoutOf(0.9, 100),
"m2": {SuccEwma: 1, Degraded: true},
"m3": readoutOf(0.9, 100),
}))
assert.True(t, agg.Degraded)
assert.Zero(t, agg.Counted, "降级时不得给出任何聚合结论")
}
func TestAggregateChannelHealth_首模型读值供自动禁用判定(t *testing.T) {
first := readoutOf(0.5, 100)
first.FirstOpenedMs = 12345
agg := aggregateChannelHealth([]string{"m1", "m2"}, staticReader(map[string]RuntimeReadout{
"m1": first,
"m2": readoutOf(0.5, 100),
}))
assert.Equal(t, int64(12345), agg.ChannelRt.FirstOpenedMs, "渠道级熔断读值取首个模型")
}
// TestRequestHealthSnapshot_不阻塞且满时丢弃 调用方是管理后台请求线程(渠道测试 /
// 重置健康度),队列满时必须丢弃而非阻塞——下一轮定时维护会兜底重算。
func TestRequestHealthSnapshot_不阻塞且满时丢弃(t *testing.T) {
drain := func() {
for {
select {
case <-healthRefresh:
default:
return
}
}
}
drain()
defer drain()
for i := range cap(healthRefresh) {
RequestHealthSnapshot(int64(i))
}
require.Len(t, healthRefresh, cap(healthRefresh))
done := make(chan struct{})
go func() {
RequestHealthSnapshot(9999)
close(done)
}()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("队列满时 RequestHealthSnapshot 必须立即丢弃,不得阻塞调用方")
}
assert.Len(t, healthRefresh, cap(healthRefresh), "满队列不应再增长")
}
// TestRefreshChannelHealth_未组装时静默返回 管理后台可能在首次 relay 请求前就触发重算,
// 此时调度引擎尚未组装(catalog / redisState 为 nil),不得 panic。
func TestRefreshChannelHealth_未组装时静默返回(t *testing.T) {
require.Nil(t, catalog, "前置条件:本包测试不组装调度引擎")
assert.NotPanics(t, func() { refreshChannelHealth(context.Background(), 1) })
}
+60 -21
View File
@@ -7,7 +7,9 @@ package dispatchadapter
import (
"context"
"math"
"strconv"
"strings"
"time"
"github.com/gogf/gf/v2/frame/g"
@@ -170,9 +172,12 @@ func (s *RedisState) ReportOutcome(o dispatch.Outcome) {
// luaHealthObserve 健康 EWMA 原子更新(基线方案 §10.2 + 修订 R6 分档衰减)。
// ARGV: success(0/1), latencyMs, decay, nowMs, keyTtlMs
// 返回 {旧succ, 新succ, 旧lat, 新lat}(tostring 字符串,避免浮点被整数回复截断),
// 供适配层打印健康分变化日志。
const luaHealthObserve = `
local succ = tonumber(redis.call('HGET', KEYS[1], 'succ_ewma') or '1.0')
local lat = tonumber(redis.call('HGET', KEYS[1], 'lat_ewma') or '0')
local oldSucc, oldLat = succ, lat
if ARGV[1] == '1' then
succ = succ * 0.9 + 0.1
local l = tonumber(ARGV[2])
@@ -184,7 +189,7 @@ else
end
redis.call('HSET', KEYS[1], 'succ_ewma', succ, 'lat_ewma', lat, 'updated_ms', ARGV[4])
redis.call('PEXPIRE', KEYS[1], ARGV[5])
return 1`
return {tostring(oldSucc), tostring(succ), tostring(oldLat), tostring(lat)}`
// luaBreakerOutcome 渠道级+模型级熔断转移(单脚本双 key 原子执行,与 dispatch/breaker.go
// 纯函数规则保持一致,改动需同步)。
@@ -313,8 +318,6 @@ return 1`
// processOutcome 消费一条结果事件:健康 EWMA + 渠道级/模型级熔断转移。
// 限流类不喂熔断窗口(429 属容量信号而非故障信号),改喂 softLimit 估计器。
// 探测失败(Probe)不喂健康 EWMA 只喂熔断窗口:周期性探测持续失败会把无流量
// 渠道的健康分指数拖垮(0.93^N)且无真实流量对冲;探测成功照常回升健康。
func (s *RedisState) processOutcome(ctx context.Context, o dispatch.Outcome) {
pol := s.policy()
now := time.Now().UnixMilli()
@@ -325,12 +328,12 @@ func (s *RedisState) processOutcome(ctx context.Context, o dispatch.Outcome) {
if o.Success {
success = "1"
}
if !o.Probe || o.Success {
if _, err := g.Redis().Do(ctx, "EVAL", luaHealthObserve, 1, healthKey,
success, o.LatencyMs, healthDecayFor(o.Class), now, stateKeyTTLMs); err != nil {
s.local.observe(o) // 降级:实例本地健康镜像
return
}
// 健康观察成功即静默:本函数消费每个请求的结果事件,任何逐请求日志都会在生产流量下
// 淹没日志;健康分轨迹排查走维护轮落盘日志 + ReadRuntime 直接读值。
if _, err := g.Redis().Do(ctx, "EVAL", luaHealthObserve, 1, healthKey,
success, o.LatencyMs, healthDecayFor(o.Class), now, stateKeyTTLMs); err != nil {
s.local.observe(o) // 降级:实例本地健康镜像
return
}
// 429 → softLimit 自动估计器(基线方案 §8.2)
@@ -423,6 +426,8 @@ func (s *RedisState) ResetHealth(ctx context.Context, channelID int64, models []
g.Log().Debugf(ctx, "[Dispatch] 健康重置失败: channel=%d model=%s err=%v", channelID, model, err)
}
}
g.Log().Infof(ctx, "[ChannelHealth] 渠道 %d 健康度已重置: %d 个模型 succ_ewma → %.2f(健康分≈%.1f)、熔断复位",
channelID, len(models), resetHealthSuccEwma, math.Pow(resetHealthSuccEwma, s.policy().Health.Alpha)*100)
}
// resetBreakerState 熔断复位为 CLOSED 并记录 recovered_ms(开启爬坡窗口)。
@@ -566,6 +571,21 @@ func (s *RedisState) CoolCredential(ctx context.Context, keyID int64, ttl time.D
}
}
// CredentialCooldownRemaining 返回渠道 Key 的凭证冷却剩余秒数(0 = 未在冷却)。
// 凭证冷却是调度器内部的瞬态状态(Redis credcd 标记 + TTL),无任何管理页面出口;
// 本函数供管理后台 Key 列表/渠道详情展示。读 TTL 而非 EXISTS,让前端能显示剩余时间。
// Redis 故障时返回 0:仅影响展示精度,不影响调度行为(调度走 IsCredentialCooled 的本地镜像降级)。
func CredentialCooldownRemaining(ctx context.Context, keyID int64) int {
v, err := g.Redis().Do(ctx, "TTL", keyCredCD+strconv.FormatInt(keyID, 10))
if err != nil || v == nil {
return 0
}
if s := v.Int(); s > 0 {
return s
}
return 0
}
// ---------------------------------------------------------------------------
// 运行时读值(供目录快照合并健康/熔断/负载)
// ---------------------------------------------------------------------------
@@ -580,27 +600,45 @@ type RuntimeReadout struct {
RecoveredMs int64 // 渠道级熔断最近恢复时间(0 = 无记录)
FirstOpenedMs int64 // 渠道级本轮故障期起点(0 = 非故障期),供自动禁用判定
Onset429Ewma float64 // 429 起始水位 EWMA(0 = 无 429 历史),供 softLimit 自动估计
// HasHealth 健康 hash 中读到了可用的 succ_ewma(该渠道×模型有过真实上报)。
// false 表示 SuccEwma/LatEwmaMs 是乐观默认值而非实测值——聚合展示分时必须排除,
// 否则大量无流量模型的"满分"会把真正故障模型的低分稀释掉。
HasHealth bool
// Degraded Redis 健康读取失败,本次读值完全不可信(既非实测也非"确实无数据")。
// 调用方必须区分对待:调度侧沿用上一份快照,展示侧跳过本轮落盘。
Degraded bool
}
// ReadRuntime 读取渠道×模型的运行时状态(目录刷新循环调用,非请求热路径)。
// Redis 故障 → 返回乐观默认值(健康满分 + CLOSED),由 last-known 快照语义兜底。
// Redis 故障 → 返回乐观默认值(健康满分 + CLOSED)并置 Degraded,由调用方决定兜底策略。
func (s *RedisState) ReadRuntime(ctx context.Context, channelID int64, model string) RuntimeReadout {
out := RuntimeReadout{SuccEwma: 1}
chStr := strconv.FormatInt(channelID, 10)
now := time.Now().UnixMilli()
if v, err := g.Redis().Do(ctx, "HMGET", keyHealth+chStr+":"+model, "succ_ewma", "lat_ewma"); err == nil {
if v, err := g.Redis().Do(ctx, "HMGET", keyHealth+chStr+":"+model, "succ_ewma", "lat_ewma"); err != nil {
// 读失败与"字段不存在"必须区分:前者读值不可信(Degraded),后者是确实没有
// 真实上报(HasHealth=false)。二者都不能当成"这个渠道很健康"落盘或重建快照。
out.Degraded = true
} else {
vals := v.Vars()
if len(vals) == 2 {
// gf redis 驱动对 HMGET 缺失字段返回空字符串而非 nil(IsNil()==false),
// 必须按无数据处理保留乐观默认值:否则无流量/键过期(TTL 24h)的模型
// 读成 succ=0,健康快照取平均被拖到个位数(调度打分有 succ<=0 兜底,
// 快照聚合没有)。
if str := vals[0].String(); str != "" {
out.SuccEwma = vals[0].Float64()
// 必须按字符串显式判空:HMGET 对不存在的 key/字段返回 Redis nil,但 gredis 把它
// 包装成非 nil 空 gvar(IsNil() 恒为 false、Float64() 静默返回 0)。曾用
// `if !vals[0].IsNil()` 判空,导致所有从未被真实流量访问过的模型被读成 0 分而非
// 默认满分,拖垮渠道聚合健康分与调度 healthFactor。详见 goframe-conventions.md。
// 值域校验:EWMA 由正数衰减而来,数学上不会精确到达 0,精确 0 视为异常数据。
if s := strings.TrimSpace(vals[0].String()); s != "" {
if f, perr := strconv.ParseFloat(s, 64); perr == nil && f > 0 && f <= 1 {
out.SuccEwma = f
out.HasHealth = true
}
}
if str := vals[1].String(); str != "" {
out.LatEwmaMs = vals[1].Float64()
if s := strings.TrimSpace(vals[1].String()); s != "" {
if f, perr := strconv.ParseFloat(s, 64); perr == nil && f >= 0 {
out.LatEwmaMs = f
}
}
}
}
@@ -622,8 +660,9 @@ func (s *RedisState) readBreaker(ctx context.Context, key string, nowMs int64) (
return dispatch.BreakerClosed, 0, 0
}
vals := v.Vars()
// 缺失字段经 gf 驱动转换后为空字符串(IsNil()==false),与 nil 同判为无快照
if len(vals) != 5 || vals[0].IsNil() || vals[0].String() == "" {
// 判空必须用 IsEmpty():HMGET 缺失字段的 gvar IsNil() 恒为 false(见 ReadRuntime 注释)。
// 此处即使漏判也侥幸无害(默认 BreakerClosed 恰为零值 0),但保持与健康读取一致的正确写法。
if len(vals) != 5 || vals[0].IsEmpty() {
return dispatch.BreakerClosed, 0, 0
}
snap := dispatch.BreakerSnapshot{
@@ -486,6 +486,98 @@ func TestReportOutcome异步消费(t *testing.T) {
}, 3*time.Second, 20*time.Millisecond, "后台 worker 必须消费上报事件")
}
// TestReadRuntime无数据回落默认值 从未被访问过的渠道×模型(Redis key 不存在)必须回落
// 默认满分 1.0。HMGET 缺失字段经 gredis 包装成非 nil 空 gvar(IsNil() 恒为 false、
// Float64() 静默返回 0),旧代码用 IsNil() 判空失效,把全部冷模型读成 0 分,
// 拖垮渠道聚合健康分与调度 healthFactor。
func TestReadRuntime无数据回落默认值(t *testing.T) {
ctx := context.Background()
s := newTestState(t, nil)
rt := s.ReadRuntime(ctx, 99, "never-touched")
assert.Equal(t, 1.0, rt.SuccEwma, "key 不存在必须回落默认满分,不能是 0")
assert.Equal(t, 0.0, rt.LatEwmaMs, "无延迟数据为 0")
assert.Equal(t, dispatch.BreakerClosed, rt.Breaker)
// 只有 lat_ewma、无 succ_ewma 字段时,succ 同样回落默认值
_, err := g.Redis().Do(ctx, "HSET", keyHealth+"99:partial", "lat_ewma", "100")
require.NoError(t, err)
rt = s.ReadRuntime(ctx, 99, "partial")
assert.Equal(t, 1.0, rt.SuccEwma, "字段缺失必须回落默认满分")
assert.Equal(t, 100.0, rt.LatEwmaMs)
}
// TestReadRuntime异常值防御 succ_ewma 为空串/非法值/精确 0 时视为无数据(保持默认 1.0)。
func TestReadRuntime异常值防御(t *testing.T) {
ctx := context.Background()
s := newTestState(t, nil)
for _, bad := range []string{"", " ", "abc", "0"} {
_, err := g.Redis().Do(ctx, "HSET", keyHealth+"30:dirty", "succ_ewma", bad, "lat_ewma", "100")
require.NoError(t, err)
rt := s.ReadRuntime(ctx, 30, "dirty")
assert.Equal(t, 1.0, rt.SuccEwma, "succ_ewma=%q 应视为无数据", bad)
}
// 真实低分(正数)不受影响
_, err := g.Redis().Do(ctx, "HSET", keyHealth+"30:low", "succ_ewma", "0.0008", "lat_ewma", "100")
require.NoError(t, err)
rt := s.ReadRuntime(ctx, 30, "low")
assert.Equal(t, 0.0008, rt.SuccEwma, "合法低分必须如实读取")
// lat_ewma 同样防御:脏值回落 0
_, err = g.Redis().Do(ctx, "HSET", keyHealth+"30:dirtylat", "succ_ewma", "0.5", "lat_ewma", "xyz")
require.NoError(t, err)
rt = s.ReadRuntime(ctx, 30, "dirtylat")
assert.Equal(t, 0.5, rt.SuccEwma)
assert.Equal(t, 0.0, rt.LatEwmaMs)
}
// TestReadRuntime区分无数据与读失败 「没读到」与「读失败」必须可区分:前者是真实的
// 「该模型没有流量」,后者是 Redis 不可用。二者都会回落成默认满分,若调用方无法区分,
// Redis 抖动期间维护快照会把全渠道健康分刷成 100,事故现场被销毁。
func TestReadRuntime区分无数据与读失败(t *testing.T) {
ctx := context.Background()
s := newTestState(t, nil)
// 从未上报:读值是乐观默认满分,但必须标记为「无真实数据」
rt := s.ReadRuntime(ctx, 77, "cold")
assert.Equal(t, 1.0, rt.SuccEwma)
assert.False(t, rt.HasHealth, "无上报的默认满分不得冒充实测值")
assert.False(t, rt.Degraded)
// 有真实上报
_, err := g.Redis().Do(ctx, "HSET", keyHealth+"77:hot", "succ_ewma", "0.42", "lat_ewma", "900")
require.NoError(t, err)
rt = s.ReadRuntime(ctx, 77, "hot")
assert.Equal(t, 0.42, rt.SuccEwma)
assert.True(t, rt.HasHealth)
assert.False(t, rt.Degraded)
// Redis 故障:同样回落满分,但标记 Degraded
mr.SetError("LOADING Redis is loading the dataset in memory")
defer mr.SetError("")
rt = s.ReadRuntime(ctx, 77, "hot")
assert.True(t, rt.Degraded, "读失败必须显式标记")
assert.False(t, rt.HasHealth, "读失败不是「有数据」")
assert.Equal(t, 1.0, rt.SuccEwma)
}
// TestReadRuntime脏值不算有效数据 精确 0 / 非法值视为无数据,HasHealth 保持 false,
// 避免脏数据以「实测值」身份参与渠道聚合(2026-08-27 全渠道健康分坍缩事故的读侧防线)。
func TestReadRuntime脏值不算有效数据(t *testing.T) {
ctx := context.Background()
s := newTestState(t, nil)
for _, bad := range []string{"", "abc", "0"} {
_, err := g.Redis().Do(ctx, "HSET", keyHealth+"78:dirty", "succ_ewma", bad, "lat_ewma", "100")
require.NoError(t, err)
rt := s.ReadRuntime(ctx, 78, "dirty")
assert.False(t, rt.HasHealth, "succ_ewma=%q 不得算作有效实测数据", bad)
assert.Equal(t, 1.0, rt.SuccEwma)
}
}
// ---------------------------------------------------------------------------
// 本地降级(修订 R4 + 基线方案 §13)
// ---------------------------------------------------------------------------
+27
View File
@@ -3,6 +3,7 @@ package dispatchadapter
import (
"context"
"math/rand"
"slices"
"sync"
"github.com/gogf/gf/v2/frame/g"
@@ -62,6 +63,21 @@ func Coordinator(ctx context.Context) *dispatch.Coordinator {
// CatalogInstance 返回目录单例(handler 取 ForwardMeta 用,需先调用过 Coordinator)。
func CatalogInstance() *Catalog { return catalog }
// HealthAlpha 返回当前路由策略的健康指数 α(健康分 = succ_ewma^α × 100)。
//
// 管理后台展示健康分必须与调度 healthFactor、维护快照落盘同源:三处都用 succ^α,
// 唯独模型级能力列表曾用线性 succ×100,而前端两处共用同一套 80/50 阈值,导致
// succ=0.89 时模型行显示 89「健康」、渠道行显示 79「降级」,看起来像 bug。
//
// 调度引擎未组装(首次 relay 请求前)时返回默认策略的 α,不触发组装
// ——管理后台读接口不应有装配整个调度引擎的副作用(与 CatalogInstance 同策略)。
func HealthAlpha() float64 {
if coordinator == nil {
return dispatch.DefaultRoutingPolicy().Health.Alpha
}
return coordinator.Policy().Health.Alpha
}
// RefreshDispatchLease 长请求(流式/websocket)续期调度租约。
// 供 handler 的租约续期器直接调用(RouteSession 非并发安全,不经会话)。
func RefreshDispatchLease(ctx context.Context, channelID int64, requestID string) {
@@ -86,6 +102,17 @@ func ReportProbeOutcome(ctx context.Context, channelID int64, model string, succ
redisState.ReportOutcome(o)
}
// CatalogHasModel 判断渠道×模型当前是否在调度目录中(探测可观测性检查用)。
// 不在目录的探测目标(渠道无启用的能力行、或模型未配置能力)健康 EWMA 写入后无人消费:
// 调度不会选它、维护快照也不会落盘该渠道的健康分——管理后台将看不到探测效果。
// 需已调用过 Coordinator 完成组装;未组装时返回 false。
func CatalogHasModel(channelID int64, model string) bool {
if catalog == nil {
return false
}
return slices.Contains(catalog.ChannelModels()[channelID], model)
}
// MarkChannelRecovered 渠道被手动启用/恢复时复位熔断并开启爬坡窗口(管理后台调用)。
func MarkChannelRecovered(ctx context.Context, channelID int64) {
if redisState != nil {
+31 -17
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
"math"
"time"
do "github.com/qianfree/team-api/internal/model/do"
@@ -148,6 +149,8 @@ func (s *sAdmin) ListChannels(ctx context.Context, req *v1.ChannelListReq) (*v1.
if rs, ok := runtimeStates[ch.ID]; ok {
item.BreakerState = int(rs.Breaker)
item.BreakerModels = rs.BreakerModels
item.WorstModel = rs.WorstModel
item.WorstModelScore = rs.WorstModelScore
}
list = append(list, item)
}
@@ -593,6 +596,8 @@ func (s *sAdmin) GetChannelDetail(ctx context.Context, req *v1.ChannelDetailReq)
if rs, ok := channelRuntimeStates()[req.ID]; ok {
res.BreakerState = int(rs.Breaker)
res.BreakerModels = rs.BreakerModels
res.WorstModel = rs.WorstModel
res.WorstModelScore = rs.WorstModelScore
}
if res.KeyType == "" {
res.KeyType = "apikey"
@@ -791,6 +796,9 @@ func (s *sAdmin) batchGetModelHealth(ctx context.Context, channelID int64, abili
}
redis := g.Redis()
// 健康分折算口径必须与调度 healthFactor / 维护快照落盘一致(succ^α×100),
// α 取当前生效的路由策略而非硬编码——策略支持热更新。
alpha := dispatchadapter.HealthAlpha()
for _, ab := range abilities {
healthKey := fmt.Sprintf("dispatch:v1:health:%d:%s", channelID, ab.ModelName)
breakerKey := fmt.Sprintf("dispatch:v1:breaker:%d:%s", channelID, ab.ModelName)
@@ -808,8 +816,8 @@ func (s *sAdmin) batchGetModelHealth(ctx context.Context, channelID int64, abili
var succEWMA float64
if _, err := fmt.Sscanf(succStr, "%f", &succEWMA); err == nil && succEWMA >= 0 {
data.SuccEWMA = &succEWMA
// 健康分 = succ_ewma * 100
healthScore := succEWMA * 100
// 健康分 = succ_ewma^α × 100(与调度 healthFactor 同源)
healthScore := math.Pow(succEWMA, alpha) * 100
data.HealthScore = &healthScore
}
}
@@ -912,10 +920,12 @@ func defaultProviderURL(t int) string {
}
// GetChannelHealthTrend 获取渠道健康趋势数据(健康度、延迟在 SQL 层四舍五入取整,避免展示小数)
// 不再返回 stability_score / consecutive_failures:健康体系重写后这两列已停止写入,
// 返回恒定值只会让前端画出两条毫无意义的常量线。
func (s *sAdmin) GetChannelHealthTrend(ctx context.Context, req *v1.ChannelHealthTrendReq) (*v1.ChannelHealthTrendRes, error) {
var points []v1.HealthTrendPoint
err := dao.ChnHealthSnapshots.Ctx(ctx).
Fields("snapshot_at, ROUND(health_score)::int AS health_score, success_rate, ROUND(latency_ms)::int AS latency_ms, stability_score, consecutive_failures").
Fields("snapshot_at, ROUND(health_score)::int AS health_score, success_rate, ROUND(latency_ms)::int AS latency_ms").
Where("channel_id", req.ID).
Where("snapshot_at >= ?", gtime.Now().Add(-time.Duration(req.Hours)*time.Hour)).
OrderAsc("snapshot_at").
@@ -959,22 +969,25 @@ func (s *sAdmin) ResetChannelHealth(ctx context.Context, req *v1.ChannelResetHea
// 调度层:只复位指定模型的熔断 + 成功率恢复
dispatchadapter.ResetModelHealth(ctx, req.ID, req.ModelName)
// 单模型重置不直接写 chn_health_scores(渠道分是全模型聚合值,不能按单模型拍值),
// 改为请求重算——否则展示分永远停在旧值,管理员会以为重置没生效
dispatchadapter.RequestHealthSnapshot(req.ID)
return nil, nil
}
// 未指定模型名,重置整个渠道(含所有模型)
// 展示层:健康分落库为 80(调度不读此表,仅供仪表盘/审计展示;下次维护快照会按策略重算)
// 展示层:健康分落库为 80(调度不读此表,仅供仪表盘/审计展示;随后的按需重算/
// 维护快照会按 Redis 实际读值重算,此处只是让界面立即有反馈)
// stability_score / consecutive_failures 已废弃,不再写入
affected, err := dao.ChnHealthScores.Ctx(ctx).
Where("channel_id", req.ID).
Data(do.ChnHealthScores{
SuccessRate: 90.00,
LatencyMs: 0,
StabilityScore: 100.00,
ConsecutiveFailures: 0,
HealthScore: 80.00,
CalculatedAt: gtime.Now(),
SuccessRate: 90.00,
LatencyMs: 0,
HealthScore: 80.00,
CalculatedAt: gtime.Now(),
}).
UpdateAndGetAffected()
if err != nil {
@@ -983,13 +996,11 @@ func (s *sAdmin) ResetChannelHealth(ctx context.Context, req *v1.ChannelResetHea
// 健康分记录缺失时兜底插入(正常情况下创建渠道时已由 InitHealthScore 初始化)
if affected == 0 {
_, err = dao.ChnHealthScores.Ctx(ctx).Insert(do.ChnHealthScores{
ChannelId: req.ID,
SuccessRate: 90.00,
LatencyMs: 0,
StabilityScore: 100.00,
ConsecutiveFailures: 0,
HealthScore: 80.00,
CalculatedAt: gtime.Now(),
ChannelId: req.ID,
SuccessRate: 90.00,
LatencyMs: 0,
HealthScore: 80.00,
CalculatedAt: gtime.Now(),
})
if err != nil {
return nil, err
@@ -998,6 +1009,9 @@ func (s *sAdmin) ResetChannelHealth(ctx context.Context, req *v1.ChannelResetHea
// 调度层:复位渠道级/模型级熔断 + 成功率恢复,渠道立即恢复被选择能力
dispatchadapter.ResetChannelHealth(ctx, req.ID)
// 请求按 Redis 实际读值重算落盘:上面写死的 80 只是调度引擎未组装时的兜底,
// 重算后会收敛到与调度同源的 0.9^α×100
dispatchadapter.RequestHealthSnapshot(req.ID)
return nil, nil
}
+5 -12
View File
@@ -9,8 +9,6 @@ import (
"net/http"
"strings"
"github.com/gogf/gf/v2/frame/g"
v1 "github.com/qianfree/team-api/api/admin/v1"
relaycommon "github.com/qianfree/team-api/relay/common"
"github.com/qianfree/team-api/relay/constant"
@@ -52,15 +50,12 @@ func doTestHTTPRequest(ctx context.Context, method, reqURL string, headers map[s
}
// executeTest 封装通用的测试请求执行和结果处理逻辑
func executeTest(ctx context.Context, provider, method, reqURL string, headers map[string]string, body []byte, useProxy bool) testResult {
func executeTest(ctx context.Context, method, reqURL string, headers map[string]string, body []byte, useProxy bool) testResult {
maskedHdrs := maskHeaders(headers)
bodyStr := string(body)
g.Log().Infof(ctx, "[ChannelTest] %s | 发送请求 | %s %s | 代理: %v", provider, method, reqURL, useProxy)
resp, err := doTestHTTPRequest(ctx, method, reqURL, headers, body, useProxy)
if err != nil {
g.Log().Warningf(ctx, "[ChannelTest] %s | 请求失败 | %s | 错误: %v", provider, reqURL, err)
return testResult{
Error: fmt.Sprintf("请求失败: %v", err),
Request: buildReqDetail(method, reqURL, maskedHdrs, bodyStr),
@@ -72,7 +67,6 @@ func executeTest(ctx context.Context, provider, method, reqURL string, headers m
respStr := string(respBody)
if resp.StatusCode != 200 {
g.Log().Warningf(ctx, "[ChannelTest] %s | 测试失败 | HTTP %d | 响应: %s", provider, resp.StatusCode, truncateStr(respStr, 200))
return testResult{
Success: false,
Error: fmt.Sprintf("HTTP %d: %s", resp.StatusCode, truncateStr(respStr, 500)),
@@ -81,7 +75,6 @@ func executeTest(ctx context.Context, provider, method, reqURL string, headers m
}
}
g.Log().Infof(ctx, "[ChannelTest] %s | 测试成功 | HTTP %d | 响应: %s", provider, resp.StatusCode, truncateStr(respStr, 200))
return testResult{
Success: true,
Response: truncateStr(respStr, 500),
@@ -101,7 +94,7 @@ func testOpenAI(ctx context.Context, baseURL, apiKey, modelName string, useProxy
"Authorization": "Bearer " + apiKey,
"Content-Type": "application/json",
}
return executeTest(ctx, "OpenAI", "POST", reqURL, headers, bodyJSON, useProxy)
return executeTest(ctx, "POST", reqURL, headers, bodyJSON, useProxy)
}
func testClaude(ctx context.Context, baseURL, apiKey, modelName string, useProxy bool) testResult {
@@ -117,7 +110,7 @@ func testClaude(ctx context.Context, baseURL, apiKey, modelName string, useProxy
"anthropic-version": "2023-06-01",
"Content-Type": "application/json",
}
return executeTest(ctx, "Claude", "POST", reqURL, headers, bodyJSON, useProxy)
return executeTest(ctx, "POST", reqURL, headers, bodyJSON, useProxy)
}
func testGemini(ctx context.Context, baseURL, apiKey, modelName string, useProxy bool) testResult {
@@ -135,7 +128,7 @@ func testGemini(ctx context.Context, baseURL, apiKey, modelName string, useProxy
"x-goog-api-key": apiKey,
"Content-Type": "application/json",
}
return executeTest(ctx, "Gemini", "POST", reqURL, headers, bodyJSON, useProxy)
return executeTest(ctx, "POST", reqURL, headers, bodyJSON, useProxy)
}
func testZhipu(ctx context.Context, baseURL, apiKey, modelName string, useProxy bool) testResult {
@@ -150,7 +143,7 @@ func testZhipu(ctx context.Context, baseURL, apiKey, modelName string, useProxy
"Authorization": "Bearer " + apiKey,
"Content-Type": "application/json",
}
return executeTest(ctx, "Zhipu", "POST", reqURL, headers, bodyJSON, useProxy)
return executeTest(ctx, "POST", reqURL, headers, bodyJSON, useProxy)
}
func buildReqDetail(method, url string, headers map[string]string, body string) *v1.ChannelTestReqDetail {
+13 -6
View File
@@ -119,14 +119,21 @@ func (s *sAdmin) TestChannel(ctx context.Context, req *v1.ChannelTestReq) (*v1.C
// 渠道启用时 MarkChannelRecovered 会复位熔断,无需在禁用期间预热。
if ch.Status == "active" {
dispatchadapter.ReportProbeOutcome(ctx, channelID, testModel, result.Success, float64(latencyMs))
// 请求立即重算渠道健康快照:维护循环 5 分钟一轮,不触发的话管理员会看到
// "测试成功了但渠道健康度纹丝不动"(模型级能力列表直读 Redis 会立刻变,
// 渠道健康分读的是 chn_health_scores 落盘值)
dispatchadapter.RequestHealthSnapshot(channelID)
// 探测目标不在调度目录:EWMA 照常写入 Redis,但调度不读、维护快照不落盘,
// 管理后台健康分不会变化——探测看起来"生效了但没效果"的常见根因,必须告警
if !dispatchadapter.CatalogHasModel(channelID, testModel) {
g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测结果已投递健康上报,但模型 %s 不在调度目录(渠道无启用的模型能力行或该模型未配置能力)| 健康分不会落盘展示,渠道/模型不参与调度,请检查渠道的模型能力配置",
ch.Name, channelID, testModel)
}
}
// 记录测试结果日志
if result.Success {
g.Log().Infof(ctx, "[ChannelTest] 渠道 %s (%d) 测试成功 | 模型: %s (上游: %s) | 延迟: %dms | 代理: %v",
ch.Name, channelID, testModel, upstreamModel, latencyMs, useProxy)
} else {
g.Log().Warningf(ctx, "[ChannelTest] 渠道 %s (%d) 测试失败 | 模型: %s (上游: %s) | 延迟: %dms | 代理: %v | 错误: %s",
// 仅失败需要日志留痕(成功结果已通过 API 响应返回)
if !result.Success {
g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测失败 | 模型: %s (上游: %s) | 延迟: %dms | 代理: %v | 错误: %s",
ch.Name, channelID, testModel, upstreamModel, latencyMs, useProxy, result.Error)
}
+1 -2
View File
@@ -193,8 +193,7 @@ return 1
args := append([]any{activeSetKey}, staleIDs...)
g.Redis().Do(ctx, "SREM", args...)
}
g.Log().Debugf(ctx, "[PREDEDUCT SWEEP] tenant=%d frozen recomputed: %d micro (stale members: %d)",
tenantID, frozenMicro, len(staleIDs))
// 心跳不打日志(每 2 分钟一条太吵),异常路径(围栏重试耗尽/Redis 故障)仍有 Warningf
return
}
g.Log().Warningf(ctx, "[PREDEDUCT SWEEP] tenant=%d frozen recompute aborted after %d fenced retries (hot wallet)",
+42
View File
@@ -387,6 +387,8 @@ func (s *ConfigService) StartSubscriber(ctx context.Context) {
s.cache.Delete(ctx, "public_options")
s.cache.Delete(ctx, "category:"+getCategoryForKey(key))
g.Log().Debugf(ctx, "[PubSub:settings] 缓存已失效: %s", key)
// 缓存失效后再触发回调:回调内部大概率要读该配置的新值
fireSettingsChanged(ctx, key)
}
conn.Close(ctx)
@@ -394,6 +396,46 @@ func (s *ConfigService) StartSubscriber(ctx context.Context) {
}()
}
// ──────────────────────────────────────────
// 配置变更回调
// ──────────────────────────────────────────
var (
settingsHookMu sync.RWMutex
settingsHooks []func(ctx context.Context, key string)
)
// OnSettingsChanged 注册配置变更回调,用于"改了配置要立刻做点什么"的场景
// (如按新的探测间隔重排 cron),而不只是让缓存失效。
//
// 回调经 Redis pub/sub 触发,因此本实例自己的写入同样会走一遍(发布者也订阅了该频道),
// 无需在写入路径上重复调用。Redis 不可用时回调不触发——与缓存失效同等降级。
// 回调在订阅 goroutine 内串行执行,实现必须快速返回、自行兜底 panic。
func OnSettingsChanged(fn func(ctx context.Context, key string)) {
settingsHookMu.Lock()
defer settingsHookMu.Unlock()
settingsHooks = append(settingsHooks, fn)
}
// fireSettingsChanged 依次触发已注册回调,单个回调 panic 不影响其余回调与订阅循环。
func fireSettingsChanged(ctx context.Context, key string) {
settingsHookMu.RLock()
hooks := make([]func(ctx context.Context, key string), len(settingsHooks))
copy(hooks, settingsHooks)
settingsHookMu.RUnlock()
for _, fn := range hooks {
func() {
defer func() {
if r := recover(); r != nil {
g.Log().Errorf(ctx, "[PubSub:settings] 变更回调 panic (key=%s): %v", key, r)
}
}()
fn(ctx, key)
}()
}
}
// ──────────────────────────────────────────
// Validation
// ──────────────────────────────────────────
+83 -4
View File
@@ -2,6 +2,7 @@ package common
import (
"context"
"runtime/debug"
"sync"
"time"
@@ -35,6 +36,10 @@ type CronScheduler struct {
running map[string]bool
runMu sync.Mutex
cron *cron.Cron
// entryIDs job 名 → robfig/cron 条目 ID,供 Reschedule 摘除旧条目。
// 仅在 StartBackground 之后有值。
entryIDs map[string]cron.EntryID
entryMu sync.Mutex
}
var (
@@ -60,8 +65,9 @@ func GetCronScheduler() *CronScheduler {
// NewCronScheduler creates a new CronScheduler.
func NewCronScheduler() *CronScheduler {
return &CronScheduler{
running: make(map[string]bool),
cron: cron.New(),
running: make(map[string]bool),
entryIDs: make(map[string]cron.EntryID),
cron: cron.New(),
}
}
@@ -180,7 +186,7 @@ func (cs *CronScheduler) runJobInternal(ctx context.Context, name, triggeredBy s
// Execute job
startTime := time.Now()
handlerErr := job.Handler(ctx)
handlerErr := runHandlerSafely(ctx, job)
duration := time.Since(startTime)
status := "succeeded"
@@ -197,6 +203,23 @@ func (cs *CronScheduler) runJobInternal(ctx context.Context, name, triggeredBy s
return handlerErr
}
// runHandlerSafely 执行 job handler 并兜底 panic。
//
// robfig/cron v3 的 cron.New() 默认不带 Recover 链,其 startJob 是裸的
// `go func() { j.Run() }()`——任意一个 job panic 都会直接崩掉整个进程,而不只是
// 让该任务失败。此处把 panic 转成普通 error:进程存活,且失败会经 recordExecution
// 落进 sys_cron_jobs,管理后台的任务列表能看到,不会变成一次无声的重启。
func runHandlerSafely(ctx context.Context, job *CronJob) (err error) {
defer func() {
if r := recover(); r != nil {
// 堆栈只进日志:last_error 会入库并在管理后台展示,塞完整堆栈会撑爆展示
g.Log().Errorf(ctx, "cron job panicked: %s: %v\n%s", job.Name, r, debug.Stack())
err = gerror.Newf("任务执行 panic: %v", r)
}
}()
return job.Handler(ctx)
}
// recordExecution persists an execution record to the database.
// Always UPSERT sys_cron_jobs with latest status; no separate execution log table.
func (cs *CronScheduler) recordExecution(ctx context.Context, jobName, status string, startedAt time.Time, duration time.Duration, errMsg, triggeredBy string) {
@@ -268,7 +291,7 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
for _, job := range jobs {
j := job
_, err := cs.cron.AddFunc(j.Schedule, func() {
entryID, err := cs.cron.AddFunc(j.Schedule, func() {
bgCtx := gctx.New()
if err := cs.RunJob(bgCtx, j.Name); err != nil {
g.Log().Errorf(bgCtx, "background cron error: %s: %v", j.Name, err)
@@ -276,7 +299,11 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
})
if err != nil {
g.Log().Errorf(ctx, "failed to register cron job %s with schedule %s: %v", j.Name, j.Schedule, err)
continue
}
cs.entryMu.Lock()
cs.entryIDs[j.Name] = entryID
cs.entryMu.Unlock()
}
cs.cron.Start()
@@ -291,6 +318,58 @@ func (cs *CronScheduler) StartBackground(ctx context.Context) {
g.Log().Info(ctx, "cron scheduler started in background")
}
// Reschedule 更新已注册 job 的调度表达式并即时生效(配置热更新用)。
//
// 先添加新条目、成功后再摘除旧条目:表达式非法时保持原调度不变,
// 避免一次错误配置把任务从调度器里彻底摘掉。
// StartBackground 之前调用无意义(尚无条目),此时只更新 job.Schedule 记录。
func (cs *CronScheduler) Reschedule(ctx context.Context, name, schedule string) error {
cs.mu.RLock()
var job *CronJob
for _, j := range cs.jobs {
if j.Name == name {
job = j
break
}
}
cs.mu.RUnlock()
if job == nil {
return gerror.Newf("job %s not registered", name)
}
cs.entryMu.Lock()
defer cs.entryMu.Unlock()
oldID, started := cs.entryIDs[name]
if !started {
cs.mu.Lock()
job.Schedule = schedule
cs.mu.Unlock()
return nil
}
newID, err := cs.cron.AddFunc(schedule, func() {
bgCtx := gctx.New()
if err := cs.RunJob(bgCtx, name); err != nil {
g.Log().Errorf(bgCtx, "background cron error: %s: %v", name, err)
}
})
if err != nil {
return gerror.Wrapf(err, "invalid schedule %q for job %s", schedule, name)
}
cs.cron.Remove(oldID)
cs.entryIDs[name] = newID
cs.mu.Lock()
oldSchedule := job.Schedule
job.Schedule = schedule
cs.mu.Unlock()
g.Log().Infof(ctx, "cron job %s rescheduled: %s → %s", name, oldSchedule, schedule)
return nil
}
// cronLockTTL 分布式锁过期时间:仅用于实例崩溃时的兜底自动释放(正常完成会主动释放)。
// 取值需大于绝大多数 job 的执行时长;过长则崩溃后会阻塞该 job 至多这么久。
const cronLockTTL = 10 * time.Minute
+119
View File
@@ -0,0 +1,119 @@
package common
import (
"context"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestRunHandlerSafely_Panic转为错误 robfig/cron v3 的 cron.New() 不带 Recover 链,
// startJob 是裸的 `go j.Run()`——任务 panic 会崩掉整个进程。此处必须兜住并转成
// 普通 error,才能既保住进程、又让失败落进 sys_cron_jobs 供管理后台查看。
func TestRunHandlerSafely_Panic转为错误(t *testing.T) {
job := &CronJob{
Name: "panicking_job",
Handler: func(_ context.Context) error {
panic("boom")
},
}
var err error
assert.NotPanics(t, func() {
err = runHandlerSafely(context.Background(), job)
}, "panic 必须被兜住,否则整个进程会崩")
require.Error(t, err)
assert.Contains(t, err.Error(), "boom", "错误信息应保留 panic 内容供排查")
}
func TestRunHandlerSafely_正常返回值透传(t *testing.T) {
ctx := context.Background()
okJob := &CronJob{Name: "ok", Handler: func(_ context.Context) error { return nil }}
assert.NoError(t, runHandlerSafely(ctx, okJob))
failJob := &CronJob{Name: "fail", Handler: func(_ context.Context) error {
return assert.AnError
}}
assert.ErrorIs(t, runHandlerSafely(ctx, failJob), assert.AnError, "普通错误必须原样返回")
}
// TestReschedule_未启动时只更新记录 StartBackground 之前没有 cron 条目,
// 此时重排只改 job.Schedule,供随后的 StartBackground 按新表达式注册。
func TestReschedule_未启动时只更新记录(t *testing.T) {
cs := NewCronScheduler()
cs.Register("job_a", "任务A", "*/5 * * * *", func(_ context.Context) error { return nil })
require.NoError(t, cs.Reschedule(context.Background(), "job_a", "@every 30m"))
jobs := cs.ListJobs()
require.Len(t, jobs, 1)
assert.Equal(t, "@every 30m", jobs[0].Schedule)
}
func TestReschedule_未注册任务报错(t *testing.T) {
cs := NewCronScheduler()
assert.Error(t, cs.Reschedule(context.Background(), "nonexistent", "@every 5m"))
}
// TestReschedule_非法表达式保持原调度 一次错误配置不能把任务从调度器里彻底摘掉,
// 因此实现是「先加新条目、成功后再摘旧条目」。
func TestReschedule_非法表达式保持原调度(t *testing.T) {
ctx := context.Background()
cs := NewCronScheduler()
cs.Register("job_b", "任务B", "@every 5m", func(_ context.Context) error { return nil })
cs.StartBackground(ctx)
defer cs.cron.Stop()
before := cs.ListJobs()[0].Schedule
require.Error(t, cs.Reschedule(ctx, "job_b", "不是合法表达式"))
assert.Equal(t, before, cs.ListJobs()[0].Schedule, "非法表达式不得改动原调度")
// 条目仍在调度器中,未被误摘
assert.Len(t, cs.cron.Entries(), 1)
}
// TestReschedule_有效表达式替换条目 同时验证 @every 语法确实被 cron.New() 的
// 默认解析器接受——探测间隔配置正是靠它支持任意分钟数(*/N 在 N>=60 时会静默失真)。
func TestReschedule_有效表达式替换条目(t *testing.T) {
ctx := context.Background()
cs := NewCronScheduler()
cs.Register("job_c", "任务C", "@every 5m", func(_ context.Context) error { return nil })
cs.StartBackground(ctx)
defer cs.cron.Stop()
require.Len(t, cs.cron.Entries(), 1)
oldID := cs.entryIDs["job_c"]
require.NoError(t, cs.Reschedule(ctx, "job_c", "@every 90m"))
assert.Equal(t, "@every 90m", cs.ListJobs()[0].Schedule)
assert.Len(t, cs.cron.Entries(), 1, "旧条目必须被摘除,否则会双倍触发")
assert.NotEqual(t, oldID, cs.entryIDs["job_c"], "条目 ID 应已更新")
}
// TestOnSettingsChanged_回调panic不影响其余回调 回调在订阅 goroutine 内串行执行,
// 一个回调 panic 不能带崩订阅循环或阻断后续回调。
func TestOnSettingsChanged_回调panic不影响其余回调(t *testing.T) {
settingsHookMu.Lock()
saved := settingsHooks
settingsHooks = nil
settingsHookMu.Unlock()
defer func() {
settingsHookMu.Lock()
settingsHooks = saved
settingsHookMu.Unlock()
}()
var firstCalled, thirdCalled bool
OnSettingsChanged(func(_ context.Context, _ string) { firstCalled = true })
OnSettingsChanged(func(_ context.Context, _ string) { panic("hook boom") })
OnSettingsChanged(func(_ context.Context, _ string) { thirdCalled = true })
assert.NotPanics(t, func() {
fireSettingsChanged(context.Background(), "some_key")
})
assert.True(t, firstCalled)
assert.True(t, thirdCalled, "前一个回调 panic 不得阻断后续回调")
}
@@ -213,6 +213,9 @@ var Registry = []SettingDef{
// ── Channel ──
{Key: "channel_auto_test_enabled", Type: SettingTypeBool, Default: "true", Category: "channel",
Label: "渠道自动探测", Description: "定期向活跃渠道发送测试请求,检测连通性并更新健康度(会消耗少量 Token)。禁用渠道不自动探测,由管理员手动测试确认后再启用"},
{Key: "channel_auto_test_interval_minutes", Type: SettingTypeInt, Default: "5", Category: "channel",
Label: "自动探测间隔(分钟)", Validation: "min:1,max:1440",
Description: "两轮自动探测之间的间隔,保存后即时生效。探测是串行的(每渠道最多 30 秒),渠道多或上游超时多时单轮耗时会变长;上一轮未结束时下一轮自动跳过,因此间隔不宜小于「渠道数 × 30 秒」"},
{Key: "channel_auto_disable_enabled", Type: SettingTypeBool, Default: "false", Category: "channel",
Label: "渠道自动禁用", Description: "渠道熔断持续超过路由策略 breaker.autoDisableAfterSeconds(默认 10 分钟)未恢复时自动禁用"},
{Key: "channel_routing_policy", Type: SettingTypeJSON, Default: "", Category: "channel",
+8 -7
View File
@@ -230,7 +230,7 @@ func GetModelChannels(ctx context.Context, startDate, endDate, modelName string)
SUM(CASE WHEN status = 'success' THEN request_count ELSE 0 END) AS success_count,
COALESCE(SUM(sum_latency_ms), 0) AS sum_latency_ms,
COALESCE(SUM(sum_first_token_ms), 0) AS sum_first_token_ms,
COALESCE(SUM(input_tokens + output_tokens), 0) AS total_tokens,
COALESCE(SUM(output_tokens), 0) AS output_tokens,
COALESCE(SUM(cache_read_tokens), 0) AS cache_read_tokens,
COALESCE(SUM(cache_creation_tokens), 0) AS cache_creation_tokens,
COALESCE(SUM(cache_hit_request_count), 0) AS cache_hit_request_count
@@ -244,7 +244,7 @@ func GetModelChannels(ctx context.Context, startDate, endDate, modelName string)
SuccessCount int64 `json:"success_count"`
SumLatencyMs int64 `json:"sum_latency_ms"`
SumFirstTokenMs int64 `json:"sum_first_token_ms"`
TotalTokens int64 `json:"total_tokens"`
OutputTokens int64 `json:"output_tokens"`
CacheReadTokens int64 `json:"cache_read_tokens"`
CacheCreationTokens int64 `json:"cache_creation_tokens"`
CacheHitRequestCount int64 `json:"cache_hit_request_count"`
@@ -270,7 +270,7 @@ func GetModelChannels(ctx context.Context, startDate, endDate, modelName string)
COUNT(*) FILTER (WHERE status = 'success') AS success_count,
COALESCE(SUM(latency_ms), 0) AS sum_latency_ms,
COALESCE(SUM(first_token_ms), 0) AS sum_first_token_ms,
COALESCE(SUM(input_tokens + output_tokens), 0) AS total_tokens,
COALESCE(SUM(output_tokens), 0) AS output_tokens,
COALESCE(SUM(cache_read_tokens), 0) AS cache_read_tokens,
COALESCE(SUM(cache_creation_tokens), 0) AS cache_creation_tokens,
COUNT(*) FILTER (WHERE cache_read_tokens > 0) AS cache_hit_request_count
@@ -299,7 +299,7 @@ func GetModelChannels(ctx context.Context, startDate, endDate, modelName string)
base.SuccessCount += tc.SuccessCount
base.SumLatencyMs += tc.SumLatencyMs
base.SumFirstTokenMs += tc.SumFirstTokenMs
base.TotalTokens += tc.TotalTokens
base.OutputTokens += tc.OutputTokens
base.CacheReadTokens += tc.CacheReadTokens
base.CacheCreationTokens += tc.CacheCreationTokens
base.CacheHitRequestCount += tc.CacheHitRequestCount
@@ -321,7 +321,7 @@ func GetModelChannels(ctx context.Context, startDate, endDate, modelName string)
}
if err := g.DB().Ctx(ctx).Model("chn_channels").
Fields("id", "name").
Where("id IN ?", channelIDs).
WhereIn("id", channelIDs).
Scan(&channels); err != nil {
g.Log().Warningf(ctx, "query channel names failed: %v", err)
} else {
@@ -348,8 +348,9 @@ func GetModelChannels(ctx context.Context, startDate, endDate, modelName string)
}
tps := float64(0)
if r.SumLatencyMs > 0 {
// TPS = 总Token数 / (总延迟ms / 1000)
tps = float64(r.TotalTokens) / (float64(r.SumLatencyMs) / 1000.0)
// TPS 口径与模型层(buildModelPerformanceList)一致:Σ 输出 token / (Σ 延迟ms / 1000)。
// 不能用输入+输出总量:输入 token 为「含缓存总输入」口径,命中缓存的请求会把 TPS 虚高数倍
tps = float64(r.OutputTokens) / (float64(r.SumLatencyMs) / 1000.0)
}
cacheHitRate := float64(0)
totalCacheTokens := r.CacheReadTokens + r.CacheCreationTokens
+7 -7
View File
@@ -952,15 +952,15 @@ type modelInfoCached struct {
}
// InitHealthScore 初始化渠道健康度记录
// 新渠道无任何 EWMA 数据,落 100 分作为展示初值;首次有真实上报后由维护任务按
// avg(succ_ewma)^α×100 重算。stability_score / consecutive_failures 已废弃,不再写入。
func InitHealthScore(ctx context.Context, channelID int64) error {
_, err := dao.ChnHealthScores.Ctx(ctx).Insert(do.ChnHealthScores{
ChannelId: channelID,
SuccessRate: 100.00,
LatencyMs: 0,
StabilityScore: 100.00,
ConsecutiveFailures: 0,
HealthScore: 100.00,
CalculatedAt: gtime.Now(),
ChannelId: channelID,
SuccessRate: 100.00,
LatencyMs: 0,
HealthScore: 100.00,
CalculatedAt: gtime.Now(),
})
return err
}
+33 -15
View File
@@ -30,7 +30,10 @@ func AutoTestChannels(ctx context.Context) {
Fields("id, name, type, base_url, test_model").
Scan(&channels)
if err != nil {
g.Log().Errorf(ctx, "[Cron] query active channels failed: %v", err)
g.Log().Errorf(ctx, "[ChannelProbe] 自动探测查询活跃渠道失败: %v", err)
return
}
if len(channels) == 0 {
return
}
@@ -38,23 +41,38 @@ func AutoTestChannels(ctx context.Context) {
failCount := 0
for _, ch := range channels {
testCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
result, err := admin.New().TestChannel(testCtx, &v1.ChannelTestReq{ID: ch.ID})
cancel()
if err != nil {
failCount++
g.Log().Warningf(ctx, "[Cron] channel %s (%d) test error: %v", ch.Name, ch.ID, err)
continue
}
if result.Success {
if probeOneChannel(ctx, ch.ID, ch.Name) {
successCount++
} else {
failCount++
if result.Error != "" {
g.Log().Warningf(ctx, "[Cron] channel %s (%d) test failed: %s", ch.Name, ch.ID, result.Error)
}
}
// 失败明细由 TestChannel 以 Warning 打印(含模型/延迟/错误),此处仅 Debug 级轮次汇总
}
g.Log().Debugf(ctx, "[ChannelProbe] 自动探测完成: 共 %d 个渠道 | 成功 %d | 失败 %d",
len(channels), successCount, failCount)
}
// probeOneChannel 探测单个渠道,返回是否成功。
// 独立超时 + panic 兜底:单个渠道无论超时、报错还是 panic,都不影响本轮后续渠道
// (panic 若逃逸到 cron 层虽有 runHandlerSafely 兜底进程,但本轮剩余渠道会被整体跳过)。
func probeOneChannel(ctx context.Context, channelID int64, channelName string) (ok bool) {
defer func() {
if r := recover(); r != nil {
ok = false
g.Log().Errorf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测 panic,跳过该渠道继续下一个: %v",
channelName, channelID, r)
}
}()
testCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
result, err := admin.New().TestChannel(testCtx, &v1.ChannelTestReq{ID: channelID})
if err != nil {
// 探测请求尚未发出(无可用 Key/渠道不存在等),TestChannel 内部无结果日志,此处记录
g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测执行出错: %v", channelName, channelID, err)
return false
}
return result.Success
}
+12 -14
View File
@@ -13,19 +13,19 @@ import (
)
// SnapshotHealthScores 快照所有渠道的当前健康度
// stability_score / consecutive_failures 已废弃(健康体系重写后从未被更新),不再复制,
// 新快照由 DB 默认值填 0;对应迁移 000017 已为 stability_score 补上 DEFAULT。
func SnapshotHealthScores(ctx context.Context) error {
type healthRow struct {
ChannelID int64 `json:"channel_id"`
HealthScore float64 `json:"health_score"`
SuccessRate float64 `json:"success_rate"`
LatencyMs float64 `json:"latency_ms"`
StabilityScore float64 `json:"stability_score"`
ConsecutiveFailures int `json:"consecutive_failures"`
ChannelID int64 `json:"channel_id"`
HealthScore float64 `json:"health_score"`
SuccessRate float64 `json:"success_rate"`
LatencyMs float64 `json:"latency_ms"`
}
var rows []healthRow
err := dao.ChnHealthScores.Ctx(ctx).
Fields("channel_id, health_score, success_rate, latency_ms, stability_score, consecutive_failures").
Fields("channel_id, health_score, success_rate, latency_ms").
Scan(&rows)
if err != nil {
g.Log().Errorf(ctx, "[Cron] query health scores failed: %v", err)
@@ -40,13 +40,11 @@ func SnapshotHealthScores(ctx context.Context) error {
now := gtime.Now()
for _, row := range rows {
_, err = dao.ChnHealthSnapshots.Ctx(ctx).Insert(do.ChnHealthSnapshots{
ChannelId: row.ChannelID,
HealthScore: row.HealthScore,
SuccessRate: row.SuccessRate,
LatencyMs: row.LatencyMs,
StabilityScore: row.StabilityScore,
ConsecutiveFailures: row.ConsecutiveFailures,
SnapshotAt: now,
ChannelId: row.ChannelID,
HealthScore: row.HealthScore,
SuccessRate: row.SuccessRate,
LatencyMs: row.LatencyMs,
SnapshotAt: now,
})
if err != nil {
g.Log().Warningf(ctx, "[Cron] insert snapshot for channel %d failed: %v", row.ChannelID, err)
+6 -6
View File
@@ -14,12 +14,12 @@ type ChnHealthScores struct {
g.Meta `orm:"table:chn_health_scores, do:true"`
Id any // 主键ID
ChannelId any // 关联渠道ID
SuccessRate any // 请求成功率(0-100)
LatencyMs any // 平均延迟(毫秒)
StabilityScore any // 稳定性评分(0-100,基于延迟波动计算)
ConsecutiveFailures any // 连续失败次数(成功后归零)
HealthScore any // 综合健康度(0-100)= 成功率×0.40 + 延迟分×0.25 + 稳定性×0.20 + 连续失败分×0.15
CalculatedAt *gtime.Time // 最近一次计算时间
SuccessRate any // 成功率(0-100)= 该渠道各模型 succ_ewma 均值×100,仅统计有真实上报的模型
LatencyMs any // 平均延迟(毫秒)= 该渠道各模型 lat_ewma 均值,仅统计有真实上报的模型;仅展示用,不参与健康分计算
StabilityScore any // 【已废弃】稳定性评分,健康体系重写后不再写入,保留列仅为兼容历史数据
ConsecutiveFailures any // 【已废弃】连续失败次数,已由 Redis 熔断器的滑动窗口计数取代(dispatch:v1:breaker:*),保留列仅为兼容历史数据
HealthScore any // 综合健康度(0-100)= avg(succ_ewma)^α × 100,α 取路由策略 health.alpha(默认 2),与调度 healthFactor 同源;每 5 分钟由维护任务聚合落盘,另可由渠道测试/重置健康度即时触发;Redis 读失败或全部模型无真实上报时保留旧值不覆盖。调度决策不读此表,仅供管理后台展示
CalculatedAt *gtime.Time // 最近一次聚合落盘时间
CreatedAt *gtime.Time // 创建时间
UpdatedAt *gtime.Time // 更新时间
}
+3 -3
View File
@@ -14,10 +14,10 @@ type ChnHealthSnapshots struct {
g.Meta `orm:"table:chn_health_snapshots, do:true"`
Id any // 主键ID
ChannelId any // 关联渠道ID
HealthScore any // 综合健康度(0-100)
HealthScore any // 综合健康度(0-100)快照,取自 chn_health_scores.health_score
SuccessRate any // 请求成功率(0-100)
LatencyMs any // 平均延迟(毫秒)
StabilityScore any // 稳定性评分(0-100)
ConsecutiveFailures any // 连续失败次数
StabilityScore any // 【已废弃】稳定性评分,不再写入,新快照为 0
ConsecutiveFailures any // 【已废弃】连续失败次数,不再写入,新快照为 0
SnapshotAt *gtime.Time // 快照时间
}
+1 -1
View File
@@ -19,7 +19,7 @@ type MdlTenantModels struct {
Enabled any // 是否启用(禁用后该租户无法调用此模型)
CustomInputPrice *decimal.Decimal // 租户自定义输入价格(NULL 表示使用默认定价)
CustomOutputPrice *decimal.Decimal // 租户自定义输出价格(NULL 表示使用默认定价)
Multiplier any // 租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与最终价格计算:最终价格 = 基础价格 × 租户倍率。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier
Multiplier any // 租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与计费,当前实际生效公式 = 基础价格 × 租户乘数 × 时段乘数;模型乘数为预留能力(mdl_models 无 multiplier 字段、computeCost 未接入),bil_records.model_multiplier 快照恒为 1.0。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier(级别兜底仅在前两者均未设置、且级别倍率 <1.0 时生效)
CreatedAt *gtime.Time // 创建时间
UpdatedAt *gtime.Time // 更新时间
BillingMode any // 覆盖模型计费方式(NULL表示跟随模型默认)
+10 -10
View File
@@ -10,14 +10,14 @@ import (
// ChnHealthScores is the golang structure for table chn_health_scores.
type ChnHealthScores struct {
Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID
ChannelId int64 `json:"channel_id" orm:"channel_id" description:"关联渠道ID"` // 关联渠道ID
SuccessRate float64 `json:"success_rate" orm:"success_rate" description:"请求成功率(0-100)"` // 请求成功率(0-100)
LatencyMs float64 `json:"latency_ms" orm:"latency_ms" description:"平均延迟(毫秒)"` // 平均延迟(毫秒)
StabilityScore float64 `json:"stability_score" orm:"stability_score" description:"稳定性评分(0-100,基于延迟波动计算)"` // 稳定性评分(0-100,基于延迟波动计算)
ConsecutiveFailures int `json:"consecutive_failures" orm:"consecutive_failures" description:"连续失败次数(成功后归零)"` // 连续失败次数(成功后归零)
HealthScore float64 `json:"health_score" orm:"health_score" description:"综合健康度(0-100)= 成功率×0.40 + 延迟分×0.25 + 稳定性×0.20 + 连续失败分×0.15"` // 综合健康度(0-100)= 成功率×0.40 + 延迟分×0.25 + 稳定性×0.20 + 连续失败分×0.15
CalculatedAt *gtime.Time `json:"calculated_at" orm:"calculated_at" description:"最近一次计算时间"` // 最近一次计算时间
CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间
UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间
Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID
ChannelId int64 `json:"channel_id" orm:"channel_id" description:"关联渠道ID"` // 关联渠道ID
SuccessRate float64 `json:"success_rate" orm:"success_rate" description:"成功率(0-100)= 该渠道各模型 succ_ewma 均值×100,仅统计有真实上报的模型"` // 成功率(0-100)= 该渠道各模型 succ_ewma 均值×100,仅统计有真实上报的模型
LatencyMs float64 `json:"latency_ms" orm:"latency_ms" description:"平均延迟(毫秒)= 该渠道各模型 lat_ewma 均值,仅统计有真实上报的模型;仅展示用,不参与健康分计算"` // 平均延迟(毫秒)= 该渠道各模型 lat_ewma 均值,仅统计有真实上报的模型;仅展示用,不参与健康分计算
StabilityScore float64 `json:"stability_score" orm:"stability_score" description:"【已废弃】稳定性评分,健康体系重写后不再写入,保留列仅为兼容历史数据"` // 【已废弃】稳定性评分,健康体系重写后不再写入,保留列仅为兼容历史数据
ConsecutiveFailures int `json:"consecutive_failures" orm:"consecutive_failures" description:"【已废弃】连续失败次数,已由 Redis 熔断器的滑动窗口计数取代(dispatch:v1:breaker:*),保留列仅为兼容历史数据"` // 【已废弃】连续失败次数,已由 Redis 熔断器的滑动窗口计数取代(dispatch:v1:breaker:*),保留列仅为兼容历史数据
HealthScore float64 `json:"health_score" orm:"health_score" description:"综合健康度(0-100)= avg(succ_ewma)^α × 100,α 取路由策略 health.alpha(默认 2),与调度 healthFactor 同源;每 5 分钟由维护任务聚合落盘,另可由渠道测试/重置健康度即时触发;Redis 读失败或全部模型无真实上报时保留旧值不覆盖。调度决策不读此表,仅供管理后台展示"` // 综合健康度(0-100)= avg(succ_ewma)^α × 100,α 取路由策略 health.alpha(默认 2),与调度 healthFactor 同源;每 5 分钟由维护任务聚合落盘,另可由渠道测试/重置健康度即时触发;Redis 读失败或全部模型无真实上报时保留旧值不覆盖。调度决策不读此表,仅供管理后台展示
CalculatedAt *gtime.Time `json:"calculated_at" orm:"calculated_at" description:"最近一次聚合落盘时间"` // 最近一次聚合落盘时间
CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间
UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间
}
@@ -10,12 +10,12 @@ import (
// ChnHealthSnapshots is the golang structure for table chn_health_snapshots.
type ChnHealthSnapshots struct {
Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID
ChannelId int64 `json:"channel_id" orm:"channel_id" description:"关联渠道ID"` // 关联渠道ID
HealthScore float64 `json:"health_score" orm:"health_score" description:"综合健康度(0-100)"` // 综合健康度(0-100)
SuccessRate float64 `json:"success_rate" orm:"success_rate" description:"请求成功率(0-100)"` // 请求成功率(0-100)
LatencyMs float64 `json:"latency_ms" orm:"latency_ms" description:"平均延迟(毫秒)"` // 平均延迟(毫秒)
StabilityScore float64 `json:"stability_score" orm:"stability_score" description:"稳定性评分(0-100)"` // 稳定性评分(0-100)
ConsecutiveFailures int `json:"consecutive_failures" orm:"consecutive_failures" description:"连续失败次数"` // 连续失败次数
SnapshotAt *gtime.Time `json:"snapshot_at" orm:"snapshot_at" description:"快照时间"` // 快照时间
Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID
ChannelId int64 `json:"channel_id" orm:"channel_id" description:"关联渠道ID"` // 关联渠道ID
HealthScore float64 `json:"health_score" orm:"health_score" description:"综合健康度(0-100)快照,取自 chn_health_scores.health_score"` // 综合健康度(0-100)快照,取自 chn_health_scores.health_score
SuccessRate float64 `json:"success_rate" orm:"success_rate" description:"请求成功率(0-100)"` // 请求成功率(0-100)
LatencyMs float64 `json:"latency_ms" orm:"latency_ms" description:"平均延迟(毫秒)"` // 平均延迟(毫秒)
StabilityScore float64 `json:"stability_score" orm:"stability_score" description:"【已废弃】稳定性评分,不再写入,新快照为 0"` // 【已废弃】稳定性评分,不再写入,新快照为 0
ConsecutiveFailures int `json:"consecutive_failures" orm:"consecutive_failures" description:"【已废弃】连续失败次数,不再写入,新快照为 0"` // 【已废弃】连续失败次数,不再写入,新快照为 0
SnapshotAt *gtime.Time `json:"snapshot_at" orm:"snapshot_at" description:"快照时间"` // 快照时间
}
+17 -17
View File
@@ -11,21 +11,21 @@ import (
// MdlTenantModels is the golang structure for table mdl_tenant_models.
type MdlTenantModels struct {
Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID
TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID"` // 租户ID
ModelId int64 `json:"model_id" orm:"model_id" description:"模型ID"` // 模型ID
Enabled bool `json:"enabled" orm:"enabled" description:"是否启用(禁用后该租户无法调用此模型)"` // 是否启用(禁用后该租户无法调用此模型)
CustomInputPrice *decimal.Decimal `json:"custom_input_price" orm:"custom_input_price" description:"租户自定义输入价格(NULL 表示使用默认定价)"` // 租户自定义输入价格(NULL 表示使用默认定价)
CustomOutputPrice *decimal.Decimal `json:"custom_output_price" orm:"custom_output_price" description:"租户自定义输出价格(NULL 表示使用默认定价)"` // 租户自定义输出价格(NULL 表示使用默认定价)
Multiplier decimal.Decimal `json:"multiplier" orm:"multiplier" description:"租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与最终价格计算:最终价格 = 基础价格 × 租户倍率。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier"` // 租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与最终价格计算:最终价格 = 基础价格 × 租户倍率。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier
CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间
UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间
BillingMode string `json:"billing_mode" orm:"billing_mode" description:"覆盖模型计费方式(NULL表示跟随模型默认)"` // 覆盖模型计费方式(NULL表示跟随模型默认)
PerRequestPrice *decimal.Decimal `json:"per_request_price" orm:"per_request_price" description:"按次计费单价(覆盖模型默认,仅 billing_mode = per_request 时有效)"` // 按次计费单价(覆盖模型默认,仅 billing_mode = per_request 时有效)
DiscountRatio *decimal.Decimal `json:"discount_ratio" orm:"discount_ratio" description:"折扣比例(如0.8表示八折,NULL表示不打折,优先于 multiplier 使用)"` // 折扣比例(如0.8表示八折,NULL表示不打折,优先于 multiplier 使用)
MaxConcurrency int `json:"max_concurrency" orm:"max_concurrency" description:"单模型并发上限(NULL表示不限制)"` // 单模型并发上限(NULL表示不限制)
ChannelScope string `json:"channel_scope" orm:"channel_scope" description:"渠道范围覆盖(NULL跟随租户默认,[]表示全部,数组表示指定渠道ID)"` // 渠道范围覆盖(NULL跟随租户默认,[]表示全部,数组表示指定渠道ID)
CustomCacheReadPrice *decimal.Decimal `json:"custom_cache_read_price" orm:"custom_cache_read_price" description:"自定义缓存读取价格($/1M token),NULL 表示使用基础定价"` // 自定义缓存读取价格($/1M token),NULL 表示使用基础定价
CustomCacheCreationPrice *decimal.Decimal `json:"custom_cache_creation_price" orm:"custom_cache_creation_price" description:"自定义缓存创建价格($/1M token),NULL 表示使用基础定价"` // 自定义缓存创建价格($/1M token),NULL 表示使用基础定价
CustomPricingTiers string `json:"custom_pricing_tiers" orm:"custom_pricing_tiers" description:"自定义阶梯定价(JSONB 数组),格式: [{\"min_tokens\":0,\"max_tokens\":100000,\"input_price\":0.5,\"output_price\":1.5,\"cache_read_price\":0.1,\"cache_creation_price\":0.2}]"` // 自定义阶梯定价(JSONB 数组),格式: [{"min_tokens":0,"max_tokens":100000,"input_price":0.5,"output_price":1.5,"cache_read_price":0.1,"cache_creation_price":0.2}]
Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID
TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID"` // 租户ID
ModelId int64 `json:"model_id" orm:"model_id" description:"模型ID"` // 模型ID
Enabled bool `json:"enabled" orm:"enabled" description:"是否启用(禁用后该租户无法调用此模型)"` // 是否启用(禁用后该租户无法调用此模型)
CustomInputPrice *decimal.Decimal `json:"custom_input_price" orm:"custom_input_price" description:"租户自定义输入价格(NULL 表示使用默认定价)"` // 租户自定义输入价格(NULL 表示使用默认定价)
CustomOutputPrice *decimal.Decimal `json:"custom_output_price" orm:"custom_output_price" description:"租户自定义输出价格(NULL 表示使用默认定价)"` // 租户自定义输出价格(NULL 表示使用默认定价)
Multiplier decimal.Decimal `json:"multiplier" orm:"multiplier" description:"租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与计费,当前实际生效公式 = 基础价格 × 租户乘数 × 时段乘数;模型乘数为预留能力(mdl_models 无 multiplier 字段、computeCost 未接入),bil_records.model_multiplier 快照恒为 1.0。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier(级别兜底仅在前两者均未设置、且级别倍率 <1.0 时生效)"` // 租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与计费,当前实际生效公式 = 基础价格 × 租户乘数 × 时段乘数;模型乘数为预留能力(mdl_models 无 multiplier 字段、computeCost 未接入),bil_records.model_multiplier 快照恒为 1.0。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier(级别兜底仅在前两者均未设置、且级别倍率 <1.0 时生效)
CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间
UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间
BillingMode string `json:"billing_mode" orm:"billing_mode" description:"覆盖模型计费方式(NULL表示跟随模型默认)"` // 覆盖模型计费方式(NULL表示跟随模型默认)
PerRequestPrice *decimal.Decimal `json:"per_request_price" orm:"per_request_price" description:"按次计费单价(覆盖模型默认,仅 billing_mode = per_request 时有效)"` // 按次计费单价(覆盖模型默认,仅 billing_mode = per_request 时有效)
DiscountRatio *decimal.Decimal `json:"discount_ratio" orm:"discount_ratio" description:"折扣比例(如0.8表示八折,NULL表示不打折,优先于 multiplier 使用)"` // 折扣比例(如0.8表示八折,NULL表示不打折,优先于 multiplier 使用)
MaxConcurrency int `json:"max_concurrency" orm:"max_concurrency" description:"单模型并发上限(NULL表示不限制)"` // 单模型并发上限(NULL表示不限制)
ChannelScope string `json:"channel_scope" orm:"channel_scope" description:"渠道范围覆盖(NULL跟随租户默认,[]表示全部,数组表示指定渠道ID)"` // 渠道范围覆盖(NULL跟随租户默认,[]表示全部,数组表示指定渠道ID)
CustomCacheReadPrice *decimal.Decimal `json:"custom_cache_read_price" orm:"custom_cache_read_price" description:"自定义缓存读取价格($/1M token),NULL 表示使用基础定价"` // 自定义缓存读取价格($/1M token),NULL 表示使用基础定价
CustomCacheCreationPrice *decimal.Decimal `json:"custom_cache_creation_price" orm:"custom_cache_creation_price" description:"自定义缓存创建价格($/1M token),NULL 表示使用基础定价"` // 自定义缓存创建价格($/1M token),NULL 表示使用基础定价
CustomPricingTiers string `json:"custom_pricing_tiers" orm:"custom_pricing_tiers" description:"自定义阶梯定价(JSONB 数组),格式: [{\"min_tokens\":0,\"max_tokens\":100000,\"input_price\":0.5,\"output_price\":1.5,\"cache_read_price\":0.1,\"cache_creation_price\":0.2}]"` // 自定义阶梯定价(JSONB 数组),格式: [{"min_tokens":0,"max_tokens":100000,"input_price":0.5,"output_price":1.5,"cache_read_price":0.1,"cache_creation_price":0.2}]
}
+61
View File
@@ -0,0 +1,61 @@
-- +goose Up
-- 修正 mdl_tenant_models.multiplier 列注释。
--
-- 背景:本列注释历史上有两版,且线上库的实际值与迁移记录不符——
-- 000001 初始版:'...最终价格 = 基础价格 × 模型倍率 × 租户倍率'
-- 000014 Up 改写为带倍率优先级链的版本,但线上库至今仍是 000001 的文案
-- (2026-08-27 重跑 gf gen dao 时发现:生成结果回退掉了代码里的优先级链说明)。
--
-- 两版都已不准确,按当前实现重写:
-- 1) 模型乘数是预留能力,mdl_models 无 multiplier 字段,computeCost 也不纳入,
-- bil_records.model_multiplier 快照恒为 1.0(见 pricing.go 的「4.5 模型倍率」段);
-- 2) 000016 引入时段定价后,computeCost 实际乘的是 TenantMultiplier × TimeMultiplier
-- (pricing.go:381 / 406 / 546),旧文案均未体现时段乘数。
COMMENT ON COLUMN mdl_tenant_models.multiplier IS '租户价格倍率(VIP 折扣)。作为 tenant_multiplier 参与计费,当前实际生效公式 = 基础价格 × 租户乘数 × 时段乘数;模型乘数为预留能力(mdl_models 无 multiplier 字段、computeCost 未接入),bil_records.model_multiplier 快照恒为 1.0。倍率来源优先级:discount_ratio > multiplier > 租户等级 price_multiplier(级别兜底仅在前两者均未设置、且级别倍率 <1.0 时生效)';
-- 渠道健康度:修正与实现不符的列注释,并把两个僵尸列标记为已废弃。
--
-- 背景:健康体系已重写为「Redis 渠道×模型 EWMA 为唯一真相 + 每 5 分钟聚合落盘」,
-- 但 chn_health_scores 的列注释仍停留在旧的加权公式,读注释会得到完全错误的结论。
-- 实际公式:health_score = avg(succ_ewma)^α × 100
-- (α = 路由策略 health.alpha,默认 2;均值只统计有真实上报的模型)
-- 注释声称:成功率×0.40 + 延迟分×0.25 + 稳定性×0.20 + 连续失败分×0.15
--
-- stability_score / consecutive_failures 自重写后从未被更新过,只在建渠道
-- (InitHealthScore)和重置健康度时写入常量,趋势图与调度均不消费。保留列不删除
-- (避免破坏已有数据与历史快照),改为注释标注废弃 + 代码停止写入。
COMMENT ON COLUMN chn_health_scores.success_rate IS '成功率(0-100)= 该渠道各模型 succ_ewma 均值×100,仅统计有真实上报的模型';
COMMENT ON COLUMN chn_health_scores.latency_ms IS '平均延迟(毫秒)= 该渠道各模型 lat_ewma 均值,仅统计有真实上报的模型;仅展示用,不参与健康分计算';
COMMENT ON COLUMN chn_health_scores.health_score IS '综合健康度(0-100)= avg(succ_ewma)^α × 100,α 取路由策略 health.alpha(默认 2),与调度 healthFactor 同源;每 5 分钟由维护任务聚合落盘,另可由渠道测试/重置健康度即时触发;Redis 读失败或全部模型无真实上报时保留旧值不覆盖。调度决策不读此表,仅供管理后台展示';
COMMENT ON COLUMN chn_health_scores.calculated_at IS '最近一次聚合落盘时间';
COMMENT ON COLUMN chn_health_scores.stability_score IS '【已废弃】稳定性评分,健康体系重写后不再写入,保留列仅为兼容历史数据';
COMMENT ON COLUMN chn_health_scores.consecutive_failures IS '【已废弃】连续失败次数,已由 Redis 熔断器的滑动窗口计数取代(dispatch:v1:breaker:*),保留列仅为兼容历史数据';
COMMENT ON COLUMN chn_health_snapshots.health_score IS '综合健康度(0-100)快照,取自 chn_health_scores.health_score';
COMMENT ON COLUMN chn_health_snapshots.stability_score IS '【已废弃】稳定性评分,不再写入,新快照为 0';
COMMENT ON COLUMN chn_health_snapshots.consecutive_failures IS '【已废弃】连续失败次数,不再写入,新快照为 0';
-- stability_score 为 NOT NULL 且无默认值,代码停止写入后插入会失败,补默认值。
-- consecutive_failures 已有 DEFAULT 0,无需处理。
ALTER TABLE chn_health_snapshots
ALTER COLUMN stability_score SET DEFAULT 0;
-- +goose Down
ALTER TABLE chn_health_snapshots
ALTER COLUMN stability_score DROP DEFAULT;
COMMENT ON COLUMN chn_health_snapshots.consecutive_failures IS '连续失败次数';
COMMENT ON COLUMN chn_health_snapshots.stability_score IS '稳定性评分(0-100)';
COMMENT ON COLUMN chn_health_snapshots.health_score IS '综合健康度(0-100)';
COMMENT ON COLUMN chn_health_scores.consecutive_failures IS '连续失败次数(成功后归零)';
COMMENT ON COLUMN chn_health_scores.stability_score IS '稳定性评分(0-100,基于延迟波动计算)';
COMMENT ON COLUMN chn_health_scores.calculated_at IS '最近一次计算时间';
COMMENT ON COLUMN chn_health_scores.health_score IS '综合健康度(0-100)= 成功率×0.40 + 延迟分×0.25 + 稳定性×0.20 + 连续失败分×0.15';
COMMENT ON COLUMN chn_health_scores.latency_ms IS '平均延迟(毫秒)';
COMMENT ON COLUMN chn_health_scores.success_rate IS '请求成功率(0-100)';
COMMENT ON COLUMN mdl_tenant_models.multiplier IS '租户价格倍率(VIP 折扣等,最终价格 = 基础价格 × 模型倍率 × 租户倍率)';
+2 -3
View File
@@ -159,8 +159,7 @@ type Outcome struct {
Success bool
Class ErrorClass
LatencyMs float64
// Probe 标记探测类结果(管理后台手动测试 / 自动探测 cron),非真实用户流量。
// 探测失败只喂熔断窗口不衰减健康 EWMA:无流量渠道若探测持续失败(如 test_model
// 配置错误),周期性 ×decay 会把健康分指数拖垮且无真实流量对冲回升。
// Probe 标记本结果来自探测请求(管理后台渠道测试 / 自动探测 cron)而非真实流量。
// 纯核心库不消费,仅供适配层观测日志区分健康分变化来源。
Probe bool
}
+25
View File
@@ -26,6 +26,28 @@ console.log(
'color: #666;',
)
// 构建发版后旧 hash chunk 失效时自动整页刷新加载新版本;标记防止刷新死循环
const CHUNK_RELOAD_KEY = 'spa_stale_chunk_reloaded'
function reloadOnStaleChunk() {
if (!sessionStorage.getItem(CHUNK_RELOAD_KEY)) {
sessionStorage.setItem(CHUNK_RELOAD_KEY, '1')
window.location.reload()
}
}
window.addEventListener('vite:preloadError', (event) => {
event.preventDefault()
reloadOnStaleChunk()
})
router.onError((error) => {
const msg = String(error instanceof Error ? error.message : error)
if (/Failed to fetch dynamically imported module|Importing a module script failed|error loading dynamically imported module/i.test(msg)) {
reloadOnStaleChunk()
}
})
const app = createApp(App)
app.config.errorHandler = (error, _instance, info) => {
console.error('[VueError]', info, error)
@@ -34,3 +56,6 @@ app.use(createPinia())
app.use(ArcoVue)
app.use(router)
app.mount('#app')
// 页面成功启动后清除标记,后续每次发版仍可各自动刷新一次
sessionStorage.removeItem(CHUNK_RELOAD_KEY)
@@ -1069,6 +1069,13 @@ function formatHeaders(headers: Record<string, string>): string {
{{ detail.health_score.toFixed(0) }}
</span>
<span v-else style="color: #94a3b8">N/A</span>
<!-- 渠道分是各模型均值,单个模型全挂会被摊薄;最差项低于健康档时点名 -->
<span
v-if="detail.status === 'active' && detail.worst_model && detail.worst_model_score != null && detail.worst_model_score < 80"
:style="{ marginLeft: '8px', color: healthColor(detail.worst_model_score), fontSize: '12px' }"
>
⚠ 最差模型 {{ detail.worst_model }} {{ detail.worst_model_score.toFixed(0) }} 分
</span>
</ADescriptionsItem>
<ADescriptionsItem label="调度状态">
<!-- disabled / testing 渠道不在目录快照中,熔断状态无意义,置灰展示 -->
+16 -2
View File
@@ -121,10 +121,24 @@ const columns: TableColumnData[] = [
},
},
{
title: '健康度', dataIndex: 'health_score', width: 90,
title: '健康度', dataIndex: 'health_score', width: 140,
render({ record }) {
if (record.health_score === null || record.health_score === undefined) return h('span', { style: 'color:#94a3b8' }, 'N/A')
return h('span', { style: { color: healthColor(record.health_score), fontWeight: 600 } }, record.health_score.toFixed(0))
const score = h('span', { style: { color: healthColor(record.health_score), fontWeight: 600 } }, record.health_score.toFixed(0))
// 渠道分是各模型均值,单个模型全挂会被其余健康模型摊薄(10 健康 + 1 全挂 ≈ 90 分绿灯)。
// 最差模型低于健康档时补一行提示,让"是哪个模型拖后腿"在列表页直接可见。
// disabled / testing 渠道不在目录快照,无最差项数据。
const worst = record.worst_model_score
if (record.status !== 'active' || !record.worst_model || worst === null || worst === undefined || worst >= 80) {
return score
}
return h('div', { style: 'display:flex; flex-direction:column; gap:2px; line-height:1.3' }, [
score,
h('span', {
style: `color:${healthColor(worst)}; font-size:12px; overflow:hidden; text-overflow:ellipsis; white-space:nowrap`,
title: `最差模型 ${record.worst_model}:${worst.toFixed(0)} 分`,
}, `⚠ ${record.worst_model} ${worst.toFixed(0)}`),
])
},
},
{
@@ -16,6 +16,19 @@ const values = useFormValues()
@change="(v: string | number | boolean) => values['channel_auto_test_enabled'] = v"
/>
</div>
<div class="section-grid" style="margin-top: 12px">
<AFormItem label="探测间隔(分钟)">
<AInputNumber
:model-value="values['channel_auto_test_interval_minutes'] as number"
@change="(v: number | undefined) => values['channel_auto_test_interval_minutes'] = v ?? 5"
:min="1" :max="1440" style="width: 100%"
/>
</AFormItem>
</div>
<div class="section-desc">
保存后即时生效,无需重启。探测是串行的(每渠道最多 30 秒),渠道多或上游超时多时单轮耗时会变长;
上一轮未结束时下一轮自动跳过,因此间隔不宜小于「渠道数 × 30 秒」。
</div>
</div>
<!-- 自动禁用 -->
+25
View File
@@ -29,6 +29,28 @@ console.log(
'color: #666;',
)
// 构建发版后旧 hash chunk 失效时自动整页刷新加载新版本;标记防止刷新死循环
const CHUNK_RELOAD_KEY = 'spa_stale_chunk_reloaded'
function reloadOnStaleChunk() {
if (!sessionStorage.getItem(CHUNK_RELOAD_KEY)) {
sessionStorage.setItem(CHUNK_RELOAD_KEY, '1')
window.location.reload()
}
}
window.addEventListener('vite:preloadError', (event) => {
event.preventDefault()
reloadOnStaleChunk()
})
router.onError((error) => {
const msg = String(error instanceof Error ? error.message : error)
if (/Failed to fetch dynamically imported module|Importing a module script failed|error loading dynamically imported module/i.test(msg)) {
reloadOnStaleChunk()
}
})
const app = createApp(App)
app.config.errorHandler = (error, _instance, info) => {
console.error('[VueError]', info, error)
@@ -38,3 +60,6 @@ app.use(createPinia())
app.use(router)
app.use(naive)
app.mount('#app')
// 页面成功启动后清除标记,后续每次发版仍可各自动刷新一次
sessionStorage.removeItem(CHUNK_RELOAD_KEY)
+15 -18
View File
@@ -80,16 +80,11 @@ watch(selectedKeyId, loadModels, { immediate: true })
<template>
<div>
<div class="page-header">
<h1 class="page-title">API Playground</h1>
<p class="page-description">在线调试 AI 模型。不同大模型参数不一致,此处仅做简单功能演示,验证资源可用。请还用API调用获取最佳体验。</p>
</div>
<!-- Toolbar: 分类标签 + API Key 选择器 -->
<div class="card mb-6">
<div class="flex flex-col gap-3 px-5 py-4 sm:flex-row sm:items-center sm:justify-between">
<!-- 分类标签 -->
<div class="tabs">
<!-- 紧凑工具栏:左侧模型类型标签 + 右侧 API Key 单行收纳 -->
<div class="card mb-3">
<div class="flex flex-wrap items-center gap-x-3 gap-y-2 px-4 py-3 sm:px-5">
<!-- 左侧:模型类型标签 -->
<div class="tabs flex-wrap">
<button v-for="tab in tabs" :key="tab.key"
class="tab"
:class="{ 'tab-active': activeTab === tab.key }"
@@ -98,8 +93,16 @@ watch(selectedKeyId, loadModels, { immediate: true })
</button>
</div>
<!-- API Key 区域 -->
<div class="flex shrink-0 items-center gap-2">
<!-- 模型加载指示内联在工具栏中,避免独立行造成的布局跳动 -->
<transition name="fade">
<div v-if="modelsLoading" class="flex shrink-0 items-center gap-1.5 text-xs text-gray-400">
<div class="spinner h-3.5 w-3.5 text-primary-600"></div>
加载模型...
</div>
</transition>
<!-- 右侧:API Key 区域 -->
<div class="ml-auto flex shrink-0 items-center gap-2">
<!-- 加载中 -->
<template v-if="keyLoading">
<div class="spinner h-4 w-4 text-primary-600"></div>
@@ -128,12 +131,6 @@ watch(selectedKeyId, loadModels, { immediate: true })
</div>
<template v-if="revealedKey">
<transition name="fade">
<div v-if="modelsLoading" class="mb-4 flex items-center gap-2 text-sm text-gray-500">
<div class="spinner h-4 w-4 text-primary-600"></div>
加载模型...
</div>
</transition>
<ChatTab v-if="activeTab === 'chat'" :models="modelsByCategory('chat').value" :api-key="revealedKey" />
<ImageTab v-if="activeTab === 'image'" :models="modelsByCategory('image').value" :api-key="revealedKey" />
<VideoTab v-if="activeTab === 'video'" :models="modelsByCategory('video').value" :api-key="revealedKey" />
@@ -48,6 +48,7 @@ const params = reactive({
systemPrompt: '',
topP: 1.0,
stream: true,
thinking: false,
})
const imageConfig = reactive({
@@ -101,6 +102,10 @@ function buildRequestBody(apiMessages: { role: string; content: string }[]) {
max_tokens: params.maxTokens,
stream: params.stream,
}
// 思考控制:通过 thinking 参数开关({"type": "enabled"} / {"type": "disabled"}),图片模型不参与
if (!isImageModel.value) {
body.thinking = { type: params.thinking ? 'enabled' : 'disabled' }
}
if (params.stream) {
body.stream_options = { include_usage: true }
}
@@ -284,11 +289,13 @@ function clearChat() {
</script>
<template>
<div class="flex flex-col lg:h-[calc(100vh-20rem)] lg:overflow-hidden">
<div class="grid grid-cols-1 lg:grid-cols-4 gap-6 lg:min-h-0 lg:flex-1">
<!-- 高度偏移 ≈ 顶栏与页边距(约128px) + 上方工具栏卡片(约116px),保证对话卡片完整落在视口内 -->
<div class="flex flex-col lg:h-[calc(100vh-15.5rem)] lg:overflow-hidden">
<!-- grid-rows minmax(0,1fr) 强制行高等于容器高,配合左列内部滚动,避免行高被参数卡撑破导致底部输入框被裁切 -->
<div class="grid grid-cols-1 lg:grid-cols-4 lg:grid-rows-[minmax(0,1fr)] gap-6 lg:min-h-0 lg:flex-1">
<!-- Left: Parameters -->
<div class="lg:col-span-1">
<div class="card lg:sticky lg:top-6">
<div class="lg:col-span-1 lg:min-h-0 lg:overflow-y-auto lg:pr-1">
<div class="card">
<div class="card-header">
<h3 class="text-sm font-semibold text-gray-900">参数配置</h3>
</div>
@@ -313,6 +320,13 @@ function clearChat() {
<label class="input-label">System Prompt</label>
<n-input v-model:value="params.systemPrompt" type="textarea" :rows="3" placeholder="设置系统提示词..." />
</div>
<div v-if="!isImageModel">
<div class="flex items-center justify-between">
<label class="input-label mb-0">深度思考</label>
<n-switch v-model:value="params.thinking" />
</div>
<p class="mt-1 text-xs text-gray-400">通过 thinking 参数控制开关,仅对支持的模型生效</p>
</div>
<div class="flex items-center justify-between">
<label class="input-label mb-0">流式响应</label>
<n-switch v-model:value="params.stream" />
@@ -335,7 +349,7 @@ function clearChat() {
</div>
<!-- Right: Chat -->
<div class="lg:col-span-3 min-h-0 h-[65vh] lg:h-auto">
<div class="h-[65vh] lg:col-span-3 lg:h-auto lg:min-h-0">
<div class="card flex flex-col h-full overflow-hidden">
<div class="px-6 py-2 bg-amber-50 border-b border-amber-200 text-center">
<span class="text-xs text-amber-700 font-medium">Playground 模式 — 使用真实 API Key 调用,产生实际费用</span>