From a5e2ecf5278e2301b919a4fa32ca2976d5eded01 Mon Sep 17 00:00:00 2001 From: chengqian Date: Thu, 27 Aug 2026 17:00:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(dispatch):=20=E6=B8=A0=E9=81=93=E5=81=A5?= =?UTF-8?q?=E5=BA=B7=E8=81=9A=E5=90=88=E5=8F=AA=E7=BB=9F=E8=AE=A1=E6=9C=89?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=E4=B8=8A=E6=8A=A5=E7=9A=84=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=EF=BC=8C=E9=99=8D=E7=BA=A7=E6=97=B6=E4=BF=9D=E7=95=99=E5=BA=93?= =?UTF-8?q?=E4=B8=AD=E6=97=A7=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 渠道展示分是目录内全部模型 succ_ewma 的算术平均，而无流量模型读到的是乐观默认满分。 20 个模型的渠道死掉 1 个（succ→0.05）仍能算出 avgSucc≈0.95、91 分绿灯，恰好掩盖了 模型级熔断要隔离的典型场景。改为只统计 HasHealth 的模型；全渠道无真实上报时 Counted=0，保留库中上一次的分数而不是用默认满分覆盖历史结论。 Redis 读失败（Degraded）时跳过落盘并打 Warning：照常落盘会把全渠道健康分刷成 100， 连带 chn_health_snapshots 把假象写进历史趋势，事故期间的曲线反而全绿。自动禁用判定 一并跳过——该操作不可逆，不能基于不可信读值触发。 聚合逻辑抽成纯函数 aggregateChannelHealth 以便单测；另新增 RequestHealthSnapshot 按需重算入口（维护循环消费，满则丢弃由下一轮兜底），供渠道测试/重置健康度接入。 健康上报是 fire-and-forget，重算前等 500ms 让 worker 落完 Redis。 --- internal/dispatchadapter/maintenance.go | 163 ++++++++++++++++--- internal/dispatchadapter/maintenance_test.go | 127 +++++++++++++++ internal/dispatchadapter/wire.go | 15 ++ 3 files changed, 281 insertions(+), 24 deletions(-) create mode 100644 internal/dispatchadapter/maintenance_test.go diff --git a/internal/dispatchadapter/maintenance.go b/internal/dispatchadapter/maintenance.go index b42dd84..d4a2fae 100644 --- a/internal/dispatchadapter/maintenance.go +++ b/internal/dispatchadapter/maintenance.go @@ -17,14 +17,37 @@ import ( ) // 维护任务（阶段 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() { @@ -34,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 } @@ -41,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() { @@ -58,41 +110,104 @@ func runMaintenance(ctx context.Context) { 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 } + + agg := aggregateChannelHealth(models, func(model string) RuntimeReadout { + return redisState.ReadRuntime(ctx, channelID, model) + }) + + // Redis 读失败：本轮读值不可信，跳过落盘与自动禁用判定。 + // 若照常落盘，乐观默认值会把全渠道健康分刷成 100，事故期间的历史趋势反而全绿。 + if agg.Degraded { + skippedDegraded = append(skippedDegraded, channelID) + 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 // 渠道级熔断状态各模型读值相同，取第一个 - } + // 全部模型都没有真实上报（新渠道 / 长期无流量 / EWMA key 已过 TTL）： + // 保留库中上一次的分数，不要用默认满分覆盖掉历史结论。 + if agg.Counted == 0 { + skippedNoData = append(skippedNoData, channelID) + } else { + snapshotHealthScore(ctx, channelID, agg.AvgSucc, agg.AvgLat, pol.Health.Alpha, agg.Detail) } - avgSucc := succSum / float64(len(models)) - avgLat := latSum / float64(len(models)) - - snapshotHealthScore(ctx, channelID, avgSucc, avgLat, pol.Health.Alpha, strings.Join(succParts, " ")) // 自动禁用：本轮故障期持续超过阈值（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）。 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/wire.go b/internal/dispatchadapter/wire.go index 524cb13..8dd623e 100644 --- a/internal/dispatchadapter/wire.go +++ b/internal/dispatchadapter/wire.go @@ -63,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) {