com.fasterxml.jackson.datatype
jackson-datatype-jsr310
@@ -197,7 +214,7 @@
org.junit.jupiter
junit-jupiter
- ${junit.version}
+ ${junit.jupiter.version}
test
@@ -226,4 +243,67 @@
-
\ No newline at end of file
+
+
+ cache-it
+
+
+ org.testcontainers
+ testcontainers
+ 1.19.8
+ test
+
+
+ org.testcontainers
+ junit-jupiter
+ 1.19.8
+ test
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+ **/*IntegrationTest.java
+
+
+
+
+ org.codehaus.mojo
+ build-helper-maven-plugin
+ 3.5.0
+
+
+ add-cache-integration-tests
+ generate-test-sources
+ add-test-source
+
+ src/cache-it/java
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+ 2.22.2
+
+
+ **/*IntegrationTest.java
+
+
+
+
+ integration-testverify
+
+
+
+
+
+
+
+
+
diff --git a/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/AbstractJetCacheRedisIntegrationTest.java b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/AbstractJetCacheRedisIntegrationTest.java
new file mode 100644
index 0000000..f0d8b0a
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/AbstractJetCacheRedisIntegrationTest.java
@@ -0,0 +1,55 @@
+package com.Modules.NormalData.cache;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+@Testcontainers
+abstract class AbstractJetCacheRedisIntegrationTest {
+ @Container
+ static final GenericContainer REDIS = new GenericContainer<>(DockerImageName.parse("redis:7.2-alpine"))
+ .withExposedPorts(6379);
+
+ ConfigurableApplicationContext first;
+ ConfigurableApplicationContext second;
+
+ @BeforeEach
+ void startContexts() {
+ first = context();
+ second = context();
+ }
+
+ @AfterEach
+ void closeContexts() {
+ if (second != null) second.close();
+ if (first != null) first.close();
+ }
+
+ ConfigurableApplicationContext context() {
+ String uri = "redis://" + REDIS.getHost() + ':' + REDIS.getMappedPort(6379) + "/15";
+ return new SpringApplicationBuilder(CacheItApplication.class)
+ .web(WebApplicationType.NONE)
+ .properties("jetcache.statIntervalMinutes=0",
+ "jetcache.areaInCacheName=false",
+ "jetcache.penetrationProtect=true",
+ "jetcache.enableMethodCache=true",
+ "jetcache.local.default.type=caffeine",
+ "jetcache.local.default.keyConvertor=fastjson2",
+ "jetcache.remote.default.type=redis.lettuce",
+ "jetcache.remote.default.keyConvertor=fastjson2",
+ "jetcache.remote.default.serialPolicy=KRYO5",
+ "jetcache.remote.default.broadcastChannel=xdr:norm:v1:cache-invalidate",
+ "jetcache.remote.default.uri.0=" + uri)
+ .run();
+ }
+
+ CacheItApplication.CacheItFacade facade(ConfigurableApplicationContext context) {
+ return context.getBean(CacheItApplication.CacheItFacade.class);
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/CacheItApplication.java b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/CacheItApplication.java
new file mode 100644
index 0000000..53cc583
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/CacheItApplication.java
@@ -0,0 +1,39 @@
+package com.Modules.NormalData.cache;
+
+import com.alicp.jetcache.anno.CacheType;
+import com.alicp.jetcache.anno.Cached;
+import com.alicp.jetcache.anno.CacheInvalidate;
+import com.alicp.jetcache.anno.SerialPolicy;
+import com.alicp.jetcache.anno.config.EnableCreateCacheAnnotation;
+import com.alicp.jetcache.anno.config.EnableMethodCache;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@SpringBootConfiguration
+@EnableAutoConfiguration
+@EnableMethodCache(basePackages = "com.Modules.NormalData.cache")
+@EnableCreateCacheAnnotation
+class CacheItApplication {
+ @Bean CacheItFacade cacheItFacade() { return new CacheItFacade(); }
+
+ static final class CacheItFacade {
+ private final AtomicInteger loads = new AtomicInteger();
+
+ @Cached(name = "xdr:norm:v1:cache-it:", key = "#key", cacheType = CacheType.BOTH,
+ expire = 60, localExpire = 60, timeUnit = TimeUnit.SECONDS,
+ localLimit = 100, syncLocal = true, serialPolicy = SerialPolicy.KRYO5)
+ public Object get(String key, Object loadedValue) {
+ loads.incrementAndGet();
+ return loadedValue;
+ }
+
+ @CacheInvalidate(name = "xdr:norm:v1:cache-it:", key = "#key")
+ public void invalidate(String key) { }
+
+ int loads() { return loads.get(); }
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheCrossContextInvalidationIntegrationTest.java b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheCrossContextInvalidationIntegrationTest.java
new file mode 100644
index 0000000..7c2f0f5
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheCrossContextInvalidationIntegrationTest.java
@@ -0,0 +1,17 @@
+package com.Modules.NormalData.cache;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class JetCacheCrossContextInvalidationIntegrationTest extends AbstractJetCacheRedisIntegrationTest {
+ @Test
+ void invalidationBroadcastRemovesOtherContextsLocalEntry() throws Exception {
+ facade(first).get("broadcast", "before");
+ assertEquals("before", facade(second).get("broadcast", "unused"));
+ facade(first).invalidate("broadcast");
+ Thread.sleep(300L);
+ assertEquals("after", facade(second).get("broadcast", "after"));
+ assertEquals(1, facade(second).loads());
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheRedisFailureIntegrationTest.java b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheRedisFailureIntegrationTest.java
new file mode 100644
index 0000000..7ca6b9e
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheRedisFailureIntegrationTest.java
@@ -0,0 +1,28 @@
+package com.Modules.NormalData.cache;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class JetCacheRedisFailureIntegrationTest extends AbstractJetCacheRedisIntegrationTest {
+ @Test
+ void cacheFailureDoesNotReturnAFalseNullBusinessValue() {
+ REDIS.getDockerClient().pauseContainerCmd(REDIS.getContainerId()).exec();
+ RuntimeException failure = null;
+ Object value = null;
+ try {
+ value = facade(first).get("redis-down", "database-value");
+ } catch (RuntimeException expected) {
+ failure = expected;
+ } finally {
+ REDIS.getDockerClient().unpauseContainerCmd(REDIS.getContainerId()).exec();
+ }
+ // JetCache 2.7 may fail-open or propagate according to error handler; either path must be explicit.
+ if (failure != null) {
+ assertNotNull(failure.getCause());
+ } else {
+ assertEquals("database-value", value);
+ }
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheSerializationIntegrationTest.java b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheSerializationIntegrationTest.java
new file mode 100644
index 0000000..2b0591a
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheSerializationIntegrationTest.java
@@ -0,0 +1,22 @@
+package com.Modules.NormalData.cache;
+
+import com.Modules.NormalData.compiled.RuleSource;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class JetCacheSerializationIntegrationTest extends AbstractJetCacheRedisIntegrationTest {
+ @Test
+ void immutableSnapshotRoundTripsAcrossContextsWithKryo5() {
+ RuleSourceSnapshot source = new RuleSourceSnapshot(7L, Collections.singletonList(
+ new RuleSource(9L, "rule", "Rule", "json", "{\"mappers\":[]}", true, "0")),
+ Instant.EPOCH);
+ facade(first).get("snapshot", source);
+ RuleSourceSnapshot decoded = (RuleSourceSnapshot) facade(second).get("snapshot", null);
+ assertEquals(7L, decoded.getDeviceId());
+ assertEquals(source.getRules().get(0).getFingerprint(), decoded.getRules().get(0).getFingerprint());
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheTwoLevelIntegrationTest.java b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheTwoLevelIntegrationTest.java
new file mode 100644
index 0000000..5c8f1e9
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/cache-it/java/com/Modules/NormalData/cache/JetCacheTwoLevelIntegrationTest.java
@@ -0,0 +1,15 @@
+package com.Modules.NormalData.cache;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class JetCacheTwoLevelIntegrationTest extends AbstractJetCacheRedisIntegrationTest {
+ @Test
+ void secondContextReadsFirstContextRemoteValueThenItsOwnL1() {
+ assertEquals("from-a", facade(first).get("two-level", "from-a"));
+ assertEquals("from-a", facade(second).get("two-level", "from-b"));
+ assertEquals("from-a", facade(second).get("two-level", "from-c"));
+ assertEquals(0, facade(second).loads());
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java
index 7b260a1..d03436f 100644
--- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java
@@ -1,8 +1,7 @@
package com.Modules.NormalData;
import cn.hutool.core.date.DateTime;
-import com.common.entity.DeviceCollectTask;
-import com.common.entity.DeviceDevice;
+import com.Modules.NormalData.compiled.*;
import com.common.entity.RuleContent.*;
import com.common.entity.SyslogNonNormalMessage;
import com.common.mapper.DmColumnMapper;
@@ -11,6 +10,8 @@ import com.common.service.*;
import com.common.util.*;
import com.config.AppProperties;
import com.config.KafkaConsumerProperties;
+import com.config.NormalizationOptimizationProperties;
+import com.Modules.NormalData.metrics.NormalizationRuleMetrics;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.json.JSONObject;
@@ -19,7 +20,15 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.common.entity.SyslogMessage;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
/**
* 日志标准化处理器（规则引擎模块 - Spring 管理的单例 Bean）
@@ -27,6 +36,45 @@ import java.util.*;
*/
@Service
public class LogNormalProcessor {
+
+ public enum ProcessingOutcome { SUCCESS, FILTERED }
+
+ /**
+ * 一条原始日志完成规则计算后的纯结果。
+ *
+ * 这里故意只保存“应该写什么”，不负责“怎样写入达梦”。活动 Kafka 链路会先并行产生
+ * 多个 {@code NormalizationResult}，再由专用批写器集中提交。这样计算线程不会继续执行
+ * 每行一个事务，同时旧的 {@link #process(String, String, String, LocalDateTime)} 入口仍可按
+ * 单条模式兼容历史调用方。
+ */
+ public static final class NormalizationResult {
+ private final ProcessingOutcome outcome;
+ private final List> rows;
+
+ private NormalizationResult(ProcessingOutcome outcome, List> rows) {
+ this.outcome = outcome;
+ this.rows = Collections.unmodifiableList(new ArrayList<>(rows));
+ }
+
+ static NormalizationResult success(List> rows) {
+ return new NormalizationResult(ProcessingOutcome.SUCCESS, rows);
+ }
+
+ static NormalizationResult filtered() {
+ return new NormalizationResult(ProcessingOutcome.FILTERED, Collections.emptyList());
+ }
+
+ public ProcessingOutcome getOutcome() {
+ return outcome;
+ }
+
+ public List> getRows() {
+ return rows;
+ }
+ }
+
+ private static final ZoneId RECEIVE_TIME_ZONE = ZoneId.of("Asia/Shanghai");
+ private static final String LEGACY_RULE_VERSION = "legacy";
private static final Logger logger = LoggerFactory.getLogger(LogNormalProcessor.class);
// ==================== 依赖注入（Spring Bean，单例，线程安全）====================
@@ -42,6 +90,12 @@ public class LogNormalProcessor {
private final DmNormalizeRuleMapper dmNormalizeRuleMapper;
private final KafkaConsumerProperties kafkaConsumerProperties;
private final AppProperties appProperties;
+ private final CompiledRuleCache compiledRuleCache;
+ private final NormalizationOptimizationProperties optimizationProperties;
+ private final NormalizationShadowComparator shadowComparator;
+ private final NormalizationShadowExecutor shadowExecutor;
+ private final NormalizationRuleMetrics globalMetrics;
+ private final Map shadowLogWindows = new ConcurrentHashMap<>();
/**
* 请求级处理上下文 —— 每个 process() 调用创建新实例，彻底解决多线程竞态条件
@@ -55,15 +109,25 @@ public class LogNormalProcessor {
Map deviceInfoMap;
String strSyslogUUID;
String strSyslogTopic;
+ LocalDateTime fallbackFirstReceivedAt;
+ LocalDateTime initialReceivedAt;
boolean isSaveNonNormal = false;
SyslogNonNormalMessage syslogNonNormalMessage = new SyslogNonNormalMessage();
- List> dmNormalizeRuleList;
+ List dmNormalizeRuleList;
List> dmColumnList;
LinkedHashMap OrginalColumnMap;
+ int savedCount;
+ int filteredCount;
+ int rejectedCount;
+ BatchRuleSnapshot batchRuleSnapshot;
+ BatchEnrichmentContext batchEnrichmentContext;
+ RuleExecutionMetrics executionMetrics;
+ MessageParseContext messageParseContext;
+ boolean shadowMode;
+ final List> normalizedRows = new ArrayList<>();
}
// ==================== 构造器注入（Spring 自动完成）====================
- @Autowired
public LogNormalProcessor(
DmNormalizeRuleService dmNormalizeRuleService,
DmColumnService dmColumnService,
@@ -77,6 +141,56 @@ public class LogNormalProcessor {
DmNormalizeRuleMapper dmNormalizeRuleMapper,
KafkaConsumerProperties kafkaConsumerProperties,
AppProperties appProperties
+ ) {
+ this(dmNormalizeRuleService, dmColumnService, syslogNormalDataService, messageService,
+ logDataFilterService, logDataCompleteService, deviceCollectTaskService,
+ deviceDeviceService, dmColumnMapper, dmNormalizeRuleMapper, kafkaConsumerProperties,
+ appProperties, defaultCompiledRuleCache(), new NormalizationOptimizationProperties());
+ }
+
+ public LogNormalProcessor(
+ DmNormalizeRuleService dmNormalizeRuleService,
+ DmColumnService dmColumnService,
+ SyslogNormalDataService syslogNormalDataService,
+ SyslogNonNormalMessageService messageService,
+ LogDataFilterService logDataFilterService,
+ LogDataCompleteService logDataCompleteService,
+ DeviceCollectTaskService deviceCollectTaskService,
+ DeviceDeviceService deviceDeviceService,
+ DmColumnMapper dmColumnMapper,
+ DmNormalizeRuleMapper dmNormalizeRuleMapper,
+ KafkaConsumerProperties kafkaConsumerProperties,
+ AppProperties appProperties,
+ CompiledRuleCache compiledRuleCache,
+ NormalizationOptimizationProperties optimizationProperties
+ ) {
+ this(dmNormalizeRuleService, dmColumnService, syslogNormalDataService, messageService,
+ logDataFilterService, logDataCompleteService, deviceCollectTaskService,
+ deviceDeviceService, dmColumnMapper, dmNormalizeRuleMapper, kafkaConsumerProperties,
+ appProperties, compiledRuleCache, optimizationProperties,
+ new NormalizationShadowComparator(), new NormalizationShadowExecutor(),
+ new NormalizationRuleMetrics());
+ }
+
+ @Autowired
+ public LogNormalProcessor(
+ DmNormalizeRuleService dmNormalizeRuleService,
+ DmColumnService dmColumnService,
+ SyslogNormalDataService syslogNormalDataService,
+ SyslogNonNormalMessageService messageService,
+ LogDataFilterService logDataFilterService,
+ LogDataCompleteService logDataCompleteService,
+ DeviceCollectTaskService deviceCollectTaskService,
+ DeviceDeviceService deviceDeviceService,
+ DmColumnMapper dmColumnMapper,
+ DmNormalizeRuleMapper dmNormalizeRuleMapper,
+ KafkaConsumerProperties kafkaConsumerProperties,
+ AppProperties appProperties,
+ CompiledRuleCache compiledRuleCache,
+ NormalizationOptimizationProperties optimizationProperties,
+ NormalizationShadowComparator shadowComparator,
+ NormalizationShadowExecutor shadowExecutor,
+ NormalizationRuleMetrics globalMetrics
) {
this.dmNormalizeRuleService = dmNormalizeRuleService;
this.dmColumnService = dmColumnService;
@@ -90,6 +204,16 @@ public class LogNormalProcessor {
this.dmNormalizeRuleMapper = dmNormalizeRuleMapper;
this.kafkaConsumerProperties = kafkaConsumerProperties;
this.appProperties = appProperties;
+ this.compiledRuleCache = compiledRuleCache;
+ this.optimizationProperties = optimizationProperties;
+ this.shadowComparator = shadowComparator;
+ this.shadowExecutor = shadowExecutor;
+ this.globalMetrics = globalMetrics;
+ }
+
+ private static CompiledRuleCache defaultCompiledRuleCache() {
+ NormalizationOptimizationProperties properties = new NormalizationOptimizationProperties();
+ return new CompiledRuleCache(new CompiledRuleCompiler(new ObjectMapper()), properties);
}
/**
@@ -99,15 +223,67 @@ public class LogNormalProcessor {
* @param syslogUUID 消息UUID
* @param syslogTopic 消息Topic（可传 null，自动使用配置值）
*/
- public void process(String logMsg, String syslogUUID, String syslogTopic) {
+ public ProcessingOutcome process(String logMsg, String syslogUUID, String syslogTopic) {
+ return process(logMsg, syslogUUID, syslogTopic, null);
+ }
+
+ /**
+ * 处理一条日志，并显式传入 Kafka 链路已解析好的首次接收时间。
+ *
+ * 该时间会同时写入标准化表的 created_at 和 log_time。created_at 表示日志首次进入系统，
+ * log_time 是现场标准化表的 RANGE 分区键。原始业务代码用本次处理时间覆盖 log_time；这里改为
+ * 稳定的首次接收时间，使 Raw、Retry 和 Replay 始终进入同一日分区。设备事件时间继续由
+ * collect_time_ts 保存，不与分区时间混用。
+ */
+ public ProcessingOutcome process(String logMsg, String syslogUUID, String syslogTopic,
+ LocalDateTime fallbackFirstReceivedAt) {
+ NormalizationResult result = prepare(logMsg, syslogUUID, syslogTopic, fallbackFirstReceivedAt);
+
+ // 兼容历史的命令行/测试入口：这些调用方没有 Kafka 批次边界，只能继续逐条落库。
+ // 当前生产活跃的 SysLogProcessor 不走这里，而是调用 prepare() 后统一批写。
+ for (Map row : result.getRows()) {
+ syslogNormalDataService.insertDynamic(row);
+ }
+ return result.getOutcome();
+ }
+
+ /**
+ * 只执行标准化计算并返回待写入行，不在本方法中写两张标准化表。
+ *
+ * 这是批量写入的关键边界：并行线程只负责 CPU/规则计算，达梦事务由 Kafka 批次汇总后的
+ * 批写器统一管理。非标日志仍沿用原有独立保存路径，不属于本次两张标准化表的改造范围。
+ */
+ public NormalizationResult prepare(String logMsg, String syslogUUID, String syslogTopic,
+ LocalDateTime fallbackFirstReceivedAt) {
+ return prepare(logMsg, syslogUUID, syslogTopic, fallbackFirstReceivedAt,
+ null, new BatchEnrichmentContext(false, deviceCollectTaskService, deviceDeviceService),
+ new RuleExecutionMetrics());
+ }
+
+ /** 活动 Kafka 链路入口：显式使用当前 poll 的规则快照、补全上下文和聚合指标。 */
+ public NormalizationResult prepare(String logMsg, String syslogUUID, String syslogTopic,
+ LocalDateTime fallbackFirstReceivedAt,
+ BatchRuleSnapshot batchRuleSnapshot,
+ BatchEnrichmentContext batchEnrichmentContext,
+ RuleExecutionMetrics executionMetrics) {
+ // [流程节点6] 按设备加载标准化规则，完成字段解析、映射、过滤、补全和数据质量检查。
ProcessContext ctx = new ProcessContext();
ctx.strSyslogUUID = syslogUUID;
ctx.strSyslogTopic = (syslogTopic != null && !syslogTopic.isEmpty())
? syslogTopic
: kafkaConsumerProperties.getTopic();
+ ctx.fallbackFirstReceivedAt = fallbackFirstReceivedAt;
ctx.isSaveNonNormal = false;
ctx.syslogNonNormalMessage = new SyslogNonNormalMessage();
+ ctx.batchRuleSnapshot = batchRuleSnapshot;
+ ctx.batchEnrichmentContext = batchEnrichmentContext == null
+ ? new BatchEnrichmentContext(false, deviceCollectTaskService, deviceDeviceService)
+ : batchEnrichmentContext;
+ ctx.executionMetrics = executionMetrics == null ? new RuleExecutionMetrics() : executionMetrics;
+ ctx.messageParseContext = new MessageParseContext(
+ optimizationProperties.getMessageParseCache().isEnabled(), globalMetrics);
+ // [流程节点6.1] 拆分 Kafka 完整消息，分别得到设备元数据头和实际 Syslog 正文。
// 解析消息
ctx.strKafkaMessage = logMsg;
if (!logMsg.isEmpty()) {
@@ -120,68 +296,285 @@ public class LogNormalProcessor {
}
// 执行处理
- init(ctx);
+ return init(ctx);
+ }
+
+ /** 活动 Kafka 链路使用已经解析好的 envelope，避免并行任务重复解析消息头。 */
+ public NormalizationResult prepare(PreparedNormalizationInput input,
+ BatchRuleSnapshot batchRuleSnapshot,
+ BatchEnrichmentContext batchEnrichmentContext,
+ RuleExecutionMetrics executionMetrics) {
+ ProcessContext ctx = new ProcessContext();
+ ctx.strSyslogUUID = input.getEventId();
+ ctx.strSyslogTopic = input.getTopic() == null || input.getTopic().isEmpty()
+ ? kafkaConsumerProperties.getTopic() : input.getTopic();
+ ctx.fallbackFirstReceivedAt = input.getReceivedAt();
+ ctx.initialReceivedAt = input.getReceivedAt();
+ ctx.strKafkaMessage = input.getFullMessage();
+ ctx.strLogMsg = input.getSyslogBody();
+ ctx.strDeviceInfo = input.getDeviceHeader();
+ ctx.deviceInfoMap = input.getDeviceInfo();
+ ctx.strDataType = input.getDataType();
+ ctx.batchRuleSnapshot = batchRuleSnapshot;
+ ctx.batchEnrichmentContext = batchEnrichmentContext == null
+ ? new BatchEnrichmentContext(false, deviceCollectTaskService, deviceDeviceService)
+ : batchEnrichmentContext;
+ ctx.executionMetrics = executionMetrics == null ? new RuleExecutionMetrics() : executionMetrics;
+ ctx.messageParseContext = new MessageParseContext(
+ optimizationProperties.getMessageParseCache().isEnabled(), globalMetrics);
+ return init(ctx);
+ }
+
+ public BatchRuleSnapshot captureRuleSnapshot(Set deviceIds) {
+ Map> rules = new LinkedHashMap<>();
+ if (deviceIds != null) {
+ for (Long deviceId : deviceIds) {
+ if (deviceId == null || deviceId <= 0) continue;
+ rules.put(deviceId, rulesForDevice(deviceId));
+ }
+ }
+ return new BatchRuleSnapshot(rules, System.currentTimeMillis());
+ }
+
+ public BatchEnrichmentContext createBatchEnrichmentContext() {
+ return new BatchEnrichmentContext(
+ optimizationProperties.getBatchEnrichmentCache().isEnabled(),
+ deviceCollectTaskService, deviceDeviceService, globalMetrics);
}
/**
* 初始化数据处理（所有运行时状态通过 ctx 传递，线程安全）
*/
- private void init(ProcessContext ctx) {
- Map mapdev = SyslogParser.parseKeyValuePairs(ctx.strDeviceInfo);
+ private NormalizationResult init(ProcessContext ctx) {
+ // [流程节点6.2] 解析设备上下文，确定 device_id 和 data_type，作为选择标准化规则的依据。
+ Map mapdev = ctx.deviceInfoMap == null
+ ? SyslogParser.parseKeyValuePairs(ctx.strDeviceInfo) : ctx.deviceInfoMap;
ctx.deviceInfoMap = mapdev;
+ if (ctx.initialReceivedAt == null) {
+ ctx.initialReceivedAt = resolveInitialReceivedAt(
+ mapdev.get("receive_time"), ctx.fallbackFirstReceivedAt);
+ }
long deviceID = Long.parseLong(mapdev.get("device_id"));
- System.out.println("device_id:" + deviceID);
+ logger.debug("标准化开始: eventId={}, deviceId={}", ctx.strSyslogUUID, deviceID);
ctx.strDataType = mapdev.get("data_type");
try {
if (deviceID > 0) {
- ctx.dmNormalizeRuleList = dmNormalizeRuleService.selectByDeviceIdAuto(deviceID);
+ ctx.dmNormalizeRuleList = ctx.batchRuleSnapshot == null
+ ? rulesForDevice(deviceID) : ctx.batchRuleSnapshot.rulesFor(deviceID);
}
- for (int i = 0; i < ctx.dmNormalizeRuleList.size(); i++) {
- try {
- Map dmNormalizeRule = ctx.dmNormalizeRuleList.get(i);
- String data_type = dmNormalizeRule.get("data_type").toString();
- System.out.println("索引: " + i + ", 值: " + dmNormalizeRule);
- System.out.println("normalrule ID: " + dmNormalizeRule.get("id")
- + ", display_name:" + dmNormalizeRule.get("display_name")
- + " data_type:" + dmNormalizeRule.get("data_type"));
-
- ctx.OrginalColumnMap = getMessageToMap(dmNormalizeRule, ctx.strLogMsg, data_type);
- if ((ctx.OrginalColumnMap == null) || (ctx.OrginalColumnMap.size() == 0)) {
- logger.error("OrginalColumnMap 对象获取为空");
- ctx.syslogNonNormalMessage.setReason("Log解析异常");
- ctx.syslogNonNormalMessage.setReasonDetail("log解析异常,返回规则名称："
- + dmNormalizeRule.get("display_name") + ",OrginalColumnMap 对象获取为空");
- if (ctx.isSaveNonNormal == false)
- SaveNonNormalMessage(ctx, deviceID, DateTime.now());
- continue;
- }
- List> destColumnList = getRuleContentMappers(dmNormalizeRule);
- List> ruleColumnList = getNormalColumnList(ctx.OrginalColumnMap, destColumnList);
- Map destMap = getColumnMap(ruleColumnList);
-
- // 数据处理-过滤规则
- if (logDataFilterService.evaluateFilterRule(
- dmNormalizeRule.get("rule_content").toString(), destMap))
- continue;
-
- // 数据处理-补全规则
- logDataCompleteService.processDataCompletion(
- dmNormalizeRule.get("rule_content").toString(), destMap);
- SaveNormalData(ctx, deviceID, DateTime.now(), destMap,
- Long.parseLong(dmNormalizeRule.get("id").toString()),
- dmNormalizeRule.get("name").toString());
-
- } catch (Exception ex) {
- logger.error("泛化规则处理失败：" + ex.getMessage());
- System.out.println(ex.getMessage());
- }
+ if (ctx.dmNormalizeRuleList == null || ctx.dmNormalizeRuleList.isEmpty()) {
+ // 原文已在进入标准化前写入 InfluxDB。设备没有绑定规则是可预期的业务终态，
+ // 继续 Retry/DLT 不会凭空产生规则，只会重复消耗 Kafka、线程和日志 I/O。
+ // 因此只跳过达梦标准化并允许 ACK；系统依赖和规则执行异常仍由下方异常链处理。
+ return NormalizationResult.filtered();
}
+ if (optimizationProperties.getCompiledRule().isEnabled()) {
+ return executeCompiled(ctx, deviceID);
+ }
+ NormalizationResult legacy = executeLegacy(ctx, deviceID);
+ shadowCompareIfSampled(ctx, deviceID, legacy);
+ return legacy;
} catch (Exception ex) {
- logger.error("处理日志消息异常：" + ex.getMessage());
- System.out.println(ex.getMessage());
+ // [SYSLOG-ALARM][NORM][已修复] 消息级异常不再转换为成功返回，确保未达终态时不会 ACK。
+ logger.error("处理日志消息异常, eventId={}", ctx.strSyslogUUID, ex);
+ throw ex instanceof RuntimeException ? (RuntimeException) ex
+ : new IllegalStateException("标准化处理失败", ex);
}
}
+ private NormalizationResult executeLegacy(ProcessContext ctx, long deviceId) {
+ for (RuleSource source : ctx.dmNormalizeRuleList) {
+ try {
+ executeLegacyRule(ctx, deviceId, source);
+ } catch (Exception failure) {
+ logger.error("泛化规则处理失败, eventId={}, ruleId={}",
+ ctx.strSyslogUUID, source.getId(), failure);
+ throw new IllegalStateException("标准化规则处理失败", failure);
+ }
+ }
+ return finish(ctx);
+ }
+
+ private NormalizationResult executeCompiled(ProcessContext ctx, long deviceId) {
+ for (RuleSource source : ctx.dmNormalizeRuleList) {
+ CompiledRule rule;
+ try {
+ rule = compiledRuleCache.get(source, ctx.executionMetrics);
+ } catch (CompiledRuleCompiler.RuleCompilationException incompatibleRule) {
+ // 编译执行是兼容性优化，不能让一条历史规则阻塞整个 Kafka 批次。
+ // 首次失败原因由 CompiledRuleCache 按 fingerprint 限频记录；此处只回退该规则。
+ try {
+ executeLegacyRule(ctx, deviceId, source);
+ } catch (Exception legacyFailure) {
+ logger.error("编译失败规则回退旧执行器后仍失败, eventId={}, ruleId={}",
+ ctx.strSyslogUUID, source.getId(), legacyFailure);
+ throw new IllegalStateException("标准化规则回退处理失败", legacyFailure);
+ }
+ continue;
+ }
+ try {
+ ParsedMessage parsed = ctx.messageParseContext.parse(rule, ctx.strLogMsg, ctx.executionMetrics);
+ if (parsed == null || parsed.getFields().isEmpty()) {
+ rejectParsedMessage(ctx, deviceId, source);
+ continue;
+ }
+ long mappingStarted = System.nanoTime();
+ Map destination = rule.map(parsed);
+ ctx.executionMetrics.addFieldMappingNanos(System.nanoTime() - mappingStarted);
+ long filterStarted = System.nanoTime();
+ if (rule.isFiltered(destination)) {
+ ctx.executionMetrics.addFilterCompleteNanos(System.nanoTime() - filterStarted);
+ ctx.filteredCount++;
+ continue;
+ }
+ rule.complete(destination);
+ ctx.executionMetrics.addFilterCompleteNanos(System.nanoTime() - filterStarted);
+ assemble(ctx, deviceId, source, destination);
+ } catch (Exception failure) {
+ if (!ctx.shadowMode) {
+ logger.error("编译规则处理失败, eventId={}, ruleId={}",
+ ctx.strSyslogUUID, source.getId(), failure);
+ }
+ throw new IllegalStateException("编译标准化规则处理失败", failure);
+ }
+ }
+ return finish(ctx);
+ }
+
+ /** 执行单条旧格式规则，供旧主路径和编译不兼容规则的精确回退共同使用。 */
+ private void executeLegacyRule(ProcessContext ctx, long deviceId, RuleSource source) {
+ Map rule = source.asLegacyMap();
+ long parseStarted = System.nanoTime();
+ globalMetrics.ruleJsonParse();
+ ctx.OrginalColumnMap = getMessageToMap(rule, ctx.strLogMsg, source.getDataType());
+ ctx.executionMetrics.addMessageParseNanos(System.nanoTime() - parseStarted);
+ if (ctx.OrginalColumnMap == null || ctx.OrginalColumnMap.isEmpty()) {
+ rejectParsedMessage(ctx, deviceId, source);
+ return;
+ }
+ long mappingStarted = System.nanoTime();
+ globalMetrics.ruleJsonParse();
+ List> destinationColumns = getRuleContentMappers(rule);
+ Map destination = getColumnMap(
+ getNormalColumnList(ctx.OrginalColumnMap, destinationColumns));
+ ctx.executionMetrics.addFieldMappingNanos(System.nanoTime() - mappingStarted);
+
+ long filterStarted = System.nanoTime();
+ globalMetrics.ruleJsonParse();
+ if (logDataFilterService.evaluateFilterRule(source.getRuleContent(), destination)) {
+ ctx.executionMetrics.addFilterCompleteNanos(System.nanoTime() - filterStarted);
+ ctx.filteredCount++;
+ return;
+ }
+ globalMetrics.ruleJsonParse();
+ logDataCompleteService.processDataCompletion(source.getRuleContent(), destination);
+ ctx.executionMetrics.addFilterCompleteNanos(System.nanoTime() - filterStarted);
+ assemble(ctx, deviceId, source, destination);
+ }
+
+ private void assemble(ProcessContext ctx, long deviceId, RuleSource source,
+ Map destination) {
+ long started = System.nanoTime();
+ ctx.normalizedRows.add(prepareNormalData(ctx, deviceId, destination,
+ source.getId(), source.getName()));
+ ctx.executionMetrics.addAssembleNanos(System.nanoTime() - started);
+ ctx.savedCount++;
+ }
+
+ private void rejectParsedMessage(ProcessContext ctx, long deviceId, RuleSource source) {
+ ctx.syslogNonNormalMessage.setReason("Log解析异常");
+ ctx.syslogNonNormalMessage.setReasonDetail("log解析异常,返回规则名称："
+ + source.getDisplayName() + ",OrginalColumnMap 对象获取为空");
+ if (!ctx.shadowMode && !ctx.isSaveNonNormal) SaveNonNormalMessage(ctx, deviceId, DateTime.now());
+ ctx.rejectedCount++;
+ }
+
+ private NormalizationResult finish(ProcessContext ctx) {
+ if (ctx.savedCount > 0) return NormalizationResult.success(ctx.normalizedRows);
+ if (ctx.filteredCount > 0 && ctx.rejectedCount == 0) return NormalizationResult.filtered();
+ throw new IllegalArgumentException("Syslog 未产生可用标准化结果: eventId=" + ctx.strSyslogUUID);
+ }
+
+ private void shadowCompareIfSampled(ProcessContext ctx, long deviceId, NormalizationResult legacy) {
+ NormalizationOptimizationProperties.CompiledRule config = optimizationProperties.getCompiledRule();
+ if (!config.isShadowCompareEnabled()
+ || !shadowComparator.sampled(ctx.strSyslogUUID, config.getShadowSampleRate())) {
+ globalMetrics.shadow("skipped");
+ return;
+ }
+ ProcessContext shadow = shadowContext(ctx);
+ Future future;
+ try {
+ future = shadowExecutor.submit(() -> executeCompiled(shadow, deviceId));
+ } catch (RejectedExecutionException saturated) {
+ globalMetrics.shadow("skipped");
+ return;
+ }
+ try {
+ NormalizationResult candidate = future.get(config.getShadowTimeoutMs(), TimeUnit.MILLISECONDS);
+ NormalizationShadowComparator.Diff diff = shadowComparator.compare(legacy, candidate);
+ ctx.executionMetrics.shadowCompared(!diff.isMatches());
+ globalMetrics.shadow(diff.isMatches() ? "match" : "mismatch");
+ if (!diff.isMatches() && shouldLogShadow(diff.getType())) {
+ logger.warn("CompiledRule影子差异: eventHash={}, ruleId={}, type={}, fields={}",
+ shadowComparator.eventHash(ctx.strSyslogUUID), firstRuleId(ctx),
+ diff.getType(), diff.getFields());
+ }
+ } catch (TimeoutException timeout) {
+ future.cancel(true);
+ globalMetrics.shadow("timeout");
+ if (shouldLogShadow("timeout")) {
+ logger.warn("CompiledRule影子超时: eventHash={}, ruleId={}, type=timeout",
+ shadowComparator.eventHash(ctx.strSyslogUUID), firstRuleId(ctx));
+ }
+ } catch (Exception failure) {
+ globalMetrics.shadow("error");
+ if (failure instanceof InterruptedException) Thread.currentThread().interrupt();
+ if (shouldLogShadow("error")) {
+ logger.warn("CompiledRule影子失败: eventHash={}, ruleId={}, type=error, error={}",
+ shadowComparator.eventHash(ctx.strSyslogUUID), firstRuleId(ctx),
+ failure.getClass().getSimpleName());
+ }
+ }
+ }
+
+ private long firstRuleId(ProcessContext ctx) {
+ return ctx.dmNormalizeRuleList == null || ctx.dmNormalizeRuleList.isEmpty()
+ ? -1L : ctx.dmNormalizeRuleList.get(0).getId();
+ }
+
+ private boolean shouldLogShadow(String type) {
+ long now = System.currentTimeMillis();
+ AtomicLong next = shadowLogWindows.computeIfAbsent(type, ignored -> new AtomicLong());
+ long current = next.get();
+ return now >= current && next.compareAndSet(current, now + 60_000L);
+ }
+
+ private ProcessContext shadowContext(ProcessContext source) {
+ ProcessContext shadow = new ProcessContext();
+ shadow.strLogMsg = source.strLogMsg;
+ shadow.strKafkaMessage = source.strKafkaMessage;
+ shadow.strDeviceInfo = source.strDeviceInfo;
+ shadow.strDataType = source.strDataType;
+ shadow.deviceInfoMap = source.deviceInfoMap;
+ shadow.strSyslogUUID = source.strSyslogUUID;
+ shadow.strSyslogTopic = source.strSyslogTopic;
+ shadow.fallbackFirstReceivedAt = source.fallbackFirstReceivedAt;
+ shadow.initialReceivedAt = source.initialReceivedAt;
+ shadow.dmNormalizeRuleList = source.dmNormalizeRuleList;
+ shadow.batchEnrichmentContext = source.batchEnrichmentContext;
+ shadow.executionMetrics = new RuleExecutionMetrics();
+ shadow.messageParseContext = new MessageParseContext(
+ optimizationProperties.getMessageParseCache().isEnabled(), globalMetrics);
+ shadow.shadowMode = true;
+ return shadow;
+ }
+
+ private List rulesForDevice(long deviceId) {
+ List loaded = dmNormalizeRuleService.selectByDeviceIdAuto(deviceId);
+ return loaded == null ? Collections.emptyList() : loaded;
+ }
+
// ==================== 以下为原有业务方法 ====================
public Cropper_paramsType getCropperParams(Map dmNormalizeRule) {
@@ -249,17 +642,11 @@ public class LogNormalProcessor {
LinkedHashMap linkMap = null;
if ("json".equals("json")) { // 默认走 json 分支，已无实例字段 strDataType
SyslogMessage msg = SyslogParser.parse(MsgContent);
- System.out.println("解析结果-log content: " + msg.getMessage().toString());
String complexJson = "{\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\"}";
- System.out.println("complexJson content: " + complexJson);
if ("dev".equals(appProperties.getRunEnvironment())) {
- LinkedHashMap flatMap = JsonParser.jsonToMap(complexJson);
- flatMap.forEach((key, value) -> System.out.println(key + " = " + value));
- return flatMap;
+ return JsonParser.jsonToMap(complexJson);
} else {
- LinkedHashMap flatMap = JsonParser.jsonToMap(msg.getMessage());
- flatMap.forEach((key, value) -> System.out.println(key + " = " + value));
- return flatMap;
+ return JsonParser.jsonToMap(msg.getMessage());
}
}
return linkMap;
@@ -303,9 +690,7 @@ public class LogNormalProcessor {
public List> getRuleList(long device_id) {
List> rulelst = dmNormalizeRuleService.selectByDeviceId((long) 1);
- if (rulelst != null) {
- System.out.println("rulelst: " + rulelst);
- } else {
+ if (rulelst == null) {
logger.error(" List> rulelst is null！");
}
return rulelst;
@@ -318,7 +703,6 @@ public class LogNormalProcessor {
for (Map.Entry entry : orginColumnMap.entrySet()) {
for (Map map : destColumnList) {
if (map.get("origin_field").toString().equals(entry.getKey())) {
- System.out.println(map);
Map normalColumMap = new LinkedHashMap<>();
normalColumMap.put("origin_field", entry.getKey());
normalColumMap.put("dest_field", map.get("dest_field").toString());
@@ -343,9 +727,9 @@ public class LogNormalProcessor {
entry.getValue().toString(), action_param.get("timezone").toString());
normalColumMap.put("dest_field_value", longTime);
} catch (Exception e) {
- logger.error("时间类型转换错误，源值：" + entry.getValue().toString()
- + ",java_date_format:" + action_param.get("java_date_format").toString());
- e.printStackTrace();
+ logger.warn("时间类型转换错误，源字段={}, 目标字段={}, javaDateFormat={}, 原因={}",
+ entry.getKey(), map.get("dest_field"), action_param.get("java_date_format"),
+ e.getMessage());
}
}
columnlist.add((HashMap) normalColumMap);
@@ -379,55 +763,96 @@ public class LogNormalProcessor {
return columnMap;
}
- public void SaveNormalData(ProcessContext ctx, long deviceId, DateTime logtime,
- Map logColumnMap,
- long normalizeRuleId, String normalizeRuleName) {
+ /**
+ * 补齐一条标准化结果所需的所有业务字段。
+ *
+ * 本方法不允许访问标准化表 Mapper，也不允许开启事务；它必须是“构造结果”的纯边界，
+ * 否则批内并行线程仍会退化为逐条 INSERT。
+ */
+ private Map prepareNormalData(ProcessContext ctx, long deviceId,
+ Map logColumnMap,
+ long normalizeRuleId, String normalizeRuleName) {
try {
if (logColumnMap.isEmpty()) {
logger.error("SaveNormalData ->logColumnMap 为空，syslogUUID:" + ctx.strSyslogUUID);
ctx.syslogNonNormalMessage.setReason("未命中规则");
ctx.syslogNonNormalMessage.setReasonDetail("失败详情：logColumnMap对象字段为空");
- if (ctx.isSaveNonNormal == false)
+ if (!ctx.shadowMode && !ctx.isSaveNonNormal)
SaveNonNormalMessage(ctx, deviceId, DateTime.now());
- return;
+ throw new IllegalArgumentException("标准化结果字段为空");
}
Map columnMap = logColumnMap;
- //补全采集设备信息
- CompletionDeviceInfo(columnMap, deviceId);
- //补全采集探针相关信息
- CompletionCollectTaskInfo(columnMap,ctx);
+ long enrichmentStarted = System.nanoTime();
+ try {
+ //补全采集设备信息
+ CompletionDeviceInfo(columnMap, deviceId, ctx);
+ //补全采集探针相关信息
+ CompletionCollectTaskInfo(columnMap,ctx);
+ } finally {
+ ctx.executionMetrics.addEnrichmentNanos(System.nanoTime() - enrichmentStarted);
+ }
//补全日志基础信息
columnMap.put("device_id", deviceId);
- columnMap.put("log_time", logtime);
- columnMap.put("id", UUID.randomUUID().toString());
+ // 现场两张标准化主表均按 log_time 做 RANGE 分区。原始业务代码会无条件用 DateTime.now()
+ // 覆盖 log_time，其实际意图是按系统接收/处理日期分区，而不是按设备事件日期分区。
+ // 这里把 log_time 和 created_at 都固定为最初 receive_time：既保持原来的分区口径，
+ // 又避免 Kafka 积压、Retry/Replay 或跨日重试改变分区。设备事件时间保留在 collect_time_ts。
+ applyInitialReceiveTime(columnMap, ctx.initialReceivedAt);
+ // 输入字符串只用于计算固定 128 位 UUID，最终 id 仍是与旧版相同长度的 36 字符 UUID。
+ // 同一 event_id + 规则 + 版本在重试/回放时生成相同 id，与稳定 log_time 组成每日本地唯一键。
+ columnMap.put("id", deterministicResultId(ctx.strSyslogUUID, normalizeRuleId, LEGACY_RULE_VERSION));
columnMap.put("normalize_rule_id", normalizeRuleId);
+ columnMap.put("rule_version", LEGACY_RULE_VERSION);
columnMap.put("normalize_rule_name", normalizeRuleName);
columnMap.put("syslog_uuid", ctx.strSyslogUUID);
columnMap.put("syslog_topic", ctx.strSyslogTopic);
- System.out.println("columnMap：" + columnMap);
- syslogNormalDataService.insertDynamic(columnMap);
+ // [SYSLOG-ALARM][LOGIO] 逐条打印完整标准化字段会增加 stdout 锁竞争和日志 I/O，并可能输出敏感字段；
+ // 建议改为阶段计数、耗时直方图和带 event_id 的采样 DEBUG 日志。
+ logger.debug("标准化结果待入库: eventId={}, normalizeRuleId={}", ctx.strSyslogUUID, normalizeRuleId);
+ // 返回完整动态列。目标表分流和真正的 INSERT 延迟到批写阶段执行。
+ return columnMap;
} catch (Exception e) {
- logger.error("SaveNormalData失败 ");
- ctx.syslogNonNormalMessage.setReason("入库失败");
- ctx.syslogNonNormalMessage.setReasonDetail("入库失败,normalizeRuleName :" + normalizeRuleName
+ logger.error("构造标准化结果失败, eventId={}, ruleId={}", ctx.strSyslogUUID, normalizeRuleId, e);
+ ctx.syslogNonNormalMessage.setReason("标准化结果构造失败");
+ ctx.syslogNonNormalMessage.setReasonDetail("标准化结果构造失败,normalizeRuleName :" + normalizeRuleName
+ ",失败详情：" + e.getMessage());
- if (ctx.isSaveNonNormal == false)
+ if (!ctx.shadowMode && !ctx.isSaveNonNormal)
SaveNonNormalMessage(ctx, deviceId, DateTime.now());
- throw new RuntimeException("SaveNormalData 失败: " + e.getMessage(), e);
+ throw new RuntimeException("构造标准化结果失败: " + e.getMessage(), e);
}
}
+ static void applyInitialReceiveTime(Map columnMap, LocalDateTime initialReceivedAt) {
+ columnMap.put("log_time", initialReceivedAt);
+ columnMap.put("created_at", initialReceivedAt);
+ }
+
+ static String deterministicResultId(String syslogUuid, long normalizeRuleId, String ruleVersion) {
+ return UUID.nameUUIDFromBytes((syslogUuid + ":" + normalizeRuleId + ":" + ruleVersion)
+ .getBytes(java.nio.charset.StandardCharsets.UTF_8)).toString();
+ }
+
//补全采集设备信息
public void CompletionDeviceInfo(Map columnMap, Long deviceID) {
+ CompletionDeviceInfo(columnMap, deviceID, null);
+ }
+
+ private void CompletionDeviceInfo(Map columnMap, Long deviceID, ProcessContext ctx) {
try {
- DeviceDevice devInfo = deviceDeviceService.getByIdSafely(Long.valueOf(deviceID).intValue());
- if (devInfo != null) {
- columnMap.put("device_ip", devInfo.getIp());
- columnMap.put("device_manufacturer", devInfo.getVendor());
- columnMap.put("device_name", devInfo.getName());
+ DeviceSnapshot device = ctx == null
+ ? DeviceSnapshot.from(deviceDeviceService.getByIdSafely(deviceID.intValue()))
+ : ctx.batchEnrichmentContext.device(deviceID.intValue());
+ if (device != null) {
+ columnMap.put("device_ip", device.getIp());
+ columnMap.put("device_manufacturer", device.getVendor());
+ columnMap.put("device_name", device.getName());
}
} catch (Exception e) {
- logger.error("CompletionDeviceInfo 失败！ID：" + deviceID);
+ logger.error("CompletionDeviceInfo 失败！ID：{}", deviceID, e);
+ if (ctx != null && ctx.batchEnrichmentContext.isEnabled()) {
+ throw e instanceof RuntimeException ? (RuntimeException) e
+ : new IllegalStateException("设备补全失败", e);
+ }
}
}
@@ -436,16 +861,20 @@ public class LogNormalProcessor {
Integer collect_id=-1;
try {
collect_id = Integer.parseInt(ctx.deviceInfoMap.get("device_collect_id"));
- DeviceCollectTask deviceCollectTask = deviceCollectTaskService.getById(collect_id);
- if (deviceCollectTask != null) {
+ CollectTaskSnapshot collectTask = ctx.batchEnrichmentContext.collectTask(collect_id);
+ if (collectTask != null) {
columnMap.put("agent_id", collect_id);
columnMap.put("origin_agent_id",collect_id.toString() );
- columnMap.put("probe_ip",deviceCollectTask.getDeviceIp() );
- columnMap.put("agent_name", deviceCollectTask.getTaskName());
- columnMap.put("origin_agent_name", deviceCollectTask.getTaskName());
+ columnMap.put("probe_ip",collectTask.getDeviceIp() );
+ columnMap.put("agent_name", collectTask.getTaskName());
+ columnMap.put("origin_agent_name", collectTask.getTaskName());
}
} catch (Exception e) {
- logger.error("CompletionCollectTaskInfo 失败！探针ID：" + collect_id);
+ logger.error("CompletionCollectTaskInfo 失败！探针ID：{}", collect_id, e);
+ if (ctx.batchEnrichmentContext.isEnabled()) {
+ throw e instanceof RuntimeException ? (RuntimeException) e
+ : new IllegalStateException("采集任务补全失败", e);
+ }
}
}
@@ -453,7 +882,7 @@ public class LogNormalProcessor {
try {
SyslogNonNormalMessage normalMessage = new SyslogNonNormalMessage();
Integer collect_id = Integer.parseInt(ctx.deviceInfoMap.get("device_collect_id"));
- DeviceCollectTask deviceCollectTask = deviceCollectTaskService.getById(collect_id);
+ CollectTaskSnapshot collectTask = ctx.batchEnrichmentContext.collectTask(collect_id);
normalMessage.setId(UUID.randomUUID().toString());
normalMessage.setDeviceId((int) deviceId);
normalMessage.setLogTime(logtime.toLocalDateTime());
@@ -462,14 +891,17 @@ public class LogNormalProcessor {
normalMessage.setSyslogTopic(ctx.strSyslogTopic);
normalMessage.setSyslogUuid(ctx.strSyslogUUID);
normalMessage.setHeaderMessage(ctx.strDeviceInfo);
+ normalMessage.setReceiveTime(ctx.initialReceivedAt);
+ // 非标表同样按 created_at 日分区；使用同一首次接收时间，避免重试后跑到另一天的分区。
+ normalMessage.setCreatedAt(ctx.initialReceivedAt.atZone(RECEIVE_TIME_ZONE).toOffsetDateTime());
normalMessage.setEtlNode("etlgo");
normalMessage.setReason(ctx.syslogNonNormalMessage.getReason());
normalMessage.setReasonDetail(ctx.syslogNonNormalMessage.getReasonDetail());
normalMessage.setRuleResult("FAIL");
normalMessage.setDeviceName(ctx.deviceInfoMap.get("device_name"));
normalMessage.setCollectTaskId(collect_id);
- if (deviceCollectTask != null)
- normalMessage.setCollectTaskName(deviceCollectTask.getTaskName());
+ if (collectTask != null)
+ normalMessage.setCollectTaskName(collectTask.getTaskName());
this.messageService.saveMessage(normalMessage);
ctx.isSaveNonNormal = true;
} catch (Exception e) {
@@ -478,6 +910,33 @@ public class LogNormalProcessor {
}
}
+ /** 将接入头中的 receive_time 写入非标表；无效值保留为空，避免伪造处理时间。 */
+ static LocalDateTime parseReceiveTime(String receiveTime) {
+ if (receiveTime == null || receiveTime.trim().isEmpty()) {
+ return null;
+ }
+ try {
+ return IngressReceiveTimeUtil.parse(receiveTime);
+ } catch (IllegalArgumentException invalidReceiveTime) {
+ logger.warn("非标日志接收时间格式异常，receiveTime={}", receiveTime);
+ return null;
+ }
+ }
+
+ /** 接入头优先；旧消息缺失/格式异常时，使用 Kafka 原始记录的稳定首次接收时间。 */
+ static LocalDateTime resolveInitialReceivedAt(String receiveTime, LocalDateTime fallbackFirstReceivedAt) {
+ LocalDateTime parsed = parseReceiveTime(receiveTime);
+ if (parsed != null) {
+ return parsed;
+ }
+ if (fallbackFirstReceivedAt != null) {
+ return fallbackFirstReceivedAt;
+ }
+ // 只有历史非 Kafka 调用入口才可能走到这里；活动 listener 必须显式传入稳定回退时间。
+ logger.warn("日志缺失稳定首次接收时间，兼容入口回退为当前时间");
+ return LocalDateTime.now(RECEIVE_TIME_ZONE);
+ }
+
public static void main(String[] args) {
// 旧入口保留，仅用于本地测试
}
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 200386d..184ac68 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
@@ -1,12 +1,24 @@
package com.Modules.NormalData;
-import com.influx.InfluxDBClient;
+import com.Modules.NormalData.compiled.BatchEnrichmentContext;
+import com.Modules.NormalData.compiled.BatchRuleSnapshot;
+import com.Modules.NormalData.compiled.RuleExecutionMetrics;
+import com.Modules.NormalData.compiled.PreparedNormalizationInput;
import com.common.util.SyslogParser;
+import com.common.util.IngressReceiveTimeUtil;
import com.config.AppProperties;
import com.config.KafkaConsumerProperties;
+import com.common.service.DmStandardizedBatchWriter;
+import com.common.service.IngressStatsProjectionWriter;
+import com.common.entity.IngressStatsEvent;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
-import com.influxdb.client.WriteApi;
+import com.influx.raw.InfluxRawPointFactory;
+import com.influx.raw.PointTimeResolver;
+import com.influx.raw.RawInfluxMetrics;
+import com.influx.raw.RawStorageProperties;
+import com.influx.raw.ReceiveTimeParser;
+import com.influxdb.exceptions.BadRequestException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
@@ -14,19 +26,24 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
-import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.UUID;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
-import java.time.LocalDate;
-import java.time.LocalDate;
+import java.util.concurrent.atomic.AtomicLong;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import com.common.util.Sm4Util;
@@ -34,6 +51,10 @@ import com.common.util.Sm4Util;
@Component
public class SysLogProcessor {
+ private static final ZoneId RECEIVE_TIME_ZONE = ZoneId.of("Asia/Shanghai");
+ private static final DateTimeFormatter MEASUREMENT_DATE_FORMATTER =
+ DateTimeFormatter.ofPattern("yyyyMMdd");
+
@Autowired
@Qualifier("logNormalProcessorExecutor")
private Executor taskExecutor;
@@ -41,7 +62,7 @@ public class SysLogProcessor {
@Value("${app.processor.batch-size:100}")
private int batchSize;
- @Value("${app.processor.process-timeout-ms:30000}")
+ @Value("${app.processor.process-timeout-ms:120000}")
private long processTimeoutMs;
@Autowired
@@ -53,10 +74,25 @@ public class SysLogProcessor {
@Autowired
private LogNormalProcessor logNormalProcessor;
+ @Autowired
+ private SyslogFailureRouter failureRouter;
+
+ @Autowired
+ private DmStandardizedBatchWriter standardizedBatchWriter;
+
+ @Autowired
+ private IngressStatsProjectionWriter ingressStatsProjectionWriter;
+
@Autowired
private com.influx.InfluxDBClient influxClient;
+ @Autowired private InfluxRawPointFactory rawPointFactory;
+ @Autowired private PointTimeResolver pointTimeResolver;
+ @Autowired private RawStorageProperties rawStorageProperties;
+ @Autowired private RawInfluxMetrics rawInfluxMetrics;
+
private final AtomicInteger totalProcessed = new AtomicInteger(0);
+ // [SYSLOG-ALARM][KACK][已修复-主链路停用] 历史辅助方法仍保留此计数用于兼容，但唯一活动 listener 不再用全局计数决定 ACK。
private final AtomicInteger currentBatchCount = new AtomicInteger(0);
@@ -69,7 +105,7 @@ public class SysLogProcessor {
try {
// 处理业务逻辑
for (String record : records) {
- System.out.println(record);
+ log.debug("兼容入口收到消息，长度={}", record == null ? 0 : record.length());
}
// 手动提交偏移量
ack.acknowledge(); // 告诉Kafka这条消息已成功处理
@@ -78,62 +114,92 @@ public class SysLogProcessor {
}
}
@KafkaListener(
+ // Raw 只处理实时原始日志，避免未到期 Retry 占用原始消费线程。
+ id = "syslog-consumer-rule-raw",
+ groupId = "${spring.kafka.consumer.group-id}",
+ autoStartup = "${syslog.raw-consumer.enabled:true}",
topics = "${spring.kafka.consumer.topic:agent-syslog-topic}",
- concurrency = "${spring.kafka.listener.concurrency}" // 多线程消费
+ containerFactory = "rawKafkaListenerContainerFactory"
)
public void consumeAndParallelProcess(List> records, Acknowledgment ack) {
+ consumeBatch(records, ack, false, "RAW");
+ }
+
+ @KafkaListener(
+ id = "syslog-consumer-rule-retry-replay",
+ groupId = "${syslog.reliability.consumer-group-id}",
+ autoStartup = "${syslog.reliability.consumer.enabled:true}",
+ topics = {
+ "${syslog.reliability.retry-1m-topic:${spring.kafka.consumer.topic}.RETRY.1M}",
+ "${syslog.reliability.retry-10m-topic:${spring.kafka.consumer.topic}.RETRY.10M}",
+ "${syslog.reliability.retry-1h-topic:${spring.kafka.consumer.topic}.RETRY.1H}",
+ "${syslog.reliability.replay-topic:${spring.kafka.consumer.topic}.REPLAY}"
+ },
+ concurrency = "${syslog.reliability.listener.concurrency:1}"
+ )
+ public void consumeRetryAndReplay(List> records, Acknowledgment ack) {
+ consumeBatch(records, ack, true, "RETRY_REPLAY");
+ }
+
+ /**
+ * 统一批处理边界：原文批量写入 InfluxDB 后，有界并行执行标准化；
+ * 只有整批全部到达可靠终态才提交 offset。
+ */
+ private void consumeBatch(List> records, Acknowledgment ack,
+ boolean enforceRetryDue, String lane) {
if (records.isEmpty()) {
ack.acknowledge();
return;
}
- log.info("开始处理批次消息，数量: {}", records.size());
- currentBatchCount.set(records.size());
-
- try {
- List> processingFutures = new ArrayList<>();
-
- // 为每条消息创建独立的处理任务
- for (ConsumerRecord record : records) {
- CompletableFuture future = CompletableFuture.runAsync(() -> {
- try {
- // 异步处理单条消息
- log.info("收到syslogmessage："+ Sm4Util.decryptCbc(record.value(), appProperties.getSm4Key()));
- processSingleMessageAsync(record);
- } catch (Exception e) {
- log.error("处理消息失败, topic: {}, partition: {}, offset: {}",
- record.topic(), record.partition(), record.offset(), e);
- } finally {
- // 计数器递减
- if (currentBatchCount.decrementAndGet() == 0) {
- log.debug("批次所有消息处理完成");
- }
- }
- }, taskExecutor);
-
- processingFutures.add(future);
- }
-
- // 等待所有处理任务完成（带超时）
- CompletableFuture allFutures = CompletableFuture.allOf(
- processingFutures.toArray(new CompletableFuture[0])
- );
-
+ long batchStartedAt = System.currentTimeMillis();
+ long prepareStartedAt = batchStartedAt;
+ BatchMetrics metrics = new BatchMetrics();
+ List preparedMessages = new ArrayList<>();
+ for (ConsumerRecord record : records) {
+ SyslogFailureRouter.MessageContext context = failureRouter.resolve(record);
try {
- allFutures.get(processTimeoutMs, TimeUnit.MILLISECONDS);
- log.info("批次处理完成，总数: {}", records.size());
- } catch (java.util.concurrent.TimeoutException e) {
- log.warn("批次处理超时，已处理: {}/{}",
- records.size() - currentBatchCount.get(), records.size());
+ if (enforceRetryDue) {
+ failureRouter.ensureRetryDue(context);
+ }
+ preparedMessages.add(prepareMessage(record, context));
+ } catch (SyslogFailureRouter.RetryNotDueException notDue) {
+ throw notDue;
+ } catch (Exception failure) {
+ metrics.record(routeFailure(record, context, failure));
}
-
- // 提交偏移量
- ack.acknowledge();
- totalProcessed.addAndGet(records.size());
-
- } catch (Exception e) {
- log.error("消费处理异常", e);
- ack.acknowledge(); // 异常情况下也提交偏移量，避免阻塞
}
+ long prepareMs = System.currentTimeMillis() - prepareStartedAt;
+
+ // 入口状态是 Kafka 已可靠接收的事实。批内按设备/探针聚合，失败时禁止 ACK；重投只重复 MIN/MAX。
+ int ingressGroups = ingressStatsProjectionWriter.writeBatch(ingressEvents(preparedMessages));
+
+ long influxStartedAt = System.currentTimeMillis();
+ List persistedMessages = persistRawBatch(preparedMessages, metrics);
+ long influxMs = System.currentTimeMillis() - influxStartedAt;
+
+ long normalizeStartedAt = System.currentTimeMillis();
+ normalizePreparedMessages(persistedMessages, metrics);
+ long normalizeMs = System.currentTimeMillis() - normalizeStartedAt;
+
+ ack.acknowledge();
+ totalProcessed.addAndGet(records.size());
+ log.info("Syslog批次完成: lane={}, topicPartitions={}, records={}, prepared={}, persisted={}, "
+ + "success={}, filtered={}, retry={}, dlt={}, normalizedRows={}, dmSuccessfulRows={}, "
+ + "dmFailedMessages={}, ingressGroups={}, prepareMs={}, influxMs={}, normalizeMs={}, "
+ + "ruleComputeMs={}, ruleSnapshotMs={}, messageParseMs={}, fieldMappingMs={}, "
+ + "filterCompleteMs={}, enrichmentMs={}, ruleAssembleMs={}, compiledRuleHit={}, "
+ + "compiledRuleMiss={}, messageParseHit={}, messageParseMiss={}, deviceUnique={}, "
+ + "collectTaskUnique={}, shadowCompared={}, shadowMismatch={}, dmWriteMs={}, totalMs={}",
+ lane, topicPartitions(records), records.size(), preparedMessages.size(), persistedMessages.size(),
+ metrics.success.get(), metrics.filtered.get(), metrics.retry.get(), metrics.dlt.get(),
+ metrics.normalizedRows.get(), metrics.dmSuccessfulRows.get(), metrics.dmFailedMessages.get(), ingressGroups,
+ prepareMs, influxMs, normalizeMs, metrics.ruleComputeMs(), metrics.ruleSnapshotMs(),
+ metrics.messageParseMs(), metrics.fieldMappingMs(), metrics.filterCompleteMs(),
+ metrics.enrichmentMs(), metrics.ruleAssembleMs(), metrics.compiledRuleHits(),
+ metrics.compiledRuleMisses(), metrics.messageParseHits(), metrics.messageParseMisses(),
+ metrics.deviceUnique(), metrics.collectTaskUnique(), metrics.shadowCompared(),
+ metrics.shadowDifferences(), metrics.dmWriteMs(),
+ System.currentTimeMillis() - batchStartedAt);
}
/**
@@ -169,6 +235,7 @@ public class SysLogProcessor {
processMessageBatch(batch, batchIndex);
} catch (Exception e) {
log.error("批次处理异常, 批次: {}", batchIndex, e);
+ throw new java.util.concurrent.CompletionException(e);
}
}, taskExecutor);
@@ -189,7 +256,7 @@ public class SysLogProcessor {
} catch (Exception e) {
log.error("分批消费处理异常", e);
- ack.acknowledge();
+ throw new IllegalStateException("分批消费失败，禁止提交 offset", e);
}
}
@@ -251,51 +318,484 @@ public class SysLogProcessor {
} catch (Exception e) {
log.error("优先级消费处理异常", e);
- ack.acknowledge();
+ throw new IllegalStateException("优先级消费失败，禁止提交 offset", e);
}
}
/**
* 异步处理单条消息
*/
- @Async("logNormalProcessorExecutor")
public void processSingleMessageAsync(ConsumerRecord record) {
- long startTime = System.currentTimeMillis();
+ processSingleMessage(record, failureRouter.resolve(record));
+ }
+ private void processSingleMessage(ConsumerRecord record,
+ SyslogFailureRouter.MessageContext messageContext) {
+ // [流程节点5] 解密 Kafka 消息，生成事件标识，保存原始事件，并把消息转交标准化处理节点。
+ // [SYSLOG-ALARM][KACK][已修复] 活动 listener 同步调用本方法，已移除嵌套 @Async/Future，异常直接返回唯一 ACK 边界。
try {
- //Message进行SM4解密
- String Sm4message=Sm4Util.decryptCbc(record.value(), appProperties.getSm4Key());
-
- String sysLogUUID =getSysLogUUID();
- String strDeviceInfo= SyslogParser.substringBeforeFirstChar(Sm4message,']');
- Map mapdev =SyslogParser.parseKeyValuePairs(strDeviceInfo);
-
- // 初始化 InfluxDB 客户端
- Point point = Point.measurement("syslog_security_"+sysLogUUID.substring(0,8) )
- .addTag("device_id", mapdev.get("device_id")) // 添加设备ID标签
- .addTag("device_collect_id", mapdev.get("device_collect_id")) // 添加探针ID标签
- .addTag("uuid", sysLogUUID) //syslog uuid
- .addTag("topic", kafkaConsumerProperties.getTopic()) //kafka topic
- .addField("message", Sm4message) // 添加字段
- .addField("receive_time", mapdev.get("receive_time")) // 添加字段
- .addField("uuid", sysLogUUID)
- .time(System.currentTimeMillis(), WritePrecision.MS) ;// 毫秒级时间戳
- influxClient.writePoint(point);
-
- System.out.println("influxdb wirte syslog ,value:"+ record.key());
- // 日志信息插入pg XdrHoneypot 表
- //insertSingleRecord( record.value());
-
- String syslogMessage= Sm4message;
- //使用注入的 Spring Bean 进行标准化处理
- logNormalProcessor.process(syslogMessage, sysLogUUID, null);
- System.out.println("insert postgres syslog ,value:"+ record.key());
- long costTime = System.currentTimeMillis() - startTime;
- log.debug("消息处理完成, offset: {}, 耗时: {}ms", record.offset(), costTime);
+ PreparedMessage message = prepareMessage(record, messageContext);
+ influxClient.writePointBlocking(message.bucket, message.point);
+ if (message.v2) rawInfluxMetrics.v2Points(1L);
+ normalizePreparedMessage(message);
} catch (Exception e) {
+ // [SYSLOG-ALARM][KACK][已修复] 异常继续抛给唯一 ACK 边界，由可靠路由器分类处理。
log.error("异步处理消息失败, offset: {}", record.offset(), e);
- // 这里可以添加重试逻辑或死信队列处理
+ throw new IllegalStateException("Syslog 处理失败", e);
+ }
+ }
+
+ /** 解密并构造原文点；此阶段不做任何 Kafka ACK。 */
+ private PreparedMessage prepareMessage(ConsumerRecord record,
+ SyslogFailureRouter.MessageContext messageContext) {
+ long startTime = System.currentTimeMillis();
+ // [流程节点5.1] 使用约定密钥解密 Kafka 密文，恢复设备元数据头和原始 Syslog。
+ String decryptedMessage = Sm4Util.decryptCbc(messageContext.getPayload(), appProperties.getSm4Key());
+ String deviceHeader = SyslogParser.substringBeforeFirstChar(decryptedMessage, ']');
+ Map deviceInfo = SyslogParser.parseKeyValuePairs(deviceHeader);
+ messageContext.setDeviceId(deviceInfo.get("device_id"));
+
+ // [流程节点5.2] 优先使用接入端 event_id；旧消息回退到稳定 topic-partition-offset ID。
+ String eventId = deviceInfo.get("event_id");
+ if (eventId == null || eventId.trim().isEmpty()) {
+ eventId = messageContext.getEventId();
+ }
+
+ String receiveTime = deviceInfo.get("receive_time");
+ ReceiveTimeParser.ParsedReceiveTime parsed = ReceiveTimeParser.parse(receiveTime);
+ boolean headerHasPointTime = deviceInfo.containsKey("point_time_ns");
+ Long envelopePointTime = messageContext.getPointTimeNs();
+ Long headerPointTime = pointTimeResolver.resolve(headerHasPointTime,
+ deviceInfo.get("point_time_ns"), parsed.getEpochMillis(), false);
+ if (envelopePointTime != null) {
+ PointTimeResolver.validate(envelopePointTime, parsed.getEpochMillis());
+ if (headerPointTime != null && !headerPointTime.equals(envelopePointTime)) {
+ throw new IllegalArgumentException("Retry/Replay point_time_ns 与原始 Kafka 载荷不一致");
+ }
+ }
+ Long pointTimeNs = envelopePointTime != null ? envelopePointTime : headerPointTime;
+ if (pointTimeNs == null && rawStorageProperties.getWriteSchema() == RawStorageProperties.WriteSchema.V2) {
+ pointTimeNs = pointTimeResolver.resolve(false, null, parsed.getEpochMillis(), true);
+ }
+ messageContext.setPointTimeNs(pointTimeNs);
+ String originalTopic = messageContext.getOriginalTopic();
+ InfluxRawPointFactory.RawPoint rawPoint = rawPointFactory.create(eventId, decryptedMessage,
+ receiveTime, deviceInfo.get("device_id"), deviceInfo.get("device_collect_id"),
+ originalTopic, parsed, pointTimeNs);
+ ResolvedInfluxTime influxTime = new ResolvedInfluxTime(parsed.getEpochMillis(),
+ parsed.getMeasurementDate(), parsed.getLocalDateTime());
+ Long deviceId = optionalLong(deviceInfo.get("device_id"));
+ Long collectId = optionalLong(deviceInfo.get("device_collect_id"));
+ PreparedNormalizationInput normalizationInput = new PreparedNormalizationInput(
+ eventId, originalTopic, influxTime.getReceivedAt(), deviceId, collectId,
+ deviceInfo.get("data_type"), SyslogParser.substringAfterFirstCloseBracket(decryptedMessage),
+ decryptedMessage, deviceHeader, deviceInfo);
+ return new PreparedMessage(record, messageContext, decryptedMessage, eventId, rawPoint.getPoint(),
+ rawPoint.getBucket(), rawPoint.isV2(), influxTime.getReceivedAt(), deviceId, collectId,
+ startTime, normalizationInput);
+ }
+
+ /**
+ * 一次 HTTP 请求持久化当前 Kafka 批次的原文。
+ * 批量请求如果因单条数据格式被 InfluxDB 拒绝，降级为逐条写入以定位坏消息；
+ * 连接、超时或服务端故障则直接保留原 offset，不制造 DLT 风暴。
+ */
+ private List persistRawBatch(List messages, BatchMetrics metrics) {
+ if (messages.isEmpty()) {
+ return messages;
+ }
+ List points = new ArrayList<>(messages.size());
+ for (PreparedMessage message : messages) {
+ points.add(message.point);
+ }
+ try {
+ String targetBucket = messages.get(0).bucket;
+ influxClient.writePointsBlocking(targetBucket, points);
+ if (messages.get(0).v2) rawInfluxMetrics.v2Points(messages.size());
+ return messages;
+ } catch (BadRequestException badBatchData) {
+ if (messages.get(0).v2) rawInfluxMetrics.v2FailedBatch();
+ log.warn("InfluxDB 批量写入发现格式错误，降级逐条定位: batchSize={}", messages.size());
+ List persisted = new ArrayList<>(messages.size());
+ for (PreparedMessage message : messages) {
+ try {
+ influxClient.writePointBlocking(message.bucket, message.point);
+ if (message.v2) rawInfluxMetrics.v2Points(1L);
+ persisted.add(message);
+ } catch (Exception singleFailure) {
+ metrics.record(routeFailure(message.record, message.context, singleFailure));
+ }
+ }
+ return persisted;
+ } catch (Exception dependencyFailure) {
+ if (messages.get(0).v2) rawInfluxMetrics.v2FailedBatch();
+ // 整批原文的持久化结果不可确认，必须让 Kafka 从原 offset 重投。
+ throw new IllegalStateException("InfluxDB 批量原文写入未取得成功确认，禁止提交 offset", dependencyFailure);
+ }
+ }
+
+ /**
+ * 历史单条兼容入口：原文已可靠持久化后，执行标准化并立即落库。
+ * 活动 Kafka 批处理链路使用 prepareNormalization()，避免在计算线程中逐条开启达梦事务。
+ */
+ private SyslogFailureRouter.TerminalState normalizePreparedMessage(PreparedMessage message) {
+ LogNormalProcessor.ProcessingOutcome outcome =
+ logNormalProcessor.process(message.decryptedMessage, message.eventId, null, message.receivedAt);
+ log.debug("消息处理完成, offset: {}, 耗时: {}ms", message.record.offset(),
+ System.currentTimeMillis() - message.startTime);
+ return outcome == LogNormalProcessor.ProcessingOutcome.FILTERED
+ ? SyslogFailureRouter.TerminalState.FILTERED
+ : SyslogFailureRouter.TerminalState.SUCCESS;
+ }
+
+ /**
+ * 只生成待写入的标准化结果，不在并行计算线程中访问两张达梦标准化表。
+ */
+ private PreparedNormalization prepareNormalization(PreparedMessage message,
+ BatchRuleSnapshot ruleSnapshot,
+ BatchEnrichmentContext enrichmentContext,
+ RuleExecutionMetrics executionMetrics) {
+ LogNormalProcessor.NormalizationResult result = logNormalProcessor.prepare(
+ message.normalizationInput, ruleSnapshot, enrichmentContext, executionMetrics);
+ log.debug("消息标准化计算完成, offset: {}, rows={}, 耗时: {}ms", message.record.offset(),
+ result.getRows().size(), System.currentTimeMillis() - message.startTime);
+ return new PreparedNormalization(message, result, null);
+ }
+
+ /**
+ * 批内有界并行标准化。任务异常必须汇总回 listener；超时或系统性失败时整批不 ACK。
+ */
+ void normalizePreparedMessages(List messages, BatchMetrics metrics) {
+ if (messages.isEmpty()) {
+ return;
+ }
+ long snapshotStartedAt = System.nanoTime();
+ Set deviceIds = new HashSet<>();
+ for (PreparedMessage message : messages) {
+ if (message.deviceId != null) deviceIds.add(message.deviceId);
+ }
+ BatchRuleSnapshot ruleSnapshot = logNormalProcessor.captureRuleSnapshot(deviceIds);
+ BatchEnrichmentContext enrichmentContext = logNormalProcessor.createBatchEnrichmentContext();
+ RuleExecutionMetrics executionMetrics = new RuleExecutionMetrics();
+ metrics.recordRuleSnapshotMs(elapsedMillis(snapshotStartedAt));
+ long ruleComputeStartedAt = System.nanoTime();
+ List> futures = new ArrayList<>(messages.size());
+ try {
+ for (PreparedMessage message : messages) {
+ futures.add(CompletableFuture.supplyAsync(() -> {
+ try {
+ return prepareNormalization(message, ruleSnapshot, enrichmentContext, executionMetrics);
+ } catch (Exception failure) {
+ // 解析/规则计算失败仍按原消息可靠路由。只有取得 Retry/DLT Broker ACK，
+ // 这个消息才会携带 terminalState 返回并允许原批次继续。
+ return new PreparedNormalization(message, null,
+ routeFailure(message.record, message.context, failure));
+ }
+ }, taskExecutor));
+ }
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
+ .get(processTimeoutMs, TimeUnit.MILLISECONDS);
+ metrics.recordRuleComputeMs(elapsedMillis(ruleComputeStartedAt));
+ metrics.recordRuleExecution(executionMetrics, enrichmentContext);
+
+ List> writeRequests = new ArrayList<>();
+ Set pendingDatabaseMessages = new HashSet<>();
+ for (CompletableFuture future : futures) {
+ PreparedNormalization prepared = future.join();
+ if (prepared.terminalState != null) {
+ metrics.record(prepared.terminalState);
+ continue;
+ }
+ if (prepared.result.getOutcome() == LogNormalProcessor.ProcessingOutcome.FILTERED) {
+ metrics.record(SyslogFailureRouter.TerminalState.FILTERED);
+ continue;
+ }
+
+ pendingDatabaseMessages.add(prepared.message);
+ for (Map row : prepared.result.getRows()) {
+ writeRequests.add(new DmStandardizedBatchWriter.WriteRequest<>(prepared.message, row));
+ }
+ }
+
+ // 批写器只返回“可归因到具体消息”的行级失败；连接、超时、事务结果未知会直接抛出，
+ // 从而让本方法退出并保留整个 Kafka 批次 offset。
+ long dmWriteStartedAt = System.nanoTime();
+ DmStandardizedBatchWriter.WriteReport writeReport =
+ standardizedBatchWriter.write(writeRequests);
+ metrics.recordDmWriteMs(elapsedMillis(dmWriteStartedAt));
+ Set failedMessages = writeReport.getFailures().keySet();
+ metrics.recordDmWrite(writeRequests.size(), writeReport.getSuccessfulRows(), failedMessages.size());
+
+ for (Map.Entry failure : writeReport.getFailures().entrySet()) {
+ PreparedMessage message = failure.getKey();
+ metrics.record(routeFailure(message.record, message.context, failure.getValue()));
+ }
+ for (PreparedMessage message : pendingDatabaseMessages) {
+ if (!failedMessages.contains(message)) {
+ metrics.record(SyslogFailureRouter.TerminalState.SUCCESS);
+ }
+ }
+ } catch (TimeoutException timeout) {
+ cancelOutstanding(futures);
+ throw new IllegalStateException("标准化批次处理超时，禁止提交 offset", timeout);
+ } catch (InterruptedException interrupted) {
+ cancelOutstanding(futures);
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("标准化批次被中断，禁止提交 offset", interrupted);
+ } catch (ExecutionException executionFailure) {
+ throw new IllegalStateException("标准化批次存在未到可靠终态的消息，禁止提交 offset",
+ executionFailure.getCause());
+ } catch (RuntimeException submitFailure) {
+ cancelOutstanding(futures);
+ throw submitFailure;
+ }
+ }
+
+ private static long elapsedMillis(long startedAtNanos) {
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAtNanos);
+ }
+
+ private void cancelOutstanding(List> futures) {
+ for (CompletableFuture future : futures) {
+ if (!future.isDone()) {
+ future.cancel(true);
+ }
+ }
+ }
+
+ /** Retry/DLT 取得 Broker ACK 才返回；系统性故障抛出以保留原 offset。 */
+ private SyslogFailureRouter.TerminalState routeFailure(ConsumerRecord record,
+ SyslogFailureRouter.MessageContext context,
+ Exception failure) {
+ SyslogFailureRouter.TerminalState state = failureRouter.route(record, context, failure);
+ if (state == SyslogFailureRouter.TerminalState.PAUSED) {
+ throw new IllegalStateException("系统性依赖故障，保留原 offset 等待恢复", failure);
+ }
+ return state;
+ }
+
+ /** 已解密、已构造原文点，但尚未提交 Kafka offset 的批内消息。 */
+ static final class PreparedMessage {
+ private final ConsumerRecord record;
+ private final SyslogFailureRouter.MessageContext context;
+ private final String decryptedMessage;
+ private final String eventId;
+ private final Point point;
+ private final String bucket;
+ private final boolean v2;
+ private final LocalDateTime receivedAt;
+ private final Long deviceId;
+ private final Long deviceCollectId;
+ private final long startTime;
+ private final PreparedNormalizationInput normalizationInput;
+
+ PreparedMessage(ConsumerRecord record,
+ SyslogFailureRouter.MessageContext context,
+ String decryptedMessage, String eventId, Point point,
+ String bucket, boolean v2,
+ LocalDateTime receivedAt, Long deviceId, Long deviceCollectId, long startTime,
+ PreparedNormalizationInput normalizationInput) {
+ this.record = record;
+ this.context = context;
+ this.decryptedMessage = decryptedMessage;
+ this.eventId = eventId;
+ this.point = point;
+ this.bucket = bucket;
+ this.v2 = v2;
+ // 这个时间已由接入 receive_time 或 Kafka 首次接收时间解析，Retry/Replay 不得重算为当前时间。
+ this.receivedAt = receivedAt;
+ this.deviceId = deviceId;
+ this.deviceCollectId = deviceCollectId;
+ this.startTime = startTime;
+ this.normalizationInput = normalizationInput;
+ }
+
+ /** 保留既有并行单测构造入口。 */
+ PreparedMessage(ConsumerRecord record,
+ SyslogFailureRouter.MessageContext context,
+ String decryptedMessage, String eventId, Point point,
+ LocalDateTime receivedAt, Long deviceId, Long deviceCollectId, long startTime) {
+ this(record, context, decryptedMessage, eventId, point, "yelangbucket", false,
+ receivedAt, deviceId, deviceCollectId, startTime,
+ legacyNormalizationInput(decryptedMessage, eventId, receivedAt, deviceId, deviceCollectId));
+ }
+
+ private static PreparedNormalizationInput legacyNormalizationInput(
+ String message, String eventId, LocalDateTime receivedAt, Long deviceId, Long collectId) {
+ String header = "";
+ String body = message;
+ Map info = Collections.emptyMap();
+ if (message != null && message.indexOf('[') >= 0 && message.indexOf(']') > message.indexOf('[')) {
+ header = SyslogParser.substringBeforeFirstChar(message, ']');
+ body = SyslogParser.substringAfterFirstCloseBracket(message);
+ info = SyslogParser.parseKeyValuePairs(header);
+ }
+ return new PreparedNormalizationInput(eventId, null, receivedAt, deviceId, collectId,
+ info.get("data_type"), body,
+ message, header, info);
+ }
+ }
+
+ private List ingressEvents(List messages) {
+ List events = new ArrayList<>(messages.size());
+ for (PreparedMessage message : messages) {
+ events.add(new IngressStatsEvent(message.eventId, message.deviceId,
+ message.deviceCollectId, message.receivedAt));
+ }
+ return events;
+ }
+
+ private static Long optionalLong(String value) {
+ if (value == null || value.trim().isEmpty()) return null;
+ try { return Long.valueOf(value.trim()); }
+ catch (NumberFormatException invalid) { return null; }
+ }
+
+ /**
+ * 并行计算阶段的结果封装。result 与 terminalState 二选一：前者等待达梦批写，后者表示该消息
+ * 已经被 Retry/DLT 可靠接管。使用显式对象可以避免在工作线程里提前修改批次 ACK 状态。
+ */
+ private static final class PreparedNormalization {
+ private final PreparedMessage message;
+ private final LogNormalProcessor.NormalizationResult result;
+ private final SyslogFailureRouter.TerminalState terminalState;
+
+ private PreparedNormalization(PreparedMessage message,
+ LogNormalProcessor.NormalizationResult result,
+ SyslogFailureRouter.TerminalState terminalState) {
+ this.message = message;
+ this.result = result;
+ this.terminalState = terminalState;
+ }
+ }
+
+ /** 单批终态计数，仅用于聚合日志，不参与 ACK 决策。 */
+ static final class BatchMetrics {
+ private final AtomicInteger success = new AtomicInteger();
+ private final AtomicInteger filtered = new AtomicInteger();
+ private final AtomicInteger retry = new AtomicInteger();
+ private final AtomicInteger dlt = new AtomicInteger();
+ private final AtomicInteger normalizedRows = new AtomicInteger();
+ private final AtomicInteger dmSuccessfulRows = new AtomicInteger();
+ private final AtomicInteger dmFailedMessages = new AtomicInteger();
+ private final AtomicLong ruleComputeMs = new AtomicLong();
+ private final AtomicLong ruleSnapshotMs = new AtomicLong();
+ private final AtomicLong dmWriteMs = new AtomicLong();
+ private volatile RuleExecutionMetrics ruleExecutionMetrics = new RuleExecutionMetrics();
+ private volatile int deviceUnique;
+ private volatile int collectTaskUnique;
+
+ void record(SyslogFailureRouter.TerminalState state) {
+ if (state == SyslogFailureRouter.TerminalState.SUCCESS) {
+ success.incrementAndGet();
+ } else if (state == SyslogFailureRouter.TerminalState.FILTERED) {
+ filtered.incrementAndGet();
+ } else if (state == SyslogFailureRouter.TerminalState.RETRY_PUBLISHED) {
+ retry.incrementAndGet();
+ } else if (state == SyslogFailureRouter.TerminalState.DLT_PUBLISHED) {
+ dlt.incrementAndGet();
+ } else {
+ throw new IllegalStateException("未到可靠终态的消息不能计入批次完成: " + state);
+ }
+ }
+
+ int successCount() {
+ return success.get();
+ }
+
+ long ruleComputeMs() {
+ return ruleComputeMs.get();
+ }
+
+ long dmWriteMs() {
+ return dmWriteMs.get();
+ }
+
+ long ruleSnapshotMs() { return ruleSnapshotMs.get(); }
+ long messageParseMs() { return ruleExecutionMetrics.getMessageParseMs(); }
+ long fieldMappingMs() { return ruleExecutionMetrics.getFieldMappingMs(); }
+ long filterCompleteMs() { return ruleExecutionMetrics.getFilterCompleteMs(); }
+ long enrichmentMs() { return ruleExecutionMetrics.getEnrichmentMs(); }
+ long ruleAssembleMs() { return ruleExecutionMetrics.getAssembleMs(); }
+ long compiledRuleHits() { return ruleExecutionMetrics.getCompiledHits(); }
+ long compiledRuleMisses() { return ruleExecutionMetrics.getCompiledMisses(); }
+ long messageParseHits() { return ruleExecutionMetrics.getParseHits(); }
+ long messageParseMisses() { return ruleExecutionMetrics.getParseMisses(); }
+ long shadowCompared() { return ruleExecutionMetrics.getShadowCompared(); }
+ long shadowDifferences() { return ruleExecutionMetrics.getShadowDifferences(); }
+ int deviceUnique() { return deviceUnique; }
+ int collectTaskUnique() { return collectTaskUnique; }
+
+ void recordRuleComputeMs(long elapsedMs) {
+ ruleComputeMs.set(Math.max(0L, elapsedMs));
+ }
+
+ void recordRuleSnapshotMs(long elapsedMs) {
+ ruleSnapshotMs.set(Math.max(0L, elapsedMs));
+ }
+
+ void recordRuleExecution(RuleExecutionMetrics executionMetrics,
+ BatchEnrichmentContext enrichmentContext) {
+ this.ruleExecutionMetrics = executionMetrics;
+ if (enrichmentContext != null) {
+ this.deviceUnique = enrichmentContext.deviceUniqueCount();
+ this.collectTaskUnique = enrichmentContext.collectTaskUniqueCount();
+ }
+ }
+
+ void recordDmWriteMs(long elapsedMs) {
+ dmWriteMs.set(Math.max(0L, elapsedMs));
+ }
+
+ /** 仅用于批次汇总观测，不参与 ACK 判断；ACK 仍由每条消息的可靠终态决定。 */
+ void recordDmWrite(int rows, int successfulRows, int failedMessages) {
+ this.normalizedRows.addAndGet(rows);
+ this.dmSuccessfulRows.addAndGet(successfulRows);
+ this.dmFailedMessages.addAndGet(failedMessages);
+ }
+ }
+
+ private String topicPartitions(List> records) {
+ Set partitions = new HashSet<>();
+ for (ConsumerRecord record : records) {
+ partitions.add(record.topic() + "-" + record.partition());
+ }
+ return partitions.toString();
+ }
+
+ /**
+ * 为 InfluxDB 选择跨 Kafka Retry/Replay 稳定的时间戳。
+ * 优先使用接入端 receive_time；旧消息缺失或格式异常时，回退到首次 Kafka 接收时间。
+ */
+ static ResolvedInfluxTime resolveInfluxTime(String receiveTime, Instant ignoredFallbackTime) {
+ ReceiveTimeParser.ParsedReceiveTime parsed = ReceiveTimeParser.parse(receiveTime);
+ return new ResolvedInfluxTime(
+ parsed.getEpochMillis(), parsed.getMeasurementDate(), parsed.getLocalDateTime());
+ }
+
+ /** InfluxDB 点的时间戳和 measurement 日期必须由同一个接收时间计算。 */
+ static final class ResolvedInfluxTime {
+ private final long timestampMs;
+ private final String measurementDate;
+ private final LocalDateTime receivedAt;
+
+ private ResolvedInfluxTime(long timestampMs, String measurementDate, LocalDateTime receivedAt) {
+ this.timestampMs = timestampMs;
+ this.measurementDate = measurementDate;
+ this.receivedAt = receivedAt;
+ }
+
+ long getTimestampMs() {
+ return timestampMs;
+ }
+
+ String getMeasurementDate() {
+ return measurementDate;
+ }
+
+ LocalDateTime getReceivedAt() {
+ return receivedAt;
}
}
@@ -331,6 +831,7 @@ public class SysLogProcessor {
} catch (Exception e) {
log.error("批次处理异常: {}", batchIndex, e);
+ throw new IllegalStateException("批次处理失败", e);
}
}
@@ -398,19 +899,4 @@ public class SysLogProcessor {
public int getCurrentBatchCount() { return currentBatchCount; }
}
- /**
- * 获取日志信息UUID，格式: yyyyMMddxxxxxxxx
- * @return
- */
- private static String getSysLogUUID ()
- {
- // 获取当前日期
- LocalDate currentDate = LocalDate.now();
- // 定义格式 (yyyyMMdd)
- DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
- // 格式化日期
- String formattedDate = currentDate.format(formatter);
- return formattedDate +"-"+ UUID.randomUUID() ;
- }
-
}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java
new file mode 100644
index 0000000..aec1153
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java
@@ -0,0 +1,355 @@
+package com.Modules.NormalData;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.common.service.DmStandardizedBatchWriter;
+import com.influxdb.exceptions.BadRequestException;
+import com.influxdb.exceptions.InfluxException;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.KafkaException;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.dao.DataAccessResourceFailureException;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+import java.net.ConnectException;
+import java.net.SocketException;
+import java.sql.SQLException;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Locale;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Kafka 异常消息可靠路由器。
+ *
+ * 该组件把“业务处理失败”与“原 offset 是否允许推进”放在同一个同步边界内：
+ * Retry/DLT 的 send Future 只有取得 Broker ACK 后才返回终态；发布失败或结果未知一律抛出，
+ * 让原消息保持未提交。这样不会用一次异步方法调用冒充消息已经被补偿链路接管。
+ */
+@Slf4j
+@Component
+public class SyslogFailureRouter {
+
+ static final String ENVELOPE_TYPE = "SYSLOG_RELIABILITY_V1";
+ private static final ZoneId EVENT_ID_ZONE = ZoneId.of("Asia/Shanghai");
+ private static final DateTimeFormatter EVENT_ID_DATE = DateTimeFormatter.ofPattern("yyyyMMdd");
+
+ private final KafkaTemplate kafkaTemplate;
+ private final ObjectMapper objectMapper;
+
+ @Value("${syslog.reliability.retry-1m-topic:${spring.kafka.consumer.topic}.RETRY.1M}")
+ private String retry1mTopic;
+ @Value("${syslog.reliability.retry-10m-topic:${spring.kafka.consumer.topic}.RETRY.10M}")
+ private String retry10mTopic;
+ @Value("${syslog.reliability.retry-1h-topic:${spring.kafka.consumer.topic}.RETRY.1H}")
+ private String retry1hTopic;
+ @Value("${syslog.reliability.dlt-topic:${spring.kafka.consumer.topic}.DLT}")
+ private String dltTopic;
+ @Value("${syslog.reliability.publish-timeout-ms:30000}")
+ private long publishTimeoutMs;
+ @Value("${syslog.reliability.consumer-version:unknown}")
+ private String consumerVersion;
+
+ public SyslogFailureRouter(KafkaTemplate kafkaTemplate, ObjectMapper objectMapper) {
+ this.kafkaTemplate = kafkaTemplate;
+ // Retry/DLT 是跨服务、跨版本协议，不能复用 Redis 缓存启用了默认类型信息的 Mapper。
+ // 否则消息会携带 Java 专用的 @class，且安全白名单可能拒绝 FailureEnvelope，
+ // 导致重试信封被误当成原始密文再次处理。副本保留 JavaTimeModule，但关闭类型信息；
+ // FailureEnvelope 上的 ignoreUnknown 同时兼容现场已经产生的带 @class 历史消息。
+ this.objectMapper = objectMapper.copy();
+ this.objectMapper.deactivateDefaultTyping();
+ }
+
+ /** 解析原始、Retry 或 Replay 消息，重试时始终返回最初的加密载荷和 event_id。 */
+ public MessageContext resolve(ConsumerRecord record) {
+ FailureEnvelope envelope = readEnvelope(record.value());
+ if (envelope == null) {
+ Instant firstReceivedAt = receivedAt(record);
+ return new MessageContext(record.value(), fallbackEventId(record, firstReceivedAt),
+ 0, null, firstReceivedAt, record.topic(), null);
+ }
+ return new MessageContext(envelope.getOriginalPayload(), envelope.getEventId(),
+ envelope.getAttemptCount(), envelope.getNextRetryAt(),
+ envelope.getFirstReceivedAt() == null ? receivedAt(record) : envelope.getFirstReceivedAt(),
+ envelope.getOriginalTopic(), envelope.getPointTimeNs());
+ }
+
+ /**
+ * 旧 Kafka 消息没有入口 event_id 时，使用首次 Kafka 接收日期和 topic-partition-offset
+ * 生成带日期前缀的稳定 ID。重投同一原始 offset 始终得到同一个值，且不会再把随机 UUID
+ * 前 8 位误当作 InfluxDB measurement 日期。
+ */
+ static String fallbackEventId(ConsumerRecord record, Instant firstReceivedAt) {
+ String deterministicId = UUID.nameUUIDFromBytes(
+ (record.topic() + ":" + record.partition() + ":" + record.offset())
+ .getBytes(java.nio.charset.StandardCharsets.UTF_8)
+ ).toString();
+ String datePrefix = EVENT_ID_DATE.format(firstReceivedAt.atZone(EVENT_ID_ZONE));
+ return datePrefix + "-" + deterministicId;
+ }
+
+ /**
+ * Retry Topic 本身不提供延时语义。未到期时短暂退避并拒绝 ACK，避免提前执行；
+ * 部署时可用支持定时投递的 Kafka 网关替换此兼容实现。
+ */
+ public void ensureRetryDue(MessageContext context) {
+ if (context.getNextRetryAt() == null) {
+ return;
+ }
+ long remaining = context.getNextRetryAt().toEpochMilli() - System.currentTimeMillis();
+ if (remaining <= 0) {
+ return;
+ }
+ try {
+ Thread.sleep(Math.min(remaining, 1000L));
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ if (Instant.now().isBefore(context.getNextRetryAt())) {
+ throw new RetryNotDueException("Retry 消息尚未到达 next_retry_at");
+ }
+ }
+
+ /** 根据异常影响范围决定暂停原消息，或在 Broker ACK 后返回可提交终态。 */
+ public TerminalState route(ConsumerRecord record, MessageContext context, Throwable failure) {
+ FailureCategory category = classify(failure);
+ if (category == FailureCategory.SYSTEMIC) {
+ // [SYSLOG-ALARM][KACK][已修复] 系统性依赖故障不倾倒 DLT，也不提交原 offset，由容器背压并等待恢复。
+ return TerminalState.PAUSED;
+ }
+
+ int nextAttempt = Math.max(0, context.getAttemptCount()) + 1;
+ boolean permanent = category == FailureCategory.PERMANENT_DATA;
+ boolean exhausted = nextAttempt > 3;
+ String targetTopic = (permanent || exhausted) ? dltTopic : retryTopic(nextAttempt);
+ Instant nextRetryAt = (permanent || exhausted) ? null : Instant.now().plusSeconds(delaySeconds(nextAttempt));
+
+ FailureEnvelope source = readEnvelope(record.value());
+ FailureEnvelope envelope = new FailureEnvelope();
+ envelope.setEnvelopeType(ENVELOPE_TYPE);
+ envelope.setEventId(context.getEventId());
+ envelope.setDeviceId(context.getDeviceId() != null ? context.getDeviceId()
+ : (source == null ? null : source.getDeviceId()));
+ envelope.setOriginalTopic(source == null ? record.topic() : source.getOriginalTopic());
+ envelope.setOriginalPartition(source == null ? record.partition() : source.getOriginalPartition());
+ envelope.setOriginalOffset(source == null ? record.offset() : source.getOriginalOffset());
+ envelope.setCurrentTopic(record.topic());
+ // [SYSLOG-ALARM][ETIME][已修复] Retry/DLT/Replay 始终携带首次 Kafka 接收时间，
+ // 不在每次转投时重置，供 InfluxDB 幂等时间戳和全链路对账使用。
+ envelope.setFirstReceivedAt(context.getFirstReceivedAt());
+ envelope.setFailedStage(detectStage(failure));
+ envelope.setErrorCode(errorCode(category));
+ envelope.setErrorType(rootCause(failure).getClass().getName());
+ envelope.setErrorMessage(sanitize(rootCause(failure).getMessage()));
+ envelope.setAttemptCount(nextAttempt);
+ envelope.setNextRetryAt(nextRetryAt);
+ envelope.setRuleId(source == null ? null : source.getRuleId());
+ envelope.setRuleVersion(source == null ? null : source.getRuleVersion());
+ envelope.setConsumerVersion(consumerVersion);
+ envelope.setOriginalPayload(context.getPayload());
+ envelope.setPointTimeNs(context.getPointTimeNs());
+
+ publishAndAwait(targetTopic, context.getEventId(), envelope);
+ TerminalState state = (permanent || exhausted)
+ ? TerminalState.DLT_PUBLISHED : TerminalState.RETRY_PUBLISHED;
+ log.warn("异常消息已由 Kafka 可靠接管: eventId={}, state={}, target={}, errorCode={}, attempt={}",
+ context.getEventId(), state, targetTopic, envelope.getErrorCode(), nextAttempt);
+ return state;
+ }
+
+ private void publishAndAwait(String topic, String key, FailureEnvelope envelope) {
+ try {
+ String body = objectMapper.writeValueAsString(envelope);
+ // [流程节点4.3] 必须等待 Broker ACK；仅调用 send() 不构成 RETRY_PUBLISHED/DLT_PUBLISHED。
+ kafkaTemplate.send(topic, key, body).get(publishTimeoutMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new FailurePublishException("发布异常消息时线程被中断，原 offset 不得提交", interrupted);
+ } catch (ExecutionException | TimeoutException publishFailure) {
+ throw new FailurePublishException("Retry/DLT 未取得 Broker ACK，原 offset 不得提交", publishFailure);
+ } catch (Exception serializationFailure) {
+ throw new FailurePublishException("异常消息序列化失败，原 offset 不得提交", serializationFailure);
+ }
+ }
+
+ FailureCategory classify(Throwable failure) {
+ // 批写器已经把失败范围缩小到单条并完成了达梦错误判别。必须先识别这些包装异常，
+ // 不能继续让内部 SQLException 的粗粒度规则把一条永久坏数据误判成整库故障，
+ // 否则该 Kafka 分区会永久停在同一 offset。
+ if (containsType(failure, DmStandardizedBatchWriter.PermanentRowWriteException.class)) {
+ return FailureCategory.PERMANENT_DATA;
+ }
+ if (containsType(failure, DmStandardizedBatchWriter.RetryableRowWriteException.class)) {
+ return FailureCategory.TRANSIENT;
+ }
+ Throwable root = rootCause(failure);
+ // InfluxDB 400 数据格式错误只影响当前消息，可直接隔离；
+ // 连接、超时、5xx 等 InfluxDB 故障属于系统性依赖故障，必须保留原 offset，避免 DLT 风暴。
+ if (root instanceof BadRequestException) {
+ return FailureCategory.PERMANENT_DATA;
+ }
+ if (root instanceof InfluxException) {
+ return FailureCategory.SYSTEMIC;
+ }
+ if (root instanceof DataAccessResourceFailureException || root instanceof SQLException
+ || root instanceof ConnectException || root instanceof KafkaException
+ || root instanceof RejectedExecutionException) {
+ return FailureCategory.SYSTEMIC;
+ }
+ if (root instanceof SocketException || root instanceof TimeoutException) {
+ return FailureCategory.TRANSIENT;
+ }
+ if (containsInChain(failure, "规则", "rule_content", "configuration", "配置")) {
+ return FailureCategory.CONFIGURATION_RULE;
+ }
+ if (root instanceof IllegalArgumentException || root instanceof SecurityException) {
+ return FailureCategory.PERMANENT_DATA;
+ }
+ return FailureCategory.UNKNOWN;
+ }
+
+ private boolean containsType(Throwable failure, Class expectedType) {
+ Throwable current = failure;
+ while (current != null) {
+ if (expectedType.isInstance(current)) {
+ return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+
+ private boolean containsInChain(Throwable failure, String... tokens) {
+ Throwable current = failure;
+ while (current != null) {
+ String message = String.valueOf(current.getMessage()).toLowerCase(Locale.ROOT);
+ for (String token : tokens) {
+ if (message.contains(token.toLowerCase(Locale.ROOT))) return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+
+ private FailureEnvelope readEnvelope(String value) {
+ if (value == null || value.length() < 2 || value.charAt(0) != '{') {
+ return null;
+ }
+ try {
+ FailureEnvelope envelope = objectMapper.readValue(value, FailureEnvelope.class);
+ return envelope != null && ENVELOPE_TYPE.equals(envelope.getEnvelopeType()) ? envelope : null;
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ /** Kafka 时间戳是旧消息缺失 receive_time 时的稳定回退值。 */
+ private Instant receivedAt(ConsumerRecord record) {
+ return record.timestamp() > 0 ? Instant.ofEpochMilli(record.timestamp()) : Instant.now();
+ }
+
+ private String retryTopic(int attempt) {
+ if (attempt == 1) return retry1mTopic;
+ if (attempt == 2) return retry10mTopic;
+ return retry1hTopic;
+ }
+
+ private long delaySeconds(int attempt) {
+ if (attempt == 1) return 60L;
+ if (attempt == 2) return 600L;
+ return 3600L;
+ }
+
+ private String errorCode(FailureCategory category) {
+ return "SYSLOG_" + category.name();
+ }
+
+ private String detectStage(Throwable failure) {
+ String text = (failure.getMessage() + " " + rootCause(failure).getClass().getSimpleName()).toLowerCase(Locale.ROOT);
+ if (text.contains("decrypt") || text.contains("sm4") || text.contains("解密")) return "DECRYPT";
+ if (text.contains("sql") || text.contains("database") || text.contains("入库")) return "DB_WRITE";
+ if (text.contains("parse") || text.contains("解析")) return "PARSE";
+ return "NORMALIZE";
+ }
+
+ private Throwable rootCause(Throwable failure) {
+ Throwable current = failure;
+ while (current.getCause() != null && current.getCause() != current) current = current.getCause();
+ return current;
+ }
+
+ private String sanitize(String message) {
+ if (message == null) return "unknown";
+ String sanitized = message.replaceAll("(?i)(password|token|secret|key)\\s*[=:]\\s*[^,;\\s]+", "$1=***")
+ .replaceAll("[\\r\\n\\t]+", " ");
+ return sanitized.length() <= 2048 ? sanitized : sanitized.substring(0, 2048);
+ }
+
+ enum FailureCategory { PERMANENT_DATA, CONFIGURATION_RULE, TRANSIENT, SYSTEMIC, UNKNOWN }
+ public enum TerminalState { SUCCESS, FILTERED, RETRY_PUBLISHED, DLT_PUBLISHED, PAUSED }
+
+ @Data
+ public static class MessageContext {
+ private final String payload;
+ private final String eventId;
+ private final int attemptCount;
+ private final Instant nextRetryAt;
+ private final Instant firstReceivedAt;
+ private final String originalTopic;
+ private Long pointTimeNs;
+ private String deviceId;
+
+ public MessageContext(String payload, String eventId, int attemptCount,
+ Instant nextRetryAt, Instant firstReceivedAt) {
+ this(payload, eventId, attemptCount, nextRetryAt, firstReceivedAt, null, null);
+ }
+
+ public MessageContext(String payload, String eventId, int attemptCount,
+ Instant nextRetryAt, Instant firstReceivedAt,
+ String originalTopic, Long pointTimeNs) {
+ this.payload = payload; this.eventId = eventId; this.attemptCount = attemptCount;
+ this.nextRetryAt = nextRetryAt; this.firstReceivedAt = firstReceivedAt;
+ this.originalTopic = originalTopic; this.pointTimeNs = pointTimeNs;
+ }
+ }
+
+ @Data
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public static class FailureEnvelope {
+ private String envelopeType;
+ private String eventId;
+ private String deviceId;
+ private String originalTopic;
+ private Integer originalPartition;
+ private Long originalOffset;
+ private String currentTopic;
+ private Instant firstReceivedAt;
+ private String failedStage;
+ private String errorCode;
+ private String errorType;
+ private String errorMessage;
+ private int attemptCount;
+ private Instant nextRetryAt;
+ private String ruleId;
+ private String ruleVersion;
+ private String consumerVersion;
+ private String originalPayload;
+ private Long pointTimeNs;
+ }
+
+ public static class FailurePublishException extends RuntimeException {
+ public FailurePublishException(String message, Throwable cause) { super(message, cause); }
+ }
+
+ public static class RetryNotDueException extends RuntimeException {
+ public RetryNotDueException(String message) { super(message); }
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/JetCacheLookupProvider.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/JetCacheLookupProvider.java
new file mode 100644
index 0000000..f638a41
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/JetCacheLookupProvider.java
@@ -0,0 +1,284 @@
+package com.Modules.NormalData.cache;
+
+import com.Modules.NormalData.compiled.NormalizationCacheNames;
+import com.alicp.jetcache.Cache;
+import com.alicp.jetcache.CacheConfig;
+import com.alicp.jetcache.CacheManager;
+import com.alicp.jetcache.CacheResult;
+import com.alicp.jetcache.MultiLevelCacheConfig;
+import com.alicp.jetcache.anno.CacheType;
+import com.alicp.jetcache.support.Fastjson2KeyConvertor;
+import com.alicp.jetcache.support.Kryo5ValueDecoder;
+import com.alicp.jetcache.support.Kryo5ValueEncoder;
+import com.alicp.jetcache.template.QuickConfig;
+import com.common.entity.DeviceCollectTask;
+import com.common.entity.DeviceDevice;
+import com.config.NormalizationOptimizationProperties;
+import com.Modules.NormalData.metrics.NormalizationRuleMetrics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+
+/** 新 provider：只访问 JetCache BOTH，绝不嵌套 Spring Cache 或 SafeCacheUtil。 */
+@Component
+public class JetCacheLookupProvider {
+ private static final Logger log = LoggerFactory.getLogger(JetCacheLookupProvider.class);
+ private final NormalizationLookupDataLoader loader;
+ private final NormalizationOptimizationProperties properties;
+ private final NormalizationRuleMetrics metrics;
+ private final ConcurrentHashMap> ruleLoads = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap> deviceLoads = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap> collectTaskLoads =
+ new ConcurrentHashMap<>();
+ private final Set loggedCacheFailures = ConcurrentHashMap.newKeySet();
+
+ private final Cache deviceCache;
+ private final Cache collectTaskCache;
+ private final Cache rulesCache;
+
+ @Autowired
+ public JetCacheLookupProvider(NormalizationLookupDataLoader loader,
+ NormalizationOptimizationProperties properties,
+ NormalizationRuleMetrics metrics,
+ CacheManager cacheManager) {
+ this(loader, properties, metrics,
+ cacheManager.getOrCreateCache(quickConfig(
+ NormalizationCacheNames.DEVICE_BY_ID,
+ properties.getEnrichmentCache().getRemoteTtlSeconds(),
+ properties.getEnrichmentCache().getLocalTtlSeconds())),
+ cacheManager.getOrCreateCache(quickConfig(
+ NormalizationCacheNames.COLLECT_TASK_BY_ID,
+ properties.getEnrichmentCache().getRemoteTtlSeconds(),
+ properties.getEnrichmentCache().getLocalTtlSeconds())),
+ cacheManager.getOrCreateCache(quickConfig(
+ NormalizationCacheNames.RULES_BY_DEVICE,
+ properties.getRuleSourceCache().getRemoteTtlSeconds(),
+ properties.getRuleSourceCache().getLocalTtlSeconds())));
+ }
+
+ JetCacheLookupProvider(NormalizationLookupDataLoader loader,
+ NormalizationOptimizationProperties properties,
+ NormalizationRuleMetrics metrics,
+ Cache deviceCache,
+ Cache collectTaskCache,
+ Cache rulesCache) {
+ this.loader = loader;
+ this.properties = properties;
+ this.metrics = metrics;
+ this.deviceCache = deviceCache;
+ this.collectTaskCache = collectTaskCache;
+ this.rulesCache = rulesCache;
+ }
+
+ static QuickConfig quickConfig(String name, int remoteTtlSeconds, int localTtlSeconds) {
+ return QuickConfig.newBuilder(name)
+ .cacheType(CacheType.BOTH)
+ .expire(Duration.ofSeconds(remoteTtlSeconds))
+ .localExpire(Duration.ofSeconds(localTtlSeconds))
+ .localLimit(10_000)
+ .syncLocal(true)
+ .penetrationProtect(true)
+ .keyConvertor(Fastjson2KeyConvertor.INSTANCE)
+ .valueEncoder(Kryo5ValueEncoder.INSTANCE)
+ .valueDecoder(Kryo5ValueDecoder.INSTANCE)
+ .build();
+ }
+
+ public DeviceDevice device(Integer id) {
+ DeviceDevice cached = safeGet(deviceCache, id, "device");
+ if (cached != null) return cached;
+ FutureTask candidate = new FutureTask<>(() -> loader.loadDevice(id));
+ FutureTask task = deviceLoads.putIfAbsent(id, candidate);
+ if (task == null) {
+ task = candidate;
+ metrics.enrichment("device", "load");
+ } else {
+ metrics.enrichment("device", "hit");
+ }
+ return awaitAndCache(task, deviceLoads, id, deviceCache, "device");
+ }
+
+ public DeviceCollectTask collectTask(Integer id) {
+ DeviceCollectTask cached = safeGet(collectTaskCache, id, "collect_task");
+ if (cached != null) return cached;
+ FutureTask candidate = new FutureTask<>(() -> loader.loadCollectTask(id));
+ FutureTask task = collectTaskLoads.putIfAbsent(id, candidate);
+ if (task == null) {
+ task = candidate;
+ metrics.enrichment("collect_task", "load");
+ } else {
+ metrics.enrichment("collect_task", "hit");
+ }
+ return awaitAndCache(task, collectTaskLoads, id, collectTaskCache, "collect_task");
+ }
+
+ /** 动态 put 让空快照只存 10 秒；FutureTask 保证本 JVM 同 key 并发只访问一次达梦。 */
+ public RuleSourceSnapshot rules(Long deviceId) {
+ RuleSourceSnapshot cached = cachedRuleSnapshot(deviceId);
+ if (cached != null) {
+ return cached;
+ }
+ FutureTask candidate = new FutureTask<>(() -> loader.loadRules(deviceId));
+ FutureTask task = ruleLoads.putIfAbsent(deviceId, candidate);
+ if (task == null) {
+ task = candidate;
+ metrics.ruleSource("load");
+ } else {
+ metrics.ruleSource("suppressed");
+ }
+ task.run();
+ try {
+ RuleSourceSnapshot loaded = task.get();
+ int remoteTtl = loaded.isEmpty()
+ ? properties.getRuleSourceCache().getEmptyTtlSeconds()
+ : properties.getRuleSourceCache().getRemoteTtlSeconds();
+ try {
+ putRuleSnapshot(deviceId, loaded, remoteTtl);
+ } catch (RuntimeException cacheFailure) {
+ metrics.ruleSource("failure");
+ logCacheFailure("rule-source", "put", cacheFailure);
+ }
+ return loaded;
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("加载设备规则快照被中断: " + deviceId, interrupted);
+ } catch (ExecutionException failure) {
+ metrics.ruleSource("failure");
+ throw new IllegalStateException("加载设备规则快照失败: " + deviceId, failure.getCause());
+ } finally {
+ ruleLoads.remove(deviceId, task);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private RuleSourceSnapshot cachedRuleSnapshot(Long deviceId) {
+ try {
+ List> layers = cacheLayers(rulesCache);
+ if (layers.size() >= 2) {
+ RuleSourceSnapshot local = layers.get(0).get(deviceId);
+ if (local != null) {
+ metrics.ruleSource("l1_hit");
+ return local;
+ }
+ }
+ RuleSourceSnapshot value = rulesCache.get(deviceId);
+ if (value != null) metrics.ruleSource("l2_hit");
+ return value;
+ } catch (RuntimeException cacheFailure) {
+ metrics.ruleSource("failure");
+ logCacheFailure("rule-source", "get", cacheFailure);
+ return null;
+ }
+ }
+
+ private void putRuleSnapshot(Long deviceId, RuleSourceSnapshot loaded, int remoteTtl) {
+ List> layers = cacheLayers(rulesCache);
+ if (layers.size() < 2) {
+ putChecked(rulesCache, deviceId, loaded, remoteTtl);
+ return;
+ }
+ int localTtl = loaded.isEmpty()
+ ? Math.min(properties.getRuleSourceCache().getEmptyTtlSeconds(),
+ properties.getRuleSourceCache().getLocalTtlSeconds())
+ : properties.getRuleSourceCache().getLocalTtlSeconds();
+ putChecked(layers.get(0), deviceId, loaded, localTtl);
+ putChecked(layers.get(1), deviceId, loaded, remoteTtl);
+ }
+
+ private T safeGet(Cache cache, Integer id, String entity) {
+ try {
+ T value = cache.get(id);
+ if (value != null) metrics.enrichment(entity, "hit");
+ return value;
+ } catch (RuntimeException cacheFailure) {
+ metrics.enrichment(entity, "failure");
+ logCacheFailure(entity, "get", cacheFailure);
+ return null;
+ }
+ }
+
+ private T awaitAndCache(FutureTask task,
+ ConcurrentHashMap> flights,
+ Integer id, Cache cache, String entity) {
+ task.run();
+ try {
+ T value = task.get();
+ if (value != null) putEnrichment(cache, id, value, entity);
+ return value;
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ metrics.enrichment(entity, "failure");
+ throw new IllegalStateException(entity + " 缓存加载被中断: " + id, interrupted);
+ } catch (ExecutionException failure) {
+ metrics.enrichment(entity, "failure");
+ throw new IllegalStateException(entity + " 缓存加载失败: " + id, failure.getCause());
+ } finally {
+ flights.remove(id, task);
+ }
+ }
+
+ private void putEnrichment(Cache cache, Integer id, T value, String entity) {
+ try {
+ List> layers = cacheLayers(cache);
+ int localTtl = properties.getEnrichmentCache().getLocalTtlSeconds();
+ int remoteTtl = properties.getEnrichmentCache().getRemoteTtlSeconds();
+ if (layers.size() >= 2) {
+ putChecked(layers.get(0), id, value, localTtl);
+ putChecked(layers.get(1), id, value, remoteTtl);
+ } else {
+ putChecked(cache, id, value, remoteTtl);
+ }
+ } catch (RuntimeException cacheFailure) {
+ // 缓存故障降级为已成功的 mapper 事实；下次调用可以重试缓存。
+ metrics.enrichment(entity, "failure");
+ logCacheFailure(entity, "put", cacheFailure);
+ }
+ }
+
+ /**
+ * JetCache 2.7.8 的 MultiLevelCache.unwrap(MultiLevelCache.class) 会遍历子缓存，
+ * 不会返回 MultiLevelCache 自身，因此必须从配置中取得分层缓存。
+ */
+ @SuppressWarnings("unchecked")
+ static List> cacheLayers(Cache cache) {
+ if (cache == null) return Collections.emptyList();
+ CacheConfig config = cache.config();
+ if (config instanceof MultiLevelCacheConfig) {
+ List> layers = ((MultiLevelCacheConfig) config).getCaches();
+ if (layers != null && !layers.isEmpty()) return layers;
+ }
+ return Collections.singletonList(cache);
+ }
+
+ private static void putChecked(Cache cache, K key, V value, int ttlSeconds) {
+ CacheResult result = cache.PUT(key, value, ttlSeconds, TimeUnit.SECONDS);
+ if (!result.isSuccess()) {
+ throw new IllegalStateException("JetCache PUT 失败: code=" + result.getResultCode()
+ + ", reason=" + safeReason(result.getMessage()));
+ }
+ }
+
+ private void logCacheFailure(String cache, String operation, RuntimeException failure) {
+ String key = cache + ':' + operation;
+ if (loggedCacheFailures.add(key)) {
+ log.warn("JetCache 访问失败，已降级数据库: cache={}, operation={}, error={}, reason={}",
+ cache, operation, failure.getClass().getSimpleName(), safeReason(failure.getMessage()));
+ }
+ }
+
+ private static String safeReason(String reason) {
+ if (reason == null || reason.trim().isEmpty()) return "unknown";
+ String oneLine = reason.replace('\r', ' ').replace('\n', ' ').trim();
+ return oneLine.length() <= 240 ? oneLine : oneLine.substring(0, 240);
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/LegacySpringCacheLookupProvider.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/LegacySpringCacheLookupProvider.java
new file mode 100644
index 0000000..86a06bb
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/LegacySpringCacheLookupProvider.java
@@ -0,0 +1,31 @@
+package com.Modules.NormalData.cache;
+
+import com.common.entity.DeviceCollectTask;
+import com.common.entity.DeviceDevice;
+import org.springframework.cache.annotation.Cacheable;
+import org.springframework.stereotype.Component;
+
+/** 回滚 provider：每个查询只经过既有 Spring Redis Cache 一层。 */
+@Component
+public class LegacySpringCacheLookupProvider {
+ private final NormalizationLookupDataLoader loader;
+
+ public LegacySpringCacheLookupProvider(NormalizationLookupDataLoader loader) {
+ this.loader = loader;
+ }
+
+ @Cacheable(cacheNames = "device", key = "'device:id:' + #id", sync = true)
+ public DeviceDevice device(Integer id) {
+ return loader.loadDevice(id);
+ }
+
+ @Cacheable(cacheNames = "collectTask", key = "'id:' + #id", sync = true)
+ public DeviceCollectTask collectTask(Integer id) {
+ return loader.loadCollectTask(id);
+ }
+
+ @Cacheable(cacheNames = "normalizeRule", key = "'deviceId:' + #deviceId", sync = true)
+ public RuleSourceSnapshot rules(Long deviceId) {
+ return loader.loadRules(deviceId);
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCacheInvalidationService.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCacheInvalidationService.java
new file mode 100644
index 0000000..1039309
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationCacheInvalidationService.java
@@ -0,0 +1,70 @@
+package com.Modules.NormalData.cache;
+
+import com.Modules.NormalData.compiled.NormalizationCacheInvalidator;
+import org.springframework.cache.Cache;
+import org.springframework.cache.CacheManager;
+import org.springframework.stereotype.Component;
+import com.Modules.NormalData.metrics.NormalizationRuleMetrics;
+
+import java.util.Collection;
+
+/** 同时覆盖当前活动 provider 的统一精确失效入口；单级失败不覆盖数据库提交结果。 */
+@Component
+public class NormalizationCacheInvalidationService {
+ private final CacheManager springCacheManager;
+ private final NormalizationCacheInvalidator jetCacheInvalidator;
+ private final NormalizationRuleMetrics metrics;
+
+ public NormalizationCacheInvalidationService(CacheManager springCacheManager,
+ NormalizationCacheInvalidator jetCacheInvalidator,
+ NormalizationRuleMetrics metrics) {
+ this.springCacheManager = springCacheManager;
+ this.jetCacheInvalidator = jetCacheInvalidator;
+ this.metrics = metrics;
+ }
+
+ public void invalidateDevice(Integer id) {
+ evictSpring("device", "device:id:" + id);
+ metrics.invalidation("local");
+ jetCacheInvalidator.device(id);
+ metrics.invalidation("remote");
+ }
+
+ public void invalidateCollectTask(Integer id) {
+ evictSpring("collectTask", "id:" + id);
+ metrics.invalidation("local");
+ jetCacheInvalidator.collectTask(id);
+ metrics.invalidation("remote");
+ }
+
+ public void invalidateCollectTaskViews(Integer... deviceIds) {
+ evictSpring("collectTask", "all");
+ if (deviceIds == null) return;
+ for (Integer deviceId : deviceIds) {
+ if (deviceId != null) evictSpring("collectTask", "device:" + deviceId);
+ }
+ }
+
+ public void invalidateRulesForDevice(Long deviceId) {
+ evictSpring("normalizeRule", "deviceId:" + deviceId);
+ metrics.invalidation("local");
+ jetCacheInvalidator.rulesForDevice(deviceId);
+ metrics.invalidation("remote");
+ }
+
+ public void invalidateRuleEntity(Long ruleId) {
+ evictSpring("normalizeRule", "id:" + ruleId);
+ }
+
+ public void invalidateRulesForDevices(Collection deviceIds) {
+ if (deviceIds == null) return;
+ for (Long deviceId : deviceIds) {
+ if (deviceId != null) invalidateRulesForDevice(deviceId);
+ }
+ }
+
+ private void evictSpring(String cacheName, Object key) {
+ Cache cache = springCacheManager.getCache(cacheName);
+ if (cache != null) cache.evict(key);
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationLookupDataLoader.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationLookupDataLoader.java
new file mode 100644
index 0000000..dff0dd8
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationLookupDataLoader.java
@@ -0,0 +1,73 @@
+package com.Modules.NormalData.cache;
+
+import com.Modules.NormalData.compiled.RuleSource;
+import com.common.entity.DeviceCollectTask;
+import com.common.entity.DeviceDevice;
+import com.common.mapper.DeviceCollectTaskMapper;
+import com.common.mapper.DeviceDeviceMapper;
+import com.common.mapper.DmNormalizeRuleMapper;
+import org.springframework.stereotype.Component;
+
+import java.sql.Clob;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** 三个 provider 共用的无缓存数据加载器，保证 CLOB 在 mapper 返回后立即脱离 JDBC 生命周期。 */
+@Component
+public class NormalizationLookupDataLoader {
+ private final DeviceDeviceMapper deviceMapper;
+ private final DeviceCollectTaskMapper collectTaskMapper;
+ private final DmNormalizeRuleMapper ruleMapper;
+
+ public NormalizationLookupDataLoader(DeviceDeviceMapper deviceMapper,
+ DeviceCollectTaskMapper collectTaskMapper,
+ DmNormalizeRuleMapper ruleMapper) {
+ this.deviceMapper = deviceMapper;
+ this.collectTaskMapper = collectTaskMapper;
+ this.ruleMapper = ruleMapper;
+ }
+
+ public DeviceDevice loadDevice(Integer id) {
+ return deviceMapper.selectById(id);
+ }
+
+ public DeviceCollectTask loadCollectTask(Integer id) {
+ return collectTaskMapper.selectById(id);
+ }
+
+ public RuleSourceSnapshot loadRules(Long deviceId) {
+ List> rows = ruleMapper.selectByDeviceId(deviceId);
+ if (rows == null || rows.isEmpty()) {
+ return new RuleSourceSnapshot(deviceId, Collections.emptyList(), Instant.now());
+ }
+ List rules = new ArrayList<>(rows.size());
+ for (Map row : rows) {
+ if (row == null) continue;
+ Map detached = new LinkedHashMap<>(row.size());
+ for (Map.Entry entry : row.entrySet()) {
+ detached.put(entry.getKey(), detach(entry.getValue()));
+ }
+ rules.add(RuleSource.fromMap(detached));
+ }
+ return new RuleSourceSnapshot(deviceId, rules, Instant.now());
+ }
+
+ private static Object detach(Object value) {
+ if (value == null) return null;
+ if (value instanceof Clob) {
+ try {
+ Clob clob = (Clob) value;
+ long length = clob.length();
+ return length == 0 ? "" : clob.getSubString(1, Math.toIntExact(length));
+ } catch (Exception failure) {
+ throw new IllegalStateException("规则 CLOB 转 String 失败", failure);
+ }
+ }
+ if (value.getClass().getName().startsWith("dm.jdbc.")) return value.toString();
+ return value;
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationLookupFacade.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationLookupFacade.java
new file mode 100644
index 0000000..6cb5bbb
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizationLookupFacade.java
@@ -0,0 +1,35 @@
+package com.Modules.NormalData.cache;
+
+import com.common.entity.DeviceCollectTask;
+import com.common.entity.DeviceDevice;
+import com.config.NormalizationOptimizationProperties;
+import org.springframework.stereotype.Component;
+
+/** 唯一热查询路由；一次调用严格只进入一个 provider。 */
+@Component
+public class NormalizationLookupFacade {
+ private final LegacySpringCacheLookupProvider legacy;
+ private final JetCacheLookupProvider jetCache;
+ private final NormalizationOptimizationProperties properties;
+
+ public NormalizationLookupFacade(LegacySpringCacheLookupProvider legacy,
+ JetCacheLookupProvider jetCache,
+ NormalizationOptimizationProperties properties) {
+ this.legacy = legacy;
+ this.jetCache = jetCache;
+ this.properties = properties;
+ }
+
+ public DeviceDevice device(Integer id) {
+ return properties.isJetcacheEnabled() ? jetCache.device(id) : legacy.device(id);
+ }
+
+ public DeviceCollectTask collectTask(Integer id) {
+ return properties.isJetcacheEnabled() ? jetCache.collectTask(id) : legacy.collectTask(id);
+ }
+
+ public RuleSourceSnapshot rules(Long deviceId) {
+ return properties.isJetcacheEnabled() && properties.getRuleSourceCache().isEnabled()
+ ? jetCache.rules(deviceId) : legacy.rules(deviceId);
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleCacheInvalidationListener.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleCacheInvalidationListener.java
new file mode 100644
index 0000000..f6d93c8
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleCacheInvalidationListener.java
@@ -0,0 +1,56 @@
+package com.Modules.NormalData.cache;
+
+import com.Modules.NormalData.compiled.CompiledRuleCache;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+import com.Modules.NormalData.metrics.NormalizationRuleMetrics;
+
+/** 只有数据库事务真正提交后才清除 raw/compiled 缓存并广播 L1 失效。 */
+@Component
+public class NormalizeRuleCacheInvalidationListener {
+ private static final Logger log = LoggerFactory.getLogger(NormalizeRuleCacheInvalidationListener.class);
+
+ private final NormalizationCacheInvalidationService invalidationService;
+ private final CompiledRuleCache compiledRuleCache;
+ private final NormalizationRuleMetrics metrics;
+
+ public NormalizeRuleCacheInvalidationListener(NormalizationCacheInvalidationService invalidationService,
+ CompiledRuleCache compiledRuleCache,
+ NormalizationRuleMetrics metrics) {
+ this.invalidationService = invalidationService;
+ this.compiledRuleCache = compiledRuleCache;
+ this.metrics = metrics;
+ }
+
+ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+ public void afterCommit(NormalizeRuleChangedEvent event) {
+ metrics.invalidation("after_commit");
+ // 先移除跨 JVM 的规则事实，再清除本 JVM 所有历史 fingerprint 编译对象。
+ for (Long deviceId : event.getDeviceIds()) {
+ try {
+ invalidationService.invalidateRulesForDevice(deviceId);
+ } catch (RuntimeException cacheFailure) {
+ metrics.invalidation("failure");
+ log.error("规则提交后 raw cache 失效失败: ruleId={}, deviceId={}, operation={}",
+ event.getRuleId(), deviceId, event.getOperation(), cacheFailure);
+ }
+ }
+ try {
+ compiledRuleCache.invalidateRule(event.getRuleId());
+ } catch (RuntimeException cacheFailure) {
+ metrics.invalidation("failure");
+ // 数据库事实已提交；raw rules 的短 TTL 是最终一致性兜底。
+ log.error("规则提交后 compiled cache 失效失败: ruleId={}, operation={}",
+ event.getRuleId(), event.getOperation(), cacheFailure);
+ }
+ try {
+ invalidationService.invalidateRuleEntity(event.getRuleId());
+ } catch (RuntimeException cacheFailure) {
+ metrics.invalidation("failure");
+ log.error("规则提交后实体缓存失效失败: ruleId={}", event.getRuleId(), cacheFailure);
+ }
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleChangedEvent.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleChangedEvent.java
new file mode 100644
index 0000000..34072c8
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/NormalizeRuleChangedEvent.java
@@ -0,0 +1,23 @@
+package com.Modules.NormalData.cache;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** 事务内发布的低敏领域事件，不携带规则正文。 */
+public final class NormalizeRuleChangedEvent {
+ private final long ruleId;
+ private final List deviceIds;
+ private final String operation;
+
+ public NormalizeRuleChangedEvent(long ruleId, List deviceIds, String operation) {
+ this.ruleId = ruleId;
+ this.deviceIds = deviceIds == null ? Collections.emptyList()
+ : Collections.unmodifiableList(new ArrayList<>(deviceIds));
+ this.operation = operation;
+ }
+
+ public long getRuleId() { return ruleId; }
+ public List getDeviceIds() { return deviceIds; }
+ public String getOperation() { return operation; }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/RuleSourceSnapshot.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/RuleSourceSnapshot.java
new file mode 100644
index 0000000..ba12453
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/cache/RuleSourceSnapshot.java
@@ -0,0 +1,34 @@
+package com.Modules.NormalData.cache;
+
+import com.Modules.NormalData.compiled.RuleSource;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** 可安全跨 JVM 缓存的设备规则事实快照，不持有 JDBC/MyBatis 对象。 */
+public final class RuleSourceSnapshot implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final long deviceId;
+ private final List rules;
+ private final Instant loadedAt;
+
+ @SuppressWarnings("unused")
+ private RuleSourceSnapshot() {
+ this(0L, Collections.emptyList(), Instant.EPOCH);
+ }
+
+ public RuleSourceSnapshot(long deviceId, List rules, Instant loadedAt) {
+ this.deviceId = deviceId;
+ this.rules = new ArrayList<>(rules);
+ this.loadedAt = loadedAt;
+ }
+
+ public long getDeviceId() { return deviceId; }
+ public List getRules() { return Collections.unmodifiableList(rules); }
+ public Instant getLoadedAt() { return loadedAt; }
+ public boolean isEmpty() { return rules.isEmpty(); }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/BatchEnrichmentContext.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/BatchEnrichmentContext.java
new file mode 100644
index 0000000..bf0897e
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/BatchEnrichmentContext.java
@@ -0,0 +1,107 @@
+package com.Modules.NormalData.compiled;
+
+import com.common.entity.DeviceCollectTask;
+import com.common.entity.DeviceDevice;
+import com.common.service.DeviceCollectTaskService;
+import com.common.service.DeviceDeviceService;
+import com.Modules.NormalData.metrics.NormalizationRuleMetrics;
+
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+
+/**
+ * 批次内按 ID single-flight。FutureTask 允许缓存 null（通过 Optional）并把 loader 异常传播到 ACK 边界。
+ */
+public final class BatchEnrichmentContext {
+ private final boolean enabled;
+ private final DeviceCollectTaskService collectTaskService;
+ private final DeviceDeviceService deviceService;
+ private final NormalizationRuleMetrics metrics;
+ private final ConcurrentHashMap>> devices =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap>> collectTasks =
+ new ConcurrentHashMap<>();
+
+ public BatchEnrichmentContext(boolean enabled, DeviceCollectTaskService collectTaskService,
+ DeviceDeviceService deviceService) {
+ this(enabled, collectTaskService, deviceService, null);
+ }
+
+ public BatchEnrichmentContext(boolean enabled, DeviceCollectTaskService collectTaskService,
+ DeviceDeviceService deviceService, NormalizationRuleMetrics metrics) {
+ this.enabled = enabled;
+ this.collectTaskService = collectTaskService;
+ this.deviceService = deviceService;
+ this.metrics = metrics;
+ }
+
+ public DeviceSnapshot device(int id) {
+ if (!enabled) return loadDevice(id);
+ FutureTask> candidate = new FutureTask<>(() ->
+ Optional.ofNullable(loadDevice(id)));
+ FutureTask> task = devices.putIfAbsent(id, candidate);
+ if (task == null) {
+ task = candidate;
+ metric("device", "load");
+ } else {
+ metric("device", "hit");
+ }
+ return await(task, "device", id).orElse(null);
+ }
+
+ public CollectTaskSnapshot collectTask(int id) {
+ if (!enabled) return loadCollectTask(id);
+ FutureTask> candidate = new FutureTask<>(() ->
+ Optional.ofNullable(loadCollectTask(id)));
+ FutureTask> task = collectTasks.putIfAbsent(id, candidate);
+ if (task == null) {
+ task = candidate;
+ metric("collect_task", "load");
+ } else {
+ metric("collect_task", "hit");
+ }
+ return await(task, "collectTask", id).orElse(null);
+ }
+
+ private DeviceSnapshot loadDevice(int id) {
+ try {
+ DeviceDevice loaded = deviceService.getByIdSafely(id);
+ return DeviceSnapshot.from(loaded);
+ } catch (RuntimeException failure) {
+ metric("device", "failure");
+ throw failure;
+ }
+ }
+
+ private CollectTaskSnapshot loadCollectTask(int id) {
+ try {
+ DeviceCollectTask loaded = collectTaskService.getById(id);
+ return CollectTaskSnapshot.from(loaded);
+ } catch (RuntimeException failure) {
+ metric("collect_task", "failure");
+ throw failure;
+ }
+ }
+
+ private void metric(String entity, String outcome) {
+ if (metrics != null) metrics.enrichment(entity, outcome);
+ }
+
+ private static Optional await(FutureTask> task, String entity, int id) {
+ task.run();
+ try {
+ return task.get();
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(entity + " 批次补全被中断: " + id, interrupted);
+ } catch (ExecutionException failure) {
+ throw new IllegalStateException(entity + " 批次补全加载失败: " + id, failure.getCause());
+ }
+ }
+
+ public boolean isEnabled() { return enabled; }
+ public int deviceUniqueCount() { return devices.size(); }
+ public int collectTaskUniqueCount() { return collectTasks.size(); }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/BatchRuleSnapshot.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/BatchRuleSnapshot.java
new file mode 100644
index 0000000..7c48ff6
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/BatchRuleSnapshot.java
@@ -0,0 +1,30 @@
+package com.Modules.NormalData.compiled;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** 一个 Kafka poll 固定使用的原始规则快照。 */
+public final class BatchRuleSnapshot {
+ private final Map> byDevice;
+ private final long capturedAtMillis;
+
+ public BatchRuleSnapshot(Map> byDevice, long capturedAtMillis) {
+ Map> copy = new LinkedHashMap<>();
+ for (Map.Entry> entry : byDevice.entrySet()) {
+ copy.put(entry.getKey(), Collections.unmodifiableList(new ArrayList<>(entry.getValue())));
+ }
+ this.byDevice = Collections.unmodifiableMap(copy);
+ this.capturedAtMillis = capturedAtMillis;
+ }
+
+ public List rulesFor(long deviceId) {
+ List rules = byDevice.get(deviceId);
+ return rules == null ? Collections.emptyList() : rules;
+ }
+
+ public long getCapturedAtMillis() { return capturedAtMillis; }
+ public int getDeviceCount() { return byDevice.size(); }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CollectTaskSnapshot.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CollectTaskSnapshot.java
new file mode 100644
index 0000000..ac86c2e
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CollectTaskSnapshot.java
@@ -0,0 +1,21 @@
+package com.Modules.NormalData.compiled;
+
+import com.common.entity.DeviceCollectTask;
+
+/** 批次内不可变的采集任务补全数据。 */
+public final class CollectTaskSnapshot {
+ private final String deviceIp;
+ private final String taskName;
+
+ private CollectTaskSnapshot(String deviceIp, String taskName) {
+ this.deviceIp = deviceIp;
+ this.taskName = taskName;
+ }
+
+ public static CollectTaskSnapshot from(DeviceCollectTask task) {
+ return task == null ? null : new CollectTaskSnapshot(task.getDeviceIp(), task.getTaskName());
+ }
+
+ public String getDeviceIp() { return deviceIp; }
+ public String getTaskName() { return taskName; }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledCompletion.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledCompletion.java
new file mode 100644
index 0000000..dcde30e
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledCompletion.java
@@ -0,0 +1,62 @@
+package com.Modules.NormalData.compiled;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public final class CompiledCompletion {
+ private final String columnName;
+ private final Object value;
+ private final boolean orRelation;
+ private final List conditions;
+
+ public CompiledCompletion(String columnName, Object value, boolean orRelation,
+ List conditions) {
+ this.columnName = columnName;
+ this.value = value;
+ this.orRelation = orRelation;
+ this.conditions = Collections.unmodifiableList(new ArrayList<>(conditions));
+ }
+
+ public boolean apply(Map fields) {
+ if (fields.containsKey(columnName) && fields.get(columnName) != null) return false;
+ boolean matches;
+ if (conditions.isEmpty()) {
+ matches = true;
+ } else if (orRelation) {
+ matches = false;
+ for (CompiledPredicate condition : conditions) {
+ if (condition.evaluate(fields.get(condition.getColumnName()), true)) {
+ matches = true;
+ break;
+ }
+ }
+ } else {
+ matches = true;
+ for (CompiledPredicate condition : conditions) {
+ if (!condition.evaluate(fields.get(condition.getColumnName()), true)) {
+ matches = false;
+ break;
+ }
+ }
+ }
+ if (matches) fields.put(columnName, value);
+ return matches;
+ }
+
+ public String getColumnName() { return columnName; }
+ public Object getValue() { return value; }
+ public boolean isOrRelation() { return orRelation; }
+ public List getConditions() { return conditions; }
+
+ @Override
+ public String toString() {
+ return "CompiledCompletion{" +
+ "columnName='" + columnName + '\'' +
+ ", value=" + value +
+ ", orRelation=" + orRelation +
+ ", conditions=" + conditions +
+ '}';
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledCropper.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledCropper.java
new file mode 100644
index 0000000..015259d
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledCropper.java
@@ -0,0 +1,42 @@
+package com.Modules.NormalData.compiled;
+
+import com.common.entity.RuleContent.Cropper_paramsType;
+
+/** 不可变的正文截取配置。 */
+public final class CompiledCropper {
+ private final String headKey;
+ private final int headOffset;
+ private final String tailKey;
+ private final int tailOffset;
+
+ public CompiledCropper(String headKey, int headOffset, String tailKey, int tailOffset) {
+ this.headKey = headKey;
+ this.headOffset = headOffset;
+ this.tailKey = tailKey;
+ this.tailOffset = tailOffset;
+ }
+
+ public Cropper_paramsType toLegacyParams() {
+ Cropper_paramsType params = new Cropper_paramsType();
+ params.sethead_key(headKey);
+ params.sethead_offset(headOffset);
+ params.settail_key(tailKey);
+ params.settail_offset(tailOffset);
+ return params;
+ }
+
+ public String getHeadKey() { return headKey; }
+ public int getHeadOffset() { return headOffset; }
+ public String getTailKey() { return tailKey; }
+ public int getTailOffset() { return tailOffset; }
+
+ @Override
+ public String toString() {
+ return "CompiledCropper{" +
+ "headKey='" + headKey + '\'' +
+ ", headOffset=" + headOffset +
+ ", tailKey='" + tailKey + '\'' +
+ ", tailOffset=" + tailOffset +
+ '}';
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledFilterGroup.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledFilterGroup.java
new file mode 100644
index 0000000..41f8398
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledFilterGroup.java
@@ -0,0 +1,41 @@
+package com.Modules.NormalData.compiled;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public final class CompiledFilterGroup {
+ private final boolean orRelation;
+ private final List predicates;
+
+ public CompiledFilterGroup(boolean orRelation, List predicates) {
+ this.orRelation = orRelation;
+ this.predicates = Collections.unmodifiableList(new ArrayList<>(predicates));
+ }
+
+ public boolean matches(Map fields) {
+ if (predicates.isEmpty()) return false;
+ if (orRelation) {
+ for (CompiledPredicate predicate : predicates) {
+ if (predicate.evaluate(fields.get(predicate.getColumnName()), false)) return true;
+ }
+ return false;
+ }
+ for (CompiledPredicate predicate : predicates) {
+ if (!predicate.evaluate(fields.get(predicate.getColumnName()), false)) return false;
+ }
+ return true;
+ }
+
+ public boolean isOrRelation() { return orRelation; }
+ public List getPredicates() { return predicates; }
+
+ @Override
+ public String toString() {
+ return "CompiledFilterGroup{" +
+ "orRelation=" + orRelation +
+ ", predicates=" + predicates +
+ '}';
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledMapper.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledMapper.java
new file mode 100644
index 0000000..db78ef4
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledMapper.java
@@ -0,0 +1,74 @@
+package com.Modules.NormalData.compiled;
+
+import com.common.util.TimeConversionUtils;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** 单个字段映射的不可变执行计划。 */
+public final class CompiledMapper {
+ public enum Action { EQUAL, MAPPING, TIME, LEGACY_NULL }
+
+ private final String originField;
+ private final String destinationField;
+ private final Action action;
+ private final Map mapping;
+ private final String timezone;
+
+ public CompiledMapper(String originField, String destinationField, Action action,
+ Map mapping, String timezone) {
+ this.originField = originField;
+ this.destinationField = destinationField;
+ this.action = action;
+ this.mapping = mapping == null ? Collections.emptyMap()
+ : Collections.unmodifiableMap(new LinkedHashMap<>(mapping));
+ this.timezone = timezone;
+ }
+
+ public Object map(Object sourceValue) {
+ switch (action) {
+ case EQUAL:
+ return sanitize(sourceValue);
+ case MAPPING:
+ // 保持旧执行器语义：源值为 null 时规则执行失败，而不是静默映射成字符串 "null"。
+ return sanitize(mapping.get(sourceValue.toString()));
+ case TIME:
+ try {
+ return TimeConversionUtils.convertToMillis(sourceValue.toString(), timezone);
+ } catch (Exception conversionFailure) {
+ // 旧执行器在时间转换失败时仍写入目标字段，只是字段值为 null。
+ return null;
+ }
+ case LEGACY_NULL:
+ // 历史规则 action.type 为空时，旧执行器仍生成目标字段，但字段值保持 null。
+ return null;
+ default:
+ throw new IllegalStateException("不支持的映射动作: " + action);
+ }
+ }
+
+ private static Object sanitize(Object value) {
+ if (value instanceof String && ((String) value).contains("\u0000")) {
+ return ((String) value).replace("\u0000", "");
+ }
+ return value;
+ }
+
+ public String getOriginField() { return originField; }
+ public String getDestinationField() { return destinationField; }
+ public Action getAction() { return action; }
+ public Map getMapping() { return mapping; }
+ public String getTimezone() { return timezone; }
+
+ @Override
+ public String toString() {
+ return "CompiledMapper{" +
+ "originField='" + originField + '\'' +
+ ", destinationField='" + destinationField + '\'' +
+ ", action=" + action +
+ ", mapping=" + mapping +
+ ", timezone='" + timezone + '\'' +
+ '}';
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledPredicate.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledPredicate.java
new file mode 100644
index 0000000..c946e29
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledPredicate.java
@@ -0,0 +1,71 @@
+package com.Modules.NormalData.compiled;
+
+import org.springframework.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.regex.Pattern;
+
+/** 过滤和补全共用的预编译条件。 */
+public final class CompiledPredicate {
+ public enum Operation { EQUAL, NOT_EQUAL, IN, ZERO_LEN, CONTAIN, REGEXP }
+
+ private final String columnName;
+ private final Operation operation;
+ private final Object value;
+ private final List