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

1、开发agent本地存储流量信息到文件功能。

Browse Source 
2、开发agent每隔2小时上报本地存储文件功能。
This commit is contained in:
gaoyutao
2026-01-20 18:05:24 +08:00
parent 6c5010119f
commit 3820d11353
9 changed files with 207 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/tongran/agent/client/core/config/ApplicationProperties.java
+1
View File
@@ -17,4 +17,5 @@ public class ApplicationProperties {
private String tempPath;
private String frpPath;
private String poePath;
private String tempTrafficPath;
}
src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java
+2
View File
@@ -39,6 +39,8 @@ public enum MsgEnum {
网络上报("NET"),
网络上报重试("NET_RECOVER"),
挂载上报("POINT"),
交换机上报("SWITCHBOARD"),
src/main/java/com/tongran/agent/client/netty/model/Message.java
+2
View File
@@ -33,5 +33,7 @@ public class Message implements Serializable {
* 发送内容
*/
private String data;
/** 最后一次标志 */
private boolean lastTrafficFlag;
}
src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java
+3
View File
@@ -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{
src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java
+28 -1
View File
@@ -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);
}
//
src/main/java/com/tongran/agent/client/service/AgentService.java
+1
View File
@@ -46,6 +46,7 @@ public interface AgentService {
void upFrpcMsg();
void handleRebootRecovery();
void checkTraffic();
List<MacVlanVO> upMacvlanStatus();
}
src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java
+167 -2
View File
@@ -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<File> getTrafficFiles() {
File dir = new File(properties.getTempTrafficPath());
File[] files = dir.listFiles((d, name) -> name.endsWith(".conf"));
if (files == null) {
return new ArrayList<>();
}
// 直接按文件名排序（时间戳是数字，字符串排序也有效）
List<File> fileList = Arrays.asList(files);
fileList.sort(Comparator.comparing(File::getName));
return fileList;
}
@Override
public void checkTraffic() {
List<File> 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<File> 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){
src/main/resources/application-dev.yml
+1 -1
View File
@@ -17,7 +17,7 @@ logging:
netty:
server:
host: 120.211.95.173
port: 6620
port: 56620
client:
client-id: client-001
reconnect-interval: 5
src/main/resources/application.yml
+2 -1
View File
@@ -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/