ci(runner): guard gitea pressure triggers [skip ci]
This commit is contained in:
@@ -438,8 +438,26 @@ root live artifact 與 lane registration 檔名都屬 restore source,
|
||||
復活。
|
||||
|
||||
未完成 runner 搬遷、硬限流、smoke 排程前,不得解除 mask、恢復泛用 runner label、
|
||||
恢復 cd-lane / drain ELF,或把 host pressure gate 預設改成 warn-only;`cd.yaml` /
|
||||
`code-review.yaml` push trigger 維持 manual-only。
|
||||
恢復 cd-lane / drain ELF,或把 host pressure gate 預設改成 warn-only;所有會命中
|
||||
`awoooi-ubuntu` / `awoooi-host` 的 `.gitea/workflows` 都不得保留 `push`、
|
||||
`pull_request` 或 `pull_request_target` 自動事件。
|
||||
|
||||
### 第九層修復: workflow pressure source guard
|
||||
|
||||
2026-06-28 補上 source guard:
|
||||
|
||||
```bash
|
||||
python3 ops/runner/guard-gitea-runner-pressure.py --root .
|
||||
```
|
||||
|
||||
此 guard 只讀 repo 內 `.gitea/workflows/*.yml` / `.yaml`,禁止兩類回歸:
|
||||
|
||||
1. `push` / `pull_request` / `pull_request_target` 自動事件命中 `awoooi-ubuntu` 或
|
||||
`awoooi-host`。
|
||||
2. Gitea workflow 恢復 `ubuntu-latest`、`ubuntu-*` 或 `self-hosted` 泛用 label。
|
||||
|
||||
`scripts/ops/ansible-validate.sh` 會執行同一 guard。若要恢復自動事件,必須先有
|
||||
runner 搬遷或非 110 硬限流的 source-of-truth diff、rollback 與 post-apply verifier。
|
||||
|
||||
---
|
||||
版本: v2.0 | 更新: 2026-03-29 | 作者: Claude Code
|
||||
|
||||
186
ops/runner/guard-gitea-runner-pressure.py
Normal file
186
ops/runner/guard-gitea-runner-pressure.py
Normal file
@@ -0,0 +1,186 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Guard Gitea workflows from reintroducing 110 runner pressure triggers.
|
||||
|
||||
This is a source guard only. It reads committed `.gitea/workflows` files and
|
||||
fails when automatic branch events target 110 incident runner labels, or when a
|
||||
generic runner label is reintroduced.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
AUTO_BRANCH_EVENTS = {"push", "pull_request", "pull_request_target"}
|
||||
INCIDENT_RUNNER_LABELS = {"awoooi-ubuntu", "awoooi-host"}
|
||||
GENERIC_LABELS = {"ubuntu-latest", "self-hosted"}
|
||||
|
||||
RUNS_ON_RE = re.compile(r"^\s*runs-on:\s*(?P<value>.+?)\s*$")
|
||||
ON_RE = re.compile(r"^on:\s*(?P<value>.*)$")
|
||||
ON_CHILD_EVENT_RE = re.compile(r"^(?P<indent>\s+)(?P<event>[A-Za-z_][A-Za-z0-9_-]*)\s*:")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WorkflowInfo:
|
||||
path: Path
|
||||
events: set[str]
|
||||
labels: list[tuple[int, str]]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Violation:
|
||||
path: Path
|
||||
line: int
|
||||
label: str
|
||||
reason: str
|
||||
events: set[str]
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--root", type=Path, default=Path.cwd(), help="Repository root")
|
||||
parser.add_argument(
|
||||
"--workflow-dir",
|
||||
default=".gitea/workflows",
|
||||
help="Workflow directory relative to --root",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def remove_inline_comment(value: str) -> str:
|
||||
return value.split("#", 1)[0].rstrip()
|
||||
|
||||
|
||||
def strip_inline_comment(value: str) -> str:
|
||||
return remove_inline_comment(value).strip()
|
||||
|
||||
|
||||
def parse_inline_events(value: str) -> set[str]:
|
||||
clean = strip_inline_comment(value).strip()
|
||||
if not clean:
|
||||
return set()
|
||||
if clean.startswith("[") and clean.endswith("]"):
|
||||
return {part.strip().strip("'\"") for part in clean[1:-1].split(",") if part.strip()}
|
||||
if clean.startswith("{") and clean.endswith("}"):
|
||||
return {
|
||||
part.split(":", 1)[0].strip().strip("'\"")
|
||||
for part in clean[1:-1].split(",")
|
||||
if part.strip()
|
||||
}
|
||||
return {clean.strip().strip("'\"")}
|
||||
|
||||
|
||||
def parse_events(lines: list[str]) -> set[str]:
|
||||
events: set[str] = set()
|
||||
for index, line in enumerate(lines):
|
||||
match = ON_RE.match(remove_inline_comment(line))
|
||||
if not match:
|
||||
continue
|
||||
inline_events = parse_inline_events(match.group("value"))
|
||||
if inline_events:
|
||||
events.update(inline_events)
|
||||
continue
|
||||
|
||||
for child in lines[index + 1 :]:
|
||||
if not child.strip() or child.lstrip().startswith("#"):
|
||||
continue
|
||||
if child[0] not in {" ", "\t"}:
|
||||
break
|
||||
child_match = ON_CHILD_EVENT_RE.match(remove_inline_comment(child))
|
||||
if child_match:
|
||||
events.add(child_match.group("event"))
|
||||
return events
|
||||
|
||||
|
||||
def normalize_runs_on(value: str) -> list[str]:
|
||||
clean = strip_inline_comment(value).strip()
|
||||
if clean.startswith("[") and clean.endswith("]"):
|
||||
return [part.strip().strip("'\"") for part in clean[1:-1].split(",") if part.strip()]
|
||||
return [clean.strip().strip("'\"")]
|
||||
|
||||
|
||||
def parse_workflow(path: Path) -> WorkflowInfo:
|
||||
lines = path.read_text(encoding="utf-8").splitlines()
|
||||
labels: list[tuple[int, str]] = []
|
||||
for line_number, line in enumerate(lines, start=1):
|
||||
match = RUNS_ON_RE.match(line)
|
||||
if not match:
|
||||
continue
|
||||
for label in normalize_runs_on(match.group("value")):
|
||||
labels.append((line_number, label))
|
||||
return WorkflowInfo(path=path, events=parse_events(lines), labels=labels)
|
||||
|
||||
|
||||
def label_is_generic(label: str) -> bool:
|
||||
if label in GENERIC_LABELS:
|
||||
return True
|
||||
return label.startswith("ubuntu-")
|
||||
|
||||
|
||||
def inspect(workflow_dir: Path) -> tuple[list[WorkflowInfo], list[Violation]]:
|
||||
workflows = [parse_workflow(path) for path in sorted(workflow_dir.glob("*.y*ml"))]
|
||||
violations: list[Violation] = []
|
||||
|
||||
for workflow in workflows:
|
||||
auto_events = workflow.events & AUTO_BRANCH_EVENTS
|
||||
for line, label in workflow.labels:
|
||||
if label_is_generic(label):
|
||||
violations.append(
|
||||
Violation(
|
||||
path=workflow.path,
|
||||
line=line,
|
||||
label=label,
|
||||
reason="generic_runner_label_reintroduced",
|
||||
events=workflow.events,
|
||||
)
|
||||
)
|
||||
if auto_events and label in INCIDENT_RUNNER_LABELS:
|
||||
violations.append(
|
||||
Violation(
|
||||
path=workflow.path,
|
||||
line=line,
|
||||
label=label,
|
||||
reason="auto_branch_event_targets_110_incident_runner",
|
||||
events=workflow.events,
|
||||
)
|
||||
)
|
||||
return workflows, violations
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
root = args.root.expanduser().resolve()
|
||||
workflow_dir = root / args.workflow_dir
|
||||
if not workflow_dir.exists():
|
||||
print(f"GITEA_RUNNER_PRESSURE_GUARD_BLOCKED workflow_dir_missing={workflow_dir}")
|
||||
return 2
|
||||
|
||||
workflows, violations = inspect(workflow_dir)
|
||||
if violations:
|
||||
print("GITEA_RUNNER_PRESSURE_GUARD_BLOCKED")
|
||||
for item in violations:
|
||||
rel_path = item.path.relative_to(root)
|
||||
events = ",".join(sorted(item.events)) or "none"
|
||||
print(
|
||||
f"{rel_path}:{item.line} label={item.label} "
|
||||
f"reason={item.reason} events={events}"
|
||||
)
|
||||
return 1
|
||||
|
||||
scheduled = sum(1 for workflow in workflows if "schedule" in workflow.events)
|
||||
print(
|
||||
"GITEA_RUNNER_PRESSURE_GUARD_OK "
|
||||
f"workflow_files={len(workflows)} "
|
||||
f"scheduled_workflows={scheduled} "
|
||||
"auto_branch_events_on_110=0 "
|
||||
"generic_runner_labels=0"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user