gfwleak/enderbyendera-realtime-protection
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ft:

Browse Source 
指令状态获取方式修改
This commit is contained in:
PushM
2024-08-23 02:43:52 +08:00
parent f6883318e7
commit 4e9c4af247
7 changed files with 345 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/java/com/realtime/protection/configuration/entity/task/TaskCommandInfo.java
View File

@@ -131,6 +131,14 @@ public class TaskCommandInfo {
private Long totalPacketNum;
private int commandStatus;
private LocalDateTime lastTrafficQueryTime;
private LocalDateTime lastRCPQueryTime;
private String logUUID;
// private String hashValue;

15
src/main/java/com/realtime/protection/configuration/utils/enums/CommandStatusEnum.java
View File

@@ -8,19 +8,20 @@ import java.util.Map;
@Getter
public enum CommandStatusEnum {
START(1),//指令生成(研判后任务需要研判，其他任务可直接下发)
SEND(2),//（已研判）下发
SEND(2),// 下发,发送c3
RCPHIT(3),//..
FLOWHIT(4),//..
UNRCPHIT(5),//..
UNFLOWHIT(6),//..
IGNORE(7),//本次忽略
WHOLEIGNORE(8),//全部忽略
SUCCESS(9),//下发成功
FAIL(10),//下发失败
STOP(11),
PAUSE(12),
CANCEL(13),//撤销
END(14);//任务结束
JUDGED(9),// 研判通过
SUCCESS(10),//下发成功
FAIL(11),//下发失败
STOP(12),
PAUSE(13),
CANCEL(14),//撤销
END(0);//任务结束
private final int commandStatusNum;
private static final Map<Integer, CommandStatusEnum> map = new HashMap<>();

12
src/main/java/com/realtime/protection/server/command/CommandMapper.java
View File

@@ -7,6 +7,7 @@ import com.realtime.protection.configuration.entity.whitelist.WhiteListObject;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import java.time.LocalDateTime;
import java.util.List;
@Mapper
@@ -86,8 +87,6 @@ public interface CommandMapper {
List<Integer> queryRunningCommandsDistributeStatusByTaskId(Long taskId);
List<TaskCommandInfo> queryRunningCommandsRcpHitCount();
Integer queryCommandLogRcpHitCountByCommandId(@Param("command_id")String commandId);
void insertCommandHistoryWithStatus(@Param("command_id")String commandUUID,
@@ -99,4 +98,13 @@ public interface CommandMapper {
List<TaskCommandInfo> queryRunningCommandsTotalPacketNum();
Integer queryCommandLogTotalPacketNumByCommandId(@Param("command_id")String uuid);
List<TaskCommandInfo> queryCommandLogLastTwoRecord(@Param("command_id")String uuid);
List<TaskCommandInfo> queryRunningCommands();
void insertCommandHistoryWithStatusWithTime(@Param("command_id")String commandUUID,
@Param("log_id")String logId,
@Param("command_status")Integer commandStatus,
@Param("effective_time")LocalDateTime lastTrafficQueryTime);
}

15
src/main/java/com/realtime/protection/server/command/CommandService.java
View File

@@ -265,7 +265,7 @@ public class CommandService {
//研判状态也写入历史表
switch (isJudged){
case 1:
insertCommandHistory(commandId, CommandStatusEnum.SEND.getCommandStatusNum());
insertCommandHistory(commandId, CommandStatusEnum.JUDGED.getCommandStatusNum());
break;
case 2:
insertCommandHistory(commandId, CommandStatusEnum.IGNORE.getCommandStatusNum());
@@ -351,8 +351,8 @@ public class CommandService {
}
public List<TaskCommandInfo> queryRunningCommandsRcpHitCount() {
return commandMapper.queryRunningCommandsRcpHitCount();
public List<TaskCommandInfo> queryRunningCommands() {
return commandMapper.queryRunningCommands();
}
public Integer queryCommandLogRcpHitCountByCommandId(String commandI) {
@@ -366,4 +366,13 @@ public class CommandService {
public Integer queryCommandLogTotalPacketNumByCommandId(String uuid) {
return commandMapper.queryCommandLogTotalPacketNumByCommandId(uuid);
}
public List<TaskCommandInfo> queryCommandLogLastTwoRecord(String uuid) {
return commandMapper.queryCommandLogLastTwoRecord(uuid);
}
public void insertCommandHistoryWithTime(String commandUUID, Integer commandStatus, LocalDateTime lastTrafficQueryTime) {
String logId = UUID.randomUUID().toString();
commandMapper.insertCommandHistoryWithStatusWithTime(commandUUID, logId, commandStatus, lastTrafficQueryTime);
}
}

164
src/main/java/com/realtime/protection/server/task/status/StateChangeService.java
View File

@@ -157,7 +157,7 @@ public class StateChangeService {
continue;
}
//判断是否全为成功
if (commandStatusList.stream().allMatch(status -> status == null || status == 1)) {
if (commandStatusList.stream().allMatch(status -> status == null || status == 0)) {
//全为成功
try {
changeState(StateEnum.RUNNING_SUCCESS.getStateNum(), taskId, true);
@@ -165,7 +165,7 @@ public class StateChangeService {
log.warn(String.format("任务%d从%s状态变为运行中RUNNING_SUCCESS状态遭遇异常：%s",
taskId, taskService.queryTaskStatus(taskId), e.getMessage()));
}
} else if (commandStatusList.stream().anyMatch(status -> status == null || status == 1)) {
} else if (commandStatusList.stream().anyMatch(status -> status == null || status == 0)) {
//部分成功
try {
changeState(StateEnum.RUNNING_PARTIAL_SUCCESS.getStateNum(), taskId, true);
@@ -187,87 +187,113 @@ public class StateChangeService {
/**
* 更新指令rcp命中状态
* 更新每次指令下周内rcp\flow命中状态
*/
@Scheduled(cron = "0 1/5 * * * ?")
@Scheduled(cron = "5/20 * * * * ?")
@Async
protected void updateCommandRCPHitStatus() {
//获取所有正在下发的指令
List<TaskCommandInfo> commandRcpHitCountList = commandService.queryRunningCommandsRcpHitCount();
log.info("rcp命中状态更新：正在下发的指令数量：{}", commandRcpHitCountList.size());
List<TaskCommandInfo> runningCommandList = commandService.queryRunningCommands();
log.info("rcp命中状态更新：正在下发的指令数量：{}", runningCommandList.size());
//遍历所有指令
for (TaskCommandInfo taskCommandInfo : commandRcpHitCountList) {
//查询当前指令的rcp-hitcount字段
Long rcpHitCount = taskCommandInfo.getRcpHitCount();
//查询当前历史表中该指令上一条日志的rcp-hitcount字段
Integer lastRcpHitCount = commandService.queryCommandLogRcpHitCountByCommandId(taskCommandInfo.getUUID());
//比较是否有变化
if (rcpHitCount == null ) {
//更新log表，rcp没命中
commandService.insertCommandHistory(taskCommandInfo.getUUID(), CommandStatusEnum.UNRCPHIT.getCommandStatusNum());
for (TaskCommandInfo taskCommandInfo : runningCommandList) {
//查询状态表，查看本次下发周期内是否已经记录了命中状态。查询最近的两条日志记录
List<TaskCommandInfo> commandLogs = commandService.queryCommandLogLastTwoRecord(taskCommandInfo.getUUID());
if (commandLogs.size()<2){
continue;
}
if (lastRcpHitCount == null){
//更新log表，rcp命中
commandService.insertCommandHistory(taskCommandInfo.getUUID(), CommandStatusEnum.RCPHIT.getCommandStatusNum());
TaskCommandInfo commandLogLatest = commandLogs.get(0);
TaskCommandInfo commandLogLatest2 = commandLogs.get(1);
log.info("指令历史状态：{}", commandLogs);
//如果日志状态为SEND、SUCCESS 或者 SEND、FAIL，则要根据当前t_command表的最新记录，和上次下发状态的命中计数相比判断是否有变化
if (commandLogLatest2.getCommandStatus()==CommandStatusEnum.SEND.getCommandStatusNum()
&& (commandLogLatest.getCommandStatus()==CommandStatusEnum.SUCCESS.getCommandStatusNum()
|| commandLogLatest.getCommandStatus()==CommandStatusEnum.FAIL.getCommandStatusNum()))
{
log.info("rcp命中查询：{}，{}",taskCommandInfo.getRcpHitCount(), commandLogLatest.getRcpHitCount());
if (taskCommandInfo.getRcpHitCount()!=null && commandLogLatest.getRcpHitCount()==null) {
//更新log表，rcp命中
commandService.insertCommandHistoryWithTime(taskCommandInfo.getUUID(),
CommandStatusEnum.RCPHIT.getCommandStatusNum(), taskCommandInfo.getLastRCPQueryTime());
}
if (taskCommandInfo.getRcpHitCount()!=null && commandLogLatest.getRcpHitCount()!=null) {
if (taskCommandInfo.getRcpHitCount() > commandLogLatest.getRcpHitCount()) {
//更新log表，rcp命中
commandService.insertCommandHistoryWithTime(taskCommandInfo.getUUID(),
CommandStatusEnum.RCPHIT.getCommandStatusNum(), taskCommandInfo.getLastRCPQueryTime());
}
}
log.info("流量命中查询：{}，{}",taskCommandInfo.getTotalPacketNum(), commandLogLatest.getTotalPacketNum());
if (taskCommandInfo.getTotalPacketNum()==null && commandLogLatest.getTotalPacketNum()!=null) {
//更新log表，流量命中
commandService.insertCommandHistoryWithTime(taskCommandInfo.getUUID(),
CommandStatusEnum.FLOWHIT.getCommandStatusNum(),taskCommandInfo.getLastTrafficQueryTime());
}
if (taskCommandInfo.getTotalPacketNum()!=null && commandLogLatest.getTotalPacketNum()!=null) {
if (taskCommandInfo.getTotalPacketNum() > commandLogLatest.getTotalPacketNum()) {
//更新log表，流量命中
commandService.insertCommandHistoryWithTime(taskCommandInfo.getUUID(),
CommandStatusEnum.FLOWHIT.getCommandStatusNum(),taskCommandInfo.getLastTrafficQueryTime());
}
}
continue;
}
if (rcpHitCount > lastRcpHitCount) {
//更新log表，rcp命中
commandService.insertCommandHistory(taskCommandInfo.getUUID(), CommandStatusEnum.RCPHIT.getCommandStatusNum());
} else {
//更新log表，rcp没命中
commandService.insertCommandHistory(taskCommandInfo.getUUID(), CommandStatusEnum.UNRCPHIT.getCommandStatusNum());
}
//如果日志状态为SUCCESS/FAIL、RCPHIT/UNRCPHIT，则判断是否更新流量命中状态
if ((commandLogLatest2.getCommandStatus()==CommandStatusEnum.SUCCESS.getCommandStatusNum()
|| commandLogLatest2.getCommandStatus()==CommandStatusEnum.FAIL.getCommandStatusNum())
&& (commandLogLatest.getCommandStatus()==CommandStatusEnum.RCPHIT.getCommandStatusNum()
|| commandLogLatest.getCommandStatus()==CommandStatusEnum.UNRCPHIT.getCommandStatusNum()))
{
log.info("流量命中查询：{}，{}",taskCommandInfo.getTotalPacketNum(), commandLogLatest.getTotalPacketNum());
if (taskCommandInfo.getTotalPacketNum()==null && commandLogLatest.getTotalPacketNum()!=null) {
//更新log表，流量命中
commandService.insertCommandHistoryWithTime(taskCommandInfo.getUUID(),
CommandStatusEnum.FLOWHIT.getCommandStatusNum(),taskCommandInfo.getLastTrafficQueryTime());
}
if (taskCommandInfo.getTotalPacketNum()!=null && commandLogLatest.getTotalPacketNum()!=null) {
if (taskCommandInfo.getTotalPacketNum() > commandLogLatest.getTotalPacketNum()) {
//更新log表，流量命中
commandService.insertCommandHistoryWithTime(taskCommandInfo.getUUID(),
CommandStatusEnum.FLOWHIT.getCommandStatusNum(),taskCommandInfo.getLastTrafficQueryTime());
}
}
continue;
}
//如果日志状态为SUCCESS/FAIL、FLOWHIT/UNFLOWHIT，则判断是否更新rcp命中状态
if ((commandLogLatest2.getCommandStatus()==CommandStatusEnum.SUCCESS.getCommandStatusNum()
|| commandLogLatest2.getCommandStatus()==CommandStatusEnum.FAIL.getCommandStatusNum())
&& (commandLogLatest.getCommandStatus()==CommandStatusEnum.FLOWHIT.getCommandStatusNum()
|| commandLogLatest.getCommandStatus()==CommandStatusEnum.UNFLOWHIT.getCommandStatusNum()))
{
log.info("rcp命中查询：{}，{}",taskCommandInfo.getRcpHitCount(), commandLogLatest.getRcpHitCount());
if (taskCommandInfo.getRcpHitCount()!=null && commandLogLatest.getRcpHitCount()==null) {
//更新log表，rcp命中
commandService.insertCommandHistoryWithTime(taskCommandInfo.getUUID(),
CommandStatusEnum.RCPHIT.getCommandStatusNum(), taskCommandInfo.getLastRCPQueryTime());
}
if (taskCommandInfo.getRcpHitCount()!=null && commandLogLatest.getRcpHitCount()!=null) {
if (taskCommandInfo.getRcpHitCount() > commandLogLatest.getRcpHitCount()) {
//更新log表，rcp命中
commandService.insertCommandHistoryWithTime(taskCommandInfo.getUUID(),
CommandStatusEnum.RCPHIT.getCommandStatusNum(), taskCommandInfo.getLastRCPQueryTime());
}
}
continue;
}
//如果日志状态为RCPHIT/FLOWHIT、FLOWHIT/UNFLOWHIT，则不需要查询了
}
}
/**
* 更新指令rcp命中状态
*/
@Scheduled(cron = "0 3/5 * * * ?")
@Async
protected void updateCommandTrafficHitStatus() {
//查询当前历史表中该指令上一条日志的统计流量大小字段
//比较是否有变化
//更新log表，流量命中、流量未命中
//获取所有正在下发的指令
List<TaskCommandInfo> commandTotalPacketNumList = commandService.queryRunningCommandsTotalPacketNum();
log.info("流量命中状态：查询正在下发的指令数量：{}", commandTotalPacketNumList.size());
//遍历所有指令
for (TaskCommandInfo taskCommandInfo : commandTotalPacketNumList) {
//查询当前指令的统计流量大小字段
Long totalPacketNum = taskCommandInfo.getTotalPacketNum();
//查询当前历史表中该指令上一条日志的rcp-hitcount字段
Integer lastTotalPacketNum = commandService.queryCommandLogTotalPacketNumByCommandId(taskCommandInfo.getUUID());
//比较是否有变化
if (totalPacketNum == null ) {
//更新log表，rcp没命中
commandService.insertCommandHistory(taskCommandInfo.getUUID(), CommandStatusEnum.UNFLOWHIT.getCommandStatusNum());
continue;
}
if (lastTotalPacketNum == null){
//更新log表，rcp命中
commandService.insertCommandHistory(taskCommandInfo.getUUID(), CommandStatusEnum.FLOWHIT.getCommandStatusNum());
continue;
}
if (totalPacketNum > lastTotalPacketNum) {
//更新log表，rcp命中
commandService.insertCommandHistory(taskCommandInfo.getUUID(), CommandStatusEnum.FLOWHIT.getCommandStatusNum());
} else {
//更新log表，rcp没命中
commandService.insertCommandHistory(taskCommandInfo.getUUID(), CommandStatusEnum.UNFLOWHIT.getCommandStatusNum());
}
}
}
}

56
src/main/java/com/realtime/protection/server/task/status/StateHandler.java
View File

@@ -277,20 +277,38 @@ public class StateHandler {
})
.doOnError(WebClientResponseException.class, res -> success.set(false));
Map<String, Integer> response = mono.block(Duration.ofSeconds(5));
if (response == null) {
return false;
}
response.forEach((commandUUID, responseCode) -> {
log.info("指令首次下发成功, 指令uuid: " + commandUUID + ", responseCode: " + responseCode);
if (responseCode != 0) {
log.warn("指令首次下发失败, 指令uuid: " + commandUUID + ", responseCode: " + responseCode);
}
});
success.set(true);
// 异步处理响应
mono.subscribe(
response2 -> {
Map<String, Integer> response = response2;
success.set(false);
response.forEach((commandUUID, responseCode) -> {
log.info("指令首次下发成功, 指令uuid: " + commandUUID + ", responseCode: " + responseCode);
if (responseCode != 0) {
log.warn("指令首次下发失败, 指令uuid: " + commandUUID + ", responseCode: " + responseCode);
}
});
success.set(true);
},
error -> {
// 错误响应处理
System.err.println("指令首次下发错误: " + error.getMessage());
success.set(false);
}
);
// Map<String, Integer> response = mono.block(Duration.ofSeconds(5));
//
// if (response == null) {
// return false;
// }
// response.forEach((commandUUID, responseCode) -> {
// log.info("指令首次下发成功, 指令uuid: " + commandUUID + ", responseCode: " + responseCode);
// if (responseCode != 0) {
// log.warn("指令首次下发失败, 指令uuid: " + commandUUID + ", responseCode: " + responseCode);
// }
// });
//
// success.set(true);
return success.get();
}
@@ -335,16 +353,6 @@ public class StateHandler {
}
);
// if (response == null) {
// log.info("指令首次查询RCP返回为null");
// return false;
// }
// response.forEach((commandUUID, responseCode) -> {
// log.info("指令首次查询RCP成功, 指令uuid: " + commandUUID + ", responseCode: " + responseCode);
// if (responseCode != 0) {
// log.warn("指令首次查询RCP失败, 指令uuid: " + commandUUID + ", responseCode: " + responseCode);
// }
// });
success.set(true);