Compare commits
40
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
de98a9952e
|
||
|
|
3e69d102dd
|
||
|
|
0387b2592a | ||
|
|
057731ae62
|
||
|
|
529daa4c75 | ||
|
|
38946e8def
|
||
|
|
893323a89b | ||
|
|
6f149c85e2
|
||
|
|
852afbf02e | ||
|
|
6a29a10244 | ||
|
|
94a962459c | ||
|
|
d03927d73c | ||
|
|
68bb02b20a
|
||
|
|
c4aa6ee0af
|
||
|
|
401c1f9a00
|
||
|
|
6dac9897fd
|
||
|
|
787614667c
|
||
|
|
072a5bad77
|
||
|
|
b25b1fbf62
|
||
|
|
8578bee895
|
||
|
|
02e0b698c5
|
||
|
|
ac85d5fe58
|
||
|
|
bb73a48695
|
||
|
|
70a8aa68c6
|
||
|
|
ce4c9f13b5
|
||
|
|
3641e6ffb3
|
||
|
|
99ef62bfa0
|
||
|
|
33cbd23861 | ||
|
|
0dfdbd46c8
|
||
|
|
21aabe162f
|
||
|
|
cbfcb10143 | ||
|
|
3ec528623a
|
||
|
|
11669ea81a | ||
|
|
83eb87bcec
|
||
|
|
75798f4c89 | ||
|
|
4277bd907a
|
||
|
|
70f0380a3d
|
||
|
|
3fb62289fa | ||
|
|
bfe9a1be58
|
||
|
|
4d6ae62abe |
@@ -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)" = 0
|
||||
@@ -0,0 +1,139 @@
|
||||
---
|
||||
name: publish images
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
paths:
|
||||
- '.gitea/workflows/publish-images.yml'
|
||||
- 'config/**'
|
||||
- 'container/**'
|
||||
- 'scripts/**'
|
||||
- 'src/**'
|
||||
- 'pyproject.toml'
|
||||
- 'README.md'
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
publish-images:
|
||||
name: publish-images
|
||||
runs-on: self-hosted
|
||||
timeout-minutes: 45
|
||||
permissions:
|
||||
contents: read
|
||||
container:
|
||||
image: docker.io/gitea/runner-images:ubuntu-latest@sha256:fd911d7417bfbf0f454530e447da95b58001e1df41bbc5e1a8dd35d432575aae
|
||||
volumes:
|
||||
- /run/spire/agent-sockets:/run/spire/agent-sockets:ro
|
||||
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
|
||||
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: Fetch pinned SPIRE CLI
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
archive=/tmp/spire.tar.gz
|
||||
curl --fail --location --silent --show-error \
|
||||
--output "$archive" \
|
||||
https://github.com/spiffe/spire/releases/download/v1.15.3/spire-1.15.3-linux-amd64-musl.tar.gz
|
||||
printf '%s %s\n' \
|
||||
ca1a4d1155317bdd2afc7f36663828a10410c7c840e54725b90b4064b0a301c7 \
|
||||
"$archive" | sha256sum --check --status
|
||||
tar -xzf "$archive" -C /tmp spire-1.15.3/bin/spire-agent
|
||||
|
||||
- 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"
|
||||
|
||||
/tmp/spire-1.15.3/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
|
||||
@@ -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
|
||||
@@ -0,0 +1,5 @@
|
||||
---
|
||||
self-hosted-runner:
|
||||
labels:
|
||||
- pod
|
||||
- vm
|
||||
@@ -1,24 +1,43 @@
|
||||
# Gitea microVM runner
|
||||
# Gitea dynamic runner
|
||||
|
||||
为 Gitea Actions 按需启动 Cloud Hypervisor microVM。适合 kind、嵌套容器和其他不应
|
||||
在常驻 Kubernetes runner 中执行的 CI 工作负载。
|
||||
为 Gitea Actions 按需创建一次性执行环境。对 workflow 提供两种稳定的 runner
|
||||
接口:
|
||||
|
||||
```yaml
|
||||
runs-on: [self-hosted, pod]
|
||||
```
|
||||
|
||||
```yaml
|
||||
runs-on: [self-hosted, vm]
|
||||
```
|
||||
|
||||
`pod` 使用动态 Kubernetes Pod,`vm` 使用动态 Cloud Hypervisor microVM。每个环境
|
||||
只执行一个 job,并在 job 结束后连同本地状态一起销毁。完整的设计约束见
|
||||
[`docs/design-principles.md`](docs/design-principles.md)。
|
||||
|
||||
组件:
|
||||
|
||||
- `controller`:接收 Gitea `workflow_job` webhook,将指定 label 的 queued job
|
||||
发布到 NATS JetStream。
|
||||
- `worker`:在虚拟化宿主机领取任务,限制本机并发,并启动一次性 microVM。
|
||||
- `microvm-runner-launch`:为每个任务创建 COW disk、NoCloud seed 和 TAP,运行
|
||||
- `worker`:领取任务、限制并发,并通过 Pod 或 microVM backend 创建一次性环境。
|
||||
- `microvm-runner-launch`:为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP,运行
|
||||
Cloud Hypervisor,退出后完整清理。
|
||||
- `guest-runner`:在 guest 中领取一次性 runner registration token,注册 ephemeral
|
||||
runner,执行一个 job 后关机。
|
||||
- `jwt-broker`:运行在 Kubernetes runner 外层 Pod 中,以可被 SPIRE attestation
|
||||
的 PID 获取固定 `aud=zot` JWT-SVID;DinD job 通过受限 HTTP endpoint 获取短期
|
||||
token。broker 不记录响应、不缓存 token,也不接受调用方指定 audience。
|
||||
- `pod-worker`:在 Kubernetes 中创建一次性 privileged Pod;Pod 内的 workflow 使用
|
||||
host executor,Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在
|
||||
支持原生 job hooks 的 3.x 版本,在 workflow 第一步前等待实际任务对应的 SVID。
|
||||
- `jwt-broker`:早期共享 Kubernetes runner 的过渡实验;目标架构不部署它,每个
|
||||
动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
|
||||
|
||||
消息流使用一个 `WorkQueuePolicy` stream。相同 runner label 的所有 worker 共享同一
|
||||
durable consumer;扩容只需要增加 worker 或提高单机 capacity。
|
||||
|
||||
当前 webhook → NATS 流程是用于尽快验证 Pod/VM 生命周期的 bootstrap 实现,不是
|
||||
长期调度接口。长期目标是让 controller 作为兼容 Gitea Runner 协议的调度器直接注册、
|
||||
声明 labels、领取 task,并把已领取 task 交给 Pod/VM executor;路线与迁移边界见
|
||||
[`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
|
||||
|
||||
## 开发
|
||||
|
||||
```bash
|
||||
@@ -32,9 +51,16 @@ pytest
|
||||
|
||||
- NATS 密码、webhook secret 和 Gitea registration token 只从文件读取。
|
||||
- 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。
|
||||
- launcher 只接受 UUID instance ID 和 URL-safe nonce,所有临时文件都位于独立目录。
|
||||
- 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 和容量配置保留在
|
||||
`panxiao81/homelab-infra`。
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
runner:
|
||||
capacity: 1
|
||||
timeout: 3h
|
||||
shutdown_timeout: 1m
|
||||
|
||||
host:
|
||||
workdir_parent: /workspace
|
||||
|
||||
container:
|
||||
require_docker: false
|
||||
valid_volumes: []
|
||||
@@ -0,0 +1,13 @@
|
||||
runner:
|
||||
capacity: 1
|
||||
timeout: 3h
|
||||
shutdown_timeout: 1m
|
||||
hooks:
|
||||
job_started: /usr/local/libexec/gitea-job-started
|
||||
|
||||
host:
|
||||
workdir_parent: /workspace
|
||||
|
||||
container:
|
||||
require_docker: false
|
||||
valid_volumes: []
|
||||
@@ -7,9 +7,11 @@ COPY src ./src
|
||||
RUN python -m venv /venv && /venv/bin/pip install --no-cache-dir .
|
||||
|
||||
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 \
|
||||
&& adduser -S -D -H -u 65532 -G runner runner \
|
||||
&& install -d -o 65532 -g 65532 /var/run/secrets/kubernetes.io/serviceaccount
|
||||
COPY --from=build /venv /venv
|
||||
COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
|
||||
USER 65532:65532
|
||||
EXPOSE 8787
|
||||
ENTRYPOINT ["/venv/bin/gitea-microvm-controller"]
|
||||
ENTRYPOINT ["/venv/bin/gitea-dynamic-runner-controller"]
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
|
||||
|
||||
FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242 AS 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
|
||||
|
||||
COPY --from=runner /usr/local/bin/gitea-runner /usr/local/bin/gitea-runner
|
||||
COPY --from=runner /usr/local/bin/run.sh /usr/local/bin/run.sh
|
||||
COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
|
||||
COPY config/runner.yaml /etc/gitea-runner/config.yaml
|
||||
COPY --chmod=0755 scripts/gitea-job-started /usr/local/libexec/gitea-job-started
|
||||
|
||||
VOLUME ["/data"]
|
||||
WORKDIR /
|
||||
ENTRYPOINT ["/usr/local/bin/run.sh"]
|
||||
@@ -0,0 +1,88 @@
|
||||
# 动态 Runner 设计原则
|
||||
|
||||
## 对 workflow 的接口
|
||||
|
||||
Runner 只向 workflow 暴露两个执行环境:
|
||||
|
||||
```yaml
|
||||
runs-on: [self-hosted, pod]
|
||||
```
|
||||
|
||||
```yaml
|
||||
runs-on: [self-hosted, vm]
|
||||
```
|
||||
|
||||
- `self-hosted` 是固定前缀。
|
||||
- `pod` 表示一次性 Kubernetes Pod,承担常规 CI、镜像构建和 kind 等任务。
|
||||
- `vm` 表示一次性 microVM,承担需要独立内核、KVM、systemd 或更强隔离的任务。
|
||||
|
||||
执行后端是基础设施选择,不是权限角色。workflow 不需要额外声明由 controller
|
||||
维护的 role 或权限 label。
|
||||
|
||||
## 一个 job,一个环境
|
||||
|
||||
Controller 根据 Gitea `workflow_job` webhook 创建执行环境。每个 Pod 或 VM 注册一个
|
||||
ephemeral runner,只执行一个 job;任务结束后注销 runner,并删除计算环境及其全部
|
||||
本地状态。
|
||||
|
||||
`job_id` 仅用于消息去重、状态追踪、实例关联和失败清理,不进入 workload 身份,也
|
||||
不参与资源授权。
|
||||
|
||||
Gitea 不保证由某次 `queued` webhook 创建的 runner 一定领取该 webhook 对应的 job。
|
||||
因此创建环境时只赋予无业务权限的启动身份。runner 实际领取任务后,controller 根据
|
||||
`in_progress` webhook 返回的 `runner_name` 和真实 job 名称绑定业务身份;环境中的
|
||||
job-start hook 必须等目标 SVID 可用后才放行 workflow 的第一步。不能依据 queued
|
||||
事件提前赋予任务权限。
|
||||
|
||||
## 环境只提供运行边界
|
||||
|
||||
基础镜像只提供启动 runner 和执行 workflow 所需的最小环境。Docker、BuildKit、
|
||||
kind 等工具由 pipeline 按需安装和启动,而不是由 controller 预制成常驻服务。
|
||||
|
||||
例如 Pod job 可以在 Pod 内启动仅供本次任务使用的 Docker daemon。该 daemon 及其
|
||||
镜像、容器和缓存属于当前 job 的临时状态,随 Pod 一起销毁。Docker 创建的容器不是
|
||||
独立的身份边界;需要访问凭据的操作由 Pod 中的 workflow 进程完成,并通过环境变量
|
||||
或标准输入把短期凭据交给具体工具。
|
||||
|
||||
## Workload 身份
|
||||
|
||||
动态 Pod 和 VM 都直接拥有自己的 SPIFFE 身份,不继承常驻 runner 的共享身份:
|
||||
|
||||
- Pod 通过 Kubernetes workload attestation 取得身份。
|
||||
- VM 通过 VM 内的 SPIRE Agent 取得身份。
|
||||
|
||||
SPIFFE ID 由具有业务意义且稳定的 workflow 上下文派生:
|
||||
|
||||
```text
|
||||
spiffe://ddupan.top/ci/<owner>/<repository>/<job-name>
|
||||
```
|
||||
|
||||
同一种任务在不同运行中使用相同的逻辑 SPIFFE ID;每次运行取得独立、短期的 SVID。
|
||||
Pod 与 VM 是可替换的执行实现,因此默认不写入 SPIFFE ID。
|
||||
|
||||
job 名称必须经过确定性的路径规范化。规范化结果必须保留仓库边界,并在发生冲突时
|
||||
拒绝创建环境,不能静默地让两个任务共享身份。同一仓库内需要不同权限的任务应使用
|
||||
不同的 job 名称;workflow 文件只是编排载体,不进入权限身份。
|
||||
|
||||
## Self-service 与授权边界
|
||||
|
||||
新增 workflow 或 job 时,controller 自动为它派生身份,不维护第二份任务或角色
|
||||
allowlist。能够修改仓库 CI 的主体本来就能修改该仓库已有任务,因此 controller 的
|
||||
重复审批不能形成额外的安全边界,只会破坏 self-service。
|
||||
|
||||
身份不等于权限。新任务可以立即取得自己的 SPIFFE ID,但默认不会因此获得 Zot、
|
||||
OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意义的 job 身份
|
||||
配置授权策略。
|
||||
|
||||
## 非目标设计
|
||||
|
||||
目标架构不依赖以下机制:
|
||||
|
||||
- 多个 job 共享的常驻 Docker daemon。
|
||||
- 常驻 runner Pod 的共享 SPIFFE 身份。
|
||||
- 为嵌套 CI 容器转发共享身份的 JWT broker。
|
||||
- 将 Gitea 数字 job ID 编入 SPIFFE ID。
|
||||
- controller 维护的仓库任务权限 allowlist。
|
||||
|
||||
仓库中的 `jwt-broker` 是早期方案的实验实现,在 Pod/VM 动态执行环境完成迁移后不应
|
||||
部署。
|
||||
@@ -0,0 +1,59 @@
|
||||
# Gitea Runner 协议调度器路线
|
||||
|
||||
## 目标
|
||||
|
||||
长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner
|
||||
协议客户端注册,并声明 `self-hosted`、`pod` 和 `vm` labels;它只在后端存在可用容量
|
||||
时领取 task,然后将该 task 交给一个一次性 Pod 或 microVM 执行。
|
||||
|
||||
```text
|
||||
Gitea RunnerService
|
||||
│ Register / Declare / FetchTask
|
||||
▼
|
||||
dynamic-runner scheduler
|
||||
│ 已领取的 task + lease
|
||||
├── Pod executor
|
||||
└── microVM executor
|
||||
│ logs / state / result
|
||||
└──────────────────────► Gitea
|
||||
```
|
||||
|
||||
这与“收到 webhook 后临时注册另一个 act_runner”不同。`FetchTask` 已经完成任务分配,
|
||||
不能再期待 Gitea 把同一个 task 分配给随后启动的 runner。协议调度器必须让 executor
|
||||
执行已经领取的 task,并继续完成日志、状态、心跳、取消和最终结果上报。
|
||||
|
||||
## 设计约束
|
||||
|
||||
- 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。
|
||||
- scheduler 在没有对应 backend 容量时不领取 task,避免本地形成不可控积压。
|
||||
- 每个 executor 只执行一个 task,完成后销毁。
|
||||
- SPIFFE 身份从实际领取的 task 的 repository 和 job name 派生,不需要 queued 与
|
||||
in-progress webhook 的二阶段关联。
|
||||
- scheduler 的 runner registration credential 不进入 executor;executor 只得到执行
|
||||
当前 task 所需的短期 lease/capability。
|
||||
- task ACK、心跳和结果必须能够跨 scheduler 重启恢复;NATS 可以继续作为内部 handoff,
|
||||
但不是 Gitea 任务事实来源。
|
||||
- Pod 与 VM 共享 task/executor 协议,只有环境创建和销毁实现不同。
|
||||
|
||||
## 实现顺序
|
||||
|
||||
1. 固定当前 Gitea 版本所使用的 RunnerService protobuf 与 act_runner 版本,记录兼容
|
||||
范围并建立协议契约测试。
|
||||
2. 实现只注册、Declare labels 和容量感知 FetchTask 的 scheduler spike,暂不执行
|
||||
task。
|
||||
3. 从 act_runner 提取或复用 task 执行与日志上报能力,定义 scheduler 到 executor 的
|
||||
单任务协议。
|
||||
4. 首先接入 Pod executor,验证成功、失败、取消、超时和 scheduler 重启。
|
||||
5. 接入 microVM executor,并复用同一 task 协议和身份派生逻辑。
|
||||
6. 双轨运行并验证后,移除 webhook receiver、临时 runner 注册和 identity binding
|
||||
subject。
|
||||
|
||||
## Bootstrap 实现的退出条件
|
||||
|
||||
只有同时满足以下条件才能删除 webhook 路径:
|
||||
|
||||
- scheduler 能通过 RunnerService 稳定领取并执行 Pod/VM task;
|
||||
- Gitea UI 中的实时日志、取消、超时和结论与官方 runner 行为一致;
|
||||
- scheduler 重启不会丢失已领取 task,也不会重复执行;
|
||||
- SPIFFE 身份只来自实际领取 task;
|
||||
- 同一套 workflow 无需修改 `runs-on` 即可从 bootstrap 迁移。
|
||||
+7
-6
@@ -3,9 +3,9 @@ requires = ["setuptools>=75"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "gitea-microvm-runner"
|
||||
version = "0.2.0"
|
||||
description = "On-demand Cloud Hypervisor runners for Gitea Actions"
|
||||
name = "gitea-dynamic-runner"
|
||||
version = "0.3.0"
|
||||
description = "On-demand Pod and microVM execution environments for Gitea Actions"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = ["aiohttp==3.12.15", "nats-py==2.11.0"]
|
||||
|
||||
@@ -13,9 +13,10 @@ dependencies = ["aiohttp==3.12.15", "nats-py==2.11.0"]
|
||||
test = ["pytest==8.4.2", "pytest-asyncio==1.2.0"]
|
||||
|
||||
[project.scripts]
|
||||
gitea-microvm-controller = "gitea_microvm_runner.controller:main"
|
||||
gitea-microvm-worker = "gitea_microvm_runner.worker:cli"
|
||||
gitea-spire-jwt-broker = "gitea_microvm_runner.jwt_broker:main"
|
||||
gitea-dynamic-runner-controller = "gitea_dynamic_runner.controller:main"
|
||||
gitea-dynamic-runner-pod-worker = "gitea_dynamic_runner.pod_worker:cli"
|
||||
gitea-dynamic-runner-vm-worker = "gitea_dynamic_runner.worker:cli"
|
||||
gitea-spire-jwt-broker = "gitea_dynamic_runner.jwt_broker:main"
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
asyncio_mode = "auto"
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
#!/bin/sh
|
||||
set -eu
|
||||
|
||||
socket=${SPIRE_AGENT_SOCKET:-/run/spire/agent-sockets/spire-agent.sock}
|
||||
wait_seconds=${SPIFFE_IDENTITY_WAIT_SECONDS:-60}
|
||||
deadline=$(( $(date +%s) + wait_seconds ))
|
||||
|
||||
while [ "$(date +%s)" -lt "$deadline" ]; do
|
||||
if timeout 2 /opt/spire/bin/spire-agent api fetch jwt \
|
||||
-audience ci-job-ready \
|
||||
-socketPath "$socket" \
|
||||
>/dev/null 2>&1; then
|
||||
exit 0
|
||||
fi
|
||||
sleep 1
|
||||
done
|
||||
|
||||
echo "timed out waiting for the job SPIFFE identity" >&2
|
||||
exit 1
|
||||
@@ -4,8 +4,10 @@ set -eu
|
||||
token_url=${1:?token URL is required}
|
||||
instance=${2:?Gitea instance is required}
|
||||
runner_name=${3:?runner name is required}
|
||||
runner_label=${4:?runner label is required}
|
||||
runner_labels=${4:?runner labels are required}
|
||||
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() {
|
||||
rm -f -- "$token_file"
|
||||
@@ -22,7 +24,8 @@ gitea-runner register \
|
||||
--ephemeral \
|
||||
--instance "$instance" \
|
||||
--name "$runner_name" \
|
||||
--labels "$runner_label:host" \
|
||||
--labels "$runner_labels" \
|
||||
--token-file "$token_file"
|
||||
rm -f -- "$token_file"
|
||||
gitea-runner daemon
|
||||
gitea-runner daemon --config "$config_file" --once
|
||||
echo gitea-runner-job-complete >/dev/ttyS0
|
||||
|
||||
@@ -21,7 +21,7 @@ vm_timeout=${RUNNER_VM_TIMEOUT:-3h}
|
||||
cpus=${RUNNER_VM_CPUS:-4}
|
||||
memory=${RUNNER_VM_MEMORY:-3G}
|
||||
gitea_instance=${GITEA_INSTANCE:-https://git.ddupan.top}
|
||||
runner_label=${RUNNER_LABEL:-kind-microvm}
|
||||
runner_labels=${RUNNER_LABELS:-self-hosted:host,vm:host}
|
||||
token_url=${RUNNER_TOKEN_URL:-http://172.30.0.1:8787}
|
||||
|
||||
vm_dir="$state_root/instances/$instance_id"
|
||||
@@ -29,8 +29,13 @@ tap="mvr${instance_id%%-*}"
|
||||
overlay="$vm_dir/root.qcow2"
|
||||
seed="$vm_dir/seed.img"
|
||||
serial="$vm_dir/serial.log"
|
||||
log_dir="$state_root/logs"
|
||||
|
||||
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
|
||||
rm -rf -- "$vm_dir"
|
||||
}
|
||||
@@ -39,7 +44,10 @@ trap cleanup EXIT INT TERM
|
||||
test -r "$base_image"
|
||||
test -r "$firmware"
|
||||
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
|
||||
instance-id: $instance_id
|
||||
@@ -48,7 +56,8 @@ EOF
|
||||
cat >"$vm_dir/user-data" <<EOF
|
||||
#cloud-config
|
||||
runcmd:
|
||||
- [ /usr/local/libexec/gitea-microvm-guest-runner, "$token_url/token/$nonce", "$gitea_instance", "gitea-${instance_id%%-*}", "$runner_label" ]
|
||||
- [ 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
|
||||
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" up
|
||||
|
||||
timeout --signal=TERM --kill-after=30s "$vm_timeout" "$cloud_hypervisor" \
|
||||
if timeout --signal=TERM --kill-after=30s "$vm_timeout" "$cloud_hypervisor" \
|
||||
--firmware "$firmware" \
|
||||
--cpus "boot=$cpus" \
|
||||
--memory "size=$memory" \
|
||||
--disk "path=$overlay" \
|
||||
--disk "path=$seed,readonly=on" \
|
||||
--memory "size=$memory,shared=on" \
|
||||
--disk "path=$overlay,image_type=qcow2,direct=on,sparse=off" \
|
||||
--disk "path=$seed,readonly=on,image_type=raw,direct=on,sparse=off" \
|
||||
--net "tap=$tap,mac=$mac" \
|
||||
--serial "file=$serial" \
|
||||
--console off
|
||||
--console off; then
|
||||
grep -Fq gitea-runner-job-complete "$serial"
|
||||
else
|
||||
exit $?
|
||||
fi
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Dynamic Pod and microVM execution environments for Gitea Actions."""
|
||||
@@ -4,6 +4,7 @@
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import ssl
|
||||
from pathlib import Path
|
||||
@@ -13,8 +14,11 @@ from aiohttp import web
|
||||
from nats.js.api import DiscardPolicy, RetentionPolicy, StorageType, StreamConfig
|
||||
from nats.js.errors import NotFoundError
|
||||
|
||||
LABEL = os.environ.get("RUNNER_LABEL", "kind-microvm")
|
||||
SUBJECT = os.environ.get("NATS_SUBJECT", f"ci.runner.{LABEL}")
|
||||
from .models import IdentityBinding, RunnerRequest
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
|
||||
STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
|
||||
NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
|
||||
NATS_USER = os.environ.get("NATS_USER", "ci-producer")
|
||||
@@ -24,15 +28,9 @@ WEBHOOK_SECRET_FILE = Path(os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/g
|
||||
|
||||
|
||||
def accepts(payload: object) -> tuple[bool, str | None]:
|
||||
if not isinstance(payload, dict) or payload.get("action") != "queued":
|
||||
return False, None
|
||||
job = payload.get("workflow_job")
|
||||
if not isinstance(job, dict) or LABEL not in job.get("labels", []):
|
||||
return False, None
|
||||
job_id = job.get("id")
|
||||
if not isinstance(job_id, int) or isinstance(job_id, bool):
|
||||
return False, None
|
||||
return True, str(job_id)
|
||||
"""Compatibility helper for callers that only need acceptance and identity."""
|
||||
request = RunnerRequest.from_webhook(payload)
|
||||
return (request is not None, str(request.job_id) if request else None)
|
||||
|
||||
|
||||
def valid_signature(body: bytes, signature: str) -> bool:
|
||||
@@ -43,7 +41,7 @@ def valid_signature(body: bytes, signature: str) -> bool:
|
||||
async def ensure_stream(js: object) -> None:
|
||||
config = StreamConfig(
|
||||
name=STREAM,
|
||||
subjects=["ci.runner.*"],
|
||||
subjects=[f"{SUBJECT_PREFIX}.>"],
|
||||
retention=RetentionPolicy.WORK_QUEUE,
|
||||
storage=StorageType.FILE,
|
||||
discard=DiscardPolicy.OLD,
|
||||
@@ -68,16 +66,69 @@ async def webhook(request: web.Request) -> web.Response:
|
||||
payload = json.loads(body)
|
||||
except json.JSONDecodeError as error:
|
||||
raise web.HTTPBadRequest(text="invalid JSON\n") from error
|
||||
accepted, job_id = accepts(payload)
|
||||
if not accepted:
|
||||
return web.Response(status=204)
|
||||
context = _webhook_context(payload)
|
||||
LOG.info("workflow_job webhook received %s", context)
|
||||
|
||||
runner_request = RunnerRequest.from_webhook(payload)
|
||||
if runner_request is not None:
|
||||
subject = f"{SUBJECT_PREFIX}.{runner_request.backend}"
|
||||
await request.app["js"].publish(
|
||||
SUBJECT,
|
||||
body,
|
||||
headers={"Nats-Msg-Id": f"gitea-workflow-job-{job_id}"},
|
||||
subject,
|
||||
runner_request.to_json(),
|
||||
headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
|
||||
)
|
||||
LOG.info(
|
||||
"runner request published subject=%s job_id=%d run_id=%d backend=%s "
|
||||
"repository=%s job_name=%r labels=%s",
|
||||
subject,
|
||||
runner_request.job_id,
|
||||
runner_request.run_id,
|
||||
runner_request.backend,
|
||||
runner_request.repository,
|
||||
runner_request.job_name,
|
||||
runner_request.labels,
|
||||
)
|
||||
return web.Response(status=202, text="queued\n")
|
||||
|
||||
binding = IdentityBinding.from_webhook(payload)
|
||||
if binding is not None:
|
||||
subject = f"{SUBJECT_PREFIX}.{binding.backend}.binding"
|
||||
await request.app["js"].publish(
|
||||
subject,
|
||||
binding.to_json(),
|
||||
headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"},
|
||||
)
|
||||
LOG.info(
|
||||
"identity binding published subject=%s job_id=%d run_id=%d backend=%s "
|
||||
"runner_name=%s repository=%s job_name=%r",
|
||||
subject,
|
||||
binding.job_id,
|
||||
binding.run_id,
|
||||
binding.backend,
|
||||
binding.runner_name,
|
||||
binding.repository,
|
||||
binding.job_name,
|
||||
)
|
||||
return web.Response(status=202, text="binding queued\n")
|
||||
|
||||
LOG.warning("workflow_job webhook ignored %s", context)
|
||||
return web.Response(status=204)
|
||||
|
||||
|
||||
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:
|
||||
connected = request.app["nc"].is_connected
|
||||
@@ -109,6 +160,7 @@ def create_app() -> web.Application:
|
||||
|
||||
|
||||
def main() -> None:
|
||||
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")))
|
||||
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
"""Small in-cluster Kubernetes API client used by the Pod backend."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
|
||||
from aiohttp import ClientResponseError, ClientSession, TCPConnector
|
||||
import ssl
|
||||
|
||||
|
||||
class KubernetesClient:
|
||||
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) -> KubernetesClient:
|
||||
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 create_pod(self, manifest: dict[str, object]) -> dict[str, object]:
|
||||
response = await self._request("POST", self._pods_path(), json=manifest)
|
||||
return await response.json()
|
||||
|
||||
async def get_pod(self, name: str) -> dict[str, object] | None:
|
||||
try:
|
||||
response = await self._request("GET", f"{self._pods_path()}/{quote(name)}")
|
||||
except ClientResponseError as error:
|
||||
if error.status == 404:
|
||||
return None
|
||||
raise
|
||||
return await response.json()
|
||||
|
||||
async def bind_identity(self, name: str, identity_path: str) -> None:
|
||||
patch = {
|
||||
"metadata": {
|
||||
"labels": {"ci.ddupan.top/identity-bound": "true"},
|
||||
"annotations": {"ci.ddupan.top/spiffe-path": identity_path},
|
||||
}
|
||||
}
|
||||
response = await self._request(
|
||||
"PATCH",
|
||||
f"{self._pods_path()}/{quote(name)}",
|
||||
data=json.dumps(patch),
|
||||
headers={"Content-Type": "application/merge-patch+json"},
|
||||
)
|
||||
response.release()
|
||||
|
||||
async def delete_pod(self, name: str) -> None:
|
||||
try:
|
||||
response = await self._request(
|
||||
"DELETE",
|
||||
f"{self._pods_path()}/{quote(name)}",
|
||||
json={"gracePeriodSeconds": 30, "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("KubernetesClient is not open")
|
||||
return await self.session.request(method, f"{self.api_url}{path}", **kwargs)
|
||||
|
||||
def _pods_path(self) -> str:
|
||||
return f"/api/v1/namespaces/{quote(self.namespace)}/pods"
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Validated messages shared by the controller and runner backends."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import asdict, dataclass
|
||||
import json
|
||||
|
||||
|
||||
BACKENDS = frozenset({"pod", "vm"})
|
||||
REQUIRED_LABEL = "self-hosted"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RunnerRequest:
|
||||
"""A queued Gitea job that needs one disposable runner environment."""
|
||||
|
||||
job_id: int
|
||||
run_id: int
|
||||
backend: str
|
||||
repository: str
|
||||
job_name: str
|
||||
labels: tuple[str, ...]
|
||||
schema_version: int = 1
|
||||
|
||||
@classmethod
|
||||
def from_webhook(cls, payload: object) -> RunnerRequest | None:
|
||||
if not isinstance(payload, dict) or payload.get("action") != "queued":
|
||||
return None
|
||||
|
||||
job = payload.get("workflow_job")
|
||||
repository = payload.get("repository")
|
||||
if not isinstance(job, dict) or not isinstance(repository, dict):
|
||||
return None
|
||||
|
||||
labels_value = job.get("labels")
|
||||
if not isinstance(labels_value, list) or not all(
|
||||
isinstance(label, str) for label in labels_value
|
||||
):
|
||||
return None
|
||||
labels = tuple(dict.fromkeys(labels_value))
|
||||
selected = BACKENDS.intersection(labels)
|
||||
if REQUIRED_LABEL not in labels or len(selected) != 1:
|
||||
return None
|
||||
|
||||
job_id = job.get("id")
|
||||
run_id = job.get("run_id")
|
||||
job_name = job.get("name")
|
||||
full_name = repository.get("full_name")
|
||||
if not _positive_int(job_id) or not _positive_int(run_id):
|
||||
return None
|
||||
if not all(_nonempty(value) for value in (job_name, full_name)):
|
||||
return None
|
||||
|
||||
return cls(
|
||||
job_id=job_id,
|
||||
run_id=run_id,
|
||||
backend=next(iter(selected)),
|
||||
repository=full_name.strip(),
|
||||
job_name=job_name.strip(),
|
||||
labels=labels,
|
||||
)
|
||||
|
||||
def to_json(self) -> bytes:
|
||||
document = asdict(self)
|
||||
document["labels"] = list(self.labels)
|
||||
return json.dumps(
|
||||
document,
|
||||
ensure_ascii=False,
|
||||
separators=(",", ":"),
|
||||
sort_keys=True,
|
||||
).encode()
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, body: bytes) -> RunnerRequest:
|
||||
document = json.loads(body)
|
||||
if not isinstance(document, dict) or document.get("schema_version") != 1:
|
||||
raise ValueError("unsupported runner request")
|
||||
labels = document.get("labels")
|
||||
if not isinstance(labels, list) or not all(
|
||||
isinstance(label, str) for label in labels
|
||||
):
|
||||
raise ValueError("invalid runner labels")
|
||||
try:
|
||||
request = cls(
|
||||
job_id=document["job_id"],
|
||||
run_id=document["run_id"],
|
||||
backend=document["backend"],
|
||||
repository=document["repository"],
|
||||
job_name=document["job_name"],
|
||||
labels=tuple(labels),
|
||||
)
|
||||
except KeyError as error:
|
||||
raise ValueError(f"missing runner request field: {error.args[0]}") from error
|
||||
if (
|
||||
not _positive_int(request.job_id)
|
||||
or not _positive_int(request.run_id)
|
||||
or request.backend not in BACKENDS
|
||||
or not _nonempty(request.repository)
|
||||
or not _nonempty(request.job_name)
|
||||
or REQUIRED_LABEL not in request.labels
|
||||
or request.backend not in request.labels
|
||||
):
|
||||
raise ValueError("invalid runner request")
|
||||
return request
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class IdentityBinding:
|
||||
"""The actual task claimed by an ephemeral runner."""
|
||||
|
||||
job_id: int
|
||||
run_id: int
|
||||
backend: str
|
||||
runner_name: str
|
||||
repository: str
|
||||
job_name: str
|
||||
schema_version: int = 1
|
||||
|
||||
@classmethod
|
||||
def from_webhook(cls, payload: object) -> IdentityBinding | None:
|
||||
if not isinstance(payload, dict) or payload.get("action") != "in_progress":
|
||||
return None
|
||||
job = payload.get("workflow_job")
|
||||
repository = payload.get("repository")
|
||||
if not isinstance(job, dict) or not isinstance(repository, dict):
|
||||
return None
|
||||
|
||||
labels = job.get("labels")
|
||||
if not isinstance(labels, list) or not all(
|
||||
isinstance(label, str) for label in labels
|
||||
):
|
||||
return None
|
||||
selected = BACKENDS.intersection(labels)
|
||||
if REQUIRED_LABEL not in labels or len(selected) != 1:
|
||||
return None
|
||||
backend = next(iter(selected))
|
||||
|
||||
job_id = job.get("id")
|
||||
run_id = job.get("run_id")
|
||||
runner_name = job.get("runner_name")
|
||||
job_name = job.get("name")
|
||||
full_name = repository.get("full_name")
|
||||
if not _positive_int(job_id) or not _positive_int(run_id):
|
||||
return None
|
||||
if not all(
|
||||
_nonempty(value)
|
||||
for value in (runner_name, job_name, full_name)
|
||||
):
|
||||
return None
|
||||
if not runner_name.startswith(f"gitea-{backend}-"):
|
||||
return None
|
||||
|
||||
return cls(
|
||||
job_id=job_id,
|
||||
run_id=run_id,
|
||||
backend=backend,
|
||||
runner_name=runner_name.strip(),
|
||||
repository=full_name.strip(),
|
||||
job_name=job_name.strip(),
|
||||
)
|
||||
|
||||
def to_json(self) -> bytes:
|
||||
return json.dumps(
|
||||
asdict(self),
|
||||
ensure_ascii=False,
|
||||
separators=(",", ":"),
|
||||
sort_keys=True,
|
||||
).encode()
|
||||
|
||||
|
||||
def _positive_int(value: object) -> bool:
|
||||
return isinstance(value, int) and not isinstance(value, bool) and value > 0
|
||||
|
||||
|
||||
def _nonempty(value: object) -> bool:
|
||||
return isinstance(value, str) and bool(value.strip())
|
||||
@@ -0,0 +1,378 @@
|
||||
#!/usr/bin/env python3
|
||||
"""JetStream worker that creates one disposable Kubernetes Pod per job."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import ssl
|
||||
import uuid
|
||||
|
||||
import nats
|
||||
from aiohttp import ClientResponseError
|
||||
from nats.errors import TimeoutError
|
||||
from nats.js.api import AckPolicy, ConsumerConfig
|
||||
|
||||
from .kubernetes import KubernetesClient
|
||||
from .models import IdentityBinding, RunnerRequest
|
||||
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
|
||||
REQUEST_SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.pod")
|
||||
BINDING_SUBJECT = os.environ.get("NATS_BINDING_SUBJECT", "ci.runner.pod.binding")
|
||||
NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
|
||||
NATS_USER = os.environ.get("NATS_USER", "ci-worker")
|
||||
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")
|
||||
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("RUNNER_NAMESPACE", "gitea-actions")
|
||||
RUNNER_IMAGE = os.environ.get("RUNNER_IMAGE", "docker.io/gitea/runner:2")
|
||||
RUNNER_SERVICE_ACCOUNT = os.environ.get("RUNNER_SERVICE_ACCOUNT", "gitea-dynamic-runner")
|
||||
RUNNER_TOKEN_SECRET = os.environ.get("RUNNER_TOKEN_SECRET", "gitea-dynamic-runner")
|
||||
GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.ddupan.top")
|
||||
CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4"))
|
||||
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:
|
||||
parts = repository.split("/")
|
||||
if len(parts) != 2 or not all(parts):
|
||||
raise ValueError("repository must be owner/name")
|
||||
if not job_name.strip():
|
||||
raise ValueError("identity components must not be empty")
|
||||
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]:
|
||||
if request.backend != "pod":
|
||||
raise ValueError("Pod backend only accepts pod requests")
|
||||
return {
|
||||
"apiVersion": "v1",
|
||||
"kind": "Pod",
|
||||
"metadata": {
|
||||
"name": pod_name,
|
||||
"namespace": NAMESPACE,
|
||||
"labels": {
|
||||
"app.kubernetes.io/name": "gitea-dynamic-runner",
|
||||
"app.kubernetes.io/component": "runner",
|
||||
"ci.ddupan.top/backend": "pod",
|
||||
},
|
||||
"annotations": {
|
||||
"ci.ddupan.top/queued-job-id": str(request.job_id),
|
||||
"ci.ddupan.top/queued-run-id": str(request.run_id),
|
||||
},
|
||||
},
|
||||
"spec": {
|
||||
"serviceAccountName": RUNNER_SERVICE_ACCOUNT,
|
||||
"restartPolicy": "Never",
|
||||
"terminationGracePeriodSeconds": 30,
|
||||
"containers": [
|
||||
{
|
||||
"name": "runner",
|
||||
"image": RUNNER_IMAGE,
|
||||
"imagePullPolicy": "IfNotPresent",
|
||||
"securityContext": {"privileged": True},
|
||||
"env": [
|
||||
{"name": "GITEA_INSTANCE_URL", "value": GITEA_INSTANCE},
|
||||
{
|
||||
"name": "GITEA_RUNNER_NAME",
|
||||
"valueFrom": {"fieldRef": {"fieldPath": "metadata.name"}},
|
||||
},
|
||||
{
|
||||
"name": "GITEA_RUNNER_REGISTRATION_TOKEN_FILE",
|
||||
"value": "/run/secrets/gitea/token",
|
||||
},
|
||||
{"name": "GITEA_RUNNER_LABELS", "value": "self-hosted:host,pod:host"},
|
||||
{"name": "GITEA_RUNNER_EPHEMERAL", "value": "1"},
|
||||
{"name": "GITEA_RUNNER_ONCE", "value": "1"},
|
||||
{"name": "CONFIG_FILE", "value": "/etc/gitea-runner/config.yaml"},
|
||||
],
|
||||
"volumeMounts": [
|
||||
{
|
||||
"name": "registration-token",
|
||||
"mountPath": "/run/secrets/gitea",
|
||||
"readOnly": True,
|
||||
},
|
||||
{
|
||||
"name": "spire-agent-socket",
|
||||
"mountPath": "/run/spire/agent-sockets",
|
||||
"readOnly": True,
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
"volumes": [
|
||||
{
|
||||
"name": "registration-token",
|
||||
"secret": {
|
||||
"secretName": RUNNER_TOKEN_SECRET,
|
||||
"items": [{"key": "token", "path": "token"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "spire-agent-socket",
|
||||
"csi": {"driver": "csi.spiffe.io", "readOnly": True},
|
||||
},
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
async def heartbeat(message: object, stop: asyncio.Event) -> None:
|
||||
while True:
|
||||
try:
|
||||
await asyncio.wait_for(stop.wait(), timeout=60)
|
||||
return
|
||||
except asyncio.TimeoutError:
|
||||
await message.in_progress()
|
||||
|
||||
|
||||
async def wait_for_pod(client: KubernetesClient, name: str) -> bool:
|
||||
deadline = asyncio.get_running_loop().time() + POD_TIMEOUT
|
||||
while asyncio.get_running_loop().time() < deadline:
|
||||
pod = await client.get_pod(name)
|
||||
if pod is None:
|
||||
raise RuntimeError(f"runner Pod {name} disappeared")
|
||||
status = pod.get("status")
|
||||
phase = status.get("phase") if isinstance(status, dict) else None
|
||||
if phase == "Succeeded":
|
||||
return True
|
||||
if phase == "Failed":
|
||||
return False
|
||||
await asyncio.sleep(2)
|
||||
raise asyncio.TimeoutError(f"runner Pod {name} timed out")
|
||||
|
||||
|
||||
async def run_request(message: object, client: KubernetesClient) -> None:
|
||||
try:
|
||||
request = RunnerRequest.from_json(message.data)
|
||||
except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error:
|
||||
LOG.error("discarding invalid Pod request: %s", error)
|
||||
await message.ack()
|
||||
return
|
||||
if request.backend != "pod":
|
||||
LOG.error("discarding %s request received by Pod worker", request.backend)
|
||||
await message.ack()
|
||||
return
|
||||
|
||||
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()
|
||||
pulse = asyncio.create_task(heartbeat(message, stop))
|
||||
try:
|
||||
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):
|
||||
await message.ack()
|
||||
LOG.info(
|
||||
"runner request acknowledged pod=%s job_id=%d result=succeeded",
|
||||
pod_name,
|
||||
request.job_id,
|
||||
)
|
||||
else:
|
||||
LOG.error(
|
||||
"runner Pod failed pod=%s job_id=%d; request will be retried",
|
||||
pod_name,
|
||||
request.job_id,
|
||||
)
|
||||
await message.nak(delay=30)
|
||||
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)
|
||||
raise
|
||||
finally:
|
||||
stop.set()
|
||||
await pulse
|
||||
LOG.info("deleting runner Pod pod=%s job_id=%d", pod_name, request.job_id)
|
||||
await client.delete_pod(pod_name)
|
||||
|
||||
|
||||
async def bind_request(message: object, client: KubernetesClient) -> None:
|
||||
binding: IdentityBinding | None = None
|
||||
try:
|
||||
document = json.loads(message.data)
|
||||
binding = IdentityBinding(**document)
|
||||
if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"):
|
||||
raise ValueError("invalid Pod identity binding")
|
||||
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)
|
||||
except ClientResponseError as error:
|
||||
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)
|
||||
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)
|
||||
raise
|
||||
except (json.JSONDecodeError, TypeError, ValueError) as error:
|
||||
LOG.error("discarding invalid identity binding: %s", error)
|
||||
await message.ack()
|
||||
return
|
||||
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:
|
||||
active: set[asyncio.Task[None]] = set()
|
||||
while True:
|
||||
active = {task for task in active if not task.done()}
|
||||
free = CAPACITY - len(active)
|
||||
if free < 1:
|
||||
await asyncio.wait(active, return_when=asyncio.FIRST_COMPLETED)
|
||||
continue
|
||||
try:
|
||||
messages = await subscription.fetch(batch=free, timeout=5)
|
||||
except TimeoutError:
|
||||
continue
|
||||
for message in messages:
|
||||
task = asyncio.create_task(run_request(message, client))
|
||||
task.add_done_callback(_report_task)
|
||||
active.add(task)
|
||||
|
||||
|
||||
async def consume_bindings(subscription: object, client: KubernetesClient) -> None:
|
||||
while True:
|
||||
try:
|
||||
messages = await subscription.fetch(batch=16, timeout=5)
|
||||
except TimeoutError:
|
||||
continue
|
||||
await asyncio.gather(*(bind_request(message, client) for message in messages))
|
||||
|
||||
|
||||
def _report_task(task: asyncio.Task[None]) -> None:
|
||||
if not task.cancelled() and (error := task.exception()) is not None:
|
||||
LOG.error("Pod runner task failed", exc_info=(type(error), error, error.__traceback__))
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
if CAPACITY < 1:
|
||||
raise ValueError("RUNNER_CAPACITY must be at least 1")
|
||||
context = 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=context,
|
||||
name="pod-runner-worker",
|
||||
)
|
||||
js = nc.jetstream()
|
||||
requests = await js.pull_subscribe(
|
||||
REQUEST_SUBJECT,
|
||||
durable="pod",
|
||||
stream=STREAM,
|
||||
config=ConsumerConfig(
|
||||
durable_name="pod",
|
||||
filter_subject=REQUEST_SUBJECT,
|
||||
ack_policy=AckPolicy.EXPLICIT,
|
||||
ack_wait=5 * 60,
|
||||
max_ack_pending=max(CAPACITY, 1),
|
||||
max_deliver=5,
|
||||
),
|
||||
)
|
||||
bindings = await js.pull_subscribe(
|
||||
BINDING_SUBJECT,
|
||||
durable="pod-binding",
|
||||
stream=STREAM,
|
||||
config=ConsumerConfig(
|
||||
durable_name="pod-binding",
|
||||
filter_subject=BINDING_SUBJECT,
|
||||
ack_policy=AckPolicy.EXPLICIT,
|
||||
ack_wait=30,
|
||||
max_ack_pending=64,
|
||||
max_deliver=10,
|
||||
),
|
||||
)
|
||||
async with KubernetesClient(
|
||||
api_url=KUBERNETES_API,
|
||||
token_file=KUBERNETES_TOKEN_FILE,
|
||||
ca_file=KUBERNETES_CA_FILE,
|
||||
namespace=NAMESPACE,
|
||||
) as client:
|
||||
try:
|
||||
await asyncio.gather(
|
||||
consume_requests(requests, client),
|
||||
consume_bindings(bindings, client),
|
||||
)
|
||||
finally:
|
||||
await nc.drain()
|
||||
|
||||
|
||||
def cli() -> None:
|
||||
logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
|
||||
asyncio.run(main())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
cli()
|
||||
@@ -2,6 +2,7 @@
|
||||
"""Capacity-bounded JetStream consumer that launches one ephemeral VM per job."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import secrets
|
||||
@@ -14,11 +15,13 @@ from aiohttp import web
|
||||
from nats.errors import TimeoutError
|
||||
from nats.js.api import AckPolicy, ConsumerConfig
|
||||
|
||||
from .models import RunnerRequest
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "1"))
|
||||
SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.kind-microvm")
|
||||
SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.vm")
|
||||
STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
|
||||
DURABLE = os.environ.get("NATS_DURABLE", "kind-microvm")
|
||||
DURABLE = os.environ.get("NATS_DURABLE", "vm")
|
||||
MAX_INFLIGHT = int(os.environ.get("RUNNER_MAX_INFLIGHT", "64"))
|
||||
NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
|
||||
NATS_USER = os.environ.get("NATS_USER", "ci-worker")
|
||||
@@ -28,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")
|
||||
TOKEN_LISTEN = os.environ.get("TOKEN_LISTEN", "172.30.0.1")
|
||||
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] = {}
|
||||
token_lock = asyncio.Lock()
|
||||
@@ -42,6 +48,13 @@ async def token(request: web.Request) -> web.Response:
|
||||
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:
|
||||
while True:
|
||||
try:
|
||||
@@ -52,6 +65,17 @@ async def heartbeat(message: object, stop: asyncio.Event) -> None:
|
||||
|
||||
|
||||
async def run_one(message: object) -> None:
|
||||
try:
|
||||
request = RunnerRequest.from_json(message.data)
|
||||
except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error:
|
||||
LOG.error("discarding invalid runner request: %s", error)
|
||||
await message.ack()
|
||||
return
|
||||
if request.backend != "vm":
|
||||
LOG.error("discarding %s request received by VM worker", request.backend)
|
||||
await message.ack()
|
||||
return
|
||||
|
||||
instance_id = str(uuid.uuid4())
|
||||
nonce = secrets.token_urlsafe(32)
|
||||
async with token_lock:
|
||||
@@ -59,7 +83,18 @@ async def run_one(message: object) -> None:
|
||||
stop = asyncio.Event()
|
||||
pulse = asyncio.create_task(heartbeat(message, stop))
|
||||
try:
|
||||
process = await asyncio.create_subprocess_exec(LAUNCHER, instance_id, nonce)
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
LAUNCHER,
|
||||
instance_id,
|
||||
nonce,
|
||||
env={
|
||||
**os.environ,
|
||||
"RUNNER_JOB_ID": str(request.job_id),
|
||||
"RUNNER_RUN_ID": str(request.run_id),
|
||||
"RUNNER_REPOSITORY": request.repository,
|
||||
"RUNNER_JOB_NAME": request.job_name,
|
||||
},
|
||||
)
|
||||
return_code = await process.wait()
|
||||
if return_code == 0:
|
||||
await message.ack()
|
||||
@@ -137,6 +172,7 @@ async def consume() -> None:
|
||||
async def main() -> None:
|
||||
app = web.Application()
|
||||
app.router.add_get("/token/{nonce}", token)
|
||||
app.router.add_get("/assets/guest-assets.tar.gz", guest_assets)
|
||||
runner = web.AppRunner(app)
|
||||
await runner.setup()
|
||||
await web.TCPSite(runner, TOKEN_LISTEN, TOKEN_PORT).start()
|
||||
@@ -1 +0,0 @@
|
||||
"""Gitea microVM runner controller and worker."""
|
||||
@@ -7,7 +7,7 @@ Requires=microvm-runner-network.service
|
||||
[Service]
|
||||
Type=simple
|
||||
EnvironmentFile=-/etc/microvm-runner/worker.env
|
||||
ExecStart=/opt/gitea-microvm-runner/venv/bin/gitea-microvm-worker
|
||||
ExecStart=/opt/gitea-dynamic-runner/venv/bin/gitea-dynamic-runner-vm-worker
|
||||
Restart=on-failure
|
||||
RestartSec=5s
|
||||
SupplementaryGroups=kvm
|
||||
|
||||
+77
-11
@@ -1,22 +1,88 @@
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
|
||||
from gitea_microvm_runner import controller
|
||||
import pytest
|
||||
|
||||
from gitea_dynamic_runner import controller
|
||||
from gitea_dynamic_runner.models import IdentityBinding, RunnerRequest
|
||||
|
||||
|
||||
def test_accepts_matching_queued_job(monkeypatch):
|
||||
monkeypatch.setattr(controller, "LABEL", "kind-microvm")
|
||||
assert controller.accepts({
|
||||
def queued_job(**overrides):
|
||||
job = {
|
||||
"id": 47,
|
||||
"run_id": 12,
|
||||
"name": "publish-image",
|
||||
"labels": ["self-hosted", "pod"],
|
||||
}
|
||||
job.update(overrides)
|
||||
return {
|
||||
"action": "queued",
|
||||
"workflow_job": {"id": 47, "labels": ["linux", "kind-microvm"]},
|
||||
}) == (True, "47")
|
||||
"workflow_job": job,
|
||||
"repository": {"full_name": "panxiao81/example"},
|
||||
}
|
||||
|
||||
|
||||
def test_rejects_other_actions_labels_and_boolean_id(monkeypatch):
|
||||
monkeypatch.setattr(controller, "LABEL", "kind-microvm")
|
||||
assert controller.accepts({"action": "completed", "workflow_job": {"id": 1, "labels": ["kind-microvm"]}}) == (False, None)
|
||||
assert controller.accepts({"action": "queued", "workflow_job": {"id": 1, "labels": ["host"]}}) == (False, None)
|
||||
assert controller.accepts({"action": "queued", "workflow_job": {"id": True, "labels": ["kind-microvm"]}}) == (False, None)
|
||||
def test_accepts_matching_queued_job():
|
||||
assert controller.accepts(queued_job()) == (True, "47")
|
||||
|
||||
|
||||
def test_rejects_other_actions_labels_and_boolean_id():
|
||||
completed = queued_job()
|
||||
completed["action"] = "completed"
|
||||
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", "pod", "vm"])) == (False, None)
|
||||
assert controller.accepts(queued_job(id=True)) == (False, None)
|
||||
|
||||
|
||||
def test_runner_request_contains_stable_identity_context():
|
||||
request = RunnerRequest.from_webhook(queued_job())
|
||||
assert request is not None
|
||||
assert request.backend == "pod"
|
||||
assert request.repository == "panxiao81/example"
|
||||
assert request.job_name == "publish-image"
|
||||
assert request.job_id == 47
|
||||
assert json.loads(request.to_json()) == {
|
||||
"backend": "pod",
|
||||
"job_id": 47,
|
||||
"job_name": "publish-image",
|
||||
"labels": ["self-hosted", "pod"],
|
||||
"repository": "panxiao81/example",
|
||||
"run_id": 12,
|
||||
"schema_version": 1,
|
||||
}
|
||||
|
||||
|
||||
def test_runner_request_requires_complete_identity_context():
|
||||
assert RunnerRequest.from_webhook(queued_job(run_id=None)) is None
|
||||
assert RunnerRequest.from_webhook(queued_job(name=" ")) is None
|
||||
|
||||
|
||||
def test_runner_request_json_round_trip_and_validation():
|
||||
request = RunnerRequest.from_webhook(queued_job())
|
||||
assert request is not None
|
||||
assert RunnerRequest.from_json(request.to_json()) == request
|
||||
with pytest.raises(ValueError, match="unsupported"):
|
||||
RunnerRequest.from_json(b'{"schema_version":2}')
|
||||
|
||||
|
||||
def test_identity_binding_uses_actual_runner_assignment():
|
||||
payload = queued_job(runner_name="gitea-pod-6c47d03d")
|
||||
payload["action"] = "in_progress"
|
||||
binding = IdentityBinding.from_webhook(payload)
|
||||
assert binding is not None
|
||||
assert binding.backend == "pod"
|
||||
assert binding.runner_name == "gitea-pod-6c47d03d"
|
||||
assert binding.repository == "panxiao81/example"
|
||||
assert binding.job_name == "publish-image"
|
||||
assert json.loads(binding.to_json())["job_id"] == 47
|
||||
|
||||
|
||||
def test_identity_binding_rejects_runner_from_another_pool():
|
||||
payload = queued_job(runner_name="gitea-vm-6c47d03d")
|
||||
payload["action"] = "in_progress"
|
||||
assert IdentityBinding.from_webhook(payload) is None
|
||||
|
||||
|
||||
def test_signature_accepts_gitea_and_prefixed_forms(tmp_path, monkeypatch):
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import pytest
|
||||
|
||||
from gitea_microvm_runner.jwt_broker import extract_svid
|
||||
from gitea_dynamic_runner.jwt_broker import extract_svid
|
||||
|
||||
|
||||
def test_extract_svid_from_spire_json():
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
import pytest
|
||||
|
||||
from gitea_dynamic_runner.models import RunnerRequest
|
||||
from gitea_dynamic_runner import pod_worker
|
||||
|
||||
|
||||
def request() -> RunnerRequest:
|
||||
return RunnerRequest(
|
||||
job_id=47,
|
||||
run_id=12,
|
||||
backend="pod",
|
||||
repository="panxiao81/example",
|
||||
job_name="publish image",
|
||||
labels=("self-hosted", "pod"),
|
||||
)
|
||||
|
||||
|
||||
def test_identity_path_is_meaningful_and_uri_safe():
|
||||
path = pod_worker.identity_path("panxiao81/example", "publish image")
|
||||
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):
|
||||
pod_worker.identity_path("invalid", "test")
|
||||
|
||||
|
||||
def test_pod_manifest_is_disposable_host_runner(monkeypatch):
|
||||
monkeypatch.setattr(pod_worker, "NAMESPACE", "ci")
|
||||
manifest = pod_worker.pod_manifest(request(), "gitea-pod-abcd")
|
||||
assert manifest["metadata"]["name"] == "gitea-pod-abcd"
|
||||
assert manifest["metadata"]["annotations"]["ci.ddupan.top/queued-job-id"] == "47"
|
||||
spec = manifest["spec"]
|
||||
assert spec["restartPolicy"] == "Never"
|
||||
container = spec["containers"][0]
|
||||
assert container["securityContext"] == {"privileged": True}
|
||||
env = {item["name"]: item for item in container["env"]}
|
||||
assert env["GITEA_RUNNER_LABELS"]["value"] == "self-hosted:host,pod:host"
|
||||
assert env["GITEA_RUNNER_EPHEMERAL"]["value"] == "1"
|
||||
assert env["GITEA_RUNNER_ONCE"]["value"] == "1"
|
||||
assert all(item["name"] != "DOCKER_HOST" for item in container["env"])
|
||||
volumes = {item["name"]: item for item in spec["volumes"]}
|
||||
assert volumes["spire-agent-socket"]["csi"]["driver"] == "csi.spiffe.io"
|
||||
assert "runner-config" not in volumes
|
||||
assert "docker" not in volumes
|
||||
Reference in New Issue
Block a user