Compare commits

...
Author SHA1 Message Date
panxiao81 d82687382a fix: 编码 OpenSandbox assignment metadata
test / go (pull_request) Successful in 4m45s
test / shell (pull_request) Failing after 13m27s
test / python (pull_request) Failing after 13m27s
2026-09-21 07:48:10 +00:00
panxiao81 7f52cf393f Merge Pod 与 VM 独立执行容量池
publish images / publish-images (push) Failing after 11m8s
test / python (push) Successful in 10s
test / shell (push) Successful in 26s
test / go (push) Successful in 5m51s
2026-09-21 07:26:10 +00:00
panxiao81 fadc93a0bf feat: 为执行后端增加独立容量池
test / python (pull_request) Successful in 13s
test / shell (pull_request) Successful in 20s
test / go (pull_request) Successful in 5m54s
2026-09-21 07:21:38 +00:00
panxiao81 8a186dcd86 Merge Runner 无中断恢复与容量隔离
test / python (push) Successful in 25s
test / shell (push) Successful in 29s
test / go (push) Successful in 2m28s
publish images / publish-images (push) Failing after 11m13s
2026-09-21 06:48:11 +00:00
panxiao81 d06845bc3c fix: 隔离 Pod 与 VM 并发容量
test / python (pull_request) Successful in 20s
test / shell (pull_request) Successful in 23s
test / go (pull_request) Successful in 3m46s
2026-09-21 06:44:34 +00:00
panxiao81 2a23f0c62e feat: 通过 Lease 单例调度任务
test / python (pull_request) Successful in 31s
test / shell (pull_request) Successful in 34s
test / go (pull_request) Successful in 3m38s
2026-09-21 06:35:06 +00:00
panxiao81 5e94182308 fix: 从后端恢复 Runner claim
test / python (pull_request) Successful in 9s
test / shell (pull_request) Successful in 15s
test / go (pull_request) Successful in 2m18s
2026-09-21 06:18:23 +00:00
panxiao81 327c73e744 Merge Go Runner 与 OpenSandbox VM 集成
test / python (push) Successful in 15s
test / shell (push) Successful in 21s
publish images / publish-images (push) Failing after 13m58s
test / go (push) Failing after 15m46s
2026-09-21 06:11:40 +00:00
panxiao81 6bb654a4b7 feat: 持久化清理 VM 执行器
test / python (pull_request) Successful in 15s
test / shell (pull_request) Successful in 21s
test / go (pull_request) Successful in 2m57s
2026-09-21 06:00:53 +00:00
panxiao81 6a4e061815 fix: 重写 Gitea 反代 Host 2026-09-21 05:39:23 +00:00
panxiao81 14c431031d fix: 代理 Runner 仓库 HTTP 流量 2026-09-21 05:27:52 +00:00
panxiao81 be697a61cf feat: 自动清理终态 Pod 执行器 2026-09-21 05:24:38 +00:00
panxiao81 2036584e80 fix: 等待 Runner facade 就绪 2026-09-20 21:34:03 +00:00
panxiao81 44909c2ee1 fix: 绑定 Pod 到真实 SPIRE agent 2026-09-20 21:27:48 +00:00
panxiao81 a53483f2f0 fix: 注入执行器 SPIFFE socket 2026-09-20 21:10:58 +00:00
panxiao81 cbbe14d6b5 fix: gate scheduler task concurrency 2026-09-20 20:53:58 +00:00
panxiao81 bca8096491 fix: preserve split NATS permissions 2026-09-20 20:41:16 +00:00
panxiao81 9f1181a536 ci: publish Go controller source changes 2026-09-20 20:29:49 +00:00
panxiao81 e271537760 feat: assemble Go scheduler and backend workers 2026-09-20 20:24:45 +00:00
panxiao81 27599631f1 feat: run official executor behind SPIFFE mTLS 2026-09-20 20:12:03 +00:00
panxiao81 8b77b4be63 feat: bootstrap official runner through SPIFFE facade 2026-09-20 20:00:34 +00:00
panxiao81 5fdd39f7ff 实现预分配 RunnerService facade 2026-09-20 19:51:32 +00:00
panxiao81 b77245e4b9 实现 Gitea assignment 调度循环 2026-09-20 19:19:11 +00:00
panxiao81 a856176c56 收窄 assignment 队列职责 2026-09-20 19:06:04 +00:00
panxiao81 27787d7899 实现 durable assignment 消费循环 2026-09-20 18:57:50 +00:00
panxiao81 bfb939ca7a 建立单进程组件运行时 2026-09-20 18:50:19 +00:00
panxiao81 67607391ff 接入 OpenSandbox 官方 Go SDK 2026-09-20 18:37:08 +00:00
panxiao81 fccaf319ac 实现 client-go Pod backend 2026-09-20 18:29:26 +00:00
panxiao81 0d8b36380d 实现持久化 assignment handoff 2026-09-20 18:20:22 +00:00
panxiao81 22dc3fef63 建立无状态 worker reconcile 边界 2026-09-20 18:11:57 +00:00
panxiao81 3f336310af 固定 workflow job key 身份契约 2026-09-20 17:59:59 +00:00
panxiao81 3c1fca1832 建立 Go Task Scheduler 协议骨架 2026-09-20 17:52:38 +00:00
panxiao81 5e5f2cc48d Merge pull request '恢复原生 Kubernetes Pod 调度' (#29) from fix/runner-workspace-permissions into main
test / python (push) Successful in 9s
test / shell (push) Successful in 24s
publish images / publish-images (push) Failing after 1s
2026-09-20 17:13:43 +00:00
panxiao81 c72777a479 恢复原生 Kubernetes Pod 调度
test / python (pull_request) Successful in 11s
test / shell (pull_request) Successful in 17s
2026-09-20 17:08:51 +00:00
panxiao81 8a332a959d 合并 completed 消息 ACK 修复
test / python (push) Successful in 17s
test / shell (push) Successful in 21s
publish images / publish-images (push) Canceled after 0s
2026-09-18 19:45:54 +00:00
panxiao81 65587b940b 确认完成事件后 ACK 持久消息
test / python (pull_request) Successful in 17s
test / shell (pull_request) Successful in 22s
2026-09-18 19:44:01 +00:00
panxiao81 df9c7ab785 合并持久事件驱动 OpenSandbox 调度
test / python (push) Successful in 14s
test / shell (push) Successful in 20s
publish images / publish-images (push) Canceled after 0s
2026-09-18 19:34:24 +00:00
panxiao81 5348ca9891 将 OpenSandbox 调度改为持久事件循环
test / shell (pull_request) Successful in 14s
test / python (pull_request) Successful in 10s
2026-09-18 19:25:09 +00:00
panxiao81 0890bc76d9 Merge pull request '按 guest Agent 格式恢复 SPIRE 父身份' (#26) from fix/opensandbox-guest-parent into main
test / python (push) Successful in 21s
test / shell (push) Successful in 26s
publish images / publish-images (push) Canceled after 0s
2026-09-18 18:56:50 +00:00
panxiao81 fe0703fe5c 按 guest Agent 格式恢复 SPIRE 父身份
test / python (pull_request) Successful in 27s
test / shell (pull_request) Successful in 29s
2026-09-18 18:55:49 +00:00
panxiao81 66f7d797f3 Merge pull request '修正 OpenSandbox SPIRE 父身份格式' (#25) from fix/opensandbox-spire-parent into main
test / shell (push) Successful in 24s
test / python (push) Successful in 17s
publish images / publish-images (push) Failing after 1s
2026-09-18 18:29:48 +00:00
panxiao81 ab8fbad310 修正 OpenSandbox SPIRE 父身份格式
test / python (pull_request) Successful in 24s
test / shell (pull_request) Successful in 39s
2026-09-18 18:28:32 +00:00
panxiao81 58b48207fe Merge pull request '重构为直接 OpenSandbox 生命周期调度' (#24) from refactor/direct-opensandbox-scheduler into main
test / python (push) Successful in 11s
test / shell (push) Successful in 16s
publish images / publish-images (push) Failing after 1s
2026-09-18 18:19:38 +00:00
panxiao81 f756a83013 修复 runner 清理脚本检查
test / shell (pull_request) Successful in 18s
test / python (pull_request) Successful in 15s
2026-09-18 18:18:10 +00:00
panxiao81 523ccbae2f 修复 runner token 清理路径
test / python (pull_request) Successful in 15s
test / shell (pull_request) Failing after 17s
2026-09-18 18:13:48 +00:00
panxiao81 f3a199e7ba 重构为直接 OpenSandbox 生命周期调度
test / python (pull_request) Successful in 13s
test / shell (pull_request) Failing after 21s
2026-09-18 18:12:10 +00:00
panxiao81 6a58c95c5c Merge pull request '接入 OpenSandbox Kata Runner 控制面' (#23) from feat/opensandbox-runner into main
test / python (push) Successful in 13s
test / shell (push) Successful in 22s
publish images / publish-images (push) Failing after 0s
2026-09-18 16:22:58 +00:00
panxiao81 2679f81cdd fix: 使用 OpenSandbox chart 服务地址
test / python (pull_request) Successful in 9s
test / shell (pull_request) Successful in 17s
2026-09-18 01:09:37 +00:00
panxiao81 d7ba64d6f3 接入 OpenSandbox Kata Runner 控制面
test / python (pull_request) Successful in 9s
test / shell (pull_request) Successful in 21s
2026-09-18 00:56:10 +00:00
panxiao81 8d70df2128 Merge pull request '在 VM Runner 中构建发布镜像' (#17) from refactor/publish-on-vm into main
test / python (push) Successful in 10s
test / shell (push) Successful in 15s
publish images / publish-images (push) Failing after 0s
Reviewed-on: #17
2026-09-16 18:29:26 +00:00
panxiao81 3e69d102dd 在 VM Runner 中发布镜像
test / python (pull_request) Successful in 14s
test / shell (pull_request) Successful in 22s
2026-09-16 18:28:33 +00:00
panxiao81 0387b2592a Merge pull request '使用 native snapshotter 发布镜像证明' (#16) from fix/publish-native-snapshotter into main
test / python (push) Successful in 10s
test / shell (push) Successful in 16s
publish images / publish-images (push) Canceled after 9m5s
Reviewed-on: #16
2026-09-16 18:20:10 +00:00
panxiao81 057731ae62 使用 native snapshotter 生成镜像证明
test / python (pull_request) Successful in 8s
test / shell (pull_request) Successful in 14s
2026-09-16 18:19:27 +00:00
panxiao81 529daa4c75 Merge pull request '修复 Pod 内嵌套 Docker overlay 构建' (#15) from fix/publish-vfs-builder into main
test / python (push) Successful in 9s
publish images / publish-images (push) Failing after 35s
test / shell (push) Successful in 16s
Reviewed-on: #15
2026-09-16 18:17:24 +00:00
panxiao81 38946e8def 使用 vfs 构建嵌套容器镜像
test / shell (pull_request) Successful in 20s
test / python (pull_request) Successful in 9s
2026-09-16 18:16:41 +00:00
panxiao81 893323a89b Merge pull request '添加 Zot 镜像构建发布流水线' (#14) from feat/image-publish-pipeline into main
test / python (push) Successful in 9s
test / shell (push) Successful in 16s
publish images / publish-images (push) Failing after 57s
Reviewed-on: #14
2026-09-16 18:12:25 +00:00
panxiao81 6f149c85e2 添加 Zot 镜像发布流水线
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 16s
2026-09-16 18:10:02 +00:00
panxiao81 852afbf02e Merge pull request '添加动态 Pod Runner 冒烟测试' (#9) from test/pod-smoke into main
test / python (push) Successful in 10s
test / shell (push) Failing after 17s
Reviewed-on: #9
2026-09-16 18:03:01 +00:00
panxiao81 6a29a10244 Merge pull request '修复 LXC 内 microVM Runner 启动链路' (#11) from fix/microvm-lxc-bootstrap into main
test / shell (push) Failing after 19s
test / python (push) Successful in 9s
Reviewed-on: #11
2026-09-16 18:02:51 +00:00
panxiao81 94a962459c Merge pull request '为 Pod Runner 提供 Ubuntu workflow 环境' (#12) from fix/ubuntu-runner-image into main
test / python (push) Successful in 8s
test / shell (push) Successful in 18s
Reviewed-on: #12
2026-09-16 18:02:16 +00:00
panxiao81 d03927d73c Merge pull request '修复 Runner SPIFFE 身份路径并补充关联日志' (#13) from fix/identity-binding-observability into main
test / shell (push) Successful in 15s
test / python (push) Successful in 8s
Reviewed-on: #13
2026-09-16 18:01:57 +00:00
panxiao81 68bb02b20a 修复 Runner 身份路径并补充关联日志
test / python (pull_request) Successful in 8s
test / shell (pull_request) Successful in 15s
2026-09-16 18:00:50 +00:00
panxiao81 c4aa6ee0af 记录 LXC guest 内存共享要求
test / python (pull_request) Successful in 26s
test / shell (pull_request) Failing after 1m2s
2026-09-16 17:42:52 +00:00
panxiao81 401c1f9a00 共享 guest memfd 避免 LXC 内存双重计费
test / shell (pull_request) Failing after 22s
test / python (pull_request) Failing after 1m3s
2026-09-16 17:38:08 +00:00
panxiao81 6dac9897fd 稳定 kind smoke 与 microVM 内存配置
test / python (pull_request) Successful in 14s
test / shell (pull_request) Failing after 19s
2026-09-16 17:28:29 +00:00
panxiao81 787614667c 为 Pod Runner 提供 Ubuntu workflow 环境
test / python (pull_request) Successful in 11s
test / shell (pull_request) Successful in 17s
2026-09-16 17:21:41 +00:00
panxiao81 072a5bad77 为 microVM 磁盘启用 direct I/O
test / python (pull_request) Successful in 18s
test / shell (pull_request) Failing after 21s
2026-09-16 17:19:32 +00:00
panxiao81 b25b1fbf62 避免 microVM 镜像复制占满 LXC page cache
test / python (pull_request) Successful in 9s
test / shell (pull_request) Failing after 18s
2026-09-16 17:15:08 +00:00
panxiao81 8578bee895 修正 kind 二进制校验
test / python (pull_request) Successful in 9s
test / shell (pull_request) Failing after 16s
2026-09-16 16:50:52 +00:00
panxiao81 02e0b698c5 为 cloud-init Runner 设置 HOME
test / python (pull_request) Successful in 11s
test / shell (pull_request) Failing after 21s
2026-09-16 16:48:44 +00:00
panxiao81 ac85d5fe58 校验 microVM Runner 任务完成状态
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 16s
2026-09-16 16:44:05 +00:00
panxiao81 bb73a48695 将 Pod smoke 改为手动触发
test / python (pull_request) Successful in 14s
test / shell (pull_request) Successful in 20s
2026-09-16 16:42:49 +00:00
panxiao81 70a8aa68c6 将 VM smoke 改为手动触发
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 15s
2026-09-16 16:42:20 +00:00
panxiao81 ce4c9f13b5 添加 microVM kind 冒烟测试
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 17s
VM kind smoke / kind (pull_request) Failing after 2s
2026-09-16 16:39:06 +00:00
panxiao81 3641e6ffb3 修复 LXC 内 microVM Runner 启动链路
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 16s
2026-09-16 16:38:03 +00:00
panxiao81 99ef62bfa0 Merge remote-tracking branch 'origin/main' into test/pod-smoke
dynamic Pod smoke test / pod-smoke (pull_request) Successful in 5s
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 18s
2026-09-16 16:06:55 +00:00
panxiao81 33cbd23861 Merge pull request:升级动态 Runner 到 3.5.0
test / shell (push) Successful in 16s
test / python (push) Successful in 9s
2026-09-16 15:58:19 +00:00
panxiao81 0dfdbd46c8 升级动态 Runner 到 3.5.0
test / shell (pull_request) Successful in 15s
test / python (pull_request) Successful in 9s
2026-09-16 15:57:15 +00:00
panxiao81 21aabe162f 添加动态 Pod Runner 冒烟测试
test / python (pull_request) Successful in 9s
test / shell (pull_request) Successful in 16s
dynamic Pod smoke test / pod-smoke (pull_request) Failing after 1s
2026-09-16 15:32:38 +00:00
panxiao81 cbfcb10143 Merge pull request:修复服务账号令牌挂载点
test / python (push) Successful in 9s
test / shell (push) Successful in 15s
2026-09-16 15:20:02 +00:00
panxiao81 3ec528623a 修复只读根文件系统下的服务账号令牌挂载
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 21s
2026-09-16 15:18:21 +00:00
panxiao81 11669ea81a Merge pull request:统一动态 Runner 命名并记录协议调度路线
test / python (push) Successful in 9s
test / shell (push) Successful in 17s
2026-09-16 14:54:44 +00:00
67 changed files with 6587 additions and 127 deletions
+19
View File
@@ -0,0 +1,19 @@
name: dynamic Pod smoke test
on:
workflow_dispatch:
jobs:
pod-smoke:
name: pod-smoke
runs-on: [self-hosted, pod]
timeout-minutes: 5
steps:
- name: Verify disposable host environment
run: |
test -S /run/spire/agent-sockets/spire-agent.sock
/opt/spire/bin/spire-agent api fetch jwt \
-audience ci-smoke \
-socketPath /run/spire/agent-sockets/spire-agent.sock \
>/dev/null
test "$(id -u)" = 2000
+128
View File
@@ -0,0 +1,128 @@
---
name: publish images
on:
push:
branches: [main]
paths:
- '.gitea/workflows/publish-images.yml'
- 'config/**'
- 'container/**'
- 'cmd/**'
- 'internal/**'
- 'scripts/**'
- 'src/**'
- 'go.mod'
- 'go.sum'
- 'pyproject.toml'
- 'README.md'
workflow_dispatch:
jobs:
publish-images:
name: publish-images
runs-on: [self-hosted, vm]
timeout-minutes: 45
permissions:
contents: read
env:
PUSH_REGISTRY: zot-push.ad.ddupan.top
PULL_REGISTRY: zot.ad.ddupan.top
CONTROLLER_REPOSITORY: panxiao81/gitea-dynamic-runner-controller
RUNNER_REPOSITORY: panxiao81/gitea-dynamic-runner-runner
SPIRE_AGENT_SOCKET: /run/spire/agent-sockets/spire-agent.sock
steps:
- uses: actions/checkout@v4
- name: Test source
shell: bash
run: |
set -euo pipefail
go test ./...
go vet ./...
python3 -m pip install --break-system-packages -e '.[test]'
pytest -q
python3 -m compileall -q src tests
apt-get update
apt-get install --yes --no-install-recommends shellcheck
shellcheck scripts/*
- name: Build and publish
shell: bash
run: |
set -euo pipefail
set +x
: "${GITHUB_SHA:?GITHUB_SHA is required}"
image_tag="sha-${GITHUB_SHA}"
docker_config=$(mktemp -d)
jwt_file=$(mktemp)
cleanup() {
docker buildx rm ci-builder >/dev/null 2>&1 || true
rm -rf -- "$docker_config" "$jwt_file"
}
trap cleanup EXIT
export DOCKER_CONFIG="$docker_config"
/opt/spire/bin/spire-agent api fetch jwt \
-audience zot \
-socketPath "$SPIRE_AGENT_SOCKET" \
-output json >"$jwt_file"
# shellcheck disable=SC2016
jq -er '.[0].svids[0].svid' "$jwt_file" | \
docker login "$PUSH_REGISTRY" --username zot --password-stdin
docker buildx create \
--name ci-builder \
--driver docker-container \
--use
publish() {
local repository=$1
local dockerfile=$2
local metadata=$3
docker buildx build \
--builder ci-builder \
--platform linux/amd64 \
--file "$dockerfile" \
--tag "${PUSH_REGISTRY}/${repository}:${image_tag}" \
--tag "${PUSH_REGISTRY}/${repository}:main" \
--provenance=mode=max \
--sbom=true \
--metadata-file "$metadata" \
--push \
.
}
publish \
"$CONTROLLER_REPOSITORY" \
container/controller.Dockerfile \
controller-metadata.json
publish \
"$RUNNER_REPOSITORY" \
container/runner.Dockerfile \
runner-metadata.json
controller_digest=$(
# shellcheck disable=SC2016
jq -er '."containerimage.digest"' controller-metadata.json
)
runner_digest=$(
# shellcheck disable=SC2016
jq -er '."containerimage.digest"' runner-metadata.json
)
controller_ref="${PULL_REGISTRY}/${CONTROLLER_REPOSITORY}@${controller_digest}"
runner_ref="${PULL_REGISTRY}/${RUNNER_REPOSITORY}@${runner_digest}"
printf 'controller=%s\nrunner=%s\n' "$controller_ref" "$runner_ref"
if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then
{
printf '## Published images\n\n'
# shellcheck disable=SC2016
printf -- '- Controller: `%s`\n' "$controller_ref"
# shellcheck disable=SC2016
printf -- '- Runner: `%s`\n' "$runner_ref"
# shellcheck disable=SC2016
printf -- '- Source: `%s`\n' "$GITHUB_SHA"
} >>"$GITHUB_STEP_SUMMARY"
fi
+10
View File
@@ -6,6 +6,16 @@ on:
pull_request: pull_request:
jobs: jobs:
go:
runs-on: [self-hosted, pod]
steps:
- uses: actions/checkout@v4
- uses: actions/setup-go@v6
with:
go-version-file: go.mod
- run: go test ./...
- run: go vet ./...
python: python:
runs-on: self-hosted runs-on: self-hosted
steps: steps:
+31
View File
@@ -0,0 +1,31 @@
name: VM kind smoke
on:
workflow_dispatch:
jobs:
kind:
runs-on: [self-hosted, vm]
steps:
- name: Verify Docker
run: docker info
- name: Install kind
shell: bash
run: |
set -euo pipefail
version=v0.33.0
curl --fail --location --silent --show-error \
--output /tmp/kind "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64"
curl --fail --location --silent --show-error \
--output /tmp/kind.sha256sum "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64.sha256sum"
printf '%s %s\n' "$(cut -d ' ' -f1 /tmp/kind.sha256sum)" /tmp/kind | sha256sum --check
chmod 0755 /tmp/kind
- name: Create and delete kind cluster
shell: bash
run: |
set -euo pipefail
trap '/tmp/kind delete cluster --name smoke' EXIT
/tmp/kind create cluster --name smoke --wait 180s
/tmp/kind get clusters | grep -Fx smoke
+5
View File
@@ -0,0 +1,5 @@
---
self-hosted-runner:
labels:
- pod
- vm
+52 -14
View File
@@ -15,26 +15,32 @@ runs-on: [self-hosted, vm]
只执行一个 job,并在 job 结束后连同本地状态一起销毁。完整的设计约束见 只执行一个 job,并在 job 结束后连同本地状态一起销毁。完整的设计约束见
[`docs/design-principles.md`](docs/design-principles.md)。 [`docs/design-principles.md`](docs/design-principles.md)。
组件: 目标 Go controller 组件:
- `controller`:接收 Gitea `workflow_job` webhook,将指定 label 的 queued job - `scheduler`:以常驻 Gitea RunnerService 身份直接领取 task,并把完整 assignment
发布到 NATS JetStream。 持久化到 JetStream;一个 registration 下按配置启动多个并发 `FetchTask` goroutine,
- `worker`:领取任务、限制并发,并通过 Pod 或 microVM backend 创建一次性环境。 同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。
- `microvm-runner-launch`:为每个任务创建 COW disk、NoCloud seed 和 TAP,运行 - `pod-worker`:直接在 homelab Kubernetes 创建一次性 Pod。
- `vm-worker`:通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 Kata microVM。
- 三个组件默认在同一个 Go 进程启用。首轮集成期间不允许只启动 worker,因为 facade
的 assignment claim registry 仍是进程内状态;支持安全拆分前进程会明确拒绝该配置。
- `microvm-runner-launch`:为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP,运行
Cloud Hypervisor,退出后完整清理。 Cloud Hypervisor,退出后完整清理。
- `guest-runner`:在 guest 中领取一次性 runner registration token,注册 ephemeral - `guest-runner`:在 guest 中领取一次性 runner registration token,注册 ephemeral
runner,执行一个 job 后关机。 runner,执行一个 job 后关机。
- `pod-worker`:在 Kubernetes 中创建一次性 privileged Pod;Pod 内的 workflow 使用 - `opensandbox-identity`:在 sandbox 集群按实际 Pod UID 创建并清理临时 SPIFFE
host executor,Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。 entry;不持有 OpenSandbox API key、Gitea token 或 Bao 凭据。身份与 Pool 契约见
[`docs/opensandbox-runner.md`](docs/opensandbox-runner.md)。
- Pod executor:在 Kubernetes 中创建一次性 privileged Pod;Pod 内的 workflow 使用
host executor,Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在
支持原生 job hooks 的 3.x 版本,在 workflow 第一步前等待实际任务对应的 SVID。
- `jwt-broker`:早期共享 Kubernetes runner 的过渡实验;目标架构不部署它,每个 - `jwt-broker`:早期共享 Kubernetes runner 的过渡实验;目标架构不部署它,每个
动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。 动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
消息流使用一个 `WorkQueuePolicy` stream。相同 runner label 的所有 worker 共享同一 Pod 路径由 homelab 集群中的 `pod-worker` 直接创建 Kubernetes Pod。OpenSandbox 只用于
durable consumer;扩容只需要增加 worker 或提高单机 capacity。 VM/Kata workload;两个 backend 使用独立 durable consumer 和独立容量池。assignment 根据
`runs-on` 进入对应池,池满时留在 JetStream pending,不会创建超出容量的 workload;任一
当前 webhook → NATS 流程是用于尽快验证 Pod/VM 生命周期的 bootstrap 实现,不是 执行层故障不会阻塞另一条部署。长期 RunnerService 协议路线见
长期调度接口。长期目标是让 controller 作为兼容 Gitea Runner 协议的调度器直接注册、
声明 labels、领取 task,并把已领取 task 交给 Pod/VM executor;路线与迁移边界见
[`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。 [`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
## 开发 ## 开发
@@ -44,15 +50,47 @@ python -m venv .venv
. .venv/bin/activate . .venv/bin/activate
pip install -e '.[test]' pip install -e '.[test]'
pytest pytest
go test ./...
go vet ./...
``` ```
## Go controller 首次集成配置
controller 默认执行 `controller` 子命令,runner 镜像执行 `executor` 子命令。所有长期
credential 都从挂载文件读取,不接受明文环境变量:
- `GITEA_RUNNER_UUID_FILE`、`GITEA_RUNNER_TOKEN_FILE`:scheduler 的常驻 RunnerService
registration;该 credential 不下发给 executor。
- `NATS_PRODUCER_PASSWORD_FILE`、`NATS_WORKER_PASSWORD_FILE`:分别使用现有最小权限的
`ci-producer` publish 连接和 `ci-worker` pull/ACK 连接,controller 不合并权限。
- `RUNNER_FACADE_CAPABILITY_KEY_FILE`:至少 32 字节的 controller HMAC key。
- `OPENSANDBOX_API_KEY_FILE`:仅启用 `vm-worker` 时读取。
必要的非 secret 配置包括 `POD_EXECUTOR_IMAGE`(应使用 digest)、`SPIRE_AGENT_ID`、
`RUNNER_FACADE_URL`、`RUNNER_FACADE_SPIFFE_ID` 和 `SPIFFE_ENDPOINT_SOCKET`。默认
`COMPONENTS=all`、Pod 并发 4、VM 并发 1;首次 smoke test 应显式设为
`COMPONENTS=scheduler,pod-worker`,先验证 Pod 链路,避免同时消耗 VM 容量。
Pod task 的 terminal update 被 Gitea 接受后,controller 会在 Pod 上持久写入
`ci.ddupan.top/terminal=true` label。生命周期 reconciler 只清理同时带该 label 且已经
进入 `Succeeded` 或 `Failed` phase 的 Pod 及其同名 `ClusterStaticEntry`;controller
重启不影响清理恢复,上报终态前失败的 Pod 也不会被误删。
## 安全边界 ## 安全边界
- NATS 密码、webhook secret 和 Gitea registration token 只从文件读取。 - OpenSandbox API key、webhook secret 和 Gitea registration token 只从文件读取。
- registration token 不写入 seed image;worker 通过单次 nonce endpoint 交给 guest。 - registration token 不写入 seed image;worker 通过单次 nonce endpoint 交给 guest。
- guest 启动时从仅监听 microVM bridge 的 worker endpoint 获取固定版本 Runner 和配置
资产;基础镜像无需为 Runner 发布而重做。
- `runner-vm-bootstrap.yaml` 暂时只验证 VM 调度和生命周期,不提供 SPIFFE
identity;VM agent attestation 完成前不得将它当作身份链路验证结果。
- guest runner 使用 `--ephemeral`,每台 VM 只执行一个 job。 - guest runner 使用 `--ephemeral`,每台 VM 只执行一个 job。
- launcher 只接受 UUID instance ID 和 URL-safe nonce,所有临时文件都位于独立目录。 - launcher 只接受 UUID instance ID 和 URL-safe nonce,所有临时文件都位于独立目录。
- base image 不得包含 runner identity、registration token、SSH 密码或 host key。 - base image 不得包含 runner identity、registration token、SSH 密码或 host key。
- LXC 内运行 Cloud Hypervisor 必须为 guest memory 启用 `shared=on`。默认的 private
memfd 映射会在 guest 写入后同时产生 shmem 与 anonymous CoW charge,使 LXC cgroup
对 guest RAM 接近双倍计费。
homelab 的 Kubernetes、OpenBao、LXC、bridge 和容量配置保留在 homelab 的 Kubernetes、OpenBao、LXC、bridge 和容量配置保留在
`panxiao81/homelab-infra`。 `panxiao81/homelab-infra`。
+388
View File
@@ -0,0 +1,388 @@
package main
import (
"context"
"errors"
"fmt"
"log"
"net/http"
"os"
"slices"
"strconv"
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"golang.org/x/sync/errgroup"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/leaderelection"
"k8s.io/client-go/tools/leaderelection/resourcelock"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/assignmentqueue"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/backendpool"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/giteaactions"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/opensandboxbackend"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/podbackend"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerbootstrap"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskscheduler"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
)
type runComponent func(context.Context) error
func (function runComponent) Run(ctx context.Context) error { return function(ctx) }
type controllerConfig struct {
Components controller.Selection
TrustDomain, WorkloadAPIAddr string
GiteaURL, GiteaUUID, GiteaToken string
NATSURL, NATSProducerUser, NATSProducerPassword string
NATSWorkerUser, NATSWorkerPassword, NATSCA, Stream, SubjectBase string
FacadeListen, FacadeURL, FacadeSPIFFEID string
CapabilityKey []byte
PodNamespace, PodImage, PodServiceAccount, SPIRECluster, SPIREClass string
SPIREAgentID string
PodExecutorUID, PodCapacity int
OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool string
VMTimeout, VMCapacity int
}
func runController(ctx context.Context) error {
config, err := loadControllerConfig()
if err != nil {
return err
}
if !slices.Contains(config.Components, controller.Scheduler) {
return errors.New("split worker deployment is not yet safe: scheduler/facade must be enabled with workers")
}
if !slices.Contains(config.Components, controller.PodWorker) && !slices.Contains(config.Components, controller.VMWorker) {
return errors.New("scheduler requires at least one local backend worker")
}
producerConnection, err := connectNATS(config.NATSURL, config.NATSProducerUser, config.NATSProducerPassword, config.NATSCA, "gitea-dynamic-runner-producer")
if err != nil {
return fmt.Errorf("connect NATS producer: %w", err)
}
defer producerConnection.Close()
producerJS, err := jetstream.New(producerConnection)
if err != nil {
return fmt.Errorf("open producer JetStream: %w", err)
}
if _, err := producerJS.Stream(ctx, config.Stream); err != nil {
return fmt.Errorf("open assignment stream %s: %w", config.Stream, err)
}
workerConnection, err := connectNATS(config.NATSURL, config.NATSWorkerUser, config.NATSWorkerPassword, config.NATSCA, "gitea-dynamic-runner-worker")
if err != nil {
return fmt.Errorf("connect NATS worker: %w", err)
}
defer workerConnection.Close()
workerJS, err := jetstream.New(workerConnection)
if err != nil {
return fmt.Errorf("open worker JetStream: %w", err)
}
capabilities, err := runnerfacade.NewCapabilities(config.CapabilityKey)
if err != nil {
return err
}
registry := runnerfacade.NewRegistry()
podPool := backendpool.New(config.PodCapacity)
vmPool := backendpool.New(config.VMCapacity)
var podExecutorBackend *podbackend.Backend
var vmExecutorBackend *opensandboxbackend.Backend
giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken)
facade := &runnerfacade.Facade{
Registry: registry, Capabilities: capabilities, Upstream: giteaClient,
OnTerminal: func(ctx context.Context, assignment taskassignment.Assignment) error {
switch assignment.Backend {
case taskassignment.BackendPod:
if podExecutorBackend == nil {
return errors.New("Pod lifecycle backend is not configured")
}
if err := podExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
return err
}
podPool.Release(assignment.ID)
case taskassignment.BackendVM:
if vmExecutorBackend == nil {
return errors.New("VM lifecycle backend is not configured")
}
if err := vmExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
return err
}
vmPool.Release(assignment.ID)
}
return nil
},
}
bootstrap := runnerbootstrap.Bootstrap{
Capabilities: capabilities, FacadeURL: config.FacadeURL, FacadeSPIFFEID: config.FacadeSPIFFEID,
WorkloadAPIAddr: config.WorkloadAPIAddr,
}
labels := []string{"self-hosted"}
if slices.Contains(config.Components, controller.PodWorker) {
labels = append(labels, string(taskassignment.BackendPod))
}
if slices.Contains(config.Components, controller.VMWorker) {
labels = append(labels, string(taskassignment.BackendVM))
}
poller := taskscheduler.Poller{
Client: giteaClient,
Scheduler: &taskscheduler.Scheduler{TrustDomain: config.TrustDomain, Dispatcher: assignmentqueue.Publisher{
JetStream: producerJS, SubjectBase: config.SubjectBase,
}},
Config: taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels, Capacity: config.PodCapacity + config.VMCapacity},
OnError: func(err error) { log.Printf("scheduler: %v", err) },
}
kubernetesConfig, err := rest.InClusterConfig()
if err != nil {
return fmt.Errorf("load leader election Kubernetes config: %w", err)
}
kubernetesClient, err := kubernetes.NewForConfig(kubernetesConfig)
if err != nil {
return fmt.Errorf("create leader election Kubernetes client: %w", err)
}
leaderIdentity := strings.TrimSpace(os.Getenv("HOSTNAME"))
if leaderIdentity == "" {
return errors.New("HOSTNAME is required for scheduler leader election")
}
facadeServer := runnerfacade.Server{
Facade: facade, ListenAddress: config.FacadeListen, TrustDomain: config.TrustDomain,
WorkloadAPIAddr: config.WorkloadAPIAddr, UpstreamURL: config.GiteaURL,
}
components := controller.Registry{
controller.Scheduler: runComponent(func(ctx context.Context) error {
group, groupContext := errgroup.WithContext(ctx)
group.Go(func() error { return facadeServer.Run(groupContext) })
group.Go(func() error {
return runSchedulerLeader(groupContext, kubernetesClient, config.PodNamespace, leaderIdentity, poller.Run)
})
return group.Wait()
}),
}
if slices.Contains(config.Components, controller.PodWorker) {
client, err := podbackend.NewInClusterClient()
if err != nil {
return err
}
backend := podbackend.Backend{API: client, Config: podbackend.Config{
Namespace: config.PodNamespace, Image: config.PodImage, ServiceAccount: config.PodServiceAccount,
ExecutorArgs: []string{"executor"}, TrustDomain: config.TrustDomain,
SPIRECluster: config.SPIRECluster, SPIREClass: config.SPIREClass,
SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID,
}}
podExecutorBackend = &backend
assignments, err := backend.RecoverAssignments(ctx)
if err != nil {
return err
}
for _, assignment := range assignments {
podPool.Restore(assignment.ID)
if err := registry.RecoverClaimed(assignment); err != nil {
return fmt.Errorf("recover Pod facade claim %s: %w", assignment.ID, err)
}
}
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, podPool)
if err != nil {
return err
}
lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) { log.Printf("pod lifecycle: %v", err) }}
components[controller.PodWorker] = runComponent(func(ctx context.Context) error {
group, groupContext := errgroup.WithContext(ctx)
group.Go(func() error { return component.Run(groupContext) })
group.Go(func() error { return lifecycle.Run(groupContext) })
return group.Wait()
})
}
if slices.Contains(config.Components, controller.VMWorker) {
lifecycle := opensandboxbackend.NewLifecycleClient(config.OpenSandboxURL, config.OpenSandboxAPIKey, &http.Client{Timeout: 60 * time.Second})
backend := opensandboxbackend.Backend{Lifecycle: lifecycle, Config: opensandboxbackend.Config{
Pool: config.OpenSandboxPool, Timeout: config.VMTimeout,
Entrypoint: []string{"/usr/local/bin/gitea-dynamic-runner", "executor"},
Env: map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr},
}}
vmExecutorBackend = &backend
assignments, err := backend.RecoverAssignments(ctx, config.TrustDomain)
if err != nil {
return err
}
for _, assignment := range assignments {
vmPool.Restore(assignment.ID)
if err := registry.RecoverClaimed(assignment); err != nil {
return fmt.Errorf("recover VM facade claim %s: %w", assignment.ID, err)
}
}
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, vmPool)
if err != nil {
return err
}
lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) { log.Printf("VM lifecycle: %v", err) }}
components[controller.VMWorker] = runComponent(func(ctx context.Context) error {
group, groupContext := errgroup.WithContext(ctx)
group.Go(func() error { return component.Run(groupContext) })
group.Go(func() error { return lifecycleReconciler.Run(groupContext) })
return group.Wait()
})
}
return controller.Run(ctx, config.Components, components)
}
func runSchedulerLeader(ctx context.Context, client kubernetes.Interface, namespace, identity string, run func(context.Context) error) error {
if client == nil || namespace == "" || identity == "" || run == nil {
return errors.New("leader election client, namespace, identity, and scheduler are required")
}
electionContext, cancel := context.WithCancel(ctx)
defer cancel()
result := make(chan error, 1)
lock := &resourcelock.LeaseLock{
LeaseMeta: metav1.ObjectMeta{Name: "dynamic-runner-scheduler", Namespace: namespace},
Client: client.CoordinationV1(),
LockConfig: resourcelock.ResourceLockConfig{
Identity: identity,
},
}
elector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
Lock: lock, LeaseDuration: 15 * time.Second, RenewDeadline: 10 * time.Second, RetryPeriod: 2 * time.Second,
ReleaseOnCancel: true,
Callbacks: leaderelection.LeaderCallbacks{
OnStartedLeading: func(leaderContext context.Context) {
result <- run(leaderContext)
cancel()
},
OnStoppedLeading: func() {
if ctx.Err() == nil {
select {
case result <- errors.New("scheduler leadership lost"):
default:
}
}
},
},
})
if err != nil {
return fmt.Errorf("configure scheduler leader election: %w", err)
}
go elector.Run(electionContext)
select {
case err := <-result:
return err
case <-ctx.Done():
return nil
}
}
func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn, error) {
options := []nats.Option{nats.Name(clientName), nats.UserInfo(user, password)}
if caFile != "" {
options = append(options, nats.RootCAs(caFile))
}
return nats.Connect(server, options...)
}
func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims, admission assignmentqueue.Admission) (controller.Component, error) {
consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, backend, capacity)
if err != nil {
return nil, err
}
return assignmentqueue.ConsumerComponent{
Consumer: consumer, Capacity: capacity,
Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission},
OnError: func(err error) { log.Printf("%s worker: %v", backend, err) },
}, nil
}
func loadControllerConfig() (controllerConfig, error) {
selection, err := controller.ParseSelection(os.Getenv("COMPONENTS"))
if err != nil {
return controllerConfig{}, err
}
read := func(name string) (string, error) {
path := os.Getenv(name)
if path == "" {
return "", fmt.Errorf("%s is required", name)
}
value, err := os.ReadFile(path)
if err != nil {
return "", fmt.Errorf("read %s: %w", name, err)
}
return strings.TrimSpace(string(value)), nil
}
uuid, err := read("GITEA_RUNNER_UUID_FILE")
if err != nil {
return controllerConfig{}, err
}
token, err := read("GITEA_RUNNER_TOKEN_FILE")
if err != nil {
return controllerConfig{}, err
}
producerPassword, err := read("NATS_PRODUCER_PASSWORD_FILE")
if err != nil {
return controllerConfig{}, err
}
workerPassword, err := read("NATS_WORKER_PASSWORD_FILE")
if err != nil {
return controllerConfig{}, err
}
capabilityKey, err := read("RUNNER_FACADE_CAPABILITY_KEY_FILE")
if err != nil {
return controllerConfig{}, err
}
config := controllerConfig{
Components: selection, TrustDomain: env("TRUST_DOMAIN", "ddupan.top"), WorkloadAPIAddr: os.Getenv("SPIFFE_ENDPOINT_SOCKET"),
GiteaURL: env("GITEA_INSTANCE_URL", "https://git.ddupan.top"), GiteaUUID: uuid, GiteaToken: token,
NATSURL: env("NATS_URL", "tls://nats.ad.ddupan.top:4222"),
NATSProducerUser: env("NATS_PRODUCER_USER", "ci-producer"), NATSProducerPassword: producerPassword,
NATSWorkerUser: env("NATS_WORKER_USER", "ci-worker"), NATSWorkerPassword: workerPassword,
NATSCA: os.Getenv("NATS_CA_FILE"), Stream: env("NATS_STREAM", "CI_RUNNER"), SubjectBase: env("NATS_SUBJECT_BASE", "ci.runner"),
FacadeListen: env("RUNNER_FACADE_LISTEN", ":8443"), FacadeURL: os.Getenv("RUNNER_FACADE_URL"), FacadeSPIFFEID: os.Getenv("RUNNER_FACADE_SPIFFE_ID"), CapabilityKey: []byte(capabilityKey),
PodNamespace: env("POD_NAMESPACE", "gitea-actions"), PodImage: os.Getenv("POD_EXECUTOR_IMAGE"), PodServiceAccount: env("POD_SERVICE_ACCOUNT", "gitea-task-executor"),
SPIRECluster: env("SPIRE_CLUSTER", "homelab"), SPIREClass: env("SPIRE_CLASS", "spire-mgmt-spire"), SPIREAgentID: os.Getenv("SPIRE_AGENT_ID"), PodExecutorUID: envInt("POD_EXECUTOR_UID", 2000), PodCapacity: envInt("POD_CAPACITY", 4),
OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1),
}
if config.WorkloadAPIAddr == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
return controllerConfig{}, errors.New("SPIFFE_ENDPOINT_SOCKET, RUNNER_FACADE_URL, and RUNNER_FACADE_SPIFFE_ID are required")
}
if slices.Contains(selection, controller.PodWorker) && config.PodImage == "" {
return controllerConfig{}, errors.New("POD_EXECUTOR_IMAGE is required for pod-worker")
}
if slices.Contains(selection, controller.PodWorker) && config.SPIREAgentID == "" {
return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for pod-worker")
}
if slices.Contains(selection, controller.VMWorker) {
if config.OpenSandboxURL == "" {
return controllerConfig{}, errors.New("OPENSANDBOX_API is required for vm-worker")
}
config.OpenSandboxAPIKey, err = read("OPENSANDBOX_API_KEY_FILE")
if err != nil {
return controllerConfig{}, err
}
}
return config, nil
}
func env(name, fallback string) string {
if value := strings.TrimSpace(os.Getenv(name)); value != "" {
return value
}
return fallback
}
func envInt(name string, fallback int) int {
value := strings.TrimSpace(os.Getenv(name))
if value == "" {
return fallback
}
parsed, err := strconv.Atoi(value)
if err != nil || parsed < 1 {
return fallback
}
return parsed
}
@@ -0,0 +1,62 @@
package main
import (
"os"
"path/filepath"
"testing"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller"
)
func secretFile(t *testing.T, name, value string) string {
t.Helper()
path := filepath.Join(t.TempDir(), name)
if err := os.WriteFile(path, []byte(value+"\n"), 0o600); err != nil {
t.Fatal(err)
}
return path
}
func TestLoadControllerConfigUsesFileSecrets(t *testing.T) {
t.Setenv("COMPONENTS", "scheduler,pod-worker")
t.Setenv("GITEA_RUNNER_UUID_FILE", secretFile(t, "uuid", "scheduler-uuid"))
t.Setenv("GITEA_RUNNER_TOKEN_FILE", secretFile(t, "token", "scheduler-token"))
t.Setenv("NATS_PRODUCER_PASSWORD_FILE", secretFile(t, "nats-producer", "producer-password"))
t.Setenv("NATS_WORKER_PASSWORD_FILE", secretFile(t, "nats-worker", "worker-password"))
t.Setenv("RUNNER_FACADE_CAPABILITY_KEY_FILE", secretFile(t, "capability", "0123456789abcdef0123456789abcdef"))
t.Setenv("SPIFFE_ENDPOINT_SOCKET", "unix:///run/spire/agent-sockets/spire-agent.sock")
t.Setenv("RUNNER_FACADE_URL", "https://gitea-runner-facade.gitea-actions.svc:8443")
t.Setenv("RUNNER_FACADE_SPIFFE_ID", "spiffe://ddupan.top/ns/gitea-actions/sa/gitea-dynamic-runner")
t.Setenv("POD_EXECUTOR_IMAGE", "zot.ddupan.top/ci/gitea-runner@sha256:abc")
t.Setenv("SPIRE_AGENT_ID", "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/node-uid")
config, err := loadControllerConfig()
if err != nil {
t.Fatal(err)
}
if len(config.Components) != 2 || config.Components[0] != controller.Scheduler || config.Components[1] != controller.PodWorker {
t.Fatalf("components = %#v", config.Components)
}
if config.GiteaUUID != "scheduler-uuid" || config.GiteaToken != "scheduler-token" || config.NATSProducerPassword != "producer-password" || config.NATSWorkerPassword != "worker-password" {
t.Fatal("file secrets were not loaded")
}
if string(config.CapabilityKey) != "0123456789abcdef0123456789abcdef" || config.PodExecutorUID != 2000 {
t.Fatalf("config = %#v", config)
}
}
func TestLoadControllerConfigRequiresOpenSandboxSecretOnlyForVM(t *testing.T) {
t.Setenv("COMPONENTS", "scheduler,vm-worker")
t.Setenv("GITEA_RUNNER_UUID_FILE", secretFile(t, "uuid", "uuid"))
t.Setenv("GITEA_RUNNER_TOKEN_FILE", secretFile(t, "token", "token"))
t.Setenv("NATS_PRODUCER_PASSWORD_FILE", secretFile(t, "nats-producer", "producer-password"))
t.Setenv("NATS_WORKER_PASSWORD_FILE", secretFile(t, "nats-worker", "worker-password"))
t.Setenv("RUNNER_FACADE_CAPABILITY_KEY_FILE", secretFile(t, "capability", "0123456789abcdef0123456789abcdef"))
t.Setenv("SPIFFE_ENDPOINT_SOCKET", "unix:///run/spire/agent-sockets/spire-agent.sock")
t.Setenv("RUNNER_FACADE_URL", "https://facade:8443")
t.Setenv("RUNNER_FACADE_SPIFFE_ID", "spiffe://ddupan.top/controller")
t.Setenv("OPENSANDBOX_API", "http://opensandbox.internal")
if _, err := loadControllerConfig(); err == nil {
t.Fatal("expected missing OpenSandbox API key file error")
}
}
+43
View File
@@ -0,0 +1,43 @@
package main
import (
"context"
"errors"
"fmt"
"os"
"os/signal"
"syscall"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerbootstrap"
)
func main() {
if err := run(); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
}
func run() error {
if len(os.Args) > 2 {
return errors.New("usage: gitea-dynamic-runner [controller|executor]")
}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
command := "controller"
if len(os.Args) == 2 {
command = os.Args[1]
}
switch command {
case "controller":
return runController(ctx)
case "executor":
config, err := runnerbootstrap.ExecutorConfigFromEnvironment()
if err != nil {
return err
}
return runnerbootstrap.RunExecutor(ctx, config)
default:
return fmt.Errorf("unknown command %q", command)
}
}
+11
View File
@@ -0,0 +1,11 @@
runner:
capacity: 1
timeout: 3h
shutdown_timeout: 1m
host:
workdir_parent: /workspace
container:
require_docker: false
valid_volumes: []
+13 -11
View File
@@ -1,15 +1,17 @@
FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire FROM docker.io/library/golang:1.27-alpine@sha256:4cb7ac979db5fcc41cae44b2227ba5ab8a51e8807f40d9ba4dee20a0ad960b5b AS build
FROM python:3.12.11-alpine3.22 AS build
WORKDIR /src WORKDIR /src
COPY pyproject.toml README.md ./ COPY go.mod go.sum ./
COPY src ./src RUN go mod download
RUN python -m venv /venv && /venv/bin/pip install --no-cache-dir . COPY cmd ./cmd
COPY internal ./internal
RUN CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/gitea-dynamic-runner ./cmd/gitea-dynamic-runner
FROM python:3.12.11-alpine3.22 FROM python:3.12.11-alpine3.22
RUN addgroup -S -g 65532 runner && adduser -S -D -H -u 65532 -G runner runner RUN addgroup -S -g 65532 runner \
COPY --from=build /venv /venv && adduser -S -D -H -u 65532 -G runner runner \
COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent && install -d -o 65532 -g 65532 /var/run/secrets/kubernetes.io/serviceaccount
COPY --from=build /out/gitea-dynamic-runner /usr/local/bin/gitea-dynamic-runner
USER 65532:65532 USER 65532:65532
EXPOSE 8787 EXPOSE 8443
ENTRYPOINT ["/venv/bin/gitea-dynamic-runner-controller"] ENTRYPOINT ["/usr/local/bin/gitea-dynamic-runner"]
CMD ["controller"]
+33 -3
View File
@@ -1,8 +1,38 @@
FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
FROM docker.io/gitea/runner:2@sha256:66d80966792e621c9761c47919644198d35fd1c297e9a01e69ed3c1ae37db0c7 FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242 AS runner
FROM docker.io/library/golang:1.27-alpine@sha256:4cb7ac979db5fcc41cae44b2227ba5ab8a51e8807f40d9ba4dee20a0ad960b5b AS controller
WORKDIR /src
COPY go.mod go.sum ./
RUN go mod download
COPY cmd ./cmd
COPY internal ./internal
RUN CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/gitea-dynamic-runner ./cmd/gitea-dynamic-runner
# The runner daemon image is intentionally minimal and does not contain the
# Node.js runtime required by JavaScript actions such as actions/checkout.
# Run the daemon in Gitea's Ubuntu workflow image so host-mode jobs and their
# actions share a GitHub Actions-compatible userspace.
FROM docker.io/gitea/runner-images:ubuntu-latest@sha256:fd911d7417bfbf0f454530e447da95b58001e1df41bbc5e1a8dd35d432575aae
USER root USER root
RUN groupadd --gid 2000 runner \
&& useradd --uid 2000 --gid 2000 --groups docker --create-home --shell /bin/bash runner \
&& printf 'runner ALL=(ALL) NOPASSWD:ALL\n' >/etc/sudoers.d/runner \
&& chmod 0440 /etc/sudoers.d/runner \
&& install -d -o 2000 -g 2000 /data
COPY --from=runner /usr/local/bin/gitea-runner /usr/local/bin/gitea-runner
COPY --from=controller /out/gitea-dynamic-runner /usr/local/bin/gitea-dynamic-runner
COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
COPY config/runner.yaml /etc/gitea-runner/config.yaml COPY config/runner.yaml /etc/gitea-runner/config.yaml
COPY scripts/gitea-job-started /usr/local/libexec/gitea-job-started COPY --chmod=0755 scripts/gitea-job-started /usr/local/libexec/gitea-job-started
RUN chmod 0755 /usr/local/libexec/gitea-job-started COPY --chmod=0755 scripts/gitea-opensandbox-runner /usr/local/libexec/gitea-opensandbox-runner
VOLUME ["/data"]
ENV HOME=/home/runner
USER 2000:2000
WORKDIR /home/runner
ENTRYPOINT ["/usr/local/bin/gitea-dynamic-runner"]
CMD ["executor"]
+22 -12
View File
@@ -21,18 +21,20 @@ runs-on: [self-hosted, vm]
## 一个 job,一个环境 ## 一个 job,一个环境
Controller 根据 Gitea `workflow_job` webhook 创建执行环境。每个 Pod 或 VM 注册一个 Controller 通过 Gitea RunnerService 原子领取具体 task,再把版本化 assignment 交给
ephemeral runner,只执行一个 job;任务结束后注销 runner,并删除计算环境及其全部 一个一次性 Pod 或 VM。executor 直接执行已经领取的 task,不再注册临时 runner 去
本地状态。 二次竞争任务;任务结束并回报 Gitea 后删除计算环境及其全部本地状态。
`job_id` 仅用于消息去重、状态追踪、实例关联和失败清理,不进入 workload 身份,也 `job_id` 仅用于消息去重、状态追踪、实例关联和失败清理,不进入 workload 身份,也
不参与资源授权。 不参与资源授权。
Gitea 不保证由某次 `queued` webhook 创建的 runner 一定领取该 webhook 对应的 job。 assignment ID 只用于消息去重、状态追踪、实例关联和失败清理,不进入 workload 身份,
因此创建环境时只赋予无业务权限的启动身份。runner 实际领取任务后,controller 根据 也不参与资源授权。worker 通过 assignment ID 从 Kubernetes labels 或 OpenSandbox
`in_progress` webhook 返回的 `runner_name` 和真实 job 名称绑定业务身份;环境中的 metadata 恢复 executor;JetStream 不保存 executor 生命周期状态。
job-start hook 必须等目标 SVID 可用后才放行 workflow 的第一步。不能依据 queued
事件提前赋予任务权限。 Pod executor 创建后,controller 使用实际 Pod UID 创建幂等 `ClusterStaticEntry`,将
SPIFFE ID绑定到该 Pod 的 workload selector。executor 必须等目标 SVID 可用后才执行
workflow 的第一步。
## 环境只提供运行边界 ## 环境只提供运行边界
@@ -54,15 +56,15 @@ kind 等工具由 pipeline 按需安装和启动,而不是由 controller 预
SPIFFE ID 由具有业务意义且稳定的 workflow 上下文派生: SPIFFE ID 由具有业务意义且稳定的 workflow 上下文派生:
```text ```text
spiffe://ddupan.top/ci/<owner>/<repository>/<job-name> spiffe://ddupan.top/ci/<owner>/<repository>/<job-key>
``` ```
同一种任务在不同运行中使用相同的逻辑 SPIFFE ID;每次运行取得独立、短期的 SVID。 同一种任务在不同运行中使用相同的逻辑 SPIFFE ID;每次运行取得独立、短期的 SVID。
Pod 与 VM 是可替换的执行实现,因此默认不写入 SPIFFE ID。 Pod 与 VM 是可替换的执行实现,因此默认不写入 SPIFFE ID。
job 名称必须经过确定性的路径规范化。规范化结果必须保留仓库边界,并在发生冲突时 job key 必须满足 `[A-Za-z_][A-Za-z0-9_-]*`,展示名称 `name` 不参与身份计算。同一
拒绝创建环境,不能静默地让两个任务共享身份。同一仓库内需要不同权限的任务应使用 仓库内需要不同权限的任务应使用不同的 job key;workflow 文件只是编排载体,不进入
不同的 job 名称;workflow 文件只是编排载体,不进入权限身份。 权限身份。
## Self-service 与授权边界 ## Self-service 与授权边界
@@ -86,3 +88,11 @@ OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意
仓库中的 `jwt-broker` 是早期方案的实验实现,在 Pod/VM 动态执行环境完成迁移后不应 仓库中的 `jwt-broker` 是早期方案的实验实现,在 Pod/VM 动态执行环境完成迁移后不应
部署。 部署。
## 实现依赖原则
基础设施协议优先使用上游维护的成熟客户端,不在 controller 内重复实现认证、连接、
资源编码或错误语义。Kubernetes 使用 `client-go`,NATS JetStream 使用 `nats.go`,
Gitea RunnerService 使用 `actionslib`,OpenSandbox Lifecycle API 使用官方 Go SDK;
SPIFFE Workload API 与 mTLS 使用 `go-spiffe`。自定义代码只保留领域模型、reconcile
规则及上游客户端未覆盖的最小适配层。
+55
View File
@@ -0,0 +1,55 @@
# OpenSandbox VM runner
Gitea 的 `workflow_job` webhook 只负责发现带 `self-hosted,pod` 或
`self-hosted,vm` label 的 queued job。controller 不再把请求写入 NATS,而是携带由
OpenSandbox 只承载 `self-hosted,vm` workload。`self-hosted,pod` 由 homelab 原生
Kubernetes worker 创建,不经过 OpenSandbox。homelab ExternalSecret 挂载的 API key
供 VM consumer 调用 OpenSandbox Lifecycle API:
```text
http://10.60.0.13:8080/v1/sandboxes
```
请求通过 `extensions.poolRef` 选择 `ci-pod` 或 `ci-vm`。地址是 VyOS HAProxy 的
内网 TCP frontend,backend 为 sandbox 两个节点上的固定 NodePort;不通过公网或
Cloudflare Tunnel。OpenSandbox key 只从 OpenBao `kv/k8s/opensandbox-api.api_key`
进入 homelab Secret,进程通过 `OPENSANDBOX_API_KEY_FILE` 读取。
## 身份顺序
Lifecycle 请求把稳定的 repository/task SPIFFE ID 放入 task environment。sandbox
集群内的 `opensandbox-identity` controller 读取 BatchSandbox allocation 得到实际
Pod UID,然后创建:
- parent:`spiffe://ddupan.top/spire/agent/k8s_psat/sandbox-kata/pod/<pod-uid>`;
- workload:`spiffe://ddupan.top/ci/<owner>/<repository>/<task>`;
- selector:`unix:uid:2000`。
job ID 只进入诊断 label,不进入业务身份。Pool 内 runner 进程固定使用 UID 2000,
Pod 设置 `shareProcessNamespace: true`;guest-local SPIRE Agent 使用 Pod-bound PSAT,
通过内存 emptyDir 暴露 Workload API。`gitea-opensandbox-runner` 等待精确 SVID,随后
通过一次性 nonce URL 领取 Gitea registration token。token 不进入 Lifecycle 请求、
BatchSandbox、镜像或 sandbox Secret。
identity controller 只能读取 opensandbox namespace 的 BatchSandbox/Pod,并维护带
自身 label 的 ClusterStaticEntry。它不持有 OpenSandbox API key、Gitea token 或 Bao
凭据。BatchSandbox 消失后,对应 entry 在下一次 reconcile 删除;Pod 删除后,按 Pod
UID attestation 的临时 Agent 失去父级。
## Pool 契约
`ci-vm` 使用 `kata-clh-runtime-rs`;`ci-pod` 使用默认 runc。两者都要求:
- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI 和 identity gate;
- runner UID 2000,SPIRE Agent 与 privileged dockerd 使用不同 UID;
- Docker socket 通过 group 2000 共享,Docker 数据仅存在于 sandbox emptyDir;
- `self-hosted` 必须是所有 runner labels 的前缀;
- ephemeral/once runner 完成一项任务后退出。
## 清理与恢复
Gitea 接受 runner 终态后,facade 先通过 Lifecycle API 将
`ci.ddupan.top/terminal=true` 持久化到 sandbox metadata,再向 runner 返回成功。VM
lifecycle reconciler 按该 metadata 查询并调用 DELETE;controller 在标记与删除之间重启
也能恢复清理。API delete、identity entry delete 均接受对象已不存在,OpenSandbox
timeout 仍是最终兜底回收边界;生命周期状态不写入消息队列或新的数据库。
+65 -7
View File
@@ -3,8 +3,9 @@
## 目标 ## 目标
长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner 长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner
协议客户端注册,并声明 `self-hosted`、`pod` 和 `vm` labels;它只在后端存在可用容量 协议客户端注册,并声明 `self-hosted`、`pod` 和 `vm` labels;单个 registration 内按总
时领取 task,然后将该 task 交给一个一次性 Pod 或 microVM 执行。 配置容量启动多个 `FetchTask` goroutine,再将 task 按 `runs-on` 交给 Pod 或 VM 的独立
容量池,由一次性 Pod 或 microVM 执行。
```text ```text
Gitea RunnerService Gitea RunnerService
@@ -18,6 +19,28 @@ dynamic-runner scheduler
└──────────────────────► Gitea └──────────────────────► Gitea
``` ```
controller 使用单一 Go 二进制;默认在同一进程启用 `scheduler`、`pod-worker` 和
`vm-worker`,也可通过 `--components` 只启用其中一部分。组件是独立应用服务边界,
共享进程不意味着共享后端状态或把 assignment 降级为内存 channel。
首轮集成的 facade pending/claimed registry 与三个组件同进程。虽然二进制保留组件选择
接口,但当前会拒绝“worker 不带 scheduler/facade”的拆分部署:普通 Kubernetes Service
无法保证 executor 回到持有其 assignment 的 replica。后续拆分必须增加按 assignment
路由或可重建的 claim 分发,不能新增一套生命周期数据库来掩盖该问题。
executor 直接运行固定版本的官方 Gitea Runner 二进制,不 fork workflow 执行引擎。
controller 暴露兼容 RunnerService 的 facade:`FetchTask` 只返回已分配 assignment,
`UpdateTask` 与 `UpdateLog` 转发真实 Gitea。facade 同时验证逻辑 SPIFFE ID、assignment
ID,以及由 controller 密钥确定性生成的 assignment HMAC capability;该 capability
只绑定执行实例,不参与 Zot/OpenBao 等业务授权。
executor 为官方 runner 生成与 v3.5.0 schema 一致的一次性 `.runner` 文件,并以
`daemon --once` 启动。runner 只访问 executor 内的 loopback HTTP proxy;proxy 使用
`go-spiffe` 从 Workload API 持续取得和轮换 X509-SVID,再以 mTLS 连接 controller
facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把静态客户端证书写入
镜像。assignment capability 会进入一次性 executor 环境,但不会进入 label、annotation
或 OpenSandbox metadata;它只对该 assignment 有效,并且不能绕过 SPIFFE 身份校验。
这与“收到 webhook 后临时注册另一个 act_runner”不同。`FetchTask` 已经完成任务分配, 这与“收到 webhook 后临时注册另一个 act_runner”不同。`FetchTask` 已经完成任务分配,
不能再期待 Gitea 把同一个 task 分配给随后启动的 runner。协议调度器必须让 executor 不能再期待 Gitea 把同一个 task 分配给随后启动的 runner。协议调度器必须让 executor
执行已经领取的 task,并继续完成日志、状态、心跳、取消和最终结果上报。 执行已经领取的 task,并继续完成日志、状态、心跳、取消和最终结果上报。
@@ -25,22 +48,57 @@ dynamic-runner scheduler
## 设计约束 ## 设计约束
- 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。 - 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。
- scheduler 在没有对应 backend 容量时不领取 task,避免本地形成不可控积压。 - scheduler 使用单一 Gitea runner UUID/token 和一个 `Declare`,不为并发槽位重复注册;
`POD_CAPACITY + VM_CAPACITY` 决定并发 `FetchTask` goroutine 数量。
- task 领取并持久化后按 backend 进入独立 durable consumer;对应容量池已满时延迟 NAK,
assignment 保持 JetStream pending,且不得创建超出配置容量的 workload。
- scheduler Declare 后使用 RunnerService 长轮询;一旦 FetchTask 返回已分配 task,在
JetStream publish 成功前只重试该 assignment,不领取下一项。
- 每个 executor 只执行一个 task,完成后销毁。 - 每个 executor 只执行一个 task,完成后销毁。
- SPIFFE 身份从实际领取的 task 的 repository 和 job name 派生,不需要 queued 与 - SPIFFE 身份从实际领取的 task 的 repository 和 workflow job key 派生,不需要 queued 与
in-progress webhook 的二阶段关联。 in-progress webhook 的二阶段关联。
- 身份中的 task 段使用 workflow job key,而不是可带空格的展示名称;job key 必须满足
`[A-Za-z_][A-Za-z0-9_-]*`。slug + hash 只保留为旧名称的显式迁移后备方案。
- scheduler 只做确定性的身份派生与 executor 绑定,不维护业务授权 policy;Zot、
OpenBao 等资源服务继续是唯一授权决策点。
- scheduler 的 runner registration credential 不进入 executor;executor 只得到执行 - scheduler 的 runner registration credential 不进入 executor;executor 只得到执行
当前 task 所需的短期 lease/capability。 当前 task 所需的短期 lease/capability。
- task ACK、心跳和结果必须能够跨 scheduler 重启恢复;NATS 可以继续作为内部 handoff, - JetStream 只持久化和投递 assignment,不保存 executor 生命周期状态。Pod labels/annotations
但不是 Gitea 任务事实来源。 与 OpenSandbox metadata 是后端运行状态的权威来源,Gitea 是 task 终态的权威来源。
- assignment 使用版本化 envelope 保存完整 Gitea protobuf task,并从 workflow `runs-on`
严格选择 pod 或 vm subject;消费者解码后重新派生 backend 与身份,拒绝被篡改的冗余字段。
- JetStream 的 message ID 等于稳定 assignment ID `gitea-task-<task-id>`,仅用于发布去重,
不承担 executor 生命周期记录。
- worker 按稳定 assignment ID reconcile 后端资源,进程内只保留并发控制等可丢弃状态;
不新增数据库,也不依赖内存中的 runner-to-executor 映射。
- VM worker 使用 OpenSandbox 官方 Go SDK,并把 assignment ID、repository、job key 和
SPIFFE ID写入 sandbox metadata;通过 `extensions.poolRef=ci-vm` 使用既有 Kata Pool。
- executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后,facade 在后端
metadata 写入持久 terminal marker;backend reconciler 仅在执行环境也进入终态后清理,
从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。
- pod 与 vm 使用独立 durable consumer、进程内 admission pool 和并发上限。consumer 只负责将 assignment
幂等落到后端;executor 与身份恢复 metadata 持久化后立即 `DoubleAck`。尚未取得
Pod UID 等短暂未就绪状态以及临时后端错误使用延迟 NAK。
- admission pool 只保存可重建的并发状态:启动时从 Pod labels/annotations 或 OpenSandbox
metadata 恢复非终态 assignment,terminal update 持久化成功后释放槽位,不引入新存储。
- assignment ACK 后的运行、结果回报和清理由 backend reconciler 根据 Kubernetes、
OpenSandbox 与 Gitea 的事实状态驱动,不继续占用 JetStream delivery。
- consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment;无需把完整
task 写入 Pod annotation、OpenSandbox metadata 或环境变量。
- Pod 与 VM 共享 task/executor 协议,只有环境创建和销毁实现不同。 - Pod 与 VM 共享 task/executor 协议,只有环境创建和销毁实现不同。
- scheduler 在 assignment 持久化到 JetStream 后即可继续领取;Pod 与 VM 分别由 durable
consumer 的 capacity 限制并发,不共享全局执行槽位。未知后端故障由对应 consumer 的
NAK/redelivery 收敛,不能阻塞另一种 backend。
- 两种 backend 都注入同一份 runner bootstrap 环境;Pod 仍由 homelab Kubernetes 原生
创建,只有 VM 经 OpenSandbox 创建,bootstrap 机制不改变 backend 边界。
## 实现顺序 ## 实现顺序
1. 固定当前 Gitea 版本所使用的 RunnerService protobuf 与 act_runner 版本,记录兼容 1. 固定当前 Gitea 版本所使用的 RunnerService protobuf 与 act_runner 版本,记录兼容
范围并建立协议契约测试。 范围并建立协议契约测试。
2. 实现只注册、Declare labels 和容量感知 FetchTask 的 scheduler spike,暂不执行 2. 实现只注册、Declare labels 和容量感知 FetchTask 的 scheduler spike,暂不执行
task。 task。首次集成必须验证 FetchTask 后、JetStream publish 前进程崩溃时 Gitea 对同一
runner 的 task 恢复语义;该窗口未验证前不能声称 scheduler 可无损恢复。
3. 从 act_runner 提取或复用 task 执行与日志上报能力,定义 scheduler 到 executor 的 3. 从 act_runner 提取或复用 task 执行与日志上报能力,定义 scheduler 到 executor 的
单任务协议。 单任务协议。
4. 首先接入 Pod executor,验证成功、失败、取消、超时和 scheduler 重启。 4. 首先接入 Pod executor,验证成功、失败、取消、超时和 scheduler 重启。
+72
View File
@@ -0,0 +1,72 @@
module git.ddupan.top/panxiao81/gitea-dynamic-runner
go 1.27
require (
connectrpc.com/connect v1.20.0
gitea.dev/actionslib v1.0.0
github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5
github.com/nats-io/nats.go v1.54.0
github.com/spiffe/go-spiffe/v2 v2.8.2
golang.org/x/sync v0.23.0
google.golang.org/protobuf v1.36.12
k8s.io/api v0.37.0
k8s.io/apimachinery v0.37.0
k8s.io/client-go v0.37.0
)
require (
github.com/Microsoft/go-winio v0.6.2 // indirect
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/emicklei/go-restful/v3 v3.13.0 // indirect
github.com/fxamacker/cbor/v2 v2.9.1 // indirect
github.com/go-jose/go-jose/v4 v4.1.5 // indirect
github.com/go-logr/logr v1.4.3 // indirect
github.com/go-openapi/jsonpointer v1.0.0 // indirect
github.com/go-openapi/jsonreference v1.0.0 // indirect
github.com/go-openapi/swag v0.27.1 // indirect
github.com/go-openapi/swag/cmdutils v0.27.1 // indirect
github.com/go-openapi/swag/conv v0.27.1 // indirect
github.com/go-openapi/swag/fileutils v0.27.1 // indirect
github.com/go-openapi/swag/jsonutils v0.27.1 // indirect
github.com/go-openapi/swag/loading v0.27.1 // indirect
github.com/go-openapi/swag/mangling v0.27.1 // indirect
github.com/go-openapi/swag/netutils v0.27.1 // indirect
github.com/go-openapi/swag/pools v0.27.1 // indirect
github.com/go-openapi/swag/stringutils v0.27.1 // indirect
github.com/go-openapi/swag/typeutils v0.27.1 // indirect
github.com/go-openapi/swag/yamlutils v0.27.1 // indirect
github.com/google/gnostic-models v0.7.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/klauspost/compress v1.20.0 // indirect
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/nats-io/nkeys v0.4.16 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/sirupsen/logrus v1.10.2 // indirect
github.com/x448/float16 v0.8.4 // indirect
go.yaml.in/yaml/v2 v2.4.4 // indirect
go.yaml.in/yaml/v3 v3.0.5 // indirect
go.yaml.in/yaml/v4 v4.0.0-rc.6 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/net v0.58.0 // indirect
golang.org/x/oauth2 v0.36.0 // indirect
golang.org/x/sys v0.48.0 // indirect
golang.org/x/term v0.46.0 // indirect
golang.org/x/text v0.42.0 // indirect
golang.org/x/time v0.15.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect
google.golang.org/grpc v1.79.3 // indirect
gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
k8s.io/klog/v2 v2.140.0 // indirect
k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad // indirect
k8s.io/utils v0.0.0-20260626114624-be93311217bd // indirect
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
sigs.k8s.io/randfill v1.0.0 // indirect
sigs.k8s.io/structured-merge-diff/v6 v6.4.2 // indirect
sigs.k8s.io/yaml v1.6.0 // indirect
)
+170
View File
@@ -0,0 +1,170 @@
connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ=
connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4=
gitea.dev/actionslib v1.0.0 h1:l0oFJP+P4Ds1rlCI5zk618dYkuBc2mU7Gz5wPeG0lZY=
gitea.dev/actionslib v1.0.0/go.mod h1:6O8YHkqVTKSR0LL2e5VhIDePYzGTZCbfmSVqJWEhk9g=
github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5 h1:7mZNkBh4VaI+i4VOTJIknLCBkbGizTkdHYqgcMqeDBM=
github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5/go.mod h1:w0nIMCTL1L3oSS67ABFOdYxrezYQZhk01UtmKSi2UhA=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/emicklei/go-restful/v3 v3.13.0 h1:C4Bl2xDndpU6nJ4bc1jXd+uTmYPVUwkD6bFY/oTyCes=
github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ=
github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
github.com/go-jose/go-jose/v4 v4.1.5 h1:RjgjO2LOtWOJKUC5wpwY9LR3B3vwVAz6JS2YHfYU6eA=
github.com/go-jose/go-jose/v4 v4.1.5/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/go-openapi/jsonpointer v1.0.0 h1:kR9tHqY0CtZaOPVFm622dPVNhrvYpwr4uCxgL3h1H8s=
github.com/go-openapi/jsonpointer v1.0.0/go.mod h1:Z3rw7dWu1p9IgitXCFamSlA5lmDiklEB6vkaxcNZW5Y=
github.com/go-openapi/jsonreference v1.0.0 h1:jlmTr6torcd1YgDQvSfNmRtKzYDO4FGBkrAdlAVWnpY=
github.com/go-openapi/jsonreference v1.0.0/go.mod h1:jtwdyGbJk0Xhe5Y+rwtglQP6Sb1WZST4rT32LWB+sv0=
github.com/go-openapi/swag v0.27.1 h1:VotvOLWW8q/EAxB0YdsBBGC8XYyeL1YwBj2ungAGPNg=
github.com/go-openapi/swag v0.27.1/go.mod h1:GTkJPwHfhJp6MWr4/rCh64HVI3Ofu+tcsbfjfHmTxpE=
github.com/go-openapi/swag/cmdutils v0.27.1 h1:I7sYqaWVl5mq0NEmNQkAmFDyNin9ufvMX/p2zwtQaOE=
github.com/go-openapi/swag/cmdutils v0.27.1/go.mod h1:Sm1MVFMkF6guJJ+pQqHnQA3N0j9qALV3NxzDSv6bETM=
github.com/go-openapi/swag/conv v0.27.1 h1:8wi9ZG+olmY1wXphl93EWniPtbSPkXM/feH7FgjsvrU=
github.com/go-openapi/swag/conv v0.27.1/go.mod h1:QbqMivkpKhC3g1B1GGGOJ6ANewI3S62dbzYu3Duowqs=
github.com/go-openapi/swag/fileutils v0.27.1 h1:QQqBSoi5mW4XpU85nS0mLcA+zAE6vLzrb0QkmLKf9oM=
github.com/go-openapi/swag/fileutils v0.27.1/go.mod h1:VvJFZLTZS0AI854gEQz5tk7dBESdLjiNUMSZ/th2ry8=
github.com/go-openapi/swag/jsonutils v0.27.1 h1:SVgK3i4USzCU5mibOOS/l4ea2h9UQXy7J7RNLTjuXjU=
github.com/go-openapi/swag/jsonutils v0.27.1/go.mod h1:tdlEpZqdcQ17uj6J4YdK9vd8It5qWMwjWXOs0tjpRlk=
github.com/go-openapi/swag/jsonutils/fixtures_test v0.27.1 h1:mJu3COL9WEaZVp/Kf2PRMi7tPszPEJfSr/OO75ynCs8=
github.com/go-openapi/swag/jsonutils/fixtures_test v0.27.1/go.mod h1:mofwUWx70wvskwESqRJ//k/9kURmCgyJl5m5Ppoh5kY=
github.com/go-openapi/swag/loading v0.27.1 h1:/DxUgDXKbBX4bcn7r9uEXfJyzN5XpiJmZplzQTjrRCY=
github.com/go-openapi/swag/loading v0.27.1/go.mod h1:jvGh3iA2+zyUUycB5fgJWzeHnhrpvGnJJM0RVE9ZShE=
github.com/go-openapi/swag/mangling v0.27.1 h1:yC9D0HyUE8gbP+BfmGx9+AA89ikwZTMjESK3OnnoaqA=
github.com/go-openapi/swag/mangling v0.27.1/go.mod h1:jtBE2+V+3pILxOR7Vgce+Cwp6A2PgZbvVqfNntbVs0w=
github.com/go-openapi/swag/netutils v0.27.1 h1:mICMFoS82F5TZ4Zy3cqmcQk+BFeCp3Uyq3Np7GI0/qU=
github.com/go-openapi/swag/netutils v0.27.1/go.mod h1:J+WYyFMLtvtCGqa6jLv+YNUmIKI3ZRQRrvfNDMoQoEQ=
github.com/go-openapi/swag/pools v0.27.1 h1:9LeadcMyb2GJCbXX5hVQDbZ2Lq9TL4dCs/nx1j5DO0E=
github.com/go-openapi/swag/pools v0.27.1/go.mod h1:kVQefhSK5RWuRe7BXsL8htgBPAMpN7HDGpGEknqugeE=
github.com/go-openapi/swag/stringutils v0.27.1 h1:ZXePZ0r2p1qSjo8tD3Un4vFj8+FqlCkczxDrJIhYUp8=
github.com/go-openapi/swag/stringutils v0.27.1/go.mod h1:lzRN95CxXmA03XcDWHLOb6nOMcxCqR5rGY0lOgsfRoM=
github.com/go-openapi/swag/typeutils v0.27.1 h1:KSTdFlfnse4r6dP9IrEnwMldjE+zs71UeEB3//PtVXc=
github.com/go-openapi/swag/typeutils v0.27.1/go.mod h1:Srm0xFNRZ1Y+vCxJclo5qzx8aj+1pAKda/YfFPrG0dQ=
github.com/go-openapi/swag/yamlutils v0.27.1 h1:ftxv6xvXb1E3zohUc+okZ9nSqNb9StQX/FXnKZ98sQA=
github.com/go-openapi/swag/yamlutils v0.27.1/go.mod h1:bnxFIB1qewGRiZHypXGZ3fNgf13/0HfRgnS/iZBDrOo=
github.com/go-openapi/testify/enable/yaml/v2 v2.6.0 h1:gGHwAJ0R/5jU8BEGDbfRNR3hL68dAVi84WuOApp29B0=
github.com/go-openapi/testify/enable/yaml/v2 v2.6.0/go.mod h1:tY+St1SGq4NFl0QIqdTY4aEdbChAHxhyB77XQi9iJCo=
github.com/go-openapi/testify/v2 v2.6.0 h1:5PKH2HE7YJ/LuRPQGvSxBRlFXNQhSetBLlGAgUEu3ug=
github.com/go-openapi/testify/v2 v2.6.0/go.mod h1:SgsVHtfooshd0tublTtJ50FPKhujf47YRqauXXOUxfw=
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo=
github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8=
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/sirupsen/logrus v1.10.2 h1:G2SED73/qrAu6YwbdxOD6peLkCBI3z7L+ykJFTXJBBo=
github.com/sirupsen/logrus v1.10.2/go.mod h1:SLEg8TqYulVKKfIGHldVp2K2aYz2DKSVBq4g/H5bR7Q=
github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
github.com/spiffe/go-spiffe/v2 v2.8.2 h1:jUEsvCMD6fH25J8K/w3q/XnIx8W1lb8+YLaEEHIjHmc=
github.com/spiffe/go-spiffe/v2 v2.8.2/go.mod h1:w2CLWKLMTX/PPYUEUPv3ltH0RXsw5S8suwNF46w9/Aw=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4=
github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48=
go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8=
go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0=
go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs=
go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18=
go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE=
go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8=
go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew=
go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
go.yaml.in/yaml/v4 v4.0.0-rc.6 h1:1h7H1ohdUh93/FyE4YaDa1Zh64K6VVbjF4K6WUxMtH4=
go.yaml.in/yaml/v4 v4.0.0-rc.6/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk=
golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc=
golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww=
google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE=
google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ=
google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo=
gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M=
gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
k8s.io/api v0.37.0 h1:Z//Vj9N7RA/yS2sDmxyeo7h+RR4zbUrd2vrd3Z0TbB4=
k8s.io/api v0.37.0/go.mod h1:LKXgcJWMc+f4OLbP5SFR8rulEg07zZhpi/zMULiBImk=
k8s.io/apimachinery v0.37.0 h1:Np2AbDtf8x6RDHiD8T9LbKJ9gaegeVNa8yNm5FuGKm0=
k8s.io/apimachinery v0.37.0/go.mod h1:RN3nhprFSCxOi5Selxd7oMTXOe/c+ZbcE7Im+TS2zkE=
k8s.io/client-go v0.37.0 h1:nsN31fy8wBySuZ+QRnKmrjRSQLOG2rvoGN0tKd12zhQ=
k8s.io/client-go v0.37.0/go.mod h1:FcGqw+Ll/gNQiq+nPGY1Oyt9y7SgDh1d3MW3RFDEbn0=
k8s.io/klog/v2 v2.140.0 h1:Tf+J3AH7xnUzZyVVXhTgGhEKnFqye14aadWv7bzXdzc=
k8s.io/klog/v2 v2.140.0/go.mod h1:o+/RWfJ6PwpnFn7OyAG3QnO47BFsymfEfrz6XyYSSp0=
k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad h1:oXImqH8mQNk7PmvzKhmN3ddJoY6OnyM225MXwGHPm0A=
k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad/go.mod h1:0/mqHCVhlumdJ3BhCfnjSZQE037nAhNodh1/hK0T8/I=
k8s.io/utils v0.0.0-20260626114624-be93311217bd h1:Ea7fgQ5we8Y9T0OX5o0dAHzQOBRI07D/dEYRaB9ZZEs=
k8s.io/utils v0.0.0-20260626114624-be93311217bd/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk=
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg=
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU=
sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
sigs.k8s.io/structured-merge-diff/v6 v6.4.2 h1:qdOxHwrl2Kaag1aQEarlYcOA9vSyGCp3CIki3aW8c4Q=
sigs.k8s.io/structured-merge-diff/v6 v6.4.2/go.mod h1:M3W8sfWvn2HhQDIbGWj3S099YozAsymCo/wrT5ohRUE=
sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs=
sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4=
+213
View File
@@ -0,0 +1,213 @@
// Package assignmentqueue implements the durable assignment handoff with JetStream.
package assignmentqueue
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)
type publishAPI interface {
PublishMsg(context.Context, *nats.Msg, ...jetstream.PublishOpt) (*jetstream.PubAck, error)
}
// Publisher implements the scheduler dispatcher with one subject per backend.
type Publisher struct {
JetStream publishAPI
SubjectBase string
}
func (p Publisher) Dispatch(ctx context.Context, assignment taskassignment.Assignment) error {
if p.JetStream == nil {
return errors.New("JetStream publisher is required")
}
body, err := taskassignment.Marshal(assignment)
if err != nil {
return err
}
base := strings.TrimSuffix(p.SubjectBase, ".")
if base == "" {
return errors.New("assignment subject base is required")
}
message := &nats.Msg{
Subject: base + "." + string(assignment.Backend),
Header: nats.Header{jetstream.MsgIDHeader: []string{assignment.ID}},
Data: body,
}
if _, err := p.JetStream.PublishMsg(ctx, message); err != nil {
return fmt.Errorf("publish assignment %s: %w", assignment.ID, err)
}
return nil
}
type Accepter interface {
Accept(context.Context, taskassignment.Assignment) (bool, error)
}
type Claims interface {
Offer(taskassignment.Assignment) (<-chan struct{}, error)
WaitClaimed(context.Context, string) error
}
type Admission interface {
Acquire(string) bool
Release(string)
}
// Message is the subset of jetstream.Msg needed by one reconciliation.
type Message interface {
Data() []byte
DoubleAck(context.Context) error
NakWithDelay(time.Duration) error
TermWithReason(string) error
}
// Processor maps one delivery to one idempotent worker reconciliation.
type Processor struct {
TrustDomain string
Accepter Accepter
Claims Claims
Admission Admission
RetryDelay time.Duration
ClaimTimeout time.Duration
}
func (p Processor) Process(ctx context.Context, message Message) error {
if p.Accepter == nil || p.Claims == nil || p.Admission == nil {
return errors.New("assignment accepter, claim registry, and backend admission pool are required")
}
assignment, err := taskassignment.Unmarshal(message.Data(), p.TrustDomain)
if err != nil {
return errors.Join(err, message.TermWithReason("invalid assignment"))
}
if _, err := p.Claims.Offer(assignment); err != nil {
return errors.Join(err, message.TermWithReason("conflicting assignment"))
}
if !p.Admission.Acquire(assignment.ID) {
delay := p.RetryDelay
if delay <= 0 {
delay = 2 * time.Second
}
return message.NakWithDelay(delay)
}
accepted, err := p.Accepter.Accept(ctx, assignment)
if err != nil {
p.Admission.Release(assignment.ID)
delay := p.RetryDelay
if delay <= 0 {
delay = 15 * time.Second
}
return errors.Join(err, message.NakWithDelay(delay))
}
if accepted {
timeout := p.ClaimTimeout
if timeout <= 0 {
timeout = 4 * time.Minute
}
claimContext, cancel := context.WithTimeout(ctx, timeout)
err := p.Claims.WaitClaimed(claimContext, assignment.ID)
cancel()
if err != nil {
delay := p.RetryDelay
if delay <= 0 {
delay = 2 * time.Second
}
return errors.Join(err, message.NakWithDelay(delay))
}
if err := message.DoubleAck(ctx); err != nil {
return fmt.Errorf("ack assignment %s: %w", assignment.ID, err)
}
return nil
}
delay := p.RetryDelay
if delay <= 0 {
delay = 2 * time.Second
}
return message.NakWithDelay(delay)
}
type consumeAPI interface {
Consume(jetstream.MessageHandler, ...jetstream.PullConsumeOpt) (jetstream.ConsumeContext, error)
}
// ConsumerComponent runs bounded reconciliation goroutines for one durable
// backend consumer. The goroutine set is operational state, not task storage.
type ConsumerComponent struct {
Consumer consumeAPI
Processor Processor
Capacity int
OnError func(error)
}
type consumerManager interface {
CreateOrUpdateConsumer(context.Context, string, jetstream.ConsumerConfig) (jetstream.Consumer, error)
}
// OpenConsumer creates the durable backend cursor. Capacity is enforced both
// server-side and by ConsumerComponent's local semaphore.
func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectBase string, backend taskassignment.Backend, capacity int) (jetstream.Consumer, error) {
if manager == nil || stream == "" || strings.TrimSuffix(subjectBase, ".") == "" || capacity < 1 {
return nil, errors.New("JetStream manager, stream, subject base, and positive capacity are required")
}
if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
return nil, fmt.Errorf("unsupported assignment backend %q", backend)
}
consumer, err := manager.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{
Name: string(backend),
Durable: string(backend),
FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + string(backend),
AckPolicy: jetstream.AckExplicitPolicy,
AckWait: 5 * time.Minute,
MaxAckPending: capacity,
MaxDeliver: 20,
})
if err != nil {
return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)
}
return consumer, nil
}
func (c ConsumerComponent) Run(ctx context.Context) error {
if c.Consumer == nil || c.Capacity < 1 {
return errors.New("JetStream consumer and positive capacity are required")
}
semaphore := make(chan struct{}, c.Capacity)
var workers sync.WaitGroup
consumeContext, err := c.Consumer.Consume(func(message jetstream.Msg) {
select {
case semaphore <- struct{}{}:
case <-ctx.Done():
return
}
workers.Add(1)
go func() {
defer workers.Done()
defer func() { <-semaphore }()
if err := c.Processor.Process(ctx, message); err != nil && !errors.Is(err, context.Canceled) && c.OnError != nil {
c.OnError(err)
}
}()
}, jetstream.PullMaxMessages(c.Capacity))
if err != nil {
return fmt.Errorf("start JetStream consumer: %w", err)
}
select {
case <-ctx.Done():
consumeContext.Stop()
<-consumeContext.Closed()
workers.Wait()
return nil
case <-consumeContext.Closed():
workers.Wait()
return errors.New("JetStream consumer stopped unexpectedly")
}
}
+211
View File
@@ -0,0 +1,211 @@
package assignmentqueue
import (
"context"
"errors"
"testing"
"time"
runnerv1 "gitea.dev/actionslib/runner/v1"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"google.golang.org/protobuf/types/known/structpb"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)
func testAssignment(t *testing.T) taskassignment.Assignment {
t.Helper()
fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
if err != nil {
t.Fatal(err)
}
assignment, err := taskassignment.New(&runnerv1.Task{
Id: 42,
Context: fields,
WorkflowPayload: []byte("jobs:\n publish:\n runs-on: [self-hosted, pod]\n steps: []\n"),
}, "ddupan.top")
if err != nil {
t.Fatal(err)
}
return assignment
}
type fakePublisher struct{ message *nats.Msg }
func (p *fakePublisher) PublishMsg(_ context.Context, message *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) {
p.message = message
return &jetstream.PubAck{}, nil
}
func TestPublisherUsesBackendSubjectAndAssignmentDeduplication(t *testing.T) {
api := &fakePublisher{}
publisher := Publisher{JetStream: api, SubjectBase: "ci.assignment"}
if err := publisher.Dispatch(context.Background(), testAssignment(t)); err != nil {
t.Fatal(err)
}
if api.message.Subject != "ci.assignment.pod" {
t.Fatalf("subject = %q", api.message.Subject)
}
if api.message.Header.Get(jetstream.MsgIDHeader) != "gitea-task-42" {
t.Fatalf("message ID = %q", api.message.Header.Get(jetstream.MsgIDHeader))
}
}
type fakeAccepter struct {
accepted bool
err error
}
type fakeClaims struct {
claimed bool
}
type fakeAdmission struct {
allowed bool
active map[string]bool
released int
}
func (a *fakeAdmission) Acquire(assignmentID string) bool {
if !a.allowed {
return false
}
if a.active == nil {
a.active = make(map[string]bool)
}
a.active[assignmentID] = true
return true
}
func (a *fakeAdmission) Release(assignmentID string) {
delete(a.active, assignmentID)
a.released++
}
func (c *fakeClaims) Offer(taskassignment.Assignment) (<-chan struct{}, error) {
ready := make(chan struct{})
if c.claimed {
close(ready)
}
return ready, nil
}
func (c *fakeClaims) WaitClaimed(ctx context.Context, _ string) error {
if c.claimed {
return nil
}
<-ctx.Done()
return ctx.Err()
}
func (a *fakeAccepter) Accept(context.Context, taskassignment.Assignment) (bool, error) {
return a.accepted, a.err
}
type fakeMessage struct {
data []byte
acked int
nacked time.Duration
terminated int
}
func (m *fakeMessage) Data() []byte { return m.data }
func (m *fakeMessage) DoubleAck(context.Context) error { m.acked++; return nil }
func (m *fakeMessage) NakWithDelay(delay time.Duration) error { m.nacked = delay; return nil }
func (m *fakeMessage) TermWithReason(string) error { m.terminated++; return nil }
func encodedAssignment(t *testing.T) []byte {
t.Helper()
data, err := taskassignment.Marshal(testAssignment(t))
if err != nil {
t.Fatal(err)
}
return data
}
func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
message := &fakeMessage{data: encodedAssignment(t)}
processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}}
if err := processor.Process(context.Background(), message); err != nil {
t.Fatal(err)
}
if message.acked != 1 || message.nacked != 0 {
t.Fatalf("message = %#v", message)
}
}
func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
message := &fakeMessage{data: encodedAssignment(t)}
processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{}, Claims: &fakeClaims{}, Admission: &fakeAdmission{allowed: true}, RetryDelay: 2 * time.Second}
if err := processor.Process(context.Background(), message); err != nil {
t.Fatal(err)
}
if message.acked != 0 || message.nacked != 2*time.Second {
t.Fatalf("message = %#v", message)
}
}
func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T) {
retry := &fakeMessage{data: encodedAssignment(t)}
admission := &fakeAdmission{allowed: true}
processor := Processor{
TrustDomain: "ddupan.top",
Accepter: &fakeAccepter{err: errors.New("backend unavailable")},
Claims: &fakeClaims{},
Admission: admission,
RetryDelay: time.Minute,
}
if err := processor.Process(context.Background(), retry); err == nil {
t.Fatal("expected backend error")
}
if retry.nacked != time.Minute {
t.Fatalf("retry delay = %s", retry.nacked)
}
if admission.released != 1 {
t.Fatalf("released slots = %d", admission.released)
}
poison := &fakeMessage{data: []byte("not-json")}
if err := processor.Process(context.Background(), poison); err == nil {
t.Fatal("expected decode error")
}
if poison.terminated != 1 || poison.nacked != 0 {
t.Fatalf("poison message = %#v", poison)
}
}
func TestProcessorLeavesAssignmentPendingWhenBackendPoolIsFull(t *testing.T) {
message := &fakeMessage{data: encodedAssignment(t)}
accepter := &fakeAccepter{accepted: true}
processor := Processor{
TrustDomain: "ddupan.top",
Accepter: accepter,
Claims: &fakeClaims{},
Admission: &fakeAdmission{},
RetryDelay: 3 * time.Second,
}
if err := processor.Process(context.Background(), message); err != nil {
t.Fatal(err)
}
if message.acked != 0 || message.nacked != 3*time.Second {
t.Fatalf("message = %#v", message)
}
}
type fakeConsumerManager struct{ config jetstream.ConsumerConfig }
func (m *fakeConsumerManager) CreateOrUpdateConsumer(_ context.Context, _ string, config jetstream.ConsumerConfig) (jetstream.Consumer, error) {
m.config = config
return nil, nil
}
func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) {
manager := &fakeConsumerManager{}
if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.BackendPod, 4); err != nil {
t.Fatal(err)
}
if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 {
t.Fatalf("config = %#v", manager.config)
}
}
+50
View File
@@ -0,0 +1,50 @@
// Package backendpool manages runtime capacity independently for each executor backend.
package backendpool
import "sync"
type Pool struct {
mu sync.Mutex
capacity int
active map[string]struct{}
}
func New(capacity int) *Pool {
return &Pool{capacity: capacity, active: make(map[string]struct{})}
}
// Acquire reserves a backend slot without blocking. Redelivery of the same
// assignment is idempotent and succeeds even while the pool is full.
func (p *Pool) Acquire(assignmentID string) bool {
p.mu.Lock()
defer p.mu.Unlock()
if _, exists := p.active[assignmentID]; exists {
return true
}
if assignmentID == "" || len(p.active) >= p.capacity {
return false
}
p.active[assignmentID] = struct{}{}
return true
}
func (p *Pool) Restore(assignmentID string) {
if assignmentID == "" {
return
}
p.mu.Lock()
p.active[assignmentID] = struct{}{}
p.mu.Unlock()
}
func (p *Pool) Release(assignmentID string) {
p.mu.Lock()
delete(p.active, assignmentID)
p.mu.Unlock()
}
func (p *Pool) Active() int {
p.mu.Lock()
defer p.mu.Unlock()
return len(p.active)
}
+26
View File
@@ -0,0 +1,26 @@
package backendpool
import "testing"
func TestPoolSeparatesRuntimeCapacityFromDeliveries(t *testing.T) {
pool := New(2)
if !pool.Acquire("one") || !pool.Acquire("two") || pool.Acquire("three") {
t.Fatal("capacity was not enforced")
}
if !pool.Acquire("one") {
t.Fatal("redelivery must be idempotent")
}
pool.Release("one")
if !pool.Acquire("three") || pool.Active() != 2 {
t.Fatalf("active=%d", pool.Active())
}
}
func TestRestoreMayTemporarilyExceedReducedCapacity(t *testing.T) {
pool := New(1)
pool.Restore("one")
pool.Restore("two")
if pool.Active() != 2 || pool.Acquire("three") {
t.Fatalf("active=%d", pool.Active())
}
}
+75
View File
@@ -0,0 +1,75 @@
// Package controller composes independently runnable scheduler and backend workers.
package controller
import (
"context"
"errors"
"fmt"
"slices"
"strings"
"golang.org/x/sync/errgroup"
)
type ComponentName string
const (
Scheduler ComponentName = "scheduler"
PodWorker ComponentName = "pod-worker"
VMWorker ComponentName = "vm-worker"
)
var defaultComponents = []ComponentName{Scheduler, PodWorker, VMWorker}
// Selection parses --components. An empty value enables all components.
type Selection []ComponentName
func ParseSelection(value string) (Selection, error) {
if strings.TrimSpace(value) == "" || strings.TrimSpace(value) == "all" {
return append(Selection(nil), defaultComponents...), nil
}
var selected Selection
for _, raw := range strings.Split(value, ",") {
name := ComponentName(strings.TrimSpace(raw))
if !slices.Contains(defaultComponents, name) {
return nil, fmt.Errorf("unknown controller component %q", name)
}
if !slices.Contains(selected, name) {
selected = append(selected, name)
}
}
if len(selected) == 0 {
return nil, errors.New("at least one controller component is required")
}
return selected, nil
}
type Component interface {
Run(context.Context) error
}
type Registry map[ComponentName]Component
// Run starts exactly the selected components in one process. The first real
// failure cancels its peers; ordinary context cancellation is graceful.
func Run(ctx context.Context, selection Selection, registry Registry) error {
group, groupContext := errgroup.WithContext(ctx)
for _, name := range selection {
component, ok := registry[name]
if !ok || component == nil {
return fmt.Errorf("component %q is not configured", name)
}
name, component := name, component
group.Go(func() error {
err := component.Run(groupContext)
if errors.Is(err, context.Canceled) && groupContext.Err() != nil {
return nil
}
if err != nil {
return fmt.Errorf("component %s: %w", name, err)
}
return nil
})
}
return group.Wait()
}
+63
View File
@@ -0,0 +1,63 @@
package controller
import (
"context"
"errors"
"sync"
"testing"
)
func TestParseSelectionDefaultsToAll(t *testing.T) {
for _, input := range []string{"", "all"} {
selection, err := ParseSelection(input)
if err != nil {
t.Fatal(err)
}
if len(selection) != 3 || selection[0] != Scheduler || selection[1] != PodWorker || selection[2] != VMWorker {
t.Fatalf("selection = %v", selection)
}
}
}
func TestParseSelectionAllowsOneOrMoreComponents(t *testing.T) {
selection, err := ParseSelection("vm-worker,scheduler,vm-worker")
if err != nil {
t.Fatal(err)
}
if len(selection) != 2 || selection[0] != VMWorker || selection[1] != Scheduler {
t.Fatalf("selection = %v", selection)
}
if _, err := ParseSelection("webhook"); err == nil {
t.Fatal("expected obsolete component to be rejected")
}
}
type componentFunc func(context.Context) error
func (f componentFunc) Run(ctx context.Context) error { return f(ctx) }
func TestRunStartsSelectedComponentsAndCancelsPeers(t *testing.T) {
started := make(chan ComponentName, 2)
peerStopped := make(chan struct{})
var once sync.Once
registry := Registry{
Scheduler: componentFunc(func(context.Context) error {
started <- Scheduler
return errors.New("poll failed")
}),
PodWorker: componentFunc(func(ctx context.Context) error {
started <- PodWorker
<-ctx.Done()
once.Do(func() { close(peerStopped) })
return ctx.Err()
}),
}
err := Run(context.Background(), Selection{Scheduler, PodWorker}, registry)
if err == nil || !errors.Is(err, context.Canceled) && err.Error() != "component scheduler: poll failed" {
t.Fatalf("Run() error = %v", err)
}
<-peerStopped
if len(started) != 2 {
t.Fatalf("started components = %d", len(started))
}
}
+71
View File
@@ -0,0 +1,71 @@
// Package giteaactions provides the authenticated Gitea RunnerService client.
package giteaactions
import (
"context"
"net/http"
"strings"
"connectrpc.com/connect"
"gitea.dev/actionslib/pkg/protocol"
runnerv1 "gitea.dev/actionslib/runner/v1"
"gitea.dev/actionslib/runner/v1/runnerv1connect"
)
// Client is the subset of RunnerService owned by the scheduler.
type Client struct {
runner runnerv1connect.RunnerServiceClient
}
// NewClient authenticates every RPC with the persistent scheduler runner.
func NewClient(httpClient connect.HTTPClient, instanceURL, uuid, token string) *Client {
auth := connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
return func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
request.Header().Set("User-Agent", "gitea-dynamic-runner-go/0")
request.Header().Set(protocol.UUIDHeader, uuid)
request.Header().Set(protocol.TokenHeader, token)
return next(ctx, request)
}
})
baseURL := strings.TrimRight(instanceURL, "/") + "/api/actions"
return &Client{runner: runnerv1connect.NewRunnerServiceClient(
httpClient,
baseURL,
connect.WithInterceptors(auth),
)}
}
// Declare advertises the scheduler labels before tasks are fetched.
func (c *Client) Declare(ctx context.Context, version string, labels []string) error {
_, err := c.runner.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{
Version: version,
Labels: labels,
}))
return err
}
// FetchTask asks Gitea to atomically assign the next matching task.
func (c *Client) FetchTask(ctx context.Context, tasksVersion int64) (*runnerv1.FetchTaskResponse, error) {
response, err := c.runner.FetchTask(ctx, connect.NewRequest(&runnerv1.FetchTaskRequest{
TasksVersion: tasksVersion,
}))
if err != nil {
return nil, err
}
return response.Msg, nil
}
// UpdateTask forwards executor state through the scheduler runner identity.
func (c *Client) UpdateTask(ctx context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
return c.runner.UpdateTask(ctx, request)
}
// UpdateLog forwards executor log rows through the scheduler runner identity.
func (c *Client) UpdateLog(ctx context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
return c.runner.UpdateLog(ctx, request)
}
// DefaultHTTPClient is suitable for the scheduler's long-lived connection.
func DefaultHTTPClient() *http.Client {
return &http.Client{Transport: http.DefaultTransport}
}
+101
View File
@@ -0,0 +1,101 @@
package giteaactions
import (
"context"
"net/http"
"net/http/httptest"
"reflect"
"testing"
"connectrpc.com/connect"
"gitea.dev/actionslib/pkg/protocol"
runnerv1 "gitea.dev/actionslib/runner/v1"
"gitea.dev/actionslib/runner/v1/runnerv1connect"
)
type runnerService struct {
runnerv1connect.UnimplementedRunnerServiceHandler
t *testing.T
declaredLabels []string
fetchedVersion int64
updatedTask int64
updatedLog int64
expectedUUID string
expectedToken string
}
func (s *runnerService) checkAuth(request connect.AnyRequest) {
s.t.Helper()
if got := request.Header().Get(protocol.UUIDHeader); got != s.expectedUUID {
s.t.Errorf("runner UUID header = %q", got)
}
if got := request.Header().Get(protocol.TokenHeader); got != s.expectedToken {
s.t.Errorf("runner token header = %q", got)
}
}
func (s *runnerService) Declare(_ context.Context, request *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) {
s.checkAuth(request)
s.declaredLabels = request.Msg.Labels
return connect.NewResponse(&runnerv1.DeclareResponse{}), nil
}
func (s *runnerService) FetchTask(_ context.Context, request *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
s.checkAuth(request)
s.fetchedVersion = request.Msg.TasksVersion
return connect.NewResponse(&runnerv1.FetchTaskResponse{
Task: &runnerv1.Task{Id: 42},
TasksVersion: 8,
}), nil
}
func (s *runnerService) UpdateTask(_ context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
s.checkAuth(request)
s.updatedTask = request.Msg.GetState().GetId()
return connect.NewResponse(&runnerv1.UpdateTaskResponse{State: request.Msg.State}), nil
}
func (s *runnerService) UpdateLog(_ context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
s.checkAuth(request)
s.updatedLog = request.Msg.GetTaskId()
return connect.NewResponse(&runnerv1.UpdateLogResponse{}), nil
}
func TestClientUsesOfficialRunnerProtocol(t *testing.T) {
service := &runnerService{
t: t,
expectedUUID: "runner-uuid",
expectedToken: "runner-token",
}
path, handler := runnerv1connect.NewRunnerServiceHandler(service)
mux := http.NewServeMux()
mux.Handle("/api/actions"+path, http.StripPrefix("/api/actions", handler))
server := httptest.NewServer(mux)
defer server.Close()
client := NewClient(server.Client(), server.URL, service.expectedUUID, service.expectedToken)
labels := []string{"self-hosted", "pod", "vm"}
if err := client.Declare(context.Background(), "0.1.0", labels); err != nil {
t.Fatal(err)
}
response, err := client.FetchTask(context.Background(), 7)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(service.declaredLabels, labels) {
t.Fatalf("declared labels = %#v", service.declaredLabels)
}
if service.fetchedVersion != 7 || response.GetTasksVersion() != 8 || response.GetTask().GetId() != 42 {
t.Fatalf("unexpected FetchTask exchange: request=%d response=%v", service.fetchedVersion, response)
}
if _, err := client.UpdateTask(context.Background(), connect.NewRequest(&runnerv1.UpdateTaskRequest{State: &runnerv1.TaskState{Id: 42}})); err != nil {
t.Fatal(err)
}
if _, err := client.UpdateLog(context.Background(), connect.NewRequest(&runnerv1.UpdateLogRequest{TaskId: 42})); err != nil {
t.Fatal(err)
}
if service.updatedTask != 42 || service.updatedLog != 42 {
t.Fatalf("updated task=%d log=%d", service.updatedTask, service.updatedLog)
}
}
+310
View File
@@ -0,0 +1,310 @@
// Package opensandboxbackend implements the VM executor backend with the official OpenSandbox SDK.
package opensandboxbackend
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"net/http"
"sort"
"strings"
"time"
opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
)
const assignmentMetadata = "ci.ddupan.top/assignment-id"
const terminalMetadata = "ci.ddupan.top/terminal"
const annotationsMetadataPrefix = "ci.ddupan.top/annotations-"
const metadataValueLimit = 63
type Lifecycle interface {
ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error)
CreateSandbox(context.Context, opensandbox.CreateSandboxRequest) (*opensandbox.SandboxInfo, error)
GetSandbox(context.Context, string) (*opensandbox.SandboxInfo, error)
PatchSandboxMetadata(context.Context, string, opensandbox.MetadataPatch) (*opensandbox.SandboxInfo, error)
DeleteSandbox(context.Context, string) error
}
// MarkTerminal persists the accepted Gitea terminal state on the sandbox. The
// lifecycle reconciler performs deletion separately so the runner receives the
// successful UpdateTask response before its VM is stopped.
func (b Backend) MarkTerminal(ctx context.Context, assignmentID string) error {
executor, err := b.Find(ctx, assignmentID)
if err != nil || executor == nil {
return err
}
value := "true"
_, err = b.Lifecycle.PatchSandboxMetadata(ctx, executor.Name, opensandbox.MetadataPatch{
terminalMetadata: &value,
})
if err != nil {
return fmt.Errorf("mark sandbox %s terminal: %w", executor.Name, err)
}
return nil
}
// CleanupTerminated removes sandboxes whose terminal result was accepted by
// Gitea. The marker is stored by OpenSandbox, so cleanup survives restarts.
func (b Backend) CleanupTerminated(ctx context.Context) (int, error) {
if err := b.validate(); err != nil {
return 0, err
}
result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
Metadata: map[string]string{terminalMetadata: "true"},
PageSize: 100,
})
if err != nil {
return 0, fmt.Errorf("list terminal sandboxes: %w", err)
}
cleaned := 0
for _, sandbox := range result.Items {
if sandbox.Metadata[assignmentMetadata] == "" {
continue
}
if err := b.Delete(ctx, executor(sandbox)); err != nil {
return cleaned, fmt.Errorf("delete terminal sandbox %s: %w", sandbox.ID, err)
}
cleaned++
}
return cleaned, nil
}
// Lifecycle periodically reconciles durable terminal markers into deletes.
type LifecycleReconciler struct {
Backend Backend
Interval time.Duration
OnError func(error)
}
func (l LifecycleReconciler) Run(ctx context.Context) error {
interval := l.Interval
if interval <= 0 {
interval = 2 * time.Second
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
if _, err := l.Backend.CleanupTerminated(ctx); err != nil && ctx.Err() == nil && l.OnError != nil {
l.OnError(err)
}
select {
case <-ctx.Done():
return nil
case <-ticker.C:
}
}
}
type Config struct {
Pool string
Timeout int
Entrypoint []string
Env map[string]string
}
type Backend struct {
Lifecycle Lifecycle
Config Config
}
func (b Backend) RecoverAssignments(ctx context.Context, trustDomain string) ([]taskassignment.Assignment, error) {
if err := b.validate(); err != nil {
return nil, err
}
result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
Metadata: map[string]string{"ci.ddupan.top/backend": "vm"}, PageSize: 100,
})
if err != nil {
return nil, fmt.Errorf("list recoverable sandboxes: %w", err)
}
assignments := make([]taskassignment.Assignment, 0, len(result.Items))
for _, sandbox := range result.Items {
if sandbox.Metadata[terminalMetadata] == "true" {
continue
}
annotations, err := decodeAnnotations(sandbox.Metadata)
if err != nil {
return nil, fmt.Errorf("decode sandbox %s annotations: %w", sandbox.ID, err)
}
assignment, err := taskassignment.FromMetadata(sandbox.Metadata, annotations, trustDomain)
if err != nil {
return nil, fmt.Errorf("recover sandbox %s: %w", sandbox.ID, err)
}
assignments = append(assignments, assignment)
}
return assignments, nil
}
func NewLifecycleClient(baseURL, apiKey string, client *http.Client) *opensandbox.LifecycleClient {
if client != nil {
return opensandbox.NewLifecycleClient(baseURL, apiKey, opensandbox.WithHTTPClient(client))
}
return opensandbox.NewLifecycleClient(baseURL, apiKey)
}
func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Executor, error) {
if err := b.validate(); err != nil {
return nil, err
}
result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
Metadata: map[string]string{assignmentMetadata: assignmentID}, PageSize: 2,
})
if err != nil {
return nil, fmt.Errorf("list assignment sandboxes: %w", err)
}
if len(result.Items) > 1 {
return nil, fmt.Errorf("assignment %s owns %d sandboxes", assignmentID, len(result.Items))
}
if len(result.Items) == 0 {
return nil, nil
}
return executor(result.Items[0]), nil
}
func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, launch taskworker.LaunchSpec) (*taskworker.Executor, error) {
if err := b.validate(); err != nil {
return nil, err
}
if assignment.Backend != taskassignment.BackendVM {
return nil, fmt.Errorf("OpenSandbox backend cannot create %q assignment", assignment.Backend)
}
environment := clone(b.Config.Env)
for key, value := range launch.Environment {
environment[key] = value
}
sandboxMetadata := clone(launch.Metadata.Labels)
annotations, err := encodeAnnotations(launch.Metadata.Annotations)
if err != nil {
return nil, fmt.Errorf("encode sandbox annotations: %w", err)
}
for key, value := range annotations {
sandboxMetadata[key] = value
}
request := opensandbox.CreateSandboxRequest{
Timeout: &b.Config.Timeout, Entrypoint: append([]string{}, b.Config.Entrypoint...),
Env: environment, Metadata: sandboxMetadata,
Extensions: map[string]string{"poolRef": b.Config.Pool},
ResourceLimits: opensandbox.ResourceLimits{},
}
sandbox, err := b.Lifecycle.CreateSandbox(ctx, request)
if err != nil {
return nil, fmt.Errorf("create assignment sandbox: %w", err)
}
return executor(*sandbox), nil
}
func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor, identity taskidentity.Identity) error {
if executor == nil || executor.Name == "" {
return errors.New("sandbox ID is required for identity binding")
}
sandbox, err := b.Lifecycle.GetSandbox(ctx, executor.Name)
if err != nil {
return fmt.Errorf("verify sandbox identity metadata: %w", err)
}
annotations, err := decodeAnnotations(sandbox.Metadata)
if err != nil {
return fmt.Errorf("decode sandbox identity metadata: %w", err)
}
if annotations["ci.ddupan.top/spiffe-id"] != identity.SPIFFEID {
return fmt.Errorf("sandbox %s has inconsistent SPIFFE identity metadata", executor.Name)
}
return nil
}
func (b Backend) Delete(ctx context.Context, executor *taskworker.Executor) error {
if executor == nil || executor.Name == "" {
return nil
}
err := b.Lifecycle.DeleteSandbox(ctx, executor.Name)
var apiError *opensandbox.APIError
if errors.As(err, &apiError) && apiError.StatusCode == http.StatusNotFound {
return nil
}
return err
}
func (b Backend) validate() error {
if b.Lifecycle == nil || b.Config.Pool == "" || b.Config.Timeout < 1 || len(b.Config.Entrypoint) == 0 {
return errors.New("OpenSandbox Lifecycle client, pool, timeout, and entrypoint are required")
}
return nil
}
func executor(sandbox opensandbox.SandboxInfo) *taskworker.Executor {
return &taskworker.Executor{Name: sandbox.ID, IdentityTarget: sandbox.ID, Phase: phase(sandbox.Status.State)}
}
func phase(state opensandbox.SandboxState) taskworker.Phase {
switch state {
case opensandbox.StateRunning:
return taskworker.PhaseRunning
case opensandbox.StateTerminated, opensandbox.StateFailed:
// A successful executor reports Gitea before it exits. If Gitea is not
// terminal when the sandbox stops, termination is an execution failure.
return taskworker.PhaseFailed
default:
return taskworker.PhasePending
}
}
func clone(source map[string]string) map[string]string {
result := make(map[string]string, len(source))
for key, value := range source {
result[key] = value
}
return result
}
// OpenSandbox metadata follows Kubernetes label-value constraints, unlike Pod
// annotations. Store the annotation map as deterministic URL-safe base64
// chunks so repository paths and SPIFFE IDs remain lossless and recoverable.
func encodeAnnotations(annotations map[string]string) (map[string]string, error) {
data, err := json.Marshal(annotations)
if err != nil {
return nil, err
}
encoded := base64.RawURLEncoding.EncodeToString(data)
result := make(map[string]string, (len(encoded)+metadataValueLimit-1)/metadataValueLimit)
for index := 0; len(encoded) > 0; index++ {
length := min(metadataValueLimit, len(encoded))
result[fmt.Sprintf("%s%03d", annotationsMetadataPrefix, index)] = encoded[:length]
encoded = encoded[length:]
}
return result, nil
}
func decodeAnnotations(metadata map[string]string) (map[string]string, error) {
keys := make([]string, 0)
for key := range metadata {
if strings.HasPrefix(key, annotationsMetadataPrefix) {
keys = append(keys, key)
}
}
if len(keys) == 0 {
return nil, errors.New("sandbox annotation metadata is missing")
}
sort.Strings(keys)
var encoded strings.Builder
for index, key := range keys {
if key != fmt.Sprintf("%s%03d", annotationsMetadataPrefix, index) {
return nil, errors.New("sandbox annotation metadata chunks are incomplete")
}
encoded.WriteString(metadata[key])
}
data, err := base64.RawURLEncoding.DecodeString(encoded.String())
if err != nil {
return nil, err
}
var annotations map[string]string
if err := json.Unmarshal(data, &annotations); err != nil {
return nil, err
}
return annotations, nil
}
+150
View File
@@ -0,0 +1,150 @@
package opensandboxbackend
import (
"context"
"testing"
runnerv1 "gitea.dev/actionslib/runner/v1"
opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
)
type fakeLifecycle struct {
items []opensandbox.SandboxInfo
created opensandbox.CreateSandboxRequest
deleted string
}
func (f *fakeLifecycle) ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error) {
return &opensandbox.ListSandboxesResponse{Items: f.items}, nil
}
func (f *fakeLifecycle) CreateSandbox(_ context.Context, request opensandbox.CreateSandboxRequest) (*opensandbox.SandboxInfo, error) {
f.created = request
return &opensandbox.SandboxInfo{ID: "sandbox-42", Status: opensandbox.SandboxStatus{State: opensandbox.StatePending}, Metadata: request.Metadata}, nil
}
func (f *fakeLifecycle) GetSandbox(_ context.Context, id string) (*opensandbox.SandboxInfo, error) {
for _, item := range f.items {
if item.ID == id {
return &item, nil
}
}
metadata, _ := encodeAnnotations(map[string]string{"ci.ddupan.top/spiffe-id": assignment().Identity.SPIFFEID})
return &opensandbox.SandboxInfo{ID: id, Metadata: metadata}, nil
}
func (f *fakeLifecycle) PatchSandboxMetadata(_ context.Context, id string, patch opensandbox.MetadataPatch) (*opensandbox.SandboxInfo, error) {
for index := range f.items {
if f.items[index].ID != id {
continue
}
if f.items[index].Metadata == nil {
f.items[index].Metadata = map[string]string{}
}
for key, value := range patch {
if value != nil {
f.items[index].Metadata[key] = *value
}
}
return &f.items[index], nil
}
return &opensandbox.SandboxInfo{ID: id}, nil
}
func (f *fakeLifecycle) DeleteSandbox(_ context.Context, id string) error { f.deleted = id; return nil }
func backend(lifecycle Lifecycle) Backend {
return Backend{Lifecycle: lifecycle, Config: Config{
Pool: "ci-vm", Timeout: 14400,
Entrypoint: []string{"/usr/local/bin/gitea-task-executor"},
Env: map[string]string{"SPIFFE_ENDPOINT_SOCKET": "unix:///run/spire/agent-sockets/spire-agent.sock"},
}}
}
func assignment() taskassignment.Assignment {
return taskassignment.Assignment{
ID: "gitea-task-42", Backend: taskassignment.BackendVM,
Task: &runnerv1.Task{Id: 42},
Identity: taskidentity.Identity{Repository: "owner/repo", Task: "publish", SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"},
}
}
func TestFindRecoversSandboxByMetadata(t *testing.T) {
lifecycle := &fakeLifecycle{items: []opensandbox.SandboxInfo{{ID: "sandbox-42", Status: opensandbox.SandboxStatus{State: opensandbox.StateRunning}}}}
executor, err := backend(lifecycle).Find(context.Background(), assignment().ID)
if err != nil || executor.Name != "sandbox-42" || executor.Phase != taskworker.PhaseRunning {
t.Fatalf("executor=%#v err=%v", executor, err)
}
}
func TestCreateUsesPoolAndPersistsRecoveryMetadata(t *testing.T) {
lifecycle := &fakeLifecycle{}
metadata := taskworker.BackendMetadata(assignment())
executor, err := backend(lifecycle).Create(context.Background(), assignment(), taskworker.LaunchSpec{
Metadata: metadata,
Environment: map[string]string{
"CI_SPIFFE_ID": assignment().Identity.SPIFFEID,
"CI_RUNNER_CAPABILITY": "capability",
},
})
if err != nil {
t.Fatal(err)
}
if lifecycle.created.Extensions["poolRef"] != "ci-vm" || lifecycle.created.Metadata[assignmentMetadata] != assignment().ID || lifecycle.created.Env["CI_SPIFFE_ID"] != assignment().Identity.SPIFFEID || executor.Name != "sandbox-42" {
t.Fatalf("request=%#v executor=%#v", lifecycle.created, executor)
}
if lifecycle.created.Env["CI_RUNNER_CAPABILITY"] != "capability" {
t.Fatalf("environment = %#v", lifecycle.created.Env)
}
annotations, err := decodeAnnotations(lifecycle.created.Metadata)
if err != nil || annotations["ci.ddupan.top/repository"] != "owner/repo" || annotations["ci.ddupan.top/spiffe-id"] != assignment().Identity.SPIFFEID {
t.Fatalf("annotations=%#v err=%v", annotations, err)
}
for _, value := range lifecycle.created.Metadata {
if len(value) > metadataValueLimit {
t.Fatalf("metadata value exceeds %d characters: %q", metadataValueLimit, value)
}
}
}
func TestAnnotationMetadataRoundTripPreservesSlashValues(t *testing.T) {
want := taskworker.BackendMetadata(assignment()).Annotations
encoded, err := encodeAnnotations(want)
if err != nil {
t.Fatal(err)
}
got, err := decodeAnnotations(encoded)
if err != nil {
t.Fatal(err)
}
for key, value := range want {
if got[key] != value {
t.Fatalf("%s=%q, want %q", key, got[key], value)
}
}
}
func TestBindIdentityVerifiesPersistedMetadata(t *testing.T) {
lifecycle := &fakeLifecycle{}
if err := backend(lifecycle).BindIdentity(context.Background(), &taskworker.Executor{Name: "sandbox-42"}, assignment().Identity); err != nil {
t.Fatal(err)
}
}
func TestTerminalMarkerDrivesDurableCleanup(t *testing.T) {
lifecycle := &fakeLifecycle{items: []opensandbox.SandboxInfo{{
ID: "sandbox-42",
Metadata: map[string]string{assignmentMetadata: assignment().ID},
}}}
backend := backend(lifecycle)
if err := backend.MarkTerminal(context.Background(), assignment().ID); err != nil {
t.Fatal(err)
}
if lifecycle.items[0].Metadata[terminalMetadata] != "true" {
t.Fatalf("metadata = %#v", lifecycle.items[0].Metadata)
}
cleaned, err := backend.CleanupTerminated(context.Background())
if err != nil || cleaned != 1 || lifecycle.deleted != "sandbox-42" {
t.Fatalf("cleaned=%d deleted=%q err=%v", cleaned, lifecycle.deleted, err)
}
}
+266
View File
@@ -0,0 +1,266 @@
// Package podbackend implements the native homelab Kubernetes executor backend.
package podbackend
import (
"context"
"errors"
"fmt"
"net/url"
"strings"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
)
const (
assignmentLabel = "ci.ddupan.top/assignment-id"
terminalLabel = "ci.ddupan.top/terminal"
)
// Pod is the backend state required by the reconciler, not an in-memory lifecycle record.
type Pod struct {
Name string
Namespace string
UID string
Phase string
Labels map[string]string
Annotations map[string]string
}
// PodManifest leaves task execution wiring to the executor image while fixing
// the metadata contract needed for recovery.
type PodManifest struct {
Name string
Namespace string
Labels map[string]string
Annotations map[string]string
Image string
ServiceAccount string
Args []string
Environment map[string]string
}
// IdentityEntry is a ClusterStaticEntry pinned to one concrete Pod UID.
type IdentityEntry struct {
Name string
Labels map[string]string
ClassName string
ParentID string
SPIFFEID string
Selectors []string
}
// API is the narrow Kubernetes boundary used by the backend adapter.
type API interface {
ListPods(context.Context, string, string) ([]Pod, error)
CreatePod(context.Context, PodManifest) (Pod, error)
DeletePod(context.Context, string, string) error
LabelPod(context.Context, string, string, map[string]string) error
EnsureIdentityEntry(context.Context, IdentityEntry) error
DeleteIdentityEntry(context.Context, string) error
}
// MarkTerminal persists Gitea's accepted terminal state on the backend
// resource. Cleanup can therefore resume after a controller restart.
func (b Backend) MarkTerminal(ctx context.Context, assignmentID string) error {
if err := b.validate(); err != nil {
return err
}
if assignmentID == "" {
return errors.New("assignment ID is required")
}
if err := b.API.LabelPod(ctx, b.Config.Namespace, assignmentID, map[string]string{terminalLabel: "true"}); err != nil {
return fmt.Errorf("mark assignment Pod terminal: %w", err)
}
return nil
}
// CleanupTerminated removes only executors whose terminal update was accepted
// by Gitea and whose process has exited.
func (b Backend) CleanupTerminated(ctx context.Context) (int, error) {
if err := b.validate(); err != nil {
return 0, err
}
pods, err := b.API.ListPods(ctx, b.Config.Namespace, terminalLabel+"=true")
if err != nil {
return 0, fmt.Errorf("list terminal assignment Pods: %w", err)
}
cleaned := 0
for _, pod := range pods {
if pod.Phase != "Succeeded" && pod.Phase != "Failed" {
continue
}
if err := b.Delete(ctx, executor(pod)); err != nil {
return cleaned, err
}
cleaned++
}
return cleaned, nil
}
type Config struct {
Namespace string
Image string
ServiceAccount string
ExecutorArgs []string
TrustDomain string
SPIRECluster string
SPIREClass string
SPIREAgentID string
ExecutorUID int
}
type Backend struct {
API API
Config Config
}
func (b Backend) RecoverAssignments(ctx context.Context) ([]taskassignment.Assignment, error) {
if err := b.validate(); err != nil {
return nil, err
}
pods, err := b.API.ListPods(ctx, b.Config.Namespace, "ci.ddupan.top/backend=pod")
if err != nil {
return nil, fmt.Errorf("list recoverable assignment Pods: %w", err)
}
assignments := make([]taskassignment.Assignment, 0, len(pods))
for _, pod := range pods {
if pod.Labels[terminalLabel] == "true" {
continue
}
assignment, err := taskassignment.FromMetadata(pod.Labels, pod.Annotations, b.Config.TrustDomain)
if err != nil {
return nil, fmt.Errorf("recover Pod %s: %w", pod.Name, err)
}
assignments = append(assignments, assignment)
}
return assignments, nil
}
func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Executor, error) {
if err := b.validate(); err != nil {
return nil, err
}
pods, err := b.API.ListPods(ctx, b.Config.Namespace, assignmentLabel+"="+assignmentID)
if err != nil {
return nil, fmt.Errorf("list assignment Pods: %w", err)
}
if len(pods) > 1 {
return nil, fmt.Errorf("assignment %s owns %d Pods", assignmentID, len(pods))
}
if len(pods) == 0 {
return nil, nil
}
return executor(pods[0]), nil
}
func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, launch taskworker.LaunchSpec) (*taskworker.Executor, error) {
if err := b.validate(); err != nil {
return nil, err
}
if assignment.Backend != taskassignment.BackendPod {
return nil, fmt.Errorf("Pod backend cannot create %q assignment", assignment.Backend)
}
labels := clone(launch.Metadata.Labels)
labels["app.kubernetes.io/name"] = "gitea-dynamic-runner"
labels["app.kubernetes.io/component"] = "executor"
pod, err := b.API.CreatePod(ctx, PodManifest{
Name: assignment.ID,
Namespace: b.Config.Namespace,
Labels: labels,
Annotations: clone(launch.Metadata.Annotations),
Image: b.Config.Image,
ServiceAccount: b.Config.ServiceAccount,
Args: append([]string{}, b.Config.ExecutorArgs...),
Environment: clone(launch.Environment),
})
if err != nil {
return nil, fmt.Errorf("create assignment Pod: %w", err)
}
return executor(pod), nil
}
func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor, identity taskidentity.Identity) error {
if err := b.validate(); err != nil {
return err
}
if executor == nil || executor.Name == "" || executor.IdentityTarget == "" {
return errors.New("Pod name and UID are required for identity binding")
}
if _, err := identityPath(identity.SPIFFEID, b.Config.TrustDomain); err != nil {
return err
}
return b.API.EnsureIdentityEntry(ctx, IdentityEntry{
Name: executor.Name,
Labels: map[string]string{
"app.kubernetes.io/name": "gitea-dynamic-runner",
"app.kubernetes.io/component": "pod-identity",
assignmentLabel: executor.Name,
},
ClassName: b.Config.SPIREClass,
ParentID: b.Config.SPIREAgentID,
SPIFFEID: identity.SPIFFEID,
Selectors: []string{"k8s:pod-uid:" + executor.IdentityTarget},
})
}
func identityPath(spiffeID, trustDomain string) (string, error) {
parsed, err := url.Parse(spiffeID)
if err != nil || parsed.Scheme != "spiffe" || parsed.Host != trustDomain || !strings.HasPrefix(parsed.Path, "/ci/") {
return "", fmt.Errorf("invalid CI SPIFFE ID %q", spiffeID)
}
return strings.TrimPrefix(parsed.Path, "/ci/"), nil
}
func (b Backend) Delete(ctx context.Context, executor *taskworker.Executor) error {
if err := b.validate(); err != nil {
return err
}
if executor == nil || executor.Name == "" {
return nil
}
if err := b.API.DeleteIdentityEntry(ctx, executor.Name); err != nil {
return fmt.Errorf("delete Pod identity entry: %w", err)
}
if err := b.API.DeletePod(ctx, b.Config.Namespace, executor.Name); err != nil {
return fmt.Errorf("delete assignment Pod: %w", err)
}
return nil
}
func (b Backend) validate() error {
if b.API == nil || b.Config.Namespace == "" || b.Config.Image == "" || b.Config.ServiceAccount == "" || b.Config.TrustDomain == "" || b.Config.SPIRECluster == "" || b.Config.SPIREClass == "" || b.Config.SPIREAgentID == "" || b.Config.ExecutorUID < 1 {
return errors.New("Pod API and complete executor/SPIRE configuration are required")
}
return nil
}
func executor(pod Pod) *taskworker.Executor {
return &taskworker.Executor{
Name: pod.Name,
IdentityTarget: pod.UID,
Phase: phase(pod.Phase),
}
}
func phase(value string) taskworker.Phase {
switch value {
case "Succeeded":
return taskworker.PhaseSucceeded
case "Failed":
return taskworker.PhaseFailed
case "Running":
return taskworker.PhaseRunning
default:
return taskworker.PhasePending
}
}
func clone(source map[string]string) map[string]string {
target := make(map[string]string, len(source))
for key, value := range source {
target[key] = value
}
return target
}
+153
View File
@@ -0,0 +1,153 @@
package podbackend
import (
"context"
"testing"
runnerv1 "gitea.dev/actionslib/runner/v1"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker"
)
type fakeAPI struct {
pods []Pod
selector string
created PodManifest
entry IdentityEntry
entryGone string
deleted string
marked map[string]string
}
func (a *fakeAPI) ListPods(_ context.Context, _ string, selector string) ([]Pod, error) {
a.selector = selector
return a.pods, nil
}
func (a *fakeAPI) CreatePod(_ context.Context, manifest PodManifest) (Pod, error) {
a.created = manifest
return Pod{Name: manifest.Name, Namespace: manifest.Namespace, UID: "pod-uid", Phase: "Pending"}, nil
}
func (a *fakeAPI) DeletePod(_ context.Context, _, name string) error {
a.deleted = name
return nil
}
func (a *fakeAPI) LabelPod(_ context.Context, _, _ string, labels map[string]string) error {
a.marked = labels
return nil
}
func (a *fakeAPI) EnsureIdentityEntry(_ context.Context, entry IdentityEntry) error {
a.entry = entry
return nil
}
func (a *fakeAPI) DeleteIdentityEntry(_ context.Context, name string) error {
a.entryGone = name
return nil
}
func backend(api API) Backend {
return Backend{API: api, Config: Config{
Namespace: "gitea-actions", Image: "zot/ci-executor:main",
ServiceAccount: "gitea-task-executor", ExecutorArgs: []string{"executor"},
TrustDomain: "ddupan.top", SPIRECluster: "homelab",
SPIREClass: "spire-mgmt-spire",
SPIREAgentID: "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/node-uid",
ExecutorUID: 2000,
}}
}
func assignment() taskassignment.Assignment {
return taskassignment.Assignment{
ID: "gitea-task-42", Backend: taskassignment.BackendPod,
Task: &runnerv1.Task{Id: 42},
Identity: taskidentity.Identity{
Repository: "owner/repo", Task: "publish",
SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish",
},
}
}
func TestFindRecoversPodByAssignmentLabel(t *testing.T) {
api := &fakeAPI{pods: []Pod{{Name: "gitea-task-42", UID: "uid", Phase: "Running"}}}
executor, err := backend(api).Find(context.Background(), "gitea-task-42")
if err != nil {
t.Fatal(err)
}
if api.selector != "ci.ddupan.top/assignment-id=gitea-task-42" || executor.Name != "gitea-task-42" || executor.IdentityTarget != "uid" || executor.Phase != taskworker.PhaseRunning {
t.Fatalf("selector=%q executor=%#v", api.selector, executor)
}
}
func TestCreateUsesDeterministicNameAndRecoveryMetadata(t *testing.T) {
api := &fakeAPI{}
metadata := taskworker.BackendMetadata(assignment())
executor, err := backend(api).Create(context.Background(), assignment(), taskworker.LaunchSpec{
Metadata: metadata, Environment: map[string]string{"CI_RUNNER_CAPABILITY": "capability"},
})
if err != nil {
t.Fatal(err)
}
if api.created.Name != "gitea-task-42" || api.created.Labels[assignmentLabel] != "gitea-task-42" {
t.Fatalf("manifest = %#v", api.created)
}
if api.created.Annotations["ci.ddupan.top/spiffe-id"] != assignment().Identity.SPIFFEID {
t.Fatalf("annotations = %#v", api.created.Annotations)
}
if api.created.Environment["CI_RUNNER_CAPABILITY"] != "capability" {
t.Fatalf("environment = %#v", api.created.Environment)
}
if len(api.created.Args) != 1 || api.created.Args[0] != "executor" || executor.IdentityTarget != "pod-uid" {
t.Fatalf("args=%v executor=%#v", api.created.Args, executor)
}
}
func TestBindIdentityCreatesEntryPinnedToPodUID(t *testing.T) {
api := &fakeAPI{}
executor := &taskworker.Executor{Name: "gitea-task-42", IdentityTarget: "pod-uid"}
if err := backend(api).BindIdentity(context.Background(), executor, assignment().Identity); err != nil {
t.Fatal(err)
}
if api.entry.Name != executor.Name || api.entry.SPIFFEID != assignment().Identity.SPIFFEID || api.entry.ParentID != "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/node-uid" || len(api.entry.Selectors) != 1 || api.entry.Selectors[0] != "k8s:pod-uid:pod-uid" {
t.Fatalf("entry=%#v", api.entry)
}
}
func TestDeleteRemovesIdentityBeforePod(t *testing.T) {
api := &fakeAPI{}
executor := &taskworker.Executor{Name: "gitea-task-42", IdentityTarget: "pod-uid"}
if err := backend(api).Delete(context.Background(), executor); err != nil {
t.Fatal(err)
}
if api.entryGone != executor.Name || api.deleted != executor.Name {
t.Fatalf("entry=%q pod=%q", api.entryGone, api.deleted)
}
}
func TestTerminalMarkerAndCleanupUseBackendState(t *testing.T) {
api := &fakeAPI{pods: []Pod{
{Name: "running", UID: "running-uid", Phase: "Running"},
{Name: "finished", UID: "finished-uid", Phase: "Succeeded"},
}}
backend := backend(api)
if err := backend.MarkTerminal(context.Background(), "gitea-task-42"); err != nil {
t.Fatal(err)
}
if api.marked[terminalLabel] != "true" {
t.Fatalf("labels = %#v", api.marked)
}
cleaned, err := backend.CleanupTerminated(context.Background())
if err != nil {
t.Fatal(err)
}
if api.selector != terminalLabel+"=true" || cleaned != 1 || api.deleted != "finished" || api.entryGone != "finished" {
t.Fatalf("selector=%q cleaned=%d deleted=%q entry=%q", api.selector, cleaned, api.deleted, api.entryGone)
}
}
func TestFindRejectsDuplicatePods(t *testing.T) {
api := &fakeAPI{pods: []Pod{{Name: "one"}, {Name: "two"}}}
if _, err := backend(api).Find(context.Background(), "gitea-task-42"); err == nil {
t.Fatal("expected duplicate executor error")
}
}
+178
View File
@@ -0,0 +1,178 @@
package podbackend
import (
"context"
"encoding/json"
"fmt"
"reflect"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
)
var identityEntryResource = schema.GroupVersionResource{
Group: "spire.spiffe.io", Version: "v1alpha1", Resource: "clusterstaticentries",
}
// Client uses client-go's typed client for Pods and its dynamic client for the
// SPIRE Operator CRD.
type Client struct {
Kubernetes kubernetes.Interface
Dynamic dynamic.Interface
}
func NewClient(config *rest.Config) (*Client, error) {
kubernetesClient, err := kubernetes.NewForConfig(config)
if err != nil {
return nil, fmt.Errorf("create Kubernetes client: %w", err)
}
dynamicClient, err := dynamic.NewForConfig(config)
if err != nil {
return nil, fmt.Errorf("create Kubernetes dynamic client: %w", err)
}
return &Client{Kubernetes: kubernetesClient, Dynamic: dynamicClient}, nil
}
func NewInClusterClient() (*Client, error) {
config, err := rest.InClusterConfig()
if err != nil {
return nil, fmt.Errorf("load in-cluster Kubernetes config: %w", err)
}
return NewClient(config)
}
func (c *Client) ListPods(ctx context.Context, namespace, selector string) ([]Pod, error) {
list, err := c.Kubernetes.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector})
if err != nil {
return nil, err
}
pods := make([]Pod, 0, len(list.Items))
for _, item := range list.Items {
pods = append(pods, podFromKubernetes(item))
}
return pods, nil
}
func (c *Client) CreatePod(ctx context.Context, manifest PodManifest) (Pod, error) {
environment := make([]corev1.EnvVar, 0, len(manifest.Environment))
for name, value := range manifest.Environment {
environment = append(environment, corev1.EnvVar{Name: name, Value: value})
}
document := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: manifest.Name, Namespace: manifest.Namespace,
Labels: manifest.Labels, Annotations: manifest.Annotations,
},
Spec: corev1.PodSpec{
ServiceAccountName: manifest.ServiceAccount,
RestartPolicy: corev1.RestartPolicyNever,
Containers: []corev1.Container{{
Name: "executor", Image: manifest.Image, Args: manifest.Args, Env: environment,
SecurityContext: &corev1.SecurityContext{Privileged: boolPointer(true)},
VolumeMounts: []corev1.VolumeMount{{
Name: "spire-agent-socket", MountPath: "/run/spire/agent-sockets", ReadOnly: true,
}},
}},
Volumes: []corev1.Volume{{
Name: "spire-agent-socket",
VolumeSource: corev1.VolumeSource{CSI: &corev1.CSIVolumeSource{
Driver: "csi.spiffe.io", ReadOnly: boolPointer(true),
}},
}},
},
}
created, err := c.Kubernetes.CoreV1().Pods(manifest.Namespace).Create(ctx, document, metav1.CreateOptions{})
if err != nil {
return Pod{}, err
}
return podFromKubernetes(*created), nil
}
func (c *Client) DeletePod(ctx context.Context, namespace, name string) error {
policy := metav1.DeletePropagationBackground
err := c.Kubernetes.CoreV1().Pods(namespace).Delete(ctx, name, metav1.DeleteOptions{PropagationPolicy: &policy})
if apierrors.IsNotFound(err) {
return nil
}
return err
}
func (c *Client) LabelPod(ctx context.Context, namespace, name string, labels map[string]string) error {
patch, err := json.Marshal(map[string]any{"metadata": map[string]any{"labels": labels}})
if err != nil {
return err
}
_, err = c.Kubernetes.CoreV1().Pods(namespace).Patch(ctx, name, types.MergePatchType, patch, metav1.PatchOptions{})
return err
}
func (c *Client) EnsureIdentityEntry(ctx context.Context, entry IdentityEntry) error {
resource := c.Dynamic.Resource(identityEntryResource)
existing, err := resource.Get(ctx, entry.Name, metav1.GetOptions{})
if err == nil {
existingSpec, _, nestedErr := unstructured.NestedMap(existing.Object, "spec")
if nestedErr != nil {
return nestedErr
}
if !reflect.DeepEqual(existingSpec, identityEntryObject(entry).Object["spec"]) {
return fmt.Errorf("identity entry %s exists with different selectors or SPIFFE ID", entry.Name)
}
return nil
}
if !apierrors.IsNotFound(err) {
return err
}
_, err = resource.Create(ctx, identityEntryObject(entry), metav1.CreateOptions{})
return err
}
func (c *Client) DeleteIdentityEntry(ctx context.Context, name string) error {
policy := metav1.DeletePropagationBackground
err := c.Dynamic.Resource(identityEntryResource).Delete(ctx, name, metav1.DeleteOptions{PropagationPolicy: &policy})
if apierrors.IsNotFound(err) {
return nil
}
return err
}
func identityEntryObject(entry IdentityEntry) *unstructured.Unstructured {
selectors := make([]any, len(entry.Selectors))
for index, selector := range entry.Selectors {
selectors[index] = selector
}
return &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "spire.spiffe.io/v1alpha1",
"kind": "ClusterStaticEntry",
"metadata": map[string]any{
"name": entry.Name, "labels": stringMap(entry.Labels),
},
"spec": map[string]any{
"className": entry.ClassName, "parentID": entry.ParentID,
"spiffeID": entry.SPIFFEID, "selectors": selectors,
},
}}
}
func podFromKubernetes(pod corev1.Pod) Pod {
return Pod{
Name: pod.Name, Namespace: pod.Namespace, UID: string(pod.UID),
Phase: string(pod.Status.Phase), Labels: pod.Labels, Annotations: pod.Annotations,
}
}
func boolPointer(value bool) *bool { return &value }
func stringMap(values map[string]string) map[string]any {
result := make(map[string]any, len(values))
for key, value := range values {
result[key] = value
}
return result
}
+82
View File
@@ -0,0 +1,82 @@
package podbackend
import (
"context"
"testing"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
dynamicfake "k8s.io/client-go/dynamic/fake"
"k8s.io/client-go/kubernetes/fake"
)
func testClient(objects ...runtime.Object) *Client {
return &Client{
Kubernetes: fake.NewSimpleClientset(objects...),
Dynamic: dynamicfake.NewSimpleDynamicClient(runtime.NewScheme()),
}
}
func TestClientPodLifecycleUsesTypedClient(t *testing.T) {
client := testClient()
created, err := client.CreatePod(context.Background(), PodManifest{
Name: "gitea-task-42", Namespace: "gitea-actions",
Labels: map[string]string{assignmentLabel: "gitea-task-42"},
Image: "zot/ci-executor:main", ServiceAccount: "gitea-task-executor",
Args: []string{"execute", "gitea-task-42"}, Environment: map[string]string{"CI_RUNNER_CAPABILITY": "capability"},
})
if err != nil {
t.Fatal(err)
}
pod, err := client.Kubernetes.CoreV1().Pods("gitea-actions").Get(context.Background(), created.Name, metav1.GetOptions{})
if err != nil {
t.Fatal(err)
}
if got := pod.Spec.Containers[0].Env; len(got) != 1 || got[0].Name != "CI_RUNNER_CAPABILITY" || got[0].Value != "capability" {
t.Fatalf("environment = %#v", got)
}
if err := client.LabelPod(context.Background(), "gitea-actions", created.Name, map[string]string{terminalLabel: "true"}); err != nil {
t.Fatal(err)
}
pod, err = client.Kubernetes.CoreV1().Pods("gitea-actions").Get(context.Background(), created.Name, metav1.GetOptions{})
if err != nil || pod.Labels[terminalLabel] != "true" {
t.Fatalf("terminal label pod=%#v err=%v", pod, err)
}
pod.UID = types.UID("pod-uid")
pod.Status.Phase = corev1.PodRunning
if _, err := client.Kubernetes.CoreV1().Pods("gitea-actions").Update(context.Background(), pod, metav1.UpdateOptions{}); err != nil {
t.Fatal(err)
}
pods, err := client.ListPods(context.Background(), "gitea-actions", assignmentLabel+"=gitea-task-42")
if err != nil || len(pods) != 1 || pods[0].UID != "pod-uid" || pods[0].Phase != "Running" {
t.Fatalf("pods=%#v err=%v", pods, err)
}
if err := client.DeletePod(context.Background(), "gitea-actions", created.Name); err != nil {
t.Fatal(err)
}
}
func TestClientEnsuresIdempotentClusterStaticEntry(t *testing.T) {
client := testClient()
entry := IdentityEntry{
Name: "gitea-task-42", Labels: map[string]string{assignmentLabel: "gitea-task-42"},
ClassName: "spire-mgmt-spire",
ParentID: "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/pod/pod-uid",
SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish",
Selectors: []string{"unix:uid:2000"},
}
if err := client.EnsureIdentityEntry(context.Background(), entry); err != nil {
t.Fatal(err)
}
if err := client.EnsureIdentityEntry(context.Background(), entry); err != nil {
t.Fatal(err)
}
if err := client.DeleteIdentityEntry(context.Background(), entry.Name); err != nil {
t.Fatal(err)
}
if err := client.DeleteIdentityEntry(context.Background(), entry.Name); err != nil {
t.Fatal(err)
}
}
+32
View File
@@ -0,0 +1,32 @@
package podbackend
import (
"context"
"time"
)
// Lifecycle reconciles durable terminal markers into backend cleanup.
type Lifecycle struct {
Backend Backend
Interval time.Duration
OnError func(error)
}
func (l Lifecycle) Run(ctx context.Context) error {
interval := l.Interval
if interval <= 0 {
interval = 2 * time.Second
}
for {
if _, err := l.Backend.CleanupTerminated(ctx); err != nil && ctx.Err() == nil && l.OnError != nil {
l.OnError(err)
}
timer := time.NewTimer(interval)
select {
case <-ctx.Done():
timer.Stop()
return nil
case <-timer.C:
}
}
}
+91
View File
@@ -0,0 +1,91 @@
// Package runnerbootstrap configures an unmodified, one-shot Gitea Runner to
// consume exactly the task assigned by the controller facade.
package runnerbootstrap
import (
"encoding/json"
"errors"
"fmt"
"net/url"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)
const (
EnvAssignmentID = "CI_ASSIGNMENT_ID"
EnvCapability = "CI_RUNNER_CAPABILITY"
EnvFacadeURL = "CI_RUNNER_FACADE_URL"
EnvFacadeID = "CI_RUNNER_FACADE_SPIFFE_ID"
EnvSPIFFEID = "CI_SPIFFE_ID"
EnvBackend = "CI_RUNNER_BACKEND"
)
// Bootstrap emits assignment-scoped launch configuration. FacadeURL is the
// controller endpoint reached by the local SPIFFE proxy, not by Runner itself.
type Bootstrap struct {
Capabilities runnerfacade.Capabilities
FacadeURL string
FacadeSPIFFEID string
WorkloadAPIAddr string
}
func (b Bootstrap) Environment(assignment taskassignment.Assignment) (map[string]string, error) {
if assignment.ID == "" || assignment.Identity.SPIFFEID == "" || b.FacadeSPIFFEID == "" || b.WorkloadAPIAddr == "" {
return nil, errors.New("assignment ID and SPIFFE ID are required")
}
parsed, err := url.Parse(b.FacadeURL)
if err != nil || parsed.Scheme != "https" || parsed.Host == "" {
return nil, fmt.Errorf("runner facade URL must be an absolute https URL")
}
capability := b.Capabilities.Issue(assignment.ID)
if capability == "" {
return nil, errors.New("runner capability issuer is not configured")
}
return map[string]string{
EnvAssignmentID: assignment.ID,
EnvCapability: capability,
EnvFacadeURL: b.FacadeURL,
EnvFacadeID: b.FacadeSPIFFEID,
EnvSPIFFEID: assignment.Identity.SPIFFEID,
EnvBackend: string(assignment.Backend),
"SPIFFE_ENDPOINT_SOCKET": b.WorkloadAPIAddr,
}, nil
}
// Registration mirrors Gitea Runner v3.5.0's registration file schema. ID is
// intentionally zero: the facade authenticates UUID and token and never uses
// the server-issued runner database ID.
type Registration struct {
Warning string `json:"WARNING"`
ID int64 `json:"id"`
UUID string `json:"uuid"`
Name string `json:"name"`
Token string `json:"token"`
Address string `json:"address"`
Labels []string `json:"labels"`
Ephemeral bool `json:"ephemeral"`
}
func RegistrationJSON(assignmentID, capability, localProxyURL string, backend taskassignment.Backend) ([]byte, error) {
if assignmentID == "" || capability == "" {
return nil, errors.New("assignment ID and runner capability are required")
}
parsed, err := url.Parse(localProxyURL)
if err != nil || parsed.Scheme != "http" || parsed.Host == "" {
return nil, errors.New("local runner proxy URL must be an absolute http URL")
}
if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
return nil, fmt.Errorf("unsupported runner backend %q", backend)
}
registration := Registration{
Warning: "Generated for one preassigned task by gitea-dynamic-runner.",
UUID: assignmentID, Name: assignmentID, Token: capability,
Address: localProxyURL, Labels: []string{"self-hosted", string(backend)}, Ephemeral: true,
}
data, err := json.MarshalIndent(registration, "", " ")
if err != nil {
return nil, err
}
return append(data, '\n'), nil
}
@@ -0,0 +1,78 @@
package runnerbootstrap
import (
"encoding/json"
"testing"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
)
func testBootstrap(t *testing.T) Bootstrap {
t.Helper()
capabilities, err := runnerfacade.NewCapabilities([]byte("0123456789abcdef0123456789abcdef"))
if err != nil {
t.Fatal(err)
}
return Bootstrap{
Capabilities: capabilities,
FacadeURL: "https://runner-facade.gitea-actions.svc:8443",
FacadeSPIFFEID: "spiffe://ddupan.top/ns/gitea-actions/sa/gitea-dynamic-runner",
WorkloadAPIAddr: "unix:///run/spire/agent-sockets/spire-agent.sock",
}
}
func testAssignment() taskassignment.Assignment {
return taskassignment.Assignment{
ID: "gitea-task-42", Backend: taskassignment.BackendPod,
Identity: taskidentity.Identity{SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"},
}
}
func TestEnvironmentIsDeterministicAndAssignmentScoped(t *testing.T) {
bootstrap := testBootstrap(t)
first, err := bootstrap.Environment(testAssignment())
if err != nil {
t.Fatal(err)
}
second, err := bootstrap.Environment(testAssignment())
if err != nil {
t.Fatal(err)
}
if first[EnvCapability] == "" || first[EnvCapability] != second[EnvCapability] {
t.Fatalf("capabilities = %q, %q", first[EnvCapability], second[EnvCapability])
}
if first[EnvAssignmentID] != "gitea-task-42" || first[EnvSPIFFEID] != testAssignment().Identity.SPIFFEID {
t.Fatalf("environment = %#v", first)
}
if first[EnvBackend] != "pod" || first[EnvFacadeID] == "" {
t.Fatalf("environment = %#v", first)
}
if first["SPIFFE_ENDPOINT_SOCKET"] != "unix:///run/spire/agent-sockets/spire-agent.sock" {
t.Fatalf("environment = %#v", first)
}
}
func TestRegistrationMatchesOfficialRunnerSchema(t *testing.T) {
data, err := RegistrationJSON("gitea-task-42", "capability", "http://127.0.0.1:8080", taskassignment.BackendVM)
if err != nil {
t.Fatal(err)
}
var registration Registration
if err := json.Unmarshal(data, &registration); err != nil {
t.Fatal(err)
}
if registration.UUID != "gitea-task-42" || registration.Token != "capability" || registration.Address != "http://127.0.0.1:8080" || !registration.Ephemeral {
t.Fatalf("registration = %#v", registration)
}
if len(registration.Labels) != 2 || registration.Labels[0] != "self-hosted" || registration.Labels[1] != "vm" {
t.Fatalf("labels = %#v", registration.Labels)
}
}
func TestRegistrationRejectsNonLocalTLSAddress(t *testing.T) {
if _, err := RegistrationJSON("id", "capability", "https://facade.example", taskassignment.BackendPod); err == nil {
t.Fatal("expected local proxy URL validation error")
}
}
+153
View File
@@ -0,0 +1,153 @@
package runnerbootstrap
import (
"context"
"errors"
"fmt"
"net"
"net/http"
"os"
"os/exec"
"path/filepath"
"time"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)
type ExecutorConfig struct {
AssignmentID string
Capability string
Backend taskassignment.Backend
FacadeURL string
FacadeSPIFFEID string
WorkloadAPIAddr string
RunnerBinary string
ListenAddress string
WorkDir string
Stdout *os.File
Stderr *os.File
}
// RunExecutor runs the SPIFFE proxy and one unmodified official Runner process.
// The generated registration file exists only in the executor's temporary
// work directory and the runner exits after its preassigned task.
func RunExecutor(ctx context.Context, config ExecutorConfig) error {
if config.RunnerBinary == "" {
config.RunnerBinary = "gitea-runner"
}
if config.ListenAddress == "" {
config.ListenAddress = "127.0.0.1:0"
}
listener, err := net.Listen("tcp", config.ListenAddress)
if err != nil {
return fmt.Errorf("listen for local runner proxy: %w", err)
}
defer listener.Close()
address, ok := listener.Addr().(*net.TCPAddr)
if !ok || !address.IP.IsLoopback() {
return errors.New("runner proxy must listen on a loopback address")
}
proxy, err := NewProxy(ctx, config.FacadeURL, config.FacadeSPIFFEID, config.WorkloadAPIAddr)
if err != nil {
return err
}
defer proxy.Close()
workDir := config.WorkDir
removeWorkDir := false
if workDir == "" {
workDir, err = os.MkdirTemp("", "gitea-task-runner-")
if err != nil {
return fmt.Errorf("create runner work directory: %w", err)
}
removeWorkDir = true
}
if removeWorkDir {
defer os.RemoveAll(workDir)
}
registration, err := RegistrationJSON(
config.AssignmentID, config.Capability, "http://"+listener.Addr().String(), config.Backend,
)
if err != nil {
return err
}
if err := os.WriteFile(filepath.Join(workDir, ".runner"), registration, 0o600); err != nil {
return fmt.Errorf("write one-shot runner registration: %w", err)
}
server := &http.Server{Handler: proxy.Handler, ReadHeaderTimeout: 10 * time.Second}
serverErrors := make(chan error, 1)
go func() { serverErrors <- server.Serve(listener) }()
readyContext, cancelReady := context.WithTimeout(ctx, 2*time.Minute)
readyErr := waitForFacade(readyContext, "http://"+listener.Addr().String())
cancelReady()
if readyErr != nil {
shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
shutdownErr := server.Shutdown(shutdownContext)
cancel()
serverErr := <-serverErrors
if errors.Is(serverErr, http.ErrServerClosed) {
serverErr = nil
}
return errors.Join(readyErr, shutdownErr, serverErr)
}
command := exec.CommandContext(ctx, config.RunnerBinary, "daemon", "--once")
command.Dir = workDir
command.Stdout = config.Stdout
command.Stderr = config.Stderr
runnerErr := command.Run()
shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
shutdownErr := server.Shutdown(shutdownContext)
cancel()
serverErr := <-serverErrors
if errors.Is(serverErr, http.ErrServerClosed) {
serverErr = nil
}
return errors.Join(runnerErr, shutdownErr, serverErr)
}
func waitForFacade(ctx context.Context, endpoint string) error {
client := &http.Client{Timeout: 2 * time.Second}
ticker := time.NewTicker(250 * time.Millisecond)
defer ticker.Stop()
for {
request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
if err != nil {
return fmt.Errorf("create facade readiness request: %w", err)
}
response, requestErr := client.Do(request)
if requestErr == nil {
_ = response.Body.Close()
if response.StatusCode < http.StatusInternalServerError {
return nil
}
}
select {
case <-ctx.Done():
return fmt.Errorf("wait for runner facade: %w", ctx.Err())
case <-ticker.C:
}
}
}
// ExecutorConfigFromEnvironment reads the non-secret image configuration and
// the assignment-scoped values injected by the backend. The Workload API
// address follows SPIFFE_ENDPOINT_SOCKET through go-spiffe when not set here.
func ExecutorConfigFromEnvironment() (ExecutorConfig, error) {
backend := taskassignment.Backend(os.Getenv(EnvBackend))
if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
return ExecutorConfig{}, fmt.Errorf("invalid %s %q", EnvBackend, backend)
}
config := ExecutorConfig{
AssignmentID: os.Getenv(EnvAssignmentID), Capability: os.Getenv(EnvCapability),
Backend: backend, FacadeURL: os.Getenv(EnvFacadeURL), FacadeSPIFFEID: os.Getenv(EnvFacadeID),
RunnerBinary: os.Getenv("GITEA_RUNNER_BINARY"), ListenAddress: "127.0.0.1:0",
Stdout: os.Stdout, Stderr: os.Stderr,
}
if config.AssignmentID == "" || config.Capability == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
return ExecutorConfig{}, errors.New("complete runner assignment and facade environment is required")
}
return config, nil
}
+44
View File
@@ -0,0 +1,44 @@
package runnerbootstrap
import (
"context"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"time"
)
func TestWaitForFacadeRetriesTransientGatewayFailure(t *testing.T) {
var requests atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
if requests.Add(1) < 3 {
writer.WriteHeader(http.StatusBadGateway)
return
}
writer.WriteHeader(http.StatusNotFound)
}))
defer server.Close()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := waitForFacade(ctx, server.URL); err != nil {
t.Fatal(err)
}
if requests.Load() != 3 {
t.Fatalf("requests = %d", requests.Load())
}
}
func TestWaitForFacadeStopsWithContext(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
writer.WriteHeader(http.StatusServiceUnavailable)
}))
defer server.Close()
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
if err := waitForFacade(ctx, server.URL); err == nil {
t.Fatal("expected readiness timeout")
}
}
+56
View File
@@ -0,0 +1,56 @@
package runnerbootstrap
import (
"context"
"fmt"
"net/http"
"net/http/httputil"
"net/url"
"github.com/spiffe/go-spiffe/v2/spiffeid"
"github.com/spiffe/go-spiffe/v2/spiffetls/tlsconfig"
"github.com/spiffe/go-spiffe/v2/workloadapi"
)
type Proxy struct {
Handler http.Handler
source *workloadapi.X509Source
}
// NewProxy obtains rotating X509-SVIDs from the Workload API and authorizes
// one exact controller identity. The official Runner talks plain HTTP only to
// this executor-local handler.
func NewProxy(ctx context.Context, facadeURL, facadeSPIFFEID, workloadAPIAddr string) (*Proxy, error) {
target, err := url.Parse(facadeURL)
if err != nil || target.Scheme != "https" || target.Host == "" {
return nil, fmt.Errorf("runner facade URL must be an absolute https URL")
}
serverID, err := spiffeid.FromString(facadeSPIFFEID)
if err != nil {
return nil, fmt.Errorf("parse runner facade SPIFFE ID: %w", err)
}
options := []workloadapi.X509SourceOption{}
if workloadAPIAddr != "" {
options = append(options, workloadapi.WithClientOptions(workloadapi.WithAddr(workloadAPIAddr)))
}
source, err := workloadapi.NewX509Source(ctx, options...)
if err != nil {
return nil, fmt.Errorf("open SPIFFE Workload API X509 source: %w", err)
}
transport := http.DefaultTransport.(*http.Transport).Clone()
transport.TLSClientConfig = tlsconfig.MTLSClientConfig(source, source, tlsconfig.AuthorizeID(serverID))
return &Proxy{Handler: NewProxyHandler(target, transport), source: source}, nil
}
func NewProxyHandler(target *url.URL, transport http.RoundTripper) http.Handler {
proxy := httputil.NewSingleHostReverseProxy(target)
proxy.Transport = transport
return proxy
}
func (p *Proxy) Close() error {
if p == nil || p.source == nil {
return nil
}
return p.source.Close()
}
+36
View File
@@ -0,0 +1,36 @@
package runnerfacade
import (
"crypto/hmac"
"crypto/sha256"
"encoding/base64"
"errors"
)
// Capabilities are deterministic per assignment so controller restarts do not
// require a per-task token database.
type Capabilities struct{ key []byte }
func NewCapabilities(key []byte) (Capabilities, error) {
if len(key) < 32 {
return Capabilities{}, errors.New("runner facade capability key must be at least 32 bytes")
}
return Capabilities{key: append([]byte(nil), key...)}, nil
}
func (c Capabilities) Issue(assignmentID string) string {
if len(c.key) < 32 || assignmentID == "" {
return ""
}
mac := hmac.New(sha256.New, c.key)
_, _ = mac.Write([]byte("gitea-runner-assignment\x00" + assignmentID))
return base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
}
func (c Capabilities) Verify(assignmentID, token string) bool {
want := c.Issue(assignmentID)
if want == "" || token == "" {
return false
}
return hmac.Equal([]byte(want), []byte(token))
}
+145
View File
@@ -0,0 +1,145 @@
// Package runnerfacade presents pre-assigned tasks to unmodified Gitea Runner binaries.
package runnerfacade
import (
"context"
"errors"
"net/http"
"connectrpc.com/connect"
"gitea.dev/actionslib/pkg/protocol"
runnerv1 "gitea.dev/actionslib/runner/v1"
"gitea.dev/actionslib/runner/v1/runnerv1connect"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)
type identityKey struct{}
func WithSPIFFEID(ctx context.Context, id string) context.Context {
return context.WithValue(ctx, identityKey{}, id)
}
// SPIFFEMiddleware extracts the authenticated workload identity from the mTLS
// peer certificate. TLS verification itself is configured by the server with
// go-spiffe; this layer only passes the verified ID into Connect handlers.
func SPIFFEMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
if request.TLS == nil || len(request.TLS.PeerCertificates) == 0 {
http.Error(response, "client SPIFFE identity required", http.StatusUnauthorized)
return
}
var spiffeID string
for _, uri := range request.TLS.PeerCertificates[0].URIs {
if uri.Scheme == "spiffe" {
if spiffeID != "" {
http.Error(response, "multiple client SPIFFE identities", http.StatusUnauthorized)
return
}
spiffeID = uri.String()
}
}
if spiffeID == "" {
http.Error(response, "client SPIFFE identity required", http.StatusUnauthorized)
return
}
next.ServeHTTP(response, request.WithContext(WithSPIFFEID(request.Context(), spiffeID)))
})
}
type Upstream interface {
UpdateTask(context.Context, *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error)
UpdateLog(context.Context, *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error)
}
type Facade struct {
runnerv1connect.UnimplementedRunnerServiceHandler
Registry *Registry
Capabilities Capabilities
Upstream Upstream
OnTerminal func(context.Context, taskassignment.Assignment) error
}
func (f *Facade) Handler() (string, http.Handler) {
return runnerv1connect.NewRunnerServiceHandler(f)
}
func (f *Facade) Register(context.Context, *connect.Request[runnerv1.RegisterRequest]) (*connect.Response[runnerv1.RegisterResponse], error) {
return nil, connect.NewError(connect.CodePermissionDenied, errors.New("executor registration is disabled"))
}
func (f *Facade) Declare(ctx context.Context, request *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) {
assignmentID, _, err := f.authenticate(ctx, request)
if err != nil {
return nil, err
}
return connect.NewResponse(&runnerv1.DeclareResponse{Runner: &runnerv1.Runner{
Uuid: assignmentID, Name: assignmentID, Status: runnerv1.RunnerStatus_RUNNER_STATUS_IDLE,
Version: request.Msg.GetVersion(), Labels: append([]string(nil), request.Msg.GetLabels()...), Ephemeral: true,
}}), nil
}
func (f *Facade) FetchTask(ctx context.Context, request *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
assignmentID, spiffeID, err := f.authenticate(ctx, request)
if err != nil {
return nil, err
}
assignment, err := f.Registry.Claim(assignmentID, spiffeID)
if err != nil {
return nil, connect.NewError(connect.CodeFailedPrecondition, err)
}
return connect.NewResponse(&runnerv1.FetchTaskResponse{Task: assignment.Task}), nil
}
func (f *Facade) UpdateTask(ctx context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
assignment, err := f.authorizeClaimed(ctx, request)
if err != nil {
return nil, err
}
if request.Msg.GetState().GetId() != assignment.Task.GetId() {
return nil, connect.NewError(connect.CodePermissionDenied, errors.New("task update does not match assignment"))
}
response, err := f.Upstream.UpdateTask(ctx, connect.NewRequest(request.Msg))
if err == nil && request.Msg.GetState().GetResult() != runnerv1.Result_RESULT_UNSPECIFIED && f.OnTerminal != nil {
if terminalErr := f.OnTerminal(ctx, assignment); terminalErr != nil {
return nil, connect.NewError(connect.CodeUnavailable, terminalErr)
}
}
return response, err
}
func (f *Facade) UpdateLog(ctx context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
assignment, err := f.authorizeClaimed(ctx, request)
if err != nil {
return nil, err
}
if request.Msg.GetTaskId() != assignment.Task.GetId() {
return nil, connect.NewError(connect.CodePermissionDenied, errors.New("log update does not match assignment"))
}
return f.Upstream.UpdateLog(ctx, connect.NewRequest(request.Msg))
}
func (f *Facade) authorizeClaimed(ctx context.Context, request connect.AnyRequest) (taskassignment.Assignment, error) {
assignmentID, spiffeID, err := f.authenticate(ctx, request)
if err != nil {
return taskassignment.Assignment{}, err
}
assignment, err := f.Registry.Resolve(assignmentID, spiffeID)
if err != nil {
return taskassignment.Assignment{}, connect.NewError(connect.CodePermissionDenied, err)
}
return assignment, nil
}
func (f *Facade) authenticate(ctx context.Context, request connect.AnyRequest) (string, string, error) {
if f.Registry == nil || f.Upstream == nil {
return "", "", connect.NewError(connect.CodeInternal, errors.New("runner facade is not configured"))
}
assignmentID := request.Header().Get(protocol.UUIDHeader)
token := request.Header().Get(protocol.TokenHeader)
spiffeID, _ := ctx.Value(identityKey{}).(string)
if assignmentID == "" || spiffeID == "" || !f.Capabilities.Verify(assignmentID, token) {
return "", "", connect.NewError(connect.CodeUnauthenticated, errors.New("invalid executor identity or capability"))
}
return assignmentID, spiffeID, nil
}
+239
View File
@@ -0,0 +1,239 @@
package runnerfacade
import (
"context"
"crypto/tls"
"crypto/x509"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"connectrpc.com/connect"
"gitea.dev/actionslib/pkg/protocol"
runnerv1 "gitea.dev/actionslib/runner/v1"
"gitea.dev/actionslib/runner/v1/runnerv1connect"
"google.golang.org/protobuf/types/known/structpb"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)
type fakeUpstream struct {
taskUpdates int
logUpdates int
}
func (u *fakeUpstream) UpdateTask(_ context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
u.taskUpdates++
return connect.NewResponse(&runnerv1.UpdateTaskResponse{State: request.Msg.State}), nil
}
func (u *fakeUpstream) UpdateLog(_ context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
u.logUpdates++
return connect.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: request.Msg.Index + int64(len(request.Msg.Rows))}), nil
}
func facadeAssignment(t *testing.T) taskassignment.Assignment {
t.Helper()
fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
if err != nil {
t.Fatal(err)
}
assignment, err := taskassignment.New(&runnerv1.Task{
Id: 42, Context: fields,
WorkflowPayload: []byte("jobs:\n publish:\n runs-on: [self-hosted, pod]\n steps: []\n"),
}, "ddupan.top")
if err != nil {
t.Fatal(err)
}
return assignment
}
func testFacade(t *testing.T) (*Facade, taskassignment.Assignment, string) {
t.Helper()
capabilities, err := NewCapabilities([]byte("0123456789abcdef0123456789abcdef"))
if err != nil {
t.Fatal(err)
}
assignment := facadeAssignment(t)
registry := NewRegistry()
if _, err := registry.Offer(assignment); err != nil {
t.Fatal(err)
}
return &Facade{Registry: registry, Capabilities: capabilities, Upstream: &fakeUpstream{}}, assignment, capabilities.Issue(assignment.ID)
}
func authenticatedRequest[T any](message *T, assignmentID, token string) *connect.Request[T] {
request := connect.NewRequest(message)
request.Header().Set(protocol.UUIDHeader, assignmentID)
request.Header().Set(protocol.TokenHeader, token)
return request
}
func TestFacadeReturnsOnlyPreassignedTaskAndSignalsClaim(t *testing.T) {
facade, assignment, token := testFacade(t)
ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
response, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token))
if err != nil {
t.Fatal(err)
}
if response.Msg.GetTask().GetId() != assignment.Task.GetId() {
t.Fatalf("task = %#v", response.Msg.GetTask())
}
if err := facade.Registry.WaitClaimed(context.Background(), assignment.ID); err != nil {
t.Fatal(err)
}
if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); connect.CodeOf(err) != connect.CodeFailedPrecondition {
t.Fatalf("second FetchTask error = %v", err)
}
}
func TestFacadeDeclareReturnsOfficialRunnerMetadata(t *testing.T) {
facade, assignment, token := testFacade(t)
ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
response, err := facade.Declare(ctx, authenticatedRequest(&runnerv1.DeclareRequest{
Version: "v3.5.0", Labels: []string{"self-hosted", "pod"},
}, assignment.ID, token))
if err != nil {
t.Fatal(err)
}
runner := response.Msg.GetRunner()
if runner.GetUuid() != assignment.ID || runner.GetName() != assignment.ID || runner.GetVersion() != "v3.5.0" || !runner.GetEphemeral() {
t.Fatalf("runner = %#v", runner)
}
if len(runner.GetLabels()) != 2 || runner.GetLabels()[0] != "self-hosted" || runner.GetLabels()[1] != "pod" {
t.Fatalf("labels = %#v", runner.GetLabels())
}
}
func TestAPIHandlerMatchesOfficialRunnerBasePath(t *testing.T) {
facade, assignment, token := testFacade(t)
identityURL, err := url.Parse(assignment.Identity.SPIFFEID)
if err != nil {
t.Fatal(err)
}
handler := APIHandler(facade)
server := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
request.TLS = &tls.ConnectionState{PeerCertificates: []*x509.Certificate{{URIs: []*url.URL{identityURL}}}}
handler.ServeHTTP(response, request)
}))
defer server.Close()
client := runnerv1connect.NewRunnerServiceClient(server.Client(), server.URL+APIBasePath)
request := authenticatedRequest(&runnerv1.DeclareRequest{
Version: "v3.5.0", Labels: []string{"self-hosted", "pod"},
}, assignment.ID, token)
response, err := client.Declare(context.Background(), request)
if err != nil {
t.Fatal(err)
}
if response.Msg.GetRunner().GetUuid() != assignment.ID {
t.Fatalf("runner = %#v", response.Msg.GetRunner())
}
}
func TestHandlerProxiesRepositoryTrafficToGitea(t *testing.T) {
var upstream *httptest.Server
upstream = httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
upstreamHost := strings.TrimPrefix(upstream.URL, "http://")
if request.URL.Path != "/owner/repo/info/refs" || request.Header.Get("Authorization") != "Basic checkout-token" || request.Host != upstreamHost {
t.Errorf("request path=%q authorization=%q host=%q", request.URL.Path, request.Header.Get("Authorization"), request.Host)
response.WriteHeader(http.StatusBadRequest)
return
}
response.WriteHeader(http.StatusOK)
}))
defer upstream.Close()
facade, _, _ := testFacade(t)
handler, err := Handler(facade, upstream.URL)
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "http://facade/owner/repo/info/refs", nil)
request.Header.Set("Authorization", "Basic checkout-token")
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status = %d", response.Code)
}
}
func TestFacadeRejectsWrongIdentityOrCapability(t *testing.T) {
facade, assignment, token := testFacade(t)
wrongIdentity := WithSPIFFEID(context.Background(), "spiffe://ddupan.top/ci/owner/repo/other")
if _, err := facade.FetchTask(wrongIdentity, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); connect.CodeOf(err) != connect.CodeFailedPrecondition {
t.Fatalf("wrong identity error = %v", err)
}
ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, "wrong")); connect.CodeOf(err) != connect.CodeUnauthenticated {
t.Fatalf("wrong capability error = %v", err)
}
}
func TestFacadeForwardsOnlyMatchingTaskAndLogUpdates(t *testing.T) {
facade, assignment, token := testFacade(t)
ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); err != nil {
t.Fatal(err)
}
if _, err := facade.UpdateTask(ctx, authenticatedRequest(&runnerv1.UpdateTaskRequest{State: &runnerv1.TaskState{Id: 42}}, assignment.ID, token)); err != nil {
t.Fatal(err)
}
if _, err := facade.UpdateLog(ctx, authenticatedRequest(&runnerv1.UpdateLogRequest{TaskId: 42}, assignment.ID, token)); err != nil {
t.Fatal(err)
}
upstream := facade.Upstream.(*fakeUpstream)
if upstream.taskUpdates != 1 || upstream.logUpdates != 1 {
t.Fatalf("task updates=%d log updates=%d", upstream.taskUpdates, upstream.logUpdates)
}
if _, err := facade.UpdateLog(ctx, authenticatedRequest(&runnerv1.UpdateLogRequest{TaskId: 99}, assignment.ID, token)); connect.CodeOf(err) != connect.CodePermissionDenied {
t.Fatalf("mismatched log error = %v", err)
}
}
func TestFacadeSignalsTerminalTaskAfterUpstreamAcceptsIt(t *testing.T) {
facade, assignment, token := testFacade(t)
ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID)
if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); err != nil {
t.Fatal(err)
}
completed := 0
facade.OnTerminal = func(_ context.Context, got taskassignment.Assignment) error {
if got.ID != assignment.ID {
t.Fatalf("terminal assignment = %s", got.ID)
}
completed++
return nil
}
if _, err := facade.UpdateTask(ctx, authenticatedRequest(&runnerv1.UpdateTaskRequest{
State: &runnerv1.TaskState{Id: 42, Result: runnerv1.Result_RESULT_SUCCESS},
}, assignment.ID, token)); err != nil {
t.Fatal(err)
}
if completed != 1 {
t.Fatalf("terminal notifications = %d", completed)
}
}
func TestCapabilitiesAreDeterministicAndAssignmentScoped(t *testing.T) {
capabilities, err := NewCapabilities([]byte("0123456789abcdef0123456789abcdef"))
if err != nil {
t.Fatal(err)
}
token := capabilities.Issue("gitea-task-42")
if !capabilities.Verify("gitea-task-42", token) || capabilities.Verify("gitea-task-43", token) {
t.Fatal("capability scope is invalid")
}
}
func TestRegistryRecoversClaimedAssignment(t *testing.T) {
registry := NewRegistry()
assignment := facadeAssignment(t)
if err := registry.RecoverClaimed(assignment); err != nil {
t.Fatal(err)
}
resolved, err := registry.Resolve(assignment.ID, assignment.Identity.SPIFFEID)
if err != nil || resolved.Task.GetId() != assignment.Task.GetId() {
t.Fatalf("resolved=%#v err=%v", resolved, err)
}
}
+110
View File
@@ -0,0 +1,110 @@
package runnerfacade
import (
"context"
"errors"
"fmt"
"sync"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)
type claim struct {
assignment taskassignment.Assignment
claimed bool
ready chan struct{}
}
// Registry holds pending task payloads and an active authorization cache.
// Pending entries are rebuilt by JetStream redelivery; active authorization is
// reconstructable from backend metadata and is not an independent state store.
type Registry struct {
mu sync.Mutex
claims map[string]*claim
}
func NewRegistry() *Registry { return &Registry{claims: make(map[string]*claim)} }
func (r *Registry) Offer(assignment taskassignment.Assignment) (<-chan struct{}, error) {
r.mu.Lock()
defer r.mu.Unlock()
if existing := r.claims[assignment.ID]; existing != nil {
if existing.assignment.Identity != assignment.Identity || existing.assignment.Task.GetId() != assignment.Task.GetId() {
return nil, fmt.Errorf("assignment %s was offered with different task data", assignment.ID)
}
return existing.ready, nil
}
entry := &claim{assignment: assignment, ready: make(chan struct{})}
r.claims[assignment.ID] = entry
return entry.ready, nil
}
func (r *Registry) Claim(assignmentID, spiffeID string) (taskassignment.Assignment, error) {
r.mu.Lock()
defer r.mu.Unlock()
entry := r.claims[assignmentID]
if entry == nil {
return taskassignment.Assignment{}, errors.New("assignment is not pending")
}
if entry.assignment.Identity.SPIFFEID != spiffeID {
return taskassignment.Assignment{}, errors.New("executor SPIFFE ID does not match assignment")
}
if entry.claimed {
return taskassignment.Assignment{}, errors.New("assignment was already claimed")
}
entry.claimed = true
close(entry.ready)
return entry.assignment, nil
}
func (r *Registry) Resolve(assignmentID, spiffeID string) (taskassignment.Assignment, error) {
r.mu.Lock()
defer r.mu.Unlock()
entry := r.claims[assignmentID]
if entry == nil || !entry.claimed {
return taskassignment.Assignment{}, errors.New("assignment is not claimed")
}
if entry.assignment.Identity.SPIFFEID != spiffeID {
return taskassignment.Assignment{}, errors.New("executor SPIFFE ID does not match assignment")
}
return entry.assignment, nil
}
func (r *Registry) WaitClaimed(ctx context.Context, assignmentID string) error {
r.mu.Lock()
entry := r.claims[assignmentID]
r.mu.Unlock()
if entry == nil {
return errors.New("assignment is not pending")
}
select {
case <-entry.ready:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (r *Registry) Remove(assignmentID string) {
r.mu.Lock()
defer r.mu.Unlock()
delete(r.claims, assignmentID)
}
// RecoverClaimed restores authorization for an executor that already claimed
// its task before the controller restarted.
func (r *Registry) RecoverClaimed(assignment taskassignment.Assignment) error {
ready, err := r.Offer(assignment)
if err != nil {
return err
}
r.mu.Lock()
entry := r.claims[assignment.ID]
if !entry.claimed {
entry.claimed = true
close(entry.ready)
}
r.mu.Unlock()
<-ready
return nil
}
+113
View File
@@ -0,0 +1,113 @@
package runnerfacade
import (
"context"
"crypto/tls"
"errors"
"fmt"
"net"
"net/http"
"net/http/httputil"
"net/url"
"time"
"github.com/spiffe/go-spiffe/v2/spiffeid"
"github.com/spiffe/go-spiffe/v2/spiffetls/tlsconfig"
"github.com/spiffe/go-spiffe/v2/workloadapi"
)
const APIBasePath = "/api/actions"
// APIHandler exposes the facade at the base path used by the official Runner.
// SPIFFE middleware runs after the TLS listener has authenticated the peer.
func APIHandler(facade *Facade) http.Handler {
path, handler := facade.Handler()
mux := http.NewServeMux()
mux.Handle(APIBasePath+path, http.StripPrefix(APIBasePath, SPIFFEMiddleware(handler)))
return mux
}
// Handler keeps RunnerService calls inside the authenticated facade while
// forwarding repository and artifact HTTP traffic to the real Gitea server.
// Official Runner derives checkout URLs from its registration instance URL,
// which intentionally points at the executor-local SPIFFE proxy.
func Handler(facade *Facade, upstreamURL string) (http.Handler, error) {
target, err := url.Parse(upstreamURL)
if err != nil || (target.Scheme != "http" && target.Scheme != "https") || target.Host == "" {
return nil, errors.New("Gitea upstream must be an absolute HTTP URL")
}
path, service := facade.Handler()
mux := http.NewServeMux()
mux.Handle(APIBasePath+path, http.StripPrefix(APIBasePath, SPIFFEMiddleware(service)))
proxy := httputil.NewSingleHostReverseProxy(target)
director := proxy.Director
proxy.Director = func(request *http.Request) {
director(request)
request.Host = target.Host
}
mux.Handle("/", proxy)
return mux, nil
}
type Server struct {
Facade *Facade
ListenAddress string
TrustDomain string
WorkloadAPIAddr string
UpstreamURL string
}
// Run serves the RunnerService facade with workload-to-workload mTLS. Any
// identity in the local trust domain may complete TLS; the facade then requires
// the exact logical task identity stored in its assignment registry.
func (s Server) Run(ctx context.Context) error {
if s.Facade == nil || s.ListenAddress == "" || s.TrustDomain == "" || s.UpstreamURL == "" {
return errors.New("runner facade, listen address, trust domain, and Gitea upstream are required")
}
handler, err := Handler(s.Facade, s.UpstreamURL)
if err != nil {
return err
}
trustDomain, err := spiffeid.TrustDomainFromString(s.TrustDomain)
if err != nil {
return fmt.Errorf("parse facade trust domain: %w", err)
}
options := []workloadapi.X509SourceOption{}
if s.WorkloadAPIAddr != "" {
options = append(options, workloadapi.WithClientOptions(workloadapi.WithAddr(s.WorkloadAPIAddr)))
}
source, err := workloadapi.NewX509Source(ctx, options...)
if err != nil {
return fmt.Errorf("open facade SPIFFE Workload API X509 source: %w", err)
}
defer source.Close()
listener, err := net.Listen("tcp", s.ListenAddress)
if err != nil {
return fmt.Errorf("listen for runner facade: %w", err)
}
defer listener.Close()
tlsListener := tls.NewListener(listener, tlsconfig.MTLSServerConfig(
source, source, tlsconfig.AuthorizeMemberOf(trustDomain),
))
httpServer := &http.Server{Handler: handler, ReadHeaderTimeout: 10 * time.Second}
serverErrors := make(chan error, 1)
go func() { serverErrors <- httpServer.Serve(tlsListener) }()
select {
case <-ctx.Done():
shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
shutdownErr := httpServer.Shutdown(shutdownContext)
cancel()
serverErr := <-serverErrors
if errors.Is(serverErr, http.ErrServerClosed) {
serverErr = nil
}
return errors.Join(shutdownErr, serverErr)
case err := <-serverErrors:
if errors.Is(err, http.ErrServerClosed) {
return nil
}
return fmt.Errorf("serve runner facade: %w", err)
}
}
+152
View File
@@ -0,0 +1,152 @@
// Package taskassignment defines the durable handoff between the scheduler and workers.
package taskassignment
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"slices"
"strconv"
"gitea.dev/actionslib/pkg/model"
runnerv1 "gitea.dev/actionslib/runner/v1"
"google.golang.org/protobuf/proto"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
)
const wireVersion = 1
type Backend string
const (
BackendPod Backend = "pod"
BackendVM Backend = "vm"
)
// Assignment is the only document persisted in the handoff queue.
type Assignment struct {
ID string
Backend Backend
Task *runnerv1.Task
Identity taskidentity.Identity
}
// FromMetadata reconstructs the minimal assignment needed to authorize an
// already-running executor after a controller restart. Backend metadata was
// originally derived from the trusted Gitea task and is validated again here.
func FromMetadata(labels, annotations map[string]string, trustDomain string) (Assignment, error) {
taskID, err := strconv.ParseInt(labels["ci.ddupan.top/task-id"], 10, 64)
if err != nil || taskID < 1 {
return Assignment{}, errors.New("backend metadata has invalid task ID")
}
backend := Backend(labels["ci.ddupan.top/backend"])
if backend != BackendPod && backend != BackendVM {
return Assignment{}, errors.New("backend metadata has invalid backend")
}
id := labels["ci.ddupan.top/assignment-id"]
if id != fmt.Sprintf("gitea-task-%d", taskID) {
return Assignment{}, errors.New("backend metadata assignment ID does not match task ID")
}
identity, err := taskidentity.FromMetadata(
annotations["ci.ddupan.top/repository"], annotations["ci.ddupan.top/job-key"],
annotations["ci.ddupan.top/spiffe-id"], trustDomain,
)
if err != nil {
return Assignment{}, err
}
return Assignment{ID: id, Backend: backend, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil
}
type envelope struct {
Version int `json:"version"`
ID string `json:"id"`
Backend Backend `json:"backend"`
Task []byte `json:"task"`
Identity taskidentity.Identity `json:"identity"`
}
// New derives all trusted assignment fields from the task fetched from Gitea.
func New(task *runnerv1.Task, trustDomain string) (Assignment, error) {
if task == nil || task.GetId() <= 0 {
return Assignment{}, errors.New("positive Gitea task ID is required")
}
identity, err := taskidentity.FromTask(task, trustDomain)
if err != nil {
return Assignment{}, err
}
backend, err := backendFromTask(task)
if err != nil {
return Assignment{}, err
}
return Assignment{
ID: fmt.Sprintf("gitea-task-%d", task.GetId()),
Backend: backend,
Task: task,
Identity: identity,
}, nil
}
func backendFromTask(task *runnerv1.Task) (Backend, error) {
workflow, err := model.ReadWorkflow(bytes.NewReader(task.GetWorkflowPayload()))
if err != nil {
return "", fmt.Errorf("parse task workflow for backend: %w", err)
}
jobIDs := workflow.GetJobIDs()
if len(jobIDs) != 1 || workflow.GetJob(jobIDs[0]) == nil {
return "", fmt.Errorf("task workflow must contain exactly one non-empty job")
}
labels := workflow.GetJob(jobIDs[0]).RunsOnLabels()
if !slices.Contains(labels, "self-hosted") {
return "", fmt.Errorf("task runs-on labels must include self-hosted: %v", labels)
}
hasPod := slices.Contains(labels, string(BackendPod))
hasVM := slices.Contains(labels, string(BackendVM))
if hasPod == hasVM {
return "", fmt.Errorf("task runs-on labels must select exactly one of pod or vm: %v", labels)
}
if hasPod {
return BackendPod, nil
}
return BackendVM, nil
}
// Marshal encodes a versioned assignment. Protobuf preserves the exact Gitea task.
func Marshal(assignment Assignment) ([]byte, error) {
if assignment.Task == nil {
return nil, errors.New("assignment task is required")
}
task, err := proto.Marshal(assignment.Task)
if err != nil {
return nil, fmt.Errorf("marshal Gitea task: %w", err)
}
return json.Marshal(envelope{
Version: wireVersion,
ID: assignment.ID, Backend: assignment.Backend,
Task: task, Identity: assignment.Identity,
})
}
// Unmarshal re-derives trusted fields instead of trusting duplicated queue metadata.
func Unmarshal(data []byte, trustDomain string) (Assignment, error) {
var wire envelope
if err := json.Unmarshal(data, &wire); err != nil {
return Assignment{}, fmt.Errorf("decode assignment: %w", err)
}
if wire.Version != wireVersion {
return Assignment{}, fmt.Errorf("unsupported assignment version %d", wire.Version)
}
task := new(runnerv1.Task)
if err := proto.Unmarshal(wire.Task, task); err != nil {
return Assignment{}, fmt.Errorf("unmarshal Gitea task: %w", err)
}
canonical, err := New(task, trustDomain)
if err != nil {
return Assignment{}, err
}
if wire.ID != canonical.ID || wire.Backend != canonical.Backend || wire.Identity != canonical.Identity {
return Assignment{}, errors.New("assignment metadata does not match its Gitea task")
}
return canonical, nil
}
@@ -0,0 +1,90 @@
package taskassignment
import (
"bytes"
"testing"
runnerv1 "gitea.dev/actionslib/runner/v1"
"google.golang.org/protobuf/types/known/structpb"
)
func task(t *testing.T, labels string) *runnerv1.Task {
t.Helper()
context, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
if err != nil {
t.Fatal(err)
}
return &runnerv1.Task{
Id: 42,
Context: context,
WorkflowPayload: []byte("jobs:\n publish:\n runs-on: " + labels + "\n steps: []\n"),
}
}
func TestNewSelectsBackendFromRunsOn(t *testing.T) {
for _, test := range []struct {
labels string
backend Backend
}{
{"[self-hosted, pod]", BackendPod},
{"[self-hosted, vm]", BackendVM},
} {
assignment, err := New(task(t, test.labels), "ddupan.top")
if err != nil {
t.Fatal(err)
}
if assignment.Backend != test.backend || assignment.ID != "gitea-task-42" {
t.Fatalf("assignment = %#v", assignment)
}
}
}
func TestNewRejectsAmbiguousBackend(t *testing.T) {
for _, labels := range []string{
"[self-hosted]",
"[self-hosted, pod, vm]",
"[pod]",
} {
if _, err := New(task(t, labels), "ddupan.top"); err == nil {
t.Fatalf("expected labels %s to fail", labels)
}
}
}
func TestAssignmentWireRoundTripAndValidation(t *testing.T) {
want, err := New(task(t, "[self-hosted, pod]"), "ddupan.top")
if err != nil {
t.Fatal(err)
}
data, err := Marshal(want)
if err != nil {
t.Fatal(err)
}
got, err := Unmarshal(data, "ddupan.top")
if err != nil {
t.Fatal(err)
}
if got.ID != want.ID || got.Backend != want.Backend || got.Identity != want.Identity || !bytes.Equal(got.Task.WorkflowPayload, want.Task.WorkflowPayload) {
t.Fatalf("round trip = %#v, want %#v", got, want)
}
tampered := bytes.Replace(data, []byte(`"backend":"pod"`), []byte(`"backend":"vm"`), 1)
if _, err := Unmarshal(tampered, "ddupan.top"); err == nil {
t.Fatal("expected tampered backend to fail")
}
}
func TestFromMetadataRecoversMinimalAssignment(t *testing.T) {
assignment, err := FromMetadata(map[string]string{
"ci.ddupan.top/assignment-id": "gitea-task-42",
"ci.ddupan.top/task-id": "42",
"ci.ddupan.top/backend": "vm",
}, map[string]string{
"ci.ddupan.top/repository": "owner/repo",
"ci.ddupan.top/job-key": "publish",
"ci.ddupan.top/spiffe-id": "spiffe://ddupan.top/ci/owner/repo/publish",
}, "ddupan.top")
if err != nil || assignment.Task.GetId() != 42 || assignment.Backend != BackendVM {
t.Fatalf("assignment=%#v err=%v", assignment, err)
}
}
+108
View File
@@ -0,0 +1,108 @@
// Package taskidentity derives workload identities from tasks assigned by Gitea.
package taskidentity
import (
"bytes"
"crypto/sha256"
"errors"
"fmt"
"regexp"
"strings"
"gitea.dev/actionslib/pkg/model"
runnerv1 "gitea.dev/actionslib/runner/v1"
)
var safeSegment = regexp.MustCompile(`^[A-Za-z0-9._-]+$`)
var safeTaskKey = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_-]*$`)
// Identity is the trusted identity context extracted from a fetched task.
type Identity struct {
Repository string
Task string
SPIFFEID string
}
// FromMetadata validates identity fields recovered from backend-owned state.
func FromMetadata(repository, task, spiffeID, trustDomain string) (Identity, error) {
parts := strings.Split(repository, "/")
if len(parts) != 2 || parts[0] == "" || parts[1] == "" || !safeTaskKey.MatchString(task) {
return Identity{}, errors.New("invalid recovered repository or task identity")
}
expected := "spiffe://" + trustDomain + "/ci/" + strings.Join([]string{
sanitize(parts[0]), sanitize(parts[1]), task,
}, "/")
if spiffeID != expected {
return Identity{}, fmt.Errorf("recovered SPIFFE ID %q does not match %q", spiffeID, expected)
}
return Identity{Repository: repository, Task: task, SPIFFEID: spiffeID}, nil
}
// FromTask derives the repository/task SPIFFE ID from Gitea's trusted task
// context. Workflow input never supplies or overrides the resulting ID.
func FromTask(task *runnerv1.Task, trustDomain string) (Identity, error) {
if task == nil || task.Context == nil {
return Identity{}, errors.New("task context is required")
}
repository := strings.TrimSpace(task.Context.GetFields()["repository"].GetStringValue())
taskName, err := workflowTaskKey(task.WorkflowPayload)
if err != nil {
return Identity{}, err
}
parts := strings.Split(repository, "/")
if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
return Identity{}, fmt.Errorf("invalid task repository %q", repository)
}
trustDomain = strings.TrimSpace(trustDomain)
if trustDomain == "" || strings.ContainsAny(trustDomain, "/ ") {
return Identity{}, fmt.Errorf("invalid trust domain %q", trustDomain)
}
path := strings.Join([]string{
sanitize(parts[0]),
sanitize(parts[1]),
taskName,
}, "/")
return Identity{
Repository: repository,
Task: taskName,
SPIFFEID: "spiffe://" + trustDomain + "/ci/" + path,
}, nil
}
func workflowTaskKey(payload []byte) (string, error) {
workflow, err := model.ReadWorkflow(bytes.NewReader(payload))
if err != nil {
return "", fmt.Errorf("parse task workflow: %w", err)
}
jobIDs := workflow.GetJobIDs()
if len(jobIDs) != 1 {
return "", fmt.Errorf("task workflow must contain exactly one job, got %d", len(jobIDs))
}
if !safeTaskKey.MatchString(jobIDs[0]) {
return "", fmt.Errorf("task job key %q must match %s", jobIDs[0], safeTaskKey)
}
return jobIDs[0], nil
}
// BackoffTaskSegment deterministically converts a legacy display name into a
// collision-resistant path segment. Canonical task identities do not use it.
func BackoffTaskSegment(value string) string {
return sanitize(value)
}
func sanitize(value string) string {
if safeSegment.MatchString(value) {
return value
}
slug := strings.Trim(regexp.MustCompile(`[^A-Za-z0-9._-]+`).ReplaceAllString(value, "-"), "-._")
if len(slug) > 48 {
slug = strings.TrimRight(slug[:48], "-._")
}
if slug == "" {
slug = "segment"
}
digest := fmt.Sprintf("%x", sha256.Sum256([]byte(value)))[:12]
return slug + "-" + digest
}
+77
View File
@@ -0,0 +1,77 @@
package taskidentity
import (
"testing"
runnerv1 "gitea.dev/actionslib/runner/v1"
"google.golang.org/protobuf/types/known/structpb"
)
func TestFromTaskUsesFetchedContext(t *testing.T) {
ctx, err := structpb.NewStruct(map[string]any{
"repository": "panxiao81/gitea-dynamic-runner",
"job": "Publish images",
})
if err != nil {
t.Fatal(err)
}
got, err := FromTask(&runnerv1.Task{
Id: 900,
Context: ctx,
WorkflowPayload: []byte("jobs:\n publish-images:\n runs-on: [self-hosted, vm]\n steps: []\n"),
}, "ddupan.top")
if err != nil {
t.Fatal(err)
}
if got.Repository != "panxiao81/gitea-dynamic-runner" || got.Task != "publish-images" {
t.Fatalf("unexpected task identity context: %#v", got)
}
want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/publish-images"
if got.SPIFFEID != want {
t.Fatalf("SPIFFE ID = %q, want %q", got.SPIFFEID, want)
}
}
func TestFromTaskRejectsUnsafeWorkflowJobKey(t *testing.T) {
ctx, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
if err != nil {
t.Fatal(err)
}
task := &runnerv1.Task{
Context: ctx,
WorkflowPayload: []byte("jobs:\n 'Run on Ubuntu':\n runs-on: self-hosted\n steps: []\n"),
}
if _, err := FromTask(task, "ddupan.top"); err == nil {
t.Fatal("expected unsafe job key to fail")
}
}
func TestBackoffTaskSegmentIsStableAndCollisionResistant(t *testing.T) {
got := BackoffTaskSegment("Run on Ubuntu")
if got != "Run-on-Ubuntu-8b7cd4c244fb" {
t.Fatalf("backoff segment = %q", got)
}
if got == BackoffTaskSegment("Run@on Ubuntu") {
t.Fatal("different legacy names must not collide after slugging")
}
}
func TestFromTaskRejectsIncompleteServerContext(t *testing.T) {
for _, fields := range []map[string]any{
{"repository": "invalid"},
{"repository": ""},
} {
ctx, err := structpb.NewStruct(fields)
if err != nil {
t.Fatal(err)
}
task := &runnerv1.Task{
Context: ctx,
WorkflowPayload: []byte("jobs:\n test:\n runs-on: self-hosted\n steps: []\n"),
}
if _, err := FromTask(task, "ddupan.top"); err == nil {
t.Fatalf("expected invalid task context to fail: %#v", fields)
}
}
}
+136
View File
@@ -0,0 +1,136 @@
package taskscheduler
import (
"context"
"errors"
"fmt"
"sync/atomic"
"time"
"golang.org/x/sync/errgroup"
runnerv1 "gitea.dev/actionslib/runner/v1"
)
type PollClient interface {
Declare(context.Context, string, []string) error
FetchTask(context.Context, int64) (*runnerv1.FetchTaskResponse, error)
}
type PollerConfig struct {
Version string
Labels []string
EmptyBackoff time.Duration
ErrorBackoff time.Duration
Capacity int
}
// Poller is the scheduler component. Each fetcher keeps its assigned task
// until that assignment is durably dispatched; all fetchers share one runner
// declaration and a monotonic tasks version.
type Poller struct {
Client PollClient
Scheduler *Scheduler
Config PollerConfig
OnError func(error)
}
func (p Poller) Run(ctx context.Context) error {
if p.Client == nil || p.Scheduler == nil {
return errors.New("Gitea poll client and task scheduler are required")
}
if p.Config.Version == "" || len(p.Config.Labels) == 0 {
return errors.New("runner version and labels are required")
}
if err := p.Client.Declare(ctx, p.Config.Version, p.Config.Labels); err != nil {
return fmt.Errorf("declare scheduler labels: %w", err)
}
emptyBackoff := p.Config.EmptyBackoff
if emptyBackoff <= 0 {
emptyBackoff = time.Second
}
errorBackoff := p.Config.ErrorBackoff
if errorBackoff <= 0 {
errorBackoff = 5 * time.Second
}
capacity := p.Config.Capacity
if capacity < 1 {
capacity = 1
}
var tasksVersion atomic.Int64
group, groupContext := errgroup.WithContext(ctx)
for range capacity {
group.Go(func() error { return p.runFetcher(groupContext, &tasksVersion, emptyBackoff, errorBackoff) })
}
return group.Wait()
}
func (p Poller) runFetcher(ctx context.Context, tasksVersion *atomic.Int64, emptyBackoff, errorBackoff time.Duration) error {
for {
response, err := p.Client.FetchTask(ctx, tasksVersion.Load())
if err != nil {
if ctx.Err() != nil {
return nil
}
p.report(fmt.Errorf("fetch Gitea task: %w", err))
if !wait(ctx, errorBackoff) {
return nil
}
continue
}
if response == nil {
p.report(errors.New("fetch Gitea task returned an empty response"))
if !wait(ctx, errorBackoff) {
return nil
}
continue
}
storeMaximum(tasksVersion, response.GetTasksVersion())
task := response.GetTask()
if task == nil {
if !wait(ctx, emptyBackoff) {
return nil
}
continue
}
for {
if err := p.Scheduler.Run(ctx, task); err == nil {
break
} else {
if ctx.Err() != nil {
return nil
}
p.report(fmt.Errorf("dispatch Gitea task %d: %w", task.GetId(), err))
}
if !wait(ctx, errorBackoff) {
return nil
}
}
}
}
func storeMaximum(value *atomic.Int64, candidate int64) {
for current := value.Load(); candidate > current; current = value.Load() {
if value.CompareAndSwap(current, candidate) {
return
}
}
}
func (p Poller) report(err error) {
if p.OnError != nil {
p.OnError(err)
}
}
func wait(ctx context.Context, duration time.Duration) bool {
timer := time.NewTimer(duration)
defer timer.Stop()
select {
case <-ctx.Done():
return false
case <-timer.C:
return true
}
}
+166
View File
@@ -0,0 +1,166 @@
package taskscheduler
import (
"context"
"errors"
"sync"
"testing"
"time"
runnerv1 "gitea.dev/actionslib/runner/v1"
"google.golang.org/protobuf/types/known/structpb"
)
type fakePollClient struct {
mu sync.Mutex
declared int
fetches int
responses []*runnerv1.FetchTaskResponse
}
func (c *fakePollClient) Declare(context.Context, string, []string) error {
c.mu.Lock()
defer c.mu.Unlock()
c.declared++
return nil
}
func (c *fakePollClient) FetchTask(ctx context.Context, _ int64) (*runnerv1.FetchTaskResponse, error) {
c.mu.Lock()
c.fetches++
if len(c.responses) > 0 {
response := c.responses[0]
c.responses = c.responses[1:]
c.mu.Unlock()
return response, nil
}
c.mu.Unlock()
<-ctx.Done()
return nil, ctx.Err()
}
type retryDispatcher struct {
mu sync.Mutex
calls int
failures int
done chan struct{}
fetches func() int
fetchesAtSuccess int
}
func (d *retryDispatcher) Dispatch(context.Context, Assignment) error {
d.mu.Lock()
defer d.mu.Unlock()
d.calls++
if d.failures > 0 {
d.failures--
return errors.New("JetStream unavailable")
}
if d.fetches != nil {
d.fetchesAtSuccess = d.fetches()
}
select {
case <-d.done:
default:
close(d.done)
}
return nil
}
func pollTask(t *testing.T) *runnerv1.Task {
t.Helper()
fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"})
if err != nil {
t.Fatal(err)
}
return &runnerv1.Task{
Id: 42, Context: fields,
WorkflowPayload: []byte("jobs:\n test:\n runs-on: [self-hosted, pod]\n steps: []\n"),
}
}
func TestPollerRetriesAssignedTaskBeforeFetchingAnother(t *testing.T) {
client := &fakePollClient{responses: []*runnerv1.FetchTaskResponse{{Task: pollTask(t), TasksVersion: 7}}}
dispatcher := &retryDispatcher{failures: 2, done: make(chan struct{})}
dispatcher.fetches = func() int {
client.mu.Lock()
defer client.mu.Unlock()
return client.fetches
}
poller := Poller{
Client: client,
Scheduler: &Scheduler{TrustDomain: "ddupan.top", Dispatcher: dispatcher},
Config: PollerConfig{
Version: "dev", Labels: []string{"self-hosted:host", "pod:host", "vm:host"},
EmptyBackoff: time.Millisecond, ErrorBackoff: time.Millisecond,
},
}
ctx, cancel := context.WithCancel(context.Background())
finished := make(chan error, 1)
go func() { finished <- poller.Run(ctx) }()
select {
case <-dispatcher.done:
case <-time.After(time.Second):
t.Fatal("assignment was not dispatched")
}
cancel()
if err := <-finished; err != nil {
t.Fatal(err)
}
dispatcher.mu.Lock()
defer dispatcher.mu.Unlock()
client.mu.Lock()
defer client.mu.Unlock()
if dispatcher.calls != 3 || dispatcher.fetchesAtSuccess != 1 || client.declared != 1 {
t.Fatalf("dispatches=%d fetches-before-dispatch=%d declares=%d", dispatcher.calls, dispatcher.fetchesAtSuccess, client.declared)
}
}
type blockingPollClient struct {
mu sync.Mutex
declared int
started chan struct{}
}
func (c *blockingPollClient) Declare(context.Context, string, []string) error {
c.mu.Lock()
c.declared++
c.mu.Unlock()
return nil
}
func (c *blockingPollClient) FetchTask(ctx context.Context, _ int64) (*runnerv1.FetchTaskResponse, error) {
c.started <- struct{}{}
<-ctx.Done()
return nil, ctx.Err()
}
func TestPollerStartsConfiguredNumberOfFetchersAfterOneDeclare(t *testing.T) {
client := &blockingPollClient{started: make(chan struct{}, 3)}
poller := Poller{
Client: client,
Scheduler: &Scheduler{TrustDomain: "ddupan.top", Dispatcher: &retryDispatcher{done: make(chan struct{})}},
Config: PollerConfig{
Version: "dev", Labels: []string{"self-hosted:host", "pod:host", "vm:host"}, Capacity: 3,
},
}
ctx, cancel := context.WithCancel(context.Background())
finished := make(chan error, 1)
go func() { finished <- poller.Run(ctx) }()
for range 3 {
select {
case <-client.started:
case <-time.After(time.Second):
t.Fatal("configured fetchers did not start")
}
}
cancel()
if err := <-finished; err != nil {
t.Fatal(err)
}
client.mu.Lock()
defer client.mu.Unlock()
if client.declared != 1 {
t.Fatalf("declares = %d", client.declared)
}
}
+38
View File
@@ -0,0 +1,38 @@
// Package taskscheduler owns tasks fetched through Gitea's RunnerService and
// dispatches them to an executor only after their identity is known.
package taskscheduler
import (
"context"
"errors"
runnerv1 "gitea.dev/actionslib/runner/v1"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)
type Assignment = taskassignment.Assignment
// Dispatcher creates exactly one executor for an already assigned Gitea task.
// It must not register another runner or ask Gitea for a task.
type Dispatcher interface {
Dispatch(context.Context, Assignment) error
}
// Scheduler implements the TaskRunner boundary used by Gitea Runner's poller.
type Scheduler struct {
TrustDomain string
Dispatcher Dispatcher
}
// Run derives identity from the fetched task before provisioning its executor.
func (s *Scheduler) Run(ctx context.Context, task *runnerv1.Task) error {
if s.Dispatcher == nil {
return errors.New("executor dispatcher is required")
}
assignment, err := taskassignment.New(task, s.TrustDomain)
if err != nil {
return err
}
return s.Dispatcher.Dispatch(ctx, assignment)
}
+56
View File
@@ -0,0 +1,56 @@
package taskscheduler
import (
"context"
"testing"
runnerv1 "gitea.dev/actionslib/runner/v1"
"google.golang.org/protobuf/types/known/structpb"
)
type recordingDispatcher struct {
assignment Assignment
}
func (d *recordingDispatcher) Dispatch(_ context.Context, assignment Assignment) error {
d.assignment = assignment
return nil
}
func TestRunDerivesIdentityBeforeDispatch(t *testing.T) {
taskContext, err := structpb.NewStruct(map[string]any{
"repository": "panxiao81/gitea-dynamic-runner",
"job": "test",
})
if err != nil {
t.Fatal(err)
}
task := &runnerv1.Task{
Id: 42,
Context: taskContext,
WorkflowPayload: []byte("jobs:\n test:\n runs-on: [self-hosted, pod]\n steps: []\n"),
}
dispatcher := &recordingDispatcher{}
scheduler := Scheduler{TrustDomain: "ddupan.top", Dispatcher: dispatcher}
if err := scheduler.Run(context.Background(), task); err != nil {
t.Fatal(err)
}
if dispatcher.assignment.Task != task {
t.Fatal("dispatcher did not receive the fetched task")
}
if dispatcher.assignment.ID != "gitea-task-42" {
t.Fatalf("assignment ID = %q", dispatcher.assignment.ID)
}
want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/test"
if dispatcher.assignment.Identity.SPIFFEID != want {
t.Fatalf("SPIFFE ID = %q, want %q", dispatcher.assignment.Identity.SPIFFEID, want)
}
}
func TestRunRequiresDispatcher(t *testing.T) {
scheduler := Scheduler{TrustDomain: "ddupan.top"}
if err := scheduler.Run(context.Background(), &runnerv1.Task{}); err == nil {
t.Fatal("expected missing dispatcher to fail")
}
}
+191
View File
@@ -0,0 +1,191 @@
// Package taskworker reconciles assigned Gitea tasks against an executor backend.
package taskworker
import (
"context"
"errors"
"strconv"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
)
// Phase is observed from Kubernetes or OpenSandbox, never stored by the worker.
type Phase string
const (
PhasePending Phase = "pending"
PhaseRunning Phase = "running"
PhaseSucceeded Phase = "succeeded"
PhaseFailed Phase = "failed"
)
// Executor is the backend resource discovered by assignment ID.
type Executor struct {
Name string
IdentityTarget string
Phase Phase
}
// Metadata is persisted on the backend resource. Labels remain query-safe;
// annotations retain complete identity context.
type Metadata struct {
Labels map[string]string
Annotations map[string]string
}
// LaunchSpec contains the durable resource metadata and the short-lived
// executor environment. Environment values configure the one-shot runner but
// are deliberately excluded from labels and annotations.
type LaunchSpec struct {
Metadata Metadata
Environment map[string]string
}
// Bootstrap produces assignment-scoped executor configuration. Implementations
// must be deterministic so a redelivery after controller restart creates the
// same credentials without storing another lifecycle record.
type Bootstrap interface {
Environment(taskassignment.Assignment) (map[string]string, error)
}
// Backend is implemented by the native Pod and OpenSandbox adapters.
// Every method must be idempotent.
type Backend interface {
Find(context.Context, string) (*Executor, error)
Create(context.Context, taskassignment.Assignment, LaunchSpec) (*Executor, error)
BindIdentity(context.Context, *Executor, taskidentity.Identity) error
Delete(context.Context, *Executor) error
}
// TaskState uses Gitea as the authority for whether an assigned task has
// already reached a terminal state.
type TaskState interface {
Terminal(context.Context, int64) (bool, error)
Report(context.Context, int64, Phase) error
}
// Worker has no correctness-critical in-memory state. Handle may be called
// again for the same assignment after any operation.
type Worker struct {
Backend Backend
Tasks TaskState
Bootstrap Bootstrap
}
// Accept completes the durable handoff from JetStream to the backend. Once it
// returns true, all recovery information exists in Kubernetes/OpenSandbox and
// the assignment message can be acknowledged immediately.
func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment) (bool, error) {
if w.Backend == nil || w.Bootstrap == nil {
return false, errors.New("backend and runner bootstrap are required")
}
if assignment.ID == "" || assignment.Task == nil {
return false, errors.New("valid assignment is required")
}
executor, err := w.Backend.Find(ctx, assignment.ID)
if err != nil {
return false, err
}
if executor == nil {
launch, launchErr := w.launchSpec(assignment)
if launchErr != nil {
return false, launchErr
}
executor, err = w.Backend.Create(ctx, assignment, launch)
if err != nil {
return false, err
}
}
if executor.IdentityTarget == "" {
return false, nil
}
if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
return false, err
}
return true, nil
}
// Handle performs one reconciliation. Done means the queue message may be
// acknowledged. A false result should remain pending and be reconciled again.
func (w Worker) Handle(ctx context.Context, assignment taskassignment.Assignment) (done bool, err error) {
if w.Backend == nil || w.Tasks == nil || w.Bootstrap == nil {
return false, errors.New("backend, Gitea task state, and runner bootstrap are required")
}
if assignment.ID == "" || assignment.Task == nil {
return false, errors.New("valid assignment is required")
}
terminal, err := w.Tasks.Terminal(ctx, assignment.Task.GetId())
if err != nil {
return false, err
}
executor, err := w.Backend.Find(ctx, assignment.ID)
if err != nil {
return false, err
}
if terminal {
if executor != nil {
if err := w.Backend.Delete(ctx, executor); err != nil {
return false, err
}
}
return true, nil
}
if executor == nil {
launch, launchErr := w.launchSpec(assignment)
if launchErr != nil {
return false, launchErr
}
executor, err = w.Backend.Create(ctx, assignment, launch)
if err != nil {
return false, err
}
}
if executor.IdentityTarget == "" {
return false, nil
}
if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
return false, err
}
switch executor.Phase {
case PhaseSucceeded, PhaseFailed:
// Reporting first closes the delete-before-ack crash window: after a
// restart Gitea prevents this assignment from executing a second time.
if err := w.Tasks.Report(ctx, assignment.Task.GetId(), executor.Phase); err != nil {
return false, err
}
if err := w.Backend.Delete(ctx, executor); err != nil {
return false, err
}
return true, nil
default:
return false, nil
}
}
func (w Worker) launchSpec(assignment taskassignment.Assignment) (LaunchSpec, error) {
environment, err := w.Bootstrap.Environment(assignment)
if err != nil {
return LaunchSpec{}, err
}
return LaunchSpec{Metadata: BackendMetadata(assignment), Environment: environment}, nil
}
// BackendMetadata is the shared metadata contract for Pods and OpenSandbox.
func BackendMetadata(assignment taskassignment.Assignment) Metadata {
return Metadata{
Labels: map[string]string{
"ci.ddupan.top/assignment-id": assignment.ID,
"ci.ddupan.top/task-id": strconv.FormatInt(assignment.Task.GetId(), 10),
"ci.ddupan.top/backend": string(assignment.Backend),
},
Annotations: map[string]string{
"ci.ddupan.top/repository": assignment.Identity.Repository,
"ci.ddupan.top/job-key": assignment.Identity.Task,
"ci.ddupan.top/spiffe-id": assignment.Identity.SPIFFEID,
},
}
}
+135
View File
@@ -0,0 +1,135 @@
package taskworker
import (
"context"
"testing"
runnerv1 "gitea.dev/actionslib/runner/v1"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
)
type fakeBackend struct {
executor *Executor
created int
bound int
deleted int
}
type fakeBootstrap struct{}
func (fakeBootstrap) Environment(taskassignment.Assignment) (map[string]string, error) {
return map[string]string{"CI_RUNNER_CAPABILITY": "capability"}, nil
}
func (b *fakeBackend) Find(context.Context, string) (*Executor, error) { return b.executor, nil }
func (b *fakeBackend) Create(_ context.Context, _ taskassignment.Assignment, _ LaunchSpec) (*Executor, error) {
b.created++
b.executor = &Executor{Name: "executor", IdentityTarget: "pod-uid", Phase: PhaseRunning}
return b.executor, nil
}
func (b *fakeBackend) BindIdentity(context.Context, *Executor, taskidentity.Identity) error {
b.bound++
return nil
}
func (b *fakeBackend) Delete(context.Context, *Executor) error {
b.deleted++
b.executor = nil
return nil
}
type fakeTasks struct {
terminal bool
reported []Phase
}
func (t *fakeTasks) Terminal(context.Context, int64) (bool, error) { return t.terminal, nil }
func (t *fakeTasks) Report(_ context.Context, _ int64, phase Phase) error {
t.reported = append(t.reported, phase)
t.terminal = true
return nil
}
func assignment() taskassignment.Assignment {
return taskassignment.Assignment{
ID: "gitea-task-42",
Backend: taskassignment.BackendPod,
Task: &runnerv1.Task{Id: 42},
Identity: taskidentity.Identity{
Repository: "owner/repo",
Task: "publish",
SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish",
},
}
}
func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) {
backend := &fakeBackend{executor: &Executor{Name: "existing", IdentityTarget: "uid", Phase: PhaseRunning}}
worker := Worker{Backend: backend, Tasks: &fakeTasks{}, Bootstrap: fakeBootstrap{}}
done, err := worker.Handle(context.Background(), assignment())
if err != nil || done {
t.Fatalf("Handle() = (%v, %v), want pending", done, err)
}
if backend.created != 0 || backend.bound != 1 {
t.Fatalf("created=%d bound=%d", backend.created, backend.bound)
}
}
func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
backend := &fakeBackend{}
worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}}
accepted, err := worker.Accept(context.Background(), assignment())
if err != nil || !accepted {
t.Fatalf("Accept() = (%v, %v), want accepted", accepted, err)
}
if backend.created != 1 || backend.bound != 1 || backend.deleted != 0 {
t.Fatalf("created=%d bound=%d deleted=%d", backend.created, backend.bound, backend.deleted)
}
}
func TestAcceptRetriesWhileBackendIdentityTargetIsUnavailable(t *testing.T) {
backend := &fakeBackend{executor: &Executor{Name: "pending", Phase: PhasePending}}
worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}}
accepted, err := worker.Accept(context.Background(), assignment())
if err != nil || accepted {
t.Fatalf("Accept() = (%v, %v), want retry", accepted, err)
}
if backend.bound != 0 {
t.Fatalf("identity bindings = %d", backend.bound)
}
}
func TestHandleReportsBeforeCleanupAndBecomesRecoverable(t *testing.T) {
backend := &fakeBackend{executor: &Executor{Name: "finished", IdentityTarget: "uid", Phase: PhaseSucceeded}}
tasks := &fakeTasks{}
worker := Worker{Backend: backend, Tasks: tasks, Bootstrap: fakeBootstrap{}}
done, err := worker.Handle(context.Background(), assignment())
if err != nil || !done {
t.Fatalf("Handle() = (%v, %v), want done", done, err)
}
if len(tasks.reported) != 1 || backend.deleted != 1 {
t.Fatalf("reported=%v deleted=%d", tasks.reported, backend.deleted)
}
// Simulate redelivery after deletion but before the queue ACK. Gitea's
// terminal state prevents a duplicate executor from being created.
done, err = worker.Handle(context.Background(), assignment())
if err != nil || !done || backend.created != 0 {
t.Fatalf("recovery = (%v, %v), created=%d", done, err, backend.created)
}
}
func TestBackendMetadataContainsRecoveryKeys(t *testing.T) {
metadata := BackendMetadata(assignment())
if metadata.Labels["ci.ddupan.top/assignment-id"] != "gitea-task-42" {
t.Fatalf("labels = %#v", metadata.Labels)
}
if metadata.Annotations["ci.ddupan.top/spiffe-id"] != "spiffe://ddupan.top/ci/owner/repo/publish" {
t.Fatalf("annotations = %#v", metadata.Annotations)
}
}
+1
View File
@@ -16,6 +16,7 @@ test = ["pytest==8.4.2", "pytest-asyncio==1.2.0"]
gitea-dynamic-runner-controller = "gitea_dynamic_runner.controller:main" gitea-dynamic-runner-controller = "gitea_dynamic_runner.controller:main"
gitea-dynamic-runner-pod-worker = "gitea_dynamic_runner.pod_worker:cli" gitea-dynamic-runner-pod-worker = "gitea_dynamic_runner.pod_worker:cli"
gitea-dynamic-runner-vm-worker = "gitea_dynamic_runner.worker:cli" gitea-dynamic-runner-vm-worker = "gitea_dynamic_runner.worker:cli"
gitea-dynamic-runner-opensandbox-identity = "gitea_dynamic_runner.opensandbox_identity:cli"
gitea-spire-jwt-broker = "gitea_dynamic_runner.jwt_broker:main" gitea-spire-jwt-broker = "gitea_dynamic_runner.jwt_broker:main"
[tool.pytest.ini_options] [tool.pytest.ini_options]
+4 -1
View File
@@ -6,6 +6,8 @@ instance=${2:?Gitea instance is required}
runner_name=${3:?runner name is required} runner_name=${3:?runner name is required}
runner_labels=${4:?runner labels are required} runner_labels=${4:?runner labels are required}
token_file=/run/gitea-runner-registration-token token_file=/run/gitea-runner-registration-token
config_file=${GITEA_RUNNER_CONFIG_FILE:-/etc/gitea-runner/config-vm-bootstrap.yaml}
export HOME="${HOME:-/root}"
cleanup() { cleanup() {
rm -f -- "$token_file" rm -f -- "$token_file"
@@ -25,4 +27,5 @@ gitea-runner register \
--labels "$runner_labels" \ --labels "$runner_labels" \
--token-file "$token_file" --token-file "$token_file"
rm -f -- "$token_file" rm -f -- "$token_file"
gitea-runner daemon gitea-runner daemon --config "$config_file" --once
echo gitea-runner-job-complete >/dev/ttyS0
+30
View File
@@ -0,0 +1,30 @@
#!/usr/bin/env bash
set -euo pipefail
: "${CI_SPIFFE_ID:?CI_SPIFFE_ID is required}"
: "${SPIFFE_ENDPOINT_SOCKET:?SPIFFE_ENDPOINT_SOCKET is required}"
: "${GITEA_RUNNER_REGISTRATION_TOKEN_URL:?GITEA_RUNNER_REGISTRATION_TOKEN_URL is required}"
socket_path=${SPIFFE_ENDPOINT_SOCKET#unix://}
token_file=$(mktemp)
trap 'rm -f -- "$token_file"' EXIT
export GITEA_RUNNER_REGISTRATION_TOKEN_FILE="$token_file"
deadline=$((SECONDS + 120))
while (( SECONDS < deadline )); do
if /opt/spire/bin/spire-agent api fetch x509 \
-socketPath "$socket_path" \
-output json 2>/dev/null | \
jq -e --arg id "$CI_SPIFFE_ID" \
'any(.svids[]?; .spiffe_id == $id)' >/dev/null; then
umask 077
curl --fail --silent --show-error --retry 10 --retry-all-errors \
--connect-timeout 2 --max-time 30 \
"$GITEA_RUNNER_REGISTRATION_TOKEN_URL" >"$token_file"
/usr/local/bin/run.sh
exit $?
fi
sleep 1
done
printf 'timed out waiting for OpenSandbox SPIFFE identity %s\n' "$CI_SPIFFE_ID" >&2
exit 1
+24 -11
View File
@@ -29,8 +29,13 @@ tap="mvr${instance_id%%-*}"
overlay="$vm_dir/root.qcow2" overlay="$vm_dir/root.qcow2"
seed="$vm_dir/seed.img" seed="$vm_dir/seed.img"
serial="$vm_dir/serial.log" serial="$vm_dir/serial.log"
log_dir="$state_root/logs"
cleanup() { cleanup() {
if test -f "$serial"; then
install -d -m 0700 "$log_dir"
cp "$serial" "$log_dir/$instance_id.log"
fi
ip link delete "$tap" 2>/dev/null || true ip link delete "$tap" 2>/dev/null || true
rm -rf -- "$vm_dir" rm -rf -- "$vm_dir"
} }
@@ -39,7 +44,10 @@ trap cleanup EXIT INT TERM
test -r "$base_image" test -r "$base_image"
test -r "$firmware" test -r "$firmware"
install -d -m 0700 "$state_root/instances" "$vm_dir" install -d -m 0700 "$state_root/instances" "$vm_dir"
qemu-img create -q -f qcow2 -F qcow2 -b "$base_image" "$overlay" # Cloud Hypervisor cannot open qcow2 backing chains. Keep each disposable
# root disk flat and bypass the LXC page cache: caching both a 5 GiB copy and
# guest RAM can otherwise trigger the container memory limit.
qemu-img convert -q -T none -t none -f qcow2 -O qcow2 "$base_image" "$overlay"
cat >"$vm_dir/meta-data" <<EOF cat >"$vm_dir/meta-data" <<EOF
instance-id: $instance_id instance-id: $instance_id
@@ -48,7 +56,8 @@ EOF
cat >"$vm_dir/user-data" <<EOF cat >"$vm_dir/user-data" <<EOF
#cloud-config #cloud-config
runcmd: runcmd:
- [ /usr/local/libexec/gitea-microvm-guest-runner, "$token_url/token/$nonce", "$gitea_instance", "gitea-${instance_id%%-*}", "$runner_labels" ] - [ sh, -c, "curl --fail --silent --show-error --retry 10 --retry-all-errors $token_url/assets/guest-assets.tar.gz | tar -xz -C /" ]
- [ /usr/local/libexec/gitea-microvm-guest-runner, "$token_url/token/$nonce", "$gitea_instance", "gitea-vm-${instance_id%%-*}", "$runner_labels" ]
EOF EOF
cloud-localds "$seed" "$vm_dir/user-data" "$vm_dir/meta-data" cloud-localds "$seed" "$vm_dir/user-data" "$vm_dir/meta-data"
@@ -63,12 +72,16 @@ ip tuntap add dev "$tap" mode tap
ip link set "$tap" master "$bridge" ip link set "$tap" master "$bridge"
ip link set "$tap" up ip link set "$tap" up
timeout --signal=TERM --kill-after=30s "$vm_timeout" "$cloud_hypervisor" \ if timeout --signal=TERM --kill-after=30s "$vm_timeout" "$cloud_hypervisor" \
--firmware "$firmware" \ --firmware "$firmware" \
--cpus "boot=$cpus" \ --cpus "boot=$cpus" \
--memory "size=$memory" \ --memory "size=$memory,shared=on" \
--disk "path=$overlay" \ --disk "path=$overlay,image_type=qcow2,direct=on,sparse=off" \
--disk "path=$seed,readonly=on" \ --disk "path=$seed,readonly=on,image_type=raw,direct=on,sparse=off" \
--net "tap=$tap,mac=$mac" \ --net "tap=$tap,mac=$mac" \
--serial "file=$serial" \ --serial "file=$serial" \
--console off --console off; then
grep -Fq gitea-runner-job-complete "$serial"
else
exit $?
fi
+247 -59
View File
@@ -1,61 +1,77 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Gitea workflow_job webhook to NATS JetStream producer.""" """Persist Gitea workflow_job events and optionally schedule VM sandboxes."""
from __future__ import annotations
import asyncio
import hashlib import hashlib
import hmac import hmac
import json import json
import logging
import os import os
import ssl import ssl
from pathlib import Path from pathlib import Path
import nats import nats
from aiohttp import web from aiohttp import web
from nats.js.api import DiscardPolicy, RetentionPolicy, StorageType, StreamConfig from nats.errors import TimeoutError as NatsTimeoutError
from nats.js.api import (
AckPolicy,
ConsumerConfig,
DiscardPolicy,
RetentionPolicy,
StorageType,
StreamConfig,
)
from nats.js.errors import NotFoundError from nats.js.errors import NotFoundError
from .models import IdentityBinding, RunnerRequest from .models import IdentityBinding, RunnerRequest
from .opensandbox import OpenSandboxClient
from .opensandbox_worker import OpenSandboxScheduler, RegistrationTokens
SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner") LOG = logging.getLogger(__name__)
STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER") WEBHOOK_SECRET_FILE = Path(
os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret")
)
REGISTRATION_TOKEN_FILE = Path(
os.environ.get("REGISTRATION_TOKEN_FILE", "/run/secrets/gitea/registration-token")
)
OPENSANDBOX_API = os.environ.get("OPENSANDBOX_API", "http://10.60.0.13:8080")
OPENSANDBOX_API_KEY_FILE = Path(
os.environ.get(
"OPENSANDBOX_API_KEY_FILE",
"/run/secrets/opensandbox/api-key",
)
)
NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222") NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
NATS_USER = os.environ.get("NATS_USER", "ci-producer")
NATS_PASSWORD_FILE = Path(os.environ.get("NATS_PASSWORD_FILE", "/run/secrets/nats/password"))
NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt") NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt")
WEBHOOK_SECRET_FILE = Path(os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret")) NATS_PRODUCER_USER = os.environ.get("NATS_PRODUCER_USER", "ci-producer")
NATS_PRODUCER_PASSWORD_FILE = Path(
os.environ.get("NATS_PRODUCER_PASSWORD_FILE", "/run/secrets/nats/producer-password")
)
NATS_WORKER_USER = os.environ.get("NATS_WORKER_USER", "ci-worker")
NATS_WORKER_PASSWORD_FILE = Path(
os.environ.get("NATS_WORKER_PASSWORD_FILE", "/run/secrets/nats/worker-password")
)
NATS_STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
NATS_SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
VM_CONSUMER_ENABLED = os.environ.get("VM_CONSUMER_ENABLED", "false") == "true"
VM_CAPACITY = int(os.environ.get("VM_CAPACITY", "1"))
def accepts(payload: object) -> tuple[bool, str | None]: def accepts(payload: object) -> tuple[bool, str | None]:
"""Compatibility helper for callers that only need acceptance and identity."""
request = RunnerRequest.from_webhook(payload) request = RunnerRequest.from_webhook(payload)
return (request is not None, str(request.job_id) if request else None) return (request is not None, str(request.job_id) if request else None)
def valid_signature(body: bytes, signature: str) -> bool: def valid_signature(body: bytes, signature: str) -> bool:
expected = hmac.new(WEBHOOK_SECRET_FILE.read_bytes().strip(), body, hashlib.sha256).hexdigest() expected = hmac.new(
WEBHOOK_SECRET_FILE.read_bytes().strip(), body, hashlib.sha256
).hexdigest()
return hmac.compare_digest(signature.removeprefix("sha256="), expected) return hmac.compare_digest(signature.removeprefix("sha256="), expected)
async def ensure_stream(js: object) -> None:
config = StreamConfig(
name=STREAM,
subjects=[f"{SUBJECT_PREFIX}.>"],
retention=RetentionPolicy.WORK_QUEUE,
storage=StorageType.FILE,
discard=DiscardPolicy.OLD,
max_age=24 * 60 * 60,
max_msgs=10_000,
max_bytes=256 * 1024 * 1024,
duplicate_window=24 * 60 * 60,
)
try:
await js.stream_info(STREAM)
except NotFoundError:
await js.add_stream(config=config)
else:
await js.update_stream(config=config)
async def webhook(request: web.Request) -> web.Response: async def webhook(request: web.Request) -> web.Response:
body = await request.read() body = await request.read()
if not valid_signature(body, request.headers.get("X-Gitea-Signature", "")): if not valid_signature(body, request.headers.get("X-Gitea-Signature", "")):
@@ -64,58 +80,230 @@ async def webhook(request: web.Request) -> web.Response:
payload = json.loads(body) payload = json.loads(body)
except json.JSONDecodeError as error: except json.JSONDecodeError as error:
raise web.HTTPBadRequest(text="invalid JSON\n") from error raise web.HTTPBadRequest(text="invalid JSON\n") from error
runner_request = RunnerRequest.from_webhook(payload) context = _webhook_context(payload)
if runner_request is not None: LOG.info("workflow_job webhook received %s", context)
await request.app["js"].publish(
f"{SUBJECT_PREFIX}.{runner_request.backend}",
runner_request.to_json(),
headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
)
return web.Response(status=202, text="queued\n")
binding = IdentityBinding.from_webhook(payload) if isinstance(payload, dict) and payload.get("action") == "completed":
if binding is not None: job = payload.get("workflow_job")
runner_name = job.get("runner_name") if isinstance(job, dict) else None
scheduler = request.app.get("scheduler")
if (
isinstance(runner_name, str)
and runner_name.startswith("gitea-vm-")
and scheduler is not None
):
cleaned = await scheduler.complete(runner_name)
LOG.info(
"completed runner cleanup runner=%r cleaned=%s %s",
runner_name,
cleaned,
context,
)
return web.Response(status=204)
runner_request = RunnerRequest.from_webhook(payload)
if runner_request is None:
binding = IdentityBinding.from_webhook(payload)
if binding is None:
LOG.info("workflow_job webhook ignored %s", context)
return web.Response(status=204)
subject = f"{NATS_SUBJECT_PREFIX}.{binding.backend}.binding"
await request.app["js"].publish( await request.app["js"].publish(
f"{SUBJECT_PREFIX}.{binding.backend}.binding", subject,
binding.to_json(), binding.to_json(),
headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"}, headers={
"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"
},
) )
LOG.info("runner identity binding persisted subject=%s %s", subject, context)
return web.Response(status=202, text="binding queued\n") return web.Response(status=202, text="binding queued\n")
return web.Response(status=204) subject = f"{NATS_SUBJECT_PREFIX}.{runner_request.backend}"
await request.app["js"].publish(
subject,
runner_request.to_json(),
headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
)
LOG.info("runner request persisted subject=%s %s", subject, context)
return web.Response(status=202, text="queued\n")
async def registration_token(request: web.Request) -> web.Response:
scheduler = request.app.get("scheduler")
if scheduler is None:
raise web.HTTPNotFound()
value = await scheduler.tokens.consume(request.match_info["nonce"])
if value is None:
raise web.HTTPNotFound()
return web.Response(body=value, headers={"Cache-Control": "no-store"})
def _webhook_context(payload: object) -> str:
if not isinstance(payload, dict):
return f"payload_type={type(payload).__name__}"
job = payload.get("workflow_job")
repository = payload.get("repository")
job = job if isinstance(job, dict) else {}
repository = repository if isinstance(repository, dict) else {}
return (
f"action={payload.get('action')!r} job_id={job.get('id')!r} "
f"run_id={job.get('run_id')!r} runner_name={job.get('runner_name')!r} "
f"repository={repository.get('full_name')!r} job_name={job.get('name')!r} "
f"labels={job.get('labels')!r}"
)
async def health(request: web.Request) -> web.Response: async def health(request: web.Request) -> web.Response:
connected = request.app["nc"].is_connected connected = request.app["nc"].is_connected
return web.Response(text="ok\n" if connected else "disconnected\n", status=200 if connected else 503) return web.Response(
text="ok\n" if connected else "disconnected\n",
status=200 if connected else 503,
async def nats_context(app: web.Application):
tls = ssl.create_default_context(cafile=NATS_CA_FILE)
nc = await nats.connect(
NATS_URL,
user=NATS_USER,
password=NATS_PASSWORD_FILE.read_text().strip(),
tls=tls,
name="microvm-runner-controller",
) )
app["nc"] = nc
app["js"] = nc.jetstream()
async def ensure_stream(js: object) -> None:
config = StreamConfig(
name=NATS_STREAM,
subjects=[f"{NATS_SUBJECT_PREFIX}.>"],
retention=RetentionPolicy.WORK_QUEUE,
storage=StorageType.FILE,
discard=DiscardPolicy.OLD,
max_age=24 * 60 * 60,
max_msgs=10_000,
max_bytes=256 * 1024 * 1024,
duplicate_window=24 * 60 * 60,
)
try:
await js.stream_info(NATS_STREAM)
except NotFoundError:
await js.add_stream(config=config)
else:
await js.update_stream(config=config)
async def run_message(message: object, scheduler: OpenSandboxScheduler) -> None:
try:
runner_request = RunnerRequest.from_json(message.data)
await scheduler.create(runner_request)
task = scheduler.active[runner_request.job_id]
try:
while not task.done():
try:
await asyncio.wait_for(asyncio.shield(task), timeout=30)
except asyncio.TimeoutError:
await message.in_progress()
await task
except asyncio.CancelledError:
# A completed webhook cancels the lifecycle monitor after its
# Lifecycle DELETE succeeds. That is successful message handling.
# During controller shutdown the scheduler still owns the job, so
# preserve the unacked message for redelivery.
if runner_request.job_id in scheduler.active:
raise
except ValueError as error:
LOG.info("runner request already active: %s", error)
await message.nak(delay=5)
except Exception:
LOG.exception("persistent runner request failed")
await message.nak(delay=15)
else:
await message.ack()
async def consume_vm_requests(js: object, scheduler: OpenSandboxScheduler) -> None:
subscription = await js.pull_subscribe(
f"{NATS_SUBJECT_PREFIX}.vm",
durable="vm",
stream=NATS_STREAM,
config=ConsumerConfig(
durable_name="vm",
filter_subject=f"{NATS_SUBJECT_PREFIX}.vm",
ack_policy=AckPolicy.EXPLICIT,
ack_wait=5 * 60,
max_ack_pending=VM_CAPACITY,
max_deliver=20,
),
)
active: set[asyncio.Task[None]] = set()
while True:
active = {task for task in active if not task.done()}
try:
messages = await subscription.fetch(
batch=max(1, VM_CAPACITY - len(active)), timeout=5
)
except NatsTimeoutError:
continue
for message in messages:
task = asyncio.create_task(run_message(message, scheduler))
active.add(task)
async def runtime_context(app: web.Application):
tls = ssl.create_default_context(cafile=NATS_CA_FILE)
producer = await nats.connect(
NATS_URL,
user=NATS_PRODUCER_USER,
password=NATS_PRODUCER_PASSWORD_FILE.read_text().strip(),
tls=tls,
name="opensandbox-runner-controller-producer",
)
app["nc"] = producer
app["js"] = producer.jetstream()
await ensure_stream(app["js"]) await ensure_stream(app["js"])
yield if not VM_CONSUMER_ENABLED:
await nc.drain() try:
yield
finally:
await producer.drain()
return
if VM_CAPACITY < 1:
raise ValueError("VM_CAPACITY must be at least 1")
tokens = RegistrationTokens(REGISTRATION_TOKEN_FILE.read_bytes())
worker = await nats.connect(
NATS_URL,
user=NATS_WORKER_USER,
password=NATS_WORKER_PASSWORD_FILE.read_text().strip(),
tls=ssl.create_default_context(cafile=NATS_CA_FILE),
name="opensandbox-vm-runner-worker",
)
async with OpenSandboxClient(
api_url=OPENSANDBOX_API,
api_key_file=OPENSANDBOX_API_KEY_FILE,
) as client:
scheduler = OpenSandboxScheduler(client, tokens)
app["opensandbox_client"] = client
app["scheduler"] = scheduler
consumer = asyncio.create_task(
consume_vm_requests(worker.jetstream(), scheduler),
name="opensandbox-vm-consumer",
)
try:
yield
finally:
consumer.cancel()
await asyncio.gather(consumer, return_exceptions=True)
await scheduler.close()
await worker.drain()
await producer.drain()
def create_app() -> web.Application: def create_app() -> web.Application:
app = web.Application(client_max_size=1024 * 1024) app = web.Application(client_max_size=1024 * 1024)
app.cleanup_ctx.append(nats_context) app.cleanup_ctx.append(runtime_context)
app.router.add_post("/webhook", webhook) app.router.add_post("/webhook", webhook)
app.router.add_get("/token/{nonce}", registration_token)
app.router.add_get("/healthz", health) app.router.add_get("/healthz", health)
return app return app
def main() -> None: def main() -> None:
web.run_app(create_app(), host=os.environ.get("LISTEN", "0.0.0.0"), port=int(os.environ.get("PORT", "8787"))) logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
web.run_app(
create_app(),
host=os.environ.get("LISTEN", "0.0.0.0"),
port=int(os.environ.get("PORT", "8787")),
)
if __name__ == "__main__": if __name__ == "__main__":
+79
View File
@@ -0,0 +1,79 @@
"""Minimal client for the OpenSandbox lifecycle API."""
from __future__ import annotations
from pathlib import Path
from urllib.parse import quote
from aiohttp import ClientResponseError, ClientSession
class OpenSandboxClient:
def __init__(
self,
*,
api_url: str,
api_key_file: Path | None = None,
) -> None:
self.api_url = api_url.rstrip("/")
self.api_key_file = api_key_file
self.session: ClientSession | None = None
async def __aenter__(self) -> OpenSandboxClient:
headers = {}
if self.api_key_file is not None:
headers["OPEN-SANDBOX-API-KEY"] = self.api_key_file.read_text().strip()
self.session = ClientSession(headers=headers, raise_for_status=True)
return self
async def __aexit__(self, *_: object) -> None:
if self.session is not None:
await self.session.close()
async def create(
self,
*,
pool: str,
timeout: int,
entrypoint: list[str],
env: dict[str, str],
metadata: dict[str, str],
) -> dict[str, object]:
response = await self._request(
"POST",
"/v1/sandboxes",
json={
"timeout": timeout,
"entrypoint": entrypoint,
"env": env,
"metadata": metadata,
"extensions": {"poolRef": pool},
},
)
return await response.json()
async def get(self, sandbox_id: str) -> dict[str, object] | None:
try:
response = await self._request(
"GET", f"/v1/sandboxes/{quote(sandbox_id, safe='')}"
)
except ClientResponseError as error:
if error.status == 404:
return None
raise
return await response.json()
async def delete(self, sandbox_id: str) -> None:
try:
response = await self._request(
"DELETE", f"/v1/sandboxes/{quote(sandbox_id, safe='')}"
)
response.release()
except ClientResponseError as error:
if error.status != 404:
raise
async def _request(self, method: str, path: str, **kwargs: object):
if self.session is None:
raise RuntimeError("OpenSandboxClient is not open")
return await self.session.request(method, f"{self.api_url}{path}", **kwargs)
@@ -0,0 +1,183 @@
"""Reconcile OpenSandbox Pod UIDs to narrowly scoped SPIRE entries."""
from __future__ import annotations
import asyncio
import hashlib
import logging
import os
from pathlib import Path
from aiohttp import ClientResponseError
from .sandbox_kubernetes import SandboxKubernetesClient, allocated_pod_name
LOG = logging.getLogger(__name__)
KUBERNETES_API = os.environ.get("KUBERNETES_API", "https://kubernetes.default.svc")
KUBERNETES_TOKEN_FILE = Path(
os.environ.get(
"KUBERNETES_TOKEN_FILE",
"/var/run/secrets/kubernetes.io/serviceaccount/token",
)
)
KUBERNETES_CA_FILE = Path(
os.environ.get(
"KUBERNETES_CA_FILE",
"/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
)
)
NAMESPACE = os.environ.get("OPENSANDBOX_NAMESPACE", "opensandbox")
SPIFFE_TRUST_DOMAIN = os.environ.get("SPIFFE_TRUST_DOMAIN", "ddupan.top")
SPIRE_CLUSTER_NAME = os.environ.get("SPIRE_CLUSTER_NAME", "sandbox-kata")
SPIRE_CLASS_NAME = os.environ.get("SPIRE_CLASS_NAME", "spire-mgmt-spire")
RUNNER_UID = int(os.environ.get("RUNNER_UID", "2000"))
RECONCILE_INTERVAL = int(os.environ.get("RECONCILE_INTERVAL", "2"))
def entry_name(sandbox_id: str) -> str:
suffix = hashlib.sha256(sandbox_id.encode()).hexdigest()[:12]
return f"gitea-ci-{suffix}"
def sandbox_name(document: dict[str, object]) -> str | None:
metadata = document.get("metadata")
value = metadata.get("name") if isinstance(metadata, dict) else None
return value if isinstance(value, str) and value else None
def is_runner_sandbox(document: dict[str, object]) -> bool:
metadata = document.get("metadata")
labels = metadata.get("labels") if isinstance(metadata, dict) else None
return isinstance(labels, dict) and labels.get("ci.ddupan.top/runner") == "true"
def task_environment(document: dict[str, object]) -> dict[str, str]:
spec = document.get("spec")
task = spec.get("taskTemplate") if isinstance(spec, dict) else None
task_spec = task.get("spec") if isinstance(task, dict) else None
process = task_spec.get("process") if isinstance(task_spec, dict) else None
values = process.get("env") if isinstance(process, dict) else None
result: dict[str, str] = {}
if not isinstance(values, list):
return result
for item in values:
if not isinstance(item, dict):
continue
name, value = item.get("name"), item.get("value")
if isinstance(name, str) and isinstance(value, str):
result[name] = value
return result
def identity_entry(
*, sandbox_id: str, pod_uid: str, spiffe_id: str
) -> dict[str, object]:
expected_prefix = f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/"
if not spiffe_id.startswith(expected_prefix):
raise ValueError("runner SPIFFE ID is outside the CI namespace")
return {
"apiVersion": "spire.spiffe.io/v1alpha1",
"kind": "ClusterStaticEntry",
"metadata": {
"name": entry_name(sandbox_id),
"labels": {
"app.kubernetes.io/name": "gitea-dynamic-runner",
"app.kubernetes.io/component": "opensandbox-identity",
"ci.ddupan.top/sandbox-id": sandbox_id,
},
},
"spec": {
"className": SPIRE_CLASS_NAME,
"parentID": (
f"spiffe://{SPIFFE_TRUST_DOMAIN}/spire/agent/k8s_psat/"
f"{SPIRE_CLUSTER_NAME}/pod/{pod_uid}"
),
"spiffeID": spiffe_id,
"selectors": [f"unix:uid:{RUNNER_UID}"],
},
}
async def reconcile(client: SandboxKubernetesClient) -> None:
sandboxes = await client.list_batchsandboxes()
live_names = {
name
for document in sandboxes
if is_runner_sandbox(document) and (name := sandbox_name(document))
}
entries = await client.list_entries()
existing = {
name
for document in entries
if (name := sandbox_name(document)) is not None
}
for document in sandboxes:
if not is_runner_sandbox(document):
continue
name = sandbox_name(document)
pod_name = allocated_pod_name(document)
spiffe_id = task_environment(document).get("CI_SPIFFE_ID")
if not name or not pod_name or not spiffe_id or entry_name(name) in existing:
continue
pod = await client.get_pod(pod_name)
metadata = pod.get("metadata") if isinstance(pod, dict) else None
pod_uid = metadata.get("uid") if isinstance(metadata, dict) else None
if not isinstance(pod_uid, str) or not pod_uid:
continue
try:
await client.create_entry(
identity_entry(
sandbox_id=name,
pod_uid=pod_uid,
spiffe_id=spiffe_id,
)
)
except ClientResponseError as error:
if error.status != 409:
raise
LOG.info(
"SPIRE entry ready sandbox=%s pod=%s pod_uid=%s spiffe_id=%s",
name,
pod_name,
pod_uid,
spiffe_id,
)
for document in entries:
metadata = document.get("metadata")
labels = metadata.get("labels") if isinstance(metadata, dict) else None
sandbox_id = labels.get("ci.ddupan.top/sandbox-id") if isinstance(labels, dict) else None
name = metadata.get("name") if isinstance(metadata, dict) else None
if (
isinstance(name, str)
and isinstance(sandbox_id, str)
and sandbox_id not in live_names
):
await client.delete_entry(name)
LOG.info("removed stale SPIRE entry=%s sandbox=%s", name, sandbox_id)
async def main() -> None:
async with SandboxKubernetesClient(
api_url=KUBERNETES_API,
token_file=KUBERNETES_TOKEN_FILE,
ca_file=KUBERNETES_CA_FILE,
namespace=NAMESPACE,
) as client:
while True:
try:
await reconcile(client)
except Exception:
LOG.exception("OpenSandbox identity reconcile failed")
await asyncio.sleep(RECONCILE_INTERVAL)
def cli() -> None:
logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
asyncio.run(main())
if __name__ == "__main__":
cli()
@@ -0,0 +1,194 @@
"""Direct OpenSandbox lifecycle scheduler used by the webhook controller."""
from __future__ import annotations
import asyncio
import logging
import os
import secrets
import uuid
from collections.abc import Callable
from .models import RunnerRequest
from .opensandbox import OpenSandboxClient
from .pod_worker import identity_path
LOG = logging.getLogger(__name__)
SANDBOX_TIMEOUT = int(os.environ.get("RUNNER_SANDBOX_TIMEOUT", str(4 * 60 * 60)))
SPIFFE_TRUST_DOMAIN = os.environ.get("SPIFFE_TRUST_DOMAIN", "ddupan.top")
TOKEN_BASE_URL = os.environ.get(
"RUNNER_TOKEN_BASE_URL", "http://192.168.10.127:8787/token"
).rstrip("/")
def sandbox_request(
request: RunnerRequest,
runner_name: str,
registration_token_url: str,
) -> dict[str, object]:
path = identity_path(request.repository, request.job_name)
return {
"pool": f"ci-{request.backend}",
"timeout": SANDBOX_TIMEOUT,
"entrypoint": ["/usr/local/libexec/gitea-opensandbox-runner"],
"env": {
"GITEA_RUNNER_NAME": runner_name,
"GITEA_RUNNER_LABELS": f"self-hosted:host,{request.backend}:host",
"GITEA_RUNNER_REGISTRATION_TOKEN_URL": registration_token_url,
"GITEA_RUNNER_EPHEMERAL": "1",
"GITEA_RUNNER_ONCE": "1",
"CONFIG_FILE": "/etc/gitea-runner/config.yaml",
"CI_SPIFFE_ID": f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/{path}",
"SPIFFE_ENDPOINT_SOCKET": (
"unix:///run/spire/agent-sockets/spire-agent.sock"
),
},
"metadata": {
"ci.ddupan.top/runner": "true",
"ci.ddupan.top/job-id": str(request.job_id),
"ci.ddupan.top/run-id": str(request.run_id),
"ci.ddupan.top/runner-name": runner_name,
},
}
class RegistrationTokens:
"""Single-use registration-token URLs; values never enter Sandbox CRs."""
def __init__(self, token: bytes) -> None:
self._token = token.strip()
self._values: dict[str, bytes] = {}
self._lock = asyncio.Lock()
async def issue(self) -> tuple[str, str]:
nonce = secrets.token_urlsafe(32)
async with self._lock:
self._values[nonce] = self._token
return nonce, f"{TOKEN_BASE_URL}/{nonce}"
async def consume(self, nonce: str) -> bytes | None:
async with self._lock:
return self._values.pop(nonce, None)
async def revoke(self, nonce: str) -> None:
async with self._lock:
self._values.pop(nonce, None)
class OpenSandboxScheduler:
def __init__(
self,
client: OpenSandboxClient,
tokens: RegistrationTokens,
*,
on_finished: Callable[[int], None] | None = None,
) -> None:
self.client = client
self.tokens = tokens
self.on_finished = on_finished
self.active: dict[int, asyncio.Task[None]] = {}
self.runners: dict[str, tuple[int, str, str]] = {}
async def create(self, request: RunnerRequest) -> str:
if request.job_id in self.active:
raise ValueError(f"job {request.job_id} already has an active sandbox")
runner_name = f"gitea-{request.backend}-{uuid.uuid4().hex[:12]}"
nonce, token_url = await self.tokens.issue()
try:
response = await self.client.create(
**sandbox_request(request, runner_name, token_url)
)
sandbox_id = response.get("id")
if not isinstance(sandbox_id, str) or not sandbox_id:
raise RuntimeError("OpenSandbox create response has no id")
except Exception:
await self.tokens.revoke(nonce)
raise
task = asyncio.create_task(
self._monitor(request, sandbox_id, nonce, runner_name),
name=f"opensandbox-{sandbox_id}",
)
task.add_done_callback(self._report)
self.active[request.job_id] = task
self.runners[runner_name] = (request.job_id, sandbox_id, nonce)
LOG.info(
"OpenSandbox runner created sandbox=%s runner=%s job_id=%d "
"repository=%s job_name=%r pool=ci-%s",
sandbox_id,
runner_name,
request.job_id,
request.repository,
request.job_name,
request.backend,
)
return sandbox_id
async def _monitor(
self,
request: RunnerRequest,
sandbox_id: str,
nonce: str,
runner_name: str,
) -> None:
try:
deadline = asyncio.get_running_loop().time() + SANDBOX_TIMEOUT
while asyncio.get_running_loop().time() < deadline:
sandbox = await self.client.get(sandbox_id)
if sandbox is None:
return
status = sandbox.get("status")
state = status.get("state") if isinstance(status, dict) else None
if state in {"Terminated", "Failed"}:
return
await asyncio.sleep(2)
raise asyncio.TimeoutError(f"OpenSandbox {sandbox_id} timed out")
finally:
await self._cleanup(request.job_id, sandbox_id, nonce, runner_name)
async def complete(self, runner_name: str) -> bool:
"""Delete the sandbox which actually ran a completed Gitea job."""
state = self.runners.get(runner_name)
if state is None:
return False
job_id, sandbox_id, nonce = state
task = self.active.get(job_id)
if task is not None:
task.cancel()
await asyncio.gather(task, return_exceptions=True)
await self._cleanup(job_id, sandbox_id, nonce, runner_name)
return True
async def _cleanup(
self,
job_id: int,
sandbox_id: str,
nonce: str,
runner_name: str,
) -> None:
"""Revoke and delete once, including cancellation-before-start races."""
if self.runners.pop(runner_name, None) is None:
return
await self.tokens.revoke(nonce)
try:
await self.client.delete(sandbox_id)
finally:
self.active.pop(job_id, None)
if self.on_finished is not None:
self.on_finished(job_id)
@staticmethod
def _report(task: asyncio.Task[None]) -> None:
if not task.cancelled() and (error := task.exception()) is not None:
LOG.error(
"OpenSandbox lifecycle task failed",
exc_info=(type(error), error, error.__traceback__),
)
async def close(self) -> None:
tasks = list(self.active.values())
for task in tasks:
task.cancel()
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
+76 -5
View File
@@ -4,12 +4,13 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import hashlib
import json import json
import logging import logging
import os import os
from pathlib import Path from pathlib import Path
import re
import ssl import ssl
from urllib.parse import quote
import uuid import uuid
import nats import nats
@@ -49,16 +50,26 @@ RUNNER_TOKEN_SECRET = os.environ.get("RUNNER_TOKEN_SECRET", "gitea-dynamic-runne
GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.ddupan.top") GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.ddupan.top")
CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4")) CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4"))
POD_TIMEOUT = int(os.environ.get("RUNNER_POD_TIMEOUT", str(4 * 60 * 60))) POD_TIMEOUT = int(os.environ.get("RUNNER_POD_TIMEOUT", str(4 * 60 * 60)))
SPIFFE_PATH_SEGMENT = re.compile(r"^[A-Za-z0-9._-]+$")
def identity_path(repository: str, job_name: str) -> str: def identity_path(repository: str, job_name: str) -> str:
parts = repository.split("/") parts = repository.split("/")
if len(parts) != 2 or not all(parts): if len(parts) != 2 or not all(parts):
raise ValueError("repository must be owner/name") raise ValueError("repository must be owner/name")
encoded = [quote(part, safe="-._~") for part in (*parts, job_name)] if not job_name.strip():
if not job_name.strip() or any(not part for part in encoded):
raise ValueError("identity components must not be empty") raise ValueError("identity components must not be empty")
return "/".join(encoded) return "/".join(safe_identity_segment(part) for part in (*parts, job_name))
def safe_identity_segment(value: str) -> str:
"""Map arbitrary Gitea names to stable SPIFFE Operator path segments."""
if SPIFFE_PATH_SEGMENT.fullmatch(value):
return value
slug = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-._")
slug = slug[:48].rstrip("-._") or "segment"
digest = hashlib.sha256(value.encode()).hexdigest()[:12]
return f"{slug}-{digest}"
def pod_manifest(request: RunnerRequest, pod_name: str) -> dict[str, object]: def pod_manifest(request: RunnerRequest, pod_name: str) -> dict[str, object]:
@@ -174,36 +185,90 @@ async def run_request(message: object, client: KubernetesClient) -> None:
return return
pod_name = f"gitea-pod-{uuid.uuid4().hex[:12]}" pod_name = f"gitea-pod-{uuid.uuid4().hex[:12]}"
LOG.info(
"runner request received job_id=%d run_id=%d repository=%s job_name=%r "
"backend=%s pod=%s",
request.job_id,
request.run_id,
request.repository,
request.job_name,
request.backend,
pod_name,
)
stop = asyncio.Event() stop = asyncio.Event()
pulse = asyncio.create_task(heartbeat(message, stop)) pulse = asyncio.create_task(heartbeat(message, stop))
try: try:
await client.create_pod(pod_manifest(request, pod_name)) await client.create_pod(pod_manifest(request, pod_name))
LOG.info(
"runner Pod created pod=%s job_id=%d run_id=%d image=%s",
pod_name,
request.job_id,
request.run_id,
RUNNER_IMAGE,
)
if await wait_for_pod(client, pod_name): if await wait_for_pod(client, pod_name):
await message.ack() await message.ack()
LOG.info(
"runner request acknowledged pod=%s job_id=%d result=succeeded",
pod_name,
request.job_id,
)
else: else:
LOG.error("runner Pod %s failed", pod_name) LOG.error(
"runner Pod failed pod=%s job_id=%d; request will be retried",
pod_name,
request.job_id,
)
await message.nak(delay=30) await message.nak(delay=30)
except Exception: except Exception:
LOG.exception(
"runner request failed pod=%s job_id=%d; request will be retried",
pod_name,
request.job_id,
)
await message.nak(delay=30) await message.nak(delay=30)
raise raise
finally: finally:
stop.set() stop.set()
await pulse await pulse
LOG.info("deleting runner Pod pod=%s job_id=%d", pod_name, request.job_id)
await client.delete_pod(pod_name) await client.delete_pod(pod_name)
async def bind_request(message: object, client: KubernetesClient) -> None: async def bind_request(message: object, client: KubernetesClient) -> None:
binding: IdentityBinding | None = None
try: try:
document = json.loads(message.data) document = json.loads(message.data)
binding = IdentityBinding(**document) binding = IdentityBinding(**document)
if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"): if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"):
raise ValueError("invalid Pod identity binding") raise ValueError("invalid Pod identity binding")
path = identity_path(binding.repository, binding.job_name) path = identity_path(binding.repository, binding.job_name)
LOG.info(
"identity binding received job_id=%d run_id=%d runner_name=%s "
"repository=%s job_name=%r identity_path=%s",
binding.job_id,
binding.run_id,
binding.runner_name,
binding.repository,
binding.job_name,
path,
)
await client.bind_identity(binding.runner_name, path) await client.bind_identity(binding.runner_name, path)
except ClientResponseError as error: except ClientResponseError as error:
if error.status == 404: if error.status == 404:
LOG.warning(
"identity binding Pod not found runner_name=%s job_id=%s; binding will be retried",
binding.runner_name if binding else None,
binding.job_id if binding else None,
)
await message.nak(delay=2) await message.nak(delay=2)
return return
LOG.exception(
"identity binding Kubernetes request failed runner_name=%s job_id=%s status=%d",
binding.runner_name if binding else None,
binding.job_id if binding else None,
error.status,
)
await message.nak(delay=30) await message.nak(delay=30)
raise raise
except (json.JSONDecodeError, TypeError, ValueError) as error: except (json.JSONDecodeError, TypeError, ValueError) as error:
@@ -211,6 +276,12 @@ async def bind_request(message: object, client: KubernetesClient) -> None:
await message.ack() await message.ack()
return return
await message.ack() await message.ack()
LOG.info(
"identity binding applied and acknowledged runner_name=%s job_id=%d identity_path=%s",
binding.runner_name,
binding.job_id,
path,
)
async def consume_requests(subscription: object, client: KubernetesClient) -> None: async def consume_requests(subscription: object, client: KubernetesClient) -> None:
@@ -0,0 +1,148 @@
"""Kubernetes resources that bind an OpenSandbox Kata guest to SPIRE."""
from __future__ import annotations
import json
from pathlib import Path
import ssl
from urllib.parse import quote
from aiohttp import ClientResponseError, ClientSession, TCPConnector
class SandboxKubernetesClient:
def __init__(
self,
*,
api_url: str,
token_file: Path,
ca_file: Path,
namespace: str,
) -> None:
self.api_url = api_url.rstrip("/")
self.token_file = token_file
self.ca_file = ca_file
self.namespace = namespace
self.session: ClientSession | None = None
async def __aenter__(self) -> SandboxKubernetesClient:
context = ssl.create_default_context(cafile=self.ca_file)
self.session = ClientSession(
connector=TCPConnector(ssl=context),
headers={
"Authorization": f"Bearer {self.token_file.read_text().strip()}"
},
raise_for_status=True,
)
return self
async def __aexit__(self, *_: object) -> None:
if self.session is not None:
await self.session.close()
async def get_batchsandbox(self, sandbox_id: str) -> dict[str, object] | None:
try:
response = await self._request(
"GET", f"{self._batchsandboxes_path()}/{quote(sandbox_id, safe='')}"
)
except ClientResponseError as error:
if error.status == 404:
return None
raise
return await response.json()
async def list_batchsandboxes(self) -> list[dict[str, object]]:
response = await self._request("GET", self._batchsandboxes_path())
document = await response.json()
items = document.get("items") if isinstance(document, dict) else None
return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else []
async def get_pod(self, name: str) -> dict[str, object] | None:
try:
response = await self._request(
"GET", f"{self._pods_path()}/{quote(name, safe='')}"
)
except ClientResponseError as error:
if error.status == 404:
return None
raise
return await response.json()
async def create_entry(self, manifest: dict[str, object]) -> dict[str, object]:
response = await self._request(
"POST", self._entries_path(), json=manifest
)
return await response.json()
async def list_entries(self) -> list[dict[str, object]]:
response = await self._request(
"GET",
f"{self._entries_path()}?labelSelector="
"app.kubernetes.io%2Fcomponent%3Dopensandbox-identity",
)
document = await response.json()
items = document.get("items") if isinstance(document, dict) else None
return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else []
async def get_entry(self, name: str) -> dict[str, object] | None:
try:
response = await self._request(
"GET", f"{self._entries_path()}/{quote(name, safe='')}"
)
except ClientResponseError as error:
if error.status == 404:
return None
raise
return await response.json()
async def delete_entry(self, name: str) -> None:
try:
response = await self._request(
"DELETE",
f"{self._entries_path()}/{quote(name, safe='')}",
json={"propagationPolicy": "Background"},
)
response.release()
except ClientResponseError as error:
if error.status != 404:
raise
async def _request(self, method: str, path: str, **kwargs: object):
if self.session is None:
raise RuntimeError("SandboxKubernetesClient is not open")
return await self.session.request(method, f"{self.api_url}{path}", **kwargs)
def _batchsandboxes_path(self) -> str:
return (
"/apis/sandbox.opensandbox.io/v1alpha1/namespaces/"
f"{quote(self.namespace, safe='')}/batchsandboxes"
)
def _pods_path(self) -> str:
return f"/api/v1/namespaces/{quote(self.namespace, safe='')}/pods"
@staticmethod
def _entries_path() -> str:
return "/apis/spire.spiffe.io/v1alpha1/clusterstaticentries"
def allocated_pod_name(batchsandbox: dict[str, object]) -> str | None:
metadata = batchsandbox.get("metadata")
if not isinstance(metadata, dict):
return None
annotations = metadata.get("annotations")
if not isinstance(annotations, dict):
return None
raw = annotations.get("sandbox.opensandbox.io/alloc-status")
if not isinstance(raw, str):
return None
try:
allocation = json.loads(raw)
except json.JSONDecodeError:
return None
if not isinstance(allocation, dict):
return None
pods = allocation.get("pods")
if not isinstance(pods, list) or len(pods) != 1 or not isinstance(pods[0], str):
return None
return pods[0]
+11
View File
@@ -31,6 +31,9 @@ REGISTRATION_TOKEN_FILE = Path(os.environ.get("REGISTRATION_TOKEN_FILE", "/etc/m
LAUNCHER = os.environ.get("LAUNCHER", "/usr/local/libexec/microvm-runner-launch") LAUNCHER = os.environ.get("LAUNCHER", "/usr/local/libexec/microvm-runner-launch")
TOKEN_LISTEN = os.environ.get("TOKEN_LISTEN", "172.30.0.1") TOKEN_LISTEN = os.environ.get("TOKEN_LISTEN", "172.30.0.1")
TOKEN_PORT = int(os.environ.get("TOKEN_PORT", "8787")) TOKEN_PORT = int(os.environ.get("TOKEN_PORT", "8787"))
GUEST_ASSETS = Path(
os.environ.get("GUEST_ASSETS", "/opt/gitea-dynamic-runner/guest-assets.tar.gz")
)
tokens: dict[str, bytes] = {} tokens: dict[str, bytes] = {}
token_lock = asyncio.Lock() token_lock = asyncio.Lock()
@@ -45,6 +48,13 @@ async def token(request: web.Request) -> web.Response:
return web.Response(body=value, headers={"Cache-Control": "no-store"}) return web.Response(body=value, headers={"Cache-Control": "no-store"})
async def guest_assets(_: web.Request) -> web.FileResponse:
return web.FileResponse(
GUEST_ASSETS,
headers={"Cache-Control": "public, immutable"},
)
async def heartbeat(message: object, stop: asyncio.Event) -> None: async def heartbeat(message: object, stop: asyncio.Event) -> None:
while True: while True:
try: try:
@@ -162,6 +172,7 @@ async def consume() -> None:
async def main() -> None: async def main() -> None:
app = web.Application() app = web.Application()
app.router.add_get("/token/{nonce}", token) app.router.add_get("/token/{nonce}", token)
app.router.add_get("/assets/guest-assets.tar.gz", guest_assets)
runner = web.AppRunner(app) runner = web.AppRunner(app)
await runner.setup() await runner.setup()
await web.TCPSite(runner, TOKEN_LISTEN, TOKEN_PORT).start() await web.TCPSite(runner, TOKEN_LISTEN, TOKEN_PORT).start()
+46 -2
View File
@@ -1,3 +1,4 @@
import asyncio
import hashlib import hashlib
import hmac import hmac
import json import json
@@ -31,8 +32,14 @@ def test_rejects_other_actions_labels_and_boolean_id():
completed = queued_job() completed = queued_job()
completed["action"] = "completed" completed["action"] = "completed"
assert controller.accepts(completed) == (False, None) assert controller.accepts(completed) == (False, None)
assert controller.accepts(queued_job(labels=["self-hosted", "other"])) == (False, None) assert controller.accepts(queued_job(labels=["self-hosted", "other"])) == (
assert controller.accepts(queued_job(labels=["self-hosted", "pod", "vm"])) == (False, None) False,
None,
)
assert controller.accepts(queued_job(labels=["self-hosted", "pod", "vm"])) == (
False,
None,
)
assert controller.accepts(queued_job(id=True)) == (False, None) assert controller.accepts(queued_job(id=True)) == (False, None)
@@ -95,3 +102,40 @@ def test_signature_accepts_gitea_and_prefixed_forms(tmp_path, monkeypatch):
assert controller.valid_signature(body, digest) assert controller.valid_signature(body, digest)
assert controller.valid_signature(body, f"sha256={digest}") assert controller.valid_signature(body, f"sha256={digest}")
assert not controller.valid_signature(body, "bad") assert not controller.valid_signature(body, "bad")
async def test_completed_lifecycle_cancellation_acks_message():
request = RunnerRequest.from_webhook(queued_job())
assert request is not None
class Message:
data = request.to_json()
acked = False
async def ack(self):
self.acked = True
async def nak(self, **kwargs):
raise AssertionError(f"unexpected NAK: {kwargs}")
async def in_progress(self):
pass
class Scheduler:
active = {}
async def create(self, runner_request):
async def completed():
self.active.pop(runner_request.job_id, None)
raise asyncio.CancelledError
self.active[runner_request.job_id] = asyncio.create_task(completed())
message = Message()
await controller.run_message(message, Scheduler())
assert message.acked is True
def test_opensandbox_consumer_is_vm_only():
assert controller.VM_CONSUMER_ENABLED is False
assert controller.VM_CAPACITY == 1
+53
View File
@@ -0,0 +1,53 @@
import json
from gitea_dynamic_runner.opensandbox_identity import (
identity_entry,
is_runner_sandbox,
task_environment,
)
from gitea_dynamic_runner.sandbox_kubernetes import allocated_pod_name
def batchsandbox():
return {
"metadata": {
"name": "sandbox-1",
"labels": {"ci.ddupan.top/runner": "true"},
"annotations": {
"sandbox.opensandbox.io/alloc-status": json.dumps(
{"pods": ["ci-vm-pod-1"]}
)
},
},
"spec": {
"taskTemplate": {
"spec": {
"process": {
"env": [
{
"name": "CI_SPIFFE_ID",
"value": "spiffe://ddupan.top/ci/org/repo/test",
}
]
}
}
}
},
}
def test_extracts_allocation_and_identity_environment():
document = batchsandbox()
assert is_runner_sandbox(document)
assert allocated_pod_name(document) == "ci-vm-pod-1"
assert task_environment(document)["CI_SPIFFE_ID"].endswith("/org/repo/test")
def test_entry_binds_exact_pod_agent_and_runner_uid():
entry = identity_entry(
sandbox_id="sandbox-1",
pod_uid="pod-uid",
spiffe_id="spiffe://ddupan.top/ci/org/repo/test",
)
assert entry["spec"]["parentID"].endswith("/sandbox-kata/pod/pod-uid")
assert entry["spec"]["selectors"] == ["unix:uid:2000"]
+101
View File
@@ -0,0 +1,101 @@
import asyncio
import pytest
from gitea_dynamic_runner.models import RunnerRequest
from gitea_dynamic_runner.opensandbox_worker import (
OpenSandboxScheduler,
RegistrationTokens,
sandbox_request,
)
def request(backend: str = "vm") -> RunnerRequest:
return RunnerRequest(
job_id=42,
run_id=7,
backend=backend,
repository="panxiao81/example",
job_name="publish image",
labels=("self-hosted", backend),
)
def test_sandbox_request_uses_pool_identity_and_one_time_token_url():
document = sandbox_request(
request(), "gitea-vm-abcd", "http://scheduler/token/nonce"
)
assert document["pool"] == "ci-vm"
assert document["entrypoint"] == ["/usr/local/libexec/gitea-opensandbox-runner"]
assert document["env"]["GITEA_RUNNER_LABELS"] == "self-hosted:host,vm:host"
assert document["env"]["GITEA_RUNNER_REGISTRATION_TOKEN_URL"].endswith("/nonce")
assert document["env"]["CI_SPIFFE_ID"] == (
"spiffe://ddupan.top/ci/panxiao81/example/publish-image-3e72cdc4a97e"
)
assert document["metadata"]["ci.ddupan.top/runner"] == "true"
async def test_registration_token_is_single_use(monkeypatch):
monkeypatch.setattr(
"gitea_dynamic_runner.opensandbox_worker.TOKEN_BASE_URL",
"http://scheduler/token",
)
tokens = RegistrationTokens(b"secret\n")
nonce, url = await tokens.issue()
assert url == f"http://scheduler/token/{nonce}"
assert await tokens.consume(nonce) == b"secret"
assert await tokens.consume(nonce) is None
class FakeOpenSandbox:
def __init__(self):
self.created = None
self.deleted = []
async def create(self, **document):
self.created = document
return {"id": "sandbox-1"}
async def get(self, sandbox_id):
return {"status": {"state": "Terminated"}}
async def delete(self, sandbox_id):
self.deleted.append(sandbox_id)
async def test_scheduler_creates_monitors_and_deletes(monkeypatch):
monkeypatch.setattr(
"gitea_dynamic_runner.opensandbox_worker.TOKEN_BASE_URL",
"http://scheduler/token",
)
client = FakeOpenSandbox()
scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
assert await scheduler.create(request()) == "sandbox-1"
await scheduler.active[42]
assert client.created["pool"] == "ci-vm"
assert client.deleted == ["sandbox-1"]
async def test_scheduler_rejects_duplicate_active_job():
client = FakeOpenSandbox()
scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
scheduler.active[42] = asyncio.Future()
with pytest.raises(ValueError, match="already"):
await scheduler.create(request())
async def test_scheduler_deletes_sandbox_for_completed_runner():
class RunningOpenSandbox(FakeOpenSandbox):
async def get(self, sandbox_id):
return {"status": {"state": "Running"}}
client = RunningOpenSandbox()
scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
await scheduler.create(request())
runner_name = next(iter(scheduler.runners))
assert await scheduler.complete(runner_name) is True
assert client.deleted == ["sandbox-1"]
assert scheduler.active == {}
assert scheduler.runners == {}
assert await scheduler.complete(runner_name) is False
+16 -2
View File
@@ -16,8 +16,22 @@ def request() -> RunnerRequest:
def test_identity_path_is_meaningful_and_uri_safe(): def test_identity_path_is_meaningful_and_uri_safe():
assert pod_worker.identity_path("panxiao81/example", "publish image") == ( path = pod_worker.identity_path("panxiao81/example", "publish image")
"panxiao81/example/publish%20image" assert path.startswith("panxiao81/example/publish-image-")
assert "%" not in path
assert all(pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part) for part in path.split("/"))
assert pod_worker.identity_path("panxiao81/example", "lint") == "panxiao81/example/lint"
assert path == pod_worker.identity_path("panxiao81/example", "publish image")
assert path != pod_worker.identity_path("panxiao81/example", "publish-image")
real_path = pod_worker.identity_path(
"panxiao81/postgresql-tenant-operator", "Run on Ubuntu"
)
assert real_path.startswith(
"panxiao81/postgresql-tenant-operator/Run-on-Ubuntu-"
)
assert all(
pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part)
for part in real_path.split("/")
) )
with pytest.raises(ValueError): with pytest.raises(ValueError):
pod_worker.identity_path("invalid", "test") pod_worker.identity_path("invalid", "test")