diff --git a/src/gitea_dynamic_runner/controller.py b/src/gitea_dynamic_runner/controller.py index c5bd022..582c3e5 100644 --- a/src/gitea_dynamic_runner/controller.py +++ b/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"))) diff --git a/src/gitea_dynamic_runner/pod_worker.py b/src/gitea_dynamic_runner/pod_worker.py index f733081..29517a1 100644 --- a/src/gitea_dynamic_runner/pod_worker.py +++ b/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: diff --git a/tests/test_pod_worker.py b/tests/test_pod_worker.py index 00c8dc0..62baf8e 100644 --- a/tests/test_pod_worker.py +++ b/tests/test_pod_worker.py @@ -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")