建立 Pod 与 VM 调度消息契约

src/gitea_microvm_runner/controller.py

@@ -13,8 +13,10 @@ from aiohttp import web
 from nats.js.api import DiscardPolicy, RetentionPolicy, StorageType, StreamConfig
 from nats.js.errors import NotFoundError
 
-LABEL = os.environ.get("RUNNER_LABEL", "kind-microvm")
-SUBJECT = os.environ.get("NATS_SUBJECT", f"ci.runner.{LABEL}")
+from .models import IdentityBinding, RunnerRequest
+
+
+SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
 STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
 NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
 NATS_USER = os.environ.get("NATS_USER", "ci-producer")
@@ -24,15 +26,9 @@ WEBHOOK_SECRET_FILE = Path(os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/g
 
 
 def accepts(payload: object) -> 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:

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())

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()