Agent/tr-agent-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tcpdump策略增加立即执行、优化tcpdump探测

Browse Source
This commit is contained in:
gaoyutao
2026-01-30 19:18:40 +08:00
parent c5353aaafd
commit 04a51bc201
7 changed files with 264 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/tongran/agent/client/core/eo/TcpdumpEO.java
+17
View File
@@ -0,0 +1,17 @@
package com.tongran.agent.client.core.eo;
import lombok.Data;
import java.util.List;
@Data
public class TcpdumpEO {
/** 开启探测(0:否,1:是) */
private Integer detectFlag;
/** 探测频率1 每天 2立即执行 */
private String frequency;
/** 探测时间列表(多个时间用,分隔) */
private String detectTimes;
}
src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java
+13
View File
@@ -52,4 +52,17 @@ public class SchedulerConfig {
executor.initialize(); // 重要：必须调用initialize()
return executor;
}
// 在SchedulerConfig中添加专用TCPDUMP线程池
@Bean("tcpdumpTaskExecutor")
public ThreadPoolTaskExecutor tcpdumpTaskExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2); // 核心线程数
executor.setMaxPoolSize(10); // 最大线程数（根据网卡数量调整）
executor.setQueueCapacity(20); // 队列容量
executor.setKeepAliveSeconds(300);// 空闲线程存活时间
executor.setThreadNamePrefix("tcpdump-worker-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}
src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskConfig.java
+46 -3
View File
@@ -1,8 +1,12 @@
package com.tongran.agent.client.scheduler.task;
import com.tongran.agent.client.utils.AssertLog;
import lombok.Data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.config.CronTask;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
@@ -20,6 +24,9 @@ public class SpecificTimeTaskConfig implements SchedulingConfigurer {
private ScheduledTaskRegistrar taskRegistrar;
private final Map<String, ScheduledFuture<?>> taskFutures = new ConcurrentHashMap<>();
private final Map<String, Object> taskMap = new ConcurrentHashMap<>();
@Autowired
@Qualifier("tcpdumpTaskExecutor")
private ThreadPoolTaskExecutor tcpdumpExecutor;
@Override
public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
@@ -31,13 +38,36 @@ public class SpecificTimeTaskConfig implements SchedulingConfigurer {
*/
public void addCronTask(String taskId, Runnable task, String cronExpression) {
removeTaskIfExists(taskId);
CronTask cronTask = new CronTask(task, cronExpression);
ScheduledFuture<?> future = taskRegistrar.getScheduler().schedule(
cronTask.getRunnable(),
cronTask.getRunnable(),
cronTask.getTrigger()
);
taskMap.put(taskId, cronTask);
taskFutures.put(taskId, future);
}
public void addTcpdumpCronTask(String taskId, Runnable task, String cronExpression) {
removeTaskIfExists(taskId);
// 使用轻量级调度，实际任务交给专用线程池
CronTask cronTask = new CronTask(() -> {
// 异步执行，不阻塞定时任务线程
tcpdumpExecutor.execute(() -> {
try {
task.run();
} catch (Exception e) {
AssertLog.error("任务执行异常 {}: {}", taskId, e.getMessage());
}
});
}, cronExpression);
ScheduledFuture<?> future = taskRegistrar.getScheduler().schedule(
cronTask.getRunnable(),
cronTask.getTrigger()
);
taskMap.put(taskId, cronTask);
taskFutures.put(taskId, future);
}
@@ -80,6 +110,19 @@ public class SpecificTimeTaskConfig implements SchedulingConfigurer {
String cronExpression = String.format("%d %d %d * * ?", second, minute, hour);
addCronTask(taskId, task, cronExpression);
}
public void addDailyTimeTcpdumpTask(String taskId, Runnable task, String time) {
String[] timeParts = time.split(":");
if (timeParts.length < 2) {
throw new IllegalArgumentException("时间格式应为 HH:mm 或 HH:mm:ss");
}
int hour = Integer.parseInt(timeParts[0]);
int minute = Integer.parseInt(timeParts[1]);
int second = timeParts.length > 2 ? Integer.parseInt(timeParts[2]) : 0;
String cronExpression = String.format("%d %d %d * * ?", second, minute, hour);
addTcpdumpCronTask(taskId, task, cronExpression);
}
/**
* 添加延迟任务
src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java
+152 -61
View File
@@ -18,6 +18,8 @@ import com.tongran.agent.client.utils.AssertLog;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
@@ -30,6 +32,7 @@ import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
@Service
@@ -44,6 +47,9 @@ public class SpecificTimeTaskService {
@Resource
private SpecificTimeTaskConfig taskConfig;
@Resource
@Qualifier("tcpdumpTaskExecutor")
private ThreadPoolTaskExecutor tcpdumpExecutor;
public SpecificTimeTaskService() {
this.sessionManager = SessionManager.getInstance();
@@ -54,7 +60,7 @@ public class SpecificTimeTaskService {
Runnable task = createTcpdumpTaskRunnable(request);
try {
if (request.getTime() != null) {
taskConfig.addDailyTimeTask(taskId, task, request.getTime());
taskConfig.addDailyTimeTcpdumpTask(taskId, task, request.getTime());
} else {
throw new IllegalArgumentException("必须指定一种时间方式");
}
@@ -63,6 +69,22 @@ public class SpecificTimeTaskService {
throw new RuntimeException("创建任务失败: " + e.getMessage(), e);
}
}
public CompletableFuture<Boolean> executeTcpdumpImmediatelyWithResult(SpecificTimeRequest request) {
String taskId = request.getTaskId() != null ?
request.getTaskId() : "tcpdump-immediate-" + System.currentTimeMillis();
return CompletableFuture.supplyAsync(() -> {
try {
AssertLog.info("立即执行TCPDUMP任务开始: {}", taskId);
executeTcpdump(request);
AssertLog.info("立即执行TCPDUMP任务完成: {}", taskId);
return true;
} catch (Exception e) {
AssertLog.error("立即执行TCPDUMP任务异常 {}: {}", taskId, e.getMessage(), e);
return false;
}
}, tcpdumpExecutor); // 使用专用线程池
}
/**
* 创建指定时间任务
*/
@@ -89,15 +111,19 @@ public class SpecificTimeTaskService {
throw new RuntimeException("创建任务失败: " + e.getMessage(), e);
}
}
private Runnable createTcpdumpTaskRunnable(SpecificTimeRequest request) {
return () -> {
System.out.println("执行定时任务: " + request.getTaskName());
System.out.println("执行时间: " + LocalDateTime.now());
System.out.println("-----------------------------------");
// 具体的业务逻辑
executeTcpdump(request);
// 直接执行，因为已经在专用线程池中了
try {
System.out.println("执行定时任务: " + request.getTaskName());
System.out.println("执行时间: " + LocalDateTime.now());
System.out.println("-----------------------------------");
executeTcpdump(request);
} catch (Exception e) {
AssertLog.error("TCPDUMP任务执行异常: {}", e.getMessage(), e);
}
};
}
@@ -147,74 +173,139 @@ public class SpecificTimeTaskService {
/**
* 对单个网卡进行抓包和分析
*/
private List<IpCount> captureAndAnalyzeInterface(SpecificTimeRequest request, String timestamp, String interfaceName) {
// 检查目录
if (AdvancedAsyncDownloader.createSingleDirectoryIfNotExists(properties.getTempPcapPath())) {
String pcapFileName = properties.getTempPcapPath() + timestamp + "_" + interfaceName + ".pcap";
List<IpCount> ipCountList = new ArrayList<>();
try {
// 第一步：执行tcpdump命令捕获数据
ProcessBuilder captureBuilder = new ProcessBuilder();
captureBuilder.command("timeout", "10", "tcpdump", "-i", interfaceName,
"-nn", "-n", "-w", pcapFileName);
private List<IpCount> captureAndAnalyzeInterface(SpecificTimeRequest request,
String timestamp, String interfaceName) {
Process captureProcess = captureBuilder.start();
int captureExitCode = captureProcess.waitFor();
if (!AdvancedAsyncDownloader.createSingleDirectoryIfNotExists(properties.getTempPcapPath())) {
return new ArrayList<>();
}
if (captureExitCode != 0 && captureExitCode != 124) {
AssertLog.error("Tcpdump pcap包生成失败，网卡名称 {} with exit code: {}", interfaceName, captureExitCode);
return ipCountList;
String pcapFileName = properties.getTempPcapPath() + timestamp + "_" + interfaceName + ".pcap";
List<IpCount> ipCountList = new ArrayList<>();
try {
// 第一步：执行tcpdump抓包（带超时控制）
CompletableFuture<Boolean> captureFuture = CompletableFuture.supplyAsync(() -> {
try {
ProcessBuilder captureBuilder = new ProcessBuilder();
// 设置抓包时间，根据实际需求调整
captureBuilder.command("timeout", "10", "tcpdump",
"-i", interfaceName, "-nn", "-n", "-w", pcapFileName);
Process captureProcess = captureBuilder.start();
int captureExitCode = captureProcess.waitFor();
if (captureExitCode != 0 && captureExitCode != 124) {
AssertLog.error("Tcpdump抓包失败，网卡: {}，退出码: {}", interfaceName, captureExitCode);
return false;
}
return true;
} catch (Exception e) {
AssertLog.error("抓包过程异常 {}: {}", interfaceName, e.getMessage());
return false;
}
});
// 等待文件写入完成
Thread.sleep(20000);
// 等待抓包完成（最长7分钟）
boolean captureSuccess = captureFuture.get(20, TimeUnit.SECONDS);
// 检查文件是否存在且不为空
File pcapFile = new File(pcapFileName);
if (!pcapFile.exists() || pcapFile.length() == 0) {
AssertLog.warn("PCAP file is empty or does not exist for interface: {}", interfaceName);
return ipCountList;
}
if (!captureSuccess) {
return ipCountList;
}
// 第二步：分析pcap文件并统计目标IP
ProcessBuilder analyzeBuilder = new ProcessBuilder();
analyzeBuilder.command("bash", "-c",
"tcpdump -r " + pcapFileName + " -nn 2>/dev/null | " +
"awk '/^[0-9]/ && ($2 == \"IP\" || $2 == \"IP6\") {dest=$5; if(match(dest,/^([0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+)/)) {dest=substr(dest,RSTART,RLENGTH)} else if(match(dest,/([0-9a-fA-F:]+)/)) {dest=substr(dest,RSTART,RLENGTH); gsub(/:[0-9]+$/,\"\",dest)} print dest}' | " +
"sort | uniq -c | sort -rn");
// 第二步：分析PCAP文件
CompletableFuture<List<IpCount>> analyzeFuture = CompletableFuture.supplyAsync(() -> {
return analyzePcapFile(pcapFileName, interfaceName);
});
Process analyzeProcess = analyzeBuilder.start();
// 分析阶段超时时间（2分钟应该足够）
ipCountList = analyzeFuture.get(2, TimeUnit.MINUTES);
} catch (TimeoutException e) {
AssertLog.warn("网卡 {} 抓包或分析超时，跳过处理", interfaceName);
} catch (Exception e) {
AssertLog.error("处理网卡 {} 时异常: {}", interfaceName, e.getMessage());
} finally {
// 确保清理临时文件
cleanupTempFile(pcapFileName);
}
return ipCountList;
}
/**
* 分析PCAP文件并统计IP信息
*/
private List<IpCount> analyzePcapFile(String pcapFileName, String interfaceName) {
List<IpCount> ipCountList = new ArrayList<>();
File pcapFile = new File(pcapFileName);
if (!pcapFile.exists() || pcapFile.length() == 0) {
AssertLog.warn("PCAP文件不存在或为空: {}", pcapFileName);
return ipCountList;
}
try {
ProcessBuilder analyzeBuilder = new ProcessBuilder();
// 使用更可靠的命令分析PCAP文件
analyzeBuilder.command("bash", "-c",
"tcpdump -r " + pcapFileName + " -nn 2>/dev/null | " +
"awk '/^[0-9]/ && ($2 == \"IP\" || $2 == \"IP6\") { " +
" dest=$5; " +
" if(match(dest,/^([0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+)/)) { " +
" dest=substr(dest,RSTART,RLENGTH) " +
" } else if(match(dest,/([0-9a-fA-F:]+)/)) { " +
" dest=substr(dest,RSTART,RLENGTH); gsub(/:[0-9]+$/,\"\",dest) " +
" } " +
" print dest " +
"}' | " +
"sort | uniq -c | sort -rn");
Process analyzeProcess = analyzeBuilder.start();
// 读取分析结果
StringBuilder result = new StringBuilder();
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(analyzeProcess.getInputStream()))) {
// 读取分析结果
BufferedReader reader = new BufferedReader(new InputStreamReader(analyzeProcess.getInputStream()));
StringBuilder result = new StringBuilder();
String line;
while ((line = reader.readLine()) != null) {
result.append(line).append("\n");
}
int analyzeExitCode = analyzeProcess.waitFor();
if (analyzeExitCode == 0) {
AssertLog.debug("Tcpdump analysis completed successfully for interface: {}", interfaceName);
ipCountList = parseTcpdumpOutput(result.toString());
} else {
AssertLog.error("Tcpdump analysis failed for interface {} with exit code: {}", interfaceName, analyzeExitCode);
}
// 清理临时文件
if (pcapFile.exists()) {
pcapFile.delete();
}
} catch (Exception e) {
AssertLog.error("Error capturing interface {}: {}", interfaceName, e.getMessage());
}
return ipCountList;
int analyzeExitCode = analyzeProcess.waitFor();
if (analyzeExitCode == 0) {
AssertLog.debug("PCAP分析完成: {}，文件大小: {} bytes", interfaceName, pcapFile.length());
ipCountList = parseTcpdumpOutput(result.toString());
} else {
AssertLog.error("PCAP分析失败: {}，退出码: {}", interfaceName, analyzeExitCode);
}
} catch (Exception e) {
AssertLog.error("分析PCAP文件异常 {}: {}", interfaceName, e.getMessage());
}
return ipCountList;
}
/**
* 清理临时文件
*/
private void cleanupTempFile(String fileName) {
try {
File file = new File(fileName);
if (file.exists()) {
boolean deleted = file.delete();
if (!deleted) {
AssertLog.warn("无法删除临时文件: {}", fileName);
} else {
AssertLog.debug("已清理临时文件: {}", fileName);
}
}
} catch (Exception e) {
AssertLog.warn("清理临时文件异常: {}", e.getMessage());
}
return new ArrayList<>();
}
/**
src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java
+34 -3
View File
@@ -940,11 +940,42 @@ public class AgentServiceImpl implements AgentService {
}
if(jsonObject.containsKey("tcpdumpTimes")){
String tcpdumpTimes = jsonObject.getString("tcpdumpTimes");
handleTcpdump(tcpdumpTimes);
TcpdumpEO tcpdumpEO = JSON.parseObject(tcpdumpTimes, TcpdumpEO.class);
if(tcpdumpEO.getDetectFlag() == 1){
if("1".equals(tcpdumpEO.getFrequency())){
handleTcpdump(tcpdumpEO);
}else if("2".equals(tcpdumpEO.getFrequency())){
// 立即执行
SpecificTimeRequest request = SpecificTimeRequest.builder()
.taskId("tcpdumpnow")
.taskName("tcpdumpnow")
.clientId(GlobalConfig.CLIENT_ID)
.build();
taskService.executeTcpdumpImmediatelyWithResult(request);
}
}else{
// 1. 将tcpdump探测时间写入配置文件
String[] lines = {
"# tcpdump探测时间配置文件",
"# 生成时间: " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()),
"# tcpdump探测时间arr",
"tcpdumpTimes=",
};
// 检查目录并写入配置文件
if (AdvancedAsyncDownloader.createSingleDirectoryIfNotExists(properties.getConfPath())) {
AgentUtil.bufferedWriter(properties.getConfPath() + "/tcpdumptimes.conf", lines);
}
// 更新全局变量
GlobalConfig.TCPDUMP_TIMES = tcpdumpTimes;
taskService.cancleTcpdumpSpecificTimeTask();
}
}
}
private void handleTcpdump(String tcpdumpTimes) {
private void handleTcpdump(TcpdumpEO tcpdumpEo) {
String tcpdumpTimes = tcpdumpEo.getDetectTimes();
if (!tcpdumpTimes.equals(GlobalConfig.TCPDUMP_TIMES)) {
// 1. 将tcpdump探测时间写入配置文件
String[] lines = {
@@ -2597,7 +2628,7 @@ public class AgentServiceImpl implements AgentService {
props.load(input);
String tcpdumpTimes = props.getProperty("tcpdumpTimes");
if(!"".equals(tcpdumpTimes)){
if(tcpdumpTimes != null && !tcpdumpTimes.trim().isEmpty()){
String[] tcpdumpTimeArr = tcpdumpTimes.split(",");
for (String tcpdumpTime : tcpdumpTimeArr) {
SpecificTimeRequest request = SpecificTimeRequest.builder()
src/main/resources/application-dev.yml
+1 -1
View File
@@ -17,7 +17,7 @@ logging:
netty:
server:
host: 120.211.95.173
port: 56620
port: 6620
client:
client-id: client-001
reconnect-interval: 5
src/main/resources/application.yml
+1 -1
View File
@@ -6,7 +6,7 @@ spring:
matching-strategy: ant_path_matcher
application:
name: tr-agent-client
version: 1.1.13
version: 1.1.14
conf-path: /usr/local/tongran/conf
script-path: /usr/local/tongran/sbin
tmp-path: /usr/local/tongran/tmp