42e881576d
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m33s
- 新增网关限流：按本系统运营商账户（卡 carrier_id）→ 限流组 → 操作限制每秒请求数，每次 HTTP 尝试计量，默认限流、仅后台人工点击/通道阈值/自动续费/手动刷新豁免；支持只统计与拦截两种模式，默认只统计 - Asynq 遇限流不计失败、固定短延迟重试；轮询任务与停复机业务重试遇限流短延迟重排 - 停机、复机不再使用网关内部重试，只发一次请求 - 网关返回结构化错误；新增上游错误分类规则表与测试接口 - 新增按月分区的停复机记录表：系统自动停复机只写记录，不再写统一审计与集成日志；普通轮询查询不再写集成日志 - 新增限流组、错误分类、停复机记录、按账户调用统计与枚举后台接口 - 迁移 000243~000246；同步 OpenSpec 提案（含后续 6 个 change）与技术方案、工程约束
80 lines
3.0 KiB
Go
80 lines
3.0 KiB
Go
package gatewaylimit
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha1"
|
||
"encoding/hex"
|
||
"time"
|
||
|
||
"github.com/hibiken/asynq"
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/infrastructure/systemconfig"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// Runtime 汇总一个进程内的网关限流组件，API 与 Worker 各自创建一份，限流窗口经 Redis 跨进程共享。
|
||
type Runtime struct {
|
||
Settings *SettingsProvider
|
||
Groups *GroupProvider
|
||
Classifier *Classifier
|
||
Limiter *Limiter
|
||
Guard *Guard
|
||
Recorder *StopResumeRecorder
|
||
|
||
redisClient *redis.Client
|
||
logger *zap.Logger
|
||
}
|
||
|
||
// NewRuntime 创建网关限流运行组件；配置读取使用独立注册表，与后台系统配置共享同一份数据库与缓存数据。
|
||
func NewRuntime(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) *Runtime {
|
||
if logger == nil {
|
||
logger = zap.NewNop()
|
||
}
|
||
registry := systemconfig.NewRegistry()
|
||
RegisterConfigDefinitions(registry, logger)
|
||
var cache systemconfig.Cache
|
||
if redisClient != nil {
|
||
cache = systemconfig.NewRedisCache(redisClient)
|
||
}
|
||
reader := systemconfig.NewReader(db, registry, cache, systemconfig.NewLogAlertSink(logger))
|
||
settings := NewSettingsProvider(reader, logger)
|
||
groups := NewGroupProvider(postgres.NewCarrierRateLimitGroupStore(db), logger)
|
||
classifier := NewClassifier(postgres.NewGatewayErrorClassificationStore(db), logger)
|
||
limiter := NewLimiter(redisClient, groups, settings)
|
||
guard := NewGuard(limiter, NewCarrierResolver(db, redisClient), classifier, redisClient, logger)
|
||
recorder := NewStopResumeRecorder(postgres.NewCardStopResumeRecordStore(db), classifier, logger)
|
||
return &Runtime{Settings: settings, Groups: groups, Classifier: classifier, Limiter: limiter, Guard: guard, Recorder: recorder, redisClient: redisClient, logger: logger}
|
||
}
|
||
|
||
// TrackDeferral 记录 Asynq 任务因限流被推迟；同一任务首次被限流起累计推迟超过阈值时写一次错误告警，任务继续重试。
|
||
// 任务以"类型 + 载荷摘要"标识，记录 1 天后自动过期。
|
||
func (r *Runtime) TrackDeferral(task *asynq.Task) {
|
||
if r == nil || r.redisClient == nil || task == nil {
|
||
return
|
||
}
|
||
ctx := context.Background()
|
||
sum := sha1.Sum(task.Payload())
|
||
taskKey := task.Type() + ":" + hex.EncodeToString(sum[:])
|
||
key := constants.RedisRateLimitDeferStartKey(taskKey)
|
||
now := time.Now()
|
||
if ok, err := r.redisClient.SetNX(ctx, key, now.Unix(), 24*time.Hour).Result(); err != nil || ok {
|
||
return
|
||
}
|
||
startedUnix, err := r.redisClient.Get(ctx, key).Int64()
|
||
if err != nil {
|
||
return
|
||
}
|
||
deferred := now.Sub(time.Unix(startedUnix, 0))
|
||
if deferred < r.Settings.Current(ctx).DeferAlert {
|
||
return
|
||
}
|
||
if alerted, _ := r.redisClient.SetNX(ctx, key+":alerted", 1, 24*time.Hour).Result(); alerted {
|
||
r.logger.Error("异步任务因运营商账户网关限流累计推迟超过告警阈值，任务继续重试",
|
||
zap.String("task_type", task.Type()), zap.String("task_key", taskKey), zap.Duration("deferred", deferred))
|
||
}
|
||
}
|