diff --git a/src/main/java/com/tongran/agent/client/core/eo/TcpdumpEO.java b/src/main/java/com/tongran/agent/client/core/eo/TcpdumpEO.java new file mode 100644 index 0000000..1dfc721 --- /dev/null +++ b/src/main/java/com/tongran/agent/client/core/eo/TcpdumpEO.java @@ -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; +} diff --git a/src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java b/src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java index 1c4975e..928dde6 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java +++ b/src/main/java/com/tongran/agent/client/scheduler/config/SchedulerConfig.java @@ -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; + } } \ No newline at end of file diff --git a/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskConfig.java b/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskConfig.java index 31ef6cb..f850335 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskConfig.java +++ b/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskConfig.java @@ -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> taskFutures = new ConcurrentHashMap<>(); private final Map 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); + } /** * 添加延迟任务 diff --git a/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java b/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java index 0d10ce8..0033791 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java +++ b/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java @@ -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 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 captureAndAnalyzeInterface(SpecificTimeRequest request, String timestamp, String interfaceName) { - // 检查目录 - if (AdvancedAsyncDownloader.createSingleDirectoryIfNotExists(properties.getTempPcapPath())) { - String pcapFileName = properties.getTempPcapPath() + timestamp + "_" + interfaceName + ".pcap"; - List ipCountList = new ArrayList<>(); - try { - // 第一步：执行tcpdump命令捕获数据 - ProcessBuilder captureBuilder = new ProcessBuilder(); - captureBuilder.command("timeout", "10", "tcpdump", "-i", interfaceName, - "-nn", "-n", "-w", pcapFileName); + private List 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 ipCountList = new ArrayList<>(); + + try { + // 第一步：执行tcpdump抓包（带超时控制） + CompletableFuture 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> 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 analyzePcapFile(String pcapFileName, String interfaceName) { + List 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<>(); } /** diff --git a/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java b/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java index 2a5af4d..019c15d 100644 --- a/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java +++ b/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java @@ -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() diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index 48e360c..9efad49 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -17,7 +17,7 @@ logging: netty: server: host: 120.211.95.173 - port: 56620 + port: 6620 client: client-id: client-001 reconnect-interval: 5 diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index f071c10..53534df 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -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