All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m33s
- 新增网关限流:按本系统运营商账户(卡 carrier_id)→ 限流组 → 操作限制每秒请求数,每次 HTTP 尝试计量,默认限流、仅后台人工点击/通道阈值/自动续费/手动刷新豁免;支持只统计与拦截两种模式,默认只统计 - Asynq 遇限流不计失败、固定短延迟重试;轮询任务与停复机业务重试遇限流短延迟重排 - 停机、复机不再使用网关内部重试,只发一次请求 - 网关返回结构化错误;新增上游错误分类规则表与测试接口 - 新增按月分区的停复机记录表:系统自动停复机只写记录,不再写统一审计与集成日志;普通轮询查询不再写集成日志 - 新增限流组、错误分类、停复机记录、按账户调用统计与枚举后台接口 - 迁移 000243~000246;同步 OpenSpec 提案(含后续 6 个 change)与技术方案、工程约束
140 lines
4.1 KiB
Go
140 lines
4.1 KiB
Go
package gatewaylimit
|
||
|
||
import (
|
||
"context"
|
||
"regexp"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// ClassificationLoader 读取全部启用的错误分类规则(按优先级排序)。
|
||
type ClassificationLoader interface {
|
||
ListEnabled(ctx context.Context) ([]model.GatewayErrorClassification, error)
|
||
}
|
||
|
||
// Classification 是一次上游错误的分类结果。
|
||
type Classification struct {
|
||
Category string // 见 constants.GatewayClass*
|
||
RuleID uint // 命中的规则 ID,未命中为 0
|
||
Keyword string // 命中的关键字
|
||
MaybeAccepted bool // 上游可能已受理
|
||
}
|
||
|
||
type compiledRule struct {
|
||
row model.GatewayErrorClassification
|
||
ops map[string]bool
|
||
regex *regexp.Regexp
|
||
}
|
||
|
||
// Classifier 按后台维护的关键字规则对上游错误分类;规则在进程内缓存,超过 TTL 重新加载。
|
||
type Classifier struct {
|
||
loader ClassificationLoader
|
||
logger *zap.Logger
|
||
mu sync.Mutex
|
||
rules []compiledRule
|
||
loaded time.Time
|
||
}
|
||
|
||
// NewClassifier 创建错误分类器;loader 为 nil 时一律归为业务错误。
|
||
func NewClassifier(loader ClassificationLoader, logger *zap.Logger) *Classifier {
|
||
if logger == nil {
|
||
logger = zap.NewNop()
|
||
}
|
||
return &Classifier{loader: loader, logger: logger}
|
||
}
|
||
|
||
// Invalidate 使规则缓存立即失效,后台保存规则后调用。
|
||
func (c *Classifier) Invalidate() {
|
||
if c == nil {
|
||
return
|
||
}
|
||
c.mu.Lock()
|
||
c.loaded = time.Time{}
|
||
c.mu.Unlock()
|
||
}
|
||
|
||
// Classify 对某操作的上游返回信息分类;未命中任何启用规则时归为业务错误。
|
||
func (c *Classifier) Classify(ctx context.Context, op, message string) Classification {
|
||
if c == nil {
|
||
return Classification{Category: constants.GatewayClassBusiness}
|
||
}
|
||
return MatchRules(c.current(ctx), op, message)
|
||
}
|
||
|
||
// ClassifyWith 用给定规则列表分类,供测试接口在不写数据的前提下验证规则。
|
||
func ClassifyWith(rows []model.GatewayErrorClassification, op, message string) Classification {
|
||
return MatchRules(compileRules(rows, zap.NewNop()), op, message)
|
||
}
|
||
|
||
// MatchRules 按优先级顺序返回第一个命中的规则分类。
|
||
func MatchRules(rules []compiledRule, op, message string) Classification {
|
||
for _, rule := range rules {
|
||
if len(rule.ops) > 0 && !rule.ops[op] {
|
||
continue
|
||
}
|
||
if !ruleMatches(rule, message) {
|
||
continue
|
||
}
|
||
return Classification{Category: rule.row.Category, RuleID: rule.row.ID, Keyword: rule.row.Keyword, MaybeAccepted: rule.row.MaybeAccepted}
|
||
}
|
||
return Classification{Category: constants.GatewayClassBusiness}
|
||
}
|
||
|
||
func ruleMatches(rule compiledRule, message string) bool {
|
||
switch rule.row.MatchType {
|
||
case constants.GatewayMatchPrefix:
|
||
return strings.HasPrefix(message, rule.row.Keyword)
|
||
case constants.GatewayMatchRegex:
|
||
return rule.regex != nil && rule.regex.MatchString(message)
|
||
default:
|
||
return strings.Contains(message, rule.row.Keyword)
|
||
}
|
||
}
|
||
|
||
func (c *Classifier) current(ctx context.Context) []compiledRule {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
if c.loader == nil || time.Since(c.loaded) < constants.GatewayRateLimitConfigTTL {
|
||
return c.rules
|
||
}
|
||
rows, err := c.loader.ListEnabled(ctx)
|
||
c.loaded = time.Now()
|
||
if err != nil {
|
||
c.logger.Warn("加载上游错误分类规则失败,沿用上一次规则", zap.Error(err))
|
||
return c.rules
|
||
}
|
||
c.rules = compileRules(rows, c.logger)
|
||
return c.rules
|
||
}
|
||
|
||
func compileRules(rows []model.GatewayErrorClassification, logger *zap.Logger) []compiledRule {
|
||
rules := make([]compiledRule, 0, len(rows))
|
||
for _, row := range rows {
|
||
rule := compiledRule{row: row, ops: map[string]bool{}}
|
||
ops, err := DecodeOperations(row.Operations)
|
||
if err != nil {
|
||
logger.Warn("错误分类规则适用操作非法,已跳过", zap.Uint("rule_id", row.ID))
|
||
continue
|
||
}
|
||
for _, op := range ops {
|
||
rule.ops[op] = true
|
||
}
|
||
if row.MatchType == constants.GatewayMatchRegex {
|
||
re, err := regexp.Compile(row.Keyword)
|
||
if err != nil {
|
||
logger.Warn("错误分类规则正则非法,已跳过", zap.Uint("rule_id", row.ID))
|
||
continue
|
||
}
|
||
rule.regex = re
|
||
}
|
||
rules = append(rules, rule)
|
||
}
|
||
return rules
|
||
}
|