josh/electric-cherry
1
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases Wiki Activity

Cache Plutonium runtime files on hosts

Browse Source
This commit is contained in:
josh
2026-10-07 15:17:59 -04:00
parent 7fd59d32b4
commit 062ade8655
13 changed files with 1418 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/ec_agent.py
+262 -4
View File
@@ -33,7 +33,7 @@ import urllib.parse
import urllib.request
import uuid
VERSION = "1.7.0"
VERSION = "1.8.0"
LABEL = "io.electric-cherry"
MAX_FILE = 32 * 1024 * 1024
MANIFEST_MAX_ENTRIES = 10000
@@ -42,6 +42,10 @@ MANIFEST_MAX_SECONDS = 15
MANIFEST_MAX_DEPTH = 64
LIBRARY_DOWNLOAD_WORKERS = 4
HEALTH_LOCK_TIMEOUT_SECONDS = 5
PLUTONIUM_MANIFEST_URL = "https://cdn.plutoniummod.com/updater/prod/info.json"
PLUTONIUM_MANIFEST_LIMIT = 2 * 1024 * 1024
PLUTONIUM_FILE_LIMIT = 512 * 1024 * 1024
PLUTONIUM_CACHE_LIMIT = 4 * 1024 * 1024 * 1024
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
AGENT_CONFIG = Path("/etc/electric-cherry/agent.json")
AGENT_EXECUTABLE = Path("/usr/local/libexec/electric-cherry-agent.py")
@@ -487,6 +491,34 @@ def managed_write(root, value, content, uid=1000, gid=1000):
os.unlink(temp_name, dir_fd=parent)
def managed_copy(root, value, source, uid=1000, gid=1000):
"""Copy a private file, using Linux reflinks when supported, never hardlinks."""
with managed_open(Path(source).parent, Path(source).name) as incoming:
with managed_parent(root, value, create=True, uid=uid, gid=gid) as (parent, name):
temp_name = ".ec-" + uuid.uuid4().hex
fd = os.open(temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent)
try:
os.fchmod(fd, 0o644)
if os.geteuid() == 0:
os.fchown(fd, uid, gid)
try:
if not sys.platform.startswith("linux"):
raise OSError("Reflinks unavailable")
fcntl.ioctl(fd, 0x40049409, incoming.fileno()) # FICLONE
except OSError:
os.ftruncate(fd, 0)
with os.fdopen(fd, "wb", closefd=False) as outgoing:
shutil.copyfileobj(incoming, outgoing, 1024 * 1024)
outgoing.flush()
os.fsync(fd)
os.replace(temp_name, name, src_dir_fd=parent, dst_dir_fd=parent)
os.fsync(parent)
finally:
os.close(fd)
with contextlib.suppress(FileNotFoundError):
os.unlink(temp_name, dir_fd=parent)
class ReadWriteGate:
"""Let observations run together; global mutations wait for existing readers."""
def __init__(self):
@@ -921,11 +953,13 @@ class Agent:
self.read_claims = set()
self.health_probe_gate = threading.BoundedSemaphore(2)
self.cancelled_creations = frozenset()
self.defer_plutonium_preparation = False
self.observation_keep_days = 0 # Older controllers do not authorize cleanup.
self.library_progress = {}
self.library_transfers = {}
self.inventory_manifest_cache = {}
self.inventory_manifest_lock = threading.Lock()
self.plutonium_lock = threading.RLock()
self.config_path = config.get("configPath", "/etc/electric-cherry/agent.json")
self.executing_id = None
self.lane_lock = threading.Lock()
@@ -1360,8 +1394,9 @@ class Agent:
if line.startswith("PRETTY_NAME="):
os_name = line.split("=", 1)[1].strip('"')
update = self.update_inventory()
capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else [])
plutonium_cache = self.plutonium_cache_inventory()
return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})}
def safe_to_disrupt(self, instance_id, generation):
health = self.inspect(instance_id, generation)
@@ -1510,6 +1545,223 @@ class Agent:
raise Rejected("Initial files exceed the 64 MiB operation limit")
return game, library, image, port, bind, env
@staticmethod
def plutonium_manifest(data):
"""Validate the official updater format before using any paths or URLs."""
if len(data) > PLUTONIUM_MANIFEST_LIMIT:
raise AgentError("Plutonium manifest is too large")
try:
manifest = json.loads(data)
except (ValueError, UnicodeError):
raise AgentError("Invalid Plutonium manifest") from None
if not isinstance(manifest, dict) or manifest.get("product") != "plutonium-core-prod":
raise AgentError("Invalid Plutonium production manifest")
revision = manifest.get("revision")
if type(revision) is not int or not 1 <= revision <= 2147483647:
raise AgentError("Invalid Plutonium revision")
base = manifest.get("baseUrl")
if not isinstance(base, str) or not re.fullmatch(r"https://cdn\.plutoniummod\.com/updater/prod/(?:[A-Za-z0-9_-]+/)*", base):
raise AgentError("Plutonium downloads must use the official production CDN")
entries = manifest.get("files")
if not isinstance(entries, list) or not 1 <= len(entries) <= 4096:
raise AgentError("Invalid Plutonium file list")
names, total = set(), 0
for entry in entries:
if not isinstance(entry, dict):
raise AgentError("Invalid Plutonium file entry")
name, size, checksum = entry.get("name"), entry.get("size"), entry.get("hash")
if not isinstance(name, str) or len(name) > 512 or re.search(r'[\x00-\x1f\x7f\\:*?"<>|]', name):
raise AgentError("Invalid Plutonium file path")
path = PurePosixPath(name)
if not path.parts or path.is_absolute() or str(path) != name or any(part in (".", "..") or part.endswith((".", " ")) for part in path.parts) or name.lower() == "cdn_info.json":
raise AgentError("Invalid Plutonium file path")
if name.lower() in names:
raise AgentError("Duplicate Plutonium file path")
names.add(name.lower())
if type(size) is not int or not 0 <= size <= PLUTONIUM_FILE_LIMIT:
raise AgentError("Invalid Plutonium file size")
if not isinstance(checksum, str) or not re.fullmatch(r"[a-f0-9]{40}", checksum):
raise AgentError("Invalid Plutonium file hash")
total += size
if total > PLUTONIUM_CACHE_LIMIT:
raise AgentError("Plutonium cache exceeds the size limit")
for name in names:
if any(str(parent) in names for parent in PurePosixPath(name).parents if str(parent) != "."):
raise AgentError("Conflicting Plutonium file paths")
if "bin/plutonium-bootstrapper-win32.exe" not in {entry["name"] for entry in entries}:
raise AgentError("Plutonium manifest is missing the server bootstrapper")
# Match plutonium-updater's default: launcher assets are optional.
return manifest, [entry for entry in entries if not entry["name"].startswith("launcher")]
@staticmethod
def plutonium_file_matches(root, entry):
try:
with managed_open(root, entry["name"]) as stream:
if os.fstat(stream.fileno()).st_size != entry["size"]:
return False
checksum = hashlib.sha1()
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
checksum.update(chunk)
return checksum.hexdigest() == entry["hash"]
except (OSError, AgentError):
return False
def plutonium_open(self, url):
if not isinstance(url, str) or not re.fullmatch(r"https://cdn\.plutoniummod\.com/updater/prod/[A-Za-z0-9_./-]+", url):
raise AgentError("Plutonium downloads must use the official production CDN")
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, hdrs, newurl):
raise AgentError("Plutonium download redirects are not allowed")
# Never send the controller's host credentials to the public CDN.
return urllib.request.build_opener(NoRedirect()).open(url, timeout=30)
def plutonium_snapshot(self):
root = scoped(self.root / "agent", "plutonium-cache")
pointer = json.loads(managed_read(root, "current.json") or b"null")
if not isinstance(pointer, dict) or not isinstance(pointer.get("key"), str) or not re.fullmatch(r"[a-f0-9]{64}", pointer["key"]):
raise AgentError("Plutonium cache is not prepared")
content = scoped(root, pointer["key"] + "/content")
raw = managed_read(content, "cdn_info.json")
if raw is None or digest(raw) != pointer["key"]:
raise AgentError("Plutonium cache manifest is invalid")
manifest, files = self.plutonium_manifest(raw)
if pointer.get("revision") != manifest["revision"] or pointer.get("bytes") != sum(entry["size"] for entry in files):
raise AgentError("Plutonium cache metadata is invalid")
# Heartbeat checks are cheap. Every file is hashed before it is seeded.
for entry in files:
info = scoped(content, entry["name"]).lstat()
if not stat_module.S_ISREG(info.st_mode) or info.st_size != entry["size"]:
raise AgentError("Plutonium cache files are incomplete")
return pointer, content, raw, files
def plutonium_cache_inventory(self):
with self.plutonium_lock:
try:
pointer, _, _, _ = self.plutonium_snapshot()
return {"revision": pointer["revision"], "bytes": pointer["bytes"]}
except (OSError, ValueError, AgentError):
return None
def prepare_plutonium(self, payload):
target, reserve = payload.get("revision", 0), payload.get("minimumFreeBytes", 0)
if type(target) is not int or not 0 <= target <= 2147483647 or type(reserve) is not int or reserve < 0:
raise Rejected("Invalid Plutonium preparation request")
root = scoped(self.root / "agent", "plutonium-cache")
root.mkdir(mode=0o700, exist_ok=True)
os.chmod(root, 0o700)
deadline = time.monotonic() + 1800
def active():
if self.stopping.is_set() or time.monotonic() >= deadline:
raise AgentError("Plutonium cache preparation was interrupted or timed out")
if self.defer_plutonium_preparation:
raise Held("Plutonium cache preparation is waiting for other host operations")
active()
try:
with self.plutonium_open(PLUTONIUM_MANIFEST_URL) as response:
raw = response.read(PLUTONIUM_MANIFEST_LIMIT + 1)
manifest, files = self.plutonium_manifest(raw)
if manifest["revision"] < target:
raise AgentError("Plutonium CDN has not reached the requested revision")
key = digest(raw)
previous = None
with self.plutonium_lock:
with contextlib.suppress(OSError, ValueError, AgentError):
_, previous, _, _ = self.plutonium_snapshot()
if previous is not None and previous.parent.name == key and all(self.plutonium_file_matches(previous, entry) for entry in files):
return {"revision": manifest["revision"], "bytes": sum(entry["size"] for entry in files)}
staging = scoped(root, key + ".partial/content")
staging.mkdir(parents=True, exist_ok=True)
expected = {entry["name"] for entry in files}
# SIGKILL/power loss bypasses download/copy finally blocks. Remove
# abandoned temporary files before counting space or publishing.
for directory, subdirs, filenames in os.walk(staging, followlinks=False):
for name in subdirs + filenames:
path = Path(directory) / name
if path.is_symlink() or (name in filenames and path.relative_to(staging).as_posix() not in expected):
path.unlink()
# Keep at most the current snapshot and this resumable preparation.
current_key = previous.parent.name if previous is not None else None
for directory in root.iterdir():
if re.fullmatch(r"[a-f0-9]{64}(?:\.partial)?", directory.name) and directory.name not in (current_key, key + ".partial"):
if directory.is_symlink():
directory.unlink()
else:
shutil.rmtree(directory)
missing = sum(entry["size"] for entry in files if not self.plutonium_file_matches(staging, entry))
if shutil.disk_usage(root).free < reserve + missing + len(raw):
raise AgentError("Insufficient free storage for the Plutonium cache and host reserve")
for entry in files:
active()
if self.plutonium_file_matches(staging, entry):
continue
if previous is not None and self.plutonium_file_matches(previous, entry):
managed_copy(staging, entry["name"], scoped(previous, entry["name"]), os.geteuid(), os.getegid())
continue
destination = scoped(staging, entry["name"])
destination.parent.mkdir(parents=True, exist_ok=True)
temporary = destination.with_name(".download-" + uuid.uuid4().hex)
try:
with self.plutonium_open(manifest["baseUrl"] + entry["hash"]) as incoming, temporary.open("xb") as outgoing:
size, checksum = 0, hashlib.sha1()
while True:
active()
chunk = incoming.read(min(1024 * 1024, entry["size"] - size + 1))
if not chunk:
break
size += len(chunk)
if size > entry["size"]:
raise AgentError("Plutonium download exceeds its expected size")
checksum.update(chunk)
outgoing.write(chunk)
if size != entry["size"] or checksum.hexdigest() != entry["hash"]:
raise AgentError("Plutonium download failed size or checksum verification")
outgoing.flush()
os.fsync(outgoing.fileno())
os.replace(temporary, destination)
sync_parent(destination)
finally:
with contextlib.suppress(FileNotFoundError):
temporary.unlink()
# Publish metadata last. Interrupted transfers can never claim ready.
active()
atomic_bytes(staging / "cdn_info.json", raw)
pointer = {"key": key, "revision": manifest["revision"], "bytes": sum(entry["size"] for entry in files)}
with self.plutonium_lock:
final = scoped(root, key)
if final.exists():
# Invalidate before replacing a damaged/same-revision cache.
if current_key == key:
(root / "current.json").unlink(missing_ok=True)
shutil.rmtree(final)
os.replace(staging.parent, final)
sync_parent(final)
atomic_json(root / "current.json", pointer)
if previous is not None and current_key != key:
shutil.rmtree(previous.parent)
return {"revision": pointer["revision"], "bytes": pointer["bytes"]}
except (urllib.error.URLError, TimeoutError, http.client.HTTPException):
raise AgentError("Plutonium CDN download is unavailable; the previous cache is retained") from None
def seed_plutonium(self, app, image):
if not re.fullmatch(r"ghcr\.io/ayymoss/plutainer(?::[A-Za-z0-9_.-]+)?(?:@sha256:[a-f0-9]{64})?", image):
return False
with self.plutonium_lock:
try:
_, content, raw, files = self.plutonium_snapshot()
if not all(self.plutonium_file_matches(content, entry) for entry in files):
scoped(self.root / "agent", "plutonium-cache/current.json").unlink(missing_ok=True)
raise AgentError("Plutonium cache checksum verification failed")
except (OSError, ValueError, AgentError):
return False
uid, gid = int(self.config.get("runtimeUid", 1000)), int(self.config.get("runtimeGid", 1000))
marker = "runtime/plutonium/cdn_info.json"
managed_write(app, marker, None, uid, gid)
for entry in files:
self.check_creation_cancellation()
managed_copy(app, "runtime/plutonium/" + entry["name"], scoped(content, entry["name"]), uid, gid)
managed_write(app, marker, raw, uid, gid)
return True
def create(self, instance_id, generation, payload):
self.check_creation_cancellation()
directory = self.instance_dir(instance_id, generation)
@@ -1527,6 +1779,7 @@ class Agent:
app = directory / "app"
app.mkdir(parents=True, exist_ok=True)
self.make_writable(app, directory=True)
self.seed_plutonium(app, image)
self.write_initial_files(app, payload.get("files", []))
atomic_json(directory / "spec.json", payload)
env_path = directory / "environment"
@@ -2184,6 +2437,8 @@ class Agent:
return {"image": image, "ready": True}
if kind == "host.inspect":
return self.inventory()
if kind == "plutonium.prepare":
return self.prepare_plutonium(payload)
if kind == "rcon.status-batch":
return self.rcon_status_batch(operation)
if kind == "library.download":
@@ -2290,7 +2545,7 @@ class Agent:
if not self.config.get("pendingCredential") or self.recover_credential():
return self.journal.update(operation["id"], "completed", {"rotated": True, "reconciled": True})
return row
if kind == "runtime.pull":
if kind in ("runtime.pull", "plutonium.prepare"):
# Docker resumes already downloaded layers. Pull again under
# the same operation ID: an existing tag can still refer to an
# older image and cannot prove an interrupted pull completed.
@@ -2298,6 +2553,8 @@ class Agent:
try:
result = self.dispatch(operation)
return self.journal.update(operation["id"], "completed", {**result, "reconciled": True})
except Held as err:
return self.journal.update(operation["id"], "accepted", {"blocked": "host-busy", "message": str(err)})
except Exception:
return self.journal.update(operation["id"], "failed", error="Interrupted runtime download could not be resumed")
if kind == "library.import":
@@ -2407,6 +2664,7 @@ class Agent:
# Only the heartbeat thread replaces this immutable snapshot; it never
# mutates the operation journal or races Docker cleanup against create.
self.cancelled_creations = frozenset(value for value in response.get("cancelledCreations", []) if isinstance(value, str))
self.defer_plutonium_preparation = response.get("deferPlutoniumPreparation") is True
keep_days = response.get("observationKeepDays")
self.observation_keep_days = keep_days if valid_observation_keep_days(keep_days) else 0
host = response.get("host", {})