diff --git a/src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java b/src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java index 5f4e5b2..9e95269 100644 --- a/src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java +++ b/src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java @@ -40,6 +40,16 @@ public enum MsgEnum { 告警上报("ALARM"), + 建立连接("CONNECT"), + + 获取最新策略("GET_POLICY"), + + 多网IP探测上报("NETWORK_DETECT"), + + 内存详情上报("MEMORY_DETAIL"), + + 业务网络上报("BUSINESS_NET"), + 开启或更新系统采集("SYSTEM_COLLECT_START"), 开启或更新系统采集应答("SYSTEM_COLLECT_START_RSP"), diff --git a/src/main/java/com/tongran/agent/client/core/vo/NetBusinessVO.java b/src/main/java/com/tongran/agent/client/core/vo/NetBusinessVO.java new file mode 100644 index 0000000..a96b480 --- /dev/null +++ b/src/main/java/com/tongran/agent/client/core/vo/NetBusinessVO.java @@ -0,0 +1,27 @@ +package com.tongran.agent.client.core.vo; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import java.io.Serializable; + +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class NetBusinessVO implements Serializable { + private static final long serialVersionUID = 9L; + private String name; + private String mac; + private int pid; + private String processName; + private Long inSpeed; + private Long outSpeed; + private Long ipv4InSpeed; + private Long ipv4OutSpeed; + private Long ipv6InSpeed; + private Long ipv6OutSpeed; + private int connectionCount; + private long timestamp; +} diff --git a/src/main/java/com/tongran/agent/client/core/vo/NetworkInterfaceInfo.java b/src/main/java/com/tongran/agent/client/core/vo/NetworkInterfaceInfo.java new file mode 100644 index 0000000..04cb2b7 --- /dev/null +++ b/src/main/java/com/tongran/agent/client/core/vo/NetworkInterfaceInfo.java @@ -0,0 +1,74 @@ +package com.tongran.agent.client.core.vo; + +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import lombok.AllArgsConstructor; + +import java.util.List; + +/** + * 网络接口信息VO（从1.20恢复） + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class NetworkInterfaceInfo { + private String name; + private String mac; + private String type; + private String ipv4; + private String gateway; + private String publicIp; + private String carrier; + private String province; + private String city; + private String ipv6; + private String status; + private String parentInterface; + private Long netCreateTime; + private List subInterfaces; + + /** + * 兼容旧builder模式 + */ + public static NetworkInterfaceInfoBuilder builder() { + return new NetworkInterfaceInfoBuilder(); + } + + public static class NetworkInterfaceInfoBuilder { + private String name; + private String mac; + private String type; + private String ipv4; + private String gateway; + private String publicIp; + private String carrier; + private String province; + private String city; + private String ipv6; + private String status; + private String parentInterface; + private Long netCreateTime; + private List subInterfaces; + + public NetworkInterfaceInfoBuilder name(String name) { this.name = name; return this; } + public NetworkInterfaceInfoBuilder mac(String mac) { this.mac = mac; return this; } + public NetworkInterfaceInfoBuilder type(String type) { this.type = type; return this; } + public NetworkInterfaceInfoBuilder ipv4(String ipv4) { this.ipv4 = ipv4; return this; } + public NetworkInterfaceInfoBuilder gateway(String gateway) { this.gateway = gateway; return this; } + public NetworkInterfaceInfoBuilder publicIp(String publicIp) { this.publicIp = publicIp; return this; } + public NetworkInterfaceInfoBuilder carrier(String carrier) { this.carrier = carrier; return this; } + public NetworkInterfaceInfoBuilder province(String province) { this.province = province; return this; } + public NetworkInterfaceInfoBuilder city(String city) { this.city = city; return this; } + public NetworkInterfaceInfoBuilder ipv6(String ipv6) { this.ipv6 = ipv6; return this; } + public NetworkInterfaceInfoBuilder status(String status) { this.status = status; return this; } + public NetworkInterfaceInfoBuilder parentInterface(String parentInterface) { this.parentInterface = parentInterface; return this; } + public NetworkInterfaceInfoBuilder netCreateTime(Long netCreateTime) { this.netCreateTime = netCreateTime; return this; } + public NetworkInterfaceInfoBuilder subInterfaces(List subInterfaces) { this.subInterfaces = subInterfaces; return this; } + public NetworkInterfaceInfo build() { + return new NetworkInterfaceInfo(name, mac, type, ipv4, gateway, publicIp, carrier, province, city, ipv6, status, parentInterface, netCreateTime, subInterfaces); + } + } +} 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 fad0765..c99e446 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 @@ -9,6 +9,8 @@ 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.core.vo.*; +import com.tongran.agent.client.netty.MultiTargetNettyClient; +import com.tongran.agent.client.netty.config.AgentNettyConfig; import com.tongran.agent.client.netty.model.Message; import com.tongran.agent.client.service.*; import com.tongran.agent.client.utils.AgentUtil; @@ -116,6 +118,18 @@ public class BusinessTasks { @Resource private ApplicationProperties properties; + @Resource + private MultiTargetNettyClient client; + + @Resource + private AgentNettyConfig config; + + @Resource + private AgentService agentService; + + @Resource + private NetBusinessService netBusinessService; + /** * 任务1：心跳上报任务 */ @@ -125,19 +139,34 @@ public class BusinessTasks { timestamp = Math.round(timestamp / 1000.0); int count = heartbeatTask.incrementAndGet(); AssertLog.info("心跳定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); - // 业务处理 - // 判定客户端与服务端是否连接 - if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { - // 发送心跳包 - JSONObject object = new JSONObject(); - object.put("strength","31"); - object.put("name", properties.getName()); - object.put("version", properties.getVersion()); - object.put("startupTime", GlobalConfig.startupTime); - object.put("timestamp",timestamp); - Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.心跳上报.getValue()).data(object.toString()).build(); - sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); - AssertLog.info("发送心跳包={}",JSON.toJSONString(message)); + // 检查活跃连接，如果断开则尝试重连 + boolean success = true; + int activeConnect = client.getActiveConnections(); + if (activeConnect == 0) { + success = agentService.connection(); + } + if (success) { + // 判定客户端与服务端是否连接 + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) { + GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN(); + } + // 发送心跳包 + JSONObject object = new JSONObject(); + object.put("clientId", GlobalConfig.CLIENT_ID); + object.put("logicalNode", agentService.getLogicalNode()); + object.put("sn", GlobalConfig.DEVICE_SN); + object.put("strength","31"); + object.put("name", properties.getName()); + object.put("version", properties.getVersion()); + object.put("startupTime", GlobalConfig.startupTime); + object.put("timestamp",timestamp); + Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.心跳上报.getValue()).data(object.toString()).build(); + sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); + AssertLog.info("发送心跳包={}",JSON.toJSONString(message)); + } + } else { + AssertLog.info("心跳定时任务执行失败-连接断开"); } AssertLog.info("心跳定时任务执行 - task #{} completed", count); } @@ -1275,50 +1304,216 @@ public class BusinessTasks { /** * 策略更新定时任务 */ + @Async("taskExecutor") public void policyTask() { - // TODO: 从1.20恢复完整实现 + long timestamp = AgentUtil.roundMinutes(); + int count = new AtomicInteger(0).incrementAndGet(); + AssertLog.info("获取最新策略定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) { + GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN(); + } + JSONObject object = new JSONObject(); + object.put("clientId", GlobalConfig.CLIENT_ID); + object.put("sn", GlobalConfig.DEVICE_SN); + object.put("timestamp", timestamp); + Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.获取最新策略.getValue()).data(object.toString()).build(); + sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); + AssertLog.info("发送获取最新策略信息包={}", JSON.toJSONString(message)); + } + AssertLog.info("获取最新策略定时任务执行 - task #{} completed", count); } /** * 多网IP探测定时任务 */ + @Async("taskExecutor") public void networkDetectTask() { - // TODO: 从1.20恢复完整实现 + int count = new AtomicInteger(0).incrementAndGet(); + AssertLog.info("多网IP探测定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) { + GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN(); + } + long timestamps = System.currentTimeMillis(); + timestamps = Math.round(timestamps / 1000.0); + List infos = null; + try { + // NetworkInterfaceUtil may not be available in all environments + infos = com.tongran.agent.client.utils.NetworkInterfaceUtil.collectNetworkInfo(); + } catch (Exception e) { + AssertLog.warn("collectNetworkInfo failed: {}", e.getMessage()); + } + JSONObject objects = new JSONObject(); + objects.put("clientId", GlobalConfig.CLIENT_ID); + objects.put("sn", GlobalConfig.DEVICE_SN); + objects.put("networkInfo", JSON.toJSONString(infos)); + objects.put("timestamp", timestamps); + Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.多网IP探测上报.getValue()).data(objects.toString()).build(); + sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); + AssertLog.info("发送多网IP探测信息包={}", JSON.toJSONString(message)); + } + AssertLog.info("发送多网IP探测定时任务执行 - task #{} completed", count); } /** * FRPC保活定时任务 */ + @Async("taskExecutor") public void checkFrpcTask() { - // TODO: 从1.20恢复完整实现 + int count = new AtomicInteger(0).incrementAndGet(); + AssertLog.info("frpc保活机制定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + agentService.checkFrpc(); + } else { + AssertLog.info("断开连接--frpc保活机制定时任务执行 - task #{} completed", count); + } } /** * FRPC状态上报定时任务 */ + @Async("taskExecutor") public void upFrpcMsgTask() { - // TODO: 从1.20恢复完整实现 + int count = new AtomicInteger(0).incrementAndGet(); + AssertLog.info("frpc状态上报定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + agentService.upFrpcMsg(); + } else { + AssertLog.info("断开连接--frpc状态上报定时任务执行 - task #{} completed", count); + } } /** * 本地存储流量信息上报定时任务 */ + @Async("taskExecutor") public void upTempTraffic() { - // TODO: 从1.20恢复完整实现 + int count = new AtomicInteger(0).incrementAndGet(); + AssertLog.info("本地存储流量信息上报定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + agentService.checkTraffic(); + } + AssertLog.info("本地存储流量信息上报定时任务执行 - task #{} completed", count); } /** - * 注册定时任务 + * 注册重试定时任务 */ + @Async("taskExecutor") public void registerTask() { - // TODO: 从1.20恢复完整实现 + long timestamp = System.currentTimeMillis(); + timestamp = Math.round(timestamp / 1000.0); + int count = new AtomicInteger(0).incrementAndGet(); + AssertLog.info("注册重试定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + boolean success = true; + int activeConnect = client.getActiveConnections(); + AssertLog.info("注册重试定时任务执行 - 时间: {}，activeConnect={}", LocalDateTime.now(), activeConnect); + if (activeConnect == 0 || count > 1) { + success = agentService.connection(); + } + if (success) { + AssertLog.info("连接成功，发送初始连接消息"); + try { + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) { + GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN(); + } + List infos = null; + try { + infos = com.tongran.agent.client.utils.NetworkInterfaceUtil.collectNetworkInfo(); + } catch (Exception e) { + AssertLog.warn("collectNetworkInfo failed: {}", e.getMessage()); + } + JSONObject object = new JSONObject(); + object.put("clientId", GlobalConfig.CLIENT_ID); + object.put("sn", GlobalConfig.DEVICE_SN); + object.put("networkInfo", JSON.toJSONString(infos)); + object.put("timestamp", timestamp); + Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.注册.getValue()).data(object.toString()).build(); + sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); + AssertLog.info("发送注册重试={}", JSON.toJSONString(message)); + } + } catch (Exception e) { + e.printStackTrace(); + } + } else { + AssertLog.info("注册重试定时任务执行 - 时间: {}，建立连接失败", LocalDateTime.now()); + } + AssertLog.info("注册重试定时任务执行 - task #{} completed", count); } /** - * 连接重试定时任务 + * 连接重试定时任务 - 最关键的断线重连逻辑 */ + @Async("taskExecutor") public void connectionTask() { - // TODO: 从1.20恢复完整实现 + int count = new AtomicInteger(0).incrementAndGet(); + AssertLog.info("建立连接重试定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + boolean success = client.createConnection(GlobalConfig.CLIENT_ID, config.getHost(), config.getPort(), 5); + if (success) { + if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) { + GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN(); + } + long timestamp = System.currentTimeMillis(); + timestamp = Math.round(timestamp / 1000.0); + JSONObject object = new JSONObject(); + object.put("clientId", GlobalConfig.CLIENT_ID); + object.put("sn", GlobalConfig.DEVICE_SN); + object.put("timestamp", timestamp); + Message msg = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.建立连接.getValue()).data(object.toString()).build(); + client.sendMessages(GlobalConfig.CLIENT_ID, msg); + if (GlobalConfig.isRegister) { + AssertLog.info("启动心跳定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 30000); + dynamicTaskService.scheduleTask("heartbeat", () -> businessTasks.heartbeatTask(), 15000L, 30000L); + long milli = AgentUtil.getMillisToNextMinute() + 60000L; + AssertLog.info("启动更新策略定时任务 - 延迟: {}ms, 间隔: {}ms", milli, 60000); + dynamicTaskService.scheduleTask("policy", () -> businessTasks.policyTask(), milli, 60000L); + AssertLog.info("启动多网IP探测定时任务Retry - 延迟: {}ms, 间隔: {}ms", milli, 300000); + dynamicTaskService.scheduleTask("networkDetect", () -> businessTasks.networkDetectTask(), milli, 300000L); + AssertLog.info("检测监控策略配置Retry"); + agentService.checkMonitor(); + AssertLog.info("检测agent更新配置"); + agentService.checkAgentUpdate(); + AssertLog.info("启动frpc保活定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 600000); + dynamicTaskService.scheduleTask("checkFrpc", () -> businessTasks.checkFrpcTask(), 15000L, 600000L); + AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000); + dynamicTaskService.scheduleTask("upFrpcMsg", () -> businessTasks.upFrpcMsgTask(), 15000L, 300000L); + long milliFive = AgentUtil.millisecondsToNext5Minute(); + AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 0x6DDD00); + dynamicTaskService.scheduleTask("upTempTraffic", () -> businessTasks.upTempTraffic(), milliFive, 0x6DDD00L); + AssertLog.info("检测PppoE配置"); + agentService.handleRebootRecovery(); + AssertLog.info("检测tcpdump探测时间配置"); + agentService.checkTcpdumpTimes(); + } else { + try { + if (StringUtils.isBlank(GlobalConfig.DEVICE_SN)) { + GlobalConfig.DEVICE_SN = AgentUtil.getDeviceSN(); + } + long timestamps = System.currentTimeMillis(); + timestamps = Math.round(timestamps / 1000.0); + List infos = null; + try { + infos = com.tongran.agent.client.utils.NetworkInterfaceUtil.collectNetworkInfo(); + } catch (Exception e) { + AssertLog.warn("collectNetworkInfo failed: {}", e.getMessage()); + } + JSONObject objects = new JSONObject(); + objects.put("clientId", GlobalConfig.CLIENT_ID); + objects.put("sn", GlobalConfig.DEVICE_SN); + objects.put("networkInfo", JSON.toJSONString(infos)); + objects.put("timestamp", timestamps); + Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.注册.getValue()).data(objects.toString()).build(); + client.sendMessages(GlobalConfig.CLIENT_ID, message); + } catch (Exception e) { + e.printStackTrace(); + } + dynamicTaskService.scheduleTask("register", () -> businessTasks.registerTask(), 0L, 300000L); + } + dynamicTaskService.cancelTask("connection"); + } + AssertLog.info("建立连接重试定时任务执行 - task #{} completed", count); } } \ No newline at end of file diff --git a/src/main/java/com/tongran/agent/client/service/NetBusinessService.java b/src/main/java/com/tongran/agent/client/service/NetBusinessService.java new file mode 100644 index 0000000..6bbbcc7 --- /dev/null +++ b/src/main/java/com/tongran/agent/client/service/NetBusinessService.java @@ -0,0 +1,8 @@ +package com.tongran.agent.client.service; + +import com.tongran.agent.client.core.vo.NetBusinessVO; +import java.util.List; + +public interface NetBusinessService { + List netList(long var1); +} diff --git a/src/main/java/com/tongran/agent/client/utils/NetworkInterfaceUtil.java b/src/main/java/com/tongran/agent/client/utils/NetworkInterfaceUtil.java new file mode 100644 index 0000000..de1a14e --- /dev/null +++ b/src/main/java/com/tongran/agent/client/utils/NetworkInterfaceUtil.java @@ -0,0 +1,105 @@ +package com.tongran.agent.client.utils; + +import com.tongran.agent.client.core.vo.NetworkInterfaceInfo; + +import java.util.ArrayList; +import java.util.List; + +/** + * 网络接口工具类（从1.20恢复的简化版） + * 完整的网卡信息采集依赖PublicIpFetcher等多个类，此处提供基础实现 + * collectNetworkInfo用于注册消息中的网络信息上报 + */ +public class NetworkInterfaceUtil { + + public static boolean getmacVlanStatus(String interfaceName) { + try { + Process process = Runtime.getRuntime().exec( + new String[]{"ping", "-I", interfaceName, "-c", "1", "-W", "2", "www.baidu.com"}); + return process.waitFor() == 0; + } catch (Exception e) { + AssertLog.warn("接口 {} ping异常: {}", interfaceName, e.getMessage()); + return false; + } + } + + /** + * 收集网络接口信息 + * 返回所有活跃的以太网接口信息列表 + */ + public static List collectNetworkInfo() { + List result = new ArrayList<>(); + try { + // 使用 ip link show 获取网卡列表 + Process process = Runtime.getRuntime().exec("ip -o link show"); + java.io.BufferedReader reader = new java.io.BufferedReader( + new java.io.InputStreamReader(process.getInputStream())); + String line; + while ((line = reader.readLine()) != null) { + // 格式: 1: lo: ... + String[] parts = line.trim().split(":"); + if (parts.length < 2) continue; + String name = parts[1].trim(); + // 跳过回环接口 + if ("lo".equals(name)) continue; + // 检查是否UP + if (!line.contains("UP")) continue; + + NetworkInterfaceInfo info = NetworkInterfaceInfo.builder() + .name(name) + .type("Ethernet") + .build(); + + // 获取MAC地址 + try { + Process macProcess = Runtime.getRuntime().exec( + "cat /sys/class/net/" + name + "/address"); + java.io.BufferedReader macReader = new java.io.BufferedReader( + new java.io.InputStreamReader(macProcess.getInputStream())); + String mac = macReader.readLine(); + if (mac != null && !mac.trim().isEmpty()) { + info.setMac(mac.trim()); + } + macReader.close(); + } catch (Exception ignored) {} + + // 获取IPv4 + try { + Process ipProcess = Runtime.getRuntime().exec( + new String[]{"sh", "-c", "ip -4 addr show " + name + " 2>/dev/null | grep 'inet ' | awk '{print $2}' | cut -d'/' -f1"}); + java.io.BufferedReader ipReader = new java.io.BufferedReader( + new java.io.InputStreamReader(ipProcess.getInputStream())); + String ip = ipReader.readLine(); + if (ip != null && !ip.trim().isEmpty() && !"127.0.0.1".equals(ip.trim())) { + info.setIpv4(ip.trim()); + } + ipReader.close(); + } catch (Exception ignored) {} + + result.add(info); + } + reader.close(); + } catch (Exception e) { + AssertLog.warn("collectNetworkInfo失败: {}", e.getMessage()); + } + return result; + } + + public static long getInterfaceCreateTime(String interfaceName) { + try { + Process process = Runtime.getRuntime().exec( + new String[]{"stat", "-c", "%Y", "/sys/class/net/" + interfaceName}); + java.io.BufferedReader reader = new java.io.BufferedReader( + new java.io.InputStreamReader(process.getInputStream())); + String output = reader.readLine(); + reader.close(); + process.waitFor(); + if (output != null && !output.trim().isEmpty()) { + return Long.parseLong(output.trim()); + } + } catch (Exception e) { + AssertLog.error("获取网卡 {} 创建时间失败: {}", interfaceName, e.getMessage()); + } + return -1L; + } +}