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

增加tcpdump探测

Browse Source
This commit is contained in:
gaoyutao
2026-01-22 17:57:23 +08:00
parent 421b7f1073
commit 057c879b59
11 changed files with 238 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java
+1
View File
@@ -18,4 +18,5 @@ public class ApplicationProperties {
private String frpPath;
private String poePath;
private String tempTrafficPath;
private String tempPcapPath;
}
src/main/java/com/tongran/agent/client/core/config/GlobalConfig.java
+4
View File
@@ -23,6 +23,10 @@ public class GlobalConfig {
* 业务网卡名称
*/
public static String NETNAME = "";
/**
* tcpdump探测时间
*/
public static String TCPDUMP_TIMES = "";
/**
* 服务启动时间
*/
src/main/java/com/tongran/agent/client/core/vo/NetVO.java
+2
View File
@@ -69,5 +69,7 @@ public class NetVO implements Serializable {
@Schema(description = "时间戳")
private long timestamp;
/** 是否最后一次 */
private boolean lastTrafficFlag;
}
src/main/java/com/tongran/agent/client/netty/model/Message.java
-2
View File
@@ -33,7 +33,5 @@ public class Message implements Serializable {
* 发送内容
*/
private String data;
/** 最后一次标志 */
private boolean lastTrafficFlag;
}
src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java
+2
View File
@@ -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 {
src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java
+2
View File
@@ -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 {
src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java
+166 -67
View File
@@ -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<IpCount> 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<IpCount> 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<IpCount> 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<IpCount> 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<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);
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<IpCount> mergeIpCounts(List<IpCount> allIpCountList) {
Map<String, Double> 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<IpCount> 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<IpCount> parseTcpdumpOutput(String output) {
List<IpCount> 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;
}
src/main/java/com/tongran/agent/client/service/AgentService.java
+2
View File
@@ -47,6 +47,8 @@ public interface AgentService {
void upFrpcMsg();
void handleRebootRecovery();
void checkTraffic();
void checkTcpdumpTimes();
List<MacVlanVO> upMacvlanStatus();
}
src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java
+57 -4
View File
@@ -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<NetVO> 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");
src/main/java/com/tongran/agent/client/service/impl/NetServiceImpl.java
+1 -1
View File
@@ -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")
};
src/main/resources/application.yml
+1
View File
@@ -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/