调试agent本地存储流量功能

src/main/java/com/tongran/agent/client/core/enums/MsgEnum.java

@@ -75,6 +75,8 @@ public enum MsgEnum {
 
     iops结果上报("IOPS_RESULT"),
 
+    tcpdump结果上报("TCPDUMP_RESULT"),
+
     多网IP探测上报("NETWORK_DETECT");
 
     private String value;

src/main/java/com/tongran/agent/client/scheduler/service/AppInitializer.java

@@ -105,8 +105,8 @@ public class AppInitializer implements CommandLineRunner {
                 AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000);
                 dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000);
                 long milliFive = AgentUtil.millisecondsToNext5Minute();
-                AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000);
+                AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 7200000);
-                dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000);
+                dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000);
                 AssertLog.info("检测PppoE配置");
                 agentService.handleRebootRecovery();
             }else{

src/main/java/com/tongran/agent/client/scheduler/service/BusinessTasks.java

@@ -876,8 +876,8 @@ public class BusinessTasks {
                 AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000);
                 dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000);
                 long milliFive = AgentUtil.millisecondsToNext5Minute();
-                AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000);
+                AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 7200000);
-                dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000);
+                dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000);
                 AssertLog.info("检测PppoE配置");
                 agentService.handleRebootRecovery();
             }else{

src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java

@@ -1,5 +1,8 @@
 package com.tongran.agent.client.scheduler.task;
 
+import cn.hutool.core.collection.CollectionUtil;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONArray;
 import com.alibaba.fastjson2.JSONObject;
 import com.tongran.agent.client.core.config.ApplicationProperties;
 import com.tongran.agent.client.core.config.GlobalConfig;
@@ -11,15 +14,18 @@ import com.tongran.agent.client.service.AgentService;
 import com.tongran.agent.client.utils.AgentDataUtil;
 import com.tongran.agent.client.utils.AgentUtil;
 import com.tongran.agent.client.utils.AssertLog;
+import lombok.AllArgsConstructor;
+import lombok.Data;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
+import java.io.BufferedReader;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.time.LocalDateTime;
-import java.util.Arrays;
+import java.time.format.DateTimeFormatter;
-import java.util.List;
+import java.util.*;
-import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
@@ -40,6 +46,20 @@ public class SpecificTimeTaskService {
         this.sessionManager = SessionManager.getInstance();
     }
 
+    public boolean createTcpdumpSpecificTimeTask(SpecificTimeRequest request){
+        String taskId = request.getTaskId();
+        Runnable task = createTcpdumpTaskRunnable(request);
+        try {
+            if (request.getTime() != null) {
+                taskConfig.addDailyTimeTask(taskId, task, request.getTime());
+            } else {
+                throw new IllegalArgumentException("必须指定一种时间方式");
+            }
+            return true;
+        } catch (Exception e) {
+            throw new RuntimeException("创建任务失败: " + e.getMessage(), e);
+        }
+    }
     /**
      * 创建指定时间任务
      */
@@ -67,6 +87,141 @@ public class SpecificTimeTaskService {
         }
     }
     
+    private Runnable createTcpdumpTaskRunnable(SpecificTimeRequest request) {
+        return () -> {
+            System.out.println("执行定时任务: " + request.getTaskName());
+            System.out.println("执行时间: " + LocalDateTime.now());
+            System.out.println("-----------------------------------");
+            
+            // 具体的业务逻辑
+            executeTcpdump(request);
+        };
+    }
+
+    private void executeTcpdump(SpecificTimeRequest request) {
+        try {
+            String timestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"));
+            String pcapFileName = timestamp + ".pcap";
+
+            // 第一步：执行tcpdump命令捕获数据
+            ProcessBuilder captureBuilder = new ProcessBuilder();
+            captureBuilder.command("timeout", "10", "tcpdump", "-i", "eth0", "-nn", "-n", "-w", pcapFileName);
+
+            Process captureProcess = captureBuilder.start();
+            int captureExitCode = captureProcess.waitFor();
+
+            if (captureExitCode != 0) {
+                AssertLog.error("Tcpdump capture failed with exit code: {}", captureExitCode);
+                return;
+            }
+
+            // 等待文件写入完成
+            Thread.sleep(1000);
+
+            // 第二步：分析pcap文件并统计目标IP
+            ProcessBuilder analyzeBuilder = new ProcessBuilder();
+            analyzeBuilder.command("bash", "-c",
+                    "tcpdump -r " + pcapFileName + " -nn | awk '/IP/{split($5,a,\".\"); print a[1]\".\"a[2]\".\"a[3]\".\"a[4]}' | sort | uniq -c | sort -rn");
+
+            Process analyzeProcess = analyzeBuilder.start();
+
+            // 读取分析结果
+            BufferedReader reader = new BufferedReader(new InputStreamReader(analyzeProcess.getInputStream()));
+            StringBuilder result = new StringBuilder();
+            String line;
+
+            while ((line = reader.readLine()) != null) {
+                result.append(line).append("\n");
+            }
+
+            int analyzeExitCode = analyzeProcess.waitFor();
+
+            if (analyzeExitCode == 0) {
+                AssertLog.info("Tcpdump analysis completed successfully for task: {}", request.getTaskId());
+                AssertLog.info("IP statistics:\n{}", result.toString());
+
+                // 解析并保存结果
+                saveTcpdumpResult(request, pcapFileName, result.toString());
+            } else {
+                AssertLog.error("Tcpdump analysis failed with exit code: {}", analyzeExitCode);
+            }
+
+        } catch (IOException | InterruptedException e) {
+            AssertLog.error("Error executing tcpdump task: {}", request.getTaskId(), e);
+            Thread.currentThread().interrupt();
+        }
+    }
+
+    private void saveTcpdumpResult(SpecificTimeRequest request, String fileName, String statistics) {
+        try {
+            // 解析统计结果
+            List<IpCount> ipCountList = parseTcpdumpOutput(statistics);
+
+            AssertLog.info("Saving tcpdump result - Task: {}, File: {}, Total unique IPs: {}",
+                    request.getTaskId(), fileName, ipCountList.size());
+            String data = "";
+            if(CollectionUtil.isNotEmpty(ipCountList)){
+                data = JSONArray.toJSONString(ipCountList);
+            }
+            Message message = Message.builder().clientId(request.getClientId()).dataType(MsgEnum.tcpdump结果上报.getValue()).data(data).build();
+            if (Objects.nonNull(sessionManager.getSessionById(request.getClientId()))) {
+                sessionManager.writeAndFlush(sessionManager.getSessionById(request.getClientId()).getChannel(), message);
+                AssertLog.info("发送tcpdump包={}", JSON.toJSONString(message));
+            }
+        } catch (Exception e) {
+            AssertLog.error("Failed to parse tcpdump result for task: {}", request.getTaskId(), e);
+        }
+    }
+
+    /**
+     * 解析tcpdump输出为IP统计列表
+     */
+    private List<IpCount> parseTcpdumpOutput(String output) {
+        List<IpCount> ipCountList = new ArrayList<>();
+
+        if (output == null || output.trim().isEmpty()) {
+            return ipCountList;
+        }
+
+        String[] lines = output.split("\n");
+
+        for (String line : lines) {
+            line = line.trim();
+
+            // 跳过空行和标题行
+            if (line.isEmpty() || line.contains("reading from file") || line.contains("link-type")) {
+                continue;
+            }
+
+            // 解析格式： "   455 172.16.15.52"
+            String[] parts = line.trim().split("\\s+", 2); // 按空格分割，最多分成2部分
+
+            if (parts.length >= 2) {
+                try {
+                    int count = Integer.parseInt(parts[0].trim());
+                    String ip = parts[1].trim();
+                    ipCountList.add(new IpCount(ip, count));
+                } catch (NumberFormatException e) {
+                    AssertLog.warn("Failed to parse count from line: {}", line);
+                }
+            }
+        }
+
+        return ipCountList;
+    }
+
+
+    /**
+     * IP统计实体类
+     */
+    @Data
+    @AllArgsConstructor
+    public static class IpCount {
+        private String ip;
+        private int count;
+    }
+
+
     private Runnable createTaskRunnable(SpecificTimeRequest request) {
         return () -> {
             System.out.println("执行定时任务: " + request.getTaskName());
@@ -153,4 +308,16 @@ public class SpecificTimeTaskService {
             System.err.println("任务执行异常: " + e.getMessage());
         }
     }
+
+    public void cancleTcpdumpSpecificTimeTask() {
+        // 获取所有任务ID
+        Set<String> taskIds = taskConfig.getTaskMap().keySet();
+
+        // 遍历并关闭所有以"tcpdump"开头的任务
+        for (String taskId : taskIds) {
+            if (taskId.startsWith("tcpdump")) {
+                taskConfig.removeTask(taskId);
+            }
+        }
+    }
 }

src/main/java/com/tongran/agent/client/service/impl/AgentServiceImpl.java

@@ -322,8 +322,8 @@ public class AgentServiceImpl implements AgentService {
         AssertLog.info("启动frpc状态上报定时任务 - 延迟: {}ms, 间隔: {}ms", 15000, 300000);
         dynamicTaskService.scheduleTask("upFrpcMsg", businessTasks::upFrpcMsgTask, 15000, 300000);
         long milliFive = AgentUtil.millisecondsToNext5Minute();
-        AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 960000);
+        AssertLog.info("启动本地存储流量信息上报定时任务 - 延迟: {}ms, 间隔: {}ms", milliFive, 7200000);
-        dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 960000);
+        dynamicTaskService.scheduleTask("upTempTraffic", businessTasks::upTempTraffic, milliFive, 7200000);
         AssertLog.info("检测PppoE配置");
         handleRebootRecovery();
     }
@@ -920,21 +920,42 @@ public class AgentServiceImpl implements AgentService {
         }
         if(jsonObject.containsKey("trafficUsedTimestamp")){
             String trafficUsedTimestamp = jsonObject.getString("trafficUsedTimestamp");
-//            if (trafficUsedTimestamp != null) {
+            if (trafficUsedTimestamp != null) {
-//                String[] fileNames = trafficUsedTimestamp.split(",");
+                String[] fileNames = trafficUsedTimestamp.split(",");
-//
+
-//                // 删除对应文件
+                // 删除对应文件
-//                for (String fileName : fileNames) {
+                for (String fileName : fileNames) {
-//                    if (StringUtils.isNotBlank(fileName)) {
+                    if (StringUtils.isNotBlank(fileName)) {
-//                        fileName = fileName+"conf";
+                        fileName = fileName+"conf";
-//                        String filePath = properties.getTempTrafficPath() + fileName.trim();
+                        String filePath = properties.getTempTrafficPath() + fileName.trim();
-//                        AssertLog.info("已处理文件：{}", fileName);
+                        AssertLog.info("已处理文件：{}", fileName);
-//                        deleteTrafficFile(filePath);
+                        deleteTrafficFile(filePath);
-//                    }
+                    }
-//                }
+                }
-//            }
+            }
+        }
+        if(jsonObject.containsKey("tcpdumpTimes")){
+            String tcpdumpTimes = jsonObject.getString("tcpdumpTimes");
+            handleTcpdump(tcpdumpTimes);
         }
     }
+
+    private void handleTcpdump(String tcpdumpTimes) {
+        taskService.cancleTcpdumpSpecificTimeTask();
+        if(!"".equals(tcpdumpTimes)){
+            String[] tcpdumpTimeArr = tcpdumpTimes.split(",");
+            for (String tcpdumpTime : tcpdumpTimeArr) {
+                SpecificTimeRequest request = SpecificTimeRequest.builder()
+                        .taskId("tcpdump"+tcpdumpTime)
+                        .taskName("tcpdump"+tcpdumpTime)
+                        .time(tcpdumpTime)
+                        .clientId(GlobalConfig.CLIENT_ID)
+                        .build();
+                taskService.createTcpdumpSpecificTimeTask(request);
+            }
+        }
+    }
+
     private void deleteTrafficFile(String filePath) {
         try {
             File file = new File(filePath);
@@ -2314,8 +2335,8 @@ public class AgentServiceImpl implements AgentService {
 
         // 配置参数
         int batchSize = 10;           // 每批10个文件
-        long intraBatchInterval = 5000; // 批次内文件间隔5s
+        long intraBatchInterval = 1000; // 批次内文件间隔5s
-        long interBatchInterval = 1000; // 批次间间隔10秒
+        long interBatchInterval = 1000; // 批次间间隔1秒
 
         int totalProcessed = 0;
         int totalFiles = fileList.size();