feat(upgrade): 增加升级脚本和部署说明

- 新增达梦数据库结构、幂等索引、统计基线及校验脚本 - 新增 Docker Kafka Retry、DLT、Replay Topic 升级脚本 - 更新生产 Compose，启用 DIRECT 并关闭旧增量捕获和投影 - 补充数据库、Kafka、应用启动及验证说明 - 增加日志和本地生成文件忽略规则
2026-08-24 15:09:36 +08:00
parent 29445c6825
commit f53661931f
15 changed files with 694 additions and 38 deletions
@@ -0,0 +1,115 @@
+# 系统升级说明（简约版）
+
+本目录用于把旧备份版本升级到当前版本，只需要按下面顺序操作。
+
+## 1. 升级前
+
+1. 备份达梦数据库，保留旧版 JAR 和配置。
+2. 记录 Kafka 原始 Topic、当前正在使用的消费组和 offset。
+3. 停止 `syslog-consumer-rule`、`syslog-consumer` 全部实例。
+
+> 不要清空 Kafka，不要重置 offset，不要删除数据库历史数据。
+
+## 2. 升级数据库
+
+DBA 按顺序执行：
+
+```text
+database/01-schema-upgrade.sql
+database/02-idempotency-indexes.sql
+database/03-direct-stats-baseline.sql
+database/04-verify.sql
+```
+
+执行 `03`、`04` 前，把脚本中的日期占位符全部替换为升级当天：
+
+```text
+BUSINESS_DATE -> 2026-08-24
+YYYYMMDD      -> 20260824
+```
+
+日期仅为示例，以实际升级日为准。任一 SQL 报错就停止，不要从头反复执行。
+
+## 3. 升级 Kafka（Docker）
+
+先在 Kafka 宿主机确认容器名：
+
+```bash
+docker ps --format '{{.Names}}' | grep -i kafka
+```
+
+进入本升级目录的上级目录（能够看到 `upgrade-doc`），执行：
+
+```bash
+export KAFKA_CONTAINER=kafka_ycgy-bt-kafka-1
+
+docker exec -i \
+  -e KAFKA_BIN=/opt/kafka/bin \
+  -e BOOTSTRAP=localhost:9092 \
+  -e RAW_TOPIC=agent-01-syslog-topic \
+  "$KAFKA_CONTAINER" \
+  bash -s < upgrade-doc/kafka/kafka-upgrade.sh
+```
+
+如果现场容器名不同，只修改 `KAFKA_CONTAINER`。如果 Kafka 容器内的命令目录或端口不同，再调整 `KAFKA_BIN`、`BOOTSTRAP`。
+
+执行结束出现以下内容表示成功：
+
+```text
+PASS: Kafka Topic 已创建/确认；脚本未删除 Topic，也未修改消费组 offset。
+```
+
+脚本只新增以下 5 个 Topic，不会删除 Topic，也不会修改消费组 offset：
+
+```text
+agent-01-syslog-topic.RETRY.1M
+agent-01-syslog-topic.RETRY.10M
+agent-01-syslog-topic.RETRY.1H
+agent-01-syslog-topic.DLT
+agent-01-syslog-topic.REPLAY
+```
+
+## 4. 应用配置
+
+`syslog-consumer-rule`：
+
+```text
+SYSLOG_RAW_CONSUMER_ENABLED=true
+STATS_WRITE_MODE=DIRECT
+STATS_INCREMENTAL_CAPTURE_ENABLED=false
+STATS_INCREMENTAL_READ_ENABLED=true
+STATS_INGRESS_METADATA_ENABLED=true
+SPRING_KAFKA_CONSUMER_GROUP_ID=现场原来的活动消费组
+```
+
+`syslog-consumer`：
+
+```text
+SYSLOG_RAW_CONSUMER_ENABLED=false
+STATS_WRITE_MODE=DIRECT
+STATS_INCREMENTAL_CAPTURE_ENABLED=false
+STATS_INCREMENTAL_PROJECTION_ENABLED=false
+STATS_INCREMENTAL_READ_ENABLED=true
+STATS_LEGACY_FULL_SCAN_ENABLED=false
+STATS_INGRESS_METADATA_ENABLED=true
+```
+
+本次升级只使用 `DIRECT`，禁止再启用以下旧模式开关：
+
+```text
+STATS_INCREMENTAL_CAPTURE_ENABLED=true
+STATS_INCREMENTAL_PROJECTION_ENABLED=true
+```
+
+## 5. 启动和确认
+
+按顺序启动：
+
+1. `syslog-consumer`
+2. 一台 `syslog-consumer-rule` 灰度实例
+3. 其余 `syslog-consumer-rule` 实例
+4. `syslog-serve`
+
+确认 Kafka 消费延迟持续下降、达梦数据持续写入、Retry/DLT 没有异常增长后，升级完成。
+
+如需回滚，只回滚应用 JAR 和配置；数据库新增表、字段、索引及 Kafka 新 Topic 保留，不要清库或重置 offset。
@@ -0,0 +1,224 @@
+-- haobang-security-dm：旧备份基线 -> 当前版本结构升级。
+-- 前置：已完成备份，并停止所有目标表写入者。
+-- 本文件为一次性 DDL。发生任一错误立即停止，禁止整文件盲目重跑。
+
+-- 1. Retry/DLT 可检索索引和人工回放对象。
+CREATE TABLE syslog_failure_record (
+  id VARCHAR2(36) NOT NULL,
+  event_id VARCHAR2(80) NOT NULL,
+  device_id VARCHAR2(64),
+  original_topic VARCHAR2(255) NOT NULL,
+  original_partition INT NOT NULL,
+  original_offset BIGINT NOT NULL,
+  current_topic VARCHAR2(255),
+  first_received_at TIMESTAMP,
+  failed_stage VARCHAR2(64) NOT NULL,
+  error_code VARCHAR2(128),
+  error_type VARCHAR2(512),
+  error_message VARCHAR2(2048),
+  attempt_count INT DEFAULT 0 NOT NULL,
+  next_retry_at TIMESTAMP,
+  rule_id VARCHAR2(128),
+  rule_version VARCHAR2(64),
+  consumer_version VARCHAR2(64),
+  point_time_ns BIGINT,
+  original_payload CLOB NOT NULL,
+  status VARCHAR2(32) DEFAULT 'OPEN' NOT NULL,
+  replay_batch_id VARCHAR2(36),
+  replay_count INT DEFAULT 0 NOT NULL,
+  last_replay_at TIMESTAMP,
+  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_syslog_failure_record PRIMARY KEY(id),
+  CONSTRAINT uk_syslog_failure_origin
+    UNIQUE(original_topic,original_partition,original_offset,failed_stage)
+);
+
+COMMENT ON TABLE syslog_failure_record IS 'Syslog Retry/DLT 异常消息可检索索引，Kafka DLT 为原始事实源';
+COMMENT ON COLUMN syslog_failure_record.event_id IS '接入生成并在 Retry/Replay 中保持不变的事件标识';
+COMMENT ON COLUMN syslog_failure_record.point_time_ns IS 'InfluxDB V2 Point 纳秒时间戳；旧 DLT 记录允许为空';
+COMMENT ON COLUMN syslog_failure_record.original_payload IS '原始加密载荷，禁止接口明文返回';
+COMMENT ON COLUMN syslog_failure_record.status IS 'OPEN/REPLAYED/RESOLVED 等补偿状态';
+
+CREATE INDEX idx_syslog_failure_event
+ON syslog_failure_record(event_id);
+
+CREATE INDEX idx_syslog_failure_filter
+ON syslog_failure_record(device_id,error_code,rule_version,first_received_at);
+
+CREATE TABLE syslog_replay_batch (
+  id VARCHAR2(36) NOT NULL,
+  status VARCHAR2(32) NOT NULL,
+  operator_name VARCHAR2(128) NOT NULL,
+  reason VARCHAR2(1000) NOT NULL,
+  rate_per_second INT NOT NULL,
+  total_count INT NOT NULL,
+  success_count INT DEFAULT 0 NOT NULL,
+  failed_count INT DEFAULT 0 NOT NULL,
+  duplicate_count INT DEFAULT 0 NOT NULL,
+  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  completed_at TIMESTAMP,
+  CONSTRAINT pk_syslog_replay_batch PRIMARY KEY(id)
+);
+
+COMMENT ON TABLE syslog_replay_batch IS '人工补偿回放批次及操作审计';
+
+CREATE TABLE syslog_replay_item (
+  id VARCHAR2(36) NOT NULL,
+  replay_batch_id VARCHAR2(36) NOT NULL,
+  failure_id VARCHAR2(36) NOT NULL,
+  status VARCHAR2(32) NOT NULL,
+  error_message VARCHAR2(1000),
+  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_syslog_replay_item PRIMARY KEY(id),
+  CONSTRAINT uk_syslog_replay_item UNIQUE(replay_batch_id,failure_id)
+);
+
+COMMENT ON TABLE syslog_replay_item IS '回放批次逐条状态';
+
+CREATE INDEX idx_syslog_replay_pending
+ON syslog_replay_item(replay_batch_id,status,created_at);
+
+-- 2. 标准化、规则告警和算法告警幂等字段。
+-- 只增加可空列；历史数据保持 NULL，不执行无法可靠还原版本/窗口的全表 datafix。
+ALTER TABLE syslog_normal_data ADD rule_version VARCHAR2(64);
+ALTER TABLE syslog_normal_alarm ADD rule_version VARCHAR2(64);
+
+COMMENT ON COLUMN syslog_normal_data.rule_version IS '标准化结果使用的规则版本；确定性 id 的组成部分';
+COMMENT ON COLUMN syslog_normal_alarm.rule_version IS '标准化结果使用的规则版本；确定性 id 的组成部分';
+
+ALTER TABLE alarm ADD rule_id VARCHAR2(128);
+ALTER TABLE alarm ADD rule_version VARCHAR2(64);
+ALTER TABLE alarm ADD window_start TIMESTAMP;
+ALTER TABLE alarm ADD group_key_hash VARCHAR2(64);
+
+COMMENT ON COLUMN alarm.rule_id IS '生成告警的关联分析规则标识';
+COMMENT ON COLUMN alarm.rule_version IS '生成告警的关联分析规则版本';
+COMMENT ON COLUMN alarm.window_start IS '聚合告警业务窗口起点';
+COMMENT ON COLUMN alarm.group_key_hash IS '规范化分组结果的稳定哈希';
+
+ALTER TABLE alarm_visit ADD algorithm_id VARCHAR2(128);
+ALTER TABLE alarm_visit ADD algorithm_version VARCHAR2(64);
+ALTER TABLE alarm_visit ADD event_id VARCHAR2(80);
+ALTER TABLE alarm_visit ADD result_type VARCHAR2(64);
+
+COMMENT ON COLUMN alarm_visit.algorithm_id IS '生成结果的算法标识';
+COMMENT ON COLUMN alarm_visit.algorithm_version IS '生成结果的算法版本';
+COMMENT ON COLUMN alarm_visit.event_id IS '算法输入源事件的稳定 event_id';
+COMMENT ON COLUMN alarm_visit.result_type IS '算法返回结果类型';
+
+-- 3. 规则/算法窗口状态和多实例租约。
+CREATE TABLE analysis_checkpoint (
+  id VARCHAR2(36) NOT NULL,
+  task_type VARCHAR2(32) NOT NULL,
+  task_id VARCHAR2(128) NOT NULL,
+  task_version VARCHAR2(64) NOT NULL,
+  window_start TIMESTAMP NOT NULL,
+  window_end TIMESTAMP NOT NULL,
+  status VARCHAR2(32) NOT NULL,
+  attempt_count INT DEFAULT 0 NOT NULL,
+  input_count BIGINT DEFAULT 0 NOT NULL,
+  output_count BIGINT DEFAULT 0 NOT NULL,
+  error_code VARCHAR2(128),
+  error_message VARCHAR2(2000),
+  owner_id VARCHAR2(255),
+  fencing_token BIGINT DEFAULT 0 NOT NULL,
+  lease_until TIMESTAMP,
+  started_at TIMESTAMP,
+  completed_at TIMESTAMP,
+  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_analysis_checkpoint PRIMARY KEY(id),
+  CONSTRAINT uk_analysis_checkpoint_window
+    UNIQUE(task_type,task_id,task_version,window_start,window_end),
+  CONSTRAINT ck_analysis_checkpoint_status
+    CHECK(status IN ('PENDING','RUNNING','COMPLETED','FAILED','RETRYING'))
+);
+
+COMMENT ON TABLE analysis_checkpoint IS '规则和算法独立窗口状态、连续成功水位与多实例租约';
+
+CREATE INDEX idx_analysis_checkpoint_due
+ON analysis_checkpoint(task_type,status,lease_until,window_end);
+
+-- 4. 当前最终 DIRECT 统计正式表。
+-- 旧备份基线没有中间版 change-log 对象，本升级直接建立最终小型聚合模型。
+CREATE TABLE device_daily_stats (
+  stat_date DATE NOT NULL,
+  device_id BIGINT NOT NULL,
+  parse_count BIGINT DEFAULT 0 NOT NULL,
+  non_log_count BIGINT DEFAULT 0 NOT NULL,
+  last_receive_time TIMESTAMP,
+  projection_seq BIGINT DEFAULT 0 NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_device_daily_stats PRIMARY KEY(stat_date,device_id),
+  CONSTRAINT ck_device_daily_stats_count CHECK(parse_count>=0 AND non_log_count>=0)
+);
+
+CREATE TABLE normalize_rule_daily_stats (
+  stat_date DATE NOT NULL,
+  normalize_rule_id BIGINT NOT NULL,
+  max_log_time TIMESTAMP,
+  projection_seq BIGINT DEFAULT 0 NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_normalize_rule_daily_stats PRIMARY KEY(stat_date,normalize_rule_id)
+);
+
+CREATE TABLE device_collect_stats (
+  device_collect_id BIGINT NOT NULL,
+  first_success_time TIMESTAMP,
+  last_success_time TIMESTAMP,
+  last_fail_time TIMESTAMP,
+  projection_seq BIGINT DEFAULT 0 NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_device_collect_stats PRIMARY KEY(device_collect_id)
+);
+
+COMMENT ON TABLE device_daily_stats IS 'DIRECT 按业务日期和设备维护的物理行计数及入口时间';
+COMMENT ON TABLE normalize_rule_daily_stats IS 'DIRECT 按业务日期和规则维护的最大日志时间';
+COMMENT ON TABLE device_collect_stats IS 'DIRECT 维护采集任务首次成功及最后成功/失败时间';
+
+-- 5. DIRECT 短期影子表和人工重建状态。
+CREATE TABLE device_daily_stats_direct_shadow (
+  stat_date DATE NOT NULL,
+  device_id BIGINT NOT NULL,
+  parse_count BIGINT DEFAULT 0 NOT NULL,
+  non_log_count BIGINT DEFAULT 0 NOT NULL,
+  last_receive_time TIMESTAMP,
+  projection_seq BIGINT DEFAULT 0 NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_device_daily_direct_shadow PRIMARY KEY(stat_date,device_id),
+  CONSTRAINT ck_device_daily_direct_count CHECK(parse_count>=0 AND non_log_count>=0)
+);
+
+CREATE TABLE normalize_rule_daily_stats_direct_shadow (
+  stat_date DATE NOT NULL,
+  normalize_rule_id BIGINT NOT NULL,
+  max_log_time TIMESTAMP,
+  projection_seq BIGINT DEFAULT 0 NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_rule_daily_direct_shadow PRIMARY KEY(stat_date,normalize_rule_id)
+);
+
+CREATE TABLE stats_direct_control (
+  control_name VARCHAR2(64) NOT NULL,
+  rebuild_required SMALLINT DEFAULT 0 NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_stats_direct_control PRIMARY KEY(control_name),
+  CONSTRAINT ck_stats_direct_rebuild CHECK(rebuild_required IN (0,1))
+);
+
+INSERT INTO stats_direct_control(control_name,rebuild_required,updated_at)
+VALUES ('SYSLOG_STATS',0,SYSDATE);
+
+COMMENT ON TABLE device_daily_stats_direct_shadow IS 'DIRECT 短期对账设备影子表，正式 DIRECT 不读取';
+COMMENT ON TABLE normalize_rule_daily_stats_direct_shadow IS 'DIRECT 短期对账规则影子表，正式 DIRECT 不读取';
+COMMENT ON TABLE stats_direct_control IS 'DIRECT 人工降级和重建状态';
+
+-- 6. 今日计数日切保护字段。
+ALTER TABLE device_device ADD today_stats_date DATE;
+
+COMMENT ON COLUMN device_device.today_stats_date IS 'today_parse_count/today_non_log_count 对应的上海业务日期';
+
+COMMIT;
@@ -0,0 +1,35 @@
+-- haobang-security-dm 幂等索引升级。
+-- 前置：01-schema-upgrade.sql 成功，所有相关旧应用仍保持停止，DBA 已评估索引空间和维护窗口。
+-- 任一 CREATE 报错立即停止；禁止通过删除历史数据绕过重复问题。
+
+-- 两张标准化表现场均按 LOG_TIME 做 RANGE 分区。
+-- 当前 DM8 现场语法在不指定 GLOBAL 时建立分区索引；唯一键包含完整分区键，因此形成本地唯一索引。
+-- 新版 id 是 event_id + normalize_rule_id + rule_version 的确定性 UUID，长度仍为 36 字符。
+CREATE UNIQUE INDEX uk_normal_data_log_id
+ON syslog_normal_data(log_time,id);
+
+CREATE UNIQUE INDEX uk_normal_alarm_log_id
+ON syslog_normal_alarm(log_time,id);
+
+-- 规则聚合窗口和算法单事件结果需要跨日幂等，使用全局非分区唯一索引。
+-- 历史新增字段保持 NULL，索引只保护新版本写入且幂等字段完整的数据。
+CREATE UNIQUE INDEX uk_alarm_rule_window_group
+ON alarm(rule_id,rule_version,window_start,group_key_hash) GLOBAL;
+
+CREATE UNIQUE INDEX uk_alarm_visit_algorithm_event
+ON alarm_visit(algorithm_id,algorithm_version,event_id,result_type) GLOBAL;
+
+-- 结果门禁：前两个必须 UNIQUE/PARTITIONED=YES/VALID；后两个必须 UNIQUE/PARTITIONED=NO/VALID。
+SELECT index_name,table_name,uniqueness,partitioned,status
+FROM user_indexes
+WHERE index_name IN (
+  'UK_NORMAL_DATA_LOG_ID','UK_NORMAL_ALARM_LOG_ID',
+  'UK_ALARM_RULE_WINDOW_GROUP','UK_ALARM_VISIT_ALGORITHM_EVENT'
+)
+ORDER BY table_name,index_name;
+
+-- 旧全局宽字符串索引不属于目标结构。若现场已存在，立即停止并单独评审，禁止在本文件中自动 DROP。
+SELECT index_name,table_name,uniqueness,partitioned,status
+FROM user_indexes
+WHERE index_name IN ('UK_NORMAL_DATA_EVENT_RULE','UK_NORMAL_ALARM_EVENT_RULE')
+ORDER BY table_name,index_name;
@@ -0,0 +1,142 @@
+-- DIRECT 统计初始化（仅执行一次）。
+-- 执行前：停止 syslog-consumer-rule、syslog-consumer，完成 01、02；
+-- 将 BUSINESS_DATE 替换为升级日（如 2026-08-24），YYYYMMDD 替换为 20260824。
+-- 任一语句报错立即停止并 ROLLBACK；全部成功后再 COMMIT。
+
+ROLLBACK;
+SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+
+-- 1. 初始化当天设备统计。
+INSERT INTO device_daily_stats(
+  stat_date,device_id,parse_count,non_log_count,
+  last_receive_time,projection_seq,updated_at
+)
+SELECT TO_DATE('BUSINESS_DATE','YYYY-MM-DD'),device_id,
+       SUM(parse_count),SUM(non_log_count),MAX(last_receive_time),0,SYSDATE
+FROM (
+  SELECT device_id,COUNT(*) parse_count,0 non_log_count,
+         CAST(NULL AS TIMESTAMP) last_receive_time
+  FROM syslog_normal_data_p_YYYYMMDD
+  WHERE created_at>=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+    AND created_at<TO_DATE('BUSINESS_DATE','YYYY-MM-DD')+1
+    AND device_id IS NOT NULL
+  GROUP BY device_id
+
+  UNION ALL
+
+  SELECT device_id,COUNT(*),0,CAST(NULL AS TIMESTAMP)
+  FROM syslog_normal_alarm_p_YYYYMMDD
+  WHERE created_at>=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+    AND created_at<TO_DATE('BUSINESS_DATE','YYYY-MM-DD')+1
+    AND device_id IS NOT NULL
+  GROUP BY device_id
+
+  UNION ALL
+
+  SELECT device_id,0,COUNT(*),CAST(NULL AS TIMESTAMP)
+  FROM syslog_non_normal_message_p_YYYYMMDD
+  WHERE created_at>=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+    AND created_at<TO_DATE('BUSINESS_DATE','YYYY-MM-DD')+1
+    AND device_id IS NOT NULL
+  GROUP BY device_id
+
+  UNION ALL
+
+  SELECT id,0,0,last_receive_time
+  FROM device_device
+  WHERE id IS NOT NULL
+    AND last_receive_time>=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+    AND last_receive_time<TO_DATE('BUSINESS_DATE','YYYY-MM-DD')+1
+) facts
+GROUP BY device_id;
+
+-- 2. 初始化当天规则统计。
+INSERT INTO normalize_rule_daily_stats(
+  stat_date,normalize_rule_id,max_log_time,projection_seq,updated_at
+)
+SELECT TO_DATE('BUSINESS_DATE','YYYY-MM-DD'),normalize_rule_id,
+       MAX(log_time),0,SYSDATE
+FROM (
+  SELECT normalize_rule_id,log_time
+  FROM syslog_normal_data_p_YYYYMMDD
+  WHERE created_at>=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+    AND created_at<TO_DATE('BUSINESS_DATE','YYYY-MM-DD')+1
+    AND normalize_rule_id IS NOT NULL
+
+  UNION ALL
+
+  SELECT normalize_rule_id,log_time
+  FROM syslog_normal_alarm_p_YYYYMMDD
+  WHERE created_at>=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+    AND created_at<TO_DATE('BUSINESS_DATE','YYYY-MM-DD')+1
+    AND normalize_rule_id IS NOT NULL
+) facts
+GROUP BY normalize_rule_id;
+
+-- 3. 初始化采集任务统计。
+INSERT INTO device_collect_stats(
+  device_collect_id,first_success_time,last_success_time,
+  last_fail_time,projection_seq,updated_at
+)
+SELECT id,first_time,last_success_time,last_failed_time,0,SYSDATE
+FROM device_collect_task;
+
+-- 4. 回填设备今日计数。
+UPDATE device_device
+SET today_parse_count=0,
+    today_non_log_count=0,
+    today_stats_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD'),
+    updated_at=SYSDATE;
+
+UPDATE device_device d
+SET today_parse_count=(
+      SELECT s.parse_count
+      FROM device_daily_stats s
+      WHERE s.stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+        AND s.device_id=d.id
+    ),
+    today_non_log_count=(
+      SELECT s.non_log_count
+      FROM device_daily_stats s
+      WHERE s.stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+        AND s.device_id=d.id
+    ),
+    updated_at=SYSDATE
+WHERE EXISTS (
+  SELECT 1
+  FROM device_daily_stats s
+  WHERE s.stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+    AND s.device_id=d.id
+);
+
+-- 5. 回填规则最后命中时间，只允许时间向前推进。
+UPDATE dm_normalize_rule r
+SET first_data_saved_at=CASE
+      WHEN first_data_saved_at IS NULL OR first_data_saved_at<(
+        SELECT s.max_log_time
+        FROM normalize_rule_daily_stats s
+        WHERE s.stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+          AND s.normalize_rule_id=r.id
+      ) THEN (
+        SELECT s.max_log_time
+        FROM normalize_rule_daily_stats s
+        WHERE s.stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+          AND s.normalize_rule_id=r.id
+      )
+      ELSE first_data_saved_at
+    END,
+    updated_at=SYSDATE
+WHERE r.del_flag='0'
+  AND EXISTS (
+    SELECT 1
+    FROM normalize_rule_daily_stats s
+    WHERE s.stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+      AND s.normalize_rule_id=r.id
+      AND s.max_log_time IS NOT NULL
+  );
+
+UPDATE stats_direct_control
+SET rebuild_required=0,updated_at=SYSDATE
+WHERE control_name='SYSLOG_STATS';
+
+COMMIT;
@@ -0,0 +1,73 @@
+-- 将 BUSINESS_DATE、YYYYMMDD 替换为升级当天后执行。
+-- 结果要求见每段注释；不符合时不要启动全部业务实例。
+
+-- 1. 应返回 10 行。
+SELECT table_name
+FROM user_tables
+WHERE table_name IN (
+  'SYSLOG_FAILURE_RECORD','SYSLOG_REPLAY_BATCH','SYSLOG_REPLAY_ITEM',
+  'ANALYSIS_CHECKPOINT','DEVICE_DAILY_STATS','NORMALIZE_RULE_DAILY_STATS',
+  'DEVICE_COLLECT_STATS','DEVICE_DAILY_STATS_DIRECT_SHADOW',
+  'NORMALIZE_RULE_DAILY_STATS_DIRECT_SHADOW','STATS_DIRECT_CONTROL'
+)
+ORDER BY table_name;
+
+-- 2. POINT_TIME_NS 必须存在；其余升级字段也应全部列出。
+SELECT table_name,column_name
+FROM user_tab_columns
+WHERE (table_name='SYSLOG_FAILURE_RECORD' AND column_name='POINT_TIME_NS')
+   OR (table_name IN ('SYSLOG_NORMAL_DATA','SYSLOG_NORMAL_ALARM') AND column_name='RULE_VERSION')
+   OR (table_name='ALARM' AND column_name IN ('RULE_ID','RULE_VERSION','WINDOW_START','GROUP_KEY_HASH'))
+   OR (table_name='ALARM_VISIT' AND column_name IN ('ALGORITHM_ID','ALGORITHM_VERSION','EVENT_ID','RESULT_TYPE'))
+   OR (table_name='DEVICE_DEVICE' AND column_name='TODAY_STATS_DATE')
+ORDER BY table_name,column_name;
+
+-- 3. 应返回 4 行且全部为 UNIQUE、VALID；前两个 PARTITIONED=YES，后两个为 NO。
+SELECT index_name,table_name,uniqueness,partitioned,status
+FROM user_indexes
+WHERE index_name IN (
+  'UK_NORMAL_DATA_LOG_ID','UK_NORMAL_ALARM_LOG_ID',
+  'UK_ALARM_RULE_WINDOW_GROUP','UK_ALARM_VISIT_ALGORITHM_EVENT'
+)
+ORDER BY index_name;
+
+-- 4. 三个物理事实表及三张统计表行数，用于现场留档。
+SELECT 'SYSLOG_NORMAL_DATA_P_YYYYMMDD' AS item,COUNT(*) AS row_count
+FROM syslog_normal_data_p_YYYYMMDD
+UNION ALL
+SELECT 'SYSLOG_NORMAL_ALARM_P_YYYYMMDD',COUNT(*)
+FROM syslog_normal_alarm_p_YYYYMMDD
+UNION ALL
+SELECT 'SYSLOG_NON_NORMAL_MESSAGE_P_YYYYMMDD',COUNT(*)
+FROM syslog_non_normal_message_p_YYYYMMDD
+UNION ALL
+SELECT 'DEVICE_DAILY_STATS',COUNT(*)
+FROM device_daily_stats
+WHERE stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+UNION ALL
+SELECT 'NORMALIZE_RULE_DAILY_STATS',COUNT(*)
+FROM normalize_rule_daily_stats
+WHERE stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+UNION ALL
+SELECT 'DEVICE_COLLECT_STATS',COUNT(*)
+FROM device_collect_stats;
+
+-- 5. 必须返回 SYSLOG_STATS / 0。
+SELECT control_name,rebuild_required,updated_at
+FROM stats_direct_control
+WHERE control_name='SYSLOG_STATS';
+
+-- 6. 以下两个查询都必须返回 0 行。
+SELECT d.id
+FROM device_device d
+LEFT JOIN device_daily_stats s
+  ON s.device_id=d.id
+ AND s.stat_date=TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+WHERE d.today_stats_date IS NULL
+   OR d.today_stats_date<>TO_DATE('BUSINESS_DATE','YYYY-MM-DD')
+   OR COALESCE(d.today_parse_count,0)<>COALESCE(s.parse_count,0)
+   OR COALESCE(d.today_non_log_count,0)<>COALESCE(s.non_log_count,0);
+
+SELECT control_name
+FROM stats_direct_control
+WHERE control_name='SYSLOG_STATS' AND rebuild_required<>0;
@@ -0,0 +1,65 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+: "${KAFKA_BIN:?请设置 KAFKA_BIN，例如 /opt/kafka/bin}"
+: "${BOOTSTRAP:?请设置 BOOTSTRAP，例如 127.0.0.1:9092}"
+
+RAW_TOPIC="${RAW_TOPIC:-agent-01-syslog-topic}"
+TOPICS_CMD="${KAFKA_BIN}/kafka-topics.sh"
+CONFIGS_CMD="${KAFKA_BIN}/kafka-configs.sh"
+
+if [[ ! -x "${TOPICS_CMD}" ]] || [[ ! -x "${CONFIGS_CMD}" ]]; then
+  echo "ERROR: ${KAFKA_BIN} 下缺少 kafka-topics.sh 或 kafka-configs.sh" >&2
+  exit 1
+fi
+
+COMMON_ARGS=(--bootstrap-server "${BOOTSTRAP}")
+if [[ -n "${COMMAND_CONFIG:-}" ]]; then
+  COMMON_ARGS+=(--command-config "${COMMAND_CONFIG}")
+fi
+
+echo "检查原始 Topic：${RAW_TOPIC}"
+RAW_INFO=$("${TOPICS_CMD}" "${COMMON_ARGS[@]}" --describe --topic "${RAW_TOPIC}")
+printf '%s\n' "${RAW_INFO}"
+
+PARTITIONS=$(printf '%s\n' "${RAW_INFO}" | awk '
+  /PartitionCount:/ {
+    for (i=1; i<=NF; i++) if ($i=="PartitionCount:") { print $(i+1); exit }
+  }
+')
+REPLICAS=$(printf '%s\n' "${RAW_INFO}" | awk '
+  /ReplicationFactor:/ {
+    for (i=1; i<=NF; i++) if ($i=="ReplicationFactor:") { print $(i+1); exit }
+  }
+')
+
+if [[ ! "${PARTITIONS}" =~ ^[1-9][0-9]*$ ]] || [[ ! "${REPLICAS}" =~ ^[1-9][0-9]*$ ]]; then
+  echo "ERROR: 无法读取原始 Topic 的分区数或副本数" >&2
+  exit 1
+fi
+
+SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY")
+RETENTIONS=(259200000 604800000 604800000 7776000000 604800000)
+
+for i in 0 1 2 3 4; do
+  TOPIC="${RAW_TOPIC}.${SUFFIXES[$i]}"
+  echo "创建或确认 Topic：${TOPIC}"
+  "${TOPICS_CMD}" "${COMMON_ARGS[@]}" \
+    --create --if-not-exists \
+    --topic "${TOPIC}" \
+    --partitions "${PARTITIONS}" \
+    --replication-factor "${REPLICAS}" \
+    --config "retention.ms=${RETENTIONS[$i]}"
+
+  "${CONFIGS_CMD}" "${COMMON_ARGS[@]}" \
+    --alter --entity-type topics --entity-name "${TOPIC}" \
+    --add-config "retention.ms=${RETENTIONS[$i]}"
+done
+
+echo "检查升级结果"
+for SUFFIX in "${SUFFIXES[@]}"; do
+  "${TOPICS_CMD}" "${COMMON_ARGS[@]}" --describe --topic "${RAW_TOPIC}.${SUFFIX}"
+done
+
+echo "PASS: Kafka Topic 已创建/确认；脚本未删除 Topic，也未修改消费组 offset。"