恢复原生 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

src/gitea_dynamic_runner/controller.py

@@ -1,5 +1,5 @@
 #!/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
 
@@ -25,7 +25,7 @@ from nats.js.api import (
 )
 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
 
@@ -56,7 +56,8 @@ NATS_WORKER_PASSWORD_FILE = Path(
 )
 NATS_STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
 NATS_SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
-POD_CONSUMER_ENABLED = os.environ.get("POD_CONSUMER_ENABLED", "true") == "true"
+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]:
@@ -82,21 +83,16 @@ async def webhook(request: web.Request) -> web.Response:
     context = _webhook_context(payload)
     LOG.info("workflow_job webhook received %s", context)
 
-    scheduler: OpenSandboxScheduler = request.app["scheduler"]
     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
-        if isinstance(runner_name, str) and runner_name.startswith(
-            ("gitea-pod-", "gitea-vm-")
+        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)
-            await request.app["js"].publish(
-                f"{NATS_SUBJECT_PREFIX}.lifecycle.completed",
-                body,
-                headers={
-                    "Nats-Msg-Id": (f"gitea-workflow-job-{job.get('id')}-completed")
-                },
-            )
             LOG.info(
                 "completed runner cleanup runner=%r cleaned=%s %s",
                 runner_name,
@@ -107,8 +103,20 @@ async def webhook(request: web.Request) -> web.Response:
 
     runner_request = RunnerRequest.from_webhook(payload)
     if runner_request is None:
-        LOG.info("workflow_job webhook ignored %s", context)
-        return web.Response(status=204)
+        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(
@@ -121,7 +129,9 @@ async def webhook(request: web.Request) -> web.Response:
 
 
 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()
@@ -144,10 +154,10 @@ 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,
     )
 
 
@@ -176,12 +186,20 @@ async def run_message(message: object, scheduler: OpenSandboxScheduler) -> None:
         runner_request = RunnerRequest.from_json(message.data)
         await scheduler.create(runner_request)
         task = scheduler.active[runner_request.job_id]
-        while not task.done():
-            try:
-                await asyncio.wait_for(asyncio.shield(task), timeout=30)
-            except asyncio.TimeoutError:
-                await message.in_progress()
-        await task
+        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)
@@ -192,19 +210,17 @@ async def run_message(message: object, scheduler: OpenSandboxScheduler) -> None:
         await message.ack()
 
 
-async def consume_pod_requests(js: object, scheduler: OpenSandboxScheduler) -> None:
+async def consume_vm_requests(js: object, scheduler: OpenSandboxScheduler) -> None:
     subscription = await js.pull_subscribe(
-        f"{NATS_SUBJECT_PREFIX}.pod",
-        # Reuse the existing durable so no pending Pod request is stranded
-        # during the in-process consumer migration.
-        durable="pod",
+        f"{NATS_SUBJECT_PREFIX}.vm",
+        durable="vm",
         stream=NATS_STREAM,
         config=ConsumerConfig(
-            durable_name="pod",
-            filter_subject=f"{NATS_SUBJECT_PREFIX}.pod",
+            durable_name="vm",
+            filter_subject=f"{NATS_SUBJECT_PREFIX}.vm",
             ack_policy=AckPolicy.EXPLICIT,
             ack_wait=5 * 60,
-            max_ack_pending=4,
+            max_ack_pending=VM_CAPACITY,
             max_deliver=20,
         ),
     )
@@ -213,7 +229,7 @@ async def consume_pod_requests(js: object, scheduler: OpenSandboxScheduler) -> N
         active = {task for task in active if not task.done()}
         try:
             messages = await subscription.fetch(
-                batch=max(1, 4 - len(active)), timeout=5
+                batch=max(1, VM_CAPACITY - len(active)), timeout=5
             )
         except NatsTimeoutError:
             continue
@@ -223,7 +239,6 @@ async def consume_pod_requests(js: object, scheduler: OpenSandboxScheduler) -> N
 
 
 async def runtime_context(app: web.Application):
-    tokens = RegistrationTokens(REGISTRATION_TOKEN_FILE.read_bytes())
     tls = ssl.create_default_context(cafile=NATS_CA_FILE)
     producer = await nats.connect(
         NATS_URL,
@@ -232,16 +247,26 @@ async def runtime_context(app: web.Application):
         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-runner-controller-worker",
+        name="opensandbox-vm-runner-worker",
     )
-    app["nc"] = producer
-    app["js"] = producer.jetstream()
-    await ensure_stream(app["js"])
     async with OpenSandboxClient(
         api_url=OPENSANDBOX_API,
         api_key_file=OPENSANDBOX_API_KEY_FILE,
@@ -249,18 +274,15 @@ async def runtime_context(app: web.Application):
         scheduler = OpenSandboxScheduler(client, tokens)
         app["opensandbox_client"] = client
         app["scheduler"] = scheduler
-        consumer = None
-        if POD_CONSUMER_ENABLED:
-            consumer = asyncio.create_task(
-                consume_pod_requests(worker.jetstream(), scheduler),
-                name="opensandbox-pod-consumer",
-            )
+        consumer = asyncio.create_task(
+            consume_vm_requests(worker.jetstream(), scheduler),
+            name="opensandbox-vm-consumer",
+        )
         try:
             yield
         finally:
-            if consumer is not None:
-                consumer.cancel()
-                await asyncio.gather(consumer, return_exceptions=True)
+            consumer.cancel()
+            await asyncio.gather(consumer, return_exceptions=True)
             await scheduler.close()
             await worker.drain()
             await producer.drain()

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