修复 Runner 身份路径并补充关联日志

README.md

@@ -25,13 +25,19 @@ runs-on: [self-hosted, vm]
 - `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral
   runner，执行一个 job 后关机。
 - `pod-worker`：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用
-  host executor，Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。
+  host executor，Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在
+  支持原生 job hooks 的 3.x 版本，在 workflow 第一步前等待实际任务对应的 SVID。
 - `jwt-broker`：早期共享 Kubernetes runner 的过渡实验；目标架构不部署它，每个
   动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
 
 消息流使用一个 `WorkQueuePolicy` stream。相同 runner label 的所有 worker 共享同一
 durable consumer；扩容只需要增加 worker 或提高单机 capacity。
 
+当前 webhook → NATS 流程是用于尽快验证 Pod/VM 生命周期的 bootstrap 实现，不是
+长期调度接口。长期目标是让 controller 作为兼容 Gitea Runner 协议的调度器直接注册、
+声明 labels、领取 task，并把已领取 task 交给 Pod/VM executor；路线与迁移边界见
+[`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
+
 ## 开发
 
 ```bash

container/controller.Dockerfile

@@ -7,9 +7,11 @@ COPY src ./src
 RUN python -m venv /venv && /venv/bin/pip install --no-cache-dir .
 
 FROM python:3.12.11-alpine3.22
-RUN addgroup -S -g 65532 runner && adduser -S -D -H -u 65532 -G runner runner
+RUN addgroup -S -g 65532 runner \
+    && adduser -S -D -H -u 65532 -G runner runner \
+    && install -d -o 65532 -g 65532 /var/run/secrets/kubernetes.io/serviceaccount
 COPY --from=build /venv /venv
 COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
 USER 65532:65532
 EXPOSE 8787
-ENTRYPOINT ["/venv/bin/gitea-microvm-controller"]
+ENTRYPOINT ["/venv/bin/gitea-dynamic-runner-controller"]

container/runner.Dockerfile

@@ -1,6 +1,6 @@
 FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
 
-FROM docker.io/gitea/runner:2@sha256:66d80966792e621c9761c47919644198d35fd1c297e9a01e69ed3c1ae37db0c7
+FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242
 USER root
 COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
 COPY config/runner.yaml /etc/gitea-runner/config.yaml

docs/runner-protocol-roadmap.md

@@ -0,0 +1,59 @@
+# Gitea Runner 协议调度器路线
+
+## 目标
+
+长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner
+协议客户端注册，并声明 `self-hosted`、`pod` 和 `vm` labels；它只在后端存在可用容量
+时领取 task，然后将该 task 交给一个一次性 Pod 或 microVM 执行。
+
+```text
+Gitea RunnerService
+        │ Register / Declare / FetchTask
+        ▼
+dynamic-runner scheduler
+        │ 已领取的 task + lease
+        ├── Pod executor
+        └── microVM executor
+                │ logs / state / result
+                └──────────────────────► Gitea
+```
+
+这与“收到 webhook 后临时注册另一个 act_runner”不同。`FetchTask` 已经完成任务分配，
+不能再期待 Gitea 把同一个 task 分配给随后启动的 runner。协议调度器必须让 executor
+执行已经领取的 task，并继续完成日志、状态、心跳、取消和最终结果上报。
+
+## 设计约束
+
+- 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。
+- scheduler 在没有对应 backend 容量时不领取 task，避免本地形成不可控积压。
+- 每个 executor 只执行一个 task，完成后销毁。
+- SPIFFE 身份从实际领取的 task 的 repository 和 job name 派生，不需要 queued 与
+  in-progress webhook 的二阶段关联。
+- scheduler 的 runner registration credential 不进入 executor；executor 只得到执行
+  当前 task 所需的短期 lease/capability。
+- task ACK、心跳和结果必须能够跨 scheduler 重启恢复；NATS 可以继续作为内部 handoff，
+  但不是 Gitea 任务事实来源。
+- Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。
+
+## 实现顺序
+
+1. 固定当前 Gitea 版本所使用的 RunnerService protobuf 与 act_runner 版本，记录兼容
+   范围并建立协议契约测试。
+2. 实现只注册、Declare labels 和容量感知 FetchTask 的 scheduler spike，暂不执行
+   task。
+3. 从 act_runner 提取或复用 task 执行与日志上报能力，定义 scheduler 到 executor 的
+   单任务协议。
+4. 首先接入 Pod executor，验证成功、失败、取消、超时和 scheduler 重启。
+5. 接入 microVM executor，并复用同一 task 协议和身份派生逻辑。
+6. 双轨运行并验证后，移除 webhook receiver、临时 runner 注册和 identity binding
+   subject。
+
+## Bootstrap 实现的退出条件
+
+只有同时满足以下条件才能删除 webhook 路径：
+
+- scheduler 能通过 RunnerService 稳定领取并执行 Pod/VM task；
+- Gitea UI 中的实时日志、取消、超时和结论与官方 runner 行为一致；
+- scheduler 重启不会丢失已领取 task，也不会重复执行；
+- SPIFFE 身份只来自实际领取 task；
+- 同一套 workflow 无需修改 `runs-on` 即可从 bootstrap 迁移。

pyproject.toml

@@ -3,9 +3,9 @@ requires = ["setuptools>=75"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "gitea-microvm-runner"
-version = "0.2.0"
-description = "On-demand Cloud Hypervisor runners for Gitea Actions"
+name = "gitea-dynamic-runner"
+version = "0.3.0"
+description = "On-demand Pod and microVM execution environments for Gitea Actions"
 requires-python = ">=3.11"
 dependencies = ["aiohttp==3.12.15", "nats-py==2.11.0"]
 
@@ -13,10 +13,10 @@ dependencies = ["aiohttp==3.12.15", "nats-py==2.11.0"]
 test = ["pytest==8.4.2", "pytest-asyncio==1.2.0"]
 
 [project.scripts]
-gitea-microvm-controller = "gitea_microvm_runner.controller:main"
-gitea-microvm-worker = "gitea_microvm_runner.worker:cli"
-gitea-pod-worker = "gitea_microvm_runner.pod_worker:cli"
-gitea-spire-jwt-broker = "gitea_microvm_runner.jwt_broker:main"
+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-spire-jwt-broker = "gitea_dynamic_runner.jwt_broker:main"
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"

src/gitea_dynamic_runner/__init__.py

@@ -0,0 +1 @@
+"""Dynamic Pod and microVM execution environments for Gitea Actions."""

src/gitea_dynamic_runner/controller.py

@@ -4,6 +4,7 @@
 import hashlib
 import hmac
 import json
+import logging
 import os
 import ssl
 from pathlib import Path
@@ -16,6 +17,7 @@ from nats.js.errors import NotFoundError
 from .models import IdentityBinding, RunnerRequest
 
 
+LOG = logging.getLogger(__name__)
 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")
@@ -64,27 +66,70 @@ 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
+    context = _webhook_context(payload)
+    LOG.info("workflow_job webhook received %s", context)
+
     runner_request = RunnerRequest.from_webhook(payload)
     if runner_request is not None:
+        subject = f"{SUBJECT_PREFIX}.{runner_request.backend}"
         await request.app["js"].publish(
-            f"{SUBJECT_PREFIX}.{runner_request.backend}",
+            subject,
             runner_request.to_json(),
             headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
         )
+        LOG.info(
+            "runner request published subject=%s job_id=%d run_id=%d backend=%s "
+            "repository=%s job_name=%r labels=%s",
+            subject,
+            runner_request.job_id,
+            runner_request.run_id,
+            runner_request.backend,
+            runner_request.repository,
+            runner_request.job_name,
+            runner_request.labels,
+        )
         return web.Response(status=202, text="queued\n")
 
     binding = IdentityBinding.from_webhook(payload)
     if binding is not None:
+        subject = f"{SUBJECT_PREFIX}.{binding.backend}.binding"
         await request.app["js"].publish(
-            f"{SUBJECT_PREFIX}.{binding.backend}.binding",
+            subject,
             binding.to_json(),
             headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"},
         )
+        LOG.info(
+            "identity binding published subject=%s job_id=%d run_id=%d backend=%s "
+            "runner_name=%s repository=%s job_name=%r",
+            subject,
+            binding.job_id,
+            binding.run_id,
+            binding.backend,
+            binding.runner_name,
+            binding.repository,
+            binding.job_name,
+        )
         return web.Response(status=202, text="binding queued\n")
 
+    LOG.warning("workflow_job webhook ignored %s", context)
     return web.Response(status=204)
 
 
+def _webhook_context(payload: object) -> str:
+    if not isinstance(payload, dict):
+        return f"payload_type={type(payload).__name__}"
+    job = payload.get("workflow_job")
+    repository = payload.get("repository")
+    job = job if isinstance(job, dict) else {}
+    repository = repository if isinstance(repository, dict) else {}
+    return (
+        f"action={payload.get('action')!r} job_id={job.get('id')!r} "
+        f"run_id={job.get('run_id')!r} runner_name={job.get('runner_name')!r} "
+        f"repository={repository.get('full_name')!r} job_name={job.get('name')!r} "
+        f"labels={job.get('labels')!r}"
+    )
+
+
 async def health(request: web.Request) -> web.Response:
     connected = request.app["nc"].is_connected
     return web.Response(text="ok\n" if connected else "disconnected\n", status=200 if connected else 503)
@@ -115,6 +160,7 @@ def create_app() -> web.Application:
 
 
 def main() -> None:
+    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
     web.run_app(create_app(), host=os.environ.get("LISTEN", "0.0.0.0"), port=int(os.environ.get("PORT", "8787")))
 
 

src/gitea_dynamic_runner/pod_worker.py

@@ -4,12 +4,13 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import json
 import logging
 import os
 from pathlib import Path
+import re
 import ssl
-from urllib.parse import quote
 import uuid
 
 import nats
@@ -49,16 +50,26 @@ RUNNER_TOKEN_SECRET = os.environ.get("RUNNER_TOKEN_SECRET", "gitea-dynamic-runne
 GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.ddupan.top")
 CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4"))
 POD_TIMEOUT = int(os.environ.get("RUNNER_POD_TIMEOUT", str(4 * 60 * 60)))
+SPIFFE_PATH_SEGMENT = re.compile(r"^[A-Za-z0-9._-]+$")
 
 
 def identity_path(repository: str, job_name: str) -> str:
     parts = repository.split("/")
     if len(parts) != 2 or not all(parts):
         raise ValueError("repository must be owner/name")
-    encoded = [quote(part, safe="-._~") for part in (*parts, job_name)]
-    if not job_name.strip() or any(not part for part in encoded):
+    if not job_name.strip():
         raise ValueError("identity components must not be empty")
-    return "/".join(encoded)
+    return "/".join(safe_identity_segment(part) for part in (*parts, job_name))
+
+
+def safe_identity_segment(value: str) -> str:
+    """Map arbitrary Gitea names to stable SPIFFE Operator path segments."""
+    if SPIFFE_PATH_SEGMENT.fullmatch(value):
+        return value
+    slug = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-._")
+    slug = slug[:48].rstrip("-._") or "segment"
+    digest = hashlib.sha256(value.encode()).hexdigest()[:12]
+    return f"{slug}-{digest}"
 
 
 def pod_manifest(request: RunnerRequest, pod_name: str) -> dict[str, object]:
@@ -174,36 +185,90 @@ async def run_request(message: object, client: KubernetesClient) -> None:
         return
 
     pod_name = f"gitea-pod-{uuid.uuid4().hex[:12]}"
+    LOG.info(
+        "runner request received job_id=%d run_id=%d repository=%s job_name=%r "
+        "backend=%s pod=%s",
+        request.job_id,
+        request.run_id,
+        request.repository,
+        request.job_name,
+        request.backend,
+        pod_name,
+    )
     stop = asyncio.Event()
     pulse = asyncio.create_task(heartbeat(message, stop))
     try:
         await client.create_pod(pod_manifest(request, pod_name))
+        LOG.info(
+            "runner Pod created pod=%s job_id=%d run_id=%d image=%s",
+            pod_name,
+            request.job_id,
+            request.run_id,
+            RUNNER_IMAGE,
+        )
         if await wait_for_pod(client, pod_name):
             await message.ack()
+            LOG.info(
+                "runner request acknowledged pod=%s job_id=%d result=succeeded",
+                pod_name,
+                request.job_id,
+            )
         else:
-            LOG.error("runner Pod %s failed", pod_name)
+            LOG.error(
+                "runner Pod failed pod=%s job_id=%d; request will be retried",
+                pod_name,
+                request.job_id,
+            )
             await message.nak(delay=30)
     except Exception:
+        LOG.exception(
+            "runner request failed pod=%s job_id=%d; request will be retried",
+            pod_name,
+            request.job_id,
+        )
         await message.nak(delay=30)
         raise
     finally:
         stop.set()
         await pulse
+        LOG.info("deleting runner Pod pod=%s job_id=%d", pod_name, request.job_id)
         await client.delete_pod(pod_name)
 
 
 async def bind_request(message: object, client: KubernetesClient) -> None:
+    binding: IdentityBinding | None = None
     try:
         document = json.loads(message.data)
         binding = IdentityBinding(**document)
         if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"):
             raise ValueError("invalid Pod identity binding")
         path = identity_path(binding.repository, binding.job_name)
+        LOG.info(
+            "identity binding received job_id=%d run_id=%d runner_name=%s "
+            "repository=%s job_name=%r identity_path=%s",
+            binding.job_id,
+            binding.run_id,
+            binding.runner_name,
+            binding.repository,
+            binding.job_name,
+            path,
+        )
         await client.bind_identity(binding.runner_name, path)
     except ClientResponseError as error:
         if error.status == 404:
+            LOG.warning(
+                "identity binding Pod not found runner_name=%s job_id=%s; binding will be retried",
+                binding.runner_name if binding else None,
+                binding.job_id if binding else None,
+            )
             await message.nak(delay=2)
             return
+        LOG.exception(
+            "identity binding Kubernetes request failed runner_name=%s job_id=%s status=%d",
+            binding.runner_name if binding else None,
+            binding.job_id if binding else None,
+            error.status,
+        )
         await message.nak(delay=30)
         raise
     except (json.JSONDecodeError, TypeError, ValueError) as error:
@@ -211,6 +276,12 @@ async def bind_request(message: object, client: KubernetesClient) -> None:
         await message.ack()
         return
     await message.ack()
+    LOG.info(
+        "identity binding applied and acknowledged runner_name=%s job_id=%d identity_path=%s",
+        binding.runner_name,
+        binding.job_id,
+        path,
+    )
 
 
 async def consume_requests(subscription: object, client: KubernetesClient) -> None:

src/gitea_microvm_runner/__init__.py

@@ -1 +0,0 @@
-"""Gitea microVM runner controller and worker."""

systemd/gitea-microvm-worker.service

@@ -7,7 +7,7 @@ Requires=microvm-runner-network.service
 [Service]
 Type=simple
 EnvironmentFile=-/etc/microvm-runner/worker.env
-ExecStart=/opt/gitea-microvm-runner/venv/bin/gitea-microvm-worker
+ExecStart=/opt/gitea-dynamic-runner/venv/bin/gitea-dynamic-runner-vm-worker
 Restart=on-failure
 RestartSec=5s
 SupplementaryGroups=kvm

tests/test_controller.py

@@ -4,8 +4,8 @@ import json
 
 import pytest
 
-from gitea_microvm_runner import controller
-from gitea_microvm_runner.models import IdentityBinding, RunnerRequest
+from gitea_dynamic_runner import controller
+from gitea_dynamic_runner.models import IdentityBinding, RunnerRequest
 
 
 def queued_job(**overrides):

tests/test_jwt_broker.py

@@ -1,6 +1,6 @@
 import pytest
 
-from gitea_microvm_runner.jwt_broker import extract_svid
+from gitea_dynamic_runner.jwt_broker import extract_svid
 
 
 def test_extract_svid_from_spire_json():

tests/test_pod_worker.py

@@ -1,7 +1,7 @@
 import pytest
 
-from gitea_microvm_runner.models import RunnerRequest
-from gitea_microvm_runner import pod_worker
+from gitea_dynamic_runner.models import RunnerRequest
+from gitea_dynamic_runner import pod_worker
 
 
 def request() -> RunnerRequest:
@@ -16,8 +16,22 @@ def request() -> RunnerRequest:
 
 
 def test_identity_path_is_meaningful_and_uri_safe():
-    assert pod_worker.identity_path("panxiao81/example", "publish image") == (
-        "panxiao81/example/publish%20image"
+    path = pod_worker.identity_path("panxiao81/example", "publish image")
+    assert path.startswith("panxiao81/example/publish-image-")
+    assert "%" not in path
+    assert all(pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part) for part in path.split("/"))
+    assert pod_worker.identity_path("panxiao81/example", "lint") == "panxiao81/example/lint"
+    assert path == pod_worker.identity_path("panxiao81/example", "publish image")
+    assert path != pod_worker.identity_path("panxiao81/example", "publish-image")
+    real_path = pod_worker.identity_path(
+        "panxiao81/postgresql-tenant-operator", "Run on Ubuntu"
+    )
+    assert real_path.startswith(
+        "panxiao81/postgresql-tenant-operator/Run-on-Ubuntu-"
+    )
+    assert all(
+        pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part)
+        for part in real_path.split("/")
     )
     with pytest.raises(ValueError):
         pod_worker.identity_path("invalid", "test")