From ca8f8ee257501ee71fdd70fefc18b69bc3156f24 Mon Sep 17 00:00:00 2001 From: qiminbao Date: Thu, 25 Sep 2025 17:19:40 +0800 Subject: [PATCH] =?UTF-8?q?Agent=E7=89=88=E6=9C=AC=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/core/eo/AgentVersionUpdateEO.java | 3 +- .../client/netty/enpoint/AgentEndpoint.java | 136 ++++++------------ .../netty/handler/AgentDecoderHandler.java | 2 +- .../service/AdvancedAsyncDownloader.java | 4 +- .../task/SpecificTimeTaskService.java | 62 +++++--- .../agent/client/service/AgentService.java | 2 +- .../client/service/impl/AgentServiceImpl.java | 121 +++++++++------- 7 files changed, 156 insertions(+), 174 deletions(-) diff --git a/src/main/java/com/tongran/agent/client/core/eo/AgentVersionUpdateEO.java b/src/main/java/com/tongran/agent/client/core/eo/AgentVersionUpdateEO.java index 6c5b328..6408c1d 100644 --- a/src/main/java/com/tongran/agent/client/core/eo/AgentVersionUpdateEO.java +++ b/src/main/java/com/tongran/agent/client/core/eo/AgentVersionUpdateEO.java @@ -5,7 +5,6 @@ import lombok.Builder; import lombok.Data; import lombok.NoArgsConstructor; -import java.time.LocalDateTime; import java.util.List; @Data @@ -27,7 +26,7 @@ public class AgentVersionUpdateEO { private int method; //定时时间，执行方式为1、定时执行时该字段必传 - private LocalDateTime policyTime; + private long policyTime; //时间戳 private long timestamp; diff --git a/src/main/java/com/tongran/agent/client/netty/enpoint/AgentEndpoint.java b/src/main/java/com/tongran/agent/client/netty/enpoint/AgentEndpoint.java index a1ebab9..ccee18c 100644 --- a/src/main/java/com/tongran/agent/client/netty/enpoint/AgentEndpoint.java +++ b/src/main/java/com/tongran/agent/client/netty/enpoint/AgentEndpoint.java @@ -14,9 +14,6 @@ import com.tongran.agent.client.netty.basics.AgentHandler; import com.tongran.agent.client.netty.model.UpMsgResponse; import com.tongran.agent.client.scheduler.service.AdvancedAsyncDownloader; import com.tongran.agent.client.scheduler.service.AppInitializer; -import com.tongran.agent.client.scheduler.service.AsyncCommandExecutor; -import com.tongran.agent.client.scheduler.service.Java8FileDownloader; -import com.tongran.agent.client.scheduler.task.SpecificTimeRequest; import com.tongran.agent.client.scheduler.task.SpecificTimeTaskService; import com.tongran.agent.client.service.AgentService; import com.tongran.agent.client.utils.AgentDataUtil; @@ -27,9 +24,8 @@ import org.springframework.stereotype.Component; import javax.annotation.Resource; import java.io.IOException; -import java.util.*; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; +import java.util.LinkedHashMap; +import java.util.List; @Component public class AgentEndpoint { @@ -300,7 +296,7 @@ public class AgentEndpoint { GlobalConfig.DOWN_FILES.put(policy.getPolicyName(), count--); }else{ //所有文件下载完成，执行脚本命令 - agentService.command(policy, clientId); + agentService.command(policy, clientId,MsgEnum.执行脚本策略应答.getValue()); } } catch (IOException e) { e.printStackTrace(); @@ -338,12 +334,12 @@ public class AgentEndpoint { } //判断是否需要保存文件 if(!isFalse){ - agentService.command(policy, clientId); + agentService.command(policy, clientId,MsgEnum.执行脚本策略应答.getValue()); return null; }else{ JSONObject json = new JSONObject(); json.put("resCode",1); - json.put("resMsg", "A执行脚本策略文件保存中"); + json.put("resMsg", "执行脚本策略文件保存中"); return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.执行脚本策略应答.getValue()).content(json.toString()).build(); } } @@ -354,96 +350,50 @@ public class AgentEndpoint { @Override public UpMsgResponse upHandle(String data, String clientId) { AgentVersionUpdateEO versionUpdateEO = JSON.parseObject(data, AgentVersionUpdateEO.class); - boolean isFalse = false; - if(StringUtils.isNotBlank(versionUpdateEO.getFileUrl()) && StringUtils.isNotBlank(versionUpdateEO.getFilePath())){ - Java8FileDownloader.DownloadResult result = Java8FileDownloader.downloadFile(versionUpdateEO.getFileUrl(), versionUpdateEO.getFilePath(), ""); - // 验证下载是否完成 - if (result.isSuccess()) { - boolean isComplete = Java8FileDownloader.verifyFileCompletion( - result.getFilePath(), result.getExpectedSize()); - if(isComplete){ - isFalse = true; - System.out.println("Agent版本更新下载完成"); - }else{ - isFalse = false; - System.err.println("Agent版本更新下载失败"); - } - } + ScriptPolicyEO policy = ScriptPolicyEO.builder() + .method(versionUpdateEO.getMethod()) + .policyTime(versionUpdateEO.getPolicyTime()) + .build(); + JSONObject jsonObject = JSONObject.parseObject(data); + if(jsonObject.containsKey("commands")) { + String commands = jsonObject.getString("commands"); + List commandList = JSON.parseObject(commands, new TypeReference>() {}); + policy.setCommands(commandList); } - //判断文件是否保存成功 - if(isFalse){ - if(CollectionUtil.isNotEmpty(versionUpdateEO.getCommands())){ - //执行方式：0、立即执行；1、定时执行； - if(versionUpdateEO.getMethod() == 1){ - SpecificTimeRequest request = SpecificTimeRequest.builder() - .taskId("") - .taskName("") - .taskData(versionUpdateEO.getCommands()) - .specificDateTime(versionUpdateEO.getPolicyTime()) - .clientId(clientId) - .dataType(MsgEnum.Agent版本更新应答.getValue()) - .build(); - try { - taskService.createSpecificTimeTask(request); - JSONObject json = new JSONObject(); - json.put("resCode",1); - json.put("resMsg", "Agent版本更新定时任务保存成功"); -// Message message = Message.builder().clientId(clientId).dataType(MsgEnum.Agent更新应答.getValue()).data(json.toString()).build(); - return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.Agent版本更新应答.getValue()).content(json.toString()).build(); - } catch (Exception e) { - JSONObject json = new JSONObject(); - json.put("resCode",0); - json.put("resMsg", "Agent版本更新定时任务保存失败"); -// Message message = Message.builder().clientId(clientId).dataType(MsgEnum.Agent更新应答.getValue()).data(json.toString()).build(); - return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.Agent版本更新应答.getValue()).content(json.toString()).build(); - } - }else{ - List> list = new ArrayList<>(); - for (String command : versionUpdateEO.getCommands()) { - Map map = new HashMap<>(); - List cmd = Arrays.asList(command.split("\\s+")); - CompletableFuture future = - AsyncCommandExecutor.executeCommandAsync( - cmd, - 60, TimeUnit.SECONDS); - future.thenAccept(result -> { - if (result.isSuccess()) { - System.out.println("脚本执行成功"); - System.out.println("[成功resOut] " + result.getOutput()); - map.put("command",command); - map.put("resOut",result.getOutput()); - } else { - System.out.println("脚本执行失败"); - System.out.println("[失败resOut] " + result.getOutput()); - map.put("command",command); - map.put("resOut",result.getOutput()); - } - }).exceptionally(ex -> { - System.err.println("执行失败: " + ex.getMessage()); - map.put("command",command); - map.put("resOut","Policy execute filed"); - return null; - }); - list.add(map); - } - if(CollectionUtil.isNotEmpty(list)){ - JSONObject json = new JSONObject(); - json.put("resCode",1); - json.put("resMsg", ""); - json.put("result", AgentUtil.toJsonString(list)); -// Message message = Message.builder().clientId(clientId).dataType(MsgEnum.Agent版本更新应答.getValue()).data(json.toString()).build(); - return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.Agent版本更新应答.getValue()).content(json.toString()).build(); - } + boolean isFalse = false; + if(StringUtils.isNotBlank(versionUpdateEO.getFileUrl())){ + isFalse = true; + String finalSaveDir = versionUpdateEO.getFilePath(); + AdvancedAsyncDownloader.downloadWithProgress(versionUpdateEO.getFileUrl(), finalSaveDir, null, progress -> { + System.out.printf("下载进度: %.1f%%\n", progress); + }).thenAccept(filePath -> { + System.out.println("下载完成: " + filePath); + try { + //更改全局变量 + GlobalConfig.isCollect = false; + //调用采集任务 + agentService.cancelCollect(); + //所有文件下载完成，执行脚本命令 + agentService.command(policy, clientId,MsgEnum.Agent版本更新应答.getValue()); + } catch (Exception e) { + e.printStackTrace(); } - } + }).exceptionally(ex -> { + System.err.println("下载错误: " + ex.getMessage()); + return null; + }); + } + //判断是否需要保存文件 + if(!isFalse){ + agentService.command(policy, clientId,MsgEnum.Agent版本更新应答.getValue()); + return null; }else{ JSONObject json = new JSONObject(); - json.put("resCode",0); - json.put("resMsg", "Agent版本更新文件保存失败"); -// Message message = Message.builder().clientId(clientId).dataType(MsgEnum.Agent版本更新应答.getValue()).data(json.toString()).build(); + json.put("resCode",1); + json.put("resMsg", "Agent版本更新文件保存中"); return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.Agent版本更新应答.getValue()).content(json.toString()).build(); } - return null; + } } diff --git a/src/main/java/com/tongran/agent/client/netty/handler/AgentDecoderHandler.java b/src/main/java/com/tongran/agent/client/netty/handler/AgentDecoderHandler.java index 22017b7..a12ba6c 100644 --- a/src/main/java/com/tongran/agent/client/netty/handler/AgentDecoderHandler.java +++ b/src/main/java/com/tongran/agent/client/netty/handler/AgentDecoderHandler.java @@ -124,8 +124,8 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter { AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value); if (ObjectUtil.isNotEmpty(msgHandler)) { UpMsgResponse response = msgHandler.upHandle(data, clientId ); - AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message); if(Objects.nonNull(response)){ + AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message); Message agentMessage = Message.builder().build(); agentMessage.setClientId(response.getClientId()); agentMessage.setDataType(response.getDataType()); diff --git a/src/main/java/com/tongran/agent/client/scheduler/service/AdvancedAsyncDownloader.java b/src/main/java/com/tongran/agent/client/scheduler/service/AdvancedAsyncDownloader.java index 66bd47c..de7c711 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/service/AdvancedAsyncDownloader.java +++ b/src/main/java/com/tongran/agent/client/scheduler/service/AdvancedAsyncDownloader.java @@ -68,8 +68,8 @@ public class AdvancedAsyncDownloader { } }, downloadExecutor); } - - private static String extractFileNameFromUrl(String fileUrl) { + + public static String extractFileNameFromUrl(String fileUrl) { // 实现文件名提取逻辑 return fileUrl.substring(fileUrl.lastIndexOf('/') + 1); } diff --git a/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java b/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java index 8f01f41..54acc36 100644 --- a/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java +++ b/src/main/java/com/tongran/agent/client/scheduler/task/SpecificTimeTaskService.java @@ -2,13 +2,16 @@ package com.tongran.agent.client.scheduler.task; import cn.hutool.core.collection.CollectionUtil; import com.alibaba.fastjson2.JSONObject; +import com.tongran.agent.client.core.enums.MsgEnum; import com.tongran.agent.client.core.session.SessionManager; import com.tongran.agent.client.netty.model.Message; import com.tongran.agent.client.scheduler.service.AsyncCommandExecutor; import com.tongran.agent.client.utils.AgentUtil; +import org.apache.commons.lang3.StringUtils; import org.springframework.stereotype.Service; import javax.annotation.Resource; +import java.io.IOException; import java.time.LocalDateTime; import java.util.*; import java.util.concurrent.CompletableFuture; @@ -73,30 +76,43 @@ public class SpecificTimeTaskService { List> list = new ArrayList<>(); for (String command : request.getTaskData()) { Map map = new HashMap<>(); - List cmd = Arrays.asList(command.split("\\s+")); - CompletableFuture future = - AsyncCommandExecutor.executeCommandAsync( - cmd, - 60, TimeUnit.SECONDS); - future.thenAccept(result -> { - if (result.isSuccess()) { - System.out.println("脚本执行成功"); - System.out.println("[成功resOut] " + result.getOutput()); - map.put("command",command); - map.put("resOut",result.getOutput()); - } else { - System.out.println("脚本执行失败"); - System.out.println("[失败resOut] " + result.getOutput()); - map.put("command",command); - map.put("resOut",result.getOutput()); + if(StringUtils.equals(request.getDataType(), MsgEnum.Agent版本更新应答.getValue())){ + try { + System.out.println("重启进程已启动，当前服务退出"); + System.out.println("command="+command); + ProcessBuilder pb = new ProcessBuilder("/bin/sh", "-c", + command); + pb.start(); + System.exit(0); + } catch (IOException e) { + e.printStackTrace(); } - }).exceptionally(ex -> { - System.err.println("执行失败: " + ex.getMessage()); - map.put("command",command); - map.put("resOut","Policy execute filed"); - return null; - }); - list.add(map); + }else{ + List cmd = Arrays.asList(command.split("\\s+")); + CompletableFuture future = + AsyncCommandExecutor.executeCommandAsync( + cmd, + 100, TimeUnit.SECONDS); + future.thenAccept(result -> { + if (result.isSuccess()) { + System.out.println("脚本执行成功"); + System.out.println("[成功resOut] " + result.getOutput()); + map.put("command",command); + map.put("resOut",result.getOutput()); + } else { + System.out.println("脚本执行失败"); + System.out.println("[失败resOut] " + result.getOutput()); + map.put("command",command); + map.put("resOut",result.getOutput()); + } + }).exceptionally(ex -> { + System.err.println("执行失败: " + ex.getMessage()); + map.put("command",command); + map.put("resOut","Policy execute filed"); + return null; + }); + list.add(map); + } } if(CollectionUtil.isNotEmpty(list)){ JSONObject json = new JSONObject(); 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 097a36b..217d8e9 100644 --- a/src/main/java/com/tongran/agent/client/service/AgentService.java +++ b/src/main/java/com/tongran/agent/client/service/AgentService.java @@ -15,5 +15,5 @@ public interface AgentService { void alarmMonitor(); - void command(ScriptPolicyEO policy, String clientId); + void command(ScriptPolicyEO policy, String clientId, String dataType); } 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 884e35c..2133e50 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 @@ -24,7 +24,10 @@ import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import javax.annotation.Resource; -import java.util.*; +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -675,7 +678,7 @@ public class AgentServiceImpl implements AgentService { } @Override - public void command(ScriptPolicyEO policy, String clientId) { + public void command(ScriptPolicyEO policy, String clientId, String dataType) { long timestamp = System.currentTimeMillis(); timestamp = Math.round(timestamp / 1000.0); if(CollectionUtil.isNotEmpty(policy.getCommands())){ @@ -687,6 +690,7 @@ public class AgentServiceImpl implements AgentService { .taskData(policy.getCommands()) .specificDateTime(AgentUtil.toLocalDateTime(policy.getPolicyTime(),true)) .clientId(clientId) + .dataType(dataType) .build(); try { taskService.createSpecificTimeTask(request); @@ -716,57 +720,70 @@ public class AgentServiceImpl implements AgentService { } }else{ for (String command : policy.getCommands()) { - List cmd = Arrays.asList(command.split("\\s+")); - CompletableFuture future = - AsyncCommandExecutor.executeCommandAsync( - cmd, - 60, TimeUnit.SECONDS); - future.thenAccept(result -> { - if (result.isSuccess()) { - System.out.println("脚本执行成功"); - System.out.println("[成功resOut] " + result.getOutput()); - } else { - System.out.println("脚本执行失败"); - System.out.println("[失败resOut] " + result.getOutput()); + if(StringUtils.equals(dataType, MsgEnum.Agent版本更新应答.getValue())){ + try { + System.out.println("重启进程已启动，当前服务退出"); + System.out.println("command="+command); + ProcessBuilder pb = new ProcessBuilder("/bin/sh", "-c", + command); + pb.start(); + System.exit(0); + } catch (IOException e) { + e.printStackTrace(); } - JSONObject rse = new JSONObject(); - rse.put("command",command); - rse.put("resOut", result.getOutput()); - long timestamps = System.currentTimeMillis(); - timestamps = Math.round(timestamps / 1000.0); - JSONObject json = new JSONObject(); - json.put("resCode",1); - json.put("resMsg", ""); - json.put("timestamp",timestamps); - json.put("result", rse.toString()); - // 判定客户端与服务端是否连接 - if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { - Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.执行脚本策略应答.getValue()) - .data(json.toString()).build(); - sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); - AssertLog.info("发送执行脚本策略应答={}",JSON.toJSONString(message)); - } - }).exceptionally(ex -> { - System.err.println("执行失败: " + ex.getMessage()); - JSONObject rse = new JSONObject(); - rse.put("command",command); - rse.put("resOut", "脚本执行失败"); - long timestamps = System.currentTimeMillis(); - timestamps = Math.round(timestamps / 1000.0); - JSONObject json = new JSONObject(); - json.put("resCode",1); - json.put("resMsg", ""); - json.put("timestamp",timestamps); - json.put("result", rse.toString()); - // 判定客户端与服务端是否连接 - if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { - Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(MsgEnum.执行脚本策略应答.getValue()) - .data(json.toString()).build(); - sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); - AssertLog.info("发送执行脚本策略应答={}",JSON.toJSONString(message)); - } - return null; - }); + }else{ + List cmd = Arrays.asList(command.split("\\s+")); + CompletableFuture future = + AsyncCommandExecutor.executeCommandAsync( + cmd, + 100, TimeUnit.SECONDS); + future.thenAccept(result -> { + if (result.isSuccess()) { + System.out.println("脚本执行成功"); + System.out.println("[成功resOut] " + result.getOutput()); + } else { + System.out.println("脚本执行失败"); + System.out.println("[失败resOut] " + result.getOutput()); + } + JSONObject rse = new JSONObject(); + rse.put("command",command); + rse.put("resOut", result.getOutput()); + long timestamps = System.currentTimeMillis(); + timestamps = Math.round(timestamps / 1000.0); + JSONObject json = new JSONObject(); + json.put("resCode",1); + json.put("resMsg", ""); + json.put("timestamp",timestamps); + json.put("result", rse.toString()); + // 判定客户端与服务端是否连接 + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(dataType) + .data(json.toString()).build(); + sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); + AssertLog.info("发送执行脚本策略应答={}",JSON.toJSONString(message)); + } + }).exceptionally(ex -> { + System.err.println("执行失败: " + ex.getMessage()); + JSONObject rse = new JSONObject(); + rse.put("command",command); + rse.put("resOut", "脚本执行失败"); + long timestamps = System.currentTimeMillis(); + timestamps = Math.round(timestamps / 1000.0); + JSONObject json = new JSONObject(); + json.put("resCode",1); + json.put("resMsg", ""); + json.put("timestamp",timestamps); + json.put("result", rse.toString()); + // 判定客户端与服务端是否连接 + if (Objects.nonNull(sessionManager.getSessionById(GlobalConfig.CLIENT_ID))) { + Message message = Message.builder().clientId(GlobalConfig.CLIENT_ID).dataType(dataType) + .data(json.toString()).build(); + sessionManager.writeAndFlush(sessionManager.getSessionById(GlobalConfig.CLIENT_ID).getChannel(), message); + AssertLog.info("发送执行脚本策略应答={}",JSON.toJSONString(message)); + } + return null; + }); + } } } }