diff --git a/src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java b/src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java index 22ec1ec..f9f6098 100644 --- a/src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java +++ b/src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java @@ -75,6 +75,8 @@ public enum MsgEnum { iops结果上报("IOPS_RESULT"), + tcpdump结果上报("TCPDUMP_RESULT"), + 多网IP探测上报("NETWORK_DETECT"); private String value; diff --git a/src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java b/src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java index a2440d1..f919e19 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java +++ b/src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java @@ -105,8 +105,8 @@ public class AppInitializer implements CommandLineRunner { AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000); dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000); long milliFive = AgentUtil.millisecondsToNext5Minute(); - AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000); - dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000); + AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 7200000); + dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000); AssertLog.info("检测PppoE配置"); agentService.handleRebootRecovery(); }else{ diff --git a/src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java b/src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java index 3007a38..e076767 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java +++ b/src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java @@ -876,8 +876,8 @@ public class BusinessTasks { AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000); dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000); long milliFive = AgentUtil.millisecondsToNext5Minute(); - AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000); - dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000); + AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 7200000); + dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000); AssertLog.info("检测PppoE配置"); agentService.handleRebootRecovery(); }else{ 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 9410fba..7da71b8 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 @@ -1,5 +1,8 @@ package com.tongran.agent.client.scheduler.task; +import cn.hutool.core.collection.CollectionUtil; +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.tongran.agent.client.core.config.ApplicationProperties; import com.tongran.agent.client.core.config.GlobalConfig; @@ -11,15 +14,18 @@ import com.tongran.agent.client.service.AgentService; import com.tongran.agent.client.utils.AgentDataUtil; import com.tongran.agent.client.utils.AgentUtil; import com.tongran.agent.client.utils.AssertLog; +import lombok.AllArgsConstructor; +import lombok.Data; import org.apache.commons.lang3.StringUtils; import org.springframework.stereotype.Service; import javax.annotation.Resource; +import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStreamReader; import java.time.LocalDateTime; -import java.util.Arrays; -import java.util.List; -import java.util.Objects; +import java.time.format.DateTimeFormatter; +import java.util.*; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -40,6 +46,20 @@ public class SpecificTimeTaskService { this.sessionManager = SessionManager.getInstance(); } + public boolean createTcpdumpSpecificTimeTask(SpecificTimeRequest request){ + String taskId = request.getTaskId(); + Runnable task = createTcpdumpTaskRunnable(request); + try { + if (request.getTime() != null) { + taskConfig.addDailyTimeTask(taskId, task, request.getTime()); + } else { + throw new IllegalArgumentException("必须指定一种时间方式"); + } + return true; + } catch (Exception e) { + throw new RuntimeException("创建任务失败: " + e.getMessage(), e); + } + } /** * 创建指定时间任务 */ @@ -67,13 +87,148 @@ public class SpecificTimeTaskService { } } + private Runnable createTcpdumpTaskRunnable(SpecificTimeRequest request) { + return () -> { + System.out.println("执行定时任务: " + request.getTaskName()); + System.out.println("执行时间: " + LocalDateTime.now()); + System.out.println("-----------------------------------"); + + // 具体的业务逻辑 + executeTcpdump(request); + }; + } + + private void executeTcpdump(SpecificTimeRequest request) { + try { + String timestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss")); + String pcapFileName = timestamp + ".pcap"; + + // 第一步：执行tcpdump命令捕获数据 + ProcessBuilder captureBuilder = new ProcessBuilder(); + captureBuilder.command("timeout", "10", "tcpdump", "-i", "eth0", "-nn", "-n", "-w", pcapFileName); + + Process captureProcess = captureBuilder.start(); + int captureExitCode = captureProcess.waitFor(); + + if (captureExitCode != 0) { + AssertLog.error("Tcpdump capture failed with exit code: {}", captureExitCode); + return; + } + + // 等待文件写入完成 + Thread.sleep(1000); + + // 第二步：分析pcap文件并统计目标IP + ProcessBuilder analyzeBuilder = new ProcessBuilder(); + analyzeBuilder.command("bash", "-c", + "tcpdump -r " + pcapFileName + " -nn | awk '/IP/{split($5,a,\".\"); print a[1]\".\"a[2]\".\"a[3]\".\"a[4]}' | sort | uniq -c | sort -rn"); + + Process analyzeProcess = analyzeBuilder.start(); + + // 读取分析结果 + 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.info("Tcpdump analysis completed successfully for task: {}", request.getTaskId()); + AssertLog.info("IP statistics:\n{}", result.toString()); + + // 解析并保存结果 + saveTcpdumpResult(request, pcapFileName, result.toString()); + } else { + AssertLog.error("Tcpdump analysis failed with exit code: {}", analyzeExitCode); + } + + } catch (IOException | InterruptedException e) { + AssertLog.error("Error executing tcpdump task: {}", request.getTaskId(), e); + Thread.currentThread().interrupt(); + } + } + + private void saveTcpdumpResult(SpecificTimeRequest request, String fileName, String statistics) { + try { + // 解析统计结果 + List ipCountList = parseTcpdumpOutput(statistics); + + AssertLog.info("Saving tcpdump result - Task: {}, File: {}, Total unique IPs: {}", + request.getTaskId(), fileName, ipCountList.size()); + String data = ""; + if(CollectionUtil.isNotEmpty(ipCountList)){ + data = JSONArray.toJSONString(ipCountList); + } + Message message = Message.builder().clientId(request.getClientId()).dataType(MsgEnum.tcpdump结果上报.getValue()).data(data).build(); + if (Objects.nonNull(sessionManager.getSessionById(request.getClientId()))) { + sessionManager.writeAndFlush(sessionManager.getSessionById(request.getClientId()).getChannel(), message); + AssertLog.info("发送tcpdump包={}", JSON.toJSONString(message)); + } + } catch (Exception e) { + AssertLog.error("Failed to parse tcpdump result for task: {}", request.getTaskId(), e); + } + } + + /** + * 解析tcpdump输出为IP统计列表 + */ + private List parseTcpdumpOutput(String output) { + List ipCountList = new ArrayList<>(); + + if (output == null || output.trim().isEmpty()) { + return ipCountList; + } + + String[] lines = output.split("\n"); + + for (String line : lines) { + line = line.trim(); + + // 跳过空行和标题行 + if (line.isEmpty() || line.contains("reading from file") || line.contains("link-type")) { + continue; + } + + // 解析格式： " 455 172.16.15.52" + String[] parts = line.trim().split("\\s+", 2); // 按空格分割，最多分成2部分 + + if (parts.length >= 2) { + try { + int count = Integer.parseInt(parts[0].trim()); + String ip = parts[1].trim(); + ipCountList.add(new IpCount(ip, count)); + } catch (NumberFormatException e) { + AssertLog.warn("Failed to parse count from line: {}", line); + } + } + } + + return ipCountList; + } + + + /** + * IP统计实体类 + */ + @Data + @AllArgsConstructor + public static class IpCount { + private String ip; + private int count; + } + + private Runnable createTaskRunnable(SpecificTimeRequest request) { return () -> { System.out.println("执行定时任务: " + request.getTaskName()); System.out.println("任务数据: " + request.getTaskData()); System.out.println("执行时间: " + LocalDateTime.now()); System.out.println("-----------------------------------"); - + // 具体的业务逻辑 executeBusinessLogic(request); }; @@ -153,4 +308,16 @@ public class SpecificTimeTaskService { System.err.println("任务执行异常: " + e.getMessage()); } } + + public void cancleTcpdumpSpecificTimeTask() { + // 获取所有任务ID + Set taskIds = taskConfig.getTaskMap().keySet(); + + // 遍历并关闭所有以"tcpdump"开头的任务 + for (String taskId : taskIds) { + if (taskId.startsWith("tcpdump")) { + taskConfig.removeTask(taskId); + } + } + } } \ No newline at end of file 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 b9cf5d4..d1582d1 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 @@ -322,8 +322,8 @@ public class AgentServiceImpl implements AgentService { AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000); dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000); long milliFive = AgentUtil.millisecondsToNext5Minute(); - AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000); - dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000); + AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 7200000); + dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000); AssertLog.info("检测PppoE配置"); handleRebootRecovery(); } @@ -920,21 +920,42 @@ public class AgentServiceImpl implements AgentService { } if(jsonObject.containsKey("trafficUsedTimestamp")){ String trafficUsedTimestamp = jsonObject.getString("trafficUsedTimestamp"); -// if (trafficUsedTimestamp != null) { -// String[] fileNames = trafficUsedTimestamp.split(","); -// -// // 删除对应文件 -// for (String fileName : fileNames) { -// if (StringUtils.isNotBlank(fileName)) { -// fileName = fileName+"conf"; -// String filePath = properties.getTempTrafficPath() + fileName.trim(); -// AssertLog.info("已处理文件：{}", fileName); -// deleteTrafficFile(filePath); -// } -// } -// } + if (trafficUsedTimestamp != null) { + String[] fileNames = trafficUsedTimestamp.split(","); + + // 删除对应文件 + for (String fileName : fileNames) { + if (StringUtils.isNotBlank(fileName)) { + fileName = fileName+"conf"; + String filePath = properties.getTempTrafficPath() + fileName.trim(); + AssertLog.info("已处理文件：{}", fileName); + deleteTrafficFile(filePath); + } + } + } + } + if(jsonObject.containsKey("tcpdumpTimes")){ + String tcpdumpTimes = jsonObject.getString("tcpdumpTimes"); + handleTcpdump(tcpdumpTimes); } } + + private void handleTcpdump(String tcpdumpTimes) { + taskService.cancleTcpdumpSpecificTimeTask(); + if(!"".equals(tcpdumpTimes)){ + String[] tcpdumpTimeArr = tcpdumpTimes.split(","); + for (String tcpdumpTime : tcpdumpTimeArr) { + SpecificTimeRequest request = SpecificTimeRequest.builder() + .taskId("tcpdump"+tcpdumpTime) + .taskName("tcpdump"+tcpdumpTime) + .time(tcpdumpTime) + .clientId(GlobalConfig.CLIENT_ID) + .build(); + taskService.createTcpdumpSpecificTimeTask(request); + } + } + } + private void deleteTrafficFile(String filePath) { try { File file = new File(filePath); @@ -2314,8 +2335,8 @@ public class AgentServiceImpl implements AgentService { // 配置参数 int batchSize = 10; // 每批10个文件 - long intraBatchInterval = 5000; // 批次内文件间隔5s - long interBatchInterval = 1000; // 批次间间隔10秒 + long intraBatchInterval = 1000; // 批次内文件间隔5s + long interBatchInterval = 1000; // 批次间间隔1秒 int totalProcessed = 0; int totalFiles = fileList.size();