worker添加，自定义镜像worker修改，机器管理接口添加

modules/module-ci-common-pipeline/src/main/java/cd/casic/ci/common/pipeline/req/machine/MachineInfoReq.java

@@ -0,0 +1,26 @@
+package cd.casic.ci.common.pipeline.req.machine;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import lombok.Data;
+
+@Data
+public class MachineInfoReq {
+
+    /**
+     * 主机ip
+     */
+    private String machineHost;
+
+    /**
+     * 机器名称
+     */
+    private String machineName;
+
+    /**
+     * 机器状态 1有效 2无效
+     */
+    private String machineStatus;
+
+
+}

modules/module-ci-common-pipeline/src/main/java/cd/casic/ci/common/pipeline/resp/context/SingletonRunContextResp.java

@@ -19,4 +19,5 @@ public class SingletonRunContextResp {
     private Integer state;
     private LocalDateTime startTime;
     private LocalDateTime endTime;
+    private String duration;
 }

modules/module-ci-common-pipeline/src/main/java/cd/casic/ci/common/pipeline/resp/machine/MachineInfoResp.java

@@ -0,0 +1,38 @@
+package cd.casic.ci.common.pipeline.resp.machine;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import lombok.Data;
+
+@Data
+public class MachineInfoResp {
+    /**
+    * 机器描述
+    */
+    private String description;
+    /**
+     * id
+     */
+    @TableId(type = IdType.ASSIGN_ID)
+    private Long id;
+
+    /**
+     * 主机ip
+     */
+    private String machineHost;
+
+    /**
+     * 机器名称
+     */
+    private String machineName;
+
+    /**
+     * 机器状态 1有效 2无效
+     */
+    private String machineStatus;
+
+    /**
+     * 机器唯一标识
+     */
+    private String machineTag;
+}

modules/module-ci-process-biz/pom.xml

@@ -69,7 +69,27 @@
             <artifactId>httpclient5</artifactId>
             <version>5.2.1</version> <!-- 建议使用最新稳定版本 -->
         </dependency>
-
     </dependencies>
-
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>org.projectlombok</groupId>
+                            <artifactId>lombok</artifactId>
+                            <version>${lombok.version}</version>
+                        </path>
+                        <path>
+                            <groupId>org.mapstruct</groupId>
+                            <artifactId>mapstruct-processor</artifactId>
+                            <version>${mapstruct.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
 </project>

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/constant/DIYImageExecuteCommandConstant.java

@@ -8,6 +8,6 @@ public class DIYImageExecuteCommandConstant {
     /**
      * 人工卡点命令脚本
      */
-    public static final String COMMAND_SCRIPT ="commandScript";
+    public static final String COMMAND_SCRIPT ="buildScript";
     public static final String STATUS_CODE = "statusCode";
 }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/enums/ContextStateEnum.java

@@ -39,7 +39,6 @@ public enum ContextStateEnum {
             }
             return TRANSITIONS.get(from).contains(to);
         } catch (Exception e){
-            System.out.println("");
             return false;
         }
     }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/manager/impl/MemoryLogManager.java

@@ -16,10 +16,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.*;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 
 /**
  *  流水线运行时日志管理
@@ -120,23 +117,26 @@ public class MemoryLogManager implements LoggerManager {
 
     private void emitterInit(SseEmitter emitter,String taskId){
         // 维持心跳
-        scheduler.scheduleWithFixedDelay(()-> {
+        ScheduledFuture<?> scheduledFuture = scheduler.scheduleWithFixedDelay(() -> {
             try {
-                emitter.send("\n\n");
+                emitter.send(SseEmitter.event().comment("heartbeat"));
             } catch (IOException e) {
                 log.error("", e);
             }
         }, 0, 30, TimeUnit.SECONDS);
         emitter.onCompletion(()->{
             taskIdSSEMap.remove(taskId);
+            scheduledFuture.cancel(true);
             log.info("===================taskId:{}断开连接===============",taskId);
         });
         emitter.onError((e)->{
             taskIdSSEMap.remove(taskId);
+            scheduledFuture.cancel(true);
             log.error("===================错误，taskId:{}断开连接===============",taskId,e);
         });
         emitter.onTimeout(()->{
             taskIdSSEMap.remove(taskId);
+            scheduledFuture.cancel(true);
             log.error("===================超时，taskId:{}断开连接===============",taskId);
         });
     }

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

@@ -0,0 +1,141 @@
+package cd.casic.ci.process.engine.worker;
+
+import cd.casic.ci.common.pipeline.annotation.Plugin;
+import cd.casic.ci.process.engine.runContext.TaskRunContext;
+import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
+import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline;
+import cd.casic.ci.process.process.dataObject.target.TargetVersion;
+import cd.casic.ci.process.process.dataObject.task.PipTask;
+import cd.casic.ci.process.process.service.target.TargetVersionService;
+import cd.casic.framework.commons.exception.ServiceException;
+import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants;
+import com.jcraft.jsch.*;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.Map;
+
+@Plugin(taskType = "AFL")
+@Slf4j
+public class AFLWorker extends SshWorker{
+    @Resource
+    private TargetVersionService targetVersionService;
+    @Override
+    public void execute(TaskRunContext context) {
+        String filePath = "";
+        Map<String, Object> localVariables = context.getLocalVariables();
+        PipBaseElement taskContextDef = context.getContextDef();
+        if (taskContextDef instanceof PipTask pipTask){
+            // 查询并下载目标文件
+            String pipelineId = pipTask.getPipelineId();
+            //根据流水线id查询流水线信息
+            PipPipeline pipeline = (PipPipeline) getContextManager().getContext(pipelineId).getContextDef();
+            //根据目标id查询目标信息
+            if (StringUtils.isEmpty(pipeline.getTargetVersionId())){
+                throw new ServiceException(GlobalErrorCodeConstants.INTERNAL_SERVER_ERROR.getCode(),"目标文件不存在");
+            }
+            TargetVersion targetVersion = targetVersionService.getById(pipeline.getTargetVersionId());
+            filePath = targetVersion.getFilePath();
+            File file = new File(filePath);
+            if (!file.exists() || !file.canRead()) {
+                log.error("目标文件不存在或不可读");
+                localVariables.put("statusCode", "-1");
+                append(context,"目标文件不存在或不可读");
+                toBadEnding();
+            }
+            // 上传文件
+            // 执行shell脚本
+        }
+    }
+    public static void main(String[] args) {
+        String remoteHost = "your-remote-server-ip-or-hostname"; // 远程服务器IP或主机名
+        int remotePort = 22; // SFTP 默认端口通常是 22 (SSH 端口)
+        String username = "your-username"; // 远程服务器用户名
+        String password = "your-password"; // 远程服务器密码 或 SSH Key 路径
+
+        String localFilePath = "path/to/your/local/file.txt"; // 要上传的本地文件路径
+        String remoteDir = "/path/on/remote/server/"; // 远程服务器存放文件的目录
+        String remoteFileName = "uploaded_file.txt"; // 上传到远程服务器的文件名 (可以和本地不同)
+
+        Session session = null;
+        Channel channel = null;
+        ChannelSftp channelSftp = null;
+        FileInputStream fis = null;
+
+        try {
+            JSch jsch = new JSch();
+
+            // 如果使用 SSH Key 认证，可以添加身份文件：
+            // jsch.addIdentity("/path/to/your/private/key"); // 替换为你的私钥文件路径
+
+            session = jsch.getSession(username, remoteHost, remotePort);
+
+            // 如果使用密码认证：
+            session.setPassword(password);
+
+            // 设置连接不进行主机密钥检查 (生产环境不推荐，应该配置known_hosts)
+            java.util.Properties config = new java.util.Properties();
+            config.put("StrictHostKeyChecking", "no");
+            session.setConfig(config);
+
+            session.connect();
+            System.out.println("SFTP Session 连接成功.");
+
+            channel = session.openChannel("sftp");
+            channel.connect();
+            System.out.println("SFTP Channel 打开成功.");
+
+            channelSftp = (ChannelSftp) channel;
+
+            // 切换到远程目录 (如果目录不存在，可能需要先创建)
+            try {
+                channelSftp.cd(remoteDir);
+            } catch (SftpException e) {
+                System.err.println("远程目录不存在，尝试创建: " + remoteDir);
+                try {
+                    channelSftp.mkdir(remoteDir);
+                    channelSftp.cd(remoteDir);
+                } catch (SftpException e2) {
+                    System.err.println("创建远程目录失败: " + remoteDir);
+                    throw e2; // 抛出异常终止上传
+                }
+            }
+
+
+            // 上传文件
+            File localFile = new File(localFilePath);
+            fis = new FileInputStream(localFile);
+
+            channelSftp.put(fis, remoteFileName);
+            System.out.println("文件上传成功到: " + remoteDir + remoteFileName);
+
+        } catch (JSchException | SftpException | IOException e) {
+            System.err.println("SFTP 文件上传过程中发生异常: " + e.getMessage());
+            e.printStackTrace();
+        } finally {
+            // 关闭资源
+            if (fis != null) {
+                try {
+                    fis.close();
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }
+            }
+            if (channelSftp != null) {
+                channelSftp.disconnect();
+                System.out.println("SFTP Channel 已断开.");
+            }
+            if (channel != null) {
+                channel.disconnect();
+            }
+            if (session != null) {
+                session.disconnect();
+                System.out.println("SFTP Session 已断开.");
+            }
+        }
+    }
+}

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

@@ -0,0 +1,23 @@
+package cd.casic.ci.process.engine.worker;
+
+import cd.casic.ci.common.pipeline.annotation.Plugin;
+import cd.casic.ci.process.engine.runContext.TaskRunContext;
+import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
+import com.alibaba.fastjson.JSON;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ *  目标处理worker
+ *
+ * */
+@Slf4j
+@Plugin(taskType = "code")
+public class TargetHandleWorker extends BaseWorker{
+    @Override
+    public void execute(TaskRunContext context) {
+        PipBaseElement contextDef = context.getContextDef();
+        String id = contextDef.getId();
+        log.info("==============触发worker执行========");
+        log.info("==========运行context：{}===========", JSON.toJSONString(context));
+    }
+}

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

@@ -0,0 +1,12 @@
+package cd.casic.ci.process.engine.worker;
+
+import cd.casic.ci.common.pipeline.annotation.Plugin;
+import cd.casic.ci.process.engine.runContext.TaskRunContext;
+
+@Plugin(taskType = "TEST_CASE_GENERATION")
+public class TestCaseGenerationWorker extends SshWorker{
+    @Override
+    public void execute(TaskRunContext context) {
+
+    }
+}

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/converter/MachineConverter.java

@@ -0,0 +1,25 @@
+package cd.casic.ci.process.process.converter;
+
+import cd.casic.ci.common.pipeline.resp.machine.MachineInfoResp;
+import cd.casic.ci.common.pipeline.resp.pipeline.PipelineFindResp;
+import cd.casic.ci.process.process.dataObject.machine.MachineInfo;
+import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline;
+import org.mapstruct.Mapper;
+import org.mapstruct.factory.Mappers;
+
+import java.util.List;
+
+/**
+ * @author HopeLi
+ * @version v1.0
+ * @ClassName PipelineConverter
+ * @Date: 2025/5/13 14:39
+ * @Description:
+ */
+@Mapper(componentModel = "spring")
+public interface MachineConverter {
+    MachineConverter INSTANCE = Mappers.getMapper(MachineConverter.class);
+    MachineInfoResp toResp(MachineInfo pipPipeline);
+    List<MachineInfoResp> toRespList(List<MachineInfo> pipPipelines);
+
+}

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/dataObject/machine/MachineInfo.java

@@ -76,7 +76,7 @@ public class MachineInfo extends BaseDO {
      */
     private String username;
     /**
-     * 机器账号
+     * 系统类型
      */
     @TableField("os_system")
     private String osSystem;

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/dataObject/pipeline/PipPipeline.java

@@ -135,4 +135,5 @@ public class PipPipeline extends PipBaseElement {
      * 运行实例编号
      */
     private String instanceNum;
+
 }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/service/machine/MachineInfoService.java

@@ -1,6 +1,8 @@
 package cd.casic.ci.process.process.service.machine;
 
 
+import cd.casic.ci.common.pipeline.req.machine.MachineInfoReq;
+import cd.casic.ci.common.pipeline.resp.machine.MachineInfoResp;
 import cd.casic.ci.process.process.dataObject.machine.MachineInfo;
 import com.baomidou.mybatisplus.extension.service.IService;
 
@@ -13,5 +15,6 @@ import java.util.List;
  * @date 2022-09-27 10:25:29
  */
 public interface MachineInfoService extends IService<MachineInfo> {
+    List<MachineInfoResp> list(MachineInfoReq req);
 
 }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/service/machine/impl/MachineInfoServiceImpl.java

@@ -1,15 +1,21 @@
 package cd.casic.ci.process.process.service.machine.impl;
 
 
+import cd.casic.ci.common.pipeline.req.machine.MachineInfoReq;
+import cd.casic.ci.common.pipeline.resp.machine.MachineInfoResp;
+import cd.casic.ci.process.process.converter.MachineConverter;
 import cd.casic.ci.process.process.dal.machine.MachineInfoDao;
 import cd.casic.ci.process.process.dataObject.machine.MachineInfo;
 import cd.casic.ci.process.process.service.machine.MachineInfoService;
+import cn.hutool.core.util.ObjectUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 
 import org.springframework.stereotype.Service;
 
-
+import java.util.List;
 
 
 /**
@@ -20,6 +26,21 @@ import org.springframework.stereotype.Service;
 @Slf4j
 @Service
 public class MachineInfoServiceImpl extends ServiceImpl<MachineInfoDao, MachineInfo> implements MachineInfoService {
+    @Resource
+    private MachineInfoDao machineInfoDao;
 
-
+    @Override
+    public List<MachineInfoResp> list(MachineInfoReq req) {
+        LambdaQueryWrapper<MachineInfo> wrapper = new LambdaQueryWrapper<>();
+// 根据机器名称 查询
+        if (ObjectUtil.isNotEmpty(req.getMachineName())) {
+            wrapper.like(MachineInfo::getMachineName, req.getMachineName());
+        }
+        // 根据机器状态 1有效 2无效 查询
+        if (ObjectUtil.isNotEmpty(req.getMachineStatus())) {
+            wrapper.eq(MachineInfo::getMachineStatus, req.getMachineStatus());
+        }
+        List<MachineInfo> machineInfos = machineInfoDao.selectList(wrapper);
+        return MachineConverter.INSTANCE.toRespList(machineInfos);
+    }
 }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/service/pipeline/impl/PipelineServiceImpl.java

@@ -42,6 +42,7 @@ import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
 
+import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.*;
 
@@ -559,6 +560,11 @@ public class PipelineServiceImpl extends ServiceImpl<PipelineDao, PipPipeline> i
                     taskState.setEndTime(taskContext.getEndTime());
                     taskStateMap.put(taskId,taskState);
                     taskList.add(taskState);
+                    if (taskState.getEndTime()==null) {
+                        taskState.setDuration(between(taskState.getStartTime(),LocalDateTime.now()));
+                    } else {
+                        taskState.setDuration(between(taskState.getStartTime(),taskState.getEndTime()));
+                    }
                 }
             }
             return pipeline;
@@ -566,4 +572,26 @@ public class PipelineServiceImpl extends ServiceImpl<PipelineDao, PipPipeline> i
 
         return new TreeRunContextResp();
     }
+    private String between(LocalDateTime startTime,LocalDateTime endTime){
+        if (startTime==null||endTime==null) {
+            return "";
+        }
+        Duration between = Duration.between(startTime, endTime);
+        long days = between.toDays();
+        long hours = between.toHours()%24;
+        long minutes = between.toMinutes() % 60;
+        long seconds = between.toSeconds() % 60;
+        StringBuilder stringBuilder = new StringBuilder();
+        if (days>0) {
+            stringBuilder.append(days).append("天");
+        }
+        if (hours>0) {
+            stringBuilder.append(hours).append("小时");
+        }
+        if (minutes>0) {
+            stringBuilder.append(minutes).append("分");
+        }
+        stringBuilder.append(seconds).append("秒");
+        return stringBuilder.toString();
+    }
 }

modules/module-ci-worker/pom.xml

@@ -50,11 +50,11 @@
             <!--            <version>1.28</version>-->
         </dependency>
         <!-- https://mvnrepository.com/artifact/org.jenkins-ci/version-number -->
-        <dependency>
+<!--        <dependency>-->
-            <groupId>org.jenkins-ci</groupId>
+<!--            <groupId>org.jenkins-ci</groupId>-->
-            <artifactId>version-number</artifactId>
+<!--            <artifactId>version-number</artifactId>-->
-            <version>1.12</version>
+<!--            <version>1.12</version>-->
-        </dependency>
+<!--        </dependency>-->
 
         <!-- 外部依赖 -->
         <dependency>

ops-server/src/main/java/cd/casic/server/controller/MachineController.java

@@ -0,0 +1,22 @@
+package cd.casic.server.controller;
+
+import cd.casic.ci.process.process.dataObject.machine.MachineInfo;
+import cd.casic.ci.process.process.service.machine.MachineInfoService;
+import cd.casic.framework.commons.pojo.CommonResult;
+import jakarta.annotation.Resource;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+@RestController
+@RequestMapping("/machineInfo")
+public class MachineController {
+    @Resource
+    private MachineInfoService machineInfoService;
+    @GetMapping("/list")
+    public CommonResult<List<MachineInfo>> list(){
+        return CommonResult.success(machineInfoService.list());
+    }
+}

ops-server/src/main/resources/application-local.yaml

@@ -70,7 +70,7 @@ spring:
   # Redis 配置。Redisson 默认的配置足够使用，一般不需要进行调优
   data:
     redis:
-      host: 127.0.0.1 # 地址
+      host: 192.168.1.120 # 地址
       port: 6379 # 端口
       database: 0 # 数据库索引
 #    password: dev # 密码，建议生产环境开启