| 1 | import json |
| 2 | import os |
| 3 | import subprocess |
| 4 | import urllib.request |
| 5 | |
| 6 | |
| 7 | def nomad(path): |
| 8 | request = urllib.request.Request( |
| 9 | "http://127.0.0.1:4646/v1/" + path, |
| 10 | headers={"X-Nomad-Token": os.environ["NOMAD_TOKEN"]}, |
| 11 | ) |
| 12 | with urllib.request.urlopen(request, timeout=5) as response: |
| 13 | return json.load(response) |
| 14 | |
| 15 | |
| 16 | def instance(job="evil-forgejo"): |
| 17 | running = [item["ID"] for item in nomad(f"job/{job}/allocations") |
| 18 | if item["ClientStatus"] == "running" and item["DesiredStatus"] == "run"] |
| 19 | if len(running) != 1: |
| 20 | raise ValueError(f"{job} needs one running allocation") |
| 21 | allocation = running[0] |
| 22 | services = [item for item in nomad(f"service/{job}-forgejo") if item["AllocID"] == allocation] |
| 23 | if len(services) != 1: |
| 24 | raise ValueError(f"{job} has no Forgejo listener") |
| 25 | result = subprocess.run( |
| 26 | ["podman", "--url", "unix:///run/podman/podman.sock", "ps", "--format", "{{.ID}} {{.Names}}"], |
| 27 | check=True, capture_output=True, text=True, |
| 28 | ) |
| 29 | containers = [parts[0] for line in result.stdout.splitlines() |
| 30 | if len(parts := line.split()) == 2 and parts[1] == "forgejo-" + allocation] |
| 31 | if len(containers) != 1: |
| 32 | raise ValueError(f"{job} Forgejo task is unavailable") |
| 33 | return containers[0], f"http://{services[0]['Address']}:{services[0]['Port']}" |