ClickHouse CI Engine基于 Webhook、SQS 与 EC2 常驻 Runner 的自托管 CI 编排引擎解析【免费下载链接】ClickHouseClickHouse® is a real-time analytics database management system项目地址: https://gitcode.com/GitHub_Trending/cli/ClickHouse导读本文以 ClickHouse 仓库中独立 CI 引擎Praktika orchestrator的架构说明文档为核心系统讲解这套替代 GitHub Actions 调度、改用 GitHub Webhook 直接驱动、经 SQS 将任务派发到常驻 EC2 Runner 集群的 CI 编排实现。读完本文你将掌握消息从 GitHub Webhook 到 Runner 执行的全链路架构、orchestrator 与 runner 的“镜像固化 随 PR 发布”双层代码拆分、Praktika 运行时base venv的三层解析机制、本地免 AWS 的端到端联调方法、runner 日志排查工具以及基础设施部署与命名规范。概览为什么需要一套独立的 CI EngineClickHouse 的 PR 验证规模极大依赖 GitHub Actions 的调度存在明显局限队列调度不可控、Runner 池自持成本高、工作流策略更新依赖平台机制等。因此仓库在ci/praktika下实现了一套独立的 CI EnginePraktika orchestrator其核心设计决策是用GitHub WebhookHMAC 签名校验直接替代 GitHub Actions 的事件调度用SQS 队列完成“Webhook → 编排器 → Runner 池”的前向任务派发用长寿命 EC2 RunnerASG 托管替代按需拉起的临时 Runner复用克隆结果与共享运行时虚拟环境。这套引擎由 ci/praktika/orchestrator/README.md 定义整体架构配合 PROTOCOL.md 定义通信协议与测试场景以及 ai/DESIGN.md 定义 AI 顾问advisor的插拔式设计。它当前已支撑 ClickHouse 的PR工作流见 ci/workflows/pull_request.py其中namePR、eventWorkflow.Event.PULL_REQUEST。架构一条消息的完整旅程端到端流程README 给出了完整的消息流转链路GitHub webhook | v (HMAC-verified PR event) Lambda (ci/praktika/infrastructure/native/lambda_ci_engine.py) | v (enqueues {type, repo, pr_number, head_sha, ...}) SQS praktika_clickhouse_workflows | v Orchestrator ASG (praktika-workflow-orchestrator) user_data - systemctl enable --now praktika-controller |-- clone the PR head |-- install Praktika into the shared runtime venv if it is not already there |-- subprocess: praktika orchestrate workflow event.json --ci | |-- open per-workflow GitHub check run (PR, in_progress) | |-- find_workflow_for_event | |-- build_job_dag | |-- WorkflowState execution loop: | | for each JobState kicked: | | open per-job GitHub check run (queued - in_progress) | | send {type: job_task, ...} to SQS queue praktika-runs_on | |-- close per-workflow check | v (SQS: one queue per runner pool, named praktika-runs_on) Runner ASGs (e.g. praktika-arm-2xsmall) user_data - systemctl enable --now praktika-controller |-- clone the PR head |-- install Praktika into the shared runtime venv if it is not already there |-- subprocess: praktika orchestrate job task.json --ci | |-- job_runner.run_job(task) - Runner().run(...)关键点是回程通道不走 SQSjob 的完成、存活心跳与取消信号都通过S3回传详见 PROTOCOL.md 的 “Queues and channels”避免为每个 run 创建临时队列。S3 键天然持久化编排器重启后也能恢复在途任务状态。交互图从源码看编排器入口为 ci/praktika/orchestrator/init.py 中的orchestrate()/_orchestrate_event()先通过find_workflows_for_event()匹配事件对应的工作流按事件类型、分支与orchestrator_filter过滤跳过GH_ACTIONS引擎的工作流再由build_job_dag()用 Kahn 拓扑排序将 job 组织成可并行执行的层级levels最后由_drive_dag()循环驱动get_ready() → kick() → wait()。JobState.kick()state.py在就绪时创建 per-job checkqueued然后通过_dispatch()向_queue_for_runs_on()解析出的队列发送job_task。编排器与 Runner 的双层职责组件运行位置SQS / S3 角色LambdaAWS Lambda生产praktika_clickhouse_workflows向 S3 写取消信号OrchestratorEC2 ASGpraktika-workflow-orchestrator×2消费praktika_clickhouse_workflows向praktika-{runner-type}派发任务轮询 S3 获取 job 完成 / 取消 / 存活信号Job runnerEC2 ASGpraktika-{runner-type}如praktika-arm-2xsmall消费praktika-{runner-type}向 S3 写完成与心跳有意为之的代码拆分镜像固化 随 PR 发布这套引擎最独特的工程决策是编排器和 Runner 各自由两块代码组成——一块是固化在 EC2 镜像上的稳定引导包controller bootstrap另一块是随每个 PR 一起发布的编排模块固化在镜像上需 LTASG 重新部署或 wheel 刷新才能变更随每个 PR 发布普通git push即可Workflow 侧praktika-controller—— SQS 轮询、克隆、GH App token、缓存 venv 复用、S3 日志__init__.py::orchestrate、state.pyWorkflowState、JobState、JobCheckRunJob 侧praktika-controller—— SQS 轮询、克隆、GH App token、缓存 venv 复用、S3 日志job_runner.py::run_job将 task 映射到praktika.Runner.run因此调整工作流编排或作业执行策略只需要git push例如job_runner.py的模块头注释明确说明放在 orchestrator 包内意味着 job 执行策略随每个 PR 发布无需 LT/ASG 重部署只有稳定的引导层变更才需要 LT/ASG 重部署或引导 wheel 刷新。代码层面的对应关系Workflow 侧orchestrate()在 ci/praktika/orchestrator/init.py含_orchestrate_single、_orchestrate_resume、_drive_dag等核心函数状态机在state.pyWorkflowState、JobState、JobCheckRun。Job 侧run_job()在 ci/praktika/orchestrator/job_runner.py其核心工作是解析task中的workflow_name/job_name调用workflow.find_jobs()定位 Job然后以local_orchestrator_runTrue调用Runner().run(...)job 结束后把Result经Result.to_dict序列化连同rc、environment快照写入runs/run_id/job/final.jsonS3作为编排器sweep_completions的唯一完成信号。运行时解析Runtime Resolution三层机制Praktika 运行时的选择被拆成三层每一层只负责一件事层组件职责镜像固化Image bakeImageBuilder.Config.prebuilt_venvsci/infrastructure/projects.py在/opt/praktika/base-venvs/name下创建命名 base venv并把共享的 Praktika base wheel 烤进镜像仓库设置ci/settings/settings.py选择共享 base venv启动时 user datarunner/orchestrator 池配置ci/infrastructure/projects.py可选在启动 controller 前把当前 Praktika wheel 强制重装进共享 base venv引导praktika-controller/praktika_controller解析 base venv 并从中运行 Praktika当前唯一的设置项是PRAKTIKA_BASE_VENV两侧workflow 与 job共用。本仓库当前值可在 ci/settings/settings.py 看到PRAKTIKA_BASE_VENV praktika-runtime-0.1.8README 写作时的策略为 base 池固化 Praktika0.1、非 base 池在启动时强制重装0.1.1实际取值以仓库ci/settings/settings.py为准。这一机制的含义镜像提供稳定的 Python / 工具链依赖base 池测试的是烤进镜像的已发布 Praktika 版本非 base 池在启动时把共享 base venv 更新为当前 Praktika wheel。解析流程venv 布局路径创建者用途/opt/praktika/base-venvs/praktika-runtimeImage Builder共享的 workflow/job Python 基础环境运行时依赖、pytest、已发布的 base Praktika wheel从 ci/infrastructure/projects.py 可以看到镜像层如何落地这一点_image_builders()中的PrebuiltVenv(namePRAKTIKA_BASE_VENV, packages[anthropic[bedrock], fpraktika[infrastructure] {_PRAKTIKA_WHL}])把运行时依赖与 AI 顾问 SDK 一并烤进共享 venv池级_pool_user_data()在启动时执行python3.12 -m pip install --ignore-installed controller wheel更新 controller并向 base venv 执行pip install --force-reinstall praktika wheel覆盖烤进镜像的固定版本最后systemctl enable --now praktika-controller。各组件改哪里想要更换预烤共享 base venv、镜像级工具集或非 base 池的启动时 Praktika 覆盖版本 → 改 ci/infrastructure/projects.py想选择不同的共享 base venv → 改 ci/settings/settings.py只有 base-venv 解析逻辑本身需要变化时才动bootstrap/src/praktika_controller/venv_manager.py该引导层固化在镜像上。消息格式与生命周期SQS 前向派发 S3 反向回传两类队列队列用途praktika_clickhouse_workflows工作流触发每个 PR push / rerun 一条消息praktika-{runner-type}编排器向特定 Runner 池派发的 job task所有反向信号——job 完成、存活、取消——都走 S3路径前缀为s3://artifacts-bucket/runs/run_id/新 push 取消则用pr/pr/…。设计要点详见 PROTOCOL.md每个 run 一个 S3 前缀而非一个队列completionsjob/final.json、livenessjob/heartbeat.json、kill 标志cancel都归属同一前缀同 PR 的并发 run 用互不相交的前缀天然无竞争run_id 顶层 check run ID它直接作为 S3 前缀的后缀Lambda 仅凭 webhook 载荷即可定位特定 run 的取消键取消是持久的 S3 写入而非队列路由无需要创建/销毁的资源没有按 run 建的临时队列run 结束后 S3 键作为构建产物自然留存。job_task编排器 → Runner 队列编排器通过JobState.kick()→WorkflowState._dispatch()state.py发送的消息形状{ type: job_task, repo: ClickHouse/clickhouse-private, pr_number: 55743, head_sha: abc123, head_ref: my-branch, base_ref: master, sender: maxknv, title: My PR title, labels: [], workflow_name: PR, job_name: Style check, runs_on: [praktika-arm-2xsmall], cancel_s3_bucket: praktika-artifacts-eu-north-1, cancel_s3_key: runs/72611853552/cancel, heartbeat_s3_bucket: praktika-artifacts-eu-north-1, heartbeat_s3_key: runs/72611853552/Style_check/heartbeat.json, heartbeat_interval_s: 30, final_state_s3_bucket: praktika-artifacts-eu-north-1, final_state_s3_key: runs/72611853552/Style_check/final.json, check_run_id: 72611853552, environment: { WORKFLOW_CONFIG: {}, ... : ... } }要点environment在 run 的第一个 jobConfig Workflow时为null之后则携带上一个 job 序列化的ci/tmp/environment.json传播WORKFLOW_CONFIG、COMMIT_AUTHORS、JOB_KV_DATA等。cancel_s3_*、heartbeat_s3_*、final_state_s3_*三个字段把取消、存活与完成信号收拢到同一个 run 前缀下。job_completionRunner →runs/run_id/job/final.json{ type: job_completion, job_name: Style check, rc: 0, ts: 1704067200.123, repo: ClickHouse/clickhouse-private, pr_number: 55743, head_sha: abc123, workflow_name: PR, instance_id: i-0abc..., details_url: https://.../praktika.html?..., environment: { WORKFLOW_CONFIG: {}, ... : ... }, result: { name: Style check, status: OK, results: [], ... : ... } }该文件由orchestrator/job_runner.py在Runner.run返回后写入幂等可重读JobState.finish对已离开 RUNNING 的 job 是无操作因此迟到的final.json不会破坏 DAG 推进。存活信号与故障恢复S3 通道汇总路径均在s3://artifacts-bucket/下通道方向路径用途Cancel requestLambda → orchestratorruns/run_id/cancel-requestUI 取消按钮sweep_cancel置state.cancelledCancel-beforeLambda → orchestratorspr/pr/cancel-before-scope{ts}新 push 扇出所有event_ts ts的 run 自取消Kill flagOrchestrator → runnersruns/run_id/cancel一旦写入run 内所有运行中 job 杀掉子进程HeartbeatRunner → orchestratorruns/run_id/normalized-job/heartbeat.json周期性{ts, status, instance_id, phase, attempt}attempt为 SQS 接收计数递增即表示重跑Final stateRunner → orchestratorruns/run_id/normalized-job/final.json退出时{rc, environment, ...}心跳相关的超时参数都在 ci/praktika/settings.py 中定义并可被*_overrides.py覆盖HEARTBEAT_INTERVAL_S 30runner 心跳写间隔RUNNER_PICKUP_TIMEOUT_S 3600job 仍为 QUEUED 且从未出现心跳的最大等待覆盖 SQS 等待、ASG 扩容、EC2 启动、controller 启动与首次 pickupHEARTBEAT_STALL_S 300RUNNING 但心跳静默的第一阶段——标记 runner 失联、check 显示待重试但不判失败心跳恢复即解除HEARTBEAT_TIMEOUT_S 900硬超时静默超过即宣布 runner 死亡。设计上HEARTBEAT_TIMEOUT_S必须高于 Runner 队列的visibility_timeout600srunner 中途死亡时其job_task在可见窗口过后重新出现新 Runner 重新执行并向同一批S3 键写心跳首个心跳在 checkout 之前写入以重置last_heartbeat_ts——因此 per-job check 能跨恢复过程保持in_progress只有当重投递也无法产生心跳job 卡死或 SQSmaxReceiveCount耗尽进入死信队列时才判失败。Runner 侧还通过CancelWatchdog线程每 10 秒轮询 kill 标志命中即杀子进程。取消语义触发目标如何到达编排器新 pushsynchronize同一编排器 scope 内该 PR 所有event_ts new event_ts的在途 runLambda 写pr/pr/cancel-before-scope含{ts}scope 内较老的编排器经sweep_cancel自取消UI 取消按钮恰好一个 runLambda 写runs/run_id/cancel-request对应编排器sweep_cancel拾取重跑rerequested—不写取消新 run 有新的 run_id 和新的 S3 前缀取消信号因 S3 的持久性而能在编排器重启后存活——重新上线的编排器会在下一次 sweep 时拾取标志。基础设施故障 vs 真实红构编排器用退出码区分两类失败见INFRA_EXIT_CODE 100ci/praktika/orchestrator/init.py100 根本无法运行工作流AI provider 不可用、计划构建无法触达 S3/GH、token 铸造失败等启动/基础设施故障→ controller 释放消息visibility → 0并自终止ASG 启动新实例重试受PRAKTIKA_INFRA_FAILURE_MAX_RECEIVES默认 3约束1 DAG 已运行、job 真实失败真红构→不重试0 正常完成。_orchestrate_single中启动阶段AI advisor 计划构建最多按Settings.MAX_RETRIES_ORCHESTRATOR默认 3指数退避重试一旦进入running阶段dag_started True则不再重试——重启会导致 job 被重复执行。每一次尝试都会把顶层 check 最终化为failure绝不遗留in_progress并在输出中暴露attempt N/M、编排器实例 ID 与生命周期阶段starting/ai_setup/planning/running/finalizing_check_output()的实现即是对此的体现。局部重跑Partial re-run对单个失败 check 的重跑只重跑该 job 及其失败的下游在原 run 上就地完成不重建整个 DAG自标识 check每个 per-job check 的external_id携带{run_id, job}check_run.rerequestedwebhook 无需查 GitHub API 即可定位持久化状态编排器每轮循环写runs/run_id/state.json含 per-job{status, check_id, rc, …}、累计environment与finalized标志Lambda 读取finalized判断 run 是否结束状态变更将目标 job 及其FAILURE/CANCELLED传递依赖以及Finish Workflow这类终态always_run下游重置为PENDING删除其final.json与heartbeat.json丢弃旧 check 以便kick重新发布新 checkConfig Workflow不重跑复用持久化的WORKFLOW_CONFIG陈旧 head 防护Runner 检出的是实时 PR headrefs/pull/N/head而非 check 的 shaLambda 会重新拉取 PRhead 已推进则拒绝重跑fork PR 的重跑额外要求 maintainer完成握手编排器先写finalizedtrue再做一次sweep_rerun()Lambda 的请求写入发生在finalized读取之前因此请求绝不会丢失。已结束 run 的重跑由resume.lockS3 原子条件创建串行化并发点击只产生一次 resume。调试 Runner 日志fetch_job_log.sh通过 SSM 从在线 Runner 抓取完整的ci-runnersystemd journal并提取特定 job run 的日志先上传 journal 到 S3 以绕过SSM 48 KB 输出上限该脚本路径./ci/praktika/orchestrator/fetch_job_log.sh由 README 文档化引用# 列出当前 runner journal 中出现过的所有 job 名 ./ci/praktika/orchestrator/fetch_job_log.sh --list # 抓取某个 job 最近一次运行 ./ci/praktika/orchestrator/fetch_job_log.sh -j Config Workflow # 抓取倒数第二次运行 ./ci/praktika/orchestrator/fetch_job_log.sh -j Config Workflow -n 2 # 指定具体 runner 实例 ./ci/praktika/orchestrator/fetch_job_log.sh -j Config Workflow -i i-0abc123 # 保存到文件 ./ci/praktika/orchestrator/fetch_job_log.sh -j Config Workflow /tmp/cw.log未指定-i时自动探测正在运行的praktika-arm-2xsmallrunner。前置条件AWS SSM 访问权限以及向clickhouse-test-reports-private的 S3 写入权限。本地联调无需 AWS 的端到端冒烟测试Workflow 侧无需 AWSpraktika orchestrate workflow # 根据 git 状态自动构造事件 praktika orchestrate workflow event.json # 或加载预构造事件 praktika orchestrate workflow --name PR Fast # 只运行匹配的某一个工作流不带--ci时运行在local-orchestrator 模式每个 job 以同步子进程派发praktika orchestrate job task.jsonS3 后端为local-fs。适合在本机做端到端冒烟测试。CLI 参数--event-type、--repo、--head-sha、--head-ref、--base-ref、--pr-number、--sender、--name、--ci、--settings KEYVALUE在 ci/praktika/main.py 中定义--settings可重复传入用于覆盖如AWS_REGION等 Settings 值。事件文件省略时_build_event()会从 git remote /rev-parse HEAD/branch --show-current/gh pr view自动装配事件。Job 侧单 job 沙箱praktika orchestrate job task.json它以local_orchestrator_runTrue调用praktika.Runner.run执行 task 中指定的 job。task JSON 的形状即上文job_task的格式orchestrator 经 SQS 发送的内容。在本地模式run_job()会先写ci/tmp/environment.json_build_ci_environment并在Runner.run后把Result、环境快照等落到本地 S3 后端。命名约定一条规则生成整套资源新增一种 runner 类型只需在 ci/infrastructure/projects.py 中调用一次_runner_infra(name, instance_type)。名称从单一 base 派生资源名称ASGpraktika-{name}LTpraktika-{name}-ltSQS 队列praktika-{name}job 的runs_on[X]路由到队列praktika-X。没有回退——队列不存在时派发失败job 被标记为 FAILURE。从源码看队列名解析在_queue_for_runs_on()state.py跳过无意义的self-hosted标签取第一个有意义的runs_on标签拼成project-slug-label_queue_prefix()使用Settings.PROJECT_SLUG本仓库为clickhouse。kick()派发失败如QueueDoesNotExist时会在 check 输出中给出“Runner pool 未部署请praktika infrastructure --deploy后重跑”的提示。基础设施部署# Workflow 运行时基础设施修改 praktika-controller 或烤制的 orchestrator 镜像后 python3 -m praktika infrastructure --deploy --only LaunchTemplate AutoScalingGroup # 然后终止运行中的 orchestrator 实例让 ASG 按新 LT 重新拉起。 # Job 运行时基础设施新增 runner 类型或修改 praktika-controller / runner 镜像后 python3 -m praktika infrastructure --deploy --only LaunchTemplate AutoScalingGroup SQSQueue # 然后终止受影响池中的运行中 runner。Praktika 的 ASG 部署在部署时将launch_template_version$Latest解析为具体版本号因此 LT 升级也需要 ASG 重新部署——$Latest在运行时不被认可。当前能力边界与路线图README 明确列出了已实现的能力Webhook → Lambda → SQS → orchestrator pickup → clone → orchestrate 全链路Per-workflowPRGitHub check输出完整执行计划、编排器实例 ID、生命周期阶段与attempt N/M即使启动崩溃也总是被 finalize绝不遗留in_progress基础设施故障处理启动/基础设施崩溃以INFRA_EXIT_CODE退出controller 在全新实例上重试受 SQS 接收计数约束真实红构rc1不重试Per-job GitHub check计划时queued、kick 时in_progress、完成时success/skippedDAG 感知的执行循环WorkflowState.get_ready/kick/wait从JobState.kick()到按类型 Runner 队列的 SQS 派发一个已部署的 Runner 池praktika-arm-2xsmall本地沙箱praktika orchestrate job task.json可经praktika.Runner.run与 local-fs S3 后端端到端跑真实 job。同时列出了TODO即当前未完成项引用时需注意完成路径completion path当前编排器在派发瞬间就把 job 桩定为 SUCCESS真正的“runner 回传 done 事件 →wait()长轮询 → 翻转JobState”尚未落地运行Config Workflow做 job 过滤文件变更 / 缓存命中通过s3://clickhouse-builds/PRs/pr/workflow/job/...的制品流长寿命 runner 上的工作区清理git clean -ffdx、缓存重置孤儿 runner 清扫 Lambda强制标签 定期终止job 完成后的自终止Runner.run中的 EXIT trap按 runner 类型的扩缩容队列深度 → ASG target tracking——当前所有池固定为 1。延伸阅读仓库内证据链编排协议与测试场景ci/praktika/orchestrator/PROTOCOL.md编排器主实现ci/praktika/orchestrator/init.py状态机与 S3 sweepci/praktika/orchestrator/state.pyjob 执行入口ci/praktika/orchestrator/job_runner.pyAI 顾问设计ci/praktika/orchestrator/ai/DESIGN.md全局设置默认值ci/praktika/settings.py仓库级设置PRAKTIKA_BASE_VENV、Runner 标签、S3 bucket 等ci/settings/settings.py基础设施配置镜像烤制、池 user dataci/infrastructure/projects.pyCLI 入口orchestrate/infrastructure子命令ci/praktika/main.pyPR工作流定义ci/workflows/pull_request.py综上所述ClickHouse 的这套 CI Engine 用“Webhook SQS 常驻 EC2 Runner S3 回传”的组合将 CI 调度与执行完全收归自持基础设施同时通过“镜像固化稳定引导层、随 PR 发布策略层”的拆分让工作流与作业策略的迭代保持在git push的速度。其消息协议、存活探测与取消语义经过 PROTOCOL.md 的 13 个测试场景系统化验证是一套值得借鉴的自托管 CI 架构范本。【免费下载链接】ClickHouseClickHouse® is a real-time analytics database management system项目地址: https://gitcode.com/GitHub_Trending/cli/ClickHouse创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ClickHouse CI Engine：基于 Webhook、SQS 与 EC2 常驻 Runner 的自托管 CI 编排引擎解析
ClickHouse CI Engine基于 Webhook、SQS 与 EC2 常驻 Runner 的自托管 CI 编排引擎解析【免费下载链接】ClickHouseClickHouse® is a real-time analytics database management system项目地址: https://gitcode.com/GitHub_Trending/cli/ClickHouse导读本文以 ClickHouse 仓库中独立 CI 引擎Praktika orchestrator的架构说明文档为核心系统讲解这套替代 GitHub Actions 调度、改用 GitHub Webhook 直接驱动、经 SQS 将任务派发到常驻 EC2 Runner 集群的 CI 编排实现。读完本文你将掌握消息从 GitHub Webhook 到 Runner 执行的全链路架构、orchestrator 与 runner 的“镜像固化 随 PR 发布”双层代码拆分、Praktika 运行时base venv的三层解析机制、本地免 AWS 的端到端联调方法、runner 日志排查工具以及基础设施部署与命名规范。概览为什么需要一套独立的 CI EngineClickHouse 的 PR 验证规模极大依赖 GitHub Actions 的调度存在明显局限队列调度不可控、Runner 池自持成本高、工作流策略更新依赖平台机制等。因此仓库在ci/praktika下实现了一套独立的 CI EnginePraktika orchestrator其核心设计决策是用GitHub WebhookHMAC 签名校验直接替代 GitHub Actions 的事件调度用SQS 队列完成“Webhook → 编排器 → Runner 池”的前向任务派发用长寿命 EC2 RunnerASG 托管替代按需拉起的临时 Runner复用克隆结果与共享运行时虚拟环境。这套引擎由 ci/praktika/orchestrator/README.md 定义整体架构配合 PROTOCOL.md 定义通信协议与测试场景以及 ai/DESIGN.md 定义 AI 顾问advisor的插拔式设计。它当前已支撑 ClickHouse 的PR工作流见 ci/workflows/pull_request.py其中namePR、eventWorkflow.Event.PULL_REQUEST。架构一条消息的完整旅程端到端流程README 给出了完整的消息流转链路GitHub webhook | v (HMAC-verified PR event) Lambda (ci/praktika/infrastructure/native/lambda_ci_engine.py) | v (enqueues {type, repo, pr_number, head_sha, ...}) SQS praktika_clickhouse_workflows | v Orchestrator ASG (praktika-workflow-orchestrator) user_data - systemctl enable --now praktika-controller |-- clone the PR head |-- install Praktika into the shared runtime venv if it is not already there |-- subprocess: praktika orchestrate workflow event.json --ci | |-- open per-workflow GitHub check run (PR, in_progress) | |-- find_workflow_for_event | |-- build_job_dag | |-- WorkflowState execution loop: | | for each JobState kicked: | | open per-job GitHub check run (queued - in_progress) | | send {type: job_task, ...} to SQS queue praktika-runs_on | |-- close per-workflow check | v (SQS: one queue per runner pool, named praktika-runs_on) Runner ASGs (e.g. praktika-arm-2xsmall) user_data - systemctl enable --now praktika-controller |-- clone the PR head |-- install Praktika into the shared runtime venv if it is not already there |-- subprocess: praktika orchestrate job task.json --ci | |-- job_runner.run_job(task) - Runner().run(...)关键点是回程通道不走 SQSjob 的完成、存活心跳与取消信号都通过S3回传详见 PROTOCOL.md 的 “Queues and channels”避免为每个 run 创建临时队列。S3 键天然持久化编排器重启后也能恢复在途任务状态。交互图从源码看编排器入口为 ci/praktika/orchestrator/init.py 中的orchestrate()/_orchestrate_event()先通过find_workflows_for_event()匹配事件对应的工作流按事件类型、分支与orchestrator_filter过滤跳过GH_ACTIONS引擎的工作流再由build_job_dag()用 Kahn 拓扑排序将 job 组织成可并行执行的层级levels最后由_drive_dag()循环驱动get_ready() → kick() → wait()。JobState.kick()state.py在就绪时创建 per-job checkqueued然后通过_dispatch()向_queue_for_runs_on()解析出的队列发送job_task。编排器与 Runner 的双层职责组件运行位置SQS / S3 角色LambdaAWS Lambda生产praktika_clickhouse_workflows向 S3 写取消信号OrchestratorEC2 ASGpraktika-workflow-orchestrator×2消费praktika_clickhouse_workflows向praktika-{runner-type}派发任务轮询 S3 获取 job 完成 / 取消 / 存活信号Job runnerEC2 ASGpraktika-{runner-type}如praktika-arm-2xsmall消费praktika-{runner-type}向 S3 写完成与心跳有意为之的代码拆分镜像固化 随 PR 发布这套引擎最独特的工程决策是编排器和 Runner 各自由两块代码组成——一块是固化在 EC2 镜像上的稳定引导包controller bootstrap另一块是随每个 PR 一起发布的编排模块固化在镜像上需 LTASG 重新部署或 wheel 刷新才能变更随每个 PR 发布普通git push即可Workflow 侧praktika-controller—— SQS 轮询、克隆、GH App token、缓存 venv 复用、S3 日志__init__.py::orchestrate、state.pyWorkflowState、JobState、JobCheckRunJob 侧praktika-controller—— SQS 轮询、克隆、GH App token、缓存 venv 复用、S3 日志job_runner.py::run_job将 task 映射到praktika.Runner.run因此调整工作流编排或作业执行策略只需要git push例如job_runner.py的模块头注释明确说明放在 orchestrator 包内意味着 job 执行策略随每个 PR 发布无需 LT/ASG 重部署只有稳定的引导层变更才需要 LT/ASG 重部署或引导 wheel 刷新。代码层面的对应关系Workflow 侧orchestrate()在 ci/praktika/orchestrator/init.py含_orchestrate_single、_orchestrate_resume、_drive_dag等核心函数状态机在state.pyWorkflowState、JobState、JobCheckRun。Job 侧run_job()在 ci/praktika/orchestrator/job_runner.py其核心工作是解析task中的workflow_name/job_name调用workflow.find_jobs()定位 Job然后以local_orchestrator_runTrue调用Runner().run(...)job 结束后把Result经Result.to_dict序列化连同rc、environment快照写入runs/run_id/job/final.jsonS3作为编排器sweep_completions的唯一完成信号。运行时解析Runtime Resolution三层机制Praktika 运行时的选择被拆成三层每一层只负责一件事层组件职责镜像固化Image bakeImageBuilder.Config.prebuilt_venvsci/infrastructure/projects.py在/opt/praktika/base-venvs/name下创建命名 base venv并把共享的 Praktika base wheel 烤进镜像仓库设置ci/settings/settings.py选择共享 base venv启动时 user datarunner/orchestrator 池配置ci/infrastructure/projects.py可选在启动 controller 前把当前 Praktika wheel 强制重装进共享 base venv引导praktika-controller/praktika_controller解析 base venv 并从中运行 Praktika当前唯一的设置项是PRAKTIKA_BASE_VENV两侧workflow 与 job共用。本仓库当前值可在 ci/settings/settings.py 看到PRAKTIKA_BASE_VENV praktika-runtime-0.1.8README 写作时的策略为 base 池固化 Praktika0.1、非 base 池在启动时强制重装0.1.1实际取值以仓库ci/settings/settings.py为准。这一机制的含义镜像提供稳定的 Python / 工具链依赖base 池测试的是烤进镜像的已发布 Praktika 版本非 base 池在启动时把共享 base venv 更新为当前 Praktika wheel。解析流程venv 布局路径创建者用途/opt/praktika/base-venvs/praktika-runtimeImage Builder共享的 workflow/job Python 基础环境运行时依赖、pytest、已发布的 base Praktika wheel从 ci/infrastructure/projects.py 可以看到镜像层如何落地这一点_image_builders()中的PrebuiltVenv(namePRAKTIKA_BASE_VENV, packages[anthropic[bedrock], fpraktika[infrastructure] {_PRAKTIKA_WHL}])把运行时依赖与 AI 顾问 SDK 一并烤进共享 venv池级_pool_user_data()在启动时执行python3.12 -m pip install --ignore-installed controller wheel更新 controller并向 base venv 执行pip install --force-reinstall praktika wheel覆盖烤进镜像的固定版本最后systemctl enable --now praktika-controller。各组件改哪里想要更换预烤共享 base venv、镜像级工具集或非 base 池的启动时 Praktika 覆盖版本 → 改 ci/infrastructure/projects.py想选择不同的共享 base venv → 改 ci/settings/settings.py只有 base-venv 解析逻辑本身需要变化时才动bootstrap/src/praktika_controller/venv_manager.py该引导层固化在镜像上。消息格式与生命周期SQS 前向派发 S3 反向回传两类队列队列用途praktika_clickhouse_workflows工作流触发每个 PR push / rerun 一条消息praktika-{runner-type}编排器向特定 Runner 池派发的 job task所有反向信号——job 完成、存活、取消——都走 S3路径前缀为s3://artifacts-bucket/runs/run_id/新 push 取消则用pr/pr/…。设计要点详见 PROTOCOL.md每个 run 一个 S3 前缀而非一个队列completionsjob/final.json、livenessjob/heartbeat.json、kill 标志cancel都归属同一前缀同 PR 的并发 run 用互不相交的前缀天然无竞争run_id 顶层 check run ID它直接作为 S3 前缀的后缀Lambda 仅凭 webhook 载荷即可定位特定 run 的取消键取消是持久的 S3 写入而非队列路由无需要创建/销毁的资源没有按 run 建的临时队列run 结束后 S3 键作为构建产物自然留存。job_task编排器 → Runner 队列编排器通过JobState.kick()→WorkflowState._dispatch()state.py发送的消息形状{ type: job_task, repo: ClickHouse/clickhouse-private, pr_number: 55743, head_sha: abc123, head_ref: my-branch, base_ref: master, sender: maxknv, title: My PR title, labels: [], workflow_name: PR, job_name: Style check, runs_on: [praktika-arm-2xsmall], cancel_s3_bucket: praktika-artifacts-eu-north-1, cancel_s3_key: runs/72611853552/cancel, heartbeat_s3_bucket: praktika-artifacts-eu-north-1, heartbeat_s3_key: runs/72611853552/Style_check/heartbeat.json, heartbeat_interval_s: 30, final_state_s3_bucket: praktika-artifacts-eu-north-1, final_state_s3_key: runs/72611853552/Style_check/final.json, check_run_id: 72611853552, environment: { WORKFLOW_CONFIG: {}, ... : ... } }要点environment在 run 的第一个 jobConfig Workflow时为null之后则携带上一个 job 序列化的ci/tmp/environment.json传播WORKFLOW_CONFIG、COMMIT_AUTHORS、JOB_KV_DATA等。cancel_s3_*、heartbeat_s3_*、final_state_s3_*三个字段把取消、存活与完成信号收拢到同一个 run 前缀下。job_completionRunner →runs/run_id/job/final.json{ type: job_completion, job_name: Style check, rc: 0, ts: 1704067200.123, repo: ClickHouse/clickhouse-private, pr_number: 55743, head_sha: abc123, workflow_name: PR, instance_id: i-0abc..., details_url: https://.../praktika.html?..., environment: { WORKFLOW_CONFIG: {}, ... : ... }, result: { name: Style check, status: OK, results: [], ... : ... } }该文件由orchestrator/job_runner.py在Runner.run返回后写入幂等可重读JobState.finish对已离开 RUNNING 的 job 是无操作因此迟到的final.json不会破坏 DAG 推进。存活信号与故障恢复S3 通道汇总路径均在s3://artifacts-bucket/下通道方向路径用途Cancel requestLambda → orchestratorruns/run_id/cancel-requestUI 取消按钮sweep_cancel置state.cancelledCancel-beforeLambda → orchestratorspr/pr/cancel-before-scope{ts}新 push 扇出所有event_ts ts的 run 自取消Kill flagOrchestrator → runnersruns/run_id/cancel一旦写入run 内所有运行中 job 杀掉子进程HeartbeatRunner → orchestratorruns/run_id/normalized-job/heartbeat.json周期性{ts, status, instance_id, phase, attempt}attempt为 SQS 接收计数递增即表示重跑Final stateRunner → orchestratorruns/run_id/normalized-job/final.json退出时{rc, environment, ...}心跳相关的超时参数都在 ci/praktika/settings.py 中定义并可被*_overrides.py覆盖HEARTBEAT_INTERVAL_S 30runner 心跳写间隔RUNNER_PICKUP_TIMEOUT_S 3600job 仍为 QUEUED 且从未出现心跳的最大等待覆盖 SQS 等待、ASG 扩容、EC2 启动、controller 启动与首次 pickupHEARTBEAT_STALL_S 300RUNNING 但心跳静默的第一阶段——标记 runner 失联、check 显示待重试但不判失败心跳恢复即解除HEARTBEAT_TIMEOUT_S 900硬超时静默超过即宣布 runner 死亡。设计上HEARTBEAT_TIMEOUT_S必须高于 Runner 队列的visibility_timeout600srunner 中途死亡时其job_task在可见窗口过后重新出现新 Runner 重新执行并向同一批S3 键写心跳首个心跳在 checkout 之前写入以重置last_heartbeat_ts——因此 per-job check 能跨恢复过程保持in_progress只有当重投递也无法产生心跳job 卡死或 SQSmaxReceiveCount耗尽进入死信队列时才判失败。Runner 侧还通过CancelWatchdog线程每 10 秒轮询 kill 标志命中即杀子进程。取消语义触发目标如何到达编排器新 pushsynchronize同一编排器 scope 内该 PR 所有event_ts new event_ts的在途 runLambda 写pr/pr/cancel-before-scope含{ts}scope 内较老的编排器经sweep_cancel自取消UI 取消按钮恰好一个 runLambda 写runs/run_id/cancel-request对应编排器sweep_cancel拾取重跑rerequested—不写取消新 run 有新的 run_id 和新的 S3 前缀取消信号因 S3 的持久性而能在编排器重启后存活——重新上线的编排器会在下一次 sweep 时拾取标志。基础设施故障 vs 真实红构编排器用退出码区分两类失败见INFRA_EXIT_CODE 100ci/praktika/orchestrator/init.py100 根本无法运行工作流AI provider 不可用、计划构建无法触达 S3/GH、token 铸造失败等启动/基础设施故障→ controller 释放消息visibility → 0并自终止ASG 启动新实例重试受PRAKTIKA_INFRA_FAILURE_MAX_RECEIVES默认 3约束1 DAG 已运行、job 真实失败真红构→不重试0 正常完成。_orchestrate_single中启动阶段AI advisor 计划构建最多按Settings.MAX_RETRIES_ORCHESTRATOR默认 3指数退避重试一旦进入running阶段dag_started True则不再重试——重启会导致 job 被重复执行。每一次尝试都会把顶层 check 最终化为failure绝不遗留in_progress并在输出中暴露attempt N/M、编排器实例 ID 与生命周期阶段starting/ai_setup/planning/running/finalizing_check_output()的实现即是对此的体现。局部重跑Partial re-run对单个失败 check 的重跑只重跑该 job 及其失败的下游在原 run 上就地完成不重建整个 DAG自标识 check每个 per-job check 的external_id携带{run_id, job}check_run.rerequestedwebhook 无需查 GitHub API 即可定位持久化状态编排器每轮循环写runs/run_id/state.json含 per-job{status, check_id, rc, …}、累计environment与finalized标志Lambda 读取finalized判断 run 是否结束状态变更将目标 job 及其FAILURE/CANCELLED传递依赖以及Finish Workflow这类终态always_run下游重置为PENDING删除其final.json与heartbeat.json丢弃旧 check 以便kick重新发布新 checkConfig Workflow不重跑复用持久化的WORKFLOW_CONFIG陈旧 head 防护Runner 检出的是实时 PR headrefs/pull/N/head而非 check 的 shaLambda 会重新拉取 PRhead 已推进则拒绝重跑fork PR 的重跑额外要求 maintainer完成握手编排器先写finalizedtrue再做一次sweep_rerun()Lambda 的请求写入发生在finalized读取之前因此请求绝不会丢失。已结束 run 的重跑由resume.lockS3 原子条件创建串行化并发点击只产生一次 resume。调试 Runner 日志fetch_job_log.sh通过 SSM 从在线 Runner 抓取完整的ci-runnersystemd journal并提取特定 job run 的日志先上传 journal 到 S3 以绕过SSM 48 KB 输出上限该脚本路径./ci/praktika/orchestrator/fetch_job_log.sh由 README 文档化引用# 列出当前 runner journal 中出现过的所有 job 名 ./ci/praktika/orchestrator/fetch_job_log.sh --list # 抓取某个 job 最近一次运行 ./ci/praktika/orchestrator/fetch_job_log.sh -j Config Workflow # 抓取倒数第二次运行 ./ci/praktika/orchestrator/fetch_job_log.sh -j Config Workflow -n 2 # 指定具体 runner 实例 ./ci/praktika/orchestrator/fetch_job_log.sh -j Config Workflow -i i-0abc123 # 保存到文件 ./ci/praktika/orchestrator/fetch_job_log.sh -j Config Workflow /tmp/cw.log未指定-i时自动探测正在运行的praktika-arm-2xsmallrunner。前置条件AWS SSM 访问权限以及向clickhouse-test-reports-private的 S3 写入权限。本地联调无需 AWS 的端到端冒烟测试Workflow 侧无需 AWSpraktika orchestrate workflow # 根据 git 状态自动构造事件 praktika orchestrate workflow event.json # 或加载预构造事件 praktika orchestrate workflow --name PR Fast # 只运行匹配的某一个工作流不带--ci时运行在local-orchestrator 模式每个 job 以同步子进程派发praktika orchestrate job task.jsonS3 后端为local-fs。适合在本机做端到端冒烟测试。CLI 参数--event-type、--repo、--head-sha、--head-ref、--base-ref、--pr-number、--sender、--name、--ci、--settings KEYVALUE在 ci/praktika/main.py 中定义--settings可重复传入用于覆盖如AWS_REGION等 Settings 值。事件文件省略时_build_event()会从 git remote /rev-parse HEAD/branch --show-current/gh pr view自动装配事件。Job 侧单 job 沙箱praktika orchestrate job task.json它以local_orchestrator_runTrue调用praktika.Runner.run执行 task 中指定的 job。task JSON 的形状即上文job_task的格式orchestrator 经 SQS 发送的内容。在本地模式run_job()会先写ci/tmp/environment.json_build_ci_environment并在Runner.run后把Result、环境快照等落到本地 S3 后端。命名约定一条规则生成整套资源新增一种 runner 类型只需在 ci/infrastructure/projects.py 中调用一次_runner_infra(name, instance_type)。名称从单一 base 派生资源名称ASGpraktika-{name}LTpraktika-{name}-ltSQS 队列praktika-{name}job 的runs_on[X]路由到队列praktika-X。没有回退——队列不存在时派发失败job 被标记为 FAILURE。从源码看队列名解析在_queue_for_runs_on()state.py跳过无意义的self-hosted标签取第一个有意义的runs_on标签拼成project-slug-label_queue_prefix()使用Settings.PROJECT_SLUG本仓库为clickhouse。kick()派发失败如QueueDoesNotExist时会在 check 输出中给出“Runner pool 未部署请praktika infrastructure --deploy后重跑”的提示。基础设施部署# Workflow 运行时基础设施修改 praktika-controller 或烤制的 orchestrator 镜像后 python3 -m praktika infrastructure --deploy --only LaunchTemplate AutoScalingGroup # 然后终止运行中的 orchestrator 实例让 ASG 按新 LT 重新拉起。 # Job 运行时基础设施新增 runner 类型或修改 praktika-controller / runner 镜像后 python3 -m praktika infrastructure --deploy --only LaunchTemplate AutoScalingGroup SQSQueue # 然后终止受影响池中的运行中 runner。Praktika 的 ASG 部署在部署时将launch_template_version$Latest解析为具体版本号因此 LT 升级也需要 ASG 重新部署——$Latest在运行时不被认可。当前能力边界与路线图README 明确列出了已实现的能力Webhook → Lambda → SQS → orchestrator pickup → clone → orchestrate 全链路Per-workflowPRGitHub check输出完整执行计划、编排器实例 ID、生命周期阶段与attempt N/M即使启动崩溃也总是被 finalize绝不遗留in_progress基础设施故障处理启动/基础设施崩溃以INFRA_EXIT_CODE退出controller 在全新实例上重试受 SQS 接收计数约束真实红构rc1不重试Per-job GitHub check计划时queued、kick 时in_progress、完成时success/skippedDAG 感知的执行循环WorkflowState.get_ready/kick/wait从JobState.kick()到按类型 Runner 队列的 SQS 派发一个已部署的 Runner 池praktika-arm-2xsmall本地沙箱praktika orchestrate job task.json可经praktika.Runner.run与 local-fs S3 后端端到端跑真实 job。同时列出了TODO即当前未完成项引用时需注意完成路径completion path当前编排器在派发瞬间就把 job 桩定为 SUCCESS真正的“runner 回传 done 事件 →wait()长轮询 → 翻转JobState”尚未落地运行Config Workflow做 job 过滤文件变更 / 缓存命中通过s3://clickhouse-builds/PRs/pr/workflow/job/...的制品流长寿命 runner 上的工作区清理git clean -ffdx、缓存重置孤儿 runner 清扫 Lambda强制标签 定期终止job 完成后的自终止Runner.run中的 EXIT trap按 runner 类型的扩缩容队列深度 → ASG target tracking——当前所有池固定为 1。延伸阅读仓库内证据链编排协议与测试场景ci/praktika/orchestrator/PROTOCOL.md编排器主实现ci/praktika/orchestrator/init.py状态机与 S3 sweepci/praktika/orchestrator/state.pyjob 执行入口ci/praktika/orchestrator/job_runner.pyAI 顾问设计ci/praktika/orchestrator/ai/DESIGN.md全局设置默认值ci/praktika/settings.py仓库级设置PRAKTIKA_BASE_VENV、Runner 标签、S3 bucket 等ci/settings/settings.py基础设施配置镜像烤制、池 user dataci/infrastructure/projects.pyCLI 入口orchestrate/infrastructure子命令ci/praktika/main.pyPR工作流定义ci/workflows/pull_request.py综上所述ClickHouse 的这套 CI Engine 用“Webhook SQS 常驻 EC2 Runner S3 回传”的组合将 CI 调度与执行完全收归自持基础设施同时通过“镜像固化稳定引导层、随 PR 发布策略层”的拆分让工作流与作业策略的迭代保持在git push的速度。其消息协议、存活探测与取消语义经过 PROTOCOL.md 的 13 个测试场景系统化验证是一套值得借鉴的自托管 CI 架构范本。【免费下载链接】ClickHouseClickHouse® is a real-time analytics database management system项目地址: https://gitcode.com/GitHub_Trending/cli/ClickHouse创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考