perf: 支持流水线批量取消运行中构建 TencentBlueKing#10858 并发组接入批量取消功能

royalhuang · Aug 23, 2024 · 2cd4c3a · 2cd4c3a
1 parent cd32648
commit 2cd4c3a
Show file tree

Hide file tree

Showing 3 changed files with 62 additions and 17 deletions.
diff --git a/...iz-base/src/main/kotlin/com/tencent/devops/process/engine/interceptor/QueueInterceptor.kt b/...iz-base/src/main/kotlin/com/tencent/devops/process/engine/interceptor/QueueInterceptor.kt
@@ -29,6 +29,7 @@ package com.tencent.devops.process.engine.interceptor
 
 import com.tencent.devops.common.api.util.MessageUtil
 import com.tencent.devops.common.event.dispatcher.pipeline.PipelineEventDispatcher
+import com.tencent.devops.common.event.enums.ActionType
 import com.tencent.devops.common.log.utils.BuildLogPrinter
 import com.tencent.devops.common.pipeline.enums.BuildStatus
 import com.tencent.devops.common.redis.RedisOperation
@@ -46,6 +47,7 @@ import com.tencent.devops.process.engine.service.PipelineRuntimeService
 import com.tencent.devops.process.engine.service.PipelineTaskService
 import com.tencent.devops.common.pipeline.pojo.setting.PipelineRunLockType
 import com.tencent.devops.process.constant.ProcessMessageCode
+import com.tencent.devops.process.engine.pojo.event.PipelineBuildBatchCancelEvent
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Component
@@ -278,7 +280,6 @@ class QueueInterceptor @Autowired constructor(
                         concurrencyGroup = concurrencyGroup,
                         status = status
                     ).toMutableList()
-                    // #8143 兼容旧流水线版本 TODO 待模板设置补上漏洞，后期下掉 # 8143
                     if (concurrencyGroup == task.pipelineInfo.pipelineId) {
                         builds.addAll(
                             0,
@@ -289,14 +290,32 @@ class QueueInterceptor @Autowired constructor(
                             )
                         )
                     }
+                    // 按照流水线ID组织批量取消的事件
+                    val cancelUser = latestStartUser ?: task.pipelineInfo.creator
+                    val pipelineToEvents = mutableMapOf<String, MutableList<String>>()
                     builds.forEach { (pipelineId, buildId) ->
+                        val cancelBuilds = pipelineToEvents[pipelineId] ?: mutableListOf()
                         pipelineRuntimeService.concurrencyCancelBuildPipeline(
                             projectId = projectId,
                             pipelineId = pipelineId,
                             buildId = buildId,
-                            userId = latestStartUser ?: task.pipelineInfo.creator,
+                            userId = cancelUser,
                             groupName = concurrencyGroup,
-                            detailUrl = detailUrl
+                            detailUrl = detailUrl,
+                            cancelBuilds = cancelBuilds
+                        )
+                        pipelineToEvents[pipelineId] = cancelBuilds
+                    }
+                    pipelineToEvents.forEach { (pipelineId, cancelBuilds) ->
+                        pipelineEventDispatcher.dispatch(
+                            PipelineBuildBatchCancelEvent(
+                                source = "concurrencyGroupCancel",
+                                projectId = projectId,
+                                pipelineId = pipelineId,
+                                userId = cancelUser,
+                                buildIds = cancelBuilds,
+                                actionType = ActionType.END
+                            )
                         )
                     }
                     Response(data = BuildStatus.RUNNING)

diff --git a/...-base/src/main/kotlin/com/tencent/devops/process/engine/service/PipelineRuntimeService.kt b/...-base/src/main/kotlin/com/tencent/devops/process/engine/service/PipelineRuntimeService.kt
@@ -672,7 +672,8 @@ class PipelineRuntimeService @Autowired constructor(
         userId: String,
         executeCount: Int,
         buildStatus: BuildStatus,
-        terminateFlag: Boolean = false
+        terminateFlag: Boolean = false,
+        cancelBuilds: MutableList<String>? = null
     ): Boolean {
         logger.info("[$buildId]|SHUTDOWN_BUILD|userId=$userId|status=$buildStatus|terminateFlag=$terminateFlag")
         // 记录该构建取消人信息
@@ -684,17 +685,22 @@ class PipelineRuntimeService @Autowired constructor(
         )
         // 发送取消事件
         val actionType = if (terminateFlag) ActionType.TERMINATE else ActionType.END
-        // 发送取消事件
+        // 按照要求发送取消事件，广播事件可以直接发送
+        val cancelEvent = PipelineBuildCancelEvent(
+            source = javaClass.simpleName,
+            projectId = projectId,
+            pipelineId = pipelineId,
+            userId = userId,
+            buildId = buildId,
+            status = buildStatus,
+            actionType = actionType
+        )
+        if (cancelBuilds == null) {
+            pipelineEventDispatcher.dispatch(cancelEvent)
+        } else {
+            cancelBuilds.add(buildId)
+        }
         pipelineEventDispatcher.dispatch(
-            PipelineBuildCancelEvent(
-                source = javaClass.simpleName,
-                projectId = projectId,
-                pipelineId = pipelineId,
-                userId = userId,
-                buildId = buildId,
-                status = buildStatus,
-                actionType = actionType
-            ),
             PipelineBuildCancelBroadCastEvent(
                 source = "cancelBuild",
                 projectId = projectId,
@@ -2056,7 +2062,8 @@ class PipelineRuntimeService @Autowired constructor(
         buildId: String,
         userId: String,
         groupName: String,
-        detailUrl: String
+        detailUrl: String,
+        cancelBuilds: MutableList<String>? = null
     ) {
         val redisLock = BuildIdLock(redisOperation = redisOperation, buildId = buildId)
         try {
@@ -2100,7 +2107,8 @@ class PipelineRuntimeService @Autowired constructor(
                     buildId = buildId,
                     userId = userId,
                     executeCount = buildInfo?.executeCount ?: 1,
-                    buildStatus = BuildStatus.CANCELED
+                    buildStatus = BuildStatus.CANCELED,
+                    cancelBuilds = cancelBuilds
                 )
                 logger.info("Cancel the pipeline($pipelineId) of instance($buildId) by the user($userId)")
             } catch (t: Throwable) {

diff --git a/...biz-engine/src/main/kotlin/com/tencent/devops/process/engine/control/BuildStartControl.kt b/...biz-engine/src/main/kotlin/com/tencent/devops/process/engine/control/BuildStartControl.kt
@@ -86,6 +86,7 @@ import com.tencent.devops.process.engine.utils.ContainerUtils
 import com.tencent.devops.common.pipeline.pojo.setting.PipelineRunLockType
 import com.tencent.devops.common.pipeline.pojo.setting.PipelineSetting
 import com.tencent.devops.process.constant.ProcessMessageCode
+import com.tencent.devops.process.engine.pojo.event.PipelineBuildBatchCancelEvent
 import com.tencent.devops.process.service.BuildVariableService
 import com.tencent.devops.process.service.scm.ScmProxyService
 import com.tencent.devops.process.utils.BUILD_NO
@@ -339,14 +340,31 @@ class BuildStartControl @Autowired constructor(
                         stageId = null,
                         needShortUrl = false
                     )
+                    // 按照流水线ID组织批量取消的事件
+                    val pipelineToEvents = mutableMapOf<String, MutableList<String>>()
                     concurrencyGroupRunning.forEach { (pipelineId, buildId) ->
+                        val cancelBuilds = pipelineToEvents[pipelineId] ?: mutableListOf()
                         pipelineRuntimeService.concurrencyCancelBuildPipeline(
                             projectId = projectId,
                             pipelineId = pipelineId,
                             buildId = buildId,
                             userId = buildInfo.startUser,
                             groupName = concurrencyGroup,
-                            detailUrl = detailUrl
+                            detailUrl = detailUrl,
+                            cancelBuilds = cancelBuilds
+                        )
+                        pipelineToEvents[pipelineId] = cancelBuilds
+                    }
+                    pipelineToEvents.forEach { (pipelineId, cancelBuilds) ->
+                        pipelineEventDispatcher.dispatch(
+                            PipelineBuildBatchCancelEvent(
+                                source = "concurrencyGroupCancel",
+                                projectId = projectId,
+                                pipelineId = pipelineId,
+                                userId = userId,
+                                buildIds = cancelBuilds,
+                                actionType = ActionType.END
+                            )
                         )
                     }
                 }