[stream] 优化大仓触发耗时 #10861

src/backend/ci/core/stream/api-stream/src/main/kotlin/com/tencent/devops/stream/pojo/message/UserMessageType.kt

@@ -31,8 +31,14 @@ package com.tencent.devops.stream.pojo.message
  * 用户消息通知的类型
  */
 enum class UserMessageType {
-    // 有失败的消息组
+    // 有失败的消息组 最终状态
     REQUEST,
-    // 只有成功的消息组
-    ONLY_SUCCESS
+    // 只有成功的消息组 最终或中间状态
+    ONLY_SUCCESS;
+
+    companion object {
+        fun parse(message: String): UserMessageType {
+            return values().find { it.name == message } ?: REQUEST
+        }
+    }
 }

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/pojo/UserMessageData.kt

@@ -0,0 +1,36 @@
+/*
+ * Tencent is pleased to support the open source community by making BK-CI 蓝鲸持续集成平台 available.
+ *
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company.  All rights reserved.
+ *
+ * BK-CI 蓝鲸持续集成平台 is licensed under the MIT license.
+ *
+ * A copy of the MIT License is included in this file.
+ *
+ *
+ * Terms of the MIT License:
+ * ---------------------------------------------------
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
+ * the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
+ * LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+ * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+ * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package com.tencent.devops.stream.pojo
+
+data class UserMessageData(
+    val projectId: String,
+    val userId: String,
+    val messageId: String,
+    val messageType: String,
+    val messageTitle: String
+)

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/BaseManualTriggerService.kt

@@ -103,6 +103,7 @@ abstract class BaseManualTriggerService @Autowired constructor(
             triggerBuildReq = triggerBuildReq
         )
 
+        action.data.watcherStart("baseManualTriggerService.triggerBuild")
         val buildPipeline = gitPipelineResourceDao.getPipelineById(
             dslContext = dslContext,
             gitProjectId = action.data.getGitProjectId().toLong(),
@@ -250,7 +251,7 @@ abstract class BaseManualTriggerService @Autowired constructor(
         )
 
         var buildId: BuildId? = null
-        StreamTriggerExceptionHandlerUtil.handleManualTrigger {
+        StreamTriggerExceptionHandlerUtil.handleManualTrigger(action) {
             buildId = trigger(action, originYaml, triggerBuildReq)
         }
         return buildId
@@ -262,6 +263,7 @@ abstract class BaseManualTriggerService @Autowired constructor(
         originYaml: String,
         triggerBuildReq: TriggerBuildReq
     ): BuildId? {
+        action.data.watcherStart("baseManualTriggerService.trigger")
         val yamlReplaceResult = streamYamlTrigger.prepareCIBuildYaml(action)!!
         val parsedYaml = if (action.metaData.streamObjectKind.needInput()) {
             YamlCommonUtils.toYamlNotNull(

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/StreamTriggerRequestRepoService.kt

@@ -90,6 +90,7 @@ class StreamTriggerRequestRepoService @Autowired constructor(
             actionSetting = null
         )!!
 
+        action.data.watcherStart("streamTriggerRequestRepoService.repoTriggerBuild")
         action.data.context.repoTrigger = RepoTrigger("", triggerPipelineList)
 
         logger.info(
@@ -139,6 +140,7 @@ class StreamTriggerRequestRepoService @Autowired constructor(
     private fun triggerPerPipeline(
         action: BaseAction
     ): Boolean {
+        action.data.watcherStart("streamTriggerRequestRepoService.triggerPerPipeline")
         logger.info(
             "StreamTriggerRequestRepoService|triggerPerPipeline" +
                 "|requestEventId|${action.data.context.requestEventId}"

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/StreamTriggerRequestService.kt

@@ -134,6 +134,7 @@ class StreamTriggerRequestService @Autowired constructor(
             logger.warn("StreamTriggerRequestService|start|request event not support|$event")
             return false
         }
+        action.data.watcherStart("streamTriggerRequestService.start")
         val eventCommon = action.data.eventCommon
 
         // 初始化setting
@@ -205,6 +206,7 @@ class StreamTriggerRequestService @Autowired constructor(
     private fun checkRequest(
         action: BaseAction
     ): Boolean {
+        action.data.watcherStart("streamTriggerRequestService.checkRequest")
         logger.info(
             "StreamTriggerRequestService|checkRequest" +
                 "|requestEventId|${action.data.context.requestEventId}|action|${action.format()}"
@@ -257,6 +259,7 @@ class StreamTriggerRequestService @Autowired constructor(
         action: BaseAction,
         path2PipelineExists: Map<String, StreamTriggerPipeline>
     ): Boolean {
+        action.data.watcherStart("streamTriggerRequestService.matchAndTriggerPipeline")
         logger.info(
             "StreamTriggerRequestService|matchAndTriggerPipeline" +
                 "|requestEventId|${action.data.context.requestEventId}|action|${action.format()}"

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/StreamYamlBaseBuild.kt

@@ -341,6 +341,7 @@ class StreamYamlBaseBuild @Autowired constructor(
             gitRequestEventDao.updateChangeYamlList(dslContext, action.data.context.requestEventId!!, forkMrYamlList)
         }
 
+        action.data.watcherStart("streamYamlBaseBuild.startBuild.StreamBuildLock")
         // 修改流水线并启动构建，需要加锁保证事务性
         val buildLock = StreamBuildLock(
             redisOperation = redisOperation,
@@ -471,6 +472,7 @@ class StreamYamlBaseBuild @Autowired constructor(
         gitBuildId: Long,
         yamlTransferData: YamlTransferData?
     ) {
+        action.data.watcherStart("streamYamlBaseBuild.afterStartBuild")
         try {
             val event = gitRequestEventDao.getWithEvent(
                 dslContext = dslContext, id = action.data.context.requestEventId!!
@@ -480,8 +482,7 @@ class StreamYamlBaseBuild @Autowired constructor(
                 projectCode = action.getProjectCode(),
                 event = event,
                 gitProjectId = action.data.getGitProjectId().toLong(),
-                messageType = UserMessageType.ONLY_SUCCESS,
-                isSave = true
+                messageType = UserMessageType.ONLY_SUCCESS
             )
 
             if (action is StreamRepoTriggerAction) {

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/StreamYamlBuild.kt

@@ -155,6 +155,7 @@ class StreamYamlBuild @Autowired constructor(
         yamlTransferData: YamlTransferData?,
         manualInputs: Map<String, String>?
     ): BuildId? {
+        action.data.watcherStart("streamYamlBuild.gitStartBuild")
         logger.info(
             "StreamYamlBuild|gitStartBuild" +
                 "|eventId|${action.data.context.requestEventId}|action|${action.format()}"
@@ -325,6 +326,7 @@ class StreamYamlBuild @Autowired constructor(
         yamlTransferData: YamlTransferData?,
         manualInputs: Map<String, String>?
     ): BuildId? {
+        action.data.watcherStart("streamYamlBuild.startBuildPipeline")
         logger.info(
             "StreamYamlBuild|startBuildPipeline" +
                 "|requestEventId|${action.data.context.requestEventId}|action|${action.format()}"
@@ -404,6 +406,7 @@ class StreamYamlBuild @Autowired constructor(
 
         // 判断是否更新最后修改人
         val updateLastModifyUser = action.needUpdateLastModifyUser(pipeline.filePath)
+        action.data.watcherStart("streamYamlBuild.savePipeline.StreamBuildLock")
         StreamBuildLock(
             redisOperation = redisOperation,
             gitProjectId = action.data.getGitProjectId().toLong(),

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/StreamYamlTrigger.kt

@@ -111,9 +111,9 @@ class StreamYamlTrigger @Autowired constructor(
         action: BaseAction,
         trigger: String?
     ) {
-        logger.info("|${action.data.context.requestEventId}|checkAndTrigger|action|${action.format()}")
         val buildPipeline = action.data.context.pipeline!!
-
+        action.data.watcherStart("|${buildPipeline.pipelineId}|streamYamlTrigger.checkAndTrigger")
+        logger.info("|${action.data.context.requestEventId}|checkAndTrigger|action|${action.format()}")
         val filePath = buildPipeline.filePath
         // 流水线未启用则跳过
         if (!buildPipeline.enabled) {
@@ -177,6 +177,7 @@ class StreamYamlTrigger @Autowired constructor(
         yamlSchemaCheck.check(action = action, templateType = null, isCiFile = true)
 
         // 进入触发流程
+        action.data.watcherStart("streamYamlTrigger.trigger")
         trigger(action, triggerEvent)
     }
 
@@ -193,6 +194,7 @@ class StreamYamlTrigger @Autowired constructor(
         action: BaseAction,
         triggerEvent: Pair<List<Any>?, TriggerResult>?
     ): Boolean {
+        action.data.watcherStart("streamYamlTrigger.triggerBuild")
         logger.info(
             "StreamYamlTrigger|triggerBuild|requestEventId" +
                 "|${action.data.context.requestEventId}|action|${action.format()}"

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/actions/EventActionFactory.kt

@@ -30,8 +30,10 @@ package com.tencent.devops.stream.trigger.actions
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.kotlin.readValue
 import com.tencent.devops.common.api.enums.ScmType
+import com.tencent.devops.common.api.util.Watcher
 import com.tencent.devops.common.client.Client
 import com.tencent.devops.common.redis.RedisOperation
+import com.tencent.devops.common.service.trace.TraceTag
 import com.tencent.devops.common.webhook.pojo.code.CodeWebhookEvent
 import com.tencent.devops.common.webhook.pojo.code.git.GitEvent
 import com.tencent.devops.common.webhook.pojo.code.git.GitIssueEvent
@@ -79,6 +81,7 @@ import com.tencent.devops.stream.trigger.service.StreamTriggerTokenService
 import com.tencent.devops.stream.trigger.timer.service.StreamTimerService
 import org.jooq.DSLContext
 import org.slf4j.LoggerFactory
+import org.slf4j.MDC
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 
@@ -111,7 +114,6 @@ class EventActionFactory @Autowired constructor(
 
     fun load(event: CodeWebhookEvent): BaseAction? {
         val action = loadEvent(event) ?: return null
-
         return action.init()
     }
 
@@ -133,6 +135,7 @@ class EventActionFactory @Autowired constructor(
                     return null
                 }
             }
+
             ScmType.GITHUB -> {
                 when (actionCommonData.eventType) {
                     GithubPushEvent.classType -> objectMapper.readValue<GithubPushEvent>(eventStr)
@@ -143,6 +146,7 @@ class EventActionFactory @Autowired constructor(
                     }
                 }
             }
+
             else -> TODO("对接其他Git平台时需要补充")
         }
 
@@ -177,6 +181,7 @@ class EventActionFactory @Autowired constructor(
         ScmType.CODE_GIT -> {
             objectMapper.readValue<GitEvent>(event)
         }
+
         ScmType.GITHUB -> {
             when (objectKind) {
                 StreamObjectKind.PULL_REQUEST.value -> objectMapper.readValue<GithubPullRequestEvent>(event)
@@ -185,6 +190,7 @@ class EventActionFactory @Autowired constructor(
                 else -> throw IllegalArgumentException("$objectKind in github load action not support yet")
             }
         }
+
         else -> TODO("对接其他Git平台时需要补充")
     }
 
@@ -206,6 +212,7 @@ class EventActionFactory @Autowired constructor(
                 )
                 tGitPushAction
             }
+
             is GitMergeRequestEvent -> {
                 val tGitMrAction = TGitMrActionGit(
                     dslContext = dslContext,
@@ -218,13 +225,15 @@ class EventActionFactory @Autowired constructor(
                 )
                 tGitMrAction
             }
+
             is GitTagPushEvent -> {
                 val tGitTagPushAction = TGitTagPushActionGit(
                     apiService = tGitApiService,
                     gitCheckService = gitCheckService
                 )
                 tGitTagPushAction
             }
+
             is GitIssueEvent -> {
                 val tGitIssueAction = TGitIssueActionGit(
                     dslContext = dslContext,
@@ -234,6 +243,7 @@ class EventActionFactory @Autowired constructor(
                 )
                 tGitIssueAction
             }
+
             is GitReviewEvent -> {
                 val tGitReviewAction = TGitReviewActionGit(
                     dslContext = dslContext,
@@ -243,6 +253,7 @@ class EventActionFactory @Autowired constructor(
                 )
                 tGitReviewAction
             }
+
             is GitNoteEvent -> {
                 val tGitNoteAction = TGitNoteActionGit(
                     dslContext = dslContext,
@@ -252,6 +263,7 @@ class EventActionFactory @Autowired constructor(
                 )
                 tGitNoteAction
             }
+
             is GithubPushEvent -> {
                 when {
                     event.ref.startsWith("refs/heads/") -> GithubPushActionGit(
@@ -266,13 +278,16 @@ class EventActionFactory @Autowired constructor(
                         pipelineDelete = pipelineDelete,
                         gitCheckService = gitCheckService
                     )
+
                     event.ref.startsWith("refs/tags/") -> GithubTagPushActionGit(
                         apiService = githubApiService,
                         gitCheckService = gitCheckService
                     )
+
                     else -> return null
                 }
             }
+
             is GithubPullRequestEvent -> {
                 GithubPRActionGit(
                     apiService = githubApiService,
@@ -284,6 +299,7 @@ class EventActionFactory @Autowired constructor(
                     dslContext = dslContext
                 )
             }
+
             else -> {
                 return null
             }
@@ -295,9 +311,11 @@ class EventActionFactory @Autowired constructor(
             gitAction.isStreamDeleteAction() -> {
                 StreamDeleteAction(gitAction)
             }
+
             else -> gitAction
         }
 
+        action.data.watcher = Watcher("stream_action_watcher|${MDC.get(TraceTag.BIZID)}")
         return action
     }
 

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/actions/data/ActionData.kt

@@ -27,6 +27,7 @@
 
 package com.tencent.devops.stream.trigger.actions.data
 
+import com.tencent.devops.common.api.util.Watcher
 import com.tencent.devops.common.webhook.pojo.code.CodeWebhookEvent
 import com.tencent.devops.stream.trigger.actions.data.context.StreamTriggerContext
 
@@ -42,10 +43,18 @@ data class ActionData(
 ) {
     // 需要根据各事件源的event去拿的通用数据，随event改变可能会不同
     lateinit var eventCommon: EventCommonData
+    lateinit var watcher: Watcher
 
     // Stream触发时需要的配置信息
     lateinit var setting: StreamTriggerSetting
     val isSettingInitialized get() = this::setting.isInitialized
+    val isWatcherInitialized get() = this::watcher.isInitialized
+
+    fun watcherStart(id: String) {
+        if (isWatcherInitialized) {
+            watcher.start(id)
+        }
+    }
 
     // 方便日志打印
     fun format() = "${event::class.qualifiedName}|$context|$eventCommon|$setting"

src/backend/ci/core/stream/biz-stream/src/main/kotlin/com/tencent/devops/stream/trigger/exception/handler/StreamTriggerExceptionHandler.kt

@@ -28,6 +28,7 @@
 package com.tencent.devops.stream.trigger.exception.handler
 
 import com.tencent.devops.common.api.exception.ErrorCodeException
+import com.tencent.devops.common.service.utils.LogUtils
 import com.tencent.devops.process.yaml.v2.enums.StreamObjectKind
 import com.tencent.devops.stream.pojo.enums.TriggerReason
 import com.tencent.devops.stream.trigger.actions.BaseAction
@@ -84,6 +85,11 @@ class StreamTriggerExceptionHandler @Autowired constructor(
                 logger.error("BKSystemErrorMonitor|StreamTriggerExceptionHandler|action|${action.format()}", e)
                 return null
             }
+        } finally {
+            if (action.data.isWatcherInitialized) {
+                action.data.watcher.stop()
+                LogUtils.printCostTimeWE(action.data.watcher, warnThreshold = 1000, errorThreshold = 5000)
+            }
         }
     }