fix: 新增 NetBusinessServiceImpl 实现类，修复 Spring Bean 注入失败
问题：BusinessTasks依赖注入NetBusinessService，但缺少@Service实现类 导致启动报错 NoSuchBeanDefinitionException 解决：从1.20反编译代码恢复 NetBusinessServiceImpl - 多进程 cgroup + iptables 网络流量监控 - System.out改为 SLF4J Logger - @Service注解确保Spring自动装配
This commit is contained in:
lee
@@ -0,0 +1,370 @@
|
||||
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 lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import javax.annotation.PreDestroy;
|
||||
import javax.annotation.Resource;
|
||||
import java.io.*;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
@Slf4j
|
||||
@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 final ConcurrentHashMap<String, Long> processClassIdMap = new ConcurrentHashMap<>();
|
||||
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 {
|
||||
this.loadProcessListFromFile();
|
||||
for (String processName : this.processList) {
|
||||
List<Integer> pids = this.getProcessPids(processName);
|
||||
if (!pids.isEmpty()) {
|
||||
this.createProcessCgroup(processName);
|
||||
this.initProcessChains("iptables", processName);
|
||||
this.initProcessChains("ip6tables", processName);
|
||||
this.addProcessToCgroup(processName, pids);
|
||||
this.activeProcesses.add(processName);
|
||||
log.info("进程 {} (PID: {}) 初始化完成", processName, pids);
|
||||
} else {
|
||||
log.info("进程 {} 当前无PID运行，跳过初始化", processName);
|
||||
}
|
||||
}
|
||||
log.info("=== cgroup多进程监控初始化完成, 监控进程列表: {} ===", this.processList);
|
||||
} catch (Exception e) {
|
||||
log.error("初始化失败: {}", e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
public void cleanup() {
|
||||
try {
|
||||
for (String processName : this.activeProcesses) {
|
||||
this.cleanupProcessRules("iptables", processName);
|
||||
this.cleanupProcessRules("ip6tables", processName);
|
||||
this.removeProcessCgroup(processName);
|
||||
}
|
||||
log.info("已清理所有 iptables 规则和 cgroup");
|
||||
} catch (Exception e) {
|
||||
log.error("清理失败: {}", e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<NetBusinessVO> netList(long timestamp) {
|
||||
List<NetBusinessVO> resultList = new ArrayList<>();
|
||||
try {
|
||||
List<String> oldProcessList = new ArrayList<>(this.processList);
|
||||
this.loadProcessListFromFile();
|
||||
|
||||
Set<String> removedProcesses = new HashSet<>(oldProcessList);
|
||||
removedProcesses.removeAll(this.processList);
|
||||
if (!removedProcesses.isEmpty()) {
|
||||
log.info("检测到配置文件中已删除的进程: {}，开始清理...", removedProcesses);
|
||||
for (String processName : removedProcesses) {
|
||||
this.cleanupProcessRules("iptables", processName);
|
||||
this.cleanupProcessRules("ip6tables", processName);
|
||||
this.removeProcessCgroup(processName);
|
||||
this.activeProcesses.remove(processName);
|
||||
log.info("已清理进程 {} 的所有规则", processName);
|
||||
}
|
||||
}
|
||||
|
||||
Set<String> addedProcesses = new HashSet<>(this.processList);
|
||||
addedProcesses.removeAll(oldProcessList);
|
||||
if (!addedProcesses.isEmpty()) {
|
||||
log.info("检测到新增的进程: {}", addedProcesses);
|
||||
for (String processName : addedProcesses) {
|
||||
List<Integer> pids = this.getProcessPids(processName);
|
||||
if (!pids.isEmpty()) {
|
||||
this.createProcessCgroup(processName);
|
||||
this.initProcessChains("iptables", processName);
|
||||
this.initProcessChains("ip6tables", processName);
|
||||
this.addProcessToCgroup(processName, pids);
|
||||
this.activeProcesses.add(processName);
|
||||
log.info("已为进程 {} (PID: {}) 创建规则", processName, pids);
|
||||
} else {
|
||||
log.info("进程 {} 当前无PID，暂不创建规则", processName);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (String processName : this.processList) {
|
||||
List<Integer> pids = this.getProcessPids(processName);
|
||||
if (!pids.isEmpty()) {
|
||||
if (!this.activeProcesses.contains(processName)) {
|
||||
log.info("进程 {} 已有PID运行但无规则，开始创建规则...", processName);
|
||||
this.createProcessCgroup(processName);
|
||||
this.initProcessChains("iptables", processName);
|
||||
this.initProcessChains("ip6tables", processName);
|
||||
this.activeProcesses.add(processName);
|
||||
}
|
||||
this.addProcessToCgroup(processName, pids);
|
||||
this.ensureProcessChainsExist("iptables", processName);
|
||||
this.ensureProcessChainsExist("ip6tables", processName);
|
||||
long[] v4Stats = this.getTrafficStatsForProcess("iptables", processName);
|
||||
long[] v6Stats = this.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) {
|
||||
log.debug("进程 {} (PID: {}): IPv4收={} 发={}, IPv6收={} 发={}", processName, pids, v4Stats[0], v4Stats[1], v6Stats[0], v6Stats[1]);
|
||||
}
|
||||
} else {
|
||||
log.debug("进程 {} 当前无PID运行，保留已有规则等待进程启动", processName);
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("获取网络统计失败: {}", e.getMessage(), e);
|
||||
}
|
||||
return resultList;
|
||||
}
|
||||
|
||||
private long generateClassId(String processName) {
|
||||
int hash = Math.abs(processName.hashCode()) & 0xFFFFF;
|
||||
if (hash == 0) hash = 1;
|
||||
return (long) hash & 0xFFFFFFFFL;
|
||||
}
|
||||
|
||||
private void createProcessCgroup(String processName) {
|
||||
try {
|
||||
long classId = this.generateClassId(processName);
|
||||
String cgroupPath = CGROUP_BASE + "/" + CGROUP_PREFIX + processName;
|
||||
this.executeCommand("mkdir -p " + cgroupPath);
|
||||
this.executeCommand("echo " + classId + " > " + cgroupPath + "/net_cls.classid");
|
||||
this.processClassIdMap.put(processName, classId);
|
||||
this.processCgroupPathMap.put(processName, cgroupPath);
|
||||
log.info("进程 {} cgroup 创建完成, classid={}, path={}", processName, classId, cgroupPath);
|
||||
} catch (Exception e) {
|
||||
log.error("创建进程 cgroup 失败 {}: {}", processName, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void removeProcessCgroup(String processName) {
|
||||
try {
|
||||
String cgroupPath = this.processCgroupPathMap.get(processName);
|
||||
if (cgroupPath != null) {
|
||||
this.executeCommand("cat " + cgroupPath + "/tasks | while read pid; do echo $pid > " + CGROUP_BASE + "/tasks 2>/dev/null; done");
|
||||
this.executeCommand("rmdir " + cgroupPath + " 2>/dev/null");
|
||||
this.processClassIdMap.remove(processName);
|
||||
this.processCgroupPathMap.remove(processName);
|
||||
log.info("已删除进程 {} 的 cgroup", processName);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("删除进程 cgroup 失败 {}: {}", processName, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized 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) {
|
||||
if ((line = line.trim()).isEmpty() || line.startsWith("#")) continue;
|
||||
newProcessList.add(line);
|
||||
}
|
||||
log.info("从 {} 加载进程列表: {}", path, newProcessList);
|
||||
} catch (Exception e) {
|
||||
log.error("读取配置文件失败 {}: {}", path, e.getMessage());
|
||||
}
|
||||
}
|
||||
this.processList.clear();
|
||||
this.processList.addAll(newProcessList);
|
||||
}
|
||||
|
||||
private List<Integer> getProcessPids(String processName) {
|
||||
List<Integer> pids = new ArrayList<>();
|
||||
try {
|
||||
Process p = new ProcessBuilder("sh", "-c", "pgrep -x " + processName + " 2>/dev/null").start();
|
||||
try (BufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream()))) {
|
||||
String line;
|
||||
while ((line = br.readLine()) != null) {
|
||||
if (!line.trim().isEmpty()) pids.add(Integer.parseInt(line));
|
||||
}
|
||||
}
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
return pids;
|
||||
}
|
||||
|
||||
private void addProcessToCgroup(String processName, List<Integer> pids) {
|
||||
String cgroupPath = this.processCgroupPathMap.get(processName);
|
||||
if (cgroupPath == null) return;
|
||||
for (Integer pid : pids) {
|
||||
try {
|
||||
Process check = new ProcessBuilder("sh", "-c", "cat " + cgroupPath + "/tasks 2>/dev/null | grep -q " + pid).start();
|
||||
if (check.waitFor() == 0) continue;
|
||||
this.executeCommand("echo " + pid + " > " + cgroupPath + "/tasks");
|
||||
log.debug("进程 {} ({}) 已加入 cgroup", pid, processName);
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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) {
|
||||
String baseName = "am_" + safeName;
|
||||
int maxNeeded = Math.max(baseName.length() + 6, baseName.length() + 7);
|
||||
if (maxNeeded > 28) {
|
||||
int baseMaxLen = 21;
|
||||
if (baseName.length() > baseMaxLen) {
|
||||
String prefix = baseName.substring(0, 8);
|
||||
String suffixPart = baseName.substring(baseName.length() - (baseMaxLen - 9));
|
||||
baseName = prefix + "_" + suffixPart;
|
||||
}
|
||||
}
|
||||
return baseName + "_" + suffix + "_" + direction;
|
||||
}
|
||||
|
||||
private long getProcessClassId(String processName) {
|
||||
Long classId = this.processClassIdMap.get(processName);
|
||||
if (classId == null) {
|
||||
classId = this.generateClassId(processName);
|
||||
this.processClassIdMap.put(processName, classId);
|
||||
}
|
||||
return classId;
|
||||
}
|
||||
|
||||
private void initProcessChains(String command, String processName) {
|
||||
long classId = this.getProcessClassId(processName);
|
||||
String safeName = this.generateSafeChainName(processName);
|
||||
String suffix = command.equals("iptables") ? "v4" : "v6";
|
||||
String inChain = this.buildChainName(safeName, suffix, "in");
|
||||
String outChain = this.buildChainName(safeName, suffix, "out");
|
||||
this.initChain(command, inChain, "INPUT", classId);
|
||||
this.initChain(command, outChain, "OUTPUT", classId);
|
||||
}
|
||||
|
||||
private void ensureProcessChainsExist(String command, String processName) {
|
||||
long classId = this.getProcessClassId(processName);
|
||||
String safeName = this.generateSafeChainName(processName);
|
||||
String suffix = command.equals("iptables") ? "v4" : "v6";
|
||||
String inChain = this.buildChainName(safeName, suffix, "in");
|
||||
String outChain = this.buildChainName(safeName, suffix, "out");
|
||||
this.ensureChainExists(command, inChain, "INPUT", classId);
|
||||
this.ensureChainExists(command, outChain, "OUTPUT", classId);
|
||||
}
|
||||
|
||||
private void ensureChainExists(String command, String chainName, String hookChain, long classId) {
|
||||
try {
|
||||
Process check = new ProcessBuilder("sh", "-c", command + " -L " + chainName + " -n >/dev/null 2>&1").start();
|
||||
if (check.waitFor() != 0) {
|
||||
this.executeCommand(command + " -N " + chainName);
|
||||
this.executeCommand(command + " -A " + chainName + " -m cgroup --cgroup " + classId + " -j RETURN");
|
||||
this.executeCommand(command + " -A " + chainName + " -j RETURN");
|
||||
this.executeCommand(command + " -I " + hookChain + " 1 -m cgroup --cgroup " + classId + " -j " + chainName);
|
||||
log.info("{} 链 {} 初始化完成 (classid={})", command, chainName, classId);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("确保链存在失败 {}: {}", chainName, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void initChain(String command, String chainName, String hookChain, long classId) {
|
||||
try {
|
||||
this.executeCommand(command + " -N " + chainName + " 2>/dev/null");
|
||||
this.executeCommand(command + " -F " + chainName);
|
||||
this.executeCommand(command + " -A " + chainName + " -m cgroup --cgroup " + classId + " -j RETURN");
|
||||
this.executeCommand(command + " -A " + chainName + " -j RETURN");
|
||||
Process check = new ProcessBuilder("sh", "-c", command + " -C " + hookChain + " -m cgroup --cgroup " + classId + " -j " + chainName + " 2>/dev/null").start();
|
||||
if (check.waitFor() != 0) {
|
||||
this.executeCommand(command + " -I " + hookChain + " 1 -m cgroup --cgroup " + classId + " -j " + chainName);
|
||||
log.debug("{} 插入规则到 {} 成功", command, hookChain);
|
||||
} else {
|
||||
log.debug("{} 规则已存在于 {}，跳过插入", command, hookChain);
|
||||
}
|
||||
log.debug("{} 链 {} 初始化完成 (classid={})", command, chainName, classId);
|
||||
} catch (Exception e) {
|
||||
log.error("初始化链 {} 失败: {}", chainName, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private long[] getTrafficStatsForProcess(String command, String processName) {
|
||||
long classId = this.getProcessClassId(processName);
|
||||
String safeName = this.generateSafeChainName(processName);
|
||||
String suffix = command.equals("iptables") ? "v4" : "v6";
|
||||
String inChain = this.buildChainName(safeName, suffix, "in");
|
||||
String outChain = this.buildChainName(safeName, suffix, "out");
|
||||
return new long[]{this.getChainBytes(command, inChain), this.getChainBytes(command, outChain)};
|
||||
}
|
||||
|
||||
private void cleanupProcessRules(String command, String processName) {
|
||||
try {
|
||||
Long classId = this.processClassIdMap.get(processName);
|
||||
if (classId == null) return;
|
||||
String safeName = this.generateSafeChainName(processName);
|
||||
String suffix = command.equals("iptables") ? "v4" : "v6";
|
||||
String inChain = this.buildChainName(safeName, suffix, "in");
|
||||
String outChain = this.buildChainName(safeName, suffix, "out");
|
||||
this.executeCommand(command + " -D INPUT -m cgroup --cgroup " + classId + " -j " + inChain + " 2>/dev/null");
|
||||
this.executeCommand(command + " -D OUTPUT -m cgroup --cgroup " + classId + " -j " + outChain + " 2>/dev/null");
|
||||
this.executeCommand(command + " -F " + inChain + " 2>/dev/null");
|
||||
this.executeCommand(command + " -F " + outChain + " 2>/dev/null");
|
||||
this.executeCommand(command + " -X " + inChain + " 2>/dev/null");
|
||||
this.executeCommand(command + " -X " + outChain + " 2>/dev/null");
|
||||
log.info("已清理 {} 中进程 {} 的规则", command, processName);
|
||||
} catch (Exception e) {
|
||||
log.error("清理进程规则失败 {}: {}", processName, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private long getChainBytes(String command, String chainName) {
|
||||
try {
|
||||
Process p = new ProcessBuilder("sh", "-c", command + " -L " + chainName + " -v -n -x | grep RETURN | head -1").start();
|
||||
try (BufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream()))) {
|
||||
String line = br.readLine();
|
||||
if (line == null || line.isEmpty()) return 0L;
|
||||
String[] parts = line.trim().split("\\s+");
|
||||
if (parts.length < 2) return 0L;
|
||||
return Long.parseLong(parts[1]);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("获取链 {} 统计失败: {}", chainName, e.getMessage());
|
||||
}
|
||||
return 0L;
|
||||
}
|
||||
|
||||
private void executeCommand(String command) throws Exception {
|
||||
Process p = new ProcessBuilder("sh", "-c", command).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")) {
|
||||
log.warn("命令执行警告: {} -> {}", command, error);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user