fix

doc/Syslog-Trap接入与重放.md

@@ -1,87 +0,0 @@
-# Syslog-Trap 接入与重放
-
-## 接入目标
-
-`logs` 服务负责接收 Syslog 与 SNMP Trap，按字典和规则解析后写入 `logs_events`，并通过 `logs_alert_outbox` 异步转发到 `alert` 的原始事件池：
-
-```text
-Syslog / Trap -> logs_events -> logs_alert_outbox -> Alert/v1/raw-events/ingest
-```
-
-转发使用 `X-Internal-Key`，配置来自 `AlertForward.internal_key`。解析成功的事件 `parse_status=parsed`，未命中字典或规则的事件仍保存原始报文，并以 `parse_status=unparsed` 入队，便于规则调整后重放。
-
-## 部署配置
-
-`logs` 当前内置 UDP 接收器：
-
-```yaml
-Ingest:
-  syslog_listen_addr: "0.0.0.0:5140"
-  trap_listen_addr: "0.0.0.0:1620"
-  rule_refresh_secs: 30
-
-AlertForward:
-  enabled: true
-  base_url: "http://127.0.0.1:18080"
-  internal_key: "change-me"
-  default_policy_id: 1
-```
-
-生产环境如需标准端口 `514/162`，建议由 systemd socket、firewalld rich rule、iptables REDIRECT 或外层采集网关转发到非特权端口。TCP Syslog 接入建议在网关层启用 TCP listener，再转发到 UDP 或调用后续 HTTP ingest 入口；开启 TCP 时必须保留原始来源 IP 和 trace ID。
-
-## 字典与规则
-
-Trap 字典字段：
-
-- `vendor`：厂商，例如 `H3C`。
-- `oid`：精确 Trap OID。
-- `oid_prefix`：OID 前缀，兼容旧字典。
-- `name` / `title`：展示名称。
-- `severity_mapping_json`：级别映射 JSON。
-- `parse_expression`：解析 varbind 的表达式或正则。
-
-Syslog 规则字段：
-
-- `source_match`：来源 IP、主机名或原始行子串。
-- `message_regex`：消息正文正则。
-- `severity_mapping_json`：按正则映射平台级别。
-- `resource_uid_extract_regex`：提取 `resource_uid`，优先使用命名分组 `resource_uid`。
-
-示例 Syslog：
-
-```text
-<189>Jun 24 10:00:01 h3c-core-01 IFNET/4/LINK_DOWN: Interface GigabitEthernet1/0/1 is down, resource_uid=network:h3c-core-01
-```
-
-示例 H3C Trap OID：
-
-```text
-1.3.6.1.6.3.1.1.5.3
-```
-
-## 未解析队列与重放
-
-未解析事件仍写入 `logs_events`，并创建 outbox payload：
-
-- `source_type=syslog` 或 `trap`
-- `parse_status=unparsed`
-- `raw_payload` 保存原始报文或 varbind 摘要
-
-重放接口：
-
-```http
-POST /Logs/v1/entries/{id}/replay
-Authorization: Bearer <jwt>
-```
-
-成功响应会返回新的 `outbox_id`。重放 payload 使用 `parse_status=replayed`，并带上 `labels.replay_of_log_event_id`，前端可在“日志查询 -> 重放结果”中查看发送结果，失败任务可人工重试。
-
-## Smoke 样例
-
-输出 H3C Syslog 与 Trap 示例载荷：
-
-```powershell
-C:\Users\27105\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe scripts\test_alert_receive_smoke.py --print-log-samples
-```
-
-这些样例用于准备 UDP/TCP 接收器 smoke 数据，也可作为联调 alert 原始事件池时的期望字段参考。

doc/日志监控.md

@@ -1,498 +0,0 @@
-# Ops Logs 前端页面设计文档（Log Mgmt）
-
-## 1. 背景与目标
-
-`Logs` 服务负责采集并归一化设备侧日志（Syslog / SNMP Trap），并提供规则与字典等配置能力。前端需要在统一的后台界面中完成：
-
-1. 日志查询（查看归一化后的日志事件及详情）
-2. Syslog 规则配置
-3. Trap 规则配置
-4. Trap 字典配置
-5. Trap 屏蔽/抑制规则配置
-
-本设计以当前代码库的后端模型与前端实现为准：后端路由在 `internal/routers/register.go`，前端页面在 `front/src/views/ops/pages/log-mgmt/**/index.vue`。
-
----
-
-## 2. 范围（页面数量与路由）
-
-本模块共 5 个页面，对应后端 5 组资源（列表+CRUD 或列表+详情抽屉）。
-
-| 页面 | 菜单/路由路径 | 前端组件 |
-|---|---|---|
-| 日志查询 | `/log-mgmt/entries` | `front/src/views/ops/pages/log-mgmt/entries/index.vue` |
-| Syslog 匹配规则 | `/log-mgmt/syslog-rules` | `front/src/views/ops/pages/log-mgmt/syslog-rules/index.vue` |
-| SNMP Trap 匹配规则 | `/log-mgmt/trap-rules` | `front/src/views/ops/pages/log-mgmt/trap-rules/index.vue` |
-| Trap 字典 | `/log-mgmt/trap-dictionary` | `front/src/views/ops/pages/log-mgmt/trap-dictionary/index.vue` |
-| Trap 屏蔽/抑制 | `/log-mgmt/trap-suppressions` | `front/src/views/ops/pages/log-mgmt/trap-suppressions/index.vue` |
-
-路由与菜单配置参考：
-- `front/src/router/local-menu-flat.ts` / `front/src/router/local-menu-items.ts`
-- `front/src/views/ops/pages/system-settings/system-logs/index.vue`（页面入口按钮）
-- `front/src/views/ops/pages/monitor/log/index.vue`（嵌入 `LogMgmtEntries`）
-
----
-
-## 3. 数据对象与接口映射
-
-后端认证：API 路由组启用 `middleware.JwtAuth(true)`。
-
-前端请求的 API Base：`front/src/api/ops/logs.ts` 中为 `/Logs/v1`。
-
-### 3.1 日志事件（entries）
-
-- 接口：`GET /Logs/v1/entries`
-- 返回结构（前端类型）：`LogEntriesResult`（`total`、`page`、`page_size`、`items`）
-- 日志事件字段（前端类型 `LogEvent`）：
-  - `id`
-  - `created_at`
-  - `source_kind`（`syslog` / `snmp_trap`）
-  - `remote_addr`
-  - `raw_payload`
-  - `normalized_summary`
-  - `normalized_detail`
-  - `device_name`
-  - `severity_code`
-  - `trap_oid`
-  - `alert_sent`
-
-后端实现：`internal/models/log_event.go`、`internal/logic/controllers/crud.go`（`ListLogEvents`）。
-
-### 3.2 Syslog 规则（syslog-rules）
-
-- 接口：
-  - `GET /Logs/v1/syslog-rules`
-  - `POST /Logs/v1/syslog-rules`
-  - `PUT /Logs/v1/syslog-rules/:id`
-  - `DELETE /Logs/v1/syslog-rules/:id`
-- 规则字段（前端类型 `SyslogRule` / 后端 `SyslogRule`）：
-  - `id`、`created_at`、`updated_at`
-  - `name`
-  - `enabled`
-  - `priority`
-  - `device_name_contains`
-  - `keyword_regex`
-  - `alert_name`
-  - `severity_code`
-  - `policy_id`
-
-后端实现：`internal/models/syslog_rule.go`、`internal/logic/controllers/crud.go`。
-
-### 3.3 Trap 规则（trap-rules）
-
-- 接口：
-  - `GET /Logs/v1/trap-rules`
-  - `POST /Logs/v1/trap-rules`
-  - `PUT /Logs/v1/trap-rules/:id`
-  - `DELETE /Logs/v1/trap-rules/:id`
-- 规则字段（前端类型 `TrapRule` / 后端 `TrapRule`）：
-  - `name`
-  - `enabled`
-  - `priority`
-  - `oid_prefix`
-  - `varbind_match_regex`
-  - `alert_name`
-  - `severity_code`
-  - `policy_id`
-
-后端实现：`internal/models/trap_rule.go`、`internal/logic/controllers/crud.go`。
-
-### 3.4 Trap 字典（trap-dictionary）
-
-- 接口：
-  - `GET /Logs/v1/trap-dictionary`
-  - `POST /Logs/v1/trap-dictionary`
-  - `PUT /Logs/v1/trap-dictionary/:id`
-  - `DELETE /Logs/v1/trap-dictionary/:id`
-- 字典条目字段（前端类型 `TrapDictionaryEntry` / 后端 `TrapDictionaryEntry`）：
-  - `oid_prefix`（后端约束：uniqueIndex）
-  - `title`
-  - `description`
-  - `severity_code`
-  - `recovery_message`
-  - `enabled`
-
-后端实现：`internal/models/trap_dictionary.go`、`internal/logic/controllers/crud.go`。
-
-### 3.5 Trap 屏蔽/抑制（trap-suppressions）
-
-- 接口：
-  - `GET /Logs/v1/trap-suppressions`
-  - `POST /Logs/v1/trap-suppressions`
-  - `PUT /Logs/v1/trap-suppressions/:id`
-  - `DELETE /Logs/v1/trap-suppressions/:id`
-- 屏蔽规则字段（前端类型 `TrapShield` / 后端 `TrapShield`）：
-  - `name`
-  - `enabled`
-  - `source_ip_cidr`
-  - `oid_prefix`
-  - `interface_hint`
-  - `time_windows_json`（JSON 字符串）
-
-后端实现：`internal/models/trap_shield.go`、`internal/logic/controllers/crud.go`。
-
----
-
-## 4. 页面设计详情（逐页）
-
-### 4.1 日志查询页（`/log-mgmt/entries`）
-
-目标：以“可筛选的列表 + 详情抽屉”方式查看归一化日志事件。
-
-#### 1）顶部筛选区
-
-- 使用 `search-table` 组件
-- 筛选项：`source_kind`（下拉）
-  - `全部`（value=''）
-  - `Syslog`（value='syslog'）
-  - `SNMP Trap`（value='snmp_trap'）
-
-筛选触发：`@search` 调用 `handleSearch`，重置则 `@reset` 调用 `handleReset`。
-
-#### 2）列表表格列（Columns）
-
-表格由 `columns` 定义，主要列：
-
-- `ID`
-- `来源`：`source_kind`，通过 `sourceKindLabel()` 显示（`syslog`->`Syslog`、`snmp_trap`->`SNMP Trap`）
-- `时间`：`created_at`
-- `来源地址`：`remote_addr`
-- `设备`：`device_name`
-- `级别`：`severity_code`
-- `OID`：`trap_oid`
-- `原始报文`：`raw_payload`
-  - 使用 slot `raw_payload`：省略显示，保留 `tooltip`
-- `已告警`：`alert_sent`
-  - 使用 slot `alert_sent`：`a-tag`（已转发/否）
-- `操作`：slot `operations`
-  - `详情`：打开右侧抽屉
-
-#### 3）详情抽屉（a-drawer）
-
-- 打开逻辑：点击表格行操作中的 `详情`，调用 `openDetail(record)`
-- 抽屉展示：`a-descriptions`（1 列，bordered）
-- 展示字段：
-  - 来源类型（`source_kind`）
-  - 采集时间（`created_at`）
-  - 来源地址（`remote_addr`，空则 `-`）
-  - 设备名（`device_name`）
-  - 严重级别（`severity_code`）
-  - Trap OID（`trap_oid`）
-  - 已转发告警（`alert_sent`）
-  - 摘要（`normalized_summary`）
-  - 详情（`normalized_detail`，`pre-block` 预格式化展示）
-  - 原始报文（`raw_payload`，`pre-block` 预格式化展示）
-
-#### 4）分页策略
-
-- 分页参数由前端 `pagination.current/pageSize` 控制，并随筛选条件一起请求后端：
-  - 调用 `fetchLogEntries({ page, page_size, source_kind })`
-
-### 4.2 Syslog 规则页（`/log-mgmt/syslog-rules`）
-
-目标：规则的“列表 + 新建/编辑弹窗 + 删除确认”。
-
-#### 1）通用列表与本地过滤
-
-- 使用 `search-table`，并在前端进行“关键词本地过滤”，过滤字段：
-  - `name`
-  - `alert_name`
-  - `keyword_regex`
-- 搜索输入字段：
-  - `keyword`（label：`关键词`，placeholder：`规则名 / 告警名`）
-
-说明：该页（以及 trap-*、dictionary、suppressions 三类列表页）采用“先拉取全量 -> 本地过滤 -> 切片分页”的方式。
-
-#### 2）表格列
-
-- `ID`
-- `名称`：`name`
-- `优先级`：`priority`
-- `启用`：`enabled`（slot `enabled`，tag：启用/禁用）
-- `设备名包含`：`device_name_contains`
-- `关键字正则`：`keyword_regex`
-- `告警名`：`alert_name`
-- `级别`：`severity_code`
-- `策略ID`：`policy_id`
-- `操作`：编辑/删除
-
-#### 3）新建/编辑弹窗（a-modal）
-
-- 弹窗标题：
-  - 新建：`新建 Syslog 规则`
-  - 编辑：`编辑规则 #${editingId}`
-- 表单 `a-form`（布局 `vertical`）
-- 表单字段：
-  - `name`：`a-input`（必填）
-  - `enabled`：`a-switch`
-  - `priority`：`a-input-number`
-  - `device_name_contains`：`a-input`
-  - `keyword_regex`：`a-input`
-  - `alert_name`：`a-input`
-  - `severity_code`：`a-input`
-  - `policy_id`：`a-input-number`（min=0）
-
-提交逻辑：
-- 编辑：`updateSyslogRule(editingId, { ...formData })`
-- 新建：`createSyslogRule({ ...formData })`
-- 成功后关闭弹窗并刷新列表 `fetchList()`。
-
-#### 4）删除确认
-
-- `Modal.confirm` 二次确认
-- 删除接口：`deleteSyslogRule(id)`
-
-### 4.3 Trap 规则页（`/log-mgmt/trap-rules`）
-
-目标：TrapRule 的列表+弹窗 CRUD，与 Syslog 规则页同构。
-
-#### 1）本地过滤关键词
-
-- 字段：`keyword`
-- 匹配来源：
-  - `name`
-  - `oid_prefix`
-  - `alert_name`
-
-#### 2）表格列
-
-- `ID`、`名称`、`优先级`、`启用`
-- `OID 前缀`：`oid_prefix`
-- `Varbind 正则`：`varbind_match_regex`
-- `告警名`：`alert_name`
-- `级别`：`severity_code`
-- `策略ID`：`policy_id`
-- 操作：编辑/删除
-
-#### 3）弹窗表单字段
-
-- `name`（必填）
-- `enabled`
-- `priority`
-- `oid_prefix`
-- `varbind_match_regex`
-- `alert_name`
-- `severity_code`
-- `policy_id`（min=0）
-
-### 4.4 Trap 字典页（`/log-mgmt/trap-dictionary`）
-
-目标：TrapDictionaryEntry 的列表+弹窗 CRUD。
-
-#### 1）本地过滤关键词
-
-- 匹配字段：
-  - `oid_prefix`
-  - `title`
-  - `description`
-
-#### 2）表格列
-
-- `ID`
-- `OID 前缀`：`oid_prefix`
-- `标题`：`title`
-- `级别`：`severity_code`
-- `启用`：`enabled`
-- `描述`：`description`
-- 操作：编辑/删除
-
-#### 3）弹窗表单字段
-
-- `oid_prefix`（必填，建议提示“唯一前缀”）
-- `title`（必填）
-- `description`：`a-textarea`（rows=3）
-- `severity_code`
-- `enabled`
-- `recovery_message`：`a-textarea`（rows=2）
-
-### 4.5 Trap 屏蔽/抑制页（`/log-mgmt/trap-suppressions`）
-
-目标：TrapShield 的列表+弹窗 CRUD，并对 `time_windows_json` 做前端校验。
-
-#### 1）本地过滤关键词
-
-- 匹配字段：
-  - `name`
-  - `oid_prefix`
-  - `source_ip_cidr`
-
-#### 2）表格列
-
-- `ID`
-- `名称`：`name`
-- `启用`：`enabled`
-- `源 IP/CIDR`：`source_ip_cidr`
-- `OID 前缀`：`oid_prefix`
-- `接口提示`：`interface_hint`
-- 操作：编辑/删除
-
-#### 3）弹窗表单字段
-
-- `name`（必填）
-- `enabled`
-- `source_ip_cidr`
-- `oid_prefix`
-- `interface_hint`
-- `time_windows_json`：`a-textarea`（rows=4，placeholder=`{}`）
-
-#### 4）time_windows_json JSON 校验
-
-- 当 `time_windows_json` 非空时：
-  - 去 `trim` 后尝试 `JSON.parse(tw)`
-  - 校验失败：`Message.warning('时间窗 JSON 格式无效')` 并阻止提交
-
----
-
-## 5. 页面交互一致性要求（实现要点）
-
-为了保证各列表页体验一致，本模块约定：
-
-1. 列表页使用统一的 `search-table` 布局（顶部搜索、表格、分页、刷新）
-2. 规则类/字典/屏蔽页采用“拉取全量 -> 本地过滤 -> 切片分页”的方式
-3. 创建/编辑统一使用 `a-modal`，提交按钮触发 `formRef.validate()`
-4. 删除统一使用 `Modal.confirm`，成功后刷新列表并给出 `Message.success`
-5. `trap-suppressions` 对 `time_windows_json` 进行 JSON 字符串合法性校验
-
----
-
-## 6. 数据流（简图）
-
-```mermaid
-flowchart LR
-  UI[前端页面（search-table + 表格/弹窗/抽屉）] --> API[front/src/api/ops/logs.ts]
-  API --> BE[后端路由 internal/routers/register.go]
-  BE --> DB[(Postgres)]
-  BE --> Refresh[ingest.Global.Refresh()（规则/字典/屏蔽变更后触发）]
-```
-
----
-
-## 7. 中优先级待办（已立项，未完成）
-
-本节用于记录当前版本可用但尚未产品化完善的中优先级项，作为后续迭代输入。
-
-### 7.1 Outbox 可观测性增强
-
-当前状态：
-- 已支持 `alert_outbox` 入队、重试、死信、手动重试；
-- 已有基础列表查询接口和前端入口。
-
-待完善内容：
-- 增加 outbox 指标接口或埋点：
-  - `pending_count`
-  - `retrying_count`
-  - `dead_count`
-  - `dispatch_success_rate`
-  - `dispatch_latency_p95`
-- 增加失败原因聚合视图（按 `last_error` 分类统计）。
-- 增加任务生命周期字段（首次入队时间、最后发送时间）用于问题排查。
-
-建议落地文件：
-- 后端：`internal/logic/controllers/outbox.go`、`internal/ingest/alert_outbox.go`
-- 前端：`front/src/views/ops/pages/log-mgmt/entries/index.vue`
-
-### 7.2 分发状态模型统一（替代 bool）
-
-当前状态：
-- `logs_events` 已新增 `dispatch_status`，并在 outbox 流程中维护状态。
-- 历史字段 `alert_sent` 仍保留，用于兼容旧页面展示。
-
-待完善内容：
-- 明确状态枚举为：`not_applicable/pending/retrying/sent/dead`。
-- 前后端统一以 `dispatch_status` 作为主状态字段，`alert_sent` 逐步降级为派生字段或移除。
-- 页面文案由“已告警”升级为“分发状态”主展示，避免语义歧义。
-
-建议落地文件：
-- 后端：`internal/models/log_event.go`、`internal/logic/controllers/crud.go`
-- 前端：`front/src/api/ops/logs.ts`、`front/src/views/ops/pages/log-mgmt/entries/index.vue`
-
-### 7.3 关键路径测试补齐
-
-当前状态：
-- 已有基础单测覆盖核心函数。
-
-待完善内容：
-- 增加资源事件安全链路测试：
-  - 验签失败/成功
-  - 超时事件拒绝
-  - 幂等事件重复提交
-- 增加 outbox 重试链路测试：
-  - 发送成功更新状态
-  - 重试次数递增
-  - 超过阈值转 `dead`
-- 增加资源冲突优先级测试：
-  - `server > collector > device`
-
-建议落地文件：
-- `internal/logic/controllers/resource_event_test.go`
-- `internal/ingest/alert_outbox_test.go`
-- `internal/ingest/resource_resolver_test.go`
-
----
-
-## 8. 后续产品化规划（Phase 3）
-
-本节对应“可运维与产品化”阶段，优先级低于中优先级修复项，但会显著提升系统可管理性。
-
-### 8.1 规则发布流（draft / publish / rollback）
-
-目标：
-- 规则配置与生效状态解耦，降低误操作风险。
-
-范围：
-- 引入规则草稿态与发布态；
-- 支持发布记录、回滚到历史版本；
-- 变更需记录操作人、时间、变更说明。
-
-接口建议：
-- `POST /Logs/v1/rule-sets/:id/publish`
-- `POST /Logs/v1/rule-sets/:id/rollback`
-- `GET /Logs/v1/rule-sets/:id/history`
-
-### 8.2 规则仿真/回放能力
-
-目标：
-- 上线前可验证规则命中结果，减少误报漏报。
-
-范围：
-- 输入样本报文（syslog/trap）执行仿真；
-- 返回命中链路（命中/未命中原因）；
-- 支持历史事件回放。
-
-接口建议：
-- `POST /Logs/v1/rule-sets/:id/simulate`
-- `POST /Logs/v1/rule-sets/:id/replay`
-
-### 8.3 指标与审计面板
-
-目标：
-- 建立“采集-匹配-分发”全链路可观测性。
-
-范围：
-- 采集侧：接收速率、解析失败率；
-- 匹配侧：命中率、规则耗时；
-- 分发侧：成功率、重试率、死信量；
-- 安全侧：验签失败次数、重放拦截次数。
-
-前端建议：
-- 在日志管理模块增加“运行指标”页签；
-- 对死信和验签失败提供快捷定位入口。
-
----
-
-## 9. 未完成项执行顺序（建议）
-
-为降低风险，建议按以下顺序推进：
-
-1. **中优先级先完成**
-   - outbox 指标与失败聚合
-   - `dispatch_status` 主状态化
-   - 关键路径测试补齐
-2. **再做产品化**
-   - 规则发布流
-   - 规则仿真/回放
-   - 指标与审计面板
-
-验收建议：
-- 每项功能完成后执行“单项验证 + 回归验证”，最后统一做端到端联调。
-

etc/logs_dev.yaml

@@ -20,12 +20,12 @@ Ingest:
   trap_listen_addr: "0.0.0.0:9162"
   rule_refresh_secs: 30
 
-AlertForward:
-  enabled: true
-  base_url: https://ops-api.apinb.com
-  internal_key: "ops-alert"
-  default_policy_id: 0
-
-ResourceEvent:
-  hmac_secret: "replace-with-dc-control-shared-secret"
-  max_skew_secs: 300
+AlertForward:
+  enabled: true
+  base_url: https://ops-api.apinb.com
+  internal_key: ${LOGS_ALERT_SECRET}
+  default_policy_id: 0
+
+ResourceEvent:
+  hmac_secret: ${DC_CONTROL_LOGS_EVENT_SECRET}
+  max_skew_secs: 300

etc/logs_prod.yaml

@@ -20,12 +20,12 @@ Ingest:
   trap_listen_addr: "0.0.0.0:9162"
   rule_refresh_secs: 30
 
-AlertForward:
-  enabled: true
-  base_url: https://ops-api.apinb.com
-  internal_key: "ops-alert"
-  default_policy_id: 0
-
-ResourceEvent:
-  hmac_secret: "replace-with-dc-control-shared-secret"
-  max_skew_secs: 300
+AlertForward:
+  enabled: true
+  base_url: https://ops-api.apinb.com
+  internal_key: ${LOGS_ALERT_SECRET}
+  default_policy_id: 0
+
+ResourceEvent:
+  hmac_secret: ${DC_CONTROL_LOGS_EVENT_SECRET}
+  max_skew_secs: 300

etc/ops-logs.service

@@ -0,0 +1,28 @@
+[Unit]
+Description=OPS Logs Service
+Wants=network-online.target
+Requires=ops-mgt.service
+After=network-online.target ops-mgt.service
+PartOf=ops-stack.target
+
+[Service]
+Type=simple
+WorkingDirectory=/data/app
+EnvironmentFile=/data/app/etc/ops.env
+Environment=BSM_RuntimeMode=prod
+Environment=RUN_MODE=prod
+Environment=BSM_Prefix=/data/app
+ExecStart=/data/app/ops-logs
+ExecStartPost=/data/app/systemd/wait-http.sh ops-logs http://127.0.0.1:12440/Logs/v1/ping/hello 60
+Restart=on-failure
+RestartSec=5s
+TimeoutStartSec=75s
+TimeoutStopSec=30s
+KillSignal=SIGTERM
+StandardOutput=journal
+StandardError=journal
+SyslogIdentifier=ops-logs
+LimitNOFILE=1048576
+
+[Install]
+WantedBy=ops-stack.target

internal/config/config.go

@@ -1,51 +1,58 @@
-package config
-
-import (
-	"net"
-
-	"git.apinb.com/bsm-sdk/core/conf"
-)
-
-var Spec SrvConfig
-
-type AlertForwardConf struct {
-	BaseURL         string `yaml:"base_url"`
-	InternalKey     string `yaml:"internal_key"`
-	Enabled         bool   `yaml:"enabled"`
-	DefaultPolicyID uint   `yaml:"default_policy_id"`
-}
-
-type IngestConf struct {
-	SyslogListenAddr string `yaml:"syslog_listen_addr"`
-	TrapListenAddr   string `yaml:"trap_listen_addr"`
-	RuleRefreshSecs  int    `yaml:"rule_refresh_secs"`
-}
-
-type ResourceEventConf struct {
-	// HMACSecret 用于校验 dc-control 推送签名（X-Event-Signature）。
-	HMACSecret string `yaml:"hmac_secret"`
-	// MaxSkewSecs 允许事件时间与服务端时间的最大偏差（秒）。
-	MaxSkewSecs int `yaml:"max_skew_secs"`
-}
-
-type SrvConfig struct {
-	conf.Base      `yaml:",inline"`
-	Databases      *conf.DBConf            `yaml:"Databases"`
-	MicroService   *conf.MicroServiceConf  `yaml:"MicroService"`
-	Rpc            map[string]conf.RpcConf `yaml:"Rpc"`
-	Gateway        *conf.GatewayConf       `yaml:"Gateway"`
-	Apm            *conf.ApmConf           `yaml:"APM"`
-	Etcd           *conf.EtcdConf          `yaml:"Etcd"`
-	AlertForward   *AlertForwardConf       `yaml:"AlertForward"`
-	Ingest         IngestConf              `yaml:"Ingest"`
-	ResourceEvent  ResourceEventConf       `yaml:"ResourceEvent"`
-}
-
-func New(srvKey string) {
-	conf.New(srvKey, &Spec)
-	Spec.Port = conf.CheckPort(Spec.Port)
-	Spec.BindIP = conf.CheckIP(Spec.BindIP)
-	Spec.Addr = net.JoinHostPort(Spec.BindIP, Spec.Port)
-	conf.NotNil(Spec.Service, Spec.Cache)
-	conf.PrintInfo(Spec.Addr)
-}
+package config
+
+import (
+	"net"
+	"strings"
+
+	"git.apinb.com/bsm-sdk/core/conf"
+)
+
+var Spec SrvConfig
+
+type AlertForwardConf struct {
+	BaseURL         string `yaml:"base_url"`
+	InternalKey     string `yaml:"internal_key"`
+	Enabled         bool   `yaml:"enabled"`
+	DefaultPolicyID uint   `yaml:"default_policy_id"`
+}
+
+type IngestConf struct {
+	SyslogListenAddr string `yaml:"syslog_listen_addr"`
+	TrapListenAddr   string `yaml:"trap_listen_addr"`
+	RuleRefreshSecs  int    `yaml:"rule_refresh_secs"`
+}
+
+type ResourceEventConf struct {
+	// HMACSecret 用于校验 dc-control 推送签名（X-Event-Signature）。
+	HMACSecret string `yaml:"hmac_secret"`
+	// MaxSkewSecs 允许事件时间与服务端时间的最大偏差（秒）。
+	MaxSkewSecs int `yaml:"max_skew_secs"`
+}
+
+type SrvConfig struct {
+	conf.Base     `yaml:",inline"`
+	Databases     *conf.DBConf            `yaml:"Databases"`
+	MicroService  *conf.MicroServiceConf  `yaml:"MicroService"`
+	Rpc           map[string]conf.RpcConf `yaml:"Rpc"`
+	Gateway       *conf.GatewayConf       `yaml:"Gateway"`
+	Apm           *conf.ApmConf           `yaml:"APM"`
+	Etcd          *conf.EtcdConf          `yaml:"Etcd"`
+	AlertForward  *AlertForwardConf       `yaml:"AlertForward"`
+	Ingest        IngestConf              `yaml:"Ingest"`
+	ResourceEvent ResourceEventConf       `yaml:"ResourceEvent"`
+}
+
+func New(srvKey string) {
+	conf.New(srvKey, &Spec)
+	Spec.Port = conf.CheckPort(Spec.Port)
+	Spec.BindIP = conf.CheckIP(Spec.BindIP)
+	Spec.Addr = net.JoinHostPort(Spec.BindIP, Spec.Port)
+	Spec.ResourceEvent.HMACSecret = strings.TrimSpace(Spec.ResourceEvent.HMACSecret)
+	conf.NotNil(Spec.Service, Spec.Cache, Spec.ResourceEvent.HMACSecret)
+	if Spec.AlertForward != nil && Spec.AlertForward.Enabled {
+		Spec.AlertForward.BaseURL = strings.TrimSpace(Spec.AlertForward.BaseURL)
+		Spec.AlertForward.InternalKey = strings.TrimSpace(Spec.AlertForward.InternalKey)
+		conf.NotNil(Spec.AlertForward.BaseURL, Spec.AlertForward.InternalKey)
+	}
+	conf.PrintInfo(Spec.Addr)
+}

internal/impl/connections.go

@@ -0,0 +1,29 @@
+package impl
+
+import (
+	"fmt"
+
+	"git.apinb.com/bsm-sdk/core/cache/redis"
+	"git.apinb.com/bsm-sdk/core/conf"
+	"git.apinb.com/bsm-sdk/core/database"
+	"git.apinb.com/bsm-sdk/core/vars"
+	"gorm.io/gorm"
+)
+
+func newDatabase(cfg *conf.DBConf) *gorm.DB {
+	if cfg == nil || len(cfg.Source) == 0 {
+		panic("database source is required")
+	}
+	db, err := database.NewDatabase(cfg.Driver, cfg.Source, nil)
+	if err != nil {
+		panic(fmt.Sprintf("database init failed: %v", err))
+	}
+	return db
+}
+
+func newRedisCache(dsn string) *redis.RedisClient {
+	if dsn == "" {
+		return nil
+	}
+	return redis.New(dsn, vars.ServiceKey)
+}

internal/impl/impl.go

@@ -1,32 +1,31 @@
-package impl
-
-import (
-	"fmt"
-
-	"git.apinb.com/bsm-sdk/core/cache/redis"
-	"git.apinb.com/bsm-sdk/core/logger"
-	"git.apinb.com/bsm-sdk/core/with"
-	"git.apinb.com/ops/logs/internal/config"
-	"git.apinb.com/ops/logs/internal/models"
-	"gorm.io/gorm"
-)
-
-var (
-	RedisService *redis.RedisClient
-	DBService    *gorm.DB
-)
-
-func NewImpl() {
-	RedisService = with.RedisCache(config.Spec.Cache)
-	DBService = with.Databases(config.Spec.Databases, nil)
-	logger.New(nil)
-
-	if DBService != nil {
-		if err := DBService.AutoMigrate(models.GetAllModels()...); err != nil {
-			panic(fmt.Sprintf("logs migrate: %v", err))
-		}
-		if err := models.InitData(DBService); err != nil {
-			panic(fmt.Sprintf("logs init data: %v", err))
-		}
-	}
-}
+package impl
+
+import (
+	"fmt"
+
+	"git.apinb.com/bsm-sdk/core/cache/redis"
+	"git.apinb.com/bsm-sdk/core/logger"
+	"git.apinb.com/ops/logs/internal/config"
+	"git.apinb.com/ops/logs/internal/models"
+	"gorm.io/gorm"
+)
+
+var (
+	RedisService *redis.RedisClient
+	DBService    *gorm.DB
+)
+
+func NewImpl() {
+	RedisService = newRedisCache(config.Spec.Cache)
+	DBService = newDatabase(config.Spec.Databases)
+	logger.New(nil)
+
+	if DBService != nil {
+		if err := DBService.AutoMigrate(models.GetAllModels()...); err != nil {
+			panic(fmt.Sprintf("logs migrate: %v", err))
+		}
+		if err := models.InitData(DBService); err != nil {
+			panic(fmt.Sprintf("logs init data: %v", err))
+		}
+	}
+}

internal/ingest/alert_forward.go

@@ -2,7 +2,10 @@ package ingest
 
 import (
 	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -12,40 +15,57 @@ import (
 	"git.apinb.com/ops/logs/internal/config"
 )
 
-const alertResponseBodyLimit = 1 << 20
+const (
+	alertResponseBodyLimit = 1 << 20
+	maxAlertTraceIDLength  = 96
+)
+
+var errAlertForwardDisabled = errors.New("Alert 转发未启用或 base_url 为空")
+
+type alertForwardResponse struct {
+	Code    *int32          `json:"code"`
+	Details json.RawMessage `json:"details"`
+}
 
 // AlertReceiveBody 与 alert ReceiveRequest 对齐（含必填 raw_data）
 type AlertReceiveBody struct {
-	AlertName    string            `json:"alert_name"`
-	Summary      string            `json:"summary"`
-	Description  string            `json:"description"`
-	SeverityCode string            `json:"severity_code"`
-	Value        string            `json:"value"`
-	Threshold    string            `json:"threshold"`
-	Labels       map[string]string `json:"labels"`
-	Agent        string            `json:"agent"`
-	PolicyID     uint              `json:"policy_id"`
-	RawData      json.RawMessage   `json:"raw_data"`
+	AlertName      string            `json:"alert_name"`
+	Summary        string            `json:"summary"`
+	Description    string            `json:"description"`
+	SeverityCode   string            `json:"severity_code"`
+	Value          string            `json:"value"`
+	Threshold      string            `json:"threshold"`
+	Labels         map[string]string `json:"labels"`
+	Agent          string            `json:"agent"`
+	PolicyID       uint              `json:"policy_id"`
+	Fingerprint    string            `json:"fingerprint,omitempty"`
+	State          string            `json:"state,omitempty"`
+	SourceEventKey string            `json:"source_event_key"`
+	TraceID        string            `json:"trace_id"`
+	OccurredAt     time.Time         `json:"occurred_at"`
+	RawData        json.RawMessage   `json:"raw_data"`
 }
 
 type RawEventIngestBody struct {
-	SourceType  string            `json:"source_type"`
-	ResourceUID string            `json:"resource_uid,omitempty"`
-	EventTime   time.Time         `json:"event_time"`
-	Severity    string            `json:"severity"`
-	Title       string            `json:"title"`
-	Message     string            `json:"message"`
-	Labels      map[string]string `json:"labels,omitempty"`
-	Annotations map[string]string `json:"annotations,omitempty"`
-	ParseStatus string            `json:"parse_status"`
-	RawPayload  json.RawMessage   `json:"raw_payload"`
-	TraceID     string            `json:"trace_id,omitempty"`
+	SourceType     string            `json:"source_type"`
+	SourceEventKey string            `json:"source_event_key"`
+	State          string            `json:"state,omitempty"`
+	ResourceUID    string            `json:"resource_uid,omitempty"`
+	EventTime      time.Time         `json:"event_time"`
+	Severity       string            `json:"severity"`
+	Title          string            `json:"title"`
+	Message        string            `json:"message"`
+	Labels         map[string]string `json:"labels,omitempty"`
+	Annotations    map[string]string `json:"annotations,omitempty"`
+	ParseStatus    string            `json:"parse_status"`
+	RawPayload     json.RawMessage   `json:"raw_payload"`
+	TraceID        string            `json:"trace_id,omitempty"`
 }
 
 func forwardAlert(body AlertReceiveBody) error {
 	cfg := config.Spec.AlertForward
 	if cfg == nil || !cfg.Enabled || cfg.BaseURL == "" {
-		return nil
+		return errAlertForwardDisabled
 	}
 	if len(body.RawData) == 0 {
 		return fmt.Errorf("raw_data 不能为空")
@@ -56,19 +76,48 @@ func forwardAlert(body AlertReceiveBody) error {
 	if body.PolicyID == 0 && cfg.DefaultPolicyID > 0 {
 		body.PolicyID = cfg.DefaultPolicyID
 	}
+	body.TraceID = ensureAlertTraceID(body.TraceID, body.SourceEventKey)
 	raw, err := json.Marshal(body)
 	if err != nil {
 		return err
 	}
-	return postAlertPayload(cfg, "/Alert/v1/alerts/receive", raw)
+	result, err := postAlertPayload(cfg, "/Alert/v1/alerts/receive", raw, body.TraceID)
+	if err != nil {
+		return err
+	}
+	var details struct {
+		ID            uint   `json:"id"`
+		RawEventID    uint   `json:"raw_event_id"`
+		AlertRecordID uint   `json:"alert_record_id"`
+		IncidentID    uint   `json:"incident_id"`
+		Status        string `json:"status"`
+		Fingerprint   string `json:"fingerprint"`
+	}
+	if err := json.Unmarshal(result.Details, &details); err != nil {
+		return fmt.Errorf("Alert 响应 details 无效：%v；请稍后重试", err)
+	}
+	alertRecordID := details.AlertRecordID
+	if alertRecordID == 0 {
+		alertRecordID = details.ID
+	}
+	if details.RawEventID == 0 || alertRecordID == 0 || details.IncidentID == 0 ||
+		(details.ID != 0 && details.AlertRecordID != 0 && details.ID != details.AlertRecordID) ||
+		(details.Status != "firing" && details.Status != "resolved") || !validFingerprint(details.Fingerprint) {
+		return fmt.Errorf("Alert 响应缺少完整写入结果，无法确认转发成功；请稍后重试")
+	}
+	if body.Fingerprint != "" && details.Fingerprint != body.Fingerprint {
+		return fmt.Errorf("Alert 返回的告警指纹与请求不一致；请稍后重试")
+	}
+	return nil
 }
 
-func postAlertPayload(cfg *config.AlertForwardConf, path string, payload []byte) error {
-	req, err := http.NewRequest(http.MethodPost, cfg.BaseURL+path, bytes.NewReader(payload))
+func postAlertPayload(cfg *config.AlertForwardConf, path string, payload []byte, traceID string) (*alertForwardResponse, error) {
+	req, err := http.NewRequest(http.MethodPost, strings.TrimRight(strings.TrimSpace(cfg.BaseURL), "/")+path, bytes.NewReader(payload))
 	if err != nil {
-		return fmt.Errorf("创建 Alert 转发请求失败：%w", err)
+		return nil, fmt.Errorf("创建 Alert 转发请求失败：%w", err)
 	}
 	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Trace-Id", traceID)
 	if cfg.InternalKey != "" {
 		req.Header.Set("X-Internal-Key", cfg.InternalKey)
 	}
@@ -80,34 +129,35 @@ func postAlertPayload(cfg *config.AlertForwardConf, path string, payload []byte)
 	}
 	resp, err := client.Do(req)
 	if err != nil {
-		return fmt.Errorf("发送 Alert 转发请求失败：%w", err)
+		return nil, fmt.Errorf("发送 Alert 转发请求失败：%w", err)
 	}
 	defer resp.Body.Close()
 
 	responseBody, err := io.ReadAll(io.LimitReader(resp.Body, alertResponseBodyLimit+1))
 	if err != nil {
-		return fmt.Errorf("读取 Alert 响应失败：%w", err)
+		return nil, fmt.Errorf("读取 Alert 响应失败：%w", err)
 	}
 	if len(responseBody) > alertResponseBodyLimit {
-		return fmt.Errorf("Alert 响应体超过 %d 字节限制，请稍后重试", alertResponseBodyLimit)
+		return nil, fmt.Errorf("Alert 响应体超过 %d 字节限制，请稍后重试", alertResponseBodyLimit)
 	}
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("Alert 返回 HTTP %d，请稍后重试", resp.StatusCode)
+		return nil, fmt.Errorf("Alert 返回 HTTP %d，请稍后重试", resp.StatusCode)
 	}
 
-	var result struct {
-		Code *int32 `json:"code"`
-	}
+	var result alertForwardResponse
 	if err := json.Unmarshal(responseBody, &result); err != nil {
-		return fmt.Errorf("Alert 响应不是有效 JSON：%v；请稍后重试", err)
+		return nil, fmt.Errorf("Alert 响应不是有效 JSON：%v；请稍后重试", err)
 	}
 	if result.Code == nil {
-		return fmt.Errorf("Alert 响应缺少 code，无法确认转发成功；请稍后重试")
+		return nil, fmt.Errorf("Alert 响应缺少 code，无法确认转发成功；请稍后重试")
 	}
 	if *result.Code != 0 {
-		return fmt.Errorf("Alert 拒绝转发，业务 code=%d；请稍后重试", *result.Code)
+		return nil, fmt.Errorf("Alert 拒绝转发，业务 code=%d；请稍后重试", *result.Code)
 	}
-	return nil
+	if len(result.Details) == 0 || string(result.Details) == "null" || string(result.Details) == `""` {
+		return nil, fmt.Errorf("Alert 响应缺少 details，无法确认转发成功；请稍后重试")
+	}
+	return &result, nil
 }
 
 func buildRawEventIngestBody(body AlertReceiveBody, parseStatus string) RawEventIngestBody {
@@ -122,19 +172,39 @@ func buildRawEventIngestBody(body AlertReceiveBody, parseStatus string) RawEvent
 		"agent":       body.Agent,
 	}
 	return RawEventIngestBody{
-		SourceType:  sourceType,
-		ResourceUID: rawEventResourceUID(body.Labels),
-		EventTime:   time.Now().UTC(),
-		Severity:    body.SeverityCode,
-		Title:       firstNonEmpty(body.AlertName, "日志事件"),
-		Message:     firstNonEmpty(body.Summary, body.Description),
-		Labels:      body.Labels,
-		Annotations: annotations,
-		ParseStatus: parseStatus,
-		RawPayload:  body.RawData,
+		SourceType:     sourceType,
+		SourceEventKey: body.SourceEventKey,
+		State:          body.State,
+		ResourceUID:    rawEventResourceUID(body.Labels),
+		EventTime:      body.OccurredAt,
+		Severity:       body.SeverityCode,
+		Title:          firstNonEmpty(body.AlertName, "日志事件"),
+		Message:        firstNonEmpty(body.Summary, body.Description),
+		Labels:         body.Labels,
+		Annotations:    annotations,
+		ParseStatus:    parseStatus,
+		RawPayload:     body.RawData,
+		TraceID:        ensureAlertTraceID(body.TraceID, body.SourceEventKey),
 	}
 }
 
+func ensureAlertTraceID(traceID, sourceEventKey string) string {
+	traceID = strings.TrimSpace(traceID)
+	if traceID != "" && len(traceID) <= maxAlertTraceIDLength {
+		return traceID
+	}
+	sum := sha256.Sum256([]byte(strings.TrimSpace(sourceEventKey)))
+	return hex.EncodeToString(sum[:16])
+}
+
+func validFingerprint(value string) bool {
+	if len(value) != 64 {
+		return false
+	}
+	decoded, err := hex.DecodeString(value)
+	return err == nil && len(decoded) == 32
+}
+
 func rawEventSourceType(body AlertReceiveBody) string {
 	if body.Labels != nil {
 		switch strings.TrimSpace(body.Labels["source_subtype"]) {

internal/ingest/alert_outbox.go

@@ -2,160 +2,340 @@ package ingest
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
+	"log"
+	"os"
 	"strings"
 	"time"
 
 	"git.apinb.com/ops/logs/internal/config"
 	"git.apinb.com/ops/logs/internal/impl"
 	"git.apinb.com/ops/logs/internal/models"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 const (
-	outboxStatusPending  = "pending"
-	outboxStatusRetrying = "retrying"
-	outboxStatusSent     = "sent"
-	outboxStatusDead     = "dead"
+	outboxStatusPending    = "pending"
+	outboxStatusProcessing = "processing"
+	outboxStatusRetrying   = "retrying"
+	outboxStatusSent       = "sent"
+	outboxStatusDead       = "dead"
 )
 
-func enqueueAlert(logEventID uint, body AlertReceiveBody) error {
+func enqueueAlertWithDB(db *gorm.DB, logEventID uint, body AlertReceiveBody) (uint, error) {
+	body.TraceID = ensureAlertTraceID(body.TraceID, body.SourceEventKey)
 	payload, err := json.Marshal(body)
 	if err != nil {
-		return err
+		return 0, err
 	}
-	return enqueuePayload(logEventID, string(payload))
+	return enqueuePayloadWithDB(db, logEventID, string(payload))
 }
 
-func enqueueRawEvent(logEventID uint, body AlertReceiveBody, parseStatus string) error {
+func enqueueRawEventWithDB(db *gorm.DB, logEventID uint, body AlertReceiveBody, parseStatus string) (uint, error) {
 	payload, err := json.Marshal(buildRawEventIngestBody(body, parseStatus))
 	if err != nil {
-		return err
+		return 0, err
 	}
-	return enqueuePayload(logEventID, string(payload))
+	return enqueuePayloadWithDB(db, logEventID, string(payload))
 }
 
-func enqueuePayload(logEventID uint, payloadJSON string) error {
+func enqueuePayloadWithDB(db *gorm.DB, logEventID uint, payloadJSON string) (uint, error) {
+	if db == nil {
+		return 0, fmt.Errorf("database is not initialized")
+	}
+	now, err := databaseNow(db)
+	if err != nil {
+		return 0, err
+	}
 	row := models.AlertOutbox{
 		LogEventID:  logEventID,
 		PayloadJSON: payloadJSON,
 		Status:      outboxStatusPending,
 		RetryCount:  0,
-		NextRetryAt: time.Now(),
+		NextRetryAt: now,
 		LastError:   "",
 	}
-	return impl.DBService.Create(&row).Error
+	if err := db.Create(&row).Error; err != nil {
+		return 0, err
+	}
+	return row.ID, nil
 }
 
 func StartAlertDispatcher() {
+	owner := dispatcherOwner()
 	go func() {
 		ticker := time.NewTicker(2 * time.Second)
 		defer ticker.Stop()
 		for range ticker.C {
-			processAlertOutboxBatch(20)
+			if _, err := ProcessAlertOutboxBatch(impl.DBService, 20, owner); err != nil {
+				log.Printf("logs: alert outbox dispatch: %v", err)
+			}
 		}
 	}()
 }
 
-func processAlertOutboxBatch(limit int) {
+func dispatcherOwner() string {
+	host, _ := os.Hostname()
+	return fmt.Sprintf("%s:%d:%d", host, os.Getpid(), time.Now().UnixNano())
+}
+
+// ProcessAlertOutboxBatch 原子领取并处理一批任务，可安全用于多实例 worker。
+func ProcessAlertOutboxBatch(db *gorm.DB, limit int, owner string) (int, error) {
+	if db == nil {
+		return 0, fmt.Errorf("database is not initialized")
+	}
 	if limit <= 0 {
 		limit = 20
 	}
-	var rows []models.AlertOutbox
-	now := time.Now()
-	err := impl.DBService.
-		Where("status IN ? AND next_retry_at <= ?", []string{outboxStatusPending, outboxStatusRetrying}, now).
-		Order("id asc").
-		Limit(limit).
-		Find(&rows).Error
-	if err != nil || len(rows) == 0 {
-		return
+	owner = strings.TrimSpace(owner)
+	if owner == "" {
+		return 0, fmt.Errorf("outbox lease owner is required")
 	}
-	for _, row := range rows {
-		processOneOutbox(row)
+	processed := 0
+	for processed < limit {
+		rows, err := claimAlertOutboxBatch(db, 1, owner)
+		if err != nil {
+			return processed, err
+		}
+		if len(rows) == 0 {
+			break
+		}
+		if err := processOneOutbox(db, rows[0]); err != nil {
+			return processed, err
+		}
+		processed++
 	}
+	return processed, nil
 }
 
-func processOneOutbox(row models.AlertOutbox) {
+func claimAlertOutboxBatch(db *gorm.DB, limit int, owner string) ([]models.AlertOutbox, error) {
+	var rows []models.AlertOutbox
+	now, err := databaseNow(db)
+	if err != nil {
+		return nil, err
+	}
+	leaseUntil := now.Add(30 * time.Second)
+	err = db.Transaction(func(tx *gorm.DB) error {
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"}).
+			Where("(status IN ? AND next_retry_at <= ?) OR (status = ? AND lease_until IS NOT NULL AND lease_until <= ?)", []string{outboxStatusPending, outboxStatusRetrying}, now, outboxStatusProcessing, now).
+			Order("id asc").Limit(limit).Find(&rows).Error; err != nil {
+			return err
+		}
+		if len(rows) == 0 {
+			return nil
+		}
+		ids := make([]uint, 0, len(rows))
+		for i := range rows {
+			ids = append(ids, rows[i].ID)
+		}
+		if err := tx.Model(&models.AlertOutbox{}).Where("id IN ?", ids).Updates(map[string]interface{}{
+			"status":      outboxStatusProcessing,
+			"lease_until": leaseUntil,
+			"lease_owner": owner,
+		}).Error; err != nil {
+			return err
+		}
+		for i := range rows {
+			rows[i].Status = outboxStatusProcessing
+			rows[i].LeaseUntil = &leaseUntil
+			rows[i].LeaseOwner = owner
+		}
+		return nil
+	})
+	return rows, err
+}
+
+func processOneOutbox(db *gorm.DB, row models.AlertOutbox) error {
 	var body AlertReceiveBody
 	if err := json.Unmarshal([]byte(row.PayloadJSON), &body); err != nil {
-		markOutboxDead(row.ID, row.RetryCount, "invalid_payload: "+err.Error())
-		return
+		now, nowErr := databaseNow(db)
+		if nowErr != nil {
+			return nowErr
+		}
+		return markOutboxDead(db, row, row.RetryCount, "invalid_payload: "+err.Error(), now)
 	}
-	if err := forwardOutboxPayload(row.PayloadJSON, body); err != nil {
-		markOutboxRetry(row, err.Error())
-		return
+	fallbackKey := fmt.Sprintf("logs-outbox:%d", row.ID)
+	occurredAt := row.CreatedAt.UTC()
+	if occurredAt.IsZero() {
+		occurredAt = time.Now().UTC()
 	}
-	_ = impl.DBService.Model(&models.AlertOutbox{}).Where("id = ?", row.ID).Updates(map[string]interface{}{
-		"status":        outboxStatusSent,
-		"last_error":    "",
-		"next_retry_at": time.Now(),
-	}).Error
-	_ = impl.DBService.Model(&models.LogEvent{}).Where("id = ?", row.LogEventID).Updates(map[string]interface{}{
-		"alert_sent":      true,
-		"dispatch_status": "sent",
-	}).Error
+	forwardErr := forwardOutboxPayload(row.PayloadJSON, body, fallbackKey, occurredAt)
+	now, err := databaseNow(db)
+	if err != nil {
+		return err
+	}
+	if forwardErr != nil {
+		if errors.Is(forwardErr, errAlertForwardDisabled) {
+			return markOutboxWaiting(db, row, forwardErr.Error(), now)
+		}
+		return markOutboxRetry(db, row, forwardErr.Error(), now)
+	}
+	return markOutboxSent(db, row, now)
 }
 
-func forwardOutboxPayload(payloadJSON string, legacyBody AlertReceiveBody) error {
+func databaseNow(db *gorm.DB) (time.Time, error) {
+	var now time.Time
+	if err := db.Raw("SELECT CURRENT_TIMESTAMP").Scan(&now).Error; err != nil {
+		return time.Time{}, err
+	}
+	return now.UTC(), nil
+}
+
+func forwardOutboxPayload(payloadJSON string, legacyBody AlertReceiveBody, fallbackSourceEventKey string, occurredAt time.Time) error {
 	var rawEvent RawEventIngestBody
 	if err := json.Unmarshal([]byte(payloadJSON), &rawEvent); err == nil && rawEvent.SourceType != "" && len(rawEvent.RawPayload) > 0 {
+		if strings.TrimSpace(rawEvent.SourceEventKey) == "" {
+			rawEvent.SourceEventKey = fallbackSourceEventKey
+		}
+		if rawEvent.EventTime.IsZero() {
+			rawEvent.EventTime = occurredAt
+		}
+		rawEvent.TraceID = ensureAlertTraceID(rawEvent.TraceID, firstNonEmpty(rawEvent.SourceEventKey, fallbackSourceEventKey))
 		return forwardRawEvent(rawEvent)
 	}
+	if strings.TrimSpace(legacyBody.SourceEventKey) == "" {
+		legacyBody.SourceEventKey = fallbackSourceEventKey
+	}
+	if legacyBody.OccurredAt.IsZero() {
+		legacyBody.OccurredAt = occurredAt
+	}
+	legacyBody.TraceID = ensureAlertTraceID(legacyBody.TraceID, firstNonEmpty(legacyBody.SourceEventKey, fallbackSourceEventKey))
 	return forwardAlert(legacyBody)
 }
 
+func markOutboxWaiting(db *gorm.DB, row models.AlertOutbox, msg string, now time.Time) error {
+	return updateClaimedOutbox(db, row, map[string]interface{}{
+		"status": outboxStatusRetrying, "next_retry_at": now.Add(30 * time.Second),
+		"last_error": truncateError(msg, 1024), "lease_until": nil, "lease_owner": "",
+	}, "retrying")
+}
+
 func forwardRawEvent(body RawEventIngestBody) error {
 	cfg := config.Spec.AlertForward
 	if cfg == nil || !cfg.Enabled || cfg.BaseURL == "" {
-		return nil
+		return errAlertForwardDisabled
 	}
 	if len(body.RawPayload) == 0 {
 		return fmt.Errorf("raw_payload 不能为空")
 	}
+	body.TraceID = ensureAlertTraceID(body.TraceID, body.SourceEventKey)
 	raw, err := json.Marshal(body)
 	if err != nil {
 		return err
 	}
-	return postAlertPayload(cfg, "/Alert/v1/raw-events/ingest", raw)
+	result, err := postAlertPayload(cfg, "/Alert/v1/raw-events/ingest", raw, body.TraceID)
+	if err != nil {
+		return err
+	}
+	var details struct {
+		ID            uint   `json:"id"`
+		RawEventID    uint   `json:"raw_event_id"`
+		AlertRecordID uint   `json:"alert_record_id"`
+		IncidentID    uint   `json:"incident_id"`
+		ParseStatus   string `json:"parse_status"`
+		Status        string `json:"status"`
+		Fingerprint   string `json:"fingerprint"`
+	}
+	if err := json.Unmarshal(result.Details, &details); err != nil {
+		return fmt.Errorf("Alert 原始事件响应 details 无效：%v；请稍后重试", err)
+	}
+	if body.ParseStatus == "unparsed" {
+		if details.ID == 0 || (details.ParseStatus != "unparsed" && details.ParseStatus != "replayed") {
+			return fmt.Errorf("Alert 原始事件响应缺少持久化结果，无法确认转发成功；请稍后重试")
+		}
+		return nil
+	}
+	rawEventID := details.RawEventID
+	if rawEventID == 0 {
+		rawEventID = details.ID
+	}
+	if rawEventID == 0 || details.AlertRecordID == 0 || details.IncidentID == 0 ||
+		(details.ID != 0 && details.RawEventID != 0 && details.ID != details.RawEventID) ||
+		(details.Status != "firing" && details.Status != "resolved") || !validFingerprint(details.Fingerprint) {
+		return fmt.Errorf("Alert 原始事件响应缺少完整处理结果，无法确认转发成功；请稍后重试")
+	}
+	return nil
 }
 
-func markOutboxRetry(row models.AlertOutbox, msg string) {
+func markOutboxSent(db *gorm.DB, row models.AlertOutbox, now time.Time) error {
+	return db.Transaction(func(tx *gorm.DB) error {
+		result := tx.Model(&models.AlertOutbox{}).
+			Where("id = ? AND status = ? AND lease_owner = ?", row.ID, outboxStatusProcessing, row.LeaseOwner).
+			Updates(map[string]interface{}{"status": outboxStatusSent, "last_error": "", "next_retry_at": now, "lease_until": nil, "lease_owner": ""})
+		if result.Error != nil {
+			return result.Error
+		}
+		if result.RowsAffected == 0 {
+			return nil
+		}
+		eventResult := tx.Model(&models.LogEvent{}).Where("id = ? AND dispatch_outbox_id = ?", row.LogEventID, row.ID).Updates(map[string]interface{}{
+			"alert_sent": true, "dispatch_status": "sent",
+		})
+		if eventResult.Error != nil {
+			return eventResult.Error
+		}
+		if eventResult.RowsAffected != 1 {
+			return fmt.Errorf("log event %d does not belong to outbox %d", row.LogEventID, row.ID)
+		}
+		return nil
+	})
+}
+
+func markOutboxRetry(db *gorm.DB, row models.AlertOutbox, msg string, now time.Time) error {
 	retry := row.RetryCount + 1
 	const maxRetry = 5
 	if retry > maxRetry {
-		markOutboxDead(row.ID, retry, msg)
-		return
+		return markOutboxDead(db, row, retry, msg, now)
 	}
 	backoff := time.Duration(retry*retry) * time.Second
 	if backoff > 60*time.Second {
 		backoff = 60 * time.Second
 	}
-	_ = impl.DBService.Model(&models.AlertOutbox{}).Where("id = ?", row.ID).Updates(map[string]interface{}{
-		"status":        outboxStatusRetrying,
-		"retry_count":   retry,
-		"next_retry_at": time.Now().Add(backoff),
-		"last_error":    truncateError(msg, 1024),
-	}).Error
-	_ = impl.DBService.Model(&models.LogEvent{}).Where("id = ?", row.LogEventID).Update("dispatch_status", "retrying").Error
+	return updateClaimedOutbox(db, row, map[string]interface{}{
+		"status": outboxStatusRetrying, "retry_count": retry, "next_retry_at": now.Add(backoff),
+		"last_error": truncateError(msg, 1024), "lease_until": nil, "lease_owner": "",
+	}, "retrying")
 }
 
-func markOutboxDead(id uint, retry int, msg string) {
-	_ = impl.DBService.Model(&models.AlertOutbox{}).Where("id = ?", id).Updates(map[string]interface{}{
-		"status":        outboxStatusDead,
-		"retry_count":   retry,
-		"next_retry_at": time.Now(),
-		"last_error":    truncateError(msg, 1024),
-	}).Error
-	var row models.AlertOutbox
-	if err := impl.DBService.Select("log_event_id").First(&row, id).Error; err == nil && row.LogEventID > 0 {
-		_ = impl.DBService.Model(&models.LogEvent{}).Where("id = ?", row.LogEventID).Update("dispatch_status", "dead").Error
-	}
+func markOutboxDead(db *gorm.DB, row models.AlertOutbox, retry int, msg string, now time.Time) error {
+	return updateClaimedOutbox(db, row, map[string]interface{}{
+		"status": outboxStatusDead, "retry_count": retry, "next_retry_at": now,
+		"last_error": truncateError(msg, 1024), "lease_until": nil, "lease_owner": "",
+	}, "dead")
+}
+
+func updateClaimedOutbox(db *gorm.DB, row models.AlertOutbox, updates map[string]interface{}, eventStatus string) error {
+	return db.Transaction(func(tx *gorm.DB) error {
+		result := tx.Model(&models.AlertOutbox{}).
+			Where("id = ? AND status = ? AND lease_owner = ?", row.ID, outboxStatusProcessing, row.LeaseOwner).
+			Updates(updates)
+		if result.Error != nil {
+			return result.Error
+		}
+		if result.RowsAffected == 0 {
+			return nil
+		}
+		eventResult := tx.Model(&models.LogEvent{}).Where("id = ? AND dispatch_outbox_id = ?", row.LogEventID, row.ID).Update("dispatch_status", eventStatus)
+		if eventResult.Error != nil {
+			return eventResult.Error
+		}
+		if eventResult.RowsAffected != 1 {
+			return fmt.Errorf("log event %d does not belong to outbox %d", row.LogEventID, row.ID)
+		}
+		return nil
+	})
 }
 
 func truncateError(s string, n int) string {
 	s = strings.TrimSpace(s)
-	if len(s) <= n {
+	if n <= 0 {
+		return ""
+	}
+	runes := []rune(s)
+	if len(runes) <= n {
 		return s
 	}
-	return s[:n]
+	return string(runes[:n])
 }

internal/ingest/engine.go

@@ -1,8 +1,11 @@
 package ingest
 
 import (
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"log"
 	"net"
 	"regexp"
 	"sort"
@@ -15,6 +18,7 @@ import (
 	"git.apinb.com/ops/logs/internal/impl"
 	"git.apinb.com/ops/logs/internal/models"
 	"github.com/gosnmp/gosnmp"
+	"gorm.io/gorm"
 )
 
 type Engine struct {
@@ -29,9 +33,12 @@ type Engine struct {
 }
 
 type resourceRef struct {
-	ResourceType string
-	ResourceID   string
-	ResourceName string
+	ResourceType     string
+	ResourceUID      string
+	ResourceCategory string
+	ServiceIdentity  string
+	ResourceID       string
+	ResourceName     string
 }
 
 func resourceTypePriority(resourceType string) int {
@@ -88,6 +95,12 @@ func (e *Engine) Refresh() error {
 			ResourceID:   m.ResourceID,
 			ResourceName: m.ResourceName,
 		}
+		var mappingLabels map[string]string
+		if err := json.Unmarshal([]byte(m.LabelsJSON), &mappingLabels); err == nil {
+			ref.ResourceUID = strings.TrimSpace(mappingLabels["resource_uid"])
+			ref.ResourceCategory = strings.TrimSpace(mappingLabels["resource_category"])
+			ref.ServiceIdentity = strings.TrimSpace(mappingLabels["service_identity"])
+		}
 		var ips []string
 		if err := json.Unmarshal([]byte(m.IPsJSON), &ips); err == nil {
 			for _, ip := range ips {
@@ -146,6 +159,7 @@ func normOID(s string) string {
 }
 
 func (e *Engine) HandleSyslog(addr *net.UDPAddr, payload []byte) {
+	occurredAt := time.Now().UTC()
 	parsed := parseSyslogPayload(payload)
 	device := parsed.Hostname
 	if device == "" {
@@ -193,35 +207,38 @@ func (e *Engine) HandleSyslog(addr *net.UDPAddr, payload []byte) {
 		}
 	}
 
-	if err := impl.DBService.Create(&ev).Error; err != nil {
-		return
-	}
-
 	if matched == nil {
 		rawBytes, mErr := json.Marshal(string(payload))
 		if mErr != nil {
 			return
 		}
+		labels := map[string]string{
+			"source_type":    "log",
+			"source_subtype": "syslog",
+			"device":         device,
+			"remote_addr":    addr.String(),
+			"ip":             addr.IP.String(),
+			"instance":       firstNonEmpty(device, addr.String()),
+			"job":            "logs-syslog",
+		}
+		applyResourceIdentity(&ev, labels, "", ref, "syslog", addr.IP.String())
 		body := AlertReceiveBody{
 			AlertName:    "未解析 Syslog",
 			Summary:      summary,
 			Description:  parsed.RawLine,
 			SeverityCode: sev,
 			Value:        parsed.Message,
-			Labels: map[string]string{
-				"source_type":    "log",
-				"source_subtype": "syslog",
-				"device":         device,
-				"remote_addr":    addr.String(),
-				"ip":             addr.IP.String(),
-				"instance":       firstNonEmpty(device, addr.String()),
-				"job":            "logs-syslog",
-			},
-			Agent:   "logs-syslog",
-			RawData: rawBytes,
+			Labels:       labels,
+			Agent:        "logs-syslog",
+			RawData:      rawBytes,
 		}
-		if err := enqueueRawEvent(ev.ID, body, "unparsed"); err == nil {
-			_ = impl.DBService.Model(&ev).Update("dispatch_status", "pending").Error
+		if err := persistLogEventAndOutbox(&ev, func(tx *gorm.DB, stored *models.LogEvent) (uint, error) {
+			body.SourceEventKey = logSourceEventKey(stored, "raw")
+			body.OccurredAt = occurredAt
+			body.State = "firing"
+			return enqueueRawEventWithDB(tx, stored.ID, body, "unparsed")
+		}); err != nil {
+			log.Printf("logs: persist syslog raw event: %v", err)
 		}
 		return
 	}
@@ -241,38 +258,33 @@ func (e *Engine) HandleSyslog(addr *net.UDPAddr, payload []byte) {
 	if matchDetails.ResourceUID != "" {
 		labels["resource_uid"] = matchDetails.ResourceUID
 	}
-	rawObj := map[string]interface{}{
-		"source":       "syslog",
-		"received_at":  time.Now().UTC().Format(time.RFC3339),
-		"source_ip":    addr.IP.String(),
-		"rule_id":      matched.ID,
-		"log_entry_id": ev.ID,
-		"raw_packet":   string(payload),
-		"parsed":       detailObj,
-		"match":        matchDetails.Captures,
-	}
-	rawBytes, mErr := json.Marshal(rawObj)
-	if mErr != nil {
-		return
-	}
-	body := AlertReceiveBody{
-		AlertName:    matched.AlertName,
-		Summary:      summary,
-		Description:  summary,
-		SeverityCode: firstNonEmpty(matchDetails.SeverityCode, firstNonEmpty(matched.SeverityCode, sev)),
-		Value:        parsed.Message,
-		Labels:       labels,
-		Agent:        "logs-syslog",
-		PolicyID:     matched.PolicyID,
-		RawData:      rawBytes,
-	}
-	if err := enqueueAlert(ev.ID, body); err == nil {
-		_ = impl.DBService.Model(&ev).Update("dispatch_status", "pending").Error
+	applyResourceIdentity(&ev, labels, matchDetails.ResourceUID, ref, "syslog", addr.IP.String())
+	if err := persistLogEventAndOutbox(&ev, func(tx *gorm.DB, stored *models.LogEvent) (uint, error) {
+		rawObj := map[string]interface{}{
+			"source": "syslog", "received_at": occurredAt.Format(time.RFC3339), "source_ip": addr.IP.String(),
+			"rule_id": matched.ID, "log_entry_id": stored.ID, "raw_packet": string(payload),
+			"parsed": detailObj, "match": matchDetails.Captures,
+		}
+		rawBytes, err := json.Marshal(rawObj)
+		if err != nil {
+			return 0, err
+		}
+		body := AlertReceiveBody{
+			AlertName: matched.AlertName, Summary: summary, Description: summary,
+			SeverityCode: firstNonEmpty(matchDetails.SeverityCode, firstNonEmpty(matched.SeverityCode, sev)),
+			Value:        parsed.Message, Labels: labels, Agent: "logs-syslog", PolicyID: matched.PolicyID,
+			State: matchDetails.State, SourceEventKey: logSourceEventKey(stored, "alert"), OccurredAt: occurredAt, RawData: rawBytes,
+		}
+		body.Fingerprint = alertLifecycleFingerprint(body, matched.LifecycleKey, "")
+		return enqueueAlertWithDB(tx, stored.ID, body)
+	}); err != nil {
+		log.Printf("logs: persist matched syslog event: %v", err)
 	}
 }
 
 type syslogRuleMatch struct {
 	Matched      bool
+	State        string
 	ResourceUID  string
 	SeverityCode string
 	Captures     map[string]string
@@ -283,11 +295,12 @@ func syslogRuleMatches(rule *models.SyslogRule, device, message, rawLine string)
 }
 
 func syslogRuleMatchDetails(rule *models.SyslogRule, device, message, rawLine string) syslogRuleMatch {
-	result := syslogRuleMatch{Captures: map[string]string{}}
+	result := syslogRuleMatch{State: "firing", Captures: map[string]string{}}
 	deviceContains := strings.TrimSpace(rule.DeviceNameContains)
 	sourceMatch := strings.TrimSpace(rule.SourceMatch)
 	keywordRegex := strings.TrimSpace(rule.KeywordRegex)
 	messageRegex := strings.TrimSpace(rule.MessageRegex)
+	recoveryRegex := strings.TrimSpace(rule.RecoveryMatchRegex)
 	if deviceContains == "" && sourceMatch == "" && keywordRegex == "" && messageRegex == "" {
 		return result
 	}
@@ -304,24 +317,35 @@ func syslogRuleMatchDetails(rule *models.SyslogRule, device, message, rawLine st
 			return result
 		}
 	}
-	for _, pattern := range []string{keywordRegex, messageRegex} {
-		if pattern == "" {
-			continue
-		}
-		re, err := regexp.Compile(pattern)
+	if recoveryRegex != "" {
+		re, err := regexp.Compile(recoveryRegex)
 		if err != nil {
 			return result
 		}
-		matches := re.FindStringSubmatch(message)
-		if matches == nil {
-			matches = re.FindStringSubmatch(rawLine)
+		matches := firstRegexMatch(re, message, rawLine)
+		if matches != nil {
+			mergeNamedCaptures(result.Captures, re, matches)
+			result.State = "resolved"
+			result.Matched = true
 		}
-		if matches == nil {
-			return result
-		}
-		mergeNamedCaptures(result.Captures, re, matches)
 	}
-	result.Matched = true
+	if !result.Matched {
+		for _, pattern := range []string{keywordRegex, messageRegex} {
+			if pattern == "" {
+				continue
+			}
+			re, err := regexp.Compile(pattern)
+			if err != nil {
+				return result
+			}
+			matches := firstRegexMatch(re, message, rawLine)
+			if matches == nil {
+				return result
+			}
+			mergeNamedCaptures(result.Captures, re, matches)
+		}
+		result.Matched = true
+	}
 	if uid := extractWithNamedRegex(rule.ResourceUIDExtractRegex, "resource_uid", message, rawLine); uid != "" {
 		result.ResourceUID = normalizeExtractedResourceUID(uid)
 	} else if uid := result.Captures["resource_uid"]; uid != "" {
@@ -331,6 +355,15 @@ func syslogRuleMatchDetails(rule *models.SyslogRule, device, message, rawLine st
 	return result
 }
 
+func firstRegexMatch(re *regexp.Regexp, values ...string) []string {
+	for _, value := range values {
+		if matches := re.FindStringSubmatch(value); matches != nil {
+			return matches
+		}
+	}
+	return nil
+}
+
 func mergeNamedCaptures(dst map[string]string, re *regexp.Regexp, matches []string) {
 	names := re.SubexpNames()
 	for i, name := range names {
@@ -358,7 +391,9 @@ func extractWithNamedRegex(pattern, groupName, message, rawLine string) string {
 		names := re.SubexpNames()
 		for i, name := range names {
 			if i > 0 && name == groupName && i < len(matches) {
-				return strings.TrimSpace(matches[i])
+				if value := strings.TrimSpace(matches[i]); value != "" {
+					return value
+				}
 			}
 		}
 		for i := 1; i < len(matches); i++ {
@@ -372,9 +407,12 @@ func extractWithNamedRegex(pattern, groupName, message, rawLine string) string {
 
 func normalizeExtractedResourceUID(uid string) string {
 	uid = strings.TrimSpace(uid)
-	if uid == "" || strings.Contains(uid, ":") {
+	if uid == "" {
 		return uid
 	}
+	if category, identity, ok := splitKnownResourceUID(uid); ok {
+		return category + ":" + identity
+	}
 	return "network:" + uid
 }
 
@@ -445,6 +483,7 @@ func lookupTrapDict(e *Engine, trapOID string) *models.TrapDictionaryEntry {
 }
 
 func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
+	occurredAt := time.Now().UTC()
 	trapOID := extractTrapOID(pkt)
 	if trapShielded(e, addr, trapOID, pkt) {
 		return
@@ -490,43 +529,42 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
 		SeverityCode:      sev,
 		TrapOID:           trapOID,
 	}
-	if err := impl.DBService.Create(&ev).Error; err != nil {
-		return
-	}
-
 	e.mu.RLock()
 	rules := e.trapRules
 	e.mu.RUnlock()
 
-	matched := firstMatchingTrapRule(rules, trapOID, fp)
-	if matched == nil {
+	match := firstMatchingTrapRule(rules, trapOID, fp)
+	if match.Rule == nil {
 		rawBytes, mErr := json.Marshal(fp)
 		if mErr != nil {
 			return
 		}
+		labels := map[string]string{
+			"source_type": "log", "source_subtype": "snmp_trap", "trap_oid": trapOID,
+			"remote_addr": addr.String(), "ip": addr.IP.String(), "instance": addr.IP.String(), "job": "logs-trap",
+		}
+		applyResourceIdentity(&ev, labels, "", ref, "snmp_trap", addr.IP.String())
 		body := AlertReceiveBody{
 			AlertName:    "未解析 SNMP Trap",
 			Summary:      readable,
 			Description:  fp,
 			SeverityCode: sev,
 			Value:        string(vbJSON),
-			Labels: map[string]string{
-				"source_type":    "log",
-				"source_subtype": "snmp_trap",
-				"trap_oid":       trapOID,
-				"remote_addr":    addr.String(),
-				"ip":             addr.IP.String(),
-				"instance":       addr.IP.String(),
-				"job":            "logs-trap",
-			},
-			Agent:   "logs-trap",
-			RawData: rawBytes,
+			Labels:       labels,
+			Agent:        "logs-trap",
+			RawData:      rawBytes,
 		}
-		if err := enqueueRawEvent(ev.ID, body, "unparsed"); err == nil {
-			_ = impl.DBService.Model(&ev).Update("dispatch_status", "pending").Error
+		if err := persistLogEventAndOutbox(&ev, func(tx *gorm.DB, stored *models.LogEvent) (uint, error) {
+			body.State = "firing"
+			body.SourceEventKey = logSourceEventKey(stored, "raw")
+			body.OccurredAt = occurredAt
+			return enqueueRawEventWithDB(tx, stored.ID, body, "unparsed")
+		}); err != nil {
+			log.Printf("logs: persist trap raw event: %v", err)
 		}
 		return
 	}
+	matched := match.Rule
 
 	desc := readable
 	if dict != nil && dict.RecoveryMessage != "" {
@@ -541,6 +579,10 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
 		"instance":       addr.IP.String(),
 		"job":            "logs-trap",
 	}
+	trapInstance := trapInstanceKey(pkt)
+	if trapInstance != "" {
+		labels["trap_instance"] = trapInstance
+	}
 	if matched.ID != 0 {
 		labels["resource_type"] = "trap_rule"
 		labels["resource_id"] = strconv.FormatUint(uint64(matched.ID), 10)
@@ -551,6 +593,7 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
 			labels["resource_id"] = trapOID
 		}
 	}
+	applyResourceIdentity(&ev, labels, "", ref, "snmp_trap", addr.IP.String())
 	resolved := map[string]interface{}{}
 	if dict != nil {
 		resolved["vendor"] = dict.Vendor
@@ -561,36 +604,28 @@ func (e *Engine) HandleTrap(addr *net.UDPAddr, pkt *gosnmp.SnmpPacket) {
 		resolved["severity_mapping"] = dict.SeverityMappingJSON
 		resolved["parse_expression"] = dict.ParseExpression
 	}
-	rawObj := map[string]interface{}{
-		"source":       "snmp_trap",
-		"received_at":  time.Now().UTC().Format(time.RFC3339),
-		"source_ip":    addr.IP.String(),
-		"log_entry_id": ev.ID,
-		"trap_oid":     trapOID,
-		"varbinds":     trapVarbinds(pkt),
-		"resolved":     resolved,
-		"pdu_summary":  fp,
-	}
-	if matched.ID != 0 {
-		rawObj["rule_id"] = matched.ID
-	}
-	rawBytes, mErr := json.Marshal(rawObj)
-	if mErr != nil {
-		return
-	}
-	body := AlertReceiveBody{
-		AlertName:    firstNonEmpty(matched.AlertName, "SNMP Trap"),
-		Summary:      readable,
-		Description:  desc,
-		SeverityCode: firstNonEmpty(matched.SeverityCode, sev),
-		Value:        string(vbJSON),
-		Labels:       labels,
-		Agent:        "logs-trap",
-		PolicyID:     matched.PolicyID,
-		RawData:      rawBytes,
-	}
-	if err := enqueueAlert(ev.ID, body); err == nil {
-		_ = impl.DBService.Model(&ev).Update("dispatch_status", "pending").Error
+	if err := persistLogEventAndOutbox(&ev, func(tx *gorm.DB, stored *models.LogEvent) (uint, error) {
+		rawObj := map[string]interface{}{
+			"source": "snmp_trap", "received_at": occurredAt.Format(time.RFC3339), "source_ip": addr.IP.String(),
+			"log_entry_id": stored.ID, "trap_oid": trapOID, "varbinds": trapVarbinds(pkt), "resolved": resolved, "pdu_summary": fp,
+		}
+		if matched.ID != 0 {
+			rawObj["rule_id"] = matched.ID
+		}
+		rawBytes, err := json.Marshal(rawObj)
+		if err != nil {
+			return 0, err
+		}
+		body := AlertReceiveBody{
+			AlertName: firstNonEmpty(matched.AlertName, "SNMP Trap"), Summary: readable, Description: desc,
+			SeverityCode: firstNonEmpty(matched.SeverityCode, sev), Value: string(vbJSON), Labels: labels,
+			Agent: "logs-trap", PolicyID: matched.PolicyID, State: match.State,
+			SourceEventKey: logSourceEventKey(stored, "alert"), OccurredAt: occurredAt, RawData: rawBytes,
+		}
+		body.Fingerprint = alertLifecycleFingerprint(body, matched.LifecycleKey, trapInstance)
+		return enqueueAlertWithDB(tx, stored.ID, body)
+	}); err != nil {
+		log.Printf("logs: persist matched trap event: %v", err)
 	}
 }
 
@@ -632,6 +667,27 @@ func trapVarbinds(pkt *gosnmp.SnmpPacket) []map[string]string {
 	return out
 }
 
+func trapInstanceKey(pkt *gosnmp.SnmpPacket) string {
+	if pkt == nil {
+		return ""
+	}
+	for _, prefix := range []string{
+		"1.3.6.1.2.1.31.1.1.1.1.",
+		"1.3.6.1.2.1.2.2.1.2.",
+		"1.3.6.1.2.1.2.2.1.1.",
+	} {
+		for _, variable := range pkt.Variables {
+			name := normOID(variable.Name)
+			if strings.HasPrefix(name, prefix) {
+				if index := strings.TrimSpace(strings.TrimPrefix(name, prefix)); index != "" {
+					return index
+				}
+			}
+		}
+	}
+	return ""
+}
+
 func buildTrapReadable(trapOID string, dict *models.TrapDictionaryEntry, varbindSummary string) string {
 	if dict != nil && firstNonEmpty(dict.Name, dict.Title) != "" {
 		return firstNonEmpty(dict.Name, dict.Title) + " (" + trapOID + ")"
@@ -642,34 +698,49 @@ func buildTrapReadable(trapOID string, dict *models.TrapDictionaryEntry, varbind
 	return truncate(varbindSummary, 256)
 }
 
-func trapRuleMatches(rule *models.TrapRule, trapOID, varbindFP string) bool {
+type trapRuleMatch struct {
+	Rule  *models.TrapRule
+	State string
+}
+
+func trapRuleState(rule *models.TrapRule, trapOID, varbindFP string) (string, bool) {
 	hasOID := strings.TrimSpace(rule.OIDPrefix) != ""
 	hasRE := strings.TrimSpace(rule.VarbindMatchRegex) != ""
+	hasRecoveryRE := strings.TrimSpace(rule.RecoveryMatchRegex) != ""
 	if !hasOID && !hasRE {
-		return false
+		return "", false
 	}
 	if hasOID && !strings.HasPrefix(normOID(trapOID), normOID(rule.OIDPrefix)) {
-		return false
+		return "", false
+	}
+	if hasRecoveryRE {
+		re, err := regexp.Compile(rule.RecoveryMatchRegex)
+		if err != nil {
+			return "", false
+		}
+		if re.MatchString(trapOID) || re.MatchString(varbindFP) {
+			return "resolved", true
+		}
 	}
 	if hasRE {
 		re, err := regexp.Compile(rule.VarbindMatchRegex)
 		if err != nil {
-			return false
+			return "", false
 		}
 		if !re.MatchString(varbindFP) {
-			return false
+			return "", false
 		}
 	}
-	return true
+	return "firing", true
 }
 
-func firstMatchingTrapRule(rules []models.TrapRule, trapOID, varbindFP string) *models.TrapRule {
+func firstMatchingTrapRule(rules []models.TrapRule, trapOID, varbindFP string) trapRuleMatch {
 	for i := range rules {
-		if trapRuleMatches(&rules[i], trapOID, varbindFP) {
-			return &rules[i]
+		if state, matched := trapRuleState(&rules[i], trapOID, varbindFP); matched {
+			return trapRuleMatch{Rule: &rules[i], State: state}
 		}
 	}
-	return nil
+	return trapRuleMatch{}
 }
 
 func firstNonEmpty(a, b string) string {
@@ -679,6 +750,29 @@ func firstNonEmpty(a, b string) string {
 	return b
 }
 
+func alertLifecycleFingerprint(body AlertReceiveBody, lifecycleKey, instance string) string {
+	lifecycleKey = strings.TrimSpace(lifecycleKey)
+	if lifecycleKey == "" {
+		return ""
+	}
+	resourceUID := ""
+	sourceIP := ""
+	if body.Labels != nil {
+		resourceUID = strings.TrimSpace(body.Labels["resource_uid"])
+		sourceIP = strings.TrimSpace(body.Labels["ip"])
+	}
+	identity := strings.Join([]string{
+		strings.TrimSpace(body.Agent),
+		resourceUID,
+		sourceIP,
+		strconv.FormatUint(uint64(body.PolicyID), 10),
+		lifecycleKey,
+		strings.TrimSpace(instance),
+	}, "\x00")
+	sum := sha256.Sum256([]byte(identity))
+	return hex.EncodeToString(sum[:])
+}
+
 func (e *Engine) resolveResource(sourceIP, hostname string) (resourceRef, string) {
 	e.mu.RLock()
 	ipMap := e.resourceByIP
@@ -693,3 +787,116 @@ func (e *Engine) resolveResource(sourceIP, hostname string) (resourceRef, string
 	}
 	return resourceRef{}, "none"
 }
+
+func persistLogEventAndOutbox(ev *models.LogEvent, enqueue func(*gorm.DB, *models.LogEvent) (uint, error)) error {
+	if impl.DBService == nil {
+		return fmt.Errorf("database is not initialized")
+	}
+	if ev == nil || enqueue == nil {
+		return fmt.Errorf("log event and outbox builder are required")
+	}
+	return impl.DBService.Transaction(func(tx *gorm.DB) error {
+		if err := tx.Create(ev).Error; err != nil {
+			return err
+		}
+		outboxID, err := enqueue(tx, ev)
+		if err != nil {
+			return err
+		}
+		ev.DispatchStatus = "pending"
+		ev.DispatchOutboxID = outboxID
+		return tx.Model(ev).Updates(map[string]interface{}{
+			"dispatch_status":    ev.DispatchStatus,
+			"dispatch_outbox_id": ev.DispatchOutboxID,
+		}).Error
+	})
+}
+
+func logSourceEventKey(ev *models.LogEvent, purpose string) string {
+	return fmt.Sprintf("logs:%s:%d:%s", strings.TrimSpace(ev.SourceKind), ev.ID, strings.TrimSpace(purpose))
+}
+
+func applyResourceIdentity(ev *models.LogEvent, labels map[string]string, explicitUID string, ref resourceRef, subtype, sourceIP string) {
+	uid := strings.TrimSpace(explicitUID)
+	category := ""
+	identity := ""
+	if uid == "" {
+		uid = strings.TrimSpace(ref.ResourceUID)
+	}
+	if uid != "" {
+		if knownCategory, knownIdentity, ok := splitKnownResourceUID(uid); ok {
+			category, identity = knownCategory, knownIdentity
+			uid = category + ":" + identity
+		} else {
+			category = strings.ToLower(strings.TrimSpace(ref.ResourceCategory))
+			if category == "" {
+				category = canonicalLogResourceCategory(ref.ResourceType, subtype)
+			}
+			identity = uid
+			uid = category + ":" + identity
+		}
+	}
+	if uid == "" && strings.TrimSpace(ref.ResourceCategory) != "" && strings.TrimSpace(ref.ServiceIdentity) != "" {
+		category = strings.ToLower(strings.TrimSpace(ref.ResourceCategory))
+		identity = strings.TrimSpace(ref.ServiceIdentity)
+		uid = category + ":" + identity
+	}
+	if uid == "" && strings.TrimSpace(ref.ResourceID) != "" {
+		identity = strings.TrimSpace(ref.ResourceID)
+		category = canonicalLogResourceCategory(ref.ResourceType, subtype)
+		if knownCategory, knownIdentity, ok := splitKnownResourceUID(identity); ok {
+			category, identity = knownCategory, knownIdentity
+			uid = category + ":" + identity
+		} else {
+			uid = category + ":" + identity
+		}
+	}
+	if uid == "" {
+		category = canonicalLogResourceCategory("", subtype)
+		identity = strings.TrimSpace(sourceIP)
+		uid = category + ":" + identity
+	}
+	labels["resource_uid"] = uid
+	labels["resource_category"] = category
+	labels["service_identity"] = identity
+	ev.ResourceUID = uid
+	if ev.ResourceType == "" || ev.ResourceID == "" {
+		ev.ResourceType = category
+		ev.ResourceID = identity
+	}
+}
+
+func canonicalLogResourceCategory(resourceType, subtype string) string {
+	switch strings.ToLower(strings.TrimSpace(resourceType)) {
+	case "server", "host":
+		return "host"
+	case "device", "network", "network_device":
+		return "network"
+	case "collector":
+		return "collector"
+	case "database", "middleware", "security", "storage", "room_device":
+		return strings.ToLower(strings.TrimSpace(resourceType))
+	}
+	if strings.TrimSpace(subtype) == "snmp_trap" || strings.TrimSpace(subtype) == "trap" {
+		return "network"
+	}
+	return "log_source"
+}
+
+func splitKnownResourceUID(value string) (string, string, bool) {
+	prefix, identity, found := strings.Cut(strings.TrimSpace(value), ":")
+	if !found {
+		return "", "", false
+	}
+	category := strings.ToLower(strings.TrimSpace(prefix))
+	identity = strings.TrimSpace(identity)
+	if identity == "" {
+		return "", "", false
+	}
+	switch category {
+	case "host", "network", "collector", "database", "middleware", "security", "storage", "room_device", "log_source":
+		return category, identity, true
+	default:
+		return "", "", false
+	}
+}

internal/ingest/replay.go

@@ -9,6 +9,7 @@ import (
 
 	"git.apinb.com/ops/logs/internal/impl"
 	"git.apinb.com/ops/logs/internal/models"
+	"gorm.io/gorm"
 )
 
 func BuildReplayRawEventPayload(ev models.LogEvent) (RawEventIngestBody, error) {
@@ -16,9 +17,10 @@ func BuildReplayRawEventPayload(ev models.LogEvent) (RawEventIngestBody, error)
 	if sourceType == "" {
 		return RawEventIngestBody{}, fmt.Errorf("unsupported source kind %q", ev.SourceKind)
 	}
+	now := time.Now().UTC()
 	rawObj := map[string]interface{}{
 		"source":       sourceType,
-		"replayed_at":  time.Now().UTC().Format(time.RFC3339),
+		"replayed_at":  now.Format(time.RFC3339Nano),
 		"log_entry_id": ev.ID,
 		"source_ip":    ev.SourceIP,
 		"remote_addr":  ev.RemoteAddr,
@@ -46,23 +48,29 @@ func BuildReplayRawEventPayload(ev models.LogEvent) (RawEventIngestBody, error)
 		"device":                 ev.DeviceName,
 		"job":                    "logs-replay",
 	}
-	if uid := replayResourceUID(ev); uid != "" {
-		labels["resource_uid"] = uid
-	}
+	resourceUID := replayResourceUID(ev)
+	category, identity := splitResourceUID(resourceUID)
+	labels["resource_uid"] = resourceUID
+	labels["resource_category"] = category
+	labels["service_identity"] = identity
+	sourceEventKey := fmt.Sprintf("logs:replay:%d:%d", ev.ID, now.UnixNano())
 	return RawEventIngestBody{
-		SourceType:  sourceType,
-		ResourceUID: replayResourceUID(ev),
-		EventTime:   time.Now().UTC(),
-		Severity:    firstNonEmpty(ev.SeverityCode, "warning"),
-		Title:       replayTitle(ev),
-		Message:     firstNonEmpty(ev.NormalizedSummary, ev.RawPayload),
-		Labels:      labels,
+		SourceType:     sourceType,
+		SourceEventKey: sourceEventKey,
+		State:          "firing",
+		ResourceUID:    resourceUID,
+		EventTime:      now,
+		Severity:       firstNonEmpty(ev.SeverityCode, "warning"),
+		Title:          replayTitle(ev),
+		Message:        firstNonEmpty(ev.NormalizedSummary, ev.RawPayload),
+		Labels:         labels,
 		Annotations: map[string]string{
 			"replay":          "true",
 			"dispatch_status": ev.DispatchStatus,
 		},
-		ParseStatus: "replayed",
+		ParseStatus: "parsed",
 		RawPayload:  rawBytes,
+		TraceID:     ensureAlertTraceID("", sourceEventKey),
 	}, nil
 }
 
@@ -80,16 +88,42 @@ func EnqueueReplayLogEvent(ev models.LogEvent) (uint, error) {
 		PayloadJSON: string(payload),
 		Status:      outboxStatusPending,
 		RetryCount:  0,
-		NextRetryAt: time.Now(),
 	}
-	if err := enqueueOutboxRow(&row); err != nil {
+	if impl.DBService == nil {
+		return 0, fmt.Errorf("database is not initialized")
+	}
+	if err := impl.DBService.Transaction(func(tx *gorm.DB) error {
+		now, err := databaseNow(tx)
+		if err != nil {
+			return err
+		}
+		row.NextRetryAt = now
+		if err := enqueueOutboxRowWithDB(tx, &row); err != nil {
+			return err
+		}
+		result := tx.Model(&models.LogEvent{}).Where("id = ?", ev.ID).Updates(map[string]interface{}{
+			"dispatch_status":    "pending",
+			"dispatch_outbox_id": row.ID,
+			"alert_sent":         false,
+		})
+		if result.Error != nil {
+			return result.Error
+		}
+		if result.RowsAffected == 0 {
+			return fmt.Errorf("log event %d does not exist", ev.ID)
+		}
+		return nil
+	}); err != nil {
 		return 0, err
 	}
 	return row.ID, nil
 }
 
-func enqueueOutboxRow(row *models.AlertOutbox) error {
-	return impl.DBService.Create(row).Error
+func enqueueOutboxRowWithDB(db *gorm.DB, row *models.AlertOutbox) error {
+	if db == nil {
+		return fmt.Errorf("database is not initialized")
+	}
+	return db.Create(row).Error
 }
 
 func replaySourceType(kind string) string {
@@ -111,13 +145,32 @@ func replaySubtype(kind string) string {
 }
 
 func replayResourceUID(ev models.LogEvent) string {
-	if strings.Contains(ev.ResourceID, ":") {
-		return ev.ResourceID
+	if uid := strings.TrimSpace(ev.ResourceUID); uid != "" {
+		if category, identity, ok := splitKnownResourceUID(uid); ok {
+			return category + ":" + identity
+		}
+		return canonicalLogResourceCategory(ev.ResourceType, replaySubtype(ev.SourceKind)) + ":" + uid
+	}
+	if category, identity, ok := splitKnownResourceUID(ev.ResourceID); ok {
+		return category + ":" + identity
 	}
 	if ev.ResourceType != "" && ev.ResourceID != "" {
-		return ev.ResourceType + ":" + ev.ResourceID
+		return canonicalLogResourceCategory(ev.ResourceType, replaySubtype(ev.SourceKind)) + ":" + ev.ResourceID
 	}
-	return ""
+	category := canonicalLogResourceCategory("", replaySubtype(ev.SourceKind))
+	identity := firstNonEmpty(strings.TrimSpace(ev.SourceIP), strings.TrimSpace(ev.DeviceName))
+	if identity == "" {
+		identity = fmt.Sprintf("log-event-%d", ev.ID)
+	}
+	return category + ":" + identity
+}
+
+func splitResourceUID(uid string) (string, string) {
+	parts := strings.SplitN(strings.TrimSpace(uid), ":", 2)
+	if len(parts) != 2 {
+		return "log_source", strings.TrimSpace(uid)
+	}
+	return parts[0], parts[1]
 }
 
 func replayTitle(ev models.LogEvent) string {

internal/logic/controllers/crud.go

@@ -344,10 +344,6 @@ func ReplayLogEvent(ctx *gin.Context) {
 		infra.Response.Error(ctx, err)
 		return
 	}
-	if err := impl.DBService.Model(&models.LogEvent{}).Where("id = ?", id).Update("dispatch_status", "pending").Error; err != nil {
-		infra.Response.Error(ctx, err)
-		return
-	}
 	infra.Response.Success(ctx, gin.H{
 		"log_event_id": id,
 		"outbox_id":    outboxID,

internal/logic/controllers/rule_validation.go

@@ -16,6 +16,7 @@ func validateSyslogRule(rule *models.SyslogRule) error {
 	}{
 		{name: "keyword_regex", pattern: rule.KeywordRegex},
 		{name: "message_regex", pattern: rule.MessageRegex},
+		{name: "recovery_match_regex", pattern: rule.RecoveryMatchRegex},
 		{name: "resource_uid_extract_regex", pattern: rule.ResourceUIDExtractRegex},
 	}
 	for _, field := range regexFields {
@@ -48,18 +49,25 @@ func validateSyslogRule(rule *models.SyslogRule) error {
 		strings.TrimSpace(rule.MessageRegex) == "" {
 		return fmt.Errorf("Syslog 规则的匹配条件全部为空，运行时永远不会命中；请至少填写 device_name_contains、source_match、keyword_regex、message_regex 中的一项")
 	}
+	if strings.TrimSpace(rule.RecoveryMatchRegex) != "" && strings.TrimSpace(rule.LifecycleKey) == "" {
+		return fmt.Errorf("配置 recovery_match_regex 时 lifecycle_key 不能为空")
+	}
 	return nil
 }
 
 func validateTrapRule(rule *models.TrapRule) error {
-	if strings.TrimSpace(rule.VarbindMatchRegex) != "" {
-		if _, err := regexp.Compile(rule.VarbindMatchRegex); err != nil {
-			return fmt.Errorf("varbind_match_regex 不是有效正则表达式：%v；请修正 varbind_match_regex 后重试", err)
-		}
+	if err := validateOptionalRegex("varbind_match_regex", rule.VarbindMatchRegex); err != nil {
+		return err
+	}
+	if err := validateOptionalRegex("recovery_match_regex", rule.RecoveryMatchRegex); err != nil {
+		return err
 	}
 	if strings.TrimSpace(rule.OIDPrefix) == "" && strings.TrimSpace(rule.VarbindMatchRegex) == "" {
 		return fmt.Errorf("Trap 规则的匹配条件全部为空，运行时永远不会命中；请至少填写 oid_prefix、varbind_match_regex 中的一项")
 	}
+	if strings.TrimSpace(rule.RecoveryMatchRegex) != "" && strings.TrimSpace(rule.LifecycleKey) == "" {
+		return fmt.Errorf("配置 recovery_match_regex 时 lifecycle_key 不能为空")
+	}
 	return nil
 }
 

internal/models/alert_outbox.go

@@ -1,29 +1,31 @@
-package models
-
-import "time"
-
-// AlertOutbox 表示待发送或重试中的告警任务。
-type AlertOutbox struct {
-	ID uint `gorm:"primaryKey" json:"id"`
-
-	CreatedAt time.Time `json:"created_at"`
-	UpdatedAt time.Time `json:"updated_at"`
-
-	// LogEventID 关联日志事件 ID。
-	LogEventID uint `gorm:"index" json:"log_event_id"`
-	// PayloadJSON 保存 AlertReceiveBody 的 JSON 文本。
-	PayloadJSON string `gorm:"type:text" json:"payload_json"`
-	// Status 任务状态：pending/retrying/sent/dead。
-	Status string `gorm:"size:32;index" json:"status"`
-	// RetryCount 已重试次数。
-	RetryCount int `json:"retry_count"`
-	// NextRetryAt 下一次可重试时间。
-	NextRetryAt time.Time `gorm:"index" json:"next_retry_at"`
-	// LastError 最近一次错误信息。
-	LastError string `gorm:"type:text" json:"last_error"`
-}
-
-func (AlertOutbox) TableName() string {
-	return "logs_alert_outbox"
-}
-
+package models
+
+import "time"
+
+// AlertOutbox 表示待发送或重试中的告警任务。
+type AlertOutbox struct {
+	ID uint `gorm:"primaryKey;index:idx_logs_alert_outbox_event_latest,priority:2" json:"id"`
+
+	CreatedAt time.Time `json:"created_at"`
+	UpdatedAt time.Time `json:"updated_at"`
+
+	// LogEventID 关联日志事件 ID。
+	LogEventID uint `gorm:"index;index:idx_logs_alert_outbox_event_latest,priority:1" json:"log_event_id"`
+	// PayloadJSON 保存 AlertReceiveBody 的 JSON 文本。
+	PayloadJSON string `gorm:"type:text" json:"payload_json"`
+	// Status 任务状态：pending/processing/retrying/sent/dead。
+	Status string `gorm:"size:32;index" json:"status"`
+	// RetryCount 已重试次数。
+	RetryCount int `json:"retry_count"`
+	// NextRetryAt 下一次可重试时间。
+	NextRetryAt time.Time `gorm:"index" json:"next_retry_at"`
+	// LeaseUntil/LeaseOwner 用于多实例原子领取，避免同一任务并发发送。
+	LeaseUntil *time.Time `gorm:"index" json:"lease_until,omitempty"`
+	LeaseOwner string     `gorm:"size:128;index" json:"lease_owner"`
+	// LastError 最近一次错误信息。
+	LastError string `gorm:"type:text" json:"last_error"`
+}
+
+func (AlertOutbox) TableName() string {
+	return "logs_alert_outbox"
+}

internal/models/log_event.go

@@ -1,45 +1,49 @@
-package models
-
-import "time"
-
-// LogEvent 表示一条归一化/存储后的日志事件。
-type LogEvent struct {
-	// ID 是数据库主键。
-	ID uint `gorm:"primaryKey" json:"id"`
-	// CreatedAt 记录创建时间（写入日志事件时）。
-	CreatedAt time.Time `json:"created_at"`
-	// SourceKind 表示日志来源类型（例如 trap/syslog 等）。
-	SourceKind string `gorm:"size:16;index" json:"source_kind"`
-	// RemoteAddr 表示日志发送方地址。
-	RemoteAddr string `gorm:"size:64" json:"remote_addr"`
-	// RawPayload 保存原始负载内容。
-	RawPayload string `gorm:"type:text" json:"raw_payload"`
-	// NormalizedSummary 保存归一化后的摘要信息。
-	NormalizedSummary string `gorm:"type:text" json:"normalized_summary"`
-	// NormalizedDetail 保存归一化后的详细信息。
-	NormalizedDetail string `gorm:"type:text" json:"normalized_detail"`
-	// DeviceName 表示关联设备名称。
-	DeviceName string `gorm:"size:512;index" json:"device_name"`
-	// SourceIP 表示原始来源 IP（不含端口）。
-	SourceIP string `gorm:"size:64;index" json:"source_ip"`
-	// ResourceType 表示关联到的资源类型。
-	ResourceType string `gorm:"size:32;index" json:"resource_type"`
-	// ResourceID 表示关联到的资源 ID。
-	ResourceID string `gorm:"size:128;index" json:"resource_id"`
-	// ResourceName 表示关联到的资源名称。
-	ResourceName string `gorm:"size:256" json:"resource_name"`
-	// MatchMethod 表示资源命中方式（ip/hostname/none）。
-	MatchMethod string `gorm:"size:32" json:"match_method"`
-	// DispatchStatus 表示告警分发状态（not_applicable/pending/retrying/sent/dead）。
-	DispatchStatus string `gorm:"size:32;index" json:"dispatch_status"`
-	// SeverityCode 表示告警/严重度编码。
-	SeverityCode string `gorm:"size:32" json:"severity_code"`
-	// TrapOID 表示关联的 Trap OID（若来源为 trap）。
-	TrapOID string `gorm:"size:512;index" json:"trap_oid"`
-	// AlertSent 表示是否已将告警发送出去。
-	AlertSent bool `json:"alert_sent"`
-}
-
-func (LogEvent) TableName() string {
-	return "logs_events"
-}
+package models
+
+import "time"
+
+// LogEvent 表示一条归一化/存储后的日志事件。
+type LogEvent struct {
+	// ID 是数据库主键。
+	ID uint `gorm:"primaryKey" json:"id"`
+	// CreatedAt 记录创建时间（写入日志事件时）。
+	CreatedAt time.Time `json:"created_at"`
+	// SourceKind 表示日志来源类型（例如 trap/syslog 等）。
+	SourceKind string `gorm:"size:16;index" json:"source_kind"`
+	// RemoteAddr 表示日志发送方地址。
+	RemoteAddr string `gorm:"size:64" json:"remote_addr"`
+	// RawPayload 保存原始负载内容。
+	RawPayload string `gorm:"type:text" json:"raw_payload"`
+	// NormalizedSummary 保存归一化后的摘要信息。
+	NormalizedSummary string `gorm:"type:text" json:"normalized_summary"`
+	// NormalizedDetail 保存归一化后的详细信息。
+	NormalizedDetail string `gorm:"type:text" json:"normalized_detail"`
+	// DeviceName 表示关联设备名称。
+	DeviceName string `gorm:"size:512;index" json:"device_name"`
+	// SourceIP 表示原始来源 IP（不含端口）。
+	SourceIP string `gorm:"size:64;index" json:"source_ip"`
+	// ResourceType 表示关联到的资源类型。
+	ResourceType string `gorm:"size:32;index" json:"resource_type"`
+	// ResourceUID 是跨服务使用的规范资源标识。
+	ResourceUID string `gorm:"size:255;index" json:"resource_uid"`
+	// ResourceID 表示关联到的资源 ID。
+	ResourceID string `gorm:"size:128;index" json:"resource_id"`
+	// ResourceName 表示关联到的资源名称。
+	ResourceName string `gorm:"size:256" json:"resource_name"`
+	// MatchMethod 表示资源命中方式（ip/hostname/none）。
+	MatchMethod string `gorm:"size:32" json:"match_method"`
+	// DispatchStatus 表示告警分发状态（not_applicable/pending/retrying/sent/dead）。
+	DispatchStatus string `gorm:"size:32;index" json:"dispatch_status"`
+	// DispatchOutboxID 标识当前一次分发，防止旧任务覆盖较新的重放状态。
+	DispatchOutboxID uint `gorm:"not null;default:0;index" json:"dispatch_outbox_id"`
+	// SeverityCode 表示告警/严重度编码。
+	SeverityCode string `gorm:"size:32" json:"severity_code"`
+	// TrapOID 表示关联的 Trap OID（若来源为 trap）。
+	TrapOID string `gorm:"size:512;index" json:"trap_oid"`
+	// AlertSent 表示是否已将告警发送出去。
+	AlertSent bool `json:"alert_sent"`
+}
+
+func (LogEvent) TableName() string {
+	return "logs_events"
+}

internal/models/query.go

@@ -1,6 +1,10 @@
 package models
 
-import "gorm.io/gorm"
+import (
+	"errors"
+
+	"gorm.io/gorm"
+)
 
 // GetAllModels 数据库迁移用模型列表
 func GetAllModels() []interface{} {
@@ -23,6 +27,9 @@ func InitData(db *gorm.DB) error {
 	if db == nil {
 		return nil
 	}
+	if err := backfillDispatchOutboxIDs(db); err != nil {
+		return err
+	}
 	if err := seedDefaultSyslogRules(db); err != nil {
 		return err
 	}
@@ -35,14 +42,53 @@ func InitData(db *gorm.DB) error {
 	return nil
 }
 
+func backfillDispatchOutboxIDs(db *gorm.DB) error {
+	const postgresUpdate = `
+WITH latest AS (
+    SELECT target_event.id AS log_event_id, current_outbox.id, current_outbox.status
+    FROM logs_events AS target_event
+    JOIN LATERAL (
+        SELECT id, status
+        FROM logs_alert_outbox
+        WHERE log_event_id = target_event.id
+        ORDER BY id DESC
+        LIMIT 1
+    ) AS current_outbox ON TRUE
+    WHERE target_event.dispatch_outbox_id = 0
+)
+UPDATE logs_events AS target_event
+SET dispatch_outbox_id = latest.id,
+    dispatch_status = CASE latest.status WHEN 'processing' THEN 'retrying' ELSE latest.status END,
+    alert_sent = CASE WHEN latest.status = 'sent' THEN TRUE ELSE FALSE END
+FROM latest
+WHERE target_event.id = latest.log_event_id
+  AND target_event.dispatch_outbox_id = 0`
+	const mysqlUpdate = `
+UPDATE logs_events AS target_event
+JOIN (
+    SELECT candidate.id AS log_event_id, MAX(outbox.id) AS id
+    FROM logs_events AS candidate
+    JOIN logs_alert_outbox AS outbox ON outbox.log_event_id = candidate.id
+    WHERE candidate.dispatch_outbox_id = 0
+    GROUP BY candidate.id
+) AS latest_id ON latest_id.log_event_id = target_event.id
+JOIN logs_alert_outbox AS latest ON latest.id = latest_id.id
+SET dispatch_outbox_id = latest.id,
+    dispatch_status = CASE latest.status WHEN 'processing' THEN 'retrying' ELSE latest.status END,
+    alert_sent = CASE WHEN latest.status = 'sent' THEN TRUE ELSE FALSE END
+WHERE target_event.dispatch_outbox_id = 0`
+
+	switch db.Dialector.Name() {
+	case "postgres":
+		return db.Exec(postgresUpdate).Error
+	case "mysql":
+		return db.Exec(mysqlUpdate).Error
+	default:
+		return gorm.ErrUnsupportedDriver
+	}
+}
+
 func seedDefaultSyslogRules(db *gorm.DB) error {
-	var cnt int64
-	if err := db.Model(&SyslogRule{}).Count(&cnt).Error; err != nil {
-		return err
-	}
-	if cnt > 0 {
-		return nil
-	}
 	rows := []SyslogRule{
 		{
 			Name:               "默认-系统严重错误",
@@ -62,6 +108,8 @@ func seedDefaultSyslogRules(db *gorm.DB) error {
 			KeywordRegex:            "(?i)(link down|interface .* down|port .* down)",
 			SourceMatch:             "",
 			MessageRegex:            "(?i)(link down|interface .* down|port .* down|LINK_DOWN)",
+			RecoveryMatchRegex:      `(?i)(link[ _-]?up|interface .* up|port .* up|ifup)`,
+			LifecycleKey:            "syslog-link-state",
 			AlertName:               "Syslog链路中断",
 			SeverityCode:            "major",
 			SeverityMappingJSON:     `{"(?i)(critical|fatal|emergency)":"critical","(?i)(error|LINK_DOWN|down)":"major","(?i)(warning|warn)":"warning"}`,
@@ -72,8 +120,10 @@ func seedDefaultSyslogRules(db *gorm.DB) error {
 			Name:                    "H3C-Syslog-接口中断",
 			Enabled:                 true,
 			Priority:                120,
-			SourceMatch:             "h3c",
+			DeviceNameContains:      "h3c",
 			MessageRegex:            `(?i)(LINK_DOWN|Interface .* down|port .* down)`,
+			RecoveryMatchRegex:      `(?i)(link[ _-]?up|interface .* up|port .* up|ifup)`,
+			LifecycleKey:            "h3c-syslog-interface-state",
 			AlertName:               "H3C Syslog接口中断",
 			SeverityCode:            "major",
 			SeverityMappingJSON:     `{"(?i)(LINK_DOWN|down)":"major","(?i)(LINK_UP|up)":"info"}`,
@@ -81,40 +131,86 @@ func seedDefaultSyslogRules(db *gorm.DB) error {
 			PolicyID:                0,
 		},
 	}
-	return db.Create(&rows).Error
+	for _, row := range rows {
+		var existing SyslogRule
+		err := db.Where("name = ?", row.Name).First(&existing).Error
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			if err := db.Create(&row).Error; err != nil {
+				return err
+			}
+			continue
+		}
+		if err != nil {
+			return err
+		}
+		if err := db.Model(&existing).Omit("updated_at").Select(
+			"name",
+			"enabled",
+			"priority",
+			"device_name_contains",
+			"source_match",
+			"keyword_regex",
+			"message_regex",
+			"recovery_match_regex",
+			"lifecycle_key",
+			"alert_name",
+			"severity_code",
+			"severity_mapping_json",
+			"resource_uid_extract_regex",
+			"policy_id",
+		).Updates(&row).Error; err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 func seedDefaultTrapRules(db *gorm.DB) error {
-	var cnt int64
-	if err := db.Model(&TrapRule{}).Count(&cnt).Error; err != nil {
-		return err
-	}
-	if cnt > 0 {
-		return nil
-	}
 	rows := []TrapRule{
 		{
-			Name:              "默认-Trap链路中断",
-			Enabled:           true,
-			Priority:          100,
-			OIDPrefix:         "1.3.6.1.6.3.1.1.5",
-			VarbindMatchRegex: "(?i)(linkdown|ifdown|down)",
-			AlertName:         "SNMP Trap链路中断",
-			SeverityCode:      "major",
-			PolicyID:          0,
+			Name:               "默认-Trap链路中断",
+			Enabled:            true,
+			Priority:           100,
+			OIDPrefix:          "1.3.6.1.6.3.1.1.5",
+			VarbindMatchRegex:  `(?i)(1\.3\.6\.1\.6\.3\.1\.1\.5\.3([^0-9]|$)|\b(linkdown|ifdown|down)\b)`,
+			RecoveryMatchRegex: `(?i)(1\.3\.6\.1\.6\.3\.1\.1\.5\.4([^0-9]|$)|\b(linkup|ifup)\b)`,
+			LifecycleKey:       "snmp-interface-link-state",
+			AlertName:          "SNMP Trap链路中断",
+			SeverityCode:       "major",
+			PolicyID:           0,
 		},
 	}
-	return db.Create(&rows).Error
+	for _, row := range rows {
+		var existing TrapRule
+		err := db.Where("name = ?", row.Name).First(&existing).Error
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			if err := db.Create(&row).Error; err != nil {
+				return err
+			}
+			continue
+		}
+		if err != nil {
+			return err
+		}
+		if err := db.Model(&existing).Omit("updated_at").Select(
+			"name",
+			"enabled",
+			"priority",
+			"o_id_prefix",
+			"varbind_match_regex",
+			"recovery_match_regex",
+			"lifecycle_key",
+			"alert_name",
+			"severity_code",
+			"policy_id",
+		).Updates(&row).Error; err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 func seedDefaultTrapDictionary(db *gorm.DB) error {
-	var cnt int64
-	if err := db.Model(&TrapDictionaryEntry{}).Count(&cnt).Error; err != nil {
-		return err
-	}
-	if cnt > 0 {
-		return nil
-	}
 	rows := []TrapDictionaryEntry{
 		{
 			Vendor:              "H3C",
@@ -143,5 +239,33 @@ func seedDefaultTrapDictionary(db *gorm.DB) error {
 			Enabled:             true,
 		},
 	}
-	return db.Create(&rows).Error
+	for _, row := range rows {
+		var existing TrapDictionaryEntry
+		err := db.Where("o_id_prefix = ?", row.OIDPrefix).First(&existing).Error
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			if err := db.Create(&row).Error; err != nil {
+				return err
+			}
+			continue
+		}
+		if err != nil {
+			return err
+		}
+		if err := db.Model(&existing).Omit("updated_at").Select(
+			"o_id_prefix",
+			"vendor",
+			"o_id",
+			"name",
+			"title",
+			"description",
+			"severity_code",
+			"severity_mapping_json",
+			"parse_expression",
+			"recovery_message",
+			"enabled",
+		).Updates(&row).Error; err != nil {
+			return err
+		}
+	}
+	return nil
 }

internal/models/syslog_rule.go

@@ -24,6 +24,10 @@ type SyslogRule struct {
 	KeywordRegex string `gorm:"size:512" json:"keyword_regex"`
 	// MessageRegex 表示消息正文匹配的正则表达式。
 	MessageRegex string `gorm:"size:1024" json:"message_regex"`
+	// RecoveryMatchRegex 匹配同一生命周期的恢复消息。
+	RecoveryMatchRegex string `gorm:"size:1024" json:"recovery_match_regex"`
+	// LifecycleKey 将故障和恢复事件绑定到同一告警生命周期。
+	LifecycleKey string `gorm:"size:256" json:"lifecycle_key"`
 	// AlertName 表示告警名称。
 	AlertName string `gorm:"size:256" json:"alert_name"`
 	// SeverityCode 表示严重级别编码。

internal/models/trap_rule.go

@@ -1,33 +1,37 @@
-package models
-
-import "time"
-
-// TrapRule 表示一条 SNMP Trap 规则，用于匹配并触发告警策略。
-type TrapRule struct {
-	// ID 是数据库主键。
-	ID uint `gorm:"primaryKey" json:"id"`
-	// CreatedAt 记录创建时间（GORM 自动维护）。
-	CreatedAt time.Time `json:"created_at"`
-	// UpdatedAt 记录更新时间（GORM 自动维护）。
-	UpdatedAt time.Time `json:"updated_at"`
-	// Name 规则名称，用于展示/标识。
-	Name string `gorm:"size:256" json:"name"`
-	// Enabled 表示该规则是否启用。
-	Enabled bool `gorm:"default:true" json:"enabled"`
-	// Priority 表示匹配优先级（数值越高/低需以业务约定为准）。
-	Priority int `gorm:"index" json:"priority"`
-	// OIDPrefix 表示匹配的 OID 前缀。
-	OIDPrefix string `gorm:"size:512" json:"oid_prefix"`
-	// VarbindMatchRegex 表示对 varbind 内容的正则匹配条件。
-	VarbindMatchRegex string `gorm:"size:512" json:"varbind_match_regex"`
-	// AlertName 表示告警名称。
-	AlertName string `gorm:"size:256" json:"alert_name"`
-	// SeverityCode 表示严重级别编码。
-	SeverityCode string `gorm:"size:32" json:"severity_code"`
-	// PolicyID 表示关联的告警/处理策略 ID。
-	PolicyID uint `json:"policy_id"`
-}
-
-func (TrapRule) TableName() string {
-	return "logs_trap_rules"
-}
+package models
+
+import "time"
+
+// TrapRule 表示一条 SNMP Trap 规则，用于匹配并触发告警策略。
+type TrapRule struct {
+	// ID 是数据库主键。
+	ID uint `gorm:"primaryKey" json:"id"`
+	// CreatedAt 记录创建时间（GORM 自动维护）。
+	CreatedAt time.Time `json:"created_at"`
+	// UpdatedAt 记录更新时间（GORM 自动维护）。
+	UpdatedAt time.Time `json:"updated_at"`
+	// Name 规则名称，用于展示/标识。
+	Name string `gorm:"size:256" json:"name"`
+	// Enabled 表示该规则是否启用。
+	Enabled bool `gorm:"default:true" json:"enabled"`
+	// Priority 表示匹配优先级（数值越高/低需以业务约定为准）。
+	Priority int `gorm:"index" json:"priority"`
+	// OIDPrefix 表示匹配的 OID 前缀。
+	OIDPrefix string `gorm:"size:512" json:"oid_prefix"`
+	// VarbindMatchRegex 表示对 varbind 内容的正则匹配条件。
+	VarbindMatchRegex string `gorm:"size:512" json:"varbind_match_regex"`
+	// RecoveryMatchRegex 匹配同一生命周期的恢复 Trap OID 或 varbind。
+	RecoveryMatchRegex string `gorm:"size:1024" json:"recovery_match_regex"`
+	// LifecycleKey 将故障和恢复事件绑定到同一告警生命周期。
+	LifecycleKey string `gorm:"size:256" json:"lifecycle_key"`
+	// AlertName 表示告警名称。
+	AlertName string `gorm:"size:256" json:"alert_name"`
+	// SeverityCode 表示严重级别编码。
+	SeverityCode string `gorm:"size:32" json:"severity_code"`
+	// PolicyID 表示关联的告警/处理策略 ID。
+	PolicyID uint `json:"policy_id"`
+}
+
+func (TrapRule) TableName() string {
+	return "logs_trap_rules"
+}