316 lines
12 KiB
Go
316 lines
12 KiB
Go
package dispatchadapter
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"math"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/gtime"
|
||
|
||
"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 一次，另可由
|
||
// 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() {
|
||
ticker := time.NewTicker(maintenanceInterval)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ticker.C:
|
||
runMaintenance(ctx)
|
||
case channelID := <-healthRefresh:
|
||
time.Sleep(healthRefreshDelay)
|
||
refreshChannelHealth(ctx, channelID)
|
||
case <-stop:
|
||
return
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
|
||
// 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() {
|
||
if r := recover(); r != nil {
|
||
g.Log().Warningf(ctx, "[Dispatch] 维护任务 panic: %v", r)
|
||
}
|
||
}()
|
||
if catalog == nil || redisState == nil {
|
||
return
|
||
}
|
||
|
||
pol := coordinator.Policy()
|
||
autoDisable := lcommon.Config().GetBool(ctx, "channel_auto_disable_enabled")
|
||
now := time.Now().UnixMilli()
|
||
|
||
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)
|
||
|
||
// 全部模型都没有真实上报（新渠道 / 长期无流量 / 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 && agg.ChannelRt.FirstOpenedMs > 0 &&
|
||
now-agg.ChannelRt.FirstOpenedMs >= int64(pol.Breaker.AutoDisableAfterSeconds)*1000 {
|
||
autoDisableChannel(ctx, channelID, now-agg.ChannelRt.FirstOpenedMs)
|
||
}
|
||
}
|
||
|
||
// 覆盖摘要：目录只装载 active × 有启用能力行的渠道，未覆盖渠道的健康分不会落盘
|
||
// （探测却正常的渠道长期停在历史分数时，先看这里是否被跳过）
|
||
g.Log().Debugf(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().Debugf(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 同源。
|
||
// 每轮固定输出一行明细（计算值 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{
|
||
SuccessRate: succ * 100,
|
||
LatencyMs: latMs,
|
||
HealthScore: healthScore,
|
||
CalculatedAt: gtime.Now(),
|
||
}).
|
||
UpdateAndGetAffected()
|
||
if err != nil {
|
||
// 写入失败必须可见（此前 Debug 级静默，健康分冻结难以定位）
|
||
g.Log().Warningf(ctx, "[ChannelHealth] 渠道 %d 健康快照更新失败: %v", channelID, err)
|
||
return
|
||
}
|
||
if affected == 0 {
|
||
_, err = dao.ChnHealthScores.Ctx(ctx).Data(do.ChnHealthScores{
|
||
ChannelId: channelID,
|
||
SuccessRate: succ * 100,
|
||
LatencyMs: latMs,
|
||
HealthScore: healthScore,
|
||
CalculatedAt: gtime.Now(),
|
||
}).Insert()
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "[ChannelHealth] 渠道 %d 健康快照插入失败: %v", channelID, err)
|
||
return
|
||
}
|
||
}
|
||
|
||
// 分数有实际变化才记 Info；稳态渠道降为 Debug（避免每 5 分钟 × 渠道数的固定噪音）。
|
||
// 两种情况都带各模型 succ 明细：排查"健康分不符合预期"时可直接定位是哪个模型拉低了均值。
|
||
if math.Abs(healthScore-prevScore) >= 0.05 {
|
||
g.Log().Debugf(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 供自动恢复探测识别）
|
||
// + 清绑定 + 跨实例目录失效 + 全租户通知（沿用 channel_auto_disabled 通知模板）。
|
||
func autoDisableChannel(ctx context.Context, channelID int64, openDurationMs int64) {
|
||
affected, err := dao.ChnChannels.Ctx(ctx).
|
||
Where("id", channelID).
|
||
Where("status", "active").
|
||
Data(do.ChnChannels{Status: "disabled", AutoDisabled: 1}).
|
||
UpdateAndGetAffected()
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "[Dispatch] 自动禁用渠道失败: channel=%d err=%v", channelID, err)
|
||
return
|
||
}
|
||
if affected == 0 {
|
||
return // 已被禁用或删除
|
||
}
|
||
g.Log().Warningf(ctx, "[Dispatch] 渠道 %d 熔断持续 %ds 未恢复，已自动禁用（channel_auto_disable_enabled）",
|
||
channelID, openDurationMs/1000)
|
||
InvalidateChannel(ctx, channelID)
|
||
|
||
// 全租户通知（异步，失败仅记日志）
|
||
var chName *string
|
||
_ = dao.ChnChannels.Ctx(ctx).Where("id", channelID).Fields("name").Scan(&chName)
|
||
channelName := ""
|
||
if chName != nil {
|
||
channelName = *chName
|
||
}
|
||
go func() {
|
||
bgCtx := context.Background()
|
||
engine := lcommon.NewNotificationEngine()
|
||
if err := engine.SendToAllTenants(bgCtx, "channel_auto_disabled", g.Map{
|
||
"channel_name": channelName,
|
||
"threshold": 0, // 新语义：按熔断持续时长而非连续失败次数
|
||
}, ""); err != nil {
|
||
g.Log().Errorf(bgCtx, "[Dispatch] 自动禁用通知发送失败: %v", err)
|
||
}
|
||
}()
|
||
}
|
||
|
||
func clampFloat(v, lo, hi float64) float64 {
|
||
if v < lo {
|
||
return lo
|
||
}
|
||
if v > hi {
|
||
return hi
|
||
}
|
||
return v
|
||
}
|