diff --git a/README.md b/README.md index ec508f6..2fe01cc 100644 --- a/README.md +++ b/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)。 diff --git a/docs/opensandbox-runner.md b/docs/opensandbox-runner.md index b95c28d..908fb3d 100644 --- a/docs/opensandbox-runner.md +++ b/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 diff --git a/src/gitea_dynamic_runner/controller.py b/src/gitea_dynamic_runner/controller.py index a867326..044f758 100644 --- a/src/gitea_dynamic_runner/controller.py +++ b/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, ) @@ -200,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, ), ) @@ -221,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 @@ -231,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, @@ -240,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, @@ -257,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() diff --git a/tests/test_controller.py b/tests/test_controller.py index 163de7d..0581b00 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -134,3 +134,8 @@ async def test_completed_lifecycle_cancellation_acks_message(): 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