fix(dispatch): 修复 HMGET 判空失效导致冷渠道健康分被读成 0
ReadRuntime 用 `!vals[0].IsNil()` 判断 HMGET 结果是否存在，但 gredis 把 Redis nil 数组元素包装成非 nil 空 gvar（IsNil() 恒为 false、Float64() 静默返回 0）， 导致所有从未被真实流量访问过的渠道×模型被读成健康 0 分而非默认满分 1.0：管理 后台渠道健康度全线坍缩（如某渠道 (1.0+0)/2 → 0.5²×100=25 长期不动），调度侧 healthFactor 同样归零，冷模型路由权重被压死。 GET/HGET 单值回复的 IsNil() 正常，仅数组回复有此坑；仓库内 HMGET 调用点仅 ReadRuntime 与 readBreaker 两处，后者因默认值 BreakerClosed 恰为零值而侥幸 无害，一并改为按字符串判空。 - ReadRuntime：TrimSpace 判空 + ParseFloat 值域校验（succ 取 0<f<=1，EWMA 由正数衰减而来不会精确到达 0，精确 0 视为异常数据） - 回归测试：key 不存在须回落默认值、空串/非法值/精确 0 防御 - 通用规则写入 goframe-conventions.md 框架使用错误记录 同时补齐探测与健康分的可观测性（本次排查绕行三次的直接原因是全链路静默）： - [ChannelProbe] 探测全流程：轮次起止、模型与上游映射、结果、是否上报健康 - [ChannelHealth] 健康分变化：EWMA 变化点（Outcome.Probe 区分探测/真实流量 来源）、快照落盘含各模型 succ 明细、维护轮次覆盖摘要 - 快照落库失败由 Debug 升为 Warning（此前静默，健康分冻结无从定位） - 探测目标不在调度目录时告警（该场景健康分不落盘且渠道不参与调度） - 变化类日志按「分数实际移动」触发，稳态降为 Debug，避免生产流量刷屏
This commit is contained in:
chengqian authored and qian committed
1 parent
08c885aa6c
commit
3035fa118c
9 files changed
+240
-37
No files matched your search
@@ -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 读取路径。
|
||||
@@ -2,7 +2,9 @@ package dispatchadapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
@@ -11,6 +13,7 @@ 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）：
|
||||
@@ -53,18 +56,23 @@ 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))
|
||||
for channelID, models := range channelModels {
|
||||
if len(models) == 0 {
|
||||
continue
|
||||
}
|
||||
covered = append(covered, channelID)
|
||||
|
||||
// 按渠道聚合各模型的健康读值
|
||||
var succSum, latSum float64
|
||||
var channelRt RuntimeReadout
|
||||
succParts := make([]string, 0, len(models))
|
||||
for i, model := range models {
|
||||
rt := redisState.ReadRuntime(ctx, channelID, model)
|
||||
succSum += rt.SuccEwma
|
||||
latSum += rt.LatEwmaMs
|
||||
succParts = append(succParts, fmt.Sprintf("%s=%.3f", model, rt.SuccEwma))
|
||||
if i == 0 {
|
||||
channelRt = rt // 渠道级熔断状态各模型读值相同，取第一个
|
||||
}
|
||||
@@ -72,7 +80,7 @@ func runMaintenance(ctx context.Context) {
|
||||
avgSucc := succSum / float64(len(models))
|
||||
avgLat := latSum / float64(len(models))
|
||||
|
||||
snapshotHealthScore(ctx, channelID, avgSucc, avgLat, pol.Health.Alpha)
|
||||
snapshotHealthScore(ctx, channelID, avgSucc, avgLat, pol.Health.Alpha, strings.Join(succParts, " "))
|
||||
|
||||
// 自动禁用：本轮故障期持续超过阈值（first_opened_ms 只在故障期存在，
|
||||
// 探测失败不重置——探测成功即被清除，见 luaBreakerFail/luaBreakerSuccess）
|
||||
@@ -81,13 +89,29 @@ func runMaintenance(ctx context.Context) {
|
||||
autoDisableChannel(ctx, channelID, now-channelRt.FirstOpenedMs)
|
||||
}
|
||||
}
|
||||
|
||||
// 覆盖摘要：目录只装载 active × 有启用能力行的渠道，未覆盖渠道的健康分不会落盘
|
||||
// （探测却正常的渠道长期停在历史分数时，先看这里是否被跳过）
|
||||
g.Log().Infof(ctx, "[ChannelHealth] 健康快照维护轮次完成: 覆盖渠道 %d 个 %v", len(covered), covered)
|
||||
}
|
||||
|
||||
// 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 +122,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 +135,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 供自动恢复探测识别）
|
||||
|
||||
@@ -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,11 +328,19 @@ 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
|
||||
v, err := g.Redis().Do(ctx, "EVAL", luaHealthObserve, 1, healthKey,
|
||||
success, o.LatencyMs, healthDecayFor(o.Class), now, stateKeyTTLMs)
|
||||
if err != nil {
|
||||
s.local.observe(o) // 降级：实例本地健康镜像
|
||||
return
|
||||
}
|
||||
if vals := v.Strings(); len(vals) == 4 {
|
||||
oldSucc, e1 := strconv.ParseFloat(vals[0], 64)
|
||||
newSucc, e2 := strconv.ParseFloat(vals[1], 64)
|
||||
oldLat, e3 := strconv.ParseFloat(vals[2], 64)
|
||||
newLat, e4 := strconv.ParseFloat(vals[3], 64)
|
||||
if e1 == nil && e2 == nil && e3 == nil && e4 == nil {
|
||||
logHealthChange(ctx, o, pol, oldSucc, newSucc, oldLat, newLat)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -368,6 +379,34 @@ func (s *RedisState) processOutcome(ctx context.Context, o dispatch.Outcome) {
|
||||
}
|
||||
}
|
||||
|
||||
// logHealthChange 健康读值变化日志：按调度同源公式（succ^alpha 百分制，与 healthFactor
|
||||
// 及 chn_health_scores 落盘一致）折算健康分，**仅在健康分实际移动时**记 Info——稳态成功流量
|
||||
// 的 succ 恒为 1.0（只有延迟 EWMA 在抖），若每次上报都打印会在生产流量下淹没日志。
|
||||
// 健康分持平时降为 Debug，保留延迟变化的排查能力。
|
||||
func logHealthChange(ctx context.Context, o dispatch.Outcome, pol *dispatch.RoutingPolicy, oldSucc, newSucc, oldLat, newLat float64) {
|
||||
if oldSucc == newSucc && oldLat == newLat {
|
||||
return
|
||||
}
|
||||
src := "真实流量"
|
||||
if o.Probe {
|
||||
src = "探测"
|
||||
}
|
||||
result := "成功"
|
||||
if !o.Success {
|
||||
result = "失败(" + o.Class.String() + ")"
|
||||
}
|
||||
oldScore := math.Pow(oldSucc, pol.Health.Alpha) * 100
|
||||
newScore := math.Pow(newSucc, pol.Health.Alpha) * 100
|
||||
|
||||
if math.Abs(newScore-oldScore) < 0.05 {
|
||||
g.Log().Debugf(ctx, "[ChannelHealth] 渠道 %d 模型 %s 健康分持平 %.1f | lat_ewma %.1fms → %.1fms | 来源: %s %s",
|
||||
o.ChannelID, o.Model, newScore, oldLat, newLat, src, result)
|
||||
return
|
||||
}
|
||||
g.Log().Infof(ctx, "[ChannelHealth] 渠道 %d 模型 %s 健康分变化: %.1f → %.1f | succ_ewma %.4f → %.4f | lat_ewma %.1fms → %.1fms | 来源: %s %s",
|
||||
o.ChannelID, o.Model, oldScore, newScore, oldSucc, newSucc, oldLat, newLat, src, result)
|
||||
}
|
||||
|
||||
// luaBreakerManualReset 手动恢复：熔断复位为 CLOSED 并记录 recovered_ms（触发爬坡因子）。
|
||||
// ARGV: nowMs, keyTtlMs
|
||||
const luaBreakerManualReset = `
|
||||
@@ -423,6 +462,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 +607,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
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 运行时读值（供目录快照合并健康/熔断/负载）
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -592,15 +648,20 @@ func (s *RedisState) ReadRuntime(ctx context.Context, channelID int64, model str
|
||||
if v, err := g.Redis().Do(ctx, "HMGET", keyHealth+chStr+":"+model, "succ_ewma", "lat_ewma"); err == nil {
|
||||
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
|
||||
}
|
||||
}
|
||||
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 +683,9 @@ func (s *RedisState) readBreaker(ctx context.Context, key string, nowMs int64) (
|
||||
return dispatch.BreakerClosed, 0, 0
|
||||
}
|
||||
vals := v.Vars()
|
||||
// 缺失字段经 gf 驱动转换后为空字符串（IsNil()==false），与 nil 同判为无快照
|
||||
if len(vals) != 5 || vals[0].IsNil() || vals[0].String() == "" {
|
||||
// 判空必须用 IsEmpty()：HMGET 缺失字段的 gvar IsNil() 恒为 false（见 ReadRuntime 注释）。
|
||||
// 此处即使漏判也侥幸无害（默认 BreakerClosed 恰为零值 0），但保持与健康读取一致的正确写法。
|
||||
if len(vals) != 5 || vals[0].IsEmpty() {
|
||||
return dispatch.BreakerClosed, 0, 0
|
||||
}
|
||||
snap := dispatch.BreakerSnapshot{
|
||||
|
||||
@@ -486,6 +486,53 @@ 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)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 本地降级（修订 R4 + 基线方案 §13）
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -3,6 +3,7 @@ package dispatchadapter
|
||||
import (
|
||||
"context"
|
||||
"math/rand"
|
||||
"slices"
|
||||
"sync"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
@@ -86,6 +87,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 {
|
||||
|
||||
@@ -56,11 +56,11 @@ func executeTest(ctx context.Context, provider, method, reqURL string, headers m
|
||||
maskedHdrs := maskHeaders(headers)
|
||||
bodyStr := string(body)
|
||||
|
||||
g.Log().Infof(ctx, "[ChannelTest] %s | 发送请求 | %s %s | 代理: %v", provider, method, reqURL, useProxy)
|
||||
g.Log().Infof(ctx, "[ChannelProbe]%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)
|
||||
g.Log().Warningf(ctx, "[ChannelProbe]%s | 请求失败 | %s | 错误: %v", provider, reqURL, err)
|
||||
return testResult{
|
||||
Error: fmt.Sprintf("请求失败: %v", err),
|
||||
Request: buildReqDetail(method, reqURL, maskedHdrs, bodyStr),
|
||||
@@ -72,7 +72,7 @@ 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))
|
||||
g.Log().Warningf(ctx, "[ChannelProbe]%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 +81,7 @@ 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))
|
||||
g.Log().Infof(ctx, "[ChannelProbe]%s | 测试成功 | HTTP %d | 响应: %s", provider, resp.StatusCode, truncateStr(respStr, 200))
|
||||
return testResult{
|
||||
Success: true,
|
||||
Response: truncateStr(respStr, 500),
|
||||
|
||||
@@ -108,6 +108,8 @@ func (s *sAdmin) TestChannel(ctx context.Context, req *v1.ChannelTestReq) (*v1.C
|
||||
}
|
||||
|
||||
// 发送最小测试请求
|
||||
g.Log().Infof(ctx, "[ChannelProbe] 渠道 %s (%d) 开始探测 | 模型: %s (上游: %s) | 渠道状态: %s | 代理: %v",
|
||||
ch.Name, channelID, testModel, upstreamModel, ch.Status, useProxy)
|
||||
startTime := time.Now()
|
||||
result := sendTestRequest(ctx, ch.Type, ch.BaseURL, apiKey, upstreamModel, useProxy)
|
||||
latencyMs := time.Since(startTime).Milliseconds()
|
||||
@@ -119,14 +121,26 @@ 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))
|
||||
if dispatchadapter.CatalogHasModel(channelID, testModel) {
|
||||
g.Log().Infof(ctx, "[ChannelProbe] 渠道 %s (%d) 探测结果已投递健康上报 | 模型: %s | success: %v | 延迟: %dms（EWMA 变化将打印 [ChannelHealth] 日志，5 分钟内落盘刷新）",
|
||||
ch.Name, channelID, testModel, result.Success, latencyMs)
|
||||
} else {
|
||||
// 探测目标不在调度目录：EWMA 照常写入 Redis，但调度不读、维护快照不落盘，
|
||||
// 管理后台健康分不会变化——探测看起来"生效了但没效果"的常见根因
|
||||
g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测结果已投递健康上报，但模型 %s 不在调度目录（渠道无启用的模型能力行或该模型未配置能力）| 健康分不会落盘展示，渠道/模型不参与调度，请检查渠道的模型能力配置",
|
||||
ch.Name, channelID, testModel)
|
||||
}
|
||||
} else {
|
||||
g.Log().Infof(ctx, "[ChannelProbe] 渠道 %s (%d) 状态为 %s（非 active），探测结果不上报健康体系",
|
||||
ch.Name, channelID, ch.Status)
|
||||
}
|
||||
|
||||
// 记录测试结果日志
|
||||
if result.Success {
|
||||
g.Log().Infof(ctx, "[ChannelTest] 渠道 %s (%d) 测试成功 | 模型: %s (上游: %s) | 延迟: %dms | 代理: %v",
|
||||
g.Log().Infof(ctx, "[ChannelProbe] 渠道 %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",
|
||||
g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测失败 | 模型: %s (上游: %s) | 延迟: %dms | 代理: %v | 错误: %s",
|
||||
ch.Name, channelID, testModel, upstreamModel, latencyMs, useProxy, result.Error)
|
||||
}
|
||||
|
||||
|
||||
@@ -30,9 +30,14 @@ 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 {
|
||||
g.Log().Infof(ctx, "[ChannelProbe] 自动探测开始: 无待探测渠道（活跃且配置了 test_model 的渠道为 0）")
|
||||
return
|
||||
}
|
||||
g.Log().Infof(ctx, "[ChannelProbe] 自动探测开始: 待探测活跃渠道 %d 个", len(channels))
|
||||
|
||||
successCount := 0
|
||||
failCount := 0
|
||||
@@ -44,7 +49,8 @@ func AutoTestChannels(ctx context.Context) {
|
||||
|
||||
if err != nil {
|
||||
failCount++
|
||||
g.Log().Warningf(ctx, "[Cron] channel %s (%d) test error: %v", ch.Name, ch.ID, err)
|
||||
// 探测请求尚未发出（无可用 Key/渠道不存在等），TestChannel 内部无结果日志，此处记录
|
||||
g.Log().Warningf(ctx, "[ChannelProbe] 渠道 %s (%d) 探测执行出错: %v", ch.Name, ch.ID, err)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -52,9 +58,10 @@ func AutoTestChannels(ctx context.Context) {
|
||||
successCount++
|
||||
} else {
|
||||
failCount++
|
||||
if result.Error != "" {
|
||||
g.Log().Warningf(ctx, "[Cron] channel %s (%d) test failed: %s", ch.Name, ch.ID, result.Error)
|
||||
}
|
||||
}
|
||||
// 成功/失败明细日志由 TestChannel 统一打印（含模型/延迟/错误），此处只做轮次汇总
|
||||
}
|
||||
|
||||
g.Log().Infof(ctx, "[ChannelProbe] 自动探测完成: 共 %d 个渠道 | 成功 %d | 失败 %d",
|
||||
len(channels), successCount, failCount)
|
||||
}
|
||||
@@ -159,8 +159,7 @@ type Outcome struct {
|
||||
Success bool
|
||||
Class ErrorClass
|
||||
LatencyMs float64
|
||||
// Probe 标记探测类结果（管理后台手动测试 / 自动探测 cron），非真实用户流量。
|
||||
// 探测失败只喂熔断窗口不衰减健康 EWMA：无流量渠道若探测持续失败（如 test_model
|
||||
// 配置错误），周期性 ×decay 会把健康分指数拖垮且无真实流量对冲回升。
|
||||
// Probe 标记本结果来自探测请求（管理后台渠道测试 / 自动探测 cron）而非真实流量。
|
||||
// 纯核心库不消费，仅供适配层观测日志区分健康分变化来源。
|
||||
Probe bool
|
||||
}
|
||||
Reference in new issue
Block a user