From c392f7b47f1ccb36413b0d2417abde30820906ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Sun, 13 Sep 2026 04:01:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(syslog-consumer-rule):=20=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E6=85=A2=E6=89=B9=E6=AC=A1=E5=85=A8=E9=93=BE=E8=B7=AF?= =?UTF-8?q?=E8=80=97=E6=97=B6=E8=AF=8A=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 批次端到端耗时达到500ms时输出SLOW_SYSLOG_BATCH日志 - 记录入口统计、规则计算、InfluxDB和达梦写入的分阶段耗时 - 使用MANUAL_IMMEDIATE模式准确统计Kafka offset同步提交耗时 - 记录Kafka提交状态、异常类型和端到端总耗时 - 在常规批次日志中补充ingressMs - 增加Kafka监听器ACK模式回归测试 --- .../Modules/NormalData/SysLogProcessor.java | 138 ++++++++++++++++-- .../com/config/KafkaReliabilityConfig.java | 11 +- .../src/main/resources/application.properties | 2 + .../config/KafkaReliabilityConfigTest.java | 67 +++++++++ 4 files changed, 208 insertions(+), 10 deletions(-) create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/KafkaReliabilityConfigTest.java 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 b75a812..617c937 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 @@ -74,6 +74,12 @@ public class SysLogProcessor { @Value("${syslog.failure-continuity.enabled:true}") private boolean failureContinuityEnabled; + @Value("${syslog.slow-batch.threshold-ms:500}") + private long slowBatchThresholdMs; + + /** listener、业务处理和 Kafka commit 均在同一个 consumer 线程，ThreadLocal 用于拼接端到端慢批次。 */ + private final ThreadLocal slowBatchTrace = new ThreadLocal<>(); + @Autowired private AppProperties appProperties; @@ -186,11 +192,18 @@ public class SysLogProcessor { ack.acknowledge(); return; } - for (List> chunk : partitionRecordsByBytes( - records, Math.max(1L, largeProcessingBatchMaxBytes))) { - processBatchWithContinuity(chunk, enforceRetryDue, lane, largeTaskExecutor); + beginSlowBatchTrace(lane, records); + long processingStartedAt = System.nanoTime(); + try { + for (List> chunk : partitionRecordsByBytes( + records, Math.max(1L, largeProcessingBatchMaxBytes))) { + processBatchWithContinuity(chunk, enforceRetryDue, lane, largeTaskExecutor); + } + long processingMs = elapsedMillis(processingStartedAt); + acknowledgeAndCompleteSlowBatchTrace(ack, processingMs); + } finally { + slowBatchTrace.remove(); } - ack.acknowledge(); } @KafkaListener( @@ -220,8 +233,15 @@ public class SysLogProcessor { ack.acknowledge(); return; } - processBatchWithContinuity(records, enforceRetryDue, lane, normalizationExecutor); - ack.acknowledge(); + beginSlowBatchTrace(lane, records); + long processingStartedAt = System.nanoTime(); + try { + processBatchWithContinuity(records, enforceRetryDue, lane, normalizationExecutor); + long processingMs = elapsedMillis(processingStartedAt); + acknowledgeAndCompleteSlowBatchTrace(ack, processingMs); + } finally { + slowBatchTrace.remove(); + } } private void processBatchWithContinuity(List> records, @@ -263,7 +283,9 @@ public class SysLogProcessor { long prepareMs = System.currentTimeMillis() - prepareStartedAt; // 入口状态是 Kafka 已可靠接收的事实。批内按设备/探针聚合，失败时禁止 ACK；重投只重复 MIN/MAX。 + long ingressStartedAt = System.nanoTime(); int ingressGroups = ingressStatsProjectionWriter.writeBatch(ingressEvents(preparedMessages)); + long ingressMs = elapsedMillis(ingressStartedAt); // 必须先完成规则计算，才能确保告警全量保留，并让未保留消息同时跳过 InfluxDB 和达梦。 long normalizeStartedAt = System.currentTimeMillis(); @@ -283,8 +305,10 @@ public class SysLogProcessor { long normalizeMs = System.currentTimeMillis() - normalizeStartedAt; // 告警是最高优先级：先独立提交告警表，不能让 InfluxDB 或普通表的慢写阻塞告警入库。 + long persistencePlanStartedAt = System.nanoTime(); StandardizedPersistencePlan standardizedPlan = prepareStandardizedPersistence( samplingPlan.normalizations, metrics); + long persistencePlanMs = elapsedMillis(persistencePlanStartedAt); long alarmWriteStartedAt = System.nanoTime(); DmStandardizedBatchWriter.WriteReport alarmReport = standardizedBatchWriter.write(standardizedPlan.alarmWriteRequests); @@ -295,14 +319,20 @@ public class SysLogProcessor { samplingPlan.rawMessages, metrics, lane); long influxMs = System.currentTimeMillis() - influxStartedAt; + long remainingDmStartedAt = System.nanoTime(); persistRemainingStandardizedData(standardizedPlan, alarmReport, alarmWriteMs, metrics); + long remainingDmMs = elapsedMillis(remainingDmStartedAt); totalProcessed.addAndGet(records.size()); + long businessTotalMs = System.currentTimeMillis() - batchStartedAt; + recordSlowBatchStages(records.size(), preparedMessages.size(), persistedMessages.size(), + metrics.normalizedRows.get(), prepareMs, ingressMs, normalizeMs, + persistencePlanMs, alarmWriteMs, influxMs, remainingDmMs, metrics); log.info("Syslog批次完成: lane={}, topicPartitions={}, records={}, prepared={}, persisted={}, " + "success={}, filtered={}, retry={}, dlt={}, normalizedRows={}, dmSuccessfulRows={}, " + "dmFailedMessages={}, ingressGroups={}, rawStorageBypassed={}, backlogDegraded={}, " + "rawSampledOut={}, normalRowsSampledOut={}, nonNormalRowsSampledOut={}, " - + "prepareMs={}, influxMs={}, normalizeMs={}, " + + "prepareMs={}, ingressMs={}, influxMs={}, normalizeMs={}, " + "ruleComputeMs={}, ruleSnapshotMs={}, messageParseMs={}, fieldMappingMs={}, " + "filterCompleteMs={}, enrichmentMs={}, ruleAssembleMs={}, compiledRuleHit={}, " + "compiledRuleMiss={}, messageParseHit={}, messageParseMiss={}, deviceUnique={}, " @@ -315,7 +345,7 @@ public class SysLogProcessor { metrics.rawStorageBypassed.get(), metrics.backlogDegraded.get(), metrics.rawSampledOut.get(), metrics.normalRowsSampledOut.get(), metrics.nonNormalRowsSampledOut.get(), - prepareMs, influxMs, normalizeMs, metrics.ruleComputeMs(), metrics.ruleSnapshotMs(), + prepareMs, ingressMs, influxMs, normalizeMs, metrics.ruleComputeMs(), metrics.ruleSnapshotMs(), metrics.messageParseMs(), metrics.fieldMappingMs(), metrics.filterCompleteMs(), metrics.enrichmentMs(), metrics.ruleAssembleMs(), metrics.compiledRuleHits(), metrics.compiledRuleMisses(), metrics.messageParseHits(), metrics.messageParseMisses(), @@ -324,7 +354,67 @@ public class SysLogProcessor { metrics.nonNormalQueued(), metrics.nonNormalRejected(), nonNormalAsyncCompletedTotal.get(), nonNormalAsyncFailedTotal.get(), nonNormalAsyncRejectedTotal.get(), - System.currentTimeMillis() - batchStartedAt); + businessTotalMs); + } + + private void beginSlowBatchTrace(String lane, List> records) { + slowBatchTrace.set(new SlowBatchTrace(lane, topicPartitions(records), records.size(), + System.nanoTime())); + } + + private void recordSlowBatchStages(int records, int prepared, int persisted, int normalizedRows, + long prepareMs, long ingressMs, long normalizeMs, + long persistencePlanMs, long alarmDmMs, long influxMs, + long remainingDmMs, BatchMetrics metrics) { + SlowBatchTrace trace = slowBatchTrace.get(); + if (trace == null) return; + trace.chunks++; + trace.recordsProcessed += records; + trace.prepared += prepared; + trace.persisted += persisted; + trace.normalizedRows += normalizedRows; + trace.prepareMs += prepareMs; + trace.ingressMs += ingressMs; + trace.normalizeMs += normalizeMs; + trace.persistencePlanMs += persistencePlanMs; + trace.alarmDmMs += alarmDmMs; + trace.influxMs += influxMs; + trace.remainingDmMs += remainingDmMs; + trace.ruleComputeMs += metrics.ruleComputeMs(); + trace.messageParseMs += metrics.messageParseMs(); + trace.fieldMappingMs += metrics.fieldMappingMs(); + } + + private void acknowledgeAndCompleteSlowBatchTrace(Acknowledgment ack, long businessProcessMs) { + long commitStartedAt = System.nanoTime(); + try { + ack.acknowledge(); + completeSlowBatchTrace(businessProcessMs, elapsedMillis(commitStartedAt), null); + } catch (RuntimeException commitFailure) { + completeSlowBatchTrace(businessProcessMs, elapsedMillis(commitStartedAt), commitFailure); + throw commitFailure; + } + } + + private void completeSlowBatchTrace(long businessProcessMs, long kafkaCommitMs, + RuntimeException commitFailure) { + SlowBatchTrace trace = slowBatchTrace.get(); + if (trace == null) return; + long endToEndMs = elapsedMillis(trace.startedAtNanos); + if (slowBatchThresholdMs <= 0L || endToEndMs < slowBatchThresholdMs) return; + log.warn("[SLOW_SYSLOG_BATCH] lane={}, topicPartitions={}, polledRecords={}, processedRecords={}, " + + "chunks={}, prepared={}, persisted={}, normalizedRows={}, prepareMs={}, ingressMs={}, " + + "normalizeMs={}, ruleComputeWorkerMs={}, messageParseWorkerMs={}, " + + "fieldMappingWorkerMs={}, persistencePlanMs={}, alarmDmWriteMs={}, influxMs={}, " + + "remainingDmWriteMs={}, businessProcessMs={}, kafkaCommitMs={}, " + + "kafkaCommitStatus={}, kafkaCommitErrorType={}, endToEndMs={}", + trace.lane, trace.topicPartitions, trace.polledRecords, trace.recordsProcessed, + trace.chunks, trace.prepared, trace.persisted, trace.normalizedRows, + trace.prepareMs, trace.ingressMs, trace.normalizeMs, trace.ruleComputeMs, + trace.messageParseMs, trace.fieldMappingMs, trace.persistencePlanMs, + trace.alarmDmMs, trace.influxMs, trace.remainingDmMs, + businessProcessMs, kafkaCommitMs, commitFailure == null ? "SUCCESS" : "FAILED", + commitFailure == null ? "none" : commitFailure.getClass().getName(), endToEndMs); } private void routeContinuityBatch(List> records, @@ -1190,6 +1280,36 @@ public class SysLogProcessor { } } + private static final class SlowBatchTrace { + private final String lane; + private final String topicPartitions; + private final int polledRecords; + private final long startedAtNanos; + private int chunks; + private int recordsProcessed; + private int prepared; + private int persisted; + private int normalizedRows; + private long prepareMs; + private long ingressMs; + private long normalizeMs; + private long ruleComputeMs; + private long messageParseMs; + private long fieldMappingMs; + private long persistencePlanMs; + private long alarmDmMs; + private long influxMs; + private long remainingDmMs; + + private SlowBatchTrace(String lane, String topicPartitions, int polledRecords, + long startedAtNanos) { + this.lane = lane; + this.topicPartitions = topicPartitions; + this.polledRecords = polledRecords; + this.startedAtNanos = startedAtNanos; + } + } + /** 单批终态计数，仅用于聚合日志，不参与 ACK 决策。 */ static final class BatchMetrics { private final AtomicInteger success = new AtomicInteger(); 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 0347d36..0c84123 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 @@ -9,6 +9,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.DefaultErrorHandler; import org.springframework.util.backoff.FixedBackOff; @@ -33,7 +34,11 @@ public class KafkaReliabilityConfig { ConcurrentKafkaListenerContainerFactoryConfigurer configurer, ConsumerFactory consumerFactory, DefaultErrorHandler syslogDefaultErrorHandler) { - return configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler); + ConcurrentKafkaListenerContainerFactory factory = + configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler); + // 默认工厂承载普通 Retry/Replay listener，同样需要让 acknowledge() 覆盖真实提交耗时。 + factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); + return factory; } /** @@ -52,6 +57,9 @@ public class KafkaReliabilityConfig { ConcurrentKafkaListenerContainerFactory factory = configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler); factory.setConcurrency(concurrency); + // 在 listener 线程调用 acknowledge() 时立即同步提交，使慢批次日志能够准确覆盖 + // “业务处理完成 -> Kafka offset 提交完成”的全部时间；可靠性语义仍是处理成功后提交。 + factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); return factory; } @@ -78,6 +86,7 @@ public class KafkaReliabilityConfig { ConcurrentKafkaListenerContainerFactory factory = configuredFactory(configurer, largeConsumerFactory, syslogDefaultErrorHandler); factory.setConcurrency(concurrency); + factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); return factory; } 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 3b700ec..0055ec8 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 @@ -82,6 +82,8 @@ spring.kafka.consumer.large-topic=${KAFKA_CONSUMER_LARGE_TOPIC:${spring.kafka.co syslog.raw-consumer.enabled=true # Raw Topic 当前有 3 个分区，使用独立工厂显式创建 3 个 consumer；可通过环境变量灰度调整。 syslog.raw-listener.concurrency=${SYSLOG_RAW_LISTENER_CONCURRENCY:3} +# 端到端批次耗时达到该阈值时，打印业务各阶段及 Kafka offset 同步提交耗时；0 表示关闭。 +syslog.slow-batch.threshold-ms=${SYSLOG_SLOW_BATCH_THRESHOLD_MS:500} # 连续性优先：特殊/系统性异常优先写 DLT；即使 DLT 失败也记录告警并推进原 offset，避免阻塞分区。 syslog.failure-continuity.enabled=${SYSLOG_FAILURE_CONTINUITY_ENABLED:true} # DLT 故障时最多快速等待 1 秒；首次失败后整批直接记录并放行。 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 new file mode 100644 index 0000000..530d20b --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/KafkaReliabilityConfigTest.java @@ -0,0 +1,67 @@ +package com.config; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.listener.DefaultErrorHandler; + +import java.util.Collections; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class KafkaReliabilityConfigTest { + + @Test + void retryListenerCommitsInsideAcknowledgment() { + KafkaReliabilityConfig config = new KafkaReliabilityConfig(); + ConcurrentKafkaListenerContainerFactoryConfigurer configurer = + mock(ConcurrentKafkaListenerContainerFactoryConfigurer.class); + @SuppressWarnings("unchecked") + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + + ConcurrentKafkaListenerContainerFactory factory = + config.kafkaListenerContainerFactory(configurer, consumerFactory, + config.syslogDefaultErrorHandler()); + + assertEquals(ContainerProperties.AckMode.MANUAL_IMMEDIATE, + factory.getContainerProperties().getAckMode()); + } + + @Test + void rawListenerCommitsInsideAcknowledgmentForAccurateSlowBatchTiming() { + KafkaReliabilityConfig config = new KafkaReliabilityConfig(); + ConcurrentKafkaListenerContainerFactoryConfigurer configurer = + mock(ConcurrentKafkaListenerContainerFactoryConfigurer.class); + @SuppressWarnings("unchecked") + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + DefaultErrorHandler errorHandler = config.syslogDefaultErrorHandler(); + + ConcurrentKafkaListenerContainerFactory factory = + config.rawKafkaListenerContainerFactory( + configurer, consumerFactory, errorHandler, 3); + + assertEquals(ContainerProperties.AckMode.MANUAL_IMMEDIATE, + factory.getContainerProperties().getAckMode()); + } + + @Test + void largeListenerAlsoCommitsInsideAcknowledgment() { + KafkaReliabilityConfig config = new KafkaReliabilityConfig(); + ConcurrentKafkaListenerContainerFactoryConfigurer configurer = + mock(ConcurrentKafkaListenerContainerFactoryConfigurer.class); + @SuppressWarnings("unchecked") + ConsumerFactory consumerFactory = mock(ConsumerFactory.class); + when(consumerFactory.getConfigurationProperties()).thenReturn(Collections.emptyMap()); + + ConcurrentKafkaListenerContainerFactory factory = + config.largeRawKafkaListenerContainerFactory(configurer, consumerFactory, + config.syslogDefaultErrorHandler(), 1, 5, 33554432, 67108864); + + assertEquals(ContainerProperties.AckMode.MANUAL_IMMEDIATE, + factory.getContainerProperties().getAckMode()); + } +}