From 70f0380a3da03420c521745e61b3be3678920a38 Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Wed, 16 Sep 2026 14:22:09 +0000 Subject: [PATCH 1/2] =?UTF-8?q?=E5=BB=BA=E7=AB=8B=20Pod=20=E4=B8=8E=20VM?= =?UTF-8?q?=20=E8=B0=83=E5=BA=A6=E6=B6=88=E6=81=AF=E5=A5=91=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/design-principles.md | 15 ++- scripts/gitea-microvm-guest-runner | 4 +- scripts/microvm-runner-launch | 4 +- src/gitea_microvm_runner/controller.py | 48 ++++--- src/gitea_microvm_runner/models.py | 176 +++++++++++++++++++++++++ src/gitea_microvm_runner/worker.py | 31 ++++- tests/test_controller.py | 86 ++++++++++-- 7 files changed, 322 insertions(+), 42 deletions(-) create mode 100644 src/gitea_microvm_runner/models.py diff --git a/docs/design-principles.md b/docs/design-principles.md index cb5a143..61746a2 100644 --- a/docs/design-principles.md +++ b/docs/design-principles.md @@ -28,6 +28,12 @@ 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、 @@ -48,14 +54,15 @@ kind 等工具由 pipeline 按需安装和启动，而不是由 controller 预 SPIFFE ID 由具有业务意义且稳定的 workflow 上下文派生： ```text -spiffe://ddupan.top/ci//// +spiffe://ddupan.top/ci/// ``` 同一种任务在不同运行中使用相同的逻辑 SPIFFE ID；每次运行取得独立、短期的 SVID。 Pod 与 VM 是可替换的执行实现，因此默认不写入 SPIFFE ID。 -workflow 和 job 名称必须经过确定性的路径规范化。规范化结果必须保留仓库边界，并在 -发生冲突时拒绝创建环境，不能静默地让两个任务共享身份。 +job 名称必须经过确定性的路径规范化。规范化结果必须保留仓库边界，并在发生冲突时 +拒绝创建环境，不能静默地让两个任务共享身份。同一仓库内需要不同权限的任务应使用 +不同的 job 名称；workflow 文件只是编排载体，不进入权限身份。 ## Self-service 与授权边界 @@ -64,7 +71,7 @@ allowlist。能够修改仓库 CI 的主体本来就能修改该仓库已有任 重复审批不能形成额外的安全边界，只会破坏 self-service。 身份不等于权限。新任务可以立即取得自己的 SPIFFE ID，但默认不会因此获得 Zot、 -OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意义的 workflow/job 身份 +OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意义的 job 身份 配置授权策略。 ## 非目标设计 diff --git a/scripts/gitea-microvm-guest-runner b/scripts/gitea-microvm-guest-runner index 4405c11..7d8d7ba 100755 --- a/scripts/gitea-microvm-guest-runner +++ b/scripts/gitea-microvm-guest-runner @@ -4,7 +4,7 @@ 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 cleanup() { @@ -22,7 +22,7 @@ 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 diff --git a/scripts/microvm-runner-launch b/scripts/microvm-runner-launch index 7f02cb9..b84d0ce 100755 --- a/scripts/microvm-runner-launch +++ b/scripts/microvm-runner-launch @@ -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" @@ -48,7 +48,7 @@ EOF cat >"$vm_dir/user-data" < 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 +39,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,15 +64,25 @@ 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) - 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") + runner_request = RunnerRequest.from_webhook(payload) + if runner_request is not None: + 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 binding is not None: + await request.app["js"].publish( + f"{SUBJECT_PREFIX}.{binding.backend}.binding", + binding.to_json(), + headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"}, + ) + return web.Response(status=202, text="binding queued\n") + + return web.Response(status=204) async def health(request: web.Request) -> web.Response: diff --git a/src/gitea_microvm_runner/models.py b/src/gitea_microvm_runner/models.py new file mode 100644 index 0000000..9b0178b --- /dev/null +++ b/src/gitea_microvm_runner/models.py @@ -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()) diff --git a/src/gitea_microvm_runner/worker.py b/src/gitea_microvm_runner/worker.py index 4d3543a..b25d402 100644 --- a/src/gitea_microvm_runner/worker.py +++ b/src/gitea_microvm_runner/worker.py @@ -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") @@ -52,6 +55,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 +73,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() diff --git a/tests/test_controller.py b/tests/test_controller.py index 9f6cf96..9419598 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -1,22 +1,88 @@ import hashlib import hmac +import json + +import pytest from gitea_microvm_runner import controller +from gitea_microvm_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): -- 2.54.0 From 4277bd907a76666a6a9a2d55c3f33209308757df Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Wed, 16 Sep 2026 14:26:07 +0000 Subject: [PATCH 2/2] =?UTF-8?q?=E5=AE=9E=E7=8E=B0=E4=B8=80=E6=AC=A1?= =?UTF-8?q?=E6=80=A7=20Pod=20Runner=20=E5=90=8E=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 + config/runner.yaml | 13 ++ container/runner.Dockerfile | 8 + pyproject.toml | 1 + scripts/gitea-job-started | 19 ++ src/gitea_microvm_runner/kubernetes.py | 89 +++++++ src/gitea_microvm_runner/pod_worker.py | 307 +++++++++++++++++++++++++ tests/test_pod_worker.py | 43 ++++ 8 files changed, 482 insertions(+) create mode 100644 config/runner.yaml create mode 100644 container/runner.Dockerfile create mode 100644 scripts/gitea-job-started create mode 100644 src/gitea_microvm_runner/kubernetes.py create mode 100644 src/gitea_microvm_runner/pod_worker.py create mode 100644 tests/test_pod_worker.py diff --git a/README.md b/README.md index b3f239c..9d91702 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,8 @@ runs-on: [self-hosted, vm] Cloud Hypervisor，退出后完整清理。 - `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral runner，执行一个 job 后关机。 +- `pod-worker`：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用 + host executor，Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。 - `jwt-broker`：早期共享 Kubernetes runner 的过渡实验；目标架构不部署它，每个 动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。 diff --git a/config/runner.yaml b/config/runner.yaml new file mode 100644 index 0000000..e46dc88 --- /dev/null +++ b/config/runner.yaml @@ -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: [] diff --git a/container/runner.Dockerfile b/container/runner.Dockerfile new file mode 100644 index 0000000..f354c49 --- /dev/null +++ b/container/runner.Dockerfile @@ -0,0 +1,8 @@ +FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire + +FROM docker.io/gitea/runner:2@sha256:66d80966792e621c9761c47919644198d35fd1c297e9a01e69ed3c1ae37db0c7 +USER root +COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent +COPY config/runner.yaml /etc/gitea-runner/config.yaml +COPY scripts/gitea-job-started /usr/local/libexec/gitea-job-started +RUN chmod 0755 /usr/local/libexec/gitea-job-started diff --git a/pyproject.toml b/pyproject.toml index 8c1e745..39062c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,7 @@ 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-pod-worker = "gitea_microvm_runner.pod_worker:cli" gitea-spire-jwt-broker = "gitea_microvm_runner.jwt_broker:main" [tool.pytest.ini_options] diff --git a/scripts/gitea-job-started b/scripts/gitea-job-started new file mode 100644 index 0000000..b6b3b83 --- /dev/null +++ b/scripts/gitea-job-started @@ -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 diff --git a/src/gitea_microvm_runner/kubernetes.py b/src/gitea_microvm_runner/kubernetes.py new file mode 100644 index 0000000..9cecb71 --- /dev/null +++ b/src/gitea_microvm_runner/kubernetes.py @@ -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" diff --git a/src/gitea_microvm_runner/pod_worker.py b/src/gitea_microvm_runner/pod_worker.py new file mode 100644 index 0000000..f733081 --- /dev/null +++ b/src/gitea_microvm_runner/pod_worker.py @@ -0,0 +1,307 @@ +#!/usr/bin/env python3 +"""JetStream worker that creates one disposable Kubernetes Pod per job.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +from pathlib import Path +import ssl +from urllib.parse import quote +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))) + + +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") + encoded = [quote(part, safe="-._~") for part in (*parts, job_name)] + if not job_name.strip() or any(not part for part in encoded): + raise ValueError("identity components must not be empty") + return "/".join(encoded) + + +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]}" + stop = asyncio.Event() + pulse = asyncio.create_task(heartbeat(message, stop)) + try: + await client.create_pod(pod_manifest(request, pod_name)) + if await wait_for_pod(client, pod_name): + await message.ack() + else: + LOG.error("runner Pod %s failed", pod_name) + await message.nak(delay=30) + except Exception: + await message.nak(delay=30) + raise + finally: + stop.set() + await pulse + await client.delete_pod(pod_name) + + +async def bind_request(message: object, client: KubernetesClient) -> 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) + await client.bind_identity(binding.runner_name, path) + except ClientResponseError as error: + if error.status == 404: + await message.nak(delay=2) + return + 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() + + +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() diff --git a/tests/test_pod_worker.py b/tests/test_pod_worker.py new file mode 100644 index 0000000..ab67270 --- /dev/null +++ b/tests/test_pod_worker.py @@ -0,0 +1,43 @@ +import pytest + +from gitea_microvm_runner.models import RunnerRequest +from gitea_microvm_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(): + assert pod_worker.identity_path("panxiao81/example", "publish image") == ( + "panxiao81/example/publish%20image" + ) + 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 -- 2.54.0