feat(syslog): 增加普通通道字段截断并完善慢写诊断
- 新增普通通道字段上限配置，默认2048个UTF-8字节，截断尾部保留标记 - 规则计算后执行截断，告警不应用新增2KB限制，原有256KB策略不变 - 非标原文、头部及详细原因应用新增限制，大日志通道和InfluxDB策略不变 - Retry/Replay按原始Topic识别通道，启动日志输出生效上限 - 补充达梦实际写入内容大小、最大行及最大字段诊断 - 慢日志默认阈值统一调整为1000毫秒，异常和失败仍记录 - 更新v4配置说明及回归测试 验证：JDK 8构建成功，250项测试通过 升级：替换JAR并重启，无需数据库或Kafka脚本
This commit is contained in:
王绍宏
+16
-2
@@ -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<String, Object> row : prepared.result.getRows()) {
|
||||
boolean alarm = syslogNormalDataService.isAlarmData(row);
|
||||
if (alarm || retained.keepNormal) {
|
||||
Map<String, Object> storedRow = row;
|
||||
if (limitRawFields && !alarm) {
|
||||
storedRow = new LinkedHashMap<>(row);
|
||||
rawLaneFieldLimiter.truncateRow(storedRow);
|
||||
}
|
||||
DmStandardizedBatchWriter.WriteRequest<PreparedMessage> request =
|
||||
new DmStandardizedBatchWriter.WriteRequest<>(prepared.message, row);
|
||||
new DmStandardizedBatchWriter.WriteRequest<>(prepared.message, storedRow);
|
||||
if (alarm) alarmWriteRequests.add(request);
|
||||
else normalWriteRequests.add(request);
|
||||
retainedStandardRows++;
|
||||
|
||||
+14
-1
@@ -46,20 +46,33 @@ public class DmPreparedBatchInserter {
|
||||
|
||||
public BatchExecution insert(boolean alarmTable, List<Map<String, Object>> rows,
|
||||
int statementTimeoutSeconds) {
|
||||
return insert(alarmTable, rows, statementTimeoutSeconds, new DmWriteSizeEstimate());
|
||||
}
|
||||
|
||||
BatchExecution insert(boolean alarmTable, List<Map<String, Object>> 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<BatchExecution>) connection ->
|
||||
execute(connection, table, rows, statementTimeoutSeconds));
|
||||
execute(connection, table, rows, statementTimeoutSeconds, writeSize));
|
||||
}
|
||||
|
||||
BatchExecution execute(Connection connection, String table, List<Map<String, Object>> rows,
|
||||
int statementTimeoutSeconds) throws SQLException {
|
||||
return execute(connection, table, rows, statementTimeoutSeconds, new DmWriteSizeEstimate());
|
||||
}
|
||||
|
||||
BatchExecution execute(Connection connection, String table, List<Map<String, Object>> 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;
|
||||
|
||||
+9
-8
@@ -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(); }
|
||||
|
||||
+1
-1
@@ -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}")
|
||||
|
||||
+107
@@ -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<Map<String, Object>> rows, List<DmPreparedBatchInserter.ColumnMetadata> columns) {
|
||||
long started = System.nanoTime();
|
||||
try {
|
||||
for (Map<String, Object> 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;
|
||||
}
|
||||
}
|
||||
+62
@@ -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<String> 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<String, Object> row) {
|
||||
if (row == null || maxBytes == 0) return;
|
||||
for (Map.Entry<String, Object> 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();
|
||||
}
|
||||
}
|
||||
+3
-2
@@ -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) {
|
||||
|
||||
@@ -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}
|
||||
|
||||
+49
@@ -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<String, String> 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<String, Object> normal = normalizedRow();
|
||||
normal.put("custom_field", content);
|
||||
java.util.Map<String, Object> alarm = normalizedRow();
|
||||
alarm.put("event_level", "HIGH");
|
||||
alarm.put("description", content);
|
||||
SyslogNonNormalMessage nonNormal = new SyslogNonNormalMessage();
|
||||
nonNormal.setSyslogMessage(content);
|
||||
List<SysLogProcessor.PreparedNormalization> 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<DmStandardizedBatchWriter.WriteRequest<SysLogProcessor.PreparedMessage>> requests =
|
||||
(List<DmStandardizedBatchWriter.WriteRequest<SysLogProcessor.PreparedMessage>>)
|
||||
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<SysLogProcessor.PreparedMessage> oldMessages(int count) {
|
||||
List<SysLogProcessor.PreparedMessage> old = new ArrayList<>();
|
||||
for (int i = 0; i < count; i++) {
|
||||
|
||||
+12
-1
@@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
+28
@@ -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<DmStandardizedBatchWriter.WriteRequest<String>> 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<String> 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<DmStandardizedBatchWriter.WriteRequest<String>> requests = requests(100, 73);
|
||||
|
||||
+87
@@ -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<ILoggingEvent> appender = new ListAppender<>();
|
||||
appender.start();
|
||||
logger.addAppender(appender);
|
||||
List<Map<String, Object>> rows = Collections.singletonList(
|
||||
Collections.<String, Object>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();
|
||||
}
|
||||
}
|
||||
}
|
||||
+109
@@ -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<String, Object> first = new LinkedHashMap<>();
|
||||
first.put("id", "a");
|
||||
first.put("detail", "敏感内容😀"); // 16 UTF-8 bytes
|
||||
first.put("ignored", "not-in-sql");
|
||||
Map<String, Object> second = new LinkedHashMap<>();
|
||||
second.put("id", "bc");
|
||||
second.put("detail", new byte[3]);
|
||||
Map<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> bad = new HashMap<String, Object>() {
|
||||
@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.<String, Object>singletonMap("detail", text)),
|
||||
columns("detail"));
|
||||
assertEquals(2050L * 1024 * 1024, size.getPayloadBytes());
|
||||
}
|
||||
|
||||
static List<DmPreparedBatchInserter.ColumnMetadata> columns(String... names) {
|
||||
List<DmPreparedBatchInserter.ColumnMetadata> columns = new ArrayList<>();
|
||||
for (String name : names) columns.add(new DmPreparedBatchInserter.ColumnMetadata(name, Types.VARCHAR));
|
||||
return columns;
|
||||
}
|
||||
}
|
||||
+112
@@ -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<String, Object> 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<String, Object> 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<String, Object> 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();
|
||||
}
|
||||
}
|
||||
+4
@@ -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"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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随进程退出停止。
|
||||
|
||||
Reference in New Issue
Block a user