feat: dispatch-sdk新增调度状态功能 #12388

src/backend/ci/core/common/common-dispatch-sdk/src/main/kotlin/com/tencent/devops/common/dispatch.sdk/config/SDKConfiguration.kt

@@ -29,6 +29,7 @@ package com.tencent.devops.common.dispatch.sdk.config
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.tencent.devops.common.client.Client
+import com.tencent.devops.common.dispatch.sdk.service.DispatchMessageTracking
 import com.tencent.devops.common.dispatch.sdk.service.DispatchService
 import com.tencent.devops.common.dispatch.sdk.service.DockerRoutingSdkService
 import com.tencent.devops.common.dispatch.sdk.service.JobQuotaService
@@ -74,6 +75,12 @@ class SDKConfiguration {
     ) =
         JobQuotaService(client, buildLogPrinter)
 
+    @Bean
+    fun dispatchMessageTracking(
+        @Autowired client: Client
+    ) =
+        DispatchMessageTracking(client)
+
     @Bean
     fun dockerRoutingSdkService(
         @Autowired redisOperation: RedisOperation

src/backend/ci/core/common/common-dispatch-sdk/src/main/kotlin/com/tencent/devops/common/dispatch.sdk/listener/BuildListener.kt

@@ -32,6 +32,7 @@ import com.tencent.devops.common.api.pojo.ErrorType
 import com.tencent.devops.common.client.Client
 import com.tencent.devops.common.dispatch.sdk.BuildFailureException
 import com.tencent.devops.common.dispatch.sdk.DispatchSdkErrorCode
+import com.tencent.devops.common.dispatch.sdk.service.DispatchMessageTracking
 import com.tencent.devops.common.dispatch.sdk.pojo.DispatchMessage
 import com.tencent.devops.common.dispatch.sdk.service.DispatchService
 import com.tencent.devops.common.dispatch.sdk.service.JobQuotaService
@@ -43,6 +44,7 @@ import com.tencent.devops.common.pipeline.type.DispatchType
 import com.tencent.devops.common.service.prometheus.BkTimed
 import com.tencent.devops.common.service.utils.SpringContextUtil
 import com.tencent.devops.common.web.utils.I18nUtil
+import com.tencent.devops.dispatch.pojo.dto.InitMessageTrackingRequest
 import com.tencent.devops.dispatch.pojo.enums.JobQuotaVmType
 import com.tencent.devops.notify.api.service.ServiceNotifyResource
 import com.tencent.devops.notify.pojo.EmailNotifyMessage
@@ -79,19 +81,31 @@ interface BuildListener {
         }
         DispatcherContext.setEvent(event)
         val dispatchService = getDispatchService()
+        val dispatchMessageTracking = getDispatchMessageTracking()
 
         var startTime = 0L
         var errorCode = 0
         var errorMessage = ""
         var errorType: ErrorType? = null
 
+        // 初始化消息追踪
+        dispatchMessageTracking.initMessageTracking(
+            InitMessageTrackingRequest(
+                projectId = event.projectId,
+                pipelineId = event.pipelineId,
+                buildId = event.buildId,
+                vmSeqId = event.vmSeqId.toInt(),
+                executeCount = event.executeCount ?: 1,
+                dispatchType = event.dispatchType::class.java.simpleName
+            )
+        )
+
         try {
             logger.info("Start to handle the startup message -(${DispatcherContext.getEvent()})")
-
             startTime = System.currentTimeMillis()
             DispatchLogRedisUtils.setRedisExecuteCount(event.buildId, event.executeCount)
 
-            // 校验流水线是否在运行中，且处在构建机未启动状态
+            // 校验流水线是否已结束运行，且处在构建机未启动状态
             if (!dispatchService.checkRunning(event)) {
                 if (event.retryTime > 1) {
                     // 重试的请求如果流水线已结束，主动把配额记录删除
@@ -115,11 +129,28 @@ interface BuildListener {
                     demoteQueueRouteKeySuffix = getStartupDemoteQueue()
                 )
             ) {
+                // 追踪配额不足
+                dispatchMessageTracking.trackQuotaQueue(
+                    buildId = event.buildId,
+                    vmSeqId = event.vmSeqId,
+                    executeCount = event.executeCount ?: 1,
+                    operator = "JobQuotaService.checkAndAddRunningJob"
+                )
+
                 return
             }
 
             onStartup(dispatchService.buildDispatchMessage(event))
+
+            // 追踪处理成功
+            dispatchMessageTracking.trackConsumeSuccess(
+                buildId = event.buildId,
+                vmSeqId = event.vmSeqId,
+                executeCount = event.executeCount ?: 1,
+                operator = "${this::class.java.simpleName}.handleStartup"
+            )
         } catch (e: BuildFailureException) {
+            // BuildFailureException是交付中异常，不在此处上报状态
             dispatchService.logRed(
                 buildId = event.buildId,
                 containerHashId = event.containerHashId,
@@ -149,6 +180,17 @@ interface BuildListener {
             errorMessage = "Fail to handle the start up message"
             errorType = ErrorType.SYSTEM
 
+            // 追踪处理失败
+            getDispatchMessageTracking().trackConsumeFailed(
+                buildId = event.buildId,
+                vmSeqId = event.vmSeqId,
+                executeCount = event.executeCount ?: 1,
+                errorCode = errorCode.toString(),
+                errorMessage = t.message ?: "Unknown error",
+                errorType = "SYSTEM",
+                operator = "${this::class.java.simpleName}.handleStartup"
+            )
+
             dispatchService.onFailure(
                 event = event,
                 e = BuildFailureException(
@@ -360,6 +402,10 @@ interface BuildListener {
         return SpringContextUtil.getBean(JobQuotaService::class.java)
     }
 
+    private fun getDispatchMessageTracking(): DispatchMessageTracking {
+        return SpringContextUtil.getBean(DispatchMessageTracking::class.java)
+    }
+
     private fun getClient() = SpringContextUtil.getBean(Client::class.java)
 
     companion object {

src/backend/ci/core/common/common-dispatch-sdk/src/main/kotlin/com/tencent/devops/common/dispatch.sdk/service/DispatchMessageTracking.kt

@@ -0,0 +1,246 @@
+/*
+ * Tencent is pleased to support the open source community by making BK-CI 蓝鲸持续集成平台 available.
+ *
+ * Copyright (C) 2019 Tencent.  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.common.dispatch.sdk.service
+
+import com.tencent.devops.common.client.Client
+import com.tencent.devops.dispatch.api.ServiceDispatchMessageTrackingResource
+import com.tencent.devops.dispatch.pojo.DispatchMessageStatus
+import com.tencent.devops.dispatch.pojo.dto.InitMessageTrackingRequest
+import com.tencent.devops.dispatch.pojo.dto.UpdateMessageStatusRequest
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.stereotype.Component
+
+/**
+ * Dispatch 消息追踪实现
+ */
+@Component
+class DispatchMessageTracking @Autowired constructor(
+    private val client: Client
+) {
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(DispatchMessageTracking::class.java)
+    }
+
+    fun initMessageTracking(
+        initMessageTrackingRequest: InitMessageTrackingRequest
+    ) {
+        try {
+            client.get(ServiceDispatchMessageTrackingResource::class)
+                .initMessageTracking(initMessageTrackingRequest)
+        } catch (e: Exception) {
+            logger.warn(
+                "[${initMessageTrackingRequest.buildId}|${initMessageTrackingRequest.vmSeqId}] " +
+                        "Failed to init message tracking",
+                e
+            )
+        }
+    }
+
+    fun updateMessageStatus(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        newStatus: DispatchMessageStatus,
+        statusMsg: String?,
+        errorCode: String? = "",
+        errorMessage: String? = "",
+        errorType: String? = "",
+        operator: String? = "",
+        remark: String? = ""
+    ): Boolean {
+        return try {
+            val request = UpdateMessageStatusRequest(
+                buildId = buildId,
+                vmSeqId = vmSeqId.toInt(),
+                executeCount = executeCount,
+                newStatus = newStatus.status,
+                statusMsg = statusMsg,
+                errorCode = errorCode,
+                errorMessage = errorMessage,
+                errorType = errorType,
+                operator = operator,
+                remark = remark
+            )
+            client.get(ServiceDispatchMessageTrackingResource::class).updateMessageStatus(request).data ?: false
+        } catch (e: Exception) {
+            logger.warn("[$buildId|$vmSeqId] Failed to update message status", e)
+            false
+        }
+    }
+
+    fun trackConsumeFailed(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        errorCode: String,
+        errorMessage: String,
+        errorType: String,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.CONSUME_FAILED,
+        statusMsg = DispatchMessageStatus.CONSUME_SUCCESS.description,
+        errorCode = errorCode,
+        errorMessage = errorMessage,
+        errorType = errorType,
+        operator = operator,
+        remark = "Message consumption failed: $errorMessage"
+    )
+
+    fun trackResourceQueue(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.RESOURCE_QUEUE,
+        statusMsg = DispatchMessageStatus.RESOURCE_QUEUE.description,
+        operator = operator,
+        remark = "Queuing for resources"
+    )
+
+    fun trackQuotaQueue(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.QUOTA_QUEUE,
+        statusMsg = DispatchMessageStatus.QUOTA_QUEUE.description,
+        operator = operator
+    )
+
+    fun trackQuotaInSufficient(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.QUOTA_INSUFFICIENT,
+        statusMsg = DispatchMessageStatus.QUOTA_INSUFFICIENT.description,
+        operator = operator
+    )
+
+    fun trackResourcePreparing(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.RESOURCE_PREPARING,
+        statusMsg = DispatchMessageStatus.RESOURCE_PREPARING.description,
+        operator = operator
+    )
+
+    fun trackResourcePreparingFailed(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.RESOURCE_PREPARE_FAILED,
+        statusMsg = DispatchMessageStatus.RESOURCE_PREPARE_FAILED.description,
+        operator = operator
+    )
+
+    fun trackResourceDelivering(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.RESOURCE_DELIVERING,
+        statusMsg = DispatchMessageStatus.RESOURCE_DELIVERING.description,
+        operator = operator
+    )
+
+    fun trackResourceDeliveringFailed(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.RESOURCE_DELIVERY_FAILED,
+        statusMsg = DispatchMessageStatus.RESOURCE_DELIVERY_FAILED.description,
+        operator = operator
+    )
+
+    fun trackResourceReady(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.RESOURCE_READY,
+        statusMsg = DispatchMessageStatus.RESOURCE_READY.description,
+        operator = operator
+    )
+
+    fun trackConsumeSuccess(
+        buildId: String,
+        vmSeqId: String,
+        executeCount: Int,
+        operator: String? = ""
+    ) = updateMessageStatus(
+        buildId = buildId,
+        vmSeqId = vmSeqId,
+        executeCount = executeCount,
+        newStatus = DispatchMessageStatus.CONSUME_SUCCESS,
+        statusMsg = DispatchMessageStatus.CONSUME_SUCCESS.description,
+        operator = operator,
+        remark = "Message consumption completed successfully"
+    )
+}