From d7ba64d6f32ca9505ca04185e45077146eca7d37 Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Fri, 18 Sep 2026 00:56:10 +0000 Subject: [PATCH 1/2] =?UTF-8?q?=E6=8E=A5=E5=85=A5=20OpenSandbox=20Kata=20R?= =?UTF-8?q?unner=20=E6=8E=A7=E5=88=B6=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 3 + container/runner.Dockerfile | 1 + docs/opensandbox-runner.md | 55 ++++ pyproject.toml | 1 + scripts/gitea-opensandbox-runner | 21 ++ src/gitea_dynamic_runner/opensandbox.py | 79 +++++ .../opensandbox_worker.py | 274 ++++++++++++++++++ .../sandbox_kubernetes.py | 132 +++++++++ tests/test_opensandbox_worker.py | 114 ++++++++ 9 files changed, 680 insertions(+) create mode 100644 docs/opensandbox-runner.md create mode 100644 scripts/gitea-opensandbox-runner create mode 100644 src/gitea_dynamic_runner/opensandbox.py create mode 100644 src/gitea_dynamic_runner/opensandbox_worker.py create mode 100644 src/gitea_dynamic_runner/sandbox_kubernetes.py create mode 100644 tests/test_opensandbox_worker.py diff --git a/README.md b/README.md index 0e2178b..0f10256 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,9 @@ runs-on: [self-hosted, vm] Cloud Hypervisor，退出后完整清理。 - `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral runner，执行一个 job 后关机。 +- `opensandbox-worker`：在正式 sandbox 集群通过 OpenSandbox `ci-vm` Pool 分配 + Kata guest，按实际 Pod UID 创建临时 SPIFFE entry，并在任务结束后回收两者。身份 + 与 Pool 契约见 [`docs/opensandbox-runner.md`](docs/opensandbox-runner.md)。 - `pod-worker`：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用 host executor，Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在 支持原生 job hooks 的 3.x 版本，在 workflow 第一步前等待实际任务对应的 SVID。 diff --git a/container/runner.Dockerfile b/container/runner.Dockerfile index 0859d7d..cc5aa4a 100644 --- a/container/runner.Dockerfile +++ b/container/runner.Dockerfile @@ -14,6 +14,7 @@ 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 +COPY --chmod=0755 scripts/gitea-opensandbox-runner /usr/local/libexec/gitea-opensandbox-runner VOLUME ["/data"] WORKDIR / diff --git a/docs/opensandbox-runner.md b/docs/opensandbox-runner.md new file mode 100644 index 0000000..783e968 --- /dev/null +++ b/docs/opensandbox-runner.md @@ -0,0 +1,55 @@ +# OpenSandbox Kata runner + +The OpenSandbox worker replaces the legacy direct Cloud Hypervisor launcher for +jobs labelled `self-hosted, vm`. It runs in the sandbox Kubernetes cluster and +uses two local control planes: + +1. OpenSandbox Lifecycle API creates a sandbox from the `ci-vm` Pool. +2. Kubernetes API exposes the concrete BatchSandbox allocation and manages its + short-lived `ClusterStaticEntry`. + +## Identity ordering + +The worker sends the stable repository/job identity and a generated runner name +as task environment. Once OpenSandbox has allocated a Pool Pod, the worker reads +the Pod UID and creates an entry with: + +- parent: `spiffe://ddupan.top/spire/agent/k8s_psat/sandbox-kata/pod/`; +- workload: `spiffe://ddupan.top/ci///`; +- selector: `unix:uid:2000`. + +The Pool must run the runner task as UID 2000 and set +`shareProcessNamespace: true`. Its guest-local SPIRE Agent uses a Pod-bound PSAT +and exposes the Workload API through the shared `spire-agent-socket` emptyDir. +The runner image starts through `gitea-opensandbox-runner`, which waits until the +exact expected SVID is available before it registers with Gitea. This prevents a +job from starting between Pod allocation and entry reconciliation. + +The UID selector is the boundary between containers in the same Kata Pod. The +SPIRE Agent and privileged Docker daemon must not run as UID 2000. The runner may +access Docker only through a group-owned Unix socket. + +## Required Pool contract + +The `ci-vm` Pool template owns infrastructure that callers cannot override in +Pool mode: + +- `runtimeClassName: kata-clh-runtime-rs` with block-backed emptyDir storage; +- runner image containing `gitea-opensandbox-runner` and SPIRE CLI; +- guest-local SPIRE Agent sidecar and projected audience `spire-server` token; +- `shareProcessNamespace: true`; +- runner UID 2000 and a distinct UID for every sidecar; +- ephemeral Gitea registration token delivery; +- Docker/BuildKit storage and socket entirely inside the Kata guest. + +The worker ServiceAccount needs read access to BatchSandboxes and Pods and +create/get/delete access to ClusterStaticEntries. OpenSandbox API credentials, +when enabled, are mounted from a Secret and read from +`OPENSANDBOX_API_KEY_FILE`. + +## Cleanup + +On success, failure, timeout, or cancellation the worker deletes the +ClusterStaticEntry before deleting the sandbox. Both deletes accept `404`, so a +JetStream redelivery can safely repeat cleanup. The SPIRE Agent registration is +bound to the Pod UID and is removed by SPIRE after the Pod disappears. diff --git a/pyproject.toml b/pyproject.toml index 780f3cd..40b8091 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ test = ["pytest==8.4.2", "pytest-asyncio==1.2.0"] gitea-dynamic-runner-controller = "gitea_dynamic_runner.controller:main" gitea-dynamic-runner-pod-worker = "gitea_dynamic_runner.pod_worker:cli" gitea-dynamic-runner-vm-worker = "gitea_dynamic_runner.worker:cli" +gitea-dynamic-runner-opensandbox-worker = "gitea_dynamic_runner.opensandbox_worker:cli" gitea-spire-jwt-broker = "gitea_dynamic_runner.jwt_broker:main" [tool.pytest.ini_options] diff --git a/scripts/gitea-opensandbox-runner b/scripts/gitea-opensandbox-runner new file mode 100644 index 0000000..9d40499 --- /dev/null +++ b/scripts/gitea-opensandbox-runner @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +set -euo pipefail + +: "${CI_SPIFFE_ID:?CI_SPIFFE_ID is required}" +: "${SPIFFE_ENDPOINT_SOCKET:?SPIFFE_ENDPOINT_SOCKET is required}" + +socket_path=${SPIFFE_ENDPOINT_SOCKET#unix://} +deadline=$((SECONDS + 120)) +while (( SECONDS < deadline )); do + if /opt/spire/bin/spire-agent api fetch x509 \ + -socketPath "$socket_path" \ + -output json 2>/dev/null | \ + jq -e --arg id "$CI_SPIFFE_ID" \ + 'any(.svids[]?; .spiffe_id == $id)' >/dev/null; then + exec /usr/local/bin/run.sh + fi + sleep 1 +done + +printf 'timed out waiting for OpenSandbox SPIFFE identity %s\n' "$CI_SPIFFE_ID" >&2 +exit 1 diff --git a/src/gitea_dynamic_runner/opensandbox.py b/src/gitea_dynamic_runner/opensandbox.py new file mode 100644 index 0000000..b3cebb2 --- /dev/null +++ b/src/gitea_dynamic_runner/opensandbox.py @@ -0,0 +1,79 @@ +"""Minimal client for the OpenSandbox lifecycle API.""" + +from __future__ import annotations + +from pathlib import Path +from urllib.parse import quote + +from aiohttp import ClientResponseError, ClientSession + + +class OpenSandboxClient: + def __init__( + self, + *, + api_url: str, + api_key_file: Path | None = None, + ) -> None: + self.api_url = api_url.rstrip("/") + self.api_key_file = api_key_file + self.session: ClientSession | None = None + + async def __aenter__(self) -> OpenSandboxClient: + headers = {} + if self.api_key_file is not None: + headers["OPEN-SANDBOX-API-KEY"] = self.api_key_file.read_text().strip() + self.session = ClientSession(headers=headers, raise_for_status=True) + return self + + async def __aexit__(self, *_: object) -> None: + if self.session is not None: + await self.session.close() + + async def create( + self, + *, + pool: str, + timeout: int, + entrypoint: list[str], + env: dict[str, str], + metadata: dict[str, str], + ) -> dict[str, object]: + response = await self._request( + "POST", + "/v1/sandboxes", + json={ + "timeout": timeout, + "entrypoint": entrypoint, + "env": env, + "metadata": metadata, + "extensions": {"poolRef": pool}, + }, + ) + return await response.json() + + async def get(self, sandbox_id: str) -> dict[str, object] | None: + try: + response = await self._request( + "GET", f"/v1/sandboxes/{quote(sandbox_id, safe='')}" + ) + except ClientResponseError as error: + if error.status == 404: + return None + raise + return await response.json() + + async def delete(self, sandbox_id: str) -> None: + try: + response = await self._request( + "DELETE", f"/v1/sandboxes/{quote(sandbox_id, safe='')}" + ) + response.release() + except ClientResponseError as error: + if error.status != 404: + raise + + async def _request(self, method: str, path: str, **kwargs: object): + if self.session is None: + raise RuntimeError("OpenSandboxClient is not open") + return await self.session.request(method, f"{self.api_url}{path}", **kwargs) diff --git a/src/gitea_dynamic_runner/opensandbox_worker.py b/src/gitea_dynamic_runner/opensandbox_worker.py new file mode 100644 index 0000000..4bcb4ca --- /dev/null +++ b/src/gitea_dynamic_runner/opensandbox_worker.py @@ -0,0 +1,274 @@ +#!/usr/bin/env python3 +"""JetStream worker that provisions disposable OpenSandbox Kata runners.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import logging +import os +from pathlib import Path +import ssl +import uuid + +import nats +from nats.errors import TimeoutError +from nats.js.api import AckPolicy, ConsumerConfig + +from .models import RunnerRequest +from .opensandbox import OpenSandboxClient +from .pod_worker import heartbeat, identity_path +from .sandbox_kubernetes import SandboxKubernetesClient, allocated_pod_name + + +LOG = logging.getLogger(__name__) +STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER") +REQUEST_SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.vm") +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("OPENSANDBOX_NAMESPACE", "opensandbox") +OPENSANDBOX_API = os.environ.get("OPENSANDBOX_API", "http://opensandbox-server.opensandbox.svc:8080") +OPENSANDBOX_API_KEY_FILE_VALUE = os.environ.get("OPENSANDBOX_API_KEY_FILE") +OPENSANDBOX_API_KEY_FILE = Path(OPENSANDBOX_API_KEY_FILE_VALUE) if OPENSANDBOX_API_KEY_FILE_VALUE else None +OPENSANDBOX_POOL = os.environ.get("OPENSANDBOX_POOL", "ci-vm") +SANDBOX_TIMEOUT = int(os.environ.get("RUNNER_SANDBOX_TIMEOUT", str(4 * 60 * 60))) +ALLOCATION_TIMEOUT = int(os.environ.get("RUNNER_ALLOCATION_TIMEOUT", "120")) +CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "2")) +SPIFFE_TRUST_DOMAIN = os.environ.get("SPIFFE_TRUST_DOMAIN", "ddupan.top") +SPIRE_CLUSTER_NAME = os.environ.get("SPIRE_CLUSTER_NAME", "sandbox-kata") +SPIRE_CLASS_NAME = os.environ.get("SPIRE_CLASS_NAME", "spire-mgmt-spire") +RUNNER_UID = int(os.environ.get("RUNNER_UID", "2000")) + + +def sandbox_request(request: RunnerRequest, runner_name: str) -> dict[str, object]: + if request.backend != "vm": + raise ValueError("OpenSandbox backend only accepts vm requests") + path = identity_path(request.repository, request.job_name) + return { + "pool": OPENSANDBOX_POOL, + "timeout": SANDBOX_TIMEOUT, + "entrypoint": ["/usr/local/libexec/gitea-opensandbox-runner"], + "env": { + "GITEA_RUNNER_NAME": runner_name, + "GITEA_RUNNER_LABELS": "self-hosted:host,vm:host", + "CI_SPIFFE_ID": f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/{path}", + "SPIFFE_ENDPOINT_SOCKET": "unix:///run/spire/agent-sockets/spire-agent.sock", + }, + "metadata": { + "ci.ddupan.top/job-id": str(request.job_id), + "ci.ddupan.top/run-id": str(request.run_id), + "ci.ddupan.top/runner-name": runner_name, + }, + } + + +def entry_name(sandbox_id: str) -> str: + suffix = hashlib.sha256(sandbox_id.encode()).hexdigest()[:12] + return f"gitea-ci-{suffix}" + + +def identity_entry( + request: RunnerRequest, + *, + sandbox_id: str, + pod_uid: str, +) -> dict[str, object]: + path = identity_path(request.repository, request.job_name) + return { + "apiVersion": "spire.spiffe.io/v1alpha1", + "kind": "ClusterStaticEntry", + "metadata": { + "name": entry_name(sandbox_id), + "labels": { + "app.kubernetes.io/name": "gitea-dynamic-runner", + "app.kubernetes.io/component": "opensandbox-identity", + "ci.ddupan.top/sandbox-id": sandbox_id, + }, + }, + "spec": { + "className": SPIRE_CLASS_NAME, + "parentID": ( + f"spiffe://{SPIFFE_TRUST_DOMAIN}/spire/agent/k8s_psat/" + f"{SPIRE_CLUSTER_NAME}/pod/{pod_uid}" + ), + "spiffeID": f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/{path}", + "selectors": [f"unix:uid:{RUNNER_UID}"], + }, + } + + +async def wait_for_allocation( + client: SandboxKubernetesClient, sandbox_id: str +) -> tuple[str, str]: + deadline = asyncio.get_running_loop().time() + ALLOCATION_TIMEOUT + while asyncio.get_running_loop().time() < deadline: + batchsandbox = await client.get_batchsandbox(sandbox_id) + if batchsandbox is not None and (pod_name := allocated_pod_name(batchsandbox)): + pod = await client.get_pod(pod_name) + if pod is not None: + metadata = pod.get("metadata") + pod_uid = metadata.get("uid") if isinstance(metadata, dict) else None + if isinstance(pod_uid, str) and pod_uid: + return pod_name, pod_uid + await asyncio.sleep(1) + raise asyncio.TimeoutError(f"OpenSandbox {sandbox_id} allocation timed out") + + +async def wait_for_sandbox(client: OpenSandboxClient, sandbox_id: str) -> bool: + deadline = asyncio.get_running_loop().time() + SANDBOX_TIMEOUT + while asyncio.get_running_loop().time() < deadline: + sandbox = await client.get(sandbox_id) + if sandbox is None: + raise RuntimeError(f"OpenSandbox {sandbox_id} disappeared") + status = sandbox.get("status") + state = status.get("state") if isinstance(status, dict) else None + if state == "Terminated": + return True + if state == "Failed": + return False + await asyncio.sleep(2) + raise asyncio.TimeoutError(f"OpenSandbox {sandbox_id} timed out") + + +async def run_request( + message: object, + sandbox_client: OpenSandboxClient, + kubernetes_client: SandboxKubernetesClient, +) -> None: + try: + request = RunnerRequest.from_json(message.data) + except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error: + LOG.error("discarding invalid OpenSandbox request: %s", error) + await message.ack() + return + if request.backend != "vm": + LOG.error("discarding %s request received by OpenSandbox worker", request.backend) + await message.ack() + return + + runner_name = f"gitea-vm-{uuid.uuid4().hex[:12]}" + sandbox_id: str | None = None + static_entry: str | None = None + stop = asyncio.Event() + pulse = asyncio.create_task(heartbeat(message, stop)) + try: + create = await sandbox_client.create(**sandbox_request(request, runner_name)) + candidate = create.get("id") + if not isinstance(candidate, str) or not candidate: + raise RuntimeError("OpenSandbox create response has no id") + sandbox_id = candidate + pod_name, pod_uid = await wait_for_allocation(kubernetes_client, sandbox_id) + manifest = identity_entry(request, sandbox_id=sandbox_id, pod_uid=pod_uid) + static_entry = entry_name(sandbox_id) + await kubernetes_client.create_entry(manifest) + LOG.info( + "OpenSandbox identity created sandbox=%s pod=%s pod_uid=%s entry=%s runner=%s job_id=%d", + sandbox_id, pod_name, pod_uid, static_entry, runner_name, request.job_id, + ) + if await wait_for_sandbox(sandbox_client, sandbox_id): + await message.ack() + else: + await message.nak(delay=30) + except Exception: + LOG.exception("OpenSandbox runner failed sandbox=%s job_id=%d", sandbox_id, request.job_id) + await message.nak(delay=30) + raise + finally: + stop.set() + await pulse + try: + if static_entry is not None: + await kubernetes_client.delete_entry(static_entry) + finally: + if sandbox_id is not None: + await sandbox_client.delete(sandbox_id) + + +async def consume_requests( + subscription: object, + sandbox_client: OpenSandboxClient, + kubernetes_client: SandboxKubernetesClient, +) -> 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, sandbox_client, kubernetes_client) + ) + task.add_done_callback(_report_task) + active.add(task) + + +def _report_task(task: asyncio.Task[None]) -> None: + if not task.cancelled() and (error := task.exception()) is not None: + LOG.error( + "OpenSandbox 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="opensandbox-runner-worker", + ) + js = nc.jetstream() + subscription = await js.pull_subscribe( + REQUEST_SUBJECT, + durable="vm", + stream=STREAM, + config=ConsumerConfig( + durable_name="vm", + filter_subject=REQUEST_SUBJECT, + ack_policy=AckPolicy.EXPLICIT, + ack_wait=5 * 60, + max_ack_pending=max(CAPACITY, 1), + max_deliver=5, + ), + ) + async with ( + OpenSandboxClient( + api_url=OPENSANDBOX_API, + api_key_file=OPENSANDBOX_API_KEY_FILE, + ) as sandbox_client, + SandboxKubernetesClient( + api_url=KUBERNETES_API, + token_file=KUBERNETES_TOKEN_FILE, + ca_file=KUBERNETES_CA_FILE, + namespace=NAMESPACE, + ) as kubernetes_client, + ): + try: + await consume_requests(subscription, sandbox_client, kubernetes_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/src/gitea_dynamic_runner/sandbox_kubernetes.py b/src/gitea_dynamic_runner/sandbox_kubernetes.py new file mode 100644 index 0000000..8a6551b --- /dev/null +++ b/src/gitea_dynamic_runner/sandbox_kubernetes.py @@ -0,0 +1,132 @@ +"""Kubernetes resources that bind an OpenSandbox Kata guest to SPIRE.""" + +from __future__ import annotations + +import json +from pathlib import Path +import ssl +from urllib.parse import quote + +from aiohttp import ClientResponseError, ClientSession, TCPConnector + + +class SandboxKubernetesClient: + def __init__( + self, + *, + api_url: str, + token_file: Path, + ca_file: Path, + namespace: str, + ) -> None: + self.api_url = api_url.rstrip("/") + self.token_file = token_file + self.ca_file = ca_file + self.namespace = namespace + self.session: ClientSession | None = None + + async def __aenter__(self) -> SandboxKubernetesClient: + context = ssl.create_default_context(cafile=self.ca_file) + self.session = ClientSession( + connector=TCPConnector(ssl=context), + headers={ + "Authorization": f"Bearer {self.token_file.read_text().strip()}" + }, + raise_for_status=True, + ) + return self + + async def __aexit__(self, *_: object) -> None: + if self.session is not None: + await self.session.close() + + async def get_batchsandbox(self, sandbox_id: str) -> dict[str, object] | None: + try: + response = await self._request( + "GET", f"{self._batchsandboxes_path()}/{quote(sandbox_id, safe='')}" + ) + except ClientResponseError as error: + if error.status == 404: + return None + raise + return await response.json() + + async def get_pod(self, name: str) -> dict[str, object] | None: + try: + response = await self._request( + "GET", f"{self._pods_path()}/{quote(name, safe='')}" + ) + except ClientResponseError as error: + if error.status == 404: + return None + raise + return await response.json() + + async def create_entry(self, manifest: dict[str, object]) -> dict[str, object]: + response = await self._request( + "POST", self._entries_path(), json=manifest + ) + return await response.json() + + async def get_entry(self, name: str) -> dict[str, object] | None: + try: + response = await self._request( + "GET", f"{self._entries_path()}/{quote(name, safe='')}" + ) + except ClientResponseError as error: + if error.status == 404: + return None + raise + return await response.json() + + async def delete_entry(self, name: str) -> None: + try: + response = await self._request( + "DELETE", + f"{self._entries_path()}/{quote(name, safe='')}", + json={"propagationPolicy": "Background"}, + ) + response.release() + except ClientResponseError as error: + if error.status != 404: + raise + + async def _request(self, method: str, path: str, **kwargs: object): + if self.session is None: + raise RuntimeError("SandboxKubernetesClient is not open") + return await self.session.request(method, f"{self.api_url}{path}", **kwargs) + + def _batchsandboxes_path(self) -> str: + return ( + "/apis/sandbox.opensandbox.io/v1alpha1/namespaces/" + f"{quote(self.namespace, safe='')}/batchsandboxes" + ) + + def _pods_path(self) -> str: + return f"/api/v1/namespaces/{quote(self.namespace, safe='')}/pods" + + @staticmethod + def _entries_path() -> str: + return "/apis/spire.spiffe.io/v1alpha1/clusterstaticentries" + + +def allocated_pod_name(batchsandbox: dict[str, object]) -> str | None: + metadata = batchsandbox.get("metadata") + if not isinstance(metadata, dict): + return None + annotations = metadata.get("annotations") + if not isinstance(annotations, dict): + return None + raw = annotations.get("sandbox.opensandbox.io/alloc-status") + if not isinstance(raw, str): + return None + try: + allocation = json.loads(raw) + except json.JSONDecodeError: + return None + if not isinstance(allocation, dict): + return None + pods = allocation.get("pods") + if not isinstance(pods, list) or len(pods) != 1 or not isinstance(pods[0], str): + return None + return pods[0] diff --git a/tests/test_opensandbox_worker.py b/tests/test_opensandbox_worker.py new file mode 100644 index 0000000..9fef46e --- /dev/null +++ b/tests/test_opensandbox_worker.py @@ -0,0 +1,114 @@ +import asyncio +import json + +import pytest + +from gitea_dynamic_runner.models import RunnerRequest +from gitea_dynamic_runner import opensandbox_worker +from gitea_dynamic_runner.sandbox_kubernetes import allocated_pod_name + + +def request() -> RunnerRequest: + return RunnerRequest( + job_id=47, + run_id=12, + backend="vm", + repository="panxiao81/example", + job_name="publish image", + labels=("self-hosted", "vm"), + ) + + +def test_sandbox_request_uses_pool_and_identity_gate(monkeypatch): + monkeypatch.setattr(opensandbox_worker, "OPENSANDBOX_POOL", "ci-vm") + document = opensandbox_worker.sandbox_request(request(), "gitea-vm-abcd") + assert document["pool"] == "ci-vm" + assert document["entrypoint"] == [ + "/usr/local/libexec/gitea-opensandbox-runner" + ] + assert document["env"]["GITEA_RUNNER_NAME"] == "gitea-vm-abcd" + assert document["env"]["GITEA_RUNNER_LABELS"] == "self-hosted:host,vm:host" + assert document["env"]["CI_SPIFFE_ID"].startswith( + "spiffe://ddupan.top/ci/panxiao81/example/publish-image-" + ) + assert document["metadata"]["ci.ddupan.top/job-id"] == "47" + + +def test_identity_entry_is_bound_to_kata_pod_uid(monkeypatch): + monkeypatch.setattr(opensandbox_worker, "SPIRE_CLUSTER_NAME", "sandbox-kata") + monkeypatch.setattr(opensandbox_worker, "RUNNER_UID", 2000) + manifest = opensandbox_worker.identity_entry( + request(), sandbox_id="sandbox-123", pod_uid="pod-uid-456" + ) + assert manifest["metadata"]["name"] == opensandbox_worker.entry_name( + "sandbox-123" + ) + spec = manifest["spec"] + assert spec["parentID"] == ( + "spiffe://ddupan.top/spire/agent/k8s_psat/" + "sandbox-kata/pod/pod-uid-456" + ) + assert spec["selectors"] == ["unix:uid:2000"] + assert spec["spiffeID"].startswith( + "spiffe://ddupan.top/ci/panxiao81/example/publish-image-" + ) + + +@pytest.mark.parametrize( + ("annotation", "expected"), + [ + (json.dumps({"pods": ["pool-pod-1"], "poolRef": "ci-vm"}), "pool-pod-1"), + (json.dumps({"pods": []}), None), + (json.dumps({"pods": ["one", "two"]}), None), + ("not-json", None), + ], +) +def test_allocated_pod_name(annotation, expected): + batchsandbox = { + "metadata": { + "annotations": {"sandbox.opensandbox.io/alloc-status": annotation} + } + } + assert allocated_pod_name(batchsandbox) == expected + + +class FakeKubernetesClient: + def __init__(self): + self.calls = 0 + + async def get_batchsandbox(self, sandbox_id): + self.calls += 1 + if self.calls == 1: + return {"metadata": {"annotations": {}}} + return { + "metadata": { + "annotations": { + "sandbox.opensandbox.io/alloc-status": json.dumps( + {"pods": ["pool-pod-1"], "poolRef": "ci-vm"} + ) + } + } + } + + async def get_pod(self, name): + assert name == "pool-pod-1" + return {"metadata": {"uid": "pod-uid-1"}} + + +async def test_wait_for_allocation_returns_real_pod_uid(monkeypatch): + async def no_sleep(_): + return None + + monkeypatch.setattr(asyncio, "sleep", no_sleep) + client = FakeKubernetesClient() + assert await opensandbox_worker.wait_for_allocation(client, "sandbox-1") == ( + "pool-pod-1", + "pod-uid-1", + ) + + +def test_entry_name_is_stable_and_dns_safe(): + name = opensandbox_worker.entry_name("sandbox/with unsafe characters") + assert name == opensandbox_worker.entry_name("sandbox/with unsafe characters") + assert name.startswith("gitea-ci-") + assert "/" not in name -- 2.54.0 From 2679f81cdd1af9b379291b0cd87852450f6cff81 Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Fri, 18 Sep 2026 01:09:37 +0000 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20=E4=BD=BF=E7=94=A8=20OpenSandbox=20c?= =?UTF-8?q?hart=20=E6=9C=8D=E5=8A=A1=E5=9C=B0=E5=9D=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/gitea_dynamic_runner/opensandbox_worker.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/gitea_dynamic_runner/opensandbox_worker.py b/src/gitea_dynamic_runner/opensandbox_worker.py index 4bcb4ca..deb407d 100644 --- a/src/gitea_dynamic_runner/opensandbox_worker.py +++ b/src/gitea_dynamic_runner/opensandbox_worker.py @@ -33,7 +33,9 @@ KUBERNETES_API = os.environ.get("KUBERNETES_API", "https://kubernetes.default.sv KUBERNETES_TOKEN_FILE = Path(os.environ.get("KUBERNETES_TOKEN_FILE", "/var/run/secrets/kubernetes.io/serviceaccount/token")) KUBERNETES_CA_FILE = Path(os.environ.get("KUBERNETES_CA_FILE", "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt")) NAMESPACE = os.environ.get("OPENSANDBOX_NAMESPACE", "opensandbox") -OPENSANDBOX_API = os.environ.get("OPENSANDBOX_API", "http://opensandbox-server.opensandbox.svc:8080") +OPENSANDBOX_API = os.environ.get( + "OPENSANDBOX_API", "http://opensandbox-server.opensandbox-system.svc" +) OPENSANDBOX_API_KEY_FILE_VALUE = os.environ.get("OPENSANDBOX_API_KEY_FILE") OPENSANDBOX_API_KEY_FILE = Path(OPENSANDBOX_API_KEY_FILE_VALUE) if OPENSANDBOX_API_KEY_FILE_VALUE else None OPENSANDBOX_POOL = os.environ.get("OPENSANDBOX_POOL", "ci-vm") -- 2.54.0