From c868e3befd52cf91d30374c23c0bf649d9424c11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Sat, 12 Sep 2026 19:24:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(syslog):=20=E4=BC=98=E5=8C=96=E5=A4=A7?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E9=80=9A=E9=81=93=E5=B9=B6=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=E5=BC=82=E5=B8=B8=E8=BF=9E=E7=BB=AD=E6=B6=88=E8=B4=B9=E5=85=9C?= =?UTF-8?q?=E5=BA=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 128 KiB 及以上日志默认进入大日志 Topic - InfluxDB 原文及达梦大字段超过 256 KiB 时截断保存 - 截断内容追加 HB_XDR_TRUNCATED 标记 - InfluxDB 写入失败时继续执行规则解析和达梦写入 - 未知或系统性异常优先写入普通/大日志 DLT - DLT 写入失败时记录特殊告警并提交 offset，避免阻塞后续消息 - DLT 失败等待时间默认限制为 1 秒 - 增加连续性放行、DLT 成功及失败监控指标 - 更新 Docker Compose、v3 升级文档及相关测试 --- .../docker-compose-consumer-rule-prod.yaml | 7 +- .../Modules/NormalData/SysLogProcessor.java | 127 ++++++++++++++++-- .../NormalData/SyslogFailureRouter.java | 59 +++++++- .../common/service/DmLargeTextLimiter.java | 4 +- .../java/com/influx/raw/RawInfluxMetrics.java | 14 ++ .../com/influx/raw/RawStorageProperties.java | 2 +- .../src/main/resources/application.properties | 8 +- .../SysLogProcessorParallelTest.java | 68 +++++++++- .../SyslogFailureRouterReliabilityTest.java | 35 +++++ .../docker-compose-serve-prod.yaml | 4 +- .../com/haobang/config/AppProperties.java | 4 +- .../haobang/dispatch/SyslogWalDispatcher.java | 2 +- .../src/main/resources/application.properties | 4 +- .../dispatch/SyslogWalDispatcherTest.java | 4 +- haobang-security-dm/upgrade-doc/v3/README.md | 13 +- .../v3/config/01-new-environment-variables.md | 30 ++++- 16 files changed, 342 insertions(+), 43 deletions(-) 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 62f5ae0..036e952 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 @@ -58,6 +58,9 @@ services: - KAFKA_CONSUMER_LARGE_TOPIC=agent-01-syslog-topic.large # consumer-rule 是活跃原始日志消费节点，多实例通过同一消费组分担分区。 - SYSLOG_RAW_CONSUMER_ENABLED=true + # 特殊异常优先写入 DLT；DLT 失败也记录后放行，避免单条/单批异常永久阻塞分区。 + - SYSLOG_FAILURE_CONTINUITY_ENABLED=true + - SYSLOG_FAILURE_CONTINUITY_DLT_PUBLISH_TIMEOUT_MS=1000 # 大日志的 poll、fetch、批次和规则计算线程池均独立且有界。 - SYSLOG_LARGE_LISTENER_ENABLED=true - SYSLOG_LARGE_LISTENER_CONCURRENCY=1 @@ -91,9 +94,9 @@ services: - INFLUXDB_RAW_LEGACY_POINT_TIME_FALLBACK_ENABLED=true - INFLUXDB_RAW_QUERY_MAX_ROWS=10000 - INFLUXDB_RAW_QUERY_MAX_RESPONSE_BYTES=10485760 - - INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES=1048576 + - INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES=262144 - INFLUXDB_RAW_WRITE_BATCH_MAX_BYTES=8388608 - - DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES=65536 + - DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES=262144 - DM_NON_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES=262144 - INFLUXDB_BATCH_SIZE=1000 - INFLUXDB_FLUSH_INTERVAL=1000 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 c090903..5289b62 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 @@ -67,6 +67,9 @@ public class SysLogProcessor { @Value("${syslog.large-listener.processing-batch-max-bytes:8388608}") private long largeProcessingBatchMaxBytes; + @Value("${syslog.failure-continuity.enabled:true}") + private boolean failureContinuityEnabled; + @Autowired private AppProperties appProperties; @@ -178,7 +181,7 @@ public class SysLogProcessor { } for (List> chunk : partitionRecordsByBytes( records, Math.max(1L, largeProcessingBatchMaxBytes))) { - processBatch(chunk, enforceRetryDue, lane, largeTaskExecutor); + processBatchWithContinuity(chunk, enforceRetryDue, lane, largeTaskExecutor); } ack.acknowledge(); } @@ -201,8 +204,8 @@ public class SysLogProcessor { /** * 统一批处理边界：原文批量写入 InfluxDB 后，有界并行执行标准化。 - * 正常消息仍需到达可靠终态才提交 offset；非标消息按现场临时策略异步落库， - * 其写库结果不参与 Kafka ACK。 + * 失败时优先进入 Retry/DLT；最终兜底按连续性策略记录并推进 offset。非标消息按现场临时策略 + * 异步落库，其写库结果不参与 Kafka ACK。 */ private void consumeBatch(List> records, Acknowledgment ack, boolean enforceRetryDue, String lane, Executor normalizationExecutor) { @@ -210,12 +213,29 @@ public class SysLogProcessor { ack.acknowledge(); return; } - processBatch(records, enforceRetryDue, lane, normalizationExecutor); + processBatchWithContinuity(records, enforceRetryDue, lane, normalizationExecutor); ack.acknowledge(); } - private void processBatch(List> records, - boolean enforceRetryDue, String lane, Executor normalizationExecutor) { + private void processBatchWithContinuity(List> records, + boolean enforceRetryDue, String lane, + Executor normalizationExecutor) { + try { + processBatch(records, enforceRetryDue, lane, normalizationExecutor); + } catch (SyslogFailureRouter.RetryNotDueException notDue) { + // Retry Topic 的计划时间尚未到达，不属于处理异常，仍按原延时语义保留 offset。 + throw notDue; + } catch (RuntimeException failure) { + if (!failureContinuityEnabled) throw failure; + routeContinuityBatch(records, failure, "BATCH_" + lane); + log.error("[SYSLOG_CONTINUITY] 批次特殊异常已记录，继续消费后续消息: lane={}, " + + "records={}, errorType={}", + lane, records.size(), failure.getClass().getName(), failure); + } + } + + void processBatch(List> records, + boolean enforceRetryDue, String lane, Executor normalizationExecutor) { long batchStartedAt = System.currentTimeMillis(); long prepareStartedAt = batchStartedAt; BatchMetrics metrics = new BatchMetrics(); @@ -239,7 +259,21 @@ public class SysLogProcessor { int ingressGroups = ingressStatsProjectionWriter.writeBatch(ingressEvents(preparedMessages)); long influxStartedAt = System.currentTimeMillis(); - List persistedMessages = persistRawBatch(preparedMessages, metrics); + List persistedMessages; + try { + persistedMessages = persistRawBatch(preparedMessages, metrics); + } catch (RuntimeException rawStorageFailure) { + if (!failureContinuityEnabled) throw rawStorageFailure; + routeContinuityPrepared(preparedMessages, rawStorageFailure, "RAW_STORAGE"); + rawInfluxMetrics.rawWriteBypassed(preparedMessages.size()); + metrics.rawStorageBypassed.addAndGet(preparedMessages.size()); + // 无论 DLT 是否可用，都继续使用内存中的完整内容执行规则和达梦写入。 + persistedMessages = preparedMessages; + log.error("[SYSLOG_CONTINUITY] InfluxDB 原文写入失败，已尝试记录 DLT 并继续标准化: " + + "lane={}, records={}, errorType={}", + lane, preparedMessages.size(), rawStorageFailure.getClass().getName(), + rawStorageFailure); + } long influxMs = System.currentTimeMillis() - influxStartedAt; long normalizeStartedAt = System.currentTimeMillis(); @@ -249,7 +283,7 @@ public class SysLogProcessor { totalProcessed.addAndGet(records.size()); log.info("Syslog批次完成: lane={}, topicPartitions={}, records={}, prepared={}, persisted={}, " + "success={}, filtered={}, retry={}, dlt={}, normalizedRows={}, dmSuccessfulRows={}, " - + "dmFailedMessages={}, ingressGroups={}, prepareMs={}, influxMs={}, normalizeMs={}, " + + "dmFailedMessages={}, ingressGroups={}, rawStorageBypassed={}, prepareMs={}, influxMs={}, normalizeMs={}, " + "ruleComputeMs={}, ruleSnapshotMs={}, messageParseMs={}, fieldMappingMs={}, " + "filterCompleteMs={}, enrichmentMs={}, ruleAssembleMs={}, compiledRuleHit={}, " + "compiledRuleMiss={}, messageParseHit={}, messageParseMiss={}, deviceUnique={}, " @@ -259,6 +293,7 @@ public class SysLogProcessor { lane, topicPartitions(records), records.size(), preparedMessages.size(), persistedMessages.size(), metrics.success.get(), metrics.filtered.get(), metrics.retry.get(), metrics.dlt.get(), metrics.normalizedRows.get(), metrics.dmSuccessfulRows.get(), metrics.dmFailedMessages.get(), ingressGroups, + metrics.rawStorageBypassed.get(), prepareMs, influxMs, normalizeMs, metrics.ruleComputeMs(), metrics.ruleSnapshotMs(), metrics.messageParseMs(), metrics.fieldMappingMs(), metrics.filterCompleteMs(), metrics.enrichmentMs(), metrics.ruleAssembleMs(), metrics.compiledRuleHits(), @@ -271,6 +306,73 @@ public class SysLogProcessor { System.currentTimeMillis() - batchStartedAt); } + private void routeContinuityBatch(List> records, + RuntimeException failure, + String stage) { + int published = 0; + for (ConsumerRecord record : records) { + RuntimeException dltFailure = tryRouteContinuityRecord(record, null, failure, stage); + if (dltFailure != null) { + recordContinuityLoss(records.size(), published, record, failure, dltFailure, stage); + return; + } + published++; + } + rawInfluxMetrics.continuityDltPublished(published); + } + + private void routeContinuityPrepared(List messages, + RuntimeException failure, + String stage) { + int published = 0; + for (PreparedMessage message : messages) { + RuntimeException dltFailure = tryRouteContinuityRecord( + message.record, message.context, failure, stage); + if (dltFailure != null) { + recordContinuityLoss(messages.size(), published, message.record, + failure, dltFailure, stage); + return; + } + published++; + } + rawInfluxMetrics.continuityDltPublished(published); + } + + /** + * 连续性优先的最后边界。优先把原消息写入 DLT；即使 DLT 不可用，也只记录特殊告警和计数， + * 不再把异常抛回 Kafka listener，避免一个坏批次永久阻塞整个分区。 + */ + private RuntimeException tryRouteContinuityRecord(ConsumerRecord record, + SyslogFailureRouter.MessageContext context, + RuntimeException sourceFailure, + String stage) { + try { + SyslogFailureRouter.MessageContext resolved = context == null + ? failureRouter.resolve(record) : context; + failureRouter.routeContinuityDlt(record, resolved, sourceFailure, stage); + return null; + } catch (RuntimeException dltFailure) { + return dltFailure; + } + } + + private void recordContinuityLoss(int total, int published, + ConsumerRecord firstFailedRecord, + RuntimeException sourceFailure, + RuntimeException dltFailure, + String stage) { + if (published > 0) rawInfluxMetrics.continuityDltPublished(published); + int releasedWithoutDlt = total - published; + rawInfluxMetrics.continuityDltPublishFailed(releasedWithoutDlt); + // 首次失败后不再逐条等待或打印，剩余消息用批次计数记录，避免 Kafka 故障时放大阻塞和日志量。 + log.error("[SYSLOG_CONTINUITY_LOSS] 特殊异常写入 DLT 失败，但按连续性策略仍将提交原 offset: " + + "topic={}, partition={}, firstFailedOffset={}, stage={}, batchRecords={}, " + + "dltPublished={}, releasedWithoutDlt={}, sourceErrorType={}, dltErrorType={}", + firstFailedRecord.topic(), firstFailedRecord.partition(), firstFailedRecord.offset(), stage, + total, published, releasedWithoutDlt, sourceFailure.getClass().getName(), + dltFailure.getClass().getName(), dltFailure); + } + static List>> partitionRecordsByBytes( List> records, long maxBytes) { if (maxBytes <= 0L) throw new IllegalArgumentException("maxBytes 必须大于 0"); @@ -495,7 +597,7 @@ public class SysLogProcessor { /** * 一次 HTTP 请求持久化当前 Kafka 批次的原文。 * 批量请求如果因单条数据格式被 InfluxDB 拒绝，降级为逐条写入以定位坏消息； - * 连接、超时或服务端故障则直接保留原 offset，不制造 DLT 风暴。 + * 连接、超时或服务端故障会上抛到连续性边界，由其尝试 DLT 后继续标准化。 */ private List persistRawBatch(List messages, BatchMetrics metrics) { if (messages.isEmpty()) { @@ -536,7 +638,7 @@ public class SysLogProcessor { return persisted; } catch (Exception dependencyFailure) { if (messages.get(0).v2) rawInfluxMetrics.v2FailedBatch(); - // 整批原文的持久化结果不可确认，必须让 Kafka 从原 offset 重投。 + // 整批原文的持久化结果不可确认，交给外层连续性策略统一处理。 throw new IllegalStateException("InfluxDB 批量原文写入未取得成功确认，禁止提交 offset", dependencyFailure); } } @@ -595,7 +697,7 @@ public class SysLogProcessor { } /** - * 批内有界并行标准化。任务异常必须汇总回 listener；超时或系统性失败时整批不 ACK。 + * 批内有界并行标准化。任务异常汇总回 listener 外层的连续性边界。 */ void normalizePreparedMessages(List messages, BatchMetrics metrics) { normalizePreparedMessages(messages, metrics, taskExecutor); @@ -783,7 +885,7 @@ public class SysLogProcessor { } } - /** Retry/DLT 取得 Broker ACK 才返回；系统性故障抛出以保留原 offset。 */ + /** 常规 Retry/DLT 路由；未到终态的异常交给外层连续性边界记录并放行。 */ private SyslogFailureRouter.TerminalState routeFailure(ConsumerRecord record, SyslogFailureRouter.MessageContext context, Exception failure) { @@ -905,6 +1007,7 @@ public class SysLogProcessor { private final AtomicInteger normalizedRows = new AtomicInteger(); private final AtomicInteger dmSuccessfulRows = new AtomicInteger(); private final AtomicInteger dmFailedMessages = new AtomicInteger(); + private final AtomicInteger rawStorageBypassed = new AtomicInteger(); private final AtomicLong ruleComputeMs = new AtomicLong(); private final AtomicLong ruleSnapshotMs = new AtomicLong(); private final AtomicLong dmWriteMs = new AtomicLong(); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java index 44b4561..f3899ca 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java @@ -30,9 +30,8 @@ import java.util.concurrent.TimeoutException; /** * Kafka 异常消息可靠路由器。 * - *

该组件把“业务处理失败”与“原 offset 是否允许推进”放在同一个同步边界内： - * Retry/DLT 的 send Future 只有取得 Broker ACK 后才返回终态；发布失败或结果未知一律抛出， - * 让原消息保持未提交。这样不会用一次异步方法调用冒充消息已经被补偿链路接管。

+ *

常规 Retry/DLT 的 send Future 只有取得 Broker ACK 后才返回终态；发布失败或结果未知会抛给 + * 调用方。活动 listener 外层另有连续性优先边界，可按现场策略记录后放行。

*/ @Slf4j @Component @@ -65,6 +64,8 @@ public class SyslogFailureRouter { private String largeDltTopic; @Value("${syslog.reliability.publish-timeout-ms:30000}") private long publishTimeoutMs; + @Value("${syslog.failure-continuity.dlt-publish-timeout-ms:1000}") + private long continuityDltPublishTimeoutMs; @Value("${syslog.reliability.consumer-version:unknown}") private String consumerVersion; @@ -132,7 +133,7 @@ public class SyslogFailureRouter { public TerminalState route(ConsumerRecord record, MessageContext context, Throwable failure) { FailureCategory category = classify(failure); if (category == FailureCategory.SYSTEMIC) { - // [SYSLOG-ALARM][KACK][已修复] 系统性依赖故障不倾倒 DLT，也不提交原 offset，由容器背压并等待恢复。 + // 常规路由把系统性故障上抛；活动 listener 的连续性边界再决定是否转 DLT 并放行。 return TerminalState.PAUSED; } @@ -179,11 +180,59 @@ public class SyslogFailureRouter { return state; } + /** + * 连续性优先兜底：把无法安全完成当前处理阶段的原消息直接写入 DLT。 + * 本方法仍显式报告发布失败；是否继续提交原 offset 由调用方的连续性策略决定。 + */ + public TerminalState routeContinuityDlt(ConsumerRecord record, + MessageContext context, + Throwable failure, + String failedStage) { + int nextAttempt = Math.max(0, context.getAttemptCount()) + 1; + boolean largeLane = largeOriginalTopic != null + && largeOriginalTopic.equals(context.getOriginalTopic()); + String targetTopic = largeLane ? largeDltTopic : dltTopic; + FailureEnvelope source = readEnvelope(record.value()); + FailureEnvelope envelope = new FailureEnvelope(); + envelope.setEnvelopeType(ENVELOPE_TYPE); + envelope.setEventId(context.getEventId()); + envelope.setDeviceId(context.getDeviceId() != null ? context.getDeviceId() + : (source == null ? null : source.getDeviceId())); + envelope.setOriginalTopic(source == null ? record.topic() : source.getOriginalTopic()); + envelope.setOriginalPartition(source == null ? record.partition() : source.getOriginalPartition()); + envelope.setOriginalOffset(source == null ? record.offset() : source.getOriginalOffset()); + envelope.setCurrentTopic(record.topic()); + envelope.setFirstReceivedAt(context.getFirstReceivedAt()); + envelope.setFailedStage(failedStage == null ? "UNKNOWN" : failedStage); + envelope.setErrorCode("SYSLOG_CONTINUITY_BYPASS"); + Throwable root = rootCause(failure); + envelope.setErrorType(root.getClass().getName()); + envelope.setErrorMessage(sanitize(root.getMessage())); + envelope.setAttemptCount(nextAttempt); + envelope.setNextRetryAt(null); + envelope.setRuleId(source == null ? null : source.getRuleId()); + envelope.setRuleVersion(source == null ? null : source.getRuleVersion()); + envelope.setConsumerVersion(consumerVersion); + envelope.setOriginalPayload(context.getPayload()); + envelope.setPointTimeNs(context.getPointTimeNs()); + + publishAndAwait(targetTopic, context.getEventId(), envelope, + Math.max(1L, continuityDltPublishTimeoutMs)); + log.debug("[SYSLOG_CONTINUITY] 特殊异常已可靠写入 DLT: eventId={}, " + + "stage={}, target={}, errorType={}", + context.getEventId(), envelope.getFailedStage(), targetTopic, envelope.getErrorType()); + return TerminalState.DLT_PUBLISHED; + } + private void publishAndAwait(String topic, String key, FailureEnvelope envelope) { + publishAndAwait(topic, key, envelope, publishTimeoutMs); + } + + private void publishAndAwait(String topic, String key, FailureEnvelope envelope, long timeoutMs) { try { String body = objectMapper.writeValueAsString(envelope); // [流程节点4.3] 必须等待 Broker ACK；仅调用 send() 不构成 RETRY_PUBLISHED/DLT_PUBLISHED。 - kafkaTemplate.send(topic, key, body).get(publishTimeoutMs, TimeUnit.MILLISECONDS); + kafkaTemplate.send(topic, key, body).get(timeoutMs, TimeUnit.MILLISECONDS); } catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); throw new FailurePublishException("发布异常消息时线程被中断，原 offset 不得提交", interrupted); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmLargeTextLimiter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmLargeTextLimiter.java index f21003f..6b20561 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmLargeTextLimiter.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmLargeTextLimiter.java @@ -18,13 +18,13 @@ import java.util.concurrent.atomic.LongAdder; public class DmLargeTextLimiter { private static final long MIN_ENABLED_LIMIT = 64L; - @Value("${dm.normal.large-field-max-utf8-bytes:0}") + @Value("${dm.normal.large-field-max-utf8-bytes:262144}") private long normalMaxBytes; @Value("${dm.normal.large-fields:http_req_body,http_resp_body,http_req_header,http_resp_header,detail,description,cmdline,shell_cmdline,parent_cmdline,mail_message}") private String normalFieldsValue; - @Value("${dm.non-normal.large-field-max-utf8-bytes:0}") + @Value("${dm.non-normal.large-field-max-utf8-bytes:262144}") private long nonNormalMaxBytes; private volatile Set normalFields = Collections.emptySet(); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawInfluxMetrics.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawInfluxMetrics.java index aa052f2..a454f80 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawInfluxMetrics.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawInfluxMetrics.java @@ -19,6 +19,10 @@ public class RawInfluxMetrics { private final LongAdder messagesTruncated = new LongAdder(); private final LongAdder originalBytesTruncated = new LongAdder(); private final LongAdder storedBytesAfterTruncation = new LongAdder(); + private final LongAdder rawWriteBypassedBatches = new LongAdder(); + private final LongAdder rawWriteBypassedMessages = new LongAdder(); + private final LongAdder continuityDltPublished = new LongAdder(); + private final LongAdder continuityDltPublishFailed = new LongAdder(); public void legacyMissing() { legacyMissing.increment(); } public void legacyAssigned() { legacyAssigned.increment(); } @@ -32,6 +36,12 @@ public class RawInfluxMetrics { originalBytesTruncated.add(originalBytes); storedBytesAfterTruncation.add(storedBytes); } + public void rawWriteBypassed(long messages) { + rawWriteBypassedBatches.increment(); + rawWriteBypassedMessages.add(messages); + } + public void continuityDltPublished(long messages) { continuityDltPublished.add(messages); } + public void continuityDltPublishFailed(long messages) { continuityDltPublishFailed.add(messages); } public Map snapshot() { Map values = new LinkedHashMap<>(); @@ -47,6 +57,10 @@ public class RawInfluxMetrics { values.put("raw_message_truncated_total", messagesTruncated.sum()); values.put("raw_message_truncated_original_bytes_total", originalBytesTruncated.sum()); values.put("raw_message_truncated_stored_bytes_total", storedBytesAfterTruncation.sum()); + values.put("raw_write_bypassed_batches_total", rawWriteBypassedBatches.sum()); + values.put("raw_write_bypassed_messages_total", rawWriteBypassedMessages.sum()); + values.put("continuity_dlt_published_total", continuityDltPublished.sum()); + values.put("continuity_dlt_publish_failed_total", continuityDltPublishFailed.sum()); return values; } } 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 5f3d6b6..f69c2cd 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 @@ -20,7 +20,7 @@ public class RawStorageProperties { @Value("${influxdb.raw.query-max-rows:10000}") private int queryMaxRows; @Value("${influxdb.raw.query-max-response-bytes:10485760}") private long queryMaxResponseBytes; /** 原始 message 单字段的 UTF-8 存储上限；0 表示不限制。 */ - @Value("${influxdb.raw.message-max-utf8-bytes:0}") private long messageMaxUtf8Bytes; + @Value("${influxdb.raw.message-max-utf8-bytes:262144}") private long messageMaxUtf8Bytes; /** 一次同步写请求的估算 Line Protocol 上限；按字节切批，不能只按条数。 */ @Value("${influxdb.raw.write-batch-max-bytes:8388608}") private long writeBatchMaxBytes; 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 922b5f5..f1cb58f 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 @@ -35,7 +35,7 @@ influxdb.raw.legacy-point-time-fallback-enabled=${INFLUXDB_RAW_LEGACY_POINT_TIME influxdb.raw.query-max-rows=${INFLUXDB_RAW_QUERY_MAX_ROWS:10000} influxdb.raw.query-max-response-bytes=${INFLUXDB_RAW_QUERY_MAX_RESPONSE_BYTES:10485760} # 0 表示不限制；启用后最终 message（含截断标记）不超过该 UTF-8 字节数。 -influxdb.raw.message-max-utf8-bytes=${INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES:1048576} +influxdb.raw.message-max-utf8-bytes=${INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES:262144} # Influx 同步写按 Line Protocol 估算字节切批，避免少量大日志形成超大 HTTP 请求。 influxdb.raw.write-batch-max-bytes=${INFLUXDB_RAW_WRITE_BATCH_MAX_BYTES:8388608} # InfluxDB 2.7 连接超时配置 @@ -79,6 +79,10 @@ 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} +# 连续性优先：特殊/系统性异常优先写 DLT；即使 DLT 失败也记录告警并推进原 offset，避免阻塞分区。 +syslog.failure-continuity.enabled=${SYSLOG_FAILURE_CONTINUITY_ENABLED:true} +# DLT 故障时最多快速等待 1 秒；首次失败后整批直接记录并放行。 +syslog.failure-continuity.dlt-publish-timeout-ms=${SYSLOG_FAILURE_CONTINUITY_DLT_PUBLISH_TIMEOUT_MS:1000} # 大日志使用独立 Topic、消费组和 ConsumerFactory；建议生产环境独立 JVM 部署。 syslog.large-listener.enabled=${SYSLOG_LARGE_LISTENER_ENABLED:true} syslog.large-listener.group-id=${SYSLOG_LARGE_LISTENER_GROUP_ID:${spring.kafka.consumer.group-id}-large} @@ -180,7 +184,7 @@ dm.standardized-write.batch-size=${DM_STANDARDIZED_WRITE_BATCH_SIZE:500} dm.standardized-write.writer-concurrency=2 dm.standardized-write.writer-queue-capacity=100 # 标准化大字段和非标原文均在达梦持久化边界截断；0 表示不限制。 -dm.normal.large-field-max-utf8-bytes=${DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES:65536} +dm.normal.large-field-max-utf8-bytes=${DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES:262144} dm.normal.large-fields=${DM_NORMAL_LARGE_FIELDS:http_req_body,http_resp_body,http_req_header,http_resp_header,detail,description,cmdline,shell_cmdline,parent_cmdline,mail_message} dm.non-normal.large-field-max-utf8-bytes=${DM_NON_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES:262144} dm.standardized-write.transaction-timeout-seconds=45 diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java index ade3588..5b569f3 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java @@ -8,12 +8,14 @@ import com.common.service.DeviceCollectTaskService; import com.common.service.DeviceDeviceService; import com.common.service.DmStandardizedBatchWriter; import com.common.service.SyslogNonNormalMessageService; +import com.influx.raw.RawInfluxMetrics; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.kafka.support.Acknowledgment; import java.time.Instant; import java.time.LocalDateTime; @@ -21,6 +23,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; @@ -31,13 +34,17 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.never; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.when; -/** 验证恢复并行后仍保留“全部可靠终态才允许返回”的处理边界。 */ +/** 验证批内并行、常规可靠路由及连续性优先的最终放行边界。 */ class SysLogProcessorParallelTest { private ExecutorService executor; @@ -47,6 +54,7 @@ class SysLogProcessorParallelTest { private SyslogFailureRouter failureRouter; private DmStandardizedBatchWriter batchWriter; private SyslogNonNormalMessageService nonNormalMessageService; + private RawInfluxMetrics rawInfluxMetrics; @BeforeEach void setUp() { @@ -57,6 +65,7 @@ class SysLogProcessorParallelTest { failureRouter = mock(SyslogFailureRouter.class); batchWriter = mock(DmStandardizedBatchWriter.class); nonNormalMessageService = mock(SyslogNonNormalMessageService.class); + rawInfluxMetrics = mock(RawInfluxMetrics.class); ReflectionTestUtils.setField(processor, "taskExecutor", executor); ReflectionTestUtils.setField(processor, "processTimeoutMs", 5000L); ReflectionTestUtils.setField(processor, "logNormalProcessor", normalProcessor); @@ -66,6 +75,8 @@ class SysLogProcessorParallelTest { ReflectionTestUtils.setField(processor, "nonNormalWriterExecutor", nonNormalExecutor); ReflectionTestUtils.setField(processor, "nonNormalMaxConcurrencyRetries", 3); ReflectionTestUtils.setField(processor, "nonNormalConcurrencyRetryBackoffMs", 0L); + ReflectionTestUtils.setField(processor, "failureContinuityEnabled", true); + ReflectionTestUtils.setField(processor, "rawInfluxMetrics", rawInfluxMetrics); when(batchWriter.write(any())).thenReturn(new DmStandardizedBatchWriter.WriteReport<>()); when(normalProcessor.captureRuleSnapshot(any())).thenReturn( new BatchRuleSnapshot(Collections.emptyMap(), System.currentTimeMillis())); @@ -232,6 +243,61 @@ class SysLogProcessorParallelTest { assertEquals(3, metrics.filteredCount()); } + @Test + void unexpectedBatchFailureIsRecordedAndOffsetIsAcknowledged() { + SysLogProcessor spyProcessor = spy(processor); + RuntimeException sourceFailure = new IllegalStateException("unknown processing failure"); + doThrow(sourceFailure).when(spyProcessor) + .processBatch(anyList(), eq(false), eq("RAW"), any(Executor.class)); + when(failureRouter.resolve(any())).thenAnswer(invocation -> { + ConsumerRecord record = invocation.getArgument(0); + return new SyslogFailureRouter.MessageContext(record.value(), "event-" + record.offset(), + 0, null, Instant.now(), record.topic(), null); + }); + when(failureRouter.routeContinuityDlt(any(), any(), eq(sourceFailure), eq("BATCH_RAW"))) + .thenReturn(SyslogFailureRouter.TerminalState.DLT_PUBLISHED); + Acknowledgment ack = mock(Acknowledgment.class); + + spyProcessor.consumeAndParallelProcess(records(2), ack); + + verify(failureRouter, times(2)) + .routeContinuityDlt(any(), any(), eq(sourceFailure), eq("BATCH_RAW")); + verify(rawInfluxMetrics).continuityDltPublished(2L); + verify(ack).acknowledge(); + } + + @Test + void dltPublishFailureIsLoggedAndOffsetIsStillAcknowledged() { + SysLogProcessor spyProcessor = spy(processor); + RuntimeException sourceFailure = new IllegalStateException("unknown processing failure"); + doThrow(sourceFailure).when(spyProcessor) + .processBatch(anyList(), eq(false), eq("RAW"), any(Executor.class)); + when(failureRouter.resolve(any())).thenAnswer(invocation -> { + ConsumerRecord record = invocation.getArgument(0); + return new SyslogFailureRouter.MessageContext(record.value(), "event-" + record.offset(), + 0, null, Instant.now(), record.topic(), null); + }); + when(failureRouter.routeContinuityDlt(any(), any(), eq(sourceFailure), eq("BATCH_RAW"))) + .thenThrow(new SyslogFailureRouter.FailurePublishException( + "broker unavailable", new IllegalStateException("send failed"))); + Acknowledgment ack = mock(Acknowledgment.class); + + spyProcessor.consumeAndParallelProcess(records(2), ack); + + verify(failureRouter) + .routeContinuityDlt(any(), any(), eq(sourceFailure), eq("BATCH_RAW")); + verify(rawInfluxMetrics).continuityDltPublishFailed(2L); + verify(ack).acknowledge(); + } + + private List> records(int count) { + List> records = new ArrayList<>(); + for (int i = 0; i < count; i++) { + records.add(new ConsumerRecord<>("raw", 0, i, "key-" + i, "ciphertext-" + i)); + } + return records; + } + private List messages(int count) { List messages = new ArrayList<>(); for (int i = 0; i < count; i++) { diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogFailureRouterReliabilityTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogFailureRouterReliabilityTest.java index 105fc66..84379ab 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogFailureRouterReliabilityTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogFailureRouterReliabilityTest.java @@ -117,6 +117,41 @@ class SyslogFailureRouterReliabilityTest { () -> router.route(record, router.resolve(record), new SocketException("reset"))); } + @Test + void continuityFailureIsWrittenDirectlyToDltWithOriginalPayload() throws Exception { + SettableListenableFuture> acknowledged = new SettableListenableFuture<>(); + acknowledged.set(null); + when(template.send(eq("raw.DLT"), anyString(), anyString())).thenReturn(acknowledged); + ArgumentCaptor body = ArgumentCaptor.forClass(String.class); + + SyslogFailureRouter.TerminalState state = router.routeContinuityDlt( + record, router.resolve(record), new InfluxException("influx unavailable"), "RAW_STORAGE"); + + assertEquals(SyslogFailureRouter.TerminalState.DLT_PUBLISHED, state); + verify(template).send(eq("raw.DLT"), anyString(), body.capture()); + SyslogFailureRouter.FailureEnvelope envelope = new ObjectMapper().findAndRegisterModules() + .readValue(body.getValue(), SyslogFailureRouter.FailureEnvelope.class); + assertEquals("SYSLOG_CONTINUITY_BYPASS", envelope.getErrorCode()); + assertEquals("RAW_STORAGE", envelope.getFailedStage()); + assertEquals("ciphertext", envelope.getOriginalPayload()); + assertEquals("raw", envelope.getOriginalTopic()); + assertEquals(42L, envelope.getOriginalOffset()); + } + + @Test + void continuityFailureFromLargeLaneUsesLargeDlt() { + ConsumerRecord largeRecord = new ConsumerRecord<>( + "raw.large", 1, 7L, "event", "large-ciphertext"); + SettableListenableFuture> acknowledged = new SettableListenableFuture<>(); + acknowledged.set(null); + when(template.send(eq("raw.large.DLT"), anyString(), anyString())).thenReturn(acknowledged); + + router.routeContinuityDlt(largeRecord, router.resolve(largeRecord), + new IllegalStateException("unknown"), "BATCH_RAW_LARGE"); + + verify(template).send(eq("raw.large.DLT"), anyString(), anyString()); + } + @Test void influxDependencyFailurePausesWithoutDltStorm() { SyslogFailureRouter.TerminalState state = router.route( diff --git a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml index b9be4ba..e0269a0 100644 --- a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml +++ b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml @@ -23,10 +23,10 @@ services: - SYSLOG_SERVER_ENABLED=true # 设备上送日志为 UTF-8；显式声明，Netty 与 WAL 共用 Spring 配置。 - SYSLOG_CHARSET=UTF-8 - # 256KiB 以上走大日志 Topic，单条最多接纳 6MiB；WAL/Dispatcher 均按字节背压。 + # 128KiB 及以上走大日志 Topic，单条最多接纳 6MiB；WAL/Dispatcher 均按字节背压。 - SYSLOG_MAX_FRAME_LENGTH=6291456 - SYSLOG_MESSAGE_MAX_BYTES=6291456 - - SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES=262144 + - SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES=131072 - SYSLOG_WAL_MAX_PENDING_BYTES=268435456 - SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES=268435456 - TZ=Asia/Shanghai diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java index 41072a0..8686589 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java @@ -26,8 +26,8 @@ public class AppProperties { @Value("${syslog.message.max-bytes:6291456}") private long syslogMessageMaxBytes; - /** 超过该值投递大日志 Topic；0 表示关闭分流。 */ - @Value("${syslog.large-message.threshold-bytes:262144}") + /** 达到该值投递大日志 Topic；0 表示关闭分流。 */ + @Value("${syslog.large-message.threshold-bytes:131072}") private long syslogLargeMessageThresholdBytes; @Value("${syslog.buffer.size:1024}") diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java index 969bb4b..97a7789 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java @@ -167,7 +167,7 @@ public class SyslogWalDispatcher { // v2/v3 继续调用旧重载，确保升级前已落盘记录产生的 Kafka 消息完全不变。 boolean large = properties.getSyslogLargeMessageThresholdBytes() > 0L - && record.getDispatchPayloadBytes() > properties.getSyslogLargeMessageThresholdBytes(); + && record.getDispatchPayloadBytes() >= properties.getSyslogLargeMessageThresholdBytes(); java.util.concurrent.CompletableFuture deliveryFuture = large ? producer.messagePushLarge( resolution.getDevice(), record.getRawMessage(), receiveTime, eventTime, diff --git a/haobang-security-dm/syslog-serve/src/main/resources/application.properties b/haobang-security-dm/syslog-serve/src/main/resources/application.properties index 78db766..e84cb88 100644 --- a/haobang-security-dm/syslog-serve/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-serve/src/main/resources/application.properties @@ -23,8 +23,8 @@ spring.kafka.producer.ack-timeout-ms=130000 syslog.max.frame.length=${SYSLOG_MAX_FRAME_LENGTH:6291456} # 单条原文硬上限；0 表示不限制。超过阈值的消息在进入 WAL 前被明确拒绝。 syslog.message.max-bytes=${SYSLOG_MESSAGE_MAX_BYTES:6291456} -# 超过 256KiB 的已接纳消息投递独立大日志 Topic；0 表示关闭分流。 -syslog.large-message.threshold-bytes=${SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES:262144} +# 达到 128KiB 的已接纳消息投递独立大日志 Topic；0 表示关闭分流。 +syslog.large-message.threshold-bytes=${SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES:131072} syslog.buffer.size=1000 # syslog 消息字符编码，国内安全设备普遍使用 GBK，如需 UTF-8 改为 syslog.charset=UTF-8 syslog.charset=GBK diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java index d561b3d..35dc984 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java @@ -130,7 +130,7 @@ class SyslogWalDispatcherTest { properties.setSyslogDispatchMaxInflightBytes(64L); wal = new SyslogWalService(properties, new ObjectMapper()); wal.initialize(); - wal.tryAppend(event("20260811-large", "12345")); + wal.tryAppend(event("20260811-large", "1234")); waitUntil(() -> wal.getAppendedCount() == 1L); DeviceDevice device = new DeviceDevice(); @@ -147,7 +147,7 @@ class SyslogWalDispatcherTest { dispatcher.start(); waitUntil(() -> wal.getCheckpointSequence() == 1L); - verify(producer).messagePushLarge(eq(device), eq("12345"), + verify(producer).messagePushLarge(eq(device), eq("1234"), eq("20260811200000000"), eq("20260811200000000"), eq("20260811-large"), eq(null)); verify(producer, never()).messagePush(any(), anyString(), anyString(), anyString(), any()); diff --git a/haobang-security-dm/upgrade-doc/v3/README.md b/haobang-security-dm/upgrade-doc/v3/README.md index 28d5d5e..fa76d1d 100644 --- a/haobang-security-dm/upgrade-doc/v3/README.md +++ b/haobang-security-dm/upgrade-doc/v3/README.md @@ -16,10 +16,12 @@ v3 同时增加大日志独立处理链路： - `syslog-serve` 在进入 WAL 前执行单条硬上限检查；超过上限明确拒绝，不把原文写入应用日志； - 接纳后的消息仍先进入 `syslog-serve` 本地 WAL，WAL 入队和 Kafka 投递同时受条数、字节数约束； -- 超过分流阈值的消息进入独立 `.large` Topic，其 Retry、DLT、Replay 也使用独立 Topic； +- 达到分流阈值的消息进入独立 `.large` Topic，其 Retry、DLT、Replay 也使用独立 Topic； - `syslog-consumer-rule` 使用独立的大日志 listener、消费组、ConsumerFactory、poll 批量和规则计算线程池； - 规则解析使用完整内容，写 InfluxDB 和达梦前才按 UTF-8 字节数截断，结尾带 `HB_XDR_TRUNCATED` 标记；InfluxDB 另存原长度、保存长度和原文 SHA-256。 +- 消费端默认采用连续性优先策略：无法理解或系统性异常先尝试写入 DLT；即使 DLT 失败，也记录 + `[SYSLOG_CONTINUITY_LOSS]` 后推进 offset，避免单条异常永久阻塞整个分区。 ## 文件与执行顺序 @@ -173,9 +175,16 @@ ORDER BY status; `dispatch_inflight_bytes`、`kafka_large_dispatched`； - `GET /xdrrule/api/syslog/raw-storage/metrics`：检查 `raw_message_truncated_total` 及截断前后字节累计值； - `GET /xdrrule/api/syslog/storage-limits/metrics`：检查达梦正常/非标字段截断计数； -- 发送一条略大于 256 KiB、但小于 6 MiB 的测试消息，确认 `.large` Topic 有数据且普通 Topic 无该事件； +- 发送一条不小于 128 KiB、但小于 6 MiB 的测试消息，确认 `.large` Topic 有数据且普通 Topic 无该事件； - 查询 InfluxDB/达梦，确认被截断值以 `...[HB_XDR_TRUNCATED original_utf8_bytes=...]` 结尾。 +连续性指标也位于 `GET /xdrrule/api/syslog/raw-storage/metrics`： + +- `raw_write_bypassed_messages_total`：InfluxDB 原文写失败后仍继续标准化的消息数； +- `continuity_dlt_published_total`：特殊异常成功写入 DLT 的消息数； +- `continuity_dlt_publish_failed_total`：DLT 写失败但仍放行的消息数。该值增加时应立即检查 Kafka/DLT， + 对应消息可能无法自动重放。 + ## 回滚 回滚最后一台新版本前，先确认 Outbox 没有未应用事件： diff --git a/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md b/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md index 727c85a..84072b7 100644 --- a/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md +++ b/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md @@ -18,10 +18,10 @@ syslog-serve/docker-compose/docker-compose-serve-prod.yaml - SYSLOG_SERVER_ENABLED=true - SYSLOG_CHARSET=UTF-8 - # 单条最多接收 6 MiB，超过 256 KiB 时发送到大日志 Topic。 + # 单条最多接收 6 MiB，达到 128 KiB 时发送到大日志 Topic。 - SYSLOG_MAX_FRAME_LENGTH=6291456 - SYSLOG_MESSAGE_MAX_BYTES=6291456 - - SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES=262144 + - SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES=131072 # WAL 和 Kafka 投递最多各占用 256 MiB 在途内存。 - SYSLOG_WAL_MAX_PENDING_BYTES=268435456 @@ -36,7 +36,7 @@ syslog-serve/docker-compose/docker-compose-serve-prod.yaml - `SYSLOG_MAX_FRAME_LENGTH`：TCP 单帧最大长度； - `SYSLOG_MESSAGE_MAX_BYTES`：所有来源的单条日志硬上限，超过后拒绝； -- `SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES`：超过该值后走大日志 Topic； +- `SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES`：达到该值后走大日志 Topic； - `SYSLOG_WAL_MAX_PENDING_BYTES`、`SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES`：限制 JVM 在途数据量。 `SYSLOG_MAX_FRAME_LENGTH` 必须大于或等于 `SYSLOG_MESSAGE_MAX_BYTES`。如果设备发送 GBK，应将 @@ -70,12 +70,16 @@ syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml # Retry、DLT、Replay 也允许传输大日志。 - KAFKA_RELIABILITY_MAX_REQUEST_SIZE=33554432 - # InfluxDB 原文最多保存 1 MiB，单批写入最多 8 MiB。 - - INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES=1048576 + # 连续性优先：特殊异常即使无法写入 DLT，也记录告警并继续消费。 + - SYSLOG_FAILURE_CONTINUITY_ENABLED=true + - SYSLOG_FAILURE_CONTINUITY_DLT_PUBLISH_TIMEOUT_MS=1000 + + # InfluxDB 原文最多保存 256 KiB，单批写入最多 8 MiB。 + - INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES=262144 - INFLUXDB_RAW_WRITE_BATCH_MAX_BYTES=8388608 - # 达梦正常日志大字段最多保存 64 KiB，非标日志原文最多保存 256 KiB。 - - DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES=65536 + # 达梦正常日志大字段和非标日志原文最多保存 256 KiB。 + - DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES=262144 - DM_NON_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES=262144 ``` @@ -95,6 +99,18 @@ DM_NON_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES 规则解析始终使用完整原文，只在写入 InfluxDB 或达梦时截断。 +`SYSLOG_FAILURE_CONTINUITY_ENABLED=true` 时： + +- InfluxDB 原文写入失败，会先尝试把原消息写入普通/大日志 DLT，再继续规则解析和达梦写入； +- 其他无法完成的特殊异常，会先尝试写 DLT，然后提交原 Kafka offset，避免阻塞整个分区； +- 如果 DLT 也不可用，仍会输出 `[SYSLOG_CONTINUITY_LOSS]` 日志并提交 offset。此时消息可能无法重放， + 但不会阻塞后续新消息。 + +DLT 首次写入失败最多等待 `SYSLOG_FAILURE_CONTINUITY_DLT_PUBLISH_TIMEOUT_MS` 毫秒；随后不再逐条等待， +整批记录一次后立即放行。 + +设置为 `false` 可恢复原来的失败保留 offset 行为。 + ## 3. 日统计投影 日统计投影默认开启，一般不需要新增配置。只需确认现场没有配置：