feature: 原子操作能力支持滚动执行 TencentBlueKing#446

wangyu096 · Jan 17, 2022 · 3ce1583 · 3ce1583
1 parent 84a94b0
commit 3ce1583
Show file tree

Hide file tree

Showing 12 changed files with 38 additions and 16 deletions.
diff --git a/...ain/java/com/tencent/bk/job/execute/api/esb/v2/impl/EsbGetJobInstanceLogResourceImpl.java b/...ain/java/com/tencent/bk/job/execute/api/esb/v2/impl/EsbGetJobInstanceLogResourceImpl.java
@@ -81,8 +81,9 @@ public EsbResp<List<EsbStepInstanceResultAndLog>> getJobInstanceLogUsingPost(Esb
             taskInstanceService.listStepInstanceByTaskInstanceId(taskInstanceId);
         List<EsbStepInstanceResultAndLog> stepInstResultAndLogList = Lists.newArrayList();
         for (StepInstanceBaseDTO stepInstance : stepInstanceList) {
+            // TODO Rolling
             GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstance.getId(),
-                stepInstance.getExecuteCount());
+                stepInstance.getExecuteCount(), 0);
             if (null == gseTask) {
                 EsbStepInstanceResultAndLog stepInstResultAndLog = new EsbStepInstanceResultAndLog();
                 stepInstResultAndLog.setFinished(false);

diff --git a/...java/com/tencent/bk/job/execute/api/esb/v2/impl/EsbGetStepInstanceStatusResourceImpl.java b/...java/com/tencent/bk/job/execute/api/esb/v2/impl/EsbGetStepInstanceStatusResourceImpl.java
@@ -92,7 +92,8 @@ public EsbResp<EsbStepInstanceStatusDTO> getJobStepInstanceStatus(EsbGetStepInst
         }
 
         EsbStepInstanceStatusDTO resultData = new EsbStepInstanceStatusDTO();
-        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, stepInstance.getExecuteCount());
+        // TODO Rolling
+        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, stepInstance.getExecuteCount(), 0);
         if (null == gseTask) {
             resultData.setIsFinished(false);
             return EsbResp.buildSuccessResp(null);

diff --git a/...-execute/service-job-execute/src/main/java/com/tencent/bk/job/execute/dao/GseTaskDAO.java b/...-execute/service-job-execute/src/main/java/com/tencent/bk/job/execute/dao/GseTaskDAO.java
@@ -26,10 +26,26 @@
 
 import com.tencent.bk.job.execute.model.GseTaskDTO;
 
+/**
+ * GseTaskDAO
+ */
 public interface GseTaskDAO {
 
+    /**
+     * 保存/更新 GSE 任务
+     *
+     * @param gseTask GSE 任务
+     */
     void saveGseTask(GseTaskDTO gseTask);
 
-    GseTaskDTO getGseTask(long stepInstanceId, int executeCount);
+    /**
+     * 获取 GSE 任务
+     *
+     * @param stepInstanceId 步骤实例ID
+     * @param executeCount   步骤执行次数
+     * @param batch          滚动执行批次
+     * @return GSE 任务
+     */
+    GseTaskDTO getGseTask(long stepInstanceId, int executeCount, int batch);
 
 }
diff --git a/...service-job-execute/src/main/java/com/tencent/bk/job/execute/dao/impl/GseTaskDAOImpl.java b/...service-job-execute/src/main/java/com/tencent/bk/job/execute/dao/impl/GseTaskDAOImpl.java
@@ -96,11 +96,11 @@ public void saveGseTask(GseTaskDTO gseTask) {
     }
 
     @Override
-    public GseTaskDTO getGseTask(long stepInstanceId, int executeCount) {
+    public GseTaskDTO getGseTask(long stepInstanceId, int executeCount, int batch) {
         Record record = dslContext.select(ALL_FIELDS).from(TABLE)
             .where(TABLE.STEP_INSTANCE_ID.eq(stepInstanceId))
             .and(TABLE.EXECUTE_COUNT.eq(executeCount))
-            .and(TABLE.BATCH.eq((short) 0))
+            .and(TABLE.BATCH.eq((short) batch))
             .fetchOne();
         return extractInfo(record);
     }

diff --git a/...e/service-job-execute/src/main/java/com/tencent/bk/job/execute/engine/GseTaskManager.java b/...e/service-job-execute/src/main/java/com/tencent/bk/job/execute/engine/GseTaskManager.java
@@ -397,7 +397,7 @@ public void stopStep(long stepInstanceId, String requestId) {
 
         TaskInstanceDTO taskInstance = taskInstanceService.getTaskInstance(stepInstance.getTaskInstanceId());
         int executeCount = stepInstance.getExecuteCount();
-        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount);
+        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount, stepInstance.getBatch());
         if (null == gseTask) {
             log.info("Get gseTask return null, stepInstanceId: {}, executeCount:{}", stepInstanceId, executeCount);
             return;

diff --git a/...ute/src/main/java/com/tencent/bk/job/execute/engine/executor/AbstractGseTaskExecutor.java b/...ute/src/main/java/com/tencent/bk/job/execute/engine/executor/AbstractGseTaskExecutor.java
@@ -341,7 +341,7 @@ public void execute() {
         watch.stop();
 
         watch.start("get-gse-task-log-from-db");
-        gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount);
+        gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount, rollingBatch);
         watch.stop();
 
         boolean shouldSendTaskToGseServer = (gseTask == null || StringUtils.isEmpty(gseTask.getGseTaskId()));

diff --git a/...ob-execute/src/main/java/com/tencent/bk/job/execute/engine/executor/FileTaskExecutor.java b/...ob-execute/src/main/java/com/tencent/bk/job/execute/engine/executor/FileTaskExecutor.java
@@ -366,7 +366,7 @@ public GseTaskExecuteResult stopGseTask() {
             agentList.add(src);
         }
 
-        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount);
+        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount, rollingBatch);
         if (gseTask == null || StringUtils.isEmpty(gseTask.getGseTaskId())) {
             log.warn("Gse Task not send to gse server, not support stop");
             return new GseTaskExecuteResult(GseTaskExecuteResult.RESULT_CODE_STOP_FAILED, "Termination failed");

diff --git a/...-execute/src/main/java/com/tencent/bk/job/execute/engine/executor/ScriptTaskExecutor.java b/...-execute/src/main/java/com/tencent/bk/job/execute/engine/executor/ScriptTaskExecutor.java
@@ -605,7 +605,7 @@ public GseTaskExecuteResult stopGseTask() {
         List<api_agent> agentList = GseRequestUtils.buildAgentList(jobIpSet, accountInfo.getAccount(),
             accountInfo.getPassword());
         api_stop_task_request stopTaskRequest = new api_stop_task_request();
-        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount);
+        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount, rollingBatch);
         if (gseTask == null || StringUtils.isEmpty(gseTask.getGseTaskId())) {
             log.warn("Gse Task not send to gse server, not support stop");
             return new GseTaskExecuteResult(GseTaskExecuteResult.RESULT_CODE_STOP_FAILED, "Termination failed");

diff --git a/.../src/main/java/com/tencent/bk/job/execute/engine/listener/ResultHandleResumeListener.java b/.../src/main/java/com/tencent/bk/job/execute/engine/listener/ResultHandleResumeListener.java
@@ -144,7 +144,7 @@ public void handleEvent(StepEvent stepEvent) {
         try {
             StepInstanceDTO stepInstance = taskInstanceService.getStepInstanceDetail(stepInstanceId);
             TaskInstanceDTO taskInstance = taskInstanceService.getTaskInstance(stepInstance.getTaskInstanceId());
-            GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount);
+            GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstanceId, executeCount, stepInstance.getBatch());
 
             if (!checkIsTaskResumeable(stepInstance, gseTask)) {
                 log.warn("Task can not resume, stepStatus: {}, gseTaskStatus: {}",

diff --git a/.../service-job-execute/src/main/java/com/tencent/bk/job/execute/service/GseTaskService.java b/.../service-job-execute/src/main/java/com/tencent/bk/job/execute/service/GseTaskService.java
@@ -50,9 +50,10 @@ public interface GseTaskService {
      *
      * @param stepInstanceId 步骤实例ID
      * @param executeCount   步骤执行次数
+     * @param batch          滚动执行批次
      * @return GSE 任务
      */
-    GseTaskDTO getGseTask(long stepInstanceId, int executeCount);
+    GseTaskDTO getGseTask(long stepInstanceId, int executeCount, int batch);
 
     /**
      * 批量保存 GSE Agent 任务
@@ -116,7 +117,8 @@ void batchUpdateGseAgentTasks(long stepInstanceId,
     List<GseAgentTaskDTO> listGseAgentTasksByResultType(Long stepInstanceId, Integer executeCount, Integer resultType,
                                                         String tag);
 
-    List<GseAgentTaskDTO> getGseAgentTaskContentByResultType(Long stepInstanceId, Integer executeCount, Integer resultType,
+    List<GseAgentTaskDTO> getGseAgentTaskContentByResultType(Long stepInstanceId, Integer executeCount,
+                                                             Integer resultType,
                                                              String tag);
 
     /**

diff --git a/...job-execute/src/main/java/com/tencent/bk/job/execute/service/impl/GseTaskServiceImpl.java b/...job-execute/src/main/java/com/tencent/bk/job/execute/service/impl/GseTaskServiceImpl.java
@@ -77,8 +77,8 @@ public void saveGseTask(GseTaskDTO gseTask) {
     }
 
     @Override
-    public GseTaskDTO getGseTask(long stepInstanceId, int executeCount) {
-        return gseTaskDao.getGseTask(stepInstanceId, executeCount);
+    public GseTaskDTO getGseTask(long stepInstanceId, int executeCount, int batch) {
+        return gseTaskDao.getGseTask(stepInstanceId, executeCount, batch);
     }
 
     @Override

diff --git a/...-execute/src/main/java/com/tencent/bk/job/execute/service/impl/TaskResultServiceImpl.java b/...-execute/src/main/java/com/tencent/bk/job/execute/service/impl/TaskResultServiceImpl.java
@@ -504,7 +504,8 @@ public StepExecutionDetailDTO getStepExecutionResult(String username, Long appId
         query.setExecuteCount(finalExecuteCount);
         watch.stop();
 
-        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstance.getId(), finalExecuteCount);
+        // TODO Rolling
+        GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstance.getId(), finalExecuteCount, 0);
         if (gseTask == null) {
             return buildNotStartStepExecutionResult(appId, stepInstance, finalExecuteCount,
                 query.getMaxAgentTasksForResultGroup(), query.getSearchIp());
@@ -885,8 +886,9 @@ public List<IpDTO> getHostsByResultType(String username, Long appId, Long stepIn
             }
         }
 
+        // TODO Rolling
         GseTaskDTO gseTask = gseTaskService.getGseTask(stepInstance.getId(),
-            stepInstance.getExecuteCount());
+            stepInstance.getExecuteCount(), 0);
         if (gseTask == null) {
             if (stepInstance.getTargetServers().getIpList() != null) {
                 return stepInstance.getTargetServers().getIpList();