1560 lines
63 KiB
Python
1560 lines
63 KiB
Python
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import shlex
|
||
import subprocess
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from io import StringIO
|
||
from pathlib import Path
|
||
|
||
try:
|
||
import paramiko
|
||
except ImportError: # pragma: no cover
|
||
paramiko = None
|
||
|
||
from app.core.config import settings
|
||
from app.services.ops_agent_service import append_ops_job_event
|
||
from app.services.ops_job_service import create_ops_job, dispatch_ops_job, list_managed_nodes
|
||
|
||
|
||
_MIGRATION_TIMEOUT_SECONDS = 180
|
||
_SSH_CONNECT_TIMEOUT_SECONDS = 12
|
||
_MIGRATION_CONFIRMATION_TTL_SECONDS = 1800
|
||
_MIGRATION_ACTION = "migration.execute"
|
||
_MIGRATION_JOB_CONTEXTS: dict[int, dict] = {}
|
||
_MIGRATION_JOB_CONTEXTS_LOCK = threading.Lock()
|
||
|
||
|
||
def get_ops_migration_source_profile() -> dict:
|
||
config_files = {
|
||
"domain_env": str(Path(settings.domain_root) / ".env"),
|
||
"api_env": "/etc/default/domaincheck-api",
|
||
"worker_env": "/etc/default/domaincheck-worker",
|
||
"ops_center_env": "/etc/default/domaincheck-ops-center",
|
||
"web_env_production": str(Path(settings.domain_root).resolve().parent / "domain-web" / ".env.production"),
|
||
"api_service": "/etc/systemd/system/domaincheck-api.service",
|
||
"worker_service": "/etc/systemd/system/domaincheck-worker.service",
|
||
"sync_agent_service": "/etc/systemd/system/domaincheck-sync-agent.service",
|
||
"node_agent_service": "/etc/systemd/system/domaincheck-node-agent.service",
|
||
"postgresql_service": "/etc/systemd/system/postgresql.service",
|
||
}
|
||
file_status = {
|
||
key: {
|
||
"path": path,
|
||
"exists": Path(path).exists(),
|
||
}
|
||
for key, path in config_files.items()
|
||
}
|
||
return {
|
||
"source_node": {
|
||
"node_code": settings.node_code,
|
||
"region": settings.node_region,
|
||
"role": settings.node_role,
|
||
"worker_mode": settings.worker_mode,
|
||
},
|
||
"workspace": {
|
||
"repo_path": str(Path(settings.domain_root).resolve().parent),
|
||
"domain_root": str(Path(settings.domain_root).resolve()),
|
||
"api_root": str((Path(settings.domain_root).resolve().parent / "domain-api").resolve()),
|
||
"web_root": str((Path(settings.domain_root).resolve().parent / "domain-web").resolve()),
|
||
},
|
||
"database": {
|
||
"host": settings.db_host,
|
||
"port": int(settings.db_port),
|
||
"database": settings.db_database,
|
||
"user": settings.db_user,
|
||
"password_configured": bool(str(settings.db_password or "").strip()),
|
||
},
|
||
"redis": {
|
||
"host": settings.redis_host,
|
||
"port": int(settings.redis_port),
|
||
"db": int(settings.redis_db),
|
||
"password_configured": bool(str(settings.redis_password or "").strip()),
|
||
},
|
||
"config_files": file_status,
|
||
}
|
||
|
||
|
||
def preview_ops_migration(payload: dict | None = None) -> tuple[bool, str, dict]:
|
||
normalized = _normalize_payload(payload or {})
|
||
ok, message, context = _build_migration_context(normalized, include_remote_file_reads=True)
|
||
if not ok:
|
||
return False, message, _sanitize_migration_context(context)
|
||
return True, "迁移预检查完成", _sanitize_migration_context(context)
|
||
|
||
|
||
def execute_ops_migration(payload: dict | None = None) -> tuple[bool, str, dict]:
|
||
normalized = _normalize_payload(payload or {})
|
||
ok, message, context = _build_migration_context(normalized, include_remote_file_reads=True)
|
||
if not ok:
|
||
return False, message, _sanitize_migration_context(context)
|
||
confirmation_ok, confirmation_message, confirmation_data = _validate_execution_confirmation(normalized, context)
|
||
if not confirmation_ok:
|
||
return False, confirmation_message, _sanitize_migration_context({**context, **confirmation_data})
|
||
|
||
job_payload = _build_migration_job_payload(normalized)
|
||
metadata = {
|
||
"job_kind": "ops_migration",
|
||
"source_node_code": str(context.get("source_profile", {}).get("source_node", {}).get("node_code") or ""),
|
||
"plan_steps": list(context.get("plan_steps") or []),
|
||
"target_paths": dict(context.get("target_paths") or {}),
|
||
"target_db": {
|
||
"host": str(context.get("target_db_config", {}).get("host") or ""),
|
||
"port": int(context.get("target_db_config", {}).get("port") or 0),
|
||
"database": str(context.get("target_db_config", {}).get("database") or ""),
|
||
"user": str(context.get("target_db_config", {}).get("user") or ""),
|
||
"password_configured": bool(context.get("target_db_config", {}).get("password_configured", False)),
|
||
},
|
||
"execution_guard": {
|
||
"context_hash": str(context.get("execution_guard", {}).get("context_hash") or ""),
|
||
"expires_at": str(context.get("execution_guard", {}).get("expires_at") or ""),
|
||
"requires_confirmation": bool(context.get("execution_guard", {}).get("requires_confirmation", False)),
|
||
},
|
||
}
|
||
create_ok, create_message, create_data = create_ops_job(
|
||
{
|
||
"action": _MIGRATION_ACTION,
|
||
"target_type": "node",
|
||
"target_node_code": str(normalized.get("target_node_code") or ""),
|
||
"requested_by": "migration-ui",
|
||
"execution_mode": "control-plane",
|
||
"auto_approve": True,
|
||
"run_now": False,
|
||
"payload": job_payload,
|
||
"metadata": metadata,
|
||
}
|
||
)
|
||
if not create_ok:
|
||
return False, create_message, _sanitize_migration_context({**context, **dict(create_data or {})})
|
||
|
||
job = dict(create_data.get("job") or {})
|
||
job_id = int(job.get("id") or 0)
|
||
if job_id <= 0:
|
||
return False, "迁移任务创建成功，但缺少 job_id。", _sanitize_migration_context({**context, "job": job})
|
||
|
||
_remember_migration_job_context(job_id, normalized, context)
|
||
_start_migration_dispatch_thread(job_id)
|
||
return True, "迁移任务已创建，后台开始执行。", {
|
||
**_sanitize_migration_context(context),
|
||
"job": job,
|
||
"job_id": job_id,
|
||
"executed_immediately": False,
|
||
"execution_steps": [],
|
||
}
|
||
|
||
|
||
def execute_ops_migration_job(
|
||
*,
|
||
job_id: int,
|
||
target_node_code: str,
|
||
payload: dict | None = None,
|
||
requested_by: str = "api",
|
||
metadata: dict | None = None,
|
||
) -> tuple[bool, str, dict]:
|
||
cached = _pop_migration_job_context(job_id)
|
||
normalized = dict(cached.get("payload") or {})
|
||
if not normalized:
|
||
normalized = _normalize_payload(payload or {})
|
||
if target_node_code and not str(normalized.get("target_node_code") or "").strip():
|
||
normalized["target_node_code"] = str(target_node_code or "").strip()
|
||
context = dict(cached.get("context") or {})
|
||
if not context:
|
||
ok, message, context = _build_migration_context(normalized, include_remote_file_reads=True)
|
||
if not ok:
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(normalized.get("target_node_code") or ""),
|
||
event_type="migration_prepare_failed",
|
||
level="error",
|
||
message=message,
|
||
payload={"blocking_reasons": list(context.get("blocking_reasons") or [])},
|
||
)
|
||
return False, message, _sanitize_migration_context(context)
|
||
return _run_migration_plan(
|
||
normalized,
|
||
context,
|
||
job_id=job_id,
|
||
requested_by=requested_by,
|
||
metadata=dict(metadata or {}),
|
||
)
|
||
|
||
|
||
def _normalize_payload(payload: dict) -> dict:
|
||
repo_path = str(payload.get("target_repo_path") or "").strip() or "/www/wwwroot/getDomain"
|
||
domain_root = str(payload.get("target_domain_root") or "").strip() or "/opt/domaincheck/domainCheck"
|
||
api_root = str(payload.get("target_api_root") or "").strip() or "/opt/domaincheck/domain-api"
|
||
web_root = str(payload.get("target_web_root") or "").strip() or "/opt/domaincheck/domain-web"
|
||
return {
|
||
"target_node_code": str(payload.get("target_node_code") or "").strip(),
|
||
"target_repo_path": repo_path,
|
||
"target_domain_root": domain_root,
|
||
"target_api_root": api_root,
|
||
"target_web_root": web_root,
|
||
"sync_env_files": bool(payload.get("sync_env_files", True)),
|
||
"sync_systemd_units": bool(payload.get("sync_systemd_units", True)),
|
||
"build_frontend": bool(payload.get("build_frontend", True)),
|
||
"overwrite_database": bool(payload.get("overwrite_database", False)),
|
||
"backup_target_database": bool(payload.get("backup_target_database", True)),
|
||
"restart_services": bool(payload.get("restart_services", True)),
|
||
"target_db_host": str(payload.get("target_db_host") or "").strip(),
|
||
"target_db_port": int(payload.get("target_db_port") or 0),
|
||
"target_db_name": str(payload.get("target_db_name") or "").strip(),
|
||
"target_db_user": str(payload.get("target_db_user") or "").strip(),
|
||
"target_db_password": str(payload.get("target_db_password") or ""),
|
||
"execute_confirmation_token": str(payload.get("execute_confirmation_token") or "").strip(),
|
||
"execute_confirmation_text": str(payload.get("execute_confirmation_text") or "").strip(),
|
||
}
|
||
|
||
|
||
def _build_migration_context(payload: dict, *, include_remote_file_reads: bool) -> tuple[bool, str, dict]:
|
||
target_node_code = str(payload.get("target_node_code") or "").strip()
|
||
if not target_node_code:
|
||
return False, "缺少目标节点编码", {"blocking_reasons": ["target_node_code 不能为空"]}
|
||
|
||
node = _find_managed_node(target_node_code)
|
||
if not node:
|
||
return False, "目标节点不存在或未纳管", {"blocking_reasons": [f"未找到托管节点 {target_node_code}"]}
|
||
|
||
ssh_ready = bool(str(node.get("ssh_host") or "").strip() and str(node.get("ssh_user") or "").strip())
|
||
if not ssh_ready:
|
||
return False, "目标节点 SSH 信息不完整", {
|
||
"target_node": _serialize_target_node(node),
|
||
"blocking_reasons": ["目标节点缺少 SSH Host 或 SSH User"],
|
||
}
|
||
|
||
remote_checks = _collect_remote_checks(node, payload, include_file_reads=include_remote_file_reads)
|
||
source_profile = get_ops_migration_source_profile()
|
||
remote_db_config = dict(remote_checks.get("remote_db_config") or {})
|
||
|
||
target_db_config = {
|
||
"host": str(payload.get("target_db_host") or remote_db_config.get("DB_HOST") or "127.0.0.1").strip() or "127.0.0.1",
|
||
"port": int(payload.get("target_db_port") or remote_db_config.get("DB_PORT") or 5432),
|
||
"database": str(payload.get("target_db_name") or remote_db_config.get("DB_DATABASE") or "").strip(),
|
||
"user": str(payload.get("target_db_user") or remote_db_config.get("DB_USER") or "").strip(),
|
||
"password": str(payload.get("target_db_password") or remote_db_config.get("DB_PASSWORD") or ""),
|
||
}
|
||
target_db_config["password_configured"] = bool(str(target_db_config.get("password") or "").strip())
|
||
|
||
blocking_reasons = list(remote_checks.get("blocking_reasons") or [])
|
||
warnings = list(remote_checks.get("warnings") or [])
|
||
target_db_inspection = _inspect_target_database(node, target_db_config, remote_checks)
|
||
if bool(payload.get("overwrite_database", False)):
|
||
if not target_db_config["database"] or not target_db_config["user"]:
|
||
blocking_reasons.append("已勾选覆盖数据库，但目标机尚未识别到有效 DB_DATABASE / DB_USER。")
|
||
if not bool(remote_checks.get("tools", {}).get("psql", False)):
|
||
blocking_reasons.append("已勾选覆盖数据库，但目标机缺少 psql。")
|
||
if not bool(remote_checks.get("tools", {}).get("pg_dump", False)) and bool(payload.get("backup_target_database", True)):
|
||
blocking_reasons.append("已勾选“覆盖前先备份目标库”，但目标机缺少 pg_dump。")
|
||
elif not bool(remote_checks.get("tools", {}).get("pg_dump", False)):
|
||
warnings.append("目标机缺少 pg_dump，将无法在覆盖前生成远端数据库备份。")
|
||
if bool(target_db_inspection.get("available", False)) and bool(target_db_inspection.get("has_business_data", False)):
|
||
warnings.append("目标数据库已存在业务表或统计行数，执行覆盖前必须做二次确认。")
|
||
elif not bool(target_db_inspection.get("available", False)):
|
||
warnings.append("暂时无法确认目标数据库是否为空，请先检查目标机数据库配置与连通性。")
|
||
|
||
context = {
|
||
"payload": dict(payload or {}),
|
||
"source_profile": source_profile,
|
||
"target_node": _serialize_target_node(node),
|
||
"target_paths": {
|
||
"repo_path": str(payload.get("target_repo_path") or ""),
|
||
"domain_root": str(payload.get("target_domain_root") or ""),
|
||
"api_root": str(payload.get("target_api_root") or ""),
|
||
"web_root": str(payload.get("target_web_root") or ""),
|
||
},
|
||
"remote_checks": remote_checks,
|
||
"target_db_config": target_db_config,
|
||
"target_db_inspection": target_db_inspection,
|
||
"plan_steps": _build_plan_steps(payload),
|
||
"blocking_reasons": blocking_reasons,
|
||
"warnings": warnings,
|
||
}
|
||
context["execution_guard"] = _build_execution_guard(context)
|
||
if blocking_reasons:
|
||
return False, "迁移预检查未通过", context
|
||
return True, "ok", context
|
||
|
||
|
||
def _build_plan_steps(payload: dict) -> list[dict]:
|
||
steps = []
|
||
if bool(payload.get("sync_env_files", True)):
|
||
steps.append({"key": "sync_env_files", "title": "同步 .env / /etc/default 配置"})
|
||
if bool(payload.get("sync_systemd_units", True)):
|
||
steps.append({"key": "sync_systemd_units", "title": "同步 systemd 单元文件"})
|
||
if bool(payload.get("overwrite_database", False)):
|
||
steps.append({"key": "overwrite_database", "title": "备份并覆盖目标 PostgreSQL"})
|
||
if bool(payload.get("build_frontend", True)):
|
||
steps.append({"key": "build_frontend", "title": "在目标机重新构建前端"})
|
||
if bool(payload.get("restart_services", True)):
|
||
steps.append({"key": "restart_services", "title": "daemon-reload 并重启 domaincheck-* 服务"})
|
||
steps.append({"key": "health_check", "title": "校验目标机 /health"})
|
||
return steps
|
||
|
||
|
||
def _collect_remote_checks(node: dict, payload: dict, *, include_file_reads: bool) -> dict:
|
||
repo_path = str(payload.get("target_repo_path") or "")
|
||
domain_root = str(payload.get("target_domain_root") or "")
|
||
api_root = str(payload.get("target_api_root") or "")
|
||
web_root = str(payload.get("target_web_root") or "")
|
||
python_script = f"""
|
||
import json
|
||
import os
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
repo_path = Path({json.dumps(repo_path)})
|
||
domain_root = Path({json.dumps(domain_root)})
|
||
api_root = Path({json.dumps(api_root)})
|
||
web_root = Path({json.dumps(web_root)})
|
||
|
||
def command_exists(name):
|
||
completed = subprocess.run(["bash", "-lc", f"command -v {{name}} >/dev/null 2>&1"], capture_output=True, text=True)
|
||
return completed.returncode == 0
|
||
|
||
def read_env(path):
|
||
values = {{}}
|
||
if not path.exists():
|
||
return values
|
||
for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
||
stripped = line.strip()
|
||
if not stripped or stripped.startswith("#") or "=" not in stripped:
|
||
continue
|
||
key, value = stripped.split("=", 1)
|
||
values[key.strip()] = value.strip().strip("'").strip('"')
|
||
return values
|
||
|
||
git_commit = ""
|
||
if repo_path.exists():
|
||
completed = subprocess.run(
|
||
["bash", "-lc", f"cd {{repo_path}} && git rev-parse HEAD"],
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
if completed.returncode == 0:
|
||
git_commit = str(completed.stdout or "").strip()
|
||
|
||
payload = {{
|
||
"paths": {{
|
||
"repo_exists": repo_path.exists(),
|
||
"repo_git": (repo_path / ".git").exists(),
|
||
"domain_root_exists": domain_root.exists(),
|
||
"api_root_exists": api_root.exists(),
|
||
"web_root_exists": web_root.exists(),
|
||
}},
|
||
"git_commit": git_commit,
|
||
"tools": {{
|
||
"python3": command_exists("python3"),
|
||
"python": command_exists("python"),
|
||
"node": command_exists("node"),
|
||
"npm": command_exists("npm"),
|
||
"systemctl": command_exists("systemctl"),
|
||
"psql": command_exists("psql"),
|
||
"pg_dump": command_exists("pg_dump"),
|
||
"curl": command_exists("curl"),
|
||
}},
|
||
"env": read_env(domain_root / ".env"),
|
||
}}
|
||
print(json.dumps(payload, ensure_ascii=False))
|
||
"""
|
||
ok, stdout, stderr, _meta = _run_remote_python(node, python_script, timeout=60)
|
||
if not ok:
|
||
return {
|
||
"ok": False,
|
||
"blocking_reasons": [f"远端预检查失败: {stderr or stdout or 'unknown error'}"],
|
||
"stderr": stderr,
|
||
"stdout": stdout,
|
||
}
|
||
try:
|
||
parsed = json.loads(stdout or "{}")
|
||
except Exception:
|
||
parsed = {}
|
||
|
||
paths = dict(parsed.get("paths") or {})
|
||
tools = dict(parsed.get("tools") or {})
|
||
remote_db_config = dict(parsed.get("env") or {})
|
||
blocking_reasons: list[str] = []
|
||
warnings: list[str] = []
|
||
if not bool(paths.get("repo_exists", False)):
|
||
blocking_reasons.append(f"目标机项目路径不存在: {repo_path}")
|
||
if not bool(paths.get("repo_git", False)):
|
||
blocking_reasons.append("目标机项目路径存在，但不是 Git 仓库。")
|
||
for tool_name in ("python3", "node", "npm", "systemctl", "curl"):
|
||
if not bool(tools.get(tool_name, False)):
|
||
blocking_reasons.append(f"目标机缺少 {tool_name}。")
|
||
if not bool(paths.get("domain_root_exists", False)):
|
||
warnings.append(f"目标 domain_root 不存在: {domain_root}")
|
||
if not bool(paths.get("api_root_exists", False)):
|
||
warnings.append(f"目标 api_root 不存在: {api_root}")
|
||
if not bool(paths.get("web_root_exists", False)):
|
||
warnings.append(f"目标 web_root 不存在: {web_root}")
|
||
env_files = {}
|
||
if include_file_reads:
|
||
env_files = _collect_remote_file_presence(node)
|
||
return {
|
||
"ok": len(blocking_reasons) == 0,
|
||
"git_commit": str(parsed.get("git_commit") or ""),
|
||
"paths": paths,
|
||
"tools": tools,
|
||
"remote_db_config": remote_db_config,
|
||
"remote_file_presence": env_files,
|
||
"blocking_reasons": blocking_reasons,
|
||
"warnings": warnings,
|
||
}
|
||
|
||
|
||
def _inspect_target_database(node: dict, target_db_config: dict, remote_checks: dict) -> dict:
|
||
tools = dict(remote_checks.get("tools") or {})
|
||
if not bool(tools.get("psql", False)):
|
||
return {
|
||
"available": False,
|
||
"reason": "target missing psql",
|
||
}
|
||
if not str(target_db_config.get("database") or "").strip() or not str(target_db_config.get("user") or "").strip():
|
||
return {
|
||
"available": False,
|
||
"reason": "target database identity incomplete",
|
||
}
|
||
password_export = ""
|
||
if str(target_db_config.get("password") or "").strip():
|
||
password_export = f"export PGPASSWORD={shlex.quote(str(target_db_config.get('password') or ''))}\n"
|
||
sql = """
|
||
SELECT json_build_object(
|
||
'database', current_database(),
|
||
'public_table_count', COALESCE((
|
||
SELECT count(*)
|
||
FROM information_schema.tables
|
||
WHERE table_schema = 'public'
|
||
), 0),
|
||
'business_table_count', COALESCE((
|
||
SELECT count(*)
|
||
FROM information_schema.tables
|
||
WHERE table_schema = 'public'
|
||
AND table_name = ANY(ARRAY['domains','detect_jobs','detect_job_items','detect_worker_nodes','ops_jobs'])
|
||
), 0),
|
||
'approx_total_rows', COALESCE((
|
||
SELECT sum(GREATEST(s.n_live_tup::bigint, 0))
|
||
FROM pg_stat_user_tables s
|
||
), 0),
|
||
'approx_domains_rows', COALESCE((
|
||
SELECT GREATEST(s.n_live_tup::bigint, 0)
|
||
FROM pg_stat_user_tables s
|
||
WHERE s.relname = 'domains'
|
||
LIMIT 1
|
||
), 0),
|
||
'has_domains_table', EXISTS(
|
||
SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'domains'
|
||
),
|
||
'has_detect_jobs_table', EXISTS(
|
||
SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'detect_jobs'
|
||
)
|
||
);
|
||
""".strip()
|
||
script = "\n".join(
|
||
[
|
||
"set -euo pipefail",
|
||
password_export.rstrip("\n"),
|
||
"psql "
|
||
f"-h {shlex.quote(str(target_db_config.get('host') or '127.0.0.1'))} "
|
||
f"-p {shlex.quote(str(target_db_config.get('port') or 5432))} "
|
||
f"-U {shlex.quote(str(target_db_config.get('user') or ''))} "
|
||
f"-d {shlex.quote(str(target_db_config.get('database') or ''))} "
|
||
"-At "
|
||
f"-c {shlex.quote(sql)}",
|
||
]
|
||
)
|
||
ok, stdout, stderr, _meta = _run_remote_shell(node, script, timeout=90)
|
||
if not ok:
|
||
return {
|
||
"available": False,
|
||
"reason": "query_failed",
|
||
"stderr": str(stderr or "").strip(),
|
||
}
|
||
try:
|
||
payload = json.loads(str(stdout or "").strip() or "{}")
|
||
except Exception:
|
||
return {
|
||
"available": False,
|
||
"reason": "invalid_json",
|
||
"stdout": str(stdout or "").strip(),
|
||
}
|
||
public_table_count = int(payload.get("public_table_count", 0) or 0)
|
||
business_table_count = int(payload.get("business_table_count", 0) or 0)
|
||
approx_total_rows = int(payload.get("approx_total_rows", 0) or 0)
|
||
approx_domains_rows = int(payload.get("approx_domains_rows", 0) or 0)
|
||
has_business_data = business_table_count > 0 or approx_total_rows > 0 or approx_domains_rows > 0
|
||
return {
|
||
"available": True,
|
||
"database": str(payload.get("database") or ""),
|
||
"public_table_count": public_table_count,
|
||
"business_table_count": business_table_count,
|
||
"approx_total_rows": approx_total_rows,
|
||
"approx_domains_rows": approx_domains_rows,
|
||
"has_domains_table": bool(payload.get("has_domains_table", False)),
|
||
"has_detect_jobs_table": bool(payload.get("has_detect_jobs_table", False)),
|
||
"has_business_data": has_business_data,
|
||
}
|
||
|
||
|
||
def _build_execution_guard(context: dict) -> dict:
|
||
payload = dict(context.get("payload") or {})
|
||
target_node = dict(context.get("target_node") or {})
|
||
target_db_config = dict(context.get("target_db_config") or {})
|
||
target_db_inspection = dict(context.get("target_db_inspection") or {})
|
||
context_hash = _build_execution_context_hash(context)
|
||
issued_at = int(time.time())
|
||
expires_at = issued_at + _MIGRATION_CONFIRMATION_TTL_SECONDS
|
||
required_confirmation_text = ""
|
||
if bool(payload.get("overwrite_database", False)) and bool(target_db_inspection.get("has_business_data", False)):
|
||
db_name = str(target_db_config.get("database") or "target-db").strip() or "target-db"
|
||
required_confirmation_text = f"OVERWRITE {db_name}"
|
||
token_payload = {
|
||
"target_node_code": str(target_node.get("node_code") or ""),
|
||
"context_hash": context_hash,
|
||
"issued_at": issued_at,
|
||
"expires_at": expires_at,
|
||
"required_confirmation_text": required_confirmation_text,
|
||
}
|
||
encoded_payload = _urlsafe_b64encode(json.dumps(token_payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8"))
|
||
signature = hmac.new(_migration_confirmation_secret(), encoded_payload.encode("utf-8"), hashlib.sha256).hexdigest()
|
||
return {
|
||
"token": f"{encoded_payload}.{signature}",
|
||
"context_hash": context_hash,
|
||
"issued_at": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(issued_at)),
|
||
"expires_at": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(expires_at)),
|
||
"required_confirmation_text": required_confirmation_text,
|
||
"requires_confirmation": bool(required_confirmation_text),
|
||
}
|
||
|
||
|
||
def _validate_execution_confirmation(payload: dict, context: dict) -> tuple[bool, str, dict]:
|
||
guard = dict(context.get("execution_guard") or {})
|
||
token = str(payload.get("execute_confirmation_token") or "").strip()
|
||
if not token:
|
||
return False, "执行前必须先跑一次预检查并携带确认令牌。", {
|
||
"blocking_reasons": ["missing execute_confirmation_token"],
|
||
"execution_guard": guard,
|
||
}
|
||
encoded_payload, dot, signature = token.partition(".")
|
||
if not encoded_payload or not dot or not signature:
|
||
return False, "确认令牌格式无效。", {
|
||
"blocking_reasons": ["invalid confirmation token format"],
|
||
"execution_guard": guard,
|
||
}
|
||
expected_signature = hmac.new(_migration_confirmation_secret(), encoded_payload.encode("utf-8"), hashlib.sha256).hexdigest()
|
||
if not hmac.compare_digest(expected_signature, signature):
|
||
return False, "确认令牌签名校验失败，请重新预检。", {
|
||
"blocking_reasons": ["invalid confirmation token signature"],
|
||
"execution_guard": guard,
|
||
}
|
||
try:
|
||
token_payload = json.loads(_urlsafe_b64decode(encoded_payload).decode("utf-8"))
|
||
except Exception:
|
||
return False, "确认令牌内容无法解析，请重新预检。", {
|
||
"blocking_reasons": ["invalid confirmation token payload"],
|
||
"execution_guard": guard,
|
||
}
|
||
if int(token_payload.get("expires_at", 0) or 0) < int(time.time()):
|
||
return False, "确认令牌已过期，请重新预检。", {
|
||
"blocking_reasons": ["expired confirmation token"],
|
||
"execution_guard": guard,
|
||
}
|
||
if str(token_payload.get("context_hash") or "").strip() != str(guard.get("context_hash") or "").strip():
|
||
return False, "迁移参数已变化，请重新预检生成新的执行令牌。", {
|
||
"blocking_reasons": ["confirmation token context mismatch"],
|
||
"execution_guard": guard,
|
||
}
|
||
required_confirmation_text = str(guard.get("required_confirmation_text") or "").strip()
|
||
if required_confirmation_text:
|
||
provided_text = str(payload.get("execute_confirmation_text") or "").strip()
|
||
if provided_text != required_confirmation_text:
|
||
return False, "缺少数据库覆盖确认文案，执行被拒绝。", {
|
||
"blocking_reasons": ["missing execute_confirmation_text"],
|
||
"execution_guard": guard,
|
||
}
|
||
return True, "ok", {}
|
||
|
||
|
||
def _build_execution_context_hash(context: dict) -> str:
|
||
payload = dict(context.get("payload") or {})
|
||
serializable = {
|
||
"target_node": str(((context.get("target_node") or {}).get("node_code")) or ""),
|
||
"target_paths": dict(context.get("target_paths") or {}),
|
||
"target_db": {
|
||
"host": str(((context.get("target_db_config") or {}).get("host")) or ""),
|
||
"port": int(((context.get("target_db_config") or {}).get("port")) or 0),
|
||
"database": str(((context.get("target_db_config") or {}).get("database")) or ""),
|
||
"user": str(((context.get("target_db_config") or {}).get("user")) or ""),
|
||
},
|
||
"sync_env_files": bool(payload.get("sync_env_files", True)),
|
||
"sync_systemd_units": bool(payload.get("sync_systemd_units", True)),
|
||
"build_frontend": bool(payload.get("build_frontend", True)),
|
||
"restart_services": bool(payload.get("restart_services", True)),
|
||
"overwrite_database": bool(payload.get("overwrite_database", False)),
|
||
"backup_target_database": bool(payload.get("backup_target_database", True)),
|
||
}
|
||
raw = json.dumps(serializable, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _migration_confirmation_secret() -> bytes:
|
||
parts = [
|
||
str(settings.node_code or ""),
|
||
str(settings.admin_password or ""),
|
||
str(settings.db_password or ""),
|
||
str(settings.sync_shared_token or ""),
|
||
]
|
||
return "|".join(parts).encode("utf-8")
|
||
|
||
|
||
def _collect_remote_file_presence(node: dict) -> dict:
|
||
files = [
|
||
"/etc/default/domaincheck-api",
|
||
"/etc/default/domaincheck-worker",
|
||
"/etc/default/domaincheck-ops-center",
|
||
"/etc/systemd/system/domaincheck-api.service",
|
||
"/etc/systemd/system/domaincheck-worker.service",
|
||
"/etc/systemd/system/domaincheck-sync-agent.service",
|
||
"/etc/systemd/system/postgresql.service",
|
||
]
|
||
python_script = f"""
|
||
import json
|
||
from pathlib import Path
|
||
|
||
files = {json.dumps(files, ensure_ascii=False)}
|
||
print(json.dumps({{item: Path(item).exists() for item in files}}, ensure_ascii=False))
|
||
"""
|
||
ok, stdout, _stderr, _meta = _run_remote_python(node, python_script, timeout=30)
|
||
if not ok:
|
||
return {}
|
||
try:
|
||
return json.loads(stdout or "{}")
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def _serialize_target_node(node: dict) -> dict:
|
||
return {
|
||
"node_code": str(node.get("node_code") or ""),
|
||
"title": str(node.get("title") or ""),
|
||
"region": str(node.get("region") or ""),
|
||
"role": str(node.get("role") or ""),
|
||
"ssh_host": str(node.get("ssh_host") or ""),
|
||
"ssh_port": int(node.get("ssh_port") or 22),
|
||
"ssh_user": str(node.get("ssh_user") or ""),
|
||
"auth_mode": str(node.get("auth_mode") or "key"),
|
||
}
|
||
|
||
|
||
def _find_managed_node(node_code: str) -> dict | None:
|
||
normalized = str(node_code or "").strip()
|
||
if not normalized:
|
||
return None
|
||
for item in list_managed_nodes():
|
||
if str(item.get("node_code") or "").strip() == normalized:
|
||
return dict(item)
|
||
return None
|
||
|
||
|
||
def _build_migration_job_payload(payload: dict) -> dict:
|
||
normalized = _normalize_payload(payload or {})
|
||
sanitized = {
|
||
**normalized,
|
||
"target_db_password": "",
|
||
"target_db_password_configured": bool(str(normalized.get("target_db_password") or "").strip()),
|
||
"execute_confirmation_token": "",
|
||
"execute_confirmation_text": "",
|
||
}
|
||
return sanitized
|
||
|
||
|
||
def _sanitize_migration_context(context: dict) -> dict:
|
||
sanitized = dict(context or {})
|
||
target_db_config = dict(sanitized.get("target_db_config") or {})
|
||
if "password" in target_db_config:
|
||
target_db_config["password"] = ""
|
||
sanitized["target_db_config"] = target_db_config
|
||
return sanitized
|
||
|
||
|
||
def _remember_migration_job_context(job_id: int, payload: dict, context: dict) -> None:
|
||
if int(job_id or 0) <= 0:
|
||
return
|
||
with _MIGRATION_JOB_CONTEXTS_LOCK:
|
||
_MIGRATION_JOB_CONTEXTS[int(job_id)] = {
|
||
"payload": dict(payload or {}),
|
||
"context": dict(context or {}),
|
||
"remembered_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
|
||
|
||
def _pop_migration_job_context(job_id: int) -> dict:
|
||
normalized_job_id = int(job_id or 0)
|
||
if normalized_job_id <= 0:
|
||
return {}
|
||
with _MIGRATION_JOB_CONTEXTS_LOCK:
|
||
return dict(_MIGRATION_JOB_CONTEXTS.pop(normalized_job_id, {}) or {})
|
||
|
||
|
||
def _start_migration_dispatch_thread(job_id: int) -> None:
|
||
normalized_job_id = int(job_id or 0)
|
||
if normalized_job_id <= 0:
|
||
return
|
||
thread = threading.Thread(
|
||
target=_dispatch_migration_job_async,
|
||
args=(normalized_job_id,),
|
||
name=f"ops-migration-{normalized_job_id}",
|
||
daemon=True,
|
||
)
|
||
thread.start()
|
||
|
||
|
||
def _dispatch_migration_job_async(job_id: int) -> None:
|
||
try:
|
||
ok, message, data = dispatch_ops_job(int(job_id))
|
||
if not ok:
|
||
target_node_code = str((data.get("job") or {}).get("target_node_code") or "")
|
||
_append_migration_job_event(
|
||
job_id=int(job_id),
|
||
target_node_code=target_node_code,
|
||
event_type="migration_dispatch_failed",
|
||
level="error",
|
||
message=message,
|
||
payload={"result": dict(data or {})},
|
||
)
|
||
except Exception as exc: # pragma: no cover - safety net
|
||
_append_migration_job_event(
|
||
job_id=int(job_id),
|
||
target_node_code="",
|
||
event_type="migration_dispatch_failed",
|
||
level="error",
|
||
message=f"迁移后台派发异常: {exc}",
|
||
payload={"exception": repr(exc)},
|
||
)
|
||
|
||
|
||
def _run_migration_plan(
|
||
payload: dict,
|
||
context: dict,
|
||
*,
|
||
job_id: int = 0,
|
||
requested_by: str = "api",
|
||
metadata: dict | None = None,
|
||
) -> tuple[bool, str, dict]:
|
||
normalized_payload = dict(payload or {})
|
||
context = dict(context or {})
|
||
node = dict(context.get("target_node") or {})
|
||
execution_results: list[dict] = []
|
||
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_started",
|
||
message=f"迁移任务开始执行，目标节点 {str(node.get('node_code') or '-')}",
|
||
payload={
|
||
"requested_by": str(requested_by or "api").strip() or "api",
|
||
"metadata": dict(metadata or {}),
|
||
"plan_steps": list(context.get("plan_steps") or []),
|
||
"target_paths": dict(context.get("target_paths") or {}),
|
||
},
|
||
)
|
||
|
||
def run_step(step_key: str, title: str, fn) -> bool:
|
||
started_at = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_step_started",
|
||
message=f"{title} 开始",
|
||
payload={
|
||
"step_key": step_key,
|
||
"title": title,
|
||
"started_at": started_at,
|
||
},
|
||
)
|
||
try:
|
||
step_ok, step_message, step_data = fn()
|
||
except Exception as exc: # pragma: no cover - safety net
|
||
step_ok = False
|
||
step_message = f"{title} 失败: {exc}"
|
||
step_data = {"exception": repr(exc)}
|
||
finished_at = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
step_record = {
|
||
"step_key": step_key,
|
||
"title": title,
|
||
"ok": bool(step_ok),
|
||
"message": str(step_message or ""),
|
||
"data": dict(step_data or {}),
|
||
"started_at": started_at,
|
||
"finished_at": finished_at,
|
||
}
|
||
execution_results.append(step_record)
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_step_completed",
|
||
level="success" if step_ok else "error",
|
||
message=str(step_message or title),
|
||
payload={
|
||
"step_key": step_key,
|
||
"title": title,
|
||
"started_at": started_at,
|
||
"finished_at": finished_at,
|
||
"ok": bool(step_ok),
|
||
"result": dict(step_data or {}),
|
||
},
|
||
)
|
||
return bool(step_ok)
|
||
|
||
if bool(normalized_payload.get("sync_env_files", True)):
|
||
if not run_step(
|
||
"sync_env_files",
|
||
"同步环境配置",
|
||
lambda: _sync_env_files(node, context),
|
||
):
|
||
result = _sanitize_migration_context({**context, "execution_steps": execution_results})
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_failed",
|
||
level="error",
|
||
message="迁移中断：同步环境配置失败",
|
||
payload={"execution_steps": execution_results},
|
||
)
|
||
return False, "迁移中断：同步环境配置失败", result
|
||
|
||
if bool(normalized_payload.get("sync_systemd_units", True)):
|
||
if not run_step(
|
||
"sync_systemd_units",
|
||
"同步 systemd 单元",
|
||
lambda: _sync_systemd_units(node, context),
|
||
):
|
||
result = _sanitize_migration_context({**context, "execution_steps": execution_results})
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_failed",
|
||
level="error",
|
||
message="迁移中断：同步 systemd 单元失败",
|
||
payload={"execution_steps": execution_results},
|
||
)
|
||
return False, "迁移中断：同步 systemd 单元失败", result
|
||
|
||
if bool(normalized_payload.get("overwrite_database", False)):
|
||
if not run_step(
|
||
"overwrite_database",
|
||
"覆盖目标数据库",
|
||
lambda: _overwrite_target_database(node, context, job_id=job_id),
|
||
):
|
||
result = _sanitize_migration_context({**context, "execution_steps": execution_results})
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_failed",
|
||
level="error",
|
||
message="迁移中断：目标数据库覆盖失败",
|
||
payload={"execution_steps": execution_results},
|
||
)
|
||
return False, "迁移中断：目标数据库覆盖失败", result
|
||
|
||
if bool(normalized_payload.get("build_frontend", True)):
|
||
if not run_step(
|
||
"build_frontend",
|
||
"构建前端资源",
|
||
lambda: _build_remote_frontend(node, context, job_id=job_id),
|
||
):
|
||
result = _sanitize_migration_context({**context, "execution_steps": execution_results})
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_failed",
|
||
level="error",
|
||
message="迁移中断：前端构建失败",
|
||
payload={"execution_steps": execution_results},
|
||
)
|
||
return False, "迁移中断：前端构建失败", result
|
||
|
||
if bool(normalized_payload.get("restart_services", True)):
|
||
if not run_step(
|
||
"restart_services",
|
||
"刷新并重启服务",
|
||
lambda: _restart_remote_services(node, job_id=job_id),
|
||
):
|
||
result = _sanitize_migration_context({**context, "execution_steps": execution_results})
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_failed",
|
||
level="error",
|
||
message="迁移中断：服务重启失败",
|
||
payload={"execution_steps": execution_results},
|
||
)
|
||
return False, "迁移中断：服务重启失败", result
|
||
|
||
if not run_step(
|
||
"health_check",
|
||
"执行远端健康检查",
|
||
lambda: _run_remote_health_check(node, job_id=job_id),
|
||
):
|
||
result = _sanitize_migration_context({**context, "execution_steps": execution_results})
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_failed",
|
||
level="error",
|
||
message="迁移完成，但健康检查失败",
|
||
payload={"execution_steps": execution_results},
|
||
)
|
||
return False, "迁移完成，但健康检查失败", result
|
||
|
||
result = _sanitize_migration_context({**context, "execution_steps": execution_results})
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_finished",
|
||
level="success",
|
||
message="迁移执行完成",
|
||
payload={"execution_steps": execution_results},
|
||
)
|
||
return True, "迁移执行完成", result
|
||
|
||
|
||
def _append_migration_job_event(
|
||
*,
|
||
job_id: int,
|
||
target_node_code: str,
|
||
event_type: str,
|
||
message: str,
|
||
level: str = "info",
|
||
payload: dict | None = None,
|
||
) -> None:
|
||
if int(job_id or 0) <= 0:
|
||
return
|
||
append_ops_job_event(
|
||
job_id=int(job_id),
|
||
node_code=str(target_node_code or "").strip(),
|
||
event_type=str(event_type or "migration_event").strip() or "migration_event",
|
||
level=str(level or "info").strip() or "info",
|
||
message=str(message or "").strip()[:2000],
|
||
payload={
|
||
**dict(payload or {}),
|
||
"occurred_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"summary_text": str(message or "").strip()[:2000],
|
||
},
|
||
)
|
||
|
||
|
||
def _run_remote_python(node: dict, script: str, *, timeout: int = 60) -> tuple[bool, str, str, dict]:
|
||
remote_command = "\n".join(
|
||
[
|
||
"set -euo pipefail",
|
||
"if command -v python3 >/dev/null 2>&1; then PYTHON_BIN=python3; elif command -v python >/dev/null 2>&1; then PYTHON_BIN=python; else echo 'python missing' >&2; exit 127; fi",
|
||
'"${PYTHON_BIN}" - <<\'PY\'',
|
||
script.rstrip("\n"),
|
||
"PY",
|
||
]
|
||
)
|
||
return _run_remote_shell(node, remote_command, timeout=timeout)
|
||
|
||
|
||
def _run_remote_shell(
|
||
node: dict,
|
||
remote_script: str,
|
||
*,
|
||
timeout: int = _MIGRATION_TIMEOUT_SECONDS,
|
||
stdin_data: bytes | None = None,
|
||
) -> tuple[bool, str, str, dict]:
|
||
ssh_host = str(node.get("ssh_host") or "").strip()
|
||
ssh_user = str(node.get("ssh_user") or "").strip()
|
||
ssh_port = int(node.get("ssh_port") or 22)
|
||
auth_mode = str(node.get("auth_mode") or "key").strip() or "key"
|
||
secret = _load_node_secret(node)
|
||
ssh_password = str(secret.get("ssh_password") or "")
|
||
ssh_private_key = str(secret.get("ssh_private_key") or "")
|
||
|
||
if paramiko is not None and ((auth_mode == "password" and ssh_password) or ssh_private_key):
|
||
completed = _run_paramiko_command(
|
||
ssh_host=ssh_host,
|
||
ssh_port=ssh_port,
|
||
ssh_user=ssh_user,
|
||
remote_command=remote_script,
|
||
timeout_seconds=timeout,
|
||
ssh_password=ssh_password if auth_mode == "password" else "",
|
||
ssh_private_key=ssh_private_key if auth_mode == "key" else "",
|
||
stdin_data=stdin_data,
|
||
)
|
||
return completed.returncode == 0, completed.stdout, completed.stderr, {
|
||
"executor": "paramiko",
|
||
"returncode": int(completed.returncode or 0),
|
||
}
|
||
|
||
wrapped_command = f"bash -lc {shlex.quote(remote_script)}"
|
||
completed = subprocess.run(
|
||
[
|
||
"ssh",
|
||
"-o",
|
||
"BatchMode=yes",
|
||
"-o",
|
||
"StrictHostKeyChecking=accept-new",
|
||
"-o",
|
||
f"ConnectTimeout={_SSH_CONNECT_TIMEOUT_SECONDS}",
|
||
"-p",
|
||
str(ssh_port),
|
||
f"{ssh_user}@{ssh_host}",
|
||
wrapped_command,
|
||
],
|
||
input=stdin_data,
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
)
|
||
return completed.returncode == 0, completed.stdout.decode("utf-8", errors="replace"), completed.stderr.decode("utf-8", errors="replace"), {
|
||
"executor": "ssh",
|
||
"returncode": int(completed.returncode or 0),
|
||
}
|
||
|
||
|
||
def _upload_text(node: dict, remote_path: str, content: str, *, mode: int = 0o600) -> tuple[bool, str, dict]:
|
||
payload = base64.b64encode(str(content or "").encode("utf-8")).decode("ascii")
|
||
python_script = f"""
|
||
import base64
|
||
import os
|
||
from pathlib import Path
|
||
|
||
target = Path({json.dumps(remote_path)})
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
target.write_bytes(base64.b64decode({json.dumps(payload)}))
|
||
os.chmod(target, {int(mode)})
|
||
print("ok")
|
||
"""
|
||
ok, stdout, stderr, meta = _run_remote_python(node, python_script, timeout=60)
|
||
return ok, stderr or stdout or ("ok" if ok else "upload failed"), meta
|
||
|
||
|
||
def _upload_file(node: dict, local_path: str, remote_path: str) -> tuple[bool, str]:
|
||
local = Path(local_path)
|
||
if not local.exists():
|
||
return False, f"本地文件不存在: {local_path}"
|
||
ssh_host = str(node.get("ssh_host") or "").strip()
|
||
ssh_user = str(node.get("ssh_user") or "").strip()
|
||
ssh_port = int(node.get("ssh_port") or 22)
|
||
auth_mode = str(node.get("auth_mode") or "key").strip() or "key"
|
||
secret = _load_node_secret(node)
|
||
ssh_password = str(secret.get("ssh_password") or "")
|
||
ssh_private_key = str(secret.get("ssh_private_key") or "")
|
||
|
||
if paramiko is not None and ((auth_mode == "password" and ssh_password) or ssh_private_key):
|
||
client = _open_paramiko_client(
|
||
ssh_host=ssh_host,
|
||
ssh_port=ssh_port,
|
||
ssh_user=ssh_user,
|
||
ssh_password=ssh_password if auth_mode == "password" else "",
|
||
ssh_private_key=ssh_private_key if auth_mode == "key" else "",
|
||
)
|
||
try:
|
||
sftp = client.open_sftp()
|
||
remote_parent = str(Path(remote_path).parent)
|
||
_ensure_sftp_dir(sftp, remote_parent)
|
||
sftp.put(str(local), remote_path)
|
||
sftp.close()
|
||
finally:
|
||
client.close()
|
||
return True, "ok"
|
||
|
||
mkdir_script = f"mkdir -p {shlex.quote(str(Path(remote_path).parent))}"
|
||
ok, _stdout, stderr, _meta = _run_remote_shell(node, mkdir_script, timeout=30)
|
||
if not ok:
|
||
return False, stderr or "无法创建远端目录"
|
||
completed = subprocess.run(
|
||
[
|
||
"scp",
|
||
"-P",
|
||
str(ssh_port),
|
||
str(local),
|
||
f"{ssh_user}@{ssh_host}:{remote_path}",
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=_MIGRATION_TIMEOUT_SECONDS,
|
||
)
|
||
if completed.returncode != 0:
|
||
return False, str(completed.stderr or completed.stdout or "scp failed").strip()
|
||
return True, "ok"
|
||
|
||
|
||
def _truncate_output(value: object, limit: int = 4000) -> str:
|
||
text = str(value or "").strip()
|
||
if len(text) <= limit:
|
||
return text
|
||
return text[-limit:]
|
||
|
||
|
||
def _sync_env_files(node: dict, context: dict) -> tuple[bool, str, dict]:
|
||
target_paths = dict(context.get("target_paths") or {})
|
||
local_domain_root = Path(context.get("source_profile", {}).get("workspace", {}).get("domain_root", settings.domain_root))
|
||
local_web_root = Path(context.get("source_profile", {}).get("workspace", {}).get("web_root", local_domain_root.parent / "domain-web"))
|
||
files = [
|
||
(str(local_domain_root / ".env"), f"{target_paths.get('domain_root')}/.env", 0o600),
|
||
("/etc/default/domaincheck-api", "/etc/default/domaincheck-api", 0o644),
|
||
("/etc/default/domaincheck-worker", "/etc/default/domaincheck-worker", 0o644),
|
||
("/etc/default/domaincheck-ops-center", "/etc/default/domaincheck-ops-center", 0o644),
|
||
(str(local_web_root / ".env.production"), f"{target_paths.get('web_root')}/.env.production", 0o644),
|
||
]
|
||
uploaded: list[dict] = []
|
||
for local_path, remote_path, mode in files:
|
||
path = Path(local_path)
|
||
if not path.exists():
|
||
continue
|
||
ok, message, _meta = _upload_text(node, remote_path, path.read_text(encoding="utf-8", errors="ignore"), mode=mode)
|
||
uploaded.append({"local_path": local_path, "remote_path": remote_path, "ok": ok, "message": message})
|
||
if not ok:
|
||
return False, f"同步文件失败: {remote_path}", {"uploaded": uploaded}
|
||
return True, "环境配置已同步", {"uploaded": uploaded}
|
||
|
||
|
||
def _sync_systemd_units(node: dict, _context: dict) -> tuple[bool, str, dict]:
|
||
files = [
|
||
"/etc/systemd/system/domaincheck-api.service",
|
||
"/etc/systemd/system/domaincheck-worker.service",
|
||
"/etc/systemd/system/domaincheck-sync-agent.service",
|
||
"/etc/systemd/system/domaincheck-node-agent.service",
|
||
"/etc/systemd/system/postgresql.service",
|
||
]
|
||
uploaded: list[dict] = []
|
||
for file_path in files:
|
||
path = Path(file_path)
|
||
if not path.exists():
|
||
continue
|
||
ok, message, _meta = _upload_text(node, file_path, path.read_text(encoding="utf-8", errors="ignore"), mode=0o644)
|
||
uploaded.append({"path": file_path, "ok": ok, "message": message})
|
||
if not ok:
|
||
return False, f"同步 systemd 单元失败: {file_path}", {"uploaded": uploaded}
|
||
return True, "systemd 单元已同步", {"uploaded": uploaded}
|
||
|
||
|
||
def _overwrite_target_database(node: dict, context: dict, *, job_id: int = 0) -> tuple[bool, str, dict]:
|
||
source_db = dict(context.get("source_profile", {}).get("database") or {})
|
||
dump_dir = Path(tempfile.mkdtemp(prefix="domaincheck-migration-"))
|
||
dump_path = dump_dir / "source.sql"
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_db_dump_started",
|
||
message="开始导出源 PostgreSQL",
|
||
payload={
|
||
"database": str(source_db.get("database") or ""),
|
||
"host": str(source_db.get("host") or ""),
|
||
"port": int(source_db.get("port") or 5432),
|
||
"user": str(source_db.get("user") or ""),
|
||
},
|
||
)
|
||
dump_command = [
|
||
"pg_dump",
|
||
"-h",
|
||
str(source_db.get("host") or "127.0.0.1"),
|
||
"-p",
|
||
str(source_db.get("port") or 5432),
|
||
"-U",
|
||
str(source_db.get("user") or ""),
|
||
"--clean",
|
||
"--if-exists",
|
||
"--no-owner",
|
||
"--no-privileges",
|
||
"-d",
|
||
str(source_db.get("database") or ""),
|
||
"-f",
|
||
str(dump_path),
|
||
]
|
||
env = os.environ.copy()
|
||
if str(settings.db_password or "").strip():
|
||
env["PGPASSWORD"] = str(settings.db_password or "")
|
||
dump_completed = subprocess.run(dump_command, capture_output=True, text=True, env=env, timeout=3600)
|
||
if dump_completed.returncode != 0:
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_db_dump_failed",
|
||
level="error",
|
||
message="源数据库导出失败",
|
||
payload={
|
||
"stdout": _truncate_output(dump_completed.stdout),
|
||
"stderr": _truncate_output(dump_completed.stderr),
|
||
},
|
||
)
|
||
return False, "源数据库导出失败", {
|
||
"stderr": str(dump_completed.stderr or "").strip(),
|
||
"stdout": str(dump_completed.stdout or "").strip(),
|
||
}
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_db_dump_completed",
|
||
level="success",
|
||
message="源 PostgreSQL 导出完成",
|
||
payload={
|
||
"dump_path": str(dump_path),
|
||
"stdout": _truncate_output(dump_completed.stdout),
|
||
"stderr": _truncate_output(dump_completed.stderr),
|
||
},
|
||
)
|
||
|
||
remote_dump_path = f"/tmp/domaincheck-migration-{int(time.time())}.sql"
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_db_upload_started",
|
||
message="开始上传数据库 dump 到目标机",
|
||
payload={"remote_dump_path": remote_dump_path},
|
||
)
|
||
upload_ok, upload_message = _upload_file(node, str(dump_path), remote_dump_path)
|
||
if not upload_ok:
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_db_upload_failed",
|
||
level="error",
|
||
message="数据库 dump 上传失败",
|
||
payload={"remote_dump_path": remote_dump_path, "message": upload_message},
|
||
)
|
||
return False, "数据库 dump 上传失败", {"message": upload_message}
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_db_upload_completed",
|
||
level="success",
|
||
message="数据库 dump 已上传到目标机",
|
||
payload={"remote_dump_path": remote_dump_path},
|
||
)
|
||
|
||
backup_result = {}
|
||
target_db = dict(context.get("target_db_config") or {})
|
||
can_backup_target = bool(target_db.get("database")) and bool(target_db.get("user")) and bool(target_db.get("host"))
|
||
wants_backup = bool(context.get("payload", {}).get("backup_target_database", True))
|
||
if wants_backup and can_backup_target:
|
||
backup_result = _backup_remote_database(node, context)
|
||
if not bool(backup_result.get("ok", True)):
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_target_backup_failed",
|
||
level="error",
|
||
message="目标数据库备份失败",
|
||
payload=dict(backup_result or {}),
|
||
)
|
||
return False, "目标数据库备份失败", backup_result
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_target_backup_completed",
|
||
level="success",
|
||
message="目标数据库备份完成",
|
||
payload=dict(backup_result or {}),
|
||
)
|
||
|
||
restore_result = _restore_remote_database(node, context, remote_dump_path)
|
||
if not bool(restore_result.get("ok", False)):
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_target_restore_failed",
|
||
level="error",
|
||
message="目标数据库恢复失败",
|
||
payload=dict(restore_result or {}),
|
||
)
|
||
return False, "目标数据库恢复失败", restore_result
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_target_restore_completed",
|
||
level="success",
|
||
message="目标数据库恢复完成",
|
||
payload=dict(restore_result or {}),
|
||
)
|
||
return True, "目标数据库已覆盖", {
|
||
"remote_dump_path": remote_dump_path,
|
||
"backup": backup_result,
|
||
"restore": restore_result,
|
||
}
|
||
|
||
|
||
def _backup_remote_database(node: dict, context: dict) -> dict:
|
||
target_db = dict(context.get("target_db_config") or {})
|
||
backup_path = f"/tmp/domaincheck-target-backup-{int(time.time())}.sql"
|
||
password_export = ""
|
||
if str(target_db.get("password") or "").strip():
|
||
password_export = f"export PGPASSWORD={shlex.quote(str(target_db.get('password') or ''))}\n"
|
||
script = "\n".join(
|
||
[
|
||
"set -euo pipefail",
|
||
password_export.rstrip("\n"),
|
||
"mkdir -p /tmp",
|
||
"pg_dump "
|
||
f"-h {shlex.quote(str(target_db.get('host') or '127.0.0.1'))} "
|
||
f"-p {shlex.quote(str(target_db.get('port') or 5432))} "
|
||
f"-U {shlex.quote(str(target_db.get('user') or ''))} "
|
||
f"-d {shlex.quote(str(target_db.get('database') or ''))} "
|
||
f"-f {shlex.quote(backup_path)}",
|
||
f"echo {shlex.quote(backup_path)}",
|
||
]
|
||
)
|
||
ok, stdout, stderr, _meta = _run_remote_shell(node, script, timeout=1800)
|
||
return {
|
||
"ok": ok,
|
||
"backup_path": str(stdout or "").strip() if ok else "",
|
||
"stderr": str(stderr or "").strip(),
|
||
}
|
||
|
||
|
||
def _restore_remote_database(node: dict, context: dict, remote_dump_path: str) -> dict:
|
||
target_db = dict(context.get("target_db_config") or {})
|
||
password_export = ""
|
||
if str(target_db.get("password") or "").strip():
|
||
password_export = f"export PGPASSWORD={shlex.quote(str(target_db.get('password') or ''))}\n"
|
||
script = "\n".join(
|
||
[
|
||
"set -euo pipefail",
|
||
password_export.rstrip("\n"),
|
||
f"cat {shlex.quote(remote_dump_path)} | psql "
|
||
f"-h {shlex.quote(str(target_db.get('host') or '127.0.0.1'))} "
|
||
f"-p {shlex.quote(str(target_db.get('port') or 5432))} "
|
||
f"-U {shlex.quote(str(target_db.get('user') or ''))} "
|
||
f"-d {shlex.quote(str(target_db.get('database') or ''))}",
|
||
"echo restored",
|
||
]
|
||
)
|
||
ok, stdout, stderr, _meta = _run_remote_shell(node, script, timeout=3600)
|
||
return {
|
||
"ok": ok,
|
||
"stdout": str(stdout or "").strip(),
|
||
"stderr": str(stderr or "").strip(),
|
||
}
|
||
|
||
|
||
def _build_remote_frontend(node: dict, context: dict, *, job_id: int = 0) -> tuple[bool, str, dict]:
|
||
repo_path = str(context.get("target_paths", {}).get("repo_path") or "")
|
||
install_script = "\n".join(
|
||
[
|
||
"set -euo pipefail",
|
||
f"cd {shlex.quote(repo_path)}",
|
||
"if command -v npm >/dev/null 2>&1; then",
|
||
" cd domain-web",
|
||
" npm install",
|
||
" npm run build",
|
||
"else",
|
||
" echo 'npm missing' >&2",
|
||
" exit 127",
|
||
"fi",
|
||
]
|
||
)
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_frontend_build_started",
|
||
message="开始执行目标机前端构建",
|
||
payload={"repo_path": repo_path},
|
||
)
|
||
ok, stdout, stderr, _meta = _run_remote_shell(node, install_script, timeout=3600)
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_frontend_build_completed" if ok else "migration_frontend_build_failed",
|
||
level="success" if ok else "error",
|
||
message="前端构建完成" if ok else "前端构建失败",
|
||
payload={
|
||
"repo_path": repo_path,
|
||
"stdout": _truncate_output(stdout),
|
||
"stderr": _truncate_output(stderr),
|
||
},
|
||
)
|
||
return ok, "前端构建完成" if ok else "前端构建失败", {
|
||
"stdout": _truncate_output(stdout),
|
||
"stderr": _truncate_output(stderr),
|
||
}
|
||
|
||
|
||
def _restart_remote_services(node: dict, *, job_id: int = 0) -> tuple[bool, str, dict]:
|
||
script = "\n".join(
|
||
[
|
||
"set -euo pipefail",
|
||
"systemctl daemon-reload",
|
||
"systemctl enable postgresql domaincheck-api domaincheck-worker domaincheck-sync-agent >/dev/null 2>&1 || true",
|
||
"systemctl restart postgresql",
|
||
"systemctl restart domaincheck-api domaincheck-worker domaincheck-sync-agent",
|
||
"systemctl --no-pager --full status domaincheck-api domaincheck-worker domaincheck-sync-agent | sed -n '1,120p'",
|
||
]
|
||
)
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_service_restart_started",
|
||
message="开始刷新并重启目标机服务",
|
||
payload={},
|
||
)
|
||
ok, stdout, stderr, _meta = _run_remote_shell(node, script, timeout=300)
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_service_restart_completed" if ok else "migration_service_restart_failed",
|
||
level="success" if ok else "error",
|
||
message="远端服务已重启" if ok else "远端服务重启失败",
|
||
payload={
|
||
"stdout": _truncate_output(stdout),
|
||
"stderr": _truncate_output(stderr),
|
||
},
|
||
)
|
||
return ok, "远端服务已重启" if ok else "远端服务重启失败", {
|
||
"stdout": _truncate_output(stdout),
|
||
"stderr": _truncate_output(stderr),
|
||
}
|
||
|
||
|
||
def _run_remote_health_check(node: dict, *, job_id: int = 0) -> tuple[bool, str, dict]:
|
||
script = "curl -fsS http://127.0.0.1:8100/health"
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_health_check_started",
|
||
message="开始执行目标机健康检查",
|
||
payload={},
|
||
)
|
||
ok, stdout, stderr, _meta = _run_remote_shell(node, script, timeout=60)
|
||
data = {}
|
||
try:
|
||
data = json.loads(stdout or "{}")
|
||
except Exception:
|
||
data = {"raw": str(stdout or "").strip()}
|
||
_append_migration_job_event(
|
||
job_id=job_id,
|
||
target_node_code=str(node.get("node_code") or ""),
|
||
event_type="migration_health_check_completed" if ok else "migration_health_check_failed",
|
||
level="success" if ok else "error",
|
||
message="远端健康检查通过" if ok else "远端健康检查失败",
|
||
payload={
|
||
"response": data,
|
||
"stderr": _truncate_output(stderr),
|
||
},
|
||
)
|
||
return ok, "远端健康检查通过" if ok else "远端健康检查失败", {
|
||
"response": data,
|
||
"stderr": str(stderr or "").strip(),
|
||
}
|
||
|
||
|
||
def _load_node_secret(node: dict) -> dict:
|
||
from app.services.ops_runtime_executor_service import _load_ssh_secret # noqa: PLC0415
|
||
|
||
return dict(_load_ssh_secret(str(node.get("node_code") or "").strip()) or {})
|
||
|
||
|
||
def _open_paramiko_client(
|
||
*,
|
||
ssh_host: str,
|
||
ssh_port: int,
|
||
ssh_user: str,
|
||
ssh_password: str = "",
|
||
ssh_private_key: str = "",
|
||
):
|
||
if paramiko is None:
|
||
raise RuntimeError("paramiko is not installed")
|
||
client = paramiko.SSHClient()
|
||
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
connect_kwargs = {
|
||
"hostname": ssh_host,
|
||
"port": int(ssh_port),
|
||
"username": ssh_user,
|
||
"timeout": _SSH_CONNECT_TIMEOUT_SECONDS,
|
||
"banner_timeout": _SSH_CONNECT_TIMEOUT_SECONDS,
|
||
"auth_timeout": _SSH_CONNECT_TIMEOUT_SECONDS,
|
||
"look_for_keys": False,
|
||
"allow_agent": False,
|
||
}
|
||
if ssh_password:
|
||
connect_kwargs["password"] = ssh_password
|
||
elif ssh_private_key:
|
||
connect_kwargs["pkey"] = _load_private_key(ssh_private_key)
|
||
else:
|
||
connect_kwargs["look_for_keys"] = True
|
||
connect_kwargs["allow_agent"] = True
|
||
client.connect(**connect_kwargs)
|
||
return client
|
||
|
||
|
||
def _run_paramiko_command(
|
||
*,
|
||
ssh_host: str,
|
||
ssh_port: int,
|
||
ssh_user: str,
|
||
remote_command: str,
|
||
timeout_seconds: int,
|
||
ssh_password: str = "",
|
||
ssh_private_key: str = "",
|
||
stdin_data: bytes | None = None,
|
||
) -> subprocess.CompletedProcess:
|
||
client = _open_paramiko_client(
|
||
ssh_host=ssh_host,
|
||
ssh_port=ssh_port,
|
||
ssh_user=ssh_user,
|
||
ssh_password=ssh_password,
|
||
ssh_private_key=ssh_private_key,
|
||
)
|
||
try:
|
||
transport = client.get_transport()
|
||
channel = transport.open_session(timeout=timeout_seconds)
|
||
channel.settimeout(timeout_seconds)
|
||
channel.exec_command(f"bash -lc {shlex.quote(remote_command)}")
|
||
if stdin_data:
|
||
channel.sendall(stdin_data)
|
||
channel.shutdown_write()
|
||
stdout = channel.makefile("rb").read().decode("utf-8", errors="replace")
|
||
stderr = channel.makefile_stderr("rb").read().decode("utf-8", errors="replace")
|
||
return subprocess.CompletedProcess(
|
||
args=["paramiko", f"{ssh_user}@{ssh_host}"],
|
||
returncode=int(channel.recv_exit_status()),
|
||
stdout=stdout,
|
||
stderr=stderr,
|
||
)
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def _load_private_key(private_key_text: str):
|
||
if paramiko is None:
|
||
raise RuntimeError("paramiko is not installed")
|
||
key_text = str(private_key_text or "")
|
||
for key_cls in (paramiko.Ed25519Key, paramiko.RSAKey, paramiko.ECDSAKey, paramiko.DSSKey):
|
||
try:
|
||
return key_cls.from_private_key(StringIO(key_text))
|
||
except Exception:
|
||
continue
|
||
raise ValueError("无法识别 SSH 私钥格式")
|
||
|
||
|
||
def _ensure_sftp_dir(sftp, remote_directory: str) -> None:
|
||
normalized = str(remote_directory or "").strip()
|
||
if not normalized or normalized == "/":
|
||
return
|
||
parts = []
|
||
current = Path(normalized)
|
||
while str(current) not in {"", ".", "/"}:
|
||
parts.append(str(current))
|
||
current = current.parent
|
||
for directory in reversed(parts):
|
||
try:
|
||
sftp.stat(directory)
|
||
except IOError:
|
||
sftp.mkdir(directory)
|
||
|
||
|
||
def _urlsafe_b64encode(value: bytes) -> str:
|
||
return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=")
|
||
|
||
|
||
def _urlsafe_b64decode(value: str) -> bytes:
|
||
normalized = str(value or "").strip()
|
||
padding = "=" * (-len(normalized) % 4)
|
||
return base64.urlsafe_b64decode(normalized + padding)
|