nanChen/ai-security-xdr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(haobang-security-rule)修改json错误处理方式

Browse Source
This commit is contained in:
王绍宏
2026-09-11 18:24:32 +08:00
parent 77580ea0bf
commit 2def445537
4 changed files with 122 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java
+86 -3
View File
@@ -118,6 +118,8 @@ public class LogNormalProcessor {
private final NormalizationRuleMetrics globalMetrics;
private final RuleSelectionEngine ruleSelectionEngine;
private final Map<String, AtomicLong> 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<Map<String, Object>> normalizedRows = new ArrayList<>();
final List<SyslogNonNormalMessage> 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<RuleSelectionEngine.Candidate<ExecutionPlan>> candidates = collectCandidates(ctx);
if (ctx.invalidJson) {
return invalidJsonResult(ctx, deviceId);
}
RuleSelectionEngine.Decision<ExecutionPlan> decision =
ruleSelectionEngine.select(candidates, ctx.strLogMsg, false);
if (!decision.isSelected()) {
@@ -493,6 +506,7 @@ public class LogNormalProcessor {
private void shadowSelect(ProcessContext ctx) {
try {
List<RuleSelectionEngine.Candidate<ExecutionPlan>> 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<String, Object> 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<String, Object> result1 = NestedJsonParserUtil.safeParseJson(strJson);
if (result1 == null) return null;
LinkedHashMap<String, Object> flattened = NestedJsonUtils.flattenNestedJson(result1);
return flattened;
} else if (dataType.equals("kv")) {
haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/NestedJsonParserUtil.java
+4 -13
View File
@@ -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<Map<String, Object>> typeRef = new TypeReference<Map<String, Object>>() {};
Map<String, Object> 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();