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

cgroup+iptables获取业务流量方法优化为区分多个进程模式。

Browse Source
This commit is contained in:
gaoyutao
2026-03-17 18:00:03 +08:00
parent 2ba1134c82
commit 30d44561d1
2 changed files with 451 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/tongran/agent/client/core/vo/NetBusinessVO.java
+11
View File
@@ -23,6 +23,17 @@ public class NetBusinessVO implements Serializable {
private Long inSpeed; // 累计接收字节数
private Long outSpeed; // 累计发送字节数
/** IPv4接收流量 */
private Long ipv4InSpeed;
/** IPv4发送流量 */
private Long ipv4OutSpeed;
/** IPv6接收流量 */
private Long ipv6InSpeed;
/** IPv6发送流量 */
private Long ipv6OutSpeed;
private int connectionCount; // 当前连接数
src/main/java/com/tongran/agent/client/service/impl/NetBusinessServiceImpl.java
+440 -44
View File
@@ -1,96 +1,492 @@
package com.tongran.agent.client.service.impl;
import com.tongran.agent.client.core.config.ApplicationProperties;
import com.tongran.agent.client.core.vo.NetBusinessVO;
import com.tongran.agent.client.service.NetBusinessService;
import com.tongran.agent.client.utils.AssertLog;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.HashSet;
import java.util.Set;
@Service
public class NetBusinessServiceImpl implements NetBusinessService {
private static final String CGROUP_BASE = "/sys/fs/cgroup/net_cls";
private static final String CGROUP_PREFIX = "agent_";
private static final int MAX_CHAIN_NAME_LEN = 28;
// 存储每个进程的 classid
private final ConcurrentHashMap<String, Long> processClassIdMap = new ConcurrentHashMap<>();
// 存储每个进程的 cgroup 路径
private final ConcurrentHashMap<String, String> processCgroupPathMap = new ConcurrentHashMap<>();
// 记录当前所有已创建的进程（无论是否有进程在运行）
private final Set<String> activeProcesses = ConcurrentHashMap.newKeySet();
private List<String> processList = new ArrayList<>();
@Resource
private ApplicationProperties properties;
@PostConstruct
public void init() {
try {
loadProcessListFromFile();
// 为配置文件中有的进程创建 cgroup 和 iptables 链
for (String processName : processList) {
createProcessCgroup(processName);
initProcessChains("iptables", processName);
initProcessChains("ip6tables", processName);
activeProcesses.add(processName);
}
System.out.println("=== cgroup多进程监控初始化完成 ===");
System.out.println("监控进程列表: " + processList);
} catch (Exception e) {
System.err.println("初始化失败: " + e.getMessage());
}
}
@PreDestroy
public void cleanup() {
try {
// 清理所有已创建的规则和 cgroup
for (String processName : activeProcesses) {
cleanupProcessRules("iptables", processName);
cleanupProcessRules("ip6tables", processName);
removeProcessCgroup(processName);
}
System.out.println("已清理所有 iptables 规则和 cgroup");
} catch (Exception e) {
System.err.println("清理失败: " + e.getMessage());
}
}
@Override
public List<NetBusinessVO> netList(long timestamp) {
List<NetBusinessVO> list = new ArrayList<>();
List<NetBusinessVO> resultList = new ArrayList<>();
try {
// 1. 获取连接数最多的进程PID
Integer pid = getTopConnectionPid();
if (pid == null) return list;
// 保存旧的进程列表用于对比
List<String> oldProcessList = new ArrayList<>(processList);
// 2. 读取进程网络统计
List<String> lines = Files.readAllLines(Paths.get("/proc/" + pid + "/net/dev"));
// 重新加载配置文件
loadProcessListFromFile();
System.out.println("\n===== 进程 " + pid + " 所有网卡流量 =====");
for (String line : lines) {
line = line.trim();
if (line.isEmpty() || line.startsWith("Inter") || line.startsWith("face")) continue;
// 找出被剔除的进程（在旧列表中存在，在新列表中不存在）
Set<String> removedProcesses = new HashSet<>(oldProcessList);
removedProcesses.removeAll(processList);
String[] parts = line.split("\\s+");
if (parts.length >= 10) {
String iface = parts[0].replace(":", "");
if(iface.startsWith("lo")) continue;
long recv = Long.parseLong(parts[1]); // 接收字节
long sent = Long.parseLong(parts[9]); // 发送字节
// 清理被剔除的进程（只有这些才清理）
if (!removedProcesses.isEmpty()) {
System.out.println("检测到配置文件中已删除的进程: " + removedProcesses + "，开始清理...");
for (String processName : removedProcesses) {
cleanupProcessRules("iptables", processName);
cleanupProcessRules("ip6tables", processName);
removeProcessCgroup(processName);
activeProcesses.remove(processName);
System.out.println("已清理进程 " + processName + " 的所有规则");
}
}
list.add(NetBusinessVO.builder()
.name(iface)
.pid(pid)
.processName(Files.readString(Paths.get("/proc/" + pid + "/comm")).trim())
.inSpeed(recv)
.outSpeed(sent)
.connectionCount(getProcessConnectionCount(pid))
.timestamp(timestamp)
.build());
// 打印每个网卡的流量（方便调试）
System.out.printf("%-15s 接收: %-20d 发送: %-20d%n", iface, recv, sent);
// 找出新增的进程（在新列表中存在，在旧列表中不存在）
Set<String> addedProcesses = new HashSet<>(processList);
addedProcesses.removeAll(oldProcessList);
// 为新增的进程创建规则（但只创建有PID的）
if (!addedProcesses.isEmpty()) {
System.out.println("检测到新增的进程: " + addedProcesses);
for (String processName : addedProcesses) {
List<Integer> pids = getProcessPids(processName);
if (!pids.isEmpty()) {
// 只有有PID的才创建规则
createProcessCgroup(processName);
initProcessChains("iptables", processName);
initProcessChains("ip6tables", processName);
addProcessToCgroup(processName, pids);
activeProcesses.add(processName);
System.out.println("已为进程 " + processName + " (PID: " + pids + ") 创建规则");
} else {
System.out.println("进程 " + processName + " 当前无PID，暂不创建规则");
}
}
}
// 遍历当前配置文件中的进程列表获取统计信息
for (String processName : processList) {
List<Integer> pids = getProcessPids(processName);
// 如果有PID，确保规则存在（可能之前无PID时没创建，现在需要创建）
if (!pids.isEmpty()) {
// 检查是否已有规则，如果没有则需要创建
if (!activeProcesses.contains(processName)) {
System.out.println("进程 " + processName + " 已有PID运行但无规则，开始创建规则...");
createProcessCgroup(processName);
initProcessChains("iptables", processName);
initProcessChains("ip6tables", processName);
activeProcesses.add(processName);
}
// 确保进程在cgroup中
addProcessToCgroup(processName, pids);
// 确保链存在
ensureProcessChainsExist("iptables", processName);
ensureProcessChainsExist("ip6tables", processName);
// 获取该进程的流量统计
long[] v4Stats = getTrafficStatsForProcess("iptables", processName);
long[] v6Stats = getTrafficStatsForProcess("ip6tables", processName);
NetBusinessVO vo = new NetBusinessVO();
vo.setIpv4InSpeed(v4Stats[0]);
vo.setIpv4OutSpeed(v4Stats[1]);
vo.setIpv6InSpeed(v6Stats[0]);
vo.setIpv6OutSpeed(v6Stats[1]);
vo.setInSpeed(v4Stats[0] + v6Stats[0]);
vo.setOutSpeed(v4Stats[1] + v6Stats[1]);
vo.setTimestamp(timestamp);
vo.setName("total");
vo.setProcessName(processName);
resultList.add(vo);
if (v4Stats[0] > 0 || v4Stats[1] > 0 || v6Stats[0] > 0 || v6Stats[1] > 0) {
System.out.printf("进程 %s (PID: %s): IPv4收=%d 发=%d, IPv6收=%d 发=%d%n",
processName, pids, v4Stats[0], v4Stats[1], v6Stats[0], v6Stats[1]);
}
} else {
// 没有PID，但规则保留（因为配置还在）
System.out.println("进程 " + processName + " 当前无PID运行，保留已有规则等待进程启动");
}
}
} catch (Exception e) {
e.printStackTrace();
System.err.println("获取网络统计失败: " + e.getMessage());
}
return list;
return resultList;
}
/**
* 获取连接数最多的进程PID
* 为进程生成唯一的 classid
*/
private Integer getTopConnectionPid() {
private long generateClassId(String processName) {
// classid 必须在 0x00000001 - 0xFFFFFFFF 之间
// 使用进程名的hash，但限制在 0x1 - 0xFFFFF 之间
int hash = Math.abs(processName.hashCode()) & 0xFFFFF;
if (hash == 0) hash = 1; // 避免0值
return hash & 0xFFFFFFFFL; // 确保在32位无符号范围内
}
/**
* 为进程创建独立的 cgroup
*/
private void createProcessCgroup(String processName) {
try {
long classId = generateClassId(processName);
// 使用原始的 processName 来构建路径，不要用 safeName
String cgroupPath = CGROUP_BASE + "/" + CGROUP_PREFIX + processName;
executeCommand("mkdir -p " + cgroupPath);
executeCommand("echo " + classId + " > " + cgroupPath + "/net_cls.classid");
processClassIdMap.put(processName, classId);
processCgroupPathMap.put(processName, cgroupPath);
System.out.println("进程 " + processName + " cgroup 创建完成, classid=" + classId + ", path=" + cgroupPath);
} catch (Exception e) {
System.err.println("创建进程 cgroup 失败 " + processName + ": " + e.getMessage());
}
}
/**
* 删除进程的 cgroup
*/
private void removeProcessCgroup(String processName) {
try {
String cgroupPath = processCgroupPathMap.get(processName);
if (cgroupPath != null) {
// 先移动所有进程到根 cgroup
executeCommand("cat " + cgroupPath + "/tasks | while read pid; do echo $pid > " + CGROUP_BASE + "/tasks 2>/dev/null; done");
// 删除目录
executeCommand("rmdir " + cgroupPath + " 2>/dev/null");
processClassIdMap.remove(processName);
processCgroupPathMap.remove(processName);
System.out.println("已删除进程 " + processName + " 的 cgroup");
}
} catch (Exception e) {
System.err.println("删除进程 cgroup 失败 " + processName + ": " + e.getMessage());
}
}
private void loadProcessListFromFile() {
List<String> newProcessList = new ArrayList<>();
String path = properties.getConfPath() + "/process.conf";
File file = new File(path);
if (file.exists()) {
try (BufferedReader br = new BufferedReader(new FileReader(file))) {
String line;
while ((line = br.readLine()) != null) {
line = line.trim();
if (!line.isEmpty() && !line.startsWith("#")) {
newProcessList.add(line);
}
}
System.out.println("从 " + path + " 加载进程列表: " + newProcessList);
} catch (Exception e) {
System.err.println("读取配置文件失败 " + path + ": " + e.getMessage());
}
}
// 更新 processList
synchronized (this) {
processList.clear();
processList.addAll(newProcessList);
}
}
private List<Integer> getProcessPids(String processName) {
List<Integer> pids = new ArrayList<>();
try {
ProcessBuilder pb = new ProcessBuilder("sh", "-c",
"ss -tunap | awk '{print $7}' | grep -o 'pid=[0-9]*' | sort | uniq -c | sort -nr | head -1 | grep -o 'pid=[0-9]*' | cut -d= -f2");
"pgrep -x " + processName + " 2>/dev/null");
Process p = pb.start();
try (BufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream()))) {
String line = br.readLine();
return line != null ? Integer.parseInt(line.trim()) : null;
String line;
while ((line = br.readLine()) != null) {
line = line.trim();
if (!line.isEmpty()) {
pids.add(Integer.parseInt(line));
}
}
}
} catch (Exception e) {
return null;
// 忽略
}
return pids;
}
private void addProcessToCgroup(String processName, List<Integer> pids) {
String cgroupPath = processCgroupPathMap.get(processName);
if (cgroupPath == null) return;
for (Integer pid : pids) {
try {
ProcessBuilder checkPb = new ProcessBuilder("sh", "-c",
"cat " + cgroupPath + "/tasks 2>/dev/null | grep -q " + pid);
if (checkPb.start().waitFor() != 0) {
executeCommand("echo " + pid + " > " + cgroupPath + "/tasks");
System.out.println("进程 " + pid + "(" + processName + ") 已加入 cgroup");
}
} catch (Exception e) {
// 忽略
}
}
}
/**
* 获取进程的连接数
*/
private int getProcessConnectionCount(int pid) {
private String generateSafeChainName(String processName) {
String safe = processName.replaceAll("[^a-zA-Z0-9]", "_");
String hash = Integer.toHexString(processName.hashCode());
while (hash.length() < 8) {
hash = "0" + hash;
}
if (hash.length() > 8) {
hash = hash.substring(0, 8);
}
String prefix = safe.length() > 12 ? safe.substring(0, 12) : safe;
return prefix + "_" + hash;
}
private String buildChainName(String safeName, String suffix, String direction) {
// 先为 in 和 out 计算一个公共的 baseName（取较短的那个）
String baseName = "am_" + safeName;
// 计算 in 和 out 分别需要的长度
int inLen = baseName.length() + 6; // _v4_in
int outLen = baseName.length() + 7; // _v4_out
// 取较长的作为标准，确保两者都满足
int maxNeeded = Math.max(inLen, outLen);
if (maxNeeded > MAX_CHAIN_NAME_LEN) {
// 需要截断 baseName
int baseMaxLen = MAX_CHAIN_NAME_LEN - 7; // 按最长的 out 算
if (baseName.length() > baseMaxLen) {
String prefix = baseName.substring(0, 8);
String suffix_part = baseName.substring(baseName.length() - (baseMaxLen - 9));
baseName = prefix + "_" + suffix_part;
}
}
return baseName + "_" + suffix + "_" + direction;
}
private long getProcessClassId(String processName) {
Long classId = processClassIdMap.get(processName);
if (classId == null) {
classId = generateClassId(processName);
processClassIdMap.put(processName, classId);
}
return classId;
}
private void initProcessChains(String command, String processName) {
long classId = getProcessClassId(processName);
String safeName = generateSafeChainName(processName);
String suffix = command.equals("iptables") ? "v4" : "v6";
String inChain = buildChainName(safeName, suffix, "in");
String outChain = buildChainName(safeName, suffix, "out");
initChain(command, inChain, "INPUT", classId);
initChain(command, outChain, "OUTPUT", classId);
}
private void ensureProcessChainsExist(String command, String processName) {
long classId = getProcessClassId(processName);
String safeName = generateSafeChainName(processName);
String suffix = command.equals("iptables") ? "v4" : "v6";
String inChain = buildChainName(safeName, suffix, "in");
String outChain = buildChainName(safeName, suffix, "out");
ensureChainExists(command, inChain, "INPUT", classId);
ensureChainExists(command, outChain, "OUTPUT", classId);
}
private void ensureChainExists(String command, String chainName, String hookChain, long classId) {
try {
ProcessBuilder pb = new ProcessBuilder("sh", "-c", "ss -tunap | grep 'pid=" + pid + "' | wc -l");
ProcessBuilder checkPb = new ProcessBuilder("sh", "-c",
command + " -L " + chainName + " -n >/dev/null 2>&1");
if (checkPb.start().waitFor() != 0) {
executeCommand(command + " -N " + chainName);
executeCommand(command + " -A " + chainName +
" -m cgroup --cgroup " + classId + " -j RETURN");
executeCommand(command + " -A " + chainName + " -j RETURN");
executeCommand(command + " -I " + hookChain + " 1 -m cgroup --cgroup " +
classId + " -j " + chainName);
System.out.println(command + " 链 " + chainName + " 初始化完成 (classid=" + classId + ")");
}
} catch (Exception e) {
System.err.println("确保链存在失败 " + chainName + ": " + e.getMessage());
}
}
private void initChain(String command, String chainName, String hookChain, long classId) {
try {
// 1. 直接创建链（如果已存在会报错，但忽略）
executeCommand(command + " -N " + chainName + " 2>/dev/null");
// 2. 清空并添加规则
executeCommand(command + " -F " + chainName);
executeCommand(command + " -A " + chainName +
" -m cgroup --cgroup " + classId + " -j RETURN");
executeCommand(command + " -A " + chainName + " -j RETURN");
// 3. 检查是否已存在相同的规则
String checkRuleCmd = command + " -C " + hookChain + " -m cgroup --cgroup " +
classId + " -j " + chainName + " 2>/dev/null";
ProcessBuilder checkPb = new ProcessBuilder("sh", "-c", checkRuleCmd);
if (checkPb.start().waitFor() != 0) {
// 规则不存在，才插入
executeCommand(command + " -I " + hookChain + " 1 -m cgroup --cgroup " +
classId + " -j " + chainName);
System.out.println(command + " 插入规则到 " + hookChain + " 成功");
} else {
System.out.println(command + " 规则已存在于 " + hookChain + "，跳过插入");
}
System.out.println(command + " 链 " + chainName + " 初始化完成 (classid=" + classId + ")");
} catch (Exception e) {
System.err.println("初始化链 " + chainName + " 失败: " + e.getMessage());
}
}
private long[] getTrafficStatsForProcess(String command, String processName) {
long classId = getProcessClassId(processName);
String safeName = generateSafeChainName(processName);
String suffix = command.equals("iptables") ? "v4" : "v6";
String inChain = buildChainName(safeName, suffix, "in");
String outChain = buildChainName(safeName, suffix, "out");
long recvBytes = getChainBytes(command, inChain);
long sentBytes = getChainBytes(command, outChain);
return new long[]{recvBytes, sentBytes};
}
private void cleanupProcessRules(String command, String processName) {
try {
Long classId = processClassIdMap.get(processName);
if (classId == null) return;
String safeName = generateSafeChainName(processName);
String suffix = command.equals("iptables") ? "v4" : "v6";
String inChain = buildChainName(safeName, suffix, "in");
String outChain = buildChainName(safeName, suffix, "out");
executeCommand(command + " -D INPUT -m cgroup --cgroup " + classId +
" -j " + inChain + " 2>/dev/null");
executeCommand(command + " -D OUTPUT -m cgroup --cgroup " + classId +
" -j " + outChain + " 2>/dev/null");
executeCommand(command + " -F " + inChain + " 2>/dev/null");
executeCommand(command + " -F " + outChain + " 2>/dev/null");
executeCommand(command + " -X " + inChain + " 2>/dev/null");
executeCommand(command + " -X " + outChain + " 2>/dev/null");
System.out.println("已清理 " + command + " 中进程 " + processName + " 的规则");
} catch (Exception e) {
System.err.println("清理进程规则失败 " + processName + ": " + e.getMessage());
}
}
private long getChainBytes(String command, String chainName) {
try {
ProcessBuilder pb = new ProcessBuilder("sh", "-c",
command + " -L " + chainName + " -v -n -x | grep RETURN | head -1");
Process p = pb.start();
try (BufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream()))) {
String line = br.readLine();
return line != null ? Integer.parseInt(line.trim()) : 0;
if (line != null && !line.isEmpty()) {
String[] parts = line.trim().split("\\s+");
if (parts.length >= 2) {
return Long.parseLong(parts[1]);
}
}
}
} catch (Exception e) {
return 0;
System.err.println("获取链 " + chainName + " 统计失败: " + e.getMessage());
}
return 0;
}
private void executeCommand(String command) throws Exception {
ProcessBuilder pb = new ProcessBuilder("sh", "-c", command);
Process p = pb.start();
int exitCode = p.waitFor();
if (exitCode != 0) {
try (BufferedReader br = new BufferedReader(new InputStreamReader(p.getErrorStream()))) {
String error = br.lines().reduce("", (a, b) -> a + "\n" + b);
if (!error.isEmpty() && !error.contains("No such file") && !error.contains("File exists")) {
System.err.println("命令执行警告: " + command + " -> " + error);
}
}
}
}
}