160 lines
7.3 KiB
Python
160 lines
7.3 KiB
Python
"""Pull run requests from Git; attempt each request at most once, never retry a scrape."""
|
|
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
import uuid
|
|
|
|
|
|
IMAGE_REPOSITORY = "gitea.vandeplanque.com/tonio/car-tracker"
|
|
|
|
|
|
def validate_request(data):
|
|
if not isinstance(data, dict) or not all(isinstance(data.get(key), str)
|
|
for key in ("request_id", "revision", "image")):
|
|
raise ValueError("Invalid request types")
|
|
if set(data) != {"schema_version", "request_id", "revision", "image"} or data["schema_version"] != 1:
|
|
raise ValueError("Invalid request schema")
|
|
if str(uuid.UUID(data["request_id"])) != data["request_id"]:
|
|
raise ValueError("Invalid request identifier")
|
|
if not re.fullmatch(r"[0-9a-f]{40}", data["revision"]):
|
|
raise ValueError("Invalid revision")
|
|
if not re.fullmatch(re.escape(IMAGE_REPOSITORY) + r"@sha256:[0-9a-f]{64}", data["image"]):
|
|
raise ValueError("Invalid image")
|
|
return data
|
|
|
|
|
|
def atomic_json(path, data):
|
|
with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as output:
|
|
temporary = Path(output.name)
|
|
try:
|
|
json.dump(data, output, indent=2)
|
|
output.write("\n")
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
except BaseException:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|
|
os.replace(temporary, path)
|
|
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
|
|
|
|
def command(*args, env=None, capture=True):
|
|
return subprocess.run(args, env=env, check=True, text=True,
|
|
stdout=subprocess.PIPE if capture else None,
|
|
stderr=subprocess.PIPE if capture else None).stdout
|
|
|
|
|
|
def compose_environment(image):
|
|
env = os.environ.copy()
|
|
# Shell variables override --env-file; don't inherit stale app settings.
|
|
for key in ("APP_UID", "APP_GID", "RAW_DATA_DIR", "DB_CONFIG_DIR", "APP_SUBNET",
|
|
"PGHOST", "PGPORT", "PGDATABASE", "PGUSER", "PGSSLMODE", "COMPOSE_FILE",
|
|
"COMPOSE_PROJECT_NAME", "COMPOSE_PROFILES", "COMPOSE_ENV_FILES"):
|
|
env.pop(key, None)
|
|
for key in list(env):
|
|
if key.startswith("COLLECT_"):
|
|
env.pop(key)
|
|
env["CAR_TRACKER_IMAGE"] = image
|
|
return env
|
|
|
|
|
|
def attempt(request, state, compose_file, env_file, private_log=False, app_command=None):
|
|
outcome = state / f"{request['request_id']}.json"
|
|
if outcome.exists():
|
|
recorded = json.loads(outcome.read_text())
|
|
if recorded["request"] != request:
|
|
raise ValueError("Request identifier reused with different content")
|
|
if recorded.get("status") != "finished" or not isinstance(recorded.get("exit_code"), int):
|
|
raise ValueError("Previous request outcome is unknown or manually resolved; do not retry")
|
|
print(f"Previously finished request {request['request_id']}: exit {recorded['exit_code']}", flush=True)
|
|
return recorded["exit_code"]
|
|
env = compose_environment(request["image"])
|
|
compose = ["docker", "compose", "--env-file", str(env_file), "-f", str(compose_file)]
|
|
# Fail-closed checks occur before consuming a request. Network/pull/config
|
|
# failures here may be retried: no collector has started.
|
|
command(*compose, "config", "--quiet", env=env)
|
|
if command(*compose, "ps", "--all", "--status", "running", "--quiet", env=env).strip():
|
|
raise ValueError("An application job is already running")
|
|
command("docker", "pull", request["image"])
|
|
image = json.loads(command("docker", "image", "inspect", request["image"]))[0]
|
|
if (image["Config"].get("Labels") or {}).get("org.opencontainers.image.revision") != request["revision"]:
|
|
raise ValueError("Image revision mismatch")
|
|
# Persist before launch. A crash here can lose a run, but can never silently
|
|
# repeat one. Unknown/failed attempts require explicit investigation.
|
|
result = {"request": request, "status": "attempted", "exit_code": None}
|
|
atomic_json(outcome, result)
|
|
print(f"Starting request {request['request_id']}", flush=True)
|
|
try:
|
|
args = [*compose, "run", "--rm", "--no-deps", "--pull", "never",
|
|
"--name", f"car-tracker-run-{request['request_id']}", "app"]
|
|
if app_command is not None:
|
|
args.extend(app_command)
|
|
if private_log:
|
|
log_path = state / f"{request['request_id']}.log"
|
|
descriptor = os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
|
|
os.fchmod(descriptor, 0o600)
|
|
with os.fdopen(descriptor, "a") as output:
|
|
child = subprocess.run(args, env=env, stdout=output, stderr=subprocess.STDOUT)
|
|
else:
|
|
child = subprocess.run(args, env=env)
|
|
result.update(status="finished", exit_code=child.returncode)
|
|
atomic_json(outcome, result)
|
|
print(f"Request {request['request_id']} finished with pipeline exit {child.returncode}", flush=True)
|
|
return child.returncode if child.returncode >= 0 else 70
|
|
except (OSError, KeyboardInterrupt):
|
|
# Keep attempted/unknown instead of retrying: the child or DB transaction
|
|
# may still be running or have committed. Recovery uses the raw manifest.
|
|
print("Run outcome unknown; investigate the container and manifest before another request.", flush=True)
|
|
raise
|
|
|
|
|
|
def reconcile(root):
|
|
state = root / "state/requests"
|
|
state.mkdir(parents=True, mode=0o700, exist_ok=True)
|
|
with (root / "state/agent.lock").open("a") as lock:
|
|
try:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError:
|
|
return
|
|
for outcome in state.glob("*.json"):
|
|
if json.loads(outcome.read_text()).get("status") == "attempted":
|
|
raise ValueError("An earlier run has an unknown outcome; resolve it before consuming more requests")
|
|
repo = root / "desired.git"
|
|
if not repo.exists():
|
|
raise ValueError("Initialize the dedicated read-only Git checkout first")
|
|
# No git checkout, reset, or modification of a developer's worktree.
|
|
command("git", "-C", str(repo), "fetch", "origin",
|
|
"+refs/heads/gitops:refs/remotes/origin/gitops")
|
|
commits = command("git", "-C", str(repo), "log", "--reverse", "--format=%H",
|
|
"refs/remotes/origin/gitops", "--", "deploy/request.json").splitlines()
|
|
# Read every request commit, not just the latest: two button presses
|
|
# between polls must not silently discard one request.
|
|
for commit in commits:
|
|
raw = command("git", "-C", str(repo), "show", f"{commit}:deploy/request.json")
|
|
request = validate_request(json.loads(raw))
|
|
attempt(request, state, root / "compose.runtime.yaml", root / "app.env")
|
|
|
|
|
|
def main():
|
|
os.umask(0o077)
|
|
root = Path(os.environ.get("CAR_TRACKER_STACK_DIR", str(Path.home() / "containers/car-tracker"))).resolve()
|
|
try:
|
|
reconcile(root)
|
|
return 0
|
|
except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError):
|
|
print("Agent failed: check local configuration, Git/registry access and request state. No automatic scrape retry.")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|