649 lines
24 KiB
Python
649 lines
24 KiB
Python
from __future__ import annotations
|
||
|
||
import base64
|
||
import copy
|
||
from datetime import datetime
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
import shlex
|
||
import shutil
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from typing import Any, Callable
|
||
import uuid
|
||
|
||
from core.config import (
|
||
BUILD_DOCKER_PLATFORM,
|
||
CHATAPP_CRON_BINARY_PATH,
|
||
CHATAPP_CRON_COMPOSE_FILE,
|
||
CHATAPP_CRON_COMPOSE_SERVICE,
|
||
CHATAPP_CRON_DEFAULT_GIT_REF,
|
||
CHATAPP_CRON_ENABLED,
|
||
CHATAPP_CRON_HOST,
|
||
CHATAPP_CRON_IP,
|
||
CHATAPP_CRON_INSTANCE_ID,
|
||
CHATAPP_CRON_RELEASE_META_PATH,
|
||
CHATAPP_CRON_REMOTE_DIR,
|
||
CHATAPP_CRON_REPO_ROOT,
|
||
CHATAPP_CRON_REPO_URL,
|
||
CHATAPP_CRON_SERVICE_NAME,
|
||
CHATAPP_CRON_SSH_HOST,
|
||
CHATAPP_CRON_SSH_PORT,
|
||
CHATAPP_CRON_SSH_USE_SUDO,
|
||
CHATAPP_CRON_SSH_USER,
|
||
ROOT,
|
||
SERVICE_BUILD_TIMEOUT_SECONDS,
|
||
SERVICE_PULL_TIMEOUT_SECONDS,
|
||
utc_now,
|
||
)
|
||
from .service_release import (
|
||
create_temp_worktree,
|
||
emit_progress,
|
||
ensure_repo_root,
|
||
fetch_repo_refs,
|
||
local_command_output,
|
||
remove_temp_worktree,
|
||
resolve_repo_commit,
|
||
run_logged_command,
|
||
sanitized_ref_fragment,
|
||
short_text,
|
||
)
|
||
from .ssh_remote import copy_local_path_to_host, read_remote_text, run_host_script_checked, target_display
|
||
|
||
|
||
# 后台任务的日志回调统一走这个签名。
|
||
ProgressLogger = Callable[[str, str, str], None]
|
||
# 内存里最多保留最近若干次 cron 部署记录。
|
||
MAX_CRON_DEPLOY_OPERATIONS = 10
|
||
# 任务状态和完整日志统一落到 run 目录。
|
||
CRON_DEPLOY_JOB_DIR = ROOT / "run" / "chatapp_cron_deploy_jobs"
|
||
# 统一保护任务状态的锁。
|
||
CRON_DEPLOY_JOB_LOCK = threading.Lock()
|
||
# 当前进程内的 cron 部署任务表。
|
||
CRON_DEPLOY_JOBS: dict[str, dict[str, Any]] = {}
|
||
# 记录任务插入顺序，便于淘汰最旧项。
|
||
CRON_DEPLOY_JOB_ORDER: list[str] = []
|
||
|
||
|
||
def chatapp_cron_release_enabled() -> bool:
|
||
# 只有目标机、运行目录和仓库信息齐备时才开放部署按钮。
|
||
return bool(
|
||
CHATAPP_CRON_ENABLED
|
||
and str(CHATAPP_CRON_SERVICE_NAME or "").strip()
|
||
and str(CHATAPP_CRON_HOST or "").strip()
|
||
and str(CHATAPP_CRON_REMOTE_DIR or "").strip()
|
||
and str(CHATAPP_CRON_BINARY_PATH or "").strip()
|
||
and str(CHATAPP_CRON_COMPOSE_FILE or "").strip()
|
||
and str(CHATAPP_CRON_COMPOSE_SERVICE or "").strip()
|
||
and (str(CHATAPP_CRON_SSH_HOST or "").strip() or str(CHATAPP_CRON_IP or "").strip())
|
||
and (str(CHATAPP_CRON_REPO_URL or "").strip() or str(CHATAPP_CRON_REPO_ROOT or "").strip())
|
||
)
|
||
|
||
|
||
def chatapp_cron_host_record() -> dict[str, Any]:
|
||
# chatapp-cron 不在 Yumi 业务服务 targets 里，部署主机单独从 .env 读取。
|
||
if not chatapp_cron_release_enabled():
|
||
raise RuntimeError("chatapp-cron deploy is not configured")
|
||
|
||
return {
|
||
"host": CHATAPP_CRON_HOST,
|
||
"ip": CHATAPP_CRON_IP,
|
||
"instanceId": CHATAPP_CRON_INSTANCE_ID,
|
||
"sshHost": CHATAPP_CRON_SSH_HOST or CHATAPP_CRON_IP,
|
||
"sshUser": CHATAPP_CRON_SSH_USER,
|
||
"sshUseSudo": CHATAPP_CRON_SSH_USE_SUDO,
|
||
"sshPort": CHATAPP_CRON_SSH_PORT,
|
||
}
|
||
|
||
|
||
def ensure_chatapp_cron_repo_root(*, progress: ProgressLogger | None = None) -> Path:
|
||
# 第一次使用时允许按配置仓库地址自动 clone 到 deploy 机。
|
||
repo_root = Path(CHATAPP_CRON_REPO_ROOT).expanduser().resolve()
|
||
if (repo_root / ".git").exists():
|
||
return repo_root
|
||
|
||
repo_url = str(CHATAPP_CRON_REPO_URL or "").strip()
|
||
if not repo_url:
|
||
raise RuntimeError("missing CHATAPP_CRON_REPO_URL in .env")
|
||
|
||
repo_root.parent.mkdir(parents=True, exist_ok=True)
|
||
emit_progress(progress, "cron.repo", f"初始化 chatapp-cron 仓库：{repo_url} -> {repo_root}")
|
||
local_command_output(
|
||
["git", "clone", repo_url, str(repo_root)],
|
||
cwd=repo_root.parent,
|
||
label="chatapp-cron clone repo",
|
||
timeout_seconds=900,
|
||
)
|
||
return ensure_repo_root(repo_root, "chatapp-cron")
|
||
|
||
|
||
def read_chatapp_cron_metadata() -> dict[str, Any]:
|
||
# 元数据只用于展示最近部署版本，读取失败不影响页面打开。
|
||
if not chatapp_cron_release_enabled():
|
||
return {}
|
||
|
||
try:
|
||
text = read_remote_text(chatapp_cron_host_record(), CHATAPP_CRON_RELEASE_META_PATH, timeout_seconds=30)
|
||
return json.loads(text or "{}")
|
||
except Exception as exc: # noqa: BLE001
|
||
return {"error": str(exc)}
|
||
|
||
|
||
def build_chatapp_cron_overview_payload() -> dict[str, Any]:
|
||
# 通用页加载时返回配置摘要和最近部署版本。
|
||
enabled = chatapp_cron_release_enabled()
|
||
settings = {
|
||
"enabled": enabled,
|
||
"serviceName": CHATAPP_CRON_SERVICE_NAME,
|
||
"host": CHATAPP_CRON_HOST,
|
||
"sshHost": CHATAPP_CRON_SSH_HOST or CHATAPP_CRON_IP,
|
||
"instanceId": CHATAPP_CRON_INSTANCE_ID,
|
||
"repoRoot": str(Path(CHATAPP_CRON_REPO_ROOT).expanduser()),
|
||
"repoUrl": CHATAPP_CRON_REPO_URL,
|
||
"defaultGitRef": CHATAPP_CRON_DEFAULT_GIT_REF,
|
||
"remoteDir": CHATAPP_CRON_REMOTE_DIR,
|
||
"composeFile": CHATAPP_CRON_COMPOSE_FILE,
|
||
"composeService": CHATAPP_CRON_COMPOSE_SERVICE,
|
||
"binaryPath": CHATAPP_CRON_BINARY_PATH,
|
||
"dockerPlatform": BUILD_DOCKER_PLATFORM,
|
||
}
|
||
return {
|
||
"ok": True,
|
||
"enabled": enabled,
|
||
"configured": enabled,
|
||
"updatedAt": utc_now(),
|
||
"settings": settings,
|
||
"metadata": read_chatapp_cron_metadata() if enabled else {},
|
||
}
|
||
|
||
|
||
def docker_builder_image_name(commit: str) -> str:
|
||
# 本地临时 builder 镜像只用于 docker cp 导出二进制。
|
||
stamp = datetime.utcnow().strftime("%Y%m%d%H%M%S")
|
||
tag = sanitized_ref_fragment(f"{commit[:12]}-{stamp}").lower()
|
||
return f"hy-app-monitor/chatapp-cron-builder:{tag}"
|
||
|
||
|
||
def build_chatapp_cron_binary(git_ref: str, *, progress: ProgressLogger | None = None) -> dict[str, Any]:
|
||
# 使用仓库 Dockerfile 的 builder stage 产出 linux/amd64 二进制，避免依赖宿主机 Go 版本。
|
||
repo_root = ensure_chatapp_cron_repo_root(progress=progress)
|
||
resolved_ref = str(git_ref or "").strip() or CHATAPP_CRON_DEFAULT_GIT_REF
|
||
emit_progress(progress, "cron.build", f"刷新 chatapp-cron refs：{resolved_ref}")
|
||
fetch_repo_refs(repo_root, "chatapp-cron")
|
||
commit = resolve_repo_commit(repo_root, resolved_ref, "chatapp-cron")
|
||
emit_progress(progress, "cron.build", f"chatapp-cron ref 已解析：{resolved_ref} -> {commit[:12]}")
|
||
|
||
worktree_root = create_temp_worktree(repo_root, commit, "chatapp-cron")
|
||
artifact_root = Path(tempfile.mkdtemp(prefix="hy-app-monitor-chatapp-cron-artifact-"))
|
||
builder_image = docker_builder_image_name(commit)
|
||
container_id = ""
|
||
binary_path = artifact_root / "week-star-job"
|
||
|
||
try:
|
||
emit_progress(progress, "cron.build", f"开始 Docker builder 构建：{BUILD_DOCKER_PLATFORM}")
|
||
run_logged_command(
|
||
["docker", "build", "--platform", BUILD_DOCKER_PLATFORM, "--target", "builder", "-t", builder_image, "."],
|
||
cwd=worktree_root,
|
||
label="chatapp-cron docker builder",
|
||
timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS,
|
||
progress_stage="cron.build.docker",
|
||
progress=progress,
|
||
)
|
||
|
||
container_id = local_command_output(
|
||
["docker", "create", builder_image],
|
||
cwd=worktree_root,
|
||
label="chatapp-cron create builder container",
|
||
timeout_seconds=120,
|
||
)
|
||
local_command_output(
|
||
["docker", "cp", f"{container_id}:/out/week-star-job", str(binary_path)],
|
||
cwd=worktree_root,
|
||
label="chatapp-cron export binary",
|
||
timeout_seconds=180,
|
||
)
|
||
binary_path.chmod(0o755)
|
||
emit_progress(progress, "cron.build", f"二进制已生成：{binary_path}")
|
||
return {
|
||
"kind": "chatapp-cron",
|
||
"gitRef": resolved_ref,
|
||
"commit": commit,
|
||
"binaryPath": str(binary_path),
|
||
"builderImage": builder_image,
|
||
}
|
||
finally:
|
||
if container_id:
|
||
try:
|
||
local_command_output(
|
||
["docker", "rm", "-f", container_id],
|
||
cwd=worktree_root,
|
||
label="chatapp-cron remove builder container",
|
||
timeout_seconds=60,
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
local_command_output(
|
||
["docker", "rmi", "-f", builder_image],
|
||
cwd=worktree_root,
|
||
label="chatapp-cron remove builder image",
|
||
timeout_seconds=120,
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
remove_temp_worktree(repo_root, worktree_root)
|
||
|
||
|
||
def activate_chatapp_cron_binary(build_result: dict[str, Any], *, progress: ProgressLogger | None = None) -> dict[str, Any]:
|
||
# 把新二进制传到 cron 机器，再用 docker compose 重建 week-star-job。
|
||
host = chatapp_cron_host_record()
|
||
binary_path = Path(str(build_result["binaryPath"]))
|
||
if not binary_path.exists():
|
||
raise RuntimeError(f"chatapp-cron binary missing: {binary_path}")
|
||
|
||
remote_tmp = f"/tmp/hy-app-monitor-chatapp-cron-{build_result['commit'][:12]}-{int(time.time())}"
|
||
emit_progress(progress, "cron.deploy.copy", f"上传二进制到 {target_display(host)}:{remote_tmp}")
|
||
copy_local_path_to_host(
|
||
host,
|
||
binary_path,
|
||
remote_tmp,
|
||
timeout_seconds=SERVICE_PULL_TIMEOUT_SECONDS,
|
||
)
|
||
|
||
metadata = {
|
||
"service": CHATAPP_CRON_SERVICE_NAME,
|
||
"component": CHATAPP_CRON_COMPOSE_SERVICE,
|
||
"host": CHATAPP_CRON_HOST,
|
||
"gitRef": build_result["gitRef"],
|
||
"commit": build_result["commit"],
|
||
"deployedAt": utc_now(),
|
||
"binaryPath": CHATAPP_CRON_BINARY_PATH,
|
||
"composeFile": CHATAPP_CRON_COMPOSE_FILE,
|
||
"remoteDir": CHATAPP_CRON_REMOTE_DIR,
|
||
}
|
||
metadata_blob = base64.b64encode(json.dumps(metadata, ensure_ascii=False, indent=2).encode("utf-8")).decode("ascii")
|
||
|
||
script = f"""set -euo pipefail
|
||
REMOTE_DIR={shlex.quote(CHATAPP_CRON_REMOTE_DIR)}
|
||
COMPOSE_FILE={shlex.quote(CHATAPP_CRON_COMPOSE_FILE)}
|
||
COMPOSE_SERVICE={shlex.quote(CHATAPP_CRON_COMPOSE_SERVICE)}
|
||
BINARY_PATH={shlex.quote(CHATAPP_CRON_BINARY_PATH)}
|
||
REMOTE_TMP={shlex.quote(remote_tmp)}
|
||
META_PATH={shlex.quote(CHATAPP_CRON_RELEASE_META_PATH)}
|
||
META_BLOB={shlex.quote(metadata_blob)}
|
||
|
||
compose() {{
|
||
if docker compose version >/dev/null 2>&1; then
|
||
docker compose "$@"
|
||
else
|
||
docker-compose "$@"
|
||
fi
|
||
}}
|
||
|
||
mkdir -p "$(dirname "$BINARY_PATH")" "$REMOTE_DIR/.backups"
|
||
if [ -f "$BINARY_PATH" ]; then
|
||
cp "$BINARY_PATH" "$REMOTE_DIR/.backups/week-star-job.$(date +%Y%m%d-%H%M%S)"
|
||
fi
|
||
install -m 0755 "$REMOTE_TMP" "$BINARY_PATH"
|
||
rm -f "$REMOTE_TMP"
|
||
|
||
cd "$REMOTE_DIR"
|
||
compose -f "$COMPOSE_FILE" up -d --no-deps --force-recreate "$COMPOSE_SERVICE"
|
||
|
||
for _ in $(seq 1 30); do
|
||
if compose -f "$COMPOSE_FILE" ps "$COMPOSE_SERVICE" | grep -Eiq 'running|up'; then
|
||
break
|
||
fi
|
||
sleep 2
|
||
done
|
||
|
||
STATUS_OUTPUT="$(compose -f "$COMPOSE_FILE" ps "$COMPOSE_SERVICE")"
|
||
printf '%s\\n' "$STATUS_OUTPUT"
|
||
if ! printf '%s\\n' "$STATUS_OUTPUT" | grep -Eiq 'running|up'; then
|
||
exit 1
|
||
fi
|
||
|
||
mkdir -p "$(dirname "$META_PATH")"
|
||
printf '%s' "$META_BLOB" | base64 -d > "$META_PATH"
|
||
"""
|
||
|
||
emit_progress(progress, "cron.deploy.activate", f"重建远端服务：{CHATAPP_CRON_COMPOSE_SERVICE}")
|
||
result = run_host_script_checked(
|
||
host,
|
||
script,
|
||
command_name="chatapp-cron-activate",
|
||
timeout_seconds=SERVICE_PULL_TIMEOUT_SECONDS,
|
||
)
|
||
emit_progress(progress, "cron.deploy.activate", "chatapp-cron 重启完成")
|
||
return {
|
||
"host": CHATAPP_CRON_HOST,
|
||
"target": result.get("target"),
|
||
"status": result.get("status"),
|
||
"output": short_text(str(result.get("output") or ""), 1000),
|
||
"metadata": metadata,
|
||
}
|
||
|
||
|
||
def deploy_chatapp_cron_payload(git_ref: str, *, progress: ProgressLogger | None = None) -> dict[str, Any]:
|
||
# 对外的部署主流程：构建二进制 -> 上传目标机 -> compose 重建。
|
||
if not chatapp_cron_release_enabled():
|
||
raise RuntimeError("chatapp-cron deploy is not configured")
|
||
|
||
started_at = time.time()
|
||
build_result = build_chatapp_cron_binary(git_ref, progress=progress)
|
||
try:
|
||
deploy_result = activate_chatapp_cron_binary(build_result, progress=progress)
|
||
return {
|
||
"ok": True,
|
||
"message": "chatapp-cron deploy complete",
|
||
"updatedAt": utc_now(),
|
||
"durationSeconds": round(time.time() - started_at, 2),
|
||
"build": {
|
||
"kind": build_result["kind"],
|
||
"gitRef": build_result["gitRef"],
|
||
"commit": build_result["commit"],
|
||
"services": [CHATAPP_CRON_SERVICE_NAME],
|
||
},
|
||
"deploy": deploy_result,
|
||
}
|
||
finally:
|
||
artifact_root = Path(str(build_result.get("binaryPath") or "")).parent
|
||
if str(artifact_root).startswith(tempfile.gettempdir()):
|
||
shutil.rmtree(artifact_root, ignore_errors=True)
|
||
|
||
|
||
def ensure_cron_deploy_job_dir() -> None:
|
||
# 持久化目录不存在时先创建。
|
||
CRON_DEPLOY_JOB_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def cron_deploy_job_state_path(operation_id: str) -> Path:
|
||
# 每个任务一份状态 JSON。
|
||
return CRON_DEPLOY_JOB_DIR / f"{operation_id}.json"
|
||
|
||
|
||
def cron_deploy_job_log_path(operation_id: str) -> Path:
|
||
# 每个任务一份完整文本日志。
|
||
return CRON_DEPLOY_JOB_DIR / f"{operation_id}.log"
|
||
|
||
|
||
def valid_operation_id(operation_id: str) -> bool:
|
||
# 只接受简单安全字符，避免路径拼接被滥用。
|
||
return re.fullmatch(r"[0-9A-Za-z_-]{6,64}", str(operation_id or "").strip()) is not None
|
||
|
||
|
||
def cron_operation_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
# 对外返回和落盘都统一走这个结构。
|
||
return {
|
||
"operationId": payload["operationId"],
|
||
"kind": payload["kind"],
|
||
"status": payload["status"],
|
||
"stage": payload["stage"],
|
||
"message": payload["message"],
|
||
"error": payload["error"],
|
||
"startedAt": payload["startedAt"],
|
||
"updatedAt": payload["updatedAt"],
|
||
"finishedAt": payload["finishedAt"],
|
||
"logs": copy.deepcopy(payload.get("logs") or []),
|
||
"result": copy.deepcopy(payload.get("result")),
|
||
"request": copy.deepcopy(payload.get("request") or {}),
|
||
"logUrl": f"/api/chatapp-cron/deploy-log?id={payload['operationId']}",
|
||
}
|
||
|
||
|
||
def persist_cron_job_unlocked(payload: dict[str, Any]) -> None:
|
||
# 每次状态变化都落盘，服务重启后还能恢复最近任务。
|
||
ensure_cron_deploy_job_dir()
|
||
cron_deploy_job_state_path(payload["operationId"]).write_text(
|
||
json.dumps({"ok": True, "operation": cron_operation_payload(payload)}, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def append_cron_log_file(operation_id: str, line: str) -> None:
|
||
# 完整日志单独写入文本文件，供页面查看。
|
||
ensure_cron_deploy_job_dir()
|
||
with cron_deploy_job_log_path(operation_id).open("a", encoding="utf-8") as handle:
|
||
handle.write(line)
|
||
|
||
|
||
def trim_cron_jobs_unlocked() -> None:
|
||
# 只保留最近若干次任务，老数据直接淘汰内存引用。
|
||
while len(CRON_DEPLOY_JOB_ORDER) > MAX_CRON_DEPLOY_OPERATIONS:
|
||
operation_id = CRON_DEPLOY_JOB_ORDER.pop(0)
|
||
CRON_DEPLOY_JOBS.pop(operation_id, None)
|
||
|
||
|
||
def create_cron_deploy_job_payload(git_ref: str) -> dict[str, Any]:
|
||
# 后台任务一创建就落一份初始状态，前端可以马上开始轮询。
|
||
resolved_ref = str(git_ref or "").strip() or CHATAPP_CRON_DEFAULT_GIT_REF
|
||
operation_id = uuid.uuid4().hex[:12]
|
||
now = utc_now()
|
||
payload = {
|
||
"operationId": operation_id,
|
||
"kind": "chatapp-cron-deploy",
|
||
"status": "queued",
|
||
"stage": "queued",
|
||
"message": "chatapp-cron deploy queued",
|
||
"error": "",
|
||
"startedAt": now,
|
||
"updatedAt": now,
|
||
"finishedAt": "",
|
||
"logs": [],
|
||
"result": None,
|
||
"request": {
|
||
"gitRef": resolved_ref,
|
||
"service": CHATAPP_CRON_SERVICE_NAME,
|
||
},
|
||
}
|
||
with CRON_DEPLOY_JOB_LOCK:
|
||
ensure_cron_deploy_job_dir()
|
||
CRON_DEPLOY_JOBS[operation_id] = payload
|
||
CRON_DEPLOY_JOB_ORDER.append(operation_id)
|
||
cron_deploy_job_log_path(operation_id).write_text("", encoding="utf-8")
|
||
persist_cron_job_unlocked(payload)
|
||
trim_cron_jobs_unlocked()
|
||
return snapshot_cron_deploy_job(operation_id)
|
||
|
||
|
||
def append_cron_job_log(operation_id: str, stage: str, message: str, *, level: str = "info") -> None:
|
||
# 每条日志都带时间和阶段，页面和状态文件都保留日志。
|
||
text = str(message or "").rstrip()
|
||
if not text:
|
||
return
|
||
|
||
with CRON_DEPLOY_JOB_LOCK:
|
||
payload = CRON_DEPLOY_JOBS.get(operation_id)
|
||
if payload is None:
|
||
return
|
||
|
||
payload["updatedAt"] = utc_now()
|
||
payload["stage"] = str(stage or payload.get("stage") or "running")
|
||
payload["logs"].append(
|
||
{
|
||
"time": payload["updatedAt"],
|
||
"stage": str(stage or "").strip(),
|
||
"level": str(level or "info").strip() or "info",
|
||
"message": text,
|
||
}
|
||
)
|
||
append_cron_log_file(
|
||
operation_id,
|
||
f"[{payload['updatedAt']}] [{str(stage or '').strip() or payload['stage']}] [{str(level or 'info').strip() or 'info'}] {text}\n",
|
||
)
|
||
persist_cron_job_unlocked(payload)
|
||
|
||
|
||
def patch_cron_job(operation_id: str, **fields: Any) -> None:
|
||
# 任务元信息更新统一走这里，避免漏掉 updatedAt。
|
||
with CRON_DEPLOY_JOB_LOCK:
|
||
payload = CRON_DEPLOY_JOBS.get(operation_id)
|
||
if payload is None:
|
||
return
|
||
payload.update(fields)
|
||
payload["updatedAt"] = utc_now()
|
||
persist_cron_job_unlocked(payload)
|
||
|
||
|
||
def finish_cron_job(operation_id: str, *, result: dict[str, Any], status: str, error: str = "") -> None:
|
||
# 结束态统一收口，前端轮询时不用自己猜。
|
||
now = utc_now()
|
||
with CRON_DEPLOY_JOB_LOCK:
|
||
payload = CRON_DEPLOY_JOBS.get(operation_id)
|
||
if payload is None:
|
||
return
|
||
payload["status"] = status
|
||
payload["stage"] = status
|
||
payload["message"] = (
|
||
str(result.get("message") or "").strip()
|
||
or ("chatapp-cron deploy complete" if status == "success" else "chatapp-cron deploy failed")
|
||
)
|
||
payload["error"] = str(error or "").strip()
|
||
payload["finishedAt"] = now
|
||
payload["updatedAt"] = now
|
||
payload["result"] = result
|
||
persist_cron_job_unlocked(payload)
|
||
|
||
|
||
def snapshot_cron_deploy_job(operation_id: str) -> dict[str, Any]:
|
||
# HTTP 返回前复制一份，避免把锁外的可变对象直接暴露出去。
|
||
with CRON_DEPLOY_JOB_LOCK:
|
||
payload = CRON_DEPLOY_JOBS.get(operation_id)
|
||
if payload is None:
|
||
raise KeyError(operation_id)
|
||
return {"ok": True, "operation": cron_operation_payload(payload)}
|
||
|
||
|
||
def cron_job_progress_logger(operation_id: str) -> ProgressLogger:
|
||
# 部署主流程只关心 stage / level / message，落盘细节都交给这里。
|
||
def _logger(stage: str, level: str, message: str) -> None:
|
||
append_cron_job_log(operation_id, stage, message, level=level)
|
||
|
||
return _logger
|
||
|
||
|
||
def run_chatapp_cron_deploy_job(operation_id: str, git_ref: str) -> None:
|
||
# 真正的部署逻辑放后台线程里执行。
|
||
patch_cron_job(operation_id, status="running", stage="prepare", message="chatapp-cron deploy running")
|
||
append_cron_job_log(operation_id, "prepare", f"开始部署 chatapp-cron：{git_ref}")
|
||
|
||
try:
|
||
result = deploy_chatapp_cron_payload(git_ref, progress=cron_job_progress_logger(operation_id))
|
||
append_cron_job_log(operation_id, "success", "chatapp-cron 部署完成")
|
||
finish_cron_job(operation_id, result=result, status="success")
|
||
except Exception as exc: # noqa: BLE001
|
||
error = str(exc)
|
||
append_cron_job_log(operation_id, "failed", error, level="error")
|
||
finish_cron_job(
|
||
operation_id,
|
||
result={"ok": False, "updatedAt": utc_now(), "error": error},
|
||
status="failed",
|
||
error=error,
|
||
)
|
||
|
||
|
||
def start_chatapp_cron_deploy_job(git_ref: str) -> dict[str, Any]:
|
||
# 创建任务后立刻拉起后台线程，并把 operationId 返回给前端。
|
||
if not chatapp_cron_release_enabled():
|
||
raise RuntimeError("chatapp-cron deploy is not configured")
|
||
|
||
payload = create_cron_deploy_job_payload(git_ref)
|
||
operation_id = payload["operation"]["operationId"]
|
||
resolved_ref = payload["operation"]["request"]["gitRef"]
|
||
|
||
worker = threading.Thread(
|
||
target=run_chatapp_cron_deploy_job,
|
||
kwargs={"operation_id": operation_id, "git_ref": resolved_ref},
|
||
name=f"chatapp-cron-deploy-{operation_id}",
|
||
daemon=True,
|
||
)
|
||
worker.start()
|
||
return snapshot_cron_deploy_job(operation_id)
|
||
|
||
|
||
def recover_lost_cron_job_unlocked(payload: dict[str, Any]) -> None:
|
||
# 进程重启后，原后台线程已经不存在；把悬挂中的任务收口成失败态。
|
||
if payload.get("status") not in {"queued", "running"}:
|
||
return
|
||
|
||
now = utc_now()
|
||
message = "monitor process restarted before chatapp-cron deploy status could be finalized"
|
||
payload["status"] = "failed"
|
||
payload["stage"] = "failed"
|
||
payload["message"] = "chatapp-cron deploy interrupted"
|
||
payload["error"] = message
|
||
payload["updatedAt"] = now
|
||
payload["finishedAt"] = now
|
||
payload["logs"].append(
|
||
{
|
||
"time": now,
|
||
"stage": "failed",
|
||
"level": "error",
|
||
"message": message,
|
||
}
|
||
)
|
||
append_cron_log_file(payload["operationId"], f"[{now}] [failed] [error] {message}\n")
|
||
persist_cron_job_unlocked(payload)
|
||
|
||
|
||
def load_cron_job_from_disk(operation_id: str) -> dict[str, Any] | None:
|
||
# 任务不在内存时，从持久化状态恢复。
|
||
if not valid_operation_id(operation_id):
|
||
return None
|
||
|
||
state_path = cron_deploy_job_state_path(operation_id)
|
||
if not state_path.exists():
|
||
return None
|
||
|
||
payload = json.loads(state_path.read_text(encoding="utf-8"))
|
||
operation = dict(payload.get("operation") or {})
|
||
if not operation:
|
||
return None
|
||
|
||
restored = {
|
||
"operationId": str(operation.get("operationId") or operation_id).strip(),
|
||
"kind": str(operation.get("kind") or "chatapp-cron-deploy").strip() or "chatapp-cron-deploy",
|
||
"status": str(operation.get("status") or "").strip(),
|
||
"stage": str(operation.get("stage") or "").strip(),
|
||
"message": str(operation.get("message") or "").strip(),
|
||
"error": str(operation.get("error") or "").strip(),
|
||
"startedAt": str(operation.get("startedAt") or "").strip(),
|
||
"updatedAt": str(operation.get("updatedAt") or "").strip(),
|
||
"finishedAt": str(operation.get("finishedAt") or "").strip(),
|
||
"logs": list(operation.get("logs") or []),
|
||
"result": operation.get("result"),
|
||
"request": dict(operation.get("request") or {}),
|
||
}
|
||
recover_lost_cron_job_unlocked(restored)
|
||
return restored
|
||
|
||
|
||
def get_chatapp_cron_deploy_job_payload(operation_id: str) -> dict[str, Any]:
|
||
# 供 HTTP 轮询接口直接读取任务状态。
|
||
try:
|
||
return snapshot_cron_deploy_job(operation_id)
|
||
except KeyError:
|
||
restored = load_cron_job_from_disk(operation_id)
|
||
if restored is None:
|
||
raise
|
||
|
||
with CRON_DEPLOY_JOB_LOCK:
|
||
CRON_DEPLOY_JOBS[operation_id] = restored
|
||
if operation_id not in CRON_DEPLOY_JOB_ORDER:
|
||
CRON_DEPLOY_JOB_ORDER.append(operation_id)
|
||
trim_cron_jobs_unlocked()
|
||
return snapshot_cron_deploy_job(operation_id)
|
||
|
||
|
||
def get_chatapp_cron_deploy_log_text(operation_id: str) -> str:
|
||
# 下载或弹层展示完整任务日志。
|
||
if not valid_operation_id(operation_id):
|
||
raise KeyError(operation_id)
|
||
log_path = cron_deploy_job_log_path(operation_id)
|
||
if not log_path.exists():
|
||
if load_cron_job_from_disk(operation_id) is None:
|
||
raise KeyError(operation_id)
|
||
return log_path.read_text(encoding="utf-8") if log_path.exists() else ""
|