From 8f1a8b628318ce2e8a4dfa209e41c86f8cfbbb82 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Wed, 16 Sep 2026 17:32:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(syslog):=20=E5=A2=9E=E5=8A=A0=20WAL=20?= =?UTF-8?q?=E4=B8=8E=E5=88=86=20IP=20=E6=8E=A5=E5=85=A5=E8=AF=8A=E6=96=AD?= =?UTF-8?q?=20-=20=E6=AF=8F=E5=88=86=E9=92=9F=E4=BF=9D=E5=AD=98=20WAL=20?= =?UTF-8?q?=E5=85=A8=E5=B1=80=E7=8A=B6=E6=80=81=E5=92=8C=E5=88=86=20IP=20?= =?UTF-8?q?=E6=8E=A5=E6=94=B6=E7=BB=9F=E8=AE=A1=EF=BC=8C=E5=8E=86=E5=8F=B2?= =?UTF-8?q?=E4=BF=9D=E7=95=99=207=20=E5=A4=A9=20-=20=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E6=8E=A5=E6=94=B6=E3=80=81WAL=20=E8=90=BD=E7=9B=98=E3=80=81Kaf?= =?UTF-8?q?ka=20ACK=E3=80=81=E8=BF=87=E6=BB=A4=E3=80=81=E8=83=8C=E5=8E=8B?= =?UTF-8?q?=E5=8F=8A=E4=B8=A2=E5=BC=83=E6=95=B0=E9=87=8F=20-=20=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=20wal=5Fmax=5Fpending=5Fbytes=20=E5=92=8C=20wal=5Fadm?= =?UTF-8?q?ission=5Funavailable=20=E6=8C=87=E6=A0=87=20-=20=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=20TCP=20pending=20=E5=BD=93=E5=89=8D=E5=80=BC?= =?UTF-8?q?=E5=8F=8A=E8=BF=9B=E7=A8=8B=E5=86=85=E5=B3=B0=E5=80=BC=E7=BB=9F?= =?UTF-8?q?=E8=AE=A1=20-=20=E7=BD=91=E7=BB=9C=E5=BC=82=E5=B8=B8=E5=90=8E?= =?UTF-8?q?=E5=90=8C=E4=B8=80=20IP=20=E6=81=A2=E5=A4=8D=E6=8E=A5=E6=94=B6?= =?UTF-8?q?=E6=97=B6=E6=89=93=E5=8D=B0=E6=81=A2=E5=A4=8D=E6=97=A5=E5=BF=97?= =?UTF-8?q?=20-=20=E6=8F=90=E4=BE=9B=E5=BD=93=E5=89=8D=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E5=8F=8A=E6=8C=89=E6=97=A5=E6=9C=9F=E6=9F=A5=E8=AF=A2=E8=AF=8A?= =?UTF-8?q?=E6=96=AD=E5=8E=86=E5=8F=B2=E7=9A=84=E6=8E=A5=E5=8F=A3=20-=20?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E5=85=B3=E9=97=AD=E6=97=B6=E6=B8=85=E7=90=86?= =?UTF-8?q?=20TCP=20pending=20=E8=AE=A1=E6=95=B0=E5=B9=B6=E5=81=9C?= =?UTF-8?q?=E6=AD=A2=E6=97=A0=E6=95=88=E9=87=8D=E8=AF=95=20-=20=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E7=9B=B8=E5=85=B3=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E5=92=8C=E7=8E=B0=E5=9C=BA=E6=8E=92=E6=9F=A5=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../compiled/CompiledRuleCompiler.java | 11 +- ...ormalProcessorCompiledEquivalenceTest.java | 10 +- .../compiled/CompiledRuleCompilerTest.java | 16 +- .../docker-compose-serve-nw.yaml | 3 +- .../docker-compose-serve-prod.yaml | 2 + .../docker-compose/docker-compose-serve.yaml | 3 +- .../com/haobang/config/AppProperties.java | 24 + .../SyslogDiagnosticsController.java | 74 +++ .../SyslogDiagnosticsPersistence.java | 284 ++++++++++ .../diagnostics/SyslogIpMetricsRegistry.java | 496 ++++++++++++++++++ .../haobang/dispatch/SyslogWalDispatcher.java | 28 +- .../com/haobang/wal/SyslogWalService.java | 27 +- .../wal/SyslogWalStatusController.java | 59 +-- .../haobang/wal/SyslogWalStatusProvider.java | 78 +++ .../java/com/netty/SyslogMessageHandler.java | 126 ++++- .../src/main/java/com/netty/SyslogServer.java | 7 +- .../main/java/com/netty/SyslogServerBoth.java | 10 +- .../java/com/netty/SyslogServerLifecycle.java | 8 +- .../netty/SyslogTcpChannelInitializer.java | 8 +- .../netty/SyslogUdpChannelInitializer.java | 8 +- .../resources/application-prod.properties | 12 + .../src/main/resources/application.properties | 13 + .../SyslogDiagnosticsPersistenceTest.java | 90 ++++ .../SyslogIpMetricsRegistryTest.java | 107 ++++ .../dispatch/SyslogWalDispatcherTest.java | 8 + .../haobang/wal/SyslogWalDiagnosticsTest.java | 112 ++++ .../com/haobang/wal/SyslogWalServiceTest.java | 3 + .../SyslogMessageHandlerBackpressureTest.java | 41 +- .../v5/03-syslog-wal-diagnostics.md | 142 +++++ haobang-security-dm/upgrade-doc/v5/README.md | 3 + 30 files changed, 1722 insertions(+), 91 deletions(-) create mode 100644 haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogDiagnosticsController.java create mode 100644 haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogDiagnosticsPersistence.java create mode 100644 haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogIpMetricsRegistry.java create mode 100644 haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusProvider.java create mode 100644 haobang-security-dm/syslog-serve/src/test/java/com/haobang/diagnostics/SyslogDiagnosticsPersistenceTest.java create mode 100644 haobang-security-dm/syslog-serve/src/test/java/com/haobang/diagnostics/SyslogIpMetricsRegistryTest.java create mode 100644 haobang-security-dm/syslog-serve/src/test/java/com/haobang/wal/SyslogWalDiagnosticsTest.java create mode 100644 haobang-security-dm/upgrade-doc/v5/03-syslog-wal-diagnostics.md diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java index 55d61f1..11caaff 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java @@ -170,10 +170,10 @@ public class CompiledRuleCompiler { } private List compileFilters(JsonNode node, RuleSource source) { + node = decodeEmbeddedJson(node, source, "filters_params"); if (node == null || node.isNull()) return Collections.emptyList(); - // 旧执行器遇到 []、空字符串等非对象结构时捕获异常并按“无过滤条件”继续。 - // 字符串即使内容是 JSON 对象也不会被旧执行器二次解析，必须同样忽略。 - // 编译路径必须保持相同语义，不能让历史脏格式阻塞整条消息。 + // Fastjson 的 JSONObject.getJSONObject 会二次解析合法的内嵌 JSON 字符串， + // 因此编译路径也必须支持该历史存储格式。其余非对象结构仍按旧执行器语义忽略。 if (!node.isObject()) return Collections.emptyList(); JsonNode groups = node.get("expressions"); if (groups == null || groups.isNull()) return Collections.emptyList(); @@ -191,9 +191,10 @@ public class CompiledRuleCompiler { } private List compileCompletions(JsonNode node, RuleSource source) { + node = decodeEmbeddedJson(node, source, "complete_columns"); if (node == null || node.isNull()) return Collections.emptyList(); - // 旧补全服务对 {}、空字符串等非数组结构捕获异常后按“无补全配置”继续。 - // 字符串即使内容是 JSON 数组也不会被旧执行器二次解析，必须同样忽略。 + // Fastjson 的 JSONObject.getJSONArray 会二次解析合法的内嵌 JSON 字符串， + // 现场规则普遍沿用这种格式；其余非数组结构仍按旧执行器语义忽略。 if (!node.isArray()) return Collections.emptyList(); List result = new ArrayList<>(); for (JsonNode completion : node) { diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorCompiledEquivalenceTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorCompiledEquivalenceTest.java index f9fef7d..a391b46 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorCompiledEquivalenceTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorCompiledEquivalenceTest.java @@ -61,6 +61,7 @@ class LogNormalProcessorCompiledEquivalenceTest { assertEquals(legacyResult.getRows(), compiledResult.getRows()); assertEquals(1, compiledResult.getRows().size()); assertEquals(31L, compiledResult.getRows().get(0).get("normalize_rule_id")); + assertEquals("CN", compiledResult.getRows().get(0).get("country")); assertEquals(1, metrics.getParseMisses()); assertEquals(1, metrics.getParseHits()); assertEquals(2, metrics.getCompiledMisses()); @@ -98,10 +99,11 @@ class LogNormalProcessorCompiledEquivalenceTest { + "\"mappers\":[" + "{\"origin_field\":\"src_ip\",\"dest_field\":\"src_ip\",\"action\":{\"type\":\"equal\",\"param\":{}}}," + "{\"origin_field\":\"level\",\"dest_field\":\"severity\",\"action\":{\"type\":\"mapping\",\"param\":{\"mapping\":{\"high\":3}}}}]," - + "\"filters_params\":{\"expressions\":[]}," - + "\"complete_columns\":[{\"column_name\":\"country\",\"value\":\"CN\"," - + "\"filters_relation\":\"and\",\"conditions\":[{\"column_name\":\"src_ip\"," - + "\"op\":\"regexp\",\"value\":\"10\\\\..*\"}]}]}"; + + "\"filters_params\":\"{\\\"expressions\\\":[]}\"," + + "\"complete_columns\":\"[{\\\"column_name\\\":\\\"country\\\"," + + "\\\"value\\\":\\\"CN\\\",\\\"filters_relation\\\":\\\"and\\\"," + + "\\\"conditions\\\":[{\\\"column_name\\\":\\\"src_ip\\\"," + + "\\\"op\\\":\\\"regexp\\\",\\\"value\\\":\\\"10\\\\\\\\..*\\\"}]}]\"}"; return new RuleSource(id, name, name, "json", content, true, "0"); } diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/compiled/CompiledRuleCompilerTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/compiled/CompiledRuleCompilerTest.java index 2ea8fc1..f945cf1 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/compiled/CompiledRuleCompilerTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/compiled/CompiledRuleCompilerTest.java @@ -110,7 +110,7 @@ class CompiledRuleCompilerTest { } @Test - void decodesStringMappersButIgnoresStringFiltersAndCompletionsLikeLegacy() { + void decodesStringMappersFiltersAndCompletionsLikeLegacyFastjsonAccessors() { RuleSource legacyShape = source(1996235894300086273L, "embedded-json", "json", "{" + "\"mappers\":\"[{\\\"origin_field\\\":\\\"src\\\"," + "\\\"dest_field\\\":\\\"dest\\\"," @@ -124,8 +124,18 @@ class CompiledRuleCompilerTest { CompiledRule compiled = compiler.compile(legacyShape); assertEquals(1, compiled.getMapperByOriginField().get("src").size()); - assertTrue(compiled.getFilters().isEmpty()); - assertTrue(compiled.getCompletions().isEmpty()); + assertEquals(1, compiled.getFilters().size()); + assertEquals(1, compiled.getCompletions().size()); + + Map allowed = new java.util.LinkedHashMap<>(); + allowed.put("dest", "allowed"); + assertFalse(compiled.isFiltered(allowed)); + compiled.complete(allowed); + assertEquals("CN", allowed.get("country")); + + Map blocked = new java.util.LinkedHashMap<>(); + blocked.put("dest", "blocked"); + assertTrue(compiled.isFiltered(blocked)); } @Test diff --git a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-nw.yaml b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-nw.yaml index 7d5a66c..93ccbb0 100644 --- a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-nw.yaml +++ b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-nw.yaml @@ -23,6 +23,8 @@ services: - TZ=Asia/Shanghai # 探针标识 - DEVICE_COLLECT_ID=1 + - SYSLOG_DIAGNOSTICS_DIRECTORY=/app/logs/syslog-monitor + - SYSLOG_DIAGNOSTICS_RETENTION_DAYS=7 # 数据库配置 - SPRING_DATASOURCE_URL=jdbc:postgresql://117.72.68.72:54329/ecosys - SPRING_DATASOURCE_USERNAME=postgres @@ -76,4 +78,3 @@ services: networks: xdr-network: driver: bridge - diff --git a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml index 11aae30..4703c6c 100644 --- a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml +++ b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml @@ -25,6 +25,8 @@ services: - SYSLOG_CHARSET=UTF-8 # WAL 与应用日志共用宿主机挂载，但使用独立子目录，便于识别和清理。 - SYSLOG_WAL_DIRECTORY=/app/logs/syslog-wal + - SYSLOG_DIAGNOSTICS_DIRECTORY=/app/logs/syslog-monitor + - SYSLOG_DIAGNOSTICS_RETENTION_DAYS=7 # 128KiB 及以上走大日志 Topic，单条最多接纳 6MiB；WAL/Dispatcher 均按字节背压。 - SYSLOG_MAX_FRAME_LENGTH=6291456 - SYSLOG_MESSAGE_MAX_BYTES=6291456 diff --git a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve.yaml b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve.yaml index 74afea3..6eb6fda 100644 --- a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve.yaml +++ b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve.yaml @@ -23,6 +23,8 @@ services: - TZ=Asia/Shanghai # 探针标识 - DEVICE_COLLECT_ID=1 + - SYSLOG_DIAGNOSTICS_DIRECTORY=/app/logs/syslog-monitor + - SYSLOG_DIAGNOSTICS_RETENTION_DAYS=7 # 数据库配置 - SPRING_DATASOURCE_URL=jdbc:postgresql://117.72.68.72:54329/ecosys - SPRING_DATASOURCE_USERNAME=postgres @@ -76,4 +78,3 @@ services: networks: xdr-network: driver: bridge - diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java index 2e50865..cfcae0e 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java @@ -146,6 +146,30 @@ public class AppProperties { @Value("${syslog.status-log-interval-ms:60000}") private long syslogStatusLogIntervalMs; + // ========== 接入诊断历史 ========== + /** 独立于 Kafka/数据库保存诊断快照，避免故障链路同时带走排障证据。 */ + @Value("${syslog.diagnostics.enabled:true}") + private boolean syslogDiagnosticsEnabled; + + /** 必须位于宿主机持久化卷；生产默认映射到 /home/syslog/logs/syslog-monitor。 */ + @Value("${syslog.diagnostics.directory:${SYSLOG_DIAGNOSTICS_DIRECTORY:/data/syslog-monitor}}") + private String syslogDiagnosticsDirectory; + + @Value("${syslog.diagnostics.snapshot-interval-ms:60000}") + private long syslogDiagnosticsSnapshotIntervalMs; + + /** 保留今天及此前 6 个自然日。 */ + @Value("${syslog.diagnostics.retention-days:7}") + private int syslogDiagnosticsRetentionDays; + + /** 防止伪造 UDP 来源产生无界 IP 基数；超出的来源统一汇总到 __OTHER__。 */ + @Value("${syslog.diagnostics.max-source-ips:10000}") + private int syslogDiagnosticsMaxSourceIps; + + /** 网络线程只执行 offer；队列满时绝不阻塞接入线程。 */ + @Value("${syslog.diagnostics.event-queue-capacity:10000}") + private int syslogDiagnosticsEventQueueCapacity; + // ========== 旧 v1 原文归档精确导入（默认关闭） ========== @Value("${syslog.wal.legacy-import.enabled:false}") private boolean syslogWalLegacyImportEnabled; diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogDiagnosticsController.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogDiagnosticsController.java new file mode 100644 index 0000000..a64b025 --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogDiagnosticsController.java @@ -0,0 +1,74 @@ +package com.haobang.diagnostics; + +import com.haobang.wal.SyslogWalStatusProvider; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; +import org.springframework.format.annotation.DateTimeFormat; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.LocalDate; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.ArrayList; +import java.util.Map; + +/** 现场查询当前分 IP 计数，并按日期下载持久化诊断历史。 */ +@RestController +@RequestMapping("/api/syslog/wal/diagnostics") +public class SyslogDiagnosticsController { + private final SyslogWalStatusProvider statusProvider; + private final SyslogIpMetricsRegistry ipMetrics; + private final SyslogDiagnosticsPersistence persistence; + + public SyslogDiagnosticsController(SyslogWalStatusProvider statusProvider, + SyslogIpMetricsRegistry ipMetrics, + SyslogDiagnosticsPersistence persistence) { + this.statusProvider = statusProvider; + this.ipMetrics = ipMetrics; + this.persistence = persistence; + } + + @GetMapping("/current") + public Map current( + @RequestParam(name = "source_ip", required = false) String sourceIp) { + Map value = new LinkedHashMap<>(); + value.put("persistence", persistence.status()); + value.put("global", statusProvider.status()); + List> rows = ipMetrics.currentSnapshot(); + if (sourceIp != null && !sourceIp.trim().isEmpty()) { + List> filtered = new ArrayList<>(); + for (Map row : rows) { + if (sourceIp.trim().equals(row.get("source_ip"))) filtered.add(row); + } + rows = filtered; + } + value.put("ip_stats", rows); + return value; + } + + @GetMapping("/history") + public ResponseEntity history( + @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate date, + @RequestParam(defaultValue = "snapshots") String type) { + Path file; + if ("snapshots".equals(type)) file = persistence.snapshotFile(date); + else if ("ip".equals(type)) file = persistence.ipSnapshotFile(date); + else if ("events".equals(type)) file = persistence.eventFile(date); + else return ResponseEntity.badRequest().build(); + if (file == null || !Files.isRegularFile(file)) return ResponseEntity.notFound().build(); + Resource resource = new FileSystemResource(file.toFile()); + return ResponseEntity.ok() + .contentType(MediaType.parseMediaType("application/x-ndjson")) + .header(HttpHeaders.CONTENT_DISPOSITION, + "attachment; filename=\"" + file.getFileName() + "\"") + .body(resource); + } +} diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogDiagnosticsPersistence.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogDiagnosticsPersistence.java new file mode 100644 index 0000000..92bff4a --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogDiagnosticsPersistence.java @@ -0,0 +1,284 @@ +package com.haobang.diagnostics; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.haobang.config.AppProperties; +import com.haobang.wal.SyslogWalStatusProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardOpenOption; +import java.time.LocalDate; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** 将全局/分 IP 快照和网络异常事件写入独立 JSONL 文件。 */ +@Component +public class SyslogDiagnosticsPersistence { + private static final Logger log = LoggerFactory.getLogger(SyslogDiagnosticsPersistence.class); + private static final ZoneId ZONE = ZoneId.of("Asia/Shanghai"); + private static final String SNAPSHOT_PREFIX = "syslog-status-"; + private static final String EVENT_PREFIX = "syslog-events-"; + private static final String IP_SNAPSHOT_PREFIX = "syslog-ip-status-"; + + private final AppProperties properties; + private final SyslogIpMetricsRegistry ipMetrics; + private final SyslogWalStatusProvider statusProvider; + private final ObjectMapper objectMapper; + private final AtomicBoolean running = new AtomicBoolean(); + private final AtomicLong snapshotWriteFailures = new AtomicLong(); + private final AtomicLong eventWriteFailures = new AtomicLong(); + private final AtomicLong initializationFailures = new AtomicLong(); + private volatile String lastSnapshotTime; + private volatile String lastEventTime; + private volatile String lastWriteError; + private volatile LocalDate lastCleanupDate; + private Path rootDirectory; + private Path snapshotDirectory; + private Path eventDirectory; + private Path ipSnapshotDirectory; + private ScheduledExecutorService snapshotScheduler; + private Thread eventWriterThread; + + public SyslogDiagnosticsPersistence(AppProperties properties, + SyslogIpMetricsRegistry ipMetrics, + SyslogWalStatusProvider statusProvider, + ObjectMapper objectMapper) { + this.properties = properties; + this.ipMetrics = ipMetrics; + this.statusProvider = statusProvider; + this.objectMapper = objectMapper.copy().deactivateDefaultTyping(); + } + + @PostConstruct + public void start() { + if (!properties.isSyslogDiagnosticsEnabled()) { + log.warn("Syslog 诊断历史已关闭"); + return; + } + try { + validateConfiguration(); + rootDirectory = Paths.get(properties.getSyslogDiagnosticsDirectory()) + .toAbsolutePath().normalize(); + snapshotDirectory = rootDirectory.resolve("snapshots"); + eventDirectory = rootDirectory.resolve("events"); + ipSnapshotDirectory = rootDirectory.resolve("ip-snapshots"); + Files.createDirectories(snapshotDirectory); + Files.createDirectories(eventDirectory); + Files.createDirectories(ipSnapshotDirectory); + cleanupExpiredFiles(LocalDate.now(ZONE)); + } catch (Throwable failure) { + initializationFailures.incrementAndGet(); + lastWriteError = failure.toString(); + log.error("Syslog 诊断历史初始化失败；接入链路继续运行，诊断持久化不可用: directory={}", + properties.getSyslogDiagnosticsDirectory(), failure); + return; + } + running.set(true); + eventWriterThread = new Thread(this::eventWriterLoop, "syslog-diagnostics-event-writer"); + eventWriterThread.setDaemon(true); + eventWriterThread.start(); + snapshotScheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "syslog-diagnostics-snapshot-writer"); + thread.setDaemon(true); + return thread; + }); + snapshotScheduler.scheduleWithFixedDelay(this::writeSnapshotSafely, + 1000L, properties.getSyslogDiagnosticsSnapshotIntervalMs(), TimeUnit.MILLISECONDS); + log.info("Syslog 诊断历史已启用: directory={}, snapshotIntervalMs={}, retentionDays={}, maxSourceIps={}", + rootDirectory, properties.getSyslogDiagnosticsSnapshotIntervalMs(), + properties.getSyslogDiagnosticsRetentionDays(), properties.getSyslogDiagnosticsMaxSourceIps()); + } + + private void eventWriterLoop() { + Map pending = null; + while (running.get() || pending != null || hasQueuedEvent()) { + try { + if (pending == null) { + pending = running.get() + ? ipMetrics.pollEvent(1000L, TimeUnit.MILLISECONDS) + : ipMetrics.pollEvent(); + } + if (pending != null) { + writeEvent(pending); + pending = null; + } + } catch (InterruptedException interrupted) { + if (running.get()) Thread.currentThread().interrupt(); + } catch (Throwable failure) { + recordFailure(false, failure); + try { Thread.sleep(1000L); } + catch (InterruptedException interrupted) { + if (running.get()) Thread.currentThread().interrupt(); + } + } + } + } + + private boolean hasQueuedEvent() { + return ipMetrics.getEventQueueSize() > 0; + } + + void writeSnapshotSafely() { + if (!running.get()) return; + try { + ZonedDateTime now = ZonedDateTime.now(ZONE); + Map snapshot = new LinkedHashMap<>(); + snapshot.put("record_type", "snapshot"); + snapshot.put("snapshot_time", now.toInstant().toString()); + snapshot.put("instance_id", ipMetrics.getInstanceId()); + snapshot.put("process_start_time", ipMetrics.getProcessStartTime().toString()); + snapshot.put("device_collect_id", properties.getAppServiceDeviceCollectId()); + snapshot.put("device_collect_name", properties.getAppServiceDeviceCollectName()); + snapshot.put("global", statusProvider.status()); + SyslogIpMetricsRegistry.SnapshotBatch batch = ipMetrics.snapshotWithDeltas(); + snapshot.put("ip_stats_count", batch.getRows().size()); + appendJson(snapshotFile(now.toLocalDate()), snapshot); + appendIpSnapshots(ipSnapshotFile(now.toLocalDate()), now, batch.getRows()); + ipMetrics.commitSnapshot(batch); + lastSnapshotTime = now.toInstant().toString(); + cleanupExpiredFiles(now.toLocalDate()); + } catch (Throwable failure) { + recordFailure(true, failure); + } + } + + private void writeEvent(Map event) throws IOException { + LocalDate date = LocalDate.now(ZONE); + appendJson(eventFile(date), event); + Object eventTime = event.get("event_time"); + lastEventTime = eventTime == null ? ZonedDateTime.now(ZONE).toInstant().toString() + : String.valueOf(eventTime); + } + + private void appendJson(Path path, Map value) throws IOException { + byte[] json = (objectMapper.writeValueAsString(value) + System.lineSeparator()) + .getBytes(StandardCharsets.UTF_8); + Files.write(path, json, StandardOpenOption.CREATE, StandardOpenOption.WRITE, + StandardOpenOption.APPEND); + } + + private void appendIpSnapshots(Path path, ZonedDateTime now, + List> rows) throws IOException { + if (rows.isEmpty()) return; + StringBuilder output = new StringBuilder(rows.size() * 512); + for (Map row : rows) { + Map record = new LinkedHashMap<>(); + record.put("record_type", "ip_snapshot"); + record.put("snapshot_time", now.toInstant().toString()); + record.put("instance_id", ipMetrics.getInstanceId()); + record.put("process_start_time", ipMetrics.getProcessStartTime().toString()); + record.put("device_collect_id", properties.getAppServiceDeviceCollectId()); + record.put("device_collect_name", properties.getAppServiceDeviceCollectName()); + record.putAll(row); + output.append(objectMapper.writeValueAsString(record)).append(System.lineSeparator()); + } + Files.write(path, output.toString().getBytes(StandardCharsets.UTF_8), + StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.APPEND); + } + + private void cleanupExpiredFiles(LocalDate today) throws IOException { + if (today.equals(lastCleanupDate)) return; + LocalDate oldestRetained = today.minusDays(properties.getSyslogDiagnosticsRetentionDays() - 1L); + cleanupDirectory(snapshotDirectory, SNAPSHOT_PREFIX, oldestRetained); + cleanupDirectory(eventDirectory, EVENT_PREFIX, oldestRetained); + cleanupDirectory(ipSnapshotDirectory, IP_SNAPSHOT_PREFIX, oldestRetained); + lastCleanupDate = today; + } + + private void cleanupDirectory(Path directory, String prefix, LocalDate oldestRetained) throws IOException { + try (DirectoryStream files = Files.newDirectoryStream(directory, prefix + "*.jsonl")) { + for (Path file : files) { + LocalDate fileDate = parseDate(file.getFileName().toString(), prefix); + if (fileDate != null && fileDate.isBefore(oldestRetained)) Files.deleteIfExists(file); + } + } + } + + private LocalDate parseDate(String fileName, String prefix) { + if (!fileName.startsWith(prefix) || !fileName.endsWith(".jsonl")) return null; + String date = fileName.substring(prefix.length(), fileName.length() - ".jsonl".length()); + try { return LocalDate.parse(date); } + catch (RuntimeException ignored) { return null; } + } + + private void recordFailure(boolean snapshot, Throwable failure) { + long count = snapshot ? snapshotWriteFailures.incrementAndGet() : eventWriteFailures.incrementAndGet(); + lastWriteError = failure.toString(); + if (count == 1L || count % 100L == 0L) { + log.error("Syslog 诊断{}写入失败: failures={}", snapshot ? "快照" : "事件", count, failure); + } + } + + private void validateConfiguration() { + if (properties.getSyslogDiagnosticsSnapshotIntervalMs() <= 0L + || properties.getSyslogDiagnosticsRetentionDays() <= 0 + || properties.getSyslogDiagnosticsMaxSourceIps() <= 0 + || properties.getSyslogDiagnosticsEventQueueCapacity() <= 0) { + throw new IllegalArgumentException("Syslog 诊断间隔、保留天数、IP 上限和事件队列容量必须大于 0"); + } + if (properties.getSyslogDiagnosticsDirectory() == null + || properties.getSyslogDiagnosticsDirectory().trim().isEmpty()) { + throw new IllegalArgumentException("syslog.diagnostics.directory 不能为空"); + } + } + + public Map status() { + Map value = new LinkedHashMap<>(); + value.put("enabled", properties.isSyslogDiagnosticsEnabled()); + value.put("active", running.get()); + value.put("directory", rootDirectory == null ? null : rootDirectory.toString()); + value.put("retention_days", properties.getSyslogDiagnosticsRetentionDays()); + value.put("snapshot_interval_ms", properties.getSyslogDiagnosticsSnapshotIntervalMs()); + value.put("last_snapshot_time", lastSnapshotTime); + value.put("last_event_time", lastEventTime); + value.put("snapshot_write_failures", snapshotWriteFailures.get()); + value.put("event_write_failures", eventWriteFailures.get()); + value.put("initialization_failures", initializationFailures.get()); + value.put("event_queue_dropped", ipMetrics.getEventQueueDropped()); + value.put("last_write_error", lastWriteError); + return value; + } + + public Path snapshotFile(LocalDate date) { + return snapshotDirectory == null ? null + : snapshotDirectory.resolve(SNAPSHOT_PREFIX + date + ".jsonl"); + } + + public Path eventFile(LocalDate date) { + return eventDirectory == null ? null + : eventDirectory.resolve(EVENT_PREFIX + date + ".jsonl"); + } + + public Path ipSnapshotFile(LocalDate date) { + return ipSnapshotDirectory == null ? null + : ipSnapshotDirectory.resolve(IP_SNAPSHOT_PREFIX + date + ".jsonl"); + } + + @PreDestroy + public void shutdown() { + if (!running.compareAndSet(true, false)) return; + if (snapshotScheduler != null) snapshotScheduler.shutdownNow(); + if (eventWriterThread != null) { + eventWriterThread.interrupt(); + try { eventWriterThread.join(5000L); } + catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); } + } + } +} diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogIpMetricsRegistry.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogIpMetricsRegistry.java new file mode 100644 index 0000000..0b4d7f3 --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/diagnostics/SyslogIpMetricsRegistry.java @@ -0,0 +1,496 @@ +package com.haobang.diagnostics; + +import com.common.entity.DeviceDevice; +import com.haobang.config.AppProperties; +import io.netty.handler.codec.TooLongFrameException; +import org.springframework.stereotype.Component; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.LongAdder; + +/** + * 进程内、按来源 IP 聚合的接入诊断指标。 + * + *

所有接入热路径只做无锁计数和非阻塞队列 offer。诊断文件 IO 由 + * {@link SyslogDiagnosticsPersistence} 的独立线程完成。

+ */ +@Component +public class SyslogIpMetricsRegistry { + public static final String OVERFLOW_SOURCE_IP = "__OTHER__"; + + private final ConcurrentHashMap metrics = new ConcurrentHashMap<>(); + private final ConcurrentHashMap previousSnapshot = new ConcurrentHashMap<>(); + private final IpMetrics overflow = new IpMetrics(OVERFLOW_SOURCE_IP); + private final ArrayBlockingQueue> eventQueue; + private final LongAdder eventQueueDropped = new LongAdder(); + private final AtomicLong tcpPendingFramesCurrent = new AtomicLong(); + private final AtomicLong tcpPendingFramesMax = new AtomicLong(); + private final int maxSourceIps; + private final int deviceCollectId; + private final String deviceCollectName; + private final String instanceId = UUID.randomUUID().toString(); + private final Instant processStartTime = Instant.now(); + + public SyslogIpMetricsRegistry(AppProperties properties) { + this.maxSourceIps = Math.max(1, properties.getSyslogDiagnosticsMaxSourceIps()); + this.deviceCollectId = properties.getAppServiceDeviceCollectId(); + this.deviceCollectName = properties.getAppServiceDeviceCollectName(); + this.eventQueue = new ArrayBlockingQueue<>( + Math.max(1, properties.getSyslogDiagnosticsEventQueueCapacity())); + } + + /** + * 记录收到的报文；如果该 IP 此前发生过网络接入异常，返回本次恢复信息。 + * + *

这里不判断设备是否离线，也不生成中断统计，只用于让接入层在异常后的第一条报文到达时 + * 打印一次恢复日志。

+ */ + public NetworkRecovery recordNetworkReceived(String sourceIp, int sourcePort, String protocol) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, protocol); + value.networkReceived.increment(); + long receivedAt = System.currentTimeMillis(); + value.lastReceiveTime.set(receivedAt); + // __OTHER__ 会合并多个来源，不能用其中一个 IP 的异常推断另一个 IP 已恢复。 + if (value == overflow) return null; + long exceptionAt = value.pendingRecoveryExceptionTime.getAndSet(0L); + return exceptionAt <= 0L ? null : new NetworkRecovery(exceptionAt, receivedAt); + } + + public void recordWalAppended(String sourceIp, Integer sourcePort, String protocol) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort == null ? 0 : sourcePort, protocol); + value.walAppended.increment(); + } + + public void recordWalBackpressure(String sourceIp, int sourcePort, String protocol) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, protocol); + value.walAdmissionBackpressure.increment(); + } + + public void recordWalUnavailable(String sourceIp, Integer sourcePort, String protocol) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort == null ? 0 : sourcePort, protocol); + value.walAdmissionUnavailable.increment(); + } + + public void recordTcpPendingAdded(String sourceIp, int sourcePort) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, "TCP"); + long current = tcpPendingFramesCurrent.incrementAndGet(); + updateMax(tcpPendingFramesMax, current); + long ipCurrent = value.tcpPendingFramesCurrent.incrementAndGet(); + updateMax(value.tcpPendingFramesMax, ipCurrent); + } + + public void recordTcpPendingRemoved(String sourceIp, int sourcePort) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, "TCP"); + decrementIfPositive(tcpPendingFramesCurrent); + decrementIfPositive(value.tcpPendingFramesCurrent); + } + + public void recordKafkaAck(String sourceIp, int sourcePort, String protocol, DeviceDevice device) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, protocol); + value.assignDevice(device); + value.kafkaAcked.increment(); + } + + public void recordInvalidDeviceDropped(String sourceIp, String protocol) { + IpMetrics value = metric(sourceIp); + value.touch(0, protocol); + value.invalidDeviceDropped.increment(); + } + + public void recordDeviceLookupFailed(String sourceIp, String protocol) { + IpMetrics value = metric(sourceIp); + value.touch(0, protocol); + value.deviceLookupFailed.increment(); + } + + public void recordOversizedDropped(String sourceIp, int sourcePort, String protocol, int chars) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, protocol); + value.oversizedDropped.increment(); + enqueueEvent(event("OVERSIZED_DROPPED", sourceIp, sourcePort, protocol, + "chars=" + chars, null, value)); + } + + public void recordUdpDropped(String sourceIp, int sourcePort) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, "UDP"); + value.udpDropped.increment(); + } + + public void recordTcpPause(String sourceIp, int sourcePort) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, "TCP"); + value.tcpPauseTotal.increment(); + } + + public void recordDispatchRetry(String sourceIp, int sourcePort, String protocol) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, protocol); + value.dispatchRetry.increment(); + } + + public void recordTerminalDropped(String sourceIp, int sourcePort, String protocol, + DeviceDevice device, String reason) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, protocol); + value.assignDevice(device); + value.dispatchTerminalDropped.increment(); + enqueueEvent(event("DISPATCH_TERMINAL_DROPPED", sourceIp, sourcePort, protocol, + abbreviate(reason, 1024), null, value)); + } + + public void recordNetworkException(String sourceIp, int sourcePort, String protocol, + Throwable failure) { + IpMetrics value = metric(sourceIp); + value.touch(sourcePort, protocol); + value.networkException.increment(); + long exceptionAt = System.currentTimeMillis(); + value.lastExceptionTime.set(exceptionAt); + // 多次连续异常只保留第一次，恢复日志才能反映从首次发现异常到重新收到数据的间隔。 + if (value != overflow) value.pendingRecoveryExceptionTime.compareAndSet(0L, exceptionAt); + boolean tcp = "TCP".equalsIgnoreCase(protocol); + String type = classify(failure, tcp); + if (tcp) { + value.tcpNetworkException.increment(); + if ("TOO_LONG_FRAME".equals(type)) value.tcpTooLongFrame.increment(); + else if ("CONNECTION_RESET".equals(type)) value.tcpConnectionReset.increment(); + else value.tcpOtherException.increment(); + } else { + value.udpNetworkException.increment(); + } + Throwable root = rootCause(failure); + enqueueEvent(event(type, sourceIp, sourcePort, protocol, + root == null ? null : abbreviate(root.getMessage(), 1024), + root == null ? null : root.getClass().getName(), value)); + } + + /** 当前累计值；读取接口调用不会改变分钟增量基线。 */ + public List> currentSnapshot() { + return snapshot(false); + } + + /** 持久化线程专用；返回累计值并推进下一分钟的 delta 基线。 */ + public List> snapshotAndAdvanceDeltas() { + SnapshotBatch batch = snapshotWithDeltas(); + commitSnapshot(batch); + return batch.getRows(); + } + + /** 先捕获一致的 delta，只有两个诊断文件都写成功后才由持久化线程提交基线。 */ + public SnapshotBatch snapshotWithDeltas() { + List values = sortedMetrics(); + List> rows = new ArrayList<>(values.size()); + Map baselines = new LinkedHashMap<>(); + for (IpMetrics value : values) { + MetricValues current = value.values(); + rows.add(value.toMap(current, current.minus(previousSnapshot.get(value.sourceIp)))); + baselines.put(value.sourceIp, current); + } + return new SnapshotBatch(rows, baselines); + } + + public void commitSnapshot(SnapshotBatch batch) { + if (batch != null) previousSnapshot.putAll(batch.baselines); + } + + private List> snapshot(boolean advanceDeltas) { + List values = sortedMetrics(); + List> result = new ArrayList<>(values.size()); + for (IpMetrics value : values) { + MetricValues current = value.values(); + MetricValues previous = previousSnapshot.get(value.sourceIp); + result.add(value.toMap(current, advanceDeltas ? current.minus(previous) : null)); + if (advanceDeltas) previousSnapshot.put(value.sourceIp, current); + } + return result; + } + + private List sortedMetrics() { + List values = new ArrayList<>(metrics.values()); + if (!overflow.isEmpty()) values.add(overflow); + Collections.sort(values, Comparator.comparing(value -> value.sourceIp)); + return values; + } + + public Map summary() { + Map value = new LinkedHashMap<>(); + value.put("instance_id", instanceId); + value.put("process_start_time", processStartTime.toString()); + value.put("device_collect_id", deviceCollectId); + value.put("device_collect_name", deviceCollectName); + value.put("tracked_source_ips", metrics.size()); + value.put("max_source_ips", maxSourceIps); + value.put("event_queue_size", eventQueue.size()); + value.put("event_queue_dropped", eventQueueDropped.sum()); + value.put("network_exception_total", sum(Metric.NETWORK_EXCEPTION)); + value.put("tcp_network_exception_total", sum(Metric.TCP_NETWORK_EXCEPTION)); + value.put("udp_network_exception_total", sum(Metric.UDP_NETWORK_EXCEPTION)); + value.put("tcp_connection_reset_total", sum(Metric.TCP_CONNECTION_RESET)); + value.put("tcp_too_long_frame_total", sum(Metric.TCP_TOO_LONG_FRAME)); + value.put("tcp_other_exception_total", sum(Metric.TCP_OTHER_EXCEPTION)); + value.put("tcp_pending_frames_current", tcpPendingFramesCurrent.get()); + value.put("tcp_pending_frames_max", tcpPendingFramesMax.get()); + return value; + } + + public Map pollEvent(long timeout, TimeUnit unit) throws InterruptedException { + return eventQueue.poll(timeout, unit); + } + + public Map pollEvent() { return eventQueue.poll(); } + public String getInstanceId() { return instanceId; } + public Instant getProcessStartTime() { return processStartTime; } + public int getEventQueueSize() { return eventQueue.size(); } + public long getEventQueueDropped() { return eventQueueDropped.sum(); } + public long getTcpPendingFramesCurrent() { return tcpPendingFramesCurrent.get(); } + public long getTcpPendingFramesMax() { return tcpPendingFramesMax.get(); } + + public static final class NetworkRecovery { + private final long exceptionAtEpochMillis; + private final long recoveredAtEpochMillis; + + private NetworkRecovery(long exceptionAtEpochMillis, long recoveredAtEpochMillis) { + this.exceptionAtEpochMillis = exceptionAtEpochMillis; + this.recoveredAtEpochMillis = recoveredAtEpochMillis; + } + + public Instant getExceptionAt() { return Instant.ofEpochMilli(exceptionAtEpochMillis); } + public Instant getRecoveredAt() { return Instant.ofEpochMilli(recoveredAtEpochMillis); } + public long getElapsedMillis() { + return Math.max(0L, recoveredAtEpochMillis - exceptionAtEpochMillis); + } + } + + private IpMetrics metric(String sourceIp) { + String normalized = sourceIp == null || sourceIp.trim().isEmpty() ? "unknown" : sourceIp.trim(); + IpMetrics existing = metrics.get(normalized); + if (existing != null) return existing; + synchronized (metrics) { + existing = metrics.get(normalized); + if (existing != null) return existing; + if (metrics.size() >= maxSourceIps) return overflow; + IpMetrics created = new IpMetrics(normalized); + metrics.put(normalized, created); + return created; + } + } + + private void enqueueEvent(Map event) { + if (!eventQueue.offer(event)) eventQueueDropped.increment(); + } + + private static void updateMax(AtomicLong maximum, long candidate) { + long current = maximum.get(); + while (candidate > current && !maximum.compareAndSet(current, candidate)) { + current = maximum.get(); + } + } + + private static void decrementIfPositive(AtomicLong value) { + long current = value.get(); + while (current > 0L && !value.compareAndSet(current, current - 1L)) { + current = value.get(); + } + } + + private Map event(String eventType, String sourceIp, int sourcePort, + String protocol, String message, String exceptionClass, + IpMetrics metrics) { + Map value = new LinkedHashMap<>(); + value.put("record_type", "event"); + value.put("event_time", Instant.now().toString()); + value.put("instance_id", instanceId); + value.put("device_collect_id", deviceCollectId); + value.put("device_collect_name", deviceCollectName); + value.put("event_type", eventType); + value.put("source_ip", sourceIp); + value.put("source_port", sourcePort); + value.put("protocol", protocol); + value.put("device_id", metrics.deviceId.get() == 0 ? null : metrics.deviceId.get()); + value.put("device_name", metrics.deviceName); + value.put("exception_class", exceptionClass); + value.put("message", message); + return value; + } + + private long sum(Metric metric) { + long total = metric.value(overflow); + for (IpMetrics value : metrics.values()) total += metric.value(value); + return total; + } + + private static String classify(Throwable failure, boolean tcp) { + if (!tcp) return "UDP_NETWORK_EXCEPTION"; + for (Throwable current = failure; current != null; current = current.getCause()) { + if (current instanceof TooLongFrameException) return "TOO_LONG_FRAME"; + String text = String.valueOf(current.getMessage()).toLowerCase(); + if (text.contains("connection reset") || text.contains("broken pipe") + || text.contains("forcibly closed")) return "CONNECTION_RESET"; + } + return "TCP_NETWORK_EXCEPTION"; + } + + private static Throwable rootCause(Throwable failure) { + Throwable current = failure; + while (current != null && current.getCause() != null && current.getCause() != current) { + current = current.getCause(); + } + return current; + } + + private static String abbreviate(String value, int limit) { + if (value == null || value.length() <= limit) return value; + return value.substring(0, limit); + } + + private enum Metric { + NETWORK_EXCEPTION { long value(IpMetrics value) { return value.networkException.sum(); } }, + TCP_NETWORK_EXCEPTION { long value(IpMetrics value) { return value.tcpNetworkException.sum(); } }, + UDP_NETWORK_EXCEPTION { long value(IpMetrics value) { return value.udpNetworkException.sum(); } }, + TCP_CONNECTION_RESET { long value(IpMetrics value) { return value.tcpConnectionReset.sum(); } }, + TCP_TOO_LONG_FRAME { long value(IpMetrics value) { return value.tcpTooLongFrame.sum(); } }, + TCP_OTHER_EXCEPTION { long value(IpMetrics value) { return value.tcpOtherException.sum(); } }; + abstract long value(IpMetrics value); + } + + public static final class SnapshotBatch { + private final List> rows; + private final Map baselines; + + private SnapshotBatch(List> rows, + Map baselines) { + this.rows = rows; + this.baselines = baselines; + } + + public List> getRows() { return rows; } + } + + private static final class IpMetrics { + private final String sourceIp; + private final Set protocols = ConcurrentHashMap.newKeySet(); + private final AtomicInteger lastSourcePort = new AtomicInteger(); + private final AtomicInteger deviceId = new AtomicInteger(); + private volatile String deviceName; + private final AtomicLong lastReceiveTime = new AtomicLong(); + private final AtomicLong lastExceptionTime = new AtomicLong(); + private final AtomicLong pendingRecoveryExceptionTime = new AtomicLong(); + private final AtomicLong tcpPendingFramesCurrent = new AtomicLong(); + private final AtomicLong tcpPendingFramesMax = new AtomicLong(); + private final LongAdder networkReceived = new LongAdder(); + private final LongAdder walAppended = new LongAdder(); + private final LongAdder walAdmissionBackpressure = new LongAdder(); + private final LongAdder walAdmissionUnavailable = new LongAdder(); + private final LongAdder kafkaAcked = new LongAdder(); + private final LongAdder invalidDeviceDropped = new LongAdder(); + private final LongAdder deviceLookupFailed = new LongAdder(); + private final LongAdder oversizedDropped = new LongAdder(); + private final LongAdder udpDropped = new LongAdder(); + private final LongAdder tcpPauseTotal = new LongAdder(); + private final LongAdder networkException = new LongAdder(); + private final LongAdder tcpNetworkException = new LongAdder(); + private final LongAdder udpNetworkException = new LongAdder(); + private final LongAdder tcpConnectionReset = new LongAdder(); + private final LongAdder tcpTooLongFrame = new LongAdder(); + private final LongAdder tcpOtherException = new LongAdder(); + private final LongAdder dispatchTerminalDropped = new LongAdder(); + private final LongAdder dispatchRetry = new LongAdder(); + + private IpMetrics(String sourceIp) { this.sourceIp = sourceIp; } + + private void touch(int sourcePort, String protocol) { + if (sourcePort > 0) lastSourcePort.set(sourcePort); + if (protocol != null && !protocol.trim().isEmpty()) protocols.add(protocol); + } + + private void assignDevice(DeviceDevice device) { + if (device == null) return; + if (device.getId() != null) deviceId.set(device.getId()); + if (device.getName() != null) deviceName = device.getName(); + } + + private MetricValues values() { + return new MetricValues(networkReceived.sum(), walAppended.sum(), + walAdmissionBackpressure.sum(), walAdmissionUnavailable.sum(), kafkaAcked.sum(), + invalidDeviceDropped.sum(), deviceLookupFailed.sum(), oversizedDropped.sum(), + udpDropped.sum(), tcpPauseTotal.sum(), networkException.sum(), + tcpNetworkException.sum(), udpNetworkException.sum(), + tcpConnectionReset.sum(), tcpTooLongFrame.sum(), tcpOtherException.sum(), + dispatchTerminalDropped.sum(), dispatchRetry.sum()); + } + + private Map toMap(MetricValues total, MetricValues delta) { + Map value = new LinkedHashMap<>(); + value.put("source_ip", sourceIp); + value.put("source_port", lastSourcePort.get()); + value.put("protocols", new ArrayList<>(protocols)); + value.put("device_id", deviceId.get() == 0 ? null : deviceId.get()); + value.put("device_name", deviceName); + value.put("last_receive_time", instant(lastReceiveTime.get())); + value.put("last_exception_time", instant(lastExceptionTime.get())); + value.put("tcp_pending_frames_current", tcpPendingFramesCurrent.get()); + value.put("tcp_pending_frames_max", tcpPendingFramesMax.get()); + value.putAll(total.toMap()); + if (delta != null) value.put("delta", delta.toMap()); + return value; + } + + private boolean isEmpty() { return values().isZero(); } + private static String instant(long epochMillis) { + return epochMillis <= 0L ? null : Instant.ofEpochMilli(epochMillis).toString(); + } + } + + private static final class MetricValues { + private final long[] values; + + private MetricValues(long... values) { this.values = values; } + + private MetricValues minus(MetricValues previous) { + long[] result = new long[values.length]; + for (int index = 0; index < values.length; index++) { + long old = previous == null ? 0L : previous.values[index]; + result[index] = Math.max(0L, values[index] - old); + } + return new MetricValues(result); + } + + private boolean isZero() { + for (long value : values) if (value != 0L) return false; + return true; + } + + private Map toMap() { + String[] names = {"network_received", "wal_appended", "wal_admission_backpressure", + "wal_admission_unavailable", "kafka_acked", + "invalid_device_dropped", "device_lookup_failed", "oversized_dropped", + "udp_dropped", "tcp_pause_total", "network_exception_total", + "tcp_network_exception_total", "udp_network_exception_total", + "tcp_connection_reset_total", "tcp_too_long_frame_total", + "tcp_other_exception_total", "dispatch_terminal_dropped", "dispatch_retry"}; + Map result = new LinkedHashMap<>(); + for (int index = 0; index < names.length; index++) result.put(names[index], values[index]); + return result; + } + } +} diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java index 71212fd..4c343bd 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java @@ -1,6 +1,7 @@ package com.haobang.dispatch; import com.haobang.config.AppProperties; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import com.haobang.wal.SyslogWalRecord; import com.haobang.wal.SyslogWalService; import com.kafka.kafkaProducer; @@ -8,6 +9,7 @@ import com.netty.SyslogMessageHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; +import org.springframework.beans.factory.annotation.Autowired; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -51,6 +53,7 @@ public class SyslogWalDispatcher { private final LongAdder retryCount = new LongAdder(); private final LongAdder largeDispatched = new LongAdder(); private final AtomicLong inflight = new AtomicLong(); + private SyslogIpMetricsRegistry ipMetrics; private Semaphore permits; private ByteBudget byteBudget; @@ -71,6 +74,11 @@ public class SyslogWalDispatcher { this.properties = properties; } + @Autowired(required = false) + void setIpMetrics(SyslogIpMetricsRegistry ipMetrics) { + this.ipMetrics = ipMetrics; + } + @PostConstruct public void start() throws Exception { validateConfiguration(); @@ -157,6 +165,10 @@ public class SyslogWalDispatcher { walService.recordTerminal(record, resolution.getType().name(), resolution.getReason()); } else { terminalDropped.increment(); + if (ipMetrics != null) { + ipMetrics.recordTerminalDropped(record.getSourceIp(), record.getSourcePort(), + record.getProtocol(), resolution.getDevice(), resolution.getReason()); + } } terminalFiltered.increment(); markComplete(record); @@ -200,6 +212,10 @@ public class SyslogWalDispatcher { .whenComplete((delivery, failure) -> { if (failure == null) { kafkaAcked.increment(); + if (ipMetrics != null) { + ipMetrics.recordKafkaAck(record.getSourceIp(), record.getSourcePort(), + record.getProtocol(), resolution.getDevice()); + } if (large) largeDispatched.increment(); markComplete(record); } else { @@ -214,6 +230,9 @@ public class SyslogWalDispatcher { private void scheduleRetry(SyslogWalRecord record, int attempt, String stage, Throwable failure) { retryCount.increment(); + if (ipMetrics != null) { + ipMetrics.recordDispatchRetry(record.getSourceIp(), record.getSourcePort(), record.getProtocol()); + } long retries = retryCount.sum(); if (attempt == 1 || retries % 100 == 0) { log.warn("WAL 事件投递失败，将保留原 event_id 重试: eventId={}, sequence={}, stage={}, " @@ -261,18 +280,21 @@ public class SyslogWalDispatcher { private void logStatus() { if (!running.get()) return; - log.info("Syslog WAL状态: networkReceived={}, appended={}, admissionOversized={}, walPendingApprox={}, walPendingBytes={}, walQueue={}/{}, durableSequence={}, " + log.info("Syslog WAL状态: networkReceived={}, appended={}, admissionOversized={}, admissionUnavailable={}, " + + "walPendingApprox={}, walPendingBytes={}/{}, walQueue={}/{}, durableSequence={}, " + "checkpointSequence={}, kafkaInflight={}, kafkaInflightBytes={}, kafkaAcked={}, largeDispatched={}, " + "terminalRecordEnabled={}, terminalFiltered={}, terminalDropped={}, " - + "retry={}, tcpPauseTotal={}, udpDropped={}, deviceCacheSize={}, deviceCacheHits={}, " + + "retry={}, tcpPauseTotal={}, tcpPendingFrames={}/{}, udpDropped={}, deviceCacheSize={}, deviceCacheHits={}, " + "deviceCacheMisses={}", SyslogMessageHandler.getNetworkReceivedCount(), walService.getAppendedCount(), - walService.getOversizedCount(), walService.getPendingRecordsApprox(), walService.getPendingAppendBytes(), + walService.getOversizedCount(), walService.getUnavailableCount(), + walService.getPendingRecordsApprox(), walService.getPendingAppendBytes(), walService.getMaxPendingBytes(), walService.getQueueSize(), walService.getQueueCapacity(), walService.getDurableSequence(), walService.getCheckpointSequence(), inflight.get(), byteBudget == null ? 0L : byteBudget.used(), kafkaAcked.sum(), largeDispatched.sum(), properties.isSyslogDispatchTerminalRecordEnabled(), terminalFiltered.sum(), terminalDropped.sum(), retryCount.sum(), SyslogMessageHandler.getTcpPauseCount(), + SyslogMessageHandler.getTcpPendingFramesCurrent(), SyslogMessageHandler.getTcpPendingFramesMax(), SyslogMessageHandler.getUdpDroppedCount(), deviceResolver.getCacheSize(), deviceResolver.getHitCount(), deviceResolver.getMissCount()); } diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalService.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalService.java index bf7918c..baa76e0 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalService.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalService.java @@ -3,6 +3,7 @@ package com.haobang.wal; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.haobang.config.AppProperties; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -70,12 +71,14 @@ public class SyslogWalService { private final ObjectMapper objectMapper; private final Charset syslogCharset; private final ForceOperation forceOperation; + private SyslogIpMetricsRegistry ipMetrics; private final PointTimeAllocator pointTimeAllocator = new PointTimeAllocator(); private final AtomicBoolean initialized = new AtomicBoolean(); private final AtomicLong nextSequence = new AtomicLong(); private final AtomicLong durableSequence = new AtomicLong(); private final LongAdder appended = new LongAdder(); private final LongAdder admissionBackpressure = new LongAdder(); + private final LongAdder admissionUnavailable = new LongAdder(); private final LongAdder admissionOversized = new LongAdder(); private final AtomicLong pendingAppendBytes = new AtomicLong(); private final LongAdder quarantined = new LongAdder(); @@ -147,6 +150,12 @@ public class SyslogWalService { this.forceOperation = forceOperation; } + /** 测试可不注册诊断 Bean；生产环境由 Spring 注入。 */ + @Autowired(required = false) + void setIpMetrics(SyslogIpMetricsRegistry ipMetrics) { + this.ipMetrics = ipMetrics; + } + @PostConstruct public void initialize() { if (!initialized.compareAndSet(false, true)) return; @@ -232,6 +241,11 @@ public class SyslogWalService { } refreshDiskState(false); if (!running || fatalFailure != null || diskState == DiskState.CRITICAL) { + admissionUnavailable.increment(); + if (ipMetrics != null) { + ipMetrics.recordWalUnavailable(event.getSourceIp(), event.getSourcePort(), + event.getTransport()); + } return AppendAttempt.rejected(AdmissionStatus.UNAVAILABLE); } int rawBytes = event.getRawMessage().getBytes(event.getCharset()).length; @@ -250,7 +264,16 @@ public class SyslogWalService { admissionBackpressure.increment(); return AppendAttempt.rejected(AdmissionStatus.BACKPRESSURE); } - return AppendAttempt.accepted(pending.getDurableFuture()); + AppendAttempt accepted = AppendAttempt.accepted(pending.getDurableFuture()); + if (ipMetrics != null) { + accepted.getDurableStage().whenComplete((receipt, failure) -> { + if (failure == null) { + ipMetrics.recordWalAppended(event.getSourceIp(), event.getSourcePort(), + event.getTransport()); + } + }); + } + return accepted; } public boolean shouldPauseTcp() { @@ -281,8 +304,10 @@ public class SyslogWalService { public int getQueueCapacity() { return properties.getSyslogWalQueueCapacity(); } public long getAppendedCount() { return appended.sum(); } public long getBackpressureCount() { return admissionBackpressure.sum(); } + public long getUnavailableCount() { return admissionUnavailable.sum(); } public long getOversizedCount() { return admissionOversized.sum(); } public long getPendingAppendBytes() { return pendingAppendBytes.get(); } + public long getMaxPendingBytes() { return properties.getSyslogWalMaxPendingBytes(); } public long getQuarantinedCount() { return quarantined.sum(); } public long getTerminalRecordedCount() { return terminalRecorded.sum(); } public long getIngressPointTimeAssignedCount() { return pointTimeAllocator.getAssignedCount(); } diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusController.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusController.java index e6c96ac..397ecd2 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusController.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusController.java @@ -1,74 +1,23 @@ package com.haobang.wal; -import com.haobang.dispatch.DeviceContextResolver; -import com.haobang.dispatch.SyslogWalDispatcher; -import com.netty.SyslogMessageHandler; 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/wal") public class SyslogWalStatusController { - private final SyslogWalService wal; - private final SyslogWalDispatcher dispatcher; - private final DeviceContextResolver resolver; + private final SyslogWalStatusProvider statusProvider; - public SyslogWalStatusController(SyslogWalService wal, SyslogWalDispatcher dispatcher, - DeviceContextResolver resolver) { - this.wal = wal; - this.dispatcher = dispatcher; - this.resolver = resolver; + public SyslogWalStatusController(SyslogWalStatusProvider statusProvider) { + this.statusProvider = statusProvider; } @GetMapping("/status") public Map status() { - Map value = new LinkedHashMap<>(); - value.put("network_received", SyslogMessageHandler.getNetworkReceivedCount()); - value.put("network_oversized_dropped", SyslogMessageHandler.getOversizedDroppedCount()); - value.put("network_invalid_device_dropped", SyslogMessageHandler.getInvalidDeviceDroppedCount()); - value.put("network_device_lookup_failed", SyslogMessageHandler.getDeviceLookupFailedCount()); - value.put("wal_appended", wal.getAppendedCount()); - value.put("wal_queue_size", wal.getQueueSize()); - value.put("wal_queue_capacity", wal.getQueueCapacity()); - value.put("wal_pending_bytes", wal.getPendingAppendBytes()); - value.put("wal_durable_sequence", wal.getDurableSequence()); - value.put("wal_checkpoint_sequence", wal.getCheckpointSequence()); - value.put("wal_pending_approx", wal.getPendingRecordsApprox()); - value.put("wal_admission_backpressure", wal.getBackpressureCount()); - value.put("wal_admission_oversized", wal.getOversizedCount()); - value.put("wal_terminal_recorded", wal.getTerminalRecordedCount()); - value.put("wal_quarantined", wal.getQuarantinedCount()); - Map pointTimeAssigned = new LinkedHashMap<>(); - pointTimeAssigned.put("ingress", wal.getIngressPointTimeAssignedCount()); - value.put("point_time_assigned_total", pointTimeAssigned); - value.put("wal_disk_critical", wal.isDiskCritical()); - value.put("wal_disk_state", wal.getDiskState()); - value.put("wal_disk_total_bytes", wal.getDiskTotalBytes()); - value.put("wal_disk_usable_bytes", wal.getDiskUsableBytes()); - value.put("wal_disk_stop_threshold_bytes", wal.getDiskStopThresholdBytes()); - value.put("wal_fatal_error", wal.getFatalFailure() == null - ? null : wal.getFatalFailure().toString()); - value.put("dispatch_inflight", dispatcher.getInflight()); - value.put("dispatch_inflight_bytes", dispatcher.getInflightBytes()); - value.put("kafka_acked", dispatcher.getKafkaAcked()); - value.put("kafka_large_dispatched", dispatcher.getLargeDispatched()); - value.put("dispatch_terminal_record_enabled", dispatcher.isTerminalRecordEnabled()); - value.put("dispatch_terminal_filtered", dispatcher.getTerminalFiltered()); - value.put("dispatch_terminal_dropped", dispatcher.getTerminalDropped()); - value.put("dispatch_retry", dispatcher.getRetryCount()); - value.put("tcp_pause_total", SyslogMessageHandler.getTcpPauseCount()); - value.put("udp_dropped_total", SyslogMessageHandler.getUdpDroppedCount()); - value.put("device_cache_size", resolver.getCacheSize()); - value.put("device_cache_hits", resolver.getHitCount()); - value.put("device_cache_misses", resolver.getMissCount()); - value.put("device_ingress_snapshot_ready", resolver.isIngressSnapshotReady()); - value.put("device_ingress_snapshot_size", resolver.getIngressSnapshotSize()); - value.put("device_ingress_snapshot_refresh_failed", resolver.getIngressSnapshotRefreshFailedCount()); - return value; + return statusProvider.status(); } } diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusProvider.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusProvider.java new file mode 100644 index 0000000..ab387ff --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusProvider.java @@ -0,0 +1,78 @@ +package com.haobang.wal; + +import com.haobang.diagnostics.SyslogIpMetricsRegistry; +import com.haobang.dispatch.DeviceContextResolver; +import com.haobang.dispatch.SyslogWalDispatcher; +import com.netty.SyslogMessageHandler; +import org.springframework.stereotype.Component; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** 生成 HTTP 状态与持久化快照共用的单一口径。 */ +@Component +public class SyslogWalStatusProvider { + private final SyslogWalService wal; + private final SyslogWalDispatcher dispatcher; + private final DeviceContextResolver resolver; + private final SyslogIpMetricsRegistry ipMetrics; + + public SyslogWalStatusProvider(SyslogWalService wal, SyslogWalDispatcher dispatcher, + DeviceContextResolver resolver, SyslogIpMetricsRegistry ipMetrics) { + this.wal = wal; + this.dispatcher = dispatcher; + this.resolver = resolver; + this.ipMetrics = ipMetrics; + } + + public Map status() { + Map value = new LinkedHashMap<>(); + value.put("network_received", SyslogMessageHandler.getNetworkReceivedCount()); + value.put("network_oversized_dropped", SyslogMessageHandler.getOversizedDroppedCount()); + value.put("network_invalid_device_dropped", SyslogMessageHandler.getInvalidDeviceDroppedCount()); + value.put("network_device_lookup_failed", SyslogMessageHandler.getDeviceLookupFailedCount()); + value.put("wal_appended", wal.getAppendedCount()); + value.put("wal_queue_size", wal.getQueueSize()); + value.put("wal_queue_capacity", wal.getQueueCapacity()); + value.put("wal_pending_bytes", wal.getPendingAppendBytes()); + value.put("wal_max_pending_bytes", wal.getMaxPendingBytes()); + value.put("wal_durable_sequence", wal.getDurableSequence()); + value.put("wal_checkpoint_sequence", wal.getCheckpointSequence()); + value.put("wal_pending_approx", wal.getPendingRecordsApprox()); + value.put("wal_admission_backpressure", wal.getBackpressureCount()); + value.put("wal_admission_unavailable", wal.getUnavailableCount()); + value.put("wal_admission_oversized", wal.getOversizedCount()); + value.put("wal_terminal_recorded", wal.getTerminalRecordedCount()); + value.put("wal_quarantined", wal.getQuarantinedCount()); + Map pointTimeAssigned = new LinkedHashMap<>(); + pointTimeAssigned.put("ingress", wal.getIngressPointTimeAssignedCount()); + value.put("point_time_assigned_total", pointTimeAssigned); + value.put("wal_disk_critical", wal.isDiskCritical()); + value.put("wal_disk_state", wal.getDiskState()); + value.put("wal_disk_total_bytes", wal.getDiskTotalBytes()); + value.put("wal_disk_usable_bytes", wal.getDiskUsableBytes()); + value.put("wal_disk_stop_threshold_bytes", wal.getDiskStopThresholdBytes()); + value.put("wal_fatal_error", wal.getFatalFailure() == null + ? null : wal.getFatalFailure().toString()); + value.put("dispatch_inflight", dispatcher.getInflight()); + value.put("dispatch_inflight_bytes", dispatcher.getInflightBytes()); + value.put("kafka_acked", dispatcher.getKafkaAcked()); + value.put("kafka_large_dispatched", dispatcher.getLargeDispatched()); + value.put("dispatch_terminal_record_enabled", dispatcher.isTerminalRecordEnabled()); + value.put("dispatch_terminal_filtered", dispatcher.getTerminalFiltered()); + value.put("dispatch_terminal_dropped", dispatcher.getTerminalDropped()); + value.put("dispatch_retry", dispatcher.getRetryCount()); + value.put("tcp_pause_total", SyslogMessageHandler.getTcpPauseCount()); + value.put("tcp_pending_frames_current", SyslogMessageHandler.getTcpPendingFramesCurrent()); + value.put("tcp_pending_frames_max", SyslogMessageHandler.getTcpPendingFramesMax()); + value.put("udp_dropped_total", SyslogMessageHandler.getUdpDroppedCount()); + value.put("device_cache_size", resolver.getCacheSize()); + value.put("device_cache_hits", resolver.getHitCount()); + value.put("device_cache_misses", resolver.getMissCount()); + value.put("device_ingress_snapshot_ready", resolver.isIngressSnapshotReady()); + value.put("device_ingress_snapshot_size", resolver.getIngressSnapshotSize()); + value.put("device_ingress_snapshot_refresh_failed", resolver.getIngressSnapshotRefreshFailedCount()); + value.put("diagnostics", ipMetrics.summary()); + return value; + } +} diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogMessageHandler.java b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogMessageHandler.java index c356031..871df1a 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogMessageHandler.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogMessageHandler.java @@ -1,12 +1,14 @@ package com.netty; import com.haobang.dispatch.DeviceContextResolver; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import com.haobang.wal.SyslogIngressEvent; import com.haobang.wal.SyslogWalService; import com.kafka.kafkaProducer; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.socket.DatagramPacket; +import io.netty.channel.socket.DatagramChannel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -19,6 +21,7 @@ import java.time.ZoneId; import java.util.ArrayDeque; import java.util.Deque; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.LongAdder; /** @@ -37,6 +40,8 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { private static final LongAdder OVERSIZED_DROPPED = new LongAdder(); private static final LongAdder INVALID_DEVICE_DROPPED = new LongAdder(); private static final LongAdder DEVICE_LOOKUP_FAILED = new LongAdder(); + private static final AtomicLong TCP_PENDING_FRAMES_CURRENT = new AtomicLong(); + private static final AtomicLong TCP_PENDING_FRAMES_MAX = new AtomicLong(); /** 每个 TCP Channel 独立 Handler，pending 只保存关闭 AUTO_READ 前已进入 Pipeline 的帧。 */ private final Deque pendingTcp = new ArrayDeque<>(); @@ -44,15 +49,22 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { private final SyslogWalService walService; private final DeviceContextResolver deviceResolver; private final boolean dropInvalidBeforeWal; + private final SyslogIpMetricsRegistry ipMetrics; private boolean retryScheduled; private boolean autoReadPaused; SyslogMessageHandler(Charset syslogCharset, SyslogWalService walService) { - this(syslogCharset, walService, null, false); + this(syslogCharset, walService, null, false, null); } SyslogMessageHandler(Charset syslogCharset, SyslogWalService walService, DeviceContextResolver deviceResolver, boolean dropInvalidBeforeWal) { + this(syslogCharset, walService, deviceResolver, dropInvalidBeforeWal, null); + } + + SyslogMessageHandler(Charset syslogCharset, SyslogWalService walService, + DeviceContextResolver deviceResolver, boolean dropInvalidBeforeWal, + SyslogIpMetricsRegistry ipMetrics) { this.syslogCharset = java.util.Objects.requireNonNull(syslogCharset, "syslogCharset"); this.walService = java.util.Objects.requireNonNull(walService, "walService"); if (dropInvalidBeforeWal && deviceResolver == null) { @@ -60,6 +72,7 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { } this.deviceResolver = deviceResolver; this.dropInvalidBeforeWal = dropInvalidBeforeWal; + this.ipMetrics = ipMetrics; } @Override @@ -93,6 +106,16 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { } NETWORK_RECEIVED.increment(); + if (ipMetrics != null) { + SyslogIpMetricsRegistry.NetworkRecovery recovery = + ipMetrics.recordNetworkReceived(sourceIp, sourcePort, protocol); + if (recovery != null) { + log.info("Syslog 网络接入异常后已恢复接收: sourceIp={}, sourcePort={}, protocol={}, " + + "exceptionTime={}, recoveryTime={}, elapsedMs={}", + sourceIp, sourcePort, protocol, recovery.getExceptionAt(), + recovery.getRecoveredAt(), recovery.getElapsedMillis()); + } + } if (dropInvalidBeforeWal && shouldDropBeforeWal(sourceIp, protocol)) return; Instant receivedInstant = Instant.now(); @@ -114,7 +137,11 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { recordOversized(event, "UDP"); return; } + if (result == SyslogWalService.AdmissionResult.BACKPRESSURE && ipMetrics != null) { + ipMetrics.recordWalBackpressure(event.getSourceIp(), event.getSourcePort(), "UDP"); + } UDP_DROPPED.increment(); + if (ipMetrics != null) ipMetrics.recordUdpDropped(event.getSourceIp(), event.getSourcePort()); long dropped = UDP_DROPPED.sum(); if (dropped == 1 || dropped % 100 == 0) { log.error("UDP Syslog 无法进入 WAL；UDP 不支持背压，已发生明确丢弃: eventId={}, sourceIp={}, " @@ -126,7 +153,7 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { private void acceptTcp(ChannelHandlerContext ctx, SyslogWalService wal, SyslogIngressEvent event) { // pending 非空时必须维持同一 TCP Channel 的原始顺序，不能让新帧越过旧帧。 if (!pendingTcp.isEmpty()) { - pendingTcp.addLast(event); + addPendingTcp(event); pauseAndSchedule(ctx, wal); return; } @@ -137,7 +164,10 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { return; } if (result != SyslogWalService.AdmissionResult.ACCEPTED) { - pendingTcp.addLast(event); + if (result == SyslogWalService.AdmissionResult.BACKPRESSURE && ipMetrics != null) { + ipMetrics.recordWalBackpressure(event.getSourceIp(), event.getSourcePort(), "TCP"); + } + addPendingTcp(event); pauseAndSchedule(ctx, wal); return; } @@ -149,6 +179,12 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { ctx.channel().config().setAutoRead(false); autoReadPaused = true; TCP_PAUSE_COUNT.increment(); + if (ipMetrics != null) { + InetSocketAddress remote = remoteAddress(ctx); + ipMetrics.recordTcpPause(remote == null ? String.valueOf(ctx.channel().remoteAddress()) + : remote.getAddress().getHostAddress(), + remote == null ? 0 : remote.getPort()); + } log.warn("TCP Syslog 已按 WAL 接入保护暂停读取: channel={}, pendingFrames={}, walQueue={}/{}, " + "diskState={}, pauseTotal={}", ctx.channel().remoteAddress(), pendingTcp.size(), wal.getQueueSize(), wal.getQueueCapacity(), @@ -158,7 +194,7 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { } private void scheduleDrain(ChannelHandlerContext ctx, SyslogWalService wal) { - if (retryScheduled) return; + if (retryScheduled || !ctx.channel().isActive()) return; retryScheduled = true; ctx.executor().schedule(() -> { retryScheduled = false; @@ -167,19 +203,27 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { } private void drainPending(ChannelHandlerContext ctx, SyslogWalService wal) { + if (!ctx.channel().isActive()) { + clearPendingTcp(); + autoReadPaused = false; + return; + } while (!pendingTcp.isEmpty()) { SyslogIngressEvent event = pendingTcp.peekFirst(); SyslogWalService.AdmissionResult result = wal.tryAppend(event); if (result == SyslogWalService.AdmissionResult.OVERSIZED) { - pendingTcp.removeFirst(); + removeFirstPendingTcp(); recordOversized(event, "TCP"); continue; } if (result != SyslogWalService.AdmissionResult.ACCEPTED) { + if (result == SyslogWalService.AdmissionResult.BACKPRESSURE && ipMetrics != null) { + ipMetrics.recordWalBackpressure(event.getSourceIp(), event.getSourcePort(), "TCP"); + } scheduleDrain(ctx, wal); return; } - pendingTcp.removeFirst(); + removeFirstPendingTcp(); } if (wal.canResumeTcp()) { if (autoReadPaused && ctx.channel().isActive()) { @@ -196,33 +240,103 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + if (ipMetrics != null) { + InetSocketAddress remote = remoteAddress(ctx); + String protocol = ctx.channel() instanceof DatagramChannel ? "UDP" : "TCP"; + ipMetrics.recordNetworkException( + remote == null ? String.valueOf(ctx.channel().remoteAddress()) + : remote.getAddress().getHostAddress(), + remote == null ? 0 : remote.getPort(), protocol, cause); + } log.error("Syslog 网络接入异常: channel={}", ctx.channel().remoteAddress(), cause); ctx.close(); } + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + clearPendingTcp(); + super.channelInactive(ctx); + } + + @Override + public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { + clearPendingTcp(); + super.handlerRemoved(ctx); + } + public static long getNetworkReceivedCount() { return NETWORK_RECEIVED.sum(); } public static long getTcpPauseCount() { return TCP_PAUSE_COUNT.sum(); } public static long getUdpDroppedCount() { return UDP_DROPPED.sum(); } public static long getOversizedDroppedCount() { return OVERSIZED_DROPPED.sum(); } public static long getInvalidDeviceDroppedCount() { return INVALID_DEVICE_DROPPED.sum(); } public static long getDeviceLookupFailedCount() { return DEVICE_LOOKUP_FAILED.sum(); } + public static long getTcpPendingFramesCurrent() { return TCP_PENDING_FRAMES_CURRENT.get(); } + public static long getTcpPendingFramesMax() { return TCP_PENDING_FRAMES_MAX.get(); } + + private void addPendingTcp(SyslogIngressEvent event) { + pendingTcp.addLast(event); + long current = TCP_PENDING_FRAMES_CURRENT.incrementAndGet(); + updateMax(TCP_PENDING_FRAMES_MAX, current); + if (ipMetrics != null) { + ipMetrics.recordTcpPendingAdded(event.getSourceIp(), event.getSourcePort()); + } + } + + private SyslogIngressEvent removeFirstPendingTcp() { + SyslogIngressEvent event = pendingTcp.pollFirst(); + if (event == null) return null; + decrementIfPositive(TCP_PENDING_FRAMES_CURRENT); + if (ipMetrics != null) { + ipMetrics.recordTcpPendingRemoved(event.getSourceIp(), event.getSourcePort()); + } + return event; + } + + private void clearPendingTcp() { + while (!pendingTcp.isEmpty()) removeFirstPendingTcp(); + } + + private static void updateMax(AtomicLong maximum, long candidate) { + long current = maximum.get(); + while (candidate > current && !maximum.compareAndSet(current, candidate)) { + current = maximum.get(); + } + } + + private static void decrementIfPositive(AtomicLong value) { + long current = value.get(); + while (current > 0L && !value.compareAndSet(current, current - 1L)) { + current = value.get(); + } + } private boolean shouldDropBeforeWal(String sourceIp, String protocol) { DeviceContextResolver.IngressStatus status = deviceResolver.resolveIngress(sourceIp, protocol); if (status == DeviceContextResolver.IngressStatus.REGISTERED) return false; if (status == DeviceContextResolver.IngressStatus.INVALID) { INVALID_DEVICE_DROPPED.increment(); + if (ipMetrics != null) ipMetrics.recordInvalidDeviceDropped(sourceIp, protocol); return true; } // 后台快照暂不可用时不访问数据库、不阻塞网络线程；保守进入 WAL 等待 Dispatcher 判断。 DEVICE_LOOKUP_FAILED.increment(); + if (ipMetrics != null) ipMetrics.recordDeviceLookupFailed(sourceIp, protocol); return false; } private void recordOversized(SyslogIngressEvent event, String protocol) { OVERSIZED_DROPPED.increment(); + if (ipMetrics != null) { + ipMetrics.recordOversizedDropped(event.getSourceIp(), event.getSourcePort(), protocol, + event.getRawMessage().length()); + } log.error("Syslog 超过单条消息硬上限，已拒绝且不记录原文: eventId={}, protocol={}, sourceIp={}, chars={}, rejectedTotal={}", event.getEventId(), protocol, event.getSourceIp(), event.getRawMessage().length(), OVERSIZED_DROPPED.sum()); } + + private InetSocketAddress remoteAddress(ChannelHandlerContext ctx) { + SocketAddress address = ctx.channel().remoteAddress(); + return address instanceof InetSocketAddress ? (InetSocketAddress) address : null; + } } diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServer.java b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServer.java index b3f9dc4..9fd2056 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServer.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServer.java @@ -2,6 +2,7 @@ package com.netty; import com.haobang.config.AppProperties; import com.haobang.dispatch.DeviceContextResolver; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import com.haobang.util.SpringContextUtil; import com.haobang.wal.SyslogWalService; import io.netty.bootstrap.Bootstrap; @@ -43,6 +44,7 @@ public class SyslogServer { AppProperties properties = SpringContextUtil.getBean(AppProperties.class); SyslogWalService walService = SpringContextUtil.getBean(SyslogWalService.class); DeviceContextResolver deviceResolver = SpringContextUtil.getBean(DeviceContextResolver.class); + SyslogIpMetricsRegistry ipMetrics = SpringContextUtil.getBean(SyslogIpMetricsRegistry.class); Charset charset = walService.getSyslogCharset(); EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); @@ -53,7 +55,7 @@ public class SyslogServer { .channel(NioServerSocketChannel.class) .childHandler(new SyslogTcpChannelInitializer( properties.getSyslogMaxFrameLength(), charset, walService, deviceResolver, - !properties.isSyslogDispatchTerminalRecordEnabled())) + !properties.isSyslogDispatchTerminalRecordEnabled(), ipMetrics)) .option(ChannelOption.SO_BACKLOG, 128) .childOption(ChannelOption.SO_KEEPALIVE, true); @@ -71,6 +73,7 @@ public class SyslogServer { AppProperties properties = SpringContextUtil.getBean(AppProperties.class); SyslogWalService walService = SpringContextUtil.getBean(SyslogWalService.class); DeviceContextResolver deviceResolver = SpringContextUtil.getBean(DeviceContextResolver.class); + SyslogIpMetricsRegistry ipMetrics = SpringContextUtil.getBean(SyslogIpMetricsRegistry.class); Charset charset = walService.getSyslogCharset(); EventLoopGroup group = new NioEventLoopGroup(); @@ -79,7 +82,7 @@ public class SyslogServer { b.group(group) .channel(NioDatagramChannel.class) .handler(new SyslogUdpChannelInitializer(charset, walService, deviceResolver, - !properties.isSyslogDispatchTerminalRecordEnabled())) + !properties.isSyslogDispatchTerminalRecordEnabled(), ipMetrics)) .option(ChannelOption.SO_BROADCAST, true); ChannelFuture f = b.bind(udpPort).sync(); diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java index 072af57..c6c1ac5 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java @@ -17,6 +17,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicReference; import com.haobang.config.AppProperties; import com.haobang.dispatch.DeviceContextResolver; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import com.haobang.wal.SyslogWalService; import java.nio.charset.Charset; @@ -29,6 +30,7 @@ public class SyslogServerBoth { private final AppProperties properties; private final SyslogWalService walService; private final DeviceContextResolver deviceResolver; + private final SyslogIpMetricsRegistry ipMetrics; private final Charset syslogCharset; private final int tcpPort; private final int udpPort; @@ -44,10 +46,12 @@ public class SyslogServerBoth { private final ExecutorService serverExecutor = Executors.newFixedThreadPool(2); public SyslogServerBoth(AppProperties properties, SyslogWalService walService, - DeviceContextResolver deviceResolver, Charset syslogCharset) { + DeviceContextResolver deviceResolver, SyslogIpMetricsRegistry ipMetrics, + Charset syslogCharset) { this.properties = java.util.Objects.requireNonNull(properties, "properties"); this.walService = java.util.Objects.requireNonNull(walService, "walService"); this.deviceResolver = java.util.Objects.requireNonNull(deviceResolver, "deviceResolver"); + this.ipMetrics = java.util.Objects.requireNonNull(ipMetrics, "ipMetrics"); this.syslogCharset = java.util.Objects.requireNonNull(syslogCharset, "syslogCharset"); this.tcpPort = properties.getSyslogTcpPort(); this.udpPort = properties.getSyslogUdpPort(); @@ -118,7 +122,7 @@ public class SyslogServerBoth { .channel(NioServerSocketChannel.class) .childHandler(new SyslogTcpChannelInitializer( properties.getSyslogMaxFrameLength(), syslogCharset, walService, deviceResolver, - !properties.isSyslogDispatchTerminalRecordEnabled())) + !properties.isSyslogDispatchTerminalRecordEnabled(), ipMetrics)) // [SYSLOG-ALARM][NINGRESS][已修复] backlog/接收缓冲配置化；实际生效上限仍受操作系统 somaxconn/rmem_max 约束。 .option(ChannelOption.SO_BACKLOG, properties.getSyslogTcpBacklog()) .childOption(ChannelOption.SO_KEEPALIVE, true) @@ -141,7 +145,7 @@ public class SyslogServerBoth { b.group(udpGroup) .channel(NioDatagramChannel.class) .handler(new SyslogUdpChannelInitializer(syslogCharset, walService, deviceResolver, - !properties.isSyslogDispatchTerminalRecordEnabled())) + !properties.isSyslogDispatchTerminalRecordEnabled(), ipMetrics)) // [SYSLOG-ALARM][NINGRESS][已修复] 显式配置 UDP 接收缓冲；UDP 仍无端到端确认，强可靠设备应改用 TCP/TLS 或 RELP。 .option(ChannelOption.SO_BROADCAST, true) .option(ChannelOption.SO_RCVBUF, properties.getSyslogUdpReceiveBufferBytes()); diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerLifecycle.java b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerLifecycle.java index e32124b..0029dcd 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerLifecycle.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerLifecycle.java @@ -2,6 +2,7 @@ package com.netty; import com.haobang.config.AppProperties; import com.haobang.dispatch.DeviceContextResolver; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import com.haobang.dispatch.SyslogWalDispatcher; import com.haobang.wal.SyslogWalService; import org.slf4j.Logger; @@ -26,17 +27,20 @@ public class SyslogServerLifecycle implements SmartLifecycle { private final AppProperties properties; private final SyslogWalService walService; private final DeviceContextResolver deviceResolver; + private final SyslogIpMetricsRegistry ipMetrics; @SuppressWarnings("unused") private final SyslogWalDispatcher dispatcher; private final AtomicBoolean running = new AtomicBoolean(); private SyslogServerBoth server; public SyslogServerLifecycle(AppProperties properties, SyslogWalService walService, - SyslogWalDispatcher dispatcher, DeviceContextResolver deviceResolver) { + SyslogWalDispatcher dispatcher, DeviceContextResolver deviceResolver, + SyslogIpMetricsRegistry ipMetrics) { this.properties = properties; this.walService = walService; this.dispatcher = dispatcher; this.deviceResolver = deviceResolver; + this.ipMetrics = ipMetrics; } @Override @@ -44,7 +48,7 @@ public class SyslogServerLifecycle implements SmartLifecycle { if (!running.compareAndSet(false, true)) return; try { Charset charset = walService.getSyslogCharset(); - server = new SyslogServerBoth(properties, walService, deviceResolver, charset); + server = new SyslogServerBoth(properties, walService, deviceResolver, ipMetrics, charset); server.start(); log.info("Syslog TCP/UDP 端口已在 WAL 和 Dispatcher 就绪后开放: charset={}", charset.name()); } catch (Exception failure) { diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogTcpChannelInitializer.java b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogTcpChannelInitializer.java index be3a75c..06f0469 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogTcpChannelInitializer.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogTcpChannelInitializer.java @@ -2,6 +2,7 @@ package com.netty; import com.haobang.dispatch.DeviceContextResolver; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import com.haobang.wal.SyslogWalService; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; @@ -21,15 +22,18 @@ public class SyslogTcpChannelInitializer extends ChannelInitializer Files.isRegularFile(persistence.eventFile(today))); + + String snapshot = new String(Files.readAllBytes(persistence.snapshotFile(today)), + StandardCharsets.UTF_8); + String ipSnapshot = new String(Files.readAllBytes(persistence.ipSnapshotFile(today)), + StandardCharsets.UTF_8); + String event = new String(Files.readAllBytes(persistence.eventFile(today)), + StandardCharsets.UTF_8); + assertTrue(snapshot.contains("\"ip_stats_count\":1")); + assertTrue(ipSnapshot.contains("\"source_ip\":\"10.56.170.107\"")); + assertTrue(ipSnapshot.contains("\"delta\"")); + assertTrue(event.contains("\"event_type\":\"TOO_LONG_FRAME\"")); + assertFalse(Files.exists(expired)); + assertTrue(Files.exists(retained)); + } + + private AppProperties properties() { + AppProperties value = new AppProperties(); + value.setSyslogDiagnosticsEnabled(true); + value.setSyslogDiagnosticsDirectory(tempDirectory.toString()); + value.setSyslogDiagnosticsSnapshotIntervalMs(60_000L); + value.setSyslogDiagnosticsRetentionDays(7); + value.setSyslogDiagnosticsMaxSourceIps(100); + value.setSyslogDiagnosticsEventQueueCapacity(100); + value.setAppServiceDeviceCollectId(7); + value.setAppServiceDeviceCollectName("probe-7"); + return value; + } + + private void waitUntil(Check condition) throws Exception { + long deadline = System.currentTimeMillis() + 3000L; + while (!condition.done() && System.currentTimeMillis() < deadline) Thread.sleep(10L); + assertTrue(condition.done(), "condition not reached before timeout"); + } + + private interface Check { boolean done() throws Exception; } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/diagnostics/SyslogIpMetricsRegistryTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/diagnostics/SyslogIpMetricsRegistryTest.java new file mode 100644 index 0000000..a7c4abf --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/diagnostics/SyslogIpMetricsRegistryTest.java @@ -0,0 +1,107 @@ +package com.haobang.diagnostics; + +import com.common.entity.DeviceDevice; +import com.haobang.config.AppProperties; +import io.netty.handler.codec.TooLongFrameException; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SyslogIpMetricsRegistryTest { + @Test + void keepsPerIpTotalsDeltasAndClassifiedExceptionEvent() throws Exception { + SyslogIpMetricsRegistry registry = new SyslogIpMetricsRegistry(properties()); + DeviceDevice device = new DeviceDevice(); + device.setId(38); + device.setName("天眼-省侧分析平台"); + + registry.recordNetworkReceived("10.56.170.107", 35064, "TCP"); + registry.recordWalAppended("10.56.170.107", 35064, "TCP"); + registry.recordKafkaAck("10.56.170.107", 35064, "TCP", device); + registry.recordNetworkException("10.56.170.107", 35064, "TCP", + new TooLongFrameException("frame exceeds 6 MiB")); + + Map first = registry.snapshotAndAdvanceDeltas().get(0); + assertEquals(38, first.get("device_id")); + assertEquals(1L, first.get("network_received")); + assertEquals(1L, first.get("wal_appended")); + assertEquals(1L, first.get("kafka_acked")); + assertEquals(1L, first.get("tcp_too_long_frame_total")); + Map firstDelta = (Map) first.get("delta"); + assertEquals(1L, firstDelta.get("network_received")); + + Map second = registry.snapshotAndAdvanceDeltas().get(0); + Map secondDelta = (Map) second.get("delta"); + assertEquals(0L, secondDelta.get("network_received")); + assertEquals(0L, secondDelta.get("tcp_too_long_frame_total")); + + Map event = registry.pollEvent(1L, TimeUnit.SECONDS); + assertNotNull(event); + assertEquals("TOO_LONG_FRAME", event.get("event_type")); + assertEquals("10.56.170.107", event.get("source_ip")); + } + + @Test + void boundsSourceIpCardinalityAndUsesOverflowBucket() { + AppProperties properties = properties(); + properties.setSyslogDiagnosticsMaxSourceIps(1); + SyslogIpMetricsRegistry registry = new SyslogIpMetricsRegistry(properties); + + registry.recordNetworkReceived("10.0.0.1", 1, "UDP"); + registry.recordNetworkReceived("10.0.0.2", 2, "UDP"); + + List> rows = registry.currentSnapshot(); + assertEquals(2, rows.size()); + assertTrue(rows.stream().anyMatch(row -> SyslogIpMetricsRegistry.OVERFLOW_SOURCE_IP + .equals(row.get("source_ip")))); + } + + @Test + void reportsRecoveryOnlyForFirstMessageAfterNetworkException() { + SyslogIpMetricsRegistry registry = new SyslogIpMetricsRegistry(properties()); + + assertNull(registry.recordNetworkReceived("10.56.170.107", 35064, "TCP")); + registry.recordNetworkException("10.56.170.107", 35064, "TCP", + new IllegalStateException("connection reset")); + + SyslogIpMetricsRegistry.NetworkRecovery recovery = + registry.recordNetworkReceived("10.56.170.107", 35100, "TCP"); + assertNotNull(recovery); + assertTrue(recovery.getRecoveredAt().compareTo(recovery.getExceptionAt()) >= 0); + assertTrue(recovery.getElapsedMillis() >= 0L); + assertNull(registry.recordNetworkReceived("10.56.170.107", 35100, "TCP")); + } + + @Test + void tracksWalUnavailableAndTcpPendingCurrentAndPeakPerIp() { + SyslogIpMetricsRegistry registry = new SyslogIpMetricsRegistry(properties()); + + registry.recordWalUnavailable("10.56.170.107", 35064, "TCP"); + registry.recordTcpPendingAdded("10.56.170.107", 35064); + registry.recordTcpPendingAdded("10.56.170.107", 35064); + registry.recordTcpPendingRemoved("10.56.170.107", 35064); + + Map row = registry.snapshotAndAdvanceDeltas().get(0); + assertEquals(1L, row.get("wal_admission_unavailable")); + assertEquals(1L, row.get("tcp_pending_frames_current")); + assertEquals(2L, row.get("tcp_pending_frames_max")); + Map delta = (Map) row.get("delta"); + assertEquals(1L, delta.get("wal_admission_unavailable")); + assertEquals(1L, registry.getTcpPendingFramesCurrent()); + assertEquals(2L, registry.getTcpPendingFramesMax()); + } + + private AppProperties properties() { + AppProperties value = new AppProperties(); + value.setSyslogDiagnosticsMaxSourceIps(100); + value.setSyslogDiagnosticsEventQueueCapacity(100); + return value; + } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java index 2906485..a712119 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java @@ -3,6 +3,7 @@ package com.haobang.dispatch; import com.common.entity.DeviceDevice; import com.fasterxml.jackson.databind.ObjectMapper; import com.haobang.config.AppProperties; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import com.haobang.wal.SyslogIngressEvent; import com.haobang.wal.IngressEvent; import com.haobang.wal.IngressType; @@ -84,6 +85,8 @@ class SyslogWalDispatcherTest { .thenAnswer(invocation -> "20260811-event-1".equals(invocation.getArgument(3)) ? first : second); dispatcher = new SyslogWalDispatcher(wal, resolver, producer, properties); + SyslogIpMetricsRegistry metrics = new SyslogIpMetricsRegistry(properties); + dispatcher.setIpMetrics(metrics); dispatcher.start(); waitUntil(() -> dispatcher.getInflight() == 2L); @@ -94,6 +97,8 @@ class SyslogWalDispatcherTest { first.complete(null); waitUntil(() -> wal.getCheckpointSequence() == 2L); assertEquals(2L, dispatcher.getKafkaAcked()); + assertEquals(2L, metrics.currentSnapshot().get(0).get("kafka_acked")); + assertEquals(126, metrics.currentSnapshot().get(0).get("device_id")); assertEquals(0L, dispatcher.getRetryCount()); verify(producer).messagePush(any(DeviceDevice.class), eq("first"), eq("20260811200000000"), eq("20260811-event-1"), eq(firstEvent.getPointTimeNs())); @@ -137,11 +142,14 @@ class SyslogWalDispatcherTest { kafkaProducer producer = mock(kafkaProducer.class); dispatcher = new SyslogWalDispatcher(wal, resolver, producer, properties); + SyslogIpMetricsRegistry metrics = new SyslogIpMetricsRegistry(properties); + dispatcher.setIpMetrics(metrics); dispatcher.start(); waitUntil(() -> wal.getCheckpointSequence() == 1L); assertEquals(1L, dispatcher.getTerminalFiltered()); assertEquals(1L, dispatcher.getTerminalDropped()); + assertEquals(1L, metrics.currentSnapshot().get(0).get("dispatch_terminal_dropped")); assertEquals(0L, wal.getTerminalRecordedCount()); try (Stream files = Files.list(tempDirectory.resolve("terminal"))) { assertEquals(0L, files.count()); diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/wal/SyslogWalDiagnosticsTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/wal/SyslogWalDiagnosticsTest.java new file mode 100644 index 0000000..a18dfae --- /dev/null +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/wal/SyslogWalDiagnosticsTest.java @@ -0,0 +1,112 @@ +package com.haobang.wal; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.haobang.config.AppProperties; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; +import com.haobang.dispatch.DeviceContextResolver; +import com.haobang.dispatch.SyslogWalDispatcher; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.time.LocalDateTime; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; + +class SyslogWalDiagnosticsTest { + @TempDir Path tempDirectory; + private SyslogWalService wal; + + @AfterEach + void tearDown() { + if (wal != null) wal.shutdown(); + } + + @Test + void durableWalReceiptIncrementsMatchingSourceIp() throws Exception { + AppProperties properties = properties(); + SyslogIpMetricsRegistry metrics = new SyslogIpMetricsRegistry(properties); + wal = new SyslogWalService(properties, new ObjectMapper()); + wal.setIpMetrics(metrics); + wal.initialize(); + + SyslogIngressEvent event = new SyslogIngressEvent("event-1", + LocalDateTime.of(2026, 9, 16, 10, 0), + "10.56.170.107", 35064, "TCP", "message"); + assertEquals(SyslogWalService.AdmissionResult.ACCEPTED, wal.tryAppend(event)); + waitUntil(() -> wal.getAppendedCount() == 1L); + + Map row = metrics.currentSnapshot().get(0); + assertEquals("10.56.170.107", row.get("source_ip")); + assertEquals(1L, row.get("wal_appended")); + } + + @Test + void statusIncludesSelectedCapacityAndPendingFields() { + AppProperties properties = properties(); + properties.setSyslogWalMaxPendingBytes(123456L); + SyslogIpMetricsRegistry metrics = new SyslogIpMetricsRegistry(properties); + wal = new SyslogWalService(properties, new ObjectMapper()); + wal.initialize(); + SyslogWalStatusProvider provider = new SyslogWalStatusProvider(wal, + mock(SyslogWalDispatcher.class), mock(DeviceContextResolver.class), metrics); + + Map status = provider.status(); + assertEquals(123456L, status.get("wal_max_pending_bytes")); + assertEquals(0L, status.get("wal_admission_unavailable")); + assertTrue(status.containsKey("tcp_pending_frames_current")); + assertTrue(status.containsKey("tcp_pending_frames_max")); + } + + @Test + void unavailableAdmissionIsCountedGloballyAndForSourceIp() { + AppProperties properties = properties(); + SyslogIpMetricsRegistry metrics = new SyslogIpMetricsRegistry(properties); + wal = new SyslogWalService(properties, new ObjectMapper()); + wal.setIpMetrics(metrics); + SyslogIngressEvent event = new SyslogIngressEvent("event-unavailable", + LocalDateTime.of(2026, 9, 16, 10, 0), + "10.56.170.107", 35064, "TCP", "message"); + + assertEquals(SyslogWalService.AdmissionResult.UNAVAILABLE, wal.tryAppend(event)); + assertEquals(1L, wal.getUnavailableCount()); + assertEquals(1L, metrics.currentSnapshot().get(0).get("wal_admission_unavailable")); + } + + private AppProperties properties() { + AppProperties value = new AppProperties(); + value.setSyslogWalEnabled(true); + value.setSyslogWalDirectory(tempDirectory.toString()); + value.setSyslogWalQueueCapacity(16); + value.setSyslogWalMaxPendingBytes(0L); + value.setSyslogWalHighWatermarkPercent(70); + value.setSyslogWalLowWatermarkPercent(40); + value.setSyslogWalRetryIntervalMs(5L); + value.setSyslogWalDiskRetryIntervalMs(50L); + value.setSyslogWalFlushRecords(1); + value.setSyslogWalFlushIntervalMs(5L); + value.setSyslogWalSegmentMaxBytes(1024 * 1024L); + value.setSyslogWalRetentionDays(0); + value.setSyslogWalShutdownTimeoutMs(5000L); + value.setSyslogWalCheckpointFlushRecords(1); + value.setSyslogWalCheckpointFlushIntervalMs(5L); + value.setSyslogWalDiskHighWatermarkPercent(98); + value.setSyslogWalMinFreeBytes(0L); + value.setSyslogCharset("UTF-8"); + value.setSyslogDiagnosticsMaxSourceIps(100); + value.setSyslogDiagnosticsEventQueueCapacity(100); + return value; + } + + private void waitUntil(Check condition) throws Exception { + long deadline = System.currentTimeMillis() + 3000L; + while (!condition.done() && System.currentTimeMillis() < deadline) Thread.sleep(10L); + assertTrue(condition.done(), "condition not reached before timeout"); + } + + private interface Check { boolean done() throws Exception; } +} diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/wal/SyslogWalServiceTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/wal/SyslogWalServiceTest.java index 1f58d5a..b86456c 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/haobang/wal/SyslogWalServiceTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/haobang/wal/SyslogWalServiceTest.java @@ -96,6 +96,9 @@ class SyslogWalServiceTest { AppendAttempt attempt = service.tryAppendUnified(event); assertEquals(AdmissionStatus.UNAVAILABLE, attempt.getStatus()); assertEquals(null, attempt.getDurableStage()); + assertEquals(1L, service.getUnavailableCount()); + assertEquals(properties(tempDirectory).getSyslogWalMaxPendingBytes(), + service.getMaxPendingBytes()); } @Test diff --git a/haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java b/haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java index 103f746..78933d6 100644 --- a/haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java +++ b/haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java @@ -1,14 +1,19 @@ package com.netty; import com.haobang.dispatch.DeviceContextResolver; +import com.haobang.config.AppProperties; +import com.haobang.diagnostics.SyslogIpMetricsRegistry; import com.haobang.wal.SyslogWalService; import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.TooLongFrameException; import org.junit.jupiter.api.Test; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; @@ -17,6 +22,27 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; class SyslogMessageHandlerBackpressureTest { + @Test + void recordsPerIpReceiptAndNetworkException() { + SyslogWalService wal = mock(SyslogWalService.class); + when(wal.prepareForWal(any())).thenAnswer(invocation -> invocation.getArgument(0)); + when(wal.tryAppend(any())).thenReturn(SyslogWalService.AdmissionResult.ACCEPTED); + AppProperties properties = new AppProperties(); + properties.setSyslogDiagnosticsMaxSourceIps(10); + properties.setSyslogDiagnosticsEventQueueCapacity(10); + SyslogIpMetricsRegistry metrics = new SyslogIpMetricsRegistry(properties); + EmbeddedChannel channel = new EmbeddedChannel(new SyslogMessageHandler( + StandardCharsets.UTF_8, wal, null, false, metrics)); + channel.connect(new InetSocketAddress("192.168.4.19", 5514)); + + channel.writeInbound("event"); + channel.pipeline().fireExceptionCaught(new TooLongFrameException("too long")); + + assertEquals(1L, metrics.currentSnapshot().get(0).get("network_received")); + assertEquals(1L, metrics.currentSnapshot().get(0).get("tcp_too_long_frame_total")); + channel.close(); + } + @Test void tcpWalBackpressureDisablesAutoReadInsteadOfDroppingCurrentFrame() { SyslogWalService wal = mock(SyslogWalService.class); @@ -24,13 +50,26 @@ class SyslogMessageHandlerBackpressureTest { when(wal.tryAppend(any())).thenReturn(SyslogWalService.AdmissionResult.BACKPRESSURE); when(wal.getRetryIntervalMs()).thenReturn(60_000L); when(wal.getQueueCapacity()).thenReturn(10_000); - EmbeddedChannel channel = new EmbeddedChannel(new SyslogMessageHandler(StandardCharsets.UTF_8, wal)); + AppProperties properties = new AppProperties(); + properties.setSyslogDiagnosticsMaxSourceIps(10); + properties.setSyslogDiagnosticsEventQueueCapacity(10); + SyslogIpMetricsRegistry metrics = new SyslogIpMetricsRegistry(properties); + EmbeddedChannel channel = new EmbeddedChannel(new SyslogMessageHandler( + StandardCharsets.UTF_8, wal, null, false, metrics)); channel.connect(new InetSocketAddress("127.0.0.1", 5514)); + long globalPendingBefore = SyslogMessageHandler.getTcpPendingFramesCurrent(); channel.writeInbound("<134>Aug 11 20:00:00 device event"); assertFalse(channel.config().isAutoRead()); + assertEquals(globalPendingBefore + 1L, SyslogMessageHandler.getTcpPendingFramesCurrent()); + assertTrue(SyslogMessageHandler.getTcpPendingFramesMax() >= globalPendingBefore + 1L); + assertEquals(1L, metrics.currentSnapshot().get(0).get("wal_admission_backpressure")); + assertEquals(1L, metrics.currentSnapshot().get(0).get("tcp_pending_frames_current")); + assertEquals(1L, metrics.currentSnapshot().get(0).get("tcp_pending_frames_max")); channel.close(); + assertEquals(globalPendingBefore, SyslogMessageHandler.getTcpPendingFramesCurrent()); + assertEquals(0L, metrics.currentSnapshot().get(0).get("tcp_pending_frames_current")); } @Test diff --git a/haobang-security-dm/upgrade-doc/v5/03-syslog-wal-diagnostics.md b/haobang-security-dm/upgrade-doc/v5/03-syslog-wal-diagnostics.md new file mode 100644 index 0000000..7486ef6 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v5/03-syslog-wal-diagnostics.md @@ -0,0 +1,142 @@ +# Syslog WAL 与分 IP 诊断历史 + +## 功能范围 + +`syslog-serve` 每分钟保存一次全局 WAL 状态，并按来源 IP 保存接收、WAL 持久化、Kafka ACK、过滤、背压、丢弃和 TCP 异常计数。网络异常发生时立即写入事件文件。 + +某个来源 IP 抛出网络接入异常后，如果后续重新收到该 IP 的数据，服务会打印一次 +`Syslog 网络接入异常后已恢复接收` 日志，并包含异常时间、恢复时间和两者间隔。该机制不根据无数据时长推断断网，也不生成离线统计；连续正常数据不会重复打印恢复日志。 + +诊断文件不经过 Kafka 和数据库，默认写入容器目录 `/app/logs/syslog-monitor`。生产编排已将 `/app/logs` 映射到宿主机 `/home/syslog/logs`，因此容器重启不会删除历史。文件按自然日生成，保留今天及此前 6 天，共 7 天。 + +诊断记录不包含日志正文、凭据或 Kafka 消息内容。 + +## 文件目录 + +```text +/home/syslog/logs/syslog-monitor/ +├── snapshots/syslog-status-YYYY-MM-DD.jsonl +├── ip-snapshots/syslog-ip-status-YYYY-MM-DD.jsonl +└── events/syslog-events-YYYY-MM-DD.jsonl +``` + +- `snapshots`：全局 WAL、磁盘和 Kafka 投递状态，每分钟一行。 +- `ip-snapshots`：每个来源 IP 每分钟一行，包含进程累计值与相邻快照增量 `delta`。 +- `events`：`TOO_LONG_FRAME`、`CONNECTION_RESET`、超长消息和最终设备过滤等即时事件。 + +每次进程启动生成新的 `instance_id`，并记录 `process_start_time`。对累计值进行跨重启分析时必须按 `instance_id` 分段。 + +## 部署配置 + +生产默认值： + +```yaml +environment: + - SYSLOG_DIAGNOSTICS_DIRECTORY=/app/logs/syslog-monitor + - SYSLOG_DIAGNOSTICS_RETENTION_DAYS=7 +``` + +可选参数： + +```text +SYSLOG_DIAGNOSTICS_ENABLED=true +SYSLOG_DIAGNOSTICS_SNAPSHOT_INTERVAL_MS=60000 +SYSLOG_DIAGNOSTICS_MAX_SOURCE_IPS=10000 +SYSLOG_DIAGNOSTICS_EVENT_QUEUE_CAPACITY=10000 +``` + +来源 IP 超过上限后，新 IP 的指标汇总到 `__OTHER__`，避免伪造 UDP 来源导致内存无限增长。事件队列满时不会阻塞 Netty 接入线程，丢弃的诊断事件数量通过 `event_queue_dropped` 暴露。 + +## 现场查询 + +### 第一步：检查诊断持久化是否正常 + +```bash +docker exec syslog-serve-dm sh -c \ + "curl -fsS http://127.0.0.1:8189/syslogserve/api/syslog/wal/diagnostics/current" +``` + +确认以下字段： + +```text +persistence.enabled = true +persistence.active = true +persistence.retention_days = 7 +persistence.initialization_failures = 0 +persistence.snapshot_write_failures = 0 +persistence.event_write_failures = 0 +persistence.event_queue_dropped = 0 +persistence.last_write_error = null +``` + +### 第二步：查询单个来源 IP + +```bash +docker exec syslog-serve-dm sh -c \ + "curl -fsS 'http://127.0.0.1:8189/syslogserve/api/syslog/wal/diagnostics/current?source_ip=10.56.170.107'" +``` + +主要字段： + +```text +network_received +wal_appended +wal_admission_backpressure +wal_admission_unavailable +kafka_acked +invalid_device_dropped +oversized_dropped +udp_dropped +tcp_pause_total +tcp_pending_frames_current +tcp_pending_frames_max +network_exception_total +tcp_network_exception_total +udp_network_exception_total +tcp_connection_reset_total +tcp_too_long_frame_total +dispatch_terminal_dropped +dispatch_retry +last_receive_time +last_exception_time +``` + +`wal_appended` 只在 WAL 完成持久化后增加；`kafka_acked` 只在 Kafka 明确 ACK 后增加。 + +全局状态还包含 `wal_pending_bytes`、`wal_max_pending_bytes`。前者是当前待落盘字节数，后者是配置上限，配置为 `0` 表示不限制。`wal_admission_unavailable` 表示 WAL 因未运行、致命错误或磁盘进入 `CRITICAL` 而拒绝接入尝试的次数；TCP 会保留并重试待处理帧，因此该值不等于 TCP 丢失消息数。`tcp_pending_frames_current` 是当前在 TCP Handler 内等待进入 WAL 的帧数，`tcp_pending_frames_max` 是当前进程启动以来的峰值；连接关闭时未处理帧会从 current 中回减。 + +### 第三步：查询当天历史 + +直接在宿主机按 IP 查询： + +```bash +grep -F '"source_ip":"10.56.170.107"' \ + /home/syslog/logs/syslog-monitor/ip-snapshots/syslog-ip-status-2026-09-16.jsonl +``` + +查询该 IP 的异常事件： + +```bash +grep -F '"source_ip":"10.56.170.107"' \ + /home/syslog/logs/syslog-monitor/events/syslog-events-2026-09-16.jsonl +``` + +也可以通过 HTTP 下载： + +```text +GET /syslogserve/api/syslog/wal/diagnostics/history?date=2026-09-16&type=snapshots +GET /syslogserve/api/syslog/wal/diagnostics/history?date=2026-09-16&type=ip +GET /syslogserve/api/syslog/wal/diagnostics/history?date=2026-09-16&type=events +``` + +## 计数对账 + +同一 `instance_id` 下： + +```text +有效设备 WAL 持久化量 ≈ wal_appended +成功发往 Kafka 的数量 = kafka_acked +当前实例尚未 ACK 的近似数量 = wal_appended - kafka_acked - dispatch_terminal_dropped +``` + +重启后可能重放旧 WAL，因此分 IP 的 `kafka_acked` 可能暂时大于本实例的 `wal_appended`。此时以全局 `wal_pending_approx`、WAL 序号以及 `instance_id` 为准，不应把负差值解释为异常。 diff --git a/haobang-security-dm/upgrade-doc/v5/README.md b/haobang-security-dm/upgrade-doc/v5/README.md index d996172..e2c594b 100644 --- a/haobang-security-dm/upgrade-doc/v5/README.md +++ b/haobang-security-dm/upgrade-doc/v5/README.md @@ -3,6 +3,9 @@ 天眼/微步 JSON 异常的代码入口和现场开发排查顺序见 [`02-json-format-error-debug-guide.md`](./02-json-format-error-debug-guide.md)。 +Syslog 接收、WAL、Kafka ACK、设备过滤和 TCP 异常的全局/分 IP 七天诊断历史见 +[`03-syslog-wal-diagnostics.md`](./03-syslog-wal-diagnostics.md)。 + ## 功能说明 v5 为 `syslog-consumer-rule` 增加按设备短时采集原始日志的诊断接口，并提供自动落盘脚本。