diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml new file mode 100644 index 0000000..785a120 --- /dev/null +++ b/.gitea/workflows/test.yml @@ -0,0 +1,27 @@ +name: test +on: + push: + branches: + - main + pull_request: + +jobs: + python: + runs-on: self-hosted + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - run: pip install -e '.[test]' + - run: pytest + - run: python -m compileall -q src + + shell: + runs-on: self-hosted + steps: + - uses: actions/checkout@v4 + - run: | + sudo apt-get update + sudo apt-get install -y shellcheck + - run: shellcheck scripts/* diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..179ad9c --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +.venv/ +dist/ +build/ +*.egg-info/ diff --git a/README.md b/README.md index af4d3ee..a4f0274 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,39 @@ -# gitea-microvm-runner +# Gitea microVM runner -按需启动 Cloud Hypervisor microVM 的 Gitea Actions runner autoscaler \ No newline at end of file +为 Gitea Actions 按需启动 Cloud Hypervisor microVM。适合 kind、嵌套容器和其他不应 +在常驻 Kubernetes runner 中执行的 CI 工作负载。 + +组件： + +- `controller`：接收 Gitea `workflow_job` webhook，将指定 label 的 queued job + 发布到 NATS JetStream。 +- `worker`：在虚拟化宿主机领取任务，限制本机并发，并启动一次性 microVM。 +- `microvm-runner-launch`：为每个任务创建 COW disk、NoCloud seed 和 TAP，运行 + Cloud Hypervisor，退出后完整清理。 +- `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral + runner，执行一个 job 后关机。 + +消息流使用一个 `WorkQueuePolicy` stream。相同 runner label 的所有 worker 共享同一 +durable consumer；扩容只需要增加 worker 或提高单机 capacity。 + +## 开发 + +```bash +python -m venv .venv +. .venv/bin/activate +pip install -e '.[test]' +pytest +``` + +## 安全边界 + +- NATS 密码、webhook secret 和 Gitea registration token 只从文件读取。 +- registration token 不写入 seed image；worker 通过单次 nonce endpoint 交给 guest。 +- guest runner 使用 `--ephemeral`，每台 VM 只执行一个 job。 +- launcher 只接受 UUID instance ID 和 URL-safe nonce，所有临时文件都位于独立目录。 +- base image 不得包含 runner identity、registration token、SSH 密码或 host key。 + +homelab 的 Kubernetes、OpenBao、LXC、bridge 和容量配置保留在 +`panxiao81/homelab-infra`。 + +按需启动 Cloud Hypervisor microVM 的 Gitea Actions runner autoscaler diff --git a/container/controller.Dockerfile b/container/controller.Dockerfile new file mode 100644 index 0000000..993de1c --- /dev/null +++ b/container/controller.Dockerfile @@ -0,0 +1,12 @@ +FROM python:3.12.11-alpine3.22 AS build +WORKDIR /src +COPY pyproject.toml README.md ./ +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 +COPY --from=build /venv /venv +USER 65532:65532 +EXPOSE 8787 +ENTRYPOINT ["/venv/bin/gitea-microvm-controller"] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ac088c0 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["setuptools>=75"] +build-backend = "setuptools.build_meta" + +[project] +name = "gitea-microvm-runner" +version = "0.1.0" +description = "On-demand Cloud Hypervisor runners for Gitea Actions" +requires-python = ">=3.11" +dependencies = ["aiohttp==3.12.15", "nats-py==2.11.0"] + +[project.optional-dependencies] +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" + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/scripts/gitea-microvm-guest-runner b/scripts/gitea-microvm-guest-runner new file mode 100755 index 0000000..4405c11 --- /dev/null +++ b/scripts/gitea-microvm-guest-runner @@ -0,0 +1,28 @@ +#!/bin/sh +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} +token_file=/run/gitea-runner-registration-token + +cleanup() { + rm -f -- "$token_file" + sync + poweroff -f +} +trap cleanup EXIT INT TERM + +umask 077 +curl --fail --silent --show-error --retry 10 --retry-all-errors \ + --connect-timeout 2 --max-time 30 "$token_url" >"$token_file" +gitea-runner register \ + --no-interactive \ + --ephemeral \ + --instance "$instance" \ + --name "$runner_name" \ + --labels "$runner_label:host" \ + --token-file "$token_file" +rm -f -- "$token_file" +gitea-runner daemon diff --git a/scripts/microvm-runner-launch b/scripts/microvm-runner-launch new file mode 100755 index 0000000..7f02cb9 --- /dev/null +++ b/scripts/microvm-runner-launch @@ -0,0 +1,74 @@ +#!/bin/sh +set -eu + +instance_id=${1:?instance ID is required} +nonce=${2:?registration nonce is required} + +case "$instance_id" in + ????????-????-????-????-????????????) ;; + *) echo "invalid instance ID" >&2; exit 64 ;; +esac +case "$nonce" in + *[!A-Za-z0-9_-]*|'') echo "invalid nonce" >&2; exit 64 ;; +esac + +state_root=${RUNNER_STATE_ROOT:-/var/lib/microvm-runner} +base_image=${RUNNER_BASE_IMAGE:-$state_root/images/runner-base.qcow2} +firmware=${RUNNER_FIRMWARE:-$state_root/firmware/CLOUDHV.fd} +bridge=${RUNNER_BRIDGE:-mvrbr0} +cloud_hypervisor=${CLOUD_HYPERVISOR:-/usr/local/bin/cloud-hypervisor} +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} +token_url=${RUNNER_TOKEN_URL:-http://172.30.0.1:8787} + +vm_dir="$state_root/instances/$instance_id" +tap="mvr${instance_id%%-*}" +overlay="$vm_dir/root.qcow2" +seed="$vm_dir/seed.img" +serial="$vm_dir/serial.log" + +cleanup() { + ip link delete "$tap" 2>/dev/null || true + rm -rf -- "$vm_dir" +} +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" + +cat >"$vm_dir/meta-data" <"$vm_dir/user-data" </dev/null 2>&1 || ip link add "$bridge" type bridge + ip address show dev "$bridge" | grep -Fq "$gateway/24" || ip address add "$gateway/24" dev "$bridge" + ip link set "$bridge" up + sysctl -q -w net.ipv4.ip_forward=1 + iptables -t nat -C POSTROUTING -s "$subnet" -j MASQUERADE 2>/dev/null || \ + iptables -t nat -A POSTROUTING -s "$subnet" -j MASQUERADE + dnsmasq --interface="$bridge" --bind-interfaces --except-interface=lo \ + --dhcp-range="$dhcp_start,$dhcp_end,255.255.255.0,1h" \ + --dhcp-option="option:router,$gateway" --pid-file="$pid_file" + ;; +down) + if test -r "$pid_file"; then + kill "$(cat "$pid_file")" 2>/dev/null || true + rm -f -- "$pid_file" + fi + iptables -t nat -D POSTROUTING -s "$subnet" -j MASQUERADE 2>/dev/null || true + ip link delete "$bridge" 2>/dev/null || true + ;; +*) + echo "expected up or down" >&2 + exit 64 + ;; +esac diff --git a/src/gitea_microvm_runner/__init__.py b/src/gitea_microvm_runner/__init__.py new file mode 100644 index 0000000..89dab5d --- /dev/null +++ b/src/gitea_microvm_runner/__init__.py @@ -0,0 +1 @@ +"""Gitea microVM runner controller and worker.""" diff --git a/src/gitea_microvm_runner/controller.py b/src/gitea_microvm_runner/controller.py new file mode 100644 index 0000000..6acc29e --- /dev/null +++ b/src/gitea_microvm_runner/controller.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +"""Gitea workflow_job webhook to NATS JetStream producer.""" + +import hashlib +import hmac +import json +import os +import ssl +from pathlib import Path + +import nats +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}") +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") +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") +WEBHOOK_SECRET_FILE = Path(os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret")) + + +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) + + +def valid_signature(body: bytes, signature: str) -> bool: + expected = hmac.new(WEBHOOK_SECRET_FILE.read_bytes().strip(), body, hashlib.sha256).hexdigest() + return hmac.compare_digest(signature.removeprefix("sha256="), expected) + + +async def ensure_stream(js: object) -> None: + config = StreamConfig( + name=STREAM, + subjects=["ci.runner.*"], + 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: + body = await request.read() + if not valid_signature(body, request.headers.get("X-Gitea-Signature", "")): + raise web.HTTPUnauthorized() + try: + 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) + await request.app["js"].publish( + SUBJECT, + body, + headers={"Nats-Msg-Id": f"gitea-workflow-job-{job_id}"}, + ) + return web.Response(status=202, text="queued\n") + + +async def health(request: web.Request) -> web.Response: + connected = request.app["nc"].is_connected + 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() + await ensure_stream(app["js"]) + yield + await nc.drain() + + +def create_app() -> web.Application: + app = web.Application(client_max_size=1024 * 1024) + app.cleanup_ctx.append(nats_context) + app.router.add_post("/webhook", webhook) + app.router.add_get("/healthz", health) + return app + + +def main() -> None: + web.run_app(create_app(), host=os.environ.get("LISTEN", "0.0.0.0"), port=int(os.environ.get("PORT", "8787"))) + + +if __name__ == "__main__": + main() diff --git a/src/gitea_microvm_runner/worker.py b/src/gitea_microvm_runner/worker.py new file mode 100644 index 0000000..4d3543a --- /dev/null +++ b/src/gitea_microvm_runner/worker.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +"""Capacity-bounded JetStream consumer that launches one ephemeral VM per job.""" + +import asyncio +import logging +import os +import secrets +import ssl +import uuid +from pathlib import Path + +import nats +from aiohttp import web +from nats.errors import TimeoutError +from nats.js.api import AckPolicy, ConsumerConfig + +LOG = logging.getLogger(__name__) +CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "1")) +SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.kind-microvm") +STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER") +DURABLE = os.environ.get("NATS_DURABLE", "kind-microvm") +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") +NATS_PASSWORD_FILE = Path(os.environ.get("NATS_PASSWORD_FILE", "/etc/microvm-runner/nats-password")) +NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt") +REGISTRATION_TOKEN_FILE = Path(os.environ.get("REGISTRATION_TOKEN_FILE", "/etc/microvm-runner/registration-token")) +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")) + +tokens: dict[str, bytes] = {} +token_lock = asyncio.Lock() + + +async def token(request: web.Request) -> web.Response: + nonce = request.match_info["nonce"] + async with token_lock: + value = tokens.pop(nonce, None) + if value is None: + raise web.HTTPNotFound() + return web.Response(body=value, headers={"Cache-Control": "no-store"}) + + +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 run_one(message: object) -> None: + instance_id = str(uuid.uuid4()) + nonce = secrets.token_urlsafe(32) + async with token_lock: + tokens[nonce] = REGISTRATION_TOKEN_FILE.read_bytes().strip() + stop = asyncio.Event() + pulse = asyncio.create_task(heartbeat(message, stop)) + try: + process = await asyncio.create_subprocess_exec(LAUNCHER, instance_id, nonce) + return_code = await process.wait() + if return_code == 0: + await message.ack() + else: + LOG.error("launcher for %s exited with %d", instance_id, return_code) + await message.nak(delay=30) + except Exception: + await message.nak(delay=30) + raise + finally: + stop.set() + await pulse + async with token_lock: + tokens.pop(nonce, None) + + +def _report_task(task: asyncio.Task[None]) -> None: + if task.cancelled(): + return + error = task.exception() + if error is not None: + LOG.error( + "runner task failed", + exc_info=(type(error), error, error.__traceback__), + ) + + +async def consume() -> None: + if CAPACITY < 1: + raise ValueError("RUNNER_CAPACITY must be at least 1") + 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=DURABLE, + ) + js = nc.jetstream() + subscription = await js.pull_subscribe( + SUBJECT, + durable=DURABLE, + stream=STREAM, + config=ConsumerConfig( + durable_name=DURABLE, + filter_subject=SUBJECT, + ack_policy=AckPolicy.EXPLICIT, + ack_wait=5 * 60, + max_ack_pending=MAX_INFLIGHT, + max_deliver=5, + ), + ) + active: set[asyncio.Task[None]] = set() + try: + while True: + active = {task for task in active if not task.done()} + free = CAPACITY - len(active) + if free == 0: + 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_one(message)) + task.add_done_callback(_report_task) + active.add(task) + finally: + if active: + await asyncio.gather(*active, return_exceptions=True) + await nc.drain() + + +async def main() -> None: + app = web.Application() + app.router.add_get("/token/{nonce}", token) + runner = web.AppRunner(app) + await runner.setup() + await web.TCPSite(runner, TOKEN_LISTEN, TOKEN_PORT).start() + try: + await consume() + finally: + await runner.cleanup() + + +def cli() -> None: + logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO")) + asyncio.run(main()) + + +if __name__ == "__main__": + cli() diff --git a/systemd/gitea-microvm-worker.service b/systemd/gitea-microvm-worker.service new file mode 100644 index 0000000..abe0bd7 --- /dev/null +++ b/systemd/gitea-microvm-worker.service @@ -0,0 +1,20 @@ +[Unit] +Description=Gitea microVM runner worker +After=network-online.target microvm-runner-network.service +Wants=network-online.target +Requires=microvm-runner-network.service + +[Service] +Type=simple +EnvironmentFile=-/etc/microvm-runner/worker.env +ExecStart=/opt/gitea-microvm-runner/venv/bin/gitea-microvm-worker +Restart=on-failure +RestartSec=5s +SupplementaryGroups=kvm +PrivateTmp=yes +ProtectHome=yes +ProtectSystem=strict +ReadWritePaths=/var/lib/microvm-runner /run/microvm-runner + +[Install] +WantedBy=multi-user.target diff --git a/systemd/microvm-runner-network.service b/systemd/microvm-runner-network.service new file mode 100644 index 0000000..d1fc77a --- /dev/null +++ b/systemd/microvm-runner-network.service @@ -0,0 +1,12 @@ +[Unit] +Description=Network bridge for Gitea microVM runners +Before=gitea-microvm-worker.service + +[Service] +Type=oneshot +RemainAfterExit=yes +ExecStart=/usr/local/libexec/microvm-runner-network up +ExecStop=/usr/local/libexec/microvm-runner-network down + +[Install] +WantedBy=multi-user.target diff --git a/tests/test_controller.py b/tests/test_controller.py new file mode 100644 index 0000000..9f6cf96 --- /dev/null +++ b/tests/test_controller.py @@ -0,0 +1,31 @@ +import hashlib +import hmac + +from gitea_microvm_runner import controller + + +def test_accepts_matching_queued_job(monkeypatch): + monkeypatch.setattr(controller, "LABEL", "kind-microvm") + assert controller.accepts({ + "action": "queued", + "workflow_job": {"id": 47, "labels": ["linux", "kind-microvm"]}, + }) == (True, "47") + + +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_signature_accepts_gitea_and_prefixed_forms(tmp_path, monkeypatch): + secret = b"test-secret" + body = b'{"action":"queued"}' + secret_file = tmp_path / "secret" + secret_file.write_bytes(secret) + monkeypatch.setattr(controller, "WEBHOOK_SECRET_FILE", secret_file) + digest = hmac.new(secret, body, hashlib.sha256).hexdigest() + assert controller.valid_signature(body, digest) + assert controller.valid_signature(body, f"sha256={digest}") + assert not controller.valid_signature(body, "bad")