执行流程调试暂时完毕

.idea/compiler.xml

@@ -14,9 +14,9 @@
         <outputRelativeToContentRoot value="true" />
         <processorPath useClasspath="false">
           <entry name="$PROJECT_DIR$/../../apache-maven-3.8.6-bin/repository/org/springframework/boot/spring-boot-configuration-processor/3.3.4/spring-boot-configuration-processor-3.3.4.jar" />
-          <entry name="$PROJECT_DIR$/../../apache-maven-3.8.6-bin/repository/org/projectlombok/lombok/1.18.34/lombok-1.18.34.jar" />
           <entry name="$PROJECT_DIR$/../../apache-maven-3.8.6-bin/repository/org/mapstruct/mapstruct-processor/1.6.2/mapstruct-processor-1.6.2.jar" />
           <entry name="$PROJECT_DIR$/../../apache-maven-3.8.6-bin/repository/org/mapstruct/mapstruct/1.6.2/mapstruct-1.6.2.jar" />
+          <entry name="$PROJECT_DIR$/../../apache-maven-3.8.6-bin/repository/org/projectlombok/lombok/1.18.34/lombok-1.18.34.jar" />
         </processorPath>
         <module name="spring-boot-starter-protection" />
         <module name="module-ci-environment" />

framework/commons/src/main/java/cd/casic/framework/commons/dataobject/BaseDO.java

@@ -52,7 +52,7 @@ public abstract class BaseDO implements Serializable, TransPojo {
     /**
      * 是否删除
      */
-    @TableLogic
+//    @TableLogic
     private Boolean deleted;
 
 }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/dispatcher/impl/ParallelDispatcher.java

@@ -51,7 +51,7 @@ public class ParallelDispatcher implements BaseDispatcher{
             CountDownLatch latch = new CountDownLatch(stageList.size());
             for (PipStage secondStage : stageList) {
                 // 二阶段下所有task是串行所以不用关心线程安全相关信息
-                SecondStageRunContext context = new SecondStageRunContext(secondStage,pipelineRunContext,new ConcurrentHashMap<>());
+                SecondStageRunContext context = new SecondStageRunContext(secondStage,secondStage.getTaskValues().size(),pipelineRunContext,new ConcurrentHashMap<>());
                 runContextManager.contextRegister(context);
                 SerialDispatcher serialDispatcher = new SerialDispatcher(context,latch,runContextManager,redisMQTemplate);
                 // 给线程池进行执行

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

@@ -2,65 +2,46 @@ package cd.casic.ci.process.engine.enums;
 
 import lombok.Getter;
 
-import java.util.HashSet;
+import java.util.*;
-import java.util.Objects;
+
-import java.util.Set;
 @Getter
 public enum ContextStateEnum {
-    INIT(0,"初始化", new HashSet<>(){
+    INIT(0,"初始化"),
-        {
+    READY(1,"就绪"),
-            add(READY);
+    RUNNING(2,"运行"),
-            add(SUSPEND);
+    SUSPEND(3,"挂起"),
-            add(BAD_ENDING);
+    STOP(-1,"停止"),
-            add(STOP);
+    HAPPY_ENDING(4,"执行成功"),
-        }
+    BAD_ENDING(5,"执行失败")
-    }),
-    READY(1,"就绪", new HashSet<>(){
-        {
-            add(RUNNING);
-            add(SUSPEND);
-            add(BAD_ENDING);
-            add(STOP);
-        }
-    }),
-    RUNNING(2,"运行", new HashSet<>(){
-        {
-            add(SUSPEND);
-            add(STOP);
-            add(HAPPY_ENDING);
-            add(BAD_ENDING);
-        }
-    }),
-    SUSPEND(3,"挂起", new HashSet<>(){
-        {
-            add(INIT);
-            add(READY);
-            add(RUNNING);
-            add(BAD_ENDING);
-        }
-    }),
-    STOP(-1,"停止", new HashSet<>()),
-    HAPPY_ENDING(4,"执行成功", new HashSet<>()),
-    BAD_ENDING(5,"执行失败", new HashSet<>())
     ;
 
     private Integer code;
     private String msg;
-    /**
+    private static final Map<ContextStateEnum, Set<ContextStateEnum>> TRANSITIONS = new EnumMap<>(ContextStateEnum.class);
-     *  包含当前所有合法的下一个状态
-     * */
-    private Set<ContextStateEnum> nextStep;
 
-    ContextStateEnum(Integer code, String msg, Set<ContextStateEnum> nextStep) {
+    static {
+        TRANSITIONS.put(INIT, Set.of(READY, SUSPEND, BAD_ENDING, STOP));
+        TRANSITIONS.put(READY, Set.of(READY,RUNNING, SUSPEND, BAD_ENDING, STOP));
+        TRANSITIONS.put(RUNNING, Set.of(RUNNING,SUSPEND, HAPPY_ENDING, BAD_ENDING, STOP));
+        TRANSITIONS.put(SUSPEND, Set.of(SUSPEND,INIT, READY, BAD_ENDING, RUNNING,STOP));
+        //...初始化其他状态转移关系
+    }
+
+    ContextStateEnum(Integer code, String msg) {
         this.code = code;
         this.msg = msg;
-        this.nextStep = nextStep;
+
     }
     public static Boolean canGoto(ContextStateEnum from,ContextStateEnum to){
+        try {
             if (Objects.isNull(from) || Objects.isNull(to)) {
                 return false;
             }
-        return from.nextStep.contains(to);
+            return TRANSITIONS.get(from).contains(to);
+        } catch (Exception e){
+            System.out.println("");
+            return false;
+        }
     }
     public static ContextStateEnum getByCode(Integer code){
         for (ContextStateEnum value : values()) {

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/executor/impl/DefaultPipelineExecutor.java

@@ -52,8 +52,12 @@ public class DefaultPipelineExecutor implements PipelineExecutor {
         if (CollectionUtils.isEmpty(mainStage)) {
             throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"未找到有效阶段信息");
         }
+        Integer childCount= 0;
+        for (PipStage stage : mainStage) {
+            childCount+=stage.getStageList().size();
+        }
         // 如果要做 容灾就需要重新将数据库存的记录按顺序加载入
-        PipelineRunContext pipelineRunContext = new PipelineRunContext(null,pipeline,new ConcurrentHashMap<>(),new ConcurrentHashMap<>());
+        PipelineRunContext pipelineRunContext = new PipelineRunContext(pipeline,childCount,null,new ConcurrentHashMap<>(),new ConcurrentHashMap<>());
         ParallelDispatcher parallelDispatcher = new ParallelDispatcher(mainStage,pipelineRunContext,runContextManager,redisMQTemplate,serialExecutor);
         parallelExecutor.execute(parallelDispatcher);
         return pipelineRunContext;

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

@@ -76,6 +76,7 @@ public class DefaultWorkerManager extends AbstractRedisStreamMessageListener<Tas
             workerExecutor.execute(baseWorker);
         }catch (Exception e){
             // TODO 后期可以考虑专门整一个队列
+            log.error("=======================workerManager：",e);
         }
     }
 

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/runContext/BaseRunContext.java

@@ -4,7 +4,9 @@ import cd.casic.ci.process.engine.enums.ContextStateEnum;
 import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
 import cd.casic.framework.commons.exception.ServiceException;
 import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import lombok.Data;
+import org.springframework.data.annotation.Transient;
 
 import java.time.LocalDateTime;
 import java.util.List;
@@ -16,7 +18,10 @@ public abstract class BaseRunContext {
      *  当前上下文的定义
      * */
     private PipBaseElement contextDef;
+    @Transient
+    @JsonIgnore
     private BaseRunContext parentContext;
+    private Integer childCount;
     /**
      *  运行状态
      * */
@@ -42,8 +47,9 @@ public abstract class BaseRunContext {
     private Map<String,Object> localVariables;
     private Map<String,BaseRunContext> childContext;
 
-    public BaseRunContext(PipBaseElement contextDef,BaseRunContext parentContext, LocalDateTime startTime, String resourceId, String targetId, String targetType, Map<String, Object> globalVariables, Map<String, Object> localVariables, Map<String, BaseRunContext> childContext) {
+    public BaseRunContext(PipBaseElement contextDef,Integer childCount,BaseRunContext parentContext, LocalDateTime startTime, String resourceId, String targetId, String targetType, Map<String, Object> globalVariables, Map<String, Object> localVariables, Map<String, BaseRunContext> childContext) {
         this.contextDef = contextDef;
+        this.childCount = childCount;
         this.parentContext = parentContext;
         this.state = new AtomicInteger(ContextStateEnum.INIT.getCode());
         this.startTime = startTime;
@@ -88,6 +94,10 @@ public abstract class BaseRunContext {
      *  查找子类是否全部完成，如果子类全部完成则父类也完成
      * */
     public void checkChildEnd() throws ServiceException{
+        Map<String, BaseRunContext> childContext = getChildContext();
+        if (childContext.size()!=childCount) {
+            return;
+        }
         int result = ContextStateEnum.HAPPY_ENDING.getCode();
         for (Map.Entry<String, BaseRunContext> entry : childContext.entrySet()) {
             BaseRunContext child = entry.getValue();
@@ -100,18 +110,16 @@ public abstract class BaseRunContext {
         boolean end = false;
         if (ContextStateEnum.HAPPY_ENDING.getCode()==result) {
             if (ContextStateEnum.canGoto(ContextStateEnum.getByCode(state.get()),ContextStateEnum.HAPPY_ENDING)) {
-                this.state.compareAndExchange(state.get(),ContextStateEnum.HAPPY_ENDING.getCode());
+                this.changeContextState(ContextStateEnum.HAPPY_ENDING);
                 end = true;
             }
         } else {
             if (ContextStateEnum.canGoto(ContextStateEnum.getByCode(state.get()),ContextStateEnum.BAD_ENDING)) {
-                this.state.compareAndExchange(state.get(),ContextStateEnum.BAD_ENDING.getCode());
+                this.changeContextState(ContextStateEnum.BAD_ENDING);
                 end = true;
             }
         }
-        if (end) {
+        // 当前执行结束看看是否要执行后置处理
-            this.changeContextState(ContextStateEnum.getByCode(result));
-        }
     }
     /**
      *  查找子类是否全部就绪，如果子类全部完成则父类也就绪
@@ -127,16 +135,11 @@ public abstract class BaseRunContext {
             }
             result&=state;
         }
-        boolean ready = false;
+        if (ContextStateEnum.READY.getCode().equals(result)) {
-        if (ContextStateEnum.READY.getCode()==result) {
             if (ContextStateEnum.canGoto(ContextStateEnum.getByCode(state.get()),ContextStateEnum.READY)) {
-                this.state.compareAndExchange(state.get(),ContextStateEnum.READY.getCode());
+                this.changeContextState(ContextStateEnum.READY);
-                ready = true;
             }
         }
-        if (!ready) {
-            throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"状态有误");
-        }
     }
     /**
      *  查找子类是否存在开始运行的，如果有则父状态变成running
@@ -147,14 +150,14 @@ public abstract class BaseRunContext {
         for (Map.Entry<String, BaseRunContext> entry : childContext.entrySet()) {
             BaseRunContext child = entry.getValue();
             int state = child.getState().get();
-            if (ContextStateEnum.READY.getCode().equals(state)) {
+            if (ContextStateEnum.RUNNING.getCode().equals(state)) {
                 runningFlag = true;
                 break;
             }
         }
         if (runningFlag) {
             if (ContextStateEnum.canGoto(ContextStateEnum.getByCode(state.get()),ContextStateEnum.RUNNING)) {
-                this.state.compareAndExchange(state.get(),ContextStateEnum.RUNNING.getCode());
+                this.changeContextState(ContextStateEnum.RUNNING);
             } else{
                 throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"状态有误");
             }

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/runContext/PipelineRunContext.java

@@ -1,5 +1,6 @@
 package cd.casic.ci.process.engine.runContext;
 
+import cd.casic.ci.process.engine.enums.ContextStateEnum;
 import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
 import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline;
 import cd.casic.framework.commons.exception.ServiceException;
@@ -11,16 +12,17 @@ import java.util.concurrent.ConcurrentHashMap;
 
 public class PipelineRunContext extends BaseRunContext{
 
-    public PipelineRunContext(PipPipeline pipeline) {
+    public PipelineRunContext(PipPipeline pipeline,Integer childCount) {
-        this(null,pipeline,new ConcurrentHashMap<>(),new ConcurrentHashMap<>());
+        this(pipeline,childCount,null,new ConcurrentHashMap<>(),new ConcurrentHashMap<>());
     }
 
-    public PipelineRunContext(BaseRunContext parentContext, PipPipeline pipeline, Map<String, Object> globalVariables, Map<String, Object> localVariables) {
+    public PipelineRunContext(PipPipeline pipeline,Integer childCount,BaseRunContext parentContext, Map<String, Object> globalVariables, Map<String, Object> localVariables) {
-        this(pipeline,parentContext,LocalDateTime.now(),pipeline.getResourceId(),pipeline.getTargetId(),pipeline.getTargetType(),globalVariables,localVariables,new ConcurrentHashMap<>());
+        this(pipeline,childCount,parentContext,LocalDateTime.now(),pipeline.getResourceId(),pipeline.getTargetId(),pipeline.getTargetType(),globalVariables,localVariables,new ConcurrentHashMap<>());
     }
 
-    private PipelineRunContext(PipBaseElement contextDef, BaseRunContext parentContext, LocalDateTime startTime, String resourceId, String targetId, String targetType, Map<String, Object> globalVariables, Map<String, Object> localVariables, Map<String, BaseRunContext> childContext) {
+    private PipelineRunContext(PipBaseElement contextDef,Integer childCount, BaseRunContext parentContext, LocalDateTime startTime, String resourceId, String targetId, String targetType, Map<String, Object> globalVariables, Map<String, Object> localVariables, Map<String, BaseRunContext> childContext) {
         super(  contextDef
+                ,childCount
                 ,parentContext
                 ,startTime
                 , resourceId

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/runContext/SecondStageRunContext.java

@@ -11,8 +11,8 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class SecondStageRunContext extends BaseRunContext{
-    public SecondStageRunContext(PipStage contextDef, PipelineRunContext parentContext, Map<String, Object> localVariables) {
+    public SecondStageRunContext(PipStage contextDef,Integer childCount, PipelineRunContext parentContext, Map<String, Object> localVariables) {
-        super(contextDef, parentContext, LocalDateTime.now(), parentContext.getResourceId(), parentContext.getTargetId(), parentContext.getTargetType(), parentContext.getGlobalVariables(), localVariables, new ConcurrentHashMap<>());
+        super(contextDef,childCount, parentContext, LocalDateTime.now(), parentContext.getResourceId(), parentContext.getTargetId(), parentContext.getTargetType(), parentContext.getGlobalVariables(), localVariables, new ConcurrentHashMap<>());
     }
 
     @Override

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/runContext/TaskRunContext.java

@@ -12,10 +12,10 @@ import java.util.Map;
 
 public class TaskRunContext extends BaseRunContext{
     public TaskRunContext(PipTask contextDef, SecondStageRunContext parentContext,Map<String,Object> localVariable) {
-        super(contextDef, parentContext, LocalDateTime.now(), parentContext.getResourceId(), parentContext.getTargetId(), parentContext.getTargetType(), parentContext.getGlobalVariables(),localVariable, new HashMap<>());
+        super(contextDef,0, parentContext, LocalDateTime.now(), parentContext.getResourceId(), parentContext.getTargetId(), parentContext.getTargetType(), parentContext.getGlobalVariables(),localVariable, new HashMap<>());
     }
     private TaskRunContext(PipBaseElement contextDef, BaseRunContext parentContext, LocalDateTime startTime, String resourceId, String targetId, String targetType, Map<String, Object> globalVariables, Map<String, Object> localVariables, Map<String, BaseRunContext> childContext) {
-        super(contextDef, parentContext, startTime, resourceId, targetId, targetType, globalVariables, localVariables, childContext);
+        super(contextDef,0, parentContext, startTime, resourceId, targetId, targetType, globalVariables, localVariables, childContext);
     }
 
     /**

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

@@ -17,7 +17,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 
 @Data
-
 public abstract class BaseWorker implements Runnable{
     // 一些属性
     @Resource
@@ -36,10 +35,12 @@ public abstract class BaseWorker implements Runnable{
         BaseRunContext context = contextManager.getContext(contextKey);
         if (context instanceof TaskRunContext taskRunContext){
             try {
+                taskRunContext.changeContextState(ContextStateEnum.READY);
                 taskRunContext.changeContextState(ContextStateEnum.RUNNING);
                 execute(taskRunContext);
             } catch (Exception e) {
                 taskRunContext.changeContextState(ContextStateEnum.BAD_ENDING);
+                return;
             }
             // TODO 执行结束修改context的state,并且通知父类
             taskRunContext.changeContextState(ContextStateEnum.HAPPY_ENDING);

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

@@ -0,0 +1,22 @@
+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;
+
+
+@Slf4j
+@Plugin(taskType = "Github")
+public class TestGitWorker 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/TestWorker.java

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

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

@@ -6,6 +6,8 @@ import cd.casic.ci.common.pipeline.req.pipeline.PipelineReq;
 import cd.casic.ci.common.pipeline.req.pipeline.PipelineUpdateReq;
 import cd.casic.ci.common.pipeline.resp.pipeline.PipelineFindResp;
 import cd.casic.ci.common.pipeline.utils.PageResult;
+import cd.casic.ci.process.engine.executor.PipelineExecutor;
+import cd.casic.ci.process.engine.runContext.PipelineRunContext;
 import cd.casic.ci.process.process.service.pipeline.PipelineService;
 import cd.casic.framework.commons.pojo.CommonResult;
 import jakarta.annotation.Resource;
@@ -32,6 +34,8 @@ public class PipelineController {
 
     @Resource
     private PipelineService pipelineService;
+    @Resource
+    private PipelineExecutor pipelineExecutor;
 
     @PermitAll
     @PostMapping(path="/createPipeline")
@@ -96,4 +100,9 @@ public class PipelineController {
 
         return CommonResult.success();
     }
+    @PostMapping("/executePipeline")
+    public CommonResult<PipelineRunContext> executePipeline(String pipelineId){
+        PipelineRunContext execute = pipelineExecutor.execute(pipelineId);
+        return CommonResult.success(execute);
+    }
 }

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

@@ -41,11 +41,11 @@ public class RedisMqTest {
         pipTask.setStageId("testStage");
         PipPipeline pipeline = new PipPipeline();
         pipeline.setId("testPipeline");
-        PipelineRunContext pipelineRunContext = new PipelineRunContext(pipeline);
+        PipelineRunContext pipelineRunContext = new PipelineRunContext(pipeline,1);
         PipStage stage = new PipStage();
         stage.setId("testStage");
         stage.setParentId("testPipeline");
-        SecondStageRunContext secondStageRunContext = new SecondStageRunContext(stage,pipelineRunContext,new ConcurrentHashMap<>());
+        SecondStageRunContext secondStageRunContext = new SecondStageRunContext(stage,1,pipelineRunContext,new ConcurrentHashMap<>());
         TaskRunContext taskRunContext = new TaskRunContext(pipTask,secondStageRunContext,new HashMap<>());
         contextManager.contextRegister(pipelineRunContext);
         contextManager.contextRegister(secondStageRunContext);