Merge pull request #5536 from sawyersong2/issue5525_github_branch

feat: 适配代码拉取调度优化 #5525

src/backend/ci/core/dispatch-docker/api-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/api/op/OPDispatchDockerResource.kt

@@ -34,6 +34,7 @@ import com.tencent.devops.dispatch.docker.pojo.DockerHostLoadConfig
 import com.tencent.devops.dispatch.docker.pojo.DockerIpInfoVO
 import com.tencent.devops.dispatch.docker.pojo.DockerIpListPage
 import com.tencent.devops.dispatch.docker.pojo.DockerIpUpdateVO
+import com.tencent.devops.dispatch.docker.pojo.HostDriftLoad
 import io.swagger.annotations.Api
 import io.swagger.annotations.ApiOperation
 import io.swagger.annotations.ApiParam
@@ -167,6 +168,6 @@ interface OPDispatchDockerResource {
         @HeaderParam(AUTH_HEADER_DEVOPS_USER_ID)
         userId: String,
         @ApiParam("阈值", required = true)
-        thresholdMap: Map<String, String>
+        hostDriftLoad: HostDriftLoad
     ): Result<Boolean>
 }

src/backend/ci/core/dispatch-docker/api-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/pojo/HostDriftLoad.kt

+/*
+ * 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.dispatch.docker.pojo
+
+data class HostDriftLoad(
+    val memory: Int,
+    val cpu: Int,
+    val disk: Int,
+    val diskIo: Int,
+    val usedNum: Int
+)

src/backend/ci/core/dispatch-docker/biz-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/client/DockerHostClient.kt

@@ -177,7 +177,7 @@ class DockerHostClient @Autowired constructor(
             containerHashId = dispatchMessage.containerHashId,
             customBuildEnv = dispatchMessage.customBuildEnv,
             dockerResource = getDockerResource(dispatchType),
-            qpcUniquePath = getQpcUniquePath(dispatchMessage.projectId)
+            qpcUniquePath = getQpcUniquePath(dispatchMessage)
         )
 
         pipelineDockerTaskSimpleDao.createOrUpdate(
@@ -500,7 +500,8 @@ class DockerHostClient @Autowired constructor(
         }
     }
 
-    private fun getQpcUniquePath(projectId: String): String? {
+    private fun getQpcUniquePath(dispatchMessage: DispatchMessage): String? {
+        val projectId = dispatchMessage.projectId
         return if (projectId.startsWith("git_") &&
             dockerHostQpcService.checkQpcWhitelist(projectId.removePrefix("git_"))
         ) {

src/backend/ci/core/dispatch-docker/biz-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/controller/OPDispatchDockerResourceImpl.kt

@@ -34,6 +34,7 @@ import com.tencent.devops.dispatch.docker.pojo.DockerHostLoadConfig
 import com.tencent.devops.dispatch.docker.pojo.DockerIpInfoVO
 import com.tencent.devops.dispatch.docker.pojo.DockerIpListPage
 import com.tencent.devops.dispatch.docker.pojo.DockerIpUpdateVO
+import com.tencent.devops.dispatch.docker.pojo.HostDriftLoad
 import com.tencent.devops.dispatch.docker.service.DispatchDockerService
 
 @RestResource
@@ -88,7 +89,7 @@ class OPDispatchDockerResourceImpl constructor(
         return Result(dispatchDockerService.getDockerDriftThreshold(userId))
     }
 
-    override fun updateDockerDriftThreshold(userId: String, thresholdMap: Map<String, String>): Result<Boolean> {
-        return Result(dispatchDockerService.updateDockerDriftThreshold(userId, thresholdMap))
+    override fun updateDockerDriftThreshold(userId: String, hostDriftLoad: HostDriftLoad): Result<Boolean> {
+        return Result(dispatchDockerService.updateDockerDriftThreshold(userId, hostDriftLoad))
     }
 }

src/backend/ci/core/dispatch-docker/biz-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/dao/PipelineDockerHostDao.kt

@@ -115,7 +115,7 @@ class PipelineDockerHostDao {
         dslContext: DSLContext,
         projectId: String,
         type: DockerHostType = DockerHostType.BUILD
-    ): List<String> {
+    ): Set<String> {
         with(TDispatchPipelineDockerHost.T_DISPATCH_PIPELINE_DOCKER_HOST) {
             val result = dslContext.select(HOST_IP).from(this)
                 .where(PROJECT_CODE.eq(projectId))
@@ -123,9 +123,9 @@ class PipelineDockerHostDao {
                 .fetchOne(HOST_IP)
 
             return if (result != null && result.isNotEmpty()) {
-                result.split(",")
+                result.split(",").toSet()
             } else {
-                emptyList()
+                emptySet()
             }
         }
     }

src/backend/ci/core/dispatch-docker/biz-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/listener/DockerVMListener.kt

@@ -114,8 +114,8 @@ class DockerVMListener @Autowired constructor(
         var poolNo = 0
         try {
             // 先判断是否OP已配置专机，若配置了专机，看当前ip是否在专机列表中，若在 选择当前IP并检查负载，若不在从专机列表中选择一个容量最小的
-            val specialIpSet = pipelineDockerHostDao.getHostIps(dslContext, dispatchMessage.projectId).toSet()
-            logger.info("${dispatchMessage.projectId}| specialIpSet: $specialIpSet")
+            val specialIpSet = pipelineDockerHostDao.getHostIps(dslContext, dispatchMessage.projectId)
+            logger.info("${dispatchMessage.projectId}| specialIpSet: $specialIpSet -- ${specialIpSet.size}")
 
             val taskHistory = pipelineDockerTaskSimpleDao.getByPipelineIdAndVMSeq(
                 dslContext = dslContext,
@@ -138,31 +138,9 @@ class DockerVMListener @Autowired constructor(
                     )
                 } else {
                     driftIpInfo = JsonUtil.toJson(dockerIpInfo.intoMap())
-
-                    dockerPair = if (specialIpSet.isNotEmpty() && specialIpSet.toString() != "[]") {
-                        // 该项目工程配置了专机
-                        if (specialIpSet.contains(taskHistory.dockerIp) && dockerIpInfo.enable) {
-                            // 上一次构建IP在专机列表中，直接重用
-                            Pair(taskHistory.dockerIp, dockerIpInfo.dockerHostPort)
-                        } else {
-                            // 不在专机列表中，重新依据专机列表去选择负载最小的
-                            driftIpInfo = "专机漂移"
-
-                            dockerHostUtils.getAvailableDockerIpWithSpecialIps(
-                                dispatchMessage.projectId,
-                                dispatchMessage.pipelineId,
-                                dispatchMessage.vmSeqId,
-                                specialIpSet
-                            )
-                        }
-                    } else {
-                        // 没有配置专机，根据当前IP负载选择IP
-                        val triple = dockerHostUtils.checkAndSetIP(dispatchMessage, specialIpSet, dockerIpInfo, poolNo)
-                        if (triple.third.isNotEmpty()) {
-                            driftIpInfo = triple.third
-                        }
-                        Pair(triple.first, triple.second)
-                    }
+                    // 根据当前IP负载选择IP
+                    val pair = dockerHostUtils.checkAndSetIP(dispatchMessage, specialIpSet, dockerIpInfo, poolNo)
+                    dockerPair = Pair(pair.first, pair.second)
                 }
             } else {
                 // 第一次构建，根据负载条件选择可用IP

src/backend/ci/core/dispatch-docker/biz-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/service/DispatchDockerService.kt

@@ -34,6 +34,7 @@ import com.tencent.devops.dispatch.docker.pojo.DockerHostLoadConfig
 import com.tencent.devops.dispatch.docker.pojo.DockerIpInfoVO
 import com.tencent.devops.dispatch.docker.pojo.DockerIpListPage
 import com.tencent.devops.dispatch.docker.pojo.DockerIpUpdateVO
+import com.tencent.devops.dispatch.docker.pojo.HostDriftLoad
 import com.tencent.devops.dispatch.docker.pojo.enums.DockerHostClusterType
 import com.tencent.devops.dispatch.docker.utils.CommonUtils
 import com.tencent.devops.dispatch.docker.utils.DockerHostUtils
@@ -213,14 +214,10 @@ class DispatchDockerService @Autowired constructor(
         return mapOf("threshold" to dockerHostUtils.getDockerDriftThreshold().toString())
     }
 
-    fun updateDockerDriftThreshold(userId: String, thresholdMap: Map<String, String>): Boolean {
-        logger.info("$userId updateDockerDriftThreshold $thresholdMap")
-        val threshold = (thresholdMap["threshold"] ?: error("Parameter threshold must in (0-100).")).toInt()
-        if (threshold < 0 || threshold > 100) {
-            throw IllegalArgumentException("Parameter threshold must in (0-100).")
-        }
+    fun updateDockerDriftThreshold(userId: String, hostDriftLoad: HostDriftLoad): Boolean {
+        logger.info("$userId updateDockerDriftThreshold $hostDriftLoad")
 
-        dockerHostUtils.updateDockerDriftThreshold(threshold)
+        dockerHostUtils.updateDockerDriftThreshold(hostDriftLoad)
         return true
     }
 }

src/backend/ci/core/dispatch-docker/biz-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/service/DockerHostQpcService.kt

@@ -27,6 +27,7 @@
 
 package com.tencent.devops.dispatch.docker.service
 
+import com.tencent.devops.common.dispatch.sdk.pojo.DispatchMessage
 import com.tencent.devops.common.redis.RedisOperation
 import com.tencent.devops.dispatch.docker.common.Constants
 import org.slf4j.LoggerFactory
@@ -67,6 +68,17 @@ class DockerHostQpcService @Autowired constructor(
         return redisOperation.isMember(Constants.QPC_WHITE_LIST_KEY_PREFIX, gitProjectId)
     }
 
+    fun getQpcUniquePath(dispatchMessage: DispatchMessage): String? {
+        val projectId = dispatchMessage.projectId
+        return if (projectId.startsWith("git_") &&
+            checkQpcWhitelist(projectId.removePrefix("git_"))
+        ) {
+            return projectId.removePrefix("git_")
+        } else {
+            null
+        }
+    }
+
     companion object {
         private val LOG = LoggerFactory.getLogger(DockerHostQpcService::class.java)
     }

src/backend/ci/core/dispatch-docker/biz-dispatch-docker/src/main/kotlin/com/tencent/devops/dispatch/docker/utils/DockerHostUtils.kt

@@ -43,13 +43,14 @@ import com.tencent.devops.dispatch.docker.dao.PipelineDockerTaskDriftDao
 import com.tencent.devops.dispatch.docker.dao.PipelineDockerTaskSimpleDao
 import com.tencent.devops.dispatch.docker.exception.DockerServiceException
 import com.tencent.devops.dispatch.docker.pojo.DockerHostLoadConfig
+import com.tencent.devops.dispatch.docker.pojo.HostDriftLoad
 import com.tencent.devops.dispatch.docker.pojo.enums.DockerHostClusterType
+import com.tencent.devops.dispatch.docker.service.DockerHostQpcService
 import com.tencent.devops.dispatch.pojo.enums.PipelineTaskStatus
 import com.tencent.devops.model.dispatch.tables.records.TDispatchPipelineDockerIpInfoRecord
 import org.jooq.DSLContext
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
-import org.springframework.beans.factory.annotation.Value
 import org.springframework.stereotype.Component
 import java.util.Random
 
@@ -64,19 +65,17 @@ class DockerHostUtils @Autowired constructor(
     private val pipelineDockerTaskDriftDao: PipelineDockerTaskDriftDao,
     private val pipelineDockerTaskSimpleDao: PipelineDockerTaskSimpleDao,
     private val pipelineDockerBuildDao: PipelineDockerBuildDao,
+    private val dockerHostQpcService: DockerHostQpcService,
     private val dslContext: DSLContext
 ) {
     companion object {
         private const val LOAD_CONFIG_KEY = "dockerhost-load-config"
-        private const val DOCKER_DRIFT_THRESHOLD_KEY = "docker-drift-threshold-spKyQ86qdYhAkDDR"
+        private const val DOCKER_DRIFT_THRESHOLD_KEY = "dispatchdocker:drift-threshold-spKyQ86qdYhAkDDR"
         private const val BUILD_POOL_SIZE = 100 // 单个流水线可同时执行的任务数量
 
         private val logger = LoggerFactory.getLogger(DockerHostUtils::class.java)
     }
 
-    @Value("\${dispatch.defaultAgentLessIp:127.0.0.1}")
-    val defaultAgentLessIp: String = ""
-
     fun getAvailableDockerIpWithSpecialIps(
         projectId: String,
         pipelineId: String,
@@ -124,13 +123,6 @@ class DockerHostUtils @Autowired constructor(
         }
 
         if (dockerPair.first.isEmpty()) {
-            // agentless方案升级兼容
-            logger.info("defaultAgentLessIp: $defaultAgentLessIp")
-            if (clusterName == DockerHostClusterType.AGENT_LESS && defaultAgentLessIp.isNotEmpty()) {
-                val defaultAgentLessIpList = defaultAgentLessIp.split(",")
-                return Pair(defaultAgentLessIpList[Random().nextInt(defaultAgentLessIpList.size)], 80)
-            }
-
             if (specialIpSet.isNotEmpty()) {
                 throw DockerServiceException(errorType = ErrorCodeEnum.NO_SPECIAL_VM_ERROR.errorType,
                     errorCode = ErrorCodeEnum.NO_SPECIAL_VM_ERROR.errorCode,
@@ -248,45 +240,58 @@ class DockerHostUtils @Autowired constructor(
         specialIpSet: Set<String>,
         dockerIpInfo: TDispatchPipelineDockerIpInfoRecord,
         poolNo: Int
-    ): Triple<String, Int, String> {
-        val dockerIp = dockerIpInfo.dockerIp
-
+    ): Pair<String, Int> {
         // 查看当前IP负载情况，当前IP不可用或者负载超额或者设置为专机独享或者是否灰度已被切换，重新选择构建机
-        val threshold = getDockerDriftThreshold()
+        val hostDriftLoad = getDockerDriftThreshold()
         if (!dockerIpInfo.enable ||
-            dockerIpInfo.diskLoad > 90 ||
-            dockerIpInfo.diskIoLoad > 85 ||
-            dockerIpInfo.memLoad > threshold ||
-            dockerIpInfo.specialOn ||
+            dockerIpInfo.diskLoad > hostDriftLoad.disk ||
+            dockerIpInfo.diskIoLoad > hostDriftLoad.diskIo ||
+            dockerIpInfo.memLoad > hostDriftLoad.memory ||
+            dockerIpInfo.cpuLoad > hostDriftLoad.cpu ||
+            (dockerIpInfo.specialOn && !specialIpSet.contains(dockerIpInfo.dockerIp)) ||
             (dockerIpInfo.grayEnv != gray.isGray()) ||
-            (dockerIpInfo.usedNum > 40 && dockerIpInfo.memLoad > 60)) {
-            val pair = getAvailableDockerIpWithSpecialIps(
+            (dockerIpInfo.usedNum > hostDriftLoad.usedNum) ||
+            dockerHostQpcService.getQpcUniquePath(dispatchMessage) != null) {
+            return getAvailableDockerIpWithSpecialIps(
                 dispatchMessage.projectId,
                 dispatchMessage.pipelineId,
                 dispatchMessage.vmSeqId,
                 specialIpSet
             )
-            return Triple(pair.first, pair.second, "")
         }
 
-        return Triple(dockerIp, dockerIpInfo.dockerHostPort, "")
+        return Pair(dockerIpInfo.dockerIp, dockerIpInfo.dockerHostPort)
     }
 
-    fun updateDockerDriftThreshold(threshold: Int) {
-        redisOperation.set(DOCKER_DRIFT_THRESHOLD_KEY, threshold.toString())
+    fun updateDockerDriftThreshold(hostDriftLoad: HostDriftLoad) {
+        redisOperation.set(
+            key = DOCKER_DRIFT_THRESHOLD_KEY,
+            value = JsonUtil.toJson(hostDriftLoad),
+            expired = false
+        )
     }
 
-    fun getDockerDriftThreshold(): Int {
+    fun getDockerDriftThreshold(): HostDriftLoad {
         val thresholdStr = redisOperation.get(DOCKER_DRIFT_THRESHOLD_KEY)
         return if (thresholdStr != null && thresholdStr.isNotEmpty()) {
-            thresholdStr.toInt()
+            JsonUtil.to(thresholdStr, HostDriftLoad::class.java)
         } else {
-            90
+            HostDriftLoad(
+                cpu = 80,
+                memory = 70,
+                disk = 80,
+                diskIo = 80,
+                usedNum = 40
+            )
         }
     }
 
     fun createLoadConfig(loadConfigMap: Map<String, DockerHostLoadConfig>) {
-        redisOperation.set(LOAD_CONFIG_KEY, JsonUtil.toJson(loadConfigMap))
+        redisOperation.set(
+            key = LOAD_CONFIG_KEY,
+            value = JsonUtil.toJson(loadConfigMap),
+            expired = false
+        )
     }
 
     fun getLoadConfig(): Triple<DockerHostLoadConfig, DockerHostLoadConfig, DockerHostLoadConfig> {