feat: 流水线支持设置并发组执行策略 #6521

src/backend/ci/core/log/api-log/src/main/kotlin/com/tencent/devops/common/log/utils/BuildLogPrinter.kt

@@ -114,17 +114,21 @@ class BuildLogPrinter(
         executeCount: Int,
         subTag: String? = null
     ) {
-        genLogPrintPrintResource().addLogLine(
-            buildId = buildId,
-            logMessage = genLogMessage(
-                message = "$LOG_ERROR_FLAG${message.replace("\n", "\n$LOG_ERROR_FLAG")}",
-                tag = tag,
-                subTag = subTag,
-                jobId = jobId,
-                logType = LogType.ERROR,
-                executeCount = executeCount
+        try {
+            genLogPrintPrintResource().addLogLine(
+                buildId = buildId,
+                logMessage = genLogMessage(
+                    message = "$LOG_ERROR_FLAG${message.replace("\n", "\n$LOG_ERROR_FLAG")}",
+                    tag = tag,
+                    subTag = subTag,
+                    jobId = jobId,
+                    logType = LogType.ERROR,
+                    executeCount = executeCount
+                )
             )
-        )
+        } catch (e: Exception) {
+            logger.error("[$buildId]|addErrorLine error|message=$message", e)
+        }
     }
 
     fun addDebugLine(
@@ -135,17 +139,21 @@ class BuildLogPrinter(
         executeCount: Int,
         subTag: String? = null
     ) {
-        genLogPrintPrintResource().addLogLine(
-            buildId = buildId,
-            logMessage = genLogMessage(
-                message = "$LOG_DEBUG_FLAG${message.replace("\n", "\n$LOG_DEBUG_FLAG")}",
-                tag = tag,
-                subTag = subTag,
-                jobId = jobId,
-                logType = LogType.DEBUG,
-                executeCount = executeCount
+        try {
+            genLogPrintPrintResource().addLogLine(
+                buildId = buildId,
+                logMessage = genLogMessage(
+                    message = "$LOG_DEBUG_FLAG${message.replace("\n", "\n$LOG_DEBUG_FLAG")}",
+                    tag = tag,
+                    subTag = subTag,
+                    jobId = jobId,
+                    logType = LogType.DEBUG,
+                    executeCount = executeCount
+                )
             )
-        )
+        } catch (e: Exception) {
+            logger.error("[$buildId]|addDebugLine error|message=$message", e)
+        }
     }
 
     fun addYellowLine(
@@ -245,6 +253,7 @@ class BuildLogPrinter(
         private const val LOG_DEBUG_FLAG = "##[debug]"
 
         private const val LOG_ERROR_FLAG = "##[error]"
+
         @Suppress("UNUSED")
         private const val LOG_WARN_FLAG = "##[warning]"
     }

src/backend/ci/core/process/api-process/src/main/kotlin/com/tencent/devops/process/engine/pojo/BuildInfo.kt

@@ -53,7 +53,8 @@ data class BuildInfo(
     val channelCode: ChannelCode,
     val buildParameters: List<BuildParameters>?,
     var errorInfoList: List<ErrorInfo>?,
-    val retryFlag: Boolean? = null
+    val retryFlag: Boolean? = null,
+    val concurrencyGroup: String? = null
 ) {
 
     fun isFinish() = when {

src/backend/ci/core/process/api-process/src/main/kotlin/com/tencent/devops/process/pojo/BuildHistory.kt

@@ -91,5 +91,7 @@ data class BuildHistory(
     @ApiModelProperty("自定义构建版本号", required = false)
     val buildNumAlias: String? = null,
     @ApiModelProperty("更新时间", required = false)
-    val updateTime: Long? = null
+    val updateTime: Long? = null,
+    @ApiModelProperty("并发时,设定的group", required = false)
+    var concurrencyGroup: String? = null
 )

src/backend/ci/core/process/api-process/src/main/kotlin/com/tencent/devops/process/pojo/setting/PipelineSetting.kt

@@ -64,6 +64,10 @@ data class PipelineSetting(
     val waitQueueTimeMinute: Int = PIPELINE_SETTING_WAIT_QUEUE_TIME_MINUTE_DEFAULT,
     @ApiModelProperty("最大排队数量", required = false)
     val maxQueueSize: Int = PIPELINE_SETTING_MAX_QUEUE_SIZE_DEFAULT,
+    @ApiModelProperty("并发时,设定的group", required = false)
+    var concurrencyGroup: String? = null,
+    @ApiModelProperty("并发时,是否相同group取消正在执行的流水线", required = false)
+    val concurrencyCancelInProgress: Boolean = false,
     @ApiModelProperty("是否有操作权限", required = false)
     var hasPermission: Boolean? = null,
     @ApiModelProperty("保存流水线编排的最大个数", required = false)

src/backend/ci/core/process/api-process/src/main/kotlin/com/tencent/devops/process/utils/PipelineVarUtil.kt

@@ -241,6 +241,17 @@ object PipelineVarUtil {
         }
     }
 
+    /**
+     * 填充CI预置变量
+     */
+    fun fillContextVarMap(buildVar: Map<String, String>): Map<String, String> {
+        val varMap = mutableMapOf<String, String>()
+        contextVarMappingBuildVar.forEach { (contextKey, varKey) ->
+            if (!buildVar[varKey].isNullOrBlank()) varMap[contextKey] = buildVar[varKey]!!
+        }
+        return buildVar.plus(varMap)
+    }
+
     /**
      * 获取CI预置变量
      */

src/backend/ci/core/process/biz-base/src/main/kotlin/com/tencent/devops/process/dao/PipelineSettingDao.kt

@@ -134,7 +134,9 @@ class PipelineSettingDao {
                 IS_TEMPLATE,
                 MAX_PIPELINE_RES_NUM,
                 MAX_CON_RUNNING_QUEUE_SIZE,
-                BUILD_NUM_RULE
+                BUILD_NUM_RULE,
+                CONCURRENCY_GROUP,
+                CONCURRENCY_CANCEL_IN_PROGRESS
             )
                 .values(
                     setting.projectId,
@@ -163,7 +165,9 @@ class PipelineSettingDao {
                     isTemplate,
                     setting.maxPipelineResNum,
                     setting.maxConRunningQueueSize,
-                    setting.buildNumRule
+                    setting.buildNumRule,
+                    setting.concurrencyGroup,
+                    setting.concurrencyCancelInProgress
                 ).onDuplicateKeyUpdate()
                 .set(NAME, setting.pipelineName)
                 .set(DESC, setting.desc)
@@ -190,6 +194,8 @@ class PipelineSettingDao {
                 .set(MAX_PIPELINE_RES_NUM, setting.maxPipelineResNum)
                 .set(MAX_CON_RUNNING_QUEUE_SIZE, setting.maxConRunningQueueSize)
                 .set(BUILD_NUM_RULE, setting.buildNumRule)
+                .set(CONCURRENCY_GROUP, setting.concurrencyGroup)
+                .set(CONCURRENCY_CANCEL_IN_PROGRESS, setting.concurrencyCancelInProgress)
                 .execute()
         }
     }

src/backend/ci/core/process/biz-base/src/main/kotlin/com/tencent/devops/process/engine/dao/PipelineBuildDao.kt

@@ -51,6 +51,7 @@ import org.springframework.stereotype.Repository
 import java.sql.Timestamp
 import java.time.LocalDateTime
 import javax.ws.rs.core.Response
+import org.jooq.Record2
 
 @Suppress("ALL")
 @Repository
@@ -80,7 +81,8 @@ class PipelineBuildDao {
         webhookType: String?,
         webhookInfo: String?,
         buildMsg: String?,
-        buildNumAlias: String? = null
+        buildNumAlias: String? = null,
+        concurrencyGroup: String? = null
     ) {
         try {
             with(T_PIPELINE_BUILD_HISTORY) {
@@ -106,7 +108,8 @@ class PipelineBuildDao {
                     WEBHOOK_TYPE,
                     WEBHOOK_INFO,
                     BUILD_MSG,
-                    BUILD_NUM_ALIAS
+                    BUILD_NUM_ALIAS,
+                    CONCURRENCY_GROUP
                 ).values(
                     buildId,
                     buildNum,
@@ -128,7 +131,8 @@ class PipelineBuildDao {
                     webhookType,
                     webhookInfo,
                     buildMsg,
-                    buildNumAlias
+                    buildNumAlias,
+                    concurrencyGroup
                 ).execute()
             }
         } catch (t: Throwable) {
@@ -164,6 +168,21 @@ class PipelineBuildDao {
         }
     }
 
+    fun getBuildTasksByConcurrencyGroup(
+        dslContext: DSLContext,
+        projectId: String,
+        concurrencyGroup: String,
+        statusSet: List<BuildStatus>
+    ): List<Record2<String, String>> {
+        return with(T_PIPELINE_BUILD_HISTORY) {
+            dslContext.select(PIPELINE_ID, BUILD_ID).from(this)
+                .where(PROJECT_ID.eq(projectId))
+                .and(STATUS.`in`(statusSet.map { it.ordinal }))
+                .and(CONCURRENCY_GROUP.eq(concurrencyGroup))
+                .fetch()
+        }
+    }
+
     fun getBuildInfo(
         dslContext: DSLContext,
         projectId: String,
@@ -497,7 +516,8 @@ class PipelineBuildDao {
                     JsonUtil.getObjectMapper().readValue(self) as List<BuildParameters>
                 },
                 retryFlag = t.isRetry,
-                executeTime = t.executeTime ?: 0
+                executeTime = t.executeTime ?: 0,
+                concurrencyGroup = t.concurrencyGroup
             )
         }
     }

src/backend/ci/core/process/biz-base/src/main/kotlin/com/tencent/devops/process/engine/interceptor/InterceptData.kt

@@ -30,6 +30,7 @@ package com.tencent.devops.process.engine.interceptor
 import com.tencent.devops.common.pipeline.Model
 import com.tencent.devops.common.pipeline.enums.StartType
 import com.tencent.devops.process.engine.pojo.PipelineInfo
+import com.tencent.devops.process.pojo.setting.PipelineSetting
 
 /**
  *
@@ -39,5 +40,6 @@ import com.tencent.devops.process.engine.pojo.PipelineInfo
 data class InterceptData(
     val pipelineInfo: PipelineInfo,
     val model: Model?,
-    val startType: StartType
+    val startType: StartType,
+    val setting: PipelineSetting?
 )