From faa9ffe2d9f018c331dd22f1fb1b2e995007fef0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Tue, 15 Sep 2026 22:10:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(syslog):=20=E4=BC=98=E5=8C=96WAL=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E5=B9=B6=E5=9C=A8=E5=85=A5=E7=AB=99=E9=98=B6=E6=AE=B5?= =?UTF-8?q?=E4=B8=A2=E5=BC=83=E6=97=A0=E6=95=88=E8=AE=BE=E5=A4=87=E6=97=A5?= =?UTF-8?q?=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 默认在进入WAL前丢弃未注册、重复或未分配到当前探针的Syslog日志 - 使用后台设备内存快照，避免Netty网络线程访问Redis或达梦 - 增加Terminal审计开关，允许恢复原有逐条审计文件模式 - 历史WAL中的无效设备记录直接推进checkpoint - 增加无效设备丢弃、快照状态及刷新失败监控指标 - 生产WAL队列调整为20000条 - WAL待写预算和分段大小调整为2GiB - 同步生产配置、Docker Compose及升级文档 - 增加无效设备前置过滤和Terminal开关测试 chore(syslog): 统一生产环境 WAL 挂载目录 - 将生产环境 WAL 默认目录调整为 /app/logs/syslog-wal - Docker Compose 显式设置 SYSLOG_WAL_DIRECTORY - WAL 与应用日志共用 /app/logs 宿主机挂载 - 移除旧的 /data/syslog-archive 独立挂载 - 宿主机 WAL 路径统一为 /home/syslog/logs/syslog-wal --- .../docker-compose-serve-prod.yaml | 25 ++- .../com/haobang/config/AppProperties.java | 4 + .../dispatch/DeviceContextResolver.java | 115 ++++++++++- .../haobang/dispatch/SyslogWalDispatcher.java | 22 +- .../wal/SyslogWalStatusController.java | 7 + .../java/com/netty/SyslogMessageHandler.java | 40 +++- .../src/main/java/com/netty/SyslogServer.java | 9 +- .../main/java/com/netty/SyslogServerBoth.java | 12 +- .../java/com/netty/SyslogServerLifecycle.java | 7 +- .../netty/SyslogTcpChannelInitializer.java | 11 +- .../netty/SyslogUdpChannelInitializer.java | 11 +- .../resources/application-prod.properties | 18 ++ .../src/main/resources/application.properties | 2 + .../dispatch/DeviceContextResolverTest.java | 28 +++ .../dispatch/SyslogWalDispatcherTest.java | 28 +++ .../SyslogMessageHandlerBackpressureTest.java | 40 ++++ .../v3/config/01-new-environment-variables.md | 12 +- .../v5/02-json-format-error-debug-guide.md | 191 ++++++++++++++++++ haobang-security-dm/upgrade-doc/v5/README.md | 3 + 19 files changed, 553 insertions(+), 32 deletions(-) create mode 100644 haobang-security-dm/upgrade-doc/v5/02-json-format-error-debug-guide.md 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 e0269a0..11aae30 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 @@ -23,12 +23,25 @@ services: - SYSLOG_SERVER_ENABLED=true # 设备上送日志为 UTF-8；显式声明，Netty 与 WAL 共用 Spring 配置。 - SYSLOG_CHARSET=UTF-8 + # WAL 与应用日志共用宿主机挂载，但使用独立子目录，便于识别和清理。 + - SYSLOG_WAL_DIRECTORY=/app/logs/syslog-wal # 128KiB 及以上走大日志 Topic，单条最多接纳 6MiB；WAL/Dispatcher 均按字节背压。 - SYSLOG_MAX_FRAME_LENGTH=6291456 - SYSLOG_MESSAGE_MAX_BYTES=6291456 - SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES=131072 - - SYSLOG_WAL_MAX_PENDING_BYTES=268435456 - - SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES=268435456 + # 200KiB/条容量基线：2GiB 字节预算在 70% 时约 7,300 条 / 1.4GiB 触发 TCP 背压。 + - SYSLOG_WAL_QUEUE_CAPACITY=20000 + - SYSLOG_WAL_MAX_PENDING_BYTES=2147483648 + - SYSLOG_WAL_HIGH_WATERMARK_PERCENT=70 + - SYSLOG_WAL_LOW_WATERMARK_PERCENT=40 + - SYSLOG_WAL_FLUSH_RECORDS=200 + - SYSLOG_WAL_FLUSH_INTERVAL_MS=100 + - SYSLOG_WAL_SEGMENT_MAX_BYTES=2147483648 + - SYSLOG_DISPATCH_WORKER_THREADS=8 + - SYSLOG_DISPATCH_MAX_INFLIGHT=5000 + - SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES=1073741824 + # false：无效设备正文不进 WAL，仅按缓存周期打印 IP；true：写 WAL 和逐条 Terminal 审计。 + - SYSLOG_DISPATCH_TERMINAL_RECORD_ENABLED=false - TZ=Asia/Shanghai # 探针标识ID - APP_SERVICE_DEVICE_COLLECT_ID=1 @@ -52,6 +65,7 @@ services: - SPRING_KAFKA_PRODUCER_TOPIC=agent-01-syslog-topic - KAFKA_PRODUCER_LARGE_TOPIC=agent-01-syslog-topic.large - KAFKA_PRODUCER_MAX_REQUEST_SIZE=33554432 + - SPRING_KAFKA_PRODUCER_PROPERTIES_BUFFER_MEMORY=536870912 # 仅兼容仍使用独立原文归档的旧版 JAR；当前 WAL 版本已删除该 Java 实现。 - SYSLOG_RAW_ARCHIVE_ENABLED=false - SYSLOG_RAW_ARCHIVE_DIRECTORY=/data/syslog-archive @@ -78,16 +92,15 @@ services: # 定时任务配置 - SPRING_TASK_SCHEDULING_POOL_SIZE=5 - # JVM配置 - - JAVA_OPTS=-Xms2048m -Xmx4096m -XX:+UseG1GC + # JAVA_TOOL_OPTIONS 会被 JVM 自动读取；Dockerfile 的启动命令不会展开 JAVA_OPTS。 + - JAVA_TOOL_OPTIONS=-Xms8g -Xmx24g -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:MaxDirectMemorySize=8g -XX:+ExitOnOutOfMemoryError ports: - "8189:8189" # API端口 - "514:514/udp" # Syslog UDP - "514:514/tcp" # Syslog TCP volumes: + # 宿主机 /home/syslog/logs/syslog-wal 对应容器内 WAL 目录 /app/logs/syslog-wal。 - /home/syslog/logs:/app/logs - # 原始日志必须放在宿主机持久化盘，不能写入容器临时层。 - - /home/syslog/raw-archive:/data/syslog-archive networks: - xdr-network cap_add: 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 8686589..2e50865 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 @@ -136,6 +136,10 @@ public class AppProperties { @Value("${syslog.dispatch.shutdown-timeout-ms:30000}") private long syslogDispatchShutdownTimeoutMs; + /** false 时 Syslog 无效设备正文在 WAL 前丢弃，历史无效记录也不创建 Terminal 审计文件。 */ + @Value("${syslog.dispatch.terminal-record-enabled:false}") + private boolean syslogDispatchTerminalRecordEnabled; + @Value("${syslog.device-context.cache-ttl-ms:300000}") private long syslogDeviceContextCacheTtlMs; diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java index b6bb28b..664c704 100644 --- a/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java +++ b/haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java @@ -9,9 +9,17 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.time.LocalDateTime; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.LongAdder; /** @@ -31,6 +39,9 @@ public class DeviceContextResolver { private final Object[] locks = new Object[64]; private final LongAdder hits = new LongAdder(); private final LongAdder misses = new LongAdder(); + private final LongAdder ingressSnapshotRefreshFailed = new LongAdder(); + private volatile IngressSnapshot ingressSnapshot = IngressSnapshot.unavailable(); + private ScheduledExecutorService ingressSnapshotRefresher; public DeviceContextResolver(DeviceDeviceService deviceService, DeviceUnknownService unknownService, AppProperties properties) { @@ -40,6 +51,77 @@ public class DeviceContextResolver { for (int i = 0; i < locks.length; i++) locks[i] = new Object(); } + /** 网络线程只读取该快照，设备表加载固定在独立后台线程，禁止阻塞 Netty EventLoop。 */ + @PostConstruct + public void initializeIngressSnapshot() { + if (properties.isSyslogDispatchTerminalRecordEnabled()) return; + long refreshIntervalMs = Math.max(1L, properties.getSyslogDeviceContextCacheTtlMs()); + ingressSnapshotRefresher = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "syslog-ingress-device-snapshot"); + thread.setDaemon(true); + return thread; + }); + // 不阻塞 Spring/Netty 启动；首次快照就绪前消息保守进入 WAL。 + ingressSnapshotRefresher.scheduleWithFixedDelay( + this::refreshIngressSnapshot, 0L, refreshIntervalMs, TimeUnit.MILLISECONDS); + } + + public void refreshIngressSnapshot() { + if (properties.isSyslogDispatchTerminalRecordEnabled()) return; + try { + List devices = deviceService.getActiveDevices(); + if (devices == null) throw new IllegalStateException("活动设备列表为空引用"); + Map> grouped = new HashMap<>(); + for (DeviceDevice device : devices) { + if (device == null || device.getIp() == null || device.getIp().trim().isEmpty()) continue; + grouped.computeIfAbsent(device.getIp(), ignored -> new ArrayList<>()).add(device); + } + Map resolutions = new HashMap<>(); + for (Map.Entry> entry : grouped.entrySet()) { + List sameIp = entry.getValue(); + Resolution resolution = sameIp.size() == 1 + ? validateAssignment(sameIp.get(0)) + : Resolution.terminal(Type.DUPLICATE_DEVICE, null, + "来源 IP 存在多条设备记录: " + entry.getKey()); + resolutions.put(entry.getKey(), resolution); + } + ingressSnapshot = IngressSnapshot.ready(resolutions); + log.info("Syslog 入站设备内存快照已刷新: devices={}, sourceIps={}", + devices.size(), resolutions.size()); + } catch (Throwable failure) { + ingressSnapshotRefreshFailed.increment(); + log.error("Syslog 入站设备内存快照刷新失败，继续使用上一版本；无可用快照时消息保守进入 WAL", failure); + } + } + + @PreDestroy + public void shutdownIngressSnapshotRefresher() { + if (ingressSnapshotRefresher != null) ingressSnapshotRefresher.shutdownNow(); + } + + /** 纯内存判断；绝不访问 Redis、达梦或 device_unknown。 */ + public IngressStatus resolveIngress(String sourceIp, String protocol) { + IngressSnapshot snapshot = ingressSnapshot; + if (!snapshot.ready) return IngressStatus.UNAVAILABLE; + Resolution resolution = snapshot.resolutions.get(sourceIp); + if (resolution == null) { + logIngressInvalidOnce(snapshot, Type.UNKNOWN_DEVICE, sourceIp, protocol, + "来源 IP 未注册: " + sourceIp); + return IngressStatus.INVALID; + } + if (resolution.isRegistered()) return IngressStatus.REGISTERED; + logIngressInvalidOnce(snapshot, resolution.type, sourceIp, protocol, resolution.reason); + return IngressStatus.INVALID; + } + + private void logIngressInvalidOnce(IngressSnapshot snapshot, Type type, String sourceIp, + String protocol, String reason) { + if (snapshot.loggedInvalidIps.putIfAbsent(sourceIp, Boolean.TRUE) == null) { + log.warn("Syslog 无效设备日志将直接丢弃: type={}, sourceIp={}, protocol={}, reason={}", + type, sourceIp, protocol, reason); + } + } + public Resolution resolve(String sourceIp, String protocol) { long now = System.currentTimeMillis(); CacheEntry cached = cache.get(sourceIp); @@ -78,8 +160,14 @@ public class DeviceContextResolver { cache.put(sourceIp, new CacheEntry(resolution, now + properties.getSyslogDeviceContextCacheTtlMs())); if (resolution.type != Type.REGISTERED) { - log.warn("Syslog 设备接入终态: type={}, sourceIp={}, reason={}", - resolution.type, sourceIp, resolution.reason); + if (properties.isSyslogDispatchTerminalRecordEnabled()) { + log.warn("Syslog 无效设备日志将记录 Terminal 审计文件: type={}, sourceIp={}, reason={}", + resolution.type, sourceIp, resolution.reason); + } else { + // 无效结果会缓存；同一 IP 只在首次判定或缓存刷新时打印，避免逐条日志再次形成 IO 瓶颈。 + log.warn("Syslog 无效设备日志将直接丢弃: type={}, sourceIp={}, reason={}", + resolution.type, sourceIp, resolution.reason); + } } return resolution; } @@ -134,6 +222,9 @@ public class DeviceContextResolver { public long getHitCount() { return hits.sum(); } public long getMissCount() { return misses.sum(); } public int getCacheSize() { return cache.size() + deviceIdCache.size(); } + public boolean isIngressSnapshotReady() { return ingressSnapshot.ready; } + public int getIngressSnapshotSize() { return ingressSnapshot.resolutions.size(); } + public long getIngressSnapshotRefreshFailedCount() { return ingressSnapshotRefreshFailed.sum(); } /** 未知设备仍保留原业务登记能力，但最多每个本地缓存 TTL 执行一次，不占用网络线程。 */ private void recordUnknown(String sourceIp, String protocol) { @@ -165,6 +256,26 @@ public class DeviceContextResolver { } public enum Type { REGISTERED, UNKNOWN_DEVICE, DUPLICATE_DEVICE, NOT_ASSIGNED } + public enum IngressStatus { REGISTERED, INVALID, UNAVAILABLE } + + private static final class IngressSnapshot { + private final boolean ready; + private final Map resolutions; + private final ConcurrentHashMap loggedInvalidIps = new ConcurrentHashMap<>(); + + private IngressSnapshot(boolean ready, Map resolutions) { + this.ready = ready; + this.resolutions = java.util.Collections.unmodifiableMap(new HashMap<>(resolutions)); + } + + private static IngressSnapshot unavailable() { + return new IngressSnapshot(false, java.util.Collections.emptyMap()); + } + + private static IngressSnapshot ready(Map resolutions) { + return new IngressSnapshot(true, resolutions); + } + } public static final class Resolution { private final Type type; 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 97a7789..71212fd 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 @@ -47,6 +47,7 @@ public class SyslogWalDispatcher { private final Map completedOutOfOrder = new HashMap<>(); private final LongAdder kafkaAcked = new LongAdder(); private final LongAdder terminalFiltered = new LongAdder(); + private final LongAdder terminalDropped = new LongAdder(); private final LongAdder retryCount = new LongAdder(); private final LongAdder largeDispatched = new LongAdder(); private final AtomicLong inflight = new AtomicLong(); @@ -88,10 +89,15 @@ public class SyslogWalDispatcher { scheduler.scheduleWithFixedDelay(this::logStatus, properties.getSyslogStatusLogIntervalMs(), properties.getSyslogStatusLogIntervalMs(), TimeUnit.MILLISECONDS); - log.info("Syslog WAL Dispatcher 已启动: workerThreads={}, maxInflight={}, maxInflightBytes={}, checkpointSequence={}", + log.info("Syslog WAL Dispatcher 已启动: workerThreads={}, maxInflight={}, maxInflightBytes={}, " + + "terminalRecordEnabled={}, checkpointSequence={}", properties.getSyslogDispatchWorkerThreads(), properties.getSyslogDispatchMaxInflight(), properties.getSyslogDispatchMaxInflightBytes(), + properties.isSyslogDispatchTerminalRecordEnabled(), walService.getCheckpointSequence()); + if (!properties.isSyslogDispatchTerminalRecordEnabled()) { + log.warn("无效设备日志直接丢弃已启用: 新 Syslog 正文不进 WAL，历史记录不创建 Terminal 文件"); + } } private void readLoop() { @@ -147,7 +153,11 @@ public class SyslogWalDispatcher { if (!resolution.isRegistered()) { try { - walService.recordTerminal(record, resolution.getType().name(), resolution.getReason()); + if (properties.isSyslogDispatchTerminalRecordEnabled()) { + walService.recordTerminal(record, resolution.getType().name(), resolution.getReason()); + } else { + terminalDropped.increment(); + } terminalFiltered.increment(); markComplete(record); } catch (Throwable failure) { @@ -252,14 +262,16 @@ public class SyslogWalDispatcher { private void logStatus() { if (!running.get()) return; log.info("Syslog WAL状态: networkReceived={}, appended={}, admissionOversized={}, walPendingApprox={}, walPendingBytes={}, walQueue={}/{}, durableSequence={}, " - + "checkpointSequence={}, kafkaInflight={}, kafkaInflightBytes={}, kafkaAcked={}, largeDispatched={}, terminalFiltered={}, " + + "checkpointSequence={}, kafkaInflight={}, kafkaInflightBytes={}, kafkaAcked={}, largeDispatched={}, " + + "terminalRecordEnabled={}, terminalFiltered={}, terminalDropped={}, " + "retry={}, tcpPauseTotal={}, udpDropped={}, deviceCacheSize={}, deviceCacheHits={}, " + "deviceCacheMisses={}", SyslogMessageHandler.getNetworkReceivedCount(), walService.getAppendedCount(), walService.getOversizedCount(), walService.getPendingRecordsApprox(), walService.getPendingAppendBytes(), walService.getQueueSize(), walService.getQueueCapacity(), walService.getDurableSequence(), walService.getCheckpointSequence(), inflight.get(), byteBudget == null ? 0L : byteBudget.used(), - kafkaAcked.sum(), largeDispatched.sum(), terminalFiltered.sum(), + kafkaAcked.sum(), largeDispatched.sum(), properties.isSyslogDispatchTerminalRecordEnabled(), + terminalFiltered.sum(), terminalDropped.sum(), retryCount.sum(), SyslogMessageHandler.getTcpPauseCount(), SyslogMessageHandler.getUdpDroppedCount(), deviceResolver.getCacheSize(), deviceResolver.getHitCount(), deviceResolver.getMissCount()); @@ -318,6 +330,8 @@ public class SyslogWalDispatcher { public long getInflight() { return inflight.get(); } public long getKafkaAcked() { return kafkaAcked.sum(); } public long getTerminalFiltered() { return terminalFiltered.sum(); } + public long getTerminalDropped() { return terminalDropped.sum(); } + public boolean isTerminalRecordEnabled() { return properties.isSyslogDispatchTerminalRecordEnabled(); } public long getRetryCount() { return retryCount.sum(); } public long getLargeDispatched() { return largeDispatched.sum(); } public long getInflightBytes() { return byteBudget == null ? 0L : byteBudget.used(); } 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 ce238b8..e6c96ac 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 @@ -30,6 +30,8 @@ public class SyslogWalStatusController { 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()); @@ -55,13 +57,18 @@ public class SyslogWalStatusController { 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; } } 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 59d7180..c356031 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,5 +1,6 @@ package com.netty; +import com.haobang.dispatch.DeviceContextResolver; import com.haobang.wal.SyslogIngressEvent; import com.haobang.wal.SyslogWalService; import com.kafka.kafkaProducer; @@ -23,9 +24,9 @@ import java.util.concurrent.atomic.LongAdder; /** * Syslog 网络接入处理器。 * - *

这里只解帧、生成首次 event_id/receive_time 并提交 WAL，不再查达梦、等待 Kafka ACK - * 或写接收审计。TCP 在 WAL 达到高水位时关闭 AUTO_READ，并保留已经解出的当前帧，待 WAL - * 恢复到低水位后按原顺序继续追加。

+ *

性能优先模式会在生成事件和提交 WAL 前按来源 IP 判断设备；无效设备正文直接丢弃。 + * 有效设备随后生成首次 event_id/receive_time 并提交 WAL。TCP 在 WAL 达到高水位时关闭 + * AUTO_READ，并保留已经解出的当前帧，待 WAL 恢复到低水位后按原顺序继续追加。

*/ public class SyslogMessageHandler extends SimpleChannelInboundHandler { private static final Logger log = LoggerFactory.getLogger(SyslogMessageHandler.class); @@ -34,17 +35,31 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { private static final LongAdder TCP_PAUSE_COUNT = new LongAdder(); private static final LongAdder UDP_DROPPED = new LongAdder(); 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(); /** 每个 TCP Channel 独立 Handler，pending 只保存关闭 AUTO_READ 前已进入 Pipeline 的帧。 */ private final Deque pendingTcp = new ArrayDeque<>(); private final Charset syslogCharset; private final SyslogWalService walService; + private final DeviceContextResolver deviceResolver; + private final boolean dropInvalidBeforeWal; private boolean retryScheduled; private boolean autoReadPaused; SyslogMessageHandler(Charset syslogCharset, SyslogWalService walService) { + this(syslogCharset, walService, null, false); + } + + SyslogMessageHandler(Charset syslogCharset, SyslogWalService walService, + DeviceContextResolver deviceResolver, boolean dropInvalidBeforeWal) { this.syslogCharset = java.util.Objects.requireNonNull(syslogCharset, "syslogCharset"); this.walService = java.util.Objects.requireNonNull(walService, "walService"); + if (dropInvalidBeforeWal && deviceResolver == null) { + throw new IllegalArgumentException("性能优先模式必须提供 DeviceContextResolver"); + } + this.deviceResolver = deviceResolver; + this.dropInvalidBeforeWal = dropInvalidBeforeWal; } @Override @@ -77,13 +92,14 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { return; } + NETWORK_RECEIVED.increment(); + if (dropInvalidBeforeWal && shouldDropBeforeWal(sourceIp, protocol)) return; + Instant receivedInstant = Instant.now(); SyslogIngressEvent event = walService.prepareForWal(new SyslogIngressEvent( kafkaProducer.newEventId(receivedInstant), LocalDateTime.ofInstant(receivedInstant, RECEIVE_TIME_ZONE), sourceIp, sourcePort, protocol, message)); - NETWORK_RECEIVED.increment(); - if ("UDP".equals(protocol)) { acceptUdp(walService, event); } else { @@ -188,6 +204,20 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler { 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(); } + + 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(); + return true; + } + // 后台快照暂不可用时不访问数据库、不阻塞网络线程；保守进入 WAL 等待 Dispatcher 判断。 + DEVICE_LOOKUP_FAILED.increment(); + return false; + } private void recordOversized(SyslogIngressEvent event, String protocol) { OVERSIZED_DROPPED.increment(); 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 a927c55..b3f9dc4 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 @@ -1,6 +1,7 @@ package com.netty; import com.haobang.config.AppProperties; +import com.haobang.dispatch.DeviceContextResolver; import com.haobang.util.SpringContextUtil; import com.haobang.wal.SyslogWalService; import io.netty.bootstrap.Bootstrap; @@ -41,6 +42,7 @@ public class SyslogServer { private void startTcpServer() throws InterruptedException { AppProperties properties = SpringContextUtil.getBean(AppProperties.class); SyslogWalService walService = SpringContextUtil.getBean(SyslogWalService.class); + DeviceContextResolver deviceResolver = SpringContextUtil.getBean(DeviceContextResolver.class); Charset charset = walService.getSyslogCharset(); EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); @@ -50,7 +52,8 @@ public class SyslogServer { b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childHandler(new SyslogTcpChannelInitializer( - properties.getSyslogMaxFrameLength(), charset, walService)) + properties.getSyslogMaxFrameLength(), charset, walService, deviceResolver, + !properties.isSyslogDispatchTerminalRecordEnabled())) .option(ChannelOption.SO_BACKLOG, 128) .childOption(ChannelOption.SO_KEEPALIVE, true); @@ -67,6 +70,7 @@ public class SyslogServer { private void startUdpServer() throws InterruptedException { AppProperties properties = SpringContextUtil.getBean(AppProperties.class); SyslogWalService walService = SpringContextUtil.getBean(SyslogWalService.class); + DeviceContextResolver deviceResolver = SpringContextUtil.getBean(DeviceContextResolver.class); Charset charset = walService.getSyslogCharset(); EventLoopGroup group = new NioEventLoopGroup(); @@ -74,7 +78,8 @@ public class SyslogServer { Bootstrap b = new Bootstrap(); b.group(group) .channel(NioDatagramChannel.class) - .handler(new SyslogUdpChannelInitializer(charset, walService)) + .handler(new SyslogUdpChannelInitializer(charset, walService, deviceResolver, + !properties.isSyslogDispatchTerminalRecordEnabled())) .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 add6901..072af57 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 @@ -16,6 +16,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicReference; import com.haobang.config.AppProperties; +import com.haobang.dispatch.DeviceContextResolver; import com.haobang.wal.SyslogWalService; import java.nio.charset.Charset; @@ -27,6 +28,7 @@ public class SyslogServerBoth { private final AppProperties properties; private final SyslogWalService walService; + private final DeviceContextResolver deviceResolver; private final Charset syslogCharset; private final int tcpPort; private final int udpPort; @@ -41,9 +43,11 @@ public class SyslogServerBoth { private final ExecutorService serverExecutor = Executors.newFixedThreadPool(2); - public SyslogServerBoth(AppProperties properties, SyslogWalService walService, Charset syslogCharset) { + public SyslogServerBoth(AppProperties properties, SyslogWalService walService, + DeviceContextResolver deviceResolver, 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.syslogCharset = java.util.Objects.requireNonNull(syslogCharset, "syslogCharset"); this.tcpPort = properties.getSyslogTcpPort(); this.udpPort = properties.getSyslogUdpPort(); @@ -113,7 +117,8 @@ public class SyslogServerBoth { b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childHandler(new SyslogTcpChannelInitializer( - properties.getSyslogMaxFrameLength(), syslogCharset, walService)) + properties.getSyslogMaxFrameLength(), syslogCharset, walService, deviceResolver, + !properties.isSyslogDispatchTerminalRecordEnabled())) // [SYSLOG-ALARM][NINGRESS][已修复] backlog/接收缓冲配置化；实际生效上限仍受操作系统 somaxconn/rmem_max 约束。 .option(ChannelOption.SO_BACKLOG, properties.getSyslogTcpBacklog()) .childOption(ChannelOption.SO_KEEPALIVE, true) @@ -135,7 +140,8 @@ public class SyslogServerBoth { Bootstrap b = new Bootstrap(); b.group(udpGroup) .channel(NioDatagramChannel.class) - .handler(new SyslogUdpChannelInitializer(syslogCharset, walService)) + .handler(new SyslogUdpChannelInitializer(syslogCharset, walService, deviceResolver, + !properties.isSyslogDispatchTerminalRecordEnabled())) // [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 0d0110e..e32124b 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 @@ -1,6 +1,7 @@ package com.netty; import com.haobang.config.AppProperties; +import com.haobang.dispatch.DeviceContextResolver; import com.haobang.dispatch.SyslogWalDispatcher; import com.haobang.wal.SyslogWalService; import org.slf4j.Logger; @@ -24,16 +25,18 @@ public class SyslogServerLifecycle implements SmartLifecycle { private static final Logger log = LoggerFactory.getLogger(SyslogServerLifecycle.class); private final AppProperties properties; private final SyslogWalService walService; + private final DeviceContextResolver deviceResolver; @SuppressWarnings("unused") private final SyslogWalDispatcher dispatcher; private final AtomicBoolean running = new AtomicBoolean(); private SyslogServerBoth server; public SyslogServerLifecycle(AppProperties properties, SyslogWalService walService, - SyslogWalDispatcher dispatcher) { + SyslogWalDispatcher dispatcher, DeviceContextResolver deviceResolver) { this.properties = properties; this.walService = walService; this.dispatcher = dispatcher; + this.deviceResolver = deviceResolver; } @Override @@ -41,7 +44,7 @@ public class SyslogServerLifecycle implements SmartLifecycle { if (!running.compareAndSet(false, true)) return; try { Charset charset = walService.getSyslogCharset(); - server = new SyslogServerBoth(properties, walService, charset); + server = new SyslogServerBoth(properties, walService, deviceResolver, 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 50ecad6..be3a75c 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 @@ -1,6 +1,7 @@ package com.netty; +import com.haobang.dispatch.DeviceContextResolver; import com.haobang.wal.SyslogWalService; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; @@ -18,12 +19,17 @@ public class SyslogTcpChannelInitializer extends ChannelInitializer { private final Charset syslogCharset; private final SyslogWalService walService; + private final DeviceContextResolver deviceResolver; + private final boolean dropInvalidBeforeWal; - SyslogUdpChannelInitializer(Charset syslogCharset, SyslogWalService walService) { + SyslogUdpChannelInitializer(Charset syslogCharset, SyslogWalService walService, + DeviceContextResolver deviceResolver, boolean dropInvalidBeforeWal) { this.syslogCharset = java.util.Objects.requireNonNull(syslogCharset, "syslogCharset"); this.walService = java.util.Objects.requireNonNull(walService, "walService"); + this.deviceResolver = java.util.Objects.requireNonNull(deviceResolver, "deviceResolver"); + this.dropInvalidBeforeWal = dropInvalidBeforeWal; } @Override @@ -29,6 +35,7 @@ public class SyslogUdpChannelInitializer extends ChannelInitializer wal.getAppendedCount() == 1L); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + when(resolver.resolve(anyString(), anyString())).thenReturn(DeviceContextResolver.Resolution.terminal( + DeviceContextResolver.Type.UNKNOWN_DEVICE, null, "not registered")); + kafkaProducer producer = mock(kafkaProducer.class); + + dispatcher = new SyslogWalDispatcher(wal, resolver, producer, properties); + dispatcher.start(); + + waitUntil(() -> wal.getCheckpointSequence() == 1L); + assertEquals(1L, dispatcher.getTerminalFiltered()); + assertEquals(1L, dispatcher.getTerminalDropped()); + assertEquals(0L, wal.getTerminalRecordedCount()); + try (Stream files = Files.list(tempDirectory.resolve("terminal"))) { + assertEquals(0L, files.count()); + } + verify(producer, never()).messagePush(any(), anyString(), anyString(), anyString(), any()); + } + @Test void largeRecordUsesDedicatedKafkaLaneAndReleasesByteBudgetAfterAck() throws Exception { AppProperties properties = properties(tempDirectory); 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 461b199..103f746 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,5 +1,6 @@ package com.netty; +import com.haobang.dispatch.DeviceContextResolver; import com.haobang.wal.SyslogWalService; import io.netty.channel.embedded.EmbeddedChannel; import org.junit.jupiter.api.Test; @@ -9,7 +10,10 @@ import java.nio.charset.StandardCharsets; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; class SyslogMessageHandlerBackpressureTest { @@ -28,4 +32,40 @@ class SyslogMessageHandlerBackpressureTest { assertFalse(channel.config().isAutoRead()); channel.close(); } + + @Test + void invalidDeviceIsDroppedBeforeWalInPerformanceMode() { + SyslogWalService wal = mock(SyslogWalService.class); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + when(resolver.resolveIngress(anyString(), anyString())) + .thenReturn(DeviceContextResolver.IngressStatus.INVALID); + EmbeddedChannel channel = new EmbeddedChannel( + new SyslogMessageHandler(StandardCharsets.UTF_8, wal, resolver, true)); + channel.connect(new InetSocketAddress("192.168.4.19", 5514)); + + channel.writeInbound("<134>Aug 11 20:00:00 invalid device event"); + + verify(resolver).resolveIngress("embedded", "TCP"); + verify(wal, never()).prepareForWal(any()); + verify(wal, never()).tryAppend(any()); + channel.close(); + } + + @Test + void unavailableDeviceSnapshotFallsBackToWalWithoutBlockingLookup() { + SyslogWalService wal = mock(SyslogWalService.class); + when(wal.prepareForWal(any())).thenAnswer(invocation -> invocation.getArgument(0)); + when(wal.tryAppend(any())).thenReturn(SyslogWalService.AdmissionResult.ACCEPTED); + DeviceContextResolver resolver = mock(DeviceContextResolver.class); + when(resolver.resolveIngress(anyString(), anyString())) + .thenReturn(DeviceContextResolver.IngressStatus.UNAVAILABLE); + EmbeddedChannel channel = new EmbeddedChannel( + new SyslogMessageHandler(StandardCharsets.UTF_8, wal, resolver, true)); + channel.connect(new InetSocketAddress("192.168.4.19", 5514)); + + channel.writeInbound("<134>Aug 11 20:00:00 lookup failure event"); + + verify(wal).tryAppend(any()); + channel.close(); + } } diff --git a/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md b/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md index 45b0535..d085374 100644 --- a/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md +++ b/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md @@ -23,19 +23,21 @@ syslog-serve/docker-compose/docker-compose-serve-prod.yaml - SYSLOG_MESSAGE_MAX_BYTES=6291456 - SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES=131072 - # 200KiB/条容量基线；WAL 高水位约在 14,000 条、2.7~2.8GiB 对齐触发。 + # 200KiB/条容量基线；2GiB 字节预算在 70% 时约 7,300 条、1.4GiB 触发背压。 - SYSLOG_WAL_QUEUE_CAPACITY=20000 - - SYSLOG_WAL_MAX_PENDING_BYTES=4294967296 + - SYSLOG_WAL_MAX_PENDING_BYTES=2147483648 - SYSLOG_WAL_HIGH_WATERMARK_PERCENT=70 - SYSLOG_WAL_LOW_WATERMARK_PERCENT=40 - SYSLOG_WAL_FLUSH_RECORDS=200 - SYSLOG_WAL_FLUSH_INTERVAL_MS=100 - - SYSLOG_WAL_SEGMENT_MAX_BYTES=4294967296 + - SYSLOG_WAL_SEGMENT_MAX_BYTES=2147483648 # Kafka 投递限制正文最多 1GiB 在途，并增加加密后的 Producer 缓冲。 - SYSLOG_DISPATCH_WORKER_THREADS=8 - SYSLOG_DISPATCH_MAX_INFLIGHT=5000 - SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES=1073741824 + # 性能优先：无效设备正文不进 WAL，仅按设备缓存周期打印来源 IP。 + - SYSLOG_DISPATCH_TERMINAL_RECORD_ENABLED=false - SPRING_KAFKA_PRODUCER_PROPERTIES_BUFFER_MEMORY=536870912 # Dockerfile 不展开 JAVA_OPTS，使用 JVM 自动读取的 JAVA_TOOL_OPTIONS。 @@ -53,10 +55,12 @@ syslog-serve/docker-compose/docker-compose-serve-prod.yaml - `SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES`：达到该值后走大日志 Topic； - `SYSLOG_WAL_QUEUE_CAPACITY`、`SYSLOG_WAL_MAX_PENDING_BYTES`：共同限制 WAL 接入条数和原文字节量； - `SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES`：限制 Kafka Dispatcher 在途正文总量； +- `SYSLOG_DISPATCH_TERMINAL_RECORD_ENABLED`：`false` 在 WAL 前丢弃 Syslog 无效设备正文并按缓存周期打印 IP；`true` 写 WAL 并逐条保存 Terminal 审计文件； - `JAVA_TOOL_OPTIONS`：设置实际 JVM 堆和直接内存上限，不能替换成当前 Dockerfile 不会展开的 `JAVA_OPTS`。 `SYSLOG_MAX_FRAME_LENGTH` 必须大于或等于 `SYSLOG_MESSAGE_MAX_BYTES`。如果设备发送 GBK，应将 -`SYSLOG_CHARSET` 改为 `GBK`。 +`SYSLOG_CHARSET` 改为 `GBK`。无效设备判断只读取后台定时刷新的内存快照，不会在 Netty 网络线程访问 Redis 或达梦； +首次快照暂未就绪或刷新失败时，消息保守进入 WAL。 ## 2. syslog-consumer-rule diff --git a/haobang-security-dm/upgrade-doc/v5/02-json-format-error-debug-guide.md b/haobang-security-dm/upgrade-doc/v5/02-json-format-error-debug-guide.md new file mode 100644 index 0000000..ba2c30c --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v5/02-json-format-error-debug-guide.md @@ -0,0 +1,191 @@ +# 天眼/微步 JSON 解析异常排查入口 + +## 目标 + +定位一条日志为什么进入 `syslog_non_normal_message` 并标记为“JSON解析异常”。 + +只分析当前设备绑定的规则： + +- 天眼设备重点看规则 `129 / skyeye_alarm_log`； +- 微步设备看该设备实际绑定的规则，现场已见 `150 / weibu_TDP_alarmlog_JSON`、 + `158 / weibujsomv2`、`159 / weibujson`； +- `Threatbook[pid]: {...}` 属于微步/TDP 报文，不要拿规则129分析； +- 设备与规则的绑定以 `dm_normalize_rule_device` 为准，不要模糊查询全表后把所有规则都拿来匹配。 + +## 第一步：取得两个输入 + +### 1. 解析前的完整原文 + +不要使用达梦非标表里带有以下标记的内容判断 JSON 是否合法： + +```text +[HB_XDR_TRUNCATED original_utf8_bytes=...] +``` + +该内容已经在规则计算完成后按 2KB 限制截断，不是 Jackson 实际收到的报文。 + +使用 v5 脚本按设备采集解析前原文： + +```bash +chmod 700 upgrade-doc/v5/capture-device-raw-logs.sh + +upgrade-doc/v5/capture-device-raw-logs.sh \ + <设备ID> 120 20 \ + http://127.0.0.1:8289/xdrrule \ + /data/haobang/raw-log-captures +``` + +调试需要生成目录中的 `result.json` 和 `raw/*.log`。如果接口不可用，再从原始 InfluxDB 或 +`syslog-serve` 接收日志按 `device_id + receive_time/event_id` 导出完整 `message`。 + +### 2. 当前设备绑定的完整规则 + +`rule_content` 最长约一万多字符，在达梦 `disql` 中设置 20000 字符并 spool 到文件： + +```sql +SET LONG 20000 +SET LONGCHUNKSIZE 20000 +SET LINESIZE 32767 +SET PAGESIZE 0 +SET TRIMSPOOL ON +SPOOL /tmp/device-normalize-rules.txt + +SELECT b.device_id, + r.id AS rule_id, + r.name, + r.display_name, + r.data_type, + DBMS_LOB.GETLENGTH(r.rule_content) AS rule_content_length, + DBMS_LOB.SUBSTR(r.rule_content, 20000, 1) AS rule_content + FROM dm_normalize_rule_device b + JOIN dm_normalize_rule r ON r.id = b.normalize_rule_id + WHERE b.device_id = <设备ID> + AND r.del_flag = '0' + ORDER BY r.id; + +SPOOL OFF +``` + +若长度超过 20000，再取下一段： + +```sql +DBMS_LOB.SUBSTR(rule_content, 20000, 20001) +``` + +## 第二步：在代码中拿到 Jackson 的真实输入 + +当前主链路： + +```text +SysLogProcessor.prepareNormalization() + -> LogNormalProcessor.prepare() + -> LogNormalProcessor.collectCandidates() + -> CompiledRule.isApplicableTo() + -> CompiledRule.parse() + -> logNormalData.ParserMessageJsonType() + -> JsonParser.extractJsonFromLogMessage()（必要时） + -> NestedJsonParserUtil.safeParseJson() + -> Jackson +``` + +建议断点： + +| 位置 | 观察内容 | +| --- | --- | +| `LogNormalProcessor.java:516` | 当前设备加载了哪些规则，每条规则是否进入候选解析 | +| `CompiledRule.java:72` | 当前 `ruleId`、原始 `message` 和 cropper 配置 | +| `CompiledRule.java:94` | cropper 结果是否需要再次截取 `{...}` | +| `CompiledRule.java:99` | 变量 `cropped`，即 Compiled 链路交给 Jackson 的完整字符串 | +| `CompiledRule.java:124` | `head_key` 是否真的能标识这类报文 | +| `NestedJsonParserUtil.java:41` | Jackson 的首次解析结果及准确行列错误 | +| `LogNormalProcessor.java:1014` | 关闭 Compiled 优化时的 legacy 入口，观察 `strJson` | + +相关源码： + +```text +syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java +syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java +syslog-consumer-rule/src/main/java/com/Modules/NormalData/logNormalData.java +syslog-consumer-rule/src/main/java/com/common/util/JsonParser.java +syslog-consumer-rule/src/main/java/com/common/util/NestedJsonParserUtil.java +syslog-consumer-rule/src/main/java/com/common/service/RawLaneFieldLimiter.java +``` + +对每条失败样本保存以下信息即可： + +```text +deviceId +ruleId / ruleName +完整原文 message +cropper_params +截取结果 cropped（或 legacy 的 strJson） +Jackson 原始错误及行列号 +``` + +不要为了调试在生产 `INFO` 日志中持续打印完整原文；使用采集文件、断点或只针对一条脱敏样本的测试。 + +## 第三步：根据真实输入分类 + +| 现象 | 结论 | 处理位置 | +| --- | --- | --- | +| `cropped` 仍含 Syslog 头或天眼分隔前缀 | 截取未完成 | `cropper_params` 或 JSON 对象边界截取 | +| `cropped` 缺少开头 `{`、结尾 `}` 或被截在中间 | cropper 范围错误 | 优先修当前规则 cropper | +| 报文不含规则的格式特征，却进入该规则 | 候选规则归属错误 | `isApplicableTo()`/规则特征配置 | +| `cropped` 是完整外层 JSON，但 `req_body` 等字段内有未转义双引号、反斜杠或真实换行 | 设备报文/嵌套字段转义错误 | 优先让设备修；必要时做设备/规则专用预处理 | +| 完整原文在进入规则前已经不完整 | 上游提前截断 | 查 `syslog-serve`、Kafka 和原始存储链路 | +| 只有达梦非标字段被截断 | 与解析失败无关 | 不修改解析代码 | + +`JsonParser.extractJsonFromLogMessage()` 的默认行为是取第一个 `{` 到最后一个 `}`。配置了 cropper 时， +先执行 `ParserMessageJsonType()`；如果结果不是完整 `{...}`，当前 Compiled JSON 路径才会再次执行对象边界截取。 + +## 天眼与微步的现有复现入口 + +测试文件： + +```text +syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java +``` + +直接参考两个用例： + +- `threatbookUsesTdpRuleAfterFixedOffsetHoneypotRulesAreRejected`：微步 Threatbook 报文应选择规则150， + 不能被158/159的固定偏移规则提前判为非标； +- `skyeyeCropperPrefixIsNarrowedToOuterJsonBeforeJackson`：天眼规则129截取后，Jackson 输入必须是完整外层 JSON。 + +把现场完整原文脱敏后增加为 fixture，并使用现场规则的真实 `cropper_params`。至少断言： + +```text +选中的 normalize_rule_id +一个关键字段的映射值 +未生成“JSON解析异常”非标记录 +``` + +使用 JDK 8 执行： + +```bash +mvn -Dtest=LogNormalProcessorRuleSelectionTest test +``` + +## 必看的历史修改 + +```bash +git show a70707c72969c7979aef9c0221fd55360c365227 +git show a70707c^:syslog-consumer-rule/src/main/java/com/common/util/NestedJsonParserUtil.java +git show c6e3505 +``` + +- `a70707c...` 删除了首次失败后的全局坏 JSON 猜测修复和第二次 Jackson 解析； +- 旧版会依次尝试修复未转义引号、单引号、反斜杠、缺失引号、数值、布尔和 null； +- `c6e3505` 修改了候选规则失败处理及 JSON 对象边界截取，避免前一条不适用规则失败后立即转非标。 + +判断问题是否由删除旧修复引起，必须对同一份完整原文记录四项结果： + +```text +当前首次解析结果 +旧版首次解析结果 +旧版修复后的字符串（并指出实际生效步骤） +旧版第二次解析结果 +``` + +没有这四项证据，不要恢复旧版全局修复。全局恢复会给每条坏 JSON 增加多次扫描、正则替换、字符串分配和 +第二次 Jackson 解析。若确实需要兼容设备缺陷，优先限定到指定设备及指定规则，最后才考虑通用解析器。 diff --git a/haobang-security-dm/upgrade-doc/v5/README.md b/haobang-security-dm/upgrade-doc/v5/README.md index 53b5c8a..d996172 100644 --- a/haobang-security-dm/upgrade-doc/v5/README.md +++ b/haobang-security-dm/upgrade-doc/v5/README.md @@ -1,5 +1,8 @@ # v5 现场设备原始日志采样 +天眼/微步 JSON 异常的代码入口和现场开发排查顺序见 +[`02-json-format-error-debug-guide.md`](./02-json-format-error-debug-guide.md)。 + ## 功能说明 v5 为 `syslog-consumer-rule` 增加按设备短时采集原始日志的诊断接口，并提供自动落盘脚本。