candidate, RuleMatchProfile profile) {
+ this.candidate = candidate;
+ this.profile = profile;
+ }
+ }
+
+ private static final class AutomaticScore {
+ private static final AutomaticScore ZERO = new AutomaticScore(0.0d, 0.0d, 0.0d, 0);
+ private final double ownershipScore;
+ private final double matchedWeight;
+ private final double coverage;
+ private final int matchedCount;
+
+ private AutomaticScore(double ownershipScore, double matchedWeight,
+ double coverage, int matchedCount) {
+ this.ownershipScore = ownershipScore;
+ this.matchedWeight = matchedWeight;
+ this.coverage = coverage;
+ this.matchedCount = matchedCount;
+ }
+ }
+}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/metrics/NormalizationRuleMetrics.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/metrics/NormalizationRuleMetrics.java
index a6084c7..09797b6 100644
--- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/metrics/NormalizationRuleMetrics.java
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/metrics/NormalizationRuleMetrics.java
@@ -5,6 +5,7 @@ import org.springframework.stereotype.Component;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.DoubleAdder;
import java.util.concurrent.atomic.LongAdder;
/** 固定维度的进程内规则指标;不以 rule/device/event 作为长期标签。 */
@@ -18,7 +19,15 @@ public class NormalizationRuleMetrics {
"collect_task_hit", "collect_task_load", "collect_task_failure");
private final Counters invalidation = new Counters("local", "remote", "after_commit", "failure");
private final Counters messageParse = new Counters("hit", "miss");
+ private final Counters ruleSelection = new Counters("selected", "explicit_selected",
+ "automatic_selected", "ambiguous", "equivalent_tie", "conflicting_tie",
+ "no_match", "invalid_match_config", "shadow_selected", "shadow_no_match");
private final LongAdder ruleJsonParse = new LongAdder();
+ private final LongAdder selectionObservationCount = new LongAdder();
+ private final LongAdder selectionCandidateTotal = new LongAdder();
+ private final AtomicLong selectionCandidateMax = new AtomicLong();
+ private final DoubleAdder selectionMatchRateTotal = new DoubleAdder();
+ private final DoubleAdder selectionScoreMarginTotal = new DoubleAdder();
private final LongAdder compileDurationTotalNanos = new LongAdder();
private final LongAdder compileDurationCount = new LongAdder();
private final AtomicLong compileDurationMaxNanos = new AtomicLong();
@@ -39,6 +48,15 @@ public class NormalizationRuleMetrics {
}
public void invalidation(String outcome) { invalidation.increment(outcome); }
public void messageParse(String outcome) { messageParse.increment(outcome); }
+ public void ruleSelection(String outcome) { ruleSelection.increment(outcome); }
+ public void ruleSelectionObservation(int candidateCount, double matchRate, double scoreMargin) {
+ int candidates = Math.max(0, candidateCount);
+ selectionObservationCount.increment();
+ selectionCandidateTotal.add(candidates);
+ updateMax(selectionCandidateMax, candidates);
+ selectionMatchRateTotal.add(Math.max(0.0d, Math.min(1.0d, matchRate)));
+ selectionScoreMarginTotal.add(Math.max(0.0d, scoreMargin));
+ }
public void ruleJsonParse() { ruleJsonParse.increment(); }
public void compiledCacheGauge(long size, long evictions) {
compiledCacheSize.set(Math.max(0L, size));
@@ -61,6 +79,15 @@ public class NormalizationRuleMetrics {
result.put("enrichment_cache_get_total", enrichment.snapshot());
result.put("rule_invalidation_total", invalidation.snapshot());
result.put("message_parse_cache_get_total", messageParse.snapshot());
+ result.put("rule_selection_total", ruleSelection.snapshot());
+ result.put("rule_selection_observation_count", selectionObservationCount.sum());
+ result.put("rule_selection_candidate_count_total", selectionCandidateTotal.sum());
+ result.put("rule_selection_candidate_count_max", selectionCandidateMax.get());
+ long observations = selectionObservationCount.sum();
+ result.put("rule_selection_match_rate_avg", observations == 0L ? 0.0d
+ : selectionMatchRateTotal.sum() / observations);
+ result.put("rule_selection_score_margin_avg", observations == 0L ? 0.0d
+ : selectionScoreMarginTotal.sum() / observations);
result.put("rule_json_parse_total", ruleJsonParse.sum());
return result;
}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java
index 6cc7a12..deacf5d 100644
--- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java
@@ -32,6 +32,7 @@ public class DmStandardizedBatchExecutor {
private final StatsChangeCaptureService statsChangeCaptureService;
private final DirectStatsProjectionWriter directStatsProjectionWriter;
private final DmDirectTransactionMetrics transactionMetrics;
+ private final LateAlarmBucketWriter lateAlarmBucketWriter;
@org.springframework.beans.factory.annotation.Value("${stats.direct.transaction-log-enabled:false}")
private boolean transactionLogEnabled;
@@ -46,7 +47,8 @@ public class DmStandardizedBatchExecutor {
PlatformTransactionManager transactionManager,
StatsChangeCaptureService statsChangeCaptureService,
DirectStatsProjectionWriter directStatsProjectionWriter,
- DmDirectTransactionMetrics transactionMetrics) {
+ DmDirectTransactionMetrics transactionMetrics,
+ LateAlarmBucketWriter lateAlarmBucketWriter) {
// SqlSessionTemplate 本身线程安全。ExecutorType.BATCH 会把相同 SQL 的参数暂存在当前事务会话中,
// 直到 flushStatements()/commit 时才真正交给 JDBC 驱动批量执行。
this.batchSqlSession = new SqlSessionTemplate(sqlSessionFactory, ExecutorType.BATCH);
@@ -54,6 +56,7 @@ public class DmStandardizedBatchExecutor {
this.statsChangeCaptureService = statsChangeCaptureService;
this.directStatsProjectionWriter = directStatsProjectionWriter;
this.transactionMetrics = transactionMetrics;
+ this.lateAlarmBucketWriter = lateAlarmBucketWriter;
}
/**
@@ -126,6 +129,12 @@ public class DmStandardizedBatchExecutor {
batchSqlSession.flushStatements();
timing.factFlushMs = elapsedMs(flushStarted);
+ // 与标准化告警事实共用同一事务写入事件时间桶。事实成功而通知失败时必须整体回滚,
+ // 否则迟到数据可能永久失去重放机会。
+ if (alarmTable) {
+ lateAlarmBucketWriter.record(rows);
+ }
+
// DIRECT/SHADOW 模式在当前事务内按设备、规则聚合写统计;LEGACY 模式会直接返回。
// 任一统计 SQL 失败都会回滚本事务的事实行,禁止产生无法定位的统计缺口。
timing.statsResult = directStatsProjectionWriter.writeStandardizedRows(rows);
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/LateAlarmBucketWriter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/LateAlarmBucketWriter.java
new file mode 100644
index 0000000..8d7ddf6
--- /dev/null
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/LateAlarmBucketWriter.java
@@ -0,0 +1,162 @@
+package com.common.service;
+
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.dao.CannotAcquireLockException;
+import org.springframework.dao.DeadlockLoserDataAccessException;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * 在标准化告警事实事务内登记事件时间桶。
+ *
+ * 该通知不是“告警一定迟到”的结论。通知与事实在同一事务提交;下游能领取通知时,事实必然
+ * 已经可见,再根据领取时已经完成的规则窗口决定是否重放。这里刻意不使用事实的
+ * {@code created_at} 判断提交先后,因为 Kafka 积压时接收时间可能早于实际入库时间。
+ */
+@Component
+public class LateAlarmBucketWriter {
+
+ private static final String UPSERT_SQL =
+ "MERGE INTO analysis_late_alarm_bucket t "
+ + "USING (SELECT ? device_id, ? bucket_start FROM dual) s "
+ + "ON (t.device_id=s.device_id AND t.bucket_start=s.bucket_start) "
+ + "WHEN MATCHED THEN UPDATE SET "
+ + "t.requested_generation=t.requested_generation+1, "
+ + "t.last_observed_at=SYSDATE, "
+ + "t.status=CASE WHEN t.status='RUNNING' THEN 'RUNNING' ELSE 'PENDING' END, "
+ + "t.updated_at=SYSDATE "
+ + "WHEN NOT MATCHED THEN INSERT "
+ + "(id,device_id,bucket_start,bucket_end,last_observed_at,requested_generation,"
+ + "claimed_generation,processed_generation,status,attempt_count,fencing_token,created_at,updated_at) "
+ + "VALUES (?,?,?,?,SYSDATE,1,0,0,'PENDING',0,0,SYSDATE,SYSDATE)";
+
+ private final JdbcTemplate jdbcTemplate;
+
+ @Value("${analysis.late-replay.outbox-enabled:true}")
+ private boolean enabled;
+
+ public LateAlarmBucketWriter(JdbcTemplate jdbcTemplate) {
+ this.jdbcTemplate = jdbcTemplate;
+ }
+
+ /**
+ * 每个设备每分钟只写一条通知。调用方必须已经开启标准化事实事务。
+ *
+ * @param rows 已确认写入 {@code syslog_normal_alarm} 的同构小批次
+ */
+ public void record(List