From f53661931ff1d28113701c143fc70fcef83d46f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Mon, 24 Aug 2026 15:09:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(upgrade):=20=E5=A2=9E=E5=8A=A0=20=E5=8D=87?= =?UTF-8?q?=E7=BA=A7=E8=84=9A=E6=9C=AC=E5=92=8C=E9=83=A8=E7=BD=B2=E8=AF=B4?= =?UTF-8?q?=E6=98=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增达梦数据库结构、幂等索引、统计基线及校验脚本 - 新增 Docker Kafka Retry、DLT、Replay Topic 升级脚本 - 更新生产 Compose，启用 DIRECT 并关闭旧增量捕获和投影 - 补充数据库、Kafka、应用启动及验证说明 - 增加日志和本地生成文件忽略规则 --- .../syslog-client/logs/syslog-client.log | 18 +- .../docker-compose-consumer-rule-prod.yaml | 9 +- .../com/influx/raw/RawStorageProperties.java | 4 +- .../src/main/resources/application.properties | 4 +- .../influx/raw/InfluxRawV2ContractTest.java | 12 +- .../docker-compose-consumer-prod.yaml | 11 +- .../com/influx/raw/RawStorageProperties.java | 4 +- .../src/main/resources/application.properties | 4 +- .../influx/raw/InfluxRawV2ContractTest.java | 12 +- haobang-security-dm/upgrade-doc/README.md | 115 +++++++++ .../database/01-schema-upgrade.sql | 224 ++++++++++++++++++ .../database/02-idempotency-indexes.sql | 35 +++ .../database/03-direct-stats-baseline.sql | 142 +++++++++++ .../upgrade-doc/database/04-verify.sql | 73 ++++++ .../upgrade-doc/kafka/kafka-upgrade.sh | 65 +++++ 15 files changed, 694 insertions(+), 38 deletions(-) create mode 100644 haobang-security-dm/upgrade-doc/README.md create mode 100644 haobang-security-dm/upgrade-doc/database/01-schema-upgrade.sql create mode 100644 haobang-security-dm/upgrade-doc/database/02-idempotency-indexes.sql create mode 100644 haobang-security-dm/upgrade-doc/database/03-direct-stats-baseline.sql create mode 100644 haobang-security-dm/upgrade-doc/database/04-verify.sql create mode 100755 haobang-security-dm/upgrade-doc/kafka/kafka-upgrade.sh diff --git a/haobang-security-dm/syslog-client/logs/syslog-client.log b/haobang-security-dm/syslog-client/logs/syslog-client.log index 0748e83..96c20e4 100644 --- a/haobang-security-dm/syslog-client/logs/syslog-client.log +++ b/haobang-security-dm/syslog-client/logs/syslog-client.log @@ -1,9 +1,9 @@ -2026-08-15 04:48:58.158 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Neither @ContextConfiguration nor @ContextHierarchy found for test class [com.haobang.syslog.SysjavacollectApplicationTests], using SpringBootContextLoader -2026-08-15 04:48:58.162 [main] INFO o.s.t.c.s.AbstractContextLoader - Could not detect default resource locations for test class [com.haobang.syslog.SysjavacollectApplicationTests]: no resource found for suffixes {-context.xml, Context.groovy}. -2026-08-15 04:48:58.162 [main] INFO o.s.t.c.s.AnnotationConfigContextLoaderUtils - Could not detect default configuration classes for test class [com.haobang.syslog.SysjavacollectApplicationTests]: SysjavacollectApplicationTests does not declare any static, non-private, non-final, nested classes annotated with @Configuration. -2026-08-15 04:48:58.225 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Found @SpringBootConfiguration com.haobang.syslog.syslogClientApplication for test class com.haobang.syslog.SysjavacollectApplicationTests -2026-08-15 04:48:58.288 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Loaded default TestExecutionListener class names from location [META-INF/spring.factories]: [org.springframework.boot.test.mock.mockito.MockitoTestExecutionListener, org.springframework.boot.test.mock.mockito.ResetMocksTestExecutionListener, org.springframework.boot.test.autoconfigure.restdocs.RestDocsTestExecutionListener, org.springframework.boot.test.autoconfigure.web.client.MockRestServiceServerResetTestExecutionListener, org.springframework.boot.test.autoconfigure.web.servlet.MockMvcPrintOnlyOnFailureTestExecutionListener, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverTestExecutionListener, org.springframework.boot.test.autoconfigure.webservices.client.MockWebServiceServerTestExecutionListener, org.springframework.test.context.web.ServletTestExecutionListener, org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener, org.springframework.test.context.event.ApplicationEventsTestExecutionListener, org.springframework.test.context.support.DependencyInjectionTestExecutionListener, org.springframework.test.context.support.DirtiesContextTestExecutionListener, org.springframework.test.context.transaction.TransactionalTestExecutionListener, org.springframework.test.context.jdbc.SqlScriptsTestExecutionListener, org.springframework.test.context.event.EventPublishingTestExecutionListener] -2026-08-15 04:48:58.294 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Using TestExecutionListeners: [org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener@22f59fa, org.springframework.test.context.event.ApplicationEventsTestExecutionListener@78fa769e, org.springframework.boot.test.mock.mockito.MockitoTestExecutionListener@16612a51, org.springframework.boot.test.autoconfigure.SpringBootDependencyInjectionTestExecutionListener@54e041a4, org.springframework.test.context.support.DirtiesContextTestExecutionListener@2c78324b, org.springframework.test.context.event.EventPublishingTestExecutionListener@79defdc, org.springframework.boot.test.mock.mockito.ResetMocksTestExecutionListener@18317edc, org.springframework.boot.test.autoconfigure.restdocs.RestDocsTestExecutionListener@4e0ae11f, org.springframework.boot.test.autoconfigure.web.client.MockRestServiceServerResetTestExecutionListener@238d68ff, org.springframework.boot.test.autoconfigure.web.servlet.MockMvcPrintOnlyOnFailureTestExecutionListener@4b86805d, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverTestExecutionListener@5852c06f, org.springframework.boot.test.autoconfigure.webservices.client.MockWebServiceServerTestExecutionListener@4149c063] -2026-08-15 04:48:58.433 [main] INFO c.h.s.SysjavacollectApplicationTests - Starting SysjavacollectApplicationTests using Java 1.8.0_392 on wangshaohongdeMacBook-Pro.local with PID 34847 (started by wangshaohong in /Users/wangshaohong/WorkSpace/fufeng/ai-security-xdr/haobang-security-dm/syslog-client) -2026-08-15 04:48:58.434 [main] INFO c.h.s.SysjavacollectApplicationTests - No active profile set, falling back to 1 default profile: "default" -2026-08-15 04:48:58.712 [main] INFO c.h.s.SysjavacollectApplicationTests - Started SysjavacollectApplicationTests in 0.397 seconds (JVM running for 0.904) +2026-08-20 00:07:55.926 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Neither @ContextConfiguration nor @ContextHierarchy found for test class [com.haobang.syslog.SysjavacollectApplicationTests], using SpringBootContextLoader +2026-08-20 00:07:55.931 [main] INFO o.s.t.c.s.AbstractContextLoader - Could not detect default resource locations for test class [com.haobang.syslog.SysjavacollectApplicationTests]: no resource found for suffixes {-context.xml, Context.groovy}. +2026-08-20 00:07:55.931 [main] INFO o.s.t.c.s.AnnotationConfigContextLoaderUtils - Could not detect default configuration classes for test class [com.haobang.syslog.SysjavacollectApplicationTests]: SysjavacollectApplicationTests does not declare any static, non-private, non-final, nested classes annotated with @Configuration. +2026-08-20 00:07:56.000 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Found @SpringBootConfiguration com.haobang.syslog.syslogClientApplication for test class com.haobang.syslog.SysjavacollectApplicationTests +2026-08-20 00:07:56.089 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Loaded default TestExecutionListener class names from location [META-INF/spring.factories]: [org.springframework.boot.test.mock.mockito.MockitoTestExecutionListener, org.springframework.boot.test.mock.mockito.ResetMocksTestExecutionListener, org.springframework.boot.test.autoconfigure.restdocs.RestDocsTestExecutionListener, org.springframework.boot.test.autoconfigure.web.client.MockRestServiceServerResetTestExecutionListener, org.springframework.boot.test.autoconfigure.web.servlet.MockMvcPrintOnlyOnFailureTestExecutionListener, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverTestExecutionListener, org.springframework.boot.test.autoconfigure.webservices.client.MockWebServiceServerTestExecutionListener, org.springframework.test.context.web.ServletTestExecutionListener, org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener, org.springframework.test.context.event.ApplicationEventsTestExecutionListener, org.springframework.test.context.support.DependencyInjectionTestExecutionListener, org.springframework.test.context.support.DirtiesContextTestExecutionListener, org.springframework.test.context.transaction.TransactionalTestExecutionListener, org.springframework.test.context.jdbc.SqlScriptsTestExecutionListener, org.springframework.test.context.event.EventPublishingTestExecutionListener] +2026-08-20 00:07:56.098 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Using TestExecutionListeners: [org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener@7fd7a283, org.springframework.test.context.event.ApplicationEventsTestExecutionListener@22f59fa, org.springframework.boot.test.mock.mockito.MockitoTestExecutionListener@78fa769e, org.springframework.boot.test.autoconfigure.SpringBootDependencyInjectionTestExecutionListener@16612a51, org.springframework.test.context.support.DirtiesContextTestExecutionListener@54e041a4, org.springframework.test.context.event.EventPublishingTestExecutionListener@2c78324b, org.springframework.boot.test.mock.mockito.ResetMocksTestExecutionListener@79defdc, org.springframework.boot.test.autoconfigure.restdocs.RestDocsTestExecutionListener@18317edc, org.springframework.boot.test.autoconfigure.web.client.MockRestServiceServerResetTestExecutionListener@4e0ae11f, org.springframework.boot.test.autoconfigure.web.servlet.MockMvcPrintOnlyOnFailureTestExecutionListener@238d68ff, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverTestExecutionListener@4b86805d, org.springframework.boot.test.autoconfigure.webservices.client.MockWebServiceServerTestExecutionListener@5852c06f] +2026-08-20 00:07:56.280 [main] INFO c.h.s.SysjavacollectApplicationTests - Starting SysjavacollectApplicationTests using Java 1.8.0_392 on wangshaohongdeMacBook-Pro.local with PID 23533 (started by wangshaohong in /Users/wangshaohong/WorkSpace/fufeng/ai-security-xdr-orig/ai-security-xdr/haobang-security-dm/syslog-client) +2026-08-20 00:07:56.280 [main] INFO c.h.s.SysjavacollectApplicationTests - No active profile set, falling back to 1 default profile: "default" +2026-08-20 00:07:56.613 [main] INFO c.h.s.SysjavacollectApplicationTests - Started SysjavacollectApplicationTests in 0.489 seconds (JVM running for 1.098) diff --git a/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml b/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml index c6bcc3f..be90849 100644 --- a/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml +++ b/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml @@ -20,8 +20,9 @@ services: # 环境配置 - SPRING_PROFILES_ACTIVE=prod - TZ=Asia/Shanghai - # 全停升级且已完成增量统计 DDL/基线后，所有 rule 节点统一开启捕获。 - - STATS_INCREMENTAL_CAPTURE_ENABLED=true + # 生产环境只使用 DIRECT：事实与统计同事务写入，不再写旧 change-log/pending。 + - STATS_WRITE_MODE=DIRECT + - STATS_INCREMENTAL_CAPTURE_ENABLED=false - STATS_INGRESS_METADATA_ENABLED=true - STATS_INCREMENTAL_READ_ENABLED=true - STATS_DEVICE_RECEIVE_CHANGE_CAPTURE_ENABLED=false @@ -71,9 +72,9 @@ services: - INFLUXDB_ORG=influxdb - INFLUXDB_BUCKET=yelangbucket - INFLUXDB_RAW_WRITE_SCHEMA=V2 - - INFLUXDB_RAW_READ_MODE=V1_V2 + - INFLUXDB_RAW_READ_MODE=V2_ONLY - INFLUXDB_RAW_V1_BUCKET=yelangbucket - - INFLUXDB_RAW_V2_BUCKET=yelangbucket_v2 + - INFLUXDB_RAW_V2_BUCKET=yelangbucket - INFLUXDB_RAW_UUID_BUCKET_COUNT=4096 - INFLUXDB_RAW_LEGACY_POINT_TIME_FALLBACK_ENABLED=true - INFLUXDB_RAW_QUERY_MAX_ROWS=10000 diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawStorageProperties.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawStorageProperties.java index c2f1b8e..9ce5774 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawStorageProperties.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawStorageProperties.java @@ -12,9 +12,9 @@ public class RawStorageProperties { public static final int UUID_BUCKET_COUNT = 4096; @Value("${influxdb.raw.write-schema:V1}") private String writeSchemaValue; - @Value("${influxdb.raw.read-mode:V1_ONLY}") private String readModeValue; + @Value("${influxdb.raw.read-mode:V2_ONLY}") private String readModeValue; @Value("${influxdb.raw.v1-bucket:${influxdb.bucket:yelangbucket}}") private String v1Bucket; - @Value("${influxdb.raw.v2-bucket:yelangbucket_v2}") private String v2Bucket; + @Value("${influxdb.raw.v2-bucket:yelangbucket}") private String v2Bucket; @Value("${influxdb.raw.uuid-bucket-count:4096}") private int uuidBucketCount; @Value("${influxdb.raw.legacy-point-time-fallback-enabled:true}") private boolean legacyFallbackEnabled; @Value("${influxdb.raw.query-max-rows:10000}") private int queryMaxRows; diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties index c05f5c8..6d14542 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties @@ -27,9 +27,9 @@ influxdb.retry.attempts=3 influxdb.retry.delay=1000 # 原始 Syslog V2 分阶段开关。兼容部署先保持 V1/V1_ONLY，切换时统一写 V2。 influxdb.raw.write-schema=${INFLUXDB_RAW_WRITE_SCHEMA:V2} -influxdb.raw.read-mode=${INFLUXDB_RAW_READ_MODE:V1_V2} +influxdb.raw.read-mode=${INFLUXDB_RAW_READ_MODE:V2_ONLY} influxdb.raw.v1-bucket=${INFLUXDB_RAW_V1_BUCKET:${influxdb.bucket}} -influxdb.raw.v2-bucket=${INFLUXDB_RAW_V2_BUCKET:yelangbucket_v2} +influxdb.raw.v2-bucket=${INFLUXDB_RAW_V2_BUCKET:yelangbucket} influxdb.raw.uuid-bucket-count=${INFLUXDB_RAW_UUID_BUCKET_COUNT:4096} influxdb.raw.legacy-point-time-fallback-enabled=${INFLUXDB_RAW_LEGACY_POINT_TIME_FALLBACK_ENABLED:true} influxdb.raw.query-max-rows=${INFLUXDB_RAW_QUERY_MAX_ROWS:10000} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java index 4bdce65..674e4c0 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java @@ -29,9 +29,9 @@ class InfluxRawV2ContractTest { @Test void configurationRejectsChangingPersistentBucketCount() { RawStorageProperties properties=new RawStorageProperties(); ReflectionTestUtils.setField(properties,"writeSchemaValue","V2"); - ReflectionTestUtils.setField(properties,"readModeValue","V1_V2"); + ReflectionTestUtils.setField(properties,"readModeValue","V2_ONLY"); ReflectionTestUtils.setField(properties,"v1Bucket","yelangbucket"); - ReflectionTestUtils.setField(properties,"v2Bucket","yelangbucket_v2"); + ReflectionTestUtils.setField(properties,"v2Bucket","yelangbucket"); ReflectionTestUtils.setField(properties,"uuidBucketCount",1024); ReflectionTestUtils.setField(properties,"queryMaxRows",10000); ReflectionTestUtils.setField(properties,"queryMaxResponseBytes",10485760L); @@ -62,9 +62,9 @@ class InfluxRawV2ContractTest { } @Test void v2HasUuidFieldButNeverUuidTagAndUsesNanoseconds() { - RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket_v2"); + RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket"); InfluxRawPointFactory.RawPoint raw=new InfluxRawPointFactory(properties).create("20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5","message","20260814000000000","126","1","agent-01-syslog-topic",ReceiveTimeParser.parse("20260814000000000"),1_786_636_800_000_000_123L); - String line=raw.getPoint().toLineProtocol(); assertFalse(line.substring(0,line.indexOf(' ')).contains("uuid=")); assertTrue(line.contains("event_bucket=2085")); assertTrue(line.contains("uuid=\"20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5\"")); assertEquals(WritePrecision.NS,raw.getPoint().getPrecision()); assertEquals("yelangbucket_v2",raw.getBucket()); + String line=raw.getPoint().toLineProtocol(); assertFalse(line.substring(0,line.indexOf(' ')).contains("uuid=")); assertTrue(line.contains("event_bucket=2085")); assertTrue(line.contains("uuid=\"20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5\"")); assertEquals(WritePrecision.NS,raw.getPoint().getPrecision()); assertEquals("yelangbucket",raw.getBucket()); } @Test void onlyMissingPointTimeFallsBack() { @@ -83,7 +83,7 @@ class InfluxRawV2ContractTest { } @Test void mixedLegacyAndNewMessagesAllProduceDistinctV2PointsAndRetryIsIdempotent() { - RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.isLegacyFallbackEnabled()).thenReturn(true); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket_v2"); + RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.isLegacyFallbackEnabled()).thenReturn(true); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket"); RawInfluxMetrics metrics=new RawInfluxMetrics(); PointTimeResolver resolver=new PointTimeResolver(properties,metrics); InfluxRawPointFactory factory=new InfluxRawPointFactory(properties); ReceiveTimeParser.ParsedReceiveTime parsed=ReceiveTimeParser.parse("20260814000000000"); long first=resolver.resolve(false,null,parsed.getEpochMillis(),true); long second=resolver.resolve(false,null,parsed.getEpochMillis(),true); assertNotEquals(first,second); String id="20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5"; String originalTopic="agent-01-syslog-topic"; @@ -100,7 +100,7 @@ class InfluxRawV2ContractTest { @Test void uuidCandidateFluxComputesBucketEscapesUuidAndSelectsEarliestGlobally() { String uuid="20260814-a\" |> die() //"; - String query=InfluxDBClient.buildUuidV2CandidateQuery("yelangbucket_v2", + String query=InfluxDBClient.buildUuidV2CandidateQuery("yelangbucket", "syslog_security_20260814",uuid,Instant.parse("2026-08-13T16:00:00Z"), Instant.parse("2026-08-14T16:00:00Z")); assertTrue(query.contains("r.event_bucket == \""+InfluxEventBucket.bucketOf(uuid)+"\"")); diff --git a/haobang-security-dm/syslog-consumer/docker-compose/docker-compose-consumer-prod.yaml b/haobang-security-dm/syslog-consumer/docker-compose/docker-compose-consumer-prod.yaml index 7fdf895..3dfba92 100644 --- a/haobang-security-dm/syslog-consumer/docker-compose/docker-compose-consumer-prod.yaml +++ b/haobang-security-dm/syslog-consumer/docker-compose/docker-compose-consumer-prod.yaml @@ -21,9 +21,10 @@ services: # 环境配置 - SPRING_PROFILES_ACTIVE=prod - TZ=Asia/Shanghai - # 首次启动保持影子模式：维护增量投影，但正式字段仍由旧全量任务更新。 - - STATS_INCREMENTAL_CAPTURE_ENABLED=true - - STATS_INCREMENTAL_PROJECTION_ENABLED=true + # 生产环境只使用 DIRECT：事实与统计同事务写入，不再写旧 change-log/pending。 + - 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 @@ -77,9 +78,9 @@ services: - INFLUXDB_ORG=influxdb - INFLUXDB_BUCKET=yelangbucket - INFLUXDB_RAW_WRITE_SCHEMA=V2 - - INFLUXDB_RAW_READ_MODE=V1_V2 + - INFLUXDB_RAW_READ_MODE=V2_ONLY - INFLUXDB_RAW_V1_BUCKET=yelangbucket - - INFLUXDB_RAW_V2_BUCKET=yelangbucket_v2 + - INFLUXDB_RAW_V2_BUCKET=yelangbucket - INFLUXDB_RAW_UUID_BUCKET_COUNT=4096 - INFLUXDB_RAW_LEGACY_POINT_TIME_FALLBACK_ENABLED=true - INFLUXDB_RAW_QUERY_MAX_ROWS=10000 diff --git a/haobang-security-dm/syslog-consumer/src/main/java/com/influx/raw/RawStorageProperties.java b/haobang-security-dm/syslog-consumer/src/main/java/com/influx/raw/RawStorageProperties.java index c2f1b8e..9ce5774 100644 --- a/haobang-security-dm/syslog-consumer/src/main/java/com/influx/raw/RawStorageProperties.java +++ b/haobang-security-dm/syslog-consumer/src/main/java/com/influx/raw/RawStorageProperties.java @@ -12,9 +12,9 @@ public class RawStorageProperties { public static final int UUID_BUCKET_COUNT = 4096; @Value("${influxdb.raw.write-schema:V1}") private String writeSchemaValue; - @Value("${influxdb.raw.read-mode:V1_ONLY}") private String readModeValue; + @Value("${influxdb.raw.read-mode:V2_ONLY}") private String readModeValue; @Value("${influxdb.raw.v1-bucket:${influxdb.bucket:yelangbucket}}") private String v1Bucket; - @Value("${influxdb.raw.v2-bucket:yelangbucket_v2}") private String v2Bucket; + @Value("${influxdb.raw.v2-bucket:yelangbucket}") private String v2Bucket; @Value("${influxdb.raw.uuid-bucket-count:4096}") private int uuidBucketCount; @Value("${influxdb.raw.legacy-point-time-fallback-enabled:true}") private boolean legacyFallbackEnabled; @Value("${influxdb.raw.query-max-rows:10000}") private int queryMaxRows; diff --git a/haobang-security-dm/syslog-consumer/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer/src/main/resources/application.properties index 8d5163d..3b0ff7c 100644 --- a/haobang-security-dm/syslog-consumer/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-consumer/src/main/resources/application.properties @@ -28,9 +28,9 @@ influxdb.retry.attempts=3 influxdb.retry.delay=1000 # 原始 Syslog V2 分阶段开关。兼容部署先保持 V1/V1_ONLY，切换时统一写 V2。 influxdb.raw.write-schema=${INFLUXDB_RAW_WRITE_SCHEMA:V2} -influxdb.raw.read-mode=${INFLUXDB_RAW_READ_MODE:V1_V2} +influxdb.raw.read-mode=${INFLUXDB_RAW_READ_MODE:V2_ONLY} influxdb.raw.v1-bucket=${INFLUXDB_RAW_V1_BUCKET:${influxdb.bucket}} -influxdb.raw.v2-bucket=${INFLUXDB_RAW_V2_BUCKET:yelangbucket_v2} +influxdb.raw.v2-bucket=${INFLUXDB_RAW_V2_BUCKET:yelangbucket} influxdb.raw.uuid-bucket-count=${INFLUXDB_RAW_UUID_BUCKET_COUNT:4096} influxdb.raw.legacy-point-time-fallback-enabled=${INFLUXDB_RAW_LEGACY_POINT_TIME_FALLBACK_ENABLED:true} influxdb.raw.query-max-rows=${INFLUXDB_RAW_QUERY_MAX_ROWS:10000} diff --git a/haobang-security-dm/syslog-consumer/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java b/haobang-security-dm/syslog-consumer/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java index 4bdce65..674e4c0 100644 --- a/haobang-security-dm/syslog-consumer/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java +++ b/haobang-security-dm/syslog-consumer/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java @@ -29,9 +29,9 @@ class InfluxRawV2ContractTest { @Test void configurationRejectsChangingPersistentBucketCount() { RawStorageProperties properties=new RawStorageProperties(); ReflectionTestUtils.setField(properties,"writeSchemaValue","V2"); - ReflectionTestUtils.setField(properties,"readModeValue","V1_V2"); + ReflectionTestUtils.setField(properties,"readModeValue","V2_ONLY"); ReflectionTestUtils.setField(properties,"v1Bucket","yelangbucket"); - ReflectionTestUtils.setField(properties,"v2Bucket","yelangbucket_v2"); + ReflectionTestUtils.setField(properties,"v2Bucket","yelangbucket"); ReflectionTestUtils.setField(properties,"uuidBucketCount",1024); ReflectionTestUtils.setField(properties,"queryMaxRows",10000); ReflectionTestUtils.setField(properties,"queryMaxResponseBytes",10485760L); @@ -62,9 +62,9 @@ class InfluxRawV2ContractTest { } @Test void v2HasUuidFieldButNeverUuidTagAndUsesNanoseconds() { - RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket_v2"); + RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket"); InfluxRawPointFactory.RawPoint raw=new InfluxRawPointFactory(properties).create("20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5","message","20260814000000000","126","1","agent-01-syslog-topic",ReceiveTimeParser.parse("20260814000000000"),1_786_636_800_000_000_123L); - String line=raw.getPoint().toLineProtocol(); assertFalse(line.substring(0,line.indexOf(' ')).contains("uuid=")); assertTrue(line.contains("event_bucket=2085")); assertTrue(line.contains("uuid=\"20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5\"")); assertEquals(WritePrecision.NS,raw.getPoint().getPrecision()); assertEquals("yelangbucket_v2",raw.getBucket()); + String line=raw.getPoint().toLineProtocol(); assertFalse(line.substring(0,line.indexOf(' ')).contains("uuid=")); assertTrue(line.contains("event_bucket=2085")); assertTrue(line.contains("uuid=\"20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5\"")); assertEquals(WritePrecision.NS,raw.getPoint().getPrecision()); assertEquals("yelangbucket",raw.getBucket()); } @Test void onlyMissingPointTimeFallsBack() { @@ -83,7 +83,7 @@ class InfluxRawV2ContractTest { } @Test void mixedLegacyAndNewMessagesAllProduceDistinctV2PointsAndRetryIsIdempotent() { - RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.isLegacyFallbackEnabled()).thenReturn(true); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket_v2"); + RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.isLegacyFallbackEnabled()).thenReturn(true); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket"); RawInfluxMetrics metrics=new RawInfluxMetrics(); PointTimeResolver resolver=new PointTimeResolver(properties,metrics); InfluxRawPointFactory factory=new InfluxRawPointFactory(properties); ReceiveTimeParser.ParsedReceiveTime parsed=ReceiveTimeParser.parse("20260814000000000"); long first=resolver.resolve(false,null,parsed.getEpochMillis(),true); long second=resolver.resolve(false,null,parsed.getEpochMillis(),true); assertNotEquals(first,second); String id="20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5"; String originalTopic="agent-01-syslog-topic"; @@ -100,7 +100,7 @@ class InfluxRawV2ContractTest { @Test void uuidCandidateFluxComputesBucketEscapesUuidAndSelectsEarliestGlobally() { String uuid="20260814-a\" |> die() //"; - String query=InfluxDBClient.buildUuidV2CandidateQuery("yelangbucket_v2", + String query=InfluxDBClient.buildUuidV2CandidateQuery("yelangbucket", "syslog_security_20260814",uuid,Instant.parse("2026-08-13T16:00:00Z"), Instant.parse("2026-08-14T16:00:00Z")); assertTrue(query.contains("r.event_bucket == \""+InfluxEventBucket.bucketOf(uuid)+"\"")); diff --git a/haobang-security-dm/upgrade-doc/README.md b/haobang-security-dm/upgrade-doc/README.md new file mode 100644 index 0000000..3b77e02 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/README.md @@ -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。 diff --git a/haobang-security-dm/upgrade-doc/database/01-schema-upgrade.sql b/haobang-security-dm/upgrade-doc/database/01-schema-upgrade.sql new file mode 100644 index 0000000..fe48911 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/database/01-schema-upgrade.sql @@ -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; diff --git a/haobang-security-dm/upgrade-doc/database/02-idempotency-indexes.sql b/haobang-security-dm/upgrade-doc/database/02-idempotency-indexes.sql new file mode 100644 index 0000000..ce590a5 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/database/02-idempotency-indexes.sql @@ -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; diff --git a/haobang-security-dm/upgrade-doc/database/03-direct-stats-baseline.sql b/haobang-security-dm/upgrade-doc/database/03-direct-stats-baseline.sql new file mode 100644 index 0000000..b31b90d --- /dev/null +++ b/haobang-security-dm/upgrade-doc/database/03-direct-stats-baseline.sql @@ -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') + AND created_at=TO_DATE('BUSINESS_DATE','YYYY-MM-DD') + AND created_at=TO_DATE('BUSINESS_DATE','YYYY-MM-DD') + AND last_receive_time=TO_DATE('BUSINESS_DATE','YYYY-MM-DD') + AND created_at=TO_DATE('BUSINESS_DATE','YYYY-MM-DD') + AND created_atTO_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; diff --git a/haobang-security-dm/upgrade-doc/kafka/kafka-upgrade.sh b/haobang-security-dm/upgrade-doc/kafka/kafka-upgrade.sh new file mode 100755 index 0000000..0f418b1 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/kafka/kafka-upgrade.sh @@ -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。"