94cbf2adab
- 网关客户端新增原始数据收集器，记录最后一次尝试的明文请求、HTTP 状态、响应体与 code/msg/trace_id - tb_card_traffic_collection 只写成功读数，新增 carrier_id 与原始数据列；新增 tb_card_traffic_collection_failure，失败不判断有无套餐并带错误详情 - 采集记录在线保留当月加最近 2 个整月，到期分区摘出、导出 csv.gz、校验行数与 SHA-256 后删除，本地保留 3 个月 - 新增采集成功/失败、卡日流量、套餐日流量详单、冲正流水、退款后处理尝试列表接口：可查单个也可查全部，分页与条件筛选 - 停复机记录支持资产标识、场景、原因、指令号筛选；单卡历史支持动作、结果、来源筛选 - 单卡采集记录支持筛选与合并失败记录；套餐日流量详单支持分页；退款列表支持后处理状态筛选 - 测试环境 api、worker 日志分目录，保留 40 个压缩备份；worker 挂载 archive 目录 - 迁移 000272、000273 需维护者在测试库执行后再部署 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
397 lines
16 KiB
Go
397 lines
16 KiB
Go
package polling
|
||
|
||
import (
|
||
"context"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// 分钟计数 Hash 字段（与 gatewaylimit.Guard 写入口径一致）。
|
||
const (
|
||
statAttempts = "attempts"
|
||
statDurationSum = "duration_ms_sum"
|
||
statWouldLimit = "would_limit"
|
||
statPrefixDur = "duration_le:"
|
||
statPrefixExem = "exempt:"
|
||
statPrefixUp = "upstream:"
|
||
)
|
||
|
||
// statDurationBuckets 与写入端分桶上界一致（毫秒）。
|
||
var statDurationBuckets = []string{"100", "300", "500", "1000", "3000", "10000", "60000", "inf"}
|
||
|
||
// ListStopResumeRecords 按条件查询停复机记录；时间范围必填且跨度不超过 31 天。
|
||
func (s *GatewayRateLimitService) ListStopResumeRecords(ctx context.Context, req *dto.StopResumeRecordListRequest) (*dto.StopResumeRecordPageResult, error) {
|
||
if err := requirePlatformReader(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
if req.StartTime.IsZero() || req.EndTime.IsZero() {
|
||
return nil, errors.New(errors.CodeInvalidParam, "停复机记录查询参数不合法：开始时间与结束时间必填")
|
||
}
|
||
if !req.EndTime.After(req.StartTime) || req.EndTime.Sub(req.StartTime) > time.Duration(constants.StopResumeRecordQueryMaxDays)*24*time.Hour {
|
||
return nil, errors.New(errors.CodeInvalidParam, "停复机记录查询参数不合法：时间跨度须大于 0 且不超过 31 天")
|
||
}
|
||
page, size := normalizePage(req.Page, req.PageSize)
|
||
filter := postgres.StopResumeRecordFilter{
|
||
ICCID: strings.TrimSpace(req.ICCID), CarrierID: req.CarrierID, Action: req.Action, Source: req.Source,
|
||
Result: req.Result, ErrorCategory: req.ErrorCategory, StartTime: req.StartTime, EndTime: req.EndTime,
|
||
Scene: req.Scene, Reason: req.Reason, CommandID: strings.TrimSpace(req.CommandID), Page: page, PageSize: size,
|
||
}
|
||
if identifier := strings.TrimSpace(req.Identifier); identifier != "" {
|
||
cardIDs, err := s.cardIDsByIdentifier(ctx, identifier)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
filter.CardIDs = cardIDs
|
||
}
|
||
return s.listRecords(ctx, filter)
|
||
}
|
||
|
||
// ListCardStopResumeRecords 查询单卡停复机历史；未传时间默认最近 31 天。
|
||
func (s *GatewayRateLimitService) ListCardStopResumeRecords(ctx context.Context, req *dto.CardStopResumeRecordListRequest) (*dto.StopResumeRecordPageResult, error) {
|
||
if err := requirePlatformReader(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
var count int64
|
||
if err := s.db.WithContext(ctx).Model(&model.IotCard{}).Where("id = ?", req.ID).Count(&count).Error; err != nil {
|
||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询 IoT 卡失败")
|
||
}
|
||
if count == 0 {
|
||
return nil, errors.New(errors.CodeNotFound, "IoT 卡不存在")
|
||
}
|
||
end := req.EndTime
|
||
if end.IsZero() {
|
||
end = time.Now()
|
||
}
|
||
start := req.StartTime
|
||
if start.IsZero() {
|
||
start = end.AddDate(0, 0, -constants.StopResumeRecordQueryMaxDays)
|
||
}
|
||
if !end.After(start) || end.Sub(start) > time.Duration(constants.StopResumeRecordQueryMaxDays)*24*time.Hour {
|
||
return nil, errors.New(errors.CodeInvalidParam, "停复机记录查询参数不合法：时间跨度须大于 0 且不超过 31 天")
|
||
}
|
||
page, size := normalizePage(req.Page, req.PageSize)
|
||
cardID := req.ID
|
||
return s.listRecords(ctx, postgres.StopResumeRecordFilter{CardID: &cardID, StartTime: start, EndTime: end, Page: page, PageSize: size,
|
||
Action: req.Action, Result: req.Result, Source: req.Source})
|
||
}
|
||
|
||
// cardIDsByIdentifier 把资产标识解析为卡：卡按虚拟号、ICCID、MSISDN 匹配，设备按虚拟号、IMEI、SN 匹配后取其绑定卡。
|
||
// 匹配不到时返回空切片（查询结果为空页）。
|
||
func (s *GatewayRateLimitService) cardIDsByIdentifier(ctx context.Context, identifier string) ([]uint, error) {
|
||
cardIDs := []uint{}
|
||
if err := s.db.WithContext(ctx).Model(&model.IotCard{}).
|
||
Where("iccid = ? OR virtual_no = ? OR msisdn = ?", identifier, identifier, identifier).
|
||
Pluck("id", &cardIDs).Error; err != nil {
|
||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "按资产标识查询卡失败")
|
||
}
|
||
if len(cardIDs) > 0 {
|
||
return cardIDs, nil
|
||
}
|
||
if err := s.db.WithContext(ctx).Model(&model.DeviceSimBinding{}).
|
||
Joins("JOIN tb_device d ON d.id = tb_device_sim_binding.device_id AND d.deleted_at IS NULL").
|
||
Where("(d.virtual_no = ? OR d.imei = ? OR d.sn = ?) AND tb_device_sim_binding.bind_status = ?", identifier, identifier, identifier, constants.BindStatusBound).
|
||
Pluck("tb_device_sim_binding.iot_card_id", &cardIDs).Error; err != nil {
|
||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "按资产标识查询设备绑定卡失败")
|
||
}
|
||
return cardIDs, nil
|
||
}
|
||
|
||
func (s *GatewayRateLimitService) listRecords(ctx context.Context, filter postgres.StopResumeRecordFilter) (*dto.StopResumeRecordPageResult, error) {
|
||
rows, total, err := postgres.NewCardStopResumeRecordStore(s.db).List(ctx, filter)
|
||
if err != nil {
|
||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询停复机记录失败")
|
||
}
|
||
items := make([]dto.StopResumeRecordResponse, 0, len(rows))
|
||
for _, row := range rows {
|
||
items = append(items, dto.StopResumeRecordResponse{
|
||
ID: row.ID, CardID: row.CardID, ICCID: row.ICCID, CarrierID: row.CarrierID, Action: row.Action,
|
||
ActionName: stopResumeActionName(row.Action), Source: row.Source, Scene: row.Scene, Reason: row.Reason,
|
||
CommandID: row.CommandID, Attempt: row.Attempt, Result: row.Result, ErrorCategory: row.ErrorCategory,
|
||
ErrorCategoryName: errorCategoryName(row.ErrorCategory), ProviderCode: row.ProviderCode,
|
||
ProviderMessage: row.ProviderMessage, DurationMS: row.DurationMS, CreatedAt: row.CreatedAt,
|
||
})
|
||
}
|
||
return &dto.StopResumeRecordPageResult{List: items, Total: total, Page: filter.Page, PageSize: filter.PageSize}, nil
|
||
}
|
||
|
||
// CarrierCallStats 汇总最近若干分钟各运营商账户的网关调用（只返回有调用的账户）。
|
||
func (s *GatewayRateLimitService) CarrierCallStats(ctx context.Context, redisClient *redis.Client, req *dto.GatewayCallStatsRequest) (*dto.GatewayCallStatsResponse, error) {
|
||
if err := requirePlatformReader(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
minutes := req.Minutes
|
||
if minutes <= 0 {
|
||
minutes = 15
|
||
}
|
||
names, err := s.carrierNames(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
carrierIDs := []uint{0}
|
||
for id := range names {
|
||
carrierIDs = append(carrierIDs, id)
|
||
}
|
||
sort.Slice(carrierIDs, func(i, j int) bool { return carrierIDs[i] < carrierIDs[j] })
|
||
stamps := minuteStamps(time.Now(), minutes)
|
||
data, err := loadStatHashes(ctx, redisClient, carrierIDs, stamps)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
resp := &dto.GatewayCallStatsResponse{Mode: s.currentMode(ctx), Minutes: minutes, Carriers: []dto.GatewayCarrierStat{}}
|
||
for _, carrierID := range carrierIDs {
|
||
var ops []dto.GatewayOperationStat
|
||
for _, op := range constants.GatewayCardOperations {
|
||
stat := aggregateOperation(op, data[statKey{carrierID, op}], len(stamps))
|
||
if stat.Attempts > 0 || stat.RateLimited > 0 || stat.WouldLimit > 0 {
|
||
ops = append(ops, stat)
|
||
}
|
||
}
|
||
if len(ops) == 0 {
|
||
continue
|
||
}
|
||
name := names[carrierID]
|
||
if carrierID == 0 {
|
||
name = "未知账户"
|
||
}
|
||
resp.Carriers = append(resp.Carriers, dto.GatewayCarrierStat{CarrierID: carrierID, CarrierName: name, Operations: ops})
|
||
}
|
||
return resp, nil
|
||
}
|
||
|
||
// CarrierCallTrend 返回单个运营商账户按时间桶聚合的网关调用趋势。
|
||
func (s *GatewayRateLimitService) CarrierCallTrend(ctx context.Context, redisClient *redis.Client, req *dto.GatewayCallTrendRequest) (*dto.GatewayCallTrendResponse, error) {
|
||
if err := requirePlatformReader(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
minutes, bucket := req.Minutes, req.BucketMinutes
|
||
if minutes <= 0 {
|
||
minutes = 60
|
||
}
|
||
if bucket <= 0 {
|
||
bucket = 1
|
||
}
|
||
stamps := minuteStamps(time.Now(), minutes)
|
||
data, err := loadStatHashes(ctx, redisClient, []uint{req.CarrierID}, stamps)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
resp := &dto.GatewayCallTrendResponse{CarrierID: req.CarrierID, Points: []dto.GatewayCallTrendPoint{}}
|
||
for start := 0; start < len(stamps); start += bucket {
|
||
end := start + bucket
|
||
if end > len(stamps) {
|
||
end = len(stamps)
|
||
}
|
||
point := dto.GatewayCallTrendPoint{Time: formatStamp(stamps[start]), Operations: []dto.GatewayOperationStat{}}
|
||
for _, op := range constants.GatewayCardOperations {
|
||
series := data[statKey{req.CarrierID, op}]
|
||
if series == nil {
|
||
continue
|
||
}
|
||
stat := aggregateOperation(op, series[start:end], end-start)
|
||
if stat.Attempts > 0 || stat.RateLimited > 0 || stat.WouldLimit > 0 {
|
||
point.Operations = append(point.Operations, stat)
|
||
}
|
||
}
|
||
resp.Points = append(resp.Points, point)
|
||
}
|
||
return resp, nil
|
||
}
|
||
|
||
func (s *GatewayRateLimitService) currentMode(ctx context.Context) string {
|
||
if s.runtime == nil {
|
||
return constants.RateLimitModeObserve
|
||
}
|
||
return s.runtime.Settings.Current(ctx).Mode
|
||
}
|
||
|
||
type statKey struct {
|
||
carrierID uint
|
||
op string
|
||
}
|
||
|
||
// minuteStamps 返回从最早到最近的分钟标识（上海时区 yyyyMMddHHmm）。
|
||
func minuteStamps(now time.Time, minutes int) []string {
|
||
location := shanghaiLocation()
|
||
current := now.In(location).Truncate(time.Minute)
|
||
stamps := make([]string, minutes)
|
||
for i := 0; i < minutes; i++ {
|
||
stamps[minutes-1-i] = current.Add(-time.Duration(i) * time.Minute).Format("200601021504")
|
||
}
|
||
return stamps
|
||
}
|
||
|
||
func formatStamp(stamp string) string {
|
||
t, err := time.ParseInLocation("200601021504", stamp, shanghaiLocation())
|
||
if err != nil {
|
||
return stamp
|
||
}
|
||
return t.Format("2006-01-02 15:04")
|
||
}
|
||
|
||
// loadStatHashes 用管道批量读取分钟计数；返回每个账户 × 操作按分钟顺序的 Hash 列表。
|
||
func loadStatHashes(ctx context.Context, redisClient *redis.Client, carrierIDs []uint, stamps []string) (map[statKey][]map[string]string, error) {
|
||
result := map[statKey][]map[string]string{}
|
||
if redisClient == nil {
|
||
return result, nil
|
||
}
|
||
type pending struct {
|
||
key statKey
|
||
idx int
|
||
cmd *redis.MapStringStringCmd
|
||
}
|
||
pipe := redisClient.Pipeline()
|
||
cmds := make([]pending, 0, len(carrierIDs)*len(constants.GatewayCardOperations)*len(stamps))
|
||
for _, carrierID := range carrierIDs {
|
||
for _, op := range constants.GatewayCardOperations {
|
||
key := statKey{carrierID, op}
|
||
result[key] = make([]map[string]string, len(stamps))
|
||
for idx, stamp := range stamps {
|
||
cmds = append(cmds, pending{key: key, idx: idx, cmd: pipe.HGetAll(ctx, constants.RedisGatewayCallStatKey(carrierID, op, stamp))})
|
||
}
|
||
}
|
||
}
|
||
if _, err := pipe.Exec(ctx); err != nil && err != redis.Nil {
|
||
return nil, errors.Wrap(errors.CodeRedisError, err, "读取网关调用统计失败")
|
||
}
|
||
for _, item := range cmds {
|
||
values, err := item.cmd.Result()
|
||
if err == nil && len(values) > 0 {
|
||
result[item.key][item.idx] = values
|
||
}
|
||
}
|
||
return result, nil
|
||
}
|
||
|
||
// aggregateOperation 汇总一个操作在若干分钟内的计数。
|
||
func aggregateOperation(op string, minutes []map[string]string, minuteCount int) dto.GatewayOperationStat {
|
||
stat := dto.GatewayOperationStat{Operation: op, OperationName: constants.GatewayOperationName(op), Failures: map[string]int64{}, Exempt: map[string]int64{}}
|
||
buckets := map[string]int64{}
|
||
var durationSum int64
|
||
for _, values := range minutes {
|
||
var minuteAttempts int64
|
||
for field, raw := range values {
|
||
value, err := strconv.ParseInt(raw, 10, 64)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
switch {
|
||
case field == statAttempts:
|
||
stat.Attempts += value
|
||
minuteAttempts = value
|
||
case field == statDurationSum:
|
||
durationSum += value
|
||
case field == statWouldLimit:
|
||
stat.WouldLimit += value
|
||
case field == constants.GatewayErrorRateLimited:
|
||
stat.RateLimited += value
|
||
case field == constants.GatewayResultSuccess:
|
||
stat.Success += value
|
||
case strings.HasPrefix(field, statPrefixDur):
|
||
buckets[strings.TrimPrefix(field, statPrefixDur)] += value
|
||
case strings.HasPrefix(field, statPrefixExem):
|
||
stat.Exempt[strings.TrimPrefix(field, statPrefixExem)] += value
|
||
default:
|
||
stat.Failures[field] += value
|
||
}
|
||
}
|
||
if qps := float64(minuteAttempts) / 60; qps > stat.PeakMinuteQPS {
|
||
stat.PeakMinuteQPS = qps
|
||
}
|
||
}
|
||
if minuteCount > 0 {
|
||
stat.AvgQPS = float64(stat.Attempts) / float64(minuteCount*60)
|
||
}
|
||
if stat.Attempts > 0 {
|
||
stat.SuccessRate = float64(stat.Success) / float64(stat.Attempts)
|
||
stat.AvgDurationMS = float64(durationSum) / float64(stat.Attempts)
|
||
stat.P95DurationMS = percentileBucket(buckets, stat.Attempts, 0.95)
|
||
}
|
||
return stat
|
||
}
|
||
|
||
// percentileBucket 返回累计占比达到 p 的分桶上界；落在最后一个分桶时返回 -1。
|
||
func percentileBucket(buckets map[string]int64, total int64, p float64) int64 {
|
||
threshold := int64(float64(total)*p + 0.5)
|
||
var cumulative int64
|
||
for _, upper := range statDurationBuckets {
|
||
cumulative += buckets[upper]
|
||
if cumulative >= threshold {
|
||
if upper == "inf" {
|
||
return -1
|
||
}
|
||
value, _ := strconv.ParseInt(upper, 10, 64)
|
||
return value
|
||
}
|
||
}
|
||
return -1
|
||
}
|
||
|
||
func shanghaiLocation() *time.Location {
|
||
location, err := time.LoadLocation("Asia/Shanghai")
|
||
if err != nil {
|
||
return time.FixedZone("CST", 8*3600)
|
||
}
|
||
return location
|
||
}
|
||
|
||
func stopResumeActionName(action string) string {
|
||
if action == constants.StopResumeActionStart {
|
||
return "复机"
|
||
}
|
||
return "停机"
|
||
}
|
||
|
||
func errorCategoryName(category string) string {
|
||
switch category {
|
||
case "":
|
||
return ""
|
||
case constants.GatewayErrorNetwork:
|
||
return "网络错误"
|
||
case constants.GatewayErrorTimeout:
|
||
return "请求超时"
|
||
case constants.GatewayErrorHTTP:
|
||
return "HTTP 异常"
|
||
case constants.GatewayErrorInvalidResponse:
|
||
return "响应无效"
|
||
default:
|
||
return constants.GatewayErrorClassName(category)
|
||
}
|
||
}
|
||
|
||
// PollingOptions 返回轮询与网关限流相关的前端枚举，前端不写死取值。
|
||
func (s *GatewayRateLimitService) PollingOptions(ctx context.Context) (*dto.PollingOptionsResponse, error) {
|
||
if err := requirePlatformReader(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
resp := &dto.PollingOptionsResponse{}
|
||
for _, op := range constants.GatewayCardOperations {
|
||
resp.GatewayOperations = append(resp.GatewayOperations, dto.PollingOptionItem{Value: op, Label: constants.GatewayOperationName(op)})
|
||
}
|
||
for _, class := range constants.GatewayErrorClasses {
|
||
resp.ErrorCategories = append(resp.ErrorCategories, dto.PollingOptionItem{Value: class, Label: constants.GatewayErrorClassName(class)})
|
||
}
|
||
resp.MatchTypes = []dto.PollingOptionItem{{Value: constants.GatewayMatchContains, Label: "包含"}, {Value: constants.GatewayMatchPrefix, Label: "前缀"}, {Value: constants.GatewayMatchRegex, Label: "正则"}}
|
||
resp.RateLimitModes = []dto.PollingOptionItem{{Value: constants.RateLimitModeObserve, Label: "只统计不拦截"}, {Value: constants.RateLimitModeEnforce, Label: "拦截"}}
|
||
resp.ExemptSources = []dto.PollingOptionItem{
|
||
{Value: constants.RateLimitExemptManualClick, Label: "后台人工点击"}, {Value: constants.RateLimitExemptChannelThreshold, Label: "通道阈值"},
|
||
{Value: constants.RateLimitExemptAutoRenewal, Label: "自动续费"}, {Value: constants.RateLimitExemptManualRefresh, Label: "手动刷新"},
|
||
}
|
||
resp.StopResumeResults = []dto.PollingOptionItem{
|
||
{Value: constants.StopResumeRecordSuccess, Label: "成功"}, {Value: constants.StopResumeRecordFailed, Label: "失败"},
|
||
{Value: constants.StopResumeRecordUnknown, Label: "结果未知"}, {Value: constants.StopResumeRecordAlreadyTarget, Label: "已是目标状态"},
|
||
{Value: constants.StopResumeRecordSkipped, Label: "跳过"},
|
||
}
|
||
resp.StopResumeActions = []dto.PollingOptionItem{{Value: constants.StopResumeActionStop, Label: "停机"}, {Value: constants.StopResumeActionStart, Label: "复机"}}
|
||
return resp, nil
|
||
}
|