feat(updater): support updates after verified SeaweedFS migration

This commit is contained in:
2026-09-28 15:47:06 +02:00
parent 790f20b499
commit 2b947c5e8f
20 changed files with 1074 additions and 67 deletions

View File

@@ -1,11 +1,12 @@
#!/usr/bin/env python3
"""FEDEO host-side update engine. Python 3.10+, Docker Compose 2.24+; no pip packages.
The CLI and a future host service share Updater. No daemon or HTTP API is installed.
The CLI and the optional local host service share Updater and the exclusive lock.
All commands operate on an existing local selfhost installation, never a git checkout.
"""
import argparse
import contextlib
import copy
import datetime
import difflib
import fcntl
@@ -31,6 +32,103 @@ FILES = ("release-manifest.json", "docker-compose.selfhost.yml", ".env.example",
GIB = 1024 ** 3
LEGACY_MATRIX_SERVICES = {"matrix-db", "matrix-synapse", "matrix-redis", "matrix-element",
"matrix-livekit", "matrix-rtc-jwt", "matrix-turn", "matrix-well-known"}
CORE_SERVICES = {"backend", "frontend", "db", "traefik", "node-exporter"}
IMAGE_ID = re.compile(r"sha256:[a-f0-9]{64}\Z")
def compose_strings(value, *, escape=False):
"""Round-trip resolved Compose output, including literal dollars in secrets."""
if isinstance(value, str):
return value.replace("$", "$$") if escape else value.replace("$$", "$")
if isinstance(value, list):
return [compose_strings(v, escape=escape) for v in value]
if isinstance(value, dict):
return {k: compose_strings(v, escape=escape) for k, v in value.items()}
return value
def file_sha(path):
digest = hashlib.sha256()
with Path(path).open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
# Runs with the actual application's SDK. Never prints keys, URLs or credentials.
S3_READ_CHECK = r"""
const {S3Client, HeadBucketCommand, ListObjectsV2Command, GetObjectCommand} = require('@aws-sdk/client-s3');
const client = new S3Client({endpoint:process.env.S3_ENDPOINT, region:process.env.S3_REGION,
forcePathStyle:true, maxAttempts:8, credentials:{accessKeyId:process.env.S3_ACCESS_KEY,secretAccessKey:process.env.S3_SECRET_KEY}});
const timer = setTimeout(()=>process.exit(1), 180000);
(async()=>{
const Bucket=process.env.S3_BUCKET;
await client.send(new HeadBucketCommand({Bucket}));
const listed=await client.send(new ListObjectsV2Command({Bucket,MaxKeys:1}));
const first=listed.Contents?.[0];
if(first) {
const object=await client.send(new GetObjectCommand({Bucket,Key:first.Key,...(first.Size ? {Range:'bytes=0-0'} : {})}));
await object.Body.transformToByteArray();
}
})().catch(()=>{console.error('S3-Zugriffsprüfung fehlgeschlagen');process.exitCode=1})
.finally(()=>{clearTimeout(timer);client.destroy()});
"""
S3_UPGRADE_CHECK = r"""
const s3 = require('@aws-sdk/client-s3');
const {getSignedUrl} = require('@aws-sdk/s3-request-presigner');
const {Readable} = require('node:stream');
const assert = require('node:assert/strict');
const client = new s3.S3Client({endpoint:process.env.S3_ENDPOINT, region:process.env.S3_REGION,
forcePathStyle:true, maxAttempts:8,
credentials:{accessKeyId:process.env.S3_ACCESS_KEY,secretAccessKey:process.env.S3_SECRET_KEY}});
const Bucket=process.env.S3_BUCKET, Key=process.env.FEDEO_UPDATE_PROBE_KEY;
const timer=setTimeout(()=>process.exit(1), 180000);
(async()=>{
assert.match(Key, /^\.fedeo-update-check\/[a-f0-9]{32}\.pdf$/);
try {
await client.send(new s3.HeadBucketCommand({Bucket}));
const body=Buffer.from('FEDEO S3 update check');
await client.send(new s3.PutObjectCommand({Bucket,Key,Body:Readable.from([body]),ContentLength:body.length,
ContentType:'application/octet-stream',Metadata:{fedeo:'update-check'}}));
const head=await client.send(new s3.HeadObjectCommand({Bucket,Key}));
assert.equal(head.ContentType,'application/octet-stream');
assert.equal(head.Metadata.fedeo,'update-check');
assert.ok(!head.ContentEncoding);
const result=await client.send(new s3.GetObjectCommand({Bucket,Key}));
assert.deepEqual(Buffer.from(await result.Body.transformToByteArray()),body);
const signed=await getSignedUrl(client,new s3.GetObjectCommand({Bucket,Key}),{expiresIn:86400});
const response=await fetch(signed);
assert.equal(response.status,200);
assert.deepEqual(Buffer.from(await response.arrayBuffer()),body);
const anonymous=new URL(signed); anonymous.search='';
const denied=await fetch(anonymous);
assert.equal(denied.status,403); await denied.body?.cancel();
} finally { await client.send(new s3.DeleteObjectCommand({Bucket,Key})); }
try { await client.send(new s3.HeadObjectCommand({Bucket,Key})); throw new Error('probe remains'); }
catch(error) { if(error.$metadata?.httpStatusCode!==404) throw error; }
})().catch(()=>{console.error('S3-Vertrag des Zielbackends fehlgeschlagen');process.exitCode=1})
.finally(()=>{clearTimeout(timer);client.destroy()});
"""
S3_CLEANUP_CHECK = r"""
const s3=require('@aws-sdk/client-s3'), assert=require('node:assert/strict');
const client=new s3.S3Client({endpoint:process.env.S3_ENDPOINT,region:process.env.S3_REGION,forcePathStyle:true,
credentials:{accessKeyId:process.env.S3_ACCESS_KEY,secretAccessKey:process.env.S3_SECRET_KEY}});
const Bucket=process.env.S3_BUCKET, Key=process.env.FEDEO_UPDATE_PROBE_KEY;
const timer=setTimeout(()=>process.exit(1),60000);
(async()=>{
assert.match(Key,/^\.fedeo-update-check\/[a-f0-9]{32}\.pdf$/);
let head;
try { head=await client.send(new s3.HeadObjectCommand({Bucket,Key})); }
catch(error) { if(error.$metadata?.httpStatusCode===404) return; throw error; }
assert.equal(head.Metadata?.fedeo,'update-check');
await client.send(new s3.DeleteObjectCommand({Bucket,Key}));
try { await client.send(new s3.HeadObjectCommand({Bucket,Key})); throw new Error('probe remains'); }
catch(error) { if(error.$metadata?.httpStatusCode!==404) throw error; }
})().catch(()=>{console.error('Protokollierten S3-Prüfschlüssel manuell prüfen');process.exitCode=1})
.finally(()=>{clearTimeout(timer);client.destroy()});
"""
class UpdateError(Exception):
@@ -283,7 +381,7 @@ class Updater:
return self.runner.run(prefix + list(args), **kwargs)
def config(self, candidate):
return json.loads(self.compose(candidate, "config", "--format", "json"))
return compose_strings(json.loads(self.compose(candidate, "config", "--format", "json")))
def inspect(self, name):
return json.loads(self.runner.run(["docker", "inspect", name]))[0]
@@ -292,6 +390,166 @@ class Updater:
if shutil.disk_usage(directory).free < required:
raise UpdateError(f"Zu wenig freier Speicher auf {directory}: mindestens {required // GIB + 1} GiB erforderlich.")
def storage_record(self):
path = self.home / "storage-provider.json"
if path.is_symlink():
raise UpdateError("Speicherbeschreibung darf kein symbolischer Link sein.")
record = read_json(path, {})
if record and (record.get("schemaVersion") != 1 or record.get("provider") != "seaweedfs"
or not IMAGE_ID.fullmatch(str(record.get("image", "")))):
raise UpdateError("Unbekannte Speicherbeschreibung; Speichermigration auf dem Host prüfen.")
return record
def validate_storage(self, config, record, *, running=True):
"""The adopted storage service is immutable across application releases."""
store = config.get("services", {}).get("objectstore", {})
if store != record.get("service") or store.get("image") != record.get("image"):
raise UpdateError("SeaweedFS-Konfiguration oder Image verändert; separate Speichermigration erforderlich.")
for directory in ("storage", "storage/seaweedfs", "storage/config"):
path = self.root / directory
if not path.is_dir() or path.resolve() != path:
raise UpdateError("SeaweedFS benötigt reguläre lokale storage/-Verzeichnisse.")
if any(p.is_symlink() for p in (self.root / "storage").rglob("*")):
raise UpdateError("Symbolische Links in storage/ können nicht vollständig gesichert werden.")
expected = {"/data": str(self.root / "storage/seaweedfs"), "/etc/fedeo": str(self.root / "storage/config")}
volumes = store.get("volumes", [])
if (len(volumes) != 2 or any(v.get("type") != "bind" for v in volumes)
or {v.get("target"): v.get("source") for v in volumes} != expected
or not next(v for v in volumes if v["target"] == "/etc/fedeo").get("read_only")):
raise UpdateError("SeaweedFS-Datenmounts entsprechen nicht der übernommenen lokalen Ablage.")
for name in ("s3.json", "start-seaweedfs.sh"):
path = self.root / "storage/config" / name
if not path.is_file() or file_sha(path) != record.get("configSha256", {}).get(name):
raise UpdateError("Geschützte SeaweedFS-Konfiguration verändert oder unvollständig.")
if store.get("environment", {}).get("FEDEO_S3_CONFIG_SHA256") != record["configSha256"]["s3.json"]:
raise UpdateError("SeaweedFS-Konfigurationsprüfsumme stimmt nicht.")
network = config.get("networks", {}).get("internal", {}).get("name")
if (network != record.get("network") or store.get("ports") or set(store.get("networks", {})) != {"internal"}
or "minio" not in store["networks"]["internal"].get("aliases", [])):
raise UpdateError("SeaweedFS muss ausschließlich am bisherigen internen S3-Anschluss liegen.")
old = self.inspect(record["retainedMinio"]["id"])
if (old.get("Image") != record["retainedMinio"]["image"]
or any(old.get("State", {}).get(k) for k in ("Running", "Restarting", "Paused"))
or old.get("HostConfig", {}).get("RestartPolicy", {}).get("Name") != "no"
or network in old.get("NetworkSettings", {}).get("Networks", {})):
raise UpdateError("Der erhaltene MinIO-Container ist nicht unverändert stillgelegt.")
if running:
cid = self.compose(False, "ps", "--quiet", "objectstore").strip()
actual = self.inspect(cid) if cid and "\n" not in cid else {}
mounts = actual.get("Mounts", [])
if (actual.get("Image") != record["image"] or not actual.get("State", {}).get("Running")
or actual.get("State", {}).get("Health", {}).get("Status") != "healthy"
or actual.get("HostConfig", {}).get("PortBindings")
or set(actual.get("NetworkSettings", {}).get("Networks", {})) != {network}
or "minio" not in actual["NetworkSettings"]["Networks"][network].get("Aliases", [])
or len(mounts) != 2 or any(v.get("Type") != "bind" for v in mounts)
or {v.get("Destination"): v.get("Source") for v in mounts} != expected
or any(v.get("RW") for v in mounts if v.get("Destination") == "/etc/fedeo")
or ("FEDEO_S3_CONFIG_SHA256=" + record["configSha256"]["s3.json"]) not in actual.get("Config", {}).get("Env", [])):
raise UpdateError("Laufender SeaweedFS-Container weicht vom geprüften Speicher ab.")
def adopt_storage(self):
"""Explicit host-only handover; cannot resolve an interrupted migration/update."""
with self.lock():
guard = read_json(self.home / "status.json", {})
if guard.get("phase") != "storage_migration" or not guard.get("recoveryRequired"):
if self.storage_record() and not guard.get("recoveryRequired"):
return {"status": "already_adopted", "storageProvider": "seaweedfs"}
raise UpdateError("Keine abgeschlossene Pilotmigration zur Übernahme vorhanden.")
migration = self.home / "storage-migration"
if migration.is_symlink() or guard.get("job") != str(migration):
raise UpdateError("Speichermigrationsauftrag stimmt nicht.")
state = read_json(migration / "state.json", {})
if (state.get("schemaVersion") != 1 or state.get("phase") != "complete"
or state.get("writesReleased") is not True or state.get("provider") != "seaweedfs"
or not state.get("restoreVerifiedAt") or not IMAGE_ID.fullmatch(str(state.get("weedImage", "")))):
raise UpdateError("Speichermigration und geprüften Restore zuerst vollständig abschließen; kein resolve verwenden.")
if any((self.root / n).exists() for n in ("docker-compose.override.yml", "docker-compose.override.yaml", "compose.yml", "compose.yaml")):
raise UpdateError("Pilotübernahme unterstützt keine zusätzlichen Compose-Dateien.")
before = {n: file_sha(self.root / n) for n in (".env", "docker-compose.yml")}
if any((self.root / n).is_symlink() for n in before):
raise UpdateError("Konfigurationsdateien dürfen keine symbolischen Links sein.")
if (before["docker-compose.yml"] != file_sha(migration / "target.json")
or before[".env"] != state.get("fingerprint", {}).get(".env")):
raise UpdateError("Installation nach der Speichermigration verändert; Übernahme manuell prüfen.")
verified = read_json(migration / "copy/verified.json", {})
if (not verified.get("verifiedAt") or verified.get("count") != state.get("count")
or verified.get("bytes") != state.get("bytes")
or len(verified.get("objects", [])) != state.get("count")):
raise UpdateError("Vollständiger Kopiernachweis der Speichermigration fehlt.")
for name in ("restore-contract", "acceptance", "backend-acceptance"):
if not {"PUT", "HEAD", "GET", "DELETE", "SigV4", "presigned-24h", "anonymous-denied", "wrong-key-denied"} <= set(
read_json(migration / name / "result.json", {}).get("operations", [])):
raise UpdateError("S3-Abnahme der Speichermigration fehlt: " + name)
if (not (migration / "backup/complete.json").is_file()
or file_sha(migration / "backup/seaweedfs.tar") != state.get("seaweedBackupSha256")):
raise UpdateError("Geprüfte Speichersicherung fehlt oder wurde verändert.")
self.job, self.override = migration, None
self.runner = self.runner_factory(migration / "adoption.log")
endpoint = self.runner.run(["docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"] ).strip()
if not os.environ.get("DOCKER_HOST", endpoint).startswith("unix://"):
raise UpdateError("Speicherübernahme muss direkt auf dem lokalen Docker-Host laufen.")
config = self.config(False)
self.project = config["name"]
required = CORE_SERVICES | {"objectstore"}
if not required <= set(config["services"]) or set(config["services"]) - required - LEGACY_MATRIX_SERVICES:
raise UpdateError("Speicherübernahme benötigt den Standard-Stack; nur bekannte alte Matrix-Dienste sind zusätzlich erlaubt.")
original = read_json(migration / "original.json")
record = {"schemaVersion": 1, "provider": "seaweedfs", "adoptedAt": now(),
"image": state["weedImage"], "network": state["network"],
"service": config["services"]["objectstore"],
"databaseSourceImage": original["services"]["db"]["image"],
"databaseImage": state["containers"]["db"]["image"],
"retainedMinio": {k: state["containers"]["minio"][k] for k in ("id", "image")},
"configSha256": {n: file_sha(self.root / "storage/config" / n) for n in ("s3.json", "start-seaweedfs.sh")}}
self.validate_storage(config, record)
for name in CORE_SERVICES:
cid = self.compose(False, "ps", "--quiet", name).strip()
actual = self.inspect(cid) if cid and "\n" not in cid else {}
if (actual.get("Image") != state["containers"][name]["image"]
or not actual.get("State", {}).get("Running")
or actual.get("State", {}).get("Health", {}).get("Status") not in (None, "healthy")):
raise UpdateError("Anwendung nach Speichermigration nicht gesund/unverändert: " + name)
self.compose(False, "exec", "-T", "backend", "node", "-e", S3_READ_CHECK, timeout=210)
if any(file_sha(self.root / n) != digest for n, digest in before.items()):
raise UpdateError("Konfiguration während Speicherübernahme verändert.")
# The old guard remains until both records are durable; retry after a crash is safe.
write_json(self.home / "storage-provider.json", record)
installed = read_json(self.home / "installed.json", {})
installed.update(composeSha256=before["docker-compose.yml"], storageProvider="seaweedfs")
write_json(self.home / "installed.json", installed)
write_json(migration / "updater-adoption.json", {"adoptedAt": record["adoptedAt"], "provider": "seaweedfs"})
write_json(self.home / "status.json", {"status": "recovered", "phase": "storage_adopted",
"updatedAt": now(), "recoveryRequired": False, "maintenance": False, "mutationStarted": False})
return {"status": "adopted", "storageProvider": "seaweedfs", "version": state["version"]}
def adapt_storage_candidate(self):
self.storage = self.storage_record()
self.storage_service = "objectstore" if self.storage else "minio"
self.storage_paths = ["storage"] if self.storage else ["minio"]
if not self.storage:
if "objectstore" in self.current["services"]:
raise UpdateError("SeaweedFS zuerst mit adopt-storage aus der abgeschlossenen Pilotmigration übernehmen.")
return
self.validate_storage(self.current, self.storage)
services = self.candidate["services"]
if set(services) != CORE_SERVICES | {"minio", "createbuckets"}:
raise UpdateError("Release verwendet einen unbekannten Speicheraufbau; Updater-Unterstützung erforderlich.")
if services["db"]["image"] != self.storage["databaseSourceImage"]:
raise UpdateError("Ein Wechsel der PostgreSQL-Version benötigt eine separate Datenbankmigration.")
services["db"]["image"] = self.storage["databaseImage"]
del services["minio"], services["createbuckets"]
services["objectstore"] = copy.deepcopy(self.storage["service"])
dependencies = services["backend"].setdefault("depends_on", {})
dependencies.pop("minio", None)
dependencies.pop("createbuckets", None)
dependencies["objectstore"] = {"condition": "service_healthy", "required": True}
services["backend"].setdefault("environment", {}).update(
AWS_REQUEST_CHECKSUM_CALCULATION="WHEN_REQUIRED", FEDEO_RUN_MIGRATIONS="false", FEDEO_STORAGE_PROVIDER="seaweedfs")
write_json(self.job / "target.compose.yml", compose_strings(self.candidate, escape=True))
self.candidate = self.config(True)
self.validate_storage(self.candidate, self.storage)
def prepare(self, version, *, allow_current=False):
for name in (".env", "docker-compose.yml"):
path = self.root / name
@@ -345,14 +603,51 @@ class Updater:
raise UpdateError(f"Release-Compose enthält kein eindeutiges {name}-Image.")
atomic(self.job / "target.compose.yml", compose)
atomic(self.job / "target.env", set_version((self.root / ".env").read_bytes(), self.target))
diff = "".join(difflib.unified_diff(original.decode().splitlines(True), compose.splitlines(True),
fromfile="installed/docker-compose.yml", tofile="release/docker-compose.yml"))
atomic(self.job / "compose.diff", diff)
self.candidate = self.config(True)
self.adapt_storage_candidate()
diff = "".join(difflib.unified_diff(original.decode().splitlines(True),
(self.job / "target.compose.yml").read_text().splitlines(True),
fromfile="installed/docker-compose.yml", tofile="release/docker-compose.yml"))
atomic(self.job / "compose.diff", diff)
self.validate_configs()
ids = self.compose(False, "ps", "--all", "--quiet").split()
# Compose ps can omit services removed from the current file. Retained
# Matrix/MinIO containers still belong to the project and its backup plan.
ids = self.runner.run(["docker", "ps", "--all", "--quiet", "--filter",
"label=com.docker.compose.project=" + self.project]).split()
self.containers = [self.inspect(container) for container in ids]
write_json(self.job / "original-containers.json", self.containers)
retained = {}
seen = set()
for container in self.containers:
labels = container.get("Config", {}).get("Labels", {})
name = labels.get("com.docker.compose.service")
if labels.get("com.docker.compose.oneoff", "False").lower() == "true":
if container.get("State", {}).get("Running"):
raise UpdateError("Ein Einmalcontainer läuft noch; dessen Arbeit zuerst abschließen.")
continue
if not name or name in seen:
raise UpdateError("Unbekannte oder doppelte Projektcontainer vorhanden.")
seen.add(name)
if name not in self.current["services"]:
if self.storage and name in ("minio", "createbuckets"):
continue # MinIO's exact identity and retirement are checked by validate_storage.
if name not in LEGACY_MATRIX_SERVICES:
raise UpdateError("Unbekannter Projektcontainer: " + name)
state, host = container.get("State", {}), container.get("HostConfig", {})
if (any(state.get(k) for k in ("Running", "Restarting", "Paused"))
or host.get("RestartPolicy", {}).get("Name") != "no" or host.get("AutoRemove")):
raise UpdateError("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"]})
elif mount["Type"] != "volume" or (name, mount["Destination"]) not in {
("matrix-redis", "/data"), ("matrix-turn", "/var/lib/coturn")}:
raise UpdateError("Abweichender Datenmount eines alten Matrix-Containers: " + name)
retained[name] = {"volumes": volumes}
self.current["services"].update(retained)
self.retired_services = sorted(set(self.retired_services) | set(retained))
self.validate_legacy_matrix()
self.retired_containers = [c for c in self.containers
if c["Config"]["Labels"].get("com.docker.compose.service") in self.retired_services]
if any(c.get("HostConfig", {}).get("AutoRemove") for c in self.retired_containers):
@@ -369,16 +664,22 @@ class Updater:
raise UpdateError("Ein Einmalcontainer läuft noch; dessen Arbeit zuerst abschließen.")
self.running = [c["Config"]["Labels"]["com.docker.compose.service"] for c in self.containers
if c["State"]["Running"] and c["Config"]["Labels"].get("com.docker.compose.oneoff", "False").lower() != "true"]
if not {"backend", "frontend", "db", "minio"}.issubset(self.running):
raise UpdateError("Backend, Frontend, Datenbank und MinIO müssen vor dem Update laufen.")
if not {"backend", "frontend", "db", self.storage_service}.issubset(self.running):
raise UpdateError("Backend, Frontend, Datenbank und Dateispeicher müssen vor dem Update laufen.")
if not set(self.running).issubset(self.current["services"]):
raise UpdateError("Nicht konfigurierte Container dieses Compose-Projekts laufen noch.")
self.before = {str(p): sha(p.read_bytes()) for p in
[self.root / ".env", self.root / "docker-compose.yml", *([self.override] if self.override else [])]}
if self.storage:
for path in [self.home / "storage-provider.json", *[self.root / "storage/config" / n for n in ("s3.json", "start-seaweedfs.sh")]]:
self.before[str(path)] = file_sha(path)
database_size = int(self.compose(False, "exec", "-T", "db", "sh", "-ec",
'exec psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -Atc "SELECT pg_database_size(current_database())"').strip())
minio_size = int(self.runner.run(["du", "-sb", str(self.root / "minio")]).split()[0])
self.backup_size = database_size * 2 + minio_size + GIB
storage_size = sum(int(self.runner.run(["du", "-sb", str(self.root / path)]).split()[0]) for path in self.storage_paths)
self.backup_size = database_size * 2 + storage_size + GIB
if self.storage:
image = json.loads(self.runner.run(["docker", "image", "inspect", self.storage["image"]]))[0]
self.backup_size += image["Size"] * 2
if self.matrix_backup_paths:
self.backup_size += int(self.runner.run(["du", "-sb", str(self.root / "matrix")]).split()[0])
self.check_space(self.backup_size, self.home)
@@ -387,6 +688,8 @@ class Updater:
"releaseNotes": str(self.job / "CHANGELOG.md"), "requiresComposeAdoption": not self.managed,
"estimatedBackupBytes": self.backup_size, "services": list(self.candidate["services"]),
"retiredServices": self.retired_services, "retainedDataPaths": self.matrix_backup_paths,
"storageProvider": "seaweedfs" if self.storage else "minio",
"storageImageRetained": bool(self.storage), "storageBackupPaths": self.storage_paths,
"updateAvailable": update_available,
# Kept in private job records; the service never exposes configuration fingerprints.
"approvalFingerprint": sha(json.dumps({"manifest": self.manifest, "configuration": self.before,
@@ -396,7 +699,7 @@ class Updater:
def validate_configs(self):
old, new = self.current["services"], self.candidate["services"]
required = {"backend", "frontend", "db", "minio", "createbuckets", "traefik", "node-exporter"}
required = CORE_SERVICES | ({"objectstore"} if self.storage else {"minio", "createbuckets"})
differences = []
for label, services in (("Bestehende Konfiguration", old), ("Zielkonfiguration", new)):
missing, additional = required - set(services), set(services) - required
@@ -410,7 +713,7 @@ class Updater:
if details:
differences.append(label + " (" + "; ".join(details) + ")")
if differences:
raise UpdateError("Dieser Updater unterstützt den Standard-Selfhost-Stack mit lokalen Daten und MinIO. "
raise UpdateError("Dieser Updater unterstützt den Standard-Selfhost-Stack mit lokalem Dateispeicher. "
+ ". ".join(differences) + ". Zusätzliche Dienste benötigen Unterstützung für Erhalt, Backup und Wiederherstellung.")
self.retired_services = sorted(set(old) & LEGACY_MATRIX_SERVICES)
self.matrix_backup_paths = []
@@ -431,7 +734,10 @@ class Updater:
raise UpdateError(f"Container-Identität von {name} würde verändert.")
if self.current.get("networks") != self.candidate.get("networks"):
raise UpdateError("Docker-Netzwerke würden verändert.")
for name, target, directory in (("db", "/var/lib/postgresql/data", "postgres"), ("minio", "/data", "minio")):
data_services = [("db", "/var/lib/postgresql/data", "postgres")]
if not self.storage:
data_services.append(("minio", "/data", "minio"))
for name, target, directory in data_services:
volumes = new[name].get("volumes", [])
if len(volumes) != 1 or volumes[0].get("type") != "bind" or volumes[0].get("target") != target or volumes[0].get("source") != str(self.root / directory):
raise UpdateError(f"{name} benötigt das lokale Standard-Datenverzeichnis.")
@@ -448,7 +754,7 @@ class Updater:
healthcheck = new[name].get("healthcheck", {})
if new[name]["image"] != self.images[name] or healthcheck.get("disable") or not healthcheck.get("test") or healthcheck["test"] == ["NONE"]:
raise UpdateError(f"{name}: Release-Digest oder Healthcheck wurde überschrieben.")
for name in ("minio", "createbuckets"):
for name in (() if self.storage else ("minio", "createbuckets")):
if not re.search(r"@sha256:[a-f0-9]{64}$", new[name]["image"]):
raise UpdateError("MinIO und mc müssen auf feste Digests gepinnt sein.")
env = new["backend"].get("environment", {})
@@ -522,13 +828,21 @@ class Updater:
atomic(directory / "config" / name, (self.root / name).read_bytes())
if self.override:
atomic(directory / "config" / self.override.name, self.override.read_bytes())
if self.storage:
self.validate_storage(self.current, self.storage, running=False)
cid = self.compose(False, "ps", "--all", "--quiet", "objectstore").strip()
store_state = self.inspect(cid).get("State", {}) if cid else {"Running": True}
if any(store_state.get(k) for k in ("Running", "Restarting", "Paused")):
raise UpdateError("SeaweedFS muss vor der kalten Sicherung gestoppt sein.")
atomic(directory / "config/.updates/storage-provider.json", (self.home / "storage-provider.json").read_bytes())
self.runner.run(["docker", "image", "save", "-o", str(directory / "storage-image.tar"), self.storage["image"]], timeout=3600)
write_json(directory / "containers.json", self.containers)
self.compose(False, "exec", "-T", "db", "sh", "-ec",
'exec pg_dump -U "$POSTGRES_USER" -d "$POSTGRES_DB" --format=custom', output=directory / "database.dump")
self.compose(False, "exec", "-T", "db", "pg_restore", "--list", input_file=directory / "database.dump")
self.compose(False, "exec", "-T", "db", "sh", "-ec",
'exec pg_dumpall -U "$POSTGRES_USER" --globals-only', output=directory / "globals.sql")
paths = ["minio", *self.matrix_backup_paths]
paths = [*self.storage_paths, *self.matrix_backup_paths]
if (self.root / "traefik/letsencrypt").is_dir():
paths.append("traefik/letsencrypt")
self.runner.run(["tar", "-cpf", str(directory / "files.tar"), "-C", str(self.root), *paths], timeout=7200)
@@ -543,7 +857,13 @@ class Updater:
os.fsync(stream.fileno())
checksums[str(path.relative_to(directory))] = digest.hexdigest()
self.verify_matrix_stopped()
if self.storage:
cid = self.compose(False, "ps", "--all", "--quiet", "objectstore").strip()
if not cid or any(self.inspect(cid).get("State", {}).get(k) for k in ("Running", "Restarting", "Paused")):
raise UpdateError("SeaweedFS wurde während der kalten Sicherung gestartet.")
write_json(directory / "complete.json", {"createdAt": now(), "database": "database.dump", "files": "files.tar",
"storageProvider": "seaweedfs" if self.storage else "minio", "storageDataPaths": self.storage_paths,
"storageImage": self.storage["image"] if self.storage else None,
"retiredServices": self.retired_services, "retainedContainerIds": self.state["retiredContainerIds"],
"matrixDataPaths": self.matrix_backup_paths, "sha256": checksums})
self.save("backed_up", backup=str(directory))
@@ -570,12 +890,18 @@ class Updater:
# Mutable infrastructure tags must still be refreshed before entering maintenance.
for policy in ("missing", "always"):
services = [name for name, service in self.candidate["services"].items()
if bool(re.search(r"@sha256:[a-f0-9]{64}$", service["image"])) == (policy == "missing")]
if not IMAGE_ID.fullmatch(service["image"])
and bool(re.search(r"@sha256:[a-f0-9]{64}$", service["image"])) == (policy == "missing")]
if services:
self.compose(True, "pull", "--policy", policy, *services, timeout=3600)
for service in self.candidate["services"].values():
if IMAGE_ID.fullmatch(service["image"]):
self.runner.run(["docker", "image", "inspect", service["image"]])
self.check_space(self.backup_size, self.home)
if any(sha(Path(path).read_bytes()) != value for path, value in self.before.items()):
raise UpdateError("Konfiguration wurde während der Vorbereitung verändert; erneut prüfen.")
if self.storage:
self.validate_storage(self.current, self.storage)
self.save("maintenance", recoveryRequired=True, maintenance=True, runningServices=self.running)
# Stop every writer and public entry point. The database stays up for pg_dump.
self.compose(False, "stop", "--timeout", "60", *[s for s in self.running if s != "db"])
@@ -596,8 +922,16 @@ class Updater:
atomic(self.root / ".env.example", (self.job / ".env.example").read_bytes())
atomic(self.root / "scripts/selfhost-setup.sh", (self.job / "selfhost-setup.sh").read_bytes())
os.chmod(self.root / "scripts/selfhost-setup.sh", 0o700)
self.compose(True, "up", "--pull", "never", "-d", "--no-deps", "--wait", "--wait-timeout", "180", "db", "minio", "node-exporter")
self.compose(True, "run", "--pull", "never", "--rm", "--no-deps", "createbuckets")
self.compose(True, "up", "--pull", "never", "-d", "--no-deps", "--wait", "--wait-timeout", "180", "db", self.storage_service, "node-exporter")
if self.storage:
self.save("checking_storage", storageCheckContainer="fedeo-storage-check-" + self.job.name,
storageProbeKey=".fedeo-update-check/" + uuid.uuid4().hex + ".pdf")
self.compose(True, "run", "--pull", "never", "--rm", "--no-deps", "--name", self.state["storageCheckContainer"],
"-e", "FEDEO_UPDATE_PROBE_KEY=" + self.state["storageProbeKey"],
"--entrypoint", "node", "backend", "-e", S3_UPGRADE_CHECK, timeout=210)
self.save(storageProbeKey=None)
else:
self.compose(True, "run", "--pull", "never", "--rm", "--no-deps", "createbuckets")
self.save("migrating", migrationContainer="fedeo-update-" + self.job.name)
self.compose(True, "run", "--pull", "never", "--rm", "--no-deps", "--name", self.state["migrationContainer"],
"-e", "FEDEO_RUN_MIGRATIONS=false", "--entrypoint", "npm", "backend", "run", "migrate")
@@ -605,6 +939,8 @@ class Updater:
self.compose(True, "up", "--pull", "never", "-d", "--wait", "--wait-timeout", "600", timeout=900)
self.save("verifying")
self.verify_matrix_stopped(retired=True)
if self.storage:
self.validate_storage(self.candidate, self.storage)
for name, image in self.images.items():
container_id = self.compose(True, "ps", "--quiet", name).strip()
container = self.inspect(container_id)
@@ -618,6 +954,8 @@ class Updater:
def execute(self, command, version=None, accept_compose=False, *, expected_fingerprint=None, request_id=None):
with self.lock():
previous = read_json(self.home / "status.json", {})
if previous.get("phase") == "storage_migration":
raise UpdateError("Speichermigration prüfen und nach vollständigem Abschluss adopt-storage ausführen; kein resolve verwenden.")
if previous.get("recoveryRequired"):
raise UpdateError("Ein vorheriges Update benötigt Wiederherstellung. status und docs/selfhost-updates.md prüfen; danach resolve ausführen.")
self.job = self.home / (datetime.datetime.now().strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:8])
@@ -657,6 +995,8 @@ class Updater:
raise UpdateError("Die manuell wiederhergestellte Version mit --version angeben.")
with self.lock():
state = read_json(self.home / "status.json", {})
if state.get("phase") == "storage_migration":
raise UpdateError("Speichermigration kann nicht mit resolve freigegeben werden; Status prüfen und nach Abschluss adopt-storage verwenden.")
if not state.get("recoveryRequired"):
raise UpdateError("Kein offener Wiederherstellungsfall.")
self.job = Path(state["job"])
@@ -664,11 +1004,16 @@ class Updater:
self.runner = self.runner_factory(self.job / "commands.log")
self.override = next((self.root / name for name in ("docker-compose.override.yml", "docker-compose.override.yaml") if (self.root / name).exists()), None)
current = self.config(False)
self.storage = self.storage_record()
self.storage_service = "objectstore" if self.storage else "minio"
if self.storage:
self.validate_storage(current, self.storage)
self.public = current["services"]["frontend"]["environment"]["NUXT_PUBLIC_API_BASE"].removesuffix("/backend")
if state.get("migrationContainer"):
remaining = self.runner.run(["docker", "ps", "--all", "--quiet", "--filter", "name=^/" + state["migrationContainer"] + "$"]).strip()
if remaining:
raise UpdateError("Den verbliebenen Migrationscontainer zuerst prüfen und manuell entfernen.")
for key in ("migrationContainer", "storageCheckContainer"):
if state.get(key):
remaining = self.runner.run(["docker", "ps", "--all", "--quiet", "--filter", "name=^/" + state[key] + "$"]).strip()
if remaining:
raise UpdateError("Den verbliebenen Prüf-/Migrationscontainer zuerst prüfen und manuell entfernen.")
# Older releases have /version and /health, but no frontend health endpoint.
version_info = self.http.json(self.public + "/backend/version")
if version_info.get("product") != "FEDEO" or version_info.get("version") != version:
@@ -676,10 +1021,16 @@ class Updater:
if self.http.json(self.public + "/backend/health").get("status") != "ok":
raise UpdateError("Das wiederhergestellte Backend ist nicht erreichbar.")
self.http.get(self.public + "/")
if state.get("storageProbeKey"):
if not self.storage or not re.fullmatch(r"\.fedeo-update-check/[a-f0-9]{32}\.pdf", state["storageProbeKey"]):
raise UpdateError("Ungültiger protokollierter S3-Prüfschlüssel; manuell prüfen.")
self.compose(False, "exec", "-T", "-e", "FEDEO_UPDATE_PROBE_KEY=" + state["storageProbeKey"],
"backend", "node", "-e", S3_CLEANUP_CHECK, timeout=90)
self.save(storageProbeKey=None)
original_images = {c["Config"]["Labels"]["com.docker.compose.service"]: c["Image"]
for c in read_json(self.job / "original-containers.json", [])}
manifest = read_json(self.job / "release-manifest.json")
for name in ("backend", "frontend", "db", "minio"):
for name in ("backend", "frontend", "db", self.storage_service):
container_id = self.compose(False, "ps", "--quiet", name).strip()
container = self.inspect(container_id) if container_id else {}
container_state = container.get("State", {})
@@ -702,9 +1053,7 @@ class Updater:
for original in read_json(self.job / "original-containers.json", []):
if original.get("Id") not in state.get("retiredContainerIds", []):
continue
name = original["Config"]["Labels"]["com.docker.compose.service"]
container_id = self.compose(False, "ps", "--all", "--quiet", name).strip()
restored = self.inspect(container_id) if container_id else {}
restored = self.inspect(original["Id"])
if (restored.get("Image") != original["Image"]
or restored.get("State", {}).get("Running") != original["State"]["Running"]
or (original["State"]["Running"] and restored.get("State", {}).get("Health", {}).get("Status") not in (None, "healthy"))
@@ -724,7 +1073,7 @@ def main(argv=None):
print("Python ab Version 3.10 wird benötigt.", file=sys.stderr)
return 1
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("command", choices=("check", "apply", "status", "resolve"))
parser.add_argument("command", choices=("check", "apply", "status", "resolve", "adopt-storage"))
parser.add_argument("--directory", default=os.environ.get("FEDEO_DIR", "/opt/fedeo"))
parser.add_argument("--version", help="Zielversion, z. B. 3.1.0; check wählt sonst den neuesten stabilen Release")
parser.add_argument("--accept-compose-replacement", action="store_true", help="Zuvor geprüften Austausch einer nicht verwalteten Compose-Datei akzeptieren")
@@ -739,6 +1088,8 @@ def main(argv=None):
result = updater.status()
elif args.command == "resolve":
result = updater.resolve(args.version)
elif args.command == "adopt-storage":
result = updater.adopt_storage()
else:
result = updater.execute(args.command, args.version, args.accept_compose_replacement)
print(json.dumps(result, ensure_ascii=False, indent=2))