"""Container operations for bundle updates. No bootstrap, migration, forced kill, or volume removal.""" import json import os from pathlib import Path import re import shlex import subprocess import time from bundle_format import APP_SERVICES, BundleError, INFRA_SERVICES, OPTIONAL_INFRA_SERVICES, check_database, digest def read_env(path): values = {} for line in Path(path).read_text().splitlines(): line = line.strip() if not line or line.startswith("#"): continue name, separator, raw = line.partition("=") if not separator or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name): raise BundleError("unsupported .env syntax; use literal KEY=VALUE assignments") if raw.startswith(("'", '"')): parts = shlex.split(raw, comments=True) if len(parts) != 1: raise BundleError("unsupported quoted .env value: " + name) raw = parts[0] else: raw = raw.split(" #", 1)[0].rstrip() if "${" in raw or "$(" in raw or "`" in raw: raise BundleError("shell expansion in .env is unsupported: " + name) values[name] = raw return values class Runtime: def __init__(self, root, mode, helper_dir, timeout=300): self.root = Path(root) self.mode = mode self.helpers = Path(helper_dir) self.timeout = timeout self.env = read_env(self.root / ".env") self.project = self.env.get("COMPOSE_PROJECT_NAME", "") if not re.fullmatch(r"[a-z0-9][a-z0-9_-]*", self.project): raise BundleError("set the existing COMPOSE_PROJECT_NAME explicitly in .env before registration") self.engine = shlex.split(os.environ.get("ENGINE", "podman" if mode == "podman-run" else "docker")) if not self.engine or ("podman" if mode == "podman-run" else "docker") not in self.engine: raise BundleError("ENGINE does not match deployment mode") self.env["NGINX_UPSTREAM"] = "podman" if mode == "podman-run" else "docker" self.shell_env = {k: v for k, v in os.environ.items() if k in ("PATH", "LANG", "LC_ALL", "DOCKER_HOST", "XDG_RUNTIME_DIR", "HOME")} self.stop_record = None def command(self, args, *, data=None, input_file=None, check=True, output=None, timeout=None): result = subprocess.run(self.engine + args, input=data, stdout=output or subprocess.PIPE, stdin=input_file, stderr=subprocess.PIPE, timeout=timeout or self.timeout) if check and result.returncode: # Commands may contain generated env arguments; do not echo them or app stderr. raise BundleError("container operation failed: " + " ".join(args[:2])) return result def inspect(self, name, optional=False): result = self.command(["container", "inspect", name], check=not optional) return json.loads(result.stdout)[0] if result.returncode == 0 else None def names(self, plan, service): base = plan["services"][service]["name"] return [f"{base}-{n}" if self.mode == "docker-compose" or n > 1 else base for n in range(1, plan["services"][service]["replicas"] + 1)] def plan(self, source, app_ref=None): environment = {**self.shell_env, **self.env} state_env = self.root / ".bundle-state/images.env" if state_env.exists(): environment.update(read_env(state_env)) if app_ref: environment["APP_IMAGE"] = app_ref arguments = ["python3", str(self.helpers / "_compose_to_plan.py"), str(Path(source) / "compose.yml"), str(self.root), "--engine", "podman" if self.mode == "podman-run" else "docker"] for name in ("compose.patches.yml", "compose.ports.yml"): if (self.root / name).exists(): arguments += ["--overlay", str(self.root / name)] result = subprocess.run(arguments, env=environment, capture_output=True) if result.returncode: raise BundleError("cannot resolve Compose plan; check missing environment values/overlays") plan = json.loads(result.stdout) if plan["project"] != self.project: raise BundleError("Compose project identity changed") required = set(APP_SERVICES) | set(INFRA_SERVICES) | {"nginx", "bootstrap"} if not required <= set(plan["services"]) <= required | set(OPTIONAL_INFRA_SERVICES): raise BundleError("unsupported service topology; review updater before changing services") return plan def image_id(self, ref): result = self.command(["image", "inspect", ref, "--format", "{{.Id}}"]) value = result.stdout.decode().strip() return value if value.startswith("sha256:") else "sha256:" + value def snapshot(self, plan): result = {} for service in plan["order"]: if service in ("bootstrap", "minio-init"): continue for name in self.names(plan, service): item = self.inspect(name) state = item["State"] if not state.get("Running") or state.get("Paused"): raise BundleError("installation is not fully running: " + name) image = item["Image"] if item["Image"].startswith("sha256:") else "sha256:" + item["Image"] result[name] = {"service": service, "image": image, "id": item["Id"], "environment": item["Config"]["Env"]} if service in APP_SERVICES + ("nginx",): self.check_stop_signal(service, item["Config"].get("StopSignal")) if image != self.image_id(plan["services"][service]["image"]): raise BundleError("running image differs from installation configuration: " + name) if service == "nginx": sources = {m["Source"] for m in item["Mounts"] if m["Type"] == "bind"} if str(self.root / "conf/nginx/app.conf") not in sources: raise BundleError("running edge belongs to a different installation path") if service == "postgres": pg_env = dict(x.split("=", 1) for x in item["Config"]["Env"] if "=" in x) if pg_env.get("POSTGRES_DB") != self.env["POSTGRESQL_DB"] or pg_env.get("POSTGRES_USER") != self.env["POSTGRESQL_ID"]: raise BundleError("running PostgreSQL does not match installation DB identity") # Actual DB/cache/storage bindings must be this stack, never host prod settings. if service == "web": env = dict(x.split("=", 1) for x in item["Config"]["Env"] if "=" in x) for key, value in {"POSTGRESQL_IP": "postgres", "REDIS_MAIN_IP": "redis", "REDIS_CACHE_IP": "redis", "MINIO_ENDPOINT": "minio:9000"}.items(): if env.get(key) != value: raise BundleError("unsupported external infrastructure binding: " + key) return result @staticmethod def check_stop_signal(service, signal): value = str(signal or "TERM").upper().removeprefix("SIG") accepted = ("QUIT", "3") if service == "nginx" else ("TERM", "15") if value not in accepted: raise BundleError("unsupported image/container stop signal: " + service) def validate_stop_signals(self, plan): for service in APP_SERVICES + ("nginx",): image = plan["services"][service]["image"] item = json.loads(self.command(["image", "inspect", image]).stdout)[0] self.check_stop_signal(service, item["Config"].get("StopSignal")) def database(self, plan): postgres = self.names(plan, "postgres")[0] script = r'''psql -X -qAt -v ON_ERROR_STOP=1 -U "$POSTGRES_USER" -d "$POSTGRES_DB" <<'SQL' BEGIN READ ONLY; SELECT json_build_object( 'migrations', (SELECT json_agg(json_build_array(app,name) ORDER BY app,name) FROM django_migrations), 'tables', (SELECT json_object_agg(tab,cols) FROM ( SELECT c.relname tab, json_object_agg(a.attname,json_build_object('type',format_type(a.atttypid,a.atttypmod),'nullable',NOT a.attnotnull)) cols FROM pg_attribute a JOIN pg_class c ON c.oid=a.attrelid JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='public' AND c.relkind IN ('r','p') AND a.attnum>0 AND NOT a.attisdropped GROUP BY c.relname ) s)); COMMIT; SQL''' result = self.command(["exec", "-i", postgres, "sh", "-c", script]) return json.loads(result.stdout) def python(self, plan, file, config=None): web = self.names(plan, "web")[0] code = Path(file).read_bytes() payload = json.dumps(config or {}).encode() + b"\n" + code result = self.command(["exec", "-i", web, "/opt/venv/bin/python", "-c", "import sys,json; cfg=json.loads(sys.stdin.readline()); exec(compile(sys.stdin.read(),'','exec'))"], data=payload) return result.stdout def rq(self, plan, action): return json.loads(self.python(plan, self.helpers / "checks/update_rq.py", {"action": action})) def signal_stop(self, plan, services): # Signal exactly once: a second TERM can force an RQ worker to kill its work horse. names = [name for service in services for name in self.names(plan, service)] for name in names: item = self.inspect(name, optional=True) if item and item["State"].get("Running"): # Persist before signaling. A crash in between conservatively requires # operator inspection rather than risking a second TERM to an RQ horse. token = item["Id"] + ":" + item["State"].get("StartedAt", "") if self.stop_record and not self.stop_record(token): continue if item["State"].get("Paused"): self.command(["unpause", name]) # Both engines define -1 as infinite stop wait: no fallback SIGKILL. # stop also suppresses restart policies. App images use TERM; nginx QUIT. try: self.command(["stop", "--time", "-1", name]) except subprocess.TimeoutExpired as exc: raise BundleError("graceful stop timed out; no forced termination was attempted") from exc self.wait_stopped(plan, services) def wait_stopped(self, plan, services): names = [name for service in services for name in self.names(plan, service)] deadline = time.monotonic() + self.timeout while any((self.inspect(n, optional=True) or {}).get("State", {}).get("Running") for n in names): if time.monotonic() >= deadline: raise BundleError("graceful stop timed out; no forced termination was attempted") time.sleep(1) def start(self, plan, services): for service in services: for name in self.names(plan, service): item = self.inspect(name) if item["State"].get("Paused"): self.command(["unpause", name]) if not item["State"].get("Running"): self.command(["start", name]) self.healthy(name) def healthy(self, name): deadline = time.monotonic() + self.timeout while True: item = self.inspect(name) state = item["State"] health = state.get("Health", state.get("Healthcheck", {})).get("Status") or "healthy" if state.get("Running") and health == "healthy": return if not state.get("Running") or time.monotonic() >= deadline: raise BundleError("service did not become healthy: " + name) time.sleep(1) def pause_scheduler(self, plan): for name in self.names(plan, "scheduler"): if not self.inspect(name)["State"].get("Paused"): self.command(["pause", name]) def drain(self, plan, seconds): deadline = time.monotonic() + seconds quiet = 0 while True: queues = self.rq(plan, "status") if not all(q["suspended"] for q in queues.values()): raise BundleError("RQ suspension was lost during drain") quiet = quiet + 1 if all(q["active"] == 0 and q["busy"] == 0 for q in queues.values()) else 0 if quiet >= 2: return if time.monotonic() >= deadline: raise BundleError("RQ drain timeout; update deferred") time.sleep(min(1, max(0, deadline - time.monotonic()))) def recreate(self, plan, services, images, transaction): for service in services: if service not in APP_SERVICES + ("nginx",): raise BundleError("ordinary update cannot recreate infrastructure/bootstrap") for name in self.names(plan, service): item = self.inspect(name, optional=True) if item and item["State"].get("Running"): raise BundleError("refusing to replace a running container: " + name) if self.mode == "docker-compose": override = Path(transaction) / "compose-images.json" overrides = {s: {"image": images[s]} for s in services} for s, values in plan.get("runtime_environment", {}).items(): if s in overrides: # Restore the exact previous app environment without overwriting .env. overrides[s]["environment"] = {**{k: None for k in self.env}, **dict(v.split("=", 1) for v in values)} override.write_text(json.dumps({"services": overrides})) override.chmod(0o600) args = ["compose", "--project-directory", str(self.root), "--env-file", str(self.root / ".env"), "-p", self.project, "-f", str(self.root / "compose.yml")] for filename in ("compose.patches.yml", "compose.ports.yml"): if (self.root / filename).exists(): args += ["-f", str(self.root / filename)] args += ["-f", str(override), "up", "--no-start", "--no-deps", "--no-build", "--pull", "never", "--force-recreate", *services] # Compose interpolation must use exactly this installation's env. env = {**self.shell_env, **self.env} result = subprocess.run(self.engine + args, env=env, capture_output=True, timeout=self.timeout) if result.returncode: raise BundleError("Compose service recreation failed") else: for service in services: spec = plan["services"][service] opts = spec["opts"] if service in plan.get("runtime_environment", {}): opts, skip = [], False for arg in spec["opts"]: if skip: skip = False elif arg in ("-e", "--env-file"): skip = True else: opts.append(arg) for value in plan["runtime_environment"][service]: opts += ["-e", value] for name in self.names(plan, service): if self.inspect(name, optional=True): self.command(["rm", name]) # stopped containers only; no -f or -v self.command(["create", "--name", name, "--network", spec["networks"][0], "--network-alias", spec["alias"], *opts, images[service], *spec["command"]]) for network in spec["networks"][1:]: self.command(["network", "connect", "--alias", spec["alias"], network, name]) # Multiple web entrypoints run collectstatic. Start/wait one at a time. self.start(plan, [s for s in services if s not in ("nginx", "scheduler")]) def remove_excess(self, before, after, services): for service in services: for name in set(self.names(before, service)) - set(self.names(after, service)): item = self.inspect(name, optional=True) if item: if item["State"].get("Running"): raise BundleError("refusing to remove running excess replica: " + name) self.command(["rm", name]) def static_volume(self, plan): item = self.inspect(self.names(plan, "web")[0]) mounts = [m for m in item["Mounts"] if m["Destination"] == "/app/static"] if len(mounts) != 1 or mounts[0]["Type"] != "volume": raise BundleError("static files must use a dedicated named volume") return mounts[0]["Name"] def static_backup(self, volume, path, image): with open(path, "wb") as out: self.command(["run", "--rm", "--network", "none", "--entrypoint", "tar", "-v", volume + ":/snapshot:ro", image, "-C", "/snapshot", "-cf", "-", "."], output=out) def static_restore(self, volume, path, image): # This is only the verified static volume, never DB/MinIO/media/RQ volumes. with open(path, "rb") as archive: self.command(["run", "--rm", "-i", "--network", "none", "--entrypoint", "sh", "-v", volume + ":/snapshot", image, "-ceu", "find /snapshot -mindepth 1 -maxdepth 1 -exec rm -rf -- {} +; tar -C /snapshot -xf -"], input_file=archive) def check_mounts(self, plan, manifest, services): for service in services: for name in self.names(plan, service): self.check_container_mounts(name, manifest) def check_container_mounts(self, name, manifest): for mount in self.inspect(name)["Mounts"]: if mount["Type"] != "bind": continue try: relative = Path(mount["Source"]).relative_to(self.root).as_posix() except ValueError: continue for file, entry in manifest["files"].items(): if file == relative or file.startswith(relative + "/"): suffix = file[len(relative):].lstrip("/") destination = str(Path(mount["Destination"]) / suffix) if suffix else mount["Destination"] got = self.command(["exec", name, "cat", destination]).stdout if digest(got) != entry["sha256"]: raise BundleError("stale bind mount content: " + file) def nginx_options(self, plan, source=None, manifest=None): opts, skip = [], False original = plan["services"]["nginx"]["opts"] for i, arg in enumerate(original): if skip: skip = False elif arg in ("-p", "--publish", "--restart"): skip = True elif arg == "-v" and source: value = original[i + 1] host, suffix = value.split(":", 1) try: relative = Path(host).relative_to(self.root).as_posix() except ValueError: relative = None if relative in manifest["files"]: host = str(Path(source) / relative) opts += ["-v", host + ":" + suffix] skip = True else: opts.append(arg) return opts def validate_nginx(self, plan, source, manifest): spec = plan["services"]["nginx"] self.command(["run", "--rm", "--network", spec["networks"][0], "--entrypoint", "nginx", *self.nginx_options(plan, source, manifest), spec["image"], "-t"]) def private_probe(self, plan, manifest, credentials, transaction): name = self.project + "-update-probe" spec = plan["services"]["nginx"] opts = self.nginx_options(plan) self.remove_probe(transaction) try: self.command(["run", "-d", "--name", name, "--network", spec["networks"][0], "--label", "pi-continuum.update=" + Path(transaction).name, "--network-alias", name, *opts, spec["image"], *spec["command"]]) self.healthy(name) self.command(["exec", name, "nginx", "-t"]) # Wait for nginx's listener, not merely a running container process. deadline = time.monotonic() + self.timeout while self.command(["exec", name, "sh", "-c", "nc -z 127.0.0.1 80 || nc -z 127.0.0.1 443"], check=False).returncode: if time.monotonic() >= deadline: raise BundleError("private nginx listener did not become ready") time.sleep(1) # Check the actual mounted bytes on the private replacement before reopening the edge. self.check_container_mounts(name, manifest) scheme = "https" if any("/conf/nginx/https.conf:" in arg for arg in opts) else "http" web_env = dict(v.split("=", 1) for v in self.inspect(self.names(plan, "web")[0])["Config"]["Env"]) config = {**credentials, "url": scheme + "://" + name, "secure_cookie": web_env.get("VITE_API_SERVER_URL", "").startswith("https://")} self.python(plan, self.helpers / "checks/update_login.py", config) self.python(plan, self.helpers / "checks/update_storage.py") self.python(plan, self.helpers / "checks/processing.py") finally: self.remove_probe(transaction) def remove_probe(self, transaction): name = self.project + "-update-probe" item = self.inspect(name, optional=True) if not item: return if (item["Config"].get("Labels") or {}).get("pi-continuum.update") != Path(transaction).name: raise BundleError("private verifier name belongs to another transaction") if item["State"].get("Running"): self.command(["kill", "--signal", "QUIT", name], check=False) deadline = time.monotonic() + self.timeout while self.inspect(name)["State"].get("Running"): if time.monotonic() >= deadline: raise BundleError("private verifier failed to stop normally") time.sleep(1) self.command(["rm", name]) def data_update(self, plan, task): web = self.names(plan, "web")[0] self.command(["exec", web, "/opt/venv/bin/python", "manage.py", task["command"]]) def check_schema(self, plan, manifest): check_database(manifest["database"], self.database(plan))