fix(dispatch): 渠道健康聚合只统计有真实上报的模型，降级时保留库中旧值
渠道展示分是目录内全部模型 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。
This commit is contained in:
chengqian authored and qian committed
1 parent
3f3fc43856
commit
a5e2ecf527
3 files changed
+281
-24
No files matched your search
@@ -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）。
|
||||
|
||||
@@ -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) })
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
Reference in new issue
Block a user