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 c718d5e..2fbc046 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 @@ -8,6 +8,7 @@ import com.common.util.SyslogParser; import com.config.AppProperties; import com.config.KafkaConsumerProperties; import com.common.service.DmStandardizedBatchWriter; +import com.common.service.RawLaneFieldLimiter; import com.common.service.DirectStatsProjectionWriter; import com.common.service.SyslogNonNormalMessageService; import com.common.service.SyslogNormalDataService; @@ -54,6 +55,9 @@ import com.common.util.Sm4Util; @Component public class SysLogProcessor { + @Autowired + private RawLaneFieldLimiter rawLaneFieldLimiter; + @Autowired @Qualifier("logNormalProcessorExecutor") private Executor taskExecutor; @@ -85,7 +89,7 @@ public class SysLogProcessor { @Value("${syslog.failure-continuity.enabled:true}") private boolean failureContinuityEnabled; - @Value("${syslog.slow-batch.threshold-ms:500}") + @Value("${syslog.slow-batch.threshold-ms:1000}") private long slowBatchThresholdMs; /** listener、业务处理和 Kafka commit 均在同一个 consumer 线程，ThreadLocal 用于拼接端到端慢批次。 */ @@ -1120,15 +1124,25 @@ public class SysLogProcessor { continue; } + boolean limitRawFields = rawLaneFieldLimiter.appliesTo(prepared.message.record.topic(), + prepared.message.context == null ? null : prepared.message.context.getOriginalTopic()); if (retained.keepNonNormal) { + if (limitRawFields) { + prepared.result.getNonNormalMessages().forEach(rawLaneFieldLimiter::truncateNonNormal); + } nonNormalMessages.addAll(prepared.result.getNonNormalMessages()); } int retainedStandardRows = 0; for (Map row : prepared.result.getRows()) { boolean alarm = syslogNormalDataService.isAlarmData(row); if (alarm || retained.keepNormal) { + Map storedRow = row; + if (limitRawFields && !alarm) { + storedRow = new LinkedHashMap<>(row); + rawLaneFieldLimiter.truncateRow(storedRow); + } DmStandardizedBatchWriter.WriteRequest request = - new DmStandardizedBatchWriter.WriteRequest<>(prepared.message, row); + new DmStandardizedBatchWriter.WriteRequest<>(prepared.message, storedRow); if (alarm) alarmWriteRequests.add(request); else normalWriteRequests.add(request); retainedStandardRows++; diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmPreparedBatchInserter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmPreparedBatchInserter.java index e5f38b1..03282a7 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmPreparedBatchInserter.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmPreparedBatchInserter.java @@ -46,20 +46,33 @@ public class DmPreparedBatchInserter { public BatchExecution insert(boolean alarmTable, List> rows, int statementTimeoutSeconds) { + return insert(alarmTable, rows, statementTimeoutSeconds, new DmWriteSizeEstimate()); + } + + BatchExecution insert(boolean alarmTable, List> rows, + int statementTimeoutSeconds, DmWriteSizeEstimate writeSize) { if (rows == null || rows.isEmpty()) { return BatchExecution.EMPTY; } final String table = alarmTable ? ALARM_TABLE : NORMAL_TABLE; return jdbcTemplate.execute((ConnectionCallback) connection -> - execute(connection, table, rows, statementTimeoutSeconds)); + execute(connection, table, rows, statementTimeoutSeconds, writeSize)); } BatchExecution execute(Connection connection, String table, List> rows, int statementTimeoutSeconds) throws SQLException { + return execute(connection, table, rows, statementTimeoutSeconds, new DmWriteSizeEstimate()); + } + + BatchExecution execute(Connection connection, String table, List> rows, + int statementTimeoutSeconds, DmWriteSizeEstimate writeSize) throws SQLException { long planStarted = System.nanoTime(); TableMetadata metadata = metadata(connection, table); BatchPlan plan = buildPlan(table, rows, metadata); long planMs = elapsedMillis(planStarted); + // 使用元数据白名单选中的列，排除规则携带但不会实际绑定的字段。 + // 执行前完成测量，使JDBC绑定/执行失败时仍可输出这次尝试的数据量。 + writeSize.measure(rows, plan.columns); long bindMs; long executeMs; diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java index 044bd1e..0a10dec 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java @@ -36,7 +36,7 @@ public class DmStandardizedBatchExecutor { @org.springframework.beans.factory.annotation.Value("${stats.direct.transaction-log-enabled:false}") private boolean transactionLogEnabled; - @org.springframework.beans.factory.annotation.Value("${syslog.slow-batch.threshold-ms:500}") + @org.springframework.beans.factory.annotation.Value("${syslog.slow-batch.threshold-ms:1000}") private long slowTransactionThresholdMs; @org.springframework.beans.factory.annotation.Value("${stats.direct.max-concurrency-retries:1}") @@ -118,7 +118,7 @@ public class DmStandardizedBatchExecutor { // 事实表使用一个固定 PreparedStatement 直接 addBatch，彻底绕开逐行 OGNL 动态 SQL。 timing.phase("JDBC_FACT_WRITE"); timing.batchExecution = preparedBatchInserter.insert( - alarmTable, rows, transactionTimeoutSeconds); + alarmTable, rows, transactionTimeoutSeconds, timing.writeSize); // 兼容期变更日志仍与事实行共用当前事务；默认关闭时完全不创建 MyBatis BatchSession。 if (statsChangeCaptureService.isEnabled()) { @@ -145,11 +145,11 @@ public class DmStandardizedBatchExecutor { } catch (RuntimeException failure) { log.error("[SLOW_DM_TRANSACTION] batchId={}, table={}, factRows={}, status=FAILED, " + "failureStage={}, phaseElapsedMs={}, transactionMs={}, connectionBeginMs={}, " - + "jdbcExecuteMs={}, lateAlarmBucketMs={}, retry={}, errorType={}", + + "jdbcExecuteMs={}, lateAlarmBucketMs={}, retry={}, errorType={}, {}", org.slf4j.MDC.get("syslogBatchId"), alarmTable ? "syslog_normal_alarm" : "syslog_normal_data", rows.size(), timing.phase, elapsedMs(timing.phaseStarted), elapsedMs(transactionStarted), timing.beginMs, timing.batchExecution.getExecuteMs(), timing.lateBucketMs, retries, - failure.getClass().getName()); + failure.getClass().getName(), timing.writeSize); throw failure; } long transactionMs = elapsedMs(transactionStarted); @@ -159,22 +159,22 @@ public class DmStandardizedBatchExecutor { if (slowTransactionThresholdMs > 0 && transactionMs >= slowTransactionThresholdMs) { log.warn("[SLOW_DM_TRANSACTION] batchId={}, table={}, factRows={}, status=SUCCESS, connectionBeginMs={}, jdbcPlanMs={}, " + "jdbcBindMs={}, jdbcExecuteMs={}, lateAlarmBucketMs={}, statsProjectionStageMs={}, " - + "commitMs={}, transactionMs={}, retry={}", + + "commitMs={}, transactionMs={}, retry={}, {}", org.slf4j.MDC.get("syslogBatchId"), alarmTable ? "syslog_normal_alarm" : "syslog_normal_data", rows.size(), timing.beginMs, timing.batchExecution.getPlanMs(), timing.batchExecution.getBindMs(), timing.batchExecution.getExecuteMs(), timing.lateBucketMs, timing.statsResult == null ? 0L : timing.statsResult.getMergeMs(), - timing.commitMs, transactionMs, retries); + timing.commitMs, transactionMs, retries, timing.writeSize); } if (transactionLogEnabled && timing.statsResult != null && timing.statsResult.isDirect()) { log.info("DIRECT 达梦事务完成: factRows={}, deviceGroups={}, ruleGroups={}, aggregateMs={}, " + "jdbcPlanMs={}, jdbcBindMs={}, jdbcExecuteMs={}, factFlushMs={}, " - + "statsProjectionStageMs={}, materializeMs={}, commitMs={}, transactionMs={}, retry={}", + + "statsProjectionStageMs={}, materializeMs={}, commitMs={}, transactionMs={}, retry={}, {}", rows.size(), timing.statsResult.getDeviceGroups(), timing.statsResult.getRuleGroups(), timing.statsResult.getAggregateMs(), timing.batchExecution.getPlanMs(), timing.batchExecution.getBindMs(), timing.batchExecution.getExecuteMs(), timing.factFlushMs, timing.statsResult.getMergeMs(), timing.statsResult.getMaterializeMs(), - timing.commitMs, transactionMs, retries); + timing.commitMs, transactionMs, retries, timing.writeSize); } } @@ -189,6 +189,7 @@ public class DmStandardizedBatchExecutor { } private static final class TransactionTiming { + private final DmWriteSizeEstimate writeSize = new DmWriteSizeEstimate(); private String phase = "CONNECTION_BEGIN"; private long phaseStarted = System.nanoTime(); private void phase(String value) { phase = value; phaseStarted = System.nanoTime(); } 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 bce7847..d6a69b7 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 @@ -69,7 +69,7 @@ public class DmStandardizedBatchWriter { @Value("${dm.standardized-write.batch-size:500}") private int batchSize; - @Value("${syslog.slow-batch.threshold-ms:500}") + @Value("${syslog.slow-batch.threshold-ms:1000}") private long slowQueueThresholdMs; @Value("${dm.standardized-write.transaction-timeout-seconds:45}") diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmWriteSizeEstimate.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmWriteSizeEstimate.java new file mode 100644 index 0000000..15a3f0e --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmWriteSizeEstimate.java @@ -0,0 +1,107 @@ +package com.common.service; + +import java.math.BigDecimal; +import java.math.BigInteger; +import java.sql.Timestamp; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; + +/** + * JDBC参数内容大小估算，不是网络包、数据库页或REDO大小。 + * 只统计实际INSERT字段，调用位置在截断/字段准备之后、绑定/执行之前。 + * 不读取LOB或流，不序列化任意对象，不保留字段值，不创建大字符串的UTF-8副本。 + */ +final class DmWriteSizeEstimate { + private String status = "NOT_MEASURED"; + private long payloadBytes; + private long maxRowBytes; + private long maxFieldBytes; + private String maxFieldName = "none"; + private long unmeasuredValues; + private long elapsedMs; + + void measure(List> rows, List columns) { + long started = System.nanoTime(); + try { + for (Map row : rows) { + long rowBytes = 0L; + for (DmPreparedBatchInserter.ColumnMetadata column : columns) { + long bytes = valueBytes(row.get(column.name)); + if (bytes < 0L) { + unmeasuredValues++; + continue; + } + rowBytes += bytes; + if (bytes > maxFieldBytes) { + maxFieldBytes = bytes; + maxFieldName = column.name; + } + } + payloadBytes += rowBytes; + maxRowBytes = Math.max(maxRowBytes, rowBytes); + } + status = unmeasuredValues == 0L ? "COMPLETE" : "PARTIAL"; + } catch (RuntimeException diagnosticFailure) { + // 观测失败不能改变事实写入行为；不记录异常消息，避免携带字段内容。 + status = "FAILED"; + } finally { + elapsedMs = (System.nanoTime() - started) / 1_000_000L; + } + } + + static long valueBytes(Object value) { + if (value == null) return 0L; + if (value instanceof String) return utf8Length((String) value); + if (value instanceof byte[]) return ((byte[]) value).length; + // 只对已知标量计其文本表示，避免任意对象toString触发IO或泄漏内容。 + Class type = value.getClass(); + if (type == Boolean.class || type == Character.class || type == Byte.class + || type == Short.class || type == Integer.class || type == Long.class + || type == Float.class || type == Double.class || type == BigDecimal.class + || type == BigInteger.class || type == LocalDate.class || type == LocalDateTime.class + || type == Timestamp.class || type == java.sql.Date.class || type == java.sql.Time.class) { + return utf8Length(value.toString()); + } + if (type == java.util.Date.class) { + return utf8Length(new Timestamp(((java.util.Date) value).getTime()).toString()); + } + return -1L; + } + + /** 与JDK8 String.getBytes(UTF_8)一致：有效代理对4字节，孤立代理字符替换为一个问号字节。 */ + static long utf8Length(String text) { + long bytes = 0L; + for (int i = 0; i < text.length(); i++) { + char ch = text.charAt(i); + if (ch < 0x80) bytes++; + else if (ch < 0x800) bytes += 2L; + else if (Character.isHighSurrogate(ch) && i + 1 < text.length() + && Character.isLowSurrogate(text.charAt(i + 1))) { + bytes += 4L; + i++; + } else if (Character.isSurrogate(ch)) bytes++; + else bytes += 3L; + } + return bytes; + } + + long getPayloadBytes() { return payloadBytes; } + long getMaxRowBytes() { return maxRowBytes; } + long getMaxFieldBytes() { return maxFieldBytes; } + String getMaxFieldName() { return maxFieldName; } + long getUnmeasuredValues() { return unmeasuredValues; } + String getStatus() { return status; } + + @Override + public String toString() { + return "sizeEstimateStatus=" + status + + ", estimatedPayloadBytes=" + payloadBytes + + ", maxRowEstimatedBytes=" + maxRowBytes + + ", maxFieldEstimatedBytes=" + maxFieldBytes + + ", maxFieldName=" + maxFieldName + + ", unmeasuredValues=" + unmeasuredValues + + ", sizeEstimateMs=" + elapsedMs; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLaneFieldLimiter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLaneFieldLimiter.java new file mode 100644 index 0000000..28c99fd --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLaneFieldLimiter.java @@ -0,0 +1,62 @@ +package com.common.service; + +import com.common.entity.SyslogNonNormalMessage; +import com.common.util.Utf8TextTruncator; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +/** 普通来源消息完成规则和采样后，限制非告警达梦输出文本；调用方排除告警，不修改规则输入和InfluxDB原文。 */ +@Component +public class RawLaneFieldLimiter { + // 这些字段参与幂等、路由、分区或关联，不允许截断改变其语义。 + private static final Set IDENTITY_FIELDS = new HashSet<>(Arrays.asList( + "id", "syslog_uuid", "syslog_topic", "normalize_rule_id", "rule_version", + "device_id", "tenant_id", "agent_id", "origin_agent_id", "collect_task_id", + "log_time", "created_at", "receive_time", "event_level")); + + private final long maxBytes; + private final String rawTopic; + + public RawLaneFieldLimiter( + @Value("${syslog.raw-listener.field-max-utf8-bytes:2048}") long maxBytes, + @Value("${spring.kafka.consumer.topic}") String rawTopic) { + if (maxBytes < 0 || (maxBytes > 0 && maxBytes < 64)) { + throw new IllegalArgumentException("syslog.raw-listener.field-max-utf8-bytes 必须为0或至少64"); + } + this.maxBytes = maxBytes; + this.rawTopic = rawTopic; + } + + public boolean appliesTo(String currentTopic, String originalTopic) { + String sourceTopic = originalTopic == null || originalTopic.isEmpty() ? currentTopic : originalTopic; + return maxBytes > 0 && rawTopic != null && rawTopic.equals(sourceTopic); + } + + /** 包括自定义映射文本，不限于旧的大字段白名单；数值、日期、二进制值保持原类型。 */ + public void truncateRow(Map row) { + if (row == null || maxBytes == 0) return; + for (Map.Entry entry : row.entrySet()) { + if (entry.getValue() instanceof String && !IDENTITY_FIELDS.contains(entry.getKey())) { + entry.setValue(truncate((String) entry.getValue())); + } + } + } + + public void truncateNonNormal(SyslogNonNormalMessage message) { + if (message == null || maxBytes == 0) return; + message.setSyslogMessage(truncate(message.getSyslogMessage())); + message.setHeaderMessage(truncate(message.getHeaderMessage())); + message.setReasonDetail(truncate(message.getReasonDetail())); + } + + private String truncate(String value) { + // 每个UTF-16代码单元最多占3个UTF-8字节，短字段无需分配UTF-8数组。 + if (value == null || (long) value.length() * 3L <= maxBytes) return value; + return Utf8TextTruncator.truncate(value, maxBytes).getValue(); + } +} 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 bc64e7a..3099b7e 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 @@ -12,7 +12,7 @@ import org.springframework.stereotype.Component; public class SyslogBuildStartupLogger { // 确认包含此功能修订，不代表唯一构建；后续功能修订时同步更新此标识。 - static final String CODE_REVISION = "ASYNC_STATS_ADAPTIVE_RAW_V20260913_56"; + static final String CODE_REVISION = "ASYNC_STATS_ADAPTIVE_RAW_V20260913_57"; private final Environment environment; public SyslogBuildStartupLogger(Environment environment) { @@ -32,12 +32,13 @@ public class SyslogBuildStartupLogger { + ", 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") + + ", rawFieldMaxUtf8Bytes=" + value("syslog.raw-listener.field-max-utf8-bytes", "2048") + ", samplingEnabled=" + value("syslog.backlog-sampling.enabled", "true") + ", statsWriteMode=" + value("stats.write-mode", "DIRECT") + ", ingressStatsEnabled=" + value("stats.ingress-metadata.enabled", "true") + ", statsPublisherEnabled=" + value("stats.projection-async.enabled", "true") + ", lateBucketEnabled=" + value("analysis.late-replay.outbox-enabled", "true") - + ", slowBatchThresholdMs=" + value("syslog.slow-batch.threshold-ms", "500"); + + ", slowBatchThresholdMs=" + value("syslog.slow-batch.threshold-ms", "1000"); } private String value(String key, String fallback) { 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 8045d9e..09736f6 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,8 +82,9 @@ 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} +# 慢日志统一阈值，默认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} @@ -220,6 +221,12 @@ 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 表示不限制。 +# 普通Kafka来源字段上限：默认2KB=2048个UTF-8字节（含截断标记），0关闭本层限制，启用时至少64。 +# 在规则计算/采样之后限制普通表的文本字段（幂等ID/路由/分区等技术字段除外），告警不应用本层2KB限制， +# 以及非标表syslog_message/header_message/reason_detail；不影响InfluxDB原文，大日志不应用2KB限制。 +# 重试/重放按originalTopic识别；改为4KB填4096。已有同名环境变量会覆盖JAR内默认值。 +syslog.raw-listener.field-max-utf8-bytes=${SYSLOG_RAW_LISTENER_FIELD_MAX_UTF8_BYTES:2048} +# 以下旧限制继续生效（告警也保留原有大字段限制）；同时命中时遵循较小上限，关闭上面的限制不会关闭以下限制。 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} 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 d24e8f1..fb0f20e 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 @@ -65,6 +65,8 @@ class SysLogProcessorParallelTest { executor = Executors.newFixedThreadPool(4); nonNormalExecutor = Executors.newSingleThreadExecutor(); processor = new SysLogProcessor(); + ReflectionTestUtils.setField(processor, "rawLaneFieldLimiter", + new com.common.service.RawLaneFieldLimiter(2048, "raw")); normalProcessor = mock(LogNormalProcessor.class); failureRouter = mock(SyslogFailureRouter.class); batchWriter = mock(DmStandardizedBatchWriter.class); @@ -459,6 +461,53 @@ class SysLogProcessorParallelTest { return messages; } + @Test + @SuppressWarnings("unchecked") + void rawOutputIsLimitedAfterRulesAndLargeOutputIsUnchangedIncludingRetry() { + for (String originalTopic : java.util.Arrays.asList("raw", "raw.large")) { + ConsumerRecord record = new ConsumerRecord<>("retry", 0, 0L, "key", "ciphertext"); + SyslogFailureRouter.MessageContext context = new SyslogFailureRouter.MessageContext( + "ciphertext", "stable-id", 1, null, Instant.EPOCH, originalTopic, null); + String content = String.join("", Collections.nCopies(3000, "中")); + SysLogProcessor.PreparedMessage source = new SysLogProcessor.PreparedMessage(record, context, + content, "stable-id", null, LocalDateTime.now(), 7L, 3L, System.currentTimeMillis()); + java.util.Map normal = normalizedRow(); + normal.put("custom_field", content); + java.util.Map alarm = normalizedRow(); + alarm.put("event_level", "HIGH"); + alarm.put("description", content); + SyslogNonNormalMessage nonNormal = new SyslogNonNormalMessage(); + nonNormal.setSyslogMessage(content); + List computed = java.util.Arrays.asList( + new SysLogProcessor.PreparedNormalization(source, + LogNormalProcessor.NormalizationResult.success(java.util.Arrays.asList(normal, alarm)), null), + new SysLogProcessor.PreparedNormalization(source, + LogNormalProcessor.NormalizationResult.filtered(Collections.singletonList(nonNormal)), null)); + Object retainedPlan = ReflectionTestUtils.invokeMethod(processor, "allRetainedPlan", computed); + Object persistencePlan = ReflectionTestUtils.invokeMethod(processor, "prepareStandardizedPersistence", + ReflectionTestUtils.getField(retainedPlan, "normalizations"), new SysLogProcessor.BatchMetrics()); + for (String requestsName : java.util.Arrays.asList("normalWriteRequests", "alarmWriteRequests")) { + List> requests = + (List>) + ReflectionTestUtils.getField(persistencePlan, requestsName); + assertEquals(1, requests.size(), "告警和普通记录数量不变"); + String stored = (String) requests.get(0).getData().get( + "alarmWriteRequests".equals(requestsName) ? "description" : "custom_field"); + if ("raw".equals(originalTopic) && "normalWriteRequests".equals(requestsName)) { + assertTrue(stored.getBytes(java.nio.charset.StandardCharsets.UTF_8).length <= 2048); + assertTrue(stored.contains("HB_XDR_TRUNCATED")); + } else { + assertEquals(content, stored); + } + } + assertEquals(content, normal.get("custom_field"), "规则计算结果不被原地覆盖"); + assertEquals(content, alarm.get("description")); + assertEquals(content, ReflectionTestUtils.getField(source, "decryptedMessage"), "Influx原文不改变"); + if ("raw".equals(originalTopic)) assertTrue(nonNormal.getSyslogMessage().contains("HB_XDR_TRUNCATED")); + else assertEquals(content, nonNormal.getSyslogMessage()); + } + } + private List oldMessages(int count) { List old = new ArrayList<>(); for (int i = 0; i < count; i++) { diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmPreparedBatchInserterTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmPreparedBatchInserterTest.java index eac6628..f4019a6 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmPreparedBatchInserterTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmPreparedBatchInserterTest.java @@ -113,13 +113,24 @@ class DmPreparedBatchInserterTest { DmPreparedBatchInserter inserter = new DmPreparedBatchInserter(mock( org.springframework.jdbc.core.JdbcTemplate.class)); + DmWriteSizeEstimate size = new DmWriteSizeEstimate(); DmPreparedBatchInserter.BatchExecution result = inserter.execute(connection, - DmPreparedBatchInserter.NORMAL_TABLE, Arrays.asList(first, second), 45); + DmPreparedBatchInserter.NORMAL_TABLE, Arrays.asList(first, second), 45, size); assertEquals(2, result.getRows()); assertEquals(2, result.getColumns()); + assertEquals(18L, size.getPayloadBytes()); + assertEquals(13L, size.getMaxRowBytes()); + assertEquals("src_ip", size.getMaxFieldName()); verify(insertStatement, times(2)).addBatch(); verify(insertStatement).executeBatch(); verify(insertStatement).setNull(2, Types.VARCHAR); + + when(insertStatement.executeBatch()).thenThrow(new java.sql.SQLException("test failure")); + DmWriteSizeEstimate failedSize = new DmWriteSizeEstimate(); + assertThrows(java.sql.SQLException.class, () -> inserter.execute(connection, + DmPreparedBatchInserter.NORMAL_TABLE, Arrays.asList(first, second), 45, failedSize)); + assertEquals("COMPLETE", failedSize.getStatus()); + assertEquals(18L, failedSize.getPayloadBytes()); } } 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 b893dd4..7d2b9c0 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,34 @@ class DmStandardizedBatchWriterTest { when(normalDataService.isAlarmData(any())).thenReturn(false); } + @Test + void alarmsKeepLegacyTruncationInBothBatchAndSingleMode() { + DmLargeTextLimiter limiter = new DmLargeTextLimiter(); + ReflectionTestUtils.setField(limiter, "normalMaxBytes", 64L); + ReflectionTestUtils.setField(limiter, "normalFieldsValue", "description"); + limiter.validate(); + ReflectionTestUtils.setField(writer, "largeTextLimiter", limiter); + when(normalDataService.isAlarmData(any())).thenAnswer(invocation -> + ((Map) invocation.getArgument(0)).get("event_level") != null); + when(normalDataService.insertDynamic(any())).thenReturn(1); + String content = String.join("", java.util.Collections.nCopies(300000, "x")); + for (String mode : java.util.Arrays.asList("BATCH", "SINGLE")) { + ReflectionTestUtils.setField(writer, "mode", mode); + List> requests = requests(2, -1); + requests.get(0).getData().put("event_level", "HIGH"); + requests.get(0).getData().put("description", content); + requests.get(1).getData().put("description", content); + DmStandardizedBatchWriter.WriteReport report = writer.write(requests); + assertEquals(2, report.getSuccessfulRows()); + String alarm = (String) requests.get(0).getData().get("description"); + assertTrue(alarm.contains("HB_XDR_TRUNCATED"), "告警仍遵守旧的大字段上限"); + assertTrue(alarm.getBytes(java.nio.charset.StandardCharsets.UTF_8).length <= 64); + String ordinary = (String) requests.get(1).getData().get("description"); + assertTrue(ordinary.contains("HB_XDR_TRUNCATED")); + assertTrue(ordinary.getBytes(java.nio.charset.StandardCharsets.UTF_8).length <= 64); + } + } + @Test void oneBadRowIsIsolatedWithoutDowngradingEveryGoodRowToSingleInsert() { List> requests = requests(100, 73); diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmTransactionWriteSizeLogTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmTransactionWriteSizeLogTest.java new file mode 100644 index 0000000..b2c6507 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmTransactionWriteSizeLogTest.java @@ -0,0 +1,87 @@ +package com.common.service; + +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import org.apache.ibatis.session.Configuration; +import org.apache.ibatis.session.SqlSessionFactory; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.AbstractPlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionStatus; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +class DmTransactionWriteSizeLogTest { + @Test + void successfulSlowNormalAndAlarmTransactionsIncludeOnlySizeMetadata() { + check(false, false); + check(true, false); + } + + @Test + void failedWriteIncludesAttemptedPayloadAndStillThrows() { + check(false, true); + } + + private void check(boolean alarm, boolean fail) { + SqlSessionFactory sessions = mock(SqlSessionFactory.class); + Configuration configuration = new Configuration(); + configuration.setEnvironment(new org.apache.ibatis.mapping.Environment("test", + new org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory(), mock(javax.sql.DataSource.class))); + when(sessions.getConfiguration()).thenReturn(configuration); + DmPreparedBatchInserter inserter = mock(DmPreparedBatchInserter.class); + when(inserter.insert(eq(alarm), anyList(), anyInt(), any(DmWriteSizeEstimate.class))) + .thenAnswer(invocation -> { + DmWriteSizeEstimate size = invocation.getArgument(3); + size.measure(invocation.getArgument(1), DmWriteSizeEstimateTest.columns("detail")); + if (fail) throw new IllegalStateException("write failed"); + return new DmPreparedBatchInserter.BatchExecution(1, 1, 0, 0, 0); + }); + AbstractPlatformTransactionManager transactions = new AbstractPlatformTransactionManager() { + protected Object doGetTransaction() { return new Object(); } + protected void doBegin(Object transaction, TransactionDefinition definition) { } + protected void doCommit(DefaultTransactionStatus status) { + // 使成功事务确定超过1ms阈值，不依赖真实数据库。 + java.util.concurrent.locks.LockSupport.parkNanos(5_000_000L); + } + protected void doRollback(DefaultTransactionStatus status) { } + }; + DmStandardizedBatchExecutor executor = new DmStandardizedBatchExecutor(sessions, transactions, + inserter, mock(StatsChangeCaptureService.class), mock(DirectStatsProjectionWriter.class), + new DmDirectTransactionMetrics(), mock(LateAlarmBucketWriter.class)); + ReflectionTestUtils.setField(executor, "slowTransactionThresholdMs", 1L); + Logger logger = (Logger) LoggerFactory.getLogger(DmStandardizedBatchExecutor.class); + ListAppender appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + List> rows = Collections.singletonList( + Collections.singletonMap("detail", "秘密😀")); + try { + if (fail) assertThrows(IllegalStateException.class, () -> executor.execute(alarm, rows, 45)); + else executor.execute(alarm, rows, 45); + String message = appender.list.stream().map(ILoggingEvent::getFormattedMessage) + .filter(text -> text.contains("[SLOW_DM_TRANSACTION]")).findFirst().orElse(""); + assertTrue(message.contains("table=" + (alarm ? "syslog_normal_alarm" : "syslog_normal_data"))); + assertTrue(message.contains("status=" + (fail ? "FAILED" : "SUCCESS"))); + assertTrue(message.contains("sizeEstimateStatus=COMPLETE")); + assertTrue(message.contains("estimatedPayloadBytes=10")); + assertTrue(message.contains("maxRowEstimatedBytes=10")); + assertTrue(message.contains("maxFieldEstimatedBytes=10")); + assertTrue(message.contains("maxFieldName=detail")); + assertTrue(message.contains("sizeEstimateMs=")); + assertFalse(message.contains("秘密")); + } finally { + logger.detachAppender(appender); + appender.stop(); + } + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmWriteSizeEstimateTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmWriteSizeEstimateTest.java new file mode 100644 index 0000000..5259335 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmWriteSizeEstimateTest.java @@ -0,0 +1,109 @@ +package com.common.service; + +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; + +import java.nio.charset.StandardCharsets; +import java.sql.Types; +import java.util.*; + +import static org.junit.jupiter.api.Assertions.*; + +class DmWriteSizeEstimateTest { + @Test + void utf8MatchesJdkForAsciiChineseEmojiMalformedSurrogatesAndRandomText() { + for (String text : Arrays.asList("", "ascii", "中文😀", "\uD800", "\uDC00", "\uD800x\uDC00")) { + assertEquals(text.getBytes(StandardCharsets.UTF_8).length, DmWriteSizeEstimate.utf8Length(text)); + } + Random random = new Random(42); + for (int i = 0; i < 200; i++) { + char[] chars = new char[1024]; + for (int j = 0; j < chars.length; j++) chars[j] = (char) random.nextInt(65536); + String text = new String(chars); + assertEquals(text.getBytes(StandardCharsets.UTF_8).length, DmWriteSizeEstimate.utf8Length(text)); + } + } + + @Test + void measuresOnlyBoundColumnsAndDoesNotMutateRowsOrLogValues() { + Map first = new LinkedHashMap<>(); + first.put("id", "a"); + first.put("detail", "敏感内容😀"); // 16 UTF-8 bytes + first.put("ignored", "not-in-sql"); + Map second = new LinkedHashMap<>(); + second.put("id", "bc"); + second.put("detail", new byte[3]); + Map snapshot = new LinkedHashMap<>(first); + DmWriteSizeEstimate size = new DmWriteSizeEstimate(); + size.measure(Arrays.asList(first, second), columns("id", "detail")); + assertEquals(22L, size.getPayloadBytes()); + assertEquals(17L, size.getMaxRowBytes()); + assertEquals(16L, size.getMaxFieldBytes()); + assertEquals("detail", size.getMaxFieldName()); + assertEquals("COMPLETE", size.getStatus()); + assertEquals(snapshot, first); + assertFalse(size.toString().contains("敏感内容")); + assertFalse(size.toString().contains("not-in-sql")); + } + + @Test + void measuresTruncatedValuesNotOriginalAndKeepsTruncationMarker() { + DmLargeTextLimiter limiter = new DmLargeTextLimiter(); + ReflectionTestUtils.setField(limiter, "normalMaxBytes", 64L); + ReflectionTestUtils.setField(limiter, "normalFieldsValue", "detail"); + limiter.validate(); + Map row = new LinkedHashMap<>(); + row.put("detail", String.join("", Collections.nCopies(100, "中"))); + limiter.truncateNormalRow(row); + String truncated = (String) row.get("detail"); + assertTrue(truncated.getBytes(StandardCharsets.UTF_8).length <= 64); + DmWriteSizeEstimate size = new DmWriteSizeEstimate(); + size.measure(Collections.singletonList(row), columns("detail")); + assertEquals(truncated.getBytes(StandardCharsets.UTF_8).length, size.getPayloadBytes()); + assertSame(truncated, row.get("detail")); + } + + @Test + void handlesNullScalarsAndUnknownWithoutCallingArbitraryToString() { + Object unknown = new Object() { + @Override public String toString() { throw new AssertionError("must not stringify"); } + }; + Map row = new LinkedHashMap<>(); + row.put("id", 123); + row.put("flag", true); + row.put("unknown", unknown); + row.put("missing", null); + DmWriteSizeEstimate size = new DmWriteSizeEstimate(); + size.measure(Collections.singletonList(row), columns("id", "flag", "unknown", "missing")); + assertEquals(7L, size.getPayloadBytes()); + assertEquals(1L, size.getUnmeasuredValues()); + assertEquals("PARTIAL", size.getStatus()); + } + + @Test + void diagnosticFailureIsMarkedWithoutThrowingAndUnmeasuredIsExplicit() { + DmWriteSizeEstimate size = new DmWriteSizeEstimate(); + assertEquals("NOT_MEASURED", size.getStatus()); + Map bad = new HashMap() { + @Override public Object get(Object key) { throw new IllegalStateException("sensitive value"); } + }; + assertDoesNotThrow(() -> size.measure(Collections.singletonList(bad), columns("id"))); + assertEquals("FAILED", size.getStatus()); + assertFalse(size.toString().contains("sensitive value")); + } + + @Test + void totalUsesLongBeyondTwoGiBWithoutAllocatingAnEncodedCopy() { + String text = String.join("", Collections.nCopies(1024 * 1024, "a")); + DmWriteSizeEstimate size = new DmWriteSizeEstimate(); + size.measure(Collections.nCopies(2050, Collections.singletonMap("detail", text)), + columns("detail")); + assertEquals(2050L * 1024 * 1024, size.getPayloadBytes()); + } + + static List columns(String... names) { + List columns = new ArrayList<>(); + for (String name : names) columns.add(new DmPreparedBatchInserter.ColumnMetadata(name, Types.VARCHAR)); + return columns; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/RawLaneFieldLimiterTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/RawLaneFieldLimiterTest.java new file mode 100644 index 0000000..b0bb700 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/RawLaneFieldLimiterTest.java @@ -0,0 +1,112 @@ +package com.common.service; + +import com.common.entity.SyslogNonNormalMessage; +import org.junit.jupiter.api.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.core.env.MapPropertySource; + +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; + +class RawLaneFieldLimiterTest { + @Test + void springDefaultIsTwoKiBAndExplicitOverrideWorks() { + checkSpringLimit(null, 2048); + checkSpringLimit("4096", 4096); + checkSpringLimit("0", 0); + } + + private void checkSpringLimit(String override, int expected) { + try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) { + Map properties = new HashMap<>(); + properties.put("spring.kafka.consumer.topic", "raw"); + if (override != null) properties.put("syslog.raw-listener.field-max-utf8-bytes", override); + context.getEnvironment().getPropertySources().addFirst(new MapPropertySource("test", properties)); + context.register(RawLaneFieldLimiter.class); + context.refresh(); + RawLaneFieldLimiter limiter = context.getBean(RawLaneFieldLimiter.class); + Map row = new HashMap<>(); + String content = repeat("a", 5000); + row.put("custom_field", content); + limiter.truncateRow(row); + assertEquals(expected == 0 ? content : expected, + expected == 0 ? row.get("custom_field") : bytes(row.get("custom_field"))); + assertEquals(expected != 0, limiter.appliesTo("raw", null)); + } + } + + @Test + void truncatesCustomTextByUtf8WithMarkerAndPreservesTypesAndIdentity() { + RawLaneFieldLimiter limiter = new RawLaneFieldLimiter(2048, "raw"); + Map row = new HashMap<>(); + String content = repeat("中文🙂", 1000); + row.put("custom_field", content); + row.put("id", content); + row.put("event_level", "HIGH"); + row.put("number", 123L); + row.put("nothing", null); + byte[] binary = new byte[3000]; + row.put("binary", binary); + row.put("exact", repeat("a", 2048)); + row.put("short", "短字段"); + limiter.truncateRow(row); + String stored = (String) row.get("custom_field"); + assertTrue(bytes(stored) <= 2048); + assertTrue(stored.endsWith("...[HB_XDR_TRUNCATED original_utf8_bytes=10000]")); + assertEquals(stored, new String(stored.getBytes(StandardCharsets.UTF_8), StandardCharsets.UTF_8)); + assertEquals(content, row.get("id")); + assertEquals("HIGH", row.get("event_level")); + assertEquals(123L, row.get("number")); + assertNull(row.get("nothing")); + assertSame(binary, row.get("binary")); + assertEquals(repeat("a", 2048), row.get("exact")); + assertEquals("短字段", row.get("short")); + limiter.truncateRow(row); + assertEquals(stored, row.get("custom_field"), "重复处理不能重复追加标记"); + } + + @Test + void onlyOriginalRawTopicAppliesIncludingRetryAndReplay() { + RawLaneFieldLimiter limiter = new RawLaneFieldLimiter(2048, "custom-raw"); + assertTrue(limiter.appliesTo("custom-raw", null)); + assertTrue(limiter.appliesTo("retry", "custom-raw")); + assertTrue(limiter.appliesTo("replay", "custom-raw")); + assertFalse(limiter.appliesTo("custom-large", null)); + assertFalse(limiter.appliesTo("retry", "custom-large")); + assertFalse(limiter.appliesTo("custom-raw", "custom-large")); + assertFalse(limiter.appliesTo("retry", null)); + } + + @Test + void nonNormalContentLimitedButIdentityUnchanged() { + RawLaneFieldLimiter limiter = new RawLaneFieldLimiter(2048, "raw"); + SyslogNonNormalMessage message = new SyslogNonNormalMessage(); + String content = repeat("x", 4096); + message.setSyslogMessage(content); + message.setHeaderMessage(content); + message.setReasonDetail(content); + message.setSyslogUuid("stable-id"); + limiter.truncateNonNormal(message); + assertEquals(2048, bytes(message.getSyslogMessage())); + assertEquals(2048, bytes(message.getHeaderMessage())); + assertEquals(2048, bytes(message.getReasonDetail())); + assertEquals("stable-id", message.getSyslogUuid()); + } + + @Test + void validatesMarkerBudget() { + assertThrows(IllegalArgumentException.class, () -> new RawLaneFieldLimiter(-1, "raw")); + assertThrows(IllegalArgumentException.class, () -> new RawLaneFieldLimiter(63, "raw")); + assertDoesNotThrow(() -> new RawLaneFieldLimiter(64, "raw")); + } + + private static int bytes(Object text) { return ((String) text).getBytes(StandardCharsets.UTF_8).length; } + private static String repeat(String text, int count) { + StringBuilder value = new StringBuilder(); + for (int i = 0; i < count; i++) value.append(text); + return value.toString(); + } +} 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 3837632..45a2be7 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,7 @@ class SyslogBuildStartupLoggerTest { MockEnvironment environment = new MockEnvironment() .withProperty("syslog.raw-listener.max-poll-records", "700") .withProperty("syslog.raw-listener.adaptive-max-records", "2500") + .withProperty("syslog.slow-batch.threshold-ms", "1500") .withProperty("stats.projection-async.enabled", "false"); environment.setActiveProfiles("prod"); SyslogBuildStartupLogger reporter = new SyslogBuildStartupLogger(environment); @@ -32,6 +33,7 @@ class SyslogBuildStartupLoggerTest { assertTrue(message.contains("baseProcessingRecords=700")); assertTrue(message.contains("adaptiveMaxRecords=2500")); assertTrue(message.contains("statsPublisherEnabled=false")); + assertTrue(message.contains("slowBatchThresholdMs=1500")); } finally { logger.detachAppender(appender); appender.stop(); @@ -46,6 +48,8 @@ class SyslogBuildStartupLoggerTest { assertFalse(summary.contains("buildTime=")); assertTrue(summary.contains("baseProcessingRecords=500")); assertTrue(summary.contains("adaptiveMaxRecords=3000")); + 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 5107d5c..254e896 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 @@ -125,12 +125,43 @@ docker exec syslog-consumer-rule-dm sh -c 'curl -fsS http://127.0.0.1:8289/xdrru `dropped_fact_rows_total` 指受统计缺失影响的事实行数/尝试行数，并不表示事实数据被删除。 这些内存指标会在重启后清零，不能覆盖崩溃时丢失的统计。损失日志最多每10秒一次，计数逐次累加。 `published_total` 仅表示Kafka确认，不代表统计已落达梦；统计最终入库还需检查投影消费组积压。 -`SLOW_DM_TRANSACTION` 自动沿用500ms阈值，无须打开所有事务INFO日志： +`SLOW_DM_TRANSACTION` 自动沿用1000ms（1秒）默认阈值，无须打开所有事务INFO日志： `connectionBeginMs`包括取连接/事务初始化，`lateAlarmBucketMs`仅为分钟桶小键准备/注册提交回调， `jdbcExecuteMs`是JDBC批执行，`statsProjectionStageMs`在DIRECT下是统计事件构造/注册回调阶段， 不含后台Kafka/统计写库；`commitMs`是事实事务提交阶段。状态中的统计成功指标表示聚合/注册完成，不保证投影落库。 `SLOW_DM_QUEUE` 单独说明批写任务排队超过阈值。 +### 达梦写入内容大小诊断 + +`SLOW_DM_TRANSACTION` 的普通表/告警表成功慢事务、失败事务日志增加以下字段，无需新配置、数据库或Kafka脚本： + +| 字段 | 含义 | +| --- | --- | +| sizeEstimateStatus | COMPLETE：支持的参数内容已测量；PARTIAL：有未测量类型；FAILED：估算异常；NOT_MEASURED：尚未执行到测量阶段 | +| estimatedPayloadBytes | 本次小事务实际INSERT列的非空参数内容估算总字节数 | +| maxRowEstimatedBytes | 单行内容估算最大字节数 | +| maxFieldEstimatedBytes / maxFieldName | 单个字段值最大估算字节数及列名，不打印内容；同大小取首个，均为零时名称为none | +| unmeasuredValues | 不支持测量的非空参数个数，如LOB/流/自定义对象；不会为了测量读取它们 | +| sizeEstimateMs | 大小估算自身耗时，包含在事务总耗时中，不计入jdbcExecuteMs或commitMs | + +测量在已有字段截断和字段准备之后、按表元数据确定INSERT列之后、绑定和执行之前进行。 +String按UTF-8计数（含中文、emoji，不创建完整编码字节数组），byte[]按原始长度计数， +常用数值/布尔/日期按文本表示估算，NULL计零。不包含SQL文本、驱动协议、索引、数据库页及REDO开销， +也不等于数据库字符集编码后的物理长度。COMPLETE表示参数内容估算完整，不代表精确磁盘字节数。 +失败事务表示本次尝试量，不是成功落库量；失败/二分/重试的多条日志不能累加为净新增量。 +NOT_MEASURED或FAILED中的零/部分值不能理解为本次写入内容为空。 +仅增加观测，不修改字段值、截断阈值、保存条数或Kafka确认策略；不覆盖现有非标异步写库路径。 +若需要快慢批次对照，已有stats.direct.transaction-log-enabled=true可打印DIRECT成功事务信息（日志量会增加）。 + +客户仍在宿主机执行： + +```bash +docker logs --since 10m syslog-consumer-rule-dm 2>&1 | grep 'SLOW_DM_TRANSACTION' | tail -30 +``` + +将estimatedPayloadBytes、maxRowEstimatedBytes与jdbcExecuteMs、commitMs一起比较， +通过batchId关联SLOW_SYSLOG_BATCH；不能将polledMessageBytes视为达梦写入大小。 + 在有Kafka脚本的现场节点执行（已确认bootstrap地址）： ```bash @@ -182,7 +213,12 @@ docker exec syslog-consumer-rule-dm sh -c 'curl -fsS http://127.0.0.1:8289/xdrru `stageMs`各阶段为互斥墙钟时间，跨业务块累加；旧有normalizeMs等字段包含部分子阶段，不能再与stageMs相加。 兼容字段ruleComputeWorkerMs实际是整批规则计算墙钟耗时；新增ruleWorkerBusySumMs才是worker耗时之和，二者不要混淆。 -业务异常和提交异常即使小于500ms也输出诊断。`processedRecords/prepared`等旧计数仅覆盖完成的业务块； +慢日志统一使用 `syslog.slow-batch.threshold-ms`，默认1000ms，适用于 `SLOW_SYSLOG_BATCH`、 +`SLOW_DM_TRANSACTION`、`SLOW_DM_QUEUE` 和 `SLOW_NON_NORMAL_WRITE`。 +如容器已有 `SYSLOG_SLOW_BATCH_THRESHOLD_MS=500`，需删除该覆盖或改为1000；仅替换JAR不会覆盖环境变量。 +启动日志 `SYSLOG_BUILD_READY` 中的 `slowBatchThresholdMs` 可确认生效值。无需执行数据库或Kafka脚本。 + +业务异常和提交异常即使小于1000ms也输出诊断。`processedRecords/prepared`等旧计数仅覆盖完成的业务块； 出现failedChunks时不能把它们理解为整次poll已成功入库。无限等待中的操作尚未结束，不会提前输出最终慢批次。 采样审计接口新增 `audit_async`，直接统计接口的 `dm_transaction.late_alarm_bucket_async` 查看通知队列。 @@ -192,6 +228,30 @@ docker exec syslog-consumer-rule-dm sh -c 'curl -fsS http://127.0.0.1:8289/xdrru 新审计指标 `audit_lost_or_unconfirmed_messages_total` 表示未记录或提交结果未知的消息数， 旧 `audit_fallback_messages_total` 本路径不再增加。 +## 普通通道字段截断（默认2KB） + +配置文件：`syslog-consumer-rule/src/main/resources/application.properties`（prod也继承）。 + +```properties +syslog.raw-listener.field-max-utf8-bytes=${SYSLOG_RAW_LISTENER_FIELD_MAX_UTF8_BYTES:2048} +``` + +- 默认2048个UTF-8字节，不是2048个汉字；改为4KB时填4096。0仅关闭本层限制，启用时至少64。 +- 对原始Topic等于 `spring.kafka.consumer.topic` 的消息生效。Retry/Replay优先按信封的originalTopic判断； + 缺失originalTopic时只按当前Topic判断，不猜测大日志来源。 +- 在规则计算及采样决策后、达梦写入前处理：`syslog_normal_data` 的文本内容字段 + （包括自定义映射字段），以及非标表的 `syslog_message`、`header_message`、`reason_detail`。 +- 不截断数值、日期、二进制对象；标准化结果的ID、syslog_uuid、Topic、规则版本、设备/租户/采集ID、 + 分区时间和event_level等技术字段保持不变。`syslog_normal_alarm` 跳过新增2KB限制， + 但仍保留原有 `dm.normal.large-field-max-utf8-bytes`（默认256KB，配置字段白名单）限制，本次不改旧策略。 +- 超限内容尾部追加 `...[HB_XDR_TRUNCATED original_utf8_bytes=原字节数]`，标记包含在2048字节内， + 不切断中文或emoji。不会修改规则输入或InfluxDB原文限制；大日志通道不应用2KB限制，仍沿用旧策略。 +- 旧的 `dm.normal.large-field-max-utf8-bytes` 和 `dm.non-normal.large-field-max-utf8-bytes` 继续生效； + 同时命中时最终遵循较小上限。此配置不修改历史记录，已截掉的字段内容不能靠回滚JAR恢复。 + +替换JAR并重启即可，无需数据库或Kafka脚本。若容器已有 `SYSLOG_RAW_LISTENER_FIELD_MAX_UTF8_BYTES`， +环境变量优先于JAR内默认值。启动日志 `SYSLOG_BUILD_READY` 的 `rawFieldMaxUtf8Bytes=2048` 可确认实际值。 + ## 回滚 回滚旧JAR会恢复主消费同步入口统计、事实计数Outbox、同步审计及分钟桶保护，新独立listener随进程退出停止。