refactor(remote_env): 依赖安装改远程分离式——nohup逐包安装+进度文件,轮询无状态
This commit is contained in:
ymq
parent
023d4a67fa
commit
ffe19ab95b
@ -239,126 +239,102 @@ def collect_skill_deps(skills_base: str, org_id: str, heavy_skip: str) -> dict:
|
||||
return {"specs": sorted(specs.values()), "heavy_skipped": sorted(heavy_skipped)}
|
||||
|
||||
|
||||
async def install_remote_deps(env: dict, specs: list, pip_index: str,
|
||||
timeout: int = 1200) -> dict:
|
||||
"""在远程 ~/<remote_dir>/.venv-skills 建 venv 并安装依赖。幂等。
|
||||
async def launch_remote_install(env: dict, specs: list, pip_index: str, rel: str) -> bool:
|
||||
"""在远程以 nohup 分离进程启动逐包安装（脱离 SSH 会话，不怕断连/重启）。
|
||||
|
||||
specs 经 base64 传输（含 >=[] 等特殊字符，避开 shell 展开）。
|
||||
返回 {ok, installed:[...], failed:[...], error}
|
||||
进度写 ~/<rel>/.venv-skills-install.progress（OK:/FAILED: 行），
|
||||
完成写 ~/<rel>/.venv-skills-install.done（汇总）。
|
||||
本调用秒级返回，是否成功启动由返回值给出。
|
||||
"""
|
||||
if not specs:
|
||||
return {"ok": True, "installed": [], "failed": [], "skipped_reason": "无可安装依赖"}
|
||||
err = _validate_remote_dir(env.get("remote_dir", ""))
|
||||
if err:
|
||||
return {"ok": False, "error": err}
|
||||
|
||||
rel = _rel_remote_dir(env.get("remote_dir", ""))
|
||||
payload = base64.b64encode(json.dumps(specs).encode()).decode()
|
||||
# 不用 set -e / 管道：管道会吞掉 pip 的退出码（$? 变成 tail 的状态）。
|
||||
# set -f 关 glob：pip 规格里的 * / [...] 不能被远程 shell 展开。
|
||||
# 完整输出由 SSH 捕获，报告时截取尾部。
|
||||
script = (
|
||||
'RD="$HOME/' + rel + '"; VENV="$RD/.venv-skills"; '
|
||||
'command -v python3 >/dev/null || { echo "NO_PYTHON3"; exit 3; }; '
|
||||
'if [ ! -x "$VENV/bin/python" ]; then python3 -m venv "$VENV" || { echo "VENV_FAIL"; exit 4; }; fi; '
|
||||
'set -f; RC=0; "$VENV/bin/pip" install --no-input --disable-pip-version-check '
|
||||
+ (('-i ' + shlex.quote(pip_index) + ' ') if pip_index else '')
|
||||
+ '$(echo "' + payload + '" | base64 -d | python3 -c '
|
||||
'"import json,sys;print(chr(32).join(json.load(sys.stdin)))") '
|
||||
'|| RC=$?; '
|
||||
'echo "INSTALL_RC=$RC"'
|
||||
)
|
||||
ssh_cmd = ["ssh"] + _ssh_common_args(env) + [_ssh_target(env), script]
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*ssh_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
|
||||
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout + 30)
|
||||
except asyncio.TimeoutError:
|
||||
return {"ok": False, "error": f"依赖安装超时（{timeout}s），venv 可能装了一半，可重跑"}
|
||||
except Exception as e:
|
||||
return {"ok": False, "error": "SSH 执行失败: " + str(e)[:300]}
|
||||
|
||||
out = stdout.decode("utf-8", "replace")
|
||||
if "NO_PYTHON3" in out:
|
||||
return {"ok": False, "error": "远程主机缺少 python3"}
|
||||
if proc.returncode not in (0, 1):
|
||||
return {"ok": False, "error": "venv 创建失败: "
|
||||
+ (out + stderr.decode('utf-8', 'replace'))[-600:]}
|
||||
|
||||
# 解析 pip 退出码（由脚本末尾 echo INSTALL_RC=$RC 输出）
|
||||
rc = -1
|
||||
for line in out.splitlines():
|
||||
if line.startswith("INSTALL_RC="):
|
||||
try:
|
||||
rc = int(line.split("=", 1)[1])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# pip 逐包结果解析：新版 pip 解析器失败时要么全装要么全不装，
|
||||
# 从 ERROR 行提取包名。常见格式：
|
||||
# ERROR: Could not find a version that satisfies the requirement <name> ...
|
||||
# ERROR: No matching distribution found for <name>
|
||||
failed = []
|
||||
for line in out.splitlines():
|
||||
if "ERROR:" not in line:
|
||||
continue
|
||||
m = (re.search(r"the requirement ([A-Za-z0-9._\[\]-]+)", line)
|
||||
or re.search(r"No matching distribution found for ([A-Za-z0-9._\[\]-]+)", line))
|
||||
failed.append(m.group(1) if m else line.strip()[:120])
|
||||
if rc != 0 and not failed:
|
||||
failed = ["部分包安装失败，详见日志末尾"]
|
||||
|
||||
if rc == 0:
|
||||
installed = list(specs)
|
||||
failed = []
|
||||
else:
|
||||
# 批量失败 → 逐包降级：一次 SSH 会话内逐包安装（pip 幂等，已装的秒过），
|
||||
# 部分成功优于全败。
|
||||
return await _install_per_package(env, specs, pip_index, rel, timeout)
|
||||
return {"ok": rc == 0, "installed": installed, "failed": failed,
|
||||
"venv": "~/" + rel + "/.venv-skills",
|
||||
"tail": out[-800:]}
|
||||
|
||||
|
||||
async def _install_per_package(env: dict, specs: list, pip_index: str,
|
||||
rel: str, timeout: int) -> dict:
|
||||
"""逐包安装（批量失败后的降级路径）。返回格式与 install_remote_deps 一致。"""
|
||||
payload = base64.b64encode(json.dumps(specs).encode()).decode()
|
||||
index_arg = ('-i ' + shlex.quote(pip_index) + ' ') if pip_index else ''
|
||||
script = (
|
||||
# 分离脚本：建 venv → 逐包安装 → 写进度/完成标记。
|
||||
# 用 set -f 防 pip 规格里 * / [...] 被远程 shell 展开。
|
||||
inner = (
|
||||
'RD="$HOME/' + rel + '"; VENV="$RD/.venv-skills"; '
|
||||
'[ -x "$VENV/bin/pip" ] || { echo "VENV_MISSING"; exit 4; }; '
|
||||
'set -f; '
|
||||
'PROG="$RD/.venv-skills-install.progress"; DONE="$RD/.venv-skills-install.done"; '
|
||||
'if [ ! -x "$VENV/bin/python" ]; then '
|
||||
'python3 -m venv "$VENV" || { echo "VENV_FAIL" > "$DONE"; exit 4; }; fi; '
|
||||
'set -f; : > "$PROG"; '
|
||||
'echo "' + payload + '" | base64 -d | python3 -c '
|
||||
'"import json,sys;print(chr(10).join(json.load(sys.stdin)))" | '
|
||||
'while IFS= read -r pkg; do '
|
||||
' [ -n "$pkg" ] || continue; '
|
||||
' if "$VENV/bin/pip" install --no-input --disable-pip-version-check '
|
||||
+ index_arg + '"$pkg" >/dev/null 2>&1; then echo "OK:$pkg"; '
|
||||
' else echo "FAILED:$pkg"; fi; '
|
||||
'done'
|
||||
'[ -n "$pkg" ] || continue; '
|
||||
'if "$VENV/bin/pip" install --no-input --disable-pip-version-check '
|
||||
+ index_arg + '"$pkg" >/dev/null 2>&1; then echo "OK:$pkg" >> "$PROG"; '
|
||||
'else echo "FAILED:$pkg" >> "$PROG"; fi; done; '
|
||||
'OKC=$(grep -c "^OK:" "$PROG" 2>/dev/null || echo 0); '
|
||||
'FC=$(grep -c "^FAILED:" "$PROG" 2>/dev/null || echo 0); '
|
||||
'echo "OK=$OKC FAILED=$FC" > "$DONE"'
|
||||
)
|
||||
script = (
|
||||
'RD="$HOME/' + rel + '"; '
|
||||
'command -v python3 >/dev/null || { echo "NO_PYTHON3"; exit 3; }; '
|
||||
'mkdir -p "$RD"; '
|
||||
'nohup bash -c ' + shlex.quote(inner) +
|
||||
' </dev/null >"$RD/.venv-skills-install.log" 2>&1 & echo LAUNCHED'
|
||||
)
|
||||
ssh_cmd = ["ssh"] + _ssh_common_args(env) + [_ssh_target(env), script]
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*ssh_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
|
||||
stdout, _stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout + 30)
|
||||
except asyncio.TimeoutError:
|
||||
return {"ok": False, "error": f"逐包安装超时（{timeout}s），可重跑"}
|
||||
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
|
||||
except Exception as e:
|
||||
return {"ok": False, "error": "SSH 执行失败: " + str(e)[:300]}
|
||||
logger.warning("launch_remote_install 启动失败: %s", e)
|
||||
return False
|
||||
out = stdout.decode("utf-8", "replace")
|
||||
if "NO_PYTHON3" in out:
|
||||
return False
|
||||
return "LAUNCHED" in out
|
||||
|
||||
|
||||
async def poll_remote_install(env: dict, rel: str) -> dict:
|
||||
"""轮询远程安装进度。短 SSH 调用。
|
||||
|
||||
返回 {done:bool, ok:bool, installed:[], failed:[], venv, ...}
|
||||
"""
|
||||
script = (
|
||||
'RD="$HOME/' + rel + '"; DONE="$RD/.venv-skills-install.done"; '
|
||||
'PROG="$RD/.venv-skills-install.progress"; '
|
||||
'if [ -f "$DONE" ]; then echo "STATE=done"; cat "$DONE"; echo "===PROG==="; '
|
||||
'cat "$PROG" 2>/dev/null; '
|
||||
'else echo "STATE=running"; cat "$PROG" 2>/dev/null; fi'
|
||||
)
|
||||
ssh_cmd = ["ssh"] + _ssh_common_args(env) + [_ssh_target(env), script]
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*ssh_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
|
||||
stdout, _stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
|
||||
except Exception as e:
|
||||
return {"done": False, "error": "轮询失败: " + str(e)[:200]}
|
||||
|
||||
out = stdout.decode("utf-8", "replace")
|
||||
done = "STATE=done" in out
|
||||
body = out.split("STATE=", 1)[1] if "STATE=" in out else out
|
||||
if "===PROG===" in body:
|
||||
summary, prog = body.split("===PROG===", 1)
|
||||
else:
|
||||
summary, prog = body, ""
|
||||
|
||||
installed, failed = [], []
|
||||
for line in out.splitlines():
|
||||
for line in prog.splitlines():
|
||||
line = line.strip()
|
||||
if line.startswith("OK:"):
|
||||
installed.append(line[3:])
|
||||
elif line.startswith("FAILED:"):
|
||||
failed.append(line[7:])
|
||||
return {"ok": len(failed) == 0 and len(installed) > 0,
|
||||
"installed": installed, "failed": failed,
|
||||
"venv": "~/" + rel + "/.venv-skills",
|
||||
"mode": "per-package", "tail": out[-500:]}
|
||||
|
||||
result = {
|
||||
"done": done,
|
||||
"ok": done and len(failed) == 0 and len(installed) > 0,
|
||||
"installed": installed,
|
||||
"failed": failed,
|
||||
"venv": "~/" + rel + "/.venv-skills",
|
||||
"mode": "per-package-detached",
|
||||
}
|
||||
if done:
|
||||
result["summary"] = summary.strip()
|
||||
return result
|
||||
|
||||
|
||||
# ── 总编排 ────────────────────────────────────────────────────
|
||||
@ -399,55 +375,33 @@ async def init_remote_env(env: dict, org_id: str = "", sor=None,
|
||||
report["error"] = "技能同步失败: " + str(sk.get("error", ""))
|
||||
return report
|
||||
|
||||
# 4. 依赖解析（快）+ 安装（慢，后台执行）
|
||||
# 4. 依赖解析（快）+ 启动远程分离式安装（nohup，脱离 SSH 会话，
|
||||
# 进度落远程文件，轮询无状态——服务重启/断连都不影响）
|
||||
cfg = await _load_cfg(sor)
|
||||
deps = collect_skill_deps(skills_base or _default_skills_base(), org_id,
|
||||
cfg["heavy_skip"])
|
||||
report["steps"]["deps_found"] = deps
|
||||
task_key = _start_deps_install(env, deps["specs"], cfg)
|
||||
report["deps_task"] = task_key
|
||||
report["steps"]["deps_install"] = {"status": "running", "task": task_key}
|
||||
launched = await launch_remote_install(env, deps["specs"], cfg["pip_index"], rel)
|
||||
report["steps"]["deps_launch"] = {"ok": launched, "count": len(deps["specs"])}
|
||||
if not launched:
|
||||
report["warning"] = "依赖安装进程启动失败（远程可能缺 python3），请重新运行初始化"
|
||||
return report
|
||||
|
||||
|
||||
# ── 依赖安装后台任务与状态 ────────────────────────────────────
|
||||
# 依赖安装可能耗时数分钟（超过 nginx 600s 代理超时），不能在 API 请求内同步等。
|
||||
# 编排只启动后台任务并返回任务号，前端用 init_remote_env_status 轮询结果。
|
||||
# 状态存进程内存（单进程部署；重启丢状态则重跑，幂等无副作用）。
|
||||
# ── 依赖安装状态查询（无状态，轮询远程进度文件）────────────────
|
||||
# 安装进程由远程 nohup 托管，进度写 ~/<remote_dir>/.venv-skills-install.*。
|
||||
# 查询只需 SSH 读文件，不依赖本进程内存，服务重启后仍可查。
|
||||
|
||||
_INSTALL_TASKS = {} # {task_key: {"status": ..., "result": ..., "started": ts}}
|
||||
|
||||
|
||||
def _start_deps_install(env: dict, specs: list, cfg: dict) -> str:
|
||||
from appPublic.uniqueID import getID
|
||||
task_key = "deps_" + getID()
|
||||
_INSTALL_TASKS[task_key] = {"status": "running", "result": None,
|
||||
"specs": list(specs)}
|
||||
|
||||
async def _run():
|
||||
try:
|
||||
r = await install_remote_deps(env, specs, cfg["pip_index"],
|
||||
cfg["install_timeout"])
|
||||
_INSTALL_TASKS[task_key]["result"] = r
|
||||
_INSTALL_TASKS[task_key]["status"] = "done"
|
||||
except Exception as e:
|
||||
_INSTALL_TASKS[task_key]["result"] = {"ok": False, "error": str(e)[:500]}
|
||||
_INSTALL_TASKS[task_key]["status"] = "done"
|
||||
|
||||
asyncio.create_task(_run())
|
||||
return task_key
|
||||
|
||||
|
||||
def get_deps_install_status(task_key: str) -> dict:
|
||||
t = _INSTALL_TASKS.get(task_key or "")
|
||||
if not t:
|
||||
return {"ok": False, "error": "任务不存在（可能服务重启过，请重新初始化）"}
|
||||
if t["status"] == "running":
|
||||
return {"ok": True, "status": "running",
|
||||
"message": f"依赖安装中（{len(t.get('specs', []))} 个包），请稍候再查询"}
|
||||
r = dict(t.get("result") or {})
|
||||
r["status"] = "done"
|
||||
return r
|
||||
async def get_deps_install_status(sor, owner_type: str, owner_id: str) -> dict:
|
||||
"""查询远程依赖安装进度。从 sd_work_envs 取远程配置后轮询进度文件。"""
|
||||
from .work_env import get_work_env
|
||||
env = await get_work_env(
|
||||
sor, owner_id if owner_type == "user" else "",
|
||||
owner_id if owner_type == "org" else "")
|
||||
if env.get("mode") != "remote" or not env.get("remote_dir"):
|
||||
return {"ok": False, "error": "当前非远程模式，无安装任务"}
|
||||
rel = _rel_remote_dir(env.get("remote_dir", ""))
|
||||
return await poll_remote_install(env, rel)
|
||||
|
||||
|
||||
def _default_skills_base() -> str:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user