mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-24 12:42:49 +00:00
Own demo update and recovery through sustained host observation
Compose the existing signed stable demo paths with the Relay host lock, durable systemd ownership, full persistent estate capture and five-minute forward and recovery watches. Preserve uncertain attempts and failed rollback, refuse destructive headroom pruning, and keep native installed acceptance separate from connected source controls. Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
@@ -0,0 +1,504 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Demo-only host transaction. Called by the reviewed CI dispatcher, not a worker tool."""
|
||||
from __future__ import annotations
|
||||
|
||||
import fcntl
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import signal
|
||||
import socket
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
WINDOW = 300
|
||||
INTERVAL = 5
|
||||
READY_TIMEOUT = 60
|
||||
SERVICES = ("pulse", "pulse-relay", "caddy")
|
||||
ROOT = Path("/var/lib/pulse-deploy/demo")
|
||||
LOCK = Path("/var/lib/pulse-deploy/relay/deploy.lock")
|
||||
PATHS = {"binary": Path("/opt/pulse/bin/pulse"),
|
||||
"unit": Path("/etc/systemd/system/pulse.service"),
|
||||
"dropins": Path("/etc/systemd/system/pulse.service.d"),
|
||||
"data": Path("/etc/pulse")}
|
||||
RELAY_HEALTH = "https://relay.pulserelay.pro/health"
|
||||
COUNT_KEYS = ("nodes", "vms_per_node", "lxcs_per_node", "docker_hosts",
|
||||
"docker_containers", "generic_hosts", "k8s_clusters", "k8s_nodes",
|
||||
"k8s_pods", "k8s_deployments")
|
||||
DURATION_KEYS = ("seed_duration", "sample_interval", "update_interval")
|
||||
# Only generated demo operational memory is reset on the already-unhealthy
|
||||
# recovery route. Its complete original estate remains in the private snapshot.
|
||||
DEMO_HISTORY = ("ai_incidents.json", "ai_incidents.json.tmp",
|
||||
"alerts/alert-history.json", "alerts/alert-history.backup.json",
|
||||
"alerts/alert-history.json.imported", "alerts/alert-history.backup.json.imported",
|
||||
"alerts/events.db", "alerts/events.db-shm", "alerts/events.db-wal")
|
||||
CRASH = re.compile(r"\bpanic\b|\bfatal error\b|segmentation fault|core dumped|watchdog timeout", re.I)
|
||||
TERMINAL = {"committed", "healthy_noop", "rolled_back", "refused"}
|
||||
|
||||
|
||||
class Failure(Exception):
|
||||
"""Only fixed stage names, never command output or private response bodies."""
|
||||
|
||||
|
||||
def atomic_json(path, value):
|
||||
tmp = path.with_suffix(".tmp")
|
||||
with open(tmp, "w", encoding="utf-8") as stream:
|
||||
os.chmod(tmp, 0o600)
|
||||
json.dump(value, stream, sort_keys=True)
|
||||
stream.flush()
|
||||
os.fsync(stream.fileno())
|
||||
os.replace(tmp, path)
|
||||
fd = os.open(path.parent, os.O_DIRECTORY)
|
||||
try:
|
||||
os.fsync(fd)
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
|
||||
def validate(request):
|
||||
if set(request) != {"mode", "hostname", "local_url", "public_url", "version", "profile", "control_sha", "run_id", "run_attempt", "installer_sha256"}:
|
||||
raise Failure("request-shape")
|
||||
if request["mode"] not in {"update", "recover"}:
|
||||
raise Failure("request-mode")
|
||||
for field, pattern in (("version", r"v[0-9]+\.[0-9]+\.[0-9]+"),
|
||||
("hostname", r"[a-zA-Z0-9][a-zA-Z0-9.-]{0,252}"),
|
||||
("control_sha", r"[0-9a-f]{40}"),
|
||||
("run_id", r"[1-9][0-9]*"), ("run_attempt", r"[1-9][0-9]*")):
|
||||
if not isinstance(request[field], str) or not re.fullmatch(pattern, request[field]):
|
||||
raise Failure("request-identity")
|
||||
for field, public in (("local_url", False), ("public_url", True)):
|
||||
try:
|
||||
url = urlsplit(request[field])
|
||||
valid = (url.scheme == ("https" if public else "http")
|
||||
and url.hostname and not url.username and not url.password
|
||||
and not url.query and not url.fragment
|
||||
and (url.path == "/api/health" if public else url.path in {"", "/"})
|
||||
and (public or url.hostname in {"localhost", "127.0.0.1", "::1"})
|
||||
and 0 < (url.port or (443 if public else 80)) < 65536)
|
||||
except (TypeError, ValueError):
|
||||
valid = False
|
||||
if not valid:
|
||||
raise Failure("request-url")
|
||||
profile = request["profile"]
|
||||
if not isinstance(profile, dict) or set(profile) != set(COUNT_KEYS + DURATION_KEYS):
|
||||
raise Failure("request-profile")
|
||||
for name, value in profile.items():
|
||||
pattern = r"[1-9][0-9]{0,3}" if name in COUNT_KEYS else r"[1-9][0-9]{0,3}[smhd]"
|
||||
if not isinstance(value, str) or not re.fullmatch(pattern, value):
|
||||
raise Failure("request-profile")
|
||||
digest = request["installer_sha256"]
|
||||
if (request["mode"] == "update" and not re.fullmatch(r"[0-9a-f]{64}", digest)) or (request["mode"] == "recover" and digest != ""):
|
||||
raise Failure("request-installer")
|
||||
|
||||
|
||||
def copy_path(src, dst):
|
||||
"""Copy without following symlinks, retaining ownership as well as modes."""
|
||||
info = src.lstat()
|
||||
if stat.S_ISLNK(info.st_mode):
|
||||
dst.symlink_to(os.readlink(src))
|
||||
elif stat.S_ISDIR(info.st_mode):
|
||||
dst.mkdir(mode=0o700)
|
||||
for child in src.iterdir():
|
||||
copy_path(child, dst / child.name)
|
||||
shutil.copystat(src, dst, follow_symlinks=False)
|
||||
elif stat.S_ISREG(info.st_mode):
|
||||
shutil.copy2(src, dst, follow_symlinks=False)
|
||||
else:
|
||||
raise Failure("snapshot-special-file")
|
||||
os.chown(dst, info.st_uid, info.st_gid, follow_symlinks=False)
|
||||
if not stat.S_ISLNK(info.st_mode):
|
||||
fd = os.open(dst, os.O_RDONLY | (os.O_DIRECTORY if stat.S_ISDIR(info.st_mode) else 0))
|
||||
try:
|
||||
os.fsync(fd)
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
|
||||
def remove_path(path):
|
||||
if path.is_symlink() or path.is_file():
|
||||
path.unlink()
|
||||
elif path.is_dir():
|
||||
shutil.rmtree(path)
|
||||
|
||||
|
||||
def estate_hash(paths):
|
||||
rows = []
|
||||
for label, root in sorted(paths.items()):
|
||||
exists = root.exists() or root.is_symlink()
|
||||
if not exists:
|
||||
rows.append((label, "absent"))
|
||||
continue
|
||||
for path in [root] + (sorted(root.rglob("*")) if root.is_dir() and not root.is_symlink() else []):
|
||||
info = path.lstat()
|
||||
kind = stat.S_IFMT(info.st_mode)
|
||||
content = os.readlink(path) if stat.S_ISLNK(info.st_mode) else ""
|
||||
if stat.S_ISREG(info.st_mode):
|
||||
digest = hashlib.sha256()
|
||||
with open(path, "rb") as stream:
|
||||
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
digest.update(block)
|
||||
content = digest.hexdigest()
|
||||
rows.append((label, str(path.relative_to(root)), kind, stat.S_IMODE(info.st_mode), info.st_uid, info.st_gid, content))
|
||||
return hashlib.sha256(json.dumps(rows, sort_keys=True).encode()).hexdigest()
|
||||
|
||||
|
||||
class Host:
|
||||
now = staticmethod(time.monotonic)
|
||||
sleep = staticmethod(time.sleep)
|
||||
|
||||
def command(self, argv, *, timeout=20, env=None):
|
||||
try:
|
||||
result = subprocess.run(argv, capture_output=True, timeout=timeout, env=env, check=False)
|
||||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||||
raise Failure("command-unavailable") from exc
|
||||
if result.returncode or len(result.stdout) > 1024 * 1024:
|
||||
raise Failure("command-failed")
|
||||
return result.stdout.decode("utf-8", errors="strict")
|
||||
|
||||
def identity(self, hostname):
|
||||
if socket.gethostname() != hostname or os.geteuid() != 0:
|
||||
raise Failure("host-identity")
|
||||
unit = self.command(["systemctl", "show", "pulse", "--property=FragmentPath,User,Group,ExecStart,DropInPaths", "--no-pager"])
|
||||
fields = dict(line.split("=", 1) for line in unit.splitlines() if "=" in line)
|
||||
if (fields.get("FragmentPath") != str(PATHS["unit"])
|
||||
or fields.get("User") != "pulse" or fields.get("Group") != "pulse"
|
||||
or str(PATHS["binary"]) not in fields.get("ExecStart", "")):
|
||||
raise Failure("service-identity")
|
||||
for name in ("binary", "unit", "data"):
|
||||
if PATHS[name].is_symlink() or not PATHS[name].exists():
|
||||
raise Failure("estate-identity")
|
||||
if PATHS["dropins"].is_symlink():
|
||||
raise Failure("dropin-identity")
|
||||
for path in fields.get("DropInPaths", "").split():
|
||||
if Path(path).parent != PATHS["dropins"] or Path(path).is_symlink():
|
||||
raise Failure("unsupported-dropin")
|
||||
if self.command(["systemctl", "is-enabled", "pulse"]).strip() != "enabled":
|
||||
raise Failure("service-disabled")
|
||||
|
||||
def state(self):
|
||||
states = {}
|
||||
for service in SERVICES:
|
||||
text = self.command(["systemctl", "show", service, "--property=ActiveState,SubState,MainPID,NRestarts", "--no-pager"])
|
||||
fields = dict(line.split("=", 1) for line in text.splitlines() if "=" in line)
|
||||
if (set(fields) != {"ActiveState", "SubState", "MainPID", "NRestarts"}
|
||||
or not fields["MainPID"].isdigit() or not fields["NRestarts"].isdigit()):
|
||||
raise Failure("service-observation")
|
||||
states[service] = fields
|
||||
return states
|
||||
|
||||
def status(self, url):
|
||||
# No redirects, response bodies, credential arguments or ambient curl config.
|
||||
try:
|
||||
result = self.command(["curl", "--disable", "--silent", "--connect-timeout", "3", "--max-time", "8", "--output", "/dev/null", "--write-out", "%{http_code}", "--", url], timeout=10)
|
||||
except Failure:
|
||||
return "000"
|
||||
return result if re.fullmatch(r"[1-5][0-9]{2}", result) else "000"
|
||||
|
||||
def version(self, url):
|
||||
try:
|
||||
body = self.command(["curl", "--disable", "--silent", "--fail", "--connect-timeout", "3", "--max-time", "8", "--", url.rstrip("/") + "/api/version"], timeout=10)
|
||||
if len(body) > 8192:
|
||||
raise Failure("version-observation")
|
||||
value = json.loads(body)["version"]
|
||||
if not isinstance(value, str) or not re.fullmatch(r"v?[0-9]+\.[0-9]+\.[0-9]+", value):
|
||||
raise Failure("version-observation")
|
||||
return value.lstrip("v")
|
||||
except (ValueError, KeyError, TypeError) as exc:
|
||||
raise Failure("version-observation") from exc
|
||||
|
||||
def cursors(self):
|
||||
cursors = {}
|
||||
for service in SERVICES:
|
||||
text = self.command(["journalctl", "-u", service, "-n", "1", "--output=json", "--show-cursor", "--no-pager"])
|
||||
matches = re.findall(r"^-- cursor: ([^\s]+)$", text, re.M)
|
||||
if len(matches) != 1:
|
||||
raise Failure("journal-cursor")
|
||||
cursors[service] = matches[0]
|
||||
return cursors
|
||||
|
||||
def check_journal(self, cursors):
|
||||
for service, cursor in cursors.items():
|
||||
text = self.command(["journalctl", "-u", service, "--after-cursor", cursor, "--output=json", "--no-pager"])
|
||||
for line in text.splitlines():
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
if not isinstance(entry.get("MESSAGE"), str) or not isinstance(entry.get("__CURSOR"), str):
|
||||
raise Failure("journal-observation")
|
||||
except (ValueError, TypeError, AttributeError) as exc:
|
||||
raise Failure("journal-observation") from exc
|
||||
if CRASH.search(entry["MESSAGE"]):
|
||||
raise Failure("new-service-crash")
|
||||
cursors[service] = entry["__CURSOR"]
|
||||
|
||||
def stop(self):
|
||||
self.command(["systemctl", "stop", "pulse"], timeout=90)
|
||||
|
||||
def start(self):
|
||||
self.command(["systemctl", "daemon-reload"])
|
||||
self.command(["systemctl", "start", "pulse"], timeout=90)
|
||||
|
||||
def install(self, attempt, request):
|
||||
installer = attempt / "installer.sh"
|
||||
if hashlib.sha256(installer.read_bytes()).hexdigest() != request["installer_sha256"]:
|
||||
raise Failure("installer-identity")
|
||||
self.command(["bash", str(installer), "--version", request["version"]], timeout=900,
|
||||
env={"PATH": "/usr/sbin:/usr/bin:/sbin:/bin", "HOME": "/root", "PULSE_SERVICE_NAME": "pulse"})
|
||||
|
||||
def profile_values(self, request):
|
||||
values = {"DEMO_MODE": "true", "PULSE_MOCK_MODE": "true", "PULSE_MOCK_RANDOM_METRICS": "true",
|
||||
"PULSE_MOCK_STOPPED_PERCENT": "6", "PULSE_MOCK_SEED_METRICS_STORE": "false"}
|
||||
for key, value in request["profile"].items():
|
||||
key = {"seed_duration": "trends_seed_duration", "sample_interval": "trends_sample_interval"}.get(key, key)
|
||||
values["PULSE_MOCK_" + key.upper()] = value
|
||||
return values
|
||||
|
||||
def profile_matches(self, request):
|
||||
text = (PATHS["data"] / ".env").read_text().splitlines()
|
||||
values = self.profile_values(request)
|
||||
return (all(text.count(key + "=" + value) == 1 for key, value in values.items())
|
||||
and "demo_fixtures" in json.loads((PATHS["data"] / "billing.json").read_text()).get("capabilities", []))
|
||||
|
||||
def profile(self, request, unhealthy):
|
||||
env_file = PATHS["data"] / ".env"
|
||||
billing_file = PATHS["data"] / "billing.json"
|
||||
if env_file.is_symlink() or billing_file.is_symlink():
|
||||
raise Failure("configuration-symlink")
|
||||
values = self.profile_values(request)
|
||||
text = env_file.read_text()
|
||||
for key, value in values.items():
|
||||
pattern = re.compile(r"^[ \t]*" + re.escape(key) + r"=.*$", re.M)
|
||||
text = pattern.sub(key + "=" + value, text) if pattern.search(text) else text.rstrip("\n") + "\n" + key + "=" + value + "\n"
|
||||
env_file.write_text(text)
|
||||
os.chmod(env_file, 0o600)
|
||||
billing = json.loads(billing_file.read_text())
|
||||
if not isinstance(billing, dict) or not isinstance(billing.get("capabilities", []), list):
|
||||
raise Failure("billing-shape")
|
||||
billing["capabilities"] = sorted(set(billing.get("capabilities", []) + ["demo_fixtures"]))
|
||||
billing.pop("integrity", None)
|
||||
billing_file.write_text(json.dumps(billing))
|
||||
os.chmod(billing_file, 0o600)
|
||||
if unhealthy:
|
||||
for relative in DEMO_HISTORY:
|
||||
path = PATHS["data"] / relative
|
||||
# Never follow a writable parent symlink to a path outside the estate.
|
||||
if any(p.is_symlink() for p in path.parents if p != PATHS["data"].parent):
|
||||
raise Failure("history-symlink")
|
||||
if path.is_symlink():
|
||||
raise Failure("history-symlink")
|
||||
if path.is_file():
|
||||
path.unlink()
|
||||
|
||||
|
||||
def active(state):
|
||||
return state["ActiveState"] == "active" and state["SubState"] == "running" and int(state["MainPID"]) > 0
|
||||
|
||||
|
||||
class Transaction:
|
||||
def __init__(self, host, request, attempt):
|
||||
self.host, self.request, self.attempt = host, request, attempt
|
||||
self.receipt = {"schema_version": 2, "status": "intent", "mutated": False,
|
||||
"control_sha": request["control_sha"], "run_id": request["run_id"],
|
||||
"run_attempt": request["run_attempt"], "mode": request["mode"],
|
||||
"expected_version": request["version"], "installer_sha256": request["installer_sha256"],
|
||||
"forward": {}, "recovery": {}, "recovery_required": False}
|
||||
|
||||
def save(self, status):
|
||||
self.receipt["status"] = status
|
||||
atomic_json(self.attempt / "receipt.json", self.receipt)
|
||||
|
||||
def urls(self):
|
||||
return (self.request["local_url"].rstrip("/") + "/api/health", self.request["public_url"], RELAY_HEALTH)
|
||||
|
||||
def healthy(self, states=None):
|
||||
return all(active(s) for s in (states or self.host.state()).values()) and all(self.host.status(url) == "200" for url in self.urls())
|
||||
|
||||
def ready(self, cursors):
|
||||
deadline = self.host.now() + READY_TIMEOUT
|
||||
while True:
|
||||
self.host.check_journal(cursors)
|
||||
if self.healthy():
|
||||
return
|
||||
if self.host.now() >= deadline:
|
||||
raise Failure("listener-readiness")
|
||||
self.host.sleep(INTERVAL)
|
||||
|
||||
def watch(self, phase, expected_version, cursors):
|
||||
baseline = self.host.state()
|
||||
started = self.host.now()
|
||||
evidence = self.receipt[phase]
|
||||
evidence.update({"required_seconds": WINDOW, "samples": 0, "elapsed_seconds": 0})
|
||||
while True:
|
||||
self.host.check_journal(cursors)
|
||||
state = self.host.state()
|
||||
if state != baseline or not self.healthy(state):
|
||||
raise Failure("sustained-health")
|
||||
if self.host.version(self.request["local_url"]) != expected_version:
|
||||
raise Failure("runtime-version")
|
||||
evidence["samples"] += 1
|
||||
elapsed = self.host.now() - started
|
||||
evidence["elapsed_seconds"] = elapsed
|
||||
if elapsed >= WINDOW:
|
||||
return
|
||||
self.host.sleep(min(INTERVAL, WINDOW - elapsed))
|
||||
|
||||
def capture(self):
|
||||
snapshot = self.attempt / "snapshot"
|
||||
size = sum(path.lstat().st_size for root in PATHS.values() if root.exists()
|
||||
for path in ([root] + (list(root.rglob("*")) if root.is_dir() else []))
|
||||
if path.is_file() and not path.is_symlink())
|
||||
# Both our independent estate and the signed installer's own backup need
|
||||
# space. Refuse shortage; never delete backups or databases for headroom.
|
||||
if any(shutil.disk_usage(parent).free < 2 * size + 64 * 1024 * 1024
|
||||
for parent in (self.attempt, PATHS["data"].parent)):
|
||||
raise Failure("snapshot-headroom")
|
||||
snapshot.mkdir(mode=0o700)
|
||||
manifest = {}
|
||||
for label, path in PATHS.items():
|
||||
manifest[label] = path.exists() or path.is_symlink()
|
||||
if manifest[label]:
|
||||
copy_path(path, snapshot / label)
|
||||
atomic_json(snapshot / "paths.json", manifest)
|
||||
digest = estate_hash(PATHS)
|
||||
self.receipt["snapshot_sha256"] = digest
|
||||
self.receipt["snapshot_retained"] = True
|
||||
return digest
|
||||
|
||||
def restore(self):
|
||||
snapshot = self.attempt / "snapshot"
|
||||
for label, exists in json.loads((snapshot / "paths.json").read_text()).items():
|
||||
remove_path(PATHS[label])
|
||||
if exists:
|
||||
copy_path(snapshot / label, PATHS[label])
|
||||
if estate_hash(PATHS) != self.receipt["snapshot_sha256"]:
|
||||
raise Failure("restored-estate-identity")
|
||||
|
||||
def run(self):
|
||||
old_handlers = {}
|
||||
healthy_baseline = False
|
||||
stopped = False
|
||||
original_version = ""
|
||||
states = {}
|
||||
|
||||
def cancel(_signum, _frame):
|
||||
raise Failure("cancelled-forward")
|
||||
|
||||
try:
|
||||
validate(self.request)
|
||||
self.host.identity(self.request["hostname"])
|
||||
for entry in ROOT.glob("attempts/*"):
|
||||
if entry == self.attempt:
|
||||
continue
|
||||
receipt = entry / "receipt.json"
|
||||
if not receipt.exists() or json.loads(receipt.read_text()).get("status") not in TERMINAL:
|
||||
raise Failure("prior-recovery-required")
|
||||
states = self.host.state()
|
||||
if not all(active(states[name]) for name in ("pulse-relay", "caddy")) or self.host.status(RELAY_HEALTH) != "200":
|
||||
raise Failure("cohost-unhealthy")
|
||||
healthy_baseline = self.healthy(states)
|
||||
if healthy_baseline:
|
||||
original_version = self.host.version(self.request["local_url"])
|
||||
if self.request["mode"] == "recover" and original_version != self.request["version"].lstrip("v"):
|
||||
raise Failure("recovery-source-mismatch")
|
||||
elif self.request["mode"] == "update":
|
||||
raise Failure("update-baseline-unhealthy")
|
||||
self.receipt["healthy_baseline"] = healthy_baseline
|
||||
# The no-op is observed too; an immediate healthy sample is not success.
|
||||
if self.request["mode"] == "recover" and healthy_baseline and self.host.profile_matches(self.request):
|
||||
self.watch("forward", original_version, self.host.cursors())
|
||||
self.save("healthy_noop")
|
||||
return 0
|
||||
old_handlers = {sig: signal.getsignal(sig) for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP)}
|
||||
for sig in old_handlers:
|
||||
signal.signal(sig, cancel)
|
||||
self.save("capturing")
|
||||
stopped = True
|
||||
self.host.stop()
|
||||
self.capture()
|
||||
self.save("applying")
|
||||
self.receipt["mutated"] = True
|
||||
cursors = self.host.cursors()
|
||||
if self.request["mode"] == "update":
|
||||
self.host.install(self.attempt, self.request)
|
||||
# The installer may start the unit; finish profile work quiescent.
|
||||
self.host.stop()
|
||||
self.host.profile(self.request, not healthy_baseline)
|
||||
self.host.start()
|
||||
self.ready(cursors)
|
||||
self.watch("forward", self.request["version"].lstrip("v"), cursors)
|
||||
# No unrelated Relay/Caddy restart is owned by this transaction.
|
||||
final = self.host.state()
|
||||
for service in ("pulse-relay", "caddy"):
|
||||
if final[service] != states[service]:
|
||||
raise Failure("cohost-identity-changed")
|
||||
self.save("committed")
|
||||
return 0
|
||||
except Exception as exc:
|
||||
self.receipt["failure"] = str(exc) if isinstance(exc, Failure) else "transaction-observation"
|
||||
# Owned restoration, observation and the terminal receipt survive cancellation.
|
||||
for sig in old_handlers:
|
||||
signal.signal(sig, signal.SIG_IGN)
|
||||
if stopped:
|
||||
try:
|
||||
self.save("recovering")
|
||||
self.host.stop()
|
||||
if "snapshot_sha256" in self.receipt:
|
||||
self.restore()
|
||||
if not healthy_baseline:
|
||||
self.receipt["recovery_required"] = True
|
||||
self.receipt["rollback"] = "unavailable-unhealthy-baseline"
|
||||
self.save("recovery_required")
|
||||
else:
|
||||
cursors = self.host.cursors()
|
||||
self.host.start()
|
||||
self.ready(cursors)
|
||||
self.watch("recovery", original_version, cursors)
|
||||
final = self.host.state()
|
||||
if any(final[name] != states[name] for name in ("pulse-relay", "caddy")):
|
||||
raise Failure("cohost-identity-changed")
|
||||
self.receipt["rollback"] = "verified"
|
||||
self.save("rolled_back")
|
||||
except Exception as recovery:
|
||||
self.receipt["rollback"] = "failed"
|
||||
self.receipt["recovery_failure"] = str(recovery) if isinstance(recovery, Failure) else "recovery-observation"
|
||||
self.receipt["recovery_required"] = True
|
||||
self.save("rollback_failed")
|
||||
else:
|
||||
self.save("refused")
|
||||
return 1
|
||||
finally:
|
||||
for sig, handler in old_handlers.items():
|
||||
signal.signal(sig, handler)
|
||||
|
||||
|
||||
def main():
|
||||
# Attempt paths are generated by the fixed CI bootstrap, not CLI overrides.
|
||||
if len(sys.argv) != 2:
|
||||
return 2
|
||||
attempt = Path(sys.argv[1])
|
||||
if attempt.parent != ROOT / "attempts" or not re.fullmatch(r"[0-9a-f]{64}", attempt.name) or attempt.is_symlink():
|
||||
return 2
|
||||
request = json.loads((attempt / "request.json").read_text())
|
||||
transaction = Transaction(Host(), request, attempt)
|
||||
fd = None
|
||||
try:
|
||||
LOCK.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
fd = os.open(LOCK, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
|
||||
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
return transaction.run()
|
||||
except (OSError, ValueError):
|
||||
transaction.receipt["failure"] = "host-lock-unavailable"
|
||||
transaction.save("refused")
|
||||
return 1
|
||||
finally:
|
||||
if fd is not None:
|
||||
os.close(fd)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,113 @@
|
||||
#!/usr/bin/env python3
|
||||
"""The two existing demo CI paths share one retained, independently owned child."""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
BOOTSTRAP = r'''
|
||||
import base64, fcntl, hashlib, json, os, pathlib, subprocess, sys, time
|
||||
root = pathlib.Path('/var/lib/pulse-deploy/demo')
|
||||
payload = json.loads(sys.stdin.read(512 * 1024))
|
||||
assert os.geteuid() == 0 and set(payload) == {'request', 'source', 'installer'}
|
||||
source = base64.b64decode(payload['source'], validate=True)
|
||||
installer = base64.b64decode(payload['installer'], validate=True)
|
||||
assert len(source) < 128 * 1024 and len(installer) < 256 * 1024
|
||||
assert (hashlib.sha256(installer).hexdigest() == payload['request']['installer_sha256']) if installer else (payload['request']['installer_sha256'] == '')
|
||||
identity = hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
||||
root.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
assert not root.is_symlink() and root.stat().st_uid == 0
|
||||
os.chmod(root, 0o700)
|
||||
(root / 'attempts').mkdir(mode=0o700, exist_ok=True)
|
||||
assert not (root / 'attempts').is_symlink()
|
||||
fd = os.open(root / 'submit.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
|
||||
fcntl.flock(fd, fcntl.LOCK_EX)
|
||||
attempt = root / 'attempts' / identity
|
||||
if not attempt.exists():
|
||||
attempt.mkdir(mode=0o700)
|
||||
for name, data in [('transaction.py', source), ('installer.sh', installer),
|
||||
('request.json', json.dumps(payload['request']).encode()),
|
||||
('intent.json', json.dumps({'schema_version': 1, 'identity': identity, 'state': 'intent'}).encode())]:
|
||||
with open(attempt / name, 'xb') as stream:
|
||||
os.chmod(attempt / name, 0o600)
|
||||
stream.write(data); stream.flush(); os.fsync(stream.fileno())
|
||||
# Intent is durable before starting the unit. An uncertain launch is never replayed.
|
||||
dfd = os.open(attempt, os.O_DIRECTORY); os.fsync(dfd); os.close(dfd)
|
||||
try:
|
||||
child = subprocess.run(['systemd-run', '--quiet', '--collect',
|
||||
'--unit=pulse-demo-' + identity[:32],
|
||||
'--property=RuntimeMaxSec=45min',
|
||||
'--property=TimeoutStopSec=20min', '--property=KillMode=mixed',
|
||||
'/usr/bin/python3', str(attempt / 'transaction.py'), str(attempt)],
|
||||
capture_output=True, timeout=30)
|
||||
launch = {'schema_version': 1, 'state': 'submitted' if child.returncode == 0 else 'uncertain',
|
||||
'unit': 'pulse-demo-' + identity[:32]}
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
launch = {'schema_version': 1, 'state': 'uncertain'}
|
||||
with open(attempt / 'launch.json', 'x') as stream:
|
||||
os.chmod(attempt / 'launch.json', 0o600)
|
||||
json.dump(launch, stream); stream.flush(); os.fsync(stream.fileno())
|
||||
os.close(fd)
|
||||
# Closing or losing this observer never stops the independently owned unit.
|
||||
deadline = time.monotonic() + 4000
|
||||
terminal = {'committed', 'healthy_noop', 'rolled_back', 'rollback_failed', 'recovery_required', 'refused'}
|
||||
while time.monotonic() < deadline:
|
||||
try:
|
||||
receipt = json.loads((attempt / 'receipt.json').read_text())
|
||||
except (FileNotFoundError, ValueError):
|
||||
receipt = None
|
||||
if receipt and receipt.get('status') in terminal:
|
||||
receipt['request_identity'] = identity
|
||||
print(json.dumps(receipt))
|
||||
sys.exit(0 if receipt['status'] in {'committed', 'healthy_noop'} else 1)
|
||||
launch = attempt / 'launch.json'
|
||||
if launch.exists() and json.loads(launch.read_text()).get('state') == 'uncertain':
|
||||
break
|
||||
time.sleep(5)
|
||||
print(json.dumps({'schema_version': 2, 'status': 'uncertain', 'request_identity': identity,
|
||||
'failure': 'terminal-receipt-unavailable', 'recovery_required': True}))
|
||||
sys.exit(1)
|
||||
'''
|
||||
|
||||
|
||||
def payload(mode, environ=os.environ):
|
||||
profile = {key: environ['MOCK_' + key.upper()] for key in
|
||||
('nodes', 'vms_per_node', 'lxcs_per_node', 'docker_hosts', 'docker_containers',
|
||||
'generic_hosts', 'k8s_clusters', 'k8s_nodes', 'k8s_pods', 'k8s_deployments',
|
||||
'seed_duration', 'sample_interval', 'update_interval')}
|
||||
installer = Path('/tmp/pulse-install.sh').read_bytes() if mode == 'update' else b''
|
||||
request = {'mode': mode, 'hostname': environ['DEMO_EXPECTED_HOSTNAME'],
|
||||
'local_url': environ['DEMO_LOCAL_BASE_URL'], 'public_url': environ['DEMO_PUBLIC_HEALTH_URL'],
|
||||
'version': environ['EXPECTED_VERSION'], 'profile': profile,
|
||||
'control_sha': environ['GITHUB_SHA'], 'run_id': environ['GITHUB_RUN_ID'],
|
||||
'run_attempt': environ['GITHUB_RUN_ATTEMPT'],
|
||||
'installer_sha256': hashlib.sha256(installer).hexdigest() if installer else ''}
|
||||
return {'request': request, 'source': base64.b64encode(Path(__file__).with_name('demo-runtime-transaction.py').read_bytes()).decode(),
|
||||
'installer': base64.b64encode(installer).decode()}
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) != 2 or sys.argv[1] not in {'update', 'recover'}:
|
||||
return 2
|
||||
# Same project CI SSH identity and pinned host verification as before. This
|
||||
# is not an installed worker dispatcher or a new project credential.
|
||||
command = ['ssh', '-i', str(Path.home() / '.ssh/id_ed25519'), '-o', 'IdentitiesOnly=yes',
|
||||
'-o', 'StrictHostKeyChecking=yes', '-o', 'UserKnownHostsFile=' + str(Path.home() / '.ssh/known_hosts'),
|
||||
os.environ['DEMO_SERVER_USER'] + '@' + os.environ['DEMO_SERVER_HOST'],
|
||||
'sudo python3 -c ' + shlex.quote(BOOTSTRAP)]
|
||||
try:
|
||||
result = subprocess.run(command, input=json.dumps(payload(sys.argv[1])).encode(), timeout=4100, check=False)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
print(json.dumps({'schema_version': 2, 'status': 'uncertain', 'failure': 'observer-unavailable', 'recovery_required': True}))
|
||||
return 1
|
||||
return result.returncode
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
@@ -1,395 +1,4 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
if [ "$#" -ne 16 ]; then
|
||||
echo "usage: recover-demo-runtime.sh EXPECTED_HOSTNAME LOCAL_BASE_URL EXPECTED_VERSION MOCK_NODES MOCK_VMS_PER_NODE MOCK_LXCS_PER_NODE MOCK_DOCKER_HOSTS MOCK_DOCKER_CONTAINERS MOCK_GENERIC_HOSTS MOCK_K8S_CLUSTERS MOCK_K8S_NODES MOCK_K8S_PODS MOCK_K8S_DEPLOYMENTS MOCK_SEED_DURATION MOCK_SAMPLE_INTERVAL MOCK_UPDATE_INTERVAL" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
EXPECTED_HOSTNAME="$1"
|
||||
LOCAL_BASE_URL="${2%/}"
|
||||
EXPECTED_VERSION="${3#v}"
|
||||
MOCK_NODES="$4"
|
||||
MOCK_VMS_PER_NODE="$5"
|
||||
MOCK_LXCS_PER_NODE="$6"
|
||||
MOCK_DOCKER_HOSTS="$7"
|
||||
MOCK_DOCKER_CONTAINERS="$8"
|
||||
MOCK_GENERIC_HOSTS="$9"
|
||||
MOCK_K8S_CLUSTERS="${10}"
|
||||
MOCK_K8S_NODES="${11}"
|
||||
MOCK_K8S_PODS="${12}"
|
||||
MOCK_K8S_DEPLOYMENTS="${13}"
|
||||
MOCK_SEED_DURATION="${14}"
|
||||
MOCK_SAMPLE_INTERVAL="${15}"
|
||||
MOCK_UPDATE_INTERVAL="${16}"
|
||||
SERVICE_NAME="pulse"
|
||||
RELAY_SERVICE_NAME="pulse-relay"
|
||||
EXPECTED_BINARY="/opt/pulse/bin/pulse"
|
||||
EXPECTED_UNIT="/etc/systemd/system/pulse.service"
|
||||
MUTATED=false
|
||||
BACKUP_DIR=""
|
||||
HISTORY_BACKUP_ARCHIVE=""
|
||||
DEMO_HISTORY_PATHS=(
|
||||
ai_incidents.json
|
||||
ai_incidents.json.tmp
|
||||
alerts/alert-history.json
|
||||
alerts/alert-history.backup.json
|
||||
alerts/alert-history.json.imported
|
||||
alerts/alert-history.backup.json.imported
|
||||
alerts/events.db
|
||||
alerts/events.db-shm
|
||||
alerts/events.db-wal
|
||||
)
|
||||
|
||||
for fixture_count in \
|
||||
"$MOCK_NODES" \
|
||||
"$MOCK_VMS_PER_NODE" \
|
||||
"$MOCK_LXCS_PER_NODE" \
|
||||
"$MOCK_DOCKER_HOSTS" \
|
||||
"$MOCK_DOCKER_CONTAINERS" \
|
||||
"$MOCK_GENERIC_HOSTS" \
|
||||
"$MOCK_K8S_CLUSTERS" \
|
||||
"$MOCK_K8S_NODES" \
|
||||
"$MOCK_K8S_PODS" \
|
||||
"$MOCK_K8S_DEPLOYMENTS"; do
|
||||
[[ "$fixture_count" =~ ^[1-9][0-9]*$ ]]
|
||||
done
|
||||
[[ "$MOCK_SEED_DURATION" =~ ^[1-9][0-9]*[smhd]$ ]]
|
||||
[[ "$MOCK_SAMPLE_INTERVAL" =~ ^[1-9][0-9]*[smh]$ ]]
|
||||
[[ "$MOCK_UPDATE_INTERVAL" =~ ^[1-9][0-9]*[smh]$ ]]
|
||||
|
||||
log() {
|
||||
printf '[demo-recovery] %s\n' "$*" >&2
|
||||
}
|
||||
|
||||
unit_hash() {
|
||||
sudo sha256sum "$EXPECTED_UNIT" | awk '{print $1}'
|
||||
}
|
||||
|
||||
binary_hash() {
|
||||
sudo sha256sum "$EXPECTED_BINARY" | awk '{print $1}'
|
||||
}
|
||||
|
||||
dropin_manifest_hash() {
|
||||
local paths path
|
||||
paths="$(systemctl show "$SERVICE_NAME" --property=DropInPaths --value)"
|
||||
if [ -z "$paths" ]; then
|
||||
printf '%s' 'none' | sha256sum | awk '{print $1}'
|
||||
return
|
||||
fi
|
||||
for path in $paths; do
|
||||
sudo test -f "$path"
|
||||
printf '%s\t' "$path"
|
||||
sudo sha256sum "$path" | awk '{print $1}'
|
||||
done | sort | sha256sum | awk '{print $1}'
|
||||
}
|
||||
|
||||
runtime_config_hash() {
|
||||
local path
|
||||
for path in /etc/pulse/.env /etc/pulse/billing.json; do
|
||||
sudo test -f "$path"
|
||||
printf '%s\t' "$path"
|
||||
sudo sha256sum "$path" | awk '{print $1}'
|
||||
done | sha256sum | awk '{print $1}'
|
||||
}
|
||||
|
||||
set_env_value() {
|
||||
local key="$1"
|
||||
local value="$2"
|
||||
local env_file="/etc/pulse/.env"
|
||||
if sudo grep -Eq "^[[:space:]]*${key}=" "$env_file"; then
|
||||
sudo sed -i "s|^[[:space:]]*${key}=.*|${key}=${value}|" "$env_file"
|
||||
else
|
||||
printf '\n%s=%s\n' "$key" "$value" | sudo tee -a "$env_file" >/dev/null
|
||||
fi
|
||||
}
|
||||
|
||||
runtime_profile_matches() {
|
||||
sudo grep -Fxq "PULSE_MOCK_NODES=${MOCK_NODES}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_VMS_PER_NODE=${MOCK_VMS_PER_NODE}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_LXCS_PER_NODE=${MOCK_LXCS_PER_NODE}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_DOCKER_HOSTS=${MOCK_DOCKER_HOSTS}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_DOCKER_CONTAINERS=${MOCK_DOCKER_CONTAINERS}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_GENERIC_HOSTS=${MOCK_GENERIC_HOSTS}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_K8S_CLUSTERS=${MOCK_K8S_CLUSTERS}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_K8S_NODES=${MOCK_K8S_NODES}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_K8S_PODS=${MOCK_K8S_PODS}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_K8S_DEPLOYMENTS=${MOCK_K8S_DEPLOYMENTS}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_SEED_METRICS_STORE=false" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_TRENDS_SEED_DURATION=${MOCK_SEED_DURATION}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_TRENDS_SAMPLE_INTERVAL=${MOCK_SAMPLE_INTERVAL}" /etc/pulse/.env &&
|
||||
sudo grep -Fxq "PULSE_MOCK_UPDATE_INTERVAL=${MOCK_UPDATE_INTERVAL}" /etc/pulse/.env
|
||||
}
|
||||
|
||||
cleanup_config_backup() {
|
||||
[ -n "$BACKUP_DIR" ] || return 0
|
||||
sudo unlink "$BACKUP_DIR/runtime.env" 2>/dev/null || true
|
||||
sudo unlink "$BACKUP_DIR/billing.json" 2>/dev/null || true
|
||||
if [ -n "$HISTORY_BACKUP_ARCHIVE" ]; then
|
||||
sudo unlink "$HISTORY_BACKUP_ARCHIVE" 2>/dev/null || true
|
||||
HISTORY_BACKUP_ARCHIVE=""
|
||||
fi
|
||||
rmdir "$BACKUP_DIR" 2>/dev/null || true
|
||||
BACKUP_DIR=""
|
||||
}
|
||||
|
||||
clear_demo_operational_history() {
|
||||
local rel_path
|
||||
local -a existing=()
|
||||
for rel_path in "${DEMO_HISTORY_PATHS[@]}"; do
|
||||
if sudo test -f "/etc/pulse/${rel_path}"; then
|
||||
existing+=("$rel_path")
|
||||
fi
|
||||
done
|
||||
((${#existing[@]} > 0)) || return 0
|
||||
|
||||
HISTORY_BACKUP_ARCHIVE="$BACKUP_DIR/demo-operational-history.tar"
|
||||
sudo tar --create --preserve-permissions \
|
||||
--file "$HISTORY_BACKUP_ARCHIVE" --directory /etc/pulse -- "${existing[@]}"
|
||||
for rel_path in "${existing[@]}"; do
|
||||
sudo unlink "/etc/pulse/${rel_path}"
|
||||
done
|
||||
}
|
||||
|
||||
restore_demo_operational_history() {
|
||||
local rel_path
|
||||
[ -n "$HISTORY_BACKUP_ARCHIVE" ] || return 0
|
||||
sudo test -f "$HISTORY_BACKUP_ARCHIVE" || return 0
|
||||
for rel_path in "${DEMO_HISTORY_PATHS[@]}"; do
|
||||
sudo unlink "/etc/pulse/${rel_path}" 2>/dev/null || true
|
||||
done
|
||||
sudo tar --extract --preserve-permissions \
|
||||
--file "$HISTORY_BACKUP_ARCHIVE" --directory /etc/pulse
|
||||
}
|
||||
|
||||
restore_runtime_config() {
|
||||
[ -n "$BACKUP_DIR" ] || return 0
|
||||
sudo cp --archive "$BACKUP_DIR/runtime.env" /etc/pulse/.env
|
||||
sudo cp --archive "$BACKUP_DIR/billing.json" /etc/pulse/billing.json
|
||||
}
|
||||
|
||||
local_version() {
|
||||
curl --connect-timeout 3 --max-time 8 -fsS \
|
||||
"${LOCAL_BASE_URL}/api/version" | jq -er '.version'
|
||||
}
|
||||
|
||||
local_healthy() {
|
||||
curl --connect-timeout 3 --max-time 8 -fsS \
|
||||
"${LOCAL_BASE_URL}/api/health" >/dev/null && local_version >/dev/null
|
||||
}
|
||||
|
||||
emit_evidence() {
|
||||
local status="$1"
|
||||
local before_pid="$2"
|
||||
local after_pid="$3"
|
||||
local before_relay_pid="$4"
|
||||
local after_relay_pid="$5"
|
||||
local before_binary_sha="$6"
|
||||
local after_binary_sha="$7"
|
||||
local before_unit_sha="$8"
|
||||
local after_unit_sha="$9"
|
||||
local before_dropins_sha="${10}"
|
||||
local after_dropins_sha="${11}"
|
||||
local version="${12}"
|
||||
local before_config_sha="${13}"
|
||||
local after_config_sha="${14}"
|
||||
jq -n \
|
||||
--arg status "$status" \
|
||||
--argjson mutated "$MUTATED" \
|
||||
--arg hostname "$(hostname)" \
|
||||
--arg service "$SERVICE_NAME" \
|
||||
--arg before_pid "$before_pid" \
|
||||
--arg after_pid "$after_pid" \
|
||||
--arg relay_service "$RELAY_SERVICE_NAME" \
|
||||
--arg before_relay_pid "$before_relay_pid" \
|
||||
--arg after_relay_pid "$after_relay_pid" \
|
||||
--arg binary "$EXPECTED_BINARY" \
|
||||
--arg before_binary_sha256 "$before_binary_sha" \
|
||||
--arg after_binary_sha256 "$after_binary_sha" \
|
||||
--arg unit "$EXPECTED_UNIT" \
|
||||
--arg before_unit_sha256 "$before_unit_sha" \
|
||||
--arg after_unit_sha256 "$after_unit_sha" \
|
||||
--arg before_dropins_sha256 "$before_dropins_sha" \
|
||||
--arg after_dropins_sha256 "$after_dropins_sha" \
|
||||
--arg before_runtime_config_sha256 "$before_config_sha" \
|
||||
--arg after_runtime_config_sha256 "$after_config_sha" \
|
||||
--arg version "$version" \
|
||||
--arg expected_version "$EXPECTED_VERSION" \
|
||||
--arg mock_nodes "$MOCK_NODES" \
|
||||
--arg mock_vms_per_node "$MOCK_VMS_PER_NODE" \
|
||||
--arg mock_lxcs_per_node "$MOCK_LXCS_PER_NODE" \
|
||||
--arg mock_docker_hosts "$MOCK_DOCKER_HOSTS" \
|
||||
--arg mock_docker_containers "$MOCK_DOCKER_CONTAINERS" \
|
||||
--arg mock_generic_hosts "$MOCK_GENERIC_HOSTS" \
|
||||
--arg mock_k8s_clusters "$MOCK_K8S_CLUSTERS" \
|
||||
--arg mock_k8s_nodes "$MOCK_K8S_NODES" \
|
||||
--arg mock_k8s_pods "$MOCK_K8S_PODS" \
|
||||
--arg mock_k8s_deployments "$MOCK_K8S_DEPLOYMENTS" \
|
||||
--arg mock_seed_duration "$MOCK_SEED_DURATION" \
|
||||
--arg mock_sample_interval "$MOCK_SAMPLE_INTERVAL" \
|
||||
--arg mock_update_interval "$MOCK_UPDATE_INTERVAL" \
|
||||
'{schema_version: 1, status: $status, mutated: $mutated,
|
||||
hostname: $hostname, service: $service,
|
||||
before_pid: $before_pid, after_pid: $after_pid,
|
||||
relay_service: $relay_service,
|
||||
before_relay_pid: $before_relay_pid,
|
||||
after_relay_pid: $after_relay_pid,
|
||||
binary: $binary,
|
||||
before_binary_sha256: $before_binary_sha256,
|
||||
after_binary_sha256: $after_binary_sha256,
|
||||
unit: $unit,
|
||||
before_unit_sha256: $before_unit_sha256,
|
||||
after_unit_sha256: $after_unit_sha256,
|
||||
before_dropins_sha256: $before_dropins_sha256,
|
||||
after_dropins_sha256: $after_dropins_sha256,
|
||||
before_runtime_config_sha256: $before_runtime_config_sha256,
|
||||
after_runtime_config_sha256: $after_runtime_config_sha256,
|
||||
version: $version, expected_version: $expected_version,
|
||||
runtime_profile: {mock_nodes: $mock_nodes,
|
||||
mock_vms_per_node: $mock_vms_per_node,
|
||||
mock_lxcs_per_node: $mock_lxcs_per_node,
|
||||
mock_docker_hosts: $mock_docker_hosts,
|
||||
mock_docker_containers: $mock_docker_containers,
|
||||
mock_generic_hosts: $mock_generic_hosts,
|
||||
mock_k8s_clusters: $mock_k8s_clusters,
|
||||
mock_k8s_nodes: $mock_k8s_nodes,
|
||||
mock_k8s_pods: $mock_k8s_pods,
|
||||
mock_k8s_deployments: $mock_k8s_deployments,
|
||||
mock_seed_duration: $mock_seed_duration,
|
||||
mock_sample_interval: $mock_sample_interval,
|
||||
mock_update_interval: $mock_update_interval}}'
|
||||
}
|
||||
|
||||
[ "$(hostname)" = "$EXPECTED_HOSTNAME" ] || {
|
||||
log "target hostname does not match the governed demo environment"
|
||||
exit 1
|
||||
}
|
||||
[ "$(systemctl show "$SERVICE_NAME" --property=FragmentPath --value)" = "$EXPECTED_UNIT" ]
|
||||
[ "$(systemctl show "$SERVICE_NAME" --property=User --value)" = "pulse" ]
|
||||
[ "$(systemctl show "$SERVICE_NAME" --property=Group --value)" = "pulse" ]
|
||||
[ "$(systemctl is-enabled "$SERVICE_NAME")" = "enabled" ]
|
||||
[ "$(systemctl is-active "$RELAY_SERVICE_NAME")" = "active" ]
|
||||
systemctl show "$SERVICE_NAME" --property=ExecStart --value | grep -Fq "$EXPECTED_BINARY"
|
||||
sudo test -x "$EXPECTED_BINARY"
|
||||
sudo test -f "$EXPECTED_UNIT"
|
||||
|
||||
BEFORE_PID="$(systemctl show "$SERVICE_NAME" --property=MainPID --value)"
|
||||
BEFORE_RELAY_PID="$(systemctl show "$RELAY_SERVICE_NAME" --property=MainPID --value)"
|
||||
BEFORE_BINARY_SHA="$(binary_hash)"
|
||||
BEFORE_UNIT_SHA="$(unit_hash)"
|
||||
BEFORE_DROPINS_SHA="$(dropin_manifest_hash)"
|
||||
BEFORE_CONFIG_SHA="$(runtime_config_hash)"
|
||||
|
||||
if local_healthy && runtime_profile_matches; then
|
||||
VERSION="$(local_version)"
|
||||
[ "${VERSION#v}" = "$EXPECTED_VERSION" ]
|
||||
emit_evidence healthy_noop "$BEFORE_PID" "$BEFORE_PID" \
|
||||
"$BEFORE_RELAY_PID" "$BEFORE_RELAY_PID" \
|
||||
"$BEFORE_BINARY_SHA" "$BEFORE_BINARY_SHA" \
|
||||
"$BEFORE_UNIT_SHA" "$BEFORE_UNIT_SHA" \
|
||||
"$BEFORE_DROPINS_SHA" "$BEFORE_DROPINS_SHA" "$VERSION" \
|
||||
"$BEFORE_CONFIG_SHA" "$BEFORE_CONFIG_SHA"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
rollback_on_error() {
|
||||
local rc=$?
|
||||
local failed_pid
|
||||
trap - ERR
|
||||
if [ "$MUTATED" = true ]; then
|
||||
failed_pid="$(systemctl show "$SERVICE_NAME" --property=MainPID --value || true)"
|
||||
if [[ "$failed_pid" =~ ^[1-9][0-9]*$ ]] && sudo kill -0 "$failed_pid" 2>/dev/null; then
|
||||
# The HTTP listener never opened, so pprof is unavailable. SIGQUIT gives
|
||||
# one bounded Go goroutine dump in the service journal before the already
|
||||
# failed process is stopped; it does not widen recovery to any other unit.
|
||||
log "capturing failed Pulse startup goroutines before compensation"
|
||||
sudo kill -QUIT "$failed_pid" || true
|
||||
sleep 2
|
||||
fi
|
||||
log "recovery validation failed; compensating by stopping only pulse.service"
|
||||
sudo systemctl stop "$SERVICE_NAME" || true
|
||||
restore_runtime_config || true
|
||||
restore_demo_operational_history || true
|
||||
fi
|
||||
emit_evidence compensated_unavailable "$BEFORE_PID" \
|
||||
"$(systemctl show "$SERVICE_NAME" --property=MainPID --value || true)" \
|
||||
"$BEFORE_RELAY_PID" \
|
||||
"$(systemctl show "$RELAY_SERVICE_NAME" --property=MainPID --value || true)" \
|
||||
"$BEFORE_BINARY_SHA" "$(binary_hash || true)" \
|
||||
"$BEFORE_UNIT_SHA" "$(unit_hash || true)" \
|
||||
"$BEFORE_DROPINS_SHA" "$(dropin_manifest_hash || true)" "" \
|
||||
"$BEFORE_CONFIG_SHA" "$(runtime_config_hash || true)"
|
||||
cleanup_config_backup
|
||||
exit "$rc"
|
||||
}
|
||||
BACKUP_DIR="$(mktemp -d)"
|
||||
sudo cp --archive /etc/pulse/.env "$BACKUP_DIR/runtime.env"
|
||||
sudo cp --archive /etc/pulse/billing.json "$BACKUP_DIR/billing.json"
|
||||
trap rollback_on_error ERR
|
||||
|
||||
log "applying the target-compatible demo runtime profile and restarting only pulse.service"
|
||||
MUTATED=true
|
||||
set_env_value PULSE_MOCK_NODES "$MOCK_NODES"
|
||||
set_env_value PULSE_MOCK_VMS_PER_NODE "$MOCK_VMS_PER_NODE"
|
||||
set_env_value PULSE_MOCK_LXCS_PER_NODE "$MOCK_LXCS_PER_NODE"
|
||||
set_env_value PULSE_MOCK_DOCKER_HOSTS "$MOCK_DOCKER_HOSTS"
|
||||
set_env_value PULSE_MOCK_DOCKER_CONTAINERS "$MOCK_DOCKER_CONTAINERS"
|
||||
set_env_value PULSE_MOCK_GENERIC_HOSTS "$MOCK_GENERIC_HOSTS"
|
||||
set_env_value PULSE_MOCK_K8S_CLUSTERS "$MOCK_K8S_CLUSTERS"
|
||||
set_env_value PULSE_MOCK_K8S_NODES "$MOCK_K8S_NODES"
|
||||
set_env_value PULSE_MOCK_K8S_PODS "$MOCK_K8S_PODS"
|
||||
set_env_value PULSE_MOCK_K8S_DEPLOYMENTS "$MOCK_K8S_DEPLOYMENTS"
|
||||
# Persistent mock-store backfill is reserved for the isolated local-dev data
|
||||
# directory. The governed demo already has persistent runtime data, so a stale
|
||||
# opt-in here can turn startup into an unbounded SQLite backfill.
|
||||
set_env_value PULSE_MOCK_SEED_METRICS_STORE false
|
||||
set_env_value PULSE_MOCK_TRENDS_SEED_DURATION "$MOCK_SEED_DURATION"
|
||||
set_env_value PULSE_MOCK_TRENDS_SAMPLE_INTERVAL "$MOCK_SAMPLE_INTERVAL"
|
||||
set_env_value PULSE_MOCK_UPDATE_INTERVAL "$MOCK_UPDATE_INTERVAL"
|
||||
sudo chown pulse:pulse /etc/pulse/.env
|
||||
sudo chmod 600 /etc/pulse/.env
|
||||
# Alert lifecycle and AI incident history on this host are generated demo
|
||||
# operational memory. A failed 50-node bootstrap can leave thousands of
|
||||
# lifecycle events whose v6.4 replay queues one serialized incident save per
|
||||
# event before HTTP opens. Back up and clear only those fixed demo-history
|
||||
# files, including the retired JSON sources that repopulate a cleared event
|
||||
# log; compensation restores them if the bounded profile still fails.
|
||||
clear_demo_operational_history
|
||||
sudo systemctl restart "$SERVICE_NAME"
|
||||
|
||||
for _ in $(seq 1 12); do
|
||||
if [ "$(systemctl is-active "$SERVICE_NAME" || true)" = "active" ] && local_healthy; then
|
||||
break
|
||||
fi
|
||||
sleep 5
|
||||
done
|
||||
|
||||
[ "$(systemctl is-active "$SERVICE_NAME")" = "active" ]
|
||||
[ "$(systemctl show "$SERVICE_NAME" --property=SubState --value)" = "running" ]
|
||||
local_healthy
|
||||
runtime_profile_matches
|
||||
|
||||
AFTER_PID="$(systemctl show "$SERVICE_NAME" --property=MainPID --value)"
|
||||
AFTER_RELAY_PID="$(systemctl show "$RELAY_SERVICE_NAME" --property=MainPID --value)"
|
||||
AFTER_BINARY_SHA="$(binary_hash)"
|
||||
AFTER_UNIT_SHA="$(unit_hash)"
|
||||
AFTER_DROPINS_SHA="$(dropin_manifest_hash)"
|
||||
AFTER_CONFIG_SHA="$(runtime_config_hash)"
|
||||
VERSION="$(local_version)"
|
||||
|
||||
[ "$AFTER_PID" -gt 0 ]
|
||||
[ "$AFTER_PID" != "$BEFORE_PID" ]
|
||||
[ "$AFTER_RELAY_PID" = "$BEFORE_RELAY_PID" ]
|
||||
[ "$AFTER_BINARY_SHA" = "$BEFORE_BINARY_SHA" ]
|
||||
[ "$AFTER_UNIT_SHA" = "$BEFORE_UNIT_SHA" ]
|
||||
[ "$AFTER_DROPINS_SHA" = "$BEFORE_DROPINS_SHA" ]
|
||||
[ "${VERSION#v}" = "$EXPECTED_VERSION" ]
|
||||
|
||||
trap - ERR
|
||||
cleanup_config_backup
|
||||
emit_evidence recovered "$BEFORE_PID" "$AFTER_PID" \
|
||||
"$BEFORE_RELAY_PID" "$AFTER_RELAY_PID" \
|
||||
"$BEFORE_BINARY_SHA" "$AFTER_BINARY_SHA" \
|
||||
"$BEFORE_UNIT_SHA" "$AFTER_UNIT_SHA" \
|
||||
"$BEFORE_DROPINS_SHA" "$AFTER_DROPINS_SHA" "$VERSION" \
|
||||
"$BEFORE_CONFIG_SHA" "$AFTER_CONFIG_SHA"
|
||||
# CI observer only; remote systemd owns mutation, restoration and both watches.
|
||||
exec python3 .github/scripts/dispatch-demo-runtime.py recover
|
||||
|
||||
@@ -0,0 +1,443 @@
|
||||
import copy
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import signal
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
SCRIPTS = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def load(name, filename):
|
||||
spec = importlib.util.spec_from_file_location(name, SCRIPTS / filename)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
engine = load("demo_transaction", "demo-runtime-transaction.py")
|
||||
dispatcher = load("demo_dispatcher", "dispatch-demo-runtime.py")
|
||||
|
||||
|
||||
def request(mode="recover"):
|
||||
return {"mode": mode, "hostname": "fixture", "local_url": "http://127.0.0.1:7655",
|
||||
"public_url": "https://demo.example/api/health", "version": "v6.4.5",
|
||||
"control_sha": "a" * 40, "run_id": "123", "run_attempt": "1",
|
||||
"installer_sha256": hashlib.sha256(b"signed-installer-fixture").hexdigest() if mode == "update" else "",
|
||||
"profile": {**{name: "2" for name in engine.COUNT_KEYS},
|
||||
"seed_duration": "2h", "sample_interval": "5m", "update_interval": "15s"}}
|
||||
|
||||
|
||||
class FixtureHost(engine.Host):
|
||||
"""Real filesystem estate with a virtual *unchanged* 300-second clock.
|
||||
|
||||
Commands/services/journal/HTTP are adapters, not native or installed proof.
|
||||
"""
|
||||
def __init__(self):
|
||||
self.clock = 0
|
||||
self.pid = 100
|
||||
self.running = True
|
||||
self.runtime_version = "6.4.5"
|
||||
self.starts = 0
|
||||
self.stops = 0
|
||||
self.install_count = 0
|
||||
self.fail_at = None
|
||||
self.failed = False
|
||||
self.rollback_failure = False
|
||||
self.unhealthy = False
|
||||
self.journal_failure = False
|
||||
self.ready_until = 0
|
||||
self.cancel_at = None
|
||||
self.cancel_in_recovery = False
|
||||
self.co_host_change = False
|
||||
self.co_pid = 200
|
||||
self.transaction = None
|
||||
|
||||
def now(self):
|
||||
return self.clock
|
||||
|
||||
def sleep(self, seconds):
|
||||
self.clock += seconds
|
||||
|
||||
def identity(self, hostname):
|
||||
if hostname != "fixture":
|
||||
raise engine.Failure("host-identity")
|
||||
|
||||
def state(self):
|
||||
state = {"ActiveState": "active", "SubState": "running", "MainPID": str(self.pid), "NRestarts": "0"}
|
||||
result = {"pulse": state, "pulse-relay": {**state, "MainPID": "200"}, "caddy": {**state, "MainPID": str(self.co_pid)}}
|
||||
if not self.running:
|
||||
result["pulse"] = {**state, "ActiveState": "inactive", "SubState": "dead", "MainPID": "0"}
|
||||
return result
|
||||
|
||||
def status(self, url):
|
||||
if url == engine.RELAY_HEALTH:
|
||||
return "200"
|
||||
if self.clock < self.ready_until or self.unhealthy:
|
||||
return "503"
|
||||
if self.failed and self.rollback_failure:
|
||||
return "503"
|
||||
return "200" if self.running else "000"
|
||||
|
||||
def version(self, url):
|
||||
return self.runtime_version
|
||||
|
||||
def cursors(self):
|
||||
return {name: str(self.clock) for name in engine.SERVICES}
|
||||
|
||||
def check_journal(self, cursors):
|
||||
phase = self.transaction.receipt["status"] if self.transaction else ""
|
||||
if self.journal_failure:
|
||||
raise engine.Failure("journal-observation")
|
||||
if self.co_host_change and self.clock >= 55:
|
||||
self.co_pid = 201
|
||||
if self.cancel_at is not None and self.clock >= self.cancel_at and not self.failed:
|
||||
self.failed = True
|
||||
handler = signal.getsignal(signal.SIGTERM)
|
||||
handler(signal.SIGTERM, None)
|
||||
if self.cancel_in_recovery and phase == "recovering" and self.clock >= 350:
|
||||
# Exercise the actual installed handler decision; do not signal the
|
||||
# source-proof executor or kill another process to prove it.
|
||||
assert signal.getsignal(signal.SIGTERM) == signal.SIG_IGN
|
||||
self.cancel_in_recovery = False
|
||||
if self.fail_at is not None and self.clock >= self.fail_at and not self.failed:
|
||||
self.failed = True
|
||||
raise engine.Failure("new-service-crash")
|
||||
|
||||
def stop(self):
|
||||
self.stops += 1
|
||||
self.running = False
|
||||
|
||||
def start(self):
|
||||
self.starts += 1
|
||||
self.pid += 1
|
||||
self.running = True
|
||||
if self.transaction.receipt["status"] == "recovering":
|
||||
self.runtime_version = "6.4.5"
|
||||
self.unhealthy = False
|
||||
|
||||
def install(self, attempt, req):
|
||||
if hashlib.sha256((attempt / "installer.sh").read_bytes()).hexdigest() != req["installer_sha256"]:
|
||||
raise engine.Failure("installer-identity")
|
||||
self.install_count += 1
|
||||
self.runtime_version = req["version"].lstrip("v")
|
||||
engine.PATHS["binary"].write_bytes(b"new executable")
|
||||
engine.PATHS["unit"].write_bytes(b"new unit")
|
||||
(engine.PATHS["data"] / "persisted-history").write_bytes(b"forward history")
|
||||
|
||||
def profile(self, req, unhealthy):
|
||||
super().profile(req, unhealthy)
|
||||
if self.unhealthy:
|
||||
self.unhealthy = False
|
||||
|
||||
|
||||
class TransactionTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
root = Path(self.tmp.name)
|
||||
self.paths = {name: root / "estate" / name for name in engine.PATHS}
|
||||
for name, path in self.paths.items():
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if name in {"dropins", "data"}:
|
||||
path.mkdir()
|
||||
else:
|
||||
path.write_bytes(b"old " + name.encode())
|
||||
path.chmod(0o755 if name == "binary" else 0o644)
|
||||
(self.paths["dropins"] / "limits.conf").write_bytes(b"old limits")
|
||||
(self.paths["data"] / ".env").write_text("PRIVATE_TOKEN=not-logged\n")
|
||||
(self.paths["data"] / "billing.json").write_text('{"capabilities":[],"integrity":"private"}')
|
||||
(self.paths["data"] / "persisted-history").write_bytes(b"original history")
|
||||
(self.paths["data"] / "alerts").mkdir()
|
||||
(self.paths["data"] / "alerts/events.db").write_bytes(b"synthetic opaque persistent file")
|
||||
self.root = root / "control"
|
||||
self.attempt = self.root / "attempts" / ("a" * 64)
|
||||
self.attempt.mkdir(parents=True)
|
||||
(self.attempt / "installer.sh").write_bytes(b"signed-installer-fixture")
|
||||
self.addCleanup(patch.stopall)
|
||||
patch.object(engine, "ROOT", self.root).start()
|
||||
patch.object(engine, "PATHS", self.paths).start()
|
||||
self.before = engine.estate_hash(self.paths)
|
||||
self.host = FixtureHost()
|
||||
|
||||
def run_transaction(self, mode="recover", version=None):
|
||||
req = request(mode)
|
||||
if version:
|
||||
req["version"] = version
|
||||
tx = engine.Transaction(self.host, req, self.attempt)
|
||||
self.host.transaction = tx
|
||||
result = tx.run()
|
||||
self.assertEqual(tx.receipt, json.loads((self.attempt / "receipt.json").read_text()))
|
||||
return result, tx.receipt
|
||||
|
||||
def test_healthy_profile_change_keeps_data_and_watches_complete_window(self):
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(code, 0)
|
||||
self.assertEqual(receipt["status"], "committed")
|
||||
self.assertEqual(receipt["forward"]["elapsed_seconds"], 300)
|
||||
self.assertEqual(receipt["forward"]["samples"], 61)
|
||||
self.assertEqual((self.paths["data"] / "alerts/events.db").read_bytes(), b"synthetic opaque persistent file")
|
||||
self.assertTrue((self.attempt / "snapshot/data/persisted-history").exists())
|
||||
self.assertNotIn("not-logged", json.dumps(receipt))
|
||||
|
||||
def test_healthy_noop_requires_profile_and_full_watch_without_restart(self):
|
||||
self.host.profile(request(), False)
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(code, 0)
|
||||
self.assertEqual(receipt["status"], "healthy_noop")
|
||||
self.assertEqual(receipt["forward"]["elapsed_seconds"], 300)
|
||||
self.assertFalse(receipt["mutated"])
|
||||
self.assertEqual(self.host.stops, 0)
|
||||
|
||||
def test_update_restores_complete_changed_executable_unit_and_data_after_late_failure(self):
|
||||
self.host.fail_at = 55
|
||||
code, receipt = self.run_transaction("update", "v6.4.6")
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["status"], "rolled_back")
|
||||
self.assertEqual(receipt["failure"], "new-service-crash")
|
||||
self.assertEqual(receipt["recovery"]["elapsed_seconds"], 300)
|
||||
self.assertEqual(engine.estate_hash(self.paths), self.before)
|
||||
self.assertEqual(self.host.runtime_version, "6.4.5")
|
||||
|
||||
def test_failure_at_299_seconds_still_collects_entire_recovery_and_cancellation(self):
|
||||
self.host.fail_at = 299
|
||||
self.host.cancel_in_recovery = True
|
||||
previous = signal.getsignal(signal.SIGTERM)
|
||||
code, receipt = self.run_transaction("update", "v6.4.6")
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["recovery"]["elapsed_seconds"], 300)
|
||||
self.assertGreaterEqual(self.host.clock, 600)
|
||||
self.assertFalse(self.host.cancel_in_recovery)
|
||||
self.assertEqual(signal.getsignal(signal.SIGTERM), previous)
|
||||
self.assertEqual(engine.estate_hash(self.paths), self.before)
|
||||
|
||||
def test_forward_cancel_owns_full_recovery(self):
|
||||
self.host.cancel_at = 55
|
||||
code, receipt = self.run_transaction("update", "v6.4.6")
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["failure"], "cancelled-forward")
|
||||
self.assertEqual(receipt["status"], "rolled_back")
|
||||
self.assertEqual(receipt["recovery"]["elapsed_seconds"], 300)
|
||||
|
||||
def test_failed_rollback_retains_estate_and_blocks_next_transaction(self):
|
||||
self.host.fail_at = 55
|
||||
self.host.rollback_failure = True
|
||||
code, receipt = self.run_transaction("update", "v6.4.6")
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["status"], "rollback_failed")
|
||||
self.assertTrue(receipt["snapshot_retained"])
|
||||
next_attempt = self.root / "attempts" / ("b" * 64)
|
||||
next_attempt.mkdir()
|
||||
self.attempt = next_attempt
|
||||
self.host.rollback_failure = False
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(receipt["failure"], "prior-recovery-required")
|
||||
self.assertFalse(receipt["mutated"])
|
||||
|
||||
def test_unhealthy_recovery_retains_original_generated_history(self):
|
||||
self.host.unhealthy = True
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(code, 0)
|
||||
self.assertFalse(receipt["healthy_baseline"])
|
||||
self.assertFalse((self.paths["data"] / "alerts/events.db").exists())
|
||||
self.assertEqual((self.attempt / "snapshot/data/alerts/events.db").read_bytes(), b"synthetic opaque persistent file")
|
||||
self.assertEqual(receipt["forward"]["elapsed_seconds"], 300)
|
||||
|
||||
def test_failed_unhealthy_recovery_is_not_a_healthy_rollback(self):
|
||||
self.host.unhealthy = True
|
||||
self.host.fail_at = 55
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["status"], "recovery_required")
|
||||
self.assertEqual(receipt["rollback"], "unavailable-unhealthy-baseline")
|
||||
self.assertEqual(engine.estate_hash(self.paths), self.before)
|
||||
self.assertFalse(self.host.running)
|
||||
|
||||
def test_update_refuses_unhealthy_baseline_without_stopping_or_pruning(self):
|
||||
self.host.unhealthy = True
|
||||
code, receipt = self.run_transaction("update", "v6.4.6")
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["failure"], "update-baseline-unhealthy")
|
||||
self.assertEqual(self.host.stops, 0)
|
||||
self.assertEqual(engine.estate_hash(self.paths), self.before)
|
||||
|
||||
def test_recovery_refuses_different_healthy_source(self):
|
||||
code, receipt = self.run_transaction(version="v6.4.6")
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["failure"], "recovery-source-mismatch")
|
||||
self.assertEqual(self.host.stops, 0)
|
||||
|
||||
def test_listener_readiness_does_not_spend_the_watch_window(self):
|
||||
self.host.ready_until = 30
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(code, 0)
|
||||
self.assertEqual(receipt["forward"]["elapsed_seconds"], 300)
|
||||
self.assertGreaterEqual(self.host.clock, 330)
|
||||
self.assertEqual(self.host.starts, 1)
|
||||
|
||||
def test_missing_journal_fails_closed_and_retains_failure(self):
|
||||
self.host.journal_failure = True
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["status"], "rollback_failed")
|
||||
self.assertEqual(receipt["recovery_failure"], "journal-observation")
|
||||
|
||||
def test_cohost_restart_is_not_silently_accepted_as_recovery(self):
|
||||
self.host.co_host_change = True
|
||||
code, receipt = self.run_transaction("update", "v6.4.6")
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["status"], "rollback_failed")
|
||||
self.assertEqual(receipt["recovery_failure"], "cohost-identity-changed")
|
||||
|
||||
def test_snapshot_headroom_failure_never_deletes_persistent_files(self):
|
||||
with patch.object(engine.shutil, "disk_usage", return_value=type("Usage", (), {"free": 0})()):
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["status"], "rolled_back")
|
||||
self.assertFalse(receipt["mutated"])
|
||||
self.assertEqual(receipt["failure"], "snapshot-headroom")
|
||||
self.assertEqual(engine.estate_hash(self.paths), self.before)
|
||||
|
||||
def test_linked_history_cannot_remove_files_outside_estate(self):
|
||||
(self.paths["data"] / "alerts/events.db").unlink()
|
||||
outside = self.root / "private-file"
|
||||
outside.write_text("keep")
|
||||
(self.paths["data"] / "alerts/events.db").symlink_to(outside)
|
||||
self.host.unhealthy = True
|
||||
code, receipt = self.run_transaction()
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(receipt["failure"], "history-symlink")
|
||||
self.assertEqual(outside.read_text(), "keep")
|
||||
|
||||
def test_shared_host_lock_refusal_is_terminal_and_has_no_mutation(self):
|
||||
lock = self.root / "host.lock"
|
||||
fd = os.open(lock, os.O_CREAT | os.O_RDWR, 0o600)
|
||||
self.addCleanup(os.close, fd)
|
||||
engine.fcntl.flock(fd, engine.fcntl.LOCK_EX | engine.fcntl.LOCK_NB)
|
||||
(self.attempt / "request.json").write_text(json.dumps(request()))
|
||||
with patch.object(engine, "LOCK", lock), patch.object(engine.sys, "argv", ["transaction", str(self.attempt)]):
|
||||
self.assertEqual(engine.main(), 1)
|
||||
receipt = json.loads((self.attempt / "receipt.json").read_text())
|
||||
self.assertEqual(receipt["status"], "refused")
|
||||
self.assertFalse(receipt["mutated"])
|
||||
self.assertEqual(engine.estate_hash(self.paths), self.before)
|
||||
|
||||
|
||||
class InputAndCommandTest(unittest.TestCase):
|
||||
def test_valid_current_envelope(self):
|
||||
engine.validate(request())
|
||||
engine.validate(request("update"))
|
||||
|
||||
def test_no_candidate_prerelease_arbitrary_service_url_or_profile(self):
|
||||
cases = [("mode", "shell"), ("version", "v6.4.6-rc.1"), ("control_sha", "main"),
|
||||
("local_url", "http://other-host:7655"), ("public_url", "http://demo.example/api/health"),
|
||||
("public_url", "https://user:secret@demo.example/api/health"),
|
||||
("public_url", "https://demo.example/api/health?token=secret"),
|
||||
("installer_sha256", "arbitrary")]
|
||||
for name, value in cases:
|
||||
with self.subTest(name=name, value=value):
|
||||
req = request(); req[name] = value
|
||||
with self.assertRaises(engine.Failure):
|
||||
engine.validate(req)
|
||||
req = request(); req["service"] = "license"
|
||||
with self.assertRaises(engine.Failure):
|
||||
engine.validate(req)
|
||||
req = request(); req["profile"]["nodes"] = "2; restart caddy"
|
||||
with self.assertRaises(engine.Failure):
|
||||
engine.validate(req)
|
||||
|
||||
def test_http_observation_rejects_redirect_and_transport_failure(self):
|
||||
host = engine.Host()
|
||||
with patch.object(host, "command", return_value="302") as command:
|
||||
self.assertEqual(host.status("https://demo.example/api/health"), "302")
|
||||
self.assertNotIn("--location", command.call_args.args[0])
|
||||
self.assertIn("--disable", command.call_args.args[0])
|
||||
self.assertIn("/dev/null", command.call_args.args[0])
|
||||
with patch.object(host, "command", side_effect=engine.Failure("command-failed")):
|
||||
self.assertEqual(host.status("https://demo.example/api/health"), "000")
|
||||
|
||||
def test_malformed_or_unavailable_journal_is_not_no_crash(self):
|
||||
host = engine.Host()
|
||||
for text in ["not-json", '{"MESSAGE":"fine"}', '{"MESSAGE":[],"__CURSOR":"c"}']:
|
||||
with self.subTest(text=text), patch.object(host, "command", return_value=text):
|
||||
with self.assertRaises(engine.Failure):
|
||||
host.check_journal({"pulse": "c"})
|
||||
with patch.object(host, "command", return_value='{"MESSAGE":"panic: fixture","__CURSOR":"d"}'):
|
||||
with self.assertRaisesRegex(engine.Failure, "new-service-crash"):
|
||||
host.check_journal({"pulse": "c"})
|
||||
|
||||
def test_journal_cursor_advances_without_returning_private_message(self):
|
||||
host = engine.Host(); cursors = {"pulse": "c"}
|
||||
with patch.object(host, "command", return_value='{"MESSAGE":"private normal message","__CURSOR":"d"}'):
|
||||
host.check_journal(cursors)
|
||||
self.assertEqual(cursors, {"pulse": "d"})
|
||||
|
||||
def test_no_journal_cursor_and_malformed_service_state_fail_closed(self):
|
||||
host = engine.Host()
|
||||
with patch.object(host, "command", return_value=""):
|
||||
with self.assertRaises(engine.Failure):
|
||||
host.cursors()
|
||||
with self.assertRaises(engine.Failure):
|
||||
host.state()
|
||||
|
||||
|
||||
class DispatchLifecycleTest(unittest.TestCase):
|
||||
def test_observer_budget_covers_child_and_recovery_and_no_wait_owned_by_ssh(self):
|
||||
source = dispatcher.BOOTSTRAP
|
||||
self.assertIn("RuntimeMaxSec=45min", source)
|
||||
self.assertIn("TimeoutStopSec=20min", source)
|
||||
self.assertIn("KillMode=mixed", source)
|
||||
self.assertIn("+ 4000", source)
|
||||
self.assertGreater(4000, 45 * 60 + 20 * 60)
|
||||
self.assertNotIn("'--wait'", source)
|
||||
self.assertNotIn("systemctl', 'stop'", source)
|
||||
|
||||
def test_bootstrap_retains_intent_and_repeated_or_uncertain_launch_does_not_replay(self):
|
||||
# Execute the real bootstrap with only its filesystem/root/systemd/clock
|
||||
# adapters rebound, including a lost submission response. No host tools.
|
||||
for uncertain in (False, True):
|
||||
with self.subTest(uncertain=uncertain), tempfile.TemporaryDirectory() as directory:
|
||||
root = Path(directory) / "state"
|
||||
source = dispatcher.BOOTSTRAP.replace("pathlib.Path('/var/lib/pulse-deploy/demo')", repr(root))
|
||||
# repr(Path) is not a Python literal: bind just the fixed path.
|
||||
source = dispatcher.BOOTSTRAP.replace("'/var/lib/pulse-deploy/demo'", repr(str(root)))
|
||||
source = source.replace("root.stat().st_uid == 0", "root.stat().st_uid == os.getuid()")
|
||||
clock = [0]
|
||||
calls = []
|
||||
payload = {"request": request(), "source": "c291cmNl", "installer": ""}
|
||||
|
||||
def launch(argv, **kwargs):
|
||||
calls.append(argv)
|
||||
attempt = next((root / "attempts").iterdir())
|
||||
if uncertain:
|
||||
raise subprocess.TimeoutExpired("systemd-run", 30)
|
||||
(attempt / "receipt.json").write_text(json.dumps({"status": "committed"}))
|
||||
return subprocess.CompletedProcess(argv, 0)
|
||||
|
||||
class Input:
|
||||
def read(self, _limit):
|
||||
return json.dumps(payload)
|
||||
|
||||
with patch.object(os, "geteuid", return_value=0), patch.object(subprocess, "run", side_effect=launch), \
|
||||
patch.object(dispatcher.sys, "stdin", Input()), patch("time.monotonic", side_effect=lambda: clock[0]), \
|
||||
patch("time.sleep", side_effect=lambda seconds: clock.__setitem__(0, clock[0] + seconds)), patch("builtins.print"):
|
||||
for _ in range(2):
|
||||
with self.assertRaises(SystemExit):
|
||||
exec(compile(source, "fixed-bootstrap", "exec"), {})
|
||||
self.assertEqual(len(calls), 1)
|
||||
attempt = next((root / "attempts").iterdir())
|
||||
self.assertTrue((attempt / "intent.json").exists())
|
||||
self.assertEqual((attempt / "request.json").stat().st_mode & 0o777, 0o600)
|
||||
self.assertEqual(attempt.stat().st_mode & 0o777, 0o700)
|
||||
if uncertain:
|
||||
self.assertEqual(json.loads((attempt / "launch.json").read_text())["state"], "uncertain")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user