恢复原生 Kubernetes Pod 调度

README.md

@@ -17,7 +17,8 @@ runs-on: [self-hosted, vm]
 
 组件：
 
-- `controller`：接收 Gitea `workflow_job` webhook，直接调用 OpenSandbox Lifecycle
+- `controller`：接收 Gitea `workflow_job` webhook，将任务持久化到 NATS JetStream；仅在
+  显式启用 VM consumer 时调用 OpenSandbox Lifecycle
   API，从 `ci-pod` 或 `ci-vm` Pool 创建一次性环境。
 - `microvm-runner-launch`：为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP，运行
   Cloud Hypervisor，退出后完整清理。
@@ -32,7 +33,8 @@ runs-on: [self-hosted, vm]
 - `jwt-broker`：早期共享 Kubernetes runner 的过渡实验；目标架构不部署它，每个
   动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
 
-OpenSandbox 路径不使用 NATS。旧 Pod/microVM worker 只作为迁移期代码保留，不应重新
+Pod 路径由 homelab 集群中的 `pod-worker` 直接创建 Kubernetes Pod。OpenSandbox 只用于
+VM/Kata workload；两个 backend 使用独立 durable consumer，任一执行层故障不会阻塞另一条
 部署。长期 RunnerService 协议路线见
 [`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
 

docs/opensandbox-runner.md

@@ -1,8 +1,10 @@
-# OpenSandbox runner
+# OpenSandbox VM runner
 
 Gitea 的 `workflow_job` webhook 只负责发现带 `self-hosted,pod` 或
 `self-hosted,vm` label 的 queued job。controller 不再把请求写入 NATS，而是携带由
-homelab ExternalSecret 挂载的 API key 直接调用 OpenSandbox Lifecycle API：
+OpenSandbox 只承载 `self-hosted,vm` workload。`self-hosted,pod` 由 homelab 原生
+Kubernetes worker 创建，不经过 OpenSandbox。homelab ExternalSecret 挂载的 API key
+供 VM consumer 调用 OpenSandbox Lifecycle API：
 
 ```text
 http://10.60.0.13:8080/v1/sandboxes
@@ -19,7 +21,7 @@ Lifecycle 请求把稳定的 repository/task SPIFFE ID 放入 task environment
 集群内的 `opensandbox-identity` controller 读取 BatchSandbox allocation 得到实际
 Pod UID，然后创建：
 
-- parent：`spiffe://ddupan.top/spire/agent/k8s_psat/sandbox-kata/<pod-uid>`；
+- parent：`spiffe://ddupan.top/spire/agent/k8s_psat/sandbox-kata/pod/<pod-uid>`；
 - workload：`spiffe://ddupan.top/ci/<owner>/<repository>/<task>`；
 - selector：`unix:uid:2000`。
 

src/gitea_dynamic_runner/controller.py

@@ -1,18 +1,31 @@
 #!/usr/bin/env python3
-"""Gitea workflow_job webhook to the OpenSandbox Lifecycle API."""
+"""Persist Gitea workflow_job events and optionally schedule VM sandboxes."""
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import hmac
 import json
 import logging
 import os
+import ssl
 from pathlib import Path
 
+import nats
 from aiohttp import web
+from nats.errors import TimeoutError as NatsTimeoutError
+from nats.js.api import (
+    AckPolicy,
+    ConsumerConfig,
+    DiscardPolicy,
+    RetentionPolicy,
+    StorageType,
+    StreamConfig,
+)
+from nats.js.errors import NotFoundError
 
-from .models import RunnerRequest
+from .models import IdentityBinding, RunnerRequest
 from .opensandbox import OpenSandboxClient
 from .opensandbox_worker import OpenSandboxScheduler, RegistrationTokens
 
@@ -22,19 +35,29 @@ WEBHOOK_SECRET_FILE = Path(
     os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret")
 )
 REGISTRATION_TOKEN_FILE = Path(
-    os.environ.get(
-        "REGISTRATION_TOKEN_FILE", "/run/secrets/gitea/registration-token"
-    )
-)
-OPENSANDBOX_API = os.environ.get(
-    "OPENSANDBOX_API", "http://10.60.0.13:8080"
+    os.environ.get("REGISTRATION_TOKEN_FILE", "/run/secrets/gitea/registration-token")
 )
+OPENSANDBOX_API = os.environ.get("OPENSANDBOX_API", "http://10.60.0.13:8080")
 OPENSANDBOX_API_KEY_FILE = Path(
     os.environ.get(
         "OPENSANDBOX_API_KEY_FILE",
         "/run/secrets/opensandbox/api-key",
     )
 )
+NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
+NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt")
+NATS_PRODUCER_USER = os.environ.get("NATS_PRODUCER_USER", "ci-producer")
+NATS_PRODUCER_PASSWORD_FILE = Path(
+    os.environ.get("NATS_PRODUCER_PASSWORD_FILE", "/run/secrets/nats/producer-password")
+)
+NATS_WORKER_USER = os.environ.get("NATS_WORKER_USER", "ci-worker")
+NATS_WORKER_PASSWORD_FILE = Path(
+    os.environ.get("NATS_WORKER_PASSWORD_FILE", "/run/secrets/nats/worker-password")
+)
+NATS_STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
+NATS_SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
+VM_CONSUMER_ENABLED = os.environ.get("VM_CONSUMER_ENABLED", "false") == "true"
+VM_CAPACITY = int(os.environ.get("VM_CAPACITY", "1"))
 
 
 def accepts(payload: object) -> tuple[bool, str | None]:
@@ -60,25 +83,55 @@ async def webhook(request: web.Request) -> web.Response:
     context = _webhook_context(payload)
     LOG.info("workflow_job webhook received %s", context)
 
-    runner_request = RunnerRequest.from_webhook(payload)
-    if runner_request is None:
-        LOG.info("workflow_job webhook ignored %s", context)
+    if isinstance(payload, dict) and payload.get("action") == "completed":
+        job = payload.get("workflow_job")
+        runner_name = job.get("runner_name") if isinstance(job, dict) else None
+        scheduler = request.app.get("scheduler")
+        if (
+            isinstance(runner_name, str)
+            and runner_name.startswith("gitea-vm-")
+            and scheduler is not None
+        ):
+            cleaned = await scheduler.complete(runner_name)
+            LOG.info(
+                "completed runner cleanup runner=%r cleaned=%s %s",
+                runner_name,
+                cleaned,
+                context,
+            )
         return web.Response(status=204)
 
-    scheduler: OpenSandboxScheduler = request.app["scheduler"]
-    try:
-        sandbox_id = await scheduler.create(runner_request)
-    except ValueError as error:
-        LOG.info("duplicate workflow_job webhook ignored: %s", error)
-        return web.Response(status=202, text="already scheduled\n")
-    except Exception:
-        LOG.exception("failed to create OpenSandbox runner %s", context)
-        raise web.HTTPServiceUnavailable(text="sandbox unavailable\n")
-    return web.Response(status=202, text=f"sandbox={sandbox_id}\n")
+    runner_request = RunnerRequest.from_webhook(payload)
+    if runner_request is None:
+        binding = IdentityBinding.from_webhook(payload)
+        if binding is None:
+            LOG.info("workflow_job webhook ignored %s", context)
+            return web.Response(status=204)
+        subject = f"{NATS_SUBJECT_PREFIX}.{binding.backend}.binding"
+        await request.app["js"].publish(
+            subject,
+            binding.to_json(),
+            headers={
+                "Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"
+            },
+        )
+        LOG.info("runner identity binding persisted subject=%s %s", subject, context)
+        return web.Response(status=202, text="binding queued\n")
+
+    subject = f"{NATS_SUBJECT_PREFIX}.{runner_request.backend}"
+    await request.app["js"].publish(
+        subject,
+        runner_request.to_json(),
+        headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
+    )
+    LOG.info("runner request persisted subject=%s %s", subject, context)
+    return web.Response(status=202, text="queued\n")
 
 
 async def registration_token(request: web.Request) -> web.Response:
-    scheduler: OpenSandboxScheduler = request.app["scheduler"]
+    scheduler = request.app.get("scheduler")
+    if scheduler is None:
+        raise web.HTTPNotFound()
     value = await scheduler.tokens.consume(request.match_info["nonce"])
     if value is None:
         raise web.HTTPNotFound()
@@ -101,15 +154,119 @@ def _webhook_context(payload: object) -> str:
 
 
 async def health(request: web.Request) -> web.Response:
-    client: OpenSandboxClient = request.app["opensandbox_client"]
+    connected = request.app["nc"].is_connected
     return web.Response(
-        text="ok\n" if client.session is not None else "disconnected\n",
-        status=200 if client.session is not None else 503,
+        text="ok\n" if connected else "disconnected\n",
+        status=200 if connected else 503,
     )
 
 
-async def opensandbox_context(app: web.Application):
+async def ensure_stream(js: object) -> None:
+    config = StreamConfig(
+        name=NATS_STREAM,
+        subjects=[f"{NATS_SUBJECT_PREFIX}.>"],
+        retention=RetentionPolicy.WORK_QUEUE,
+        storage=StorageType.FILE,
+        discard=DiscardPolicy.OLD,
+        max_age=24 * 60 * 60,
+        max_msgs=10_000,
+        max_bytes=256 * 1024 * 1024,
+        duplicate_window=24 * 60 * 60,
+    )
+    try:
+        await js.stream_info(NATS_STREAM)
+    except NotFoundError:
+        await js.add_stream(config=config)
+    else:
+        await js.update_stream(config=config)
+
+
+async def run_message(message: object, scheduler: OpenSandboxScheduler) -> None:
+    try:
+        runner_request = RunnerRequest.from_json(message.data)
+        await scheduler.create(runner_request)
+        task = scheduler.active[runner_request.job_id]
+        try:
+            while not task.done():
+                try:
+                    await asyncio.wait_for(asyncio.shield(task), timeout=30)
+                except asyncio.TimeoutError:
+                    await message.in_progress()
+            await task
+        except asyncio.CancelledError:
+            # A completed webhook cancels the lifecycle monitor after its
+            # Lifecycle DELETE succeeds. That is successful message handling.
+            # During controller shutdown the scheduler still owns the job, so
+            # preserve the unacked message for redelivery.
+            if runner_request.job_id in scheduler.active:
+                raise
+    except ValueError as error:
+        LOG.info("runner request already active: %s", error)
+        await message.nak(delay=5)
+    except Exception:
+        LOG.exception("persistent runner request failed")
+        await message.nak(delay=15)
+    else:
+        await message.ack()
+
+
+async def consume_vm_requests(js: object, scheduler: OpenSandboxScheduler) -> None:
+    subscription = await js.pull_subscribe(
+        f"{NATS_SUBJECT_PREFIX}.vm",
+        durable="vm",
+        stream=NATS_STREAM,
+        config=ConsumerConfig(
+            durable_name="vm",
+            filter_subject=f"{NATS_SUBJECT_PREFIX}.vm",
+            ack_policy=AckPolicy.EXPLICIT,
+            ack_wait=5 * 60,
+            max_ack_pending=VM_CAPACITY,
+            max_deliver=20,
+        ),
+    )
+    active: set[asyncio.Task[None]] = set()
+    while True:
+        active = {task for task in active if not task.done()}
+        try:
+            messages = await subscription.fetch(
+                batch=max(1, VM_CAPACITY - len(active)), timeout=5
+            )
+        except NatsTimeoutError:
+            continue
+        for message in messages:
+            task = asyncio.create_task(run_message(message, scheduler))
+            active.add(task)
+
+
+async def runtime_context(app: web.Application):
+    tls = ssl.create_default_context(cafile=NATS_CA_FILE)
+    producer = await nats.connect(
+        NATS_URL,
+        user=NATS_PRODUCER_USER,
+        password=NATS_PRODUCER_PASSWORD_FILE.read_text().strip(),
+        tls=tls,
+        name="opensandbox-runner-controller-producer",
+    )
+    app["nc"] = producer
+    app["js"] = producer.jetstream()
+    await ensure_stream(app["js"])
+    if not VM_CONSUMER_ENABLED:
+        try:
+            yield
+        finally:
+            await producer.drain()
+        return
+
+    if VM_CAPACITY < 1:
+        raise ValueError("VM_CAPACITY must be at least 1")
     tokens = RegistrationTokens(REGISTRATION_TOKEN_FILE.read_bytes())
+    worker = await nats.connect(
+        NATS_URL,
+        user=NATS_WORKER_USER,
+        password=NATS_WORKER_PASSWORD_FILE.read_text().strip(),
+        tls=ssl.create_default_context(cafile=NATS_CA_FILE),
+        name="opensandbox-vm-runner-worker",
+    )
     async with OpenSandboxClient(
         api_url=OPENSANDBOX_API,
         api_key_file=OPENSANDBOX_API_KEY_FILE,
@@ -117,13 +274,23 @@ async def opensandbox_context(app: web.Application):
         scheduler = OpenSandboxScheduler(client, tokens)
         app["opensandbox_client"] = client
         app["scheduler"] = scheduler
-        yield
-        await scheduler.close()
+        consumer = asyncio.create_task(
+            consume_vm_requests(worker.jetstream(), scheduler),
+            name="opensandbox-vm-consumer",
+        )
+        try:
+            yield
+        finally:
+            consumer.cancel()
+            await asyncio.gather(consumer, return_exceptions=True)
+            await scheduler.close()
+            await worker.drain()
+            await producer.drain()
 
 
 def create_app() -> web.Application:
     app = web.Application(client_max_size=1024 * 1024)
-    app.cleanup_ctx.append(opensandbox_context)
+    app.cleanup_ctx.append(runtime_context)
     app.router.add_post("/webhook", webhook)
     app.router.add_get("/token/{nonce}", registration_token)
     app.router.add_get("/healthz", health)

src/gitea_dynamic_runner/opensandbox_identity.py

@@ -91,7 +91,7 @@ def identity_entry(
             "className": SPIRE_CLASS_NAME,
             "parentID": (
                 f"spiffe://{SPIFFE_TRUST_DOMAIN}/spire/agent/k8s_psat/"
-                f"{SPIRE_CLUSTER_NAME}/{pod_uid}"
+                f"{SPIRE_CLUSTER_NAME}/pod/{pod_uid}"
             ),
             "spiffeID": spiffe_id,
             "selectors": [f"unix:uid:{RUNNER_UID}"],

src/gitea_dynamic_runner/opensandbox_worker.py

@@ -88,6 +88,7 @@ class OpenSandboxScheduler:
         self.tokens = tokens
         self.on_finished = on_finished
         self.active: dict[int, asyncio.Task[None]] = {}
+        self.runners: dict[str, tuple[int, str, str]] = {}
 
     async def create(self, request: RunnerRequest) -> str:
         if request.job_id in self.active:
@@ -106,11 +107,12 @@ class OpenSandboxScheduler:
             raise
 
         task = asyncio.create_task(
-            self._monitor(request, sandbox_id, nonce),
+            self._monitor(request, sandbox_id, nonce, runner_name),
             name=f"opensandbox-{sandbox_id}",
         )
         task.add_done_callback(self._report)
         self.active[request.job_id] = task
+        self.runners[runner_name] = (request.job_id, sandbox_id, nonce)
         LOG.info(
             "OpenSandbox runner created sandbox=%s runner=%s job_id=%d "
             "repository=%s job_name=%r pool=ci-%s",
@@ -124,7 +126,11 @@ class OpenSandboxScheduler:
         return sandbox_id
 
     async def _monitor(
-        self, request: RunnerRequest, sandbox_id: str, nonce: str
+        self,
+        request: RunnerRequest,
+        sandbox_id: str,
+        nonce: str,
+        runner_name: str,
     ) -> None:
         try:
             deadline = asyncio.get_running_loop().time() + SANDBOX_TIMEOUT
@@ -139,13 +145,38 @@ class OpenSandboxScheduler:
                 await asyncio.sleep(2)
             raise asyncio.TimeoutError(f"OpenSandbox {sandbox_id} timed out")
         finally:
-            await self.tokens.revoke(nonce)
-            try:
-                await self.client.delete(sandbox_id)
-            finally:
-                self.active.pop(request.job_id, None)
-                if self.on_finished is not None:
-                    self.on_finished(request.job_id)
+            await self._cleanup(request.job_id, sandbox_id, nonce, runner_name)
+
+    async def complete(self, runner_name: str) -> bool:
+        """Delete the sandbox which actually ran a completed Gitea job."""
+        state = self.runners.get(runner_name)
+        if state is None:
+            return False
+        job_id, sandbox_id, nonce = state
+        task = self.active.get(job_id)
+        if task is not None:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+        await self._cleanup(job_id, sandbox_id, nonce, runner_name)
+        return True
+
+    async def _cleanup(
+        self,
+        job_id: int,
+        sandbox_id: str,
+        nonce: str,
+        runner_name: str,
+    ) -> None:
+        """Revoke and delete once, including cancellation-before-start races."""
+        if self.runners.pop(runner_name, None) is None:
+            return
+        await self.tokens.revoke(nonce)
+        try:
+            await self.client.delete(sandbox_id)
+        finally:
+            self.active.pop(job_id, None)
+            if self.on_finished is not None:
+                self.on_finished(job_id)
 
     @staticmethod
     def _report(task: asyncio.Task[None]) -> None:

tests/test_controller.py

@@ -1,3 +1,4 @@
+import asyncio
 import hashlib
 import hmac
 import json
@@ -31,8 +32,14 @@ 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(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)
 
 
@@ -95,3 +102,40 @@ def test_signature_accepts_gitea_and_prefixed_forms(tmp_path, monkeypatch):
     assert controller.valid_signature(body, digest)
     assert controller.valid_signature(body, f"sha256={digest}")
     assert not controller.valid_signature(body, "bad")
+
+
+async def test_completed_lifecycle_cancellation_acks_message():
+    request = RunnerRequest.from_webhook(queued_job())
+    assert request is not None
+
+    class Message:
+        data = request.to_json()
+        acked = False
+
+        async def ack(self):
+            self.acked = True
+
+        async def nak(self, **kwargs):
+            raise AssertionError(f"unexpected NAK: {kwargs}")
+
+        async def in_progress(self):
+            pass
+
+    class Scheduler:
+        active = {}
+
+        async def create(self, runner_request):
+            async def completed():
+                self.active.pop(runner_request.job_id, None)
+                raise asyncio.CancelledError
+
+            self.active[runner_request.job_id] = asyncio.create_task(completed())
+
+    message = Message()
+    await controller.run_message(message, Scheduler())
+    assert message.acked is True
+
+
+def test_opensandbox_consumer_is_vm_only():
+    assert controller.VM_CONSUMER_ENABLED is False
+    assert controller.VM_CAPACITY == 1

tests/test_opensandbox_identity.py

@@ -49,5 +49,5 @@ def test_entry_binds_exact_pod_agent_and_runner_uid():
         pod_uid="pod-uid",
         spiffe_id="spiffe://ddupan.top/ci/org/repo/test",
     )
-    assert entry["spec"]["parentID"].endswith("/sandbox-kata/pod-uid")
+    assert entry["spec"]["parentID"].endswith("/sandbox-kata/pod/pod-uid")
     assert entry["spec"]["selectors"] == ["unix:uid:2000"]

tests/test_opensandbox_worker.py

@@ -26,13 +26,9 @@ def test_sandbox_request_uses_pool_identity_and_one_time_token_url():
         request(), "gitea-vm-abcd", "http://scheduler/token/nonce"
     )
     assert document["pool"] == "ci-vm"
-    assert document["entrypoint"] == [
-        "/usr/local/libexec/gitea-opensandbox-runner"
-    ]
+    assert document["entrypoint"] == ["/usr/local/libexec/gitea-opensandbox-runner"]
     assert document["env"]["GITEA_RUNNER_LABELS"] == "self-hosted:host,vm:host"
-    assert document["env"]["GITEA_RUNNER_REGISTRATION_TOKEN_URL"].endswith(
-        "/nonce"
-    )
+    assert document["env"]["GITEA_RUNNER_REGISTRATION_TOKEN_URL"].endswith("/nonce")
     assert document["env"]["CI_SPIFFE_ID"] == (
         "spiffe://ddupan.top/ci/panxiao81/example/publish-image-3e72cdc4a97e"
     )
@@ -86,3 +82,20 @@ async def test_scheduler_rejects_duplicate_active_job():
     scheduler.active[42] = asyncio.Future()
     with pytest.raises(ValueError, match="already"):
         await scheduler.create(request())
+
+
+async def test_scheduler_deletes_sandbox_for_completed_runner():
+    class RunningOpenSandbox(FakeOpenSandbox):
+        async def get(self, sandbox_id):
+            return {"status": {"state": "Running"}}
+
+    client = RunningOpenSandbox()
+    scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
+    await scheduler.create(request())
+    runner_name = next(iter(scheduler.runners))
+
+    assert await scheduler.complete(runner_name) is True
+    assert client.deleted == ["sandbox-1"]
+    assert scheduler.active == {}
+    assert scheduler.runners == {}
+    assert await scheduler.complete(runner_name) is False