1、适配部署环境变量自动注入

2、适配南网达梦数据库安全版SQL语句Group by 不支持自定义函数
3、达梦数据库驱动改为达梦提供的指定版本
4、修改syslog-consumer模块yaml 文件kafka 变量名称
5、修改访问日志告警表新增4个字段,完善异常行为详情页面显示字段
This commit is contained in:
2026-07-22 18:50:44 +08:00
parent 5fe0e124bb
commit 16b73dffee
89 changed files with 4507 additions and 2186 deletions
@@ -1,462 +1,361 @@
package com.Modules.NormalData;
import cn.hutool.core.date.DateTime;
import com.common.entity.XdrHoneypot;
import com.common.mapper.XdrHoneypotMapper;
import com.common.service.SyslogNonNormalMessageService;
import com.common.util.*;
import com.config.AppConfig;
import org.apache.ibatis.session.SqlSession;
import org.springframework.beans.factory.annotation.Autowired;
import com.common.entity.SyslogMessage;
import com.influx.SyslogToInfluxApp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.common.service.DmNormalizeRuleService;
import com.common.service.DmColumnService;
import com.common.service.SyslogNormalDataService;
import com.common.entity.DmColumn;
import com.common.service.impl.DmColumnServiceImpl;
import com.common.mapper.DmColumnMapper;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.RestController;
import com.common.mapper.DmColumnMapper;
import com.common.mapper.DmNormalizeRuleMapper;
import java.sql.Timestamp;
import java.util.*;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.json.JSONObject;
import com.common.entity.RuleContent.*;
import com.common.entity.SyslogNonNormalMessage;
import org.joda.time.LocalDateTime;
import com.common.service.LogDataFilterService;
import com.common.service.LogDataCompleteService;
import com.common.service.DeviceCollectTaskService;
import com.common.entity.DeviceCollectTask;
import com.common.entity.DeviceDevice;
import com.common.util.TimeConversionUtils;
import com.common.service.DeviceDeviceService;
import com.common.entity.RuleContent.*;
import com.common.entity.SyslogNonNormalMessage;
import com.common.mapper.DmColumnMapper;
import com.common.mapper.DmNormalizeRuleMapper;
import com.common.service.*;
import com.common.util.*;
import com.config.AppProperties;
import com.config.KafkaConsumerProperties;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.joda.time.LocalDateTime;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.sql.Timestamp;
import java.util.*;
import java.util.stream.Collectors;
import com.common.entity.SyslogMessage;
/**
* 日志标准化处理器
* 线程安全版本:运行时状态封装为 ProcessContext,每次 process() 调用独立(解决多线程 NPE 问题)
*/
@Service
public class LogNormalProcessor {
private static final Logger logger = LoggerFactory.getLogger(LogNormalProcessor.class);
private String strLogMsg ;
private String strKafkaMessage;
private String strDeviceInfo;
private String strDataType ="json" ;
private Map<String, Object> messageMap ;
private Map<String, String> deviceInfoMap ;
private String strSyslogUUID ;
private String strSyslogTopic ;
private boolean isSaveNonNormal =false ;
// ==================== 依赖注入(Spring Bean,单例,线程安全)====================
private final DmNormalizeRuleService dmNormalizeRuleService;
private final DmColumnService dmColumnService;
private final SyslogNormalDataService syslogNormalDataService;
private final SyslogNonNormalMessageService messageService;
private final LogDataFilterService logDataFilterService;
private final LogDataCompleteService logDataCompleteService;
private final DeviceCollectTaskService deviceCollectTaskService;
private final DeviceDeviceService deviceDeviceService;
private final DmColumnMapper dmColumnMapper;
private final DmNormalizeRuleMapper dmNormalizeRuleMapper;
private final KafkaConsumerProperties kafkaConsumerProperties;
private final AppProperties appProperties;
@Autowired
public DmNormalizeRuleService dmNormalizeRuleService ;
@Autowired
public DmColumnService dmColumnService ;
@Autowired
public SyslogNormalDataService syslogNormalDataService =SpringContextUtil.getBean(SyslogNormalDataService.class);
@Autowired
private SyslogNonNormalMessageService messageService =SpringContextUtil.getBean(SyslogNonNormalMessageService.class);
@Autowired
private LogDataFilterService logDataFilterService= SpringContextUtil.getBean(LogDataFilterService.class);
@Autowired
private LogDataCompleteService logDataCompleteService= SpringContextUtil.getBean(LogDataCompleteService.class);
@Autowired
private DeviceCollectTaskService deviceCollectTaskService= SpringContextUtil.getBean(DeviceCollectTaskService.class);
@Autowired
private DeviceDeviceService deviceDeviceService= SpringContextUtil.getBean(DeviceDeviceService.class);
@Autowired
SyslogNonNormalMessage syslogNonNormalMessage=new SyslogNonNormalMessage();
@Autowired
DmColumnMapper dmColumnMapper;
@Autowired
DmNormalizeRuleMapper dmNormalizeRuleMapper;
private List<Map<String, Object>> dmNormalizeRuleList;
private List<Map<String, Object>> dmColumnList;
private LinkedHashMap<String, Object> OrginalColumnMap ;
public LogNormalProcessor( String LogMsg, String syslogUUID,String syslogTopic) {
/** 标准化数据处理步骤 */
//初始化 (获取syslog_normal_data 表全部字段的属性、设备ID对应的规则设置、转化成JSON)
//判断数据解析类型(json、键值、xml、正则表达式、分割符)解析syslogMessage 字段转成 HashMAP
//匹配规则内容,获取字段映射关系配置,抽取命中的字段名称、数值
//判断字段内容及类型,根据数据类型属性,进行内容转换。
//生成insert SQL语句、执行入库操作。
strKafkaMessage=LogMsg;
strSyslogUUID=syslogUUID;
strSyslogTopic=syslogTopic;
if(!LogMsg.isEmpty()) {
strLogMsg = SyslogParser.substringAfterFirstCloseBracket(LogMsg);
strDeviceInfo=SyslogParser.substringBeforeFirstChar(LogMsg,']');
}
else{
LogMsg="[receive_time=20251118165909470 device_id=1 device_name=honeypot vendor=changting data_type=json device_collect_id=1]<14>1 2025-09-24T11:52:26Z 5f46d3be75e1 supermario 128 honeypot_event - {\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\",\"start_time\":\"2023-09-03T11:07:02.50167643Z\",\"time\":\"2023-09-03T11:16:18.883885281Z\",\"risk_level\":4,\"connection\":\"b18f3fbe-3fbf-4495-815f-ff26f6fb0bdf\",\"file_info\":null,\"extra\":{\"payload\":{\"format\":\"line\",\"name\":{\"cn\":\"攻击载荷\",\"en\":\"payload\"},\"value\":\"\"},\"uid\":{\"format\":\"line\",\"name\":{\"cn\":\"\",\"en\":\"\"},\"uid\":\"b4cbc73c-25d0-4429-ae1b-a856cdf1a651\",\"value\":\"\"}},\"type\":\"WEB_ATTACK_SCANNER\",\"agent_sn\":\"caa7da42-0cca-4cb1-b501-1f1eb2b588d5\",\"agent_name\":\" 教育局蜜罐探针\",\"honeypot_id\":\"11a9ac6bdf38ae2aaa49ec4f1b4a921bff71952cb9f175bdd8ee1f0497057bc6\",\"honeypot_name\":\"茂名市中小学管理平台管理后台\",\"src_ip\":\"117.50.189.7\",\"src_port\":58512,\"src_mac\":\"\",\"dest_ip\":\"192.168.222.2\",\"dest_port\":9200,\"proxy_ip\":null,\"node\":\"WRx3\"}";
strLogMsg="<14>1 2025-09-24T11:52:26Z 5f46d3be75e1 supermario 128 honeypot_event - {\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\",\"start_time\":\"2023-09-03T11:07:02.50167643Z\",\"time\":\"2023-09-03T11:16:18.883885281Z\",\"risk_level\":4,\"connection\":\"b18f3fbe-3fbf-4495-815f-ff26f6fb0bdf\",\"file_info\":null,\"extra\":{\"payload\":{\"format\":\"line\",\"name\":{\"cn\":\"攻击载荷\",\"en\":\"payload\"},\"value\":\"\"},\"uid\":{\"format\":\"line\",\"name\":{\"cn\":\"\",\"en\":\"\"},\"uid\":\"b4cbc73c-25d0-4429-ae1b-a856cdf1a651\",\"value\":\"\"}},\"type\":\"WEB_ATTACK_SCANNER\",\"agent_sn\":\"caa7da42-0cca-4cb1-b501-1f1eb2b588d5\",\"agent_name\":\" 教育局蜜罐探针\",\"honeypot_id\":\"11a9ac6bdf38ae2aaa49ec4f1b4a921bff71952cb9f175bdd8ee1f0497057bc6\",\"honeypot_name\":\"茂名市中小学管理平台管理后台\",\"src_ip\":\"117.50.189.7\",\"src_port\":58512,\"src_mac\":\"\",\"dest_ip\":\"192.168.222.2\",\"dest_port\":9200,\"proxy_ip\":null,\"node\":\"WRx3\"}";
strDeviceInfo=SyslogParser.substringBeforeFirstChar(LogMsg,']');
}
}
/**
* 初始化数据init()
* 请求级处理上下文 —— 每个 process() 调用创建新实例,彻底解决多线程竞态条件
*/
public void init()
{
Map<String,String> mapdev =SyslogParser.parseKeyValuePairs(strDeviceInfo);
deviceInfoMap=mapdev;
//获取日志对应的设备ID
long deviceID=Long.parseLong(mapdev.get("device_id"));
System.out.println("device_id:"+deviceID );
strDataType=mapdev.get("data_type");
//getDeviceID(strLogMsg);
try{
// 通过工具类获取Service实例
dmColumnService= SpringContextUtil.getBean(DmColumnService.class);
dmNormalizeRuleService= SpringContextUtil.getBean(DmNormalizeRuleService.class);
if(deviceID>0) {
dmNormalizeRuleList = dmNormalizeRuleService.selectByDeviceIdAuto(deviceID);
private static class ProcessContext {
String strLogMsg;
String strKafkaMessage;
String strDeviceInfo;
String strDataType = "json";
Map<String, Object> messageMap;
Map<String, String> deviceInfoMap;
String strSyslogUUID;
String strSyslogTopic;
boolean isSaveNonNormal = false;
SyslogNonNormalMessage syslogNonNormalMessage = new SyslogNonNormalMessage();
List<Map<String, Object>> dmNormalizeRuleList;
List<Map<String, Object>> dmColumnList;
LinkedHashMap<String, Object> OrginalColumnMap;
}
// ==================== 构造器注入(Spring 自动完成)====================
@Autowired
public LogNormalProcessor(
DmNormalizeRuleService dmNormalizeRuleService,
DmColumnService dmColumnService,
SyslogNormalDataService syslogNormalDataService,
SyslogNonNormalMessageService messageService,
LogDataFilterService logDataFilterService,
LogDataCompleteService logDataCompleteService,
DeviceCollectTaskService deviceCollectTaskService,
DeviceDeviceService deviceDeviceService,
DmColumnMapper dmColumnMapper,
DmNormalizeRuleMapper dmNormalizeRuleMapper,
KafkaConsumerProperties kafkaConsumerProperties,
AppProperties appProperties
) {
this.dmNormalizeRuleService = dmNormalizeRuleService;
this.dmColumnService = dmColumnService;
this.syslogNormalDataService = syslogNormalDataService;
this.messageService = messageService;
this.logDataFilterService = logDataFilterService;
this.logDataCompleteService = logDataCompleteService;
this.deviceCollectTaskService = deviceCollectTaskService;
this.deviceDeviceService = deviceDeviceService;
this.dmColumnMapper = dmColumnMapper;
this.dmNormalizeRuleMapper = dmNormalizeRuleMapper;
this.kafkaConsumerProperties = kafkaConsumerProperties;
this.appProperties = appProperties;
}
/**
* 处理一条日志消息(线程安全:每次调用创建独立的 ProcessContext
*
* @param logMsg 完整日志消息(含头部)
* @param syslogUUID 消息UUID
* @param syslogTopic 消息Topic(可传 null,自动使用配置值)
*/
public void process(String logMsg, String syslogUUID, String syslogTopic) {
ProcessContext ctx = new ProcessContext();
ctx.strSyslogUUID = syslogUUID;
ctx.strSyslogTopic = (syslogTopic != null && !syslogTopic.isEmpty())
? syslogTopic
: kafkaConsumerProperties.getTopic();
ctx.isSaveNonNormal = false;
ctx.syslogNonNormalMessage = new SyslogNonNormalMessage();
// 解析消息
ctx.strKafkaMessage = logMsg;
if (!logMsg.isEmpty()) {
ctx.strLogMsg = SyslogParser.substringAfterFirstCloseBracket(logMsg);
ctx.strDeviceInfo = SyslogParser.substringBeforeFirstChar(logMsg, ']');
} else {
// 测试用例硬编码数据
logMsg = "[receive_time=20251118165909470 device_id=1 device_name=honeypot vendor=changting data_type=json device_collect_id=1]<14>1 2025-09-24T11:52:26Z 5f46d3be75e1 supermario 128 honeypot_event - \uFEFF{\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\"}";
ctx.strLogMsg = "<14>1 2025-09-24T11:52:26Z 5f46d3be75e1 supermario 128 honeypot_event - \uFEFF{\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\"}";
ctx.strDeviceInfo = SyslogParser.substringBeforeFirstChar(logMsg, ']');
}
// 执行处理
init(ctx);
}
/**
* 初始化数据处理(所有运行时状态通过 ctx 传递,线程安全)
*/
private void init(ProcessContext ctx) {
Map<String, String> mapdev = SyslogParser.parseKeyValuePairs(ctx.strDeviceInfo);
ctx.deviceInfoMap = mapdev;
long deviceID = Long.parseLong(mapdev.get("device_id"));
System.out.println("device_id:" + deviceID);
ctx.strDataType = mapdev.get("data_type");
try {
if (deviceID > 0) {
ctx.dmNormalizeRuleList = dmNormalizeRuleService.selectByDeviceIdAuto(deviceID);
}
//dmColumnList=dmColumnService.selectAllNormal();
//System.out.println("dmColumnList size:"+ dmColumnList.size());
//OrginalColumnMap=getMessageToMap(strLogMsg);
//解析SyslogMessage
//SyslogMessage logMsg = SyslogParser.parse(strLogMsg);
//设备对应的规则normal rule Map 进行标准化数据处理
for (int i = 0; i < dmNormalizeRuleList.size(); i++) {
for (int i = 0; i < ctx.dmNormalizeRuleList.size(); i++) {
try {
Map<String, Object> dmNormalizeRule = dmNormalizeRuleList.get(i);
Map<String, Object> dmNormalizeRule = ctx.dmNormalizeRuleList.get(i);
String data_type = dmNormalizeRule.get("data_type").toString();
//数据类型及格式不符,则break;
System.out.println("索引: " + i + ", 值: " + dmNormalizeRule);
System.out.println("normalrule ID: " + dmNormalizeRule.get("id") + ", display_name:" + dmNormalizeRule.get("display_name") + " data_type:" + dmNormalizeRule.get("data_type"));
//数据类型不匹配,则跳过规则
//if (!data_type.equals(strDataType)) break;
System.out.println("normalrule ID: " + dmNormalizeRule.get("id")
+ ", display_name:" + dmNormalizeRule.get("display_name")
+ " data_type:" + dmNormalizeRule.get("data_type"));
//获取syslog message 文本解析配置项
OrginalColumnMap = getMessageToMap(dmNormalizeRule, strLogMsg, data_type);
if ((OrginalColumnMap == null) || (OrginalColumnMap.size()==0)) {
ctx.OrginalColumnMap = getMessageToMap(dmNormalizeRule, ctx.strLogMsg, data_type);
if ((ctx.OrginalColumnMap == null) || (ctx.OrginalColumnMap.size() == 0)) {
logger.error("OrginalColumnMap 对象获取为空");
//保存非标日志信息
syslogNonNormalMessage.setReason("Log解析异常");
syslogNonNormalMessage.setReasonDetail("log解析异常,返回规则名称:"+dmNormalizeRule.get("display_name") +",OrginalColumnMap 对象获取为空");
if(isSaveNonNormal==false)
SaveNonNormalMessage(deviceID, DateTime.now());
ctx.syslogNonNormalMessage.setReason("Log解析异常");
ctx.syslogNonNormalMessage.setReasonDetail("log解析异常,返回规则名称:"
+ dmNormalizeRule.get("display_name") + ",OrginalColumnMap 对象获取为空");
if (ctx.isSaveNonNormal == false)
SaveNonNormalMessage(ctx, deviceID, DateTime.now());
continue;
}
List<HashMap<String, Object>> destColumnList = getRuleContentMappers(dmNormalizeRule);
//解析字段匹配已命中的配置规则字段
List<HashMap<String, Object>> ruleColumnList = getNormalColumnList(OrginalColumnMap, destColumnList);
//System.out.println("ruleColumnList :"+ ruleColumnList);
List<HashMap<String, Object>> ruleColumnList = getNormalColumnList(ctx.OrginalColumnMap, destColumnList);
Map<String, Object> destMap = getColumnMap(ruleColumnList);
//List<HashMap<String ,Object>> destMap2= getCompleteColumnsList(dmNormalizeRule);
//数据处理-过滤规则
if (logDataFilterService.evaluateFilterRule(dmNormalizeRule.get("rule_content").toString(), destMap))
// 数据处理-过滤规则
if (logDataFilterService.evaluateFilterRule(
dmNormalizeRule.get("rule_content").toString(), destMap))
continue;
//数据处理-补全规则
logDataCompleteService.processDataCompletion(dmNormalizeRule.get("rule_content").toString(), destMap);
//long ruleid=Long.parseLong(dmNormalizeRule.get("id").toString());
SaveNormalData(deviceID, DateTime.now(), destMap, Long.parseLong(dmNormalizeRule.get("id").toString()), dmNormalizeRule.get("name").toString());
// 数据处理-补全规则
logDataCompleteService.processDataCompletion(
dmNormalizeRule.get("rule_content").toString(), destMap);
SaveNormalData(ctx, deviceID, DateTime.now(), destMap,
Long.parseLong(dmNormalizeRule.get("id").toString()),
dmNormalizeRule.get("name").toString());
} catch (Exception ex) {
logger.error("泛化规则处理失败:" + ex.getMessage());
System.out.println(ex.getMessage());
/**
//保存非标日志信息
syslogNonNormalMessage.setReason("泛化规则处理失败");
syslogNonNormalMessage.setReasonDetail("泛化规则处理失败,失败详情:"+ ex.getMessage());
SaveNonNormalMessage(deviceID,DateTime.now());
**/
}
}
}
catch (Exception ex)
{
logger.error("处理日志消息异常:"+ex.getMessage());
} catch (Exception ex) {
logger.error("处理日志消息异常:" + ex.getMessage());
System.out.println(ex.getMessage());
}
}
// ==================== 以下为原有业务方法 ====================
//获取数据泛化规则的列表字段,提取 Cropper_params
public Cropper_paramsType getCropperParams( Map<String, Object> dmNormalizeRule )
{
Cropper_paramsType cropperParams = new Cropper_paramsType();
public Cropper_paramsType getCropperParams(Map<String, Object> dmNormalizeRule) {
Cropper_paramsType cropperParams = new Cropper_paramsType();
try {
JSONObject jsonObject = new JSONObject( dmNormalizeRule.get("rule_content").toString());
JSONObject jsonObject = new JSONObject(dmNormalizeRule.get("rule_content").toString());
if (jsonObject.isEmpty()) return null;
cropperParams.sethead_key(jsonObject.getJSONObject("cropper_params").get("head_key").toString());
//需要判断是否字符串head_offset是否空
cropperParams.sethead_offset(Integer.parseInt(jsonObject.getJSONObject("cropper_params").get("head_offset").toString()));
cropperParams.sethead_offset(Integer.parseInt(
jsonObject.getJSONObject("cropper_params").get("head_offset").toString()));
cropperParams.settail_key(jsonObject.getJSONObject("cropper_params").get("tail_key").toString());
//需要判断是否字符串tail_offset是否空
cropperParams.settail_offset( Integer.parseInt(jsonObject.getJSONObject("cropper_params").get("tail_offset").toString()));
cropperParams.settail_offset(Integer.parseInt(
jsonObject.getJSONObject("cropper_params").get("tail_offset").toString()));
return cropperParams;
}catch (Exception ex) {
logger.error("getCropperParams"+ex.getMessage());
} catch (Exception ex) {
logger.error("getCropperParams" + ex.getMessage());
return null;
}
}
public List<HashMap<String ,Object>> getRuleContentMappers( Map<String, Object> dmNormalizeRule )
{
List<HashMap<String, Object>> filedList =null;
public List<HashMap<String, Object>> getRuleContentMappers(Map<String, Object> dmNormalizeRule) {
List<HashMap<String, Object>> filedList = null;
ObjectMapper objectMapper = new ObjectMapper();
try {
// 转换为list Map
//Map<String, Object> rule_contentMap = objectMapper.readValue( dmNormalizeRule.get("rule_content").toString(), Map.class);
//if (rule_contentMap.isEmpty()) return null;
JSONObject jsonObject = new JSONObject( dmNormalizeRule.get("rule_content").toString());
JSONObject jsonObject = new JSONObject(dmNormalizeRule.get("rule_content").toString());
if (jsonObject.isEmpty()) return null;
ObjectMapper objectMapper_mappers = new ObjectMapper();
filedList =readJsonToList( jsonObject.get("mappers").toString() );
}catch (Exception ex) {
logger.error("getRuleContentMappers异常:"+ex.getMessage());
filedList = readJsonToList(jsonObject.get("mappers").toString());
} catch (Exception ex) {
logger.error("getRuleContentMappers异常:" + ex.getMessage());
}
return filedList;
}
public List<HashMap<String, Object>> getCompleteColumnsList( Map<String, Object> dmNormalizeRule )
{
List<HashMap<String, Object>> filedList =null;
ObjectMapper objectMapper = new ObjectMapper();
public List<HashMap<String, Object>> getCompleteColumnsList(Map<String, Object> dmNormalizeRule) {
List<HashMap<String, Object>> filedList = null;
try {
// 转换为list Map
//Map<String, Object> rule_contentMap = objectMapper.readValue( dmNormalizeRule.get("rule_content").toString(), Map.class);
//if (rule_contentMap.isEmpty()) return null;
JSONObject jsonObject = new JSONObject( dmNormalizeRule.get("rule_content").toString());
JSONObject jsonObject = new JSONObject(dmNormalizeRule.get("rule_content").toString());
if (jsonObject.isEmpty()) return null;
ObjectMapper objectMapper_mappers = new ObjectMapper();
filedList =readJsonToList( jsonObject.get("complete_columns").toString() );
}catch (Exception ex) {
logger.error("getCompleteColumnsList"+ex.getMessage());
filedList = readJsonToList(jsonObject.get("complete_columns").toString());
} catch (Exception ex) {
logger.error("getCompleteColumnsList" + ex.getMessage());
}
return filedList;
}
/**
* 将JSON 字符串转换为 List<HashMap<String, Object>>
*/
public static List<HashMap<String, Object>> readJsonToList(String jsonStr) {
ObjectMapper objectMapper = new ObjectMapper();
try {
return objectMapper.readValue(
jsonStr,
new TypeReference<List<HashMap<String, Object>>>() {}
);
return objectMapper.readValue(jsonStr,
new TypeReference<List<HashMap<String, Object>>>() {});
} catch (Exception e) {
logger.error("readJsonToList 解析失败:"+jsonStr);
//throw new RuntimeException("JSON 解析失败: " + e.getMessage(), e);
logger.error("readJsonToList 解析失败:" + jsonStr);
return null;
}
}
/**\
* 判断日志信息的数据类型
* @param logMsg
* @return
*/
public String getLogDataType( String logMsg)
{
//判断日志信息的数据类型(json、regex、kv、sep、xml
//默认返回json
public String getLogDataType(String logMsg) {
return "json";
}
public int getDeviceID( String logMsg)
{
//解析日志信息头部,获取device_id
// 默认返回device_id =1
public int getDeviceID(String logMsg) {
return 1;
}
//获取log 内容字段及数据值
public LinkedHashMap<String, Object> getMessageToMap( String MsgContent )
{
LinkedHashMap<String, Object> linkMap =null;
if(strDataType.equals("json")) {
//json 类型
//解析SyslogMessage
public LinkedHashMap<String, Object> getMessageToMap(String MsgContent) {
LinkedHashMap<String, Object> linkMap = null;
if ("json".equals("json")) { // 默认走 json 分支
SyslogMessage msg = SyslogParser.parse(MsgContent);
System.out.println("解析结果-log content: " + msg.getMessage().toString());
String complexJson = "{\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\",\"start_time\":\"2023-09-03T11:07:02.50167643Z\",\"time\":\"2023-09-03T11:16:18.883885281Z\",\"risk_level\":4,\"connection\":\"b18f3fbe-3fbf-4495-815f-ff26f6fb0bdf\",\"file_info\":null,\"extra\":{\"payload\":{\"format\":\"line\",\"name\":{\"cn\":\"攻击载荷\",\"en\":\"payload\"},\"value\":\"\"},\"uid\":{\"format\":\"line\",\"name\":{\"cn\":\"\",\"en\":\"\"},\"uid\":\"b4cbc73c-25d0-4429-ae1b-a856cdf1a651\",\"value\":\"\"}},\"type\":\"WEB_ATTACK_SCANNER\",\"agent_sn\":\"caa7da42-0cca-4cb1-b501-1f1eb2b588d5\",\"agent_name\":\" 教育局蜜罐探针\",\"honeypot_id\":\"11a9ac6bdf38ae2aaa49ec4f1b4a921bff71952cb9f175bdd8ee1f0497057bc6\",\"honeypot_name\":\"茂名市中小学管理平台管理后台\",\"src_ip\":\"117.50.189.7\",\"src_port\":58512,\"src_mac\":\"\",\"dest_ip\":\"192.168.222.2\",\"dest_port\":9200,\"proxy_ip\":null,\"node\":\"WRx3\"}";
String complexJson = "{\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\"}";
System.out.println("complexJson content: " + complexJson);
//静态字符串、编码有问题,临时用 静态字符串做测试,流程环境没问题
if(AppConfig.geRunEnvironment().equals("dev")) {
//LinkedHashMap<String, Object> flatMap = JsonParser.parseJsonToFlatMap(complexJson);
LinkedHashMap<String, Object> flatMap =JsonParser.jsonToMap(complexJson);
if ("dev".equals(appProperties.getRunEnvironment())) {
LinkedHashMap<String, Object> flatMap = JsonParser.jsonToMap(complexJson);
flatMap.forEach((key, value) -> System.out.println(key + " = " + value));
return flatMap;
}
else
{
//LinkedHashMap<String, Object> flatMap = JsonParser.parseJsonToFlatMap(msg.getMessage());
LinkedHashMap<String, Object> flatMap =JsonParser.jsonToMap(msg.getMessage());
} else {
LinkedHashMap<String, Object> flatMap = JsonParser.jsonToMap(msg.getMessage());
flatMap.forEach((key, value) -> System.out.println(key + " = " + value));
return flatMap;
}
}
else if(strDataType.equals("kv")) //key-value 键值类型
{
return null;
}
else if(strDataType.equals("sep")) //分隔符
{
return null;
}
else if(strDataType.equals("xml")) //类型 xml
{
return null;
}
else if(strDataType.equals("regex")) //正则表达式
{
return null;
}
// List<Map<String, Object>> rulelst=dmNormalizeRuleService.selectByDeviceId((long)1);
return linkMap;
}
//获取log message内容字段及数据值
public LinkedHashMap<String, Object> getMessageToMap( Map<String, Object> dmNormalizeRule, String MsgContent , String dataType)
{
public LinkedHashMap<String, Object> getMessageToMap(
Map<String, Object> dmNormalizeRule, String MsgContent, String dataType) {
LinkedHashMap<String, Object> linkMap = new LinkedHashMap<>();
try {
//String decode =dmNormalizeRule.get("decode").toString();
Cropper_paramsType cropperParams = getCropperParams(dmNormalizeRule);
Cropper_paramsType cropperParams = getCropperParams(dmNormalizeRule);
if (dataType.equals("json")) {
String strJson = logNormalData.ParserMessageJsonType(MsgContent, cropperParams);
LinkedHashMap<String, Object> result1 = NestedJsonParserUtil.safeParseJson(strJson);
LinkedHashMap<String, Object> flattened = NestedJsonUtils.flattenNestedJson(result1);
return flattened;
//parseJsonToFlatMap 复杂的json转换、解析过程有异常
//return JsonParser.parseJsonToFlatMap(strJson);
// return JsonParser.jsonToMap(strJson);
}
else if (dataType.equals("kv")) //key-value 键值类型
{
} else if (dataType.equals("kv")) {
String strKeyVal = logNormalData.ParserMessageJsonType(MsgContent, cropperParams);
KvTextParser kvTextParser =new KvTextParser();
kv_paramsType kvparams=logNormalData.getkv_paramsType( dmNormalizeRule);
linkMap= kvTextParser.parseKvText(strKeyVal,kvparams);
KvTextParser kvTextParser = new KvTextParser();
kv_paramsType kvparams = logNormalData.getkv_paramsType(dmNormalizeRule);
linkMap = kvTextParser.parseKvText(strKeyVal, kvparams);
return linkMap;
}
else if (dataType.equals("sep")) //分隔符
{
//获取分隔符
String SepKey=logNormalData.sepType(dmNormalizeRule);
} else if (dataType.equals("sep")) {
String SepKey = logNormalData.sepType(dmNormalizeRule);
String strSep = logNormalData.ParserMessageJsonType(MsgContent, cropperParams);
linkMap= TextParserUtil.parseSeparatedText(strSep, SepKey);
linkMap = TextParserUtil.parseSeparatedText(strSep, SepKey);
return linkMap;
}
else if (dataType.equals("xml")) //类型 xml
{
} else if (dataType.equals("xml")) {
return linkMap;
}
else if (dataType.equals("regex")) //正则表达式
{
} else if (dataType.equals("regex")) {
String strRegex = logNormalData.ParserMessageJsonType(MsgContent, cropperParams);
String regexp= logNormalData.Regexp(dmNormalizeRule);
linkMap= RegexTextParser.parseWithRegex(strRegex,regexp );
String regexp = logNormalData.Regexp(dmNormalizeRule);
linkMap = RegexTextParser.parseWithRegex(strRegex, regexp);
return linkMap;
}
}catch (Exception ex) {
logger.error("getMessageToMap"+ex.getMessage());
return null;
} catch (Exception ex) {
logger.error("getMessageToMap" + ex.getMessage());
return null;
}
return linkMap;
}
/**
* 根据设备ID获取配置规则
* @param device_id
* @return List<Map<String, Object>>
*/
public List<Map<String, Object>> getRuleList( long device_id)
{
List<Map<String, Object>> rulelst=dmNormalizeRuleService.selectByDeviceId((long)1);
if (rulelst!=null)
{
public List<Map<String, Object>> getRuleList(long device_id) {
List<Map<String, Object>> rulelst = dmNormalizeRuleService.selectByDeviceId((long) 1);
if (rulelst != null) {
System.out.println("rulelst: " + rulelst);
}
else{
logger.error(" List<Map<String, Object>> rulelst is null" );
} else {
logger.error(" List<Map<String, Object>> rulelst is null");
}
return rulelst;
}
/**
* 查找命中配置规则的字段及数值
* @param destColumnList 泛化目标字段list
* @param destColumnList 日志源字段及数值list
* @return
*/
public List<HashMap<String,Object>> getNormalColumnList( LinkedHashMap<String, Object> orginColumnMap ,List<HashMap<String, Object>> destColumnList )
{
List<HashMap<String, Object>> columnlist =new ArrayList<>();
// 原始解析字段遍历泛化规则目标字段
for (Map.Entry<String, Object > entry : orginColumnMap.entrySet()) {
//System.out.println(entry.getKey() + ": " + entry.getValue());
public List<HashMap<String, Object>> getNormalColumnList(
LinkedHashMap<String, Object> orginColumnMap,
List<HashMap<String, Object>> destColumnList) {
List<HashMap<String, Object>> columnlist = new ArrayList<>();
for (Map.Entry<String, Object> entry : orginColumnMap.entrySet()) {
for (Map<String, Object> map : destColumnList) {
//System.out.println( "origin_field: " + map.get("origin_field").toString());
if ( map.get("origin_field").toString().equals(entry.getKey()) ) {
if (map.get("origin_field").toString().equals(entry.getKey())) {
System.out.println(map);
Map<String, Object> normalColumMap =new LinkedHashMap<>();
normalColumMap.put("origin_field",entry.getKey() );
normalColumMap.put("dest_field",map.get("dest_field").toString());
normalColumMap.put("action",(HashMap<String, Object>)map.get("action") );
//normalColumMap.put("mapping ",entry.getKey() );
normalColumMap.put("origin_field_value",entry.getValue() );
// System.out.println("action: " + map.get("action").toString());
if( ((HashMap<String, Object>)map.get("action")).get("type").equals("equal"))
//直接赋值
normalColumMap.put("dest_field_value",entry.getValue() );
else if(((HashMap<String, Object>)map.get("action")).get("type").equals("mapping"))
{
//mapping 映射枚举值
normalColumMap.put("dest_field_value",entry.getValue() );
normalColumMap.put("action_param",((HashMap<String, Object>)map.get("action")).get("param") );
HashMap<String, Object> action_param=(HashMap<String, Object>)((HashMap<String, Object>)map.get("action")).get("param") ;
//匹配并获取映射枚举值
normalColumMap.put("dest_field_value",getMappingValue(action_param ,entry.getValue().toString() ));
}
else if(((HashMap<String, Object>)map.get("action")).get("type").equals("time"))
{
//time 类型
//normalColumMap.put("dest_field_value",entry.getValue() );
normalColumMap.put("action_param",((HashMap<String, Object>)map.get("action")).get("param") );
HashMap<String, Object> action_param=(HashMap<String, Object>)((HashMap<String, Object>)map.get("action")).get("param") ;
//匹配时间格式并转成换整型格式
Map<String, Object> normalColumMap = new LinkedHashMap<>();
normalColumMap.put("origin_field", entry.getKey());
normalColumMap.put("dest_field", map.get("dest_field").toString());
normalColumMap.put("action", (HashMap<String, Object>) map.get("action"));
normalColumMap.put("origin_field_value", entry.getValue());
if (((HashMap<String, Object>) map.get("action")).get("type").equals("equal")) {
normalColumMap.put("dest_field_value", entry.getValue());
} else if (((HashMap<String, Object>) map.get("action")).get("type").equals("mapping")) {
normalColumMap.put("action_param",
((HashMap<String, Object>) map.get("action")).get("param"));
HashMap<String, Object> action_param =
(HashMap<String, Object>) ((HashMap<String, Object>) map.get("action")).get("param");
normalColumMap.put("dest_field_value",
getMappingValue(action_param, entry.getValue().toString()));
} else if (((HashMap<String, Object>) map.get("action")).get("type").equals("time")) {
normalColumMap.put("action_param",
((HashMap<String, Object>) map.get("action")).get("param"));
HashMap<String, Object> action_param =
(HashMap<String, Object>) ((HashMap<String, Object>) map.get("action")).get("param");
try {
long longTime = TimeConversionUtils.convertToMillis(entry.getValue().toString(), action_param.get("timezone").toString());
long longTime = TimeConversionUtils.convertToMillis(
entry.getValue().toString(), action_param.get("timezone").toString());
normalColumMap.put("dest_field_value", longTime);
} catch (Exception e) {
logger.error("时间类型转换错误,源值:" + entry.getValue().toString() + ",java_date_format:" + action_param.get("java_date_format").toString());
logger.error("时间类型转换错误,源值:" + entry.getValue().toString()
+ ",java_date_format:" + action_param.get("java_date_format").toString());
e.printStackTrace();
}
}
columnlist.add((HashMap<String, Object>)normalColumMap);
//System.out.println( "normalColumMap: " +normalColumMap);
//存在源字段配置多个目标字段,使用continue,而不是break
columnlist.add((HashMap<String, Object>) normalColumMap);
continue;
}
}
@@ -464,156 +363,125 @@ public class LogNormalProcessor {
return columnlist;
}
/**
* paramMapping Map匹配查找对应的值
* @param paramMappingValueMap
* @param
* @return
*/
public Object getMappingValue(HashMap<String, Object> paramMappingValueMap,String value)
{
HashMap<String, Object> Map= (HashMap<String, Object>)paramMappingValueMap.get("mapping");
for (Map.Entry<String, Object > entry : Map.entrySet()) {
if ( entry.getKey().equals(value) ) {
return entry.getValue();
public Object getMappingValue(HashMap<String, Object> paramMappingValueMap, String value) {
HashMap<String, Object> Map = (HashMap<String, Object>) paramMappingValueMap.get("mapping");
for (Map.Entry<String, Object> entry : Map.entrySet()) {
if (entry.getKey().equals(value)) {
return entry.getValue();
}
}
return null;
}
/**
* 获取字段Map,包含字段field及Value值
* @param normalColumnList
* @return
*/
public Map<String, Object > getColumnMap( List<HashMap<String,Object>> normalColumnList )
{
Map<String, Object > columnMap= new HashMap<>();
public Map<String, Object> getColumnMap(List<HashMap<String, Object>> normalColumnList) {
Map<String, Object> columnMap = new HashMap<>();
for (Map<String, Object> map : normalColumnList) {
Object destFieldValue = map.get("dest_field_value");
// 判断 dest_field_value 是否为 String 且包含 "\u0000"
if (destFieldValue instanceof String && ((String) destFieldValue).contains("\u0000")) {
// 替换掉所有 "\u0000" 字符
if (destFieldValue instanceof String
&& ((String) destFieldValue).contains("\u0000")) {
destFieldValue = ((String) destFieldValue).replace("\u0000", "");
}
columnMap.put(map.get("dest_field").toString(), destFieldValue);
//columnMap.put(map.get("dest_field").toString(),map.get("dest_field_value"));
}
return columnMap;
}
/**
* 保存数据到标准化表
* @param deviceId
* @param logtime
* @param logColumnMap
*/
public void SaveNormalData(long deviceId , DateTime logtime, Map<String, Object > logColumnMap, long normalizeRuleId ,String normalizeRuleName)
{
public void SaveNormalData(ProcessContext ctx, long deviceId, DateTime logtime,
Map<String, Object> logColumnMap,
long normalizeRuleId, String normalizeRuleName) {
try {
if(logColumnMap.isEmpty() )
{
logger.error("SaveNormalData ->logColumnMap 为空,syslogUUID:" +this.strSyslogUUID);
//保存非标日志信息
syslogNonNormalMessage.setReason("未命中规则");
syslogNonNormalMessage.setReasonDetail("失败详情:logColumnMap对象字段为空" );
if(isSaveNonNormal==false)
SaveNonNormalMessage(deviceId,DateTime.now());
return ;
if (logColumnMap.isEmpty()) {
logger.error("SaveNormalData ->logColumnMap 为空,syslogUUID:" + ctx.strSyslogUUID);
ctx.syslogNonNormalMessage.setReason("未命中规则");
ctx.syslogNonNormalMessage.setReasonDetail("失败详情:logColumnMap对象字段为空");
if (ctx.isSaveNonNormal == false)
SaveNonNormalMessage(ctx, deviceId, DateTime.now());
return;
}
Map<String, Object> columnMap = logColumnMap;
//补全设备信息字段
//补全采集设备信息
CompletionDeviceInfo(columnMap, deviceId);
//补全采集探针相关信息
CompletionCollectTaskInfo(columnMap,ctx);
//补全日志基础信息
columnMap.put("device_id", deviceId);
columnMap.put("log_time", logtime);
columnMap.put("id", UUID.randomUUID().toString());
columnMap.put("normalize_rule_id", normalizeRuleId);
columnMap.put("normalize_rule_name", normalizeRuleName);
columnMap.put("syslog_uuid", this.strSyslogUUID);
columnMap.put("syslog_topic", this.strSyslogTopic);
System.out.println("columnMap"+columnMap);
columnMap.put("syslog_uuid", ctx.strSyslogUUID);
columnMap.put("syslog_topic", ctx.strSyslogTopic);
System.out.println("columnMap" + columnMap);
syslogNormalDataService.insertDynamic(columnMap);
} catch (Exception e) {
logger.error("SaveNormalData失败 " );
//保存非标日志信息
syslogNonNormalMessage.setReason("入库失败");
syslogNonNormalMessage.setReasonDetail("入库失败,normalizeRuleName :"+normalizeRuleName+",失败详情:"+ e.getMessage());
if(isSaveNonNormal==false)
SaveNonNormalMessage(deviceId,DateTime.now());
logger.error("SaveNormalData失败 ");
ctx.syslogNonNormalMessage.setReason("入库失败");
ctx.syslogNonNormalMessage.setReasonDetail("入库失败,normalizeRuleName :" + normalizeRuleName
+ ",失败详情:" + e.getMessage());
if (ctx.isSaveNonNormal == false)
SaveNonNormalMessage(ctx, deviceId, DateTime.now());
throw new RuntimeException("SaveNormalData 失败: " + e.getMessage(), e);
}
}
/**
* 补全设备信息字段
* @param columnMap
* @param deviceID
*/
public void CompletionDeviceInfo(Map<String, Object> columnMap ,Long deviceID)
{
//补全采集设备信息
public void CompletionDeviceInfo(Map<String, Object> columnMap, Long deviceID) {
try {
DeviceDevice devInfo = deviceDeviceService.getByIdSafely( Long.valueOf(deviceID).intValue() );
DeviceDevice devInfo = deviceDeviceService.getByIdSafely(Long.valueOf(deviceID).intValue());
if (devInfo != null) {
columnMap.put("device_ip", devInfo.getIp());
columnMap.put("device_manufacturer", devInfo.getVendor());
columnMap.put("device_name", devInfo.getName());
}
} catch (Exception e) {
logger.error("CompletionDeviceInfo 失败!ID"+deviceID );
logger.error("CompletionDeviceInfo 失败!ID" + deviceID);
}
}
/**
* 保存非标日志记录
* @param deviceId
* @param logtime
*/
public void SaveNonNormalMessage(long deviceId , DateTime logtime)
{
//补全采集探针相关信息(ID、IP、名称)
public void CompletionCollectTaskInfo(Map<String, Object> columnMap,ProcessContext ctx) {
Integer collect_id=-1;
try {
SyslogNonNormalMessage normalMessage =new SyslogNonNormalMessage() ;
//获取日志对应的设备ID
Integer collect_id= Integer.parseInt(this.deviceInfoMap.get("device_collect_id"));
DeviceCollectTask deviceCollectTask=deviceCollectTaskService.getById(collect_id);
normalMessage.setId( UUID.randomUUID().toString());
normalMessage.setDeviceId((int)deviceId);
collect_id = Integer.parseInt(ctx.deviceInfoMap.get("device_collect_id"));
DeviceCollectTask deviceCollectTask = deviceCollectTaskService.getById(collect_id);
if (deviceCollectTask != null) {
columnMap.put("agent_id", collect_id);
columnMap.put("origin_agent_id",collect_id.toString() );
columnMap.put("probe_ip",deviceCollectTask.getDeviceIp() );
columnMap.put("agent_name", deviceCollectTask.getTaskName());
columnMap.put("origin_agent_name", deviceCollectTask.getTaskName());
}
} catch (Exception e) {
logger.error("CompletionCollectTaskInfo 失败!探针ID" + collect_id);
}
}
public void SaveNonNormalMessage(ProcessContext ctx, long deviceId, DateTime logtime) {
try {
SyslogNonNormalMessage normalMessage = new SyslogNonNormalMessage();
Integer collect_id = Integer.parseInt(ctx.deviceInfoMap.get("device_collect_id"));
DeviceCollectTask deviceCollectTask = deviceCollectTaskService.getById(collect_id);
normalMessage.setId(UUID.randomUUID().toString());
normalMessage.setDeviceId((int) deviceId);
normalMessage.setLogTime(logtime.toLocalDateTime());
normalMessage.setRuleTime(logtime.toLocalDateTime());
normalMessage.setSyslogMessage(this.strLogMsg);
normalMessage.setSyslogTopic(this.strSyslogTopic);
normalMessage.setSyslogUuid(this.strSyslogUUID);
normalMessage.setHeaderMessage(this.strDeviceInfo);
normalMessage.setSyslogMessage(ctx.strLogMsg);
normalMessage.setSyslogTopic(ctx.strSyslogTopic);
normalMessage.setSyslogUuid(ctx.strSyslogUUID);
normalMessage.setHeaderMessage(ctx.strDeviceInfo);
normalMessage.setEtlNode("etlgo");
normalMessage.setReason( syslogNonNormalMessage.getReason());
normalMessage.setReasonDetail( syslogNonNormalMessage.getReasonDetail());
//采集探针名称
normalMessage.setReason(ctx.syslogNonNormalMessage.getReason());
normalMessage.setReasonDetail(ctx.syslogNonNormalMessage.getReasonDetail());
normalMessage.setRuleResult("FAIL");
normalMessage.setDeviceName(deviceInfoMap.get("device_name"));
normalMessage.setDeviceName(ctx.deviceInfoMap.get("device_name"));
normalMessage.setCollectTaskId(collect_id);
if(deviceCollectTask!=null)
normalMessage.setCollectTaskName(deviceCollectTask.getTaskName());
this.messageService.saveMessage(normalMessage );
isSaveNonNormal=true;
if (deviceCollectTask != null)
normalMessage.setCollectTaskName(deviceCollectTask.getTaskName());
this.messageService.saveMessage(normalMessage);
ctx.isSaveNonNormal = true;
} catch (Exception e) {
logger.error("SaveNonNormalMessage 失败:" );
logger.error("SaveNonNormalMessage 失败:");
throw new RuntimeException("SaveNonNormalMessage 失败: " + e.getMessage(), e);
}
}
public static void main(String[] args) {
String strlogMessage= "<14>1 2025-09-24T11:52:26Z 5f46d3be75e1 supermario 128 honeypot_event - {\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\",\"start_time\":\"2023-09-03T11:07:02.50167643Z\",\"time\":\"2023-09-03T11:16:18.883885281Z\",\"risk_level\":4,\"connection\":\"b18f3fbe-3fbf-4495-815f-ff26f6fb0bdf\",\"file_info\":null,\"extra\":{\"payload\":{\"format\":\"line\",\"name\":{\"cn\":\"攻击载荷\",\"en\":\"payload\"},\"value\":\"\"},\"uid\":{\"format\":\"line\",\"name\":{\"cn\":\"\",\"en\":\"\"},\"uid\":\"b4cbc73c-25d0-4429-ae1b-a856cdf1a651\",\"value\":\"\"}},\"type\":\"WEB_ATTACK_SCANNER\",\"agent_sn\":\"caa7da42-0cca-4cb1-b501-1f1eb2b588d5\",\"agent_name\":\" 教育局蜜罐探针\",\"honeypot_id\":\"11a9ac6bdf38ae2aaa49ec4f1b4a921bff71952cb9f175bdd8ee1f0497057bc6\",\"honeypot_name\":\"茂名市中小学管理平台管理后台\",\"src_ip\":\"117.50.189.7\",\"src_port\":58512,\"src_mac\":\"\",\"dest_ip\":\"192.168.222.2\",\"dest_port\":9200,\"proxy_ip\":null,\"node\":\"WRx3\"}";
String strMsgContent= "{\"source\":\"honeypot1\",\"id\":\"f6a13c35-bf9d-4da6-a181-50ce23e7ef6a\",\"start_time\":\"2023-09-03T11:07:02.50167643Z\",\"time\":\"2023-09-03T11:16:18.883885281Z\",\"risk_level\":4,\"connection\":\"b18f3fbe-3fbf-4495-815f-ff26f6fb0bdf\",\"file_info\":null,\"extra\":{\"payload\":{\"format\":\"line\",\"name\":{\"cn\":\"攻击载荷\",\"en\":\"payload\"},\"value\":\"\"},\"uid\":{\"format\":\"line\",\"name\":{\"cn\":\"\",\"en\":\"\"},\"uid\":\"b4cbc73c-25d0-4429-ae1b-a856cdf1a651\",\"value\":\"\"}},\"type\":\"WEB_ATTACK_SCANNER\",\"agent_sn\":\"caa7da42-0cca-4cb1-b501-1f1eb2b588d5\",\"agent_name\":\" 教育局蜜罐探针\",\"honeypot_id\":\"11a9ac6bdf38ae2aaa49ec4f1b4a921bff71952cb9f175bdd8ee1f0497057bc6\",\"honeypot_name\":\"茂名市中小学管理平台管理后台\",\"src_ip\":\"117.50.189.7\",\"src_port\":58512,\"src_mac\":\"\",\"dest_ip\":\"192.168.222.2\",\"dest_port\":9200,\"proxy_ip\":null,\"node\":\"WRx3\"}";
//Map<String, Object> flatMap =(new logNormalData()).getMessageToMap(strMsgContent);
//flatMap.forEach((key, value) -> System.out.println(key + " = " + value));
//LogNormalProcessor logData =new LogNormalProcessor(strlogMessage,);
//logData.init();
}
}
@@ -2,7 +2,8 @@ package com.Modules.NormalData;
import com.influx.InfluxDBClient;
import com.common.util.SyslogParser;
import com.config.AppConfig;
import com.config.AppProperties;
import com.config.KafkaConsumerProperties;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.client.WriteApi;
@@ -29,7 +30,6 @@ import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import com.common.util.Sm4Util;
import com.config.AppConfig;
@Slf4j
@Component
public class SysLogProcessor {
@@ -44,11 +44,20 @@ public class SysLogProcessor {
@Value("${app.processor.process-timeout-ms:30000}")
private long processTimeoutMs;
private static String strhexKey=AppConfig.getSM4Key();
@Autowired
private AppProperties appProperties;
@Autowired
private KafkaConsumerProperties kafkaConsumerProperties;
@Autowired
private LogNormalProcessor logNormalProcessor;
@Autowired
private com.influx.InfluxDBClient influxClient;
private final AtomicInteger totalProcessed = new AtomicInteger(0);
private final AtomicInteger currentBatchCount = new AtomicInteger(0);
// 初始化 InfluxDB 客户端
private final com.influx.InfluxDBClient influxClient = new InfluxDBClient();
@@ -88,7 +97,7 @@ public class SysLogProcessor {
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
try {
// 异步处理单条消息
log.info("收到syslogmessage"+ Sm4Util.decryptCbc(record.value(), strhexKey));
log.info("收到syslogmessage"+ Sm4Util.decryptCbc(record.value(), appProperties.getSm4Key()));
processSingleMessageAsync(record);
} catch (Exception e) {
log.error("处理消息失败, topic: {}, partition: {}, offset: {}",
@@ -254,13 +263,8 @@ public class SysLogProcessor {
long startTime = System.currentTimeMillis();
try {
// 模拟消息解析
//String message = parseMessage(record);
// 模拟业务处理
//processBusinessLogic(message);
//Message进行SM4解密
String Sm4message=Sm4Util.decryptCbc(record.value(), strhexKey);
String Sm4message=Sm4Util.decryptCbc(record.value(), appProperties.getSm4Key());
System.out.println("Sm4message:"+Sm4message);
@@ -273,7 +277,7 @@ public class SysLogProcessor {
.addTag("device_id", mapdev.get("device_id")) // 添加设备ID标签
.addTag("device_collect_id", mapdev.get("device_collect_id")) // 添加探针ID标签
.addTag("uuid", sysLogUUID) //syslog uuid
.addTag("topic", AppConfig.getTopic()) //kafka topic
.addTag("topic", kafkaConsumerProperties.getTopic()) //kafka topic
.addField("message", Sm4message) // 添加字段
.addField("receive_time", mapdev.get("receive_time")) // 添加字段
.addField("uuid", sysLogUUID)
@@ -286,12 +290,10 @@ public class SysLogProcessor {
// 日志信息插入pg XdrHoneypot 表
//insertSingleRecord( record.value());
//String syslogMessage= AppConfig.geRunEnvironment().equals("test")? record.value().substring(34) : record.value();
//String syslogMessage= appProperties.getRunEnvironment().equals("test")? record.value().substring(34) : record.value();
String syslogMessage= Sm4message;
//剔除测试环境本机syslog新增的头部信息
LogNormalProcessor logNormalProcessor = new LogNormalProcessor(syslogMessage,sysLogUUID,AppConfig.getTopic());
//LogNormalProcessor logNormalProcessor =new LogNormalProcessor(record.value());
logNormalProcessor.init();
//使用注入的 Spring Bean 进行标准化处理
logNormalProcessor.process(syslogMessage, sysLogUUID, null);
System.out.println("insert postgres syslog ,value:"+ record.key());
long costTime = System.currentTimeMillis() - startTime;
log.debug("消息处理完成, offset: {}, 耗时: {}ms", record.offset(), costTime);
@@ -75,4 +75,8 @@ public class AlarmVisit {
private String[] httpReqBody;
private String[] httpRespHeader;
private String[] httpRespBody;
private String reason;
private String dtype;
private String originField;
private String originLog;
}
@@ -0,0 +1,18 @@
package com.common.entity;
import lombok.Builder;
import lombok.Data;
import java.time.LocalDateTime;
@Data
@Builder
public class AlarmVisitProcesslog {
private Long id;
private String batchNo;
private Long logCount;
private LocalDateTime startTime;
private LocalDateTime endTime;
private LocalDateTime createdAt;
private String remark;
}
@@ -20,6 +20,10 @@ public class DeviceCollectTask {
private Integer taskCount;
private OffsetDateTime recentDiscoverTime;
private Integer epmUpperLimit;
private OffsetDateTime expireTime;
private String remark;
private String deviceIp;
private Long organizationId;
// Getter and Setter 方法
public Integer getId() { return id; }
@@ -73,6 +77,18 @@ public class DeviceCollectTask {
public Integer getEpmUpperLimit() { return epmUpperLimit; }
public void setEpmUpperLimit(Integer epmUpperLimit) { this.epmUpperLimit = epmUpperLimit; }
public OffsetDateTime getExpireTime() { return expireTime; }
public void setExpireTime(OffsetDateTime expireTime) { this.expireTime = expireTime; }
public String getRemark() { return remark; }
public void setRemark(String remark) { this.remark = remark; }
public String getDeviceIp() { return deviceIp; }
public void setDeviceIp(String deviceIp) { this.deviceIp = deviceIp; }
public Long getOrganizationId() { return organizationId; }
public void setOrganizationId(Long organizationId) { this.organizationId = organizationId; }
@Override
public String toString() {
return "DeviceCollectTask{" +
@@ -93,6 +109,10 @@ public class DeviceCollectTask {
", taskCount=" + taskCount +
", recentDiscoverTime=" + recentDiscoverTime +
", epmUpperLimit=" + epmUpperLimit +
", expireTime=" + expireTime +
", remark='" + remark + '\'' +
", deviceIp='" + deviceIp + '\'' +
", organizationId=" + organizationId +
'}';
}
}
@@ -23,7 +23,8 @@ public interface AlarmVisitMapper {
"attack_port, victim_port, attack_method, etl_time, log_count, ",
"attack_chain_phase, disposition_advice, attack_direction, ",
"judged_state, disposed_state, attack_result, fall, payload, dns_info, engine_type, " ,
"http_req_header , http_req_body,http_resp_header , http_resp_body ",
"http_req_header , http_req_body,http_resp_header , http_resp_body, ",
"reason, dtype, origin_field, origin_log ",
") VALUES ",
"<foreach collection='list' item='item' separator=','>",
"(#{item.id}, #{item.createdAt}, #{item.alarmName}, #{item.alarmLevel}, ",
@@ -34,7 +35,7 @@ public interface AlarmVisitMapper {
"#{item.deviceId, typeHandler=com.Modules.etl.handler.ArrayIntegerTypeHandler}, ",
"#{item.comment}, " ,
"#{item.originLogIds, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, ",
"#{item.logStartAt}, #{item.logEndAt},, #{item.windowTime} #{item.httpStatus}, ",
"#{item.logStartAt}, #{item.logEndAt}, #{item.windowTime}, #{item.httpStatus}, ",
"#{item.attackPort, typeHandler=com.Modules.etl.handler.ArrayIntegerTypeHandler}, ",
"#{item.victimPort, typeHandler=com.Modules.etl.handler.ArrayIntegerTypeHandler}, ",
"#{item.attackMethod}, #{item.etlTime}, #{item.logCount}, ",
@@ -45,7 +46,8 @@ public interface AlarmVisitMapper {
"#{item.httpReqHeader, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, ",
"#{item.httpReqBody, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, ",
"#{item.httpRespHeader, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, ",
"#{item.httpRespBody, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}) ",
"#{item.httpRespBody, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, ",
"#{item.reason}, #{item.dtype}, #{item.originField}, #{item.originLog}) ",
"</foreach>",
"</script>"})
void batchInsert(@Param("list") List<AlarmVisit> alarmList);
@@ -60,7 +62,8 @@ public interface AlarmVisitMapper {
"attack_port, victim_port, attack_method, etl_time, log_count, " +
"attack_chain_phase, disposition_advice, attack_direction, " +
"judged_state, disposed_state, attack_result, fall, payload, dns_info,engine_type, " +
"http_req_header , http_req_body,http_resp_header , http_resp_body " +
"http_req_header , http_req_body,http_resp_header , http_resp_body, " +
"reason, dtype, origin_field, origin_log " +
") VALUES (" +
"#{id}, #{createdAt}, #{alarmName}, #{alarmLevel}, " +
"#{alarmType}, #{alarmMajorType}, #{alarmMinorType}, #{alarmAreaId}, " +
@@ -80,7 +83,8 @@ public interface AlarmVisitMapper {
"#{httpReqHeader, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, " +
"#{httpReqBody, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, " +
"#{httpRespHeader, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, " +
"#{httpRespBody, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler} " +
"#{httpRespBody, typeHandler=com.Modules.etl.handler.ArrayStringTypeHandler}, " +
"#{reason}, #{dtype}, #{originField}, #{originLog} " +
")")
void insert(AlarmVisit alarm);
}
@@ -0,0 +1,27 @@
package com.common.mapper;
import com.common.entity.AlarmVisitProcesslog;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Select;
@Mapper
public interface AlarmVisitProcesslogMapper {
/**
* 插入处理记录
*/
@Insert("INSERT INTO alarm_visit_processlog (" +
"batch_no, log_count, start_time, end_time, remark" +
") VALUES (" +
"#{batchNo}, #{logCount}, #{startTime}, #{endTime}, #{remark}" +
")")
void insert(AlarmVisitProcesslog processlog);
/**
* 查询当天最大批次序号
*/
@Select("SELECT COALESCE(MAX(batch_no), '') FROM alarm_visit_processlog " +
"WHERE batch_no LIKE CONCAT(#{todayPrefix}, '%')")
String getTodayMaxBatchNo(String todayPrefix);
}
@@ -66,10 +66,12 @@ public interface DeviceCollectTaskMapper extends BaseMapper<DeviceCollectTask>{
*/
@Insert("INSERT INTO device_collect_task (created_at, updated_at, device_id, method, task_name, " +
"first_time, last_success_time, last_failed_time, detail_id, epm, epm_peak, " +
"process_architecture, task_count, recent_discover_time, epm_upper_limit) " +
"process_architecture, task_count, recent_discover_time, epm_upper_limit, " +
"expire_time, remark, device_ip, organization_id) " +
"VALUES (NOW(), NOW(), #{deviceId}, #{method}, #{taskName}, #{firstTime}, " +
"#{lastSuccessTime}, #{lastFailedTime}, #{detailId}, #{epm}, #{epmPeak}, " +
"#{processArchitecture}, #{taskCount}, #{recentDiscoverTime}, #{epmUpperLimit})")
"#{processArchitecture}, #{taskCount}, #{recentDiscoverTime}, #{epmUpperLimit}, " +
"#{expireTime}, #{remark}, #{deviceIp}, #{organizationId})")
@Options(useGeneratedKeys = true, keyProperty = "id")
int insert(DeviceCollectTask task);
@@ -90,7 +92,11 @@ public interface DeviceCollectTaskMapper extends BaseMapper<DeviceCollectTask>{
"process_architecture = #{processArchitecture}, " +
"task_count = #{taskCount}, " +
"recent_discover_time = #{recentDiscoverTime}, " +
"epm_upper_limit = #{epmUpperLimit} " +
"epm_upper_limit = #{epmUpperLimit}, " +
"expire_time = #{expireTime}, " +
"remark = #{remark}, " +
"device_ip = #{deviceIp}, " +
"organization_id = #{organizationId} " +
"WHERE id = #{id}")
int update(DeviceCollectTask task);
@@ -189,4 +195,10 @@ public interface DeviceCollectTaskMapper extends BaseMapper<DeviceCollectTask>{
*/
@Select("SELECT * FROM device_collect_task")
List<DeviceCollectTask> selectAllTasks();
/**
* 查询 first_time 为空的设备采集任务 ID 列表(用于首次成功时间填充)
*/
@Select("SELECT id FROM device_collect_task WHERE first_time IS NULL")
List<Integer> selectTasksWithoutFirstTime();
}
@@ -100,13 +100,19 @@ public interface DeviceReceiveLogMapper {
List<DeviceCollectTaskTime> selectDailyFailTimes();
/**
* 查询每个设备的首次成功时间(历史最早
* 按 ID 列表精确查询首次成功时间(仅查 first_time 为空的任务,避免全表扫描
*/
@Select("SELECT device_collect_id, MIN(created_at) AS first_success_time " +
@Select("<script>" +
"SELECT device_collect_id, MIN(created_at) AS first_success_time " +
"FROM device_receive_log " +
"WHERE push_success = 1 " +
"GROUP BY device_collect_id")
List<DeviceCollectTaskTime> selectFirstSuccessTimes();
"AND device_collect_id IN " +
"<foreach collection='ids' item='id' open='(' separator=',' close=')'>" +
"#{id}" +
"</foreach>" +
" GROUP BY device_collect_id" +
"</script>")
List<DeviceCollectTaskTime> selectFirstSuccessTimesByIds(@Param("ids") List<Integer> ids);
/**
* 查询指定时间范围内的成功记录
@@ -85,7 +85,7 @@ public interface SyslogNormalDataMapper {
* 查询指定时间后的所需字段(使用XML映射)
* 注意:参数类型保持为LocalDateTime,但实际映射时使用类型处理器转换
*/
List<SyslogNormalData> findRequiredFieldsAfterTime(@Param("startTime") LocalDateTime startTime);
List<SyslogNormalData> findRequiredFieldsAfterTime(@Param("startTime") LocalDateTime startTime,@Param("endTime") LocalDateTime endTime );
/**
* 分页查询所需字段
@@ -116,9 +116,9 @@ public class ETLOrchestrator {
public void cleanupOldLogs() {
try {
//默认删除7天内接收日志记录
LocalDateTime cutoffTime = LocalDateTime.now().minusDays(7);
LocalDateTime cutoffTime = LocalDateTime.now().minusDays(2);
int deleted = deviceReceiveLogService.deleteOldLogs(cutoffTime);
log.info("定时清理任务完成,删除{}条7天前的日志", deleted);
log.info("定时清理任务完成,删除{}条2天前的日志", deleted);
} catch (Exception e) {
log.error("定时清理日志失败", e);
}
@@ -6,13 +6,9 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.common.service.PartitionTableService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
public class PartitionTableSchedule {
@@ -21,11 +17,15 @@ public class PartitionTableSchedule {
@Autowired
private PartitionTableService partitionTableService;
// ============================================================
// 原有定时任务 — 每月15日凌晨1点(CREATE TABLE ... PARTITION OF 方式)(取消)
// ============================================================
/**
* 每月15日凌晨1点执行
* cron表达式: 秒 分 时 日 月 周
*/
@Scheduled(cron = "0 0 1 15 * ?")
//@Scheduled(cron = "0 0 1 15 * ?")
public void createNextMonthPartitionTables() {
logger.info("开始执行创建下个月分区表任务...");
@@ -41,6 +41,61 @@ public class PartitionTableSchedule {
}
}
// ============================================================
// 新增定时任务 — 通过达梦存储过程预创建下月分区(凌晨2:00)
// 涉及5个主表: alarm / alarm_visit / syslog_normal_data
// / syslog_normal_alarm / syslog_non_normal_message
// 存储过程: CREATE_TABLE_DAILY_PARTITIONS
// ============================================================
/**
* 每月15日凌晨2:00 — 预创建下月分区表
* cron: 秒 分 时 日 月 周
*/
@Scheduled(cron = "0 0 2 15 * ?")
public void createNextMonthPartitionsOn15th() {
executeMonthlyPartitionCreation("15日");
}
/**
* 每月28日凌晨2:00 — 预创建下月分区表
*/
@Scheduled(cron = "0 0 2 28 * ?")
public void createNextMonthPartitionsOn28th() {
executeMonthlyPartitionCreation("28日");
}
/**
* 每月最后一天凌晨2:00 — 预创建下月分区表
* Spring CronExpression 支持 L 表示最后一天
*/
@Scheduled(cron = "0 0 2 L * ?")
public void createNextMonthPartitionsOnLastDay() {
executeMonthlyPartitionCreation("月末最后一天");
}
/**
* 统一的月度分区创建执行逻辑
* 通过达梦存储过程 CREATE_TABLE_DAILY_PARTITIONS 批量创建
*/
private void executeMonthlyPartitionCreation(String triggerDesc) {
logger.info("==================== 月度分区预创建任务触发 ====================");
logger.info("触发时机: 每月{} 凌晨2:00", triggerDesc);
try {
partitionTableService.createNextMonthPartitionsByProcedure();
logger.info("月度分区预创建任务执行完成 (触发: {})", triggerDesc);
} catch (Exception e) {
logger.error("月度分区预创建任务执行失败 (触发: {}), 错误: {}", triggerDesc, e.getMessage(), e);
}
logger.info("================================================================");
}
// ============================================================
// 已注释的备选定时任务
// ============================================================
/**
* 测试任务 - 每分钟执行一次(开发环境使用)
* 生产环境注释或删除此方法
@@ -6,6 +6,7 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.common.entity.AlarmVisit;
import com.common.entity.AlarmVisitProcesslog;
import com.common.entity.DeviceDevice;
import com.common.entity.SecExceptionAlgorithm;
import com.common.entity.SyslogNormalData;
@@ -32,6 +33,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import com.common.mapper.*;
import java.net.InetAddress;
import com.common.util.AlgorithmResultParser;
import java.time.temporal.ChronoUnit;
@Slf4j
@Service
@EnableScheduling
@@ -55,6 +57,9 @@ public class AccessLogAlertService {
@Autowired
private AlarmVisitMapper alarmVisitMapper;
@Autowired
private AlarmVisitProcesslogMapper alarmVisitProcesslogMapper;
// 记录上次处理的时间戳
private LocalDateTime lastProcessTime;
@@ -67,8 +72,8 @@ public class AccessLogAlertService {
@PostConstruct
public void init() {
// 初始化时设置为当前时间减2分钟
lastProcessTime = LocalDateTime.now().minusMinutes(1);
// 初始化时设置为当前时间减1分钟
lastProcessTime = LocalDateTime.now().minusMinutes(1).truncatedTo(ChronoUnit.MINUTES);
log.info("初始化AccessLogAlertService,上次处理时间: {}", lastProcessTime);
// 加载启用的算法配置到缓存
@@ -111,9 +116,9 @@ public class AccessLogAlertService {
}
}
/**
* 定时任务入口 - 每2分钟执行一次
* 定时任务入口 - 每1分钟执行一次
*/
@Scheduled(cron = "0 */1 * * * ?")
//@Scheduled(cron = "0 */1 * * * ?")
@Async
public void processAccessLogAlert() {
log.info("开始执行访问日志告警处理任务");
@@ -128,9 +133,9 @@ public class AccessLogAlertService {
}
// 2. 获取上次处理时间之后的日志数据
LocalDateTime currentTime = LocalDateTime.now();
LocalDateTime currentTime = LocalDateTime.now().truncatedTo(ChronoUnit.MINUTES);
//List<SyslogNormalData> newLogs = syslogNormalDataMapper.findAfterTime(lastProcessTime);
List<SyslogNormalData> newLogs = syslogNormalDataMapper.findRequiredFieldsAfterTime(lastProcessTime);
List<SyslogNormalData> newLogs = syslogNormalDataMapper.findRequiredFieldsAfterTime(lastProcessTime,currentTime );
//List<SyslogNormalData> newLogs =algorithmResultParser.buildNewLogsFromExample();
if (newLogs.isEmpty()) {
log.info("没有发现新的日志数据,上次处理时间: {}", lastProcessTime);
@@ -140,6 +145,20 @@ public class AccessLogAlertService {
log.info("获取到 {} 条新的日志数据,时间范围: {} 到 {}",
newLogs.size(), lastProcessTime, currentTime);
// 2.1 记录本次处理批次信息
try {
String batchNo = generateBatchNo();
AlarmVisitProcesslog processlog = AlarmVisitProcesslog.builder()
.batchNo(batchNo)
.logCount((long) newLogs.size())
.startTime(lastProcessTime)
.endTime(currentTime)
.build();
alarmVisitProcesslogMapper.insert(processlog);
log.info("写入处理批次记录: batchNo={}, logCount={}", batchNo, newLogs.size());
} catch (Exception e) {
log.error("记录处理批次失败: {}", e.getMessage(), e);
}
// 3. 处理每条启用的算法配置
for (SecExceptionAlgorithm algorithm : algorithmCache.values()) {
try {
@@ -360,6 +379,10 @@ public class AccessLogAlertService {
.disposedState(0)
.dispositionAdvice("研判后处置")
.dnsInfo(alarmResult.getString("host"))
.reason(alarmResult.getString("reason"))
.dtype(alarmResult.getString("dtype"))
.originField(alarmResult.getString("origin_field"))
.originLog(alarmResult.getString("origin_log"))
.build();
//补充返回结果的原始日志字段
@@ -419,6 +442,9 @@ public class AccessLogAlertService {
log.debug("算法:{},ID:{} ,AlarmNme:{} 没有返回 origin_log节点.",AlgorithmName, alarmVisit.getId(), alarmVisit.getAlarmName());
return false;
}
// 存储原始日志JSON字符串到 origin_log 字段
alarmVisit.setOriginLog(originLogObject.toJSONString());
// _source.sport/dport 在 JSON 中为字符串类型,需要用 getString() 读取后 parseInt
String sportStr = originLogObject.getString("_source.sport");
if (sportStr != null && !sportStr.isEmpty()) {
@@ -461,6 +487,28 @@ public class AccessLogAlertService {
return deviceList.get(0).getId();
}
/**
* 生成批次号:yyyyMMdd_NNNN(时间戳字符串 + 当天序号)
* 如:20260605_0001
*/
private String generateBatchNo() {
String todayPrefix = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
String maxBatchNo = alarmVisitProcesslogMapper.getTodayMaxBatchNo(todayPrefix);
int seq = 1;
if (maxBatchNo != null && !maxBatchNo.isEmpty()) {
// batch_no 格式: yyyyMMdd_XXXX,解析序号部分
int underscoreIdx = maxBatchNo.lastIndexOf('_');
if (underscoreIdx >= 0 && underscoreIdx < maxBatchNo.length() - 1) {
try {
seq = Integer.parseInt(maxBatchNo.substring(underscoreIdx + 1)) + 1;
} catch (NumberFormatException e) {
log.warn("解析批次序号失败: {}, 使用默认序号1", maxBatchNo);
}
}
}
return todayPrefix + "_" + String.format("%04d", seq);
}
@@ -194,6 +194,10 @@ public class DataTransformer {
.httpReqBody(groupedData.getHttpReqBodys())
.httpRespHeader(groupedData.getHttpRespHeaders())
.httpRespBody(groupedData.getHttpRespBodys())
.reason(null)
.dtype(null)
.originField(null)
.originLog(null)
.build();
} catch (Exception e) {
@@ -7,11 +7,12 @@ import com.common.mapper.DeviceReceiveLogMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.CollectionUtils;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -24,11 +25,11 @@ public class DeviceCollectTaskUpdateService {
private final DeviceReceiveLogMapper deviceReceiveLogMapper;
private final DeviceCollectTaskMapper deviceCollectTaskMapper;
private final TransactionTemplate transactionTemplate;
/**
* 主处理方法:更新所有设备任务的时间信息
*/
@Transactional(rollbackFor = Exception.class)
public void updateAllDeviceTaskTimes() {
log.info("开始更新设备采集任务时间,当前时间: {}", LocalDateTime.now());
@@ -97,30 +98,29 @@ public class DeviceCollectTaskUpdateService {
/**
* 批量更新版本(性能更好)
* 注意:不使用 @Transactional 避免长时间持锁,读操作无需事务,
* 仅批写通过 TransactionTemplate 精确控制事务范围
*/
@Transactional(rollbackFor = Exception.class)
public void batchUpdateDeviceTaskTimes() {
log.info("开始批量更新设备采集任务时间,当前时间: {}", LocalDateTime.now());
// 1. 查询所有设备任务
// 1. 查询所有设备任务(只读,无需事务)
List<DeviceCollectTask> allTasks = deviceCollectTaskMapper.selectAllTasks();
if (CollectionUtils.isEmpty(allTasks)) {
return;
}
// 2. 查询时间信息
// 2. 查询时间信息(只读,无需事务)
Map<String, DeviceCollectTaskTime> successTimes = getDeviceSuccessTimes();
Map<String, DeviceCollectTaskTime> failTimes = getDeviceFailTimes();
// 3. 批量更新
// 3. 内存中组装更新数据
LocalDateTime updateTime = LocalDateTime.now();
for (DeviceCollectTask task : allTasks) {
String deviceCollectId = task.getId().toString();
// 获取成功和失败时间
DeviceCollectTaskTime successTime = successTimes.get(deviceCollectId);
DeviceCollectTaskTime failTime = failTimes.get(deviceCollectId);
DeviceCollectTaskTime successTime = successTimes.get(task.getId().toString());
DeviceCollectTaskTime failTime = failTimes.get(task.getId().toString());
// 更新成功时间
if (successTime != null) {
@@ -143,22 +143,32 @@ public class DeviceCollectTaskUpdateService {
task.setUpdatedAt(updateTime.atOffset(ZoneOffset.UTC));
}
// 4. 批量更新到数据库
int updatedCount = deviceCollectTaskMapper.batchUpdateTimes(allTasks);
// 4. 批量更新到数据库(仅写操作加事务,用完即释放连接)
int updatedCount = transactionTemplate.execute(status ->
deviceCollectTaskMapper.batchUpdateTimes(allTasks));
log.info("批量更新完成,总计: {},已更新: {}", allTasks.size(), updatedCount);
}
/**
* 获取设备的成功时间信息
* 优化:first_time 为恒定值,只对 first_time IS NULL 的新任务查询首次成功时间,
* 避免对 device_receive_log进行全表扫描
*/
private Map<String, DeviceCollectTaskTime> getDeviceSuccessTimes() {
// 查询当天的最新成功时间
// 1. 查询当天的最新成功时间
List<DeviceCollectTaskTime> dailySuccess = deviceReceiveLogMapper.selectDailySuccessTimes();
// 查询首次成功时间(历史)
List<DeviceCollectTaskTime> firstSuccess = deviceReceiveLogMapper.selectFirstSuccessTimes();
// 2. 仅查询 first_time 为空的任务 ID,避免全表扫描
List<Integer> tasksWithoutFirstTime = deviceCollectTaskMapper.selectTasksWithoutFirstTime();
List<DeviceCollectTaskTime> firstSuccess;
if (!CollectionUtils.isEmpty(tasksWithoutFirstTime)) {
log.info("发现 {} 个任务尚未设置首次成功时间,进行精确查询", tasksWithoutFirstTime.size());
firstSuccess = deviceReceiveLogMapper.selectFirstSuccessTimesByIds(tasksWithoutFirstTime);
} else {
firstSuccess = Collections.emptyList();
}
// 合并结果
// 3. 合并结果
Map<String, DeviceCollectTaskTime> result = new HashMap<>();
// 处理首次成功时间
@@ -205,7 +215,6 @@ public class DeviceCollectTaskUpdateService {
/**
* 增量更新:只处理最近一段时间的数据
*/
@Transactional(rollbackFor = Exception.class)
public void incrementalUpdate(int hours) {
LocalDateTime endTime = LocalDateTime.now();
LocalDateTime startTime = endTime.minusHours(hours);
@@ -8,7 +8,7 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -34,6 +34,9 @@ public class DeviceStatsUpdateService {
@Autowired
private JdbcTemplate jdbcTemplate;
@Autowired
private TransactionTemplate transactionTemplate;
// SQL 模板定义
private static final String COLLECT_DEVICE_STATS_SQL =
"WITH device_receive_stats AS ( " +
@@ -86,9 +89,10 @@ public class DeviceStatsUpdateService {
/**
* 每5分钟执行一次设备统计更新(秒:0,分:*,时:*)
* 注意:不使用 @Transactional 避免长时间持锁,统计查询为只读无需事务,
* 写操作通过 TransactionTemplate 精确控制事务范围
*/
@Scheduled(cron = "0 */5 * * * ?")
@Transactional
public void updateDeviceStats() {
long startTime = System.currentTimeMillis();
log.info("开始执行设备统计更新任务...");
@@ -97,22 +101,23 @@ public class DeviceStatsUpdateService {
// 获取当前日期,用于动态表名
String today = LocalDateTime.now().format(DATE_FORMATTER);
// 统计每个设备的当天数据
// 1. 统计每个设备的当天数据(只读,无需事务)
List<DeviceStatsDTO> deviceStats = collectDeviceStats(today);
// 批量更新 device_device 表
batchUpdateDeviceStats(deviceStats);
// 2. 批量更新 device_device 表(通过事务模板控制事务边界,用完即释放连接)
transactionTemplate.executeWithoutResult(status -> {
batchUpdateDeviceStats(deviceStats);
});
long endTime = System.currentTimeMillis();
log.info("设备统计更新完成,处理设备数:{},耗时:{}ms",
deviceStats.size(), (endTime - startTime));
// 批量更新 device_collect_task 表
// 3. 批量更新 device_collect_task 表(独立事务,由该 Service 自己管理)
updateDeviceCollectTaskTimes();
} catch (Exception e) {
log.error("设备统计更新任务执行失败", e);
throw e;
}
}
@@ -120,18 +125,16 @@ public class DeviceStatsUpdateService {
* 每天执行一次设备统计数值清零更新(秒:1,分:0,时:0)
*/
@Scheduled(cron = "1 0 0 * * ?")
@Transactional
public void updateDeviceCount() {
long startTime = System.currentTimeMillis();
log.info("开始执行设备统计数值清零更新任务...");
try {
int devCount= jdbcTemplate.update(UPDATE_DEVICE_COUNT_SQL);
int devCount = transactionTemplate.execute(status ->
jdbcTemplate.update(UPDATE_DEVICE_COUNT_SQL));
long endTime = System.currentTimeMillis();
log.info("设备统计数值清零更新完成,处理设备数:{} ,耗时:{}ms",devCount, (endTime - startTime));
log.info("设备统计数值清零更新完成,处理设备数:{} ,耗时:{}ms", devCount, (endTime - startTime));
} catch (Exception e) {
log.error("设备统计数值清零更新任务执行失败", e);
throw e;
}
}
@@ -20,11 +20,8 @@ import java.util.Map;
@Service
public class InfluxSyslogService {
//@Autowired
//private InfluxDBClient influxDBClient;
// 初始化 InfluxDB 客户端
//@Autowired
private final com.influx.InfluxDBClient influxDBClient = new InfluxDBClient();
@Autowired
private com.influx.InfluxDBClient influxDBClient;
private static final DateTimeFormatter TABLE_DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd");
private static final DateTimeFormatter INPUT_DATE_FORMATTER = DateTimeFormatter.ISO_DATE_TIME;
@@ -177,6 +177,58 @@ public class PartitionTableService {
}
}
// ============================================================
// 达梦(DM)存储过程方式创建分区表 — 下月预创建
// 存储过程: CREATE_TABLE_DAILY_PARTITIONS(table_name, startDate, endDate)
// ============================================================
/** 需要创建月度分区的主表列表 */
private static final String[] MONTHLY_PARTITION_TABLES = {
"alarm",
"alarm_visit",
"syslog_normal_data",
"syslog_normal_alarm",
"syslog_non_normal_message"
};
/**
* 通过达梦存储过程创建下个月所有分区表(5个主表)
* 存储过程内部已处理"分区已存在"异常,可安全重复调用
*/
public void createNextMonthPartitionsByProcedure() {
LocalDate nextMonthFirstDay = LocalDate.now()
.plusMonths(1)
.withDayOfMonth(1);
LocalDate nextMonthLastDay = nextMonthFirstDay
.with(TemporalAdjusters.lastDayOfMonth());
String startDate = nextMonthFirstDay.format(SQL_DATE_FORMATTER);
String endDate = nextMonthLastDay.format(SQL_DATE_FORMATTER);
logger.info("开始通过存储过程创建下月分区表: {} ~ {}", startDate, endDate);
int successCount = 0;
int failCount = 0;
for (String tableName : MONTHLY_PARTITION_TABLES) {
try {
String sql = String.format(
"CALL CREATE_TABLE_DAILY_PARTITIONS('%s', '%s', '%s')",
tableName, startDate, endDate);
jdbcTemplate.execute(sql);
successCount++;
logger.info("存储过程执行成功: 表={}, 日期范围={} ~ {}", tableName, startDate, endDate);
} catch (Exception e) {
failCount++;
logger.error("存储过程执行失败: 表={}, 日期范围={} ~ {}, 错误={}",
tableName, startDate, endDate, e.getMessage(), e);
}
}
logger.info("下月分区表创建完成: 成功={}, 失败={}, 总计={}",
successCount, failCount, MONTHLY_PARTITION_TABLES.length);
}
/**
* 获取下个月需要创建的分区表数量
*/
@@ -135,7 +135,7 @@ public class DeviceReceiveLogServiceImpl implements DeviceReceiveLogService {
}
LocalDateTime endTime = LocalDateTime.now().minusDays(7); // 默认保留7
LocalDateTime endTime = LocalDateTime.now().minusDays(3); // 默认保留3
if (beforeTime.isAfter(endTime)) {
log.warn("删除时间点{}晚于默认保留时间{},使用默认时间", beforeTime, endTime);
beforeTime = endTime;
@@ -15,10 +15,17 @@ import java.time.format.DateTimeFormatter;
import java.util.*;
/**
* 实时分析引擎实现
* 实时分析引擎实现(平铺SQL模式)
*
* 当 analysis.db.dameng-subquery-mode=true 时,本Bean不会被注册,
* 改为使用 RealtimeAnalysisEngineDm(子查询包装模式)适配达梦安全版。
*/
@Slf4j
@Service("realtimeAnalysisEngine")
@org.springframework.boot.autoconfigure.condition.ConditionalOnProperty(
name = "analysis.db.dameng-subquery-mode",
havingValue = "false",
matchIfMissing = true)
public class RealtimeAnalysisEngine implements AnalysisEngine {
@Autowired
@@ -0,0 +1,817 @@
package com.common.service.impl;
import com.common.entity.*;
import com.common.mapper.*;
import com.common.service.AnalysisEngine;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
/**
* 实时分析引擎实现(达梦安全版适配)
*
* 与 RealtimeAnalysisEngine 功能完全相同,唯一差异:
* 注入 SqlGeneratorServiceImplDm(子查询包装模式SQL生成器),
* 以适配达梦安全版数据库 GROUP BY 不支持自定义函数(TUMBLE/HOP/SESSION)的问题。
*
* 通过 @ConditionalOnProperty 与 RealtimeAnalysisEngine 二选一注册,
* 由配置 analysis.db.dameng-subquery-mode=true 激活本实现。
*
* @see RealtimeAnalysisEngine 原始实现(平铺SQL模式)
* @see SqlGeneratorServiceImplDm 子查询包装SQL生成器
*/
@Slf4j
@Service("realtimeAnalysisEngine")
@org.springframework.boot.autoconfigure.condition.ConditionalOnProperty(
name = "analysis.db.dameng-subquery-mode",
havingValue = "true")
public class RealtimeAnalysisEngineDm implements AnalysisEngine {
// 直接实例化(非Spring Bean),避免与父类SqlGeneratorServiceImpl产生类型冲突
private final SqlGeneratorServiceImplDm sqlGeneratorService = new SqlGeneratorServiceImplDm();
@Autowired
private AnalysisAnalysisRuleMapper ruleMapper;
@Autowired
private AnalysisFieldMapper fieldMapper;
@Autowired
private AnalysisWhereConditionMapper whereConditionMapper;
@Autowired
private AnalysisGroupByColumnMapper groupByColumnMapper;
@Autowired
private AnalysisGroupByWindowMapper groupByWindowMapper;
@Autowired
private AnalysisFilterMapper filterMapper;
@Autowired
private AnalysisGroupByHavingMapper groupByHavingMapper;
@Autowired
private AnalysisTaskHistoryMapper taskHistoryMapper;
@Autowired
private AlarmMapper alarmMapper;
@Autowired
private AnalysisGroupByMapper groupByMapper;
@Autowired
private JdbcTemplate jdbcTemplate;
private static final String RUN_MODE = "realtime";
private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
@Override
public Map<String, Object> executeRule(AnalysisAnalysisRule rule) {
String batchNo = generateBatchNo();
LocalDateTime startTime = LocalDateTime.now().withSecond(0).withNano(0);
LocalDateTime dataEndTime = startTime;
// 加载分组和窗口配置(在计算时间范围之前)
AnalysisGroupByWindow groupByWindow = null;
List<AnalysisGroupBy> groupByList = groupByMapper.selectByRuleId(rule.getRuleId());
if (groupByList != null && !groupByList.isEmpty()) {
AnalysisGroupBy groupBy = groupByList.get(0);
if (groupBy.getId() != null) {
groupByWindow = groupByWindowMapper.selectByGroupById(groupBy.getId().intValue());
}
}
// 根据窗口类型动态计算数据查询时间范围
LocalDateTime dataStartTime = calculateDataStartTime(dataEndTime, groupByWindow);
log.info("开始执行实时规则(DM子查询模式): ruleId={}, ruleName={}, batchNo={}, windowType={}, dataStartTime={}, dataEndTime={}",
rule.getRuleId(), rule.getRuleName(), batchNo,
groupByWindow != null ? groupByWindow.getWindowType() : "NONE",
dataStartTime.format(DATE_FORMATTER),
dataEndTime.format(DATE_FORMATTER));
// 创建任务历史记录
AnalysisTaskHistory history = AnalysisTaskHistory.builder()
.id(System.currentTimeMillis())
.ruleId(rule.getRuleId())
.startTime(startTime)
.status("RUNNING")
.progressPercent(0)
.inputCount(0L)
.outputCount(0L)
.delFlag("0")
.createTime(startTime)
.updateTime(startTime)
.tenantId("000000")
.remark("实时分析任务(达梦适配) - " + batchNo)
.build();
taskHistoryMapper.insert(history);
Map<String, Object> result = new HashMap<>();
result.put("ruleId", rule.getRuleId());
result.put("ruleName", rule.getRuleName());
result.put("runMode", RUN_MODE);
result.put("batchNo", batchNo);
try {
// 更新规则状态为运行中
ruleMapper.updateTaskStatus(rule.getRuleId(), "running", 1L);
// 加载规则配置
List<AnalysisField> fields = fieldMapper.selectByRuleId(rule.getRuleId());
List<AnalysisWhereCondition> whereConditions = whereConditionMapper.selectByRuleId(rule.getRuleId());
List<AnalysisFilter> filters = filterMapper.selectByRuleId(rule.getRuleId());
List<AnalysisGroupByColumn> groupByColumns = groupByColumnMapper.selectByRuleId(rule.getRuleId());
List<AnalysisGroupByHaving> havingConditions = groupByHavingMapper.selectByRuleId(rule.getRuleId());
// 生成SQL(调用达梦适配版SQL生成器,子查询包装模式)
String sql = sqlGeneratorService.generateSql(
rule,
fields,
whereConditions,
filters,
groupByColumns,
havingConditions,
groupByWindow,
dataStartTime.format(DATE_FORMATTER),
dataEndTime.format(DATE_FORMATTER)
);
log.info("生成的SQL(达梦子查询模式): {}", sql);
// 执行SQL
List<Map<String, Object>> queryResult = jdbcTemplate.queryForList(sql);
result.put("queryResult", queryResult);
// 处理结果,生成告警
long alarmCount = 0;
if (!queryResult.isEmpty()) {
List<Alarm> alarms = convertToAlarms(rule, queryResult);
if (!alarms.isEmpty()) {
String tableName = "alarm";
alarmMapper.batchInsert(alarms);
alarmCount = alarms.size();
}
}
// 更新任务历史
LocalDateTime endTime = LocalDateTime.now();
long durationSeconds = java.time.Duration.between(startTime, endTime).getSeconds();
history.setEndTime(endTime);
history.setDurationTime(durationSeconds);
history.setProgressPercent(100);
history.setInputCount((long) queryResult.size());
history.setOutputCount(alarmCount);
history.setStatus("COMPLETED");
history.setUpdateTime(endTime);
taskHistoryMapper.update(history);
result.put("processedCount", queryResult.size());
result.put("alarmCount", alarmCount);
result.put("status", "success");
// 更新规则状态为等待
ruleMapper.updateTaskStatus(rule.getRuleId(), "waiting", 1L);
log.info("规则执行成功(达梦子查询模式): ruleId={}, processedCount={}, alarmCount={}",
rule.getRuleId(), queryResult.size(), alarmCount);
} catch (Exception e) {
log.error("规则执行失败: ruleId={}", rule.getRuleId(), e);
// 更新任务历史
LocalDateTime endTime = LocalDateTime.now();
long durationSeconds = java.time.Duration.between(startTime, endTime).getSeconds();
history.setEndTime(endTime);
history.setDurationTime(durationSeconds);
history.setStatus("FAILED");
history.setRemark("执行失败: " + (e.getMessage().length() > 480 ? e.getMessage().substring(0, 480) : e.getMessage()));
history.setUpdateTime(endTime);
taskHistoryMapper.update(history);
// 更新规则状态
ruleMapper.updateTaskStatus(rule.getRuleId(), "stopped", 1L);
result.put("status", "failed");
result.put("errorMsg", e.getMessage());
}
return result;
}
@Override
public List<Map<String, Object>> executeRules(List<AnalysisAnalysisRule> rules) {
List<Map<String, Object>> results = new ArrayList<>();
for (AnalysisAnalysisRule rule : rules) {
try {
Map<String, Object> result = executeRule(rule);
results.add(result);
} catch (Exception e) {
log.error("执行规则失败: ruleId={}", rule.getRuleId(), e);
Map<String, Object> errorResult = new HashMap<>();
errorResult.put("ruleId", rule.getRuleId());
errorResult.put("ruleName", rule.getRuleName());
errorResult.put("status", "failed");
errorResult.put("errorMsg", e.getMessage());
results.add(errorResult);
}
}
return results;
}
@Override
public void stopRule(String ruleId) {
try {
ruleMapper.updateTaskStatus(ruleId, "stopped", 1L);
log.info("已停止规则: ruleId={}", ruleId);
} catch (Exception e) {
log.error("停止规则失败: ruleId={}", ruleId, e);
}
}
@Override
public String getRunMode() {
return RUN_MODE;
}
/**
* 生成批次号
*/
private String generateBatchNo() {
return LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS"));
}
/**
* 根据窗口类型动态计算数据查询开始时间
*/
private LocalDateTime calculateDataStartTime(LocalDateTime dataEndTime, AnalysisGroupByWindow groupByWindow) {
if (groupByWindow == null) {
log.warn("未配置窗口类型,使用默认查询范围:最近30分钟");
return dataEndTime.minusMinutes(30);
}
String windowType = groupByWindow.getWindowType();
if (windowType == null || windowType.trim().isEmpty()) {
log.warn("窗口类型为空,使用默认查询范围:最近30分钟");
return dataEndTime.minusMinutes(30);
}
switch (windowType.toUpperCase()) {
case "TUMBLE":
return calculateTumbleWindowStartTime(dataEndTime, groupByWindow);
case "HOP":
return calculateHopWindowStartTime(dataEndTime, groupByWindow);
case "SESSION":
return calculateSessionWindowStartTime(dataEndTime, groupByWindow);
default:
log.warn("未知窗口类型: {},使用默认查询范围:最近30分钟", windowType);
return dataEndTime.minusMinutes(30);
}
}
/**
* 计算滚动窗口的数据查询开始时间
*/
private LocalDateTime calculateTumbleWindowStartTime(LocalDateTime dataEndTime, AnalysisGroupByWindow groupByWindow) {
Integer windowSize = groupByWindow.getTumbleWindowSize();
String windowSizeUnit = groupByWindow.getTumbleWindowSizeUnit();
if (windowSize == null || windowSize <= 0) {
log.warn("滚动窗口大小配置无效,使用默认值:5分钟");
windowSize = 5;
windowSizeUnit = "m";
}
if (windowSizeUnit == null || windowSizeUnit.trim().isEmpty()) {
windowSizeUnit = "m";
}
LocalDateTime dataStartTime;
switch (windowSizeUnit.toLowerCase()) {
case "s":
dataStartTime = dataEndTime.minusSeconds(windowSize);
break;
case "m":
dataStartTime = dataEndTime.minusMinutes(windowSize);
break;
case "h":
dataStartTime = dataEndTime.minusHours(windowSize);
break;
case "d":
dataStartTime = dataEndTime.minusDays(windowSize);
break;
default:
log.warn("滚动窗口单位无效: {},使用默认单位:分钟", windowSizeUnit);
dataStartTime = dataEndTime.minusMinutes(windowSize);
}
log.info("滚动窗口查询范围: 窗口大小={}{},查询时间范围=[{}, {}]",
windowSize, windowSizeUnit,
dataStartTime.format(DATE_FORMATTER),
dataEndTime.format(DATE_FORMATTER));
return dataStartTime;
}
/**
* 计算滑动窗口的数据查询开始时间
*/
private LocalDateTime calculateHopWindowStartTime(LocalDateTime dataEndTime, AnalysisGroupByWindow groupByWindow) {
Integer windowSize = groupByWindow.getHopWindowSize();
String windowSizeUnit = groupByWindow.getHopWindowSizeUnit();
if (windowSize == null || windowSize <= 0) {
log.warn("滑动窗口大小配置无效,使用默认值:5分钟");
windowSize = 5;
windowSizeUnit = "m";
}
if (windowSizeUnit == null || windowSizeUnit.trim().isEmpty()) {
windowSizeUnit = "m";
}
LocalDateTime dataStartTime;
switch (windowSizeUnit.toLowerCase()) {
case "s":
dataStartTime = dataEndTime.minusSeconds(windowSize);
break;
case "m":
dataStartTime = dataEndTime.minusMinutes(windowSize);
break;
case "h":
dataStartTime = dataEndTime.minusHours(windowSize);
break;
case "d":
dataStartTime = dataEndTime.minusDays(windowSize);
break;
default:
log.warn("滑动窗口单位无效: {},使用默认单位:分钟", windowSizeUnit);
dataStartTime = dataEndTime.minusMinutes(windowSize);
}
log.info("滑动窗口查询范围: 窗口大小={}{},查询时间范围=[{}, {}]",
windowSize, windowSizeUnit,
dataStartTime.format(DATE_FORMATTER),
dataEndTime.format(DATE_FORMATTER));
return dataStartTime;
}
/**
* 计算会话窗口的数据查询开始时间
*/
private LocalDateTime calculateSessionWindowStartTime(LocalDateTime dataEndTime, AnalysisGroupByWindow groupByWindow) {
Integer sessionTimeout = groupByWindow.getSessionWindowSize();
String sessionTimeoutUnit = groupByWindow.getSessionWindowSizeUnit();
if (sessionTimeout == null || sessionTimeout <= 0) {
log.warn("会话窗口超时时间配置无效,使用默认值:30分钟");
sessionTimeout = 30;
sessionTimeoutUnit = "m";
}
if (sessionTimeoutUnit == null || sessionTimeoutUnit.trim().isEmpty()) {
sessionTimeoutUnit = "m";
}
LocalDateTime dataStartTime;
switch (sessionTimeoutUnit.toLowerCase()) {
case "s":
dataStartTime = dataEndTime.minusSeconds(sessionTimeout);
break;
case "m":
dataStartTime = dataEndTime.minusMinutes(sessionTimeout);
break;
case "h":
dataStartTime = dataEndTime.minusHours(sessionTimeout);
break;
case "d":
dataStartTime = dataEndTime.minusDays(sessionTimeout);
break;
default:
log.warn("会话窗口超时单位无效: {},使用默认单位:分钟", sessionTimeoutUnit);
dataStartTime = dataEndTime.minusMinutes(sessionTimeout);
}
// 会话窗口额外增加1天缓冲时间
dataStartTime = dataStartTime.minusDays(1);
log.info("会话窗口查询范围: 超时时间={}{},额外缓冲1天,查询时间范围=[{}, {}]",
sessionTimeout, sessionTimeoutUnit,
dataStartTime.format(DATE_FORMATTER),
dataEndTime.format(DATE_FORMATTER));
return dataStartTime;
}
/**
* 转换查询结果为告警对象
*/
private List<Alarm> convertToAlarms(AnalysisAnalysisRule rule, List<Map<String, Object>> queryResult) {
List<Alarm> alarms = new ArrayList<>();
for (Map<String, Object> row : queryResult) {
Alarm alarm = Alarm.builder()
.id(UUID.randomUUID().toString())
.createdAt(LocalDateTime.now())
.updatedAt(LocalDateTime.now())
.alarmName(rule.getRuleName())
.engineType("realtime")
.attackResult(-1)
.focused(false)
.fall(0)
.alarmLevel("未知")
.baseFocused(false)
.isUpdated(false)
.alarmSource(1)
.dispositionAdvice("研判后处置")
.disposedState(0)
.attackDirection("other")
.etlTime(LocalDateTime.now())
.alarmAreaId(0)
.comment(buildComment(row))
.attackChainPhase(getIntegerArray(-1))
.judgedState(0)
.build();
// 映射查询结果字段到告警对象
if (row.containsKey("log_start_at")) {
alarm.setLogStartAt(getTimestampValue(row.get("log_start_at")));
}
if (row.containsKey("log_end_at")) {
alarm.setLogEndAt(getTimestampValue(row.get("log_end_at")));
}
if (row.containsKey("alarm_name")) {
alarm.setAlarmName(getStringValue(row.get("alarm_name")));
}
if (row.containsKey("alarm_type")) {
alarm.setAlarmType(getStringValue(row.get("alarm_type")));
}
if (row.containsKey("alarm_level")) {
alarm.setAlarmLevel(convertAlarmLevel(getIntegerValue(row.get("alarm_level"))));
}
if (row.containsKey("attack_ip")) {
alarm.setAttackIp(getStringArray(row.get("attack_ip")));
}
if (row.containsKey("victim_ip")) {
alarm.setVictimIp(getStringArray(row.get("victim_ip")));
}
if (row.containsKey("victim_web_url")) {
alarm.setVictimWebUrl(getStringArray(row.get("victim_web_url")));
}
if (row.containsKey("attack_chain_phase")) {
alarm.setAttackChainPhase(getIntegerArray(row.get("attack_chain_phase")));
}
if (row.containsKey("device_id")) {
alarm.setDeviceId(getIntegerArray(row.get("device_id")));
}
if (row.containsKey("tag")) {
alarm.setTag(getStringArray(row.get("tag")));
}
if (row.containsKey("comment")) {
alarm.setComment(getStringValue(row.get("comment")));
}
if (row.containsKey("origin_log_ids")) {
alarm.setOriginLogIds(getStringArray(row.get("origin_log_ids")));
}
if (row.containsKey("query_id")) {
alarm.setQueryId(getStringValue(row.get("query_id")));
}
if (row.containsKey("attack_result")) {
alarm.setAttackResult(getIntegerValue(row.get("attack_result")));
}
if (row.containsKey("fall")) {
alarm.setFall(getIntegerValue(row.get("fall")));
}
if (row.containsKey("payload")) {
alarm.setPayload(getBytesValue(row.get("payload")));
}
if (row.containsKey("operate_event")) {
alarm.setOperateEvent(getIntegerArray(row.get("operate_event")));
}
if (row.containsKey("attack_port")) {
alarm.setAttackPort(getIntegerArray(row.get("attack_port")));
}
if (row.containsKey("victim_port")) {
alarm.setVictimPort(getIntegerArray(row.get("victim_port")));
}
if (row.containsKey("attack_method")) {
alarm.setAttackMethod(getStringValue(row.get("attack_method")));
}
if (row.containsKey("business_ext")) {
alarm.setBusinessExt(getStringValue(row.get("business_ext")));
}
if (row.containsKey("http_status")) {
alarm.setHttpStatus(getStringValue(row.get("http_status")));
}
if (row.containsKey("dns_info")) {
alarm.setDnsInfo(getStringValue(row.get("dns_info")));
}
if (row.containsKey("account_info")) {
alarm.setAccountInfo(getStringValue(row.get("account_info")));
}
if (row.containsKey("attacker_info")) {
alarm.setAttackerInfo(getStringValue(row.get("attacker_info")));
}
if (row.containsKey("victim_info")) {
alarm.setVictimInfo(getStringValue(row.get("victim_info")));
}
if (row.containsKey("suspicious_action")) {
alarm.setSuspiciousAction(getStringValue(row.get("suspicious_action")));
}
if (row.containsKey("vuln_info")) {
alarm.setVulnInfo(getStringValue(row.get("vuln_info")));
}
if (row.containsKey("weak_pwd")) {
alarm.setWeakPwd(getStringValue(row.get("weak_pwd")));
}
if (row.containsKey("compliance_baseline")) {
alarm.setComplianceBaseline(getStringValue(row.get("compliance_baseline")));
}
if (row.containsKey("file_info")) {
alarm.setFileInfo(getStringValue(row.get("file_info")));
}
if (row.containsKey("file_tags")) {
alarm.setFileTags(getStringValue(row.get("file_tags")));
}
if (row.containsKey("endpoint_info")) {
alarm.setEndpointInfo(getStringValue(row.get("endpoint_info")));
}
if (row.containsKey("origin_info")) {
alarm.setOriginInfo(getStringValue(row.get("origin_info")));
}
if (row.containsKey("protocol_info")) {
alarm.setProtocolInfo(getStringValue(row.get("protocol_info")));
}
if (row.containsKey("email_info")) {
alarm.setEmailInfo(getStringValue(row.get("email_info")));
}
if (row.containsKey("sensitive_data")) {
alarm.setSensitiveData(getStringValue(row.get("sensitive_data")));
}
if (row.containsKey("hit_intelligence")) {
alarm.setHitIntelligence(getIntegerValue(row.get("hit_intelligence")));
}
if (row.containsKey("window_time")) {
alarm.setWindowTime(getStringValue(row.get("window_time")));
}
if (row.containsKey("attack_ip_pic")) {
alarm.setAttackIpPic(getStringValue(row.get("attack_ip_pic")));
}
if (row.containsKey("victim_ip_pic")) {
alarm.setVictimIpPic(getStringValue(row.get("victim_ip_pic")));
}
if (row.containsKey("operation_at")) {
alarm.setOperationAt(getTimestampValue(row.get("operation_at")));
}
if (row.containsKey("attack_direction")) {
alarm.setAttackDirection(getStringValue(row.get("attack_direction")));
}
if (row.containsKey("etl_time")) {
alarm.setEtlTime(getTimestampValue(row.get("etl_time")));
}
if (row.containsKey("log_count")) {
alarm.setLogCount(getIntegerValue(row.get("log_count")));
}
if (row.containsKey("is_asset_hit")) {
alarm.setIsAssetHit(getIntegerValue(row.get("is_asset_hit")));
}
if (row.containsKey("http_req_header")) {
alarm.setHttpReqHeader(getStringArray(row.get("http_req_header")));
}
if (row.containsKey("http_req_body")) {
alarm.setHttpReqBody(getStringArray(row.get("http_req_body")));
}
if (row.containsKey("http_resp_header")) {
alarm.setHttpRespHeader(getStringArray(row.get("http_resp_header")));
}
if (row.containsKey("http_resp_body")) {
alarm.setHttpRespBody(getStringArray(row.get("http_resp_body")));
}
if (row.containsKey("window_time")) {
alarm.setLogEndAt(getTimestampValue(row.get("window_time")));
}
alarms.add(alarm);
}
return alarms;
}
private byte[] getBytesValue(Object value) {
if (value == null) {
return null;
}
if (value instanceof byte[]) {
return (byte[]) value;
}
return value.toString().getBytes();
}
private String getStringValue(Object value) {
if (value == null) {
return null;
}
return value.toString();
}
private Long getLongValue(Object value) {
if (value == null) {
return 0L;
}
if (value instanceof Number) {
return ((Number) value).longValue();
}
return Long.parseLong(value.toString());
}
private Integer getIntegerValue(Object value) {
if (value == null) {
return -1;
}
if (value instanceof Number) {
return ((Number) value).intValue();
}
return Integer.parseInt(value.toString());
}
private LocalDateTime getTimestampValue(Object value) {
if (value == null) {
return null;
}
if (value instanceof LocalDateTime) {
return (LocalDateTime) value;
}
String strValue = value.toString().trim();
String[] patterns = {
"yyyy-MM-dd HH:mm:ss.SSS",
"yyyy-MM-dd HH:mm:ss",
"yyyy-MM-dd'T'HH:mm:ss.SSS",
"yyyy-MM-dd'T'HH:mm:ss",
"yyyy-MM-dd HH:mm:ss.SSSSSS",
"yyyy-MM-dd HH:mm:ss.SSSSS",
"yyyy-MM-dd HH:mm:ss.S",
"yyyy-MM-dd HH:mm:ss.SS",
"yyyy-MM-dd'T'HH:mm:ss.SSSSSSS",
"yyyy-MM-dd",
"yyyy/MM/dd HH:mm:ss",
"yyyy/MM/dd HH:mm:ss.SSS"
};
for (String pattern : patterns) {
try {
return LocalDateTime.parse(strValue, java.time.format.DateTimeFormatter.ofPattern(pattern));
} catch (Exception ignored) {
}
}
try {
return LocalDateTime.parse(strValue);
} catch (Exception e) {
log.warn("无法解析时间字符串: {}", strValue);
return null;
}
}
@SuppressWarnings("unchecked")
private String[] getStringArray(Object value) {
if (value == null) {
return new String[0];
}
if (value instanceof String[]) {
return (String[]) value;
}
if (value instanceof Object[]) {
Object[] arr = (Object[]) value;
String[] result = new String[arr.length];
for (int i = 0; i < arr.length; i++) {
result[i] = arr[i] != null ? arr[i].toString() : null;
}
return result;
}
String str = value.toString();
if (str.startsWith("[") && str.endsWith("]")) {
str = str.substring(1, str.length() - 1).trim();
if (str.isEmpty()) {
return new String[0];
}
String[] parts = str.split(",");
String[] result = new String[parts.length];
for (int i = 0; i < parts.length; i++) {
String part = parts[i].trim();
if (part.startsWith("\"") && part.endsWith("\"")) {
part = part.substring(1, part.length() - 1);
}
result[i] = part;
}
return result;
}
if (str.startsWith("{") && str.endsWith("}")) {
str = str.substring(1, str.length() - 1);
return str.split(",");
}
return new String[]{str};
}
@SuppressWarnings("unchecked")
private Integer[] getIntegerArray(Object value) {
if (value == null) {
return new Integer[0];
}
String[] strArray = getStringArray(value);
Integer[] result = new Integer[strArray.length];
for (int i = 0; i < strArray.length; i++) {
try {
result[i] = Integer.parseInt(strArray[i]);
} catch (NumberFormatException e) {
result[i] = null;
}
}
return result;
}
@SuppressWarnings("unchecked")
private byte[][] getByteArrayArray(Object value) {
if (value == null) {
return new byte[0][];
}
if (value instanceof byte[][]) {
return (byte[][]) value;
}
if (value instanceof Object[]) {
Object[] arr = (Object[]) value;
byte[][] result = new byte[arr.length][];
for (int i = 0; i < arr.length; i++) {
if (arr[i] instanceof byte[]) {
result[i] = (byte[]) arr[i];
} else {
result[i] = arr[i] != null ? arr[i].toString().getBytes() : null;
}
}
return result;
}
return new byte[0][];
}
private String convertAlarmLevel(Integer eventLevel) {
if (eventLevel == null) return "未知";
switch (eventLevel) {
case 0: return "安全(无威胁)";
case 1: return "低危";
case 2: return "中危";
case 3: return "高危";
case 4: return "超危";
default: return "未知";
}
}
private String buildComment(Map<String, Object> row) {
String victimIpsStr;
if (row.containsKey("victim_ip")) {
victimIpsStr = String.join(",", getStringArray(row.get("victim_ip")));
} else {
victimIpsStr = "未知";
}
String alarmName = "";
if (row.containsKey("alarm_name")) {
alarmName = getStringValue(row.get("alarm_name"));
}
String AttackIps = "";
if (row.containsKey("attack_ip")) {
AttackIps = String.join(",", getStringArray(row.get("attack_ip")));
}
return String.format(
"24小时内,检测到%s上产生%s告警:\n告警名称:%s\n攻击IP%s\n攻击结果:%d",
victimIpsStr,
convertAlarmLevel(getIntegerValue(row.get("alarm_level"))),
alarmName,
AttackIps,
determineAttackResult(row)
);
}
private String convertAttackIps(String[] attackIps) {
if (attackIps == null || attackIps.length == 0) {
return null;
}
return String.join(",", attackIps);
}
private Integer determineAttackResult(Map<String, Object> row) {
if (row.containsKey("attack_result")) {
return getIntegerValue(row.get("attack_result"));
} else {
return -1;
}
}
}
@@ -395,8 +395,9 @@ public class SqlGeneratorServiceImpl implements SqlGeneratorService {
/**
* 构建时间窗口表达式(用于SELECT)
* 改为protected以便子类复用(达梦安全版子查询包装模式)
*/
private String buildTimeWindowExpression(AnalysisGroupByWindow groupByWindow) {
protected String buildTimeWindowExpression(AnalysisGroupByWindow groupByWindow) {
if (groupByWindow == null) {
return "";
}
@@ -0,0 +1,200 @@
package com.common.service.impl;
import com.common.entity.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import java.util.*;
/**
* 达梦安全版 SQL生成服务实现(子查询包装模式)
*
* 继承自 SqlGeneratorServiceImpl,重写 generateSql() 方法。
* 当配置了时间窗口(TUMBLE/HOP/SESSION)时,采用子查询包装方式生成SQL,
* 将时间窗口函数 TUMBLE()/HOP()/SESSION() 的计算下沉到内层子查询中,
* 外层 GROUP BY 直接引用别名 window_time,规避达梦安全版 GROUP BY 不支持自定义函数的问题。
*
* 注意:本类不注册为 Spring Bean(不加 @Service),由 RealtimeAnalysisEngineDm 直接实例化,
* 避免与父类 SqlGeneratorServiceImpl 产生类型冲突影响 OfflineAnalysisEngine。
*
* 产出示例:
* SELECT ...聚合函数..., window_time
* FROM (
* SELECT 原始列..., TUMBLE(log_time, INTERVAL '5 MINUTE') AS window_time
* FROM syslog_normal_alarm AS t
* WHERE ...
* )
* GROUP BY src_ip, dest_ip, window_time
*/
@Slf4j
public class SqlGeneratorServiceImplDm extends SqlGeneratorServiceImpl {
/**
* 重写:根据规则动态生成SQL
* 有窗口配置 → 子查询包装模式
* 无窗口配置 → 委托父类原有逻辑
*/
@Override
public String generateSql(AnalysisAnalysisRule rule,
List<AnalysisField> fields,
List<AnalysisWhereCondition> whereConditions,
List<AnalysisFilter> filters,
List<AnalysisGroupByColumn> groupByColumns,
List<AnalysisGroupByHaving> havingConditions,
AnalysisGroupByWindow groupByWindow,
String startTime,
String endTime) {
// 如果有时间窗口配置 → 走子查询包装模式(达梦安全版兼容方案)
if (groupByWindow != null && StringUtils.isNotBlank(groupByWindow.getWindowType())) {
return generateSubquerySql(fields, filters, groupByColumns,
havingConditions, groupByWindow, startTime, endTime);
}
// 无窗口配置 → 委托父类原有逻辑(平铺SQL)
return super.generateSql(rule, fields, whereConditions, filters,
groupByColumns, havingConditions, groupByWindow, startTime, endTime);
}
// ==================== 子查询包装模式核心方法 ====================
/**
* 生成子查询包装的完整SQL(达梦安全版兼容方案)
*
* SQL结构:
* SELECT [外层聚合列...], window_time
* FROM (
* SELECT [内层原始列...], TUMBLE(log_time, ...) AS window_time
* FROM syslog_normal_alarm AS t
* WHERE ...
* )
* GROUP BY [分组列...], window_time
* HAVING ...
*/
private String generateSubquerySql(List<AnalysisField> fields,
List<AnalysisFilter> filters,
List<AnalysisGroupByColumn> groupByColumns,
List<AnalysisGroupByHaving> havingConditions,
AnalysisGroupByWindow groupByWindow,
String startTime,
String endTime) {
// 1. 内层子查询SELECT:原始列 + TUMBLE(...) AS window_time
String innerSelect = buildInnerSelectForWindow(fields, groupByColumns, groupByWindow);
// 2. FROM + WHERE(在内层子查询中)
String fromClause = generateFromClause(fields, null);
String whereClause = generateWhereClauseFromFilters(filters, startTime, endTime);
// 3. 外层SELECT:聚合列 + window_time
String outerSelect = buildOuterSelectForWindow(fields);
// 4. 外层GROUP BY:分组列 + window_time(别名,非函数调用)
String groupByClause = buildGroupByForWindow(groupByColumns);
// 5. HAVING(不变,复用父类)
String havingClause = generateHavingClauseFromConditions(havingConditions);
// 6. 拼接最终SQL
return assembleSubquerySql(outerSelect, innerSelect, fromClause,
whereClause, groupByClause, havingClause);
}
/**
* 构建内层子查询的SELECT子句
* 产出:
* SELECT src_ip, dest_ip, origin_event_name, ..., TUMBLE(log_time, INTERVAL '5 MINUTE') AS window_time
*/
private String buildInnerSelectForWindow(List<AnalysisField> fields,
List<AnalysisGroupByColumn> groupByColumns,
AnalysisGroupByWindow groupByWindow) {
Set<String> columns = new LinkedHashSet<>();
// 1. 从 fields 中提取所有原始列名(不含聚合函数)
if (fields != null) {
for (AnalysisField field : fields) {
if (StringUtils.isNotBlank(field.getColumnName())) {
columns.add(field.getColumnName());
}
}
}
// 2. 从 groupByColumns 中提取原始列名
if (groupByColumns != null) {
for (AnalysisGroupByColumn col : groupByColumns) {
if (StringUtils.isNotBlank(col.getColumnName())) {
columns.add(col.getColumnName());
}
}
}
// 3. 追加时间窗口表达式(带 AS window_time 别名)
// 复用父类 protected 方法 buildTimeWindowExpression
String windowExpr = buildTimeWindowExpression(groupByWindow);
List<String> parts = new ArrayList<>(columns);
if (StringUtils.isNotBlank(windowExpr)) {
parts.add(windowExpr);
}
return "SELECT " + String.join(",\n ", parts);
}
/**
* 构建外层SELECT子句(聚合层)
* 产出:
* SELECT JSONB_AGG(DISTINCT src_port) AS attack_port, ..., window_time
*/
private String buildOuterSelectForWindow(List<AnalysisField> fields) {
// 复用父类 generateSelectClause(fields, null)
// 传 null 窗口 → 不追加 TUMBLE(...) AS window_time,仅生成聚合函数列
String baseSelect = generateSelectClause(fields, null);
// 手动追加 window_time(此时是普通列引用,不是函数调用)
return baseSelect + ",\n window_time";
}
/**
* 构建外层GROUP BY子句
* 产出:
* GROUP BY origin_event_name, src_ip, dest_ip, window_time
*
* 关键差异:使用别名 window_time,而非 TUMBLE(log_time, ...) 函数调用
*/
private String buildGroupByForWindow(List<AnalysisGroupByColumn> groupByColumns) {
// 复用父类 generateGroupByClause(columns, null)
// 传 null 窗口 → 不追加 TUMBLE(...) 函数调用,仅生成业务分组列
String baseGroupBy = generateGroupByClause(groupByColumns, null);
if (StringUtils.isBlank(baseGroupBy)) {
return "GROUP BY window_time";
}
// 在现有 GROUP BY 后追加 window_time
return baseGroupBy + ", window_time";
}
/**
* 拼接子查询包装SQL
*/
private String assembleSubquerySql(String outerSelect, String innerSelect,
String fromClause, String whereClause,
String groupByClause, String havingClause) {
StringBuilder sql = new StringBuilder();
// 外层SELECT
sql.append(outerSelect).append("\n");
// 子查询开始
sql.append("FROM (\n");
sql.append(" ").append(innerSelect).append("\n");
sql.append(" ").append(fromClause).append("\n");
if (StringUtils.isNotBlank(whereClause)) {
sql.append(" ").append(whereClause).append("\n");
}
sql.append(")\n");
// 外层GROUP BY
if (StringUtils.isNotBlank(groupByClause)) {
sql.append(groupByClause).append("\n");
}
// HAVING
if (StringUtils.isNotBlank(havingClause)) {
sql.append(havingClause).append("\n");
}
return sql.toString();
}
}
@@ -0,0 +1,70 @@
package com.config;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
/**
* 应用配置属性 - 替代静态 AppConfig
* 使用 Spring Boot 标准 @Value 机制,支持 docker-compose 环境变量覆盖
*/
@Data
@Component
public class AppProperties {
// ========== Syslog 配置 ==========
@Value("${syslog.tcp.port:514}")
private int syslogTcpPort;
@Value("${syslog.udp.port:515}")
private int syslogUdpPort;
@Value("${syslog.max.frame.length:65536}")
private int syslogMaxFrameLength;
@Value("${syslog.buffer.size:1024}")
private int syslogBufferSize;
// ========== InfluxDB 配置 ==========
@Value("${influxdb.url:http://localhost:8086}")
private String influxUrl;
@Value("${influxdb.token:}")
private String influxToken;
@Value("${influxdb.org:}")
private String influxOrg;
@Value("${influxdb.bucket:syslog}")
private String influxBucket;
@Value("${influxdb.batch.size:1000}")
private int influxBatchSize;
@Value("${influxdb.flush.interval:1000}")
private int influxFlushInterval;
@Value("${influxdb.retry.attempts:3}")
private int influxRetryAttempts;
@Value("${influxdb.retry.delay:100}")
private int influxRetryDelay;
// ========== 应用配置 ==========
@Value("${app.worker.threads:4}")
private int workerThreads;
@Value("${app.max.queue.size:2000}")
private int maxQueueSize;
@Value("${app.metrics.enabled:false}")
private boolean metricsEnabled;
// ========== 运行环境 ==========
@Value("${server.run.environment:prod}")
private String runEnvironment;
// ========== SM4 加密配置 ==========
@Value("${syslog.sm4.generateKey:}")
private String sm4Key;
}
@@ -0,0 +1,27 @@
package com.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
/**
* Kafka Consumer 配置属性类
* 替代 AppConfig 的静态方法读取,支持 docker-compose 环境变量覆盖
*/
@Data
@Component
@ConfigurationProperties(prefix = "spring.kafka.consumer")
public class KafkaConsumerProperties {
/** Kafka 集群地址 */
private String bootstrapServers = "192.168.4.99:9092";
/** 消费者组ID */
private String groupId = "agent-syslog-group";
/** 消费位置重置策略 */
private String autoOffsetReset = "latest";
/** 是否自动提交 */
private boolean enableAutoCommit = false;
/** 自动提交间隔(ms) */
private String autoCommitInterval = "1000";
/** 消费Topic */
private String topic = "agent-syslog-topic";
}
@@ -28,6 +28,8 @@ public class DmColumnController {
private DmColumnService dmColumnService;
@Autowired
private DmNormalizeRuleService dmNormalizeRuleService;
@Autowired
private LogNormalProcessor logNormalProcessor;
private static HashMap OrginalColumnMap ;
@@ -53,8 +55,7 @@ public class DmColumnController {
//List<Map<String, Object>> columnList=dmColumnService.selectAllNormal();
//List<Map<String, Object>> rulelst= dmNormalizeRuleService.selectByDeviceIdAuto((long)1);
//OrginalColumnMap= getMessageToMap(strLogMsg);
LogNormalProcessor logNormalProcessor=new LogNormalProcessor( "", UUID.randomUUID().toString(),"test-topic");
logNormalProcessor.init();
logNormalProcessor.process("", UUID.randomUUID().toString(), "test-topic");
return dmColumn.getDisplayName() ;
//return dmColumn.getDisplayName() ;
@@ -34,6 +34,8 @@ public class SyslogController {
private InfluxSyslogService syslogService;
@Autowired
private SyslogNonNormalMessageService messageService;
@Autowired
private LogNormalProcessor logNormalProcessor;
/**
* 接口1:按时间范围查询
* GET /api/syslog/time-range?startTime=2023-11-20T00:00:00Z&endTime=2023-11-20T23:59:59Z
@@ -191,10 +193,8 @@ public class SyslogController {
log.debug("开始泛化处理,syslog_uuid: {}, topic: {}", syslogUuid, syslogTopic);
// 调用泛化处理类
LogNormalProcessor logNormalProcessor = new LogNormalProcessor(
combinedMessage, syslogUuid, syslogTopic);
logNormalProcessor.init();
// 调用泛化处理类Spring管理的Bean
logNormalProcessor.process(combinedMessage, syslogUuid, syslogTopic);
processSuccess = true;
processMessage = "泛化处理成功";
@@ -314,9 +314,8 @@ public class SyslogController {
firstMessage.getSyslogMessage() : "";
String combinedMessage = headerMessage + syslogMessage;
LogNormalProcessor processor = new LogNormalProcessor(
combinedMessage, syslogUuid, firstMessage.getSyslogTopic());
processor.init();
logNormalProcessor.process(combinedMessage, syslogUuid,
firstMessage.getSyslogTopic());
processSuccess = true;
processMessage = "处理成功";
@@ -5,9 +5,11 @@ import com.influxdb.client.WriteApi;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.HealthCheck;
import com.influxdb.client.write.Point;
import com.config.AppConfig;
import com.config.AppProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import java.util.ArrayList;
@@ -19,6 +21,7 @@ import java.time.Instant;
import java.time.format.DateTimeFormatter;
@Component
public class InfluxDBClient implements AutoCloseable {
private static final Logger logger = LoggerFactory.getLogger(InfluxDBClient.class);
@@ -29,18 +32,22 @@ public class InfluxDBClient implements AutoCloseable {
private final String org;
private final QueryApi queryApi;
public InfluxDBClient() {
private final InfluxDBConfig influxDBConfig;
@Autowired
public InfluxDBClient(AppProperties appProperties, InfluxDBConfig influxDBConfig) {
this.influxDBConfig = influxDBConfig;
this.influxDB = InfluxDBClientFactory.create(
AppConfig.getInfluxUrl(),
AppConfig.getInfluxToken().toCharArray(),
AppConfig.getInfluxOrg(),
AppConfig.getInfluxBucket()
appProperties.getInfluxUrl(),
appProperties.getInfluxToken().toCharArray(),
appProperties.getInfluxOrg(),
appProperties.getInfluxBucket()
);
this.writeApi = influxDB.makeWriteApi(InfluxDBConfig.getWriteOptions());
this.writeApi = influxDB.makeWriteApi(influxDBConfig.getWriteOptions());
this.writeApiBlocking = influxDB.getWriteApiBlocking();
this.bucket = AppConfig.getInfluxBucket();
this.org = AppConfig.getInfluxOrg();
this.bucket = appProperties.getInfluxBucket();
this.org = appProperties.getInfluxOrg();
this.queryApi = influxDB.getQueryApi();
// 检查连接状态
@@ -92,7 +99,7 @@ public class InfluxDBClient implements AutoCloseable {
*/
public void writeRecord(String lineProtocol) {
try {
writeApi.writeRecord(InfluxDBConfig.getWritePrecision(),lineProtocol);
writeApi.writeRecord(influxDBConfig.getWritePrecision(),lineProtocol);
} catch (Exception e) {
logger.error("Failed to write record to InfluxDB: {}", e.getMessage());
}
@@ -1,18 +1,31 @@
package com.influx;
import com.config.AppProperties;
import com.influxdb.client.WriteOptions;
import com.influxdb.client.domain.WritePrecision;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class InfluxDBConfig {
public static WriteOptions getWriteOptions() {
private final AppProperties appProperties;
@Autowired
public InfluxDBConfig(AppProperties appProperties) {
this.appProperties = appProperties;
}
public WriteOptions getWriteOptions() {
return WriteOptions.builder()
.batchSize(com.config.AppConfig.getInfluxBatchSize())
.flushInterval(com.config.AppConfig.getInfluxFlushInterval())
.batchSize(appProperties.getInfluxBatchSize())
.flushInterval(appProperties.getInfluxFlushInterval())
.jitterInterval(1000)
.retryInterval(com.config.AppConfig.getInfluxRetryDelay())
.retryInterval(appProperties.getInfluxRetryDelay())
.build();
}
public static WritePrecision getWritePrecision() {
public WritePrecision getWritePrecision() {
return WritePrecision.MS; // 毫秒精度
}
}
@@ -1,7 +1,7 @@
package com.influx;
import com.config.AppConfig;
import com.common.util.SpringContextUtil;
import com.influx.InfluxDBClient;
import com.influxdb.client.*;
import com.influxdb.client.domain.WritePrecision;
@@ -30,7 +30,7 @@ public class SyslogToInfluxApp {
logger.info("Starting Syslog to InfluxDB application...");
// 初始化 InfluxDB 客户端
influxClient = new InfluxDBClient();
influxClient = SpringContextUtil.getBean(InfluxDBClient.class);
influxClient.writeRecord(syslog);
Point point = Point.measurement("syslog_normal")
.addTag("ip", "localhost") // 添加标签
@@ -18,7 +18,9 @@ import com.common.util.JsonParser;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.common.util.JsonParser;
import com.config.AppConfig;
import com.common.util.SpringContextUtil;
import com.config.AppProperties;
import com.config.KafkaConsumerProperties;
import com.Modules.NormalData.LogNormalProcessor;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
@@ -38,34 +40,28 @@ public class kafkalogconsumer {
}
public static void Run()
{
// 配置消费者属性
//Properties props = new Properties();
//props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.222.130:9092");
// props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group-app");
AppProperties appProperties = SpringContextUtil.getBean(AppProperties.class);
KafkaConsumerProperties kafkaProps = SpringContextUtil.getBean(KafkaConsumerProperties.class);
LogNormalProcessor logNormalProcessor = SpringContextUtil.getBean(LogNormalProcessor.class);
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, AppConfig.getBootstrapServers());
props.put(ConsumerConfig.GROUP_ID_CONFIG, AppConfig.getGroupId());
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProps.getBootstrapServers());
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProps.getGroupId());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 可选配置
//props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none"); // 从最早的消息开始消费
//props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的消息开始消费
//props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); // 自动提交偏移量
//props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 自动提交间隔
//props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000); // 设置单次拉取最大消息数[citation:6]
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, AppConfig.getAutoOffsetReset()); // 从last开始消费
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, AppConfig.getEnableAutoCommit()); // 自动提交偏移量
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,AppConfig.getAutoCommitIntervalMS()); // 自动提交间隔
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaProps.getAutoOffsetReset());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaProps.isEnableAutoCommit());
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaProps.getAutoCommitInterval());
// 创建消费者实例
Consumer<String, String> consumer = new KafkaConsumer<>(props);
try {
// 订阅主题
consumer.subscribe(Collections.singletonList(AppConfig.getTopic()));
consumer.subscribe(Collections.singletonList(kafkaProps.getTopic()));
System.out.println("开始消费消息...");
com.influx.InfluxDBClient influxClient = new InfluxDBClient();
com.influx.InfluxDBClient influxClient = SpringContextUtil.getBean(com.influx.InfluxDBClient.class);
// 持续消费消息
while (true) {
// 拉取消息(等待最多100毫秒)
@@ -88,7 +84,7 @@ public class kafkalogconsumer {
Point point = Point.measurement("syslog_security")
.addTag("deviceid", mapdev.get("device_id")) // 添加标签
.addTag("uuid", sysLogUUID) //syslog uuid
.addTag("topic", AppConfig.getTopic()) //kafka topic
.addTag("topic", kafkaProps.getTopic()) //kafka topic
.addField("message", record.value()) // 添加字段
.time(System.currentTimeMillis(), WritePrecision.MS) ;// 毫秒级时间戳
influxClient.writePointBlocking(point);
@@ -97,12 +93,9 @@ public class kafkalogconsumer {
//insertSingleRecord( record.value());
System.out.println("insert postgres syslog ,value:"+ record.key());
//String syslogMessage= AppConfig.geRunEnvironment().equals("test")? record.value().substring(34) : record.value();
String syslogMessage= record.value();
//剔除测试环境本机syslog新增的头部信息
LogNormalProcessor logNormalProcessor = new LogNormalProcessor(syslogMessage,sysLogUUID,AppConfig.getTopic());
//LogNormalProcessor logNormalProcessor =new LogNormalProcessor(record.value());
logNormalProcessor.init();
//使用注入的 Spring Bean 进行标准化处理
logNormalProcessor.process(syslogMessage, sysLogUUID, null);
}
// 手动提交偏移量(如果禁用自动提交)
consumer.commitSync();
@@ -5,8 +5,10 @@ import com.common.entity.XdrHoneypot;
import com.common.mapper.XdrHoneypotMapper;
import com.common.util.JsonParser;
import com.common.util.MyBatisUtil;
import com.common.util.SpringContextUtil;
import com.common.util.SyslogParser;
import com.config.AppConfig;
import com.config.AppProperties;
import com.config.KafkaConsumerProperties;
import com.influx.InfluxDBClient;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
@@ -36,34 +38,28 @@ public class kafkalogconsumerThead {
}
public static void Run()
{
// 配置消费者属性
//Properties props = new Properties();
//props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.222.130:9092");
// props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group-app");
KafkaConsumerProperties kafkaProps = SpringContextUtil.getBean(KafkaConsumerProperties.class);
LogNormalProcessor logNormalProcessor = SpringContextUtil.getBean(LogNormalProcessor.class);
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, AppConfig.getBootstrapServers());
props.put(ConsumerConfig.GROUP_ID_CONFIG, AppConfig.getGroupId());
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProps.getBootstrapServers());
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProps.getGroupId());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 可选配置
//props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none"); // 从最早的消息开始消费
//props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的消息开始消费
//props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); // 自动提交偏移量
//props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 自动提交间隔
//props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000); // 设置单次拉取最大消息数[citation:6]
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, AppConfig.getAutoOffsetReset()); // 从last开始消费
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, AppConfig.getEnableAutoCommit()); // 自动提交偏移量
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,AppConfig.getAutoCommitIntervalMS()); // 自动提交间隔
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaProps.getAutoOffsetReset());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaProps.isEnableAutoCommit());
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaProps.getAutoCommitInterval());
// 创建消费者实例
Consumer<String, String> consumer = new KafkaConsumer<>(props);
try {
// 订阅主题
consumer.subscribe(Collections.singletonList(AppConfig.getTopic()));
consumer.subscribe(Collections.singletonList(kafkaProps.getTopic()));
System.out.println("开始消费消息...");
InfluxDBClient influxClient = new InfluxDBClient(); ;
InfluxDBClient influxClient = SpringContextUtil.getBean(InfluxDBClient.class);
// 持续消费消息
while (true) {
// 拉取消息(等待最多100毫秒)
@@ -86,7 +82,7 @@ public class kafkalogconsumerThead {
Point point = Point.measurement("syslog_security")
.addTag("deviceid", mapdev.get("device_id")) // 添加标签
.addTag("uuid", sysLogUUID) //syslog uuid
.addTag("topic", AppConfig.getTopic()) //kafka topic
.addTag("topic", kafkaProps.getTopic()) //kafka topic
.addField("message", record.value()) // 添加字段
.time(System.currentTimeMillis(), WritePrecision.MS) ;// 毫秒级时间戳
influxClient.writePointBlocking(point);
@@ -95,12 +91,9 @@ public class kafkalogconsumerThead {
//insertSingleRecord( record.value());
System.out.println("insert postgres syslog ,value:"+ record.key());
//String syslogMessage= AppConfig.geRunEnvironment().equals("test")? record.value().substring(34) : record.value();
String syslogMessage= record.value();
//剔除测试环境本机syslog新增的头部信息
LogNormalProcessor logNormalProcessor = new LogNormalProcessor(syslogMessage,sysLogUUID,AppConfig.getTopic());
//LogNormalProcessor logNormalProcessor =new LogNormalProcessor(record.value());
logNormalProcessor.init();
//使用注入的 Spring Bean 进行标准化处理
logNormalProcessor.process(syslogMessage, sysLogUUID, null);
}
// 手动提交偏移量(如果禁用自动提交)
// consumer.commitSync();
@@ -39,7 +39,7 @@ spring.datasource.url=jdbc:dm://192.168.4.99:5237
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=ECOSYS
# mybatis Configuration
mybatis.mapper-locations=classpath:mapper/*.xml
mybatis.type-aliases-package=com.common.entity
@@ -57,7 +57,7 @@ spring.kafka.consumer.group-id=test-group
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-commit-interval=1000
spring.kafka.consumer.topic=agent-syslog-topic
spring.kafka.consumer.topic=test-topic
spring.kafka.consumer.max-poll-records=1000
spring.kafka.consumer.properties.max.poll.interval.ms=300000
@@ -137,7 +137,7 @@ spring.datasource.hikari.max-lifetime=900000
spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL
spring.datasource.hikari.validation-timeout=5000
spring.datasource.hikari.leak-detection-threshold=30000
spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer
spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-dev
spring.datasource.hikari.auto-commit=false
# 关联分析规则配置
@@ -145,7 +145,10 @@ analysis.realtime.enabled= true
# 检查间隔(秒) - 默认10秒
analysis.realtime.check-interval-seconds: 10
# 关联分析规则达梦安全版子查询模式开关(默认false,生产环境达梦安全版设为true)
# true → 使用 RealtimeAnalysisEngineDm,子查询包装SQL,适配 GROUP BY 不支持自定义函数
# false → 使用 RealtimeAnalysisEngine,平铺SQL(原有行为)
analysis.db.dameng-subquery-mode=true
# ============================================
# 探针联动API配置
@@ -150,6 +150,9 @@ analysis.realtime.enabled= true
# 检查间隔(秒) - 默认10秒
analysis.realtime.check-interval-seconds: 10
# 达梦安全版子查询模式开关(预发环境若同为达梦安全定制版,按需开启)
analysis.db.dameng-subquery-mode=false
# ============================================
# 探针联动API配置
# ============================================
@@ -148,4 +148,7 @@ spring.datasource.hikari.auto-commit=false
# 关联分析规则配置
analysis.realtime.enabled= true
# 检查间隔(秒) - 默认10秒
analysis.realtime.check-interval-seconds: 10
analysis.realtime.check-interval-seconds: 10
# 达梦安全版子查询模式开关(生产环境达梦为安全定制版,需开启)
analysis.db.dameng-subquery-mode=true
@@ -150,6 +150,12 @@ analysis.realtime.enabled= true
# 检查间隔(秒) - 默认10秒
analysis.realtime.check-interval-seconds: 10
# 关联分析规则达梦安全版子查询模式开关(默认false,生产环境达梦安全版设为true)
# true → 使用 RealtimeAnalysisEngineDm,子查询包装SQL,适配 GROUP BY 不支持自定义函数
# false → 使用 RealtimeAnalysisEngine,平铺SQL(原有行为)
analysis.db.dameng-subquery-mode=true
# ============================================
# 探针联动API配置
# ============================================
@@ -162,9 +168,9 @@ interlocking.api.base-url=http://192.168.4.99:8089/xdrservice/interlocking
# 告警健康检查配置
# ============================================
# 告警表无数据阈值(小时)
alarm.health-check.alarm-hours=2
alarm.health-check.alarm-hours=4
# 告警日志表无数据阈值(小时)
alarm.health-check.alarm-visit-hours=4
alarm.health-check.alarm-visit-hours=2
# 是否启用定时巡检
alarm.health-check.enabled=true
@@ -150,6 +150,12 @@ analysis.realtime.enabled= true
# 检查间隔(秒) - 默认10秒
analysis.realtime.check-interval-seconds: 10
# 关联分析规则达梦安全版子查询模式开关(默认false,生产环境达梦安全版设为true)
# true → 使用 RealtimeAnalysisEngineDm,子查询包装SQL,适配 GROUP BY 不支持自定义函数
# false → 使用 RealtimeAnalysisEngine,平铺SQL(原有行为)
analysis.db.dameng-subquery-mode=true
# ============================================
# 探针联动API配置
# ============================================
@@ -162,9 +168,9 @@ interlocking.api.base-url=http://192.168.4.99:8089/xdrservice/interlocking
# 告警健康检查配置
# ============================================
# 告警表无数据阈值(小时)
alarm.health-check.alarm-hours=2
alarm.health-check.alarm-hours=4
# 告警日志表无数据阈值(小时)
alarm.health-check.alarm-visit-hours=4
alarm.health-check.alarm-visit-hours=2
# 是否启用定时巡检
alarm.health-check.enabled=true
@@ -22,12 +22,17 @@
<result column="task_count" property="taskCount" />
<result column="recent_discover_time" property="recentDiscoverTime" />
<result column="epm_upper_limit" property="epmUpperLimit" />
<result column="expire_time" property="expireTime" />
<result column="remark" property="remark" />
<result column="device_ip" property="deviceIp" />
<result column="organization_id" property="organizationId" />
</resultMap>
<sql id="Base_Column_List">
id, created_at, updated_at, deleted_at, device_id, method, task_name,
first_time, last_success_time, last_failed_time, detail_id, epm, epm_peak,
process_architecture, task_count, recent_discover_time, epm_upper_limit
process_architecture, task_count, recent_discover_time, epm_upper_limit,
expire_time, remark, device_ip, organization_id
</sql>
<!-- 多条件组合查询 -->
@@ -60,6 +65,16 @@
<if test="lastSuccessTime != null">
AND last_success_time >= #{lastSuccessTime}
</if>
<!-- 新增字段查询 -->
<if test="deviceIp != null and deviceIp != ''">
AND device_ip = #{deviceIp}
</if>
<if test="organizationId != null">
AND organization_id = #{organizationId}
</if>
<if test="remark != null and remark != ''">
AND remark LIKE CONCAT('%', #{remark}, '%')
</if>
</where>
ORDER BY updated_at DESC
</select>
@@ -54,46 +54,47 @@
<!-- 只查询构建API请求体所需的字段 -->
<select id="findRequiredFieldsAfterTime" resultMap="requiredFieldsMap">
SELECT
id,
created_at,
log_time,
device_id,
device_ip,
dest_ip,
dest_port,
dest_mac,
src_ip,
src_port,
src_mac,
host_file_path,
file_md5,
file_name,
dest_city,
dest_country,
dest_lat,
dest_lon,
src_city,
src_country,
src_country_code,
src_lat,
src_lon,
http_host,
host_file_md5,
http_req_header_raw,
http_method,
http_resp_content_type,
proto,
http_referer,
http_url,
http_status_code,
-- 可选字段
event_date,
attack_result,
engine_type,
syslog_uuid,
syslog_topic
id,
created_at,
log_time,
device_id,
device_ip,
dest_ip,
dest_port,
dest_mac,
src_ip,
src_port,
src_mac,
host_file_path,
file_md5,
file_name,
dest_city,
dest_country,
dest_lat,
dest_lon,
src_city,
src_country,
src_country_code,
src_lat,
src_lon,
http_host,
host_file_md5,
http_req_header_raw,
http_method,
http_resp_content_type,
proto,
http_referer,
http_url,
http_status_code,
-- 可选字段
event_date,
attack_result,
engine_type,
syslog_uuid,
syslog_topic
FROM syslog_normal_data
WHERE created_at >= #{startTime}
WHERE created_at >= #{startTime} and created_at &lt; #{endTime}
and (proto='http' or proto='https')
ORDER BY created_at ASC
<!-- LIMIT 1000 -->
</select>