From c1472b9a9570a2b62db9dc007da50e1852c1f082 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 20:42:22 +0800 Subject: [PATCH] =?UTF-8?q?fix(syslog-consumer-rule):=20=E8=A7=A3=E8=80=A6?= =?UTF-8?q?=E7=BB=9F=E8=AE=A1=E7=89=A9=E5=8C=96=E5=B9=B6=E8=A7=A3=E5=86=B3?= =?UTF-8?q?=E5=A4=9A=E5=AE=9E=E4=BE=8B=E8=BE=BE=E6=A2=A6=E6=AD=BB=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将设备、规则和采集任务业务统计改为专用 Kafka Topic 异步物化 - 使用维度 ID 作为 Kafka key，保证同一统计维度串行更新 - 物化过程读取小型统计表精确值，确保重复消费不会重复计数 - 增加提交后发布、失败重试及定时小表对账补偿 - 固定日统计聚合键写入顺序，完善达梦死锁识别和事务重试 - JSON 解析失败直接转为非标日志，避免进入 DLT 重试链路 - 非标日志使用独立异步队列，不再阻塞正常消息 Kafka ACK - Kafka 升级脚本新增 STATS.MATERIALIZE Topic - 增加统计物化状态指标、配置示例及单元测试 --- .../Modules/NormalData/SysLogProcessor.java | 39 ++- .../NormalData/compiled/CompiledRule.java | 15 +- .../service/DirectStatsProjectionWriter.java | 80 +++++- .../service/IngressStatsProjectionWriter.java | 56 ++-- .../common/service/LateAlarmBucketWriter.java | 18 +- .../service/StatsMaterializationCommand.java | 63 +++++ .../service/StatsMaterializationConsumer.java | 255 ++++++++++++++++++ .../StatsMaterializationPublisher.java | 128 +++++++++ .../StatsMaterializationReconciler.java | 154 +++++++++++ .../com/config/KafkaReliabilityConfig.java | 16 ++ .../controllers/DirectStatsController.java | 17 +- .../src/main/resources/application.properties | 20 +- .../LogNormalProcessorRuleSelectionTest.java | 17 ++ .../SysLogProcessorParallelTest.java | 28 ++ .../NormalData/SyslogKafkaTopologyTest.java | 17 ++ .../DirectStatsProjectionWriterTest.java | 48 +++- .../StatsMaterializationCommandTest.java | 28 ++ .../StatsMaterializationConsumerTest.java | 59 ++++ .../StatsMaterializationPublisherTest.java | 41 +++ .../StatsMaterializationReconcilerTest.java | 43 +++ .../v1/kafka/kafka-cluster-upgrade.sh | 6 +- .../upgrade-doc/v1/kafka/kafka-upgrade.sh | 6 +- haobang-security-dm/upgrade-doc/v2/README.md | 13 + .../config/syslog-consumer-rule.env.example | 11 + 24 files changed, 1123 insertions(+), 55 deletions(-) create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationCommand.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationConsumer.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationPublisher.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationReconciler.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationCommandTest.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationConsumerTest.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationPublisherTest.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationReconcilerTest.java 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 f60c1c3..483239f 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 @@ -9,6 +9,7 @@ import com.config.AppProperties; import com.config.KafkaConsumerProperties; import com.common.service.DmStandardizedBatchWriter; import com.common.service.IngressStatsProjectionWriter; +import com.common.service.DirectStatsProjectionWriter; import com.common.service.SyslogNonNormalMessageService; import com.common.entity.IngressStatsEvent; import com.common.entity.SyslogNonNormalMessage; @@ -80,6 +81,12 @@ public class SysLogProcessor { @Qualifier("nonNormalWriterExecutor") private Executor nonNormalWriterExecutor; + @Value("${dm.non-normal-write.max-concurrency-retries:3}") + private int nonNormalMaxConcurrencyRetries; + + @Value("${dm.non-normal-write.concurrency-retry-backoff-ms:50}") + private long nonNormalConcurrencyRetryBackoffMs; + @Autowired private IngressStatsProjectionWriter ingressStatsProjectionWriter; @@ -585,7 +592,7 @@ public class SysLogProcessor { nonNormalWriterExecutor.execute(() -> { long startedAt = System.nanoTime(); try { - int inserted = nonNormalMessageService.saveBatchMessages(batch); + int inserted = saveNonNormalBatchWithConcurrencyRetry(batch); if (inserted != batch.size()) { nonNormalAsyncFailedTotal.addAndGet(batch.size()); log.error("非标日志异步批写结果异常，Kafka offset 不回退: rows={}, inserted={}, " @@ -615,6 +622,36 @@ public class SysLogProcessor { } } + /** + * 每次 service 调用都经过 Spring 事务代理；达梦死锁会先整体回滚，随后用同一批 ID + * 开启新事务有限重试。该等待只占用独立非标线程，不进入 Kafka ACK 边界。 + */ + private int saveNonNormalBatchWithConcurrencyRetry(List batch) { + int attempt = 0; + while (true) { + try { + return nonNormalMessageService.saveBatchMessages(batch); + } catch (RuntimeException failure) { + if (!DirectStatsProjectionWriter.isConcurrentFailure(failure) + || attempt >= Math.max(0, nonNormalMaxConcurrencyRetries)) { + throw failure; + } + attempt++; + long delayMs = Math.max(0L, nonNormalConcurrencyRetryBackoffMs) * attempt; + log.warn("非标日志批写遇到达梦并发冲突，准备重试: rows={}, attempt={}/{}, backoffMs={}", + batch.size(), attempt, Math.max(0, nonNormalMaxConcurrencyRetries), delayMs); + if (delayMs > 0L) { + try { + Thread.sleep(delayMs); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("非标日志死锁重试等待被中断", interrupted); + } + } + } + } + } + private static long elapsedMillis(long startedAtNanos) { return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAtNanos); } diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java index c0da9d2..3c674ea 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java @@ -69,8 +69,19 @@ public final class CompiledRule { } public ParsedMessage parse(String message) { - String cropped = logNormalData.ParserMessageJsonType(message, - cropper == null ? null : cropper.toLegacyParams()); + String cropped; + try { + cropped = logNormalData.ParserMessageJsonType(message, + cropper == null ? null : cropper.toLegacyParams()); + } catch (IllegalArgumentException invalidJsonContent) { + // JSON 截取器在消息中没有完整 {...} 时会直接抛异常。对 JSON 规则而言， + // 这与 Jackson 解析失败属于同一个确定性业务结果，由上层直接转非标， + // 不能冒泡成系统异常后进入 Retry/DLT。 + if ("json".equals(dataType)) { + return null; + } + throw invalidJsonContent; + } LinkedHashMap parsed; switch (dataType) { case "json": diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DirectStatsProjectionWriter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DirectStatsProjectionWriter.java index 0115cef..759cf21 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DirectStatsProjectionWriter.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DirectStatsProjectionWriter.java @@ -11,6 +11,7 @@ import org.springframework.stereotype.Service; import org.springframework.transaction.support.TransactionSynchronizationManager; import java.sql.Date; +import java.sql.SQLException; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.ZoneId; @@ -61,6 +62,7 @@ public class DirectStatsProjectionWriter { private final JdbcTemplate jdbc; private final StatsWriteModeProvider modeProvider; private final DirectStatsWriteMetrics metrics; + private final StatsMaterializationPublisher materializationPublisher; @Value("${stats.incremental.read-enabled:false}") private boolean readEnabled; @@ -88,10 +90,12 @@ public class DirectStatsProjectionWriter { public DirectStatsProjectionWriter(JdbcTemplate jdbc, StatsWriteModeProvider modeProvider, - DirectStatsWriteMetrics metrics) { + DirectStatsWriteMetrics metrics, + StatsMaterializationPublisher materializationPublisher) { this.jdbc = jdbc; this.modeProvider = modeProvider; this.metrics = metrics; + this.materializationPublisher = materializationPublisher; } public WriteResult writeStandardizedRows(List> rows) { @@ -154,6 +158,8 @@ public class DirectStatsProjectionWriter { status.put("device_table", selectedDeviceTable()); status.put("rule_table", selectedRuleTable()); status.put("today_stats_date_enabled", todayStatsDateEnabled); + status.put("business_materialization_mode", asynchronousMaterialization() + ? "KAFKA_DIRTY_KEY" : "SYNCHRONOUS"); return status; } @@ -202,8 +208,12 @@ public class DirectStatsProjectionWriter { if (modeProvider.materializesBusinessFields() && readEnabled) { long materializeStarted = System.nanoTime(); - materializeDevices(devices); - materializeRules(rules); + if (asynchronousMaterialization()) { + materializationPublisher.enqueueAfterCommit(materializationCommands(devices, rules)); + } else { + materializeDevices(devices); + materializeRules(rules); + } materializeMs = elapsedMs(materializeStarted); } metrics.recordSuccess(accumulator.getInputRows(), devices.size(), rules.size(), @@ -216,7 +226,7 @@ public class DirectStatsProjectionWriter { aggregateMs, mergeMs, materializeMs); } catch (RuntimeException failure) { metrics.recordFailure(); - if (isConcurrentWrite(failure)) { + if (isConcurrentFailure(failure)) { throw new ConcurrentStatsWriteException("直接统计聚合键并发写入失败，可安全回滚后有限重试", failure); } throw new DirectStatsWriteException("直接统计写入失败，事实事务必须整体回滚", failure); @@ -261,7 +271,11 @@ public class DirectStatsProjectionWriter { long materializeMs = 0L; if (modeProvider.materializesBusinessFields() && readEnabled) { long materializeStarted = System.nanoTime(); - materializeDeviceKeys(keys); + if (asynchronousMaterialization()) { + materializationPublisher.enqueueAfterCommit(deviceMaterializationCommands(keys)); + } else { + materializeDeviceKeys(keys); + } materializeMs = elapsedMs(materializeStarted); } metrics.recordSuccess(accumulator.getInputRows(), keys.size(), 0, @@ -270,7 +284,7 @@ public class DirectStatsProjectionWriter { aggregateMs, mergeMs, materializeMs); } catch (RuntimeException failure) { metrics.recordFailure(); - if (isConcurrentWrite(failure)) { + if (isConcurrentFailure(failure)) { throw new ConcurrentStatsWriteException("直接统计局部分组并发写入失败，可安全回滚后有限重试", failure); } throw new DirectStatsWriteException("直接统计局部分组重算失败，事实事务必须整体回滚", failure); @@ -389,6 +403,41 @@ public class DirectStatsProjectionWriter { "WHERE id=? AND del_flag='0'", parameters); } + private boolean asynchronousMaterialization() { + return modeProvider.materializesBusinessFields() && readEnabled + && materializationPublisher.isEnabled(); + } + + private List materializationCommands( + List> devices, + List> rules) { + List commands = new ArrayList<>(devices.size() + rules.size()); + LocalDate today = LocalDate.now(ZoneId.of(businessZone)); + for (Map.Entry entry : devices) { + DirectStatsAccumulator.DailyKey key = entry.getKey(); + if (today.equals(key.getDate())) { + commands.add(StatsMaterializationCommand.device(key.getDate(), key.getDimensionId())); + } + } + for (Map.Entry entry : rules) { + DirectStatsAccumulator.DailyKey key = entry.getKey(); + commands.add(StatsMaterializationCommand.rule(key.getDate(), key.getDimensionId())); + } + return commands; + } + + private List deviceMaterializationCommands( + List keys) { + List commands = new ArrayList<>(keys.size()); + LocalDate today = LocalDate.now(ZoneId.of(businessZone)); + for (DirectStatsAccumulator.DailyKey key : keys) { + if (today.equals(key.getDate())) { + commands.add(StatsMaterializationCommand.device(key.getDate(), key.getDimensionId())); + } + } + return commands; + } + private String selectedDeviceTable() { return modeProvider.writesShadowProjection() ? shadowDeviceTable : directDeviceTable; } @@ -410,7 +459,12 @@ public class DirectStatsProjectionWriter { } } - private static boolean isConcurrentWrite(Throwable failure) { + /** + * 识别 Spring 与不同 JDBC 驱动对并发冲突的各种包装形式。 + * 达梦可能把死锁翻译成 DataIntegrityViolationException，并仅在内层 + * BatchUpdateException 的 SQLState/消息中保留 Deadlock 信息。 + */ + public static boolean isConcurrentFailure(Throwable failure) { Throwable current = failure; while (current != null) { if (current instanceof DuplicateKeyException @@ -418,6 +472,18 @@ public class DirectStatsProjectionWriter { || current instanceof DeadlockLoserDataAccessException) { return true; } + if (current instanceof SQLException) { + String sqlState = ((SQLException) current).getSQLState(); + if (sqlState != null && sqlState.startsWith("40")) { + return true; + } + } + String message = String.valueOf(current.getMessage()) + .toLowerCase(java.util.Locale.ROOT); + if (message.contains("deadlock") || message.contains("死锁") + || message.contains("unique constraint") || message.contains("duplicate key")) { + return true; + } current = current.getCause(); } return false; diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/IngressStatsProjectionWriter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/IngressStatsProjectionWriter.java index d42ed75..93e873b 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/IngressStatsProjectionWriter.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/IngressStatsProjectionWriter.java @@ -2,9 +2,6 @@ package com.common.service; import com.common.entity.IngressStatsEvent; import org.springframework.beans.factory.annotation.Value; -import org.springframework.dao.DuplicateKeyException; -import org.springframework.dao.CannotAcquireLockException; -import org.springframework.dao.DeadlockLoserDataAccessException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; import org.springframework.transaction.PlatformTransactionManager; @@ -15,6 +12,7 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.time.ZoneId; import java.util.Comparator; +import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -30,6 +28,7 @@ public class IngressStatsProjectionWriter { private final JdbcTemplate jdbc; private final PlatformTransactionManager transactionManager; private final StatsWriteModeProvider modeProvider; + private final StatsMaterializationPublisher materializationPublisher; @Value("${stats.ingress-metadata.enabled:false}") private boolean enabled; @@ -50,10 +49,12 @@ public class IngressStatsProjectionWriter { public IngressStatsProjectionWriter(JdbcTemplate jdbc, PlatformTransactionManager transactionManager, - StatsWriteModeProvider modeProvider) { + StatsWriteModeProvider modeProvider, + StatsMaterializationPublisher materializationPublisher) { this.jdbc = jdbc; this.transactionManager = transactionManager; this.modeProvider = modeProvider; + this.materializationPublisher = materializationPublisher; } public int writeBatch(List events) { @@ -89,6 +90,8 @@ public class IngressStatsProjectionWriter { status.put("read_enabled", readEnabled); status.put("target_device_table", modeProvider.writesShadowProjection() ? "device_daily_stats+" + shadowDeviceTable : "device_daily_stats"); + status.put("business_materialization_mode", asynchronousMaterialization() + ? "KAFKA_DIRTY_KEY" : "SYNCHRONOUS"); status.put("successful_batches", successfulBatches.get()); status.put("received_events", receivedEvents.get()); status.put("updated_groups", updatedGroups.get()); @@ -99,9 +102,7 @@ public class IngressStatsProjectionWriter { } private static boolean retryableConcurrentWrite(RuntimeException failure) { - return failure instanceof DuplicateKeyException - || failure instanceof CannotAcquireLockException - || failure instanceof DeadlockLoserDataAccessException; + return DirectStatsProjectionWriter.isConcurrentFailure(failure); } private int writeAggregation(Aggregation aggregation) { @@ -109,6 +110,11 @@ public class IngressStatsProjectionWriter { transaction.setTimeout(Math.max(1, transactionTimeoutSeconds)); Integer updated = transaction.execute(status -> { int groups = 0; + boolean asyncMaterialization = asynchronousMaterialization(); + List commands = asyncMaterialization + ? new ArrayList<>(aggregation.devices.size() + aggregation.collects.size()) + : null; + LocalDate today = LocalDate.now(ZoneId.of(businessZone)); for (Map.Entry entry : aggregation.devices.entrySet()) { DeviceKey key = entry.getKey(); LocalDateTime receiveTime = entry.getValue(); @@ -118,11 +124,15 @@ public class IngressStatsProjectionWriter { if (modeProvider.writesShadowProjection()) { mergeDeviceReceive(safeTable(shadowDeviceTable), key, receiveTime); } - if (readEnabled && LocalDate.now(ZoneId.of(businessZone)).equals(key.date)) { - jdbc.update("UPDATE device_device SET last_receive_time=CASE " + - "WHEN last_receive_time IS NULL OR last_receive_time? THEN ? ELSE first_time END," + - "last_success_time=CASE WHEN last_success_time IS NULL OR last_success_time? THEN ? ELSE first_time END," + + "last_success_time=CASE WHEN last_success_time IS NULL OR last_success_timeTopic key 是维度类型和 ID；Kafka 保证同一维度只落在一个分区，消费组保证同一时刻只由 + * 一个实例处理该分区。命令不携带 delta，因此重复投递和故障重放不会造成重复计数。

+ */ +@Slf4j +@Component +public class StatsMaterializationConsumer { + private static final String DEVICE_MATERIALIZE_WITH_DATE = + "MERGE INTO device_device d USING (SELECT stat_date,device_id,parse_count,non_log_count," + + "last_receive_time FROM %s WHERE stat_date=? AND device_id=?) s " + + "ON (d.id=s.device_id) WHEN MATCHED THEN UPDATE SET " + + "d.today_parse_count=s.parse_count,d.today_non_log_count=s.non_log_count," + + "d.today_stats_date=s.stat_date,d.last_receive_time=CASE " + + "WHEN s.last_receive_time IS NULL THEN d.last_receive_time " + + "WHEN d.last_receive_time IS NULL OR d.last_receive_times.first_success_time " + + "THEN s.first_success_time ELSE c.first_time END," + + "c.last_success_time=CASE WHEN s.last_success_time IS NULL THEN c.last_success_time " + + "WHEN c.last_success_time IS NULL OR c.last_success_time> records, Acknowledgment acknowledgment) { + if (records == null || records.isEmpty()) { + acknowledgment.acknowledge(); + return; + } + consumed.addAndGet(records.size()); + Map commands = parseAndCoalesce(records); + if (commands.isEmpty()) { + acknowledgment.acknowledge(); + return; + } + try { + int count = materializeWithRetry(commands.values()); + acknowledgment.acknowledge(); + applied.addAndGet(count); + successfulBatches.incrementAndGet(); + lastSuccessTime = LocalDateTime.now(); + log.info("统计物化批次完成: received={}, coalesced={}, applied={}, invalidTotal={}", + records.size(), commands.size(), count, invalid.get()); + } catch (RuntimeException failure) { + failures.incrementAndGet(); + lastFailureTime = LocalDateTime.now(); + throw failure; + } + } + + private Map parseAndCoalesce( + List> records) { + Map result = new TreeMap<>(); + for (ConsumerRecord record : records) { + try { + StatsMaterializationCommand command = objectMapper.readValue( + record.value(), StatsMaterializationCommand.class); + command.validate(); + String key = command.key(); + if (record.key() == null || !key.equals(record.key())) { + throw new IllegalArgumentException("Kafka key 与统计物化命令不一致"); + } + result.put(key, command); + } catch (Exception badCommand) { + invalid.incrementAndGet(); + log.error("忽略非法内部统计物化命令: topic={}, partition={}, offset={}, key={}", + record.topic(), record.partition(), record.offset(), record.key(), badCommand); + } + } + return result; + } + + int materializeWithRetry(Collection commands) { + int attempt = 0; + while (true) { + try { + return materializeOnce(commands); + } catch (RuntimeException failure) { + if (!DirectStatsProjectionWriter.isConcurrentFailure(failure) + || attempt >= Math.max(0, maxConcurrencyRetries)) { + throw failure; + } + attempt++; + retries.incrementAndGet(); + sleepBeforeRetry(attempt); + } + } + } + + private int materializeOnce(Collection commands) { + if (!modeProvider.materializesBusinessFields() || !readEnabled) { + throw new IllegalStateException("统计物化消费者只允许在 DIRECT 且 read-enabled 模式运行"); + } + final LocalDate today = LocalDate.now(ZoneId.of(businessZone)); + final List deviceParams = new ArrayList<>(); + final List ruleParams = new ArrayList<>(); + final List collectParams = new ArrayList<>(); + for (StatsMaterializationCommand command : commands) { + command.validate(); + switch (command.getType()) { + case DEVICE: + // yesterday 的迟到事实只保存在日统计表，不允许覆盖今日展示字段。 + if (today.equals(command.date())) { + deviceParams.add(new Object[]{Date.valueOf(today), command.getDimensionId()}); + } + break; + case RULE: + // 按 rule id 从所有日统计取 MAX，命令即使跨日期合并也不会丢失较大值。 + ruleParams.add(new Object[]{command.getDimensionId()}); + break; + case COLLECT: + collectParams.add(new Object[]{command.getDimensionId()}); + break; + default: + throw new IllegalArgumentException("未知统计物化命令类型: " + command.getType()); + } + } + TransactionTemplate transaction = new TransactionTemplate(transactionManager); + transaction.setTimeout(Math.max(1, transactionTimeoutSeconds)); + Integer count = transaction.execute(status -> { + if (!deviceParams.isEmpty()) { + String template = todayStatsDateEnabled + ? DEVICE_MATERIALIZE_WITH_DATE : DEVICE_MATERIALIZE_WITHOUT_DATE; + jdbc.batchUpdate(String.format(template, safeTable(deviceTable)), deviceParams); + } + if (!ruleParams.isEmpty()) { + jdbc.batchUpdate(String.format(RULE_MATERIALIZE, safeTable(ruleTable)), ruleParams); + } + if (!collectParams.isEmpty()) { + jdbc.batchUpdate(COLLECT_MATERIALIZE, collectParams); + } + return deviceParams.size() + ruleParams.size() + collectParams.size(); + }); + return count == null ? 0 : count; + } + + private void sleepBeforeRetry(int attempt) { + try { + Thread.sleep(Math.max(0L, retryBackoffMs) * attempt); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("统计物化死锁重试等待被中断", interrupted); + } + } + + public Map status() { + Map result = new LinkedHashMap<>(); + result.put("consumed_total", consumed.get()); + result.put("invalid_total", invalid.get()); + result.put("applied_total", applied.get()); + result.put("successful_batches", successfulBatches.get()); + result.put("failures_total", failures.get()); + result.put("concurrency_retries_total", retries.get()); + result.put("last_success_time", lastSuccessTime); + result.put("last_failure_time", lastFailureTime); + return result; + } + + private static String safeTable(String table) { + if (table == null || !table.matches("[A-Za-z][A-Za-z0-9_]*")) { + throw new IllegalStateException("非法统计物化源表名: " + table); + } + return table; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationPublisher.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationPublisher.java new file mode 100644 index 0000000..a374400 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationPublisher.java @@ -0,0 +1,128 @@ +package com.common.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.transaction.support.TransactionSynchronizationAdapter; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.util.concurrent.ListenableFuture; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** 事务提交后合并统计脏键并批量发布；失败的 key 留在内存等待下轮，小表校准负责崩溃窗口。 */ +@Slf4j +@Component +public class StatsMaterializationPublisher { + private final KafkaTemplate kafkaTemplate; + private final ObjectMapper objectMapper; + private final ConcurrentMap pending = new ConcurrentHashMap<>(); + private final AtomicBoolean flushing = new AtomicBoolean(); + private final AtomicLong enqueued = new AtomicLong(); + private final AtomicLong published = new AtomicLong(); + private final AtomicLong failures = new AtomicLong(); + private volatile long lastSuccessAt; + private volatile long lastFailureAt; + + @Value("${stats.materialization.enabled:true}") private boolean enabled; + @Value("${stats.materialization.topic:${spring.kafka.consumer.topic}.STATS.MATERIALIZE}") private String topic; + @Value("${stats.materialization.publish-timeout-ms:10000}") private long publishTimeoutMs; + @Value("${stats.materialization.publish-max-batch-size:1000}") private int maxBatchSize; + + public StatsMaterializationPublisher(KafkaTemplate kafkaTemplate, ObjectMapper objectMapper) { + this.kafkaTemplate = kafkaTemplate; + this.objectMapper = objectMapper.copy(); + this.objectMapper.deactivateDefaultTyping(); + } + + public boolean isEnabled() { return enabled; } + + public void enqueueAfterCommit(Collection commands) { + if (!enabled || commands == null || commands.isEmpty()) return; + if (!TransactionSynchronizationManager.isSynchronizationActive()) { + throw new IllegalStateException("统计物化通知必须在 Spring 事务内登记"); + } + List snapshot = new ArrayList<>(commands); + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { + @Override public void afterCommit() { enqueueNow(snapshot); } + }); + } + + public void enqueueNow(Collection commands) { + if (!enabled || commands == null) return; + for (StatsMaterializationCommand command : commands) { + if (command == null) continue; + pending.put(command.key(), command); + enqueued.incrementAndGet(); + } + } + + @Scheduled(fixedDelayString = "${stats.materialization.publish-interval-ms:200}") + public void flush() { + if (!enabled || pending.isEmpty() || !flushing.compareAndSet(false, true)) return; + try { + List snapshot = new ArrayList<>(pending.values()); + snapshot.sort(Comparator.comparing(StatsMaterializationCommand::key)); + int size = Math.min(snapshot.size(), Math.max(1, maxBatchSize)); + List sends = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + StatsMaterializationCommand command = snapshot.get(i); + sends.add(new PendingSend(command, kafkaTemplate.send( + topic, command.key(), objectMapper.writeValueAsString(command)))); + } + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(Math.max(1L, publishTimeoutMs)); + for (PendingSend send : sends) { + long remaining = deadline - System.nanoTime(); + if (remaining <= 0L) throw new java.util.concurrent.TimeoutException("统计物化通知发布超时"); + send.future.get(remaining, TimeUnit.NANOSECONDS); + if (pending.remove(send.command.key(), send.command)) published.incrementAndGet(); + } + lastSuccessAt = System.currentTimeMillis(); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + recordFailure(interrupted); + } catch (Exception failure) { + recordFailure(failure); + } finally { + flushing.set(false); + } + } + + private void recordFailure(Exception failure) { + failures.incrementAndGet(); + lastFailureAt = System.currentTimeMillis(); + log.error("统计物化通知发布失败，脏键保留等待重试: pending={}", pending.size(), failure); + } + + public Map status() { + Map result = new LinkedHashMap<>(); + result.put("enabled", enabled); result.put("topic", topic); + result.put("pending_keys", pending.size()); result.put("enqueued_total", enqueued.get()); + result.put("published_total", published.get()); result.put("publish_failures_total", failures.get()); + result.put("last_success_epoch_ms", lastSuccessAt == 0L ? null : lastSuccessAt); + result.put("last_failure_epoch_ms", lastFailureAt == 0L ? null : lastFailureAt); + return result; + } + + private static final class PendingSend { + private final StatsMaterializationCommand command; + private final ListenableFuture> future; + private PendingSend(StatsMaterializationCommand command, + ListenableFuture> future) { + this.command = command; this.future = future; + } + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationReconciler.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationReconciler.java new file mode 100644 index 0000000..52ff747 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/StatsMaterializationReconciler.java @@ -0,0 +1,154 @@ +package com.common.service; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.sql.Date; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 扫描三个小型聚合表与业务展示字段的差异，补偿“数据库已提交、进程却在发布通知前退出”的窗口。 + * 不扫描 syslog 事实宽表，不参与事实消费事务。 + */ +@Slf4j +@Component +public class StatsMaterializationReconciler { + private final JdbcTemplate jdbc; + private final StatsWriteModeProvider modeProvider; + private final StatsMaterializationPublisher publisher; + private final AtomicBoolean running = new AtomicBoolean(); + private final AtomicLong runs = new AtomicLong(); + private final AtomicLong detected = new AtomicLong(); + private final AtomicLong failures = new AtomicLong(); + private volatile LocalDateTime lastSuccessTime; + private volatile LocalDateTime lastFailureTime; + + @Value("${stats.materialization.enabled:true}") private boolean enabled; + @Value("${stats.incremental.read-enabled:false}") private boolean readEnabled; + @Value("${stats.direct.device-table:device_daily_stats}") private String deviceTable; + @Value("${stats.direct.rule-table:normalize_rule_daily_stats}") private String ruleTable; + @Value("${stats.business-zone:Asia/Shanghai}") private String businessZone; + + public StatsMaterializationReconciler(JdbcTemplate jdbc, + StatsWriteModeProvider modeProvider, + StatsMaterializationPublisher publisher) { + this.jdbc = jdbc; + this.modeProvider = modeProvider; + this.publisher = publisher; + } + + @Scheduled(initialDelayString = "${stats.materialization.reconcile-initial-delay-ms:30000}", + fixedDelayString = "${stats.materialization.reconcile-interval-ms:60000}") + public void reconcile() { + if (!enabled || !readEnabled || !modeProvider.materializesBusinessFields() + || !running.compareAndSet(false, true)) { + return; + } + try { + List commands = findDifferences(); + publisher.enqueueNow(commands); + runs.incrementAndGet(); + detected.addAndGet(commands.size()); + lastSuccessTime = LocalDateTime.now(); + if (!commands.isEmpty()) { + log.warn("统计物化对账发现差异并已重新投递: dirtyKeys={}", commands.size()); + } + } catch (RuntimeException failure) { + failures.incrementAndGet(); + lastFailureTime = LocalDateTime.now(); + log.error("统计物化小表对账失败，下个周期自动重试", failure); + } finally { + running.set(false); + } + } + + List findDifferences() { + LocalDate today = LocalDate.now(ZoneId.of(businessZone)); + List commands = new ArrayList<>(); + String safeDeviceTable = safeTable(deviceTable); + String safeRuleTable = safeTable(ruleTable); + + List> devices = jdbc.queryForList( + "SELECT s.device_id dimension_id FROM " + safeDeviceTable + " s " + + "JOIN device_device d ON d.id=s.device_id WHERE s.stat_date=? AND (" + + "NVL(d.today_parse_count,0)<>NVL(s.parse_count,0) OR " + + "NVL(d.today_non_log_count,0)<>NVL(s.non_log_count,0) OR " + + "d.today_stats_date IS NULL OR d.today_stats_date<>s.stat_date OR " + + "(s.last_receive_time IS NOT NULL AND (d.last_receive_time IS NULL " + + "OR d.last_receive_time row : devices) { + commands.add(StatsMaterializationCommand.device(today, longValue(row, "dimension_id"))); + } + + List> rules = jdbc.queryForList( + "SELECT s.normalize_rule_id dimension_id FROM (SELECT normalize_rule_id," + + "MAX(max_log_time) max_log_time FROM " + safeRuleTable + + " WHERE max_log_time IS NOT NULL GROUP BY normalize_rule_id) s " + + "JOIN dm_normalize_rule r ON r.id=s.normalize_rule_id AND r.del_flag='0' " + + "WHERE r.first_data_saved_at IS NULL OR r.first_data_saved_at row : rules) { + commands.add(StatsMaterializationCommand.rule(today, longValue(row, "dimension_id"))); + } + + List> collects = jdbc.queryForList( + "SELECT s.device_collect_id dimension_id FROM device_collect_stats s " + + "JOIN device_collect_task c ON c.id=s.device_collect_id WHERE " + + "(s.first_success_time IS NOT NULL AND (c.first_time IS NULL " + + "OR c.first_time>s.first_success_time)) OR " + + "(s.last_success_time IS NOT NULL AND (c.last_success_time IS NULL " + + "OR c.last_success_time row : collects) { + commands.add(StatsMaterializationCommand.collect(longValue(row, "dimension_id"))); + } + return commands; + } + + public Map status() { + Map result = new LinkedHashMap<>(); + result.put("enabled", enabled); + result.put("runs_total", runs.get()); + result.put("dirty_keys_detected_total", detected.get()); + result.put("failures_total", failures.get()); + result.put("running", running.get()); + result.put("last_success_time", lastSuccessTime); + result.put("last_failure_time", lastFailureTime); + return result; + } + + private static long longValue(Map row, String name) { + Object value = row.get(name); + if (value == null) value = row.get(name.toUpperCase(Locale.ROOT)); + if (value == null) { + for (Map.Entry entry : row.entrySet()) { + if (entry.getKey().equalsIgnoreCase(name)) { + value = entry.getValue(); + break; + } + } + } + if (!(value instanceof Number)) { + throw new IllegalStateException("统计物化对账结果缺少数字列 " + name + ": " + value); + } + return ((Number) value).longValue(); + } + + private static String safeTable(String table) { + if (table == null || !table.matches("[A-Za-z][A-Za-z0-9_]*")) { + throw new IllegalStateException("非法统计物化对账表名: " + table); + } + return table; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java index 0d93837..243cf49 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/KafkaReliabilityConfig.java @@ -50,6 +50,22 @@ public class KafkaReliabilityConfig { return factory; } + /** 统计脏键与原始日志隔离；同一维度 key 由 Kafka 分区保证只被一个组成员处理。 */ + @Bean("statsMaterializationKafkaListenerContainerFactory") + public ConcurrentKafkaListenerContainerFactory statsMaterializationKafkaListenerContainerFactory( + ConcurrentKafkaListenerContainerFactoryConfigurer configurer, + ConsumerFactory consumerFactory, + DefaultErrorHandler syslogDefaultErrorHandler, + @Value("${stats.materialization.listener-concurrency:1}") int concurrency) { + if (concurrency < 1) { + throw new IllegalArgumentException("stats.materialization.listener-concurrency 必须大于 0"); + } + ConcurrentKafkaListenerContainerFactory factory = + configuredFactory(configurer, consumerFactory, syslogDefaultErrorHandler); + factory.setConcurrency(concurrency); + return factory; + } + private ConcurrentKafkaListenerContainerFactory configuredFactory( ConcurrentKafkaListenerContainerFactoryConfigurer configurer, ConsumerFactory consumerFactory, diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/DirectStatsController.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/DirectStatsController.java index 5ae320a..192a61f 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/DirectStatsController.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/DirectStatsController.java @@ -4,6 +4,9 @@ import com.common.service.DirectStatsProjectionWriter; import com.common.service.DirectStatsDayRolloverService; import com.common.service.IngressStatsProjectionWriter; import com.common.service.DmStandardizedBatchExecutor; +import com.common.service.StatsMaterializationConsumer; +import com.common.service.StatsMaterializationPublisher; +import com.common.service.StatsMaterializationReconciler; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; @@ -19,15 +22,24 @@ public class DirectStatsController { private final IngressStatsProjectionWriter ingressWriter; private final DirectStatsDayRolloverService dayRolloverService; private final DmStandardizedBatchExecutor batchExecutor; + private final StatsMaterializationPublisher materializationPublisher; + private final StatsMaterializationConsumer materializationConsumer; + private final StatsMaterializationReconciler materializationReconciler; public DirectStatsController(DirectStatsProjectionWriter writer, IngressStatsProjectionWriter ingressWriter, DirectStatsDayRolloverService dayRolloverService, - DmStandardizedBatchExecutor batchExecutor) { + DmStandardizedBatchExecutor batchExecutor, + StatsMaterializationPublisher materializationPublisher, + StatsMaterializationConsumer materializationConsumer, + StatsMaterializationReconciler materializationReconciler) { this.writer = writer; this.ingressWriter = ingressWriter; this.dayRolloverService = dayRolloverService; this.batchExecutor = batchExecutor; + this.materializationPublisher = materializationPublisher; + this.materializationConsumer = materializationConsumer; + this.materializationReconciler = materializationReconciler; } @GetMapping("/status") @@ -36,6 +48,9 @@ public class DirectStatsController { status.put("ingress_metadata", ingressWriter.status()); status.put("direct_day_rollover", dayRolloverService.status()); status.put("dm_transaction", batchExecutor.status()); + status.put("materialization_publisher", materializationPublisher.status()); + status.put("materialization_consumer", materializationConsumer.status()); + status.put("materialization_reconciler", materializationReconciler.status()); return ResponseEntity.ok(status); } } diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties index 1943d00..222e1d4 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties @@ -160,6 +160,8 @@ dm.standardized-write.circuit-breaker-failure-threshold=3 dm.standardized-write.circuit-breaker-backoff-ms=30000 dm.non-normal-write.batch-size=${DM_NON_NORMAL_WRITE_BATCH_SIZE:500} dm.non-normal-write.queue-capacity=${DM_NON_NORMAL_WRITE_QUEUE_CAPACITY:500} +dm.non-normal-write.max-concurrency-retries=${DM_NON_NORMAL_WRITE_MAX_CONCURRENCY_RETRIES:3} +dm.non-normal-write.concurrency-retry-backoff-ms=${DM_NON_NORMAL_WRITE_CONCURRENCY_RETRY_BACKOFF_MS:50} # ETL配置 etl.batch.page-size=1000 @@ -193,12 +195,26 @@ stats.direct.shadow-device-table=device_daily_stats_direct_shadow stats.direct.shadow-rule-table=normalize_rule_daily_stats_direct_shadow stats.direct.control-table=stats_direct_control stats.direct.today-stats-date-enabled=true -stats.direct.max-concurrency-retries=1 -stats.direct.concurrency-retry-backoff-ms=20 +stats.direct.max-concurrency-retries=${STATS_DIRECT_MAX_CONCURRENCY_RETRIES:3} +stats.direct.concurrency-retry-backoff-ms=${STATS_DIRECT_CONCURRENCY_RETRY_BACKOFF_MS:50} stats.direct.transaction-log-enabled=false stats.direct.day-rollover-check-cron=5 * * * * ? stats.business-zone=Asia/Shanghai stats.fact-only.confirm-rebuild-required=false +# 事实事务只维护小型精确统计表；提交后按维度脏键异步刷新业务展示字段。 +# 三个实例必须配置相同 topic/group-id，Topic 分区数建议与原始 Topic 相同。 +stats.materialization.enabled=${STATS_MATERIALIZATION_ENABLED:true} +stats.materialization.topic=${STATS_MATERIALIZATION_TOPIC:${spring.kafka.consumer.topic}.STATS.MATERIALIZE} +stats.materialization.group-id=${STATS_MATERIALIZATION_GROUP_ID:${spring.kafka.consumer.group-id}-stats-materialize} +# 每个实例一个物化 consumer；现场三个实例合计三个，足以处理低基数脏键并减少达梦并发。 +stats.materialization.listener-concurrency=${STATS_MATERIALIZATION_LISTENER_CONCURRENCY:1} +stats.materialization.publish-interval-ms=${STATS_MATERIALIZATION_PUBLISH_INTERVAL_MS:200} +stats.materialization.publish-timeout-ms=${STATS_MATERIALIZATION_PUBLISH_TIMEOUT_MS:10000} +stats.materialization.publish-max-batch-size=${STATS_MATERIALIZATION_PUBLISH_MAX_BATCH_SIZE:1000} +stats.materialization.max-concurrency-retries=${STATS_MATERIALIZATION_MAX_CONCURRENCY_RETRIES:3} +stats.materialization.concurrency-retry-backoff-ms=${STATS_MATERIALIZATION_CONCURRENCY_RETRY_BACKOFF_MS:50} +stats.materialization.reconcile-initial-delay-ms=${STATS_MATERIALIZATION_RECONCILE_INITIAL_DELAY_MS:30000} +stats.materialization.reconcile-interval-ms=${STATS_MATERIALIZATION_RECONCILE_INTERVAL_MS:60000} # 标准化规则编译与批次缓存灰度开关 normalization.jetcache.enabled=true 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 0233f33..02e947e 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 @@ -122,6 +122,23 @@ class LogNormalProcessorRuleSelectionTest { verify(fixture.nonNormalService, never()).saveMessage(any()); } + @Test + void jsonWithoutObjectBoundaryAlsoGoesDirectlyToDeferredNonNormal() { + RuleSource invalidJson = new RuleSource(10L, "json-no-cropper", "json-no-cropper", + "json", "{\"mappers\":[" + equalMapping() + + "],\"complete_columns\":[]}", true, "0"); + RuleSource regexWouldMatch = regexRule(20L, "regex-second", "value=(\\S+)"); + Fixture fixture = fixture(Arrays.asList(invalidJson, regexWouldMatch), config(true)); + + LogNormalProcessor.NormalizationResult result = fixture.prepareActive("value=ok"); + + 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 legacyInvalidJsonIsNonNormalTerminalInsteadOfException() { Fixture fixture = fixture(Arrays.asList( diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java index ed5e055..ade3588 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java @@ -13,6 +13,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.dao.DataIntegrityViolationException; import java.time.Instant; import java.time.LocalDateTime; @@ -63,6 +64,8 @@ class SysLogProcessorParallelTest { ReflectionTestUtils.setField(processor, "standardizedBatchWriter", batchWriter); ReflectionTestUtils.setField(processor, "nonNormalMessageService", nonNormalMessageService); ReflectionTestUtils.setField(processor, "nonNormalWriterExecutor", nonNormalExecutor); + ReflectionTestUtils.setField(processor, "nonNormalMaxConcurrencyRetries", 3); + ReflectionTestUtils.setField(processor, "nonNormalConcurrencyRetryBackoffMs", 0L); when(batchWriter.write(any())).thenReturn(new DmStandardizedBatchWriter.WriteReport<>()); when(normalProcessor.captureRuleSnapshot(any())).thenReturn( new BatchRuleSnapshot(Collections.emptyMap(), System.currentTimeMillis())); @@ -182,6 +185,31 @@ class SysLogProcessorParallelTest { assertEquals(2, metrics.filteredCount()); } + @Test + void asynchronousNonNormalDeadlockRetriesInANewServiceTransaction() { + when(normalProcessor.prepare(any(PreparedNormalizationInput.class), + any(), any(), any())).thenAnswer(invocation -> { + SyslogNonNormalMessage message = new SyslogNonNormalMessage(); + message.setId(java.util.UUID.randomUUID().toString()); + return LogNormalProcessor.NormalizationResult.filtered( + Collections.singletonList(message)); + }); + AtomicInteger attempts = new AtomicInteger(); + when(nonNormalMessageService.saveBatchMessages(anyList())).thenAnswer(invocation -> { + if (attempts.incrementAndGet() == 1) { + throw new RuntimeException(new DataIntegrityViolationException("Deadlock")); + } + return ((List) invocation.getArgument(0)).size(); + }); + + SysLogProcessor.BatchMetrics metrics = new SysLogProcessor.BatchMetrics(); + processor.normalizePreparedMessages(messages(2), metrics); + + verify(nonNormalMessageService, timeout(1000).times(2)).saveBatchMessages(anyList()); + assertEquals(2, attempts.get()); + assertEquals(2, metrics.nonNormalQueued()); + } + @Test void rejectedNonNormalTaskDoesNotFailKafkaBatch() { when(normalProcessor.prepare(any(PreparedNormalizationInput.class), diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java index 2c09a51..5fa2659 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SyslogKafkaTopologyTest.java @@ -9,6 +9,7 @@ import java.util.List; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import com.common.service.StatsMaterializationConsumer; /** 防止多节点 consumer-rule 的监听器 ID 或显式消费组在后续合并中回退。 */ class SyslogKafkaTopologyTest { @@ -48,4 +49,20 @@ class SyslogKafkaTopologyTest { "${syslog.reliability.replay-topic:${spring.kafka.consumer.topic}.REPLAY}" }, listener.topics()); } + + @Test + void statsMaterializerUsesDedicatedTopicAndGroup() throws Exception { + Method method = StatsMaterializationConsumer.class.getMethod( + "consume", List.class, org.springframework.kafka.support.Acknowledgment.class); + KafkaListener listener = method.getAnnotation(KafkaListener.class); + + assertNotNull(listener); + assertEquals("syslog-stats-materializer", listener.id()); + assertEquals("${stats.materialization.group-id:${spring.kafka.consumer.group-id}-stats-materialize}", + listener.groupId()); + assertEquals("statsMaterializationKafkaListenerContainerFactory", listener.containerFactory()); + assertArrayEquals(new String[]{ + "${stats.materialization.topic:${spring.kafka.consumer.topic}.STATS.MATERIALIZE}" + }, listener.topics()); + } } diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DirectStatsProjectionWriterTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DirectStatsProjectionWriterTest.java index ebfb1ed..efc5b60 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DirectStatsProjectionWriterTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DirectStatsProjectionWriterTest.java @@ -4,6 +4,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.mockito.Answers; +import org.springframework.dao.DataIntegrityViolationException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.transaction.support.TransactionSynchronizationManager; @@ -11,6 +12,7 @@ import org.springframework.transaction.support.TransactionSynchronizationManager import java.time.LocalDateTime; import java.time.LocalDate; import java.time.ZoneId; +import java.sql.BatchUpdateException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -19,7 +21,9 @@ import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyCollection; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; @@ -105,10 +109,51 @@ class DirectStatsProjectionWriterTest { assertTrue(sql.getAllValues().get(1).contains("UPDATE device_device")); } + @Test + void damengBatchDeadlockWrappedAsIntegrityViolationIsRetryable() { + BatchUpdateException deadlock = new BatchUpdateException("Deadlock", null, 0, new int[0]); + DataIntegrityViolationException wrapped = + new DataIntegrityViolationException("PreparedStatementCallback", deadlock); + + assertTrue(DirectStatsProjectionWriter.isConcurrentFailure(wrapped)); + } + + @Test + void kafkaMaterializationRemovesBusinessTableUpdatesFromFactTransaction() { + JdbcTemplate jdbc = mock(JdbcTemplate.class); + StatsMaterializationPublisher publisher = mock(StatsMaterializationPublisher.class); + when(publisher.isEnabled()).thenReturn(true); + DirectStatsProjectionWriter writer = new DirectStatsProjectionWriter( + jdbc, new StatsWriteModeProvider("DIRECT"), new DirectStatsWriteMetrics(), publisher); + configure(writer, true); + TransactionSynchronizationManager.setActualTransactionActive(true); + + Map row = new HashMap<>(); + row.put("created_at", LocalDateTime.now()); + row.put("log_time", LocalDateTime.now()); + row.put("device_id", 7L); + row.put("normalize_rule_id", 11L); + writer.writeStandardizedRow(row); + + ArgumentCaptor sql = ArgumentCaptor.forClass(String.class); + verify(jdbc, times(2)).batchUpdate(sql.capture(), anyList()); + assertTrue(sql.getAllValues().get(0).contains("MERGE INTO device_daily_stats")); + assertTrue(sql.getAllValues().get(1).contains("MERGE INTO normalize_rule_daily_stats")); + verify(publisher).enqueueAfterCommit(anyCollection()); + assertTrue("KAFKA_DIRTY_KEY".equals(writer.status().get("business_materialization_mode"))); + } + private DirectStatsProjectionWriter writer(JdbcTemplate jdbc, String mode, boolean readEnabled) { StatsWriteModeProvider modeProvider = new StatsWriteModeProvider(mode); + StatsMaterializationPublisher publisher = mock(StatsMaterializationPublisher.class); + when(publisher.isEnabled()).thenReturn(false); DirectStatsProjectionWriter writer = new DirectStatsProjectionWriter( - jdbc, modeProvider, new DirectStatsWriteMetrics()); + jdbc, modeProvider, new DirectStatsWriteMetrics(), publisher); + configure(writer, readEnabled); + return writer; + } + + private void configure(DirectStatsProjectionWriter writer, boolean readEnabled) { ReflectionTestUtils.setField(writer, "readEnabled", readEnabled); ReflectionTestUtils.setField(writer, "directDeviceTable", "device_daily_stats"); ReflectionTestUtils.setField(writer, "directRuleTable", "normalize_rule_daily_stats"); @@ -117,6 +162,5 @@ class DirectStatsProjectionWriterTest { ReflectionTestUtils.setField(writer, "controlTable", "stats_direct_control"); ReflectionTestUtils.setField(writer, "todayStatsDateEnabled", true); ReflectionTestUtils.setField(writer, "businessZone", "Asia/Shanghai"); - return writer; } } diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationCommandTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationCommandTest.java new file mode 100644 index 0000000..7691993 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationCommandTest.java @@ -0,0 +1,28 @@ +package com.common.service; + +import org.junit.jupiter.api.Test; + +import java.time.LocalDate; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class StatsMaterializationCommandTest { + @Test + void usesStableDimensionKeyAndValidatesDate() { + StatsMaterializationCommand command = StatsMaterializationCommand.device( + LocalDate.of(2026, 9, 11), 7L); + + assertEquals("DEVICE:7", command.key()); + assertEquals(LocalDate.of(2026, 9, 11), command.date()); + } + + @Test + void rejectsMissingDateForDailyDimension() { + StatsMaterializationCommand command = new StatsMaterializationCommand(); + command.setType(StatsMaterializationCommand.Type.RULE); + command.setDimensionId(8L); + + assertThrows(IllegalArgumentException.class, command::validate); + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationConsumerTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationConsumerTest.java new file mode 100644 index 0000000..be579a7 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationConsumerTest.java @@ -0,0 +1,59 @@ +package com.common.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.SimpleTransactionStatus; + +import java.time.LocalDate; +import java.time.ZoneId; +import java.util.Arrays; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class StatsMaterializationConsumerTest { + @Test + void materializesExactValuesInOneTransaction() { + JdbcTemplate jdbc = mock(JdbcTemplate.class); + PlatformTransactionManager manager = mock(PlatformTransactionManager.class); + when(manager.getTransaction(any(TransactionDefinition.class))) + .thenReturn(new SimpleTransactionStatus()); + StatsMaterializationConsumer consumer = new StatsMaterializationConsumer( + jdbc, manager, new ObjectMapper(), new StatsWriteModeProvider("DIRECT")); + ReflectionTestUtils.setField(consumer, "readEnabled", true); + ReflectionTestUtils.setField(consumer, "deviceTable", "device_daily_stats"); + ReflectionTestUtils.setField(consumer, "ruleTable", "normalize_rule_daily_stats"); + ReflectionTestUtils.setField(consumer, "todayStatsDateEnabled", true); + ReflectionTestUtils.setField(consumer, "businessZone", "Asia/Shanghai"); + ReflectionTestUtils.setField(consumer, "transactionTimeoutSeconds", 45); + ReflectionTestUtils.setField(consumer, "maxConcurrencyRetries", 3); + ReflectionTestUtils.setField(consumer, "retryBackoffMs", 0L); + LocalDate today = LocalDate.now(ZoneId.of("Asia/Shanghai")); + + int count = consumer.materializeWithRetry(Arrays.asList( + StatsMaterializationCommand.device(today, 7L), + StatsMaterializationCommand.rule(today, 8L), + StatsMaterializationCommand.collect(2L))); + + assertEquals(3, count); + ArgumentCaptor sql = ArgumentCaptor.forClass(String.class); + verify(jdbc, times(3)).batchUpdate(sql.capture(), anyList()); + List statements = sql.getAllValues(); + assertTrue(statements.get(0).contains("MERGE INTO device_device")); + assertTrue(statements.get(1).contains("MAX(max_log_time)")); + assertTrue(statements.get(2).contains("MERGE INTO device_collect_task")); + verify(manager).commit(any()); + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationPublisherTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationPublisherTest.java new file mode 100644 index 0000000..afa3913 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationPublisherTest.java @@ -0,0 +1,41 @@ +package com.common.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import java.time.LocalDate; +import java.util.Collections; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; + +class StatsMaterializationPublisherTest { + @AfterEach + void clearTransactionState() { + TransactionSynchronizationManager.clear(); + } + + @Test + void dirtyKeyBecomesPublishableOnlyAfterDatabaseCommit() { + @SuppressWarnings("unchecked") + KafkaTemplate kafka = mock(KafkaTemplate.class); + StatsMaterializationPublisher publisher = new StatsMaterializationPublisher(kafka, new ObjectMapper()); + ReflectionTestUtils.setField(publisher, "enabled", true); + TransactionSynchronizationManager.initSynchronization(); + + publisher.enqueueAfterCommit(Collections.singletonList( + StatsMaterializationCommand.device(LocalDate.of(2026, 9, 11), 7L))); + assertEquals(0, publisher.status().get("pending_keys")); + + for (TransactionSynchronization synchronization + : TransactionSynchronizationManager.getSynchronizations()) { + synchronization.afterCommit(); + } + assertEquals(1, publisher.status().get("pending_keys")); + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationReconcilerTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationReconcilerTest.java new file mode 100644 index 0000000..91a095e --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/StatsMaterializationReconcilerTest.java @@ -0,0 +1,43 @@ +package com.common.service; + +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; + +class StatsMaterializationReconcilerTest { + @Test + void scansOnlyAggregateTablesAndBuildsExactDirtyKeys() { + JdbcTemplate jdbc = mock(JdbcTemplate.class, invocation -> { + if (!"queryForList".equals(invocation.getMethod().getName())) { + return org.mockito.Answers.RETURNS_DEFAULTS.answer(invocation); + } + String sql = (String) invocation.getArguments()[0]; + Map row = new HashMap<>(); + if (sql.contains("device_daily_stats")) row.put("DIMENSION_ID", 7L); + else if (sql.contains("normalize_rule_daily_stats")) row.put("dimension_id", 8L); + else if (sql.contains("device_collect_stats")) row.put("DIMENSION_ID", 2L); + else return Collections.emptyList(); + return Collections.singletonList(row); + }); + StatsMaterializationReconciler reconciler = new StatsMaterializationReconciler( + jdbc, new StatsWriteModeProvider("DIRECT"), mock(StatsMaterializationPublisher.class)); + ReflectionTestUtils.setField(reconciler, "deviceTable", "device_daily_stats"); + ReflectionTestUtils.setField(reconciler, "ruleTable", "normalize_rule_daily_stats"); + ReflectionTestUtils.setField(reconciler, "businessZone", "Asia/Shanghai"); + + List commands = reconciler.findDifferences(); + + assertEquals(3, commands.size()); + assertEquals("DEVICE:7", commands.get(0).key()); + assertEquals("RULE:8", commands.get(1).key()); + assertEquals("COLLECT:2", commands.get(2).key()); + } +} diff --git a/haobang-security-dm/upgrade-doc/v1/kafka/kafka-cluster-upgrade.sh b/haobang-security-dm/upgrade-doc/v1/kafka/kafka-cluster-upgrade.sh index 100b1ff..2119476 100755 --- a/haobang-security-dm/upgrade-doc/v1/kafka/kafka-cluster-upgrade.sh +++ b/haobang-security-dm/upgrade-doc/v1/kafka/kafka-cluster-upgrade.sh @@ -57,11 +57,11 @@ fi echo "新 Topic 使用分区数=${PARTITIONS}，副本数=${REPLICAS}" -SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY") -RETENTIONS=(259200000 604800000 604800000 7776000000 604800000) +SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY" "STATS.MATERIALIZE") +RETENTIONS=(259200000 604800000 604800000 7776000000 604800000 604800000) echo "[3/4] 创建或更新升级 Topic" -for i in 0 1 2 3 4; do +for i in 0 1 2 3 4 5; do TOPIC="${RAW_TOPIC}.${SUFFIXES[$i]}" RETENTION_MS="${RETENTIONS[$i]}" diff --git a/haobang-security-dm/upgrade-doc/v1/kafka/kafka-upgrade.sh b/haobang-security-dm/upgrade-doc/v1/kafka/kafka-upgrade.sh index 0f418b1..5b378f3 100755 --- a/haobang-security-dm/upgrade-doc/v1/kafka/kafka-upgrade.sh +++ b/haobang-security-dm/upgrade-doc/v1/kafka/kafka-upgrade.sh @@ -39,10 +39,10 @@ if [[ ! "${PARTITIONS}" =~ ^[1-9][0-9]*$ ]] || [[ ! "${REPLICAS}" =~ ^[1-9][0-9] exit 1 fi -SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY") -RETENTIONS=(259200000 604800000 604800000 7776000000 604800000) +SUFFIXES=("RETRY.1M" "RETRY.10M" "RETRY.1H" "DLT" "REPLAY" "STATS.MATERIALIZE") +RETENTIONS=(259200000 604800000 604800000 7776000000 604800000 604800000) -for i in 0 1 2 3 4; do +for i in 0 1 2 3 4 5; do TOPIC="${RAW_TOPIC}.${SUFFIXES[$i]}" echo "创建或确认 Topic：${TOPIC}" "${TOPICS_CMD}" "${COMMON_ARGS[@]}" \ diff --git a/haobang-security-dm/upgrade-doc/v2/README.md b/haobang-security-dm/upgrade-doc/v2/README.md index a56e02b..339f1f2 100644 --- a/haobang-security-dm/upgrade-doc/v2/README.md +++ b/haobang-security-dm/upgrade-doc/v2/README.md @@ -162,6 +162,17 @@ services: ## 发布顺序 +升级 `syslog-consumer-rule` 前，先在 Kafka 集群执行一次 +`upgrade-doc/v1/kafka/kafka-cluster-upgrade.sh`，确认已经创建： + +```text +agent-01-syslog-topic.STATS.MATERIALIZE +``` + +实际前缀以现场 `spring.kafka.consumer.topic` 为准。统计 Topic 的分区数、副本数会沿用原始 Topic； +三个 `syslog-consumer-rule` 实例必须使用相同的 `STATS_MATERIALIZATION_TOPIC` 和 +`STATS_MATERIALIZATION_GROUP_ID`。 + 三个 JAR 一起升级： 1. `syslog-consumer` @@ -178,6 +189,8 @@ API 任务先执行 validate。先启用告警任务，确认数量、时间和 - 三个服务启动正常； - Kafka lag 能回到 0，Retry/DLT 没有持续增长； +- 统计物化 Topic 没有持续积压，日志持续出现 `统计物化批次完成`； +- `/api/syslog/stats/direct/status` 中物化发布、消费、对账均无持续失败； - WAL checkpoint 能追上 durable sequence； - API validate 的配置、DNS、TLS、认证和最小查询全部通过； - API run 中 `recordsFetched` 与 `recordsDurable` 一致，`recordsQuarantined=0`； diff --git a/haobang-security-dm/upgrade-doc/v2/config/syslog-consumer-rule.env.example b/haobang-security-dm/upgrade-doc/v2/config/syslog-consumer-rule.env.example index 98ca2b0..143b3a9 100644 --- a/haobang-security-dm/upgrade-doc/v2/config/syslog-consumer-rule.env.example +++ b/haobang-security-dm/upgrade-doc/v2/config/syslog-consumer-rule.env.example @@ -17,3 +17,14 @@ NORMALIZATION_RULE_SELECTION_AMBIGUOUS_FALLBACK=LOWEST_RULE_ID NORMALIZATION_RULE_SELECTION_DETAIL_LOG_ENABLED=false NORMALIZATION_RULE_SELECTION_EXPLICIT_ENABLED=true NORMALIZATION_RULE_SELECTION_AUTOMATIC_FALLBACK_ENABLED=true + +# 统计投影解耦：三个 syslog-consumer-rule 实例保持完全相同的 Topic 与 Group ID。 +STATS_MATERIALIZATION_ENABLED=true +STATS_MATERIALIZATION_TOPIC=agent-01-syslog-topic.STATS.MATERIALIZE +STATS_MATERIALIZATION_GROUP_ID=agent-01-syslog-group-stats-materialize +STATS_MATERIALIZATION_LISTENER_CONCURRENCY=1 +STATS_MATERIALIZATION_PUBLISH_INTERVAL_MS=200 +STATS_MATERIALIZATION_PUBLISH_MAX_BATCH_SIZE=1000 +STATS_MATERIALIZATION_MAX_CONCURRENCY_RETRIES=3 +STATS_MATERIALIZATION_CONCURRENCY_RETRY_BACKOFF_MS=50 +STATS_MATERIALIZATION_RECONCILE_INTERVAL_MS=60000