220 lines
8.8 KiB
Python
220 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Persistent, narrowly-scoped blunix deployment runner.
|
|
|
|
The builder only creates a durable request. This process owns the switch and
|
|
therefore is not a child of agent-codex-builder.service.
|
|
"""
|
|
import json
|
|
import fcntl
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
|
|
def payload_path(directory, job_id):
|
|
return Path(directory) / f"{job_id}.json"
|
|
|
|
|
|
def write_payload(path, payload):
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
|
|
try:
|
|
with temporary.open("x", encoding="utf-8") as stream:
|
|
temporary.chmod(0o600)
|
|
stream.write(json.dumps(payload, indent=2) + "\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
temporary.replace(path)
|
|
fd = os.open(path.parent, os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
|
|
|
|
def run_switch(request, environment=None, runner=None):
|
|
environment = environment or os.environ
|
|
runner = runner or subprocess.run
|
|
if request.get("host") != "blunix":
|
|
raise ValueError("Deployment ist ausschließlich für blunix erlaubt")
|
|
repo = Path(request["repo"])
|
|
expected_repo = Path(environment.get("AGENT_BUILDER_DEPLOY_REPO", "/home/drone/nixos-config"))
|
|
if repo != expected_repo:
|
|
raise ValueError(f"Deployment-Repository muss exakt {expected_repo} sein")
|
|
|
|
timeout = int(environment.get("AGENT_BUILDER_DEPLOY_TIMEOUT_SECONDS", "1200"))
|
|
revision = str(request.get("commit_hash", ""))
|
|
if not re.fullmatch(r"[0-9a-f]{7,40}", revision):
|
|
raise ValueError("Deployment requires a reviewed commit hash")
|
|
resolved = runner(
|
|
["git", "-C", str(repo), "rev-parse", "--verify", f"{revision}^{{commit}}"],
|
|
capture_output=True, text=True, timeout=timeout, check=False,
|
|
)
|
|
exact_revision = resolved.stdout.strip()
|
|
if resolved.returncode or not re.fullmatch(r"[0-9a-f]{40}", exact_revision):
|
|
return False, "Reviewed commit is unavailable or ambiguous; no activation attempted."
|
|
request["resolved_commit_hash"] = exact_revision
|
|
state_dir = environment.get("AGENT_BUILDER_STATE_DIR")
|
|
manifest = Path(state_dir) / "last-verified-deployment.json" if state_dir else None
|
|
if manifest and manifest.exists():
|
|
previous = json.loads(manifest.read_text(encoding="utf-8"))["deployed_commit_hash"]
|
|
ancestry = runner(
|
|
["git", "-C", str(repo), "merge-base", "--is-ancestor", previous, exact_revision],
|
|
capture_output=True, text=True, timeout=timeout, check=False,
|
|
)
|
|
if ancestry.returncode:
|
|
return False, "Automatic deployment would roll back or diverge from the last verified revision; manual reconciliation required."
|
|
|
|
source = f"git+file://{repo}?rev={exact_revision}"
|
|
build = runner(
|
|
["nix", "build", "--no-link", "--print-out-paths", source + "#nixosConfigurations.blunix.config.system.build.toplevel"],
|
|
capture_output=True, text=True, timeout=timeout, check=False,
|
|
)
|
|
expected_system = build.stdout.strip()
|
|
if build.returncode or not re.fullmatch(r"/nix/store/[^/\s]+", expected_system):
|
|
return False, "Approved revision failed to build; no activation attempted.\n" + build.stderr[-4000:]
|
|
command = [
|
|
environment["AGENT_BUILDER_SUDO_BIN"],
|
|
environment["AGENT_BUILDER_ENV_BIN"],
|
|
"HOME=/tmp",
|
|
environment["AGENT_BUILDER_NIXOS_REBUILD_BIN"],
|
|
"switch",
|
|
"--flake",
|
|
source + "#blunix",
|
|
]
|
|
result = runner(
|
|
command,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=int(environment.get("AGENT_BUILDER_DEPLOY_TIMEOUT_SECONDS", "1200")),
|
|
check=False,
|
|
)
|
|
output = (result.stdout + result.stderr).strip()
|
|
if result.returncode != 0:
|
|
return False, output
|
|
|
|
health = runner(
|
|
[
|
|
environment.get(
|
|
"AGENT_BUILDER_SYSTEMCTL_BIN",
|
|
"/run/current-system/sw/bin/systemctl",
|
|
),
|
|
"is-active",
|
|
"board-agent-trigger.service",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
health_output = (health.stdout + health.stderr).strip()
|
|
combined = "\n".join(part for part in (output, health_output) if part)
|
|
healthy = health.returncode == 0 and health.stdout.strip() == "active"
|
|
if healthy:
|
|
generation = runner(
|
|
["readlink", "-f", "/run/current-system"], capture_output=True,
|
|
text=True, timeout=30, check=False,
|
|
)
|
|
closure = generation.stdout.strip()
|
|
if generation.returncode or closure != expected_system:
|
|
return False, combined + "\nActivated system does not match the approved revision's built closure."
|
|
request["activated_system"] = closure
|
|
request["deployed_commit_hash"] = exact_revision
|
|
return healthy, combined
|
|
|
|
|
|
def process(path, environment=None, runner=None):
|
|
environment = environment or os.environ
|
|
with (path.parent / '.delivery.lock').open('a+') as handle:
|
|
fcntl.flock(handle, fcntl.LOCK_EX)
|
|
request = json.loads(path.read_text(encoding="utf-8"))
|
|
write_payload(path.parent / '.claimed' / path.name, {'claimed': True})
|
|
state_dir = Path(environment["AGENT_BUILDER_STATE_DIR"])
|
|
state_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
state_path = state_dir / path.name
|
|
if state_path.exists():
|
|
saved = json.loads(state_path.read_text(encoding="utf-8"))
|
|
if saved.get("state") in {"deployed", "deploy_failed"}:
|
|
_write_completion(saved)
|
|
path.unlink(missing_ok=True)
|
|
return saved
|
|
if request.get("state") == "deployed":
|
|
path.unlink(missing_ok=True)
|
|
return request
|
|
max_retries = int(environment.get("AGENT_BUILDER_DEPLOY_MAX_RETRIES", "3"))
|
|
backoff = float(environment.get("AGENT_BUILDER_DEPLOY_BACKOFF_SECONDS", "5"))
|
|
retries = int(request.get("retry_count", 0))
|
|
ok = False
|
|
output = ""
|
|
while retries < max_retries and not ok:
|
|
retries += 1
|
|
request.update(state="deploying", retry_count=retries, updated_at=time.time())
|
|
write_payload(path, request)
|
|
write_payload(state_path, request)
|
|
try:
|
|
ok, output = run_switch(request, environment, runner)
|
|
except (OSError, ValueError, subprocess.TimeoutExpired) as error:
|
|
output = str(error)
|
|
if not ok and retries < max_retries and backoff:
|
|
time.sleep(backoff * retries)
|
|
if not ok and retries >= max_retries:
|
|
output = output or "maximale Deployment-Wiederholungen erreicht"
|
|
request.update(
|
|
state="deployed" if ok else "deploy_failed",
|
|
output=output[-4000:],
|
|
updated_at=time.time(),
|
|
)
|
|
write_payload(state_path, request)
|
|
if ok:
|
|
write_payload(state_dir / "last-verified-deployment.json", {
|
|
"deployed_commit_hash": request["deployed_commit_hash"],
|
|
"activated_system": request["activated_system"], "job_id": request["job_id"],
|
|
})
|
|
_write_completion(request)
|
|
path.unlink(missing_ok=True)
|
|
return request
|
|
|
|
|
|
def _write_completion(request):
|
|
completion_dir = request.get("on_complete_dir")
|
|
if completion_dir:
|
|
target = Path(completion_dir) / f"{request['job_id']}.json"
|
|
if not target.exists():
|
|
target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
write_payload(
|
|
target,
|
|
{
|
|
"job_id": request["job_id"],
|
|
"status": "committed" if request["state"] == "deployed" else "deploy_failed",
|
|
"deployment_status": request["state"],
|
|
"deployed_commit_hash": request.get("deployed_commit_hash"),
|
|
"activated_system": request.get("activated_system"),
|
|
"failure_detail": request.get("output") if request["state"] != "deployed" else None,
|
|
"commits": [{"repo": Path(request["repo"]).name, "branch": "dev", "hash": request["commit_hash"]}],
|
|
"completed_at": time.time(),
|
|
"runtime": request.get("runtime", "codex"),
|
|
"pipeline_run_id": request.get("pipeline_run_id"),
|
|
"board_slug": request.get("board_slug"),
|
|
"configured_profile": request.get("configured_profile") or {},
|
|
"telemetry": request.get("telemetry") or {},
|
|
},
|
|
)
|
|
|
|
|
|
def main():
|
|
queue = Path(os.environ["AGENT_BUILDER_DEPLOY_QUEUE_DIR"])
|
|
for path in sorted(queue.glob("*.json")):
|
|
result = process(path)
|
|
print(f"{result['job_id']}: {result['state']}", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|