diff --git a/src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java b/src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java index b2160f9..5d5e1ec 100644 --- a/src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java +++ b/src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java @@ -18,4 +18,5 @@ public class ApplicationProperties { private String frpPath; private String poePath; private String tempTrafficPath; + private String tempPcapPath; } \ No newline at end of file diff --git a/src/main/java/com/tongran/agent/client/core/config/GlobalConfig.java b/src/main/java/com/tongran/agent/client/core/config/GlobalConfig.java index 47fa5cc..70e825e 100644 --- a/src/main/java/com/tongran/agent/client/core/config/GlobalConfig.java +++ b/src/main/java/com/tongran/agent/client/core/config/GlobalConfig.java @@ -23,6 +23,10 @@ public class GlobalConfig { * 业务网卡名称 */ public static String NETNAME = ""; + /** + * tcpdump探测时间 + */ + public static String TCPDUMP_TIMES = ""; /** * 服务启动时间 */ diff --git a/src/main/java/com/tongran/agent/client/core/vo/NetVO.java b/src/main/java/com/tongran/agent/client/core/vo/NetVO.java index e248188..fc40110 100644 --- a/src/main/java/com/tongran/agent/client/core/vo/NetVO.java +++ b/src/main/java/com/tongran/agent/client/core/vo/NetVO.java @@ -69,5 +69,7 @@ public class NetVO implements Serializable { @Schema(description = "时间戳") private long timestamp; + /** 是否最后一次 */ + private boolean lastTrafficFlag; } \ No newline at end of file diff --git a/src/main/java/com/tongran/agent/client/netty/model/Message.java b/src/main/java/com/tongran/agent/client/netty/model/Message.java index d755548..e676e2f 100644 --- a/src/main/java/com/tongran/agent/client/netty/model/Message.java +++ b/src/main/java/com/tongran/agent/client/netty/model/Message.java @@ -33,7 +33,5 @@ public class Message implements Serializable { * 发送内容 */ private String data; - /** 最后一次标志 */ - private boolean lastTrafficFlag; } 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 f919e19..c3024a4 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 @@ -109,6 +109,8 @@ public class AppInitializer implements CommandLineRunner { dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000); AssertLog.info("检测PppoE配置"); agentService.handleRebootRecovery(); + AssertLog.info("检测tcpdump探测时间配置"); + agentService.checkTcpdumpTimes(); }else{ //未注册,发送注册 try { 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 e076767..9d5d09e 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 @@ -880,6 +880,8 @@ public class BusinessTasks { dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000); AssertLog.info("检测PppoE配置"); agentService.handleRebootRecovery(); + AssertLog.info("检测tcpdump探测时间配置"); + agentService.checkTcpdumpTimes(); }else{ //未注册,发送注册 try { 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 7da71b8..0d10ce8 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 @@ -9,6 +9,7 @@ import com.tongran.agent.client.core.config.GlobalConfig; import com.tongran.agent.client.core.enums.MsgEnum; import com.tongran.agent.client.core.session.SessionManager; import com.tongran.agent.client.netty.model.Message; +import com.tongran.agent.client.scheduler.service.AdvancedAsyncDownloader; import com.tongran.agent.client.scheduler.service.AsyncCommandExecutor; import com.tongran.agent.client.service.AgentService; import com.tongran.agent.client.utils.AgentDataUtil; @@ -21,6 +22,7 @@ import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.io.BufferedReader; +import java.io.File; import java.io.IOException; import java.io.InputStreamReader; import java.time.LocalDateTime; @@ -28,6 +30,7 @@ import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; @Service public class SpecificTimeTaskService { @@ -101,80 +104,173 @@ public class SpecificTimeTaskService { private void executeTcpdump(SpecificTimeRequest request) { try { String timestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss")); - String pcapFileName = timestamp + ".pcap"; + List allIpCountList = new ArrayList<>(); - // 第一步:执行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); + // 获取业务网卡列表 + String businessNetName = GlobalConfig.NETNAME; + if (businessNetName == null || businessNetName.trim().isEmpty()) { + AssertLog.error("No business network interfaces configured"); return; } - // 等待文件写入完成 - Thread.sleep(1000); + String[] businessNetNameArr = businessNetName.split(";"); - // 第二步:分析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"); + // 遍历所有业务网卡进行抓包 + for (String interfaceName : businessNetNameArr) { + if (interfaceName.trim().isEmpty()) { + continue; + } - Process analyzeProcess = analyzeBuilder.start(); + String cleanInterfaceName = interfaceName.trim(); + AssertLog.info("Starting tcpdump on interface: {}", cleanInterfaceName); - // 读取分析结果 - BufferedReader reader = new BufferedReader(new InputStreamReader(analyzeProcess.getInputStream())); - StringBuilder result = new StringBuilder(); - String line; - - while ((line = reader.readLine()) != null) { - result.append(line).append("\n"); + try { + List interfaceResults = captureAndAnalyzeInterface(request, timestamp, cleanInterfaceName); + allIpCountList.addAll(interfaceResults); + } catch (Exception e) { + AssertLog.error("Failed to capture interface {}: {}", cleanInterfaceName, e.getMessage()); + // 继续处理其他网卡,不中断整个流程 + } } - int analyzeExitCode = analyzeProcess.waitFor(); + // 合并所有网卡的统计结果(按IP聚合) + List mergedResults = mergeIpCounts(allIpCountList); - if (analyzeExitCode == 0) { - AssertLog.info("Tcpdump analysis completed successfully for task: {}", request.getTaskId()); - AssertLog.info("IP statistics:\n{}", result.toString()); + // 保存合并后的结果 + saveTcpdumpResult(request, timestamp + "_merged", mergedResults); - // 解析并保存结果 - 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); + AssertLog.error("Error executing tcpdump task: {}", request.getTaskId(), e); } } /** - * 解析tcpdump输出为IP统计列表 + * 对单个网卡进行抓包和分析 + */ + 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); + + Process captureProcess = captureBuilder.start(); + int captureExitCode = captureProcess.waitFor(); + + if (captureExitCode != 0 && captureExitCode != 124) { + AssertLog.error("Tcpdump pcap包生成失败,网卡名称 {} with exit code: {}", interfaceName, captureExitCode); + return ipCountList; + } + + // 等待文件写入完成 + Thread.sleep(20000); + + // 检查文件是否存在且不为空 + 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; + } + + // 第二步:分析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"); + + 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.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; + } + return new ArrayList<>(); + } + + /** + * 合并多个网卡的IP统计结果(相同IP的数量相加) + */ + private List mergeIpCounts(List allIpCountList) { + Map ipCountMap = new HashMap<>(); + + for (IpCount ipCount : allIpCountList) { + String ip = ipCount.getIp(); + Double count = ipCount.getCount(); + + if (ipCountMap.containsKey(ip)) { + ipCountMap.put(ip, ipCountMap.get(ip) + count); + } else { + ipCountMap.put(ip, count); + } + } + + // 转换为列表并按数量降序排序 + return ipCountMap.entrySet().stream() + .map(entry -> new IpCount(entry.getKey(), entry.getValue())) + .sorted((a, b) -> Double.compare(b.getCount(), a.getCount())) // 降序 + .collect(Collectors.toList()); + } + + /** + * 保存合并后的结果 + */ + private void saveTcpdumpResult(SpecificTimeRequest request, String fileName, List ipCountList) { + try { + AssertLog.info("Saving merged 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包统计,共{}个IP", ipCountList.size()); + } + } catch (Exception e) { + AssertLog.error("Failed to save merged tcpdump result for task: {}", request.getTaskId(), e); + } + } + + /** + * 解析tcpdump输出为IP统计列表(优化版本) */ private List parseTcpdumpOutput(String output) { List ipCountList = new ArrayList<>(); @@ -194,15 +290,18 @@ public class SpecificTimeTaskService { } // 解析格式: " 455 172.16.15.52" - String[] parts = line.trim().split("\\s+", 2); // 按空格分割,最多分成2部分 + // 使用正则表达式更精确地匹配 + if (line.matches("^\\s*\\d+\\s+\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\s*$")) { + String[] parts = line.trim().split("\\s+", 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); + if (parts.length >= 2) { + try { + Double count = Double.valueOf(parts[0].trim()); + String ip = parts[1].trim(); + ipCountList.add(new IpCount(ip, count)); + } catch (NumberFormatException e) { + AssertLog.debug("Failed to parse count from line: {}", line); + } } } } @@ -218,7 +317,7 @@ public class SpecificTimeTaskService { @AllArgsConstructor public static class IpCount { private String ip; - private int count; + private Double count; } diff --git a/src/main/java/com/tongran/agent/client/service/AgentService.java b/src/main/java/com/tongran/agent/client/service/AgentService.java index e93e635..bf816bc 100644 --- a/src/main/java/com/tongran/agent/client/service/AgentService.java +++ b/src/main/java/com/tongran/agent/client/service/AgentService.java @@ -47,6 +47,8 @@ public interface AgentService { void upFrpcMsg(); void handleRebootRecovery(); void checkTraffic(); + + void checkTcpdumpTimes(); List upMacvlanStatus(); } 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 d1582d1..2a5af4d 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 @@ -2,6 +2,7 @@ package com.tongran.agent.client.service.impl; import cn.hutool.core.collection.CollectionUtil; import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.TypeReference; import com.tongran.agent.client.core.config.ApplicationProperties; @@ -10,6 +11,7 @@ import com.tongran.agent.client.core.enums.MsgEnum; import com.tongran.agent.client.core.eo.*; import com.tongran.agent.client.core.session.SessionManager; import com.tongran.agent.client.core.vo.MacVlanVO; +import com.tongran.agent.client.core.vo.NetVO; import com.tongran.agent.client.core.vo.NetworkInterfaceInfo; import com.tongran.agent.client.netty.MultiTargetNettyClient; import com.tongran.agent.client.netty.config.AgentNettyConfig; @@ -326,6 +328,8 @@ public class AgentServiceImpl implements AgentService { dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000); AssertLog.info("检测PppoE配置"); handleRebootRecovery(); + AssertLog.info("检测tcpdump探测时间配置"); + checkTcpdumpTimes(); } @Override @@ -926,8 +930,8 @@ public class AgentServiceImpl implements AgentService { // 删除对应文件 for (String fileName : fileNames) { if (StringUtils.isNotBlank(fileName)) { - fileName = fileName+"conf"; - String filePath = properties.getTempTrafficPath() + fileName.trim(); + fileName = fileName+".conf"; + String filePath = properties.getTempTrafficPath() + "/" + fileName.trim(); AssertLog.info("已处理文件:{}", fileName); deleteTrafficFile(filePath); } @@ -941,6 +945,23 @@ public class AgentServiceImpl implements AgentService { } private void handleTcpdump(String tcpdumpTimes) { + if (!tcpdumpTimes.equals(GlobalConfig.TCPDUMP_TIMES)) { + // 1. 将tcpdump探测时间写入配置文件 + String[] lines = { + "# tcpdump探测时间配置文件", + "# 生成时间: " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()), + "# tcpdump探测时间arr", + "tcpdumpTimes=" + tcpdumpTimes, + }; + + // 检查目录并写入配置文件 + if (AdvancedAsyncDownloader.createSingleDirectoryIfNotExists(properties.getConfPath())) { + AgentUtil.bufferedWriter(properties.getConfPath() + "/tcpdumptimes.conf", lines); + } + + // 更新全局变量 + GlobalConfig.TCPDUMP_TIMES = tcpdumpTimes; + } taskService.cancleTcpdumpSpecificTimeTask(); if(!"".equals(tcpdumpTimes)){ String[] tcpdumpTimeArr = tcpdumpTimes.split(","); @@ -2409,13 +2430,18 @@ public class AgentServiceImpl implements AgentService { props.load(reader); // 使用指定编码的Reader加载 String data = props.getProperty("traffic"); - if (StringUtils.isNotBlank(data)) { + List list = JSONArray.parseArray(data, NetVO.class); + if(list != null){ + for (NetVO netVO : list) { + netVO.setLastTrafficFlag(lastTraffic); + } + } + data = JSONObject.toJSONString(list); Message message = Message.builder() .clientId(GlobalConfig.CLIENT_ID) .dataType(MsgEnum.网络上报重试.getValue()) .data(data) - .lastTrafficFlag(lastTraffic) .build(); sessionManager.writeAndFlush( @@ -2562,7 +2588,34 @@ public class AgentServiceImpl implements AgentService { return null; } + @Override + public void checkTcpdumpTimes(){ + File tcpdumpTimesFile = new File(properties.getConfPath() + "/tcpdumptimes.conf"); + if(tcpdumpTimesFile.exists()){ + Properties props = new Properties(); + try (InputStream input = Files.newInputStream(tcpdumpTimesFile.toPath())) { + props.load(input); + String tcpdumpTimes = props.getProperty("tcpdumpTimes"); + 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); + } + AssertLog.info("应用启动时加载tcpdump探测时间配置成功: {}", tcpdumpTimes); + } + + } catch (Exception e) { + AssertLog.error("应用启动时加载tcpdump探测时间配置成功", e); + } + } + } @Override public void checkAgentUpdate() { File agentUpdateFile = new File(properties.getConfPath() + "/agentupdate.conf"); diff --git a/src/main/java/com/tongran/agent/client/service/impl/NetServiceImpl.java b/src/main/java/com/tongran/agent/client/service/impl/NetServiceImpl.java index 88c2e62..48d8599 100644 --- a/src/main/java/com/tongran/agent/client/service/impl/NetServiceImpl.java +++ b/src/main/java/com/tongran/agent/client/service/impl/NetServiceImpl.java @@ -149,7 +149,7 @@ public class NetServiceImpl implements NetService { private Double getPingLossRateSync(String interfaceName) { String[] cmd = { "/bin/sh", "-c", - String.format("ping -I %s -c 290 -i 1 -W 1 -w 290 %s 2>&1 | grep 'packet loss' | awk -F'[ ,%]+' '{print $6}'", + String.format("ping -I %s -c 290 -i 1 -W 1 -w 290 %s 2>&1 | grep 'packet loss' | awk -F'[ ,%%]+' '{print $6}'", interfaceName, "223.5.5.5") }; diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 1be839a..f071c10 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -14,6 +14,7 @@ spring: frp-path: /usr/local/tongran/frp poe-path: /usr/local/tongran/sbin/pppoe1 temp-traffic-path: /usr/local/tongran/traffictemp + temp-pcap-path: /usr/local/tongran/pcaptemp/ web: resources: static-locations: classpath*:/META-INF/resources/