feat: 采集记录成功失败分表并保存网关原始数据,到期归档 csv.gz,新增记录分页查询接口
- 网关客户端新增原始数据收集器,记录最后一次尝试的明文请求、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>
This commit is contained in:
@@ -2,6 +2,7 @@ package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -90,20 +91,25 @@ func (h *PollingCarddataHandler) Step(ctx context.Context, cardID uint) StepOutc
|
||||
}
|
||||
priority.MarkExecution(ctx)
|
||||
attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayTraffic, constants.CardObservationSceneTrafficPolling)
|
||||
result, err := h.gateway.QueryFlow(gateway.WithCarrier(ctx, card.CarrierID), &gateway.FlowQueryReq{CardNo: card.ICCID})
|
||||
queryCtx, rawCapture := gateway.WithRawCapture(gateway.WithCarrier(ctx, card.CarrierID))
|
||||
queryStartedAt := time.Now()
|
||||
result, err := h.gateway.QueryFlow(queryCtx, &gateway.FlowQueryReq{CardNo: card.ICCID})
|
||||
unreadable := unreadableDetail{raw: domainGatewayRaw(rawCapture), failure: collectionFailure(err, time.Since(queryStartedAt))}
|
||||
if gateway.IsRateLimited(err) {
|
||||
h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultRateLimited)
|
||||
h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultRateLimited, unreadable)
|
||||
priority.FailRetryable(ctx, "运营商账户网关限流,稍后重试")
|
||||
return StepOutcome{Result: StepRateLimited}
|
||||
}
|
||||
if err != nil {
|
||||
h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultFailed)
|
||||
h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultFailed, unreadable)
|
||||
_ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询流量失败", priority)
|
||||
priority.FailRetryable(ctx, "查询流量失败")
|
||||
return h.fail(ctx, cardID, startedAt, "查询流量失败", err)
|
||||
}
|
||||
if strings.TrimSpace(result.ICCID) == "" {
|
||||
h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultInvalidPayload)
|
||||
unreadable.failure = &carddomain.CollectionFailure{ErrorCategory: constants.GatewayErrorInvalidResponse,
|
||||
ProviderMessage: "流量查询响应缺少 ICCID", DurationMS: int(time.Since(queryStartedAt).Milliseconds())}
|
||||
h.recordUnreadable(ctx, cardID, attempt.IntegrationID, constants.TrafficCollectionResultInvalidPayload, unreadable)
|
||||
_ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "流量查询响应缺少 ICCID", priority)
|
||||
priority.FailRetryable(ctx, "流量查询响应缺少 ICCID")
|
||||
return h.fail(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil)
|
||||
@@ -116,7 +122,8 @@ func (h *PollingCarddataHandler) Step(ctx context.Context, cardID uint) StepOutc
|
||||
now := time.Now()
|
||||
decision, err := h.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{
|
||||
CardID: cardID, GatewayReadingMB: float64(result.Used), ReadingResult: gatewayReadingResult(result),
|
||||
ResetDay: h.carrier.GetDataResetDay(ctx, card.CarrierID),
|
||||
ResetDay: h.carrier.GetDataResetDay(ctx, card.CarrierID), Raw: unreadable.raw,
|
||||
Failure: &carddomain.CollectionFailure{DurationMS: int(time.Since(queryStartedAt).Milliseconds())},
|
||||
Metadata: carddomain.ObservationMetadata{
|
||||
ObservationID: attempt.IntegrationID, Source: constants.CardObservationSourcePolling,
|
||||
Scene: constants.CardObservationSceneTrafficPolling, ObservedAt: now, CorrelationID: attempt.IntegrationID,
|
||||
@@ -159,12 +166,18 @@ func (h *PollingCarddataHandler) fail(ctx context.Context, cardID uint, startedA
|
||||
}
|
||||
|
||||
// recordUnreadable 为没拿到有效读数的采集写采集记录(只记采集时刻有套餐的卡),失败只影响留痕。
|
||||
func (h *PollingCarddataHandler) recordUnreadable(ctx context.Context, cardID uint, observationID, result string) {
|
||||
// unreadableDetail 是读数不可用时随采集失败记录保存的原始数据与错误详情。
|
||||
type unreadableDetail struct {
|
||||
raw *carddomain.GatewayRaw
|
||||
failure *carddomain.CollectionFailure
|
||||
}
|
||||
|
||||
func (h *PollingCarddataHandler) recordUnreadable(ctx context.Context, cardID uint, observationID, result string, detail unreadableDetail) {
|
||||
if h.observation == nil {
|
||||
return
|
||||
}
|
||||
if _, err := h.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{
|
||||
CardID: cardID, ReadingResult: result, ResetDay: 1,
|
||||
CardID: cardID, ReadingResult: result, ResetDay: 1, Raw: detail.raw, Failure: detail.failure,
|
||||
Metadata: carddomain.ObservationMetadata{
|
||||
ObservationID: observationID, Source: constants.CardObservationSourcePolling,
|
||||
Scene: constants.CardObservationSceneTrafficPolling, ObservedAt: time.Now(), CorrelationID: observationID,
|
||||
@@ -181,3 +194,35 @@ func gatewayReadingResult(result *gateway.FlowUsageResp) string {
|
||||
}
|
||||
return constants.TrafficCollectionResultOK
|
||||
}
|
||||
|
||||
// domainGatewayRaw 把网关原始数据收集器转为领域层原始数据。
|
||||
func domainGatewayRaw(capture *gateway.RawCapture) *carddomain.GatewayRaw {
|
||||
if capture == nil || (capture.Attempts == 0 && len(capture.Request) == 0) {
|
||||
return nil
|
||||
}
|
||||
return &carddomain.GatewayRaw{
|
||||
Request: capture.Request, HTTPStatus: capture.HTTPStatus, Response: capture.Response,
|
||||
GatewayCode: capture.GatewayCode, GatewayMsg: capture.GatewayMsg, TraceID: capture.TraceID, Attempts: capture.Attempts,
|
||||
}
|
||||
}
|
||||
|
||||
// collectionFailure 从网关错误提取采集失败详情;无错误时只带耗时。
|
||||
func collectionFailure(err error, duration time.Duration) *carddomain.CollectionFailure {
|
||||
failure := &carddomain.CollectionFailure{DurationMS: int(duration.Milliseconds())}
|
||||
if err == nil {
|
||||
return failure
|
||||
}
|
||||
failure.ProviderMessage = err.Error()
|
||||
if gwErr, ok := gateway.AsGatewayError(err); ok {
|
||||
failure.ErrorCategory = gwErr.Category
|
||||
if gwErr.ProviderCode != 0 {
|
||||
failure.ProviderCode = strconv.Itoa(gwErr.ProviderCode)
|
||||
}
|
||||
if gwErr.ProviderMessage != "" {
|
||||
failure.ProviderMessage = gwErr.ProviderMessage
|
||||
}
|
||||
} else if gateway.IsRateLimited(err) {
|
||||
failure.ErrorCategory = constants.TrafficCollectionResultRateLimited
|
||||
}
|
||||
return failure
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user