feat: 引擎等MQ场景接入SCS框架 #7443

.github/workflows/release.yml

@@ -31,7 +31,7 @@ jobs:
     - uses: actions/setup-go@v3
       with:
         go-version: 1.19.6
-    - run: make clean all
+    - run: make BUILD_OUT_TAG=out clean all
       working-directory: src/agent/
     - uses: actions/upload-artifact@v1
       with:

.gitignore

@@ -38,6 +38,7 @@ build.yml
 *.ipa
 build/
 bkci-slim.tar.gz
+*.log.gz
 
 # 容器化忽略文件
 helm-charts/core/ci/bk-ci*.tgz

helm-charts/core/ci/Chart.lock

@@ -23,8 +23,11 @@ dependencies:
 - name: mongodb
   repository: https://raw.githubusercontent.com/bitnami/charts/eb5f9a9513d987b519f0ecd732e7031241c50328/bitnami
   version: 10.30.6
+- name: pulsar
+  repository: https://pulsar.apache.org/charts
+  version: 2.9.4
 - name: kubernetes-manager
   repository: file://./local_chart/kubernetes-management
   version: 0.0.36
-digest: sha256:521a93d5e92e46288b16ef26391656c9a7041dd880a01cf5c04389f1bb4f4186
-generated: "2023-03-14T17:35:25.523011238+08:00"
+digest: sha256:1ce4b50494ead247d371806c09c0c8c823c8857a7a1f1d7214410d279803e8a2
+generated: "2023-03-20T22:05:18.340401263+08:00"

helm-charts/core/ci/Chart.yaml

@@ -38,6 +38,10 @@ dependencies:
   version: 10.30.6
   repository: https://raw.githubusercontent.com/bitnami/charts/eb5f9a9513d987b519f0ecd732e7031241c50328/bitnami
   condition: mongodb.enabled
+- name: pulsar
+  version: 2.9.4
+  repository: https://pulsar.apache.org/charts
+  condition: pulsar.enabled
 - name: kubernetes-manager
   version: 0.0.36
   repository: "file://./local_chart/kubernetes-management"

helm-charts/core/ci/base/default_env.yaml

@@ -43,3 +43,4 @@ bkCiNotifyEmailSendChannel: blueking
 bkCiInitLocale: zh-CN
 bkCiRedisSsl: false
 bkCiEsHttps: false
+bkCiMqDefaultBinder: pulsar

helm-charts/core/ci/base/values.yaml

@@ -257,6 +257,38 @@ mongodb:
   # extraEnvVars:
   # - name: MONGODB_CLIENT_EXTRA_FLAGS
   #   value: "--ipv6"
+pulsar:
+  enabled: true
+  toolset:
+    additionalCommand: 
+      bin/pulsar-admin tenants create default-tenant;
+      bin/pulsar-admin namespaces create default-tenant/default-ns;
+  ## deployed withh emptyDir
+  volumes:
+    persistence: false
+  # disabled AntiAffinity
+  affinity:
+    anti_affinity: false
+  # disable auto recovery
+  components:
+    autorecovery: false
+    pulsar_manager: true
+  zookeeper:
+    replicaCount: 1
+  bookkeeper:
+    replicaCount: 1
+  broker:
+    replicaCount: 1
+    configData:
+      ## Enable `autoSkipNonRecoverableData` since bookkeeper is running
+      ## without persistence
+      autoSkipNonRecoverableData: "true"
+      # storage settings
+      managedLedgerDefaultEnsembleSize: "1"
+      managedLedgerDefaultWriteQuorum: "1"
+      managedLedgerDefaultAckQuorum: "1"
+  proxy:
+    replicaCount: 1
 kubernetes-manager:
   # 是否部署kubernetes-manager
   enabled: true

helm-charts/core/ci/build_chart.py

@@ -41,6 +41,10 @@
     'bkCiInfluxdbUser': '{{ include "bkci.influxdbUsername" . }}',
     'bkCiInfluxdbPassword': '{{ include "bkci.influxdbPassword" . }}',
     'bkCiInfluxdbAddr': 'http://{{ include "bkci.influxdbHost" . }}:{{ include "bkci.influxdbPort" . }}',
+    'bkCiPulsarUrl': '{{ include "bkci.pulsarServiceUrl" . }}',
+    'bkCiPulsarAuthToken': '{{ include "bkci.pulsarTokenAuthValue" . }}',
+    'bkCiPulsarTenant': '{{ include "bkci.pulsarTenant" . }}',
+    'bkCiPulsarNamespace': '{{ include "bkci.pulsarNamespace" . }}',
     'bkCiVersion': '{{ .Chart.AppVersion }}',
     'bkCiDispatchKubernetesNs': '{{ .Release.Namespace }}',
     'bkCiConsulDiscoveryTag': '{{ .Release.Namespace }}',

helm-charts/core/ci/templates/_helpers.tpl

@@ -60,6 +60,18 @@ Return the influxdb fullname
 {{- end -}}
 {{- end -}}
 
+{{/*
+Return the pulsar fullname
+*/}}
+{{- define "bkci.pulsar.fullname" -}}
+{{- if .Values.pulsar.fullnameOverride -}}
+{{- .Values.pulsar.fullnameOverride | trunc 63 | trimSuffix "-" -}}
+{{- else -}}
+{{- $name := default "pulsar" .Values.pulsar.nameOverride -}}
+{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" -}}
+{{- end -}}
+{{- end -}}
+
 
 {{/*
 Return the mongodb fullname
@@ -294,6 +306,50 @@ Return the db_quartz mongodb connection uri
 {{- end -}}
 {{- end -}}
 
+{{/*
+Return th pulsar service url
+*/}}
+{{- define "bkci.pulsarServiceUrl" -}}
+{{- if eq .Values.pulsar.enabled true -}}
+http://{{- include "bkci.pulsar.fullname" . -}}-proxy:{{- .Values.pulsar.proxy.ports.http -}}
+{{- else -}}
+{{- .Values.externalPulsar.serviceUrl -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Return th pulsar namespace
+*/}}
+{{- define "bkci.pulsarNamespace" -}}
+{{- if eq .Values.pulsar.enabled true -}}
+default-ns
+{{- else -}}
+{{- .Values.externalPulsar.namespace -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Return th pulsar tenant
+*/}}
+{{- define "bkci.pulsarTenant" -}}
+{{- if eq .Values.pulsar.enabled true -}}
+default-tenant
+{{- else -}}
+{{- .Values.externalPulsar.tenant -}}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Return th pulsar tokenAuthValue
+*/}}
+{{- define "bkci.pulsarTokenAuthValue" -}}
+{{- if eq .Values.pulsar.enabled true -}}
+""
+{{- else -}}
+{{- .Values.externalPulsar.tokenAuthValue -}}
+{{- end -}}
+{{- end -}}
+
 {{/*
 bkci standard labels
 */}}

scripts/bkenv.properties

@@ -355,6 +355,16 @@ BK_CI_INFLUXDB_PASSWORD=
 BK_CI_INFLUXDB_PORT=8086
 # BK_CI_INFLUXDB_USER默认为bk_ci. 按需修改.
 BK_CI_INFLUXDB_USER=bk_ci
+# BK_CI_PULSAR_URL
+BK_CI_PULSAR_URL=
+# BK_CI_PULSAR_AUTH_TOKEN
+BK_CI_PULSAR_AUTH_TOKEN=
+# BK_CI_PULSAR_TENANT
+BK_CI_PULSAR_TENANT=
+# BK_CI_PULSAR_NAMESPACE
+BK_CI_PULSAR_NAMESPACE=
+# BK_CI_MQ_DEFAULT_BINDER
+BK_CI_MQ_DEFAULT_BINDER=rabbit
 # BK_CI_JFROG_FQDN无默认值. 按需修改. jFrog完全合格域名
 BK_CI_JFROG_FQDN=
 # BK_CI_JFROG_HTTP_PORT默认为8080. 按需修改. jFrog构件下载服务的http端口

src/backend/ci/buildSrc/src/main/kotlin/plugins/task-deploy-to-maven.gradle.kts

@@ -176,40 +176,31 @@ signing {
     sign(publishing.publications["mavenJava"])
 }
 
+val shouldPublish = project.the<SourceSetContainer>()["main"].allSource.files.isNotEmpty()
+        || project.name == "common-dependencies"
+
 tasks.getByName("publish") {
-    onlyIf {
-        project.the<SourceSetContainer>()["main"].allSource.files.isNotEmpty()
-    }
+    onlyIf { shouldPublish }
 }
 
 tasks.getByName("generateMetadataFileForMavenJavaPublication") {
-    onlyIf {
-        project.the<SourceSetContainer>()["main"].allSource.files.isNotEmpty()
-    }
+    onlyIf { shouldPublish }
 }
 
 tasks.getByName("generatePomFileForMavenJavaPublication") {
-    onlyIf {
-        project.the<SourceSetContainer>()["main"].allSource.files.isNotEmpty()
-    }
+    onlyIf { shouldPublish }
 }
 
 tasks.getByName("publishMavenJavaPublicationToOssRepository") {
-    onlyIf {
-        project.the<SourceSetContainer>()["main"].allSource.files.isNotEmpty()
-    }
+    onlyIf { shouldPublish }
 }
 
 tasks.getByName("publishMavenJavaPublicationToMavenLocal") {
-    onlyIf {
-        project.the<SourceSetContainer>()["main"].allSource.files.isNotEmpty()
-    }
+    onlyIf { shouldPublish }
 }
 
 tasks.getByName("publishToMavenLocal") {
-    onlyIf {
-        project.the<SourceSetContainer>()["main"].allSource.files.isNotEmpty()
-    }
+    onlyIf { shouldPublish }
 }
 
 tasks.getByName("signMavenJavaPublication") {

src/backend/ci/core/artifactory/biz-artifactory/src/main/kotlin/com/tencent/devops/artifactory/mq/ArtifactoryPipelineExtendConfiguration.kt

@@ -27,22 +27,13 @@
 
 package com.tencent.devops.artifactory.mq
 
-import com.tencent.devops.common.event.dispatcher.pipeline.mq.MQ
-import com.tencent.devops.common.event.dispatcher.pipeline.mq.MQEventDispatcher
-import com.tencent.devops.common.event.dispatcher.pipeline.mq.Tools
-import org.springframework.amqp.core.Binding
-import org.springframework.amqp.core.BindingBuilder
-import org.springframework.amqp.core.FanoutExchange
-import org.springframework.amqp.core.Queue
-import org.springframework.amqp.rabbit.connection.ConnectionFactory
-import org.springframework.amqp.rabbit.core.RabbitAdmin
-import org.springframework.amqp.rabbit.core.RabbitTemplate
-import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer
-import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter
+import com.tencent.devops.common.event.annotation.EventConsumer
+import com.tencent.devops.common.event.pojo.pipeline.PipelineBuildFinishBroadCastEvent
+import com.tencent.devops.common.stream.constants.StreamBinding
 import org.springframework.beans.factory.annotation.Autowired
-import org.springframework.beans.factory.annotation.Value
-import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
+import org.springframework.messaging.Message
+import java.util.function.Consumer
 
 /**
  * 流水线构建扩展配置
@@ -51,59 +42,19 @@ import org.springframework.context.annotation.Configuration
 @Suppress("ALL")
 class ArtifactoryPipelineExtendConfiguration {
 
-    @Bean
-    fun rabbitAdmin(connectionFactory: ConnectionFactory): RabbitAdmin {
-        return RabbitAdmin(connectionFactory)
+    companion object {
+        const val STREAM_CONSUMER_GROUP = "artifactory-service"
     }
 
-    @Bean
-    fun pipelineEventDispatcher(rabbitTemplate: RabbitTemplate) = MQEventDispatcher(rabbitTemplate)
-
     /**
      * 构建广播交换机
      */
-    @Bean
-    fun pipelineBuildFanoutExchange(): FanoutExchange {
-        val fanoutExchange = FanoutExchange(MQ.EXCHANGE_PIPELINE_BUILD_FINISH_FANOUT, true, false)
-        fanoutExchange.isDelayed = true
-        return fanoutExchange
-    }
-
-    @Value("\${queueConcurrency.buildFinishExt:1}")
-    private val buildFinishExtConcurrency: Int? = null
-
-    /**
-     *  构建结束，刷额外数据--- 并发小
-     */
-    @Bean
-    fun buildFinishExtQueue() = Queue(MQ.QUEUE_PIPELINE_BUILD_FINISH_EXT)
-
-    @Bean
-    fun buildFinishExtQueueQueueBind(
-        @Autowired buildFinishExtQueue: Queue,
-        @Autowired pipelineBuildFanoutExchange: FanoutExchange
-    ): Binding {
-        return BindingBuilder.bind(buildFinishExtQueue).to(pipelineBuildFanoutExchange)
-    }
-
-    @Bean
-    fun pipelineBuildFinishExtListenerContainer(
-        @Autowired connectionFactory: ConnectionFactory,
-        @Autowired buildFinishExtQueue: Queue,
-        @Autowired rabbitAdmin: RabbitAdmin,
-        @Autowired listener: PipelineBuildArtifactoryListener,
-        @Autowired messageConverter: Jackson2JsonMessageConverter
-    ): SimpleMessageListenerContainer {
-        return Tools.createSimpleMessageListenerContainer(
-            connectionFactory = connectionFactory,
-            queue = buildFinishExtQueue,
-            rabbitAdmin = rabbitAdmin,
-            buildListener = listener,
-            messageConverter = messageConverter,
-            startConsumerMinInterval = 120000,
-            consecutiveActiveTrigger = 15,
-            concurrency = buildFinishExtConcurrency!!,
-            maxConcurrency = 3
-        )
+    @EventConsumer(StreamBinding.EXCHANGE_PIPELINE_BUILD_FINISH_FANOUT, STREAM_CONSUMER_GROUP)
+    fun buildFinishListener(
+        @Autowired listener: PipelineBuildArtifactoryListener
+    ): Consumer<Message<PipelineBuildFinishBroadCastEvent>> {
+        return Consumer { event: Message<PipelineBuildFinishBroadCastEvent> ->
+            listener.onBuildFinished(event.payload)
+        }
     }
 }

src/backend/ci/core/artifactory/biz-artifactory/src/main/kotlin/com/tencent/devops/artifactory/mq/PipelineBuildArtifactoryListener.kt

@@ -32,8 +32,6 @@ import com.tencent.devops.artifactory.service.PipelineBuildArtifactoryService
 import com.tencent.devops.common.api.exception.RemoteServiceException
 import com.tencent.devops.common.api.util.JsonUtil
 import com.tencent.devops.common.client.Client
-import com.tencent.devops.common.event.dispatcher.pipeline.PipelineEventDispatcher
-import com.tencent.devops.common.event.listener.pipeline.BaseListener
 import com.tencent.devops.common.event.pojo.pipeline.PipelineBuildFinishBroadCastEvent
 import com.tencent.devops.process.api.service.ServiceBuildResource
 import com.tencent.devops.process.api.service.ServicePipelineRuntimeResource
@@ -47,16 +45,15 @@ import org.springframework.stereotype.Component
 @Component
 @Suppress("ALL")
 class PipelineBuildArtifactoryListener @Autowired constructor(
-    pipelineEventDispatcher: PipelineEventDispatcher,
     private val pipelineBuildArtifactoryService: PipelineBuildArtifactoryService,
     private val client: Client
-) : BaseListener<PipelineBuildFinishBroadCastEvent>(pipelineEventDispatcher) {
+) {
 
     companion object {
         private val logger = LoggerFactory.getLogger(PipelineBuildArtifactoryListener::class.java)!!
     }
 
-    override fun run(event: PipelineBuildFinishBroadCastEvent) {
+    fun onBuildFinished(event: PipelineBuildFinishBroadCastEvent) {
         logger.info("PipelineBuildArtifactoryListener.run, event: $event")
         val userId = event.userId
         val projectId = event.projectId