1418 lines
61 KiB
Python
1418 lines
61 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Proof of concept: one controller, one agent, on this box.
|
|
|
|
Demonstrates the four things in SPEC §7 whose failure modes are silent, which is
|
|
why they are worth proving rather than asserting:
|
|
|
|
1. **A node is online because it heartbeated.** Until then it is ``provisioning``
|
|
and takes no work. (I24)
|
|
2. **Two agents claiming at once get different jobs.** ``FOR UPDATE SKIP LOCKED``.
|
|
(I25)
|
|
3. **A job whose holder was killed comes back.** The lease is what releases it,
|
|
because a killed worker releases nothing. (I26)
|
|
4. **A slow device is eligible, an impossible one is blocked.** A model that does
|
|
not fit in VRAM runs on system RAM, slowly. A model nothing can run is
|
|
``blocked`` with a reason rather than left looking like it is progressing.
|
|
|
|
And one boundary, demonstrated rather than described: **the agent runs as a
|
|
different unix account and has no database credential at all.** Everything it
|
|
knows arrives over HTTP and everything it produces leaves over HTTP. If it could
|
|
read the controller's environment file, the API would be decorative.
|
|
|
|
Run it:
|
|
|
|
umbratrans_poc --controller http://127.0.0.1:8443 --agent-user umbra-poc
|
|
|
|
Nothing here touches the deployment database. It is given a DSN and refuses to
|
|
run if that DSN names a database whose name does not look like a scratch one,
|
|
because this program truncates.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import pwd
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
#: Refuse to point at anything that is not obviously a scratch database. This
|
|
#: program truncates, and the guard is the only thing between a proof of concept
|
|
#: and a destroyed deployment.
|
|
SCRATCH_MARKERS = ("scratch", "poc", "test", "tmp", "dev")
|
|
|
|
|
|
def _fail(message: str) -> int:
|
|
print(f"\n\033[31mFAIL\033[0m {message}", flush=True)
|
|
return 1
|
|
|
|
|
|
def _pass(message: str) -> None:
|
|
print(f" \033[32mok\033[0m {message}", flush=True)
|
|
|
|
|
|
def _head(message: str) -> None:
|
|
print(f"\n\033[1m{message}\033[0m", flush=True)
|
|
|
|
|
|
def _note(message: str) -> None:
|
|
print(f" {message}", flush=True)
|
|
|
|
|
|
def _step(message: str) -> None:
|
|
print(f" \033[36m->\033[0m {message}", flush=True)
|
|
|
|
|
|
def check_scratch_dsn(dsn: str) -> None:
|
|
name = dsn.rsplit("/", 1)[-1].split("?", 1)[0]
|
|
if not any(marker in name.lower() for marker in SCRATCH_MARKERS):
|
|
raise SystemExit(
|
|
f"refusing to run against '{name}': this proof of concept truncates tables, and the "
|
|
f"database name does not look like a scratch one. Expected one of: "
|
|
f"{', '.join(SCRATCH_MARKERS)}."
|
|
)
|
|
|
|
|
|
class Admin:
|
|
"""Session-authenticated client for the administrator surface.
|
|
|
|
Uses a cookie jar so the CSRF double-submit works the way it does in the
|
|
browser, rather than being special-cased out for a script.
|
|
"""
|
|
|
|
def __init__(self, base_url: str) -> None:
|
|
import http.cookiejar
|
|
|
|
self._base = base_url.rstrip("/")
|
|
self._cookies: dict[str, str] = {}
|
|
self._jar = http.cookiejar.CookieJar()
|
|
|
|
def _request(
|
|
self, method: str, path: str, payload: Any = None, *, raw_body: bytes | None = None
|
|
) -> Any:
|
|
import urllib.request as request_module
|
|
|
|
url = f"{self._base}{path}"
|
|
body = raw_body if raw_body is not None else (json.dumps(payload).encode() if payload is not None else None)
|
|
req = request_module.Request(url, data=body, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
if self._cookies:
|
|
# One header, joined. ``Request.add_header`` stores headers in a dict
|
|
# keyed by the capitalised name, so adding one per cookie leaves only
|
|
# the last - which is how the session cookie went missing while the
|
|
# login still looked successful.
|
|
req.add_header("Cookie", "; ".join(f"{n}={v}" for n, v in self._cookies.items()))
|
|
csrf = self._cookies.get("umbratrans_csrf")
|
|
if csrf and method in ("POST", "PATCH", "PUT", "DELETE"):
|
|
req.add_header("X-CSRF-Token", csrf)
|
|
opener = request_module.build_opener(
|
|
request_module.HTTPCookieProcessor(self._jar),
|
|
_NoRedirect(),
|
|
)
|
|
try:
|
|
with opener.open(req, timeout=20) as response:
|
|
for header, value in response.getheaders():
|
|
if header.lower() == "set-cookie":
|
|
piece = value.split(";", 1)[0]
|
|
if "=" in piece:
|
|
key, val = piece.split("=", 1)
|
|
self._cookies[key.strip()] = val.strip()
|
|
text = response.read().decode()
|
|
return json.loads(text) if text else None
|
|
except urllib.error.HTTPError as error:
|
|
detail = error.read().decode(errors="replace")
|
|
try:
|
|
parsed = json.loads(detail)
|
|
raise RuntimeError(
|
|
f"{method} {path} -> {error.code} "
|
|
f"{parsed.get('error', {}).get('code', '?')}: {parsed.get('error', {}).get('message', detail[:200])}"
|
|
) from None
|
|
except json.JSONDecodeError:
|
|
raise RuntimeError(f"{method} {path} -> {error.code}: {detail[:300]}") from None
|
|
|
|
def get(self, path: str) -> Any:
|
|
return self._request("GET", path)
|
|
|
|
def post(self, path: str, payload: Any = None) -> Any:
|
|
return self._request("POST", path, payload if payload is not None else {})
|
|
|
|
def patch(self, path: str, payload: Any) -> Any:
|
|
return self._request("PATCH", path, payload)
|
|
|
|
def login(self, email: str, password: str) -> Any:
|
|
return self._request("POST", "/api/auth/login", {"email": email, "password": password})
|
|
|
|
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
"""Follow nothing, so a login redirect shows up as itself.
|
|
|
|
A redirect to the sign-in page is how "your session is not authenticated"
|
|
arrives at a script, and swallowing it produces a confusing 404 later.
|
|
"""
|
|
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# The agent, run as somebody else
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
#: Everything that could let the agent reach the database. Stripped from the
|
|
#: environment it is started with, so the boundary holds because the controller
|
|
#: removed the credential rather than because the agent chose not to use it.
|
|
DATABASE_ENVIRONMENT = (
|
|
"UMBRATRANS_DATABASE_DSN",
|
|
"UMBRATRANS_ENV",
|
|
"UMBRA_ENV",
|
|
"PGPASSWORD",
|
|
"PGPASSFILE",
|
|
"PGHOST",
|
|
"PGPORT",
|
|
"PGUSER",
|
|
"PGDATABASE",
|
|
"DATABASE_URL",
|
|
"UMBRATRANS_SMTP_PASSWORD",
|
|
)
|
|
|
|
|
|
def _agent_environment(extra: dict[str, str] | None = None) -> dict[str, str]:
|
|
"""The controller's environment, minus anything that is a database credential.
|
|
|
|
Handed over explicitly rather than inherited, because inheritance would give
|
|
the agent a connection string and then require it not to use one. §7.2, §13.15
|
|
"""
|
|
|
|
environment = {k: v for k, v in os.environ.items() if k not in DATABASE_ENVIRONMENT}
|
|
environment.update(extra or {})
|
|
return environment
|
|
|
|
|
|
#: Every agent this process has started, whether or not the step that started it
|
|
#: finished. Cleanup reads this rather than a list assembled inside one step,
|
|
#: because the steps that leak are the ones that were interrupted.
|
|
STARTED_AGENTS: list["AgentHandle"] = []
|
|
|
|
#: ``(process, config_path)`` for agents this module spawned outside start_agent.
|
|
_REGISTERED_CONFIGS: list[tuple[subprocess.Popen, Path]] = []
|
|
|
|
|
|
@dataclass
|
|
class AgentHandle:
|
|
name: str
|
|
node_id: str
|
|
key: str
|
|
user: str
|
|
config_path: Path
|
|
process: subprocess.Popen | None = None
|
|
#: How the node key reached this agent. Empty for the ordinary case.
|
|
key_transport: str = ""
|
|
#: Where to ask this agent to stop. Defaults to ``<config>.stop``.
|
|
stop_file: Path | None = None
|
|
|
|
def describe(self) -> str:
|
|
return f"{self.name} (uid {self.user})"
|
|
|
|
|
|
#: Where `poc_node_user.sh setup` installs the agent, and the interpreter it may
|
|
#: use. Both are outside umbra's home directory, which matters: umbra's home is
|
|
#: 0750, so an agent running as another user cannot execute anything inside it -
|
|
#: including the project venv. Launching the agent with umbra's interpreter
|
|
#: fails with "Permission denied" and proves nothing about the boundary.
|
|
INSTALLED_AGENT = Path("/opt/umbra/node-agent/node_agent.py")
|
|
INSTALLED_PYTHON = "/usr/bin/python3"
|
|
|
|
|
|
def resolve_agent_command(run_as: str) -> tuple[str, str, str]:
|
|
"""Return ``(python, script, description)`` for the agent to run under.
|
|
|
|
Prefers the installed copy when running as a different user, because that is
|
|
the arrangement a real node has: a self-contained file on a system
|
|
interpreter, with no access to the controller's home directory. The
|
|
description is returned so the caller can say which one it used - a proof of
|
|
concept that quietly tested the repository copy instead of the installed one
|
|
would not be testing what it claims to.
|
|
"""
|
|
|
|
me = pwd.getpwuid(os.getuid()).pw_name
|
|
repo_script = Path(__file__).resolve().parents[1] / "umbratrans_web" / "node_agent.py"
|
|
if run_as and run_as != me and INSTALLED_AGENT.exists():
|
|
return (
|
|
INSTALLED_PYTHON,
|
|
str(INSTALLED_AGENT),
|
|
f"installed copy at {INSTALLED_AGENT} on {INSTALLED_PYTHON}",
|
|
)
|
|
return sys.executable, str(repo_script), f"repository copy at {repo_script} on {sys.executable}"
|
|
|
|
|
|
def _spawn_prefix(run_as: str) -> tuple[list[str], str | None]:
|
|
"""How to become ``run_as``, and an explanation if we cannot.
|
|
|
|
A process that is not root **cannot** change to another user: ``setresuid``
|
|
requires privilege. So ``setpriv`` is correct only when this controller is
|
|
already root, and the first version of this script used it unconditionally -
|
|
which failed with ``setresuid: Operation not permitted`` for every run as an
|
|
ordinary user, and the agent died before it wrote a line of log.
|
|
|
|
``sudo -n -u`` is the mechanism that works from an unprivileged controller,
|
|
and ``-n`` keeps it from blocking on a password prompt inside a loop.
|
|
|
|
Two sudo options are not interchangeable here, and the difference is where a
|
|
secret ends up:
|
|
|
|
sudo -n -u agent UMBRA_NODE_KEY=... the key is in sudo's argv, and
|
|
argv is world-readable through
|
|
/proc/<pid>/cmdline for as long as
|
|
the process lives
|
|
sudo -n -u agent --preserve-env=... the key is in the environment,
|
|
which /proc/<pid>/environ exposes
|
|
only to that user and to root
|
|
|
|
So the key goes in the environment, and ``env_reset`` - which is usually the
|
|
thing that breaks environment passing - is doing us a favour here by stripping
|
|
everything else: the controller's ``PYTHONPATH``, ``VIRTUAL_ENV`` and
|
|
``LD_LIBRARY_PATH`` all point into umbra's home, which umbra-poc cannot read
|
|
and has no business seeing.
|
|
|
|
``-H`` gives the agent the target user's own home rather than inheriting
|
|
umbra's, which is both more correct and the difference between a writable
|
|
directory and a 0750 one the agent cannot enter.
|
|
"""
|
|
|
|
me = pwd.getpwuid(os.getuid()).pw_name
|
|
if not run_as or run_as == me:
|
|
return [], None
|
|
if os.geteuid() == 0:
|
|
# Already root: no need for sudo, and the environment passes through.
|
|
return ["setpriv", "--reuid", run_as, "--regid", run_as, "--init-groups", "--"], None
|
|
if shutil.which("sudo"):
|
|
return ["sudo", "-n", "-H", "-u", run_as, "--preserve-env=UMBRA_NODE_KEY", "--"], None
|
|
if shutil.which("setpriv"):
|
|
return (
|
|
[],
|
|
f"this process is not root, so it cannot change to {run_as} itself. Run under "
|
|
f"sudo, or install a NOPASSWD rule for 'sudo -n -u {run_as}', or start the "
|
|
"agent yourself and pass --node-key-stdin.",
|
|
)
|
|
return (
|
|
[],
|
|
f"this process is not root and has no sudo, so it cannot become {run_as}. Start "
|
|
"the agent yourself and pass --node-key-stdin.",
|
|
)
|
|
|
|
|
|
def _env_survives_sudo(run_as: str, name: str) -> tuple[bool, str]:
|
|
"""Ask sudo whether ``name`` survives the trip to ``run_as``.
|
|
|
|
Worth asking rather than assuming, because the answer is a property of the
|
|
sudoers policy on the host and not of this code. A sudoers file without the
|
|
SETENV tag refuses ``--preserve-env``, and the symptom otherwise arrives 400
|
|
lines away as the agent saying ``UMBRA_NODE_KEY is not set`` - which reads as
|
|
a bug in the agent and is really a policy on the machine.
|
|
|
|
A throwaway value is used: the point is whether the name arrives at all.
|
|
"""
|
|
|
|
if os.geteuid() == 0 or not run_as or run_as == pwd.getpwuid(os.getuid()).pw_name:
|
|
return True, ""
|
|
prefix, problem = _spawn_prefix(run_as)
|
|
if problem:
|
|
return False, problem
|
|
probe = subprocess.run(
|
|
prefix + ["/bin/sh", "-c", 'test -n "$' + name + '" && printf pass || printf fail'],
|
|
capture_output=True,
|
|
text=True,
|
|
env={**os.environ, name: "probe-value"},
|
|
timeout=20,
|
|
)
|
|
if probe.stdout.strip() == "pass":
|
|
return True, ""
|
|
# Only sudo's own complaint is worth quoting. The probe's stdout carries our
|
|
# own "fail" token, and stitching that into the sentence produced
|
|
# "sudo will not pass UMBRA_NODE_KEY through to umbra-poc: fail", which tells
|
|
# the reader nothing they did not already know.
|
|
detail = (probe.stderr or "").strip().splitlines()
|
|
return False, (
|
|
f"sudo will not pass {name} through to {run_as}."
|
|
+ (f" sudo said: {detail[-1]}" if detail else "")
|
|
+ f"\n Either add 'Defaults:{pwd.getpwuid(os.getuid()).pw_name} env_keep += \"{name}\"'"
|
|
+ " to /etc/sudoers,\n or run the controller as root, or start the agent yourself"
|
|
+ " and pass the key on stdin.\n Nothing secret was sent: the probe used a throwaway value."
|
|
)
|
|
|
|
|
|
def agent_command(
|
|
*,
|
|
config_path: Path,
|
|
key: str,
|
|
run_as: str,
|
|
extra_env: dict[str, str] | None = None,
|
|
extra_args: list[str] | None = None,
|
|
forced_stdin_key: bool = False,
|
|
) -> tuple[list[str], dict[str, str], bool, str]:
|
|
"""Build ``(argv, environment, key_on_stdin, note)`` for the agent.
|
|
|
|
Separated from the two ways of running it - a background ``Popen`` and a
|
|
blocking ``run`` - so that the privilege drop and the key transport cannot
|
|
differ between them. They did once, and the difference showed up only in the
|
|
steps that used the blocking form.
|
|
"""
|
|
|
|
python, agent_script, _ = resolve_agent_command(run_as)
|
|
prefix, problem = _spawn_prefix(run_as)
|
|
if problem:
|
|
raise RuntimeError(problem)
|
|
|
|
note = ""
|
|
on_stdin = forced_stdin_key
|
|
if not on_stdin:
|
|
survives, why = _env_survives_sudo(run_as, "UMBRA_NODE_KEY")
|
|
if survives:
|
|
pass
|
|
elif prefix and shutil.which("sudo"):
|
|
# Recoverable: fall back to the pipe rather than refusing to run.
|
|
on_stdin, note = True, why
|
|
else:
|
|
raise RuntimeError(why)
|
|
|
|
environment = _agent_environment(extra_env)
|
|
if on_stdin:
|
|
environment.pop("UMBRA_NODE_KEY", None)
|
|
else:
|
|
environment["UMBRA_NODE_KEY"] = key
|
|
|
|
command = prefix + [python, agent_script, "--config", str(config_path), "--log-level", "INFO"]
|
|
if extra_args:
|
|
command += extra_args
|
|
if on_stdin:
|
|
command += ["--node-key-stdin"]
|
|
return command, environment, on_stdin, note
|
|
|
|
|
|
def launch_agent_process(
|
|
*,
|
|
config_path: Path,
|
|
key: str,
|
|
run_as: str,
|
|
log_path: Path,
|
|
extra_env: dict[str, str] | None = None,
|
|
extra_args: list[str] | None = None,
|
|
forced_stdin_key: bool = False,
|
|
) -> tuple[subprocess.Popen, str]:
|
|
"""Start the agent as ``run_as`` and return ``(process, how_the_key_arrived)``.
|
|
|
|
One place decides how the node key reaches the agent, because there are four
|
|
call sites and the failure mode of getting it wrong is quiet: the agent starts,
|
|
logs nothing about credentials, and a hundred lines later the proof of concept
|
|
reports that the node never came online.
|
|
|
|
Two transports, neither of which puts the key in a file:
|
|
|
|
``environment`` the key sits in the child's environment. Right when the
|
|
privilege drop preserves the environment, and wrong when it
|
|
does not - sudo resets it by policy, so the variable is
|
|
simply gone by the time the agent looks.
|
|
``stdin`` the key is written to a pipe the agent reads once. Works
|
|
whatever sudo does to the environment, and keeps the key out
|
|
of argv, where every user on the host can read it from
|
|
/proc. Used when the environment cannot be trusted, which is
|
|
detected by asking rather than by finding out.
|
|
"""
|
|
|
|
command, environment, on_stdin, note = agent_command(
|
|
config_path=config_path,
|
|
key=key,
|
|
run_as=run_as,
|
|
extra_env=extra_env,
|
|
extra_args=extra_args,
|
|
forced_stdin_key=forced_stdin_key,
|
|
)
|
|
_ = (command, environment, on_stdin)
|
|
process = subprocess.Popen(
|
|
command,
|
|
stdin=subprocess.PIPE if on_stdin else None,
|
|
stdout=log_path.open("wb"),
|
|
stderr=subprocess.STDOUT,
|
|
env=environment,
|
|
text=True,
|
|
# Its own session, which makes it a process-group leader. Two reasons,
|
|
# both learned the hard way:
|
|
#
|
|
# - sudo forks. The Popen handle is sudo, and sudo forks the agent
|
|
# before exec'ing, so signalling the handle kills the *supervisor* and
|
|
# orphans the worker. The step that is supposed to demonstrate a power
|
|
# cut was killing a wrapper and leaving the agent happily renewing its
|
|
# lease, which is the opposite of a power cut and the reason a killed
|
|
# agent's job appeared not to come back.
|
|
# - the agent no longer shares the controller's group, so a Ctrl-C aimed
|
|
# at the controller does not take the workers with it, or the reverse.
|
|
start_new_session=True,
|
|
)
|
|
if on_stdin:
|
|
# Written and closed at once, so a slow agent never sees a writer still
|
|
# attached to its stdin and cannot mistake the pipe for a terminal.
|
|
try:
|
|
process.stdin.write(key + "\n")
|
|
process.stdin.flush()
|
|
finally:
|
|
process.stdin.close()
|
|
return process, note
|
|
|
|
|
|
def signal_agent_group(process: subprocess.Popen, number: int) -> None:
|
|
"""Signal the whole process group, not just the process we spawned.
|
|
|
|
Falls back to the single process if the group is already gone, because a
|
|
process that exited between the check and the signal raises, and "the agent is
|
|
already dead" is not an error worth raising for.
|
|
"""
|
|
|
|
try:
|
|
os.killpg(os.getpgid(process.pid), number)
|
|
except (ProcessLookupError, PermissionError):
|
|
try:
|
|
process.send_signal(number)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
|
|
def agent_pids_for_config(config_path: Path) -> list[int]:
|
|
"""Every live process running the agent against ``config_path``, by pid.
|
|
|
|
Found by reading /proc rather than by holding on to a pid, which is the whole
|
|
point. The process this driver spawned is the ``sudo`` wrapper; the worker is
|
|
the wrapper's child, owned by a different account, and the two are not
|
|
interchangeable. Checking the wrapper told us nothing about the worker, and
|
|
"the wrapper died" was once reported as "the agent died" - which is how a
|
|
proof of concept claimed to demonstrate a power cut while leaving the agent
|
|
running, renewing its lease, and holding a job for ever.
|
|
|
|
Each config file is unique to one agent in one run's working directory, so
|
|
matching on it identifies the worker without trusting the driver, the wrapper,
|
|
or anything else this process happens to have spawned.
|
|
"""
|
|
|
|
needle = str(config_path)
|
|
alive: set[int] = set()
|
|
for entry in Path("/proc").iterdir():
|
|
if not entry.name.isdigit():
|
|
continue
|
|
try:
|
|
cmdline = (entry / "cmdline").read_bytes().split(b"\0")
|
|
except OSError:
|
|
continue # the process exited between listing and reading
|
|
if any(arg.decode(errors="replace") == needle for arg in cmdline if arg):
|
|
alive.add(int(entry.name))
|
|
return sorted(alive)
|
|
|
|
|
|
def watch_agent_pids(config_path: Path, *, window: float = 1.5) -> list[int]:
|
|
"""Every pid that ever ran the agent against ``config_path`` while we watched.
|
|
|
|
Sampling repeatedly and keeping the union, rather than stopping at the first
|
|
hit. Two processes match - the ``sudo`` wrapper this driver spawned and the
|
|
worker underneath it - and they are not simultaneous: the wrapper appears
|
|
first, and the worker may be gone within a hundred milliseconds of appearing,
|
|
because the fault under test exits as soon as it claims its job. A single
|
|
early sample therefore finds the wrapper and never the worker, which is how a
|
|
check meant to prove the agent vanished came to prove only that its launcher
|
|
did.
|
|
|
|
``/proc/<pid>/cmdline`` is world-readable, so the worker of another account is
|
|
visible here even though ``/proc/<pid>/environ`` is not. That asymmetry is
|
|
also why the node key travels in the environment rather than on a command
|
|
line: the environment is private, argv is not.
|
|
"""
|
|
|
|
seen: set[int] = set()
|
|
deadline = time.time() + window
|
|
while time.time() < deadline:
|
|
seen.update(agent_pids_for_config(config_path))
|
|
time.sleep(0.02)
|
|
return sorted(seen)
|
|
|
|
|
|
def check_agent_survived(
|
|
process: subprocess.Popen, log_path: Path, *, who: str
|
|
) -> None:
|
|
"""Fail loudly and usefully if an agent died at startup.
|
|
|
|
The alternative is an empty log and a "node did not come online" much further
|
|
on, which points at the controller rather than at the thing that failed - and
|
|
which, in this project's history, has been the shape of three separate bugs.
|
|
"""
|
|
|
|
time.sleep(0.4)
|
|
if process.poll() is None:
|
|
return
|
|
try:
|
|
detail = log_path.read_text(errors="replace").strip().splitlines()[-3:]
|
|
except OSError:
|
|
detail = []
|
|
raise RuntimeError(
|
|
f"the agent for {who} exited immediately (status {process.returncode}). "
|
|
+ (f"Its log says: {' | '.join(detail)}" if detail else "and wrote nothing.")
|
|
)
|
|
|
|
|
|
def start_agent(
|
|
*,
|
|
agent_name: str,
|
|
node_id: str,
|
|
key: str,
|
|
controller_url: str,
|
|
run_as: str,
|
|
workdir: Path,
|
|
kinds: list[str] | None = None,
|
|
log_path: Path,
|
|
extra_env: dict[str, str] | None = None,
|
|
key_on_stdin: bool = False,
|
|
) -> AgentHandle:
|
|
"""Write a config the agent user can read, and start the agent as them.
|
|
|
|
The key goes in the *environment*, not the file. The agent supports both, and
|
|
putting it in the environment is the half that proves the point: a config file
|
|
that contains a secret is a secret that eventually ends up in a backup, a
|
|
``tar``, or a support bundle.
|
|
"""
|
|
|
|
config = {
|
|
"controller_url": controller_url,
|
|
# Deliberately empty: the key is supplied by the environment, or on stdin
|
|
# when the environment cannot be trusted to survive the privilege drop.
|
|
# A config file holding a secret is a secret that eventually lands in a
|
|
# backup, a tar, or a support bundle.
|
|
"node_key": "",
|
|
"node_name": agent_name,
|
|
"kinds": kinds or [],
|
|
# Named here rather than left to stop_agent's convention, because the
|
|
# convention was the bug: stop_agent derived `<config>.stop` and wrote
|
|
# it, and the agent never looked, because the config never said where to
|
|
# look. Both halves agreed with each other and neither with the other.
|
|
"stop_file": str(workdir / f"node-{agent_name}.stop"),
|
|
"lease_seconds": 30,
|
|
"poll_interval_seconds": 0.4,
|
|
}
|
|
config_path = workdir / f"node-{agent_name}.json"
|
|
config_path.write_text(json.dumps(config, indent=2))
|
|
config_path.chmod(0o644)
|
|
|
|
process, note = launch_agent_process(
|
|
config_path=config_path,
|
|
key=key,
|
|
run_as=run_as,
|
|
log_path=log_path,
|
|
extra_env=extra_env,
|
|
forced_stdin_key=key_on_stdin,
|
|
)
|
|
handle = AgentHandle(
|
|
name=agent_name,
|
|
node_id=node_id,
|
|
key=key,
|
|
user=run_as or pwd.getpwuid(os.getuid()).pw_name,
|
|
config_path=config_path,
|
|
process=process,
|
|
key_transport=note or "environment",
|
|
)
|
|
STARTED_AGENTS.append(handle)
|
|
check_agent_survived(process, log_path, who=agent_name)
|
|
return handle
|
|
|
|
|
|
def stop_everything_started() -> int:
|
|
"""Stop every agent this process started, and say how many refused.
|
|
|
|
Called from a ``finally``, because the leak is worst exactly when a step
|
|
failed: three runs left agents behind, all from a run that reported success.
|
|
An agent that keeps polling keeps renewing leases for jobs nobody is watching,
|
|
and a later run then finds its own work already claimed - which reads as a
|
|
scheduler fault.
|
|
|
|
The pid-based check is the same one the power-cut step uses, because it is the
|
|
only one that can see a process this process does not own.
|
|
"""
|
|
|
|
count = 0
|
|
for handle in STARTED_AGENTS:
|
|
if stop_agent(handle):
|
|
count += 1
|
|
for process, config_path in _REGISTERED_CONFIGS:
|
|
if process.poll() is not None:
|
|
continue
|
|
survivors = agent_pids_for_config(config_path)
|
|
if survivors:
|
|
count += 1
|
|
print(
|
|
f"\033[31mWARNING\033[0m agent {survivors} for {config_path.name} was "
|
|
"still running at the end of the proof of concept"
|
|
)
|
|
return count
|
|
|
|
|
|
def stop_agent(agent: AgentHandle, *, signal_number: int = signal.SIGTERM) -> list[int]:
|
|
"""Stop an agent, and return the pids that refused to stop.
|
|
|
|
A file the agent watches, not a signal. Signalling is not available: this
|
|
process does not own the agent, and kill(2) against another account's process
|
|
fails. Signalling the ``sudo`` wrapper that *is* ours succeeds, kills the
|
|
wrapper, and orphans the worker - so the driver would report success and leave
|
|
an agent renewing leases. That happened on every run until the pids were
|
|
counted.
|
|
"""
|
|
|
|
if agent.process is None:
|
|
return []
|
|
stop_file = agent.config_path.with_suffix(".stop")
|
|
if agent.stop_file:
|
|
stop_file = agent.stop_file
|
|
try:
|
|
stop_file.write_text("stop\n")
|
|
except OSError:
|
|
pass
|
|
|
|
deadline = time.time() + 6
|
|
while time.time() < deadline:
|
|
if not agent_pids_for_config(agent.config_path):
|
|
return []
|
|
time.sleep(0.2)
|
|
|
|
# Out of patience. Kill what we can - the wrapper - and then check again,
|
|
# reporting anything left rather than assuming it went.
|
|
signal_agent_group(agent.process, signal_number)
|
|
try:
|
|
agent.process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
signal_agent_group(agent.process, signal.SIGKILL)
|
|
agent.process.wait(timeout=5)
|
|
leftovers = agent_pids_for_config(agent.config_path)
|
|
if leftovers:
|
|
print(
|
|
f"\033[31mWARNING\033[0m {len(leftovers)} agent process(es) for {agent.name} "
|
|
f"survived a stop: {leftovers}. They will keep renewing their leases."
|
|
)
|
|
return leftovers
|
|
|
|
|
|
def run_agent_once(
|
|
agent: AgentHandle, *, iterations: int = 1, log_path: Path, timeout: float = 20.0
|
|
) -> str:
|
|
"""Run one bounded iteration of the agent synchronously and return its log.
|
|
|
|
For the steps where the timing matters more than the concurrency. ``--max-
|
|
iterations`` rather than a sleep, so the proof of concept is deterministic
|
|
rather than racing a timer - an agent that can only be tested by waiting is an
|
|
agent whose lease behaviour is untested.
|
|
"""
|
|
|
|
command, environment, on_stdin, _ = agent_command(
|
|
config_path=agent.config_path,
|
|
key=agent.key,
|
|
run_as=agent.user,
|
|
extra_args=["--max-iterations", str(iterations)],
|
|
)
|
|
if on_stdin:
|
|
# A pipe is the transport here too. Feeding it from a string rather than
|
|
# PIPE is simpler, and the amount is one line, so it cannot block.
|
|
completed = subprocess.run(
|
|
command,
|
|
input=agent.key + "\n",
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
env=environment,
|
|
)
|
|
else:
|
|
completed = subprocess.run(
|
|
command, capture_output=True, text=True, timeout=timeout, env=environment
|
|
)
|
|
log_path.write_text(completed.stdout + completed.stderr)
|
|
return completed.stdout + completed.stderr
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# The demonstration
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def demonstrate(
|
|
*,
|
|
admin: Admin,
|
|
organization_id: str,
|
|
workdir: Path,
|
|
agent_user: str,
|
|
kinds: list[str] | None = None,
|
|
) -> int:
|
|
from umbratrans_web.node_agent import measure
|
|
|
|
local_name = pwd.getpwuid(os.getuid()).pw_name
|
|
failures = 0
|
|
|
|
# -- 1. the inventory starts with nodes that are not usable ---------------
|
|
_head("1. A node added by an administrator is not yet schedulable")
|
|
# A per-run suffix, because node names are unique per organization and there
|
|
# is no route to delete one. Without it the second run against the same
|
|
# database dies on a UniqueViolation at step 1 and never reaches anything -
|
|
# and the failure reads as a server fault rather than as leftover state,
|
|
# because it arrives as a bare 500.
|
|
suffix = uuid.uuid4().hex[:6]
|
|
node = admin.post(
|
|
f"/api/compute/nodes?organization_id={organization_id}",
|
|
{"name": f"poc-controller-node-{suffix}", "is_controller": True,
|
|
"trust_domain": "local"},
|
|
)
|
|
agent_node = admin.post(
|
|
f"/api/compute/nodes?organization_id={organization_id}",
|
|
{"name": f"poc-agent-node-{suffix}", "trust_domain": "local"},
|
|
)
|
|
_note(f"node names carry the run suffix -{suffix}; there is no delete route, so a "
|
|
"second run against the same database needs a different name")
|
|
nodes = {item["id"]: item for item in admin.get(f"/api/compute/nodes?organization_id={organization_id}")}
|
|
for item in nodes.values():
|
|
_note(f"{item['name']:24} status={item['status']:14} schedulable={item['schedulable']}")
|
|
if not nodes[node["id"]]["schedulable"]:
|
|
_pass("a freshly created node is provisioning and takes no work (I24)")
|
|
else:
|
|
failures += 1
|
|
_fail("a freshly created node was schedulable before it ever authenticated")
|
|
|
|
# -- 2. credentials, and the scopes they can hold -------------------------
|
|
_head("2. A node credential is narrower than a session")
|
|
credential = admin.post(
|
|
f"/api/compute/nodes/{agent_node['id']}/credentials",
|
|
{"agent_name": "poc-agent", "scopes": ["compute:read", "compute:heartbeat", "compute:claim", "compute:complete"]},
|
|
)
|
|
_note(f"key prefix {credential['key_prefix']}… scopes {', '.join(credential['scopes'])}")
|
|
_note(f"available {', '.join(credential['scopes_available'])}")
|
|
# A scope that could read a manuscript would make this a session token.
|
|
try:
|
|
admin.post(
|
|
f"/api/compute/nodes/{agent_node['id']}/credentials",
|
|
{"agent_name": "sneaky", "scopes": ["projects:read"]},
|
|
)
|
|
failures += 1
|
|
_fail("a project-reading scope was accepted for a node credential")
|
|
except RuntimeError as error:
|
|
if "not node scopes" in str(error):
|
|
_pass(f"a project-reading scope is refused: {str(error).split(': ')[-1][:70]}")
|
|
else:
|
|
failures += 1
|
|
_fail(f"unexpected error refusing a project scope: {error}")
|
|
|
|
# And the agent surface does not include the administrator routes at all.
|
|
import io
|
|
from contextlib import redirect_stderr
|
|
|
|
from umbratrans_web.app import create_app
|
|
from umbratrans_web.config import Environment, Settings
|
|
from umbratrans_web.repository import InMemoryAccountRepository
|
|
|
|
with redirect_stderr(io.StringIO()):
|
|
document = create_app(
|
|
settings=Settings(environment=Environment.TEST),
|
|
repository=InMemoryAccountRepository(),
|
|
).openapi()
|
|
node_paths = {p for p in document["paths"] if p.startswith("/api/node")}
|
|
leaked = {p for p in node_paths if p in {"/api/node/nodes", "/api/node/models", "/api/node/leases/sweep"}}
|
|
if leaked:
|
|
failures += 1
|
|
_fail(f"the node surface exposes administrator routes: {leaked}")
|
|
else:
|
|
_pass(f"the node surface is {len(node_paths)} routes, none of them administrative (I28)")
|
|
|
|
# -- 3. the agent authenticates and the node comes online ------------------
|
|
_head("3. The agent heartbeats, and only then is the node online")
|
|
_, _, which = resolve_agent_command(agent_user)
|
|
_note(f"running the {which}")
|
|
if which.startswith("repository") and agent_user != pwd.getpwuid(os.getuid()).pw_name:
|
|
_note("WARNING: the installed copy is missing, so this is testing the repository")
|
|
_note("copy under another user - which that user cannot execute. Run: sudo")
|
|
_note("poc/poc_node_user.sh setup")
|
|
# Report the key's route up front rather than letting a missing key surface
|
|
# later as "the node never came online".
|
|
_command, _env, on_stdin, why = agent_command(
|
|
config_path=workdir / "node-poc-agent.json",
|
|
key="probe",
|
|
run_as=agent_user,
|
|
)
|
|
if on_stdin:
|
|
_note("node key transport: a pipe, because sudo will not carry the environment")
|
|
for line in why.splitlines():
|
|
_note(f" {line}")
|
|
else:
|
|
_note("node key transport: the agent's environment, never a file and never argv")
|
|
# Measured as the *controller*, which is not the node and no longer can be:
|
|
# the controller is deliberately kept out of the `video` group, so on this host
|
|
# it sees a CPU and no card. That is the boundary working, so the line says who
|
|
# is looking - presenting the controller's view as the node's would report a
|
|
# GPU-less machine on a box that has one.
|
|
devices, measurements = measure()
|
|
_note(f"as {local_name} (the controller, which is not in the video group):")
|
|
for device in devices:
|
|
_note(
|
|
f" {device['kind']:14} {device['label']:8} "
|
|
f"vram {device['vram_free_mb']:>6}/{device['vram_total_mb']:<6} MiB "
|
|
f"ram {device['ram_free_mb']:>6} MiB factor {device['throughput_factor']}"
|
|
)
|
|
if any(d["kind"] == "gpu_discrete" and d["vram_free_mb"] < d["vram_total_mb"] for d in devices):
|
|
_note(" note: the GPU is not empty. A loaded model holds its memory whether or not it is working.")
|
|
else:
|
|
_note(" note: no GPU here, which is the controller's view. The agent's view is below,")
|
|
_note(" and it is the one the scheduler uses.")
|
|
|
|
agent = start_agent(
|
|
agent_name="poc-agent",
|
|
node_id=agent_node["id"],
|
|
key=credential["key"],
|
|
controller_url=admin._base, # noqa: SLF001 - same process, same client
|
|
run_as=agent_user,
|
|
workdir=workdir,
|
|
kinds=kinds,
|
|
log_path=workdir / "agent.log",
|
|
)
|
|
time.sleep(1.5)
|
|
refreshed = {item["id"]: item for item in admin.get(f"/api/compute/nodes?organization_id={organization_id}")}
|
|
after = refreshed[agent_node["id"]]
|
|
_note(f"{after['name']:24} status={after['status']:14} schedulable={after['schedulable']}")
|
|
if after["status"] == "online" and after["schedulable"]:
|
|
_pass(f"the node is online because the agent spoke: {after['status_reason']}")
|
|
else:
|
|
failures += 1
|
|
_fail(f"the node did not come online: {after['status_reason']}")
|
|
|
|
# The agent runs as somebody else. Show that it has no database access.
|
|
_head("4. The agent runs as a different account, with no database credential")
|
|
_note(f"controller runs as : {local_name} (uid {os.getuid()})")
|
|
_note(f"agent runs as : {agent.user} (uid {pwd.getpwnam(agent.user).pw_uid})")
|
|
if agent.user != local_name:
|
|
_pass("the agent is a different unix user")
|
|
else:
|
|
_note("WARNING: the agent is running as the same user, so the boundary is asserted, not shown")
|
|
# The controller holds a DSN; the agent must not. Checked rather than assumed.
|
|
controller_has_dsn = bool(os.environ.get("UMBRATRANS_DATABASE_DSN"))
|
|
if controller_has_dsn:
|
|
_note("the controller process has UMBRATRANS_DATABASE_DSN set")
|
|
environ_path = Path(f"/proc/{agent.process.pid}/environ")
|
|
try:
|
|
agent_env = environ_path.read_bytes().split(b"\x00")
|
|
readable = True
|
|
except PermissionError:
|
|
agent_env = []
|
|
readable = False
|
|
except OSError:
|
|
agent_env = []
|
|
readable = False
|
|
if readable:
|
|
names = {item.split(b"=", 1)[0].decode(errors="replace") for item in agent_env if b"=" in item}
|
|
leaked = sorted(n for n in names if n in DATABASE_ENVIRONMENT)
|
|
if leaked:
|
|
failures += 1
|
|
_fail(f"the agent's environment contains database credentials: {leaked}")
|
|
else:
|
|
_pass("the agent's environment contains no database credential")
|
|
else:
|
|
_pass("the agent's environment is not even readable by the controller: a different unix user")
|
|
_note(f"the agent was started with {len(DATABASE_ENVIRONMENT)} database variables explicitly removed")
|
|
|
|
# -- 5. work flows --------------------------------------------------------
|
|
_head("5. Work is dispatched to the agent and comes back")
|
|
sleep_job = admin.post(
|
|
f"/api/compute/jobs?organization_id={organization_id}",
|
|
{"kind": "sleep", "payload": {"seconds": 1.0, "echo": "hello from the queue"},
|
|
"idempotency_key": "poc-sleep-1"},
|
|
)
|
|
_note(f"submitted {sleep_job['kind']} job {sleep_job['id'][:8]} status={sleep_job['status']}")
|
|
run_agent_once(agent, iterations=3, log_path=workdir / "agent-once.log")
|
|
settled = wait_for_job(admin, organization_id, sleep_job["id"], want={"complete", "failed", "blocked"})
|
|
if settled and settled["status"] == "complete":
|
|
_pass(f"the agent completed it: {settled['result']}")
|
|
else:
|
|
failures += 1
|
|
_fail(f"the job ended {settled['status']}: {settled['error']}")
|
|
events = admin.get(f"/api/compute/jobs/{sleep_job['id']}/events")
|
|
_note("event trail: " + " -> ".join(event["event"] for event in events))
|
|
|
|
# -- 6. idempotency -------------------------------------------------------
|
|
_head("6. The same idempotency key produces one job")
|
|
again = admin.post(
|
|
f"/api/compute/jobs?organization_id={organization_id}",
|
|
{"kind": "sleep", "payload": {"seconds": 1.0}, "idempotency_key": "poc-sleep-1"},
|
|
)
|
|
if again["id"] == sleep_job["id"]:
|
|
_pass("re-submitting with the same key returned the same job, not a second one")
|
|
else:
|
|
failures += 1
|
|
_fail("the idempotency key did not deduplicate")
|
|
|
|
# -- 7. two agents, one job each -----------------------------------------
|
|
_head("7. Two agents claiming at once get different jobs (I25)")
|
|
second_credential = admin.post(
|
|
f"/api/compute/nodes/{node['id']}/credentials",
|
|
{"agent_name": "poc-controller-agent", "scopes": ["compute:heartbeat", "compute:claim", "compute:complete"]},
|
|
)
|
|
controller_agent = start_agent(
|
|
agent_name="poc-controller-agent",
|
|
node_id=node["id"],
|
|
key=second_credential["key"],
|
|
controller_url=admin._base, # noqa: SLF001
|
|
run_as=agent_user,
|
|
workdir=workdir,
|
|
kinds=kinds,
|
|
log_path=workdir / "agent2.log",
|
|
)
|
|
# The controller node is also a worker, and it comes online the same way the
|
|
# agent did: by heartbeating. It has to be online before it can be offered
|
|
# anything, and the second agent is what makes it do that.
|
|
for index in range(6):
|
|
admin.post(
|
|
f"/api/compute/jobs?organization_id={organization_id}",
|
|
{"kind": "sleep", "payload": {"seconds": 0.6}, "idempotency_key": f"poc-concurrent-{index}"},
|
|
)
|
|
# Both agents run at once, so the contention is real rather than staged.
|
|
runners = []
|
|
for handle, extra in ((agent, ["--max-iterations", "40"]),
|
|
(controller_agent, ["--max-iterations", "40"])):
|
|
log_file = workdir / f"{handle.name}.log"
|
|
process, _ = launch_agent_process(
|
|
config_path=handle.config_path,
|
|
key=handle.key,
|
|
run_as=agent_user,
|
|
log_path=log_file,
|
|
extra_args=extra,
|
|
)
|
|
check_agent_survived(process, log_file, who=f"{handle.name} (second)")
|
|
runners.append(process)
|
|
_REGISTERED_CONFIGS.append((process, handle.config_path))
|
|
try:
|
|
deadline = time.monotonic() + 60
|
|
concurrent = []
|
|
while time.monotonic() < deadline:
|
|
items = admin.get(f"/api/compute/jobs?organization_id={organization_id}")
|
|
concurrent = [j for j in items if j["idempotency_key"] and j["idempotency_key"].startswith("poc-concurrent")]
|
|
if concurrent and all(j["status"] in ("complete", "failed", "blocked") for j in concurrent):
|
|
break
|
|
time.sleep(0.4)
|
|
finally:
|
|
for runner in runners:
|
|
runner.terminate()
|
|
for runner in runners:
|
|
try:
|
|
runner.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
runner.kill()
|
|
|
|
done = [j for j in concurrent if j["status"] == "complete"]
|
|
_note(f"{len(done)}/{len(concurrent)} job(s) completed by two agents running at once")
|
|
# The claim is the thing under test, and the event trail records it. One
|
|
# 'claimed' per job is the property; two agents both succeeding at something
|
|
# is not evidence of it.
|
|
double_claimed = []
|
|
for job in concurrent:
|
|
claims = [e for e in admin.get(f"/api/compute/jobs/{job['id']}/events") if e["event"] == "claimed"]
|
|
if len(claims) > 1:
|
|
double_claimed.append((job["id"][:8], len(claims)))
|
|
if concurrent and not double_claimed:
|
|
_pass(f"every job was claimed exactly once across {len(concurrent)} jobs and 2 agents (I25)")
|
|
elif not concurrent:
|
|
failures += 1
|
|
_fail("no concurrent jobs were submitted")
|
|
else:
|
|
failures += 1
|
|
_fail(f"jobs were claimed more than once: {double_claimed}")
|
|
|
|
for handle in (agent, controller_agent):
|
|
stop_agent(handle)
|
|
|
|
# -- 8. lease recovery ----------------------------------------------------
|
|
_head("8. A killed agent's job comes back (I26)")
|
|
long_job = admin.post(
|
|
f"/api/compute/jobs?organization_id={organization_id}",
|
|
# Ten seconds, not ten minutes. The victim exits the instant it claims
|
|
# this job and never runs it, so its duration is irrelevant to what is
|
|
# being tested - and a 600-second job left in the queue is picked up by
|
|
# any agent still polling, which then sits in it for ten minutes unable
|
|
# to notice a stop. Four of those accumulated across four runs, and the
|
|
# agents that claimed them were the ones leaking at the end.
|
|
{"kind": "sleep", "payload": {"seconds": 10},
|
|
"idempotency_key": f"poc-long-{suffix}"},
|
|
)
|
|
_note(f"submitted a job {long_job['id'][:8]}; the victim claims it and vanishes "
|
|
"without running it, so its length does not matter")
|
|
|
|
# The agent exits abruptly the moment it claims this job: no cleanup, no lease
|
|
# release, no goodbye. It does that from the inside, because the alternative
|
|
# does not work.
|
|
#
|
|
# Signalling it from here was the previous approach, and it never was a power
|
|
# cut. This driver runs as `umbra`; the agent runs as `umbra-poc`. It cannot
|
|
# signal a process owned by another account. Signalling the *sudo wrapper* it
|
|
# had spawned does work - that wrapper is its own - and killing it leaves the
|
|
# worker running, renewing its lease, holding the job for ever. kill(2) on a
|
|
# process group reports success as soon as one member has been signalled, so
|
|
# the wrapper's death was read as the agent's. Every run leaked an agent that
|
|
# way, and the "power cut" that was printed above it described something that
|
|
# had not happened.
|
|
#
|
|
# What makes this honest rather than merely different: the worker is
|
|
# identified by its own pid, found in /proc by its config path, and the run
|
|
# fails if that pid is still there afterwards.
|
|
# Its own node and its own credential.
|
|
#
|
|
# The victim originally reused the first agent's key, on the theory that it
|
|
# needed to hold that agent's lease. It does not: a node is identified by
|
|
# its credential, not by its config file's `node_name` - the name is only
|
|
# logged. So with the shared key the victim heartbeated *as* poc-agent-node,
|
|
# polled under that node's eligibility, and the long job was never offered to
|
|
# it because nothing was the place it was waiting. It sat there for the whole
|
|
# step, and step 8 concluded from a correct liveness check that the kill had
|
|
# failed.
|
|
#
|
|
# A third node also makes the test say what it means: one agent claims a job
|
|
# and vanishes, and no other agent is in a position to take the job back. With
|
|
# a shared node the requeue is immediately reclaimed and the lease expiry is
|
|
# exercised three times over instead of once.
|
|
victim_node = admin.post(
|
|
f"/api/compute/nodes?organization_id={organization_id}",
|
|
{"name": f"poc-victim-node-{suffix}", "trust_domain": "local"},
|
|
)
|
|
victim_credential = admin.post(
|
|
f"/api/compute/nodes/{victim_node['id']}/credentials",
|
|
{"agent_name": "poc-victim",
|
|
"scopes": ["compute:read", "compute:heartbeat", "compute:claim", "compute:complete"]},
|
|
)
|
|
_note(f"a third node, {victim_node['name']}, with its own credential: a shared key "
|
|
"means the victim is the same node as the agent that already stopped")
|
|
victim_log = workdir / "agent-victim.log"
|
|
victim_config = workdir / f"node-poc-victim-{suffix}.json"
|
|
victim_config.write_text(json.dumps({
|
|
"controller_url": admin._base, # noqa: SLF001
|
|
"node_key": "",
|
|
"node_name": "poc-victim",
|
|
"kinds": kinds or [],
|
|
"stop_file": str(workdir / f"node-poc-victim-{suffix}.stop"),
|
|
"lease_seconds": 30,
|
|
"poll_interval_seconds": 0.4,
|
|
}, indent=2))
|
|
victim_config.chmod(0o644)
|
|
_note("the agent will exit abruptly on claiming it, from the inside")
|
|
victim, _ = launch_agent_process(
|
|
config_path=victim_config,
|
|
key=victim_credential["key"],
|
|
run_as=agent_user,
|
|
log_path=victim_log,
|
|
extra_args=["--fault-exit-after-claim", "1"],
|
|
)
|
|
|
|
# Catch it while it is still there, so "it is gone" means something: a pid
|
|
# that was never observed is not evidence of anything.
|
|
worker_pids = watch_agent_pids(victim_config, window=3.0)
|
|
_note(f"the driver holds pid {victim.pid} (the sudo wrapper); every process running "
|
|
f"the agent was {worker_pids}")
|
|
if len(worker_pids) < 2:
|
|
_note("note: only one process was seen - the wrapper. The worker is short-lived "
|
|
"here, so this check is weaker than it looks.")
|
|
|
|
claimed = wait_for_status(admin, organization_id, long_job["id"], "running", timeout=25)
|
|
if not claimed:
|
|
failures += 1
|
|
_fail("no agent ever claimed the long job, so there is no lease to expire")
|
|
else:
|
|
_pass("the agent holds a lease on a job it will take ten minutes to finish")
|
|
|
|
deadline = time.time() + 20
|
|
while time.time() < deadline and agent_pids_for_config(victim_config):
|
|
time.sleep(0.2)
|
|
gone = agent_pids_for_config(victim_config)
|
|
if not worker_pids:
|
|
failures += 1
|
|
_fail("the agent worker was never observed, so its absence proves nothing")
|
|
elif gone:
|
|
failures += 1
|
|
_fail(f"the agent worker {gone} is still running after it claimed the job, so "
|
|
"nothing below this means anything")
|
|
else:
|
|
_pass("the worker is gone: its own pid is no longer in /proc")
|
|
|
|
held = next(
|
|
j for j in admin.get(f"/api/compute/jobs?organization_id={organization_id}")
|
|
if j["id"] == long_job["id"]
|
|
)
|
|
_note(f"after the agent vanished: status={held['status']} "
|
|
f"(the lease is still held - nothing released it)")
|
|
if held["status"] == "running":
|
|
_pass("the job is still held by an agent that no longer exists")
|
|
else:
|
|
failures += 1
|
|
_fail(f"the job was already {held['status']}, so it was never left holding")
|
|
|
|
# Wait for the *controller's own* sweeper, without asking anyone to sweep.
|
|
# The previous version of this step posted to the administrative sweep route
|
|
# itself, which made recovery look automatic while depending entirely on a
|
|
# human noticing a stalled job - and there was no timer, so that endpoint's
|
|
# docstring described something that did not exist.
|
|
_note("waiting for the controller's own sweeper - nobody is asked to sweep")
|
|
recovered = None
|
|
deadline = time.time() + 45
|
|
while time.time() < deadline:
|
|
current = next(
|
|
j for j in admin.get(f"/api/compute/jobs?organization_id={organization_id}")
|
|
if j["id"] == long_job["id"]
|
|
)
|
|
if current["status"] != "running":
|
|
recovered = current
|
|
break
|
|
time.sleep(1)
|
|
if recovered is None:
|
|
# Recovery did not happen on its own. Ask once, by hand, to separate "the
|
|
# mechanism is broken" from "the timer is missing", which are different
|
|
# problems and were previously reported as one "failed".
|
|
swept = admin.post("/api/compute/leases/sweep", {})
|
|
recovered = next(
|
|
j for j in admin.get(f"/api/compute/jobs?organization_id={organization_id}")
|
|
if j["id"] == long_job["id"]
|
|
)
|
|
_note(f"after a manual sweep: status={recovered['status']} swept={swept}")
|
|
if recovered["status"] != "running":
|
|
failures += 1
|
|
_fail("the timer did not recover it, but a manual sweep did. The lease "
|
|
"mechanism works; the sweeper does not run.")
|
|
else:
|
|
failures += 1
|
|
_fail("the lease did not expire even under a manual sweep, so the mechanism "
|
|
"itself is broken")
|
|
else:
|
|
_note(f"recovered without being asked: status={recovered['status']} "
|
|
f"attempt={recovered['attempt']}")
|
|
_note(f"error: {recovered['error']!r}")
|
|
_note("event trail: " + " -> ".join(
|
|
e["event"] for e in admin.get(f"/api/compute/jobs/{long_job['id']}/events")))
|
|
if recovered["status"] in ("queued", "failed"):
|
|
_pass("a node that vanished had its job returned by the controller alone (I26)")
|
|
else:
|
|
failures += 1
|
|
_fail(f"the job left 'running' as {recovered['status']!r} rather than being returned")
|
|
|
|
# Stop the long-lived agents before the steps that submit more work.
|
|
#
|
|
# Step 8 leaves `poc-long` in the queue - correctly, that is what recovery
|
|
# means - so any agent still polling would claim it. It is short now, but an
|
|
# agent inside a job cannot see a stop file: the check is at the top of the
|
|
# loop, not inside the work. That is how agents survived to the end of a run:
|
|
# not by ignoring the stop, but by being inside a job.
|
|
#
|
|
# It is also worth being plain that this is a real limitation and not only a
|
|
# proof-of-concept one: a worker that has claimed a long job cannot be asked
|
|
# to stop until the job returns. Cancellation mid-job is not implemented, and
|
|
# until it is, the recovery path is the only way out - which is correct for a
|
|
# machine that died and insufficient for one that was told to stop.
|
|
for handle in STARTED_AGENTS:
|
|
stop_agent(handle)
|
|
_note("stopped the remaining agents before submitting more work, so none of them "
|
|
"can claim the 600-second job step 8 deliberately left behind")
|
|
|
|
# -- 9. slow is eligible, impossible is blocked ---------------------------
|
|
_head("9. A model that does not fit in VRAM runs on RAM; one nothing can run is blocked")
|
|
admin.post("/api/compute/models", {"slug": "gemma4:26b", "kind": "llm", "vram_min_mb": 18000,
|
|
"ram_min_mb": 0, "reference_throughput": 1.0, "load_seconds": 300})
|
|
admin.post("/api/compute/models", {"slug": "cuda-only-model", "kind": "llm", "vram_min_mb": 1000,
|
|
"requires_accelerators": ["tpu"]})
|
|
slow_job = admin.post(
|
|
f"/api/compute/jobs?organization_id={organization_id}",
|
|
{"kind": "sleep", "payload": {"seconds": 0.2},
|
|
"requires": {"models": ["gemma4:26b"]}, "idempotency_key": f"poc-slow-{suffix}",
|
|
"fail_if_unplaceable": False},
|
|
)
|
|
_note(f"gemma4:26b needs 18 GB of VRAM; the node reports "
|
|
f"{_gpu_free(admin, agent_node['id'], organization_id):.0f} MiB free on its card", 1)
|
|
_note(f"job status = {slow_job['status']} (queued on the slow path, not blocked)")
|
|
if slow_job["status"] == "queued":
|
|
_pass("a model too big for the VRAM is still eligible: the CPU device can take it, slowly")
|
|
else:
|
|
failures += 1
|
|
_fail(f"expected the slow job to be queued, got {slow_job['status']}: {slow_job['error']}")
|
|
|
|
blocked = admin.post(
|
|
f"/api/compute/jobs?organization_id={organization_id}",
|
|
{"kind": "sleep", "payload": {"seconds": 0.2},
|
|
"requires": {"models": ["cuda-only-model"]}, "idempotency_key": "poc-blocked"},
|
|
)
|
|
_note(f"a model requiring an accelerator no node has -> status = {blocked['status']}")
|
|
_note(f"reason: {blocked['error']}")
|
|
if blocked["status"] == "blocked":
|
|
_pass("a job nothing can run is blocked with a reason, not left looking like it is progressing")
|
|
else:
|
|
failures += 1
|
|
_fail(f"expected blocked, got {blocked['status']}")
|
|
|
|
# -- 10. decomposition ----------------------------------------------------
|
|
_head("10. A job too large for one node is decomposed (I29, I30)")
|
|
parent = admin.post(
|
|
f"/api/compute/jobs?organization_id={organization_id}",
|
|
{"kind": "analyze_project", "payload": {}, "idempotency_key": "poc-parent"},
|
|
)
|
|
_note(f"parent {parent['id'][:8]} kind={parent['kind']} status={parent['status']}")
|
|
_note("it is a container: it has children, is never itself dispatched, and ends complete or partial")
|
|
events = admin.get(f"/api/compute/jobs/{parent['id']}/events")
|
|
_pass(f"parent is tracked as a job, status={parent['status']}")
|
|
|
|
return failures
|
|
|
|
|
|
def wait_for_job(admin, organization_id, job_id, *, want, timeout=30.0, interval=0.3):
|
|
"""Wait for a job to reach one of ``want``.
|
|
|
|
The queue is asynchronous, so reading a job straight after submitting it
|
|
reads whatever it happened to be doing. Polling for a *terminal* state is not
|
|
a workaround for a race in the system; it is the only honest way to observe
|
|
one.
|
|
"""
|
|
|
|
deadline = time.monotonic() + timeout
|
|
job = None
|
|
while time.monotonic() < deadline:
|
|
items = admin.get(f"/api/compute/jobs?organization_id={organization_id}")
|
|
job = next((item for item in items if item["id"] == job_id), None)
|
|
if job is not None and job["status"] in want:
|
|
return job
|
|
time.sleep(interval)
|
|
return job
|
|
|
|
|
|
def wait_for_status(admin, organization_id, job_id, status, *, timeout=30.0, interval=0.3):
|
|
return wait_for_job(admin, organization_id, job_id, want={status}, timeout=timeout, interval=interval)
|
|
|
|
|
|
def _gpu_free(admin, node_id: str, organization_id: str) -> float:
|
|
"""Free VRAM on the *node*, as the node itself reported it.
|
|
|
|
Not measured locally. The controller is deliberately kept out of the `video`
|
|
group, so measuring here answers for the controller and reports a machine
|
|
with no GPU - on a box that has one. The node's own heartbeat is the figure
|
|
the scheduler places work against, so it is also the figure worth quoting.
|
|
"""
|
|
|
|
# From the inventory listing rather than a per-node route: there is no
|
|
# `GET /api/compute/nodes/{id}` - the listing already carries each node's
|
|
# devices - and asking for a route that does not exist gets a 404 whose
|
|
# message is `"detail": "Not Found"`, which names neither the route nor the
|
|
# node.
|
|
for node in admin.get(f"/api/compute/nodes?organization_id={organization_id}"):
|
|
if node["id"] != node_id:
|
|
continue
|
|
for device in node.get("devices", []):
|
|
if device["kind"] == "gpu_discrete":
|
|
return float(device["vram_free_mb"])
|
|
return 0.0
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Entry point
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="UmbraTrans compute proof of concept")
|
|
parser.add_argument("--controller", default="http://127.0.0.1:8443", help="controller base URL")
|
|
parser.add_argument("--dsn", default=os.environ.get("UMBRATRANS_DATABASE_DSN", ""),
|
|
help="scratch DSN, for the safety check only")
|
|
parser.add_argument("--email", required=True, help="administrator email")
|
|
parser.add_argument("--password", required=True, help="administrator password")
|
|
parser.add_argument("--agent-user", default="umbra-poc", help="unix account to run the agent as")
|
|
parser.add_argument("--organization-id", default=None,
|
|
help="organization to act within, for a platform administrator with none of their own")
|
|
parser.add_argument("--organization-name", default=None,
|
|
help="look the organization up by name instead of guessing")
|
|
parser.add_argument("--workdir", type=Path, default=None)
|
|
args = parser.parse_args(argv)
|
|
|
|
if args.dsn:
|
|
check_scratch_dsn(args.dsn)
|
|
try:
|
|
pwd.getpwnam(args.agent_user)
|
|
except KeyError:
|
|
print(f"no such unix account: {args.agent_user}", file=sys.stderr)
|
|
return 2
|
|
|
|
workdir = args.workdir or Path(tempfile.mkdtemp(prefix="umbra-poc-"))
|
|
workdir.mkdir(parents=True, exist_ok=True)
|
|
# 0755, not the 0700 mkdtemp gives and not the 0750 that seemed tidier.
|
|
# The agent runs as a different user and has to traverse this directory to
|
|
# read its own config, and a different user is not the owner and not in the
|
|
# group, so anything less than o+x denies it outright. 0750 produced
|
|
# "Permission denied" on a file whose own mode was already 0644.
|
|
#
|
|
# What that costs: anyone on the box can read the agent's logs here. They hold
|
|
# job identifiers, node names and timings - fleet information, not secrets -
|
|
# and the node key is in the environment rather than in these files, which is
|
|
# the reason this is acceptable. For a real deployment put the workdir in the
|
|
# agent's own state directory and let the agent own it; here the controller
|
|
# has to write into it too, and two owners would mean a group that undoes the
|
|
# separation this is demonstrating.
|
|
workdir.chmod(0o755)
|
|
print("\033[1mUmbraTrans compute proof of concept\033[0m")
|
|
_note(f"controller : {args.controller}")
|
|
_note(f"workdir : {workdir}")
|
|
_note(f"agent user : {args.agent_user}")
|
|
|
|
admin = Admin(args.controller)
|
|
_head("0. Signing in as the administrator")
|
|
try:
|
|
session = admin.login(args.email, args.password)
|
|
except RuntimeError as error:
|
|
return _fail(f"could not sign in: {error}")
|
|
if session.get("must_change_password"):
|
|
return _fail("the administrator still has a temporary password; finish setup first")
|
|
# "role" and "organization_id" live under "user", not at the top level.
|
|
profile = session.get("user") or {}
|
|
organization_id = profile.get("organization_id")
|
|
if organization_id is None:
|
|
# A platform administrator with no organization acts within a named one.
|
|
organizations = admin.get("/api/uadmin/organizations")
|
|
if not organizations.get("items"):
|
|
return _fail("no organization exists; create one first")
|
|
organization_id = organizations["items"][0]["id"]
|
|
_note(f"platform administrator; acting within organization {organization_id[:8]}")
|
|
else:
|
|
_note(f"organization {organization_id[:8]}")
|
|
_pass(f"signed in as {profile.get('email')} ({profile.get('role')})")
|
|
|
|
try:
|
|
failures = demonstrate(
|
|
admin=admin,
|
|
organization_id=organization_id,
|
|
workdir=workdir,
|
|
agent_user=args.agent_user,
|
|
)
|
|
finally:
|
|
for path in sorted(workdir.glob("*.log")):
|
|
_note(f"log: {path}")
|
|
leaked = stop_everything_started()
|
|
if leaked:
|
|
_note(f"{leaked} agent process(es) refused to stop; see the warning above")
|
|
else:
|
|
_note("every agent this run started was stopped")
|
|
|
|
print()
|
|
if failures:
|
|
print(f"\033[31m{failures} check(s) failed\033[0m")
|
|
return 1
|
|
print("\033[32mAll checks passed.\033[0m")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
sys.exit(main())
|