124 lines
4.2 KiB
Python
124 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Conclude a board job when systemd fails before worker.py can run."""
|
|
|
|
import json
|
|
import state_store
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,60}$")
|
|
JOB_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,120}$")
|
|
MAX_ERROR_DETAIL = 240
|
|
|
|
|
|
def now_iso():
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def clean_detail(value):
|
|
value = re.sub(r"[\x00-\x1f\x7f]+", " ", str(value or ""))
|
|
value = re.sub(r"\s+", " ", value).strip()
|
|
return value[:MAX_ERROR_DETAIL] or "Systemd konnte den Worker nicht starten."
|
|
|
|
|
|
def load_json(path):
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return None
|
|
return value if isinstance(value, dict) else None
|
|
|
|
|
|
def write_json(path, value):
|
|
tmp = path.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
tmp.chmod(0o600)
|
|
tmp.replace(path)
|
|
|
|
|
|
def candidate_jobs():
|
|
candidates = []
|
|
for env_name in ("BOARD_INBOX_DIR", "BOARD_PROCESSING_DIR"):
|
|
directory = Path(os.environ[env_name])
|
|
if not directory.is_dir():
|
|
continue
|
|
for path in sorted(directory.glob("*.json")):
|
|
payload = load_json(path)
|
|
if not payload:
|
|
continue
|
|
job_id = str(payload.get("job_id", ""))
|
|
slug = str(payload.get("slug", ""))
|
|
intent_path = str(payload.get("intent_path", ""))
|
|
source = str(payload.get("source", "intent"))
|
|
if (
|
|
not JOB_ID_RE.fullmatch(job_id)
|
|
or not SLUG_RE.fullmatch(slug)
|
|
or source not in ("intent", "research")
|
|
or not re.fullmatch(r"docs/(?:intent|research-intent)/[a-z0-9][a-z0-9-]{0,60}\.md", intent_path)
|
|
or Path(intent_path).stem != slug
|
|
):
|
|
continue
|
|
candidates.append((path, payload, job_id, slug, source))
|
|
return candidates
|
|
|
|
|
|
@state_store.atomic
|
|
def update_state(slug, job_id, stage, detail):
|
|
state_path = Path(os.environ["BOARD_STATE_FILE"])
|
|
state = state_store.load()
|
|
cards = state.setdefault("cards", {})
|
|
card = cards.setdefault(slug, {})
|
|
if card.get("last_job_id") != job_id or card.get("last_run_status") in ("done", "no_changes"):
|
|
return False
|
|
card.update(
|
|
{
|
|
"last_job_id": job_id,
|
|
"last_run_status": "failed",
|
|
"last_run_at": now_iso(),
|
|
"failure_stage": stage,
|
|
"failure_detail": clean_detail(detail),
|
|
}
|
|
)
|
|
state_store.save(state)
|
|
return True
|
|
|
|
|
|
def main():
|
|
# systemd runs ExecStopPost after successful services as well. In that
|
|
# case a newly queued job must never be mistaken for the completed run.
|
|
if os.environ.get("SERVICE_RESULT") == "success":
|
|
return 0
|
|
stage = os.environ.get("BOARD_FAILURE_STAGE", "pre_start")
|
|
if stage not in ("pre_start", "sandbox_setup", "worker"):
|
|
stage = "pre_start"
|
|
candidates = candidate_jobs()
|
|
if len(candidates) != 1:
|
|
print(
|
|
f"board failure handler: expected exactly one valid queue candidate, found {len(candidates)}",
|
|
file=sys.stderr,
|
|
)
|
|
return 0
|
|
path, payload, job_id, slug, _source = candidates[0]
|
|
service_result = os.environ.get("SERVICE_RESULT", "unknown")
|
|
exit_code = os.environ.get("EXIT_CODE", "unknown")
|
|
exit_status = os.environ.get("EXIT_STATUS", "unknown")
|
|
detail = os.environ.get(
|
|
"BOARD_FAILURE_DETAIL",
|
|
f"Systemd konnte den Worker nicht starten ({service_result}, {exit_code} {exit_status}).",
|
|
)
|
|
if update_state(slug, job_id, stage, detail):
|
|
failed = Path(os.environ["BOARD_FAILED_DIR"])
|
|
failed.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
payload.update({"failed_at": now_iso(), "failure_stage": stage, "error": clean_detail(detail)})
|
|
write_json(path, payload)
|
|
path.replace(failed / path.name)
|
|
else:
|
|
print("board failure handler: job is no longer the card's current run", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|