diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/AdaptiveRawBatchSizer.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/AdaptiveRawBatchSizer.java index fb149d4..c9c26af 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/AdaptiveRawBatchSizer.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/AdaptiveRawBatchSizer.java @@ -7,8 +7,8 @@ public final class AdaptiveRawBatchSizer { private final int maxRecords; public AdaptiveRawBatchSizer( - @org.springframework.beans.factory.annotation.Value("${syslog.raw-listener.max-poll-records:500}") int baseRecords, - @org.springframework.beans.factory.annotation.Value("${syslog.raw-listener.adaptive-max-records:3000}") int maxRecords) { + @org.springframework.beans.factory.annotation.Value("${syslog.raw-listener.max-poll-records:3000}") int baseRecords, + @org.springframework.beans.factory.annotation.Value("${syslog.raw-listener.adaptive-max-records:5000}") int maxRecords) { if (baseRecords < 1 || maxRecords < baseRecords) { throw new IllegalArgumentException("普通自适应批量要求 1 <= 基础条数 <= 上限条数"); } diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java index 2fbc046..05032be 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java @@ -79,7 +79,7 @@ public class SysLogProcessor { private long rawProcessingBatchMaxBytes = 16777216L; @Autowired - private AdaptiveRawBatchSizer adaptiveRawBatchSizer = new AdaptiveRawBatchSizer(500, 3000); + private AdaptiveRawBatchSizer adaptiveRawBatchSizer = new AdaptiveRawBatchSizer(3000, 5000); // 每个普通listener线程独立学习上一个业务块的阶段，避免不同节点/通道互相放大批量。 private final ThreadLocal rawAdaptiveState = ThreadLocal.withInitial(RawAdaptiveState::new); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchWriter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchWriter.java index d6a69b7..198f889 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchWriter.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchWriter.java @@ -66,7 +66,7 @@ public class DmStandardizedBatchWriter { @Value("${dm.standardized-write.mode:BATCH}") private String mode; - @Value("${dm.standardized-write.batch-size:500}") + @Value("${dm.standardized-write.batch-size:2000}") private int batchSize; @Value("${syslog.slow-batch.threshold-ms:1000}") diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/SyslogNonNormalMessageServiceImpl.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/SyslogNonNormalMessageServiceImpl.java index 8c3d56b..a24ba36 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/SyslogNonNormalMessageServiceImpl.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/SyslogNonNormalMessageServiceImpl.java @@ -44,7 +44,7 @@ public class SyslogNonNormalMessageServiceImpl @Autowired(required = false) private DmLargeTextLimiter largeTextLimiter; - @Value("${dm.non-normal-write.batch-size:500}") + @Value("${dm.non-normal-write.batch-size:2000}") private int nonNormalBatchSize; @Override diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java index 3d1c621..016ea6c 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java @@ -53,8 +53,8 @@ public class KafkaReliabilityConfig { ConsumerFactory consumerFactory, DefaultErrorHandler syslogDefaultErrorHandler, @Value("${syslog.raw-listener.concurrency:3}") int concurrency, - @Value("${syslog.raw-listener.max-poll-records:500}") int baseRecords, - @Value("${syslog.raw-listener.adaptive-max-records:3000}") int maxPollRecords) { + @Value("${syslog.raw-listener.max-poll-records:3000}") int baseRecords, + @Value("${syslog.raw-listener.adaptive-max-records:5000}") int maxPollRecords) { if (concurrency < 1) { throw new IllegalArgumentException("syslog.raw-listener.concurrency 必须大于 0"); } diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/SyslogBuildStartupLogger.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/SyslogBuildStartupLogger.java index 3099b7e..19d4065 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/SyslogBuildStartupLogger.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/SyslogBuildStartupLogger.java @@ -30,8 +30,10 @@ public class SyslogBuildStartupLogger { + ", profiles=" + String.join(",", environment.getActiveProfiles()) + ", rawEnabled=" + value("syslog.raw-consumer.enabled", "true") + ", rawConcurrency=" + value("syslog.raw-listener.concurrency", "3") - + ", baseProcessingRecords=" + value("syslog.raw-listener.max-poll-records", "500") - + ", adaptiveMaxRecords=" + value("syslog.raw-listener.adaptive-max-records", "3000") + + ", baseProcessingRecords=" + value("syslog.raw-listener.max-poll-records", "3000") + + ", adaptiveMaxRecords=" + value("syslog.raw-listener.adaptive-max-records", "5000") + + ", dmStandardizedBatchSize=" + value("dm.standardized-write.batch-size", "2000") + + ", dmNonNormalBatchSize=" + value("dm.non-normal-write.batch-size", "2000") + ", rawFieldMaxUtf8Bytes=" + value("syslog.raw-listener.field-max-utf8-bytes", "2048") + ", samplingEnabled=" + value("syslog.backlog-sampling.enabled", "true") + ", statsWriteMode=" + value("stats.write-mode", "DIRECT") diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties index 43c7881..7517b3a 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties @@ -63,7 +63,7 @@ spring.kafka.consumer.topic=agent-01-syslog-topic syslog.raw-consumer.enabled=true -# 通用/Retry消费者上限。RAW底层上限由adaptive-max-records决定（3000），业务基础批量由max-poll-records决定（500）。 +# 通用/Retry消费者上限。RAW底层上限由adaptive-max-records决定（5000），业务基础批量由max-poll-records决定（3000）。 spring.kafka.consumer.max-poll-records=500 spring.kafka.consumer.properties.max.poll.interval.ms=900000 spring.kafka.consumer.properties.session.timeout.ms=45000 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 09736f6..fb93523 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 @@ -85,12 +85,12 @@ syslog.raw-listener.concurrency=${SYSLOG_RAW_LISTENER_CONCURRENCY:3} # 慢日志统一阈值，默认1000毫秒（1秒）：端到端批次、达梦事务、写入排队及非标异步写入。 # 达到阈值时打印诊断；0关闭耗时触发的慢日志，但异常/失败仍记录。已有同名环境变量优先。 syslog.slow-batch.threshold-ms=${SYSLOG_SLOW_BATCH_THRESHOLD_MS:1000} -# 普通业务基础批量500（保留旧配置名）；不再表示底层poll上限。已有同名环境变量仍覆盖基础值。 -# 目标业务批量=ceil(基础条数*10000/max(普通保留万分比,非标保留万分比))，并受下面3000上限限制。 -syslog.raw-listener.max-poll-records=${SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS:500} -# 底层RAW max.poll.records固定为此值；无积压时也可能poll到3000，再按500分块，全部处理后才ACK。 +# 普通业务基础批量3000（保留旧配置名）；不再表示底层poll上限。已有同名环境变量仍覆盖基础值。 +# 目标业务批量=ceil(基础条数*10000/max(普通保留万分比,非标保留万分比))，并受下面5000上限限制。 +syslog.raw-listener.max-poll-records=${SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS:3000} +# 底层RAW max.poll.records固定为此值；无积压时也可能poll到5000，再按3000分块，全部处理后才ACK。 # 基础值必须大于0且不超过上限；Retry/大日志/独立统计不采用此自适应策略。 -syslog.raw-listener.adaptive-max-records=${SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS:3000} +syslog.raw-listener.adaptive-max-records=${SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS:5000} # 大批次按 Kafka value 序列化字节拆成16MiB业务块，全部完成后提交本次poll；单条不会切断。 syslog.raw-listener.processing-batch-max-bytes=${SYSLOG_RAW_PROCESSING_BATCH_MAX_BYTES:16777216} # 每个业务块只向共享规则线程池提交8个worker，避免2000条对应2000个排队任务。 @@ -217,7 +217,8 @@ app.processor.process-timeout-ms=120000 # 失败时只对数据类错误有限二分； # 连接、超时或提交结果未知会立即停止并保留 Kafka offset，禁止放大为逐条请求风暴。 dm.standardized-write.mode=BATCH -dm.standardized-write.batch-size=${DM_STANDARDIZED_WRITE_BATCH_SIZE:500} +# 普通表/告警表每个分组的小批次上限2000行；不足2000直接写入，不等待凑满。 +dm.standardized-write.batch-size=${DM_STANDARDIZED_WRITE_BATCH_SIZE:2000} dm.standardized-write.writer-concurrency=2 dm.standardized-write.writer-queue-capacity=100 # 标准化大字段和非标原文均在达梦持久化边界截断；0 表示不限制。 @@ -236,7 +237,8 @@ dm.standardized-write.max-split-depth=10 dm.standardized-write.max-fallback-operations=100 dm.standardized-write.circuit-breaker-failure-threshold=3 dm.standardized-write.circuit-breaker-backoff-ms=30000 -dm.non-normal-write.batch-size=${DM_NON_NORMAL_WRITE_BATCH_SIZE:500} +# 非标表每条多行INSERT上限2000行；同名环境变量会覆盖JAR默认值。 +dm.non-normal-write.batch-size=${DM_NON_NORMAL_WRITE_BATCH_SIZE:2000} dm.non-normal-write.queue-capacity=${DM_NON_NORMAL_WRITE_QUEUE_CAPACITY:500} dm.non-normal-write.max-concurrency-retries=${DM_NON_NORMAL_WRITE_MAX_CONCURRENCY_RETRIES:3} dm.non-normal-write.concurrency-retry-backoff-ms=${DM_NON_NORMAL_WRITE_CONCURRENCY_RETRY_BACKOFF_MS:50} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/AdaptiveRawBatchSizerTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/AdaptiveRawBatchSizerTest.java index f4b34c4..80a4ec9 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/AdaptiveRawBatchSizerTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/AdaptiveRawBatchSizerTest.java @@ -4,6 +4,22 @@ import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.*; class AdaptiveRawBatchSizerTest { + @Test void newBaseStartsAt3000AndDegradedRatesAreCappedAt5000() { + AdaptiveRawBatchSizer sizer = new AdaptiveRawBatchSizer(3000, 5000); + assertEquals(3000, sizer.targetRecords(10000, 10000)); + assertEquals(3750, sizer.targetRecords(8000, 8000)); + assertEquals(5000, sizer.targetRecords(6000, 6000)); + assertEquals(5000, sizer.targetRecords(1000, 2000)); + } + + @Test void largerCapKeepsDefaultRetentionTargetAndAllowsFiveThousand() { + AdaptiveRawBatchSizer sizer = new AdaptiveRawBatchSizer(500, 5000); + assertEquals(500, sizer.targetRecords(10000, 10000)); + assertEquals(2500, sizer.targetRecords(1000, 2000)); + assertEquals(5000, sizer.targetRecords(1000, 1000)); + assertEquals(5000, sizer.targetRecords(0, 0)); + } + @Test void derivesMultiplierAndCapsWithoutPerStageConfiguration() { AdaptiveRawBatchSizer sizer = new AdaptiveRawBatchSizer(500, 3000); assertEquals(500, sizer.targetRecords(10000, 10000)); diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogAdaptiveRawBatchTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogAdaptiveRawBatchTest.java index 4dd250c..0cb4432 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogAdaptiveRawBatchTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogAdaptiveRawBatchTest.java @@ -23,6 +23,8 @@ class SysLogAdaptiveRawBatchTest { @BeforeEach void setup() { processor = spy(new SysLogProcessor()); + // 保留对旧基础值覆盖配置的回归场景；新默认值另行验证。 + ReflectionTestUtils.setField(processor, "adaptiveRawBatchSizer", new AdaptiveRawBatchSizer(500, 5000)); ReflectionTestUtils.setField(processor, "taskExecutor", (Executor) Runnable::run); ReflectionTestUtils.setField(processor, "largeTaskExecutor", (Executor) Runnable::run); ReflectionTestUtils.setField(processor, "largeProcessingBatchMaxBytes", 16777216L); @@ -30,6 +32,25 @@ class SysLogAdaptiveRawBatchTest { ack = mock(Acknowledgment.class); } + @Test void newBaseSplitsAt3000ThenBacklogCanUse5000WithoutEarlyAck() { + ReflectionTestUtils.setField(processor, "adaptiveRawBatchSizer", new AdaptiveRawBatchSizer(3000, 5000)); + List sizes = new ArrayList<>(); + doAnswer(call -> { + sizes.add(((List) call.getArgument(0)).size()); + verifyNoInteractions(ack); + processor.observeRawRetention(2000); + return null; + }).when(processor).processBatch(anyList(), anyBoolean(), anyString(), any()); + processor.consumeAndParallelProcess(records(5000), ack); + assertEquals(Arrays.asList(3000, 2000), sizes); + verify(ack).acknowledge(); + clearInvocations(ack); + sizes.clear(); + processor.consumeAndParallelProcess(records(5000), ack); + assertEquals(Collections.singletonList(5000), sizes); + verify(ack).acknowledge(); + } + @Test void livePollSplitsInto500WithoutEarlyAckOrRecordLoss() { List sizes = new ArrayList<>(); List offsets = new ArrayList<>(); @@ -41,9 +62,9 @@ class SysLogAdaptiveRawBatchTest { processor.observeRawRetention(10000); return null; }).when(processor).processBatch(anyList(), anyBoolean(), anyString(), any()); - processor.consumeAndParallelProcess(records(3000), ack); - assertEquals(Arrays.asList(500, 500, 500, 500, 500, 500), sizes); - for (int i = 0; i < 3000; i++) assertEquals((long) i, offsets.get(i).longValue()); + processor.consumeAndParallelProcess(records(5000), ack); + assertEquals(Collections.nCopies(10, 500), sizes); + for (int i = 0; i < 5000; i++) assertEquals((long) i, offsets.get(i).longValue()); verify(ack).acknowledge(); } @@ -70,14 +91,14 @@ class SysLogAdaptiveRawBatchTest { verify(ack, times(3)).acknowledge(); } - @Test void tenPercentHits3000CapAndIncompleteTailIsNotHeldWaitingForMore() { + @Test void tenPercentHits5000CapAndIncompleteTailIsNotHeldWaitingForMore() { processor.observeRawRetention(1000); List sizes = new ArrayList<>(); doAnswer(call -> { sizes.add(((List) call.getArgument(0)).size()); return null; }) .when(processor).processBatch(anyList(), anyBoolean(), anyString(), any()); - processor.consumeAndParallelProcess(records(3000), ack); + processor.consumeAndParallelProcess(records(5000), ack); processor.consumeAndParallelProcess(records(17), ack); - assertEquals(Arrays.asList(3000, 17), sizes); + assertEquals(Arrays.asList(5000, 17), sizes); } @Test void bytesLimitStillWinsAndOversizedSingleRecordIsNotLost() { diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogSlowBatchTraceTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogSlowBatchTraceTest.java index 824ec19..9c5f709 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogSlowBatchTraceTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogSlowBatchTraceTest.java @@ -56,7 +56,7 @@ class SysLogSlowBatchTraceTest { String log = onlyTrace(); assertTrue(log.contains("PREPARE=")); assertTrue(log.contains("ADAPTIVE_CHUNK_SELECTION=")); - assertTrue(log.contains("adaptiveTargetHistogram={500=1}")); + assertTrue(log.contains("adaptiveTargetHistogram={3000=1}")); assertTrue(log.contains("actualChunkHistogram={2=1}")); assertTrue(log.contains("ALARM_DM=")); assertTrue(log.contains("raw:0=12..14")); diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmStandardizedBatchWriterTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmStandardizedBatchWriterTest.java index 7d2b9c0..20aca38 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmStandardizedBatchWriterTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmStandardizedBatchWriterTest.java @@ -54,6 +54,25 @@ class DmStandardizedBatchWriterTest { when(normalDataService.isAlarmData(any())).thenReturn(false); } + @Test + void normalAndAlarmBatchesUse2000LimitAndWriteTailImmediately() { + ReflectionTestUtils.setField(writer, "batchSize", 2000); + List sizes = new ArrayList<>(); + doAnswer(invocation -> { + sizes.add(((List) invocation.getArgument(1)).size()); + return null; + }).when(batchExecutor).execute(anyBoolean(), anyList(), anyInt()); + for (boolean alarm : new boolean[]{false, true}) { + when(normalDataService.isAlarmData(any())).thenReturn(alarm); + sizes.clear(); + assertEquals(3000, writer.write(requests(3000, -1)).getSuccessfulRows()); + assertEquals(java.util.Arrays.asList(2000, 1000), sizes); + sizes.clear(); + assertEquals(500, writer.write(requests(500, -1)).getSuccessfulRows()); + assertEquals(java.util.Collections.singletonList(500), sizes); + } + } + @Test void alarmsKeepLegacyTruncationInBothBatchAndSingleMode() { DmLargeTextLimiter limiter = new DmLargeTextLimiter(); diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/SyslogNonNormalMessageServiceBatchTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/SyslogNonNormalMessageServiceBatchTest.java index 19e5178..71593f2 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/SyslogNonNormalMessageServiceBatchTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/SyslogNonNormalMessageServiceBatchTest.java @@ -39,6 +39,24 @@ class SyslogNonNormalMessageServiceBatchTest { .thenAnswer(invocation -> ((List) invocation.getArgument(0)).size()); } + @Test + void writes2000RowsPerInsertAndDoesNotHoldIncompleteTail() { + ReflectionTestUtils.setField(service, "nonNormalBatchSize", 2000); + List messages = new ArrayList<>(); + for (int i = 0; i < 3000; i++) { + SyslogNonNormalMessage message = new SyslogNonNormalMessage(); + message.setId("non-normal-" + i); + messages.add(message); + } + assertEquals(3000, service.saveBatchMessages(messages)); + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); + verify(mapper, times(2)).batchInsert(captor.capture()); + assertEquals(2000, captor.getAllValues().get(0).size()); + assertEquals(1000, captor.getAllValues().get(1).size()); + verify(directStatsProjectionWriter).writeNonNormalRows(messages); + } + @Test void chunksLargePollInsideOneServiceTransactionAndAggregatesStatsOnce() { List messages = new ArrayList<>(); diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/DmBatchDefaultsTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/DmBatchDefaultsTest.java new file mode 100644 index 0000000..325e15f --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/DmBatchDefaultsTest.java @@ -0,0 +1,30 @@ +package com.config; + +import com.common.service.DmStandardizedBatchWriter; +import com.common.service.impl.SyslogNonNormalMessageServiceImpl; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Value; + +import java.io.InputStream; +import java.util.Properties; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class DmBatchDefaultsTest { + @Test + void propertiesAndInjectionFallbacksBothDefaultTo2000() throws Exception { + Properties properties = new Properties(); + try (InputStream input = getClass().getResourceAsStream("/application.properties")) { + properties.load(input); + } + assertEquals("${DM_STANDARDIZED_WRITE_BATCH_SIZE:2000}", + properties.getProperty("dm.standardized-write.batch-size")); + assertEquals("${DM_NON_NORMAL_WRITE_BATCH_SIZE:2000}", + properties.getProperty("dm.non-normal-write.batch-size")); + assertEquals("${dm.standardized-write.batch-size:2000}", + DmStandardizedBatchWriter.class.getDeclaredField("batchSize").getAnnotation(Value.class).value()); + assertEquals("${dm.non-normal-write.batch-size:2000}", + SyslogNonNormalMessageServiceImpl.class.getDeclaredField("nonNormalBatchSize") + .getAnnotation(Value.class).value()); + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/KafkaReliabilityConfigTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/KafkaReliabilityConfigTest.java index 1534b36..2db23bb 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/KafkaReliabilityConfigTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/KafkaReliabilityConfigTest.java @@ -19,21 +19,21 @@ import static org.mockito.Mockito.when; class KafkaReliabilityConfigTest { @Test - void rawDefaultIs500BaseAnd3000NativePollCeiling() throws Exception { + void rawDefaultIs3000BaseAnd5000NativePollCeiling() throws Exception { java.util.Properties properties = new java.util.Properties(); try (java.io.InputStream input = getClass().getResourceAsStream("/application.properties")) { properties.load(input); } - assertEquals("${SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS:500}", + assertEquals("${SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS:3000}", properties.getProperty("syslog.raw-listener.max-poll-records")); java.lang.reflect.Method method = java.util.Arrays.stream(KafkaReliabilityConfig.class.getDeclaredMethods()) .filter(candidate -> candidate.getName().equals("rawKafkaListenerContainerFactory")).findFirst().get(); org.springframework.beans.factory.annotation.Value fallback = method.getParameters()[4] .getAnnotation(org.springframework.beans.factory.annotation.Value.class); - assertEquals("${syslog.raw-listener.max-poll-records:500}", fallback.value()); - assertEquals("${SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS:3000}", + assertEquals("${syslog.raw-listener.max-poll-records:3000}", fallback.value()); + assertEquals("${SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS:5000}", properties.getProperty("syslog.raw-listener.adaptive-max-records")); - assertEquals("${syslog.raw-listener.adaptive-max-records:3000}", method.getParameters()[5] + assertEquals("${syslog.raw-listener.adaptive-max-records:5000}", method.getParameters()[5] .getAnnotation(org.springframework.beans.factory.annotation.Value.class).value()); } @@ -66,13 +66,13 @@ class KafkaReliabilityConfigTest { ConcurrentKafkaListenerContainerFactory factory = config.rawKafkaListenerContainerFactory( - configurer, consumerFactory, errorHandler, 3, 500, 3000); + configurer, consumerFactory, errorHandler, 3, 3000, 5000); assertEquals(ContainerProperties.AckMode.MANUAL_IMMEDIATE, factory.getContainerProperties().getAckMode()); ArgumentCaptor actual = ArgumentCaptor.forClass(ConsumerFactory.class); verify(configurer).configure(any(), actual.capture()); - assertEquals(3000, actual.getValue().getConfigurationProperties().get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG)); + assertEquals(5000, actual.getValue().getConfigurationProperties().get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG)); assertEquals(false, actual.getValue().getConfigurationProperties().get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)); assertEquals(500, consumerFactory.getConfigurationProperties().get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG)); } diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/SyslogBuildStartupLoggerTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/SyslogBuildStartupLoggerTest.java index 45a2be7..323f013 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/SyslogBuildStartupLoggerTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/SyslogBuildStartupLoggerTest.java @@ -15,6 +15,8 @@ class SyslogBuildStartupLoggerTest { MockEnvironment environment = new MockEnvironment() .withProperty("syslog.raw-listener.max-poll-records", "700") .withProperty("syslog.raw-listener.adaptive-max-records", "2500") + .withProperty("dm.standardized-write.batch-size", "800") + .withProperty("dm.non-normal-write.batch-size", "900") .withProperty("syslog.slow-batch.threshold-ms", "1500") .withProperty("stats.projection-async.enabled", "false"); environment.setActiveProfiles("prod"); @@ -32,6 +34,8 @@ class SyslogBuildStartupLoggerTest { assertTrue(message.contains("profiles=prod")); assertTrue(message.contains("baseProcessingRecords=700")); assertTrue(message.contains("adaptiveMaxRecords=2500")); + assertTrue(message.contains("dmStandardizedBatchSize=800")); + assertTrue(message.contains("dmNonNormalBatchSize=900")); assertTrue(message.contains("statsPublisherEnabled=false")); assertTrue(message.contains("slowBatchThresholdMs=1500")); } finally { @@ -46,8 +50,10 @@ class SyslogBuildStartupLoggerTest { .withProperty("codeRevision", "fake")).startupSummary(); assertTrue(summary.contains("codeRevision=" + SyslogBuildStartupLogger.CODE_REVISION)); assertFalse(summary.contains("buildTime=")); - assertTrue(summary.contains("baseProcessingRecords=500")); - assertTrue(summary.contains("adaptiveMaxRecords=3000")); + assertTrue(summary.contains("baseProcessingRecords=3000")); + assertTrue(summary.contains("adaptiveMaxRecords=5000")); + assertTrue(summary.contains("dmStandardizedBatchSize=2000")); + assertTrue(summary.contains("dmNonNormalBatchSize=2000")); assertTrue(summary.contains("slowBatchThresholdMs=1000")); assertTrue(summary.contains("rawFieldMaxUtf8Bytes=2048")); } diff --git a/haobang-security-dm/upgrade-doc/v4/config/02-throughput-upgrade.md b/haobang-security-dm/upgrade-doc/v4/config/02-throughput-upgrade.md index 254e896..90ec94a 100644 --- a/haobang-security-dm/upgrade-doc/v4/config/02-throughput-upgrade.md +++ b/haobang-security-dm/upgrade-doc/v4/config/02-throughput-upgrade.md @@ -1,12 +1,12 @@ -# 吞吐优化：统计退出事实写库路径、普通业务批次自适应500～3000条 +# 吞吐优化：统计退出事实写库路径、普通业务批次自适应3000～5000条 只升级 `syslog-consumer-rule`，保持 JDK 8。无需新增表或 Topic；不能删除现有 `stats_projection_outbox`、`stats_projection_applied_event`：前者用于排空旧事件及兼容回滚，后者继续用于统计去重。 ## 本次行为 -- 普通 RAW 底层每次poll上限固定3000，业务块以500为基础按采样阶段动态扩大（默认最低档2500，最多3000）。 - 无积压时也可能一次poll返回3000条，随后分成500条业务块；Retry仍使用通用配置，大日志仍默认5条。 +- 普通 RAW 底层每次poll上限默认5000，业务块以3000为基础按采样阶段动态扩大（默认最低保留率时为5000）。 + 无积压时也可能一次poll返回5000条，随后按最多3000条分块；Retry仍使用通用配置，大日志仍默认5条。 - 普通业务处理按 Kafka value 序列化长度分成最多16MiB的块，单条不拆分；一个poll的全部块完成后ACK。 - 每个业务块最多8个规则worker，按原顺序汇总结果。不会为2000条消息同时排队2000个任务。 - 主处理方法不再调用入口统计写库，因此主批次日志 `ingressMs=0`。 @@ -35,8 +35,10 @@ | properties | 环境变量 | 默认值 | | --- | --- | --- | -| syslog.raw-listener.max-poll-records | SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS | 500 | -| syslog.raw-listener.adaptive-max-records | SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS | 3000 | +| syslog.raw-listener.max-poll-records | SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS | 3000 | +| syslog.raw-listener.adaptive-max-records | SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS | 5000 | +| dm.standardized-write.batch-size | DM_STANDARDIZED_WRITE_BATCH_SIZE | 2000 | +| dm.non-normal-write.batch-size | DM_NON_NORMAL_WRITE_BATCH_SIZE | 2000 | | syslog.raw-listener.processing-batch-max-bytes | SYSLOG_RAW_PROCESSING_BATCH_MAX_BYTES | 16777216 | | syslog.normalization.workers-per-batch | SYSLOG_NORMALIZATION_WORKERS_PER_BATCH | 8 | | stats.ingress-metadata.group-id | STATS_INGRESS_GROUP_ID | 主消费组名加 `-ingress-v1` | @@ -47,20 +49,24 @@ | stats.projection-async.memory-max-pending-groups | STATS_PROJECTION_MEMORY_MAX_PENDING_GROUPS | 50000 | | analysis.late-replay.notify-timeout-seconds | ANALYSIS_LATE_REPLAY_NOTIFY_TIMEOUT_SECONDS | 5 | -只替换JAR也会使用上述默认值；已有 `SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS=2000` 时须删除或改成500。 +只替换JAR也会使用上述默认值；已有 `SYSLOG_RAW_LISTENER_MAX_POLL_RECORDS` 覆盖时须删除或改成3000， +已有 `SYSLOG_RAW_LISTENER_ADAPTIVE_MAX_RECORDS` 覆盖时须删除或改成5000。无需数据库或Kafka脚本。 `spring.kafka.consumer.max-poll-records=500` 留给通用消费者，不再决定普通RAW批次上限。 -Kafka上限不保证每次拉满；业务块还受到16MiB限制。达梦每个小事务仍最多500行。 +Kafka上限不保证每次拉满；业务块还受到16MiB限制。达梦普通表/告警表按目标表、日期分组后，每个小事务默认最多2000行；非标表每条多行INSERT默认最多2000行，同一次服务调用的多个INSERT仍在同一个事务中。 +不足2000行直接写入，不跨业务块等待凑满。已有 `DM_STANDARDIZED_WRITE_BATCH_SIZE` 或 `DM_NON_NORMAL_WRITE_BATCH_SIZE` 环境变量时，删除覆盖或改成2000才会采用新值。 +例如同表同日保留3000行会按2000+1000写入；采样后只保留500行就只写500行。更大批次不保证性能提升，需对比现场执行及提交耗时。 保留旧配置名 `syslog.raw-listener.max-poll-records`，但其含义现在是**业务基础条数**。 底层RAW的 `max.poll.records` 固定使用 `adaptive-max-records`，不重建consumer、不反射修改Kafka内部参数。 基础值必须大于0且不超过上限，非法配置拒绝启动。独立入口统计默认2000条（不做规则计算），不采用该策略。 业务块目标条数 = min(上限, ceil(基础条数 × 10000 / 较高保留万分比))。 -普通最低1000、非标最低2000，取较高2000，500×10000÷2000=2500；两者都为1000时命中3000上限。 +普通最低1000、非标最低2000，取较高2000，3000×10000÷2000=15000，受5000上限限制； +较高保留率为8000时目标3750，为6000或更低时目标5000。 两者都为0时也只到上限，避免除0；告警仍全部保留。各阶段自动沿用现有采样决策，不另配倍率。 -实现采用**上一业务块的阶段反馈决定下一块大小**。每个RAW消费线程独立，首次或超过1分钟无有效反馈时从500探测。 -一个块内存在多个延迟阶段时取最高保留率，准备/规则异常保守回到500；恢复全量保存后的下一块回到500。 -例如首次poll3000条且均为默认最低档：先处理500，确认保留率后处理余下2500；后续poll可按2500+500处理。 +实现采用**上一业务块的阶段反馈决定下一块大小**。每个RAW消费线程独立，首次或超过1分钟无有效反馈时从3000探测。 +一个块内存在多个延迟阶段时取最高保留率，准备/规则异常保守回到3000；恢复全量保存后的下一块回到3000。 +例如首次poll5000条且均为默认最低档：先处理3000，确认保留率后处理余下2000；后续poll目标可到5000。 不会为了凑足目标条数等待新消息，也不会跳过已poll的尾部记录；每个poll所有块处理完才提交offset。 启用连续性放行时失败块仍沿用既有DLT/记录策略，之后以基础批量继续，未增加新的业务丢弃规则。 所有消息仍要解密、跑规则才能确认告警，扩大批量会增加整体poll处理耗时、内存和GC压力。 @@ -116,10 +122,11 @@ docker logs --since 10m syslog-consumer-rule-dm 2>&1 | grep -E 'STATS_ASYNC_LOSS docker exec syslog-consumer-rule-dm sh -c 'curl -fsS http://127.0.0.1:8289/xdrrule/api/syslog/stats/direct/status' ``` -确认普通配置输出 `maxPollRecords=3000, baseProcessingRecords=500, adaptiveMaxRecords=3000`， +确认普通配置输出 `maxPollRecords=5000, baseProcessingRecords=3000, adaptiveMaxRecords=5000`， +启动日志应包含 `dmStandardizedBatchSize=2000, dmNonNormalBatchSize=2000`， 主批次 `ingressMs=0`，入口状态模式为 `INDEPENDENT_KAFKA_DIRECT_MERGE`。 -`polledRecords`可大于500；`adaptiveTargetHistogram`和`actualChunkHistogram`分别表示业务目标与实际块大小。 -例如 `{500=1, 2500=1}` 表示对应大小各1块，不代表保留率。实际大小受剩余条数及字节上限限制。 +`polledRecords`可大于3000；`adaptiveTargetHistogram`和`actualChunkHistogram`分别表示业务目标与实际块大小。 +例如 `{3000=1, 2000=1}` 表示对应大小各1块，不代表保留率。实际大小受剩余条数及字节上限限制。 直接统计状态中 `daily_projection_write_mode=BEST_EFFORT_AFTER_COMMIT_KAFKA`， `async_publisher` 下有排队事件/分组数、已发布、失败重试、丢弃计数。 `dropped_fact_rows_total` 指受统计缺失影响的事实行数/尝试行数，并不表示事实数据被删除。