From 2def44553785a1ec380ba3db5ee5841a534ffa60 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Fri, 11 Sep 2026 18:24:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(haobang-security-rule)=E4=BF=AE=E6=94=B9jso?= =?UTF-8?q?n=E9=94=99=E8=AF=AF=E5=A4=84=E7=90=86=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../syslog-client/logs/syslog-client.log | 9 -- .../NormalData/LogNormalProcessor.java | 89 ++++++++++++++++++- .../com/common/util/NestedJsonParserUtil.java | 17 +--- .../LogNormalProcessorRuleSelectionTest.java | 32 +++++++ 4 files changed, 122 insertions(+), 25 deletions(-) diff --git a/haobang-security-dm/syslog-client/logs/syslog-client.log b/haobang-security-dm/syslog-client/logs/syslog-client.log index a0b742a..c2cfe4e 100644 --- a/haobang-security-dm/syslog-client/logs/syslog-client.log +++ b/haobang-security-dm/syslog-client/logs/syslog-client.log @@ -7,12 +7,3 @@ 2026-09-11 16:29:59.280 [main] INFO c.h.s.SysjavacollectApplicationTests - Starting SysjavacollectApplicationTests using Java 1.8.0_351 on wangshaohongdeMacBook-Pro.local with PID 63617 (started by wangshaohong in /Users/wangshaohong/WorkSpace/fufeng/ai-security-xdr/haobang-security-dm/syslog-client) 2026-09-11 16:29:59.286 [main] INFO c.h.s.SysjavacollectApplicationTests - No active profile set, falling back to 1 default profile: "default" 2026-09-11 16:30:00.361 [main] INFO c.h.s.SysjavacollectApplicationTests - Started SysjavacollectApplicationTests in 1.499 seconds (JVM running for 3.309) -2026-09-11 16:34:00.334 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Neither @ContextConfiguration nor @ContextHierarchy found for test class [com.haobang.syslog.SysjavacollectApplicationTests], using SpringBootContextLoader -2026-09-11 16:34:00.338 [main] INFO o.s.t.c.s.AbstractContextLoader - Could not detect default resource locations for test class [com.haobang.syslog.SysjavacollectApplicationTests]: no resource found for suffixes {-context.xml, Context.groovy}. -2026-09-11 16:34:00.339 [main] INFO o.s.t.c.s.AnnotationConfigContextLoaderUtils - Could not detect default configuration classes for test class [com.haobang.syslog.SysjavacollectApplicationTests]: SysjavacollectApplicationTests does not declare any static, non-private, non-final, nested classes annotated with @Configuration. -2026-09-11 16:34:00.401 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Found @SpringBootConfiguration com.haobang.syslog.syslogClientApplication for test class com.haobang.syslog.SysjavacollectApplicationTests -2026-09-11 16:34:00.490 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Loaded default TestExecutionListener class names from location [META-INF/spring.factories]: [org.springframework.boot.test.mock.mockito.MockitoTestExecutionListener, org.springframework.boot.test.mock.mockito.ResetMocksTestExecutionListener, org.springframework.boot.test.autoconfigure.restdocs.RestDocsTestExecutionListener, org.springframework.boot.test.autoconfigure.web.client.MockRestServiceServerResetTestExecutionListener, org.springframework.boot.test.autoconfigure.web.servlet.MockMvcPrintOnlyOnFailureTestExecutionListener, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverTestExecutionListener, org.springframework.boot.test.autoconfigure.webservices.client.MockWebServiceServerTestExecutionListener, org.springframework.test.context.web.ServletTestExecutionListener, org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener, org.springframework.test.context.event.ApplicationEventsTestExecutionListener, org.springframework.test.context.support.DependencyInjectionTestExecutionListener, org.springframework.test.context.support.DirtiesContextTestExecutionListener, org.springframework.test.context.transaction.TransactionalTestExecutionListener, org.springframework.test.context.jdbc.SqlScriptsTestExecutionListener, org.springframework.test.context.event.EventPublishingTestExecutionListener] -2026-09-11 16:34:00.496 [main] INFO o.s.b.t.c.SpringBootTestContextBootstrapper - Using TestExecutionListeners: [org.springframework.test.context.support.DirtiesContextBeforeModesTestExecutionListener@68746f22, org.springframework.test.context.event.ApplicationEventsTestExecutionListener@2f01783a, org.springframework.boot.test.mock.mockito.MockitoTestExecutionListener@68878f6d, org.springframework.boot.test.autoconfigure.SpringBootDependencyInjectionTestExecutionListener@41488b16, org.springframework.test.context.support.DirtiesContextTestExecutionListener@a8ef162, org.springframework.test.context.event.EventPublishingTestExecutionListener@2eea88a1, org.springframework.boot.test.mock.mockito.ResetMocksTestExecutionListener@7ac296f6, org.springframework.boot.test.autoconfigure.restdocs.RestDocsTestExecutionListener@7fd7a283, org.springframework.boot.test.autoconfigure.web.client.MockRestServiceServerResetTestExecutionListener@22f59fa, org.springframework.boot.test.autoconfigure.web.servlet.MockMvcPrintOnlyOnFailureTestExecutionListener@78fa769e, org.springframework.boot.test.autoconfigure.web.servlet.WebDriverTestExecutionListener@16612a51, org.springframework.boot.test.autoconfigure.webservices.client.MockWebServiceServerTestExecutionListener@54e041a4] -2026-09-11 16:34:00.647 [main] INFO c.h.s.SysjavacollectApplicationTests - Starting SysjavacollectApplicationTests using Java 1.8.0_392 on wangshaohongdeMacBook-Pro.local with PID 65327 (started by wangshaohong in /Users/wangshaohong/WorkSpace/fufeng/ai-security-xdr-orig-apipull/haobang-security-dm/syslog-client) -2026-09-11 16:34:00.647 [main] INFO c.h.s.SysjavacollectApplicationTests - No active profile set, falling back to 1 default profile: "default" -2026-09-11 16:34:00.957 [main] INFO c.h.s.SysjavacollectApplicationTests - Started SysjavacollectApplicationTests in 0.436 seconds (JVM running for 1.012) 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 7355ad6..22cd616 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 @@ -118,6 +118,8 @@ public class LogNormalProcessor { private final NormalizationRuleMetrics globalMetrics; private final RuleSelectionEngine ruleSelectionEngine; private final Map shadowLogWindows = new ConcurrentHashMap<>(); + private final AtomicLong invalidJsonLastWarnAt = new AtomicLong(); + private final AtomicLong invalidJsonSinceLastWarn = new AtomicLong(); /** * 请求级处理上下文 —— 每个 process() 调用创建新实例，彻底解决多线程竞态条件 @@ -148,6 +150,8 @@ public class LogNormalProcessor { MessageParseContext messageParseContext; boolean shadowMode; boolean deferNonNormalPersistence; + boolean invalidJson; + RuleSource invalidJsonRule; final List> normalizedRows = new ArrayList<>(); final List deferredNonNormalMessages = new ArrayList<>(); } @@ -407,6 +411,9 @@ public class LogNormalProcessor { } if (ctx.dmNormalizeRuleList.size() > 1 && selectionConfig.isShadowEnabled()) { shadowSelect(ctx); + if (ctx.invalidJson) { + return invalidJsonResult(ctx, deviceID); + } } if (optimizationProperties.getCompiledRule().isEnabled()) { return executeCompiled(ctx, deviceID); @@ -426,6 +433,9 @@ public class LogNormalProcessor { for (RuleSource source : ctx.dmNormalizeRuleList) { try { executeLegacyRule(ctx, deviceId, source); + if (ctx.invalidJson) { + return invalidJsonResult(ctx, deviceId); + } } catch (Exception failure) { logger.error("泛化规则处理失败, eventId={}, ruleId={}", ctx.strSyslogUUID, source.getId(), failure); @@ -440,6 +450,9 @@ public class LogNormalProcessor { */ private NormalizationResult executeSelected(ProcessContext ctx, long deviceId) { List> candidates = collectCandidates(ctx); + if (ctx.invalidJson) { + return invalidJsonResult(ctx, deviceId); + } RuleSelectionEngine.Decision decision = ruleSelectionEngine.select(candidates, ctx.strLogMsg, false); if (!decision.isSelected()) { @@ -493,6 +506,7 @@ public class LogNormalProcessor { private void shadowSelect(ProcessContext ctx) { try { List> candidates = collectCandidates(ctx); + if (ctx.invalidJson) return; ruleSelectionEngine.select(candidates, ctx.strLogMsg, true); } catch (Exception failure) { // 影子链路不能影响生产结果。 @@ -515,7 +529,14 @@ public class LogNormalProcessor { ParsedMessage parsed = ctx.messageParseContext.parse( compiled, ctx.strLogMsg, ctx.executionMetrics); ctx.executionMetrics.addMessageParseNanos(System.nanoTime() - parseStarted); - if (parsed == null || parsed.getFields().isEmpty()) continue; + if (parsed == null) { + if (isJsonRule(source)) { + markInvalidJson(ctx, source); + return candidates; + } + continue; + } + if (parsed.getFields().isEmpty()) continue; plan = ExecutionPlan.compiled(source, compiled, parsed); } catch (CompiledRuleCompiler.RuleCompilationException incompatibleRule) { plan = legacyCandidate(ctx, source); @@ -543,8 +564,11 @@ public class LogNormalProcessor { LinkedHashMap parsed = getMessageToMap( source.asLegacyMap(), ctx.strLogMsg, source.getDataType()); ctx.executionMetrics.addMessageParseNanos(System.nanoTime() - parseStarted); - return parsed == null || parsed.isEmpty() ? null - : ExecutionPlan.legacy(source, parsed); + if (parsed == null) { + if (isJsonRule(source)) markInvalidJson(ctx, source); + return null; + } + return parsed.isEmpty() ? null : ExecutionPlan.legacy(source, parsed); } private BusinessResult evaluateBusiness(ProcessContext ctx, ExecutionPlan plan) { @@ -656,6 +680,9 @@ public class LogNormalProcessor { // 首次失败原因由 CompiledRuleCache 按 fingerprint 限频记录；此处只回退该规则。 try { executeLegacyRule(ctx, deviceId, source); + if (ctx.invalidJson) { + return invalidJsonResult(ctx, deviceId); + } } catch (Exception legacyFailure) { logger.error("编译失败规则回退旧执行器后仍失败, eventId={}, ruleId={}", ctx.strSyslogUUID, source.getId(), legacyFailure); @@ -665,6 +692,10 @@ public class LogNormalProcessor { } try { ParsedMessage parsed = ctx.messageParseContext.parse(rule, ctx.strLogMsg, ctx.executionMetrics); + if (parsed == null && isJsonRule(source)) { + markInvalidJson(ctx, source); + return invalidJsonResult(ctx, deviceId); + } if (parsed == null || parsed.getFields().isEmpty()) { rejectParsedMessage(ctx, deviceId, source); continue; @@ -699,6 +730,10 @@ public class LogNormalProcessor { globalMetrics.ruleJsonParse(); ctx.OrginalColumnMap = getMessageToMap(rule, ctx.strLogMsg, source.getDataType()); ctx.executionMetrics.addMessageParseNanos(System.nanoTime() - parseStarted); + if (ctx.OrginalColumnMap == null && isJsonRule(source)) { + markInvalidJson(ctx, source); + return; + } if (ctx.OrginalColumnMap == null || ctx.OrginalColumnMap.isEmpty()) { rejectParsedMessage(ctx, deviceId, source); return; @@ -740,6 +775,53 @@ public class LogNormalProcessor { ctx.rejectedCount++; } + /** + * JSON 语法失败是当前消息可确定的业务终态：只产生一条非标记录， + * 不再尝试后续规则，也不抛异常进入 Retry/DLT。 + */ + private NormalizationResult invalidJsonResult(ProcessContext ctx, long deviceId) { + RuleSource source = ctx.invalidJsonRule; + ctx.syslogNonNormalMessage.setReason("JSON解析异常"); + ctx.syslogNonNormalMessage.setReasonDetail("发现非法JSON，已停止后续标准化规则处理, ruleId=" + + (source == null ? "unknown" : source.getId()) + ", ruleName=" + + (source == null ? "unknown" : source.getDisplayName())); + if (!ctx.shadowMode) { + if (!ctx.isSaveNonNormal) { + SaveNonNormalMessage(ctx, deviceId, DateTime.now()); + } + logInvalidJsonRateLimited(ctx, deviceId, source); + } + return NormalizationResult.filtered(ctx.deferredNonNormalMessages); + } + + private void markInvalidJson(ProcessContext ctx, RuleSource source) { + if (!ctx.invalidJson) { + ctx.invalidJson = true; + ctx.invalidJsonRule = source; + } + } + + private static boolean isJsonRule(RuleSource source) { + return source != null && "json".equalsIgnoreCase(source.getDataType()); + } + + /** 现场坏 JSON 可能高频出现，WARN 每分钟最多一条，避免日志 I/O 反向拖慢消费。 */ + private void logInvalidJsonRateLimited(ProcessContext ctx, long deviceId, RuleSource source) { + invalidJsonSinceLastWarn.incrementAndGet(); + long now = System.currentTimeMillis(); + long previous = invalidJsonLastWarnAt.get(); + if (now - previous < TimeUnit.MINUTES.toMillis(1) + || !invalidJsonLastWarnAt.compareAndSet(previous, now)) { + return; + } + long count = invalidJsonSinceLastWarn.getAndSet(0L); + logger.warn("JSON解析失败已直接转非标: countSinceLastLog={}, eventId={}, deviceId={}, " + + "ruleId={}, messageLength={}", + count, ctx.strSyslogUUID, deviceId, + source == null ? null : source.getId(), + ctx.strLogMsg == null ? 0 : ctx.strLogMsg.length()); + } + private NormalizationResult finish(ProcessContext ctx) { if (ctx.savedCount > 0) { return NormalizationResult.success(ctx.normalizedRows, ctx.deferredNonNormalMessages); @@ -916,6 +998,7 @@ public class LogNormalProcessor { if (dataType.equals("json")) { String strJson = logNormalData.ParserMessageJsonType(MsgContent, cropperParams); LinkedHashMap result1 = NestedJsonParserUtil.safeParseJson(strJson); + if (result1 == null) return null; LinkedHashMap flattened = NestedJsonUtils.flattenNestedJson(result1); return flattened; } else if (dataType.equals("kv")) { diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/NestedJsonParserUtil.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/NestedJsonParserUtil.java index db47e69..5314518 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/NestedJsonParserUtil.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/NestedJsonParserUtil.java @@ -72,19 +72,10 @@ public class NestedJsonParserUtil { return postProcessParsedData(result, maxDepth); } catch (JsonProcessingException e) { - logger.warn("JSON解析失败，长度={}, 原因={}", jsonStr.length(), e.getMessage()); - - // 5. 尝试修复并重新解析 - try { - String fixedJson = tryFixNestedJson(jsonStr, maxDepth); - if (!fixedJson.equals(jsonStr)) { - TypeReference> typeRef = new TypeReference>() {}; - Map result = objectMapper.readValue(fixedJson, typeRef); - return postProcessParsedData(result, maxDepth); - } - } catch (Exception ex) { - logger.debug("JSON修复后解析仍然失败: {}", ex.getMessage()); - } + // 标准化主链路不再对坏 JSON 做猜测修复：失败后由上层直接转非标。 + // getMessage() 会附带 Source 原文片段，大量异常报文会放大 Docker 日志 I/O； + // 仅在 DEBUG 保留不含原文的 Jackson 原始错误。 + logger.debug("JSON解析失败，长度={}, 原因={}", jsonStr.length(), e.getOriginalMessage()); return null; //return Collections.emptyMap(); diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java index 560917c..0233f33 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java @@ -105,6 +105,38 @@ class LogNormalProcessorRuleSelectionTest { verify(fixture.nonNormalService, never()).saveMessage(any()); } + @Test + void invalidJsonStopsAtFirstJsonRuleAndGoesDirectlyToDeferredNonNormal() { + RuleSource invalidJson = jsonRule(10L, "json-first", "", equalMapping(), ""); + RuleSource regexWouldMatch = regexRule(20L, "regex-second", "value=(\\S+)"); + Fixture fixture = fixture(Arrays.asList(invalidJson, regexWouldMatch), config(true)); + + LogNormalProcessor.NormalizationResult result = fixture.prepareActive( + "value=ok {\"broken\":\"unterminated"); + + assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome()); + assertTrue(result.getRows().isEmpty()); + assertEquals(1, result.getNonNormalMessages().size()); + assertEquals("JSON解析异常", result.getNonNormalMessages().get(0).getReason()); + assertTrue(result.getNonNormalMessages().get(0).getReasonDetail().contains("ruleId=10")); + verify(fixture.nonNormalService, never()).saveMessage(any()); + } + + @Test + void legacyInvalidJsonIsNonNormalTerminalInsteadOfException() { + Fixture fixture = fixture(Arrays.asList( + jsonRule(10L, "legacy-json", "", equalMapping(), "")), config(false)); + + LogNormalProcessor.NormalizationResult result = fixture.prepareActive( + "{\"broken\":\"unterminated"); + + assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome()); + assertTrue(result.getRows().isEmpty()); + assertEquals(1, result.getNonNormalMessages().size()); + assertEquals("JSON解析异常", result.getNonNormalMessages().get(0).getReason()); + verify(fixture.nonNormalService, never()).saveMessage(any()); + } + @Test void nonNormalSaveFailureRemainsRetryable() { Fixture fixture = fixture(Arrays.asList(