Compare commits

..
Author SHA1 Message Date
panxiao81 68bb02b20a 修复 Runner 身份路径并补充关联日志
test / python (pull_request) Successful in 8s
test / shell (pull_request) Successful in 15s
2026-09-16 18:00:50 +00:00
4 changed files with 143 additions and 24 deletions
+3 -15
View File
@@ -1,20 +1,8 @@
FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242 AS runner FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242
# The runner daemon image is intentionally minimal and does not contain the
# Node.js runtime required by JavaScript actions such as actions/checkout.
# Run the daemon in Gitea's Ubuntu workflow image so host-mode jobs and their
# actions share a GitHub Actions-compatible userspace.
FROM docker.io/gitea/runner-images:ubuntu-latest@sha256:fd911d7417bfbf0f454530e447da95b58001e1df41bbc5e1a8dd35d432575aae
USER root USER root
COPY --from=runner /usr/local/bin/gitea-runner /usr/local/bin/gitea-runner
COPY --from=runner /usr/local/bin/run.sh /usr/local/bin/run.sh
COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
COPY config/runner.yaml /etc/gitea-runner/config.yaml COPY config/runner.yaml /etc/gitea-runner/config.yaml
COPY --chmod=0755 scripts/gitea-job-started /usr/local/libexec/gitea-job-started COPY scripts/gitea-job-started /usr/local/libexec/gitea-job-started
RUN chmod 0755 /usr/local/libexec/gitea-job-started
VOLUME ["/data"]
WORKDIR /
ENTRYPOINT ["/usr/local/bin/run.sh"]
+48 -2
View File
@@ -4,6 +4,7 @@
import hashlib import hashlib
import hmac import hmac
import json import json
import logging
import os import os
import ssl import ssl
from pathlib import Path from pathlib import Path
@@ -16,6 +17,7 @@ from nats.js.errors import NotFoundError
from .models import IdentityBinding, RunnerRequest from .models import IdentityBinding, RunnerRequest
LOG = logging.getLogger(__name__)
SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner") SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER") STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222") 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) payload = json.loads(body)
except json.JSONDecodeError as error: except json.JSONDecodeError as error:
raise web.HTTPBadRequest(text="invalid JSON\n") from 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) runner_request = RunnerRequest.from_webhook(payload)
if runner_request is not None: if runner_request is not None:
subject = f"{SUBJECT_PREFIX}.{runner_request.backend}"
await request.app["js"].publish( await request.app["js"].publish(
f"{SUBJECT_PREFIX}.{runner_request.backend}", subject,
runner_request.to_json(), runner_request.to_json(),
headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"}, 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") return web.Response(status=202, text="queued\n")
binding = IdentityBinding.from_webhook(payload) binding = IdentityBinding.from_webhook(payload)
if binding is not None: if binding is not None:
subject = f"{SUBJECT_PREFIX}.{binding.backend}.binding"
await request.app["js"].publish( await request.app["js"].publish(
f"{SUBJECT_PREFIX}.{binding.backend}.binding", subject,
binding.to_json(), binding.to_json(),
headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"}, 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") return web.Response(status=202, text="binding queued\n")
LOG.warning("workflow_job webhook ignored %s", context)
return web.Response(status=204) 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: async def health(request: web.Request) -> web.Response:
connected = request.app["nc"].is_connected connected = request.app["nc"].is_connected
return web.Response(text="ok\n" if connected else "disconnected\n", status=200 if connected else 503) 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: 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"))) web.run_app(create_app(), host=os.environ.get("LISTEN", "0.0.0.0"), port=int(os.environ.get("PORT", "8787")))
+76 -5
View File
@@ -4,12 +4,13 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import hashlib
import json import json
import logging import logging
import os import os
from pathlib import Path from pathlib import Path
import re
import ssl import ssl
from urllib.parse import quote
import uuid import uuid
import nats 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") GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.ddupan.top")
CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4")) CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4"))
POD_TIMEOUT = int(os.environ.get("RUNNER_POD_TIMEOUT", str(4 * 60 * 60))) 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: def identity_path(repository: str, job_name: str) -> str:
parts = repository.split("/") parts = repository.split("/")
if len(parts) != 2 or not all(parts): if len(parts) != 2 or not all(parts):
raise ValueError("repository must be owner/name") raise ValueError("repository must be owner/name")
encoded = [quote(part, safe="-._~") for part in (*parts, job_name)] if not job_name.strip():
if not job_name.strip() or any(not part for part in encoded):
raise ValueError("identity components must not be empty") 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]: 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 return
pod_name = f"gitea-pod-{uuid.uuid4().hex[:12]}" 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() stop = asyncio.Event()
pulse = asyncio.create_task(heartbeat(message, stop)) pulse = asyncio.create_task(heartbeat(message, stop))
try: try:
await client.create_pod(pod_manifest(request, pod_name)) 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): if await wait_for_pod(client, pod_name):
await message.ack() await message.ack()
LOG.info(
"runner request acknowledged pod=%s job_id=%d result=succeeded",
pod_name,
request.job_id,
)
else: 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) await message.nak(delay=30)
except Exception: 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) await message.nak(delay=30)
raise raise
finally: finally:
stop.set() stop.set()
await pulse await pulse
LOG.info("deleting runner Pod pod=%s job_id=%d", pod_name, request.job_id)
await client.delete_pod(pod_name) await client.delete_pod(pod_name)
async def bind_request(message: object, client: KubernetesClient) -> None: async def bind_request(message: object, client: KubernetesClient) -> None:
binding: IdentityBinding | None = None
try: try:
document = json.loads(message.data) document = json.loads(message.data)
binding = IdentityBinding(**document) binding = IdentityBinding(**document)
if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"): if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"):
raise ValueError("invalid Pod identity binding") raise ValueError("invalid Pod identity binding")
path = identity_path(binding.repository, binding.job_name) 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) await client.bind_identity(binding.runner_name, path)
except ClientResponseError as error: except ClientResponseError as error:
if error.status == 404: 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) await message.nak(delay=2)
return 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) await message.nak(delay=30)
raise raise
except (json.JSONDecodeError, TypeError, ValueError) as error: except (json.JSONDecodeError, TypeError, ValueError) as error:
@@ -211,6 +276,12 @@ async def bind_request(message: object, client: KubernetesClient) -> None:
await message.ack() await message.ack()
return return
await message.ack() 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: async def consume_requests(subscription: object, client: KubernetesClient) -> None:
+16 -2
View File
@@ -16,8 +16,22 @@ def request() -> RunnerRequest:
def test_identity_path_is_meaningful_and_uri_safe(): def test_identity_path_is_meaningful_and_uri_safe():
assert pod_worker.identity_path("panxiao81/example", "publish image") == ( path = pod_worker.identity_path("panxiao81/example", "publish image")
"panxiao81/example/publish%20image" 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): with pytest.raises(ValueError):
pod_worker.identity_path("invalid", "test") pod_worker.identity_path("invalid", "test")