diff --git a/src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java b/src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java index 6c52bbd..b2160f9 100644 --- a/src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java +++ b/src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java @@ -17,4 +17,5 @@ public class ApplicationProperties { private String tempPath; private String frpPath; private String poePath; + private String tempTrafficPath; } \ No newline at end of file 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 404153a..22ec1ec 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 @@ -39,6 +39,8 @@ public enum MsgEnum { 网络上报("NET"), + 网络上报重试("NET_RECOVER"), + 挂载上报("POINT"), 交换机上报("SWITCHBOARD"), diff --git a/src/main/java/com/tongran/agent/client/netty/model/Message.java b/src/main/java/com/tongran/agent/client/netty/model/Message.java index e676e2f..d755548 100644 --- a/src/main/java/com/tongran/agent/client/netty/model/Message.java +++ b/src/main/java/com/tongran/agent/client/netty/model/Message.java @@ -33,5 +33,7 @@ public class Message implements Serializable { * 发送内容 */ private String data; + /** 最后一次标志 */ + private boolean lastTrafficFlag; } diff --git a/src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java b/src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java index c7c942a..a2440d1 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java +++ b/src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java @@ -104,6 +104,9 @@ public class AppInitializer implements CommandLineRunner { dynamicTaskService.scheduleTask("checkFrpc", businessTasks::checkFrpcTask, 15000, 600000); AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000); dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000); + long milliFive = AgentUtil.millisecondsToNext5Minute(); + AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000); + dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000); AssertLog.info("检测PppoE配置"); agentService.handleRebootRecovery(); }else{ 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 25bfb38..3007a38 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 @@ -22,7 +22,9 @@ import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; import javax.annotation.Resource; +import java.text.SimpleDateFormat; import java.time.LocalDateTime; +import java.util.Date; import java.util.List; import java.util.Objects; import java.util.concurrent.atomic.AtomicInteger; @@ -64,6 +66,7 @@ public class BusinessTasks { private final AtomicInteger checkFrpcTask = new AtomicInteger(0); private final AtomicInteger upFrpcMsgTask = new AtomicInteger(0); private final AtomicInteger upMacvlanStatusTask = new AtomicInteger(0); + private final AtomicInteger upTempTrafficTask = new AtomicInteger(0); @@ -263,6 +266,15 @@ public class BusinessTasks { data = JSONArray.toJSONString(list); } Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.网络上报.getValue()).data(data).build(); + // 将本次流量存储到本地文件 + String[] lines = { + "traffic=" + data, + }; + + // 检查目录并写入配置文件 + if (AdvancedAsyncDownloader.createSingleDirectoryIfNotExists(properties.getTempTrafficPath())) { + AgentUtil.bufferedWriter(properties.getTempTrafficPath() + "/" + timestamp+".conf", lines); + } sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); AssertLog.info("发送网络信息包={}",JSON.toJSONString(message)); } @@ -863,6 +875,9 @@ public class BusinessTasks { dynamicTaskService.scheduleTask("checkFrpc", businessTasks::checkFrpcTask, 15000, 600000); AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000); dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000); + long milliFive = AgentUtil.millisecondsToNext5Minute(); + AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000); + dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000); AssertLog.info("检测PppoE配置"); agentService.handleRebootRecovery(); }else{ @@ -1007,7 +1022,19 @@ public class BusinessTasks { } AssertLog.info("发送macvlan状态上报定时任务执行 - task #{} completed", count); } - + /** + * 任务34：本地存储流量信息上报 + */ + @Async("taskExecutor") + public void upTempTraffic() { + int count = upTempTrafficTask.incrementAndGet(); + AssertLog.info("本地存储流量信息上报定时任务执行 - 时间: {}，task #{}", LocalDateTime.now(), count); + // 判定客户端与服务端是否连接 + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + agentService.checkTraffic(); + } + AssertLog.info("本地存储流量信息上报定时任务执行 - task #{} completed", count); + } // diff --git a/src/main/java/com/tongran/agent/client/service/AgentService.java b/src/main/java/com/tongran/agent/client/service/AgentService.java index b3c51bc..e93e635 100644 --- a/src/main/java/com/tongran/agent/client/service/AgentService.java +++ b/src/main/java/com/tongran/agent/client/service/AgentService.java @@ -46,6 +46,7 @@ public interface AgentService { void upFrpcMsg(); void handleRebootRecovery(); + void checkTraffic(); List upMacvlanStatus(); } diff --git a/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java b/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java index 67bcf0e..b9cf5d4 100644 --- a/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java +++ b/src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java @@ -321,6 +321,9 @@ public class AgentServiceImpl implements AgentService { dynamicTaskService.scheduleTask("checkFrpc", businessTasks::checkFrpcTask, 15000, 600000); AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000); dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000); + long milliFive = AgentUtil.millisecondsToNext5Minute(); + AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000); + dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000); AssertLog.info("检测PppoE配置"); handleRebootRecovery(); } @@ -915,9 +918,39 @@ public class AgentServiceImpl implements AgentService { String diskName = jsonObject.getString("umountDisk"); handleUmountDisk(diskName); } - + if(jsonObject.containsKey("trafficUsedTimestamp")){ + String trafficUsedTimestamp = jsonObject.getString("trafficUsedTimestamp"); +// if (trafficUsedTimestamp != null) { +// String[] fileNames = trafficUsedTimestamp.split(","); +// +// // 删除对应文件 +// for (String fileName : fileNames) { +// if (StringUtils.isNotBlank(fileName)) { +// fileName = fileName+"conf"; +// String filePath = properties.getTempTrafficPath() + fileName.trim(); +// AssertLog.info("已处理文件：{}", fileName); +// deleteTrafficFile(filePath); +// } +// } +// } + } + } + private void deleteTrafficFile(String filePath) { + try { + File file = new File(filePath); + if (file.exists()) { + if (file.delete()) { + AssertLog.info("成功删除流量文件: {}", filePath); + } else { + AssertLog.warn("删除流量文件失败: {}", filePath); + } + } else { + AssertLog.info("流量文件不存在，无需删除: {}", filePath); + } + } catch (Exception e) { + AssertLog.error("删除流量文件异常，文件: {}, 异常信息: {}", filePath, e.getMessage()); + } } - private void handleUmountDisk(String diskName) { try { // 查询所有挂载点 @@ -2249,7 +2282,139 @@ public class AgentServiceImpl implements AgentService { AssertLog.debug("业务网卡配置文件不存在: {}", businessNetFile.getPath()); } } + // 获取所有流量打点文件 + // 更简洁的排序方法（假设文件名都是纯时间戳） + public List getTrafficFiles() { + File dir = new File(properties.getTempTrafficPath()); + File[] files = dir.listFiles((d, name) -> name.endsWith(".conf")); + if (files == null) { + return new ArrayList<>(); + } + + // 直接按文件名排序（时间戳是数字，字符串排序也有效） + List fileList = Arrays.asList(files); + fileList.sort(Comparator.comparing(File::getName)); + + return fileList; + } + @Override + public void checkTraffic() { + List fileList = getTrafficFiles(); // 获取已排序的文件列表 + if (fileList == null || fileList.isEmpty()) { + return; + } + + // 显示排序信息 + String firstTimestamp = fileList.get(0).getName().replace(".conf", ""); + String lastTimestamp = fileList.get(fileList.size() - 1).getName().replace(".conf", ""); + + AssertLog.info("开始处理流量重试文件，共 {} 个（时间戳范围: {} - {}，小的先上报）", + fileList.size(), firstTimestamp, lastTimestamp); + + // 配置参数 + int batchSize = 10; // 每批10个文件 + long intraBatchInterval = 5000; // 批次内文件间隔5s + long interBatchInterval = 1000; // 批次间间隔10秒 + + int totalProcessed = 0; + int totalFiles = fileList.size(); + + // 按批次处理（文件已按时间戳从小到大排序） + for (int batchStart = 0; batchStart < totalFiles; batchStart += batchSize) { + int batchEnd = Math.min(batchStart + batchSize, totalFiles); + List currentBatch = fileList.subList(batchStart, batchEnd); + + // 获取当前批次的时间戳范围 + String batchStartTime = currentBatch.get(0).getName().replace(".conf", ""); + String batchEndTime = currentBatch.get(currentBatch.size() - 1).getName().replace(".conf", ""); + + AssertLog.info("处理第 {} 批文件（时间戳: {} - {}），共 {} 个文件", + (batchStart / batchSize) + 1, batchStartTime, batchEndTime, currentBatch.size()); + + // 处理当前批次 + int batchProcessed = 0; + for (File file : currentBatch) { + try { + // 判断是否是最后一个文件 + boolean isLastTraffic = (totalProcessed + 1 == totalFiles); + + // 获取当前文件的时间戳用于日志 + String currentTimestamp = file.getName().replace(".conf", ""); + + processSingleFile(file, isLastTraffic); + batchProcessed++; + totalProcessed++; + + // 批次内文件间隔（最后一个文件不等待） + if (batchProcessed < currentBatch.size()) { + AssertLog.info("文件 {} 处理完成，等待 {}ms 后处理下一个", + currentTimestamp, intraBatchInterval); + Thread.sleep(intraBatchInterval); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + AssertLog.warn("文件处理被中断，已处理 {} 个文件", totalProcessed); + return; + } catch (Exception e) { + AssertLog.error("处理文件异常，文件：{}, 异常信息: {}", + file.getName(), e.getMessage()); + } + } + + // 批次间间隔（最后一批不等待） + if (batchEnd < totalFiles) { + try { + AssertLog.info("第 {} 批处理完成，共 {} 个文件，等待 {}ms 后处理下一批", + (batchStart / batchSize) + 1, batchProcessed, interBatchInterval); + Thread.sleep(interBatchInterval); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + AssertLog.warn("批次处理被中断，已处理 {} 个文件", totalProcessed); + return; + } + } + } + + AssertLog.info("流量重试文件处理完成，共处理 {} 个文件（时间戳从 {} 到 {}）", + totalProcessed, firstTimestamp, lastTimestamp); + } + + private void processSingleFile(File file, boolean lastTraffic) { + // 使用Properties类读取，并指定UTF-8编码 + Properties props = new Properties(); + try (InputStream input = Files.newInputStream(file.toPath()); + InputStreamReader reader = new InputStreamReader(input, StandardCharsets.UTF_8)) { // 关键：指定UTF-8编码 + + props.load(reader); // 使用指定编码的Reader加载 + String data = props.getProperty("traffic"); + + if (StringUtils.isNotBlank(data)) { + Message message = Message.builder() + .clientId(GlobalConfig.CLIENT_ID) + .dataType(MsgEnum.网络上报重试.getValue()) + .data(data) + .lastTrafficFlag(lastTraffic) + .build(); + + sessionManager.writeAndFlush( + sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), + message + ); + + if (lastTraffic) { + AssertLog.info("发送最后一条重试网络信息包={}", JSON.toJSONString(message)); + } else { + AssertLog.info("发送重试recover网络信息包={}", JSON.toJSONString(message)); + } + } else { + AssertLog.info("文件中没有有效数据，删除文件：{}", file.getName()); + file.delete(); + } + } catch (Exception e) { + AssertLog.error("loadNetFile异常，文件：{},异常信息:{}", file.getName(), e.getMessage()); + } + } public void caseTypeBySystem(String type, int interval, boolean collect){ diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index 9efad49..48e360c 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -17,7 +17,7 @@ logging: netty: server: host: 120.211.95.173 - port: 6620 + port: 56620 client: client-id: client-001 reconnect-interval: 5 diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index fd11e1e..1be839a 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -6,13 +6,14 @@ spring: matching-strategy: ant_path_matcher application: name: tr-agent-client - version: 1.1.10 + version: 1.1.13 conf-path: /usr/local/tongran/conf script-path: /usr/local/tongran/sbin tmp-path: /usr/local/tongran/tmp temp-path: /usr/local/tongran/temp frp-path: /usr/local/tongran/frp poe-path: /usr/local/tongran/sbin/pppoe1 + temp-traffic-path: /usr/local/tongran/traffictemp web: resources: static-locations: classpath*:/META-INF/resources/