nanChen/ai-security-xdr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(syslog): 优化WAL配置并在入站阶段丢弃无效设备日志

Browse Source 
- 默认在进入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
This commit is contained in:
王绍宏
2026-09-15 22:10:46 +08:00
parent 5a86a4f950
commit faa9ffe2d9
19 changed files with 553 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml
+19 -6
View File
@@ -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:
haobang-security-dm/syslog-serve/src/main/java/com/haobang/config/AppProperties.java
+4
View File
@@ -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;
haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/DeviceContextResolver.java
+113 -2
View File
@@ -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<DeviceDevice> devices = deviceService.getActiveDevices();
if (devices == null) throw new IllegalStateException("活动设备列表为空引用");
Map<String, List<DeviceDevice>> 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<String, Resolution> resolutions = new HashMap<>();
for (Map.Entry<String, List<DeviceDevice>> entry : grouped.entrySet()) {
List<DeviceDevice> 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<String, Resolution> resolutions;
private final ConcurrentHashMap<String, Boolean> loggedInvalidIps = new ConcurrentHashMap<>();
private IngressSnapshot(boolean ready, Map<String, Resolution> 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<String, Resolution> resolutions) {
return new IngressSnapshot(true, resolutions);
}
}
public static final class Resolution {
private final Type type;
haobang-security-dm/syslog-serve/src/main/java/com/haobang/dispatch/SyslogWalDispatcher.java
+18 -4
View File
@@ -47,6 +47,7 @@ public class SyslogWalDispatcher {
private final Map<Long, SyslogWalRecord> 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(); }
haobang-security-dm/syslog-serve/src/main/java/com/haobang/wal/SyslogWalStatusController.java
+7
View File
@@ -30,6 +30,8 @@ public class SyslogWalStatusController {
Map<String, Object> 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;
}
}
haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogMessageHandler.java
+35 -5
View File
@@ -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 网络接入处理器。
*
* <p>这里只解帧、生成首次 event_id/receive_time 并提交 WAL，不再查达梦、等待 Kafka ACK
* 或写接收审计。TCP 在 WAL 达到高水位时关闭 AUTO_READ，并保留已经解出的当前帧，待 WAL
* 恢复到低水位后按原顺序继续追加。</p>
* <p>性能优先模式会在生成事件和提交 WAL 前按来源 IP 判断设备；无效设备正文直接丢弃。
* 有效设备随后生成首次 event_id/receive_time 并提交 WAL。TCP 在 WAL 达到高水位时关闭
* AUTO_READ，并保留已经解出的当前帧，待 WAL 恢复到低水位后按原顺序继续追加。</p>
*/
public class SyslogMessageHandler extends SimpleChannelInboundHandler<Object> {
private static final Logger log = LoggerFactory.getLogger(SyslogMessageHandler.class);
@@ -34,17 +35,31 @@ public class SyslogMessageHandler extends SimpleChannelInboundHandler<Object> {
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<SyslogIngressEvent> 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<Object> {
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<Object> {
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();
haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServer.java
+7 -2
View File
@@ -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();
haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerBoth.java
+9 -3
View File
@@ -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());
haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogServerLifecycle.java
+5 -2
View File
@@ -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) {
haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogTcpChannelInitializer.java
+9 -2
View File
@@ -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<SocketChanne
private final int maxFrameLength;
private final Charset syslogCharset;
private final SyslogWalService walService;
private final DeviceContextResolver deviceResolver;
private final boolean dropInvalidBeforeWal;
SyslogTcpChannelInitializer(int maxFrameLength, Charset syslogCharset, SyslogWalService walService) {
SyslogTcpChannelInitializer(int maxFrameLength, Charset syslogCharset, SyslogWalService walService,
DeviceContextResolver deviceResolver, boolean dropInvalidBeforeWal) {
if (maxFrameLength <= 0) throw new IllegalArgumentException("maxFrameLength must be positive");
this.maxFrameLength = maxFrameLength;
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
@@ -40,6 +46,7 @@ public class SyslogTcpChannelInitializer extends ChannelInitializer<SocketChanne
pipeline.addLast("encoder", new StringEncoder(syslogCharset));
// 添加业务处理器
pipeline.addLast("handler", new SyslogMessageHandler(syslogCharset, walService));
pipeline.addLast("handler", new SyslogMessageHandler(
syslogCharset, walService, deviceResolver, dropInvalidBeforeWal));
}
}
haobang-security-dm/syslog-serve/src/main/java/com/netty/SyslogUdpChannelInitializer.java
+9 -2
View File
@@ -1,5 +1,6 @@
package com.netty;
import com.haobang.dispatch.DeviceContextResolver;
import com.haobang.wal.SyslogWalService;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
@@ -14,10 +15,15 @@ import java.nio.charset.Charset;
public class SyslogUdpChannelInitializer extends ChannelInitializer<DatagramChannel> {
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<DatagramChan
pipeline.addLast("encoder", new StringEncoder(syslogCharset));
// 添加业务处理器
pipeline.addLast("handler", new SyslogMessageHandler(syslogCharset, walService));
pipeline.addLast("handler", new SyslogMessageHandler(
syslogCharset, walService, deviceResolver, dropInvalidBeforeWal));
}
}
haobang-security-dm/syslog-serve/src/main/resources/application-prod.properties
+18
View File
@@ -14,6 +14,24 @@ syslog.buffer.size=10000
# syslog 消息字符编码，国内安全设备普遍使用 GBK，如需 UTF-8 改为 syslog.charset=UTF-8
syslog.charset=UTF-8
syslog.sm4.generateKey=f79548ab6fa8a304fc0115e17230358a
# 单探针约 200KiB/条的生产容量基线。2GiB 字节预算在 70% 高水位时
# 约 7,300 条、1.4GiB 触发背压；实际可持续吞吐仍必须以现场 NVMe 压测为准。
syslog.wal.directory=${SYSLOG_WAL_DIRECTORY:/app/logs/syslog-wal}
syslog.wal.queue-capacity=${SYSLOG_WAL_QUEUE_CAPACITY:20000}
syslog.wal.max-pending-bytes=${SYSLOG_WAL_MAX_PENDING_BYTES:2147483648}
syslog.wal.high-watermark-percent=${SYSLOG_WAL_HIGH_WATERMARK_PERCENT:70}
syslog.wal.low-watermark-percent=${SYSLOG_WAL_LOW_WATERMARK_PERCENT:40}
syslog.wal.flush-records=${SYSLOG_WAL_FLUSH_RECORDS:200}
syslog.wal.flush-interval-ms=${SYSLOG_WAL_FLUSH_INTERVAL_MS:100}
syslog.wal.segment-max-bytes=${SYSLOG_WAL_SEGMENT_MAX_BYTES:2147483648}
syslog.dispatch.worker-threads=${SYSLOG_DISPATCH_WORKER_THREADS:8}
syslog.dispatch.max-inflight=${SYSLOG_DISPATCH_MAX_INFLIGHT:5000}
syslog.dispatch.max-inflight-bytes=${SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES:1073741824}
# 默认在 WAL 前丢弃无效设备正文，仅按缓存周期打印来源 IP；需要 WAL/Terminal 审计时显式开启。
syslog.dispatch.terminal-record-enabled=${SYSLOG_DISPATCH_TERMINAL_RECORD_ENABLED:false}
spring.kafka.producer.properties.buffer.memory=${SPRING_KAFKA_PRODUCER_PROPERTIES_BUFFER_MEMORY:536870912}
# APP Service Configuration
app.service.device_id=1
app.service.device_name=honeypot
haobang-security-dm/syslog-serve/src/main/resources/application.properties
+2
View File
@@ -61,6 +61,8 @@ syslog.dispatch.max-inflight=${SYSLOG_DISPATCH_MAX_INFLIGHT:5000}
syslog.dispatch.max-inflight-bytes=${SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES:268435456}
syslog.dispatch.retry-backoff-ms=${SYSLOG_DISPATCH_RETRY_BACKOFF_MS:1000}
syslog.dispatch.shutdown-timeout-ms=${SYSLOG_DISPATCH_SHUTDOWN_TIMEOUT_MS:30000}
# 性能优先默认 false：Syslog 无效设备正文在 WAL 前丢弃，仅按缓存周期打印 IP；true 时写 WAL 和 Terminal 审计。
syslog.dispatch.terminal-record-enabled=${SYSLOG_DISPATCH_TERMINAL_RECORD_ENABLED:false}
syslog.device-context.cache-ttl-ms=${SYSLOG_DEVICE_CONTEXT_CACHE_TTL_MS:300000}
syslog.status-log-interval-ms=${SYSLOG_STATUS_LOG_INTERVAL_MS:60000}
haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/DeviceContextResolverTest.java
+28
View File
@@ -55,4 +55,32 @@ class DeviceContextResolverTest {
assertEquals(DeviceContextResolver.Type.NOT_ASSIGNED, result.getType());
}
@Test
void ingressResolutionUsesPreloadedSnapshotWithoutPerMessageDatabaseLookup() {
DeviceDeviceService service = mock(DeviceDeviceService.class);
DeviceDevice device = new DeviceDevice();
device.setId(126);
device.setIp("192.168.4.19");
device.setDeviceCollectId(1);
when(service.getActiveDevices()).thenReturn(Collections.singletonList(device));
AppProperties properties = new AppProperties();
properties.setAppServiceDeviceCollectId(1);
properties.setSyslogDispatchTerminalRecordEnabled(false);
properties.setSyslogDeviceContextCacheTtlMs(60000L);
DeviceContextResolver resolver = new DeviceContextResolver(
service, mock(DeviceUnknownService.class), properties);
resolver.refreshIngressSnapshot();
assertEquals(DeviceContextResolver.IngressStatus.REGISTERED,
resolver.resolveIngress("192.168.4.19", "TCP"));
assertEquals(DeviceContextResolver.IngressStatus.INVALID,
resolver.resolveIngress("192.168.4.20", "TCP"));
assertEquals(DeviceContextResolver.IngressStatus.INVALID,
resolver.resolveIngress("192.168.4.20", "TCP"));
verify(service, times(1)).getActiveDevices();
assertTrue(resolver.isIngressSnapshotReady());
assertEquals(1, resolver.getIngressSnapshotSize());
}
}
haobang-security-dm/syslog-serve/src/test/java/com/haobang/dispatch/SyslogWalDispatcherTest.java
+28
View File
@@ -18,6 +18,7 @@ import java.time.LocalDateTime;
import java.time.Instant;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -103,6 +104,7 @@ class SyslogWalDispatcherTest {
@Test
void terminalDeviceStateIsPersistedBeforeCheckpointAdvances() throws Exception {
AppProperties properties = properties(tempDirectory);
properties.setSyslogDispatchTerminalRecordEnabled(true);
wal = new SyslogWalService(properties, new ObjectMapper());
wal.initialize();
wal.tryAppend(event("20260811-event-1", "first"));
@@ -121,6 +123,32 @@ class SyslogWalDispatcherTest {
verify(producer, never()).messagePush(any(), anyString(), anyString(), anyString(), any());
}
@Test
void terminalDeviceStateIsDroppedWithoutFileWhenRecordingDisabled() throws Exception {
AppProperties properties = properties(tempDirectory);
properties.setSyslogDispatchTerminalRecordEnabled(false);
wal = new SyslogWalService(properties, new ObjectMapper());
wal.initialize();
wal.tryAppend(event("20260811-event-drop", "invalid-device"));
waitUntil(() -> 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<Path> 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);
haobang-security-dm/syslog-serve/src/test/java/com/netty/SyslogMessageHandlerBackpressureTest.java
+40
View File
@@ -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();
}
}