feat(syslog): 增加积压采样免采样窗口

- 新增 start-lag-ms 配置，默认15分钟内全量保留
- 在15分钟至1小时之间按10级逐步降低保留率
- 延迟达到1小时后使用普通10%、非标20%的最低保留率
- 校验起始延迟必须小于最低保留率阈值
- 状态接口新增 start_lag_ms 配置展示
- 将采样策略版本升级为 v2
- 更新 Docker Compose 和 v4 升级配置说明
- 补充采样时间边界及非法配置单元测试

haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml

@@ -61,6 +61,16 @@ services:
       # 特殊异常优先写入 DLT；DLT 失败也记录后放行，避免单条/单批异常永久阻塞分区。
       - SYSLOG_FAILURE_CONTINUITY_ENABLED=true
       - SYSLOG_FAILURE_CONTINUITY_DLT_PUBLISH_TIMEOUT_MS=1000
+      # 积压采样默认开启；升级应用前先执行 upgrade-doc/v4/database 脚本。
+      - SYSLOG_BACKLOG_SAMPLING_ENABLED=true
+      - SYSLOG_BACKLOG_SAMPLING_START_LAG_MS=900000
+      - SYSLOG_BACKLOG_SAMPLING_LAG_THRESHOLD_MS=3600000
+      - SYSLOG_BACKLOG_SAMPLING_NORMAL_RETAIN_BPS=1000
+      - SYSLOG_BACKLOG_SAMPLING_NON_NORMAL_RETAIN_BPS=2000
+      - SYSLOG_BACKLOG_SAMPLING_DEGRADE_STEPS=10
+      - SYSLOG_BACKLOG_SAMPLING_POLICY_VERSION=v2
+      - SYSLOG_BACKLOG_SAMPLING_AUDIT_TIMEOUT_SECONDS=5
+      - SYSLOG_BACKLOG_SAMPLING_AUDIT_RETRY_BACKOFF_MS=60000
       # 大日志的 poll、fetch、批次和规则计算线程池均独立且有界。
       - SYSLOG_LARGE_LISTENER_ENABLED=true
       - SYSLOG_LARGE_LISTENER_CONCURRENCY=1
@@ -96,6 +106,9 @@ services:
       - INFLUXDB_RAW_QUERY_MAX_RESPONSE_BYTES=10485760
       - INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES=262144
       - INFLUXDB_RAW_WRITE_BATCH_MAX_BYTES=8388608
+      # InfluxDB 原文写是非关键旁路：请求超时后熔断退避，主消费继续。
+      - INFLUXDB_RAW_WRITE_REQUEST_TIMEOUT=5s
+      - INFLUXDB_RAW_FAILURE_BACKOFF_MS=30000
       - DM_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES=262144
       - DM_NON_NORMAL_LARGE_FIELD_MAX_UTF8_BYTES=262144
       - INFLUXDB_BATCH_SIZE=1000

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/BacklogSamplingDecider.java

@@ -0,0 +1,99 @@
+package com.Modules.NormalData;
+
+import com.config.BacklogSamplingProperties;
+import org.springframework.stereotype.Component;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.temporal.ChronoUnit;
+
+/** 按延迟阶梯计算当前保留率，再以 event_id 做确定性哈希采样。 */
+@Component
+public class BacklogSamplingDecider {
+    private static final ZoneId RECEIVE_TIME_ZONE = ZoneId.of("Asia/Shanghai");
+    private final BacklogSamplingProperties properties;
+
+    public BacklogSamplingDecider(BacklogSamplingProperties properties) {
+        this.properties = properties;
+    }
+
+    public BacklogSamplingDecision decide(String eventId, LocalDateTime receivedAt,
+                                           boolean hasAlarm, boolean hasNormal,
+                                           boolean hasNonNormal) {
+        return decide(eventId, receivedAt, hasAlarm, hasNormal, hasNonNormal, Instant.now());
+    }
+
+    BacklogSamplingDecision decide(String eventId, LocalDateTime receivedAt,
+                                   boolean hasAlarm, boolean hasNormal,
+                                   boolean hasNonNormal, Instant now) {
+        LocalDateTime effectiveReceivedAt = receivedAt == null
+                ? LocalDateTime.ofInstant(now, RECEIVE_TIME_ZONE) : receivedAt;
+        long receivedEpochMs = effectiveReceivedAt.atZone(RECEIVE_TIME_ZONE).toInstant().toEpochMilli();
+        long lagMs = Math.max(0L, now.toEpochMilli() - receivedEpochMs);
+        LocalDateTime sourceHour = effectiveReceivedAt.truncatedTo(ChronoUnit.HOURS);
+        int steps = Math.max(1, properties.getDegradeSteps());
+        int currentStep = currentStep(
+                lagMs, properties.getStartLagMs(), properties.getLagThresholdMs(), steps);
+        int normalRetainBps = effectiveRetainBps(
+                properties.getNormalRetainBps(), currentStep, steps);
+        int nonNormalRetainBps = effectiveRetainBps(
+                properties.getNonNormalRetainBps(), currentStep, steps);
+        boolean degraded = properties.isEnabled() && currentStep > 0
+                && (normalRetainBps < BacklogSamplingProperties.BASIS_POINTS
+                || nonNormalRetainBps < BacklogSamplingProperties.BASIS_POINTS);
+        if (!degraded) {
+            return new BacklogSamplingDecision(false, true, true, true, lagMs, sourceHour,
+                    properties.getPolicyVersion(), BacklogSamplingProperties.BASIS_POINTS,
+                    BacklogSamplingProperties.BASIS_POINTS);
+        }
+
+        // 没有产出行的 FILTERED 消息仍视为普通消息，使其 InfluxDB 原文遵守普通保留率。
+        boolean normalCandidate = hasNormal || (!hasAlarm && !hasNonNormal);
+        boolean keepNormal = !normalCandidate || sampled(eventId, "NORMAL",
+                normalRetainBps, properties.getPolicyVersion());
+        boolean keepNonNormal = !hasNonNormal || sampled(eventId, "NON_NORMAL",
+                nonNormalRetainBps, properties.getPolicyVersion());
+        boolean keepRaw = hasAlarm
+                || (normalCandidate && keepNormal)
+                || (hasNonNormal && keepNonNormal);
+        return new BacklogSamplingDecision(true, keepNormal, keepNonNormal, keepRaw, lagMs,
+                sourceHour, properties.getPolicyVersion(), normalRetainBps,
+                nonNormalRetainBps);
+    }
+
+    static int currentStep(long lagMs, long startLagMs, long thresholdMs, int steps) {
+        if (lagMs <= startLagMs) return 0;
+        if (lagMs >= thresholdMs) return steps;
+        long degradableLagMs = lagMs - startLagMs;
+        long degradableWindowMs = thresholdMs - startLagMs;
+        return Math.min(steps, (int) Math.floor(
+                ((double) degradableLagMs / (double) degradableWindowMs) * steps));
+    }
+
+    static int effectiveRetainBps(int minimumRetainBps, int currentStep, int steps) {
+        if (currentStep <= 0) return BacklogSamplingProperties.BASIS_POINTS;
+        if (currentStep >= steps) return minimumRetainBps;
+        long totalDrop = BacklogSamplingProperties.BASIS_POINTS - minimumRetainBps;
+        long appliedDrop = totalDrop * currentStep / steps;
+        return (int) (BacklogSamplingProperties.BASIS_POINTS - appliedDrop);
+    }
+
+    static boolean sampled(String eventId, String category, int retainBps, String policyVersion) {
+        if (retainBps <= 0) return false;
+        if (retainBps >= BacklogSamplingProperties.BASIS_POINTS) return true;
+        String input = String.valueOf(policyVersion) + '\u0000' + category + '\u0000' + String.valueOf(eventId);
+        try {
+            byte[] digest = MessageDigest.getInstance("SHA-256")
+                    .digest(input.getBytes(StandardCharsets.UTF_8));
+            int value = ByteBuffer.wrap(digest, 0, 4).getInt() & Integer.MAX_VALUE;
+            return value % BacklogSamplingProperties.BASIS_POINTS < retainBps;
+        } catch (NoSuchAlgorithmException impossible) {
+            throw new IllegalStateException("JVM 缺少 SHA-256", impossible);
+        }
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/BacklogSamplingDecision.java

@@ -0,0 +1,40 @@
+package com.Modules.NormalData;
+
+import java.time.LocalDateTime;
+
+/** 一条原始 Kafka 消息完成规则计算后的稳定保留决策。 */
+public final class BacklogSamplingDecision {
+    private final boolean degraded;
+    private final boolean keepNormal;
+    private final boolean keepNonNormal;
+    private final boolean keepRaw;
+    private final long lagMs;
+    private final LocalDateTime sourceHour;
+    private final String policyVersion;
+    private final int normalRetainBps;
+    private final int nonNormalRetainBps;
+
+    BacklogSamplingDecision(boolean degraded, boolean keepNormal, boolean keepNonNormal,
+                            boolean keepRaw, long lagMs, LocalDateTime sourceHour,
+                            String policyVersion, int normalRetainBps, int nonNormalRetainBps) {
+        this.degraded = degraded;
+        this.keepNormal = keepNormal;
+        this.keepNonNormal = keepNonNormal;
+        this.keepRaw = keepRaw;
+        this.lagMs = lagMs;
+        this.sourceHour = sourceHour;
+        this.policyVersion = policyVersion;
+        this.normalRetainBps = normalRetainBps;
+        this.nonNormalRetainBps = nonNormalRetainBps;
+    }
+
+    public boolean isDegraded() { return degraded; }
+    public boolean isKeepNormal() { return keepNormal; }
+    public boolean isKeepNonNormal() { return keepNonNormal; }
+    public boolean isKeepRaw() { return keepRaw; }
+    public long getLagMs() { return lagMs; }
+    public LocalDateTime getSourceHour() { return sourceHour; }
+    public String getPolicyVersion() { return policyVersion; }
+    public int getNormalRetainBps() { return normalRetainBps; }
+    public int getNonNormalRetainBps() { return nonNormalRetainBps; }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java

@@ -11,6 +11,9 @@ import com.common.service.DmStandardizedBatchWriter;
 import com.common.service.IngressStatsProjectionWriter;
 import com.common.service.DirectStatsProjectionWriter;
 import com.common.service.SyslogNonNormalMessageService;
+import com.common.service.SyslogNormalDataService;
+import com.common.service.BacklogRetentionDelta;
+import com.common.service.BacklogSamplingAuditService;
 import com.common.entity.IngressStatsEvent;
 import com.common.entity.SyslogNonNormalMessage;
 import com.influxdb.client.write.Point;
@@ -32,6 +35,7 @@ import org.springframework.stereotype.Component;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -108,6 +112,9 @@ public class SysLogProcessor {
     @Autowired private PointTimeResolver pointTimeResolver;
     @Autowired private RawStorageProperties rawStorageProperties;
     @Autowired private RawInfluxMetrics rawInfluxMetrics;
+    @Autowired private BacklogSamplingDecider backlogSamplingDecider;
+    @Autowired private BacklogSamplingAuditService backlogSamplingAuditService;
+    @Autowired private SyslogNormalDataService syslogNormalDataService;
 
     private final AtomicInteger totalProcessed = new AtomicInteger(0);
     private final AtomicLong nonNormalAsyncCompletedTotal = new AtomicLong();
@@ -203,7 +210,7 @@ public class SysLogProcessor {
     }
 
     /**
-     * 统一批处理边界：原文批量写入 InfluxDB 后，有界并行执行标准化。
+     * 统一批处理边界：先有界并行完成规则计算和积压采样决策，再持久化被保留的数据。
      * 失败时优先进入 Retry/DLT；最终兜底按连续性策略记录并推进 offset。非标消息按现场临时策略
      * 异步落库，其写库结果不参与 Kafka ACK。
      */
@@ -258,32 +265,44 @@ public class SysLogProcessor {
         // 入口状态是 Kafka 已可靠接收的事实。批内按设备/探针聚合，失败时禁止 ACK；重投只重复 MIN/MAX。
         int ingressGroups = ingressStatsProjectionWriter.writeBatch(ingressEvents(preparedMessages));
 
-        long influxStartedAt = System.currentTimeMillis();
+        // 必须先完成规则计算，才能确保告警全量保留，并让未保留消息同时跳过 InfluxDB 和达梦。
-        List<PreparedMessage> persistedMessages;
+        long normalizeStartedAt = System.currentTimeMillis();
-        try {
+        List<PreparedNormalization> computed = computePreparedNormalizations(
-            persistedMessages = persistRawBatch(preparedMessages, metrics);
+                preparedMessages, metrics, normalizationExecutor);
-        } catch (RuntimeException rawStorageFailure) {
+        SamplingPlan samplingPlan = buildSamplingPlan(computed);
-            if (!failureContinuityEnabled) throw rawStorageFailure;
+        if (!samplingPlan.auditDeltas.isEmpty()) {
-            routeContinuityPrepared(preparedMessages, rawStorageFailure, "RAW_STORAGE");
+            boolean audited = backlogSamplingAuditService.recordBeforeDropping(samplingPlan.auditDeltas);
-            rawInfluxMetrics.rawWriteBypassed(preparedMessages.size());
+            if (!audited) {
-            metrics.rawStorageBypassed.addAndGet(preparedMessages.size());
+                samplingPlan = allRetainedPlan(computed);
-            // 无论 DLT 是否可用，都继续使用内存中的完整内容执行规则和达梦写入。
+                log.error("[BACKLOG_SAMPLING_FALLBACK] 采样审计不可用，当前批次已回退全量保留: "
-            persistedMessages = preparedMessages;
+                        + "lane={}, records={}", lane, records.size());
-            log.error("[SYSLOG_CONTINUITY] InfluxDB 原文写入失败，已尝试记录 DLT 并继续标准化: "
+            } else {
-                            + "lane={}, records={}, errorType={}",
+                metrics.recordSampling(samplingPlan);
-                    lane, preparedMessages.size(), rawStorageFailure.getClass().getName(),
+            }
-                    rawStorageFailure);
         }
+        long normalizeMs = System.currentTimeMillis() - normalizeStartedAt;
+
+        // 告警是最高优先级：先独立提交告警表，不能让 InfluxDB 或普通表的慢写阻塞告警入库。
+        StandardizedPersistencePlan standardizedPlan = prepareStandardizedPersistence(
+                samplingPlan.normalizations, metrics);
+        long alarmWriteStartedAt = System.nanoTime();
+        DmStandardizedBatchWriter.WriteReport<PreparedMessage> alarmReport =
+                standardizedBatchWriter.write(standardizedPlan.alarmWriteRequests);
+        long alarmWriteMs = elapsedMillis(alarmWriteStartedAt);
+
+        long influxStartedAt = System.currentTimeMillis();
+        List<PreparedMessage> persistedMessages = persistRawWithoutBlockingMainFlow(
+                samplingPlan.rawMessages, metrics, lane);
         long influxMs = System.currentTimeMillis() - influxStartedAt;
 
-        long normalizeStartedAt = System.currentTimeMillis();
+        persistRemainingStandardizedData(standardizedPlan, alarmReport, alarmWriteMs, metrics);
-        normalizePreparedMessages(persistedMessages, metrics, normalizationExecutor);
-        long normalizeMs = System.currentTimeMillis() - normalizeStartedAt;
 
         totalProcessed.addAndGet(records.size());
         log.info("Syslog批次完成: lane={}, topicPartitions={}, records={}, prepared={}, persisted={}, "
                         + "success={}, filtered={}, retry={}, dlt={}, normalizedRows={}, dmSuccessfulRows={}, "
-                        + "dmFailedMessages={}, ingressGroups={}, rawStorageBypassed={}, prepareMs={}, influxMs={}, normalizeMs={}, "
+                        + "dmFailedMessages={}, ingressGroups={}, rawStorageBypassed={}, backlogDegraded={}, "
+                        + "rawSampledOut={}, normalRowsSampledOut={}, nonNormalRowsSampledOut={}, "
+                        + "prepareMs={}, influxMs={}, normalizeMs={}, "
                         + "ruleComputeMs={}, ruleSnapshotMs={}, messageParseMs={}, fieldMappingMs={}, "
                         + "filterCompleteMs={}, enrichmentMs={}, ruleAssembleMs={}, compiledRuleHit={}, "
                         + "compiledRuleMiss={}, messageParseHit={}, messageParseMiss={}, deviceUnique={}, "
@@ -294,6 +313,8 @@ public class SysLogProcessor {
                 metrics.success.get(), metrics.filtered.get(), metrics.retry.get(), metrics.dlt.get(),
                 metrics.normalizedRows.get(), metrics.dmSuccessfulRows.get(), metrics.dmFailedMessages.get(), ingressGroups,
                 metrics.rawStorageBypassed.get(),
+                metrics.backlogDegraded.get(), metrics.rawSampledOut.get(),
+                metrics.normalRowsSampledOut.get(), metrics.nonNormalRowsSampledOut.get(),
                 prepareMs, influxMs, normalizeMs, metrics.ruleComputeMs(), metrics.ruleSnapshotMs(),
                 metrics.messageParseMs(), metrics.fieldMappingMs(), metrics.filterCompleteMs(),
                 metrics.enrichmentMs(), metrics.ruleAssembleMs(), metrics.compiledRuleHits(),
@@ -599,7 +620,27 @@ public class SysLogProcessor {
      * 批量请求如果因单条数据格式被 InfluxDB 拒绝，降级为逐条写入以定位坏消息；
      * 连接、超时或服务端故障会上抛到连续性边界，由其尝试 DLT 后继续标准化。
      */
-    private List<PreparedMessage> persistRawBatch(List<PreparedMessage> messages, BatchMetrics metrics) {
+    /**
+     * InfluxDB 是非关键旁路：任何原文写入异常都不能越过本方法阻塞 Kafka 主消费。
+     * DLT 仅用于尽力保留重放能力；DLT 也失败时 routeContinuityPrepared 会批量记录后返回。
+     */
+    List<PreparedMessage> persistRawWithoutBlockingMainFlow(
+            List<PreparedMessage> messages, BatchMetrics metrics, String lane) {
+        try {
+            return persistRawBatch(messages, metrics);
+        } catch (RuntimeException rawStorageFailure) {
+            routeContinuityPrepared(messages, rawStorageFailure, "RAW_STORAGE");
+            rawInfluxMetrics.rawWriteBypassed(messages.size());
+            metrics.rawStorageBypassed.addAndGet(messages.size());
+            // 告警已经优先写入；无论 DLT 是否可用，都继续写普通表和非标表并推进 offset。
+            log.error("[SYSLOG_CONTINUITY] InfluxDB 原文写入失败，已尝试记录 DLT，"
+                            + "当前批次继续其余标准化写入: lane={}, records={}, errorType={}",
+                    lane, messages.size(), rawStorageFailure.getClass().getName(), rawStorageFailure);
+            return messages;
+        }
+    }
+
+    List<PreparedMessage> persistRawBatch(List<PreparedMessage> messages, BatchMetrics metrics) {
         if (messages.isEmpty()) {
             return messages;
         }
@@ -632,7 +673,9 @@ public class SysLogProcessor {
                     if (message.v2) rawInfluxMetrics.v2Points(1L);
                     persisted.add(message);
                 } catch (Exception singleFailure) {
-                    metrics.record(routeFailure(message.record, message.context, singleFailure));
+                    // 原文格式问题进入 DLT，但不能因此丢掉已经识别出的告警/标准化结果。
+                    routeFailure(message.record, message.context, singleFailure);
+                    persisted.add(message);
                 }
             }
             return persisted;
@@ -705,9 +748,14 @@ public class SysLogProcessor {
 
     private void normalizePreparedMessages(List<PreparedMessage> messages, BatchMetrics metrics,
                                            Executor normalizationExecutor) {
-        if (messages.isEmpty()) {
+        List<PreparedNormalization> computed = computePreparedNormalizations(
-            return;
+                messages, metrics, normalizationExecutor);
-        }
+        persistComputedNormalizations(allRetainedPlan(computed).normalizations, metrics);
+    }
+
+    private List<PreparedNormalization> computePreparedNormalizations(
+            List<PreparedMessage> messages, BatchMetrics metrics, Executor normalizationExecutor) {
+        if (messages.isEmpty()) return Collections.emptyList();
         long snapshotStartedAt = System.nanoTime();
         Set<Long> deviceIds = new HashSet<>();
         for (PreparedMessage message : messages) {
@@ -736,55 +784,11 @@ public class SysLogProcessor {
                     .get(processTimeoutMs, TimeUnit.MILLISECONDS);
             metrics.recordRuleComputeMs(elapsedMillis(ruleComputeStartedAt));
             metrics.recordRuleExecution(executionMetrics, enrichmentContext);
-
+            List<PreparedNormalization> computed = new ArrayList<>(futures.size());
-            List<DmStandardizedBatchWriter.WriteRequest<PreparedMessage>> writeRequests = new ArrayList<>();
-            List<SyslogNonNormalMessage> nonNormalMessages = new ArrayList<>();
-            Set<PreparedMessage> pendingDatabaseMessages = new HashSet<>();
-            int filteredMessages = 0;
             for (CompletableFuture<PreparedNormalization> future : futures) {
-                PreparedNormalization prepared = future.join();
+                computed.add(future.join());
-                if (prepared.terminalState != null) {
-                    metrics.record(prepared.terminalState);
-                    continue;
-                }
-                nonNormalMessages.addAll(prepared.result.getNonNormalMessages());
-                if (prepared.result.getOutcome() == LogNormalProcessor.ProcessingOutcome.FILTERED) {
-                    filteredMessages++;
-                    continue;
-                }
-
-                pendingDatabaseMessages.add(prepared.message);
-                for (Map<String, Object> row : prepared.result.getRows()) {
-                    writeRequests.add(new DmStandardizedBatchWriter.WriteRequest<>(prepared.message, row));
-                }
             }
-
+            return computed;
-            // 批写器只返回“可归因到具体消息”的行级失败；连接、超时、事务结果未知会直接抛出，
-            // 从而让本方法退出并保留整个 Kafka 批次 offset。
-            long dmWriteStartedAt = System.nanoTime();
-            DmStandardizedBatchWriter.WriteReport<PreparedMessage> writeReport =
-                    standardizedBatchWriter.write(writeRequests);
-            metrics.recordDmWriteMs(elapsedMillis(dmWriteStartedAt));
-            Set<PreparedMessage> failedMessages = writeReport.getFailures().keySet();
-            metrics.recordDmWrite(writeRequests.size(), writeReport.getSuccessfulRows(), failedMessages.size());
-
-            for (Map.Entry<PreparedMessage, RuntimeException> failure : writeReport.getFailures().entrySet()) {
-                PreparedMessage message = failure.getKey();
-                metrics.record(routeFailure(message.record, message.context, failure.getValue()));
-            }
-            for (PreparedMessage message : pendingDatabaseMessages) {
-                if (!failedMessages.contains(message)) {
-                    metrics.record(SyslogFailureRouter.TerminalState.SUCCESS);
-                }
-            }
-            for (int i = 0; i < filteredMessages; i++) {
-                metrics.record(SyslogFailureRouter.TerminalState.FILTERED);
-            }
-
-            // 正常链路先达到可靠终态，再把非标日志交给独立单线程，避免正常写入失败重试时
-            // 提前制造非标重复数据。这里只提交任务，不等待结果，非标写入不参与 Kafka ACK。
-            // 队列满、写库失败或进程异常退出都可能丢非标数据，这是现场选择的性能优先策略。
-            submitNonNormalMessages(nonNormalMessages, metrics);
         } catch (TimeoutException timeout) {
             cancelOutstanding(futures);
             throw new IllegalStateException("标准化批次处理超时，禁止提交 offset", timeout);
@@ -801,6 +805,143 @@ public class SysLogProcessor {
         }
     }
 
+    /** 根据已审计的保留决策写两张标准化表和非标表。 */
+    private void persistComputedNormalizations(List<RetentionPreparedNormalization> normalizations,
+                                               BatchMetrics metrics) {
+        StandardizedPersistencePlan plan = prepareStandardizedPersistence(normalizations, metrics);
+        long alarmWriteStartedAt = System.nanoTime();
+        DmStandardizedBatchWriter.WriteReport<PreparedMessage> alarmReport =
+                standardizedBatchWriter.write(plan.alarmWriteRequests);
+        long alarmWriteMs = elapsedMillis(alarmWriteStartedAt);
+        persistRemainingStandardizedData(plan, alarmReport, alarmWriteMs, metrics);
+    }
+
+    private StandardizedPersistencePlan prepareStandardizedPersistence(
+            List<RetentionPreparedNormalization> normalizations, BatchMetrics metrics) {
+        List<DmStandardizedBatchWriter.WriteRequest<PreparedMessage>> alarmWriteRequests = new ArrayList<>();
+        List<DmStandardizedBatchWriter.WriteRequest<PreparedMessage>> normalWriteRequests = new ArrayList<>();
+        List<SyslogNonNormalMessage> nonNormalMessages = new ArrayList<>();
+        Set<PreparedMessage> pendingDatabaseMessages = new HashSet<>();
+        int filteredMessages = 0;
+
+        for (RetentionPreparedNormalization retained : normalizations) {
+            PreparedNormalization prepared = retained.prepared;
+            if (prepared.terminalState != null) {
+                metrics.record(prepared.terminalState);
+                continue;
+            }
+
+            if (retained.keepNonNormal) {
+                nonNormalMessages.addAll(prepared.result.getNonNormalMessages());
+            }
+            int retainedStandardRows = 0;
+            for (Map<String, Object> row : prepared.result.getRows()) {
+                boolean alarm = syslogNormalDataService.isAlarmData(row);
+                if (alarm || retained.keepNormal) {
+                    DmStandardizedBatchWriter.WriteRequest<PreparedMessage> request =
+                            new DmStandardizedBatchWriter.WriteRequest<>(prepared.message, row);
+                    if (alarm) alarmWriteRequests.add(request);
+                    else normalWriteRequests.add(request);
+                    retainedStandardRows++;
+                }
+            }
+            if (retainedStandardRows > 0) {
+                pendingDatabaseMessages.add(prepared.message);
+            } else {
+                filteredMessages++;
+            }
+        }
+        return new StandardizedPersistencePlan(alarmWriteRequests, normalWriteRequests,
+                nonNormalMessages, pendingDatabaseMessages, filteredMessages);
+    }
+
+    private void persistRemainingStandardizedData(
+            StandardizedPersistencePlan plan,
+            DmStandardizedBatchWriter.WriteReport<PreparedMessage> alarmReport,
+            long alarmWriteMs, BatchMetrics metrics) {
+        long normalWriteStartedAt = System.nanoTime();
+        DmStandardizedBatchWriter.WriteReport<PreparedMessage> normalReport =
+                standardizedBatchWriter.write(plan.normalWriteRequests);
+        metrics.recordDmWriteMs(alarmWriteMs + elapsedMillis(normalWriteStartedAt));
+        Map<PreparedMessage, RuntimeException> failures = new LinkedHashMap<>();
+        failures.putAll(alarmReport.getFailures());
+        for (Map.Entry<PreparedMessage, RuntimeException> failure : normalReport.getFailures().entrySet()) {
+            failures.putIfAbsent(failure.getKey(), failure.getValue());
+        }
+        Set<PreparedMessage> failedMessages = failures.keySet();
+        int requestedRows = plan.alarmWriteRequests.size() + plan.normalWriteRequests.size();
+        int successfulRows = alarmReport.getSuccessfulRows() + normalReport.getSuccessfulRows();
+        metrics.recordDmWrite(requestedRows, successfulRows, failedMessages.size());
+
+        for (Map.Entry<PreparedMessage, RuntimeException> failure : failures.entrySet()) {
+            PreparedMessage message = failure.getKey();
+            metrics.record(routeFailure(message.record, message.context, failure.getValue()));
+        }
+        for (PreparedMessage message : plan.pendingDatabaseMessages) {
+            if (!failedMessages.contains(message)) {
+                metrics.record(SyslogFailureRouter.TerminalState.SUCCESS);
+            }
+        }
+        for (int i = 0; i < plan.filteredMessages; i++) {
+            metrics.record(SyslogFailureRouter.TerminalState.FILTERED);
+        }
+
+        // 非标日志仍使用现有独立线程；采样决策已经在进入队列前完成。
+        submitNonNormalMessages(plan.nonNormalMessages, metrics);
+    }
+
+    SamplingPlan buildSamplingPlan(List<PreparedNormalization> computed) {
+        List<RetentionPreparedNormalization> retained = new ArrayList<>(computed.size());
+        List<PreparedMessage> rawMessages = new ArrayList<>(computed.size());
+        List<BacklogRetentionDelta> auditDeltas = new ArrayList<>();
+        for (PreparedNormalization prepared : computed) {
+            if (prepared.terminalState != null) {
+                retained.add(new RetentionPreparedNormalization(prepared, false, false));
+                // 规则异常时无法证明该消息不是告警，保留已构造的 InfluxDB 原文点。
+                rawMessages.add(prepared.message);
+                continue;
+            }
+            long alarmRows = 0L;
+            long normalRows = 0L;
+            for (Map<String, Object> row : prepared.result.getRows()) {
+                if (syslogNormalDataService.isAlarmData(row)) alarmRows++;
+                else normalRows++;
+            }
+            long nonNormalRows = prepared.result.getNonNormalMessages().size();
+            BacklogSamplingDecision decision = backlogSamplingDecider.decide(
+                    prepared.message.eventId, prepared.message.receivedAt,
+                    alarmRows > 0L, normalRows > 0L, nonNormalRows > 0L);
+            retained.add(new RetentionPreparedNormalization(
+                    prepared, decision.isKeepNormal(), decision.isKeepNonNormal()));
+            if (decision.isKeepRaw()) rawMessages.add(prepared.message);
+            if (decision.isDegraded()) {
+                auditDeltas.add(new BacklogRetentionDelta(
+                        decision.getSourceHour(), prepared.message.record.topic(),
+                        prepared.message.record.partition(), decision.getPolicyVersion(),
+                        decision.getNormalRetainBps(), decision.getNonNormalRetainBps(),
+                        1L, alarmRows, alarmRows,
+                        normalRows, decision.isKeepNormal() ? normalRows : 0L,
+                        decision.isKeepNormal() ? 0L : normalRows,
+                        nonNormalRows, decision.isKeepNonNormal() ? nonNormalRows : 0L,
+                        decision.isKeepNonNormal() ? 0L : nonNormalRows,
+                        1L, decision.isKeepRaw() ? 1L : 0L,
+                        decision.isKeepRaw() ? 0L : 1L,
+                        decision.getLagMs(), decision.getLagMs()));
+            }
+        }
+        return new SamplingPlan(retained, rawMessages, auditDeltas);
+    }
+
+    private SamplingPlan allRetainedPlan(List<PreparedNormalization> computed) {
+        List<RetentionPreparedNormalization> retained = new ArrayList<>(computed.size());
+        List<PreparedMessage> rawMessages = new ArrayList<>(computed.size());
+        for (PreparedNormalization prepared : computed) {
+            retained.add(new RetentionPreparedNormalization(prepared, true, true));
+            rawMessages.add(prepared.message);
+        }
+        return new SamplingPlan(retained, rawMessages, Collections.emptyList());
+    }
+
     /**
      * 把非标批次交给独立线程后立即返回。任何提交或执行失败只记录，不传播到 Kafka ACK 边界。
      */
@@ -984,20 +1125,71 @@ public class SysLogProcessor {
      * 并行计算阶段的结果封装。result 与 terminalState 二选一：前者等待达梦批写，后者表示该消息
      * 已经被 Retry/DLT 可靠接管。使用显式对象可以避免在工作线程里提前修改批次 ACK 状态。
      */
-    private static final class PreparedNormalization {
+    static final class PreparedNormalization {
         private final PreparedMessage message;
         private final LogNormalProcessor.NormalizationResult result;
         private final SyslogFailureRouter.TerminalState terminalState;
 
-        private PreparedNormalization(PreparedMessage message,
+        PreparedNormalization(PreparedMessage message,
-                                      LogNormalProcessor.NormalizationResult result,
+                              LogNormalProcessor.NormalizationResult result,
-                                      SyslogFailureRouter.TerminalState terminalState) {
+                              SyslogFailureRouter.TerminalState terminalState) {
             this.message = message;
             this.result = result;
             this.terminalState = terminalState;
         }
     }
 
+    static final class RetentionPreparedNormalization {
+        private final PreparedNormalization prepared;
+        private final boolean keepNormal;
+        private final boolean keepNonNormal;
+
+        private RetentionPreparedNormalization(PreparedNormalization prepared,
+                                                boolean keepNormal, boolean keepNonNormal) {
+            this.prepared = prepared;
+            this.keepNormal = keepNormal;
+            this.keepNonNormal = keepNonNormal;
+        }
+    }
+
+    static final class SamplingPlan {
+        private final List<RetentionPreparedNormalization> normalizations;
+        private final List<PreparedMessage> rawMessages;
+        private final List<BacklogRetentionDelta> auditDeltas;
+
+        private SamplingPlan(List<RetentionPreparedNormalization> normalizations,
+                             List<PreparedMessage> rawMessages,
+                             List<BacklogRetentionDelta> auditDeltas) {
+            this.normalizations = normalizations;
+            this.rawMessages = rawMessages;
+            this.auditDeltas = auditDeltas;
+        }
+
+        int rawMessageCount() { return rawMessages.size(); }
+        int auditMessageCount() { return auditDeltas.size(); }
+        List<BacklogRetentionDelta> auditDeltas() { return auditDeltas; }
+    }
+
+    private static final class StandardizedPersistencePlan {
+        private final List<DmStandardizedBatchWriter.WriteRequest<PreparedMessage>> alarmWriteRequests;
+        private final List<DmStandardizedBatchWriter.WriteRequest<PreparedMessage>> normalWriteRequests;
+        private final List<SyslogNonNormalMessage> nonNormalMessages;
+        private final Set<PreparedMessage> pendingDatabaseMessages;
+        private final int filteredMessages;
+
+        private StandardizedPersistencePlan(
+                List<DmStandardizedBatchWriter.WriteRequest<PreparedMessage>> alarmWriteRequests,
+                List<DmStandardizedBatchWriter.WriteRequest<PreparedMessage>> normalWriteRequests,
+                List<SyslogNonNormalMessage> nonNormalMessages,
+                Set<PreparedMessage> pendingDatabaseMessages, int filteredMessages) {
+            this.alarmWriteRequests = alarmWriteRequests;
+            this.normalWriteRequests = normalWriteRequests;
+            this.nonNormalMessages = nonNormalMessages;
+            this.pendingDatabaseMessages = pendingDatabaseMessages;
+            this.filteredMessages = filteredMessages;
+        }
+    }
+
     /** 单批终态计数，仅用于聚合日志，不参与 ACK 决策。 */
     static final class BatchMetrics {
         private final AtomicInteger success = new AtomicInteger();
@@ -1008,6 +1200,10 @@ public class SysLogProcessor {
         private final AtomicInteger dmSuccessfulRows = new AtomicInteger();
         private final AtomicInteger dmFailedMessages = new AtomicInteger();
         private final AtomicInteger rawStorageBypassed = new AtomicInteger();
+        private final AtomicInteger backlogDegraded = new AtomicInteger();
+        private final AtomicInteger rawSampledOut = new AtomicInteger();
+        private final AtomicInteger normalRowsSampledOut = new AtomicInteger();
+        private final AtomicInteger nonNormalRowsSampledOut = new AtomicInteger();
         private final AtomicLong ruleComputeMs = new AtomicLong();
         private final AtomicLong ruleSnapshotMs = new AtomicLong();
         private final AtomicLong dmWriteMs = new AtomicLong();
@@ -1091,6 +1287,15 @@ public class SysLogProcessor {
             nonNormalRejected.addAndGet(Math.max(0, rejected));
         }
 
+        void recordSampling(SamplingPlan plan) {
+            for (BacklogRetentionDelta delta : plan.auditDeltas) {
+                backlogDegraded.addAndGet((int) delta.getMessageTotal());
+                rawSampledOut.addAndGet((int) delta.getRawDropped());
+                normalRowsSampledOut.addAndGet((int) delta.getNormalRowsDropped());
+                nonNormalRowsSampledOut.addAndGet((int) delta.getNonNormalRowsDropped());
+            }
+        }
+
         /** 仅用于批次汇总观测，不参与 ACK 判断；ACK 仍由每条消息的可靠终态决定。 */
         void recordDmWrite(int rows, int successfulRows, int failedMessages) {
             this.normalizedRows.addAndGet(rows);

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java

@@ -254,8 +254,8 @@ public class SyslogFailureRouter {
             return FailureCategory.TRANSIENT;
         }
         Throwable root = rootCause(failure);
-        // InfluxDB 400 数据格式错误只影响当前消息，可直接隔离；
+        // InfluxDB 400 数据格式错误只影响当前消息，可直接隔离；连接、超时、5xx 属于系统性依赖故障。
-        // 连接、超时、5xx 等 InfluxDB 故障属于系统性依赖故障，必须保留原 offset，避免 DLT 风暴。
+        // 活动原文批写链路会在外层非阻塞旁路边界覆盖其 offset 行为，不能因此阻塞主消费。
         if (root instanceof BadRequestException) {
             return FailureCategory.PERMANENT_DATA;
         }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/BacklogRetentionAuditMapper.java

@@ -0,0 +1,9 @@
+package com.common.mapper;
+
+import com.common.service.BacklogRetentionDelta;
+import org.apache.ibatis.annotations.Mapper;
+
+@Mapper
+public interface BacklogRetentionAuditMapper {
+    int upsert(BacklogRetentionDelta delta);
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogRetentionDelta.java

@@ -0,0 +1,118 @@
+package com.common.service;
+
+import java.time.LocalDateTime;
+import java.util.Objects;
+
+/** 一批积压消息按来源小时、Topic、Partition 聚合后的采样审计增量。 */
+public final class BacklogRetentionDelta {
+    private final LocalDateTime sourceHour;
+    private final String sourceTopic;
+    private final int kafkaPartition;
+    private final String policyVersion;
+    private final int normalRetainBps;
+    private final int nonNormalRetainBps;
+    private long messageTotal;
+    private long alarmRowsTotal;
+    private long alarmRowsSaved;
+    private long normalRowsTotal;
+    private long normalRowsSaved;
+    private long normalRowsDropped;
+    private long nonNormalRowsTotal;
+    private long nonNormalRowsSaved;
+    private long nonNormalRowsDropped;
+    private long rawTotal;
+    private long rawSaved;
+    private long rawDropped;
+    private long minLagMs;
+    private long maxLagMs;
+
+    public BacklogRetentionDelta(LocalDateTime sourceHour, String sourceTopic, int kafkaPartition,
+                                 String policyVersion, int normalRetainBps, int nonNormalRetainBps,
+                                 long messageTotal, long alarmRowsTotal, long alarmRowsSaved,
+                                 long normalRowsTotal, long normalRowsSaved, long normalRowsDropped,
+                                 long nonNormalRowsTotal, long nonNormalRowsSaved,
+                                 long nonNormalRowsDropped, long rawTotal, long rawSaved,
+                                 long rawDropped, long minLagMs, long maxLagMs) {
+        this.sourceHour = Objects.requireNonNull(sourceHour, "sourceHour");
+        this.sourceTopic = Objects.requireNonNull(sourceTopic, "sourceTopic");
+        this.kafkaPartition = kafkaPartition;
+        this.policyVersion = Objects.requireNonNull(policyVersion, "policyVersion");
+        this.normalRetainBps = normalRetainBps;
+        this.nonNormalRetainBps = nonNormalRetainBps;
+        this.messageTotal = messageTotal;
+        this.alarmRowsTotal = alarmRowsTotal;
+        this.alarmRowsSaved = alarmRowsSaved;
+        this.normalRowsTotal = normalRowsTotal;
+        this.normalRowsSaved = normalRowsSaved;
+        this.normalRowsDropped = normalRowsDropped;
+        this.nonNormalRowsTotal = nonNormalRowsTotal;
+        this.nonNormalRowsSaved = nonNormalRowsSaved;
+        this.nonNormalRowsDropped = nonNormalRowsDropped;
+        this.rawTotal = rawTotal;
+        this.rawSaved = rawSaved;
+        this.rawDropped = rawDropped;
+        this.minLagMs = minLagMs;
+        this.maxLagMs = maxLagMs;
+    }
+
+    public void merge(BacklogRetentionDelta other) {
+        if (!sameKey(other)) throw new IllegalArgumentException("不能合并不同维度的采样审计增量");
+        messageTotal += other.messageTotal;
+        alarmRowsTotal += other.alarmRowsTotal;
+        alarmRowsSaved += other.alarmRowsSaved;
+        normalRowsTotal += other.normalRowsTotal;
+        normalRowsSaved += other.normalRowsSaved;
+        normalRowsDropped += other.normalRowsDropped;
+        nonNormalRowsTotal += other.nonNormalRowsTotal;
+        nonNormalRowsSaved += other.nonNormalRowsSaved;
+        nonNormalRowsDropped += other.nonNormalRowsDropped;
+        rawTotal += other.rawTotal;
+        rawSaved += other.rawSaved;
+        rawDropped += other.rawDropped;
+        minLagMs = Math.min(minLagMs, other.minLagMs);
+        maxLagMs = Math.max(maxLagMs, other.maxLagMs);
+    }
+
+    public BacklogRetentionDelta copy() {
+        return new BacklogRetentionDelta(sourceHour, sourceTopic, kafkaPartition, policyVersion,
+                normalRetainBps, nonNormalRetainBps, messageTotal, alarmRowsTotal,
+                alarmRowsSaved, normalRowsTotal, normalRowsSaved, normalRowsDropped,
+                nonNormalRowsTotal, nonNormalRowsSaved, nonNormalRowsDropped,
+                rawTotal, rawSaved, rawDropped, minLagMs, maxLagMs);
+    }
+
+    private boolean sameKey(BacklogRetentionDelta other) {
+        return other != null && kafkaPartition == other.kafkaPartition
+                && normalRetainBps == other.normalRetainBps
+                && nonNormalRetainBps == other.nonNormalRetainBps
+                && sourceHour.equals(other.sourceHour)
+                && sourceTopic.equals(other.sourceTopic)
+                && policyVersion.equals(other.policyVersion);
+    }
+
+    public String aggregationKey() {
+        return sourceHour + "\u0000" + sourceTopic + "\u0000" + kafkaPartition + "\u0000"
+                + policyVersion + "\u0000" + normalRetainBps + "\u0000" + nonNormalRetainBps;
+    }
+
+    public LocalDateTime getSourceHour() { return sourceHour; }
+    public String getSourceTopic() { return sourceTopic; }
+    public int getKafkaPartition() { return kafkaPartition; }
+    public String getPolicyVersion() { return policyVersion; }
+    public int getNormalRetainBps() { return normalRetainBps; }
+    public int getNonNormalRetainBps() { return nonNormalRetainBps; }
+    public long getMessageTotal() { return messageTotal; }
+    public long getAlarmRowsTotal() { return alarmRowsTotal; }
+    public long getAlarmRowsSaved() { return alarmRowsSaved; }
+    public long getNormalRowsTotal() { return normalRowsTotal; }
+    public long getNormalRowsSaved() { return normalRowsSaved; }
+    public long getNormalRowsDropped() { return normalRowsDropped; }
+    public long getNonNormalRowsTotal() { return nonNormalRowsTotal; }
+    public long getNonNormalRowsSaved() { return nonNormalRowsSaved; }
+    public long getNonNormalRowsDropped() { return nonNormalRowsDropped; }
+    public long getRawTotal() { return rawTotal; }
+    public long getRawSaved() { return rawSaved; }
+    public long getRawDropped() { return rawDropped; }
+    public long getMinLagMs() { return minLagMs; }
+    public long getMaxLagMs() { return maxLagMs; }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogSamplingAuditService.java

@@ -0,0 +1,81 @@
+package com.common.service;
+
+import com.common.mapper.BacklogRetentionAuditMapper;
+import com.config.BacklogSamplingProperties;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** 先提交小时审计，再允许调用方执行有损采样；审计失败时调用方必须回退全量保留。 */
+@Slf4j
+@Service
+public class BacklogSamplingAuditService {
+    private final BacklogRetentionAuditMapper mapper;
+    private final BacklogSamplingProperties properties;
+    private final BacklogSamplingMetrics metrics;
+    private final PlatformTransactionManager transactionManager;
+    private volatile long retryAfterMs;
+
+    public BacklogSamplingAuditService(BacklogRetentionAuditMapper mapper,
+                                       BacklogSamplingProperties properties,
+                                       BacklogSamplingMetrics metrics,
+                                       PlatformTransactionManager transactionManager) {
+        this.mapper = mapper;
+        this.properties = properties;
+        this.metrics = metrics;
+        this.transactionManager = transactionManager;
+    }
+
+    /** 返回 false 表示不能安全采样，调用方应保留当前批次全部数据。 */
+    public boolean recordBeforeDropping(List<BacklogRetentionDelta> deltas) {
+        if (deltas == null || deltas.isEmpty()) return true;
+        long affectedMessages = messageCount(deltas);
+        long now = System.currentTimeMillis();
+        if (now < retryAfterMs) {
+            metrics.auditFailure(affectedMessages);
+            return false;
+        }
+
+        List<BacklogRetentionDelta> aggregated = aggregate(deltas);
+        try {
+            TransactionTemplate transaction = new TransactionTemplate(transactionManager);
+            transaction.setTimeout(Math.max(1, properties.getAuditTimeoutSeconds()));
+            transaction.execute(status -> {
+                for (BacklogRetentionDelta delta : aggregated) mapper.upsert(delta);
+                return null;
+            });
+            retryAfterMs = 0L;
+            metrics.record(aggregated);
+            return true;
+        } catch (RuntimeException failure) {
+            retryAfterMs = System.currentTimeMillis() + Math.max(1L, properties.getAuditRetryBackoffMs());
+            metrics.auditFailure(affectedMessages);
+            log.error("[BACKLOG_SAMPLING_AUDIT_FAILURE] 小时采样审计写入失败，当前批次自动回退全量保留: "
+                            + "messages={}, groups={}, retryBackoffMs={}",
+                    affectedMessages, aggregated.size(), properties.getAuditRetryBackoffMs(), failure);
+            return false;
+        }
+    }
+
+    static List<BacklogRetentionDelta> aggregate(List<BacklogRetentionDelta> deltas) {
+        Map<String, BacklogRetentionDelta> grouped = new LinkedHashMap<>();
+        for (BacklogRetentionDelta delta : deltas) {
+            BacklogRetentionDelta existing = grouped.get(delta.aggregationKey());
+            if (existing == null) grouped.put(delta.aggregationKey(), delta.copy());
+            else existing.merge(delta);
+        }
+        return new ArrayList<>(grouped.values());
+    }
+
+    private long messageCount(List<BacklogRetentionDelta> deltas) {
+        long total = 0L;
+        for (BacklogRetentionDelta delta : deltas) total += delta.getMessageTotal();
+        return total;
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogSamplingMetrics.java

@@ -0,0 +1,56 @@
+package com.common.service;
+
+import org.springframework.stereotype.Component;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.LongAdder;
+
+/** 积压采样热路径指标；表审计失败时仍可从接口和日志发现。 */
+@Component
+public class BacklogSamplingMetrics {
+    private final LongAdder degradedMessages = new LongAdder();
+    private final LongAdder alarmRowsSaved = new LongAdder();
+    private final LongAdder normalRowsSaved = new LongAdder();
+    private final LongAdder normalRowsDropped = new LongAdder();
+    private final LongAdder nonNormalRowsSaved = new LongAdder();
+    private final LongAdder nonNormalRowsDropped = new LongAdder();
+    private final LongAdder rawSaved = new LongAdder();
+    private final LongAdder rawDropped = new LongAdder();
+    private final LongAdder auditWriteFailures = new LongAdder();
+    private final LongAdder auditFallbackMessages = new LongAdder();
+
+    public void record(List<BacklogRetentionDelta> deltas) {
+        for (BacklogRetentionDelta delta : deltas) {
+            degradedMessages.add(delta.getMessageTotal());
+            alarmRowsSaved.add(delta.getAlarmRowsSaved());
+            normalRowsSaved.add(delta.getNormalRowsSaved());
+            normalRowsDropped.add(delta.getNormalRowsDropped());
+            nonNormalRowsSaved.add(delta.getNonNormalRowsSaved());
+            nonNormalRowsDropped.add(delta.getNonNormalRowsDropped());
+            rawSaved.add(delta.getRawSaved());
+            rawDropped.add(delta.getRawDropped());
+        }
+    }
+
+    public void auditFailure(long affectedMessages) {
+        auditWriteFailures.increment();
+        auditFallbackMessages.add(Math.max(0L, affectedMessages));
+    }
+
+    public Map<String, Object> snapshot() {
+        Map<String, Object> values = new LinkedHashMap<>();
+        values.put("degraded_messages_total", degradedMessages.sum());
+        values.put("alarm_rows_saved_total", alarmRowsSaved.sum());
+        values.put("normal_rows_saved_total", normalRowsSaved.sum());
+        values.put("normal_rows_dropped_total", normalRowsDropped.sum());
+        values.put("non_normal_rows_saved_total", nonNormalRowsSaved.sum());
+        values.put("non_normal_rows_dropped_total", nonNormalRowsDropped.sum());
+        values.put("raw_saved_total", rawSaved.sum());
+        values.put("raw_dropped_total", rawDropped.sum());
+        values.put("audit_write_failures_total", auditWriteFailures.sum());
+        values.put("audit_fallback_messages_total", auditFallbackMessages.sum());
+        return values;
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/AppProperties.java

@@ -61,6 +61,14 @@ public class AppProperties {
     @Value("${influxdb.connection.write-timeout:60s}")
     private Duration influxWriteTimeout;
 
+    /** 原文写请求的单阶段上限；避免 InfluxDB 变慢长期占住 Kafka listener。 */
+    @Value("${influxdb.raw.write-request-timeout:5s}")
+    private Duration influxRawWriteRequestTimeout;
+
+    /** 原文写失败后的快速失败时间，期间不再向故障 InfluxDB 发起请求。 */
+    @Value("${influxdb.raw.failure-backoff-ms:30000}")
+    private long influxRawFailureBackoffMs;
+
     // ========== 应用配置 ==========
     @Value("${app.worker.threads:4}")
     private int workerThreads;

haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/BacklogSamplingProperties.java

@@ -0,0 +1,80 @@
+package com.config;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+/** Kafka 积压期间的可审计降级采样配置。 */
+@Component
+public class BacklogSamplingProperties {
+
+    public static final int BASIS_POINTS = 10_000;
+
+    @Value("${syslog.backlog-sampling.enabled:true}")
+    private boolean enabled;
+
+    @Value("${syslog.backlog-sampling.lag-threshold-ms:3600000}")
+    private long lagThresholdMs;
+
+    @Value("${syslog.backlog-sampling.start-lag-ms:900000}")
+    private long startLagMs;
+
+    @Value("${syslog.backlog-sampling.normal-retain-bps:1000}")
+    private int normalRetainBps;
+
+    @Value("${syslog.backlog-sampling.non-normal-retain-bps:2000}")
+    private int nonNormalRetainBps;
+
+    @Value("${syslog.backlog-sampling.degrade-steps:10}")
+    private int degradeSteps;
+
+    @Value("${syslog.backlog-sampling.policy-version:v2}")
+    private String policyVersion;
+
+    @Value("${syslog.backlog-sampling.audit-timeout-seconds:5}")
+    private int auditTimeoutSeconds;
+
+    @Value("${syslog.backlog-sampling.audit-retry-backoff-ms:60000}")
+    private long auditRetryBackoffMs;
+
+    @PostConstruct
+    public void validate() {
+        if (lagThresholdMs < 1L) {
+            throw new IllegalArgumentException("syslog.backlog-sampling.lag-threshold-ms 必须大于 0");
+        }
+        if (startLagMs < 0L || startLagMs >= lagThresholdMs) {
+            throw new IllegalArgumentException("syslog.backlog-sampling.start-lag-ms 必须大于等于 0，且小于 lag-threshold-ms");
+        }
+        validateRate("normal-retain-bps", normalRetainBps);
+        validateRate("non-normal-retain-bps", nonNormalRetainBps);
+        if (degradeSteps < 1 || degradeSteps > 1000) {
+            throw new IllegalArgumentException("syslog.backlog-sampling.degrade-steps 必须在 1~1000 之间");
+        }
+        if (policyVersion == null || policyVersion.trim().isEmpty() || policyVersion.length() > 64) {
+            throw new IllegalArgumentException("syslog.backlog-sampling.policy-version 必须为 1~64 个字符");
+        }
+        if (auditTimeoutSeconds < 1) {
+            throw new IllegalArgumentException("syslog.backlog-sampling.audit-timeout-seconds 必须大于 0");
+        }
+        if (auditRetryBackoffMs < 1L) {
+            throw new IllegalArgumentException("syslog.backlog-sampling.audit-retry-backoff-ms 必须大于 0");
+        }
+    }
+
+    private void validateRate(String name, int value) {
+        if (value < 0 || value > BASIS_POINTS) {
+            throw new IllegalArgumentException("syslog.backlog-sampling." + name + " 必须在 0~10000 之间");
+        }
+    }
+
+    public boolean isEnabled() { return enabled; }
+    public long getLagThresholdMs() { return lagThresholdMs; }
+    public long getStartLagMs() { return startLagMs; }
+    public int getNormalRetainBps() { return normalRetainBps; }
+    public int getNonNormalRetainBps() { return nonNormalRetainBps; }
+    public int getDegradeSteps() { return degradeSteps; }
+    public String getPolicyVersion() { return policyVersion; }
+    public int getAuditTimeoutSeconds() { return auditTimeoutSeconds; }
+    public long getAuditRetryBackoffMs() { return auditRetryBackoffMs; }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/BacklogSamplingController.java

@@ -0,0 +1,37 @@
+package com.controllers;
+
+import com.common.service.BacklogSamplingMetrics;
+import com.config.BacklogSamplingProperties;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+@RestController
+@RequestMapping("/api/syslog/backlog-sampling")
+public class BacklogSamplingController {
+    private final BacklogSamplingProperties properties;
+    private final BacklogSamplingMetrics metrics;
+
+    public BacklogSamplingController(BacklogSamplingProperties properties,
+                                     BacklogSamplingMetrics metrics) {
+        this.properties = properties;
+        this.metrics = metrics;
+    }
+
+    @GetMapping("/status")
+    public Map<String, Object> status() {
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("enabled", properties.isEnabled());
+        result.put("start_lag_ms", properties.getStartLagMs());
+        result.put("lag_threshold_ms", properties.getLagThresholdMs());
+        result.put("normal_retain_bps", properties.getNormalRetainBps());
+        result.put("non_normal_retain_bps", properties.getNonNormalRetainBps());
+        result.put("degrade_steps", properties.getDegradeSteps());
+        result.put("policy_version", properties.getPolicyVersion());
+        result.putAll(metrics.snapshot());
+        return result;
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/InfluxDBClient.java

@@ -5,6 +5,7 @@ import com.influxdb.client.InfluxDBClientOptions;
 import com.influxdb.client.WriteApi;
 import com.influxdb.client.WriteApiBlocking;
 import com.influxdb.client.domain.HealthCheck;
+import com.influxdb.exceptions.BadRequestException;
 import com.influxdb.client.write.Point;
 import com.config.AppProperties;
 import org.slf4j.Logger;
@@ -17,6 +18,8 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import com.influxdb.client.QueryApi;
 import java.time.Instant;
 import java.time.format.DateTimeFormatter;
@@ -41,6 +44,9 @@ public class InfluxDBClient implements AutoCloseable {
 
     private final InfluxDBConfig influxDBConfig;
     private final RawStorageProperties rawStorage;
+    private final long rawWriteFailureBackoffMs;
+    private final AtomicInteger consecutiveRawWriteFailures = new AtomicInteger();
+    private volatile long rawWriteRetryAfterMs;
 
     @Autowired
     public InfluxDBClient(AppProperties appProperties, InfluxDBConfig influxDBConfig,
@@ -48,10 +54,21 @@ public class InfluxDBClient implements AutoCloseable {
         this.influxDBConfig = influxDBConfig;
         this.rawStorage = rawStorage;
 
+        final int rawWriteTimeoutMs = durationMillisAsInt(appProperties.getInfluxRawWriteRequestTimeout());
+        this.rawWriteFailureBackoffMs = Math.max(1L, appProperties.getInfluxRawFailureBackoffMs());
         OkHttpClient.Builder httpClient = new OkHttpClient.Builder()
                 .connectTimeout(appProperties.getInfluxConnectionTimeout())
                 .readTimeout(appProperties.getInfluxReadTimeout())
-                .writeTimeout(appProperties.getInfluxWriteTimeout());
+                .writeTimeout(appProperties.getInfluxWriteTimeout())
+                .addInterceptor(chain -> {
+                    if ("/api/v2/write".equals(chain.request().url().encodedPath())) {
+                        return chain.withConnectTimeout(rawWriteTimeoutMs, TimeUnit.MILLISECONDS)
+                                .withReadTimeout(rawWriteTimeoutMs, TimeUnit.MILLISECONDS)
+                                .withWriteTimeout(rawWriteTimeoutMs, TimeUnit.MILLISECONDS)
+                                .proceed(chain.request());
+                    }
+                    return chain.proceed(chain.request());
+                });
         InfluxDBClientOptions options = InfluxDBClientOptions.builder()
                 .url(appProperties.getInfluxUrl())
                 .authenticateToken(appProperties.getInfluxToken().toCharArray())
@@ -79,8 +96,9 @@ public class InfluxDBClient implements AutoCloseable {
                 logger.warn("InfluxDB health check: {}", health.getMessage());
             }
         } catch (Exception e) {
-            logger.error("Failed to connect to InfluxDB: {}", e.getMessage());
+            // 原文存储是旁路依赖。启动时不可用不能阻止 Kafka 消费和告警入库。
-            throw new RuntimeException("InfluxDB connection failed", e);
+            logger.error("[INFLUX_NON_BLOCKING] InfluxDB 启动健康检查失败，消费者仍将启动: {}",
+                    e.getMessage(), e);
         }
     }
 
@@ -105,9 +123,12 @@ public class InfluxDBClient implements AutoCloseable {
     }
 
     public void writePointBlocking(String targetBucket, Point point) {
+        ensureRawWriteAvailable();
         try {
             writeApiBlocking.writePoint(targetBucket, org, point);
+            recordRawWriteSuccess();
         } catch (Exception e) {
+            recordRawWriteFailure(e);
             logger.error("Failed to write point to InfluxDB (blocking): {}", e.getMessage());
             throw e;
         }
@@ -116,7 +137,7 @@ public class InfluxDBClient implements AutoCloseable {
 
     /**
      * 批量同步写入原始日志点。
-     * 调用返回表示 InfluxDB 已确认整批写入；任何异常都会上抛到 Kafka ACK 边界。
+     * 调用返回表示 InfluxDB 已确认整批写入；任何异常都会上抛给主流程的非阻塞旁路边界。
      */
     public void writePointsBlocking(List<Point> points) {
         writePointsBlocking(bucket, points);
@@ -126,15 +147,44 @@ public class InfluxDBClient implements AutoCloseable {
         if (points == null || points.isEmpty()) {
             return;
         }
+        ensureRawWriteAvailable();
         try {
             writeApiBlocking.writePoints(targetBucket, org, points);
+            recordRawWriteSuccess();
         } catch (Exception e) {
+            recordRawWriteFailure(e);
             logger.error("Failed to write {} points to InfluxDB (blocking batch): {}",
                     points.size(), e.getMessage());
             throw e;
         }
     }
 
+    private void ensureRawWriteAvailable() {
+        long remainingMs = rawWriteRetryAfterMs - System.currentTimeMillis();
+        if (remainingMs > 0L) {
+            throw new IllegalStateException("InfluxDB 原文写入熔断中，剩余 " + remainingMs + "ms");
+        }
+    }
+
+    private void recordRawWriteSuccess() {
+        consecutiveRawWriteFailures.set(0);
+        rawWriteRetryAfterMs = 0L;
+    }
+
+    private void recordRawWriteFailure(Exception failure) {
+        // 400 是当前数据格式问题，外层会逐条隔离，不代表 InfluxDB 整体不可用。
+        if (failure instanceof BadRequestException) return;
+        int failures = consecutiveRawWriteFailures.incrementAndGet();
+        rawWriteRetryAfterMs = System.currentTimeMillis() + rawWriteFailureBackoffMs;
+        logger.error("[INFLUX_WRITE_CIRCUIT_OPEN] InfluxDB 原文写入失败，进入快速失败退避: "
+                        + "failures={}, backoffMs={}", failures, rawWriteFailureBackoffMs);
+    }
+
+    private int durationMillisAsInt(java.time.Duration duration) {
+        long millis = duration == null ? 5000L : Math.max(1L, duration.toMillis());
+        return (int) Math.min(Integer.MAX_VALUE, millis);
+    }
+
     /**
      * 写入原始 Line Protocol（异步）
      */

haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties

@@ -42,6 +42,9 @@ influxdb.raw.write-batch-max-bytes=${INFLUXDB_RAW_WRITE_BATCH_MAX_BYTES:8388608}
 influxdb.connection.timeout=30s
 influxdb.connection.read-timeout=30s
 influxdb.connection.write-timeout=60s
+# 原文写是旁路：单阶段最多等待 5 秒，失败后 30 秒内快速跳过，不能阻塞 Kafka 主消费。
+influxdb.raw.write-request-timeout=${INFLUXDB_RAW_WRITE_REQUEST_TIMEOUT:5s}
+influxdb.raw.failure-backoff-ms=${INFLUXDB_RAW_FAILURE_BACKOFF_MS:30000}
 # Application Configuration
 app.worker.threads=8
 app.max.queue.size=10000
@@ -83,6 +86,21 @@ syslog.raw-listener.concurrency=${SYSLOG_RAW_LISTENER_CONCURRENCY:3}
 syslog.failure-continuity.enabled=${SYSLOG_FAILURE_CONTINUITY_ENABLED:true}
 # DLT 故障时最多快速等待 1 秒；首次失败后整批直接记录并放行。
 syslog.failure-continuity.dlt-publish-timeout-ms=${SYSLOG_FAILURE_CONTINUITY_DLT_PUBLISH_TIMEOUT_MS:1000}
+# Kafka 积压降级采样默认开启；升级应用前必须先创建 syslog_backlog_retention_hourly。
+syslog.backlog-sampling.enabled=${SYSLOG_BACKLOG_SAMPLING_ENABLED:true}
+# 使用 syslog-serve 生成的 receive_time 判断端到端延迟；15 分钟内全量保留。
+syslog.backlog-sampling.start-lag-ms=${SYSLOG_BACKLOG_SAMPLING_START_LAG_MS:900000}
+# 从免采样窗口结束后开始分级，延迟达到 1 小时时降到最低保留率。
+syslog.backlog-sampling.lag-threshold-ms=${SYSLOG_BACKLOG_SAMPLING_LAG_THRESHOLD_MS:3600000}
+# 万分比：这是达到延迟阈值后的最低保留率；识别出的告警永远全量保留。
+syslog.backlog-sampling.normal-retain-bps=${SYSLOG_BACKLOG_SAMPLING_NORMAL_RETAIN_BPS:1000}
+syslog.backlog-sampling.non-normal-retain-bps=${SYSLOG_BACKLOG_SAMPLING_NON_NORMAL_RETAIN_BPS:2000}
+# 默认十级下降：15 分钟到 1 小时的 45 分钟区间，每 4.5 分钟调整一级。
+syslog.backlog-sampling.degrade-steps=${SYSLOG_BACKLOG_SAMPLING_DEGRADE_STEPS:10}
+# 调整保留率时同步修改策略版本，便于小时审计区分不同策略。
+syslog.backlog-sampling.policy-version=${SYSLOG_BACKLOG_SAMPLING_POLICY_VERSION:v2}
+syslog.backlog-sampling.audit-timeout-seconds=${SYSLOG_BACKLOG_SAMPLING_AUDIT_TIMEOUT_SECONDS:5}
+syslog.backlog-sampling.audit-retry-backoff-ms=${SYSLOG_BACKLOG_SAMPLING_AUDIT_RETRY_BACKOFF_MS:60000}
 # 大日志使用独立 Topic、消费组和 ConsumerFactory；建议生产环境独立 JVM 部署。
 syslog.large-listener.enabled=${SYSLOG_LARGE_LISTENER_ENABLED:true}
 syslog.large-listener.group-id=${SYSLOG_LARGE_LISTENER_GROUP_ID:${spring.kafka.consumer.group-id}-large}

haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/BacklogRetentionAuditMapper.xml

@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.common.mapper.BacklogRetentionAuditMapper">
+    <update id="upsert" parameterType="com.common.service.BacklogRetentionDelta">
+        MERGE INTO syslog_backlog_retention_hourly t
+        USING (
+            SELECT
+                #{sourceHour} AS source_hour,
+                #{sourceTopic} AS source_topic,
+                #{kafkaPartition} AS kafka_partition,
+                #{policyVersion} AS policy_version,
+                #{normalRetainBps} AS normal_retain_bps,
+                #{nonNormalRetainBps} AS non_normal_retain_bps,
+                #{messageTotal} AS message_total,
+                #{alarmRowsTotal} AS alarm_rows_total,
+                #{alarmRowsSaved} AS alarm_rows_saved,
+                #{normalRowsTotal} AS normal_rows_total,
+                #{normalRowsSaved} AS normal_rows_saved,
+                #{normalRowsDropped} AS normal_rows_dropped,
+                #{nonNormalRowsTotal} AS non_normal_rows_total,
+                #{nonNormalRowsSaved} AS non_normal_rows_saved,
+                #{nonNormalRowsDropped} AS non_normal_rows_dropped,
+                #{rawTotal} AS raw_total,
+                #{rawSaved} AS raw_saved,
+                #{rawDropped} AS raw_dropped,
+                #{minLagMs} AS min_lag_ms,
+                #{maxLagMs} AS max_lag_ms
+            FROM dual
+        ) s
+        ON (
+            t.source_hour = s.source_hour
+            AND t.source_topic = s.source_topic
+            AND t.kafka_partition = s.kafka_partition
+            AND t.policy_version = s.policy_version
+            AND t.normal_retain_bps = s.normal_retain_bps
+            AND t.non_normal_retain_bps = s.non_normal_retain_bps
+        )
+        WHEN MATCHED THEN UPDATE SET
+            t.message_total = t.message_total + s.message_total,
+            t.alarm_rows_total = t.alarm_rows_total + s.alarm_rows_total,
+            t.alarm_rows_saved = t.alarm_rows_saved + s.alarm_rows_saved,
+            t.normal_rows_total = t.normal_rows_total + s.normal_rows_total,
+            t.normal_rows_saved = t.normal_rows_saved + s.normal_rows_saved,
+            t.normal_rows_dropped = t.normal_rows_dropped + s.normal_rows_dropped,
+            t.non_normal_rows_total = t.non_normal_rows_total + s.non_normal_rows_total,
+            t.non_normal_rows_saved = t.non_normal_rows_saved + s.non_normal_rows_saved,
+            t.non_normal_rows_dropped = t.non_normal_rows_dropped + s.non_normal_rows_dropped,
+            t.raw_total = t.raw_total + s.raw_total,
+            t.raw_saved = t.raw_saved + s.raw_saved,
+            t.raw_dropped = t.raw_dropped + s.raw_dropped,
+            t.min_lag_ms = CASE WHEN t.min_lag_ms &lt; s.min_lag_ms THEN t.min_lag_ms ELSE s.min_lag_ms END,
+            t.max_lag_ms = CASE WHEN t.max_lag_ms &gt; s.max_lag_ms THEN t.max_lag_ms ELSE s.max_lag_ms END,
+            t.updated_at = SYSDATE
+        WHEN NOT MATCHED THEN INSERT (
+            source_hour, source_topic, kafka_partition, policy_version,
+            normal_retain_bps, non_normal_retain_bps, message_total,
+            alarm_rows_total, alarm_rows_saved, normal_rows_total, normal_rows_saved,
+            normal_rows_dropped, non_normal_rows_total, non_normal_rows_saved,
+            non_normal_rows_dropped, raw_total, raw_saved, raw_dropped,
+            min_lag_ms, max_lag_ms, created_at, updated_at
+        ) VALUES (
+            s.source_hour, s.source_topic, s.kafka_partition, s.policy_version,
+            s.normal_retain_bps, s.non_normal_retain_bps, s.message_total,
+            s.alarm_rows_total, s.alarm_rows_saved, s.normal_rows_total, s.normal_rows_saved,
+            s.normal_rows_dropped, s.non_normal_rows_total, s.non_normal_rows_saved,
+            s.non_normal_rows_dropped, s.raw_total, s.raw_saved, s.raw_dropped,
+            s.min_lag_ms, s.max_lag_ms, SYSDATE, SYSDATE
+        )
+    </update>
+</mapper>

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/BacklogSamplingDeciderTest.java

@@ -0,0 +1,126 @@
+package com.Modules.NormalData;
+
+import com.config.BacklogSamplingProperties;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class BacklogSamplingDeciderTest {
+    private BacklogSamplingProperties properties;
+    private BacklogSamplingDecider decider;
+    private final Instant now = Instant.parse("2026-09-12T12:00:00Z");
+
+    @BeforeEach
+    void setUp() {
+        properties = new BacklogSamplingProperties();
+        ReflectionTestUtils.setField(properties, "enabled", true);
+        ReflectionTestUtils.setField(properties, "startLagMs", 900_000L);
+        ReflectionTestUtils.setField(properties, "lagThresholdMs", 3_600_000L);
+        ReflectionTestUtils.setField(properties, "normalRetainBps", 0);
+        ReflectionTestUtils.setField(properties, "nonNormalRetainBps", 0);
+        ReflectionTestUtils.setField(properties, "degradeSteps", 10);
+        ReflectionTestUtils.setField(properties, "policyVersion", "test-v1");
+        decider = new BacklogSamplingDecider(properties);
+    }
+
+    @Test
+    void recentMessageIsFullyRetained() {
+        BacklogSamplingDecision decision = decider.decide("event", LocalDateTime.of(2026, 9, 12, 19, 58),
+                false, true, false, now);
+
+        assertFalse(decision.isDegraded());
+        assertTrue(decision.isKeepNormal());
+        assertTrue(decision.isKeepRaw());
+    }
+
+    @Test
+    void oldNormalMessageCanDropBothDmRowAndRawPoint() {
+        BacklogSamplingDecision decision = decider.decide("event", LocalDateTime.of(2026, 9, 12, 18, 0),
+                false, true, false, now);
+
+        assertTrue(decision.isDegraded());
+        assertFalse(decision.isKeepNormal());
+        assertFalse(decision.isKeepRaw());
+    }
+
+    @Test
+    void oldAlarmAlwaysKeepsRawEvenAtZeroNormalRate() {
+        BacklogSamplingDecision decision = decider.decide("event", LocalDateTime.of(2026, 9, 12, 18, 0),
+                true, false, false, now);
+
+        assertTrue(decision.isDegraded());
+        assertTrue(decision.isKeepRaw());
+    }
+
+    @Test
+    void stableHashReturnsSameDecisionForReplay() {
+        boolean first = BacklogSamplingDecider.sampled("event-42", "NORMAL", 3210, "v7");
+        boolean second = BacklogSamplingDecider.sampled("event-42", "NORMAL", 3210, "v7");
+
+        assertTrue(first == second);
+    }
+
+    @Test
+    void disabledPolicyRetainsOldMessagesInFull() {
+        ReflectionTestUtils.setField(properties, "enabled", false);
+
+        BacklogSamplingDecision decision = decider.decide("event",
+                LocalDateTime.of(2000, 1, 1, 0, 0), false, true, true, now);
+
+        assertFalse(decision.isDegraded());
+        assertTrue(decision.isKeepNormal());
+        assertTrue(decision.isKeepNonNormal());
+        assertTrue(decision.isKeepRaw());
+    }
+
+    @Test
+    void fifteenMinuteGraceAndRemainingWindowAreSplitIntoTenEqualSteps() {
+        ReflectionTestUtils.setField(properties, "normalRetainBps", 1000);
+        ReflectionTestUtils.setField(properties, "nonNormalRetainBps", 2000);
+
+        BacklogSamplingDecision fifteenMinutes = decider.decide("event-15m",
+                LocalDateTime.of(2026, 9, 12, 19, 45), false, true, false, now);
+        BacklogSamplingDecision nineteenMinutes = decider.decide("event-19m",
+                LocalDateTime.of(2026, 9, 12, 19, 41), false, true, false, now);
+        BacklogSamplingDecision twentyMinutes = decider.decide("event-20m",
+                LocalDateTime.of(2026, 9, 12, 19, 40), false, true, false, now);
+        BacklogSamplingDecision thirtyEightMinutes = decider.decide("event-38m",
+                LocalDateTime.of(2026, 9, 12, 19, 22), false, true, false, now);
+        BacklogSamplingDecision oneHour = decider.decide("event-60m",
+                LocalDateTime.of(2026, 9, 12, 19, 0), false, true, false, now);
+
+        assertFalse(fifteenMinutes.isDegraded());
+        assertFalse(nineteenMinutes.isDegraded());
+        assertEquals(9100, twentyMinutes.getNormalRetainBps());
+        assertEquals(5500, thirtyEightMinutes.getNormalRetainBps());
+        assertEquals(1000, oneHour.getNormalRetainBps());
+        assertEquals(2000, oneHour.getNonNormalRetainBps());
+    }
+
+    @Test
+    void currentStepUsesGraceWindowAndReachesMinimumAtThreshold() {
+        assertEquals(0, BacklogSamplingDecider.currentStep(900_000L, 900_000L, 3_600_000L, 10));
+        assertEquals(0, BacklogSamplingDecider.currentStep(1_169_999L, 900_000L, 3_600_000L, 10));
+        assertEquals(1, BacklogSamplingDecider.currentStep(1_170_000L, 900_000L, 3_600_000L, 10));
+        assertEquals(9, BacklogSamplingDecider.currentStep(3_599_999L, 900_000L, 3_600_000L, 10));
+        assertEquals(10, BacklogSamplingDecider.currentStep(3_600_000L, 900_000L, 3_600_000L, 10));
+    }
+
+    @Test
+    void startLagMustBeLowerThanMinimumRateThreshold() {
+        ReflectionTestUtils.setField(properties, "startLagMs", 3_600_000L);
+
+        IllegalArgumentException failure = assertThrows(
+                IllegalArgumentException.class, properties::validate);
+
+        assertTrue(failure.getMessage().contains("start-lag-ms"));
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java

@@ -8,6 +8,9 @@ import com.common.service.DeviceCollectTaskService;
 import com.common.service.DeviceDeviceService;
 import com.common.service.DmStandardizedBatchWriter;
 import com.common.service.SyslogNonNormalMessageService;
+import com.common.service.SyslogNormalDataService;
+import com.common.service.BacklogRetentionDelta;
+import com.config.BacklogSamplingProperties;
 import com.influx.raw.RawInfluxMetrics;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.junit.jupiter.api.AfterEach;
@@ -55,6 +58,7 @@ class SysLogProcessorParallelTest {
     private DmStandardizedBatchWriter batchWriter;
     private SyslogNonNormalMessageService nonNormalMessageService;
     private RawInfluxMetrics rawInfluxMetrics;
+    private SyslogNormalDataService normalDataService;
 
     @BeforeEach
     void setUp() {
@@ -66,6 +70,7 @@ class SysLogProcessorParallelTest {
         batchWriter = mock(DmStandardizedBatchWriter.class);
         nonNormalMessageService = mock(SyslogNonNormalMessageService.class);
         rawInfluxMetrics = mock(RawInfluxMetrics.class);
+        normalDataService = mock(SyslogNormalDataService.class);
         ReflectionTestUtils.setField(processor, "taskExecutor", executor);
         ReflectionTestUtils.setField(processor, "processTimeoutMs", 5000L);
         ReflectionTestUtils.setField(processor, "logNormalProcessor", normalProcessor);
@@ -77,6 +82,9 @@ class SysLogProcessorParallelTest {
         ReflectionTestUtils.setField(processor, "nonNormalConcurrencyRetryBackoffMs", 0L);
         ReflectionTestUtils.setField(processor, "failureContinuityEnabled", true);
         ReflectionTestUtils.setField(processor, "rawInfluxMetrics", rawInfluxMetrics);
+        ReflectionTestUtils.setField(processor, "syslogNormalDataService", normalDataService);
+        when(normalDataService.isAlarmData(any())).thenAnswer(invocation ->
+                ((java.util.Map<?, ?>) invocation.getArgument(0)).get("event_level") != null);
         when(batchWriter.write(any())).thenReturn(new DmStandardizedBatchWriter.WriteReport<>());
         when(normalProcessor.captureRuleSnapshot(any())).thenReturn(
                 new BatchRuleSnapshot(Collections.emptyMap(), System.currentTimeMillis()));
@@ -290,6 +298,92 @@ class SysLogProcessorParallelTest {
         verify(ack).acknowledge();
     }
 
+    @Test
+    void influxFailureNeverBlocksMainFlowEvenWhenGeneralContinuityIsDisabled() {
+        SysLogProcessor spyProcessor = spy(processor);
+        ReflectionTestUtils.setField(spyProcessor, "failureContinuityEnabled", false);
+        RuntimeException influxFailure = new IllegalStateException("influx unavailable");
+        doThrow(influxFailure).when(spyProcessor)
+                .persistRawBatch(anyList(), any(SysLogProcessor.BatchMetrics.class));
+        when(failureRouter.routeContinuityDlt(any(), any(), eq(influxFailure), eq("RAW_STORAGE")))
+                .thenReturn(SyslogFailureRouter.TerminalState.DLT_PUBLISHED);
+        List<SysLogProcessor.PreparedMessage> messages = messages(2);
+
+        List<SysLogProcessor.PreparedMessage> result = spyProcessor.persistRawWithoutBlockingMainFlow(
+                messages, new SysLogProcessor.BatchMetrics(), "RAW");
+
+        assertEquals(messages, result);
+        verify(failureRouter, times(2))
+                .routeContinuityDlt(any(), any(), eq(influxFailure), eq("RAW_STORAGE"));
+        verify(rawInfluxMetrics).rawWriteBypassed(2L);
+    }
+
+    @Test
+    void influxAndDltDoubleFailureStillReturnsToMainFlow() {
+        SysLogProcessor spyProcessor = spy(processor);
+        RuntimeException influxFailure = new IllegalStateException("influx unavailable");
+        doThrow(influxFailure).when(spyProcessor)
+                .persistRawBatch(anyList(), any(SysLogProcessor.BatchMetrics.class));
+        when(failureRouter.routeContinuityDlt(any(), any(), eq(influxFailure), eq("RAW_STORAGE")))
+                .thenThrow(new SyslogFailureRouter.FailurePublishException(
+                        "broker unavailable", new IllegalStateException("send failed")));
+        List<SysLogProcessor.PreparedMessage> messages = messages(2);
+
+        List<SysLogProcessor.PreparedMessage> result = spyProcessor.persistRawWithoutBlockingMainFlow(
+                messages, new SysLogProcessor.BatchMetrics(), "RAW");
+
+        assertEquals(messages, result);
+        verify(failureRouter).routeContinuityDlt(any(), any(), eq(influxFailure), eq("RAW_STORAGE"));
+        verify(rawInfluxMetrics).continuityDltPublishFailed(2L);
+        verify(rawInfluxMetrics).rawWriteBypassed(2L);
+    }
+
+    @Test
+    void backlogSamplingKeepsAlarmAndDropsNormalAndNonNormalFromRawPlan() {
+        BacklogSamplingProperties properties = new BacklogSamplingProperties();
+        ReflectionTestUtils.setField(properties, "enabled", true);
+        ReflectionTestUtils.setField(properties, "lagThresholdMs", 1L);
+        ReflectionTestUtils.setField(properties, "normalRetainBps", 0);
+        ReflectionTestUtils.setField(properties, "nonNormalRetainBps", 0);
+        ReflectionTestUtils.setField(properties, "degradeSteps", 10);
+        ReflectionTestUtils.setField(properties, "policyVersion", "test-v1");
+        ReflectionTestUtils.setField(processor, "backlogSamplingDecider",
+                new BacklogSamplingDecider(properties));
+
+        List<SysLogProcessor.PreparedMessage> source = oldMessages(3);
+        java.util.Map<String, Object> alarm = normalizedRow();
+        alarm.put("event_level", "HIGH");
+        java.util.Map<String, Object> normal = normalizedRow();
+        SyslogNonNormalMessage nonNormal = new SyslogNonNormalMessage();
+        nonNormal.setId("non-normal");
+        List<SysLogProcessor.PreparedNormalization> computed = new ArrayList<>();
+        computed.add(new SysLogProcessor.PreparedNormalization(source.get(0),
+                LogNormalProcessor.NormalizationResult.success(Collections.singletonList(alarm)), null));
+        computed.add(new SysLogProcessor.PreparedNormalization(source.get(1),
+                LogNormalProcessor.NormalizationResult.success(Collections.singletonList(normal)), null));
+        computed.add(new SysLogProcessor.PreparedNormalization(source.get(2),
+                LogNormalProcessor.NormalizationResult.filtered(Collections.singletonList(nonNormal)), null));
+
+        SysLogProcessor.SamplingPlan plan = processor.buildSamplingPlan(computed);
+
+        assertEquals(1, plan.rawMessageCount(), "只有告警原文必须保留");
+        assertEquals(3, plan.auditMessageCount());
+        long alarmSaved = 0L;
+        long normalDropped = 0L;
+        long nonNormalDropped = 0L;
+        long rawDropped = 0L;
+        for (BacklogRetentionDelta delta : plan.auditDeltas()) {
+            alarmSaved += delta.getAlarmRowsSaved();
+            normalDropped += delta.getNormalRowsDropped();
+            nonNormalDropped += delta.getNonNormalRowsDropped();
+            rawDropped += delta.getRawDropped();
+        }
+        assertEquals(1L, alarmSaved);
+        assertEquals(1L, normalDropped);
+        assertEquals(1L, nonNormalDropped);
+        assertEquals(2L, rawDropped);
+    }
+
     private List<ConsumerRecord<String, String>> records(int count) {
         List<ConsumerRecord<String, String>> records = new ArrayList<>();
         for (int i = 0; i < count; i++) {
@@ -314,6 +408,20 @@ class SysLogProcessorParallelTest {
         return messages;
     }
 
+    private List<SysLogProcessor.PreparedMessage> oldMessages(int count) {
+        List<SysLogProcessor.PreparedMessage> old = new ArrayList<>();
+        for (int i = 0; i < count; i++) {
+            ConsumerRecord<String, String> record =
+                    new ConsumerRecord<>("raw", i % 2, i, "key-" + i, "ciphertext");
+            SyslogFailureRouter.MessageContext context = new SyslogFailureRouter.MessageContext(
+                    "ciphertext", "old-event-" + i, 0, null, Instant.EPOCH);
+            old.add(new SysLogProcessor.PreparedMessage(record, context, "message-" + i,
+                    "old-event-" + i, null, LocalDateTime.of(2000, 1, 1, 0, 0),
+                    7L, 3L, System.currentTimeMillis()));
+        }
+        return old;
+    }
+
     private java.util.Map<String, Object> normalizedRow() {
         java.util.Map<String, Object> row = new java.util.HashMap<>();
         row.put("id", java.util.UUID.randomUUID().toString());

haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/BacklogSamplingAuditServiceTest.java

@@ -0,0 +1,99 @@
+package com.common.service;
+
+import com.common.mapper.BacklogRetentionAuditMapper;
+import com.config.BacklogSamplingProperties;
+import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
+
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class BacklogSamplingAuditServiceTest {
+
+    @Test
+    void aggregatesSameHourTopicPartitionAndPolicyWithoutMutatingInput() {
+        BacklogRetentionDelta first = delta(1, 1, 0, 1, 0);
+        BacklogRetentionDelta second = delta(1, 0, 1, 0, 1);
+
+        List<BacklogRetentionDelta> result = BacklogSamplingAuditService.aggregate(
+                Arrays.asList(first, second));
+
+        assertEquals(1, result.size());
+        assertEquals(2L, result.get(0).getMessageTotal());
+        assertEquals(1L, result.get(0).getNormalRowsSaved());
+        assertEquals(1L, result.get(0).getNormalRowsDropped());
+        assertEquals(1L, result.get(0).getRawSaved());
+        assertEquals(1L, result.get(0).getRawDropped());
+        assertEquals(1L, first.getMessageTotal(), "聚合不能修改调用方的逐消息审计对象");
+    }
+
+    @Test
+    void auditFailureReturnsFalseAndRecordsFullRetentionFallback() {
+        BacklogRetentionAuditMapper mapper = mock(BacklogRetentionAuditMapper.class);
+        when(mapper.upsert(any())).thenThrow(new IllegalStateException("table unavailable"));
+        PlatformTransactionManager transactionManager = transactionManager();
+        BacklogSamplingMetrics metrics = new BacklogSamplingMetrics();
+        BacklogSamplingAuditService service = new BacklogSamplingAuditService(
+                mapper, properties(), metrics, transactionManager);
+
+        boolean audited = service.recordBeforeDropping(Arrays.asList(
+                delta(2, 1, 1, 1, 1)));
+
+        assertFalse(audited);
+        Map<String, Object> snapshot = metrics.snapshot();
+        assertEquals(1L, snapshot.get("audit_write_failures_total"));
+        assertEquals(2L, snapshot.get("audit_fallback_messages_total"));
+    }
+
+    @Test
+    void successfulAuditCommitsAndRecordsSamplingMetrics() {
+        BacklogRetentionAuditMapper mapper = mock(BacklogRetentionAuditMapper.class);
+        when(mapper.upsert(any())).thenReturn(1);
+        PlatformTransactionManager transactionManager = transactionManager();
+        BacklogSamplingMetrics metrics = new BacklogSamplingMetrics();
+        BacklogSamplingAuditService service = new BacklogSamplingAuditService(
+                mapper, properties(), metrics, transactionManager);
+
+        assertTrue(service.recordBeforeDropping(Arrays.asList(
+                delta(1, 0, 1, 0, 1))));
+
+        verify(mapper).upsert(any());
+        assertEquals(1L, metrics.snapshot().get("degraded_messages_total"));
+        assertEquals(1L, metrics.snapshot().get("normal_rows_dropped_total"));
+    }
+
+    private BacklogSamplingProperties properties() {
+        BacklogSamplingProperties properties = new BacklogSamplingProperties();
+        ReflectionTestUtils.setField(properties, "auditTimeoutSeconds", 5);
+        ReflectionTestUtils.setField(properties, "auditRetryBackoffMs", 60_000L);
+        return properties;
+    }
+
+    private PlatformTransactionManager transactionManager() {
+        PlatformTransactionManager transactionManager = mock(PlatformTransactionManager.class);
+        TransactionStatus status = mock(TransactionStatus.class);
+        when(transactionManager.getTransaction(any(TransactionDefinition.class))).thenReturn(status);
+        return transactionManager;
+    }
+
+    private BacklogRetentionDelta delta(long messages, long normalSaved, long normalDropped,
+                                        long rawSaved, long rawDropped) {
+        return new BacklogRetentionDelta(LocalDateTime.of(2026, 9, 12, 18, 0),
+                "raw", 0, "v1", 1000, 10000,
+                messages, 0, 0, 1, normalSaved, normalDropped,
+                0, 0, 0, 1, rawSaved, rawDropped, 3_600_000L, 3_600_000L);
+    }
+}

haobang-security-dm/upgrade-doc/README.md

@@ -7,11 +7,13 @@
 | `v1/` | 原有升级资料 |
 | `v2/` | 从提交 `b61ec2e` 升级到当前源码所需的增量资料 |
 | `v3/` | 日统计异步投影，以及大日志独立 Kafka 通道、字节背压和存储截断升级资料 |
+| `v4/` | Kafka 积压分级采样、告警优先落库和 InfluxDB 非阻塞旁路升级资料 |
 
-新环境按 v1、v2、v3 的顺序执行；已经完成 v2 的现场只执行 v3。已经执行过早期 v3 日统计升级的
+新环境按 v1、v2、v3、v4 的顺序执行；已经完成 v3 的现场只执行 v4。已经执行过早期 v3 日统计升级的
-现场，只需补做 v3 中的大日志 Topic、应用升级和验证步骤，不要重复执行数据库升级脚本。
+现场，只需补做 v3 中的大日志 Topic、应用升级和验证步骤，再执行 v4；不要重复执行数据库升级脚本。
 
 数据库脚本不要重复执行。执行过程中有报错时，先确认已经落库的对象，再决定从哪里继续。
 
 v2 的执行步骤见 [v2/README.md](./v2/README.md)。
 v3 的执行步骤见 [v3/README.md](./v3/README.md)。
+v4 的执行步骤见 [v4/README.md](./v4/README.md)。

haobang-security-dm/upgrade-doc/v1/kafka/04-clear-topic-messages.sh

@@ -0,0 +1,162 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+# 物理清空指定 Kafka Topic 中当前已有的消息，但保留 Topic 本身、分区、副本和配置。
+#
+# 默认值对应当前客户现场，复制到 Kafka 主机后可直接执行：
+#   bash 04-clear-topic-messages.sh
+#
+# 清理其他 Topic 时可覆盖 TARGET_TOPIC：
+#   TARGET_TOPIC=other-topic bash 04-clear-topic-messages.sh
+#
+# 有认证的集群需额外设置：
+#   COMMAND_CONFIG=/path/to/client.properties bash 04-clear-topic-messages.sh
+#
+# 注意：
+# - 执行前必须停止目标 Topic 的生产者和消费者；
+# - 消息删除后通常不可恢复；
+# - 本脚本不会删除 Topic，也不会修改任何消费组 offset；
+# - 若消费组 offset 低于删除后的日志起始 offset，消费者后续行为取决于 auto.offset.reset。
+
+SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)
+BOOTSTRAP="${BOOTSTRAP:-10.141.121.35:9092,10.141.121.50:9092,10.141.121.59:9092}"
+RAW_TOPIC="${RAW_TOPIC:-agent-01-syslog-topic}"
+TARGET_TOPIC="${TARGET_TOPIC:-${RAW_TOPIC}}"
+
+if [[ ! "${TARGET_TOPIC}" =~ ^[a-zA-Z0-9._-]+$ ]]; then
+  echo "ERROR: TARGET_TOPIC 包含非法字符：${TARGET_TOPIC}" >&2
+  exit 1
+fi
+
+# 优先使用调用方显式指定的目录；否则自动识别脚本同目录、现场 Kafka 安装目录或 PATH。
+if [[ -z "${KAFKA_BIN:-}" ]]; then
+  for CANDIDATE_BIN in \
+    "${SCRIPT_DIR}" \
+    "/opt/kafka/kafka_2.13-3.6.1/bin" \
+    "/opt/kafka/bin"; do
+    if [[ -x "${CANDIDATE_BIN}/kafka-topics.sh" ]] \
+        && [[ -x "${CANDIDATE_BIN}/kafka-delete-records.sh" ]]; then
+      KAFKA_BIN="${CANDIDATE_BIN}"
+      break
+    fi
+  done
+fi
+
+if [[ -z "${KAFKA_BIN:-}" ]] \
+    && command -v kafka-topics.sh >/dev/null 2>&1 \
+    && command -v kafka-delete-records.sh >/dev/null 2>&1; then
+  KAFKA_BIN=$(dirname -- "$(command -v kafka-topics.sh)")
+fi
+
+if [[ -z "${KAFKA_BIN:-}" ]]; then
+  echo "ERROR: 未找到 Kafka CLI；现场预期路径为 /opt/kafka/kafka_2.13-3.6.1/bin" >&2
+  exit 1
+fi
+
+TOPICS_CMD="${KAFKA_BIN}/kafka-topics.sh"
+DELETE_RECORDS_CMD="${KAFKA_BIN}/kafka-delete-records.sh"
+
+if [[ ! -x "${TOPICS_CMD}" ]] || [[ ! -x "${DELETE_RECORDS_CMD}" ]]; then
+  echo "ERROR: ${KAFKA_BIN} 下缺少 kafka-topics.sh 或 kafka-delete-records.sh" >&2
+  exit 1
+fi
+
+COMMON_ARGS=(--bootstrap-server "${BOOTSTRAP}")
+if [[ -n "${COMMAND_CONFIG:-}" ]]; then
+  if [[ ! -r "${COMMAND_CONFIG}" ]]; then
+    echo "ERROR: 无法读取认证配置 ${COMMAND_CONFIG}" >&2
+    exit 1
+  fi
+  COMMON_ARGS+=(--command-config "${COMMAND_CONFIG}")
+fi
+
+RUN_ID=$(date '+%Y%m%d_%H%M%S')
+AUDIT_DIR="${CLEAR_AUDIT_DIR:-./kafka-message-clear-${RUN_ID}}"
+mkdir -p "${AUDIT_DIR}"
+
+BEFORE_FILE="${AUDIT_DIR}/topic-before.txt"
+DELETE_JSON="${AUDIT_DIR}/delete-records.json"
+DELETE_RESULT="${AUDIT_DIR}/delete-result.txt"
+AFTER_FILE="${AUDIT_DIR}/topic-after.txt"
+
+echo "目标 Kafka：${BOOTSTRAP}"
+echo "目标 Topic：${TARGET_TOPIC}"
+echo "Kafka CLI：${KAFKA_BIN}"
+echo "审计目录：${AUDIT_DIR}"
+
+echo "[1/5] 测试 Kafka 连接并确认 Topic 存在"
+"${TOPICS_CMD}" "${COMMON_ARGS[@]}" --list >/dev/null
+TOPIC_INFO=$("${TOPICS_CMD}" "${COMMON_ARGS[@]}" --describe --topic "${TARGET_TOPIC}")
+printf '%s\n' "${TOPIC_INFO}" | tee "${BEFORE_FILE}"
+
+PARTITION_COUNT=$(printf '%s\n' "${TOPIC_INFO}" | awk '
+  /PartitionCount:/ {
+    for (i=1; i<=NF; i++) if ($i=="PartitionCount:") { print $(i+1); exit }
+  }
+')
+
+if [[ ! "${PARTITION_COUNT}" =~ ^[1-9][0-9]*$ ]]; then
+  echo "ERROR: 无法确认 ${TARGET_TOPIC} 的分区数，拒绝继续" >&2
+  exit 1
+fi
+
+echo "[2/5] 为 ${PARTITION_COUNT} 个分区生成删除清单"
+printf '%s\n' "${TOPIC_INFO}" | awk -v topic="${TARGET_TOPIC}" '
+  BEGIN {
+    printf "{\"partitions\":["
+    first=1
+  }
+  /Partition:/ {
+    for (i=1; i<=NF; i++) {
+      if ($i=="Partition:" && $(i+1)~/^[0-9]+$/) {
+        if (!first) printf ","
+        printf "{\"topic\":\"%s\",\"partition\":%s,\"offset\":-1}", topic, $(i+1)
+        first=0
+      }
+    }
+  }
+  END {
+    print "],\"version\":1}"
+  }
+' > "${DELETE_JSON}"
+
+DELETE_ROWS=$(awk -F'"partition":' '{count+=NF-1} END {print count+0}' "${DELETE_JSON}")
+if [[ "${DELETE_ROWS}" -ne "${PARTITION_COUNT}" ]]; then
+  echo "ERROR: 删除清单只识别到 ${DELETE_ROWS}/${PARTITION_COUNT} 个分区，拒绝继续" >&2
+  exit 1
+fi
+
+cat "${DELETE_JSON}"
+
+EXPECTED_CONFIRM="CLEAR:${TARGET_TOPIC}:${PARTITION_COUNT}"
+echo "[3/5] 等待人工确认"
+echo "警告：将永久删除 ${TARGET_TOPIC} 当前已有的全部消息，但保留 Topic。"
+echo "请确认生产者和消费者均已停止。"
+echo "如确认执行，请完整输入：${EXPECTED_CONFIRM}"
+
+if [[ -n "${CONFIRM_CLEAR:-}" ]]; then
+  CONFIRM_INPUT="${CONFIRM_CLEAR}"
+elif [[ -t 0 ]]; then
+  read -r CONFIRM_INPUT
+else
+  echo "ERROR: 当前不是交互终端；必须通过 CONFIRM_CLEAR 提供精确确认串" >&2
+  exit 1
+fi
+
+if [[ "${CONFIRM_INPUT}" != "${EXPECTED_CONFIRM}" ]]; then
+  echo "CANCELLED: 确认串不匹配，未删除任何消息。" >&2
+  exit 1
+fi
+
+echo "[4/5] 清空 Topic 当前已有消息"
+"${DELETE_RECORDS_CMD}" "${COMMON_ARGS[@]}" \
+  --offset-json-file "${DELETE_JSON}" | tee "${DELETE_RESULT}"
+
+echo "[5/5] 再次检查 Topic"
+"${TOPICS_CMD}" "${COMMON_ARGS[@]}" \
+  --describe --topic "${TARGET_TOPIC}" | tee "${AFTER_FILE}"
+
+echo "PASS: ${TARGET_TOPIC} 当前已有消息已提交物理删除。"
+echo "Topic、分区、副本和配置均保留，消费组 offset 未修改。"
+echo "执行记录：${AUDIT_DIR}"

haobang-security-dm/upgrade-doc/v3/README.md

@@ -10,7 +10,7 @@ v3 将以下统计表的热点更新移出 `syslog-consumer-rule` 事实事务
 
 事实事务只批量写事实表并追加一条聚合 Outbox。独立的 Kafka 投影消费者串行更新日统计，之后继续通过
 已有的 `STATS.MATERIALIZE` Topic 刷新 `device_device`、`dm_normalize_rule` 和
-`device_collect_task`。这样三台 `syslog-consumer-rule` 可以并行写事实，不再同时争抢相同日统计行锁。
+`device_collect_task`。单节点和多节点部署都不再让事实写入争抢日统计行锁。
 
 v3 同时增加大日志独立处理链路：
 
@@ -36,8 +36,8 @@ v3 同时增加大日志独立处理链路：
 
 数据库脚本必须使用 `syslog-consumer-rule` 实际连接的达梦用户执行。
 
-如果现场已经执行过早期 v3 日统计异步投影升级，只执行 `kafka/02-create-large-log-topics.sh` 和本页的
+如果现场已经执行过早期 v3 日统计异步投影升级，只执行大日志 Kafka 脚本和应用配置，不要重复执行
-大日志应用配置、验证步骤；不要重复执行数据库升级脚本。
+数据库升级脚本。
 
 ## 第一步：达梦
 
@@ -117,10 +117,10 @@ PASS: 大日志 Topic 已创建并统一配置。
 
 ## 第三步：升级应用
 
-升级 `syslog-serve` 和三台 `syslog-consumer-rule`；`syslog-consumer` 不需要因为本次改造升级。
+升级 `syslog-serve` 和现场的 `syslog-consumer-rule`；`syslog-consumer` 不需要因为本次改造升级。
 
 升级前先按 [新增变量配置清单](./config/01-new-environment-variables.md) 修改现场实际使用的 Compose 文件。
-三台 `syslog-consumer-rule` 的 Topic、消费组和截断上限必须保持一致。
+如果部署了多个 `syslog-consumer-rule`，它们的 Topic、消费组和截断上限必须保持一致。
 
 统计投影默认已经启用。确认现场没有显式配置：
 
@@ -128,9 +128,7 @@ PASS: 大日志 Topic 已创建并统一配置。
 STATS_PROJECTION_ASYNC_ENABLED=false
 ```
 
-先逐台升级 `syslog-consumer-rule`，可以启动一台观察后再升级另外两台；确认大日志 listener 正常后再
+先升级 `syslog-consumer-rule`；多实例时逐台滚动。确认大日志 listener 正常后再升级 `syslog-serve`。
-升级 `syslog-serve`。新旧版本滚动期间，同一条原始 Kafka 消息只会被消费组中的一个实例处理，不会因为
-滚动升级重复统计。
 
 ## 升级后验证
 

haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md

@@ -50,7 +50,7 @@ syslog-serve/docker-compose/docker-compose-serve-prod.yaml
 syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml
 ```
 
-在 `services.syslog-consumer-rule.environment` 中增加，三台实例保持一致：
+在 `services.syslog-consumer-rule.environment` 中增加；如果以后部署多实例，各实例保持一致：
 
 ```yaml
       # 大日志独立消费通道。
@@ -160,4 +160,4 @@ docker compose -f docker-compose-consumer-rule-prod.yaml config
 docker compose -f docker-compose-consumer-rule-prod.yaml up -d --force-recreate syslog-consumer-rule
 ```
 
-建议先升级三台 `syslog-consumer-rule`，确认大日志消费者正常后，再升级 `syslog-serve`。
+建议先升级 `syslog-consumer-rule`，确认大日志消费者正常后，再升级 `syslog-serve`。

haobang-security-dm/upgrade-doc/v4/README.md

@@ -0,0 +1,124 @@
+# v4 积压分级采样与告警优先升级说明
+
+## 升级范围
+
+v4 只升级 `syslog-consumer-rule` 和达梦数据库，不修改 `syslog-serve`，也不新增 Kafka Topic。
+
+本次升级包含：
+
+- 根据消息 `receive_time` 与当前时间的差值，对积压消息分级采样；
+- `syslog_normal_alarm` 始终全量保留，并在 InfluxDB 原文、普通表和非标表之前单独提交；
+- 未被保留的 `syslog_normal_data`、`syslog_non_normal_message` 及对应原文不写入存储；
+- 按消息来源小时、Topic、Partition、策略版本和实际保留率记录采样审计；
+- 审计表不可用时自动回退为全量保留，不执行有损采样；
+- InfluxDB 原文写入变为非关键旁路，启动检查或运行时写入失败只记录并继续主消费。
+
+## 文件与执行顺序
+
+| 顺序 | 文件 | 作用 |
+| --- | --- | --- |
+| 1 | `database/01-backlog-sampling-precheck.sql` | 只读检查目标表、约束和索引是否已经存在 |
+| 2 | `database/02-backlog-sampling-upgrade.sql` | 创建积压采样小时审计表，仅执行一次 |
+| 3 | `database/03-backlog-sampling-verify.sql` | 只读验证表结构，并查看实际保留情况 |
+| 4 | `config/01-new-environment-variables.md` | 核对新增配置，再替换 `syslog-consumer-rule` JAR |
+
+数据库脚本必须使用 `syslog-consumer-rule` 实际连接的达梦用户执行。
+
+## 第一步：升级达梦
+
+先执行：
+
+```text
+database/01-backlog-sampling-precheck.sql
+```
+
+- 如果目标表、约束和索引都不存在，执行 `02-backlog-sampling-upgrade.sql`，再执行
+  `03-backlog-sampling-verify.sql`；
+- 如果目标表已经存在，不要重复执行 `02`，直接执行 `03` 核对结构；
+- 如果只存在部分对象，先停止升级并确认此前执行到哪一步，不要盲目重跑建表脚本。
+
+`03` 验证应满足：
+
+- 返回 `SYSLOG_BACKLOG_RETENTION_HOURLY` 一张表；
+- 表包含 22 个字段；
+- 主键和保留率检查约束状态为 `ENABLED`；
+- 更新时间索引状态为 `VALID`。
+
+## 第二步：配置并升级应用
+
+按 [新增变量配置清单](./config/01-new-environment-variables.md) 核对配置，然后替换并重启
+`syslog-consumer-rule`。只替换 JAR、不增加环境变量时，积压采样也会按 JAR 内置默认值开启，因此必须先
+完成数据库升级。
+
+默认策略如下：
+
+| 消息延迟 | 普通记录保留率 | 非标记录保留率 |
+| --- | ---: | ---: |
+| 不大于 15 分钟 | 100% | 100% |
+| 大于 15、且小于 19.5 分钟 | 100% | 100% |
+| 19.5～小于 24 分钟 | 91% | 92% |
+| 24～小于 28.5 分钟 | 82% | 84% |
+| 28.5～小于 33 分钟 | 73% | 76% |
+| 33～小于 37.5 分钟 | 64% | 68% |
+| 37.5～小于 42 分钟 | 55% | 60% |
+| 42～小于 46.5 分钟 | 46% | 52% |
+| 46.5～小于 51 分钟 | 37% | 44% |
+| 51～小于 55.5 分钟 | 28% | 36% |
+| 55.5～小于 60 分钟 | 19% | 28% |
+| 不小于 60 分钟 | 10% | 20% |
+
+默认前 15 分钟为免采样窗口。之后使用 `（60 分钟 - 15 分钟）÷ 10 = 4.5 分钟` 的阶梯宽度；
+达到 1 小时才使用最低保留率。
+
+采样采用 `event_id` 的确定性哈希：同一策略版本下，同一消息重复处理时选择结果一致。规则计算始终执行，
+只有完成规则计算、确认是否产生告警并成功写入小时审计后，才执行丢弃。
+
+## 写入顺序与故障行为
+
+每个批次按以下顺序处理：
+
+1. 完整规则计算并提交采样审计；
+2. 单独提交全部 `syslog_normal_alarm`；
+3. 写入被保留消息的 InfluxDB 原文；
+4. 写入被保留的 `syslog_normal_data` 和 `syslog_non_normal_message`。
+
+InfluxDB 原文写入默认单次网络阶段最多等待 5 秒；发生依赖故障后，30 秒内快速跳过后续原文请求。启动
+健康检查失败、写入失败以及特殊 DLT 写入失败都只记录日志，不阻塞 Kafka 主消费进度。
+
+## 升级后验证
+
+查看消费和异常日志：
+
+```bash
+docker logs --since 10m syslog-consumer-rule-dm 2>&1 \
+| grep -E 'Syslog批次完成|BACKLOG_SAMPLING|INFLUX_RAW_BYPASS|SYSLOG_CONTINUITY_LOSS' \
+| tail -100
+```
+
+查看运行状态：
+
+```text
+GET /xdrrule/api/syslog/backlog-sampling/status
+GET /xdrrule/api/syslog/raw-storage/metrics
+```
+
+然后执行 `database/03-backlog-sampling-verify.sql` 的最后一段查询。没有延迟消息时，审计查询为空属于正常；
+发生积压后，应能按来源小时看到告警保存数、普通/非标实际保留率和原文丢弃数。
+
+重点确认：
+
+- `alarm_rows_saved` 始终等于 `alarm_rows_total`；
+- `audit_write_failures_total` 和 `audit_fallback_messages_total` 不持续增加；
+- InfluxDB 故障时 Kafka offset 仍继续推进，告警和达梦保留数据仍持续写入；
+- `raw_write_bypassed_messages_total` 增加表示原文旁路失败，不表示主消费停止。
+
+## 回滚
+
+紧急关闭采样可配置：
+
+```yaml
+- SYSLOG_BACKLOG_SAMPLING_ENABLED=false
+```
+
+重建 `syslog-consumer-rule` 容器后，普通、非标及原文恢复全量保留。也可回滚旧 JAR。无论采用哪种方式，
+都保留 `syslog_backlog_retention_hourly` 表，不要删除历史审计数据。

haobang-security-dm/upgrade-doc/v4/config/01-new-environment-variables.md

@@ -0,0 +1,54 @@
+# v4 新增变量配置
+
+配置文件：
+
+```text
+syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml
+```
+
+在 `services.syslog-consumer-rule.environment` 中增加：
+
+```yaml
+      # 前15分钟全量保留，之后到1小时之间分10级降低保留率；告警始终全量保留。
+      - SYSLOG_BACKLOG_SAMPLING_ENABLED=true
+      - SYSLOG_BACKLOG_SAMPLING_START_LAG_MS=900000
+      - SYSLOG_BACKLOG_SAMPLING_LAG_THRESHOLD_MS=3600000
+      - SYSLOG_BACKLOG_SAMPLING_NORMAL_RETAIN_BPS=1000
+      - SYSLOG_BACKLOG_SAMPLING_NON_NORMAL_RETAIN_BPS=2000
+      - SYSLOG_BACKLOG_SAMPLING_DEGRADE_STEPS=10
+      - SYSLOG_BACKLOG_SAMPLING_POLICY_VERSION=v2
+
+      # 小时审计最多等待 5 秒；失败后 60 秒内不重试，并回退全量保留。
+      - SYSLOG_BACKLOG_SAMPLING_AUDIT_TIMEOUT_SECONDS=5
+      - SYSLOG_BACKLOG_SAMPLING_AUDIT_RETRY_BACKOFF_MS=60000
+
+      # InfluxDB 原文写入失败时快速返回并继续主消费。
+      - INFLUXDB_RAW_WRITE_REQUEST_TIMEOUT=5s
+      - INFLUXDB_RAW_FAILURE_BACKOFF_MS=30000
+```
+
+其中：
+
+- `NORMAL_RETAIN_BPS` 和 `NON_NORMAL_RETAIN_BPS` 是达到延迟阈值后的最低保留万分比；`1000` 表示 10%，
+  `2000` 表示 20%；
+- `START_LAG_MS=900000` 是15分钟免采样窗口，该时间内普通、非标及原文全部保留；
+- `LAG_THRESHOLD_MS=3600000` 是达到最低保留率的时间；扣除15分钟后，剩余45分钟由10级平均分配，
+  每级4.5分钟；
+- `START_LAG_MS` 必须大于等于0并且小于 `LAG_THRESHOLD_MS`，配置非法时应用拒绝启动；
+- 修改起始延迟、结束阈值、最低保留率或级数时，必须同时修改 `POLICY_VERSION`，以便审计区分策略；
+- `AUDIT_TIMEOUT_SECONDS` 是单次达梦审计事务超时；审计失败时本批次全量保留；
+- `INFLUXDB_RAW_WRITE_REQUEST_TIMEOUT` 是连接、写入、读取各网络阶段的上限，不是整个批次的总超时；
+- `INFLUXDB_RAW_FAILURE_BACKOFF_MS` 是确认 InfluxDB 依赖故障后的快速跳过时间。
+
+这些默认值已经写在 JAR 的 `application.properties` 中。现场只替换 JAR 时同样会默认开启采样；显式写入
+Compose 是为了便于后续核查和调整。启动新 JAR 前必须先创建
+`SYSLOG_BACKLOG_RETENTION_HOURLY`，否则应用会因审计失败回退全量保留，并周期性记录错误日志。
+
+关闭采样只需设置：
+
+```yaml
+      - SYSLOG_BACKLOG_SAMPLING_ENABLED=false
+```
+
+InfluxDB 非阻塞旁路不受通用的 `SYSLOG_FAILURE_CONTINUITY_ENABLED` 开关控制：InfluxDB 写入失败始终记录并
+放行，避免原文存储故障阻塞告警和后续 Kafka 消息。

haobang-security-dm/upgrade-doc/v4/database/01-backlog-sampling-precheck.sql

@@ -0,0 +1,21 @@
+-- v4 达梦升级预检：只读，不修改任何对象。
+-- 必须使用 syslog-consumer-rule 实际连接的达梦用户执行。
+
+-- 首次升级应返回 0 行；如果返回 1 行，不要重复执行 02，直接执行 03 验证。
+SELECT table_name
+FROM user_tables
+WHERE table_name = 'SYSLOG_BACKLOG_RETENTION_HOURLY';
+
+-- 首次升级应返回 0 行。
+SELECT constraint_name, constraint_type, status, table_name
+FROM user_constraints
+WHERE constraint_name IN (
+  'PK_SYSLOG_BACKLOG_RET_HOURLY',
+  'CK_SYSLOG_BACKLOG_RET_BPS'
+)
+ORDER BY constraint_name;
+
+-- 首次升级应返回 0 行。
+SELECT index_name, table_name, uniqueness, status
+FROM user_indexes
+WHERE index_name = 'IDX_SYSLOG_BACKLOG_RET_UPDATED';

haobang-security-dm/upgrade-doc/v4/database/02-backlog-sampling-upgrade.sql

@@ -0,0 +1,49 @@
+-- v4 达梦升级：创建 Kafka 积压分级采样的小时审计表。
+-- 仅在 01 预检确认目标对象均不存在时执行一次，不要重复执行。
+
+CREATE TABLE syslog_backlog_retention_hourly (
+  source_hour TIMESTAMP NOT NULL,
+  source_topic VARCHAR2(255) NOT NULL,
+  kafka_partition INT NOT NULL,
+  policy_version VARCHAR2(64) NOT NULL,
+  normal_retain_bps INT NOT NULL,
+  non_normal_retain_bps INT NOT NULL,
+  message_total BIGINT DEFAULT 0 NOT NULL,
+  alarm_rows_total BIGINT DEFAULT 0 NOT NULL,
+  alarm_rows_saved BIGINT DEFAULT 0 NOT NULL,
+  normal_rows_total BIGINT DEFAULT 0 NOT NULL,
+  normal_rows_saved BIGINT DEFAULT 0 NOT NULL,
+  normal_rows_dropped BIGINT DEFAULT 0 NOT NULL,
+  non_normal_rows_total BIGINT DEFAULT 0 NOT NULL,
+  non_normal_rows_saved BIGINT DEFAULT 0 NOT NULL,
+  non_normal_rows_dropped BIGINT DEFAULT 0 NOT NULL,
+  raw_total BIGINT DEFAULT 0 NOT NULL,
+  raw_saved BIGINT DEFAULT 0 NOT NULL,
+  raw_dropped BIGINT DEFAULT 0 NOT NULL,
+  min_lag_ms BIGINT DEFAULT 0 NOT NULL,
+  max_lag_ms BIGINT DEFAULT 0 NOT NULL,
+  created_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  updated_at TIMESTAMP DEFAULT SYSDATE NOT NULL,
+  CONSTRAINT pk_syslog_backlog_ret_hourly PRIMARY KEY (
+    source_hour, source_topic, kafka_partition, policy_version,
+    normal_retain_bps, non_normal_retain_bps
+  ),
+  CONSTRAINT ck_syslog_backlog_ret_bps CHECK (
+    normal_retain_bps BETWEEN 0 AND 10000
+    AND non_normal_retain_bps BETWEEN 0 AND 10000
+  )
+);
+
+CREATE INDEX idx_syslog_backlog_ret_updated
+ON syslog_backlog_retention_hourly(updated_at);
+
+COMMENT ON TABLE syslog_backlog_retention_hourly IS
+'Kafka积压分级采样小时审计；按来源小时、Topic、Partition和策略聚合';
+COMMENT ON COLUMN syslog_backlog_retention_hourly.source_hour IS
+'syslog-serve生成的receive_time所在小时（Asia/Shanghai）';
+COMMENT ON COLUMN syslog_backlog_retention_hourly.normal_retain_bps IS
+'普通标准化记录该级实际保留万分比，10000表示100%';
+COMMENT ON COLUMN syslog_backlog_retention_hourly.non_normal_retain_bps IS
+'非标记录该级实际保留万分比，10000表示100%';
+
+COMMIT;

haobang-security-dm/upgrade-doc/v4/database/03-backlog-sampling-verify.sql

@@ -0,0 +1,56 @@
+-- v4 Kafka 积压分级采样升级验证：只读。
+
+-- 应返回 SYSLOG_BACKLOG_RETENTION_HOURLY 一行。
+SELECT table_name
+FROM user_tables
+WHERE table_name = 'SYSLOG_BACKLOG_RETENTION_HOURLY';
+
+-- 应返回 22 个字段。
+SELECT column_id, column_name, data_type, nullable
+FROM user_tab_columns
+WHERE table_name = 'SYSLOG_BACKLOG_RETENTION_HOURLY'
+ORDER BY column_id;
+
+-- 两个约束都应为 ENABLED。
+SELECT constraint_name, constraint_type, status
+FROM user_constraints
+WHERE constraint_name IN (
+  'PK_SYSLOG_BACKLOG_RET_HOURLY',
+  'CK_SYSLOG_BACKLOG_RET_BPS'
+)
+ORDER BY constraint_name;
+
+-- 索引应为 VALID。
+SELECT index_name, uniqueness, status
+FROM user_indexes
+WHERE index_name = 'IDX_SYSLOG_BACKLOG_RET_UPDATED';
+
+-- 结构验证后应能执行；首次建表且还未发生积压采样时返回 0。
+SELECT COUNT(*) AS audit_row_count
+FROM syslog_backlog_retention_hourly;
+
+-- 查看每个来源小时的实际选择率。该值反映采样选择，不代表后续存储一定写入成功。
+-- 没有发生积压采样时，本查询返回空属于正常。
+SELECT
+  source_hour,
+  SUM(message_total) AS message_total,
+  SUM(alarm_rows_total) AS alarm_rows_total,
+  SUM(alarm_rows_saved) AS alarm_rows_saved,
+  SUM(normal_rows_total) AS normal_rows_total,
+  SUM(normal_rows_saved) AS normal_rows_saved,
+  SUM(normal_rows_dropped) AS normal_rows_dropped,
+  CASE WHEN SUM(normal_rows_total) = 0 THEN 1
+       ELSE SUM(normal_rows_saved) * 1.0 / SUM(normal_rows_total) END AS normal_actual_rate,
+  SUM(non_normal_rows_total) AS non_normal_rows_total,
+  SUM(non_normal_rows_saved) AS non_normal_rows_saved,
+  SUM(non_normal_rows_dropped) AS non_normal_rows_dropped,
+  CASE WHEN SUM(non_normal_rows_total) = 0 THEN 1
+       ELSE SUM(non_normal_rows_saved) * 1.0 / SUM(non_normal_rows_total) END AS non_normal_actual_rate,
+  SUM(raw_total) AS raw_total,
+  SUM(raw_saved) AS raw_saved,
+  SUM(raw_dropped) AS raw_dropped,
+  MIN(min_lag_ms) AS min_lag_ms,
+  MAX(max_lag_ms) AS max_lag_ms
+FROM syslog_backlog_retention_hourly
+GROUP BY source_hour
+ORDER BY source_hour DESC;