Worker 构建执行器模块架构指南

模块概述

Worker 模块是 BK-CI 构建执行器的核心组件，运行在构建机（Agent）上，负责接收并执行流水线任务。它是连接 CI 服务端与构建环境的桥梁，实现了任务调度、插件执行、日志上报、心跳维护等关键功能。

核心职责

1. 任务执行：接收并执行流水线中的各类任务（脚本、插件等）
2. 日志服务：收集构建日志并上报到 Log 服务
3. 心跳维护：定期向服务端发送心跳，维持构建状态
4. 变量管理：管理构建过程中的环境变量和上下文
5. 制品归档：将构建产物上传到制品库
6. 插件运行：下载并执行研发商店中的插件

运行模式

Worker 支持三种构建类型：

目录结构

src/backend/ci/core/worker/
├── build.gradle.kts              # 模块构建配置
├── worker-agent/                 # Agent 启动入口
│   ├── src/main/kotlin/
│   │   └── com/tencent/devops/agent/
│   │       ├── Application.kt   # 主入口，根据构建类型启动
│   │       ├── AgentVersion.kt  # 版本信息
│   │       └── service/
│   │           └── BuildLessStarter.kt  # 无编译构建启动器
│   └── src/test/kotlin/          # 测试代码
├── worker-common/                # 公共组件库
│   ├── src/main/kotlin/
│   │   └── com/tencent/devops/worker/common/
│   │       ├── Runner.kt         # 核心运行器
│   │       ├── WorkRunner.kt     # 第三方构建机运行器
│   │       ├── api/              # API 客户端
│   │       ├── env/              # 环境变量管理
│   │       ├── heartbeat/        # 心跳服务
│   │       ├── logger/           # 日志服务
│   │       ├── service/          # 业务服务
│   │       ├── task/             # 任务执行框架
│   │       ├── utils/            # 工具类
│   │       └── constants/        # 常量定义
│   └── src/main/kotlin/
│       └── com/tencent/devops/plugin/worker/task/
│           ├── archive/          # 归档任务
│           ├── scm/              # 代码拉取任务
│           └── image/            # 镜像任务
└── worker-api-sdk/               # API SDK 实现
    └── src/main/kotlin/
        └── com/tencent/devops/worker/common/api/
            ├── archive/          # 制品库 API
            ├── atom/             # 插件 API
            ├── log/              # 日志 API
            ├── process/          # 流水线 API
            └── ...               # 其他 API

核心组件

1. Runner – 核心运行器

Runner.kt 是 Worker 的核心入口，负责整个构建生命周期管理：

object Runner {
    fun run(workspaceInterface: WorkspaceInterface, systemExit: Boolean = true) {
        // 1. 获取构建变量
        val buildVariables = getBuildVariables()
        
        // 2. 准备工作空间、启动日志服务、启动心跳
        val workspacePathFile = prepareWorker(buildVariables, workspaceInterface)
        
        // 3. 循环获取并执行任务
        loopPickup(workspacePathFile, buildVariables)
        
        // 4. 清理并结束构建
        finishWorker(buildVariables)
    }
}

核心流程：

┌─────────────────────────────────────────────────────────────┐
│                    Worker 启动流程                           │
├─────────────────────────────────────────────────────────────┤
│  1. EngineService.setStarted()  →  上报启动状态，获取构建变量   │
│  2. LoggerService.start()       →  启动日志服务                │
│  3. Heartbeat.start()           →  启动心跳监控                │
│  4. loopPickup()                →  循环领取任务                │
│     ├── claimTask()             →  获取待执行任务              │
│     ├── TaskFactory.create()    →  创建任务执行器              │
│     ├── taskDaemon.run()        →  执行任务                   │
│     └── completeTask()          →  上报任务结果                │
│  5. finishWorker()              →  结束构建                   │
└─────────────────────────────────────────────────────────────┘

2. TaskFactory – 任务工厂

任务工厂负责根据任务类型创建对应的执行器：

object TaskFactory {
    private val taskMap = ConcurrentHashMap<String, KClass<out ITask>>()
    
    fun init() {
        // 注册内置任务类型
        register(LinuxScriptElement.classType, LinuxScriptTask::class)
        register(WindowsScriptElement.classType, WindowsScriptTask::class)
        register(MarketBuildAtomElement.classType, MarketAtomTask::class)
        
        // 通过反射扫描注册插件任务
        val reflections = Reflections("com.tencent.devops.plugin.worker.task")
        // ...
    }
    
    fun create(type: String): ITask {
        val clazz = taskMap[type] ?: return EmptyTask(type)
        return clazz.primaryConstructor?.call() ?: EmptyTask(type)
    }
}

3. ITask – 任务接口

所有任务执行器的基类：

abstract class ITask {
    private val environment = HashMap<String, String>()
    
    // 任务执行入口
    fun run(buildTask: BuildTask, buildVariables: BuildVariables, workspace: File) {
        execute(buildTask, buildVariables, workspace)
    }
    
    // 子类实现具体执行逻辑
    protected abstract fun execute(
        buildTask: BuildTask,
        buildVariables: BuildVariables,
        workspace: File
    )
    
    // 添加环境变量（输出变量）
    protected fun addEnv(env: Map<String, String>) {
        // 校验只读变量、变量名合法性
        environment.putAll(env)
    }
    
    fun getAllEnv(): Map<String, String> = environment
}

4. LoggerService – 日志服务

负责构建日志的收集、缓冲和上报：

object LoggerService {
    private val uploadQueue = LinkedBlockingQueue<LogMessage>(2000)
    
    fun addNormalLine(message: String) {
        // 1. 处理日志前缀（DEBUG/ERROR/WARN）
        // 2. 敏感信息过滤
        // 3. 本地落盘（如需要）
        // 4. 放入上报队列
        uploadQueue.put(logMessage)
    }
    
    fun addErrorLine(message: String) {
        addNormalLine("$LOG_ERROR_FLAG$message")
    }
    
    fun addWarnLine(message: String) {
        addNormalLine("$LOG_WARN_FLAG$message")
    }
    
    // 日志折叠
    fun addFoldStartLine(foldName: String) {
        addLog(LogMessage(message = "##[group]$foldName", ...))
    }
    
    fun addFoldEndLine(foldName: String) {
        addLog(LogMessage(message = "##[endgroup]$foldName", ...))
    }
}

5. Heartbeat – 心跳服务

维持与服务端的连接，监控任务超时：

object Heartbeat {
    fun start(jobTimeoutMills: Long, executeCount: Int) {
        // 定时心跳（每2秒）
        executor.scheduleWithFixedDelay({
            val heartBeatInfo = EngineService.heartbeat(...)
            // 处理取消任务
            if (!heartBeatInfo.cancelTaskIds.isNullOrEmpty()) {
                KillBuildProcessTree.killProcessTree(...)
            }
        }, 10, 2, TimeUnit.SECONDS)
        
        // Job 超时监控
        executor.scheduleWithFixedDelay({
            LoggerService.addErrorLine("Job timeout")
            EngineService.timeout()
            exitProcess(99)
        }, jobTimeoutMills, jobTimeoutMills, TimeUnit.MILLISECONDS)
    }
}

6. EngineService – 引擎服务

与 Process 服务通信的客户端：

object EngineService {
    // 上报启动状态
    fun setStarted(): BuildVariables
    
    // 领取任务
    fun claimTask(): BuildTask
    
    // 完成任务
    fun completeTask(taskResult: BuildTaskResult)
    
    // 结束构建
    fun endBuild(variables: Map<String, String>, result: BuildJobResult)
    
    // 心跳
    fun heartbeat(executeCount: Int, jobHeartbeatRequest: JobHeartbeatRequest): HeartBeatInfo
    
    // 超时上报
    fun timeout()
    
    // 错误上报
    fun submitError(errorInfo: ErrorInfo)
}

任务类型

内置任务

插件任务（plugin 包）

MarketAtomTask – 插件执行

研发商店插件的执行流程：

┌─────────────────────────────────────────────────────────────┐
│                   插件执行流程                               │
├─────────────────────────────────────────────────────────────┤
│  1. 获取插件信息                                             │
│     atomApi.getAtomEnv(projectCode, atomCode, atomVersion)  │
│                                                             │
│  2. 准备执行环境                                             │
│     ├── 创建临时工作目录                                     │
│     ├── 下载/缓存插件包                                      │
│     └── 校验 SHA 完整性                                      │
│                                                             │
│  3. 准备输入参数                                             │
│     ├── 解析 input.json                                     │
│     ├── 替换变量和凭据                                       │
│     └── 写入 .sdk.json（SDK 环境信息）                       │
│                                                             │
│  4. 执行插件                                                 │
│     ├── Linux: ShellUtil.execute()                          │
│     └── Windows: BatScriptUtil.execute()                    │
│                                                             │
│  5. 处理输出                                                 │
│     ├── 读取 output.json                                    │
│     ├── 处理输出变量                                         │
│     ├── 归档制品（artifact 类型）                            │
│     └── 归档报告（report 类型）                              │
└─────────────────────────────────────────────────────────────┘

插件 SDK 环境

Worker 为插件提供的 SDK 环境信息（.sdk.json）：

data class SdkEnv(
    val buildType: BuildType,    // 构建类型
    val projectId: String,       // 项目ID（english_name）
    val agentId: String,         // 构建机ID
    val secretKey: String,       // 密钥
    val gateway: String,         // 网关地址
    val buildId: String,         // 构建ID
    val vmSeqId: String,         // 虚拟机序号
    val fileGateway: String,     // 文件网关
    val taskId: String,          // 任务ID
    val executeCount: Int        // 执行次数
)

插件输出处理

// output.json 格式
{
    "status": "success",  // success/failure
    "data": {
        "outVar1": {
            "type": "string",
            "value": "xxx"
        },
        "outVar2": {
            "type": "artifact",
            "value": ["file1.zip", "file2.tar.gz"]
        },
        "outVar3": {
            "type": "report",
            "label": "测试报告",
            "path": "reports/",
            "target": "index.html"
        }
    },
    "qualityData": { ... }  // 质量红线数据
}

API 客户端

AbstractBuildResourceApi

所有 API 客户端的基类，提供 HTTP 请求能力：

abstract class AbstractBuildResourceApi : WorkerRestApiSDK {
    // 请求重试机制
    protected fun requestForResponse(
        request: Request,
        retryCount: Int = DEFAULT_RETRY_TIME
    ): Response {
        // 支持 502/503/504 自动重试
        // 支持 DNS 错误、连接超时重试
    }
    
    // 构建请求头（包含认证信息）
    private fun getAllHeaders(headers: Map<String, String>): Map<String, String> {
        return buildArgs.plus(headers).plus(
            AUTH_HEADER_DEVOPS_BUILD_ID to buildInfo.buildId,
            AUTH_HEADER_DEVOPS_VM_SEQ_ID to buildInfo.vmSeqId
        )
    }
}

主要 API 客户端

工具类

ShellUtil – Shell 脚本执行

object ShellUtil {
    fun execute(
        buildId: String,
        script: String,
        dir: File,
        workspace: File,
        buildEnvs: List<BuildEnv>,
        runtimeVariables: Map<String, String>,
        errorMessage: String = "Fail to run the script"
    ): String {
        // 1. 创建临时脚本文件
        // 2. 设置环境变量
        // 3. 执行脚本
        // 4. 处理输出和错误
    }
}

BatScriptUtil – Bat 脚本执行

object BatScriptUtil {
    fun execute(
        buildId: String,
        script: String,
        runtimeVariables: Map<String, String>,
        dir: File,
        workspace: File,
        errorMessage: String
    ): String {
        // Windows 批处理脚本执行
    }
}

CredentialUtils – 凭据工具

object CredentialUtils {
    // 解析凭据占位符
    fun String.parseCredentialValue(
        context: Map<String, String>?,
        acrossProjectId: String?
    ): String {
        // 解析 ${{credentials.xxx}} 格式
    }
}

WorkspaceUtils – 工作空间工具

object WorkspaceUtils {
    // 获取流水线工作空间
    fun getPipelineWorkspace(pipelineId: String, workspace: String): File
    
    // 获取日志目录
    fun getPipelineLogDir(pipelineId: String): File
    
    // 获取构建日志属性
    fun getBuildLogProperty(...): TaskBuildLogProperty
}

环境变量

AgentEnv – 构建机环境

object AgentEnv {
    fun getProjectId(): String      // 项目ID
    fun getAgentId(): String        // 构建机ID
    fun getAgentSecretKey(): String // 密钥
    fun getGateway(): String        // 网关地址
    fun getOS(): OSType             // 操作系统类型
    fun getLocaleLanguage(): String // 语言设置
    fun getRuntimeJdkPath(): String // JDK 路径
}

BuildEnv – 构建环境

object BuildEnv {
    fun getBuildType(): BuildType   // 获取构建类型
    fun isThirdParty(): Boolean     // 是否第三方构建机
}

数据模型

BuildTask – 构建任务

data class BuildTask(
    val buildId: String,
    val vmSeqId: String,
    val status: BuildTaskStatus,    // DO/WAIT/END
    val taskId: String?,
    val elementId: String?,
    val elementName: String?,
    val type: String?,              // 任务类型
    val params: Map<String, String>?,
    val buildVariable: Map<String, String>?,
    val containerType: String?,
    val executeCount: Int?,
    val stepId: String?,
    val signToken: String?
)

BuildVariables – 构建变量

data class BuildVariables(
    val buildId: String,
    val vmSeqId: String,
    val vmName: String,
    val projectId: String,          // 项目ID（english_name）
    val pipelineId: String,
    val variables: Map<String, String>,
    val variablesWithType: List<BuildParameters>,
    val buildEnvs: List<BuildEnv>,
    val containerId: String,
    val containerHashId: String?,
    val jobId: String?,
    val timeoutMills: Long
)

BuildTaskResult – 任务结果

data class BuildTaskResult(
    val buildId: String,
    val vmSeqId: String,
    val taskId: String,
    val elementId: String,
    val success: Boolean,
    val buildVariable: Map<String, String>?,
    val errorType: String?,
    val errorCode: Int?,
    val message: String?,
    val type: String?,
    val monitorData: Map<String, Any>?,
    val platformCode: String?,
    val platformErrorCode: Int?
)

构建流程时序图

┌──────┐     ┌────────┐     ┌─────────┐     ┌────────┐
│Worker│     │Process │     │   Log   │     │  Store │
└──┬───┘     └───┬────┘     └────┬────┘     └───┬────┘
   │             │               │              │
   │ setStarted  │               │              │
   │────────────>│               │              │
   │ BuildVariables              │              │
   │<────────────│               │              │
   │             │               │              │
   │ claimTask   │               │              │
   │────────────>│               │              │
   │ BuildTask   │               │              │
   │<────────────│               │              │
   │             │               │              │
   │             │  addLogLine   │              │
   │             │──────────────>│              │
   │             │               │              │
   │             │               │  getAtomEnv  │
   │             │               │─────────────>│
   │             │               │   AtomEnv    │
   │             │               │<─────────────│
   │             │               │              │
   │ heartbeat   │               │              │
   │────────────>│               │              │
   │ HeartBeatInfo               │              │
   │<────────────│               │              │
   │             │               │              │
   │ completeTask│               │              │
   │────────────>│               │              │
   │             │               │              │
   │ endBuild    │               │              │
   │────────────>│               │              │
   └─────────────┴───────────────┴──────────────┘

与其他模块的关系

                    ┌─────────────┐
                    │   Process   │
                    │  (流水线引擎) │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
        ┌─────────┐  ┌─────────┐  ┌─────────┐
        │Dispatch │  │   Log   │  │  Store  │
        │(调度服务)│  │(日志服务)│  │(研发商店)│
        └────┬────┘  └────┬────┘  └────┬────┘
             │            │            │
             └────────────┼────────────┘
                          │
                    ┌─────▼─────┐
                    │  Worker   │
                    │ (构建执行器)│
                    └─────┬─────┘
                          │
         ┌────────────────┼────────────────┐
         │                │                │
         ▼                ▼                ▼
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │Artifactory│    │  Ticket  │    │Repository│
   │ (制品库)  │    │ (凭据)   │    │ (代码库) │
   └──────────┘    └──────────┘    └──────────┘

依赖关系说明

开发规范

新增任务类型

1. 创建任务类继承 ITask：

@TaskClassType(classTypes = ["myTaskType"], priority = 1)
class MyTask : ITask() {
    override fun execute(
        buildTask: BuildTask,
        buildVariables: BuildVariables,
        workspace: File
    ) {
        // 实现任务逻辑
        LoggerService.addNormalLine("执行任务...")
        
        // 输出变量
        addEnv(mapOf("OUTPUT_VAR" to "value"))
    }
}

2. 放置在 com.tencent.devops.plugin.worker.task 包下，自动注册

日志输出规范

// 普通日志
LoggerService.addNormalLine("正常信息")

// 警告日志（黄色）
LoggerService.addWarnLine("警告信息")

// 错误日志（红色）
LoggerService.addErrorLine("错误信息")

// 调试日志（灰色）
LoggerService.addDebugLine("调试信息")

// 折叠日志
LoggerService.addFoldStartLine("[安装依赖]")
// ... 详细日志
LoggerService.addFoldEndLine("-----")

错误处理

throw TaskExecuteException(
    errorMsg = "任务执行失败",
    errorType = ErrorType.USER,        // USER/SYSTEM/THIRD_PARTY/PLUGIN
    errorCode = ErrorCode.USER_TASK_OPERATE_FAIL
)

敏感信息处理

// 自动过滤敏感信息
SensitiveValueService.addSensitiveValue("password123")

// 日志中会自动替换为 ******
LoggerService.addNormalLine("密码是: password123")
// 输出: 密码是: ******

常见问题

Q1: Worker 如何与服务端通信？

Worker 通过 HTTP API 与服务端通信，所有请求都带有认证头：

• X-DEVOPS-BUILD-TYPE: 构建类型
• X-DEVOPS-PROJECT-ID: 项目ID
• X-DEVOPS-AGENT-ID: 构建机ID
• X-DEVOPS-AGENT-SECRET-KEY: 密钥
• X-DEVOPS-BUILD-ID: 构建ID
• X-DEVOPS-VM-SEQ-ID: 虚拟机序号

Q2: 插件执行失败如何排查？

1. 检查 output.json 输出
2. 查看 Worker 日志（构建机上的 logs/ 目录）
3. 检查插件 SHA 校验是否通过
4. 确认插件运行环境（Python/Node.js 版本等）

Q3: 心跳失败会怎样？

连续 12 次心跳失败后，Worker 会自动退出（exitProcess(-1)）。服务端会将构建标记为异常终止。

Q4: 如何调试 Worker？

1. 设置环境变量 DEVOPS_AGENT_LOG_DEBUG=true 开启调试日志
2. 使用 LoggerService.addDebugLine() 输出调试信息
3. 检查构建机上的日志文件

Q5: 插件包缓存机制？

• 第三方构建机：缓存在 {user.dir}/store/cache/plugins/
• 公共构建机：缓存在 {workspace}/../cache/store/cache/plugins/
• 缓存 Key：{atomCode}-{version}-{fileName}
• 测试版本插件不缓存

国际化

Worker 模块的国际化文件位于：

• support-files/i18n/worker/message_zh_CN.properties
• support-files/i18n/worker/message_en_US.properties

常量定义在 WorkerMessageCode.kt：

object WorkerMessageCode {
    const val BK_PREPARE_TO_BUILD = "bkPrepareToRunBuild"
    const val PARAMETER_ERROR = "parameterError"
    const val UNKNOWN_ERROR = "unknownError"
    const val AGENT_DNS_ERROR = "agentDnsError"
    const val AGENT_NETWORK_TIMEOUT = "agentNetworkTimeout"
    // ...
}