From 87fee6fe6946a3f00e9c0aa270174379f8f369a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Sat, 12 Sep 2026 23:43:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(syslog):=20=E5=A2=9E=E5=8A=A0=E7=A7=AF?= =?UTF-8?q?=E5=8E=8B=E9=87=87=E6=A0=B7=E5=85=8D=E9=87=87=E6=A0=B7=E7=AA=97?= =?UTF-8?q?=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 start-lag-ms 配置，默认15分钟内全量保留 - 在15分钟至1小时之间按10级逐步降低保留率 - 延迟达到1小时后使用普通10%、非标20%的最低保留率 - 校验起始延迟必须小于最低保留率阈值 - 状态接口新增 start_lag_ms 配置展示 - 将采样策略版本升级为 v2 - 更新 Docker Compose 和 v4 升级配置说明 - 补充采样时间边界及非法配置单元测试 --- .../docker-compose-consumer-rule-prod.yaml | 13 + .../NormalData/BacklogSamplingDecider.java | 99 +++++ .../NormalData/BacklogSamplingDecision.java | 40 ++ .../Modules/NormalData/SysLogProcessor.java | 357 ++++++++++++++---- .../NormalData/SyslogFailureRouter.java | 4 +- .../mapper/BacklogRetentionAuditMapper.java | 9 + .../common/service/BacklogRetentionDelta.java | 118 ++++++ .../service/BacklogSamplingAuditService.java | 81 ++++ .../service/BacklogSamplingMetrics.java | 56 +++ .../main/java/com/config/AppProperties.java | 8 + .../com/config/BacklogSamplingProperties.java | 80 ++++ .../BacklogSamplingController.java | 37 ++ .../main/java/com/influx/InfluxDBClient.java | 58 ++- .../src/main/resources/application.properties | 18 + .../mapper/BacklogRetentionAuditMapper.xml | 70 ++++ .../BacklogSamplingDeciderTest.java | 126 +++++++ .../SysLogProcessorParallelTest.java | 108 ++++++ .../BacklogSamplingAuditServiceTest.java | 99 +++++ haobang-security-dm/upgrade-doc/README.md | 6 +- .../v1/kafka/04-clear-topic-messages.sh | 162 ++++++++ haobang-security-dm/upgrade-doc/v3/README.md | 14 +- .../v3/config/01-new-environment-variables.md | 4 +- haobang-security-dm/upgrade-doc/v4/README.md | 124 ++++++ .../v4/config/01-new-environment-variables.md | 54 +++ .../database/01-backlog-sampling-precheck.sql | 21 ++ .../database/02-backlog-sampling-upgrade.sql | 49 +++ .../database/03-backlog-sampling-verify.sql | 56 +++ 27 files changed, 1777 insertions(+), 94 deletions(-) create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/BacklogSamplingDecider.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/BacklogSamplingDecision.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/BacklogRetentionAuditMapper.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogRetentionDelta.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogSamplingAuditService.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogSamplingMetrics.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/BacklogSamplingProperties.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/BacklogSamplingController.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/BacklogRetentionAuditMapper.xml create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/BacklogSamplingDeciderTest.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/BacklogSamplingAuditServiceTest.java create mode 100755 haobang-security-dm/upgrade-doc/v1/kafka/04-clear-topic-messages.sh create mode 100644 haobang-security-dm/upgrade-doc/v4/README.md create mode 100644 haobang-security-dm/upgrade-doc/v4/config/01-new-environment-variables.md create mode 100644 haobang-security-dm/upgrade-doc/v4/database/01-backlog-sampling-precheck.sql create mode 100644 haobang-security-dm/upgrade-doc/v4/database/02-backlog-sampling-upgrade.sql create mode 100644 haobang-security-dm/upgrade-doc/v4/database/03-backlog-sampling-verify.sql diff --git a/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml b/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml index 036e952..6133c6c 100644 --- a/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-rule-prod.yaml +++ b/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 diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/BacklogSamplingDecider.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/BacklogSamplingDecider.java new file mode 100644 index 0000000..7549427 --- /dev/null +++ b/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); + } + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/BacklogSamplingDecision.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/BacklogSamplingDecision.java new file mode 100644 index 0000000..2948bc6 --- /dev/null +++ b/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; } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java index 5289b62..b75a812 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java @@ -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(); - List persistedMessages; - try { - persistedMessages = persistRawBatch(preparedMessages, metrics); - } catch (RuntimeException rawStorageFailure) { - if (!failureContinuityEnabled) throw rawStorageFailure; - routeContinuityPrepared(preparedMessages, rawStorageFailure, "RAW_STORAGE"); - rawInfluxMetrics.rawWriteBypassed(preparedMessages.size()); - metrics.rawStorageBypassed.addAndGet(preparedMessages.size()); - // 无论 DLT 是否可用，都继续使用内存中的完整内容执行规则和达梦写入。 - persistedMessages = preparedMessages; - log.error("[SYSLOG_CONTINUITY] InfluxDB 原文写入失败，已尝试记录 DLT 并继续标准化: " - + "lane={}, records={}, errorType={}", - lane, preparedMessages.size(), rawStorageFailure.getClass().getName(), - rawStorageFailure); + // 必须先完成规则计算，才能确保告警全量保留，并让未保留消息同时跳过 InfluxDB 和达梦。 + long normalizeStartedAt = System.currentTimeMillis(); + List computed = computePreparedNormalizations( + preparedMessages, metrics, normalizationExecutor); + SamplingPlan samplingPlan = buildSamplingPlan(computed); + if (!samplingPlan.auditDeltas.isEmpty()) { + boolean audited = backlogSamplingAuditService.recordBeforeDropping(samplingPlan.auditDeltas); + if (!audited) { + samplingPlan = allRetainedPlan(computed); + log.error("[BACKLOG_SAMPLING_FALLBACK] 采样审计不可用，当前批次已回退全量保留: " + + "lane={}, records={}", lane, records.size()); + } else { + metrics.recordSampling(samplingPlan); + } } + long normalizeMs = System.currentTimeMillis() - normalizeStartedAt; + + // 告警是最高优先级：先独立提交告警表，不能让 InfluxDB 或普通表的慢写阻塞告警入库。 + StandardizedPersistencePlan standardizedPlan = prepareStandardizedPersistence( + samplingPlan.normalizations, metrics); + long alarmWriteStartedAt = System.nanoTime(); + DmStandardizedBatchWriter.WriteReport alarmReport = + standardizedBatchWriter.write(standardizedPlan.alarmWriteRequests); + long alarmWriteMs = elapsedMillis(alarmWriteStartedAt); + + long influxStartedAt = System.currentTimeMillis(); + List persistedMessages = persistRawWithoutBlockingMainFlow( + samplingPlan.rawMessages, metrics, lane); long influxMs = System.currentTimeMillis() - influxStartedAt; - long normalizeStartedAt = System.currentTimeMillis(); - normalizePreparedMessages(persistedMessages, metrics, normalizationExecutor); - long normalizeMs = System.currentTimeMillis() - normalizeStartedAt; + persistRemainingStandardizedData(standardizedPlan, alarmReport, alarmWriteMs, metrics); 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 persistRawBatch(List messages, BatchMetrics metrics) { + /** + * InfluxDB 是非关键旁路：任何原文写入异常都不能越过本方法阻塞 Kafka 主消费。 + * DLT 仅用于尽力保留重放能力；DLT 也失败时 routeContinuityPrepared 会批量记录后返回。 + */ + List persistRawWithoutBlockingMainFlow( + List 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 persistRawBatch(List 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 messages, BatchMetrics metrics, Executor normalizationExecutor) { - if (messages.isEmpty()) { - return; - } + List computed = computePreparedNormalizations( + messages, metrics, normalizationExecutor); + persistComputedNormalizations(allRetainedPlan(computed).normalizations, metrics); + } + + private List computePreparedNormalizations( + List messages, BatchMetrics metrics, Executor normalizationExecutor) { + if (messages.isEmpty()) return Collections.emptyList(); long snapshotStartedAt = System.nanoTime(); Set 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> writeRequests = new ArrayList<>(); - List nonNormalMessages = new ArrayList<>(); - Set pendingDatabaseMessages = new HashSet<>(); - int filteredMessages = 0; + List computed = new ArrayList<>(futures.size()); for (CompletableFuture future : futures) { - PreparedNormalization prepared = 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 row : prepared.result.getRows()) { - writeRequests.add(new DmStandardizedBatchWriter.WriteRequest<>(prepared.message, row)); - } + computed.add(future.join()); } - - // 批写器只返回“可归因到具体消息”的行级失败；连接、超时、事务结果未知会直接抛出， - // 从而让本方法退出并保留整个 Kafka 批次 offset。 - long dmWriteStartedAt = System.nanoTime(); - DmStandardizedBatchWriter.WriteReport writeReport = - standardizedBatchWriter.write(writeRequests); - metrics.recordDmWriteMs(elapsedMillis(dmWriteStartedAt)); - Set failedMessages = writeReport.getFailures().keySet(); - metrics.recordDmWrite(writeRequests.size(), writeReport.getSuccessfulRows(), failedMessages.size()); - - for (Map.Entry 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); + return computed; } catch (TimeoutException timeout) { cancelOutstanding(futures); throw new IllegalStateException("标准化批次处理超时，禁止提交 offset", timeout); @@ -801,6 +805,143 @@ public class SysLogProcessor { } } + /** 根据已审计的保留决策写两张标准化表和非标表。 */ + private void persistComputedNormalizations(List normalizations, + BatchMetrics metrics) { + StandardizedPersistencePlan plan = prepareStandardizedPersistence(normalizations, metrics); + long alarmWriteStartedAt = System.nanoTime(); + DmStandardizedBatchWriter.WriteReport alarmReport = + standardizedBatchWriter.write(plan.alarmWriteRequests); + long alarmWriteMs = elapsedMillis(alarmWriteStartedAt); + persistRemainingStandardizedData(plan, alarmReport, alarmWriteMs, metrics); + } + + private StandardizedPersistencePlan prepareStandardizedPersistence( + List normalizations, BatchMetrics metrics) { + List> alarmWriteRequests = new ArrayList<>(); + List> normalWriteRequests = new ArrayList<>(); + List nonNormalMessages = new ArrayList<>(); + Set 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 row : prepared.result.getRows()) { + boolean alarm = syslogNormalDataService.isAlarmData(row); + if (alarm || retained.keepNormal) { + DmStandardizedBatchWriter.WriteRequest 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 alarmReport, + long alarmWriteMs, BatchMetrics metrics) { + long normalWriteStartedAt = System.nanoTime(); + DmStandardizedBatchWriter.WriteReport normalReport = + standardizedBatchWriter.write(plan.normalWriteRequests); + metrics.recordDmWriteMs(alarmWriteMs + elapsedMillis(normalWriteStartedAt)); + Map failures = new LinkedHashMap<>(); + failures.putAll(alarmReport.getFailures()); + for (Map.Entry failure : normalReport.getFailures().entrySet()) { + failures.putIfAbsent(failure.getKey(), failure.getValue()); + } + Set 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 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 computed) { + List retained = new ArrayList<>(computed.size()); + List rawMessages = new ArrayList<>(computed.size()); + List 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 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 computed) { + List retained = new ArrayList<>(computed.size()); + List 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, - LogNormalProcessor.NormalizationResult result, - SyslogFailureRouter.TerminalState terminalState) { + PreparedNormalization(PreparedMessage message, + LogNormalProcessor.NormalizationResult result, + 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 normalizations; + private final List rawMessages; + private final List auditDeltas; + + private SamplingPlan(List normalizations, + List rawMessages, + List auditDeltas) { + this.normalizations = normalizations; + this.rawMessages = rawMessages; + this.auditDeltas = auditDeltas; + } + + int rawMessageCount() { return rawMessages.size(); } + int auditMessageCount() { return auditDeltas.size(); } + List auditDeltas() { return auditDeltas; } + } + + private static final class StandardizedPersistencePlan { + private final List> alarmWriteRequests; + private final List> normalWriteRequests; + private final List nonNormalMessages; + private final Set pendingDatabaseMessages; + private final int filteredMessages; + + private StandardizedPersistencePlan( + List> alarmWriteRequests, + List> normalWriteRequests, + List nonNormalMessages, + Set 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); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java index f3899ca..63a3d13 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SyslogFailureRouter.java +++ b/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 数据格式错误只影响当前消息，可直接隔离； - // 连接、超时、5xx 等 InfluxDB 故障属于系统性依赖故障，必须保留原 offset，避免 DLT 风暴。 + // InfluxDB 400 数据格式错误只影响当前消息，可直接隔离；连接、超时、5xx 属于系统性依赖故障。 + // 活动原文批写链路会在外层非阻塞旁路边界覆盖其 offset 行为，不能因此阻塞主消费。 if (root instanceof BadRequestException) { return FailureCategory.PERMANENT_DATA; } diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/BacklogRetentionAuditMapper.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/mapper/BacklogRetentionAuditMapper.java new file mode 100644 index 0000000..70c39d8 --- /dev/null +++ b/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); +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogRetentionDelta.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogRetentionDelta.java new file mode 100644 index 0000000..325d894 --- /dev/null +++ b/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; } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogSamplingAuditService.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogSamplingAuditService.java new file mode 100644 index 0000000..41ad01c --- /dev/null +++ b/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 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 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 aggregate(List deltas) { + Map 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 deltas) { + long total = 0L; + for (BacklogRetentionDelta delta : deltas) total += delta.getMessageTotal(); + return total; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogSamplingMetrics.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/BacklogSamplingMetrics.java new file mode 100644 index 0000000..2ddf078 --- /dev/null +++ b/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 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 snapshot() { + Map 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; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/AppProperties.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/AppProperties.java index 14c3f02..82cd41e 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/AppProperties.java +++ b/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; diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/BacklogSamplingProperties.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/BacklogSamplingProperties.java new file mode 100644 index 0000000..d4af09c --- /dev/null +++ b/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; } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/BacklogSamplingController.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/BacklogSamplingController.java new file mode 100644 index 0000000..022d08f --- /dev/null +++ b/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 status() { + Map 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; + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/InfluxDBClient.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/InfluxDBClient.java index 3534008..5ccc873 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/InfluxDBClient.java +++ b/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()); - throw new RuntimeException("InfluxDB connection failed", e); + // 原文存储是旁路依赖。启动时不可用不能阻止 Kafka 消费和告警入库。 + 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 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（异步） */ diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties index f1cb58f..ae091d5 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties @@ -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} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/BacklogRetentionAuditMapper.xml b/haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/BacklogRetentionAuditMapper.xml new file mode 100644 index 0000000..2f2d749 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/mapper/BacklogRetentionAuditMapper.xml @@ -0,0 +1,70 @@ + + + + + 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 < 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 > 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 + ) + + diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/BacklogSamplingDeciderTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/BacklogSamplingDeciderTest.java new file mode 100644 index 0000000..b44f285 --- /dev/null +++ b/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")); + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java index 5b569f3..9afb0f0 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorParallelTest.java @@ -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 messages = messages(2); + + List 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 messages = messages(2); + + List 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 source = oldMessages(3); + java.util.Map alarm = normalizedRow(); + alarm.put("event_level", "HIGH"); + java.util.Map normal = normalizedRow(); + SyslogNonNormalMessage nonNormal = new SyslogNonNormalMessage(); + nonNormal.setId("non-normal"); + List 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> records(int count) { List> records = new ArrayList<>(); for (int i = 0; i < count; i++) { @@ -314,6 +408,20 @@ class SysLogProcessorParallelTest { return messages; } + private List oldMessages(int count) { + List old = new ArrayList<>(); + for (int i = 0; i < count; i++) { + ConsumerRecord 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 normalizedRow() { java.util.Map row = new java.util.HashMap<>(); row.put("id", java.util.UUID.randomUUID().toString()); diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/BacklogSamplingAuditServiceTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/BacklogSamplingAuditServiceTest.java new file mode 100644 index 0000000..c5670ee --- /dev/null +++ b/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 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 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); + } +} diff --git a/haobang-security-dm/upgrade-doc/README.md b/haobang-security-dm/upgrade-doc/README.md index b05f2be..73d5074 100644 --- a/haobang-security-dm/upgrade-doc/README.md +++ b/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 日统计升级的 -现场，只需补做 v3 中的大日志 Topic、应用升级和验证步骤，不要重复执行数据库升级脚本。 +新环境按 v1、v2、v3、v4 的顺序执行；已经完成 v3 的现场只执行 v4。已经执行过早期 v3 日统计升级的 +现场，只需补做 v3 中的大日志 Topic、应用升级和验证步骤，再执行 v4；不要重复执行数据库升级脚本。 数据库脚本不要重复执行。执行过程中有报错时，先确认已经落库的对象，再决定从哪里继续。 v2 的执行步骤见 [v2/README.md](./v2/README.md)。 v3 的执行步骤见 [v3/README.md](./v3/README.md)。 +v4 的执行步骤见 [v4/README.md](./v4/README.md)。 diff --git a/haobang-security-dm/upgrade-doc/v1/kafka/04-clear-topic-messages.sh b/haobang-security-dm/upgrade-doc/v1/kafka/04-clear-topic-messages.sh new file mode 100755 index 0000000..c741ac5 --- /dev/null +++ b/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}" diff --git a/haobang-security-dm/upgrade-doc/v3/README.md b/haobang-security-dm/upgrade-doc/v3/README.md index fa76d1d..18d617b 100644 --- a/haobang-security-dm/upgrade-doc/v3/README.md +++ b/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-serve`。新旧版本滚动期间，同一条原始 Kafka 消息只会被消费组中的一个实例处理，不会因为 -滚动升级重复统计。 +先升级 `syslog-consumer-rule`；多实例时逐台滚动。确认大日志 listener 正常后再升级 `syslog-serve`。 ## 升级后验证 diff --git a/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md b/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md index 84072b7..f4a0978 100644 --- a/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md +++ b/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`。 diff --git a/haobang-security-dm/upgrade-doc/v4/README.md b/haobang-security-dm/upgrade-doc/v4/README.md new file mode 100644 index 0000000..ae43e01 --- /dev/null +++ b/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` 表，不要删除历史审计数据。 diff --git a/haobang-security-dm/upgrade-doc/v4/config/01-new-environment-variables.md b/haobang-security-dm/upgrade-doc/v4/config/01-new-environment-variables.md new file mode 100644 index 0000000..58ad7ba --- /dev/null +++ b/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 消息。 diff --git a/haobang-security-dm/upgrade-doc/v4/database/01-backlog-sampling-precheck.sql b/haobang-security-dm/upgrade-doc/v4/database/01-backlog-sampling-precheck.sql new file mode 100644 index 0000000..dd029a6 --- /dev/null +++ b/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'; diff --git a/haobang-security-dm/upgrade-doc/v4/database/02-backlog-sampling-upgrade.sql b/haobang-security-dm/upgrade-doc/v4/database/02-backlog-sampling-upgrade.sql new file mode 100644 index 0000000..bb5c7e2 --- /dev/null +++ b/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; diff --git a/haobang-security-dm/upgrade-doc/v4/database/03-backlog-sampling-verify.sql b/haobang-security-dm/upgrade-doc/v4/database/03-backlog-sampling-verify.sql new file mode 100644 index 0000000..02e499b --- /dev/null +++ b/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;