647 lines
40 KiB
Python
647 lines
40 KiB
Python
#!/usr/bin/env python3
|
|
"""Explicit, offline-capable pilot migration of an installer stack; Python 3.10+."""
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import shutil
|
|
import sys
|
|
import time
|
|
import urllib.parse
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
spec = importlib.util.spec_from_file_location("fedeo_update", HERE / "selfhost-update.py")
|
|
u = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(u)
|
|
Error = u.UpdateError
|
|
GIB = 1024 ** 3
|
|
|
|
|
|
class TransientStorageError(Error):
|
|
pass
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise Error(message)
|
|
|
|
|
|
def sha_file(path):
|
|
h = hashlib.sha256()
|
|
with Path(path).open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def literal(value):
|
|
"""Resolved Compose JSON still undergoes interpolation when used as a file."""
|
|
if isinstance(value, str):
|
|
return value.replace("$", "$$")
|
|
if isinstance(value, list):
|
|
return [literal(v) for v in value]
|
|
if isinstance(value, dict):
|
|
return {k: literal(v) for k, v in value.items()}
|
|
return value
|
|
|
|
|
|
def resolved(value):
|
|
"""Compose config serializes literal dollar signs escaped for another Compose read."""
|
|
if isinstance(value, str):
|
|
return value.replace("$$", "$")
|
|
if isinstance(value, list):
|
|
return [resolved(v) for v in value]
|
|
if isinstance(value, dict):
|
|
return {k: resolved(v) for k, v in value.items()}
|
|
return value
|
|
|
|
|
|
def validate_minio_listing(root, bucket, inventory, *, exact=False):
|
|
"""MinIO can hide a/b from LIST when a exists, while GET a/b still works.
|
|
|
|
Only inspect paths in the known single-drive layout; never decode or modify
|
|
xl.meta. A cold equality check guards the completeness of the S3 inventory.
|
|
"""
|
|
source = root / "minio"
|
|
format_info = u.read_json(source / ".minio.sys/format.json", {})
|
|
require(format_info.get("format") == "xl-single" and format_info.get("xl", {}).get("version") == "3",
|
|
"MinIO-Datenformat ist für den zusätzlichen Vollständigkeitsabgleich nicht freigegeben.")
|
|
bucket_root = source / bucket
|
|
physical = {str(p.parent.relative_to(bucket_root)) for p in bucket_root.rglob("xl.meta") if p.is_file()}
|
|
for key in physical:
|
|
parts = key.split("/")
|
|
require(all("/".join(parts[:n]) not in physical for n in range(1, len(parts))),
|
|
"MinIO enthält einen Datei-/Verzeichniskonflikt; die S3-Auflistung kann Objekte verbergen.")
|
|
if exact:
|
|
require(physical == {o["Key"] for o in inventory["objects"]},
|
|
"Kalte MinIO-Metadaten und S3-Inventar weichen ab; keine vollständige Übernahme nachweisbar.")
|
|
|
|
|
|
def validate_config(config, root):
|
|
services = config.get("services", {})
|
|
required = {"db", "minio", "createbuckets", "backend", "frontend", "traefik", "node-exporter"}
|
|
require(required <= services.keys() and not (services.keys() - required - u.LEGACY_MATRIX_SERVICES),
|
|
"Nur der Installer-Stack wird unterstützt; zusätzliche/fehlende Dienste zuerst klären.")
|
|
for service in services.values():
|
|
require(not any(service.get(k) for k in ("build", "profiles", "secrets", "configs", "volumes_from")),
|
|
"Builds, Profile und externe Konfigurationen benötigen einen eigenen Migrationsplan.")
|
|
store = services["minio"]
|
|
require(not store.get("ports") and set(store.get("networks", {})) == {"internal"},
|
|
"MinIO muss ausschließlich im internen Docker-Netz erreichbar sein.")
|
|
require(not services["db"].get("ports"), "Veröffentlichte Datenbankports benötigen einen gesonderten Schreiber-Nachweis.")
|
|
require(set(store.get("environment", {})) == {"MINIO_ROOT_USER", "MINIO_ROOT_PASSWORD"},
|
|
"Zusätzliche MinIO-Konfiguration benötigt eine gesonderte Prüfung.")
|
|
for name, target, directory in (("minio", "/data", "minio"), ("db", "/var/lib/postgresql/data", "postgres")):
|
|
mounts = services[name].get("volumes", [])
|
|
require(len(mounts) == 1 and mounts[0].get("type") == "bind" and mounts[0].get("target") == target
|
|
and mounts[0].get("source") == str(root / directory), "Standard-Datenpfade fehlen: " + name)
|
|
require((root / directory).is_dir() and (root / directory).resolve() == root / directory,
|
|
"Datenpfad fehlt oder ist ein Link: " + directory)
|
|
env = services["backend"].get("environment", {})
|
|
require(env.get("FEDEO_FILE_BACKEND", "s3") == "s3" and not env.get("INFISICAL_CLIENT_ID"),
|
|
"Externe Speicher/Secret-Provider benötigen einen eigenen Adapter.")
|
|
require(env.get("S3_ENDPOINT", "").rstrip("/") == "http://minio:9000", "Abweichender S3-Endpunkt.")
|
|
require(all(env.get(k) for k in ("S3_BUCKET", "S3_REGION", "S3_ACCESS_KEY", "S3_SECRET_KEY")), "S3-Konfiguration unvollständig.")
|
|
require(re.fullmatch(r"[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]", env["S3_BUCKET"]), "Ungültiger Bucketname.")
|
|
require(env["S3_ACCESS_KEY"] == store["environment"]["MINIO_ROOT_USER"] and
|
|
env["S3_SECRET_KEY"] == store["environment"]["MINIO_ROOT_PASSWORD"],
|
|
"Getrennte MinIO-Benutzer vorhanden; deren Rechte zunächst vollständig inventarisieren.")
|
|
db = urllib.parse.urlsplit(env.get("DATABASE_URL", ""))
|
|
db_env = services["db"].get("environment", {})
|
|
require(db.hostname == "db" and db.port in (None, 5432) and
|
|
urllib.parse.unquote(db.path.lstrip("/")) == db_env.get("POSTGRES_DB") and
|
|
urllib.parse.unquote(db.username or "") == db_env.get("POSTGRES_USER") and
|
|
urllib.parse.unquote(db.password or "") == db_env.get("POSTGRES_PASSWORD"),
|
|
"DATABASE_URL stimmt nicht mit der lokalen Sicherung überein.")
|
|
return env
|
|
|
|
|
|
def index_containers(config, containers):
|
|
"""Keep previously retired Matrix containers without reactivating their services."""
|
|
indexed, retained = {}, {}
|
|
for container in containers:
|
|
labels = container["Config"].get("Labels", {})
|
|
name = labels.get("com.docker.compose.service")
|
|
require(name and name not in indexed and labels.get("com.docker.compose.oneoff", "False").lower() != "true",
|
|
"Unbekannte oder doppelte Projektcontainer vorhanden.")
|
|
if name not in config["services"]:
|
|
require(name in u.LEGACY_MATRIX_SERVICES, "Unbekannter Projektcontainer: " + name)
|
|
state, host = container["State"], container["HostConfig"]
|
|
require(not any(state.get(k) for k in ("Running", "Restarting", "Paused")) and
|
|
host.get("RestartPolicy", {}).get("Name") == "no" and not host.get("AutoRemove"),
|
|
"Alter Matrix-Container ist nicht dauerhaft stillgelegt: " + name)
|
|
volumes = []
|
|
for mount in container.get("Mounts", []):
|
|
if mount["Type"] == "bind":
|
|
volumes.append({"type": "bind", "source": mount["Source"], "target": mount["Destination"]})
|
|
else:
|
|
# These image-declared volumes belong to already retired services.
|
|
# Preserve their containers/volumes; never remove or restart them.
|
|
require(mount["Type"] == "volume" and (name, mount["Destination"]) in {
|
|
("matrix-redis", "/data"), ("matrix-turn", "/var/lib/coturn")},
|
|
"Abweichender Datenmount eines alten Matrix-Containers: " + name)
|
|
retained[name] = {"volumes": volumes}
|
|
indexed[name] = container
|
|
return indexed, retained
|
|
|
|
|
|
class Migration:
|
|
def __init__(self, root, tools_image=None):
|
|
os.umask(0o077)
|
|
self.engine = u.Updater(root)
|
|
self.root = self.engine.root
|
|
self.home = self.engine.home / "storage-migration"
|
|
require(not self.home.is_symlink(), "Migrationsverzeichnis darf kein Link sein.")
|
|
self.home.mkdir(mode=0o700, exist_ok=True)
|
|
self.runner = u.Runner(self.home / "commands.log")
|
|
self.state = u.read_json(self.home / "state.json", {})
|
|
self.tools_image = tools_image or self.state.get("toolsImage")
|
|
require(not self.state or tools_image in (None, self.state.get("toolsImage")), "Image eines bestehenden Auftrags darf nicht ersetzt werden.")
|
|
self.weed = u.read_json(HERE / "storage/seaweedfs.json")
|
|
self.helper_name = "fedeo-storage-helper-" + hashlib.sha256(str(self.root).encode()).hexdigest()[:12]
|
|
self.restore_name = self.helper_name + "-restore"
|
|
|
|
def docker(self, *args, **kwargs):
|
|
return self.runner.run(["docker", *args], **kwargs)
|
|
|
|
def compose(self, file, *args, **kwargs):
|
|
return self.docker("compose", "--project-directory", str(self.root), "--env-file", str(self.root / ".env"),
|
|
"-p", self.state["project"], "-f", str(file), *args, **kwargs)
|
|
|
|
def save(self, phase=None, **fields):
|
|
self.state.update(fields, updatedAt=u.now())
|
|
if phase:
|
|
self.state["phase"] = phase
|
|
u.write_json(self.home / "state.json", self.state)
|
|
|
|
def guard(self):
|
|
# Compatible with the already installed 3.3 updater: it refuses recoveryRequired.
|
|
# Kept after cutover until a provider-aware application updater is installed.
|
|
u.write_json(self.engine.home / "status.json", {
|
|
"status": "failed", "phase": "storage_migration", "job": str(self.home),
|
|
"startedAt": self.state["startedAt"], "updatedAt": u.now(), "recoveryRequired": True,
|
|
"maintenance": not self.state.get("writesReleased", False), "mutationStarted": True,
|
|
"error": "Speichermigration: selfhost-storage-migrate.py status prüfen. Herkömmliche Updates sind gesperrt."})
|
|
|
|
def fingerprint(self):
|
|
return {name: sha_file(self.root / name) for name in ("docker-compose.yml", ".env")}
|
|
|
|
def inspect(self, container):
|
|
return json.loads(self.docker("inspect", container))[0]
|
|
|
|
def wait_ready(self, container, timeout=180):
|
|
for _ in range(timeout):
|
|
state = self.inspect(container)["State"]
|
|
require(state.get("Running"), "Ein benötigter Container wurde beendet.")
|
|
if state.get("Health", {}).get("Status") in (None, "healthy"):
|
|
return
|
|
time.sleep(1)
|
|
raise Error("Container wurde innerhalb der Wartezeit nicht bereit.")
|
|
|
|
def cleanup_helper(self):
|
|
ids = self.docker("ps", "-aq", "--filter", "name=^/" + self.helper_name + "$").split()
|
|
for container in ids:
|
|
require(self.inspect(container)["Config"].get("Labels", {}).get("io.fedeo.storage-job") == str(self.home),
|
|
"Fremder Container belegt den Hilfsnamen.")
|
|
self.docker("rm", "-f", container)
|
|
|
|
def helper(self, *args, **kwargs):
|
|
deadline = time.monotonic() + 180
|
|
while True:
|
|
try:
|
|
return self.helper_once(*args, **kwargs)
|
|
except TransientStorageError:
|
|
if time.monotonic() >= deadline:
|
|
raise
|
|
time.sleep(2)
|
|
|
|
def helper_once(self, command, *, source=False, admin=False, endpoint="http://objectstore:9000", directory="copy", backend_sdk=False):
|
|
self.cleanup_helper()
|
|
config = u.read_json(self.home / "credentials.json")
|
|
job = {"bucket": config["bucket"], "directory": "/work", "result": "/work/result.json"}
|
|
if command == "contract":
|
|
job["foreignBucket"] = "fedeo-denied-" + secrets.token_hex(8)
|
|
if source:
|
|
job["source"] = config["source"]
|
|
job["target"] = {**config["admin" if admin else "target"], "endpoint": endpoint}
|
|
u.write_json(self.home / "job.json", job)
|
|
work = self.home / directory
|
|
work.mkdir(mode=0o700, exist_ok=True)
|
|
(work / "error.json").unlink(missing_ok=True)
|
|
runtime = [self.tools_image, command]
|
|
if backend_sdk:
|
|
runtime = ["--entrypoint", "node", "-e", "FEDEO_STORAGE_SDK_ROOT=" + self.state["backendWorkingDir"],
|
|
"-v", str(HERE / "storage") + ":/fedeo-check:ro", self.state["containers"]["backend"]["image"],
|
|
"/fedeo-check/cli.mjs", command]
|
|
try:
|
|
self.docker("run", "--rm", "--pull", "never", "--name", self.helper_name,
|
|
"--label", "io.fedeo.storage-job=" + str(self.home), "--network", self.state["network"],
|
|
"--user", str(os.getuid()) + ":" + str(os.getgid()),
|
|
"--read-only", "--cap-drop=ALL", "--security-opt", "no-new-privileges", "--tmpfs", "/tmp",
|
|
"-v", str(self.home / "job.json") + ":/config/job.json:ro", "-v", str(work) + ":/work",
|
|
*runtime, timeout=24 * 3600, output=self.home / "progress.jsonl")
|
|
return u.read_json(work / "result.json")
|
|
except Error:
|
|
detail = u.read_json(work / "error.json", {})
|
|
if detail.get("status") in (500, 502, 503, 504) or detail.get("code") in ("ECONNREFUSED", "ECONNRESET", "ETIMEDOUT", "EHOSTUNREACH"):
|
|
raise TransientStorageError("S3 noch nicht bereit; Wiederholung blieb erfolglos.") from None
|
|
raise
|
|
finally:
|
|
self.cleanup_helper()
|
|
|
|
def plan(self):
|
|
require(not self.state, "Ein Plan existiert bereits; status prüfen oder vorhandenen Auftrag fortsetzen.")
|
|
require(not self.engine.status().get("recoveryRequired"), "Zuerst den offenen Update-Auftrag klären.")
|
|
require(self.tools_image, "--tools-image mit lokalem Hilfsimage angeben.")
|
|
require(all((self.root / n).is_file() and not (self.root / n).is_symlink() for n in (".env", "docker-compose.yml")),
|
|
"Reguläre .env und docker-compose.yml erforderlich.")
|
|
require(not any((self.root / n).exists() for n in ("docker-compose.override.yml", "docker-compose.override.yaml", "compose.yml", "compose.yaml")),
|
|
"Compose-Overrides zuerst in einem gesonderten Plan berücksichtigen.")
|
|
endpoint = self.docker("context", "inspect", "--format", "{{.Endpoints.docker.Host}}").strip()
|
|
require(os.environ.get("DOCKER_HOST", endpoint).startswith("unix://"),
|
|
"Nur ein lokaler Docker-Daemon wird unterstützt.")
|
|
info = json.loads(self.docker("info", "--format", "{{json .}}"))
|
|
require(info["OSType"] == "linux" and info["Architecture"] in ("amd64", "x86_64"), "Pilot benötigt Linux amd64.")
|
|
config = resolved(json.loads(self.docker("compose", "--project-directory", str(self.root), "--env-file", str(self.root / ".env"),
|
|
"-f", str(self.root / "docker-compose.yml"), "config", "--format", "json")))
|
|
env = validate_config(config, self.root)
|
|
project = config["name"]
|
|
ids = self.docker("ps", "-aq", "--filter", "label=com.docker.compose.project=" + project).split()
|
|
containers = [self.inspect(i) for i in ids]
|
|
by_service, retained = index_containers(config, containers)
|
|
# Validate retained bind mounts with the same strict path rules as configured
|
|
# Matrix services. Do not put retired services back into either Compose file.
|
|
self.engine.current = copy.deepcopy(config)
|
|
self.engine.current["services"].update(retained)
|
|
self.engine.retired_services = sorted(set(self.engine.current["services"]) & u.LEGACY_MATRIX_SERVICES)
|
|
self.engine.matrix_backup_paths = []
|
|
self.engine.validate_legacy_matrix()
|
|
require(all(by_service.get(n, {}).get("State", {}).get("Running") for n in ("db", "minio", "backend", "frontend", "traefik")),
|
|
"Die bestehende Instanz muss vor der Planung laufen.")
|
|
for name in ("minio", "backend", "db"):
|
|
actual = dict(v.split("=", 1) for v in by_service[name]["Config"]["Env"] if "=" in v)
|
|
require(all(actual.get(k) == str(v) for k, v in config["services"][name].get("environment", {}).items()),
|
|
"Laufender Container und Compose-Umgebung weichen ab: " + name)
|
|
network = config["networks"]["internal"]["name"]
|
|
peers = json.loads(self.docker("network", "inspect", network))[0].get("Containers", {})
|
|
require(set(peers) <= {c["Id"] for c in containers}, "Weitere Container im internen Netzwerk; externe Schreiber zuerst klären.")
|
|
for name, target in (("minio", "/data"), ("db", "/var/lib/postgresql/data")):
|
|
mounts = [m for m in by_service[name]["Mounts"] if m["Destination"] == target]
|
|
require(len(mounts) == 1 and mounts[0]["Source"] == str(self.root / ("minio" if name == "minio" else "postgres")),
|
|
"Laufender Datenmount weicht ab: " + name)
|
|
actual_store = by_service["minio"]
|
|
require(not actual_store["HostConfig"].get("PortBindings") and set(actual_store["NetworkSettings"]["Networks"]) == {network},
|
|
"Laufendes MinIO hat zusätzliche Netzwerkzugänge.")
|
|
require(len(actual_store["Mounts"]) == 1, "Laufendes MinIO hat zusätzliche Mounts.")
|
|
require(not any(p.is_symlink() for p in (self.root / "minio").rglob("*")), "Links in MinIO-Daten benötigen eigene Backup-Prüfung.")
|
|
tools = json.loads(self.docker("image", "inspect", self.tools_image))[0]
|
|
weed = json.loads(self.docker("image", "inspect", self.weed["image"]))[0]
|
|
self.tools_image = tools["Id"]
|
|
# Containerd may discard an untagged image index after a mutable build tag
|
|
# moves. Keep an independent immutable job reference throughout the pilot.
|
|
tools_tag = "fedeo-storage-tools:job-" + secrets.token_hex(12)
|
|
self.docker("image", "tag", self.tools_image, tools_tag)
|
|
self.state = {"schemaVersion": 1, "phase": "planning", "startedAt": u.now(), "project": project,
|
|
"network": network, "toolsImage": self.tools_image, "weedImage": weed["Id"],
|
|
"toolsRetainTag": tools_tag,
|
|
"version": env.get("FEDEO_VERSION"), "fingerprint": self.fingerprint(), "writesReleased": False,
|
|
"backendWorkingDir": by_service["backend"]["Config"].get("WorkingDir") or "/",
|
|
"matrixPaths": self.engine.matrix_backup_paths, "retainedServices": sorted(retained),
|
|
"containers": {name: {"id": c["Id"], "image": c["Image"],
|
|
"running": c["State"]["Running"], "restart": c["HostConfig"]["RestartPolicy"],
|
|
"health": c["State"].get("Health", {}).get("Status"),
|
|
"networks": c["NetworkSettings"]["Networks"]} for name, c in by_service.items()}}
|
|
# Planning errors must not leave a usable partial state file.
|
|
u.write_json(self.home / "original.json", config)
|
|
for name in ("docker-compose.yml", ".env"):
|
|
u.atomic(self.home / ("original" + name.replace("/", "-")), (self.root / name).read_bytes())
|
|
u.write_json(self.home / "previous-update.json", u.read_json(self.engine.home / "status.json", {"status": "idle"}))
|
|
admin = {"accessKeyId": "fedeo-migration-" + secrets.token_hex(8), "secretAccessKey": secrets.token_urlsafe(36)}
|
|
app = {"accessKeyId": env["S3_ACCESS_KEY"], "secretAccessKey": env["S3_SECRET_KEY"]}
|
|
connection = {"region": env["S3_REGION"], "credentials": app}
|
|
u.write_json(self.home / "credentials.json", {"bucket": env["S3_BUCKET"],
|
|
"source": {**connection, "endpoint": env["S3_ENDPOINT"]}, "target": connection,
|
|
"admin": {**connection, "credentials": admin}})
|
|
inventory = self.helper("inventory", source=True, directory="preflight")
|
|
validate_minio_listing(self.root, env["S3_BUCKET"], inventory)
|
|
physical = int(self.runner.run(["du", "-sb", str(self.root / "minio")]).split()[0])
|
|
db_size = int(self.docker("exec", by_service["db"]["Id"], "sh", "-ec",
|
|
'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -Atc "SELECT pg_database_size(current_database())"').strip())
|
|
image_ids = sorted({c["Image"] for c in containers} | {tools["Id"], weed["Id"]})
|
|
image_bytes = sum(i["Size"] for i in json.loads(self.docker("image", "inspect", *image_ids)))
|
|
matrix_bytes = sum(int(self.runner.run(["du", "-sb", str(self.root / p)]).split()[0]) for p in self.state["matrixPaths"])
|
|
# Source backup + target + restored target + offline images + DB + spool + generous metadata reserve.
|
|
needed = physical + inventory["bytes"] * 3 + inventory["count"] * 256 * 1024 + db_size * 2 + image_bytes * 2 + matrix_bytes + 5 * GIB
|
|
require(shutil.disk_usage(self.root).free >= needed, f"Zu wenig freier Platz; mindestens {needed} Bytes erforderlich.")
|
|
self.state.update(imageIds=image_ids, requiredFreeBytes=needed, freeBytes=shutil.disk_usage(self.root).free,
|
|
count=inventory["count"], bytes=inventory["bytes"])
|
|
require(not (self.root / "storage").exists(), "storage/ existiert bereits; kein leeres Migrationsziel.")
|
|
s3config = {"identities": [
|
|
{"name": "migration-admin", "credentials": [{"accessKey": admin["accessKeyId"], "secretKey": admin["secretAccessKey"]}],
|
|
"actions": ["Admin", "Read", "Write", "List", "Tagging"]},
|
|
{"name": "fedeo", "credentials": [{"accessKey": app["accessKeyId"], "secretKey": app["secretAccessKey"]}],
|
|
"actions": [action + ":" + env["S3_BUCKET"] for action in ("Read", "Write", "List")]}
|
|
]}
|
|
u.write_json(self.home / "s3.json", s3config)
|
|
self.make_compose(config)
|
|
self.save("planned")
|
|
|
|
def make_compose(self, original):
|
|
target = copy.deepcopy(original)
|
|
store = copy.deepcopy(self.weed)
|
|
store["image"] = self.state["weedImage"]
|
|
store.update(container_name="fedeo-objectstore", restart="no", networks={"internal": {}},
|
|
user=str(os.getuid()) + ":" + str(os.getgid()),
|
|
environment={"FEDEO_S3_CONFIG_SHA256": sha_file(self.home / "s3.json")},
|
|
volumes=[{"type": "bind", "source": str(self.root / "storage/seaweedfs"), "target": "/data"},
|
|
{"type": "bind", "source": str(self.root / "storage/config"), "target": "/etc/fedeo", "read_only": True}])
|
|
stage = {"name": self.state["project"], "services": {"objectstore": store},
|
|
"networks": {"internal": {"external": True, "name": self.state["network"]}}}
|
|
u.write_json(self.home / "stage.json", literal(stage))
|
|
del target["services"]["minio"]
|
|
del target["services"]["createbuckets"]
|
|
target["services"]["objectstore"] = copy.deepcopy(store)
|
|
target["services"]["objectstore"]["networks"]["internal"] = {"aliases": ["minio"]}
|
|
target["services"]["objectstore"]["restart"] = "unless-stopped"
|
|
dependencies = target["services"]["backend"].setdefault("depends_on", {})
|
|
dependencies.pop("minio", None)
|
|
dependencies.pop("createbuckets", None)
|
|
dependencies["objectstore"] = {"condition": "service_healthy"}
|
|
# Keep exactly the running application images; never couple this to an application/DB update.
|
|
for name, c in self.state["containers"].items():
|
|
if name in target["services"]:
|
|
target["services"][name]["image"] = c["image"]
|
|
target["services"]["backend"]["environment"]["FEDEO_RUN_MIGRATIONS"] = "false"
|
|
# SDK 3.962 streaming trailers leave aws-chunked in SeaweedFS object metadata.
|
|
# SigV4 remains active; the migration tool supplies an explicit SHA-256 checksum.
|
|
target["services"]["backend"]["environment"]["AWS_REQUEST_CHECKSUM_CALCULATION"] = "WHEN_REQUIRED"
|
|
u.write_json(self.home / "target.json", literal(target))
|
|
# Fail before maintenance if Compose cannot parse the generated configurations.
|
|
self.compose(self.home / "stage.json", "config", "--quiet")
|
|
self.compose(self.home / "target.json", "config", "--quiet")
|
|
|
|
def verify_original(self):
|
|
require(self.fingerprint() == self.state["fingerprint"], "Compose/.env wurde nach der Planung verändert.")
|
|
for name, c in self.state["containers"].items():
|
|
actual = self.inspect(c["id"])
|
|
require(actual["Image"] == c["image"], "Containeridentität verändert: " + name)
|
|
if name in self.state.get("retainedServices", []):
|
|
index_containers({"services": {}}, [actual])
|
|
|
|
def freeze(self):
|
|
self.save("freezing")
|
|
self.guard()
|
|
for name, c in self.state["containers"].items():
|
|
if name != "db":
|
|
self.docker("update", "--restart=no", c["id"])
|
|
ids = [c["id"] for name, c in self.state["containers"].items() if name != "db"]
|
|
if ids:
|
|
self.docker("stop", "-t", "60", *ids)
|
|
self.assert_frozen()
|
|
|
|
def assert_frozen(self):
|
|
require(self.inspect(self.state["containers"]["minio"]["id"])["HostConfig"]["RestartPolicy"]["Name"] == "no",
|
|
"MinIO könnte automatisch wieder starten.")
|
|
for name, c in self.state["containers"].items():
|
|
if name in ("db", "minio"):
|
|
continue
|
|
actual = self.inspect(c["id"])
|
|
require(not actual["State"]["Running"] and actual["HostConfig"]["RestartPolicy"]["Name"] == "no",
|
|
"Schreibpause wurde aufgehoben: " + name)
|
|
known = {c["id"] for c in self.state["containers"].values()}
|
|
peers = json.loads(self.docker("network", "inspect", self.state["network"]))[0].get("Containers", {})
|
|
for cid in set(peers) - known:
|
|
c = self.inspect(cid)
|
|
labels = c["Config"].get("Labels", {})
|
|
target = (labels.get("com.docker.compose.project") == self.state["project"] and
|
|
labels.get("com.docker.compose.service") == "objectstore" and c["Image"] == self.state["weedImage"])
|
|
own_helper = labels.get("io.fedeo.storage-job") == str(self.home)
|
|
require(target or own_helper, "Ein zusätzlicher Container hat sich mit dem Speichernetz verbunden.")
|
|
|
|
def check_backup(self):
|
|
backup = self.home / "backup"
|
|
complete = u.read_json(backup / "complete.json", {})
|
|
require(complete.get("sha256"), "Abgeschlossene Sicherung fehlt.")
|
|
for name, expected in complete["sha256"].items():
|
|
require(Path(name).name == name and sha_file(backup / name) == expected,
|
|
"Backup-Prüfsumme stimmt nicht: " + name)
|
|
|
|
def backup(self):
|
|
self.save("backing_up")
|
|
backup = self.home / "backup"
|
|
backup.mkdir(mode=0o700, exist_ok=True)
|
|
require(not self.inspect(self.state["containers"]["minio"]["id"])["State"]["Running"], "MinIO für kalte Sicherung stoppen.")
|
|
db = self.state["containers"]["db"]["id"]
|
|
self.docker("exec", db, "sh", "-ec", 'exec pg_dump -U "$POSTGRES_USER" -d "$POSTGRES_DB" -Fc', output=backup / "database.dump")
|
|
self.docker("exec", "-i", db, "pg_restore", "--list", input_file=backup / "database.dump", output=backup / "database.list")
|
|
self.docker("exec", db, "sh", "-ec", 'exec pg_dumpall -U "$POSTGRES_USER" --globals-only', output=backup / "globals.sql")
|
|
paths = ["minio", *self.state["matrixPaths"], "storage/config", ".env", "docker-compose.yml"]
|
|
if (self.root / "traefik/letsencrypt").is_dir():
|
|
paths.append("traefik/letsencrypt")
|
|
self.runner.run(["tar", "-cpf", str(backup / "files.tar"), "-C", str(self.root), *paths], timeout=7200)
|
|
self.docker("image", "save", "-o", str(backup / "images.tar"), *self.state["imageIds"], timeout=7200)
|
|
u.write_json(backup / "complete.json", {"at": u.now(), "sha256": {p.name: sha_file(p) for p in backup.iterdir() if p.is_file() and p.name != "complete.json"}})
|
|
self.save("backed_up")
|
|
|
|
def migrate(self, writers_exclusive):
|
|
require(writers_exclusive, "--writers-exclusive bestätigt: keine externen S3-/DB-Schreiber und kein paralleles manuelles Compose.")
|
|
require(self.state.get("phase") in ("planned", "freezing", "backing_up", "backed_up", "copying", "verifying", "verified"),
|
|
"Migration kann in dieser Phase nicht gestartet werden.")
|
|
self.cleanup_helper()
|
|
self.verify_original()
|
|
require(shutil.disk_usage(self.root).free >= (self.state["requiredFreeBytes"] if self.state["phase"] == "planned" else GIB),
|
|
"Nicht genug freier Speicher für Migration/Fortsetzung.")
|
|
if self.state["phase"] in ("planned", "freezing", "backing_up"):
|
|
(self.root / "storage/config").mkdir(parents=True, mode=0o700, exist_ok=True)
|
|
(self.root / "storage/seaweedfs/master").mkdir(parents=True, mode=0o700, exist_ok=True)
|
|
u.atomic(self.root / "storage/config/s3.json", (self.home / "s3.json").read_bytes())
|
|
u.atomic(self.root / "storage/config/start-seaweedfs.sh", (HERE / "storage/start-seaweedfs.sh").read_bytes())
|
|
self.freeze()
|
|
self.docker("start", self.state["containers"]["minio"]["id"])
|
|
self.wait_ready(self.state["containers"]["minio"]["id"])
|
|
current = self.helper("inventory", source=True, directory="frozen-inventory")
|
|
extra = max(0, current["bytes"] - self.state["bytes"]) * 4 + max(0, current["count"] - self.state["count"]) * 256 * 1024
|
|
require(shutil.disk_usage(self.root).free >= self.state["requiredFreeBytes"] + extra,
|
|
"Bestand seit Planung gewachsen; Speicherreserve reicht nicht mehr.")
|
|
self.save(count=current["count"], bytes=current["bytes"])
|
|
self.docker("stop", "-t", "60", self.state["containers"]["minio"]["id"])
|
|
validate_minio_listing(self.root, u.read_json(self.home / "credentials.json")["bucket"], current, exact=True)
|
|
self.backup()
|
|
self.assert_frozen()
|
|
self.guard()
|
|
self.check_backup()
|
|
self.docker("start", self.state["containers"]["minio"]["id"])
|
|
self.wait_ready(self.state["containers"]["minio"]["id"])
|
|
self.compose(self.home / "stage.json", "up", "-d", "--pull", "never", "--wait", "--wait-timeout", "180")
|
|
self.helper("init", admin=True, directory="init")
|
|
self.helper("contract", directory="contract")
|
|
self.helper("contract", directory="backend-contract", backend_sdk=True)
|
|
self.save("copying")
|
|
self.helper("copy", source=True)
|
|
self.assert_frozen()
|
|
self.save("verifying")
|
|
self.helper("verify", source=True)
|
|
self.assert_frozen()
|
|
self.save("verified")
|
|
|
|
def restore_test(self):
|
|
# Start a separate container on newly extracted directories. Never count a
|
|
# successful archive listing or restart of the original directory as restore.
|
|
self.save("restore_testing")
|
|
restore = self.home / "restore-test"
|
|
if restore.exists():
|
|
# A previous interrupted restore is retained; use a fresh directory.
|
|
restore = self.home / ("restore-test-" + secrets.token_hex(4))
|
|
restore.mkdir(mode=0o700)
|
|
self.runner.run(["tar", "-xpf", str(self.home / "backup/seaweedfs.tar"), "-C", str(restore)], timeout=7200)
|
|
known = self.docker("ps", "-aq", "--filter", "name=^/" + self.restore_name + "$").split()
|
|
for cid in known:
|
|
require(self.inspect(cid)["Config"].get("Labels", {}).get("io.fedeo.storage-job") == str(self.home), "Fremder Restore-Container.")
|
|
self.docker("rm", "-f", cid)
|
|
target = u.read_json(self.home / "stage.json")["services"]["objectstore"]
|
|
try:
|
|
self.docker("run", "-d", "--pull", "never", "--name", self.restore_name,
|
|
"--label", "io.fedeo.storage-job=" + str(self.home), "--network", self.state["network"],
|
|
"--user", str(os.getuid()) + ":" + str(os.getgid()),
|
|
"-v", str(restore / "storage/seaweedfs") + ":/data",
|
|
"-v", str(restore / "storage/config") + ":/etc/fedeo:ro",
|
|
"-e", "FEDEO_S3_CONFIG_SHA256=" + sha_file(self.home / "s3.json"),
|
|
"--entrypoint", target["entrypoint"][0], self.state["weedImage"],
|
|
*target["entrypoint"][1:], *target["command"])
|
|
self.helper("restore-verify", endpoint="http://" + self.restore_name + ":9000")
|
|
self.helper("contract", endpoint="http://" + self.restore_name + ":9000", directory="restore-contract")
|
|
self.save("cutover", restoreVerifiedAt=u.now(), seaweedBackupSha256=sha_file(self.home / "backup/seaweedfs.tar"))
|
|
finally:
|
|
self.docker("rm", "-f", self.restore_name)
|
|
|
|
def activate(self):
|
|
require(self.state.get("phase") in ("verified", "cutover", "restore_testing"), "Erst Kopie und Prüfung abschließen.")
|
|
require(not self.state.get("writesReleased"), "Schreibfreigabe ist bereits dauerhaft erfolgt.")
|
|
self.cleanup_helper()
|
|
self.assert_frozen()
|
|
# A killed contract test can leave only its journalled probe keys behind.
|
|
# Bring the target up without the legacy alias, remove those exact keys,
|
|
# then take the cold backup. Never delete unknown target objects.
|
|
self.compose(self.home / "stage.json", "up", "-d", "--pull", "never", "--wait", "--wait-timeout", "180")
|
|
for directory in ("contract", "backend-contract", "acceptance", "backend-acceptance"):
|
|
if (self.home / directory / "probe.json").is_file():
|
|
self.helper("cleanup-probe", directory=directory)
|
|
if self.state["phase"] == "verified":
|
|
self.verify_original()
|
|
self.helper("verify", source=True)
|
|
self.save("cutover")
|
|
self.guard()
|
|
self.check_backup()
|
|
old = self.state["containers"]["minio"]["id"]
|
|
self.docker("stop", "-t", "60", old)
|
|
validate_minio_listing(self.root, u.read_json(self.home / "credentials.json")["bucket"],
|
|
u.read_json(self.home / "copy/inventory.json"), exact=True)
|
|
if self.state["network"] in self.inspect(old)["NetworkSettings"]["Networks"]:
|
|
self.docker("network", "disconnect", self.state["network"], old)
|
|
self.compose(self.home / "stage.json", "stop", "-t", "60", "objectstore")
|
|
# A cold backup is required before serving any business write on the new provider.
|
|
self.runner.run(["tar", "-cpf", str(self.home / "backup/seaweedfs.tar"), "-C", str(self.root), "storage"], timeout=7200)
|
|
self.restore_test()
|
|
u.atomic(self.root / "docker-compose.yml", (self.home / "target.json").read_bytes())
|
|
self.compose(self.root / "docker-compose.yml", "up", "-d", "--no-deps", "--pull", "never", "--wait", "--wait-timeout", "180", "objectstore")
|
|
self.helper("restore-verify", endpoint="http://minio:9000")
|
|
self.helper("contract", endpoint="http://minio:9000", directory="acceptance")
|
|
self.helper("contract", endpoint="http://minio:9000", directory="backend-acceptance", backend_sdk=True)
|
|
# Persist before starting any backend/background process; no automatic rollback beyond here.
|
|
self.save("released", writesReleased=True, releasedAt=u.now(), provider="seaweedfs")
|
|
self.guard()
|
|
self.start_application()
|
|
|
|
def start_application(self):
|
|
require(self.state.get("writesReleased"), "Schreibfreigabe fehlt.")
|
|
require(sha_file(self.root / "docker-compose.yml") == sha_file(self.home / "target.json"), "Aktive Compose-Datei verändert.")
|
|
names = [n for n, c in self.state["containers"].items() if c["running"] and n not in ("minio", "createbuckets", "traefik")]
|
|
# Preserve Matrix without letting an already failing optional healthcheck
|
|
# cause a new FEDEO outage after cutover. Core services must always pass.
|
|
degraded = [n for n in names if n in u.LEGACY_MATRIX_SERVICES
|
|
and self.state["containers"][n].get("health") == "unhealthy"]
|
|
self.compose(self.root / "docker-compose.yml", "up", "-d", "--pull", "never", *names)
|
|
self.compose(self.root / "docker-compose.yml", "up", "-d", "--no-deps", "--pull", "never",
|
|
"--wait", "--wait-timeout", "240", *[n for n in names if n not in degraded])
|
|
for name in degraded:
|
|
cid = self.compose(self.root / "docker-compose.yml", "ps", "--all", "--quiet", name).strip()
|
|
require(cid and "\n" not in cid and self.inspect(cid)["State"]["Running"],
|
|
"Alter Matrix-Dienst wurde nicht wieder gestartet: " + name)
|
|
self.compose(self.root / "docker-compose.yml", "up", "-d", "--no-deps", "--pull", "never", "traefik")
|
|
self.save("complete", preexistingUnhealthyServices=degraded)
|
|
|
|
def rollback(self):
|
|
require(self.state and not self.state.get("writesReleased"), "Nach Schreibfreigabe ist eine geprüfte Rückmigration erforderlich.")
|
|
require(sha_file(self.root / ".env") == self.state["fingerprint"][".env"], ".env verändert; Rückkehr manuell prüfen.")
|
|
require(sha_file(self.root / "docker-compose.yml") in {self.state["fingerprint"]["docker-compose.yml"], sha_file(self.home / "target.json")},
|
|
"Compose-Datei außerhalb der Migration verändert; Rückkehr manuell prüfen.")
|
|
self.cleanup_helper()
|
|
for cid in self.docker("ps", "-aq", "--filter", "name=^/" + self.restore_name + "$").split():
|
|
require(self.inspect(cid)["Config"].get("Labels", {}).get("io.fedeo.storage-job") == str(self.home), "Fremder Restore-Container.")
|
|
self.docker("rm", "-f", cid)
|
|
require(self.state["phase"] != "planning", "Unvollständiger Plan.")
|
|
self.save("rolling_back")
|
|
self.guard()
|
|
# May not exist when the failure occurred before the first target start.
|
|
ids = self.docker("ps", "-aq", "--filter", "label=com.docker.compose.project=" + self.state["project"],
|
|
"--filter", "label=com.docker.compose.service=objectstore").split()
|
|
for cid in ids:
|
|
self.docker("update", "--restart=no", cid)
|
|
self.docker("stop", "-t", "60", cid)
|
|
if self.state["network"] in self.inspect(cid)["NetworkSettings"]["Networks"]:
|
|
self.docker("network", "disconnect", self.state["network"], cid)
|
|
u.atomic(self.root / "docker-compose.yml", (self.home / "originaldocker-compose.yml").read_bytes())
|
|
for name, c in self.state["containers"].items():
|
|
actual = self.inspect(c["id"])
|
|
if name == "minio" and self.state["network"] not in actual["NetworkSettings"]["Networks"]:
|
|
aliases = c["networks"][self.state["network"]].get("Aliases") or ["minio"]
|
|
args = [arg for alias in aliases for arg in ("--alias", alias)]
|
|
self.docker("network", "connect", *args, self.state["network"], c["id"])
|
|
policy = c["restart"]["Name"]
|
|
if policy == "on-failure" and c["restart"].get("MaximumRetryCount"):
|
|
policy += ":" + str(c["restart"]["MaximumRetryCount"])
|
|
self.docker("update", "--restart=" + policy, c["id"])
|
|
for name in ["db", "minio", *[n for n in self.state["containers"] if n not in ("db", "minio", "traefik")], "traefik"]:
|
|
c = self.state["containers"].get(name)
|
|
if c and c["running"]:
|
|
self.docker("start", c["id"])
|
|
self.wait_ready(c["id"])
|
|
self.save("rolled_back")
|
|
u.write_json(self.engine.home / "status.json", u.read_json(self.home / "previous-update.json"))
|
|
|
|
def public_status(self):
|
|
return {k: self.state[k] for k in ("phase", "version", "count", "bytes", "requiredFreeBytes", "freeBytes",
|
|
"writesReleased", "releasedAt", "updatedAt") if k in self.state}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("command", choices=["plan", "migrate", "activate", "resume-start", "rollback", "status"])
|
|
parser.add_argument("--directory", default="/opt/fedeo")
|
|
parser.add_argument("--tools-image")
|
|
parser.add_argument("--writers-exclusive", action="store_true")
|
|
args = parser.parse_args()
|
|
migration = Migration(args.directory, args.tools_image)
|
|
try:
|
|
if args.command == "status":
|
|
print(json.dumps(migration.public_status(), indent=2)); return
|
|
with migration.engine.lock():
|
|
if args.command == "plan": migration.plan()
|
|
elif args.command == "migrate": migration.migrate(args.writers_exclusive)
|
|
elif args.command == "activate": migration.activate()
|
|
elif args.command == "resume-start": migration.start_application()
|
|
elif args.command == "rollback": migration.rollback()
|
|
print(json.dumps(migration.public_status(), indent=2))
|
|
except (Error, OSError, ValueError, KeyboardInterrupt) as error:
|
|
print("Migration angehalten: " + str(error) + "\nGeschütztes Protokoll: " + str(migration.home / "commands.log"), file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|