在 GCP 上动态创建 Forgejo Runner

背景

此前一直把部分信息敏感内容放在 Github 的 Private Repository 上，最终还是想自己运行一个实例托管。最后选用了 Forgejo，但有些 Repo 需要使用到 Actions 功能，于是我把目光看向了 Google Cloud。我将从头开始，一步一步实现按需创建一次性 GCP 实例，执行构建任务的目标。

目前你看到的网站，就是由 GCP 上的 Forgejo Runner 构建。

工作方式

整个方案由三部分组成：

Forgejo 中的 Job 使用 runs-on: gcp-linux 进入队列；
一台自己的 VPS 常驻运行 scaler，轮询 Forgejo 队列并创建 GCP VM；
GCP VM 注册为全局一次性 Runner，只执行一个 Job，完成后关机并由 scaler 删除。

Scaler 持有 Forgejo 管理 Token 和 GCP Service Account JSON，但这些凭据不会进入仓库、Actions Secret 或临时 VM。

全局 Runner 可以执行站点内任意仓库的匹配任务。任何有权运行 Actions 的用户，也就有能力消耗 GCP 额度。因此不建议在允许陌生用户自由注册的 Forgejo 实例中直接使用本方案。

创建 Forgejo 管理 Token

建议先创建一个专用用户，例如 forgejo-scaler，然后在：

Site Administration
→ Identity & access
→ User accounts

将它设为 Site Administrator。

再使用该账号进入：

User Settings
→ Applications
→ Generate New Token

Token 配置如下：

Token name：forgejo-gcp-sitewide-scaler
Repository and organization access：All (public, private, and limited)
Scope：write:admin

保存生成的 Token，稍后放到 scaler VPS 上。

如果 Forgejo 位于 Cloudflare 后面，需要确保 scaler 的请求不会被 Bot Fight、WAF 或 Access 拦截。最好只对 /api/v1/ 创建精确的放行规则，而不是关闭整个站点的保护。

配置 GCP

下面的命令均在 Google Cloud Shell 中执行。先替换项目 ID，并设置本文需要的变量：

export PROJECT_ID="YOUR_GCP_PROJECT_ID"
export REGION="us-south1"
export ZONE="us-south1-a"

export SCALER_SA_NAME="forgejo-scaler"
export SCALER_SA_EMAIL="${SCALER_SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"

export NETWORK_NAME="forgejo-runner-net"
export SUBNET_NAME="forgejo-runner-subnet"

export BUILDER_NAME="forgejo-image-builder-v1"
export IMAGE_NAME="forgejo-runner-debian12-v20260719-1"
export IMAGE_FAMILY="forgejo-runner-debian12"
export TEMPLATE_NAME="forgejo-runner-v1"

gcloud config set project "$PROJECT_ID"
gcloud config set compute/region "$REGION"
gcloud config set compute/zone "$ZONE"

Cloud Shell 重新打开后，export 的变量不会保留，需要重新执行。

启用 Compute Engine API

gcloud services enable compute.googleapis.com \
  --project="$PROJECT_ID"

创建 scaler Service Account

gcloud iam service-accounts create "$SCALER_SA_NAME" \
  --display-name="Forgejo site-wide ephemeral runner scaler"

gcloud projects add-iam-policy-binding "$PROJECT_ID" \
  --member="serviceAccount:${SCALER_SA_EMAIL}" \
  --role="roles/compute.instanceAdmin.v1"

临时 VM 不附加 Service Account，因此 scaler 不需要 iam.serviceAccounts.actAs。最好使用一个只存放 Runner 资源的独立 GCP Project，把 compute.instanceAdmin.v1 的影响限制在该项目中。

创建 Service Account JSON

gcloud iam service-accounts keys create \
  "$HOME/forgejo-scaler.json" \
  --iam-account="$SCALER_SA_EMAIL"

chmod 0600 "$HOME/forgejo-scaler.json"

如果出现：

Key creation is not allowed on this service account
constraints/iam.disableServiceAccountKeyCreation

说明组织策略禁止创建长期密钥。可以在这个专用 Project 中暂时关闭该策略：

gcloud resource-manager org-policies disable-enforce \
  iam.disableServiceAccountKeyCreation \
  --project="$PROJECT_ID"

创建并下载密钥后，重新启用：

gcloud resource-manager org-policies enable-enforce \
  iam.disableServiceAccountKeyCreation \
  --project="$PROJECT_ID"

如果当前账号没有修改组织策略的权限，就需要让组织管理员操作。密钥创建失败后还可能留下一个 0 字节文件，重新执行前先检查并删除它。

将 forgejo-scaler.json 下载并传到 scaler VPS 后，删除 Cloud Shell 中的副本：

shred -u "$HOME/forgejo-scaler.json"

不要把 JSON 转成 Base64，也不要放进 Forgejo Actions Secret。

创建专用网络

gcloud compute networks create "$NETWORK_NAME" \
  --subnet-mode=custom

gcloud compute networks subnets create "$SUBNET_NAME" \
  --network="$NETWORK_NAME" \
  --region="$REGION" \
  --range="10.50.0.0/24"

检查子网区域：

gcloud compute networks subnets list \
  --filter="name=$SUBNET_NAME" \
  --format="table(name,region,network,ipCidrRange)"

这里必须是 us-south1。不需要创建 SSH、RDP 或其他公网入站规则，Runner 只需要主动访问 Forgejo、Docker Hub 和依赖源。

烘焙 Runner 镜像

为了避免每台临时 VM 启动后都重新安装 Docker、Git 和 Runner，先制作一个自定义镜像。不要从执行过仓库 Job 的 Runner VM 制作镜像，否则可能把工作目录、构建缓存甚至 Actions Secret 一起写入镜像。

创建 bake-runner-image.sh：

cat > bake-runner-image.sh <<'BAKE'
#!/usr/bin/env bash
set -Eeuo pipefail

exec > >(
  tee -a /var/log/forgejo-image-build.log |
  logger -t forgejo-image-build -s 2>/dev/console
) 2>&1

export DEBIAN_FRONTEND=noninteractive

apt-get update
apt-get install -y --no-install-recommends \
  ca-certificates \
  curl \
  docker.io \
  git \
  iptables

systemctl enable --now docker

if ! id forgejo-runner >/dev/null 2>&1; then
  useradd \
    --create-home \
    --shell /bin/bash \
    forgejo-runner
fi

usermod -aG docker forgejo-runner

ARCH="$(
  uname -m |
  sed \
    -e 's/x86_64/amd64/' \
    -e 's/aarch64/arm64/'
)"

RUNNER_VERSION="12.13.0"
RUNNER_URL="https://code.forgejo.org/forgejo/runner/releases/download/v${RUNNER_VERSION}/forgejo-runner-${RUNNER_VERSION}-linux-${ARCH}"

curl \
  --fail \
  --location \
  --retry 5 \
  --retry-all-errors \
  --output /usr/local/bin/forgejo-runner \
  "$RUNNER_URL"

chmod 0755 /usr/local/bin/forgejo-runner

docker pull docker.io/library/node:22-bookworm

apt-get clean
rm -rf /var/lib/apt/lists/*
sync

echo "IMAGE_BUILD_READY"
shutdown -h now
BAKE

chmod +x bake-runner-image.sh

创建一台普通按需 Builder VM，避免烘焙过程中被 Spot 抢占：

gcloud compute instances create "$BUILDER_NAME" \
  --project="$PROJECT_ID" \
  --zone="$ZONE" \
  --machine-type="e2-medium" \
  --image-family="debian-12" \
  --image-project="debian-cloud" \
  --boot-disk-size="30GB" \
  --boot-disk-type="pd-balanced" \
  --subnet="$SUBNET_NAME" \
  --no-service-account \
  --no-scopes \
  --no-restart-on-failure \
  --metadata="block-project-ssh-keys=true,enable-oslogin=false,serial-port-enable=true" \
  --metadata-from-file="startup-script=bake-runner-image.sh"

查看日志：

gcloud compute instances get-serial-port-output "$BUILDER_NAME" \
  --project="$PROJECT_ID" \
  --zone="$ZONE" \
  --port=1 | tail -n 100

看到 IMAGE_BUILD_READY，并确认 VM 状态为 TERMINATED 后，创建镜像：

gcloud compute images create "$IMAGE_NAME" \
  --project="$PROJECT_ID" \
  --source-disk="$BUILDER_NAME" \
  --source-disk-zone="$ZONE" \
  --family="$IMAGE_FAMILY"

gcloud compute instances delete "$BUILDER_NAME" \
  --project="$PROJECT_ID" \
  --zone="$ZONE" \
  --quiet

创建 V1 Instance Template

临时 VM 启动时只需要读取 scaler 写入的动态 metadata，启动一次性 Runner，任务完成后关机。

创建 startup-runner-slim-v1.sh：

cat > startup-runner-slim-v1.sh <<'STARTUP'
#!/usr/bin/env bash
set -Eeuo pipefail

exec > >(
  tee -a /var/log/forgejo-ephemeral-runner.log |
  logger -t forgejo-ephemeral-runner -s 2>/dev/console
) 2>&1

METADATA_BASE="http://metadata.google.internal/computeMetadata/v1"
METADATA_HEADER="Metadata-Flavor: Google"

metadata_get() {
  curl \
    --fail \
    --silent \
    --show-error \
    --connect-timeout 2 \
    --max-time 5 \
    -H "$METADATA_HEADER" \
    "${METADATA_BASE}/instance/attributes/$1"
}

wait_for_metadata() {
  local key="$1"
  local value=""

  for _ in $(seq 1 150); do
    value="$(metadata_get "$key" 2>/dev/null || true)"

    if [ -n "$value" ]; then
      printf '%s' "$value"
      return 0
    fi

    sleep 2
  done

  echo "Timed out waiting for metadata key: $key" >&2
  return 1
}

test -x /usr/local/bin/forgejo-runner
id forgejo-runner >/dev/null

systemctl start docker

FORGEJO_URL="$(wait_for_metadata forgejo-url)"
RUNNER_UUID="$(wait_for_metadata runner-uuid)"
RUNNER_TOKEN="$(wait_for_metadata runner-token)"
RUNNER_LABEL="$(wait_for_metadata runner-label)"

install \
  -d \
  -o forgejo-runner \
  -g forgejo-runner \
  -m 0700 \
  /run/forgejo-runner

printf '%s' "$RUNNER_TOKEN" > /run/forgejo-runner/token
chown forgejo-runner:forgejo-runner /run/forgejo-runner/token
chmod 0600 /run/forgejo-runner/token

for _ in $(seq 1 30); do
  if iptables -nL DOCKER-USER >/dev/null 2>&1; then
    break
  fi

  sleep 1
done

# 169.254.169.254 同时承载 GCE Metadata HTTP 和默认 DNS。
# 只阻止 Job 容器访问 TCP 80/443，保留 UDP/TCP 53。
iptables -C DOCKER-USER \
  -d 169.254.169.254/32 \
  -p tcp \
  -m multiport \
  --dports 80,443 \
  -j REJECT 2>/dev/null ||
iptables -I DOCKER-USER \
  -d 169.254.169.254/32 \
  -p tcp \
  -m multiport \
  --dports 80,443 \
  -j REJECT

unset RUNNER_TOKEN

echo "FORGEJO_RUNNER_READY"

set +e

timeout \
  --signal=TERM \
  --kill-after=30s \
  20m \
  runuser \
    -u forgejo-runner \
    -- \
    /usr/local/bin/forgejo-runner one-job \
      --url "$FORGEJO_URL" \
      --uuid "$RUNNER_UUID" \
      --token-url "file:///run/forgejo-runner/token" \
      --label "${RUNNER_LABEL}:docker://docker.io/library/node:22-bookworm" \
      --wait

RUNNER_EXIT_CODE=$?

set -e

rm -f /run/forgejo-runner/token

echo "FORGEJO_RUNNER_EXIT_CODE=${RUNNER_EXIT_CODE}"

shutdown -h now || true
exit "$RUNNER_EXIT_CODE"
STARTUP

chmod +x startup-runner-slim-v1.sh

这里不能直接封锁整个 169.254.169.254，因为 GCE 的默认 DNS 也使用这个地址。只封锁容器到 TCP 80/443 的访问，既能阻止 Job 读取 metadata，又不会导致 actions/checkout 出现 Could not resolve host。

创建不可变的 V1 模板：

gcloud compute instance-templates create "$TEMPLATE_NAME" \
  --project="$PROJECT_ID" \
  --machine-type="e2-standard-2" \
  --image="$IMAGE_NAME" \
  --image-project="$PROJECT_ID" \
  --boot-disk-size="30GB" \
  --boot-disk-type="pd-balanced" \
  --region="$REGION" \
  --subnet="$SUBNET_NAME" \
  --provisioning-model="SPOT" \
  --maintenance-policy="TERMINATE" \
  --instance-termination-action="DELETE" \
  --max-run-duration="2h" \
  --no-restart-on-failure \
  --no-service-account \
  --no-scopes \
  --shielded-secure-boot \
  --shielded-vtpm \
  --shielded-integrity-monitoring \
  --metadata="block-project-ssh-keys=true,enable-oslogin=false,serial-port-enable=true" \
  --metadata-from-file="startup-script=startup-runner-slim-v1.sh"

Instance Template 不能原地修改。以后需要更新脚本或镜像时，应创建 forgejo-runner-v4、v5 等新模板。

检查模板：

gcloud compute instance-templates describe "$TEMPLATE_NAME" \
  --project="$PROJECT_ID" \
  --format="yaml(
    name,
    properties.disks[0].initializeParams.sourceImage,
    properties.networkInterfaces,
    properties.metadata,
    properties.scheduling,
    properties.serviceAccounts
  )"

需要确认：

引用了刚刚创建的 Custom Image；
子网是 us-south1/forgejo-runner-subnet；
metadata 中存在 startup-script；
没有 serviceAccounts；
provisioningModel 为 SPOT。

部署 scaler

Scaler 运行在自己的一台常驻 VPS 上，需要提前安装 Docker 和 Docker Compose。运行时直接调用 Compute Engine REST API，并使用 google-auth 复用 OAuth Token 和 HTTP 连接，不再安装或启动 gcloud。

准备目录和密钥

sudo install -d \
  -o root \
  -g root \
  -m 0755 \
  /usr/local/forgejo-gcp-scaler

sudo install -d \
  -o root \
  -g root \
  -m 0700 \
  /usr/local/forgejo-gcp-scaler/config \
  /usr/local/forgejo-gcp-scaler/secrets

sudo install \
  -o root \
  -g root \
  -m 0600 \
  forgejo-scaler.json \
  /usr/local/forgejo-gcp-scaler/secrets/gcp.json

写入 Forgejo Token，粘贴后按 Ctrl+D：

sudo sh -c 'umask 077; cat > /usr/local/forgejo-gcp-scaler/secrets/forgejo.token'

创建配置

替换 YOUR_GCP_PROJECT_ID 和 Forgejo 地址：

sudo tee /usr/local/forgejo-gcp-scaler/config/scaler.env >/dev/null <<'EOF'
FORGEJO_URL=https://forgejo.example.com
FORGEJO_TOKEN_FILE=/run/secrets/forgejo.token

GCP_PROJECT_ID=YOUR_GCP_PROJECT_ID
GCP_ZONE=us-south1-a
GCP_INSTANCE_TEMPLATE=forgejo-runner-v1
GCP_KEY_FILE=/run/secrets/gcp.json

RUNNER_LABEL=gcp-linux
MAX_INSTANCES=2
POLL_INTERVAL_SECONDS=15
ORPHAN_GRACE_SECONDS=600
EOF

sudo chown root:root /usr/local/forgejo-gcp-scaler/config/scaler.env
sudo chmod 0600 /usr/local/forgejo-gcp-scaler/config/scaler.env

MAX_INSTANCES 是成本控制中最重要的开关。个人实例建议先设为 1，确认稳定后再提高。

创建 scaler.py

sudo tee /usr/local/forgejo-gcp-scaler/scaler.py >/dev/null <<'PYTHON'
#!/usr/bin/env python3

from __future__ import annotations

import datetime as dt
import json
import os
import secrets
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any

try:
    import requests
    from google.auth.transport.requests import AuthorizedSession
    from google.oauth2 import service_account
except ImportError as exc:
    raise RuntimeError(
        "Missing Python dependencies. Install them with: "
        "pip install google-auth requests"
    ) from exc


def required_env(name: str) -> str:
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value


FORGEJO_URL = required_env("FORGEJO_URL").rstrip("/")
FORGEJO_TOKEN_FILE = Path(required_env("FORGEJO_TOKEN_FILE"))

GCP_PROJECT_ID = required_env("GCP_PROJECT_ID")
GCP_ZONE = required_env("GCP_ZONE")
GCP_INSTANCE_TEMPLATE = required_env("GCP_INSTANCE_TEMPLATE")
GCP_KEY_FILE = Path(required_env("GCP_KEY_FILE"))

RUNNER_LABEL = os.environ.get("RUNNER_LABEL", "gcp-linux").strip()
MAX_INSTANCES = int(os.environ.get("MAX_INSTANCES", "2"))
POLL_INTERVAL_SECONDS = int(os.environ.get("POLL_INTERVAL_SECONDS", "15"))
ORPHAN_GRACE_SECONDS = int(os.environ.get("ORPHAN_GRACE_SECONDS", "600"))

RUNNER_NAME_PREFIX = "gcp-ephemeral-"
MANAGED_BY_LABEL = "forgejo-gcp-scaler"

COMPUTE_API_BASE = "https://compute.googleapis.com/compute/v1"
GCP_SCOPES = ("https://www.googleapis.com/auth/cloud-platform",)

ACTIVE_VM_STATES = {
    "PROVISIONING",
    "STAGING",
    "RUNNING",
    "STOPPING",
    "SUSPENDING",
    "SUSPENDED",
    "REPAIRING",
}

GCP_SESSION: AuthorizedSession | None = None


def log(message: str) -> None:
    timestamp = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
    print(f"{timestamp} {message}", flush=True)


def read_secret(path: Path) -> str:
    value = path.read_text(encoding="utf-8").strip()
    if not value:
        raise RuntimeError(f"Secret file is empty: {path}")
    return value


FORGEJO_TOKEN = read_secret(FORGEJO_TOKEN_FILE)


def forgejo_api(
    method: str,
    path: str,
    payload: dict[str, Any] | None = None,
    *,
    ignore_not_found: bool = False,
) -> Any:
    url = f"{FORGEJO_URL}{path}"

    body = None
    headers = {
        "Accept": "application/json",
        "Authorization": f"token {FORGEJO_TOKEN}",
        "User-Agent": "forgejo-gcp-scaler/1.0",
    }

    if payload is not None:
        body = json.dumps(payload).encode("utf-8")
        headers["Content-Type"] = "application/json"

    request = urllib.request.Request(
        url=url,
        data=body,
        headers=headers,
        method=method,
    )

    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            raw = response.read()

            if not raw:
                return None

            return json.loads(raw.decode("utf-8"))

    except urllib.error.HTTPError as exc:
        error_body = exc.read().decode("utf-8", errors="replace")

        if ignore_not_found and exc.code == 404:
            return None

        raise RuntimeError(
            f"Forgejo API {method} {path} failed: "
            f"HTTP {exc.code}: {error_body[:500]}"
        ) from exc


def authenticate_gcp() -> None:
    global GCP_SESSION

    log("Loading GCP service-account credentials")

    credentials = service_account.Credentials.from_service_account_file(
        str(GCP_KEY_FILE),
        scopes=GCP_SCOPES,
    )

    # AuthorizedSession 会缓存并自动刷新 OAuth access token，
    # 所有 Compute Engine 请求复用同一个 HTTP Session。
    GCP_SESSION = AuthorizedSession(credentials)


def compute_api(
    method: str,
    path: str,
    *,
    params: dict[str, str | int] | None = None,
    payload: dict[str, Any] | None = None,
    ignore_not_found: bool = False,
) -> dict[str, Any] | None:
    if GCP_SESSION is None:
        raise RuntimeError("GCP session has not been initialized")

    url = f"{COMPUTE_API_BASE}{path}"

    try:
        response = GCP_SESSION.request(
            method=method,
            url=url,
            params=params,
            json=payload,
            timeout=30,
        )
    except requests.RequestException as exc:
        raise RuntimeError(
            f"Compute API {method} {path} request failed: {exc}"
        ) from exc

    if ignore_not_found and response.status_code == 404:
        return None

    if not response.ok:
        body = response.text[:1000]
        raise RuntimeError(
            f"Compute API {method} {path} failed: "
            f"HTTP {response.status_code}: {body}"
        )

    if not response.content:
        return None

    result = response.json()

    if not isinstance(result, dict):
        raise RuntimeError(
            f"Compute API {method} {path} returned invalid JSON"
        )

    return result


def operation_error_message(operation: dict[str, Any]) -> str:
    errors = (operation.get("error") or {}).get("errors") or []

    messages: list[str] = []

    for error in errors:
        code = str(error.get("code", "")).strip()
        message = str(error.get("message", "")).strip()

        if code and message:
            messages.append(f"{code}: {message}")
        elif code or message:
            messages.append(code or message)

    return "; ".join(messages) or "unknown Compute Engine operation error"


def wait_for_zone_operation(
    operation: dict[str, Any],
    *,
    timeout_seconds: int = 600,
) -> dict[str, Any]:
    operation_name = str(operation.get("name", "")).strip()

    if not operation_name:
        raise RuntimeError("Compute Engine operation response lacks name")

    deadline = time.monotonic() + timeout_seconds
    path = (
        f"/projects/{GCP_PROJECT_ID}/zones/{GCP_ZONE}"
        f"/operations/{operation_name}"
    )

    while True:
        current = compute_api("GET", path)

        if current is None:
            raise RuntimeError(
                f"Compute Engine operation {operation_name} disappeared"
            )

        if current.get("status") == "DONE":
            if current.get("error"):
                raise RuntimeError(
                    f"Compute Engine operation {operation_name} failed: "
                    f"{operation_error_message(current)}"
                )

            return current

        if time.monotonic() >= deadline:
            raise TimeoutError(
                f"Timed out waiting for Compute Engine operation "
                f"{operation_name}"
            )

        time.sleep(1)


def instance_template_resource() -> str:
    value = GCP_INSTANCE_TEMPLATE.strip()

    if value.startswith(("https://", "http://")):
        return value

    if value.startswith(("projects/", "global/")):
        return value

    return (
        f"projects/{GCP_PROJECT_ID}/global/instanceTemplates/{value}"
    )


def list_jobs() -> list[dict[str, Any]]:
    encoded_label = urllib.parse.quote(RUNNER_LABEL, safe="")

    jobs = forgejo_api(
        "GET",
        f"/api/v1/admin/actions/runners/jobs?labels={encoded_label}",
    )

    # Forgejo 在没有匹配 Job 时可能返回 JSON null，而不是 []。
    if jobs is None:
        return []

    if not isinstance(jobs, list):
        preview = json.dumps(
            jobs,
            ensure_ascii=False,
            separators=(",", ":"),
        )[:1000]

        raise RuntimeError(
            "Forgejo jobs API returned "
            f"{type(jobs).__name__}, expected list: {preview}"
        )

    return [
        job
        for job in jobs
        if job.get("status") in {"waiting", "running"}
    ]


def list_managed_instances() -> list[dict[str, Any]]:
    path = (
        f"/projects/{GCP_PROJECT_ID}/zones/{GCP_ZONE}/instances"
    )

    result: list[dict[str, Any]] = []
    page_token: str | None = None

    while True:
        params: dict[str, str | int] = {
            "maxResults": 500,
            "fields": (
                "items(name,status,zone,labels,creationTimestamp),"
                "nextPageToken"
            ),
        }

        if page_token:
            params["pageToken"] = page_token

        response = compute_api("GET", path, params=params) or {}

        for instance in response.get("items") or []:
            labels = instance.get("labels") or {}

            if labels.get("managed-by") != MANAGED_BY_LABEL:
                continue

            result.append(instance)

        page_token = response.get("nextPageToken")

        if not page_token:
            break

    return result


def list_global_runners() -> list[dict[str, Any]]:
    runners = forgejo_api(
        "GET",
        "/api/v1/admin/actions/runners?visible=false&limit=100",
    )

    # Forgejo 在没有 Runner 时可能返回 JSON null，而不是 []。
    if runners is None:
        return []

    if not isinstance(runners, list):
        preview = json.dumps(
            runners,
            ensure_ascii=False,
            separators=(",", ":"),
        )[:1000]

        raise RuntimeError(
            "Forgejo runners API returned "
            f"{type(runners).__name__}, expected list: {preview}"
        )

    return runners


def delete_runner(runner_id: str | int) -> None:
    forgejo_api(
        "DELETE",
        f"/api/v1/admin/actions/runners/{runner_id}",
        ignore_not_found=True,
    )


def delete_vm(
    name: str,
    *,
    ignore_not_found: bool = True,
) -> None:
    path = (
        f"/projects/{GCP_PROJECT_ID}/zones/{GCP_ZONE}"
        f"/instances/{name}"
    )

    operation = compute_api(
        "DELETE",
        path,
        ignore_not_found=ignore_not_found,
    )

    if operation is not None:
        wait_for_zone_operation(operation)


def delete_instance(instance: dict[str, Any]) -> None:
    name = str(instance["name"])
    labels = instance.get("labels") or {}
    runner_id = labels.get("runner-id")

    if runner_id:
        try:
            delete_runner(runner_id)
        except Exception as exc:
            log(f"Could not delete stale Runner {runner_id}: {exc}")

    log(f"Deleting VM {name}")
    delete_vm(name)


def register_runner(name: str) -> dict[str, Any]:
    response = forgejo_api(
        "POST",
        "/api/v1/admin/actions/runners",
        {
            "name": name,
            "description": (
                f"GCP global ephemeral runner for label {RUNNER_LABEL}"
            ),
            "ephemeral": True,
        },
    )

    if not isinstance(response, dict):
        raise RuntimeError("Forgejo register runner API returned invalid JSON")

    for key in ("id", "uuid", "token"):
        if not response.get(key):
            raise RuntimeError(f"Runner registration response lacks {key}")

    return response


def add_instance_metadata(
    instance_name: str,
    values: dict[str, str],
) -> None:
    instance_path = (
        f"/projects/{GCP_PROJECT_ID}/zones/{GCP_ZONE}"
        f"/instances/{instance_name}"
    )

    instance = compute_api(
        "GET",
        instance_path,
        params={"fields": "metadata"},
    )

    if instance is None:
        raise RuntimeError(f"VM {instance_name} was not found")

    current_metadata = instance.get("metadata") or {}
    fingerprint = str(current_metadata.get("fingerprint", "")).strip()

    if not fingerprint:
        raise RuntimeError(
            f"VM {instance_name} metadata lacks fingerprint"
        )

    merged: dict[str, str] = {}

    for item in current_metadata.get("items") or []:
        key = str(item.get("key", "")).strip()

        if key:
            merged[key] = str(item.get("value", ""))

    merged.update(values)

    payload = {
        "fingerprint": fingerprint,
        "items": [
            {"key": key, "value": value}
            for key, value in merged.items()
        ],
    }

    operation = compute_api(
        "POST",
        f"{instance_path}/setMetadata",
        payload=payload,
    )

    if operation is None:
        raise RuntimeError(
            f"Setting metadata on VM {instance_name} returned no operation"
        )

    wait_for_zone_operation(operation)


def create_instance() -> None:
    now = dt.datetime.now(dt.timezone.utc)
    timestamp = now.strftime("%Y%m%d-%H%M%S")
    suffix = secrets.token_hex(3)

    runner_name = f"{RUNNER_NAME_PREFIX}{timestamp}-{suffix}"
    instance_name = f"fj-{int(now.timestamp())}-{suffix}"

    runner = register_runner(runner_name)
    runner_id = str(runner["id"])
    instance_created = False

    metadata = {
        "forgejo-url": FORGEJO_URL,
        "runner-uuid": str(runner["uuid"]),
        "runner-token": str(runner["token"]),
        "runner-label": RUNNER_LABEL,
    }

    try:
        log(
            f"Creating VM {instance_name} for Runner {runner_id} "
            f"with label {RUNNER_LABEL}"
        )

        # 这里只指定 name 和 labels，metadata 留空。
        # Instance Template 中的 startup-script 因此会被保留。
        operation = compute_api(
            "POST",
            (
                f"/projects/{GCP_PROJECT_ID}/zones/{GCP_ZONE}"
                "/instances"
            ),
            params={
                "sourceInstanceTemplate": instance_template_resource(),
            },
            payload={
                "name": instance_name,
                "labels": {
                    "managed-by": MANAGED_BY_LABEL,
                    "runner-id": runner_id,
                },
            },
        )

        if operation is None:
            raise RuntimeError(
                f"Creating VM {instance_name} returned no operation"
            )

        # 请求已被 Compute Engine 接受后，即使等待阶段失败，
        # 清理流程也会尝试删除可能已创建的实例。
        instance_created = True
        wait_for_zone_operation(operation)

        # 与原来的 gcloud add-metadata 一致：
        # 先读取模板带来的 metadata，再合并动态 Runner 字段。
        log(f"Adding Runner metadata to VM {instance_name}")
        add_instance_metadata(instance_name, metadata)

    except Exception:
        if instance_created:
            try:
                log(
                    f"Deleting VM {instance_name} after "
                    "metadata setup failure"
                )
                delete_vm(instance_name)
            except Exception as cleanup_exc:
                log(
                    f"Could not delete VM {instance_name}: "
                    f"{cleanup_exc}"
                )

        try:
            delete_runner(runner_id)
        except Exception as cleanup_exc:
            log(
                f"Could not delete Runner {runner_id} "
                f"after VM failure: {cleanup_exc}"
            )

        raise


def runner_name_age_seconds(name: str) -> float | None:
    if not name.startswith(RUNNER_NAME_PREFIX):
        return None

    value = name[len(RUNNER_NAME_PREFIX):]

    try:
        timestamp_text = value.rsplit("-", 1)[0]
        created = dt.datetime.strptime(
            timestamp_text,
            "%Y%m%d-%H%M%S",
        ).replace(tzinfo=dt.timezone.utc)

        return (
            dt.datetime.now(dt.timezone.utc) - created
        ).total_seconds()

    except ValueError:
        return None


def cleanup_terminated_instances(
    instances: list[dict[str, Any]],
) -> None:
    for instance in instances:
        if instance.get("status") == "TERMINATED":
            try:
                delete_instance(instance)
            except Exception as exc:
                log(f"Failed to delete VM {instance.get('name')}: {exc}")


def cleanup_orphan_runners(
    instances: list[dict[str, Any]],
) -> None:
    instance_runner_ids = {
        str((instance.get("labels") or {}).get("runner-id"))
        for instance in instances
        if (instance.get("labels") or {}).get("runner-id")
    }

    for runner in list_global_runners():
        runner_id = str(runner.get("id", ""))
        name = str(runner.get("name", ""))
        status = str(runner.get("status", ""))

        if not name.startswith(RUNNER_NAME_PREFIX):
            continue

        if runner_id in instance_runner_ids:
            continue

        if status != "offline":
            continue

        age = runner_name_age_seconds(name)

        if age is None or age < ORPHAN_GRACE_SECONDS:
            continue

        log(f"Deleting orphan global Runner {runner_id} ({name})")

        try:
            delete_runner(runner_id)
        except Exception as exc:
            log(f"Failed to delete orphan Runner {runner_id}: {exc}")


def reconcile() -> None:
    jobs = list_jobs()
    instances = list_managed_instances()

    cleanup_terminated_instances(instances)

    # 删除动作后重新查询，避免把已删除 VM 计入容量。
    instances = list_managed_instances()

    cleanup_orphan_runners(instances)

    active_instances = [
        instance
        for instance in instances
        if instance.get("status") in ACTIVE_VM_STATES
    ]

    desired = min(len(jobs), MAX_INSTANCES)
    missing = max(0, desired - len(active_instances))

    waiting = sum(1 for job in jobs if job.get("status") == "waiting")
    running = sum(1 for job in jobs if job.get("status") == "running")

    log(
        "Reconcile: "
        f"waiting_jobs={waiting}, "
        f"running_jobs={running}, "
        f"active_vms={len(active_instances)}, "
        f"desired_vms={desired}"
    )

    for _ in range(missing):
        create_instance()


def main() -> None:
    if MAX_INSTANCES < 1:
        raise RuntimeError("MAX_INSTANCES must be at least 1")

    if POLL_INTERVAL_SECONDS < 5:
        raise RuntimeError("POLL_INTERVAL_SECONDS must be at least 5")

    authenticate_gcp()

    log(
        f"Starting scaler for {FORGEJO_URL}; "
        f"label={RUNNER_LABEL}; "
        f"max_instances={MAX_INSTANCES}; "
        "gcp_backend=compute-rest"
    )

    while True:
        try:
            reconcile()
        except Exception as exc:
            log(f"Reconcile failed: {exc}")

        time.sleep(POLL_INTERVAL_SECONDS)


if __name__ == "__main__":
    main()
PYTHON

sudo chown root:root /usr/local/forgejo-gcp-scaler/scaler.py
sudo chmod 0755 /usr/local/forgejo-gcp-scaler/scaler.py

这个版本包含几个必要的处理：Forgejo 无任务时返回的 null 会被视为空列表；创建 VM 后读取最新的 metadata fingerprint，再合并动态 Runner 字段，不会覆盖模板中的 startup-script；已停止 VM 和离线的遗留 Runner 会自动清理。

创建 Python 依赖和镜像

sudo tee /usr/local/forgejo-gcp-scaler/requirements-rest.txt >/dev/null <<'REQUIREMENTS'
google-auth>=2.38,<3
requests>=2.31,<3
REQUIREMENTS

sudo tee /usr/local/forgejo-gcp-scaler/Dockerfile >/dev/null <<'DOCKERFILE'
FROM python:3.13-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

COPY requirements-rest.txt ./requirements-rest.txt

RUN python -m pip install \
        --disable-pip-version-check \
        --no-cache-dir \
        --requirement requirements-rest.txt

COPY scaler.py ./scaler.py

ENTRYPOINT ["python", "/app/scaler.py"]
DOCKERFILE

sudo tee /usr/local/forgejo-gcp-scaler/.dockerignore >/dev/null <<'DOCKERIGNORE'
*
!Dockerfile
!requirements-rest.txt
!scaler.py
DOCKERIGNORE

sudo chown root:root \
  /usr/local/forgejo-gcp-scaler/requirements-rest.txt \
  /usr/local/forgejo-gcp-scaler/Dockerfile \
  /usr/local/forgejo-gcp-scaler/.dockerignore

sudo chmod 0644 \
  /usr/local/forgejo-gcp-scaler/requirements-rest.txt \
  /usr/local/forgejo-gcp-scaler/Dockerfile \
  /usr/local/forgejo-gcp-scaler/.dockerignore

.dockerignore 只允许构建所需的三个文件进入 Docker 构建上下文，避免把 secrets/ 发送给 Docker daemon。

创建 Docker Compose

sudo tee /usr/local/forgejo-gcp-scaler/compose.yaml >/dev/null <<'YAML'
services:
  scaler:
    build:
      context: .
      dockerfile: Dockerfile
    image: forgejo-gcp-scaler:rest
    container_name: forgejo-gcp-scaler
    init: true
    stop_grace_period: 5s
    restart: unless-stopped

    env_file:
      - /usr/local/forgejo-gcp-scaler/config/scaler.env

    environment:
      PYTHONDONTWRITEBYTECODE: "1"
      PYTHONUNBUFFERED: "1"

    volumes:
      - /usr/local/forgejo-gcp-scaler/secrets/gcp.json:/run/secrets/gcp.json:ro
      - /usr/local/forgejo-gcp-scaler/secrets/forgejo.token:/run/secrets/forgejo.token:ro

    tmpfs:
      - /tmp

    cap_drop:
      - ALL

    security_opt:
      - no-new-privileges:true
YAML

sudo chown root:root /usr/local/forgejo-gcp-scaler/compose.yaml
sudo chmod 0644 /usr/local/forgejo-gcp-scaler/compose.yaml

启动：

cd /usr/local/forgejo-gcp-scaler

sudo docker compose up -d --build
sudo docker compose logs -f scaler

正常情况下可以看到：

Starting scaler for https://forgejo.example.com; label=gcp-linux; max_instances=2; gcp_backend=compute-rest
Reconcile: waiting_jobs=0, running_jobs=0, active_vms=0, desired_vms=0

如果修改了 scaler.env，不能只执行 docker compose restart。容器环境变量在创建时确定，需要重新创建容器：

sudo docker compose up -d --force-recreate scaler

在仓库中使用

任意仓库创建 .forgejo/workflows/build.yml：

name: Build on GCP

on:
  workflow_dispatch:

  push:
    branches:
      - main

jobs:
  build:
    runs-on: gcp-linux
    timeout-minutes: 90

    steps:
      - name: Checkout
        uses: actions/checkout@v6
        with:
          persist-credentials: false

      - name: Build
        shell: bash
        run: |
          set -Eeuo pipefail
          ./build.sh

不需要为仓库添加 GCP JSON 或 Forgejo Admin Token。仓库自己的部署密钥仍然按正常方式放进 Actions Secrets。

任务进入队列后，scaler 日志会出现：

Reconcile: waiting_jobs=1, running_jobs=0, active_vms=0, desired_vms=1
Creating VM fj-...

在 Forgejo 的：

Site Administration
→ Actions
→ Runners

可以看到名称以 gcp-ephemeral- 开头的临时 Runner。任务完成后，它会退出并关机；scaler 下一轮轮询时会删除 VM 和可能残留的 Runner 注册。

验证与排错

先在 scaler VPS 上测试 Forgejo API：

sudo TOKEN="$(cat /usr/local/forgejo-gcp-scaler/secrets/forgejo.token)"

curl \
  --fail \
  --silent \
  --show-error \
  -H "Authorization: token ${TOKEN}" \
  "https://forgejo.example.com/api/v1/admin/actions/runners/jobs?labels=gcp-linux"

unset TOKEN

没有任务时可能返回 []，Forgejo 15 也可能返回 null。如果返回 HTML 或 403，通常是 Token 权限或 Cloudflare 规则有问题。

查看 scaler：

cd /usr/local/forgejo-gcp-scaler
sudo docker compose ps
sudo docker compose logs --tail=200 scaler

查看临时 VM：

gcloud compute instances list \
  --project="$PROJECT_ID" \
  --filter='labels.managed-by=forgejo-gcp-scaler'

VM 启动失败时，可以查看串口日志：

gcloud compute instances get-serial-port-output VM_NAME \
  --project="$PROJECT_ID" \
  --zone="$ZONE" \
  --port=1 | tail -n 200

如果日志显示 No startup scripts to run，说明创建 VM 时覆盖了模板 metadata。本文中的 scaler 会先创建 VM，再通过 instances.setMetadata REST API 读取并提交最新 fingerprint，合并动态字段，不能把两步合并成创建时只提交动态 metadata。

如果 Job 容器出现 Could not resolve host，检查 DOCKER-USER：

iptables -S DOCKER-USER

规则应只拒绝 169.254.169.254 的 TCP 80 和 443，不能拒绝整个 IP。

费用与维护

Project、Service Account、VPC、Subnet 和 Instance Template 本身不会因为闲置而持续产生计算费用。主要费用来自：

Spot VM 运行期间的 CPU 和内存；
VM 存在期间的启动磁盘和外部 IPv4；
Custom Image 的存储；
构建产生的公网出站流量。

建议同时设置：

MAX_INSTANCES=1 或 2；
GCP Billing Budget 告警；
较低的区域 vCPU 配额；
VM 的 --max-run-duration=2h；
工作流的 timeout-minutes。

自定义镜像不会自动安装后续的 Debian 安全更新，需要定期重新烘焙，并创建新的 Instance Template。旧镜像会持续产生存储费用，确认新版本稳定后即可删除。

定期检查是否有遗留资源：

gcloud compute instances list --project="$PROJECT_ID"
gcloud compute disks list --project="$PROJECT_ID"
gcloud compute images list --project="$PROJECT_ID" --no-standard-images
gcloud compute instance-templates list --project="$PROJECT_ID"
gcloud compute addresses list --project="$PROJECT_ID"

总结

最终仓库只需要写：

runs-on: gcp-linux

常驻 scaler 根据等待和运行中的 Job 数量创建 GCP Spot VM；临时 VM 使用全局 ephemeral Runner，只执行一个任务，随后关机并删除。这样不需要为每个仓库重复保存 GCP 凭据，也不需要长期运行一台空闲 Runner。