mianbin/ops-pro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: mianbin:bfeacb12a2919eea9c7a76b731baae698a4fa4bb
Branches Tags
mianbin:master
mianbin:jenkins-engin
mianbin:temp
..
compare: mianbin:3eff2984ebc47c9c01a9b05d7ce301c4642e6da1
Branches Tags
mianbin:master
mianbin:jenkins-engin
mianbin:temp

No commits in common. "bfeacb12a2919eea9c7a76b731baae698a4fa4bb" and "3eff2984ebc47c9c01a9b05d7ce301c4642e6da1" have entirely different histories.
bfeacb12a2 ... 3eff2984eb

33 changed files with 92 additions and 477 deletions
Show Stats Expand all files Collapse all files

23
framework/commons/pom.xml
View File

@ -194,27 +194,6 @@
</dependency> </dependency>
</dependencies> </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> </project>

14
framework/commons/src/main/java/cd/casic/framework/commons/util/util/WebFrameworkUtils.java
View File

@ -20,7 +20,6 @@ import org.springframework.web.context.request.ServletRequestAttributes;
public class WebFrameworkUtils { public class WebFrameworkUtils {
private static final String REQUEST_ATTRIBUTE_LOGIN_USER_ID = "login_user_id"; private static final String REQUEST_ATTRIBUTE_LOGIN_USER_ID = "login_user_id";
private static final String REQUEST_ATTRIBUTE_LOGIN_USER_NAME = "login_user_name";
private static final String REQUEST_ATTRIBUTE_LOGIN_USER_TYPE = "login_user_type"; private static final String REQUEST_ATTRIBUTE_LOGIN_USER_TYPE = "login_user_type";
private static final String REQUEST_ATTRIBUTE_COMMON_RESULT = "common_result"; private static final String REQUEST_ATTRIBUTE_COMMON_RESULT = "common_result";
@ -39,6 +38,7 @@ public class WebFrameworkUtils {
public WebFrameworkUtils(WebProperties webProperties) { public WebFrameworkUtils(WebProperties webProperties) {
WebFrameworkUtils.properties = webProperties; WebFrameworkUtils.properties = webProperties;
} }
/** /**
* 获得租户编号，从 header 中 * 获得租户编号，从 header 中
* 考虑到其它 framework 组件也会使用到租户编号，所以不得不放在 WebFrameworkUtils 统一提供 * 考虑到其它 framework 组件也会使用到租户编号，所以不得不放在 WebFrameworkUtils 统一提供
@ -54,18 +54,6 @@ public class WebFrameworkUtils {
public static void setLoginUserId(ServletRequest request, Long userId) { public static void setLoginUserId(ServletRequest request, Long userId) {
request.setAttribute(REQUEST_ATTRIBUTE_LOGIN_USER_ID, userId); request.setAttribute(REQUEST_ATTRIBUTE_LOGIN_USER_ID, userId);
} }
public static void setLoginNickName(ServletRequest request, String userName) {
request.setAttribute(REQUEST_ATTRIBUTE_LOGIN_USER_ID, userName);
}
public static String getLoginNickName() {
return getLoginNickName(getRequest());
}
public static String getLoginNickName(HttpServletRequest request) {
if (request == null) {
return null;
}
return (String) request.getAttribute(REQUEST_ATTRIBUTE_LOGIN_USER_NAME);
}
/** /**
* 设置用户类型 * 设置用户类型

2
framework/spring-boot-starter-biz-data-permission/src/main/java/cd/casic/framework/datapermission/convert/TenantConvert.java
View File

@ -10,7 +10,7 @@ import org.mapstruct.factory.Mappers;
* *
* @author mianbin modified from yudao * @author mianbin modified from yudao
*/ */
@Mapper(componentModel = "spring") @Mapper
public interface TenantConvert { public interface TenantConvert {
TenantConvert INSTANCE = Mappers.getMapper(TenantConvert.class); TenantConvert INSTANCE = Mappers.getMapper(TenantConvert.class);

1
framework/spring-boot-starter-security/src/main/java/cd/casic/framework/security/core/util/SecurityFrameworkUtils.java
View File

@ -127,7 +127,6 @@ public class SecurityFrameworkUtils {
// 原因是，Spring Security 的 Filter 在 ApiAccessLogFilter 后面，在它记录访问日志时，线上上下文已经没有用户编号等信息 // 原因是，Spring Security 的 Filter 在 ApiAccessLogFilter 后面，在它记录访问日志时，线上上下文已经没有用户编号等信息
WebFrameworkUtils.setLoginUserId(request, loginUser.getId()); WebFrameworkUtils.setLoginUserId(request, loginUser.getId());
WebFrameworkUtils.setLoginUserType(request, loginUser.getUserType()); WebFrameworkUtils.setLoginUserType(request, loginUser.getUserType());
WebFrameworkUtils.setLoginNickName(request,loginUser.getInfo().get("nickName"));
} }
private static Authentication buildAuthentication(LoginUser loginUser, HttpServletRequest request) { private static Authentication buildAuthentication(LoginUser loginUser, HttpServletRequest request) {

32
modules/module-ci-process-api/src/main/java/cd/casic/ci/api/PipHistoryController.java
View File

@ -1,32 +0,0 @@
package cd.casic.ci.api;
import cd.casic.ci.process.process.dataObject.history.PipPipelineHisInstance;
import cd.casic.ci.process.process.dataObject.log.PipTaskLog;
import cd.casic.ci.process.process.service.history.PipelineHistoryService;
import cd.casic.ci.process.process.service.taskLog.TaskLogService;
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("/history")
public class PipHistoryController {
@Resource
TaskLogService taskLogService;
@Resource
PipelineHistoryService pipelineHistoryService;
@GetMapping("/getLogById")
public CommonResult<PipTaskLog> getLogById(String id){
PipTaskLog byId = taskLogService.getById(id);
return CommonResult.success(byId);
}
@GetMapping("/list")
public CommonResult<List<PipPipelineHisInstance>> list(String pipelineId){
List<PipPipelineHisInstance> list = pipelineHistoryService.list();
return CommonResult.success(list);
}
}

12
modules/module-ci-process-api/src/main/java/cd/casic/ci/api/PipelineController.java
View File

@ -9,19 +9,13 @@ import cd.casic.ci.process.dal.resp.context.TreeRunContextResp;
import cd.casic.ci.process.dal.resp.pipeline.PipelineFindResp; import cd.casic.ci.process.dal.resp.pipeline.PipelineFindResp;
import cd.casic.ci.process.engine.executor.PipelineExecutor; import cd.casic.ci.process.engine.executor.PipelineExecutor;
import cd.casic.ci.process.engine.runContext.PipelineRunContext; import cd.casic.ci.process.engine.runContext.PipelineRunContext;
import cd.casic.ci.process.enums.PiplineTriggerModeEnum;
import cd.casic.ci.process.process.service.pipeline.PipelineService; import cd.casic.ci.process.process.service.pipeline.PipelineService;
import cd.casic.framework.commons.pojo.CommonResult; import cd.casic.framework.commons.pojo.CommonResult;
import cd.casic.framework.commons.pojo.PageResult; import cd.casic.framework.commons.pojo.PageResult;
import cd.casic.framework.commons.util.util.WebFrameworkUtils;
import cd.casic.framework.security.core.LoginUser;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
import jakarta.annotation.security.PermitAll; import jakarta.annotation.security.PermitAll;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.validation.Valid; import jakarta.validation.Valid;
import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.NotNull;
import org.springframework.boot.actuate.autoconfigure.metrics.MetricsProperties;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
import java.util.List; import java.util.List;
@ -105,11 +99,9 @@ public class PipelineController {
return CommonResult.success(); return CommonResult.success();
} }
@Resource
private HttpServletRequest request;
@PostMapping("/executePipeline/{pipelineId}") @PostMapping("/executePipeline/{pipelineId}")
public CommonResult<PipelineRunContext> executePipeline(@PathVariable("pipelineId") String pipelineId,@RequestParam(required = false,defaultValue = "false") Boolean fromError){ public CommonResult<PipelineRunContext> executePipeline(@PathVariable String pipelineId){
PipelineRunContext execute = pipelineExecutor.execute(pipelineId, PiplineTriggerModeEnum.HAND,fromError); PipelineRunContext execute = pipelineExecutor.execute(pipelineId);
return CommonResult.success(execute); return CommonResult.success(execute);
} }
@PostMapping("/getStageRunState/{pipelineId}") @PostMapping("/getStageRunState/{pipelineId}")

2
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/constant/EngineRuntimeConstant.java
View File

@ -1,5 +1,5 @@
package cd.casic.ci.process.engine.constant; package cd.casic.ci.process.engine.constant;
public class EngineRuntimeConstant { public class EngineRuntimeConstant {
public static final String HIS_LOG_KEY = "logId"; public static final String LOG_KEY = "logContent";
} }

2
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/constant/PipelineBehaviorConstant.java
View File

@ -10,5 +10,5 @@ public class PipelineBehaviorConstant {
/** /**
* 流水线是否从上次错误处执行 * 流水线是否从上次错误处执行
* */ * */
public static final String PIPELINE_EXECUTE_FROM_ERROR="fromError"; public static final String PIPELINE_EXECUTE_FROM_ERROR="";
} }

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

@ -1,22 +1,17 @@
package cd.casic.ci.process.engine.dispatcher.impl; package cd.casic.ci.process.engine.dispatcher.impl;
import cd.casic.ci.process.engine.constant.PipelineBehaviorConstant;
import cd.casic.ci.process.engine.dispatcher.BaseDispatcher; import cd.casic.ci.process.engine.dispatcher.BaseDispatcher;
import cd.casic.ci.process.engine.enums.ContextStateEnum; import cd.casic.ci.process.engine.enums.ContextStateEnum;
import cd.casic.ci.process.engine.manager.RunContextManager; import cd.casic.ci.process.engine.manager.RunContextManager;
import cd.casic.ci.process.engine.runContext.BaseRunContext;
import cd.casic.ci.process.engine.runContext.PipelineRunContext; import cd.casic.ci.process.engine.runContext.PipelineRunContext;
import cd.casic.ci.process.engine.runContext.SecondStageRunContext; import cd.casic.ci.process.engine.runContext.SecondStageRunContext;
import cd.casic.ci.process.engine.runContext.TaskRunContext;
import cd.casic.ci.process.process.dataObject.stage.PipStage; import cd.casic.ci.process.process.dataObject.stage.PipStage;
import cd.casic.framework.mq.redis.core.RedisMQTemplate; import cd.casic.framework.mq.redis.core.RedisMQTemplate;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
@Slf4j @Slf4j
@ -49,24 +44,9 @@ public class ParallelDispatcher implements BaseDispatcher{
} }
CountDownLatch latch = new CountDownLatch(stageList.size()); CountDownLatch latch = new CountDownLatch(stageList.size());
for (PipStage secondStage : stageList) { for (PipStage secondStage : stageList) {
Map<String, Object> globalVariables = pipelineRunContext.getGlobalVariables(); // 二阶段下所有task是串行所以不用关心线程安全相关信息
Object fromError = globalVariables.get(PipelineBehaviorConstant.PIPELINE_EXECUTE_FROM_ERROR); SecondStageRunContext context = new SecondStageRunContext(secondStage,secondStage.getTaskValues().size(),pipelineRunContext,new ConcurrentHashMap<>());
SecondStageRunContext context = null; runContextManager.contextRegister(context);
if (!Boolean.TRUE.equals(fromError)) {
// 注册taskContext，且发送消息至消息队列给work执行, 如果需要则传入参数
// 二阶段下所有task是串行所以不用关心线程安全相关信息
context = new SecondStageRunContext(secondStage,secondStage.getTaskValues().size(),pipelineRunContext,new ConcurrentHashMap<>());
runContextManager.contextRegister(context);
} else {
BaseRunContext baseRunContext = runContextManager.getContext(secondStage.getId());
if (baseRunContext!=null) {
context = (SecondStageRunContext) baseRunContext;
} else {
// 防止运行停止以后添加节点 ,删除节点的同时也要删除上下文
context = new SecondStageRunContext(secondStage,secondStage.getTaskValues().size(),pipelineRunContext,new ConcurrentHashMap<>());
runContextManager.contextRegister(context);
}
}
SerialDispatcher serialDispatcher = new SerialDispatcher(context,latch,runContextManager,redisMQTemplate); SerialDispatcher serialDispatcher = new SerialDispatcher(context,latch,runContextManager,redisMQTemplate);
// 给线程池进行执行 // 给线程池进行执行
taskExecutor.execute(serialDispatcher); taskExecutor.execute(serialDispatcher);

23
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/dispatcher/impl/SerialDispatcher.java
View File

@ -1,11 +1,9 @@
package cd.casic.ci.process.engine.dispatcher.impl; package cd.casic.ci.process.engine.dispatcher.impl;
import cd.casic.ci.process.engine.constant.PipelineBehaviorConstant;
import cd.casic.ci.process.engine.dispatcher.BaseDispatcher; import cd.casic.ci.process.engine.dispatcher.BaseDispatcher;
import cd.casic.ci.process.engine.enums.ContextStateEnum; import cd.casic.ci.process.engine.enums.ContextStateEnum;
import cd.casic.ci.process.engine.manager.RunContextManager; import cd.casic.ci.process.engine.manager.RunContextManager;
import cd.casic.ci.process.engine.message.TaskRunMessage; import cd.casic.ci.process.engine.message.TaskRunMessage;
import cd.casic.ci.process.engine.runContext.BaseRunContext;
import cd.casic.ci.process.engine.runContext.SecondStageRunContext; import cd.casic.ci.process.engine.runContext.SecondStageRunContext;
import cd.casic.ci.process.engine.runContext.TaskRunContext; import cd.casic.ci.process.engine.runContext.TaskRunContext;
import cd.casic.ci.process.process.dataObject.base.PipBaseElement; import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
@ -17,7 +15,6 @@ import lombok.extern.slf4j.Slf4j;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
@Slf4j @Slf4j
@ -44,23 +41,9 @@ public class SerialDispatcher implements BaseDispatcher {
@Override @Override
public void dispatch() throws InterruptedException { public void dispatch() throws InterruptedException {
for (PipTask pipTask : taskList) { for (PipTask pipTask : taskList) {
TaskRunContext taskRunContext = null; // 注册taskContext，且发送消息至消息队列给work执行, 如果需要则传入参数
Map<String, Object> globalVariables = stageRunContext.getGlobalVariables(); TaskRunContext taskRunContext = new TaskRunContext(pipTask,stageRunContext,new HashMap<>());
Object fromError = globalVariables.get(PipelineBehaviorConstant.PIPELINE_EXECUTE_FROM_ERROR); contextManager.contextRegister(taskRunContext);
if (!Boolean.TRUE.equals(fromError)) {
// 注册taskContext，且发送消息至消息队列给work执行, 如果需要则传入参数
taskRunContext = new TaskRunContext(pipTask,stageRunContext,new HashMap<>());
contextManager.contextRegister(taskRunContext);
} else {
BaseRunContext context = contextManager.getContext(pipTask.getId());
if (context != null) {
taskRunContext= (TaskRunContext) context;
} else {
// 防止运行停止以后添加节点,删除节点的同时也要删除上下文
taskRunContext = new TaskRunContext(pipTask,stageRunContext,new HashMap<>());
contextManager.contextRegister(taskRunContext);
}
}
taskRunContext.changeContextState(ContextStateEnum.READY); taskRunContext.changeContextState(ContextStateEnum.READY);
TaskRunMessage taskRunMessage = new TaskRunMessage(pipTask); TaskRunMessage taskRunMessage = new TaskRunMessage(pipTask);
redisMQTemplate.send(taskRunMessage); redisMQTemplate.send(taskRunMessage);

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

@ -27,7 +27,6 @@ public enum ContextStateEnum {
TRANSITIONS.put(SUSPEND, Set.of(SUSPEND,INIT, READY, BAD_ENDING, RUNNING,STOP)); TRANSITIONS.put(SUSPEND, Set.of(SUSPEND,INIT, READY, BAD_ENDING, RUNNING,STOP));
//...初始化其他状态转移关系 //...初始化其他状态转移关系
TRANSITIONS.put(SKIP_TO,Collections.emptySet()); TRANSITIONS.put(SKIP_TO,Collections.emptySet());
// TRANSITIONS.put(HAPPY_ENDING,Set.of(RUNNING));
} }
ContextStateEnum(Integer code, String msg) { ContextStateEnum(Integer code, String msg) {

3
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/executor/PipelineExecutor.java
View File

@ -1,8 +1,7 @@
package cd.casic.ci.process.engine.executor; package cd.casic.ci.process.engine.executor;
import cd.casic.ci.process.engine.runContext.PipelineRunContext; import cd.casic.ci.process.engine.runContext.PipelineRunContext;
import cd.casic.ci.process.enums.PiplineTriggerModeEnum;
public interface PipelineExecutor { public interface PipelineExecutor {
PipelineRunContext execute(String pipelineId, PiplineTriggerModeEnum triggerModeEnum,Boolean formError); PipelineRunContext execute(String pipelineId);
} }

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

@ -1,13 +1,9 @@
package cd.casic.ci.process.engine.executor.impl; package cd.casic.ci.process.engine.executor.impl;
import cd.casic.ci.process.engine.constant.PipelineBehaviorConstant;
import cd.casic.ci.process.engine.dispatcher.impl.ParallelDispatcher; import cd.casic.ci.process.engine.dispatcher.impl.ParallelDispatcher;
import cd.casic.ci.process.engine.executor.PipelineExecutor; import cd.casic.ci.process.engine.executor.PipelineExecutor;
import cd.casic.ci.process.engine.manager.LoggerManager;
import cd.casic.ci.process.engine.manager.RunContextManager; import cd.casic.ci.process.engine.manager.RunContextManager;
import cd.casic.ci.process.engine.runContext.BaseRunContext;
import cd.casic.ci.process.engine.runContext.PipelineRunContext; import cd.casic.ci.process.engine.runContext.PipelineRunContext;
import cd.casic.ci.process.enums.PiplineTriggerModeEnum;
import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline; 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.stage.PipStage;
import cd.casic.ci.process.process.service.pipeline.PipelineService; import cd.casic.ci.process.process.service.pipeline.PipelineService;
@ -37,10 +33,8 @@ public class DefaultPipelineExecutor implements PipelineExecutor {
private ThreadPoolTaskExecutor serialExecutor; private ThreadPoolTaskExecutor serialExecutor;
@Resource @Resource
private RedisMQTemplate redisMQTemplate; private RedisMQTemplate redisMQTemplate;
@Resource
private LoggerManager loggerManager;
@Override @Override
public PipelineRunContext execute(String pipelineId, PiplineTriggerModeEnum triggerModeEnum,Boolean formError) { public PipelineRunContext execute(String pipelineId) {
PipPipeline pipeline = pipelineService.getById(pipelineId); PipPipeline pipeline = pipelineService.getById(pipelineId);
if (pipeline==null) { if (pipeline==null) {
return null; return null;
@ -59,24 +53,8 @@ public class DefaultPipelineExecutor implements PipelineExecutor {
for (PipStage stage : mainStage) { for (PipStage stage : mainStage) {
childCount+=stage.getStageList().size(); childCount+=stage.getStageList().size();
} }
PipelineRunContext pipelineRunContext = null; // 如果要做 容灾就需要重新将数据库存的记录按顺序加载入
if (!formError) { PipelineRunContext pipelineRunContext = new PipelineRunContext(pipeline,childCount,null,new ConcurrentHashMap<>(),new ConcurrentHashMap<>());
// 从头执行清理缓存日志
loggerManager.clear(pipelineId);
ConcurrentHashMap<String, Object> globalVariable = null;
globalVariable = new ConcurrentHashMap<>();
// 如果要做 容灾就需要重新将数据库存的记录按顺序加载入
pipelineRunContext=new PipelineRunContext(pipeline,childCount,null,globalVariable,new ConcurrentHashMap<>());
} else {
try {
pipelineRunContext = (PipelineRunContext)runContextManager.getContext(pipelineId);
} catch (Exception e) {
// 以后可以改成从 数据库重载
throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"未找到上次的记录");
}
}
pipelineRunContext.setTriggerMode(triggerModeEnum);
pipelineRunContext.getGlobalVariables().put(PipelineBehaviorConstant.PIPELINE_EXECUTE_FROM_ERROR,formError);
runContextManager.contextRegister(pipelineRunContext); runContextManager.contextRegister(pipelineRunContext);
ParallelDispatcher parallelDispatcher = new ParallelDispatcher(mainStage,pipelineRunContext,runContextManager,redisMQTemplate,serialExecutor); ParallelDispatcher parallelDispatcher = new ParallelDispatcher(mainStage,pipelineRunContext,runContextManager,redisMQTemplate,serialExecutor);
parallelExecutor.execute(parallelDispatcher); parallelExecutor.execute(parallelDispatcher);

4
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/manager/LoggerManager.java
View File

@ -1,6 +1,5 @@
package cd.casic.ci.process.engine.manager; package cd.casic.ci.process.engine.manager;
import cd.casic.ci.process.engine.runContext.TaskRunContext;
import jakarta.servlet.http.HttpServletRequest; import jakarta.servlet.http.HttpServletRequest;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@ -14,6 +13,5 @@ public interface LoggerManager {
SseEmitter subscribe(String taskId,HttpServletRequest request); SseEmitter subscribe(String taskId,HttpServletRequest request);
void append(String taskId,String logContent); void append(String taskId,String logContent);
String getLogContent(String taskId); String getLogContent(String taskId);
void flushMemory(List<TaskRunContext> taskContextList); void flushMemory(List<String> taskIdList);
void clear(String pipelineId);
} }

2
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/manager/RunContextManager.java
View File

@ -28,6 +28,4 @@ public interface RunContextManager {
* */ * */
void contextRegister(BaseRunContext context); void contextRegister(BaseRunContext context);
BaseRunContext getContext(String key); BaseRunContext getContext(String key);
public void toHistory(String pipelineId);
} }

84
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/manager/impl/DefaultRunContextManager.java
View File

@ -1,6 +1,5 @@
package cd.casic.ci.process.engine.manager.impl; package cd.casic.ci.process.engine.manager.impl;
import cd.casic.ci.process.dal.req.pipeline.PipelineQueryReq;
import cd.casic.ci.process.engine.enums.ContextStateEnum; import cd.casic.ci.process.engine.enums.ContextStateEnum;
import cd.casic.ci.process.engine.manager.LoggerManager; import cd.casic.ci.process.engine.manager.LoggerManager;
import cd.casic.ci.process.engine.manager.RunContextManager; import cd.casic.ci.process.engine.manager.RunContextManager;
@ -8,47 +7,27 @@ import cd.casic.ci.process.engine.runContext.BaseRunContext;
import cd.casic.ci.process.engine.runContext.PipelineRunContext; import cd.casic.ci.process.engine.runContext.PipelineRunContext;
import cd.casic.ci.process.engine.runContext.SecondStageRunContext; import cd.casic.ci.process.engine.runContext.SecondStageRunContext;
import cd.casic.ci.process.engine.runContext.TaskRunContext; import cd.casic.ci.process.engine.runContext.TaskRunContext;
import cd.casic.ci.process.enums.PiplineTriggerModeEnum;
import cd.casic.ci.process.process.dal.history.PipPipelineHisInstanceDao;
import cd.casic.ci.process.process.dal.pipeline.PipTaskDao; import cd.casic.ci.process.process.dal.pipeline.PipTaskDao;
import cd.casic.ci.process.process.dataObject.base.PipBaseElement; import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
import cd.casic.ci.process.process.dataObject.history.PipPipelineHisInstance;
import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline;
import cd.casic.ci.process.process.dataObject.task.PipTask; import cd.casic.ci.process.process.dataObject.task.PipTask;
import cd.casic.ci.process.process.service.pipeline.PipelineService;
import cd.casic.framework.commons.exception.ServiceException; import cd.casic.framework.commons.exception.ServiceException;
import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants; import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants;
import cd.casic.framework.commons.util.util.WebFrameworkUtils;
import cd.casic.framework.security.core.LoginUser;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import java.util.Collection; import java.util.Collection;
import java.util.LinkedList;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@Component @Component
@Slf4j
public class DefaultRunContextManager implements RunContextManager { public class DefaultRunContextManager implements RunContextManager {
private final Map<String,PipelineRunContext> contextMap= new ConcurrentHashMap(); private final Map<String,PipelineRunContext> contextMap= new ConcurrentHashMap();
@Resource @Resource
private LoggerManager loggerManager; private LoggerManager loggerManager;
@Resource @Resource
private PipTaskDao taskDao; private PipTaskDao taskDao;
@Resource
private PipPipelineHisInstanceDao hisInstanceDao;
@Resource
private PipelineService pipelineService;
@Override @Override
public Boolean stopPipeline(String pipelineId) { public Boolean stopPipeline(String pipelineId) {
return null; return null;
@ -86,26 +65,13 @@ public class DefaultRunContextManager implements RunContextManager {
if (context instanceof PipelineRunContext pipelineRunContext) { if (context instanceof PipelineRunContext pipelineRunContext) {
if (contextMap.containsKey(id)) { if (contextMap.containsKey(id)) {
PipelineRunContext oldPipeline = contextMap.get(id); PipelineRunContext oldPipeline = contextMap.get(id);
if (!ContextStateEnum.HAPPY_ENDING.getCode().equals(oldPipeline.getState().get()) oldPipeline.changeContextStateAndChild(ContextStateEnum.BAD_ENDING);
&&!ContextStateEnum.BAD_ENDING.getCode().equals(oldPipeline.getState().get())) { LambdaQueryWrapper<PipTask> wrapper = new LambdaQueryWrapper<>();
throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"有未完成执行的流水线请稍后再试"); List<PipTask> taskList = taskDao.selectList(wrapper);
} List<String> taskIdList = taskList.stream().map(PipTask::getId).toList();
// 清空上一次的日志
loggerManager.flushMemory(taskIdList);
} }
pipelineRunContext.setLoggerManager(loggerManager);
pipelineRunContext.setHisInstanceDao(hisInstanceDao);
SecurityContext securityContext = SecurityContextHolder.getContext();
if (securityContext !=null) {
Authentication authentication = securityContext.getAuthentication();
if (authentication!=null) {
if (authentication.getPrincipal() instanceof LoginUser user) {
pipelineRunContext.setUserId(String.valueOf(user.getId()));
pipelineRunContext.setNickName(user.getInfo().get("nickname"));
}
}
}
pipelineRunContext.setContextManager(this);
contextMap.put(id,pipelineRunContext); contextMap.put(id,pipelineRunContext);
} else { } else {
if (parentContext==null) { if (parentContext==null) {
@ -146,43 +112,5 @@ public class DefaultRunContextManager implements RunContextManager {
} }
pipelineRunContext.changeContextStateAndChild(stateEnum); pipelineRunContext.changeContextStateAndChild(stateEnum);
} }
@Transactional(rollbackFor = Exception.class)
public void toHistory(String pipelineId){
log.info("========================开始入库");
BaseRunContext context = getContext(pipelineId);
PipBaseElement contextDef = context.getContextDef();
if (context instanceof PipelineRunContext pipelineRunContext) {
if (contextDef instanceof PipPipeline pipeline){
Collection<BaseRunContext> values = pipelineRunContext.getChildContext().values();
List<TaskRunContext> taskContextList = new LinkedList<>();
for (BaseRunContext value : values) {
Collection<BaseRunContext> childContext = value.getChildContext().values();
List<TaskRunContext> list = childContext.stream().map(it -> (TaskRunContext) it).toList();
taskContextList.addAll(list);
}
loggerManager.flushMemory(taskContextList);
PipPipelineHisInstance pipPipelineHisInstance = new PipPipelineHisInstance();
pipPipelineHisInstance.setPipelineId(pipeline.getId());
pipPipelineHisInstance.setPipelineName(pipeline.getName());
AtomicInteger state = pipelineRunContext.getState();
pipPipelineHisInstance.setState(String.valueOf(state.get()));
pipPipelineHisInstance.setStateName(ContextStateEnum.getByCode(state.get()).getMsg());
pipPipelineHisInstance.setStartTime(pipelineRunContext.getStartTime());
pipPipelineHisInstance.setExecutorUserId(pipelineRunContext.getUserId());
pipPipelineHisInstance.setExecutorUserName(pipelineRunContext.getNickName());
pipPipelineHisInstance.setEndTime(pipelineRunContext.getEndTime());
PiplineTriggerModeEnum triggerMode = pipelineRunContext.getTriggerMode();
pipPipelineHisInstance.setTriggerMode(triggerMode.getCode());
pipPipelineHisInstance.setContextTree(JSON.toJSONString(pipelineRunContext));
PipelineQueryReq pipelineQueryReq = new PipelineQueryReq();
pipelineQueryReq.setId(pipelineId);
pipPipelineHisInstance.setDefTree(JSON.toJSONString(pipelineService.findPipelineById(pipelineQueryReq)));
pipPipelineHisInstance.setStateTree(JSON.toJSONString(pipelineService.getPipelineRunState(pipelineId)));
pipPipelineHisInstance.setTargetVersionName("");
pipPipelineHisInstance.setTargetVersionId(pipeline.getTargetVersionId());
hisInstanceDao.insert(pipPipelineHisInstance);
}
}
}
} }

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

@ -1,14 +1,8 @@
package cd.casic.ci.process.engine.manager.impl; package cd.casic.ci.process.engine.manager.impl;
import cd.casic.ci.process.engine.constant.EngineRuntimeConstant;
import cd.casic.ci.process.engine.manager.LoggerManager; import cd.casic.ci.process.engine.manager.LoggerManager;
import cd.casic.ci.process.engine.runContext.TaskRunContext;
import cd.casic.ci.process.process.dal.pipeline.PipTaskLogDao; import cd.casic.ci.process.process.dal.pipeline.PipTaskLogDao;
import cd.casic.ci.process.process.dataObject.log.PipTaskLog; import cd.casic.ci.process.process.dataObject.log.PipTaskLog;
import cd.casic.ci.process.process.dataObject.task.PipTask;
import cd.casic.ci.process.process.service.task.TaskService;
import cd.casic.ci.process.util.snowflake.SnowflakeIdWorker;
import cd.casic.ci.process.util.snowflake.SnowflakeIdentifierGenerator;
import cd.casic.framework.commons.util.network.IpUtil; import cd.casic.framework.commons.util.network.IpUtil;
import cd.casic.framework.commons.util.util.WebFrameworkUtils; import cd.casic.framework.commons.util.util.WebFrameworkUtils;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
@ -37,10 +31,6 @@ public class MemoryLogManager implements LoggerManager {
private final Map<String, List<SseEmitter>> taskIdSSEMap = new ConcurrentHashMap<>(); private final Map<String, List<SseEmitter>> taskIdSSEMap = new ConcurrentHashMap<>();
private final Map<String,StringBuffer> taskIdMemoryLogMap = new ConcurrentHashMap<>(); private final Map<String,StringBuffer> taskIdMemoryLogMap = new ConcurrentHashMap<>();
@Resource
private SnowflakeIdentifierGenerator identifierGenerator;
@Resource
private TaskService taskService;
public final Integer FLUSH_DB_SIZE=2*1024*1024; public final Integer FLUSH_DB_SIZE=2*1024*1024;
// public final Integer FLUSH_DB_SIZE=1000; // public final Integer FLUSH_DB_SIZE=1000;
@ -117,43 +107,41 @@ public class MemoryLogManager implements LoggerManager {
} }
} }
} }
public void flushMemory(List<TaskRunContext> taskContextList){ public void flushMemory(List<String> taskIdList){
List<PipTaskLog> insertList = new ArrayList<>(); List<PipTaskLog> insertList = new ArrayList<>();
List<PipTaskLog> updateList = new ArrayList<>(); List<PipTaskLog> updateList = new ArrayList<>();
for (TaskRunContext taskRunContext : taskContextList) { for (String taskId : taskIdList) {
String taskId = taskRunContext.getContextDef().getId();
StringBuffer logCache = taskIdMemoryLogMap.get(taskId); StringBuffer logCache = taskIdMemoryLogMap.get(taskId);
if (taskIdDbMap.containsKey(taskId)) { if (logCache!=null) {
// 之前已经入库过 if (taskIdDbMap.containsKey(taskId)) {
// 存在则更新 // 之前已经入库过
String id = taskIdDbMap.get(taskId); // 存在则更新
PipTaskLog pipTaskLog = logDao.selectById(id); String id = taskIdDbMap.get(taskId);
if (logCache!=null) { PipTaskLog pipTaskLog = logDao.selectById(id);
pipTaskLog.setContent(pipTaskLog.getContent()+logCache); // TODO 之后优化
} pipTaskLog.setContent(pipTaskLog.getContent()+logCache.toString());
updateList.add(pipTaskLog); updateList.add(pipTaskLog);
taskIdMemoryLogMap.remove(taskId); taskIdDbMap.remove(taskId);
taskRunContext.getLocalVariables().put(EngineRuntimeConstant.HIS_LOG_KEY,pipTaskLog.getId()); taskIdMemoryLogMap.remove(taskId);
List<SseEmitter> sseEmitters = taskIdSSEMap.get(taskId); List<SseEmitter> sseEmitters = taskIdSSEMap.get(taskId);
if (!CollectionUtils.isEmpty(sseEmitters)) { if (!CollectionUtils.isEmpty(sseEmitters)) {
sseEmitters.forEach(ResponseBodyEmitter::complete); sseEmitters.forEach(ResponseBodyEmitter::complete);
} }
// logDao.updateById(pipTaskLog); // logDao.updateById(pipTaskLog);
} else { } else {
// 不存在就新增 // 不存在就新增
PipTaskLog pipTaskLog = new PipTaskLog(); PipTaskLog pipTaskLog = new PipTaskLog();
pipTaskLog.setTaskId(taskId); pipTaskLog.setTaskId(taskId);
if (logCache!=null) {
pipTaskLog.setContent(logCache.toString()); pipTaskLog.setContent(logCache.toString());
} // logDao.insert(pipTaskLog);
pipTaskLog.setId(identifierGenerator.nextUUID(null)); // taskIdDbMap.put(taskId,pipTaskLog.getId());
insertList.add(pipTaskLog); insertList.add(pipTaskLog);
taskRunContext.getLocalVariables().put(EngineRuntimeConstant.HIS_LOG_KEY,pipTaskLog.getId()); taskIdDbMap.remove(taskId);
taskIdMemoryLogMap.remove(taskId); taskIdMemoryLogMap.remove(taskId);
taskIdDbMap.put(taskId,pipTaskLog.getId()); List<SseEmitter> sseEmitters = taskIdSSEMap.get(taskId);
List<SseEmitter> sseEmitters = taskIdSSEMap.get(taskId); if (!CollectionUtils.isEmpty(sseEmitters)) {
if (!CollectionUtils.isEmpty(sseEmitters)) { sseEmitters.forEach(ResponseBodyEmitter::complete);
sseEmitters.forEach(ResponseBodyEmitter::complete); }
} }
} }
} }
@ -176,12 +164,6 @@ public class MemoryLogManager implements LoggerManager {
} }
return logCache.toString(); return logCache.toString();
} }
public void clear(String pipelineId){
PipTask query = new PipTask();
query.setPipelineId(pipelineId);
List<PipTask> taskList = taskService.getTask(query);
taskList.forEach(it->taskIdDbMap.remove(it.getId()));
}
private void emitterInit(SseEmitter emitter,String taskId){ private void emitterInit(SseEmitter emitter,String taskId){
// 维持心跳 // 维持心跳

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

@ -4,7 +4,6 @@ import cd.casic.ci.process.engine.enums.ContextStateEnum;
import cd.casic.ci.process.process.dataObject.base.PipBaseElement; import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
import cd.casic.framework.commons.exception.ServiceException; import cd.casic.framework.commons.exception.ServiceException;
import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants; import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants;
import com.alibaba.fastjson.annotation.JSONField;
import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Data; import lombok.Data;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -23,41 +22,39 @@ public abstract class BaseRunContext {
/** /**
* 当前上下文的定义 * 当前上下文的定义
* */ * */
protected PipBaseElement contextDef; private PipBaseElement contextDef;
@Transient @Transient
@JsonIgnore @JsonIgnore
@JSONField(serialize = false) private BaseRunContext parentContext;
protected BaseRunContext parentContext; private Integer childCount;
protected Integer childCount;
/** /**
* 运行状态 * 运行状态
* */ * */
protected final AtomicInteger state; private final AtomicInteger state;
/** /**
* 启动时间 * 启动时间
* */ * */
protected LocalDateTime startTime; private LocalDateTime startTime;
/** /**
* 结束时间 * 结束时间
* */ * */
protected LocalDateTime endTime; private LocalDateTime endTime;
protected String resourceId; private String resourceId;
protected String targetVersionId; private String targetVersionId;
protected String targetType; private String targetType;
/** /**
* 整个流水线全局的变量 * 整个流水线全局的变量
* */ * */
protected final Map<String,Object> globalVariables; private Map<String,Object> globalVariables;
/** /**
* 当前上下文局部变量 * 当前上下文局部变量
* */ * */
protected final Map<String,Object> localVariables; private Map<String,Object> localVariables;
protected Map<String,BaseRunContext> childContext; private Map<String,BaseRunContext> childContext;
/** /**
* 用来在控制其他地方的阻塞放行,countDown为1 * 用来在控制其他地方的阻塞放行,countDown为1
* */ * */
@JsonIgnore private CountDownLatch countDownLatch;
protected CountDownLatch countDownLatch;
public BaseRunContext(PipBaseElement contextDef,Integer childCount,BaseRunContext parentContext, LocalDateTime startTime, String resourceId, String targetVersionId, 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 targetVersionId, String targetType, Map<String, Object> globalVariables, Map<String, Object> localVariables, Map<String, BaseRunContext> childContext) {
this.contextDef = contextDef; this.contextDef = contextDef;
@ -80,10 +77,9 @@ public abstract class BaseRunContext {
||ContextStateEnum.BAD_ENDING.equals(stateEnum) ||ContextStateEnum.BAD_ENDING.equals(stateEnum)
||ContextStateEnum.SKIP_TO.equals(stateEnum)) { ||ContextStateEnum.SKIP_TO.equals(stateEnum)) {
this.endTime=LocalDateTime.now(); this.endTime=LocalDateTime.now();
if(this instanceof PipelineRunContext pipelineRunContext){ }
// 流水线执行结束 进行入库 if(this instanceof PipelineRunContext){
log.info("debug专用");
}
} }
callParentChange(stateEnum); callParentChange(stateEnum);
} else { } else {
@ -197,8 +193,7 @@ public abstract class BaseRunContext {
if (ContextStateEnum.canGoto(ContextStateEnum.getByCode(state.get()),ContextStateEnum.RUNNING)) { if (ContextStateEnum.canGoto(ContextStateEnum.getByCode(state.get()),ContextStateEnum.RUNNING)) {
this.changeContextState(ContextStateEnum.RUNNING); this.changeContextState(ContextStateEnum.RUNNING);
} else{ } else{
log.error("非正常状态扭转{}->{}",ContextStateEnum.getByCode(state.get()),ContextStateEnum.RUNNING); throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"状态有误");
// throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"状态有误");
} }
} }
} }

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

@ -1,45 +1,18 @@
package cd.casic.ci.process.engine.runContext; package cd.casic.ci.process.engine.runContext;
import cd.casic.ci.process.engine.enums.ContextStateEnum; import cd.casic.ci.process.engine.enums.ContextStateEnum;
import cd.casic.ci.process.engine.manager.LoggerManager;
import cd.casic.ci.process.engine.manager.RunContextManager;
import cd.casic.ci.process.enums.PiplineTriggerModeEnum;
import cd.casic.ci.process.process.dal.history.PipPipelineHisInstanceDao;
import cd.casic.ci.process.process.dataObject.base.PipBaseElement; import cd.casic.ci.process.process.dataObject.base.PipBaseElement;
import cd.casic.ci.process.process.dataObject.history.PipPipelineHisInstance;
import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline; import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline;
import cd.casic.framework.commons.exception.ServiceException; import cd.casic.framework.commons.exception.ServiceException;
import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants; import cd.casic.framework.commons.exception.enums.GlobalErrorCodeConstants;
import cd.casic.framework.commons.util.util.WebFrameworkUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.annotation.JSONField;
import com.fasterxml.jackson.annotation.JsonIgnore;
import lombok.Data;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
@Setter
@Getter
@Slf4j
public class PipelineRunContext extends BaseRunContext{
@JsonIgnore
@JSONField(serialize = false)
private PipPipelineHisInstanceDao hisInstanceDao;
@JsonIgnore
@JSONField(serialize = false)
private LoggerManager loggerManager;
private PiplineTriggerModeEnum triggerMode;
private String userId;
private String nickName;
@JSONField(serialize = false)
private RunContextManager contextManager;
public PipelineRunContext(PipPipeline pipeline, Integer childCount) { public class PipelineRunContext extends BaseRunContext{
public PipelineRunContext(PipPipeline pipeline,Integer childCount) {
this(pipeline,childCount,null,new ConcurrentHashMap<>(),new ConcurrentHashMap<>()); this(pipeline,childCount,null,new ConcurrentHashMap<>(),new ConcurrentHashMap<>());
} }
@ -87,22 +60,4 @@ public class PipelineRunContext extends BaseRunContext{
throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"不支持类型"); throw new ServiceException(GlobalErrorCodeConstants.PIPELINE_ERROR.getCode(),"不支持类型");
} }
} }
@Override
public void changeContextState(ContextStateEnum stateEnum){
ContextStateEnum curr = ContextStateEnum.getByCode(getState().get());
if (ContextStateEnum.canGoto(curr,stateEnum)) {
state.compareAndExchange(curr.getCode(),stateEnum.getCode());
if (ContextStateEnum.HAPPY_ENDING.equals(stateEnum)
||ContextStateEnum.BAD_ENDING.equals(stateEnum)
||ContextStateEnum.SKIP_TO.equals(stateEnum)) {
this.endTime=LocalDateTime.now();
// 入库保存
contextManager.toHistory(getContextDef().getId());
}
callParentChange(stateEnum);
} else {
log.error("非法状态扭转直接忽略");
}
}
} }

3
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/scheduler/job/PipelineSchedulingJob.java
View File

@ -1,7 +1,6 @@
package cd.casic.ci.process.engine.scheduler.job; package cd.casic.ci.process.engine.scheduler.job;
import cd.casic.ci.process.engine.executor.PipelineExecutor; import cd.casic.ci.process.engine.executor.PipelineExecutor;
import cd.casic.ci.process.enums.PiplineTriggerModeEnum;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
import org.joda.time.LocalDateTime; import org.joda.time.LocalDateTime;
import org.quartz.Job; import org.quartz.Job;
@ -28,7 +27,7 @@ public class PipelineSchedulingJob implements Job {
String pipelineId = dataMap.getString(JOB_PARAM_PIPELINE_ID); String pipelineId = dataMap.getString(JOB_PARAM_PIPELINE_ID);
if (pipelineId != null && executor != null) { if (pipelineId != null && executor != null) {
executor.execute(pipelineId, PiplineTriggerModeEnum.TIMING,null); executor.execute(pipelineId);
System.out.println("定时任务开始执行，当前执行时间为" + new LocalDateTime()); System.out.println("定时任务开始执行，当前执行时间为" + new LocalDateTime());
} }
} }

2
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/worker/ApplicationWorker.java
View File

@ -58,6 +58,7 @@ public class ApplicationWorker extends HttpWorker {
public void execute(TaskRunContext context) { public void execute(TaskRunContext context) {
int statusCode = 0; int statusCode = 0;
Map<String, Object> localVariables = context.getLocalVariables(); Map<String, Object> localVariables = context.getLocalVariables();
PipTaskLog pipTaskLog = (PipTaskLog) localVariables.get(EngineRuntimeConstant.LOG_KEY);
PipBaseElement contextDef = context.getContextDef(); PipBaseElement contextDef = context.getContextDef();
append(context,"SCA-应用包审查分析节点开始执行"); append(context,"SCA-应用包审查分析节点开始执行");
@ -104,6 +105,7 @@ public class ApplicationWorker extends HttpWorker {
} }
} }
localVariables.put("statusCode", statusCode + ""); localVariables.put("statusCode", statusCode + "");
localVariables.put(EngineRuntimeConstant.LOG_KEY, pipTaskLog);
} }
private void handleUpload(Map<String,Object> applicationConfigInfo, File file, BaseRunContext context) throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException { private void handleUpload(Map<String,Object> applicationConfigInfo, File file, BaseRunContext context) throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException {

2
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/worker/CodingWorker.java
View File

@ -49,6 +49,7 @@ public class CodingWorker extends HttpWorker {
public void execute(TaskRunContext context) { public void execute(TaskRunContext context) {
int statusCode = 0; int statusCode = 0;
Map<String, Object> localVariables = context.getLocalVariables(); Map<String, Object> localVariables = context.getLocalVariables();
PipTaskLog pipTaskLog = (PipTaskLog) localVariables.get(EngineRuntimeConstant.LOG_KEY);
PipBaseElement contextDef = context.getContextDef(); PipBaseElement contextDef = context.getContextDef();
append(context,"SCA-代码仓库管理节点开始执行"); append(context,"SCA-代码仓库管理节点开始执行");
@ -131,6 +132,7 @@ public class CodingWorker extends HttpWorker {
} }
localVariables.put("statusCode", statusCode + ""); localVariables.put("statusCode", statusCode + "");
localVariables.put(EngineRuntimeConstant.LOG_KEY, pipTaskLog);
} }

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

@ -29,6 +29,7 @@ public class DIYImageExecuteCommandWorker extends SshWorker {
public void execute(TaskRunContext context) { public void execute(TaskRunContext context) {
int statusCode = -1; int statusCode = -1;
Map<String, Object> localVariables = context.getLocalVariables(); Map<String, Object> localVariables = context.getLocalVariables();
PipTaskLog taskLog = (PipTaskLog) localVariables.get(EngineRuntimeConstant.LOG_KEY);
if (context.getContextDef() instanceof PipTask taskDef) { if (context.getContextDef() instanceof PipTask taskDef) {
log.info(taskDef.getTaskName()); log.info(taskDef.getTaskName());
Map<String, Object> taskProperties = taskDef.getTaskProperties(); Map<String, Object> taskProperties = taskDef.getTaskProperties();

2
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/worker/ScaSbomWorker.java
View File

@ -58,6 +58,7 @@ public class ScaSbomWorker extends HttpWorker {
public void execute(TaskRunContext context) { public void execute(TaskRunContext context) {
int statusCode = 0; int statusCode = 0;
Map<String, Object> localVariables = context.getLocalVariables(); Map<String, Object> localVariables = context.getLocalVariables();
PipTaskLog pipTaskLog = (PipTaskLog) localVariables.get(EngineRuntimeConstant.LOG_KEY);
PipBaseElement contextDef = context.getContextDef(); PipBaseElement contextDef = context.getContextDef();
append(context,"SCA-SBOM节点开始执行"); append(context,"SCA-SBOM节点开始执行");
@ -106,6 +107,7 @@ public class ScaSbomWorker extends HttpWorker {
} }
localVariables.put("statusCode", statusCode + ""); localVariables.put("statusCode", statusCode + "");
localVariables.put(EngineRuntimeConstant.LOG_KEY, pipTaskLog);
} }
private void handleUpload(Map<String,Object> scaSbomConfigInfo, File file, BaseRunContext context) throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException { private void handleUpload(Map<String,Object> scaSbomConfigInfo, File file, BaseRunContext context) throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException {

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

@ -61,20 +61,13 @@ public abstract class BaseWorker implements Runnable{
if (context instanceof TaskRunContext taskRunContext){ if (context instanceof TaskRunContext taskRunContext){
try { try {
taskRunContext.changeContextState(ContextStateEnum.READY); taskRunContext.changeContextState(ContextStateEnum.READY);
Map<String, Object> globalVariables = context.getGlobalVariables(); if (this instanceof PassableWorker passableWorker) {
Object fromError = globalVariables.get(PipelineBehaviorConstant.PIPELINE_EXECUTE_FROM_ERROR); taskRunContext.changeContextState(ContextStateEnum.SUSPEND);
if (Boolean.TRUE.equals(fromError)&&ContextStateEnum.HAPPY_ENDING.getCode().equals(taskRunContext.getState().get())) { passableWorker.waitForPermission();
append(taskRunContext,"跳过执行"+CommandConstant.ENTER); } else {
return; taskRunContext.changeContextState(ContextStateEnum.RUNNING);
} else{
if (this instanceof PassableWorker passableWorker) {
taskRunContext.changeContextState(ContextStateEnum.SUSPEND);
passableWorker.waitForPermission();
} else {
taskRunContext.changeContextState(ContextStateEnum.RUNNING);
}
execute(taskRunContext);
} }
execute(taskRunContext);
} catch (Exception e) { } catch (Exception e) {
log.error("================worker执行报错：",e); log.error("================worker执行报错：",e);
// todo 根据配置决定失败是跳过还是直接失败，如果直接跳过，状态改为跳过，如果直接失败状态就改为失败 // todo 根据配置决定失败是跳过还是直接失败，如果直接跳过，状态改为跳过，如果直接失败状态就改为失败

7
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/dal/history/PipPipelineHisInstanceDao.java
View File

@ -1,7 +0,0 @@
package cd.casic.ci.process.process.dal.history;
import cd.casic.ci.process.process.dataObject.history.PipPipelineHisInstance;
import cd.casic.framework.mybatis.core.mapper.BaseMapperX;
public interface PipPipelineHisInstanceDao extends BaseMapperX<PipPipelineHisInstance> {
}

33
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/dataObject/history/PipPipelineHisInstance.java
View File

@ -1,33 +0,0 @@
package cd.casic.ci.process.process.dataObject.history;
import cd.casic.framework.commons.dataobject.BaseDO;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.time.LocalDateTime;
@EqualsAndHashCode(callSuper = true)
@Data
@TableName("pip_pipeline_his_instance")
public class PipPipelineHisInstance extends BaseDO {
@TableId(type = IdType.ASSIGN_ID)
private String id;
private String pipelineId;
private String contextTree;
private String stateTree;
private String defTree;
private String executorUserId;
private String executorUserName;
private String triggerMode;
private LocalDateTime startTime;
private LocalDateTime endTime;
private String state;
private String stateName;
private String pipelineName;
private String targetVersionId;
private String targetVersionName;
private String description;
}

11
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/service/history/PipelineHistoryService.java
View File

@ -1,11 +0,0 @@
package cd.casic.ci.process.process.service.history;
import cd.casic.ci.process.process.dataObject.history.PipPipelineHisInstance;
import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline;
import com.baomidou.mybatisplus.extension.service.IService;
import java.util.List;
public interface PipelineHistoryService extends IService<PipPipelineHisInstance> {
List<PipPipelineHisInstance> getListByPipelineId(String pipelineId);
}

27
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/service/history/impl/PipelineHistoryServiceImpl.java
View File

@ -1,27 +0,0 @@
package cd.casic.ci.process.process.service.history.impl;
import cd.casic.ci.process.process.dal.history.PipPipelineHisInstanceDao;
import cd.casic.ci.process.process.dal.pipeline.PipelineDao;
import cd.casic.ci.process.process.dataObject.history.PipPipelineHisInstance;
import cd.casic.ci.process.process.dataObject.pipeline.PipPipeline;
import cd.casic.ci.process.process.service.history.PipelineHistoryService;
import cd.casic.ci.process.process.service.pipeline.PipelineService;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;
import java.util.List;
@Service
public class PipelineHistoryServiceImpl extends ServiceImpl<PipPipelineHisInstanceDao, PipPipelineHisInstance> implements PipelineHistoryService {
@Resource
private PipPipelineHisInstanceDao pipelineHisInstanceDao;
@Override
public List<PipPipelineHisInstance> getListByPipelineId(String pipelineId) {
LambdaQueryWrapper<PipPipelineHisInstance> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(PipPipelineHisInstance::getPipelineId,pipelineId);
return pipelineHisInstanceDao.selectList(wrapper);
}
}

8
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/service/taskLog/TaskLogService.java
View File

@ -1,8 +0,0 @@
package cd.casic.ci.process.process.service.taskLog;
import cd.casic.ci.process.process.dataObject.log.PipTaskLog;
import cd.casic.ci.process.process.dataObject.task.PipTask;
import com.baomidou.mybatisplus.extension.service.IService;
public interface TaskLogService extends IService<PipTaskLog> {
}

11
modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/service/taskLog/impl/TaskLogServiceImpl.java
View File

@ -1,11 +0,0 @@
package cd.casic.ci.process.process.service.taskLog.impl;
import cd.casic.ci.process.process.dal.pipeline.PipTaskLogDao;
import cd.casic.ci.process.process.dataObject.log.PipTaskLog;
import cd.casic.ci.process.process.service.taskLog.TaskLogService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.stereotype.Service;
@Service
public class TaskLogServiceImpl extends ServiceImpl<PipTaskLogDao, PipTaskLog> implements TaskLogService {
}

2
ops-server/src/main/java/cd/casic/server/OpsServerApplication.java
View File

@ -5,7 +5,6 @@ import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.ComponentScans; import org.springframework.context.annotation.ComponentScans;
import org.springframework.security.core.context.SecurityContextHolder;
/** /**
* 项目的启动类 * 项目的启动类
@ -18,6 +17,7 @@ import org.springframework.security.core.context.SecurityContextHolder;
@MapperScan(basePackages = {"cd.casic.**.dal","cd.casic.**.dao"}) @MapperScan(basePackages = {"cd.casic.**.dal","cd.casic.**.dao"})
@SpringBootApplication(scanBasePackages = {"${ops.info.base-package}.server", "${ops.info.base-package}.module"}) @SpringBootApplication(scanBasePackages = {"${ops.info.base-package}.server", "${ops.info.base-package}.module"})
public class OpsServerApplication { public class OpsServerApplication {
public static void main(String[] args) { public static void main(String[] args) {
SpringApplication.run(OpsServerApplication.class, args); SpringApplication.run(OpsServerApplication.class, args);
} }

12
ops-server/src/test/java/cd/casic/server/PipelineExecuteTest.java
View File

@ -1,7 +1,6 @@
package cd.casic.server; package cd.casic.server;
import cd.casic.ci.process.engine.executor.PipelineExecutor; import cd.casic.ci.process.engine.executor.PipelineExecutor;
import cd.casic.ci.process.enums.PiplineTriggerModeEnum;
import cd.casic.ci.process.process.service.pipeline.PipelineService; import cd.casic.ci.process.process.service.pipeline.PipelineService;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
@ -18,7 +17,7 @@ public class PipelineExecuteTest {
// 执行pipeline // 执行pipeline
@Test @Test
public void executePipeline(){ public void executePipeline(){
pipelineExecutor.execute("716299522803896320", PiplineTriggerModeEnum.HAND,false); pipelineExecutor.execute("716299522803896320");
} }
// 获取pipeline执行状态 // 获取pipeline执行状态
@Test @Test
@ -28,18 +27,11 @@ public class PipelineExecuteTest {
@Test @Test
public void taskSkipExecute(){ public void taskSkipExecute(){
// 这个流水线包含了故意报错的worker，可以验证跳过效果 // 这个流水线包含了故意报错的worker，可以验证跳过效果
pipelineExecutor.execute("718104543308681216",PiplineTriggerModeEnum.HAND,false); pipelineExecutor.execute("718104543308681216");
} }
@Test @Test
public void taskSkipGetState(){ public void taskSkipGetState(){
// 这个流水线包含了故意报错的worker，可以验证跳过效果 // 这个流水线包含了故意报错的worker，可以验证跳过效果
pipelineService.getPipelineRunState("718104543308681216"); pipelineService.getPipelineRunState("718104543308681216");
} }
@Test
public void fromError(){
// 这个流水线包含了故意报错的worker，可以验证跳过效果
// 传false 就是从头执行，传false就是跳过正确的节点从错误处执行
pipelineExecutor.execute("718104543308681216",PiplineTriggerModeEnum.HAND,false);
pipelineExecutor.execute("718104543308681216",PiplineTriggerModeEnum.HAND,true);
}
} }