# -*- coding: utf-8 -*- """ source.py — Orchestrator режима SOURCE (подготовка к переносу на старом сервере) Строгая защита: каждый шаг в try/except, state сохраняется при ошибке. """ import os import json import tarfile import time import sys from datetime import datetime from core.color import header, subheader, success, warn, error as cerror, info, step, prompt, prompt_password, confirm, divider, log_cmd from core import state from core.runner import run from discover.docker import discover_docker, get_container_pid from discover.nginx import discover_nginx, get_nginx_systemd_unit from discover.network import find_sidecar_processes, gather_host_network_info, get_process_details, find_systemd_units_related, gather_cron_jobs from manifest.manifest import build_manifest, save_manifest, review_manifest _ARCHIVE_DIR = "/tmp/docker-migrate-archives" # Сервисы, которые НЕЛЬЗЯ останавливать/включать в архив — они управляют средой выполнения SYSTEM_CRITICAL = { "docker.service", "containerd.service", "docker.socket", "systemd-journald.service", "systemd-networkd.service", "systemd-timesyncd.service", "systemd-resolved.service", "systemd-logind.service", "ssh.service", "sshd.service", } def run_source_mode(): state.reset_state(mode="source") from core.fsm import FSM fsm = FSM(mode="source") try: fsm.resume_from("INIT") except Exception as e: cerror(f"Source-режим прерван: {e}") sys.exit(1) def do_discovery(): step(1, "АВТО-ПОИСК ЗАВИСИМОСТЕЙ") state.set_stage("SOURCE_DISCOVER") service_name = prompt("Введите имя Docker-сервиса или контейнера (например, marzban)") if not service_name: raise RuntimeError("Имя сервиса не указано") try: docker_data = discover_docker(service_name) except RuntimeError: # Если exact не найден, discover_docker уже предложил список raise except Exception as e: raise RuntimeError(f"Ошибка поиска Docker: {e}") cid = docker_data.get("container_id") if not cid: raise RuntimeError("Не удалось определить ID контейнера") # Подсказки для nginx ports = list(docker_data.get("ports", {}).keys()) ports_list = [] for p in ports: try: port_num = p.split(":")[-1] ports_list.append(int(port_num)) except Exception: pass domain_hints = [] labels = docker_data.get("labels", {}) for k, v in labels.items(): if "DOMAIN" in k.upper() or "HOST" in k.upper(): domain_hints.append(str(v)) # Также из Env контейнера env_list = docker_data.get("env", []) for e in env_list: if "DOMAIN" in e.upper() or "HOST" in e.upper(): try: val = e.split("=", 1)[1] if val and "." in val: domain_hints.append(val) except Exception: pass # Sidecar discovery (не fatal если не получилось) sidecars = [] try: sidecars = find_sidecar_processes(cid, ports_list) for s in sidecars: if s.get("host_pid"): try: details = get_process_details(s["host_pid"]) s["details"] = details except Exception: warn(f"Не удалось получить детали процесса {s.get('host_pid')}") if sidecars: success(f"Найдено sidecar/loopback зависимостей: {len(sidecars)}") except Exception as e: warn(f"Sidecar discovery не удался: {e}. Продолжаем без sidecar.") # Nginx nginx_data = [] try: nginx_data = discover_nginx(service_ports=ports_list, service_domain_hints=domain_hints) if nginx_data: success(f"Найдено связанных nginx конфигов: {len(nginx_data)}") except Exception as e: warn(f"Nginx discovery не удался: {e}. Продолжаем без nginx.") # Systemd sidecar_procs = [{"pid": s["host_pid"], "process": s["host_process"]} for s in sidecars if s.get("host_pid")] systemd_units = [] try: systemd_units = find_systemd_units_related(sidecar_procs) nginx_units = get_nginx_systemd_unit() for nu in nginx_units: systemd_units.append({ "name": nu["unit"], "path": nu["path"], "related_to": "nginx", }) if systemd_units: success(f"Найдено systemd units: {len(systemd_units)}") except Exception as e: warn(f"Systemd discovery не удался: {e}") # Cron hints = [docker_data.get("container_name", "")] + [s.get("host_process", "") for s in sidecars] cron_jobs = [] try: cron_jobs = gather_cron_jobs(user_hint=hints) if cron_jobs: success(f"Найдено cron заданий: {len(cron_jobs)}") except Exception as e: warn(f"Cron discovery не удался: {e}") # Host network host_network = {} try: host_network = gather_host_network_info() except Exception as e: warn(f"Сетевая информация не собрана: {e}") extra_hints = [] manifest = build_manifest( docker_data=docker_data, nginx_data=nginx_data, sidecars=sidecars, host_network=host_network, systemd_units=systemd_units, cron_jobs=cron_jobs, extra_hints=extra_hints, ) os.makedirs(_ARCHIVE_DIR, exist_ok=True) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") safe_name = service_name.replace("/", "_").replace(":", "_") manifest_path = os.path.join(_ARCHIVE_DIR, f"{safe_name}_{stamp}_manifest.json") save_manifest(manifest, manifest_path) state.set_stage("SOURCE_MANIFEST_REVIEW", manifest_path=manifest_path) info(f"Этап discovery завершён. Manifest: {manifest_path}") # review вызывается отдельным шагом FSM, не тут def do_manifest_review(): manifest_path = state.load_state().get("manifest_path") if not manifest_path or not os.path.isfile(manifest_path): raise RuntimeError("Manifest не найден. Сначала выполните discovery.") from manifest.manifest import load_manifest manifest = load_manifest(manifest_path) review_manifest(manifest) if not confirm("Подтвердить и продолжить сборку архива", default="y"): raise RuntimeError("Прервано пользователем") def do_pack(): step(2, "АРХИВИРОВАНИЕ") st = state.load_state() manifest_path = st.get("manifest_path") manifest = json.load(open(manifest_path, "r", encoding="utf-8")) svc_name = manifest["service"]["name"] stamp = datetime.now().strftime("%Y%m%d_%H%M%S") archive_name = f"{svc_name}_{stamp}_migrate.tar.gz" archive_path = os.path.join(_ARCHIVE_DIR, archive_name) files_to_pack = set() # compose cf = manifest["docker"].get("compose_file") if cf and os.path.isfile(cf): files_to_pack.add(cf) # env ef = manifest["docker"].get("env_file") if ef and os.path.isfile(ef): files_to_pack.add(ef) # bind mounts for m in manifest["docker"].get("mounts", []): if m["type"] == "bind": src = m["source"] if os.path.isdir(src) or os.path.isfile(src): files_to_pack.add(src) elif m["type"] == "volume": vol_name = m["source"] info(f"Архивируем Docker volume: {vol_name} ...") try: run(f"docker run --rm -v {vol_name}:/data -v {_ARCHIVE_DIR}:/out alpine tar czf /out/vol_{vol_name}_{stamp}.tar.gz -C /data .", check=False) files_to_pack.add(f"vol_{vol_name}_{stamp}.tar.gz") success(f"Volume {vol_name} заархивирован") except Exception as e: warn(f"Не удалось архивировать volume {vol_name}: {e}") # nginx configs + SSL for n in manifest.get("nginx", []): f = n.get("file") if f and os.path.isfile(f): files_to_pack.add(f) cert = n.get("ssl_certificate") key = n.get("ssl_certificate_key") if cert and os.path.isfile(cert): files_to_pack.add(cert) if key and os.path.isfile(key): files_to_pack.add(key) # sidecar configs for s in manifest.get("sidecars", []): details = s.get("details", {}) for f in details.get("files", []): if os.path.isfile(f): # Пропускаем runtime/system пути — они не относятся к сервису if any(f.startswith(p) for p in ( "/proc/", "/sys/", "/dev/", "/var/lib/containerd/", "/var/lib/docker/", "/run/containerd/", "/run/docker/", )): continue files_to_pack.add(f) unit = details.get("unit") if unit and unit not in SYSTEM_CRITICAL: try: uout = run(f"systemctl show {unit} -p FragmentPath --value", check=False) path = uout.stdout.strip() if path and os.path.isfile(path): files_to_pack.add(path) except Exception: pass # systemd units for u in manifest.get("systemd_units", []): uname = u.get("name") if uname and uname in SYSTEM_CRITICAL: continue p = u.get("path") if p and os.path.isfile(p): files_to_pack.add(p) # cron files for c in manifest.get("cron_jobs", []): f = c.get("file") if f and os.path.isfile(f): files_to_pack.add(f) # host_network snapshot netfile = os.path.join(_ARCHIVE_DIR, f"{svc_name}_{stamp}_network.json") try: with open(netfile, "w", encoding="utf-8") as f: json.dump(manifest.get("host_network", {}), f, indent=2, ensure_ascii=False) files_to_pack.add(netfile) except Exception as e: warn(f"Не удалось сохранить snapshot сети: {e}") # manifest itself files_to_pack.add(manifest_path) # Build tar.gz — сохраняем пути относительно корня (без начального /) info(f"Создаём архив: {archive_path}") with tarfile.open(archive_path, "w:gz") as tar: for fp in files_to_pack: if os.path.exists(fp): arcname = fp.lstrip("/") tar.add(fp, arcname=arcname) info(f" + {arcname}") else: warn(f" ! Не найден файл для архива: {fp}") success(f"Архив создан: {archive_path}") state.set_stage("SOURCE_PACK", archive_path=archive_path) # Автопереход к следующему шагу делает FSM def do_stop_service(): step(3, "ОСТАНОВКА СЕРВИСА (опционально)") st = state.load_state() manifest_path = st.get("manifest_path") manifest = json.load(open(manifest_path, "r", encoding="utf-8")) svc_name = manifest["service"]["name"] if confirm(f"Остановить Docker-сервис '{svc_name}' перед переносом", default="n"): cid = manifest["docker"].get("container_id") if cid: try: run(f"docker stop {cid} -t 30", check=False) success(f"Контейнер {cid[:12]} остановлен") except Exception as e: warn(f"Не удалось остановить контейнер: {e}") # Дедупликация + фильтр system-critical stopped = set() for u in manifest.get("systemd_units", []): uname = u.get("name") if not uname or uname in stopped: continue stopped.add(uname) if uname in SYSTEM_CRITICAL: warn(f"Пропуск system-critical unit: {uname}") continue try: run(f"systemctl stop {uname}", check=False) info(f"Остановлен systemd unit: {uname}") except Exception: pass else: info("Пропуск остановки (сервис продолжает работать)") state.set_stage("SOURCE_STOP") def do_transfer_offer(): step(4, "ПЕРЕНОС НА НОВЫЙ СЕРВЕР (опционально)") st = state.load_state() # При resume: проверяем, были ли уже введены target-параметры resume_host = st.get("target_host") resume_user = st.get("target_user") resume_port = st.get("target_port", 22) if resume_host and resume_host not in ("", "None"): info(f"Обнаружены ранее введённые параметры target: {resume_user}@{resume_host}:{resume_port}") if confirm("Использовать эти параметры повторно (если ошибка была в SSH-ключе, выберите Y, после фикса)", default="y"): host, user, port_int = resume_host, resume_user or "root", int(resume_port) else: # Сбрасываем, чтобы запросить заново state.set_stage("SOURCE_STOP", target_host=None, target_user=None, target_port=None, ssh_key=None) host = user = None else: host = user = None if not host: if not confirm("Перенести архив на новый сервер сейчас", default="y"): success("Готово! Архив оставлен на текущем сервере.") success(f"Manifest и архив лежат в: {_ARCHIVE_DIR}") state.set_stage("DONE") return host = prompt("IP или домен нового сервера") if not host: warn("IP/домен не указан. Перенос пропущен.") success(f"Архив оставлен в: {_ARCHIVE_DIR}") state.set_stage("DONE") return user = prompt("SSH user (root или обычный пользователь)") if not user: user = "root" port = prompt("SSH порт (Enter=22)") or "22" port_int = int(port) # SSH-диагностика: ищем ключи, проверяем доступ from transfer.ssh import pick_or_setup_ssh_key key_path = None try: key_path = pick_or_setup_ssh_key(host, user, port_int) except RuntimeError as exc: # pick_or_setup уже дал инструкции, помечаем ошибку и выходим state.set_error("ssh_key_setup", "", str(exc), suggestion="Настройте SSH-ключ и запустите: docker-migrate --resume") raise if key_path: state.set_stage("TRANSFER", target_host=host, target_user=user, target_port=port_int, ssh_key=key_path) else: # Нет SSH-ключа — запрашиваем пароль от TARGET сервера password = prompt_password(f"Пароль от target-сервера {user}@{host} (Enter для отмены)") if not password: state.set_error("ssh_key_setup", "", "Пароль не введён", suggestion="Запустите docker-migrate --resume или настройте SSH-ключ на target") raise RuntimeError("Пароль от target не введён.") from transfer.ssh import ensure_sshpass if not ensure_sshpass(): state.set_error("ssh_key_setup", "", "sshpass не найден", suggestion="apt-get install -y sshpass") raise RuntimeError("sshpass не установлен.") state.set_stage("TRANSFER", target_host=host, target_user=user, target_port=port_int, ssh_key=None, ssh_password=password) from transfer.transfer import do_transfer do_transfer()