worker修改

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/worker/BaseWorker.java

@@ -17,6 +17,8 @@ import cd.casic.ci.process.process.service.machine.MachineInfoService;
 import cd.casic.ci.process.ssh.SshClient;
 import cd.casic.ci.process.ssh.SshClientFactory;
 import cd.casic.ci.process.ssh.WinRMHelper;
+import cd.casic.framework.commons.exception.ServiceException;
+import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants;
 import jakarta.annotation.Resource;
 
 import lombok.Data;
@@ -67,5 +69,7 @@ public abstract class BaseWorker implements Runnable{
         }
     }
     public abstract void execute(TaskRunContext context);
-
+    public void toBadEnding(){
+        throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"");
+    }
 }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/worker/DIYImageExecuteCommandWorker.java

@@ -2,8 +2,10 @@ package cd.casic.ci.process.engine.worker;
 
 import cd.casic.ci.common.pipeline.annotation.Plugin;
 import cd.casic.ci.process.engine.constant.DIYImageExecuteCommandConstant;
+import cd.casic.ci.process.engine.constant.EngineRuntimeConstant;
 import cd.casic.ci.process.engine.runContext.TaskRunContext;
 import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
+import cd.casic.ci.process.process.dataObject.log.PipTaskLog;
 import cd.casic.ci.process.process.dataObject.machine.MachineInfo;
 import cd.casic.ci.process.process.dataObject.task.PipTask;
 import cd.casic.framework.commons.exception.ServiceException;
@@ -23,15 +25,15 @@ import java.util.concurrent.atomic.AtomicInteger;
  * @date 2022-11-08 9:59
  */
 @Slf4j
-@Plugin(taskType = "CUSTOM_IMAGE_EXECUTION_COMMAND")
+//@Plugin(taskType = "CUSTOM_IMAGE_EXECUTION_COMMAND")
+@Plugin(taskType = "test")
 public class DIYImageExecuteCommandWorker extends SshWorker {
 
-
-
     @Override
     public void execute(TaskRunContext context) {
         int statusCode = -1;
-        AtomicInteger state = context.getState();
+        Map<String, Object> localVariables = context.getLocalVariables();
+        PipTaskLog taskLog = (PipTaskLog) localVariables.get(EngineRuntimeConstant.LOG_KEY);
         if (context.getContextDef() instanceof PipTask taskDef) {
             log.info(taskDef.getTaskName());
             Map<String, Object> taskProperties = taskDef.getTaskProperties();
@@ -43,9 +45,11 @@ public class DIYImageExecuteCommandWorker extends SshWorker {
                 machineId=Long.valueOf(String.valueOf(machineIdObj));
             } catch (NumberFormatException e) {
                 log.error("缺少参数：{}",DIYImageExecuteCommandConstant.MACHINE_ID);
+                toBadEnding();
             }
             if (StringUtils.isEmpty(commandScript) ||machineIdObj == null) {
-                throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"缺少参数");
+//                缺少参数
+                toBadEnding();
             }
 
             try {
@@ -56,21 +60,22 @@ public class DIYImageExecuteCommandWorker extends SshWorker {
 
                 //获取机器
                 MachineInfo machineInfoDO = this.getMachineInfoService().getById(machineId);
-                statusCode = shell(machineInfoDO,
+                statusCode = shell(machineInfoDO,taskLog,
                         "echo \"自定义镜像执行命令\"",
                         commandScript
                 );
             } catch (Exception e) {
                 String errorMessage = "该节点配置信息为空，请先配置该节点信息" + "\r\n";
                 log.error("执行ssh失败:",e);
-//                errorHandle(e, errorMessage);
+                taskLog.append(errorMessage);
+                toBadEnding();
             }
             if (statusCode == 0) {
-//                log.info("{}节点执行完成", getName());
+                log.info("节点执行完成");
             } else {
-//                log.error("{}节点执行失败", getName());
+                log.error("节点执行失败");
             }
-            context.getLocalVariables().put(DIYImageExecuteCommandConstant.STATUS_CODE,statusCode);
+            localVariables.put(DIYImageExecuteCommandConstant.STATUS_CODE,statusCode);
         }
     }
 }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/worker/SshWorker.java

@@ -2,6 +2,7 @@ package cd.casic.ci.process.engine.worker;
 
 import cd.casic.ci.process.constant.CommandConstant;
 import cd.casic.ci.process.enums.MachineSystemEnum;
+import cd.casic.ci.process.process.dataObject.log.PipTaskLog;
 import cd.casic.ci.process.process.dataObject.machine.MachineInfo;
 import cd.casic.ci.process.ssh.SshClient;
 import cd.casic.ci.process.ssh.SshClientFactory;
@@ -10,6 +11,9 @@ import lombok.extern.slf4j.Slf4j;
 
 import java.util.Arrays;
 import java.util.List;
+/**
+ *  使用ssh的worker的基类
+ * */
 @Slf4j
 public abstract class SshWorker extends BaseWorker{
     /**
@@ -18,7 +22,7 @@ public abstract class SshWorker extends BaseWorker{
      * @param commands 命令
      * @return 0 成功；其他值 失败
      */
-    public int shell(MachineInfo machineInfo, String... commands) {
+    public int shell(MachineInfo machineInfo, PipTaskLog taskLog, String... commands) {
         List<String> commandList = Arrays.asList(commands);
         if(MachineSystemEnum.WINDOWS.getSystem().equals(machineInfo.getOsSystem())){
             return powerShell(machineInfo, commandList);
@@ -32,17 +36,16 @@ public abstract class SshWorker extends BaseWorker{
         try {
             ssh = SshClientFactory.createSsh(machineInfo);
             //执行命令，并且把命令的执行回传到前端
-            statusCode = ssh.execNew(commandList, var -> {
+            // TODO 记录日志
-                // TODO 记录日志
+            //                loggerService.sendMessage(key, var);
-//                loggerService.sendMessage(key, var);
+            statusCode = ssh.execNew(commandList, taskLog::append);
-//                nodeLogger.append(var);
-            });
             log.info("exit-status: " + statusCode);
             //主动释放当前socket连接
 //            loggerService.close(key);
         } catch (Exception e) {
             String errorMessage = "与机器建立SSH连接出错" + CommandConstant.ENTER;
 //            errorHandle(e, errorMessage);
+            taskLog.append(errorMessage);
         } finally {
             if(ssh!=null) {
                 ssh.disconnect();

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/worker/TestWorker.java

@@ -9,7 +9,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 
 @Slf4j
-@Plugin(taskType = "test")
+//@Plugin(taskType = "test")
 public class TestWorker extends BaseWorker{
 
 

ops-server/src/test/java/cd/casic/server/MachineTest.java

@@ -0,0 +1,32 @@
+package cd.casic.server;
+
+import cd.casic.ci.process.engine.manager.RunContextManager;
+import cd.casic.ci.process.engine.message.TaskRunMessage;
+import cd.casic.ci.process.engine.runContext.PipelineRunContext;
+import cd.casic.ci.process.engine.runContext.SecondStageRunContext;
+import cd.casic.ci.process.engine.runContext.TaskRunContext;
+import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline;
+import cd.casic.ci.process.process.dataObject.stage.PipStage;
+import cd.casic.ci.process.process.dataObject.task.PipTask;
+import cd.casic.ci.process.util.CryptogramUtil;
+import cd.casic.framework.mq.redis.core.RedisMQTemplate;
+import jakarta.annotation.Resource;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
+
+@SpringBootTest(classes = {OpsServerApplication.class})
+@ActiveProfiles("local")
+public class MachineTest {
+    @Test
+    public void test01(){
+        System.out.println("h w !");
+        String password = "Hnidc@0626cn!@#zyx";
+        String encrypt = CryptogramUtil.doEncrypt(password);
+        System.out.println(encrypt);
+    }
+
+}