diff --git a/modules/module-ci-common-pipeline/src/main/java/cd/casic/ci/common/pipeline/pojo/BuildParameters.java b/modules/module-ci-common-pipeline/src/main/java/cd/casic/ci/common/pipeline/pojo/BuildParameters.java
index d71480d..0dec43e 100644
--- a/modules/module-ci-common-pipeline/src/main/java/cd/casic/ci/common/pipeline/pojo/BuildParameters.java
+++ b/modules/module-ci-common-pipeline/src/main/java/cd/casic/ci/common/pipeline/pojo/BuildParameters.java
@@ -38,4 +38,9 @@ public class BuildParameters {
this.desc = desc;
this.defaultValue = defaultValue;
}
+
+ public BuildParameters(String key, String value) {
+ this.key = key;
+ this.value = value;
+ }
}
diff --git a/modules/module-ci-process-api/src/main/java/cd/casic/ci/process/api/process/pojo/BuildTask.java b/modules/module-ci-process-api/src/main/java/cd/casic/ci/process/api/process/pojo/BuildTask.java
index e3d49a4..e333fbb 100644
--- a/modules/module-ci-process-api/src/main/java/cd/casic/ci/process/api/process/pojo/BuildTask.java
+++ b/modules/module-ci-process-api/src/main/java/cd/casic/ci/process/api/process/pojo/BuildTask.java
@@ -2,6 +2,7 @@ package cd.casic.ci.process.api.process.pojo;
import cd.casic.ci.common.pipeline.enums.BuildTaskStatus;
import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
import java.util.Map;
@@ -14,6 +15,7 @@ import java.util.Map;
* @Filename：BuildTask
* @description：Todo
*/
+@Data
@Schema(title = "流水线模型-构建任务")
public class BuildTask {
@Schema(title = "构建ID", required = true)
diff --git a/modules/module-ci-worker/pom.xml b/modules/module-ci-worker/pom.xml
index 47fd3cc..6f19bac 100644
--- a/modules/module-ci-worker/pom.xml
+++ b/modules/module-ci-worker/pom.xml
@@ -13,6 +13,12 @@
module-ci-worker
+
+
+ cd.casic.boot
+ module-ci-process-api
+
+
cd.casic.boot
module-ci-common-pipeline
diff --git a/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/api/log/LogSDKApi.java b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/api/log/LogSDKApi.java
new file mode 100644
index 0000000..099fe5e
--- /dev/null
+++ b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/api/log/LogSDKApi.java
@@ -0,0 +1,28 @@
+package cd.casic.devops.common.worker.api.log;
+
+import cd.casic.ci.log.dal.pojo.TaskBuildLogProperty;
+import cd.casic.ci.log.dal.pojo.enums.LogStorageMode;
+import cd.casic.ci.log.dal.pojo.message.LogMessage;
+
+import java.util.List;
+
+public interface LogSDKApi extends WorkerRestApiSDK {
+
+
+ Result addLogMultiLine(String buildId, List logMessages);
+
+
+ Result finishLog(
+ String tag,
+ String jobId,
+ Integer executeCount,
+ String subTag,
+ LogStorageMode logMode
+ );
+
+
+ Result updateStorageMode(
+ List propertyList,
+ int executeCount
+ );
+}
\ No newline at end of file
diff --git a/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/EmptyTask.java b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/EmptyTask.java
new file mode 100644
index 0000000..0dbb682
--- /dev/null
+++ b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/EmptyTask.java
@@ -0,0 +1,27 @@
+package cd.casic.devops.common.worker.task;
+
+import cd.casic.ci.common.pipeline.enums.ErrorCode;
+import cd.casic.ci.common.pipeline.pojo.ErrorType;
+import java.io.File;
+import cd.casic.ci.process.api.process.pojo.BuildTask;
+import cd.casic.ci.process.api.process.pojo.BuildVariables;
+
+public class EmptyTask extends ITask {
+ private final String type;
+
+ public EmptyTask(String type) {
+ this.type = type;
+ }
+
+ @Override
+ public void execute(BuildTask buildTask, BuildVariables buildVariables, File workspace) {
+ throw new TaskExecuteException(
+ "Received unimplemented build task: " + type,
+ ErrorCode.USER_INPUT_INVAILD,
+ ErrorType.USER
+ );
+ }
+}
+
+
+
diff --git a/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/ITask.java b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/ITask.java
new file mode 100644
index 0000000..af0100c
--- /dev/null
+++ b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/ITask.java
@@ -0,0 +1,149 @@
+package cd.casic.devops.common.worker.task;
+
+import cd.casic.ci.common.pipeline.enums.ErrorCode;
+import cd.casic.ci.common.pipeline.pojo.BuildParameters;
+import cd.casic.ci.common.pipeline.pojo.ErrorType;
+import cd.casic.ci.process.api.process.pojo.BuildTask;
+import cd.casic.ci.process.api.process.pojo.BuildVariables;
+import cd.casic.devops.common.worker.env.BuildEnv;
+import cd.casic.devops.common.worker.env.BuildType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@SuppressWarnings({"NestedMethodCall", "MethodCount"})
+public abstract class ITask {
+
+ private final Logger logger = LoggerFactory.getLogger(ITask.class);
+
+ private final Map environment = new HashMap<>();
+
+ private final Map monitorData = new HashMap<>();
+
+ private String platformCode;
+
+ private Integer platformErrorCode;
+
+ private Boolean finishKillFlag;
+
+ /* 存储常量的key */
+ private List constVar;
+
+ public void run(BuildTask buildTask, BuildVariables buildVariables, File workspace) {
+ // 过滤只读变量并收集键名到列表
+ constVar = buildVariables.getVariablesWithType().stream()
+ .filter(variable -> variable.isReadOnly())
+ .map(variable -> variable.getKey())
+ .collect(Collectors.toList());
+
+ execute(buildTask, buildVariables, workspace);
+ }
+
+ private BuildVariables combineVariables(BuildTask buildTask, BuildVariables buildVariables) {
+
+ Map buildVariable = buildTask.getBuildVariable();
+ if (buildVariable == null) {
+ return buildVariables;
+ }
+
+ Map newVariables = buildVariables.getVariables();
+ newVariables.putAll(buildVariable);
+
+ List buildParameters = buildVariable.entrySet().stream().map(entry -> new BuildParameters(entry.getKey(), entry.getValue())
+ ).collect(Collectors.toList());
+
+ Map newBuildParameters =
+ buildVariables.getVariablesWithType().stream()
+ .collect(Collectors.toMap(
+ BuildParameters::getKey,
+ param -> param,
+ (a, b) -> b // 合并函数（新值优先）
+ ));
+ buildParameters.forEach(param -> newBuildParameters.put(param.getKey(), param));
+
+ BuildVariables restBuildVariables = new BuildVariables();
+ restBuildVariables.setVariables(newVariables);
+ restBuildVariables.setVariablesWithType(newBuildParameters.values().stream().collect(Collectors.toList()));
+ return restBuildVariables;
+ }
+
+ abstract void execute(BuildTask buildTask, BuildVariables buildVariables, File workspace);
+
+ protected void addEnv(Map env) {
+ if (!constVar.isEmpty()) {
+ boolean errFlag = false;
+ for (Map.Entry entry : env.entrySet()) {
+ String key = entry.getKey();
+ if (constVar.contains(key)) {
+ LoggerService.addErrorLine("Variable " + key + " is read-only and cannot be modified.");
+ errFlag = true;
+ }
+ }
+
+ if (errFlag) {
+ throw new TaskExecuteException(
+ "[Finish task] status: false, errorType: " + ErrorType.USER.getNum() +
+ ", errorCode: " + ErrorCode.USER_INPUT_INVAILD +
+ ", message: read-only cannot be modified.",
+ ErrorType.USER,
+ ErrorCode.USER_INPUT_INVAILD
+ );
+ }
+ environment.putAll(env);
+ }
+ }
+
+ protected void addEnv(String key, String value) {
+ environment.put(key, value);
+ }
+
+ protected String getEnv(String key) {
+ return environment.getOrDefault(key, "");
+ }
+
+ public Map getAllEnv() {
+ return environment;
+ }
+
+ protected void addMonitorData(Map monitorDataMap) {
+ monitorData.putAll(monitorDataMap);
+ }
+
+ public Map getMonitorData() {
+ return new HashMap<>(monitorData);
+ }
+
+ protected void addPlatformCode(String taskPlatformCode) {
+ this.platformCode = taskPlatformCode;
+ }
+
+ public String getPlatformCode() {
+ return platformCode;
+ }
+
+ protected void addPlatformErrorCode(int taskPlatformErrorCode) {
+ this.platformErrorCode = taskPlatformErrorCode;
+ }
+
+ public Integer getPlatformErrorCode() {
+ return platformErrorCode;
+ }
+
+ protected void addFinishKillFlag(boolean taskFinishKillFlag) {
+ this.finishKillFlag = taskFinishKillFlag;
+ }
+
+ public Boolean getFinishKillFlag() {
+ return finishKillFlag;
+ }
+
+ protected boolean isThirdAgent() {
+ return BuildEnv.getBuildType() == BuildType.AGENT;
+ }
+
+}
\ No newline at end of file
diff --git a/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskClassType.java b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskClassType.java
new file mode 100644
index 0000000..06f55a5
--- /dev/null
+++ b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskClassType.java
@@ -0,0 +1,17 @@
+package cd.casic.devops.common.worker.task;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+
+@Target({ElementType.TYPE, ElementType.FIELD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface TaskClassType {
+
+ String[] classTypes();
+
+
+ int priority() default 0;
+}
diff --git a/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskDaemon.java b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskDaemon.java
new file mode 100644
index 0000000..b804844
--- /dev/null
+++ b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskDaemon.java
@@ -0,0 +1,127 @@
+package cd.casic.devops.common.worker.task;
+import cd.casic.ci.common.pipeline.enums.ErrorCode;
+import cd.casic.ci.common.pipeline.pojo.ErrorType;
+import cd.casic.ci.process.api.process.pojo.BuildTask;
+import cd.casic.ci.process.api.process.pojo.BuildTaskResult;
+import cd.casic.ci.process.api.process.pojo.BuildVariables;
+import com.github.xiaoymin.knife4j.core.util.CommonUtils;
+import java.io.File;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.concurrent.*;
+
+import static cd.casic.ci.process.api.process.utils.Constants.PIPELINE_TASK_MESSAGE_STRING_LENGTH_MAX;
+
+public class TaskDaemon implements Callable> {
+ private final ITask task;
+ private final BuildTask buildTask;
+ private final BuildVariables buildVariables;
+ private final File workspace;
+ private static final int PARAM_MAX_LENGTH = 4000; // 流水线参数最大长度
+
+ public TaskDaemon(ITask task, BuildTask buildTask, BuildVariables buildVariables, File workspace) {
+ this.task = task;
+ this.buildTask = buildTask;
+ this.buildVariables = buildVariables;
+ this.workspace = workspace;
+ }
+
+ @Override
+ public Map call() {
+ task.run(buildTask, buildVariables, workspace);
+ return task.getAllEnv();
+ }
+
+ public void runWithTimeout() {
+ int timeout = TaskUtil.getTimeOut(buildTask);
+ ExecutorService executor = Executors.newCachedThreadPool();
+ String taskId = buildTask.getTaskId();
+
+ if (taskId != null) {
+ TaskExecutorCache.put(taskId, executor);
+ }
+ Future> f1 = executor.submit(this);
+ try {
+ if (f1.get(timeout, TimeUnit.MINUTES) == null) {
+ throw new TimeoutException("Task[" + buildTask.getElementName() +
+ "] timeout: " + timeout + " minutes");
+ }
+ } catch (Exception e) {
+ throw new TaskExecuteException(
+ ErrorType.USER,
+ ErrorCode.USER_TASK_OUTTIME_LIMIT,
+ e.getMessage() != null ? e.getMessage() :
+ "Task[" + buildTask.getElementName() + "] timeout: " + timeout + " minutes"
+ );
+ } finally {
+ executor.shutdownNow();
+ if (taskId != null) {
+ TaskExecutorCache.invalidate(taskId);
+ }
+ }
+ }
+
+ private Map getAllEnv() {
+ return task.getAllEnv();
+ }
+
+ private Map getMonitorData() {
+ return task.getMonitorData();
+ }
+
+ public BuildTaskResult getBuildResult(
+ Boolean isSuccess,
+ String errorMessage,
+ String errorType,
+ Integer errorCode) {
+ Map allEnv = getAllEnv();
+ Map buildResult = new HashMap<>();
+
+ if (allEnv != null && !allEnv.isEmpty()) {
+ for (Map.Entry entry : allEnv.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ // 检查值长度
+ if (value != null && value.length() > PARAM_MAX_LENGTH) {
+ LoggerService.getInstance().addWarnLine(
+ "Warning, assignment to variable [" + key + "] failed, " +
+ "more than " + PARAM_MAX_LENGTH + " characters(len=" + value.length() + ")"
+ );
+ continue;
+ }
+
+ if (SensitiveValueService.matchSensitiveValue(value)) {
+ LoggerService.getInstance().addWarnLine(
+ "Warning, credentials cannot be assigned to variable[" + key + "]"
+ );
+ continue;
+ }
+
+ buildResult.put(key, value);
+ }
+ }
+
+ return new BuildTaskResult(
+ buildTask.getTaskId(),
+ buildTask.getTaskId(),
+ buildVariables.getContainerHashId(),
+ buildTask.getElementVersion(),
+ isSuccess,
+ buildTask.getExecuteCount(),
+ buildResult,
+ errorMessage != null ?
+ CommonUtils.interceptStringInLength(
+ SensitiveValueService.fixSensitiveContent(errorMessage),
+ PIPELINE_TASK_MESSAGE_STRING_LENGTH_MAX
+ ) : null,
+ buildTask.getType(),
+ errorType,
+ errorCode,
+ task.getPlatformCode(),
+ task.getPlatformErrorCode(),
+ getMonitorData()
+ );
+
+ }
+}
\ No newline at end of file
diff --git a/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskExecutorCache.java b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskExecutorCache.java
new file mode 100644
index 0000000..18d392a
--- /dev/null
+++ b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskExecutorCache.java
@@ -0,0 +1,44 @@
+package cd.casic.devops.common.worker.task;
+
+import cd.casic.ci.process.api.engine.common.Timeout;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Cache;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public final class TaskExecutorCache {
+
+ private static final TaskExecutorCache INSTANCE = new TaskExecutorCache();
+
+ // Caffeine 缓存
+ private final Cache taskExecutorCache = Caffeine.newBuilder()
+ .maximumSize(50)
+ .expireAfterWrite(Timeout.MAX_JOB_RUN_DAYS, TimeUnit.DAYS)
+ .build();
+
+ // 私有构造函数
+ private TaskExecutorCache() {}
+
+ // 获取单例实例
+ public static TaskExecutorCache getInstance() {
+ return INSTANCE;
+ }
+
+ public void invalidate(String taskId) {
+ taskExecutorCache.invalidate(taskId);
+ }
+
+ public void put(String taskId, ExecutorService executor) {
+ taskExecutorCache.put(taskId, executor);
+ }
+
+ public ExecutorService getIfPresent(String taskId) {
+ return taskExecutorCache.getIfPresent(taskId);
+ }
+
+ public Map getAllPresent(Set taskIds) {
+ return taskExecutorCache.getAllPresent(taskIds);
+ }
+}
\ No newline at end of file
diff --git a/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskFactory.java b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskFactory.java
new file mode 100644
index 0000000..083d916
--- /dev/null
+++ b/modules/module-ci-worker/src/main/java/cd/casic/devops/common/worker/task/TaskFactory.java
@@ -0,0 +1,99 @@
+package cd.casic.devops.common.worker.task;
+
+import cd.casic.ci.common.pipeline.pojo.element.agent.LinuxScriptElement;
+import cd.casic.ci.common.pipeline.pojo.element.agent.WindowsScriptElement;
+import cd.casic.ci.common.pipeline.pojo.element.market.MarketBuildAtomElement;
+import cd.casic.ci.common.pipeline.pojo.element.market.MarketBuildLessAtomElement;
+import org.reflections.Reflections;
+
+import java.lang.reflect.Modifier;
+import java.util.HashMap;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.Map;
+
+public final class TaskFactory {
+ // 单例实例
+ private static final TaskFactory INSTANCE = new TaskFactory();
+
+ // 任务类型映射表
+ private final Map> taskMap = new ConcurrentHashMap<>();
+
+ // 私有构造方法
+ private TaskFactory() {}
+
+ // 获取单例实例
+ public static TaskFactory getInstance() {
+ return INSTANCE;
+ }
+
+ @SuppressWarnings("all")
+ public void init() {
+ // 注册基础任务类型
+ register(LinuxScriptElement.classType, LinuxScriptTask.class);
+ register(WindowsScriptElement.classType, WindowsScriptTask.class);
+ register(MarketBuildAtomElement.classType, MarketAtomTask.class);
+ register(MarketBuildLessAtomElement.classType, MarketAtomTask.class);
+
+ // 使用反射扫描任务类
+ Reflections reflections = new Reflections("com.tencent.devops.plugin.worker.task");
+ Set> taskClasses = reflections.getSubTypesOf(ITask.class);
+
+ Map candidatePriorityMap = new HashMap<>();
+ Map> candidateMap = new HashMap<>();
+
+ if (taskClasses != null) {
+ for (Class taskClazz : taskClasses) {
+ if (!Modifier.isAbstract(taskClazz.getModifiers())) {
+ TaskClassType taskClassType = taskClazz.getAnnotation(TaskClassType.class);
+ if (taskClassType != null) {
+ for (String classType : taskClassType.classTypes()) {
+ boolean find = false;
+ Integer priority = candidatePriorityMap.get(classType);
+
+ if (taskClassType.priority() > (priority != null ? priority : 0)) {
+ priority = taskClassType.priority();
+ find = true;
+ }
+
+ if (priority == null || find) {
+ candidatePriorityMap.put(classType, priority != null ? priority : 0);
+ candidateMap.put(classType, taskClazz);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ // 注册扫描到的任务类
+ for (Map.Entry> entry : candidateMap.entrySet()) {
+ register(entry.getKey(), entry.getValue());
+ }
+ }
+
+
+ // 注册任务类型
+ private void register(String classType, Class taskClass) {
+ taskMap.put(classType, taskClass);
+ }
+
+ // 创建任务实例
+ public ITask create(String type) {
+ Class clazz = taskMap.get(type);
+ if (clazz == null) {
+ return new EmptyTask(type);
+ }
+
+ try {
+ // 尝试获取无参构造器
+ java.lang.reflect.Constructor ctor = clazz.getDeclaredConstructor();
+ if (ctor != null) {
+ return ctor.newInstance();
+ }
+ } catch (Exception e) {
+ // 构造失败时返回EmptyTask
+ }
+ return new EmptyTask(type);
+ }
+}
\ No newline at end of file