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

新增syslog-collector-api 采集模块，通过api 方式采集天眼、微步的设备日志。

Browse Source
This commit is contained in:
nanChen
2026-09-07 15:34:17 +08:00
parent a3852fc0f9
commit 037cca32dd
53 changed files with 6358 additions and 488 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
+1 -2
View File
@@ -28,8 +28,7 @@ services:
- SPRING_DATASOURCE_USERNAME=SYSDBA
- SPRING_DATASOURCE_PASSWORD=caZ2TcmXNSW8L2Ap
- SPRING_DATASOURCE_DRIVER_CLASS_NAME=dm.jdbc.driver.DmDriver
- SPRING_DATASOURCE_HIKARI_SCHEMA=\"PUBLIC\"
- SPRING_DATASOURCE_HIKARI_SCHEMA=PUBLIC
# Redis配置
- SPRING_REDIS_HOST=192.168.4.99
haobang-security-dm/syslog-serve/src/main/java/com/Modules/Device/DeviceProcess.java
+1 -1
View File
@@ -107,7 +107,7 @@ public class DeviceProcess {
if(deviceList.size()>0)
{
deviceUnknownService.updateLastTime( deviceList.get(0).getId(),currentTime );
System.out.println("更新未知设备的最后发现时间,IP:"+this.sourceIP);
logger.info("更新未知设备的最后发现时间,IP:"+this.sourceIP);
}
else {
//创建未知设备记录
haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java
+192 -102
View File
@@ -5,6 +5,15 @@ import com.haobang.util.Sm4Util;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
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.format.DateTimeFormatter;
import java.util.LinkedHashMap;
import java.util.Properties;
@@ -20,157 +29,238 @@ public class kafkaProducer {
private static final Logger logger = LoggerFactory.getLogger(kafkaProducer.class);
// KafkaProducer 全局单例，线程安全，避免每次 send 重建连接
private static volatile Producer<String, String> producer;
// 缓存的 topic 名称（初始化时从配置读取一次，避免每次 send 都查 Spring）
private static volatile String cachedTopic;
// 缓存的 SM4 密钥
private static volatile String cachedSm4Key;
private static final Object lock = new Object();
// ==================== 兜底备份配置 ====================
/** 兜底文件存储根目录 */
private static final String FALLBACK_DIR = "/app/syslog-fallback/";
/** 兜底文件日期格式 */
private static final DateTimeFormatter FALLBACK_DATE_FORMAT = DateTimeFormatter.ofPattern("yyyyMMdd");
/** 兜底写入最大重试次数 */
private static final int FALLBACK_MAX_RETRIES = 3;
/** 兜底写入重试间隔（毫秒） */
private static final long FALLBACK_RETRY_INTERVAL_MS = 100;
static {
// 注册 JVM 关闭钩子，确保 Producer 正常 flush + close
Runtime.getRuntime().addShutdownHook(new Thread(() -> closeProducer()));
}
private static AppProperties getProps() {
return SpringContextUtil.getBean(AppProperties.class);
}
/**
* 获取全局 KafkaProducer 单例（双重检查锁定）
*/
private static Producer<String, String> getProducer() {
if (producer == null) {
synchronized (lock) {
if (producer == null) {
AppProperties appProps = getProps();
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, appProps.getKafkaProducerBootstrap());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// 生产者缓冲区大小，减少网络 IO 次数
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
producer = new KafkaProducer<>(props);
cachedTopic = appProps.getKafkaProducerTopic();
cachedSm4Key = appProps.getSm4Key();
logger.info("KafkaProducer 单例初始化完成, topic={}, brokers={}",
cachedTopic, appProps.getKafkaProducerBootstrap());
}
}
}
return producer;
}
/**
* 优雅关闭 KafkaProducer
*/
public static void closeProducer() {
if (producer != null) {
synchronized (lock) {
if (producer != null) {
try {
logger.info("正在关闭 KafkaProducer...");
producer.flush();
producer.close();
logger.info("KafkaProducer 已关闭");
} catch (Exception e) {
logger.error("关闭 KafkaProducer 异常", e);
}
producer = null;
}
}
}
}
public static void main(String[] args) {
AppProperties appProps = getProps();
// 配置生产者属性
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, appProps.getKafkaProducerBootstrap());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 可选配置：提高可靠性
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// 创建生产者实例
Producer<String, String> producer = new KafkaProducer<>(props);
try {
// 发送10条消息
//for (int i = 0; i < 1; i++) {
String key = "key-" + "1";
String value = "message-" + "1" + " at " + System.currentTimeMillis();
// 创建生产者记录
ProducerRecord<String, String> record =
new ProducerRecord<>(appProps.getKafkaProducerTopic(), key, value);
// 发送消息（异步方式）
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception == null) {
System.out.println("消息发送成功: " +
"主题=" + metadata.topic() +
", 分区=" + metadata.partition() +
", 偏移量=" + metadata.offset());
} else {
System.err.println("消息发送失败: " + exception.getMessage());
}
try (Producer<String, String> producer = new KafkaProducer<>(props)) {
String key = "key-" + "1";
String value = "message-" + "1" + " at " + System.currentTimeMillis();
ProducerRecord<String, String> record =
new ProducerRecord<>(appProps.getKafkaProducerTopic(), key, value);
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception == null) {
System.out.println("消息发送成功: " +
"主题=" + metadata.topic() +
", 分区=" + metadata.partition() +
", 偏移量=" + metadata.offset());
} else {
System.err.println("消息发送失败: " + exception.getMessage());
}
});
// 同步发送方式（如果需要）
// RecordMetadata metadata = producer.send(record).get();
// System.out.println("同步发送成功: " + metadata.offset());
// }
}
});
producer.flush();
} catch (Exception e) {
e.printStackTrace();
} finally {
// 关闭生产者
producer.close();
}
}
/**
* * syslog 文本消息
* * @param strSyslog
* syslog 文本消息推送到 Kafka（复用全局单例 Producer）
* @param deviceDevice 设备信息
* @param strSyslog 原始 syslog 消息
* @param strReceiveTime 接收时间
*/
public static void messagePush(DeviceDevice deviceDevice, String strSyslog,String strReceiveTime)
{
AppProperties appProps = getProps();
// 配置生产者属性
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, appProps.getKafkaProducerBootstrap());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 可选配置：提高可靠性
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// 创建生产者实例
Producer<String, String> producer = new KafkaProducer<>(props);
public static void messagePush(DeviceDevice deviceDevice, String strSyslog, String strReceiveTime) {
Producer<String, String> p = getProducer();
try {
String key = "key-" + System.currentTimeMillis();
String value = DeviceInfoUtil.getFullLogString(deviceDevice, strSyslog, strReceiveTime);
String key = "key-" + System.currentTimeMillis();
//String value = DeviceInfoUtil.getFullLogString(strSyslog);
//采用动态获取Syslog 请求的设备信息
String value = DeviceInfoUtil.getFullLogString(deviceDevice,strSyslog,strReceiveTime);
// SM4 加密
String sm4Message = Sm4Util.encryptCbc(value, cachedSm4Key);
//Message进行SM4加密写入kafka
String Sm4message= Sm4Util.encryptCbc(value, appProps.getSm4Key());
//System.out.println("Sm4message:"+Sm4message);
ProducerRecord<String, String> record =
new ProducerRecord<>(cachedTopic, key, sm4Message);
// 创建生产者记录
ProducerRecord<String, String> record =
new ProducerRecord<>(appProps.getKafkaProducerTopic(), key, Sm4message);
// 发送消息（异步方式）
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception == null) {
System.out.println("消息发送成功: " +
"主题=" + metadata.topic() +
", 分区=" + metadata.partition() +
", 偏移量=" + metadata.offset());
} else {
System.err.println("消息发送失败: " + exception.getMessage());
logger.error("消息发送失败: " + exception.getMessage());
}
// 异步发送 + 失败回调兜底：闭包捕获 value（原始明文），失败时写入本地文件
p.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception == null) {
logger.debug("消息发送成功: 主题={}, 分区={}, 偏移量={}",
metadata.topic(), metadata.partition(), metadata.offset());
} else {
logger.error("消息发送失败, 开始兜底写入本地文件: topic={}, 异常={}",
cachedTopic, exception.getMessage());
fallbackToLocal(value);
}
});
// 同步发送方式（如果需要）
// RecordMetadata metadata = producer.send(record).get();
// System.out.println("同步发送成功: " + metadata.offset());
}
});
} catch (Exception e) {
e.printStackTrace();
} finally {
// 关闭生产者
producer.close();
// 加密/组包阶段异常，同样兜底
logger.error("messagePush 预处理异常, 开始兜底写入本地文件", e);
try {
String value = DeviceInfoUtil.getFullLogString(deviceDevice, strSyslog, strReceiveTime);
fallbackToLocal(value);
} catch (Exception ex) {
logger.error("兜底写入时组包也失败, 消息已丢失", ex);
}
}
// 注意：不再 close，Producer 是全局单例，由 shutdown hook 关闭
}
/**
* 生成json字符串，补充设备配置信息
* @param strSyslog
* @return
*/
private static String getLogJsonString(String strSyslog)
{
private static String getLogJsonString(String strSyslog) {
ObjectMapper objectMapper = new ObjectMapper();
// 创建设备配置信息MAP
Map<String, Object> map = new LinkedHashMap<>();
AppProperties appProps = getProps();
map.put("device_id", appProps.getAppServiceDeviceId());
map.put("device_name", appProps.getAppServiceDeviceName());
map.put("vendor", appProps.getAppServiceVendor());
map.put("data_type", appProps.getAppServiceDataType());
// 添加syslogMessage
map.put("syslogMessage", strSyslog);
try {
//生成json消息字符串
String json = objectMapper.writeValueAsString(map);
return json;
}
catch(Exception ex)
{
logger.error("kafkaProducer getLogJsonString(）生成json字符串 异常: syslog: "+ strSyslog);
logger.error("Exception: : "+ ex.getMessage());
return objectMapper.writeValueAsString(map);
} catch (Exception ex) {
logger.error("kafkaProducer getLogJsonString() 生成json字符串异常, syslog: {}", strSyslog);
logger.error("Exception: {}", ex.getMessage());
}
return null;
}
// ==================== 兜底备份 ====================
/**
* 将原始消息写入本地兜底文件（按日期分文件，追加写入）
* <p>
* 文件路径：/data/syslog-fallback/kafka-failed-yyyyMMdd.log
* </p>
* @param message 原始明文消息（完整日志字符串）
*/
private static void fallbackToLocal(String message) {
if (message == null || message.isEmpty()) {
logger.warn("兜底消息为空，跳过写入");
return;
}
for (int i = 0; i < FALLBACK_MAX_RETRIES; i++) {
try {
String dateStr = LocalDate.now().format(FALLBACK_DATE_FORMAT);
String fileName = "kafka-failed-" + dateStr + ".log";
Path filePath = Paths.get(FALLBACK_DIR, fileName);
// 确保目录存在
File dir = new File(FALLBACK_DIR);
if (!dir.exists()) {
boolean created = dir.mkdirs();
if (!created) {
logger.error("创建兜底目录失败: {}", FALLBACK_DIR);
return;
}
}
// 追加写入，每条一行
String line = message + "\n";
Files.write(filePath, line.getBytes(StandardCharsets.UTF_8),
StandardOpenOption.CREATE, StandardOpenOption.APPEND);
logger.debug("消息已兜底写入文件: {}", fileName);
return;
} catch (IOException e) {
logger.error("兜底文件写入失败(第{}次重试): {}", i + 1, e.getMessage());
if (i < FALLBACK_MAX_RETRIES - 1) {
try {
Thread.sleep(FALLBACK_RETRY_INTERVAL_MS);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
break;
}
}
}
}
logger.error("兜底写入彻底失败，消息丢失！消息长度={}", message.length());
}
}
haobang-security-dm/syslog-serve/src/main/resources/application-prod.properties
+3 -3
View File
@@ -9,8 +9,8 @@ server.error.include-binding-errors=always
# Syslog Server Configuration
syslog.tcp.port=514
syslog.udp.port=514
syslog.max.frame.length=2621440
syslog.buffer.size=10000
syslog.max.frame.length=262144
syslog.buffer.size=1000
# syslog 消息字符编码，国内安全设备普遍使用 GBK，如需 UTF-8 改为 syslog.charset=UTF-8
syslog.charset=UTF-8
syslog.sm4.generateKey=f79548ab6fa8a304fc0115e17230358a
@@ -23,7 +23,7 @@ app.service.product_name=diting
app.service.data_type=json
#采集探针ID
app.service.device_collect_id=${DEVICE_COLLECT_ID:1}
app.service.version=${APP_SERVICE_VER:V1.0.0-20260527}
app.service.version=${APP_SERVICE_VER:V1.0.0-20260811}
app.service.device_collect_name=${DEVICE_COLLECT_NAME:DevCollect-01}
# kafka Configuration
haobang-security-dm/syslog-serve/src/main/resources/logback.xml
+4 -4
View File
@@ -16,10 +16,10 @@
<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>logs/syslog-serve-dm.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 每天滚动 -->
<fileNamePattern>logs/syslog-serve-dm.%d{yyyy-MM-dd}.log</fileNamePattern>
<!-- 保留的日志文件的最大天数 -->
<maxHistory>1</maxHistory>
<!-- 每小时滚动 -->
<fileNamePattern>logs/syslog-serve-dm.%d{yyyy-MM-dd-HH}.log</fileNamePattern>
<!-- 保留的日志文件的最大小时数 -->
<maxHistory>24</maxHistory>
<!-- 所有归档日志文件的总大小上限 -->
<totalSizeCap>15GB</totalSizeCap>
<!-- 启动时是否清理过期日志 -->