From 5ab9a52628d1f1ed6e91c20cf57a38d995519812 Mon Sep 17 00:00:00 2001 From: lee Date: Wed, 22 Jul 2026 17:32:55 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=96=B0=E5=A2=9E=20NetBusinessServiceI?= =?UTF-8?q?mpl=20=E5=AE=9E=E7=8E=B0=E7=B1=BB=EF=BC=8C=E4=BF=AE=E5=A4=8D=20?= =?UTF-8?q?Spring=20Bean=20=E6=B3=A8=E5=85=A5=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题：BusinessTasks依赖注入NetBusinessService，但缺少@Service实现类 导致启动报错 NoSuchBeanDefinitionException 解决：从1.20反编译代码恢复 NetBusinessServiceImpl - 多进程 cgroup + iptables 网络流量监控 - System.out改为 SLF4J Logger - @Service注解确保Spring自动装配 --- .../service/impl/NetBusinessServiceImpl.java | 370 ++++++++++++++++++ 1 file changed, 370 insertions(+) create mode 100644 src/main/java/com/tongran/agent/client/service/impl/NetBusinessServiceImpl.java diff --git a/src/main/java/com/tongran/agent/client/service/impl/NetBusinessServiceImpl.java b/src/main/java/com/tongran/agent/client/service/impl/NetBusinessServiceImpl.java new file mode 100644 index 0000000..aafd8a2 --- /dev/null +++ b/src/main/java/com/tongran/agent/client/service/impl/NetBusinessServiceImpl.java @@ -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 processClassIdMap = new ConcurrentHashMap<>(); + private final ConcurrentHashMap processCgroupPathMap = new ConcurrentHashMap<>(); + private final Set activeProcesses = ConcurrentHashMap.newKeySet(); + private List processList = new ArrayList<>(); + @Resource + private ApplicationProperties properties; + + @PostConstruct + public void init() { + try { + this.loadProcessListFromFile(); + for (String processName : this.processList) { + List 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 netList(long timestamp) { + List resultList = new ArrayList<>(); + try { + List oldProcessList = new ArrayList<>(this.processList); + this.loadProcessListFromFile(); + + Set 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 addedProcesses = new HashSet<>(this.processList); + addedProcesses.removeAll(oldProcessList); + if (!addedProcesses.isEmpty()) { + log.info("检测到新增的进程: {}", addedProcesses); + for (String processName : addedProcesses) { + List 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 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 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 getProcessPids(String processName) { + List 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 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); + } + } + } + } +}