fix(channel):修复渠道健康分数一直下降的问题
This commit is contained in:
chengqian committed
1 parent
b34c77951c
commit
d2e3794e3a
6 files changed
+87
-11
No files matched your search
@@ -779,3 +779,13 @@ err := g.DB().Transaction(dbCtx, func(txCtx context.Context, tx gdb.TX) error {
|
||||
- 「读-改-写」JSONB 配置的合并更新块，读失败时应中止更新并返回错误，而不是带着默认值继续。
|
||||
|
||||
排查信号：JSONB 配置列中「本次没提交的字段莫名回到默认值」；`grep 'Scan(&' `命中基础类型指针。
|
||||
|
||||
### 2026-08-26（二）：gf redis 驱动 HMGET 缺失字段返回空字符串而非 nil，IsNil() 判缺失失效
|
||||
|
||||
**问题**：渠道健康度一段时间后自动掉到个位数。排查发现健康快照（`runMaintenance` → `ReadRuntime` 平均渠道下所有模型的 succ EWMA）把**无健康键的模型读成 succ=0** 平均了进去；健康键 TTL 24h，无流量模型的键过期后即变成 0，渠道健康度随键过期逐渐坍缩到个位数。
|
||||
|
||||
**原因**：`g.Redis().Do(ctx, "HMGET", key, "f1", "f2")` 对**不存在的 key/字段**，经 gf redis contrib 驱动转换后返回的是 `[]string{"", ""}`（空字符串），`.Vars()` 后每个元素 **`IsNil()==false`**、`Float64()==0`。`ReadRuntime` 用 `!vals[0].IsNil()` 判「有数据才覆盖乐观默认值 1」，判断永远为真 → 默认值被 0 覆盖。调度打分侧 `healthFactor` 恰好有 `succ<=0 && lat==0 视为满分` 的兜底所以转发不受影响，但快照聚合没有兜底，展示值被拖垮。`readBreaker` 同模式（空串 state 恰好 Int()==0==CLOSED 侥幸正确）。
|
||||
|
||||
**修复**：`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`），两者行为不一致，混用时尤其注意。
|
||||
@@ -313,6 +313,8 @@ 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()
|
||||
@@ -323,10 +325,12 @@ func (s *RedisState) processOutcome(ctx context.Context, o dispatch.Outcome) {
|
||||
if o.Success {
|
||||
success = "1"
|
||||
}
|
||||
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
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
// 429 → softLimit 自动估计器（基线方案 §8.2）
|
||||
@@ -588,10 +592,14 @@ 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 {
|
||||
if !vals[0].IsNil() {
|
||||
// gf redis 驱动对 HMGET 缺失字段返回空字符串而非 nil（IsNil()==false），
|
||||
// 必须按无数据处理保留乐观默认值：否则无流量/键过期（TTL 24h）的模型
|
||||
// 读成 succ=0，健康快照取平均被拖到个位数（调度打分有 succ<=0 兜底，
|
||||
// 快照聚合没有）。
|
||||
if str := vals[0].String(); str != "" {
|
||||
out.SuccEwma = vals[0].Float64()
|
||||
}
|
||||
if !vals[1].IsNil() {
|
||||
if str := vals[1].String(); str != "" {
|
||||
out.LatEwmaMs = vals[1].Float64()
|
||||
}
|
||||
}
|
||||
@@ -614,7 +622,8 @@ func (s *RedisState) readBreaker(ctx context.Context, key string, nowMs int64) (
|
||||
return dispatch.BreakerClosed, 0, 0
|
||||
}
|
||||
vals := v.Vars()
|
||||
if len(vals) != 5 || vals[0].IsNil() {
|
||||
// 缺失字段经 gf 驱动转换后为空字符串（IsNil()==false），与 nil 同判为无快照
|
||||
if len(vals) != 5 || vals[0].IsNil() || vals[0].String() == "" {
|
||||
return dispatch.BreakerClosed, 0, 0
|
||||
}
|
||||
snap := dispatch.BreakerSnapshot{
|
||||
|
||||
@@ -104,6 +104,55 @@ func Test健康衰减分档(t *testing.T) {
|
||||
assert.InEpsilon(t, 800, rt.LatEwmaMs, 1e-9)
|
||||
}
|
||||
|
||||
// Test探测失败不衰减健康EWMA 探测类（Probe）失败只喂熔断窗口，健康分不变——
|
||||
// 防止每 5 分钟的自动探测持续失败把无流量渠道健康分指数拖垮（0.93^N → 个位数）；
|
||||
// 探测成功照常回升健康并记录延迟。
|
||||
func Test探测失败不衰减健康EWMA(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newTestState(t, func(p *dispatch.RoutingPolicy) {
|
||||
p.Breaker.FailThreshold = 1000 // 阈值调高，只验证窗口计数不触发熔断
|
||||
p.Breaker.ModelFailThreshold = 1000
|
||||
})
|
||||
|
||||
// 连续探测失败：健康 EWMA 保持满分，熔断窗口照常计数
|
||||
for range 5 {
|
||||
s.processOutcome(ctx, dispatch.Outcome{
|
||||
ChannelID: 9, Model: "m", Success: false,
|
||||
Class: dispatch.ErrClassTransient, Probe: true,
|
||||
})
|
||||
}
|
||||
rt := s.ReadRuntime(ctx, 9, "m")
|
||||
assert.InEpsilon(t, 1.0, rt.SuccEwma, 1e-9, "探测失败不得衰减健康分")
|
||||
|
||||
v, err := g.Redis().Do(ctx, "HGET", keyBreaker+"9", "fail_count")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 5, v.Int(), "探测失败仍喂熔断窗口")
|
||||
|
||||
// 真实流量失败拉低健康分后，探测成功照常回升
|
||||
s.processOutcome(ctx, dispatch.Outcome{ChannelID: 9, Model: "m", Success: false, Class: dispatch.ErrClassTransient})
|
||||
rt = s.ReadRuntime(ctx, 9, "m")
|
||||
require.InEpsilon(t, 0.93, rt.SuccEwma, 1e-9)
|
||||
|
||||
s.processOutcome(ctx, dispatch.Outcome{ChannelID: 9, Model: "m", Success: true, LatencyMs: 500, Probe: true})
|
||||
rt = s.ReadRuntime(ctx, 9, "m")
|
||||
assert.InEpsilon(t, 0.93*0.9+0.1, rt.SuccEwma, 1e-9, "探测成功照常回升健康分")
|
||||
assert.InEpsilon(t, 500, rt.LatEwmaMs, 1e-9)
|
||||
}
|
||||
|
||||
// Test缺失健康键读默认值 gf 驱动对 HMGET 缺失字段返回空字符串而非 nil（IsNil()==false）。
|
||||
// ReadRuntime 必须按无数据处理保留乐观默认（succ=1）：无流量模型或 TTL（24h）过期键
|
||||
// 若读成 succ=0，健康快照取平均会把渠道健康度拖到个位数。
|
||||
func Test缺失健康键读默认值(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newTestState(t, nil)
|
||||
|
||||
rt := s.ReadRuntime(ctx, 999, "no-traffic-model")
|
||||
assert.InEpsilon(t, 1.0, rt.SuccEwma, 1e-9, "无健康数据必须默认满分")
|
||||
assert.Zero(t, rt.LatEwmaMs)
|
||||
assert.Equal(t, dispatch.BreakerClosed, rt.Breaker)
|
||||
assert.Equal(t, dispatch.BreakerClosed, rt.ModelBreaker)
|
||||
}
|
||||
|
||||
func Test限流不喂熔断窗口(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newTestState(t, nil)
|
||||
|
||||
@@ -71,13 +71,15 @@ func RefreshDispatchLease(ctx context.Context, channelID int64, requestID string
|
||||
}
|
||||
|
||||
// ReportProbeOutcome 渠道测试（管理后台手动测试 / 自动探测 cron）结果喂给调度健康体系。
|
||||
// 探测失败按 TRANSIENT 轻罚（0.93 档），成功正常回升并可关闭 HALF_OPEN 熔断。
|
||||
// model 必须传平台模型名（chn_abilities.model_name），与调度健康键一致，不要传上游名。
|
||||
// 探测失败只喂熔断窗口不衰减健康分（防周期性探测把无流量渠道健康分拖垮）；
|
||||
// 探测成功正常回升健康并可关闭 HALF_OPEN 熔断。
|
||||
func ReportProbeOutcome(ctx context.Context, channelID int64, model string, success bool, latencyMs float64) {
|
||||
Coordinator(ctx) // 确保已组装（含健康上报 worker）
|
||||
if redisState == nil {
|
||||
return
|
||||
}
|
||||
o := dispatch.Outcome{ChannelID: channelID, Model: model, Success: success, LatencyMs: latencyMs}
|
||||
o := dispatch.Outcome{ChannelID: channelID, Model: model, Success: success, LatencyMs: latencyMs, Probe: true}
|
||||
if !success {
|
||||
o.Class = dispatch.ErrClassTransient
|
||||
}
|
||||
|
||||
@@ -112,11 +112,13 @@ func (s *sAdmin) TestChannel(ctx context.Context, req *v1.ChannelTestReq) (*v1.C
|
||||
result := sendTestRequest(ctx, ch.Type, ch.BaseURL, apiKey, upstreamModel, useProxy)
|
||||
latencyMs := time.Since(startTime).Milliseconds()
|
||||
|
||||
// 更新健康度（喂给调度引擎的健康体系；探测失败按瞬时错误轻罚）。
|
||||
// 更新健康度（喂给调度引擎的健康体系；探测失败只喂熔断窗口不衰减健康分）。
|
||||
// 必须上报平台模型名 testModel：调度健康键与目录聚合均按 chn_abilities.model_name
|
||||
// 组织，上报上游名会写进调度器永远不读的孤立 key，探测结果对健康度完全无效。
|
||||
// 仅 active 渠道上报：禁用/测试中渠道不在调度目录，喂健康分只会无意义拉低展示值；
|
||||
// 渠道启用时 MarkChannelRecovered 会复位熔断，无需在禁用期间预热。
|
||||
if ch.Status == "active" {
|
||||
dispatchadapter.ReportProbeOutcome(ctx, channelID, upstreamModel, result.Success, float64(latencyMs))
|
||||
dispatchadapter.ReportProbeOutcome(ctx, channelID, testModel, result.Success, float64(latencyMs))
|
||||
}
|
||||
|
||||
// 记录测试结果日志
|
||||
|
||||
@@ -159,4 +159,8 @@ type Outcome struct {
|
||||
Success bool
|
||||
Class ErrorClass
|
||||
LatencyMs float64
|
||||
// Probe 标记探测类结果（管理后台手动测试 / 自动探测 cron），非真实用户流量。
|
||||
// 探测失败只喂熔断窗口不衰减健康 EWMA：无流量渠道若探测持续失败（如 test_model
|
||||
// 配置错误），周期性 ×decay 会把健康分指数拖垮且无真实流量对冲回升。
|
||||
Probe bool
|
||||
}
|
||||
Reference in new issue
Block a user