feat: 审计动作按落库策略分级写入,系统过程动作改写应用日志
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 12m8s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 12m8s
- 审计注册表为可由系统任务、定时任务发起的动作声明落库策略(always/on_success/outcome/log_only/none),未声明按 log_only - 人工、开放接口、个人客户、外部系统发起的动作始终写审计表 - log_only 写所在服务 app.log(logger=audit),不含快照 - 对接日志不再补写 integration.attempt_started,无上游审计时 audit_event_id 为空 - 新增 audit.persist_overrides 按动作覆盖策略 - 审计覆盖扫描要求机器动作必须声明策略 - 新增 OpenSpec 提案 refactor-audit-log-persistence 与技术方案文档 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -2,6 +2,7 @@ package main
|
||||
|
||||
import (
|
||||
stdErrors "errors"
|
||||
auditInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
||||
"os"
|
||||
"os/signal"
|
||||
"reflect"
|
||||
@@ -146,6 +147,7 @@ func initLogger(cfg *config.Config) *zap.Logger {
|
||||
|
||||
appLogger := logger.GetAppLogger()
|
||||
appLogger.Info("应用程序启动中...", zap.String("address", cfg.Server.Address))
|
||||
applyAuditPersistOverrides(cfg, appLogger)
|
||||
return appLogger
|
||||
}
|
||||
|
||||
@@ -472,3 +474,10 @@ func initGateway(cfg *config.Config, db *gorm.DB, appLogger *zap.Logger) *gatewa
|
||||
|
||||
return client
|
||||
}
|
||||
|
||||
// applyAuditPersistOverrides 加载审计落库策略的运行时覆盖;非法项忽略并告警。
|
||||
func applyAuditPersistOverrides(cfg *config.Config, appLogger *zap.Logger) {
|
||||
if invalid := auditInfra.SetPersistOverrides(cfg.Audit.PersistOverrides); len(invalid) > 0 {
|
||||
appLogger.Warn("审计落库策略覆盖存在非法项,已忽略", zap.Strings("invalid", invalid))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -151,6 +151,9 @@ func runWorker(cfg *config.Config) {
|
||||
defer func() {
|
||||
_ = logger.Sync() // 忽略 sync 错误
|
||||
}()
|
||||
if invalid := auditInfra.SetPersistOverrides(cfg.Audit.PersistOverrides); len(invalid) > 0 {
|
||||
logger.GetAppLogger().Warn("审计落库策略覆盖存在非法项,已忽略", zap.Strings("invalid", invalid))
|
||||
}
|
||||
retentionLogger, syncRetentionLogger := logger.NewRetentionLogger(cfg.Logging.Level, logger.LogRotationConfig{
|
||||
Filename: cfg.Logging.RetentionLog.Filename, MaxSize: cfg.Logging.RetentionLog.MaxSize,
|
||||
MaxBackups: cfg.Logging.RetentionLog.MaxBackups, MaxAge: cfg.Logging.RetentionLog.MaxAge, Compress: cfg.Logging.RetentionLog.Compress,
|
||||
|
||||
440
docs/2026-11-05-审计日志写入改造技术方案.md
Normal file
440
docs/2026-11-05-审计日志写入改造技术方案.md
Normal file
@@ -0,0 +1,440 @@
|
||||
# 审计日志写入改造技术方案
|
||||
|
||||
> 日期:2026-11-05
|
||||
> 状态:方案已定,未实施(结论见第 10 节)
|
||||
> 关联:`openspec/changes/add-audit-integration-write-switches`(全局写入开关,本方案是其后的"按动作分级")
|
||||
> **范围:本方案基于 `Iteration/8-11` 迭代实施,只部署到测试环境**(推送 8-11 由 `.gitea/workflows/deploy.yaml` 自动部署)。本方案不修改生产环境的代码、配置或服务;第 1 节的生产数据只用来说明问题和估算效果。
|
||||
|
||||
## 1. 背景
|
||||
|
||||
### 1.1 现状数据(生产近 14 天,只读统计)
|
||||
|
||||
| 指标 | 数值 |
|
||||
|---|---|
|
||||
| `tb_audit_event` 日均新增 | 约 122 万条(含停机重试风暴);扣掉风暴约 20 万条 |
|
||||
| 机器发起(`system_task` / `scheduled_job`)占比 | 99% 以上 |
|
||||
| 人、开放接口、个人客户、外部系统发起 | 约 1,210 条/天 |
|
||||
| 每条事件含索引 | 约 1.9 KB;每条资源快照约 1.2 KB;平均 2–3 条资源 |
|
||||
| 日均新增存储 | 约 6 GB(风暴期)/ 约 1 GB(平时) |
|
||||
| 索引数 | `tb_audit_event` 14 个、`tb_audit_event_resource` 7 个、`tb_integration_log` 13 个 |
|
||||
|
||||
测试环境日均约 12.6 万条,几乎全是系统任务:流量同步 5.6 万、扣减 2.6 万、网络状态同步 2.5 万、优先轮询约 1.4 万。
|
||||
|
||||
### 1.2 问题
|
||||
|
||||
1. **审计表被当成过程日志用。** 每次轮询同步、每次扣减、每次对接调用开始、每次失败重试都写一条事件加 2–3 条资源快照。一次业务动作产生多次写入,而且每次写入要维护 21 个索引。
|
||||
2. **失败重试没有收敛。** 退款后处理停机失败(`refund.process_asset`)和自动停机失败(`iot_card.auto_stop`),每重试一次就写一条审计。11-05 10:16–11:47,86 张卡产生了 26.1 万次停机失败,一个风暴就占了当天审计量的六成以上。
|
||||
3. **对接日志顺带写审计。** `integrationlog.Repository.Start` 在没有上游审计事件时,会补写一条 `integration.attempt_started`(`internal/infrastructure/integrationlog/repository.go:119`),这一项每天约 45.6 万条。它与 `tb_integration_log` 本身完全重复。
|
||||
4. **后果。** 11-03 生产 PostgreSQL IO 打满,轮询停摆约 1.7 天,审计和对接日志的写入与留存扫描是主要 IO 来源之一。
|
||||
|
||||
### 1.3 老系统做法(`/home/zoujin/works/golang/kg/qc`,供参考)
|
||||
|
||||
- **人工后台操作**:写 `tbl_system_log`。
|
||||
- **业务过程**:按业务各写一张表,如 `tbl_order_log`、`tbl_card_wallet_auto_renew_log`、`tbl_next_month_log`,以及充值、短信日志。
|
||||
- **下游代理调用我方接口**:只这一类写 `tbl_agent_pool_api_call_log`。
|
||||
- **网关调用和轮询**:全部写 log4j 按天滚动的文本日志,不进数据库。
|
||||
|
||||
## 2. 目标与原则
|
||||
|
||||
**目标**:审计表只回答两个问题:
|
||||
|
||||
1. **人(或代表人的外部系统)做了什么**;
|
||||
2. **钱和套餐、卡的业务状态是怎么变的**。
|
||||
|
||||
审计日均降到几千条,一年不到 10 GB,不再依赖归档和清理来扛量。
|
||||
|
||||
**原则**:
|
||||
|
||||
| 原则 | 说明 |
|
||||
|---|---|
|
||||
| 人工操作全留 | 后台账号、个人客户、开放接口、外部系统发起的动作,成功和失败都写审计 |
|
||||
| 机器动作只留"结果" | 系统任务和定时任务只在资金或业务状态**真正改变**时写审计;无变化、同步、读取不写 |
|
||||
| 每次失败都可查,审计只记结论 | 自动重试的每一次尝试都完整记在业务尝试记录表里(时间、第几次、错误码、上游返回),可分页查询;审计只记这件事的最终结论:成功,或重试用尽转人工 |
|
||||
| 过程数据去业务表 | 已有专门记录表的(停复机记录、采集记录、扣减效果、日流量),审计不再重复写 |
|
||||
| 排查用的走文本日志 | 只用于排查、不需要按业务查询的,写入现有应用日志 `app.log`,沿用现有轮换,由日志同步服务器归集 |
|
||||
| 不改变业务结果 | 写不写审计,都不影响业务事务成功与否(与现状 `Append` 吞错一致) |
|
||||
|
||||
## 3. 数据分层
|
||||
|
||||
| 层 | 存储 | 放什么 | 保留 |
|
||||
|---|---|---|---|
|
||||
| L1 审计 | `tb_audit_event` + `tb_audit_event_resource` | 人工操作;机器发起的资金、套餐生命周期、停复机成功等状态变化 | 按现有留存策略 |
|
||||
| L2 业务记录 | 各业务专表 | 停复机调用 `tb_card_stop_resume_record`、流量采集 `tb_card_traffic_collection`、扣减效果与日流量、对接调用 `tb_integration_log`、退款单上的后处理状态 | 按各表策略 |
|
||||
| L3 文本日志 | 现有应用日志 `app.log`(zap JSON + lumberjack 轮换) | 机器过程类动作的结构化日志,例如停机原因更新、网络状态同步、通知投递、优先轮询、流量重置明细 | 沿用 `logging.app_log` 轮换;长期留存由日志同步服务器负责 |
|
||||
| L4 不写 | — | 与 L1/L2 完全重复的,例如 `integration.attempt_started` | — |
|
||||
|
||||
## 4. 动作去向清单
|
||||
|
||||
### 4.1 保留在审计(L1)
|
||||
|
||||
**人工及外部发起**(约 1,210 条/天,全部保留):
|
||||
|
||||
| 发起方 | 日均 | 主要动作 |
|
||||
|---|---|---|
|
||||
| 后台账号 | 约 404 | 手动刷新卡、设备切卡、回收、设 WiFi、手动复机、改套餐到期、设备重启、下单、退款、审批、登录等 |
|
||||
| 开放接口 | 约 612 | 下单、代理钱包扣款、开放接口复机 |
|
||||
| 个人客户 | 约 182 | 绑资产、建手机号关联、更新微信身份、绑手机号 |
|
||||
| 外部系统 | 约 9 | 支付确认、外部回调接收 |
|
||||
|
||||
**机器发起但改变资金或业务状态**(方案 A 约 3,510 条/天):
|
||||
|
||||
| 动作 | 日均 | 方案 B 去向 |
|
||||
|---|---|---|
|
||||
| `iot_card.auto_start` 自动复机 | 1,127 | 移到 L2 停复机记录 |
|
||||
| `package_usage.reset_traffic` 流量重置 | 1,049 | 移到 L2(套餐自身的重置时间加日流量记录) |
|
||||
| `iot_card.auto_stop` 自动停机,**仅成功** | 380 | 移到 L2 停复机记录 |
|
||||
| `commission.calculate` 分佣计算 | 238 | 保留 |
|
||||
| `package_usage.expire` / `activate` / 退款失效 | 176 / 90 / 6 | 保留 |
|
||||
| 导入建卡、建设备、设备分配 | 93 / 52 / 44 | 保留 |
|
||||
| 系统下单、订单套餐失效明细、员工代收账单 | 77 / 74 / 51 | 保留 |
|
||||
| 外部回调接收、审批同步、退款批准、佣金回溯判定 | 27 / 13 / 6 / 6 | 保留 |
|
||||
| `refund.process_asset`,**仅成功** | 4 | 保留 |
|
||||
|
||||
### 4.2 移出审计
|
||||
|
||||
| 动作 | 现在日均(估算) | 去向 | 说明 |
|
||||
|---|---|---|---|
|
||||
| `integration.attempt_started` | 约 45.6 万 | L4 不写 | `tb_integration_log` 已有同样内容;`audit_event_id` 留空 |
|
||||
| `refund.process_asset` 失败 | 约 32.6 万 | L2 退款后处理尝试记录 | 每次尝试一条,记在新表 `tb_refund_post_process_attempt`;退款单上保留当前状态、次数和最后错误;审计只在最终成功或转人工时写一条 |
|
||||
| `iot_card.auto_stop` 失败 | 约 31.5 万 | L2 停复机记录 | `tb_card_stop_resume_record` 每次实际调用一条,已含错误分类 |
|
||||
| `iot_card.auto_stop_reason_update` | 约 3.8 万 | L3 文本 | 只改停机原因、不改停复机状态 |
|
||||
| `iot_card.worker_traffic_sync` | 约 3.1 万(正常时 6–8 万) | L2 采集记录 | `tb_card_traffic_collection` 已经记录每次采集 |
|
||||
| `iot_card.worker_network_sync` | 约 2.6 万(正常时 6–8 万) | L3 文本 | 结果就是卡上的网络状态字段 |
|
||||
| `package_usage.deduct_traffic` | 约 2.1 万 | L2 | 采集记录、扣减效果、日流量三张表已经能还原 |
|
||||
| `notification.deliver`、`polling_priority.*` 等 | 约 0.5 万 | L3 文本 | — |
|
||||
|
||||
### 4.3 方案选择
|
||||
|
||||
采用**方案 B**。停复机和流量重置都已经有更合适的业务表,审计里只留人工操作和资金、套餐生命周期类动作。
|
||||
|
||||
| 方案 | 审计事件/天 | 占现在 | 每天新增存储 |
|
||||
|---|---|---|---|
|
||||
| 现在 | 约 122 万 | 100% | 约 6 GB |
|
||||
| 方案 A:4.1 全留 | 约 4,700 | 0.39% | 约 24 MB |
|
||||
| **方案 B**:停复机、流量重置移到 L2 | 约 **2,200** | 0.18% | 约 11 MB |
|
||||
|
||||
## 5. 技术设计
|
||||
|
||||
### 5.1 注册表增加"落库策略"
|
||||
|
||||
`internal/infrastructure/audit/registry.go` 的 `ActionDefinition` 新增字段:
|
||||
|
||||
```go
|
||||
// PersistPolicy 决定动作是否写入审计表。
|
||||
type PersistPolicy string
|
||||
|
||||
const (
|
||||
PersistAlways PersistPolicy = "always" // 成功失败都写(人工动作默认)
|
||||
PersistOnSuccess PersistPolicy = "on_success" // 只写成功;失败写文本日志
|
||||
PersistOutcome PersistPolicy = "outcome" // 只写最终结论:成功、或重试用尽转人工;中间失败不写审计(每次尝试在业务尝试记录表)
|
||||
PersistLogOnly PersistPolicy = "log_only" // 不写表,写 L3 文本日志
|
||||
PersistNone PersistPolicy = "none" // 什么都不写
|
||||
)
|
||||
|
||||
type ActionDefinition struct {
|
||||
// ……现有字段
|
||||
Persist PersistPolicy // 为空时按操作者决定,见 5.2
|
||||
}
|
||||
```
|
||||
|
||||
动作的策略在注册表中**集中定义**,调用方不改代码。这样一份清单就能看清哪些动作进审计,评审和后续调整也只需改这一处。
|
||||
|
||||
### 5.2 默认规则
|
||||
|
||||
`Persist` 为空时,按本次写入的操作者类型决定:
|
||||
|
||||
| 操作者 `actor_kind` | 默认策略 |
|
||||
|---|---|
|
||||
| `account`、`personal_customer`、`openapi`、`external_system` | `always` |
|
||||
| `system_task`、`scheduled_job` | `log_only` |
|
||||
|
||||
因此新增机器动作**默认不进审计**,要进审计必须在注册表显式声明 `always`、`on_success` 或 `outcome`。这能防止后面新增的轮询、同步类动作再次把审计表写满。
|
||||
|
||||
同一动作可能同时由人和机器触发,例如 `order.create`、`iot_card.manual_start` 与开放接口复机。策略按"动作 + 操作者"判断:注册表里的 `Persist` 只约束机器操作者,人工操作者始终是 `always`。
|
||||
|
||||
### 5.3 Writer 改造
|
||||
|
||||
只改 `internal/infrastructure/audit/writer.go` 的 `AppendAndGet`,在注册校验通过后、组装事件前分流:
|
||||
|
||||
```go
|
||||
action, ok := w.registry.Action(input.ActionCode)
|
||||
// ……现有校验
|
||||
switch w.persistDecision(ctx, tx, action, input) {
|
||||
case decisionSkip:
|
||||
return nil, nil
|
||||
case decisionLog:
|
||||
w.machineLog.Info("audit", auditLogFields(action, input)...)
|
||||
return nil, nil
|
||||
}
|
||||
// ……现有写表逻辑
|
||||
```
|
||||
|
||||
要点:
|
||||
|
||||
1. **调用方兼容**:`Append` 本来就吞错并返回 nil,调用方不受影响。`AppendAndGet` 有 11 处调用(`refund/clawback.go`、`refund/service.go`、`businessusergroup`、`packagetrafficalert`、`distribution`、`refund_channel`、`integrationlog`),实施时逐一确认能处理返回的 `nil` 事件。这 11 处大多是人工或资金动作,仍然是 `always`,实际不会拿到 nil。
|
||||
2. **`outcome` 的判定**:由调用方在写审计时标明这次是不是最终结论,`AppendInput` 增加 `Final bool`。
|
||||
- 中间失败(`Final=false`)不写审计表,只写一行应用日志,日志里带尝试记录 ID 方便对照;
|
||||
- 最终成功,或重试用尽转人工(`Final=true`)写一条审计,`metadata` 带总尝试次数、首次失败时间、最后错误码;
|
||||
- 用业务键作 `EventID`(例如 `refund-asset:{refund_id}:outcome`),保证同一件事的结论只有一条。
|
||||
- **每一次尝试的细节都不丢**,都在业务尝试记录表里(见 5.5),可分页查询(见 5.9)。
|
||||
3. **文本日志写入现有应用日志**:不新建日志文件,也不新增轮换配置。Writer 直接使用 `logger.GetAppLogger().Named("audit")`,写入现有的 `app.log`,轮换沿用 `logging.app_log`(测试环境配置见 5.8),由日志同步服务器自动归集。`NewWriter` 签名不变,调用方不用改。
|
||||
- 每行固定 `"logger":"audit"`、`"msg":"audit_event"`,便于在同步端按 logger 名过滤和建索引;
|
||||
- 字段:`action_code`、`actor_kind`、`actor_id`、`source`、`result`、`resource_type`、`resource_key`、`correlation_id`、`request_id`、`error_code`、`summary`;中间失败额外带 `attempt`(第几次)和尝试记录 ID;
|
||||
- 级别:成功为 Info,失败为 Warn;
|
||||
- **不写** before/after 快照与主体数据,避免敏感数据进入文本日志;需要变化值的动作(如流量重置前后用量)只写必要的数值字段。
|
||||
4. **运行时覆盖**:在 `add-audit-integration-write-switches` 的全局开关之外,增加一个按动作覆盖的配置 `audit.persist_overrides`(动作编码 → 策略)。出问题时不发版就能把某个动作临时降为 `log_only`,或恢复为 `always`。
|
||||
|
||||
### 5.4 对接日志不再补写审计
|
||||
|
||||
`internal/infrastructure/integrationlog/repository.go` 的 `Start`:
|
||||
|
||||
- **删除** `input.AuditEventID == nil` 时补写 `integration.attempt_started` 的逻辑(第 118–120 行)。上游有审计事件时照常关联;没有时 `audit_event_id` 留空。
|
||||
- 注册表里的 `integration.attempt_started` 设为 `none`,作为兜底。
|
||||
- `auditEventIDForSeries` 按 `trigger_series` 回查审计的逻辑保留,但需要确认它不会走慢查询(它会查审计表)。
|
||||
|
||||
### 5.5 失败重试收敛(与停机风暴止血配套)
|
||||
|
||||
审计只记结论之后,排查问题靠的是每一次尝试的记录,同时重试本身要收敛。这两件可以单独开 Change,但**必须先于或同时上线**:
|
||||
|
||||
1. **退款后处理**(`internal/service/refund/approval_decision.go` 的 `ProcessAssetPostProcessing`,`stopAsset` → `ManualStopCard`):
|
||||
- 退款单增加或复用后处理字段:`asset_post_status`、`asset_post_attempts`、`asset_post_last_error`、`asset_post_next_at`,表示当前状态;
|
||||
- **新增尝试记录表** `tb_refund_post_process_attempt`,每次尝试一条,不修改、不删除。字段:`refund_id`、`step`(停机、套餐失效、钱包回收等)、`asset_type`、`asset_id`、`iccid`、`attempt`、`trigger`(outbox 投递、补偿扫描、人工重试)、`result`、`error_code`、`error_message`、`stop_resume_record_id`(关联到那次停机调用)、`duration_ms`、`created_at`。按月分区。每单最多 10 次,量可控;
|
||||
- 失败按指数退避,超过 N 次(建议 10 次)标记为"需人工处理",不再自动重试;
|
||||
- outbox 与 `RecoverRefundPostProcessing` 都按这几个字段判断,不再无限重投。
|
||||
2. **自动停机**:8-11 上的停机已经走停复机指令队列(`stop-resume-command-queue`),按错误分类处理:终态不可操作不重试,限流按退避处理。退款停机只请求停机评估。本次不改这部分。
|
||||
- 每次实际调用上游,已经在 `tb_card_stop_resume_record` 记一条,包括第几次、错误类别、上游返回码与信息、耗时。
|
||||
- 指令在调用上游前就结束的情况(已是目标状态、条件不再满足、卡不允许停复机)现在没有记录。改为在停复机记录里写 `result=skipped`,并写明原因,保证"为什么没停"也能查到。限流推迟后还会再执行,不写记录,沿用现有的分钟计数。
|
||||
|
||||
### 5.6 方案 B 的业务表补位
|
||||
|
||||
| 移出的动作 | 补位方式 | 现状 |
|
||||
|---|---|---|
|
||||
| `iot_card.auto_start` / `auto_stop` 成功 | `tb_card_stop_resume_record`(每次实际调用一条,含来源、场景、原因、结果) | Iteration/8-11 已有 |
|
||||
| `package_usage.reset_traffic` | 套餐使用记录的 `last_reset_at` / `next_reset_at` 加日流量记录 | 已有;重置前后用量写一行 `app.log`(`logger=audit`)备查 |
|
||||
| `iot_card.auto_start` / `auto_stop` 的停机原因、触发场景 | `tb_card_stop_resume_record` 的 `source`、`scene`、`reason` | 已有;原审计里的 before/after 卡状态不再保留,卡当前状态以卡表为准 |
|
||||
|
||||
后台"卡操作记录"之类的页面如果要看到停复机时间线,查询侧改为合并审计和停复机记录两个来源,不要回头再写审计。
|
||||
|
||||
### 5.7 审计覆盖治理
|
||||
|
||||
`internal/governance/auditcoverage/scanner.go` 扫描每个入口的 `AuditEvent`、`DomainLedger`、`IntegrationLog` 分类。本次需要做三件事:
|
||||
|
||||
- 对移到 L2、L3、L4 的入口,把 `AuditEvent` 分类改为不写,并在 `DomainLedger` 里写明对应的业务表;
|
||||
- 覆盖扫描增加一条校验:机器操作者的动作必须在注册表显式声明 `Persist`,否则扫描失败;
|
||||
- 同步更新 `openspec/specs/operations-audit` 和 `external-integration` 中"每个入口都写审计"的要求。
|
||||
|
||||
### 5.8 应用日志增量与轮换(测试环境)
|
||||
|
||||
测试环境由 `docker-compose.prod.yml` 部署,api、worker 两个容器的现状:
|
||||
|
||||
- 日志级别是 `LEVEL=info`、`DEVELOPMENT=false`,即 JSON 格式,可以按 `"logger":"audit"` 过滤;
|
||||
- 两个容器都把宿主机 `./logs` 挂载到 `/app/logs`,且都**没有设置** `JUNHONG_LOGGING_APP_LOG_FILENAME`,所以都使用默认的 `/app/logs/app.log`;
|
||||
- 轮换参数也是默认值:100 MB 一个文件、3 个备份、7 天、压缩。
|
||||
|
||||
**问题**:api 和 worker 写的是宿主机上**同一个** `/opt/junhong_cmp/logs/app.log`。两个进程各自用 lumberjack 按大小轮换,会互相把对方正在写的文件改名、压缩,导致丢日志和备份错乱。这个问题与审计改造无关,但 L3 日志要落在 `app.log` 里,所以要一并修。
|
||||
|
||||
**调整原则**:每个服务写在自己运行目录下的 `logs/`。审计日志不新增路径、不改文件名,跟着所在进程的应用日志走。测试环境的问题只在于两个容器共用了同一个宿主机目录,所以只改挂载:
|
||||
|
||||
| 服务 | 挂载(宿主机 → 容器) | 宿主机日志位置 |
|
||||
|---|---|---|
|
||||
| api | `./api/logs:/app/logs` | `/opt/junhong_cmp/api/logs/app.log`、`access.log` |
|
||||
| worker | `./worker/logs:/app/logs` | `/opt/junhong_cmp/worker/logs/app.log` |
|
||||
|
||||
- 容器内仍然是默认的 `/app/logs/app.log`,**不设置** `APP_LOG_FILENAME`;目录结构与生产 `/opt/junhong_cmp/api/logs`、`/opt/junhong_cmp/worker/logs` 一致;
|
||||
- 轮换只加两项:`JUNHONG_LOGGING_APP_LOG_MAX_BACKUPS=40`、`JUNHONG_LOGGING_APP_LOG_MAX_AGE=14`,其余用默认值(100 MB、压缩);
|
||||
- 容量:两个进程合计约 1 GB,每个进程是 100 MB 当前文件加 40 个压缩备份。每个备份约 10 MB 是估算值,部署后按实际备份大小微调;
|
||||
- `.gitea/workflows/deploy.yaml` 里 `mkdir -p ${DEPLOY_DIR}/logs` 改为创建 `api/logs`、`worker/logs`;
|
||||
- 原来的 `/opt/junhong_cmp/logs` 不再写入,旧文件保留不动。
|
||||
|
||||
- **新增量**:方案 B 在测试环境移到 L3 的动作,每天约 8 万–10 万行:流量同步改由采集记录承担,不写日志;需要写日志的有网络状态同步约 2.5 万、优先轮询约 1.4 万、停机原因更新与通知若干。按每行约 400 B 计算,每天约 30–40 MB(压缩前),相对单文件 100 MB 没有压力。
|
||||
|
||||
文本日志是排查补充。丢几行不影响业务,也不影响审计表。写日志失败时不重试、不报错。
|
||||
|
||||
### 5.9 记录查询接口
|
||||
|
||||
审计里移出去的数据,要能在后台查得到,否则等于丢了。凡是独立成表的记录,都要提供**分页 + 条件筛选**的查询接口。L3 文本日志不提供接口,由日志同步服务器检索。
|
||||
|
||||
**公共约定**(与现有 `ListStopResumeRecords` 一致):
|
||||
|
||||
| 项 | 约定 |
|
||||
|---|---|
|
||||
| 分页 | `page` 默认 1;`page_size` 默认 20,最大 100(采集记录沿用现有的最大 200);返回 `total`、`page`、`page_size`、`items` |
|
||||
| 时间范围 | `start_time`、`end_time` 必填,跨度不超过 31 天,按上海时间解析;单卡接口为空时默认取最近 31 天 |
|
||||
| 卡标识 | `identifier` 支持虚拟号、ICCID、MSISDN,先解析成卡 ID 再查,保证走"卡 ID + 时间"索引;找不到卡时直接返回空页 |
|
||||
| 排序 | 按记录时间倒序,同一时间按 ID 倒序 |
|
||||
| 单个与全部 | 每类记录提供一个列表接口,`identifier` 选填:传了就查单卡或单资产,不传就查全部。现有的单卡接口保留给前端兼容,与列表接口共用同一个查询 |
|
||||
| 筛选 | 按业务排查需要来定,下面逐表列出,不受现有索引限制;为这些筛选新建或重建索引 |
|
||||
| 时间范围的作用 | 表按月分区,时间范围用于分区裁剪,所以查全部时也只带时间范围、不要求其他条件 |
|
||||
| 权限 | 平台账号;代理和企业账号是否开放、按归属范围过滤,另行确定 |
|
||||
|
||||
**接口清单**:
|
||||
|
||||
| 记录 | 表 | 列表接口 | 筛选条件 | 现状 |
|
||||
|---|---|---|---|---|
|
||||
| 运营商流量采集记录(成功,失败见 5.10) | `tb_card_traffic_collection` | `GET /traffic-collections` | `identifier`、`carrier_id`(运营商账户)、`decision`(扣减判定)、`source`、`scene`、`has_delta`(是否产生增量)、`stale_guarded` | 只有单卡接口 `GET /assets/:identifier/traffic-collections`,只能按时间分页 |
|
||||
| 停复机记录 | `tb_card_stop_resume_record` | `GET /card-stop-resume-records` | `identifier`、`carrier_id`、`action`、`source`、`scene`、`reason`、`result`、`error_category`、`command_id` | 已有跨卡接口,但只能用 ICCID 查卡,也不能按场景、原因、指令号筛选;单卡接口没有筛选 |
|
||||
| 套餐日流量详单 | `tb_package_usage_daily_record` | `GET /package-usage-daily-records` | `identifier`、`package_usage_id`、`package_id`、`carrier_id`、日期范围 | 只有单套餐接口 `GET /package-usages/:id/daily-records`,不分页 |
|
||||
| 卡日流量 | `tb_card_daily_usage` | `GET /card-daily-usages` | `identifier`、`carrier_id`、日期范围、`min_usage_mb`(用量下限,排查异常大流量) | 无接口 |
|
||||
| 套餐流量冲正流水 | `tb_package_usage_adjustment` | `GET /package-traffic-adjustments` | `identifier`、`package_usage_id`、操作人、调整方向(补扣或退回)、时间范围 | 只有单套餐接口,没有时间筛选 |
|
||||
| 对接日志 | `tb_integration_log` | 现有 `GET /audit/integrations` | 现有条件已覆盖 | 不改 |
|
||||
| 退款后处理尝试记录 | `tb_refund_post_process_attempt`(见 5.5) | `GET /refund-post-process-attempts` | `refund_id`、退款单号、`identifier`、`step`、`result`、`error_code`、`trigger`、时间范围 | 新增。退款列表同时增加 `asset_post_status` 筛选(可查"需人工处理"),列表和详情返回后处理次数、最后错误、下次重试时间 |
|
||||
|
||||
**索引**:按上面的筛选设计,需要时新建或重建。分区表的索引建在父表上,会自动应用到各个分区。
|
||||
|
||||
| 表 | 索引 |
|
||||
|---|---|
|
||||
| `tb_card_traffic_collection` | 保留 `(iot_card_id, observed_at DESC)`;新增冗余列 `carrier_id`,采集时写入;新增 `(observed_at DESC)`、`(carrier_id, observed_at DESC)`、`(decision, observed_at DESC)`;失败表索引见 5.10 |
|
||||
| `tb_card_stop_resume_record` | 现有 `(card_id, created_at)`、`(carrier_id, created_at)`、`(result, created_at)` 重建为倒序;新增 `(created_at DESC)`、`(scene, created_at DESC)`、`(command_id)` |
|
||||
| `tb_card_daily_usage` | 现有唯一键 `(iot_card_id, date)`;新增 `(date, usage_mb DESC)`,支撑查全部和按用量下限筛选 |
|
||||
| `tb_package_usage_daily_record` | 新增 `(package_usage_id, date DESC)`、`(date)` |
|
||||
| `tb_package_usage_adjustment` | 新增 `(iot_card_id, created_at DESC)`、`(created_at DESC)` |
|
||||
| `tb_refund_post_process_attempt` | `(refund_id, created_at DESC)`、`(created_at DESC)`、`(result, created_at DESC)`、`(error_code, created_at DESC)` |
|
||||
|
||||
其余低选择性的条件,如 `source`、`stale_guarded`、调整方向,和上面的索引组合使用即可,不单独建索引。上线后在测试环境看慢查询日志,按实际情况调整。
|
||||
|
||||
**说明**:
|
||||
|
||||
- **流量扣减结果**直接从采集记录查:每条采集记录都带 `decision`、`delta_mb`、`month_used_mb`、`event_id`,再结合日流量详单,能还原每次扣了多少。
|
||||
- **扣减效果表** `tb_card_observation_effect` 不对外提供接口。它只保存 outbox 事件的处理进度,没有卡 ID。排查扣减卡住时,在采集记录里按 `event_id` 关联查。
|
||||
- 新增接口都要在路由 `RouteSpec` 中声明输入、输出 DTO,并同步生成接口文档。
|
||||
|
||||
### 5.10 流量采集记录成功与失败分表
|
||||
|
||||
**现状**(测试库近 3 天,只读统计):
|
||||
|
||||
| 读数结果 | 扣减判定 | 条数 | 平均行大小 |
|
||||
|---|---|---|---|
|
||||
| `ok` | `written`(写入增量) | 104.2 万 | 432 B |
|
||||
| `ok` | `queue_not_full`(队列未满,暂不判断) | 17.3 万 | 285 B |
|
||||
| `ok` | `slope_negative`、`baseline`、`settled_existing` | 800 | 约 420 B |
|
||||
| `failed` | — | 14.5 万 | 136 B |
|
||||
| `no_reading` | — | 400 | 144 B |
|
||||
|
||||
- 成功和失败写在同一张 `tb_card_traffic_collection` 里,失败约占 11%,测试环境每天约 4.8 万条。
|
||||
- 失败记录只存了失败类型。`polling_carddata_handler` 在调用 `recordUnreadable` 时**丢掉了具体错误**,查不到失败原因,只能再去对接日志里按 `observation_id` 找。
|
||||
- 失败记录也只给"采集时刻有套餐"的卡写,为此每条失败都要额外查一次有效套餐。
|
||||
|
||||
**调整**:
|
||||
|
||||
| 表 | 写什么 | 保留 |
|
||||
|---|---|---|
|
||||
| `tb_card_traffic_collection`(现有) | 只写成功读数(`result=ok`),包括所有扣减判定;新增原始请求与响应字段 | 在线保留 2 个月,到期月分区导出为压缩包后删除(见下文"到期归档") |
|
||||
| `tb_card_traffic_collection_failure`(新增) | 只写失败:`failed`、`rate_limited`、`invalid_payload`、`no_reading`、`invalid_reading`;带原始请求与响应 | 与成功表相同:在线 2 个月,到期归档后删除 |
|
||||
|
||||
失败表字段:`iot_card_id`、`iccid`、`carrier_id`、`observed_at`、`observation_id`(等于对接日志的 `integration_id`,可直接关联)、`source`、`scene`、`result`(失败类型)、`error_category`(按网关错误分类:超时、限流、上游报错、响应格式错等)、`provider_code`、`provider_message`(截断到 500 字)、`duration_ms`、`created_at`,加上下面的原始数据字段。按 `observed_at` 按月分区,只新增不修改。
|
||||
|
||||
**原始数据**:成功表和失败表都保存运营商网关的原始请求与原始响应,有据可查。
|
||||
|
||||
| 字段 | 内容 |
|
||||
|---|---|
|
||||
| `raw_request` | 加密前的业务请求体(明文 JSON,只有卡号等查询参数),jsonb |
|
||||
| `http_status` | HTTP 状态码;没拿到响应(超时、连接失败)时为空 |
|
||||
| `gateway_code`、`gateway_msg`、`gateway_trace_id` | 网关响应外层的 `code`、`msg`、`trace_id` |
|
||||
| `raw_response` | 网关原始响应体,原样保存。能解析成 JSON 的存 jsonb,不能解析的按文本存在 `raw_response_text`。超过 4 KB 时截断,并标记 `raw_truncated=true` |
|
||||
| `gateway_attempts` | 网关客户端内部重试次数;保存的是最后一次的请求和响应 |
|
||||
|
||||
- **获取方式**:网关客户端(`internal/gateway/client.go` 的 `executeWithRetry`)已经读到了完整响应体和外层的 `code`、`msg`、`trace_id`,但只返回 `data`。新增按请求上下文挂载的原始数据收集器 `gateway.WithRawCapture(ctx)`,调用方挂上后,客户端把最后一次的请求体、状态码、响应体和重试次数写进收集器。没挂收集器的调用不受影响。
|
||||
- **写入**:`polling_carddata_handler` 调 `QueryFlow` 前挂上收集器,成功读数和各类失败都把收集器内容带进 `TrafficObservation`,写入对应的表。
|
||||
- **安全**:只存明文业务请求体,不存加密后的报文、签名、`app_id`、密钥和请求头。
|
||||
- **量的估算**:流量查询响应很小(`iccid`、`used`、`unit` 加外层字段,约 150–250 B),每行增加约 300–400 B。测试环境成功读数每天约 40 万条,每天增加约 120–160 MB;在线保留 2 个月,原始数据部分约 7–10 GB。失败每天约 4.8 万条,2 个月约 1 GB。部署后按实际行大小复核。
|
||||
|
||||
**到期归档**:
|
||||
|
||||
- **在线保留**:当月加上最近 2 个整月。月分区的月末早于"当前时间减 2 个月"时,判定为到期。例如 2027-01-01 起,2026-10 的分区到期。常量 `TrafficCollectionRetentionDays=90` 改为 `TrafficCollectionRetentionMonths=2`,成功表和失败表共用。
|
||||
- **执行位置**:沿用 leader worker 现有的分区维护循环 `startTrafficCollectionPartitionMaintainer`,每 12 小时运行一次。原来"到期直接 `DROP TABLE`"的逻辑,改为下面的四步:
|
||||
1. **摘出分区**:`ALTER TABLE ... DETACH PARTITION`,到期分区不再参与在线查询。
|
||||
2. **导出压缩包**:用 `COPY (SELECT * FROM 分区 ORDER BY observed_at, id) TO STDOUT WITH (FORMAT csv, HEADER)` 流式读出,边读边 gzip 写到 worker 运行目录下的 `archive/traffic-collection/` 目录,不在内存里攒整个月的数据。文件名为 `tb_card_traffic_collection_202610.csv.gz` 或 `tb_card_traffic_collection_failure_202610.csv.gz`。原始请求与响应以 JSON 文本形式保存在 CSV 列里。
|
||||
3. **校验**:写完后重新读一遍压缩包,统计行数并与分区 `count(*)` 比对;同时生成同名的 `.manifest.json`,记录表名、月份、行数、最早和最晚采集时间、文件大小、SHA-256 和导出时间。
|
||||
4. **删除分区**:行数一致后才 `DROP TABLE` 已摘出的分区。导出或校验失败时保留已摘出的表,记 Error 日志,下一轮重试;不会出现"数据删了但没有压缩包"的情况。
|
||||
- **文件同步**:压缩包和 manifest 由你们的同步服务器拉走。本地只保留最近 3 个月的压缩包,更早的自动删除,个数可配置。先写成 `.tmp`,校验通过后再改名为正式文件名,避免同步到写了一半的文件。
|
||||
- **测试环境挂载**:worker 容器增加 `./worker/archive:/app/archive`,宿主机目录为 `/opt/junhong_cmp/worker/archive/traffic-collection/`。
|
||||
- **压缩包大小估算**:测试环境成功表每月约 1,200 万行,CSV 约 8–10 GB,gzip 后约 1–1.5 GB;失败表每月约 150 万行,压缩后约 50 MB。
|
||||
- **回查**:要查归档月份的数据,解压后用 CSV 工具查看,或者临时 `COPY FROM` 导回一张临时表。线上接口只查在线的 2 个月。
|
||||
|
||||
|
||||
**写入规则**:
|
||||
|
||||
- 失败记录**不再判断是否有套餐**,所有轮询到的卡都写,省掉每条失败的有效套餐查询。运营商接口出问题和有没有套餐无关,排查时需要看全量。
|
||||
- `TrafficObservation` 增加失败详情(错误类别、上游返回码与信息、耗时),由 `polling_carddata_handler` 在网关调用失败时填入,不再丢弃错误。
|
||||
- 成功表仍在观测事务之外写入,规则不变;写入失败只告警,不影响扣减。
|
||||
- 成功表已有的失败历史数据保留,到期随分区删除;新数据按新规则分表写入。
|
||||
|
||||
**查询接口**(与 5.9 公共约定一致:可查单个也可查全部,分页,时间范围必填):
|
||||
|
||||
| 接口 | 筛选条件 | 索引 |
|
||||
|---|---|---|
|
||||
| `GET /traffic-collections`(成功) | `identifier`、`carrier_id`、`decision`、`source`、`scene`、`has_delta`、`stale_guarded` | 见 5.9 |
|
||||
| `GET /traffic-collection-failures`(失败) | `identifier`、`carrier_id`、`result`、`error_category`、`provider_code`、`source`、`scene` | `(iot_card_id, observed_at DESC)`、`(observed_at DESC)`、`(carrier_id, observed_at DESC)`、`(error_category, observed_at DESC)`、`(observation_id)` |
|
||||
|
||||
成功表不再有失败数据,所以 5.9 里成功表的 `result` 筛选和 `(result, observed_at DESC)` 索引都去掉。单卡接口 `GET /assets/:identifier/traffic-collections` 增加 `include_failures` 参数,默认不含失败;传了就把两张表合并后按时间倒序返回,方便在一张卡上连起来看。
|
||||
|
||||
## 6. 存量数据
|
||||
|
||||
- **不删除、不迁移**已有审计数据,由现有留存和归档任务按策略自然过期。
|
||||
- 如果要尽快释放空间,可以另行评估"按动作编码清理机器过程类历史数据"的一次性任务。本方案不包含它。
|
||||
|
||||
## 7. 分支与范围
|
||||
|
||||
- 基于 **Iteration/8-11** 实施。该分支已有 `tb_card_stop_resume_record`、采集记录和扣减效果表,方案 B 的补位表都存在。
|
||||
- **只部署测试环境**:8-11 推送后自动部署到测试环境;不在生产执行任何代码替换、配置修改或服务重启。生产什么时候、以什么方式合入,不在本方案范围内。
|
||||
- 全局写入开关 `add-audit-integration-write-switches` 已在 8-11 上,本方案与它互不替代:全局开关用于应急**全停**,本方案是常态下的**按动作分级**。
|
||||
|
||||
## 8. 实施步骤
|
||||
|
||||
| 步骤 | 内容 | 涉及 |
|
||||
|---|---|---|
|
||||
| 1 | 注册表增加 `Persist`,补齐全部机器动作的策略,并加上 5.2 的默认规则 | `internal/infrastructure/audit/registry.go`、`pkg/constants/audit.go` |
|
||||
| 2 | Writer 按策略分流:`outcome` 只写最终结论;`log_only` 写现有应用日志(`logger=audit`) | `internal/infrastructure/audit/writer.go` |
|
||||
| 3 | 对接日志去掉 `attempt_started` 补写 | `internal/infrastructure/integrationlog/repository.go` |
|
||||
| 4 | 退款后处理加重试上限和退避,新增尝试记录表;停机业务性失败不重投,调用前被拦截的写 `skipped` 停复机记录 | `internal/service/refund`、`internal/service/iot_card`、`migrations` |
|
||||
| 5 | 增加 `audit.persist_overrides` 运行时覆盖配置 | `pkg/config` |
|
||||
| 6 | 审计覆盖扫描和 OpenSpec 规格同步 | `internal/governance/auditcoverage`、`openspec/specs` |
|
||||
| 7 | 查询侧:卡、设备时间线合并停复机记录(方案 B) | `internal/query/audit` |
|
||||
| 8 | 测试环境 api、worker 各自挂载自己的 `logs/` 目录,轮换改为 40 个备份(见 5.8) | `docker-compose.prod.yml`、`.gitea/workflows/deploy.yaml` |
|
||||
| 9 | 记录查询接口:每类记录一个列表接口,可查单个也可查全部;按 5.9 的筛选条件建立或重建索引;采集记录增加 `carrier_id` 冗余列 | `internal/routes`、`internal/handler/admin`、`internal/model/dto`、`internal/store/postgres`、`migrations` |
|
||||
| 10 | 采集记录成功与失败分表:新增失败表,补全失败详情,失败不再判断有无套餐;两张表保存网关原始请求与响应(网关客户端新增原始数据收集器);提供失败查询接口(见 5.10) | `internal/gateway/client.go`、`internal/application/cardobservation`、`internal/task/polling_carddata_handler.go`、`internal/model`、`internal/store/postgres`、`migrations` |
|
||||
| 11 | 采集记录到期归档:分区维护改为摘出、导出 `csv.gz`、校验、删除四步;测试环境 worker 挂载 `archive` 目录 | `internal/store/postgres/card_traffic_collection_store.go`、`cmd/worker/main.go`、`pkg/constants`、`docker-compose.prod.yml` |
|
||||
|
||||
建议按"1–3 → 4 → 5–11"分三次提交,每次推送 8-11 后在测试环境验证。
|
||||
|
||||
## 9. 测试与验证(测试环境)
|
||||
|
||||
**测试环境逐项验证**(项目规定自动化测试为 N/A,按 ENG-TEST-001 在测试环境验证,以下各项是验证要点):
|
||||
|
||||
- 每种策略在 `account`、`openapi`、`system_task` 下的写入结果;
|
||||
- `outcome`:中间失败不写审计,尝试记录表每次都有一条;最终成功或转人工写一条审计,带总次数;同一件事重复写结论只保留一条;
|
||||
- `log_only` 不写表、文本日志有记录且不含快照;
|
||||
- 覆盖配置能把 `always` 降为 `log_only`,也能反过来恢复。
|
||||
|
||||
- 对接调用没有上游审计时,`tb_integration_log.audit_event_id` 为空,审计表无新增;
|
||||
- 退款后处理失败超过上限后状态为"需人工处理",outbox 不再重投。
|
||||
- 流量查询成功写成功表、各类失败写失败表,两张表都带原始请求与响应;失败记录有错误类别和上游返回码;无套餐的卡失败也写;
|
||||
- 网关响应超过 4 KB 时截断并标记;原始请求里没有密钥、签名或加密报文;
|
||||
- 到期归档:构造一个过期月分区,执行维护后生成 `csv.gz` 和 manifest,行数与 SHA-256 一致后分区被删除;模拟导出失败时,已摘出的表保留,下一轮重试成功;
|
||||
- 记录查询接口:每个接口都能查单卡、也能查全部,分页总数正确,各筛选条件生效。
|
||||
|
||||
**测试环境观察**(部署后 24 小时):
|
||||
|
||||
- 审计日增从约 12.6 万降到约 300 条(只读查询走 `created_at` 索引、按小时统计);
|
||||
- `app.log` 中出现 `logger=audit` 的行,轮换正常,日志同步服务器能收到并按 logger 名过滤;
|
||||
- 停复机记录、扣减效果写入量没有变化,业务结果没有变化;成功表加失败表的写入量合计,与改造前采集记录的量相当(失败表多出无套餐卡的失败);
|
||||
- 新接口在测试库数据量下没有慢查询;
|
||||
- `/opt/junhong_cmp/worker/archive/traffic-collection/` 目录可写,同步服务器能拉到文件。第一次真实归档要等到在线数据满 2 个月之后,在此之前用单元测试和集成测试覆盖。
|
||||
|
||||
## 10. 已定事项(2026-11-05)
|
||||
|
||||
| 事项 | 结论 |
|
||||
|---|---|
|
||||
| 方案 | 方案 B |
|
||||
| 范围 | 基于 Iteration/8-11,只部署测试环境 |
|
||||
| L3 文本日志 | 写入所在服务自己运行目录下的 `logs/app.log`,沿用现有轮换,由日志同步服务器归集 |
|
||||
| 测试环境日志保留 | api、worker 各自的 `logs/` 目录,各保留 40 个压缩备份,合计约 1 GB;部署后按实际备份大小微调 |
|
||||
| 退款后处理重试上限 | 10 次,指数退避,最长间隔 1 小时;超过后标记"需人工处理" |
|
||||
| 失败记录 | 每次尝试都记录在业务尝试记录表(退款后处理尝试、停复机记录),可分页查询;审计只记最终结论,不按首次失败去重 |
|
||||
| 停机业务性失败 | 按网关错误分类判定为不可重试,不再重投 |
|
||||
| 运行时覆盖 | 提供 `audit.persist_overrides`,可按动作临时调整策略 |
|
||||
| 存量审计数据 | 不删除、不迁移,按现有留存策略自然过期 |
|
||||
| 采集记录分表 | 成功写现有表,失败写新表 `tb_card_traffic_collection_failure`,带完整错误详情,所有卡都写;两张表都保存网关原始请求与响应 |
|
||||
| 采集记录保留 | 两张表在线保留 2 个月(当月加最近 2 个整月);到期月分区先导出为 `csv.gz` 压缩包并校验,再删除;压缩包由同步服务器拉走,本地保留 3 个月 |
|
||||
| 记录查询 | 所有独立成表的记录都提供分页加条件筛选接口,可查单个也可查全部;筛选条件按排查需要来定,索引随之新建或重建,清单见 5.9;文本日志不提供接口 |
|
||||
| 页面依赖 | 实施第 7 步(时间线合并停复机记录)前,与前端或运营确认是否有页面依赖机器动作审计 |
|
||||
@@ -13,6 +13,8 @@ import (
|
||||
"strings"
|
||||
|
||||
"github.com/bytedance/sonic"
|
||||
|
||||
auditinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
||||
)
|
||||
|
||||
// Entry 是一个必须经过审计分类的 HTTP、Worker 或定时任务入口。
|
||||
@@ -46,6 +48,9 @@ type Entry struct {
|
||||
|
||||
// Scan 扫描当前仓库中对外 HTTP、Asynq Worker 和定时任务注册入口。
|
||||
func Scan(root string) ([]Entry, error) {
|
||||
if missing := auditinfra.NewRegistry().MachineActionsWithoutPolicy(); len(missing) > 0 {
|
||||
return nil, fmt.Errorf("以下可由系统任务或定时任务发起的审计动作未声明落库策略:%s", strings.Join(missing, ", "))
|
||||
}
|
||||
var entries []Entry
|
||||
files := []string{
|
||||
"internal/routes", "internal/application", "internal/domain", "internal/service",
|
||||
@@ -201,7 +206,7 @@ func classifyWorker(file string, line int, taskType, handler string) Entry {
|
||||
entry := Entry{
|
||||
Key: fmt.Sprintf("worker:%s:%d:%s", file, line, taskType), Kind: "worker",
|
||||
CodeEntry: fmt.Sprintf("%s:%d %s", file, line, handler), Owner: owner,
|
||||
Summary: "处理异步任务 " + taskType, AuditEvent: "按状态变化、人工触发、连续失败或高风险异常决定",
|
||||
Summary: "处理异步任务 " + taskType, AuditEvent: "按审计注册表落库策略决定:系统任务未声明策略时只写应用日志",
|
||||
DomainLedger: ledgerDecision(owner), IntegrationLog: workerIntegrationDecision(taskType),
|
||||
Outbox: "任务来源 Outbox/业务任务事实;消费端按稳定事件或任务 ID 幂等",
|
||||
ActionCode: actionCode(owner, handler), ActionName: "处理异步任务(" + taskType + ")",
|
||||
|
||||
130
internal/infrastructure/audit/persist_decision.go
Normal file
130
internal/infrastructure/audit/persist_decision.go
Normal file
@@ -0,0 +1,130 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/logger"
|
||||
"github.com/break/junhong_cmp_fiber/pkg/sanitizer"
|
||||
)
|
||||
|
||||
// persistDecision 是一次审计写入的去向。
|
||||
type persistDecision int
|
||||
|
||||
const (
|
||||
decisionWrite persistDecision = iota
|
||||
decisionLog
|
||||
decisionSkip
|
||||
)
|
||||
|
||||
// auditLoggerName 是应用日志中审计行的固定 logger 名,日志同步端按它过滤。
|
||||
const auditLoggerName = "audit"
|
||||
|
||||
// persistOverrides 是运行时按动作覆盖的落库策略(只作用于系统任务、定时任务操作者)。
|
||||
var persistOverrides atomic.Pointer[map[string]PersistPolicy]
|
||||
|
||||
// SetPersistOverrides 解析并设置运行时覆盖,格式为 "动作编码=策略,动作编码=策略";返回被忽略的非法项。
|
||||
func SetPersistOverrides(raw string) []string {
|
||||
overrides := map[string]PersistPolicy{}
|
||||
var invalid []string
|
||||
for _, item := range strings.Split(raw, ",") {
|
||||
item = strings.TrimSpace(item)
|
||||
if item == "" {
|
||||
continue
|
||||
}
|
||||
code, policy, ok := strings.Cut(item, "=")
|
||||
code, policy = strings.TrimSpace(code), strings.TrimSpace(policy)
|
||||
if !ok || code == "" || !validPersistPolicy(PersistPolicy(policy)) {
|
||||
invalid = append(invalid, item)
|
||||
continue
|
||||
}
|
||||
overrides[code] = PersistPolicy(policy)
|
||||
}
|
||||
persistOverrides.Store(&overrides)
|
||||
return invalid
|
||||
}
|
||||
|
||||
// effectivePolicy 返回机器操作者下动作实际采用的策略:运行时覆盖 > 注册表声明 > 默认只写应用日志。
|
||||
func effectivePolicy(action ActionDefinition) PersistPolicy {
|
||||
if overrides := persistOverrides.Load(); overrides != nil {
|
||||
if policy, ok := (*overrides)[action.Code]; ok {
|
||||
return policy
|
||||
}
|
||||
}
|
||||
if action.Persist != "" {
|
||||
return action.Persist
|
||||
}
|
||||
return PersistLogOnly
|
||||
}
|
||||
|
||||
// decidePersist 按"动作 × 操作者"决定写表、写应用日志还是跳过;人工与外部操作者始终写表。
|
||||
func decidePersist(action ActionDefinition, input AppendInput) persistDecision {
|
||||
if !isMachineActor(input.Actor.Kind) {
|
||||
return decisionWrite
|
||||
}
|
||||
switch effectivePolicy(action) {
|
||||
case PersistAlways:
|
||||
return decisionWrite
|
||||
case PersistOnSuccess:
|
||||
if input.Result == constants.AuditResultSuccess {
|
||||
return decisionWrite
|
||||
}
|
||||
return decisionLog
|
||||
case PersistOutcome:
|
||||
if input.Final {
|
||||
return decisionWrite
|
||||
}
|
||||
return decisionLog
|
||||
case PersistNone:
|
||||
return decisionSkip
|
||||
default:
|
||||
return decisionLog
|
||||
}
|
||||
}
|
||||
|
||||
// writeAuditLog 把未落表的审计写成一行应用日志;只含标识类字段,不含快照与主体数据。
|
||||
func writeAuditLog(ctx context.Context, action ActionDefinition, input AppendInput) {
|
||||
resourceType, resourceKey := primaryResource(input)
|
||||
fields := []zap.Field{
|
||||
zap.String("action_code", action.Code),
|
||||
zap.String("actor_kind", input.Actor.Kind),
|
||||
zap.String("actor_id", input.Actor.ID),
|
||||
zap.String("source", input.Source),
|
||||
zap.String("result", input.Result),
|
||||
zap.String("resource_type", resourceType),
|
||||
zap.String("resource_key", resourceKey),
|
||||
zap.String("correlation_id", input.CorrelationID),
|
||||
zap.String("request_id", input.RequestID),
|
||||
zap.String("error_code", input.ErrorCode),
|
||||
zap.String("summary", sanitizer.SanitizeText(input.Summary)),
|
||||
}
|
||||
if input.Attempt > 0 {
|
||||
fields = append(fields, zap.Int("attempt", input.Attempt))
|
||||
}
|
||||
if input.AttemptRecordID != "" {
|
||||
fields = append(fields, zap.String("attempt_record_id", input.AttemptRecordID))
|
||||
}
|
||||
auditLogger := logger.GetAppLogger().Named(auditLoggerName)
|
||||
if input.Result == constants.AuditResultSuccess {
|
||||
auditLogger.Info("audit_event", fields...)
|
||||
return
|
||||
}
|
||||
auditLogger.Warn("audit_event", fields...)
|
||||
}
|
||||
|
||||
// primaryResource 返回主资源的类型与键,没有主资源时取第一条资源。
|
||||
func primaryResource(input AppendInput) (string, string) {
|
||||
for _, resource := range input.Resources {
|
||||
if resource.Relation == constants.AuditResourceRelationPrimary {
|
||||
return resource.Type, resource.Key
|
||||
}
|
||||
}
|
||||
if len(input.Resources) > 0 {
|
||||
return input.Resources[0].Type, input.Resources[0].Key
|
||||
}
|
||||
return "", ""
|
||||
}
|
||||
250
internal/infrastructure/audit/persist_policy.go
Normal file
250
internal/infrastructure/audit/persist_policy.go
Normal file
@@ -0,0 +1,250 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"sort"
|
||||
|
||||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||||
)
|
||||
|
||||
// PersistPolicy 决定系统任务、定时任务发起的审计动作是否写入审计表。
|
||||
// 后台账号、个人客户、开放接口、外部系统发起的动作始终写表,不受该策略约束。
|
||||
type PersistPolicy string
|
||||
|
||||
const (
|
||||
// PersistAlways 成功、失败都写审计表。
|
||||
PersistAlways PersistPolicy = "always"
|
||||
// PersistOnSuccess 只写成功;失败写应用日志。
|
||||
PersistOnSuccess PersistPolicy = "on_success"
|
||||
// PersistOutcome 只写最终结论(AppendInput.Final=true);中间失败写应用日志,逐次细节由业务尝试记录保存。
|
||||
PersistOutcome PersistPolicy = "outcome"
|
||||
// PersistLogOnly 不写审计表,只写所在服务的应用日志。
|
||||
PersistLogOnly PersistPolicy = "log_only"
|
||||
// PersistNone 不写审计表,也不写应用日志。
|
||||
PersistNone PersistPolicy = "none"
|
||||
)
|
||||
|
||||
// validPersistPolicy 判断策略取值是否合法。
|
||||
func validPersistPolicy(policy PersistPolicy) bool {
|
||||
switch policy {
|
||||
case PersistAlways, PersistOnSuccess, PersistOutcome, PersistLogOnly, PersistNone:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// machinePersistPolicies 是可由系统任务或定时任务发起的动作的落库策略。
|
||||
// 只登记资金、套餐生命周期与业务状态变化等需要留在审计表的动作,以及明确不写的动作;
|
||||
// 未登记的机器动作按 PersistLogOnly 写应用日志(见 docs/2026-11-05-审计日志写入改造技术方案.md 4.1、4.2)。
|
||||
var machinePersistPolicies = func() map[string]PersistPolicy {
|
||||
policies := map[string]PersistPolicy{
|
||||
// 对接调用开始:与 tb_integration_log 完全重复,不写。
|
||||
constants.AuditActionIntegrationAttemptStarted: PersistNone,
|
||||
// 退款资产后处理:只写最终结论(成功或转人工),逐次尝试见 tb_refund_post_process_attempt。
|
||||
"refund.process_asset": PersistOutcome,
|
||||
}
|
||||
// 资金、审批、订单、套餐生命周期、导入与资产归属等业务状态变化:留在审计表。
|
||||
for _, code := range []string{
|
||||
"agent_distribution.registration_approve",
|
||||
"agent_distribution.registration_reject",
|
||||
"agent_recharge.close",
|
||||
"agent_recharge.create",
|
||||
"agent_recharge.credit",
|
||||
"agent_wallet.order_complete",
|
||||
"agent_wallet.order_debit",
|
||||
"agent_wallet.order_release",
|
||||
"agent_wallet.order_reserve",
|
||||
"approval.recover_submission",
|
||||
"approval.sync_decision",
|
||||
"approval.sync_submission",
|
||||
"asset.first_purchase_ended",
|
||||
"asset.first_purchase_recorded",
|
||||
"asset_auto_renewal.failed",
|
||||
"asset_auto_renewal.renewed",
|
||||
"asset_package_batch_order_task.complete",
|
||||
"asset_recharge.auto_purchase",
|
||||
"audit.retention_cleanup",
|
||||
"commission.calculate",
|
||||
"commission.credit",
|
||||
"commission_withdrawal.anomaly_flag",
|
||||
"commission_withdrawal.approve",
|
||||
"commission_withdrawal.attempt_approve",
|
||||
"commission_withdrawal.attempt_close",
|
||||
"commission_withdrawal.attempt_reject",
|
||||
"commission_withdrawal.attempt_submit",
|
||||
"commission_withdrawal.reject",
|
||||
"commission_withdrawal.request",
|
||||
"device.allocate",
|
||||
"device.allocate_batch",
|
||||
"device.create",
|
||||
"device.recall",
|
||||
"device.recall_batch",
|
||||
"device.series_binding",
|
||||
"device.series_binding_batch",
|
||||
"device.update_polling_status",
|
||||
"device_import_task.complete",
|
||||
"employee_collection.application.approve",
|
||||
"employee_collection.application.reject",
|
||||
"employee_collection.application.revoke",
|
||||
"employee_collection.application.submit",
|
||||
"employee_collection.bill.close",
|
||||
"employee_collection.bill.create",
|
||||
"employee_collection.bill.refund_offset",
|
||||
"integration.inbound_received",
|
||||
// 自动停复机:系统自动发起的已由 StopResumeService.commandAuditRequired 改写停复机记录,
|
||||
// 仍写审计的只剩通道阈值、自动续费等豁免链路(gateway-call-observability 要求保留)。
|
||||
"iot_card.auto_start",
|
||||
"iot_card.auto_stop",
|
||||
"iot_card.batch_update_polling_status",
|
||||
"iot_card.create",
|
||||
"iot_card.terminal_disable_polling",
|
||||
"iot_card.update_polling_status",
|
||||
"iot_card_import_task.complete",
|
||||
"order.cancel",
|
||||
"order.create",
|
||||
"order.expire_close",
|
||||
"order.online_pay",
|
||||
"order.wallet_pay",
|
||||
"order_package_invalidate_task.complete",
|
||||
"order_package_invalidate_task.item",
|
||||
"package.silence_default_changed",
|
||||
"package.silence_switch_changed",
|
||||
"package_silence_policy.created",
|
||||
"package_silence_policy.status_changed",
|
||||
"package_silence_policy.updated",
|
||||
"package_traffic_alert.trigger",
|
||||
"package_usage.activate",
|
||||
"package_usage.adjust_traffic",
|
||||
"package_usage.expire",
|
||||
"package_usage.invalidate_asset",
|
||||
"package_usage.invalidate_refund",
|
||||
"package_usage.silence_adjust_approved",
|
||||
"package_usage.silence_adjust_rejected",
|
||||
"package_usage.silence_adjust_requested",
|
||||
"package_usage.silence_backfilled",
|
||||
"package_usage.silence_ended",
|
||||
"package_usage.silence_expired",
|
||||
"package_usage.silence_extension_applied",
|
||||
"package_usage.silence_extension_approved",
|
||||
"package_usage.silence_extension_exception",
|
||||
"package_usage.silence_extension_order_changed",
|
||||
"package_usage.silence_extension_rejected",
|
||||
"package_usage.silence_extension_requested",
|
||||
"package_usage.silence_migrated_on_exchange",
|
||||
"package_usage.silence_set",
|
||||
"package_usage.update_expires_at",
|
||||
"payment.confirm",
|
||||
"payment.create",
|
||||
"payment.fail",
|
||||
"phone_asset_association.import_unbind",
|
||||
"phone_asset_unbind_import_task.complete",
|
||||
"refund.anomaly_flag",
|
||||
"refund.approve",
|
||||
"refund.attempt_approve",
|
||||
"refund.attempt_close",
|
||||
"refund.channel_call",
|
||||
"refund.channel_recover",
|
||||
"refund.clawback_commission",
|
||||
"refund.clawback_not_required",
|
||||
"refund.invalidate_commission",
|
||||
"refund.reject",
|
||||
"shop.import_business_owner",
|
||||
"shop_business_owner_import_task.complete",
|
||||
"withdrawal_qualification.approve",
|
||||
"withdrawal_qualification.invalidate",
|
||||
"withdrawal_qualification.reject",
|
||||
"withdrawal_qualification.submit",
|
||||
"withdrawal_qualification.submit_reject",
|
||||
"withdrawal_qualification.void",
|
||||
} {
|
||||
policies[code] = PersistAlways
|
||||
}
|
||||
// 过程类动作:写应用日志。采集与扣减见采集记录与日流量,流量重置见套餐重置时间与日流量。
|
||||
for _, code := range []string{
|
||||
"device.worker_observation_sync",
|
||||
"iot_card.auto_stop_reason_update",
|
||||
"iot_card.worker_network_sync",
|
||||
"iot_card.worker_realname_sync",
|
||||
"iot_card.worker_traffic_sync",
|
||||
"notification.cleanup",
|
||||
"notification.cleanup_item",
|
||||
"notification.deliver",
|
||||
"package_usage.deduct_traffic",
|
||||
"package_usage.reset_traffic",
|
||||
"polling_priority.claim",
|
||||
"polling_priority.complete",
|
||||
"polling_priority.dequeue",
|
||||
"polling_priority.enqueue",
|
||||
"polling_priority.expire",
|
||||
"polling_priority.fail",
|
||||
"polling_priority.retry",
|
||||
} {
|
||||
policies[code] = PersistLogOnly
|
||||
}
|
||||
return policies
|
||||
}()
|
||||
|
||||
// applyPersistPolicies 把机器动作落库策略写入已注册的动作定义。
|
||||
func applyPersistPolicies(registry *Registry, policies map[string]PersistPolicy) {
|
||||
for code, policy := range policies {
|
||||
action, ok := registry.actionsByCode[code]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
action.Persist = policy
|
||||
registry.actionsByCode[code] = action
|
||||
for operation, byOperation := range registry.actionsByOperation {
|
||||
if byOperation.Code == code {
|
||||
byOperation.Persist = policy
|
||||
registry.actionsByOperation[operation] = byOperation
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// isMachineActor 判断操作者是否为系统任务或定时任务。
|
||||
func isMachineActor(actorKind string) bool {
|
||||
return actorKind == constants.AuditActorSystemTask || actorKind == constants.AuditActorScheduledJob
|
||||
}
|
||||
|
||||
// allowsMachineOrigin 判断动作是否允许由系统任务或定时任务发起。
|
||||
func allowsMachineOrigin(action ActionDefinition) bool {
|
||||
if isMachineActor(action.AllowedActor) {
|
||||
return true
|
||||
}
|
||||
for _, origin := range action.AllowedOrigins {
|
||||
if isMachineActor(origin.Actor) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// MachineActionsWithoutPolicy 返回可由系统任务或定时任务发起、但未声明落库策略的动作编码(审计覆盖扫描使用)。
|
||||
func (r *Registry) MachineActionsWithoutPolicy() []string {
|
||||
if r == nil {
|
||||
return nil
|
||||
}
|
||||
var codes []string
|
||||
for code, action := range r.actionsByCode {
|
||||
if allowsMachineOrigin(action) && action.Persist == "" {
|
||||
codes = append(codes, code)
|
||||
}
|
||||
}
|
||||
sort.Strings(codes)
|
||||
return codes
|
||||
}
|
||||
|
||||
// MachineActions 返回可由系统任务或定时任务发起的全部动作及其声明的策略(空表示未声明)。
|
||||
func (r *Registry) MachineActions() map[string]PersistPolicy {
|
||||
if r == nil {
|
||||
return nil
|
||||
}
|
||||
actions := map[string]PersistPolicy{}
|
||||
for code, action := range r.actionsByCode {
|
||||
if allowsMachineOrigin(action) {
|
||||
actions[code] = action.Persist
|
||||
}
|
||||
}
|
||||
return actions
|
||||
}
|
||||
@@ -18,6 +18,8 @@ type ActionDefinition struct {
|
||||
SubjectFields []string
|
||||
SensitiveRead bool
|
||||
AllowedOrigins []ActionOrigin
|
||||
// Persist 是系统任务、定时任务发起时的落库策略,空表示未声明(按 PersistLogOnly 处理)。
|
||||
Persist PersistPolicy
|
||||
}
|
||||
|
||||
// ActionOrigin 定义动作允许的操作者与入口组合。
|
||||
@@ -1185,6 +1187,7 @@ func NewRegistry() *Registry {
|
||||
},
|
||||
}
|
||||
registerPackageSilenceAudit(registry)
|
||||
applyPersistPolicies(registry, machinePersistPolicies)
|
||||
return registry
|
||||
}
|
||||
|
||||
|
||||
@@ -812,6 +812,11 @@ type AppendInput struct {
|
||||
FailCount int
|
||||
Metadata map[string]any
|
||||
Resources []ResourceInput
|
||||
// Final 只对"只写结论"策略生效:true 表示本次是最终结论(成功或重试用尽转人工),写审计表;false 只写应用日志。
|
||||
Final bool
|
||||
// Attempt 与 AttemptRecordID 是自动重试类动作的尝试序号与业务尝试记录标识,只写入应用日志。
|
||||
Attempt int
|
||||
AttemptRecordID string
|
||||
}
|
||||
|
||||
// Writer 只提供不可变 Audit Event 追加能力。
|
||||
@@ -956,6 +961,13 @@ func (w *Writer) AppendAndGet(ctx context.Context, tx *gorm.DB, input AppendInpu
|
||||
if !validResult(input.Result) || len(input.Resources) == 0 {
|
||||
return nil, pkgerrors.New(pkgerrors.CodeInvalidParam, "审计结果非法或缺少资源")
|
||||
}
|
||||
switch decidePersist(action, input) {
|
||||
case decisionSkip:
|
||||
return nil, nil
|
||||
case decisionLog:
|
||||
writeAuditLog(ctx, action, input)
|
||||
return nil, nil
|
||||
}
|
||||
resources, err := w.buildResources(input.Resources, action)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -115,9 +115,7 @@ func (r *Repository) Start(ctx context.Context, input Attempt) (*model.Integrati
|
||||
if input.AuditEventID == nil && input.TriggerSeries != nil {
|
||||
input.AuditEventID = r.auditEventIDForSeries(ctx, *input.TriggerSeries)
|
||||
}
|
||||
if input.AuditEventID == nil {
|
||||
input.AuditEventID = r.recordAuditEvent(ctx, constants.AuditActionIntegrationAttemptStarted, input.IntegrationID, input.Provider, input.Direction, input.Operation, input.ResourceType, input.ResourceID, input.ResourceKey, input.CorrelationID)
|
||||
}
|
||||
// 没有上游审计事件时 audit_event_id 留空,不再补写 integration.attempt_started(与本表内容重复)。
|
||||
autoAttempt := input.Attempt <= 0 && input.TriggerSeries != nil
|
||||
if input.Attempt <= 0 {
|
||||
input.Attempt = 1
|
||||
@@ -321,6 +319,10 @@ func (r *Repository) recordAuditEvent(ctx context.Context, actionCode, integrati
|
||||
zap.L().Warn("Integration Log 缺少审计关联", zap.String("integration_id", integrationID), zap.Error(err))
|
||||
return nil
|
||||
}
|
||||
if event == nil {
|
||||
// 落库策略决定不写审计表(只写应用日志或不写)时没有事件可关联。
|
||||
return nil
|
||||
}
|
||||
return &event.ID
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
schema: spec-driven
|
||||
created: 2026-11-05
|
||||
261
openspec/changes/refactor-audit-log-persistence/design.md
Normal file
261
openspec/changes/refactor-audit-log-persistence/design.md
Normal file
@@ -0,0 +1,261 @@
|
||||
## Context
|
||||
|
||||
技术方案见 `docs/2026-11-05-审计日志写入改造技术方案.md`(方案 B)。本设计基于 `Iteration/8-11` 的现状。
|
||||
|
||||
**审计写入**
|
||||
|
||||
- 写入链路:业务调用 `audit.Writer.Append` / `AppendAndGet`(`internal/infrastructure/audit/writer.go:935`),经注册表 `Registry` 校验动作与操作者来源后,写入 `tb_audit_event` 和 `tb_audit_event_resource`。
|
||||
- 写入失败已经被吞掉,记为二次失败,不影响业务。
|
||||
- `AppendAndGet` 共 11 处调用,大多是人工或资金类动作。
|
||||
|
||||
**对接日志**
|
||||
|
||||
- `integrationlog.Repository.Start`(`internal/infrastructure/integrationlog/repository.go:118`)在没有上游审计事件时,会补写一条 `integration.attempt_started` 审计并回填 `audit_event_id`。
|
||||
|
||||
**停复机**
|
||||
|
||||
- 8-11 上,退款停机在停复机指令队列开启时只请求停机评估(`internal/service/refund/service.go` 的 `stopAsset`)。
|
||||
- 停机失败已由指令队列按错误分类处理:终态不重试,限流按退避重试。
|
||||
- 停复机调用记录写在 `tb_card_stop_resume_record`,每次实际调用上游写一条,结果枚举已含 `skipped`,但目前没有写入点。
|
||||
|
||||
**退款后处理**
|
||||
|
||||
- `processRefundAsset` 依次执行:套餐失效、接续激活、判断有无生效主套餐、停机评估,最后标记完成。
|
||||
- 任一步失败都会触发 Outbox 重投和 `RecoverRefundPostProcessing` 补偿扫描,二者都没有次数上限和退避。
|
||||
|
||||
**流量采集**
|
||||
|
||||
- `polling_carddata_handler` 调用 `gateway.QueryFlow`。失败时 `recordUnreadable` 只传失败类型,具体错误被丢弃。
|
||||
- 采集记录写在 `tb_card_traffic_collection`:成功和失败同表,只写有套餐的卡。
|
||||
- 表按月分区,由 leader worker 的 `startTrafficCollectionPartitionMaintainer` 每 12 小时维护一次,到期分区直接 `DROP`。
|
||||
- 网关客户端 `executeWithRetry` 已经读到完整响应体以及外层的 `code`、`msg`、`trace_id`,但只返回 `data`。
|
||||
|
||||
**记录查询**
|
||||
|
||||
- 已有的列表接口:采集记录(单卡)、停复机记录(跨卡和单卡)、冲正流水(单套餐)、日流量详单(单套餐,不分页)、对接日志(完整)。
|
||||
|
||||
**测试环境**
|
||||
|
||||
- 由 `docker-compose.prod.yml` 部署,api 和 worker 都把宿主机 `./logs` 挂载到 `/app/logs`,共用同一个 `app.log`。
|
||||
- 日志级别为 info,格式为 JSON。
|
||||
|
||||
## Goals / Non-Goals
|
||||
|
||||
**Goals:**
|
||||
|
||||
- 审计表只记人工操作,以及资金、套餐生命周期的状态变化。测试环境日增从约 12.6 万降到约 300 条。
|
||||
- 每一次失败都有记录可查:采集失败、停复机跳过、退款后处理尝试。
|
||||
- 运营商原始请求与响应有据可查。
|
||||
- 独立记录都能按单个或全部、分页、按条件查询。
|
||||
|
||||
**Non-Goals:**
|
||||
|
||||
- 不修改生产环境的代码、配置与服务。
|
||||
- 不删除、不迁移已有审计数据。
|
||||
- 不改 `add-audit-integration-write-switches` 的全局开关语义。
|
||||
- 不改停复机指令队列的重试与错误分类规则。
|
||||
- 不调整测试环境的日志级别。
|
||||
|
||||
## Decisions
|
||||
|
||||
### D1 落库策略放在注册表,按"动作 × 操作者"决定
|
||||
|
||||
在 `ActionDefinition` 上新增 `Persist` 字段:`always`、`on_success`、`outcome`、`log_only`、`none`。`AppendAndGet` 完成注册与来源校验后,由 `persistDecision` 决定写表、写日志还是跳过。
|
||||
|
||||
- 人工操作者(`account`、`personal_customer`、`openapi`、`external_system`)一律写表。
|
||||
- 系统任务和定时任务按 `Persist` 处理,未声明时按 `log_only`。
|
||||
- `persist_overrides` 运行时配置只覆盖系统任务和定时任务操作者。
|
||||
- `iot_card.auto_start` / `auto_stop` 声明为 `always`:8-11 上系统自动停复机已由 `commandAuditRequired` 改为只写停复机记录,仍会写这两个动作的只剩通道阈值、自动续费等豁免链路,`gateway-call-observability` 要求这些链路保留审计。
|
||||
|
||||
**为什么这样做**:
|
||||
- 调用方不改代码,所有动作的去向集中在一处,评审只需看注册表。
|
||||
- 默认 `log_only` 能防止新增的机器动作再次写满审计表。
|
||||
|
||||
**备选**:在各调用方删除 `Append` 调用。
|
||||
- 改动面大,散落多处,后续新增动作也容易复发,因此不采用。
|
||||
|
||||
### D2 `log_only` 写所在进程的应用日志
|
||||
|
||||
使用 `logger.GetAppLogger().Named("audit")`,写一行 JSON,`msg=audit_event`。不新建日志文件,`NewWriter` 签名不变。日志里只放标识类字段,不写快照。
|
||||
|
||||
- 轮换沿用 `logging.app_log`,由日志同步服务器归集。
|
||||
|
||||
**备选**:单独建 `audit-machine.log`。
|
||||
- 需要额外配置轮换和同步路径,用户已明确要求写在服务自己的日志目录,因此不采用。
|
||||
|
||||
### D3 `outcome` 由调用方标明最终结论
|
||||
|
||||
`AppendInput` 增加 `Final bool`:
|
||||
|
||||
- `Final=false`:只写应用日志,带尝试序号和尝试记录 ID。
|
||||
- `Final=true`:写表。`EventID` 使用业务键,例如 `refund-asset:{refund_id}:outcome`,依靠现有的 `ON CONFLICT (event_id) DO NOTHING` 保证同一结论只有一条。
|
||||
|
||||
**备选**:Writer 自己按次数判断是否为最终结论。
|
||||
- Writer 不知道业务上限,判断会出错,因此由调用方决定。
|
||||
|
||||
**备选**:只写首次失败。
|
||||
- 用户否决:只记首次失败很难排查问题。
|
||||
|
||||
### D4 去掉对接日志的补写审计
|
||||
|
||||
删除 `Start` 中 `AuditEventID == nil` 时调用 `recordAuditEvent` 的分支;注册表中 `integration.attempt_started` 设为 `none` 兜底。
|
||||
|
||||
- `auditEventIDForSeries` 保留。实施时确认它按 `trigger_series` 回查审计的 SQL 走索引。
|
||||
- 同步修改 `operations-audit` 中"新外部交互日志必须有非空 `audit_event_id`"的场景。
|
||||
|
||||
### D5 退款后处理:尝试表 + 退款单状态字段
|
||||
|
||||
**新表 `tb_refund_post_process_attempt`**
|
||||
|
||||
- 按 `created_at` 按月分区,只新增。
|
||||
- 每个步骤失败或整体成功时各写一条记录。
|
||||
|
||||
**退款单新增字段**
|
||||
|
||||
- `asset_post_status`:处理中、已完成、需人工处理;
|
||||
- `asset_post_attempts`;
|
||||
- `asset_post_last_error`;
|
||||
- `asset_post_next_at`。
|
||||
|
||||
现有的 `asset_reset` 完成标记保留,并与 `asset_post_status=已完成` 同步写入,保证兼容。
|
||||
|
||||
**重试与人工处理**
|
||||
|
||||
- 失败时 `next_at = now + min(2^(n-1) 分钟, 60 分钟)`。
|
||||
- `n ≥ 10` 时置为需人工处理,写 `outcome` 失败结论审计。
|
||||
- Outbox 处理器和 `RecoverRefundPostProcessing` 在执行前检查状态与 `next_at`:
|
||||
- 未到重试时间:处理器返回成功,不再重投,由补偿扫描在到期后拉起;
|
||||
- 需人工处理:直接跳过。
|
||||
- 新增后台人工重试接口 `POST /refunds/:id/asset-post-processing/retry`,记为人工重试,只允许在需人工处理状态下发起。
|
||||
|
||||
**备选**:只在退款单上记最后一次错误。
|
||||
- 每次尝试的细节会丢失,用户要求每次失败都可查,因此不采用。
|
||||
|
||||
### D6 停复机跳过记录写在指令执行入口
|
||||
|
||||
在指令执行前的重新判断处写 `skipped` 记录,覆盖三种结束原因:已是目标状态、条件不再满足、卡不允许停复机。
|
||||
|
||||
- 限流推迟不写记录,沿用现有分钟计数。
|
||||
- 记录写入失败只告警。
|
||||
|
||||
### D7 采集记录分表,并由网关原始数据收集器补齐原始数据
|
||||
|
||||
**原始数据收集器**
|
||||
|
||||
- 新增 `gateway.WithRawCapture(ctx) (*RawCapture, context.Context)`。
|
||||
- `executeWithRetry` 在每次尝试后,用最后一次的明文请求、HTTP 状态码、响应体、外层 `code` / `msg` / `trace_id` 和尝试次数覆盖收集器。
|
||||
- 只记录加密前的业务 JSON。不挂收集器的调用没有额外开销。
|
||||
|
||||
**采集写入**
|
||||
|
||||
- `polling_carddata_handler` 调用 `QueryFlow` 前挂上收集器。
|
||||
- `TrafficObservation` 增加 `Raw *gateway.RawCapture` 和 `Failure`(错误类别、上游码与信息、耗时)。
|
||||
- `writeCollectionRecord` 按结果分流:
|
||||
- 成功:保持"有套餐才写"的规则;
|
||||
- 失败:直接写失败表,不查套餐。
|
||||
|
||||
**原始数据列**(两张表相同)
|
||||
|
||||
- `raw_request` jsonb
|
||||
- `http_status` int
|
||||
- `gateway_code`、`gateway_msg`、`gateway_trace_id`
|
||||
- `raw_response` jsonb 或 `raw_response_text` text
|
||||
- `raw_truncated` bool
|
||||
- `gateway_attempts` int
|
||||
|
||||
**分表上线后的查询**
|
||||
|
||||
- 采集记录表增加 `carrier_id` 冗余列,在采集时写入。
|
||||
- 新数据写入后,成功表只会有成功记录。历史上写在成功表里的失败数据不迁移,随分区过期。
|
||||
- 成功表接口按 `result='ok'` 过滤,避免历史失败数据混入。
|
||||
|
||||
**备选**:原始数据只放对接日志。
|
||||
- 现有规格要求普通轮询查询不逐条写对接日志,而且对接日志不按卡分区,查询成本高,因此不采用。
|
||||
|
||||
### D8 到期归档:摘出、导出、校验、删除
|
||||
|
||||
改写 `MaintainPartitions`,成功表和失败表共用同一流程,保留期常量改为 `TrafficCollectionRetentionMonths = 2`。
|
||||
|
||||
1. **摘出分区**:执行 `ALTER TABLE ... DETACH PARTITION`。
|
||||
2. **导出**:通过 pgx 的 `CopyTo` 执行 `COPY (SELECT * FROM <分区> ORDER BY observed_at, id) TO STDOUT WITH (FORMAT csv, HEADER)`,经 `gzip.Writer` 流式写入 `<工作目录>/archive/traffic-collection/<表名>_<yyyymm>.csv.gz.tmp`。
|
||||
3. **校验**:重新读取压缩包统计行数,与 `count(*)` 比对;计算 SHA-256,写入 `.manifest.json`。校验通过后,把 `.tmp` 改名为正式文件名。
|
||||
4. **删除**:执行 `DROP TABLE`。
|
||||
|
||||
**失败处理**
|
||||
|
||||
- 任一步失败,保留已摘出的表,记录 Error 日志。
|
||||
- 下一轮维护时,先处理"已摘出但未删除"的表,按命名前缀加 `pg_inherits` 中不存在来识别。
|
||||
|
||||
**本地清理**
|
||||
|
||||
- 本地压缩包按月份只保留最近 N 个(默认 3),超出的删除。
|
||||
|
||||
**备选**:先 DROP 再由外部工具导出。
|
||||
- 可能丢数据,因此不采用。
|
||||
|
||||
**备选**:导出为 JSONL。
|
||||
- 用 `COPY CSV` 可以在数据库端直接流式输出,效率更高;原始数据作为 JSON 文本保存在 CSV 列中,可读性足够,因此选 CSV。
|
||||
|
||||
### D9 记录查询接口共用一套查询骨架
|
||||
|
||||
每类记录在 `internal/store/postgres` 中提供 `List(ctx, Filter)`。Filter 统一包含:
|
||||
|
||||
- `CardIDs`:由资产标识解析而来;
|
||||
- `From` / `To`;
|
||||
- 分页参数;
|
||||
- 各表自身的条件。
|
||||
|
||||
资产标识统一通过现有的资产解析服务转换为卡 ID 或设备 ID;解析不到时直接返回空页。原有的单卡接口复用同一个 `List`。
|
||||
|
||||
**索引**(迁移中创建;分区表在父表上建,自动作用到各分区):
|
||||
|
||||
| 表 | 索引 |
|
||||
|---|---|
|
||||
| 采集成功表 | `(observed_at DESC)`、`(carrier_id, observed_at DESC)`、`(decision, observed_at DESC)`,保留现有 `(iot_card_id, observed_at DESC)` |
|
||||
| 采集失败表 | `(iot_card_id, observed_at DESC)`、`(observed_at DESC)`、`(carrier_id, observed_at DESC)`、`(error_category, observed_at DESC)`、`(observation_id)` |
|
||||
| 停复机记录 | 现有三个索引重建为倒序;新增 `(created_at DESC)`、`(scene, created_at DESC)`、`(command_id)` |
|
||||
| 卡日流量 | 新增 `(date, usage_mb DESC)` |
|
||||
| 套餐日流量详单 | 新增 `(package_usage_id, date DESC)`、`(date)` |
|
||||
| 冲正流水 | 新增 `(iot_card_id, created_at DESC)`、`(created_at DESC)` |
|
||||
| 退款后处理尝试 | `(refund_id, created_at DESC)`、`(created_at DESC)`、`(result, created_at DESC)`、`(error_code, created_at DESC)` |
|
||||
|
||||
**总数统计**:使用同一组条件执行 `count(*)`。时间范围上限 31 天,配合分区裁剪,成本可控。
|
||||
|
||||
### D10 测试环境部署配置
|
||||
|
||||
`docker-compose.prod.yml` 的调整:
|
||||
|
||||
| 服务 | 调整 |
|
||||
|---|---|
|
||||
| api | 挂载改为 `./api/logs:/app/logs` |
|
||||
| worker | 挂载改为 `./worker/logs:/app/logs`,新增 `./worker/archive:/app/archive` |
|
||||
| 两者 | 增加 `JUNHONG_LOGGING_APP_LOG_MAX_BACKUPS=40`、`JUNHONG_LOGGING_APP_LOG_MAX_AGE=14` |
|
||||
|
||||
`.gitea/workflows/deploy.yaml` 改为创建 `api/logs`、`worker/logs`、`worker/archive` 三个目录。
|
||||
|
||||
## Risks / Trade-offs
|
||||
|
||||
| 风险 | 应对 |
|
||||
|---|---|
|
||||
| 后台页面依赖机器动作的审计(例如卡时间线里显示流量同步),改造后数据消失 | 实施时间线合并任务前,先与前端或运营确认;卡和设备时间线合并停复机记录 |
|
||||
| `AppendAndGet` 在策略为不写时返回 nil 事件,调用方可能解引用空指针 | 逐一检查 11 处调用;这些调用都属于人工或资金动作,策略为 `always`;补充返回 nil 时的单元测试 |
|
||||
| 采集原始数据使成功表变大 | 在线只保留 2 个月;部署后按实际行大小复核;必要时可把原始数据列的保留期单独缩短 |
|
||||
| 归档导出期间 IO 升高 | 每月只执行一次,流式读写;放在 leader 的维护循环中串行执行,一次只处理一个分区 |
|
||||
| 归档文件在同步前被本地清理删除 | 本地保留 3 个月,远大于同步周期;清理只删比保留个数更早的文件 |
|
||||
| 分表上线前,成功表中混有历史失败数据 | 成功表接口按 `result='ok'` 过滤;历史数据随分区过期 |
|
||||
| 退款后处理退避期间,Outbox 处理器返回成功导致事件不再重投 | 由补偿扫描按 `next_at` 拉起,并在补偿扫描中覆盖"待重试"状态 |
|
||||
| 本 Change 修改的 `card-traffic-metering` 和 `gateway-call-observability` 来自未归档的 Change | 本 Change 必须在 `add-card-traffic-queue-and-observed-deduction`、`add-carrier-gateway-rate-limit` 之后归档 |
|
||||
|
||||
## Migration Plan
|
||||
|
||||
只部署测试环境。推送 `Iteration/8-11` 后由 `.gitea/workflows/deploy.yaml` 自动部署,分三次提交:
|
||||
|
||||
1. 审计落库策略、Writer 分流、对接日志去掉补写审计、覆盖扫描、运行时覆盖配置。
|
||||
2. 退款后处理尝试记录与重试上限、停复机跳过记录。
|
||||
3. 采集分表与原始数据、到期归档、记录查询接口与索引、时间线合并、测试环境部署配置。
|
||||
|
||||
每次部署后在测试环境观察 24 小时,确认审计日增、应用日志和各业务记录写入正常。数据库迁移只新增表、列和索引,不改已有数据。
|
||||
|
||||
## Open Questions
|
||||
|
||||
- 后台是否有页面依赖机器动作的审计?实施时间线合并任务前与前端或运营确认。
|
||||
81
openspec/changes/refactor-audit-log-persistence/proposal.md
Normal file
81
openspec/changes/refactor-audit-log-persistence/proposal.md
Normal file
@@ -0,0 +1,81 @@
|
||||
## Why
|
||||
|
||||
统一审计表被当作过程日志使用。系统任务每次同步、扣减、对接调用开始、失败重试,都会写一条审计事件,外加 2–3 条资源快照。生产近 14 天日均写入约 122 万条,其中 99% 以上由机器发起;测试环境日均约 12.6 万条。两张审计表合计 21 个索引,写入成本高,是 11-03 生产数据库 IO 打满、轮询停摆的主要原因之一。
|
||||
|
||||
同时,从审计中移出的过程数据必须仍能查到。现有采集记录、停复机记录等业务表大多只能按单卡查询,筛选条件少。采集失败的具体原因被丢弃,也没有保存运营商原始响应,出了问题无据可查。
|
||||
|
||||
方案依据:`docs/2026-11-05-审计日志写入改造技术方案.md`(方案 B)。本 Change 基于 `Iteration/8-11` 实施,只部署到测试环境。
|
||||
|
||||
## What Changes
|
||||
|
||||
**审计写入**
|
||||
|
||||
- 统一审计注册表为每个动作声明落库策略:总是写、只写成功、只写结论、只写应用日志、不写。
|
||||
- 未声明策略的系统任务、定时任务动作,默认只写应用日志。
|
||||
- 后台账号、个人客户、OpenAPI、外部系统发起的动作始终写审计。
|
||||
- 系统任务的过程类动作不再写审计表,改写到所在服务的应用日志(`logs/app.log`,带 `logger=audit` 标记,不含变更前后快照)。涉及的动作包括:流量同步、网络状态同步、流量扣减、停机原因更新、通知投递、优先轮询、流量重置明细等。
|
||||
- 自动重试类动作只写最终结论:最终成功,或重试用尽转人工。每一次尝试都记录在业务尝试记录表中。
|
||||
- **BREAKING** 对接日志不再补写 `integration.attempt_started` 审计事件。没有上游审计事件的对接日志,其 `audit_event_id` 允许为空。
|
||||
- 提供按动作的运行时覆盖配置,不发版即可临时调整某个动作的落库策略。
|
||||
|
||||
**退款资产后处理**
|
||||
|
||||
- 新增退款后处理尝试记录,每次尝试一条,只增不改。
|
||||
- 自动重试最多 10 次,指数退避,间隔最长 1 小时。超过上限后标记"需人工处理",不再自动重投。
|
||||
|
||||
**停复机记录**
|
||||
|
||||
- 停复机指令在调用上游之前就结束时(已是目标状态、条件不再满足、不允许停复机),写一条结果为"跳过"的停复机记录,并写明原因。
|
||||
|
||||
**运营商流量采集记录**
|
||||
|
||||
- **BREAKING** 成功与失败分表:现有采集记录表只写成功读数,新增采集失败表。
|
||||
- 失败记录带错误类别、上游返回码与信息、耗时。
|
||||
- 失败记录不再判断卡有没有套餐,所有轮询到的卡都写。
|
||||
- 成功表和失败表都保存网关原始请求(加密前明文业务参数)与原始响应。
|
||||
- **BREAKING** 在线保留由 90 天改为 2 个月(当月加最近 2 个整月)。到期的月分区按"摘出分区、导出 `csv.gz`、校验、删除"四步处理,压缩包和说明文件留在 worker 运行目录,由同步服务器拉走。
|
||||
|
||||
**记录查询接口**
|
||||
|
||||
- 每类独立记录各提供一个列表接口:传卡标识查单个,不传查全部。
|
||||
- 都支持分页和条件筛选,时间范围必填。所需索引随筛选条件新建或重建。
|
||||
- 覆盖的记录:采集成功、采集失败、停复机、套餐日流量详单、卡日流量、流量冲正、退款后处理尝试。退款列表增加后处理状态筛选。
|
||||
|
||||
**测试环境部署配置**
|
||||
|
||||
- api、worker 各自挂载自己的 `logs/` 目录,不再共用同一个 `app.log`。
|
||||
- 每个服务保留 40 个压缩备份。
|
||||
- worker 挂载 `archive/` 目录。
|
||||
|
||||
## Capabilities
|
||||
|
||||
### New Capabilities
|
||||
|
||||
- `operational-record-query`:独立业务记录的统一分页条件查询。覆盖采集成功与失败、停复机、套餐日流量、卡日流量、流量冲正、退款后处理尝试;可查单个,也可查全部。
|
||||
- `refund-asset-post-processing`:退款资产后处理的逐次尝试记录、重试上限、退避与转人工。
|
||||
|
||||
### Modified Capabilities
|
||||
|
||||
- `operations-audit`:审计落库策略分级;系统过程动作改写应用日志;结论型写入;对接日志允许无审计关联;轮询审计边界调整。
|
||||
- `card-traffic-metering`:采集记录成功与失败分表、保存原始请求与响应、失败记录不再判断有无套餐、在线保留 2 个月并到期归档为压缩包。该 capability 来自尚未归档的 `add-card-traffic-queue-and-observed-deduction`,本 Change 须在其归档后再归档。
|
||||
- `gateway-call-observability`:停复机指令在调用前结束时写"跳过"记录。该 capability 来自尚未归档的 `add-carrier-gateway-rate-limit`,本 Change 须在其归档后再归档。
|
||||
|
||||
## Impact
|
||||
|
||||
- **审计**:`internal/infrastructure/audit`(注册表、Writer)、`pkg/constants/audit.go`、`internal/governance/auditcoverage`、`internal/query/audit`(时间线合并停复机记录)。
|
||||
- **对接日志**:`internal/infrastructure/integrationlog/repository.go`。
|
||||
- **退款**:`internal/service/refund`、`internal/infrastructure/commissiondelivery`(补偿扫描),新增尝试记录表。
|
||||
- **停复机**:停复机指令执行路径中写入跳过记录。
|
||||
- **采集**:
|
||||
- 原始数据收集器:`internal/gateway/client.go`
|
||||
- 采集写入:`internal/task/polling_carddata_handler.go`、`internal/application/cardobservation`
|
||||
- 存储与分区维护:`internal/store/postgres/card_traffic_collection_store.go`、`cmd/worker/main.go`
|
||||
- **接口**:`internal/routes`、`internal/handler/admin`、`internal/model/dto`、`internal/store/postgres`,并同步生成接口文档。
|
||||
- **数据库迁移**:
|
||||
- 新增采集失败表、退款后处理尝试表;
|
||||
- 采集成功表增加 `carrier_id` 与原始数据列;
|
||||
- 退款单增加后处理状态字段;
|
||||
- 各记录表新建或重建索引。
|
||||
- **配置与部署**:`pkg/config`(`audit.persist_overrides`)、`docker-compose.prod.yml`、`.gitea/workflows/deploy.yaml`。
|
||||
- **与已有 Change 的关系**:`add-audit-integration-write-switches` 的全局写入开关保持不变。全局开关用于应急全停;本 Change 是常态下按动作分级,两者互不替代。
|
||||
- **不涉及**:生产环境的代码、配置与服务;已有审计数据的删除或迁移。
|
||||
@@ -0,0 +1,95 @@
|
||||
## MODIFIED Requirements
|
||||
|
||||
### Requirement: 采集记录
|
||||
|
||||
系统 SHALL 把每次流量采集按结果分别写入两张表。
|
||||
|
||||
**成功表 `tb_card_traffic_collection`**
|
||||
|
||||
- 只写读数结果为有效的采集。
|
||||
- 记录内容:采集时间、采集 ID、来源、场景、运营商账户、运营商原始读数、入队后的队列、写入判断(已写入、建起点、队列未满、读数下降、乱序、激活前结算用已有快照)。
|
||||
- 写入成功时,还记录换算后的本月累计、是否触发防旧值、本次增量与流量事件 ID。
|
||||
- 只为采集时刻有套餐的卡写成功记录;没有套餐的卡 MUST NOT 写成功记录。
|
||||
|
||||
**失败表 `tb_card_traffic_collection_failure`**
|
||||
|
||||
- 写读数结果为失败、限流、响应无效、无读数、读数无效的采集。
|
||||
- 记录内容:采集时间、采集 ID、来源、场景、运营商账户、ICCID、失败类型、错误类别、上游返回码、上游返回信息(截断)、耗时。
|
||||
- 失败记录 MUST 为所有被采集的卡写入,不判断卡是否有套餐。
|
||||
|
||||
**原始数据**
|
||||
|
||||
- 两张表都 SHALL 保存本次网关调用的原始数据:加密前的明文业务请求、HTTP 状态码、网关响应的 `code`、`msg`、`trace_id`、原始响应体,以及网关客户端内部重试次数。
|
||||
- 原始响应能解析为 JSON 时按 JSON 保存,否则按文本保存。超过 4 KB 时截断,并标记已截断。
|
||||
- 网关客户端内部发生过重试时,保存最后一次的请求与响应。
|
||||
- 系统 MUST NOT 保存加密后的报文、签名、应用标识、密钥或请求头。
|
||||
|
||||
**通用规则**
|
||||
|
||||
- 采集 ID 等于对接日志的 `integration_id`,MUST 随流量事件与扣费审计传递,使扣费可追溯到对应采集记录。
|
||||
- 采集记录写入失败 MUST 只记录告警,不影响扣减与卡状态。
|
||||
|
||||
#### Scenario: 被队列拦下的采集
|
||||
- **WHEN** 有套餐的卡本次读数使队列出现下降
|
||||
- **THEN** 系统在成功表写一条采集记录,判断为读数下降,记录队列内容与原始响应,不写快照
|
||||
|
||||
#### Scenario: 从扣费追到采集
|
||||
- **WHEN** 维护者从一笔扣费排查
|
||||
- **THEN** 可经流量事件的采集 ID 找到成功表中对应的原始读数、队列与原始响应
|
||||
|
||||
#### Scenario: 没套餐的卡采集成功
|
||||
- **WHEN** 没套餐的卡被采集且读数有效
|
||||
- **THEN** 系统不写成功记录
|
||||
|
||||
#### Scenario: 没套餐的卡采集失败
|
||||
- **WHEN** 没套餐的卡查流量超时
|
||||
- **THEN** 系统在失败表写一条记录,失败类型为失败,错误类别为超时,HTTP 状态码为空
|
||||
|
||||
#### Scenario: 上游返回业务错误
|
||||
- **WHEN** 网关返回非成功的业务码
|
||||
- **THEN** 失败表记录上游返回码、返回信息、`trace_id` 与原始响应体
|
||||
|
||||
#### Scenario: 原始响应过长
|
||||
- **WHEN** 网关原始响应体超过 4 KB
|
||||
- **THEN** 系统保存前 4 KB,并标记已截断
|
||||
|
||||
#### Scenario: 不保存敏感报文
|
||||
- **WHEN** 任一采集记录写入原始数据
|
||||
- **THEN** 记录中不含加密报文、签名、应用标识与密钥
|
||||
|
||||
## ADDED Requirements
|
||||
|
||||
### Requirement: 采集记录在线保留与到期归档
|
||||
|
||||
采集成功表与失败表 SHALL 按采集时间按月分区,在线保留当月加最近 2 个整月。月分区的月末早于"当前时间减 2 个月"时即为到期。
|
||||
|
||||
到期分区 SHALL 由 leader worker 的分区维护任务按以下顺序处理:
|
||||
|
||||
1. 把分区从父表摘出,使其不再参与在线查询;
|
||||
2. 按采集时间与 ID 排序,流式导出为带表头的 CSV,并以 gzip 压缩,写入 worker 运行目录下的 `archive/traffic-collection/`;
|
||||
3. 重新读取压缩包统计行数,与分区行数比对;同时生成同名说明文件,记录表名、月份、行数、最早与最晚采集时间、文件大小、SHA-256 与导出时间;
|
||||
4. 行数一致后删除已摘出的分区。
|
||||
|
||||
约束:
|
||||
|
||||
- 压缩包与说明文件 MUST 先以临时文件名写入,校验通过后再改为正式文件名。
|
||||
- 导出或校验失败时,系统 MUST 保留已摘出的分区,记录错误日志,并在下一轮维护时重试;MUST NOT 在没有完成校验的压缩包时删除分区。
|
||||
- 本地 SHALL 只保留最近 3 个月的压缩包,个数可配置。
|
||||
- 维护任务 SHALL 继续预建未来月份的分区。
|
||||
|
||||
#### Scenario: 到期分区归档
|
||||
- **GIVEN** 当前为 2027-01-01,成功表存在 2026-10 分区
|
||||
- **WHEN** 分区维护任务运行
|
||||
- **THEN** 系统生成 `tb_card_traffic_collection_202610.csv.gz` 与说明文件,行数与分区一致后删除该分区
|
||||
|
||||
#### Scenario: 导出失败
|
||||
- **WHEN** 归档过程中写压缩包失败
|
||||
- **THEN** 已摘出的分区保留,不生成正式文件名的压缩包,下一轮维护重试
|
||||
|
||||
#### Scenario: 在线查询不包含已归档月份
|
||||
- **WHEN** 查询的时间范围落在已归档月份
|
||||
- **THEN** 接口返回空结果,不报错
|
||||
|
||||
#### Scenario: 本地压缩包超过保留个数
|
||||
- **WHEN** 本地已有压缩包的月份数超过配置的保留个数
|
||||
- **THEN** 系统删除最早月份的压缩包与说明文件
|
||||
@@ -0,0 +1,20 @@
|
||||
## ADDED Requirements
|
||||
|
||||
### Requirement: 停复机指令调用前结束的跳过记录
|
||||
|
||||
停复机指令在调用上游之前结束时,系统 SHALL 写一条结果为跳过(`skipped`)的停复机记录,并在原因中写明结束原因。适用情况:
|
||||
|
||||
- 执行前判断卡已是目标状态;
|
||||
- 执行前判断停复机条件不再满足;
|
||||
- 卡不允许停复机(如关闭轮询、规则不允许)。
|
||||
|
||||
被本系统限流推迟、稍后会再次执行的指令 MUST NOT 写跳过记录;限流推迟仍按现有分钟计数观测。跳过记录 MUST NOT 写统一审计。
|
||||
|
||||
#### Scenario: 执行前已是目标状态
|
||||
- **GIVEN** 一条停机指令待执行
|
||||
- **WHEN** 执行前判断卡已经是停机状态并结束指令
|
||||
- **THEN** 系统写一条结果为跳过、原因为已是目标状态的停复机记录,不调用上游
|
||||
|
||||
#### Scenario: 限流推迟
|
||||
- **WHEN** 停机指令因所在限流组本秒额度用尽被推迟
|
||||
- **THEN** 系统不写停复机记录
|
||||
@@ -0,0 +1,116 @@
|
||||
## ADDED Requirements
|
||||
|
||||
### Requirement: 记录查询公共约定
|
||||
|
||||
系统 SHALL 为每类独立业务记录提供一个后台列表接口,接口遵循以下约定。
|
||||
|
||||
**查询范围**
|
||||
|
||||
- 资产标识选填。传入时只查该卡或该资产,支持虚拟号、ICCID、MSISDN,设备支持 IMEI、SN;不传时查全部。
|
||||
- 资产标识解析不到资产时,接口 MUST 返回空页,不报错。
|
||||
- 原有的单卡、单套餐接口 SHALL 保留,与列表接口使用同一查询逻辑。
|
||||
|
||||
**时间范围**
|
||||
|
||||
- 开始时间与结束时间 MUST 必填,跨度不超过 31 天,按上海时间解析;结束时间不含。
|
||||
- 单卡、单套餐接口未传时间时,默认取最近 31 天。
|
||||
- 查全部时除时间范围外,MUST NOT 要求其他筛选条件。
|
||||
|
||||
**分页与排序**
|
||||
|
||||
- 页码默认 1;每页条数默认 20,最大 100。采集成功记录沿用现有上限 200。
|
||||
- 返回总数、页码、每页条数与当前页记录。
|
||||
- 按记录时间倒序排列,时间相同时按 ID 倒序。
|
||||
|
||||
**其他**
|
||||
|
||||
- 系统 SHALL 为本能力列出的筛选条件建立所需索引,使带时间范围的查询不扫描时间范围外的分区。
|
||||
- 接口 MUST 需要平台账号登录。
|
||||
- 接口 MUST 在路由声明中登记输入与输出结构,并同步生成接口文档。
|
||||
|
||||
#### Scenario: 查单卡
|
||||
- **WHEN** 请求带 ICCID 和 7 天时间范围
|
||||
- **THEN** 系统只返回该卡在范围内的记录,并分页、按时间倒序返回
|
||||
|
||||
#### Scenario: 查全部
|
||||
- **WHEN** 请求只带 7 天时间范围
|
||||
- **THEN** 系统返回范围内全部记录的第一页与总数
|
||||
|
||||
#### Scenario: 缺少时间范围
|
||||
- **WHEN** 列表接口请求未带开始或结束时间
|
||||
- **THEN** 系统拒绝请求并提示时间范围必填
|
||||
|
||||
#### Scenario: 时间跨度超限
|
||||
- **WHEN** 请求的时间跨度超过 31 天
|
||||
- **THEN** 系统拒绝请求并提示跨度上限
|
||||
|
||||
#### Scenario: 标识找不到资产
|
||||
- **WHEN** 请求的资产标识不存在
|
||||
- **THEN** 系统返回总数为 0 的空页
|
||||
|
||||
### Requirement: 流量采集记录查询
|
||||
|
||||
系统 SHALL 提供两个采集记录列表接口:
|
||||
|
||||
- 采集成功记录:`GET /traffic-collections`。筛选条件:资产标识、运营商账户、写入判断、来源、场景、是否产生增量、是否触发防旧值。
|
||||
- 采集失败记录:`GET /traffic-collection-failures`。筛选条件:资产标识、运营商账户、失败类型、错误类别、上游返回码、来源、场景。
|
||||
|
||||
两者都 MUST 返回原始请求与原始响应字段。单卡采集记录接口 SHALL 支持"包含失败"参数:默认只返回成功记录;开启时合并成功与失败记录,并按采集时间倒序返回。
|
||||
|
||||
#### Scenario: 按运营商账户查失败
|
||||
- **WHEN** 查询某运营商账户近 1 天、错误类别为超时的采集失败记录
|
||||
- **THEN** 系统返回该账户下所有卡的超时失败记录,含上游返回与原始数据
|
||||
|
||||
#### Scenario: 查产生增量的采集
|
||||
- **WHEN** 查询近 1 天产生增量的采集成功记录
|
||||
- **THEN** 系统只返回增量大于 0 的记录
|
||||
|
||||
#### Scenario: 单卡合并查看
|
||||
- **WHEN** 单卡采集记录接口开启"包含失败"
|
||||
- **THEN** 成功与失败记录合并后按采集时间倒序分页返回,每条标明是成功还是失败
|
||||
|
||||
### Requirement: 停复机记录查询
|
||||
|
||||
停复机记录列表 SHALL 支持以下筛选条件:资产标识(不再只支持 ICCID)、运营商账户、动作、来源、场景、原因、结果(含跳过)、错误类别、指令号。单卡停复机历史 SHALL 支持动作、结果、来源筛选。
|
||||
|
||||
#### Scenario: 按场景与原因查
|
||||
- **WHEN** 查询近 7 天场景为退款、结果为跳过的停复机记录
|
||||
- **THEN** 系统返回匹配记录,并带跳过原因
|
||||
|
||||
#### Scenario: 用虚拟号查
|
||||
- **WHEN** 用卡的虚拟号查询停复机记录
|
||||
- **THEN** 系统返回该卡的停复机记录
|
||||
|
||||
### Requirement: 流量与冲正记录查询
|
||||
|
||||
系统 SHALL 提供以下列表接口:
|
||||
|
||||
| 记录 | 筛选条件 |
|
||||
|---|---|
|
||||
| 套餐日流量详单 | 资产标识、套餐使用记录、套餐、运营商账户、日期范围 |
|
||||
| 卡日流量 | 资产标识、运营商账户、日期范围、用量下限 |
|
||||
| 套餐流量冲正流水 | 资产标识、套餐使用记录、操作人、调整方向(补扣或退回)、时间范围 |
|
||||
|
||||
现有套餐日流量详单接口 SHALL 增加分页。现有单套餐冲正流水接口 SHALL 增加时间范围筛选。
|
||||
|
||||
#### Scenario: 查异常大流量
|
||||
- **WHEN** 查询近 3 天、单日用量不低于 10240 MB 的卡日流量
|
||||
- **THEN** 系统返回所有满足条件的卡与日期,按日期倒序分页
|
||||
|
||||
#### Scenario: 查全部冲正
|
||||
- **WHEN** 查询近 30 天的全部退回类冲正流水
|
||||
- **THEN** 系统返回所有卡的退回类冲正记录,含操作人与调整前后用量
|
||||
|
||||
### Requirement: 退款后处理尝试查询
|
||||
|
||||
系统 SHALL 提供退款后处理尝试记录列表接口。筛选条件:退款单 ID、退款单号、资产标识、处理步骤、结果、错误码、触发方式、时间范围。
|
||||
|
||||
退款列表 SHALL 增加后处理状态筛选,可筛出"需人工处理"。退款列表与详情 SHALL 返回后处理次数、最后错误与下次重试时间。
|
||||
|
||||
#### Scenario: 查需人工处理的退款
|
||||
- **WHEN** 退款列表按后处理状态"需人工处理"筛选
|
||||
- **THEN** 系统返回所有已转人工的退款单,并带最后错误
|
||||
|
||||
#### Scenario: 查一张退款单的全部尝试
|
||||
- **WHEN** 按退款单号查询尝试记录
|
||||
- **THEN** 系统按时间倒序返回该单每一次尝试的步骤、结果、错误码与触发方式
|
||||
@@ -0,0 +1,160 @@
|
||||
## MODIFIED Requirements
|
||||
|
||||
### Requirement: 审计时间线
|
||||
|
||||
系统 SHALL 支持按事件、操作者、资源、请求、关联标识和资金维度查询已记录的审计事实。新建的 `tb_integration_log` 在已取得稳定审计事件时 SHALL 写入其内部 ID 作为 `audit_event_id`;没有上游审计事件时 `audit_event_id` SHALL 为空,系统 MUST NOT 为补齐关联而另写审计事件,并且该日志的非空 `integration_id` SHALL 出现在对应事件列表和事件详情的 `investigation_refs.integration_refs` 中;关联生成失败时系统 MUST 保留 Integration Log 并记录可排查告警,且 MUST NOT 按名称、时间或摘要推断关联。历史 Integration Log 不在本要求的回填范围内。审计事件的构造、校验或持久化失败 MUST 记录可关联的结构化错误日志和二次失败记录,且 MUST NOT 改变已通过业务校验的业务操作结果或接口响应。尚无完成物理清理记录时,审计调查接口 MUST 继续查询在线审计事实,且返回的留存边界不得将数据标记为已归档。
|
||||
|
||||
#### Scenario: 审计时间线
|
||||
|
||||
- **GIVEN** 审计事实已存在
|
||||
- **WHEN** 使用对应维度查询
|
||||
- **THEN** 返回匹配的事实与稳定动作编码,不用访问日志替代
|
||||
|
||||
#### Scenario: 事件返回已关联外部交互引用
|
||||
|
||||
- **GIVEN** 一个在线审计事件有 `tb_integration_log.audit_event_id` 指向其内部 ID 的外部交互记录
|
||||
- **WHEN** 查询全局审计事件列表或该事件详情
|
||||
- **THEN** 该事件的 `investigation_refs.integration_refs` 返回该记录的 `integration_id`
|
||||
|
||||
#### Scenario: 事件没有关联外部交互引用
|
||||
|
||||
- **GIVEN** 一个在线审计事件没有稳定关联的外部交互记录
|
||||
- **WHEN** 查询全局审计事件列表或该事件详情
|
||||
- **THEN** `investigation_refs.integration_refs` 返回空数组
|
||||
|
||||
#### Scenario: 新外部交互日志具有稳定审计关联
|
||||
|
||||
- **GIVEN** 系统即将记录一次新的外部调用、入站回调或未发送裁决,且该调用由一个已写入的审计事件发起
|
||||
- **WHEN** 写入对应 Integration Log
|
||||
- **THEN** 该记录保存非空 `audit_event_id`,且其目标 Audit Event 已存在
|
||||
|
||||
#### Scenario: 没有上游审计事件的外部交互日志
|
||||
|
||||
- **GIVEN** 一次系统任务发起的外部调用没有上游审计事件
|
||||
- **WHEN** 写入对应 Integration Log
|
||||
- **THEN** 该记录的 `audit_event_id` 为空
|
||||
- **AND** 审计事件表不新增 `integration.attempt_started` 事件
|
||||
|
||||
#### Scenario: 缺少审计关联时保留外部交互日志
|
||||
|
||||
- **GIVEN** 一次新的外部交互日志没有稳定审计事件关联
|
||||
- **WHEN** 系统尝试写入该 Integration Log
|
||||
- **THEN** 系统持久化该 Integration Log、记录可排查告警,并返回空 `integration_refs`
|
||||
|
||||
#### Scenario: 审计写入失败不阻断业务
|
||||
|
||||
- **GIVEN** 一个业务操作已通过自身输入、权限和状态校验
|
||||
- **WHEN** 该操作的审计事件构造、校验或持久化失败
|
||||
- **THEN** 系统提交或返回该业务操作原本的结果,并以请求关联标识、动作编码和资源标识记录审计失败
|
||||
|
||||
#### Scenario: 尚无物理清理记录时查询在线审计
|
||||
|
||||
- **GIVEN** 审计或外部交互日志尚无完成物理清理的归档运行记录
|
||||
- **WHEN** 调用任一依赖留存边界的审计调查接口
|
||||
- **THEN** 系统返回当前在线数据或空列表
|
||||
- **AND** 响应中的 `archived_before` 为空
|
||||
- **AND** 系统不得因空留存边界返回服务端错误
|
||||
|
||||
### Requirement: 轮询审计事实保留边界
|
||||
|
||||
系统 SHALL 仅为人工触发的轮询操作,以及由轮询结果导致的资金或套餐生命周期变化(如套餐到期、激活、退款失效)持久化统一审计事件及资源快照。系统任务发起的轮询同步本身(流量同步、网络状态同步、流量扣减、停复机原因更新)MUST NOT 创建 `tb_audit_event` 或 `tb_audit_event_resource`,无论成功、失败或结果未知;其可查证据由采集记录、采集失败记录、停复机记录、日流量记录与应用日志提供。审计调查接口仅返回实际已保留的历史事实。
|
||||
|
||||
#### Scenario: 自动轮询成功但无业务变化
|
||||
- **GIVEN** 一次自动轮询成功且没有改变任何业务状态、有效流量或风险结论
|
||||
- **WHEN** 系统结束该轮询处理
|
||||
- **THEN** 审计事件和审计资源表不新增该次轮询记录
|
||||
|
||||
#### Scenario: 自动轮询同步流量并扣减
|
||||
- **GIVEN** 一次自动轮询读到有效增量并完成扣减
|
||||
- **WHEN** 系统提交该变化
|
||||
- **THEN** 审计事件表不新增流量同步或扣减事件
|
||||
- **AND** 采集记录保存该次读数、扣减判定、增量与原始响应
|
||||
|
||||
#### Scenario: 自动轮询调用失败
|
||||
- **WHEN** 一次自动轮询查流量超时或返回无效响应
|
||||
- **THEN** 审计事件表不新增记录
|
||||
- **AND** 采集失败记录保存该次失败的错误类别、上游返回与原始数据
|
||||
|
||||
#### Scenario: 轮询导致套餐到期
|
||||
- **GIVEN** 轮询或定时任务判定一个套餐已到期
|
||||
- **WHEN** 系统把套餐置为已过期
|
||||
- **THEN** 系统保留套餐到期的审计事实与资源快照
|
||||
|
||||
#### Scenario: 人工触发刷新
|
||||
- **WHEN** 后台账号手动刷新一张卡
|
||||
- **THEN** 系统为该操作写审计事件
|
||||
|
||||
## ADDED Requirements
|
||||
|
||||
### Requirement: 审计动作落库策略
|
||||
|
||||
系统 SHALL 在统一审计注册表中为每个动作声明落库策略,取值为:总是写(`always`)、只写成功(`on_success`)、只写结论(`outcome`)、只写应用日志(`log_only`)、不写(`none`)。
|
||||
|
||||
- 操作者为后台账号、个人客户、OpenAPI 或外部系统时,系统 MUST 按"总是写"处理,不受动作策略限制。
|
||||
- 操作者为系统任务或定时任务、且动作未声明策略时,系统 MUST 按"只写应用日志"处理。
|
||||
- 审计覆盖扫描 MUST 在存在可由系统任务或定时任务发起、但未声明落库策略的动作时失败。
|
||||
- 系统 SHALL 提供按动作编码覆盖落库策略的运行时配置;覆盖只作用于系统任务与定时任务操作者。
|
||||
- 是否写入审计 MUST NOT 改变业务操作的结果或接口响应。
|
||||
|
||||
#### Scenario: 人工操作始终写审计
|
||||
- **GIVEN** 动作 `order.create` 对系统任务声明为只写应用日志
|
||||
- **WHEN** 后台账号创建订单
|
||||
- **THEN** 系统写入审计事件与资源快照
|
||||
|
||||
#### Scenario: 新增机器动作未声明策略
|
||||
- **WHEN** 系统任务写一个未声明落库策略的动作
|
||||
- **THEN** 审计事件表不新增记录,应用日志新增一行该动作的结构化日志
|
||||
|
||||
#### Scenario: 覆盖扫描发现未声明策略
|
||||
- **GIVEN** 一个可由系统任务发起的动作没有声明落库策略
|
||||
- **WHEN** 运行审计覆盖扫描
|
||||
- **THEN** 扫描失败并列出该动作编码
|
||||
|
||||
#### Scenario: 运行时覆盖
|
||||
- **GIVEN** 运行时配置把 `package_usage.expire` 覆盖为只写应用日志
|
||||
- **WHEN** 定时任务使一个套餐到期
|
||||
- **THEN** 审计事件表不新增记录,应用日志新增一行
|
||||
- **AND** 删除该覆盖后,下一次到期重新写审计事件
|
||||
|
||||
### Requirement: 系统过程动作写应用日志
|
||||
|
||||
落库策略为"只写应用日志",或"只写成功"时的失败结果,系统 SHALL 在所在服务的应用日志中写一行结构化日志,并 MUST NOT 写入审计事件表与资源表。
|
||||
|
||||
- 日志行 MUST 带固定 logger 名 `audit`。
|
||||
- 日志行 MUST 包含:动作编码、操作者类型与标识、来源、结果、资源类型与资源键、关联标识、请求标识、错误码、摘要。
|
||||
- 日志行 MUST NOT 包含变更前后快照、主体数据、密钥、令牌或加密报文。
|
||||
- 成功写 Info 级别,失败写 Warn 级别。
|
||||
- 应用日志写入失败 MUST NOT 重试,也 MUST NOT 影响业务结果。
|
||||
|
||||
#### Scenario: 网络状态同步
|
||||
- **WHEN** 系统任务同步一张卡的网络状态
|
||||
- **THEN** 应用日志新增一行 `logger=audit`、动作编码为 `iot_card.worker_network_sync` 的记录
|
||||
- **AND** 该行不含变更前后快照
|
||||
|
||||
#### Scenario: 不新增审计表数据
|
||||
- **WHEN** 系统任务写任一只写应用日志的动作
|
||||
- **THEN** `tb_audit_event` 与 `tb_audit_event_resource` 均无新增
|
||||
|
||||
### Requirement: 自动重试动作只写结论
|
||||
|
||||
落库策略为"只写结论"的动作,系统 SHALL 只在最终结论产生时写一条审计事件。最终结论指最终成功,或重试用尽转人工处理。
|
||||
|
||||
- 中间失败 MUST NOT 写审计事件,只写应用日志;日志行 MUST 带尝试序号与对应尝试记录标识。
|
||||
- 结论事件的元数据 MUST 包含总尝试次数、首次失败时间与最后错误码。
|
||||
- 同一业务对象的同一结论 MUST 只有一条审计事件。
|
||||
- 每一次尝试的细节 MUST 由对应业务尝试记录保存,并可分页查询。
|
||||
|
||||
#### Scenario: 重试后成功
|
||||
- **GIVEN** 一张退款单的资产后处理前两次失败、第三次成功
|
||||
- **WHEN** 第三次处理完成
|
||||
- **THEN** 审计事件表只有一条结论为成功的事件,元数据中总尝试次数为 3
|
||||
- **AND** 尝试记录中有三条记录
|
||||
|
||||
#### Scenario: 重试用尽转人工
|
||||
- **GIVEN** 一张退款单的资产后处理连续失败达到上限
|
||||
- **WHEN** 系统将其标记为需人工处理
|
||||
- **THEN** 审计事件表写入一条结论为失败的事件,包含最后错误码
|
||||
|
||||
#### Scenario: 结论重复写入
|
||||
- **WHEN** 同一退款单的成功结论被重复提交
|
||||
- **THEN** 审计事件表仍只有一条该结论事件
|
||||
@@ -0,0 +1,42 @@
|
||||
## ADDED Requirements
|
||||
|
||||
### Requirement: 退款资产后处理尝试记录
|
||||
|
||||
系统 SHALL 在每次执行退款资产后处理时写一条尝试记录,只新增,不修改、不删除。
|
||||
|
||||
- 记录内容:退款单、处理步骤(套餐失效、接续激活、停机评估或停机)、资产类型与资产 ID、ICCID、尝试序号、触发方式(Outbox 投递、补偿扫描、人工重试)、结果、错误码、错误信息(截断)、关联的停复机记录、耗时、记录时间。
|
||||
- 尝试记录 SHALL 按月分区。
|
||||
- 尝试记录写入失败 MUST 只记录告警,不影响后处理结果。
|
||||
|
||||
#### Scenario: 一次尝试失败
|
||||
- **WHEN** 退款资产后处理在套餐失效步骤失败
|
||||
- **THEN** 系统写一条尝试记录,步骤为套餐失效,结果为失败,并带错误码与错误信息
|
||||
|
||||
#### Scenario: 一次尝试成功
|
||||
- **WHEN** 退款资产后处理全部步骤成功
|
||||
- **THEN** 系统写一条结果为成功的尝试记录
|
||||
|
||||
### Requirement: 退款资产后处理重试上限
|
||||
|
||||
系统 SHALL 在退款单上维护资产后处理的当前状态、已尝试次数、最后错误与下次重试时间。
|
||||
|
||||
- 后处理失败时,系统 SHALL 按指数退避计算下次重试时间,间隔最长 1 小时。
|
||||
- 已尝试次数达到 10 次仍未成功时,系统 MUST 把状态置为"需人工处理",并停止自动重试。
|
||||
- Outbox 投递与补偿扫描 MUST 按上述状态与下次重试时间判断是否执行:未到下次重试时间的不执行,"需人工处理"的不执行。
|
||||
- 后台人工重试 SHALL 允许重新执行"需人工处理"的退款单,并记为人工重试触发;执行前已尝试次数不清零。
|
||||
- 重复执行 MUST 保持既有幂等:不重复失效套餐,不重复处理资产。
|
||||
|
||||
#### Scenario: 退避重试
|
||||
- **GIVEN** 一张退款单的资产后处理第 3 次失败
|
||||
- **WHEN** 补偿扫描在下次重试时间之前运行
|
||||
- **THEN** 系统不执行该退款单的后处理
|
||||
|
||||
#### Scenario: 达到上限转人工
|
||||
- **GIVEN** 一张退款单的资产后处理已失败 9 次
|
||||
- **WHEN** 第 10 次仍失败
|
||||
- **THEN** 退款单状态为需人工处理,Outbox 与补偿扫描不再执行它,审计写一条结论为失败的事件
|
||||
|
||||
#### Scenario: 人工重试成功
|
||||
- **GIVEN** 一张需人工处理的退款单
|
||||
- **WHEN** 运营在后台发起人工重试并成功
|
||||
- **THEN** 退款单状态为已完成,尝试记录新增一条触发方式为人工重试、结果为成功的记录
|
||||
71
openspec/changes/refactor-audit-log-persistence/tasks.md
Normal file
71
openspec/changes/refactor-audit-log-persistence/tasks.md
Normal file
@@ -0,0 +1,71 @@
|
||||
## 1. 审计落库策略(第 1 次提交)
|
||||
|
||||
- [x] 1.1 `internal/infrastructure/audit/registry.go`:`ActionDefinition` 新增 `Persist`,取值为 `always`、`on_success`、`outcome`、`log_only`、`none`;按技术方案 4.1、4.2 为全部可由系统任务或定时任务发起的动作声明策略,`integration.attempt_started` 设为 `none`。
|
||||
- [x] 1.2 `writer.go`:
|
||||
- 新增 `persistDecision`:人工操作者一律写表;系统任务和定时任务按 `Persist` 处理,未声明时按 `log_only`。
|
||||
- `AppendInput` 增加 `Final`;`outcome` 策略在 `Final=false` 时只写日志,`Final=true` 时按业务键 `EventID` 写表。
|
||||
- [x] 1.3 `log_only` 和 `on_success` 的失败结果写 `logger.GetAppLogger().Named("audit")`,`msg=audit_event`。字段按 spec 写入,不含快照;成功记 Info,失败记 Warn。
|
||||
- [x] 1.4 逐一检查 `AppendAndGet` 的 11 处调用,确保返回 nil 事件时不会出错。
|
||||
- [x] 1.5 `pkg/config` 新增 `audit.persist_overrides`(动作编码 → 策略),运行时生效,只作用于系统任务和定时任务操作者。
|
||||
- [x] 1.6 `internal/infrastructure/integrationlog/repository.go`:删除 `Start` 中补写 `integration.attempt_started` 的分支;确认 `auditEventIDForSeries` 的查询走索引。
|
||||
- [x] 1.7 `internal/governance/auditcoverage`:可由系统任务发起但未声明 `Persist` 的动作,扫描判为失败;更新已移出审计入口的 `AuditEvent` / `DomainLedger` 分类。
|
||||
- [x] 1.8 `gofmt`、`go build ./cmd/api ./cmd/worker`、`openspec validate refactor-audit-log-persistence`;提交并推送 8-11。
|
||||
|
||||
## 2. 退款后处理与停复机跳过记录(第 2 次提交)
|
||||
|
||||
- [ ] 2.1 迁移:
|
||||
- 新建 `tb_refund_post_process_attempt`,按月分区,并建立索引;
|
||||
- 退款单新增 `asset_post_status`、`asset_post_attempts`、`asset_post_last_error`、`asset_post_next_at`;
|
||||
- 补齐 down 脚本。
|
||||
- [ ] 2.2 `internal/service/refund`:
|
||||
- `processRefundAsset` 每次执行都写尝试记录(步骤、触发方式、结果、错误码与错误信息、耗时、关联停复机记录);
|
||||
- 失败时按指数退避(最长 1 小时)更新退款单;
|
||||
- 达到 10 次后置为需人工处理,并写一条 `outcome` 失败结论审计;
|
||||
- 成功时写 `outcome` 成功结论审计,`asset_reset` 同步置为完成。
|
||||
- [ ] 2.3 Outbox 处理器与 `RecoverRefundPostProcessing` 在执行前检查状态和 `next_at`:未到期、已完成、需人工处理的都跳过;补偿扫描负责拉起已到期的待重试单。
|
||||
- [ ] 2.4 新增 `POST /refunds/:id/asset-post-processing/retry` 人工重试接口,只允许在需人工处理状态下调用,记为人工重试;在路由 `RouteSpec` 中登记,并生成接口文档。
|
||||
- [ ] 2.5 停复机指令执行前的重新判断处,按三种结束原因写 `skipped` 停复机记录(已是目标状态、条件不再满足、不允许停复机);限流推迟不写。
|
||||
- [ ] 2.6 构建、校验、提交并推送 8-11。
|
||||
|
||||
## 3. 流量采集分表与原始数据(第 3 次提交)
|
||||
|
||||
- [ ] 3.1 `internal/gateway/client.go`:新增 `WithRawCapture`,`executeWithRetry` 按尝试写入最后一次的明文请求、HTTP 状态、响应体、`code`、`msg`、`trace_id` 和尝试次数;不记录加密报文、签名、`app_id`、密钥和请求头。
|
||||
- [ ] 3.2 迁移:
|
||||
- 新建 `tb_card_traffic_collection_failure`,按月分区,并建立索引;
|
||||
- 采集成功表增加 `carrier_id` 和原始数据列(`raw_request`、`http_status`、`gateway_code`、`gateway_msg`、`gateway_trace_id`、`raw_response`、`raw_response_text`、`raw_truncated`、`gateway_attempts`),新增 D9 中列出的索引。
|
||||
- [ ] 3.3 `polling_carddata_handler`:
|
||||
- 调用 `QueryFlow` 前挂上收集器;
|
||||
- 网关失败时填入错误类别、上游码与信息、耗时;
|
||||
- 成功和失败都把原始数据带进 `TrafficObservation`。
|
||||
- [ ] 3.4 `internal/application/cardobservation`:`writeCollectionRecord` 按结果分流。成功写成功表(保持"有套餐才写"),失败直接写失败表(不查套餐);原始响应超过 4 KB 时截断并标记。
|
||||
- [ ] 3.5 `MaintainPartitions` 改为成功表、失败表共用的四步归档(摘出、导出 `csv.gz.tmp`、校验并写 manifest 后改名、删除),并处理上一轮遗留的已摘出表、清理本地压缩包(默认保留 3 个月);常量改为 `TrafficCollectionRetentionMonths=2`。
|
||||
|
||||
## 4. 记录查询接口(第 3 次提交)
|
||||
|
||||
- [ ] 4.1 在 `internal/store/postgres` 中为各记录表实现统一的 `List(ctx, Filter)`;资产标识复用现有资产解析,解析不到时返回空页。
|
||||
- [ ] 4.2 采集成功列表 `GET /traffic-collections`、采集失败列表 `GET /traffic-collection-failures`;单卡采集接口增加筛选条件和"包含失败"参数,成功表查询按 `result='ok'` 过滤。
|
||||
- [ ] 4.3 停复机记录列表:`iccid` 扩展为资产标识,增加场景、原因、指令号筛选;单卡历史增加动作、结果、来源筛选;新建或重建 D9 中列出的索引。
|
||||
- [ ] 4.4 新增卡日流量列表;新增套餐日流量详单列表,原单套餐接口改为分页;新增冲正流水列表,原单套餐接口增加时间范围筛选。
|
||||
- [ ] 4.5 新增退款后处理尝试列表;退款列表增加 `asset_post_status` 筛选,列表和详情返回后处理次数、最后错误、下次重试时间。
|
||||
- [ ] 4.6 统一校验规则:时间范围必填、跨度不超过 31 天、分页上限;所有接口在 `RouteSpec` 中登记输入输出 DTO,并生成接口文档。
|
||||
|
||||
## 5. 时间线与测试环境部署配置(第 3 次提交)
|
||||
|
||||
- [ ] 5.1 先与前端或运营确认是否有页面依赖机器动作的审计,再在 `internal/query/audit` 中让卡和设备时间线合并停复机记录。
|
||||
- [ ] 5.2 `docker-compose.prod.yml`:api 挂载 `./api/logs`,worker 挂载 `./worker/logs` 和 `./worker/archive`;两者增加 `MAX_BACKUPS=40`、`MAX_AGE=14`。`.gitea/workflows/deploy.yaml` 改为创建这三个目录。
|
||||
- [ ] 5.3 构建、`openspec validate --all`、`./scripts/context-health.sh`;提交并推送 8-11。
|
||||
|
||||
## 6. 测试环境验证(AGENTS.md 规定自动化测试为 N/A;按 ENG-TEST-001 在测试环境验证)
|
||||
|
||||
- [ ] 6.1 审计:部署后 24 小时,审计日增约 300 条(按小时只读统计);人工操作照常写审计;`app.log` 中有 `logger=audit` 的行且不含快照;新对接日志的 `audit_event_id` 可为空。
|
||||
- [ ] 6.2 运行时覆盖:把一个机器动作覆盖为 `log_only`,再恢复,确认写入行为随之切换。
|
||||
- [ ] 6.3 退款后处理:在测试环境造一张后处理会失败的退款单,确认每次尝试都有记录、退避生效、第 10 次后转人工并写结论审计,人工重试成功后状态变为已完成。
|
||||
- [ ] 6.4 停复机:造一条执行前已是目标状态的停机指令,确认写入 `skipped` 记录。
|
||||
- [ ] 6.5 采集:成功记录带原始数据,失败记录带错误详情和原始数据,无套餐卡的失败也写入;记录中没有密钥、签名或加密报文。
|
||||
- [ ] 6.6 归档:在测试库用一个过期的空分区或小分区执行一次维护,确认生成 `csv.gz` 和 manifest、行数一致后分区被删除;模拟目录不可写时,已摘出的分区仍保留。
|
||||
- [ ] 6.7 查询接口:每个接口分别验证查单个、查全部、分页总数和各筛选条件;用 `EXPLAIN` 确认走索引并做了分区裁剪。
|
||||
- [ ] 6.8 部署:确认 `/opt/junhong_cmp/api/logs`、`/opt/junhong_cmp/worker/logs`、`/opt/junhong_cmp/worker/archive` 生成且轮换正常,同步服务器能拉到文件。
|
||||
|
||||
## 7. 归档顺序
|
||||
|
||||
- [ ] 7.1 本 Change 必须在 `add-card-traffic-queue-and-observed-deduction` 和 `add-carrier-gateway-rate-limit` 归档之后再归档。
|
||||
@@ -32,6 +32,14 @@ type Config struct {
|
||||
Worker WorkerConfig `mapstructure:"worker"`
|
||||
WeCom WeComConfig `mapstructure:"wecom"`
|
||||
Approval ApprovalConfig `mapstructure:"approval"`
|
||||
Audit AuditConfig `mapstructure:"audit"`
|
||||
}
|
||||
|
||||
// AuditConfig 统一审计写入配置。
|
||||
type AuditConfig struct {
|
||||
// PersistOverrides 按动作覆盖系统任务、定时任务的落库策略,格式 "动作编码=策略,动作编码=策略",
|
||||
// 策略取值 always、on_success、outcome、log_only、none;为空不覆盖。环境变量:JUNHONG_AUDIT_PERSIST_OVERRIDES
|
||||
PersistOverrides string `mapstructure:"persist_overrides"`
|
||||
}
|
||||
|
||||
// ServerConfig HTTP 服务器配置
|
||||
|
||||
@@ -162,6 +162,11 @@ worker:
|
||||
# 审计归档、Integration 归档与日留存默认关闭,低峰期按单日推进
|
||||
audit_archive_tasks_enabled: false
|
||||
|
||||
# 统一审计写入配置
|
||||
audit:
|
||||
# 按动作覆盖系统任务、定时任务的落库策略,例如 "package_usage.expire=log_only";为空不覆盖
|
||||
persist_overrides: ""
|
||||
|
||||
# 审批新旧入口切换配置
|
||||
approval:
|
||||
legacy_refund_manual_enabled: true
|
||||
|
||||
@@ -144,6 +144,7 @@ func bindEnvVariables(v *viper.Viper) {
|
||||
"worker.audit_archive_tasks_enabled",
|
||||
"approval.legacy_refund_manual_enabled",
|
||||
"approval.legacy_offline_recharge_pay_enabled",
|
||||
"audit.persist_overrides",
|
||||
"wecom.base_url",
|
||||
"wecom.timeout",
|
||||
"wechat.official_account.app_id",
|
||||
|
||||
Reference in New Issue
Block a user