60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
"""List or remove stopped deployment rollback containers; never touches volumes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import shlex
|
|
from pathlib import Path
|
|
|
|
import paramiko
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
NAME = re.compile(r"^mini-openclaw-rollback-\d{14}$")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--apply", action="store_true")
|
|
args = parser.parse_args()
|
|
values = {}
|
|
for raw in (ROOT / "server-deploy.conf").read_text(encoding="utf-8-sig").splitlines():
|
|
line = raw.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
key, value = line.split("=", 1)
|
|
values[key.strip()] = value.strip()
|
|
client = paramiko.SSHClient()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
options = {"hostname": values["SERVER_HOST"], "port": int(values.get("SERVER_PORT", "22")), "username": values["SERVER_USER"], "timeout": 15}
|
|
options.update({"key_filename": values["SERVER_KEYFILE"]} if values.get("SERVER_KEYFILE") else {"password": values["SERVER_PASSWORD"]})
|
|
client.connect(**options)
|
|
_, stdout, _ = client.exec_command("docker ps -a --format '{{.Names}}|{{.State}}'", timeout=30)
|
|
lines = stdout.read().decode("utf-8", errors="replace").splitlines()
|
|
status = stdout.channel.recv_exit_status()
|
|
if status:
|
|
client.close()
|
|
raise RuntimeError("Unable to list containers")
|
|
targets = []
|
|
for line in lines:
|
|
name, _, state = line.partition("|")
|
|
if NAME.fullmatch(name) and state != "running":
|
|
targets.append(name)
|
|
print("rollback_targets=" + (",".join(targets) if targets else "none"))
|
|
if args.apply and targets:
|
|
command = "docker rm " + " ".join(shlex.quote(name) for name in targets)
|
|
_, stdout, stderr = client.exec_command(command, timeout=60)
|
|
output = stdout.read().decode("utf-8", errors="replace")
|
|
error = stderr.read().decode("utf-8", errors="replace")
|
|
code = stdout.channel.recv_exit_status()
|
|
if code:
|
|
client.close()
|
|
raise RuntimeError(error.strip() or "Unable to remove rollback containers")
|
|
print("removed=" + ",".join(output.splitlines()))
|
|
client.close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|