diff --git a/api/admin/v1/channel.go b/api/admin/v1/channel.go index 796a674..b151590 100644 --- a/api/admin/v1/channel.go +++ b/api/admin/v1/channel.go @@ -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 导出渠道列表请求 diff --git a/docs/reference/goframe-conventions.md b/docs/reference/goframe-conventions.md index 8ca619e..c89df67 100644 --- a/docs/reference/goframe-conventions.md +++ b/docs/reference/goframe-conventions.md @@ -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 读取路径。 diff --git a/docs/reference/多项目渠道调度对比.md b/docs/reference/多项目渠道调度对比.md index 5262b52..f26d9c9 100644 --- a/docs/reference/多项目渠道调度对比.md +++ b/docs/reference/多项目渠道调度对比.md @@ -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 两侧的分析不受影响,横向对比结论仍有参考价值。 + --- ## 目录 diff --git a/internal/cmd/cmd.go b/internal/cmd/cmd.go index 337e1df..88a6a1a 100644 --- a/internal/cmd/cmd.go +++ b/internal/cmd/cmd.go @@ -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) } } diff --git a/internal/cmd/spa_handler_test.go b/internal/cmd/spa_handler_test.go new file mode 100644 index 0000000..01e030c --- /dev/null +++ b/internal/cmd/spa_handler_test.go @@ -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("
"), + }, + "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(), "") { + 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(), "") { + 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") +} diff --git a/internal/dao/internal/chn_health_scores.go b/internal/dao/internal/chn_health_scores.go index e550c6a..bb1cf8b 100644 --- a/internal/dao/internal/chn_health_scores.go +++ b/internal/dao/internal/chn_health_scores.go @@ -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 // 更新时间 } diff --git a/internal/dao/internal/chn_health_snapshots.go b/internal/dao/internal/chn_health_snapshots.go index 7034d1e..02e7702 100644 --- a/internal/dao/internal/chn_health_snapshots.go +++ b/internal/dao/internal/chn_health_snapshots.go @@ -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 // 快照时间 } diff --git a/internal/dao/internal/mdl_tenant_models.go b/internal/dao/internal/mdl_tenant_models.go index 885b674..9e278c6 100644 --- a/internal/dao/internal/mdl_tenant_models.go +++ b/internal/dao/internal/mdl_tenant_models.go @@ -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表示跟随模型默认) diff --git a/internal/dispatchadapter/catalog.go b/internal/dispatchadapter/catalog.go index abb9ce7..26386d3 100644 --- a/internal/dispatchadapter/catalog.go +++ b/internal/dispatchadapter/catalog.go @@ -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): diff --git a/internal/dispatchadapter/catalog_test.go b/internal/dispatchadapter/catalog_test.go index dafb0d8..966efd8 100644 --- a/internal/dispatchadapter/catalog_test.go +++ b/internal/dispatchadapter/catalog_test.go @@ -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) + } +} diff --git a/internal/dispatchadapter/maintenance.go b/internal/dispatchadapter/maintenance.go index de59ff1..d4a2fae 100644 --- a/internal/dispatchadapter/maintenance.go +++ b/internal/dispatchadapter/maintenance.go @@ -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 供自动恢复探测识别) diff --git a/internal/dispatchadapter/maintenance_test.go b/internal/dispatchadapter/maintenance_test.go new file mode 100644 index 0000000..f734416 --- /dev/null +++ b/internal/dispatchadapter/maintenance_test.go @@ -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) }) +} diff --git a/internal/dispatchadapter/state_redis.go b/internal/dispatchadapter/state_redis.go index dde4a30..8e6b7fe 100644 --- a/internal/dispatchadapter/state_redis.go +++ b/internal/dispatchadapter/state_redis.go @@ -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{ diff --git a/internal/dispatchadapter/state_redis_test.go b/internal/dispatchadapter/state_redis_test.go index d1c7987..1c4adb3 100644 --- a/internal/dispatchadapter/state_redis_test.go +++ b/internal/dispatchadapter/state_redis_test.go @@ -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) // --------------------------------------------------------------------------- diff --git a/internal/dispatchadapter/wire.go b/internal/dispatchadapter/wire.go index f43c63e..8dd623e 100644 --- a/internal/dispatchadapter/wire.go +++ b/internal/dispatchadapter/wire.go @@ -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 { diff --git a/internal/logic/admin/channel.go b/internal/logic/admin/channel.go index c914d7e..69ab27d 100644 --- a/internal/logic/admin/channel.go +++ b/internal/logic/admin/channel.go @@ -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 } diff --git a/internal/logic/admin/channel_test_request.go b/internal/logic/admin/channel_test_request.go index 06f11d9..576c90f 100644 --- a/internal/logic/admin/channel_test_request.go +++ b/internal/logic/admin/channel_test_request.go @@ -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 { diff --git a/internal/logic/admin/channel_testing.go b/internal/logic/admin/channel_testing.go index 3669e2a..b0b0550 100644 --- a/internal/logic/admin/channel_testing.go +++ b/internal/logic/admin/channel_testing.go @@ -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) } diff --git a/internal/logic/billing/sweeper.go b/internal/logic/billing/sweeper.go index c8b360f..5708fb9 100644 --- a/internal/logic/billing/sweeper.go +++ b/internal/logic/billing/sweeper.go @@ -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)", diff --git a/internal/logic/common/config.go b/internal/logic/common/config.go index e648ed2..3c82cc7 100644 --- a/internal/logic/common/config.go +++ b/internal/logic/common/config.go @@ -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 // ────────────────────────────────────────── diff --git a/internal/logic/common/cron.go b/internal/logic/common/cron.go index acc0d2e..8ca7098 100644 --- a/internal/logic/common/cron.go +++ b/internal/logic/common/cron.go @@ -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 diff --git a/internal/logic/common/cron_test.go b/internal/logic/common/cron_test.go new file mode 100644 index 0000000..ae5ff0e --- /dev/null +++ b/internal/logic/common/cron_test.go @@ -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 不得阻断后续回调") +} diff --git a/internal/logic/common/settings_registry.go b/internal/logic/common/settings_registry.go index 67accf2..128f1bc 100644 --- a/internal/logic/common/settings_registry.go +++ b/internal/logic/common/settings_registry.go @@ -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", diff --git a/internal/logic/monitor/api_metrics.go b/internal/logic/monitor/api_metrics.go index 72dadd6..6759a47 100644 --- a/internal/logic/monitor/api_metrics.go +++ b/internal/logic/monitor/api_metrics.go @@ -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 diff --git a/internal/logic/relay/provider.go b/internal/logic/relay/provider.go index f9fc2c7..0661b5a 100644 --- a/internal/logic/relay/provider.go +++ b/internal/logic/relay/provider.go @@ -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 } diff --git a/internal/logic/task/channel_auto_test_cron.go b/internal/logic/task/channel_auto_test_cron.go index c30a8cb..4138afe 100644 --- a/internal/logic/task/channel_auto_test_cron.go +++ b/internal/logic/task/channel_auto_test_cron.go @@ -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 } diff --git a/internal/logic/task/health_snapshot.go b/internal/logic/task/health_snapshot.go index 46e4c54..b4aa842 100644 --- a/internal/logic/task/health_snapshot.go +++ b/internal/logic/task/health_snapshot.go @@ -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) diff --git a/internal/model/do/chn_health_scores.go b/internal/model/do/chn_health_scores.go index 6812339..6247223 100644 --- a/internal/model/do/chn_health_scores.go +++ b/internal/model/do/chn_health_scores.go @@ -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 // 更新时间 } diff --git a/internal/model/do/chn_health_snapshots.go b/internal/model/do/chn_health_snapshots.go index aa0ed6d..c832a21 100644 --- a/internal/model/do/chn_health_snapshots.go +++ b/internal/model/do/chn_health_snapshots.go @@ -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 // 快照时间 } diff --git a/internal/model/do/mdl_tenant_models.go b/internal/model/do/mdl_tenant_models.go index e510c84..6064204 100644 --- a/internal/model/do/mdl_tenant_models.go +++ b/internal/model/do/mdl_tenant_models.go @@ -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表示跟随模型默认) diff --git a/internal/model/entity/chn_health_scores.go b/internal/model/entity/chn_health_scores.go index b3afa6f..0b696ca 100644 --- a/internal/model/entity/chn_health_scores.go +++ b/internal/model/entity/chn_health_scores.go @@ -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:"更新时间"` // 更新时间 } diff --git a/internal/model/entity/chn_health_snapshots.go b/internal/model/entity/chn_health_snapshots.go index 5ddd15a..97b9e63 100644 --- a/internal/model/entity/chn_health_snapshots.go +++ b/internal/model/entity/chn_health_snapshots.go @@ -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:"快照时间"` // 快照时间 } diff --git a/internal/model/entity/mdl_tenant_models.go b/internal/model/entity/mdl_tenant_models.go index 4ef62d5..752893a 100644 --- a/internal/model/entity/mdl_tenant_models.go +++ b/internal/model/entity/mdl_tenant_models.go @@ -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}] } diff --git a/migrations/000017_v0_2_14_version.sql b/migrations/000017_v0_2_14_version.sql new file mode 100644 index 0000000..72859c8 --- /dev/null +++ b/migrations/000017_v0_2_14_version.sql @@ -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 折扣等,最终价格 = 基础价格 × 模型倍率 × 租户倍率)'; diff --git a/relaykit/dispatch/types.go b/relaykit/dispatch/types.go index b0ef3d2..a9f769f 100644 --- a/relaykit/dispatch/types.go +++ b/relaykit/dispatch/types.go @@ -159,8 +159,7 @@ type Outcome struct { Success bool Class ErrorClass LatencyMs float64 - // Probe 标记探测类结果(管理后台手动测试 / 自动探测 cron),非真实用户流量。 - // 探测失败只喂熔断窗口不衰减健康 EWMA:无流量渠道若探测持续失败(如 test_model - // 配置错误),周期性 ×decay 会把健康分指数拖垮且无真实流量对冲回升。 + // Probe 标记本结果来自探测请求(管理后台渠道测试 / 自动探测 cron)而非真实流量。 + // 纯核心库不消费,仅供适配层观测日志区分健康分变化来源。 Probe bool } diff --git a/web/admin/src/main.ts b/web/admin/src/main.ts index 4ba0800..0de17d8 100644 --- a/web/admin/src/main.ts +++ b/web/admin/src/main.ts @@ -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) diff --git a/web/admin/src/views/ai/ChannelDetailPage.vue b/web/admin/src/views/ai/ChannelDetailPage.vue index ae1ade5..352ad99 100644 --- a/web/admin/src/views/ai/ChannelDetailPage.vue +++ b/web/admin/src/views/ai/ChannelDetailPage.vue @@ -1069,6 +1069,13 @@ function formatHeaders(headers: Record在线调试 AI 模型。不同大模型参数不一致,此处仅做简单功能演示,验证资源可用。请还用API调用获取最佳体验。
-通过 thinking 参数控制开关,仅对支持的模型生效
+