ai-security-xdr/haobang-security-dm/upgrade-doc/v3

v3 日统计异步投影与大日志通道升级说明

升级内容

v3 将以下统计表的热点更新移出 syslog-consumer-rule 事实事务：

• device_daily_stats
• normalize_rule_daily_stats
• device_collect_stats

事实事务只批量写事实表并追加一条聚合 Outbox。独立的 Kafka 投影消费者串行更新日统计，之后继续通过 已有的 STATS.MATERIALIZE Topic 刷新 device_device、dm_normalize_rule 和 device_collect_task。单节点和多节点部署都不再让事实写入争抢日统计行锁。

v3 同时增加大日志独立处理链路：

• syslog-serve 在进入 WAL 前执行单条硬上限检查；超过上限明确拒绝，不把原文写入应用日志；
• 接纳后的消息仍先进入 syslog-serve 本地 WAL，WAL 入队和 Kafka 投递同时受条数、字节数约束；
• 达到分流阈值的消息进入独立 .large Topic，其 Retry、DLT、Replay 也使用独立 Topic；
• syslog-consumer-rule 使用独立的大日志 listener、消费组、ConsumerFactory、poll 批量和规则计算线程池；
• 规则解析使用完整内容，写 InfluxDB 和达梦前才按 UTF-8 字节数截断，结尾带 HB_XDR_TRUNCATED 标记；InfluxDB 另存原长度、保存长度和原文 SHA-256。
• 消费端默认采用连续性优先策略：无法理解或系统性异常先尝试写入 DLT；即使 DLT 失败，也记录 [SYSLOG_CONTINUITY_LOSS] 后推进 offset，避免单条异常永久阻塞整个分区。

文件与执行顺序

顺序	文件	作用
1	database/01-stats-async-projection-precheck.sql	只读预检
2	database/02-stats-async-projection-upgrade.sql	创建 Outbox 和消费幂等表
3	database/03-stats-async-projection-verify.sql	只读验证表、约束和索引
4	kafka/01-create-stats-projection-topic.sh	创建独立统计投影 Topic
5	kafka/02-create-large-log-topics.sh	创建大日志主 Topic 及独立 Retry、DLT、Replay Topic
6	config/01-new-environment-variables.md	升级应用前配置新增环境变量，并核对 Kafka Broker 参数

数据库脚本必须使用 syslog-consumer-rule 实际连接的达梦用户执行。

如果现场已经执行过早期 v3 日统计异步投影升级，只执行大日志 Kafka 脚本和应用配置，不要重复执行 数据库升级脚本。

第一步：达梦

依次执行：

01-stats-async-projection-precheck.sql
02-stats-async-projection-upgrade.sql
03-stats-async-projection-verify.sql

预检判断：

• 第一组应返回 DEVICE_DAILY_STATS、NORMALIZE_RULE_DAILY_STATS、DEVICE_COLLECT_STATS 三行；
• 后面三组必须为空；
• 如果目标表、约束或索引已经存在，不要执行 02，先执行 03 判断是否已经升级完成。

验证判断：

• 返回两张新表；
• 三个约束状态为 ENABLED；
• 两个索引状态为 VALID；
• 首次建表时两张表的行数都是 0。

第二步：Kafka

统计投影 Topic

在 Kafka 主机的 kafka_2.13-3.6.1/bin 目录执行：

bash /脚本所在路径/01-create-stats-projection-topic.sh

默认创建：

agent-01-syslog-topic.STATS.PROJECTION

新建时为 1 个分区，副本数沿用 agent-01-syslog-topic。如果现场此前已经建成 6 个分区，脚本只告警， 不会修改或删除；应用使用固定 Kafka key，投影事件仍然只落入其中一个稳定分区。

必须看到：

PASS: agent-01-syslog-topic.STATS.PROJECTION 已创建/确认。

STATS.PROJECTION 不能与已有的 STATS.MATERIALIZE 配成同一个 Topic。

大日志 Topic

先确认所有 Kafka Broker 的 replica.fetch.max.bytes 不小于 33554432，并按现场 Kafka 版本要求使 配置生效。然后执行：

bash /脚本所在路径/02-create-large-log-topics.sh

脚本默认从 agent-01-syslog-topic 读取分区数和副本数，并创建：

agent-01-syslog-topic.large
agent-01-syslog-topic.large.RETRY.1M
agent-01-syslog-topic.large.RETRY.10M
agent-01-syslog-topic.large.RETRY.1H
agent-01-syslog-topic.large.DLT
agent-01-syslog-topic.large.REPLAY

必须看到：

PASS: 大日志 Topic 已创建并统一配置。

第三步：升级应用

升级 syslog-serve 和现场的 syslog-consumer-rule；syslog-consumer 不需要因为本次改造升级。

升级前先按 新增变量配置清单 修改现场实际使用的 Compose 文件。 如果部署了多个 syslog-consumer-rule，它们的 Topic、消费组和截断上限必须保持一致。

统计投影默认已经启用。确认现场没有显式配置：

STATS_PROJECTION_ASYNC_ENABLED=false

先升级 syslog-consumer-rule；多实例时逐台滚动。确认大日志 listener 正常后再升级 syslog-serve。

升级后验证

查看处理日志：

docker logs --since 2m syslog-consumer-rule-dm 2>&1 \
| grep -E 'Syslog批次完成|统计投影批次完成|统计物化批次完成|Outbox|Deadlock' \
| tail -100

查看 Outbox 状态：

SELECT status, COUNT(*)
FROM stats_projection_outbox
GROUP BY status
ORDER BY status;

正常表现：

• 日志持续出现 统计投影批次完成；
• APPLIED 持续增长；
• PENDING、PUBLISHED 只短暂停留，不持续增长；
• FAILED 不持续增长；
• Syslog批次完成 中 dmWriteMs 明显下降；
• 不再出现 daily stats 相关死锁。

查看新消费组积压：

./kafka-consumer-groups.sh \
  --bootstrap-server 10.141.121.35:9092,10.141.121.50:9092,10.141.121.59:9092 \
  --group agent-01-syslog-group-dm-stats-projection \
  --describe

验证大日志链路：

• GET /syslogserve/api/syslog/wal/status：检查 wal_admission_oversized、wal_pending_bytes、 dispatch_inflight_bytes、kafka_large_dispatched；
• GET /xdrrule/api/syslog/raw-storage/metrics：检查 raw_message_truncated_total 及截断前后字节累计值；
• GET /xdrrule/api/syslog/storage-limits/metrics：检查达梦正常/非标字段截断计数；
• 发送一条不小于 128 KiB、但小于 6 MiB 的测试消息，确认 .large Topic 有数据且普通 Topic 无该事件；
• 查询 InfluxDB/达梦，确认被截断值以 ...[HB_XDR_TRUNCATED original_utf8_bytes=...] 结尾。

连续性指标也位于 GET /xdrrule/api/syslog/raw-storage/metrics：

• raw_write_bypassed_messages_total：InfluxDB 原文写失败后仍继续标准化的消息数；
• continuity_dlt_published_total：特殊异常成功写入 DLT 的消息数；
• continuity_dlt_publish_failed_total：DLT 写失败但仍放行的消息数。该值增加时应立即检查 Kafka/DLT， 对应消息可能无法自动重放。

回滚

回滚最后一台新版本前，先确认 Outbox 没有未应用事件：

SELECT status, COUNT(*)
FROM stats_projection_outbox
WHERE status <> 'APPLIED'
GROUP BY status;

查询为空后再回滚最后一台。两张新表、Kafka Topic 和消费组 offset 全部保留，不要删除或重置。

回滚旧版本前，还要先停止新生产者并排空 .large 主 Topic；大日志 Topic 和消费组 offset 同样保留， 不要删除或重置。