Merge branch 'master' of https://github.com/Tencent/bk-ci into featur…

…e_remoteEnv
terlinhe · Jul 16, 2021 · dfd1ce7 · dfd1ce7
2 parents c18e37f + 058d4c3
commit dfd1ce7
Show file tree

Hide file tree

Showing 33 changed files with 535 additions and 376 deletions.
diff --git a/README.md b/README.md
@@ -44,7 +44,7 @@ bk-ci提供了流水线、代码库、凭证管理、环境管理、研发商店
 
 ## Support
 1. [GitHub讨论区](https://github.com/Tencent/bk-ci/discussions)
-2. QQ群：744672165
+2. QQ群：495299374
 
 ## BlueKing Community
 - [BK-BCS](https://github.com/Tencent/bk-bcs)：蓝鲸容器管理平台是以容器技术为基础，为微服务业务提供编排管理的基础服务平台。

diff --git a/README_EN.md b/README_EN.md
@@ -49,7 +49,7 @@ bk-ci provides five core services, namely Process, Repository, Ticket, Environme
 
 ## Support
 1. [GitHub Discussions](https://github.com/Tencent/bk-ci/discussions)
-2. QQ Group: 744672165
+2. QQ Group: 495299374
 
 ## BlueKing Community
 

diff --git a/src/backend/ci/build.gradle.kts b/src/backend/ci/build.gradle.kts
@@ -1,9 +1,8 @@
 plugins {
     kotlin("jvm")
+    detektCheck
 }
 
-apply("$rootDir/detekt.gradle.kts")
-
 allprojects {
     // 包路径
     group = "com.tencent.bk.devops.ci"

diff --git a/src/backend/ci/detekt.gradle.kts → ...rc/src/main/kotlin/detektCheck.gradle.kts b/src/backend/ci/detekt.gradle.kts → ...rc/src/main/kotlin/detektCheck.gradle.kts
diff --git a/...log/biz-log/src/main/kotlin/com/tencent/devops/log/resources/BuildLogPrintResourceImpl.kt b/...log/biz-log/src/main/kotlin/com/tencent/devops/log/resources/BuildLogPrintResourceImpl.kt
@@ -56,37 +56,41 @@ class BuildLogPrintResourceImpl @Autowired constructor(
             logger.error("Invalid build ID[$buildId]")
             return Result(false)
         }
-        return buildLogPrintService.asyncDispatchEvent(LogEvent(buildId, listOf(logMessage)))
+        buildLogPrintService.dispatchEvent(LogEvent(buildId, listOf(logMessage)))
+        return Result(true)
     }
 
     override fun addRedLogLine(buildId: String, logMessage: LogMessage): Result<Boolean> {
         if (buildId.isBlank()) {
             logger.error("Invalid build ID[$buildId]")
             return Result(false)
         }
-        return buildLogPrintService.asyncDispatchEvent(LogEvent(
+        buildLogPrintService.dispatchEvent(LogEvent(
             buildId = buildId,
             logs = listOf(logMessage.copy(message = Ansi().bold().fgRed().a(logMessage.message).reset().toString()))
         ))
+        return Result(true)
     }
 
     override fun addYellowLogLine(buildId: String, logMessage: LogMessage): Result<Boolean> {
         if (buildId.isBlank()) {
             logger.error("Invalid build ID[$buildId]")
             return Result(false)
         }
-        return buildLogPrintService.asyncDispatchEvent(LogEvent(
+        buildLogPrintService.dispatchEvent(LogEvent(
             buildId = buildId,
             logs = listOf(logMessage.copy(message = Ansi().bold().fgYellow().a(logMessage.message).reset().toString()))
         ))
+        return Result(true)
     }
 
     override fun addLogMultiLine(buildId: String, logMessages: List<LogMessage>): Result<Boolean> {
         if (buildId.isBlank()) {
             logger.error("Invalid build ID[$buildId]")
             return Result(false)
         }
-        return buildLogPrintService.asyncDispatchEvent(LogEvent(buildId, logMessages))
+        buildLogPrintService.dispatchEvent(LogEvent(buildId, logMessages))
+        return Result(true)
     }
 
     override fun addLogStatus(

diff --git a/.../src/main/kotlin/com/tencent/devops/misc/cron/process/PipelineBuildHistoryDataClearJob.kt b/.../src/main/kotlin/com/tencent/devops/misc/cron/process/PipelineBuildHistoryDataClearJob.kt
@@ -50,6 +50,9 @@ import java.time.LocalDateTime
 import java.util.concurrent.Callable
 import java.util.concurrent.Executors
 import java.util.concurrent.Future
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
 
 @Component
 @Suppress("ALL")
@@ -73,22 +76,38 @@ class PipelineBuildHistoryDataClearJob @Autowired constructor(
             "pipeline:build:history:data:clear:project:id"
         private const val PIPELINE_BUILD_HISTORY_DATA_CLEAR_PROJECT_LIST_KEY =
             "pipeline:build:history:data:clear:project:list"
+        private const val PIPELINE_BUILD_HISTORY_DATA_CLEAR_THREAD_SET_KEY =
+            "pipeline:build:history:data:clear:thread:set"
+        private var executor: ThreadPoolExecutor? = null
     }
 
     @Value("\${process.deletedPipelineStoreDays:30}")
     private val deletedPipelineStoreDays: Long = 30 // 回收站已删除流水线保存天数
 
-    private val executor = Executors.newFixedThreadPool(miscBuildDataClearConfig.maxThreadHandleProjectNum)
-
     @Scheduled(initialDelay = 10000, fixedDelay = 12000)
     fun pipelineBuildHistoryDataClear() {
         if (!miscBuildDataClearConfig.switch.toBoolean()) {
             // 如果清理构建历史数据开关关闭，则不清理
             return
         }
         logger.info("pipelineBuildHistoryDataClear start")
-        val lock = RedisLock(redisOperation,
-            LOCK_KEY, 3000)
+        if (executor == null) {
+            // 创建带有边界队列的线程池，防止内存爆掉
+            logger.info("pipelineBuildHistoryDataClear create executor")
+            executor = ThreadPoolExecutor(
+                miscBuildDataClearConfig.maxThreadHandleProjectNum,
+                miscBuildDataClearConfig.maxThreadHandleProjectNum,
+                0L,
+                TimeUnit.MILLISECONDS,
+                LinkedBlockingQueue(10),
+                Executors.defaultThreadFactory(),
+                ThreadPoolExecutor.DiscardPolicy()
+            )
+        }
+        val lock = RedisLock(
+            redisOperation,
+            LOCK_KEY, 3000
+        )
         try {
             if (!lock.tryLock()) {
                 logger.info("get lock failed, skip")
@@ -108,7 +127,6 @@ class PipelineBuildHistoryDataClearJob @Autowired constructor(
             }
             // 获取清理项目构建数据的线程数量
             val maxThreadHandleProjectNum = miscBuildDataClearConfig.maxThreadHandleProjectNum
-            val futureList = mutableListOf<Future<Boolean>>()
             val avgProjectNum = maxProjectNum / maxThreadHandleProjectNum
             for (index in 1..maxThreadHandleProjectNum) {
                 // 计算线程能处理的最大项目主键ID
@@ -117,17 +135,15 @@ class PipelineBuildHistoryDataClearJob @Autowired constructor(
                 } else {
                     index * avgProjectNum + maxProjectNum % maxThreadHandleProjectNum
                 }
-                futureList.add(
+                // 判断线程是否正在处理任务，如正在处理则不分配新任务(定时任务12秒执行一次，线程启动到往set集合设置编号耗费时间很短，故不加锁)
+                if (!redisOperation.isMember(PIPELINE_BUILD_HISTORY_DATA_CLEAR_THREAD_SET_KEY, index.toString())) {
                     doClearBus(
                         threadNo = index,
                         projectIdList = projectIdList,
                         minThreadProjectPrimaryId = (index - 1) * avgProjectNum,
                         maxThreadProjectPrimaryId = maxThreadProjectPrimaryId
                     )
-                )
-            }
-            futureList.forEachIndexed { index, future ->
-                logger.info("future-$index doClearBus result:${future.get()}")
+                }
             }
         } catch (t: Throwable) {
             logger.warn("pipelineBuildHistoryDataClear failed", t)
@@ -143,7 +159,7 @@ class PipelineBuildHistoryDataClearJob @Autowired constructor(
         maxThreadProjectPrimaryId: Long
     ): Future<Boolean> {
         val threadName = "Thread-$threadNo"
-        return executor.submit(Callable<Boolean> {
+        return executor!!.submit(Callable<Boolean> {
             var handleProjectPrimaryId =
                 redisOperation.get("$threadName:$PIPELINE_BUILD_HISTORY_DATA_CLEAR_PROJECT_ID_KEY")?.toLong()
             if (handleProjectPrimaryId == null) {
@@ -152,43 +168,52 @@ class PipelineBuildHistoryDataClearJob @Autowired constructor(
                 if (handleProjectPrimaryId >= maxThreadProjectPrimaryId) {
                     // 已经清理完全部项目的流水线的过期构建记录，再重新开始清理
                     redisOperation.delete("$threadName:$PIPELINE_BUILD_HISTORY_DATA_CLEAR_PROJECT_ID_KEY")
-                    logger.info("pipelineBuildHistoryDataClear reStart")
+                    logger.info("pipelineBuildHistoryDataClear $threadName reStart")
                     return@Callable true
                 }
             }
-            val maxEveryProjectHandleNum = miscBuildDataClearConfig.maxEveryProjectHandleNum
-            var maxHandleProjectPrimaryId = handleProjectPrimaryId ?: 0L
-            val projectInfoList = if (projectIdList.isNullOrEmpty()) {
-                val channelCodeList = miscBuildDataClearConfig.clearChannelCodes.split(",")
-                maxHandleProjectPrimaryId = handleProjectPrimaryId + maxEveryProjectHandleNum
-                projectMiscService.getProjectInfoList(
-                    minId = handleProjectPrimaryId,
-                    maxId = maxHandleProjectPrimaryId,
-                    channelCodeList = channelCodeList
-                )
-            } else {
-                projectMiscService.getProjectInfoList(projectIdList = projectIdList)
-            }
-            // 根据项目依次查询T_PIPELINE_INFO表中的流水线数据处理
-            projectInfoList?.forEach { projectInfo ->
-                val channel = projectInfo.channel
-                // 获取项目对应的流水线数据清理配置类，如果不存在说明无需清理该项目下的构建记录
-                val projectDataClearConfigService =
-                    ProjectDataClearConfigFactory.getProjectDataClearConfigService(channel) ?: return@forEach
-                val projectPrimaryId = projectInfo.id
-                if (projectPrimaryId > maxHandleProjectPrimaryId) {
-                    maxHandleProjectPrimaryId = projectPrimaryId
+            // 将线程编号存入redis集合
+            redisOperation.sadd(PIPELINE_BUILD_HISTORY_DATA_CLEAR_THREAD_SET_KEY, threadNo.toString())
+            try {
+                val maxEveryProjectHandleNum = miscBuildDataClearConfig.maxEveryProjectHandleNum
+                var maxHandleProjectPrimaryId = handleProjectPrimaryId ?: 0L
+                val projectInfoList = if (projectIdList.isNullOrEmpty()) {
+                    val channelCodeList = miscBuildDataClearConfig.clearChannelCodes.split(",")
+                    maxHandleProjectPrimaryId = handleProjectPrimaryId + maxEveryProjectHandleNum
+                    projectMiscService.getProjectInfoList(
+                        minId = handleProjectPrimaryId,
+                        maxId = maxHandleProjectPrimaryId,
+                        channelCodeList = channelCodeList
+                    )
+                } else {
+                    projectMiscService.getProjectInfoList(projectIdList = projectIdList)
                 }
-                val projectId = projectInfo.projectId
-                // 清理流水线构建数据
-                clearPipelineBuildData(projectId, projectDataClearConfigService)
+                // 根据项目依次查询T_PIPELINE_INFO表中的流水线数据处理
+                projectInfoList?.forEach { projectInfo ->
+                    val channel = projectInfo.channel
+                    // 获取项目对应的流水线数据清理配置类，如果不存在说明无需清理该项目下的构建记录
+                    val projectDataClearConfigService =
+                        ProjectDataClearConfigFactory.getProjectDataClearConfigService(channel) ?: return@forEach
+                    val projectPrimaryId = projectInfo.id
+                    if (projectPrimaryId > maxHandleProjectPrimaryId) {
+                        maxHandleProjectPrimaryId = projectPrimaryId
+                    }
+                    val projectId = projectInfo.projectId
+                    // 清理流水线构建数据
+                    clearPipelineBuildData(projectId, projectDataClearConfigService)
+                }
+                // 将当前已处理完的最大项目Id存入redis
+                redisOperation.set(
+                    key = "$threadName:$PIPELINE_BUILD_HISTORY_DATA_CLEAR_PROJECT_ID_KEY",
+                    value = maxHandleProjectPrimaryId.toString(),
+                    expired = false
+                )
+            } catch (ignore: Exception) {
+                logger.warn("pipelineBuildHistoryDataClear doClearBus failed", ignore)
+            } finally {
+                // 释放redis集合中的线程编号
+                redisOperation.sremove(PIPELINE_BUILD_HISTORY_DATA_CLEAR_THREAD_SET_KEY, threadNo.toString())
             }
-            // 将当前已处理完的最大项目Id存入redis
-            redisOperation.set(
-                key = "$threadName:$PIPELINE_BUILD_HISTORY_DATA_CLEAR_PROJECT_ID_KEY",
-                value = maxHandleProjectPrimaryId.toString(),
-                expired = false
-            )
             return@Callable true
         })
     }

diff --git a/...i-openapi/src/main/kotlin/com/tencent/devops/openapi/api/apigw/v3/ApigwBuildResourceV3.kt b/...i-openapi/src/main/kotlin/com/tencent/devops/openapi/api/apigw/v3/ApigwBuildResourceV3.kt
@@ -30,13 +30,16 @@ import com.tencent.devops.common.api.auth.AUTH_HEADER_DEVOPS_APP_CODE
 import com.tencent.devops.common.api.auth.AUTH_HEADER_DEVOPS_APP_CODE_DEFAULT_VALUE
 import com.tencent.devops.common.api.auth.AUTH_HEADER_DEVOPS_USER_ID
 import com.tencent.devops.common.api.auth.AUTH_HEADER_DEVOPS_USER_ID_DEFAULT_VALUE
+import com.tencent.devops.common.api.auth.AUTH_HEADER_USER_ID
+import com.tencent.devops.common.api.auth.AUTH_HEADER_USER_ID_DEFAULT_VALUE
 import com.tencent.devops.common.api.pojo.BuildHistoryPage
 import com.tencent.devops.common.api.pojo.Result
 import com.tencent.devops.common.pipeline.pojo.StageReviewRequest
 import com.tencent.devops.process.pojo.BuildHistory
 import com.tencent.devops.process.pojo.BuildHistoryWithVars
 import com.tencent.devops.process.pojo.BuildId
 import com.tencent.devops.process.pojo.BuildManualStartupInfo
+import com.tencent.devops.process.pojo.BuildTaskPauseInfo
 import com.tencent.devops.process.pojo.pipeline.ModelDetail
 import io.swagger.annotations.Api
 import io.swagger.annotations.ApiOperation
@@ -294,4 +297,23 @@ interface ApigwBuildResourceV3 {
         @ApiParam("变量名列表", required = true)
         variableNames: List<String>
     ): Result<Map<String, String>>
+
+    @ApiOperation("操作暂停插件")
+    @POST
+    @Path("/{buildId}/execute/pause")
+    fun executionPauseAtom(
+        @ApiParam(value = "用户ID", required = true, defaultValue = AUTH_HEADER_USER_ID_DEFAULT_VALUE)
+        @HeaderParam(AUTH_HEADER_USER_ID)
+        userId: String,
+        @ApiParam("项目ID", required = true)
+        @PathParam("projectId")
+        projectId: String,
+        @ApiParam("流水线ID", required = true)
+        @PathParam("pipelineId")
+        pipelineId: String,
+        @ApiParam("构建ID", required = true)
+        @PathParam("buildId")
+        buildId: String,
+        taskPauseExecute: BuildTaskPauseInfo
+    ): Result<Boolean>
 }
diff --git a/...src/main/kotlin/com/tencent/devops/openapi/resources/apigw/v3/ApigwBuildResourceV3Impl.kt b/...src/main/kotlin/com/tencent/devops/openapi/resources/apigw/v3/ApigwBuildResourceV3Impl.kt
@@ -38,6 +38,7 @@ import com.tencent.devops.process.pojo.BuildHistory
 import com.tencent.devops.process.pojo.BuildHistoryWithVars
 import com.tencent.devops.process.pojo.BuildId
 import com.tencent.devops.process.pojo.BuildManualStartupInfo
+import com.tencent.devops.process.pojo.BuildTaskPauseInfo
 import com.tencent.devops.process.pojo.pipeline.ModelDetail
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
@@ -222,6 +223,23 @@ class ApigwBuildResourceV3Impl @Autowired constructor(
         )
     }
 
+    override fun executionPauseAtom(
+        userId: String,
+        projectId: String,
+        pipelineId: String,
+        buildId: String,
+        taskPauseExecute: BuildTaskPauseInfo
+    ): Result<Boolean> {
+        logger.info("$pipelineId| $buildId| $userId |executionPauseAtom $taskPauseExecute")
+        return client.get(ServiceBuildResource::class).executionPauseAtom(
+            userId = userId,
+            projectId = projectId,
+            pipelineId = pipelineId,
+            buildId = buildId,
+            taskPauseExecute = taskPauseExecute
+        )
+    }
+
     companion object {
         private val logger = LoggerFactory.getLogger(ApigwBuildResourceV3Impl::class.java)
     }

diff --git a/...pi-process/src/main/kotlin/com/tencent/devops/process/api/service/ServiceBuildResource.kt b/...pi-process/src/main/kotlin/com/tencent/devops/process/api/service/ServiceBuildResource.kt
@@ -42,6 +42,7 @@ import com.tencent.devops.process.pojo.BuildHistoryVariables
 import com.tencent.devops.process.pojo.BuildHistoryWithVars
 import com.tencent.devops.process.pojo.BuildId
 import com.tencent.devops.process.pojo.BuildManualStartupInfo
+import com.tencent.devops.process.pojo.BuildTaskPauseInfo
 import com.tencent.devops.process.pojo.ReviewParam
 import com.tencent.devops.process.pojo.VmInfo
 import com.tencent.devops.process.pojo.pipeline.ModelDetail
@@ -550,4 +551,23 @@ interface ServiceBuildResource {
         @ApiParam("审核请求体", required = false)
         reviewRequest: StageReviewRequest? = null
     ): Result<Boolean>
+
+    @ApiOperation("操作暂停插件")
+    @POST
+    @Path("/projects/{projectId}/pipelines/{pipelineId}/builds/{buildId}/execution/pause")
+    fun executionPauseAtom(
+        @ApiParam(value = "用户ID", required = true, defaultValue = AUTH_HEADER_USER_ID_DEFAULT_VALUE)
+        @HeaderParam(AUTH_HEADER_USER_ID)
+        userId: String,
+        @ApiParam("项目ID", required = true)
+        @PathParam("projectId")
+        projectId: String,
+        @ApiParam("流水线ID", required = true)
+        @PathParam("pipelineId")
+        pipelineId: String,
+        @ApiParam("构建ID", required = true)
+        @PathParam("buildId")
+        buildId: String,
+        taskPauseExecute: BuildTaskPauseInfo
+    ): Result<Boolean>
 }
diff --git a/...cess/api-process/src/main/kotlin/com/tencent/devops/process/api/user/UserBuildResource.kt b/...cess/api-process/src/main/kotlin/com/tencent/devops/process/api/user/UserBuildResource.kt
@@ -505,8 +505,8 @@ interface UserBuildResource {
         @ApiParam("任务ID", required = true)
         @PathParam("taskId")
         taskId: String,
-        @ApiParam("待执行插件元素", required = true)
-        element: Element,
+        @ApiParam("待执行插件元素", required = false)
+        element: Element?,
         @ApiParam("执行类型, true 继续, false 停止", required = true)
         @QueryParam("isContinue")
         isContinue: Boolean,